Módulo 7: Production con Pinecone — la migración de "demo funcional" a "servicio 24/7"

Proyecto: Production RAG System con Pinecone

Siguiente →

Descripción del proyecto

Este es el proyecto culminante del módulo 7 y el cierre operacional de tu trayecto en RAG avanzado: tomar el sistema que construiste a lo largo de los módulos 2-6 (chunking inteligente, query optimization, hybrid search, re-ranking, metadata filtering) y migrarlo de un entorno de desarrollo local a una arquitectura de producción sobre Pinecone serverless con FastAPI, caching seguro multi-tenant y observabilidad.

La meta no es "usar Pinecone" como checklist técnico. Es demostrar que sabes tomar decisiones de infraestructura con datos: cuándo migrar, cuándo no, cómo medir el impacto, cómo aislar tenants sin filtrar datos entre ellos, y cómo entregar un sistema que un equipo de ingeniería puede operar sin que tú estés despierto a las 3am.

Al terminar tendrás un repositorio público con código modular, tests automatizados, benchmark reproducible y un documento de decisión técnica que puedes mostrar a un comité de arquitectura o a un entrevistador senior. Es el portfolio piece que diferencia a un ingeniero que "leyó sobre RAG" de uno que "operó RAG en producción".

Objetivo del proyecto

Implementar un sistema RAG production-grade sobre Pinecone con seis dimensiones operacionales:

  1. Migración validada desde ChromaDB sin pérdida de datos ni degradación de calidad
  2. Aislamiento multi-tenant garantizado por namespaces + TenantContext + tests automatizados
  3. Pipeline avanzado preservado: query expansion + hybrid search + re-ranking + metadata filtering
  4. API HTTP con FastAPI: endpoints /ask, /health, /metrics listos para Kubernetes
  5. Caching seguro que respeta aislamiento entre tenants
  6. Benchmark + análisis de costo documentados con recomendación final defendible

Especificaciones técnicas

Stack obligatorio

ComponenteTecnologíaPor qué
Vector DBPinecone serverlessBackend gestionado, auto-scale
EmbeddingsOpenAI text-embedding-3-smallEstándar industria, 1536 dims
GeneraciónOpenAI gpt-4o-miniCosto/calidad equilibrados
APIFastAPI + uvicornAsync nativo, OpenAPI auto
CacheRedis o in-memoryReducir read units redundantes
ValidaciónPydantic v2Schema enforcement
Observabilidadstructlog + PrometheusLogs JSON + métricas
Testspytest + pytest-asyncioEstándar Python
BM25 (opcional)rank-bm25Si mantienes hybrid local

Setup del proyecto

mkdir production-rag && cd production-rag
python -m venv venv && source venv/bin/activate
pip install pinecone openai fastapi uvicorn[standard] pydantic redis structlog \
            python-dotenv pytest pytest-asyncio httpx

Estructura de directorios sugerida:

production-rag/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI app
│   ├── config.py            # Pydantic Settings
│   ├── tenant.py            # TenantContext, secure_query
│   ├── retriever.py         # ProductionRetriever
│   ├── pipeline.py          # query → expand → retrieve → rerank → generate
│   ├── cache.py             # Cache layer with tenant isolation
│   ├── metadata.py          # DocMetadata, FilterSpec
│   └── observability.py     # logging, metrics
├── scripts/
│   ├── migrate_from_chroma.py
│   ├── benchmark.py
│   └── seed_test_data.py
├── tests/
│   ├── test_isolation.py
│   ├── test_filters.py
│   └── test_pipeline.py
├── BENCHMARK.md             # Reporte final
├── DECISION.md              # Documento de decisión técnica
├── docker-compose.yml       # Redis + app
└── pyproject.toml

Funcionalidades obligatorias

1) Creación idempotente del índice (scripts/setup_index.py)

from pinecone import Pinecone, ServerlessSpec
import time

def ensure_production_index(pc: Pinecone, name: str, dimension: int = 1536) -> None:
    existing = [i["name"] for i in pc.list_indexes()]
    if name in existing:
        info = pc.describe_index(name)
        assert info.dimension == dimension, f"Dimension mismatch: {info.dimension} vs {dimension}"
        return
    pc.create_index(
        name=name,
        dimension=dimension,
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1"),
    )
    while not pc.describe_index(name).status["ready"]:
        time.sleep(1)

2) Migración desde ChromaDB (scripts/migrate_from_chroma.py)

  • Lectura paginada de ChromaDB
  • Validación con DocMetadata Pydantic antes de upsert
  • Upsert por lotes de 100 vectores
  • Retry exponencial en errores transitorios (429, 5xx)
  • Checkpoint en disco para reanudar
  • Validación final: assert chroma_count == pinecone_namespace_count

3) Aislamiento multi-tenant (app/tenant.py)

