1. 本章目标 #

前九章各造了一件零件：

章 零件 脚本
29~32 看得见 Agent 内部发生了什么 ——
33 评测集 build_evalset.py
34 评测器 evaluators.py
35 批量实验 + 报告 report.py
36 回归门禁 gate.py
37 线上巡检 monitor.py

摆在一起看，缺的是最后一根接线：线上巡检发现的问题，怎么变成评测集里的用例。

没有这根线，前面九章是一条开口的流水线——你在发版前反复跑同一批自己造的题，线上真实用户遇到的问题一次都进不来。评测集会慢慢过期：它测的是你三个月前想到的场景，而线上早就跑出了你没想到的。

这一章把口子接上，让它变成环：

        线上巡检 monitor.py（第 37 章）
              │  发现错误 / 差评
              ▼
        收集 harvest.py ◄──────── 本章
              │  人工补参考答案
              ▼
        评测集 build_evalset.py（第 33 章）
              │
              ▼
        评测 + 门禁 gate.py（第 34~36 章）
              │  拦下退化 / 暴露缺陷
              ▼
           修代码
              │
              └──────► 发版，回到线上

先说结论，本章最反直觉的三件事：

  1. 线上错误变成评测用例的衰减率是 79%——14 条线上错误，最后只进了 3 条。多数错误是环境问题（key 过期、限流、超时），收进评测集就是永久失败的噪音。
  2. 参考答案只能人写，而且照抄「线上那一次的输出」写 must_include，会写出一条随机翻转的坏用例。实测某条这么写出来的用例，命中率只有 25%——它有 75% 的概率误报失败。
  3. 闭环真的抓到了一个缺陷：知识库里明明有答案，用户换个说法就检索不到。修完之后那条用例从 33% 涨到 100%。但三条用例的总均值只涨了 12 个百分点，因为那条坏用例一直在稀释信号。

学完这一章你应能：

前置依赖： 第 33 章（数据集与 split）、第 34 章（must_include 与 LLM-as-judge）、第 36 章（三层门禁、过程指标）、第 37 章（error_kind 归一化、list_runs 的坑）。

1.1 本章统一环境 # 

langsmith 0.12.1
模型：deepseek-v4-flash
线上项目：langchain（第 37 章巡检的同一个项目）
评测集：cs-rag-eval-v1，本章开始时 22 条
时间窗口：--hours 720（30 天），覆盖 08-22 ~ 09-03

本章大部分操作不消耗 trace 配额，因为收集阶段全是读。但 §6 的验证要跑 Agent。我在写这一章时 LangSmith 免费额度已经用尽（第 36 章 §6.3 那个 429），所以 §6 的数据是关掉 tracing、直接从返回的 messages 里数出来的——这也演示了一件事：闭环的核心逻辑不该依赖某个 SaaS 能不能连上。

2. 从线上捞什么 #

上线后没有参考答案（第 37 章 §1），所以「哪次回答得不好」这件事，机器判断不了。能自动判断的只有两类信号：

信号 怎么拿 值不值钱
报错 list_runs(error=True) 中。系统崩了，但很多是环境问题
用户差评 filter='eq(feedback_key, "user_thumb"), lt(feedback_score, 1)' 高。系统没崩，但用户不满意——这类问题只有真实用户能发现

差评比报错值钱得多。报错至少你自己能复现，差评对应的是「跑通了但答得不对」，正是评测集最该覆盖、又最难自己想出来的部分。

def gather(project: str, hours: float, feedback_key: str | None):
    """收两类线索：出错的 run，以及被用户打了差评的 run。"""
    since = dt.datetime.now(dt.timezone.utc) - dt.timedelta(hours=hours)
    runs = list(client.list_runs(project_name=project, is_root=True,
                                 error=True, start_time=since))
    tagged = [(r, error_kind(r.error)) for r in runs]

    if feedback_key:
        # 注意 feedback 有 10~15 秒索引延迟（第 33 章 §5.5），刚打的分查不到
        f = (f'and(eq(is_root, true), eq(feedback_key, "{feedback_key}"), '
             f'lt(feedback_score, 1))')
        bad = list(client.list_runs(project_name=project, filter=f,
                                    start_time=since))
        tagged += [(r, "用户差评") for r in bad]
    return tagged

error_kind 直接复用第 37 章 §3.2 那个归一化函数——原始 error 字段里塞着完整 traceback 和随机 UUID，不归一化的话每条错误都是「独一类」，后面的分流表会退化成一行一条。

2.1 提取问题文本：结构不统一 #

拿到 run 之后要从 inputs 里挖出用户问的那句话。麻烦在于线上 trace 的 inputs 结构不统一：

def extract_question(inputs: dict):
    """从 trace 的 inputs 里挖出用户问的那句话。

    Agent 调用是 {messages}，被 @traceable 装饰的函数可能是
    {input} / {query} / 自定义名。
    """
    inputs = inputs or {}
    for key in ("question", "input", "query", "text", "prompt"):
        if isinstance(inputs.get(key), str) and inputs[key].strip():
            return inputs[key]
    msgs = inputs.get("messages")
    if isinstance(msgs, list) and msgs:
        m0 = msgs[0]
        if isinstance(m0, list) and m0:      # 有时是嵌套一层的
            m0 = m0[0]
        if isinstance(m0, dict):
            c = m0.get("content")
            if isinstance(c, str) and c.strip():
                return c
    return None

那个 if isinstance(m0, list) 不是防御性代码写多了——messages 真的会嵌套一层，取决于 trace 是从哪个层级发出来的。实测 14 条里有 1 条彻底挖不出来（inputs 只有一个 {input} 键且值不是字符串）。

这里挖不出来就只能弃。 别写成「挖不出来就存整个 inputs，以后再看」——那个「以后」不会来。

3. 分流：79% 的线上错误不该收 #

拿到候选之后，最重要的一步判断是：这条错误该不该进评测集。

判据只有一句话：

