Metadata Filtering en Pinecone

Descripción de la cápsula

Hasta este momento, tu sistema RAG en Pinecone aísla por tenant usando namespaces y opera con índices serverless escalables. Pero los usuarios de producción rara vez quieren "todo el conocimiento del workspace": quieren tutoriales recientes, no anuncios viejos. Quieren políticas con etiqueta legal, no FAQs públicas. Quieren documentos creados después de la última auditoría.

Esa capa de selección la resuelve metadata filtering nativo: el parámetro filter que Pinecone evalúa durante la búsqueda, antes de calcular similitud final. Es el equivalente robusto y performante del where que ya conoces de ChromaDB, pero ejecutado en infraestructura gestionada.

En esta cápsula vas a construir filtros expresivos para producción, validar payloads antes de indexar (porque un campo mal escrito rompe queries de manera silenciosa), combinar namespace + filter como patrón canónico multi-tenant, y diseñar fallbacks para evitar el "cero resultados" que destruye la experiencia de usuario.

Al final tendrás una capa de filtros mantenible que no es un diccionario que crece sin control, sino un constructor disciplinado con validación de schema y comportamiento documentado.

El problema real: filtros frágiles en producción

Consideremos cómo se ve metadata filtering en código mal pensado:

# Anti-patrón: filtros construidos ad-hoc en cada handler
def handle_search(query, filters):
    vector = embed(query)
    pinecone_filter = {}
    if "type" in filters:
        pinecone_filter["type"] = filters["type"]
    if "tags" in filters:
        pinecone_filter["tags"] = filters["tags"]
    if "date" in filters:
        pinecone_filter["created_at"] = filters["date"]
    return index.query(vector=vector, filter=pinecone_filter, top_k=10)

Tres problemas que vas a vivir:

1. Sin operadores explícitos: {"type": "tutorial"} funciona como $eq, pero {"tags": ["api", "auth"]} no funciona como $in automáticamente. Pinecone va a devolver matches inesperados o ninguno.
2. Sin validación: si filters["date"] viene como string ISO en lugar de timestamp Unix, el filtro pasa pero no compara correctamente y obtienes silencio.
3. Sin reuso: cada endpoint construye su propio diccionario y duplicas lógica con sutiles divergencias.

La solución es un constructor centralizado, validado y testeado que pueda evolucionar sin romper handlers.

Operadores de filter en Pinecone

Pinecone soporta un subconjunto de operadores tipo MongoDB sobre metadata indexada:

# Ejemplo combinando varios operadores
filter_expr = {
    "$and": [
        {"type": {"$in": ["tutorial", "reference"]}},
        {"created_at": {"$gte": 1735689600}},
        {"deprecated": {"$ne": True}},
    ]
}

Punto importante: los campos sobre los que filtras deben existir en metadata al momento de upsert. Pinecone no infiere campos, no permite joins entre vectores, y campos faltantes simplemente no matchean filtros que los referencian.

Validación de metadata con Pydantic

Antes de indexar, define un schema explícito. Esto previene el 80% de los bugs de filtering en producción:

from pydantic import BaseModel, Field, field_validator
from typing import Literal

DocType = Literal["tutorial", "reference", "policy", "faq", "announcement"]

class DocMetadata(BaseModel):
    doc_id: str
    workspace_id: str
    type: DocType
    created_at: int
    updated_at: int
    tags: list[str] = Field(default_factory=list)
    author: str | None = None
    deprecated: bool = False

    @field_validator("tags")
    @classmethod
    def normalize_tags(cls, v: list[str]) -> list[str]:
        return sorted({tag.strip().lower() for tag in v if tag.strip()})

    def to_pinecone(self) -> dict:
        return self.model_dump(exclude_none=True)

Por qué importa esta normalización:

• tags se convierten a minúsculas y sin duplicados antes de indexar. Sin esto, una query buscando "API" no encontraría documentos taggeados "api".
• type está restringido a un Literal. Si alguien intenta indexar type="Tutorial", Pydantic falla rápido en lugar de crear silenciosamente un valor que ningún filtro va a encontrar.
• created_at es int (timestamp Unix). Si alguien manda string ISO, Pydantic lo rechaza.

