Hybrid Search — BM25 + vector
Hybridní vyhledávání představuje průlomový přístup kombinující tradiční BM25 algoritmus s moderními vektorovými embeddings. Tato technika umožňuje dosáhnout přesnějších výsledků než jednotlivé metody samostatně. Zjistěte, jak implementovat hybridní vyhledávání pro vaše LLM aplikace a AI agenty.
Hybrid Search: Kombinace BM25 a Vector Search pro lepší vyhledávání
V moderních RAG (Retrieval-Augmented Generation) systémech se často setkáváme s dilematem: použít tradiční keyword-based vyhledávání nebo sémantické vector search? Odpověď je jednoduchá – proč si vybírat, když můžeme využít oba přístupy současně? Hybrid search kombinuje sílu BM25 algoritmu s vector embeddings a poskytuje tak nejlepší výsledky z obou světů.
Proč hybrid search?
Každý z přístupů má své silné a slabé stránky:
- BM25 – vynikající pro exact match, specifické termíny, názvy produktů, kódy
- Vector search – perfektní pro sémantickou podobnost, synonyma, kontextové vyhledávání
Kombinací těchto metod dosáhneme lepší precision i recall, zejména v případech, kdy potřebujeme najít dokumenty obsahující jak specifické klíčová slova, tak sémanticky podobný obsah.
Implementace hybrid search
Příprava dat a indexů
Nejdříve si připravíme data a vytvoříme potřebné indexy. Budeme potřebovat jak inverzní index pro BM25, tak vector database pro embeddings.
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sentence_transformers import SentenceTransformer
import faiss
class HybridSearchEngine:
def __init__(self, model_name='sentence-transformers/all-MiniLM-L6-v2'):
self.embedding_model = SentenceTransformer(model_name)
self.tfidf_vectorizer = TfidfVectorizer(
stop_words='english',
max_features=10000,
ngram_range=(1, 2)
)
self.documents = []
self.embeddings = None
self.tfidf_matrix = None
self.faiss_index = None
def index_documents(self, documents):
"""Indexuje dokumenty pro oba typy vyhledávání"""
self.documents = documents
# Vector embeddings
embeddings = self.embedding_model.encode(documents)
self.embeddings = embeddings
# FAISS index pro rychlé vector search
dimension = embeddings.shape[1]
self.faiss_index = faiss.IndexFlatIP(dimension)
# Normalizujeme embeddings pro cosine similarity
faiss.normalize_L2(embeddings)
self.faiss_index.add(embeddings.astype('float32'))
# TF-IDF pro BM25-like vyhledávání
self.tfidf_matrix = self.tfidf_vectorizer.fit_transform(documents)
BM25 implementace
Implementujeme BM25 algoritmus, který je vylepšenou verzí TF-IDF s lepším normalizováním délky dokumentu.
import math
from collections import Counter
class BM25:
def __init__(self, corpus, k1=1.5, b=0.75):
self.k1 = k1
self.b = b
self.corpus = corpus
self.doc_len = [len(doc.split()) for doc in corpus]
self.avgdl = sum(self.doc_len) / len(self.doc_len)
self.doc_freqs = []
self.idf = {}
# Vytvoříme inverzní index
for doc in corpus:
frequencies = Counter(doc.lower().split())
self.doc_freqs.append(frequencies)
for word in frequencies:
if word not in self.idf:
self.idf[word] = 0
self.idf[word] += 1
# Spočítáme IDF hodnoty
for word in self.idf:
self.idf[word] = math.log((len(corpus) - self.idf[word] + 0.5) /
(self.idf[word] + 0.5))
def search(self, query, k=10):
"""Vyhledá k nejrelevantnějších dokumentů pomocí BM25"""
query_words = query.lower().split()
scores = []
for i, doc_freq in enumerate(self.doc_freqs):
score = 0
for word in query_words:
if word in doc_freq:
tf = doc_freq[word]
idf = self.idf.get(word, 0)
# BM25 formula
numerator = tf * (self.k1 + 1)
denominator = (tf + self.k1 * (1 - self.b +
self.b * self.doc_len[i] / self.avgdl))
score += idf * (numerator / denominator)
scores.append((i, score))
# Seřadíme podle skóre
scores.sort(key=lambda x: x[1], reverse=True)
return scores[:k]
Kombinování výsledků
Klíčová část hybrid search je v kombinování výsledků z obou metod. Existuje několik přístupů – můžeme použít váženou sumu, rank fusion nebo reciprocal rank fusion (RRF).
class HybridSearchEngine:
# ... předchozí kód ...
def hybrid_search(self, query, k=10, alpha=0.7):
"""
Kombinuje BM25 a vector search
alpha: váha pro vector search (1-alpha pro BM25)
"""
# Vector search
query_embedding = self.embedding_model.encode([query])
faiss.normalize_L2(query_embedding)
vector_scores, vector_indices = self.faiss_index.search(
query_embedding.astype('float32'), k*2
)
# BM25 search
if not hasattr(self, 'bm25'):
self.bm25 = BM25(self.documents)
bm25_results = self.bm25.search(query, k*2)
# Normalizujeme skóre na [0,1]
def normalize_scores(scores):
if not scores or max(scores) == min(scores):
return [0.0] * len(scores)
min_score, max_score = min(scores), max(scores)
return [(s - min_score) / (max_score - min_score) for s in scores]
# Vytvoříme dictionary se všemi skóre
combined_scores = {}
# Vector search skóre
norm_vector_scores = normalize_scores(vector_scores[0].tolist())
for idx, score in zip(vector_indices[0], norm_vector_scores):
combined_scores[idx] = alpha * score
# BM25 skóre
bm25_scores = [score for _, score in bm25_results]
norm_bm25_scores = normalize_scores(bm25_scores)
for (idx, _), score in zip(bm25_results, norm_bm25_scores):
if idx in combined_scores:
combined_scores[idx] += (1 - alpha) * score
else:
combined_scores[idx] = (1 - alpha) * score
# Seřadíme podle kombinovaného skóre
sorted_results = sorted(combined_scores.items(),
key=lambda x: x[1], reverse=True)
return [(idx, score, self.documents[idx])
for idx, score in sorted_results[:k]]
Reciprocal Rank Fusion (RRF)
Alternativním přístupem k váženému průměru je RRF, který kombinuje výsledky na základě jejich pořadí namísto absolutních skóre. Tento přístup je často robustnější vůči různým škálám skóre.
def rrf_hybrid_search(self, query, k=10, rrf_constant=60):
"""Hybrid search s Reciprocal Rank Fusion"""
# Získáme výsledky z obou metod
query_embedding = self.embedding_model.encode([query])
faiss.normalize_L2(query_embedding)
vector_scores, vector_indices = self.faiss_index.search(
query_embedding.astype('float32'), k*2
)
bm25_results = self.bm25.search(query, k*2)
# RRF skóre pro každý dokument
rrf_scores = {}
# Vector search - přidáme RRF skóre podle pozice
for rank, doc_idx in enumerate(vector_indices[0]):
rrf_scores[doc_idx] = 1.0 / (rrf_constant + rank + 1)
# BM25 - přidáme RRF skóre
for rank, (doc_idx, _) in enumerate(bm25_results):
if doc_idx in rrf_scores:
rrf_scores[doc_idx] += 1.0 / (rrf_constant + rank + 1)
else:
rrf_scores[doc_idx] = 1.0 / (rrf_constant + rank + 1)
# Seřadíme podle RRF skóre
sorted_results = sorted(rrf_scores.items(),
key=lambda x: x[1], reverse=True)
return [(idx, score, self.documents[idx])
for idx, score in sorted_results[:k]]
Praktické využití
Ukažme si, jak hybrid search engine použít v praxi:
# Inicializace a indexování
documents = [
"Python je programovací jazyk pro data science",
"JavaScript se používá pro webový vývoj",
"Machine learning algoritmy v Pythonu",
"React framework pro frontend development",
"Analýza dat pomocí pandas knihovny"
]
search_engine = HybridSearchEngine()
search_engine.index_documents(documents)
# Vyhledávání
query = "programování v pythonu"
# Klasické BM25
print("BM25 výsledky:")
bm25_results = search_engine.bm25.search(query, k=3)
for idx, score in bm25_results:
print(f"Skóre: {score:.3f} - {documents[idx]}")
print("\nHybrid search výsledky:")
hybrid_results = search_engine.hybrid_search(query, k=3)
for idx, score, doc in hybrid_results:
print(f"Skóre: {score:.3f} - {doc}")
print("\nRRF výsledky:")
rrf_results = search_engine.rrf_hybrid_search(query, k=3)
for idx, score, doc in rrf_results:
print(f"RRF skóre: {score:.3f} - {doc}")
Optimalizace a tunning
Pro dosažení nejlepších výsledků je důležité správně nastavit parametry:
- Alpha parameter – váha mezi vector (alpha) a BM25 (1-alpha) search. Začněte s hodnotou 0.7
- BM25 parametry – k1 (1.2-2.0) a b (0.75) podle typu dokumentů
- Vector model – vyberte model vhodný pro váš domain a jazyk
- RRF konstanta – typicky 60, ale může být třeba upravit podle dat
def evaluate_hybrid_search(self, test_queries, relevance_judgments, alphas):
"""Evaluace různých alpha hodnot"""
best_alpha = 0.7
best_score = 0
for alpha in alphas:
scores = []
for query in test_queries:
results = self.hybrid_search(query, alpha=alpha)
# Spočítáme precision@k nebo jiné metriky
score = calculate_precision_at_k(results, relevance_judgments[query])
scores.append(score)
avg_score = sum(scores) / len(scores)
if avg_score > best_score:
best_score = avg_score
best_alpha = alpha
return best_alpha, best_score
Shrnutí
Hybrid search kombinující BM25 a vector search poskytuje robustní řešení pro moderní vyhledávací systémy. Zatímco BM25 vyniká v přesném vyhledávání specifických termínů, vector search dokáže zachytit sémantickou podobnost. Jejich kombinace pomocí váženého průměru nebo RRF přináší lepší výsledky než kterýkoli z přístupů samostatně. Klíčem k úspěchu je správné nastavení parametrů a testování na reálných datech vašeho domain.