Si trabajas con Machine Learning, seguro conoces esta situación: entrenas un modelo, consigues una métrica decente, lo guardas en un .pkl... y ahí se queda. El modelo vive en tu notebook, pero nadie más puede usarlo. No hay forma de que una app, un frontend o un compañero de equipo lo consuman.
Ahí es donde entra FastAPI. Es uno de los frameworks que más se usan últimamente para resolver justo ese problema: convertir un modelo entrenado en algo que cualquier sistema pueda llamar mediante una simple petición HTTP. En este artículo te voy a enseñar cómo construir una API que sirva un modelo de Scikit-learn, con validación de datos, manejo de errores y buenas prácticas que puedes usar en tu día a día.
No es un tutorial exhaustivo de FastAPI (eso daría para un libro), sino una guía práctica centrada en el caso de uso más común: exponer un modelo de ML como servicio.
¿Por qué FastAPI?
Antes de meternos en código, vale la pena entender por qué FastAPI se ha convertido en el estándar de facto para servir modelos de ML.
Es rápido, dentro de lo que cabe en Python. Está construido sobre Starlette (para la parte async) y Pydantic (para la validación), lo que le da un rendimiento muy por encima de Flask o Django en benchmarks típicos. No vas a igualar a Node.js o Go en throughput puro (ningún framework de Python lo hace), pero dentro del ecosistema Python es de los más rápidos disponibles.
Valida datos automáticamente. Esto es enorme cuando hablamos de ML. Si tu modelo espera 4 features numéricas y alguien te manda un string, FastAPI lo rechaza antes de que llegue a tu código. Te ahorras la mitad de los try/except que escribirías en Flask.
Documentación automática. Cada endpoint que escribes genera documentación interactiva en /docs (Swagger UI) sin que tengas que tocar nada. Para un equipo que necesita probar tu modelo sin escribir código, esto es oro.
Tipado nativo de Python. Usa type hints para todo, lo cual significa mejor autocompletado, menos bugs y código que se explica solo.
El punto de partida: un modelo entrenado
Vamos a trabajar con un caso clásico como un modelo de clasificación entrenado con Scikit-learn. Imagina que ya tienes esto:
# train_model.py
import joblib
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
# Cargamos datos y entrenamos
data = load_iris()
X_train, X_test, y_train, y_test = train_test_split(
data.data, data.target, test_size=0.2, random_state=42
)
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)
print(f"Accuracy: {model.score(X_test, y_test):.3f}")
# Guardamos el modelo entrenado
joblib.dump(model, "modelo_iris.pkl")
Hasta aquí, nada que no hayas visto mil veces. Lo interesante empieza ahora: cómo convertimos ese .pkl en un servicio que cualquiera pueda consumir.
Estructura del proyecto
Antes de escribir código, una estructura (aunque sea básica) es muy importante.
ml-api/
├── app/
│ ├── __init__.py
│ ├── main.py # Punto de entrada de FastAPI
│ ├── models.py # Esquemas de Pydantic
│ ├── ml_model.py # Lógica de carga e inferencia
│ └── config.py # Configuración
├── modelo_iris.pkl
├── requirements.txt
└── .env
Paso 1: definir qué espera y qué devuelve la API
Aquí es donde Pydantic entra en juego, y es probablemente la parte más importante de todo el artículo. Pydantic te permite definir un esquema de datos con tipos, y FastAPI lo usa automáticamente para validar cada petición.
# app/models.py
from pydantic import BaseModel, ConfigDict, Field, field_validator
class IrisInput(BaseModel):
model_config = ConfigDict(
json_schema_extra={
"example": {
"sepal_length": 5.1,
"sepal_width": 3.5,
"petal_length": 1.4,
"petal_width": 0.2
}
}
)
sepal_length: float = Field(..., gt=0, le=15, description="Longitud del sépalo en cm")
sepal_width: float = Field(..., gt=0, le=15, description="Ancho del sépalo en cm")
petal_length: float = Field(..., gt=0, le=15, description="Longitud del pétalo en cm")
petal_width: float = Field(..., gt=0, le=15, description="Ancho del pétalo en cm")
@field_validator("*")
@classmethod
def check_no_nan(cls, value: float) -> float:
if value != value: # forma rápida de detectar NaN
raise ValueError("El valor no puede ser NaN")
return value
class PredictionOutput(BaseModel):
prediction: str
confidence: float
probabilities: dict[str, float]
Fíjate en un par de detalles que marcan la diferencia entre un esquema "que funciona" y uno bien pensado:
Field(..., gt=0, le=15)no solo dice que el campo es obligatorio (el...), también define un rango razonable. Si alguien manda un sépalo de -3 cm, FastAPI rechaza la petición sin que tu modelo tenga que enterarse. En este caso concreto, ese rango ya descarta NaN por sí solo (nan <= 15siempre daFalseen Python), así que elfield_validatorde abajo es estrictamente redundante aquí. Lo dejo igualmente porque es el patrón que vas a necesitar en cuanto tengas un campo sin límites numéricos claros, donde un NaN sí podría colarse sin que ningúnFieldlo detecte.El
model_configconjson_schema_extrano afecta a la lógica, pero hace que la documentación automática (/docs) muestre un ejemplo realista, lo cual ayuda muchísimo a quien va a consumir tu API por primera vez. UsaConfigDicty no la antiguaclass Config, que Pydantic 2 marca como obsoleta.
Paso 2: encapsular la lógica del modelo
Es tentador cargar el modelo directamente en main.py y listo. No lo hagas. Separar esta lógica te permite testear el modelo sin levantar un servidor, y cambiar de implementación (por ejemplo, pasar de Scikit-learn a ONNX) sin tocar la API.
# app/ml_model.py
import joblib
import numpy as np
from pathlib import Path
from functools import lru_cache
MODEL_PATH = Path(__file__).parent.parent / "modelo_iris.pkl"
CLASS_NAMES = ["setosa", "versicolor", "virginica"]
class IrisModel:
def __init__(self, model_path: Path):
self.model = joblib.load(model_path)
def predict(self, features: list[float]) -> dict:
X = np.array([features])
prediction = self.model.predict(X)[0]
probabilities = self.model.predict_proba(X)[0]
return {
"prediction": CLASS_NAMES[prediction],
"confidence": float(max(probabilities)),
"probabilities": {
name: float(prob)
for name, prob in zip(CLASS_NAMES, probabilities)
}
}
@lru_cache
def get_model() -> IrisModel:
return IrisModel(MODEL_PATH)
El @lru_cache en get_model() es un truco pequeño pero útil: garantiza que el modelo se cargue una sola vez en memoria, sin importar cuántas veces se llame a esta función. Esto se combina muy bien con el sistema de inyección de dependencias de FastAPI, que vemos ahora.
Paso 3: el endpoint
# app/main.py
from fastapi import FastAPI, Depends
from app.models import IrisInput, PredictionOutput
from app.ml_model import IrisModel, get_model
app = FastAPI(
title="API de Clasificación de Iris",
description="Sirve un modelo de Random Forest entrenado con el dataset Iris",
version="1.0.0"
)
@app.get("/health")
def health_check():
return {"status": "ok"}
@app.post("/predict", response_model=PredictionOutput)
def predict(
data: IrisInput,
model: IrisModel = Depends(get_model)
):
features = [
data.sepal_length,
data.sepal_width,
data.petal_length,
data.petal_width
]
return model.predict(features)
Hay varias decisiones de diseño aquí que conviene explicar:
Depends(get_model) es la inyección de dependencias de FastAPI. En vez de cargar el modelo dentro de la función predict, se lo pides como argumento. FastAPI se encarga de resolverlo. La ventaja real aparece cuando escribes tests: puedes sustituir get_model por una versión falsa sin tocar el endpoint.
response_model=PredictionOutput le dice a FastAPI qué forma debe tener la respuesta. Esto cumple dos funciones: valida que tu código no esté devolviendo algo inesperado, y documenta automáticamente el formato de salida en /docs.
El endpoint /health parece trivial, pero es imprescindible en producción. Cualquier orquestador (Docker, Kubernetes, un balanceador de carga) necesita una forma de preguntar "¿estás viva?" antes de enviarte tráfico.
Fíjate en que el endpoint no tiene ningún try/except. Es intencional, y es justo lo que vemos en la siguiente sección: si algo falla dentro de model.predict(), la excepción sube sola hasta donde realmente debe resolverse.
Manejo de errores
El error más común que he visto (y que yo mismo he cometido) es envolver el endpoint en un try/except Exception genérico que convierte todo en un 500. El problema es que eso "traga" cualquier excepción más específica antes de que pueda tratarse como lo que realmente es.
# app/main.py (continuación)
from fastapi import Request
from fastapi.responses import JSONResponse
from app.ml_model import PredictionError
class ModelNotLoadedError(Exception):
pass
@app.exception_handler(ModelNotLoadedError)
async def model_not_loaded_handler(request: Request, exc: ModelNotLoadedError):
return JSONResponse(
status_code=503,
content={"detail": "El modelo no está disponible actualmente. Inténtalo en unos segundos."}
)
@app.exception_handler(PredictionError)
async def prediction_error_handler(request: Request, exc: PredictionError):
return JSONResponse(
status_code=422,
content={"detail": f"No se pudo generar la predicción: {str(exc)}"}
)
Fíjate en que PredictionError no se define aquí, sino que se importa desde app.ml_model. Tiene que vivir junto a quien la lanza (IrisModel.predict()), no en main.py; si la defines en los dos sitios acabas con dos clases distintas con el mismo nombre, y el except de una no captura las instancias de la otra.
Para que esto funcione de verdad, hay que volver a IrisModel.predict() (la del Paso 2) y hacer que traduzca cualquier fallo de NumPy o Scikit-learn en PredictionError, en lugar de dejar que se propague tal cual:
# app/ml_model.py (versión final, con PredictionError definida aquí)
import joblib
import numpy as np
from pathlib import Path
from functools import lru_cache
MODEL_PATH = Path(__file__).parent.parent / "modelo_iris.pkl"
CLASS_NAMES = ["setosa", "versicolor", "virginica"]
class PredictionError(Exception):
"""Error al generar una predicción con el modelo ya cargado."""
pass
class IrisModel:
def __init__(self, model_path: Path):
self.model = joblib.load(model_path)
def predict(self, features: list[float]) -> dict:
try:
X = np.array([features])
prediction = self.model.predict(X)[0]
probabilities = self.model.predict_proba(X)[0]
except Exception as e:
raise PredictionError(str(e)) from e
return {
"prediction": CLASS_NAMES[prediction],
"confidence": float(max(probabilities)),
"probabilities": {
name: float(prob)
for name, prob in zip(CLASS_NAMES, probabilities)
}
}
@lru_cache
def get_model() -> IrisModel:
return IrisModel(MODEL_PATH)
El try/except Exception solo tiene sentido en la capa donde se genera la predicción, para transformar cualquier error inesperado en una excepción de dominio (PredictionError). El endpoint no necesita capturar errores genéricos: simplemente deja que las excepciones lleguen a su exception handler correspondiente.
Así, la API devuelve respuestas claras y útiles en lugar de un 500 genérico. Por ejemplo, un 422 indica que el problema está en los datos enviados por el cliente, mientras que un 503 señala un fallo temporal del servicio.
Cargar el modelo solo una vez
Un error común es recargar el modelo en cada petición, lo cual es lentísimo y, en modelos grandes. FastAPI tiene un mecanismo pensado justo para esto: el ciclo de vida de la aplicación (lifespan).
# app/main.py (versión con lifespan)
from contextlib import asynccontextmanager
from fastapi import FastAPI
ml_models = {}
@asynccontextmanager
async def lifespan(app: FastAPI):
# Se ejecuta UNA vez, al arrancar la app
ml_models["iris"] = IrisModel(MODEL_PATH)
print("Modelo cargado correctamente")
yield
# Se ejecuta al apagar la app (limpieza de recursos)
ml_models.clear()
print("Modelo descargado, cerrando aplicación")
app = FastAPI(lifespan=lifespan)
Este patrón es preferible al @lru_cache que vimos antes cuando el modelo es pesado o cuando necesitas hacer algo más al arrancar (por ejemplo, conectar a una base de datos o verificar que el modelo carga sin errores antes de aceptar tráfico). Si el modelo falla al cargar, la aplicación ni siquiera arranca, lo cual es exactamente el comportamiento que quieres: prefieres que falle al desplegar, no en mitad de una petición real.
Probando la API
Una vez que tienes el servidor corriendo (uvicorn app.main:app --reload), FastAPI genera automáticamente documentación interactiva en dos formatos:
Swagger UI en
http://localhost:8000/docs: interfaz donde puedes probar cada endpoint directamente desde el navegador.ReDoc en
http://localhost:8000/redoc: documentación más orientada a lectura, útil para compartir con otros equipos.
Para probar el endpoint de predicción desde la terminal:
curl -X POST "http://localhost:8000/predict" \
-H "Content-Type: application/json" \
-d '{
"sepal_length": 5.1,
"sepal_width": 3.5,
"petal_length": 1.4,
"petal_width": 0.2
}'
Y la respuesta debería verse así:
{
"prediction": "setosa",
"confidence": 0.98,
"probabilities": {
"setosa": 0.98,
"versicolor": 0.02,
"virginica": 0.0
}
}
Si mandas un dato fuera de rango, por ejemplo "sepal_length": -5, FastAPI te devuelve un 422 con el detalle exacto de qué campo falló y por qué, sin que tu código tenga que hacer nada extra. Esto es validación automática en su máxima expresión.
Escribiendo tests para la API
FastAPI viene preparado para esto gracias a TestClient, que te permite simular peticiones HTTP sin levantar un servidor real.
# tests/test_main.py
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_health_check():
response = client.get("/health")
assert response.status_code == 200
assert response.json() == {"status": "ok"}
def test_predict_valid_input():
payload = {
"sepal_length": 5.1,
"sepal_width": 3.5,
"petal_length": 1.4,
"petal_width": 0.2
}
response = client.post("/predict", json=payload)
assert response.status_code == 200
data = response.json()
assert data["prediction"] in ["setosa", "versicolor", "virginica"]
assert 0 <= data["confidence"] <= 1
def test_predict_invalid_input_out_of_range():
payload = {
"sepal_length": -3, # valor fuera de rango
"sepal_width": 3.5,
"petal_length": 1.4,
"petal_width": 0.2
}
response = client.post("/predict", json=payload)
assert response.status_code == 422
def test_predict_missing_field():
payload = {
"sepal_length": 5.1,
"sepal_width": 3.5
# faltan petal_length y petal_width
}
response = client.post("/predict", json=payload)
assert response.status_code == 422
Async vs sync
En FastAPI no siempre conviene usar async def. Si el endpoint solo ejecuta una predicción local de CPU (por ejemplo, con Scikit-learn), lo adecuado es usar def, ya que FastAPI la ejecuta en un thread pool sin bloquear el servidor.
async def solo aporta ventajas cuando el endpoint realiza operaciones de I/O, como consultas a bases de datos, llamadas a APIs o servicios externos. Declarar un endpoint como async y ejecutar dentro código síncrono y bloqueante (como model.predict()) puede perjudicar el rendimiento al bloquear el bucle de eventos.
La regla práctica es sencilla: si todo el código es síncrono, usa def; si trabajas con librerías realmente asíncronas, entonces utiliza async def.
Conclusiones
FastAPI no resuelve el problema de hacer buen Machine Learning, pero resuelve casi perfectamente el problema de exponerlo. La combinación de validación automática con Pydantic, documentación generada sola y un sistema de inyección de dependencias simple lo convierte en la opción más razonable hoy en día para este caso de uso, sobre todo cuando vienes del mundo de los notebooks y quieres dar el salto a algo que otros puedan consumir de forma fiable.
Lo que hemos visto aquí es la base: un endpoint, un modelo cargado correctamente y errores manejados con cabeza. A partir de aquí, el camino natural es añadir autenticación, logging, tests y, eventualmente, contenerizar todo para desplegarlo. Pero si llegas hasta aquí, ya tienes algo mucho más sólido que el típico .pkl abandonado en una carpeta.