Cápsula 03: Diseño de metadata schema — la decisión que tu yo del futuro va a agradecer (o maldecir)

Descripción de la cápsula

Pre-filtering solo funciona si tu metadata está bien diseñada. Y "bien diseñada" significa pensada antes del primer ingest masivo, no después de descubrir que necesitas un campo nuevo cuando ya tienes 10M documentos indexados. Esta cápsula te da el playbook para diseñar el schema correcto desde el día 1, evitando las dos trampas opuestas: schema muy minimalista (que te obliga a re-indexar al primer change) y schema inflado (donde 80% de los campos nunca se usan pero ralentizan todo).

El secreto está en el principio "design for filters you'll need in the next 12 months, not for what you have today". Esa proyección requiere un workshop con stakeholders, no es decisión solo del ingeniero. Esta cápsula te enseña cómo conducir ese workshop, qué campos son universales (siempre van), y cómo validar metadata antes de indexar para que la deuda técnica no se acumule.

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

• ✅ Identificar los 5 campos universales que casi todo RAG production necesita
• ✅ Conducir un workshop de schema design con stakeholders en menos de 1 hora
• ✅ Aplicar las cuatro convenciones de modelado (snake_case, enums controlados, tipos correctos, naming consistente)
• ✅ Implementar validación de metadata pre-indexación con Pydantic
• ✅ Diseñar migración de metadata legacy sin downtime
• ✅ Anticipar el error más caro: agregar un campo de filtro 6 meses después y tener que re-embeber 10M docs

Tiempo estimado: 30-35 minutos

Los 5 campos universales

Casi todo sistema RAG production-ready necesita estos cinco campos. Empieza con ellos, agrega específicos según tu dominio:

Campo 1: tenant_id (o workspace_id)

Tipo: string Obligatorio: sí, en sistemas multi-tenant Razón: aislamiento de seguridad. Cubierto en cápsulas 02 y 05. Ejemplo: "tenant_id": "acme_corp"

Campo 2: doc_id y chunk_index

Tipo: string + int Obligatorio: sí Razón: trazabilidad. Cuando un retrieval devuelve un chunk, quieres saber de qué documento original vino y qué posición ocupa. Ejemplo: "doc_id": "manual_v3_chapter_4", "chunk_index": 12

Campo 3: source

Tipo: string Obligatorio: sí Razón: debugging. Cuando algo está mal, quieres saber qué documento original lo causó. Ejemplo: "source": "/docs/legal/contracts/v3/section_4.md"

Campo 4: created_at (timestamp)

Tipo: int (Unix epoch seconds) Obligatorio: sí Razón: filtros temporales. "solo docs del último año", "ignorar docs anteriores a la versión 3.0", retention policies, freshness en ranking. Ejemplo: "created_at": 1714521600

Importante: Unix epoch como int, no string ISO. Permite filtros por rango ($gte, $lt) que strings no permiten.

Campo 5: type o category

Tipo: string (enum cerrado) Obligatorio: recomendado Razón: segmentación por intención. "buscar solo en FAQ", "solo tutoriales", "ignorar marketing". Mejora precision dramáticamente. Ejemplo: "type": "tutorial" (de un enum {tutorial, reference, faq, changelog, marketing})

Campos específicos del dominio

Estos los agregas según tu caso. Algunos comunes:

Regla: agrega un campo si vas a filtrar por él. No lo agregues "por si acaso" — un campo nunca usado es overhead permanente.

El workshop de schema design

Antes de indexar 1M documentos, conduce un workshop de 60-90 minutos con stakeholders del producto:

Preguntas a hacer

1. ¿Tenemos múltiples clientes que NO deben ver datos de otros? (Si sí: tenant_id obligatorio)
2. ¿Vamos a tener documentos de distintos tipos? (Si sí: type enum)
3. ¿Importa la fecha del documento para algunas queries? (Si sí: created_at)
4. ¿Qué filtros nos van a pedir en los próximos 6-12 meses? (Acá es donde el producto piensa)
5. ¿Hay categorías/segmentos del corpus que se manejan distinto? (Si sí: agregar campo)
6. ¿Hay regulaciones que requieren campos específicos? (Compliance: GDPR, HIPAA, SOX)

