Cápsula 06: Observabilidad y Deployment

Descripción de la cápsula

Tu sistema RAG funciona en local. Ahora lo empaquetarás para ejecución reproducible y añadirás telemetría base para operar con confianza en cualquier entorno.

En esta cápsula implementarás:

• Dockerfile multi-stage para la API RAG
• docker-compose.yml con rag-api y chromadb-server
• Variables de entorno para configuración sensible
• Health checks en API y contenedores
• Integración con Prometheus para métricas
• Logging estructurado con trace_id para trazabilidad

Al final podrás levantar todo el sistema con un solo comando y monitorear latencia, errores y volumen de consultas.

---

Por qué Deployment Reproducible Importa

El problema del "funciona en mi máquina"

Desarrollador A: Python 3.11, ChromaDB 0.4.22, Ubuntu
Desarrollador B: Python 3.9, ChromaDB 0.3.x, macOS
Staging: Docker, pero sin ChromaDB como servicio
Producción: ???

Sin empaquetado consistente, cada entorno es una lotería. Docker y docker-compose te dan una sola fuente de verdad para ejecutar el sistema.

Por qué observabilidad desde el día uno

En producción necesitas responder:

• ¿El sistema está vivo? → Health checks
• ¿Cuántas preguntas llegan? → Contadores
• ¿Qué tan lento es /ask? → Latencia p95
• ¿Dónde falló esta request? → Logs con trace_id

Sin esto, debugging en producción es adivinar.

---

Arquitectura de Deployment

                    ┌─────────────────────────────────────────┐
                    │           docker-compose                 │
                    │                                           │
   Cliente          │  ┌─────────────────┐  ┌──────────────┐  │
   HTTP ────────────┼─►│   rag-api       │  │ chromadb-   │  │
                    │  │   (FastAPI)     │──►│ server      │  │
                    │  │   :8000         │  │ :8001       │  │
                    │  └────────┬────────┘  └──────────────┘  │
                    │           │                              │
                    │           ▼                              │
                    │  ┌─────────────────┐                    │
                    │  │   Volumen       │  (persistencia)    │
                    │  │   chroma_data   │                    │
                    │  └─────────────────┘                    │
                    └─────────────────────────────────────────┘

---

Dockerfile Multi-Stage

Objetivo: imagen pequeña, reproducible, sin artifacts de build.

# Dockerfile
# ========== Stage 1: Builder ==========
FROM python:3.11-slim as builder

WORKDIR /app

# Instalar dependencias de compilación solo para build
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

# Crear virtualenv y instalar dependencias
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

# ========== Stage 2: Runtime ==========
FROM python:3.11-slim

WORKDIR /app

# Copiar solo lo necesario desde builder
COPY --from=builder /root/.local /root/.local
ENV PATH=/root/.local/bin:$PATH

# Usuario no-root por seguridad
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser

# Código de la aplicación
COPY --chown=appuser:appuser app/ ./app/
COPY --chown=appuser:appuser main.py .

# Variables por defecto (override con env)
ENV HOST=0.0.0.0
ENV PORT=8000

EXPOSE 8000

# Healthcheck
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Explicación de instrucciones clave

Instrucción	Propósito
FROM python:3.11-slim	Imagen base mínima (~50MB vs ~900MB de full)
as builder	Stage intermedio, no se incluye en imagen final
COPY --from=builder	Solo bins y libs, no compiladores
USER appuser	No ejecutar como root dentro del contenedor
HEALTHCHECK	Docker/K8s pueden detectar si el proceso está vivo
EXPOSE 8000	Documenta el puerto (no lo abre; eso lo hace run/compose)

---

docker-compose.yml

# docker-compose.yml
version: "3.8"

services:
  # ========== API RAG ==========
  rag-api:
    build:
      context: .
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    environment:
      - CHROMA_HOST=chromadb
      - CHROMA_PORT=8001
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - LOG_LEVEL=info
    depends_on:
      chromadb:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 10s
    restart: unless-stopped

  # ========== ChromaDB Server ==========
  chromadb:
    image: chromadb/chroma:latest
    ports:
      - "8001:8000"
    volumes:
      - chroma_data:/chroma/chroma
    environment:
      - IS_PERSISTENT=TRUE
      - ANONYMIZED_TELEMETRY=FALSE
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/api/v1/heartbeat"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 5s
    restart: unless-stopped

