17. Persistenza dello stato e checkpointing

Nel Capitolo 15 abbiamo ricostruito il ciclo dell'agente come StateGraph, e nel Capitolo 16 abbiamo usato componenti predefiniti e il routing multi-ramo per costruire qualcosa di più elaborato. Ogni grafo che abbiamo scritto finora condivide una limitazione: il grafo non conserva il proprio stato.

Il grafo conserva e gestisce lo stato per la durata di una singola chiamata a invoke(), e non un istante di più. Quando lo chiami, LangGraph crea uno stato nuovo, esegue i nodi, unisce ciascun valore di ritorno nello stato secondo le regole del reducer e restituisce lo stato finale al chiamante. Una volta restituito quello stato, il grafo non lo ricorda più. La successiva chiamata a invoke() parte da uno stato completamente nuovo, senza alcun collegamento con la chiamata precedente.

Ne derivano due problemi. Primo, anche messages fa parte dello stato, quindi l'agente non può ricordare nulla di ciò che hai detto in precedenza. Secondo, se un'esecuzione fallisce a metà strada, tutto ciò che aveva realizzato fino a quel punto scompare. Poniamo che il terzo nodo sollevi un'eccezione: i risultati prodotti dai primi due nodi se ne vanno con esso, e devi ricominciare dall'inizio. Nel paragrafo 15.1 avevamo elencato il "recupero dalle interruzioni" come uno dei motivi per ricorrere a LangGraph — è proprio questo il problema che avevamo in mente.

LangGraph gestisce tutto ciò a livello di framework. Collega un checkpointer a un grafo e LangGraph salverà automaticamente uno snapshot dello stato a ogni passo dell'esecuzione. Quegli snapshot salvati sopravvivono alla chiamata a invoke(), così la chiamata successiva può riprendere da dove si era interrotta l'ultima. Questa proprietà—lo stato che sopravvive oltre una singola esecuzione—si chiama persistenza.

In realtà hai già usato un checkpointer. Nel Capitolo 11, quando abbiamo dato all'agente RAG conversazionale la memoria multi-turno, abbiamo passato create_agent(..., checkpointer=InMemorySaver()) e un thread_id. All'epoca, tutto ciò che dovevi sapere era che il checkpointer conserva la cronologia della conversazione per ciascun thread_id; non abbiamo mai spiegato come. E nel Capitolo 16, quando abbiamo introdotto il parametro checkpointer, abbiamo detto "spiegheremo come funziona nel Capitolo 17." Questo è quel capitolo.

Si articola in tre parti. Nel 17.1 colleghiamo un checkpointer a un grafo e teniamo conversazioni multi-turno con thread_id. Nel 17.2 apriamo i checkpoint salvati per vedere cosa l'agente sapeva in un dato momento—i checkpoint sono il tuo strumento principale per rintracciare il motivo per cui un agente si è comportato male. Nel 17.3 prendiamo un grafo fallito a metà esecuzione e lo riprendiamo da dove si era fermato invece che dall'inizio.

17.1) Trasportare lo stato tra le chiamate

17.1.1) Un grafo che dimentica

Abbiamo aperto questo capitolo dicendo che un grafo non conserva il proprio stato. Confermiamolo con il codice.

Il grafo qui sotto ha la stessa forma del grafo say_hello del paragrafo 15.2. L'unica differenza è che il nodo restituisce una risposta dell'LLM invece di una stringa fissa.

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()   # nessun checkpointer

Ora teniamo una conversazione a due turni. Diciamo al modello il nostro nome nella prima chiamata, poi gli chiediamo qual è il nostro nome nella seconda.

# Prima chiamata — diamo il nostro nome
graph.invoke({"messages": [{"role": "user", "content": "Ciao, mi chiamo Bob."}]})
 
# Seconda chiamata — lo richiediamo
result = graph.invoke({"messages": [{"role": "user", "content": "Come mi chiamo?"}]})
print(result["messages"][-1].content)

Output:

Mi dispiace, ma non conosco il tuo nome. Potresti dirmi qual è?

L'unico messaggio che il modello ha ricevuto nella seconda chiamata era "Come mi chiamo?". Lo stato della prima chiamata non è più nel grafo, quindi i messaggi precedenti—quelli che portavano il nome—non sono mai arrivati al modello.

