04 - RAG avanzato: re-ranking, hybrid search, chunk strategy

Il recupero dell'articolo precedente sbagliava cinque domande su quindici: il passaggio giusto non era in cima. Davanti a un numero così, la tentazione è provare tutte le tecniche che si leggono nei blog (ordinamento BM25, ricerca ibrida, riscrittura della domanda, riordino con un secondo modello) e tenere quella che sembra migliore. Il rischio è scegliere per fortuna: con quindici domande, una sola risposta che cambia sposta il hit@1 di 6,7 punti percentuali. Questo articolo prova quattro interventi sullo stesso corpus (più un quinto, il riordino, descritto senza eseguirlo) e dichiara per ciascuno quale errore dovrebbe curare, quanto lo cura davvero, e quando una differenza di un'unità non è una prova. Il risultato anticipa la conclusione: sul corpus di prova nessuna tecnica «sofisticata» fa molto, perché l'errore che resta è un problema di vocabolario, e le tecniche che lo curano devono portare conoscenza che il corpus non ha.

I listati sono stati eseguiti due volte con uscita identica con numpy e scikit-learn 1.9.0; il confronto con la libreria BM25 è stato fatto con rank-bm25 0.2.2, installato in un ambiente a parte. Nessun modello è stato scaricato o eseguito. Serve il modulo corpus.py dell'articolo 03 - RAG fondamentale, da salvare accanto ai file di questo.

Che cosa misurare e quanto fidarsi

Il modulo qui sotto è l'infrastruttura comune. tokens separa il testo in parole minuscole e, a richiesta, toglie un elenco di parole funzionali scritto a mano e volutamente corto. La classe BM25 è l'ordinamento che si introduce nella sezione seguente. valuta prende una funzione che dà un punteggio a ciascun passaggio per una domanda e restituisce hit@1, hit@3, il rango reciproco medio (MRR) e il rango del passaggio giusto per ogni domanda; a parità di punteggio vince l'indice più basso, per avere un risultato ripetibile. Il file non stampa nulla: i risultati compaiono nei listati successivi.

import math
import re
from collections import Counter

import numpy as np
from corpus import PASSAGGI, DOMANDE

TOKEN = re.compile(r"(?u)\b\w\w+\b")
# Elenco scritto a mano, volutamente corto: articoli, preposizioni, pronomi e ausiliari più frequenti.
FUNZIONALI = set("""il lo la i gli le un una uno di a da in con su per tra fra e o ma che chi si è non al allo alla ai
alle del dello della dei delle dal dalla dai nel nella nei nelle sul sulla ho hai ha mio mia miei quanto quanti quale
quali come cui ne ci sono può deve devono va vanno è""".split())


def tokens(testo, senza_funzionali=False):
    t = TOKEN.findall(testo.lower())
    return [w for w in t if w not in FUNZIONALI] if senza_funzionali else t


class BM25:
    """BM25 con l'idf della libreria rank_bm25 (con pavimento a 0,25 volte l'idf medio): k1=1,5, b=0,75."""

    def __init__(self, corpus_tokens, k1=1.5, b=0.75, eps=0.25):
        self.k1, self.b = k1, b
        self.tf = [Counter(d) for d in corpus_tokens]
        self.len = np.array([len(d) for d in corpus_tokens], dtype=float)
        self.avgdl = self.len.mean()
        n = len(corpus_tokens)
        df = Counter(w for d in self.tf for w in d)
        idf = {w: math.log(n - f + 0.5) - math.log(f + 0.5) for w, f in df.items()}
        medio = sum(idf.values()) / len(idf)
        self.idf = {w: (v if v >= 0 else eps * medio) for w, v in idf.items()}

    def punteggi(self, query_tokens):
        s = np.zeros(len(self.tf))
        norma = self.k1 * (1 - self.b + self.b * self.len / self.avgdl)
        for w in query_tokens:
            f = np.array([d.get(w, 0) for d in self.tf], dtype=float)
            s += self.idf.get(w, 0.0) * f * (self.k1 + 1) / (f + norma)
        return s


