Home Fondamenti Token Modelli AI Deep Learning Tecniche RAG RAG Avanzato MCP Orchestrazione LangChain LangGraph Prompt Engineering Usare l'AI ChipsBot News
Home / LangGraph

LangGraph: Agenti Stateful come Grafi Diretti

LangGraph trasforma la costruzione di agenti AI complessi: invece di sperare che il modello faccia le cose giuste in un loop libero, tu disegni esplicitamente il flusso come un grafo. Stato persistente, branch condizionali, approvazioni umane — tutto sotto controllo.

Cos’è LangGraph: agenti come grafi

Immagina un processo aziendale di approvazione acquisti. C’è una richiesta iniziale, va al manager, che può approvarla (e passa alla contabilità) oppure rifiutarla (e torna al richiedente con una motivazione) oppure chiedere chiarimenti (torna al richiedente, poi quando risponde ritorna al manager). Non è un flusso lineare — ci sono branch, loop, stati intermedi e coinvolgimento umano.

LangGraph è esattamente lo strumento per costruire questo tipo di sistema, ma con agenti AI. È la risposta di LangChain a una domanda concreta: come costruire agenti che si comportano in modo prevedibile e affidabile in produzione?

I semplici ReAct agent (un loop LLM + tool) hanno tre limiti fondamentali che LangGraph risolve:

  1. Stato volatile: i loop classici "ricordano" solo la conversazione. LangGraph ha uno State esplicito — un dizionario tipizzato che ogni nodo può leggere e modificare, che persiste tra step e sopravvive ai crash.
  2. Flusso imprevedibile: in un loop libero, il modello decide autonomamente cosa fare dopo. Con LangGraph, sei tu a disegnare le transizioni — nessuna sorpresa su quale nodo viene eseguito dopo quale condizione.
  3. No human oversight: LangGraph ha il human-in-the-loop come cittadino di prima classe — puoi mettere in pausa il grafo, far approvare un’azione da un umano, e riprendere esattamente da dove si era fermato.

LangGraph fa parte dell’ecosistema LangChain ma può essere usato completamente indipendentemente, anche senza aver mai toccato LangChain.

I concetti fondamentali: State, Nodes, Edges

LangGraph ha tre concetti base. Capirli bene è tutto.

State

Lo State è la "memoria di lavoro" del grafo — un TypedDict Python che rappresenta tutto ciò che il grafo sa in un dato momento. Ogni nodo può leggerlo e scriverne parti. Lo State è l’unico modo in cui i nodi si "comunicano" tra loro.

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class AgentState(TypedDict):
    # add_messages è un "reducer": aggiunge invece di sostituire
    messages: Annotated[list, add_messages]
    # Campi custom per la logica applicativa
    task_completato: bool
    tentativi: int
    dati_raccolti: dictpython

La parola chiave è Annotated: permette di specificare un reducer, ovvero una funzione che determina come lo State viene aggiornato quando più nodi scrivono sullo stesso campo. add_messages è un reducer predefinito che aggiunge i nuovi messaggi alla lista invece di sovrascriverla — il comportamento ovvio per una conversazione.

Nodes

Un nodo è una funzione Python che riceve lo State attuale e restituisce un dizionario con i campi da aggiornare. Non restituisce uno State completo — solo le differenze (delta). I campi non menzionati rimangono invariati.

def nodo_analisi(state: AgentState) -> dict:
    # Leggi dallo state
    ultima_domanda = state["messages"][-1].content

    # Logica del nodo: chiamata LLM, tool, calcolo, ecc.
    risposta = llm.invoke(ultima_domanda)

    # Ritorna SOLO i campi da aggiornare (non tutto lo state)
    return {
        "messages": [risposta],             # add_messages lo aggiunge
        "tentativi": state["tentativi"] + 1  # incrementa il contatore
    }python

Edges

Gli edges (archi) definiscono come il grafo si muove da un nodo all’altro. Ci sono due tipi:

  • Edge normale: sempre da A a B, senza condizioni. builder.add_edge("A", "B")
  • Conditional edge: una funzione Python decide dove andare in base allo State. builder.add_conditional_edges(...)

Il tuo primo grafo

