04 - Golden dataset: come costruire eval set che predice la production quality

Un dataset comodo dà una stima comoda

Cinquanta domande scelte al volo, punteggi verdi, rilascio. Due settimane dopo gli utenti si lamentano di un terzo delle risposte. Il dataset non era sbagliato per un difetto di calcolo: era stato scritto da chi conosce il sistema, e chi conosce il sistema formula le domande che il sistema capisce. Un golden dataset utile deve soddisfare tre condizioni diverse, e ciascuna si verifica con un controllo preciso: rispecchiare la miscela di domande che arriverà davvero, contenere i casi in cui il sistema fallisce abbastanza da misurarli separatamente, e restare confrontabile nel tempo, cioè non cambiare di nascosto sotto i piedi. Questo articolo costruisce un dataset in questo ordine e usa, ove possibile, listati eseguiti; i numeri provengono da un modello didattico con ipotesi dichiarate e non misurano nessun sistema reale.

Prerequisiti: i concetti di intervallo di confidenza e dimensione del campione dell'episodio 1 e l'idea di un cancello di qualità dell'episodio 3. Il passo successivo, sulle metriche del solo retriever, userà il dataset costruito qui.

Quanto sbaglia la stima, a seconda di come scegli le domande

Prima di ogni tecnica conviene vedere il danno. Il listato simula un sistema con quattro tipi di domanda: facili, medie, difficili e senza risposta nel corpus. La miscela del traffico e la probabilità di risposta corretta in ogni tipo sono ipotesi del modello (60/25/10/5 per cento e 0,95/0,80/0,45/0,30), da cui il tasso di successo reale è 0,830. Poi confronta tre modi di scegliere quaranta domande, ripetuti cinquemila volte con un seme fisso. «Convenienza» imita chi scrive le domande a mano: sceglie prevalentemente quelle facili. «Casuale» le estrae dal traffico. «Stratificato» ne prende dieci per tipo e pesa ogni tipo con la sua quota nel traffico.

# b2.py - modello didattico: quanto sbaglia la stima del tasso di successo con tre modi di scegliere 40 domande
import numpy as np

PESI = np.array([0.60, 0.25, 0.10, 0.05])        # miscela del traffico: facile, media, difficile, senza risposta
P_OK = np.array([0.95, 0.80, 0.45, 0.30])        # IPOTESI: probabilita che il sistema risponda bene in ogni strato
VERO = float(PESI @ P_OK)                          # tasso di successo reale in produzione
N, RIPETIZIONI = 40, 5000
rng = np.random.default_rng(2026)

def convenienza() -> float:
    """Chi scrive le domande a mano sceglie le sue: 85% facili, 15% medie, mai difficili."""
    strati = rng.choice(4, size=N, p=[0.85, 0.15, 0.0, 0.0])
    return rng.binomial(1, P_OK[strati]).mean()

def casuale() -> float:
    """Campione casuale dal traffico, media semplice."""
    strati = rng.choice(4, size=N, p=PESI)
    return rng.binomial(1, P_OK[strati]).mean()

def stratificato() -> float:
    """10 domande per strato, poi media pesata con la miscela del traffico."""
    per_strato = rng.binomial(10, P_OK) / 10
    return float(PESI @ per_strato)

print(f"tasso di successo reale (ipotesi del modello): {VERO:.3f}")
for nome, f in (("convenienza", convenienza), ("casuale", casuale), ("stratificato", stratificato)):
    stime = np.array([f() for _ in range(RIPETIZIONI)])
    print(f"{nome:13s} media delle stime={stime.mean():.3f}  scarto={stime.mean() - VERO:+.3f}  deviazione standard={stime.std():.3f}")

# Output:
# tasso di successo reale (ipotesi del modello): 0.830
# convenienza   media delle stime=0.928  scarto=+0.098  deviazione standard=0.041
# casuale       media delle stime=0.831  scarto=+0.001  deviazione standard=0.060
# stratificato  media delle stime=0.830  scarto=-0.000  deviazione standard=0.055

