83 - asyncio: l'infrastruttura su cui gira un agente

asyncio serve a una cosa sola: non stare fermi mentre si aspetta. Un agente passa quasi tutto il suo tempo in attesa di risposte di rete — il modello, le API dei tool, il database — e durante quell'attesa il processore non fa nulla. Con il codice sincrono quell'attesa blocca tutto; con una coroutine il controllo torna all'event loop, che intanto fa avanzare le altre attese. Non è parallelismo (gira un solo thread, un'istruzione alla volta): è sovrapposizione delle attese, e per un agente è esattamente la risorsa che conta.

L'analogia: il cuoco e i quattro fornelli

Un cuoco sincrono mette la pasta, resta davanti alla pentola per nove minuti, poi accende il sugo, aspetta, poi scalda il pane. Trenta minuti (tempi d'esempio, per fissare l'idea). Un cuoco asincrono mette la pasta, e mentre bolle accende il sugo e infila il pane nel forno; poi gira fra i fornelli man mano che qualcosa è pronto. Undici minuti. È sempre un cuoco solo — non ha imparato a tagliare due cipolle insieme (quello sarebbe parallelismo, e richiederebbe un secondo cuoco, cioè un altro processo). Ha solo smesso di fissare le pentole.

Da questa analogia si deduce già il limite: se il lavoro è tagliare cipolle e non aspettare che bolla, asyncio non serve a niente. Lavoro I/O bound (rete, disco, DB) → asyncio. Lavoro CPU bound (moltiplicare matrici, parsare 2 GB di JSON) → processi separati.

Tre parole che vengono confuse: concorrenza, parallelismo, thread

Prima del codice conviene sistemare il vocabolario, perché la maggior parte degli errori con asyncio nasce da una confusione terminologica e non da un problema tecnico.

Concorrenza significa che più attività sono in corso nello stesso intervallo di tempo. Non dice nulla su quante ne stiano avanzando in questo preciso istante. Il cuoco dell'analogia è concorrente: pasta, sugo e pane sono tutti «in corso», anche se lui in ogni singolo momento sta facendo una cosa sola.

Parallelismo significa che più attività avanzano fisicamente insieme, su unità di calcolo distinte. Due cuochi. Richiede più core e, in Python, più processi.

Thread sono un meccanismo del sistema operativo per ottenere concorrenza (e su altri linguaggi anche parallelismo), in cui a decidere chi gira è il sistema, non il tuo codice. È la differenza chiave con asyncio: i thread sono preemptivi — possono essere interrotti in qualunque punto, anche a metà di un'operazione che tu consideravi indivisibile — mentre le coroutine sono cooperative e cedono il controllo solo nei punti di await, che sono visibili nel codice.

Questa differenza ha una conseguenza pratica enorme e spesso ignorata: con i thread non puoi sapere, leggendo una funzione, dove potrebbe essere interrotta; con le coroutine puoi, perché i punti di interruzione sono scritti. Il codice asincrono è più facile da ragionare sulle sezioni critiche, e questo compensa la difficoltà maggiore nel debug.

Resta il terzo protagonista, che in Python va nominato: il Global Interpreter Lock, il lucchetto che storicamente ha impedito a due thread di eseguire bytecode Python contemporaneamente nello stesso processo. È la ragione per cui, in Python, i thread non danno parallelismo sul calcolo puro — danno concorrenza sull'attesa, perché il lucchetto viene rilasciato durante le operazioni di I/O. Il panorama si sta muovendo: Python 3.13 ha introdotto build sperimentali senza GIL e la PEP 779 ne definisce i criteri per lo stato di «supportato» (consultata il 2026-11-12). Sulla macchina su cui ho eseguito gli esempi di questa lezione (Python 3.14.4) il GIL risulta ancora attivo (sys._is_gil_enabled() restituisce True), e comunque la regola pratica per scrivere un agente oggi non cambia: per l'attesa si usa asyncio, per il calcolo si usano processi separati.

Da cui la tabella mentale, che vale la pena fissare. Ricorda che ogni voce risponde alla stessa domanda: che cosa sta facendo il programma mentre «aspetta»?

• Molte attese di rete (chiamate al modello, API, query): asyncio. È il caso di un agente, quasi al 100%.
• Calcolo pesante (embedding su CPU, parsing di file enormi, trasformazioni numeriche): processi separati, con ProcessPoolExecutor o un servizio a parte.
• Librerie sincrone che non puoi cambiare (un driver di database legacy, un SDK senza versione async): asyncio.to_thread, che sposta la chiamata bloccante su un thread e restituisce il controllo all'event loop.
• Un misto, che è il caso reale: codice async per il flusso, to_thread per le isole sincrone, un processo separato per il calcolo.

Se l'attesa è di rete, asyncio; se è calcolo, un altro processo; se hai una libreria che non sai cambiare, la isoli in un thread con asyncio.to_thread senza riscrivere nulla. Il caso «misto» è il più frequente nei progetti veri e vale la pena dichiararlo, perché nessuna delle tre soluzioni da sola copre un servizio che chiama un modello in rete, legge da un driver legacy e calcola embedding.

Dove un agente aspetta davvero: anatomia di una latenza

Vale la pena guardare da dove viene il tempo in una run tipica, perché determina dove conviene intervenire e dove è inutile. Prendiamo un agente che fa tre giri e chiama quattro tool in totale, su un modello servito in rete. I tempi qui sotto sono un esempio costruito per ragionare, non una misura: i tuoi saranno diversi, e sono quelli che devi misurare.