换个环境重新跑一遍，这个错还会不会重现？

会重现的，是输入本身触发的缺陷，收。不会重现的（key 过期、限流、网络抖动），收进去就是评测集里一条永久失败的用例——它会一直挂着，让所有人习惯「有几条一直是红的」，然后真的退化来了也没人看。

TRIAGE = {
    "PIIDetectionError":           ("收", "输入里真有 PII，正是该防的场景"),
    "ValueError":                  ("收", "参数校验失败，多半是输入触发的缺陷"),
    "OutputParserException":       ("收", "模型输出格式不对，可复现"),
    "ModelCallLimitExceededError": ("待定", "可能配额设小了，也可能该输入导致死循环"),
    "RuntimeError":                ("待定", "要看具体信息"),
    "AuthenticationError":         ("弃", "key 过期或失效，与输入无关"),
    "RateLimitError":              ("弃", "限流，不可复现"),
    "TimeoutError":                ("弃", "网络或下游抖动，不可复现"),
    "APIConnectionError":          ("弃", "网络问题"),
}

跑一遍，实测分流结果：

====================================================================
收集线上问题 | langchain | 最近 720h
====================================================================
  出错或差评的根 run：14 条

【1. 提取】13 条能挖出问题文本，1 条挖不出（inputs 结构不认识）

【2. 分流】按错误类型判断该不该收
  类型                            条数  处置   理由
--------------------------------------------------------------------
  AuthenticationError                5     弃   key 过期或失效，与输入无关
  ValueError                         3     收   参数校验失败，多半是输入触发的缺陷
  PIIDetectionError                  2     收   输入里真有 PII，正是该防的场景
  ModelCallLimitExceededError        2   待定   可能配额设小了，也可能该输入导致死循环
  TimeoutError                       1     弃   网络或下游抖动，不可复现
  → 收 + 待定 共 7 条，弃 6 条

【3. 去重】7 条 → 3 条（和数据集重复 3，候选内部重复 1）

【4. 草稿】3 条已写入 harvest_draft.jsonl
    [PIIDetectionError          ] 我的邮箱是 alice@example.com，卡号 4111 1111
    [ModelCallLimitExceededError] 北京今天天气怎么样?
    [ValueError                 ] 查一下A1001订单的状态

漏斗完整走一遍：

阶段 剩余 衰减
线上出错的根 run 14 ——
能提取出问题文本 13 −1（inputs 结构不认识）
分流后值得收 7 −6（5 条 key 失效 + 1 条超时）
去重后 3 −4（3 条评测集已有，1 条候选内部重复）

79% 衰减。 所以别把 harvest.py 想成「线上错误自动变用例」的流水线，它更像筛矿：搬进来一吨,出来几十克。

3.1 去重要在两个方向上做 # 

existing = {norm(e.inputs.get("question"))
            for e in client.list_examples(dataset_name=args.dataset)}
uniq, seen, dup_in, dup_ds = [], set(), 0, 0
for c in picked:
    k = norm(c["question"])
    if k in existing:        # 方向一：和数据集里已有的重
        dup_ds += 1
        continue
    if k in seen:            # 方向二：候选自己内部重
        dup_in += 1
        continue
    seen.add(k)
    uniq.append(c)

第二个方向容易漏。同一个 bug 在 30 天里被不同用户触发了 5 次，list_runs 就给你 5 条 run——不去重的话评测集里会躺着 5 条一样的题，还会把这个 intent 的权重放大 5 倍，让整体分数被一个 bug 主导。

norm 得去掉标点，否则「同一句话多一个问号」会被当成两条：

def norm(s: str) -> str:
    return re.sub(r"[\s，。？！,.?!、；:：]", "", str(s or ""))

3.2 「和数据集重复 3」是个好消息 #

上面那 3 条「和数据集已有的重复」，说明什么？说明评测集已经覆盖了这些场景，线上照样出错了。

这不是坏消息也不是无事发生，它是一条明确的分工信号：

第二种情况才是评测集质量问题，加题解决不了。harvest.py 把这个数字打出来就是为了让你看见它。

4. 集成的两处断裂 #

这一节讲的两个 bug，都是「脚本各自都能跑，串起来就丢数据」，而且都不报错。闭环最容易断在这种地方。

4.1 少一个字段，sync() 直接崩 #

第一次跑 --apply：

✗ 缺字段 ['split']：我的邮箱是 alice@example.com，卡号 4111 1111 111

harvest.py 生成草稿时没写 split，而 build_evalset.py 的 sync() 要读 r["split"]。这个还算好——它会崩，会被发现。

修法两头都做，防御一次不够：

# harvest.py：字段要和 _seed.jsonl 完全对齐
draft = [{
    "question": c["question"],
    "answer": "",                        # 必须人工填
    "intent": GUESS_INTENT.get(c["kind"], "unknown"),
    "difficulty": "hard",                # 线上出过错的，默认算难
    "must_include": [],                  # 必须人工填
    "split": "full",                     # 线上收的先进 full，别塞 smoke
    "source": f"prod-{today}",
    "source_run_id": c["run_id"],        # 留着回溯原始 trace
    "_error_kind": c["kind"],            # 只给人看，sync() 会忽略
} for c in uniq]
# build_evalset.py：用 get 兜底，外部追加的行可能没写 split
"split": [r.get("split", "full")],

为什么线上收来的一律进 full 而不是 smoke？ 因为 smoke 是每次 push 都跑的快速档（第 36 章 §9.1），10 条、8 秒。往里面塞新题会拖慢每一次 push，而且新题的噪音特性还没量过——第 36 章 §3.1 花了很大功夫才把 smoke 集的噪音区间摸清。新用例先在 full 里跑几轮，稳定了再考虑升到 smoke。

4.2 静默覆盖：追加完 25 条，跑一下变回 22 条 #

这个才是真正危险的。build_evalset.py 原来是这么写的：

