7. Deployment y Optimización

Descripción

Tienes los componentes individuales: DocumentProcessor, VisionAnalyzer, RAGModule, AudioModule. Funcionan en tu máquina local. Ahora necesitas que funcionen en producción: empaquetados en Docker, expuestos como API REST con FastAPI, con variables de entorno seguras, logging estructurado, rate limiting, monitoring de costos, y optimizaciones de latencia. Esta cápsula transforma tu proyecto local en un servicio deployable.

Por qué importa: Un sistema que solo funciona en tu laptop no es portfolio-worthy. La diferencia entre "hice un proyecto" y "construí un servicio" es el deployment: Docker, API REST, manejo de errores HTTP, logging, health checks, y configuración externalizada. Estos patrones son los que un hiring manager busca.

Conexión con el módulo: En el Módulo 7 aprendiste patrones de producción: selección de caso de uso, fallbacks, optimización de costos. Aquí los aplicas al Document Analyzer completo: cada patrón se materializa en código, configuración, o infraestructura.

Checklist de Producción

Infraestructura

• Dockerfile multi-stage para build optimizado
• docker-compose.yml con servicio + volúmenes
• Variables de entorno desde .env (no hardcodeadas)
• API keys como secrets, nunca en código
• Health check endpoint funcional
• Logging estructurado en JSON

API REST

• Validación de inputs (tipo, tamaño, formato)
• Rate limiting por IP (10 requests/minuto)
• Timeout configurable por operación
• Manejo de errores con códigos HTTP apropiados
• CORS configurado para frontend (si aplica)
• Documentación automática con Swagger/OpenAPI

Optimización

• Cache de descripciones de imágenes (no re-describir)
• Cache de audio generado (no re-sintetizar)
• Usar gpt-4o-mini donde sea suficiente
• Reducir resolución de imágenes antes de Vision
• Índice RAG persistente (no re-indexar cada request)
• Procesamiento en background para documentos grandes

Monitoring

• Métricas: latencia por operación, errores, costos estimados
• Logging de cada request con duración y resultado
• Tracking de costos acumulados por API key
• Alertas por umbral de error (>5% en 5 min)

FastAPI: Endpoints Completos

Estructura de la aplicación

import logging
import os
import time
import uuid
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Optional

from fastapi import FastAPI, File, Form, HTTPException, UploadFile, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse
from pydantic import BaseModel, Field

from modules.document_processor import DocumentProcessor
from modules.vision_analyzer import VisionAnalyzer
from modules.rag_module import RAGModule
from modules.audio_module import AudioModule
from models.schemas import AnalyzeResponse, QARequest, QAResult

logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.processor = DocumentProcessor()
    app.state.analyzer = VisionAnalyzer()
    app.state.rag = RAGModule(persist_directory="./chroma_data")
    app.state.audio = AudioModule(output_dir="./audio_output")
    app.state.start_time = time.time()
    logger.info("Servicios inicializados")
    yield
    logger.info("Servicios cerrados")


