4 mar 2026

Pattern Multi-Agente: Orchestratori, Worker e Pipeline

I singoli agenti sono bravi nei compiti ben delimitati. Nel momento in cui un compito richiede conoscenze specialistiche attraverso più domini, lavoro parallelo su input di grandi dimensioni, o decisioni che dovrebbero essere validate prima dell’esecuzione, hai bisogno di più agenti.

I sistemi multi-agente non sono intrinsecamente più complessi — sono solo strutturati diversamente. La chiave è scegliere il pattern giusto per il problema. Tre pattern coprono la maggior parte dei casi d’uso: orchestratore-worker, pipeline e fan-out parallelo.

Perché Più Agenti

L’argomento per i sistemi multi-agente si riduce a tre ragioni pratiche.

Specializzazione. Un singolo agente con 50 strumenti nel contesto si confonde. Un agente specializzato con 5 strumenti focalizzati performa meglio. Dividi per dominio — ricerca, scrittura, codice, verifica — e ogni agente fa una cosa bene.

Parallelismo. Alcuni compiti si scompongono in sotto-compiti indipendenti. Analizzare 20 documenti in sequenza è lento; analizzarli contemporaneamente con agenti paralleli è veloce.

Verifica. Avere un agente che produce output e un secondo agente che lo critica indipendentemente cattura errori che la revisione automatica perde. Il revisore non ha interesse a difendere la risposta originale.

Pattern 1: Orchestratore-Worker

Un agente orchestratore pianifica e delega. Gli agenti worker eseguono compiti specifici e restituiscono risultati. L’orchestratore assembla l’output finale.

Questo è il pattern più flessibile. L’orchestratore può adattare il suo piano in base ai risultati intermedi, riprovare compiti falliti, o escalare a un worker diverso.

import anthropic
import json

client = anthropic.Anthropic()

def run_worker(system_prompt: str, task: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system=system_prompt,
        messages=[{"role": "user", "content": task}]
    )
    return response.content[0].text

def orchestrator(user_request: str) -> str:
    # Step 1: plan the work
    plan_response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system="""You are a planning agent. Given a user request, break it into
2-4 specific subtasks. Return a JSON array of task descriptions only.
Example: ["Research X", "Analyze Y", "Synthesize findings"]""",
        messages=[{"role": "user", "content": user_request}]
    )

    tasks = json.loads(plan_response.content[0].text)

    # Step 2: run each task with a specialized worker
    results = []
    for task in tasks:
        result = run_worker(
            system_prompt="You are a focused execution agent. Complete the assigned task thoroughly.",
            task=task
        )
        results.append({"task": task, "result": result})

    # Step 3: synthesize
    synthesis_prompt = f"""Original request: {user_request}

Worker results:
{json.dumps(results, indent=2)}

Synthesize these results into a cohesive final response."""

    final = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{"role": "user", "content": synthesis_prompt}]
    )
    return final.content[0].text

Il pattern orchestratore funziona meglio quando la struttura del compito non è nota in anticipo. Se non sai quanti sotto-compiti ti serviranno fino a quando non hai analizzato il problema, usa un orchestratore.

Una insidia: gli orchestratori possono allucinare sotto-compiti che non hanno senso. Vincola il formato dell’output (array JSON, lista numerata) e validalo prima di eseguire i worker. Un blocco try/except attorno all’analisi JSON con un passaggio di ripianificazione di fallback gestisce questo con grazia.

Pattern 2: Pipeline

Gli agenti formano una catena sequenziale. Ogni agente trasforma l’input e passa il suo output al successivo. Nessun agente è a conoscenza degli altri — ricevono input e producono output.

Questo è il pattern più semplice da implementare e ragionare. Funziona bene per compiti di trasformazione con fasi ben definite.

def run_pipeline(input_text: str) -> str:
    stages = [
        {
            "name": "Researcher",
            "system": "Extract and organize all key facts from the input. "
                      "Format as a structured list with sources noted where available.",
        },
        {
            "name": "Writer",
            "system": "Transform the research notes into clear, readable prose. "
                      "Maintain all factual content. Target a technical audience.",
        },
        {
            "name": "Editor",
            "system": "Improve clarity and concision. Remove redundancy. "
                      "Do not change facts. Return only the improved text.",
        },
        {
            "name": "Fact Checker",
            "system": "Review for internal consistency. Flag any claims that "
                      "contradict each other or seem unsupported. "
                      "If no issues, return 'VERIFIED: ' followed by the original text.",
        },
    ]

    current = input_text
    for stage in stages:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=stage["system"],
            messages=[{"role": "user", "content": current}]
        )
        current = response.content[0].text
        print(f"[{stage['name']}] complete ({len(current)} chars)")

    return current

Le pipeline accumulano errori. Se il ricercatore perde qualcosa, lo scrittore non può aggiungerlo. Progetta le tue fasi in modo additivo piuttosto che lossy — evita fasi che eliminano informazioni di cui la fase successiva potrebbe aver bisogno.

Un aggiustamento pratico: passa l’input originale insieme all’output di ogni fase quando gli agenti downstream hanno bisogno di contesto che le fasi precedenti potrebbero aver compresso.

