89 - Agent-as-tool: orchestrazione gerarchica invece della delega

Trasformare un agente in un tool significa che chi lo chiama non cede il controllo: riceve una risposta, la legge e continua a ragionare. È l'opposto dell'handoff dell'articolo precedente, dove il controllo passava e non tornava. Con l'agent-as-tool nasce una gerarchia, un orchestratore e dei sotto-agenti, che permette di confrontare pareri, eseguire lavoro in parallelo e sintetizzare. Il prezzo è un conto in input che cresce in fretta e un modo nuovo di sbagliare: la gerarchia che chiama sé stessa.

Versione provata. openai-agents 0.24.0 su Python 3.14.4, il 2026-11-12, con il modello finto descritto nell'articolo precedente (un agents.Model che risponde con un copione e registra ciò che riceve; il file fake.py è riportato lì). I listati provano la meccanica dell'SDK, non la qualità di un modello vero: le decisioni dell'orchestratore le prende il copione. Il metodo Agent.as_tool(...) è nella documentazione dei tool; fissa la versione e rileggi il changelog prima di aggiornare.

L'analogia: il consulente e il perito

Nell'handoff eri il cliente passato dal centralino all'ufficio rimborsi: a quel punto parlavi con loro. Qui invece sei seduto davanti a un consulente che, per rispondere alla tua domanda, esce dalla stanza, chiede un parere a un perito, torna con la risposta in mano e continua a parlare con te. Tu non hai mai incontrato il perito. Il consulente può anche interpellarne tre, confrontare i pareri e dirti che due su tre sconsigliano l'operazione.

La differenza funzionale si riduce a una domanda sola: ti serve la risposta dello specialista, o ti serve che lo specialista parli all'utente? Se ti serve la risposta, per elaborarla, confrontarla, inserirla in un documento, vuoi un agente come tool. Se ti serve che prosegua lui la conversazione, vuoi un handoff.

Le due forme, in un disegno

Quattro topologie, e quando si usa ciascuna

Handoff e agent-as-tool sono i due mattoni; con due mattoni si costruiscono quattro forme ricorrenti. Conviene conoscerle come forme, perché la domanda «quale architettura multi-agente uso?» ha spesso una di queste quattro come risposta, e sceglierla in anticipo evita di scoprire a metà che il grafo disegnato non può fare ciò che serve. Le prime tre corrispondono a schemi descritti da Anthropic in Building effective agents: il routing, l'orchestrator-workers e il prompt chaining.

La stella (routing). Un agente di triage, N specialisti, handoff solo dal centro verso la periferia. È la forma dell'articolo precedente. Serve quando una richiesta appartiene a uno dominio e lo specialista può concluderla da solo. Proprietà che la rendono preferibile: nessun ciclo possibile per costruzione, un solo punto in cui si decide, e specialisti indipendenti che si possono sviluppare e testare separatamente. Limite: lo specialista parla direttamente all'utente, quindi non c'è un punto unico dove imporre tono e controlli di uscita.

La gerarchia (orchestrazione). Un orchestratore, N sotto-agenti esposti come tool. È la forma di questo articolo. Serve quando servono più contributi da combinare, o quando il risultato di uno specialista deve essere elaborato prima di arrivare all'utente; Anthropic la raccomanda quando i sotto-compiti non sono prevedibili in anticipo. Proprietà: un unico punto di responsabilità sulla risposta finale, che è il posto dove mettere il guardrail di uscita, il formato, la voce. Limite: il moltiplicatore nell'input e la latenza, se non parallelizzi.

La catena (pipeline di agenti). Agente A produce, agente B trasforma, agente C verifica. Il grafo è fisso e sta nel tuo codice: non è un pattern agentico, è una pipeline i cui stadi sono agenti. Serve quando la sequenza è nota e ogni stadio ha un mandato diverso (estrai, normalizza, valida). Proprietà: prevedibile, testabile stadio per stadio, nessuna decisione di routing da sbagliare. Se stai pensando a un'orchestrazione e scopri che l'ordine delle chiamate è sempre lo stesso, quello che vuoi è una catena.

La maglia (agenti che si parlano fra loro). Ogni agente può passare a ogni altro. Serve raramente. Ha un'attrattiva intellettuale forte perché somiglia a un gruppo di lavoro, e ha due costi che si pagano sempre: i cicli diventano possibili e il comportamento complessivo diventa difficile da prevedere e da testare. Vale la pena adottarla solo dopo aver verificato che le altre tre non bastano; la maglia che «serviva» è spesso una stella con un router scritto meglio.

Un criterio di scelta da applicare in quest'ordine: la sequenza è nota? Catena. Una richiesta appartiene a un solo dominio? Stella. Serve combinare contributi? Gerarchia. Nessuna delle tre? Riconsidera il problema prima di passare alla maglia.

Per il lato industriale di queste scelte (pattern di coordinamento, scalabilità, osservabilità di un sistema a più agenti in esercizio) la serie dedicata approfondisce in orchestrazione multi-agente: architetture per la produzione, e i framework che implementano queste topologie come primitive sono trattati in CrewAI: team di agenti collaborativi e AutoGen: conversazione multi-agente. Qui resto sul meccanismo e sui conti.

Perché l'orchestrazione affidata al modello fallisce in modi silenziosi