app = FastAPI(
    title="Document Analyzer Multimodal",
    description="API para análisis multimodal de documentos",
    version="1.0.0",
    lifespan=lifespan
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


UPLOAD_DIR = "/tmp/uploads"
MAX_FILE_SIZE = int(os.getenv("MAX_FILE_SIZE_MB", "50")) * 1024 * 1024
SUPPORTED_EXTENSIONS = {".pdf", ".png", ".jpg", ".jpeg", ".webp"}

Endpoint POST /analyze

@app.post("/analyze", response_model=AnalyzeResponse)
async def analyze_document(
    request: Request,
    file: UploadFile = File(...),
    question: Optional[str] = Form(None),
    extract_structured: bool = Form(True),
    generate_summary: bool = Form(True),
    generate_audio_summary: bool = Form(False),
    index_for_qa: bool = Form(True),
    audio_voice: str = Form("nova")
):
    start = time.time()
    doc_id = str(uuid.uuid4())
    errors: list[str] = []

    ext = Path(file.filename).suffix.lower()
    if ext not in SUPPORTED_EXTENSIONS:
        raise HTTPException(400, f"Formato '{ext}' no soportado")

    os.makedirs(UPLOAD_DIR, exist_ok=True)
    file_path = os.path.join(UPLOAD_DIR, f"{doc_id}{ext}")

    try:
        content_bytes = await file.read()
        if len(content_bytes) > MAX_FILE_SIZE:
            raise HTTPException(413, "Archivo excede el tamaño máximo")

        with open(file_path, "wb") as f:
            f.write(content_bytes)

        processor = request.app.state.processor
        analyzer = request.app.state.analyzer
        rag = request.app.state.rag
        audio = request.app.state.audio

        processed = processor.process(file_path)

        doc_type = "otro"
        try:
            doc_type = analyzer.classify(processed)
        except Exception as e:
            errors.append(f"Clasificación falló: {e}")
            logger.warning(f"Clasificación falló para {doc_id}: {e}")

        extracted_data = None
        if extract_structured:
            try:
                extraction = analyzer.extract_structured(processed, doc_type)
                extracted_data = {
                    "document_type": extraction.document_type,
                    "fields": extraction.fields,
                    "confidence": extraction.confidence
                }
            except Exception as e:
                errors.append(f"Extracción estructurada falló: {e}")
                logger.warning(f"Extracción falló para {doc_id}: {e}")

        summary = None
        if generate_summary:
            try:
                summary = _generate_summary(analyzer, processed)
            except Exception as e:
                errors.append(f"Resumen falló: {e}")

        indexed = False
        if index_for_qa:
            try:
                image_descriptions = None
                if processed.has_image_pages:
                    image_descriptions = analyzer.describe_for_rag(
                        processed.get_images_for_vision()
                    )
                rag.index(
                    doc_id=doc_id,
                    content=processed,
                    filename=file.filename,
                    document_type=doc_type,
                    image_descriptions=image_descriptions
                )
                indexed = True
            except Exception as e:
                errors.append(f"Indexación falló: {e}")

        qa_result = None
        if question:
            try:
                if indexed:
                    qa = rag.query(question=question, doc_id=doc_id)
                    qa_result = {
                        "question": qa.question,
                        "answer": qa.answer,
                        "sources": qa.sources,
                        "confidence": qa.confidence
                    }
                else:
                    qa_result = _direct_qa(analyzer, processed, question)
            except Exception as e:
                errors.append(f"Q&A falló: {e}")

        audio_url = None
        if generate_audio_summary and summary:
            try:
                audio_filename = f"{doc_id}_resumen.mp3"
                audio.generate_summary_audio(
                    text=summary,
                    filename=audio_filename,
                    voice=audio_voice
                )
                audio_url = f"/audio/{audio_filename}"
            except Exception as e:
                errors.append(f"Audio falló: {e}")

        latency = round(time.time() - start, 2)

        return AnalyzeResponse(
            success=True,
            doc_id=doc_id,
            extracted_data=extracted_data,
            summary=summary,
            qa_result=qa_result,
            audio_summary_url=audio_url,
            metadata={
                "doc_id": doc_id,
                "filename": file.filename,
                "file_type": ext.lstrip("."),
                "pages_processed": processed.total_pages,
                "document_type": doc_type,
                "indexed": indexed,
                "latency_seconds": latency,
                "estimated_cost_usd": _estimate_total_cost(processed, extract_structured, generate_audio_summary)
            },
            errors=errors
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error procesando {doc_id}: {e}", exc_info=True)
        raise HTTPException(500, f"Error interno: {str(e)}")
    finally:
        Path(file_path).unlink(missing_ok=True)

Endpoint POST /ask

@app.post("/ask")
async def ask_question(request: Request, body: QARequest):
    rag = request.app.state.rag

    if not body.question or len(body.question.strip()) < 5:
        raise HTTPException(400, "Pregunta demasiado corta (mínimo 5 caracteres)")

    try:
        result = rag.query(
            question=body.question,
            doc_id=body.doc_id
        )
        return result
    except Exception as e:
        logger.error(f"Error en Q&A: {e}", exc_info=True)
        raise HTTPException(500, f"Error generando respuesta: {str(e)}")

Endpoint GET /health

@app.get("/health")
async def health_check(request: Request):
    from openai import OpenAI

    checks = {}

    try:
        client = OpenAI()
        client.models.list()
        checks["openai"] = "connected"
    except Exception as e:
        checks["openai"] = f"error: {str(e)}"

    try:
        rag = request.app.state.rag
        count = rag.collection.count()
        checks["chromadb"] = "connected"
        checks["indexed_documents"] = count
    except Exception as e:
        checks["chromadb"] = f"error: {str(e)}"

    uptime = round(time.time() - request.app.state.start_time, 1)
    checks["uptime_seconds"] = uptime

    all_ok = all(
        v == "connected" for k, v in checks.items()
        if k in ("openai", "chromadb")
    )

    status_code = 200 if all_ok else 503
    checks["status"] = "ok" if all_ok else "degraded"

    return JSONResponse(content=checks, status_code=status_code)

Endpoint GET /audio/{filename}

@app.get("/audio/{filename}")
async def serve_audio(filename: str):
    if not filename.endswith((".mp3", ".wav")):
        raise HTTPException(400, "Formato no soportado")

    file_path = os.path.join("./audio_output", filename)
    if not os.path.exists(file_path):
        raise HTTPException(404, "Audio no encontrado")

    return FileResponse(file_path, media_type="audio/mpeg", filename=filename)

Funciones auxiliares

def _generate_summary(analyzer, content) -> str:
    from openai import OpenAI
    client = OpenAI()

    if content.full_text:
        text = content.full_text[:6000]
    elif content.has_image_pages:
        descriptions = analyzer.describe_for_rag(content.get_images_for_vision()[:5])
        text = " ".join(descriptions)
    else:
        return "Documento sin contenido extraíble."

    r = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"Resume este documento en 3-5 puntos clave:\n\n{text}"
        }],
        max_tokens=500
    )
    return r.choices[0].message.content


