Agentes com LangGraph: quando multiagente é a resposta errada

Lede. LangGraph não conserta prompt ruim. Ele conserta o que o prompt não conserta: controle de fluxo, estado explícito e retomada depois de uma falha no meio do caminho. Neste artigo você monta um grafo de agentes em Python, vê o estado tipado e os reducers decidirem como dois nós escrevem no mesmo lugar, e termina com a seção que importa mais: os três casos em que não use agente. Os blocos com LangGraph não são executados aqui porque a biblioteca não é dependência deste repositório; todo o resto roda. Onde você está na linha. Este é o passo 10 de 13 — orquestração da geração. Antes dele: 01 · Anatomia do pipeline — de onde vêm o Scored, o Budget e a montagem. Depois dele: 11 · Observabilidade, 12 · Guia de decisão e 13 · Produção. Este artigo é autossuficiente: quem não leu o 01 lê os contratos aqui de novo.

---

1. Quando uma cadeia linear já resolve

Vamos começar pelo que não precisa de framework. Uma resposta de RAG tem, na

forma mais simples, três etapas: recuperar, montar, gerar. Isso é uma função que chama

outra função. Sem grafo, sem estado, sem nada.

from __future__ import annotations

from collections.abc import Callable, Sequence
from dataclasses import dataclass, field
from typing import Any


@dataclass(frozen=True)
class Resultado:
    """A saída de uma requisição: a resposta e o que ela usou para sair."""

    resposta: str
    trechos_usados: tuple[str, ...] = ()
    chamadas: int = 0


def recuperar(pergunta: str, k: int) -> list[tuple[str, str]]:
    """Retorna (ref_id, trecho). Num sistema real, a busca híbrida do artigo 04."""
    return [("t1", "O prazo de pagamento padrão é de 30 dias."),
            ("t2", "A renovação automática vale sem recusa no prazo anterior.")]


def montar(trechos: Sequence[tuple[str, str]]) -> str:
    """Junta os trechos no texto que o modelo vai ler."""
    return "\n\n".join(f"[{ref}] {texto}" for ref, texto in trechos)


def gerar(prompt: str) -> str:
    """Placeholder da chamada ao modelo. Nenhuma parte do artigo depende de rede."""
    return f"[o modelo recebeu {len(prompt)} caracteres]"


def responder(pergunta: str) -> Resultado:
    """A cadeia linear: três chamadas, uma atrás da outra, sem estado.

    Funciona para a maioria dos casos de RAG. Quando ela deixa de funcionar é
    quando algum passo precisa decidir algo, voltar atrás, esperar um humano, ou
    sobreviver a uma falha — e é aí que a seção 3 entra.
    """
    trechos = recuperar(pergunta, k=2)
    if not trechos:
        return Resultado(resposta="Não encontrei nada na base.", chamadas=0)
    prompt = f"{montar(trechos)}\n\nPergunta: {pergunta}\nResposta:"
    return Resultado(resposta=gerar(prompt), chamadas=1)


print(responder("qual o prazo de pagamento?"))

Quatro linhas de orquestração, e funciona. O que essa versão não tem:

• Nenhum estado explícito. As variáveis vivem no escopo da função. Se você quiser

  ver o que o sistema tinha em mãos no meio do caminho, precisa de print ou de

  breakpoint.

• Nenhuma retomada. Se a terceira chamada estoura timeout, a requisição morre. Não

  há como voltar ao passo 2 e tentar de novo só ele.

• Nenhum ciclo. Se o passo 2 quiser refazer a busca com outra consulta, alguém

  escreve um while à mão, e esse while é um grafo sem formalism.

• Nenhum humano no meio. Se a resposta precisa de aprovação antes de virar ação,

  não há onde parar.

Cada uma dessas quatro ausências custa caro em um sistema real, e é cada uma delas que

o LangGraph resolve. Mas resolver as quatro é um preço alto, e é por isso que a última

seção deste artigo é a mais importante.