Giro 1: chiamata al modello            1,8 s   (attesa di rete)
        esecuzione di 2 tool            0,4 s   (attesa: 2 query DB, in parallelo)
Giro 2: chiamata al modello            2,4 s   (attesa; il prompt è più lungo)
        esecuzione di 1 tool            1,1 s   (attesa: una API esterna lenta)
Giro 3: chiamata al modello            1,3 s   (attesa; risposta finale, breve)
        il tuo codice (validazione,
        serializzazione, logging)      0,02 s   (calcolo)
                                      -------
                                       7,02 s   di cui 7,00 di attesa

Il rapporto è quello che conta: oltre il 99% del tempo è attesa. Da qui tre conclusioni operative che non sono ovvie.

Primo: ottimizzare il tuo codice non serve a niente. Quei 20 millisecondi di validazione e logging sono irrilevanti. Il tempo sta nelle chiamate di rete, e l'unico modo di ridurlo è farne meno (meno giri, tool che restituiscono meno contesto) o farle insieme quando sono indipendenti.

Secondo: il ciclo dell'agente non si parallelizza. Il giro 2 ha bisogno della risposta del giro 1: non c'è niente da sovrapporre. I 5,5 secondi di chiamate al modello sono un limite inferiore per questa run, e si abbassano solo riducendo il numero di giri — che è un problema di progetto dell'agente, non di concorrenza.

Terzo: ciò che si parallelizza sono i tool dentro lo stesso giro e le run fra loro. Nel giro 1 i due tool vanno insieme: 0,4 secondi invece di 0,7. E se devi processare cinquanta documenti, le cinquanta run sono indipendenti: con un semaforo a dieci il tempo totale è circa un decimo di quello sequenziale (cinquanta run in cinque ondate da dieci); nell'esempio eseguito più sotto, 0,5 secondi contro 5,0.

C'è anche una quarta conclusione, che riguarda il server più che il singolo utente, e che è la ragione principale per cui tutto questo modulo è scritto in codice asincrono. Con codice sincrono, un processo che serve un utente resta bloccato per sette secondi, durante i quali non può fare nulla per nessun altro: per servire dieci utenti contemporaneamente ti servono dieci processi, con dieci volte la memoria. Con codice asincrono lo stesso processo, in quei sette secondi di attesa, fa avanzare le richieste di tutti gli altri. Il vantaggio non l'ho misurato su un server reale; l'ordine di grandezza dipende dal rapporto fra attesa e calcolo, e in una run come quella sopra, in cui l'attesa supera il 99%, è grande.

La differenza, disegnata su una linea del tempo

Tre chiamate da un secondo l'una. In alto il codice sincrono, in mezzo l'errore più comune, in basso la versione che funziona. Le barre grigie sono attesa pura: tempo in cui il processore non fa nulla e potrebbe fare altro.

Il meccanismo in quattro oggetti

• Coroutine. Una funzione dichiarata async def. Chiamarla non la esegue: restituisce un oggetto coroutine, inerte, come una ricetta scritta su un foglio. Dimenticarsi di attenderla è l'errore numero uno di chi comincia (Python in genere ti avverte con un RuntimeWarning: coroutine was never awaited).
• await. Il punto in cui la coroutine dice «sto per aspettare, intanto prendi tu il controllo». È l'unico posto dove l'event loop può passare ad altro. Nessun await dentro una funzione async significa nessuna concorrenza.
• Event loop. Lo scheduler cooperativo: tiene la lista delle attese e, quando il sistema operativo gli dice che un socket è pronto, riprende la coroutine corrispondente. asyncio.run() lo crea, esegue, chiude.
• Task. Una coroutine pianificata. asyncio.create_task(coro()) la mette in corsa subito; da quel momento procede anche se tu stai facendo altro. Questa è la differenza fondamentale fra «ci sono due coroutine» e «due cose stanno davvero avanzando insieme».

I quattro oggetti si leggono in sequenza: una coroutine è la ricetta, un task è la ricetta messa sul fuoco, l'event loop è il cuoco che gira fra i fornelli, e l'await è il momento in cui il cuoco può spostarsi. Se uno dei quattro manca, la concorrenza sparisce senza avvisare, ed è la ragione per cui gli errori di questa lezione producono codice che funziona, solo lentamente.

La misura che chiarisce tutto

