15. Costruire il tuo primo grafo con LangGraph

Nella Parte IV, abbiamo definito gli strumenti, li abbiamo collegati a un LLM e completato un ciclo dell'agente che ripete il ciclo decidi-ed-esegui. Guidare il ciclo, eseguire gli strumenti quando l'LLM li richiedeva, sapere quando fermarsi — abbiamo scritto a mano ogni parte di quel flusso.

In questo capitolo, costruiremo lo stesso agente in un modo completamente diverso. Invece di scrivere direttamente il flusso, registreremo passi (nodi) e regole di connessione (edge) con il framework LangGraph e lasceremo che sia esso a gestire l'esecuzione. Il comportamento è identico a quello del Capitolo 14, ma cambia il modo in cui lo costruiamo.

Questo capitolo tratta quattro concetti fondamentali — StateGraph, nodi, edge e State — e poi rifattorizza il ciclo dell'agente del Capitolo 14 in un grafo LangGraph. Nei capitoli successivi, il Capitolo 16 tratta il routing condizionale e i componenti predefiniti, mentre il Capitolo 17 tratta la persistenza dello stato che permette a un agente di riprendere dal punto in cui è stato interrotto.

15.1) Perché i grafi?

15.1.1) Limiti del ciclo dell'agente esistente

Rivediamo il ciclo dell'agente del Capitolo 14. Eliminando la gestione degli errori e altri dettagli, la struttura fondamentale era questa:

# Ciclo dell'agente del Capitolo 14 — struttura fondamentale (semplificata)
messages = [
    SystemMessage(content="You are a helpful assistant."),
    HumanMessage(content=user_input),
]
 
for step in range(max_steps):
    # Chiedi all'LLM di decidere l'azione successiva
    ai_message = llm_with_tools.invoke(messages)
    messages.append(ai_message)
 
    # Se non viene richiesta alcuna chiamata a uno strumento, restituisci la risposta finale
    if not ai_message.tool_calls:
        return ai_message.content
 
    # Esegui gli strumenti richiesti
    for tool_call in ai_message.tool_calls:
        selected_tool = tool_map[tool_call["name"]]
        tool_message = selected_tool.invoke(tool_call)
        messages.append(tool_message)

Questo codice copre solo le basi e nient'altro. In un vero ambiente di produzione, però, serve molto di più. Ecco alcuni esempi.

• Ripristino dopo un crash — Se un agente si blocca al passo 7 di un'attività di ricerca in 10 passi, dovrebbe essere in grado di riprendere dal passo 7 invece di ricominciare da capo.
• Richieste di approvazione — Prima che un agente esegua un'operazione critica, dovrebbe essere in grado di mettersi in pausa e chiedere a un umano "Va bene procedere?"
• Monitoraggio in tempo reale — Gli utenti dovrebbero poter vedere cosa sta facendo attualmente l'agente e quali strumenti sta chiamando.
• Visualizzazione e debug — Dovrebbe essere disponibile un diagramma che mostra come opera l'agente, così che quando sorgono problemi si possa tracciare quale passo è andato storto.

Implementare queste funzionalità da soli non è impossibile, ma non è nemmeno facile. Il solo ripristino dopo un crash richiede di scrivere codice per serializzare lo stato a ogni passo, salvarlo su disco, ripristinarlo e riprendere esattamente nella posizione corretta. Potresti ritrovarti con più codice infrastrutturale che logica di business.

LangGraph è stato costruito per fornire queste funzionalità a livello di framework. Ripristino dopo un crash, richieste di approvazione, monitoraggio, visualizzazione — il framework gestisce tutto questo. Ma c'è un requisito: devi costruire il tuo agente in una struttura che il framework possa comprendere.

Il ciclo dell'agente del Capitolo 14 gestisce tutta la logica direttamente, quindi non c'è nulla a cui il framework possa agganciarsi. Per sfruttare ciò che LangGraph offre, dobbiamo ricostruire l'agente in una struttura che LangGraph comprende — un grafo. Di questo tratta il presente capitolo.

15.1.2) Cos'è LangGraph?

LangGraph è un framework di orchestrazione che definisce ed esegue i flussi di lavoro degli agenti come grafi. Un grafo, in questo contesto, indica una struttura in cui ogni nodo (passo) che l'agente esegue è collegato tramite edge (regole di connessione).

In LangGraph, scomponi il flusso di lavoro in nodi indipendenti e li colleghi con gli edge. LangGraph poi percorre il grafo, eseguendo ogni nodo lungo il percorso. Ecco come appare il ciclo dell'agente del Capitolo 14 espresso come grafo:

I riquadri rettangolari sono i nodi, e le frecce sono gli edge. Il rombo rappresenta un edge condizionale che si dirama verso percorsi diversi in base a una condizione.

Nel Capitolo 14, l'intero flusso di lavoro viveva in cicli for, controlli if e altro codice scritto a mano. Con LangGraph, definisci cosa fa ogni nodo e colleghi i nodi tra loro con gli edge. In breve, passi dal programmare il flusso di lavoro al dichiararlo come struttura.

LangGraph non sostituisce nulla di ciò che hai imparato nei Capitoli 12–14. Definizioni degli strumenti, bind_tools(), tool_calls, ToolMessage — tutti questi elementi vengono ancora usati all'interno dei nodi, esattamente come prima.

La sezione successiva tratta i componenti fondamentali di LangGraph — StateGraph, State, nodi ed edge — uno alla volta.

15.2) Componenti di LangGraph: StateGraph, State, nodi, edge

Questa sezione illustra i quattro componenti fondamentali di LangGraph uno alla volta. Inizieremo con StateGraph — la classe che raggruppa State, nodi ed edge in un grafo — e poi tratteremo ciascuna delle parti (State, nodi, edge) che vi risiedono all'interno.

15.2.1) StateGraph

StateGraph è la classe usata per costruire i grafi in LangGraph. Specifichi lo State che il grafo gestirà, aggiungi i nodi, li colleghi con gli edge e poi compili per produrre un grafo eseguibile.

Vediamo come funziona.

Creare un'istanza di StateGraph

Chiama il costruttore StateGraph per creare un'istanza. Devi passare lo schema dello State (la classe stessa) come parametro. Qui stiamo usando MessagesState, uno State predefinito che LangGraph fornisce per gestire le liste di messaggi. Tratteremo i dettagli in 15.2.2.

from langgraph.graph import StateGraph, MessagesState
 
builder = StateGraph(MessagesState)

Aggiungere i nodi

Usa add_node() per registrare un nodo. Un nodo è una funzione Python che prende lo State corrente e restituisce le parti che desidera modificare. Tratteremo in dettaglio le funzioni dei nodi in 15.2.3.

def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}
 
builder.add_node(say_hello)    # il nome del nodo diventa "say_hello"

Collegare gli edge

Usa add_edge(source, target) per collegare i nodi. source è dove inizia l'edge; target è dove arriva. START e END sono marcatori speciali per i punti di ingresso e uscita del grafo. Tratteremo gli edge in 15.2.4.

from langgraph.graph import START, END
 
builder.add_edge(START, "say_hello")   # il grafo inizia → esegui say_hello
builder.add_edge("say_hello", END)     # say_hello completa → termina il grafo

Compilare ed eseguire

Chiamare compile() convalida la struttura del grafo e produce un oggetto eseguibile. Esegui il grafo compilato con invoke(), passando i valori iniziali dello State.

graph = builder.compile()
 
initial_state = {"messages": [{"role": "user", "content": "hi!"}]}
result = graph.invoke(initial_state)

Ora mettiamo tutto insieme e costruiamo un semplice grafo:

from langgraph.graph import StateGraph, MessagesState, START, END
 
def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}
 
builder = StateGraph(MessagesState)
 
builder.add_node(say_hello)
builder.add_edge(START, "say_hello")
builder.add_edge("say_hello", END)
 
graph = builder.compile()
 
initial_state = {"messages": [{"role": "user", "content": "hi!"}]}
result = graph.invoke(initial_state)
print(result["messages"][-1].content)

Output:

hello world

Quando chiami invoke(), il grafo viene eseguito nell'ordine START → say_hello → END. say_hello ha restituito un dizionario con messages come chiave, e quel valore è stato aggiunto alla lista messages in MessagesState. Approfondiremo come funziona in 15.2.2. Il risultato è che estraendo il contenuto dell'ultimo messaggio otteniamo "hello world".

15.2.2) State: i dati che scorrono attraverso il grafo

Lo State è il dato che ogni nodo nel grafo condivide. Quando un nodo viene eseguito, riceve lo State corrente, svolge il suo lavoro e restituisce solo le parti che desidera modificare. LangGraph reintegra queste modifiche nello State e consegna la versione aggiornata al nodo successivo.

Definire lo State

Definisci lo State creando una sottoclasse di TypedDict. Scegli i campi e i tipi che corrispondono a ciò che il tuo agente deve tracciare. Ecco un esempio semplice:

from typing_extensions import TypedDict
 
class AgentState(TypedDict):
    messages: list       # lista dei messaggi
    llm_calls: int       # conteggio delle chiamate all'LLM

