Módulo 5: Hybrid Search — combinando keyword + semantic para queries que necesitan ambas
Cápsula 08: Proyecto integrador — hybrid search engine con A/B testing
Descripción del proyecto
Este es el cierre del módulo. Vas a construir un motor hybrid search funcional que combina BM25 + semantic + RRF en una arquitectura que puedes llevar a producción. No es un tutorial — es el proyecto que va a tu portfolio o a tu codebase real.
El sistema implementa retrieval dual (con rank_bm25 para BM25 e ChromaDB para semantic), fusión con RRF, y un script de A/B testing que compara la calidad contra semantic-only baseline. Al cerrar el proyecto, vas a tener un sistema reusable + un reporte que justifica con datos por qué hybrid search es la elección correcta para tu caso (o, si los datos lo demuestran, por qué no lo es).
Al finalizar este proyecto vas a tener:
- ✅ Motor hybrid con interfaz reutilizable (tu pipeline puede swappear estrategias)
- ✅ Eval set propio con ground truth para validación
- ✅ A/B test reproducible sobre tu corpus real
- ✅ Reporte comparativo (semantic-only vs hybrid) con métricas claras
- ✅ Documentación que justifica la decisión final
- ✅ Pipeline RAG integrado con la estrategia ganadora
Tiempo estimado: 3-4 horas para implementación + 1 hora para análisis.
Arquitectura del proyecto
hybrid_search_project/
├── src/
│ ├── retrievers/
│ │ ├── __init__.py
│ │ ├── base.py # Interfaz Retriever
│ │ ├── semantic.py # ChromaDB semantic
│ │ ├── bm25.py # rank_bm25 keyword
│ │ └── hybrid.py # Hybrid con RRF
│ ├── fusion/
│ │ └── rrf.py # Reciprocal Rank Fusion
│ ├── evaluation/
│ │ ├── eval_set.py
│ │ ├── metrics.py
│ │ └── ab_test.py
│ └── pipeline.py # Pipeline RAG integrado
├── data/
│ ├── corpus/ # Tus documentos
│ └── golden_set.json # Eval set con ground truth
├── benchmarks/
│ ├── run_benchmark.py
│ └── reports/
├── .env
└── requirements.txt
Paso 1: Interfaz unificada de retrievers
# 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):
"""Interfaz unificada para todos los retrievers."""
@abstractmethod
def search(self, query: str, top_k: int) -> List[RetrievalResult]:
"""Retorna documentos rankeados por relevancia."""
pass
@property
@abstractmethod
def name(self) -> str:
passPaso 2: Implementaciones de retrievers
Semantic retriever
# src/retrievers/semantic.py
import chromadb
from chromadb.utils import embedding_functions
import os
from .base import Retriever, RetrievalResult
class SemanticRetriever(Retriever):
"""Retriever usando 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, no distance
document=results["documents"][0][i],
metadata=results["metadatas"][0][i] or {},
)
for i in range(len(results["ids"][0]))
]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]:
"""Tokenizer que preserva 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])
# Identificadores con underscore
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):
"""Retriever usando rank_bm25 in-memory."""
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
# Construir índice
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 # filtrar matches con score 0
]Hybrid retriever con 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]]:
"""Fusiona N rankings usando 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 con 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 en paralelo (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()
# Extraer rankings (lista de IDs ordenados)
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)
# Construir resultados finales: necesito el contenido de cada doc
# (cualquier retriever debería tenerlo; uso semantic por su acceso a 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 resultsPaso 3: Eval set con 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" # ej: "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)Ejemplo de 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"
}
]Paso 4: Métricas y 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:
"""Evalúa un retriever sobre el eval set completo."""
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)
# Métricas por categoría
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)
# Stats globales
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():
"""Ejecuta A/B test comparando semantic-only vs hybrid."""
# Cargar eval set
eval_set = load_eval_set()
print(f"Eval set: {len(eval_set)} queries")
# Setup retrievers
semantic = SemanticRetriever()
# Para BM25 necesitamos los documentos
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)
# Evaluar cada uno
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):
"""Imprime reporte comparativo legible."""
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}"
)
# Por categoría (solo hybrid vs semantic)
print("\n\nBreakdown por categoría (Recall@5):\n")
print(f"{'Category':<20} {'Semantic':>15} {'Hybrid':>15} {'Mejora':>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"
)Paso 5: Ejecutar y reportar
# 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)
# Guardar reporte JSON para auditoría
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"\nReporte guardado: {report_path}")
if __name__ == "__main__":
main()Output esperado:
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 por categoría (Recall@5):
Category Semantic Hybrid Mejora
----------------------------------------------------------------------
exact_match 54.2% 88.6% +34.4p
conceptual 85.1% 87.4% +2.3p
mixed 71.0% 82.2% +11.2p
Paso 6: Análisis y reporte final
Con los datos del benchmark, escribir un reporte de decisión:
# Decisión: Hybrid retrieval con RRF
## Resultados clave
- **Hybrid mejora recall +15.2 puntos** sobre semantic-only (68.4% → 83.6%).
- **Mejora dramática en queries exact-match: +34 puntos** (54% → 88%).
- **Mejora moderada en mixed: +11 puntos.**
- **Queries puramente conceptuales casi no cambian (+2pts)** — semantic ya gana ahí.
## Por qué hybrid gana en este corpus
El log de queries muestra que ~50% del tráfico tiene componentes exact-match
(identificadores de código, error codes, comandos). Para esa categoría, BM25
encuentra docs específicos que cosine similarity subordinaba a paráfrasis cercanas.
## Costo del cambio
- **Latencia:** +30ms (de 245ms a 310ms p95). Aceptable para nuestro SLA <500ms.
- **Costo recurrente:** $0 (rank_bm25 corre local).
- **Mantenimiento:** medio (tokenizer custom, monitorear quality).
## Recomendación
**Deployar hybrid retrieval con RRF (k=60).** El A/B test produce mejora medible en
las dos métricas que importan (recall + MRR), sin caída en precision, dentro del SLA.
## Plan de rollout
1. Feature flag con 10% del tráfico → monitorear 3 días.
2. Si métricas de producción confirman las del benchmark, escalar a 50% → 100%.
3. Métrica de protección: precision@5 no debe caer >2% vs semantic-only.
## Plan B si producción no replica
- Si recall mejora en eval set pero NO en producción: las queries reales pueden
diferir de las del eval set. Investigar log y refinar eval set.
- Si latencia sube más de lo esperado: paralelizar queries con ThreadPoolExecutor
(ya implementado), o reducir candidates_per_method.Checklist de entrega del proyecto
Antes de considerar el proyecto cerrado:
- Implementación de los 3 retrievers con interfaz unificada.
- Eval set de 50+ queries con ground truth y categorías.
- RRF correctamente implementado y testeado.
- A/B test ejecutable con
python benchmarks/run_benchmark.py. - Reporte comparativo con tabla por categoría.
- Pipeline RAG actualizado con la estrategia ganadora.
- README con setup, comando para benchmark, y decisión.
- Tests (al menos sanity checks de cada retriever).
Extensiones opcionales
Extensión 1: agregar weighted blending
Después del proyecto base, agrega un retriever weighted (cápsula 05) y compara contra RRF. Si gana >3% sobre RRF, considerar deployar.
Extensión 2: routing dinámico de α
Detectar tipo de query y elegir α apropiado (cápsula 05). Mejora típica: +2-3% en corpus mixto.
Extensión 3: agregar re-ranking a cascade
Hybrid retrieve top-30 → cross-encoder rerank → top-5. Mejora típica: +5-8% precision adicional.
Extensión 4: migrar BM25 a Elasticsearch
Cuando el corpus pase 1M docs, migrar de rank_bm25 a Elasticsearch (cápsula 06). Sin cambiar la interfaz Retriever — solo la implementación.
Trampas y errores comunes en el proyecto
Trampa 1: eval set chico o solo de un tipo
Si todas tus queries de eval son semánticas, hybrid no va a mostrar ventaja sobre semantic. Eval set debe reflejar la distribución real del log de producción.
Trampa 2: comparar contra baseline injusto
Si el baseline tiene rerank y la versión hybrid no, la comparación es invalida. Mantener el resto del pipeline igual entre las dos versiones.
Trampa 3: tokenización diferente entre indexing y query
Si el corpus se indexó con text.lower().split() y la query usa el technical_tokenizer, los matches fallan. Mismo tokenizer en ambos lados.
Trampa 4: olvidar paralelizar queries
Sin paralelización, latencia hybrid = latencia semantic + latencia BM25. Con ThreadPoolExecutor, latencia hybrid = max de las dos.
Trampa 5: deployar sin A/B en producción
Aún si el benchmark dice hybrid gana, ejecutar A/B test en producción 1-2 semanas antes de 100% rollout. Las queries reales pueden tener distribución distinta al eval set.
Resumen del proyecto
Construiste:
- ✅ Sistema modular de retrieval con interfaz unificada
- ✅ Tres implementaciones probadas (semantic, BM25, hybrid)
- ✅ Eval set robusto con ground truth y categorización
- ✅ Benchmark reproducible con métricas por tipo de query
- ✅ Reporte que justifica la decisión con datos
- ✅ Pipeline RAG actualizado y listo para deploy
Mejora típica esperada con hybrid integrado:
Métrica Semantic-only Hybrid Mejora
Precision@5 82% 92% +10 pts
Recall@5 68% 84% +16 pts
MRR 0.74 0.85 +0.11
Recall en exact_match 54% 88% +34 pts
Esto cierra el módulo de hybrid search y te prepara para M06 (metadata filtering), donde combinas hybrid + filters por metadata para retrieval avanzado.
Próximos pasos
Inmediato:
- Correr el benchmark sobre tu eval set propio.
- Decidir si hybrid vale para tu corpus (si la mejora es >5% en recall, generalmente sí).
- Integrar al pipeline RAG y deployar a staging.
- A/B test producción 1-2 semanas.
Módulo 6 (Metadata Filtering):
Hybrid retrieve + metadata filter es un patrón muy poderoso: filtras primero por categoría/idioma/fecha (reduce search space dramáticamente), después corres hybrid sobre el subset filtrado. M06 cubre el patrón.
Recursos
- rank-bm25 — Librería usada
- ChromaDB Documentation — Vector DB
- LangChain — EnsembleRetriever — Patrón de referencia
- RRF Paper — Algoritmo de fusión
- BEIR Benchmark — Comparaciones empíricas
- Anthropic — Contextual Retrieval — Técnica complementaria
Tiempo estimado: 3-4 horas + 1 hora de análisis Siguiente módulo: Módulo 6 — Metadata Filtering