Lo stato è l'API: LangGraph dopo tre riscritture

Perché conta

I tutorial LangGraph ti mostrano il grafo. Quello che non ti mostrano è cosa succede a sei settimane dal go-live di un sistema multi-agente in produzione, quando hai aggiunto quindici campi al tuo state dict, tre agenti scrivono sulla stessa chiave con assunzioni diverse, e un checkpoint replay un valore che era valido due ore fa ma adesso è incoerente.

Ho riscritto il modello di stato per lo stesso sistema tre volte in quattro mesi. Non perché LangGraph cambiasse API — non lo ha fatto — ma perché i miei primi due design erano sbagliati in modi che non riuscivo a vedere finché il sistema non girava sotto carico reale. Ogni riscrittura ci è costata una settimana di migrazione. Il terzo design ha retto otto mesi e quattro agenti aggiuntivi senza toccare il contratto dello schema.

Lo schema di stato è l'API tra i tuoi agenti. Se lo progetti come la maggior parte dei tutorial suggerisce — un sacco di valori mutabili — lo paghi. Ecco cosa ho imparato.

1. Prima iterazione: dict piatto e perché collassa

Il primo design sembrava perfettamente ragionevole. Un semplice TypedDict con una dozzina di chiavi, ogni agente legge ciò che gli serve, scrive ciò che produce. Pulito, semplice, tutorial-approved.

class AgentState(TypedDict):
    messages: list[str]
    research_results: list[dict]
    draft: str
    critique: str
    iteration_count: int
    should_continue: bool

Cosa si è rotto: due agenti che scrivevano su messages in parallelo — uno appendendo i risultati dei tool, uno appendendo i messaggi utente — producevano un ordering indefinito. should_continue veniva impostato a True dal planner e immediatamente sovrascritto a False dal critic che girava un passo dopo, perché LangGraph fonde gli output dei nodi in modo sequenziale e l'ultima scrittura vince.

Il problema più profondo: un dict piatto con semantica di scrittura implicita trasforma ogni nodo in una potenziale race condition. Funziona nell'happy path. Si rompe nel momento in cui due nodi toccano lo stesso campo nello stesso step, o quando un nodo assume che un campo non sia stato modificato dall'ultima volta che lo ha letto.

La soluzione non è leggere in modo difensivo o coordinare i nodi. La soluzione è cambiare come si applicano gli aggiornamenti di stato — passare da una semantica di sovrascrittura a una semantica di merge esplicita per campo. Non puoi appiccicare quella cosa su un dict piatto a posteriori senza migrare ogni nodo che tocca lo stato.

Abbiamo provato a rattopparlo. Abbiamo aggiunto un campo last_writer per tracciare chi avesse modificato cosa, abbiamo aggiunto lock a livello di nodo, abbiamo aggiunto una convenzione "claim" dove un nodo svuotava un campo prima di scriverci. Niente ha funzionato in modo affidabile. Il problema era il design, non l'implementazione.

2. Seconda iteranza: Pydantic annidato e il problema della rigidità

Il secondo tentativo è andato nella direzione opposta. Ho avvolto tutto in modelli Pydantic, li ho annidati e ho aggiunto validazione ovunque.

class ResearchState(BaseModel):
    query: str
    sources: list[Source]
    confidence: float

class DraftState(BaseModel):
    content: str
    word_count: int
    critique_history: list[Critique]

class AgentState(TypedDict):
    research: ResearchState | None
    draft: DraftState | None
    final_output: str | None

Ha risolto il problema dei conflitti in scrittura — ogni agente possedeva la sua sezione dello schema. Ma ha creato un altro failure mode: le funzioni edge condizionali dovevano ispezionare research is not None and draft is not None and draft.critique_history[-1].approved prima di fare routing. Tre settimane dopo, quei predicati edge erano diventati business logic load-bearing, testati da nessuna parte, leggibili da nessuno.

L'altro problema: None come sentinel è semanticamente ambiguo in LangGraph. research: None vuol dire "non iniziato", "fallito" o "saltato di proposito"? Stavamo incorporando lo stato del workflow nello stato dei dati, il che significava che cambiare il workflow richiedeva cambiare lo schema — una migrazione completa ogni volta.

Quando abbiamo aggiunto un quarto agente che doveva fare un override parziale dei risultati di ricerca senza scartare le source, abbiamo dovuto aggiungere research_override: ResearchState | None perché non potevamo mutare in modo sicuro il campo esistente. Lo schema accumulava campi non perché il dominio crescesse, ma perché il design non aveva un meccanismo per update parziali sicuri.

Pydantic annidato è lo strumento giusto per i tool I/O contract. È lo strumento sbagliato per lo stato di LangGraph quando la struttura del grafo è ancora in evoluzione.

3. Terza iterazione: channel con reducer

Il design che ha funzionato è quello che LangGraph documenta davvero ma che la maggior parte dei tutorial salta: campi Annotated con funzioni reducer.

