Cápsula 07: Hardening Final

Descripción de la cápsula

Antes de cerrar el proyecto, aplicarás una pasada de hardening: robustez ante fallos, seguridad básica y mantenibilidad. Un sistema RAG "funciona" cuando retrieval y generation responden; un sistema "production-ready" sigue funcionando cuando ChromaDB se cae, el LLM timeout, o un atacante envía 10.000 requests por segundo.

En esta cápsula implementarás:

• Manejo de errores con retry logic, fallback responses y degradación controlada
• Caching (embedding cache + result cache) con métricas de hit rate
• Seguridad (API keys, rate limiting, validación de entradas)
• Documentación (README con diagrama, API reference, guía de deployment)

Al final tendrás un sistema que se defiende ante fallos y abusos, y documentación que permite a cualquier developer operarlo.

---

Por qué Hardening Importa

Escenarios reales de fallo

Día 1: Todo bien
Día 2: ChromaDB reinicia por deploy → /ask devuelve 500 durante 30s
Día 3: OpenAI tiene latencia p99 de 15s → timeouts en cascada
Día 4: Un script mal escrito hace 500 req/seg → servicio caído para todos
Día 5: Alguien descubre que no hay auth → extrae todo el corpus

Sin hardening, cada uno de estos es un incidente. Con hardening: retries, fallbacks, rate limit y auth reducen el impacto.

---

Áreas de Hardening

Área	Qué hacer	Prioridad
Confiabilidad	Retries, timeouts, fallbacks cuando servicio externo falla	Alta
Seguridad	API key, rate limiting, validación de inputs	Alta
Performance	Caching de embeddings y resultados	Media
Errores	Formato uniforme 4xx/5xx, no exponer stack traces	Alta
Documentación	README, API reference, runbook de operación	Media

---

Error Handling Robusto

Retry Logic para ChromaDB y OpenAI

# app/utils/retry.py
import asyncio
import functools
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

def with_retry(max_attempts=3, exceptions=(ConnectionError, TimeoutError)):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            last_exc = None
            for attempt in range(max_attempts):
                try:
                    return await func(*args, **kwargs)
                except exceptions as e:
                    last_exc = e
                    if attempt < max_attempts - 1:
                        await asyncio.sleep(2 ** attempt)  # 1s, 2s, 4s
            raise last_exc
        return wrapper
    return decorator

# Uso
@with_retry(max_attempts=3)
async def get_chroma_results(collection, query_embedding, top_k=5):
    return collection.query(query_embeddings=[query_embedding], n_results=top_k)

Con tenacity (opción más completa):

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type((ConnectionError, TimeoutError))
)
async def call_openai(messages):
    # ...

Fallback Responses

Cuando retrieval o generation fallan, devolver una respuesta controlada en lugar de 500 genérico.

# app/api/ask.py
FALLBACK_MESSAGES = {
    "retrieval_failed": "No pude buscar en la base de conocimientos. Por favor intenta de nuevo en unos segundos.",
    "generation_failed": "Encontré información relevante pero tuve un problema al generar la respuesta. Intenta reformular tu pregunta.",
    "no_evidence": "No encontré información suficiente para responder tu pregunta con confianza. ¿Puedes ser más específico?",
}

async def ask_endpoint(payload: dict):
    question = payload.get("question", "").strip()[:500]  # límite
    if not question:
        raise HTTPException(400, "question is required")

    try:
        docs = await retrieve(question, top_k=5)
    except Exception as e:
        logger.warning(f"Retrieval failed: {e}")
        return {
            "answer": FALLBACK_MESSAGES["retrieval_failed"],
            "sources": [],
            "confidence": 0,
            "fallback": True,
        }

    if not docs or not docs.get("documents") or not docs["documents"][0]:
        return {
            "answer": FALLBACK_MESSAGES["no_evidence"],
            "sources": [],
            "confidence": 0,
        }

    try:
        answer = await generate_answer(question, docs["documents"][0])
    except Exception as e:
        logger.warning(f"Generation failed: {e}")
        return {
            "answer": FALLBACK_MESSAGES["generation_failed"],
            "sources": [m for m in docs.get("metadatas", [[]])[0]],
            "confidence": 0.5,
            "fallback": True,
        }

    return {"answer": answer, "sources": docs["metadatas"][0], "confidence": 0.85}

Graceful Degradation

