Módulo 6: Metadata Filtering — el componente que casi nadie implementa primero pero todos terminan necesitando
Cápsula 07: Integración con hybrid search — el patrón "estado del arte" final
Descripción de la cápsula
Cubrimos cada componente por separado: chunking (M02), query optimization (M03), re-ranking (M04), hybrid search (M05), metadata filtering (este módulo). Ahora viene la pregunta operativa: ¿cómo se conectan todos en un solo pipeline?
Esta cápsula te enseña la arquitectura "estado del arte" para RAG production-ready de mayo 2026: query optimization → metadata filter → hybrid retrieval → reranking → generation. Cada componente reduce un problema distinto. Combinados, llevan precision@5 desde ~70% (RAG MVP) hasta 92-95% en producción real.
Al finalizar esta cápsula serás capaz de:
- ✅ Implementar el pipeline completo
filter → hybrid → rerankend-to-end - ✅ Aplicar metadata filtering tanto a vector search como a BM25 (ambos componentes del hybrid)
- ✅ Diseñar fallback escalonado que preserva aislamiento mientras relaja optional filters
- ✅ Benchmarkear las cuatro arquitecturas en orden incremental: semantic-only, +hybrid, +rerank, +filter
- ✅ Anticipar puntos de falla cuando los componentes interactúan
- ✅ Decidir qué componentes son obligatorios vs opcionales según tu contexto
Tiempo estimado: 30 minutos
La arquitectura completa
Query del usuario
│
▼
┌──────────────────────────────────────┐
│ Query Optimization (M03) │
│ ├─ Detectar tipo de query │
│ ├─ Rewriting/expansion si aplica │
│ └─ Output: query(s) procesada(s) │
└──────────────────┬───────────────────┘
│
▼
┌──────────────────────────────────────┐
│ Metadata Filter (M06 - este módulo) │
│ ├─ workspace_id obligatorio │
│ ├─ visibility según rol │
│ ├─ filtros temporales (recencia) │
│ └─ tags relevantes │
└──────────────────┬───────────────────┘
│ filter aplicado
▼
┌──────────────────────────────────────┐
│ Hybrid Retrieval (M05) │
│ │
│ ┌──────────────────┐ ┌────────────┐│
│ │ Semantic search │ │ BM25 ││
│ │ (con filter) │ │ (con filter│
│ │ Top-30 │ │ Top-30 │
│ └────────┬─────────┘ └─────┬──────┘│
│ │ │ │
│ └──────┬───────────┘ │
│ ▼ │
│ Reciprocal Rank Fusion │
│ │ │
│ ▼ │
│ Top-30 fused │
└──────────────────┬───────────────────┘
│
▼
┌──────────────────────────────────────┐
│ Re-ranking (M04) │
│ Cross-encoder o LLM rerank │
│ → Top-5 final │
└──────────────────┬───────────────────┘
│
▼
┌──────────────────────────────────────┐
│ LLM Generation │
│ Prompt: query + top-5 docs │
│ → Respuesta │
└──────────────────────────────────────┘
Cada paso reduce un problema distinto:
| Componente | Problema que ataca |
|---|---|
| Query optimization | Queries ambiguas o incompletas del usuario |
| Metadata filter | Buscar en menos docs (latencia + aislamiento) |
| Semantic search | Encontrar docs por significado |
| BM25 | Encontrar docs por keywords exactos |
| RRF | Combinar señales de retrieval |
| Re-ranking | Refinar top-K con mejor scoring |
Implementación del pipeline completo
# pipeline.py
from concurrent.futures import ThreadPoolExecutor
from typing import Optional
import chromadb
from rank_bm25 import BM25Okapi
from sentence_transformers import CrossEncoder
class FilteredHybridRagPipeline:
"""Pipeline completo: filter → hybrid → rerank."""
def __init__(
self,
collection,
bm25_index: BM25Okapi,
all_doc_ids: list[str],
all_metadatas: list[dict],
reranker_model: str = "cross-encoder/ms-marco-MiniLM-L-12-v2",
):
self.collection = collection
self.bm25_index = bm25_index
self.all_doc_ids = all_doc_ids
self.all_metadatas = all_metadatas
self.reranker = CrossEncoder(reranker_model)
def search(
self,
query: str,
tenant: TenantContext,
additional_filters: Optional[dict] = None,
n_candidates: int = 30,
n_final: int = 5,
) -> list[dict]:
"""
Pipeline end-to-end: metadata filter → hybrid (sem + BM25) → RRF → rerank.
"""
# Paso 1: Construir filter seguro
where = self._build_filter(tenant, additional_filters)
# Paso 2: Hybrid retrieval en paralelo (con filter aplicado en ambos)
with ThreadPoolExecutor(max_workers=2) as executor:
future_sem = executor.submit(
self._semantic_search, query, where, n_candidates
)
future_bm25 = executor.submit(
self._bm25_search, query, where, n_candidates
)
sem_ids = future_sem.result()
bm25_ids = future_bm25.result()
# Paso 3: Fusión con RRF
fused_ids = self._reciprocal_rank_fusion([sem_ids, bm25_ids])
top_candidates = fused_ids[:n_candidates]
# Paso 4: Recuperar contenido para rerank
candidates_data = self.collection.get(
ids=top_candidates,
include=["documents", "metadatas"],
)
# Paso 5: Re-ranking con cross-encoder
reranked = self._cross_encoder_rerank(
query=query,
candidates=candidates_data,
top_k=n_final,
)
return reranked
def _build_filter(
self, tenant: TenantContext, additional: Optional[dict]
) -> dict:
"""Construye filter con workspace_id obligatorio."""
if not tenant or not tenant.workspace_id:
raise TenantIsolationError("workspace_id obligatorio")
base = {"workspace_id": tenant.workspace_id}
if additional and "workspace_id" in additional:
raise TenantIsolationError("No sobrescribir workspace_id")
if additional:
return {"$and": [base, additional]}
return base
def _semantic_search(self, query: str, where: dict, n: int) -> list[str]:
"""Vector search con filter aplicado."""
results = self.collection.query(
query_texts=[query],
where=where,
n_results=n,
)
return results["ids"][0]
def _bm25_search(self, query: str, where: dict, n: int) -> list[str]:
"""BM25 search con filter manual sobre IDs."""
query_tokens = query.lower().split()
all_scores = self.bm25_index.get_scores(query_tokens)
# IDs que pasan el filter (pre-computar set para fast lookup)
allowed_ids = self._get_filtered_ids(where)
# Tomar top-N solo entre los allowed
scored = [
(self.all_doc_ids[i], all_scores[i])
for i in range(len(all_scores))
if self.all_doc_ids[i] in allowed_ids and all_scores[i] > 0
]
scored.sort(key=lambda x: -x[1])
return [doc_id for doc_id, _ in scored[:n]]
def _get_filtered_ids(self, where: dict) -> set[str]:
"""Aplicar filter manualmente sobre los metadatos para BM25."""
# Usar collection.get con where para obtener IDs allowed
result = self.collection.get(where=where, include=[])
return set(result["ids"])
def _reciprocal_rank_fusion(
self, rankings: list[list[str]], k: int = 60
) -> list[str]:
"""RRF: combinar rankings ignorando scores absolutos."""
from collections import defaultdict
scores = defaultdict(float)
for ranking in rankings:
for rank, doc_id in enumerate(ranking, start=1):
scores[doc_id] += 1.0 / (k + rank)
sorted_ids = sorted(scores.items(), key=lambda x: -x[1])
return [doc_id for doc_id, _ in sorted_ids]
def _cross_encoder_rerank(
self, query: str, candidates: dict, top_k: int
) -> list[dict]:
"""Cross-encoder rerank sobre los candidates fusionados."""
if not candidates["documents"]:
return []
pairs = [(query, doc) for doc in candidates["documents"]]
scores = self.reranker.predict(pairs, batch_size=32, show_progress_bar=False)
ranked_indices = sorted(
range(len(scores)),
key=lambda i: -scores[i],
)[:top_k]
return [
{
"doc_id": candidates["ids"][i],
"document": candidates["documents"][i],
"metadata": candidates["metadatas"][i],
"score": float(scores[i]),
}
for i in ranked_indices
]Uso
# Asumiendo collection y bm25 ya configurados
pipeline = FilteredHybridRagPipeline(
collection=chroma_collection,
bm25_index=bm25,
all_doc_ids=all_ids,
all_metadatas=all_metas,
)
tenant = TenantContext(
workspace_id="acme",
user_id="user_1",
user_role="member",
)
results = pipeline.search(
query="¿cómo configuro OAuth2?",
tenant=tenant,
additional_filters={
"$and": [
{"language": "es"},
{"created_at": {"$gte": six_months_ago_ts}},
]
},
n_candidates=30,
n_final=5,
)
# Pasar al LLM
context = "\n\n".join(r["document"] for r in results)
answer = generate_answer(query, context)Fallback escalonado en pipeline integrado
Si el filter es muy restrictivo y devuelve pocos resultados, puedes escalar fallback a nivel pipeline:
def search_with_pipeline_fallback(
pipeline,
query: str,
tenant: TenantContext,
optional_filters: dict,
n_final: int = 5,
):
"""Pipeline con fallback escalonado de optional filters."""
fallback_steps = [
optional_filters, # full
{k: v for k, v in optional_filters.items() if k != "tags"}, # sin tags
{k: v for k, v in optional_filters.items() if k != "created_at"},# sin time
{}, # solo tenant
]
for step_filters in fallback_steps:
try:
results = pipeline.search(
query=query,
tenant=tenant,
additional_filters=step_filters or None,
n_final=n_final,
)
if len(results) >= 3:
return {"results": results, "filters_used": step_filters}
except Exception as e:
print(f"Step failed: {e}")
continue
return {"results": [], "filters_used": None}workspace_id SIEMPRE preservado — está en pipeline._build_filter().
Benchmarking incremental
Demostrar el impacto de cada componente con benchmarks:
def benchmark_incremental(eval_set):
"""Mide cada componente incremental sobre el mismo eval set."""
results = {}
# Architecture A: solo semantic baseline
print("Architecture A (baseline)...")
results["A_semantic_only"] = evaluate(
eval_set,
retriever_fn=lambda q, t: collection.query(query_texts=[q], n_results=5),
)
# Architecture B: + hybrid
print("Architecture B (+ hybrid)...")
results["B_hybrid"] = evaluate(
eval_set,
retriever_fn=lambda q, t: hybrid_search(q, n_results=5),
)
# Architecture C: + rerank
print("Architecture C (+ rerank)...")
results["C_hybrid_rerank"] = evaluate(
eval_set,
retriever_fn=lambda q, t: hybrid_search_with_rerank(q, n_final=5),
)
# Architecture D: + metadata filter (la final)
print("Architecture D (+ metadata filter — full pipeline)...")
pipeline = FilteredHybridRagPipeline(...)
results["D_full_pipeline"] = evaluate(
eval_set,
retriever_fn=lambda q, t: pipeline.search(q, t, n_final=5),
)
return results
# Output esperado típico:
# A_semantic_only: Recall@5 = 65%, Precision@5 = 72%, Latency p95 = 220ms
# B_hybrid: Recall@5 = 78%, Precision@5 = 82%, Latency p95 = 340ms
# C_hybrid_rerank: Recall@5 = 85%, Precision@5 = 91%, Latency p95 = 480ms
# D_full_pipeline: Recall@5 = 87%, Precision@5 = 92%, Latency p95 = 290ms ← ¡filter mejora latencia!Lectura clave: metadata filter NO solo agrega seguridad, también mejora latencia (busca en menos vectores) y a veces precision (menos ruido).
Trampas comunes en la integración
Trampa 1: BM25 no respeta el filter
El error: vector search aplica where, pero BM25 busca sobre todo el corpus.
Síntoma: RRF fusiona resultados de tenant A (vector) con resultados de varios tenants (BM25). Data leak parcial.
Cómo prevenir: el approach del pipeline arriba — _get_filtered_ids aplica filter manualmente sobre los IDs antes de scoring BM25.
Trampa 2: filter cambia el ranking de RRF
Si vector retrieval con filter devuelve 10 docs y BM25 con filter devuelve 25, el ranking RRF está sesgado hacia BM25 (más candidatos).
Cómo prevenir: asegurar que ambos retrievers devuelven n_candidates consistente. Si un retriever no encuentra suficientes docs en el subset filtrado, anotarlo como warning pero seguir.
Trampa 3: rerank scoring sobre poco contexto
Después del filter agresivo, puedes tener solo 5-8 candidates. Reranker sobre tan pocos no aporta mucho.
Cómo prevenir: si len(candidates) < min_for_rerank, skip rerank — los pocos que hay ya pasaron filter + RRF, ranking adicional no agrega valor.
Trampa 4: latencia subiendo a pesar del filter
Filter aplicado correctamente debería bajar latencia. Si sube:
- ¿
tenant_idestá indexado en ChromaDB? Sin índice, filter es lineal. - ¿BM25 está aplicando el filter eficientemente o re-scanning todo?
- ¿Rerank batch_size correcto?
Cómo prevenir: profile cada paso del pipeline. Identificar el componente lento.
Ejercicio aplicado
Escenario: eres AI Engineer en una empresa SaaS de DevOps. Tu pipeline actual:
- 50 tenants, ~1M docs total
- Sistema actual: solo semantic + cross-encoder rerank
- Métricas: Precision@5 = 80%, Recall@5 = 68%, Latency p95 = 320ms
- Tickets reportando: "vi un doc de otra empresa", "el bot no encuentra error codes específicos"
Tu trabajo:
- Diseña el pipeline final que ataca los dos problemas (data leak + recall en queries con identificadores).
- Justifica qué componentes agregas y en qué orden.
- Estima impacto esperado.
Solución
1. Pipeline propuesto: filter → hybrid → rerank (full architecture D)
class FullPipeline:
def search(self, query, tenant: TenantContext, n_final=5):
# Paso 1: filter por tenant_id + visibility (CRÍTICO para data leak)
where = build_secure_filter(tenant)
# Paso 2: hybrid retrieval en paralelo
with ThreadPoolExecutor(max_workers=2) as ex:
sem_future = ex.submit(semantic_search, query, where, n=30)
bm25_future = ex.submit(bm25_search_filtered, query, where, n=30)
sem_ids = sem_future.result()
bm25_ids = bm25_future.result()
# Paso 3: RRF
fused = reciprocal_rank_fusion([sem_ids, bm25_ids])[:30]
# Paso 4: recuperar contenido + rerank
candidates = collection.get(ids=fused)
return cross_encoder_rerank(query, candidates, top_k=n_final)2. Justificación de cada componente
| Componente | Por qué | Resuelve qué problema |
|---|---|---|
| Metadata filter | tenant_id obligatorio, visibility según rol | "vi un doc de otra empresa" |
| BM25 (en hybrid) | Identificadores exactos: error codes, comandos | "no encuentra error codes específicos" |
| Semantic search | (ya estaba, mantener) — queries conceptuales | Cubierto por baseline |
| RRF fusion | Combina señales de semantic + BM25 | Mejora recall sin perder precision |
| Cross-encoder rerank | (ya estaba, mantener) — refinar top-K | Calidad final del retrieval |
Orden de impacto:
- Metadata filter (CRÍTICO): elimina data leak. Sin esto, el sistema viola compliance.
- BM25: ataca el problema de "error codes específicos". Mejora recall +15-20%.
- Mantener rerank: ya funciona, no romper.
3. Estimación de impacto
Métrica Antes Después Cambio
─────────────────────────────────────────────────────────
Precision@5 80% 92% (+12 pts) Mejora
Recall@5 68% 88% (+20 pts) Mejora dramática
Latency p95 320ms 220ms (-100ms) Mejora (filter reduce search space)
Data leak risk ALTO CERO Eliminado
Costo $/mes (igual) +$30 (BM25 infra) Aceptable
Plan de migración (4 semanas):
Semana 1: Implementar metadata filter + secure_query
- Tests de aislamiento automatizados
- A/B test interno
- Compliance review
Semana 2: Integrar BM25 (rank_bm25 si <1M docs, ES si más)
- Construir índice
- Implementar bm25_search_filtered con _get_filtered_ids
- Tests de performance
Semana 3: Pipeline completo + RRF
- Integrar todo
- Benchmark vs baseline
- Validar mejora en eval set propio
Semana 4: Rollout gradual
- Feature flag al 10%
- Monitorear data_leak_count (debería ser 0)
- Escalar a 50% → 100%
Métricas a monitorear post-deploy:
- Crítico: zero data leak (count debe ser 0).
- Recall@5 sobre eval set diario.
- Precision@5.
- Latency p50/p95/p99.
- BM25 contribution rate (% de docs en top-K que vinieron del BM25 ranking).
- Filter coverage (% queries con filter aplicado correctamente).
Resumen y siguiente paso
Lo que aprendiste:
- Pipeline completo: query opt → filter → hybrid (semantic + BM25 con RRF) → rerank → LLM.
- Cada componente reduce un problema distinto. Combinados, llevan precision@5 de 70% a 92-95%.
- BM25 también debe respetar el filter — no solo vector search.
- Fallback escalonado preserva tenant_id mientras relaja optional filters.
- Metadata filter mejora seguridad Y latencia (busca en menos docs).
- Benchmark incremental: medir cada componente para justificar su valor.
Checkpoint: antes de avanzar, deberías poder:
- Implementar pipeline completo
filter → hybrid → rerank. - Asegurar que BM25 respeta el filter (no solo vector search).
- Diseñar fallback escalonado a nivel pipeline.
Siguiente cápsula: 08 — Proyecto integrador metadata-filtered RAG.
Cierre del módulo: vas a construir el pipeline completo de este módulo — filter + hybrid + rerank + tests de aislamiento + benchmark — como proyecto del portfolio. Es la culminación de M01-M06.
Recursos
- Anthropic — Contextual Retrieval — Técnica complementaria
- Pinecone — Hybrid Search — Tutorial visual
- LangChain — EnsembleRetriever — Implementación de referencia
- LlamaIndex — Multi-Step Query Engine — Patrones avanzados
- BEIR Benchmark — Comparaciones empíricas
- Cross-Encoders Guide — Reranking
Tiempo estimado: 30 minutos Siguiente: 08-project-metadata-filtered-rag.md