Memoria de Conversación

Implementa memoria de conversación en LangGraph usando el reductor add_messages, almacenando y gestionando el historial de conversación entre turnos.

La memoria de conversación es lo que permite al chatbot recordar lo que se dijo antes en la conversa. LangGraph lo maneja de forma natural mediante el estado persistente y el reductor add_messages.

El Reductor add_messages

LangGraph proporciona un reductor add_messages integrado, diseñado específicamente para listas de mensajes de conversación:

python

from langgraph.graph import add_messages
from typing_extensions import TypedDict, Annotated
from typing import List, Any

class ConversationState(TypedDict):
    messages: Annotated[List[Any], add_messages]
    context: str

A diferencia de operator.add (que simplemente concatena), add_messages es inteligente:

Añade mensajes nuevos a la lista
Actualiza los mensajes existentes si se añade un mensaje con el mismo ID
Conserva el orden de los mensajes

python

from langchain_core.messages import AIMessage, HumanMessage, ToolMessage

# add_messages handles these intelligently:
updates = [
    HumanMessage(content="Hello", id="1"),
    AIMessage(content="Hi!", id="2"),
    # If we add a message with the same ID, it updates in place:
    AIMessage(content="Hello there!", id="2")  # Replaces previous AIMessage
]

ℹ️Nota

El reductor add_messages se prefiere a operator.add para las listas de mensajes porque gestiona correctamente la deduplicación y las actualizaciones in situ.

Agregando Mensajes al Estado

Hay dos patrones para actualizar mensajes en el estado:

Patrón 1: Anexar a la lista existente

python

def chat_node(state: ConversationState) -> dict:
    response = llm.invoke(state["messages"])
    # Returns the new message(s) — add_messages handles appending
    return {"messages": [response]}

Patrón 2: Devolver todos los mensajes (también funciona)

python

def chat_node(state: ConversationState) -> dict:
    response = llm.invoke(state["messages"])
    # The reducer concatenates these with existing messages
    return {"messages": state["messages"] + [response]}
    # Both patterns produce the same result with add_messages

💡Consejo

El Patrón 1 es más eficiente: solo devuelves los mensajes nuevos en lugar de la lista completa. El reductor se encarga de la concatenación.

Ejemplo Completo de Memoria de Conversación

python

from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from typing_extensions import TypedDict, Annotated
from typing import List, Any

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

class ChatState(TypedDict):
    messages: Annotated[List[Any], add_messages]
    user_name: str

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

# Build with persistence
checkpointer = MemorySaver()
builder = StateGraph(ChatState)
builder.add_node("chat", chatbot)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
app = builder.compile(checkpointer=checkpointer)

# Multi-turn conversation
session = {"configurable": {"thread_id": "user-1"}}

app.invoke({"messages": [HumanMessage("Hi, my name is Alice")]}, session)
app.invoke({"messages": [HumanMessage("What's my name?")]}, session)
# The bot remembers: "Your name is Alice"

✅Éxito

Con persistencia + add_messages, el historial de conversación crece entre invocaciones. El LLM recibe el historial completo y mantiene el contexto.

Recortando Historial de Conversación

El historial de conversación puede crecer más allá de la ventana de contexto del LLM. Trunca los mensajes antes de enviarlos al LLM:

python

from langchain_core.messages import trim_messages

def chat_with_trimming(state: ChatState) -> dict:
    # Keep last 10 messages or 4000 tokens
    trimmed = trim_messages(
        state["messages"],
        strategy="last",
        max_tokens=4000,
        token_counter=len,  # Use char count as proxy
        start_on="human",   # Always start with a human message
        include_system=True
    )

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

Estrategia de Recorte Personalizada

python

def trim_to_token_limit(messages: list, max_tokens: int = 4000) -> list:
    """Keep messages within token budget, preserving the most recent."""
    import tiktoken
    enc = tiktoken.encoding_for_model("gpt-4o")

    total_tokens = 0
    keep = []

    for msg in reversed(messages):
        msg_tokens = len(enc.encode(msg.content))
        if total_tokens + msg_tokens > max_tokens:
            break
        total_tokens += msg_tokens
        keep.insert(0, msg)

    # Always include the system message if present
    if messages and messages[0].type == "system":
        keep.insert(0, messages[0])

    return keep

⚠️Advertencia

El truncado elimina el contexto antiguo. El LLM olvidará las partes iniciales de la conversa. Para conversaciones muy largas, usa la memoria basada en resúmenes (Lección 4).

Persistencia de Mensaje del Sistema

Mantén un mensaje de sistema al inicio de la conversación:

python

class ChatState(TypedDict):
    messages: Annotated[List[Any], add_messages]

def ensure_system_message(state: ChatState) -> dict:
    """Ensure the system message is always first."""
    messages = state["messages"]
    system_msg = SystemMessage(
        "You are a helpful assistant. The current date is 2024."
    )

    if not messages or messages[0].type != "system":
        return {"messages": [system_msg] + messages}

    return {}  # System message already present

Extracción Estructurada de Memoria

Extrae y guarda información estructurada de las conversaciones:

python

from pydantic import BaseModel, Field

class UserProfile(BaseModel):
    name: str = Field(default="")
    preferences: list = Field(default_factory=list)
    pain_points: list = Field(default_factory=list)

class MemoryState(TypedDict):
    messages: Annotated[List[Any], add_messages]
    profile: UserProfile

def extract_profile(state: MemoryState) -> dict:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Extract user info from the conversation. Update the profile:\n"
                   "{current_profile}"),
        ("human", "{last_message}")
    ])

    parser = PydanticOutputParser(pydantic_object=UserProfile)
    chain = prompt | llm | parser

    last_msg = state["messages"][-1].content
    profile = chain.invoke({
        "current_profile": state["profile"].dict(),
        "last_message": last_msg
    })

    return {"profile": profile}

Ejemplo Completo: Chatbot con Memoria Mejorada

python

from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from typing_extensions import TypedDict, Annotated
from typing import List, Any

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

class State(TypedDict):
    messages: Annotated[List[Any], add_messages]
    summary: str  # For long-term memory

def call_model(state: State) -> dict:
    system_prompt = "You are a helpful assistant."
    if state.get("summary"):
        system_prompt += f"\nConversation summary: {state['summary']}"

    messages = [SystemMessage(system_prompt)] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

checkpointer = MemorySaver()
builder = StateGraph(State)
builder.add_node("model", call_model)
builder.add_edge(START, "model")
builder.add_edge("model", END)
app = builder.compile(checkpointer=checkpointer)

session = {"configurable": {"thread_id": "memory-demo"}}
app.invoke({"messages": [HumanMessage("I love Python!")]}, session)
app.invoke({"messages": [HumanMessage("What do I love?")]}, session)
# → "You said you love Python!"