06 - Orchestrazione Multi-Agente: Architetture Production

Il problema: tre agenti nel prototipo, una dozzina in produzione

Un prototipo con due o tre agenti funziona quasi sempre, perché la persona che lo guarda fa da supervisore: vede il loop, interrompe, riavvia. In produzione quel supervisore non c'è, e i fastidi diventano guasti: due agenti che si rimbalzano la stessa richiesta, una conversazione che non finisce, un errore che arriva intatto alla risposta finale, una bolletta di token imprevista. Questo articolo dà un criterio per decidere se ti serve un sistema multi-agente, descrive i cinque modi di coordinare il lavoro e le tre architetture per collegare gli agenti, e poi quattro difese che valgono qualunque scelta tu faccia: un tetto a ogni ciclo, un registro di ciò che è successo, ritentativi con un interruttore, tracce.

I listati sono eseguiti in locale con la sola libreria standard di Python 3.14.4, e gli agenti sono funzioni finte che seguono una regola fissa: nessun modello chiamato, nessun broker o database avviato. Mostrano la meccanica dell'orchestrazione, non il comportamento di un modello vero. Se non hai letto come ragiona un singolo agente, parti da Fondamenti: OODA Loop, ReAct e Tool Calling; i framework confrontati prima sono LangChain e LangGraph, CrewAI e AutoGen.

Prima di orchestrare: ti servono più agenti?

La guida di Microsoft sui pattern di orchestrazione apre con una scala di complessità: una chiamata diretta al modello, un singolo agente con strumenti, un'orchestrazione multi-agente. Il consiglio è usare il livello più basso che soddisfa i requisiti in modo affidabile, perché ogni livello aggiunge coordinamento, latenza e costo. Il singolo agente con strumenti è indicato come scelta spesso giusta per i casi aziendali, più semplice da debuggare e da testare. Giustificano il salto un prompt troppo complesso per un agente solo, troppi strumenti insieme, oppure confini di sicurezza diversi per ciascun agente.

Due fonti misurano il costo del salto. Anthropic, descrivendo il proprio sistema di ricerca con un agente coordinatore e sottoagenti in parallelo, riporta che nei suoi dati gli agenti usano circa 4 volte i token di una chat e i sistemi multi-agente circa 15 volte, e ne ricava che servono compiti di valore abbastanza alto da pagarlo. Il lavoro «Towards a Science of Scaling Agent Systems» (260 configurazioni, sei benchmark, cinque architetture) misura invece gli errori: nelle sue prove agenti indipendenti, senza verifica centrale, amplificano gli errori di traccia di 17,2 volte rispetto a un agente singolo; un coordinamento centralizzato che valida prima di aggregare li contiene a 4,4 volte; decentralizzato e ibrido stanno in mezzo (7,8 e 5,1). Gli autori trovano anche che dove l'agente singolo supera già il 45% di accuratezza aggiungere agenti dà rendimenti negativi.

Quelle cifre descrivono i benchmark degli autori, non il tuo sistema, e non vanno copiate come soglie. Dicono però che i modi in cui un sistema multi-agente va male sono prevedibili e si riconducono a un elenco corto.

• un ciclo senza tetto (handoff che rimbalza, conversazione che non converge, agente che richiama lo stesso strumento);
• un errore che nessuno controlla e che passa intatto da un agente al successivo;
• un contesto che cresce a ogni passaggio, con costo e latenza che crescono insieme;
• un guasto di un componente (modello, strumento, broker) che si propaga agli altri;
• un comportamento che non puoi ricostruire dopo, perché nessuno ha registrato che cosa è successo.

I cinque punti sono indipendenti e l'ordine non conta, ma non pesano allo stesso modo: i primi due sono difetti di progetto che si prevengono con regole nel codice (tetti, verifica), il terzo con la scelta di che cosa passare da un agente all'altro, gli ultimi due con infrastruttura (ritentativi, interruttori, tracce). Il resto dell'articolo segue questo ordine.