def _direct_qa(analyzer, content, question: str) -> dict:
    from openai import OpenAI
    client = OpenAI()

    if content.full_text:
        context = content.full_text[:6000]
    else:
        return {"question": question, "answer": "Documento sin texto para Q&A directo."}

    r = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"Contexto:\n{context}\n\nPregunta: {question}\n\nResponde basándote solo en el contexto."
        }],
        max_tokens=300
    )
    return {
        "question": question,
        "answer": r.choices[0].message.content,
        "sources": [],
        "confidence": None
    }


def _estimate_total_cost(content, extract: bool, audio: bool) -> float:
    cost = 0.002  # base: clasificación + resumen
    if extract:
        cost += 0.01 if content.has_image_pages else 0.003
    if audio:
        cost += 0.01
    cost += content.image_page_count * 0.005  # descripciones de imagen
    cost += 0.001  # embeddings
    return round(cost, 4)

Docker

Dockerfile

FROM python:3.11-slim AS base

WORKDIR /app

RUN apt-get update && \
    apt-get install -y --no-install-recommends \
        poppler-utils \
        ffmpeg \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

RUN mkdir -p /tmp/uploads /tmp/audio ./audio_output ./chroma_data

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

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

docker-compose.yml

version: "3.8"

services:
  document-analyzer:
    build: .
    ports:
      - "8000:8000"
    env_file:
      - .env
    volumes:
      - chroma_data:/app/chroma_data
      - audio_output:/app/audio_output
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

volumes:
  chroma_data:
  audio_output:

requirements.txt

openai>=1.0.0
pymupdf>=1.24.0
pillow>=10.0.0
pydantic>=2.0.0
chromadb>=0.5.0
fastapi>=0.110.0
uvicorn>=0.29.0
python-multipart>=0.0.9
python-dotenv>=1.0.0
pydub>=0.25.0
slowapi>=0.1.9

Build y ejecución

docker build -t document-analyzer .

docker run -d \
  --name doc-analyzer \
  -p 8000:8000 \
  --env-file .env \
  -v doc_chroma:/app/chroma_data \
  -v doc_audio:/app/audio_output \
  document-analyzer

Con docker-compose:

docker compose up -d
docker compose logs -f

Variables de Entorno

Archivo .env

# Obligatorias
OPENAI_API_KEY=sk-...

# Opcionales — fallback providers
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=...

# Configuración del servicio
LOG_LEVEL=INFO
MAX_FILE_SIZE_MB=50
REQUEST_TIMEOUT=120

# ChromaDB
CHROMA_PERSIST_DIR=./chroma_data

# Audio
AUDIO_OUTPUT_DIR=./audio_output
DEFAULT_TTS_VOICE=nova

Cargar configuración

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    openai_api_key: str
    anthropic_api_key: str = ""
    google_api_key: str = ""
    log_level: str = "INFO"
    max_file_size_mb: int = 50
    request_timeout: int = 120
    chroma_persist_dir: str = "./chroma_data"
    audio_output_dir: str = "./audio_output"
    default_tts_voice: str = "nova"

    class Config:
        env_file = ".env"


settings = Settings()

Logging Estructurado

Configuración

import logging
import json
import sys


class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
        }
        if record.exc_info and record.exc_info[0]:
            log_data["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_data)


def setup_logging(level: str = "INFO"):
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JSONFormatter())

    root = logging.getLogger()
    root.setLevel(getattr(logging, level.upper()))
    root.addHandler(handler)

