Phase 11 - Lesson 16

LangGraph — Máquinas de Estado para Agentes

Um loop ReAct escrito à mão é um while True. Um loop ReAct escrito no LangGraph é um gráfico pelo qual você pode criar checkpoints, interromper, ramificar e viajar no tempo. O agente não mudou. A estrutura ao redor dele sim.

Tipo: Construção Linguagens: Python Pré-requisitos: Fase 11 · 09 (Function Calling), Fase 11 · 14 (Model Context Protocol) Tempo: ~75 minutos

O Problema

Você coloca em produção um agente de chamada de função. Ele funciona por três turnos, então algo dá errado: o modelo tenta uma ferramenta que retorna 500, o usuário muda de ideia no meio da tarefa ou o agente decide reembolsar um pedido sem a autorização de um humano. O loop while True: não tem ganchos. Você não pode pausá-lo, não pode retrocedê-lo e não pode ramificá-lo para explorar "e se o modelo tivesse escolhido a outra ferramenta". No momento em que você lança isso além de uma demonstração, o agente se torna uma caixa preta que ou funcionou ou não.

O próximo passo é óbvio assim que você o vê. O agente já é uma máquina de estados — prompt do sistema mais histórico de mensagens mais chamadas de ferramentas pendentes mais a próxima ação. Torne a máquina de estados explícita: nós para "o modelo pensa", "uma ferramenta executa", "um humano aprova" e arestas para as transições condicionais entre eles. Uma vez que o grafo é explícito, a estrutura ganha quatro coisas de graça: checkpointing (salvar o estado entre as etapas), interrupts (pausar para um humano), streaming (transmitir tokens e eventos intermediários) e time-travel (retroceder a um estado anterior e tentar um caminho diferente).

LangGraph é a biblioteca que entrega essa abstração. Não é um framework de agentes no sentido do LangChain ("aqui está um AgentExecutor, boa sorte"). É um runtime de grafos com estado de primeira classe, persistência de primeira classe e interrupções de primeira classe. O loop do agente é algo que você desenha, não algo que você escreve à mão.

O Conceito

Um StateGraph tem três coisas.

1. State (Estado). Um dicionário tipado (TypedDict ou modelo Pydantic) que flui pelo grafo. Cada nó recebe o estado completo e retorna uma atualização parcial, que o LangGraph mescla usando um reducer por campo — operator.add para listas que devem acumular, sobrescrevendo por padrão.
2. Nodes (Nós). Funções Python state -> partial_state. Cada um é uma etapa discreta: "chamar o modelo", "executar ferramentas", "resumir".
3. Edges (Arestas). Transições entre nós. Arestas estáticas vão para um único lugar. Arestas condicionais recebem uma função de roteador state -> next_node_name para que o grafo possa ramificar com base na saída do modelo.

Você compila o grafo. A compilação define a topologia, anexa um checkpointer (opcional, mas essencial para produção) e retorna um executável (runnable). Você o invoca com um estado inicial e um thread_id. Cada etapa da execução persiste um checkpoint indexado por (thread_id, checkpoint_id).

Os quatro superpoderes

Checkpointing (Criação de checkpoints). Cada transição de nó grava o novo estado em um armazenamento (em memória para testes, Postgres/Redis/SQLite para produção). Retome chamando o grafo novamente com o mesmo thread_id. O grafo recomeça exatamente de onde parou.

Interrupts (Interrupções). Marque um nó com interrupt_before=["human_review"] e a execução parará antes que esse nó seja executado. O estado persiste. Sua API responde ao usuário com "aguardando aprovação". Uma requisição posterior para o mesmo thread_id com Command(resume=...) retoma a execução.

Streaming. graph.stream(state, mode="updates") gera deltas de estado conforme eles acontecem. O mode="messages" transmite os tokens do LLM dentro dos nós do modelo. O mode="values" retorna snapshots completos. Você escolhe o que exibir na sua interface.

Time-travel (Viagem no tempo). graph.get_state_history(thread_id) retorna o histórico completo de checkpoints. Passe qualquer checkpoint_id anterior para graph.invoke e você criará uma ramificação a partir desse ponto. Excelente para depuração ("e se o modelo tivesse escolhido a ferramenta B?") e para testes de regressão que reproduzem execuções de produção.

Os reducers são o ponto principal