⚠️ "Vou usar LangGraph porque é o padrão" é a decisão invertida
O LangGraph não é um framework de RAG. Ele é um orquestrador de estado com

checkpoint. Se a sua pipeline é recuperar -> montar -> gerar, ele traz três

coisas que você não precisa (dependência, conceito, superfície de bug) e não traz

nada que você já não tenha. O teste honesto não é "o meu caso é complexo o

bastante?", é "qual das quatro ausências da seção 1 eu tenho?". Se você não

responder pelo menos uma delas com um exemplo concreto, não use agente.

---

2. Os quatro recursos que justificam o grafo

Cada recurso abaixo tem um sintoma próprio. O que importa é que nenhum deles é

"resolver melhor": todos são sobre controle, não sobre qualidade de resposta. É por

isso que um agente com busca ruim produz uma resposta ruim que pode estar em três

lugares diferentes, e é por isso que a medição (artigo 11) vem antes do agente, não

depois.

Vale notar de onde veio a ideia: o padrão "pensar, observar, agir" do

ReAct pressupõe que a observação seja boa — o

ciclo só raciocina sobre o que a busca devolveu. Ciclo de raciocínio em cima de

observação ruim é multiplicador de erro, não de qualidade.

Recurso	O que resolve	O sintoma que indica que você precisa dele
state • reducer	dois nós escrevendo no mesmo campo sem se atropelar	um _resultado_accumulado global, ou um dict gigante passado por argumento
ciclo	o passo 2 refaz a busca com outra consulta	um for tentativas dentro de uma função, com a busca embaixo
checkpointer • thread_id	retomar depois de falha, ou continuar amanhã	o cliente reenvia a pergunta e você paga tudo de novo
interrupt	humano decide no meio do fluxo	um input() no meio do servidor, ou um e-mail pedindo aprovação

Nenhum desses quatro é "o modelo fica mais inteligente". Todos os quatro são sobre o

ciclo de vida da requisição, e é por isso que eles valem a dependência em uns casos e

não em outros. O state e o reducer você resolve com um dataclass e uma função de

merge, e fica com 90% do benefício. O ciclo você resolve com um for. Já o

checkpointer com thread_id é difícil de fazer fora de um orquestrador, porque exige

persistir o estado em fronteira de passo — e é ele que muda a conversa de "sistema

síncrono" para "processo".

---

3. O que entra no state, e o que não entra

O state é o estado tipado que o grafo carrega de nó em nó. No LangGraph ele é um

TypedDict, um dataclass ou um modelo Pydantic, e é o schema das chaves: o que

existe, e de que tipo.

A decisão que importa mais não é o que entra, é o que não entra. Um state que

carrega o segredo do banco e o volume bruto dos documentos é um state que vai para o

checkpoint — ou seja, vai para o disco, ou para o Postgres, ou para o log de

observabilidade.

from typing import Annotated, TypedDict


def fundir_listas(a: list[str], b: list[str]) -> list[str]:
    """Acumula sem duplicar. É o reducer padrão para "achados"."""
    return a + [item for item in b if item not in a]


def somar(a: int, b: int) -> int:
    """Soma contadores: tentativas, tokens, chamadas de ferramenta."""
    return a + b


class Estado(TypedDict, total=False):
    """O contrato do grafo: o que circula entre os nós.

    `total=False` porque o estado nasce vazio e cada nó preenche o que precisa.
    A chave `pergunta` não tem reducer: só um nó a escreve, e o LangGraph
    recusa que dois nós escrevam a mesma chave sem reducer no mesmo passo.
    """

    pergunta: str
    # Acumuladores: vários nós escrevem, o reducer combina.
    trechos: Annotated[list[str], fundir_listas]
    chamadas: Annotated[int, somar]
    # Substituição: o último a escrever vence, e é isso que se quer.
    resposta_final: str
    # Controle do ciclo: o roteador lê isto para decidir para onde ir.
    pode_responder: bool

