Questo articolo \u00e8 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.<\/p>\n\n\n\n
Obiettivo:<\/strong> Costruire due progetti da zero:<\/p>\n\n\n\n Tratteremo le best practice: gestione della configurazione, selezione dei modelli e gestione degli errori per creare sistemi robusti.<\/p>\n\n\n\n Prima di immergerci nel codice, fissiamo due concetti:<\/p>\n\n\n\n Crea e attiva un ambiente virtuale. Successivamente, crea un file Installa le dipendenze:<\/p>\n\n\n\n OPENAI_API_KEY=”sk-\u2026” python<\/p>\n\n\n\n import os load_dotenv()<\/p>\n\n\n\n class ModelProvider(Enum): def get_llm(provider: ModelProvider, model_name: str = None): if name<\/strong> == “main<\/strong>“: python<\/p>\n\n\n\n from typing import TypedDict, Dict<\/p>\n\n\n\n class ClassificationState(TypedDict): python class VacancyClassifierAgent: python<\/p>\n\n\n\n from langgraph.graph import StateGraph, END<\/p>\n\n\n\n async def main(): if name<\/strong> == “main<\/strong>“: python<\/p>\n\n\n\n from dataclasses import dataclass, field @dataclass python<\/p>\n\n\n\n from typing import TypedDict, Annotated, Sequence class AgentState(TypedDict): python class MCPAgent: python<\/p>\n\n\n\n import asyncio @retry(stop=stop_after_attempt(3), wait=wait_fixed(1)) async def main(): if name<\/strong> == “main<\/strong>“: Abbiamo costruito due tipi di agenti utilizzando pratiche moderne:<\/p>\n\n\n\n I pattern architetturali presentati \u2014 la factory di modelli, la gestione della configurazione, la separazione della logica in nodi e l’uso di grafi di stati \u2014 sono i fondamenti per la costruzione di sistemi IA scalabili e robusti.<\/p>\n","protected":false},"excerpt":{"rendered":" Questo articolo \u00e8 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: Tratteremo le best practice: gestione della configurazione, selezione dei modelli e… <\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"_acf_changed":false,"neve_meta_sidebar":"","neve_meta_container":"","neve_meta_enable_content_width":"","neve_meta_content_width":0,"neve_meta_title_alignment":"","neve_meta_author_avatar":"","neve_post_elements_order":"","neve_meta_disable_header":"","neve_meta_disable_footer":"","neve_meta_disable_title":"","_themeisle_gutenberg_block_has_review":false,"footnotes":""},"categories":[540],"tags":[104,863,864,169,108],"class_list":["post-2473","post","type-post","status-publish","format-standard","hentry","category-llm","tag-tag-ai-agent","tag-tag-langchain","tag-tag-langgraph","tag-169","tag-tag-mcp"],"acf":[],"_links":{"self":[{"href":"https:\/\/it.davidka.net\/it\/wp-json\/wp\/v2\/posts\/2473","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/it.davidka.net\/it\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/it.davidka.net\/it\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/it.davidka.net\/it\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/it.davidka.net\/it\/wp-json\/wp\/v2\/comments?post=2473"}],"version-history":[{"count":0,"href":"https:\/\/it.davidka.net\/it\/wp-json\/wp\/v2\/posts\/2473\/revisions"}],"wp:attachment":[{"href":"https:\/\/it.davidka.net\/it\/wp-json\/wp\/v2\/media?parent=2473"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/it.davidka.net\/it\/wp-json\/wp\/v2\/categories?post=2473"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/it.davidka.net\/it\/wp-json\/wp\/v2\/tags?post=2473"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}\n
Breve panoramica dei concetti: L’Agente e il ponte MCP<\/h3>\n\n\n\n
\n
Parte 1: Configurazione di un ambiente robusto<\/h3>\n\n\n\n
Passo 1: Ambiente virtuale e dipendenze<\/h4>\n\n\n\n
requirements.txt<\/code>:<\/p>\n\n\n\n# Framework principali\nlangchain\nlanggraph\n\n# Adattatori per modelli\nlangchain-openai\nlangchain-google-genai\nlangchain-mistralai\nlangchain-community # Per Ollama\n\n# Strumenti e protocolli\nlangchain-mcp-adapters\nmcp\nollama\n\n# Utilit\u00e0\npython-dotenv\ntenacity # Per una gestione robusta degli errori<\/code><\/pre>\n\n\n\npip install -r requirements.txt```\n\n#### Passo 2: Configurazione delle chiavi API\n\nCrea un file `.env` per memorizzare le tue chiavi:<\/code><\/pre>\n\n\n\n
GOOGLE_API_KEY=”AIzaSy\u2026”
MISTRAL_API_KEY=”\u2026”
BRAVE_API_KEY=”\u2026” # Per lo strumento di ricerca web tramite MCP<\/p>\n\n\n\n#### Passo 3: Il pattern "Model Factory"\n\nPer passare in modo flessibile tra modelli cloud e locali senza modificare il codice dell'agente, utilizzeremo il pattern factory.<\/code><\/pre>\n\n\n\nllm_factory.py<\/h1>\n\n\n\n
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<\/p>\n\n\n\n
OPENAI = “openai”
GEMINI = “gemini”
MISTRAL_API = “mistral_api”
OLLAMA = “ollama”<\/p>\n\n\n\n
“””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}”)<\/p>\n\n\n\nEsempio di utilizzo<\/h1>\n\n\n\n
# 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)<\/p>\n\n\n\n### Parte 2: Progetto 1 \u2014 Agente di Classificazione Annunci di Lavoro\n\nQuesto agente dimostra come utilizzare LangGraph per creare un **grafo lineare** con stato gestito. Prender\u00e0 una descrizione di un annuncio di lavoro e la classificher\u00e0 sequenzialmente in base a tre parametri.\n\n#### Passo 1: Definizione dello Stato\n\nLo stato \u00e8 la "memoria" del nostro grafo, passata da un nodo all'altro.<\/code><\/pre>\n\n\n\nvacancy_classifier.py<\/h1>\n\n\n\n
“””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<\/p>\n\n\n\n#### Passo 2: Implementazione dei Nodi del Grafo\n\nOgni nodo \u00e8 una funzione che prende lo stato, esegue la sua parte di lavoro e restituisce lo stato aggiornato.<\/code><\/pre>\n\n\n\n
import asyncio
import json
from langchain_core.prompts import ChatPromptTemplate
from llm_factory import get_llm, ModelProvider<\/p>\n\n\n\n
def init<\/strong>(self):
self.llm = get_llm(ModelProvider.OPENAI, model_name=”gpt-4o-mini”)<\/p>\n\n\n\nasync def _classify_job_type(self, state: ClassificationState) -> ClassificationState:\n """Nodo 1: Determina il tipo di lavoro."""\n prompt = ChatPromptTemplate.from_messages([\n ("system", "Determina il tipo di lavoro. La risposta deve essere 'a progetto' o 'permanente'."),\n ("human", "Descrizione dell'annuncio:\\n\\n{description}")\n ])\n chain = prompt | self.llm\n result = await chain.ainvoke({"description": state["description"]})\n\n state["job_type"] = result.content.strip()\n state["classification_log"].append("Tipo di lavoro determinato.")\n return state\n\nasync def _classify_category(self, state: ClassificationState) -> ClassificationState:\n """Nodo 2: Determina la categoria professionale."""\n # Le categorie possono essere caricate da un file o da un database\n categories = ["Sviluppatore Python", "Designer", "Marketer", "Animatore 3D"]\n prompt = ChatPromptTemplate.from_messages([\n ("system", f"Scegli la categoria pi\u00f9 adatta dalla lista: {', '.join(categories)}."),\n ("human", "Descrizione dell'annuncio:\\n\\n{description}")\n ])\n chain = prompt | self.llm\n result = await chain.ainvoke({"description": state["description"]})\n\n state["category"] = result.content.strip()\n state["classification_log"].append("Categoria determinata.")\n return state\n\nasync def _classify_search_type(self, state: ClassificationState) -> ClassificationState:\n """Nodo 3: Determina lo scopo della ricerca."""\n prompt = ChatPromptTemplate.from_messages([\n ("system", "Determina lo scopo dell'autore. La risposta deve essere 'cerca lavoro' o 'cerca prestatore'."),\n ("human", "Descrizione dell'annuncio:\\n\\n{description}")\n ])\n chain = prompt | self.llm\n result = await chain.ainvoke({"description": state["description"]})\n\n state["search_type"] = result.content.strip()\n state["classification_log"].append("Scopo della ricerca determinato.")\n return state<\/code><\/pre>\n\n\n\n#### Passo 3: Assemblaggio ed Esecuzione del Grafo\n\nAssembliamo i nodi in un unico flusso di lavoro.<\/code><\/pre>\n\n\n\n\u2026 continuazione della classe VacancyClassifierAgent \u2026<\/h1>\n\n\n\n
def build_graph(self):\n """Assembla il grafo di stati."""\n workflow = StateGraph(ClassificationState)\n\n workflow.add_node("job_type_classifier", self._classify_job_type)\n workflow.add_node("category_classifier", self._classify_category)\n workflow.add_node("search_type_classifier", self._classify_search_type)\n\n workflow.set_entry_point("job_type_classifier")\n workflow.add_edge("job_type_classifier", "category_classifier")\n workflow.add_edge("category_classifier", "search_type_classifier")\n workflow.add_edge("search_type_classifier", END)\n\n return workflow.compile()<\/code><\/pre>\n\n\n\n
agent = VacancyClassifierAgent()
graph = agent.build_graph()<\/p>\n\n\n\ndescription = "Cerchiamo uno sviluppatore Python esperto per unirsi al nostro team a tempo pieno per lavorare su un progetto fintech."\n\ninitial_state = ClassificationState(\n description=description,\n job_type="", category="", search_type="",\n classification_log=[]\n)\n\nfinal_state = await graph.ainvoke(initial_state)\n\nprint("--- Risultato della Classificazione ---")\nprint(json.dumps(final_state, indent=2, ensure_ascii=False))<\/code><\/pre>\n\n\n\n
asyncio.run(main())<\/p>\n\n\n\n### Parte 3: Progetto 2 \u2014 Agente Assistente con Strumenti (MCP)\n\nQuesto agente dimostra una **logica ciclica**, in cui pu\u00f2 chiamare ripetutamente degli strumenti per risolvere un compito.\n\n#### Passo 1: Gestione della Configurazione\n\nPer gli agenti che interagiscono con il mondo esterno, una configurazione robusta \u00e8 essenziale.<\/code><\/pre>\n\n\n\nmcp_agent_config.py<\/h1>\n\n\n\n
import os
from llm_factory import ModelProvider<\/p>\n\n\n\n
class AgentConfig:
workdir: str = “.\/agent_workdir”
model_provider: ModelProvider = ModelProvider.OLLAMA<\/p>\n\n\n\ndef __post_init__(self):\n """Validazione post-inizializzazione."""\n os.makedirs(self.workdir, exist_ok=True)<\/code><\/pre>\n\n\n\n#### Passo 2: Definizione dello Stato per il Dialogo\n\nLo stato ora memorizzer\u00e0 la cronologia dei messaggi.<\/code><\/pre>\n\n\n\nmcp_agent.py<\/h1>\n\n\n\n
from langchain_core.messages import BaseMessage
import operator<\/p>\n\n\n\n
messages: Annotated[Sequence[BaseMessage], operator.add]<\/p>\n\n\n\n#### Passo 3: Implementazione del Grafo Ciclico\n\nIl grafo consister\u00e0 in due nodi principali e un arco condizionale che crea il ciclo "ragionamento-azione".<\/code><\/pre>\n\n\n\n
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<\/p>\n\n\n\n
def init<\/strong>(self, config: AgentConfig):
self.config = config
self.llm = get_llm(config.model_provider)
self.tools = []
self.tool_executor = None<\/p>\n\n\n\nasync def setup_tools(self):\n """Inizializza gli strumenti tramite MCP."""\n mcp_config = {\n "filesystem": {\n "command": "npx", "args": ["-y", "@modelcontextprotocol\/server-filesystem", self.config.workdir],\n "transport": "stdio"\n },\n # Aggiungi brave-search se hai una BRAVE_API_KEY\n }\n mcp_client = MultiServerMCPClient(mcp_config)\n self.tools = await mcp_client.get_tools()\n self.tool_executor = ToolExecutor([V1ToolExecutor(tool) for tool in self.tools])\n\n # Collega gli strumenti al modello\n self.llm = self.llm.bind_tools(self.tools)\n\ndef _should_continue(self, state: AgentState):\n """Arco condizionale: decide se chiamare uno strumento."""\n last_message = state['messages'][-1]\n if not last_message.tool_calls:\n return "end"\n return "continue"\n\ndef _call_model(self, state: AgentState):\n """Nodo 1: Chiama il LLM per prendere una decisione."""\n response = self.llm.invoke(state['messages'])\n return {"messages": [response]}\n\ndef _call_tool(self, state: AgentState):\n """Nodo 2: Esegue la chiamata allo strumento."""\n last_message = state['messages'][-1]\n tool_call = last_message.tool_calls[0]\n\n action = {"tool": tool_call["name"], "tool_input": tool_call["args"], "log": ""}\n response = self.tool_executor.invoke(action)\n\n return {"messages": [response]}\n\ndef build_graph(self):\n workflow = StateGraph(AgentState)\n workflow.add_node("agent", self._call_model)\n workflow.add_node("action", self._call_tool)\n\n workflow.set_entry_point("agent")\n workflow.add_conditional_edges(\n "agent",\n self._should_continue,\n {"continue": "action", "end": END}\n )\n workflow.add_edge("action", "agent")\n\n return workflow.compile()<\/code><\/pre>\n\n\n\n#### Passo 4: Esecuzione e Interazione<\/code><\/pre>\n\n\n\n\u2026 continuazione di mcp_agent.py \u2026<\/h1>\n\n\n\n
from langchain_core.messages import HumanMessage
from tenacity import retry, stop_after_attempt, wait_fixed<\/p>\n\n\n\n
async def run_agent_task(graph, task):
“””Esegue un compito con gestione degli errori.”””
return await graph.ainvoke({“messages”: [HumanMessage(content=task)]})<\/p>\n\n\n\n
config = AgentConfig(model_provider=ModelProvider.OPENAI) # o OLLAMA
agent = MCPAgent(config)
await agent.setup_tools()
graph = agent.build_graph()<\/p>\n\n\n\ntask = "Crea un file chiamato 'ciao.txt' nella directory di lavoro e scrivici dentro 'Ciao, mondo!'."\nresult = await run_agent_task(graph, task)\n\nprint("\\n--- Risposta Finale dell'Agente ---")\nprint(result['messages'][-1].content)<\/code><\/pre>\n\n\n\n
asyncio.run(main())`` Qui abbiamo aggiunto il decoratore<\/code>tenacity` per la robustezza: se la chiamata all’agente fallisce a causa di un errore di rete temporaneo, verr\u00e0 automaticamente ritentata.<\/p>\n\n\n\nConclusione<\/h3>\n\n\n\n
\n