Potremmo naturalmente risolverlo da soli. Conservare i messages restituiti dalla prima chiamata e passarli insieme alla seconda. È esattamente così che gestivamo la cronologia della conversazione nel Capitolo 8. Ma allora scriveremmo il nostro codice per memorizzare e recuperare la cronologia di ogni conversazione e di ogni utente. È proprio questo il lavoro di cui un checkpointer ti solleva.

17.1.2) Checkpoint e checkpointer

Un checkpointer è un oggetto il cui compito è salvare lo stato. Crei un'istanza—InMemorySaver(), per esempio—e la passi a builder.compile(checkpointer=...) per collegarla al tuo grafo.

Una volta collegato un checkpointer, il grafo copia l'intero stato e lo salva man mano che l'esecuzione procede. Ciascuna di quelle copie salvate si chiama checkpoint. Pensalo come una fotografia: l'intero stato in quell'istante, conservato esattamente com'era.

Il salvataggio automatico in un videogioco è l'immagine mentale giusta. Il gioco registra silenziosamente i tuoi progressi ogni volta che superi un punto significativo, così puoi uscire e tornare più tardi, o morire senza dover ricominciare dall'inizio. Un checkpointer fa esattamente questo per un grafo.

Quindi, quand'è un "punto significativo"? LangGraph divide l'esecuzione di un grafo in fasi, e ogni fase si chiama super-step. Un checkpoint viene salvato ogni volta che un super-step finisce.

Il motivo per cui è un super-step e non semplicemente un passo è che una singola fase può eseguire più nodi contemporaneamente. In un grafo come quello del paragrafo 17.1.1, dove i nodi formano una linea retta, eseguire un nodo è un super-step. Ma in un grafo dove più nodi vengono eseguiti in parallelo, tutti quei nodi insieme costituiscono un unico super-step.

Quindi anche una singola chiamata a invoke() lascia dietro di sé diversi checkpoint. Li estrarremo e guarderemo esattamente cosa contiene ciascuno nel 17.2.

L'InMemorySaver che abbiamo usato come esempio è il checkpointer più semplice che esista. Come suggerisce il nome, memorizza i checkpoint nella memoria del processo (RAM). Non c'è nulla da installare e nulla da configurare, il che lo rende adatto all'apprendimento e allo sviluppo locale. Il compromesso è che ogni checkpoint salvato scompare quando il processo si riavvia. Esamineremo le alternative di produzione nel 17.1.5.

17.1.3) Aggiungere un checkpointer

Collegare un checkpointer richiede solo due cose.

1. Creare un'istanza di checkpointer e passarla a compile().
2. Passare un config contenente un thread_id ogni volta che chiami invoke().

Arriveremo tra poco al motivo per cui la seconda è necessaria. Per ora, sappi solo che indica al checkpointer quale delle conversazioni salvate vuoi continuare.

Applichiamo entrambe al grafo del paragrafo 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. Crea un checkpointer e passalo a compile()
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
 
# 2. Passa a invoke() un config che porta un thread_id
config = {"configurable": {"thread_id": "1"}}
 
graph.invoke(
    {"messages": [{"role": "user", "content": "Ciao, mi chiamo Bob."}]},
    config,
)
result = graph.invoke(
    {"messages": [{"role": "user", "content": "Come mi chiamo?"}]},
    config,
)
print(result["messages"][-1].content)

Output:

Ti chiami Bob.

Le stesse due chiamate del paragrafo 17.1.1, e un esito diverso. Questa volta il nome resta.

Ecco perché. Un grafo con un checkpointer collegato carica lo stato salvato prima di eseguire il nodo llm_call. Quello stato contiene già il primo scambio. Il nuovo messaggio che abbiamo passato viene poi unito in esso. Come abbiamo visto nel paragrafo 15.2.2, il campo messages porta il reducer add_messages, quindi il nuovo messaggio viene aggiunto alla lista esistente. Il modello finisce per ricevere tre messaggi: il saluto, la sua prima risposta e la nuova domanda.

Inviamo solo il nuovo messaggio, e LangGraph carica la conversazione precedente dall'ultimo checkpoint. La cronologia della conversazione che gestivamo a mano nel Capitolo 8 ora è gestita da LangGraph.

17.1.4) thread_id: l'identificatore che separa le conversazioni