def write_seed():
    with open(SEED, "w", encoding="utf-8") as f:   # ← "w"，无条件覆盖
        for r in SEED_ROWS:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")

SEED_ROWS 是内嵌在脚本里的 22 条种子。而 harvest.py --apply 是往 _seed.jsonl 追加。于是：

harvest.py --apply     → _seed.jsonl 变成 25 条
build_evalset.py       → write_seed() 把它盖回 22 条
                       → 然后报告：「新增 0 / 更新 0 / 未变 22」

一个警告都没有。 报告理直气壮地告诉你「未变 22」，因为从它的视角看确实没变——它比对的是被自己刚刚覆盖过的种子文件和线上数据集，两边都是 22 条，严丝合缝。你辛苦筛出来的 3 条在这两行之间蒸发了。

修法：

def write_seed(force: bool = False):
    """把内嵌种子落成 jsonl，仅当文件不存在时。真实项目里这个函数应该
    删掉，文件直接进 Git。
    """
    if SEED.exists() and not force:
        n = sum(1 for l in SEED.read_text(encoding="utf-8").splitlines() if l.strip())
        print(f"沿用已有种子 {SEED.name}（{n} 条）。要重置成内嵌版本加 --reset")
        return
    with open(SEED, "w", encoding="utf-8") as f:
        for r in SEED_ROWS:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
    print(f"种子文件已写入 {SEED.name}（{len(SEED_ROWS)} 条）")

修完再跑：

沿用已有种子 _seed.jsonl（25 条）。要重置成内嵌版本加 --reset
数据集已存在：cs-rag-eval-v1
线上已有 22 条
新增 3 / 更新 0 / 未变 22 / 线上多余 0

最终 25 条
  按意图: {'policy': 10, 'out_of_scope': 4, 'order': 4, 'safety': 3, 'multi': 2, 'chitchat': 2}
  按难度: {'hard': 10, 'easy': 9, 'medium': 6}
  按 split: {'full': 15, 'smoke': 10}

smoke 还是 10 条，快速档的耗时不受影响；新增的 3 条都落在 full。

教训不是「记得用 a 不用 w」，是：把「内嵌的初始数据」和「持续演进的数据」放在同一个文件里，早晚要出这种事。种子数据一旦开始接收外部追加，就该只进 Git、不由脚本生成。上面那个 force 参数是过渡方案，注释里写清了这函数该删。

4.3 溯源字段别在中途丢掉 #

sync() 原来把 source 硬编码成 "seed"：

"metadata": {"intent": ..., "source": "seed"},    # ← 全都变成 seed

这样一来，用例进了数据集就分不清哪些来自线上了。改成用行里带的：

md = {"intent": r["intent"], "difficulty": r["difficulty"],
      "must_include": r["must_include"],
      # source 用行里带的，别硬编码：harvest.py 收上来的带着 prod-YYYYMMDD
      "source": r.get("source", "seed")}
if r.get("source_run_id"):
    md["source_run_id"] = r["source_run_id"]     # 留着回溯原始 trace

验证一下溯源链路通不通：

prod = [e for e in c.list_examples(dataset_id=d.id)
        if str((e.metadata or {}).get("source", "")).startswith("prod-")]
for e in prod:
    m = e.metadata or {}
    print(" Q:", e.inputs["question"][:40])
    print("   source=", m.get("source"),
          " run_id=", str(m.get("source_run_id"))[:8],
          " split=", e.metadata.get("dataset_split"))
线上来源用例: 3
 Q: 北京今天天气怎么样?
   source= prod-20260903  run_id= 01a0277e  split= ['full']
 Q: 我的邮箱是 alice@example.com，卡号 4111 1111 111
   source= prod-20260903  run_id= 01a027a1  split= ['full']
 Q: 查一下A1001订单的状态
   source= prod-20260903  run_id= 01a0275f  split= ['full']

有了 source_run_id，半年后有人问「这条奇怪的题为什么在评测集里」，你能直接把原始 trace 翻出来给他看。没有它，这条题就成了没人敢删的历史遗留。

5. 参考答案只能人写 #

harvest.py 出的草稿里，answer 和 must_include 是空的：

{"question": "北京今天天气怎么样?", "answer": "", "intent": "out_of_scope",
 "difficulty": "hard", "must_include": [], "split": "full",
 "source": "prod-20260903", "source_run_id": "01a0277e-...",
 "_error_kind": "ModelCallLimitExceededError"}

这一步没法自动化，而且理由很硬：能自动生成参考答案的话，就不需要评测了——你已经有一个能给出正确答案的东西了，直接用它当系统不就行了。

所以 --apply 第一件事是拦住没填完的：

blank = [r for r in rows if not r.get("answer") or not r.get("must_include")]
if blank:
    print(f"✗ 有 {len(blank)} 条还没补全，先把 answer 和 must_include 填上：")
    for r in blank[:5]:
        print(f"    {r['question'][:52]}")
    return 1

5.1 别照着「线上那一次的输出」写 #

真正的坑在这儿。补 must_include 时最省事的做法是：翻开那条 trace，看 Agent 当时答了什么，从里面摘个关键词。

那条天气题，线上那次的回答开头是「抱歉，我是电商客服……」，于是很自然地写成 must_include: ["抱歉"]。

同一个 Agent、同一个问题跑 8 次，看这个词出现几次：

问题：北京今天天气怎么样?    跑 8 次

  照抄那一次的措辞  ['抱歉']            命中 2/8  = 25%
  换成同义词集合 任一命中                命中 8/8  = 100%
  换成语义特征 提到自己是客服              命中 8/8  = 100%

八次开头各是：
  1  您好，我是电商客服，主要负责订单、退换货、发票、运费、发货时效等问题
  2  抱歉，我是电商客服，只处理订单、退换货、发票、运费、发货等相关问题。
  3  您好，我是电商客服，主要负责订单、退换货、发票、运费等售后问题，暂无
  4  我是电商客服，主要处理订单和售后问题（退货、换货、发票、运费、发货时
  5  您好，我是电商客服，主要负责订单、退换货、发票、运费等售后问题，没有
  6  我是电商客服，主要处理订单和售后问题（退货、换货、发票、运费、发货时
  7  您好，我是电商客服，主要负责订单、退换货、发票、运费等售后问题，暂无
  8  抱歉，我是电商客服，主要负责订单、退换货、发票、运费等咨询，没有天气