def ranghi_da_punteggi(s):
    """Posizione (1 = migliore) di ogni passaggio; a parità vince l'indice più basso."""
    ordine = np.argsort(-s, kind="stable")
    r = np.empty(len(s), dtype=int)
    r[ordine] = np.arange(1, len(s) + 1)
    return r


def valuta(punteggi_di):
    """punteggi_di(domanda) -> un punteggio per passaggio. Restituisce hit@1, hit@3, MRR e il rango di ogni domanda."""
    ranghi = [int(ranghi_da_punteggi(punteggi_di(d))[gold]) for d, gold in DOMANDE]
    return sum(r == 1 for r in ranghi), sum(r <= 3 for r in ranghi), float(np.mean([1 / r for r in ranghi])), ranghi

La regola di lettura vale per tutto l'articolo: quindici domande sono una prova dimostrativa, non una valutazione. Un cambio di un'unità in hit@1 (9 contro 10) corrisponde a una domanda e non basta a dire che un metodo è migliore; un cambio di quattro unità (9 contro 13) è un segnale, ma resta una misura su un corpus inventato. Per un corpus vero servono centinaia di domande, scritte da chi le farà davvero, con il passaggio giusto indicato da una persona.

BM25: la ripetizione non vale in modo lineare

Il TF-IDF dell'articolo precedente premia in modo proporzionale le parole che si ripetono: un passaggio che dice «copie» due volte ha un punteggio doppio per quella parola, anche se la seconda ripetizione non aggiunge pertinenza. BM25 (una funzione di ordinamento classica della ricerca di testo) corregge due cose: la ripetizione di una parola satura (con k₁ = 1,5 il contributo cresce sempre meno) e un passaggio più lungo della media viene penalizzato in funzione di b = 0,75. La classe del modulo usa la stessa definizione di idf della libreria rank_bm25, compreso il pavimento a 0,25 volte l'idf medio per le parole che compaiono in più di metà dei passaggi.

Il listato calcola i punteggi due volte, a mano e con BM25Okapi, e stampa lo scarto massimo su tutte le domande; poi valuta il recupero con e senza le parole funzionali. Se l'implementazione a mano fosse sbagliata, lo scarto non sarebbe dell'ordine dell'arrotondamento. Il secondo giro risponde a una domanda lasciata aperta: togliere «una», «per» e «con» cura il primo difetto dell'articolo precedente, quello dell'albergo?

import numpy as np
from rank_bm25 import BM25Okapi
from corpus import PASSAGGI, DOMANDE
from metriche import BM25, tokens, valuta

testi = [t for _, _, t in PASSAGGI]
print(f"{'configurazione':>34} {'hit@1':>6} {'hit@3':>6} {'MRR':>6}")
for senza in (False, True):
    corpus = [tokens(t, senza) for t in testi]
    mio = BM25(corpus)
    lib = BM25Okapi(corpus)
    # controllo: i punteggi a mano coincidono con quelli della libreria su tutte le domande
    scarto = max(np.abs(mio.punteggi(tokens(d, senza)) - lib.get_scores(tokens(d, senza))).max() for d, _ in DOMANDE)
    h1, h3, mrr, ranghi = valuta(lambda d: mio.punteggi(tokens(d, senza)))
    nome = "BM25, senza parole funzionali" if senza else "BM25, tutte le parole"
    print(f"{nome:>34} {h1:>6} {h3:>6} {mrr:>6.3f}   (scarto con rank_bm25: {scarto:.1e})")
    print("   ranghi del passaggio giusto:", ranghi)

# Output:
#                     configurazione  hit@1  hit@3    MRR
#              BM25, tutte le parole      9     14  0.772   (scarto con rank_bm25: 6.2e-15)
#    ranghi del passaggio giusto: [1, 2, 1, 2, 2, 1, 1, 1, 1, 1, 3, 1, 2, 1, 4]
#      BM25, senza parole funzionali      9     13  0.746   (scarto con rank_bm25: 8.9e-16)
#    ranghi del passaggio giusto: [1, 2, 1, 9, 2, 1, 1, 1, 1, 1, 12, 1, 2, 1, 2]

