Cápsula 04: Metadata Filtering Implementation — del concepto al código en ChromaDB

Descripción de la cápsula

En el Módulo 3 (cápsula 02) aprendiste qué es metadata filtering y por qué es la feature más crítica de un sistema RAG en producción: reduce el espacio de búsqueda 10x, mejora accuracy y, en sistemas multi-tenant, es la línea de defensa contra data leakage entre clientes.

Esta cápsula cierra el loop con el código real. Vas a aprender la sintaxis de where clauses de ChromaDB (que se basa en MongoDB query operators), los seis operadores que necesitas conocer, cómo combinar filters con AND/OR/NOT, y cómo benchmarkear el impacto en tu propia data — porque "filtering acelera 10x" es la regla general, pero el speedup real depende de qué porcentaje de tu dataset queda después del filter, y eso solo lo descubres midiendo.

Cuando termines, vas a saber escribir filters complejos sin consultar la documentación cada vez, anticipar los errores de sintaxis más comunes (especialmente con tipos numéricos vs strings), y diseñar el schema de metadata desde el inicio para que los filters que vas a necesitar sean fáciles de expresar.

Al finalizar esta cápsula serás capaz de:

• ✅ Escribir where clauses con los seis operadores principales: equality, $ne, $gt/$gte/$lt/$lte, $in/$nin, $and/$or
• ✅ Combinar múltiples filters con lógica booleana correcta
• ✅ Diferenciar cuándo el filter va en where (metadata) vs where_document (text)
• ✅ Benchmarkear el impacto real de filters sobre tu dataset (no asumir "10x speedup")
• ✅ Diseñar metadata schema considerando los filters que vas a necesitar
• ✅ Anticipar los tres errores de sintaxis más caros: tipos mezclados, claves inexistentes, AND implícito mal entendido

Tiempo estimado: 30-40 minutos

Insertar metadata correctamente

Antes de filtrar, hay que insertar metadata estructurada. ChromaDB acepta dict por documento con valores de tipo str, int, float, bool. No acepta listas, dicts anidados ni None — eso es la primera limitación a conocer.

import chromadb
from chromadb.utils import embedding_functions
import os

openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key=os.getenv("OPENAI_API_KEY"),
    model_name="text-embedding-3-small"
)

client = chromadb.PersistentClient(path="./chroma_filtering")
collection = client.get_or_create_collection(
    name="docs_with_metadata",
    embedding_function=openai_ef,
    metadata={"hnsw:space": "cosine"}
)

collection.add(
    documents=[
        "Password reset instructions for Pro plan users",
        "Q1 2026 marketing campaign report",
        "API documentation for v2.0 endpoints",
        "Product roadmap 2026 — confidential",
        "Customer support FAQ — billing section",
        "Engineering blog post about HNSW configuration",
    ],
    metadatas=[
        {"category": "support", "language": "en", "priority": 1, "plan": "pro"},
        {"category": "marketing", "language": "en", "priority": 3, "year": 2026},
        {"category": "docs", "language": "en", "priority": 2, "version": "2.0"},
        {"category": "product", "language": "en", "priority": 1, "confidential": True},
        {"category": "support", "language": "es", "priority": 2, "topic": "billing"},
        {"category": "engineering", "language": "en", "priority": 3, "topic": "hnsw"},
    ],
    ids=["1", "2", "3", "4", "5", "6"],
)

print(f"Total docs: {collection.count()}")  # 6

Convenciones de schema importantes:

1. Mantener tipos consistentes por clave. Si una clave es int en un doc y str en otro, los filters por rango fallan en silencio.
2. Naming en snake_case minúsculo. priority, no Priority o priorityLevel.
3. Booleanos para flags. confidential: True en vez de confidential: "yes".
4. Strings cortos para valores categóricos. category: "support" (no oraciones largas).

Los seis operadores que tenés que conocer

1. Equality (default)

# Solo docs en category="support"
results = collection.query(
    query_texts=["how to reset password"],
    where={"category": "support"},
    n_results=5,
)

print(f"Results: {len(results['documents'][0])}")
# 2 docs: "Password reset..." y "Customer support FAQ..."