O que não entra, e por quê:

• Segredo de conexão e chave de API. Eles vão para o checkpoint. Um checkpoint em

  Postgres é um log de auditoria, e log de auditoria é justamente onde você não quer a

  sua chave.

• Volume bruto (o texto de todos os documentos recuperados). O state é

  serializado a cada passo; o bruto multiplica o tamanho do checkpoint por um fator que

  ninguém note.

• Handle de conexão e objeto de sessão. Não é serializável, e o LangGraph avisa.

• O que dá para recalcular barato. Se o passo 2 pode ser reexecutado a partir do

  passo 1 em milissegundos, ele não precisa estar no checkpoint — precisa é que

  reexecutar seja barato, e isso é a seção 6.

O que entra, em ordem de prioridade: a pergunta, o resultado de cada passo (não o

material bruto dele), os contadores que viram métrica, e o controle do ciclo.

E há um terceiro caminho para o item "não entra": o campo entra no state, para o

nó usar, mas não é checkpointado. O marcador é UntrackedValue, uma anotação que

diz "esta chave existe em execução e some no checkpoint".

def estado_com_cache():   # <- não é chamada de propósito
    """Um campo que existe em execução e nunca é checkpointado.

    `UntrackedValue` é a resposta para o item "handle de conexão" da lista
    acima: o nó precisa do cliente HTTP durante a execução, e o cliente não pode
    ir para o disco. O valor é lido e escrito normalmente pelos nós, e o
    checkpointer o ignora.

    NÃO vou escrever a linha de declaração aqui. A documentação da API de
    estado do LangGraph traz a seção "Untracked values" só com exemplo em
    TypeScript (`new UntrackedValue(...)` dentro de um `StateSchema`), e a
    grafia em Python mudou de forma entre versões -- o que você viu em tutorial
    antigo pode ser `Annotated[...]`, o que a doc descreve hoje é outro
    desenho. O que é estável é o **comportamento**, e é ele que importa:
    durante a execução o valor existe e é legível; no checkpoint ele é excluído;
    na retomada ele volta ao estado inicial (ou não existe).
    """
    import langgraph.graph as lg   # noqa: PLC0415

    #vei a declaração da SUA versão na doc antes de copiar:
    #https://docs.langchain.com/oss/python/langgraph/graph-api
    assert hasattr(lg, "StateGraph"), "confira a doc da sua versão"
    raise NotImplementedError("veja a doc: a sintaxe mudou entre versões")

O ganho é duplo: o checkpoint não cresce com o que não precisa viajar, e o

serializador não quebra com o que não sabe serializar. O preço é o mesmo de qualquer

campo não rastreado: na retomada ele não está lá. Se o próximo nó precisa dele, ou

ele é recalculável no começo do nó, ou ele não deveria ser não rastreado — deveria

ser um campo comum.

💡 Regra prática: se o valor não é necessário para o próximo nó decidir, ele não
entra no state

O state é lido por todo nó seguinte e gravado a cada passo. Cada campo é uma

taxa por requisição. "A resposta final" é necessária (é a saída). "Os 40

documentos recuperados" não é — o próximo nó só precisa dos 5 que entraram no

orçamento. Quando em dúvida, deixe o valor fora e passe pelo closure do nó.

---

4. Reducers: a parte que ninguém configura e todo mundo tropeça

O reducer é a regra que decide como o estado de dois nós se combina quando ambos

escrevem a mesma chave no mesmo passo. Sem ele, o LangGraph recusa a execução

(InvalidUpdateError na chave). Com ele, você escolhe a regra.

Por que Annotated é obrigatório: o reducer não pode ser adivinhado a partir do tipo.

list pode significar "substitui" ou "acumula", e as duas coisas estão certas em

situações diferentes. O tipo não sabe; a anotação diz.

def unir_dicionarios(a: dict[str, float], b: dict[str, float]) -> dict[str, float]:
    """Funde scores por chave, somando. Para "score por trecho"."""
    return {**a, **{k: a.get(k, 0.0) + v for k, v in b.items()}}


