用 Evidently、SageMaker AI 和 MLflow 搭一条可追踪的模型监控链路

2026-07-08 40 预计阅读时间: 1 分钟
来源: aws.amazon.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间：10 分钟

模型上线以后，真正的工作才刚开始。输入数据分布会变，业务人群会变，特征管道也可能悄悄变。来源文章讨论的是一条比较务实的路线：用开源 Evidently 生成数据与模型监控报告，用 Amazon SageMaker AI 承载执行与扩展，再把结果组织到 MLflow 里做对比，并通过流水线和漂移通知把监控变成持续机制。

监控不只是看一张准确率曲线

对判别式机器学习模型来说，常见目标是分类、排序、回归这类预测任务。上线后的风险通常来自两类变化：

  • 数据漂移：线上输入特征的分布和训练或基准数据不再一致。
  • 模型表现退化：真实标签回流后，准确率、AUC、F1、RMSE 等指标开始下滑。

Evidently 适合承担“生成报告”的角色。它可以拿参考数据和当前数据做比较，输出数据漂移、列级统计、模型质量等报告。SageMaker AI 适合承担“运行环境和规模化”的角色，例如把监控任务放到 Processing Job 或 Pipeline 里跑。MLflow 则适合做“实验与监控结果台账”：每次监控运行都记录参数、指标、HTML 报告或 JSON 产物，后续可以按时间比较。

这三个组件组合起来，重点不是多加一个仪表盘，而是把监控结果纳入机器学习工程的版本化流程：哪一天开始漂移、哪批数据触发告警、哪次模型发布后指标变化，都应该能被追溯。

一条合理的运行链路

可以把这套方案拆成四步：

  1. 准备 reference dataset：通常来自训练集、验证集，或一段稳定线上数据。
  2. 收集 current dataset：来自近期线上请求、批量预测输入，或带标签的回流数据。
  3. 用 Evidently 生成监控报告：输出 HTML、JSON 或指标摘要。
  4. 用 MLflow 记录结果：把漂移分数、漂移列数量、报告文件、数据窗口等记录为一次 run。

在 SageMaker AI 中，这段逻辑可以被封装成 Processing Job，然后由 SageMaker Pipelines 定时或事件触发。漂移超过阈值时，再发出通知，例如通过 Amazon SNS、EventBridge 或团队已有的告警系统。

一个关键设计点是：不要只保存最终结论。保存报告文件、数据时间窗口、模型版本、特征 schema、阈值配置，才能在问题发生后复盘。

可以这样实践：本地先跑通 Evidently + MLflow

下面示例是一个最小可改造脚本：读取 reference 和 current 两份 CSV，用 Evidently 生成数据漂移报告，并把漂移相关指标与 HTML 报告记录到 MLflow。

运行前需要准备：

  • reference.csv：基准数据。
  • current.csv：当前窗口数据。
  • 两份 CSV 的列名尽量一致。
  • 将 target、prediction 按你的数据实际情况修改；如果只做数据漂移监控，可以先删除相关列配置。
pip install evidently mlflow pandas
python monitor_drift.py

monitor_drift.py：

from pathlib import Path

import mlflow
import pandas as pd
from evidently import Report
from evidently.presets import DataDriftPreset

REFERENCE_PATH = "reference.csv"
CURRENT_PATH = "current.csv"
REPORT_PATH = "evidently_data_drift_report.html"


def main():
    reference = pd.read_csv(REFERENCE_PATH)
    current = pd.read_csv(CURRENT_PATH)

    report = Report(metrics=[DataDriftPreset()])
    result = report.run(reference_data=reference, current_data=current)

    result.save_html(REPORT_PATH)
    summary = result.dict()

    # Evidently 的具体 JSON 结构会随版本演进；生产中建议固定版本并封装解析逻辑。
    drifted_columns = 0
    dataset_drift = False

    for metric in summary.get("metrics", []):
        metric_result = metric.get("result", {})
        if "dataset_drift" in metric_result:
            dataset_drift = bool(metric_result["dataset_drift"])
        if "number_of_drifted_columns" in metric_result:
            drifted_columns = int(metric_result["number_of_drifted_columns"])

    mlflow.set_experiment("sagemaker-model-monitoring")
    with mlflow.start_run(run_name="daily-data-drift-check"):
        mlflow.log_param("reference_path", REFERENCE_PATH)
        mlflow.log_param("current_path", CURRENT_PATH)
        mlflow.log_param("monitor_type", "data_drift")
        mlflow.log_metric("dataset_drift", int(dataset_drift))
        mlflow.log_metric("drifted_columns", drifted_columns)
        mlflow.log_artifact(REPORT_PATH)

    print(f"dataset_drift={dataset_drift}, drifted_columns={drifted_columns}")
    print(f"report={Path(REPORT_PATH).resolve()}")