Sin operador explícito, ChromaDB asume equality. Es la sintaxis que vas a usar en 70% de los casos.

2. $ne (not equal)

# Todo menos category="marketing"
results = collection.query(
    query_texts=["product information"],
    where={"category": {"$ne": "marketing"}},
    n_results=5,
)

Útil para excluir categorías específicas (ej: "no me muestres docs de marketing en queries de soporte").

3. $gt, $gte, $lt, $lte (rangos numéricos)

# Docs con priority <= 2 (alta prioridad)
results = collection.query(
    query_texts=["documentation"],
    where={"priority": {"$lte": 2}},
    n_results=5,
)
# Match: docs con priority=1 (Pro support, Product roadmap) y priority=2 (API docs, Spanish support)

# Docs con priority > 1 (no críticos)
results = collection.query(
    query_texts=["..."],
    where={"priority": {"$gt": 1}},
    n_results=5,
)

Importante: los operadores de rango solo funcionan con tipos numéricos (int, float). Si la clave es string, ChromaDB lanza error o devuelve resultados incorrectos sin avisar (depende de la versión).

4. $in, $nin (membership)

# Docs en categories soporte O documentación
results = collection.query(
    query_texts=["help with the API"],
    where={"category": {"$in": ["support", "docs"]}},
    n_results=5,
)
# Match: support docs + docs category

# Docs que NO sean support ni marketing
results = collection.query(
    query_texts=["..."],
    where={"category": {"$nin": ["support", "marketing"]}},
    n_results=5,
)

$in es preferible a múltiples $or con equality. Más legible y más rápido.

5. $and, $or (lógica booleana)

AND implícito cuando pasás múltiples claves:

# AND implícito: category="support" Y language="en"
results = collection.query(
    query_texts=["password help"],
    where={
        "category": "support",
        "language": "en",
    },
    n_results=5,
)

AND explícito (necesario para combinaciones más complejas):

results = collection.query(
    query_texts=["..."],
    where={
        "$and": [
            {"category": "support"},
            {"priority": {"$lte": 2}},
            {"language": "en"},
        ]
    },
    n_results=5,
)

OR explícito:

# (category="support") O (priority=1)
results = collection.query(
    query_texts=["urgent issue"],
    where={
        "$or": [
            {"category": "support"},
            {"priority": 1},
        ]
    },
    n_results=5,
)

Combinaciones complejas:

# (category in [support, docs]) AND (priority <= 2 OR confidential = true)
results = collection.query(
    query_texts=["..."],
    where={
        "$and": [
            {"category": {"$in": ["support", "docs"]}},
            {
                "$or": [
                    {"priority": {"$lte": 2}},
                    {"confidential": True},
                ]
            }
        ]
    },
    n_results=5,
)

6. where_document (filter por contenido del texto)

Aparte de filtrar por metadata, podés filtrar por substring del documento mismo:

results = collection.query(
    query_texts=["..."],
    where={"category": "engineering"},
    where_document={"$contains": "HNSW"},
    n_results=5,
)
# Solo docs de engineering que mencionan "HNSW" en el texto

where_document solo soporta $contains y $not_contains. Es útil pero no es búsqueda full-text — es substring matching simple. Para hybrid search (BM25 + semantic), necesitas otro approach (cubierto en M03/03 hybrid search y en guía #8 Advanced RAG).

Benchmark del impacto real

La regla general dice "metadata filtering acelera 10x". Eso es verdad cuando el filter reduce el espacio de búsqueda al 10%. Si tu filter solo reduce al 50%, el speedup es ~2x. Si tu filter deja menos candidatos que n_results, puede ser más lento (vimos esa trampa en M04/06).

Vamos a medir sobre 10K docs sintéticos.

# benchmark_filtering.py
import chromadb
from chromadb.utils import embedding_functions
import os
import time
import statistics

openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key=os.getenv("OPENAI_API_KEY"),
    model_name="text-embedding-3-small"
)
client = chromadb.PersistentClient(path="./chroma_bench_filter")
collection = client.get_or_create_collection(
    name="bench_filtering",
    embedding_function=openai_ef
)

