03 - LangChain e LangGraph: Agenti con AgentExecutor e Tools

Il problema: i tutorial su LangChain non girano più

Se copi un esempio di agente LangChain scritto un anno fa, probabilmente comincia con from langchain.agents import AgentExecutor e oggi si ferma con un errore di importazione. Non è un tuo errore: l'ecosistema è stato riorganizzato. La risposta breve a «come costruisco un agente oggi» è: con create_agent del pacchetto langchain se ti basta il ciclo già pronto, con un StateGraph di LangGraph se vuoi controllare tu i passi, gli stati salvati e le pause per l'approvazione umana. Questo articolo mostra entrambe le strade, e per non dipendere da un servizio a pagamento usa un modello finto che segue una regola fissa: i listati sono eseguiti davvero, ma provano la struttura, non l'intelligenza di un modello.

Prima di tutto la misura. Il blocco seguente tenta l'importazione di AgentExecutor da langchain.agents (la strada dei vecchi tutorial) e poi dal pacchetto langchain-classic, e stampa le versioni installate in un ambiente virtuale pulito. Gli esempi dell'articolo sono stati eseguiti con Python 3.14.4 e queste versioni, installate da PyPI il 2026-11-12.

import importlib.metadata as md

try:
    from langchain.agents import AgentExecutor
    print("langchain.agents.AgentExecutor: importato")
except ImportError as e:
    print("langchain.agents.AgentExecutor:", type(e).__name__, "da", e.name)

from langchain_classic.agents import AgentExecutor
print("langchain_classic.agents.AgentExecutor:", AgentExecutor.__module__)
for pacchetto in ("langchain", "langchain-classic", "langgraph"):
    print(pacchetto, md.version(pacchetto))

# Output:
# langchain.agents.AgentExecutor: ImportError da langchain.agents
# langchain_classic.agents.AgentExecutor: langchain_classic.agents.agent
# langchain 1.5.0
# langchain-classic 1.0.8
# langgraph 1.2.21

Con langchain 1.5.0 l'importazione dalla vecchia posizione fallisce con ImportError, mentre langchain-classic 1.0.8 espone ancora la classe. La guida ufficiale alla migrazione alla versione 1 dice che il codice legacy del pacchetto langchain (catene come LLMChain, recuperatori, indicizzazione, hub, funzionalità deprecate) è stato spostato in langchain-classic, e che al posto di langgraph.prebuilt.create_react_agent si raccomanda langchain.agents.create_agent. Non abbiamo eseguito un ciclo completo di AgentExecutor: il blocco prova solo dove vive la classe. Per codice nuovo non c'è motivo di cominciare da lì.

I pacchetti: chi fa che cosa

La tabella spiega perché le versioni contano: i numeri non sono allineati tra pacchetti (langgraph 1.2.21, langchain 1.5.0, langchain-classic 1.0.8, langchain-ollama 1.1.0, langchain-openai 1.7.4, langchain-anthropic 1.7.10 al 2026-11-12), e un esempio che non dichiara quali installa è fragile. Le integrazioni sono pacchetti a parte: se usi un modello locale servito da Ollama, il pacchetto è langchain-ollama, non il vecchio import da langchain-community che compare nei tutorial. Nominare un fornitore qui è questione di compatibilità: l'articolo non ne presuppone nessuno.

Il modello finto, e perché lo usiamo

Per provare un agente senza chiamare un modello basta una sottoclasse di BaseChatModel con due metodi: bind_tools, che qui restituisce se stessa perché il modello finto non legge gli schemi, e _generate, che decide la risposta guardando l'ultimo messaggio. Se l'ultimo è un risultato di strumento (ToolMessage), risponde con una frase finale; altrimenti propone la chiamata a cerca_ordine con l'argomento A-1042. È la stessa forma di scambio di un modello vero (richiesta, chiamata di strumento, risultato, risposta), ma senza ragionamento: lo useremo solo per verificare che il cablaggio funzioni. Non dimostra che un modello reale sceglierebbe lo strumento giusto.

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.outputs import ChatGeneration, ChatResult


