Sistemi Multi-Agente: Orchestrare Più AI che Collaborano

Cos’è un sistema multi-agente?

Quando lavori con un singolo modello AI, stai essenzialmente affidando tutto il ragionamento a un unico cervello. Funziona per task semplici, ma appena la complessità aumenta — pipeline lunghe, decisioni ramificate, operazioni parallele — emerge il limite fondamentale: un solo agente deve fare tutto.

I sistemi multi-agente risolvono questo problema dividendo il lavoro tra più istanze AI specializzate che collaborano. Ogni agente ha un ruolo preciso, riceve istruzioni chiare, usa strumenti specifici e passa i risultati ad altri agenti. Il risultato finale emerge dalla coordinazione, non dal genio di un singolo modello.

In pratica: immagina di dover scrivere un report di mercato completo. Un agente ricerca competitor, un altro analizza i dati finanziari, un terzo sintetizza le fonti e un quarto redige il documento finale. Ognuno è ottimizzato per il suo compito, lavora in parallelo dove possibile, e l’orchestratore coordina il flusso.

Perché i multi-agente sono il futuro dell’automazione AI

Il passaggio da single-agent a multi-agent non è solo una questione di performance. Cambia radicalmente cosa puoi costruire. Ecco i vantaggi chiave:

  • Parallelizzazione reale: task indipendenti vengono eseguiti simultaneamente, riducendo drammaticamente i tempi.
  • Specializzazione: ogni agente può essere finemente istruito per il suo dominio, con prompt specifici e strumenti dedicati.
  • Scalabilità: aggiungere capacità significa aggiungere agenti, non riscrivere tutto.
  • Isolamento degli errori: se un agente fallisce, non compromette l’intera pipeline.
  • Ragionamento distribuito: problemi complessi vengono decomposti in sotto-problemi gestibili.

Framework come LangGraph, CrewAI, AutoGen e le API native di Claude stanno rendendo questi sistemi accessibili a qualsiasi sviluppatore. Non serve essere un ricercatore AI per costruire pipeline multi-agente funzionanti.

Architetture fondamentali

Prima di scrivere codice, è utile capire i pattern architetturali principali. Ogni progetto richiede un’architettura diversa a seconda della complessità e del tipo di task.

1. Orchestratore + Sub-agenti

È il pattern più comune. Un agente orchestratore riceve l’obiettivo principale, lo decompone in sotto-task e li delega a sub-agenti specializzati. L’orchestratore raccoglie i risultati e li integra. Ideale per workflow con step sequenziali ma logicamente separabili.

2. Pipeline sequenziale

Gli agenti sono disposti in catena: l’output di uno diventa l’input del successivo. Semplice da implementare e debuggare. Ottimo quando ogni step trasforma l’output del precedente (es. ricerca → analisi → scrittura → revisione).

3. Rete peer-to-peer

Gli agenti possono comunicare direttamente tra loro senza un orchestratore centrale. Più flessibile ma anche più difficile da controllare. Adatto a sistemi dove gli agenti devono negoziare o collaborare in modo non strutturato.

4. Specializzazione con router

Un agente router classifica la richiesta in ingresso e la indirizza all’agente più adatto. Utile quando hai una gamma ampia di task eterogenei e vuoi evitare che ogni agente gestisca tutto.

Tool use: il meccanismo che abilita l’autonomia

Un agente AI diventa davvero potente quando può usare strumenti: chiamare API, leggere file, eseguire codice, fare ricerche web. Con Claude di Anthropic, il meccanismo di tool use è integrato nativamente nell’API.

Ecco un esempio base di agente con tool use usando l’API di Claude:

import anthropic
import json

client = anthropic.Anthropic()

# Definizione degli strumenti disponibili
tools = [
    {
        "name": "search_web",
        "description": "Cerca informazioni sul web e restituisce i risultati",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "La query di ricerca"
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "analyze_data",
        "description": "Analizza un dataset e restituisce statistiche",
        "input_schema": {
            "type": "object",
            "properties": {
                "data": {
                    "type": "array",
                    "description": "Array di valori numerici da analizzare"
                }
            },
            "required": ["data"]
        }
    }
]

