Proyecto: API LLM escalable production-ready

Llegaste al final del módulo. Vas a integrar todo lo aprendido — imagen con vLLM, GPU A10G, volume persistente, FastAPI, autoscaling configurado, autenticación, monitoring básico — en un entregable que podrías llevar a tu portfolio o usar en un proyecto real.

Este es el deliverable que el Módulo 8 (Unified Client) va a usar como uno de los providers detrás del cliente unificado. Después de este proyecto, ya tienes un endpoint LLM tuyo, en infraestructura tuya, listo para producción.

Al terminar este proyecto vas a tener:

• Un endpoint HTTPS público /chat con auth por token
• Healthcheck público /health con estado del modelo
• Métricas observables vía /metrics y logs estructurados
• Autoscaling configurado para latencia P50 razonable y costo acotado
• Tests que validan contrato y autenticación
• README con setup, costos esperados, y cómo usarlo

---

Especificación funcional

Endpoints

Método	Path	Auth	Descripción
GET	/health	No	Estado del servicio: modelo cargado, versión, uptime
GET	/metrics	Bearer admin	Contadores: total requests, errores, latencia P50/P95
POST	/chat	Bearer user	Genera respuesta del LLM para un prompt
POST	/chat-batch	Bearer user	Procesa hasta 10 prompts en una llamada (batching)

Contratos

POST /chat

// Request
{
  "prompt": "Explica REST en 2 frases.",
  "max_tokens": 256,
  "temperature": 0.7,
  "request_id": "uuid-opcional-del-cliente"
}

// Response 200
{
  "respuesta": "REST es un estilo...",
  "modelo": "mistralai/Mistral-7B-Instruct-v0.3",
  "tokens_generados": 78,
  "duracion_ms": 1820,
  "request_id": "uuid-opcional-del-cliente"
}

// Response 401 / 403 / 422
{
  "detail": "Mensaje específico del error"
}

No-funcionales

• Healthcheck disponible incluso si el container GPU está apagado
• Logs estructurados en JSON (request_id, latencia, modelo, status)
• P50 < 3s, P99 < 30s bajo carga normal (con warm pool activo)
• Tope de costo: máximo 5 containers paralelos

---

Implementación completa

Crea proyecto_final.py:

# proyecto_final.py
import os
import json
import time
import uuid
import modal
from collections import deque
from typing import Annotated
from fastapi import FastAPI, HTTPException, Header
from pydantic import BaseModel, Field

# ============================================================
# Configuración
# ============================================================
MODELO = "mistralai/Mistral-7B-Instruct-v0.3"
CACHE_DIR = "/cache/huggingface"
VERSION_APP = "1.0.0"

app = modal.App("llm-api-final")
pesos_volume = modal.Volume.from_name("mistral-pesos", create_if_missing=True)

imagen = (
    modal.Image.debian_slim(python_version="3.11")
    .pip_install(
        "vllm==0.6.3",
        "huggingface_hub[hf_transfer]==0.26.2",
        "fastapi[standard]",
    )
    .env({"HF_HOME": CACHE_DIR, "HF_HUB_ENABLE_HF_TRANSFER": "1"})
)

# ============================================================
# Contratos Pydantic
# ============================================================
class ChatRequest(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=4000)
    max_tokens: int = Field(256, ge=1, le=2048)
    temperature: float = Field(0.7, ge=0.0, le=2.0)
    request_id: str | None = None


class ChatResponse(BaseModel):
    respuesta: str
    modelo: str
    tokens_generados: int
    duracion_ms: int
    request_id: str


class BatchRequest(BaseModel):
    prompts: list[str] = Field(..., min_length=1, max_length=10)
    max_tokens: int = Field(256, ge=1, le=2048)
    temperature: float = Field(0.7, ge=0.0, le=2.0)


class BatchResponse(BaseModel):
    respuestas: list[str]
    modelo: str
    duracion_ms: int


class HealthResponse(BaseModel):
    status: str
    modelo: str
    version: str
    uptime_seconds: int


class MetricsResponse(BaseModel):
    total_requests: int
    total_errores: int
    latencia_p50_ms: int
    latencia_p95_ms: int
    ultima_actualizacion: str


# ============================================================
# Estado en proceso (métricas simples)
# ============================================================
INICIO_PROCESO = time.time()
metricas = {
    "total_requests": 0,
    "total_errores": 0,
    "latencias": deque(maxlen=1000),
}


def registrar_request(duracion_ms: int, error: bool = False):
    metricas["total_requests"] += 1
    if error:
        metricas["total_errores"] += 1
    else:
        metricas["latencias"].append(duracion_ms)


def percentil(valores, p):
    if not valores:
        return 0
    ordenados = sorted(valores)
    idx = int(len(ordenados) * p / 100)
    return ordenados[min(idx, len(ordenados) - 1)]


