2.6 — Grafos de Estado para Fluxos de Trabalho com Agentes (LangGraph)

Objetivo da Aula

Ao concluir esta aula, você será capaz de:

Modelar um workflow de agente como grafo dirigido com nós e arestas

Implementar nós de planejamento, execução e revisão com TypedDict de estado

Adicionar aresta condicional com add_conditional_edges para retry automático

Comparar a abordagem LangGraph vs agent loop manual

Executar o grafo com grafo.invoke() e interpretar o estado final

Por que isso importa

O agent loop manual (while True: tool_call) tem limitações sérias em produção: - Sem visualização: impossível desenhar o fluxo para um stakeholder - Sem paralelismo nativo: não consegue executar nós em paralelo - Difícil de testar: mock de estados intermediários requer muito boilerplate - Retry manual: implementar “se o reviewer rejeitar, tentar novamente” vira código complexo

Decisão de Arquitetura: Agente como Grafo de Estado

O while loop manual de agente (chamar LLM → executar tool → repetir) é frágil: sem estado persistido, qualquer falha reinicia do zero. LangGraph modela o agente como grafo de computação com estado tipado compartilhado entre os nós — análogo a Step Functions da AWS ou Airflow com XCom. Cada nó é uma função; arestas condicionais decidem o próximo nó baseado no estado. O checkpointing salva o estado após cada nó por thread_id: se o processo morrer, você retoma do último checkpoint. interrupt_before pausa antes de um nó crítico para aprovação humana — o pattern HITL sem precisar redesenhar o fluxo. Use grafos de estado quando o fluxo tem mais de 3-4 etapas, precisa de retry granular, ou tem pontos de aprovação humana.

LangGraph trata o agente como um grafo de computação, onde cada nó é uma função Python e cada aresta define quando ir para qual próximo nó. O estado é um TypedDict compartilhado por todos os nós — sem variáveis globais, sem dependências implícitas.

O padrão é usado em produção por empresas como LangChain, Notion e Replit para pipelines de agentes complexos.

Conceitos Fundamentais

Estado como TypedDict

from typing import TypedDict



class EstadoAgente(TypedDict):

    tarefa: str        # Input do usuário

    plano: list[str]   # Gerado pelo planner

    resultado: str     # Gerado pelo executor

    tentativas: int    # Número de tentativas do executor

    aprovado: bool     # Decisão do reviewer

    feedback: str      # Feedback do reviewer para o executor

Cada campo do TypedDict é uma “coluna” do estado que flui pelo grafo. Os nós leem e escrevem nesse estado — é o equivalente funcional das variáveis locais do while loop.

A grande vantagem: o estado é serializable (é um dict). Você pode: - Persistir entre execuções (checkpointing) - Inspecionar o estado em qualquer ponto - Fazer replay de qualquer estado para debugging

Nós do Grafo

Cada nó é uma função que recebe o estado atual e retorna uma atualização parcial:

def no_planner(state: EstadoAgente) -> EstadoAgente:

    """Decompõe a tarefa em passos."""

    response = client.messages.create(

        model='claude-haiku-4-5-20251001',

        max_tokens=300,

        messages=[{

            'role': 'user',

            'content': f'Decomponha em 3-5 passos. JSON: {{"passos": [...]}}\nTAREFA: {state["tarefa"]}',

        }],

    )

    try:

        plano = json.loads(response.content[0].text)['passos']

    except Exception:

        plano = ['Executar tarefa diretamente']

    

    return {**state, 'plano': plano}  # retorna estado atualizado

{**state, 'plano': plano} — copia o estado e atualiza apenas plano. Os outros campos permanecem intactos.

Construção do Grafo

from langgraph.graph import StateGraph, END, START



def criar_grafo():

    grafo = StateGraph(EstadoAgente)

    

    # Adicionar nós

    grafo.add_node('planner', no_planner)

    grafo.add_node('executor', no_executor)

    grafo.add_node('reviewer', no_reviewer)

    

    # Arestas fixas

    grafo.add_edge(START, 'planner')    # sempre começa no planner

    grafo.add_edge('planner', 'executor')  # planner → executor

    grafo.add_edge('executor', 'reviewer') # executor → reviewer

    

    # Aresta condicional (TODO do starter)

    grafo.add_conditional_edges(

        'reviewer',           # nó de origem

        rotear_apos_review,   # função que decide o próximo nó

        {

            'executor': 'executor',  # se retornar 'executor', vai para executor

            '__end__': END,          # se retornar '__end__', termina

        }

    )

    

    return grafo.compile()