Da qui, passi AgentState quando crei uno StateGraph e lo usi come type hint per le tue funzioni dei nodi.

Reducer

Quando un nodo restituisce un valore, il campo State corrispondente viene aggiornato. Il comportamento predefinito è la sovrascrittura — se un nodo restituisce {"llm_calls": 3}, llm_calls diventa semplicemente 3, indipendentemente da quale fosse prima.

Ma alcuni campi hanno bisogno dell'aggiunta, non della sovrascrittura. Cosa succede se messages viene sovrascritto? Ogni volta che un nodo restituisce un nuovo messaggio, l'intera cronologia della conversazione svanisce. Per messages, l'aggiunta è il comportamento corretto.

LangGraph ti permette di impostare una strategia di aggiornamento diversa per ogni campo tramite una funzione reducer. Specifichi il reducer come secondo argomento in Annotated:

from typing_extensions import TypedDict, Annotated
from langgraph.graph.message import add_messages
 
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]   # reducer: aggiungi
    llm_calls: int                            # nessun reducer: sovrascrivi

add_messages è un reducer fornito da LangGraph. Invece di sostituire la lista, aggiunge i nuovi messaggi a quelli già presenti. Poiché questo reducer è impostato sul campo messages, qualsiasi valore che un nodo restituisce per messages viene aggiunto. llm_calls non ha alcun reducer, quindi i valori restituiti sovrascrivono semplicemente ciò che era presente prima.

È esattamente per questo che il messaggio che say_hello ha restituito in 15.2.1 è stato aggiunto a messages invece di sostituirlo — se ne è occupato il reducer.

MessagesState

LangGraph include uno State predefinito chiamato MessagesState. Ecco come appare sotto il cofano:

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

La stessa struttura che abbiamo appena trattato — un campo messages con il reducer add_messages già collegato.

Se hai bisogno di campi aggiuntivi, basta creare una sottoclasse:

from langgraph.graph import MessagesState
 
class AgentState(MessagesState):
    llm_calls: int  # comportamento predefinito di sovrascrittura

15.2.3) Nodi: funzioni che aggiornano lo State

Un nodo è una funzione Python che esegue un singolo, specifico compito all'interno del grafo.

def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}

Due cose da sapere quando scrivi le funzioni dei nodi:

Regola 1: Riceve lo State corrente come argomento. LangGraph passa l'oggetto State corrente quando esegue il nodo.

Regola 2: Restituisce solo le parti che desidera modificare, non l'intero State. I nodi non modificano lo State direttamente. Restituisci semplicemente i campi che desideri aggiornare, e LangGraph li unisce allo State esistente secondo le regole del reducer di ogni campo.

Usa add_node() per aggiungere un nodo allo StateGraph:

builder.add_node(say_hello)          # il nome della funzione "say_hello" diventa il nome del nodo
builder.add_node("my_node", my_func) # puoi anche specificare il nome esplicitamente

15.2.4) Edge: regole che collegano i nodi

Un edge determina "dopo che questo nodo termina, cosa viene eseguito dopo?" Ce ne sono di due tipi.

Edge normali

Un edge normale collega un "vai a" fisso tra due nodi. Usa add_edge(source, target) — source è il nodo di partenza, target è la destinazione.

builder.add_edge(START, "say_hello")       # quando il grafo inizia, esegui say_hello
builder.add_edge("say_hello", "llm_call")  # dopo say_hello, esegui llm_call
builder.add_edge("llm_call", END)          # dopo llm_call, termina il grafo

Edge condizionali

Un edge condizionale sceglie il nodo successivo in fase di esecuzione in base allo State corrente. Usa add_conditional_edges(source, routing_function) — source è il nodo di partenza, e routing_function è una funzione che prende lo State corrente e restituisce il nome del nodo successivo:

from langgraph.graph import END
 
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"   # chiamata a uno strumento richiesta → vai a tool_node
    return END               # nessuna chiamata a uno strumento → termina
 
builder.add_conditional_edges("llm_call", should_continue)

add_conditional_edges("llm_call", should_continue) dice a LangGraph: "quando llm_call termina, chiama should_continue per decidere cosa viene eseguito dopo." should_continue instrada verso "tool_node" se l'ultimo messaggio ha tool_calls, oppure verso END se non li ha. In pratica, questo significa che il grafo continua verso il nodo di esecuzione dello strumento quando l'LLM richiede una chiamata a uno strumento, e termina quando non lo fa.

Ora che abbiamo trattato tutti e quattro i componenti, la sezione successiva li utilizza per rifattorizzare il ciclo dell'agente del Capitolo 14 in un grafo LangGraph.

15.3) Rifattorizzare il ciclo dell'agente in un grafo