class ModelloFinto(BaseChatModel):
    """Si comporta come un modello con strumenti, ma segue una regola fissa:
    dopo una domanda chiama lo strumento; dopo il risultato risponde."""

    @property
    def _llm_type(self) -> str:
        return "finto"

    def bind_tools(self, tools, **kwargs):
        return self                      # il modello finto non ha bisogno degli schemi

    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        ultimo = messages[-1]
        if isinstance(ultimo, ToolMessage):
            msg = AIMessage(content=f"Risposta finale: {ultimo.content}")
        else:
            msg = AIMessage(content="", tool_calls=[
                {"name": "cerca_ordine", "args": {"id": "A-1042"}, "id": "call_1"}])
        return ChatResult(generations=[ChatGeneration(message=msg)])

Il file si chiama modello.py e viene importato dai blocchi che seguono; non stampa niente. Un dettaglio che conta: il campo id nella chiamata (call_1) è obbligatorio, perché serve ad abbinare il risultato alla richiesta; omettendolo, l'esecuzione si ferma con TypeError: tool_call() missing 1 required keyword-only argument: 'id' (provato).

Gli strumenti: nome, descrizione, schema

Il decoratore @tool di langchain_core.tools trasforma una funzione in uno strumento: il nome viene dalla funzione, la descrizione dal docstring, lo schema degli argomenti dai type hint. Il modello vede solo questi tre elementi, quindi il docstring è prompt a tutti gli effetti: dice quando usare lo strumento e quando no. Il blocco usa un dizionario al posto di un servizio di ricerca, per restare eseguibile offline, e stampa ciò che il modello vedrebbe, poi invoca lo strumento con argomenti giusti e con un nome di argomento sbagliato.

from langchain_core.tools import tool

ORDINI = {"A-1042": "in consegna", "A-1043": "evaso"}

@tool
def cerca_ordine(id: str) -> str:
    """Restituisce lo stato di un ordine dato il suo codice (formato A-NNNN).
    Usalo per domande sullo stato di un ordine, non per prezzi o resi."""
    return ORDINI.get(id, "ordine inesistente")

print(cerca_ordine.name)
print(cerca_ordine.description)
print(cerca_ordine.args)
print(cerca_ordine.invoke({"id": "A-1042"}))
try:
    cerca_ordine.invoke({"identificativo": "A-1042"})
except Exception as e:
    print(type(e).__name__, "-", str(e).splitlines()[0])

# Output:
# cerca_ordine
# Restituisce lo stato di un ordine dato il suo codice (formato A-NNNN).
# Usalo per domande sullo stato di un ordine, non per prezzi o resi.
# {'id': {'title': 'Id', 'type': 'string'}}
# in consegna
# ValidationError - 1 validation error for cerca_ordine

L'uscita mostra lo schema generato ({'id': {'title': 'Id', 'type': 'string'}}) e il fatto che una chiamata con l'argomento identificativo invece di id non arriva alla funzione: la validazione di Pydantic la blocca con ValidationError. In un agente reale quell'errore deve tornare al modello come testo, altrimenti non può correggersi. Se togli il type hint str, lo schema perde il tipo e il modello riceve meno indicazioni su come riempire il campo.

Strumenti che toccano dati: filtrare il testo non basta

Un esempio frequente nei tutorial è uno strumento query_database che accetta SQL e lo esegue «solo se comincia per SELECT». Quel controllo sul testo non è una barriera: la protezione vera si mette dove la scrittura è impossibile. Con SQLite, per esempio, aprire il file con sqlite3.connect("file:app.db?mode=ro", uri=True) fa fallire ogni scrittura con OperationalError: attempt to write a readonly database (provato su un database temporaneo). Lo stesso criterio vale per i file: limita la cartella leggibile e la dimensione letta, e non fidarti di un percorso scelto dal modello. Altre misure sono in Security e Safety: Jailbreaking e Guardrails.

Il ciclo come grafo: LangGraph

