AB
AB
AiBoss

All Categories

Tutorials

Evidently AI

Tutorials

用 Evidently AI 搭建批量回归模型的性能监控与漂移排查

面向批量推理场景的 Evidently AI 实操教程：从数据准备、模型验证，到生成回归性能报告、按周追踪质量下滑，再用漂移报告定位根因，并说明如何把检查接入 Airflow 或 MLflow 自动运行。

模型上线之后，真正麻烦的问题往往不是「能不能预测」，而是「预测还准不准」。数据分布会变、季节会变、上游采集口径会变，模型却不会自己告诉你它退化了。Evidently AI 就是用来补上这一环的开源工具：它把参考数据与当前数据放在一起对比，生成可视化报告，覆盖模型质量指标、特征漂移、数据分布等维度。这篇教程面向已经会用 Python 训练模型、但还没建立监控习惯的读者，用一个城市共享单车需求预测的回归案例，走完「训练 → 验证 → 上线观察 → 质量下滑 → 根因定位 → 自动化」的完整链路。如果你更想先了解这类工具的整体定位，可以看站内的 Evidently AI 条目，再回到本文动手。

需要先说明一点：本文讨论的是批量模式下的单模型监控，也就是每天或每周跑一次推理、把预测结果落库、再定期评估的场景。实时在线服务、流式特征、多模型编排都不在本文范围内。另外，Evidently 的 API 在不同大版本之间有过调整，本文示例基于较早的 0.6.x 系列接口（Report + metrics + column_mapping 的写法）。如果你安装的是更新版本，函数名与参数可能已经变化，请以官网当前文档为准。

准备工作

环境与依赖

需要一台能跑 Jupyter Notebook 的机器，Python 环境即可，不要求 GPU。核心依赖是 Evidently 本身，另外需要 pandas、scikit-learn 用于数据处理和建模。安装命令：

pip install evidently pandas scikit-learn
pip show evidently

第二条命令用来确认实际装上的版本号。如果你的环境里已经是较新的版本，而你想复现本文的写法，需要显式安装 0.6.x 系列的版本；反过来，如果你想用新接口，就应当对照官网当前文档改写代码，不要直接照抄本文的 Report(...) 调用。

数据

示例使用 UCI 机器学习仓库公开的城市共享单车租赁数据集。这个数据集按小时记录租赁量，同时带有天气、季节、温度、湿度、风速、节假日等字段，非常适合演示「特征会漂移、目标有季节性」的监控场景。下载后得到一个 CSV 文件，读入 pandas 即可。

需要提前想清楚的三件事

在写第一行监控代码之前，先回答三个问题，它们决定了监控怎么搭：

  • 模型怎么部署。 是脚本或 Notebook 里跑、是 Airflow 编排的批量推理流水线、还是暴露成 API 的实时服务。本文是批量流水线。
  • 反馈回路有多长。 预测之后多久能拿到真实值。地图导航的到达时间几乎立刻可知，翻译质量要等用户反馈，而需求预测通常要等到一周结束、真实租赁量统计出来才知道。反馈越慢，越依赖漂移指标做前置预警。
  • 服务有多关键。 推荐系统错一次的代价接近零，医疗场景错一次的代价极高。关键程度决定了阈值该设多严、告警该发给谁。

这三件事没有标准答案，但它们共同决定了一个好监控应有的样子：配置复杂度不超过模型服务本身、覆盖已知的失效模式、不制造告警疲劳、由建模的人自己就能改、并且以后能扩展或替换。

操作步骤

第一步：数据预处理与划分参考集、当前集

读入数据后做三件事。第一，把日期时间列转成 DatetimeIndex，这样后面可以按时间段切片，模拟「某一周的线上数据」。第二，把数据切成 reference（参考）和 current（当前）两部分：参考集用来训练模型，当前集用来模拟上线后的推理。第三，明确哪些列是数值特征、哪些是类别特征，后面做特征漂移分析时要用。