Si ChromaDB está caído pero tienes cache de resultados, puedes seguir respondiendo preguntas frecuentes:

async def ask_with_degradation(question: str):
    # 1. Intentar cache de resultado primero
    cached = await result_cache.get(question)
    if cached:
        return {**cached, "from_cache": True}

    # 2. Intentar retrieval normal
    try:
        docs = await retrieve(question)
    except Exception:
        # 3. Degradar: responder solo con modelo (sin RAG) o mensaje de indisponibilidad
        return {"answer": "El servicio de búsqueda no está disponible. Intenta más tarde.", "sources": []}

    # ... flujo normal

---

Caching

Embedding Cache

Evita re-computar embeddings para las mismas preguntas o textos.

# app/cache/embedding_cache.py
import hashlib
import json
from typing import Optional

# Usando Redis o dict en memoria para desarrollo
class EmbeddingCache:
    def __init__(self, redis_url: Optional[str] = None):
        self._redis = redis.from_url(redis_url) if redis_url else {}
        self._local = {} if not redis_url else None  # fallback in-memory

    def _key(self, text: str, model: str) -> str:
        h = hashlib.sha256(f"{model}:{text}".encode()).hexdigest()
        return f"emb:{h}"

    async def get(self, text: str, model: str = "text-embedding-3-small") -> Optional[list]:
        k = self._key(text, model)
        if self._redis:
            val = await self._redis.get(k)
            return json.loads(val) if val else None
        return self._local.get(k)

    async def set(self, text: str, embedding: list, model: str = "text-embedding-3-small", ttl: int = 86400):
        k = self._key(text, model)
        val = json.dumps(embedding)
        if self._redis:
            await self._redis.setex(k, ttl, val)
        else:
            self._local[k] = embedding

Result Cache

Cachea respuestas completas de /ask para preguntas idénticas.

# app/cache/result_cache.py
class ResultCache:
    def __init__(self, redis_url: Optional[str] = None, ttl: int = 3600):
        self._redis = redis.from_url(redis_url) if redis_url else {}
        self._ttl = ttl

    def _key(self, question: str) -> str:
        return f"ask:{hashlib.sha256(question.strip().lower().encode()).hexdigest()}"

    async def get(self, question: str) -> Optional[dict]:
        k = self._key(question)
        if self._redis:
            val = await self._redis.get(k)
            return json.loads(val) if val else None
        return None

    async def set(self, question: str, result: dict):
        k = self._key(question)
        if self._redis:
            await self._redis.setex(k, self._ttl, json.dumps(result))

Métricas de Hit Rate

# app/metrics.py
from prometheus_client import Counter
cache_hits = Counter("rag_cache_hits_total", "Cache hits", ["cache_type"])  # embedding, result
cache_misses = Counter("rag_cache_misses_total", "Cache misses", ["cache_type"])

# En el flujo
if cached_emb := await embedding_cache.get(text):
    cache_hits.labels(cache_type="embedding").inc()
    return cached_emb
cache_misses.labels(cache_type="embedding").inc()
emb = await get_embedding(text)
await embedding_cache.set(text, emb)

---

Seguridad

API Key en Headers

# app/auth.py
from fastapi import Security, HTTPException
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

async def verify_api_key(api_key: str = Security(api_key_header)):
    expected = os.getenv("API_KEY")
    if not expected:
        return None  # Sin API key configurada, permitir (solo para dev)
    if api_key != expected:
        raise HTTPException(403, "Invalid API key")
    return api_key

@app.post("/ask", dependencies=[Depends(verify_api_key)])
async def ask(payload: dict):
    ...

Rate Limiting

# app/middleware/rate_limit.py
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.post("/ask")
@limiter.limit("20/minute")
async def ask(request: Request, payload: dict):
    ...

Alternativa simple sin dependencias externas:

from collections import defaultdict
from time import time

class SimpleRateLimiter:
    def __init__(self, requests_per_minute=60):
        self.rpm = requests_per_minute
        self.requests = defaultdict(list)

    def is_allowed(self, client_id: str) -> bool:
        now = time()
        self.requests[client_id] = [t for t in self.requests[client_id] if now - t < 60]
        if len(self.requests[client_id]) >= self.rpm:
            return False
        self.requests[client_id].append(now)
        return True

Validación de Entradas

from pydantic import BaseModel, Field, validator