Lo scarto con la libreria è di 10^-15: il BM25 a mano è corretto. Con tutte le parole hit@3 sale da 13 (TF-IDF) a 14 e il passaggio dell'albergo passa dal rango 4 al 2 (e resta al 2 anche senza parole funzionali, quindi lo ha curato BM25 e non la rimozione), ma hit@1 resta 9 (contro 10 del TF-IDF, una domanda di differenza), quindi nessun guadagno dimostrabile. Togliere le parole funzionali ha un effetto che non mi aspettavo e che l'esperimento smentisce: il risultato peggiora (MRR 0,746 contro 0,772), perché i passaggi giusti delle domande 4 e 11 (rango da 2 a 9 e da 3 a 12) li trovava soprattutto «con» e «che», parole che non avevano niente a che fare con la risposta. Il vecchio punteggio era giusto per un caso fortunato, e la rimozione lo ha tolto senza aggiungere sinonimi. Un passaggio che si salva per le parole sbagliate è un recupero fragile, non un recupero migliore.

Ricerca ibrida: fondere due ordinamenti

La ricerca ibrida combina un ordinamento lessicale (BM25) con uno denso (qui la LSA a 12 dimensioni dell'articolo precedente, senza nessun modello neurale). L'idea è che i due sbaglino in modo diverso: il lessicale non vede i sinonimi ma è preciso sulle parole rare, il denso vede qualche vicinanza ma può confondere termini specifici. Le due strade per fonderli sono la media dei punteggi normalizzati e la Reciprocal Rank Fusion di Cormack, Clarke e Büttcher (2009): il punteggio di un passaggio è la somma di 1/(k + rango) su tutti gli ordinamenti. Gli autori scrivono che k = 60 è stato fissato in un'indagine pilota e che la scelta non era critica. La formula usa solo i ranghi e non i punteggi, quindi non richiede di rendere confrontabili scale diverse; il valore 60 è un'impostazione loro, non una legge.

Il listato valuta cinque metodi sulle stesse quindici domande e stampa, per ciascuno, il rango del passaggio giusto per ogni domanda: guardare i ranghi per domanda e non solo la media è il modo per vedere se due metodi sbagliano sulle stesse domande.

import numpy as np
from sklearn.decomposition import TruncatedSVD
from sklearn.feature_extraction.text import TfidfVectorizer
from corpus import PASSAGGI, DOMANDE
from metriche import BM25, tokens, valuta, ranghi_da_punteggi

testi = [t for _, _, t in PASSAGGI]
bm25 = BM25([tokens(t, True) for t in testi])
vec = TfidfVectorizer(tokenizer=lambda t: tokens(t, True), lowercase=False, token_pattern=None)
X = vec.fit_transform(testi)
svd = TruncatedSVD(n_components=12, random_state=0).fit(X)
P = svd.transform(X)
P /= np.linalg.norm(P, axis=1, keepdims=True)


def lessicale(d):
    return bm25.punteggi(tokens(d, True))


def denso(d):
    q = svd.transform(vec.transform([d]))[0]
    return P @ q / (np.linalg.norm(q) + 1e-12)


def rrf(d, k=60):
    """Reciprocal Rank Fusion: somma di 1/(k + rango) su tutti gli ordinamenti da fondere."""
    return sum(1.0 / (k + ranghi_da_punteggi(f(d))) for f in (lessicale, denso))


def pesata(d, alfa=0.5):
    """Fusione dei punteggi: ciascun punteggio normalizzato fra 0 e 1 sui passaggi, poi media pesata."""
    def norma(s):
        return (s - s.min()) / (s.max() - s.min() + 1e-12)
    return alfa * norma(lessicale(d)) + (1 - alfa) * norma(denso(d))


print(f"{'metodo':>26} {'hit@1':>6} {'hit@3':>6} {'MRR':>6}")
for nome, f in (("BM25", lessicale), ("denso (LSA, 12 dim.)", denso), ("RRF (k = 60)", rrf),
                ("media pesata, alfa = 0,5", pesata), ("media pesata, alfa = 0,8", lambda d: pesata(d, 0.8))):
    h1, h3, mrr, ranghi = valuta(f)
    print(f"{nome:>26} {h1:>6} {h3:>6} {mrr:>6.3f}   ranghi: {ranghi}")

# Output:
#                     metodo  hit@1  hit@3    MRR
#                       BM25      9     13  0.746   ranghi: [1, 2, 1, 9, 2, 1, 1, 1, 1, 1, 12, 1, 2, 1, 2]
#       denso (LSA, 12 dim.)      9     14  0.772   ranghi: [1, 2, 1, 2, 1, 1, 1, 1, 1, 1, 12, 1, 2, 2, 2]
#               RRF (k = 60)      9     14  0.761   ranghi: [1, 2, 1, 3, 1, 1, 1, 1, 1, 1, 12, 1, 2, 2, 2]
#   media pesata, alfa = 0,5      9     14  0.772   ranghi: [1, 2, 1, 2, 2, 1, 1, 1, 1, 1, 12, 1, 2, 1, 2]
#   media pesata, alfa = 0,8      9     14  0.772   ranghi: [1, 2, 1, 2, 2, 1, 1, 1, 1, 1, 12, 1, 2, 1, 2]

Nessuna fusione batte il metodo denso da solo: hit@1 resta 9 per tutti, e l'MRR va da 0,746 (BM25) a 0,772 (denso, media pesata) con RRF a 0,761. Le differenze sono al più di una domanda su quindici (hit@3: 13 contro 14) e di 0,026 in MRR. La causa si legge nei ranghi: sulla domanda 11 (i «backup» che nel corpus si chiamano «copie») tutti i metodi mettono il passaggio giusto al rango 12, perché né BM25 né la LSA su 18 passaggi conoscono il sinonimo, e una fusione di due ordinamenti che sbagliano insieme sbaglia insieme. La ricerca ibrida aiuta quando i due ordinamenti hanno errori diversi; qui il denso corregge il lessicale sulle domande 4 e 5 (ranghi da 9 a 2 e da 2 a 1), ma lo peggiora sulla 14 (da 1 a 2), e la RRF eredita i primi due guadagni solo in parte (domanda 4: rango 3). Con un modello di embedding preaddestrato i due ordinamenti sarebbero probabilmente più diversi, ma questo articolo non lo ha eseguito e non lo afferma.

Riscrivere la domanda, e HyDE

Se l'errore sta nel vocabolario, si può cambiare la domanda prima di cercare. La riscrittura (query rewriting) chiede a un modello linguistico di riformulare la domanda, o di aggiungere termini; HyDE, proposto da Gao, Ma, Lin e Callan (2022), chiede al modello di scrivere un documento ipotetico che risponde alla domanda, e cerca i passaggi vicini a quel documento, che può contenere dettagli falsi ma porta il tipo di parole che un passaggio vero conterrebbe. Entrambe le tecniche richiedono una chiamata a un modello per ogni domanda: costo, latenza e un modello che può aggiungere termini sbagliati.

Nessun modello viene eseguito qui. Il listato simula la riscrittura con un dizionario di sinonimi che ho scritto a mano dopo aver visto i fallimenti, e lo applica alle domande: aggiunge alla domanda le parole del documento (per esempio «telefono» per «cellulare» e «copie» per «backup»). Il numero che ne esce non è una stima dell'efficacia di nessun modello: è un limite superiore costruito, che dice quanto varrebbe una riscrittura perfetta sulle parole giuste.

from corpus import PASSAGGI, DOMANDE
from metriche import BM25, tokens, valuta

testi = [t for _, _, t in PASSAGGI]
bm25 = BM25([tokens(t, True) for t in testi])

# Riscrittura della domanda: SIMULATA con un dizionario scritto a mano dopo aver visto i fallimenti.
# Non misura nessun modello: mostra solo che cosa succede se la domanda porta le parole del documento.
SINONIMI = {
    "cellulare": ["telefono"], "codici": ["secondo", "fattore"], "backup": ["copie", "ripristinare"],
    "albergo": ["pernottamento"], "tetto": ["fino", "rimborsato"], "portatile": ["computer"],
    "automobile": ["auto"], "dentista": ["visite", "mediche"], "respinta": ["scartata"],
    "cancellato": ["ripristino"], "conservate": ["conservano"],
}


def espandi(domanda):
    t = tokens(domanda, True)
    return t + [s for w in t for s in SINONIMI.get(w, [])]


for nome, q in (("domanda originale", lambda d: tokens(d, True)), ("domanda con sinonimi", espandi)):
    h1, h3, mrr, ranghi = valuta(lambda d: bm25.punteggi(q(d)))
    print(f"{nome:>22}: hit@1 {h1}  hit@3 {h3}  MRR {mrr:.3f}  ranghi {ranghi}")

# Output:
#      domanda originale: hit@1 9  hit@3 13  MRR 0.746  ranghi [1, 2, 1, 9, 2, 1, 1, 1, 1, 1, 12, 1, 2, 1, 2]
#   domanda con sinonimi: hit@1 13  hit@3 15  MRR 0.933  ranghi [1, 1, 1, 2, 1, 1, 1, 1, 1, 1, 1, 1, 2, 1, 1]

Con il dizionario, hit@1 passa da 9 a 13 e l'MRR da 0,746 a 0,933; solo le domande 4 e 13 restano al rango 2. È il guadagno più grande dell'articolo, e proprio per questo è quello meno credibile: il dizionario è stato costruito guardando le risposte. Una riscrittura fatta da un modello vero può aggiungere i termini giusti e può anche aggiungerne di sbagliati, e la sua efficacia va misurata su domande che non ha visto. Il messaggio è un altro: quando il recupero fallisce per vocabolario, la leva è portare le parole del documento nella domanda (o la domanda nello spazio dei documenti, come fanno gli embedding), non rifinire l'ordinamento.

Unità piccole, risposta ampia

La scelta dell'unità da indicizzare ha un compromesso: un'unità piccola (una frase) somiglia di più a una domanda, un'unità grande ha il contesto per rispondere. Il recupero con documento genitore (parent document retrieval) prende il meglio dei due: si indicizzano le frasi, ma quando una frase vince si restituisce il passaggio che la contiene. Il listato divide i 18 passaggi in 32 frasi (15,8 parole in media contro 28,2), indicizza le frasi con BM25, assegna a ogni passaggio il punteggio della sua frase migliore, e confronta il risultato con l'indice sui passaggi interi.

import re
import numpy as np
from corpus import PASSAGGI, DOMANDE
from metriche import BM25, tokens, valuta

testi = [t for _, _, t in PASSAGGI]

# Indice di piccole unità (frasi), ciascuna col riferimento al passaggio che la contiene.
frasi, padre = [], []
for i, t in enumerate(testi):
    for f in re.split(r"(?<=[.;])\s+", t):
        frasi.append(f)
        padre.append(i)
padre = np.array(padre)
print(f"{len(testi)} passaggi, {len(frasi)} frasi; parole per passaggio {np.mean([len(t.split()) for t in testi]):.1f}, per frase {np.mean([len(f.split()) for f in frasi]):.1f}")

bm_passaggi = BM25([tokens(t, True) for t in testi])
bm_frasi = BM25([tokens(f, True) for f in frasi])


def via_frasi(domanda):
    s = bm_frasi.punteggi(tokens(domanda, True))
    return np.array([s[padre == i].max() for i in range(len(testi))])     # il passaggio vale quanto la sua frase migliore


for nome, f in (("indice sui passaggi", lambda d: bm_passaggi.punteggi(tokens(d, True))),
                ("indice sulle frasi, restituisce il passaggio", via_frasi)):
    h1, h3, mrr, ranghi = valuta(f)
    print(f"{nome:>46}: hit@1 {h1}  hit@3 {h3}  MRR {mrr:.3f}")

# Output:
# 18 passaggi, 32 frasi; parole per passaggio 28.2, per frase 15.8
#                            indice sui passaggi: hit@1 9  hit@3 13  MRR 0.746
#   indice sulle frasi, restituisce il passaggio: hit@1 10  hit@3 13  MRR 0.780

L'indice sulle frasi migliora di poco (hit@1 da 9 a 10, MRR da 0,746 a 0,780): una domanda in più, quindi sotto la soglia del segnale. Il vantaggio atteso è comunque reale in linea di principio: la frase in cima concentra il confronto sulle parole che contano, e il passaggio intero resta disponibile per costruire il prompt. Il prezzo è un indice con quasi il doppio delle voci e una mappa da mantenere fra frasi e genitori.

Riordinare con un cross-encoder

Il riordino (re-ranking) è una seconda fase: la prima fase recupera molti candidati con un metodo economico (BM25, fusione, vettori), la seconda li ordina con un modello più costoso che legge insieme la domanda e ciascun candidato (un cross-encoder). La ragione del doppio passaggio è il costo, che Reimers e Gurevych quantificano per il caso peggiore: leggere ogni coppia con una sola rete su 10.000 frasi richiede 49.995.000 inferenze, circa 65 ore su una GPU V100. Riordinare 20 candidati per domanda richiede 20 inferenze. Questo articolo non esegue nessun cross-encoder, perché serve scaricare un modello; il blocco è uno schema da adattare, e la sintassi di CrossEncoder e di predict (che riceve un elenco di coppie) segue la documentazione di Sentence Transformers consultata oggi.

# SCHEMA DA ADATTARE, NON ESEGUITO: richiede di scaricare un modello di riordino.
from sentence_transformers import CrossEncoder


def riordina(domanda, candidati, modello, k=3):
    """`candidati` sono i passaggi della prima fase (per esempio i primi 20 della fusione)."""
    coppie = [(domanda, c) for c in candidati]
    punteggi = modello.predict(coppie)             # una inferenza per ogni coppia (domanda, passaggio)
    ordine = sorted(range(len(candidati)), key=lambda i: -punteggi[i])
    return [candidati[i] for i in ordine[:k]]


modello = CrossEncoder("NOME-DEL-MODELLO-DI-RIORDINO")   # da scegliere e da valutare sui propri dati

Lo schema fa tre cose: costruisce le coppie, chiede i punteggi con predict (una inferenza per coppia) e ordina. Il nome del modello è volutamente un segnaposto, perché il modello va scelto e valutato sui propri dati e sulla propria lingua: un modello di riordino allenato su domande in inglese non garantisce nulla per un corpus italiano. Non ho misurato nessun guadagno e non lo dichiaro.

Che cosa sai fare adesso

Sai misurare un recupero con hit@k e MRR e leggere i ranghi per domanda, sai verificare un BM25 scritto a mano contro una libreria, sai fondere due ordinamenti con la RRF e riconoscere quando la fusione non può aiutare, e sai distinguere un guadagno da una domanda di differenza. Sai anche descrivere riscrittura, HyDE e riordino con cross-encoder con il loro costo, senza averli eseguiti. Il prossimo articolo passa dal recupero all'azione: 05 - Agenti AI: ReAct, tool-use, planning. Se ti serve prima il recupero di base, parti da 03 - RAG fondamentale; l'indice è nella pagina della serie.

Più avanti nella serie trovi anche 08 - Inferenza on-prem con Ollama, vLLM, llama.cpp, che stima la memoria e il tempo di esecuzione del modello che riceve il contesto recuperato.