Short-term Memory: Conversation History

Descripción de la cápsula

Short-term memory es el historial de conversación — los mensajes intercambiados en la sesión actual. Sin ella, cada invoke() es una isla: el agente no sabe qué pasó hace 30 segundos. Con ella, el agente mantiene contexto entre turnos y puede referirse a lo que se dijo antes.

Parece simple. No lo es. El historial crece con cada turno, y los modelos tienen context windows finitos. GPT-4.1 tiene 128K tokens — suena enorme, pero una investigación de 30 turnos con tool calls y resultados extensos puede saturarlo. Si no gestionas el crecimiento del historial, tu agente va a fallar en el peor momento: cuando la conversación se pone interesante.

Esta cápsula te da las herramientas completas: MessagesState para gestionar el historial automáticamente, thread_id para aislar conversaciones por usuario, message trimming para controlar el crecimiento, y summarization para preservar contexto sin desperdiciar tokens.

El problema: invoke() sin contexto

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

model = ChatOpenAI(model="gpt-4o-mini")

def chatbot(state: MessagesState) -> dict:
    response = model.invoke(state["messages"])
    return {"messages": [response]}

builder = StateGraph(MessagesState)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
graph = builder.compile()

result1 = graph.invoke({"messages": [{"role": "user", "content": "Me llamo Carlos y trabajo en IA"}]})
print(f"Turno 1: {result1['messages'][-1].content}")

result2 = graph.invoke({"messages": [{"role": "user", "content": "¿Cómo me llamo?"}]})
print(f"Turno 2: {result2['messages'][-1].content}")
# Output esperado:
# Turno 1: ¡Hola Carlos! Qué interesante que trabajes en IA...
# Turno 2: No tengo información sobre tu nombre...

El agente no sabe cómo se llama el usuario. Cada invoke() recibe solo los mensajes que tú le pasas explícitamente. No hay historial acumulado. Es como tener una conversación donde la otra persona se olvida de todo cada vez que parpadea.

MessagesState: estado que gestiona mensajes

MessagesState es un estado pre-built de LangGraph que maneja el historial de mensajes automáticamente:

from langgraph.graph import MessagesState

# MessagesState es equivalente a:
# class MessagesState(TypedDict):
#     messages: Annotated[list[AnyMessage], add_messages]

El detalle clave es el reducer add_messages. Cuando tu nodo retorna {"messages": [new_message]}, el reducer no reemplaza la lista — agrega el nuevo mensaje a los existentes. Cada turno acumula mensajes.

Cómo funciona add_messages

from langchain_core.messages import HumanMessage, AIMessage
from langgraph.graph import add_messages

existing = [
    HumanMessage(content="¿Qué es RAG?"),
    AIMessage(content="RAG es Retrieval-Augmented Generation...")
]
new = [HumanMessage(content="Dame un ejemplo")]

result = add_messages(existing, new)
print(f"Mensajes después del reducer: {len(result)}")
for msg in result:
    print(f"  {msg.type}: {msg.content[:50]}...")
# Output esperado:
# Mensajes después del reducer: 3
#   human: ¿Qué es RAG?...
#   ai: RAG es Retrieval-Augmented Generation......
#   human: Dame un ejemplo...

add_messages también maneja deduplicación por ID: si agregas un mensaje con el mismo id que uno existente, lo actualiza en vez de duplicarlo.

Multi-turn conversation con thread_id

El thread_id es lo que conecta invocaciones separadas. Cada thread es una conversación independiente con su propio historial.

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

model = ChatOpenAI(model="gpt-4o-mini")

def chatbot(state: MessagesState) -> dict:
    response = model.invoke(state["messages"])
    return {"messages": [response]}

builder = StateGraph(MessagesState)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "conversacion_001"}}

result1 = graph.invoke(
    {"messages": [{"role": "user", "content": "Me llamo Carlos y trabajo en IA"}]}, config
)
print(f"Turno 1: {result1['messages'][-1].content}")

