Artículos / MLOps
Sirve un modelo scikit-learn con FastAPI y Docker
Entrena Iris con scikit-learn y sírvelo con FastAPI: valida entradas, protege ambas rutas, acota lotes y prueba el contenedor Docker.
Una API de predicciones necesita un contrato de entrada claro, un artefacto de modelo confiable y una forma de rechazar llamadas no autorizadas. Este ejemplo entrena un clasificador Iris, sirve predicciones individuales y por lotes y empaqueta el mismo código en Docker. Es una demostración local, sin mediciones de capacidad ni promesa de estar lista para producción.
Archivos y versiones probadas
Usa Python 3.13 y estas dependencias directas en requirements.txt. Los comandos siguientes son PowerShell. Mantén alineadas las dependencias de entrenamiento y serving; para un release fija también las dependencias transitivas y el digest de la imagen base.
iris-api/
requirements.txt
train_model.py
test_api.py
app/
main.py
Dockerfile
fastapi==0.138.0
pydantic==2.13.4
uvicorn==0.49.0
scikit-learn==1.9.0
numpy==2.4.6
joblib==1.5.3
httpx==0.28.1
Entrena y guarda un artefacto confiable
La introducción a scikit-learn explica la evaluación. Aquí, el holdout estratificado mantiene representadas las clases de Iris. La accuracy impresa describe esta pequeña partición, no la calidad de un despliegue ni datos futuros.
# train_model.py
from pathlib import Path
import joblib
from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
from sklearn.model_selection import train_test_split
iris = load_iris()
X_train, X_test, y_train, y_test = train_test_split(
iris.data, iris.target, test_size=0.2, random_state=42, stratify=iris.target
)
model = RandomForestClassifier(n_estimators=100, random_state=42, n_jobs=1)
model.fit(X_train, y_train)
print(f"Holdout accuracy: {accuracy_score(y_test, model.predict(X_test)):.4f}")
model_dir = Path(__file__).resolve().parent / "model"
model_dir.mkdir(exist_ok=True)
joblib.dump(model, model_dir / "classifier.joblib")
joblib.dump(iris.target_names.tolist(), model_dir / "class_names.joblib")
Carga sólo artefactos producidos por tu pipeline de entrenamiento confiable: joblib puede ejecutar código al cargar. Importa mantener las versiones de scikit-learn y sus dependencias; cargar un artefacto antiguo en un entorno de serving actualizado no está soportado. El ejemplo Docker entrena dentro de su propio entorno.
Protege ambas rutas de predicción
Crea app/main.py. Las entradas son números JSON positivos y finitos; se rechazan strings, booleanos y campos extra. Un lote contiene entre 1 y 128 registros. Ambas rutas reutilizan la dependencia de API key y el helper de predicción, conservando el orden de entrada.
RandomForest convierte las entradas a float32 internamente. Un número JSON finito como 1e100 excede ese formato y puede provocar un error del servidor al predecir. El esquema aplica su límite numérico superior a cada variable; un servicio real también necesita rangos propios del dominio, además de validez numérica.
Configura API_KEY antes de importar la aplicación. Si falta o está en blanco, no arranca; las peticiones sin clave o con una incorrecta reciben 401. No agregues otro handler de la misma ruta para introducir autenticación: aplica la dependencia a las rutas originales, individual y de lotes.
# app/main.py
from contextlib import asynccontextmanager
import os
from pathlib import Path
import secrets
from typing import Annotated
import joblib
import numpy as np
from fastapi import Depends, FastAPI, HTTPException, Security
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from fastapi.security import APIKeyHeader
from pydantic import BaseModel, ConfigDict, Field
EXPECTED_API_KEY = os.environ.get("API_KEY")
if not EXPECTED_API_KEY or not EXPECTED_API_KEY.strip():
raise RuntimeError("API_KEY must be configured before starting the API")
MODEL_DIR = Path(__file__).resolve().parents[1] / "model"
ml_models = {}
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
def require_api_key(key: str | None = Security(api_key_header)):
if key is None or not secrets.compare_digest(
key.encode("utf-8"), EXPECTED_API_KEY.encode("utf-8")
):
raise HTTPException(
status_code=401, detail="Invalid API key",
headers={"WWW-Authenticate": "APIKey"},
)
MAX_FLOAT32 = float(np.finfo(np.float32).max)
class IrisFeatures(BaseModel):
model_config = ConfigDict(strict=True, extra="forbid", allow_inf_nan=False)
sepal_length: float = Field(gt=0, le=MAX_FLOAT32, description="Sepal length in cm")
sepal_width: float = Field(gt=0, le=MAX_FLOAT32, description="Sepal width in cm")
petal_length: float = Field(gt=0, le=MAX_FLOAT32, description="Petal length in cm")
petal_width: float = Field(gt=0, le=MAX_FLOAT32, description="Petal width in cm")
class PredictionResponse(BaseModel):
prediction: int
class_name: str
probabilities: list[float]
@asynccontextmanager
async def lifespan(app: FastAPI):
ml_models["classifier"] = joblib.load(MODEL_DIR / "classifier.joblib")
ml_models["class_names"] = joblib.load(MODEL_DIR / "class_names.joblib")
try:
yield
finally:
ml_models.clear()
app = FastAPI(title="Iris Classifier API", version="1.0.0", lifespan=lifespan)
@app.exception_handler(RequestValidationError)
async def invalid_request(request, error: RequestValidationError):
# Raw invalid values can contain NaN/infinity, which JSON cannot serialize.
details = [{"loc": item["loc"], "msg": item["msg"], "type": item["type"]}
for item in error.errors()]
return JSONResponse(status_code=422, content={"detail": details})
@app.get("/health")
def health_check():
if "classifier" not in ml_models or "class_names" not in ml_models:
raise HTTPException(status_code=503, detail="Model unavailable")
return {"status": "healthy", "model_loaded": True}
def predict_many(features_list: list[IrisFeatures]):
if "classifier" not in ml_models or "class_names" not in ml_models:
raise HTTPException(status_code=503, detail="Model unavailable")
matrix = np.array([[f.sepal_length, f.sepal_width, f.petal_length, f.petal_width]
for f in features_list])
model = ml_models["classifier"]
predictions = model.predict(matrix).tolist()
probabilities = model.predict_proba(matrix).tolist()
return [
PredictionResponse(
prediction=label, class_name=ml_models["class_names"][label],
probabilities=scores,
)
for label, scores in zip(predictions, probabilities)
]
@app.post("/predict", response_model=PredictionResponse,
dependencies=[Depends(require_api_key)])
def predict(features: IrisFeatures):
return predict_many([features])[0]
@app.post("/predict/batch", response_model=list[PredictionResponse],
dependencies=[Depends(require_api_key)])
def predict_batch(features_list: Annotated[list[IrisFeatures], Field(min_length=1, max_length=128)]):
return predict_many(features_list)
/health es público y devuelve 503 si el modelo no está disponible. Los artefactos faltantes impiden el arranque. El handler de validación omite la entrada inválida cruda para que NaN o infinito no conviertan una respuesta de validación en un error al serializar JSON. Las probabilidades siguen model.classes_—[setosa, versicolor, virginica] para este modelo—y no equivalen automáticamente a confianza calibrada.
Prueba la API sin arrancar un servidor
Crea test_api.py. TestClient entra en el lifespan, carga el modelo real guardado y envía peticiones a la aplicación dentro del proceso. Estos controles cubren resultados utilizables, autenticación en ambas rutas y límites de entrada.
# test_api.py
import os
from fastapi.testclient import TestClient
from app.main import app
sample = {"sepal_length": 5.1, "sepal_width": 3.5, "petal_length": 1.4, "petal_width": 0.2}
headers = {"X-API-Key": os.environ["API_KEY"]}
with TestClient(app) as client:
assert client.get("/health").status_code == 200
single = client.post("/predict", json=sample, headers=headers)
single.raise_for_status()
assert single.json()["class_name"] == "setosa"
batch = client.post("/predict/batch", json=[sample, sample], headers=headers)
batch.raise_for_status()
assert batch.json() == [single.json(), single.json()]
for route, payload in [("/predict", sample), ("/predict/batch", [sample])]:
assert client.post(route, json=payload).status_code == 401
assert client.post(route, json=payload, headers={"X-API-Key": "wrong"}).status_code == 401
assert client.post("/predict/batch", json=[], headers=headers).status_code == 422
assert client.post("/predict/batch", json=[sample] * 129, headers=headers).status_code == 422
assert client.post("/predict", json={**sample, "sepal_length": 0}, headers=headers).status_code == 422
for route, payload in [("/predict", {**sample, "sepal_length": 1e100}),
("/predict/batch", [{**sample, "sepal_length": 1e100}])]:
assert client.post(route, json=payload, headers=headers).status_code == 422
print("API contract checks passed")
Desde la raíz del proyecto, instala, genera una clave local de demostración, entrena y ejecuta las comprobaciones. La clave generada se asigna al entorno, sin escribirla en el código ni imprimirla en la consola.
python -m venv .venv
& .\.venv\Scripts\python.exe -m pip install -r requirements.txt
$env:API_KEY = & .\.venv\Scripts\python.exe -c "import secrets; print(secrets.token_urlsafe(32))"
& .\.venv\Scripts\python.exe train_model.py
& .\.venv\Scripts\python.exe test_api.py
Para explorar la API por HTTP, ejecuta el siguiente comando y abre la documentación local. Usa Authorize en Swagger con la clave configurada para las predicciones. Detén el servidor antes de iniciar el contenedor en ese mismo puerto.
& .\.venv\Scripts\python.exe -m uvicorn app.main:app --host 127.0.0.1 --port 8000
Construye y prueba el contenedor
La imagen instala las mismas versiones directas, entrena el modelo de ejemplo en ese entorno y se ejecuta con un usuario sin privilegios. La clave se suministra al arrancar. Guarda este Dockerfile:
FROM python:3.13-slim
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
ENV OMP_NUM_THREADS=1 OPENBLAS_NUM_THREADS=1 MKL_NUM_THREADS=1
WORKDIR /app
COPY requirements.txt .
RUN python -m pip install --no-cache-dir --only-binary=:all: -r requirements.txt
COPY train_model.py test_api.py ./
COPY app/ ./app/
RUN python train_model.py && useradd --create-home appuser
USER appuser
EXPOSE 8000
CMD ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
La primera ejecución prueba el contrato dentro de Docker, sin acceso a red ni servidor persistente. La segunda sirve HTTP sólo en la interfaz local del host; corre en primer plano y el contenedor se elimina al detenerlo.
docker build -t iris-classifier:demo .
docker run --rm --network none --env API_KEY iris-classifier:demo python test_api.py
docker run --rm --env API_KEY -p 127.0.0.1:8000:8000 iris-classifier:demo
Antes de exponer un servicio real
Una API key compartida no aporta identidad de usuario, roles, aislamiento de tenants ni rotación automática. Un despliegue real necesita HTTPS, secretos administrados, límites de tamaño y tasa de peticiones, procedencia de artefactos, monitoreo y capacidad medida. Cada worker carga su modelo; añadir procesos consume memoria. Los handlers síncronos permiten que FastAPI ejecute inferencia fuera del event loop, pero no aceleran el trabajo de CPU.
El repositorio histórico fast-ml-deploy usa /v1/iris/predict y otro payload. Su README y este ejemplo completo describen interfaces separadas. Continúa con orquestación de pipelines y monitoreo de calidad con labels y data drift.
Fuentes
¿Vas a construir una plataforma de IA o de datos?
Si necesitas adaptar este enfoque a tu sistema, cuéntame qué datos tienes y qué decisión técnica necesitas resolver.