17. Persistência de Estado e Checkpointing

No Capítulo 15 reconstruímos o loop do agente como um StateGraph, e no Capítulo 16 usamos componentes pré-construídos e roteamento multi-ramificação para construir algo mais elaborado. Cada grafo que escrevemos até agora compartilha uma limitação: o grafo não retém seu estado.

O grafo mantém e gerencia o estado durante a duração de uma única chamada invoke(), e não um instante a mais. Quando você o chama, o LangGraph cria um estado novo, executa os nós, mescla cada valor de retorno no estado de acordo com as regras do reducer e devolve o estado final ao chamador. Uma vez que ele tenha devolvido esse estado, o grafo não o lembra mais. O próximo invoke() começa a partir de um estado totalmente novo, sem conexão com a chamada anterior.

Dois problemas decorrem disso. Primeiro, messages também faz parte do estado, então o agente não consegue lembrar nada que você disse anteriormente. Segundo, se uma execução falha no meio do caminho, tudo o que ela realizou até aquele ponto desaparece. Digamos que o terceiro nó levante uma exceção: os resultados produzidos pelos dois primeiros nós vão junto, e você tem que começar tudo de novo do início. Lá em 15.1 listamos "recuperar-se de interrupções" como uma das razões para recorrer ao LangGraph — este é o problema que tínhamos em mente.

O LangGraph lida com isso no nível do framework. Anexe um checkpointer a um grafo e o LangGraph salvará automaticamente um snapshot do estado a cada etapa de execução. Esses snapshots salvos sobrevivem à chamada invoke(), então a próxima chamada pode continuar de onde a última parou. Essa propriedade—o estado sobrevivendo além de uma única execução—é chamada de persistência.

Você na verdade já usou um checkpointer antes. No Capítulo 11, quando demos ao agente RAG conversacional memória multi-turno, passamos create_agent(..., checkpointer=InMemorySaver()) e um thread_id. Na ocasião, tudo o que você precisava saber era que o checkpointer mantém o histórico de conversa por thread_id; nunca explicamos como. E no Capítulo 16, quando introduzimos o parâmetro checkpointer, dissemos "cobriremos como isso funciona no Capítulo 17." Este é aquele capítulo.

Ele vem em três partes. Em 17.1 anexamos um checkpointer a um grafo e mantemos conversas multi-turno com thread_id. Em 17.2 abrimos os checkpoints salvos para ver o que o agente sabia em qualquer momento dado—checkpoints são sua principal ferramenta para rastrear por que um agente se comportou mal. Em 17.3 pegamos um grafo que falhou no meio da execução e o retomamos a partir de onde parou, em vez de do início.

17.1) Carregando Estado Entre Chamadas

17.1.1) Um Grafo Que Esquece

Abrimos este capítulo dizendo que um grafo não retém seu estado. Vamos confirmar isso em código.

O grafo abaixo tem o mesmo formato que o grafo say_hello de 15.2. A única diferença é que o nó retorna uma resposta do LLM em vez de uma string fixa.

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-5-mini")
 
def llm_call(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}
 
builder = StateGraph(MessagesState)
builder.add_node(llm_call)
builder.add_edge(START, "llm_call")
builder.add_edge("llm_call", END)
 
graph = builder.compile()   # sem checkpointer

Agora vamos ter uma conversa de dois turnos. Dizemos ao modelo nosso nome na primeira chamada e depois perguntamos qual é nosso nome na segunda.

# Primeira chamada — informamos nosso nome
graph.invoke({"messages": [{"role": "user", "content": "Oi, meu nome é Bob."}]})
 
# Segunda chamada — perguntamos de volta
result = graph.invoke({"messages": [{"role": "user", "content": "Qual é meu nome?"}]})
print(result["messages"][-1].content)

Saída:

Desculpe, mas não sei seu nome. Você poderia me dizer qual é?

A única mensagem que o modelo recebeu na segunda chamada foi "Qual é meu nome?". O estado da primeira chamada não está mais no grafo, então as mensagens anteriores—as que carregavam o nome—nunca chegaram ao modelo.

Poderíamos, é claro, resolver isso por conta própria. Guardar as messages retornadas pela primeira chamada e passá-las junto com a segunda. É exatamente assim que gerenciamos o histórico de conversa lá no Capítulo 8. Mas então estaríamos escrevendo nosso próprio código para armazenar e recuperar o histórico para cada conversa e cada usuário. Esse é o trabalho que um checkpointer tira de suas mãos.

17.1.2) Checkpoints e Checkpointers