import pandas as pd

df = pd.read_csv("bike_sharing.csv")
df["datetime"] = pd.to_datetime(df["datetime"])
df = df.set_index("datetime").sort_index()

# 参考期：用于训练
reference = df.loc["2011-01-01":"2011-01-31"]
# 当前期：模拟上线后的数据
current = df.loc["2011-02-01":]

numerical_features = ["temp", "atemp", "hum", "windspeed"]
categorical_features = ["season", "holiday", "workingday", "weather"]
target = "count"

字段名请以你实际下载的数据集为准，不同年份的版本列名可能略有差异。

第二步：训练模型并做上线前验证

用一月份的数据训练一个随机森林回归模型。这里只是演示，真实项目里训练窗口应该更长，特征选择和算法选择也应该更有依据。

from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split

features = numerical_features + categorical_features
X = reference[features]
y = reference[target]

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.25, shuffle=False
)

model = RandomForestRegressor(n_estimators=100, random_state=0)
model.fit(X_train, y_train)

训练完不要直接跳到监控。先看测试集表现，这一步决定了你对「生产环境里什么样的误差算正常」的预期。没有这个基准，后面看到误差上升你无法判断是异常还是本来就该这样。

Evidently 需要一个 column_mapping 告诉它每列的角色：哪些是数值特征、哪些是类别特征、哪一列是预测值、哪一列是目标值。

from evidently import ColumnMapping

column_mapping = ColumnMapping(
    target=target,
    prediction="prediction",
    numerical_features=numerical_features,
    categorical_features=categorical_features,
)

注意 prediction 这一列此时还不存在，需要先把模型预测结果拼进 DataFrame：

X_test = X_test.copy()
X_test["prediction"] = model.predict(X_test)
X_test[target] = y_test.values

X_train_eval = X_train.copy()
X_train_eval["prediction"] = model.predict(X_train)
X_train_eval[target] = y_train.values

第三步：生成回归性能报告

Evidently 的 Report 接收一组 metric（指标组件），在 run() 时同时吃参考数据和当前数据，输出一份可视化报告。RegressionPreset() 是一个预设组合，一次性给出回归任务常用的一整套指标和图表。

from evidently.report import Report
from evidently.metric_preset import RegressionPreset

regression_performance_report = Report(metrics=[
    RegressionPreset(),
])

regression_performance_report.run(
    reference_data=X_train_eval.sort_index(),
    current_data=X_test.sort_index(),
    column_mapping=column_mapping,
)

regression_performance_report

在 Notebook 里直接输出对象，会渲染成内嵌的可视化报告。也可以导出成 HTML 文件，存到磁盘后用浏览器打开：

regression_performance_report.save_html("regression_performance.html")

报告里包含的内容大致分几类：

  • 质量指标：平均误差、MAE、RMSE 等，训练集和测试集各一份，可以直接对比。
  • 预测值 vs 实际值散点：点越贴近对角线说明拟合越好。如果训练集和测试集都沿对角线分布，说明模型没有明显过拟合。
  • 误差分布：看误差是否对称、是否近似正态。误差明显偏斜往往意味着模型在系统性地高估或低估某一类样本。
  • 分段分析：按特征区间切分后看哪些区间的误差偏大，帮助发现「整体指标还行但某个天气条件下完全失准」这类问题。

如果做的是分类任务，对应的预设是分类版本，指标会换成准确率、精确率、召回率、混淆矩阵等。

第四步：用全量参考期重训生产模型

验证通过后，用整个参考期（而不只是训练集那 75%）重新训练一次，把数据用满。重训之后建议再跑一次报告做自检，确认没有引入技术性错误。这次可以把 reference_data 设为 None，只评估当前数据：

model.fit(reference[features], reference[target])

reference_eval = reference.copy()
reference_eval["prediction"] = model.predict(reference[features])

