Module 5: Hybrid Search — combining keyword + semantic for queries that need both
Capsule 08: Capstone project — a hybrid search engine with A/B testing
Project description
This is the close of the module. You're going to build a working hybrid search engine that combines BM25 + semantic + RRF in an architecture you can take to production. This isn't a tutorial — it's the project that goes into your portfolio or your real codebase.
The system implements dual retrieval (with rank_bm25 for BM25 and ChromaDB for semantic), fusion with RRF, and an A/B testing script that compares the quality against the semantic-only baseline. When you close the project you'll have a reusable system + a report that justifies with data why hybrid search is the right call for your case (or, if the data shows it, why it isn't).
By the end of this project you'll have:
- ✅ A hybrid engine with a reusable interface (your pipeline can swap strategies)
- ✅ Your own eval set with ground truth for validation
- ✅ A reproducible A/B test over your real corpus
- ✅ A comparison report (semantic-only vs hybrid) with clear metrics
- ✅ Documentation that justifies the final decision
- ✅ A RAG pipeline integrated with the winning strategy
Estimated time: 3-4 hours for implementation + 1 hour for analysis.
Project architecture
hybrid_search_project/
├── src/
│ ├── retrievers/
│ │ ├── __init__.py
│ │ ├── base.py # Retriever interface
│ │ ├── semantic.py # ChromaDB semantic
│ │ ├── bm25.py # rank_bm25 keyword
│ │ └── hybrid.py # Hybrid with RRF
│ ├── fusion/
│ │ └── rrf.py # Reciprocal Rank Fusion
│ ├── evaluation/
│ │ ├── eval_set.py
│ │ ├── metrics.py
│ │ └── ab_test.py
│ └── pipeline.py # The integrated RAG pipeline
├── data/
│ ├── corpus/ # Your documents
│ └── golden_set.json # The eval set with ground truth
├── benchmarks/
│ ├── run_benchmark.py
│ └── reports/
├── .env
└── requirements.txt
Step 1: the unified retriever interface
# src/retrievers/base.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List
@dataclass
class RetrievalResult:
doc_id: str
score: float
document: str
metadata: dict
class Retriever(ABC):
"""Unified interface for every retriever."""
@abstractmethod
def search(self, query: str, top_k: int) -> List[RetrievalResult]:
"""Returns documents ranked by relevance."""
pass
@property
@abstractmethod
def name(self) -> str:
passStep 2: the retriever implementations
The semantic retriever
# src/retrievers/semantic.py
import chromadb
from chromadb.utils import embedding_functions
import os
from .base import Retriever, RetrievalResult
class SemanticRetriever(Retriever):
"""A retriever using ChromaDB + OpenAI embeddings."""
def __init__(self, collection_name: str = "rag_docs", db_path: str = "./chroma_db"):
self._embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
api_key=os.getenv("OPENAI_API_KEY"),
model_name="text-embedding-3-small",
)
client = chromadb.PersistentClient(path=db_path)
self.collection = client.get_collection(
name=collection_name,
embedding_function=self._embedding_fn,
)
@property
def name(self) -> str:
return "semantic_chromadb"
def search(self, query: str, top_k: int = 30) -> List[RetrievalResult]:
results = self.collection.query(query_texts=[query], n_results=top_k)
return [
RetrievalResult(
doc_id=results["ids"][0][i],
score=1.0 - results["distances"][0][i], # cosine sim, not distance
document=results["documents"][0][i],
metadata=results["metadatas"][0][i] or {},
)
for i in range(len(results["ids"][0]))
]The BM25 retriever
# src/retrievers/bm25.py
import re
from rank_bm25 import BM25Okapi
from .base import Retriever, RetrievalResult
def technical_tokenizer(text: str) -> list[str]:
"""A tokenizer that preserves CamelCase, snake_case, error codes."""
text_lower = text.lower()
tokens = re.findall(r'\b\w+\b', text_lower)
# CamelCase split
camel_matches = re.findall(r'[A-Z][a-z]+|[a-z]+|\d+', text)
tokens.extend([m.lower() for m in camel_matches])
# Identifiers with underscores
underscore_tokens = re.findall(r'\b\w+_\w+\b', text_lower)
tokens.extend(underscore_tokens)
# Error codes (ALL_CAPS_WITH_NUMBERS)
code_tokens = re.findall(r'[A-Z][A-Z_0-9]{2,}', text)
tokens.extend([t.lower() for t in code_tokens])
return tokens
class BM25Retriever(Retriever):
"""A retriever using in-memory rank_bm25."""
def __init__(self, documents: list[str], doc_ids: list[str], metadatas: list[dict]):
assert len(documents) == len(doc_ids) == len(metadatas)
self.documents = documents
self.doc_ids = doc_ids
self.metadatas = metadatas
# Build the index
tokenized_corpus = [technical_tokenizer(doc) for doc in documents]
self.bm25 = BM25Okapi(tokenized_corpus)
@property
def name(self) -> str:
return "bm25_rank_bm25"
def search(self, query: str, top_k: int = 30) -> List[RetrievalResult]:
query_tokens = technical_tokenizer(query)
scores = self.bm25.get_scores(query_tokens)
# Top-k indices
top_indices = sorted(
range(len(scores)),
key=lambda i: -scores[i],
)[:top_k]
return [
RetrievalResult(
doc_id=self.doc_ids[i],
score=float(scores[i]),
document=self.documents[i],
metadata=self.metadatas[i],
)
for i in top_indices if scores[i] > 0 # drop the matches scoring 0
]The hybrid retriever with RRF
# src/fusion/rrf.py
from typing import List
from collections import defaultdict
def reciprocal_rank_fusion(
rankings: List[List[str]],
k: int = 60,
) -> List[tuple[str, float]]:
"""Fuses N rankings using Reciprocal Rank Fusion."""
rrf_scores: dict[str, float] = defaultdict(float)
for ranking in rankings:
for rank, doc_id in enumerate(ranking, start=1):
rrf_scores[doc_id] += 1.0 / (k + rank)
return sorted(rrf_scores.items(), key=lambda x: -x[1])
# src/retrievers/hybrid.py
from concurrent.futures import ThreadPoolExecutor
from .base import Retriever, RetrievalResult
from .semantic import SemanticRetriever
from .bm25 import BM25Retriever
from src.fusion.rrf import reciprocal_rank_fusion
class HybridRetriever(Retriever):
"""Hybrid retrieval with BM25 + semantic + RRF."""
def __init__(self, semantic: SemanticRetriever, bm25: BM25Retriever, rrf_k: int = 60):
self.semantic = semantic
self.bm25 = bm25
self.rrf_k = rrf_k
@property
def name(self) -> str:
return f"hybrid_rrf_k{self.rrf_k}"
def search(self, query: str, top_k: int = 5, candidates_per_method: int = 30) -> List[RetrievalResult]:
# Queries in parallel (they're I/O bound)
with ThreadPoolExecutor(max_workers=2) as executor:
future_sem = executor.submit(self.semantic.search, query, candidates_per_method)
future_bm25 = executor.submit(self.bm25.search, query, candidates_per_method)
sem_results = future_sem.result()
bm25_results = future_bm25.result()
# Extract the rankings (a sorted list of IDs)
sem_ranking = [r.doc_id for r in sem_results]
bm25_ranking = [r.doc_id for r in bm25_results]
# RRF
fused = reciprocal_rank_fusion([sem_ranking, bm25_ranking], k=self.rrf_k)
# Build the final results: I need the content of each doc
# (any retriever should have it; I use semantic for its access to metadata)
all_docs = {r.doc_id: r for r in sem_results}
for r in bm25_results:
if r.doc_id not in all_docs:
all_docs[r.doc_id] = r
results = []
for doc_id, rrf_score in fused[:top_k]:
if doc_id in all_docs:
doc = all_docs[doc_id]
results.append(RetrievalResult(
doc_id=doc_id,
score=rrf_score,
document=doc.document,
metadata=doc.metadata,
))
return resultsStep 3: the eval set with ground truth
# src/evaluation/eval_set.py
import json
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import List
@dataclass
class EvalQuery:
query: str
expected_doc_ids: List[str]
category: str = "general" # e.g. "exact_match", "conceptual", "mixed"
def load_eval_set(path: str = "data/golden_set.json") -> List[EvalQuery]:
with open(path) as f:
data = json.load(f)
return [EvalQuery(**item) for item in data]
def save_eval_set(queries: List[EvalQuery], path: str):
with open(path, "w") as f:
json.dump([asdict(q) for q in queries], f, indent=2)An example eval set:
[
{
"query": "OAuth2PasswordBearer scopes",
"expected_doc_ids": ["doc_42", "doc_87"],
"category": "exact_match"
},
{
"query": "how do I authenticate users in my API",
"expected_doc_ids": ["doc_42", "doc_103", "doc_215"],
"category": "conceptual"
},
{
"query": "ERR_NETWORK_TIMEOUT_504",
"expected_doc_ids": ["doc_503"],
"category": "exact_match"
}
]Step 4: metrics and the A/B test
# src/evaluation/metrics.py
from typing import List
import time
import statistics
from src.retrievers.base import Retriever
from src.evaluation.eval_set import EvalQuery
def precision_at_k(retrieved_ids: List[str], expected_ids: List[str], k: int) -> float:
top_k = retrieved_ids[:k]
return sum(1 for doc_id in top_k if doc_id in expected_ids) / k
def recall_at_k(retrieved_ids: List[str], expected_ids: List[str], k: int) -> float:
if not expected_ids:
return 0.0
top_k = retrieved_ids[:k]
return sum(1 for doc_id in top_k if doc_id in expected_ids) / len(expected_ids)
def mrr(retrieved_ids: List[str], expected_ids: List[str]) -> float:
for i, doc_id in enumerate(retrieved_ids, 1):
if doc_id in expected_ids:
return 1.0 / i
return 0.0
def evaluate_retriever(
retriever: Retriever,
eval_set: List[EvalQuery],
top_k: int = 5,
) -> dict:
"""Evaluates one retriever over the full eval set."""
precisions, recalls, mrrs, latencies = [], [], [], []
by_category = {}
for item in eval_set:
start = time.perf_counter()
results = retriever.search(item.query, top_k=top_k)
elapsed = (time.perf_counter() - start) * 1000
retrieved_ids = [r.doc_id for r in results]
p = precision_at_k(retrieved_ids, item.expected_doc_ids, top_k)
r = recall_at_k(retrieved_ids, item.expected_doc_ids, top_k)
m = mrr(retrieved_ids, item.expected_doc_ids)
precisions.append(p)
recalls.append(r)
mrrs.append(m)
latencies.append(elapsed)
# Metrics by category
if item.category not in by_category:
by_category[item.category] = {"p": [], "r": [], "m": []}
by_category[item.category]["p"].append(p)
by_category[item.category]["r"].append(r)
by_category[item.category]["m"].append(m)
# Global stats
latencies.sort()
return {
"name": retriever.name,
"precision_at_k": statistics.mean(precisions),
"recall_at_k": statistics.mean(recalls),
"mrr": statistics.mean(mrrs),
"latency_p50_ms": latencies[len(latencies) // 2],
"latency_p95_ms": latencies[int(len(latencies) * 0.95)],
"by_category": {
cat: {
"precision": statistics.mean(stats["p"]),
"recall": statistics.mean(stats["r"]),
"mrr": statistics.mean(stats["m"]),
"n_queries": len(stats["p"]),
}
for cat, stats in by_category.items()
},
"n_total_queries": len(eval_set),
}# src/evaluation/ab_test.py
from src.retrievers.semantic import SemanticRetriever
from src.retrievers.bm25 import BM25Retriever
from src.retrievers.hybrid import HybridRetriever
from src.evaluation.metrics import evaluate_retriever
from src.evaluation.eval_set import load_eval_set
def run_ab_test():
"""Runs the A/B test comparing semantic-only vs hybrid."""
# Load the eval set
eval_set = load_eval_set()
print(f"Eval set: {len(eval_set)} queries")
# Set up the retrievers
semantic = SemanticRetriever()
# For BM25 we need the documents
all_docs = semantic.collection.get()
bm25 = BM25Retriever(
documents=all_docs["documents"],
doc_ids=all_docs["ids"],
metadatas=all_docs["metadatas"],
)
hybrid = HybridRetriever(semantic, bm25, rrf_k=60)
# Evaluate each one
results = {}
for retriever in [semantic, bm25, hybrid]:
print(f"\nEvaluating {retriever.name}...")
results[retriever.name] = evaluate_retriever(retriever, eval_set, top_k=5)
return results
def print_report(results: dict):
"""Prints a readable comparison report."""
print("\n" + "=" * 90)
print("A/B TEST REPORT — Semantic vs BM25 vs Hybrid")
print("=" * 90)
print(f"\n{'Retriever':<25} {'P@5':>10} {'R@5':>10} {'MRR':>10} {'p50 ms':>10} {'p95 ms':>10}")
print("-" * 90)
for name, metrics in results.items():
print(
f"{name:<25} "
f"{metrics['precision_at_k']:>9.1%} "
f"{metrics['recall_at_k']:>9.1%} "
f"{metrics['mrr']:>9.3f} "
f"{metrics['latency_p50_ms']:>9.0f} "
f"{metrics['latency_p95_ms']:>9.0f}"
)
# By category (hybrid vs semantic only)
print("\n\nBreakdown by category (Recall@5):\n")
print(f"{'Category':<20} {'Semantic':>15} {'Hybrid':>15} {'Gain':>15}")
print("-" * 70)
sem_by_cat = results.get("semantic_chromadb", {}).get("by_category", {})
hyb_by_cat = results.get("hybrid_rrf_k60", {}).get("by_category", {})
for cat in sem_by_cat:
if cat in hyb_by_cat:
sem_recall = sem_by_cat[cat]["recall"]
hyb_recall = hyb_by_cat[cat]["recall"]
diff = (hyb_recall - sem_recall) * 100
print(
f"{cat:<20} "
f"{sem_recall:>14.1%} "
f"{hyb_recall:>14.1%} "
f"{diff:>+13.1f}p"
)Step 5: run it and report
# benchmarks/run_benchmark.py
import json
from datetime import datetime
from pathlib import Path
from src.evaluation.ab_test import run_ab_test, print_report
def main():
results = run_ab_test()
print_report(results)
# Save the JSON report for auditing
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
report_dir = Path("benchmarks/reports")
report_dir.mkdir(parents=True, exist_ok=True)
report_path = report_dir / f"ab_test_{timestamp}.json"
with open(report_path, "w") as f:
json.dump(results, f, indent=2)
print(f"\nReport saved: {report_path}")
if __name__ == "__main__":
main()Expected output:
Eval set: 80 queries
Evaluating semantic_chromadb...
Evaluating bm25_rank_bm25...
Evaluating hybrid_rrf_k60...
==========================================================================================
A/B TEST REPORT — Semantic vs BM25 vs Hybrid
==========================================================================================
Retriever P@5 R@5 MRR p50 ms p95 ms
------------------------------------------------------------------------------------------
semantic_chromadb 82.5% 68.4% 0.745 185 245
bm25_rank_bm25 78.2% 62.1% 0.681 25 48
hybrid_rrf_k60 91.7% 83.6% 0.853 215 310
Breakdown by category (Recall@5):
Category Semantic Hybrid Gain
----------------------------------------------------------------------
exact_match 54.2% 88.6% +34.4p
conceptual 85.1% 87.4% +2.3p
mixed 71.0% 82.2% +11.2p
Step 6: analysis and the final report
With the benchmark data in hand, write a decision report:
# Decision: Hybrid retrieval with RRF
## Key results
- **Hybrid improves recall by +15.2 points** over semantic-only (68.4% → 83.6%).
- **A dramatic improvement on exact-match queries: +34 points** (54% → 88%).
- **A moderate improvement on mixed: +11 points.**
- **Purely conceptual queries barely change (+2pts)** — semantic already wins there.
## Why hybrid wins on this corpus
The query log shows that ~50% of the traffic has exact-match components
(code identifiers, error codes, commands). For that category, BM25 finds the
specific docs that cosine similarity was subordinating to close paraphrases.
## The cost of the change
- **Latency:** +30ms (from 245ms to 310ms p95). Acceptable for our <500ms SLA.
- **Recurring cost:** $0 (rank_bm25 runs locally).
- **Maintenance:** medium (a custom tokenizer, monitoring quality).
## Recommendation
**Deploy hybrid retrieval with RRF (k=60).** The A/B test produces a measurable gain on
the two metrics that matter (recall + MRR), with no precision drop, and within the SLA.
## Rollout plan
1. Feature flag with 10% of traffic → monitor for 3 days.
2. If the production metrics confirm the benchmark's, scale to 50% → 100%.
3. Protective metric: precision@5 must not drop >2% vs semantic-only.
## Plan B if production doesn't replicate it
- If recall improves on the eval set but NOT in production: the real queries may
differ from the eval set's. Investigate the log and refine the eval set.
- If latency rises more than expected: parallelize the queries with ThreadPoolExecutor
(already implemented), or reduce candidates_per_method.Project delivery checklist
Before you consider the project closed:
- The 3 retrievers implemented behind a unified interface.
- An eval set of 50+ queries with ground truth and categories.
- RRF correctly implemented and tested.
- A runnable A/B test with
python benchmarks/run_benchmark.py. - A comparison report with a per-category table.
- An updated RAG pipeline with the winning strategy.
- A README with setup, the benchmark command, and the decision.
- Tests (sanity checks for each retriever at minimum).
Optional extensions
Extension 1: add weighted blending
After the base project, add a weighted retriever (capsule 05) and compare it against RRF. If it wins by >3% over RRF, consider deploying it.
Extension 2: dynamic α routing
Detect the query type and pick the appropriate α (capsule 05). Typical gain: +2-3% on a mixed corpus.
Extension 3: add re-ranking as a cascade
Hybrid retrieve top-30 → cross-encoder rerank → top-5. Typical gain: +5-8% additional precision.
Extension 4: migrate BM25 to Elasticsearch
When the corpus passes 1M docs, migrate from rank_bm25 to Elasticsearch (capsule 06). Without changing the Retriever interface — only the implementation.
Traps and common mistakes in the project
Trap 1: an eval set that's small or all one type
If all your eval queries are semantic, hybrid won't show any advantage over semantic. The eval set must reflect the real distribution of the production log.
Trap 2: comparing against an unfair baseline
If the baseline has a rerank and the hybrid version doesn't, the comparison is invalid. Keep the rest of the pipeline identical between the two versions.
Trap 3: different tokenization between indexing and query
If the corpus was indexed with text.lower().split() and the query uses technical_tokenizer, the matches fail. Same tokenizer on both sides.
Trap 4: forgetting to parallelize the queries
Without parallelization, hybrid latency = semantic latency + BM25 latency. With ThreadPoolExecutor, hybrid latency = the max of the two.
Trap 5: deploying without an A/B in production
Even if the benchmark says hybrid wins, run an A/B test in production for 1-2 weeks before a 100% rollout. The real queries can have a different distribution from the eval set's.
Project summary
You built:
- ✅ A modular retrieval system with a unified interface
- ✅ Three implementations tested side by side (semantic, BM25, hybrid)
- ✅ A robust eval set with ground truth and categorization
- ✅ A reproducible benchmark with metrics per query type
- ✅ A report that justifies the decision with data
- ✅ An updated RAG pipeline, ready to deploy
The typical improvement expected with hybrid integrated:
Metric Semantic-only Hybrid Gain
Precision@5 82% 92% +10 pts
Recall@5 68% 84% +16 pts
MRR 0.74 0.85 +0.11
Recall on exact_match 54% 88% +34 pts
That closes the hybrid search module and sets you up for M06 (metadata filtering), where you combine hybrid + metadata filters for advanced retrieval.
Next steps
Right away:
- Run the benchmark over your own eval set.
- Decide whether hybrid is worth it for your corpus (if the gain is >5% in recall, it usually is).
- Integrate it into the RAG pipeline and deploy to staging.
- A/B test in production for 1-2 weeks.
Module 6 (Metadata Filtering):
Hybrid retrieve + a metadata filter is a very powerful pattern: you filter first by category/language/date (which shrinks the search space dramatically), then run hybrid over the filtered subset. M06 covers the pattern.
Resources
- rank-bm25 — The library used here
- ChromaDB Documentation — The vector DB
- LangChain — EnsembleRetriever — A reference pattern
- RRF Paper — The fusion algorithm
- BEIR Benchmark — Empirical comparisons
- Anthropic — Contextual Retrieval — A complementary technique
Estimated time: 3-4 hours + 1 hour of analysis Next module: Module 6 — Metadata Filtering