07 - Evaluation framework automation: scoring pipeline, batch processing, async evaluation

Automatizzare la valutazione: scoring in parallelo, cache e cancello sulla PR

Hai un insieme di valutazione di 500 domande, tre sistemi di retrieval da confrontare e due versioni del prompt. Moltiplica: 500 × 3 × 2 fa 3000 giudizi, e se per ciascuno chiedi tre metriche le chiamate al giudice diventano 9000. Quanto durino dipende da una sola grandezza, la latenza del tuo giudice, che dipende da modello, hardware e lunghezza dei prompt e che non posso darti io. Quello che puoi fare è ragionare su come il tempo scala, ed è ciò che questo episodio costruisce: un modello di calcolo del tempo, un valutatore asincrono con limite di concorrenza e cache, e un cancello che blocca la modifica che peggiora le metriche.

I listati girano con la sola libreria standard più numpy, senza alcun modello: il giudice è finto, con latenza simulata, perché ciò che si studia è l'orchestrazione e non il verdetto. Per cosa misurare con quel giudice vedi l'episodio sulla faithfulness; per come costruire l'insieme di valutazione, quello sul golden dataset; per il quadro d'insieme di una pipeline di valutazione, questo articolo della serie AI Engineering.

Quanto tempo risparmia il parallelismo: un modello, non una misura

Prima di scrivere l'orchestrazione conviene sapere cosa aspettarsi. Il listato genera 3000 latenze di giudice con distribuzione lognormale (mediana 8 secondi, coda lunga: poche chiamate molto lente) e calcola il tempo di completamento con k lavoratori, assegnando ogni chiamata al primo slot libero. Il seme del generatore è fisso, quindi i numeri sono riproducibili; ma sono numeri di un modello e non di una macchina. Sostituisci mediana e dispersione con quelle del tuo giudice e rifai il conto.

import heapq
import numpy as np

# Modello, non misura: latenze di un giudice con mediana 8 s e coda lunga (lognormale).
rng = np.random.default_rng(7)
latenze = rng.lognormal(mean=np.log(8.0), sigma=0.4, size=3000)   # secondi, 500 x 3 x 2


def tempo_totale(durate, lavoratori):
    """Tempo di completamento con `lavoratori` slot paralleli (assegna al primo libero)."""
    liberi = [0.0] * lavoratori
    heapq.heapify(liberi)
    for d in durate:
        inizio = heapq.heappop(liberi)
        heapq.heappush(liberi, inizio + d)
    return max(liberi)


print(f"somma delle latenze: {latenze.sum() / 3600:.2f} ore")
for k in (1, 3, 5, 10):
    t = tempo_totale(latenze, k)
    print(f"{k:2d} lavoratori: {t / 60:7.1f} min   (ideale {latenze.sum() / k / 60:7.1f})")

# La cache elimina chiamate, non le rende più veloci: si tolgono i task duplicati
for quota in (0.0, 0.2, 0.4):
    restanti = latenze[int(len(latenze) * quota):]
    print(f"cache {quota:.0%}: 5 lavoratori -> {tempo_totale(restanti, 5) / 60:6.1f} min")

# Output:
# somma delle latenze: 7.13 ore
#  1 lavoratori:   427.7 min   (ideale   427.7)
#  3 lavoratori:   142.7 min   (ideale   142.6)
#  5 lavoratori:    85.7 min   (ideale    85.5)
# 10 lavoratori:    43.0 min   (ideale    42.8)
# cache 0%: 5 lavoratori ->   85.7 min
# cache 20%: 5 lavoratori ->   69.5 min
# cache 40%: 5 lavoratori ->   52.3 min

