Desarrollo de la API con Validación y Monitoreo

Concepto clave

En producción, una API de Machine Learning no es solo un modelo envuelto en código. Es un sistema que debe ser robusto, validado y monitoreado. Imagina que eres un chef en un restaurante de alta cocina: no basta con tener una receta excelente (tu modelo). Necesitas verificar que los ingredientes sean frescos (validación de datos), asegurar que cada plato salga perfecto (monitoreo de predicciones) y tener alertas si algo falla en la cocina (logging y métricas). FastAPI facilita esto con su sistema de validación integrado usando Pydantic y su compatibilidad con herramientas de observabilidad.

La validación garantiza que los datos de entrada cumplan con las expectativas del modelo, evitando errores como valores nulos o formatos incorrectos. El monitoreo, por otro lado, te permite entender cómo se comporta tu API en tiempo real: cuántas solicitudes recibe, cuál es el tiempo de respuesta, y si las predicciones son coherentes. Sin estos elementos, desplegar un modelo en producción es como conducir un coche sin instrumentos: no sabes si vas demasiado rápido o si se está calentando el motor.

Cómo funciona en la práctica

Vamos a construir una API para un modelo de clasificación de texto (por ejemplo, detección de spam). Primero, definimos un esquema de validación con Pydantic para la entrada: el texto debe ser una cadena no vacía con longitud máxima. Luego, integramos el modelo (usaremos un ejemplo simple con scikit-learn) y agregamos endpoints para predicción y salud. Finalmente, configuramos logging básico y métricas con Prometheus (usando librerías como prometheus-fastapi-instrumentator).

Paso a paso: 1) Crear un entorno virtual e instalar dependencias (FastAPI, uvicorn, scikit-learn, pydantic, prometheus-fastapi-instrumentator). 2) Definir el esquema de datos en un archivo schemas.py. 3) Implementar el modelo y cargarlo (o entrenarlo en memoria para el ejemplo). 4) Construir la API en main.py con endpoints y validación. 5) Agregar middleware para logging y métricas. 6) Probar con herramientas como curl o Postman.

Código en acción

Aquí tienes un ejemplo funcional de una API con validación y monitoreo básico. Nota: Este código es simplificado para propósitos educativos; en un entorno real, podrías usar un modelo preentrenado y configuraciones más avanzadas.

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import logging
from prometheus_fastapi_instrumentator import Instrumentator

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

# Esquema de validación
class TextRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=1000, description="Texto a clasificar")

# Modelo de ejemplo (clasificador simple)
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
import numpy as np

# Entrenar un modelo minimalista (en producción, cargarías uno preentrenado)
vectorizer = TfidfVectorizer()
model = LogisticRegression()
# Datos de ejemplo
X_train = ["oferta gratis", "hola amigo", "gana dinero rápido", "reunión mañana"]
y_train = [1, 0, 1, 0]  # 1=spam, 0=no spam
X_train_vec = vectorizer.fit_transform(X_train)
model.fit(X_train_vec, y_train)

def predict_spam(text: str) -> dict:
    """Función de predicción"""
    try:
        X_vec = vectorizer.transform([text])
        prediction = model.predict(X_vec)[0]
        probability = model.predict_proba(X_vec)[0].max()
        return {"prediction": "spam" if prediction == 1 else "no spam", "confidence": float(probability)}
    except Exception as e:
        logger.error(f"Error en predicción: {e}")
        raise HTTPException(status_code=500, detail="Error interno del modelo")

app = FastAPI(title="API de Clasificación de Spam")

# Instrumentar para métricas de Prometheus
Instrumentator().instrument(app).expose(app)

@app.post("/predict")
async def predict(request: TextRequest):
    """Endpoint para predicción con validación automática"""
    logger.info(f"Solicitud recibida: {request.text[:50]}...")
    result = predict_spam(request.text)
    logger.info(f"Predicción: {result}")
    return result

@app.get("/health")
async def health_check():
    """Endpoint de salud para monitoreo"""
    return {"status": "healthy"}