La convenienza non è solo imprecisa, è sbilanciata: stima 0,928 contro un valore reale di 0,830, circa dieci punti di troppo, e con una dispersione bassa che la fa sembrare affidabile. È il meccanismo del «tutte le metriche erano verdi». Il campione casuale non ha distorsione, ma ha la dispersione più alta (0,060 di deviazione standard). Lo stratificato ha una dispersione solo un po' inferiore (0,055); in questo modello il guadagno sul numero complessivo è modesto, e non è la ragione principale per stratificare. La ragione è che con dieci domande per tipo vedi dove il sistema fallisce: il numero complessivo nasconde che le domande difficili stanno a 0,45, e quello è l'unico dato su cui puoi agire. La pesatura con la miscela del traffico richiede di conoscerla, quindi serve un registro delle domande reali, anche piccolo.

Un campionamento stratificato con quote esplicite

La generazione di domande sintetiche con Ragas, descritta nell'episodio 2, può fornire una prima bozza degli strati, ma con il difetto di costruzione che lì è spiegato: ogni domanda sintetica ha già una risposta nel corpus. Il campionatore fa tre cose che un random.sample non fa. Prende una quota fissa da ogni strato, ordina le domande per identificativo prima di estrarre, in modo che il risultato dipenda solo dal seme e non dall'ordine del file, e fallisce con un messaggio chiaro se uno strato ha meno domande della quota richiesta: riempire in silenzio con altre domande falserebbe la stratificazione senza che nessuno se ne accorga. Il registro inventato ha 400 domande nelle proporzioni dell'esempio precedente.

# b1.py - campionamento stratificato con quote esplicite e seme fisso
import random
from collections import Counter

STRATI = ("facile", "media", "difficile", "senza_risposta")

def log_inventato(n: int = 400, seme: int = 1) -> list:
    """Un registro di domande di esempio: id e strato, nelle proporzioni 60/25/10/5."""
    rng = random.Random(seme)
    pesi = (0.60, 0.25, 0.10, 0.05)
    return [{"id": f"d{i:03d}", "strato": rng.choices(STRATI, pesi)[0]} for i in range(n)]

def campiona(domande: list, quote: dict, seme: int = 42) -> list:
    """Estrae quote[strato] domande da ogni strato. Se uno strato ha meno domande, fallisce: non riempie in silenzio."""
    rng = random.Random(seme)
    scelte = []
    for strato, n in quote.items():
        pool = sorted((d for d in domande if d["strato"] == strato), key=lambda d: d["id"])
        if len(pool) < n:
            raise ValueError(f"strato '{strato}': servono {n} domande, ce ne sono {len(pool)}")
        scelte += rng.sample(pool, n)
    return sorted(scelte, key=lambda d: d["id"])

log = log_inventato()
print("registro:", dict(sorted(Counter(d["strato"] for d in log).items())))

casuale = random.Random(7).sample(log, 40)
print("40 a caso:      ", dict(sorted(Counter(d["strato"] for d in casuale).items())))

stratificato = campiona(log, {"facile": 10, "media": 10, "difficile": 10, "senza_risposta": 10})
print("10 per strato:  ", dict(sorted(Counter(d["strato"] for d in stratificato).items())))

try:
    campiona(log, {"senza_risposta": 500})
except ValueError as errore:
    print("quota impossibile ->", errore)

# Output:
# registro: {'difficile': 40, 'facile': 250, 'media': 89, 'senza_risposta': 21}
# 40 a caso:       {'difficile': 5, 'facile': 25, 'media': 6, 'senza_risposta': 4}
# 10 per strato:   {'difficile': 10, 'facile': 10, 'media': 10, 'senza_risposta': 10}
# quota impossibile -> strato 'senza_risposta': servono 500 domande, ce ne sono 21