Pattern 3: Fan-Out Parallelo

Dividi un grande input in chunk indipendenti, elabora ciascuno in modo concorrente con agenti separati, poi aggrega i risultati.

Questo è il pattern giusto quando stai elaborando più dati di quanti ne entrano comodamente in una singola finestra di contesto, o quando il tempo di elaborazione è importante.

import asyncio

async def analyze_document(doc: str, index: int) -> dict:
    """Analyze a single document asynchronously."""
    system = """Analyze this document and return a JSON object with:
- "sentiment": positive/negative/neutral
- "key_topics": list of 3-5 main topics
- "summary": 2-3 sentence summary
- "flags": list of any concerns (empty list if none)"""

    # asyncio.to_thread lets you call synchronous code in a thread pool
    result = await asyncio.to_thread(
        lambda: client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            system=system,
            messages=[{"role": "user", "content": doc}]
        ).content[0].text
    )
    return {"index": index, **json.loads(result)}

async def parallel_analysis(documents: list[str]) -> dict:
    # Fan out: analyze all documents concurrently
    tasks = [analyze_document(doc, i) for i, doc in enumerate(documents)]
    analyses = await asyncio.gather(*tasks)

    # Aggregate with a dedicated synthesis agent
    synthesis_input = json.dumps({
        "document_count": len(documents),
        "analyses": analyses
    })

    aggregate_result = await asyncio.to_thread(
        lambda: client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system="Synthesize document analyses into: overall sentiment distribution, "
                   "top themes across all documents, and notable patterns in flags. "
                   "Return as JSON.",
            messages=[{"role": "user", "content": synthesis_input}]
        ).content[0].text
    )

    return {
        "individual": analyses,
        "aggregate": json.loads(aggregate_result)
    }

Il passaggio di aggregazione è dove la maggior parte delle implementazioni taglia i corners. Non concatenare risultati — passali a un agente che comprende il compito di aggregazione. Una join di stringhe di 20 analisi non è utile; un riepilogo sintetizzato sì.

Nota la scelta del modello: usa claude-haiku-4-5-20251001 per l’analisi ad alto volume per documento dove la velocità è importante e il compito è semplice, e claude-sonnet-4-6 per la sintesi dove il giudizio è più importante del throughput.

Scegliere il Pattern Giusto

Situazione	Pattern
Struttura del compito sconosciuta fino all’analisi	Orchestratore-worker
Fasi di trasformazione ben definite	Pipeline
Input grande, chunk indipendenti	Fan-out parallelo
Necessità di verifica indipendente	Orchestratore o pipeline con fase di revisione
Minimizzare la latenza su input grandi	Fan-out parallelo

Questi pattern si compongono. Un sistema reale potrebbe usare un orchestratore che fa fan-out di alcuni compiti in parallelo mentre ne esegue altri attraverso una pipeline. Inizia con il pattern più semplice che si adatta e aggiungi complessità solo quando l’approccio più semplice fallisce.

Considerazioni Pratiche

Costo. I sistemi multi-agente moltiplicano le chiamate API. Una pipeline a 4 fasi potrebbe costare 4× una singola chiamata più l’overhead di sintesi. Mescola i modelli strategicamente: usa Opus per l’orchestrazione e la pianificazione dove conta il giudizio, Haiku per i compiti di esecuzione ad alto volume.

Propagazione degli errori. Decidi in anticipo come ogni agente gestisce i fallimenti. Opzioni: propaga l’errore (ferma), restituisci un oggetto errore (lascia che l’orchestratore decida), o riprova con un prompt modificato (si recupera con grazia, aggiunge latenza). Per la maggior parte dei sistemi di produzione, restituire oggetti errore strutturati e lasciare che l’orchestratore decida è il default giusto.

Tracing. Un sistema multi-agente dove non puoi vedere cosa ha fatto ogni agente è un incubo di debugging. Registra ogni chiamata dell’agente con: input, output, modello, latenza e conteggio dei token. Etichetta ogni chiamata con un trace ID in modo da poter ricostruire il percorso di esecuzione completo.

Passaggio del contesto. Sii deliberato su quale contesto riceve ogni agente. Passare la cronologia completa della conversazione a ogni agente è costoso e spesso confuso — gli agenti si distraggono con il contesto precedente irrilevante. Passa solo ciò di cui ogni agente ha bisogno per fare il suo lavoro specifico.

Cosa Costruire Dopo

I pattern qui sono il fondamento. Ciò che costruisci sopra di essi dipende dal tuo problema:

Aggiungi uso degli strumenti ai worker — lascia che agenti specializzati chiamino API, interroghino database o eseguano codice
Aggiungi checkpoint human-in-the-loop dove l’orchestratore si ferma prima di azioni ad alto rischio
Aggiungi memoria persistendo gli output degli agenti in un vector store che gli agenti futuri possono interrogare
Aggiungi valutazione instradando gli output attraverso un agente giudice prima di restituirli all’utente

I sistemi multi-agente sono dove si svolge l’ingegneria interessante nell’AI in questo momento. I pattern sono semplici; il giudizio sta nell’applicarli correttamente al tuo problema specifico.