import operator
from typing import Annotated
from langgraph.graph import StateGraph

class AgentState(TypedDict):
    messages: Annotated[list[str], operator.add]      # solo append
    tool_calls: Annotated[list[ToolCall], operator.add]  # solo append
    current_draft: str                                 # last-write-wins (intenzionale)
    iteration: Annotated[int, lambda a, b: b]          # prendi sempre l'ultimo
    approved: bool | None                              # signal di routing, una scrittura per step

Cosa è cambiato: messages e tool_calls usano operator.add come reducer — significa che l'output di ogni nodo viene appeso alla lista esistente, non la sostituisce. Se due nodi aggiungono entrambi un messaggio nello stesso step, vengono preservati entrambi, in ordine di traversal del grafo. Niente sovrascrittura silenziosa.

current_draft è semplice — last-write-wins — perché solo il nodo writer lo tocca, e vogliamo l'ultimo valore. Il modello a channel rende la semantica di scrittura esplicita e per-campo, non implicita e globale.

Il salto nel modello mentale è questo: non stai progettando una struttura dati, stai progettando un contratto di message-passing. Ogni campo è un channel con una regola per combinare scritture concorrenti. La sovrascrittura è una regola valida — deve solo essere una scelta deliberata, non il default.

Questo design ha tre proprietà che i precedenti non avevano. Primo, è sicuro sotto esecuzione parallela dei nodi — il reducer gestisce il merge senza coordination logic nei nodi. Secondo, lo schema comunica l'intent: Annotated[list, operator.add] dice a ogni reader che questo campo accumula — nessuno lo "fixerà" aggiungendo una sovrascrittura. Terzo, lo schema è stabile sotto crescita — aggiungere un nuovo agente significa aggiungere nuovi tipi di messaggio a un append channel, non nuovi campi top-level. Il contratto tra gli agenti esistenti non cambia.

La migrazione dal design Pydantic ha richiesto tre giorni invece di una settimana. Il guadagno di chiarezza è stato immediato: ogni commento di code review sullo stato è passato da "è sicuro?" a "è il reducer giusto?".

4. Gli edge condizionali non sono routing logic

Dopo aver sistemato il modello di stato, l'errore successivo che vedo costantemente — nelle code review e nei sistemi su cui faccio audit — è trattare le funzioni edge condizionali come il posto dove mettere la business logic.

Un edge condizionale in LangGraph prende lo stato e ritorna il nome del nodo successivo. Stop. Una funzione, una responsabilità: "dato questo snapshot di stato, quale nodo gira dopo?"

# Sbagliato: business logic dentro il router
def should_continue(state: AgentState) -> str:
    if state["iteration"] > 5:
        return "end"
    last_message = state["messages"][-1] if state["messages"] else ""
    if "approved" in last_message.lower() or "looks good" in last_message.lower():
        return "end"
    return "generate"

# Giusto: il routing legge un signal esplicito scritto da un nodo
def should_continue(state: AgentState) -> str:
    if state.get("approved") or state["iteration"] >= state["max_iterations"]:
        return "end"
    return "generate"

Il failure mode: funzioni edge che fanno parsing di stringhe, lookup su database o multi-step logic non sono testabili in isolamento e non sono debuggabili nei trace. Quando il routing va male — e va male, di solito in una combinazione di stato che i tuoi unit test non hanno mai coperto — non puoi fare replay della decisione perché la logica viveva dentro l'edge, non scritta nello stato.

La regola che seguo: se la condizione di routing implica più di leggere un campo e confrontare un valore, la condizione appartiene a un nodo che scrive un signal di routing nello stato. L'edge legge il signal. Lo stato è sempre la source of truth, non la funzione edge.

Un corollario: mai fare routing in base al contenuto di un messaggio. Fai routing in base a un campo che un nodo ha scritto esplicitamente dopo aver interpretato il messaggio. Il nodo interpreter è dove vive l'ambiguità ed è dove puoi fare unit test. La funzione edge dovrebbe essere una lookup table.

5. Interrupt e il problema della doppia scrittura

Lo human-in-the-loop è una delle feature più forti di LangGraph. È anche dove ho visto più incidenti in produzione — non perché l'API sia sbagliata, ma perché il design pattern che la maggior parte dei team adotta è sottilmente rotto.

Il pattern naive: aggiungi un interrupt_before al nodo "invia email", il grafo si mette in pausa, la tua UI mostra l'azione pendente, l'umano clicca approva, chiami graph.update_state() con l'approvazione, l'esecuzione riprende.

Il problema della doppia scrittura: se il tuo nodo esegue un effetto collaterale esterno prima del checkpoint, e il grafo salva lo stato dopo il completamento del nodo, il replay dal checkpoint rieseguirà quell'effetto collaterale. Invii un'email, checkpoint, l'umano rifiuta, retry — hai inviato l'email due volte prima che il rifiuto venga processato.

