Questo articolo è una guida pratica per sviluppatori sulla creazione di agenti IA autonomi in Python. Non ripeteremo la teoria su cosa siano LangChain e LangGraph. Ci concentreremo invece sul codice, sull’architettura e sulla risoluzione di problemi reali.
Obiettivo: Costruire due progetti da zero:
- Un Agente Classificatore: Un agente multi-step con stato gestito, ma senza strumenti esterni.
- Un Agente Assistente: Un agente completo con accesso al file system e alla ricerca web tramite il protocollo MCP, basato su una logica ciclica.
Tratteremo le best practice: gestione della configurazione, selezione dei modelli e gestione degli errori per creare sistemi robusti.
Breve panoramica dei concetti: L’Agente e il ponte MCP
Prima di immergerci nel codice, fissiamo due concetti:
- Agente IA: Un programma costruito attorno a un ciclo “ragionamento-azione”. Riceve un compito, utilizza un LLM per decidere cosa fare dopo (es. chiamare uno strumento), esegue l’azione e ripete il ciclo finché il compito non è completato.
- MCP (Model Context Protocol): Uno standard che funge da ponte tra la logica dell’agente e gli strumenti esterni. Permette all’agente di lavorare con file, API o ricerche in modo unificato, senza preoccuparsi dei dettagli della loro implementazione.
Parte 1: Configurazione di un ambiente robusto
Passo 1: Ambiente virtuale e dipendenze
Crea e attiva un ambiente virtuale. Successivamente, crea un file requirements.txt:
# Framework principali
langchain
langgraph
# Adattatori per modelli
langchain-openai
langchain-google-genai
langchain-mistralai
langchain-community # Per Ollama
# Strumenti e protocolli
langchain-mcp-adapters
mcp
ollama
# Utilità
python-dotenv
tenacity # Per una gestione robusta degli erroriInstalla le dipendenze:
pip install -r requirements.txt```
#### Passo 2: Configurazione delle chiavi API
Crea un file `.env` per memorizzare le tue chiavi:OPENAI_API_KEY=”sk-…”
GOOGLE_API_KEY=”AIzaSy…”
MISTRAL_API_KEY=”…”
BRAVE_API_KEY=”…” # Per lo strumento di ricerca web tramite MCP
#### Passo 3: Il pattern "Model Factory"
Per passare in modo flessibile tra modelli cloud e locali senza modificare il codice dell'agente, utilizzeremo il pattern factory.python
llm_factory.py
import os
from enum import Enum
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_mistralai import ChatMistralAI
from langchain_community.chat_models import ChatOllama
load_dotenv()
class ModelProvider(Enum):
OPENAI = “openai”
GEMINI = “gemini”
MISTRAL_API = “mistral_api”
OLLAMA = “ollama”
def get_llm(provider: ModelProvider, model_name: str = None):
“””Una factory per creare istanze di LLM.”””
if provider == ModelProvider.OPENAI:
return ChatOpenAI(model=model_name or “gpt-4o-mini”, temperature=0)
elif provider == ModelProvider.GEMINI:
return ChatGoogleGenerativeAI(model=model_name or “gemini-1.5-flash”, temperature=0)
elif provider == ModelProvider.MISTRAL_API:
return ChatMistralAI(model=model_name or “mistral-large-latest”, temperature=0)
elif provider == ModelProvider.OLLAMA:
# Assicurati che Ollama sia in esecuzione con il modello richiesto
# docker exec -it ollama ollama pull mistral
return ChatOllama(model=model_name or “mistral”, temperature=0)
raise ValueError(f”Provider del modello sconosciuto: {provider}”)
Esempio di utilizzo
if name == “main“:
# local_llm = get_llm(ModelProvider.OLLAMA)
openai_llm = get_llm(ModelProvider.OPENAI)
response = openai_llm.invoke(“Spiega il concetto di RAG in tre frasi.”)
print(response.content)
### Parte 2: Progetto 1 — Agente di Classificazione Annunci di Lavoro
Questo agente dimostra come utilizzare LangGraph per creare un **grafo lineare** con stato gestito. Prenderà una descrizione di un annuncio di lavoro e la classificherà sequenzialmente in base a tre parametri.
#### Passo 1: Definizione dello Stato
Lo stato è la "memoria" del nostro grafo, passata da un nodo all'altro.python
vacancy_classifier.py
from typing import TypedDict, Dict
class ClassificationState(TypedDict):
“””Stato per l’agente classificatore.”””
description: str # Testo di origine
job_type: str # Tipo di lavoro (progetto/permanente)
category: str # Professione
search_type: str # Scopo (cerca lavoro/prestatore)
classification_log: list # Log di debug
#### Passo 2: Implementazione dei Nodi del Grafo
Ogni nodo è una funzione che prende lo stato, esegue la sua parte di lavoro e restituisce lo stato aggiornato.python
import asyncio
import json
from langchain_core.prompts import ChatPromptTemplate
from llm_factory import get_llm, ModelProvider
class VacancyClassifierAgent:
def init(self):
self.llm = get_llm(ModelProvider.OPENAI, model_name=”gpt-4o-mini”)
async def _classify_job_type(self, state: ClassificationState) -> ClassificationState:
"""Nodo 1: Determina il tipo di lavoro."""
prompt = ChatPromptTemplate.from_messages([
("system", "Determina il tipo di lavoro. La risposta deve essere 'a progetto' o 'permanente'."),
("human", "Descrizione dell'annuncio:\n\n{description}")
])
chain = prompt | self.llm
result = await chain.ainvoke({"description": state["description"]})
state["job_type"] = result.content.strip()
state["classification_log"].append("Tipo di lavoro determinato.")
return state
async def _classify_category(self, state: ClassificationState) -> ClassificationState:
"""Nodo 2: Determina la categoria professionale."""
# Le categorie possono essere caricate da un file o da un database
categories = ["Sviluppatore Python", "Designer", "Marketer", "Animatore 3D"]
prompt = ChatPromptTemplate.from_messages([
("system", f"Scegli la categoria più adatta dalla lista: {', '.join(categories)}."),
("human", "Descrizione dell'annuncio:\n\n{description}")
])
chain = prompt | self.llm
result = await chain.ainvoke({"description": state["description"]})
state["category"] = result.content.strip()
state["classification_log"].append("Categoria determinata.")
return state
async def _classify_search_type(self, state: ClassificationState) -> ClassificationState:
"""Nodo 3: Determina lo scopo della ricerca."""
prompt = ChatPromptTemplate.from_messages([
("system", "Determina lo scopo dell'autore. La risposta deve essere 'cerca lavoro' o 'cerca prestatore'."),
("human", "Descrizione dell'annuncio:\n\n{description}")
])
chain = prompt | self.llm
result = await chain.ainvoke({"description": state["description"]})
state["search_type"] = result.content.strip()
state["classification_log"].append("Scopo della ricerca determinato.")
return state#### Passo 3: Assemblaggio ed Esecuzione del Grafo
Assembliamo i nodi in un unico flusso di lavoro.python
… continuazione della classe VacancyClassifierAgent …
from langgraph.graph import StateGraph, END
def build_graph(self):
"""Assembla il grafo di stati."""
workflow = StateGraph(ClassificationState)
workflow.add_node("job_type_classifier", self._classify_job_type)
workflow.add_node("category_classifier", self._classify_category)
workflow.add_node("search_type_classifier", self._classify_search_type)
workflow.set_entry_point("job_type_classifier")
workflow.add_edge("job_type_classifier", "category_classifier")
workflow.add_edge("category_classifier", "search_type_classifier")
workflow.add_edge("search_type_classifier", END)
return workflow.compile()async def main():
agent = VacancyClassifierAgent()
graph = agent.build_graph()
description = "Cerchiamo uno sviluppatore Python esperto per unirsi al nostro team a tempo pieno per lavorare su un progetto fintech."
initial_state = ClassificationState(
description=description,
job_type="", category="", search_type="",
classification_log=[]
)
final_state = await graph.ainvoke(initial_state)
print("--- Risultato della Classificazione ---")
print(json.dumps(final_state, indent=2, ensure_ascii=False))if name == “main“:
asyncio.run(main())
### Parte 3: Progetto 2 — Agente Assistente con Strumenti (MCP)
Questo agente dimostra una **logica ciclica**, in cui può chiamare ripetutamente degli strumenti per risolvere un compito.
#### Passo 1: Gestione della Configurazione
Per gli agenti che interagiscono con il mondo esterno, una configurazione robusta è essenziale.python
mcp_agent_config.py
from dataclasses import dataclass, field
import os
from llm_factory import ModelProvider
@dataclass
class AgentConfig:
workdir: str = “./agent_workdir”
model_provider: ModelProvider = ModelProvider.OLLAMA
def __post_init__(self):
"""Validazione post-inizializzazione."""
os.makedirs(self.workdir, exist_ok=True)#### Passo 2: Definizione dello Stato per il Dialogo
Lo stato ora memorizzerà la cronologia dei messaggi.python
mcp_agent.py
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
#### Passo 3: Implementazione del Grafo Ciclico
Il grafo consisterà in due nodi principali e un arco condizionale che crea il ciclo "ragionamento-azione".python
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor
from langchain_mcp_adapters.langchain import V1ToolExecutor
from langchain_mcp_adapters.clients import MultiServerMCPClient
from llm_factory import get_llm
from mcp_agent_config import AgentConfig
class MCPAgent:
def init(self, config: AgentConfig):
self.config = config
self.llm = get_llm(config.model_provider)
self.tools = []
self.tool_executor = None
async def setup_tools(self):
"""Inizializza gli strumenti tramite MCP."""
mcp_config = {
"filesystem": {
"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", self.config.workdir],
"transport": "stdio"
},
# Aggiungi brave-search se hai una BRAVE_API_KEY
}
mcp_client = MultiServerMCPClient(mcp_config)
self.tools = await mcp_client.get_tools()
self.tool_executor = ToolExecutor([V1ToolExecutor(tool) for tool in self.tools])
# Collega gli strumenti al modello
self.llm = self.llm.bind_tools(self.tools)
def _should_continue(self, state: AgentState):
"""Arco condizionale: decide se chiamare uno strumento."""
last_message = state['messages'][-1]
if not last_message.tool_calls:
return "end"
return "continue"
def _call_model(self, state: AgentState):
"""Nodo 1: Chiama il LLM per prendere una decisione."""
response = self.llm.invoke(state['messages'])
return {"messages": [response]}
def _call_tool(self, state: AgentState):
"""Nodo 2: Esegue la chiamata allo strumento."""
last_message = state['messages'][-1]
tool_call = last_message.tool_calls[0]
action = {"tool": tool_call["name"], "tool_input": tool_call["args"], "log": ""}
response = self.tool_executor.invoke(action)
return {"messages": [response]}
def build_graph(self):
workflow = StateGraph(AgentState)
workflow.add_node("agent", self._call_model)
workflow.add_node("action", self._call_tool)
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
"agent",
self._should_continue,
{"continue": "action", "end": END}
)
workflow.add_edge("action", "agent")
return workflow.compile()#### Passo 4: Esecuzione e Interazionepython
… continuazione di mcp_agent.py …
import asyncio
from langchain_core.messages import HumanMessage
from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
async def run_agent_task(graph, task):
“””Esegue un compito con gestione degli errori.”””
return await graph.ainvoke({“messages”: [HumanMessage(content=task)]})
async def main():
config = AgentConfig(model_provider=ModelProvider.OPENAI) # o OLLAMA
agent = MCPAgent(config)
await agent.setup_tools()
graph = agent.build_graph()
task = "Crea un file chiamato 'ciao.txt' nella directory di lavoro e scrivici dentro 'Ciao, mondo!'."
result = await run_agent_task(graph, task)
print("\n--- Risposta Finale dell'Agente ---")
print(result['messages'][-1].content)if name == “main“:
asyncio.run(main())`` Qui abbiamo aggiunto il decoratoretenacity` per la robustezza: se la chiamata all’agente fallisce a causa di un errore di rete temporaneo, verrà automaticamente ritentata.
Conclusione
Abbiamo costruito due tipi di agenti utilizzando pratiche moderne:
- Un grafo lineare è eccellente per compiti con una sequenza chiara di passaggi, come processi ETL o analisi multi-step.
- Un grafo ciclico è la base per la creazione di assistenti interattivi e agenti autonomi in grado di risolvere problemi complessi con strumenti.
I pattern architetturali presentati — la factory di modelli, la gestione della configurazione, la separazione della logica in nodi e l’uso di grafi di stati — sono i fondamenti per la costruzione di sistemi IA scalabili e robusti.