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

Реалізація гібридного пошуку (векторний + повнотекстовий пошук) для RAG

Гібридний пошук — комбінація векторного (dense) та повнотекстового (sparse/BM25) пошуку з подальшим злиттям результатів. Практика показує, що гібридний пошук послідовно перевищує будь-який із методів окремо на більшості корпоративних датасетів. Dense пошук хороший для семантично близьких запитів, BM25 — для точних термінів, чисел, абревіатур.

Чому Dense пошуку самого по собі недостатньо

Dense embedding усереднює семантику—і сила, і слабкість одночасно. Запит "договір №ДА-2023-451" матиме високу косинусну подібність з договорами взагалі, але не з конкретним документом за номером. BM25 знайде точне співпадіння рядка "ДА-2023-451" миттєво.

Dense пошук погано працює для:

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

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 дні
• Інтеграція гібридного пошуку в існуючий RAG: 3–5 днів
• Підбір оптимального alpha/k на датасеті: 2–3 дні
• Всього: 1–2 тижні