Lo stato come API: LangGraph dopo tre riscritture

Perché conta

I tutorial su LangGraph di solito mostrano il graph. Spesso non mostrano cosa succede dopo settimane in un production multi-agent system, quando lo state dict è cresciuto, più agent scrivono sulla stessa key, e un checkpoint riproduce un valore che prima era valido ma ora non lo è più.

Ho riscritto lo state model dello stesso sistema tre volte in quattro mesi. LangGraph non ha cambiato la sua API. I miei primi due design erano sbagliati in modi che ho visto solo sotto carico reale.

Il terzo design ha retto l'aggiunta di altri agent senza cambiare il contratto dello schema. La lezione è stata diretta: lo state schema è l'API tra agent. Se lo tratto come un contenitore mutabile di valori, creo problemi di migration e recovery più avanti.

1. La prima iterazione: flat dict e perché collassa

Il primo design sembrava ragionevole. Ho usato un semplice TypedDict con una dozzina di key. Ogni agent leggeva ciò che gli serviva e scriveva ciò che produceva. Era chiaro, semplice, e comune nei tutorial.

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

Cosa si è rotto: due agent scrivevano su messages in parallelo — uno aggiungendo tool results, l'altro aggiungendo user messages — e l'ordinamento non era esplicito. should_continue veniva impostato a True dal planner e poi sovrascritto a False dal critic eseguito uno step dopo, perché LangGraph fonde gli output dei node in sequenza e l'ultima scrittura vince.

Il problema più profondo era il write model implicito. Un flat dict rende ogni campo condiviso una possibile race condition. Funziona nel happy path. Fallisce quando due node toccano lo stesso campo nello stesso step, o quando un node assume che un campo non sia cambiato dall'ultima volta in cui lo ha letto.

Ho provato a correggere il design flat con un campo last_writer, lock a livello di node, e una convenzione in cui un node svuotava un campo prima di scriverlo. Quelle correzioni rendevano il graph più difficile da ragionare senza sistemare il contratto. Mi servivano merge semantics a livello di campo invece di overwrite-by-default.

2. La seconda iterazione: Pydantic annidato e il problema della rigidità

Il secondo tentativo è andato nella direzione opposta. Ho avvolto lo state in modelli Pydantic annidati e aggiunto validation a ogni boundary.

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 alcuni problemi di write conflict perché ogni agent possedeva una sezione dello schema. Ha anche spostato complessità nelle conditional edge. Le routing functions dovevano ispezionare state annidato come research is not None and draft is not None and draft.critique_history[-1].approved prima di decidere dove andare dopo.

Quei predicate degli edge sono diventati business logic critica. Erano difficili da testare in isolamento e difficili da valutare per chi faceva review.

L'altro problema: None è un sentinel ambiguo nello state di LangGraph. research: None significa "non iniziato", "fallito", o "saltato intenzionalmente"? Stavo incorporando workflow state dentro data state. Cambiare il workflow significava cambiare lo schema, e questo rendeva le migration più costose.

Quando ho aggiunto un altro agent che doveva sovrascrivere parzialmente i research results senza scartare le sources, ho aggiunto research_override: ResearchState | None perché non potevo mutare in sicurezza il campo esistente. Lo schema cresceva perché l'update model era troppo rigido, non perché il dominio richiedesse più concetti.

Pydantic annidato è adatto ai contratti di tool I/O. Per lo state LangGraph in un graph la cui struttura sta ancora evolvendo, l'ho trovato troppo rigido.

3. La terza iterazione: channels con reducers

Il design che ha funzionato è quello che LangGraph documenta ma che molti tutorial saltano: campi Annotated con reducer functions.

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

class AgentState(TypedDict):
    messages: Annotated[list[str], operator.add]      # append-only
    tool_calls: Annotated[list[ToolCall], operator.add]  # append-only
    current_draft: str                                 # last-write-wins (intentional)
    iteration: Annotated[int, lambda a, b: b]          # always take latest
    approved: bool | None                              # routing signal, write-once per step

Cosa è cambiato: messages e tool_calls usano operator.add come reducer, quindi gli output dei node vengono aggiunti invece di sostituire la lista esistente. Se due node aggiungono un message nello stesso step, entrambi i message vengono preservati nell'ordine di graph traversal. Non c'è overwrite silenzioso.

current_draft resta last-write-wins perché lo tocca solo il writer node, e voglio il valore più recente. Il punto non è vietare gli overwrite. Il punto è rendere intenzionale il comportamento di scrittura di ogni campo.

Questo ha cambiato il mio mental model. Non stavo progettando una data structure; stavo progettando un contratto di message-passing. Ogni campo è diventato un channel con una regola per combinare le scritture.

