用户反馈、漂移与根因定位

更新 2026-10-05

我的学习 →

正在读取学习记录…

漂移是系统遇到的输入、人群或“输入怎样对应正确结果”的关系随时间改变。用户反馈是观察变化的一种信号，例如“问题解决了”或“还没解决”。收到更多差评时，要先问分母是什么、人群有没有变，再检查模型、Prompt、工具和数据版本。

客服系统之前主要服务普通顾客，后来企业顾客占比变高。普通问题容易解决，企业问题涉及专属合同，平均反馈自然可能下降。与此同时，发布的新政策工具又漏了企业条目；人群变化与版本故障可以同时发生，不能只选一个顺眼的解释。

一个简单例子：e4在同一天连续反馈三次“未解决”。按请求统计，它会占三份；按用户统计，它占一份。让反馈声音最大的用户自动拿三票，会把总体体验估计带偏。

用三个切面拆开下降

固定数据有旧窗口6位用户、新窗口8位用户。新窗口共10条请求，e4重复出现三次。程序先将每位用户的解决反馈求平均，再让用户等权参与统计。

随后按普通/企业人群切分，用旧人群比例重新加权新的人群内反馈率。最后在企业人群中按工具t1/t2切版本，定位 policy_missing 的用户，并对两份政策快照执行同一企业查询。

请求反馈先合并为用户，再按人群标准化；企业人群按工具t1和t2切片，回放发现t2缺企业政策 — 图1：换分母、查人群、切版本，再回放具体失败；聚合下降与版本问题各有对应证据。

保存完整程序并运行

数据的每行包含用户ID、人群、工具版本、是否解决与原因。标签来自预置反馈记录；users() 在每位用户内求平均，原因用集合保存。一个用户若在同一窗口跨人群或版本，程序要求先明确处理规则，避免悄悄归进某一边。

在 PowerShell 创建并进入文件夹：

mkdir l7-drift
cd l7-drift

把下面完整代码保存为该文件夹的 drift_lab.py。数据已经写在程序中，直接保存即可运行。

"""按用户汇总反馈，分解人群变化，并用工具快照复现版本问题。"""

from collections import defaultdict
from statistics import mean

# 每行是请求反馈。resolved 是预置的用户反馈标签，非模型自评。
BASE = [
    ("b1", "regular", "t1", True, "ok"),
    ("b2", "regular", "t1", True, "ok"),
    ("b3", "regular", "t1", True, "ok"),
    ("b4", "regular", "t1", True, "ok"),
    ("b5", "enterprise", "t1", True, "ok"),
    ("b6", "enterprise", "t1", False, "manual"),
]
CURRENT = [
    ("u1", "regular", "t1", True, "ok"),
    ("u2", "regular", "t1", True, "ok"),
    ("e1", "enterprise", "t1", True, "ok"),
    ("e2", "enterprise", "t1", True, "ok"),
    ("e3", "enterprise", "t2", True, "ok"),
    ("e4", "enterprise", "t2", False, "policy_missing"),
    ("e5", "enterprise", "t2", False, "policy_missing"),
    ("e6", "enterprise", "t2", False, "policy_missing"),
    ("e4", "enterprise", "t2", False, "policy_missing"),
    ("e4", "enterprise", "t2", False, "policy_missing"),
]
POLICIES = {"t1": {"regular": 7, "enterprise": 30}, "t2": {"regular": 7}}


def users(rows):
    groups = defaultdict(list)
    for user, segment, version, resolved, reason in rows:
        groups[user].append((segment, version, resolved, reason))
    result = []
    for user, feedback in groups.items():
        # 同一用户本次窗口若跨人群或跨版本，需要先定义聚合策略。
        if len({(r[0], r[1]) for r in feedback}) != 1:
            raise ValueError("user crosses segment or version")
        segment, version = feedback[0][:2]
        result.append({"user": user, "segment": segment, "version": version,
                       "resolved": mean(r[2] for r in feedback),
                       "reasons": {r[3] for r in feedback}})
    return result


def rates(rows, key):
    groups = defaultdict(list)
    for row in rows:
        groups[row[key]].append(row["resolved"])
    return {name: (len(values), mean(values)) for name, values in groups.items()}


