跳至主要内容

使用 Python 中的 NannyML 构建端到端机器学习模型监控工作流

学习一个端到端的工作流，将您在 Jupyter 笔记本中的任意模型用于生产环境中的监控。
· 15分钟 阅读

使用 AI 探索

为什么要监控机器学习模型？

机器学习项目是一个迭代过程。您不可能在 Jupyter 笔记本里训练出一个成功的模型就此收工。即使模型已经上线并可供用户访问，工作也并未结束。部署之后，您还需要持续“照看”它，确保其表现与开发阶段一样优秀。

Zillow 的丑闻就是没有这样做会发生什么的典型例子。2021 年，Zillow 因其用于估算房价的机器学习模型而亏损了惊人的 3.04 亿美元。Zillow 为 7000 多套房产付了高价，随后不得不以更低价格抛售。公司被自家模型“坑了”，并被迫裁员 25%。

这类无声的模型失效在真实世界的模型中很常见，因此在生产环境的性能下滑前，模型需要不断更新。否则就会损害公司的声誉、与利益相关者的信任，最终伤及公司利润。

本文将教您如何使用 NannyML 在部署后实施端到端的机器学习模型监控工作流。

什么是 NannyML？

NannyML 是一个专注于部署后机器学习的开源库，正在不断发展。它提供了丰富的功能，帮助解决生产级 ML 环境中出现的各种问题。举例如下：

  • 漂移检测：检测训练数据与生产数据之间的数据分布变化。
  • 性能估计：在没有即时真实标签的情况下估计模型在生产环境的性能。
  • 自动化报告：生成关于已部署模型健康状况和性能的报告。
  • 告警系统：针对数据漂移和性能问题提供告警。
  • 模型公平性评估：监控模型公平性，防止偏见。
  • 与 ML 框架兼容：可与所有机器学习框架集成。
  • 用户友好接口：提供与 scikit-learn 类似的熟悉接口。

我们将逐一学习这些功能的技术细节。

前置概念回顾

我们将通过“机器人练习射箭”的类比来学习模型监控的基本概念。

在这个类比中：

  • 机器人代表我们的机器学习模型。
  • 靶子代表模型的目标或任务。
  • 我们可以将其视为一个回归问题，因为得分基于箭矢距离红心（靶心中央的红点）的接近程度来计算。
  • 箭和弓的特性，以及机器人的物理属性和环境条件（如风力和天气），就是我们模型的特征或输入变量。

那么，我们开始吧。

数据漂移

想象我们已经精心准备了弓、箭和靶（就像数据准备）。我们的机器人配备了许多传感器和摄像头，在训练中射击了 10000 次。随着时间推移，它开始频繁命中红心。我们对表现非常满意，开始将机器人及其复制品卖给射箭爱好者（部署模型）。

但很快，投诉接踵而至。一些用户报告机器人完全射不中靶子。我们很惊讶，召集团队讨论问题出在哪。

我们发现这是一个典型的数据漂移案例。机器人所处的环境发生了变化——不同的风向模式、变化的湿度水平，甚至箭（重量、平衡）和弓的物理特性也变了。

这种输入特征在真实世界中的变化打乱了机器人的准确性，就像当输入数据，尤其是特征之间的关系随时间变化时，机器学习模型可能表现不佳一样。

概念漂移

在解决这些问题后，我们发布了新一批机器人。然而几周后，又出现了类似的投诉。我们困惑之下继续深挖，结果发现用户经常更换靶子。

这些新靶子的大小不同，放置距离也不一。这种变化要求机器人采用不同的射击方法——这正是教科书式的概念漂移。

在机器学习中，概念漂移是指输入变量与目标输出之间的关系发生变化。对于我们的机器人来说，新类型的靶子意味着它们需要用不同的方式射击；同理，当训练数据的动态发生显著变化时，机器学习模型也需要进行调整。

更多真实世界中的概念漂移与数据漂移示例

为了加深理解，我们来看一些数据漂移与概念漂移在现实中的发生方式。

