Monitorizar un pipeline de IA en Python: del entrenamiento a producción

Un entrenamiento de ocho horas que llevaba divergiendo desde el epoch 2 y nadie lo vio hasta el checkpoint final. Un modelo en producción que perdió varios puntos de accuracy en tres semanas sin que saltara ninguna alerta. Monitorizar un pipeline de IA en Python no es un problema técnico único, sino al menos cuatro problemas distintos según la etapa del ciclo de vida del modelo en la que ocurra, y tratarlos todos con la misma herramienta —o con ninguna— es la razón habitual por la que se detectan tarde.

Este artículo recorre esas etapas en orden: qué vigilar durante el entrenamiento, qué registrar al comparar experimentos, y qué hacer distinto una vez el modelo sirve tráfico real, con la herramienta y el ejemplo que corresponde a cada una.

Durante el entrenamiento: qué medir por epoch y cómo detectarlo a tiempo

Un patrón de callbacks framework-agnóstico consiste en una clase base con métodos vacíos —on_epoch_begin, on_epoch_end, on_batch_begin, on_batch_end— que un bucle de entrenamiento custom invoca en cada punto del ciclo, permitiendo añadir logging, checkpointing o alertas tempranas sin tocar la lógica de entrenamiento. Es un patrón propio, no la API de ningún framework concreto, y por eso funciona igual con un bucle de PyTorch escrito a mano, con un modelo de scikit-learn envuelto en un loop externo, o con cualquier proceso iterativo que tenga una noción de "epoch":

import logging

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)


class Callback:
    def on_epoch_begin(self, epoch: int) -> None:
        pass

    def on_epoch_end(self, epoch: int, logs: dict | None = None) -> None:
        pass

    def on_batch_begin(self, batch: int) -> None:
        pass

    def on_batch_end(self, batch: int, logs: dict | None = None) -> None:
        pass


class EarlyWarningCallback(Callback):
    def on_epoch_end(self, epoch: int, logs: dict | None = None) -> None:
        logs = logs or {}
        loss = logs.get("loss")
        if loss is not None and (loss != loss or loss in (float("inf"), float("-inf"))):
            raise RuntimeError(f"loss inválido en epoch {epoch}: {loss}")
        val_loss, train_loss = logs.get("val_loss"), logs.get("loss")
        if val_loss is not None and train_loss is not None and val_loss > train_loss * 1.5:
            logger.warning(
                f"epoch {epoch}: val_loss se dispara frente a train_loss "
                f"({val_loss:.3f} vs {train_loss:.3f})"
            )

Qué medir en cada on_epoch_end: loss de entrenamiento y de validación por separado, learning rate efectivo (sobre todo con scheduler), norma del gradiente si el modelo es propenso a explotar o desvanecerse, y duración del epoch. Ese último dato parece trivial pero suele ser el primero en delatar una degradación de rendimiento: si el epoch 1 tarda 4 minutos y el epoch 8 tarda 11 sin que el modelo haya cambiado de tamaño, el data loader es uno de los primeros sospechosos —junto con throttling térmico de la GPU, acumulación de tensores sin liberar o un callback de evaluación que se ha vuelto más caro—, y el paso correcto es perfilar la iteración, no asumir la causa de entrada.

Si el bucle de entrenamiento no es propio, no hace falta reinventar esto. Keras expone su propia clase Callback con hooks equivalentes —on_epoch_begin, on_epoch_end(epoch, logs), on_train_batch_begin/on_train_batch_end— conectados directamente a model.fit(). PyTorch Lightning usa otra convención de nombres, algo que no conviene dar por sentado viniendo de Keras: no existe on_epoch_end, los hooks son on_train_epoch_start/on_train_epoch_end y on_train_batch_start/on_train_batch_end. El patrón casero de arriba tiene sentido sin framework de por medio; con Keras o Lightning, su propio sistema de callbacks hace el mismo trabajo sin la capa extra.

En validación y experimentos: qué registrar para poder comparar

Un print() por epoch sirve para ver un entrenamiento en marcha, no para comparar la corrida de hoy con la de la semana pasada: para eso hace falta un registro persistente y consultable. MLflow Tracking es la pieza estándar —log_param() para hiperparámetros, log_metric(key, value, step=epoch) para valores por epoch—, pero instalarlo y conectarlo a un bucle real queda fuera de este mapa por etapas: el tutorial de MLflow del blog cubre esa implementación con detalle, y el tracking con decoradores muestra cómo envolver funciones de entrenamiento para registrar automáticamente sin llamadas manuales repetidas.

Lo mínimo que conviene registrar por experimento, más allá de la métrica final:

• Hiperparámetros completos: learning rate, batch size, arquitectura, seed. Sin esto, un resultado bueno no es reproducible.
• Versión o hash del dataset: la causa más común de "el mismo código da resultados distintos" es que el dataset cambió entre corridas.
• Métrica por step, no solo la final: una corrida que termina bien pero pasó por una zona inestable a mitad de entrenamiento es información, no ruido.
• Commit de código y versiones de dependencias: sin esto, comparar un experimento de hace tres meses con uno de hoy es comparar dos cosas distintas con la misma etiqueta.