result2 = graph.invoke(
    {"messages": [{"role": "user", "content": "¿Cómo me llamo?"}]}, config
)
print(f"Turno 2: {result2['messages'][-1].content}")
print(f"Mensajes totales: {len(result2['messages'])}")
# Output esperado:
# Turno 1: ¡Hola Carlos! Qué interesante que trabajes en IA...
# Turno 2: Te llamas Carlos, y trabajas en IA.
# Mensajes totales: 4

Dos cambios respecto al ejemplo sin memoria: MemorySaver() como checkpointer y config con thread_id. El agente ahora acumula mensajes entre invocaciones.

Threads como aislamiento: multi-usuario

Cada thread_id es una conversación aislada. Usando el mismo grafo compilado del ejemplo anterior:

config_alice = {"configurable": {"thread_id": "alice_thread"}}
config_bob = {"configurable": {"thread_id": "bob_thread"}}

graph.invoke({"messages": [{"role": "user", "content": "Soy Alice, me interesa NLP"}]}, config_alice)
graph.invoke({"messages": [{"role": "user", "content": "Soy Bob, me interesa computer vision"}]}, config_bob)

result_alice = graph.invoke({"messages": [{"role": "user", "content": "¿Qué me interesa?"}]}, config_alice)
result_bob = graph.invoke({"messages": [{"role": "user", "content": "¿Qué me interesa?"}]}, config_bob)

print(f"Alice: {result_alice['messages'][-1].content}")
print(f"Bob: {result_bob['messages'][-1].content}")
# Output esperado:
# Alice: Te interesa NLP...
# Bob: Te interesa computer vision...

Alice y Bob usan el mismo grafo. Sus conversaciones son completamente independientes. Multi-tenancy con una línea de configuración.

El problema del historial infinito

El context window de un modelo es finito. GPT-4.1 tiene 128K tokens. Suena como mucho. No lo es.

Un turno típico de investigación:
  - Mensaje del usuario: ~50 tokens
  - Tool call del agente: ~100 tokens
  - Resultado de tool (búsqueda web): ~500-2000 tokens
  - Respuesta del agente: ~300 tokens
  Total por turno: ~1,000-2,500 tokens

Con 128K tokens de context window:
  128,000 / 2,500 = ~50 turnos máximo teórico
  Límite práctico (calidad + generación): ~30-40 turnos

30 turnos parece mucho para un chat casual. Para un agente de investigación con múltiples tools por turno, puedes llegar al límite en 10-15 interacciones. Cuando lo haces, el modelo falla con "context length exceeded" — o peor, ignora mensajes antiguos silenciosamente.

Message trimming: mantener solo lo relevante

Trimming es la estrategia más directa: mantén los últimos N mensajes y descarta el resto.

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

messages = [
    SystemMessage(content="Eres un asistente de investigación experto en IA."),
    HumanMessage(content="¿Qué es RAG?"),
    AIMessage(content="RAG es Retrieval-Augmented Generation, un patrón que combina..."),
    HumanMessage(content="¿Cómo se implementa?"),
    AIMessage(content="Se implementa con un retriever que busca documentos relevantes..."),
    HumanMessage(content="¿Qué bases de datos vectoriales recomiendas?"),
    AIMessage(content="Las principales opciones son Pinecone, Weaviate, Chroma..."),
    HumanMessage(content="¿Cuál es más fácil de empezar?"),
    AIMessage(content="Chroma es la más fácil para comenzar porque..."),
    HumanMessage(content="Dame un ejemplo de código con Chroma"),
]

trimmed = trim_messages(
    messages,
    max_tokens=200,
    token_counter=len,
    strategy="last",
    include_system=True,
    start_on="human",
)

print(f"Originales: {len(messages)} → Trimmed: {len(trimmed)}")
for msg in trimmed:
    print(f"  {msg.type}: {msg.content[:60]}...")
# Output esperado:
# Originales: 10 → Trimmed: 3
#   system: Eres un asistente de investigación experto en IA....
#   human: Dame un ejemplo de código con Chroma...

Parámetros clave de trim_messages

Integrarlo en el grafo

La forma correcta es hacer trimming dentro del nodo, antes de llamar al modelo:

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import trim_messages, SystemMessage
from dotenv import load_dotenv

load_dotenv()