Ricostruiamo il ciclo dell'agente del Capitolo 14 usando LangGraph. Il comportamento è identico a quello del Capitolo 14 — l'LLM decide, gli strumenti vengono eseguiti in base alle richieste dell'LLM, e il ciclo si ripete fino al completamento. L'unica cosa che cambia è come strutturiamo questo flusso.

Ecco come apparirà il grafo finito:

Il grafo cicla tra llm_call e tool_node finché l'LLM smette di richiedere chiamate agli strumenti, a quel punto esce verso END. Costruiamolo passo dopo passo.

15.3.1) Definire lo State

Creiamo una sottoclasse di MessagesState da 15.2.2 per definire lo State dell'agente. Il campo messages è ereditato da MessagesState, e aggiungiamo un campo llm_calls per tracciare il numero di chiamate all'LLM.

from langgraph.graph import MessagesState
 
class AgentState(MessagesState):
    llm_calls: int    # conteggio delle chiamate all'LLM (sovrascrittura)

La lista messages accumulerà in ordine gli input degli utenti (HumanMessage), le risposte dell'LLM (AIMessage) e i risultati dell'esecuzione degli strumenti (ToolMessage).

15.3.2) Costruire i nodi

Per prima cosa, configuriamo gli strumenti e il modello dal Capitolo 14:

from langchain_openai import ChatOpenAI
from langchain_core.messages import ToolMessage
from langchain.tools import tool
 
@tool
def get_weather(city: str) -> str:
    """Ottieni il meteo attuale per una città."""
    fake_data = {"Tokyo": "18°C, cloudy", "Cairo": "31°C, sunny"}
    return fake_data.get(city, f"No weather data for {city}.")
 
@tool
def calculate(expression: str) -> str:
    """Calcola una semplice espressione aritmetica. Esempio: '3 * 21'."""
    return str(eval(expression))  # Attenzione: eval() è un rischio per la sicurezza. Non usarlo in produzione.
 
tools = [get_weather, calculate]
tool_map = {t.name: t for t in tools}
 
llm = ChatOpenAI(model="gpt-5-mini")
model_with_tools = llm.bind_tools(tools)

Ora scriviamo le due funzioni dei nodi.

Nodo llm_call — Chiama l'LLM e restituisce la risposta:

def llm_call(state: AgentState):
    """Chiama l'LLM e restituisce la risposta."""
    response = model_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "llm_calls": state.get("llm_calls", 0) + 1,
    }

model_with_tools.invoke() chiama l'LLM, e la risposta viene impacchettata sotto la chiave messages nel dizionario restituito. Il reducer la aggiunge ai messages esistenti in AgentState. llm_calls restituisce il conteggio corrente più 1, sovrascrivendo il valore precedente.

Nodo tool_node — Esegue gli strumenti richiesti dall'LLM e restituisce i risultati:

def tool_node(state: AgentState):
    """Esegue gli strumenti richiesti dall'LLM."""
    last_message = state["messages"][-1]
    results = []
    for tool_call in last_message.tool_calls:
        selected_tool = tool_map[tool_call["name"]]
        tool_message = selected_tool.invoke(tool_call)
        results.append(tool_message)
    return {"messages": results}

Poiché tool_node viene sempre eseguito subito dopo llm_call, l'ultimo messaggio in messages è garantito essere l'AIMessage che l'LLM ha appena prodotto. Il campo tool_calls di quel messaggio contiene le chiamate agli strumenti che l'LLM ha richiesto. Il nodo esegue ogni strumento, raccoglie i risultati in results e li restituisce sotto la chiave messages — il reducer si occupa di aggiungerli alla lista esistente.

15.3.3) Edge condizionale

Una volta che llm_call termina, abbiamo bisogno di un edge condizionale per decidere se eseguire tool_node o terminare il grafo. Questo segue lo stesso schema di 15.2.4:

from typing import Literal
from langgraph.graph import END
 
def should_continue(state: AgentState) -> Literal["tool_node", "__end__"]:
    """Decide se eseguire gli strumenti o terminare il grafo."""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"
    return END

Se last_message.tool_calls è presente, l'LLM sta chiedendo una chiamata a uno strumento, quindi instradiamo verso tool_node. Altrimenti, instradiamo verso END e il grafo termina.

Il type hint di ritorno Literal["tool_node", "__end__"] dichiara le possibili destinazioni che questa funzione può restituire. LangGraph ha bisogno di questo hint per disegnare correttamente i percorsi degli edge condizionali nelle visualizzazioni del grafo. Non ha alcun effetto sul comportamento a runtime.