Le proporzioni del registro sono 250 facili, 89 medie, 40 difficili e 21 senza risposta. Un campione casuale di quaranta ne prende solo quattro dell'ultimo tipo e cinque dei difficili: troppo poche per dire qualcosa sullo strato, e non per sfortuna ma per costruzione. La quota fissa di dieci per strato dà a ciascuno una stima con un proprio intervallo, largo come quello di un campione di dieci. Un campione di dieci domande ha un intervallo molto ampio (con la formula dell'episodio 1, per una proporzione di 0,8 l'intervallo al 95% va da circa 0,49 a 0,94), quindi dieci per strato sono il punto di partenza, non l'obiettivo. L'ultima riga mostra l'errore che previene: chiedere 500 domande senza risposta a un registro che ne ha 21 solleva un'eccezione. Gli strati si ricavano dal tuo registro o dalle interviste agli utenti, non da una lista standard: l'esempio ne usa quattro perché bastano a mostrare il meccanismo.

Annotare: lo schema, il validatore, l'accordo fra persone

Una domanda con la sua risposta giusta è un dato che qualcuno ha deciso. Lo schema deve conservare ciò che serve a diagnosticare un errore dopo mesi: lo strato, se la domanda è rispondibile, i documenti che contengono la risposta, la risposta di riferimento, chi ha annotato e quando. Il campo rispondibile è quello che separa le domande a cui il sistema deve rispondere da quelle a cui deve rifiutarsi, e senza di esso il rifiuto corretto si conta come errore. L'impronta del corpus lega il dataset alla versione dei documenti su cui è stato annotato: se un documento cambia, l'impronta cambia e i riferimenti vanno riverificati.

# b3.py - un record del golden dataset, il suo validatore e l'impronta del corpus
import hashlib, json
from sklearn.metrics import cohen_kappa_score

CORPUS = {
    "prezzi": "Il piano Pro ha un prezzo mensile di 49 euro e include 10 utenti.",
    "rimborso": "Il rimborso e possibile entro 14 giorni dall'acquisto, con richiesta al supporto.",
    "api": "Le API accettano al massimo 100 richieste al minuto per ogni chiave.",
}

def impronta_corpus(corpus: dict) -> str:
    """Hash stabile del corpus: cambia se cambia anche una virgola di un documento."""
    blocco = json.dumps(corpus, sort_keys=True, ensure_ascii=False).encode("utf-8")
    return "sha256:" + hashlib.sha256(blocco).hexdigest()[:16]

CAMPI = {"id": str, "domanda": str, "strato": str, "rispondibile": bool,
         "id_documenti_attesi": list, "risposta_riferimento": str, "annotatore": str, "data": str}

def valida(record: dict, corpus: dict, visti: set) -> list:
    errori = []
    for campo, tipo in CAMPI.items():
        if campo not in record:
            errori.append(f"manca '{campo}'")
        elif not isinstance(record[campo], tipo):
            errori.append(f"'{campo}' deve essere {tipo.__name__}")
    if record.get("id") in visti:
        errori.append(f"id duplicato '{record['id']}'")
    visti.add(record.get("id"))
    if record.get("rispondibile") and not record.get("id_documenti_attesi"):
        errori.append("domanda rispondibile senza documenti attesi")
    if not record.get("rispondibile") and record.get("id_documenti_attesi"):
        errori.append("domanda senza risposta ma con documenti attesi")
    for d in record.get("id_documenti_attesi", []):
        if d not in corpus:
            errori.append(f"documento atteso '{d}' non esiste nel corpus")
    return errori

DATASET = {
    "versione_schema": 1,
    "corpus": impronta_corpus(CORPUS),
    "record": [
        {"id": "q001", "domanda": "Quanti utenti include il piano Pro?", "strato": "facile", "rispondibile": True,
         "id_documenti_attesi": ["prezzi"], "risposta_riferimento": "Include 10 utenti.",
         "annotatore": "A", "data": "2026-11-12"},
        {"id": "q002", "domanda": "Il piano Pro include il supporto telefonico?", "strato": "senza_risposta", "rispondibile": False,
         "id_documenti_attesi": [], "risposta_riferimento": "Il corpus non lo dice: il sistema deve rifiutarsi.",
         "annotatore": "A", "data": "2026-11-12"},
        {"id": "q002", "domanda": "Come funziona il rimborso?", "strato": "facile", "rispondibile": True,
         "id_documenti_attesi": ["rimborsi"], "risposta_riferimento": "Entro 14 giorni.",
         "annotatore": "B", "data": "2026-11-12"},
    ],
}