model = ChatOpenAI(model="gpt-4o-mini")
SYSTEM_PROMPT = SystemMessage(content="Eres un asistente de investigación experto en IA.")

def chatbot(state: MessagesState) -> dict:
    trimmed = trim_messages(
        state["messages"],
        max_tokens=4000,
        token_counter=model.get_num_tokens_from_messages,
        strategy="last",
        include_system=True,
        start_on="human",
    )
    if not any(isinstance(m, SystemMessage) for m in trimmed):
        trimmed = [SYSTEM_PROMPT] + trimmed

    response = model.invoke(trimmed)
    return {"messages": [response]}

builder = StateGraph(MessagesState)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "trimming_demo"}}

for i in range(1, 6):
    result = graph.invoke(
        {"messages": [{"role": "user", "content": f"Turno {i}: cuéntame sobre el tema {i} de IA"}]}, config
    )
    print(f"Turno {i}: {len(result['messages'])} mensajes en historial")
# Output esperado:
# Turno 1: 2 mensajes en historial
# Turno 2: 4 mensajes en historial
# Turno 3: 6 mensajes en historial
# Turno 4: 8 mensajes en historial
# Turno 5: 10 mensajes en historial

El historial completo se guarda en el state (gracias al checkpointer). Pero el modelo solo ve los mensajes que caben en el límite de tokens. Lo mejor de ambos mundos: historial completo para auditoría, ventana limitada para el modelo.

Summarization: comprimir en vez de descartar

Trimming es lossy: los mensajes descartados desaparecen. Si el usuario mencionó algo importante en el turno 3 de 50, el agente lo pierde al hacer trim. Summarization resuelve esto: comprime los mensajes antiguos en un resumen.

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, RemoveMessage
from typing import Annotated, TypedDict
from langgraph.graph import add_messages
from dotenv import load_dotenv

load_dotenv()

model = ChatOpenAI(model="gpt-4o-mini")


class ConversationState(TypedDict):
    messages: Annotated[list, add_messages]
    summary: str


def chatbot(state: ConversationState) -> dict:
    msgs = state["messages"]
    if state.get("summary"):
        msgs = [SystemMessage(content=f"Resumen de la conversación anterior: {state['summary']}")] + msgs
    return {"messages": [model.invoke(msgs)]}


def should_summarize(state: ConversationState) -> str:
    return "summarize" if len(state["messages"]) > 10 else "end"


def summarize_conversation(state: ConversationState) -> dict:
    to_summarize = state["messages"][:-2]
    prompt = f"Resume la conversación en 2-3 oraciones, capturando datos clave del usuario.\n"
    prompt += f"Resumen previo: {state.get('summary', 'Ninguno')}\n\nConversación:\n"
    for msg in to_summarize:
        role = "Usuario" if isinstance(msg, HumanMessage) else "Asistente"
        prompt += f"{role}: {msg.content}\n"

    response = model.invoke([HumanMessage(content=prompt)])
    delete_messages = [RemoveMessage(id=m.id) for m in to_summarize]
    return {"summary": response.content, "messages": delete_messages}


builder = StateGraph(ConversationState)
builder.add_node("chatbot", chatbot)
builder.add_node("summarize", summarize_conversation)
builder.add_edge(START, "chatbot")
builder.add_conditional_edges("chatbot", should_summarize, {"summarize": "summarize", "end": END})
builder.add_edge("summarize", END)

graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "summary_demo"}}

conversations = [
    "Me llamo Carlos y soy investigador en IA",
    "Mi área es NLP, especialmente transformers",
    "Estoy trabajando en un paper sobre attention mechanisms",
    "El paper compara multi-head attention con linear attention",
    "Los resultados favorecen linear attention para secuencias largas",
    "Necesito más datos de benchmark para las conclusiones",
]

for i, msg in enumerate(conversations):
    result = graph.invoke({"messages": [{"role": "user", "content": msg}]}, config)
    summary = result.get("summary", "")
    preview = summary[:80] + "..." if summary else "Sin resumen"
    print(f"Turno {i+1}: {len(result['messages'])} msgs | Resumen: {preview}")