# Uso correcto: validar antes de upsert
def upsert_document(index, vector, raw_metadata: dict, namespace: str):
    metadata = DocMetadata(**raw_metadata)  # falla si schema inválido
    index.upsert(
        vectors=[(metadata.doc_id, vector, metadata.to_pinecone())],
        namespace=namespace,
    )

Constructor de filtros tipado

Con metadata validada al indexar, ahora construye el filtro de query como una API tipada en lugar de un diccionario libre:

from dataclasses import dataclass, field
from typing import Literal

@dataclass
class FilterSpec:
    doc_types: list[DocType] | None = None
    tags_any: list[str] | None = None
    tags_all: list[str] | None = None
    min_created_at: int | None = None
    max_created_at: int | None = None
    exclude_deprecated: bool = True
    author: str | None = None

    def build(self) -> dict:
        clauses: list[dict] = []
        if self.doc_types:
            clauses.append({"type": {"$in": list(self.doc_types)}})
        if self.tags_any:
            normalized = sorted({t.lower() for t in self.tags_any})
            clauses.append({"tags": {"$in": normalized}})
        if self.tags_all:
            for tag in {t.lower() for t in self.tags_all}:
                clauses.append({"tags": {"$in": [tag]}})
        if self.min_created_at is not None:
            clauses.append({"created_at": {"$gte": self.min_created_at}})
        if self.max_created_at is not None:
            clauses.append({"created_at": {"$lte": self.max_created_at}})
        if self.exclude_deprecated:
            clauses.append({"deprecated": {"$ne": True}})
        if self.author:
            clauses.append({"author": {"$eq": self.author}})

        if not clauses:
            return {}
        if len(clauses) == 1:
            return clauses[0]
        return {"$and": clauses}

Ventajas frente al diccionario libre:

• El handler recibe un FilterSpec, no un dict[str, Any]. El IDE puede autocompletar y mypy puede tipar.
• La normalización de tags vive en un solo lugar (.lower()).
• exclude_deprecated=True por defecto previene que documentos marcados como obsoletos contaminen resultados.
• Cuando agregues un operador nuevo, todos los handlers heredan la mejora sin tocar código.

Patrón canónico: namespace + filter

Combina aislamiento (namespace) con selección (filter) en una sola función segura:

def secure_filtered_query(
    index,
    vector: list[float],
    tenant: TenantContext,
    spec: FilterSpec,
    top_k: int = 10,
) -> dict:
    namespace = tenant_namespace(tenant.tenant_id)
    filter_dict = spec.build()
    return index.query(
        vector=vector,
        top_k=top_k,
        namespace=namespace,
        filter=filter_dict if filter_dict else None,
        include_metadata=True,
    )

Tres invariantes garantizadas:

1. Aislamiento absoluto: el namespace siempre proviene del TenantContext validado, nunca de input del cliente.
2. Composición correcta: el filtro vive dentro del namespace; nunca cruzas tenants.
3. Filtro opcional: si spec.build() está vacío, mandas filter=None (Pinecone permite query sin filtro), evitando un objeto {} que en algunos clientes genera warnings.

Comparación: filtrar en app vs filtrar en motor

Hay dos lugares donde puedes aplicar filtros: en el motor (Pinecone evalúa antes de devolver) o en la app (Pinecone devuelve N resultados, tú filtras después). La elección impacta latencia, costo y seguridad.

Regla práctica: filtra en motor siempre que el dato esté en metadata. Filtra en app solo cuando necesitas datos que Pinecone no tiene (permisos calculados dinámicamente, joins con tablas externas).

Conexión con el proyecto Production RAG

Tu Production RAG va a exponer una API de búsqueda como esta:

@app.post("/search")
async def search(req: SearchRequest, user: AuthUser = Depends(get_user)):
    tenant = TenantContext(tenant_id=user.workspace_id, role=user.role)
    spec = FilterSpec(
        doc_types=req.types,
        tags_any=req.tags,
        min_created_at=req.since,
        exclude_deprecated=True,
    )
    vector = await embed_query(req.query)
    matches = secure_filtered_query(index, vector, tenant, spec, top_k=req.top_k)
    return format_response(matches)