volumes:
  chroma_data:

Variables de entorno

Crea .env (no commitear, usarlo solo local):

# .env.example (renombrar a .env y completar)
OPENAI_API_KEY=sk-...
LOG_LEVEL=info
CHROMA_HOST=chromadb
CHROMA_PORT=8001

# Levantar todo
docker-compose up -d

# Ver logs
docker-compose logs -f rag-api

# Parar y eliminar volúmenes
docker-compose down -v

---

Variables de Entorno en la Aplicación

# app/config.py
import os
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    chroma_host: str = os.getenv("CHROMA_HOST", "localhost")
    chroma_port: int = int(os.getenv("CHROMA_PORT", "8001"))
    openai_api_key: str = os.getenv("OPENAI_API_KEY", "")
    log_level: str = os.getenv("LOG_LEVEL", "info")
    
    class Config:
        env_file = ".env"

settings = Settings()

Centraliza toda la configuración aquí. No hardcodees URLs ni API keys.

---

Health Check Endpoint

# app/main.py (fragmento)
from fastapi import FastAPI, Request
import httpx

app = FastAPI(title="RAG API", version="1.0.0")

@app.get("/health")
async def health(request: Request):
    """Health check: API + dependencias."""
    status = {"status": "ok", "version": "1.0.0"}
    try:
        chroma_url = f"http://{settings.chroma_host}:{settings.chroma_port}/api/v1/heartbeat"
        async with httpx.AsyncClient() as client:
            r = await client.get(chroma_url, timeout=2.0)
            status["chromadb"] = "ok" if r.status_code == 200 else "degraded"
    except Exception as e:
        status["chromadb"] = "error"
        status["chromadb_error"] = str(e)
    return status

Kubernetes, Render, Railway y similares usan /health para saber si el servicio está listo.

---

Integración con Prometheus

Dependencias

# requirements.txt (añadir)
prometheus-client>=0.19.0

Métricas mínimas

# app/metrics.py
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from fastapi import Response

# Contadores
request_count = Counter(
    "rag_requests_total",
    "Total requests",
    ["endpoint", "method", "status"]
)
ask_latency = Histogram(
    "rag_ask_latency_seconds",
    "Latency of /ask endpoint",
    buckets=[0.5, 1.0, 1.5, 2.0, 3.0, 5.0]
)
search_latency = Histogram(
    "rag_search_latency_seconds",
    "Latency of /search endpoint",
    buckets=[0.05, 0.1, 0.2, 0.5, 1.0]
)

def get_metrics():
    return generate_latest()

@app.get("/metrics")
def metrics():
    return Response(
        content=get_metrics(),
        media_type=CONTENT_TYPE_LATEST
    )

Uso en endpoints

import time
from app.metrics import request_count, ask_latency

@app.post("/ask")
async def ask(payload: dict, request: Request):
    trace_id = request.headers.get("X-Trace-ID", "unknown")
    start = time.perf_counter()
    try:
        result = await do_ask(payload["question"], trace_id)
        request_count.labels(endpoint="/ask", method="POST", status="200").inc()
        ask_latency.observe(time.perf_counter() - start)
        return result
    except Exception as e:
        request_count.labels(endpoint="/ask", method="POST", status="500").inc()
        raise

---

Logging Estructurado

# app/logging_config.py
import logging
import json
from datetime import datetime