# Output esperado:
# Turno 1: 2 msgs | Resumen: Sin resumen
# ...
# Turno 6: 4 msgs | Resumen: Carlos es investigador en IA especializado en NLP...

El flujo de summarization

Turno N (mensajes <= 10):
  user message → chatbot → should_summarize → "end" → END

Turno N (mensajes > 10):
  user message → chatbot → should_summarize → "summarize"
                                                  ↓
                                          Resume msgs antiguos en 2-3 oraciones
                                          Elimina msgs resumidos con RemoveMessage
                                          Guarda resumen en state["summary"]
                                                  ↓
                                                 END

Los mensajes recientes se mantienen intactos (los últimos 2). Los antiguos se comprimen en un resumen que se inyecta como SystemMessage. El agente tiene contexto de toda la conversación, pero solo paga tokens por el resumen + mensajes recientes.

Trimming vs summarization: cuándo usar cada uno

Regla de decisión: Empieza con trimming — cubre el 80% de los casos. Si los usuarios se quejan de que el agente "olvidó" algo importante de turnos antiguos, migra a summarization.

Token counting: saber cuándo te acercas al límite

No esperes a que el modelo falle con "context length exceeded." Monitorea proactivamente con model.get_num_tokens_from_messages(messages) y compara contra el context window. Integra el monitoring en el nodo del chatbot para que aplique trim automáticamente al superar un umbral (por ejemplo, 75% del context window):

token_count = model.get_num_tokens_from_messages(state["messages"])
if token_count > CONTEXT_WINDOW * 0.75:
    messages = trim_messages(...)  # Trim automático

Esto previene errores y mantiene los costos bajo control.

Troubleshooting

Problema 1: "El agente no recuerda el turno anterior"

Síntoma: Cada invoke() parece empezar de cero. Causa: Falta el checkpointer o el thread_id. Solución: Verifica que compilaste con checkpointer=MemorySaver() y que cada invoke incluye config = {"configurable": {"thread_id": "..."}}.

Problema 2: "context length exceeded" después de muchos turnos

Síntoma: Error del modelo al superar el context window. Causa: No hay trimming — el historial crece sin límite. Solución: Implementa trim_messages dentro del nodo, antes de invocar al modelo. Usa max_tokens al 80% del context window.

Problema 3: "El system prompt desaparece después del trim"

Síntoma: El agente pierde su personalidad o instrucciones. Causa: trim_messages eliminó el SystemMessage. Solución: Usa include_system=True. Si aún se pierde, re-inyéctalo:

if not any(isinstance(m, SystemMessage) for m in trimmed):
    trimmed = [SYSTEM_PROMPT] + trimmed

Problema 4: "Los threads se mezclan"

Síntoma: Un usuario ve mensajes de otro. Causa/Solución: Mismo thread_id para diferentes usuarios. Genera IDs únicos: f"user_{user_id}_conv_{conversation_id}".

Ejercicios

Ejercicio 1: Chatbot multi-turno básico (Fácil)

Crea un chatbot con MessagesState y MemorySaver que mantenga conversación. Envía 3 mensajes: tu nombre, tu profesión, y "¿Qué sabes de mí?". Verifica que recuerda ambos datos.

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

model = ChatOpenAI(model="gpt-4o-mini")

def chatbot(state: MessagesState) -> dict:
    return {"messages": [model.invoke(state["messages"])]}

builder = StateGraph(MessagesState)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "ejercicio_1"}}

r1 = graph.invoke({"messages": [{"role": "user", "content": "Me llamo Ana"}]}, config)
print(f"T1: {r1['messages'][-1].content[:80]}")

r2 = graph.invoke({"messages": [{"role": "user", "content": "Soy ingeniera de datos"}]}, config)
print(f"T2: {r2['messages'][-1].content[:80]}")

r3 = graph.invoke({"messages": [{"role": "user", "content": "¿Qué sabes de mí?"}]}, config)
print(f"T3: {r3['messages'][-1].content}")
print(f"Mensajes totales: {len(r3['messages'])}")
# Output esperado:
# T1: ¡Hola Ana! Encantado de conocerte...
# T2: ¡Qué interesante! La ingeniería de datos es...
# T3: Sé que te llamas Ana y eres ingeniera de datos.
# Mensajes totales: 6