Um checkpointer é um objeto cujo trabalho é salvar estado. Você cria uma instância—InMemorySaver(), por exemplo—e a passa para builder.compile(checkpointer=...) para anexá-la ao seu grafo.

Uma vez que um checkpointer é anexado, o grafo copia todo o estado e o salva à medida que a execução avança. Cada uma dessas cópias salvas é chamada de checkpoint. Pense nisso como uma fotografia: o estado inteiro naquele instante, preservado exatamente como estava.

Um autosave em um videogame é a imagem mental correta. O jogo registra silenciosamente seu progresso sempre que você passa por um ponto significativo, para que você possa sair e voltar mais tarde, ou morrer sem ter que recomeçar do início. Um checkpointer faz exatamente isso para um grafo.

Então, quando é um "ponto significativo"? O LangGraph divide a execução de um grafo em estágios, e cada estágio é chamado de super-step. Um checkpoint é salvo toda vez que um super-step termina.

A razão de ser um super-step e não apenas um step é que um único estágio pode executar vários nós ao mesmo tempo. Em um grafo como o de 17.1.1, onde os nós formam uma linha reta, executar um nó é um super-step. Mas em um grafo onde vários nós executam em paralelo, todos esses nós juntos formam um único super-step.

Então, mesmo uma única chamada a invoke() deixa vários checkpoints para trás. Vamos extraí-los e olhar exatamente o que cada um contém em 17.2.

O InMemorySaver que temos usado como nosso exemplo é o checkpointer mais simples que existe. Como o nome sugere, ele armazena checkpoints na memória do processo (RAM). Não há nada para instalar e nada para configurar, o que o torna uma boa escolha para aprendizado e desenvolvimento local. O trade-off é que todo checkpoint salvo desaparece quando o processo reinicia. Vamos olhar as alternativas de produção em 17.1.5.

17.1.3) Adicionando um Checkpointer

Anexar um checkpointer requer apenas duas coisas.

1. Criar uma instância de checkpointer e passá-la para compile().
2. Passar um config contendo um thread_id sempre que você chamar invoke().

Chegaremos ao porquê da segunda ser necessária em breve. Por ora, apenas saiba que ela diz ao checkpointer qual de suas conversas salvas você quer continuar.

Vamos aplicar ambas ao grafo de 17.1.1.

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-5-mini")
 
def llm_call(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}
 
builder = StateGraph(MessagesState)
builder.add_node(llm_call)
builder.add_edge(START, "llm_call")
builder.add_edge("llm_call", END)
 
# 1. Cria um checkpointer e o passa para compile()
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
 
# 2. Passa um config carregando um thread_id para invoke()
config = {"configurable": {"thread_id": "1"}}
 
graph.invoke(
    {"messages": [{"role": "user", "content": "Oi, meu nome é Bob."}]},
    config,
)
result = graph.invoke(
    {"messages": [{"role": "user", "content": "Qual é meu nome?"}]},
    config,
)
print(result["messages"][-1].content)

Saída:

Seu nome é Bob.

As mesmas duas chamadas de 17.1.1, e um resultado diferente. Desta vez o nome permanece.

Eis o porquê. Um grafo com um checkpointer anexado carrega o estado salvo antes de executar o nó llm_call. Esse estado já contém a primeira troca. A nova mensagem que passamos é então mesclada nele. Como vimos em 15.2.2, o campo messages carrega o reducer add_messages, então a nova mensagem é anexada à lista existente. O modelo acaba recebendo três mensagens: a saudação, sua própria primeira resposta e a nova pergunta.

Enviamos apenas a nova mensagem, e o LangGraph carrega a conversa anterior a partir do último checkpoint. O histórico de conversa que gerenciamos manualmente no Capítulo 8 agora é gerenciado pelo LangGraph.

17.1.4) thread_id: O Identificador Que Separa Conversas

Um thread_id é um identificador que distingue uma conversa de outra. Você escolhe o valor. Usamos "1" em 17.1.3, mas qualquer string serve. Chame o grafo com o mesmo thread_id e você continua aquela conversa; chame-o com um diferente e você inicia uma conversa separada.

Vamos confirmar isso. Faremos Alice e Bob manterem conversas diferentes através do mesmo grafo.

def send(thread_id: str, text: str) -> str:
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke(
        {"messages": [{"role": "user", "content": text}]},
        config,
    )
    return result["messages"][-1].content
 
# Conversa da Alice
send("alice", "Minha cor favorita é verde-azulado.")
 
# Conversa do Bob — um thread_id diferente
send("bob", "Minha cor favorita é laranja.")
 