Antes de agregar validación y monitoreo, el código podría verse así (menos robusto):

# Antes: Sin validación ni monitoreo
from fastapi import FastAPI

app = FastAPI()

@app.post("/predict")
async def predict(text: str):
    # Sin validación, podría fallar con texto vacío o muy largo
    return {"prediction": "spam"}

Errores comunes

• No validar datos de entrada: Aceptar cualquier entrada puede causar errores en el modelo o problemas de seguridad. Solución: Usa Pydantic para definir esquemas estrictos, como en el ejemplo con TextRequest.
• Ignorar el logging: Sin logs, es difícil depurar fallos en producción. Solución: Configura logging básico (como logging.basicConfig) y registra eventos clave, como solicitudes y errores.
• No exponer métricas: Sin métricas, no puedes medir el rendimiento o detectar anomalías. Solución: Integra herramientas como Prometheus con prometheus-fastapi-instrumentator para rastrear solicitudes, tiempos de respuesta, etc.
• Manejo pobre de errores: Devolver errores genéricos al cliente puede ser poco útil. Solución: Usa HTTPException con detalles específicos y códigos de estado apropiados (por ejemplo, 400 para entradas inválidas).
• Olvidar endpoints de salud: Los sistemas de orquestación (como Kubernetes) necesitan verificar que la API esté viva. Solución: Incluye un endpoint /health simple que devuelva el estado del servicio.

Checklist de dominio

1. He definido esquemas de validación con Pydantic para todas las entradas de la API.
2. He implementado logging para registrar solicitudes, predicciones y errores críticos.
3. He integrado métricas (por ejemplo, con Prometheus) para monitorear el rendimiento en tiempo real.
4. He añadido endpoints de salud (/health) para verificaciones de vivacidad.
5. He probado la API con casos límite (por ejemplo, texto vacío, entradas muy largas) para asegurar robustez.
6. He documentado los endpoints (FastAPI lo hace automáticamente, pero verifica que sea claro).
7. He configurado manejo de errores con respuestas HTTP apropiadas (por ejemplo, 400, 500).

Extiende la API con Validación Avanzada y Monitoreo Personalizado

En este ejercicio, mejorarás la API de ejemplo agregando validación avanzada y monitoreo personalizado. Sigue estos pasos:

1. Clona el entorno: Crea un nuevo directorio y copia el código de main.py del ejemplo anterior. Instala las dependencias necesarias: fastapi, uvicorn, scikit-learn, pydantic, prometheus-fastapi-instrumentator.
2. Añade validación avanzada: Modifica el esquema TextRequest para que incluya un campo language (cadena) que solo permita valores "es" (español) o "en" (inglés). Usa el validador de Pydantic (@validator) para verificar que el texto no contenga palabras prohibidas (por ejemplo, define una lista como ["spam", "fraude"] y rechaza la solicitud si se encuentra alguna).
3. Implementa monitoreo personalizado: Crea una métrica personalizada con Prometheus para contar el número de predicciones clasificadas como spam. Usa prometheus_client (instálalo si es necesario) y registra un contador que se incremente en el endpoint /predict cuando la predicción sea "spam".
4. Prueba la API: Ejecuta la API con uvicorn main:app --reload y usa curl o Postman para enviar solicitudes. Verifica que la validación funcione (por ejemplo, envía texto con palabras prohibidas y debe devolver error 400). Accede a /metrics para ver las métricas de Prometheus, incluyendo tu contador personalizado.
5. Documenta los cambios: Actualiza los comentarios en el código para reflejar las nuevas funcionalidades y ejecuta una prueba final con al menos 3 casos diferentes.

• Recuerda importar validator de pydantic para la validación personalizada.
• Para la métrica personalizada, usa Counter de prometheus_client y asegúrate de exponerla en el endpoint /metrics (FastAPI instrumentator lo hace automáticamente).
• Prueba casos como texto vacío, lenguaje inválido, y palabras prohibidas para cubrir todos los escenarios de validación.

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.