# As quatro regras que resolvem quase todo caso real:
reducer_acumula = fundir_listas        # lista de trechos, achados, mensagens
reducer_soma = somar                   # contadores de chamada, custo, token
reducer_funde = unir_dicionarios       # score por chave, contagem por tipo
reducer_sobrescreve = None             # último a escrever vence: é o padrão


# O teste: o que acontece quando DOIS nós escrevem a MESMA chave no MESMO passo?
def colide(estado_a: list[str], estado_b: list[str]) -> list[str]:
    """Dois nós acham o mesmo trecho. `reducer_acumula` deduplica.

    Sem o reducer, isso é `InvalidUpdateError` — e a mensagem aponta a chave,
    não o nó, o que torna o erro mais chato de debugar do que parece.
    """
    return reducer_acumula(estado_a, estado_b)


print(colide(["t1", "t2"], ["t2", "t3"]))     # ['t1', 't2', 't3']
print(unir_dicionarios({"t1": 0.9}, {"t1": 0.4, "t2": 0.7}))   # {'t1': 1.3, 't2': 0.7}

A escolha do reducer por chave é a decisão de design mais barata e mais consequente

do grafo, e ela é feita no schema, uma vez. Por isso vale a pena escrever a tabela

dos seus campos antes de escrever a primeira função de nó:

Chave	Quem escreve	Reducer certo	O que acontece com o errado
pergunta	um nó só	nenhum	InvalidUpdateError se dois nós escreverem
trechos	busca, depois reranker	acumular sem duplicar	duplicata ocupa orçamento (artigo 09)
chamadas	todo nó que chama modelo	somar	vira "último valor", e a métrica mente
score_por_id	busca e grafo	fundir por chave	um ramo sobrescreve o outro silenciosamente
resposta_final	o nó final	sobrescrever	dois rascunhos viram um e ninguém sabe qual

---

5. Conditional edges: o roteador, que é a parte que precisa de teste

Edge condicional é a aresta que depende do estado: em vez de ligar o nó A no nó B

sempre, ela chama uma função que olha o estado e devolve o nome do próximo nó.