Lo crítico es que el cliente nunca construye filtros raw. La API recibe campos de negocio (types, tags, since) y el FilterSpec los traduce a operadores Pinecone. Esto previene inyección de filtros maliciosos y mantiene el contrato estable cuando cambies el backend.

Troubleshooting

Problema 1: "Filter expression invalid"

Causa: estructura JSON con operador no soportado o tipo incorrecto (string donde se espera number).
Solución: valida que min_created_at sea int, no str. Usa FilterSpec.build() en lugar de construir dicts a mano. Si necesitas debuggear, imprime filter_dict antes del query y compáralo contra la referencia oficial de operadores.

Problema 2: "Resultados inconsistentes por tags"

Causa: tags indexados sin normalizar ("API", "api", "Api" se tratan como distintos).
Solución: aplica el field_validator de DocMetadata.normalize_tags en upsert. Para datos legacy, corre un job de re-indexado que renormalice y haga upsert sobre los mismos IDs.

Problema 3: "Cero resultados con filtro complejo"

Causa: condiciones demasiado restrictivas (combinación rara de tags + fecha + tipo).
Solución: implementa fallback graduado. Primero intenta el filtro completo; si no devuelve suficientes matches, relaja la dimensión menos crítica (típicamente tags).

def query_with_graduated_fallback(index, vector, tenant, spec, top_k=10, min_results=5):
    primary = secure_filtered_query(index, vector, tenant, spec, top_k=top_k)
    if len(primary["matches"]) >= min_results:
        return primary
    relaxed = FilterSpec(
        doc_types=spec.doc_types,
        min_created_at=spec.min_created_at,
        exclude_deprecated=spec.exclude_deprecated,
    )
    return secure_filtered_query(index, vector, tenant, relaxed, top_k=top_k)

Problema 4: "Filtros lentos en namespaces grandes"

Causa: no hay índices secundarios para metadata en serverless; cardinalidad alta de un campo puede degradar.
Solución: mantén el conjunto de campos filtrables pequeño (5-10 campos máximo). Para campos free-form (descripciones largas), no los uses como filtro: ponlos en metadata pero no los referencies en filter.

Problema 5: "Campos boolean no funcionan"

Causa: Pinecone serializa boolean correctamente pero algunos SDKs viejos los convierten en string.
Solución: verifica versión del SDK (pinecone>=5.0) y prueba con {"deprecated": {"$eq": False}} explícito en lugar de {"deprecated": False}.

Ejercicios

Ejercicio 1: Schema de metadata para tu dominio

Define un DocMetadata Pydantic para un dominio de soporte técnico que indexa: tickets resueltos, artículos de KB, notas internas. Debe incluir severity (low/medium/high/critical), resolved_at opcional y product_line.

from pydantic import BaseModel, Field, field_validator
from typing import Literal

Severity = Literal["low", "medium", "high", "critical"]
DocType = Literal["ticket", "kb_article", "internal_note"]

class SupportDocMetadata(BaseModel):
    doc_id: str
    workspace_id: str
    type: DocType
    severity: Severity
    product_line: str
    created_at: int
    resolved_at: int | None = None
    tags: list[str] = Field(default_factory=list)

    @field_validator("tags")
    @classmethod
    def normalize_tags(cls, v: list[str]) -> list[str]:
        return sorted({t.strip().lower() for t in v if t.strip()})

    @field_validator("product_line")
    @classmethod
    def normalize_product(cls, v: str) -> str:
        return v.strip().lower().replace(" ", "_")

Ejercicio 2: Filter para "tickets críticos abiertos del último mes"

Construye un FilterSpec que devuelva tickets de severidad critical o high, sin resolved_at, creados en los últimos 30 días.

import time

now = int(time.time())
thirty_days_ago = now - 30 * 24 * 3600

filter_dict = {
    "$and": [
        {"type": {"$eq": "ticket"}},
        {"severity": {"$in": ["critical", "high"]}},
        {"created_at": {"$gte": thirty_days_ago}},
        {"resolved_at": {"$exists": False}},
    ]
}

result = index.query(
    vector=query_vector,
    top_k=20,
    namespace="workspace-acme",
    filter=filter_dict,
    include_metadata=True,
)

Ejercicio 3: Constructor con FilterSpec