assert len(r3["messages"]) == 6
print("✅ El agente mantiene historial correctamente")

Ejercicio 2: Threads aislados para dos usuarios (Fácil)

Crea dos threads con IDs diferentes. Cada "usuario" dice su nombre y tema de interés. Pregunta a cada uno "¿Qué me interesa?" y verifica que las respuestas no se mezclan.

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

model = ChatOpenAI(model="gpt-4o-mini")

def chatbot(state: MessagesState) -> dict:
    return {"messages": [model.invoke(state["messages"])]}

builder = StateGraph(MessagesState)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
graph = builder.compile(checkpointer=MemorySaver())

config_a = {"configurable": {"thread_id": "user_maria"}}
config_b = {"configurable": {"thread_id": "user_pedro"}}

graph.invoke({"messages": [{"role": "user", "content": "Soy María, me interesa MLOps"}]}, config_a)
graph.invoke({"messages": [{"role": "user", "content": "Soy Pedro, me interesa cybersecurity"}]}, config_b)

result_a = graph.invoke({"messages": [{"role": "user", "content": "¿Qué me interesa?"}]}, config_a)
result_b = graph.invoke({"messages": [{"role": "user", "content": "¿Qué me interesa?"}]}, config_b)

print(f"María: {result_a['messages'][-1].content}")
print(f"Pedro: {result_b['messages'][-1].content}")
# Output esperado:
# María: Te interesa MLOps...
# Pedro: Te interesa cybersecurity...

print("✅ Los threads están correctamente aislados")

Ejercicio 3: Trimming con alerta automática (Medio)

Implementa un chatbot que cuente tokens en cada turno y aplique trim_messages automáticamente cuando supere 500 tokens. Loggea cuántos mensajes se eliminaron en cada trim.

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import trim_messages, SystemMessage
from dotenv import load_dotenv

load_dotenv()

model = ChatOpenAI(model="gpt-4o-mini")
TOKEN_LIMIT = 500

def chatbot_with_trim(state: MessagesState) -> dict:
    messages = state["messages"]
    token_count = model.get_num_tokens_from_messages(messages)
    print(f"  [MONITOR] {token_count} tokens, {len(messages)} mensajes")

    if token_count > TOKEN_LIMIT:
        original_count = len(messages)
        messages = trim_messages(
            messages,
            max_tokens=TOKEN_LIMIT // 2,
            token_counter=model.get_num_tokens_from_messages,
            strategy="last",
            include_system=True,
            start_on="human",
        )
        print(f"  [TRIM] ⚠️ {original_count}→{len(messages)} mensajes, {token_count}→{model.get_num_tokens_from_messages(messages)} tokens")

    return {"messages": [model.invoke(messages)]}

builder = StateGraph(MessagesState)
builder.add_node("chatbot", chatbot_with_trim)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "trim_alert"}}

for i, topic in enumerate(["Explícame ML en detalle", "¿Y deep learning?", "¿Cuál es la diferencia?",
                           "Dame ejemplos prácticos", "¿Qué frameworks recomiendas?",
                           "Compara PyTorch vs TensorFlow", "¿Cuál para principiantes?"]):
    print(f"Turno {i+1}: '{topic}'")
    graph.invoke({"messages": [{"role": "user", "content": topic}]}, config)
print("\n✅ Trim automático funcionando")

Ejercicio 4: Summarization con verificación (Medio)

Implementa un grafo con summarization que resuma al superar 6 mensajes. Envía datos específicos en los primeros turnos (nombre, empresa, ciudad). Después de la summarización, verifica que el resumen preserva esos datos.

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, RemoveMessage
from typing import Annotated, TypedDict
from langgraph.graph import add_messages
from dotenv import load_dotenv

load_dotenv()

model = ChatOpenAI(model="gpt-4o-mini")


class SumState(TypedDict):
    messages: Annotated[list, add_messages]
    summary: str