Un thread_id è un identificatore che distingue una conversazione da un'altra. Il valore lo scegli tu. Abbiamo usato "1" nel paragrafo 17.1.3, ma qualsiasi stringa va bene. Chiama il grafo con lo stesso thread_id e continui quella conversazione; chiamalo con uno diverso e inizi una conversazione separata.

Confermiamolo. Faremo tenere ad Alice e Bob conversazioni diverse attraverso lo stesso 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
 
# Conversazione di Alice
send("alice", "Il mio colore preferito è il verde acqua.")
 
# Conversazione di Bob — un thread_id diverso
send("bob", "Il mio colore preferito è l'arancione.")
 
# Chiedi a ciascuno di loro di nuovo
print("Alice:", send("alice", "Qual è il mio colore preferito?"))
print("Bob:  ", send("bob", "Qual è il mio colore preferito?"))

Output:

Alice: Il tuo colore preferito è il verde acqua.
Bob:   Il tuo colore preferito è l'arancione.

Entrambe le conversazioni sono passate attraverso lo stesso oggetto graph e lo stesso checkpointer, eppure non si sono mai mescolate. Il thread_id è la chiave primaria che il checkpointer usa per memorizzare e recuperare lo stato. Chiavi diverse, archiviazione completamente separata.

Quindi cosa succede se ometti il valore?

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

Output:

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

Il grafo non viene eseguito affatto. Una volta collegato un checkpointer, thread_id non è opzionale—è obbligatorio.

Considerato quello che abbiamo appena visto, ha senso. Prima di eseguire un nodo, il checkpointer deve caricare lo stato salvato — e il thread_id è ciò che gli indica lo stato di quale conversazione caricare.

Questa è la forma di base di un servizio di chatbot: un grafo, un checkpointer e un thread_id per utente o per stanza di chat.

17.1.5) I limiti di InMemorySaver e le alternative di produzione

Abbiamo detto prima che InMemorySaver conserva i checkpoint in memoria. Con questa scelta arrivano due limitazioni.

Riavvia il processo e tutto sparisce. Ridistribuisci il servizio o riavvia il server, e ogni conversazione accumulata fino a quel momento scompare.

Processi separati non possono condividerlo. Un servizio reale distribuisce le richieste in arrivo su diversi processi. Ogni processo ha la propria memoria, quindi una conversazione salvata dal processo A è invisibile al processo B. Un utente può inviare lo stesso thread_id ogni volta e comunque veder cadere a pezzi la conversazione, a seconda di quale processo capita di prendere in carico la richiesta.

Ecco perché la produzione usa checkpointer che memorizzano i checkpoint in un database.

• SqliteSaver / AsyncSqliteSaver (langgraph-checkpoint-sqlite) — memorizza tutto in un unico file. Adatto a un piccolo servizio in esecuzione su un singolo server, o a un prototipo locale.
• PostgresSaver / AsyncPostgresSaver (langgraph-checkpoint-postgres) — memorizza i checkpoint in un server di database. Aggiungi altri server e ogni processo vede comunque gli stessi checkpoint. Questo è il checkpointer che la documentazione di LangGraph raccomanda per la produzione.

Tutti implementano la stessa interfaccia di InMemorySaver. Il codice del tuo grafo, i tuoi nodi e il modo in cui usi thread_id restano esattamente com'erano. L'unica cosa che cambia è come crei il checkpointer.

create_agent, che hai incontrato nel Capitolo 16, usa i checkpointer allo stesso modo. Passa un checkpointer al suo parametro checkpointer, e passa a invoke() un config che porta un thread_id. È questo che ha fatto ricordare i turni precedenti all'agente RAG conversazionale del Capitolo 11.

Continueremo a usare InMemorySaver per il resto di questo capitolo. Ovunque i checkpoint capiti che risiedano, il modo in cui ci lavori è lo stesso.

17.2) Ispezionare lo stato ed effettuare il debug

Nel paragrafo 17.1.2 abbiamo detto che il checkpointer salva lo stato a ogni super-step. Estraiamo quei checkpoint e guardiamoli.

Aprire i checkpoint salvati non è mera curiosità. Gli agenti si comportano male. Chiamano gli strumenti in un ciclo che non finisce mai, perdono traccia dei turni precedenti, prendono un ramo che non ti saresti mai aspettato. Per scoprire il perché, devi sapere che aspetto aveva lo stato in quel momento—e i checkpoint hanno la risposta.

