Puesta en producción de un modelo de deep learning

import torch
import torch.nn as nn
from torchvision import datasets, transforms
from torch.utils.data import DataLoader

# ── Modelo ───────────────────────────────────────────────
class MNISTClassifier(nn.Module):
    def __init__(self):
        super().__init__()
        self.net = nn.Sequential(
            nn.Flatten(),
            nn.Linear(28 * 28, 256),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(128, 10),
        )

    def forward(self, x):
        return self.net(x)

# ── Datos ────────────────────────────────────────────────
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.1307,), (0.3081,))
])
train_ds = datasets.MNIST("data", train=True, download=True, transform=transform)
train_loader = DataLoader(train_ds, batch_size=64, shuffle=True)

# ── Entrenamiento ────────────────────────────────────────
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = MNISTClassifier().to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.CrossEntropyLoss()

for epoch in range(5):
    model.train()
    total_loss = 0
    for batch_x, batch_y in train_loader:
        batch_x, batch_y = batch_x.to(device), batch_y.to(device)
        optimizer.zero_grad()
        loss = criterion(model(batch_x), batch_y)
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    print(f"Epoch {epoch+1}/5 — Loss: {total_loss/len(train_loader):.4f}")

# ── Guardar ──────────────────────────────────────────────
model.eval()
torch.save(model.state_dict(), "mnist_model.pt")
print("✓ Modelo guardado: mnist_model.pt")

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

# ── Datos ────────────────────────────────────────────────
(x_train, y_train), _ = keras.datasets.mnist.load_data()
x_train = x_train.astype("float32") / 255.0
x_train = x_train.reshape(-1, 28, 28, 1)

# ── Modelo ───────────────────────────────────────────────
model = keras.Sequential([
    layers.Input(shape=(28, 28, 1)),
    layers.Flatten(),
    layers.Dense(256, activation="relu"),
    layers.Dropout(0.2),
    layers.Dense(128, activation="relu"),
    layers.Dropout(0.2),
    layers.Dense(10, activation="softmax"),
])

model.compile(optimizer="adam",
              loss="sparse_categorical_crossentropy",
              metrics=["accuracy"])

model.fit(x_train, y_train, epochs=5, batch_size=64, verbose=1)

# ── Guardar ──────────────────────────────────────────────
model.save("mnist_model.h5")                    # Formato legacy HDF5
model.save("mnist_savedmodel")                  # Formato SavedModel (directorio)
print("✓ Modelos guardados: .h5 y SavedModel")

# Para exportar desde PyTorch + ejecutar ONNX
pip install onnx onnxruntime

# Para exportar desde TensorFlow (opcional)
pip install tf2onnx

import torch
import torch.nn as nn

# ── Redefinir la arquitectura (misma que en train_model.py) ──
class MNISTClassifier(nn.Module):
    def __init__(self):
        super().__init__()
        self.net = nn.Sequential(
            nn.Flatten(),
            nn.Linear(28 * 28, 256), nn.ReLU(), nn.Dropout(0.2),
            nn.Linear(256, 128),      nn.ReLU(), nn.Dropout(0.2),
            nn.Linear(128, 10),
        )
    def forward(self, x):
        return self.net(x)

# ── Cargar pesos ─────────────────────────────────────────
model = MNISTClassifier()
model.load_state_dict(torch.load("mnist_model.pt", weights_only=True))
model.eval()

# ── Exportar a ONNX ─────────────────────────────────────
dummy_input = torch.randn(1, 1, 28, 28)  # Batch de 1 imagen

torch.onnx.export(
    model,
    dummy_input,
    "mnist_model.onnx",
    input_names=["image"],
    output_names=["logits"],
    dynamic_axes={
        "image":  {0: "batch_size"},
        "logits": {0: "batch_size"},
    },
    opset_version=17,
)

print("✓ Exportado: mnist_model.onnx")

# Desde SavedModel
python -m tf2onnx.convert --saved-model mnist_savedmodel --output mnist_model.onnx --opset 17