Prima di affrontare i casi complessi, costruiamo il grafo più semplice possibile: un chatbot che risponde alle domande.

START ──▴ [chatbot] ──▴ END
from langgraph.graph import StateGraph, END, START
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage
from typing import TypedDict, Annotated
from langgraph.graph import add_messages

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")

class State(TypedDict):
    messages: Annotated[list, add_messages]

# Definisci il nodo: riceve State, restituisce delta
def chatbot(state: State) -> dict:
    risposta = llm.invoke(state["messages"])
    return {"messages": [risposta]}

# Costruisci il grafo
builder = StateGraph(State)
builder.add_node("chatbot", chatbot)

# Definisci il flusso: START → chatbot → END
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

# Compila: valida la struttura del grafo e ottimizza
graph = builder.compile()

# Esecuzione
result = graph.invoke({
    "messages": [HumanMessage(content="Ciao! Cos'è LangGraph?")]
})
print(result["messages"][-1].content)python

Il metodo builder.compile() non è solo un builder pattern — valida la struttura del grafo (nodi irraggiungibili, edge verso nodi inesistenti), produce una rappresentazione ottimizzata e abilita funzionalità avanzate come checkpointing e streaming.

Conditional edges: routing intelligente

Il vero potere di LangGraph emerge con i conditional edges. Una funzione "router" ispeziona lo State dopo ogni nodo e decide quale nodo attivare successivamente.

Il pattern classico: un agente con tool use deve decidere se continuare (ha richiesto un tool) o fermarsi (ha la risposta finale).