Il risultato mostra due fatti. Primo, in questo modello il tempo scala quasi come 1/k: con cinque lavoratori l'attesa cala da circa 428 a circa 86 minuti, perché le chiamate sono indipendenti e il valore ideale (somma divisa per k) è raggiunto quasi per intero. Secondo, nella realtà quel «k» non lo scegli liberamente: lo fissa il server del giudice, che oltre una certa concorrenza serializza internamente o rifiuta richieste, e il guadagno si ferma. La concorrenza giusta si trova misurando il throughput al crescere di k sul tuo hardware, non copiando il numero di qualcun altro. Le righe sulla cache mostrano che togliere il 40% delle chiamate porta il caso a cinque lavoratori da 86 a 52 minuti: la cache non rende veloce la chiamata, la evita.

Il valutatore asincrono: semaforo, cache e il problema dei duplicati simultanei

Un asyncio.Semaphore è un contatore interno: ogni acquire() lo decrementa, ogni release() lo incrementa, e non scende sotto zero. Chi trova zero aspetta. Usato con async with limita a N le chiamate contemporanee al giudice, qualunque sia il numero di task lanciati con gather. Questo ti protegge da un errore comune: lanciare 3000 richieste insieme e ottenere timeout a cascata.

La cache è più insidiosa. L'implementazione naturale controlla un dizionario, chiama il giudice se manca la chiave, e salva il risultato a fine chiamata. Ma con gather tutti i task partono prima che il primo finisca, quindi nessuno trova mai una chiave già presente: i duplicati attendono in parallelo e pagano tutti la chiamata. Il rimedio, chiamato single-flight, è salvare in cache il Task appena creato e non il risultato: chi arriva dopo attende lo stesso Task. Nel listato la differenza si legge in due righe, cache[k] = asyncio.create_task(...) contro cache[k] = await .... Tenere il Task nel dizionario serve anche a un secondo scopo: la documentazione di Python avverte che il ciclo di eventi conserva solo riferimenti deboli ai task, e un task non referenziato può sparire a metà esecuzione.

import asyncio
import hashlib
import json


class Giudice:
    """Giudice finto: nessun modello, solo latenza simulata e un punteggio derivato dall'hash."""

    def __init__(self, concorrenza):
        self.sem = asyncio.Semaphore(concorrenza)
        self.in_volo = 0
        self.picco = 0
        self.chiamate = 0

    async def valuta(self, elemento, metrica):
        async with self.sem:
            self.in_volo += 1
            self.picco = max(self.picco, self.in_volo)
            self.chiamate += 1
            await asyncio.sleep(0.01)                    # il "modello" impiega tempo
            self.in_volo -= 1
        h = hashlib.sha256(f"{elemento}|{metrica}".encode()).digest()
        return h[0] / 255


def chiave(elemento, metrica):
    return hashlib.sha256(json.dumps([elemento, metrica], sort_keys=True).encode()).hexdigest()


async def esegui(elementi, metriche, concorrenza, single_flight):
    giudice = Giudice(concorrenza)
    cache = {}          # chiave -> risultato (naive) oppure Task (single-flight)

    async def uno(e, m):
        k = chiave(e, m)
        if single_flight:
            if k not in cache:                           # il Task entra in cache SUBITO
                cache[k] = asyncio.create_task(giudice.valuta(e, m))
            return await cache[k]
        if k in cache:
            return cache[k]
        cache[k] = await giudice.valuta(e, m)            # entra in cache solo a fine chiamata
        return cache[k]

    risultati = await asyncio.gather(*(uno(e, m) for e in elementi for m in metriche))
    return giudice, risultati


# 100 risposte, ma solo 40 sono distinte: capita quando lo stesso test gira su più varianti
elementi = [f"risposta-{i % 40}" for i in range(100)]
metriche = ["faithfulness", "context_precision"]
for sf in (False, True):
    g, r = asyncio.run(esegui(elementi, metriche, concorrenza=5, single_flight=sf))
    nome = "single-flight" if sf else "cache naive  "
    print(f"{nome}: richieste={len(r)} chiamate al giudice={g.chiamate} picco concorrenza={g.picco}")

# Output:
# cache naive  : richieste=200 chiamate al giudice=200 picco concorrenza=5
# single-flight: richieste=200 chiamate al giudice=80 picco concorrenza=5