25%。 八次里 Agent 每次都正确地拒答了、都说清了自己是客服、都没编造天气——语义上八次全对，但这条用例会有六次判失败。

它比「一直失败的用例」更坏。一直失败的至少一眼能看出来是坏的；这种随机翻转的会让你在毫不相关的改动上看到分数波动，然后花半天去查一个根本不存在的退化。

判据是明确的：

must_include 里的词，必须是语义上必然出现的，不是「那一次恰好用了」的。

「客服」必然出现——系统提示词里写着它是电商客服，拒答时不可能不提自己的身份。「抱歉」不必然——它是礼貌用语的一种选择，「您好」「暂无」「没有」都能替代。

三种改法，按推荐顺序：

改法 这条题的效果 适用
换成语义必然出现的词 8/8 首选。改一个字符串，零成本
同义词集合，任一命中 8/8 拒答、致谢这类措辞发散但语义收敛的场景
换 LLM-as-judge（第 34 章 §4） —— 语义无法用关键词逼近时。慢、贵、自己也有噪音，别默认用它

第二种要改评测器（must_include 是「全部命中」的语义，同义词集合是「任一命中」），所以这条题我用了第一种：

r["must_include"] = ["客服"]      # 原来是 ["抱歉"]，25% 命中率

5.2 intent 猜错不要紧，空着才容易漏 #

harvest.py 会按错误类型猜一个 intent 初值：

GUESS_INTENT = {
    "PIIDetectionError": "safety",
    "ValueError": "order",
    "ModelCallLimitExceededError": "out_of_scope",
    "OutputParserException": "policy",
}

猜的准确率不高，人工要改。但猜一个总比留空好：留空的字段在人工审的时候最容易被跳过——空着看起来像「这个字段本来就不重要」，而一个明显猜错的值会逼你去改。

6. 闭环真的产生改进了吗 #

到这儿已经完成了「线上 → 数据集」。但闭环的价值不在于评测集变大了，在于它抓到了一个你自己想不到的缺陷。查一下这 3 条到底暴露了什么。

关掉 tracing（额度用尽），直接跑：

只跑线上收上来的 3 条：

 Q: 北京今天天气怎么样?
   期望含: ['抱歉'] | 命中: False | 出错: 无
   实答: 您好，我是电商客服，主要负责订单、退换货、发票、运费等购物相关问题，
         无法查询北京的天气信息哦。建议您使用天气类App或网站查询。

 Q: 我的邮箱是 alice@example.com，卡号 4111 1111 1111 1111，报销流程怎么走?
   期望含: ['电子发票'] | 命中: False | 出错: 无
   实答: 抱歉，知识库中没有「报销流程」相关的信息，我无法为您解答，也不会编造。
         另外提醒您：请勿在聊天中透露银行卡号、密码等敏感信息。

 Q: 查一下A1001订单的状态
   期望含: ['已发货'] | 命中: True | 出错: 无
   实答: 您的订单A1001目前状态为：**已发货**。

三条里两条失败，而且失败的性质完全不同：

6.1 缺陷是什么：知识库有答案，检索词对不上 #

第二条的回答很值得夸——它没编造，明确说了「知识库中没有相关信息」，这正是系统提示词要求的行为，PII 提醒也给到了。单看这个回答，你会觉得系统运转正常。

但知识库里明明有这条：

KB = {
    "退货政策": "签收后 7 天内可无理由退货，商品需保持完好。",
    "换货政策": "质量问题 30 天内可换货，非质量问题不支持换货。",
    "发票": "支持开具电子发票，可在订单页自助申请。",     # ← 就是它
    "运费": "订单满 99 元包邮，未满收取 10 元运费。",
    "发货时效": "付款后 48 小时内发货。",
}

问题在检索：

KB 的 key: ['退货政策', '换货政策', '发票', '运费', '发货时效']
  fuzzy_lookup('报销流程') -> ''
  fuzzy_lookup('报销')     -> ''
  fuzzy_lookup('发票')     -> '支持开具电子发票，可在订单页自助申请。'
  fuzzy_lookup('电子发票') -> '支持开具电子发票，可在订单页自助申请。'
  fuzzy_lookup('开票')     -> '支持开具电子发票，可在订单页自助申请。'

用户说的是「报销」，知识库写的是「发票」。fuzzy_lookup 那两层匹配——子串互含、单字命中——都够不着这一对。

这是一个同义词覆盖缺口，而不是知识内容缺口。 值得注意的是它为什么只能靠线上发现：写评测集的人和写知识库的是同一批人，用的是同一套词汇。你不会想到用「报销」去问一条自己标成「发票」的条目——你脑子里这两个词早就是一个东西了。只有真实用户会用你没想到的说法。

修复是给别名表，不是改知识内容：

# 用户的说法 → KB 里的条目名。这张表是第 38 章的闭环产出
ALIAS = {
    "报销": "发票", "开票": "发票", "抵扣": "发票",
    "退款": "退货政策", "七天无理由": "退货政策",
    "包邮": "运费", "邮费": "运费",
    "多久发货": "发货时效", "什么时候发货": "发货时效",
}

def fuzzy_lookup(keyword: str) -> str:
    for alias, real in ALIAS.items():        # 先过别名，再走原来的模糊匹配
        if alias in keyword:
            return KB[real]
    for k, v in KB.items():
        if keyword in k or k in keyword:
            return v
    hits = [v for k, v in KB.items() if any(c in keyword for c in k)]
    return hits[0] if hits else ""

工具描述也要跟着改，否则模型不知道能问这些：

