hybrid-search

What this skill does

# Hybrid Search

## Why Hybrid

Dense vectors capture meaning but miss exact terms (product SKUs, error codes, proper nouns, code tokens). BM25 nails lexical matches but misses synonyms and paraphrases. Fusion consistently beats either alone on real-world corpora — typically +10-20% recall@10.

## Fusion Algorithms

| Algorithm | Needs score calibration | Robust to list-size mismatch | Tunable |
|---|---|---|---|
| RRF (Reciprocal Rank Fusion) | No | Yes | k_constant |
| Weighted linear (alpha) | Yes (min-max normalize) | No | alpha |
| CombSUM / CombMNZ | Yes | Partial | No |
| Learned fusion (LTR) | Yes | Yes | Model |

Default to RRF unless you have a labeled dataset for alpha tuning.

## RRF: Formula and Code

```
score(d) = sum over rankings r of  1 / (k + rank_r(d))
```

`k` (the RRF constant) is typically 60 in the original TREC paper. Larger `k` = more uniform contribution from low ranks.

```python
from collections import defaultdict

def rrf(rankings: list[list[str]], k: int = 60) -> list[tuple[str, float]]:
    """Each ranking is a list of doc IDs ordered by score desc."""
    scores = defaultdict(float)
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking):
            scores[doc_id] += 1.0 / (k + rank + 1)
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)
```

Works across any number of rankers. No score calibration needed.

## BM25 + Dense with RRF (Python, from scratch)

```python
from rank_bm25 import BM25Okapi
from langchain_openai import OpenAIEmbeddings
import numpy as np

class HybridIndex:
    def __init__(self, docs: list[str], emb_model="text-embedding-3-small"):
        self.docs = docs
        self.bm25 = BM25Okapi([d.lower().split() for d in docs])
        self.embeddings = OpenAIEmbeddings(model=emb_model)
        self.vecs = np.array(self.embeddings.embed_documents(docs))

    def bm25_rank(self, q: str, k: int) -> list[int]:
        scores = self.bm25.get_scores(q.lower().split())
        return list(np.argsort(scores)[::-1][:k])

    def dense_rank(self, q: str, k: int) -> list[int]:
        qv = np.array(self.embeddings.embed_query(q))
        sims = self.vecs @ qv / (np.linalg.norm(self.vecs, axis=1) * np.linalg.norm(qv))
        return list(np.argsort(sims)[::-1][:k])

    def search(self, q: str, k_each: int = 50, top_k: int = 10, rrf_k: int = 60):
        bm = self.bm25_rank(q, k_each)
        dn = self.dense_rank(q, k_each)
        fused = rrf([[str(i) for i in bm], [str(i) for i in dn]], k=rrf_k)[:top_k]
        return [(int(doc_id), score) for doc_id, score in fused]
```

Retrieve 50 from each side, fuse, keep top 10. Both searches can run in parallel (`asyncio.gather`).

## LangChain EnsembleRetriever (RRF under the hood)

```python
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
from langchain_qdrant import QdrantVectorStore
from langchain_openai import OpenAIEmbeddings

bm25 = BM25Retriever.from_documents(docs); bm25.k = 50
dense = QdrantVectorStore.from_documents(
    docs, OpenAIEmbeddings(model="text-embedding-3-small"), collection_name="kb"
).as_retriever(search_kwargs={"k": 50})

hybrid = EnsembleRetriever(retrievers=[bm25, dense], weights=[0.4, 0.6], c=60)
results = hybrid.invoke("oauth token refresh 403 error")
```

`weights` feed into RRF as multiplicative weights: `w_i / (k + rank)`. Equal weights = vanilla RRF.

## Weighted Linear Fusion (Alpha)

Useful when you have labeled pairs to tune alpha.

```python
def normalize(scores: np.ndarray) -> np.ndarray:
    lo, hi = scores.min(), scores.max()
    return (scores - lo) / (hi - lo + 1e-9)

def alpha_fuse(bm25_scores: dict[str, float], dense_scores: dict[str, float],
               alpha: float = 0.5) -> list[tuple[str, float]]:
    ids = set(bm25_scores) | set(dense_scores)
    bm = normalize(np.array([bm25_scores.get(i, 0.0) for i in ids]))
    dn = normalize(np.array([dense_scores.get(i, 0.0) for i in ids]))
    fused = alpha * dn + (1 - alpha) * bm
    return sorted(zip(ids, fused.tolist()), key=lambda x: x[1], reverse=True)
```