# Pergunta a cada um deles novamente
print("Alice:", send("alice", "Qual é minha cor favorita?"))
print("Bob:  ", send("bob", "Qual é minha cor favorita?"))

Saída:

Alice: Sua cor favorita é verde-azulado.
Bob:   Sua cor favorita é laranja.

Ambas as conversas passaram pelo mesmo objeto graph e pelo mesmo checkpointer, mas nunca se misturaram. O thread_id é a chave primária que o checkpointer usa para armazenar e buscar estado. Chaves diferentes, armazenamento completamente separado.

Então o que acontece se você deixar o valor de fora?

graph.invoke({"messages": [{"role": "user", "content": "Olá"}]})

Saída:

ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id

O grafo não executa de forma alguma. Uma vez que um checkpointer é anexado, thread_id não é opcional—é obrigatório.

Dado o que acabamos de ver, isso faz sentido. Antes de executar um nó, o checkpointer tem que carregar o estado salvo — e o thread_id é o que diz a ele o estado de qual conversa carregar.

Este é o formato básico de um serviço de chatbot: um grafo, um checkpointer e um thread_id por usuário ou por sala de chat.

17.1.5) Os Limites do InMemorySaver e as Alternativas de Produção

Dissemos anteriormente que o InMemorySaver mantém checkpoints na memória. Duas limitações vêm com essa escolha.

Reinicie o processo e tudo se vai. Reimplante o serviço ou reinicie o servidor, e toda conversa acumulada até então desaparece.

Processos separados não podem compartilhá-lo. Um serviço real distribui as requisições recebidas entre vários processos. Cada processo tem sua própria memória, então uma conversa salva pelo processo A é invisível para o processo B. Um usuário pode enviar o mesmo thread_id toda vez e ainda assim ver a conversa desmoronar, dependendo de qual processo por acaso pega a requisição.

É por isso que a produção usa checkpointers que armazenam checkpoints em um banco de dados.

• SqliteSaver / AsyncSqliteSaver (langgraph-checkpoint-sqlite) — armazena tudo em um único arquivo. Uma boa escolha para um serviço pequeno rodando em um servidor, ou para um protótipo local.
• PostgresSaver / AsyncPostgresSaver (langgraph-checkpoint-postgres) — armazena checkpoints em um servidor de banco de dados. Adicione mais servidores e todo processo ainda verá os mesmos checkpoints. Este é o checkpointer que a documentação do LangGraph recomenda para produção.

Todos eles implementam a mesma interface que o InMemorySaver. O código do seu grafo, seus nós e a forma como você usa thread_id permanecem exatamente como estão. A única coisa que muda é como você cria o checkpointer.

O create_agent, que você conheceu no Capítulo 16, usa checkpointers da mesma forma. Passe um checkpointer para seu parâmetro checkpointer e passe um config carregando um thread_id para invoke(). Foi isso que fez o agente RAG conversacional no Capítulo 11 lembrar dos turnos anteriores.

Continuaremos usando o InMemorySaver pelo resto deste capítulo. Onde quer que os checkpoints vivam, a forma como você trabalha com eles é a mesma.

17.2) Inspecionando Estado e Depurando

Em 17.1.2 dissemos que o checkpointer salva estado a cada super-step. Vamos extrair esses checkpoints e olhá-los.

Abrir os checkpoints salvos não é curiosidade ociosa. Agentes se comportam mal. Eles chamam ferramentas em um loop que nunca termina, perdem a noção de turnos anteriores, tomam uma ramificação que você nunca esperou. Para descobrir o porquê, você precisa saber como o estado parecia naquele momento—e os checkpoints têm a resposta.

O LangGraph lhe dá dois métodos.

• graph.get_state(config) — retorna o checkpoint mais recente para aquela conversa.
• graph.get_state_history(config) — retorna todos os checkpoints para aquela conversa, do mais novo ao mais antigo.

Ambos precisam de um thread_id no config, já que precisam saber de quem são os checkpoints que você está solicitando.

Os checkpoints que esses dois métodos devolvem são representados como objetos StateSnapshot.

17.2.1) StateSnapshot: O Que Há Dentro de um Checkpoint

Vamos ver o que um checkpoint realmente contém. Não usaremos um LLM para este exemplo. Um LLM retorna algo diferente toda vez, o que não ajuda quando queremos percorrer cada campo um por um. Em vez disso, construiremos um pequeno grafo que retorna valores fixos.