def chatbot(state: SumState) -> dict:
    msgs = state["messages"]
    if state.get("summary"):
        msgs = [SystemMessage(content=f"Resumen previo: {state['summary']}")] + msgs
    return {"messages": [model.invoke(msgs)]}


def should_summarize(state: SumState) -> str:
    return "summarize" if len(state["messages"]) > 6 else "end"


def summarize(state: SumState) -> dict:
    old = state["messages"][:-2]
    prompt = "Resume en 2-3 oraciones. Incluye TODOS los datos específicos (nombres, empresas, ciudades):\n"
    for m in old:
        role = "Usuario" if isinstance(m, HumanMessage) else "Asistente"
        prompt += f"{role}: {m.content}\n"
    resp = model.invoke([HumanMessage(content=prompt)])
    return {"summary": resp.content, "messages": [RemoveMessage(id=m.id) for m in old]}


builder = StateGraph(SumState)
builder.add_node("chatbot", chatbot)
builder.add_node("summarize", summarize)
builder.add_edge(START, "chatbot")
builder.add_conditional_edges("chatbot", should_summarize, {"summarize": "summarize", "end": END})
builder.add_edge("summarize", END)
graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "sum_verify"}}

graph.invoke({"messages": [{"role": "user", "content": "Me llamo Diana y vivo en Bogotá"}]}, config)
graph.invoke({"messages": [{"role": "user", "content": "Trabajo en fintech, mi empresa es PayLatam"}]}, config)
graph.invoke({"messages": [{"role": "user", "content": "Usamos Python y FastAPI"}]}, config)
graph.invoke({"messages": [{"role": "user", "content": "Nuestro producto principal es pagos móviles"}]}, config)

result = graph.invoke(
    {"messages": [{"role": "user", "content": "¿Dónde vivo y cómo se llama mi empresa?"}]}, config
)
print(f"Respuesta: {result['messages'][-1].content}")
print(f"Resumen: {result.get('summary', 'Ninguno')[:120]}...")
# Output esperado:
# Respuesta: Vives en Bogotá y tu empresa se llama PayLatam...
# Resumen: Diana vive en Bogotá, trabaja en fintech en PayLatam usando Python y FastAPI...

print("✅ Summarization preserva datos de turnos anteriores")

Ejercicio 5: Comparar trimming vs summarization (Medio-Avanzado)

Crea dos grafos — uno con trimming agresivo y otro con summarization. Envía un dato específico en el mensaje 1 ("mi código es ALPHA-7742"), seguido de 7 mensajes sobre otros temas. En el mensaje 9, pregunta por el código. Compara cuál de los dos lo recuerda.

from langgraph.graph import StateGraph, START, END, MessagesState, add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import trim_messages, SystemMessage, HumanMessage, RemoveMessage
from typing import Annotated, TypedDict
from dotenv import load_dotenv

load_dotenv()

model = ChatOpenAI(model="gpt-4o-mini")

# Graph 1: Trimming agresivo (300 tokens max)
def chatbot_trim(state: MessagesState) -> dict:
    messages = trim_messages(
        state["messages"], max_tokens=300,
        token_counter=model.get_num_tokens_from_messages,
        strategy="last", include_system=True, start_on="human",
    )
    return {"messages": [model.invoke(messages)]}

b1 = StateGraph(MessagesState)
b1.add_node("chatbot", chatbot_trim)
b1.add_edge(START, "chatbot")
b1.add_edge("chatbot", END)
graph_trim = b1.compile(checkpointer=MemorySaver())

# Graph 2: Summarization (threshold 6 msgs)
class SumState(TypedDict):
    messages: Annotated[list, add_messages]
    summary: str

def chatbot_sum(state: SumState) -> dict:
    msgs = state["messages"]
    if state.get("summary"):
        msgs = [SystemMessage(content=f"Contexto: {state['summary']}")] + msgs
    return {"messages": [model.invoke(msgs)]}

def should_sum(state: SumState) -> str:
    return "summarize" if len(state["messages"]) > 6 else "end"