visti = set()
for r in DATASET["record"]:
    errori = valida(r, CORPUS, visti)
    print(r["id"], "OK" if not errori else errori)
print("impronta corpus:", DATASET["corpus"])
CORPUS["api"] = CORPUS["api"].replace("100", "200")
print("dopo una modifica:", impronta_corpus(CORPUS))

# due persone etichettano le stesse 10 domande come "risposta corretta" (1) o no (0): etichette inventate
a = [1, 1, 0, 1, 1, 0, 1, 0, 1, 1]
b = [1, 1, 0, 1, 0, 0, 1, 0, 1, 1]
accordo = sum(x == y for x, y in zip(a, b)) / len(a)
print(f"accordo grezzo={accordo:.0%}  kappa di Cohen={cohen_kappa_score(a, b):.2f}")

# Output:
# q001 OK
# q002 OK
# q002 ["id duplicato 'q002'", "documento atteso 'rimborsi' non esiste nel corpus"]
# impronta corpus: sha256:2e887c623b9de0e9
# dopo una modifica: sha256:8547d3d9909f2e74
# accordo grezzo=90%  kappa di Cohen=0.78

Il validatore è dove lo schema smette di essere una convenzione. Controlla che ogni campo ci sia e abbia il tipo giusto, che gli identificativi siano unici, che una domanda rispondibile abbia almeno un documento atteso e che una non rispondibile non ne abbia, e che i documenti attesi esistano davvero nel corpus. Nell'esempio il terzo record ha un identificativo duplicato e cita un documento («rimborsi») che nel corpus si chiama «rimborso»: due errori che producono un dataset silenziosamente sbagliato, perché il retriever non lo troverà mai e il punteggio crollerà per colpa dell'annotazione. L'ultima riga mostra l'impronta prima e dopo la modifica di una cifra in un documento: sono diverse, e la differenza è ciò che ti dice di rifare il controllo.

Resta l'accordo fra persone. Se non hai esperti di dominio, fai annotare lo stesso sottoinsieme a due persone e misura la concordanza con il kappa di Cohen, che corregge l'accordo grezzo per quello atteso per caso: nell'esempio, 90% di accordo grezzo diventa un kappa di 0,78. La documentazione di scikit-learn lo definisce come (p_o − p_e) / (1 − p_e), con p_o l'accordo osservato e p_e quello atteso. Un kappa basso non dice chi ha ragione, dice che la domanda o il criterio sono ambigui; discutile prima di fidarti del punteggio che ne deriva. Le etichette dell'esempio sono inventate, e il kappa su dieci domande è a sua volta molto incerto.

I casi difficili: perturbare le domande

Gli strati descrivono la difficoltà della domanda; le perturbazioni descrivono la sua forma. La stessa domanda può arrivare con un refuso, in maiuscolo, preceduta da frasi di cortesia, troncata o riformulata con parole diverse. Il listato prende dieci domande, ciascuna con il suo documento giusto in un corpus di dieci, e le sottopone al recupero TF-IDF in sei forme. La parafrasi è scritta a mano per ogni domanda, perché una parafrasi automatica introdurrebbe un secondo modello da validare.

# b4.py - le stesse domande, perturbate: quanto resiste il recupero lessicale
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer

DOCS = {
    "prezzi":   "Il piano Pro ha un prezzo mensile di 49 euro e include 10 utenti.",
    "rimborso": "Il rimborso e possibile entro 14 giorni dall'acquisto, con richiesta al supporto.",
    "export":   "L'esportazione dei dati produce un file CSV dalla pagina Impostazioni.",
    "sso":      "L'accesso con SSO e disponibile solo nel piano Enterprise tramite SAML.",
    "backup":   "I backup automatici girano ogni notte e si conservano per 30 giorni.",
    "api":      "Le API accettano al massimo 100 richieste al minuto per ogni chiave.",
    "fatture":  "Le fatture si scaricano in PDF dalla sezione Fatturazione dell'account.",
    "password": "Per reimpostare la password usa il link inviato alla tua email di registrazione.",
    "integraz": "Le integrazioni con Slack e Teams si attivano dal pannello Integrazioni.",
    "limiti":   "Ogni progetto accetta fino a 500 documenti caricati in formato PDF.",
}
# (domanda canonica, parafrasi scritta a mano, documento giusto)
CASI = [
    ("quanto costa al mese il piano Pro", "che prezzo mensile ha il Pro", "prezzi"),
    ("come scarico i miei dati in CSV", "voglio un file CSV con i dati", "export"),
    ("serve SAML per accedere", "il login SSO e nel piano Enterprise", "sso"),
    ("ogni quanto si fanno i backup", "i backup automatici sono giornalieri", "backup"),
    ("quante richieste al minuto accetta l'API", "limite di richieste al minuto per chiave", "api"),
    ("dove scarico le fatture in PDF", "scaricare fatture PDF fatturazione", "fatture"),
    ("ho dimenticato la password", "reimpostare la password dimenticata", "password"),
    ("come collego Slack", "attivare l'integrazione con Slack", "integraz"),
    ("quanti documenti posso caricare per progetto", "numero massimo di documenti per progetto", "limiti"),
    ("posso farmi restituire i soldi", "come funziona il rimborso", "rimborso"),
]

def refuso(q: str) -> str:
    """Scambia due lettere adiacenti nella parola piu lunga (il refuso tipico)."""
    parole = q.split()
    i = max(range(len(parole)), key=lambda k: (len(parole[k]), -k))
    p = parole[i]
    m = len(p) // 2
    parole[i] = p[: m - 1] + p[m] + p[m - 1] + p[m + 1:]
    return " ".join(parole)

PERTURBAZIONI = {
    "originale":   lambda c, p: c,
    "parafrasi":   lambda c, p: p,
    "refuso":      lambda c, p: refuso(c),
    "maiuscolo":   lambda c, p: c.upper(),
    "con cortesie": lambda c, p: "ciao scusa una domanda veloce, " + c + ", grazie",
    "troncata":    lambda c, p: " ".join(c.split()[:3]),
}

ids = list(DOCS)
vec = TfidfVectorizer(sublinear_tf=True)
M = vec.fit_transform(DOCS.values())
for nome, f in PERTURBAZIONI.items():
    hit = 0
    for canonica, parafrasi, giusto in CASI:
        punteggi = (M @ vec.transform([f(canonica, parafrasi)]).T).toarray().ravel()
        hit += giusto in [ids[i] for i in np.argsort(-punteggi, kind="stable") if punteggi[i] > 0][:3]
    print(f"{nome:13s} hit@3={hit}/{len(CASI)}")

# Output:
# originale     hit@3=9/10
# parafrasi     hit@3=10/10
# refuso        hit@3=9/10
# maiuscolo     hit@3=9/10
# con cortesie  hit@3=9/10
# troncata      hit@3=7/10

I risultati contraddicono l'intuizione comune: il recupero lessicale regge bene a maiuscole, refusi e cortesie (nove domande su dieci, come per il testo originale), perché molte parole della domanda restano intatte. Cala sulla versione troncata a tre parole (sette su dieci). La domanda mancata già nella forma originale, «posso farmi restituire i soldi», è un problema di vocabolario, non di forma: la parafrasi la recupera (dieci su dieci) soltanto perché l'ho scritta con la parola «rimborso», la stessa del documento, quindi quel 10/10 dimostra il meccanismo e non la robustezza. Ne ricavi una regola di lavoro: le perturbazioni servono a misurare quali forme il tuo retriever non sopporta, e la risposta cambia con il retriever; un modello a embedding si comporta in modo diverso e va rimisurato. Prova sempre ciascuna perturbazione contro l'originale e riporta la differenza, non il valore assoluto.

Versionare il dataset

Il dataset è un file, e un file si versiona. Finché è un JSON di qualche centinaio di record, la soluzione più semplice è metterlo in Git insieme al codice, con la versione dello schema e l'impronta del corpus nel file e una etichetta di Git a ogni rilascio del dataset; ogni punteggio riportato deve citare la versione con cui è stato calcolato. Quando il dataset o il corpus diventano grandi, DVC offre un'alternativa: la sua documentazione spiega che non è un sistema di controllo di versione in sé, ma manipola file .dvc che definiscono le versioni dei dati e che Git può versionare, mentre i dati stanno altrove. Non c'è bisogno di DVC prima che serva.