La documentazione di LangGraph lo descrive come framework di orchestrazione di basso livello e runtime per agenti con stato di lunga durata: definisci flussi come grafi di stato condiviso, nodi e archi, e puoi mescolare passi deterministici e passi guidati dal modello. Nel listato lo stato è MessagesState (una lista di messaggi), il nodo agent chiama il modello, il nodo tools esegue le chiamate richieste tramite ToolNode, e l'arco condizionale tools_condition manda il flusso a tools se l'ultimo messaggio contiene chiamate di strumento, altrimenti alla fine. L'arco da tools ad agent chiude il ciclo: è l'orchestratore del primo articolo, reso esplicito.

from langchain_core.tools import tool
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
from modello import ModelloFinto

@tool
def cerca_ordine(id: str) -> str:
    """Restituisce lo stato di un ordine dato il suo codice (formato A-NNNN)."""
    return {"A-1042": "in consegna"}.get(id, "ordine inesistente")

tools = [cerca_ordine]
llm_con_strumenti = ModelloFinto().bind_tools(tools)

def agente(state: MessagesState):
    """Nodo che chiama il modello e aggiunge la sua risposta allo stato."""
    return {"messages": [llm_con_strumenti.invoke(state["messages"])]}

builder = StateGraph(MessagesState)
builder.add_node("agent", agente)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)   # tool_calls -> "tools", altrimenti END
builder.add_edge("tools", "agent")
grafo = builder.compile()

risultato = grafo.invoke({"messages": [("user", "Dov'è il mio ordine A-1042?")]})
for m in risultato["messages"]:
    contenuto = m.content or f"tool_calls={[(c['name'], c['args']) for c in m.tool_calls]}"
    print(f"{m.type:>5}: {contenuto}")

# Output:
# human: Dov'è il mio ordine A-1042?
#    ai: tool_calls=[('cerca_ordine', {'id': 'A-1042'})]
#  tool: in consegna
#    ai: Risposta finale: in consegna

Eseguito due volte, con uscita identica. La stampa mostra i quattro messaggi dello scambio: la domanda, la chiamata a cerca_ordine, il risultato «in consegna» e la risposta. Se rimuovi l'arco tools verso agent, il grafo si ferma dopo l'esecuzione dello strumento e la risposta finale non arriva mai: è l'errore più comune di chi scrive il grafo a mano. Il vantaggio rispetto a un ciclo while sta nel fatto che ogni nodo è una unità osservabile e salvabile; il costo è più struttura da scrivere.

La scorciatoia: create_agent

Se non ti serve cambiare la forma del ciclo, create_agent costruisce lo stesso grafo per te. La documentazione lo descrive come un involucro minimale e configurabile attorno al ciclo del modello, a cui componi modello, strumenti, prompt e middleware. Il blocco ripete l'esempio con la scorciatoia e conferma che la sequenza dei messaggi è la stessa.

from langchain.agents import create_agent
from langchain_core.tools import tool
from modello import ModelloFinto

@tool
def cerca_ordine(id: str) -> str:
    """Restituisce lo stato di un ordine dato il suo codice (formato A-NNNN)."""
    return {"A-1042": "in consegna"}.get(id, "ordine inesistente")

agente = create_agent(ModelloFinto(), [cerca_ordine],
                      system_prompt="Sei un assistente per lo stato degli ordini.")
out = agente.invoke({"messages": [("user", "Dov'è A-1042?")]})
print([m.type for m in out["messages"]])
print(out["messages"][-1].content)

# Output:
# ['human', 'ai', 'tool', 'ai']
# Risposta finale: in consegna

La sequenza human, ai, tool, ai è quella del grafo scritto a mano. Per decidere tra le due strade: parti da create_agent e scendi a StateGraph solo quando ti servono nodi tuoi, percorsi diversi a seconda dello stato o più agenti nello stesso grafo. Passare dall'una all'altra non è un ripensamento: il grafo sotto è lo stesso.

Stato salvato: thread e checkpoint

