LangGraph e PostgreSQL: como testar retomadas e controlar checkpoints

Persistir o estado de um workflow não basta para demonstrar que ele sobreviverá a uma falha. O teste importante acontece quando o processo termina depois de realizar uma operação de negócio, mas antes de registrar que aquela etapa terminou. Ao mesmo tempo, manter todo o histórico indefinidamente não é uma estratégia de capacidade.

Considere uma plataforma interna que prepara mudanças de infraestrutura, aguarda aprovação e registra sua execução. Neste cenário proposto, precisamos provar duas propriedades: a mudança não será aplicada duas vezes após uma retomada e a limpeza não eliminará uma aprovação ainda pendente.

Este guia usa a biblioteca Python do LangGraph com PostgresSaver diretamente. Não pressupõe serviços de gerenciamento de threads do Agent Server, cuja persistência é administrada pelo próprio servidor. Checkpointers guardam estado por thread; stores atendem dados compartilhados entre threads. São responsabilidades diferentes. (docs.langchain.com)

1. Entenda o que está sendo persistido

Não confunda os checkpoints do workflow com os checkpoints internos do mecanismo PostgreSQL. Aqui, o objeto de retenção é o estado da aplicação armazenado em tabelas.

Os identificadores essenciais são:

• thread_id: identifica a continuidade de uma execução ou conversa.
• checkpoint_id: identifica um ponto persistido.
• checkpoint_ns: distingue namespaces, incluindo os utilizados por subgrafos.
• Superstep: etapa de execução que pode envolver múltiplos nós.

A durabilidade também precisa ser explícita: sync persiste antes de avançar; async permite persistência em paralelo à próxima etapa; exit persiste na saída da execução, inclusive interrupções, mas não oferece recuperação intermediária após uma queda abrupta. Neste laboratório, use sync. (docs.langchain.com)

Na implementação PostgreSQL consultada, checkpoints contém snapshots e metadados, checkpoint_blobs armazena valores de canais e checkpoint_writes mantém escritas associadas às tarefas. checkpoint_migrations registra migrações. Há referências entre checkpoints, versões de canais e blobs; uma linha não representa necessariamente todo o estado necessário para restaurar aquele ponto. (github.com)

Retomada não é replay

Uma interrupção dinâmica usa interrupt() e continua com Command(resume=...), mantendo o mesmo thread_id. O nó interrompido começa novamente; portanto, o código anterior ao interrupt() pode executar outra vez. Não envolva essa chamada em um tratamento genérico que engula sua exceção de controle. (docs.langchain.com)

Após uma falha, a continuação de trabalho pendente pode usar graph.invoke(None, config). Isso é diferente de enviar novamente o input inicial. Já selecionar um checkpoint histórico permite replay: nós posteriores podem executar novamente, incluindo chamadas externas. (docs.langchain.com)

2. Arquitetura proposta

Cliente → API autenticada → coordenador por thread → worker LangGraph
                                                    ↓
                                              PostgresSaver
                                                    ↓
                                              PostgreSQL HA

Worker → operação idempotente → sistema de negócio
Controlador de retenção → catálogo de threads → exclusão autorizada

Recomendo separar três responsabilidades:

1. Execução: avançar o grafo e persistir seu estado.
2. Coordenação: impedir avanço concorrente incompatível da mesma thread e bloquear retomadas durante exclusão.
3. Retenção: decidir quando o histórico deixa de ser necessário.

O checkpointer não deve ser tratado como autorização de negócio. A API precisa verificar tenant, identidade do aprovador, estado atual e validade da decisão antes de aceitar uma retomada.

No cenário de mudanças de infraestrutura, a sequência proposta é: preparar → aprovar → registrar/aplicar idempotentemente → encerrar. O resultado esperado é demonstrável por testes, não presumido pela presença do PostgreSQL.

3. Laboratório: interromper, matar o processo e retomar

Use um banco descartável, acessível por DATABASE_URL, e Python compatível com as versões escolhidas. Os comandos abaixo são bootstrap de laboratório; depois de validar a combinação, fixe dependências e hashes no processo de build.

python -m venv .venv
. .venv/bin/activate
python -m pip install langgraph langgraph-checkpoint-postgres 'psycopg[binary]'
python -m pip check
python -m pip freeze > requirements-tested.txt

O método setup() prepara o schema do saver. Execute-o como etapa controlada de migração, não a cada requisição. Ao construir conexões manualmente para o saver, observe os requisitos documentados de autocommit=True e row_factory=dict_row. O exemplo utiliza o gerenciador de contexto from_conn_string(). (pypi.org)