Il dataset predice la produzione? Una verifica con pochi punti

L'unico modo di sapere se il golden dataset funziona è confrontarlo con ciò che accade in produzione. Hai una serie di versioni del sistema; per ciascuna hai il punteggio sul dataset e un segnale di produzione che scegli tu (la quota di risposte giudicate utili dagli utenti, per esempio). Se i due si muovono insieme, il dataset sta guardando la cosa giusta. Il listato calcola la correlazione di Pearson con il suo intervallo di confidenza, quella di Spearman, che guarda solo l'ordine delle versioni, e mostra come si restringerebbe l'intervallo con più versioni. I dati sono inventati: servono a mostrare il metodo.

# b5.py - il golden dataset predice la produzione? La correlazione fra versioni, con il suo intervallo
import math
from scipy.stats import pearsonr, spearmanr

# DATI INVENTATI per illustrare il metodo: 8 versioni del sistema.
# punteggio sul golden dataset e quota di risposte giudicate utili in produzione (da un segnale a tua scelta)
punteggio_eval = [0.62, 0.68, 0.71, 0.74, 0.77, 0.80, 0.83, 0.86]
utili_produzione = [0.55, 0.66, 0.60, 0.72, 0.70, 0.78, 0.74, 0.81]

r = pearsonr(punteggio_eval, utili_produzione)
lo, hi = r.confidence_interval(confidence_level=0.95)
rho = spearmanr(punteggio_eval, utili_produzione)
print(f"versioni={len(punteggio_eval)}")
print(f"Pearson r={r.statistic:.2f}  intervallo 95%=[{lo:.2f}, {hi:.2f}]")
print(f"Spearman rho={rho.statistic:.2f}  (concorda sull'ordine delle versioni?)")

# se ne avessi 24 con la stessa correlazione, l'intervallo si restringerebbe cosi (Fisher z):
z = math.atanh(r.statistic)
for n in (8, 24, 60):
    s = 1 / math.sqrt(n - 3)
    print(f"n={n:3d}: [{math.tanh(z - 1.96 * s):.2f}, {math.tanh(z + 1.96 * s):.2f}]")

# Output:
# versioni=8
# Pearson r=0.92  intervallo 95%=[0.61, 0.99]
# Spearman rho=0.93  (concorda sull'ordine delle versioni?)
# n=  8: [0.61, 0.99]
# n= 24: [0.82, 0.97]
# n= 60: [0.87, 0.95]

Con otto versioni la correlazione di 0,92 ha un intervallo da 0,61 a 0,99: la direzione è chiara, l'intensità no. Spearman dà 0,93, cioè l'ordine delle versioni è quasi lo stesso nelle due serie. Con ventiquattro versioni e la stessa correlazione l'intervallo andrebbe da 0,82 a 0,97, con sessanta da 0,87 a 0,95. Non avrai mai sessanta versioni: ma puoi usare sottoinsiemi del traffico o finestre temporali come punti aggiuntivi, ricordando che non sono indipendenti. La correlazione non dimostra una causa, e un valore alto con un segnale di produzione sbagliato (per esempio la lunghezza della risposta) non vale nulla: scegli il segnale prima di guardare i risultati.

Che cosa sai fare adesso

Sai mostrare, con una simulazione che puoi rifare, perché un dataset scritto a mano sovrastima la qualità, e sai costruirne uno stratificato con quote esplicite, seme fisso ed errore chiaro quando uno strato non basta. Sai scrivere uno schema che distingue le domande a cui rispondere da quelle a cui rifiutarsi, validarlo con codice, legarlo a un'impronta del corpus e misurare l'accordo fra chi annota. Sai perturbare le domande per vedere quali forme il tuo retriever non regge, e verificare con un intervallo di confidenza se il punteggio sul dataset segue la produzione. Per la parte che riguarda l'osservazione continua dopo il rilascio puoi continuare con RAG in produzione: monitoring, evaluation, optimization.

Nella stessa serie trovi anche l'automazione della valutazione.