数据漂移示例

  1. 信用评分模型：经济变动会改变人们的消费和信贷习惯。如果信用评分模型无法适应，可能导致不合理的拒贷或高风险的批准。
  2. 健康监测系统：在医疗领域，用户人群结构或传感器校准的变化，可能导致监测病人生命体征的模型给出不准确的健康评估。
  3. 零售需求预测：在零售中，消费者行为和潮流的变化，可能使基于历史销量数据的模型无法有效预测当前产品需求。

概念漂移示例

  1. 社交媒体内容审核：内容审核模型必须持续适应不断演变的语言与文化现象，否则可能误判何为不当内容。
  2. 自动驾驶车辆：自动驾驶汽车中的模型必须针对不同地区的交通规则与路况进行更新，才能获得最佳表现。
  3. 欺诈检测模型：随着欺诈手段迭代，欺诈检测模型需要更新以识别新出现的模式。

现在，让我们看看一个端到端的 ML 监控工作流。

端到端的机器学习模型监控工作流是什么样的？

模型监控包含三个主要步骤，ML 工程师需要反复迭代执行。

1. 监控性能

第一步当然是在部署中密切关注模型性能。但说起来容易做起来难。

当生产中的模型可以立即获得真实标签时，检测模型行为变化就容易得多。比如在我们的机器人/射箭类比中，用户可以直接看靶子，告诉我们机器人没射中——这就是即时的真实标签。

相反，以预测贷款违约的模型为例。这类模型每月预测用户是否会在下期还款中违约。要验证预测，模型必须等到实际还款日。这就是延迟真实标签的例子，在真实世界的机器学习系统中最为常见。

在这种情况下，等待真实标签到来再判断模型表现往往代价高昂。因此，ML 工程师需要无需真实标签即可估计模型性能的方法。这时 CBPE 或 DLE 等算法就派上用场（稍后详述）。

监控模型也可以通过衡量直接的业务影响来实现，即监控 KPI（关键绩效指标）。在 Zillow 的案例中，完善的监控系统本可以检测到利润下滑并向工程师发出告警（假设）。

2. 根因分析

如果监控系统检测到性能下降，不论是基于已实现的性能（有真实标签）还是估计的性能（无真实标签），ML 工程师都必须找出下滑背后的原因。

这通常包括单独或组合地检查特征是否发生数据（特征）漂移，并检查目标是否出现概念漂移。

根据分析结果，他们会采用多种问题修复技术。

3. 问题修复

以下是缓解部署后性能下降所造成影响的一些常见（非穷尽）技术：

  1. 数据再平衡：如果性能下滑源于数据漂移，可调整训练数据集以反映当前条件。
  2. 特征工程：更新或创建新特征可以提升模型性能。这在概念漂移（输入与输出关系发生变化）情况下是个不错的方法。
  3. 模型重训：成本更高的一种方法是用新数据重训模型以确保其保持准确。这对数据漂移和概念漂移都有效。
  4. 模型微调：与从头重训不同，部分模型可以在近期数据集上进行微调。这对深度学习和生成式模型很适用。
  5. 异常检测：使用异常（新奇）检测方法可以更早识别生产数据中的异常模式。
  6. 引入领域专家：邀请领域专家参与，往往能深入洞察模型表现不佳的原因。

每种方法都有其适用场景，很多时候您可能会组合使用。

NannyML 主要覆盖该迭代流程的前两步。那我们就开始动手吧。

步骤一：为 NannyML 准备数据

除了训练集和验证集之外，NannyML 还需要两个额外的数据集，分别称为参考集和分析集，并需采用特定格式以便开始监控。本节将教您如何从任意数据中创建它们。

首先，我们需要一个已经训练好、可随时部署到生产环境的模型，以便进行监控。为此，我们将使用 diamonds 数据集，并训练一个 XGBoost 回归模型。

加载数据、定义特征与目标

import warnings

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns
import xgboost as xgb
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import OneHotEncoder

warnings.filterwarnings("ignore")

导入模块后的第一步是从 Seaborn 加载 Diamonds 数据集。不过，我们将使用本文特制的一个数据集版本，以更好地展示监控是什么样子。您可以使用下面的代码片段将数据集加载到您的环境中：

