06-01-02 — LangGraph: grafos de estado para agentes complexos

Por que grafos de estado?

Um agente ReAct simples (pense → aja → observe → repita) funciona para tarefas lineares. Mas quando você precisa de:

• Múltiplos agentes colaborando
• Revisão humana em pontos específicos (human-in-the-loop)
• Retry com lógica diferente dependendo do erro
• Subgrafos paralelos que se sincronizam

…um loop simples não basta. LangGraph modela isso como um directed graph onde o estado é tipado e persiste.

Conceitos fundamentais

StateGraph e TypedDict

O estado é um TypedDict que flui entre todos os nós. Cada nó recebe o estado atual e retorna um dict parcial para atualizar.

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator

# Define o estado do agente
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]  # lista acumulativa
    next_action: str

# Nó: chama o LLM
def call_llm(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# Nó: executa ferramentas
def execute_tools(state: AgentState) -> dict:
    last_message = state["messages"][-1]
    # processa tool_calls do AIMessage
    results = []
    for tool_call in last_message.tool_calls:
        result = tools_by_name[tool_call["name"]].invoke(tool_call["args"])
        results.append(ToolMessage(content=str(result), tool_call_id=tool_call["id"]))
    return {"messages": results}

# Roteamento condicional
def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END

# Monta o grafo
graph = StateGraph(AgentState)
graph.add_node("llm", call_llm)
graph.add_node("tools", execute_tools)
graph.set_entry_point("llm")
graph.add_conditional_edges("llm", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "llm")  # após ferramentas, volta ao LLM

app = graph.compile()

Diagrama do fluxo

Checkpointing — persistência de estado

LangGraph tem um sistema de checkpointing que salva o estado após cada nó. Isso permite:

• Pausar e retomar: o agente para, um humano aprova, o agente continua
• Time-travel debugging: replay a partir de qualquer checkpoint
• Multi-turn conversation: cada thread_id mantém seu histórico

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres import PostgresSaver  # para produção

# Checkpointer em memória (dev/teste)
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# Cada conversa tem um thread_id único
config = {"configurable": {"thread_id": "usuario-123"}}

# Primeira mensagem
result1 = app.invoke(
    {"messages": [HumanMessage(content="Qual a temperatura em São Paulo?")]},
    config=config
)

# Segunda mensagem — estado anterior é recuperado automaticamente
result2 = app.invoke(
    {"messages": [HumanMessage(content="E no Rio?")]},
    config=config
)

# Ver histórico de estados
for state in app.get_state_history(config):
    print(state.values["messages"][-1].content)

Human-in-the-loop

from langgraph.graph import interrupt

def aprovacao_humana(state: AgentState) -> dict:
    # Pausa o grafo aqui — aguarda input externo
    decisao = interrupt("Aprovar esta ação? (sim/não)")
    if decisao == "sim":
        return {"next_action": "executar"}
    return {"next_action": "cancelar"}

# Para retomar após aprovação:
app.invoke(Command(resume="sim"), config=config)

Multi-agent com LangGraph

LangGraph suporta subgrafos: um nó pode ser outro grafo compilado. Isso permite hierarquias de agentes (supervisor → worker) com estado isolado por agente.

from langgraph.graph import StateGraph

# Subgrafo especialista em pesquisa
researcher_graph = StateGraph(ResearchState)
# ... define nós e arestas do researcher ...
researcher_app = researcher_graph.compile()

# Grafo principal usa researcher como nó
main_graph = StateGraph(MainState)
main_graph.add_node("researcher", researcher_app)  # subgrafo como nó
main_graph.add_node("writer", writer_node)