def run_agent(task: str) -> str:
    """Esegue un agente con tool use in loop autonomo."""
    messages = [{"role": "user", "content": task}]

    while True:
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )

        # Se il modello ha finito, restituisce la risposta
        if response.stop_reason == "end_turn":
            return response.content[0].text

        # Se il modello vuole usare un tool
        if response.stop_reason == "tool_use":
            # Aggiunge la risposta dell'assistente alla conversazione
            messages.append({
                "role": "assistant",
                "content": response.content
            })

            # Esegue tutti i tool richiesti
            tool_results = []
            for content_block in response.content:
                if content_block.type == "tool_use":
                    result = execute_tool(content_block.name, content_block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": content_block.id,
                        "content": result
                    })

            # Aggiunge i risultati alla conversazione
            messages.append({
                "role": "user",
                "content": tool_results
            })

def execute_tool(name: str, inputs: dict) -> str:
    """Esegue il tool richiesto e restituisce il risultato."""
    if name == "search_web":
        # Implementazione reale: integrazione con Serper, Tavily, etc.
        return f"Risultati di ricerca per: {inputs['query']}"
    elif name == "analyze_data":
        data = inputs["data"]
        return json.dumps({
            "mean": sum(data) / len(data),
            "min": min(data),
            "max": max(data),
            "count": len(data)
        })
    return "Tool non trovato"

# Esecuzione
result = run_agent("Cerca informazioni sui sistemi multi-agente e analizza questi dati: [10, 25, 38, 42, 55]")
print(result)

Questo pattern — chiamata API → verifica stop_reason → esecuzione tool → nuovo messaggio — è il cuore di ogni agente autonomo. Il loop continua finché il modello non decide di aver completato il task.

Implementare un sistema multi-agente con Python

Vediamo ora come costruire un sistema con più agenti che collaborano. In questo esempio creiamo un pipeline di ricerca e analisi con tre agenti: un researcher, un analyst e un writer.

import anthropic
from dataclasses import dataclass
from typing import Optional

client = anthropic.Anthropic()

@dataclass
class AgentResult:
    agent_name: str
    output: str
    success: bool

class MultiAgentPipeline:
    """
    Pipeline sequenziale: Researcher → Analyst → Writer
    """

    def __init__(self, model: str = "claude-opus-4-5"):
        self.model = model

    def run_agent(
        self,
        name: str,
        system_prompt: str,
        task: str,
        context: Optional[str] = None
    ) -> AgentResult:
        """Esegue un singolo agente e restituisce il risultato."""
        messages = []

        # Aggiunge il contesto degli agenti precedenti se presente
        if context:
            messages.append({
                "role": "user",
                "content": f"Contesto dalla fase precedente:

{context}

Il tuo task: {task}"
            })
        else:
            messages.append({"role": "user", "content": task})

        try:
            response = client.messages.create(
                model=self.model,
                max_tokens=2048,
                system=system_prompt,
                messages=messages
            )
            return AgentResult(
                agent_name=name,
                output=response.content[0].text,
                success=True
            )
        except Exception as e:
            return AgentResult(
                agent_name=name,
                output=f"Errore: {str(e)}",
                success=False
            )

    def run(self, topic: str) -> dict:
        """Esegue la pipeline completa e restituisce tutti i risultati."""
        results = {}

        # Agente 1: Researcher
        print(f"[Researcher] Analizzando: {topic}")
        researcher = self.run_agent(
            name="Researcher",
            system_prompt="""Sei un ricercatore esperto. Il tuo compito è identificare:
            1. I punti chiave dell'argomento
            2. I concetti tecnici fondamentali
            3. Le applicazioni pratiche principali
            Sii preciso e strutturato. Usa elenchi puntati.""",
            task=f"Ricerca e struttura le informazioni chiave su: {topic}"
        )
        results["researcher"] = researcher

        if not researcher.success:
            return results

        # Agente 2: Analyst
        print(f"[Analyst] Elaborando i dati...")
        analyst = self.run_agent(
            name="Analyst",
            system_prompt="""Sei un analista tecnico. Devi:
            1. Valutare la complessità implementativa
            2. Identificare vantaggi e limitazioni
            3. Suggerire casi d'uso ottimali
            Fornisci un'analisi critica basata sui dati ricevuti.""",
            task="Analizza criticamente le informazioni fornite e valuta vantaggi, limitazioni e casi d'uso",
            context=researcher.output
        )
        results["analyst"] = analyst

        if not analyst.success:
            return results

        # Agente 3: Writer
        print(f"[Writer] Componendo il report finale...")
        combined_context = f"Ricerca:
{researcher.output}