# ============================================================
# Autenticación
# ============================================================
def verificar_token(authorization: str | None, tipo: str = "user"):
    var_env = "API_TOKEN_USER" if tipo == "user" else "API_TOKEN_ADMIN"
    token_esperado = os.environ.get(var_env)

    if not token_esperado:
        raise HTTPException(500, f"{var_env} no configurado en el servidor")
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(401, "Falta header 'Authorization: Bearer <token>'")
    if authorization.removeprefix("Bearer ") != token_esperado:
        raise HTTPException(403, "Token inválido")


def log_estructurado(evento: str, **kwargs):
    print(json.dumps({"evento": evento, "ts": time.time(), **kwargs}))


# ============================================================
# Servicio LLM con GPU
# ============================================================
@app.cls(
    image=imagen,
    gpu="A10G",
    volumes={CACHE_DIR: pesos_volume},
    timeout=600,
    min_containers=0,            # No pagar GPU 24/7 por defecto
    max_containers=5,            # Acotar costo
    scaledown_window=300,        # 5min de gracia
)
class MistralService:
    @modal.enter()
    def cargar_modelo(self):
        from vllm import LLM
        log_estructurado("modelo_cargando", modelo=MODELO)
        inicio = time.time()
        self.llm = LLM(model=MODELO, download_dir=CACHE_DIR)
        log_estructurado(
            "modelo_listo",
            duracion_s=round(time.time() - inicio, 2),
        )

    @modal.method()
    def generar(self, prompt: str, max_tokens: int, temperature: float) -> dict:
        from vllm import SamplingParams
        sampling = SamplingParams(temperature=temperature, max_tokens=max_tokens)
        inicio = time.time()
        prompt_fmt = f"[INST] {prompt} [/INST]"
        output = self.llm.generate(prompt_fmt, sampling)[0]
        return {
            "texto": output.outputs[0].text.strip(),
            "tokens": len(output.outputs[0].token_ids),
            "duracion_ms": int((time.time() - inicio) * 1000),
        }

    @modal.method()
    def generar_batch(
        self, prompts: list[str], max_tokens: int, temperature: float
    ) -> dict:
        from vllm import SamplingParams
        sampling = SamplingParams(temperature=temperature, max_tokens=max_tokens)
        inicio = time.time()
        prompts_fmt = [f"[INST] {p} [/INST]" for p in prompts]
        outputs = self.llm.generate(prompts_fmt, sampling)
        return {
            "respuestas": [o.outputs[0].text.strip() for o in outputs],
            "duracion_ms": int((time.time() - inicio) * 1000),
        }


# ============================================================
# Función web (FastAPI ASGI)
# ============================================================
@app.function(
    image=imagen,
    secrets=[modal.Secret.from_name("llm-api-tokens")],
    min_containers=1,           # Web tier siempre warm (es barato, CPU only)
    max_containers=3,
)
@modal.asgi_app()
def web():
    web_app = FastAPI(
        title="LLM API",
        version=VERSION_APP,
        description="Endpoint Mistral 7B sobre Modal con autoscaling y monitoring",
    )

    @web_app.get("/health", response_model=HealthResponse)
    def health():
        return HealthResponse(
            status="ok",
            modelo=MODELO,
            version=VERSION_APP,
            uptime_seconds=int(time.time() - INICIO_PROCESO),
        )

    @web_app.get("/metrics", response_model=MetricsResponse)
    def metrics(authorization: Annotated[str | None, Header()] = None):
        verificar_token(authorization, tipo="admin")
        latencias = list(metricas["latencias"])
        return MetricsResponse(
            total_requests=metricas["total_requests"],
            total_errores=metricas["total_errores"],
            latencia_p50_ms=percentil(latencias, 50),
            latencia_p95_ms=percentil(latencias, 95),
            ultima_actualizacion=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        )

    @web_app.post("/chat", response_model=ChatResponse)
    def chat(
        req: ChatRequest,
        authorization: Annotated[str | None, Header()] = None,
    ):
        verificar_token(authorization, tipo="user")
        rid = req.request_id or str(uuid.uuid4())
        inicio = time.time()
        try:
            servicio = MistralService()
            resultado = servicio.generar.remote(
                req.prompt, req.max_tokens, req.temperature
            )
            duracion_total = int((time.time() - inicio) * 1000)
            registrar_request(duracion_total)
            log_estructurado(
                "chat_ok",
                request_id=rid,
                duracion_ms=duracion_total,
                tokens=resultado["tokens"],
            )
            return ChatResponse(
                respuesta=resultado["texto"],
                modelo=MODELO,
                tokens_generados=resultado["tokens"],
                duracion_ms=duracion_total,
                request_id=rid,
            )
        except Exception as e:
            registrar_request(0, error=True)
            log_estructurado("chat_error", request_id=rid, error=str(e))
            raise HTTPException(500, f"Error generando respuesta: {e}")

    @web_app.post("/chat-batch", response_model=BatchResponse)
    def chat_batch(
        req: BatchRequest,
        authorization: Annotated[str | None, Header()] = None,
    ):
        verificar_token(authorization, tipo="user")
        inicio = time.time()
        servicio = MistralService()
        resultado = servicio.generar_batch.remote(
            req.prompts, req.max_tokens, req.temperature
        )
        duracion_total = int((time.time() - inicio) * 1000)
        registrar_request(duracion_total)
        return BatchResponse(
            respuestas=resultado["respuestas"],
            modelo=MODELO,
            duracion_ms=duracion_total,
        )

    return web_app


