ML Model Monitoring & Retraining Pipeline
Evidently + MLflow + Grafana | Production MLOps
What You'll Learn
- 1Separate data drift, target drift, and performance drift and pick the right statistical test for each
- 2Compute KS and PSI values and set thresholds that do not trigger false alarm cascades
- 3Design a retrain trigger with cooldown, minimum sample counts, and a human approval gate
- 4Export model health metrics to Prometheus so alerts describe the model, not the host
Project Overview
Problem Statement
ML models degrade over time due to data drift, concept drift, and changing business conditions. Without monitoring, degraded models silently make poor predictions, causing revenue loss and customer dissatisfaction. This system automatically detects drift, alerts stakeholders, and triggers retraining.
Objectives
- Implement comprehensive data and model drift detection
- Build automated retraining pipeline triggered by drift
- Create dashboards for model health monitoring
- Set up alerting for model performance degradation
- Maintain full experiment lineage with MLflow
| Component | Technology |
|---|---|
| Drift Detection | Evidently AI |
| Experiment Tracking | MLflow |
| Monitoring Dashboard | Grafana + Prometheus |
| Alerting | PagerDuty + Slack |
| Pipeline Orchestration | Apache Airflow |
| Data Store | PostgreSQL + S3 |
| Model Registry | MLflow Model Registry |
Design Rationale
Monitoring separates three questions that are easy to conflate: did the inputs change, did the relationship between inputs and target change, and did the outputs get worse. Data drift answers the first with two-sample tests per feature, target drift answers the second once labels arrive, and performance drift answers the third only when ground truth exists, which for churn or lending products can be weeks after the prediction was made.
The Kolmogorov-Smirnov statistic is the default for continuous features because it needs no bins:
A value above 0.10 on a feature merits an alert; above 0.20 the serving distribution has genuinely moved. Binned and categorical features use the population stability index instead, which yields one readable number per column and slices cleanly into a Grafana panel.
Statistical significance alone is a trap. Testing every feature daily at p = 0.05 with no correction produces at least one false alarm almost immediately:
That is a 78.5 percent chance of paging someone for noise in the first month. The system therefore evaluates a 7 day sliding window stepped every 3 days, requires a threshold breach on two consecutive windows, and cools down for 24 hours after any retrain. The cost is a few days of detection latency; the benefit is that alerts mean something.
The retrain decision is a small state machine with hysteresis rather than one threshold:
Cooldown stops a persistently drifting feature from firing hourly, the 1,000 prediction minimum stops a quiet overnight period from reading as collapse, and the scheduled path still fires after 30 days without training so healthy models get refreshed data anyway.
Architecture Diagram
Step-by-Step Implementation
Step 1: Environment Setup
mkdir model-monitoring && cd model-monitoring
pip install evidently mlflow prometheus-client grafana-api
pip install pandas numpy scikit-learn
pip install fastapi uvicorn psycopg2-binary
pip install apache-airflow great-expectations
Step 2: Evidently AI Drift Detection
Set up comprehensive monitoring for data quality, data drift, and model performance.
# src/monitoring/drift_detector.py
import pandas as pd
import numpy as np
from evidently import ColumnMapping
from evidently.report import Report
from evidently.metric_preset import (
DataDriftPreset, DataQualityPreset, TargetDriftPreset,
ClassificationPreset
)
from evidently.test_suite import TestSuite
from evidently.tests import *
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
import json
import logging
logger = logging.getLogger(__name__)
@dataclass
class DriftConfig:
drift_threshold: float = 0.05
missing_threshold: float = 0.1
alert_on_drift: bool = True
reference_period_days: int = 30
class DriftDetector:
def __init__(self, config: DriftConfig = None):
self.config = config or DriftConfig()
self.reference_data = None
def set_reference_data(self, reference_df: pd.DataFrame, target_col: str = None):
self.reference_data = reference_df
self.target_col = target_col
def create_column_mapping(self, df: pd.DataFrame) -> ColumnMapping:
categorical_cols = df.select_dtypes(include=["object", "category"]).columns.tolist()
numerical_cols = df.select_dtypes(include=[np.number]).columns.tolist()
mapping = ColumnMapping(
target=self.target_col if self.target_col in df.columns else None,
numerical_features=[c for c in numerical_cols if c != self.target_col],
categorical_features=categorical_cols,
)
return mapping
def run_drift_analysis(
self, current_df: pd.DataFrame,
report_name: str = "drift_report"
) -> Dict:
if self.reference_data is None:
raise ValueError("Reference data not set. Call set_reference_data first.")
column_mapping = self.create_column_mapping(current_df)
# Data drift report
drift_report = Report(metrics=[
DataDriftPreset(stattest="ks", stattest_threshold=self.config.drift_threshold),
DataQualityPreset(),
TargetDriftPreset() if self.target_col else None,
])
drift_report.run(
reference_data=self.reference_data,
current_data=current_df,
column_mapping=column_mapping,
)
# Save report
report_path = f"reports/{report_name}.html"
drift_report.save_html(report_path)
# Extract results
results = self._extract_results(drift_report)
logger.info(f"Drift analysis complete: {results}")
return results
def run_data_tests(self, current_df: pd.DataFrame) -> Dict:
test_suite = TestSuite(tests=[
TestShareOfDriftedColumns(lt=0.3),
TestColumnDrift(column_name="feature_1", stattest="ks"),
TestColumnNullShare(column_name="feature_1", lt=0.1),
TestColumnValueRange(column_name="feature_1", gt=0, lt=1000),
])
test_suite.run(
reference_data=self.reference_data,
current_data=current_df,
)
return {
"passed": test_suite.as_dict()["summary"]["all_passed"],
"tests": test_suite.as_dict()["tests"],
}
def _extract_results(self, report) -> Dict:
report_dict = report.as_dict()
results = {
"drift_detected": False,
"drifted_columns": [],
"drift_scores": {},
}
for metric in report_dict.get("metrics", []):
if metric.get("metric") == "DatasetDriftMetric":
results["drift_detected"] = metric.get("result", {}).get("drift_detected", False)
elif metric.get("metric") == "ColumnDriftMetric":
col = metric.get("result", {}).get("column_name", "")
drift_score = metric.get("result", {}).get("drift_score", 0)
results["drifted_columns"].append(col)
results["drift_scores"][col] = drift_score
return results
Step 3: Prometheus Metrics Export
Export model metrics to Prometheus for real-time monitoring and alerting.
# src/monitoring/metrics_exporter.py
from prometheus_client import Gauge, Counter, Histogram, start_http_server
import time
from typing import Dict
class ModelMetricsExporter:
def __init__(self, port: int = 8001):
self.port = port
# Define metrics
self.prediction_latency = Histogram(
"model_prediction_latency_seconds",
"Model prediction latency in seconds",
buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0],
)
self.prediction_count = Counter(
"model_predictions_total",
"Total number of predictions",
["model_version", "status"],
)
self.data_drift_score = Gauge(
"model_data_drift_score",
"Current data drift score",
["feature_name"],
)
self.model_accuracy = Gauge(
"model_accuracy",
"Current model accuracy",
["model_version"],
)
self.feature_missing_ratio = Gauge(
"feature_missing_ratio",
"Missing value ratio per feature",
["feature_name"],
)
def start(self):
start_http_server(self.port)
def record_prediction(self, latency: float, model_version: str, success: bool):
self.prediction_latency.observe(latency)
status = "success" if success else "error"
self.prediction_count.labels(model_version=model_version, status=status).inc()
def record_drift_scores(self, drift_scores: Dict[str, float]):
for feature, score in drift_scores.items():
self.data_drift_score.labels(feature_name=feature).set(score)
def record_accuracy(self, accuracy: float, model_version: str):
self.model_accuracy.labels(model_version=model_version).set(accuracy)
Step 4: Automated Retraining Trigger
# src/retraining/trigger.py
import logging
from typing import Dict, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
logger = logging.getLogger(__name__)
class RetrainReason(Enum):
DATA_DRIFT = "data_drift"
PERFORMANCE_DEGRADATION = "performance_degradation"
SCHEDULED = "scheduled"
MANUAL = "manual"
@dataclass
class RetrainTriggerConfig:
drift_threshold: float = 0.1
accuracy_drop_threshold: float = 0.05
min_samples_for_eval: int = 1000
cooldown_hours: int = 24
max_consecutive_failures: int = 3
class RetrainTrigger:
def __init__(self, config: RetrainTriggerConfig = None):
self.config = config or RetrainTriggerConfig()
self.last_retrain_time: Optional[datetime] = None
self.consecutive_failures = 0
def should_retrain(self, monitoring_results: Dict) -> Optional[RetrainReason]:
# Check cooldown
if self.last_retrain_time:
elapsed = datetime.now() - self.last_retrain_time
if elapsed < timedelta(hours=self.config.cooldown_hours):
return None
# Check data drift
if monitoring_results.get("drift_detected", False):
logger.warning("Data drift detected - triggering retrain")
return RetrainReason.DATA_DRIFT
# Check performance degradation
current_accuracy = monitoring_results.get("current_accuracy", 1.0)
baseline_accuracy = monitoring_results.get("baseline_accuracy", 1.0)
if baseline_accuracy - current_accuracy > self.config.accuracy_drop_threshold:
logger.warning("Performance degradation detected - triggering retrain")
return RetrainReason.PERFORMANCE_DEGRADATION
# Check scheduled retrain
if monitoring_results.get("days_since_last_train", 0) > 30:
return RetrainReason.SCHEDULED
return None
def execute_retrain(self, reason: RetrainReason) -> bool:
logger.info(f"Starting retraining due to: {reason.value}")
self.last_retrain_time = datetime.now()
try:
# Trigger Airflow DAG
from airflow.api.client.local_client import Client
client = Client(None, None)
client.trigger_dag(
dag_id="model_retraining",
conf={"reason": reason.value, "timestamp": datetime.now().isoformat()},
)
return True
except Exception as e:
self.consecutive_failures += 1
logger.error(f"Retrain trigger failed: {e}")
return False
Step 5: MLflow Experiment Tracking Integration
# src/tracking/mlflow_tracker.py
import mlflow
import mlflow.sklearn
from typing import Dict, Any, Optional
from datetime import datetime
class MLflowTracker:
def __init__(self, experiment_name: str = "model_monitoring"):
mlflow.set_experiment(experiment_name)
def log_drift_event(self, drift_results: Dict, retrain_triggered: bool):
with mlflow.start_run(run_name=f"drift-check-{datetime.now().strftime('%Y%m%d-%H%M')}"):
mlflow.log_params({
"check_type": "drift_detection",
"drift_detected": drift_results.get("drift_detected", False),
"num_drifted_columns": len(drift_results.get("drifted_columns", [])),
"retrain_triggered": retrain_triggered,
})
for col, score in drift_results.get("drift_scores", {}).items():
mlflow.log_metric(f"drift_score_{col}", score)
if retrain_triggered:
mlflow.set_tag("retrain_reason", "data_drift")
def log_model_performance(
self, model_version: str, metrics: Dict[str, float],
dataset_info: Dict[str, Any]
):
with mlflow.start_run(run_name=f"perf-{model_version}"):
mlflow.log_params({
"model_version": model_version,
"dataset_size": dataset_info.get("size", 0),
"evaluation_date": datetime.now().isoformat(),
})
mlflow.log_metrics(metrics)
def log_retrain_event(
self, reason: str, success: bool,
metrics_before: Dict, metrics_after: Dict
):
with mlflow.start_run(run_name=f"retrain-{datetime.now().strftime('%Y%m%d')}"):
mlflow.log_params({
"reason": reason,
"success": success,
})
for k, v in metrics_before.items():
mlflow.log_metric(f"before_{k}", v)
for k, v in metrics_after.items():
mlflow.log_metric(f"after_{k}", v)
âšī¸
Set up baseline metrics when deploying a new model. All subsequent monitoring should compare against this baseline. Store baseline metrics in a separate, immutable location.
đĄ
Use sliding windows for drift detection to avoid false positives from natural data variation. A 7-day window with a 3-day step provides good stability.
Performance Metrics
| Metric | Target | Description |
|---|---|---|
| Drift Detection Latency | < 5min | Time to detect drift |
| Retrain Pipeline | < 2hr | End-to-end retraining |
| False Alarm Rate | < 5% | Drift false positives |
| Model Refresh Rate | Weekly | Scheduled retraining |
| Monitoring Coverage | 100% | All production models |
What Breaks in Production
The most common failure is monitoring without labels. Accuracy, the metric every alert is written against, exists only when ground truth arrives, and for many products that is months later. In the gap, watch proxies that are available immediately: prediction distribution, score histograms, feature missingness, and request volume. Treat them as leading indicators, never as substitutes for the real metric.
Second, reference data that rots. Comparing today against a reference set drawn last January eventually flags ordinary seasonality as drift, and the team learns to ignore the panel. Refresh the reference quarterly, or use a rolling 30 day reference when the business is seasonal, and stamp every report with the reference version that produced it.
Third, alert fatigue. A threshold tuned to catch everything pages at 3 a.m. for a 0.11 KS movement on a feature that contributes nothing to the model. Rank features by contribution to the score, page only on the top decile, and open a low-priority ticket for the rest so the signal survives contact with a real on-call rotation.
Fourth, retraining with no approval gate. Fully automatic retraining plus fully automatic deployment turns one bad data day into a bad quarter of decisions. Keep a human step between the MLflow run and the Seldon rollout, and require the incumbent-versus-challenger comparison to clear a 5 percent margin before promotion is even offered.
Interview Talking Points
- Drift Types: Data drift (input distribution change) vs concept drift (relationship between input and target changes). Different detection methods for each.
- Evidently AI: Provides comprehensive reports with statistical tests (KS test, PSI, Jensen-Shannon) for drift detection.
- Retraining Strategy: Automatic retraining with human approval gates prevents deploying broken models.
- Prometheus + Grafana: Industry-standard monitoring stack that integrates with existing infrastructure.
- MLflow Integration: Full lineage from monitoring alerts to retraining experiments to deployed models.
- Gradual Rollout: Canary deployments and A/B testing for new model versions reduce deployment risk.
â ī¸
Aggressive drift thresholds lead to frequent, unnecessary retraining. Start with conservative thresholds and tighten based on observed false positive rates.
âšī¸
For online learning scenarios, consider using river or creme libraries for continuous model updates instead of batch retraining.
Common Mistakes
Alerting on raw p-values across every feature each day
Fix: Require a breach on two consecutive windows and only page on the top decile of features by importance.
Comparing against a reference set frozen last January
Fix: Refresh reference data quarterly or use a rolling 30 day window, and stamp each report with its reference version.
Automatic retraining straight through to production
Fix: Keep a human approval gate and require a 5 percent improvement margin before promotion is offered.
Best Practices
- Store baseline metrics in an immutable location on the day a model deploys
- Log every drift event and retrain reason to MLflow so decisions stay auditable
- Watch label-free proxies continuously and treat accuracy as a lagging confirmation
Key Takeaways
- KS above 0.10 on a continuous feature warrants an alert; above 0.20 the distribution has genuinely moved
- Daily p = 0.05 testing gives a 78.5 percent chance of a false alarm within 30 days, so require two consecutive breaches
- Cooldown and a 1,000 prediction minimum turn a noisy signal into a stable retrain trigger
- Labels lag predictions, so score histograms and missingness rates are your leading indicators
Frequently Asked Questions
Data drift or concept drift, which matters more?
Concept drift matters more because it breaks the mapping between features and target, so predictions become wrong even when every input still looks normal. It is also harder to detect, since it usually shows up first as accuracy decay once labels arrive, while data drift is visible immediately from the inputs alone.
How often should a healthy model retrain?
Retrain on evidence, not on a calendar, with a scheduled path only as a backstop: a 30 day maximum age keeps the model close to current data without churning. On stable datasets that means roughly monthly; on fast-moving ones drift and accuracy alerts fire first and the schedule never triggers.
What if ground truth labels never arrive?
Shift to proxy monitoring: track prediction distributions, score histograms, and downstream business actions such as approval or click rates, and validate the proxy against hand-labeled samples on a monthly cadence of a few hundred rows. If even that is impossible, treat the model as unsupervised and retrain on a fixed schedule with shadow evaluation only.
Interview Questions
Q1: Differentiate data drift, target drift, and performance drift with detection methods.
Data drift compares input distributions with KS or PSI and needs no labels. Target drift compares label distributions once they arrive. Performance drift compares metrics such as AUC or F1 against a stored baseline, and is the only one that proves the model got worse.
Q2: How do you keep alert volume under control?
Require the threshold on two consecutive 7-day windows, apply a 24 hour cooldown after any retrain, restrict paging to top-decile features, and route the rest to tickets. We cut pages by roughly 80 percent while catching the same three real incidents.
Q3: Describe a safe fully automated retraining flow.
Drift or accuracy alert triggers an Airflow DAG, which retrains, evaluates against the incumbent, and registers only if it wins by 5 percent. A human approves the promotion, canary serves 5 percent of traffic for 30 minutes, and error-rate alerts roll back automatically.