Analisi:
{analyst.output}"
        writer = self.run_agent(
            name="Writer",
            system_prompt="""Sei un technical writer specializzato in AI. Scrivi in italiano.
            Il tuo output deve essere:
            - Chiaro e accessibile anche a lettori non esperti
            - Strutturato con titoli e sezioni logiche
            - Pratico, con focus su applicazioni concrete
            - Lungo almeno 400 parole""",
            task="Scrivi un articolo tecnico completo basato sulla ricerca e sull'analisi fornite",
            context=combined_context
        )
        results["writer"] = writer

        return results

# Utilizzo
pipeline = MultiAgentPipeline()
results = pipeline.run("sistemi multi-agente con Claude AI")

for agent_name, result in results.items():
    print(f"
{'='*50}")
    print(f"OUTPUT - {result.agent_name}")
    print('='*50)
    print(result.output[:500] + "..." if len(result.output) > 500 else result.output)

Orchestrazione avanzata con LangGraph

I tool di automazione come n8n e Make gestiscono workflow visualmente, ma per pipeline AI più complesse serve un framework come LangGraph. Permette di definire grafi di esecuzione dove i nodi sono agenti e gli archi sono transizioni condizionali.

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# Stato condiviso tra tutti gli agenti nel grafo
class AgentState(TypedDict):
    topic: str
    research: str
    analysis: str
    final_report: str
    iteration: Annotated[int, operator.add]
    should_retry: bool

def researcher_node(state: AgentState) -> AgentState:
    """Nodo ricercatore nel grafo."""
    # Simulazione della chiamata Claude
    research_result = f"Ricerca su '{state['topic']}': [risultati dettagliati]"
    return {
        **state,
        "research": research_result,
        "iteration": 1
    }

def analyst_node(state: AgentState) -> AgentState:
    """Nodo analista nel grafo."""
    analysis_result = f"Analisi della ricerca: [punti chiave, pro/contro]"
    quality_score = len(state["research"]) > 100  # Logica di qualità semplificata
    return {
        **state,
        "analysis": analysis_result,
        "should_retry": not quality_score,
        "iteration": 1
    }

def writer_node(state: AgentState) -> AgentState:
    """Nodo scrittore nel grafo."""
    final_report = f"Report finale basato su ricerca e analisi: [contenuto completo]"
    return {
        **state,
        "final_report": final_report,
        "iteration": 1
    }

def quality_check(state: AgentState) -> str:
    """Funzione di routing condizionale."""
    if state.get("should_retry") and state.get("iteration", 0) < 3:
        return "retry_research"
    return "proceed_to_write"

# Costruzione del grafo
workflow = StateGraph(AgentState)

# Aggiunta dei nodi
workflow.add_node("researcher", researcher_node)
workflow.add_node("analyst", analyst_node)
workflow.add_node("writer", writer_node)

# Definizione del flusso
workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "analyst")

# Routing condizionale: retry o procedi
workflow.add_conditional_edges(
    "analyst",
    quality_check,
    {
        "retry_research": "researcher",   # Torna al ricercatore
        "proceed_to_write": "writer"      # Procedi alla scrittura
    }
)
workflow.add_edge("writer", END)

# Compilazione ed esecuzione
app = workflow.compile()
result = app.invoke({
    "topic": "architetture multi-agente",
    "research": "",
    "analysis": "",
    "final_report": "",
    "iteration": 0,
    "should_retry": False
})

print(result["final_report"])

Il vantaggio di LangGraph è il routing condizionale: un agente di controllo qualità può decidere se ripetere una fase o procedere. Questo permette sistemi auto-correttivi che migliorano il proprio output prima di finalizzarlo.

Model Context Protocol (MCP): il layer di comunicazione universale

Un problema concreto nei sistemi multi-agente è come gli agenti comunicano con strumenti e servizi esterni. Ogni integrazione richiedeva codice custom. Anthropic ha risolto questo con MCP — Model Context Protocol, uno standard aperto per connettere gli agenti AI a qualsiasi fonte di dati o strumento.

Con MCP puoi connettere Claude a database, GitHub, Slack, filesystem locali e centinaia di altri servizi senza scrivere integrazioni da zero. Claude Code stesso usa MCP per accedere ai file del tuo progetto.

MCP è al multi-agente quello che HTTP è al web: uno standard che permette a qualsiasi agente di parlare con qualsiasi servizio usando lo stesso protocollo.