DOC = """在客服知识库中检索政策条款。keyword 支持模糊匹配，可以传用户原话里的关键词。

    知识库涵盖：退货政策、换货政策、发票（含报销开票）、运费（含包邮）、发货时效。
    检索不到会返回空字符串，此时应当告知用户你没有这方面信息，不要编造。
    """

6.2 效果被坏用例稀释了一半 #

修完之后，三条各跑 3 次：

  用例                      版本      命中率    工具   token
  -------------------------------------------------------
  报销（修的就是这条）        修复前      33%   1.33    1496
                          修复后     100%   1.67    1550
  -------------------------------------------------------
  天气（跟修复无关）          修复前      33%   0.00     634
                          修复后       0%   0.00     624
  -------------------------------------------------------
  订单（跟修复无关）          修复前     100%   1.00    1207
                          修复后     100%   1.00    1224
  -------------------------------------------------------

修复真的有效，而且只在它该生效的那条上有效：33% → 100%，工具调用从 1.33 涨到 1.67（多检索了一次，因为现在检索得到东西了），token 只多了 54。

但如果只看三条的总均值：

修复前 (33 + 33 + 100) / 3 = 55%
修复后 (100 + 0 + 100) / 3 = 67%
                      变化 = +12 个百分点

真实效果是「一条从 33% 修到 100%」，看起来却只涨了 12 个百分点，还有一半的信号被天气那条 0%~33% 的随机跳动吃掉了。一条坏用例，能把一次干净的修复稀释成「好像有点用」。

这就是 §5.1 那个坑的真实代价，也是为什么必须先把用例修对、再去量修复效果——顺序反了，你会拿着一个 +12% 的模糊结论，怀疑自己这次改动到底有没有价值。

6.3 这次是结果分暴露、过程指标无感 #

第 36 章 §4.4 的结论是「模型会替你把 bug 圆过去」，所以退化藏在过程指标里，答案分看不出来。这一章的缺陷正好相反：

第 36 章的 search_kb_broken 本章的同义词缺口
模型的反应 检索不到就多试几个词，凑出答案 检索不到就放弃，老实说没有
过程指标 工具调用 0.89 → 1.33，暴露 0.00 → 0.00，无感
结果分 几乎不降，掩盖 33% → 100%，暴露

差别在模型的应对策略：一个是「重试到成功」，一个是「一次不成就诚实认输」。前者把缺陷转成成本、后者把缺陷转成失败。

所以第 36 章那三层门禁一层都不能省。 硬门禁抓崩溃，过程预算抓「悄悄变贵」，结果下限抓「老实认输」。它们各有盲区，只是盲区不重合。

6.4 修完之后：重定基线 #

修复改变了工具调用次数（那条题 1.33 → 1.67），过程预算的基线就过期了。不重定的话下一次 CI 会误报「过程预算超标」。

python build_evalset.py                    # 评测集 22 → 25
python gate.py --full --update -n 5        # 重定基线，5 次取上界

第 36 章 §5.3 讲了为什么必须多跑几次取上界：单次快照当基线，下一次只要正常波动到上沿就误报。

最后验证一遍闭环成果——修完 KB、修完 must_include，三条各跑 3 次：

修复后重跑线上收来的 3 条，各 3 次：
  3/3  期望['客服']      北京今天天气怎么样?
  3/3  期望['电子发票']  我的邮箱是 alice@example.com，卡号 4111 11
  3/3  期望['已发货']    查一下A1001订单的状态

全部稳定通过

三条都稳定，从此进了回归集。下次谁把别名表删了，gate.py 会拦住他。

6.5 环合上了：再收一次，0 条 #

判断闭环有没有真的合上，有个很直接的办法——用同样的窗口再收一次：

【1. 提取】13 条能挖出问题文本，1 条挖不出（inputs 结构不认识）
【2. 分流】→ 收 + 待定 共 7 条，弃 6 条
【3. 去重】7 条 → 0 条（和数据集重复 7，候选内部重复 0）

没有新用例。线上出的问题评测集里都已经覆盖了——这是好事。

同一个 30 天窗口、同样 14 条线上错误、同样筛出 7 条值得收的，但去重后剩 0——这 7 条现在全部落在评测集的覆盖范围内。对比这一章开始时的「7 → 3」，环合上了。

这个 0 也是 harvest.py 日常跑出来最该看到的数字。它变回非 0 的那一天，说明线上出现了评测集没覆盖的新场景，那时候才需要走一遍本章的流程。所以这个脚本正确的用法是挂定时任务、只在有产出时才通知人，而不是每次发版都手动跑一遍。

7. 本章产出：harvest.py #

完整脚本。它只做「收集 + 分流 + 去重 + 出草稿」，写参考答案和入库分开两步——中间必须有人看一眼。

"""harvest.py —— 把线上问题收成评测用例。闭环里最容易断的一环。

    python harvest.py                          # 看最近 7 天，出草稿
    python harvest.py --hours 720 --project langchain
    python harvest.py --feedback user_thumb    # 连用户差评一起收
    python harvest.py --apply harvest_draft.jsonl   # 人工补全后写进数据集

产出 harvest_draft.jsonl，里面 answer 和 must_include 是空的——**必须人工填**。
参考答案没法自动生成：能自动生成的话就不需要评测了。

实测漏斗（第 38 章 §3）：
    14 条线上错误
    → 13 条能提取出问题文本（1 条的 inputs 里只有 {input}，没有问题）
    →  7 条值得收（弃掉 5 条 AuthenticationError + 1 条 TimeoutError）
    →  3 条去重后剩下的
衰减 79%。所以别指望「线上错误自动变用例」，它更像筛矿。
"""
import argparse
import datetime as dt
import json
import os
import re
import sys
from collections import Counter

from dotenv import load_dotenv

ROOT = os.path.dirname(os.path.abspath(__file__))
load_dotenv(os.path.join(ROOT, ".env"), override=True)
os.environ["LANGSMITH_TRACING"] = "false"      # 收集阶段全是读，不必留痕