Output del workshop: contrato de schema

# data_contract.py
from pydantic import BaseModel, Field
from typing import Literal
from datetime import datetime


class DocumentMetadata(BaseModel):
    """Contrato de metadata para todos los documentos del sistema."""

    # Universales
    tenant_id: str = Field(..., description="ID del tenant. Aislamiento multi-tenant.")
    doc_id: str = Field(..., description="ID estable del documento padre.")
    chunk_index: int = Field(..., ge=0, description="Posición del chunk dentro del doc.")
    source: str = Field(..., description="Path o URI del documento original.")
    created_at: int = Field(..., description="Unix timestamp en segundos.")

    # Tipo (enum cerrado)
    type: Literal["tutorial", "reference", "faq", "changelog", "marketing"]

    # Específicos del dominio
    language: Literal["en", "es", "pt", "fr"]
    product_line: Literal["api", "dashboard", "mobile", "general"]
    version: str = Field(..., regex=r"^\d+\.\d+(\.\d+)?$")  # ej: "3.1.0"

    # Tags abiertos (lista de strings normalizados)
    tags: list[str] = Field(default_factory=list)

    class Config:
        # Prohibir campos extra para evitar inflar schema accidentalmente
        extra = "forbid"

Beneficios de tener este contrato:

• Cualquier documento que se indexa pasa por validación.
• Si el contrato cambia, hay un único lugar donde actualizarlo.
• Devs nuevos saben exactamente qué metadata se espera.
• Pydantic genera schema JSON automático para documentación.

Las cuatro convenciones de modelado

Convención 1: snake_case consistente

# ✅ Bien
{"tenant_id": "acme", "created_at": 1714521600, "product_line": "api"}

# ❌ Mal: mezcla de casing
{"tenantId": "acme", "createdAt": 1714521600, "ProductLine": "api"}

Por qué: consistencia hace los filtros legibles y previene typos. Algunas vector DBs (incluyendo ChromaDB) son case-sensitive en metadata keys.

Convención 2: enums cerrados para campos categóricos

# ✅ Bien
ALLOWED_TYPES = {"tutorial", "reference", "faq"}
ALLOWED_LANGUAGES = {"en", "es", "pt"}

# ❌ Mal: campo abierto
{"type": "Tutorial"}     # mismo concepto pero "Tutorial" vs "tutorial"
{"type": "TutoriaL"}     # typo
{"type": "tutorial doc"} # variante semántica

Por qué: sin enum, terminas con 47 variantes del mismo concepto y los filtros pierden cobertura. Validación pre-ingest fuerza consistencia.

Convención 3: tipos correctos para filtros de rango

# ✅ Bien
{"created_at": 1714521600}  # int, permite $gte, $lt
{"price": 99.99}             # float
{"is_published": True}       # bool

# ❌ Mal
{"created_at": "2024-04-30"}  # string, no permite filtros de rango bien
{"price": "99.99"}            # string parsing en cada filter

Por qué: filtros como where={"created_at": {"$gte": cutoff}} requieren tipos numéricos comparables. Strings no funcionan correctamente.

Convención 4: naming descriptivo y prefijado

# ✅ Bien (descriptivo, sin ambigüedad)
{"tenant_id": "X", "user_role": "admin", "doc_visibility": "private"}

# ❌ Mal (ambiguo)
{"id": "X", "role": "admin", "vis": "private"}  # ¿qué id? ¿role de qué?

Por qué: los nombres se mantienen por años. Vale la pena un par de caracteres extra de claridad.

Validación pre-indexación

Implementar validación que falle ruidosamente si la metadata no cumple el contrato:

# validation.py
from pydantic import ValidationError


def validate_and_normalize(raw_metadata: dict) -> DocumentMetadata:
    """Valida y normaliza metadata. Falla si no cumple contrato."""
    try:
        # Pydantic valida tipos, requeridos, enums, regex
        validated = DocumentMetadata(**raw_metadata)
    except ValidationError as e:
        # Logging detallado para debugging
        print(f"Metadata validation failed: {e.errors()}")
        raise

    # Normalización adicional
    metadata_dict = validated.model_dump()
    metadata_dict["tags"] = [t.lower().strip() for t in metadata_dict.get("tags", [])]

    return metadata_dict


# Uso en pipeline de ingest
def ingest_document(content: str, raw_metadata: dict):
    try:
        validated_metadata = validate_and_normalize(raw_metadata)
    except ValidationError:
        # Decisión: skip o fail loud
        print(f"Skipping doc por metadata inválida")
        return None

    collection.add(
        documents=[content],
        metadatas=[validated_metadata],
        ids=[f"{validated_metadata['doc_id']}_chunk_{validated_metadata['chunk_index']}"],
    )

Decisión clave: ¿qué hacer si la validación falla?

• Fail loud (raise): mejor en development. Forzas a fixear el problema.
• Skip + log: mejor en producción. No bloquear el pipeline por un doc malformado, pero logear para investigar.
• Quarantine: mover docs inválidos a una collection separada para revisión manual.

Migración de metadata legacy

Si llegas tarde y ya tienes 1M docs sin metadata estructurada, hay opciones:

Opción A: re-indexar con metadata completa (lo correcto pero caro)

def reindex_with_proper_metadata(old_collection, new_collection):
    """Re-indexa todo el corpus con metadata válida."""
    docs = old_collection.get(include=["documents", "metadatas", "embeddings"])

    for doc, raw_meta, emb in zip(docs["documents"], docs["metadatas"], docs["embeddings"]):
        # Inferir o agregar campos faltantes
        new_meta = backfill_metadata(raw_meta, doc)
        validated = validate_and_normalize(new_meta)

        new_collection.add(
            documents=[doc],
            embeddings=[emb],  # reusar embeddings existentes (no re-embebir)
            metadatas=[validated],
            ids=[validated["doc_id"]],
        )


def backfill_metadata(raw: dict, content: str) -> dict:
    """Inferir campos faltantes desde el contenido o defaults."""
    return {
        "tenant_id": raw.get("tenant_id", "unknown"),  # default si no existe
        "doc_id": raw.get("doc_id") or hash(content),   # generar si falta
        "chunk_index": raw.get("chunk_index", 0),
        "source": raw.get("source", raw.get("filename", "legacy")),
        "created_at": raw.get("created_at") or int(datetime.now().timestamp()),
        "type": raw.get("type", "reference"),
        "language": detect_language(content),  # ML detection
        "product_line": raw.get("product_line", "general"),
        "version": raw.get("version", "1.0"),
        "tags": raw.get("tags", []),
    }

Costo: tiempo de re-indexar, pero no de re-embebir (reusas embeddings).

Opción B: migración progresiva (no rompe nada)

def progressive_migration():
    """Marca docs con schema_version. Procesa por lotes."""
    # Procesar 10K docs por día
    batch = old_collection.get(
        where={"schema_version": None},  # solo no-migrados
        limit=10000,
    )

    for item in batch:
        # ... mismo backfill ...
        validated["schema_version"] = "v2"
        update_in_place(item, validated)

Beneficio: sin downtime. El sistema mezcla docs migrados y no-migrados durante la transición.

Costo: filters tienen que tolerar ambos schemas durante semanas.

Trampas y errores comunes

Trampa 1: schema demasiado minimalista al inicio

El error: "agreguemos solo tenant_id por ahora, después agregamos más si hace falta".