dataset_link = "https://raw.githubusercontent.com/BexTuychiev/medium_stories/master/2024/1_january/4_intro_to_nannyml/diamonds_special.csv"

diamonds_special = pd.read_csv(dataset_link)
diamonds_special.head()

image8.png

这个特制版本包含一个名为“set”的列，我们稍后会谈到它。

现在，我们先提取所有特征名、类别型特征名以及目标名：

# Extract all feature names
all_feature_names = diamonds_special.drop(["price", "set"], axis=1).columns.tolist()

# Extract the columns and cast into category
cats = diamonds_special.select_dtypes(exclude=np.number).columns

# Define the target column
target = "price"

该任务是回归——我们将根据钻石的物理属性预测价格。

diamonds 数据集相当干净。因此，我们唯一的预处理是将文本特征转换为 Pandas 的 category 类型。这是启用 XGBoost 自动处理类别型数据的要求。

for col in cats:
   diamonds_special[col] = diamonds_special[col].astype("category")

接下来进行数据划分。

将数据划分为四个集合

没错，您没有看错。我们将把数据划分为四个集合。传统上，您可能只会划分为三个：

  • 训练集用于让模型学习模式
  • 验证集用于超参数调优
  • 测试集用于部署前的最终评估

模型监控工作流需要另一个集合来模拟生产数据。这样可以确保我们的系统使用正确的算法检测性能下滑，并报告问题所在。

为此，我在 Diamonds Special 的 set 列中将行标注为四个类别：

diamonds_special.set.unique()
['train', 'val', 'test', 'prod']
Categories (4, object): ['prod', 'test', 'train', 'val']

训练集占 70%，其余各占 10%。让我们进行划分：

tr = diamonds_special[diamonds_special.set == "train"].drop("set", axis=1)
val = diamonds_special[diamonds_special.set == "validation"].drop("set", axis=1)
test = diamonds_special[diamonds_special.set == "test"].drop("set", axis=1)
prod = diamonds_special[diamonds_special.set == "prod"].drop("set", axis=1)

tr.shape
(37758, 10)

不过，真实世界的数据集不会自带集合标签，因此您需要手动将数据划分为四个集合。下面是一个使用 sklearn 的 train_test_split 来完成任务的函数：

def split_into_four(df, train_size=0.7):
   """
   A function to split a dataset into four sets:
   - Training
   - Validation
   - Testing
   - Production
   train_size is set by the user.
   The remaining data will be equally divided between the three sets.
   """
   # Do the splits
   training, the_rest = train_test_split(df, train_size=train_size)
   validation, the_rest = train_test_split(the_rest, train_size=1 / 3)
   testing, production = train_test_split(the_rest, train_size=0.5)

   # Reset the indices
   sets = (training, validation, testing, production)
   for set in sets:
       set.reset_index(inplace=True, drop=True)

   return sets


tr, val, test, prod = split_into_four(your_dataset)

注意：使用“Explain code”按钮可以获得该函数的逐行讲解。

现在，进入模型训练。

训练模型

在训练 XGBoost 模型前，我们需要将数据集转换为 DMatrix。代码如下：

dtrain = xgb.DMatrix(tr[all_feature_names], label=tr[target], enable_categorical=True)
dval = xgb.DMatrix(val[all_feature_names], label=val[target], enable_categorical=True)

dtest = xgb.DMatrix(
   test[all_feature_names], label=test[target], enable_categorical=True
)
dprod = xgb.DMatrix(
   prod[all_feature_names], label=prod[target], enable_categorical=True
)

下面是使用已调优超参数训练回归器的代码：

# Define optimized parameters
params = {
   "n_estimators": 10000,
   "learning_rate": 0.1,
   "tree_method": "gpu_hist",
   "max_depth": 6,
   "min_child_weight": 1,
   "gamma": 0,
   "subsample": 0.8,
   "colsample_bytree": 0.8,
   "objective": "reg:squarederror",
   "reg_alpha": 0.01,
   "reg_lambda": 1,
}