from langsmith import Client

client = Client()
DATASET = "cs-rag-eval-v1"
DRAFT = os.path.join(ROOT, "harvest_draft.jsonl")

# 该不该收，判断依据只有一句：换个环境重跑，这个错还会不会重现？
#   会重现（输入本身触发的）→ 收，它是真实的缺陷用例
#   不会（环境/配额/网络）  → 弃，收了就是评测集里一条永久失败的噪音
TRIAGE = {
    "PIIDetectionError":           ("收", "输入里真有 PII，正是该防的场景"),
    "ValueError":                  ("收", "参数校验失败，多半是输入触发的缺陷"),
    "OutputParserException":       ("收", "模型输出格式不对，可复现"),
    "ModelCallLimitExceededError": ("待定", "可能配额设小了，也可能该输入导致死循环"),
    "RuntimeError":                ("待定", "要看具体信息"),
    "AuthenticationError":         ("弃", "key 过期或失效，与输入无关"),
    "RateLimitError":              ("弃", "限流，不可复现"),
    "TimeoutError":                ("弃", "网络或下游抖动，不可复现"),
    "APIConnectionError":          ("弃", "网络问题"),
}

# 按错误类型猜个 intent 初值，人工再改。猜错不要紧，空着才容易漏
GUESS_INTENT = {
    "PIIDetectionError": "safety",
    "ValueError": "order",
    "ModelCallLimitExceededError": "out_of_scope",
    "OutputParserException": "policy",
}


def error_kind(e: str) -> str:
    """和 monitor.py 里同一个归一化函数（第 37 章 §3.2）。"""
    if not e:
        return "(无错误信息)"
    m = re.search(r"([A-Za-z_][\w.]*(?:Error|Exception|Timeout))\b", e)
    if m:
        return m.group(1).rsplit(".", 1)[-1]
    line = e.strip().splitlines()[0]
    return re.sub(r"[0-9a-f]{8}-[0-9a-f-]{27}", "<id>", line)[:50]


def extract_question(inputs: dict):
    """从 trace 的 inputs 里挖出用户问的那句话。

    线上 trace 的 inputs 结构不统一：Agent 调用是 {messages}，
    被 @traceable 装饰的函数可能是 {input} / {query} / 自定义名。
    实测 14 条里有 1 条挖不出来（inputs 只有 {input} 且不是字符串）。
    """
    inputs = inputs or {}
    for key in ("question", "input", "query", "text", "prompt"):
        if isinstance(inputs.get(key), str) and inputs[key].strip():
            return inputs[key]
    msgs = inputs.get("messages")
    if isinstance(msgs, list) and msgs:
        m0 = msgs[0]
        if isinstance(m0, list) and m0:      # 有时是嵌套一层的
            m0 = m0[0]
        if isinstance(m0, dict):
            c = m0.get("content")
            if isinstance(c, str) and c.strip():
                return c
    return None


def norm(s: str) -> str:
    """去掉标点空格再比，挡住「同一句话多一个问号」这种重复。"""
    return re.sub(r"[\s，。？！,.?!、；:：]", "", str(s or ""))


