Módulo 4: Re-ranking — la segunda etapa que transforma retrieval mediocre en excelente
Cápsula 08: Proyecto integrador — sistema de re-ranking con A/B testing
Descripción del proyecto
Este es el cierre del módulo. Vas a construir un sistema de re-ranking production-ready que toma todo lo aprendido en las cápsulas 02-07 y lo integra en un pipeline funcional con observabilidad. No es un tutorial — es el proyecto que va a tu portfolio o a tu codebase real.
El sistema implementa las tres técnicas (cross-encoder, LLM, Cohere) detrás de una interfaz unificada, ejecuta un A/B test sobre un eval set propio, mide precision/recall/latencia, y produce un reporte que justifica con datos qué técnica deployar.
Al finalizar este proyecto vas a tener:
- ✅ Sistema de re-ranking con interfaz
Rerankerreutilizable - ✅ Tres implementaciones (cross-encoder local, LLM, Cohere) con fallback
- ✅ Eval set construido sobre tu corpus con ground truth
- ✅ Script de A/B testing que compara las técnicas
- ✅ Reporte comparativo que justifica la decisión final
- ✅ Pipeline RAG actualizado con la técnica ganadora integrada
Tiempo estimado: 2-3 horas para implementación + 1 hora para análisis.
Arquitectura del proyecto
proyecto_reranking/
├── src/
│ ├── rerankers/
│ │ ├── __init__.py
│ │ ├── base.py # Interfaz Reranker
│ │ ├── cross_encoder.py # Cross-encoder local
│ │ ├── llm_based.py # LLM con structured outputs
│ │ ├── cohere_managed.py # Cohere Rerank API
│ │ └── factory.py # Factory para crear rerankers
│ ├── retrieval/
│ │ ├── pipeline.py # Pipeline con re-ranking
│ │ └── chromadb_setup.py
│ └── evaluation/
│ ├── eval_set.py # Construcción y carga del eval set
│ ├── metrics.py # Precision, Recall, MRR, latency
│ └── ab_test.py # A/B testing entre rerankers
├── eval_data/
│ └── golden_set.json # Eval set con ground truth
├── benchmarks/
│ ├── run_benchmark.py
│ └── reports/ # Reportes generados
├── .env
└── requirements.txt
Paso 1: Interfaz unificada
Crear una abstracción que permita intercambiar rerankers sin cambiar el resto del código.
# src/rerankers/base.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List
@dataclass
class RerankResult:
document: str
score: float
original_index: int
class Reranker(ABC):
"""Interfaz unificada para todos los rerankers."""
@abstractmethod
def rerank(
self,
query: str,
documents: List[str],
top_k: int = 5,
) -> List[RerankResult]:
"""Re-rankea documentos por relevancia a la query."""
pass
@property
@abstractmethod
def name(self) -> str:
"""Nombre identificador del reranker."""
passPaso 2: Implementaciones concretas
Cross-encoder (cápsula 03 reutilizada)
# src/rerankers/cross_encoder.py
from sentence_transformers import CrossEncoder
from .base import Reranker, RerankResult
class CrossEncoderReranker(Reranker):
"""Re-ranker local con cross-encoder de sentence-transformers."""
def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-12-v2"):
self.model = CrossEncoder(model_name)
self._model_name = model_name
@property
def name(self) -> str:
return f"cross_encoder_{self._model_name.split('/')[-1]}"
def rerank(self, query, documents, top_k=5):
if not documents:
return []
pairs = [(query, doc) for doc in documents]
scores = self.model.predict(pairs, batch_size=32, show_progress_bar=False)
indexed = list(enumerate(zip(documents, scores)))
sorted_results = sorted(indexed, key=lambda x: -x[1][1])
return [
RerankResult(document=doc, score=float(score), original_index=idx)
for idx, (doc, score) in sorted_results[:top_k]
]LLM-based (cápsula 04)
# src/rerankers/llm_based.py
from openai import OpenAI
from pydantic import BaseModel, Field
from concurrent.futures import ThreadPoolExecutor
import os
from .base import Reranker, RerankResult
class RelevanceScore(BaseModel):
score: float = Field(ge=0.0, le=10.0)
class LLMReranker(Reranker):
"""Re-ranker basado en LLM (premium, costoso)."""
SYSTEM_PROMPT = """You are an expert relevance evaluator.
Score how well the document answers the query (0-10).
Be strict. Most docs should score 4-7. Reserve 9-10 for excellent matches.
Return JSON with score field."""
def __init__(self, model: str = "gpt-4o-mini", workers: int = 5):
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.model = model
self.workers = workers
@property
def name(self) -> str:
return f"llm_{self.model}"
def _score_pair(self, args):
idx, query, doc = args
try:
response = self.client.beta.chat.completions.parse(
model=self.model,
messages=[
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "user", "content": f'Query: {query}\n\nDocument:\n{doc[:1500]}'},
],
response_format=RelevanceScore,
temperature=0.1,
)
return idx, response.choices[0].message.parsed.score
except Exception as e:
print(f"LLM rerank failed for doc {idx}: {e}")
return idx, 0.0
def rerank(self, query, documents, top_k=5):
args = [(i, query, doc) for i, doc in enumerate(documents)]
with ThreadPoolExecutor(max_workers=self.workers) as executor:
scored = list(executor.map(self._score_pair, args))
scored.sort(key=lambda x: -x[1])
return [
RerankResult(document=documents[idx], score=float(score), original_index=idx)
for idx, score in scored[:top_k]
]Cohere (cápsula 05)
# src/rerankers/cohere_managed.py
import cohere
import os
from .base import Reranker, RerankResult
class CohereReranker(Reranker):
"""Re-ranker managed via Cohere API."""
def __init__(self, model: str = "rerank-multilingual-v3"):
self.co = cohere.Client(api_key=os.getenv("COHERE_API_KEY"))
self.model = model
@property
def name(self) -> str:
return f"cohere_{self.model}"
def rerank(self, query, documents, top_k=5):
if not documents:
return []
response = self.co.rerank(
model=self.model,
query=query,
documents=documents,
top_n=top_k,
)
return [
RerankResult(
document=documents[r.index],
score=r.relevance_score,
original_index=r.index,
)
for r in response.results
]Factory
# src/rerankers/factory.py
from .cross_encoder import CrossEncoderReranker
from .llm_based import LLMReranker
from .cohere_managed import CohereReranker
def get_reranker(reranker_type: str):
"""Factory para crear reranker según tipo."""
if reranker_type == "cross_encoder":
return CrossEncoderReranker()
elif reranker_type == "cross_encoder_multilingual":
return CrossEncoderReranker(model_name="cross-encoder/mmarco-mMiniLMv2-L12-H384-v1")
elif reranker_type == "llm":
return LLMReranker()
elif reranker_type == "cohere":
return CohereReranker()
elif reranker_type == "none":
return None # baseline sin reranking
else:
raise ValueError(f"Unknown reranker type: {reranker_type}")Paso 3: Eval set con ground truth
El eval set es lo más importante del proyecto. Sin ground truth válido, los benchmarks son ciegos.
Estructura del eval set
# eval_data/golden_set.json (formato)
[
{
"query": "How do I implement OAuth2 authentication in FastAPI?",
"expected_doc_ids": ["doc_42", "doc_87", "doc_103"],
"category": "auth",
"language": "en",
"difficulty": "medium"
},
{
"query": "¿Cómo configuro PostgreSQL con SQLAlchemy?",
"expected_doc_ids": ["doc_215", "doc_301"],
"category": "database",
"language": "es",
"difficulty": "easy"
},
...
]Cómo construirlo
# src/evaluation/eval_set.py
import json
from pathlib import Path
from dataclasses import dataclass, asdict
from typing import List
@dataclass
class EvalQuery:
query: str
expected_doc_ids: List[str]
category: str = "general"
language: str = "en"
difficulty: str = "medium"
def load_eval_set(path: str = "eval_data/golden_set.json") -> List[EvalQuery]:
"""Carga eval set desde JSON."""
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):
"""Guarda eval set a JSON."""
with open(path, "w") as f:
json.dump([asdict(q) for q in queries], f, indent=2)
def build_eval_set_interactive(corpus_collection, n_queries: int = 50):
"""
Construir eval set interactivamente.
Para cada query candidata, el operador valida qué docs son relevantes.
"""
eval_set = []
candidate_queries = [
# Usar queries reales del log de producción
"How do I implement OAuth2 in FastAPI?",
# ...
][:n_queries]
for query in candidate_queries:
# Recuperar top-20 con cosine
results = corpus_collection.query(query_texts=[query], n_results=20)
print(f"\n{'='*80}")
print(f"Query: {query}")
print(f"{'='*80}")
print("Marca con [r] los relevantes:")
relevant_ids = []
for i, (doc, doc_id) in enumerate(zip(results['documents'][0], results['ids'][0])):
print(f"\n[{i}] {doc_id}")
print(f" {doc[:200]}...")
mark = input(" Relevante? (r/n/q para salir): ").strip().lower()
if mark == "r":
relevant_ids.append(doc_id)
elif mark == "q":
break
if relevant_ids:
eval_set.append(EvalQuery(
query=query,
expected_doc_ids=relevant_ids,
category="manual",
language="en",
))
save_eval_set(eval_set, "eval_data/golden_set.json")
return eval_setRecomendación: construir 50-100 queries con ground truth manual. Vale las 2-4 horas de tiempo.
Paso 4: Métricas y A/B test
# src/evaluation/metrics.py
from typing import List
from src.rerankers.base import Reranker, RerankResult
from src.evaluation.eval_set import EvalQuery
import time
import statistics
def precision_at_k(retrieved_ids: List[str], expected_ids: List[str], k: int) -> float:
"""% de top-K que son relevantes."""
top_k = retrieved_ids[:k]
relevant_in_top_k = sum(1 for doc_id in top_k if doc_id in expected_ids)
return relevant_in_top_k / k
def recall_at_k(retrieved_ids: List[str], expected_ids: List[str], k: int) -> float:
"""% de relevantes que aparecen en top-K."""
if not expected_ids:
return 0.0
top_k = retrieved_ids[:k]
relevant_in_top_k = sum(1 for doc_id in top_k if doc_id in expected_ids)
return relevant_in_top_k / len(expected_ids)
def mean_reciprocal_rank(retrieved_ids: List[str], expected_ids: List[str]) -> float:
"""1 / posición del primer relevante. 0 si ninguno está."""
for i, doc_id in enumerate(retrieved_ids, start=1):
if doc_id in expected_ids:
return 1.0 / i
return 0.0
def evaluate_reranker(
reranker: Reranker,
eval_set: List[EvalQuery],
collection,
initial_n: int = 20,
final_top_k: int = 5,
) -> dict:
"""Evalúa un reranker sobre el eval set completo."""
precisions = []
recalls = []
mrrs = []
latencies = []
for item in eval_set:
# Retrieval inicial
results = collection.query(query_texts=[item.query], n_results=initial_n)
candidates = results['documents'][0]
candidate_ids = results['ids'][0]
# Re-rank
start = time.perf_counter()
if reranker is None:
# Baseline sin re-rank
reranked_ids = candidate_ids[:final_top_k]
else:
reranked = reranker.rerank(item.query, candidates, top_k=final_top_k)
reranked_ids = [candidate_ids[r.original_index] for r in reranked]
elapsed = (time.perf_counter() - start) * 1000
# Métricas
precisions.append(precision_at_k(reranked_ids, item.expected_doc_ids, k=final_top_k))
recalls.append(recall_at_k(reranked_ids, item.expected_doc_ids, k=final_top_k))
mrrs.append(mean_reciprocal_rank(reranked_ids, item.expected_doc_ids))
latencies.append(elapsed)
latencies.sort()
return {
"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)],
"n_queries": len(eval_set),
}# src/evaluation/ab_test.py
from src.rerankers.factory import get_reranker
from src.evaluation.metrics import evaluate_reranker
from src.evaluation.eval_set import load_eval_set
def ab_test_all_rerankers(collection):
"""Compara todos los rerankers sobre el mismo eval set."""
eval_set = load_eval_set()
print(f"Eval set: {len(eval_set)} queries")
rerankers_to_test = [
("none (baseline)", "none"),
("cross_encoder", "cross_encoder"),
("llm", "llm"),
("cohere", "cohere"),
]
results = {}
for label, reranker_type in rerankers_to_test:
print(f"\nEvaluating {label}...")
reranker = get_reranker(reranker_type)
result = evaluate_reranker(reranker, eval_set, collection)
results[label] = result
print(f" Precision@5: {result['precision_at_k']:.2%}")
print(f" Recall@5: {result['recall_at_k']:.2%}")
print(f" MRR: {result['mrr']:.3f}")
print(f" Latency p95: {result['latency_p95_ms']:.0f}ms")
return results
def print_comparison_report(results: dict):
"""Imprime tabla comparativa."""
print("\n" + "="*80)
print("AB TEST REPORT")
print("="*80)
headers = ["Reranker", "P@5", "R@5", "MRR", "p50 ms", "p95 ms"]
print(f"\n{headers[0]:<25} {headers[1]:>8} {headers[2]:>8} {headers[3]:>8} {headers[4]:>10} {headers[5]:>10}")
print("-"*80)
for label, metrics in results.items():
print(f"{label:<25} {metrics['precision_at_k']:>7.1%} {metrics['recall_at_k']:>7.1%} {metrics['mrr']:>8.3f} {metrics['latency_p50_ms']:>9.0f} {metrics['latency_p95_ms']:>9.0f}")
# Identificar ganador en cada métrica
print("\nMejor en cada métrica:")
metrics_to_check = ["precision_at_k", "recall_at_k", "mrr"]
for metric in metrics_to_check:
best = max(results.items(), key=lambda x: x[1][metric])
print(f" {metric}: {best[0]} ({best[1][metric]:.3f})")
fastest = min(results.items(), key=lambda x: x[1]["latency_p95_ms"])
print(f" fastest p95: {fastest[0]} ({fastest[1]['latency_p95_ms']:.0f}ms)")Paso 5: Ejecutar y reportar
# benchmarks/run_benchmark.py
from src.evaluation.ab_test import ab_test_all_rerankers, print_comparison_report
from src.retrieval.chromadb_setup import get_collection
import json
from datetime import datetime
def main():
collection = get_collection()
results = ab_test_all_rerankers(collection)
print_comparison_report(results)
# Guardar reporte para auditoría
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
report_path = f"benchmarks/reports/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 none (baseline)...
Precision@5: 71.50%
Recall@5: 63.20%
MRR: 0.587
Latency p95: 195ms
Evaluating cross_encoder...
Precision@5: 89.20%
Recall@5: 78.40%
MRR: 0.812
Latency p95: 348ms
Evaluating llm...
Precision@5: 92.80%
Recall@5: 81.50%
MRR: 0.849
Latency p95: 1480ms
Evaluating cohere...
Precision@5: 90.40%
Recall@5: 79.10%
MRR: 0.823
Latency p95: 312ms
================================================================================
AB TEST REPORT
================================================================================
Reranker P@5 R@5 MRR p50 ms p95 ms
--------------------------------------------------------------------------------
none (baseline) 71.5% 63.2% 0.587 165 195
cross_encoder 89.2% 78.4% 0.812 285 348
llm 92.8% 81.5% 0.849 1180 1480
cohere 90.4% 79.1% 0.823 265 312
Mejor en cada métrica:
precision_at_k: llm (0.928)
recall_at_k: llm (0.815)
mrr: llm (0.849)
fastest p95: none (baseline) (195ms)
Paso 6: Análisis y decisión final
Con los datos del benchmark, aplica el decision framework de la cápsula 07:
# Decisión: Cross-encoder
## Resultados clave del A/B test
- **Cross-encoder mejora precision +17.7 puntos** sobre baseline (71.5% → 89.2%).
- **LLM rerank mejora +21.3 puntos** pero con +1100ms extra y +$200/mes en costos.
- **Cohere mejora +18.9 puntos** con costo similar a LLM.
## Por qué cross-encoder gana en este caso
1. **Costo:** $0 vs $200/mes (LLM) o $30/mes (Cohere).
2. **Latencia:** 348ms p95 vs 1480ms (LLM) — clave para UX <500ms.
3. **Diferencia de calidad vs LLM es solo 3.6%** — no justifica el costo extra.
4. **Corpus mayormente en inglés** — cross-encoder MS MARCO funciona bien.
## Cuándo migraría a Cohere o LLM
- **A Cohere:** si agregamos contenido multilingüe significativo (>20% no-inglés).
- **A LLM:** si compliance pide precision >92% y los $200/mes son aceptables.
- **A LLM en cascada:** para 10% de queries marcadas "high stakes".
## Riesgos identificados
- Cross-encoder tiene cold-start de ~5s. Mitigación: cargar modelo al startup.
- Latencia p95 sube de 195ms a 348ms. Dentro de SLA actual (<500ms) pero margen apretado.Paso 7: Integración al pipeline RAG
# src/retrieval/pipeline.py (actualizado con re-ranking)
from src.rerankers.factory import get_reranker
from src.retrieval.chromadb_setup import get_collection
class RAGPipeline:
def __init__(self, reranker_type: str = "cross_encoder"):
self.collection = get_collection()
self.reranker = get_reranker(reranker_type)
def retrieve(self, query: str, top_k: int = 5):
"""Pipeline completo con re-ranking."""
# Etapa 1: retrieval amplio
results = self.collection.query(query_texts=[query], n_results=20)
candidates = results['documents'][0]
candidate_ids = results['ids'][0]
# Etapa 2: re-rank (si está habilitado)
if self.reranker is None:
return candidates[:top_k], candidate_ids[:top_k]
reranked = self.reranker.rerank(query, candidates, top_k=top_k)
reranked_ids = [candidate_ids[r.original_index] for r in reranked]
reranked_docs = [r.document for r in reranked]
return reranked_docs, reranked_idsChecklist de entrega del proyecto
Antes de considerar el proyecto cerrado, verificar:
- Implementación de las 3 técnicas detrás de la interfaz
Reranker. - Eval set de 50+ queries con ground truth validado manualmente.
- Script
run_benchmark.pyque ejecuta el A/B test. - Reporte comparativo con tabla de métricas y decisión justificada.
- Pipeline RAG actualizado con la técnica ganadora integrada.
- README con setup, cómo correr el benchmark, y decisión final.
- Variables de entorno documentadas (
OPENAI_API_KEY,COHERE_API_KEY). - Tests de cada reranker (al menos sanity checks).
Extensiones opcionales
Cuando termines el proyecto base, considera:
Extensión 1: cascada cross-encoder + LLM
Para queries "high stakes":
class CascadeReranker(Reranker):
def __init__(self):
self.cross_encoder = CrossEncoderReranker()
self.llm = LLMReranker()
def rerank(self, query, documents, top_k=5):
# Etapa A: cross-encoder filtra a top-15
cross_top_15 = self.cross_encoder.rerank(query, documents, top_k=15)
# Etapa B: LLM refina a top-K
cross_docs = [r.document for r in cross_top_15]
return self.llm.rerank(query, cross_docs, top_k=top_k)Extensión 2: caché de resultados
Para queries repetidas, cachear resultados de re-ranking (cubierto en cápsula 06).
Extensión 3: routing por tipo de query
Detectar tipo de query y rutear a reranker apropiado (queries simples → cross-encoder, queries críticas → LLM).
Extensión 4: monitoring en producción
Loggear cada rerank con trace_id, scores, latencia. Dashboard que muestra distribución de scores.
Trampas y errores comunes en el proyecto
Trampa 1: eval set chico o sesgado
Si construyes el eval set con 10 queries, los benchmarks son ruido estadístico. Mínimo 50, ideal 100+.
Trampa 2: medir solo precision
Los rerankers pueden mejorar precision a costa de recall (o viceversa). Reportar ambas Y latencia.
Trampa 3: no medir variabilidad
Una corrida del benchmark puede dar resultados engañosos por azar. Ejecutar 3 veces y reportar mediana de medianas.
Trampa 4: ground truth pobre
Si el ground truth fue anotado por una sola persona en 10 minutos, los benchmarks reflejan ese sesgo. Múltiples anotadores cuando es posible.
Trampa 5: comparar con baseline injusto
Si el baseline es "sin re-rank pero con queries optimizadas" vs "con re-rank pero sin optimization", la comparación es invalida.
Trampa 6: deployar el ganador sin A/B en producción
Aún si el benchmark dice X gana, ejecutar A/B test en producción 1-2 semanas antes de 100% rollout.
Resumen del proyecto
Construiste:
- ✅ Sistema modular de re-ranking con interfaz unificada
- ✅ Tres implementaciones probadas en paralelo
- ✅ Eval set robusto con ground truth
- ✅ Benchmark reproducible
- ✅ Reporte que justifica decisión con datos
- ✅ Pipeline RAG actualizado y listo para deploy
Mejora típica esperada con re-ranking integrado:
Métrica Sin rerank Con rerank Mejora
Precision@5 71% 89% +18 pts
Recall@5 63% 78% +15 pts
MRR 0.587 0.812 +0.225
Esto cierra el módulo de re-ranking y te prepara para M05 (Hybrid Search), donde combinas la mejora de re-ranking con BM25 para queries con keywords exactos.
Próximos pasos
Inmediato:
- Correr el benchmark completo sobre tu eval set.
- Decidir técnica final basándote en resultados.
- Integrar al pipeline RAG y deployar a staging.
- A/B test en producción 1-2 semanas.
Módulo 5 (Hybrid Search):
El re-ranking refina los candidatos pero asume que el retrieval inicial los recuperó. Si tu corpus tiene contenido técnico con identificadores exactos (nombres de funciones, error codes), el retrieval inicial con cosine puede perder docs relevantes. M05 cubre cómo agregar BM25 al retrieval inicial para no perder esos casos.
Recursos
- Sentence Transformers — Cross-Encoders
- Cohere Rerank Documentation
- OpenAI Structured Outputs
- BEIR Benchmark
- LangChain RAG Evaluation
- Anthropic — Contextual Retrieval
Tiempo estimado: 2-3 horas + 1 hora de análisis Siguiente módulo: Módulo 5 — Hybrid Search