Questo mi ha dato tre proprietà che i design precedenti non avevano. L'esecuzione parallela dei node era più sicura perché i reducer gestivano il comportamento di merge senza coordination logic dentro i node. Lo schema comunicava l'intento: Annotated[list, operator.add] dice a chi legge che il campo accumula. Lo schema è anche diventato più stabile durante la crescita, perché aggiungere un agent spesso significava aggiungere tipi di message a un append channel invece di aggiungere nuovi campi top-level.

In quel progetto, la migration dal design Pydantic è stata più breve delle riscritture precedenti. Anche la code review è diventata più concreta: invece di domande ampie come "è sicuro?", i reviewer potevano chiedere "è questo il reducer giusto?"

4. Le conditional edge non sono routing logic

Dopo aver sistemato lo state model, l'errore successivo che vedo spesso è mettere business logic dentro le conditional edge functions.

Una conditional edge in LangGraph prende lo state e restituisce il nome del node successivo. Mantengo questa responsabilità stretta: dato questo snapshot dello state, quale node viene eseguito dopo?

# Wrong: business logic inside the 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"

# Right: routing reads an explicit signal written by a node
def should_continue(state: AgentState) -> str:
    if state.get("approved") or state["iteration"] >= state["max_iterations"]:
        return "end"
    return "generate"

La failure mode: edge functions che parsano stringhe, fanno database lookup, o contengono logica multi-step sono difficili da testare e difficili da debuggare nelle trace. Quando il routing sbaglia, la decisione è difficile da riprodurre se la logica rilevante viveva solo dentro l'edge e non era stata scritta nello state.

La regola che seguo è semplice: se una routing condition richiede più che leggere un campo e confrontare un valore, metto quella condizione in un node. Il node scrive nello state un routing signal esplicito, e l'edge legge quel signal. Nei miei design, lo state è la source of truth, non la edge function.

Evito anche di fare routing direttamente sul contenuto dei message quando posso. Faccio routing su un campo scritto da un node dopo che ha interpretato il message. L'interpreter node è dove deve stare l'ambiguità, ed è la parte che posso unit-testare. La edge function resta vicina a una lookup table.

5. Interrupt e il problema del dual-write

Human-in-the-loop è una delle funzionalità utili di LangGraph. È anche dove ho visto incidenti in production. Il problema di solito non è l'API. È il design pattern attorno a side effects e checkpoints.

Il pattern ingenuo è diretto: aggiungere un interrupt_before al node "send email", mettere in pausa il graph, mostrare la pending action nella UI, raccogliere approval, chiamare graph.update_state() con l'approval, e riprendere l'esecuzione.

Il dual-write problem: se un node esegue un side effect esterno prima del checkpoint, e il graph salva lo state dopo il completamento del node, il replay dal checkpoint può rieseguire quel side effect. Invia un'email, checkpoint, il human rifiuta, retry — l'email potrebbe già essere stata inviata prima che il rifiuto venga processato.

# Wrong: side effect before interrupt boundary
def send_draft_node(state: AgentState) -> dict:
    email_client.send(state["draft"], to=state["recipient"])  # side effect happens here
    return {"email_status": "sent"}  # checkpoint captures this — replaying from here re-sends

# Right: separate intent from execution
def prepare_send_node(state: AgentState) -> dict:
    # write the intent — no external call yet
    return {"pending_action": {"type": "email", "to": state["recipient"], "body": state["draft"]}}
    # graph interrupts here; human sees and approves 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 uso è: scrivere l'intent nello state, interrompere, lasciare che il human approvi o modifichi l'intent, poi eseguire. Il replay dal checkpoint pre-execution riesegue l'intent node, che non ha side effect esterni. L'executor viene eseguito solo se human_approved è presente nello state.

Questo è un caso specifico di una regola più ampia: i checkpoint catturano state, non side effects. Progetto i node in modo che il replay da un checkpoint sia sicuro. Se un node ha side effects esterni, quegli effetti richiedono idempotency, un approval gate prima dell'esecuzione, o entrambi. Se nessuna delle due condizioni è vera, non considero quel side effect sicuro da eseguire dentro il graph.

6. Cosa salvo nei checkpoint e cosa no

Il checkpointer di LangGraph persiste lo state a ogni boundary di node per default. Lo considero il punto di partenza sicuro. È anche un cost model che voglio capire prima di scalare un graph.

In un graph che chiama un LLM a ogni node, ogni step scrive un checkpoint. In un graph su cui ho lavorato, le scritture remote dei checkpoint aggiungevano overhead che potevo vedere nelle trace. Con una concurrency più alta, quell'overhead diventava più facile da notare. Ancora più importante, non era sempre necessario. Non avevo bisogno di stato persistente attorno a ogni node solo-compute. Ne avevo bisogno attorno a side effects e chiamate costose e non deterministiche.