L'esito è netto. Con la cache naive le 200 richieste (100 risposte × 2 metriche, di cui solo 80 distinte) producono 200 chiamate al giudice, cioè zero risparmio; con il single-flight le chiamate sono 80. In entrambi i casi il picco di concorrenza osservato è 5, il valore del semaforo, a conferma che il limite regge. Il giudice qui è finto: il punteggio deriva dall'hash dell'input, quindi la stessa risposta riceve sempre lo stesso punteggio, che è la proprietà che una cache richiede. Con un LLM vero a temperatura positiva due chiamate identiche possono dare risultati diversi, e la cache fissa la prima: è un compromesso consapevole, non un difetto.

Cosa metti nella chiave decide quando la cache è corretta:

• il testo della domanda, del contesto e della risposta, perché cambiano il verdetto;
• il nome della metrica e la versione del suo prompt di giudizio;
• il modello giudice e la sua versione, se la cambi senza saperlo i punteggi non sono più confrontabili;
• non il tempo né un identificatore di run, che renderebbero ogni chiave unica.

La regola pratica è che la chiave contenga tutto ciò che, se cambiasse, cambierebbe la risposta del giudice, e nient'altro. Una chiave troppo povera restituisce punteggi vecchi per input nuovi, l'errore più difficile da accorgersi; una troppo ricca azzera i risparmi. Ordinare le chiavi del JSON, come fa sort_keys=True, evita che lo stesso contenuto produca due hash diversi. Per i fallimenti transitori aggiungi un numero limitato di tentativi e un timeout per chiamata con asyncio.wait_for, perché una chiamata bloccata occupa uno slot del semaforo per sempre.

Il cancello: bloccare la modifica che peggiora le metriche

Automatizzare la misura serve a decidere. Il cancello è uno script che confronta i risultati con due criteri: un minimo assoluto per ogni metrica e una regressione massima rispetto alla base, cioè all'ultima versione accettata. Il secondo criterio evita un difetto del solo minimo assoluto: una metrica può scendere da 0,95 a 0,81 senza mai violare la soglia 0,80, e la modifica che l'ha causata passerebbe. Lo script restituisce un codice di uscita, 0 o 1, che è tutto ciò che l'integrazione continua deve sapere.

SOGLIE = {"faithfulness": 0.80, "context_precision": 0.75}   # minimi assoluti
MAX_CALO = 0.03                                               # regressione ammessa rispetto alla base


def cancello(risultati, base):
    """Ritorna (codice_di_uscita, motivi). 0 = ok, 1 = merge da bloccare."""
    motivi = []
    for metrica, minimo in SOGLIE.items():
        ora, prima = risultati[metrica], base[metrica]
        if ora < minimo:
            motivi.append(f"{metrica} {ora:.3f} sotto il minimo {minimo:.2f}")
        if prima - ora > MAX_CALO:
            motivi.append(f"{metrica} calata di {prima - ora:.3f} rispetto alla base")
    return (1 if motivi else 0), motivi


base = {"faithfulness": 0.86, "context_precision": 0.80}
for nome, run in [("pr-a", {"faithfulness": 0.85, "context_precision": 0.79}),
                  ("pr-b", {"faithfulness": 0.79, "context_precision": 0.80})]:
    codice, motivi = cancello(run, base)
    print(nome, "uscita", codice, motivi)

# Output:
# pr-a uscita 0 []
# pr-b uscita 1 ['faithfulness 0.790 sotto il minimo 0.80', 'faithfulness calata di 0.070 rispetto alla base']

La prima modifica passa. La seconda viola entrambi i criteri: la faithfulness è 0,79, sotto il minimo di 0,80, ed è calata di 0,07 rispetto alla base. Le soglie del listato (0,80, 0,75, calo ammesso 0,03) sono esempi: calibra la tolleranza sulla variabilità della tua metrica a parità di codice, ripetendo la misura più volte, altrimenti un cancello troppo stretto blocca per rumore e finisce disattivato da chi ne è infastidito.

