Memória de Conversação

Implemente memória de conversação em LangGraph usando o reducer add_messages, armazenando e gerenciando histórico de conversação entre turnos.

A memória de conversação é o que permite ao chatbot lembrar o que foi dito anteriormente na conversa. O LangGraph trata isso de forma natural por meio do estado persistente e do reducer add_messages.

O Redutor add_messages

O LangGraph fornece um reducer add_messages integrado, projetado especificamente para listas de mensagens de conversação:

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

Diferente do operator.add (que simplesmente concatena), o add_messages é inteligente:

Acrescenta novas mensagens à lista
Atualiza mensagens existentes se uma mensagem com o mesmo ID for adicionada
Preserva a ordem das mensagens

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

O reducer add_messages é preferido ao operator.add para listas de mensagens porque trata corretamente a deduplicação e as atualizações no local.

Adicionando Mensagens ao Estado

Há dois padrões para atualizar mensagens no estado:

Padrão 1: Anexar à 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]}

Padrão 2: Retornar todas as mensagens (também 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

💡Dica

O Padrão 1 é mais eficiente: você retorna apenas as mensagens novas, e não a lista inteira. O reducer cuida da concatenação.

Exemplo Completo de Memória de Conversação

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"

✅Sucesso

Com persistência + add_messages, o histórico de conversação cresce entre invocações. O LLM recebe o histórico completo e mantém o contexto.

Truncando Histórico de Conversação

O histórico de conversação pode crescer além da janela de contexto do LLM. Truncate as mensagens antes de enviar ao 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]}

Estratégia de Truncamento 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

⚠️Aviso

O truncamento remove contexto antigo. O LLM esquecerá as partes iniciais da conversa. Para conversas muito longas, use memória baseada em resumo (Lição 4).

Persistência de Mensagem do Sistema

Mantenha uma mensagem de sistema no início da conversa:

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

Extração Estruturada de Memória

Extraia e armazene informações estruturadas das conversas:

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}

Exemplo Completo: Chatbot com Memória Aprimorada

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!"