Scrivo più volte nell'articolo «se il piano è noto, scrivilo in Python», e vale la pena spiegare perché: i fallimenti dell'orchestrazione affidata al modello non assomigliano a errori. Producono risultati plausibili e incompleti, che è la categoria peggiore. Sono tre modi concreti, descritti qui come scenari da prevedere nel progetto, non come statistiche.

Il sotto-agente non consultato. L'orchestratore ha tre revisori e ne chiama due, perché ha giudicato il terzo non pertinente. La risposta finale riporta due pareri, è ben scritta, è coerente, e manca il parere che avrebbe bloccato la clausola. Il punto cruciale è che non c'è nessun segnale: nessuna eccezione, nessun campo vuoto, nessun avviso. Per accorgersene bisogna confrontare i tool chiamati con quelli attesi, cioè fare esattamente il controllo che l'orchestrazione in codice rende inutile.

L'aggregazione impressionistica. «Il verdetto complessivo è il più severo fra i tre» è una funzione. Se la lasci al modello, può applicarla correttamente oppure produrre una media pesata dalle sue impressioni: due «da modificare» e un «inaccettabile» diventano «da modificare», perché la maggioranza suggerisce quella direzione. Il risultato è difendibile a parole e sbagliato rispetto alla regola che avevi stabilito.

Il contesto non propagato. L'orchestratore chiama parere_legale("valuta la clausola di cui sopra") perché lui ce l'ha davanti. Il sotto-agente riceve solo quella frase e, non avendo il testo, produce un parere generico oppure inventa una clausola plausibile e la valuta. La risposta che torna ha la forma giusta, il campo citazione contiene una citazione inventata, e l'orchestratore la integra nella sintesi senza battere ciglio.

I tre fallimenti hanno una proprietà comune: la risposta finale è ben formata. Un sistema che si rompe con un'eccezione ti avvisa; un sistema che risponde bene a una domanda a cui ha risposto male non ti avvisa mai, e lo scopri quando qualcuno a valle verifica a mano. Da qui la regola operativa: ogni volta che un passo dell'orchestrazione è obbligatorio, deve essere obbligatorio nel codice. Il prompt può chiedere; soltanto il codice può garantire.

Il codice: tre revisori e un coordinatore

Prendiamo un caso concreto: una clausola contrattuale da valutare sotto tre profili diversi. Ogni revisore ha istruzioni proprie e un output_type (un modello Pydantic, visto nell'articolo sull'output strutturato); l'orchestratore li espone come tool con as_tool, li interpella e produce un verdetto unico. In questa prova i revisori sono copioni che restituiscono un parere già scritto, e il copione dell'orchestratore chiama i tre tool nello stesso giro, poi sintetizza.

import asyncio
import json
from typing import Literal
from pydantic import BaseModel, Field
from agents import Agent, Runner
from fake import ModelloFinto, chiama, testo

class Parere(BaseModel):
    verdetto: Literal["accettabile", "da_modificare", "inaccettabile"]
    motivo: str = Field(max_length=400)
    citazione: str = Field(description="La parte esatta del testo su cui ti basi")

CLAUSOLA = ("Il fornitore garantisce la disponibilità del servizio al 100% "
            "e si impegna a rimborsare l'intero corrispettivo annuale "
            "per qualsiasi interruzione superiore a un minuto.")

visto = {"legale": [], "fiscale": [], "tecnico": []}

def parere(verdetto, motivo):
    """Il 'revisore' di questa prova: un copione che restituisce un Parere gia' scritto."""
    return testo(json.dumps({"verdetto": verdetto, "motivo": motivo, "citazione": CLAUSOLA[:40]}))

revisore_legale = Agent(
    name="Revisore legale",
    instructions=("Valuti clausole contrattuali sotto il profilo civilistico. "
                  "Cita sempre la porzione di testo su cui ti basi. "
                  "Se la clausola non riguarda il tuo profilo, verdetto='accettabile' "
                  "e motivo='fuori dal mio perimetro'."),
    output_type=Parere,
    model=ModelloFinto([parere("da_modificare", "Una garanzia al 100% è un'obbligazione di risultato.")], visto["legale"]),
)
revisore_fiscale = Agent(
    name="Revisore fiscale",
    instructions="Valuti le implicazioni fiscali e di fatturazione. Stesse regole.",
    output_type=Parere,
    model=ModelloFinto([parere("accettabile", "fuori dal mio perimetro")], visto["fiscale"]),
)
revisore_tecnico = Agent(
    name="Revisore tecnico",
    instructions="Valuti la fattibilità tecnica degli impegni assunti. Stesse regole.",
    output_type=Parere,
    model=ModelloFinto([parere("inaccettabile", "Nessun sistema garantisce il 100%.")], visto["tecnico"]),
)