# O LangGraph não é dependência deste repositório, então nada deste bloco é
# executado pelo verificador da série. Para rodar de verdade, instale a
# biblioteca (`pip install langgraph`) e chame `montar_fluxo()`. As assinaturas
# usadas aqui estão no [Graph API](https://docs.langchain.com/oss/python/langgraph/graph-api).
def montar_fluxo():   # <- não é chamada de propósito
    """Monta e compila o grafo. É esta função que o seu serviço chama."""
    from langgraph.graph import END, START, StateGraph

    def rotear(estado: Estado) -> str:
        """Decide o próximo passo olhando o estado. É a função mais crítica do
        grafo, e a que precisa de teste de tabela.

        Ela decide até onde o ciclo vai — e por isso um `if` trocado aqui não
        dá erro: dá um ciclo que não termina, ou uma resposta que para antes da
        hora. O limite de chamadas é uma linha sua, não do LangGraph.
        """
        if not estado.get("trechos"):
            return "sem_evidence"         # nada recuperado: não há o que responder
        if not estado.get("pode_responder", True):
            return "pedir_revisao"        # o ciclo parou: alguém precisa aprovar
        if estado.get("chamadas", 0) > 6:
            return "dar_tempo_limite"     # orçamento de chamadas estourado
        return "gerar"

    def buscar(estado: Estado) -> dict:
        """Nó 1. Recupera e escreve em `trechos` (o reducer acumula)."""
        achados = recuperar(estado["pergunta"], k=3)
        return {"trechos": [ref for ref, _ in achados], "chamadas": 1}

    def gerar_resposta(estado: Estado) -> dict:
        """Nó 2. Monta e chama o modelo, contando a chamada."""
        prompt = f"{montar(list(estado['trechos']))}\n\nPergunta: {estado['pergunta']}"
        return {"resposta_final": gerar(prompt), "chamadas": 1}

    def pedir_revisao(estado: Estado) -> dict:
        """Nó 3. Espera um humano. Não chama modelo, não consome orçamento."""
        return {"resposta_final": "Resposta pronta, aguardando aprovação."}

    def sem_evidencia(estado: Estado) -> dict:
        """Nó 4. Sem trecho não há resposta, e insistir só é custo."""
        return {"resposta_final": "Não encontrei essa informação na base."}

    # O grafo. `StateGraph` recebe o schema; cada `add_node` recebe a função.
    fluxo = StateGraph(Estado)
    fluxo.add_node("buscar", buscar)
    fluxo.add_node("gerar", gerar_resposta)
    fluxo.add_node("pedir_revisao", pedir_revisao)
    fluxo.add_node("sem_evidence", sem_evidencia)

    # Arestas normais e arestas condicionais. A condicional vem do roteador,
    # com o mapa nome-do-nó -> nome-do-nó, para o grafo ficar legível e o
    # compilador conseguir validar os destinos.
    fluxo.add_edge(START, "buscar")
    fluxo.add_edge("buscar", "gerar")
    fluxo.add_edge("pedir_revisao", END)
    fluxo.add_edge("sem_evidence", END)
    fluxo.add_conditional_edges(
        "gerar",
        rotear,
        {
            "gerar": "gerar",                       # ciclo: refaz a geração
            "pedir_revisao": "pedir_revisao",
            "sem_evidence": "sem_evidence",
            "dar_tempo_limite": "pedir_revisao",   # sem tempo, entrega o que tem
        },
    )
    return fluxo.compile()

Três detalhes desse código que a documentação deixa na mão e que muda o resultado:

• add_conditional_edges recebe o mapa, não só a função. Com o mapa, o destino é

  um nó nomeado e o compilador valida que ele existe. Sem o mapa, um return "gerar"

  errado na função só vira erro em tempo de execução, no meio da requisição.

• O START e o END vêm do próprio LangGraph e são as arestas de entrada e saída.

• add_sequence liga uma lista de nós em cadeia, e é o atalho para

  add_edge repetido — mas não aceita condição, então serve só para a parte linear do

  fluxo.

⚠️ O ciclo é o que transforma custo em runaway
Um nó que chama o modelo e volta para si mesmo é um ciclo sem freio, e o

LangGraph não pone limite. Sem um if de contagem no roteador, uma resposta

ruim gera dez chamadas de modelo antes de qualquer erro aparecer. É por isso que

a função rotear acima lê chamadas antes de decidir gerar de novo: o

limite de chamadas não é uma proteção do LangGraph, é uma linha sua no

roteador. E o número do limite é seu — 6 aqui é arbitrário e vale para este

grafo, não para o seu.

---

6. Checkpointer, thread_id e retomada

O checkpointer é o que permite pausar e retomar uma execução. Ele grava o state

em fronteira de super-step — que é o nome que o LangGraph dá para "cada rodada em

que vários nós sem dependência rodam em paralelo". Um super-step é a sua unidade de

transação: o que rodou nele, rodou inteiro.

E aqui está a armadilha de produção, e ela precisa ficar em destaque: o checkpoint nunca é salvo no meio da função de um nó. Ao retomar, o nó roda de novo desde o começo. Então qualquer efeito colateral que você executou antes do ponto de parada

repetiu.

O que salva é o thread_id: é ele que diz ao checkpointer de qual execução estamos

falando. Sem ele, cada invocação é uma thread nova e a retomada não acontece. Com ele,

cada conversa é uma linha no checkpoint, e o segundo invoke com o mesmo id continua

de onde parou. O formato do checkpoint e do thread_id está documentado em

Persistência do LangGraph.

