Hybrid Search: BM25 + семантический поиск

pip install rank-bm25 sentence-transformers numpy

from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer
import numpy as np
from typing import Optional


# ── Документы ──────────────────────────────────────────────────────────
docs = [
    "nginx ssl сертификат настройка https",
    "tls протокол шифрование сертификат letsencrypt",
    "nginx прокси обратный proxy upstream",
    "ssl сертификат openssl генерация csr",
    "redis кеш производительность latency",
    "postgresql индексы оптимизация запросов",
    "docker nginx конфигурация reverse proxy ssl",
    "certbot letsencrypt автоматическое обновление сертификата",
]

# ── Индексирование ─────────────────────────────────────────────────────
# BM25: токенизируем по пробелам (в production — использовать нормальный токенизатор)
tokenized = [doc.split() for doc in docs]
bm25 = BM25Okapi(tokenized)

# Semantic: кодируем все документы заранее
model = SentenceTransformer("all-MiniLM-L6-v2")
doc_embeddings = model.encode(docs, normalize_embeddings=True)


def bm25_search(query: str, top_k: int = 10) -> list[tuple[int, float]]:
    """Возвращает [(doc_idx, score), ...] отсортированный по убыванию."""
    scores = bm25.get_scores(query.split())
    ranked = sorted(enumerate(scores), key=lambda x: x[1], reverse=True)
    return ranked[:top_k]


def semantic_search(query: str, top_k: int = 10) -> list[tuple[int, float]]:
    """Возвращает [(doc_idx, score), ...] отсортированный по убыванию."""
    q_emb = model.encode([query], normalize_embeddings=True)[0]
    # Cosine similarity через dot product (оба нормализованы)
    scores = doc_embeddings @ q_emb
    ranked = sorted(enumerate(scores), key=lambda x: x[1], reverse=True)
    return ranked[:top_k]


def reciprocal_rank_fusion(
    *ranked_lists: list[tuple[int, float]],
    k: int = 60,
) -> list[tuple[int, float]]:
    """
    Объединяет произвольное число ranked списков через RRF.

    Args:
        *ranked_lists: каждый список — [(doc_idx, score), ...] по убыванию
        k: константа сглаживания (обычно 60)

    Returns:
        Новый ranked список [(doc_idx, rrf_score), ...]
    """
    rrf_scores: dict[int, float] = {}

    for ranked in ranked_lists:
        for rank, (doc_idx, _score) in enumerate(ranked, start=1):
            rrf_scores[doc_idx] = rrf_scores.get(doc_idx, 0.0) + 1.0 / (k + rank)

    return sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)


def weighted_hybrid(
    bm25_results: list[tuple[int, float]],
    semantic_results: list[tuple[int, float]],
    alpha: float = 0.5,
) -> list[tuple[int, float]]:
    """
    Нормализует скоры обоих методов и складывает с весом alpha.

    alpha = 0   → только BM25
    alpha = 0.5 → равный вес
    alpha = 1   → только semantic
    """
    def min_max_normalize(results: list[tuple[int, float]]) -> dict[int, float]:
        if not results:
            return {}
        scores = [s for _, s in results]
        lo, hi = min(scores), max(scores)
        if hi == lo:
            return {idx: 1.0 for idx, _ in results}
        return {idx: (s - lo) / (hi - lo) for idx, s in results}

    bm25_norm = min_max_normalize(bm25_results)
    sem_norm  = min_max_normalize(semantic_results)

    all_ids = set(bm25_norm) | set(sem_norm)
    hybrid: dict[int, float] = {}
    for doc_idx in all_ids:
        b_score = bm25_norm.get(doc_idx, 0.0)
        s_score = sem_norm.get(doc_idx, 0.0)
        hybrid[doc_idx] = alpha * s_score + (1 - alpha) * b_score

    return sorted(hybrid.items(), key=lambda x: x[1], reverse=True)


# ── Пример использования ───────────────────────────────────────────────
query = "nginx ssl сертификат"

bm25_res    = bm25_search(query, top_k=8)
semantic_res = semantic_search(query, top_k=8)

rrf_res     = reciprocal_rank_fusion(bm25_res, semantic_res)
weighted_res = weighted_hybrid(bm25_res, semantic_res, alpha=0.5)

print("=== BM25 ===")
for rank, (idx, score) in enumerate(bm25_res[:5], 1):
    print(f"  #{rank}  [{score:.3f}]  {docs[idx][:60]}")

print("\n=== Semantic ===")
for rank, (idx, score) in enumerate(semantic_res[:5], 1):
    print(f"  #{rank}  [{score:.3f}]  {docs[idx][:60]}")

print("\n=== RRF (hybrid) ===")
for rank, (idx, score) in enumerate(rrf_res[:5], 1):
    print(f"  #{rank}  [{score:.4f}]  {docs[idx][:60]}")