# Training with early stopping
regressor = xgb.train(
   params,
   dtrain,
   num_boost_round=10000,
   evals=[(dtrain, "train"), (dval, "eval")],
   early_stopping_rounds=50,
   verbose_eval=500,
)
[0]	train-rmse:3638.77217	eval-rmse:0.00000
[49]	train-rmse:502.61376	eval-rmse:0.00000

不错——我们在验证集上的 RMSE 达到了 503 美元。让我们在测试集上做一次最终评估：

from sklearn.metrics import mean_squared_error

# Predict on the test set
y_test_pred = regressor.predict(dtest)

# Evaluate
mean_squared_error(test[target], y_test_pred, squared=False)
551.2763827068798

测试表现为 551 美元。已经够好了。

创建参考集

到目前为止，一切都很直观。现在进入重点——创建参考集与分析集。

参考集是在模型监控语境下对测试集的另一种称呼。NannyML 使用模型在测试集上的表现作为生产性能的基线。参考集除了特征外必须包含两列：

  • 目标本身——真实标签——钻石价格
  • 测试时的预测值——我们已生成在 y_test_pred 中

目前我们的测试集包含特征和目标，但还缺少 y_test_pred：

test.columns  # Ignore the `set` column
Index(['carat', 'cut', 'color', 'clarity', 'depth', 'table', 'price', 'x', 'y',
      'z'],
     dtype='object')

让我们把它加上：

test["y_pred"] = y_test_pred

现在，我们将测试集重命名为 reference：

reference = test.copy(deep=True)

reference.columns
Index(['carat', 'cut', 'color', 'clarity', 'depth', 'table', 'price', 'x', 'y',
      'z', 'y_pred'],
     dtype='object')

创建分析集

此时，让我们假设回归器已经部署到云端。相比真正部署，这样的假设对本文来说更为简单。

部署我们的钻石定价模型后，我们收到消息称将有一批大型钻石货物即将到达。在货物抵达前，这批钻石的物理测量数据作为 prod（仍在假设中）已发送给我们，以便我们为其生成价格并在网站上开始营销。那么，我们就来生成价格吧。

# Generate prices for production data
y_prod_pred = regressor.predict(dprod)

在真实钻石到达并由人工专家核验我们的模型生成的价格之前，我们必须检查模型的表现。我们不希望在网站上展示价格不准确的钻石。

为此，我们需要通过比较 y_prod_pred 与新钻石的实际价格（真实标签）来衡量模型性能。但在价格被核验之前，我们并没有真实标签。因此，我们需要在没有真实标签的情况下估计模型性能。

要完成此任务，NannyML 需要一个分析集——包含生产数据以及模型对其做出的预测的数据。

创建分析集与创建 reference 类似：

# Add the predictions of new diamonds to prod
prod["y_pred"] = y_prod_pred

analysis = prod.copy(deep=True)
analysis.columns
Index(['carat', 'cut', 'color', 'clarity', 'depth', 'table', 'price', 'x', 'y',
      'z', 'y_pred'],
     dtype='object')

现在，我们可以开始估计 regressor 的性能了。

步骤二：在 NannyML 中估计性能

NannyML 提供了两类主要算法来估计回归与分类模型的性能：

  • Direct Loss Estimation（DLE，直接损失估计）：用于回归
  • Confidence-Based Performance Estimation（CBPE，基于置信度的性能估计）：用于分类

我们的任务将使用 DLE。DLE 可以在没有真实标签的情况下衡量生产模型的性能，并报告 RMSE、RMSLE、MAE 等多种回归伪指标。

要使用 DLE，我们首先需要在 reference 上进行拟合，以建立性能基线。

在 NannyML 中使用 DLE 进行性能估计

import nannyml  # pip install nannyml

estimator = nannyml.DLE(
   feature_column_names=all_feature_names,
   y_true=target,
   y_pred="y_pred",
   metrics=["rmse"],
   chunk_size=250,
)

初始化 DLE 需要三个参数——输入特征名、包含测试真实标签的列名，以及包含测试预测的列名。