Ecco un esempio di configurazione MCP per connettere un agente Claude a un server di strumenti custom:

{
  "mcpServers": {
    "database": {
      "command": "python",
      "args": ["-m", "mcp_server_postgres"],
      "env": {
        "DATABASE_URL": "postgresql://user:pass@localhost/mydb"
      }
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxx"
      }
    },
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/Users/utente/progetti"
      ]
    }
  }
}

Con questa configurazione, un agente Claude può interrogare il database, leggere repository GitHub e accedere ai file locali — tutto attraverso lo stesso protocollo standardizzato.

Pattern avanzati: human-in-the-loop e supervisione

I sistemi multi-agente completamente autonomi vanno bene per task low-risk, ma per operazioni critiche — acquisti, pubblicazioni, modifiche irreversibili — è essenziale inserire punti di controllo umano. Questo pattern si chiama human-in-the-loop.

Come con gli strumenti di coding AI, il punto non è sostituire la supervisione umana ma renderla più efficiente: l’agente fa il lavoro pesante e l’umano approva solo i passaggi critici.

import anthropic
from enum import Enum

client = anthropic.Anthropic()

class ActionRisk(Enum):
    LOW = "low"       # Esegui automaticamente
    MEDIUM = "medium" # Notifica e procedi
    HIGH = "high"     # Richiedi approvazione esplicita

def assess_action_risk(action: str, parameters: dict) -> ActionRisk:
    """Valuta il livello di rischio di un'azione."""
    high_risk_actions = ["delete", "purchase", "publish", "send_email", "deploy"]
    medium_risk_actions = ["update", "modify", "write_file"]

    if any(risky in action.lower() for risky in high_risk_actions):
        return ActionRisk.HIGH
    elif any(medium in action.lower() for medium in medium_risk_actions):
        return ActionRisk.MEDIUM
    return ActionRisk.LOW

def supervised_agent_loop(task: str, auto_approve_low_risk: bool = True):
    """
    Loop agente con supervisione umana per azioni ad alto rischio.
    """
    tools = [
        {
            "name": "read_file",
            "description": "Legge il contenuto di un file",
            "input_schema": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Percorso del file"}
                },
                "required": ["path"]
            }
        },
        {
            "name": "publish_content",
            "description": "Pubblica contenuto su un canale",
            "input_schema": {
                "type": "object",
                "properties": {
                    "channel": {"type": "string"},
                    "content": {"type": "string"}
                },
                "required": ["channel", "content"]
            }
        }
    ]

    messages = [{"role": "user", "content": task}]
    actions_log = []

    while True:
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=2048,
            tools=tools,
            messages=messages
        )

        if response.stop_reason == "end_turn":
            print("
✅ Task completato con successo")
            break

        if response.stop_reason == "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            tool_results = []

            for block in response.content:
                if block.type == "tool_use":
                    risk = assess_action_risk(block.name, block.input)
                    approved = False

                    if risk == ActionRisk.LOW and auto_approve_low_risk:
                        approved = True
                        print(f"[AUTO] {block.name}: {block.input}")
                    elif risk == ActionRisk.MEDIUM:
                        print(f"[INFO] L'agente vuole eseguire: {block.name}")
                        print(f"       Parametri: {block.input}")
                        approved = True  # Auto-approva medium in questo esempio
                    else:  # HIGH risk
                        print(f"
⚠️  AZIONE AD ALTO RISCHIO RILEVATA")
                        print(f"   Tool: {block.name}")
                        print(f"   Parametri: {block.input}")
                        approval = input("   Approvare? (s/n): ").strip().lower()
                        approved = approval == 's'

                    if approved:
                        result = f"Azione {block.name} eseguita con successo"
                        actions_log.append({"action": block.name, "approved": True})
                    else:
                        result = f"Azione {block.name} rifiutata dall'utente"
                        actions_log.append({"action": block.name, "approved": False})

                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })

            messages.append({"role": "user", "content": tool_results})

    return actions_log