Crea una función search_support que reciba parámetros de negocio y use FilterSpec para producir el filtro Pinecone.

from dataclasses import dataclass

@dataclass
class SupportFilterSpec:
    severities: list[Severity] | None = None
    product_line: str | None = None
    only_open: bool = False
    min_created_at: int | None = None

    def build(self) -> dict:
        clauses = []
        if self.severities:
            clauses.append({"severity": {"$in": list(self.severities)}})
        if self.product_line:
            clauses.append({"product_line": {"$eq": self.product_line.lower().replace(" ", "_")}})
        if self.only_open:
            clauses.append({"resolved_at": {"$exists": False}})
        if self.min_created_at is not None:
            clauses.append({"created_at": {"$gte": self.min_created_at}})
        if not clauses:
            return {}
        return clauses[0] if len(clauses) == 1 else {"$and": clauses}

def search_support(index, vector, tenant, spec: SupportFilterSpec, top_k=10):
    return secure_filtered_query(index, vector, tenant, spec, top_k=top_k)

Ejercicio 4: Fallback graduado para soporte

Implementa un fallback que primero busca con filtros completos; si no encuentra ≥3 resultados, relaja product_line; si aún no encuentra, relaja también severities.

def search_with_graduated_relaxation(index, vector, tenant, spec, top_k=10, min_results=3):
    primary = search_support(index, vector, tenant, spec, top_k)
    if len(primary["matches"]) >= min_results:
        return primary, "primary"

    no_product = SupportFilterSpec(
        severities=spec.severities,
        only_open=spec.only_open,
        min_created_at=spec.min_created_at,
    )
    second = search_support(index, vector, tenant, no_product, top_k)
    if len(second["matches"]) >= min_results:
        return second, "relaxed_product"

    no_severity = SupportFilterSpec(
        only_open=spec.only_open,
        min_created_at=spec.min_created_at,
    )
    third = search_support(index, vector, tenant, no_severity, top_k)
    return third, "relaxed_severity"

Ejercicio 5: Auditoría de queries con filtros

Implementa logging estructurado que registre cada query con su filter aplicado, número de resultados y tenant. Útil para debugging y analytics.

import json
import logging
from time import perf_counter

logger = logging.getLogger("pinecone.query.audit")

def audited_query(index, vector, tenant, spec, top_k=10):
    namespace = tenant_namespace(tenant.tenant_id)
    filter_dict = spec.build()
    start = perf_counter()
    result = index.query(
        vector=vector,
        top_k=top_k,
        namespace=namespace,
        filter=filter_dict if filter_dict else None,
        include_metadata=True,
    )
    elapsed_ms = (perf_counter() - start) * 1000

    logger.info(json.dumps({
        "event": "pinecone_query",
        "tenant_id": tenant.tenant_id,
        "namespace": namespace,
        "filter": filter_dict,
        "top_k": top_k,
        "matches": len(result["matches"]),
        "elapsed_ms": round(elapsed_ms, 2),
    }))
    return result

Resumen

• Pinecone soporta operadores tipo MongoDB ($eq, $in, $gte, $and, $exists, etc.) sobre metadata indexada en upsert
• Valida metadata con Pydantic antes de upsert; normaliza tags y restringe type a Literal para prevenir bugs silenciosos
• Construye filtros con un FilterSpec tipado, no con diccionarios libres en cada handler
• El patrón canónico multi-tenant es secure_filtered_query(tenant, spec): namespace por tenant + filter por negocio
• Filtra en motor (pre-filter nativo) salvo cuando necesites datos externos a Pinecone
• Implementa fallback graduado para evitar "cero resultados" en queries restrictivas
• Auditoría con logging estructurado: cada query registra tenant, filter aplicado y matches devueltos

Recursos adicionales

1. Filter by Metadata - Pinecone Docs - Operadores oficiales soportados.
2. Pinecone Query API Reference - Parámetros completos del endpoint query.
3. Pydantic v2 Validators - Field validators para normalización.
4. Metadata Best Practices - Recomendaciones oficiales de schema.
5. RAG in Production - Pinecone Learn - Patrones operacionales.

Creado: Marzo 13, 2026
Versión: 2.0