# Sbagliato: side effect prima del confine di interrupt
def send_draft_node(state: AgentState) -> dict:
    email_client.send(state["draft"], to=state["recipient"])  # side effect qui
    return {"email_status": "sent"}  # il checkpoint cattura questo — replay re-invia

# Giusto: separa intent ed esecuzione
def prepare_send_node(state: AgentState) -> dict:
    # scrivi l'intent — nessuna chiamata esterna ancora
    return {"pending_action": {"type": "email", "to": state["recipient"], "body": state["draft"]}}
    # il grafo si interrompe qui; l'umano vede e approva pending_action

def execute_send_node(state: AgentState) -> dict:
    action = state.get("pending_action")
    if action and state.get("human_approved"):
        email_client.send(action["body"], to=action["to"])
    return {"pending_action": None, "human_approved": None}

Il pattern: scrivi l'intent nello stato, interrompi, lascia che l'umano approvi o modifichi l'intent, poi esegui. Il replay dal checkpoint pre-esecuzione rilancia il nodo intent — che non ha effetti collaterali. L'executor gira solo se human_approved è settato nello stato.

È un caso specifico della regola generale: i checkpoint catturano lo stato, non gli effetti collaterali. Progetta i nodi in modo che il replay da qualsiasi checkpoint sia sicuro. Se un nodo ha effetti collaterali esterni, quegli effetti devono essere idempotenti, oppure devono vivere dopo il gate di approvazione finale, oppure entrambi. Se nessuna delle due cose è vera, l'effetto collaterale non è sicuro da eseguire dentro il grafo.

6. Cosa metto a checkpoint e cosa no

Il checkpointer di LangGraph persiste lo stato a ogni confine di nodo per default. È il default sicuro e il punto di partenza giusto. È anche un cost model che vale la pena capire prima di scalare.

In un grafo che chiama un LLM a ogni nodo, ogni step scrive un checkpoint. Per un grafo di 10 nodi con 100ms a scrittura su uno store remoto, sono un secondo di overhead per traiettoria. Sotto alta concorrenza diventa latenza visibile. Più importante, è inutile: non hai bisogno di mettere a checkpoint i nodi compute-only — devi mettere a checkpoint prima e dopo gli effetti collaterali.

La classificazione che uso per ogni nodo prima del rilascio:

• Ha effetti collaterali esterni (chiamata API, scrittura DB, email): sempre checkpoint prima e dopo. Il replay deve essere sicuro — progetta per idempotenza o metti un gate di approvazione prima della prima esecuzione.
• Chiama un LLM: checkpoint degli input prima della call. Le call LLM sono costose e non-deterministiche — se il nodo fallisce a metà call, vuoi fare replay con gli stessi input, non riderivarli.
• Trasformazione pura (parsing, formatting, filtering): salta il checkpoint. Riapplicare una conversione JSON-to-TypedDict è gratis e corretto.
• Nodo di routing: nessun checkpoint necessario. La decisione di routing è riproducibile dallo stato, che è già a checkpoint.

Mettere a checkpoint un nodo di formatting e farlo replay in caso di failure è innocuo ma spreca storage e aggiunge latenza. Non mettere a checkpoint un nodo di scrittura DB e farlo replay in caso di failure crea una doppia scrittura. Sapere quali nodi hanno effetti collaterali è la stessa conoscenza che ti serve per progettare bene il grafo all'inizio — quindi l'audit dei checkpoint è anche un audit di correttezza.

Quando faccio audit di un sistema LangGraph: lista tutti i nodi, classifica ognuno (effetti collaterali / call LLM / trasformazione pura / routing). Ogni nodo nelle prime due classi deve avere un checkpoint. Ogni nodo nelle ultime due è candidato a saltarlo. Se non riesci a classificare un nodo, è la prima cosa da sistemare.

Il framing di categoria

Tre riscritture sullo stesso sistema non è una storia sulla complessità di LangGraph. LangGraph è uno strumento ben progettato. È una storia sul costo di trattare lo stato come un dettaglio implementativo invece che come l'interfaccia primaria.

Nei sistemi multi-agente, lo schema di stato è il contratto tra gli agenti. Determina come possono evolvere indipendentemente, come aggiungi un nuovo agente senza rompere quelli esistenti, e come fai debug di una traiettoria tre settimane dopo, quando un cliente apre un ticket di supporto. Sbaglialo e ogni nuovo agente diventa un rischio di migrazione. Azzeccalo ed è la parte del sistema che non ha bisogno di cambiare.

Le tre iterazioni si riducono a una sola regola: progetta lo stato come un set di channel con semantica di merge esplicita, non come un oggetto mutabile condiviso. Tutto il resto — funzioni edge condizionali, pattern di interrupt, granularità dei checkpoint — segue dall'avere quella fondazione corretta.

Se stai costruendo o facendo audit di un sistema LangGraph e ti scontri con conflitti di stato, routing fragile o problemi di interrupt, scrivimi. È la classe di problemi che trovo più interessante.