sanity_report = Report(metrics=[RegressionPreset()])
sanity_report.run(
    reference_data=None,
    current_data=reference_eval.sort_index(),
    column_mapping=column_mapping,
)
sanity_report

这一步的意义是确认模型「可以出厂」。之后进入生产观察期。

第五步：按周观察生产表现

刚上线的前几周，值得人工逐周看报告，而不是立刻全自动化。目的是校准预期：哪些指标波动属于正常范围、哪些分段需要重点关注、告警阈值大概该定在哪里。

第一周，用当前期第一周的数据跑一次完整报告，和参考期对比。平均误差可能略有上升，但只要还在合理区间内，就不必紧张。

第二周，你已经知道自己每次真正关心的是哪几个图，这时可以定制一份更精简的报告，只保留需要的组件，减少噪音：

from evidently.metrics import (
    RegressionQualityMetric,
    RegressionErrorPlot,
    RegressionErrorDistribution,
)

weekly_report = Report(metrics=[
    RegressionQualityMetric(),
    RegressionErrorPlot(),
    RegressionErrorDistribution(),
])

weekly_report.run(
    reference_data=reference_eval,
    current_data=current.loc["2011-02-07":"2011-02-14"],
    column_mapping=column_mapping,
)
weekly_report.show()

这份精简报告里，质量指标继续变差，误差分布也开始出现偏移的迹象。

第三周，情况明显恶化：误差大幅上升，而且误差分布整体左移，说明模型在系统性地低估需求。这时候光看性能报告已经不够了，需要找原因。

第六步：用漂移报告定位根因

性能下降只是症状，根因通常在上游数据。这个案例里最可疑的是季节性：训练数据只有一月，而二月之后气温、天气类型、租赁模式都在变。用数据漂移报告可以验证这个猜测。

from evidently.metric_preset import DataDriftPreset

drift_report = Report(metrics=[
    DataDriftPreset(),
])

drift_report.run(
    reference_data=reference_eval,
    current_data=current.loc["2011-02-21":"2011-02-28"],
    column_mapping=column_mapping,
)
drift_report.save_html("data_drift.html")

漂移报告会逐列给出参考分布与当前分布的对比，并标注该列是否发生了显著漂移。如果温度、天气类型等列被标记为漂移，而租赁量又恰好与这些列强相关，那么「模型没坏，是世界变了」这个结论就站得住脚。接下来的动作不是调模型参数，而是把新季节的数据补进训练集、重训模型，并考虑在监控里长期跟踪这几列的漂移。

排查顺序建议固定下来：先看性能指标是否真的下降 → 再看是哪些分段下降 → 再看这些分段对应的特征是否漂移 → 最后判断是数据问题还是模型问题。固定顺序能避免每次出事都从头摸索。

一个完整示例

把上面的步骤串成一个可以顺序执行的脚本，从读数据到输出两份 HTML 报告：

import pandas as pd
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split
from evidently import ColumnMapping
from evidently.report import Report
from evidently.metric_preset import RegressionPreset, DataDriftPreset

# 1. 数据
df = pd.read_csv("bike_sharing.csv")
df["datetime"] = pd.to_datetime(df["datetime"])
df = df.set_index("datetime").sort_index()
reference = df.loc["2011-01-01":"2011-01-31"]
current = df.loc["2011-02-01":]

numerical_features = ["temp", "atemp", "hum", "windspeed"]
categorical_features = ["season", "holiday", "workingday", "weather"]
target = "count"
features = numerical_features + categorical_features

# 2. 训练
X_train, X_test, y_train, y_test = train_test_split(
    reference[features], reference[target], test_size=0.25, shuffle=False
)
model = RandomForestRegressor(n_estimators=100, random_state=0)
model.fit(X_train, y_train)

# 3. 列映射
column_mapping = ColumnMapping(
    target=target,
    prediction="prediction",
    numerical_features=numerical_features,
    categorical_features=categorical_features,
)

