第 50 课 · 进阶 · 19 分钟

Python 模型监控与漂移检查

在用户察觉之前,先发现模型行为的变化。

模型监控数据漂移基线告警阈值

街对面新开了一家店,销售悄悄变了,代码却一行没改。上个月的模型已经悄悄错了。什么检查能让它早点被发现?

昨天有效的模型,今天可能因为输入或行为变化而悄悄失效。监控把当前窗口与可信基线比较,用阈值控制噪声,并只记录聚合指标而不是个人数据。这些小检查能在漂移变成事故之前发出信号。 你还会用分箱 PSI 比较分布,并区分输入漂移和结果漂移。

把当前输入与基线比较

从一个容易解释的指标开始,例如某个数值特征的均值。保存一段可信时期的基线窗口,再与最近窗口比较。差异只是需要调查的信号,不是模型已经坏掉的证明。

baseline = [10, 12, 14]
live = [13, 15, 17]
shift = sum(live) / len(live) - sum(baseline) / len(baseline)
print(round(shift, 2))

最简单的均值偏移

用基线单位表示差距

原始差值很难跨不同量纲的特征比较。除以基线标准差,可以把偏移表示成熟悉的单位。绝对值越大信号越强,但阈值仍取决于具体业务。

import statistics

baseline = [10, 12, 14]
live = [13, 15, 17]
gap = (sum(live) / len(live) - sum(baseline) / len(baseline)) / statistics.pstdev(baseline)
print(round(gap, 2))

三个单位约等于 1.84 个基线标准差

证据足够才告警

很小的当前样本可能因偶然产生夸张的均值。要求最小样本量、设置阈值,并显式处理基线标准差为零的情况。告警应触发有记录的检查,而不是自动改模型。

import statistics

def should_alert(baseline, live, min_n=5, threshold=1.0):
    if len(live) < min_n:
        return False
    baseline_mean = sum(baseline) / len(baseline)
    baseline_std = statistics.pstdev(baseline)
    if not baseline_std:
        return False
    gap = (sum(live) / len(live) - baseline_mean) / baseline_std
    return abs(gap) >= threshold

baseline = [10, 11, 12, 13, 14]
print(should_alert(baseline, [14, 15, 16, 17, 18]))
print(should_alert(baseline, [15, 16]))

小样本不会触发告警

监控聚合值,而不是身份

记录比率和计数,而不是原始用户文本、提示词或标识符。正例预测率的变化可以暴露输入或模型行为变化,同时不泄露个人数据。保留期限和访问权限要写清楚。

baseline = [0, 0, 0, 0, 1]
live = [0, 0, 1, 1, 1]
base_rate = sum(baseline) / len(baseline)
live_rate = sum(live) / len(live)
print(round(live_rate - base_rate, 2))

记录比率,而不是某个人

分箱 PSI 比较分布

群体稳定性指数比较基线和实时窗口中固定分箱的占比。很小的占比会被替换为下限,以避免除以零。PSI 是筛查信号,不是模型出错的证明;要检查移动最大的分箱。

import math

def shares(values, edges):
    counts = [0] * (len(edges) - 1)
    for value in values:
        for i in range(len(edges) - 1):
            if edges[i] <= value < edges[i + 1]:
                counts[i] += 1
                break
    total = sum(counts)
    return [count / total for count in counts]

edges = [0, 12, 16, 20, 30]
base = shares([10, 12, 14, 16, 18], edges)
live = shares([14, 16, 18, 20, 22], edges)
psi = sum((l - b) * math.log(max(l, 0.001) / max(b, 0.001)) for b, l in zip(base, live))
print([round(x, 2) for x in base])
print([round(x, 2) for x in live])
print(round(psi, 3))

比较占比,而不是身份

区分输入漂移和结果漂移

输入漂移表示特征分布移动了。结果漂移表示标签到达后,模型的准确率或错误成本发生了变化。两者可能同时出现,也可能各自出现,因此都要监控,并让告警阈值与最小样本量挂钩。

def accuracy(actual, predicted):
    return sum(a == p for a, p in zip(actual, predicted)) / len(actual)

baseline_actual = [1, 0, 1, 1, 0, 1]
baseline_pred = [1, 0, 1, 1, 0, 1]
live_actual = [1, 1, 0, 1, 0, 1]
live_pred = [1, 1, 0, 1, 0, 0]
print("baseline", round(accuracy(baseline_actual, baseline_pred), 2))
print("live", round(accuracy(live_actual, live_pred), 2))

输入和标签是不同信号

本课要点

  • 基线定义模型仍被信任时的正常状态。
  • 标准化差距让不同特征的偏移可以比较。
  • 最小样本量和阈值能减少误报。
  • 输入漂移和概念漂移是不同的问题。
  • 监控聚合指标,避免保留原始个人数据。
  • PSI 筛查分布移动;输入漂移和结果漂移回答的是不同问题。

常见错误

  • 只凭一个观测或极短窗口就告警。
  • 把统计偏移当成模型一定出错的证据。
  • 本来记录比率就够了,却保存完整提示词或用户记录。
  • 没有样本量或成本策略,却对每一次微小的 PSI 变化都告警。
轮到你

比较基线 [0, 0, 1, 1] 与当前 [0, 1, 1, 1] 的正例预测率,打印差值并保留两位小数。

动手试试

改改变量、先预测输出,再运行代码。第一次点击“运行代码”时会下载 Python 运行时,可能需要几秒钟。

浏览器内 Python本地运行 · 无需账号
首次运行会从 CDN 下载 Pyodide。修改后点击“运行代码”。

练习本课

12 道题,作答后立即看到解释。

查看 12 道题答案开始练习
← k-means 聚类