Leçon 2 — Construire pas à pas

BM25 et recherche vectorielle

5 min de lecture

BM25 et recherche vectorielle

Étape 1 : une tokenisation qui ne casse pas les codes

BM25 compare des jetons (tokens), c'est-à-dire des morceaux de texte. La tokenisation naïve text.split() garde la ponctuation (HEL-4040,), et un découpage sur tout ce qui n'est pas une lettre coupe HEL-4040 en hel + 4040 : le jeton rare disparaît.

files/hybrid_rag.py
def tokenize(text: str) -> list[str]:
    """Minuscules, sans accents, en gardant les identifiants entiers (hel-6030, sync.batch_size)."""
    text = unicodedata.normalize("NFKD", text.lower()).encode("ascii", "ignore").decode()
    return [t for t in re.findall(r"[a-z0-9_]+(?:[.-][a-z0-9_]+)*", text) if t not in STOPWORDS]
Entréesplit()Ce tokenize
Erreur HEL-6030,Erreur, HEL-6030,erreur, hel-6030
sync.batch_sizesync.batch_sizesync.batch_size
étéétéete

Les mots vides (stopwords : « le », « de », « comment »…) sont retirés : ils sont partout et ne discriminent rien.

flowchart LR
  T["« J'obtiens HEL-4040 ! »"] --> N["minuscules + sans accents"]
  N --> R["regex : mots et identifiants"]
  R --> S["retrait des mots vides"]
  S --> O["['obtiens', 'hel-4040']"]

Que se passerait-il pour la question « HEL-4040 » si la tokenisation coupait sur le tiret ?

Étape 2 : l'index BM25

La bibliothèque rank-bm25 fait tout le calcul. On lui donne la liste des documents tokenisés :

files/hybrid_rag.py
self.bm25 = BM25Okapi([tokenize(d["text"]) for d in docs])

def bm25_search(bm25: BM25Okapi, query: str, k: int = 20) -> list[int]:
    scores = bm25.get_scores(tokenize(query))
    return [int(i) for i in np.argsort(-scores)[:k] if scores[i] > 0]

On ne garde que les documents de score strictement positif : un document qui ne partage aucun jeton avec la question n'a rien à faire dans le classement.

Étape 3 : les embeddings Gemini

files/hybrid_rag.py
def embed(client, texts: list[str]) -> np.ndarray:
    out = []
    for i in range(0, len(texts), 50):
        contents = [types.Content(parts=[types.Part(text=t)]) for t in texts[i:i + 50]]
        r = with_retry(lambda: client.models.embed_content(
            model=EMBED_MODEL, contents=contents,
            config=types.EmbedContentConfig(output_dimensionality=EMBED_DIM)))
        out += [e.values for e in r.embeddings]
    v = np.array(out, dtype=np.float32)
    return v / np.linalg.norm(v, axis=1, keepdims=True)

Trois choix à comprendre :

  1. output_dimensionality=768 : gemini-embedding-2 produit 3072 dimensions par défaut ; on en garde 768, soit 4 fois moins de mémoire, largement assez pour 80 documents.
  2. Normalisation : une fois chaque vecteur ramené à une norme de 1, la similarité cosinus devient un simple produit scalaire doc_vecs @ q_vec.
  3. Un Content par texte : voir l'avertissement ci-dessous.

Les embeddings du corpus sont mis en cache dans .cache/docs-<empreinte>.npy : l'empreinte change si le modèle, la dimension ou un document change.

files/hybrid_rag.py
def vector_search(doc_vecs: np.ndarray, q_vec: np.ndarray, k: int = 20) -> list[int]:
    return [int(i) for i in np.argsort(-(doc_vecs @ q_vec))[:k]]

Pourquoi normaliser les vecteurs avant de les stocker ?

Résultat obtenu

Comparaison des deux moteurs seuls sur trois questions (exécution du 2026-10-15, 3 premiers identifiants renvoyés) :

HEL-6010 apparaît chaque nuit, pourquoi ?
  bm25   ['err-HEL-6010', 'faq-regions', 'par-sync.timeout_s']
  vector ['err-HEL-7010', 'err-HEL-7030', 'err-HEL-7020']      ← piégé par « nuit » (archivage nocturne)
Comment vider la mémoire temporaire d'une source de données qui renvoie des infos dépassées ?
  bm25   ['err-HEL-4040', 'err-HEL-6040', 'err-HEL-7040']      ← aucun mot utile en commun
  vector ['proc-purger-cache', 'proc-schema-evolution', 'err-HEL-6050']
Peut-on cacher les numéros de sécurité sociale dans les tableaux partagés ?
  bm25   ['err-HEL-4020', 'err-HEL-6020', 'err-HEL-7020']
  vector ['proc-masquage', 'note-3-3-0', 'proc-audit']

Chaque moteur réussit là où l'autre échoue : c'est exactement ce que la fusion va exploiter.

Exercice : tester la tokenisation

Écrivez un test qui vérifie que tokenize("Le paramètre export.timeout_s vaut 30 !") renvoie ["parametre", "export.timeout_s", "vaut", "30"], puis lancez pytest -q.

Voir la solution
files/test_hybrid_rag.py
def test_tokenize_parametre():
    assert tokenize("Le paramètre export.timeout_s vaut 30 !") == ["parametre", "export.timeout_s", "vaut", "30"]

« Le » est un mot vide, « paramètre » perd son accent, le point dans export.timeout_s est conservé parce qu'il est suivi d'un caractère de mot, et « ! » disparaît. Le fichier fourni contient un test du même genre : test_tokenize_garde_les_identifiants.

Je sais faire

  • Écrire une tokenisation qui préserve les identifiants techniques
  • Construire un index BM25Okapi et l'interroger
  • Calculer des embeddings gemini-embedding-2 réduits à 768 dimensions, un vecteur par texte
  • Faire une recherche cosinus par produit matriciel
Suivant