Compilando il grafo con un checkpointer (qui InMemorySaver, che è anche importabile come MemorySaver) lo stato viene salvato e ripreso per ciascun thread_id. Secondo la documentazione, thread_id è il puntatore con cui il checkpointer sa quale stato caricare. Il blocco usa un nodo finto che dichiara quanti messaggi vede, così si vede la persistenza senza un modello: due richieste sullo stesso thread e una su un thread diverso.

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import AIMessage

def conta(state: MessagesState):
    """Nodo finto: dice quanti messaggi ha visto, per mostrare che lo stato persiste."""
    return {"messages": [AIMessage(content=f"Messaggi nello stato: {len(state['messages']) + 1}")]}

builder = StateGraph(MessagesState)
builder.add_node("conta", conta)
builder.add_edge(START, "conta")
builder.add_edge("conta", END)
grafo = builder.compile(checkpointer=InMemorySaver())

for thread, testo in (("utente-1", "Mi chiamo Federico"), ("utente-1", "Come mi chiamo?"),
                      ("utente-2", "Ciao")):
    config = {"configurable": {"thread_id": thread}}
    out = grafo.invoke({"messages": [("user", testo)]}, config)
    print(thread, "->", out["messages"][-1].content)

# Output:
# utente-1 -> Messaggi nello stato: 2
# utente-1 -> Messaggi nello stato: 4
# utente-2 -> Messaggi nello stato: 2

Sullo stesso thread il conteggio sale da 2 a 4, perché la seconda richiesta parte dai messaggi della prima; sul thread utente-2 riparte da 2. Il salvataggio in memoria sparisce a ogni riavvio del processo: per una vera memoria tra sessioni serve un archivio duraturo, e per la memoria a lungo termine un progetto a parte, tema di Memory Systems: Context, History e Knowledge Graphs. Un altro rischio: se usi un unico thread_id per tutti gli utenti, condividi la conversazione di tutti.

Approvazione umana con interrupt

Per azioni irreversibili (inviare un'email, spostare denaro) il grafo deve potersi fermare e aspettare una persona. In LangGraph lo fa interrupt() dentro un nodo: sospende l'esecuzione, consegna al chiamante il valore passato, e riprende quando riceve Command(resume=...) con lo stesso thread_id. Serve il checkpointer, perché lo stato deve sopravvivere alla pausa. Nel listato il nodo chiede l'approvazione di un'email; il programma stampa la richiesta, lo stato in attesa, poi riprende con «approvato».

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.types import Command, interrupt

def chiedi_approvazione(state: MessagesState):
    """Si ferma e consegna al chiamante la descrizione dell'azione da approvare."""
    decisione = interrupt({"azione": "invia_email", "a": "cliente@example.com"})
    esito = "inviata" if decisione == "approvato" else "annullata"
    return {"messages": [("assistant", f"Email {esito}.")]}

builder = StateGraph(MessagesState)
builder.add_node("approva", chiedi_approvazione)
builder.add_edge(START, "approva")
builder.add_edge("approva", END)
grafo = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "t1"}}
primo = grafo.invoke({"messages": [("user", "Avvisa il cliente")]}, config)
print("interrotto:", primo["__interrupt__"][0].value)
print("prossimo nodo da eseguire:", grafo.get_state(config).next)
finale = grafo.invoke(Command(resume="approvato"), config)
print("dopo il resume:", finale["messages"][-1].content)

# Output:
# interrotto: {'azione': 'invia_email', 'a': 'cliente@example.com'}
# prossimo nodo da eseguire: ('approva',)
# dopo il resume: Email inviata.