# Mesma regra do bloco anterior: o LangGraph não é dependência deste repositório,
# então o código que o usa fica dentro de uma função não chamada.
def com_checkpoint():   # <- não é chamada de propósito
    """Compila o grafo com checkpointer em memória e mostra a retomada."""
    from langgraph.checkpoint.memory import MemorySaver

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

    # O `thread_id` é obrigatório sempre que há checkpointer: sem ele, o
    # LangGraph não sabe qual execução retomar.
    config = {"configurable": {"thread_id": "req-123"}}

    # Primeira chamada: roda do START até o fim.
    estado1 = app.invoke({"pergunta": "qual o prazo?"}, config)
    print(estado1["chamadas"], estado1["resposta_final"])

    # Segunda chamada, **mesmo** thread_id: o grafo retoma do checkpoint, não
    # reexecuta tudo. Este é o mecanismo de "o cliente voltou amanhã".
    estado2 = app.invoke({"pergunta": "e a renovação?"}, config)
    print(estado2["resposta_final"])
    return app

Em produção, o checkpointer em memória não serve — ele morre com o processo, que é

justamente o que você quer sobreviver. O de Postgres é o de produção:

def com_postgres(uri: str):   # <- não é chamada de propósito
    """A versão de produção: checkpointer compartilhado entre as réplicas."""
    from langgraph.checkpoint.postgres import PostgresSaver

    # `setup()` cria o schema do checkpoint na primeira vez. Ele é idempotente:
    # rodar de novo não quebra nada, mas rodar **nunca** é o erro, porque o
    # primeiro `invoke` falha sem as tabelas.
    with PostgresSaver.from_conn_string(uri) as checkpointer:
        checkpointer.setup()                    # uma vez, no deploy
        return montar_fluxo().compile(checkpointer=checkpointer)

E o thread_id vem de onde? Não é um número que você inventa por requisição: ele é a

identidade da conversa, e quem a possui é a sua aplicação (o id da sessão do

usuário, o id do caso, o id do ticket). Se você gerar um thread_id novo a cada

invoke, a retomada nunca acontece — porque cada chamada é uma conversa diferente.

⚠️ Checkpoint em memória com deploy horizontal escala errado
Duas réplicas do mesmo serviço, o mesmo thread_id, e o checkpointer em memória

de cada uma: a segunda réplica não sabe o que a primeira já rodou, e cada

invocação recomeça do zero (ou pior, diverge do estado que o cliente já viu).

O checkpointer precisa ser comparthado entre as réplicas — Postgres, Redis

ou equivalente. Memória é para o teste local e para o notebook, nunca para o

serviço que atende dois usuários ao mesmo tempo.

---

7. Interromper para um humano no meio

interrupt_before pausa antes de um nó, no limite de um super-step. É a forma mais

simples de humano no meio: o grafo para, e quem retoma é a sua aplicação — que pode

mostrar o rascunho numa tela, num e-mail, num Slack.

def pausar_para_revisao():   # <- não é chamada de propósito
    """Interrompe antes de `gerar` e mostra que a retomada usa o mesmo id."""
    from langgraph.checkpoint.memory import MemorySaver

    app = montar_fluxo().compile(
        checkpointer=MemorySaver(),
        interrupt_before=["gerar"],
    )
    # O `invoke` para no interrupt. O estado até `buscar` está no checkpoint.
    estado_parado = app.invoke({"pergunta": "qual o prazo?"},
                               {"configurable": {"thread_id": "req-9"}})
    print("parou antes de gerar:", estado_parado.get("trechos"))

    # Depois do humano aprovar, a mesma thread_id continua de onde parou.
    estado_final = app.invoke(None, {"configurable": {"thread_id": "req-9"}})
    print(estado_final["resposta_final"])
    return app

O padrão, então, é: pause no limite do nó, não no meio dele. E o thread_id é o

que amarra o "antes" ao "depois". Um humano que aprova no Slack e um invoke com o

mesmo id é a mesma execução, vista de dois lugares.

---