from dataclasses import dataclass
import re

VALID_TENANT_ID = re.compile(r"^[a-z0-9_-]{1,40}$")

@dataclass(frozen=True)
class TenantContext:
    tenant_id: str
    role: str

    def __post_init__(self):
        if not VALID_TENANT_ID.match(self.tenant_id):
            raise ValueError(f"Invalid tenant_id: {self.tenant_id}")

def tenant_namespace(tenant_id: str) -> str:
    if not VALID_TENANT_ID.match(tenant_id):
        raise ValueError(f"Invalid tenant_id: {tenant_id}")
    return f"workspace-{tenant_id}"

def secure_query(index, vector, tenant: TenantContext, filter_dict: dict | None = None, top_k: int = 10):
    return index.query(
        vector=vector,
        top_k=top_k,
        namespace=tenant_namespace(tenant.tenant_id),
        filter=filter_dict,
        include_metadata=True,
    )

4) Pipeline RAG completo (app/pipeline.py)

Encadena los componentes que ya construiste:

async def rag_pipeline(query: str, tenant: TenantContext, filter_spec: FilterSpec) -> dict:
    expanded = await expand_query(query)         # M03 query optimization
    candidates = await hybrid_retrieve(expanded, tenant, filter_spec, top_k=50)  # M05 hybrid
    reranked = await rerank(query, candidates, top_k=10)  # M04 reranking
    answer = await generate_answer(query, reranked)
    return {"answer": answer, "sources": reranked}

5) Cache layer con aislamiento (app/cache.py)

import hashlib
import json
from redis.asyncio import Redis

class TenantAwareCache:
    def __init__(self, redis: Redis, ttl_seconds: int = 300):
        self.redis = redis
        self.ttl = ttl_seconds

    def _key(self, tenant_id: str, query: str, filter_dict: dict | None) -> str:
        payload = json.dumps({"q": query, "f": filter_dict or {}}, sort_keys=True)
        digest = hashlib.sha256(payload.encode()).hexdigest()[:16]
        return f"rag:{tenant_id}:{digest}"

    async def get(self, tenant: TenantContext, query: str, filter_dict: dict | None):
        raw = await self.redis.get(self._key(tenant.tenant_id, query, filter_dict))
        return json.loads(raw) if raw else None

    async def set(self, tenant: TenantContext, query: str, filter_dict: dict | None, value: dict):
        await self.redis.setex(
            self._key(tenant.tenant_id, query, filter_dict),
            self.ttl,
            json.dumps(value),
        )

Punto crítico: la key incluye tenant_id siempre. Sin esto, una query idéntica de dos tenants distintos compartiría caché → leak de datos. Esto se prueba en tests/test_isolation.py.

6) API FastAPI (app/main.py)

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel

app = FastAPI(title="Production RAG")

class AskRequest(BaseModel):
    query: str
    types: list[str] | None = None
    tags: list[str] | None = None

class AskResponse(BaseModel):
    answer: str
    sources: list[dict]
    cached: bool

@app.post("/ask", response_model=AskResponse)
async def ask(req: AskRequest, tenant: TenantContext = Depends(get_tenant_from_jwt)):
    spec = FilterSpec(doc_types=req.types, tags_any=req.tags)
    cached = await cache.get(tenant, req.query, spec.build())
    if cached:
        return AskResponse(**cached, cached=True)
    result = await rag_pipeline(req.query, tenant, spec)
    await cache.set(tenant, req.query, spec.build(), result)
    return AskResponse(**result, cached=False)

@app.get("/health")
def health():
    return {"status": "ok"}

@app.get("/metrics")
def metrics():
    return prometheus_client.generate_latest()

7) Benchmark y costos

Entrega BENCHMARK.md siguiendo la metodología de la cápsula 07. Mínimo:

  • Tabla p50/p95/p99 ChromaDB vs Pinecone con 1000+ queries
  • Throughput a concurrency=[1, 10, 50]
  • Recall@10 contra ground truth de 100 queries
  • Estimación de costo a 12 meses con crecimiento proyectado

Validaciones y manejo de errores

Lista de invariantes que tu sistema debe garantizar y que tus tests deben cubrir:

  • Toda query require tenant_id válido (regex ^[a-z0-9_-]{1,40}$)
  • Toda query a Pinecone pasa por secure_query (no llamadas directas a index.query)
  • Dimensión de embedding validada antes de upsert (1536 para OpenAI small)
  • Retry exponencial con jitter en upsert (3 intentos: 1s, 2s, 4s + jitter)
  • Mismatch de conteos post-migración bloquea el deploy
  • Cache key incluye tenant_id y hash determinístico de query+filter
  • Todos los logs incluyen tenant_id, query_id, latency_ms como JSON estructurado
  • /metrics expone counters por tenant: queries totales, errores, latencia p95