Salve como recovery_demo.py:

import os
import sys
from typing import TypedDict

import psycopg
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

DB = os.environ['DATABASE_URL']
THREAD = os.environ['THREAD_ID']
CONFIG = {'configurable': {'thread_id': THREAD}}


class State(TypedDict):
    operation_id: str
    approved: bool
    status: str


def approval(state: State):
    answer = interrupt({'operation_id': state['operation_id']})
    if not isinstance(answer, bool):
        raise ValueError('Approval must be boolean')
    return {'approved': answer}


def apply_change(state: State):
    if not state['approved']:
        return {'status': 'rejected'}

    # The committed row IS the business effect in this laboratory.
    with psycopg.connect(DB) as conn:
        conn.execute(
            'INSERT INTO demo_effects (operation_id) VALUES (%s) '
            'ON CONFLICT (operation_id) DO NOTHING',
            (state['operation_id'],),
        )

    # Crash after the business commit, before the node returns.
    if os.environ.get('FAIL_AFTER_COMMIT') == '1':
        os._exit(17)
    return {'status': 'done'}


builder = StateGraph(State)
builder.add_node('approval', approval)
builder.add_node('apply_change', apply_change)
builder.add_edge(START, 'approval')
builder.add_edge('approval', 'apply_change')
builder.add_edge('apply_change', END)

mode = sys.argv[1]
with PostgresSaver.from_conn_string(DB) as saver:
    if mode == 'setup':
        saver.setup()
        with psycopg.connect(DB) as conn:
            conn.execute(
                'CREATE TABLE IF NOT EXISTS demo_effects ('
                'operation_id TEXT PRIMARY KEY)'
            )
    else:
        graph = builder.compile(checkpointer=saver)
        if mode == 'start':
            result = graph.invoke(
                {'operation_id': THREAD, 'approved': False,
                 'status': 'pending'},
                CONFIG, durability='sync',
            )
            assert result['__interrupt__']
            assert graph.get_state(CONFIG).next == ('approval',)
        elif mode == 'resume':
            graph.invoke(Command(resume=True), CONFIG, durability='sync')
        elif mode == 'recover':
            before = graph.get_state(CONFIG)
            assert before.values['approved'] is True
            assert before.next == ('apply_change',)
            result = graph.invoke(None, CONFIG, durability='sync')
            assert result['status'] == 'done'
            assert graph.get_state(CONFIG).next == ()
            with psycopg.connect(DB) as conn:
                count = conn.execute(
                    'SELECT count(*) FROM demo_effects '
                    'WHERE operation_id = %s', (THREAD,),
                ).fetchone()[0]
            assert count == 1
            print('PASS: recovered with exactly one stored effect')
        else:
            raise ValueError('Unknown mode')

Execute cada fase em um processo diferente:

set -eu
: "${DATABASE_URL:?Inject a disposable database connection string}"
export THREAD_ID="$(python -c 'import uuid; print(uuid.uuid4())')"

python recovery_demo.py setup
python recovery_demo.py start

rc=0
FAIL_AFTER_COMMIT=1 python recovery_demo.py resume || rc=$?
test "$rc" -eq 17

python recovery_demo.py recover

O teste verifica uma interrupção persistida, aprovação recuperada por outro processo, queda abrupta após commit e conclusão sem duplicar a linha de negócio. Ele não depende de memória do processo anterior.

Limite importante: a linha é o próprio efeito neste laboratório. Inserir uma chave local e depois chamar uma API não torna a API idempotente. Para operações externas, recomendo uma chave estável aceita pelo destino, reconciliação de resultado ou um padrão outbox com consumidor idempotente. A chave deve representar a operação de negócio, não uma tentativa nem um checkpoint novo.

Este exemplo não demonstra comportamento sob failover do banco, múltiplos escritores ou indisponibilidade de rede. Esses casos exigem testes adicionais.

Matriz mínima de aceitação

O reaproveitamento de trabalho de nós concluídos no mesmo superstep depende das pending writes preservadas pelo mecanismo de checkpoint. Por isso, inclua falhas em ramos paralelos na suíte. (docs.langchain.com)

4. Meça crescimento antes de apagar

Comece pelo espaço físico total das relações, incluindo índices e armazenamento TOAST, usado pelo PostgreSQL para valores grandes. Adapte o schema:

SELECT c.relname,
       pg_total_relation_size(c.oid) AS total_bytes,
       pg_size_pretty(pg_total_relation_size(c.oid)) AS total_pretty