def standardized(base_users, current_users):
    baseline = rates(base_users, "segment")
    current = rates(current_users, "segment")
    if set(baseline) - set(current):
        raise ValueError("missing current segment")
    # 用旧人群比例加权新的人群内反馈率，排除纯粹的人群占比变化。
    return sum(count / len(base_users) * current[name][1]
               for name, (count, _) in baseline.items())


def lookup(version, segment):
    return POLICIES[version].get(segment)


def main():
    base, current = users(BASE), users(CURRENT)
    print(f"baseline users={len(base)} resolved={mean(r['resolved'] for r in base):.1%}")
    print(f"current requests={len(CURRENT)} request_resolved={mean(r[3] for r in CURRENT):.1%} "
          f"users={len(current)} user_resolved={mean(r['resolved'] for r in current):.1%}")
    for name, (count, rate) in sorted(rates(current, "segment").items()):
        print(f"segment={name} users={count} resolved={rate:.1%}")
    print(f"current_with_baseline_mix={standardized(base, current):.1%}")
    enterprise = [r for r in current if r["segment"] == "enterprise"]
    for version, (count, rate) in sorted(rates(enterprise, "version").items()):
        print(f"enterprise version={version} users={count} resolved={rate:.1%}")
    failed = [r["user"] for r in current if "policy_missing" in r["reasons"]]
    print(f"policy_missing users={','.join(failed)}")
    print(f"replay enterprise: t1={lookup('t1', 'enterprise')}d "
          f"t2={lookup('t2', 'enterprise')}")


if __name__ == "__main__":
    main()

保存后列出文件，确认能看到 drift_lab.py，再运行：

ls
python -X utf8 drift_lab.py

程序输出：

baseline users=6 resolved=83.3%
current requests=10 request_resolved=50.0% users=8 user_resolved=62.5%
segment=enterprise users=6 resolved=50.0%
segment=regular users=2 resolved=100.0%
current_with_baseline_mix=83.3%
enterprise version=t1 users=2 resolved=100.0%
enterprise version=t2 users=4 resolved=25.0%
policy_missing users=e4,e5,e6
replay enterprise: t1=30d t2=None

逐项读懂结果

旧窗口有6位用户，5位反馈解决，因此为83.3%。新窗口10次请求只有5次解决，按请求是50%；合并重复用户后仍是5位解决、8位用户，按用户是62.5%。

新窗口企业6位、解决3位，为50%；普通2位全部解决，为100%。旧窗口企业只占2/6，普通占4/6；新窗口企业占6/8。因此原始总率下降包含人群占比变化。

current_with_baseline_mix=83.3% 用旧比例计算：(4/6)×100% + (2/6)×50% = 83.3%。它回答“如果新窗口仍是旧人群比例，会怎样”；观察到的62.5%仍作为原始指标保留。

企业人群的t1有2位用户、解决率100%；t2有4位、解决率25%。失败原因关联到e4、e5、e6三位。最后回放同一查询，t1查到30天，t2得到 None：源码里的t2字典确实缺企业政策。这为修复这份固定工具快照提供了明确线索。

版本切片本身仍可能受到任务难度与用户分配影响。回放确认了“该工具版本漏条目”的具体机制，没有证明线上所有反馈下降都由这个缺项导致。

改一处，再看对应结果

在 POLICIES 中给t2补上 "enterprise": 30，运行 python -X utf8 drift_lab.py。最后的回放从 t2=None 变为 t2=30；前面的旧反馈比例都保持原值，因为修代码不会重写已经发生的反馈记录。

这条变化把“修复可复现缺项”与“确认用户体验恢复”分开。若接入新运行，需要用修复版本采集后续反馈，再按相同窗口与人群复核。完成后撤回补项并运行测试。

分布变化与机制变化各怎么确认

输入漂移可以先看语言、任务类别、请求长度、人群比例；关系漂移则可能来自政策或业务规则改变，例如过去允许退款的条件现在变了。反馈变化也可能来自收集界面、抽样比例或未回复用户的变化，不能把所有下降都归给模型。

用户反馈应关联到请求或Trace ID，并保留当时的Prompt、模型、工具和数据版本。按错误类别找代表样本，用原数据快照复现；只比较“升级前后总平均”会把多个变量揉在一起。