print("\n=== Weighted (alpha=0.5) ===")
for rank, (idx, score) in enumerate(weighted_res[:5], 1):
    print(f"  #{rank}  [{score:.3f}]  {docs[idx][:60]}")

=== BM25 ===
  #1  [3.241]  nginx ssl сертификат настройка https
  #2  [2.187]  ssl сертификат openssl генерация csr
  #3  [1.943]  tls протокол шифрование сертификат letsencrypt
  #4  [1.102]  docker nginx конфигурация reverse proxy ssl
  #5  [0.871]  certbot letsencrypt автоматическое обновление сертификата

=== Semantic ===
  #1  [0.921]  tls протокол шифрование сертификат letsencrypt
  #2  [0.905]  nginx ssl сертификат настройка https
  #3  [0.887]  ssl сертификат openssl генерация csr
  #4  [0.874]  certbot letsencrypt автоматическое обновление сертификата
  #5  [0.841]  docker nginx конфигурация reverse proxy ssl

=== RRF (hybrid) ===
  #1  [0.0325]  nginx ssl сертификат настройка https        ← #1 в BM25, #2 в semantic
  #2  [0.0323]  tls протокол шифрование сертификат          ← #3 в BM25, #1 в semantic
  #3  [0.0317]  ssl сертификат openssl генерация csr        ← #2 в BM25, #3 в semantic
  #4  [0.0160]  certbot letsencrypt автоматическое...
  #5  [0.0158]  docker nginx конфигурация reverse proxy ssl

=== Weighted (alpha=0.5) ===
  #1  [0.924]  nginx ssl сертификат настройка https
  #2  [0.891]  tls протокол шифрование сертификат
  #3  [0.867]  ssl сертификат openssl генерация csr

pip install weaviate-client

import weaviate
import weaviate.classes as wvc

client = weaviate.connect_to_local()  # или connect_to_weaviate_cloud(...)

# ── Создание коллекции ─────────────────────────────────────────────────
client.collections.create(
    "KnowledgeBase",
    vectorizer_config=wvc.config.Configure.Vectorizer.text2vec_openai(),
    properties=[
        wvc.config.Property(name="content", data_type=wvc.config.DataType.TEXT),
        wvc.config.Property(name="source",  data_type=wvc.config.DataType.TEXT),
        wvc.config.Property(name="section", data_type=wvc.config.DataType.TEXT),
    ],
)

kb = client.collections.get("KnowledgeBase")

# ── Загрузка документов ────────────────────────────────────────────────
with kb.batch.dynamic() as batch:
    for doc in documents:
        batch.add_object({"content": doc["text"], "source": doc["url"], "section": doc["title"]})

# ── Hybrid search ──────────────────────────────────────────────────────
response = kb.query.hybrid(
    query="nginx ssl сертификат настройка",
    alpha=0.5,          # 0 = только BM25, 1 = только semantic, 0.5 = баланс
    limit=10,
    return_metadata=wvc.query.MetadataQuery(score=True, explain_score=True),
    return_properties=["content", "source", "section"],
)

for obj in response.objects:
    print(f"Score: {obj.metadata.score:.4f}")
    print(f"  {obj.properties['content'][:100]}")
    print(f"  Explain: {obj.metadata.explain_score}")
    print()

client.close()

# Weaviate позволяет настроить BM25-параметры на уровне коллекции
client.collections.create(
    "KnowledgeBase",
    vectorizer_config=wvc.config.Configure.Vectorizer.text2vec_openai(),
    inverted_index_config=wvc.config.Configure.inverted_index(
        bm25_b=0.75,          # нормализация по длине (default)
        bm25_k1=1.2,          # насыщение частоты (default)
        index_null_state=False,
        index_property_length=False,
        index_timestamps=False,
    ),
    properties=[
        wvc.config.Property(
            name="content",
            data_type=wvc.config.DataType.TEXT,
            tokenization=wvc.config.Tokenization.WORD,  # или LOWERCASE, FIELD
        ),
    ],
)

# ── Hybrid с фильтром ──────────────────────────────────────────────────
from weaviate.classes.query import Filter

response = kb.query.hybrid(
    query="ssl сертификат",
    alpha=0.7,          # уклон к semantic (для смысловых запросов)
    limit=5,
    filters=Filter.by_property("section").equal("Безопасность"),
)

pip install qdrant-client fastembed

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, SparseVectorParams, SparseIndexParams,
    PointStruct, SparseVector, NamedSparseVector, NamedVector,
    SearchRequest, FusionQuery, Prefetch, Fusion,
)
from fastembed import TextEmbedding, SparseTextEmbedding

# ── Клиент и модели ────────────────────────────────────────────────────
client = QdrantClient(":memory:")  # или QdrantClient(url="http://localhost:6333")