FROM pg_class AS c
JOIN pg_namespace AS n ON n.oid = c.relnamespace
WHERE n.nspname = 'public'
  AND c.relname IN (
    'checkpoints', 'checkpoint_blobs', 'checkpoint_writes'
  )
ORDER BY total_bytes DESC;

pg_total_relation_size é apropriado para acompanhar a ocupação total de cada tabela, não apenas seu heap. A soma acima não inclui todos os custos do banco, como WAL e backups. (postgresql.org)

Observe também a atividade de manutenção:

SELECT relname, n_live_tup, n_dead_tup,
       last_autovacuum, last_autoanalyze
FROM pg_stat_user_tables
WHERE schemaname = 'public'
  AND relname IN (
    'checkpoints', 'checkpoint_blobs', 'checkpoint_writes'
  );

Os contadores de tuplas são estimativas, não uma medição exata de bloat. Correlacione-os com crescimento, carga e duração das transações. (postgresql.org)

Para capacity planning, proponho medir bytes incrementais por thread encerrada sob uma carga representativa:

ocupação estimada = threads/dia × bytes incrementais/thread × dias retidos
                   + threads ativas + índices + folga operacional

Evite contabilizar índices duas vezes se a medição incremental já os inclui. Separe orçamento para Write-Ahead Log (WAL), réplicas, backups e espaço de manutenção.

Como exemplo hipotético, 10 mil threads diárias com incremento médio de 200 KiB representam aproximadamente 1,9 GiB/dia antes de componentes não incluídos na medição. Isso não é um benchmark do LangGraph.

5. Controle crescimento na origem

Recomendo quatro limites de projeto:

1. Guardar referências imutáveis para documentos grandes, acompanhadas de versão e checksum.
2. Limitar resultados de ferramentas e o tamanho máximo dos inputs.
3. Encerrar threads por unidade de trabalho, em vez de reutilizar uma thread global indefinidamente.
4. Definir uma política explícita para histórico de mensagens e evidências de auditoria.

Reduzir mensagens no estado corrente não apaga automaticamente os checkpoints históricos. Trate compactação do estado e expiração do histórico como operações distintas. As APIs de gerenciamento de mensagens atuam sobre o estado; a remoção de checkpoints exige tratamento próprio. (docs.langchain.com)

Também não troque sync por exit apenas para economizar armazenamento sem aprovar a mudança no contrato de recuperação.

6. Retenção segura: comece por threads inteiras

Minha recomendação inicial é excluir threads encerradas e expiradas, em vez de implementar SQL para conservar arbitrariamente as últimas N linhas.

Mantenha um catálogo da aplicação com tenant_id, thread_id, status, última atividade, expires_at, versão do grafo e indicação de retenção obrigatória. Esses campos são uma proposta de schema da aplicação, não colunas nativas do saver.

Protocolo recomendado

1. Selecionar um lote pequeno de candidatas encerradas e expiradas.
2. Excluir candidatas sob retenção obrigatória ou aprovação pendente.
3. Adquirir a mesma coordenação usada pelos executores.
4. Marcar a thread como deleting e impedir novas admissões.
5. Drenar execuções existentes e confirmar que não há escritores autorizados.
6. Excluir pelo saver.
7. Verificar ausência dos dados e registrar conclusão.
8. Manter um tombstone pelo prazo necessário para rejeitar entregas atrasadas.

O método público delete_thread(thread_id) remove checkpoints e escritas da thread; a implementação PostgreSQL também remove seus blobs. Isso não exclui automaticamente dados de um store independente ou do sistema de negócio. (reference.langchain.com)

Exemplo da etapa de exclusão, somente depois de aplicar o protocolo acima:

import os
from langgraph.checkpoint.postgres import PostgresSaver

thread_id = os.environ['EXPIRED_FENCED_THREAD_ID']
config = {'configurable': {'thread_id': thread_id}}

with PostgresSaver.from_conn_string(os.environ['DATABASE_URL']) as saver:
    saver.delete_thread(thread_id)
    assert saver.get_tuple(config) is None

A variável é apenas um input operacional, não um mecanismo de fencing. Um lock usado somente pelo coletor não protege contra workers que o ignoram. Projete reexecução segura quando o coletor falhar entre apagar dados e atualizar o catálogo.

E conservar somente o checkpoint mais recente?

A implementação consultada no branch main possui prune(..., strategy='keep_latest'), mas isso não comprova disponibilidade em qualquer release instalada. Ela preserva dependências necessárias, possui regras para namespaces e recusa threads com DeltaChannel. Confirme a API da versão fixada e seus limites antes de utilizá-la. Não transplante SQL interno para produção. (github.com)