O estado terá dois tipos de campos. foo não tem reducer, então é sobrescrito; bar tem um, então acumula. É o mesmo arranjo que o reducer add_messages que anexamos a messages em 15.2.2—aqui usamos o operator.add do Python como reducer para concatenar listas.

from operator import add
from typing_extensions import TypedDict, Annotated
 
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
 
class State(TypedDict):
    foo: str                        # sem reducer → sobrescreve
    bar: Annotated[list[str], add]  # reducer add → acumula
 
def node_a(state: State):
    return {"foo": "a", "bar": ["a"]}
 
def node_b(state: State):
    return {"foo": "b", "bar": ["b"]}
 
builder = StateGraph(State)
builder.add_node(node_a)
builder.add_node(node_b)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)
 
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": "", "bar": []}, config)
 
snapshot = graph.get_state(config)
print(snapshot)

Saída:

StateSnapshot(
    values={'foo': 'b', 'bar': ['a', 'b']},
    next=(),
    config={'configurable': {'thread_id': '1', 'checkpoint_ns': '',
                             'checkpoint_id': '1f17da03-9654-65d8-8002-9e59231bb481'}},
    metadata={'source': 'loop', 'step': 2, 'parents': {}},
    created_at='2026-07-12T03:17:33.637368+00:00',
    parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '',
                                    'checkpoint_id': '1f17da03-9653-6ca0-8001-7c24c8ec66f2'}},
    tasks=(),
    interrupts=()
)

Oito campos. Vamos abordá-los um de cada vez.

• values — o estado como ele estava neste checkpoint. bar é ['a', 'b'] porque o reducer add acumulou o que ambos os nós retornaram; foo é 'b' porque não tem reducer, então a última escrita venceu. Este é o campo que responde "como o estado parecia naquele momento?"
• next — uma tupla de nomes de nós a serem executados após este checkpoint. Um () vazio significa que não há mais nada a executar, ou seja, o grafo terminou. ('node_b',) significa que node_b ainda está à frente.
• config — o endereço deste checkpoint. thread_id identifica a conversa e checkpoint_id identifica qual momento dentro dela. O LangGraph atribui o checkpoint_id automaticamente toda vez que salva um checkpoint.
• metadata — informações contábeis sobre a execução. source diz de onde o checkpoint veio: "input" significa que foi construído a partir da entrada que você passou para invoke(), e "loop" significa que foi produzido enquanto o grafo estava em execução. step é o número do super-step.
• created_at — quando o checkpoint foi salvo. Útil quando você está alinhando as coisas com seus logs.
• parent_config — o config do checkpoint imediatamente anterior a este. Siga-o e você pode andar para trás pela execução. É None para o primeiro checkpoint.
• tasks — o registro de execução para os nós listados em next. No momento em que um checkpoint é salvo, esses nós ainda não foram executados; uma vez que sejam, seu resultado é anexado a este checkpoint. Um nó que teve sucesso deixa seu valor de retorno em result, e um nó que falhou deixa sua exceção em error.
• interrupts — onde o grafo pausou para devolver o controle a uma pessoa. O LangGraph pode parar no meio da execução e esperar que alguém aprove uma etapa ou forneça um valor, e este campo registra essas pausas.

Você lê os campos como atributos. metadata é um dicionário, então você extrai valores dele com uma chave.

snapshot = graph.get_state(config)
 
print(snapshot.values)            # {'foo': 'b', 'bar': ['a', 'b']}
print(snapshot.next)              # ()
print(snapshot.metadata["step"])  # 2

Destes, aquele que você mais frequentemente recorrerá durante a depuração é next. Se next não estiver vazio, o grafo não chegou ao fim—ele parou em algum lugar no meio. E quando retomamos um grafo que falhou em 17.3, este campo é onde começamos.

17.2.2) Percorrendo o Histórico de Checkpoints

get_state() mostra apenas o checkpoint mais recente. Mas depurar frequentemente significa perguntar "como chegamos aqui?"—e para isso você precisa de toda a trajetória da execução. get_state_history() a fornece para você.

for snap in graph.get_state_history(config):
    print(f"step={snap.metadata['step']:>2}  "
          f"next={str(snap.next):<16}  values={snap.values}")

Saída:

step= 2  next=()                values={'foo': 'b', 'bar': ['a', 'b']}
step= 1  next=('node_b',)       values={'foo': 'a', 'bar': ['a']}
step= 0  next=('node_a',)       values={'foo': '', 'bar': []}
step=-1  next=('__start__',)    values={'bar': []}

O checkpoint mais novo vem primeiro, então trabalhe de baixo para cima para seguir a execução em ordem.

