Комбинируем семантический поиск с keyword-поиском. BM25, Reciprocal Rank Fusion, практические примеры улучшения точности.
Комбинируем семантический поиск с keyword-поиском. BM25, Reciprocal Rank Fusion, практические примеры улучшения точности.
За 60 минут вы:
Векторный поиск:
Keyword поиск (BM25):
Гибридный поиск: лучшее из обоих миров!
pip install rank-bm25from rank_bm25 import BM25Okapi
# Документы
documents = [
"Python язык программирования",
"Django веб фреймворк на Python",
"Flask микрофреймворк для веб"
]
# Токенизация
tokenized_docs = [doc.lower().split() for doc in documents]
# Создание индекса
bm25 = BM25Okapi(tokenized_docs)
# Поиск
query = "веб Python"
query_tokens = query.lower().split()
scores = bm25.get_scores(query_tokens)
print(f"BM25 scores: {scores}")
# Топ результат
top_idx = scores.argmax()
print(f"Лучший: {documents[top_idx]}")import faiss
import numpy as np
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer
class HybridSearch:
"""Гибридный поиск: векторный + BM25."""
def __init__(self, documents: list[str]):
self.documents = documents
self.model = SentenceTransformer('all-MiniLM-L6-v2')
# BM25 индекс
tokenized_docs = [doc.lower().split() for doc in documents]
self.bm25 = BM25Okapi(tokenized_docs)
# Векторный индекс
embeddings = self.model.encode(documents)
faiss.normalize_L2(embeddings)
self.index = faiss.IndexFlatIP(embeddings.shape[1])
self.index.add(embeddings)
def search(
self,
query: str,
k: int = 10,
alpha: float = 0.5
) -> list[dict]:
"""
Гибридный поиск.
Args:
query: Поисковый запрос
k: Количество результатов
alpha: Вес векторного поиска (0 = только BM25, 1 = только векторный)
"""
# BM25 поиск
query_tokens = query.lower().split()
bm25_scores = self.bm25.get_scores(query_tokens)
# Векторный поиск
query_embedding = self.model.encode([query])
faiss.normalize_L2(query_embedding)
distances, indices = self.index.search(query_embedding, k=len(self.documents))
vector_scores = np.zeros(len(self.documents))
vector_scores[indices[0]] = distances[0]
# Нормализация к [0, 1]
bm25_norm = self._normalize(bm25_scores)
vector_norm = self._normalize(vector_scores)
# Комбинирование
combined_scores = alpha * vector_norm + (1 - alpha) * bm25_norm
# Топ-k
top_indices = np.argsort(combined_scores)[::-1][:k]
return [
{
"index": int(idx),
"score": float(combined_scores[idx]),
"text": self.documents[idx]
}
for idx in top_indices
]
def _normalize(self, scores: np.ndarray) -> np.ndarray:
"""Нормализация к [0, 1]."""
min_s, max_s = scores.min(), scores.max()
if max_s - min_s < 1e-6:
return np.zeros_like(scores)
return (scores - min_s) / (max_s - min_s)
# Использование
documents = [
"Python язык программирования общего назначения",
"Питон это змея из семейства удавов",
"Django веб фреймворк на Python для быстрой разработки",
"Flask микрофреймворк для создания веб приложений",
"FastAPI современный фреймворк для создания API"
]
searcher = HybridSearch(documents)
results = searcher.search("веб разработка на Python", k=3, alpha=0.7)
for r in results:
print(f"[{r['score']:.3f}] {r['text']}")Альтернативный способ слияния результатов:
from collections import defaultdict
def reciprocal_rank_fusion(
results_lists: list[list],
k: int = 60
) -> list[tuple]:
"""
Слияние результатов через Reciprocal Rank Fusion.
Args:
results_lists: Список ранжированных списков результатов
k: Константа для сглаживания (обычно 60)
"""
rrf_scores = defaultdict(float)
for results in results_lists:
for rank, item in enumerate(results):
doc_id = item["id"] if isinstance(item, dict) else item
rrf_scores[doc_id] += 1 / (k + rank + 1)
# Сортировка по RRF score
sorted_results = sorted(
rrf_scores.items(),
key=lambda x: x[1],
reverse=True
)
return sorted_results
# Пример
def hybrid_search_rrf(query, vector_index, bm25, model, documents, k=10):
"""Гибридный поиск с RRF."""
# Векторный поиск
query_embedding = model.encode([query])
faiss.normalize_L2(query_embedding)
_, vector_indices = vector_index.search(query_embedding, k=50)
vector_results = [{"id": idx} for idx in vector_indices[0]]
# BM25 поиск
query_tokens = query.lower().split()
bm25_scores = bm25.get_scores(query_tokens)
bm25_indices = np.argsort(bm25_scores)[::-1][:50]
bm25_results = [{"id": idx} for idx in bm25_indices]
# Слияние через RRF
fused = reciprocal_rank_fusion([vector_results, bm25_results], k=60)
return [
{"id": doc_id, "rrf_score": score, "text": documents[doc_id]}
for doc_id, score in fused[:k]
]Следующая тема — RAG-приложение.
Далее: RAG-приложение: поиск + генерация ответов