求助：Azure ML Python SDK v2下Model Drift与Data Drift实现方案

数据漂移（Data Drift）实现 #

数据漂移检测的核心是对比基线数据集（模型训练/验证用的数据集）与生产数据集的特征分布差异，以下是SDK v2的实现步骤：

1. 初始化ML客户端

from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential

# 替换为你的Azure资源信息
ml_client = MLClient(
    credential=DefaultAzureCredential(),
    subscription_id="your-sub-id",
    resource_group_name="your-rg-name",
    workspace_name="your-workspace-name",
)

2. 创建数据漂移检测器
   需要提前将基线和生产数据集注册为Azure ML的数据资产（Data Asset），确保两者schema完全匹配（列名、数据类型一致）：

from azure.ai.ml.entities import DataDriftDetector

drift_detector = DataDriftDetector(
    name="prod-data-drift-detector",
    description="监控生产用户行为数据与训练基线的漂移",
    baseline_data="baseline-user-data",  # 基线数据资产名称
    target_data="prod-user-data",  # 生产数据资产名称
    features=["age", "transaction_amount", "login_frequency"],  # 指定要监控的特征列
    compute_target="cpu-cluster",  # 用于计算漂移的集群（需提前创建）
    drift_threshold=0.3,  # 漂移阈值，超过则标记为异常
)

# 提交创建请求
ml_client.drift_detectors.create_or_update(drift_detector)

3. 手动触发漂移检测并查看结果

# 运行漂移检测作业
drift_job = ml_client.drift_detectors.run(
    name="prod-data-drift-detector",
    target_data_version="latest"  # 使用最新的生产数据版本
)

# 等待作业完成后获取结果
drift_job = ml_client.jobs.get(drift_job.name)
print(f"整体漂移分数: {drift_job.properties.drift_score}")
print(f"各特征漂移细节: {drift_job.properties.feature_drift_scores}")

4. 配置定期检测
   如果需要自动定时检测，创建检测器时添加schedule参数：

from azure.ai.ml.entities import RecurrenceSchedule

drift_detector = DataDriftDetector(
    # 其他参数同上
    schedule=RecurrenceSchedule(frequency="day", interval=1)  # 每天运行一次
)

模型漂移（Model Drift）实现 #

模型漂移分为预测漂移（模型输出分布变化）和概念漂移（真实标签与预测的匹配度变化），SDK v2支持批量和在线两种场景的检测：

批量场景（离线漂移检测） #

适合定期处理批量生产数据的场景，核心是对比基线预测结果与生产预测结果的分布：

1. 准备基线预测与生产预测数据

• 用基线数据集跑批量推理，将预测结果保存为Data Asset（比如baseline-preds）
• 定期对生产数据跑批量推理，将结果保存为Data Asset（比如prod-preds）

2. 创建模型漂移检测器

model_drift_detector = DataDriftDetector(
    name="model-prediction-drift-detector",
    description="监控模型预测结果的分布变化",
    baseline_data="baseline-preds",
    target_data="prod-preds",
    features=["predicted_label"],  # 监控预测列
    compute_target="cpu-cluster",
    drift_threshold=0.25,
)

ml_client.drift_detectors.create_or_update(model_drift_detector)

3. 运行检测并查看结果

model_drift_job = ml_client.drift_detectors.run(name="model-prediction-drift-detector")
model_drift_job = ml_client.jobs.get(model_drift_job.name)
print(f"模型漂移分数: {model_drift_job.properties.drift_score}")

在线场景（实时/定时监控） #

针对在线部署的模型，可配置自动监控规则，结合生产输入数据与真实标签检测概念漂移：

from azure.ai.ml.entities import Monitor, MonitorMetricThreshold, DataContext

# 定义监控上下文，关联在线端点与数据
data_context = DataContext(
    endpoint_name="user-churn-endpoint",  # 在线端点名称
    target_data_asset="prod-user-inputs",  # 生产输入数据资产
    ground_truth_data_asset="prod-user-labels",  # 真实标签数据资产（需定期更新）
)

# 定义漂移阈值规则
prediction_drift_threshold = MonitorMetricThreshold(
    metric_name="prediction_drift",
    threshold=0.3,
    operator="greater_than"
)

# 创建监控规则，每天运行一次
monitor = Monitor(
    name="online-model-drift-monitor",
    data_context=data_context,
    metric_thresholds=[prediction_drift_threshold],
    schedule="daily",
    compute_target="cpu-cluster"
)

ml_client.monitors.create_or_update(monitor)

关键注意事项 #

• 数据/模型漂移的核心是数据一致性：确保对比的两个数据集schema完全对齐，否则检测结果无效
• 漂移阈值需根据业务场景调整：比如金融风控场景对漂移的敏感度要远高于普通推荐场景
• 漂移分数基于统计方法（PSI、KL散度等），SDK v2会根据数据类型自动选择合适的计算方式

内容的提问来源于stack exchange，提问作者praneeth kumar