def do_sum(state: SumState) -> dict:
    old = state["messages"][:-2]
    prompt = "Resume capturando TODOS los datos específicos (códigos, números, nombres):\n"
    for m in old:
        role = "Usuario" if isinstance(m, HumanMessage) else "Asistente"
        prompt += f"{role}: {m.content}\n"
    resp = model.invoke([HumanMessage(content=prompt)])
    return {"summary": resp.content, "messages": [RemoveMessage(id=m.id) for m in old]}

b2 = StateGraph(SumState)
b2.add_node("chatbot", chatbot_sum)
b2.add_node("summarize", do_sum)
b2.add_edge(START, "chatbot")
b2.add_conditional_edges("chatbot", should_sum, {"summarize": "summarize", "end": END})
b2.add_edge("summarize", END)
graph_sum = b2.compile(checkpointer=MemorySaver())

# Test
msgs = [
    "Mi código secreto es ALPHA-7742. Recuérdalo.",
    "Hablemos de Python", "¿FastAPI o Django?", "¿Y Flask?",
    "Hablemos de bases de datos", "¿PostgreSQL o MySQL?",
    "¿Y MongoDB?", "¿Cuál es mi código secreto?",
]

ct = {"configurable": {"thread_id": "cmp_trim"}}
cs = {"configurable": {"thread_id": "cmp_sum"}}

for msg in msgs[:-1]:
    graph_trim.invoke({"messages": [{"role": "user", "content": msg}]}, ct)
    graph_sum.invoke({"messages": [{"role": "user", "content": msg}]}, cs)

rt = graph_trim.invoke({"messages": [{"role": "user", "content": msgs[-1]}]}, ct)
rs = graph_sum.invoke({"messages": [{"role": "user", "content": msgs[-1]}]}, cs)

print(f"TRIMMING: {rt['messages'][-1].content}")
print(f"SUMMARIZATION: {rs['messages'][-1].content}")
print(f"\nTrimming recuerda: {'✅' if '7742' in rt['messages'][-1].content else '❌'}")
print(f"Summarization recuerda: {'✅' if '7742' in rs['messages'][-1].content else '❌'}")
# Output esperado:
# Trimming recuerda: ❌ (mensaje 1 fue descartado)
# Summarization recuerda: ✅ (resumen capturó el dato)

Resumen

En esta cápsula aprendiste:

• Sin short-term memory, cada invoke() es una isla — el agente no sabe qué pasó en el turno anterior. Con MessagesState y un checkpointer, el historial se acumula automáticamente
• MessagesState es un estado pre-built con messages: Annotated[list[AnyMessage], add_messages]. El reducer add_messages agrega mensajes nuevos y maneja deduplicación por ID
• thread_id aísla conversaciones: mismo grafo, diferentes threads, diferentes historiales. Es la base del multi-usuario
• El historial crece sin límite y los context windows son finitos. Sin gestión, tu agente falla cuando la conversación se pone larga
• Message trimming descarta mensajes antiguos. Simple, sin costo, pero lossy
• Summarization comprime mensajes en un resumen. Preserva contexto pero cuesta una llamada LLM extra
• Token monitoring proactivo evita "context length exceeded" — monitorea antes de invocar al modelo, no después de que falle

Próxima cápsula: Checkpointing: MemorySaver — cómo LangGraph guarda el estado del agente automáticamente, cómo inspeccionar checkpoints, y cómo el thread management organiza ejecuciones.

Recursos adicionales

1. LangGraph — How to manage conversation history — Guía oficial para gestionar historial con trimming y summarization
2. LangGraph — MessagesState — Documentación del estado pre-built para mensajes
3. LangChain — trim_messages — Referencia completa de la función trim_messages
4. LangGraph — How to add summary of conversation history — Guía paso a paso para summarization
5. OpenAI — Managing tokens — Cómo contar y gestionar tokens
6. LangGraph — add_messages reducer — Documentación del reducer de mensajes

Módulo 8 — LangChain & LangGraph: From Chains to Agents

Siguiente cápsula: Checkpointing: MemorySaver — aprenderás cómo LangGraph guarda automáticamente el estado del agente después de cada nodo, cómo inspeccionar checkpoints para ver el estado exacto en cualquier momento, y cómo thread_id organiza el historial de ejecuciones.