Lo stato è l'API: LangGraph dopo tre riscritture

Perché conta

I tutorial LangGraph ti mostrano il grafo. Quello che spesso non mostrano è cosa succede sei settimane dopo il go-live di un sistema multi-agente in produzione: hai aggiunto quindici campi allo state dict, tre agenti scrivono sulla stessa chiave con assunzioni diverse, e un checkpoint fa il replay di un valore che era valido due ore prima ma ora non è più coerente.

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. Il terzo design ha retto l'aggiunta di altri agenti senza cambiare il contratto dello schema.

Lo schema di stato è l'API tra i tuoi agenti. Se lo tratti come un contenitore generico di valori mutabili, il costo arriva più tardi: nei replay, nei conflitti di scrittura e nei routing difficili da spiegare. Ecco cosa ho imparato.

1. Prima iterazione: dict piatto e perché collassa

Il primo design sembrava ragionevole. Un TypedDict semplice, una dozzina di chiavi, ogni agente legge ciò che gli serve e scrive ciò che produce. Pulito, facile da capire, molto simile a tanti esempi introduttivi.

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 scrivevano su messages in parallelo. Uno appendeva i risultati dei tool, l'altro i messaggi utente. Il risultato era un ordine non deterministico. In un altro punto, should_continue veniva impostato a True dal planner e subito sovrascritto a False dal critic, che girava un passo dopo. In assenza di reducer espliciti, il merge tende a comportarsi come last-write-wins per i campi condivisi.

Il problema più profondo era il modello di scrittura implicito. Un dict piatto trasforma ogni nodo in una potenziale race condition. Funziona nell'happy path, ma si rompe quando due nodi toccano lo stesso campo nello stesso step o quando un nodo assume che un campo non sia cambiato dall'ultima lettura.

La correzione non è leggere in modo più difensivo o coordinare ogni nodo a mano. La correzione è cambiare come si applicano gli aggiornamenti di stato: da una semantica di sovrascrittura a una semantica di merge esplicita per campo. Se provi ad aggiungerla dopo su un dict piatto, devi migrare ogni nodo che tocca lo stato.

Ho provato a rattopparlo. Ho aggiunto un campo last_writer per tracciare chi modificava cosa. Ho aggiunto lock a livello di nodo. Ho aggiunto una convenzione "claim", in cui un nodo svuotava un campo prima di scriverci. Nessuna di queste correzioni ha retto bene. Il problema era il design, non l'implementazione locale.

2. Seconda iterazione: 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

Questo ha risolto i conflitti in scrittura: ogni agente possedeva la sua sezione dello schema. Però ha creato un'altra modalità di guasto. Le funzioni edge condizionali dovevano ispezionare condizioni come research is not None and draft is not None and draft.critique_history[-1].approved prima di decidere il routing. Dopo poche settimane, quei predicati erano diventati logica di business portante: poco testati, difficili da leggere e ancora più difficili da modificare.

L'altro problema: None come sentinel è ambiguo in LangGraph. research: None vuol dire "non iniziato", "fallito" o "saltato di proposito"? Stavo incorporando lo stato del workflow nello stato dei dati. Di conseguenza, cambiare il workflow richiedeva cambiare lo schema, quindi fare una migrazione completa.

Quando ho aggiunto un quarto agente che doveva fare un override parziale dei risultati di ricerca senza scartare le source, ho dovuto aggiungere research_override: ResearchState | None. Non potevo 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 è utile per i contratti I/O dei tool. Per lo stato di LangGraph, soprattutto mentre la struttura del grafo è ancora in evoluzione, mi ha dato più rigidità che sicurezza.

3. Terza iterazione: channel con reducer

Il design che ha funzionato è quello che LangGraph documenta davvero, ma che molti tutorial saltano: 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
    max_iterations: int                                # limite configurato per il routing
    approved: bool | None                              # signal di routing, una scrittura per step

Cosa è cambiato: messages e tool_calls usano operator.add come reducer. L'output di ogni nodo viene appeso alla lista esistente invece di sostituirla. Se due nodi aggiungono entrambi un messaggio nello stesso step, entrambi vengono preservati, nell'ordine di traversal del grafo. Non c'è una sovrascrittura silenziosa.