registro = []
orchestratore = Agent(
    name="Coordinatore revisione",
    instructions=(
        "Ricevi una clausola. Devi consultare TUTTI E TRE i revisori, sempre, "
        "anche se pensi che uno non sia pertinente: è lui a doverlo dire.\n"
        "Poi sintetizza: riporta i tre verdetti, e il verdetto complessivo è il "
        "più severo fra i tre. Elenca le citazioni.\n"
        "Non esprimere opinioni tue: solo sintesi di quelle ricevute."
    ),
    tools=[
        revisore_legale.as_tool(tool_name="parere_legale",
            tool_description="Parere civilistico su una clausola. Passa il testo integrale."),
        revisore_fiscale.as_tool(tool_name="parere_fiscale",
            tool_description="Parere fiscale su una clausola. Passa il testo integrale."),
        revisore_tecnico.as_tool(tool_name="parere_tecnico",
            tool_description="Parere di fattibilità tecnica. Passa il testo integrale."),
    ],
    # Il modello finto chiama i tre tool nello stesso giro, poi sintetizza.
    model=ModelloFinto([
        [chiama("parere_legale", input=CLAUSOLA), chiama("parere_fiscale", input=CLAUSOLA),
         chiama("parere_tecnico", input=CLAUSOLA)],
        testo("Verdetti: legale da_modificare, fiscale accettabile, tecnico inaccettabile. "
              "Complessivo: inaccettabile."),
    ], registro),
)

async def main():
    esito = await Runner.run(orchestratore, CLAUSOLA, max_turns=8)
    print(esito.final_output)
    print("tool dell'orchestratore:", registro[0]["tool"], "| handoff:", registro[0]["handoff"])
    print("ultimo agente:", esito.last_agent.name)
    print("il revisore legale ha ricevuto:", visto["legale"][0]["input"])
    print("l'orchestratore, al secondo giro, ha letto:",
          [m["output"][:60] for m in registro[1]["input"] if m.get("type") == "function_call_output"])

asyncio.run(main())

# Output:
# Verdetti: legale da_modificare, fiscale accettabile, tecnico inaccettabile. Complessivo: inaccettabile.
# tool dell'orchestratore: ['parere_legale', 'parere_fiscale', 'parere_tecnico'] | handoff: []
# ultimo agente: Coordinatore revisione
# il revisore legale ha ricevuto: [{'content': "Il fornitore garantisce la disponibilità del servizio al 100% e si impegna a rimborsare l'intero corrispettivo annuale per qualsiasi interruzione superiore a un minuto.", 'role': 'user'}]
# l'orchestratore, al secondo giro, ha letto: ['verdetto=\'da_modificare\' motivo="Una garanzia al 100% è un\'o', "verdetto='accettabile' motivo='fuori dal mio perimetro' cita", "verdetto='inaccettabile' motivo='Nessun sistema garantisce i"]

L'uscita contiene tre fatti che contano. Primo: l'orchestratore ha tre tool e nessun handoff, e l'agente che chiude la run è lui, non un sotto-agente. Secondo: il revisore legale ha ricevuto soltanto la stringa passata nell'argomento input del tool, non la conversazione dell'orchestratore; per questo la tool_description dice «passa il testo integrale». Terzo, ed è un dettaglio che non si indovina: i pareri tornano all'orchestratore come il testo dell'oggetto Pydantic (verdetto='da_modificare' motivo=...), non come JSON. Funziona per un modello che li legge, ma un parsing a valle non può contarci: se ti serve JSON, lo produci tu con un estrattore di output personalizzato (custom_output_extractor di as_tool) oppure orchestrando in codice.

Le scelte di progetto che fanno funzionare il pattern sono tre.

• «Devi consultare TUTTI E TRE, sempre». Senza questa riga l'orchestratore può consultarne uno o due, perché valuta da sé la pertinenza, e quando sbaglia non te ne accorgi perché il risultato sembra completo. Se la consultazione di tutti è un requisito rigido, la forma corretta non è nemmeno un prompt: chiami i tre sotto-agenti tu e passi i tre pareri a un quarto agente che sintetizza.
• «Il verdetto complessivo è il più severo». La regola di aggregazione va dichiarata, altrimenti l'orchestratore fa una media impressionistica. Se la regola è deterministica è meglio implementarla in Python: max(verdetti, key=severita) non sbaglia mai.
• output_type sui sotto-agenti, non sull'orchestratore. Così ogni parere ha una forma controllata, con il campo citazione obbligatorio che rende la risposta verificabile, mentre la sintesi finale resta testo per l'utente.

Le tre scelte hanno lo stesso scopo: spostare dal modello al codice o allo schema tutto ciò che deve essere garantito. L'orchestratore resta libero solo di sintetizzare, che è l'unica cosa per cui serve un modello. Più sposti verso il codice, meno la qualità del risultato dipende da un modello che «si ricorda» di fare la cosa giusta.

Il parallelismo, che è il vero guadagno

Il modello può chiedere i tre tool in un unico giro, come nel copione di sopra; se il runner li esegue insieme il tempo totale è quello del revisore più lento invece della somma. Se vuoi il parallelismo garantito, non sperato, l'orchestrazione la scrivi tu con asyncio, come nell'articolo su asyncio. Nel blocco seguente i modelli finti ci mettono un secondo a rispondere (ModelloLento), per rendere visibile la differenza.

import asyncio
import json
import time
from typing import Literal
from pydantic import BaseModel
from agents import Agent, Runner
from fake import ModelloFinto, testo

class Parere(BaseModel):
    verdetto: Literal["accettabile", "da_modificare", "inaccettabile"]
    motivo: str

class ModelloLento(ModelloFinto):
    """Come il modello finto, ma ci mette `secondi` a rispondere (come una chiamata di rete)."""
    def __init__(self, copione, secondi):
        super().__init__(copione)
        self.secondi = secondi
    async def get_response(self, *args, **kwargs):
        await asyncio.sleep(self.secondi)
        return await super().get_response(*args, **kwargs)

