Memória e histórico de conversa com LangGraph

Cada pergunta começa do zero

Modelos de linguagem não têm memória entre chamadas. Tudo o que o modelo sabe sobre a conversa é o que está no prompt daquela chamada. Até agora, todas as chains da trilha recebiam uma pergunta e esqueciam tudo depois.

Na prática, isso quebra a conversa mais simples:

Sozinha, "E se eu tiver pago com Pix?" não diz nada: Pix para quê? Sem o histórico, o RAG da Parte 7 busca na documentação por "pago com Pix" e, nos testes do repositório, trouxe um trecho sobre pausar a conta. Com o histórico, a mesma pergunta encontra o trecho de cobrança em duplicidade, que explica que no Pix o valor volta para a conta de origem.

Este tutorial resolve isso em três passos: guardar a conversa, controlar quanto dela vai para o modelo e usar o histórico para melhorar a busca do RAG.

Como o padrão funciona

A ideia é simples: a cada mensagem, recuperar as mensagens anteriores daquela conversa, colocá-las no prompt e, depois da resposta, guardar a pergunta e a resposta para a próxima vez.

No LangChain 1.x, quem faz o "carregar" e o "salvar" é a persistência do LangGraph: um checkpointer guarda o estado do grafo depois de cada execução, separado por um thread_id (um id por conversa).

Se você nunca usou LangGraph, não se preocupe: aqui ele é só uma forma de organizar as etapas que já conhecemos em nós, com um estado compartilhado que pode ser salvo. As chains das partes anteriores continuam iguais, só que agora dentro dos nós.

Um chat com memória

Primeiro, o prompt ganha um lugar para o histórico. O MessagesPlaceholder recebe a lista de mensagens anteriores e as coloca entre as instruções e a pergunta nova:

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder


def create_chat_prompt() -> ChatPromptTemplate:
    return ChatPromptTemplate.from_messages(
        [
            ("system", "You are the support assistant of Nuvem. Answer in Portuguese."),
            MessagesPlaceholder("history"),
            ("human", "{question}"),
        ]
    )

Depois, o grafo. O estado é um MessagesState: um dicionário com uma lista messages em que cada nó acrescenta mensagens, em vez de substituir a lista. O grafo tem um nó só, que separa o histórico da pergunta nova e chama a chain:

from langchain_core.messages import AIMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph


def create_chat_graph(llm=None, max_messages=10, checkpointer=None):
    llm = llm or _default_llm()
    chain = create_chat_prompt() | llm | StrOutputParser()

    def respond(state: MessagesState) -> dict:
        history, question = state["messages"][:-1], state["messages"][-1].content
        answer = chain.invoke({"history": trim_history(history, max_messages), "question": question})
        return {"messages": [AIMessage(answer)]}  # acrescentada ao estado

    graph = StateGraph(MessagesState)
    graph.add_node("respond", respond)
    graph.add_edge(START, "respond")
    graph.add_edge("respond", END)
    return graph.compile(checkpointer=checkpointer or InMemorySaver())

O compile(checkpointer=...) é o que liga a memória. Na chamada, você manda só a mensagem nova e o thread_id; o checkpointer traz o resto:

from langchain_core.messages import HumanMessage

chat = create_chat_graph()
config = {"configurable": {"thread_id": "cliente-1"}}

chat.invoke({"messages": [HumanMessage("Meu nome é Ana.")]}, config=config)
state = chat.invoke({"messages": [HumanMessage("Qual é o meu nome?")]}, config=config)

state["messages"][-1].content   # a resposta lembra que o nome é Ana
len(state["messages"])          # 4: duas perguntas e duas respostas

Outro thread_id é outra conversa, com histórico vazio. Nos testes, cinco mensagens na conversa "a" e uma na "b" deixaram 10 e 2 mensagens salvas, respectivamente, sem nada vazar de uma para outra.

Limitando o histórico

Uma conversa longa não pode ir inteira para o modelo: cada mensagem custa tokens, e todo modelo tem um limite de contexto. A solução é separar duas coisas:

• O que fica salvo: a conversa inteira, no checkpointer.
• O que vai para o prompt: só as últimas mensagens.

from langchain_core.messages import trim_messages


def trim_history(messages, max_messages=10):
    return trim_messages(
        messages,
        max_tokens=max_messages,
        token_counter=len,       # conta mensagens em vez de tokens
        strategy="last",         # mantém as mais recentes
        start_on="human",        # o histórico nunca começa com uma resposta solta
    )

Com token_counter=len, o "limite de tokens" vira um limite de mensagens, que é mais fácil de entender. Para controlar por tokens de verdade, passe o próprio modelo como token_counter. O start_on="human" evita mandar ao modelo um histórico que começa no meio de uma troca, com uma resposta sem a pergunta.

Nos testes, com max_messages=4, a quinta mensagem da conversa tinha 8 mensagens anteriores salvas, mas o modelo recebeu exatamente 4 (human, ai, human, ai). O estado salvo continuou com tudo.

RAG conversacional

Colocar o histórico no prompt do RAG não basta. O problema está antes, na busca: o retriever recebe "E se eu tiver pago com Pix?" e busca exatamente isso, sem saber do que a conversa trata.

A solução é uma etapa a mais: reescrever a pergunta de acompanhamento como uma pergunta completa, usando o histórico, e buscar com a versão reescrita.

O prompt de reescrita pede explicitamente para não responder, só reformular:

def create_condense_prompt() -> ChatPromptTemplate:
    return ChatPromptTemplate.from_messages(
        [
            (
                "system",
                "Given the conversation and a follow-up question, rewrite the "
                "follow-up as a standalone question in Portuguese that can be "
                "understood without the conversation. Do not answer it. "
                "Return only the rewritten question.",
            ),
            MessagesPlaceholder("history"),
            ("human", "{question}"),
        ]
    )

O grafo tem três nós. O estado ganha dois campos além das mensagens, standalone_question e docs, para cada etapa ver o que a anterior produziu (e para você inspecionar depois):

class ConversationalRAGState(MessagesState):
    standalone_question: str
    docs: list[Document]


def create_conversational_rag_graph(llm=None, retriever=None, max_messages=10, checkpointer=None):
    ...
    def condense(state):
        history, question = _split(state["messages"])
        if not history:                       # primeira mensagem: nada a reescrever
            return {"standalone_question": question}
        rewritten = condense_chain.invoke(
            {"history": trim_history(history, max_messages), "question": question}
        )
        return {"standalone_question": rewritten}

    def retrieve(state):
        return {"docs": retriever.invoke(state["standalone_question"])}

    def answer(state):
        history, question = _split(state["messages"])
        text = answer_chain.invoke({
            "history": trim_history(history, max_messages),
            "question": question,
            "context": format_docs(state["docs"]),
        })
        return {"messages": [AIMessage(text)]}

    graph = StateGraph(ConversationalRAGState)
    graph.add_node("condense", condense)
    graph.add_node("retrieve", retrieve)
    graph.add_node("answer", answer)
    graph.add_edge(START, "condense")
    graph.add_edge("condense", "retrieve")
    graph.add_edge("retrieve", "answer")
    graph.add_edge("answer", END)
    return graph.compile(checkpointer=checkpointer or InMemorySaver())

Dois detalhes que valem a pena:

• A reescrita é pulada na primeira mensagem. Sem histórico, não há o que reformular, e você economiza uma chamada ao modelo. Nos testes, a primeira pergunta gerou 1 chamada (a resposta) e a de acompanhamento gerou 2 (reescrita + resposta).
• A resposta usa a pergunta original, não a reescrita. A reescrita existe para a busca; para responder, o modelo já tem o histórico no prompt e a pergunta como o cliente escreveu.

Os retriever, format_docs e o "não encontrei" vêm direto da Parte 7. Memória não substitui o RAG; ela melhora a pergunta que chega até ele.

Rodando o exemplo

CONVERSATION = [
    "Fui cobrado duas vezes no mesmo mês. Em quanto tempo recebo o estorno?",
    "E se eu tiver pago com Pix?",          # só faz sentido com o histórico
    "Qual foi mesmo o prazo que você falou?",
]


def ask(graph, thread_id, question):
    state = graph.invoke(
        {"messages": [HumanMessage(question)]},
        config={"configurable": {"thread_id": thread_id}},
    )
    print(f"=== [{thread_id}] {question}")
    print(f"(pergunta reescrita: {state['standalone_question']})")
    print(state["messages"][-1].content)


graph = create_conversational_rag_graph()
for question in CONVERSATION:
    ask(graph, "cliente-1", question)

# Outra conversa, sem histórico: a mesma pergunta fica ambígua
ask(graph, "cliente-2", "E se eu tiver pago com Pix?")

python src/main_memory.py

A linha "pergunta reescrita" mostra o que foi para a busca. É a melhor ferramenta para depurar um RAG conversacional: se a resposta saiu errada, comece verificando se a reescrita entendeu a pergunta.

Em produção

• Troque o InMemorySaver. Ele guarda tudo na memória do processo: reiniciou a aplicação, as conversas somem, e duas instâncias da API não compartilham nada. A documentação do LangChain recomenda um checkpointer com banco de dados, como o PostgresSaver (pacote langgraph-checkpoint-postgres). O grafo não muda: só o objeto passado em checkpointer=.
• Defina bem o thread_id. Um por atendimento ou por janela de chat, nunca um por cliente "para sempre": conversas muito antigas confundem mais do que ajudam.
• Resuma em vez de só cortar. O corte por número de mensagens esquece o começo da conversa. Para conversas longas, a alternativa é resumir as mensagens antigas e manter as recentes inteiras; o LangChain tem um SummarizationMiddleware para agentes que faz isso.
• Histórico é dado pessoal. Conversas de suporte têm nome, e-mail, dados de pagamento. Defina por quanto tempo guardar, apague quando o atendimento termina e não use o histórico de um cliente na conversa de outro.
• Cuidado com a reescrita. Ela é mais uma chamada ao modelo, e pode errar. Registre a pergunta reescrita junto com a resposta, como o exemplo faz no estado.

Resumo e referências

• O modelo não lembra nada entre chamadas; memória é colocar o histórico no prompt.
• No LangChain 1.x, a persistência é feita pelo LangGraph: MessagesState + checkpointer + thread_id. O RunnableWithMessageHistory está obsoleto.
• Salve a conversa inteira, mas mande ao modelo só as últimas mensagens (trim_messages).
• No RAG, reescreva a pergunta de acompanhamento antes de buscar; responda com a pergunta original e o histórico.
• Em produção, use um checkpointer com banco e trate o histórico como dado pessoal.

Na Parte 9 (Chain-of-Thought e Reflection), o foco volta para a qualidade da resposta: pedir o raciocínio passo a passo e fazer o modelo revisar o próprio trabalho.

Referências

• Repositório pattern_chain (código completo)
• Documentação do LangChain: short-term memory
• Referência do LangChain: runnables
• Documentação oficial do LangChain