Síntoma: 6 meses después necesitas filtrar por tipo, idioma, fecha. Re-indexar 5M docs cuesta ~$50-100 + downtime.

Cómo prevenir: workshop al inicio. Pensar 12 meses adelante. Mejor agregar 3 campos extra desde el día 1 que migrar después.

Trampa 2: schema inflado con campos nunca usados

El error: "agreguemos 30 campos por si acaso".

Síntoma: RAM usada por metadata es 2x mayor que necesaria. Indexing más lento. Confusión de devs nuevos sobre qué campos usar.

Cómo prevenir: balance — solo agregar lo que vas a usar en próximos 12 meses. Workshop con stakeholders define el corte.

Trampa 3: campos sin enum cerrado se fragmentan

El error: {"category": <free text>} permite cualquier valor.

Síntoma: después de 6 meses tienes "tutorial", "Tutorial", "tut", "tutoriaL", "tutorial-doc", "tutorials" en el campo. Filtros pierden cobertura porque solo matchean uno.

Cómo prevenir: Pydantic Literal types o validación contra ALLOWED_VALUES. Reject cualquier valor fuera de la lista.

Trampa 4: timestamps como strings

El error: {"created_at": "2024-04-30"}.

Síntoma: filter where={"created_at": {"$gte": "2024-01-01"}} puede funcionar por casualidad (string comparison) pero falla con formatos inconsistentes.

Cómo prevenir: siempre Unix timestamp int. Convertir desde ISO al ingestar:

"created_at": int(datetime.fromisoformat(date_str).timestamp())

Trampa 5: campos con cardinalidad ultra-alta

El error: {"hash": "<hash de 64 chars>", "uuid": "<uuid>"} como filter.

Síntoma: ChromaDB/Pinecone no construyen índice eficiente sobre campos con 1M+ valores únicos. Filters lentos.

Cómo prevenir: campos para filter deben tener cardinalidad <10K. Para identificadores únicos (hashes, UUIDs), guardarlos en metadata pero no usar como filter.

Trampa 6: olvidar versionar el schema

El error: sin versioning, no sabes qué docs están en schema v1 vs v2.

Síntoma: durante migración progresiva, no puedes diferenciar docs migrados de los que faltan.

Cómo prevenir: agregar "schema_version": "v2" en cada doc. Filter where={"schema_version": {"$ne": "v2"}} muestra cuáles faltan migrar.

Ejercicio aplicado

Escenario: eres AI Engineer en una empresa SaaS de gestión de clínicas médicas. Datos:

• 50 clínicas (tenants)
• ~200K documentos por clínica (historias clínicas, protocolos, papers, manuales)
• Idiomas: español, portugués, inglés
• Compliance: HIPAA + GDPR
• Queries típicas: doctores y enfermeros buscando guidelines y protocolos

Tu trabajo:

1. Diseña el metadata schema completo con Pydantic.
2. Justifica cada campo (universal vs específico del dominio).
3. Define enums para los campos categóricos.

# clinic_metadata.py
from pydantic import BaseModel, Field
from typing import Literal


class ClinicalDocumentMetadata(BaseModel):
    """Schema de metadata para documentos clínicos."""

    # === UNIVERSALES ===

    tenant_id: str = Field(..., description="ID de la clínica. HIPAA isolation.")
    doc_id: str
    chunk_index: int = Field(..., ge=0)
    source: str = Field(..., description="Path al documento original.")
    created_at: int = Field(..., description="Unix timestamp.")

    # === ESPECÍFICOS DEL DOMINIO MÉDICO ===

    # Tipo de documento (para segmentar queries)
    type: Literal[
        "patient_record",
        "protocol",
        "clinical_guideline",
        "research_paper",
        "training_material",
        "policy",
    ]

    # Audiencia (filtros por rol)
    audience: Literal[
        "clinical",      # médicos, enfermeros
        "administrative", # admin, finanzas
        "patient",       # contenido para pacientes
        "all",
    ]

    # Idioma
    language: Literal["es", "pt", "en"]

    # Especialidad médica (enum cerrado)
    specialty: Literal[
        "general",
        "cardiology",
        "pediatrics",
        "oncology",
        "neurology",
        "psychiatry",
        "internal_medicine",
        "emergency",
        "surgery",
        "obstetrics_gynecology",
    ]

    # Sensibilidad PHI (filters por rol)
    contains_phi: bool = Field(..., description="Contiene Protected Health Info?")

    # Versión del documento
    version: str = Field(..., regex=r"^\d+\.\d+$")

    # Estado regulatorio
    regulatory_status: Literal["draft", "approved", "deprecated", "under_review"]

    # Tags abiertos para sub-segmentación
    tags: list[str] = Field(default_factory=list)

    # Versioning del schema (para migraciones)
    schema_version: Literal["v1"] = "v1"

    class Config:
        extra = "forbid"