Herramientas equivalentes a MLflow (Weights & Biases, Neptune) resuelven el mismo problema con distinto UI y distinto modelo de precio; el punto no es cuál usar, sino que exista un registro persistente antes de necesitar comparar diez corridas a la vez, momento en el que reconstruirlo desde logs sueltos ya no es viable.

En producción: logging estructurado, niveles y qué no loguear

En producción el problema cambia de naturaleza: ya no se trata de instrumentar un bucle que se controla de principio a fin, sino de dejar rastro suficiente en un sistema que corre sin supervisión directa. El módulo logging de la librería estándar sigue siendo la base más estable: cinco niveles de severidad con valores numéricos crecientes —DEBUG (10), INFO (20), WARNING (30), ERROR (40) y CRITICAL (50)— que permiten filtrar por handler sin tocar el código que emite el log. En un pipeline de inferencia, DEBUG es para reproducir un fallo puntual, INFO para el ciclo de vida normal (petición recibida, modelo cargado, inferencia completada), WARNING para anomalías que no rompen la respuesta (fallback a un modelo más simple, timeout ya recuperado), y ERROR/CRITICAL para lo que sí la rompe.

Para configuración rápida sin montar handlers a mano, Loguru es estable y ampliamente usado —versión 0.7.3 publicada en diciembre de 2024, sin una release nueva desde entonces (unos veinte meses a día de hoy)— y resulta más cómodo que logging para contexto estructurado o rotación en una línea; conviene comprobar su changelog antes de sumarlo a un proyecto nuevo, no darlo por mantenido activamente sin mirar. La contrapartida es que logging da control más granular sobre handlers y formatters cuando hace falta enviar niveles distintos a destinos distintos, y no añade una dependencia externa. Ninguna rota sola sin configurarlo: la librería estándar incluye RotatingFileHandler (por tamaño, maxBytes/backupCount) y TimedRotatingFileHandler (por intervalo) en logging.handlers, imprescindibles en cualquier proceso de larga duración para que el disco no se llene.

Para que esos logs sean consumibles por un agregador en vez de solo legibles en una terminal, la salida se estructura como JSON en vez de texto plano, con un Formatter propio:

import json
import logging


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        if record.exc_info:
            payload["exc_info"] = self.formatException(record.exc_info)
        if record.stack_info:
            payload["stack_info"] = self.formatStack(record.stack_info)
        return json.dumps(payload, ensure_ascii=False)


handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])
logger = logging.getLogger(__name__)

En producción, nunca se debe loguear el input crudo del usuario sin sanitizar, el output completo del modelo sin enmascarar, ni credenciales, tokens de API o cadenas de conexión a base de datos: información sensible que un log poco cuidado convierte en una fuga de datos silenciosa. La OWASP Logging Cheat Sheet es explícita en esto y añade a la lista datos personales identificables, claves de cifrado y código fuente de la aplicación: todo eso debe enmascararse, hashearse o excluirse antes de que llegue al log, no después.

Observabilidad en producción: métricas del sistema y señales del modelo

Logging responde a "qué pasó en esta petición concreta"; observabilidad responde a "cómo se está comportando el sistema en agregado". Prometheus recoge series temporales numéricas mediante un modelo de pull sobre HTTP —el propio servicio expone un endpoint de métricas y Prometheus lo consulta a intervalos regulares, en vez de que el servicio empuje cada evento— y se combina de forma habitual con Grafana para la parte de visualización y alertas. Para un pipeline de inferencia, las métricas de sistema típicas son latencia por petición, throughput y tasa de errores; nada de esto es específico de IA, es lo mismo que se mediría en cualquier servicio HTTP.

Lo que sí es específico de un modelo en producción es una capa de señales que Prometheus y Grafana no generan solas, solo visualizan lo que se decide exportar: distribución de las predicciones (si de repente el 95% de las clasificaciones caen en una sola clase, algo cambió, en el modelo o en los datos de entrada), confianza media de la predicción, y tiempo de inferencia por request separado del tiempo total de la petición (para distinguir un modelo que se volvió más lento de una red que se volvió más lenta). Ninguna de estas señales sustituye una evaluación offline periódica contra un conjunto de referencia, pero permite notar el problema en días en vez de en el informe trimestral de accuracy.

Qué monitorizar en cada etapa

Ninguna de estas cuatro capas sustituye a las demás: un callback de entrenamiento no dice nada de un modelo ya desplegado, y un dashboard de Grafana no explica por qué un experimento de hace dos meses no se puede reproducir. El coste real no es instrumentar las cuatro desde el primer día, es descubrir en qué etapa faltaba la instrumentación justo cuando algo ya se rompió ahí.