Se replay histórico é requisito, conservar somente o último ponto é uma decisão funcional, não apenas uma otimização de armazenamento.

7. DELETE, VACUUM e Kubernetes

Após DELETE, espaço pode permanecer alocado. O VACUUM comum disponibiliza espaço para reutilização; em geral, não reduz o arquivo ao tamanho esperado pelo operador. VACUUM FULL reescreve a tabela, exige bloqueio exclusivo e espaço adicional. Não o transforme em rotina automática sem planejamento. (postgresql.org)

O manifesto abaixo é um modelo de manutenção manualmente habilitada, não um coletor de retenção. Ele complementa, mas não substitui, autovacuum:

apiVersion: batch/v1
kind: CronJob
metadata:
  name: langgraph-checkpoint-vacuum
  namespace: platform
spec:
  schedule: "30 3 * * *"
  timeZone: "Etc/UTC"
  suspend: true
  concurrencyPolicy: Forbid
  startingDeadlineSeconds: 600
  successfulJobsHistoryLimit: 1
  failedJobsHistoryLimit: 3
  jobTemplate:
    spec:
      backoffLimit: 1
      activeDeadlineSeconds: 900
      template:
        spec:
          restartPolicy: Never
          automountServiceAccountToken: false
          securityContext:
            runAsNonRoot: true
            runAsUser: 10001
            runAsGroup: 10001
            seccompProfile:
              type: RuntimeDefault
          containers:
            - name: vacuum
              image: postgres:18
              command: ["psql"]
              args:
                - "-X"
                - "-v"
                - "ON_ERROR_STOP=1"
                - "-c"
                - "VACUUM (ANALYZE) public.checkpoints;"
                - "-c"
                - "VACUUM (ANALYZE) public.checkpoint_blobs;"
                - "-c"
                - "VACUUM (ANALYZE) public.checkpoint_writes;"
              envFrom:
                - secretRef:
                    name: langgraph-maintenance-db
              env:
                - name: PGSSLMODE
                  value: verify-full
                - name: PGSSLROOTCERT
                  value: /etc/postgres-ca/ca.crt
                - name: PGCONNECT_TIMEOUT
                  value: "10"
              resources:
                requests:
                  cpu: 100m
                  memory: 64Mi
                limits:
                  cpu: 500m
                  memory: 256Mi
              securityContext:
                allowPrivilegeEscalation: false
                readOnlyRootFilesystem: true
                capabilities:
                  drop: ["ALL"]
              volumeMounts:
                - name: postgres-ca
                  mountPath: /etc/postgres-ca
                  readOnly: true
          volumes:
            - name: postgres-ca
              configMap:
                name: postgres-ca

Antes de usar, crie o namespace, o Secret com PGHOST, PGPORT, PGDATABASE, PGUSER e PGPASSWORD, e o ConfigMap com ca.crt. Ajuste schema, janela, recursos, permissões de manutenção e fixe a imagem por digest aprovado. Os limites do container não limitam a carga de VACUUM executada no servidor PostgreSQL.

suspend: true exige habilitação deliberada. concurrencyPolicy: Forbid evita sobreposição dos Jobs deste CronJob, não coordena workers nem outras rotinas. CronJobs podem ter execuções duplicadas ou ausentes em determinadas circunstâncias; mantenha tarefas reexecutáveis. (kubernetes.io)

Este Job finito usa timeout e código de saída. Probes são mais úteis nos serviços persistentes: readiness controla recepção de tráfego, liveness detecta falha do processo e startup protege inicialização demorada. Não faça liveness reiniciar todos os workers apenas porque o banco está temporariamente indisponível. (kubernetes.io)

8. Produção: confiabilidade, segurança e custo

Disponibilidade e concorrência

Para produção, recomendo PostgreSQL com alta disponibilidade entre zonas e workers distribuídos, com requests, limits, encerramento gracioso e orçamento de disrupção. Dimensione autoscaling pelo trabalho pendente e pela capacidade do banco, não somente por CPU.

Use pool de conexões com limite explícito. O Psycopg oferece max_size e timeout de aquisição; estabeleça um orçamento global considerando processos, réplicas, migrações e manutenção. (psycopg.org)

Como regra de dimensionamento proposta:

conexões potenciais = pods × processos/pod × máximo do pool
                     + conexões de negócio + manutenção + reserva

Recomendo serializar mutações por thread e usar um protocolo de lease/fencing que também seja respeitado na retenção. Mais réplicas não substituem essa coordenação.

Segurança