def revisore(nome, verdetto, secondi):
    risposta = testo(json.dumps({"verdetto": verdetto, "motivo": "..."}))
    return Agent(name=nome, instructions="Valuta la clausola.", output_type=Parere,
                 model=ModelloLento([risposta], secondi))

async def revisione_parallela(revisori: dict, clausola: str, timeout_s: float) -> dict:
    """Tre pareri in parallelo, garantiti. Nessuna decisione lasciata al modello."""
    async def uno(nome: str, agente):
        try:
            async with asyncio.timeout(timeout_s):
                esito = await Runner.run(agente, clausola, max_turns=4)
            return nome, esito.final_output
        except TimeoutError:
            return nome, None

    coppie = await asyncio.gather(*(uno(n, a) for n, a in revisori.items()))
    pareri = dict(coppie)

    mancanti = [n for n, p in pareri.items() if p is None]
    if mancanti:
        return {"ok": False, "revisori_non_pervenuti": mancanti}

    severita = {"accettabile": 0, "da_modificare": 1, "inaccettabile": 2}
    peggiore = max(pareri.values(), key=lambda p: severita[p.verdetto])
    return {"ok": True,
            "verdetto": peggiore.verdetto,                 # aggregazione deterministica
            "pareri": {n: p.model_dump() for n, p in pareri.items()}}

def squadra(secondi_tecnico: float) -> dict:
    return {"legale": revisore("Legale", "da_modificare", 1.0),
            "fiscale": revisore("Fiscale", "accettabile", 1.0),
            "tecnico": revisore("Tecnico", "inaccettabile", secondi_tecnico)}

async def main():
    clausola = "Garanzia di disponibilità al 100%."
    # Ogni revisore impiega 1 s: in parallelo il totale è circa 1 s, in serie sarebbe circa 3 s.
    t0 = time.perf_counter()
    esito = await revisione_parallela(squadra(1.0), clausola, timeout_s=2)
    durata = time.perf_counter() - t0
    print(esito["ok"], esito["verdetto"], "| meno della somma (3 s):", durata < 2.0)

    # Un revisore che non risponde in tempo: errore esplicito, non una sintesi incompleta.
    print(await revisione_parallela(squadra(5), clausola, timeout_s=1.5))

asyncio.run(main())

# Output:
# True inaccettabile | meno della somma (3 s): True
# {'ok': False, 'revisori_non_pervenuti': ['tecnico']}

Le righe che contano sono tre. asyncio.gather lancia i tre revisori insieme, quindi con un secondo ciascuno il totale è circa un secondo e non tre (la prima riga dell'uscita lo conferma con una soglia larga, perché i tempi cambiano da una macchina all'altra e da un'esecuzione all'altra). asyncio.timeout dà a ogni revisore un tempo massimo, e il ramo except TimeoutError trasforma il ritardo in un esito esplicito: nella seconda esecuzione il tecnico impiega cinque secondi, il timeout è 1,5 e l'uscita dice revisori_non_pervenuti: ['tecnico'] invece di una sintesi silenziosamente incompleta. Infine max(..., key=severita) è l'aggregazione deterministica.

Questa versione è più lunga e migliore su tre fronti: il parallelismo è certo, l'aggregazione non può essere reinterpretata, e un revisore assente è un errore visibile. La regola che ne discende: l'orchestrazione la lasci al modello solo quando il piano non è noto in anticipo. Se sai che vanno chiamati tre revisori e che il verdetto è il massimo, quello è codice.

Il costo, misurato

Qui si capisce perché il pattern va dosato. Ho misurato quanti caratteri vengono inviati ai modelli in tutta la run, sommando orchestratore e sotto-agenti, per una clausola di circa mille caratteri, contro un agente generalista che la legge una volta sola. Sono caratteri, non token, e contano il JSON della cronologia con le sue sequenze di escape: la proporzione è indicativa, non un conteggio di fatturazione.

import asyncio
from agents import Agent, Runner
from fake import ModelloFinto, chiama, testo

CLAUSOLA = ("Il fornitore garantisce la disponibilità del servizio al 100% "
            "e si impegna a rimborsare l'intero corrispettivo annuale "
            "per qualsiasi interruzione superiore a un minuto. ") * 6      # una clausola lunga
PARERE = "verdetto='da_modificare' motivo='" + "x" * 200 + "' citazione='..."   # ~230 caratteri

def caratteri(registro):
    return sum(v["caratteri"] for v in registro)

async def con_sotto_agenti(k: int) -> int:
    registro = []                       # un solo registro: orchestratore e sotto-agenti insieme
    istruzioni_revisore = "Valuti la clausola sotto il tuo profilo e cita il testo su cui ti basi."
    revisori = [Agent(name=f"Revisore {i}", instructions=istruzioni_revisore,
                      model=ModelloFinto([testo(PARERE)], registro)) for i in range(k)]
    tools = [r.as_tool(tool_name=f"parere_{i}", tool_description="Parere su una clausola. Passa il testo integrale.")
             for i, r in enumerate(revisori)]
    orchestratore = Agent(name="Coordinatore", instructions="Consulta tutti i revisori e sintetizza.",
                          tools=tools, model=ModelloFinto([
                              [chiama(f"parere_{i}", input=CLAUSOLA) for i in range(k)],
                              testo("Sintesi.")], registro))
    await Runner.run(orchestratore, CLAUSOLA)
    return caratteri(registro)

