Monitoreo de Modelos ML en Producción con Evidently
Detecta data drift y degradación de modelos ML en producción con Evidently AI: reportes automatizados y alertas antes de que tu modelo falle en silencio.
Entrenaste un gran modelo. Pasó la validación cruzada con nota sobresaliente y superó el holdout set sin problemas. Lo desplegaste. Y luego, tres meses después, alguien nota que las predicciones ya no cuadran.
Esto no es un escenario hipotético — es lo que le pasa a la mayoría de los modelos de ML en producción. El mundo cambia. Tu modelo no. Este artículo muestra cómo detectar ese drift antes de que se convierta en un problema de negocio, usando Evidently: reportes de drift, reportes de rendimiento del modelo y test suites automatizadas que puedes conectar a alertas.
Por Qué Se Degradan los Modelos
Hay dos modos principales de falla:
Data drift: la distribución estadística de las features de entrada cambia. Ejemplo: tu modelo de detección de fraude fue entrenado cuando la mayoría de las transacciones eran menores a 450. El modelo nunca vio esa región del espacio de features.
Concept drift: la relación entre las entradas y el target cambia. Ejemplo: el comportamiento del cliente después de un evento económico importante cambia — features que antes predecían churn ya no lo hacen.
Ambos se manifiestan como degradación silenciosa: la API sigue devolviendo predicciones, solo que peores. El monitoreo detecta esto antes de que los stakeholders lo hagan.
Instalando Evidently
pip install evidently scikit-learn pandas numpy
Construiremos un ejemplo completo: entrenar un modelo de predicción de churn, simular datos de producción con drift, y generar reportes de Evidently.
Entrenando el Modelo de Referencia
# train.py
import joblib
import numpy as np
import pandas as pd
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
np.random.seed(42)
n = 2000
# Simular features de clientes
df_train = pd.DataFrame({
'meses_antiguedad': np.random.exponential(24, n),
'cargo_mensual': np.random.normal(65, 20, n).clip(20, 150),
'num_productos': np.random.randint(1, 6, n),
'llamadas_soporte': np.random.poisson(1.5, n),
'dias_ultimo_acceso': np.random.exponential(15, n),
})
# Target: churn (más llamadas y menos productos → más probable que se vaya)
churn_prob = (
0.3
- 0.008 * df_train['meses_antiguedad']
- 0.05 * df_train['num_productos']
+ 0.04 * df_train['llamadas_soporte']
+ 0.003 * df_train['dias_ultimo_acceso']
).clip(0.05, 0.95)
df_train['churn'] = (np.random.random(n) < churn_prob).astype(int)
X = df_train.drop('churn', axis=1)
y = df_train['churn']
X_train, X_ref, y_train, y_ref = train_test_split(X, y, test_size=0.3, random_state=42)
model = Pipeline([
('scaler', StandardScaler()),
('clf', GradientBoostingClassifier(n_estimators=100, random_state=42)),
])
model.fit(X_train, y_train)
print(f"Accuracy en train: {model.score(X_train, y_train):.4f}")
print(f"Accuracy en val: {model.score(X_ref, y_ref):.4f}")
# Guardar modelo y datos de referencia
joblib.dump(model, 'model.joblib')
X_ref.to_parquet('datos_referencia.parquet', index=False)
y_ref.to_frame().to_parquet('labels_referencia.parquet', index=False)
El dataset de referencia es crucial: es la línea de base estadística que Evidently usa para comparar cuando llegan datos nuevos.
Simulando Datos de Producción con Drift
En un escenario real recolectarías predicciones en vivo. Aquí simulamos dos meses de datos de producción donde la distribución ha cambiado:
# simular_produccion.py
import numpy as np
import pandas as pd
np.random.seed(99)
n = 500
# Mes 1: similar al entrenamiento (sin drift aún)
df_mes1 = pd.DataFrame({
'meses_antiguedad': np.random.exponential(24, n),
'cargo_mensual': np.random.normal(65, 20, n).clip(20, 150),
'num_productos': np.random.randint(1, 6, n),
'llamadas_soporte': np.random.poisson(1.5, n),
'dias_ultimo_acceso': np.random.exponential(15, n),
})
# Mes 2: con drift — nueva tarificación subió los cargos, clientes con menos productos
df_mes2 = pd.DataFrame({
'meses_antiguedad': np.random.exponential(18, n), # antigüedades más cortas (usuarios nuevos)
'cargo_mensual': np.random.normal(95, 25, n).clip(20, 150), # cargos más altos
'num_productos': np.random.randint(1, 3, n), # menos productos
'llamadas_soporte': np.random.poisson(3.0, n), # más llamadas
'dias_ultimo_acceso': np.random.exponential(30, n), # menos engagement
})
df_mes1.to_parquet('produccion_mes1.parquet', index=False)
df_mes2.to_parquet('produccion_mes2.parquet', index=False)
Reporte de Data Drift
Evidently compara el dataset de referencia contra tus datos de producción actuales y calcula tests estadísticos por feature:
# reporte_drift.py
import pandas as pd
import joblib
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, DataQualityPreset
referencia = pd.read_parquet('datos_referencia.parquet')
mes1 = pd.read_parquet('produccion_mes1.parquet')
mes2 = pd.read_parquet('produccion_mes2.parquet')
# ── Mes 1: debería verse limpio ─────────────────────────────────────
reporte_m1 = Report(metrics=[DataDriftPreset(), DataQualityPreset()])
reporte_m1.run(reference_data=referencia, current_data=mes1)
reporte_m1.save_html('drift_mes1.html')
# ── Mes 2: debería mostrar drift ────────────────────────────────────
reporte_m2 = Report(metrics=[DataDriftPreset(), DataQualityPreset()])
reporte_m2.run(reference_data=referencia, current_data=mes2)
reporte_m2.save_html('drift_mes2.html')
Abre drift_mes2.html en tu navegador. Verás algo como:
cargo_mensual: DRIFT DETECTADO (distancia de Wasserstein = 0.42)llamadas_soporte: DRIFT DETECTADO (PSI = 0.31)meses_antiguedad: borderline- Features con drift: 3/5 (60%)
Reporte de Rendimiento del Modelo
Si tienes etiquetas para los datos de producción (común en escenarios batch con feedback diferido), también puedes monitorear la calidad del modelo directamente:
import numpy as np
from evidently import ColumnMapping
from evidently.metric_preset import ClassificationPreset
model = joblib.load('model.joblib')
# Agregar predicciones y simular labels para el mes 2
mes2_eval = mes2.copy()
mes2_eval['prediction'] = model.predict(mes2_eval)
mes2_eval['target'] = (np.random.random(len(mes2_eval)) < 0.45).astype(int) # degradado
referencia_eval = referencia.copy()
referencia_eval['prediction'] = model.predict(referencia_eval)
referencia_eval['target'] = pd.read_parquet('labels_referencia.parquet')['churn'].values
reporte_perf = Report(metrics=[ClassificationPreset()])
reporte_perf.run(
reference_data=referencia_eval,
current_data=mes2_eval,
column_mapping=ColumnMapping(target='target', prediction='prediction'),
)
reporte_perf.save_html('rendimiento_modelo.html')
Alertas Programáticas
Los reportes HTML son útiles para exploración, pero para monitoreo automatizado necesitas acceso programático a los resultados:
from evidently.test_suite import TestSuite
from evidently.tests import (
TestShareOfDriftedColumns,
TestColumnDrift,
)
suite = TestSuite(tests=[
TestShareOfDriftedColumns(lt=0.3), # Falla si >30% de las columnas tienen drift
TestColumnDrift(column_name='cargo_mensual'),
TestColumnDrift(column_name='llamadas_soporte'),
])
suite.run(reference_data=referencia, current_data=mes2)
resultados = suite.as_dict()
for test in resultados['tests']:
status = test['status']
name = test['name']
print(f"[{status}] {name}")
# [FAIL] Share of Drifted Columns
# [FAIL] Column Drift: cargo_mensual
# [FAIL] Column Drift: llamadas_soporte
Puedes conectar resultados con cualquier sistema de alertas — Slack, PagerDuty, email — para recibir notificaciones en el momento en que el drift supere tus umbrales.
Programando Chequeos de Drift Regulares
En producción normalmente corres chequeos de drift en un schedule. Un enfoque simple con cron:
# monitorear.py — ejecutar diariamente vía cron o Prefect
import datetime
import pandas as pd
from evidently.test_suite import TestSuite
from evidently.tests import TestShareOfDriftedColumns
def ejecutar_chequeo_diario(produccion_path: str, referencia_path: str) -> bool:
referencia = pd.read_parquet(referencia_path)
actual = pd.read_parquet(produccion_path)
suite = TestSuite(tests=[TestShareOfDriftedColumns(lt=0.3)])
suite.run(reference_data=referencia, current_data=actual)
paso = suite.as_dict()['summary']['all_passed']
fecha = datetime.date.today().isoformat()
if not paso:
print(f"[{fecha}] ALERTA: drift detectado en datos de producción")
else:
print(f"[{fecha}] OK: sin drift significativo")
return paso
if __name__ == '__main__':
ejecutar_chequeo_diario('produccion_hoy.parquet', 'datos_referencia.parquet')
Si ya orquestas el reentrenamiento con Prefect, este chequeo entra como un flow programado más: Orquesta Pipelines de ML con Prefect.
Qué Hacer Cuando Se Detecta Drift
La detección es solo el primer paso. Cuando se dispara una alerta, el protocolo es:
- Investigar: ¿en qué features hay drift? ¿De qué magnitud?
- Diagnosticar la causa raíz: ¿cambio en el pipeline de datos upstream? ¿Estacionalidad? ¿Evento del mundo real?
- Decidir: ¿reentrenar, adaptar el preprocesamiento, o escalar a una decisión humana?
- Reentrenar si es necesario: usa los datos con drift como nueva referencia después de reentrenar
La conclusión clave es que el monitoreo no reemplaza el reentrenamiento — te dice cuándo reentrenar, y con qué datos.
Conclusión
La degradación silenciosa de modelos es uno de los problemas más comunes y costosos en ML en producción. Evidently te da la infraestructura para detectarlo automáticamente, con configuración mínima. El patrón siempre es el mismo: recolecta un dataset de referencia representativo al momento de entrenar, compara los datos de producción contra él regularmente, y actúa cuando los tests estadísticos señalen un cambio significativo.
Para cerrar el loop de MLOps, rastrea tus experimentos de reentrenamiento desde el primer día: Gestiona el Ciclo de Vida de Machine Learning con MLflow en Python.
Sobre el autor
Rodrigo Arenas es arquitecto de software e ingeniero de machine learning en Medellín, Colombia. Construye productos, plataformas y software open source para IA, Data Engineering y Machine Learning — es el creador de Ciaren, sklearn-genetic-opt y PyWorkforce.
Lectura recomendada
Rodrigo Arenas — ML Engineer & Open Source Maintainer
¿Estás trabajando en algo similar? Puedo ayudarte a llevarlo a producción.