class StructuredFormatter(logging.Formatter):
    def format(self, record):
        log_obj = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
        }
        if hasattr(record, "trace_id"):
            log_obj["trace_id"] = record.trace_id
        if record.exc_info:
            log_obj["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_obj, ensure_ascii=False)

def setup_logging():
    handler = logging.StreamHandler()
    handler.setFormatter(StructuredFormatter())
    root = logging.getLogger()
    root.addHandler(handler)
    root.setLevel(logging.INFO)

# Uso en endpoint
logger = logging.getLogger(__name__)

async def do_ask(question: str, trace_id: str):
    logger.info("Processing ask", extra={"trace_id": trace_id, "question_len": len(question)})
    # ...

Salida ejemplo:

{"timestamp": "2026-03-13T12:00:00.000Z", "level": "INFO", "message": "Processing ask", "trace_id": "req-abc123", "question_len": 45}

---

Checklist de Deployment

• Se puede levantar con docker-compose up -d
• Healthcheck de rag-api responde OK
• Healthcheck de ChromaDB responde OK
• Logs estructurados (JSON) habilitados
• /metrics expone latencia y contadores
• Variables sensibles en .env (no en repo)
• Persistencia de ChromaDB en volumen nombrado

---

Flujo Recomendado de Deployment

1. Build imagen reproducible
   docker build -t rag-api:latest .

2. Validar en staging
   docker-compose -f docker-compose.yml up
   curl http://localhost:8000/health
   curl -X POST http://localhost:8000/ask -d '{"question":"test"}'

3. Smoke test de endpoints críticos
   - GET /health
   - GET /search?q=test
   - POST /ask con pregunta conocida

4. Promoción a producción
   - Tag de imagen: rag-api:v1.2.3
   - Desplegar en plataforma (Render, Railway, K8s)
   - Verificar métricas en dashboard

---

Métricas Mínimas Post-Deploy

Métrica	Dónde	Umbral alerta
Disponibilidad /health	Prometheus / uptime	< 99%
p95 /ask	Prometheus histogram	> 2.5s
Error rate por endpoint	Prometheus counter	> 1%
Volumen consultas/min	Prometheus counter	— (solo observación)
ChromaDB heartbeat	Health check	error

---

Ejercicios con Soluciones Detalladas

Ejercicio 1: Añadir endpoint de readiness

Objetivo: /ready que verifica ChromaDB + conexión a OpenAI (sin llamar, solo connect).

Solución:

@app.get("/ready")
async def ready():
    checks = {}
    try:
        async with httpx.AsyncClient() as c:
            r = await c.get(f"http://{settings.chroma_host}:{settings.chroma_port}/api/v1/heartbeat", timeout=2.0)
        checks["chromadb"] = r.status_code == 200
    except Exception:
        checks["chromadb"] = False
    checks["openai_configured"] = bool(settings.openai_api_key)
    all_ok = all(checks.values())
    return JSONResponse(
        status_code=200 if all_ok else 503,
        content={"ready": all_ok, "checks": checks}
    )

Ejercicio 2: Métrica de costo por consulta

Objetivo: Exponer rag_cost_per_query (Gauge o Counter) estimando tokens usados.

Solución:

from prometheus_client import Gauge
cost_per_query = Gauge("rag_estimated_cost_usd", "Estimated cost per query in USD")

# En do_ask, después de llamar a OpenAI:
# cost_per_query.set( (prompt_tokens * 0.001 + completion_tokens * 0.002) / 1000 )  # aprox

Ejercicio 3: Dockerfile para desarrollo (hot reload)

Objetivo: Dockerfile.dev que monta código y usa uvicorn --reload.

Solución:

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]

En docker-compose.yml añadir servicio rag-api-dev con build: Dockerfile.dev y volumes: ["./app:/app/app"].

Ejercicio 4: Logs con trace_id en todas las requests

Objetivo: Middleware que inyecta trace_id en cada request y lo añade a los logs.

Solución:

import uuid
from starlette.middleware.base import BaseHTTPMiddleware

class TraceMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        trace_id = request.headers.get("X-Trace-ID") or str(uuid.uuid4())[:8]
        request.state.trace_id = trace_id
        response = await call_next(request)
        response.headers["X-Trace-ID"] = trace_id
        return response

app.add_middleware(TraceMiddleware)

Ejercicio 5: Health check que valida colección existente

Objetivo: /health debe comprobar que la colección por defecto existe y tiene documentos.

Solución:

@app.get("/health")
async def health():
    try:
        client = get_chroma_client()
        col = client.get_collection("rag_docs")
        count = col.count()
        return {"status": "ok", "chromadb": "ok", "doc_count": count}
    except Exception as e:
        return JSONResponse(
            status_code=503,
            content={"status": "degraded", "chromadb": str(e)}
        )

Ejercicio 6: docker-compose con Redis para cache (opcional)

Objetivo: Añadir servicio redis y variable REDIS_URL para la API.

Solución:

  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s

  rag-api:
    environment:
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      redis:
        condition: service_healthy

---

Troubleshooting de Deployment

"Funciona local, falla en contenedor"

Revisa variables de entorno: CHROMA_HOST debe ser chromadb (nombre del servicio), no localhost. Revisa rutas de persistencia: ChromaDB debe escribir en un volumen, no en /tmp efímero.

"Healthcheck OK, pero /ask falla"

Valida dependencia de ChromaDB: ¿la colección existe? ¿Tiene datos? Verifica OPENAI_API_KEY y que el contenedor tenga acceso a internet para la API de OpenAI.

"No vemos logs útiles"

Estandariza formato JSON y agrega trace_id a cada log. Usa LOG_LEVEL=DEBUG en desarrollo y INFO en producción.

"La imagen pesa demasiado"

Usa multi-stage y slim. Evita apt-get install de paquetes pesados. Revisa con docker history rag-api:latest.

"ChromaDB pierde datos al reiniciar"

Asegúrate de que el volumen esté montado correctamente. En docker-compose, chroma_data debe mapear a la ruta que ChromaDB usa para persistir (por defecto /chroma/chroma en la imagen oficial).

---

Comandos de Referencia Rápida

# Levantar todo
docker-compose up -d

# Ver estado
docker-compose ps
docker-compose logs -f rag-api

# Rebuild tras cambios
docker-compose build --no-cache rag-api
docker-compose up -d rag-api

# Ejecutar tests contra API levantada
API_BASE_URL=http://localhost:8000 pytest tests/integration -v

# Escalar réplicas (si configurado)
docker-compose up -d --scale rag-api=2

---

Ejemplo de Log Estructurado Completo

{
  "timestamp": "2026-03-13T12:00:00.123Z",
  "level": "INFO",
  "message": "Processing ask request",
  "trace_id": "req-a1b2c3d4",
  "module": "ask",
  "question_length": 42,
  "retrieval_docs_count": 5,
  "latency_ms": 1850
}

Este formato permite buscar por trace_id en cualquier agregador de logs (Elasticsearch, Datadog, etc.).

---

Grafana Dashboard Sugerido

Si usas Prometheus + Grafana, crea un dashboard con:

1. Panel de latencia: gráfica de histogram_quantile(0.95, rag_ask_latency_seconds_bucket)
2. Panel de throughput: rate(rag_requests_total[5m])
3. Panel de errores: rate(rag_requests_total{status="500"}[5m]) / rate(rag_requests_total[5m])
4. Panel de health: up/down del target

---

Resumen

• Empaquetaste el sistema con Dockerfile multi-stage y docker-compose (rag-api + chromadb-server).
• Configuraste variables de entorno para ChromaDB, OpenAI y nivel de log.
• Implementaste health checks en API y ChromaDB para detección de fallos.
• Integraste Prometheus con métricas de latencia y contadores.
• Configuraste logging estructurado con trace_id.
• El sistema se levanta con un solo comando y es observable desde el primer día.

Próximo paso: Hardening final (errores, cache, seguridad, documentación) en Cápsula 07.

---

Recursos Adicionales

• Docker Documentation — Dockerfile, multi-stage
• Docker Compose — orquestación de servicios
• Prometheus — métricas y alerting
• prometheus-client Python — instrumentación
• ChromaDB Docker — despliegue de ChromaDB
• FastAPI Deployment — opciones de deploy
• 12-Factor App — configuración, logs, procesos
• OpenTelemetry — traces para observabilidad avanzada

---

Tiempo estimado: 50-60 minutos
Siguiente: 07-hardening-final.md