# 4. 验证报告
X_test = X_test.copy()
X_test["prediction"] = model.predict(X_test)
X_test[target] = y_test.values
X_train_eval = X_train.copy()
X_train_eval["prediction"] = model.predict(X_train)
X_train_eval[target] = y_train.values

report = Report(metrics=[RegressionPreset()])
report.run(
    reference_data=X_train_eval.sort_index(),
    current_data=X_test.sort_index(),
    column_mapping=column_mapping,
)
report.save_html("validation.html")

# 5. 全量重训
model.fit(reference[features], reference[target])
reference_eval = reference.copy()
reference_eval["prediction"] = model.predict(reference[features])

# 6. 生产周报
week3 = current.loc["2011-02-21":"2011-02-28"].copy()
week3["prediction"] = model.predict(week3[features])

prod_report = Report(metrics=[RegressionPreset()])
prod_report.run(
    reference_data=reference_eval,
    current_data=week3,
    column_mapping=column_mapping,
)
prod_report.save_html("production_week3.html")

# 7. 漂移报告
drift_report = Report(metrics=[DataDriftPreset()])
drift_report.run(
    reference_data=reference_eval,
    current_data=week3,
    column_mapping=column_mapping,
)
drift_report.save_html("drift_week3.html")

print("done")

跑完之后目录里会有三个 HTML 文件：validation.html 是上线前验证，production_week3.html 是第三周的性能，drift_week3.html 是第三周的漂移分析。对照着看，就能复现「性能下滑 → 漂移确认 → 重训」这条判断链。

注意事项

版本差异是最大的坑

Evidently 的接口在版本迭代中变化较大。本文的 Report、metric_preset、ColumnMapping 写法对应较早的版本。新版本可能改用不同的入口函数、不同的参数名，甚至把预设拆成更细的组件。动手前先 pip show evidently 确认版本，再对照官网当前文档核对 API。不要因为报 ImportError 就怀疑环境坏了，多半是版本不匹配。

参考集的选择决定监控的有效性

参考集是「正常」的基准。如果参考集本身就不具代表性（比如只覆盖一个季节、一个地区、一类用户），那么后续所有漂移判断都会失真。参考集应当尽量覆盖模型预期会遇到的数据分布，并且在业务发生结构性变化时主动更新。

反馈延迟决定了监控的重心

本文的案例里真实值要等一周才拿到，所以性能指标天然滞后。这种情况下，漂移指标是更早的预警信号：特征分布一偏，即使还没拿到真实值，也能提前判断风险。反过来，如果反馈是即时的，性能指标本身就可以承担主要预警职责。

不要一次盯几百个指标

监控的价值在于让人采取行动。指标太多、阈值太松，结果是告警疲劳，最后没人看。建议从少量核心指标起步，随着对模型行为的理解逐步增加，并且每个告警都要能对应一个明确的处理动作。

人工观察期不能省

刚上线时逐周人工看报告，是为了建立对「正常波动」的直觉。跳过这一步直接上自动化阈值，很容易把阈值定得过松或过紧。等观察几周、对指标波动范围有把握之后，再把检查固化成脚本。

关于自动化

批量场景下，监控通常和推理流水线放在一起：每天新数据到达时跑推理并落库，每周拿到反馈数据后跑一次质量评估。这类任务适合交给 Airflow 之类的编排工具按 DAG 调度，也可以把评估结果记录到 MLflow 里做历史追踪。具体接入方式取决于你现有的调度与实验管理栈，本文不展开；无论选哪种，核心都是把上面脚本里的「读数据 → 跑报告 → 判断阈值 → 通知」封装成一个可重复执行的作业。

价格与功能以官网为准

Evidently 提供开源库和托管服务两种形态，功能范围、配额、定价都会随时间调整。本文只涉及开源库的本地使用，不涉及任何付费方案的具体条款。如果你考虑使用托管版本，请以官网当前公布的信息为准。

Back