Google 的 Rules of ML讨论了数据处理差异、随时间变化和反馈循环导致的训练/服务偏差，并建议记录服务时实际使用的特征。Agent系统同样需要保留当时的检索内容和工具结果，而不只保留最终答案。

每位用户平均解决率也有假设：这里让用户等权，更符合“一个用户体验如何”的问题；若问题是客服处理每张工单的负担，应另算工单率。窗口长度、跨版本用户的归属和反馈缺失都需要预先说明。缺少反馈不能直接标为已解决。

保存并运行测试

7 项测试检查请求/用户分母、重复用户权重、旧人群标准化、版本切片、缺项回放、跨版本用户拒绝和缺人群拒绝。

把下面代码保存为同一文件夹的 test_drift_lab.py，确认两个文件都在当前目录，运行：

ls
python -X utf8 -m unittest -v test_drift_lab.py

完整测试：test_drift_lab.py

"""检查重用户去重、人群标准化、版本切片与根因复现。"""

import sys
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
import drift_lab as lab


class DriftTests(unittest.TestCase):
    def test_request_and_user_denominators_differ(self):
        current = lab.users(lab.CURRENT)
        self.assertEqual(len(current), 8)
        self.assertAlmostEqual(lab.mean(r["resolved"] for r in current), .625)
        self.assertAlmostEqual(lab.mean(r[3] for r in lab.CURRENT), .5)

    def test_repeating_failure_does_not_add_users(self):
        current = lab.users(lab.CURRENT)
        fewer = lab.users(lab.CURRENT[:8])
        self.assertEqual(lab.rates(current, "segment"), lab.rates(fewer, "segment"))

    def test_fixed_population_mix(self):
        self.assertAlmostEqual(lab.standardized(lab.users(lab.BASE),
                                                lab.users(lab.CURRENT)), 5 / 6)

    def test_version_slice_locates_three_affected_users(self):
        enterprise = [r for r in lab.users(lab.CURRENT) if r["segment"] == "enterprise"]
        self.assertEqual(lab.rates(enterprise, "version"), {"t1": (2, 1), "t2": (4, .25)})
        self.assertEqual(sum("policy_missing" in r["reasons"] for r in enterprise), 3)

    def test_policy_snapshot_reproduces_missing_entry(self):
        self.assertEqual(lab.lookup("t1", "enterprise"), 30)
        self.assertIsNone(lab.lookup("t2", "enterprise"))

    def test_cross_version_user_requires_explicit_policy(self):
        with self.assertRaises(ValueError):
            lab.users([("u", "regular", "t1", True, "ok"),
                       ("u", "regular", "t2", False, "bad")])

    def test_missing_segment_cannot_be_silently_standardized(self):
        with self.assertRaises(ValueError):
            lab.standardized(lab.users(lab.BASE), lab.users(lab.CURRENT[:2]))


if __name__ == "__main__":
    unittest.main()

继续查文档与论文

阅读 Rules of ML 的 Training-Serving Skew，特别是服务时记录特征与共享处理逻辑的建议。再看 The ML Test Score的监控检查项，思考哪项可映射到自己的工具与检索数据。

课后练习展开习题收起习题

若企业用户全都进入t2，普通用户全都进入t1，总率能区分版本与人群吗？修复工具后，为什么旧反馈不能立即改为成功？

正在读取本地记录…

不调用模型，不执行代码。简答、代码和场景题仅保存答案并展示自查标准，不生成评分、徽章或认证。输入内容不会自动保存，请点击提交；不要填写个人信息或密钥。

练习 1 / 4 · v1.1.0

选择题 · 成功率下降后，先怎样缩小排查范围

成功率下降后，先怎样缩小排查范围？

对应知识点：L7-drift

选择一个答案A. 马上把所有失败归因模型B. 只看总体平均分C. 按任务类型、输入分布和版本分组比较

练习 2 / 4 · v1.1.0

简答题 · 解释机制

数据漂移和行为退步怎样区别观察？

对应知识点：L7-drift

你的回答

练习 3 / 4 · v1.1.0

代码题 · 核对给定数据

退款样本占比从2/10变为6/10，用表达式算两个比例。

对应知识点：L7-drift

你的代码或伪代码

练习 4 / 4 · v1.1.0

场景题 · 分析具体变化

线上多了含口语缩写的查询，但旧基准集全是标准问法，应补什么？

对应知识点：L7-drift

你的回答