# Cardiólogo busca protocolos aprobados
where={
    "tenant_id": "clinic_acme",
    "specialty": "cardiology",
    "type": "protocol",
    "regulatory_status": "approved",
    "language": "es",
}

# Personal admin busca solo docs no-clínicos
where={
    "tenant_id": "clinic_acme",
    "audience": "administrative",
    "contains_phi": False,
}

# Investigador busca papers recientes
where={
    "tenant_id": "clinic_acme",
    "type": "research_paper",
    "created_at": {"$gte": one_year_ago_ts},
    "language": {"$in": ["en", "es"]},
}

def ingest_clinical_doc(content: str, raw_metadata: dict):
    try:
        validated = ClinicalDocumentMetadata(**raw_metadata)
    except ValidationError as e:
        # Quarantine para revisión
        save_to_quarantine(content, raw_metadata, e.errors())
        log_metric("ingest_validation_failed", tenant=raw_metadata.get("tenant_id"))
        return None

    # PHI validation extra (regulatory)
    if validated.contains_phi and validated.audience == "patient":
        raise SecurityError("Doc con PHI no puede tener audiencia 'patient'")

    return collection.add(
        documents=[content],
        metadatas=[validated.model_dump()],
        ids=[f"{validated.doc_id}_{validated.chunk_index}"],
    )

Resumen y siguiente paso

Lo que aprendiste:

• Cinco campos universales: tenant_id, doc_id+chunk_index, source, created_at, type.
• Campos específicos según dominio: language, version, audience, etc.
• Workshop con stakeholders al inicio define el contrato — vale 60-90 minutos.
• Cuatro convenciones: snake_case, enums cerrados, tipos numéricos para fechas, naming descriptivo.
• Validación pre-ingest con Pydantic. Decision: fail loud (dev) o skip+log (prod) o quarantine.
• Migración legacy: re-indexar con backfill, opción progresiva con schema_version.
• Trampas: schema muy minimalista (re-index futuro), inflado (overhead), enums abiertos (fragmentación), timestamps como strings.

Checkpoint: antes de avanzar, deberías poder:

• Listar los 5 campos universales y justificar cada uno.
• Conducir un workshop de schema design en menos de 90 min.
• Implementar validación con Pydantic + manejo de errores apropiado.

Siguiente cápsula: 04 — Filtros con where clause de ChromaDB.

Tienes el schema. Ahora vas a aprender la sintaxis completa de filtros: igualdad, rangos, IN, AND/OR. La cápsula 04 es la práctica concreta sobre el schema teórico de esta cápsula.

Recursos

1. Pydantic — Data Validation — Modelos y validación
2. ChromaDB — Metadata Filtering — Operadores soportados
3. JSON Schema — Validación formal
4. Martin Fowler — Data Modeling — Patrones generales
5. Anthropic — Contextual Retrieval — Patrón complementario
6. GDPR Data Mapping Guide — Compliance y metadata

Tiempo estimado: 30-35 minutos Siguiente: 04-filters-with-chromadb-where.md