Lo schema del flusso su GitHub Actions è breve: si avvia solo quando cambiano i percorsi che influenzano la qualità, esegue la valutazione su un sottoinsieme piccolo e veloce, poi lancia il cancello. Perché blocchi davvero il merge, il controllo deve essere impostato come obbligatorio nelle regole di protezione del ramo: la documentazione di GitHub spiega che i controlli richiesti devono avere stato riuscito, saltato o neutro prima che si possa modificare il ramo protetto. Il sottoinsieme per le PR e l'insieme completo notturno rispondono a due domande diverse: «ho rotto qualcosa?» e «come sta il sistema nel complesso?».

# Schema da adattare, non eseguito: le cartelle e gli script sono quelli del tuo repository.
name: rag-eval-gate
on:
  pull_request:
    paths: ["retrieval/**", "prompts/**", "eval/**"]   # parte solo se cambia qualcosa di rilevante
jobs:
  eval:
    runs-on: ubuntu-latest
    timeout-minutes: 30
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -r eval/requirements.txt
      - run: python eval/run_eval.py --dataset eval/golden_ci.json --out eval/risultati.json
      - run: python eval/cancello.py eval/risultati.json eval/base.json   # exit 1 = job fallito

Tracciare le esecuzioni e quando non serve

Un punteggio senza contesto non dice nulla tra tre mesi: serve sapere con quale dataset, quale prompt, quale modello giudice è stato ottenuto. Strumenti di tracciamento come MLflow offrono funzioni per questo; la sua documentazione descrive start_run per aprire una esecuzione e log_param e log_metric per registrare parametri e metriche. Per un team piccolo un file JSON versionato accanto al codice, con i parametri e i punteggi di ogni esecuzione, fa lo stesso lavoro senza un servizio da mantenere. Passa a uno strumento dedicato quando confronti molte esecuzioni e ti serve un'interfaccia per farlo, non prima.

Non tutto va automatizzato subito. Con un insieme di cinquanta domande e un giudice veloce, il ciclo manuale richiede minuti e l'infrastruttura costerebbe più del problema. L'ordine ragionevole è: prima lo script di valutazione riproducibile, poi la cache, poi il parallelismo, poi il cancello, infine il tracciamento. Ogni passo si giustifica con un dolore reale che hai già provato.

Domande frequenti

Quanti lavoratori concorrenti devo usare?

Quanti ne regge il server del giudice senza serializzare o rifiutare richieste. Si trova misurando il throughput al crescere della concorrenza sul proprio hardware e fermandosi dove smette di aumentare. Il valore dipende da modello, hardware e lunghezza dei prompt.

La cache può falsare i risultati?

Sì, se la chiave omette qualcosa che cambia il verdetto, come la versione del prompt o del modello giudice. Una cache corretta contiene nella chiave tutto ciò che influenza la risposta e nient'altro, e si svuota quando cambi il giudice.

Perché serve il cancello se guardo già i numeri?

Perché guardare i numeri a mano richiede che qualcuno si ricordi di farlo ad ogni modifica. Il cancello rende la regressione una condizione di merge, non un'abitudine, ed è tracciabile nella cronologia della PR.

Un LLM giudice è deterministico?

Non necessariamente. Abbassare la temperatura riduce la variabilità ma non sempre la elimina. Per questo i cancelli usano tolleranze calibrate sulla variabilità misurata e non confronti esatti.

Che cosa sai fare adesso

Sai stimare come cambia il tempo di una valutazione al variare della concorrenza, limitare le chiamate con un semaforo, evitare che la cache fallisca con i duplicati simultanei, scrivere un cancello con minimo assoluto e regressione massima e collegarlo a una PR. Sai anche quali di questi pezzi aspettare finché non servono. Nel prossimo episodio, l'ottimizzazione guidata dalla valutazione, usi questa infrastruttura per confrontare varianti con un test statistico.

Nella stessa serie trovi anche il monitoraggio in produzione.