Реализация Hybrid Search (векторный + полнотекстовый поиск) для RAG

Реализация Hybrid Search (векторный + полнотекстовый поиск) для RAG

Hybrid Search — комбинация векторного (dense) и полнотекстового (sparse/BM25) поиска с последующим слиянием результатов. Практика показывает, что hybrid search стабильно превосходит любой из методов в отдельности на большинстве корпоративных датасетов. Dense search хорош для семантически близких запросов, BM25 — для точных терминов, номеров, аббревиатур.

Почему нельзя обойтись только dense search

Dense embedding усредняет семантику — это и сила, и слабость. Запрос «договор №ДА-2023-451» будет иметь высокое косинусное сходство с договорами вообще, но не с конкретным документом по номеру. BM25 найдёт точное совпадение строки «ДА-2023-451» мгновенно.

Dense search плохо работает для:

• Точных номеров (договор, артикул, серийный номер)
• Аббревиатур и специфических акронимов
• Редких технических терминов
• Запросов на поиск точной цитаты

BM25 плохо работает для:

• Перефразированных запросов (синонимы)
• Семантически похожих концепций с разными словами
• Межъязыковых запросов
• Неточных описаний («что-то про оплату после поставки»)

Алгоритмы слияния результатов

Reciprocal Rank Fusion (RRF) — наиболее устойчивый метод:

from collections import defaultdict

def reciprocal_rank_fusion(
    dense_results: list[tuple],   # [(doc_id, score), ...]
    sparse_results: list[tuple],
    k: int = 60  # RRF константа (обычно 60)
) -> list[tuple]:
    """
    RRF score = sum(1 / (k + rank_i)) по всем спискам
    k=60 стандартное значение (Cormack et al., 2009)
    """
    scores = defaultdict(float)

    for rank, (doc_id, _) in enumerate(dense_results, 1):
        scores[doc_id] += 1 / (k + rank)

    for rank, (doc_id, _) in enumerate(sparse_results, 1):
        scores[doc_id] += 1 / (k + rank)

    return sorted(scores.items(), key=lambda x: -x[1])

Relative Score Fusion (RSF) — нормализованное объединение:

def relative_score_fusion(
    dense_results: list[tuple],
    sparse_results: list[tuple],
    alpha: float = 0.5  # Вес dense
) -> list[tuple]:
    """Нормализует оценки в [0,1] и взвешивает"""
    scores = defaultdict(float)

    # Нормализация dense
    if dense_results:
        max_d = max(s for _, s in dense_results)
        min_d = min(s for _, s in dense_results)
        for doc_id, score in dense_results:
            norm = (score - min_d) / (max_d - min_d + 1e-8)
            scores[doc_id] += alpha * norm

    # Нормализация sparse
    if sparse_results:
        max_s = max(s for _, s in sparse_results)
        min_s = min(s for _, s in sparse_results)
        for doc_id, score in sparse_results:
            norm = (score - min_s) / (max_s - min_s + 1e-8)
            scores[doc_id] += (1 - alpha) * norm

    return sorted(scores.items(), key=lambda x: -x[1])

SPLADE: продвинутый sparse encoder

SPLADE (Sparse Lexical and Expansion Model) генерирует sparse векторы с лексическим расширением — модель учится «расширять» запрос синонимами и связанными терминами:

from fastembed import SparseTextEmbedding

sparse_model = SparseTextEmbedding(
    model_name="prithivida/Splade_PP_en_v1"
)

def encode_sparse(text: str) -> dict:
    """Возвращает sparse вектор {token_id: weight}"""
    output = list(sparse_model.embed([text]))[0]
    return {
        "indices": output.indices.tolist(),
        "values": output.values.tolist(),
    }

SPLADE превосходит BM25 на большинстве BEIR бенчмарков. Для русского языка рекомендуем модель naver/efficient-splade-VI-BT-large-query или multilingual варианты.

Реализация с Qdrant (практический пример)

from qdrant_client import QdrantClient
from qdrant_client.models import (
    SparseVector, Prefetch, FusionQuery, Fusion,
    NamedVector, NamedSparseVector
)
from fastembed import TextEmbedding, SparseTextEmbedding

dense_model = TextEmbedding("BAAI/bge-m3")  # Multilingual dense
sparse_model = SparseTextEmbedding("prithivida/Splade_PP_en_v1")
client = QdrantClient(url="http://localhost:6333")

def hybrid_search(query: str, top_k: int = 5) -> list[dict]:
    # Dense embedding
    dense_vec = list(dense_model.embed([query]))[0].tolist()

    # Sparse embedding
    sparse_output = list(sparse_model.embed([query]))[0]
    sparse_vec = SparseVector(
        indices=sparse_output.indices.tolist(),
        values=sparse_output.values.tolist()
    )

    results = client.query_points(
        collection_name="hybrid_docs",
        prefetch=[
            Prefetch(query=dense_vec, using="dense", limit=50),
            Prefetch(query=sparse_vec, using="sparse", limit=50),
        ],
        query=FusionQuery(fusion=Fusion.RRF),
        limit=top_k,
        with_payload=True,
    )

    return [
        {"text": r.payload["text"], "source": r.payload["source"], "score": r.score}
        for r in results.points
    ]

Практический кейс: влияние alpha на качество retrieval

Датасет: 12 000 документов корпоративной базы знаний (договоры, регламенты, FAQ).

Тестовый набор: 400 запросов разных типов.

Hybrid RRF без reranker уже бьёт dense+reranker. Комбинация hybrid+reranker — наилучший результат.

Оптимальное k для RRF

k=60 — эмпирически устойчивое значение. Слишком малое k (10–20) даёт большой вес топ-позициям. Слишком большое (100+) нивелирует разницу между позициями. На реальных данных: проверьте k∈{20, 40, 60, 80} на валидационном наборе.

Сроки реализации

• Настройка sparse encoder + SPLADE: 2–3 дня
• Интеграция hybrid search в существующий RAG: 3–5 дней
• Подбор оптимального alpha/k на датасете: 2–3 дня
• Итого: 1–2 недели