async def generalista() -> int:
    registro = []
    agente = Agent(name="Generalista", instructions="Valuti la clausola sotto ogni profilo.",
                   model=ModelloFinto([testo("Sintesi.")], registro))
    await Runner.run(agente, CLAUSOLA)
    return caratteri(registro)

async def main():
    base = await generalista()
    print(f"generalista: {base} caratteri inviati ({len(CLAUSOLA)} sono la clausola)")
    for k in (1, 2, 3, 5):
        n = await con_sotto_agenti(k)
        print(f"orchestratore + {k} sotto-agenti: {n:>6} caratteri inviati, {n / base:.1f} volte il generalista")

asyncio.run(main())

# Output:
# generalista: 1117 caratteri inviati (1014 sono la clausola)
# orchestratore + 1 sotto-agenti:   5460 caratteri inviati, 4.9 volte il generalista
# orchestratore + 2 sotto-agenti:   8688 caratteri inviati, 7.8 volte il generalista
# orchestratore + 3 sotto-agenti:  11916 caratteri inviati, 10.7 volte il generalista
# orchestratore + 5 sotto-agenti:  18375 caratteri inviati, 16.5 volte il generalista

Il costo cresce in modo lineare con il numero dei sotto-agenti: ogni sotto-agente in più aggiunge circa 3.200 caratteri, cioè più di tre volte la lunghezza della clausola. Non ho scomposto la cifra voce per voce, ma le cause sono evidenti nel flusso: il testo viaggia come input del sotto-agente, come argomento della chiamata che l'orchestratore deve scrivere (e quindi generare) e di nuovo nella cronologia che l'orchestratore rilegge al secondo giro, insieme al parere. Non è una ragione per non usarlo: è una ragione per sapere quando la qualità aggiuntiva giustifica un moltiplicatore che, in questa prova, va da 5 volte con un sotto-agente a oltre 16 con cinque. Per il lato economico, misurazione e controllo del costo per richiesta, c'è FinOps e ottimizzazione dei costi per agenti LLM.

Profondità e ricorsione: il contatore che serve davvero

Il tetto max_turns protegge un agente da un ciclo al proprio interno. Non protegge da niente quando il problema è l'annidamento: un orchestratore con max_turns=8 che chiama un sotto-agente con max_turns=8, che a sua volta ne chiama un altro, può produrre molte chiamate al modello pur rispettando ogni singolo tetto. E se per distrazione l'agente A ha fra i propri tool l'agente B e B ha A, la ricorsione è illimitata.

La difesa è un contatore di profondità che vive nel contesto della run, cioè nell'oggetto tuo che l'SDK passa ai tool e che il modello non vede né può modificare. Nel blocco seguente come_tool sostituisce as_tool: costruisce un function tool che controlla profondità, percorso e budget prima di far girare il sotto-agente, e scrive una riga di log JSON per evento (registra_passo), con la posizione nell'albero sempre presente. La prova ha una dipendenza circolare di proposito: il legale usa il fiscale e il fiscale usa il legale. Viene eseguita con due limiti di profondità per vedere i due controlli.

import asyncio
import json
import time
from dataclasses import dataclass, field
from agents import Agent, Runner, function_tool, RunContextWrapper
from fake import ModelloFinto, chiama, testo

@dataclass
class ContestoEsecuzione:
    id_run: str
    profondita: int = 0
    max_profondita: int = 2
    chiamate_modello: int = 0
    max_chiamate_modello: int = 25
    percorso: list[str] = field(default_factory=list)   # chi ha chiamato chi

class LimiteGerarchia(RuntimeError):
    pass

def registra_passo(c: ContestoEsecuzione, evento: str, **extra) -> None:
    """Una riga per evento, con la posizione nell'albero sempre presente."""
    riga = {"run": c.id_run, "profondita": c.profondita,
            "percorso": "/".join(c.percorso) or "radice",   # "legale/fiscale"
            "evento": evento, **extra}
    print(json.dumps(riga, ensure_ascii=False))

def numero_di_giri(esito) -> int:
    return len(esito.raw_responses)          # una risposta del modello per giro

def come_tool(agente: Agent, nome: str, descrizione: str):
    """Espone un agente come tool, con controllo di profondità e di budget."""

    @function_tool(name_override=nome, description_override=descrizione)
    async def invoca(contesto: RunContextWrapper[ContestoEsecuzione], richiesta: str) -> str:
        c = contesto.context

        if c.profondita >= c.max_profondita:
            return ("ERRORE: profondità massima raggiunta. "
                    "Rispondi con i dati che hai già e dichiara cosa manca.")
        if nome in c.percorso:
            return (f"ERRORE: {nome} è già nel percorso di chiamata {c.percorso}. "
                    "Ricorsione non consentita: concludi con i dati disponibili.")
        if c.chiamate_modello >= c.max_chiamate_modello:
            raise LimiteGerarchia(f"budget di chiamate esaurito su {c.percorso}")

        c.profondita += 1
        c.percorso.append(nome)
        registra_passo(c, "sotto_agente_inizio", agente=nome)
        try:
            esito = await Runner.run(agente, richiesta, context=c, max_turns=5)
            c.chiamate_modello += numero_di_giri(esito)
            registra_passo(c, "sotto_agente_fine", agente=nome, giri=numero_di_giri(esito))
            return str(esito.final_output)
        finally:
            c.profondita -= 1            # ripristina anche in caso di eccezione
            c.percorso.pop()

    return invoca