if __name__ == "__main__":
    main()

如果你已经有 MLflow Tracking Server，可以这样指定地址：

export MLFLOW_TRACKING_URI="https://your-mlflow-tracking-server"
python monitor_drift.py

这段脚本不是 SageMaker 专属代码，但它是放进 SageMaker Processing Job 的核心逻辑。先在本地跑通，可以减少云上调试成本。

放进 SageMaker Pipeline 时要关注什么

在 SageMaker AI 里规模化时，可以这样实践：把监控脚本、依赖文件和阈值配置打包成一个处理步骤。流水线输入包括模型版本、reference 数据位置、current 数据位置和输出报告路径。

一个简化的配置可以长这样，用于表达监控作业需要的环境变量和数据位置：

monitoring_job:
  image_uri: "123456789012.dkr.ecr.us-east-1.amazonaws.com/model-monitoring:latest"
  entrypoint: "python monitor_drift.py"
  inputs:
    reference_data: "s3://your-bucket/monitoring/reference/"
    current_data: "s3://your-bucket/monitoring/current/2025-01-01/"
  outputs:
    reports: "s3://your-bucket/monitoring/reports/2025-01-01/"
  environment:
    MLFLOW_TRACKING_URI: "https://your-mlflow-tracking-server"
    DRIFTED_COLUMNS_THRESHOLD: "5"
    SNS_TOPIC_ARN: "arn:aws:sns:us-east-1:123456789012:model-drift-alerts"

真正落地时，需要把这份配置映射到 SageMaker Processing 或 Pipeline SDK。关键不是 YAML 形式，而是把输入、输出、阈值和 MLflow run 绑定起来，避免监控任务变成一次性脚本。

告警逻辑也应该保持克制。不要一有单列漂移就打爆通知渠道。更实际的规则包括：

  • 漂移列数量超过阈值。
  • 关键特征漂移，例如价格、地域、设备类型、信用分等。
  • 连续多个时间窗口出现漂移。
  • 数据漂移和模型指标下降同时出现。

MLflow 的价值在于比较，而不只是存档

把 Evidently 报告丢到对象存储当然也能查，但 MLflow 的优势是比较不同 run。你可以按日期、模型版本、数据窗口筛选，观察某个特征漂移是否从模型发布后开始，或者某次上游数据修复是否让漂移消失。

建议至少记录这些字段：

  • model_name 和 model_version
  • reference/current 数据窗口
  • 特征 schema 版本
  • 漂移列数量
  • 是否触发 dataset drift
  • 关键模型质量指标
  • Evidently HTML/JSON 报告 artifact

有标签回流时，可以把模型质量也放进同一条 run。没有标签时，就先做数据漂移和预测分布漂移，不要假装已经知道模型效果。

采用建议：先做窄，再做自动化

这套方案的边界很清楚：Evidently 负责报告，SageMaker AI 负责运行和扩展，MLflow 负责记录与比较，通知系统负责把异常推到人面前。它不能替你判断所有业务后果，也不能替代数据质量契约和特征管道测试。

落地时可以按这个顺序推进：

  1. 选一个高价值模型，不要一开始铺满所有模型。
  2. 固定 reference 数据集和监控窗口，例如按天或按小时。
  3. 先记录报告和指标，不急着自动重训。
  4. 用 MLflow 对比 2 到 4 周结果后再调阈值。
  5. 告警只发给能处理问题的人，并附上 MLflow run、报告 artifact 和数据窗口。

监控系统最怕“看起来很完整，但没人相信”。从少量关键指标开始，把每次漂移都变成可追踪、可解释、可行动的记录，比堆满仪表盘更有用。

# AI # AWS # 云平台 # 可观测性 # 安全 # 机器学习

相关推荐