此外，我们还传入了 RMSE 作为度量和 250 的块大小。让我们将估计器拟合到 reference，并在 analysis 上进行估计：

# Fit to the reference set
estimator.fit(reference)

# Estimate on the analysis set
estimated_results = estimator.estimate(analysis)
estimated_results
<nannyml.performance_estimation.direct_loss_estimation.result.Result at 0x7f3954af1c90>

我们获得了一个 NannyML 的 Result 对象，它可以被可视化绘制。来看看图：

estimated_results.plot().show()

image1.png

解释一下该图——它分为两个部分，分别显示参考集和分析集上的性能。如果估计的生产 RMSE 超出阈值，NannyML 会将其标记为告警。

如我们所见，生产数据出现了几次告警，说明在最近几个批次中有些不对劲。

步骤三：监控中的估计性能 vs. 实现性能

我们的监控系统告诉我们，模型在生产中的性能大约下降了一半。但这只是估计——我们还不能完全确定。

在我们绘制估计性能的同时，货物已经到达，我们的钻石专家也计算出了它们的实际价格。我们已将其作为 price 存入了 prod。

现在，我们可以将模型的实现性能（实际性能）与估计性能进行对比，看看我们的监控系统是否工作良好。

NannyML 提供了一个 PerformanceCalculator 类来实现这一点：

calculator = nannyml.PerformanceCalculator(
   problem_type="regression",
   y_true=target,
   y_pred="y_pred",
   metrics=["rmse"],
   chunk_size=250,
)

calculator.fit(reference)
realized_results = calculator.calculate(analysis)

该类需要四个参数：

  • problem_type：任务类型是什么？
  • y_true：真实标签是什么？
  • y_pred：到哪里找预测值？
  • metrics：用哪些指标来计算性能？

传入这些参数并在 reference 上拟合计算器后，我们在分析集上进行 calculate。

要对比 realized_results 与 estimated_results，我们再次使用可视化：

estimated_results.compare(realized_results).plot().show()

image5.png

可以看到，估计 RMSE（紫色）与实际表现（实现 RMSE，蓝色）相当接近。

这说明了一点——我们的监控系统工作良好，但模型表现不佳，这从损失上升可以看出。那么，原因是什么？

我们现在就来深入探究。

步骤四：漂移检测方法

如引言所述，模型在生产中失败的最常见原因之一就是漂移。本节我们将聚焦于数据（特征）漂移检测。

漂移检测是模型监控工作流中根因分析步骤的一部分。它通常从多变量漂移检测开始。

多变量漂移检测

最好的多变量漂移检测方法之一是使用 PCA 计算数据重构误差。它效果显著，甚至能捕捉到特征分布中极其微小的漂移。方法概览如下：

1. 使用 PCA 在 reference 上进行拟合，并将其压缩到更低维度——reference_lower。

  • 在此步骤中，由于 PCA 的特性，部分原始数据信息会丢失。

2. 将 reference_lower 解压回原始维度——reference_reconstructed。

  • 由于步骤 1 信息有损，重构数据与 reference 不会完全一致。

3. 计算 reference 与 reference_reconstructed 的差异，并将其命名为数据重构误差——reconstruct_error。

  • reconstruct_error 作为基线，用于对比生产数据的重构误差。

4. 对生产数据的各个批次应用同样的降维/重构流程。

  • 如果生产数据的重构误差高于基线，我们就认为特征已发生漂移。
  • 系统会向我们发送告警，提示需要进一步调查生产数据。

这四个步骤在 NannyML 中被实现为 DataReconstructionDriftCalculator 类。用法如下：

# Initialize
multivariate_calc = nannyml.DataReconstructionDriftCalculator(
   column_names=all_feature_names,
   chunk_size=250,
)

# Fit
multivariate_calc.fit(reference)
# Calculate error
multivariate_results = multivariate_calc.calculate(analysis)

拿到每个数据块（每 250 行）的误差后，我们可以绘图：

multivariate_results.plot().show()

image3.png

如图所示，重构误差非常高，表明特征发生了漂移。我们也可以将该误差与实现性能进行对比：