def costruisci():
    """Tre agenti con una dipendenza circolare: legale usa fiscale, fiscale usa legale."""
    visto = []
    legale = Agent(name="Legale", instructions="l", model=ModelloFinto([
        chiama("fiscale", richiesta="e l'IVA?"), testo("Parere legale completo.")]))
    fiscale = Agent(name="Fiscale", instructions="f", model=ModelloFinto([
        chiama("legale", richiesta="e il contratto?"), testo("Parere fiscale con i dati che ho.")], visto))
    legale.tools = [come_tool(fiscale, "fiscale", "Parere fiscale.")]
    fiscale.tools = [come_tool(legale, "legale", "Parere legale.")]
    coordinatore = Agent(name="Coordinatore", instructions="c", model=ModelloFinto([
        chiama("legale", richiesta="valuta la clausola"), testo("Sintesi.")]),
        tools=[come_tool(legale, "legale", "Parere legale.")])
    return coordinatore, visto

async def main():
    for max_profondita in (2, 3):
        print(f"--- max_profondita = {max_profondita}")
        coordinatore, visto = costruisci()
        contesto = ContestoEsecuzione("r-7", max_profondita=max_profondita)
        esito = await Runner.run(coordinatore, "clausola", context=contesto)
        risposta_al_fiscale = [m["output"] for m in visto[1]["input"] if m.get("type") == "function_call_output"]
        print("il fiscale ha letto:", risposta_al_fiscale)
        print("esito:", esito.final_output, "| giri contati:", contesto.chiamate_modello, "| profondità finale:", contesto.profondita)

asyncio.run(main())

# Output:
# --- max_profondita = 2
# {"run": "r-7", "profondita": 1, "percorso": "legale", "evento": "sotto_agente_inizio", "agente": "legale"}
# {"run": "r-7", "profondita": 2, "percorso": "legale/fiscale", "evento": "sotto_agente_inizio", "agente": "fiscale"}
# {"run": "r-7", "profondita": 2, "percorso": "legale/fiscale", "evento": "sotto_agente_fine", "agente": "fiscale", "giri": 2}
# {"run": "r-7", "profondita": 1, "percorso": "legale", "evento": "sotto_agente_fine", "agente": "legale", "giri": 2}
# il fiscale ha letto: ['ERRORE: profondità massima raggiunta. Rispondi con i dati che hai già e dichiara cosa manca.']
# esito: Sintesi. | giri contati: 4 | profondità finale: 0
# --- max_profondita = 3
# {"run": "r-7", "profondita": 1, "percorso": "legale", "evento": "sotto_agente_inizio", "agente": "legale"}
# {"run": "r-7", "profondita": 2, "percorso": "legale/fiscale", "evento": "sotto_agente_inizio", "agente": "fiscale"}
# {"run": "r-7", "profondita": 2, "percorso": "legale/fiscale", "evento": "sotto_agente_fine", "agente": "fiscale", "giri": 2}
# {"run": "r-7", "profondita": 1, "percorso": "legale", "evento": "sotto_agente_fine", "agente": "legale", "giri": 2}
# il fiscale ha letto: ["ERRORE: legale è già nel percorso di chiamata ['legale', 'fiscale']. Ricorsione non consentita: concludi con i dati disponibili."]
# esito: Sintesi. | giri contati: 4 | profondità finale: 0

Con max_profondita = 2 la catena legale → fiscale arriva al limite e il fiscale riceve l'errore di profondità; con 3 passa il primo controllo e interviene quello sul percorso, che rileva che legale è già nella catena. In entrambi i casi la run si chiude con la sintesi, il contatore torna a zero (finally) e nessuna eccezione sale. Quattro dettagli rendono questo involucro diverso da un semplice as_tool().

I primi due controlli restituiscono un errore come testo invece di sollevare. È la stessa scelta dell'articolo sui tool: «ho raggiunto il limite di profondità» è una situazione da cui l'orchestratore può recuperare, concludendo con i dati che ha. Un'eccezione ucciderebbe la run intera per un sotto-agente mancante, perdendo i pareri già raccolti. Il terzo controllo invece solleva: il budget complessivo di chiamate non è recuperabile, e continuare significa solo spendere. Nel codice il contatore somma i giri dei sotto-agenti (numero_di_giri legge raw_responses) e non conta i giri del coordinatore.

Il controllo sul percorso chiude la ricorsione. nome in c.percorso rileva non solo A→A ma anche A→B→A, che è la forma che nessuno prevede e che si verifica quando due tool sono stati registrati in momenti diversi dello sviluppo. Il percorso è anche l'informazione da avere nei log: «legale/fiscale» spiega in un colpo d'occhio una bolletta anomala. Il finally ripristina la profondità: senza di esso un sotto-agente che solleva lascia il contatore incrementato, e dopo qualche eccezione il sistema crede di essere profondo pur essendo alla radice, rifiutando chiamate legittime. È un guasto che si manifesta solo sotto errore, quindi solo in produzione.