# Desde .h5
python -m tf2onnx.convert --keras mnist_model.h5 --output mnist_model.onnx --opset 17

import onnx
import onnxruntime as ort
import numpy as np

# ── Verificar estructura ────────────────────────────────
onnx_model = onnx.load("mnist_model.onnx")
onnx.checker.check_model(onnx_model)
print("✓ Modelo ONNX válido")
print(f"  Inputs:  {[i.name for i in onnx_model.graph.input]}")
print(f"  Outputs: {[o.name for o in onnx_model.graph.output]}")

# ── Inferencia de prueba ─────────────────────────────────
session = ort.InferenceSession("mnist_model.onnx")

# Crear imagen de prueba (ruido aleatorio, 1×1×28×28)
test_input = np.random.randn(1, 1, 28, 28).astype(np.float32)
result = session.run(["logits"], {"image": test_input})

logits = result[0]
predicted_class = np.argmax(logits, axis=1)[0]
print(f"  Test inference → clase predicha: {predicted_class}")
print(f"  Shape output: {logits.shape}")

# ── Comparar tamaño ─────────────────────────────────────
import os
pt_size  = os.path.getsize("mnist_model.pt") / 1024
onnx_size = os.path.getsize("mnist_model.onnx") / 1024
print(f"\n  PyTorch:  {pt_size:.0f} KB")
print(f"  ONNX:     {onnx_size:.0f} KB")

ml-api/
├── app/
│   ├── __init__.py
│   ├── main.py          # FastAPI app + endpoints
│   ├── model.py         # Carga del modelo ONNX + inferencia
│   └── schemas.py       # Pydantic schemas (request/response)
├── models/
│   └── mnist_model.onnx # Modelo exportado
├── requirements.txt
├── Dockerfile
└── docker-compose.yml

fastapi==0.115.*
uvicorn[standard]==0.34.*
onnxruntime==1.20.*
numpy==2.1.*
Pillow==11.*
python-multipart==0.0.*

from pydantic import BaseModel, Field


class PredictionResponse(BaseModel):
    """Respuesta del endpoint /predict."""
    predicted_class: int = Field(..., ge=0, le=9, description="Dígito predicho (0-9)")
    confidence: float = Field(..., ge=0.0, le=1.0, description="Confianza de la predicción")
    probabilities: list[float] = Field(..., description="Probabilidades para cada clase (0-9)")


class HealthResponse(BaseModel):
    """Respuesta del endpoint /health."""
    status: str
    model_loaded: bool
    model_name: str

import numpy as np
import onnxruntime as ort
from PIL import Image
import io
import logging

logger = logging.getLogger(__name__)

# ── Sesión ONNX (singleton) ─────────────────────────────
_session: ort.InferenceSession | None = None
MODEL_PATH = "models/mnist_model.onnx"


def load_model() -> ort.InferenceSession:
    """Carga el modelo ONNX. Se llama una sola vez al inicio."""
    global _session
    if _session is None:
        logger.info(f"Cargando modelo desde {MODEL_PATH}...")
        opts = ort.SessionOptions()
        opts.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
        _session = ort.InferenceSession(MODEL_PATH, opts)
        logger.info("✓ Modelo cargado correctamente")
    return _session


def preprocess_image(image_bytes: bytes) -> np.ndarray:
    """Preprocesa una imagen para el modelo MNIST."""
    img = Image.open(io.BytesIO(image_bytes)).convert("L")  # Grayscale
    img = img.resize((28, 28))                                # Resize
    arr = np.array(img, dtype=np.float32)

    # Normalizar igual que durante el entrenamiento
    arr = (arr / 255.0 - 0.1307) / 0.3081

    # Reshape a (1, 1, 28, 28): batch, channels, height, width
    arr = arr.reshape(1, 1, 28, 28)
    return arr