current_draft rimane semplice: last-write-wins. È una scelta intenzionale, perché solo il nodo writer lo tocca e mi serve l'ultimo valore. Il modello a channel rende la semantica di scrittura esplicita e per-campo, invece che implicita e globale.

Il salto nel modello mentale è questo: non sto progettando solo una struttura dati, sto progettando un contratto di passaggio di messaggi. Ogni campo è un channel con una regola per combinare scritture concorrenti. La sovrascrittura è una regola valida, purché sia una scelta deliberata e non il default ereditato per caso.

Da quel punto in poi ho iniziato a cercare tre proprietà nello schema. La prima è la sicurezza sotto esecuzione parallela dei nodi: il reducer gestisce il merge senza logica di coordinamento dentro i nodi. La seconda è la leggibilità dell'intento: Annotated[list, operator.add] dice a ogni reader che quel campo accumula. La terza è la stabilità sotto crescita: aggiungere un agente spesso significa aggiungere nuovi tipi di messaggio a un append channel, non nuovi campi top-level.

La migrazione dal design Pydantic è stata più breve delle riscritture precedenti. Il cambiamento più visibile è stato nelle code review: le domande sono passate da "è sicuro?" a "è il reducer giusto?".

4. Gli edge condizionali non sono logica di routing

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

Un edge condizionale in LangGraph prende lo stato e ritorna il nome del nodo successivo. Per me deve rispondere a una domanda stretta: "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"

La modalità di guasto: funzioni edge che fanno parsing di stringhe, lookup su database o logica multi-step sono difficili da testare in isolamento e difficili da leggere nei trace. Quando il routing sbaglia — spesso per una combinazione di stato non coperta dai test — non puoi fare replay pulito della decisione se la logica vive solo dentro l'edge e non lascia un segnale nello stato.

La regola pratica che seguo è semplice: se la condizione di routing richiede più che leggere un campo e confrontare un valore, la metto in un nodo che interpreta lo stato e scrive un segnale di routing esplicito. L'edge legge quel segnale. In questo modo lo stato resta la fonte di verità, non la funzione edge.

Evito anche di fare routing in base al contenuto grezzo di un messaggio. Preferisco fare routing in base a un campo scritto esplicitamente da un nodo dopo aver interpretato il messaggio. L'ambiguità vive nel nodo interpreter, dove posso testarla. La funzione edge resta quasi una lookup table.

5. Interrupt e il problema della doppia scrittura

Lo human-in-the-loop è una delle feature più utili di LangGraph. È anche uno dei punti in cui ho visto più incidenti in produzione. Il problema, di solito, non è l'API: è il pattern di integrazione.

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

Il problema della doppia scrittura: se il nodo esegue un effetto collaterale esterno prima del checkpoint, e il grafo salva lo stato dopo il completamento del nodo, un replay dal checkpoint può rieseguire lo stesso effetto collaterale. Nel caso peggiore invii un'email, fai checkpoint, l'umano rifiuta, parte un retry e l'email viene inviata di nuovo 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 che preferisco è separare intent ed esecuzione. Prima scrivo l'intent nello stato. Poi interrompo. L'umano approva o modifica l'intent. Solo dopo eseguo l'effetto collaterale. Se riparto da un checkpoint pre-esecuzione, rilancio il nodo che prepara l'intent, che non ha effetti collaterali. L'executor gira solo se human_approved è presente nello stato.

È un caso specifico della regola generale: i checkpoint catturano lo stato, non gli effetti collaterali. Per questo progetto 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 entrambe le cose. Se nessuna delle due condizioni è vera, non considero sicuro eseguire quell'effetto 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. È un default sicuro e un buon punto di partenza. È anche un modello dei costi da capire prima di scalare.

In un grafo che chiama un LLM a ogni nodo, ogni step scrive un checkpoint. Su uno store remoto, le scritture dei checkpoint possono aggiungere un overhead visibile nei trace, soprattutto con maggiore concorrenza. Il punto non è disattivare i checkpoint ovunque, ma metterli dove servono: prima e dopo i punti in cui il replay ha conseguenze.

Prima del rilascio, classifico ogni nodo con questa checklist:

• Ha effetti collaterali esterni (chiamata API, scrittura DB, email): checkpoint prima e dopo. Il replay deve essere sicuro, quindi progetto per idempotenza o metto 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, voglio fare replay con gli stessi input, non riderivarli.
• Trasformazione pura (parsing, formatting, filtering): di solito salto il checkpoint. Riapplicare una conversione JSON-to-TypedDict è economico e corretto.
• Nodo di routing: di solito non richiede un checkpoint. La decisione di routing è riproducibile dallo stato, che è già a checkpoint.

Un esempio concreto: mettere a checkpoint un nodo di formatting e farlo replay in caso di failure è quasi sempre innocuo, ma spreca storage e aggiunge latenza. Non mettere a checkpoint un nodo di scrittura DB, invece, può trasformare un retry in una doppia scrittura. Sapere quali nodi hanno effetti collaterali è la stessa conoscenza che serve per progettare bene il grafo all'inizio. L'audit dei checkpoint diventa quindi anche un audit di correttezza.

Quando faccio audit di un sistema LangGraph, parto da una tabella semplice: lista dei nodi, tipo di nodo, effetti collaterali, checkpoint necessari. I nodi con effetti collaterali e le call LLM richiedono attenzione esplicita. Le trasformazioni pure e i routing sono candidati a saltare il checkpoint. Se non riesco a classificare un nodo, quello è il primo punto da sistemare.

Il framing di categoria

Tre riscritture sullo stesso sistema non sono una storia sulla complessità di LangGraph. LangGraph è uno strumento ben progettato. Sono 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 puoi aggiungere un nuovo agente senza rompere quelli esistenti e come fai debug di una traiettoria settimane dopo, quando arriva un ticket di supporto. Se lo sbagli, ogni nuovo agente diventa un rischio di migrazione. Se lo progetti bene, diventa una parte stabile del sistema.

Per me, le tre iterazioni si riducono a una regola: progetto lo stato come un set di channel con semantica di merge esplicita, non come un oggetto mutabile condiviso. Funzioni edge condizionali, pattern di interrupt e granularità dei checkpoint diventano più semplici quando questa fondazione è chiara.

Se in un sistema LangGraph emergono conflitti di stato, routing fragile o problemi di interrupt, il primo passo è diagnosticare il punto in cui il contratto di stato si rompe.

FAQ

Perché un dict piatto nello stato LangGraph crea conflitti?

Un dict piatto rende ogni nodo una potenziale race condition: se due nodi scrivono lo stesso campo nello stesso step, senza reducer espliciti il merge tende a comportarsi come last-write-wins per i campi condivisi. L'ho visto con messages e should_continue, dove l'ordine non deterministico e le sovrascritture rendevano replay e routing difficili da spiegare.

Quando conviene usare reducer nei campi di stato LangGraph?

Uso reducer quando un campo deve combinare scritture concorrenti con una semantica esplicita. Per messages e tool_calls uso operator.add, così gli output dei nodi vengono appesi invece di sostituirsi. Per campi come current_draft, invece, mantengo last-write-wins solo se è una scelta intenzionale.

Dove va messa la logica di routing in LangGraph?

Tengo gli edge condizionali stretti: leggono uno snapshot di stato e scelgono il nodo successivo. Se una condizione richiede parsing, lookup o logica multi-step, la metto in un nodo che scrive un segnale nello stato. L'edge legge quel segnale, così lo stato resta la fonte di verità.

Come evito doppie scritture con interrupt human-in-the-loop?

Separo intent ed esecuzione. Prima scrivo nello stato l'azione pendente, poi interrompo e lascio approvare o modificare l'intent. Solo dopo eseguo l'effetto collaterale. Poiché i checkpoint catturano lo stato ma non gli effetti collaterali, progetto i replay in modo che siano sicuri.

Quali nodi LangGraph meritano un checkpoint?

Classifico ogni nodo prima del rilascio. Metto attenzione esplicita su effetti collaterali esterni e call LLM: voglio checkpoint dove il replay ha conseguenze o dove servono gli stessi input. Di solito salto checkpoint su trasformazioni pure e routing, perché la decisione è riproducibile dallo stato già persistito.