模型监控实战：性能追踪/数据漂移/自动告警

模型监控实战：性能追踪/数据漂移/自动告警

1. 监控指标

模型监控指标： ├── 性能指标 │ ├── 准确率/精确率/召回率 │ ├── 延迟（P50/P95/P99） │ └── 吞吐量（QPS） ├── 数据质量 │ ├── 缺失率 │ ├── 异常值比例 │ └── 特征分布变化 ├── 数据漂移 │ ├── 特征漂移（Feature Drift） │ ├── 标签漂移（Label Drift） │ └── 概念漂移（Concept Drift） └── 系统指标 ├── CPU/GPU 使用率 ├── 内存使用 └── 错误率

2. 数据漂移检测

fromscipy.statsimportks_2sampimportnumpyasnpdefdetect_drift(reference,current,threshold=0.05):drift_report={}forcolinreference.columns:ifreference[col].dtypein['float64','int64']:stat,p_value=ks_2samp(reference[col].dropna(),current[col].dropna())else:contingency=pd.crosstab(reference[col],current[col])stat,p_value,_,_=chi2_contingency(contingency)drift_report[col]={'statistic':stat,'p_value':p_value,'drift_detected':p_value<threshold,}returndrift_report

3. 性能追踪

importtimeimportnumpyasnpfromcollectionsimportdequeclassModelMonitor:def__init__(self,window_size=1000):self.predictions=deque(maxlen=window_size)self.latencies=deque(maxlen=window_size)self.ground_truth=deque(maxlen=window_size)deflog_prediction(self,prediction,latency,ground_truth=None):self.predictions.append(prediction)self.latencies.append(latency)ifground_truthisnotNone:self.ground_truth.append(ground_truth)defget_metrics(self):metrics={'avg_latency':np.mean(self.latencies),'p95_latency':np.percentile(self.latencies,95),'prediction_count':len(self.predictions),}ifself.ground_truth:accuracy=np.mean([p==tforp,tinzip(self.predictions,self.ground_truth)])metrics['accuracy']=accuracyreturnmetrics

总结