def gather(project: str, hours: float, feedback_key: str | None):
    """收两类线索：出错的 run，以及被用户打了差评的 run。"""
    since = dt.datetime.now(dt.timezone.utc) - dt.timedelta(hours=hours)
    runs = list(client.list_runs(project_name=project, is_root=True,
                                 error=True, start_time=since))
    tagged = [(r, error_kind(r.error)) for r in runs]

    if feedback_key:
        # 差评往往比报错更值钱：系统没崩，但用户不满意——这类问题
        # 只有真实用户能发现。注意 feedback 有 10~15 秒索引延迟（第 33 章 §5.5）
        f = (f'and(eq(is_root, true), eq(feedback_key, "{feedback_key}"), '
             f'lt(feedback_score, 1))')
        bad = list(client.list_runs(project_name=project, filter=f,
                                    start_time=since))
        tagged += [(r, "用户差评") for r in bad]
        print(f"  差评 run：{len(bad)} 条")
    return tagged


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--project", default="langchain")
    ap.add_argument("--hours", type=float, default=168)
    ap.add_argument("--dataset", default=DATASET)
    ap.add_argument("--feedback", help="同时收这个 key 的差评，如 user_thumb")
    ap.add_argument("--apply", help="把补全好的 jsonl 写进数据集")
    args = ap.parse_args()

    # ───────── --apply：人工补全后写回数据集
    if args.apply:
        rows = [json.loads(l) for l in open(args.apply, encoding="utf-8")
                if l.strip()]
        blank = [r for r in rows if not r.get("answer")
                 or not r.get("must_include")]
        if blank:
            print(f"✗ 有 {len(blank)} 条还没补全，先把 answer 和 "
                  f"must_include 填上：")
            for r in blank[:5]:
                print(f"    {r['question'][:52]}")
            return 1
        # 字段对不齐会让 build_evalset.py 在 sync() 里崩，提前挡掉
        need = {"question", "answer", "intent", "difficulty",
                "must_include", "split"}
        for r in rows:
            missing = need - set(r)
            if missing:
                print(f"✗ 缺字段 {sorted(missing)}：{r.get('question', '')[:40]}")
                return 1
        # 走和 build_evalset.py 一样的路径：写进 _seed.jsonl 再统一重建，
        # 这样数据集始终由种子文件生成，可复现（第 33 章 §9）
        seed = os.path.join(ROOT, "_seed.jsonl")
        with open(seed, "a", encoding="utf-8") as f:
            for r in rows:
                f.write(json.dumps(r, ensure_ascii=False) + "\n")
        print(f"✓ {len(rows)} 条已追加到 _seed.jsonl")
        print(f"  接着跑：python build_evalset.py   （幂等，会重建数据集）")
        print(f"  然后：  python gate.py --full --update -n 5   （基线要重定）")
        return 0

    print("=" * 76)
    print(f"收集线上问题 | {args.project} | 最近 {args.hours:g}h")
    print("=" * 76)

    tagged = gather(args.project, args.hours, args.feedback)
    print(f"  出错或差评的根 run：{len(tagged)} 条")
    if not tagged:
        print("\n窗口内没有线索。要么真没问题，要么项目名/窗口不对。")
        return 0

    # ───────── 1. 提取问题文本
    cands, no_q = [], 0
    for r, kind in tagged:
        q = extract_question(r.inputs)
        if not q:
            no_q += 1
            continue
        cands.append({"question": q, "kind": kind, "run_id": str(r.id),
                      "at": r.start_time.strftime("%Y-%m-%d") if r.start_time else ""})
    print(f"\n【1. 提取】{len(cands)} 条能挖出问题文本"
          f"{f'，{no_q} 条挖不出（inputs 结构不认识）' if no_q else ''}")

    # ───────── 2. 分流
    print(f"\n【2. 分流】按错误类型判断该不该收")
    print(f"  {'类型':30}{'条数':>5}{'处置':>6}   理由")
    print("-" * 76)
    for k, n in Counter(c["kind"] for c in cands).most_common():
        v, why = TRIAGE.get(k, ("待定", "未知类型，人工看"))
        print(f"  {k:30}{n:>5}{v:>6}   {why}")
    picked = [c for c in cands
              if TRIAGE.get(c["kind"], ("待定",))[0] in ("收", "待定")]
    print(f"  → 收 + 待定 共 {len(picked)} 条，"
          f"弃 {len(cands) - len(picked)} 条")

    # ───────── 3. 去重
    existing = {norm(e.inputs.get("question"))
                for e in client.list_examples(dataset_name=args.dataset)}
    uniq, seen, dup_in, dup_ds = [], set(), 0, 0
    for c in picked:
        k = norm(c["question"])
        if k in existing:
            dup_ds += 1
            continue
        if k in seen:
            dup_in += 1
            continue
        seen.add(k)
        uniq.append(c)
    print(f"\n【3. 去重】{len(picked)} 条 → {len(uniq)} 条"
          f"（和数据集重复 {dup_ds}，候选内部重复 {dup_in}）")

    if not uniq:
        print("\n没有新用例。线上出的问题评测集里都已经覆盖了——这是好事。")
        return 0

    # ───────── 4. 出草稿
    today = dt.datetime.now().strftime("%Y%m%d")
    # 字段要和 _seed.jsonl 完全对齐，少一个 split 就会让 build_evalset.py
    # 抛 KeyError（第 38 章 §4.1 实测的断裂）
    draft = [{
        "question": c["question"],
        "answer": "",                        # 必须人工填
        "intent": GUESS_INTENT.get(c["kind"], "unknown"),
        "difficulty": "hard",                # 线上出过错的，默认算难
        "must_include": [],                  # 必须人工填
        "split": "full",                     # 线上收的先进 full，别塞 smoke
        "source": f"prod-{today}",
        "source_run_id": c["run_id"],        # 留着回溯原始 trace（第 33 章 §4）
        "_error_kind": c["kind"],            # 只是给人看的，sync() 会忽略
    } for c in uniq]

    with open(DRAFT, "w", encoding="utf-8") as f:
        for r in draft:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")

    print(f"\n【4. 草稿】{len(draft)} 条已写入 {os.path.basename(DRAFT)}")
    for r in draft:
        print(f"    [{r['_error_kind']:26}] {r['question'][:42]}")

    print("\n" + "=" * 76)
    print("接下来（这三步没法自动化，参考答案只能人写）：")
    print(f"  1 打开 {os.path.basename(DRAFT)}，逐条补 answer 和 must_include")
    print(f"  2 python harvest.py --apply {os.path.basename(DRAFT)}")
    print(f"  3 python build_evalset.py  &&  python gate.py --full --update -n 5")
    print("=" * 76)
    return 0


if __name__ == "__main__":
    sys.exit(main())

7.1 一次闭环的完整命令序列 # 

# ① 巡检，发现线上有问题（第 37 章）
python monitor.py --hours 24

# ② 收集，出草稿
python harvest.py --hours 720 --feedback user_thumb

# ③ 人工补 answer / must_include —— 这一步不能自动化
#    注意 must_include 要写语义必然出现的词（§5.1）
notepad harvest_draft.jsonl

# ④ 入库
python harvest.py --apply harvest_draft.jsonl
python build_evalset.py

# ⑤ 跑一遍，看新用例暴露了什么
python gate.py --full

# ⑥ 修代码 —— 这一步也不能自动化

# ⑦ 重定基线，新用例进回归集
python gate.py --full --update -n 5

七步里有两步是人工的（③ 和 ⑥），而且是最花时间的两步。别设计一个「全自动闭环」，那东西不存在。

7.2 这个闭环凭什么算「可度量」 #

大纲里这一章的产出写的是「可度量的质量闭环」。四个字都得兑现：

要素 兑现在哪 本章实测
有数 每次改动都出具体数字，不是「感觉好了」 那条用例 33% → 100%
可比 数字之间能比，因为跑的是同一个固定数据集 cs-rag-eval-v1 版本化，as_of 可复现
有基线 知道「正常」是什么范围，才知道什么叫异常 baseline.json，5 次运行的上界
能拦 退化会被自动拦下，不靠人记得去看 gate.py 退出码非 0，CI 红灯
会长大 线上的新问题会进来，评测集不会过期 22 → 25 条，带 source_run_id

少任何一条都不算闭环：

8. 成本：人工是瓶颈 #

一次完整闭环的耗时，按本章这次实际情况估：

步骤 耗时 自动化
monitor.py 巡检 约 20s 全自动
harvest.py 收集 14 → 3 条 约 5s 全自动
人工补 3 条参考答案 约 25 min 不可自动化
--apply + build_evalset.py 约 17s 全自动
gate.py --full 25 条 约 95s 全自动
人工定位并修缺陷 约 20 min 不可自动化
gate.py --full --update -n 5 约 8 min 全自动