class AskPayload(BaseModel):
    question: str = Field(..., min_length=1, max_length=500)

    @validator("question")
    def sanitize(cls, v):
        v = v.strip()
        if not v:
            raise ValueError("question cannot be empty")
        # Opcional: rechazar patrones peligrosos
        if "<?php" in v.lower() or "<script" in v.lower():
            raise ValueError("Invalid characters in question")
        return v

No Exponer Secretos en Logs

# Mal
logger.info(f"Calling OpenAI with key {api_key[:8]}...")  # Evitar

# Bien
logger.info("Calling OpenAI", extra={"key_prefix": api_key[:4] + "***" if api_key else "not_set"})

---

Formato Uniforme de Errores

# app/exceptions.py
from fastapi import Request, status
from fastapi.responses import JSONResponse

def error_response(status_code: int, detail: str, trace_id: str = None):
    return JSONResponse(
        status_code=status_code,
        content={
            "error": True,
            "detail": detail,
            "trace_id": trace_id,
        },
    )

@app.exception_handler(422)
async def validation_exception_handler(request: Request, exc):
    return error_response(422, "Invalid input", getattr(request.state, "trace_id", None))

@app.exception_handler(500)
async def server_exception_handler(request: Request, exc):
    logger.exception("Unhandled error")
    return error_response(500, "Internal server error", getattr(request.state, "trace_id", None))

---

Documentación

README con Arquitectura

# RAG API con ChromaDB

Sistema RAG production-ready: ingestion, retrieval, generation y API REST.

## Arquitectura