Nell'uscita, la prima chiamata non termina il grafo: restituisce nella chiave __interrupt__ il payload, e get_state(config).next dice che il prossimo nodo da eseguire è ancora approva. Dopo il resume la funzione riparte dall'inizio del nodo (non dalla riga dell'interrupt), ed è per questo che ciò che precede interrupt() nel nodo non deve avere effetti collaterali: ripeterebbe l'effetto a ogni ripresa. Qui il valore di ritorno decide soltanto il testo, non invia nulla.

Limiti e resilienza

Un agente in produzione ha due modi di fallire: gira troppo o cade per un errore esterno. Contro il primo, LangGraph ha un limite di passi, recursion_limit, passato nella configurazione. Il blocco costruisce un nodo ostinato che richiama sempre lo stesso strumento e lo ferma a 6 passi, catturando GraphRecursionError.

from langchain_core.messages import AIMessage
from langgraph.errors import GraphRecursionError
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.tools import tool

@tool
def cerca_ordine(id: str) -> str:
    """Restituisce lo stato di un ordine."""
    return "ordine inesistente"

def testardo(state: MessagesState):
    """Nodo finto: richiama sempre lo stesso strumento, senza mai rispondere."""
    chiamata = {"name": "cerca_ordine", "args": {"id": "A-9999"}, "id": f"c{len(state['messages'])}"}
    return {"messages": [AIMessage(content="", tool_calls=[chiamata])]}

builder = StateGraph(MessagesState)
builder.add_node("agent", testardo)
builder.add_node("tools", ToolNode([cerca_ordine]))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")
grafo = builder.compile()

try:
    grafo.invoke({"messages": [("user", "Dov'è A-9999?")]}, {"recursion_limit": 6})
except GraphRecursionError as e:
    print(type(e).__name__)
    print(str(e).splitlines()[0])

# Output:
# GraphRecursionError
# Recursion limit of 6 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.

Il messaggio dell'eccezione dice che il limite è stato raggiunto senza una condizione di arresto e come alzarlo. Alzarlo non è mai la prima risposta: se l'agente gira a vuoto, un limite più alto spende più chiamate per lo stesso risultato. Di solito si cattura l'eccezione, si restituisce all'utente una risposta parziale o un'escalation, e si registra la traccia per capire perché la condizione di arresto non è scattata.

Contro gli errori esterni, ogni Runnable (modelli compresi) ha with_retry e with_fallbacks. Il blocco usa un modello che fallisce con ConnectionError un numero prestabilito di volte, per mostrare tre casi: il retry che alla terza chiamata ha successo, il fallback verso un secondo modello, e il retry che si arrende e lascia salire l'errore.

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatGeneration, ChatResult

class Instabile(BaseChatModel):
    """Fallisce con ConnectionError per le prime `guasti` chiamate, poi risponde."""
    guasti: int = 2
    nome: str = "A"
    chiamate: list = []

    @property
    def _llm_type(self): return "instabile"

    def _generate(self, messages, stop=None, run_manager=None, **kw):
        self.chiamate.append(1)
        if len(self.chiamate) <= self.guasti:
            raise ConnectionError(f"{self.nome}: rete assente (tentativo {len(self.chiamate)})")
        return ChatResult(generations=[ChatGeneration(message=AIMessage(content=f"risposta di {self.nome}"))])

# retry: tre tentativi in tutto, il terzo riesce
a = Instabile(guasti=2, nome="A", chiamate=[])
r = a.with_retry(stop_after_attempt=3, wait_exponential_jitter=False,
                 retry_if_exception_type=(ConnectionError,))
print("retry   :", r.invoke("ciao").content, "| chiamate:", len(a.chiamate))

# fallback: il primo modello è sempre guasto, risponde il secondo
p = Instabile(guasti=99, nome="P", chiamate=[])
s = Instabile(guasti=0, nome="S", chiamate=[])
print("fallback:", p.with_fallbacks([s]).invoke("ciao").content)

# retry non basta: dopo i tentativi l'errore risale
c = Instabile(guasti=99, nome="C", chiamate=[])
try:
    c.with_retry(stop_after_attempt=2, wait_exponential_jitter=False,
                 retry_if_exception_type=(ConnectionError,)).invoke("ciao")
except ConnectionError as e:
    print("errore  :", e, "| chiamate:", len(c.chiamate))

# Output:
# retry   : risposta di A | chiamate: 3
# fallback: risposta di S
# errore  : C: rete assente (tentativo 2) | chiamate: 2

I tre casi si leggono nell'uscita: con tre tentativi consentiti e due guasti il modello risponde alla terza chiamata; con il fallback risponde il secondo modello perché il primo è sempre guasto; con due tentativi e guasto permanente l'errore risale dopo due chiamate. Il parametro retry_if_exception_type limita i tentativi agli errori transitori, e va scelto con cura: ritentare un errore di validazione ripete lo stesso errore a pagamento. Su modelli veri aggiungi un tetto di tempo totale, perché retry e fallback sommano le latenze. Il tempo di una chiamata a un modello reale non è stato misurato qui.

LCEL: comporre senza agente

Non ogni problema è un agente. LCEL (LangChain Expression Language) compone Runnable con l'operatore |: prompt, modello, parser. RunnableParallel esegue rami indipendenti sullo stesso input, RunnableBranch sceglie la prima catena la cui condizione è vera e usa l'ultima come predefinita. È il livello giusto per i workflow del primo articolo, dove i passi sono fissi e il modello compila solo i campi. Il blocco usa FakeListChatModel per la sequenza e funzioni semplici al posto dei modelli negli altri due casi.

from langchain_core.language_models.fake_chat_models import FakeListChatModel
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableBranch, RunnableLambda, RunnableParallel

llm = FakeListChatModel(responses=["Un grafo di stato per agenti."])   # modello finto
parser = StrOutputParser()

# Sequenza: prompt | modello | parser
catena = ChatPromptTemplate.from_messages([("human", "{domanda}")]) | llm | parser
print("sequenza :", catena.invoke({"domanda": "Cos'è LangGraph?"}))

# Parallelo: tre rami sullo stesso input (qui funzioni semplici al posto dei modelli)
parallelo = RunnableParallel(
    parole=RunnableLambda(lambda x: len(x["testo"].split())),
    maiuscolo=RunnableLambda(lambda x: x["testo"].upper()),
    lingua=RunnableLambda(lambda x: x["lingua"]),
)
print("parallelo:", parallelo.invoke({"testo": "agenti con grafi", "lingua": "it"}))

# Ramo condizionale: la prima condizione vera sceglie la catena; l'ultima è il default
ramo = RunnableBranch(
    (lambda x: x["lingua"] == "it", RunnableLambda(lambda x: "ramo italiano")),
    (lambda x: x["lingua"] == "en", RunnableLambda(lambda x: "english branch")),
    RunnableLambda(lambda x: "ramo di default"),
)
print("ramo     :", [ramo.invoke({"lingua": l}) for l in ("it", "en", "fr")])

# Output:
# sequenza : Un grafo di stato per agenti.
# parallelo: {'parole': 3, 'maiuscolo': 'AGENTI CON GRAFI', 'lingua': 'it'}
# ramo     : ['ramo italiano', 'english branch', 'ramo di default']

La sequenza restituisce la risposta predefinita del modello finto; il parallelo produce un dizionario con una chiave per ramo; il ramo sceglie in base alla lingua e cade sul default per fr. L'ordine delle condizioni conta: la prima vera vince. Se un compito si risolve così, un agente è un costo in più, e la regola del primo articolo (Introduzione, Cosa Sono e Quando Usarli) resta valida.

Domande frequenti

Che cosa sai fare adesso

Sai riconoscere un tutorial obsoleto dall'import, scegliere tra create_agent e StateGraph, definire strumenti con schema leggibile dal modello, salvare lo stato per thread, fermare un grafo per l'approvazione umana e imporre limiti e fallback. Sai anche provare il cablaggio con un modello finto prima di spendere una chiamata. Il ciclo che il framework automatizza è quello scritto a mano in Fondamenti: OODA Loop, ReAct e Tool Calling; un esempio costruito di assistente di ricerca su un grafo lo trovi in Case Study: Autonomous Research Assistant, e le architetture con più ruoli in Orchestrazione Multi-Agente. Nel prossimo articolo, CrewAI: Team di Agenti Collaborativi, cambiamo paradigma: non un grafo di passi, ma una squadra di ruoli.