Un agente ha appena aperto un ticket, poi il processo si interrompe. Il ticket esiste, ma l'agente non ha ancora salvato l'esito del passaggio. Al riavvio vede un'attività incompleta e la esegue di nuovo. Ora i ticket sono due.
È il punto da cui partirei per valutare un workflow con stato. Recuperare il lavoro serve; capire quali azioni verranno ripetute serve ancora di più. Un checkpoint permette di ritrovare lo stato del processo. L'API che riceve una richiesta deve anche riconoscere quando stiamo chiedendo di completare la stessa operazione.
Ho preparato un laboratorio Docker che rende visibile questa differenza. Il processo viene terminato subito dopo la risposta dell'API fittizia e prima che il nodo abbia restituito il risultato. Prima si ottengono due ticket; aggiungendo un'identità stabile all'operazione, la stessa ripresa restituisce il ticket già creato. Nessun modello a pagamento: qui interessa ciò che succede intorno alla decisione dell'agente.
Che cosa intendiamo per stato e grafo
Un workflow è una procedura composta da passaggi. Lo stato contiene i dati che quei passaggi si scambiano: l'operazione richiesta, i risultati disponibili e le informazioni necessarie per proseguire. Per aprire un ticket possono bastare il titolo, un identificatore dell'operazione e, dopo la creazione, l'identificatore restituito dal servizio.
Un grafo rende esplicito il percorso. I nodi sono funzioni che svolgono un passaggio; gli archi indicano quale funzione eseguire dopo. Un arco condizionale può mandare una proposta all'approvazione, un errore alla gestione del problema o un risultato incompleto a un nuovo tentativo. È questo il senso di graph engineering qui: progettare i passaggi e le transizioni di un processo. Non stiamo parlando di knowledge graph o delle rappresentazioni interne del modello.
Nel laboratorio il percorso è volutamente lineare:
START -> prepare -> create_ticket -> ENDprepare stabilisce la chiave dell'operazione; create_ticket chiama il servizio. Uso LangGraph per ispezionare lo stato e la ripresa di un nodo, senza aggiungere rami che nasconderebbero il difetto. Due funzioni in sequenza sarebbero sufficienti per aprire un ticket: il valore del grafo emerge quando i percorsi, le attese e i ritorni diventano parte del problema.
Un nodo può contenere codice normale, una chiamata a un modello o entrambe le cose. La Graph API di LangGraph definisce proprio questi elementi. La distinzione tra framework di orchestrazione e gli altri componenti è approfondita nella mappa di LangChain, LangGraph, OpenRouter e Langfuse.
Il checkpoint arriva dopo il passaggio
Un checkpoint è una fotografia persistita dello stato di esecuzione. In LangGraph lo selezioniamo attraverso un thread_id: riutilizzare quello originale permette di ritrovare il processo; cambiarlo identifica un altro thread.
Il checkpointer del laboratorio scrive su SQLite. Il file sta in un volume Docker, quindi sopravvive alla sostituzione del container. La modalità durability="sync" salva il checkpoint prima di passare al passo successivo. La documentazione dei checkpointer distingue questa modalità da async, che scrive mentre il passo successivo procede, e da exit, che salva lo stato all'uscita gestita del grafo, senza conservare gli stati intermedi necessari a recuperare un crash del processo.
Il punto critico è dentro create_ticket:
result = call("/tickets", {"title": state["title"]}, key)
emit("effect_ack", **result)
if args.crash:
os.kill(os.getpid(), signal.SIGKILL)
return {"ticket_id": result["ticket_id"]}Il servizio ha già confermato il ticket. Il nodo, invece, muore prima del return. Il checkpoint precedente dice ancora che il prossimo nodo è create_ticket. Alla ripresa viene richiamata la funzione dall'inizio, compresa la richiesta HTTP.
Spezzare il lavoro in nodi più piccoli riduce quanto dobbiamo ripetere, ma rimane questo intervallo fra due sistemi: l'azione è stata registrata dal destinatario e il chiamante non ne ha ancora salvato il risultato. Anche un task con risultato persistito deve affrontare il caso in cui l'effetto avviene prima del salvataggio di quel risultato.
Preparare il laboratorio
Servono Docker Engine e Docker Compose funzionanti, un terminale e accesso a internet per costruire l'immagine. Non occorrono Python installato sul computer, API key o account di un servizio AI. La prova è stata eseguita su Linux ARM64 in Docker Desktop, con Python 3.13.17, LangGraph 1.2.17 e langgraph-checkpoint-sqlite 3.1.2.
Crea una cartella vuota e apri lì il terminale:
mkdir stateful-agent-lab
cd stateful-agent-lab
docker version
docker compose versionCopia nella cartella i cinque file completi riportati in fondo all'articolo. Questi li scrivi tu; Docker creerà l'immagine, la rete e i due volumi:
stateful-agent-lab/
Dockerfile
requirements.txt
compose.yaml
api.py
workflow.pyapi è un ricevitore fittizio che registra ticket nel proprio database. workflow esegue i nodi e usa un database separato per i checkpoint. La rete è interna a Docker e non espone porte sul computer. Ogni comando seguente va eseguito da questa stessa cartella.
docker compose -p stateful-lab up -d --build api
docker compose -p stateful-lab run --rm workflow ticketsLa prima lista deve essere vuota. Se stai ripetendo una prova precedente e vuoi ripartire da zero, il comando di pulizia finale rimuove anche i dati del laboratorio.
Prova 1: lo stato si recupera, il ticket si duplica
Avvia il percorso senza idempotenza:
docker compose -p stateful-lab run --rm workflow start --thread naive --operation case-naive:open-ticket:v1 --mode naive --crashIl comando termina con codice 137, dopo effect_ack e crash_before_node_return. È l'interruzione che abbiamo richiesto. Nel Compose, init: true fa eseguire Python come processo figlio: è necessario per ottenere questa terminazione controllata, invece di affidarsi al comportamento speciale del PID 1 nel container.
Ora osserva entrambi i sistemi:
docker compose -p stateful-lab run --rm workflow state --thread naive
docker compose -p stateful-lab run --rm workflow ticketsLo stato mostra next: ["create_ticket"] e non contiene ticket_id. Il servizio elenca già il ticket 1. Il dato interessante è proprio la distanza fra queste due risposte.
Riprendi lo stesso thread:
docker compose -p stateful-lab run --rm workflow resume --thread naive
docker compose -p stateful-lab run --rm workflow ticketsIl nodo riparte, il servizio crea il ticket 2. prepare non viene rieseguito; l'azione dentro il nodo incompleto sì. Il recupero del checkpoint ha funzionato, ma la seconda richiesta HTTP rappresentava una nuova creazione per il destinatario.
Prova 2: una sola operazione, anche dopo il riavvio
La correzione usa una chiave di idempotenza: un identificatore che resta uguale nei tentativi della stessa operazione. Il destinatario associa atomicamente chiave, contenuto della richiesta e ticket creato. Una richiesta ripetuta restituisce lo stesso ticket.
La chiave case-safe:open-ticket:v1 identifica un'intenzione precisa. Viene salvata nell'input del workflow e inviata nell'header Idempotency-Key. Generare un nuovo UUID dentro il nodo a ogni tentativo annullerebbe questo accordo. Anche usare solo il titolo sarebbe una scelta sbagliata: due richieste legittime possono avere lo stesso testo. È il motivo per cui la Builders' Library di AWS distingue l'identità della richiesta dal semplice confronto dei parametri.
docker compose -p stateful-lab run --rm workflow start --thread safe --operation case-safe:open-ticket:v1 --mode safe --crash
docker compose -p stateful-lab run --rm workflow state --thread safe
docker compose -p stateful-lab run --rm workflow ticketsDopo l'uscita 137, il checkpoint indica ancora create_ticket; il servizio contiene il nuovo ticket 3. Riavvia anche il destinatario, poi riprendi:
docker compose -p stateful-lab restart api
docker compose -p stateful-lab run --rm workflow resume --thread safe
docker compose -p stateful-lab run --rm workflow state --thread safe
docker compose -p stateful-lab run --rm workflow ticketsLa risposta dell'azione è created: false, ticket_id: 3. Il nodo può finalmente salvare quel risultato; next diventa una lista vuota. Nel servizio rimangono tre ticket totali: due della prova precedente e uno di questa.
Nel ricevitore il vincolo UNIQUE sulla chiave e una transazione BEGIN IMMEDIATE proteggono la verifica e la creazione. Se scrivessimo soltanto un log locale dopo la risposta HTTP, un crash fra la risposta e il log riproporrebbe lo stesso problema.
I controlli che distinguono un retry da una nuova richiesta
Con la sola prova positiva potremmo costruire un servizio che rifiuta ogni richiesta successiva. Aggiungiamo due controlli:
docker compose -p stateful-lab run --rm workflow post --operation case-safe:open-ticket:v1 --title "Different payload"
docker compose -p stateful-lab run --rm workflow post --operation case-new:open-ticket:v1Il primo restituisce HTTP 409, con uscita 1: la chiave esiste ma il contenuto è diverso. Il secondo crea il ticket 4, perché identifica un'altra operazione. Il titolo può essere identico a quello precedente senza impedire una nuova creazione.
Puoi anche invocare nuovamente resume --thread safe: il workflow è già terminato e non compare un altro effect_ack.
| Passaggio | Nodo ancora da eseguire | Ticket della singola operazione |
|---|---|---|
| Crash senza chiave | create_ticket | 1 |
| Ripresa senza chiave | nessuno | 2 |
| Crash con chiave | create_ticket | 1 |
| Riavvio del ricevitore e ripresa con chiave | nessuno | 1 |
Questi sono gli esiti osservati nel laboratorio. La deduplicazione dipende dal contratto implementato nel ricevitore e dalla conservazione del suo database. Per un'API che non offre idempotenza serve una verifica dell'operazione tramite un identificatore consultabile oppure una gestione esplicita dell'esito incerto, prima di ritentare. Aggiungere un registro nel chiamante, da solo, non rende atomica una scrittura su un altro sistema.
Ripresa, replay e approvazione hanno conseguenze diverse
La prova recupera l'ultimo stato disponibile. Il replay da un checkpoint storico chiede invece di eseguire nuovamente i nodi successivi a quel punto; può ripetere chiamate API e chiamate al modello. Anche la cronologia richiede quindi una decisione sugli effetti che vogliamo riprodurre. In grafi con nodi paralleli, LangGraph può recuperare le scritture già salvate dei nodi completati; il nostro nodo è morto prima di completare quella scrittura.
Un'approvazione umana è un altro passaggio ancora. LangGraph offre interrupt e ripresa con Command(resume=...). L'approvazione deve riferirsi all'azione che la persona ha esaminato: cambiare destinatario o payload richiede di rivederla. Qui il riavvio usa lo stesso codice e gli stessi dati; migrazioni di workflow, retry concorrenti su più host e riconnessione degli stream restano fuori dalla prova.
Per un servizio reale definirei soprattutto quanto a lungo conservare le chiavi rispetto ai retry ammessi e come riconciliare un risultato incerto. Cancellare troppo presto la ricevuta dell'operazione riapre la possibilità di creare duplicati.
Quando sceglierei qualcosa di più semplice
Per leggere dati, produrre una proposta e restituirla subito può bastare una procedura lineare. Se il requisito è eseguire lavori indipendenti, ritentarli e distribuirli fra worker, una coda con stato nel database evita di modellare un grafo applicativo più ampio.
Il grafo diventa utile quando devo capire e controllare percorsi diversi: una proposta respinta torna alla revisione, una verifica fallita cambia strategia, una persona approva giorni dopo. Aiuta a esprimere queste transizioni; comporta anche uno schema di stato, regole per le migrazioni e manutenzione dei checkpoint.
Un orchestratore durevole come Temporal è un'alternativa per coordinare attività, attese e recovery. La sua documentazione sulle Activity mantiene la richiesta di idempotenza: un'attività può aver eseguito l'effetto e morire prima di registrarne il completamento.
Per scegliere partirei da una domanda molto concreta: dopo un'interruzione, quale operazione posso ripetere e quale devo prima cercare nel sistema destinatario? Il grafo viene dopo questa risposta.
Sorgenti completi e pulizia
I cinque file seguenti appartengono al laboratorio didattico, non a un servizio esposto al pubblico. L'immagine Python è fissata per digest, le due dipendenze dirette per versione. Le dipendenze transitive restano soggette alla risoluzione di pip.
Dopo le prove, questo comando elimina soltanto container, rete e volumi del progetto stateful-lab. Cancella quindi anche i ticket e i checkpoint creati qui:
docker compose -p stateful-lab down -vL'immagine locale stateful-agent-lab:local può essere rimossa con docker image rm stateful-agent-lab:local quando nessun altro laboratorio la usa. La base Python potrebbe essere condivisa con altri progetti.
Dockerfile
FROM python:3.13-slim@sha256:57b29a7c050c4e679e4bea572e14b3965cf4b5839de2da49d55e49ffc4024f39
ENV PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1 \
PIP_DISABLE_PIP_VERSION_CHECK=1 LANGGRAPH_STRICT_MSGPACK=true
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY api.py workflow.py ./
CMD ["python", "workflow.py", "--help"]requirements.txt
langgraph==1.2.17
langgraph-checkpoint-sqlite==3.1.2compose.yaml
name: stateful-agent-lab
services:
api:
image: stateful-agent-lab:local
build: .
command: ["python", "api.py"]
volumes:
- receiver-data:/data
healthcheck:
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=2)"]
interval: 2s
timeout: 3s
retries: 15
read_only: true
cap_drop: [ALL]
security_opt: ["no-new-privileges:true"]
networks: [lab]
workflow:
init: true
image: stateful-agent-lab:local
entrypoint: ["python", "workflow.py"]
environment:
API_URL: http://api:8000
volumes:
- checkpoint-data:/data
depends_on:
api:
condition: service_healthy
read_only: true
cap_drop: [ALL]
security_opt: ["no-new-privileges:true"]
networks: [lab]
volumes:
receiver-data:
checkpoint-data:
networks:
lab:
internal: trueapi.py
"""Local teaching receiver. No external service calls or credentials."""
import json
import sqlite3
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
DB = "/data/receiver.sqlite"
with sqlite3.connect(DB) as db:
db.execute("""CREATE TABLE IF NOT EXISTS tickets (
id INTEGER PRIMARY KEY,
operation_key TEXT UNIQUE,
payload TEXT NOT NULL
)""")
class Handler(BaseHTTPRequestHandler):
def respond(self, status, value):
data = json.dumps(value, sort_keys=True).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
def do_GET(self):
if self.path == "/health":
return self.respond(200, {"ok": True})
if self.path != "/tickets":
return self.respond(404, {"error": "not found"})
with sqlite3.connect(DB) as db:
rows = db.execute(
"SELECT id, operation_key, payload FROM tickets ORDER BY id"
).fetchall()
return self.respond(200, {"tickets": [
{"ticket_id": row[0], "key": row[1],
"payload": json.loads(row[2])} for row in rows
]})
def do_POST(self):
if self.path != "/tickets":
return self.respond(404, {"error": "not found"})
try:
length = int(self.headers.get("Content-Length", "0"))
if not 0 < length <= 4096:
return self.respond(400, {"error": "invalid size"})
payload = json.loads(self.rfile.read(length))
if (not isinstance(payload, dict)
or set(payload) != {"title"}
or not isinstance(payload["title"], str)
or not 1 <= len(payload["title"]) <= 200):
return self.respond(400, {"error": "invalid payload"})
canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
key = self.headers.get("Idempotency-Key")
if key is not None and not 1 <= len(key) <= 200:
return self.respond(400, {"error": "invalid key"})
with sqlite3.connect(DB, timeout=10) as db:
db.execute("BEGIN IMMEDIATE")
prior = (db.execute(
"SELECT id, payload FROM tickets WHERE operation_key = ?",
(key,)
).fetchone() if key else None)
if prior:
if prior[1] != canonical:
return self.respond(409, {"error": "key reused with different payload"})
ticket_id, created = prior[0], False
else:
cursor = db.execute(
"INSERT INTO tickets(operation_key, payload) VALUES (?, ?)",
(key, canonical)
)
ticket_id, created = cursor.lastrowid, True
# The database transaction commits BEFORE the HTTP response.
return self.respond(201 if created else 200, {
"ticket_id": ticket_id, "created": created
})
except (ValueError, json.JSONDecodeError):
return self.respond(400, {"error": "invalid JSON"})
ThreadingHTTPServer(("0.0.0.0", 8000), Handler).serve_forever()workflow.py
"""Reproducible checkpoint/effect gap, without an LLM or paid API."""
import argparse
import json
import os
import signal
import urllib.error
import urllib.request
from typing import NotRequired, TypedDict
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
API = os.environ.get("API_URL", "http://api:8000")
def emit(event, **values):
print(json.dumps({"event": event, **values}, sort_keys=True), flush=True)
def call(path, payload=None, key=None):
headers = {"Content-Type": "application/json"}
if key is not None:
headers["Idempotency-Key"] = key
data = None if payload is None else json.dumps(payload).encode()
request = urllib.request.Request(API + path, data=data, headers=headers)
with urllib.request.urlopen(request, timeout=5) as response:
return json.load(response)
class State(TypedDict):
operation_id: str
mode: str
title: str
key: NotRequired[str]
ticket_id: NotRequired[int]
parser = argparse.ArgumentParser()
parser.add_argument("action", choices=["start", "resume", "state", "tickets", "post"])
parser.add_argument("--thread", default="demo")
parser.add_argument("--operation", default="case-001:open-ticket:v1")
parser.add_argument("--mode", choices=["naive", "safe"], default="safe")
parser.add_argument("--title", default="Check supplier configuration")
parser.add_argument("--crash", action="store_true")
args = parser.parse_args()
if args.action == "tickets":
emit("tickets", **call("/tickets"))
elif args.action == "post":
try:
emit("post", **call("/tickets", {"title": args.title}, args.operation))
except urllib.error.HTTPError as error:
emit("http_error", status=error.code, body=json.loads(error.read()))
raise SystemExit(1)
else:
def prepare(state: State):
emit("prepare", operation=state["operation_id"])
return {"key": state["operation_id"]}
def create_ticket(state: State):
key = state["key"] if state["mode"] == "safe" else None
result = call("/tickets", {"title": state["title"]}, key)
emit("effect_ack", **result)
if args.crash:
emit("crash_before_node_return")
os.kill(os.getpid(), signal.SIGKILL)
return {"ticket_id": result["ticket_id"]}
builder = StateGraph(State)
builder.add_node("prepare", prepare)
builder.add_node("create_ticket", create_ticket)
builder.add_edge(START, "prepare")
builder.add_edge("prepare", "create_ticket")
builder.add_edge("create_ticket", END)
config = {"configurable": {"thread_id": args.thread}}
with SqliteSaver.from_conn_string("/data/checkpoints.sqlite") as saver:
graph = builder.compile(checkpointer=saver)
saved = graph.get_state(config)
if args.action == "state":
emit("state", values=saved.values, next=list(saved.next))
elif args.action == "start":
if saved.values:
parser.error("thread already exists; resume it or choose a new thread")
result = graph.invoke({
"operation_id": args.operation, "mode": args.mode,
"title": args.title
}, config, durability="sync")
emit("done", **result)
else:
if not saved.values:
parser.error("thread not found; use the original thread id")
result = graph.invoke(None, config, durability="sync")
emit("done", **result)