NannyML: 开源的机器学习模型部署后监控工具

nannyml

NannyML简介

NannyML是一个开源的Python库,专门用于解决机器学习模型部署后的监控问题。它的主要功能包括:

估计模型部署后的性能(无需访问真实标签)
检测数据漂移
智能关联数据漂移警报与模型性能变化

NannyML的核心贡献者开发了多种新颖的算法来估计模型性能,包括:

基于置信度的性能估计(CBPE)
直接损失估计(DLE)

此外,他们还发明了一种基于PCA数据重构的新方法来检测多变量数据漂移。

NannyML专为数据科学家设计,具有以下特点:

易用的接口
交互式可视化
完全模型无关
支持所有表格用例、分类和回归任务

为什么使用NannyML?

使用NannyML可以帮助数据科学家:

实现性能监控和部署后数据科学的闭环
快速理解和自动检测模型的静默失效
对已部署的机器学习模型保持完全可见性和信任

具体好处包括:

消除因不知道模型性能而导致的不眠之夜
随时间分析数据漂移和模型性能
发现模型表现不如预期的根本原因
避免警报疲劳,只在模型性能受到影响时做出反应
在任何环境中轻松设置

NannyML的主要功能

1. 性能估计和监控

当部署模型的实际结果延迟或缺失时,可以使用NannyML的CBPE算法(分类)或DLE算法(回归)来估计模型性能。这些算法可以提供任何您想要的估计指标,如ROC AUC或RMSE。

NannyML还可以在获得真实标签后跟踪机器学习模型的实际性能。

2. 数据漂移检测

NannyML使用基于PCA的数据重构来检测多变量特征漂移。它会监控重构误差的变化,并在超过阈值时记录数据漂移警报。

对于单变量特征漂移,NannyML采用了多种统计检验方法,如Jensen-Shannon距离和L-infinity距离等。这些检验结果会随时间跟踪,并叠加在时间特征分布上。

NannyML还使用相同的统计检验来检测模型输出漂移和目标分布漂移(需要实际值)。

3. 智能警报

由于NannyML可以估计性能,因此可以过滤掉不影响预期性能的数据漂移警报,从而减少警报疲劳。除了将数据漂移问题与性能下降联系起来外,还可以使用NannyML的Ranker根据其他标准对警报进行优先级排序。

快速上手

要开始使用NannyML,首先需要安装它:

pip install nannyml

NannyML依赖于LightGBM,可能需要安装额外的OS特定二进制文件。

下面是一个简单的使用示例:

import nannyml as nml
import pandas as pd

# 加载数据
reference_df, analysis_df, _ = nml.load_us_census_ma_employment_data()

# 设置分块大小
chunk_size = 5000

# 初始化估计器
estimator = nml.CBPE(
    problem_type='classification_binary',
    y_pred_proba='predicted_probability',
    y_pred='prediction',
    y_true='employed',
    metrics=['roc_auc'],
    chunk_size=chunk_size,
)

# 拟合估计器并估计性能
estimator = estimator.fit(reference_df)
estimated_performance = estimator.estimate(analysis_df)

# 可视化结果
figure = estimated_performance.plot()
figure.show()

# 定义特征列
features = ['AGEP', 'SCHL', 'MAR', 'RELP', 'DIS', 'ESP', 'CIT', 'MIG', 'MIL', 'ANC',
       'NATIVITY', 'DEAR', 'DEYE', 'DREM', 'SEX', 'RAC1P']

# 初始化单变量漂移计算器
univariate_calculator = nml.UnivariateDriftCalculator(
    column_names=features,
    chunk_size=chunk_size
)

# 计算漂移
univariate_calculator.fit(reference_df)
univariate_drift = univariate_calculator.calculate(analysis_df)

# 获取漂移最严重的特征
alert_count_ranker = nml.AlertCountRanker()
alert_count_ranked_features = alert_count_ranker.rank(univariate_drift)
print(alert_count_ranked_features.head())

# 绘制前3个特征的漂移结果
figure = univariate_drift.filter(column_names=['RELP','AGEP', 'SCHL']).plot()
figure.show()