# ============================================================
# Keep-warm en horario laboral (opcional)
# ============================================================
@app.function(schedule=modal.Period(minutes=5))
def keep_warm_laboral():
    from datetime import datetime
    hora_utc = datetime.utcnow().hour
    # Ajusta a tu timezone — ejemplo: 14:00-22:00 UTC ≈ 9am-5pm CDMX
    if 14 <= hora_utc < 22:
        servicio = MistralService()
        servicio.generar.remote("ping", max_tokens=1, temperature=0.0)
        log_estructurado("keep_warm", hora_utc=hora_utc)

---

Setup paso a paso

1 — Crear los secrets

# Tokens (genera valores aleatorios)
TOKEN_USER=$(python -c "import secrets; print(secrets.token_urlsafe(32))")
TOKEN_ADMIN=$(python -c "import secrets; print(secrets.token_urlsafe(32))")
echo "USER: $TOKEN_USER"
echo "ADMIN: $TOKEN_ADMIN"

modal secret create llm-api-tokens \
  API_TOKEN_USER=$TOKEN_USER \
  API_TOKEN_ADMIN=$TOKEN_ADMIN

Guarda los dos tokens en tu password manager. Sin ellos no puedes llamar el API.

2 — Deployar

modal deploy proyecto_final.py

Output esperado:

✓ Created mount /Users/.../proyecto_final.py
✓ Created function MistralService.*
✓ Created function keep_warm_laboral
✓ Created web function web => https://tu-usuario--llm-api-final-web.modal.run

✓ App deployed in 12.4s

3 — Probar endpoints

BASE=https://tu-usuario--llm-api-final-web.modal.run

# Health (sin auth)
curl $BASE/health

# Chat con auth
curl -X POST $BASE/chat \
  -H "Authorization: Bearer $TOKEN_USER" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Explica Modal en 2 frases.", "max_tokens": 150}'

# Metrics (admin)
curl $BASE/metrics -H "Authorization: Bearer $TOKEN_ADMIN"

# Batch
curl -X POST $BASE/chat-batch \
  -H "Authorization: Bearer $TOKEN_USER" \
  -H "Content-Type: application/json" \
  -d '{"prompts": ["¿Qué es FastAPI?", "¿Qué es REST?"], "max_tokens": 100}'

---

Tests automatizados

Crea test_api.py localmente (no en Modal):

# test_api.py
import os
import pytest
import httpx

BASE = os.environ["LLM_API_BASE"]                # exporta antes de correr
TOKEN_USER = os.environ["LLM_API_TOKEN_USER"]
TOKEN_ADMIN = os.environ["LLM_API_TOKEN_ADMIN"]

cliente = httpx.Client(base_url=BASE, timeout=60)


def test_health_ok():
    r = cliente.get("/health")
    assert r.status_code == 200
    data = r.json()
    assert data["status"] == "ok"
    assert data["modelo"]


def test_chat_sin_token_falla():
    r = cliente.post("/chat", json={"prompt": "hola"})
    assert r.status_code == 401


def test_chat_token_invalido_falla():
    r = cliente.post(
        "/chat",
        headers={"Authorization": "Bearer invalid"},
        json={"prompt": "hola"},
    )
    assert r.status_code == 403


def test_chat_prompt_vacio_falla():
    r = cliente.post(
        "/chat",
        headers={"Authorization": f"Bearer {TOKEN_USER}"},
        json={"prompt": ""},
    )
    assert r.status_code == 422  # Pydantic rejection


def test_chat_ok():
    r = cliente.post(
        "/chat",
        headers={"Authorization": f"Bearer {TOKEN_USER}"},
        json={"prompt": "Di solo 'hola'", "max_tokens": 10},
    )
    assert r.status_code == 200
    data = r.json()
    assert data["respuesta"]
    assert data["tokens_generados"] > 0
    assert data["request_id"]


def test_metrics_requiere_admin():
    r_user = cliente.get("/metrics", headers={"Authorization": f"Bearer {TOKEN_USER}"})
    assert r_user.status_code == 403

    r_admin = cliente.get("/metrics", headers={"Authorization": f"Bearer {TOKEN_ADMIN}"})
    assert r_admin.status_code == 200
    assert "total_requests" in r_admin.json()