LangGraph ti offre due metodi.

• graph.get_state(config) — restituisce il checkpoint più recente per quella conversazione.
• graph.get_state_history(config) — restituisce ogni checkpoint per quella conversazione, dal più recente al più vecchio.

Entrambi hanno bisogno di un thread_id nel config, poiché devono sapere di chi stai chiedendo i checkpoint.

I checkpoint che questi due metodi restituiscono sono rappresentati come oggetti StateSnapshot.

17.2.1) StateSnapshot: cosa c'è dentro un checkpoint

Vediamo cosa contiene effettivamente un checkpoint. Non useremo un LLM per questo esempio. Un LLM restituisce qualcosa di diverso ogni volta, il che non aiuta quando vogliamo esaminare ciascun campo uno per uno. Costruiremo invece un piccolo grafo che restituisce valori fissi.

Lo stato avrà due tipi di campi. foo non ha reducer, quindi viene sovrascritto; bar ne ha uno, quindi accumula. È lo stesso schema del reducer add_messages che abbiamo collegato a messages nel paragrafo 15.2.2—qui usiamo operator.add di Python come reducer per concatenare le liste.

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                        # nessun reducer → sovrascrivi
    bar: Annotated[list[str], add]  # reducer add → accumula
 
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)

Output:

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=()
)

Otto campi. Esaminiamoli uno alla volta.

• values — lo stato così com'era a questo checkpoint. bar è ['a', 'b'] perché il reducer add ha accumulato ciò che entrambi i nodi hanno restituito; foo è 'b' perché non ha reducer, quindi ha vinto l'ultima scrittura. È il campo che risponde alla domanda "che aspetto aveva lo stato in quel momento?"
• next — una tupla di nomi di nodi da eseguire dopo questo checkpoint. Un () vuoto significa che non c'è nulla da eseguire, cioè il grafo è finito. ('node_b',) significa che node_b è ancora davanti.
• config — l'indirizzo di questo checkpoint. thread_id identifica la conversazione, e checkpoint_id identifica quale momento al suo interno. LangGraph assegna automaticamente il checkpoint_id ogni volta che salva un checkpoint.
• metadata — informazioni di servizio sull'esecuzione. source ti dice da dove proviene il checkpoint: "input" significa che è stato costruito dall'input che hai passato a invoke(), e "loop" significa che è stato prodotto mentre il grafo era in esecuzione. step è il numero del super-step.
• created_at — quando il checkpoint è stato salvato. Comodo quando devi allinearlo con i tuoi log.
• parent_config — il config del checkpoint immediatamente precedente a questo. Seguilo e puoi camminare all'indietro attraverso l'esecuzione. È None per il primissimo checkpoint.
• tasks — il registro di esecuzione per i nodi elencati in next. Nel momento in cui un checkpoint viene salvato, quei nodi non sono ancora stati eseguiti; una volta eseguiti, il loro esito viene allegato a questo checkpoint. Un nodo che è riuscito lascia il suo valore di ritorno in result, e un nodo fallito lascia la sua eccezione in error.
• interrupts — dove il grafo si è messo in pausa per restituire il controllo a una persona. LangGraph può fermarsi a metà esecuzione e attendere che qualcuno approvi un passo o fornisca un valore, e questo campo registra quelle pause.

I campi si leggono come attributi. metadata è un dizionario, quindi ne estrai i valori con una chiave.

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

Di questi, quello a cui ricorrerai più spesso durante il debug è next. Se next non è vuoto, il grafo non è arrivato alla fine—si è fermato da qualche parte a metà. E quando riprenderemo un grafo fallito nel 17.3, questo campo è il punto da cui iniziamo.

17.2.2) Percorrere la cronologia dei checkpoint

get_state() ti mostra solo il checkpoint più recente. Ma fare il debug significa spesso chiedersi "come siamo finiti qui?"—e per questo ti serve l'intera traiettoria dell'esecuzione. get_state_history() te la fornisce.

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

Output:

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': []}

Il checkpoint più recente viene per primo, quindi procedi dal basso verso l'alto per seguire l'esecuzione in ordine.