Simuliamo tre chiamate a un tool che impiega un secondo ciascuna (asyncio.sleep sta al posto dell'attesa di rete). Tre versioni, stesso lavoro. Eseguito con Python 3.14.4, due volte, con la stessa uscita:

import asyncio, time

async def tool(nome: str, secondi: float) -> str:
    await asyncio.sleep(secondi)          # qui c'è l'attesa di rete, finta
    return f"{nome} ok"

async def sequenziale():
    return [await tool(f"t{i}", 1) for i in range(3)]

async def concorrente_sbagliata():
    # ERRORE: create_task c'è, ma l'await è subito dopo -> resta sequenziale
    out = []
    for i in range(3):
        t = asyncio.create_task(tool(f"t{i}", 1))
        out.append(await t)
    return out

async def concorrente():
    tasks = [asyncio.create_task(tool(f"t{i}", 1)) for i in range(3)]
    return await asyncio.gather(*tasks)

async def main():
    for f in (sequenziale, concorrente_sbagliata, concorrente):
        t0 = time.perf_counter()
        await f()
        print(f"{f.__name__:24s} {time.perf_counter() - t0:.1f}s")

asyncio.run(main())

# sequenziale              3.0s
# concorrente_sbagliata    3.0s
# concorrente              1.0s

La versione centrale è il bug più comune in assoluto: crea i task e li attende uno per uno, quindi non guadagna niente. La regola operativa: crea tutti i task, poi attendi. Se fra la creazione e l'attesa c'è un await dentro un ciclo, hai scritto codice sincrono con sintassi asincrona. Su altre cinque esecuzioni, i tempi stampati con tre decimali sono stati 3,003-3,010 s per le prime due versioni e 1,001 s per la terza: la differenza fra «sbagliata» e «sequenziale» è nulla, quella con la terza è di due secondi.

Dove serve davvero in un agente

Il ciclo dell'agente è sequenziale per natura: il giro 2 dipende dalla risposta del giro 1, non c'è niente da sovrapporre. Le occasioni sono altre tre, e sono quelle che contano. La prima: chiamate di tool parallele nello stesso giro, perché il modello può chiedere tre tool in un'unica risposta, ed eseguirli insieme porta il tempo del giro a quello del tool più lento. La seconda: più agenti, o lo stesso agente su più input, perché venti documenti da classificare sono venti run indipendenti. La terza: servire più utenti con un processo, il motivo per cui esistono le librerie HTTP async: con codice sincrono un worker bloccato in attesa del modello non può servire nessun altro.

Ecco la prima occasione, con due tool finti: uno risponde, l'altro solleva un errore. Le chiamate hanno la forma {"name": ..., "args": ...}, una semplificazione di quella dei messaggi della lezione precedente. Il blocco non si lancia da solo: lo esegue il main che trovi dopo gli esempi di concorrenza controllata, insieme alle altre prove.

import asyncio, time

async def esegui_chiamate(tool: dict, chiamate: list) -> list:
    tasks = [asyncio.create_task(tool[c["name"]](**c["args"])) for c in chiamate]
    esiti = await asyncio.gather(*tasks, return_exceptions=True)
    fuori = []
    for c, e in zip(chiamate, esiti):
        if isinstance(e, Exception):
            fuori.append({"tool": c["name"], "errore": f"{type(e).__name__}: {e}"})
        else:
            fuori.append({"tool": c["name"], "risultato": e})
    return fuori

async def giacenza(sku):
    await asyncio.sleep(0.2)
    return {"sku": sku, "giacenza": 4}

async def in_arrivo(sku):
    await asyncio.sleep(0.3)
    raise KeyError(sku)                      # un tool che fallisce

async def prova_chiamate():
    t0 = time.perf_counter()
    out = await esegui_chiamate({"giacenza": giacenza, "in_arrivo": in_arrivo},
                                [{"name": "giacenza", "args": {"sku": "A1"}},
                                 {"name": "in_arrivo", "args": {"sku": "A1"}}])
    print(out)
    print(f"durata {time.perf_counter() - t0:.1f}s (sequenziale: 0.5s)")

# [{'tool': 'giacenza', 'risultato': {'sku': 'A1', 'giacenza': 4}}, {'tool': 'in_arrivo', 'errore': "KeyError: 'A1'"}]
# durata 0.3s (sequenziale: 0.5s)

Il dettaglio che fa la differenza è return_exceptions=True. Senza, la prima eccezione fa fallire l'intero gather e i risultati dei tool andati bene li butti via — in un agente, dove un tool in errore è informazione utile da restituire al modello, è esattamente il comportamento che non vuoi. L'uscita lo conferma: il tool riuscito consegna il risultato, quello fallito consegna un errore leggibile, e la durata è quella del più lento (0,3 secondi) invece della somma (0,5).

Concorrenza controllata: semaforo, timeout, cancellazione

«Fai tutto insieme» funziona con tre chiamate e ti fa bannare con trecento. Tre strumenti, tutti in libreria standard. Gli esempi, insieme a quello delle chiamate parallele, sono sezioni di un unico file che ho eseguito due volte (Python 3.14.4, uscita identica) e il main in fondo li lancia in sequenza. Ogni blocco ha il suo async def prova_... che lo esercita; l'uscita complessiva è riportata dopo il main.

Semaforo — un tetto al numero di richieste contemporanee. Il semaforo è un contatore di posti: ogni async with sem ne occupa uno, e oltre il tetto la coroutine aspetta a quella riga. Nella prova lancio 50 lavori da 0,1 secondi con tetto 10 e conto quanti sono davvero in volo nello stesso istante.

import asyncio, time

async def mappa_limitata(coro_f, elementi, max_contemporanee=5):
    sem = asyncio.Semaphore(max_contemporanee)
    async def con_limite(x):
        async with sem:              # oltre il tetto, qui si aspetta
            return await coro_f(x)
    return await asyncio.gather(*(con_limite(x) for x in elementi),
                                return_exceptions=True)

in_volo = 0
massimo = 0

async def lavoro(x):
    global in_volo, massimo
    in_volo += 1
    massimo = max(massimo, in_volo)
    try:
        await asyncio.sleep(0.1)
        return x
    finally:
        in_volo -= 1

async def prova_semaforo():
    t0 = time.perf_counter()
    out = await mappa_limitata(lavoro, range(50), max_contemporanee=10)
    print(len(out), "risultati; massimo in volo:", massimo,
          f"; durata {time.perf_counter() - t0:.1f}s (sequenziale: 5.0s)")

Timeout — nessuna attesa è infinita. Da Python 3.11 esiste asyncio.timeout, un gestore di contesto più leggibile di wait_for. Qui TimeoutError è quella built-in: da 3.11 asyncio.TimeoutError ne è un alias. La funzione trasforma la scadenza in un valore d'errore leggibile dal modello, invece di un'eccezione che uccide la run.

async def con_scadenza(coro, secondi=20):
    try:
        async with asyncio.timeout(secondi):
            return await coro
    except TimeoutError:
        return {"errore": f"timeout dopo {secondi}s"}

async def prova_timeout():
    print(await con_scadenza(asyncio.sleep(5, "lento"), secondi=0.2))
    print(await con_scadenza(asyncio.sleep(0.05, "veloce"), secondi=0.2))

Cancellazione — quando un timeout scatta, la coroutine riceve una CancelledError nel punto in cui era in attesa. Due regole non negoziabili: non catturare CancelledError per ignorarla (se la catturi per pulire, rilanciala), e liberare le risorse in un finally. Per provarla serve un client che non risponde: ClientFinto dorme cinque secondi e io cancello il task dopo 0,2.

class ClientFinto:
    async def get(self, url, timeout):
        await asyncio.sleep(5)

async def chiamata_tracciata(client, url):
    inizio = time.perf_counter()
    try:
        r = await client.get(url, timeout=10)
        return r.json()
    except asyncio.CancelledError:
        print("annullata, pulisco")
        raise                        # SEMPRE: altrimenti rompi la cancellazione
    finally:
        print(f"durata {time.perf_counter() - inizio:.1f}s")

async def prova_cancellazione():
    t = asyncio.create_task(chiamata_tracciata(ClientFinto(), "http://x"))
    await asyncio.sleep(0.2)
    t.cancel()
    try:
        await t
    except asyncio.CancelledError:
        print("il task risulta cancellato:", t.cancelled())

Da Python 3.11 c'è anche asyncio.TaskGroup, che è la forma consigliata quando i task sono logicamente un gruppo: se uno solleva, gli altri vengono cancellati e il blocco async with non si chiude prima che siano tutti terminati. È più sicuro di gather contro i task orfani; gather resta comodo quando vuoi gli errori come valori.

async def scarica(client, u, risultati):
    await asyncio.sleep(0.1)
    risultati[u] = f"contenuto di {u}"

async def gruppo(urls, client):
    risultati = {}
    async with asyncio.TaskGroup() as tg:          # Python 3.11+
        for u in urls:
            tg.create_task(scarica(client, u, risultati))
    return risultati      # qui tutti i task sono finiti, garantito

async def prova_gruppo():
    print(await gruppo(["a", "b", "c"], client=None))

Il main che esegue le cinque prove in sequenza, stampando il nome di ciascuna, e l'uscita complessiva che ne risulta. È l'unico punto in cui parte l'event loop: tutto il resto sono definizioni, ed è così che un file di prove asincrone resta leggibile. Concatena i blocchi precedenti nell'ordine in cui compaiono e aggiungi questo in fondo.

async def main():
    for p in (prova_chiamate, prova_semaforo, prova_timeout, prova_cancellazione, prova_gruppo):
        print("##", p.__name__)
        await p()

asyncio.run(main())

# ## prova_chiamate
# [{'tool': 'giacenza', 'risultato': {'sku': 'A1', 'giacenza': 4}}, {'tool': 'in_arrivo', 'errore': "KeyError: 'A1'"}]
# durata 0.3s (sequenziale: 0.5s)
# ## prova_semaforo
# 50 risultati; massimo in volo: 10 ; durata 0.5s (sequenziale: 5.0s)
# ## prova_timeout
# {'errore': 'timeout dopo 0.2s'}
# veloce
# ## prova_cancellazione
# annullata, pulisco
# durata 0.2s
# il task risulta cancellato: True
# ## prova_gruppo
# {'a': 'contenuto di a', 'b': 'contenuto di b', 'c': 'contenuto di c'}

Leggi l'uscita come una verifica per ogni strumento. Il semaforo: cinquanta lavori, mai più di dieci insieme, in 0,5 secondi invece di 5 (cinque ondate da 0,1). Il timeout: la chiamata lenta restituisce un dizionario d'errore, quella veloce passa. La cancellazione: la riga «annullata, pulisco» compare perché la CancelledError viene catturata e rilanciata, il finally stampa comunque la durata, e il task risulta davvero cancellato. Il TaskGroup consegna il dizionario completo perché il blocco non si chiude finché i tre task non hanno finito. Nel codice originale scarica non era definita: qui lo è, ed è il motivo per cui ora il blocco gira.

Cosa succede fra due await (e perché conta)

L'event loop è cooperativo: cambia coroutine solo nei punti di await. Fra un await e il successivo il tuo codice gira senza interruzioni, come se fosse solo. Questa proprietà ha due conseguenze opposte che vanno capite insieme.

La prima è buona: fra due await non serve nessun lock. Un incremento di contatore, una scrittura in un dizionario, una lettura-modifica-scrittura senza await in mezzo: sono atomici per costruzione, perché nessun'altra coroutine può intromettersi. Un esempio che lo mostra è già quello del contatore in_volo nel blocco del semaforo: incrementa e decrementa senza alcun lock, e il massimo osservato (10) è esatto.

La seconda è cattiva: se metti un await dentro una sequenza che deve essere coerente, hai aperto una finestra in cui un'altra coroutine può cambiarti lo stato sotto i piedi. Due addebiti da 30 sul saldo 100 devono lasciare 40. Nella versione rotta, la lettura del saldo avviene prima dell'await dell'antifrode e la scrittura dopo; i due addebiti leggono entrambi 100 e scrivono entrambi 70. verifica_antifrode è qui un finto controllo che dorme 0,05 secondi.

import asyncio

async def verifica_antifrode(importo):
    await asyncio.sleep(0.05)               # finto controllo esterno

saldo = {"valore": 100}

# ROTTO: fra la lettura e la scrittura c'è un await
async def addebita_rotto(importo):
    attuale = saldo["valore"]
    await verifica_antifrode(importo)      # qui un'altra coroutine può addebitare
    saldo["valore"] = attuale - importo    # scrive sulla base di un valore vecchio

# CORRETTO: la sezione critica è protetta
lucchetto = asyncio.Lock()

async def addebita(importo):
    await verifica_antifrode(importo)      # fuori dalla sezione critica
    async with lucchetto:
        saldo["valore"] -= importo

async def prova_saldo():
    saldo["valore"] = 100
    await asyncio.gather(addebita_rotto(30), addebita_rotto(30))
    print("rotto   :", saldo["valore"], "(atteso 40)")
    saldo["valore"] = 100
    await asyncio.gather(addebita(30), addebita(30))
    print("corretto:", saldo["valore"], "(atteso 40)")

asyncio.run(prova_saldo())

# rotto   : 70 (atteso 40)
# corretto: 40 (atteso 40)

Il risultato 70 al posto di 40 è un addebito sparito senza alcun errore, ed è esattamente il tipo di bug che non si vede in un test sequenziale. La versione corretta fa anche un'altra cosa giusta: tiene fuori dal lucchetto la chiamata lenta, e dentro soltanto la modifica. Un lock che racchiude un await di rete serializza tutti gli utenti sulla latenza di quella rete, cioè distrugge esattamente il vantaggio per cui hai scritto codice asincrono.

Il pattern completo: un agente che serve molti utenti

Mettiamo insieme tutti i pezzi nella forma che userai davvero. Client condiviso, semaforo sul fornitore del modello, timeout per richiesta, errori come valori, e spegnimento pulito. Il ciclo vero dell'agente è il metodo _ciclo, lasciato da implementare (qui solleva NotImplementedError): è la funzione della lezione precedente, resa asincrona. Il httpx.AsyncClient è creato e chiuso ma in questo esempio non fa chiamate di rete.

import asyncio, time
import httpx

class ServizioAgente:
    def __init__(self, max_contemporanee=8, timeout_s=45.0):
        # un solo client per processo: riusa connessioni e handshake TLS
        self._client = httpx.AsyncClient(timeout=httpx.Timeout(timeout_s),
                                         limits=httpx.Limits(max_connections=50))
        self._sem = asyncio.Semaphore(max_contemporanee)
        self._timeout_s = timeout_s
        self._in_corso: set[asyncio.Task] = set()      # evita i task orfani

    async def _ciclo(self, richiesta: str) -> str:
        raise NotImplementedError("qui va il ciclo dell'agente")

    async def una_run(self, richiesta: str) -> dict:
        t0 = time.perf_counter()
        try:
            async with self._sem:                       # tetto sulle chiamate parallele
                async with asyncio.timeout(self._timeout_s):
                    testo = await self._ciclo(richiesta)
            return {"ok": True, "testo": testo, "durata_s": round(time.perf_counter() - t0, 2)}
        except TimeoutError:
            return {"ok": False, "errore": "timeout", "durata_s": self._timeout_s}
        except asyncio.CancelledError:
            raise                                       # mai inghiottirla
        except Exception as e:
            return {"ok": False, "errore": f"{type(e).__name__}: {e}"}

    def avvia(self, richiesta: str) -> asyncio.Task:
        t = asyncio.create_task(self.una_run(richiesta))
        self._in_corso.add(t)
        t.add_done_callback(self._in_corso.discard)      # riferimento rilasciato a fine task
        return t

    async def chiudi(self):
        for t in list(self._in_corso):
            t.cancel()
        await asyncio.gather(*self._in_corso, return_exceptions=True)
        await self._client.aclose()

Leggilo in quest'ordine: una_run è l'unico punto in cui una richiesta diventa un risultato (mai un'eccezione verso il chiamante, tranne la cancellazione), avvia è quello in cui la run diventa un task conservato, chiudi quello in cui tutto finisce. Per provarlo senza modello ho derivato una classe che sostituisce _ciclo con una finta, il cui argomento è il numero di secondi da attendere, oppure «guasto» per far sollevare un errore. Ho concesso due posti di semaforo e un timeout di mezzo secondo. Eseguito con httpx 0.28.1, due volte, uscita identica.

class ServizioDiProva(ServizioAgente):
    async def _ciclo(self, richiesta):
        if richiesta == "guasto":
            raise ValueError("tool rotto")
        await asyncio.sleep(float(richiesta))
        return f"fatto dopo {richiesta}s"

async def prova_servizio():
    s = ServizioDiProva(max_contemporanee=2, timeout_s=0.5)
    t = [s.avvia(r) for r in ("0.1", "guasto", "2", "0.1")]
    for x in t[:2]:
        r = await x
        r.pop("durata_s", None)
        print(r)
    print(await t[2])
    print({k: v for k, v in (await t[3]).items() if k != "durata_s"}, "(ha atteso in coda un posto)")
    # una run lunga ancora in corso, poi spegnimento
    lunga = s.avvia("0.4")
    await asyncio.sleep(0.05)
    print("in corso prima di chiudi:", len(s._in_corso))
    await s.chiudi()
    print("in corso dopo chiudi:", len(s._in_corso), "- run cancellata:", lunga.cancelled())

asyncio.run(prova_servizio())

# {'ok': True, 'testo': 'fatto dopo 0.1s'}
# {'ok': False, 'errore': 'ValueError: tool rotto'}
# {'ok': False, 'errore': 'timeout', 'durata_s': 0.5}
# {'ok': True, 'testo': 'fatto dopo 0.1s'} (ha atteso in coda un posto)
# in corso prima di chiudi: 1
# in corso dopo chiudi: 0 - run cancellata: True

Le sei righe raccontano ciò che il codice promette. La prima run riesce. La seconda fallisce con un errore convertito in valore, non in eccezione. La terza dura 2 secondi e viene fermata dal timeout a 0,5. La quarta trova i due posti occupati e aspetta in coda finché la prima non finisce, poi riesce. Alla fine una run lunga è ancora in volo, chiudi la cancella (run cancellata: True) e l'insieme dei task in corso torna vuoto.

Quattro righe meritano un commento, perché sono esattamente quelle che la gente omette. Primo, self._in_corso più add_done_callback è il modo canonico di non perdere un task: lo tieni in un insieme finché non finisce, poi lo togli. La documentazione di asyncio avverte che un task non referenziato altrove può essere raccolto dal garbage collector anche prima di aver finito.

Secondo, l'ordine dei due gestori di contesto decide che cosa limita il timeout. Con il semaforo fuori e il timeout dentro, come qui, il timeout parte dopo aver ottenuto il posto: limita la durata della run, non l'attesa in coda. Se vuoi garantire all'utente un tempo massimo totale, coda inclusa, inverti l'ordine: è il tempo che l'utente percepisce. Non c'è un ordine giusto in assoluto, c'è quello che corrisponde alla garanzia che vuoi dare.

Terzo, except asyncio.CancelledError: raise viene prima di except Exception. Da Python 3.8 CancelledError deriva da BaseException, quindi tecnicamente except Exception non la catturerebbe; scriverlo esplicitamente documenta l'intenzione e protegge dal caso in cui qualcuno, domani, cambi quel blocco in except BaseException. Quarto, chiudi: cancellare i task in volo e chiudere il client distingue uno spegnimento pulito da un processo che lascia connessioni mezze aperte e log pieni di avvisi.

Quando l'utente chiude la pagina: la cancellazione fatta bene

C'è un caso che in un agente capita continuamente e che il codice d'esempio non copre quasi mai: l'utente chiude la scheda del browser mentre la run è a metà. Il server riceve la disconnessione, il framework cancella il task, e a quel punto cosa succede dipende interamente da come hai scritto le coroutine. Se le hai scritte male, resta una chiamata al modello che continua e che pagherai, un lucchetto preso e non rilasciato, un posto nel semaforo occupato per sempre.

La cancellazione in asyncio funziona così: il task riceve una CancelledError nel punto in cui era in attesa, come se quell'await avesse sollevato un'eccezione. Da lì l'eccezione si propaga verso l'alto come qualunque altra, e tutti i blocchi finally e i gestori di contesto (async with) eseguono la loro pulizia. Questo è il meccanismo che rende la cancellazione sicura — a patto di non interferire.

import asyncio, time

class Registro:
    async def apri(self, id_run): print("registro: apri", id_run)
    async def chiudi(self, id_run, motivo): print("registro: chiudi", id_run, motivo)

class Metriche:
    def osserva(self, nome, valore): print("metrica", nome, "registrata")

registro, metriche, semaforo = Registro(), Metriche(), asyncio.Semaphore(2)

async def ciclo_agente(richiesta):
    await asyncio.sleep(float(richiesta))
    return {"ok": True}

async def run_con_pulizia(id_run: str, richiesta: str) -> dict:
    inizio = time.monotonic()
    await registro.apri(id_run)                   # apre una riga di audit
    try:
        async with semaforo:                      # rilasciato dal gestore di contesto
            async with asyncio.timeout(45):
                return await ciclo_agente(richiesta)

    except TimeoutError:
        await registro.chiudi(id_run, "scadenza")
        return {"ok": False, "errore": "timeout"}

    except asyncio.CancelledError:
        # l'utente se n'è andato: registra e RILANCIA
        await registro.chiudi(id_run, "annullata_dall_utente")
        raise                                     # <- senza questa riga rompi tutto

    except Exception as e:
        await registro.chiudi(id_run, f"errore:{type(e).__name__}")
        raise

    finally:
        durata = time.monotonic() - inizio
        metriche.osserva("durata_run_s", durata)  # eseguito in OGNI caso

async def prova_pulizia():
    t = asyncio.create_task(run_con_pulizia("r1", "5"))
    await asyncio.sleep(0.1)
    t.cancel()
    try:
        await t
    except asyncio.CancelledError:
        pass
    print("posti liberi nel semaforo:", semaforo._value, "su 2")

asyncio.run(prova_pulizia())

# registro: apri r1
# registro: chiudi r1 annullata_dall_utente
# metrica durata_run_s registrata
# posti liberi nel semaforo: 2 su 2

Il blocco usa un Registro, una Metriche e un semaforo finti, definiti all'inizio, così gira da solo (due esecuzioni identiche). L'uscita mostra le tre cose che contano dopo un annullamento: la riga di audit viene chiusa con il motivo giusto, la metrica della durata viene registrata comunque, e il semaforo torna a due posti liberi su due. Se dopo la tua cancellazione i posti fossero meno, c'è una risorsa che non rilasci.

Tre punti da leggere con attenzione, perché sono esattamente quelli che si sbagliano.

Il raise dopo aver catturato CancelledError non è opzionale. Catturarla e non rilanciarla significa dire al sistema «questo task si è concluso normalmente» quando in realtà è stato interrotto: il chiamante crede che sia andato tutto bene, i gestori a monte non eseguono la loro pulizia, e nei casi peggiori un TaskGroup o un gather aspetta per sempre qualcosa che non arriverà. La regola è assoluta: si cattura solo per pulire, e poi si rilancia sempre.

Il semaforo viene rilasciato dal gestore di contesto, non da te. È il motivo per cui async with semaforo è nettamente preferibile a un acquire() seguito da un release() a mano: con la forma manuale, un'eccezione fra le due righe lascia il semaforo preso, e dopo qualche richiesta annullata il sistema si blocca da solo — un guasto che si manifesta ore dopo la causa.

Il blocco finally gira anche durante la cancellazione. È la garanzia su cui si appoggia tutta la pulizia, ed è anche una trappola: se dentro il finally metti un await lento, stai rallentando la cancellazione di un task che qualcuno vuole morto. Nella pulizia si fanno cose brevi; il lavoro pesante va altrove.

La verifica che vale la pena fare è proprio quella dell'esempio: avvia una run, cancellala dopo un istante con task.cancel(), e guarda i log e il semaforo. Se non vedi la riga di chiusura e la metrica, hai delle risorse che non vengono rilasciate e non lo sapevi.

Misurare invece di credere

Ogni affermazione sulla concorrenza va verificata, perché il codice asincrono mente con facilità: sembra parallelo e non lo è. Due strumenti bastano quasi sempre. Il primo è la modalità debug di asyncio, che segnala le coroutine che occupano l'event loop troppo a lungo (la soglia predefinita è 0,1 secondi). Lo provo con una coroutine che chiama time.sleep(0.3), cioè lavoro sincrono dentro codice async.

import asyncio, time

async def bloccante():
    time.sleep(0.3)                  # lavoro sincrono dentro una coroutine: ferma l'event loop

async def main():
    await asyncio.gather(bloccante(), asyncio.sleep(0.1))

asyncio.run(main(), debug=True)

# Executing <Task finished name='Task-2' coro=<bloccante() done, defined at .../b4.py:3> result=None created at .../asyncio/tasks.py:732> took 0.300 seconds

Il messaggio (abbreviato: i percorsi dei file sono tagliati) nomina la coroutine colpevole, bloccante(), e quanto ha tenuto fermo l'event loop. Il secondo strumento è un contatore di concorrenza reale, cinque righe attorno alla coroutine da misurare. Lo uso su tre attese da 0,1 secondi, prima una alla volta poi con gather.

import asyncio

in_volo = 0
massimo = 0

async def strumentato(coro):
    global in_volo, massimo
    in_volo += 1
    massimo = max(massimo, in_volo)
    try:
        return await coro
    finally:
        in_volo -= 1

async def attesa():
    await asyncio.sleep(0.1)

async def main():
    global massimo
    for _ in range(3):                      # sequenziale: una alla volta
        await strumentato(attesa())
    print("sequenziale, massimo =", massimo)
    massimo = 0
    await asyncio.gather(*(strumentato(attesa()) for _ in range(3)))
    print("concorrente, massimo =", massimo)

asyncio.run(main())

# sequenziale, massimo = 1
# concorrente, massimo = 3

Se dopo cento richieste massimo vale 1, il tuo codice è sequenziale a prescindere da quanti async def hai scritto. È la verifica che smonta l'illusione più comune, e costa cinque righe.

Le quattro domande che evitano gli errori classici

Ogni volta che scrivo o rileggo codice asincrono per un agente, passo da queste quattro domande. Non sono una checklist burocratica: ciascuna corrisponde a una classe di bug che ho visto costare tempo.

1. «Questa attesa è davvero attesa, o è calcolo?» Un await davanti a una funzione non la rende concorrente: se dentro quella funzione c'è un ciclo che macina per due secondi senza mai cedere il controllo, l'event loop è fermo e tutti gli altri utenti aspettano. Il sintomo è caratteristico e facile da riconoscere una volta che lo conosci: la latenza media è accettabile ma i percentili alti sono disastrosi, perché le richieste si mettono in coda dietro il blocco. La verifica è asyncio.run(main(), debug=True), che segnala le coroutine che occupano l'event loop troppo a lungo.

2. «Fra la creazione e l'attesa, c'è un await?» È il bug del concorrente_sbagliata visto sopra, ed è il più frequente in assoluto perché il codice sembra corretto: ci sono i create_task, c'è il gather mentale, e la sintassi è tutta asincrona. La regola per non sbagliare è sintattica, quindi applicabile a vista: se in un ciclo for c'è un await su qualcosa che poteva partire prima, hai scritto codice sequenziale. Prima si creano tutti i task, poi si attende.

3. «Chi tiene il riferimento a questo task?» Un task creato e non conservato può essere raccolto dal garbage collector e svanire senza eseguire e senza segnalare nulla. È il bug peggiore della categoria, perché non lascia traccia: nessun errore, nessun log, semplicemente una cosa che non è accaduta. La difesa è meccanica: ogni create_task va messo in una lista o in un insieme, oppure si usa un TaskGroup, che tiene i riferimenti per te e garantisce che nessun task sopravviva al blocco.

4. «Cosa succede se questo non risponde mai?» Ogni operazione di rete deve avere un timeout, e ogni timeout deve avere un comportamento definito. Non è pessimismo: un tool che interroga un servizio esterno risponderà lentamente almeno una volta, e nel frattempo tiene occupata una connessione, un posto nel semaforo e la pazienza di un utente. La domanda da porsi non è «metto un timeout?» ma «quanto, e cosa restituisco quando scatta?» — e la risposta giusta è quasi sempre un valore d'errore leggibile dal modello, non un'eccezione che uccide la run.

Una nota che vale più di tutte e quattro, e che tiene insieme questa lezione con quella sui tool: in un agente, un errore di concorrenza non si manifesta come un errore. Non ottieni un crash, ottieni un sistema più lento del previsto, oppure un tool il cui risultato non arriva e viene silenziosamente sostituito da un'eccezione dentro gather, oppure un risultato che il modello interpreta come un dato. Il non-determinismo del modello e il non-determinismo della concorrenza si sommano, e distinguere i due è difficile a posteriori. Per questo le misure di cinque righe della sezione precedente — il contatore di concorrenza reale, la modalità debug — valgono più di un pomeriggio di ragionamento sul codice.

Trade-off

Quattro costi vanno messi in conto. Guadagni tempo, non capacità di calcolo: su lavoro CPU bound asyncio peggiora le cose, perché una funzione che macina per due secondi senza await blocca l'intero event loop, cioè tutti gli altri utenti; quella va in asyncio.to_thread o in un process pool. Contaminazione del codice: una funzione async si chiama solo da codice async, e introdurre asyncio a metà di un progetto sincrono significa riscrivere la catena dei chiamanti fino al confine.

Debug più difficile: gli stack trace attraversano l'event loop e un task che fallisce senza essere atteso può sparire in silenzio; le mitigazioni sono asyncio.run(main(), debug=True) e tenere un riferimento a ogni task creato. Ordine non garantito: gather preserva l'ordine dei risultati, non quello di completamento, quindi se il tuo codice assume che il primo finito sia il primo richiesto, è rotto.

Le trappole in cui si cade davvero

1. Chiamare una libreria sincrona dentro una coroutine. Un requests.get() o un time.sleep(5) in codice async blocca tutto per cinque secondi, utenti compresi. Dentro codice async si usano client async (per HTTP, per esempio httpx.AsyncClient), oppure si incapsula il sincrono in await asyncio.to_thread(funzione, arg).
2. Task orfani. asyncio.create_task(...) senza tenere il riferimento: il garbage collector può raccoglierlo e il task svanisce senza eseguire. Conserva i task in una lista, o usa TaskGroup.
3. Un client HTTP nuovo per ogni richiesta. Butti via il pool di connessioni e paghi un handshake TLS a ogni chiamata. Crea un AsyncClient per processo e passalo in giro.
4. Eccezioni silenziate dentro gather. Con return_exceptions=True gli errori diventano valori: se non controlli con isinstance(x, Exception), un fallimento passa per un risultato valido e finisce nel contesto dell'agente come se fosse un dato.
5. asyncio.run annidato. Chiamarlo quando un loop è già in esecuzione (tipico dentro un notebook o un server) solleva RuntimeError. Dentro codice già async si usa await, non run.
6. Credere che asyncio renda il codice thread-safe. Non ci sono thread, ma c'è interleaving: fra due await lo stato condiviso può essere cambiato da un'altra coroutine. Le sezioni critiche vanno protette con asyncio.Lock.

Le sei trappole hanno la stessa struttura: il programma continua a funzionare. La libreria sincrona dentro una coroutine, il task orfano, il client ricreato, l'eccezione scambiata per risultato e lo stato condiviso toccato fra due await non producono un crash, producono lentezza o dati sbagliati. È per questo che le due verifiche della sezione «Misurare invece di credere» valgono più della lettura del codice.

Che cosa sai fare adesso

Un agente è codice che aspetta: asyncio è il modo per far fruttare quelle attese senza thread. Tre righe da ricordare: crea tutti i task prima di attendere; metti un semaforo e un timeout su qualsiasi cosa tocchi la rete; non catturare mai CancelledError senza rilanciarla. Ora sai scegliere fra asyncio, processi e thread in base a dove un programma aspetta, scrivere le tre varianti di esecuzione e vedere la differenza con un cronometro, e proteggere un servizio con semaforo, timeout e cancellazione. Nella prossima lezione usiamo tutto questo per far girare il primo agente funzionante, prima con un SDK gestito e poi con un modello in locale.

Nella serie, la lezione precedente è Anatomia di un agente: il ciclo, le istruzioni, i tool, la condizione di stop.