Recherche hybride (sémantique + lexicale)

Objectifs

• Comprendre pourquoi combiner recherche sémantique et lexicale
• Implémenter une recherche hybride avec Reciprocal Rank Fusion
• Utiliser la recherche hybride dans les bases vectorielles

Le problème : aucune approche n’est parfaite

La recherche sémantique excelle sur les synonymes mais peut échouer sur des termes techniques précis. La recherche lexicale fait l’inverse :

# Requête : "erreur 503 nginx"
# Sémantique → trouve "problèmes de serveur web" (bien) mais pas "code 503" (manqué)
# Lexicale   → trouve "erreur 503 nginx" exactement (bien) mais pas "timeout du reverse proxy" (manqué)
# Hybride    → trouve les deux ✅

Recherche lexicale avec BM25

BM25 est l’algorithme standard de la recherche par mots-clés :

from rank_bm25 import BM25Okapi
import re

def tokeniser(texte: str) -> list[str]:
    """Tokenisation simple pour le français."""
    texte = texte.lower()
    tokens = re.findall(r"\b\w+\b", texte)
    # Supprimer les stop words courants
    stop_words = {"le", "la", "les", "de", "du", "des", "un", "une",
                  "et", "ou", "est", "dans", "pour", "avec", "sur"}
    return [t for t in tokens if t not in stop_words]

documents = [
    "Comment résoudre l'erreur 503 dans nginx",
    "Configuration du reverse proxy avec timeout",
    "Guide d'installation de nginx sur Ubuntu",
    "Les bases de la cuisine française",
]

# Indexer avec BM25
corpus_tokenise = [tokeniser(doc) for doc in documents]
bm25 = BM25Okapi(corpus_tokenise)

# Rechercher
requete_tokens = tokeniser("erreur 503 nginx")
scores_bm25 = bm25.get_scores(requete_tokens)

for i, score in enumerate(scores_bm25):
    print(f"BM25 {score:.3f} | {documents[i]}")

Reciprocal Rank Fusion (RRF)

RRF combine les classements de plusieurs systèmes de recherche. Chaque document reçoit un score basé sur son rang dans chaque liste :

def reciprocal_rank_fusion(
    resultats_listes: list[list[tuple[int, float]]],
    k: int = 60
) -> list[tuple[int, float]]:
    """Fusionne plusieurs listes de résultats avec RRF.

    Args:
        resultats_listes: liste de listes de (doc_id, score)
        k: constante RRF (60 est standard)
    """
    scores_fusionnes: dict[int, float] = {}

    for resultats in resultats_listes:
        for rang, (doc_id, _) in enumerate(resultats):
            if doc_id not in scores_fusionnes:
                scores_fusionnes[doc_id] = 0.0
            scores_fusionnes[doc_id] += 1.0 / (k + rang + 1)

    # Trier par score décroissant
    return sorted(
        scores_fusionnes.items(),
        key=lambda x: x[1],
        reverse=True
    )

Pipeline hybride complet

from openai import OpenAI
from rank_bm25 import BM25Okapi
import numpy as np

class RechercheHybride:
    def __init__(self, model: str = "text-embedding-3-large"):
        self.client = OpenAI()
        self.model = model
        self.documents: list[str] = []
        self.embeddings: np.ndarray | None = None
        self.bm25: BM25Okapi | None = None

    def indexer(self, documents: list[str]):
        self.documents = documents

        # Index sémantique
        response = self.client.embeddings.create(
            input=documents, model=self.model
        )
        vecteurs = sorted(response.data, key=lambda x: x.index)
        self.embeddings = np.array(
            [v.embedding for v in vecteurs], dtype=np.float32
        )

        # Index BM25
        corpus_tokenise = [tokeniser(d) for d in documents]
        self.bm25 = BM25Okapi(corpus_tokenise)

    def rechercher(
        self, requete: str, k: int = 5,
        poids_semantique: float = 0.7,
        poids_lexical: float = 0.3
    ) -> list[dict]:
        # Recherche sémantique
        query_emb = self.client.embeddings.create(
            input=requete, model=self.model
        ).data[0].embedding
        scores_sem = self.embeddings @ np.array(query_emb)
        classement_sem = [
            (i, float(scores_sem[i]))
            for i in np.argsort(scores_sem)[::-1]
        ]

        # Recherche lexicale
        query_tokens = tokeniser(requete)
        scores_lex = self.bm25.get_scores(query_tokens)
        classement_lex = [
            (i, float(scores_lex[i]))
            for i in np.argsort(scores_lex)[::-1]
        ]

        # Fusion RRF
        fusionne = reciprocal_rank_fusion(
            [classement_sem, classement_lex]
        )

        return [
            {
                "texte": self.documents[doc_id],
                "score_rrf": score,
                "score_sem": float(scores_sem[doc_id]),
                "score_lex": float(scores_lex[doc_id]),
            }
            for doc_id, score in fusionne[:k]
        ]

# Utilisation
moteur = RechercheHybride()
moteur.indexer([
    "Comment résoudre l'erreur 503 dans nginx",
    "Timeout et configuration du reverse proxy",
    "Guide d'installation nginx sur Ubuntu",
    "Diagnostiquer les problèmes de serveur web",
])

resultats = moteur.rechercher("erreur 503 nginx")
for r in resultats:
    print(f"RRF: {r['score_rrf']:.4f} | "
          f"Sem: {r['score_sem']:.3f} | "
          f"Lex: {r['score_lex']:.3f} | "
          f"{r['texte']}")

Recherche hybride native dans Weaviate

Weaviate propose la recherche hybride en une seule requête :

import weaviate

wclient = weaviate.connect_to_local()
collection = wclient.collections.get("Document")

# Recherche hybride native
resultats = collection.query.hybrid(
    query="erreur 503 nginx",
    alpha=0.7,  # 0 = lexicale pure, 1 = sémantique pure
    limit=5
)

for obj in resultats.objects:
    print(f"[{obj.metadata.score:.3f}] {obj.properties['texte']}")

Résumé

• La recherche hybride combine sémantique (sens) et lexicale (mots exacts)
• Reciprocal Rank Fusion (RRF) fusionne les classements efficacement
• Le paramètre alpha contrôle l’équilibre entre les deux approches
• Weaviate offre la recherche hybride nativement