FastAPI para servir modelos: del notebook a un endpoint que aguanta

El notebook miente

Un model.predict(x) en una celda de Jupyter da una sensación de “ya está”. No lo está.

En el notebook controlas todo: una petición cada vez, los datos limpios, la GPU para ti solo y nadie esperando al otro lado. Producción es lo contrario. Llegan peticiones simultáneas, con datos raros, y alguien mira el reloj. El salto del notebook al endpoint no es empaquetar el modelo, es construir lo que rodea al modelo para que no se caiga al primer pico de tráfico.

FastAPI es una buena base para ese salto. No porque sea moderno, sino porque resuelve tres cosas que importan al servir modelos: validación de entrada, concurrencia y arranque. Vamos por partes.

Carga el modelo una vez, no en cada petición

El error más caro es cargar el modelo dentro de la función del endpoint. Cada petición vuelve a leer pesos de disco, y un modelo de medio gigabyte tardando un segundo en cargar convierte tu API en una tortuga bajo carga.

El modelo se carga al arrancar el proceso y vive en memoria. En FastAPI, el sitio para eso es el lifespan:

from contextlib import asynccontextmanager
from fastapi import FastAPI
import joblib

ml = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Se ejecuta una vez al arrancar el worker
    ml["model"] = joblib.load("model.joblib")
    yield
    # Se ejecuta al apagar: liberar recursos
    ml.clear()

app = FastAPI(lifespan=lifespan)

Así el modelo se carga una vez por worker, no una vez por petición. La diferencia entre 200 ms y 5 ms por llamada suele estar aquí.

Valida la entrada antes de tocar el modelo

Un modelo es exigente con su entrada: espera el número de features correcto, en el orden correcto y del tipo correcto. Si dejas que un JSON mal formado llegue hasta predict, el error explota dentro del modelo, con un traceback ilegible y un 500 genérico para quien llama.

Pydantic mueve esa validación al borde de tu API. Defines el contrato una vez y FastAPI rechaza lo que no cumple con un 422 claro, antes de gastar un solo ciclo de inferencia:

from pydantic import BaseModel, Field

class PredictRequest(BaseModel):
    edad: int = Field(ge=0, le=120)
    ingresos: float = Field(ge=0)
    antiguedad_meses: int = Field(ge=0)

class PredictResponse(BaseModel):
    probabilidad: float
    etiqueta: str

@app.post("/predict", response_model=PredictResponse)
def predict(req: PredictRequest):
    x = [[req.edad, req.ingresos, req.antiguedad_meses]]
    prob = ml["model"].predict_proba(x)[0][1]
    return PredictResponse(
        probabilidad=round(prob, 4),
        etiqueta="alto" if prob > 0.5 else "bajo",
    )

El contrato de salida (response_model) es igual de útil: garantiza que nunca filtras un campo interno por accidente y te da documentación automática en /docs. Quien consume tu API ve exactamente qué entra y qué sale.

La trampa de async con modelos

Aquí casi todo el mundo se equivoca. FastAPI es asíncrono, pero la inferencia de un modelo es síncrona y consume CPU. Si declaras el endpoint como async def y dentro llamas a un predict pesado, bloqueas el bucle de eventos: mientras ese modelo calcula, ninguna otra petición avanza. Has convertido un servidor concurrente en uno de un solo carril.

La regla es sencilla y conviene tenerla a mano:

Tu código en el endpoint	Declara el endpoint como
Inferencia pesada de CPU (`predict`)	`def` (síncrono)
Llamada a una API externa con `await`	`async def`
Lectura de disco o consulta bloqueante	`def` (síncrono)

Cuando declaras el endpoint con def normal, FastAPI lo ejecuta en un pool de hilos y no bloquea el bucle. Es contraintuitivo: para algo que consume CPU, el def plano rinde mejor que el async def. Síguelo y la concurrencia te sale gratis.

Procesa por lotes si puedes

Un modelo casi siempre predice más rápido sobre un lote que sobre filas sueltas. Llamar predict cien veces con una fila es más lento que llamarlo una vez con cien filas, porque cada llamada arrastra su propia sobrecarga.

Si tu caso de uso lo permite, expón un endpoint de lote:

class BatchRequest(BaseModel):
    items: list[PredictRequest] = Field(max_length=500)

@app.post("/predict/batch")
def predict_batch(req: BatchRequest):
    X = [[i.edad, i.ingresos, i.antiguedad_meses] for i in req.items]
    probs = ml["model"].predict_proba(X)[:, 1]
    return [{"probabilidad": round(float(p), 4)} for p in probs]

El max_length no es decorativo: sin un tope, alguien te enviará un lote de un millón de filas y tumbará el proceso. Pon siempre un límite y devuélvelo en la documentación.

Despliegue: workers, no hilos

En desarrollo arrancas con uvicorn app:app --reload y listo. En producción necesitas varios procesos para usar todos los núcleos, porque el GIL de Python impide que los hilos aprovechen varias CPUs para cómputo puro.

La fórmula habitual es Uvicorn gestionado por varios workers, uno por núcleo disponible:

uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4

Cada worker carga su propia copia del modelo en memoria. Con un modelo grande esto importa: cuatro workers son cuatro veces la RAM del modelo. Mídelo antes de decidir el número de workers, no después de que el contenedor muera por falta de memoria.

Un par de cosas que no son opcionales en producción: un endpoint /health que confirme que el modelo está cargado, para que el orquestador sepa cuándo el proceso está listo, y logs con el tiempo de inferencia por petición. Sin esa medida, “va lento” es una sensación; con ella, es un número que puedes atacar.

La lista corta antes de publicar

Antes de poner un modelo detrás de una URL pública, repasa esto:

El modelo se carga en el lifespan, una vez por worker.
La entrada se valida con Pydantic y rechazas lo inválido con un 422.
Los endpoints de inferencia pesada son def, no async def.
Hay un tope al tamaño de lote.
Tienes /health y mides el tiempo de cada predicción.

Nada de esto es exótico. Es la diferencia entre un endpoint que aguanta el primer día con tráfico real y uno que te despierta a las tres de la mañana. El modelo era la parte fácil; servirlo bien es el oficio.