Hybrid Search (semantisch + lexikalisch)
Auf dieser Seite
Lernziele
- Verstehen, warum die Kombination aus semantischer und lexikalischer Suche wichtig ist
- Implementieren von Hybrid Search mittels Reciprocal Rank Fusion (RRF)
- Einsatz von Hybrid Search in Vektordatenbanken
Das Problem: Keine einzelne Methode ist perfekt
Die semantische Suche eignet sich hervorragend für die Behandlung von Synonymen, kann jedoch präzise technische Begriffe übersehen; die lexikalische Suche verhält sich genau umgekehrt. Betrachten Sie beispielsweise die Abfrage eines Systemadministrators: „nginx error 503”, und beobachten Sie, wie beide Engines gegen eine technische Dokumentationsbasis reagieren.
# Abfrage: "nginx error 503"
# Semantisch → findet "web server issues" (gut), verpasst aber "code 503" (Fehler)
# Lexikalisch → findet "nginx error 503" exakt (gut), verpasst aber "reverse proxy timeout" (Fehler)
# Hybrid → findet beides ✅Die semantische Suche versteht, dass es sich um einen Server-Vorfall handelt, und zeigt allgemeine Artikel an, doch die Zahl 503 entgeht ihr: Für ein Embedding-Modell sind 503 und 504 lediglich zwei dreistellige Zahlen, die semantisch nahe beieinander liegen, obwohl sie unterschiedliche Ausfälle bezeichnen. Die lexikalische Suche ruft den exakten Code ab, verpasst jedoch den Artikel über Reverse-Proxy-Zeitüberschreitungen, der oft die Ursache beschreibt. Keine der beiden Methoden allein liefert die richtigen Ergebnisse; ihre Vereinigung tut es.
Lexikalische Suche mit BM25
BM25 ist der Standardalgorithmus für die Keyword-Suche. Er gewichtet jeden Begriff basierend auf seiner Seltenheit im Korpus und seiner Häufigkeit im Dokument und bringt damit diskriminierende Begriffe wie „503” oder „nginx” natürlicherweise zur Oberfläche. Sie müssen ihm jedoch saubere Token zuführen: Die Tokenisierungsfunktion wandelt Text in Kleinbuchstaben um, extrahiert Wörter und entfernt französische Stop-Wörter, die kein Relevanzsignal hinzufügen.
from rank_bm25 import BM25Okapi
import re
def tokenize(text: str) -> list[str]:
"""Einfache Tokenisierung für Französisch."""
text = text.lower()
tokens = re.findall(r"\b\w+\b", text)
# Entfernen gängiger Stop-Wörter
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",
]
# Indexierung mit BM25
tokenized_corpus = [tokenize(doc) for doc in documents]
bm25 = BM25Okapi(tokenized_corpus)
# Suche
query_tokens = tokenize("erreur 503 nginx")
bm25_scores = bm25.get_scores(query_tokens)
for i, score in enumerate(bm25_scores):
print(f"BM25 {score:.3f} | {documents[i]}")Reciprocal Rank Fusion (RRF)
Nun müssen Sie die beiden Ranglisten kombinieren, was schwieriger aussieht, als es ist: Ein Kosinus-Score liegt zwischen 0 und 1, während ein BM25-Score kein oberes Limit hat. Eine direkte Addition würde dazu führen, dass BM25 das semantische Signal überlagert. RRF umgeht dies, indem er die rohen Scores ignoriert und nur Ränge berücksichtigt: Jedes Dokument erhält in jeder Liste 1 / (k + rank), und diese Beiträge werden summiert. Ein Dokument, das überall gut platziert ist, schlägt somit ein Dokument, das nur in einer Liste hervorragend abschneidet.
def reciprocal_rank_fusion(
result_lists: list[list[tuple[int, float]]],
k: int = 60
) -> list[tuple[int, float]]:
"""Fusioniert mehrere Ergebnislisten mittels RRF.
Args:
result_lists: Liste von Listen mit (doc_id, score)
k: RRF-Konstante (60 ist der Standard)
"""
fused_scores: dict[int, float] = {}
for results in result_lists:
for rank, (doc_id, _) in enumerate(results):
if doc_id not in fused_scores:
fused_scores[doc_id] = 0.0
fused_scores[doc_id] += 1.0 / (k + rank + 1)
# Nach Score absteigend sortieren
return sorted(
fused_scores.items(),
key=lambda x: x[1],
reverse=True
)Die Konstante k, die in der Literatur konventionell auf 60 festgelegt wird, dämpft die Lücke zwischen den oberen Positionen: Ohne sie würde der erste Rang doppelt so viel wiegen wie der zweite, was die Fusion zu empfindlich gegenüber geringfügigen Ranking-Unterschieden machen würde.
Kompletter Hybrid-Pipeline
Die Klasse unten hält beide Indizes parallel über dasselbe Korpus, führt beide Suchen für jede Abfrage durch und fusioniert die Ranglisten. Sie gibt zudem individuelle Scores zurück – ein wertvoller Detailaspekt während der Feinabstimmung, da ein Vergleich zeigt, welche Engine ein bestimmtes Ergebnis getragen hat.
from openai import OpenAI
from rank_bm25 import BM25Okapi
import numpy as np
class HybridSearch:
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 index(self, documents: list[str]):
self.documents = documents
# Semantischer Index
response = self.client.embeddings.create(
input=documents, model=self.model
)
vectors = sorted(response.data, key=lambda x: x.index)
self.embeddings = np.array(
[v.embedding for v in vectors], dtype=np.float32
)
# BM25-Index
tokenized_corpus = [tokenize(d) for d in documents]
self.bm25 = BM25Okapi(tokenized_corpus)
def search(
self, query: str, k: int = 5,
semantic_weight: float = 0.7,
lexical_weight: float = 0.3
) -> list[dict]:
# Semantische Suche
query_emb = self.client.embeddings.create(
input=query, model=self.model
).data[0].embedding
sem_scores = self.embeddings @ np.array(query_emb)
sem_ranking = [
(i, float(sem_scores[i]))
for i in np.argsort(sem_scores)[::-1]
]
# Lexikalische Suche
query_tokens = tokenize(query)
lex_scores = self.bm25.get_scores(query_tokens)
lex_ranking = [
(i, float(lex_scores[i]))
for i in np.argsort(lex_scores)[::-1]
]
# RRF-Fusion
fused = reciprocal_rank_fusion(
[sem_ranking, lex_ranking]
)
return [
{
"text": self.documents[doc_id],
"rrf_score": score,
"sem_score": float(sem_scores[doc_id]),
"lex_score": float(lex_scores[doc_id]),
}
for doc_id, score in fused[:k]
]
# Nutzung
engine = HybridSearch()
engine.index([
"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",
])
results = engine.search("erreur 503 nginx")
for r in results:
print(f"RRF: {r['rrf_score']:.4f} | "
f"Sem: {r['sem_score']:.3f} | "
f"Lex: {r['lex_score']:.3f} | "
f"{r['text']}")Auf diesem Korpus aus vier Dokumenten dominiert der Artikel „nginx error 503” die lexikalische Rangliste, während „web server issues” die semantische dominiert; die Fusion bringt beide in die Top-Drei, etwas, das keine der beiden Methoden allein erreichen würde.
Native Hybrid Search in Weaviate
Wenn Sie bereits Weaviate verwenden, müssen Sie nicht all dies selbst programmieren: Die Datenbank integriert die Fusion und stellt sie in einer einzigen Abfrage bereit, mit einem alpha-Parameter, der die beiden Engines ausbalanciert – 0 für rein lexikalisch, 1 für rein semantisch.
import weaviate
wclient = weaviate.connect_to_local()
collection = wclient.collections.get("Document")
# Native Hybrid Search
results = collection.query.hybrid(
query="erreur 503 nginx",
alpha=0.7, # 0 = rein lexikalisch, 1 = rein semantisch
limit=5
)
for obj in results.objects:
print(f"[{obj.metadata.score:.3f}] {obj.properties['texte']}")Ein alpha-Wert von 0,7 ist ein vernünftiger Ausgangspunkt, aber er ist nicht universell: Ein Korpus, das reich an technischen Referenzen und Fehlercodes ist, profitiert davon, den Wert Richtung 0,5 zu senken, während eine Wissensdatenbank, die in einfacher Sprache verfasst ist, Werte bis 0,8 verarbeiten kann. Tun Sie diesen Schieberegler nur mit einem Evaluierungssatz vor Augen – genau dies ist das Thema der nächsten Lektion.
Zusammenfassung
- Hybrid Search kombiniert semantische (Bedeutung) und lexikalische (exakte Wörter) Suche
- Reciprocal Rank Fusion (RRF) fusioniert Ranglisten effektiv
- Der
alpha-Parameter steuert das Gleichgewicht zwischen beiden Ansätzen - Weaviate bietet Hybrid Search nativ an