dense_model  = TextEmbedding("BAAI/bge-small-en-v1.5")         # 384-мерный dense
sparse_model = SparseTextEmbedding("prithvida/Splade_PP_en_v1") # SPLADE sparse

COLLECTION = "kb_hybrid"

# ── Создание коллекции с двумя типами векторов ─────────────────────────
client.create_collection(
    collection_name=COLLECTION,
    vectors_config={
        "dense": VectorParams(size=384, distance=Distance.COSINE),
    },
    sparse_vectors_config={
        "sparse": SparseVectorParams(
            index=SparseIndexParams(on_disk=False),
        ),
    },
)

# ── Индексирование: генерируем оба вектора для каждого документа ────────
docs = [
    {"id": 1, "text": "nginx ssl сертификат настройка https"},
    {"id": 2, "text": "tls протокол шифрование letsencrypt certbot"},
    {"id": 3, "text": "redis кеш производительность latency оптимизация"},
    {"id": 4, "text": "ssl openssl генерация csr приватный ключ"},
]

texts = [d["text"] for d in docs]

# fastembed возвращает генераторы — превращаем в списки
dense_vecs  = list(dense_model.embed(texts))
sparse_vecs = list(sparse_model.embed(texts))

points = []
for i, doc in enumerate(docs):
    sp = sparse_vecs[i]
    points.append(PointStruct(
        id=doc["id"],
        payload={"text": doc["text"]},
        vector={
            "dense":  dense_vecs[i].tolist(),
            "sparse": SparseVector(indices=sp.indices.tolist(), values=sp.values.tolist()),
        },
    ))

client.upsert(collection_name=COLLECTION, points=points)


# ── Hybrid поиск через Query API с Fusion ──────────────────────────────
def hybrid_search(query: str, top_k: int = 5):
    # Кодируем запрос обоими моделями
    q_dense  = list(dense_model.embed([query]))[0].tolist()
    q_sparse = list(sparse_model.embed([query]))[0]
    q_sparse_vec = SparseVector(
        indices=q_sparse.indices.tolist(),
        values=q_sparse.values.tolist(),
    )

    results = client.query_points(
        collection_name=COLLECTION,
        prefetch=[
            # Сначала получаем top-k по каждому методу
            Prefetch(query=q_dense,       using="dense",  limit=top_k * 2),
            Prefetch(query=q_sparse_vec,  using="sparse", limit=top_k * 2),
        ],
        # Объединяем через RRF
        query=FusionQuery(fusion=Fusion.RRF),
        limit=top_k,
        with_payload=True,
    )
    return results.points


# ── Пример ────────────────────────────────────────────────────────────
for point in hybrid_search("nginx ssl сертификат"):
    print(f"id={point.id}  score={point.score:.4f}  {point.payload['text']}")

from dataclasses import dataclass
import numpy as np


@dataclass
class LabeledQuery:
    query: str
    relevant_ids: set[int]  # ids документов, которые должны быть в top-k


def mean_reciprocal_rank(results: list[tuple[int, float]], relevant_ids: set[int]) -> float:
    """MRR: обратный ранг первого релевантного документа."""
    for rank, (doc_id, _) in enumerate(results, start=1):
        if doc_id in relevant_ids:
            return 1.0 / rank
    return 0.0


def evaluate_alpha(
    labeled_queries: list[LabeledQuery],
    alpha_values: list[float],
    top_k: int = 10,
) -> dict[float, float]:
    """Для каждого alpha считаем средний MRR по размеченной выборке."""
    results = {}
    for alpha in alpha_values:
        mrr_scores = []
        for lq in labeled_queries:
            bm25_res    = bm25_search(lq.query, top_k=top_k * 2)
            semantic_res = semantic_search(lq.query, top_k=top_k * 2)
            hybrid_res   = weighted_hybrid(bm25_res, semantic_res, alpha=alpha)
            mrr = mean_reciprocal_rank(hybrid_res[:top_k], lq.relevant_ids)
            mrr_scores.append(mrr)
        results[alpha] = np.mean(mrr_scores)
    return results


# ── Пример ────────────────────────────────────────────────────────────
labeled = [
    LabeledQuery("nginx ssl настройка",          relevant_ids={0, 3, 6}),
    LabeledQuery("кеш redis производительность", relevant_ids={4}),
    LabeledQuery("tls шифрование сертификат",    relevant_ids={1, 3, 7}),
]

alphas  = [0.0, 0.25, 0.5, 0.75, 1.0]
mrr_by_alpha = evaluate_alpha(labeled, alphas)

for alpha, mrr in sorted(mrr_by_alpha.items(), key=lambda x: x[1], reverse=True):
    print(f"alpha={alpha:.2f}  MRR={mrr:.3f}")

best_alpha = max(mrr_by_alpha, key=mrr_by_alpha.get)
print(f"\nОптимальный alpha: {best_alpha}")