Skip to main content Skills Marketplace Découvrez et explorez les compétences IA créées par la communauté.
Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Afficher les détails du prompt Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
npx skills add https://github.com/pluginagentmarketplace/custom-plugin-mlops --skill ml-monitoringLa commande reste sur une seule ligne. Faites défiler horizontalement pour la vérifier avant de la copier.
Vous préférez une copie locale ? Téléchargez les fichiers actuellement disponibles dans SkillsMP.
Télécharger Zip Téléchargement... Explorateur de fichiers
6 fichiers name ml-monitoring description Production-grade ML model monitoring, drift detection, and observability sasmp_version 1.3.0 version 2.0.0 bonded_agent 06-monitoring-observability bond_type PRIMARY_BOND input_schema {"type":"object","required":["monitoring_type"],"properties":{"monitoring_type":{"type":"string","enum":["drift_detection","performance_monitoring","alerting","ab_testing","root_cause_analysis"]},"model_type":{"type":"string","enum":["classification","regression","ranking","recommendation","nlp","cv"]},"drift_config":{"type":"object","properties":{"reference_window_days":{"type":"integer","default":30},"detection_method":{"type":"string","enum":["ks_test","psi","wasserstein","chi_square","js_divergence"]},"threshold":{"type":"number","default":0.1}}}}} output_schema {"type":"object","properties":{"status":{"type":"string","enum":["healthy","degraded","critical","unknown"]},"drift_results":{"type":"object"},"alerts":{"type":"array","items":{"type":"object"}},"recommendations":{"type":"array","items":{"type":"string"}},"next_steps":{"type":"array","items":{"type":"string"}}}} validation {"pre_conditions":["model_deployed_and_serving","metrics_endpoint_accessible","baseline_data_available"],"post_conditions":["monitoring_dashboard_updated","alerts_configured","drift_report_generated"]} error_handling {"common_errors":[{"type":"insufficient_baseline_data","recovery":"extend_reference_window"},{"type":"metrics_collection_failure","recovery":"fallback_to_cached_metrics"},{"type":"alert_fatigue","recovery":"adjust_thresholds_dynamically"}]}
ML Monitoring
Production-grade ML model monitoring, drift detection, and observability skill.
Learning Objectives
By mastering this skill, you will be able to:
Implement comprehensive data and model drift detection
Build production alerting systems with actionable notifications
Design and analyze A/B tests for model comparison
Create observability dashboards for ML systems
Perform root cause analysis on model degradation
Module 1: Data Drift Detection
Concept Overview
Data drift occurs when the statistical properties of model inputs change over time, potentially degrading model performance.
Drift Detection Methods
Method Best For Sensitivity Compute Cost KS Test Continuous features High Low PSI Categorical features Medium Low Wasserstein Distribution shape High Medium Chi-Square Categorical Medium Low JS Divergence Probability distributions High Medium
Implementation: Evidently AI Drift Detection
"""
Production-ready drift detection with Evidently AI.
"""
import pandas as pd
from evidently import ColumnMapping
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, TargetDriftPreset
from evidently.metrics import (
DataDriftTable,
DatasetDriftMetric,
ColumnDriftMetric
)
from typing import Dict , Any , Optional
import json
from datetime import datetime
class
DriftDetector
"""Enterprise drift detection system."""
def
__init__
self,
reference_data: pd.DataFrame,
column_mapping: Optional [ColumnMapping] = None ,
drift_threshold: float = 0.1
self
self
or
self
self
def
detect_drift
self,
current_data: pd.DataFrame,
include_target: bool = True
Dict
str
Any
"""
Detect drift between reference and current data.
Args:
current_data: Current production data
include_target: Whether to check target drift
Returns:
Drift detection results with actionable insights
"""
self
self
"timestamp"
"dataset_drift"
"metrics"
0
"result"
"dataset_drift"
"drift_share"
"metrics"
0
"result"
"drift_share"
"drifted_columns"
"severity"
"healthy"
"recommendations"
"metrics"
1
"result"
"drift_by_columns"
for
in
if
"drift_detected"
"drifted_columns"
"column"
"drift_score"
"drift_score"
"stattest_name"
"stattest_name"
"drift_share"
if
0.5
"severity"
"critical"
"recommendations"
"URGENT: Major drift detected. Consider model retraining."
elif
0.2
"severity"
"degraded"
"recommendations"
"WARNING: Moderate drift. Investigate drifted features."
self
return
def
generate_html_report
self,
current_data: pd.DataFrame,
output_path: str
str
"""Generate interactive HTML drift report."""
self
self
return
if
"__main__"
"data/reference_baseline.parquet"
"data/production_latest.parquet"
"label"
"prediction"
"feature_1"
"feature_2"
"feature_3"
"category_a"
"category_b"
0.1
print
2
Exercise 1.1: Custom Drift Detection Task : Implement a custom drift detector for a specific feature type.
from sklearn.metrics.pairwise import rbf_kernel
import numpy as np
def maximum_mean_discrepancy (
X: np.ndarray,
Y: np.ndarray,
gamma: float = 1.0
) -> float :
"""
Calculate MMD between two distributions.
TODO: Implement
1. Calculate K_XX (kernel of X with itself)
2. Calculate K_YY (kernel of Y with itself)
3. Calculate K_XY (cross kernel)
4. Return MMD = mean(K_XX) + mean(K_YY) - 2*mean(K_XY)
"""
pass
np.random.seed(42 )
X_ref = np.random.normal(0 , 1 , (1000 , 5 ))
X_drift = np.random.normal(0.5 , 1.2 , (1000 , 5 ))
X_nodrift = np.random.normal(0.02 , 1.01 , (1000 , 5 ))
Module 2: Model Performance Monitoring
Key Performance Indicators classification_metrics:
primary:
- accuracy
- f1_score
- auc_roc
secondary:
- precision
- recall
- log_loss
business:
- false_positive_cost
- false_negative_cost
regression_metrics:
primary:
- rmse
- mae
- r2_score
secondary:
- mape
- quantile_loss
business:
- prediction_interval_coverage
- business_value_captured
Implementation: Prometheus Metrics Exporter """
Production metrics exporter for ML models.
"""
from prometheus_client import (
Counter, Gauge, Histogram, Summary,
CollectorRegistry, generate_latest
)
from typing import Dict , Any , List
import time
from functools import wraps
import numpy as np
class MLMetricsExporter :
"""Export ML metrics to Prometheus."""
def __init__ (self, model_name: str , model_version: str ):
self .registry = CollectorRegistry()
self .model_name = model_name
self .model_version = model_version
self .prediction_counter = Counter(
'ml_predictions_total' ,
'Total number of predictions' ,
['model_name' , 'model_version' , 'status' ],
registry=self .registry
)
self .prediction_latency = Histogram(
'ml_prediction_latency_seconds' ,
'Prediction latency in seconds' ,
['model_name' , 'model_version' ],
buckets=[0.001 , 0.005 , 0.01 , 0.025 , 0.05 , 0.1 , 0.25 , 0.5 , 1.0 ],
registry=self .registry
)
self .prediction_confidence = Histogram(
'ml_prediction_confidence' ,
'Prediction confidence distribution' ,
['model_name' , 'model_version' , 'predicted_class' ],
buckets=[0.1 , 0.2 , 0.3 , 0.4 , 0.5 , 0.6 , 0.7 , 0.8 , 0.9 , 0.95 , 0.99 ],
registry=self .registry
)
self .feature_value = Summary(
'ml_feature_value' ,
'Feature value distribution' ,
['model_name' , 'feature_name' ],
registry=self .registry
)
self .model_accuracy = Gauge(
'ml_model_accuracy' ,
'Current model accuracy' ,
['model_name' , 'model_version' , 'window' ],
registry=self .registry
)
self .model_f1 = Gauge(
'ml_model_f1_score' ,
'Current model F1 score' ,
['model_name' , 'model_version' , 'window' ],
registry=self .registry
)
self .drift_score = Gauge(
'ml_drift_score' ,
'Current drift score' ,
['model_name' , 'drift_type' , 'feature' ],
registry=self .registry
)
def track_prediction (self, func ):
"""Decorator to track prediction metrics."""
@wraps(func )
def wrapper (*args, **kwargs ):
start_time = time.time()
try :
result = func(*args, **kwargs)
latency = time.time() - start_time
self .prediction_counter.labels(
model_name=self .model_name,
model_version=self .model_version,
status='success'
).inc()
self .prediction_latency.labels(
model_name=self .model_name,
model_version=self .model_version
).observe(latency)
if isinstance (result, dict ) and 'confidence' in result:
predicted_class = str (result.get('class' , 'unknown' ))
self .prediction_confidence.labels(
model_name=self .model_name,
model_version=self .model_version,
predicted_class=predicted_class
).observe(result['confidence' ])
return result
except Exception as e:
self .prediction_counter.labels(
model_name=self .model_name,
model_version=self .model_version,
status='error'
).inc()
raise
return wrapper
def update_performance_metrics (
self,
accuracy: float ,
f1_score: float ,
window: str = "1h"
):
"""Update performance metrics from ground truth."""
self .model_accuracy.labels(
model_name=self .model_name,
model_version=self .model_version,
window=window
).set (accuracy)
self .model_f1.labels(
model_name=self .model_name,
model_version=self .model_version,
window=window
).set (f1_score)
def update_drift_score (
self,
drift_type: str ,
feature: str ,
score: float
):
"""Update drift metrics."""
self .drift_score.labels(
model_name=self .model_name,
drift_type=drift_type,
feature=feature
).set (score)
def get_metrics (self ) -> bytes :
"""Get metrics in Prometheus format."""
return generate_latest(self .registry)
from fastapi import FastAPI, Response
app = FastAPI()
metrics_exporter = MLMetricsExporter(
model_name="fraud_detector" ,
model_version="v2.1.0"
)
@app.get("/metrics"
async def metrics ():
return Response(
content=metrics_exporter.get_metrics(),
media_type="text/plain"
)
@app.post("/predict"
@metrics_exporter.track_prediction
async def predict (request: dict ):
return {"class" : 1 , "confidence" : 0.92 }
Module 3: Alerting System Design
Alert Severity Matrix critical_alerts:
conditions:
- model_accuracy_drop > 15 %
- prediction_latency_p99 > 2s
- error_rate > 5 %
- data_drift_score > 0.5
response_time: immediate
notification: pagerduty + slack
warning_alerts:
conditions:
- model_accuracy_drop > 5 %
- prediction_latency_p95 > 500ms
- error_rate > 1 %
- data_drift_score > 0.2
response_time: 1_hour
notification: slack
info_alerts:
conditions:
- feature_distribution_shift
- traffic_pattern_change
- new_category_detected
response_time: 24_hours
notification: email
Implementation: Intelligent Alerting """
Intelligent alerting system with noise reduction.
"""
from dataclasses import dataclass, field
from typing import List , Optional , Callable , Dict , Any
from datetime import datetime, timedelta
from enum import Enum
import json
class AlertSeverity (Enum ):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
@dataclass
class Alert :
"""Alert data structure."""
id : str
severity: AlertSeverity
title: str
description: str
metric_name: str
current_value: float
threshold: float
timestamp: datetime = field(default_factory=datetime.utcnow)
labels: Dict [str , str ] = field(default_factory=dict )
runbook_url: Optional [str ] = None
@dataclass
class AlertRule :
"""Define an alert rule."""
name: str
metric_name: str
condition: Callable [[float ], bool ]
severity: AlertSeverity
description_template: str
runbook_url: Optional [str ] = None
cooldown_minutes: int = 15
consecutive_failures: int = 3
class AlertManager :
"""Manage alerts with noise reduction."""
def __init__ (self ):
self .rules: Dict [str , AlertRule] = {}
self .active_alerts: Dict [str , Alert] = {}
self .alert_history: List [Alert] = []
self .failure_counts: Dict [str , int ] = {}
self .last_alert_time: Dict [str , datetime] = {}
self .notification_handlers: Dict [AlertSeverity, List [Callable ]] = {
AlertSeverity.INFO: [],
AlertSeverity.WARNING: [],
AlertSeverity.CRITICAL: []
}
def add_rule (self, rule: AlertRule ):
"""Register an alert rule."""
self .rules[rule.name] = rule
self .failure_counts[rule.name] = 0
def add_notification_handler (
self,
severity: AlertSeverity,
handler: Callable [[Alert], None ]
):
"""Add notification handler for severity level."""
self .notification_handlers[severity].append(handler)
def evaluate_metric (
self,
metric_name: str ,
value: float ,
labels: Dict [str , str ] = None
):
"""Evaluate metric against all matching rules."""
for rule_name, rule in self .rules.items():
if rule.metric_name != metric_name:
continue
if rule.condition(value):
self ._handle_failure(rule, value, labels or {})
else :
self ._handle_recovery(rule)
def _handle_failure (
self,
rule: AlertRule,
value: float ,
labels: Dict [str , str ]
):
"""Handle rule failure."""
self .failure_counts[rule.name] += 1
if self .failure_counts[rule.name] < rule.consecutive_failures:
return
last_time = self .last_alert_time.get(rule.name)
if last_time:
cooldown = timedelta(minutes=rule.cooldown_minutes)
if datetime.utcnow() - last_time < cooldown:
return
alert = Alert(
id =f"{rule.name} _{datetime.utcnow().timestamp()} " ,
severity=rule.severity,
title=f"Alert: {rule.name} " ,
description=rule.description_template.format (
value=value,
metric=rule.metric_name
),
metric_name=rule.metric_name,
current_value=value,
threshold=0 ,
labels=labels,
runbook_url=rule.runbook_url
)
self .active_alerts[rule.name] = alert
self .alert_history.append(alert)
self .last_alert_time[rule.name] = datetime.utcnow()
self ._send_notifications(alert)
def _handle_recovery (self, rule: AlertRule ):
"""Handle rule recovery."""
self .failure_counts[rule.name] = 0
if rule.name in self .active_alerts:
del self .active_alerts[rule.name]
def _send_notifications (self, alert: Alert ):
"""Send notifications for alert."""
handlers = self .notification_handlers[alert.severity]
for handler in handlers:
try :
handler(alert)
except Exception as e:
print (f"Notification handler failed: {e} " )
def slack_notification (alert: Alert ):
"""Send Slack notification."""
import requests
color_map = {
AlertSeverity.INFO: "#36a64f" ,
AlertSeverity.WARNING: "#ff9900" ,
AlertSeverity.CRITICAL: "#ff0000"
}
payload = {
"attachments" : [{
"color" : color_map[alert.severity],
"title" : alert.title,
"text" : alert.description,
"fields" : [
{"title" : "Metric" , "value" : alert.metric_name, "short" : True },
{"title" : "Value" , "value" : str (alert.current_value), "short" : True }
],
"footer" : "ML Monitoring System" ,
"ts" : int (alert.timestamp.timestamp())
}]
}
print (f"[SLACK] {alert.severity.value} : {alert.title} " )
def pagerduty_notification (alert: Alert ):
"""Send PagerDuty notification for critical alerts."""
import requests
payload = {
"routing_key" : "YOUR_ROUTING_KEY" ,
"event_action" : "trigger" ,
"dedup_key" : alert.id ,
"payload" : {
"summary" : alert.title,
"severity" : "critical" ,
"source" : "ml-monitoring" ,
"custom_details" : {
"metric" : alert.metric_name,
"value" : alert.current_value,
"description" : alert.description
}
},
"links" : [{
"href" : alert.runbook_url,
"text" : "Runbook"
}] if alert.runbook_url else []
}
print (f"[PAGERDUTY] Critical: {alert.title} " )
alert_manager = AlertManager()
alert_manager.add_rule(AlertRule(
name="high_error_rate" ,
metric_name="prediction_error_rate" ,
condition=lambda x: x > 0.05 ,
severity=AlertSeverity.CRITICAL,
description_template="Error rate {value:.2%} exceeds 5% threshold" ,
runbook_url="https://wiki.company.com/ml-runbooks/high-error-rate" ,
consecutive_failures=3
))
alert_manager.add_rule(AlertRule(
name="model_accuracy_degradation" ,
metric_name="model_accuracy" ,
condition=lambda x: x < 0.85 ,
severity=AlertSeverity.WARNING,
description_template="Model accuracy dropped to {value:.2%}" ,
consecutive_failures=5
))
alert_manager.add_notification_handler(AlertSeverity.CRITICAL, slack_notification)
alert_manager.add_notification_handler(AlertSeverity.CRITICAL, pagerduty_notification)
alert_manager.add_notification_handler(AlertSeverity.WARNING, slack_notification)
Module 4: A/B Testing for ML Models
Statistical Framework experiment_design:
minimum_detectable_effect: 0.02
statistical_power: 0.80
significance_level: 0.05
sample_size_calculation:
formula: "2 * (z_alpha + z_beta)^2 * variance / MDE^2"
traffic_allocation:
control: 50 %
treatment: 50 %
guardrail_metrics:
- latency_p99_ms < 200
- error_rate < 0.01
Implementation: A/B Testing Framework """
Statistical A/B testing framework for ML models.
"""
import numpy as np
from scipy import stats
from dataclasses import dataclass
from typing import Dict , List , Optional , Tuple
from datetime import datetime
import hashlib
@dataclass
class ExperimentConfig :
"""A/B experiment configuration."""
experiment_id: str
control_model: str
treatment_model: str
traffic_split: float = 0.5
min_sample_size: int = 1000
significance_level: float = 0.05
guardrails: Dict [str , Tuple [float , float ]] = None
@dataclass
class ExperimentResult :
"""A/B experiment results."""
experiment_id: str
control_mean: float
treatment_mean: float
control_std: float
treatment_std: float
sample_size_control: int
sample_size_treatment: int
p_value: float
confidence_interval: Tuple [float , float ]
lift: float
is_significant: bool
recommendation: str
class ABTestingFramework :
"""Statistical A/B testing for ML models."""
def __init__ (self, config: ExperimentConfig ):
self .config = config
self .control_metrics: List [float ] = []
self .treatment_metrics: List [float ] = []
self .guardrail_violations: Dict [str , List [float ]] = {}
def assign_variant (self, user_id: str ) -> str :
"""
Deterministically assign user to variant.
Uses consistent hashing for stable assignment.
"""
hash_input = f"{self.config.experiment_id} :{user_id} "
hash_value = int (hashlib.md5(hash_input.encode()).hexdigest(), 16 )
normalized = (hash_value % 10000 ) / 10000
if normalized < self .config.traffic_split:
return "control"
else :
return "treatment"
def record_metric (
self,
variant: str ,
metric_value: float ,
guardrail_metrics: Dict [str , float ] = None
):
"""Record metric observation for variant."""
if variant == "control" :
self .control_metrics.append(metric_value)
else :
self .treatment_metrics.append(metric_value)
if guardrail_metrics and self .config.guardrails:
for metric_name, value in guardrail_metrics.items():
if metric_name in self .config.guardrails:
min_val, max_val = self .config.guardrails[metric_name]
if value < min_val or value > max_val:
if metric_name not in self .guardrail_violations:
self .guardrail_violations[metric_name] = []
self .guardrail_violations[metric_name].append(value)
def calculate_sample_size (
self,
baseline_rate: float ,
minimum_detectable_effect: float ,
power: float = 0.8 ,
significance_level: float = 0.05
) -> int :
"""Calculate required sample size per variant."""
z_alpha = stats.norm.ppf(1 - significance_level / 2 )
z_beta = stats.norm.ppf(power)
p1 = baseline_rate
p2 = baseline_rate * (1 + minimum_detectable_effect)
pooled_p = (p1 + p2) / 2
n = (2 * pooled_p * (1 - pooled_p) * (z_alpha + z_beta) ** 2 ) / \
((p1 - p2) ** 2 )
return int (np.ceil(n))
def analyze (self ) -> ExperimentResult:
"""Perform statistical analysis of experiment."""
control = np.array(self .control_metrics)
treatment = np.array(self .treatment_metrics)
control_mean = np.mean(control)
treatment_mean = np.mean(treatment)
control_std = np.std(control, ddof=1 )
treatment_std = np.std(treatment, ddof=1 )
t_stat, p_value = stats.ttest_ind(
treatment, control, equal_var=False
)
n1, n2 = len (control), len (treatment)
se = np.sqrt(control_std**2 /n1 + treatment_std**2 /n2)
z = stats.norm.ppf(1 - self .config.significance_level/2 )
diff = treatment_mean - control_mean
ci = (diff - z*se, diff + z*se)
lift = (treatment_mean - control_mean) / control_mean if control_mean != 0 else 0
is_significant = p_value < self .config.significance_level
if len (control) < self .config.min_sample_size:
recommendation = "CONTINUE: Insufficient sample size"
elif self .guardrail_violations:
recommendation = f"STOP: Guardrail violations detected: {list (self.guardrail_violations.keys())} "
elif is_significant and lift > 0 :
recommendation = f"SHIP: Treatment wins with {lift:.2 %} lift (p={p_value:.4 f} )"
elif is_significant and lift < 0 :
recommendation = f"REVERT: Control wins, treatment shows {lift:.2 %} regression"
else :
recommendation = "NO DECISION: Results not statistically significant"
return ExperimentResult(
experiment_id=self .config.experiment_id,
control_mean=control_mean,
treatment_mean=treatment_mean,
control_std=control_std,
treatment_std=treatment_std,
sample_size_control=len (control),
sample_size_treatment=len (treatment),
p_value=p_value,
confidence_interval=ci,
lift=lift,
is_significant=is_significant,
recommendation=recommendation
)
def run_sequential_analysis (
self,
alpha_spending_function: str = "obrien_fleming"
) -> Dict [str , any ]:
"""
Sequential analysis for early stopping.
Uses alpha spending to control false positive rate.
"""
current_n = len (self .control_metrics) + len (self .treatment_metrics)
max_n = self .config.min_sample_size * 2
information_fraction = current_n / max_n
if alpha_spending_function == "obrien_fleming" :
alpha_spent = 2 * (1 - stats.norm.cdf(
stats.norm.ppf(1 - self .config.significance_level/2 ) /
np.sqrt(information_fraction)
))
else :
alpha_spent = self .config.significance_level * information_fraction
result = self .analyze()
return {
"can_stop_early" : result.p_value < alpha_spent,
"alpha_spent" : alpha_spent,
"information_fraction" : information_fraction,
"current_p_value" : result.p_value,
"result" : result
}
config = ExperimentConfig(
experiment_id="model_v2_vs_v1_conversion" ,
control_model="fraud_detector_v1" ,
treatment_model="fraud_detector_v2" ,
traffic_split=0.5 ,
min_sample_size=5000 ,
guardrails={
"latency_p99_ms" : (0 , 200 ),
"error_rate" : (0 , 0.01 )
}
)
ab_test = ABTestingFramework(config)
np.random.seed(42 )
for i in range (10000 ):
user_id = f"user_{i} "
variant = ab_test.assign_variant(user_id)
if variant == "control" :
converted = np.random.binomial(1 , 0.10 )
else :
converted = np.random.binomial(1 , 0.11 )
ab_test.record_metric(
variant=variant,
metric_value=converted,
guardrail_metrics={"latency_p99_ms" : np.random.normal(100 , 20 )}
)
result = ab_test.analyze()
print (f"Recommendation: {result.recommendation} " )
print (f"Lift: {result.lift:.2 %} " )
print (f"P-value: {result.p_value:.4 f} " )
print (f"95% CI: [{result.confidence_interval[0 ]:.4 f} , {result.confidence_interval[1 ]:.4 f} ]" )
Module 5: Root Cause Analysis
Decision Tree: Model Degradation Model Performance Degraded?
├── YES
│ ├── Check Data Drift
│ │ ├── Drift Detected
│ │ │ ├── Feature drift → Investigate upstream data sources
│ │ │ ├── Target drift → Market/behavior shift
│ │ │ └── Concept drift → Retrain with recent data
│ │ └── No Drift
│ │ ├── Check Infrastructure
│ │ │ ├── Latency spike → Scale resources
│ │ │ ├── Memory pressure → Optimize batch size
│ │ │ └── GPU utilization low → Check data loading
│ │ └── Check Model
│ │ ├── Feature engineering bug → Review pipeline
│ │ ├── Model config changed → Audit deployment
│ │ └── Dependency version → Check requirements
│ └── Check Traffic Patterns
│ ├── Traffic spike → Scale horizontally
│ └── New user segment → Evaluate model coverage
└── NO
└── Continue monitoring
Automated Root Cause Analysis """
Automated root cause analysis for ML systems.
"""
from dataclasses import dataclass
from typing import List , Dict , Optional , Callable
from enum import Enum
import pandas as pd
class RootCause (Enum ):
DATA_DRIFT = "data_drift"
TARGET_DRIFT = "target_drift"
CONCEPT_DRIFT = "concept_drift"
INFRASTRUCTURE = "infrastructure"
MODEL_BUG = "model_bug"
TRAFFIC_PATTERN = "traffic_pattern"
UNKNOWN = "unknown"
@dataclass
class DiagnosticResult :
"""Result of diagnostic check."""
check_name: str
passed: bool
details: Dict
severity: str
remediation: str
@dataclass
class RCAReport :
"""Root cause analysis report."""
primary_cause: RootCause
confidence: float
diagnostics: List [DiagnosticResult]
recommended_actions: List [str ]
timeline: List [Dict ]
class RootCauseAnalyzer :
"""Automated root cause analysis."""
def __init__ (self ):
self .diagnostics: List [Callable ] = []
self .results: List [DiagnosticResult] = []
def add_diagnostic (self, diagnostic_fn: Callable ):
"""Register diagnostic check."""
self .diagnostics.append(diagnostic_fn)
def run_analysis (
self,
metrics_df: pd.DataFrame,
drift_results: Dict ,
infra_metrics: Dict
) -> RCAReport:
"""Run all diagnostics and determine root cause."""
self .results = []
for diagnostic in self .diagnostics:
result = diagnostic(metrics_df, drift_results, infra_metrics)
self .results.append(result)
primary_cause, confidence = self ._determine_cause()
recommendations = self ._generate_recommendations(primary_cause)
timeline = self ._build_timeline(metrics_df)
return RCAReport(
primary_cause=primary_cause,
confidence=confidence,
diagnostics=self .results,
recommended_actions=recommendations,
timeline=timeline
)
def _determine_cause (self ) -> tuple [RootCause, float ]:
"""Determine most likely root cause."""
failed_checks = [r for r in self .results if not r.passed]
if not failed_checks:
return RootCause.UNKNOWN, 0.0
cause_priority = {
"data_drift" : RootCause.DATA_DRIFT,
"target_drift" : RootCause.TARGET_DRIFT,
"infrastructure" : RootCause.INFRASTRUCTURE,
"model_validation" : RootCause.MODEL_BUG,
"traffic" : RootCause.TRAFFIC_PATTERN
}
for check_prefix, cause in cause_priority.items():
matching = [r for r in failed_checks if check_prefix in r.check_name.lower()]
if matching:
severity_score = sum (
{"critical" : 3 , "warning" : 2 , "info" : 1 }.get(r.severity, 0 )
for r in matching
)
confidence = min (0.95 , 0.5 + severity_score * 0.15 )
return cause, confidence
return RootCause.UNKNOWN, 0.3
def _generate_recommendations (self, cause: RootCause ) -> List [str ]:
"""Generate remediation recommendations."""
recommendations = {
RootCause.DATA_DRIFT: [
"1. Identify drifted features using drift report" ,
"2. Investigate upstream data source changes" ,
"3. Consider retraining with recent data window" ,
"4. Implement feature monitoring alerts"
],
RootCause.TARGET_DRIFT: [
"1. Analyze target distribution shift" ,
"2. Check for seasonality or market changes" ,
"3. Update training data with recent labels" ,
"4. Consider online learning approach"
],
RootCause.INFRASTRUCTURE: [
"1. Check resource utilization (CPU/GPU/Memory)" ,
"2. Review recent deployment changes" ,
"3. Scale resources if needed" ,
"4. Check for dependency version conflicts"
],
RootCause.MODEL_BUG: [
"1. Review recent model/pipeline changes" ,
"2. Validate feature engineering logic" ,
"3. Check model artifact integrity" ,
"4. Run model validation tests"
],
RootCause.TRAFFIC_PATTERN: [
"1. Analyze traffic distribution changes" ,
"2. Check for new user segments" ,
"3. Validate model coverage for edge cases" ,
"4. Consider segment-specific models"
],
RootCause.UNKNOWN: [
"1. Review all system logs" ,
"2. Check recent changes across stack" ,
"3. Engage cross-functional team" ,
"4. Consider manual investigation"
]
}
return recommendations.get(cause, recommendations[RootCause.UNKNOWN])
def _build_timeline (self, metrics_df: pd.DataFrame ) -> List [Dict ]:
"""Build timeline of events."""
return [
{"timestamp" : "T-24h" , "event" : "Baseline metrics normal" },
{"timestamp" : "T-12h" , "event" : "First anomaly detected" },
{"timestamp" : "T-6h" , "event" : "Performance degradation confirmed" },
{"timestamp" : "T-0h" , "event" : "RCA initiated" }
]
def check_data_drift (
metrics_df: pd.DataFrame,
drift_results: Dict ,
infra_metrics: Dict
) -> DiagnosticResult:
"""Check for data drift."""
drift_score = drift_results.get("drift_share" , 0 )
return DiagnosticResult(
check_name="data_drift_check" ,
passed=drift_score < 0.2 ,
details={"drift_score" : drift_score, "drifted_features" : drift_results.get("drifted_columns" , [])},
severity="critical" if drift_score > 0.5 else "warning" if drift_score > 0.2 else "info" ,
remediation="Investigate drifted features and consider retraining"
)
def check_infrastructure (
metrics_df: pd.DataFrame,
drift_results: Dict ,
infra_metrics: Dict
) -> DiagnosticResult:
"""Check infrastructure health."""
cpu_util = infra_metrics.get("cpu_utilization" , 0 )
memory_util = infra_metrics.get("memory_utilization" , 0 )
latency_p99 = infra_metrics.get("latency_p99_ms" , 0 )
issues = []
if cpu_util > 90 :
issues.append(f"High CPU: {cpu_util} %" )
if memory_util > 90 :
issues.append(f"High Memory: {memory_util} %" )
if latency_p99 > 500 :
issues.append(f"High Latency: {latency_p99} ms" )
return DiagnosticResult(
check_name="infrastructure_check" ,
passed=len (issues) == 0 ,
details={"cpu" : cpu_util, "memory" : memory_util, "latency_p99" : latency_p99, "issues" : issues},
severity="critical" if len (issues) > 1 else "warning" if issues else "info" ,
remediation="Scale resources or optimize model serving"
)
analyzer = RootCauseAnalyzer()
analyzer.add_diagnostic(check_data_drift)
analyzer.add_diagnostic(check_infrastructure)
Troubleshooting Guide
Common Issues
Issue: High False Positive Rate in Drift Detection
Frequent drift alerts without actual model degradation
Alert fatigue from monitoring system
grep -r "drift_threshold" config/
SELECT date , drift_score, model_accuracy
FROM ml_metrics
WHERE date > NOW() - INTERVAL '7 days'
ORDER BY date ;
Increase drift threshold from 0.1 to 0.15
Implement consecutive failure requirement (3+ failures)
Use statistical significance testing before alerting
Add cooldown periods between alerts
Issue: Delayed Ground Truth Labels
Performance metrics unavailable for hours/days
Cannot detect model degradation quickly
label_delays = df['label_timestamp' ] - df['prediction_timestamp' ]
print (f"Median delay: {label_delays.median()} " )
print (f"P95 delay: {label_delays.quantile(0.95 )} " )
Implement proxy metrics (confidence scores, prediction distribution)
Use weak supervision for faster labeling
Create synthetic ground truth from business rules
Deploy shadow models for comparison
Issue: A/B Test Shows No Significance
P-value stuck above significance level
Experiment running longer than expected
from statsmodels.stats.power import TTestIndPower
power_analysis = TTestIndPower()
required_n = power_analysis.solve_power(
effect_size=0.1 ,
power=0.8 ,
alpha=0.05
)
print (f"Required sample size per group: {required_n} " )
Verify minimum detectable effect is realistic
Increase traffic allocation to experiment
Extend experiment duration
Consider alternative metrics with higher signal
Debug Checklist monitoring_debug_checklist:
metrics_collection:
- [ ] Metrics endpoint returning 200
- [ ] Prometheus scraping successfully
- [ ] No gaps in time series data
- [ ] Labels correctly applied
drift_detection:
- [ ] Reference data loaded correctly
- [ ] Feature columns match between ref and current
- [ ] Statistical tests appropriate for data types
- [ ] Threshold values reasonable
alerting:
- [ ] Alert rules syntax valid
- [ ] Notification channels configured
- [ ] Cooldown periods appropriate
- [ ] Runbooks linked and accessible
ab_testing:
- [ ] Traffic split working correctly
- [ ] User assignment deterministic
- [ ] Metrics tracking both variants
- [ ] Guardrails configured
Quick Reference
Essential Commands
python -m evidently.ui.app --project-path ./evidently_workspace
curl -s localhost:9090/api/v1/query?query=ml_prediction_latency_seconds
docker-compose -f monitoring-stack.yml up -d
python scripts/run_drift_detection.py --date $(date +%Y-%m-%d)
Metric Naming Conventions prometheus_naming:
format: "{domain}_{metric}_{unit}"
examples:
- ml_prediction_latency_seconds
- ml_model_accuracy_ratio
- ml_drift_score_ratio
- ml_feature_value_total
Integration Points upstream_dependencies:
- feature_store: feast
- model_registry: mlflow
- data_warehouse: snowflake
downstream_consumers:
- alerting: pagerduty, slack
- visualization: grafana
- analytics: datadog
Métiers associés SOC
Basé sur la classification professionnelle SOC