def predict(image_bytes: bytes) -> dict:
    """Ejecuta inferencia sobre una imagen."""
    session = load_model()
    input_array = preprocess_image(image_bytes)

    # Ejecutar inferencia ONNX
    logits = session.run(["logits"], {"image": input_array})[0]

    # Softmax para obtener probabilidades
    exp = np.exp(logits - np.max(logits, axis=1, keepdims=True))
    probs = (exp / exp.sum(axis=1, keepdims=True))[0]

    predicted_class = int(np.argmax(probs))
    confidence = float(probs[predicted_class])

    return {
        "predicted_class": predicted_class,
        "confidence": round(confidence, 4),
        "probabilities": [round(float(p), 4) for p in probs],
    }

from contextlib import asynccontextmanager
from fastapi import FastAPI, File, UploadFile, HTTPException
import logging

from app.model import load_model, predict
from app.schemas import PredictionResponse, HealthResponse

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# ── Lifespan: cargar modelo al iniciar ───────────────────
@asynccontextmanager
async def lifespan(app: FastAPI):
    """Carga el modelo al arrancar, libera al cerrar."""
    load_model()
    logger.info("🚀 API lista para recibir requests")
    yield
    logger.info("👋 Cerrando API")


app = FastAPI(
    title="MNIST Classifier API",
    description="API de clasificación de dígitos con ONNX Runtime",
    version="1.0.0",
    lifespan=lifespan,
)


# ── Health check ─────────────────────────────────────────
@app.get("/health", response_model=HealthResponse)
def health():
    """Comprueba que la API y el modelo están operativos."""
    from app.model import _session
    return HealthResponse(
        status="ok",
        model_loaded=_session is not None,
        model_name="mnist_model.onnx",
    )


# ── Predicción ───────────────────────────────────────────
@app.post("/predict", response_model=PredictionResponse)
async def predict_endpoint(
    file: UploadFile = File(..., description="Imagen de un dígito (PNG/JPG)")
):
    """Recibe una imagen y devuelve el dígito predicho."""
    # Validar tipo de archivo
    if file.content_type not in ("image/png", "image/jpeg", "image/bmp", "image/webp"):
        raise HTTPException(
            status_code=400,
            detail=f"Tipo de archivo no soportado: {file.content_type}. Usa PNG, JPG, BMP o WebP."
        )

    # Leer y predecir
    image_bytes = await file.read()
    if len(image_bytes) == 0:
        raise HTTPException(status_code=400, detail="Archivo vacío")

    result = predict(image_bytes)
    return PredictionResponse(**result)

# Instalar todo
pip install -r requirements.txt

# Arrancar el servidor (modo desarrollo)
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# Health check
curl http://localhost:8000/health
# → {"status":"ok","model_loaded":true,"model_name":"mnist_model.onnx"}

# Predicción con una imagen
curl -X POST http://localhost:8000/predict \
  -F "file=@test_digit.png" \
  | python -m json.tool
# → {
#     "predicted_class": 7,
#     "confidence": 0.9834,
#     "probabilities": [0.0001, 0.0002, 0.0012, ...]
#   }

import requests

# ── Health check ─────────────────────────────────────────
r = requests.get("http://localhost:8000/health")
print(f"Health: {r.json()}")

# ── Predicción ───────────────────────────────────────────
with open("test_digit.png", "rb") as f:
    r = requests.post(
        "http://localhost:8000/predict",
        files={"file": ("digit.png", f, "image/png")}
    )

result = r.json()
print(f"Clase: {result['predicted_class']}")
print(f"Confianza: {result['confidence']:.2%}")

# ── Benchmark rápido ─────────────────────────────────────
import time
times = []
for _ in range(100):
    with open("test_digit.png", "rb") as f:
        t0 = time.perf_counter()
        requests.post("http://localhost:8000/predict",
                       files={"file": ("d.png", f, "image/png")})
        times.append(time.perf_counter() - t0)

print(f"\nLatencia media: {sum(times)/len(times)*1000:.1f} ms")
print(f"Latencia p95:   {sorted(times)[94]*1000:.1f} ms")
print(f"Throughput:     {len(times)/sum(times):.0f} req/s")