multivariate_results.compare(realized_results).plot().show(config={"staticPlot": True})

image7.png

可以看到，所有损失的尖峰都对应着重构误差中的告警。

单变量漂移检测

数据重构误差是一个衡量所有特征整体漂移的单一数值。但单个特征的漂移呢？如果我们的数据集包含数百个特征，如何找到漂移最严重的那些并采取相应措施？

这时就需要使用单变量漂移检测方法。NannyML 根据特征类型提供了多种方法：

  • 类别型特征：L-infinity、卡方（Chi2）
  • 连续型特征：Wasserstein、Kolgomor-Smirnov 检验
  • 二者皆可：Jensen-Shannon 距离、Hellinger 距离

这些方法都会将 reference 中各个特征的分布与 analysis 集中的分布进行比较。我们可以在 UnivariateDriftCalculator 类中使用其中的一些（甚至全部）：

univariate_calc = nannyml.UnivariateDriftCalculator(
   column_names=all_feature_names,
   continuous_methods=["wasserstein"],
   categorical_methods=["jensen_shannon"],
   chunk_size=250,
)

univariate_calc.fit(reference)
univariate_results = univariate_calc.calculate(analysis)

唯一必需的参数是 column_names，其余参数可使用 NannyML 的默认设置。但为简单起见，我们对连续与类别特征分别使用 wasserstein 和 jensen_shannon。

目前我们有 11 个特征，直接调用 plot().show() 可能并非最优选择。相反，我们可以使用告警计数排序器，按（考虑所有数据块时）触发告警的次数返回最“问题多”的特征。代码如下：

# Initialize the ranker
alert_count_ranker = nannyml.AlertCountRanker()

# Filter the univariate results for only columns containing the metrics
filtered_univariate = univariate_results.filter(
   continuous_methods=["wasserstein"],
   categorical_methods=["jensen_shannon"],
   only_drifting=False,
)

# Rank the alerts
ranker_results = alert_count_ranker.rank(filtered_univariate)

有了排序结果后，我们可以打印它的头部，因为它是一个 Pandas DataFrame：

ranker_results.head(10)

image4.png

我们可以看到，问题最多的特征是 color 和 depth。这并不奇怪，因为在写这篇文章前，我就是故意让它们发生漂移的。

但如果这是一个真实场景，您就需要花时间针对这些特征进行问题修复。

结语

我认为模型监控之所以迷人，是因为它打破了“有了一个好模型就算完成”的幻觉。随着地球自转、用户模式变化，没有任何模型能长期保持相关性。这使得模型监控成为每位 ML 工程师必备的核心技能。

今天，我们介绍了一个基础的模型监控工作流。我们从基本监控概念讲起，然后直入代码：将数据整理为与 NannyML 兼容的格式；绘制了第一张用于估计模型性能的图，再绘制另一张图将其与实现性能对比；收到数次性能下滑的告警；用多变量漂移检测进行核查；发现严重的特征漂移；再用单变量漂移检测进行复核；最终识别出发生漂移的特征。

遗憾的是，我们在问题修复处按下了暂停键。监控工作流的最后一步超出了本文范围。不过，我有几条很棒的推荐，涵盖这一步以及更多关于模型监控的内容：

这两门课程都出自您能想到的最佳人选——NannyML 的 CEO 与创始人。课程中有许多不容错过的精华内容。

我也推荐阅读 Nanny ML 文档，里面有实操教程。

Bexruz (Bex) Tuychiev

作者

Bexruz (Bex) Tuychiev

LinkedIn

我是一名数据科学内容创作者，拥有超过 2 年的经验，并在 Medium 上拥有较大的读者群。我喜欢用略带讽刺的笔调撰写有关 AI 和机器学习的深度文章——毕竟总得想点办法让这些内容不那么枯燥。我已发表 130 多篇文章，并制作了一门 DataCamp 课程，另有一门正在筹备中。我的内容已被超过 500 万人次阅读，其中有 2 万人在 Medium 和 LinkedIn 上成为关注者。 

主题

Python人工智能机器学习