8. Falha e retomada: o efeito colateral que se repete

Você já viu que o nó reexecuta do começo ao retomar. O que isso significa na prática é

uma armadilha que não aparece em nenhum teste, porque teste você roda uma vez.

Vamos provar. O padrão perigoso é um nó que escreve e depois chama o modelo (que

pode estourar timeout no meio). Ao retomar, o nó roda de novo, e a escrita acontece de

novo.

# Uma "ferramenta" com efeito colateral: registrar num log externo.
registros: list[str] = []


def chamar_modelo(falhar: bool = False) -> str:
    """Placeholder da chamada ao modelo. Com `falhar=True`, ela estoura.

    É assim que o provedor se comporta: timeout, rate limit, 500. O nó não tem
    como evitar, e é exatamente por isso que ele precisa ser reexecutável.
    """
    if falhar:
        raise TimeoutError("timeout do provedor")
    return "resposta do modelo"


def registrar(peca: str) -> str:
    """Efeito colateral: escreve FORA do grafo. Não é checkpointado."""
    registros.append(peca)
    return f"[gravado: {peca}]"


def no_com_efeito(estado: dict, falhar: bool = False) -> dict:
    """Nó que escreve e depois pode falhar.

    Se a chamada ao modelo depois da escrita estourar, o checkpoint guardado é o
    do super-step ANTERIOR: este nó não completou, e o estado dele não existe
    ainda. Ao retomar, o nó roda de novo desde a primeira linha — e `registrar`
    grava de novo. O estado final está consistente; o mundo externo, não.
    """
    peca = f"pedido-{estado['id']}"
    mensagem = registrar(peca)                   # efeito colateral ANTES do risco
    chamar_modelo(falhar=falhar)                  # <- o ponto de risco
    return {"peca": mensagem, "tentativas": estado.get("tentativas", 0) + 1}

Para tornar isso observável — e é aqui que a maioria dos sistemas perde o controle, já

que a repetição é silenciosa por definição — a simulação abaixo executa o mesmo nó

duas vezes com o mesmo estado, que é o que a retomada faz, e imprime quantas vezes a

gravação aconteceu.

# Simulação da retomada: o checkpointer guardou o estado de ANTES do nó, e o nó
# reexecuta do começo. É a mesma coisa que o LangGraph faz, sem a biblioteca.
checkpoint: dict = {"id": "9", "tentativas": 0}

# --- Primeira execução: a chamada ao modelo estoura depois da gravação. ---
try:
    no_com_efeito(dict(checkpoint), falhar=True)
except TimeoutError:
    print("1a execução: timeout depois do efeito colateral")

# --- Retomada: mesmo estado guardado, mesmo nó, do começo. ---
resultado = no_com_efeito(dict(checkpoint))
print("2a execução: ok,", resultado)
print("tentativas no estado:", resultado["tentativas"])
print("o mesmo pedido foi gravado", registros.count("pedido-9"), "vezes")

O resultado é o ponto inteiro da seção: 1 tentativa no estado (a que o grafo conta) e

2 gravações no log. O estado está consistente; o mundo externo não está. E a

correção não é "não use LangGraph" — é nenhum efeito colateral antes do ponto de risco:

• Escreva depois da chamada que pode falhar, não antes.

• Ou torne a escrita idempotente pelo id da execução (grava por chave, nunca

  acresenta linha nova).

• Ou mova a escrita para um nó depois do que pode falhar, e deixe esse nó ser o

  ponto de parada.

E, para fechar a parte de custo: se o nó chama o modelo três vezes, uma retomada

executa três chamadas de novo. O custo de um ciclo de 8 nós é 8 chamadas por

iteração, e três iterações dão 24 chamadas — e isso é aritmética sobre o seu grafo, não

benchmark de ninguém. É por isso que a recuperação precisa estar medida (artigo 11)

antes do agente: sem ela, o ciclo é um multiplicador de custo sobre um número que você

não conhece.

---

