Agent AI cu memorie persistentă în LangGraph: checkpoints, reluarea execuției și recuperarea după erori

Un agent AI care rulează mai mulți pași nu ar trebui să piardă tot progresul când aplicația este repornită, un API nu răspunde sau procesul este oprit pentru mentenanță. În LangGraph, această continuitate este construită în jurul checkpoints, adică instantanee ale stării agentului salvate în timpul execuției.

Când un graf este compilat cu un checkpointer, LangGraph salvează starea după fiecare etapă a grafului și o organizează în threads. Acest mecanism permite continuarea conversațiilor, aprobări umane, depanare prin „time travel” și recuperare după erori.

Checkpointer sau memorie pe termen lung?

Checkpointer-ul păstrează starea unei anumite execuții: mesaje, rezultate intermediare, noduri finalizate și următorii pași. Identificatorul principal este thread_id.

Un Store are alt rol: păstrează informații care trebuie reutilizate între thread-uri diferite, precum preferințele unui utilizator sau date comune mai multor conversații. În majoritatea aplicațiilor serioase, checkpointer-ul gestionează execuția curentă, iar Store-ul gestionează memoria pe termen lung.

Pentru teste poți folosi InMemorySaver, dar acesta păstrează datele numai în RAM. După repornirea aplicației, checkpoint-urile dispar. Pentru producție, documentația recomandă un backend persistent, precum PostgreSQL, SQLite, Redis sau MongoDB.

Configurează un checkpointer PostgreSQL

Instalează pachetele:

pip install -U langgraph "psycopg[binary,pool]" \
  langgraph-checkpoint-postgres

Construiește apoi graful:

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import RetryPolicy


class AgentState(TypedDict, total=False):
    request: str
    normalized_request: str
    result: str
    status: str


def normalize_request(state: AgentState):
    return {
        "normalized_request": state["request"].strip(),
        "status": "normalized",
    }


def call_external_service(state: AgentState):
    # Înlocuiește cu apelul real către model sau API.
    result = external_service(state["normalized_request"])

    return {
        "result": result,
        "status": "completed",
    }


builder = StateGraph(AgentState)

builder.add_node("normalize", normalize_request)
builder.add_node(
    "call_service",
    call_external_service,
    retry_policy=RetryPolicy(
        max_attempts=3,
        retry_on=ConnectionError,
    ),
)

builder.add_edge(START, "normalize")
builder.add_edge("normalize", "call_service")
builder.add_edge("call_service", END)

Conectează PostgreSQL și compilează graful:

DB_URI = (
    "postgresql://langgraph:parola@localhost:5432/agents"
)

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # Rulează setup() la prima configurare a bazei.
    checkpointer.setup()

    graph = builder.compile(checkpointer=checkpointer)

    config = {
        "configurable": {
            "thread_id": "client-1842-task-77"
        }
    }

    result = graph.invoke(
        {"request": "Analizează documentele proiectului"},
        config,
    )

    print(result)

thread_id trebuie să fie stabil și unic pentru procesul pe care vrei să îl continui. Dacă refolosești același ID, LangGraph încarcă starea thread-ului respectiv. Dacă folosești un ID nou, începi o execuție separată. Pentru PostgresSaver, documentația recomandă păstrarea identificatorului sub 255 de caractere.

Inspectează starea înainte de recuperare

Poți vedea ultimul checkpoint astfel:

snapshot = graph.get_state(config)

print(snapshot.values)
print(snapshot.next)
print(snapshot.config)

values conține starea salvată, iar next indică nodurile programate în continuare. Istoricul complet poate fi accesat prin get_state_history(), util când vrei să afli unde a apărut o eroare sau să reiei execuția dintr-un checkpoint anterior.

Retry, error handler și reluare

Pentru erori temporare, precum indisponibilitatea unui API, folosește RetryPolicy. LangGraph poate reîncerca nodul în funcție de tipul excepției și de politica de backoff.

În LangGraph 1.2 și versiunile ulterioare poți adăuga și un error_handler, executat după epuizarea încercărilor. Handler-ul poate actualiza starea și redirecționa execuția către un nod de recuperare.

Dacă procesul este oprit controlat prin mecanismul de graceful shutdown, LangGraph salvează un checkpoint la limita dintre etape. Execuția poate continua ulterior cu:

result = graph.invoke(None, config)

Trebuie folosit același thread_id, altfel runtime-ul nu știe ce checkpoint trebuie încărcat.

Evită repetarea acțiunilor ireversibile

Reluarea din checkpoint nu înseamnă că orice operație este executată exact o singură dată. Nodurile situate după checkpoint pot rula din nou, inclusiv apelurile către API-uri, modelele AI sau operațiile externe.

Pentru plăți, trimiterea emailurilor sau crearea comenzilor, folosește chei de idempotency:

thread_id + numele operației + identificatorul taskului

Înainte de executare, verifică dacă operația a fost deja înregistrată. Astfel, o reluare nu trimite același email de două ori și nu creează două plăți.

În producție, monitorizează dimensiunea checkpoint-urilor și definește o politică de retenție. Istoricul poate crește continuu, iar documentația recomandă ștergerea periodică a checkpoint-urilor vechi sau configurarea unei perioade de păstrare.

Un agent rezilient nu este doar un model conectat la câteva tool-uri. Are stare persistentă, identificatori stabili, operații idempotente, retry controlat și o procedură clară de reluare după incidente.

Surse folosite

LangChain Docs – LangGraph Persistence
LangChain Docs – Add Memory
LangChain Docs – Fault Tolerance
LangChain Docs – Interrupts
LangChain Docs – Use Time Travel
LangChain Docs – Checkpointer Integrations