Referencia: Un Pipeline de ML de Producción

Un pipeline de ML de referencia para entornos Python: entrenamiento por lotes con PyTorch 2.6+ o scikit-learn, registro de artefactos, servicio con FastAPI y monitorización de deriva y latencia, no solo proyectos de ciencia en notebooks.

Receta

Tarjeta de referencia rápida - lista para copiar y pegar.

fraud-ml/
  pipelines/train.py       # trabajo de entrenamiento programado
  pipelines/features.py    # escrituras de la tienda de características
  serving/app.py           # inferencia FastAPI
  models/                  # pesos exportados (S3 en producción)
  eval/metrics.py          # puertas de evaluación offline
  monitoring/drift.py      # comprobaciones de distribución de producción

Cuándo usar esto:

• Primer modelo que va a producción
• Línea base de RFC de MLOps
• Brecha de auditoría entre notebook y servicio

Ejemplo de Trabajo

"""pipelines/train.py - entrenamiento por lotes con puerta de evaluación."""
from __future__ import annotations
 
import json
from pathlib import Path
 
import joblib
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import train_test_split
 
from pipelines.features import load_feature_frame
 
 
def train(version: str, min_auc: float = 0.82) -> Path:
    X, y = load_feature_frame()
    X_train, X_test, y_train, y_test = train_test_split(X, y, stratify=y, random_state=42)
    model = GradientBoostingClassifier(random_state=42)
    model.fit(X_train, y_train)
    auc = roc_auc_score(y_test, model.predict_proba(X_test)[:, 1])
    if auc < min_auc:
        raise RuntimeError(f"AUC {auc:.3f} por debajo de la puerta {min_auc}")
    out = Path(f"models/{version}/model.joblib")
    out.parent.mkdir(parents=True, exist_ok=True)
    joblib.dump(model, out)
    out.with_suffix(".metrics.json").write_text(json.dumps({"auc": auc}))
    return out

"""serving/app.py - API de inferencia."""
from __future__ import annotations
 
import joblib
from fastapi import FastAPI
from pydantic import BaseModel, Field
 
app = FastAPI()
MODEL = joblib.load("models/prod/model.joblib")
 
 
class Features(BaseModel):
    amount: float = Field(gt=0)
    merchant_risk_score: float
 
 
class Score(BaseModel):
    fraud_probability: float
    model_version: str = "2026.07.09"
 
 
@app.post("/v1/score", response_model=Score)
def score(body: Features) -> Score:
    prob = float(MODEL.predict_proba([[body.amount, body.merchant_risk_score]])[0, 1])
    return Score(fraud_probability=prob)

Lo que esto demuestra:

• El trabajo de entrenamiento falla en CI si la puerta de evaluación no se cumple.
• El servicio carga artefactos versionados; Pydantic valida las entradas.
• Clara separación entre entrenamiento por lotes e inferencia en línea.
• El artefacto de métricas viaja con el modelo para auditoría.

Análisis Profundo

Cómo Funciona

• Pipeline de características - Escritura por lotes en el almacén de datos; la ruta en línea lee datos precalculados o un subconjunto de baja latencia.
• Registro - S3 o MLflow almacenan model.joblib + metadatos JSON.
• Promoción - Comparación en sombra o canary del nuevo modelo AUC/latencia antes de cambiar el alias.
• Monitorización - Registro de predicciones; comparación semanal de la distribución de entrada (PSI).
• Reversión - Versión anterior del modelo en el registro; señal para revertir el alias.

Etapas del Pipeline

Notas de Python

# monitoring/drift.py - comprobación simplificada de PSI
def population_stability_index(expected: list[float], actual: list[float]) -> float:
    ...

Errores Comunes

• Sesgo entre entrenamiento y servicio - Diferentes rutas de código de características. Solución: Paquete features.py compartido importado en ambos lados.
• Notebook como producción - Sin pruebas ni versionado. Solución: Exportar funciones a módulos pipelines/.
• Sin validación de entrada - La basura bloquea el modelo. Solución: Límites de Pydantic en la API de servicio.
• Desajuste GPU train, CPU serve - Problemas de formato de exportación. Solución: Estandarizar en ONNX o joblib por familia de modelos.
• Ignorar la deriva - Decaimiento silencioso de la calidad. Solución: Informe semanal por lotes + umbral de alerta.

Alternativas

Preguntas Frecuentes

Relacionado

• Referencia: Una Plataforma de Datos / Sistema ETL - upstream de características
• Referencia: Un Backend SaaS con FastAPI - host de servicio
• Conceptos Básicos de MLOps - profundidad de operaciones
• Pipelines Clásicos de ML - patrones de sklearn
• Benchmarks de Escalabilidad - pruebas de carga de inferencia

Versiones de Stack: Esta página fue escrita para Python 3.14.0 (estable 3.14, mantenimiento 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, y uv 0.6+.