• step -1 — logo após invoke() receber a entrada. Note que o {"foo": "", "bar": []} que passamos não aparece em values. Colocar a entrada no estado é, em si, um estágio, e esse estágio ainda não foi executado. O __start__ em next é o nó interno que faz isso.

  bar aparece como [], mas esse não é o valor que passamos. Um campo com reducer começa com um valor vazio para as escritas acumularem nele. foo não tem reducer, então não tem valor inicial algum — que é por isso que não aparece aqui.

• step 0 — __start__ foi executado e a entrada agora está no estado. foo='' e bar=[] são os valores que passamos. node_a é o próximo.
• step 1 — o resultado de executar node_a. foo agora é 'a' e bar é ['a'], com node_b a seguir.
• step 2 — o resultado de executar node_b. next está vazio, então o grafo terminou.

Um next não vazio no step 0 ou step 1 não significa que o grafo parou ali. Um checkpoint tirado enquanto o grafo ainda está em execução naturalmente tem um nó alinhado a seguir. Quando 17.2.1 disse "um next não vazio significa que o grafo parou," estava falando sobre o checkpoint no ponto onde algo deu errado. No meio do histórico, next simplesmente mostra qual caminho o grafo tomou.

17.3) Retomando de Onde Falhou

Como o estado é salvo ao final de cada super-step, uma falha no meio da execução não leva o trabalho concluído junto—ele ainda está lá nos checkpoints. Não há razão para começar tudo de novo do início. Você continua de onde parou.

17.3.1) Retomando com invoke(None, config)

Retomar é simples: passe None onde a entrada vai.

graph.invoke(None, config)

Isso significa "não há nova entrada; continue a partir do estado salvo." O config ainda precisa de um thread_id, é claro, já que o LangGraph tem que saber qual conversa continuar.

Vamos encenar uma falha e retomar dela. Construiremos um grafo de dois nós cujo segundo nó falha apenas na primeira execução. Ele tem que ter sucesso na nova tentativa, ou nunca veríamos a retomada realmente funcionando.

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
 
class State(TypedDict):
    step_1_done: bool
    step_2_done: bool
 
first_try = True   # flag para fazer a primeira execução falhar
 
def step_1(state: State):
    print("step_1 executando (trabalho caro)")
    return {"step_1_done": True}
 
def step_2(state: State):
    global first_try
    if first_try:
        first_try = False
        print("step_2 falhou (timeout da API)")
        raise RuntimeError("External API timed out")
    print("step_2 executando")
    return {"step_2_done": True}
 
builder = StateGraph(State)
builder.add_node(step_1)
builder.add_node(step_2)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", END)
 
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "job-42"}}
 
try:
    graph.invoke({"step_1_done": False, "step_2_done": False}, config)
except RuntimeError as e:
    print("Falhou:", e)

Saída:

step_1 executando (trabalho caro)
step_2 falhou (timeout da API)
Falhou: External API timed out

step_1 teve sucesso e step_2 levantou uma exceção. Vamos descobrir onde o grafo parou, usando o get_state() que aprendemos em 17.2.

snapshot = graph.get_state(config)
print("next   =", snapshot.next)
print("values =", snapshot.values)

Saída:

next   = ('step_2',)
values = {'step_1_done': True, 'step_2_done': False}

next é ('step_2',), o que nos diz que o grafo parou no meio da execução de step_2. E em values, step_1_done é True—o resultado de step_1 ainda está lá no checkpoint.

Agora retomamos com None.

result = graph.invoke(None, config)
print("Final =", result)

Saída:

step_2 executando
Final = {'step_1_done': True, 'step_2_done': True}

step_1 executando (trabalho caro) nunca foi impresso. step_1 não executou uma segunda vez. O LangGraph carregou o estado salvo e continuou em step_2. Não pagamos por aquela primeira etapa cara duas vezes.

17.3.2) O Que Observar ao Retomar

Quando você retoma, o nó que falhou executa novamente. Se aquele nó chama um LLM ou acessa uma API externa, essas chamadas acontecem novamente também—e elas podem retornar algo diferente.

É aí que está a armadilha. Se step_2 enviou um e-mail e depois falhou, retomar envia um segundo e-mail. O LangGraph garante apenas que não vai reexecutar os nós que tiveram sucesso.

Então qualquer nó que possa executar novamente precisa ser idempotente: fazer a mesma coisa duas vezes deve deixá-lo no mesmo lugar. Verifique se o e-mail já foi enviado antes de enviá-lo; coloque uma chave única na tabela do banco de dados para que uma inserção duplicada não possa acontecer.