Middleware de logging

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.time()
    request_id = str(uuid.uuid4())[:8]

    logger.info(json.dumps({
        "event": "request_start",
        "request_id": request_id,
        "method": request.method,
        "path": request.url.path,
        "client": request.client.host if request.client else "unknown"
    }))

    response = await call_next(request)

    duration = round(time.time() - start, 3)
    logger.info(json.dumps({
        "event": "request_end",
        "request_id": request_id,
        "status": response.status_code,
        "duration_seconds": duration
    }))

    response.headers["X-Request-ID"] = request_id
    response.headers["X-Duration-Seconds"] = str(duration)

    return response

Rate Limiting

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)


@app.post("/analyze", response_model=AnalyzeResponse)
@limiter.limit("10/minute")
async def analyze_document(request: Request, ...):
    ...


@app.post("/ask")
@limiter.limit("30/minute")
async def ask_question(request: Request, ...):
    ...

Optimización de Costos

Estrategias implementadas

Selección de modelo por operación

MODEL_SELECTION = {
    "classify": "gpt-4o-mini",           # barato, suficiente
    "extract_structured_text": "gpt-4o-mini",  # texto → mini suficiente
    "extract_structured_image": "gpt-4o",      # vision → necesita 4o
    "summarize": "gpt-4o-mini",           # resumen → mini suficiente
    "describe_image": "gpt-4o-mini",      # descripción → mini suficiente
    "qa_generate": "gpt-4o",              # respuesta detallada → 4o mejor
    "embeddings": "text-embedding-3-small",  # más barato
}

Cache simple

import hashlib
from functools import lru_cache


class ResultCache:
    def __init__(self):
        self._cache: dict[str, any] = {}

    def get(self, key: str):
        return self._cache.get(key)

    def set(self, key: str, value):
        self._cache[key] = value

    def make_key(self, *args) -> str:
        content = "|".join(str(a)[:200] for a in args)
        return hashlib.md5(content.encode()).hexdigest()


cache = ResultCache()


def classify_with_cache(analyzer, content) -> str:
    cache_key = cache.make_key("classify", content.file_path, content.total_pages)
    cached = cache.get(cache_key)
    if cached:
        return cached

    result = analyzer.classify(content)
    cache.set(cache_key, result)
    return result

Monitoring de Costos

Cost tracker

class CostTracker:
    def __init__(self):
        self.operations: list[dict] = []

    def track(self, operation: str, model: str, tokens_in: int = 0, tokens_out: int = 0):
        cost = self._calculate_cost(model, tokens_in, tokens_out)
        self.operations.append({
            "operation": operation,
            "model": model,
            "tokens_in": tokens_in,
            "tokens_out": tokens_out,
            "cost_usd": cost,
            "timestamp": time.time()
        })

    def get_total(self) -> float:
        return sum(op["cost_usd"] for op in self.operations)

    def get_breakdown(self) -> dict:
        by_operation = {}
        for op in self.operations:
            name = op["operation"]
            if name not in by_operation:
                by_operation[name] = {"count": 0, "cost": 0}
            by_operation[name]["count"] += 1
            by_operation[name]["cost"] += op["cost_usd"]
        return by_operation

    def _calculate_cost(self, model: str, tokens_in: int, tokens_out: int) -> float:
        rates = {
            "gpt-4o": {"in": 2.50 / 1_000_000, "out": 10.00 / 1_000_000},
            "gpt-4o-mini": {"in": 0.15 / 1_000_000, "out": 0.60 / 1_000_000},
            "text-embedding-3-small": {"in": 0.02 / 1_000_000, "out": 0},
        }
        rate = rates.get(model, {"in": 0.001, "out": 0.002})
        return round(tokens_in * rate["in"] + tokens_out * rate["out"], 6)

Troubleshooting

"Docker build falla con error de PyMuPDF"

Causa: PyMuPDF necesita compilarse y la imagen slim no tiene build tools.

Solución: Usar la imagen con build tools o instalar dependencias:

RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && pip install --no-cache-dir -r requirements.txt \
    && apt-get remove -y build-essential \
    && apt-get autoremove -y \
    && rm -rf /var/lib/apt/lists/*

"El servicio se queda sin memoria con PDFs grandes"

Solución: Limitar workers de uvicorn y agregar swap:

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1", "--limit-max-requests", "100"]

"Rate limit (429) de OpenAI en producción"

Solución: Implementar retry con backoff exponencial:

import time


def retry_with_backoff(fn, max_retries: int = 3, base_delay: float = 1.0):
    for attempt in range(max_retries):
        try:
            return fn()
        except Exception as e:
            if "429" in str(e) and attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt)
                logger.warning(f"Rate limit, retry en {delay}s (intento {attempt + 1})")
                time.sleep(delay)
            else:
                raise

"ChromaDB pierde datos al reiniciar el contenedor"

Solución: Montar volumen para persistencia:

volumes:
  - chroma_data:/app/chroma_data

Ejecución Local

Sin Docker

source venv/bin/activate
uvicorn main:app --reload --host 0.0.0.0 --port 8000

Test con curl

curl -X POST "http://localhost:8000/analyze" \
  -F "file=@factura_ejemplo.pdf" \
  -F "question=¿Cuál es el total?" \
  -F "extract_structured=true" \
  -F "generate_summary=true" \
  -F "generate_audio_summary=true" \
  -F "audio_voice=nova"

curl http://localhost:8000/health

curl -X POST "http://localhost:8000/ask" \
  -H "Content-Type: application/json" \
  -d '{"question": "¿Quién es el proveedor?", "doc_id": "doc-id-aquí"}'

Test con Python

import requests

with open("factura_ejemplo.pdf", "rb") as f:
    response = requests.post(
        "http://localhost:8000/analyze",
        files={"file": ("factura.pdf", f, "application/pdf")},
        data={
            "question": "¿Cuál es el total?",
            "extract_structured": "true",
            "generate_summary": "true",
            "generate_audio_summary": "false"
        }
    )

result = response.json()
print(f"Éxito: {result['success']}")
print(f"Tipo: {result['metadata']['document_type']}")
print(f"Resumen: {result['summary']}")
if result.get("qa_result"):
    print(f"Respuesta: {result['qa_result']['answer']}")

Ejercicios

Ejercicio 1: Rate limiting por API key

Implementa rate limiting diferenciado: usuarios con API key premium tienen 50 requests/minuto, usuarios sin key tienen 5 requests/minuto. Usa slowapi con key function personalizada.

from slowapi import Limiter


def get_rate_limit_key(request: Request) -> str:
    api_key = request.headers.get("X-API-Key", "")
    if api_key:
        return f"premium_{api_key}"
    return get_remote_address(request)


def get_rate_limit(request: Request) -> str:
    api_key = request.headers.get("X-API-Key", "")
    if api_key and api_key.startswith("premium_"):
        return "50/minute"
    return "5/minute"


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


@app.post("/analyze")
@limiter.limit(lambda: "50/minute", key_func=lambda request: (
    f"premium_{request.headers.get('X-API-Key', '')}"
    if request.headers.get("X-API-Key")
    else get_remote_address(request)
))
async def analyze_document(request: Request, ...):
    ...

Ejercicio 2: Dashboard de costos

Implementa un endpoint GET /costs que retorne: costo total acumulado, costo por operación, costo por modelo, y los últimos 10 requests con su costo individual.

cost_tracker = CostTracker()


@app.get("/costs")
async def get_costs():
    breakdown = cost_tracker.get_breakdown()

    by_model = {}
    for op in cost_tracker.operations:
        model = op["model"]
        if model not in by_model:
            by_model[model] = {"count": 0, "cost": 0}
        by_model[model]["count"] += 1
        by_model[model]["cost"] += op["cost_usd"]

    recent = sorted(cost_tracker.operations, key=lambda x: x["timestamp"], reverse=True)[:10]

    return {
        "total_cost_usd": round(cost_tracker.get_total(), 4),
        "total_operations": len(cost_tracker.operations),
        "by_operation": {
            k: {"count": v["count"], "cost_usd": round(v["cost"], 4)}
            for k, v in breakdown.items()
        },
        "by_model": {
            k: {"count": v["count"], "cost_usd": round(v["cost"], 4)}
            for k, v in by_model.items()
        },
        "recent_operations": [
            {
                "operation": op["operation"],
                "model": op["model"],
                "cost_usd": round(op["cost_usd"], 6),
                "tokens": op["tokens_in"] + op["tokens_out"]
            }
            for op in recent
        ]
    }

Resumen

• FastAPI expone 4 endpoints: /analyze, /ask, /health, /audio/{filename}.
• Docker empaqueta el servicio con todas las dependencias (poppler, ffmpeg).
• Variables de entorno gestionan API keys y configuración sin hardcodear.
• Logging estructurado en JSON facilita debugging y monitoring.
• Rate limiting protege contra abuso (slowapi).
• Optimización de costos: selección de modelo por operación, cache, reducción de resolución.
• Cost tracker registra cada operación para monitoring y presupuesto.
• El sistema es resiliente: errores parciales no crashean el request completo.

Recursos Adicionales

1. FastAPI Deployment — Guía oficial
2. Docker Best Practices — Dockerfile optimizado
3. slowapi — Rate limiting para FastAPI
4. OpenAI Rate Limits — Límites por tier
5. Pydantic Settings — Configuración tipada