# Si la collection está vacía, generar 10K docs sintéticos
if collection.count() == 0:
    docs = [f"Document {i} discusses topic {i % 100} in category {['support','marketing','docs','product'][i % 4]}." for i in range(10000)]
    metas = [
        {
            "category": ["support", "marketing", "docs", "product"][i % 4],
            "priority": (i % 3) + 1,
            "topic_id": i % 100,
        }
        for i in range(10000)
    ]
    ids = [f"doc_{i:05d}" for i in range(10000)]
    # Insertar en batches (cápsula 05)
    for batch_start in range(0, 10000, 200):
        end = min(batch_start + 200, 10000)
        collection.add(
            documents=docs[batch_start:end],
            metadatas=metas[batch_start:end],
            ids=ids[batch_start:end],
        )

# Pre-computar embedding de la query (no contar latencia OpenAI)
import openai
client_oai = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
query_emb = client_oai.embeddings.create(
    input="general topic discussion",
    model="text-embedding-3-small"
).data[0].embedding


def benchmark_filter(label, where_clause, n_runs=50):
    """Mide latencia con un filter específico."""
    # Warm-up
    for _ in range(5):
        collection.query(query_embeddings=[query_emb], n_results=10, where=where_clause)

    latencies = []
    for _ in range(n_runs):
        start = time.perf_counter()
        collection.query(query_embeddings=[query_emb], n_results=10, where=where_clause)
        latencies.append((time.perf_counter() - start) * 1000)

    latencies.sort()
    p50 = latencies[len(latencies) // 2]
    p95 = latencies[int(len(latencies) * 0.95)]
    return p50, p95


# Filters con distintos niveles de selectividad
benchmarks = [
    ("Sin filter (10K candidates)", None),
    ("category=support (~25% = 2500 candidates)", {"category": "support"}),
    ("category=support AND priority<=2 (~17% = 1700)", {"category": "support", "priority": {"$lte": 2}}),
    ("category=support AND priority=1 (~8% = 800)", {"category": "support", "priority": 1}),
    ("topic_id=42 (~1% = 100)", {"topic_id": 42}),  # Filter ultra-específico
]

print(f"{'Filter':<55} {'p50':>8} {'p95':>8}  Speedup vs no-filter")
baseline_p95 = None
for label, where in benchmarks:
    p50, p95 = benchmark_filter(label, where)
    if baseline_p95 is None:
        baseline_p95 = p95
        speedup = "1.0x (baseline)"
    else:
        speedup = f"{baseline_p95 / p95:.1f}x"
    print(f"{label:<55} {p50:>6.1f}ms {p95:>6.1f}ms  {speedup}")

Output típico:

Filter                                                   p50      p95  Speedup vs no-filter
Sin filter (10K candidates)                             8.2ms  12.4ms  1.0x (baseline)
category=support (~25% = 2500 candidates)               4.1ms   6.3ms  2.0x
category=support AND priority<=2 (~17% = 1700)          3.2ms   5.1ms  2.4x
category=support AND priority=1 (~8% = 800)             2.4ms   4.0ms  3.1x
topic_id=42 (~1% = 100)                                 5.8ms   9.2ms  1.3x  ← ojo

Lecciones:

1. El "10x" no es universal. En este dataset el speedup máximo observado es ~3x porque HNSW ya es muy rápido en 10K docs. El 10x se ve más en datasets de 100K+ donde reducir el search space realmente importa.

2. El filter ultra-específico es contra-intuitivo más lento. Reducir el espacio de 10K a 100 candidatos baja el speedup a 1.3x — porque HNSW tiene que escanear más nodos buscando candidatos que cumplan el filter. Cuando el filter deja menos candidatos que n_results × 10, considerá get() con filter directo en vez de query().

3. El benchmark sobre tu dataset real es lo único que importa. Los números arriba son ilustrativos; los tuyos van a ser distintos según distribución de metadata, tamaño de dataset y ef_search.

Diseñar el metadata schema desde el inicio

El error más caro con metadata filtering no es de sintaxis — es no haber agregado los campos correctos al ingestar. Si descubrís en producción que querés filtrar por language, pero los chunks no tienen esa metadata, tenés que re-ingestar todo.

Antes de empezar el ingestion masivo, listá los filters que vas a necesitar:

# Ejemplo: schema de metadata para un RAG corporativo
ESPERAMOS_FILTRAR_POR = [
    "category",       # ej: "support", "engineering", "legal"
    "language",       # ej: "en", "es", "pt"
    "department",     # ej: "sales", "engineering", "hr"
    "access_level",   # ej: "public", "internal", "confidential"
    "doc_date",       # ej: 1714521600 (timestamp para rango)
    "doc_id",         # para retrieval directo de chunks de un doc específico
    "chunk_index",    # para ordenar chunks de un mismo doc
]

# Esto se traduce a metadata por chunk:
def build_metadata(doc, chunk_index, total_chunks):
    return {
        "doc_id": doc.id,
        "chunk_index": chunk_index,
        "total_chunks": total_chunks,
        "category": doc.category,
        "language": doc.language,
        "department": doc.department,
        "access_level": doc.access_level,
        "doc_date": int(doc.created_at.timestamp()),  # int para filters de rango
        "source": doc.source_path,
    }

Heurísticas:

• Sobreincluir es barato, omitir es caro. Agregar 2-3 campos extra cuesta milisegundos en ingestion. Falta de 1 campo crítico cuesta horas de re-ingestion.
• Timestamps como int (Unix epoch), no str. Permite filters por rango de fechas.
• Para data multi-tenant: tenant_id siempre. Es la línea de defensa contra leaks (cápsula 04 del Módulo 3 cubrió por qué).
• Identificadores estables. doc_id debe sobrevivir re-ingests. No usar UUIDs aleatorios.

Trampas y errores comunes

Trampa 1: tipos inconsistentes en una clave

El error:

# Doc 1
{"priority": 1}  # int

# Doc 2
{"priority": "1"}  # string

Síntoma: filters por rango ({"priority": {"$lte": 2}}) devuelven solo algunos docs sin error explícito. Otros docs "desaparecen" del resultado aunque su priority numéricamente cumple la condición.

Por qué pasa: ChromaDB compara el operador con el valor real almacenado. 1 <= 2 es True, pero "1" <= 2 no se evalúa correctamente en SQLite.

Cómo prevenir: validar tipos en el ingestion pipeline:

from typing import Any

EXPECTED_TYPES = {
    "category": str,
    "language": str,
    "priority": int,
    "doc_date": int,
    "confidential": bool,
}

def validate_metadata(meta: dict[str, Any]) -> dict[str, Any]:
    cleaned = {}
    for key, value in meta.items():
        if key in EXPECTED_TYPES:
            expected = EXPECTED_TYPES[key]
            if not isinstance(value, expected):
                raise TypeError(f"Metadata key '{key}' expected {expected.__name__}, got {type(value).__name__}: {value}")
        cleaned[key] = value
    return cleaned

# Antes de collection.add():
metadatas = [validate_metadata(m) for m in raw_metadatas]

Trampa 2: filtrar por una clave que no existe

El error:

# Algunos docs tienen "language", otros no
collection.query(
    query_texts=["..."],
    where={"language": "en"},
    n_results=10,
)

Síntoma: los docs sin la clave language se excluyen del resultado, aunque deberían incluirse según la lógica del producto.

Cómo prevenir: asegurar que todas las claves filtradas existen en todos los docs. Si un campo es opcional, usá un valor por default ("unknown", null sentinel) — no la ausencia de la clave.

Trampa 3: $and confundido con AND implícito

El error:

# Esto NO es lo que pensás
where = {
    "$and": [
        {"category": "support", "priority": 1},  # ← AND implícito dentro del $and
        {"language": "en"}
    ]
}

Síntoma: el filter funciona pero cubre menos casos de los esperados.

Cómo escribir: dentro de $and, cada elemento es un filter independiente. Lo de arriba es equivalente a:

# (category=support AND priority=1) AND (language=en)
where = {
    "$and": [
        {"category": "support"},
        {"priority": 1},
        {"language": "en"},
    ]
}

Más explícito y menos propenso a malentendidos.

Trampa 4: confundir where con where_document

El error:

# Querés filtrar docs cuyo TEXTO contenga "HNSW"
collection.query(
    query_texts=["..."],
    where={"text": {"$contains": "HNSW"}},  # ❌
)

Síntoma: error o resultados vacíos. El operador $contains no aplica a metadata.

Cómo corregir:

collection.query(
    query_texts=["..."],
    where_document={"$contains": "HNSW"},  # ✅
)

where es para metadata. where_document es para el texto del documento. Son parámetros distintos.

Trampa 5: filter ultra-específico que vuelve query lenta

El error:

# Filter que deja solo 5 candidatos
collection.query(
    query_embeddings=[emb],
    n_results=10,
    where={"unique_session_id": "abc123def456"},
)

Síntoma: la query es más lenta que sin filter.

Cómo corregir: si el filter deja menos candidatos que n_results × 5, usar get() directo:

# Más rápido: get directo cuando el filter es muy restrictivo
results = collection.get(
    where={"unique_session_id": "abc123def456"},
    include=['documents', 'metadatas']
)
# No es semantic search, pero el filter ya selecciona los docs relevantes

Trampa 6: metadata schema sin pensar en filters futuros

El error: insertás 1 millón de docs con metadata {"source": path, "chunk_index": i} sin agregar category, language u otros campos. En semana 4, marketing pide "filtremos por idioma para personalizar respuestas".

Síntoma: "tenemos que re-ingestar 1M de docs porque no tenemos language".

Cómo prevenir: antes del ingestion masivo, hacer un workshop con stakeholders ("¿qué filters necesitarán los próximos 12 meses?") y agregar TODOS los campos al schema. El costo extra es trivial. El costo de re-ingestar es significativo (tiempo, dinero de re-embedding, downtime).

Ejercicio aplicado

Escenario: estás construyendo un RAG para una empresa SaaS multi-tenant donde cada cliente tiene sus propios documentos. Requisitos:

• Cada chunk pertenece a un tenant específico (ID único)
• Algunos chunks son privados (solo accesibles a ese tenant)
• Otros chunks son públicos (visibles a todos los tenants — ej: docs de soporte general)
• Los queries deben respetar permisos: tenant A nunca ve chunks privados de tenant B
• Los chunks tienen idioma (en, es, pt)
• Hay clasificación de tipo (product_doc, support_faq, legal, marketing)
• Frecuentemente se filtra por fecha (mostrar solo docs del último año)

Tarea:

1. Diseñá el schema de metadata para los chunks
2. Escribí el filter que aplicaría a una query del tenant acme_corp que busca docs en español del último año
3. Identificá un riesgo de seguridad si el filter se construye mal

def build_chunk_metadata(chunk, doc, tenant_id):
    return {
        # Identidad
        "tenant_id": tenant_id,             # str — siempre presente
        "doc_id": doc.id,                   # str — estable
        "chunk_index": chunk.index,         # int
        "total_chunks": len(doc.chunks),    # int

        # Permisos
        "visibility": doc.visibility,       # str: "private" | "public"

        # Clasificación
        "doc_type": doc.type,               # str: product_doc, support_faq, legal, marketing
        "language": doc.language,           # str: en, es, pt
        "category": doc.category,           # str: granularidad fina opcional

        # Temporal (timestamps Unix epoch para filters por rango)
        "created_at": int(doc.created_at.timestamp()),
        "updated_at": int(doc.updated_at.timestamp()),

        # Trazabilidad
        "source": doc.source_path,          # str — para citas
    }

import time

now = int(time.time())
one_year_ago = now - (365 * 24 * 60 * 60)

# El filter de seguridad: tenant_id="acme_corp" O (visibility="public" Y otro tenant)
filter_query = {
    "$and": [
        {
            "$or": [
                {"tenant_id": "acme_corp"},
                {"visibility": "public"},
            ]
        },
        {"language": "es"},
        {"created_at": {"$gte": one_year_ago}},
    ]
}

results = collection.query(
    query_texts=["¿cómo configuro la facturación?"],
    where=filter_query,
    n_results=10,
)

# ❌ FILTER ROTO: leak entre tenants
filter_query_BROKEN = {
    "$or": [                              # ← OR a nivel raíz
        {"tenant_id": "acme_corp"},
        {"visibility": "public"},
        {"language": "es"},                # ← este OR aplica también a otros tenants
    ]
}
# Esto retorna:
# - docs de acme_corp (correcto)
# - docs públicos de cualquier tenant (correcto)
# - docs en español de CUALQUIER tenant (LEAK!)

def build_secure_filter(tenant_id: str, additional_filters: dict) -> dict:
    """
    Construye un filter que SIEMPRE respeta el aislamiento por tenant.
    additional_filters se AND-ea sobre la cláusula de seguridad,
    nunca puede saltarse el tenant.
    """
    security_clause = {
        "$or": [
            {"tenant_id": tenant_id},
            {"visibility": "public"},
        ]
    }
    return {
        "$and": [security_clause, additional_filters],
    }

# Uso seguro
filter_query = build_secure_filter(
    tenant_id="acme_corp",
    additional_filters={
        "$and": [
            {"language": "es"},
            {"created_at": {"$gte": one_year_ago}},
        ]
    },
)

def test_tenant_isolation():
    # Insertar docs privados de tenant_a y tenant_b
    collection.add(
        documents=["secret_a", "secret_b", "public_doc"],
        metadatas=[
            {"tenant_id": "tenant_a", "visibility": "private"},
            {"tenant_id": "tenant_b", "visibility": "private"},
            {"tenant_id": "tenant_a", "visibility": "public"},
        ],
        ids=["1", "2", "3"],
    )

    # Query como tenant_a NO debe encontrar "secret_b"
    filter = build_secure_filter("tenant_a", {})
    results = collection.query(
        query_texts=["secret"],
        where=filter,
        n_results=10,
    )

    assert "2" not in results['ids'][0], "LEAK: tenant_a vio doc privado de tenant_b"

Resumen y siguiente paso

Lo que aprendiste:

• ChromaDB acepta metadata con tipos str/int/float/bool (no listas, dicts ni None).
• Seis operadores cubren 99% de los casos: equality (default), $ne, $gt/$gte/$lt/$lte, $in/$nin, $and, $or.
• AND es implícito cuando pasás múltiples claves; explícito con $and cuando combinás con $or.
• where filtra por metadata. where_document filtra por substring del texto.
• El speedup real depende del % del dataset que queda después del filter — benchmarkeá tu caso, no asumas "10x".
• Filter ultra-específico (deja <10 candidatos) puede ser más lento que sin filter — usar get() directo en esos casos.
• Diseñar el schema de metadata desde el inicio considerando filters futuros — agregar campos después significa re-ingestar.
• En sistemas multi-tenant, encapsular la cláusula de seguridad (tenant_id + visibility) en función reusable y testarlo con tests de aislamiento.

Checkpoint: antes de avanzar, deberías poder:

• Escribir un filter complejo combinando $and, $or, $in y operadores de rango.
• Identificar el bug de seguridad clásico en filters multi-tenant ($or a nivel raíz).
• Diseñar metadata schema para un RAG considerando filters futuros (no solo los actuales).

Siguiente cápsula: 05 — Batch Ingestion Pipeline.

Acabás de dominar cómo filtrar al hacer queries. Pero antes de poder filtrar, hay que insertar la data — y hacerlo eficientemente cuando son miles o millones de docs requiere conocer batch ingestion. La cápsula 05 cubre cómo amortizar el overhead de inserts, manejar rate limits del API de embeddings, y construir un pipeline idempotente que se pueda re-ejecutar sin duplicar.

Recursos

1. ChromaDB — Metadata Filtering — Referencia oficial de operadores
2. ChromaDB — Where Document Filtering — Filter por contenido del texto
3. MongoDB Query Operators — ChromaDB hereda de MongoDB la sintaxis; útil para casos avanzados
4. Multi-tenancy Patterns in Vector DBs — Comparación de estrategias (collection per tenant vs metadata filtering vs namespaces)
5. Schema Design Best Practices — Patrones generales aplicables a metadata schema
6. Indexed Property Filters in HNSW (paper) — Cómo filtros afectan el rendimiento de HNSW

Tiempo estimado: 30-40 minutos Siguiente: 05-batch-ingestion.md