A Função de Roteamento

from typing import Literal



def rotear_apos_review(state: EstadoAgente) -> Literal['executor', '__end__']:

    if state['aprovado'] or state['tentativas'] >= 3:

        return END      # aprovado ou esgotou tentativas → terminar

    return 'executor'   # não aprovado e ainda tem tentativas → retry

Essa função é o coração do controle de fluxo. Em vez de if/else aninhados no loop, você declara a lógica de roteamento explicitamente.

Execução do Grafo

# Inicializar com estado inicial

estado_inicial: EstadoAgente = {

    'tarefa': 'Crie um relatório sobre LLMs em produção.',

    'plano': [],

    'resultado': '',

    'tentativas': 0,

    'aprovado': False,

    'feedback': '',

}



grafo = criar_grafo()

resultado_final = grafo.invoke(estado_inicial)



print(resultado_final['resultado'])      # resposta final

print(resultado_final['tentativas'])     # quantas tentativas foram necessárias

grafo.invoke() executa o grafo até END e retorna o estado final.

Aprofundamento Técnico

Por que LangGraph vs Agent Loop Manual

Aspecto	Agent Loop (while)	LangGraph
Visualização	Não	Sim (grafo)
Teste de nós	Difícil	Fácil (funções puras)
Paralelismo	Manual	Nativo
Retry	Código manual	add_conditional_edges
Checkpointing	Manual	Nativo
Debug de estado	print()	Inspeção do dict

Checkpointing: Sobrevivendo a Falhas

from langgraph.checkpoint.memory import MemorySaver



def criar_grafo_com_checkpoint():

    grafo = StateGraph(EstadoAgente)

    # ... adicionar nós e arestas ...

    

    checkpointer = MemorySaver()  # em produção: SqliteSaver ou PostgresSaver

    return grafo.compile(checkpointer=checkpointer)



# Executar com thread_id para recuperação

config = {'configurable': {'thread_id': 'sessao-abc123'}}

grafo.invoke(estado_inicial, config=config)



# Se a execução falhar, retomar do último checkpoint:

grafo.invoke(None, config=config)  # None = retomar do checkpoint

Human-in-the-Loop com interrupt_before

grafo = StateGraph(EstadoAgente)

# ...

grafo.compile(

    interrupt_before=['executor'],  # pausa antes do executor para aprovação humana

)



# Execução até o ponto de interrupção

resultado = grafo.invoke(estado_inicial)

# → pausa antes de 'executor'



# Depois de aprovação humana:

grafo.invoke(None, config=config)  # continua

Esse é o padrão HITL (Human-in-the-Loop) do LangGraph — pausar antes de ações de alto impacto.

Paralelismo: Fan-Out e Fan-In

# Executar analista E pesquisador em paralelo

grafo.add_edge('planner', 'analista')

grafo.add_edge('planner', 'pesquisador')

# LangGraph executa os dois nós em paralelo automaticamente



# Fan-in: esperar ambos antes de continuar

grafo.add_edge(['analista', 'pesquisador'], 'synthesizer')

Exemplos Anotados

Exemplo 1: TODO do Starter — Aresta Condicional

def criar_grafo():

    if not _LANGGRAPH:

        return None

    

    grafo = StateGraph(EstadoAgente)

    

    grafo.add_node('planner', no_planner)

    grafo.add_node('executor', no_executor)

    grafo.add_node('reviewer', no_reviewer)

    

    grafo.add_edge(START, 'planner')

    grafo.add_edge('planner', 'executor')

    grafo.add_edge('executor', 'reviewer')

    

    # TODO: substituir add_edge('reviewer', END) por:

    grafo.add_conditional_edges(

        'reviewer',

        rotear_apos_review,

        # Mapeamento: valor retornado → próximo nó

        # 'executor' → nó 'executor'

        # END → fim do grafo

    )

    

    return grafo.compile()

A função rotear_apos_review já está implementada no starter:

def rotear_apos_review(state: EstadoAgente) -> Literal['executor', '__end__']:

    if state['aprovado'] or state['tentativas'] >= 3:

        return END

    return 'executor'

Exemplo 2: Trace de Execução Completo

[PLANNER] Planejando: Crie um relatório sobre LLMs em produção.

  Plano: ['Definir contexto', 'Listar casos de uso', 'Discutir desafios', 'Conclusão']