Quando rivedo il placement dei checkpoint, di solito arrivo a quattro classi:

• Ha side effects esterni (API call, DB write, email): normalmente faccio checkpoint prima e dopo. Il replay deve essere sicuro, quindi progetto per idempotency o metto un approval gate prima della prima esecuzione.
• Chiama un LLM: salvo nei checkpoint gli input prima della call. Le LLM call sono costose e non deterministiche, quindi se il node fallisce a metà call, voglio fare replay con gli stessi input invece di rideterminarli.
• Pure transform (parsing, formatting, filtering): di solito salto il checkpoint. Riprodurre una conversione JSON-to-TypedDict è deterministico ed economico.
• Routing node: di solito non aggiungo un checkpoint. La routing decision è riproducibile dallo state, che è già salvato nei checkpoint.

La differenza conta. Fare checkpoint di un formatting node e riprodurlo in caso di failure di solito è innocuo, ma spreca storage e può aggiungere latency. Non fare checkpoint di un DB write node e riprodurlo in caso di failure può creare una duplicate write.

Il checkpoint audit è anche un correctness audit. Prima di decidere cosa richiede stato persistente, identifico quali node hanno side effects, quali node sono non deterministici, e quali node sono pure transforms.

Quando faccio audit di un sistema LangGraph, elenco tutti i node e classifico ciascuno: side effect, LLM call, pure transform, o routing. Nella mia pratica, i node nelle prime due classi richiedono checkpoint coverage. I node nelle ultime due classi sono candidati a essere saltati. Se non riesco a classificare un node, sistemo prima quello.

Il quadro della categoria

Tre riscritture dello stesso sistema non sono una storia su LangGraph complesso. LangGraph mi dà le primitive di cui ho bisogno per stato persistente e replay. La storia riguarda il costo di trattare lo state come un dettaglio implementativo invece che come l'interfaccia primaria.

In un multi-agent system, lo state schema è il contratto tra agent. Determina come gli agent evolvono in modo indipendente, come aggiungo un nuovo agent senza rompere quelli esistenti, e come debuggo una trajectory più avanti. Se il contratto dello state è instabile, ogni nuovo agent diventa un rischio di migration. Se il contratto è esplicito, può diventare una delle parti meno volatili del sistema.

Le tre iterazioni si riducono a una regola che uso ora: progettare lo state come channels con merge semantics esplicite, non come un oggetto mutabile condiviso. Conditional edge functions, interrupt patterns, recovery deterministica, e granularità dei checkpoint sono più facili da ragionare partendo da questa base.

Se stai costruendo o facendo audit di un sistema LangGraph e incontri state conflicts, routing fragile, o problemi con gli interrupt, scrivimi. È la classe di problemi che trovo più interessante.

FAQ

Come dovrei modellare lo shared state LangGraph tra agent?

Modello lo shared state come channels con merge semantics esplicite, non come un oggetto mutabile condiviso. In pratica, questo significa usare reducer dove serve, come liste append-only, e mantenere il comportamento di overwrite solo dove è deliberato e posseduto da uno specifico writer.

Perché uno state flat TypedDict può fallire durante l'esecuzione parallela dei node?

Un flat dict rende ogni campo condiviso una possibile race condition. Se due node scrivono la stessa key nello stesso step senza un reducer, LangGraph oggi rifiuta l'aggiornamento concorrente invece di fare merge silenzioso last-write-wins — il failure è esplicito, ma il design richiede comunque un channel/reducer esplicito prima che quei node possano essere eseguiti insieme. L'ho incontrato la prima volta con messages condivisi e routing flags che venivano sovrascritti tra step o rifiutati in parallelo.

Quando i modelli Pydantic annidati sono troppo rigidi per lo state LangGraph?

Ho trovato Pydantic annidato utile per i contratti di tool I/O, ma troppo rigido per graph state in evoluzione. Le conditional edge dovevano ispezionare dati annidati, None diventava un sentinel ambiguo, e gli override parziali creavano nuovi campi nello schema perché l'update model non era abbastanza flessibile.

Cosa appartiene a una conditional edge function di LangGraph?

Mantengo le conditional edge strette: dato questo snapshot dello state, quale node viene eseguito dopo? Se una condizione richiede parsing, lookup, o logica multi-step, metto quella logica in un node, scrivo nello state un routing signal esplicito, e lascio che l'edge legga quel signal.

Come evito side effects duplicati con gli interrupt di LangGraph?

Separo intent ed execution. Un node scrive la pending action nello state senza fare la call esterna, il graph interrompe per approval o modifica da parte del human, e un executor successivo viene eseguito solo se l'approval è presente. I checkpoint catturano state, non side effects.