• step -1 — subito dopo che invoke() ha ricevuto l'input. Nota che il {"foo": "", "bar": []} che abbiamo passato non compare in values. Portare l'input nello stato è di per sé una fase, e quella fase non è ancora stata eseguita. Il __start__ in next è il nodo interno che se ne occupa.

  bar compare come [], ma non è il valore che abbiamo passato. Un campo con un reducer parte con un valore vuoto in cui le scritture possono accumularsi. foo non ha reducer, quindi non ha alcun valore iniziale — ecco perché non compare qui.

• step 0 — __start__ è stato eseguito e l'input è ora nello stato. foo='' e bar=[] sono i valori che abbiamo passato. node_a è il prossimo.
• step 1 — il risultato dell'esecuzione di node_a. foo è ora 'a' e bar è ['a'], con node_b come prossimo.
• step 2 — il risultato dell'esecuzione di node_b. next è vuoto, quindi il grafo è finito.

Un next non vuoto allo step 0 o allo step 1 non significa che il grafo si sia fermato lì. Un checkpoint preso mentre il grafo è ancora in esecuzione ha naturalmente un nodo in fila come prossimo. Quando nel 17.2.1 abbiamo detto "un next non vuoto significa che il grafo si è fermato", stavamo parlando del checkpoint nel punto in cui qualcosa è andato storto. Nel mezzo della cronologia, next ti mostra semplicemente quale percorso ha preso il grafo.

17.3) Riprendere da dove è fallito

Poiché lo stato viene salvato alla fine di ogni super-step, un fallimento a metà esecuzione non porta con sé il lavoro completato—è ancora lì nei checkpoint. Non c'è motivo di ricominciare dall'inizio. Riprendi da dove si è fermato.

17.3.1) Riprendere con invoke(None, config)

Riprendere è semplice: passa None dove va l'input.

graph.invoke(None, config)

Significa "non c'è alcun nuovo input; continua dallo stato salvato." Il config ha comunque bisogno di un thread_id, naturalmente, poiché LangGraph deve sapere quale conversazione continuare.

Mettiamo in scena un fallimento e riprendiamo da esso. Costruiremo un grafo a due nodi il cui secondo nodo fallisce solo alla sua prima esecuzione. Deve riuscire al nuovo tentativo, altrimenti non vedremmo mai la ripresa funzionare davvero.

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 per far fallire la prima esecuzione
 
def step_1(state: State):
    print("step_1 in esecuzione (lavoro costoso)")
    return {"step_1_done": True}
 
def step_2(state: State):
    global first_try
    if first_try:
        first_try = False
        print("step_2 fallito (timeout API)")
        raise RuntimeError("External API timed out")
    print("step_2 in esecuzione")
    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("Fallito:", e)

Output:

step_1 in esecuzione (lavoro costoso)
step_2 fallito (timeout API)
Fallito: External API timed out

step_1 è riuscito e step_2 ha sollevato un'eccezione. Scopriamo dove si è fermato il grafo, usando il get_state() che abbiamo imparato nel 17.2.

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

Output:

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

next è ('step_2',), il che ci dice che il grafo si è fermato a metà dell'esecuzione di step_2. E in values, step_1_done è True—il risultato di step_1 è ancora lì nel checkpoint.

Ora riprendiamo con None.

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

Output:

step_2 in esecuzione
Final = {'step_1_done': True, 'step_2_done': True}

step_1 in esecuzione (lavoro costoso) non è mai stato stampato. step_1 non è stato eseguito una seconda volta. LangGraph ha caricato lo stato salvato e ha ripreso da step_2. Non abbiamo pagato due volte per quel costoso primo passo.

17.3.2) A cosa fare attenzione quando si riprende

Quando riprendi, il nodo che è fallito viene eseguito di nuovo. Se quel nodo chiama un LLM o raggiunge un'API esterna, anche quelle chiamate avvengono di nuovo—e potrebbero restituire qualcosa di diverso.

È lì che si nasconde la trappola. Se step_2 ha inviato un'email e poi è fallito, riprendere invia una seconda email. LangGraph garantisce soltanto che non rieseguirà i nodi che sono riusciti.

Quindi qualsiasi nodo che potrebbe essere eseguito di nuovo deve essere idempotente: fare la stessa cosa due volte dovrebbe lasciarti nello stesso punto. Verifica se l'email è già stata inviata prima di inviarla; metti una chiave univoca sulla tabella del database affinché un inserimento duplicato non possa avvenire.