Recherche hybride (sémantique + lexicale)
Mis à jour le 29 juillet 2026
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 exactement l’inverse. Prenez une requête d’administrateur système, « erreur 503 nginx », et observez le comportement des deux moteurs sur une base de documentation technique.
# 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 ✅Le sémantique comprend qu’il s’agit d’un incident serveur et remonte les articles généraux, mais le nombre 503 lui échappe : pour un modèle d’embedding, 503 et 504 sont deux nombres à trois chiffres, sémantiquement voisins alors qu’ils désignent deux pannes différentes. Le lexical, lui, retrouve le code exact mais passe à côté de l’article sur le timeout du reverse proxy, qui décrit pourtant la cause la plus fréquente. Aucun des deux ne donne la bonne page de résultats ; leur union, si.
Recherche lexicale avec BM25
BM25 est l’algorithme standard de la recherche par mots-clés. Il pondère chaque terme selon sa rareté dans le corpus et sa fréquence dans le document, ce qui fait naturellement ressortir les termes discriminants comme « 503 » ou « nginx ». Encore faut-il lui fournir des tokens propres : la fonction de tokenisation met en minuscules, extrait les mots et retire les mots vides du français, qui n’apportent aucune information de pertinence.
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)
Reste à combiner les deux classements, et c’est plus délicat qu’il n’y paraît : un score cosinus vit entre 0 et 1, un score BM25 n’a pas de borne supérieure. Les additionner directement reviendrait à laisser BM25 écraser le sémantique. RRF contourne le problème en ignorant les scores et en ne retenant que les rangs : chaque document reçoit 1 / (k + rang) dans chaque liste, et l’on somme ces contributions. Un document bien classé partout devance ainsi un document excellent dans une seule 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
)La constante k, fixée à 60 par convention issue de la littérature, amortit l’écart entre les premières positions : sans elle, le premier rang pèserait deux fois le deuxième, ce qui rendrait la fusion trop sensible au moindre écart de classement.
Pipeline hybride complet
La classe ci-dessous maintient les deux index en parallèle sur le même corpus, lance les deux recherches à chaque requête et fusionne les classements. Elle retourne aussi les scores individuels — détail précieux en phase de mise au point, car c’est en les comparant qu’on comprend lequel des deux moteurs a porté un résultat.
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']}")Sur ce corpus de quatre documents, l’article « erreur 503 dans nginx » domine le classement lexical et l’article « problèmes de serveur web » domine le sémantique ; la fusion les fait remonter tous les deux dans le trio de tête, ce qu’aucune des deux méthodes n’aurait produit seule.
Recherche hybride native dans Weaviate
Si vous utilisez déjà Weaviate, inutile d’écrire tout cela : la base intègre la fusion et l’expose en une seule requête, avec un paramètre alpha qui règle l’équilibre entre les deux moteurs — 0 pour du lexical pur, 1 pour du sémantique pur.
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']}")Un alpha de 0,7 constitue un point de départ raisonnable, mais il n’a rien d’universel : un corpus riche en références techniques et en codes d’erreur gagnera à descendre vers 0,5, tandis qu’une base de connaissances rédigée en langage courant supporte 0,8. Ne réglez ce curseur qu’avec un jeu d’évaluation sous les yeux — c’est précisément l’objet de la leçon suivante.
Résumé
- La recherche hybride combine sémantique (sens) et lexicale (mots exacts)
- Reciprocal Rank Fusion (RRF) fusionne les classements efficacement
- Le paramètre
alphacontrôle l’équilibre entre les deux approches - Weaviate offre la recherche hybride nativement