\`\`\`
Documentos → Ingestion → ChromaDB → Retrieval → Generation → API
                ↑              ↑           ↑
            chunking      vector store    LLM (OpenAI)
\`\`\`

## Requisitos

- Python 3.11+
- Docker y Docker Compose
- OpenAI API Key

## Uso Local

\`\`\`bash
cp .env.example .env
docker-compose up -d
curl -X POST http://localhost:8000/ask -H "Content-Type: application/json" -d '{"question":"¿Qué es RAG?"}'
\`\`\`

## API Reference

| Endpoint | Método | Descripción |
|----------|--------|-------------|
| /health | GET | Health check |
| /search | GET | Búsqueda semántica (q, top_k) |
| /ask | POST | Pregunta RAG (question) |
| /ingest | POST | Ingestion de documentos |

## Deployment

Ver [DEPLOYMENT.md](./DEPLOYMENT.md).

API Reference Automática

FastAPI genera Swagger en /docs y ReDoc en /redoc. Asegúrate de documentar parámetros:

@app.post("/ask", summary="Pregunta RAG")
async def ask(
    payload: AskPayload,
    request: Request,
    api_key: str = Depends(verify_api_key)
):
    """
    Recibe una pregunta, recupera contexto de ChromaDB y genera respuesta con fuentes.
    Requiere X-API-Key en header si está configurado.
    """
    ...

---

Checklist de Hardening

• No se exponen secretos en logs
• Hay fallbacks cuando ChromaDB o OpenAI fallan
• Errores 4xx/5xx tienen formato uniforme
• Retry logic en llamadas a ChromaDB y OpenAI
• Rate limiting en endpoints públicos
• API key opcional para protección
• Validación de inputs con Pydantic
• Documentación de endpoints actualizada
• README con arquitectura, setup y troubleshooting
• Plan de rollback documentado
• Caching de embeddings y resultados (opcional pero recomendado)
• Métricas de cache hit rate

---

Priorización de Hardening

Orden	Área	Esfuerzo	Impacto
1	Confiabilidad: retries, fallbacks	Medio	Alto
2	Seguridad: auth, rate limit, validación	Medio	Alto
3	Errores: formato uniforme, no stack traces	Bajo	Medio
4	Caching	Medio	Alto (performance)
5	Documentación	Medio	Medio (mantenibilidad)
6	Plan de rollback	Bajo	Alto (operación)

---

Ejercicios con Soluciones Detalladas

Ejercicio 1: Retry con backoff exponencial para OpenAI

Objetivo: Implementar retry solo para errores 429 y 5xx de OpenAI.

Solución:

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception(lambda e: "429" in str(e) or "5" in str(e)[:1])
)
async def call_openai_completion(messages):
    # ...

Ejercicio 2: Fallback a respuesta corta cuando LLM timeout

Objetivo: Si OpenAI tarda más de 30s, devolver "Consulta muy compleja, intenta reformular".

Solución:

try:
    answer = await asyncio.wait_for(generate_answer(question, docs), timeout=30.0)
except asyncio.TimeoutError:
    answer = "Consulta muy compleja. Intenta reformular o simplificar tu pregunta."

Ejercicio 3: Rate limit por API key en lugar de IP

Objetivo: Limitar 100 req/min por API key para multi-tenant.

Solución:

def get_client_id(request: Request) -> str:
    return request.headers.get("X-API-Key", get_remote_address(request))

limiter = Limiter(key_func=get_client_id)

Ejercicio 4: Cache hit rate en endpoint /metrics

Objetivo: Exponer gauge rag_cache_hit_rate calculado como hits/(hits+misses).

Solución:

from prometheus_client import Gauge
hit_rate = Gauge("rag_cache_hit_rate", "Cache hit rate", ["cache_type"])

# Actualizar en cada hit/miss
def update_hit_rate(cache_type):
    h, m = cache_hits.labels(cache_type=cache_type)._value.get(), cache_misses.labels(cache_type=cache_type)._value.get()
    hit_rate.labels(cache_type=cache_type).set(h / (h + m) if (h + m) > 0 else 0)

Ejercicio 5: README con diagrama Mermaid

Objetivo: Incluir diagrama de flujo en README.

Solución:

## Arquitectura

\`\`\`mermaid
flowchart LR
    A[Documentos] --> B[Chunking]
    B --> C[Embeddings]
    C --> D[ChromaDB]
    D --> E[Retrieval]
    E --> F[Generation]
    F --> G[API Response]
\`\`\`

Ejercicio 6: Plan de rollback en 3 pasos

Objetivo: Documentar rollback en DEPLOYMENT.md.

Solución:

## Rollback

1. Revertir a imagen anterior: `docker pull rag-api:v1.2.2 && docker-compose up -d rag-api`
2. Si hay migración de ChromaDB: restaurar backup de volumen
3. Verificar health y métricas antes de cerrar incidente

---

Ejercicio de Hardening en 60 Minutos

Reparte el tiempo así:

Tiempo	Foco
20 min	Errores: retries, fallbacks, formato uniforme
20 min	Seguridad: API key, rate limit, validación
20 min	Docs: README, runbook, rollback

Entrega: Lista de hallazgos y acciones con prioridad (crítico / importante / nice-to-have).

---

Troubleshooting Final

"Todo parece bien, pero no confiamos en release"

Haz smoke tests + canary interno antes de entregar. Despliega en staging, corre los integration tests, simula fallo de ChromaDB y verifica que los fallbacks funcionen.

"Tenemos hardening técnico pero docs pobres"

Sin runbook ni README operativo, el proyecto no está listo. Cualquier developer debe poder levantar el sistema y entender qué hace cada endpoint. Invierte 1-2 horas en documentación.

"Rollback no está probado"

Define y valida rollback mínimo en staging: qué comando ejecutar, cómo restaurar datos si hace falta, cómo verificar que todo está bien.

"Cache está causando respuestas obsoletas"

Ajusta TTL. Para embedding cache 24h suele estar bien; para result cache 1h o menos si el corpus se actualiza con frecuencia. Añade forma de invalidar cache por patrón o flush total.

"Rate limit está bloqueando usuarios legítimos"

Aumenta límites o implementa límites por tenant. Monitorea metricas de rechazos y ajusta.

---

Resumen

• Implementaste manejo de errores robusto con retries, fallbacks y degradación controlada.
• Añadiste caching de embeddings y resultados con métricas de hit rate.
• Aplicaste seguridad básica: API key, rate limiting, validación de inputs.
• Estandarizaste errores en formato uniforme sin exponer stack traces.
• Documentaste con README (arquitectura, setup, API reference) y guía de deployment.
• Definiste plan de rollback para operación.

El sistema queda preparado para entrega técnica seria. La siguiente cápsula consolida todo como proyecto final.

---

Recursos Adicionales

• OWASP API Security — top 10 vulnerabilidades
• Tenacity — retry decorators
• SlowAPI — rate limiting para FastAPI
• Redis Cache Patterns — caching
• Pydantic Validation — validación de datos
• FastAPI Security — API keys, OAuth
• Keep a Changelog — formato de CHANGELOG
• Semantic Versioning — versionado de API

---

Tiempo estimado: 55-65 minutos
Siguiente: 08-proyecto-rag-production-ready.md