Separe credenciais de migração, execução e manutenção. Use Transport Layer Security (TLS) com verificação de identidade, banco privado, controle de acesso baseado em papéis (RBAC) e NetworkPolicy compatível com a rede do cluster.

Não coloque credenciais no Git. Secrets Kubernetes não são, por padrão, uma garantia de criptografia em repouso; configure proteção do armazenamento e acesso mínimo. Evite conceder leitura ampla de Secrets a operadores ou workloads. (kubernetes.io)

Recomendo ainda autorização por tenant fora do thread_id, rotação de credenciais, minimização de dados pessoais e proibição de payloads completos em logs de rotina.

Backup e rollback

Point-in-Time Recovery (PITR), recuperação para um instante específico, combina backup base e sequência de WAL arquivada. Replicação não substitui essa estratégia de recuperação. (postgresql.org)

Defina Recovery Point Objective (RPO), perda máxima aceitável, e Recovery Time Objective (RTO), prazo de recuperação. Teste restauração isolada antes de habilitar exclusão automática. Uma restauração de checkpoints não desfaz efeitos externos; bloqueie chamadas reais durante a validação.

Para upgrades, recomendo manter fixtures de checkpoints antigos, testar desserialização e retomada, versionar o schema de estado e documentar compatibilidade reversa. Fazer rollback da imagem não prova que o código antigo consegue ler estados escritos pela versão nova.

FinOps e observabilidade

Trate armazenamento, operações de entrada/saída, WAL, backups, tráfego entre zonas e telemetria como componentes separados de custo. Use workers Spot apenas onde a política de recuperação tiver sido demonstrada; mantenha capacidade estável para caminhos críticos.

Instrumente métricas próprias para latência de persistência e retomada, erros por fase, espera no pool, threads pendentes, backlog de expiração, bytes por tabela e duração da limpeza. Elas são propostas de instrumentação, não nomes nativos garantidos do LangGraph.

Defina um Service Level Objective (SLO) para retomadas elegíveis concluídas corretamente dentro do prazo. Em traces, correlacione tentativa, versão do grafo e checkpoint; evite thread_id como label de alta cardinalidade em métricas.

9. Troubleshooting

Para o Job Kubernetes:

kubectl get cronjobs,jobs,pods -n platform
kubectl describe cronjob langgraph-checkpoint-vacuum -n platform
kubectl logs -n platform job/<job-name>
kubectl describe pod <pod-name> -n platform
kubectl get events -n platform --sort-by=.metadata.creationTimestamp

Não exponha a connection string ao coletar evidências. Primeiro diferencie falha do grafo, autenticação, transporte, pool, armazenamento e manutenção; depois altere configuração.

10. Desenvolvimento, alternativas e critérios de adoção

Em desenvolvimento, uma réplica e um banco descartável podem bastar. Em produção, recomendo retenção automatizada com proteção de threads, métricas, backups testados, coordenação distribuída e um procedimento de rollback.

Use esta abordagem quando workflows precisam sobreviver a reinícios, esperar aprovações ou manter continuidade por thread. Para uma tarefa curta e descartável, o custo operacional pode não se justificar.

Como alternativas, avalie persistência em memória para testes efêmeros, SQLite para execução local e uma máquina de estados transacional da própria aplicação quando o fluxo for simples. As implementações de checkpointer têm objetivos distintos; não escolha uma solução local como substituta automática de uma arquitetura compartilhada. (docs.langchain.com)

Para minha recomendação de arquitetura, PostgreSQL faz sentido quando a equipe já consegue operar sua disponibilidade, recuperação e capacidade. Um serviço gerenciado reduz parte da operação, mas deve ser avaliado contra custos, controles de retenção e dependência do fornecedor.

Resumo prático

• Teste em processos diferentes e injete falha depois do efeito de negócio.
• Separe retomada, novo input e replay histórico.
• Use identidade de operação estável e valide idempotência no destino.
• Meça tabelas, índices, WAL e backups separadamente.
• Exclua threads expiradas somente após impedir novas escritas.
• Não confunda redução do estado, retenção e recuperação de espaço físico.
• Exija testes de restauração e compatibilidade antes de upgrades.

Conclusão

O objetivo não é acumular checkpoints nem apagá-los o mais rápido possível. É manter um contrato verificável entre execução, efeitos de negócio e recuperação.

Em uma plataforma DevOps e SRE, esse contrato deve aparecer no pipeline de testes, nos limites de capacidade, nos SLOs e nos procedimentos operacionais. Quando retomada e retenção são tratadas juntas, o PostgreSQL deixa de ser apenas um destino de snapshots e passa a integrar uma estratégia explícita de confiabilidade do workflow.