[EXECUTOR] Executando plano (4 passos, tentativa 1)

  Resultado: "LLMs em produção são amplamente usados..."



[REVIEWER] Avaliando resultado...

  Aprovado: False. Feedback: Faltou dados quantitativos e exemplos concretos.



[EXECUTOR] Executando plano (4 passos, tentativa 2)  ← retry automático via aresta condicional

  Resultado: "LLMs em produção: GPT-4 processa 100B tokens/dia..."



[REVIEWER] Avaliando resultado...

  Aprovado: True.

→ END



Resultado final (2 tentativas):

"LLMs em produção: GPT-4 processa 100B tokens/dia..."

Exemplo 3: Versão Sem LangGraph para Comparação

def executar_sem_langgraph(tarefa: str) -> dict:

    """Mesmo fluxo, implementado manualmente."""

    estado: EstadoAgente = {

        'tarefa': tarefa, 'plano': [], 'resultado': '',

        'tentativas': 0, 'aprovado': False, 'feedback': '',

    }

    

    estado = no_planner(estado)

    

    for _ in range(3):

        estado = no_executor(estado)

        estado = no_reviewer(estado)

        if estado['aprovado']:

            break

    

    return estado

Mesma lógica, mas sem: visualização, checkpointing, paralelismo, HITL nativo.

Padrões e Armadilhas

Padrões

Padrão 1: Estado imutável nos nós — sempre {**state, ...}

# BOM: não modifica o estado original

return {**state, 'resultado': novo_resultado}



# RUIM: mutação in-place (pode causar bugs sutis)

state['resultado'] = novo_resultado

return state

Padrão 2: Guardrail de tentativas no roteamento

def rotear_apos_review(state: EstadoAgente) -> Literal['executor', '__end__']:

    if state['aprovado'] or state['tentativas'] >= 3:  # ← sempre ter limite

        return END

    return 'executor'

Sem state['tentativas'] >= 3, um reviewer que nunca aprova cria loop infinito.

Padrão 3: Usar Haiku nos nós intermediários, modelo mais capaz apenas no executor

# Planner e Reviewer: custo baixo

model='claude-haiku-4-5-20251001'



# Executor: tarefa principal — pode usar modelo mais capaz

model='claude-sonnet-4-6'

Armadilhas

⚠️ Armadilha 1: add_edge('reviewer', END) sem condicional bloqueia retry

# ERRADO: reviewer sempre vai para END, sem retry

grafo.add_edge('reviewer', END)



# CORRETO: usar conditional_edges

grafo.add_conditional_edges('reviewer', rotear_apos_review)

⚠️ Armadilha 2: LangGraph não instalado = ImportError O starter trata isso:

try:

    from langgraph.graph import StateGraph, END, START

    _LANGGRAPH = True

except ImportError:

    _LANGGRAPH = False

Se LangGraph não estiver instalado, o criar_grafo() retorna None e o código usa executar_sem_langgraph.

⚠️ Armadilha 3: Estado inicial incompleto

# ERRADO: campo 'tentativas' faltando → KeyError em no_reviewer

estado_inicial = {'tarefa': 'Criar relatório', 'aprovado': False}



# CORRETO: todos os campos do TypedDict inicializados

estado_inicial: EstadoAgente = {

    'tarefa': tarefa, 'plano': [], 'resultado': '',

    'tentativas': 0, 'aprovado': False, 'feedback': '',

}

O grafo de estado resolve o problema de durabilidade e controle de fluxo — o próximo passo é instrumentar esse sistema para que você saiba o que ele está fazendo, quanto está custando, e quando interromper.

⚗ Laboratório prático — mozak.tech
Se não for realizar o laboratório, pule para o próximo capítulo.

Ponte para o Lab

O TODO do starter está em criar_grafo():

# Substituir:

grafo.add_edge('reviewer', END)



# Por:

grafo.add_conditional_edges(

    'reviewer',

    rotear_apos_review,

)

Após essa mudança, quando o reviewer retornar aprovado=False e tentativas < 3, o grafo vai automaticamente re-executar o nó executor com o feedback do reviewer disponível no estado.

Instalar LangGraph (se necessário):

pip install langgraph langchain-anthropic

Verificar que funciona: o trace deve mostrar [EXECUTOR] aparecendo 2x antes do END se o reviewer rejeitar na primeira tentativa.

Agora você está pronto para o lab.