# ── Base image ───────────────────────────────────────────
FROM python:3.11-slim AS base

# ── Metadatos (buena práctica) ───────────────────────────
LABEL maintainer="tu-nombre"
LABEL description="MNIST Classifier API with ONNX Runtime"

# ── Variables de entorno ─────────────────────────────────
ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1 \
    PIP_NO_CACHE_DIR=1

# ── Dependencias del sistema ─────────────────────────────
RUN apt-get update && \
    apt-get install -y --no-install-recommends libgomp1 && \
    rm -rf /var/lib/apt/lists/*

# ── Directorio de trabajo ────────────────────────────────
WORKDIR /app

# ── Instalar dependencias Python ─────────────────────────
# Copiamos requirements.txt PRIMERO para cachear esta capa
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# ── Copiar código y modelo ───────────────────────────────
COPY app/ ./app/
COPY models/ ./models/

# ── Puerto ───────────────────────────────────────────────
EXPOSE 8000

# ── Crear usuario no-root (seguridad) ───────────────────
RUN adduser --disabled-password --gecos "" appuser
USER appuser

# ── Comando de arranque ──────────────────────────────────
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

__pycache__
*.pyc
.git
.gitignore
.env
*.md
data/
*.pt
*.pth
*.h5
venv/
.venv/
.pytest_cache/
.mypy_cache/

# Construir la imagen
docker build -t mnist-api:1.0 .

# Verificar tamaño
docker images mnist-api
# → REPOSITORY   TAG   IMAGE ID       SIZE
# → mnist-api    1.0   abc123def456   178MB

# Ejecutar (prueba rápida)
docker run --rm -p 8000:8000 mnist-api:1.0

# Probar desde otra terminal
curl http://localhost:8000/health

services:
  api:
    build:
      context: .
      dockerfile: Dockerfile
    image: mnist-api:1.0
    container_name: mnist-api
    ports:
      - "8000:8000"
    environment:
      - PYTHONUNBUFFERED=1
      - LOG_LEVEL=info
    volumes:
      - ./models:/app/models:ro
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 15s
    restart: unless-stopped
    deploy:
      resources:
        limits:
          memory: 512M
          cpus: "2.0"
        reservations:
          memory: 256M

# ── Configuración de la API ──────────────────────────────
API_PORT=8000
LOG_LEVEL=info
MODEL_PATH=models/mnist_model.onnx
WORKERS=1

    ports:
      - "${API_PORT:-8000}:8000"
    environment:
      - LOG_LEVEL=${LOG_LEVEL:-info}
      - MODEL_PATH=${MODEL_PATH:-models/mnist_model.onnx}

services:
  api:
    build: .
    ports:
      - "8000:8000"
    volumes:
      - ./models:/app/models:ro
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 15s
    restart: unless-stopped

  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
    depends_on:
      api:
        condition: service_healthy
    restart: unless-stopped

# Construir y levantar en foreground (para ver logs)
docker compose up --build

# O en background (detached)
docker compose up --build -d

# Ver los logs en tiempo real
docker compose logs -f api

# Ver el estado del servicio
docker compose ps

# ── 1. Health check ──────────────────────────────────────
curl -s http://localhost:8000/health | python -m json.tool
# {
#     "status": "ok",
#     "model_loaded": true,
#     "model_name": "mnist_model.onnx"
# }

# ── 2. Documentación interactiva ─────────────────────────
# Abre en el navegador: http://localhost:8000/docs

# ── 3. Predicción ────────────────────────────────────────
curl -s -X POST http://localhost:8000/predict \
  -F "file=@test_digit.png" | python -m json.tool
# {
#     "predicted_class": 7,
#     "confidence": 0.9834,
#     "probabilities": [0.0001, 0.0002, 0.0012, ...]
# }

# ── 4. Verificar health check de Docker ──────────────────
docker inspect --format='{{.State.Health.Status}}' mnist-api
# → healthy

# ── 5. Ver recursos usados ───────────────────────────────
docker stats mnist-api --no-stream
# CONTAINER   CPU %   MEM USAGE / LIMIT   MEM %
# mnist-api   0.50%   89MiB / 512MiB      17.38%

"""Test end-to-end de la API dockerizada."""
import requests
import numpy as np
from PIL import Image
import io

API_URL = "http://localhost:8000"

# ── Test 1: Health ───────────────────────────────────────
r = requests.get(f"{API_URL}/health")
assert r.status_code == 200
data = r.json()
assert data["status"] == "ok"
assert data["model_loaded"] is True
print("✓ Health check OK")

# ── Test 2: Predicción con imagen real ───────────────────
with open("test_digit.png", "rb") as f:
    r = requests.post(f"{API_URL}/predict", files={"file": ("d.png", f, "image/png")})
assert r.status_code == 200
result = r.json()
assert 0 <= result["predicted_class"] <= 9
assert 0 <= result["confidence"] <= 1
assert len(result["probabilities"]) == 10
print(f"✓ Predicción OK: clase={result['predicted_class']}, "
      f"confianza={result['confidence']:.2%}")

# ── Test 3: Predicción con imagen generada ───────────────
# Crear un "7" sintético
img = Image.new("L", (28, 28), 0)
pixels = np.array(img)
pixels[6:22, 10:20] = 255   # Línea horizontal superior
pixels[6:22, 16:20] = 255   # Línea vertical derecha
img = Image.fromarray(pixels)
buf = io.BytesIO()
img.save(buf, format="PNG")
buf.seek(0)

r = requests.post(f"{API_URL}/predict", files={"file": ("synth.png", buf, "image/png")})
assert r.status_code == 200
print(f"✓ Predicción sintética OK: clase={r.json()['predicted_class']}")

# ── Test 4: Error handling ───────────────────────────────
r = requests.post(f"{API_URL}/predict",
                  files={"file": ("test.txt", b"not an image", "text/plain")})
assert r.status_code == 400
print(f"✓ Error handling OK: {r.json()['detail']}")

print("\n🎉 Todos los tests pasaron")

# Ver logs
docker compose logs -f api

# Entrar al contenedor (debugging)
docker compose exec api bash

# Reiniciar el servicio
docker compose restart api

# Parar todo
docker compose down

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

# Reconstruir sin cache (si algo falla)
docker compose build --no-cache

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://tu-frontend.com"],  # En producción: lista explícita
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
    allow_credentials=False,
)

import logging
import time
import uuid
from fastapi import Request

logger = logging.getLogger("mnist-api")

@app.middleware("http")
async def log_requests(request: Request, call_next):
    """Middleware que logea cada request con timing y request ID."""
    request_id = str(uuid.uuid4())[:8]
    start = time.perf_counter()

    response = await call_next(request)

    duration_ms = (time.perf_counter() - start) * 1000
    logger.info(
        f"[{request_id}] {request.method} {request.url.path} "
        f"→ {response.status_code} ({duration_ms:.1f}ms)"
    )

    response.headers["X-Request-ID"] = request_id
    return response

# pip install slowapi
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("/predict")
@limiter.limit("30/minute")  # Máximo 30 predicciones por minuto por IP
async def predict_endpoint(request: Request, file: UploadFile = File(...)):
    ...

from fastapi import File, UploadFile
from typing import List

@app.post("/predict/batch", response_model=list[PredictionResponse])
async def predict_batch(
    files: list[UploadFile] = File(..., description="Múltiples imágenes")
):
    """Procesa múltiples imágenes en un solo request."""
    if len(files) > 32:
        raise HTTPException(400, "Máximo 32 imágenes por batch")

    results = []
    for file in files:
        image_bytes = await file.read()
        result = predict(image_bytes)
        results.append(PredictionResponse(**result))

    return results

# pip install prometheus-fastapi-instrumentator
from prometheus_fastapi_instrumentator import Instrumentator

# Después de crear app:
Instrumentator().instrument(app).expose(app, endpoint="/metrics")