Cada campo do estado tem um reducer. A maioria dos padrões é suficiente — um novo valor sobrescreve o antigo. Mas listas de mensagens precisam de operator.add para que as novas mensagens sejam anexadas em vez de substituídas. Arestas paralelas mesclam suas atualizações por meio do reducer. Se dois nós atualizarem messages e você esquecer de Annotated[list, add_messages], o segundo vencerá silenciosamente e você perderá metade do turno. O reducer é a única parte sutil da biblioteca; acerte ele e o restante se encaixará.

O grafo ReAct em quatro nós

Um agente ReAct em produção é composto por quatro nós e duas arestas:

1. agent — chama o LLM com o histórico de mensagens atual. Retorna a mensagem do assistente (que pode conter chamadas de ferramentas, tool_calls).
2. tools — executa quaisquer tool_calls na última mensagem do assistente e anexa os resultados das ferramentas como mensagens de ferramenta.
3. Uma aresta condicional a partir de agent que roteia para tools se a última mensagem tiver tool_calls, caso contrário, para END.
4. Uma aresta estática de tools de volta para agent.

Isso é tudo. Você obtém o loop ReAct completo (Pensamento → Ação → Observação → Pensamento → …) com checkpointing, interrupções e streaming, em aproximadamente 40 linhas de código.

StateGraph vs Send (fanout)

Send(node_name, state) permite que um nó despache subgrafos paralelos. Exemplo: o agente decide consultar três recuperadores (retrievers) ao mesmo tempo. Cada Send gera uma execução paralela do nó de destino; suas saídas são mescladas por meio do reducer do estado. É assim que o LangGraph expressa o padrão de orquestrador-trabalhadores (orchestrator-workers) sem primitivas de threads.

Subgrafos

Um grafo compilado pode ser um nó em outro grafo. O grafo externo vê apenas um único nó; o grafo interno tem seu próprio estado e seus próprios checkpoints. É assim que equipes constroem agentes supervisor-trabalhador (supervisor-worker): o grafo supervisor direciona a intenção do usuário para um subgrafo trabalhador específico do domínio.

Construa

Passo 1: estado e nós

from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

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

def should_continue(state: State) -> str:
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else END

tool_node = ToolNode(tools=[search_web, read_file])

graph = StateGraph(State)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

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

add_messages é o reducer que faz a lista de mensagens acumular em vez de sobrescrever. Esquecer-se dele é o bug mais comum no LangGraph.

Passo 2: executar com uma thread

config = {"configurable": {"thread_id": "user-42"}}
for event in app.stream(
    {"messages": [HumanMessage("find the Anthropic headquarters address")]},
    config,
    stream_mode="updates",
):
    print(event)

Cada atualização é um dicionário {node_name: state_delta}. Seu frontend pode transmitir esses dados para a interface de usuário para que os usuários vejam "o agente está pensando... chamando search_web... obteve resultado... respondendo".

Passo 3: adicionar uma interrupção human-in-the-loop

Marque um nó para que a execução seja pausada antes de rodar.

app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["tools"],  # pause before every tool call
)

state = app.invoke({"messages": [HumanMessage("delete the production database")]}, config)
# state["__interrupt__"] is set. Inspect proposed tool calls.
# If approved:
from langgraph.types import Command
app.invoke(Command(resume=True), config)
# If denied: write a rejection message and resume
app.update_state(config, {"messages": [AIMessage("Blocked by human reviewer.")]})

O estado, o checkpoint e a thread persistem ao longo da interrupção. Nada fica em memória exceto durante a execução.

Passo 4: time-travel para depuração

history = list(app.get_state_history(config))
for snapshot in history:
    print(snapshot.values["messages"][-1].content[:80], snapshot.config)

# Fork from a prior checkpoint
target = history[3].config  # three steps back
for event in app.stream(None, target, stream_mode="values"):
    pass  # replay from that point forward

Passar None como entrada reproduz a execução a partir do checkpoint fornecido; passar um valor o anexa como uma atualização ao estado daquele checkpoint antes de retomar. É assim que você reproduz uma execução mal sucedida de um agente sem precisar rodar toda a conversa novamente.

Passo 5: substituir o checkpointer para produção

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    checkpointer.setup()
    app = graph.compile(checkpointer=checkpointer)