9. Os três casos em que NÃO usar agente

Esta é a seção mais importante do artigo, e ela vem antes da conclusão. Os três casos

são de RAG, e são os que mais aparecem.

1. A cadeia é linear e não tem decisão no meio. Se o fluxo é sempre

recuperar -> montar -> gerar, e o único "se" é "se não achou nada, diga que não

achou", você tem uma função. Um if resolve em uma linha; um grafo resolve em um

arquivo. Aurre aqui não é "flexibilidade para o futuro" — é custo agora por

flexibilidade que talvez você nunca use. Se a única decisão do fluxo é "tem resultado

ou não", essa decisão cabe em um if e em um early return.

2. Você não tem como medir se o ciclo ajudou. Um agente com busca ruim produz uma

resposta ruim que pode estar em três lugares: na recuperação, na montagem ou no

ciclo. Aí você otimiza o prompt, que é a única parte rápida de editar, e o prompt

nunca foi o problema. Sem um conjunto de 50 perguntas com resposta esperada (o

artigo 11), não existe "o ciclo ajudou":

existe "o ciclo rodou". Não coloque o agente antes da medição — com a medição primeiro,

você descobre que 80% das perguntas não precisam de ciclo, e o agente é um luxo de 20%

que você pode nem ter.

3. A tarefa é idempotente e barata de repetir. Se refazer tudo do zero custa menos

que guardar o estado e retomar — indexar 10 documentos, processar uma fila de e-mails,

gerar um relatório noturno — então o checkpointer é puro custo: você paga a

persistência, a complexidade e a superfície de bug do estado, para substituir um for

que roda de novo em dois segundos. A retomada só paga quando a unidade é cara

(uma chamada de modelo) ou o tempo entre as etapas é longo (um humano). Se a unidade é

barata e a etapa é curta, repetir é mais barato que lembrar.

💡 Se você ainda está em dúvida, comece por add_sequence****, não por um ciclo
Dá para usar o LangGraph só para o que ele faz de melhor — estado tipado e

checkpoint — com um fluxo linear de nós e sem nenhum add_conditional_edges.

Se depois de um mês o fluxo ainda for linear, você usou a ferramenta certa e

não precisou do ciclo. Se apareceu uma decisão real, ela já tem onde entrar.

Multiagente (vários agentes com estado próprio) quase nunca é o que você quer:

o que resolve o caso real é um grafo com state e alguns nós.

---

TL;DR

• LangGraph resolve controle, não qualidade: state, ciclo, checkpoint e humano no

  meio. Nenhum dos quatro deixa a resposta melhor, e é por isso que a medição vem

  antes do agente.

• Uma cadeia linear (recuperar -> montar -> gerar) é uma função. A pergunta que

  justifica o grafo é "qual das quatro ausências eu tenho?", não "o meu caso é

  complexo o bastante?".

• O reducer (Annotated) decide como duas escritas se combinam e é escolhido no

  schema, por chave. Duas escritas na mesma chave sem reducer é InvalidUpdateError —

  e o erro aponta a chave, não o nó.

• Checkpoint salva por super-step, nunca no meio do nó. Ao retomar, o nó roda de

  novo e o efeito colateral antes do ponto de parada repetiu — de forma silenciosa.

  Nenhum efeito colateral antes do ponto de risco.

• O thread_id é o que amarra a retomada, e ele vem da aplicação (a conversa), não

  de um invoke novo por requisição.

• Não use agente quando a cadeia é linear, quando você não tem como medir se o

  ciclo ajudou, ou quando a tarefa é barata de repetir.

---

Referências

• Graph API — LangGraph — StateGraph, add_node, add_conditional_edges e compile, a assinatura de cada bloco deste artigo
• ReAct: Synergizing Reasoning and Acting in Language Models — o padrão pensar/observar/agir que o ciclo formaliza, e a dependência dele em a observação ser boa
• Persisting state — LangGraph — checkpointer e thread_id, o par que habilita a retomada