Desarrollo de backend de ML con FastAPI

Descubre cómo crear APIs de ML de alto rendimiento con FastAPI.

Configuración básica

1. Instalación

pip install fastapi uvicorn pydantic numpy pandas scikit-learn

2. Estructura del proyecto

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uvicorn

app = FastAPI(
    title="API ML",
    description="API pour prédictions ML",
    version="1.0.0"
)

# Modelos de datos
class PredictionRequest(BaseModel):
    features: List[float]
    model_type: Optional[str] = "default"

class PredictionResponse(BaseModel):
    prediction: float
    confidence: float
    model_version: str

Endpoints de la API

1. Rutas básicas

@app.get("/")
async def root():
    return {"message": "API ML active"}

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

@app.get("/models")
async def list_models():
    return {
        "available_models": [
            "classification",
            "regression",
            "clustering"
        ]
    }

2. Endpoints de predicción

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    try:
        # Carga del modelo
        model = load_model(request.model_type)
        
        # Predicción
        prediction = model.predict([request.features])[0]
        confidence = model.predict_proba([request.features])[0]
        
        return PredictionResponse(
            prediction=float(prediction),
            confidence=float(max(confidence)),
            model_version="1.0.0"
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Gestión de modelos

1. Carga y gestión

from typing import Dict
import joblib
import os

class ModelManager:
    def __init__(self):
        self.models: Dict[str, object] = {}
        self.model_path = "models/"
        
    def load_model(self, model_type: str) -> object:
        if model_type not in self.models:
            path = os.path.join(self.model_path, f"{model_type}.pkl")
            if not os.path.exists(path):
                raise HTTPException(
                    status_code=404,
                    detail=f"Modèle {model_type} non trouvé"
                )
            self.models[model_type] = joblib.load(path)
        return self.models[model_type]

model_manager = ModelManager()

2. Validación de datos

from pydantic import validator
import numpy as np

class FeatureValidation(BaseModel):
    features: List[float]
    
    @validator('features')
    def validate_features(cls, v):
        if not v:
            raise ValueError("La liste des features ne peut pas être vide")
        if not all(isinstance(x, (int, float)) for x in v):
            raise ValueError("Toutes les features doivent être numériques")
        return v

Middleware y seguridad

1. Middleware personalizado

from fastapi import Request
import time

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

2. Autenticación

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_token(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token invalide"
        )

Pruebas y documentación

1. Pruebas unitarias

# test_api.py
from fastapi.testclient import TestClient
import pytest

client = TestClient(app)

def test_predict():
    response = client.post(
        "/predict",
        json={"features": [1.0, 2.0, 3.0]}
    )
    assert response.status_code == 200
    assert "prediction" in response.json()

2. Documentación de la API

# Añadir documentación personalizada
@app.get("/docs")
async def custom_docs():
    return {
        "endpoints": {
            "/predict": {
                "description": "Endpoint de prédiction",
                "parameters": {
                    "features": "Liste de features numériques",
                    "model_type": "Type de modèle à utiliser"
                }
            }
        }
    }

Buenas prácticas

1. Rendimiento

   • Caché de modelos
   • Validación eficiente
   • Gestión de la memoria
   • Monitorización

2. Seguridad

   • Autenticación
   • Validación de las entradas
   • Rate limiting
   • Logging

3. Mantenimiento

   • Pruebas automatizadas
   • Documentación
   • Versionado
   • Monitorización

Conclusión

Puntos clave que hay que recordar:

• API de alto rendimiento
• Seguridad reforzada
• Documentación completa
• Pruebas automatizadas

Recomendaciones:

• Seguir los estándares REST
• Implementar la seguridad
• Mantener la documentación
• Automatizar las pruebas
• Formarse en FastAPI