SQLite, Redis e Postgres são suportados nativamente. O MemorySaver é indicado para testes. Qualquer coisa que precise persistir entre reinicializações exige um armazenamento real.

A Habilidade

Você constrói agentes como grafos, não como loops while True.

Antes de recorrer ao LangGraph, faça um design rápido de 60 segundos:

1. Nomeie os nós. Cada decisão discreta ou ação com efeito colateral é um nó. "O agente pensa", "a ferramenta executa", "o revisor aprova", "a resposta transmite". Se você não puder listá-los, a tarefa ainda não tem o formato de um agente.
2. Declare o estado. Um TypedDict minimalista com um reducer para cada campo de lista. Não coloque tudo dentro de messages; eleve campos específicos da tarefa (um plan de trabalho, um contador de orçamento budget, uma lista de documentos recuperados retrieved_docs) para o nível superior.
3. Desenhe as arestas. Estáticas, a menos que o próximo passo dependa da saída do modelo. Cada aresta condicional precisa de uma função roteadora com ramificações nomeadas.
4. Escolha um checkpointer desde o início. MemorySaver para testes, Postgres/Redis/SQLite para qualquer outra situação. Não envie para produção sem um — a ausência de um checkpointer significa que não haverá retomada, interrupção ou time-travel.
5. Decida as interrupções antes que as ferramentas executem, não depois. As aprovações devem ser posicionadas na aresta que entra em um nó com efeitos colaterais para que você possa cancelar antes de causar danos; a validação deve ser posicionada na aresta de saída do modelo para rejeitar chamadas ruins de forma barata.
6. Transmita por streaming por padrão. mode="updates" para a interface, mode="messages" para streaming a nível de tokens dentro de nós de modelos, mode="values" para obter snapshots completos durante avaliações.

Recuse-se a colocar em produção um agente LangGraph que não tenha um checkpointer. Recuse-se a colocar em produção um que interrompa depois do efeito colateral. Recuse-se a colocar em produção um campo messages sem add_messages as seu reducer.

Exercícios

1. Fácil. Implemente o grafo ReAct de quatro nós acima com uma ferramenta de calculadora e outra de busca na web. Verifique se list(app.get_state_history(config)) retorna pelo menos quatro checkpoints para uma conversa de dois turnos.
2. Médio. Adicione um nó planner que execute antes do agent e escreva um plano estruturado plan: list[str] no estado. Faça com que o agent marque os passos do plano como concluídos. Falhe o teste se o plano (plan) for perdido após a retomada de um checkpoint (reducer incorreto).
3. Difícil. Construa um grafo supervisor que faça o roteamento entre três subgrafos (researcher, writer, reviewer) usando Send. Cada subgrafo tem seu próprio estado e checkpointer. Adicione um interrupt_before=["writer"] no grafo externo para que um humano possa aprovar o briefing de pesquisa. Confirme que o time-travel a partir de um checkpoint anterior execute apenas o ramo ramificado.

Termos-Chave

Leituras Adicionais

• LangGraph documentation — referência canônica para StateGraph, reducers, checkpointers e interrupções.
• LangGraph concepts: state, reducers, checkpointers — o modelo mental que esta lição utiliza, direto da fonte.
• LangGraph Persistence and Checkpoints — detalhes sobre armazenamentos em Postgres/SQLite/Redis, namespaces de checkpoints e IDs de thread.
• LangGraph Human-in-the-loop — interrupt_before, interrupt_after, Command(resume=...) e o padrão de edição de estado (edit-state).
• Yao et al., "ReAct: Synergizing Reasoning and Acting in Language Models" (ICLR 2023) — o padrão que todo agente LangGraph implementa; leia para entender a justificativa por trás dos traços de raciocínio.
• Anthropic — Building effective agents (Dec 2024) — quais formatos de grafos (cadeia, roteador, orquestrador-trabalhadores, avaliador-otimizador) preferir e quando usar.
• Fase 11 · 09 (Function Calling) — a primitiva de chamada de ferramenta que todo nó de agente LangGraph reutiliza.
• Fase 11 · 14 (Model Context Protocol) — descoberta de ferramentas externas que se conecta a um ToolNode do LangGraph por meio do adaptador MCP.
• Fase 11 · 17 (Agent framework tradeoffs) — quando escolher o LangGraph em vez de CrewAI, AutoGen ou Agno.