def test_batch_acepta_lista():
    r = cliente.post(
        "/chat-batch",
        headers={"Authorization": f"Bearer {TOKEN_USER}"},
        json={"prompts": ["Hola", "Adiós"], "max_tokens": 10},
    )
    assert r.status_code == 200
    assert len(r.json()["respuestas"]) == 2

Correr:

export LLM_API_BASE=https://tu-usuario--llm-api-final-web.modal.run
export LLM_API_TOKEN_USER=<el-token-user>
export LLM_API_TOKEN_ADMIN=<el-token-admin>
pip install pytest httpx
pytest test_api.py -v

---

README sugerido

Crea README.md en el repo donde guardes este proyecto:

# LLM API — Mistral 7B sobre Modal

Endpoint HTTPS production-ready para Mistral 7B Instruct, deployado en Modal con autoscaling, autenticación y monitoring.

## Endpoints

- `GET /health` — público
- `GET /metrics` — Bearer admin
- `POST /chat` — Bearer user
- `POST /chat-batch` — Bearer user, hasta 10 prompts

## Setup

```bash
pip install modal
modal token new
modal secret create llm-api-tokens API_TOKEN_USER=... API_TOKEN_ADMIN=...
modal deploy proyecto_final.py
```

## Costo aproximado

- Sin tráfico: ~$5/mes (CPU container web warm + storage)
- 100,000 req/mes: ~$60/mes
- 1,000,000 req/mes: ~$600/mes

Verifica precios actuales en [modal.com/pricing](https://modal.com/pricing).

## Configuración de escala (defaults)

| Parámetro | Valor | Por qué |
|-----------|-------|---------|
| `min_containers` (GPU) | 0 | No pagar GPU sin tráfico |
| `max_containers` (GPU) | 5 | Tope de costo |
| `scaledown_window` | 300s | Burst-friendly |
| `min_containers` (web) | 1 | Healthcheck siempre disponible |

Ajusta a tu caso real.

---

Auto-evaluación

Antes de declarar el módulo completo, verifica:

• El endpoint /health responde sin auth y muestra status: ok
• Llamadas a /chat sin token retornan 401, con token inválido 403, con token válido 200
• La primera llamada (cold start) tarda <90s, las siguientes <5s
• modal app logs muestra eventos JSON estructurados (chat_ok, chat_error)
• /metrics con token admin muestra contadores reales después de varias llamadas a /chat
• modal app stop llm-api-final apaga el deployment limpiamente
• Los tests pytest pasan

Si todos los checks pasan, tienes un endpoint production-ready.

---

Conexión con el resto del path

Este endpoint es un provider para el Unified AI Client del Módulo 8. La interfaz que diseñaste acá (/chat con request/response JSON) es compatible con:

# En el Módulo 8 vas a escribir algo así:
class ModalClient(BaseAIClient):
    def chat(self, prompt: str) -> str:
        r = httpx.post(
            f"{self.base_url}/chat",
            headers={"Authorization": f"Bearer {self.token}"},
            json={"prompt": prompt, "max_tokens": 256},
        )
        r.raise_for_status()
        return r.json()["respuesta"]

Vas a poder hacer client = UnifiedAIClient(provider="modal") y consumirlo igual que OpenAI o Ollama.

---

Módulo 6 completado

Recapitulemos: ahora puedes...

• ✅ Decidir cuándo Modal vence a managed o self-hosted
• ✅ Setupear cuenta, CLI y autenticación
• ✅ Definir imágenes con dependencias y secretos
• ✅ Deployar un LLM (Mistral 7B) con GPU A10G y caché de pesos
• ✅ Exponerlo como API HTTPS pública con FastAPI + auth
• ✅ Configurar autoscaling y cold start mitigation
• ✅ Estimar costos y aplicar optimizaciones
• ✅ Entregar un servicio production-ready con healthcheck, métricas y tests

Bien hecho. Modal es de los temas que casi nadie cubre y vas a verlo aparecer en startups AI reales.

---

Siguiente módulo

Módulo 7 — Trade-offs y Decision Matrix toma todos los proveedores que aprendiste en módulos 2-6 (OpenAI, LM Studio, Ollama, OpenRouter, Modal) y los compara con benchmarks reales que tú mismo vas a correr. Es el módulo que te da el criterio cuantitativo para elegir, no solo cualitativo.

---

Recursos

1. Repo ejemplo: modal-labs/llm-serving — ejemplos oficiales de patrones avanzados.
2. Modal — Production checklist — recomendaciones oficiales para production.
3. FastAPI — Testing — patrones de testing.
4. Prometheus client for Python — si quieres extender /metrics a formato Prometheus estándar.
5. Sentry SDK — error tracking para producción real.