""" Corso Machine Learning Engineer - Modulo 10, Lezione 05 "RAG da zero: recuperare il contesto giusto, e le risposte con una citazione verificabile" OBIETTIVO DI QUESTO FILE ------------------------ Tre esercizi sulle tre misure che decidono se un RAG funziona. Girano OFFLINE: nessuna rete, nessun modello scaricato, nessun servizio in ascolto. Il corpus è quello dell'articolo - i docstring della libreria standard, letti con `inspect` - e il recupero usa un vettorizzatore lessicale deterministico invece di un embedding neurale. Perché lessicale e non neurale: l'oggetto degli esercizi è la MACCHINA DI MISURA (tetto, recall, verifica della citazione), non la qualità del recuperatore. Un embedding migliore sposta i numeri verso l'alto; non cambia una riga del codice che li calcola. Quando avrai Ollama in ascolto, sostituire `vettorizza_lessicale` con una chiamata a `nomic-embed-text` è l'esercizio 3b in fondo al file: tutto il resto resta identico. 1. IL TETTO. Dato un recupero, quante query hanno la risposta nel contesto passato al modello? È il massimo che il generatore può raggiungere. Nessun prompt lo supera. Si calcola in dieci righe e va calcolato PRIMA di toccare qualunque altra cosa. 2. L'ANCORA TAGLIATA. Una finestra fissa che cade dentro la frase-risposta la rende irrecuperabile: nessun chunk la contiene più tutta. Misuri quanto spesso accade su una spazzata di dimensioni, e quanto costa il rimedio. 3. LA VERIFICA CHE NON SI FIDA. Un controllo programmatico che per ogni frase risale allo span citato e pretende che numeri e nomi compaiano li' dentro LETTERALMENTE. Le risposte che non passano si rifiutano. COME SI USA ----------- python3 esercizi.py Il file gira anche così com'è: stampa quali esercizi mancano ancora. Completa i TODO uno alla volta e riesegui. Le soluzioni commentate sono in fondo al file. AMBIENTE DICHIARATO ------------------- Python 3.12+ e NumPy 2.x. Nessun altro pacchetto. python3 -m venv .venv && source .venv/bin/activate pip install "numpy>=2.0" Eseguito con Python 3.14.4 e NumPy 2.3.5. Il corpus dipende dalla versione di Python: i docstring cambiano fra le minor. I numeri stampati dal file sono quindi riproducibili sulla TUA versione, non identici a quelli dell'articolo - ed è il motivo per cui il file ristampa sempre la versione con cui gira. """ import importlib import inspect import re import sys import numpy as np MODULI = ["statistics", "bisect", "heapq", "itertools", "functools", "difflib", "textwrap", "tempfile", "uuid", "base64", "secrets", "gzip", "math", "re", "json", "dataclasses", "contextlib", "ipaddress", "graphlib", "decimal", "fractions", "csv", "random", "shlex", "pathlib", "hashlib", "calendar", "queue", "copy", "pprint", "enum", "time", "array", "zlib", "struct", "pickle", "shutil", "glob", "fnmatch", "operator", "abc", "numbers", "cmath", "mimetypes", "stat", "filecmp", "zipfile", "tarfile", "bz2", "lzma", "binascii", "codecs", "io", "traceback", "warnings", "weakref", "gc", "tokenize", "tomllib", "zoneinfo"] # ====================================================================== # CORPUS - un documento per modulo, costruito offline con `inspect` # ====================================================================== def costruisci_corpus(): """Un documento per modulo: docstring del modulo + docstring pubblici.""" documenti = {} for nome in MODULI: try: m = importlib.import_module(nome) except ImportError: continue parti = [f"MODULE {nome}"] testo_modulo = inspect.getdoc(m) if testo_modulo: parti.append(testo_modulo.strip()) visti = set() for attr, obj in sorted(vars(m).items()): if attr.startswith("_"): continue if not (inspect.isfunction(obj) or inspect.isclass(obj) or inspect.isbuiltin(obj)): continue doc = inspect.getdoc(obj) if not doc or len(doc.split()) < 20: continue # la stessa funzione è ri-esportata da più moduli: tienila una volta impronta = (getattr(obj, "__module__", nome), getattr(obj, "__qualname__", attr)) if impronta in visti: continue visti.add(impronta) parti.append(f"## {nome}.{attr}\n\n{doc.strip()}") corpo = "\n\n".join(parti) if len(corpo.split()) >= 120: documenti[nome] = corpo return documenti def normalizza(testo): """Spazi collassati: serve a confrontare un'ancora con un chunk.""" return " ".join(testo.split()) # Otto domande, con il documento atteso e l'ANCORA: la stringa letterale che # contiene la risposta. Un chunk conta come corretto solo se la contiene TUTTA. # # Le domande sono PARAFRASI deliberate: non riusano i termini rari del # documento. È il caso difficile e anche quello realistico - chi fa una # domanda non conosce il vocabolario della risposta, altrimenti non la # cercherebbe. Su domande che ricopiano i termini del documento qualunque # recuperatore sembra perfetto. GOLDEN = [ ("q1", "how do I split a sample into ten groups of equal probability?", "statistics", "Set *n* to 10 for deciles."), ("q2", "what is the lowest resemblance a suggested word must reach before it is kept?", "difflib", "Optional arg cutoff (default 0.6) is a float in [0, 1]."), ("q3", "which symbols stand in for the two characters that are unsafe inside a web address?", "base64", "The alphabet uses '-' instead of '+' and '_' instead of\n'/'."), ("q4", "after making a one-off scratch file at the lowest level, whose job is cleaning it up?", "tempfile", "Caller is responsible for deleting the file when done with it."), ("q5", "how do I check a secret token without leaking how long the match took?", "secrets", "This function uses an approach designed to prevent\ntiming analysis"), ("q6", "what reading mode and what squeezing strength apply when I ask for neither?", "gzip", "The default mode is\n\"rb\", and the default compresslevel is 9."), ("q7", "how do I average two speeds held over the same distance travelled?", "statistics", ">>> harmonic_mean([40, 60])\n 48.0"), ("q8", "what is left over when I chop a sequence into equal handfuls and it does not divide evenly?", "itertools", "The last batch may be shorter than n."), ] # ====================================================================== # RECUPERATORE LESSICALE - deterministico, offline, senza dipendenze # ====================================================================== _PAROLA = re.compile(r"[a-z0-9_]+") _STOP = frozenset("a an and are as at be by do does for from how i in is it of on or " "that the to what when where which who why with not".split()) def _termini(testo): return [p for p in _PAROLA.findall(testo.lower()) if p not in _STOP and len(p) > 1] def vettorizza_lessicale(documenti_testo): """Matrice TF-IDF a righe normalizzate, scritta a mano con NumPy. Restituisce (M, vocabolario, idf): M ha una riga per documento, già divisa per la propria norma, così il prodotto scalare È il coseno. """ tokenizzati = [_termini(t) for t in documenti_testo] vocabolario = {} for toks in tokenizzati: for t in toks: vocabolario.setdefault(t, len(vocabolario)) n_doc, n_term = len(tokenizzati), len(vocabolario) tf = np.zeros((n_doc, n_term), dtype=np.float32) for i, toks in enumerate(tokenizzati): for t in toks: tf[i, vocabolario[t]] += 1.0 df = (tf > 0).sum(axis=0) idf = np.log((1.0 + n_doc) / (1.0 + df)).astype(np.float32) + 1.0 M = tf * idf norme = np.linalg.norm(M, axis=1, keepdims=True) return M / np.maximum(norme, 1e-9), vocabolario, idf def vettorizza_query(query, vocabolario, idf): v = np.zeros(len(vocabolario), dtype=np.float32) for t in _termini(query): if t in vocabolario: v[vocabolario[t]] += 1.0 v *= idf return v / max(float(np.linalg.norm(v)), 1e-9) # ====================================================================== # ESERCIZIO 1 - IL TETTO # # Il tetto è la percentuale di query per cui il contesto passato al modello # CONTIENE la risposta. È un limite superiore duro: se il tetto è 0.60, # nessun prompt, nessuna temperatura e nessun modello più grande porteranno # l'accuratezza sopra 0.60. Il resto sarebbe invenzione. # # Ti arrivano: # - `recuperati`: per ogni query, la lista ORDINATA dei testi dei chunk # restituiti dal recupero (il più simile per primo); # - `ancore`: per ogni query, la stringa che contiene la risposta; # - `k`: quanti di quei chunk finiscono davvero nel prompt. # # Devi restituire la frazione di query per cui ALMENO UNO dei primi k chunk # contiene l'ancora INTERA. # # Due trappole, entrambe reali: # - confronta sempre su testo NORMALIZZATO (usa `normalizza`): un a capo # dentro il docstring e uno spazio nel chunk sono lo stesso testo per un # lettore e due stringhe diverse per Python; # - "contiene l'ancora" non è "viene dal documento giusto". Un chunk può # venire dal documento giusto e aver perso per strada proprio la frase che # risponde: conterebbe come successo e non lo è. # ====================================================================== def tetto(recuperati, ancore, k): """Frazione di query per cui l'ancora è dentro uno dei primi k chunk.""" # TODO 1: scorri le query; per ciascuna guarda i primi k chunk e verifica # se almeno uno, normalizzato, contiene l'ancora normalizzata. # Restituisci colpi / numero di query. raise NotImplementedError("TODO 1") # ====================================================================== # ESERCIZIO 2 - L'ANCORA TAGLIATA, E QUANTO COSTA NON TAGLIARLA # # Una finestra fissa taglia il documento ogni `dim` token. Se il confine cade # dentro la frase che risponde, quella frase non esiste più INTERA in nessun # chunk: meta' sta nel chunk i, meta' nel chunk i+1, e il recupero non potrà # mai restituirla tutta. Il guasto è silenzioso - l'indice si costruisce # senza un errore - e si manifesta come una query che "stranamente non trova # mai la risposta". # # Il rimedio è la sovrapposizione: ogni finestra ripete le ultime # `sovrapposizione` parole della precedente, quindi una frase a cavallo del # confine ricompare intera nel chunk successivo. Il costo è un indice più # grande, ed è il numero che devi misurare per poter decidere. # # 2a: scrivi la finestra scorrevole. # 2b: conta, su una spazzata di dimensioni, quante ancore vengono tagliate con # e senza sovrapposizione, e di quanto cresce l'indice. # ====================================================================== def finestra_fissa(testo, dim, sovrapposizione): """Finestre di `dim` token che scorrono di `dim - sovrapposizione`.""" # TODO 2a: dividi il testo in parole; produci le finestre di `dim` parole # avanzando di `passo = dim - sovrapposizione` (mai meno di 1); # niente finestre vuote; se il testo è vuoto restituisci [testo]. raise NotImplementedError("TODO 2a") def spazzata_tagli(documenti, dimensioni, frazione_sovrapposizione=0.2): """Per ogni dimensione: ancore tagliate senza e con sovrapposizione, e numero di chunk nei due casi. Restituisce una lista di tuple (dim, tagliate_senza, tagliate_con, chunk_senza, chunk_con). """ # TODO 2b: per ogni dim, segmenta TUTTI i documenti nei due modi, poi per # ogni ancora controlla se esiste un chunk che la contiene intera # (sempre con `normalizza`). Conta quante NON ne hanno nessuno. raise NotImplementedError("TODO 2b") # ====================================================================== # ESERCIZIO 3 - LA VERIFICA CHE NON SI FIDA # # Il prompt chiede al modello di citare, per ogni frase, lo span da cui l'ha # presa: "... [S3]". Chiederlo non basta: il modello può citare uno span che # non dice quella cosa, oppure inventare un numero e attaccarci una citazione # plausibile. Serve un controllo che non creda alla citazione e la verifichi. # # La regola è volutamente grezza e per questo robusta: per ogni frase, prendi # gli span citati, ed esigi che ogni NUMERO e ogni NOME TECNICO della frase # compaia LETTERALMENTE dentro quegli span. Se manca, la risposta si rifiuta - # non si mostra con un avviso: si rifiuta. # # Cosa non fa, e va saputo: questa verifica garantisce che la risposta sia # ANCORATA al contesto, non che sia VERA. Se nel contesto è finito un # documento sbagliato, una risposta fedelmente sbagliata passa il controllo. # Contro quello serve il filtro sui metadati, non il verificatore. # # Formato: le citazioni sono "[S1]", "[S2]"... e gli indici partono da 1. # Se la risposta inizia con "NON NEL CONTESTO" è un'astensione: esito valido. # ====================================================================== NON_SO = "NON NEL CONTESTO" CITAZIONE = re.compile(r"\[S(\d+)\]") NUMERO = re.compile(r"-?\d+(?:\.\d+)?") # Un nome tecnico ha un punto, un trattino basso o una maiuscola interna. # Le parole inglesi comuni non ne hanno: è così che "According" non fa # rifiutare una risposta corretta. NOME = re.compile(r"\b(?:[A-Za-z_][A-Za-z0-9_]*\.[A-Za-z_][A-Za-z0-9_]*" r"|[A-Za-z]*[a-z][A-Z][A-Za-z]*" r"|[a-z]+_[a-z_]+)\b") # Il modello scrive spesso "... shorter than n. [S1]", con la citazione DOPO il # punto. Spezzando le frasi senza normalizzare, "[S1]" diventa un frammento a se' # e la frase prima risulta senza citazione: il verificatore rifiuterebbe una # risposta perfetta, e misurerebbe il proprio parser invece del modello. CODA_CITAZIONE = re.compile(r"([.!?])((?:\s*\[S\d+\])+)") def verifica_ancoraggio(risposta, span): """Restituisce "ancorata", "astenuta" oppure "rifiutata".""" # TODO 3: 1) astensione se la risposta inizia con NON_SO; # 2) normalizza con CODA_CITAZIONE (porta la citazione DENTRO la # frase, prima del punto), poi spezza in frasi su [.!?]; # 3) ogni frase deve avere almeno una citazione, e ogni indice # citato deve esistere (1 <= n <= len(span)); # 4) unisci i testi degli span citati, in minuscolo; # 5) togli le citazioni dalla frase, poi pretendi che ogni numero e # ogni nome tecnico della frase compaiano in quell'unione; # 6) al primo fallimento: "rifiutata". raise NotImplementedError("TODO 3") # ====================================================================== # ESECUZIONE # ====================================================================== def _recupera_tutto(documenti, dim=200, sovrapposizione=40, k_max=10): """Segmenta, indicizza e recupera: serve all'esercizio 1 e usa il 2a.""" chunk = [] for doc_id, testo in documenti.items(): for pezzo in finestra_fissa(testo, dim, sovrapposizione): chunk.append(pezzo) M, vocabolario, idf = vettorizza_lessicale(chunk) fuori = [] for _, domanda, _, _ in GOLDEN: punteggi = M @ vettorizza_query(domanda, vocabolario, idf) ordine = np.argsort(-punteggi)[:k_max] fuori.append([chunk[i] for i in ordine]) return fuori, len(chunk) def main(): print("Python %s | NumPy %s" % (sys.version.split()[0], np.__version__)) documenti = costruisci_corpus() parole = [len(t.split()) for t in documenti.values()] print("corpus: %d documenti, %d parole in totale, mediana %d per documento" % (len(documenti), sum(parole), int(np.median(parole)))) print("golden set: %d domande\n" % len(GOLDEN)) ancore = [a for _, _, _, a in GOLDEN] mancanti = [] # ---- esercizio 2 prima, perché l'1 usa la finestra ---- try: righe = spazzata_tagli(documenti, list(range(60, 420, 40))) print("ESERCIZIO 2 - l'ancora tagliata") print(" %6s %14s %12s %10s %9s" % ("dim", "tagliate s/s", "tagliate c/s", "chunk s/s", "chunk c/s")) ts = tc = cs = cc = 0 for dim, t_senza, t_con, c_senza, c_con in righe: print(" %6d %14d %12d %10d %9d" % (dim, t_senza, t_con, c_senza, c_con)) ts += t_senza; tc += t_con; cs += c_senza; cc += c_con prove = len(righe) * len(ancore) print(" totale: tagliate senza sovrapposizione %d/%d, con sovrapposizione %d/%d" % (ts, prove, tc, prove)) print(" costo della sovrapposizione: %d chunk contro %d, +%.1f%% di indice\n" % (cc, cs, 100.0 * (cc - cs) / max(cs, 1))) except NotImplementedError as e: mancanti.append(str(e)); print("ESERCIZIO 2 - non ancora svolto (%s)\n" % e) # ---- esercizio 1 ---- try: recuperati, n_chunk = _recupera_tutto(documenti) print("ESERCIZIO 1 - il tetto (indice da %d chunk)" % n_chunk) for k in (1, 3, 5, 10): print(" k = %2d -> tetto %.3f" % (k, tetto(recuperati, ancore, k))) print(" Leggilo così: con k = 3 il generatore non può superare quel") print(" valore, qualunque cosa tu scriva nel prompt. Il recuperatore di") print(" questo file è lessicale e le domande sono parafrasi: il tetto") print(" resta basso per costruzione. Con un embedding vero (vedi 3b in") print(" fondo) sale, e il codice che lo misura non cambia.\n") except NotImplementedError as e: mancanti.append(str(e)); print("ESERCIZIO 1 - non ancora svolto (%s)\n" % e) # ---- esercizio 3 ---- casi = [ ("tutto in regola", "The default cutoff is 0.6 [S1]. The maximum number returned is 3 [S1].", ["Optional arg cutoff (default 0.6) is a float in [0, 1]. Optional arg n " "(default 3) is the maximum number of close matches to return."], "ancorata"), ("numero inventato, citazione plausibile", "The default cutoff is 0.9 [S1].", ["Optional arg cutoff (default 0.6) is a float in [0, 1]."], "rifiutata"), ("frase senza citazione", "The default cutoff is 0.6.", ["Optional arg cutoff (default 0.6) is a float in [0, 1]."], "rifiutata"), ("span citato inesistente", "The default cutoff is 0.6 [S7].", ["Optional arg cutoff (default 0.6) is a float in [0, 1]."], "rifiutata"), ("nome tecnico non presente nello span citato", "The function statistics.fmean returns the arithmetic mean [S1].", ["Return the harmonic mean of data."], "rifiutata"), ("citazione dopo il punto finale", "The last batch may be shorter than n. [S1]", ["Batch data into tuples of length n. The last batch may be shorter than n."], "ancorata"), ("astensione dichiarata", "NON NEL CONTESTO", ["Un testo che non c'entra nulla con la domanda."], "astenuta"), ("fedele a uno span sbagliato: passa, e deve passare", "The default cutoff is 0.9 [S1].", ["NOTE (draft): the default cutoff of get_close_matches is 0.9."], "ancorata"), ] try: verifica_ancoraggio("NON NEL CONTESTO", ["prova"]) print("ESERCIZIO 3 - la verifica che non si fida") ok = 0 for etichetta, risposta, span, atteso in casi: esito = verifica_ancoraggio(risposta, span) segno = "ok " if esito == atteso else "NO " ok += esito == atteso print(" %s %-44s atteso %-9s ottenuto %s" % (segno, etichetta, atteso, esito)) print(" %d casi su %d.\n" % (ok, len(casi))) except NotImplementedError as e: mancanti.append(str(e)); print("ESERCIZIO 3 - non ancora svolto (%s)\n" % e) if mancanti: print("Mancano ancora: %s" % ", ".join(mancanti)) else: print("Tutti gli esercizi sono svolti.") print("Nota sull'ultimo caso dell'esercizio 3: la risposta è FALSA e il") print("verificatore la accetta, perché è fedele allo span che le è stato") print("dato. Ancoraggio e verità sono due proprietà diverse, e si") print("difendono con due meccanismi diversi.") # ====================================================================== # SOLUZIONI COMMENTATE # ====================================================================== # # --- TODO 1 ------------------------------------------------------------ # def tetto(recuperati, ancore, k): # colpi = 0 # for chunk_query, ancora in zip(recuperati, ancore): # bersaglio = normalizza(ancora) # if any(bersaglio in normalizza(c) for c in chunk_query[:k]): # colpi += 1 # return colpi / len(ancore) # # Perché `normalizza` su entrambi i lati: l'ancora viene dal docstring, dove # ci sono a capo e rientri; il chunk viene da uno split su spazi bianchi, dove # non ce ne sono più. Senza normalizzare, il tetto risulta vicino a zero e si # passa un pomeriggio a cercare un bug nel recupero che è nel confronto. # # --- TODO 2a ----------------------------------------------------------- # def finestra_fissa(testo, dim, sovrapposizione): # parole = testo.split() # if not parole: # return [testo] # passo = max(1, dim - sovrapposizione) # pezzi = [" ".join(parole[i:i + dim]) # for i in range(0, len(parole), passo) # if parole[i:i + dim]] # return pezzi or [testo] # # `max(1, ...)` non è pedanteria: con sovrapposizione >= dim il passo sarebbe # zero o negativo e `range` girerebbe all'infinito riempiendo la memoria. # # --- TODO 2b ----------------------------------------------------------- # def spazzata_tagli(documenti, dimensioni, frazione_sovrapposizione=0.2): # ancore = [normalizza(a) for _, _, _, a in GOLDEN] # righe = [] # for dim in dimensioni: # sovr = int(dim * frazione_sovrapposizione) # senza, con = [], [] # for testo in documenti.values(): # senza += finestra_fissa(testo, dim, 0) # con += finestra_fissa(testo, dim, sovr) # n_senza = [normalizza(c) for c in senza] # n_con = [normalizza(c) for c in con] # t_senza = sum(1 for a in ancore if not any(a in c for c in n_senza)) # t_con = sum(1 for a in ancore if not any(a in c for c in n_con)) # righe.append((dim, t_senza, t_con, len(senza), len(con))) # return righe # # Il risultato tipico: le ancore tagliate sono poche in percentuale, ma non # zero, e quelle poche sono IRRECUPERABILI - non è un calo graduale di # qualità, è un insieme di query che non funzionera' mai. La sovrapposizione # le azzera e costa circa un quinto di indice in più. # # --- TODO 3 ------------------------------------------------------------ # def verifica_ancoraggio(risposta, span): # if risposta.strip().startswith(NON_SO): # return "astenuta" # testo = CODA_CITAZIONE.sub(lambda m: m.group(2) + m.group(1), risposta) # frasi = [f.strip() for f in re.split(r"(?<=[.!?])\s+", testo) if f.strip()] # if not frasi: # return "rifiutata" # for frase in frasi: # citati = CITAZIONE.findall(frase) # if not citati: # return "rifiutata" # testi = [] # for c in citati: # n = int(c) # if not (1 <= n <= len(span)): # return "rifiutata" # testi.append(span[n - 1]) # unione = " ".join(testi).lower() # nudo = CITAZIONE.sub("", frase) # for numero in NUMERO.findall(nudo): # if numero.lower() not in unione: # return "rifiutata" # for nome in NOME.findall(nudo): # if nome.lower() not in unione: # return "rifiutata" # return "ancorata" # # `CITAZIONE.sub("", frase)` prima di cercare i numeri è obbligatorio: senza, # il "3" di "[S3]" verrebbe cercato dentro lo span e la risposta verrebbe # rifiutata per colpa della sua stessa citazione. # # --- 3b, quando hai Ollama in ascolto ---------------------------------- # Sostituisci `vettorizza_lessicale` con gli embedding veri e rimisura il tetto: # # import json, urllib.request # def incorpora(testi): # req = urllib.request.Request( # "http://localhost:11434/api/embed", # data=json.dumps({"model": "nomic-embed-text", "input": testi}).encode(), # headers={"Content-Type": "application/json"}) # with urllib.request.urlopen(req, timeout=600) as r: # V = np.asarray(json.load(r)["embeddings"], dtype=np.float32) # return V / np.maximum(np.linalg.norm(V, axis=1, keepdims=True), 1e-9) # # Il punto dell'esercizio: il tetto sale, il CODICE CHE LO MISURA non cambia. # Se cambiando recuperatore devi riscrivere la valutazione, la valutazione era # legata all'implementazione e non al problema. if __name__ == "__main__": main()