Contexto del problema en sistemas de LLM

Al interactuar con modelos de lenguaje grande (LLM) a través de APIs, es común generar historiales de conversación donde un mismo rol (como el usuario o el sistema) emite múltiples mensajes de forma secuencial. Muchos proveedores de modelos restringen esta práctica, exigiendo una alternancia estricta de roles (por ejemplo, system, user, assistant). Para evitar errores de validación en el endpoint y optimizar el consumo de tokens, es fundamental consolidar estos bloques de mensajes adyacentes en una sola entidad.

Implementación con LangChain Core

El ecosistema de LangChain proporciona utilidades nativas para resolver este escenario. La función merge_message_runs permite agrupar automáticamente mensajes consecutivos que comparten el mismo rol, concatenando su contenido de manera segura.

Ejemplo 1: Consolidación desde una lista estructurada

A continuación, se muestra cómo procesar un historial crudo donde se intercalan mensajes del mismo autor, transformándolo en una secuencia válida para el modelo:

from langchain_core.messages import (
    AIMessage,
    HumanMessage,
    SystemMessage,
    merge_message_runs,
)

raw_conversation = [
    SystemMessage(content="Eres un experto en arquitectura de software."),
    SystemMessage(content="Siempre proporcionas ejemplos prácticos en Python."),
    HumanMessage(content="¿Cuál es la mejor forma de manejar la concurrencia?"),
    HumanMessage(content="¿Debería usar asyncio o threading?"),
    AIMessage(content="Para operaciones de E/S, asyncio es la opción ideal."),
    AIMessage(content="Sin embargo, para tareas intensivas de CPU, multiprocessing es mejor debido al GIL."),
]

consolidated_history = merge_message_runs(raw_conversation)

for msg in consolidated_history:
    print(f"[{msg.type.upper()}]: {msg.content}\n")

Ejemplo 2: Construcción dinámica y fusión de secuencias

En arquitecturas más complejas, los mensajes pueden generarse en diferentes módulos o bloques de código. Podemos utilizar el operador de concatenación de listas (+) para unificar estas secuencias antes de aplicar la consolidación final:

from langchain_core.messages import (
    AIMessage,
    HumanMessage,
    SystemMessage,
    merge_message_runs,
)

# Simulación de mensajes generados en diferentes componentes del sistema
system_prompts = [
    SystemMessage(content="Actúa como un tutor de matemáticas avanzadas."),
    SystemMessage(content="Usa un tono amigable y explica paso a paso.")
]

user_queries = [
    HumanMessage(content="Explícame el concepto de derivada."),
    HumanMessage(content="¿Cómo se aplica este concepto en la física clásica?")
]

# Concatenación de listas de mensajes usando el operador +
full_interaction = system_prompts + user_queries + [
    AIMessage(content="La derivada mide la tasa de cambio instantánea de una función."),
    AIMessage(content="En física clásica, es fundamental para calcular velocidad instantánea y aceleración a partir de la posición.")
]

optimized_messages = merge_message_runs(full_interaction)

for msg in optimized_messages:
    print(f"--- Rol: {msg.type} ---\n{msg.content}\n")

Consideraciones técnicas y buenas prácticas

• Formato de contenido: Si los mensajes contienen estructuras complejas (como listas de diccionarios para contenido mulitmodal o herramientas), la función de fusión intentará concatenar los campos de texto. Es crucial validar que la estructura final resultante sea compatible con el esquema de entrada del modelo objetivo.
• Preservación de metadatos: Al unir mensajes, los metadatos individuales (como name, id o response_metadata) del primer mensaje en la secuencia suelen conservarse, mientras que los de los mensajes subsiguientes pueden descartarse. Asegúrese de que esta pérdida de metadatos no afecte la lógica de rastreo o auditoría de su aplicación.
• Manjo de errores: Si la lista de entrada está vacía o no contiene objetos válidos de BaseMessage, la función puede lanzar excepciones. Implemente validaciones previas en pipelines de producción donde el historial se construya dinámicamente.