Gestión de conversaciones

Uso de trim_messages

Cuando trabajamos con conversaciones largas en LangChain, uno de los principales desafíos es gestionar el límite de tokens que pueden procesar los modelos de lenguaje. Cada modelo tiene una ventana de contexto limitada, y si nuestros mensajes exceden este límite, obtendremos errores o comportamientos inesperados.

La función trim_messages de LangChain Core nos proporciona una solución elegante para recortar automáticamente los mensajes de una conversación, manteniendo solo los más relevantes según diferentes estrategias.

Configuración básica de trim_messages

Para utilizar trim_messages, primero necesitamos importarla y configurar sus parámetros principales:

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, trim_messages
from langchain_openai import ChatOpenAI

# Configuramos nuestro modelo
model = ChatOpenAI(model="gpt-4o")

# Creamos el trimmer con configuración básica
trimmer = trim_messages(
    max_tokens=100,
    strategy="last",
    token_counter=model,
    include_system=True
)

El parámetro max_tokens define el límite máximo de tokens que queremos mantener en nuestra conversación. La strategy determina qué mensajes conservar cuando necesitamos recortar, y token_counter especifica qué modelo usar para contar los tokens.

Estrategias de recorte

LangChain ofrece diferentes estrategias de recorte para adaptarse a distintos casos de uso:

Estrategia "last" - Mantiene los mensajes más recientes:

trimmer_last = trim_messages(
    max_tokens=80,
    strategy="last",
    token_counter=model,
    include_system=True
)

messages = [
    SystemMessage(content="Eres un asistente útil especializado en matemáticas"),
    HumanMessage(content="Hola, soy María"),
    AIMessage(content="¡Hola María! ¿En qué puedo ayudarte hoy?"),
    HumanMessage(content="¿Cuánto es 15 + 27?"),
    AIMessage(content="15 + 27 = 42"),
    HumanMessage(content="¿Y 42 dividido entre 6?"),
    AIMessage(content="42 ÷ 6 = 7"),
    HumanMessage(content="Perfecto, gracias por tu ayuda")
]

# Aplicamos el recorte
trimmed_messages = trimmer_last.invoke(messages)
print(f"Mensajes originales: {len(messages)}")
print(f"Mensajes después del recorte: {len(trimmed_messages)}")

Estrategia "first" - Mantiene los mensajes más antiguos:

trimmer_first = trim_messages(
    max_tokens=80,
    strategy="first",
    token_counter=model,
    include_system=True
)

# Usando los mismos mensajes del ejemplo anterior
trimmed_first = trimmer_first.invoke(messages)

Parámetros avanzados de configuración

La función trim_messages incluye varios parámetros que nos permiten personalizar el comportamiento del recorte:

Control de mensajes del sistema:

# Siempre incluir el mensaje del sistema
trimmer_with_system = trim_messages(
    max_tokens=60,
    strategy="last",
    token_counter=model,
    include_system=True  # El SystemMessage siempre se mantiene
)

# Excluir el mensaje del sistema del recorte
trimmer_without_system = trim_messages(
    max_tokens=60,
    strategy="last",
    token_counter=model,
    include_system=False  # El SystemMessage puede ser recortado
)

Configuración de punto de inicio:

# Comenzar el recorte desde mensajes humanos
trimmer_start_human = trim_messages(
    max_tokens=70,
    strategy="last",
    token_counter=model,
    start_on="human"  # Asegura que el primer mensaje sea de tipo HumanMessage
)

# Comenzar desde mensajes de IA
trimmer_start_ai = trim_messages(
    max_tokens=70,
    strategy="last",
    token_counter=model,
    start_on="ai"  # Asegura que el primer mensaje sea de tipo AIMessage
)

Manejo de mensajes parciales

El parámetro allow_partial controla si permitimos que los mensajes se corten parcialmente cuando exceden el límite de tokens:

# No permitir mensajes parciales (comportamiento por defecto)
trimmer_no_partial = trim_messages(
    max_tokens=50,
    strategy="last",
    token_counter=model,
    allow_partial=False  # Si un mensaje no cabe completo, se excluye
)

# Permitir mensajes parciales
trimmer_partial = trim_messages(
    max_tokens=50,
    strategy="last",
    token_counter=model,
    allow_partial=True  # Los mensajes pueden cortarse para ajustarse al límite
)

# Ejemplo con mensaje largo
long_messages = [
    SystemMessage(content="Eres un asistente útil"),
    HumanMessage(content="Explícame detalladamente cómo funciona el algoritmo de ordenamiento burbuja, incluyendo su complejidad temporal y espacial, así como ejemplos prácticos de implementación"),
    AIMessage(content="El algoritmo burbuja es un método de ordenamiento simple...")
]

# Comparamos los resultados
result_no_partial = trimmer_no_partial.invoke(long_messages)
result_partial = trimmer_partial.invoke(long_messages)

Integración con cadenas LCEL

Una de las ventajas principales de trim_messages es su integración perfecta con las cadenas LCEL. Podemos incorporar el recorte directamente en nuestro flujo de procesamiento:

from langchain_core.output_parsers import StrOutputParser

# Creamos una cadena que incluye recorte automático
chain = (
    trim_messages(
        max_tokens=150,
        strategy="last",
        token_counter=model,
        include_system=True,
        start_on="human"
    )
    | model
    | StrOutputParser()
)

# Usamos la cadena con una conversación larga
conversation = [
    SystemMessage(content="Eres un experto en programación Python"),
    HumanMessage(content="¿Qué es una lista en Python?"),
    AIMessage(content="Una lista es una estructura de datos mutable que puede contener elementos de diferentes tipos"),
    HumanMessage(content="¿Cómo agrego elementos?"),
    AIMessage(content="Puedes usar el método append() para agregar elementos al final de la lista"),
    HumanMessage(content="¿Y cómo los elimino?"),
    AIMessage(content="Puedes usar remove() para eliminar por valor o pop() para eliminar por índice"),
    HumanMessage(content="Dame un ejemplo práctico de uso de listas")
]

# La cadena automáticamente recorta los mensajes antes de enviarlos al modelo
response = chain.invoke(conversation)
print(response)

Monitoreo del recorte

Para supervisar el comportamiento del recorte, podemos implementar un sistema de logging que nos muestre qué mensajes se están eliminando:

def log_trimming(original_messages, trimmed_messages):
    """Función para registrar información sobre el recorte"""
    original_count = len(original_messages)
    trimmed_count = len(trimmed_messages)
    
    if original_count > trimmed_count:
        removed_count = original_count - trimmed_count
        print(f"🔄 Recorte aplicado: {removed_count} mensajes eliminados")
        print(f"📊 Mensajes: {original_count} → {trimmed_count}")
    else:
        print("✅ No se requirió recorte")

# Ejemplo de uso con logging
messages = [
    SystemMessage(content="Eres un asistente útil"),
    HumanMessage(content="Hola"),
    AIMessage(content="¡Hola! ¿Cómo puedo ayudarte?"),
    HumanMessage(content="¿Qué tiempo hace?"),
    AIMessage(content="No tengo acceso a información meteorológica en tiempo real"),
    HumanMessage(content="Entiendo, gracias")
]

trimmer = trim_messages(
    max_tokens=40,
    strategy="last",
    token_counter=model,
    include_system=True
)

trimmed = trimmer.invoke(messages)
log_trimming(messages, trimmed)

Esta funcionalidad de recorte automático es especialmente útil en aplicaciones de chat donde las conversaciones pueden extenderse indefinidamente, permitiendo mantener un rendimiento óptimo sin perder el contexto más relevante de la interacción.