Combining vector search with BM25 sparse retrieval using fusion algorithms for improved RAG accuracy
Scope: Vector + sparse retrieval, RRF fusion, parallel architectures, score normalization, metrics Lines: ~450 Last Updated: 2025-10-26
Activate this skill when:
Hybrid Search: Combine dense (vector) + sparse (keyword) retrieval
Why hybrid beats single-method:
RRF Algorithm (Cormack et al., 2009; widely adopted in 2024 RAG systems):
RRF(d) = Σ (1 / (k + rank_i(d)))d: Documentrank_i(d): Rank of document in result set ik: Constant (typically 60)Benefits:
Parallel retrieval (recommended 2024-2025):
Query → [Vector Search] → Results_V
→ [BM25 Search] → Results_B
→ RRF Fusion → Final RankingsSequential retrieval (less common):
Query → Vector Search → Top-K → BM25 Rerank → Final ResultsWeighted fusion (advanced):
Score = α * vector_score + (1-α) * bm25_score
# α tuned via evaluation (typically 0.5-0.7)| Platform | Hybrid Support | Method | |----------|----------------|--------| | Elasticsearch | Native | Dense vector + BM25 fusion | | Weaviate | Native (v1.19+) | Hybrid search API | | Qdrant | Native (v1.7+) | Sparse vector support | | Pinecone | Beta (2024) | Hybrid search indexes | | ChromaDB | Manual | Separate queries + RRF |
Min-Max normalization:
norm_score = (score - min_score) / (max_score - min_score)Z-score normalization:
norm_score = (score - mean) / std_devSoftmax normalization:
norm_score = exp(score) / Σ exp(all_scores)When needed: Weighted fusion (not RRF, which is rank-based)
import dspy
from typing import List, Dict, Tuple
from collections import defaultdict
class RRFHybridRetriever(dspy.Module):
"""Hybrid retrieval using Reciprocal Rank Fusion."""
def __init__(self, k=10, rrf_k=60):
super().__init__()
self.k = k
self.rrf_k = rrf_k
# Two separate retrievers
self.vector_retrieve = dspy.Retrieve(k=k)
# BM25 retriever (custom implementation)
self.bm25_retrieve = BM25Retriever(k=k)
def reciprocal_rank_fusion(
self,
vector_results: List[str],
bm25_results: List[str]
) -> List[str]:
"""Apply RRF to merge two ranked lists."""
scores = defaultdict(float)
# Score from vector results
for rank, doc in enumerate(vector_results, start=1):
scores[doc] += 1.0 / (self.rrf_k + rank)
# Score from BM25 results
for rank, doc in enumerate(bm25_results, start=1):
scores[doc] += 1.0 / (self.rrf_k + rank)
# Sort by combined score
ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return [doc for doc, _ in ranked[:self.k]]
def forward(self, question: str):
# Parallel retrieval
vector_passages = self.vector_retrieve(question).passages
bm25_passages = self.bm25_retrieve(question)
# Apply RRF fusion
fused_passages = self.reciprocal_rank_fusion(
vector_passages, bm25_passages
)
return dspy.Prediction(passages=fused_passages)
class HybridRAG(dspy.Module):
"""RAG with hybrid retrieval."""
def __init__(self, k=10):
super().__init__()
self.retrieve = RRFHybridRetriever(k=k)
self.generate = dspy.ChainOfThought("context, question -> answer")
def forward(self, question: str):
retrieval = self.retrieve(question)
context = "\n\n".join(retrieval.passages)
return self.generate(context=context, question=question)
# Custom BM25 retriever (example using rank_bm25)
from rank_bm25 import BM25Okapi
class BM25Retriever:
"""Simple BM25 retriever for hybrid search."""
def __init__(self, corpus: List[str], k=10):
self.k = k
self.corpus = corpus
# Tokenize corpus
tokenized = [doc.lower().split() for doc in corpus]
self.bm25 = BM25Okapi(tokenized)
def __call__(self, query: str) -> List[str]:
tokenized_query = query.lower().split()
scores = self.bm25.get_scores(tokenized_query)
# Get top-k indices
top_indices = sorted(
range(len(scores)),
key=lambda i: scores[i],
reverse=True
)[:self.k]
return [self.corpus[i] for i in top_indices]When to use:
import dspy
from elasticsearch import Elasticsearch
class ElasticsearchHybridRetriever:
"""Hybrid retrieval using Elasticsearch native capabilities."""
def __init__(self, es_client: Elasticsearch, index: str, k=10):
self.es = es_client
self.index = index
self.k = k
def __call__(self, query: str) -> List[str]:
# Hybrid query: BM25 + dense vector
search_body = {
"query": {
"bool": {
"should": [
# BM25 component
{
"match": {
"content": {
"query": query,
"boost": 0.5 # Weight for BM25
}
}
},
# Dense vector component
{
"script_score": {
"query": {"match_all": {}},
"script": {
"source": "cosineSimilarity(params.query_vector, 'embedding') + 1.0",
"params": {
"query_vector": self._embed(query)
}
},
"boost": 0.5 # Weight for vector
}
}
]
}
},
"size": self.k
}
response = self.es.search(index=self.index, body=search_body)
passages = [hit["_source"]["content"] for hit in response["hits"]["hits"]]
return passages
def _embed(self, text: str) -> List[float]:
"""Generate embedding for text (use your embedding model)."""
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("all-MiniLM-L6-v2")
return model.encode(text).tolist()
# Use with DSPy
class ESHybridRAG(dspy.Module):
def __init__(self, es_client: Elasticsearch, index: str):
super().__init__()
self.retrieve = ElasticsearchHybridRetriever(es_client, index, k=5)
self.generate = dspy.ChainOfThought("context, question -> answer")
def forward(self, question: str):
passages = self.retrieve(question)
context = "\n\n".join(passages)
return self.generate(context=context, question=question)When to use:
import dspy
import weaviate
class WeaviateHybridRetriever:
"""Hybrid retrieval using Weaviate native hybrid search."""
def __init__(self, client: weaviate.Client, class_name: str, k=10, alpha=0.5):
self.client = client
self.class_name = class_name
self.k = k
self.alpha = alpha # 0 = pure BM25, 1 = pure vector
def __call__(self, query: str) -> List[str]:
result = (
self.client.query
.get(self.class_name, ["content"])
.with_hybrid(
query=query,
alpha=self.alpha, # Balance vector vs BM25
fusion_type="relativeScoreFusion" # or "rankedFusion"
)
.with_limit(self.k)
.do()
)
passages = [
item["content"]
for item in result["data"]["Get"][self.class_name]
]
return passages
# Use with DSPy
class WeaviateHybridRAG(dspy.Module):
def __init__(self, weaviate_client: weaviate.Client, class_name: str):
super().__init__()
self.retrieve = WeaviateHybridRetriever(weaviate_client, class_name, k=5)
self.generate = dspy.ChainOfThought("context, question -> answer")
def forward(self, question: str):
passages = self.retrieve(question)
context = "\n\n".join(passages)
return self.generate(context=context, question=question)When to use:
import dspy
from qdrant_client import QdrantClient
from qdrant_client.models import SparseVector, NamedSparseVector
class QdrantHybridRetriever:
"""Hybrid retrieval using Qdrant sparse + dense vectors."""
def __init__(self, client: QdrantClient, collection: str, k=10):
self.client = client
self.collection = collection
self.k = k
def __call__(self, query: str) -> List[str]:
# Generate dense and sparse vectors
dense_vector = self._embed_dense(query)
sparse_vector = self._embed_sparse(query)
# Hybrid search
results = self.client.search(
collection_name=self.collection,
query_vector=dense_vector,
query_filter=None,
sparse_vector=NamedSparseVector(
name="sparse",
vector=sparse_vector
),
limit=self.k
)
passages = [hit.payload["content"] for hit in results]
return passages
def _embed_dense(self, text: str):
"""Dense embedding (e.g., sentence-transformers)."""
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("all-MiniLM-L6-v2")
return model.encode(text).tolist()
def _embed_sparse(self, text: str) -> SparseVector:
"""Sparse embedding (e.g., BM25-style)."""
# Use SPLADE or similar sparse encoder
# Simplified example: TF-IDF style
from sklearn.feature_extraction.text import TfidfVectorizer
vectorizer = TfidfVectorizer(max_features=1000)
# Fit on corpus (do this once, not per query)
sparse_vec = vectorizer.transform([text])
indices = sparse_vec.indices.tolist()
values = sparse_vec.data.tolist()
return SparseVector(indices=indices, values=values)When to use:
import dspy
import numpy as np
from typing import List, Tuple
class WeightedHybridRetriever(dspy.Module):
"""Hybrid retrieval with weighted score fusion."""
def __init__(self, k=10, alpha=0.6):
super().__init__()
self.k = k
self.alpha = alpha # Weight for vector scores
self.vector_retrieve = dspy.Retrieve(k=k)
self.bm25_retrieve = BM25Retriever(k=k)
def normalize_scores(self, scores: List[float]) -> List[float]:
"""Min-max normalization to [0, 1]."""
scores = np.array(scores)
min_score = scores.min()
max_score = scores.max()
if max_score == min_score:
return [1.0] * len(scores)
return ((scores - min_score) / (max_score - min_score)).tolist()
def forward(self, question: str):
# Get results with scores
vector_results = self.vector_retrieve(question)
bm25_results = self.bm25_retrieve.search_with_scores(question)
# Normalize scores separately
vector_scores = self.normalize_scores([r.score for r in vector_results])
bm25_scores = self.normalize_scores([r.score for r in bm25_results])
# Create combined score dict
combined = {}
for doc, score in zip(vector_results.passages, vector_scores):
combined[doc] = self.alpha * score
for doc, score in zip(bm25_results.passages, bm25_scores):
if doc in combined:
combined[doc] += (1 - self.alpha) * score
else:
combined[doc] = (1 - self.alpha) * score
# Sort and return top-k
ranked = sorted(combined.items(), key=lambda x: x[1], reverse=True)
passages = [doc for doc, _ in ranked[:self.k]]
return dspy.Prediction(passages=passages)When to use:
import dspy
from phoenix.trace import dsl as trace_dsl
from phoenix.evals import RetrievalEvaluator
class ObservableHybridRAG(dspy.Module):
"""Hybrid RAG with Phoenix observability."""
def __init__(self, k=10):
super().__init__()
self.retrieve = RRFHybridRetriever(k=k)
self.generate = dspy.ChainOfThought("context, question -> answer")
# Initialize Phoenix evaluator
self.evaluator = RetrievalEvaluator()
def forward(self, question: str):
# Trace retrieval
with trace_dsl.span("hybrid_retrieval"):
retrieval = self.retrieve(question)
# Log retrieval metrics
trace_dsl.log_attribute("num_passages", len(retrieval.passages))
trace_dsl.log_attribute("fusion_method", "RRF")
context = "\n\n".join(retrieval.passages)
# Trace generation
with trace_dsl.span("generation"):
result = self.generate(context=context, question=question)
# Attach retrieved passages for evaluation
result.retrieved_passages = retrieval.passages
return result
# Evaluate retrieval quality
def evaluate_hybrid_retrieval(rag_system, test_set):
"""Evaluate hybrid RAG with Phoenix metrics."""
from phoenix.evals import RetrievalEvaluator, HitRate, MRR, NDCG
evaluator = RetrievalEvaluator(
metrics=[
HitRate(k=5), # Hit rate @ 5
MRR(), # Mean Reciprocal Rank
NDCG(k=10) # Normalized Discounted Cumulative Gain
]
)
results = []
for example in test_set:
prediction = rag_system(question=example.question)
# Evaluate retrieval
metrics = evaluator.evaluate(
retrieved_docs=prediction.retrieved_passages,
relevant_docs=example.relevant_docs,
query=example.question
)
results.append(metrics)
# Aggregate metrics
avg_hit_rate = np.mean([r["hit_rate@5"] for r in results])
avg_mrr = np.mean([r["mrr"] for r in results])
avg_ndcg = np.mean([r["ndcg@10"] for r in results])
print(f"Hit Rate@5: {avg_hit_rate:.3f}")
print(f"MRR: {avg_mrr:.3f}")
print(f"NDCG@10: {avg_ndcg:.3f}")
return resultsMetrics explained:
def rrf_score(rank, k=60):
return 1.0 / (k + rank)Query has rare entities/codes? → Increase BM25 weight (alpha < 0.5)
Query is semantic/paraphrased? → Increase vector weight (alpha > 0.7)
Balanced query? → Use RRF or alpha = 0.5Elasticsearch: Best for large-scale production (>10M docs)
Weaviate: Best for managed hybrid + GraphQL
Qdrant: Best for custom sparse encoders (SPLADE)
ChromaDB: Use manual RRF (no native hybrid)1. NDCG@10 (overall quality)
2. MRR (first relevant result)
3. Hit Rate@5 (basic coverage)❌ Using only vector search for technical queries:
# Bad - misses exact product codes
retrieve = dspy.Retrieve(k=5) # Vector only
result = retrieve("Find part number ABC-12345")✅ Use hybrid:
# Good - combines semantic + exact match
retrieve = RRFHybridRetriever(k=5)
result = retrieve("Find part number ABC-12345")❌ Ignoring score normalization with weighted fusion:
# Bad - raw scores have different scales
combined_score = vector_score + bm25_score # Invalid!✅ Normalize first:
# Good - normalize then combine
norm_vector = normalize(vector_score)
norm_bm25 = normalize(bm25_score)
combined_score = 0.6 * norm_vector + 0.4 * norm_bm25❌ Not tuning alpha parameter:
# Bad - using default without evaluation
retriever = WeaviateHybridRetriever(alpha=0.5) # Is 0.5 optimal?✅ Evaluate and tune:
# Good - grid search for optimal alpha
for alpha in [0.3, 0.5, 0.7, 0.9]:
retriever = WeaviateHybridRetriever(alpha=alpha)
score = evaluate(retriever, test_set)
# Choose best alpha❌ Skipping BM25 for specialized domains:
# Bad - vector search alone for medical/legal/technical
retrieve = dspy.Retrieve(k=5) # Misses terminology✅ Use hybrid for domain-specific:
# Good - hybrid catches exact medical terms
retrieve = RRFHybridRetriever(k=5)dspy-rag.md - Basic RAG patterns and vector retrievalrag-reranking-techniques.md - Multi-stage retrieval with rerankinggraph-rag.md - Graph-based retrieval for multihop reasoninghierarchical-rag.md - Multi-level document structuresdatabase/postgres-query-optimization.md - Metadata filteringdatabase/redis-basics.md - Caching retrieval resultsHybrid search combines the strengths of vector (semantic) and BM25 (lexical) retrieval:
Hybrid search typically achieves 15-30% improvement over single-method retrieval in production RAG systems (2024-2025 benchmarks).
Last Updated: 2025-10-26 Format Version: 1.0 (Atomic)