I cinque pattern di orchestrazione

La guida di Microsoft nomina cinque pattern, ciascuno con i suoi sinonimi in uso nel settore. La tabella li riporta con la regola, presa dalla stessa guida, per riconoscere quando evitarli.

La colonna centrale è quella che conta: nel sequenziale e nel concorrente l'ordine lo decide il tuo codice e il sistema resta prevedibile; nelle altre tre lo decide un agente, cioè un modello, e la prevedibilità va recuperata con tetti e controlli. Il pattern che molti tutorial chiamano «plan-first» corrisponde al quinto: la guida lo chiama magentic e lo descrive con un gestore che costruisce un registro dei compiti e dell'avanzamento. Sono la classificazione di una guida, non una norma: altre fonti usano altri nomi per le stesse cose.

Sequenziale e concorrente: l'ordine lo decidi tu

Il blocco costruisce due funzioni sopra agenti finti, ciascuno una regola che avvolge il testo ricevuto in una lettera. sequenziale passa l'uscita di ciascuno all'ingresso del successivo e conserva una traccia; concorrente dà lo stesso ingresso a tutti con un ThreadPoolExecutor e raccoglie i risultati con pool.map, che li restituisce nell'ordine di invio e non di fine. Non è cosmetico: aggregando nell'ordine di arrivo, la stessa richiesta darebbe sintesi diverse da un'esecuzione all'altra.

from concurrent.futures import ThreadPoolExecutor

def agente(nome, risposta):
    """Un agente finto: nessun modello, risponde con una regola fissa."""
    def esegui(testo):
        return f"{risposta}({testo})"
    esegui.nome = nome
    return esegui

# Sequenziale: l'ordine lo decide il codice, non gli agenti
def sequenziale(agenti, testo):
    traccia = []
    for a in agenti:
        testo = a(testo)
        traccia.append((a.nome, testo))
    return testo, traccia

esito, traccia = sequenziale(
    [agente("bozza", "B"), agente("revisione", "R"), agente("stile", "S")], "tema")
for nome, uscita in traccia:
    print(f"{nome:10} -> {uscita}")

# Concorrente: stessi input a tutti, risultati raccolti nell'ordine di invio
def concorrente(agenti, testo):
    with ThreadPoolExecutor(max_workers=len(agenti)) as pool:
        return dict(zip((a.nome for a in agenti), pool.map(lambda a: a(testo), agenti)))

print(concorrente([agente("fondamentali", "F"), agente("tecnica", "T"),
                   agente("sentiment", "M")], "ACME"))

# Output:
# bozza      -> B(tema)
# revisione  -> R(B(tema))
# stile      -> S(R(B(tema)))
# {'fondamentali': 'F(ACME)', 'tecnica': 'T(ACME)', 'sentiment': 'M(ACME)'}