Sul fronte della diagnosi, la traccia di una run con sotto-agenti è un albero, ma gli strumenti la mostrano spesso come lista piatta. Con il campo percorso su ogni riga tre domande diventano ricerche di testo: quali sotto-agenti sono stati chiamati in questa run (filtra per run), quale ramo è costato più giri (raggruppa per percorso), perché la sintesi non cita il parere fiscale (cerca fiscale: se non compare, il sotto-agente non è stato chiamato, e il problema è di orchestrazione, non di sintesi). Il campo run deve essere lo stesso per tutta la gerarchia, ed è il motivo per cui il contesto viene passato ai sotto-agenti con context=c invece di crearne uno nuovo.

Un ultimo punto riguarda la progettazione più che il codice: due livelli bastano di solito. Un orchestratore e i suoi specialisti sono comprensibili, tracciabili e discutibili. Al terzo livello il costo si moltiplica di nuovo e la traccia diventa un albero che nessuno legge. Il valore predefinito max_profondita = 2 è una scelta di prudenza, non un limite dell'SDK: cambialo solo con una ragione misurata.

Un sotto-agente particolare: il verificatore

C'è un uso del pattern che non è «chiedere a uno specialista» ma «far controllare il proprio lavoro». Lo scrittore produce una bozza, la passa a un verificatore come tool, e se torna un rifiuto riscrive. È il ciclo produci-verifica-correggi, che Anthropic chiama evaluator-optimizer e raccomanda quando esistono criteri di valutazione chiari. Il verificatore di questa prova è un controllo deterministico sul numero di parole, perché la specifica è oggettiva (al massimo dodici parole) e il copione dello scrittore invia due bozze, una troppo lunga e una corretta.

import asyncio
import json
from typing import Literal
from pydantic import BaseModel, Field
from agents import Agent, Runner
from fake import ModelloFinto, chiama, testo

class Verifica(BaseModel):
    esito: Literal["approvato", "da_rifare"]
    problemi: list[str] = Field(description="Uno per riga, concreto e azionabile")

SPECIFICA = "Al massimo 12 parole."

def controllo_oggettivo(input_utente):
    """Il 'verificatore' di questa prova: un controllo deterministico sulla specifica."""
    messaggio = input_utente if isinstance(input_utente, str) else input_utente[-1]["content"]
    parole = len(messaggio.split("\nSPECIFICA:")[0].replace("TESTO:", "").split())
    if parole > 12:
        return testo(json.dumps({"esito": "da_rifare",
                                 "problemi": [f"{parole} parole, il limite è 12"]}))
    return testo(json.dumps({"esito": "approvato", "problemi": []}))

verificatore = Agent(
    name="Verificatore",
    instructions=(
        "Ricevi un testo e la specifica che doveva rispettare. "
        "Controlla SOLO il rispetto della specifica, punto per punto. "
        "Non riscrivere il testo, non dare consigli di stile. "
        "Se un requisito non è rispettato, esito='da_rifare' e dillo citando il punto."
    ),
    output_type=Verifica,
    model=ModelloFinto(controllo_oggettivo),
)

BOZZA_1 = "Il servizio resta attivo quasi sempre ma in rari casi può fermarsi per manutenzione programmata"
BOZZA_2 = "Il servizio può fermarsi solo per manutenzione programmata."

registro = []
scrittore = Agent(
    name="Scrittore",
    instructions=(
        "Scrivi il testo richiesto. Poi chiama SEMPRE verifica_testo passando "
        "il tuo testo E la specifica. Se torna 'da_rifare', correggi i problemi "
        "elencati e verifica di nuovo. Massimo due correzioni, poi consegna "
        "dichiarando cosa resta non conforme."
    ),
    tools=[verificatore.as_tool(
        tool_name="verifica_testo",
        tool_description=("Controlla un testo contro una specifica. "
                          "Passa sia il testo sia la specifica integrale."))],
    model=ModelloFinto([
        chiama("verifica_testo", input=f"TESTO: {BOZZA_1}\nSPECIFICA: {SPECIFICA}"),
        chiama("verifica_testo", input=f"TESTO: {BOZZA_2}\nSPECIFICA: {SPECIFICA}"),
        testo(BOZZA_2),
    ], registro),
)

async def main():
    esito = await Runner.run(scrittore, "Spiega la disponibilità del servizio.")
    print("testo consegnato:", esito.final_output)
    for m in registro[-1]["input"]:
        if m.get("type") == "function_call_output":
            print("il verificatore ha risposto:", m["output"])

asyncio.run(main())

# Output:
# testo consegnato: Il servizio può fermarsi solo per manutenzione programmata.
# il verificatore ha risposto: esito='da_rifare' problemi=['15 parole, il limite è 12']
# il verificatore ha risposto: esito='approvato' problemi=[]

Il testo consegnato è la seconda bozza, e le due ultime righe dell'uscita mostrano il percorso: la prima viene respinta con un problema concreto («15 parole, il limite è 12»), la seconda approvata. Due dettagli distinguono un ciclo utile da uno infinito. «Massimo due correzioni, poi consegna dichiarando cosa resta non conforme» impedisce il rimpallo fra scrittore e verificatore, che altrimenti continuerebbe fino a max_turns con l'utente in attesa. E «non riscrivere il testo» sul verificatore evita che diventi un secondo scrittore, la cui proposta lo scrittore adotta senza valutarla.