### Tuning Alpha with a Gold Set

```python
def recall_at_k(retrieved: list[str], relevant: set[str], k: int) -> float:
    return len(set(retrieved[:k]) & relevant) / max(len(relevant), 1)

def tune_alpha(eval_set, retriever_bm25, retriever_dense, k: int = 5):
    best_alpha, best_score = 0.5, 0.0
    for alpha in np.arange(0.0, 1.01, 0.1):
        scores = []
        for q, relevant_ids in eval_set:
            bm = retriever_bm25.scored(q, k=50)
            dn = retriever_dense.scored(q, k=50)
            fused_ids = [i for i, _ in alpha_fuse(dict(bm), dict(dn), alpha)][:k]
            scores.append(recall_at_k(fused_ids, set(relevant_ids), k))
        mean = np.mean(scores)
        if mean > best_score:
            best_score, best_alpha = mean, alpha
    return best_alpha, best_score
```

Typical sweet spot: alpha 0.5-0.75 in English docs. Code/technical corpora skew lower (more lexical).

## Learned Sparse: SPLADE

SPLADE produces sparse vectors with learned term weights (not raw counts). Outperforms BM25 on out-of-domain benchmarks but requires a GPU for indexing.

```python
from transformers import AutoTokenizer, AutoModelForMaskedLM
import torch

tokenizer = AutoTokenizer.from_pretrained("naver/splade-v3")
model = AutoModelForMaskedLM.from_pretrained("naver/splade-v3").eval()

def splade_encode(text: str) -> dict[int, float]:
    toks = tokenizer(text, return_tensors="pt", truncation=True, max_length=256)
    with torch.no_grad():
        logits = model(**toks).logits  # (1, seq, vocab)
    weights = torch.max(
        torch.log(1 + torch.relu(logits)) * toks.attention_mask.unsqueeze(-1),
        dim=1,
    ).values.squeeze(0)  # (vocab,)
    nonzero = torch.nonzero(weights).squeeze(-1)
    return {int(i): float(weights[i]) for i in nonzero}
```

Index the sparse dict in a sparse-capable store (Pinecone, Qdrant, Vespa, Elasticsearch).

## Native Hybrid in Managed Vector DBs

### Pinecone (sparse-dense hybrid)

```python
from pinecone import Pinecone

pc = Pinecone()
index = pc.index("hybrid-kb")

index.upsert([{
    "id": "doc1",
    "values": dense_vec,            # dense
    "sparse_values": {              # sparse (BM25 or SPLADE)
        "indices": [3, 17, 42],
        "values": [0.9, 0.6, 0.3],
    },
    "metadata": {"source": "manual"},
}])

# Alpha weights dense vs sparse
def hybrid_scale(dense, sparse, alpha):
    return (
        [v * alpha for v in dense],
        {"indices": sparse["indices"], "values": [v * (1 - alpha) for v in sparse["values"]]},
    )

d, s = hybrid_scale(query_dense, query_sparse, alpha=0.7)
results = index.query(vector=d, sparse_vector=s, top_k=10)
```

### Qdrant (named vectors + sparse)

```python
from qdrant_client import QdrantClient
from qdrant_client.models import (
    VectorParams, SparseVectorParams, Distance, SparseVector,
    Prefetch, FusionQuery, Fusion,
)

client = QdrantClient(url="http://localhost:6333")
client.create_collection(
    "kb",
    vectors_config={"dense": VectorParams(size=1536, distance=Distance.COSINE)},
    sparse_vectors_config={"sparse": SparseVectorParams()},
)

# Query API v1.10+ with native RRF fusion
result = client.query_points(
    collection_name="kb",
    prefetch=[
        Prefetch(query=query_dense, using="dense", limit=50),
        Prefetch(query=SparseVector(indices=idx, values=vals), using="sparse", limit=50),
    ],
    query=FusionQuery(fusion=Fusion.RRF),
    limit=10,
)
```

### Weaviate (alpha parameter)

```python
import weaviate
client = weaviate.connect_to_local()
coll = client.collections.get("KB")

result = coll.query.hybrid(
    query="oauth token refresh",
    alpha=0.75,        # 0 = pure BM25, 1 = pure vector
    fusion_type=weaviate.classes.query.HybridFusion.RELATIVE_SCORE,  # or RANKED (RRF)
    limit=10,
)
```

## Parallel Execution Pattern

```python
import asyncio

async def hybrid_async(bm25_ret, dense_ret, q: str,