Nella prima parte dell'uscita l'ultimo agente riceve il lavoro di tutti i precedenti: è il vantaggio (ogni passo vede il risultato completo del precedente) e il rischio (se il primo sbaglia, gli altri lavorano sull'errore, come nota la guida tra i motivi per evitare il pattern). Nella seconda gli agenti non si vedono tra loro, e la domanda da farsi è chi risolve il conflitto quando due risultati si contraddicono. Con agenti veri il tempo del concorrente tende a essere quello del più lento e non la somma, mentre i token si sommano; qui non l'ho misurato, perché non ho chiamato modelli.

Handoff e group chat: servono i tetti

Nell'handoff un agente riceve la richiesta e decide a chi passarla: in un customer support il primo smista, lo specialista risponde. Il guaio è il rimbalzo, due agenti che si rimandano la richiesta perché ciascuno la crede di competenza dell'altro. Il blocco mostra il difetto di proposito: legale rimanda a contabilita, che rimanda a legale; il tetto max_passaggi interrompe il ciclo e il risultato dice che cosa è successo (fermato) invece di restituire una risposta inventata.

def triage(richiesta, passaggi):
    """Instrada: restituisce (agente_successivo, None) oppure (None, risposta)."""
    if "fattura" in richiesta:
        return "contabilita", None
    return None, "Risposta generica dal triage"

def contabilita(richiesta, passaggi):
    if "rimborso" in richiesta:
        return "legale", None           # ha bisogno di un altro specialista
    return None, "Fattura rintracciata"

def legale(richiesta, passaggi):
    return "contabilita", None          # difetto voluto: rimanda indietro

AGENTI = {"triage": triage, "contabilita": contabilita, "legale": legale}

def handoff(richiesta, max_passaggi=4):
    corrente, catena = "triage", []
    while len(catena) < max_passaggi:
        catena.append(corrente)
        prossimo, risposta = AGENTI[corrente](richiesta, catena)
        if risposta is not None:
            return {"catena": catena, "risposta": risposta}
        corrente = prossimo
    return {"catena": catena, "risposta": None, "fermato": "tetto di passaggi"}

print(handoff("dov'e' la mia fattura?"))
print(handoff("fattura con rimborso"))

# Output:
# {'catena': ['triage', 'contabilita'], 'risposta': 'Fattura rintracciata'}
# {'catena': ['triage', 'contabilita', 'legale', 'contabilita'], 'risposta': None, 'fermato': 'tetto di passaggi'}

La prima richiesta si chiude in due passaggi. La seconda arriva al tetto con la catena triage, contabilita, legale, contabilita, e il None in risposta conta: chi chiama sa che non c'è risposta e può passare a una persona invece di mostrare un testo qualunque. Il tetto di 4 è arbitrario: scegli il tuo guardando le catene lecite più lunghe che osservi, più un margine. La stessa logica vale per la group chat, dove la guida suggerisce tre agenti o meno e, nei cicli maker-checker (uno propone, uno verifica), un numero massimo di iterazioni con un ripiego, per esempio l'escalation a una persona. I parametri equivalenti in AutoGen sono nell'articolo su AutoGen.

Piano e registro: il pattern magentic

Quando non conosci i passi in anticipo, un agente può costruire un piano e un gestore lo esegue tenendo un registro: che cosa è deciso, fatto, bloccato. Il blocco isola la parte deterministica, l'esecuzione di un piano con dipendenze. Il «pianificatore» è una costante (PIANO) perché la parte che richiede un modello non è eseguibile qui; l'esecuzione usa graphlib.TopologicalSorter, e un compito bloccato da un predecessore fallito viene saltato e annotato, non eseguito a vuoto.

from graphlib import TopologicalSorter

# Il "pianificatore" finto restituisce sempre lo stesso piano: compito -> dipendenze
PIANO = {
    "raccogli_dati": [],
    "analizza_a": ["raccogli_dati"],
    "analizza_b": ["raccogli_dati"],
    "sintesi": ["analizza_a", "analizza_b"],
}
GUASTI = {"analizza_b"}                  # il compito che qui fallisce

def esegui_piano(piano, guasti):
    registro = {nome: "da fare" for nome in piano}
    ts = TopologicalSorter(piano)
    ts.prepare()
    while ts.is_active():
        for nome in sorted(ts.get_ready()):
            bloccato = [d for d in piano[nome] if registro[d] != "fatto"]
            if bloccato:
                registro[nome] = f"saltato (dipende da {', '.join(bloccato)})"
            elif nome in guasti:
                registro[nome] = "fallito"
            else:
                registro[nome] = "fatto"
            ts.done(nome)
    return registro

for nome, stato in esegui_piano(PIANO, GUASTI).items():
    print(f"{nome:14} {stato}")

# Output:
# raccogli_dati  fatto
# analizza_a     fatto
# analizza_b     fallito
# sintesi        saltato (dipende da analizza_b)

Il registro dice che sintesi non è stata eseguita perché analizza_b è fallito, e non produce una sintesi costruita su metà dei dati. Se scegli di consegnare comunque un risultato parziale devi dichiararlo all'utente: è una decisione di prodotto. Il costo del pattern è la sua ragione d'essere: costruire e discutere un piano richiede tempo, e la guida lo sconsiglia con scadenze strette o quando il percorso è già noto.

Dove sta il controllo: hub, pari o coda

I pattern dicono come si coordina il lavoro; le architetture dicono come sono collegati gli agenti. Sono tre, e non sono una classificazione ufficiale: sono le forme che si incontrano più spesso, e si possono combinare.

Nell'hub-and-spoke un agente coordinatore parla con tutti gli altri, che non parlano tra loro: è la forma dell'orchestratore con sottoagenti descritta da Anthropic per il proprio sistema di ricerca. Tutto passa da un punto, quindi registrare, limitare e fermare è facile; per lo stesso motivo il coordinatore è un punto di guasto e un collo di bottiglia. Anthropic segnala che i suoi agenti principali aspettano i sottoagenti in modo sincrono, il che semplifica il coordinamento ma crea strozzature, e che l'asincrono porterebbe problemi di coerenza dello stato e di propagazione degli errori.

Nel peer-to-peer gli agenti si parlano direttamente. Non ho un listato per questa forma: richiederebbe scoperta e instradamento tra pari, che non ha senso simulare in trenta righe e che la guida di Microsoft nomina solo come «protocollo basato sui pari». Una fonte però c'è: nel lavoro sullo scaling l'architettura decentralizzata (gli agenti si scambiano messaggi alla pari) amplifica gli errori più della centralizzata, 7,8 contro 4,4 volte nei loro esperimenti, perché manca un punto di controllo. Non è una classifica: su alcuni benchmark la decentralizzata è risultata la migliore, e gli autori concludono che conta l'accordo tra forma della comunicazione e struttura del compito.

Nella coda di messaggi gli agenti comunicano pubblicando e leggendo da un broker: il mittente non sa chi leggerà, e il lettore può essere spento quando il messaggio arriva. Il listato è un broker in memoria di pochi metodi, utile a vedere tre proprietà e non a sostituire RabbitMQ o Kafka (che non ho avviato): un messaggio resta in coda finché qualcuno lo legge, uno che fallisce rientra con un contatore, e dopo due tentativi finisce in una «coda dei morti» (dead letter queue) invece di bloccare gli altri.

from collections import defaultdict, deque

class Broker:
    """Un broker minimo in memoria: code per argomento, tentativi, coda dei morti."""
    def __init__(self, max_tentativi=2):
        self.code = defaultdict(deque)
        self.morti = []
        self.max_tentativi = max_tentativi

    def pubblica(self, argomento, messaggio):
        self.code[argomento].append({"corpo": messaggio, "tentativi": 0})

    def consuma(self, argomento, gestore):
        """Elabora tutto cio' che e' in coda; un messaggio che fallisce rientra."""
        coda = self.code[argomento]
        for _ in range(len(coda)):
            msg = coda.popleft()
            try:
                gestore(msg["corpo"])
            except ValueError as errore:
                msg["tentativi"] += 1
                if msg["tentativi"] >= self.max_tentativi:
                    self.morti.append((msg["corpo"], str(errore)))
                else:
                    coda.append(msg)

def analista(corpo):
    if corpo == "malformato":
        raise ValueError("JSON non valido")
    print("analisi di:", corpo)

b = Broker()
b.pubblica("analisi", "ordine 17")      # l'analista e' "spento": il messaggio aspetta
b.pubblica("analisi", "malformato")
print("in coda prima del consumo:", len(b.code["analisi"]))
for giro in (1, 2):
    b.consuma("analisi", analista)
    print(f"giro {giro}: in coda {len(b.code['analisi'])}, morti {len(b.morti)}")
print("coda dei morti:", b.morti)

# Output:
# in coda prima del consumo: 2
# analisi di: ordine 17
# giro 1: in coda 1, morti 0
# giro 2: in coda 0, morti 1
# coda dei morti: [('malformato', 'JSON non valido')]

Il messaggio ordine 17 era in coda prima che l'analista esistesse e viene elaborato al primo giro; quello malformato fallisce, rientra, fallisce ancora e finisce in morti con il motivo. Senza il limite di tentativi girerebbe all'infinito: il rimbalzo dell'handoff in versione infrastrutturale. Un broker vero aggiunge ciò che qui manca, persistenza su disco e conferma di ricezione; non risolve invece l'ordine: due agenti che leggono dalla stessa coda non vedono i messaggi nello stesso istante, e lo stato condiviso torna un problema tuo.

La tabella non attribuisce numeri (quanti agenti, quanti millisecondi) perché non ho né una fonte né una misura che reggano. La regola che ne ricavo: parti dall'hub, la forma più facile da fermare e osservare, e passa alla coda quando hai un motivo preciso, per esempio componenti da spegnere o scalare separatamente. Il peer-to-peer ha senso solo se l'assenza di un coordinatore è un requisito, non una comodità.

Lo stato: un registro di eventi e le saghe

Con più agenti che scrivono, lo stato condiviso è dove i bug si nascondono. Un approccio è non salvare lo stato corrente ma gli eventi che l'hanno prodotto: è l'Event Sourcing, con eventi registrati in un archivio in cui si aggiunge soltanto, e lo stato ricavato rigiocandoli. Un secondo pattern serve quando un'operazione tocca più agenti e deve riuscire per intero o per niente: la saga spezza l'operazione in transazioni locali, ciascuna con la propria compensazione da eseguire se un passo successivo fallisce. Entrambi sono nel catalogo dei pattern di Azure Architecture Center.

Il blocco li unisce in un agente di viaggio: prenotare volo, albergo e auto, con l'auto non disponibile. saga esegue i passi in ordine e, al primo fallimento, le compensazioni in ordine inverso, solo per i passi già riusciti. Ogni azione scrive un evento sul registro; stato_a(n) ricostruisce le prenotazioni attive rigiocando i primi n eventi.

registro = []                       # registro eventi: solo aggiunte, mai modifiche

def evento(tipo, **dati):
    registro.append({"n": len(registro) + 1, "tipo": tipo, **dati})

def prenota(risorsa, fallisce=False):
    if fallisce:
        raise RuntimeError(f"{risorsa} non disponibile")
    evento("prenotato", risorsa=risorsa)

def annulla(risorsa):
    evento("annullato", risorsa=risorsa)

# Ogni passo e' (azione, compensazione): la compensazione non e' un "annulla" magico,
# e' un'azione scritta apposta per disfare quel passo.
PASSI = [("volo", False), ("albergo", False), ("auto", True)]

def saga(passi):
    fatti = []
    for risorsa, fallisce in passi:
        try:
            prenota(risorsa, fallisce)
            fatti.append(risorsa)
        except RuntimeError as errore:
            evento("fallito", motivo=str(errore))
            for gia in reversed(fatti):      # compensazioni in ordine inverso
                annulla(gia)
            return False
    return True

print("saga completata:", saga(PASSI))
for e in registro:
    print(e)

# Lo stato corrente non e' salvato: si ricostruisce rigiocando il registro
def stato_a(n):
    attive = set()
    for e in registro[:n]:
        if e["tipo"] == "prenotato":
            attive.add(e["risorsa"])
        elif e["tipo"] == "annullato":
            attive.discard(e["risorsa"])
    return sorted(attive)

print("dopo l'evento 2:", stato_a(2))
print("dopo l'evento 5:", stato_a(5))

# Output:
# saga completata: False
# {'n': 1, 'tipo': 'prenotato', 'risorsa': 'volo'}
# {'n': 2, 'tipo': 'prenotato', 'risorsa': 'albergo'}
# {'n': 3, 'tipo': 'fallito', 'motivo': 'auto non disponibile'}
# {'n': 4, 'tipo': 'annullato', 'risorsa': 'albergo'}
# {'n': 5, 'tipo': 'annullato', 'risorsa': 'volo'}
# dopo l'evento 2: ['albergo', 'volo']
# dopo l'evento 5: []

Il registro racconta tutto: due prenotazioni, un fallimento, due annullamenti in ordine inverso. Rigiocato fino all'evento 2 dà albergo e volo, fino al 5 nulla: guardare lo stato in un istante passato è ciò che si compra con l'Event Sourcing, e si paga rigiocando gli eventi (per questo il catalogo cita le viste materializzate). L'esempio nasconde un punto: le compensazioni reali possono fallire, e il catalogo avverte che i passi dopo il punto di non ritorno (il «pivot») devono poter essere ripetuti in sicurezza, cioè essere idempotenti. Inoltre non tutto si annulla: un messaggio già inviato a un cliente si compensa con un altro messaggio.

Quando qualcosa si rompe: ritentativi, interruttore, ripiego

I fallimenti attesi di un agente sono quelli di ogni chiamata remota: il modello non risponde, lo strumento va in timeout, l'uscita non ha il formato atteso. Il catalogo di Microsoft tratta due difese che si combinano. Il retry ritenta aspettando di più a ogni tentativo, sull'ipotesi che l'errore sia transitorio; la guida avverte che una politica aggressiva può peggiorare un servizio già al limite, e che ritentare è sicuro solo se l'operazione è idempotente. L'interruttore (circuit breaker) fa l'opposto: dopo troppi errori smette di chiamare per un periodo, poi lascia passare una prova. Ha tre stati: chiuso (le chiamate passano), aperto (falliscono subito), semiaperto (una prova: se riesce si richiude, se fallisce si riapre).

Il blocco implementa entrambi su un orologio finto che avanza solo quando glielo dici, così l'uscita non dipende dalla macchina. Interruttore.chiama è la macchina a stati; ritenta aspetta un tempo scelto a caso tra 0 e il minimo tra il tetto e base * 2 ** n (jitter pieno, che evita che tanti client ritentino nello stesso istante), con un seme fisso. Una riga conta più delle altre, except InterruttoreAperto: raise: ritentare contro un circuito aperto vanifica l'interruttore, e la pagina di Microsoft dice che il retry deve fermarsi quando l'interruttore segnala un guasto non transitorio.

import random

class Orologio:
    """Tempo finto: avanza solo quando lo dici tu, cosi' l'uscita e' sempre la stessa."""
    def __init__(self):
        self.t = 0.0
    def dormi(self, secondi):
        self.t += secondi

class InterruttoreAperto(Exception):
    pass

class Interruttore:
    def __init__(self, orologio, soglia=3, pausa=30.0):
        self.o, self.soglia, self.pausa = orologio, soglia, pausa
        self.stato, self.errori, self.aperto_da = "chiuso", 0, None

    def chiama(self, funzione):
        if self.stato == "aperto":
            if self.o.t - self.aperto_da < self.pausa:
                raise InterruttoreAperto("circuito aperto: chiamata non tentata")
            self.stato = "semiaperto"        # lascia passare UNA prova
        try:
            risultato = funzione()
        except ConnectionError:
            self.errori += 1
            if self.stato == "semiaperto" or self.errori >= self.soglia:
                self.stato, self.aperto_da = "aperto", self.o.t
            raise
        self.stato, self.errori = "chiuso", 0
        return risultato

def ritenta(funzione, orologio, attese, tentativi=4, base=1.0, tetto=8.0, seme=7):
    rng = random.Random(seme)
    for n in range(tentativi):
        try:
            return funzione()
        except InterruttoreAperto:
            raise                              # non si ritenta contro un circuito aperto
        except ConnectionError:
            if n == tentativi - 1:
                raise
            attesa = rng.uniform(0, min(tetto, base * 2 ** n))   # jitter "pieno"
            attese.append(round(attesa, 2))
            orologio.dormi(attesa)

def agente_giu():
    raise ConnectionError("modello non raggiungibile")

o = Orologio()
ib = Interruttore(o)
attese = []
try:
    ritenta(lambda: ib.chiama(agente_giu), o, attese)
except (ConnectionError, InterruttoreAperto) as e:
    print("ritentativo finito:", type(e).__name__, "| attese:", attese, "| stato:", ib.stato)
try:
    ib.chiama(agente_giu)
except InterruttoreAperto as e:
    print("subito dopo:", e)
o.dormi(31)
print("dopo 31 s, una prova riuscita:", ib.chiama(lambda: "ok"), "| stato:", ib.stato)

# Output:
# ritentativo finito: InterruttoreAperto | attese: [0.32, 0.3, 2.6] | stato: aperto
# subito dopo: circuito aperto: chiamata non tentata
# dopo 31 s, una prova riuscita: ok | stato: chiuso

Il ritentativo ha atteso tre volte (0,32, 0,3 e 2,6 secondi nel tempo finto); al terzo errore il circuito si è aperto e il quarto tentativo non è partito. Subito dopo una chiamata è respinta senza tentare; dopo 31 secondi, oltre la pausa di 30, passa una prova, riesce e il circuito si richiude. Le soglie (3 errori, 30 secondi, 4 tentativi, tetto di 8 secondi) sono esempi: dipendono dal tempo di recupero del servizio e dal costo di un errore, e vanno misurate sul tuo. Con un modello a pagamento, un tetto sul numero totale di chiamate per richiesta conta quanto il resto.

L'ultima difesa è il ripiego (graceful degradation): quando un componente non risponde, il sistema restituisce qualcosa di meno buono ma dichiarato, per esempio un agente più semplice o a regole al posto del primario, un risultato parziale con un avviso, una risposta in cache etichettata come tale. Anthropic descrive la variante per i sistemi lunghi: punti di ripresa (checkpoint) per riprendere da dove l'agente era arrivato, perché i riavvii costano, e dire all'agente che uno strumento non funziona lasciando che si adatti, che secondo loro funziona sorprendentemente bene. Il confine da non attraversare: non nascondere all'utente che la risposta è degradata.

Osservare: tracce, costo, soglie tue

Un sistema multi-agente non deterministico si debugga solo se registri che cosa ha fatto. Anthropic scrive che gli utenti segnalavano agenti che «non trovavano informazioni ovvie» e che solo la traccia completa in produzione ha permesso di capire se fosse una query sbagliata, una fonte scelta male o uno strumento rotto. L'idea è quella dei sistemi tradizionali: ogni richiesta ha un identificativo, ogni passo è un tratto (span) con un padre, e l'albero mostra dove tempo ed errori si accumulano.

Il blocco costruisce un tracciatore minimo con un gestore di contesto: ogni tratto annota padre, durata dichiarata e attributi, e se al suo interno sale un'eccezione registra il tipo d'errore e la lascia salire. L'esempio ha un pianificatore e un agente di ricerca il cui strumento va in timeout; i token sono numeri inventati per l'esempio, non misure.

from contextlib import contextmanager

class Tracciatore:
    """Tiene l'elenco dei tratti (span). Il tempo e' finto: ogni evento costa quanto dichiari."""
    def __init__(self):
        self.tratti, self.pila, self.ora = [], [], 0.0

    @contextmanager
    def tratto(self, nome, durata, **attributi):
        padre = self.pila[-1] if self.pila else None
        t = {"id": len(self.tratti) + 1, "padre": padre, "nome": nome,
             "inizio": self.ora, "attributi": attributi}
        self.tratti.append(t)
        self.pila.append(t["id"])
        try:
            yield t
        except Exception as e:
            t["attributi"]["errore"] = type(e).__name__
            raise
        finally:
            self.ora += durata
            t["fine"] = self.ora
            self.pila.pop()

tr = Tracciatore()
try:
    with tr.tratto("richiesta", 0.1, traccia="T-1"):
        with tr.tratto("agente:pianificatore", 1.5, token=420):
            pass
        with tr.tratto("agente:ricerca", 4.0, token=1800):
            with tr.tratto("strumento:cerca", 2.0):
                raise TimeoutError("lo strumento non risponde")
except TimeoutError:
    pass

figli = {t["id"]: t["padre"] for t in tr.tratti}
def profondita(i):
    return 0 if figli[i] is None else 1 + profondita(figli[i])
for t in tr.tratti:
    durata = round(t["fine"] - t["inizio"], 1)
    print("  " * profondita(t["id"]) + f'{t["nome"]} {durata}s {t["attributi"]}')
print("token totali:", sum(t["attributi"].get("token", 0) for t in tr.tratti))

# Output:
# richiesta 7.6s {'traccia': 'T-1', 'errore': 'TimeoutError'}
#   agente:pianificatore 1.5s {'token': 420}
#   agente:ricerca 6.0s {'token': 1800, 'errore': 'TimeoutError'}
#     strumento:cerca 2.0s {'errore': 'TimeoutError'}
# token totali: 2220

L'albero indica subito dove è nato l'errore (strumento:cerca) e come risale: chi guarda solo il livello esterno vede «richiesta fallita» senza sapere perché. La durata di un tratto include quella dei figli, quindi per sapere dove si è speso il tempo va sottratta. In un sistema vero non scrivi un tracciatore tuo: usi OpenTelemetry. Le sue convenzioni per l'AI generativa sono state spostate nel repository semantic-conventions-genai; non le ho confrontate con il listato, quindi consulta quel repository per i nomi degli attributi prima di adottarli.

Che cosa registrare in un tratto dipende dal sistema, ma alcune grandezze servono quasi sempre:

• l'identificativo della richiesta, uguale in tutti i tratti dello stesso lavoro;
• l'agente o lo strumento, e la versione del prompt o del modello usato;
• i token in ingresso e in uscita, per ricostruire il costo di ogni richiesta;
• l'esito (ok, errore, saltato) e il tipo di errore;
• il numero di passaggi e di tentativi, per vedere i rimbalzi e i ritentativi.

Da queste grandezze si ricavano le metriche che contano: latenza per agente, tasso d'errore, token per richiesta, richieste che arrivano al tetto, aperture dell'interruttore. Le soglie d'allarme non si copiano da un articolo: si fissano osservando un periodo di funzionamento normale del tuo sistema. Un allarme su «richieste arrivate al tetto di passaggi» o su «aperture dell'interruttore» richiede pochi dati e segnala subito che qualcosa non va.

Dove andare da qui

Come si misura che l'insieme funzioni, e non soltanto che giri, è in Testing e Evaluation; la memoria che gli agenti si passano è in Memory Systems; il rischio di agenti che si fidano l'uno dell'altro in Security e Safety; come si progettano gli strumenti che chiamano in Tool Calling Avanzato.

Domande frequenti

Che cosa sai fare adesso

Sai decidere se un problema richiede più agenti o ne basta uno con strumenti, e scegliere uno dei cinque pattern in base a chi decide il passo successivo. Sai mettere un tetto a un handoff e a una group chat, eseguire un piano con dipendenze lasciando un registro di ciò che è fallito o saltato, collegare gli agenti con un hub o con una coda sapendo che cosa ciascuna forma ti costa, ricostruire uno stato da un registro di eventi e compensare una saga. Sai combinare ritentativi e interruttore senza farli litigare, registrare tracce da cui si capisce dove è nato un errore, e fissare le soglie d'allarme sui tuoi dati e non su una tabella.