# Utilizzo
log = supervised_agent_loop(
    "Leggi il file config.json e poi pubblica un annuncio su Twitter",
    auto_approve_low_risk=True
)
print(f"
Azioni eseguite: {len(log)}")

Gestione degli errori e resilienza

Un sistema multi-agente in produzione deve gestire fallimenti in modo graceful. Gli errori sono inevitabili: timeout API, risposte malformate, tool che falliscono. La resilienza non è opzionale.

Le strategie principali sono:

  • Retry con backoff esponenziale: riprova le chiamate fallite aspettando intervalli crescenti (1s, 2s, 4s, 8s).
  • Circuit breaker: se un agente fallisce ripetutamente, bypassa quel nodo e usa un fallback.
  • Checkpoint dello stato: salva lo stato della pipeline periodicamente per poter riprendere da un punto intermedio in caso di crash.
  • Timeout per ogni agente: imposta un limite di tempo massimo per ogni nodo, evitando che un agente bloccato congeli l’intera pipeline.
  • Monitoring e alerting: traccia metriche di ogni agente (latenza, tasso di successo, token utilizzati) per rilevare anomalie.

Costi e ottimizzazione

Il vantaggio dei sistemi multi-agente ha un costo: più chiamate API, più token consumati. Una pipeline con 5 agenti può costare 5× rispetto a un singolo agente, e se non è ottimizzata anche di più (i contesti si accumulano).

Strategie di ottimizzazione dei costi:

  • Modelli misti: usa Claude Opus solo per il ragionamento complesso (orchestratore), Claude Haiku per task semplici (sub-agenti di recupero dati).
  • Prompt caching: con il prompt caching di Claude, i system prompt comuni vengono cachati e non riconteggiati ad ogni chiamata.
  • Contesti compressi: invece di passare l’intero output di un agente al successivo, usa un agente sintetizzatore per comprimere le informazioni.
  • Parallelizzazione efficiente: esegui agenti indipendenti in parallelo con asyncio per ridurre la latenza totale (non i costi, ma il tempo).

Casi d’uso reali per developer e freelance

I sistemi multi-agente non sono solo per grandi aziende. Come freelance developer, puoi costruire pipeline multi-agente per automatizzare lavori ripetitivi e offrirle come servizio.

Ecco i casi d’uso più concreti e remunerativi:

  • Pipeline di content marketing: agente SEO researcher + agente writer + agente editor → articoli ottimizzati in automatico.
  • Code review automatizzato: agente che legge PR GitHub, agente che verifica security, agente che controlla performance → report completo.
  • Lead generation B2B: agente che ricerca aziende target, agente che trova contatti, agente che personalizza l’outreach.
  • Customer support AI: agente router che classifica il ticket, agente tecnico per problemi complessi, agente CRM per aggiornare lo stato.
  • Report finanziari automatici: agente che raccoglie dati da API, agente analista, agente che genera PDF professionale.

Strumenti e framework consigliati nel 2026

L’ecosistema si è stabilizzato. Questi sono i framework più maturi e usati in produzione oggi:

  • LangGraph (Python): il più flessibile per grafi di agenti complessi con stato condiviso e routing condizionale. Ottimo per sistemi enterprise.
  • CrewAI (Python): API di alto livello orientata ai “ruoli”. Perfetto per chi vuole costruire velocemente senza gestire i dettagli dell’orchestrazione.
  • AutoGen (Python, Microsoft): forte per sistemi multi-agente conversazionali e simulazioni dove gli agenti “dialogano” tra loro.
  • Anthropic Agent SDK: toolkit ufficiale Anthropic per costruire agenti con Claude, con supporto nativo a MCP e tool use.
  • n8n / Make con AI: per chi preferisce approcci no-code/low-code, entrambi i tool hanno ora nodi AI nativi che permettono pipeline multi-agente visuale.

Conclusione: costruisci il tuo primo sistema multi-agente

I sistemi multi-agente rappresentano il salto qualitativo dall’automazione AI “toy” all’automazione AI “production-grade”. Non è più fantascienza: con Python, l’API di Claude e un framework come LangGraph o CrewAI, puoi costruire pipeline che lavorano per te 24/7 su task che prima richiedevano ore di lavoro manuale.

Il consiglio pratico: inizia semplice. Una pipeline a tre agenti — researcher, analyst, writer — è già straordinariamente potente e ti insegna tutti i pattern fondamentali. Da lì, aggiungi complessità gradualmente: routing condizionale, retry automatici, supervisione umana per le azioni critiche.

L’infrastruttura è pronta. I modelli sono abbastanza capaci. L’unico limite è la tua immaginazione su quali workflow automatizzare.

Condividi

Articoli Recenti

Categorie popolari