机器学习工程化怎样持续观察模型效果本文围绕“模型效果怎样持续观察”整理检查要点。示例仅用于说明方法请以公开、合成或已脱敏输入复跑。1. 先固定讨论边界机器学习工程化的重点是让一次结论能够被独立复核。数据版本、配置、随机状态和产物位置应当一同记录任何缺项都应视为结论的边界而不是用经验补齐。结论应同时附上适用条件和未覆盖项。若数据、依赖或执行路径发生变化应重新运行验证而不是沿用旧记录。2. 按最小闭环验证可先把问题缩小为固定输入上的单一断言再逐步加入数据切分、训练和服务环节。比较不同方案时保持其余条件不变并把失败样本作为下一轮检查材料。建议先写出可失败的断言再保存输入摘要、配置与结果摘要。这样既便于定位差异也避免在排障材料中保留不必要的内容。3. 参考实现与图示以下片段保留原有技术结构。运行前请替换为本地的非敏感示例并根据依赖版本核对接口。import time import logging import json import numpy as np from typing import Dict, Any, List, Optional # 配置结构化 JSON 记录格式 logger logging.getLogger(ml_observability) logger.setLevel(logging.INFO) handler logging.StreamHandler() handler.setFormatter(logging.Formatter(%(message)s)) logger.addHandler(handler) class ModelInferenceObserver: def __init__(self, sample_rate: float 0.05, alert_null_ratio: float 0.1): 采样观察器防止全量打印记录造成 I/O 阻塞 :param sample_rate: 采样率 0.0 ~ 1.0 :param alert_null_ratio: 特征空值率报警阈值 self.sample_rate sample_rate self.alert_null_ratio alert_null_ratio def observe(self, trace_id: str, feature_names: List[str], features: np.ndarray, predictions: np.ndarray) - Dict[str, Any]: start_time time.time() batch_size, feature_dim features.shape # 1. 检查特征空值率NaN 或 None 侵入检测 nan_mask np.isnan(features) nan_counts np.sum(nan_mask, axis0) null_ratios nan_counts / batch_size anomalous_features {} for idx, ratio in enumerate(null_ratios): if ratio self.alert_null_ratio: anomalous_features[feature_names[idx]] round(float(ratio), 4) if anomalous_features: logger.warning(json.dumps({ event: FEATURE_NULL_ALARM, trace_id: trace_id, anomalies: anomalous_features, timestamp: int(time.time()) })) # 2. 预测概率分布聚合统计 pred_mean float(np.mean(predictions)) pred_std float(np.std(predictions)) metrics_summary { trace_id: trace_id, batch_size: batch_size, pred_mean: round(pred_mean, 4), pred_std: round(pred_std, 4), latency_ms: round((time.time() - start_time) * 1000, 2) } # 3. 动态采样落盘供后续数据漂移分析 if np.random.rand() self.sample_rate: logger.info(json.dumps({ event: FEATURE_SNAPSHOT, trace_id: trace_id, metrics: metrics_summary, feature_summary: {dimension: int(features.shape[1])}, sample_pred: float(predictions[0]) })) return metrics_summary # 模拟推理测试 if __name__ __main__: observer ModelInferenceObserver(sample_rate1.0, alert_null_ratio0.05) # 模拟特征列名与特征输入其中第二列存在缺失值 feat_names [feature_1, feature_2, feature_3] dummy_input np.array([ [0.1, np.nan, 1.0], [0.2, np.nan, 2.0], [0.3, 0.5, 1.0], [0.4, np.nan, 3.0] ]) dummy_preds np.array([0.82, 0.15, 0.65, 0.41]) observer.observe( trace_idexample-trace-id, feature_namesfeat_names, featuresdummy_input, predictionsdummy_preds )import numpy as np import pandas as pd def calculate_psi(baseline: np.ndarray, actual: np.ndarray, bins: int 10) - float: 计算两个连续变量分布之间的 PSI 值 :param baseline: 基准集训练集或上一周的特征分布 :param actual: 当前观测集的特征分布 :param bins: 分箱数量 :return: float PSI 值 # 按照基准集划分等频分箱 quantiles np.linspace(0, 100, bins 1) bin_edges np.percentile(baseline, quantiles) # 微调边界处理极值 bin_edges[0] - 1e-5 bin_edges[-1] 1e-5 # 统计两组数据在各分箱的样本数 baseline_counts, _ np.histogram(baseline, binsbin_edges) actual_counts, _ np.histogram(actual, binsbin_edges) # 计算比例加入极小常数平滑避免除零 eps 1e-4 actual_pct (actual_counts / len(actual)) eps expect_pct (baseline_counts / len(baseline)) eps # 计算 PSI 公式 psi_value np.sum((actual_pct - expect_pct) * np.log(actual_pct / expect_pct)) return float(psi_value) # 验证计算逻辑 train_feature np.random.normal(loc0.0, scale1.0, size10000) online_feature np.random.normal(loc0.5, scale1.2, size2000) # 均值漂移至 0.5 psi_res calculate_psi(train_feature, online_feature) print(f当前特征的在线 PSI 指标为: {psi_res:.4f})4. 复核清单输入是否可公开、合成或完成脱敏。数据版本、依赖版本和运行配置是否可追溯。对比是否使用相同的输入范围与度量定义。失败路径是否有最小复现和可诊断的错误信息。总结“模型效果怎样持续观察”应以清晰的条件和脚本复核。先记录边界再解释结果。