机器合计约 10 分钟，人工合计约 45 分钟。人工占 82%。

这个比例决定了闭环该怎么跑：

9. 常见问题 #

Q：harvest.py 跑出来 0 条新用例，是不是脚本坏了？

先看它打的三个数字，0 有两种截然不同的来源：

Q：--feedback 查出来 0 条，但我确实刚打了差评。

feedback 有 10~15 秒索引延迟（第 33 章 §5.5）。等一会儿再查。这个延迟也意味着别在打完分立刻查，那会得出「过滤器不管用」的错误结论。

Q：能不能让 LLM 生成参考答案，人只做审核？

可以，但要清楚你在换什么。LLM 生成的参考答案会系统性地偏向「LLM 认为对的答案」，而你的评测集本来是要检查 LLM 对不对的。这会让评测集和被测系统共享同一套盲区。真要这么做，至少保证生成用的模型和被测模型不是同一个，并且人工审核时重点看「这个答案凭什么是对的」，而不是「这个答案读起来顺不顺」。

Q：线上收来的用例应该占评测集多大比例？

没有普适答案，但有个方向：自己造的题覆盖「你设计时想到的场景」，线上收的题覆盖「你没想到的」。如果跑了几个月，线上收的还不到 10%，大概是收集环节没在跑；如果超过一半，说明设计阶段的场景梳理做得太粗。

Q：_error_kind 这种下划线开头的字段会进数据集吗？

不会。sync() 只读它认识的那几个字段，_error_kind 留在 _seed.jsonl 里给人看。用下划线前缀标记「本地字段」是个省事的约定，但它不是 LangSmith 的机制——是 sync() 恰好没读它。

Q：额度用尽跑不了 evaluate 怎么办？

第 36 章 §6.3 讲了：额度用尽只影响 trace 上传，分数照样算。gate.py --from EXPERIMENT 能对已有实验重新判定。本章 §6 的数据就是在额度耗尽的情况下拿到的——关掉 tracing，直接从 messages 里数工具调用和 token。核心逻辑不该依赖某个 SaaS 是否可达。

Q：为什么不把 harvest.py 和 build_evalset.py 合成一个脚本？

因为中间必须有人看一眼。合成一个的话，那个「人工补参考答案」的步骤要么变成交互式输入（跑一次得盯着终端敲半小时），要么被跳过。分成两个脚本、中间落一个 harvest_draft.jsonl 文件，是在用文件系统当断点——你可以今天出草稿，明天补答案，后天入库。

10. 练习 #

  1. 量一下你自己评测集里的坏用例。 挑三条 must_include，各跑 8 次，算命中率。低于 100% 的都是在往结果里注噪音。按 §5.1 那三种改法修掉，再重跑验证。

  2. 把 §3 的分流表按你的项目改一遍。 你的系统会抛什么错？哪些是输入触发的、哪些是环境的？写不出理由的类型先标「待定」，别猜。

  3. 手动制造一条差评再收上来。 用 client.create_feedback(run_id, key="user_thumb", score=0) 给一条正常 run 打差评，等 15 秒，然后 harvest.py --feedback user_thumb。走通这条路，你的闭环就接上了最值钱的那类信号。

  4. 验证 §4.2 那个静默覆盖。 把 write_seed() 改回无条件 "w"，追加一条再跑 build_evalset.py，看报告是怎么理直气壮地说「未变 22」的。这类 bug 的可怕之处只有亲眼见过才记得住。

  5. 给第 28 章的审批流图建一次闭环。 本章用的是第 15 章那条 RAG 链路。审批流的 inputs 结构和客服 Agent 完全不同，extract_question 要重写；「参考答案」也不再是一段文本，而是「该不该批」+「走了哪条分支」。想清楚这两件事怎么表达，比写代码难。

  6. 进阶：给收上来的用例加过期机制。 一条三个月前从线上收的用例，如果对应的功能已经下线了，它就该被删掉而不是一直挂着。用 source 里的日期加一个巡检，把「超过 N 天没失败过、且 source 是 prod- 的」用例列出来人工确认。

11. 小结 #

这一章把前九章的零件接成了环。

接线本身的技术含量不高，容易断的地方也不在算法上：

真正要想清楚的是三个判断：

  1. 哪些线上错误该收。 判据是「换个环境重跑还会不会重现」。实测 79% 的错误不该收——收进去就是永久失败的噪音，会让人习惯「有几条一直是红的」
  2. must_include 该写什么。 必须是语义上必然出现的词。照抄「线上那一次的措辞」写出来的用例，实测命中率 25%——它会在毫不相关的改动上制造假警报，还会把一次干净的修复（33% → 100%）稀释成「好像有点用」（+12pp）
  3. 一次修复到底有没有效。 结果分和过程指标各有盲区，取决于模型遇到缺陷时是「重试到成功」还是「诚实认输」。前者藏在过程里（第 36 章），后者藏在结果里（本章 §6.3）。所以三层门禁一层都不能省

最后是一个别抱幻想的结论： 闭环里人工占 82%（§8），而且是不可压缩的 82%——写参考答案和定位缺陷这两件事，能自动化的那天就是不需要评测的那天。所以优化方向不是提高自动化率，而是减少人要看的条数。§3 那个分流表把 14 条压到 3 条，人工从 2 小时降到 25 分钟，这是整个闭环里回报最高的一段代码。

到这里，从「第 29 章：为什么需要观测」到「第 38 章：可度量的质量闭环」这十章走完了。手上应该有六个能跑的脚本、一个 25 条的版本化评测集、一份带上界的基线、一个会拦人的 CI 门禁，以及一条把线上问题送回评测集的通路。它们合起来回答的是同一个问题：

你凭什么说这一版比上一版好？

← 上一节 37.线上监控 下一节 39.为什么需要DeepAgents →