"__end__" è il valore stringa sottostante di END. Poiché Literal accetta solo letterali stringa, scriviamo "__end__" invece di END.

15.3.4) Assemblare ed eseguire il grafo

È ora di collegare tutto insieme. Assembliamo State, nodi ed edge condizionale in uno StateGraph e compiliamo:

from langgraph.graph import StateGraph, START, END
 
builder = StateGraph(AgentState)
 
builder.add_node("llm_call", llm_call)
builder.add_node("tool_node", tool_node)
 
builder.add_edge(START, "llm_call")                         # inizio → llm_call
builder.add_conditional_edges("llm_call", should_continue)  # llm_call → tool_node o END
builder.add_edge("tool_node", "llm_call")                   # tool_node → llm_call (ciclo)
 
agent = builder.compile()

L'edge da tool_node di ritorno verso llm_call crea un ciclo. L'esecuzione continua a ciclare finché l'LLM risponde con una risposta finale invece di richiedere un'altra chiamata a uno strumento, a quel punto il ciclo esce.

Eseguiamolo:

from langchain_core.messages import HumanMessage
 
result = agent.invoke({
    "messages": [HumanMessage(content="Ottieni la temperatura al Cairo, poi moltiplica il numero per 3.")],
    "llm_calls": 0,
})
 
print(result["messages"][-1].content)
print(f"\nTotale chiamate all'LLM: {result['llm_calls']}")

Output:

Temperatura attuale al Cairo: 31°C. Moltiplicata per 3 = 93.
 
Totale chiamate all'LLM: 3

L'agente ha chiamato get_weather("Cairo"), ha visto il risultato, ha chiamato calculate("31 * 3") e ha prodotto la risposta finale — lo stesso risultato ottenuto nel Capitolo 14.

Ispezionando l'intera cronologia dei messaggi si vede ogni passo registrato in messages, in ordine:

for message in result["messages"]:
    message.pretty_print()

Output:

================================ Human Message =================================
Ottieni la temperatura al Cairo, poi moltiplica il numero per 3.
================================== Ai Message ==================================
Tool Calls:
  get_weather (call_DiL9WF)
  Args:
    city: Cairo
================================= Tool Message =================================
Name: get_weather
31°C, sunny
================================== Ai Message ==================================
Tool Calls:
  calculate (call_wa6RqWST)
  Args:
    expression: 31 * 3
================================= Tool Message =================================
Name: calculate
93
================================== Ai Message ==================================
Temperatura attuale al Cairo: 31°C. Moltiplicata per 3 = 93.

15.3.5) Visualizzazione del grafo

In un notebook Jupyter, agent.get_graph().draw_mermaid_png() renderizza la struttura del grafo come immagine direttamente nell'output della cella.

from IPython.display import Image, display
 
display(Image(agent.get_graph().draw_mermaid_png()))

In un ambiente terminale, salvalo invece come file PNG.

agent.get_graph().draw_mermaid_png(output_file_path="agent_graph.png")

Immagine generata:

Le linee continue sono edge normali e le linee tratteggiate sono edge condizionali. Questo diagramma viene generato automaticamente dal codice.

15.3.6) Limite di ricorsione

Proprio come abbiamo usato max_steps per proteggerci dai cicli infiniti nel Capitolo 14, LangGraph ha una rete di sicurezza integrata. Ogni volta che un nodo viene eseguito durante l'esecuzione del grafo, un contatore interno si incrementa di uno. Quando quel contatore supera il limite configurato, LangGraph solleva un GraphRecursionError.

Per vedere come funziona il conteggio, guarda l'esecuzione precedente. Chiamare sia get_weather che calculate ha visitato i nodi in questo ordine:

llm_call(1) → tool_node(2) → llm_call(3) → tool_node(4) → llm_call(5) → END

In totale sono 5 visite ai nodi. Se imposti recursion_limit a 3, il limite scatta alla 3ª visita e l'esecuzione viene interrotta:

from langgraph.errors import GraphRecursionError
 
try:
    result = agent.invoke(
        {"messages": [HumanMessage(content="Ottieni la temperatura al Cairo, poi moltiplica il numero per 3.")],
         "llm_calls": 0},
        config={"recursion_limit": 3},
    )
except GraphRecursionError:
    print("L'agente ha raggiunto il limite di ricorsione — interruzione dell'esecuzione.")

Output:

L'agente ha raggiunto il limite di ricorsione — interruzione dell'esecuzione.

Imposta il limite passando config={"recursion_limit": numero} a invoke(). Il valore giusto dipende dal tuo caso d'uso e dalla complessità del tuo grafo. Inizia con un numero generoso e regolalo attraverso i test.