Il limite da conoscere: lo schema funziona quando la verifica è più facile della produzione e si appoggia a un criterio esterno (una specifica scritta, uno schema, un test che gira). Non funziona come «pensaci meglio». Huang et al. (Large Language Models Cannot Self-Correct Reasoning Yet) riportano che, senza un segnale esterno di correttezza, l'autocorrezione di un modello sul proprio ragionamento non migliora e può peggiorare le risposte. La conseguenza pratica è che il verificatore deve avere in mano qualcosa di oggettivo da controllare, altrimenti paghi due chiamate per aggiungere rumore. Dove un controllo oggettivo non esiste, meglio un guardrail deterministico o una revisione umana, e sono il tema dei prossimi due articoli: i guardrail e le trappole reali.

Scegliere fra le due forme

• Serve la risposta per elaborarla? Agent-as-tool. Serve che prosegua lui? Handoff.
• Serve confrontare più pareri? Agent-as-tool (l'handoff va in un posto solo).
• Il sotto-agente deve parlare con l'utente? Handoff (con l'agent-as-tool l'utente non lo vede mai).
• Conta il costo di input? Handoff (un agente alla volta, nessuna duplicazione dell'input).
• Conta avere un unico punto di responsabilità sulla risposta finale? Agent-as-tool: l'orchestratore è l'unico che parla, quindi l'unico posto dove imporre tono, formato e guardrail di uscita.

Le cinque domande ruotano attorno a chi deve avere l'ultima parola. Se l'ultima parola deve essere dello specialista, handoff; se deve restare a un punto unico che controlla la forma della risposta, agent-as-tool. Quando le risposte divergono (vuoi la voce dello specialista ma anche un controllo di uscita) il compromesso è spesso un handoff con un guardrail sull'output dello specialista, che è il tema dell'articolo sui guardrail.

Trade-off

• Qualità contro costo. Specialisti con istruzioni mirate possono sbagliare meno di un generalista, e lo paghi in input moltiplicato (misurato sopra) e in latenza, se non parallelizzi.
• Controllo contro autonomia. Orchestrare in Python ti dà parallelismo certo e aggregazione deterministica; orchestrare col modello gestisce i casi che non avevi previsto. La via di mezzo pratica: piano fisso in codice, e un ramo «altro» affidato al modello.
• Contesto duplicato. Ogni sotto-agente riceve il proprio input: non vede la conversazione dell'orchestratore e non ne eredita la confusione, ma devi passargli esplicitamente tutto ciò che gli serve. Dimenticare un pezzo è l'errore tipico.
• Profondità della gerarchia. Due livelli sono gestibili. Tre livelli moltiplicano costo e latenza e rendono le tracce difficili da leggere.

Le quattro voci tengono insieme un solo criterio: ogni volta che aggiungi un livello di autonomia aggiungi un costo di controllo. Per un sistema che deve rispondere in modo ripetibile, comincia dal lato del codice e dai al modello solo la sintesi; sposta responsabilità sul modello quando una misura (non un'impressione) mostra che il piano fisso non copre i casi reali.

Le trappole in cui si cade davvero

1. Ricorsione involontaria. Se A ha come tool B e B ha come tool A, hai una ricorsione che si manifesta come esplosione di costo. max_turns protegge il singolo agente, non la profondità dell'annidamento: serve un contatore nel contesto della run.
2. Non passare il contesto al sotto-agente. L'orchestratore chiama parere_legale("valuta la clausola") senza allegare il testo. Il sotto-agente non lo ha, e inventa. La tool_description deve dirlo («passa il testo integrale»).
3. Lasciare l'aggregazione al modello quando la regola è nota. «Il più severo dei tre» è max: costa una riga e non sbaglia.
4. Un sotto-agente che restituisce un muro di testo. La sua risposta diventa contesto dell'orchestratore e la paghi a ogni giro successivo. Vincola i sotto-agenti a output brevi e tipizzati.
5. Nessun timeout sui sotto-agenti. Un sotto-agente che gira a vuoto blocca l'intero albero. Timeout per ciascuno, e un esito esplicito per il caso «non pervenuto».
6. Tracce illeggibili. Senza un identificativo di run condiviso e un percorso, la traccia di una run gerarchica è un elenco piatto in cui non si capisce chi ha chiamato chi.

Le sei trappole condividono una causa: nella gerarchia gli errori non sono più locali. Un sotto-agente sbagliato non fa fallire la run, ne avvelena la sintesi; un ciclo non esplode subito, si accumula in fatture. Le difese viste in questo articolo (contesto esplicito, aggregazione in codice, timeout, contatore di profondità, log ad albero) servono tutte a rendere l'errore visibile nel punto in cui nasce.

Che cosa sai fare adesso

Sai usare as_tool per esporre un agente come tool e sai che il sotto-agente riceve solo la stringa che gli passi, e che i suoi risultati tornano come testo. Sai orchestrare in Python con asyncio.gather, timeout per revisore e aggregazione deterministica, e sai misurare il moltiplicatore di input della gerarchia sul tuo caso. Sai proteggere una gerarchia dalla ricorsione con un contatore di profondità nel contesto, e costruire un ciclo scrittore-verificatore con un criterio esterno. Nell'articolo successivo passiamo alla sicurezza: i guardrail, cioè i controlli che girano accanto all'agente e possono fermarlo.