Criterios de éxito

  • ✅ Migración completa sin pérdida de datos (validación de conteos + recall@10 estable)
  • ✅ p95 mejora ≥30% frente a baseline ChromaDB local bajo concurrencia ≥10
  • ✅ Tests de aislamiento pasan: no hay caso donde tenant A vea datos de tenant B
  • ✅ FastAPI corre en Docker con health check verde
  • ✅ Cache reduce latencia en ≥50% para queries repetidas sin filtrar entre tenants
  • ✅ BENCHMARK.md y DECISION.md están versionados en git con datos reales del benchmark

Rúbrica de evaluación (100 puntos)

Funcionalidad (50 pts)

  • (15 pts) Migración robusta y validada (script + checkpoint + assertion de conteos)
  • (10 pts) Pipeline RAG completo: expand + hybrid + rerank + filter
  • (10 pts) Aislamiento multi-tenant con tests dedicados
  • (10 pts) Cache con tenant_id en key + tests de no-cross-leak
  • (5 pts) API FastAPI con /ask, /health, /metrics

Operación y calidad (30 pts)

  • (10 pts) Manejo de errores: retry exponencial, fallbacks de filtros, timeout handling
  • (10 pts) Código modular: separación clara entre pipeline, retriever, cache, API
  • (10 pts) Logging estructurado JSON con tenant_id, query_id, latency, status

Benchmark y documentación (20 pts)

  • (10 pts) BENCHMARK.md reproducible con percentiles correctos y recall@k
  • (10 pts) DECISION.md con análisis de costo a 12 meses y recomendación defendible

Extra credit (+10)

  • (+5 pts) Plan de rollback documentado (cómo volver a ChromaDB si Pinecone falla 24h)
  • (+5 pts) Tests de regresión de retrieval (golden set que detecta degradación)

Errores comunes y cómo evitarlos

  1. Crear índice con dimensión equivocada → idempotencia debe verificar dimension == 1536 y fallar si no.
  2. Migrar sin validar conteos → tu CI debe bloquear deploy si chroma_count != pinecone_count.
  3. Llamar index.query directamente → enforces con un linter custom o pre-commit hook que detecte el patrón.
  4. Cache compartido entre tenants → test obligatorio: dos tenants con la misma query reciben respuestas distintas si tienen datos distintos.
  5. Comparar benchmarks con condiciones distintas → corre desde la misma máquina, mismo dataset, misma top_k.
  6. Logs con PII → nunca logueas el contenido completo de queries; solo hash + metadata operacional.
  7. No tener plan de rollback → si Pinecone tiene un outage, ¿cómo sirves tráfico? Debe estar documentado.

Documentos a entregar

BENCHMARK.md

  • Setup del benchmark (hardware, dataset, configuración)
  • Tabla de latencias percentiladas
  • Curva throughput vs concurrencia
  • Recall@10 comparado
  • Estimación de costos a 12 meses

DECISION.md

  • Contexto del producto (escala actual, esperada)
  • Resumen de hallazgos del benchmark
  • Trade-offs evaluados (latencia, costo, operación, lock-in)
  • Recomendación final con razones cuantificadas
  • Plan de rollback

README.md

  • Setup local en 5 minutos
  • Cómo correr migración
  • Cómo correr benchmark
  • Cómo correr tests
  • Cómo deploy en producción

Recursos para el proyecto

  1. Pinecone Docs - Referencia oficial completa.
  2. Upsert Data - Ingesta por lotes optimizada.
  3. Query API - Parámetros de consulta.
  4. Filter by Metadata - Operadores soportados.
  5. Implement Multitenancy - Patrones oficiales.
  6. FastAPI Production Deployment - Buenas prácticas.
  7. Prometheus Python Client - Métricas para /metrics.
  8. structlog - Logging estructurado JSON.

Conexión con el módulo 8

Con este sistema en producción, las queries empiezan a generar tráfico real y los stakeholders te van a preguntar: "¿pero responde bien?". Esa pregunta no se contesta con p95 ni costo: se contesta con métricas de calidad sistemáticas.

En el módulo 8 (RAG Evaluation) construirás el sistema de evaluación continua que falta: golden datasets versionados, RAGAS para métricas automatizadas (faithfulness, answer relevancy, context precision), quality gates en CI/CD que bloquean deploys que degradan calidad, y dashboards que detectan regresión antes de que un usuario abra un ticket.

Sin evaluación continua, tu sistema en producción es una caja negra que mejora cuando crees que mejora. Con evaluación continua, mejoras con evidencia.

Creado: Marzo 13, 2026
Versión: 2.0

← AnteriorBenchmarking y Costos: ChromaDB vs PineconeSiguiente →Introducción al Módulo: RAG Evaluation