[agente] ──▴ router ──▴ [usa_tool] ──┐ │ │ └──▴ END │ │ │ ├───────────────────┴ │ (torna all'agente) (agente)
def dovrei_continuare(state: State) -> str:
    """Router: decide se usare un tool o terminare."""
    ultimo_messaggio = state["messages"][-1]

    # Se il modello ha generato tool_calls, dobbiamo eseguirli
    if hasattr(ultimo_messaggio, "tool_calls") and ultimo_messaggio.tool_calls:
        return "usa_tool"

    # Altrimenti il modello ha la risposta finale
    return "end"

# add_conditional_edges: nodo_sorgente, funzione_router, mappa_output→nodo
builder.add_conditional_edges(
    "agente",
    dovrei_continuare,
    {
        "usa_tool": "tool_node",  # se router ritorna "usa_tool" → tool_node
        "end": END                 # se router ritorna "end" → fine
    }
)python

La funzione router può restituire qualsiasi stringa — la mappa {"output_router": "nome_nodo"} converte l’output del router nel nodo corretto. Questo disaccoppiamento è importante: la logica di routing è nel grafo, non dentro il nodo.

Implementare un ReAct agent

Ecco un agente completo con tool use, costruito con LangGraph. Rispetto agli agent di LangChain classici, hai visibilità e controllo su ogni step del loop:

from langgraph.graph import StateGraph, END, START
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool

@tool
def cerca_prodotto(nome: str) -> str:
    """Cerca informazioni su un prodotto nel catalogo."""
    catalogo = {
        "laptop": "ThinkPad X1: 16GB RAM, i7, 1299€",
        "mouse": "MX Master 3: wireless, ergonomico, 89€"
    }
    return catalogo.get(nome.lower(), "Prodotto non trovato")

@tool
def calcola_sconto(prezzo: float, percentuale: float) -> str:
    """Calcola il prezzo scontato dato prezzo base e percentuale."""
    scontato = prezzo * (1 - percentuale / 100)
    return f"Prezzo scontato: {scontato:.2f}€"

tools = [cerca_prodotto, calcola_sconto]
# bind_tools dice al modello quali tool può invocare
llm_con_tools = llm.bind_tools(tools)

class State(TypedDict):
    messages: Annotated[list, add_messages]

def agente(state: State) -> dict:
    risposta = llm_con_tools.invoke(state["messages"])
    return {"messages": [risposta]}

def router(state: State) -> str:
    ultimo = state["messages"][-1]
    if hasattr(ultimo, "tool_calls") and ultimo.tool_calls:
        return "tools"
    return "end"

builder = StateGraph(State)
builder.add_node("agente", agente)
builder.add_node("tools", ToolNode(tools))  # ToolNode esegue i tool automaticamente

builder.add_edge(START, "agente")
builder.add_conditional_edges("agente", router, {"tools": "tools", "end": END})
builder.add_edge("tools", "agente")  # dopo i tool, torna all'agente

graph = builder.compile()

result = graph.invoke({
    "messages": [HumanMessage("Cerca il laptop e dimmi quanto costa con il 20% di sconto")]
})python

ToolNode è un nodo prebuilt di LangGraph che gestisce automaticamente l’esecuzione dei tool: analizza le tool_calls nell’ultimo messaggio, chiama la funzione corrispondente e aggiunge il risultato come ToolMessage allo State.

State management avanzato

Per applicazioni reali lo State deve fare di più del semplice accumulare messaggi. I reducer personalizzati permettono di definire esattamente come ogni campo viene aggiornato:

from typing import Annotated
import operator

def mantieni_ultimi_5(left: list, right: list) -> list:
    """Finestra mobile: mantieni solo gli ultimi 5 elementi."""
    return (left + right)[-5:]

class State(TypedDict):
    messages: Annotated[list, add_messages]          # accumula messaggi
    errori: Annotated[list, operator.add]             # accumula errori (append)
    risultati: Annotated[list, mantieni_ultimi_5]     # finestra mobile
    contatore: int                                    # sostituzione semplice
    stato_corrente: str                               # sostituzione semplicepython

I campi senza Annotated usano il comportamento di default: sostituzione completa. Se un nodo scrive {"contatore": 5}, il valore diventa 5 indipendentemente da cosa c’era prima. I campi con reducer invece combinano il valore precedente con quello nuovo secondo la funzione specificata.

Checkpoints e persistenza

I checkpoint sono snapshot completi dello State salvati dopo ogni step del grafo. Sono la fondazione di tre funzionalità critiche per la produzione: resumability (riprendi dopo un crash), branching (esplora percorsi alternativi), debugging (ispeziona lo stato di ogni step passato).

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres import PostgresSaver  # per la produzione

# Checkpointer in memoria (solo per sviluppo/test)
checkpointer = MemorySaver()

# In produzione: PostgreSQL o Redis
# checkpointer = PostgresSaver.from_conn_string("postgresql://...")

# Passa il checkpointer al momento della compilazione
graph = builder.compile(checkpointer=checkpointer)

# Il thread_id identifica la "sessione" — ogni thread ha la sua storia
config = {"configurable": {"thread_id": "sessione_utente_123"}}

# Prima chiamata
graph.invoke({"messages": [HumanMessage("Cerca il laptop")]}, config)

# Seconda chiamata — riprende automaticamente dal checkpoint precedente
graph.invoke({"messages": [HumanMessage("Quanto costa con il 15% di sconto?")]}, config)

# Ispeziona la storia completa degli step
for stato in graph.get_state_history(config):
    print(stato.config["configurable"]["checkpoint_id"])
    print(f"Step: {stato.metadata.get('step')}")
    print(stato.values["messages"])python

Ogni coppia (grafo compilato, thread_id) definisce una "conversazione" persistente. Due chiamate allo stesso grafo con lo stesso thread_id si comportano come due turni della stessa sessione: la seconda vede tutto ciò che è successo nella prima.

Human-in-the-loop

Questa è probabilmente la funzionalità più differenziante di LangGraph rispetto a tutti gli altri framework.

Immagina un agente AI che gestisce acquisti online per conto dell’utente. Vuoi che lavori in autonomia per ordini piccoli, ma che chieda conferma per qualsiasi ordine sopra i 500€. Non vuoi bloccare il flusso — vuoi solo inserire un checkpoint di approvazione nel momento giusto.

LangGraph lo rende possibile con interrupt():

from langgraph.types import interrupt, Command

def nodo_approvazione(state: State) -> dict:
    """Pausa il grafo e chiede approvazione umana."""

    # interrupt() serializza lo state, lo salva nel checkpointer
    # e FERMA l'esecuzione del grafo. Chi ha chiamato invoke() riceve
    # un'eccezione GraphInterrupt con il valore passato qui.
    risposta_umana = interrupt({
        "messaggio": "Approvare questo acquisto?",
        "dettagli": state["acquisto_corrente"],
        "importo": state["importo"]
    })

    # Quando l'umano risponde, l'esecuzione RIPRENDE esattamente qui
    # risposta_umana contiene ciò che l'umano ha passato a Command(resume=...)
    if risposta_umana["approvato"]:
        return {"stato_acquisto": "approvato"}
    else:
        return {"stato_acquisto": "rifiutato", "motivo": risposta_umana["motivo"]}python
from langgraph.errors import GraphInterrupt

config = {"configurable": {"thread_id": "ordine_42"}}

# Avvia il grafo — si fermerà all'interrupt
try:
    result = graph.invoke(initial_state, config)
except GraphInterrupt as e:
    # e.value contiene i dati passati a interrupt()
    print("In attesa di approvazione:", e.value)

    # ... mostra i dettagli all'utente, aspetta la sua decisione ...
    risposta = {"approvato": True}  # o False, con motivo

    # Riprendi passando la risposta come Command(resume=...)
    # Il thread_id deve essere lo stesso! LangGraph recupera il checkpoint.
    result = graph.invoke(
        Command(resume=risposta),
        config  # stesso thread_id
    )python
Human-in-the-loop richiede un checkpointer interrupt() funziona SOLO con un checkpointer configurato. Senza checkpoint il grafo non ha dove salvare lo state tra la pausa e la ripresa. In produzione usa PostgresSaver o RedisSaver — mai MemorySaver che perde tutto al restart.

Pattern human-in-the-loop

Ci sono quattro pattern principali per inserire il controllo umano in un flusso agentico:

  • Review-before-action: il grafo mostra cosa sta per fare l’agente ("Sto per inviare questa email a 500 destinatari") e aspetta conferma. Ideale per azioni irreversibili.
  • Edit-before-action: come review, ma l’utente può anche modificare i parametri dell’azione prima di approvarla ("Cambia il soggetto da X a Y, poi invia").
  • Approval gates: in flussi multi-step con approvazione gerarchica — l’agente completa la bozza, passa al team lead per review, che poi passa al CFO per approvazione finale.
  • Escalation: il grafo tenta di risolvere autonomamente; se non riesce dopo N tentativi, o se il contesto è ambiguo, interrompe e passa a un operatore umano con tutto il contesto.

Architetture multi-agent

LangGraph è la scelta naturale per sistemi multi-agent perché lo State condiviso e gli edge condizionali rendono la comunicazione tra agenti esplicita e tracciabile.

Il pattern più comune è il supervisor: un orchestratore che riceve le richieste e le smista ad agenti specializzati, raccogliendo i risultati.

from langchain_core.messages import SystemMessage
from langgraph.graph import MessagesState
import json

def agente_ricerca(state: MessagesState) -> dict:
    """Specialista: cerca informazioni online."""
    # ... logica di ricerca con web tools ...
    return {"messages": [AIMessage(content="Risultati ricerca: ...")]}

def agente_analisi(state: MessagesState) -> dict:
    """Specialista: analizza dati e trae conclusioni."""
    # ... logica di analisi ...
    return {"messages": [AIMessage(content="Analisi completata: ...")]}

def supervisor(state: MessagesState) -> dict:
    """Orchestratore: decide quale agente chiamare."""
    sistema = """Sei un supervisor con accesso a questi agenti:
    - ricerca: cerca informazioni online e su documenti
    - analisi: analizza dati e produce conclusioni
    - FINISH: quando il task è completato e puoi rispondere all'utente

    Rispondi SOLO con JSON: {"next": "ricerca" | "analisi" | "FINISH", "ragione": "..."}"""

    risposta = llm.invoke([SystemMessage(sistema)] + state["messages"])
    dati = json.loads(risposta.content)
    return {"messages": [risposta], "next": dati["next"]}

def router_supervisor(state: MessagesState) -> str:
    return state["next"]

builder = StateGraph(MessagesState)
builder.add_node("supervisor", supervisor)
builder.add_node("ricerca", agente_ricerca)
builder.add_node("analisi", agente_analisi)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges(
    "supervisor", router_supervisor,
    {"ricerca": "ricerca", "analisi": "analisi", "FINISH": END}
)
# Gli agenti specializzati tornano sempre al supervisor
builder.add_edge("ricerca", "supervisor")
builder.add_edge("analisi", "supervisor")python

Questo pattern è quello usato da ChipsBot: un orchestratore centrale che conosce le competenze di ogni agente specialista e smista le richieste ottimizzando per la risposta migliore.

Subgraphs: modularità

I subgraph permettono di costruire grafi riutilizzabili e di comporre sistemi complessi da blocchi più semplici. Un grafo compilato può essere usato come nodo in un altro grafo:

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class RagState(TypedDict):
    messages: Annotated[list, add_messages]
    context: str

def crea_sottografo_rag():
    """Restituisce un grafo RAG riutilizzabile."""
    builder = StateGraph(RagState)
    builder.add_node("retrieval", nodo_retrieval)
    builder.add_node("generation", nodo_generation)
    builder.add_edge(START, "retrieval")
    builder.add_edge("retrieval", "generation")
    builder.add_edge("generation", END)
    return builder.compile()

# Il subgraph è un grafo compilato
rag_subgraph = crea_sottografo_rag()

# Lo usi come qualsiasi altro nodo nel grafo principale
main_builder = StateGraph(MainState)
main_builder.add_node("rag", rag_subgraph)  # il subgraph è il nodo!
main_builder.add_node("post_processing", nodo_post_processing)
main_builder.add_edge(START, "rag")
main_builder.add_edge("rag", "post_processing")python

I subgraph hanno il proprio State interno — non condividono tutto con il grafo padre. Ci deve essere un meccanismo di mappatura tra i campi del grafo padre e quelli del subgraph. LangGraph gestisce questo con trasformazioni opzionali a livello di add_node.

LangGraph vs altri framework

LangGraph non è sempre la scelta giusta. Una comparazione onesta:

Framework Modello State H-i-t-L Curva
LangGraph Grafo diretto ✅ Avanzato ✅ Nativo Alta
CrewAI Team di ruoli ❌ Limitato ❌ No Bassa
AutoGen Conversazione ✅ Medio ⚠️ Parziale Media
Anthropic SDK Loop libero ❌ Manuale ❌ Manuale Bassa

LangGraph è la scelta giusta quando:

  • Il flusso ha branch multipli e la logica di routing è importante
  • Hai bisogno di human-in-the-loop in punti specifici del flusso
  • L’agente è long-running e deve sopravvivere a crash (checkpoints)
  • Stai costruendo un sistema multi-agent con coordinazione sofisticata
  • Hai bisogno di auditability: ogni step è tracciabile e replayable

Usa qualcosa di più semplice quando:

  • Il task è lineare senza branch: una chain LCEL è sufficiente
  • Stai prototipando rapidamente: CrewAI è più veloce da configurare
  • Usi solo Claude senza tool complessi: Anthropic SDK è più diretto

In produzione: LangGraph Platform

LangGraph Platform (disponibile in versione cloud e self-hosted) aggiunge l’infrastruttura necessaria per portare i grafi in produzione senza costruirla da zero:

  • Persistence: PostgreSQL integrato per checkpoints, nessun setup manuale
  • Streaming: SSE (Server-Sent Events) per aggiornamenti real-time step-per-step
  • Async task queue: per agenti long-running (minuti, ore). Il client può disconnettersi e riconnettersi più tardi per vedere i risultati
  • Cron scheduling: esecuzioni periodiche di grafi (es. monitoraggio notturno)
  • API REST automatica: ogni grafo viene esposto come endpoint REST con documentazione OpenAPI generata automaticamente
  • LangGraph Studio: interfaccia visuale per esplorare grafi, ispezionare State, testare step-by-step

Setup locale per sviluppo:

pip install langgraph-cli

# Crea un nuovo progetto LangGraph
langgraph new my-agent

# Avvia il server di sviluppo locale su :8123
langgraph dev

# Studio disponibile su https://smith.langchain.com/studio
# Collegati al server locale per debug visualebash

Il server locale espone automaticamente tre endpoint per ogni grafo compilato: /invoke (sincrono), /stream (SSE), /threads/{id}/state (ispezione State). LangGraph Studio si connette a questi endpoint per la visualizzazione interattiva.

Per approfondire