第 13 章 RAG 评测：Context Recall、Faithfulness、Answer Relevance

“Garbage retrieved, garbage generated.” —— RAG 时代的第一定律

13.1 RAG 评测的特殊性

RAG（Retrieval-Augmented Generation）系统由两个独立子系统组成——retriever 和 generator。它的评测因此天然分两层：

flowchart LR
  Q[Question] --> Retr[Retriever]
  KB[(向量库)] --> Retr
  Retr --> Ctx[Retrieved Contexts]
  Q --> Gen[Generator LLM]
  Ctx --> Gen
  Gen --> A[Answer]
  Retr -.Recall / Precision.-> M1[Retriever 评测]
  Gen -.Faithfulness / Relevance.-> M2[Generator 评测]
  style M1 fill:#dbeafe
  style M2 fill:#dcfce7

这种”双层评测”是 RAG 与纯 LLM 应用最大的工程差异。一个普通 LLM 应用挂掉时你只需问”模型答得对不对”；RAG 挂掉时你必须问”是 retriever 没找到、还是 generator 没用好”——指标和工程修法完全不同。

如果跳过这一分层、把 RAG 当黑盒评测：“最终回答 80% 通过”。看起来稳，但你不知道：

• 是 retriever 拿到了正确文档但 generator 没 follow？（Faithfulness 问题）
• 是 retriever 漏了关键文档导致 generator 只能编？（Recall 问题）
• 是 retriever 拿了一堆无关文档把 generator 带跑偏了？（Precision 问题）

这三种失败的工程修法完全不同——分别要调 prompt、改 chunk size、改 retriever 算法。混在一个总分里看，你一辈子也找不到根因。

13.2 双层指标矩阵

把第 4、11 章的指标按”层”重新组织：

四个核心指标缺一不可，加上两个派生指标共六维。任何一个低于阈值都对应特定工程问题，下一节展开。

13.3 指标 → 失败模式 → 工程修法

flowchart TB
  Start[指标异常] --> R{Recall 低?}
  R -->|是| R1[chunk 太大丢上下文 / chunk 太小切碎语义]
  R -->|是| R2[向量化模型选错<br/>embedding 模型 lang 不匹配]
  R -->|是| R3[top_k 太小<br/>关键文档没进 top]
  R -->|否| P{Precision 低?}
  P -->|是| P1[chunk 太小, 检索碎片化]
  P -->|是| P2[query 增强缺失<br/>未做 HyDE / query expansion]
  P -->|否| F{Faithfulness 低?}
  F -->|是| F1[generator prompt 没强调<br/>'仅基于 context 回答']
  F -->|是| F2[模型推理能力不足]
  F -->|是| F3[context 内有冲突信息]
  F -->|否| AR{Answer Relevance 低?}
  AR -->|是| AR1[prompt 引导模型答非所问]
  AR -->|是| AR2[noncommittal: 模型频繁说不知道]
  style R fill:#dbeafe
  style P fill:#dcfce7
  style F fill:#fef3c7
  style AR fill:#fce7f3

每条修法都是高频踩坑积累的经验。详细展开几条最常见的：

13.3.1 Recall 低：先看 chunk size

Chunk size 是 RAG 的第一个调参旋钮。常见错误：

• chunk 太大（> 1000 tokens）：单个 chunk 信息量足，但 vector 化时被”语义平均”，匹配不准
• chunk 太小（< 100 tokens）：上下文丢失，“该结论的前提”和”该结论的结果”被切到不同 chunk

修法的标准做法：在你的真实数据集上做 chunk size 扫描实验——512 / 768 / 1024 / 1536 / 2048，每种跑一遍 Recall 评测，画曲线。绝大部分场景的最优值在 512-1024 之间。

13.3.2 Precision 低：query 重写是大杀器

用户问的 query 经常很短或表达不准（“那个退货政策”），retriever 直接用原始 query 搜会拉一堆无关 chunk。

常见修法是 query 重写：

• HyDE（Hypothetical Document Embeddings）：让 LLM 先”假设”一个回答，用这个假设回答的 embedding 去检索
• Query Expansion：让 LLM 把短 query 扩展成几个不同表达
• Multi-Query：生成多个变体 query 并行检索，结果合并去重

这些技术能把 Precision 从 0.5 拉到 0.75+。本卷不深入 RAG 工程（详见《RAG 工程》第 8、12 章），评测视角的核心是：Precision 异常时，先怀疑 query 没做好。

13.3.3 Faithfulness 低：往往是 prompt 不够”严”

Generator 的 system prompt 决定了模型对 context 的依赖程度。不严的 prompt：

"You are a helpful customer service assistant. Use the following context to answer the question:
{context}

Question: {question}
Answer:"

严格的 prompt：

"You are a customer service assistant. Answer the user's question based ONLY on the
information in the provided context. If the context does not contain enough information
to answer, say 'I don't have that information.'

Do NOT use prior knowledge. Do NOT make up information. Quote relevant policy text
when possible.

Context:
{context}

Question: {question}
Answer:"

仅这一处 prompt 改动，常能把 Faithfulness 从 0.78 拉到 0.92。第 1 章 NYC MyCity 案的失败之一就是 generator prompt 不够严——下面 §13.6 会做反事实推演。

13.4 一个端到端的 RAG 评测脚本

把 ragas + promptfoo 组合成一个完整脚本，可直接放进 CI。基于第 11、12 章的 API：

# eval_rag.py
import asyncio
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import (
    Faithfulness, AnswerRelevancy,
    ContextRecall, ContextPrecision,
)
from your_rag_app import rag_pipeline

async def run_eval(samples_jsonl):
    """运行 RAG 评测的端到端脚本"""
    # 1. 加载黄金集
    samples = load_jsonl(samples_jsonl)

    # 2. 跑端到端 RAG, 拿到 (question, context, answer) 三元组
    rows = []
    for s in samples:
        question = s["user_input"]
        contexts = await rag_pipeline.retrieve(question)
        answer = await rag_pipeline.generate(question, contexts)
        rows.append({
            "user_input": question,
            "retrieved_contexts": contexts,
            "response": answer,
            "reference": s.get("reference", ""),
        })

    # 3. ragas 评测
    ds = Dataset.from_list(rows)
    result = evaluate(
        dataset=ds,
        metrics=[
            ContextRecall(),
            ContextPrecision(),
            Faithfulness(),
            AnswerRelevancy(),
        ],
    )

    # 4. 输出报告 + 退出码
    print(result)
    df = result.to_pandas()

    failed = []
    for _, row in df.iterrows():
        if row["faithfulness"] < 0.85:
            failed.append((row["user_input"], "faithfulness", row["faithfulness"]))
        if row["context_recall"] < 0.90:
            failed.append((row["user_input"], "recall", row["context_recall"]))
        # ... 其他阈值

    if failed:
        print(f"\n❌ {len(failed)} cases below threshold:")
        for q, m, v in failed[:20]:
            print(f"  - [{m}={v:.2f}] {q}")
        exit(1)
    print("\n✓ All assertions passed")

if __name__ == "__main__":
    asyncio.run(run_eval("data/golden/v1.jsonl"))

这个脚本不到 60 行，但它做完了：

• 端到端跑 RAG（retrieve + generate）
• 四个核心指标全测
• 阈值检测失败 case
• 退出码用于 CI 集成

第 18 章会展示如何把这个脚本接入 GitHub Actions 做 PR 门禁。

13.5 一个常被忽视的指标：Noise Sensitivity

ragas 仓库中有个 metric 叫 NoiseSensitivity（metrics/_noise_sensitivity.py）——评测在 retriever 拿到错信息时，generator 是否会被带偏。

它的工作方式：

1. 对每条样例，故意往 context 里插入一段 noise（来自其他文档的片段）
2. 看 generator 是否被 noise 干扰、给出错误答案

这个指标对评测 RAG 的鲁棒性至关重要——retriever 在生产里 100% 会犯错（拉错 chunk、混入无关信息），你的 generator 必须能区分什么该用什么不该用。

工业实践把 Noise Sensitivity 作为 Faithfulness 的补充指标——前者评”理想 context 下”的忠实度，后者评”嘈杂 context 下”的稳健度。两个都过才算合格的 RAG。

13.6 反事实推演：NYC MyCity 案如果有 RAG 评测

回到第 1 章 §1.4 的 NYC MyCity 违法建议事故。如果团队上线前做了完整 RAG 评测，会发生什么？

Step 1：构造黄金集

把 nyc.gov 上”For Business Owners”section 的 200 条核心法规问答抽出来：

{"user_input": "Can I keep my employees' tips?", "reference": "No. Under NY Labor Law §196-d, employers cannot retain any portion of employee tips."}
{"user_input": "Can I require cash-only at my store?", "reference": "No. NYC Local Law 28 of 2020 prohibits retailers from refusing cash."}
...

Step 2：跑 RAG 评测

metrics:
  - context_recall: 必须 ≥ 0.90
  - context_precision: 必须 ≥ 0.70
  - faithfulness: 必须 ≥ 0.85
  - answer_correctness: 必须 ≥ 0.80

Step 3：预期失败模式

The Markup 报道的 6 条违法建议样例里，至少 4 条会在以下任一指标上失败：

• “Can I keep tips?” 答 yes：Faithfulness 极低（context 里明确 §196-d 禁止）
• “Cash-only OK?” 答 yes：Answer Correctness 0（与 NYC Local Law 28 矛盾）
• “Fire harassment-complaining employee?” 答 yes：Faithfulness 极低（Title VII 明确为 retaliation）

任何一条失败 → CI 阻止上线 → 媒体不会有”NYC chatbot 教企业违法”的新闻。

Step 4：发现的根因

跑完评测会立刻看到失败模式集中在 Faithfulness——说明 generator prompt 不严。修法是改 prompt 强调”仅基于 NYC 官方政策回答”。改完重跑评测、Faithfulness 涨到 0.92、上线。

整个过程从评测集构造到首次评测跑出，工程量约 1 人 1 周。对照实际事故的”全国媒体头版 + ACLU 谴责 + 监管介入”代价，这是性价比最高的工程投入之一。

13.7 RAG 评测的运营节奏

gantt
    title RAG 系统的评测运营节奏
    dateFormat  YYYY-MM-DD
    section 离线评测
    每次 PR 合并触发 (50 题)    :2026-05-01, 30d
    每周回归集 (200 题)         :2026-05-01, 30d
    section 在线评测
    每天采样 1% 流量做 Faithfulness :crit, 2026-05-01, 30d
    section 数据集维护
    每周 hard case mining       :2026-05-08, 30d
    每季度 retriever 重训练评测 :2026-05-08, 30d

四件事并行：

• 每次 PR 合并：50 题快速回归（~5 分钟），阻止改坏 prompt
• 每周一次完整 200 题回归：（~30 分钟）发现累积 drift
• 每天 1% 在线采样 Faithfulness：发现真实分布下的退化
• 每周 hard case mining：把生产里的失败 case 反哺评测集

第 18 章会给出这套节奏的完整 GitHub Actions / cron 配置。

13.7.5 调参手册：从指标定位到具体动作的全流程

把 §13.3 的诊断流程图落到具体动作清单。当 RAG 评测异常时，按下面的优先级排查：

第一步：先看哪一层

跑评测 → 拿到 4 个核心指标
- 如果 Recall < 0.85: retriever 层问题, 看第二步
- 如果 Recall ≥ 0.85 但 Faithfulness < 0.80: generator 层问题, 看第三步
- 如果 Recall ≥ 0.85 且 Faithfulness ≥ 0.85 但 Answer Relevance 低: prompt 引导问题

第二步（retriever 调参）：从便宜到贵依次试

1. chunk_size 扫描（成本: 0）
   → 重新切分文档为 256/512/1024/2048 token, 各自跑评测看 Recall 曲线

2. embedding 模型升级（成本: 改 API key）
   → text-embedding-ada-002 → text-embedding-3-small → text-embedding-3-large
     OpenAI 的三代模型差异显著, large 在多语言上提升尤其明显

3. top_k 调大（成本: 检索延迟略增）
   → 5 → 10 → 20 看 Recall 收益

4. 加 reranker（成本: 一次重排, 额外延迟)
   → cohere-rerank-v3 或 BAAI/bge-reranker, 显著提升 Precision

5. query 重写（成本: 多一次 LLM 调用）
   → HyDE / multi-query / step-back prompting

第三步（generator 调参）

1. system prompt 加强约束（参见 §13.3.3 严格 prompt 范本）
2. context 排序：把最相关 chunk 放最前（"lost in the middle" 现象）
3. 加结构化输出（强制 JSON schema, 减少自由发挥空间）
4. 升级模型版本（最贵但最稳）

这套从便宜到贵的排查顺序是工业团队踩坑总结的——不要一上来就换大模型，常常一个 chunk_size 调整就能解决问题。每一步调完都要重跑 evals，记录改动 → 指标变化 → 决定保留 / 回滚——这是评测体系给团队带来的”工程纪律”价值。

13.7.6 一份公开数据视角下的 retriever 横评

工业团队常问”我该用 OpenAI text-embedding-3-large 还是开源的 BGE-M3?”。MTEB（Massive Text Embedding Benchmark, Muennighoff et al. 2023, arXiv:2210.07316）给了一份持续更新的公开 leaderboard。截至 2026 年初常见模型的相对位置：

注：上表数字基于 MTEB 公开榜单的历史快照，具体值随版本变动。

工程上的判断：

• 多语言 + 数据合规：BGE-M3 自托管（中文场景几乎一致首选）
• 追求性价比 + 国际化：text-embedding-3-small
• 追求最高质量：text-embedding-3-large 或 BGE-M3 二选一，跑你自己的业务集对比

注意：MTEB leaderboard 与你业务真实表现可能差异 5-15pp——MTEB 上拿 65 的模型在你的客服场景可能只拿 50。所以”看榜单选 + 在自己 retriever 评测集上重跑”才是负责任做法。这也是 §13.3 强调”不能只看 Recall 数字、要看是否符合自己业务分布”的理由。

13.7.7 一次完整的 RAG 调试故事：从 80% 拉到 95%

把本章方法整合成一个端到端调试故事——基于多个公开 RAG 项目（如 GitHub repo langchain-ai/rag-from-scratch、run-llama/llama_index 例子）的复合演绎，所有数字与方法都来自公开实践：

起点：客服 RAG 系统初版上线，离线评测 Faithfulness = 0.78、Recall = 0.82、Latency p95 = 4.2s。用户投诉率上升。

第 1 周诊断：

跑完整评测 → 失败 case 集中在 3 类:
  - 长政策条款被切断（占 42%）→ Recall 偏低主因
  - 模型加入 "可能" / "也许" 等不确定词 → Faithfulness 偏低主因
  - 多步政策（"先 A 再 B"）只被检索到一步 → Recall 次因

第 2 周修法：

第 3 周评测：Faithfulness = 0.91、Recall = 0.92、Precision = 0.81、Latency p95 = 5.8s。

第 4 周成本平衡：Latency 涨了 1.6 秒（reranker + multi-query 各加 0.5-1s）。决策：

• 移除 multi-query → Latency 降 0.7s、Recall 微跌 1pp（Recall = 0.91）
• 保留 reranker → 它对 Precision 涨 16pp 是必需的

最终：Faithfulness 0.91、Recall 0.91、Precision 0.81、Latency p95 = 5.1s。所有指标过门禁、用户投诉率下降 40%。

这个完整流程展示了 RAG 调参的工程节奏：

flowchart LR
  A[初版上线] --> B[评测发现指标不达标]
  B --> C[失败 case 分类]
  C --> D[逐项 hypothesis + 修法]
  D --> E[重新评测验证]
  E --> F{所有指标过?}
  F -->|否| C
  F -->|是| G[成本/延迟权衡]
  G --> H[最终上线]
  style C fill:#fef3c7
  style G fill:#dbeafe
  style H fill:#dcfce7

每一步都是评测体系驱动的——失败 case 分类靠评测、修法效果靠评测、最终决策靠评测。如果没有这一整套指标信号，团队会陷入”改 prompt → 自己测两条 → 觉得好像变好了”的盲调状态——3 周后大概率指标退化更严重。

13.7.8 一份调试 checklist：当 RAG 评测异常时按这个走

把上面故事抽象成可勾选清单，新人按这个走能避免 80% 的重复踩坑：

□ Step 1: 跑完整评测, 拿到 4 个核心指标
□ Step 2: 把失败 case 按 (input, retrieved, output, expected) 导出
□ Step 3: 失败 case 按"哪个指标拉低"分桶
□ Step 4: Recall 桶 → 看是否长 chunk 被切; 短 chunk 上下文丢失
□ Step 5: Faithfulness 桶 → 看 generator 是否用了 context 没说的信息
□ Step 6: Precision 桶 → 看检索是否拿了无关 chunk
□ Step 7: 按 §13.3.0 优先级修法（chunk_size → embedding → top_k → reranker → query rewrite）
□ Step 8: 每改一个变量重跑评测, 留改动日志
□ Step 9: 多个改动叠加时, 注意检查指标是否互斥（Latency 涨太多）
□ Step 10: 成本/延迟权衡, 选定最终配置
□ Step 11: 写 ADR (Architecture Decision Record) 记录这次调参的决策
□ Step 12: 把这次发现的 hard case 反哺进黄金集（§3.6）

这份 checklist 应该贴在团队 wiki 上，新人入职第一天就能看到。它把”调试 RAG”这件事从”高级工程师的玄学”变成”可被任何工程师执行的流程”——这是评测体系给团队带来的第二个隐藏红利（第一个是质量信号本身）。

13.7.9 一个反直觉细节：reranker 不能盲目加

很多团队听说”reranker 能涨 Precision”就盲目加上去。但 reranker 在某些场景反而会降低整体效果——值得拆解一下原因：

• 小数据场景下 reranker 是过拟合：retriever 召回 top-3 的 chunk 已经基本相关，reranker 重排没什么改进空间
• 长 chunk 场景下 reranker 评分不稳定：rerank 模型多数训练数据是短段落，遇到 1k+ token 的 chunk 时分数 noise 大
• 多语言场景下 reranker 选错：用纯英文训练的 reranker 处理中文，效果差于直接 retriever

实操判断：先跑评测看 retriever 的 Precision 基线。Precision > 0.7 时 reranker 收益边际化、不值得加；Precision < 0.5 时加 reranker 几乎一定提升；Precision 0.5-0.7 之间需要实测对比。

这就是评测体系的核心价值——任何”听起来都该试”的优化，都要拿数据决定。

13.7.10 不要把”RAG 评测”等同于”找文档评测”

很多团队的 RAG 评测做着做着会陷入一种误区——只评测”找到正确文档”，忘了 RAG 的真正目标是”基于文档生成正确回答”。

具体表现：团队把 80% 工程精力投入到 retriever 调参、embedding 模型对比、reranker 选型，结果 generator 的 prompt 一直没改、出错率持续偏高。

这种偏差的根源：retriever 评测有清晰的指标（Recall / Precision）、易做 A/B；generator 评测要靠 LLM-judge、慢且贵。所以团队下意识地多调容易测的部分。

修法是设定两阶段预算：

• 50% 工程时间在 retriever（chunk size / embedding / reranker）
• 50% 工程时间在 generator（prompt / 模型 / 结构化输出）

不要让”哪个易测”决定”哪个该投入”。RAG 的最终质量是 retriever × generator 两者的乘积——任一方差，整体质量就差。第 §13.7.5 的”调参手册”已经体现这种平衡，但没显式写出来。

13.7.11 一个微妙细节：Faithfulness 高 ≠ 答案对

RAG 评测有一个看似反直觉的现象：Faithfulness 高（回答完全基于 context）不代表回答正确。

考虑这个场景：

Question: "Acme 公司今年营收多少?"
Context: "Acme 公司 2024 年营收 8000 万美元。" (注：context 本身是过期的, 实际 2025 年营收 1.2 亿)
Answer:  "Acme 公司今年营收 8000 万美元。"

Faithfulness: 1.0  (完全基于 context)
Correctness: 0.0   (答错了, 实际是 1.2 亿)

Faithfulness 只能保证”模型没编造”，但不能保证”context 本身正确”。如果 context 过期、错误、不完整，Faithfulness 高的答案仍然可能误导用户。

工程修法：

• Context 新鲜度：retriever 索引必须定期更新（每天 / 每周）
• Answer Correctness：与 reference 比对的指标，能补充 Faithfulness
• 多源验证：高合规场景（医疗 / 金融）要求 RAG 引用多个相关 chunk，单一来源不足以下结论

这一点是 RAG 评测的”高阶警觉”——团队搭好 Faithfulness 评测后还要再问一句”context 本身对吗”。

13.7.12 GraphRAG 评测：知识图谱时代的新挑战

2024 年微软推出 GraphRAG（arXiv:2404.16130）后，“在知识图谱上做 RAG”成为了主流方向之一。它的评测有几个独特挑战：

flowchart TB
  Q[Question] --> Decomp[问题分解]
  Decomp --> Sub1[子问题 1]
  Decomp --> Sub2[子问题 2]
  Sub1 --> KG1[图节点检索]
  Sub2 --> KG2[图节点检索]
  KG1 --> Path[图路径推理]
  KG2 --> Path
  Path --> Synth[答案综合]
  Synth --> A[Answer]
  Q -.评测.-> E1[Decomp 评测]
  Q -.评测.-> E2[Path 评测]
  Q -.评测.-> E3[Synth 评测]
  style Path fill:#fef3c7

GraphRAG 的评测维度比传统 RAG 多 3 个：

1. 问题分解正确性：是否合理拆分成子问题
2. 图路径合理性：从节点 A 到节点 B 的推理路径是否最优 / 准确
3. 多跳推理质量：跨多个图节点的综合答案

ragas 等传统 RAG 评测工具尚未原生支持这些维度——需要自己写自定义 metric。这是 2026 年评测领域的重要前沿之一。工业团队如果在做 GraphRAG，要预留专项评测的工程投入（约 1-2 人月）。

13.7.13 一个工程现实：Faithfulness 不应是唯一红线

最后给个反直觉的工程提醒——Faithfulness 太追求完美会损害 Helpfulness。

考虑场景：

Question: "我能加一份蘑菇吗?"
Context: "本店菜单不含蘑菇加点选项, 但接受食材自带。"
Answer A: "本店没有加蘑菇选项。" (Faithfulness=1.0, Helpfulness 中)
Answer B: "本店没有加蘑菇选项, 不过您可以自带蘑菇我们帮您加工。" (Faithfulness=1.0, Helpfulness 高)
Answer C: "本店没有加蘑菇选项, 但建议您可以试试附近 XX 餐厅。" (Faithfulness=0.5 - context 没说附近餐厅, Helpfulness 高)

A 和 B 都满分 Faithfulness，但 B 显然更好。C 的”违反 context”在某些场景下反而是 helpful 的——在不重要的细节上”超出 context”是用户期待的智能行为。

如果团队把 Faithfulness 阈值设到极严（如 ≥ 0.95），所有 C 类回答会被标失败，模型逐渐学会只给 A 类回答——产品质量在 Helpfulness 维度上反而退化。

修法：

• Faithfulness 阈值在不同场景分级：合规相关 ≥ 0.95、一般咨询 ≥ 0.85、闲聊 ≥ 0.70
• 配合 Helpfulness 指标：避免 Faithfulness 单极优化
• 业务专家 review 抽样：让人工判断哪些”超出 context 的回答”是好是坏

这是 RAG 评测从”机械合规”走向”业务智能”的关键认知。

13.7.14 Hybrid Search 的评测特殊性

工业 RAG 大量使用 Hybrid Search（结合 BM25 关键词检索 + dense vector 向量检索）。它的评测有自己的特点：

flowchart TB
  Q[Query] --> BM25[BM25 检索]
  Q --> Vec[向量检索]
  BM25 --> R1[关键词命中 top-K]
  Vec --> R2[语义相似 top-K]
  R1 --> Fusion[结果融合<br/>RRF / 加权]
  R2 --> Fusion
  Fusion --> Final[最终 top-K]
  Final -.分别评测.-> Sub[Sub-evaluations]
  Sub --> S1[BM25 单独 Recall]
  Sub --> S2[Vector 单独 Recall]
  Sub --> S3[Fusion 后 Recall]
  style Fusion fill:#fef3c7
  style Sub fill:#dcfce7

评测三个独立维度：

1. BM25 单独表现：判断关键词检索召回了多少
2. Vector 单独表现：判断语义检索召回了多少
3. Fusion 收益：混合后比单纯任一种好多少

这种分层评测让你能精确判断”哪一半在工作、哪一半在拖后腿”。常见发现：

• Vector 在专业术语场景表现差（医疗 / 法律领域专有名词）→ 需要 BM25 兜底
• BM25 在口语化场景表现差（用户表达模糊）→ 需要 vector 救场
• Fusion 权重需要业务专属调整：默认 RRF 不一定最优，要按业务集做扫描

ragas 等工具默认评测的是”端到端 RAG”，对 hybrid 内部分解不直接支持。需要团队自己把 BM25 / Vector 各自结果分别送评 —— 这是工业 RAG 评测里被忽视的细节。

13.7.15 一个 RAG 真实事故的反事实推演（结合本章方法）

回到第 1 章 §1.3 Bard 案——更细的反事实：如果 Google 当时的 Bard 是个 RAG 系统而非纯生成，本章的双层评测会怎样拦住事故？

假设场景：Bard 接入 NASA 的天文事实数据库做 RAG。用户问 “JWST 的发现”。

评测 1：Context Recall

• 期望 retriever 召回”JWST 主要发现”相关文档
• 但召回时如果检索到了 “JWST 拍到行星” 类相关文档（即使是太阳系内行星 / 模糊描述），Recall 看起来是高的
• 真实失败：retriever 没区分 “exoplanet first image”（VLT 2004）vs “JWST exoplanet observation”（不是 first）

评测 2：Faithfulness

• 如果模型答 “JWST 拍到第一张系外行星照片”，但 retrieved context 没说”第一张”——Faithfulness 评测应该判失败
• 这个评测能直接拦住事故

评测 3：Answer Correctness（与黄金答案对比）

• 黄金答案：“JWST 在系外行星观测领域有突破，但第一张直接成像是 2004 年的 VLT”
• 模型答 “JWST took the first picture” → Correctness 0

3 道评测中第 2 和第 3 都能拦住。Bard 没用 RAG（当时是纯生成）所以这套方法不直接适用，但故事说明：RAG + 完整评测 = 阻断事实错误的最强防线。这就是为什么 2024 年起几乎所有”知识型 chatbot”都迁移到 RAG 架构 + 配套评测体系。

13.7.16 一个 RAG 评测的工程边界：什么是它解决不了的

诚实告诉读者 RAG 评测的局限。即使做齐 Faithfulness + Recall + Precision + Relevance + Correctness 5 件套，仍有一些场景 RAG 评测无能为力：

1. 多源信息冲突：context 里两份文档说法不一致，评测无法判断 generator 选哪边对。需要”信源可信度”的额外维度
2. 时间敏感答案：context 来自 2024 年文档，用户问的是 2026 年情况——答案应该承认”这是 2024 年信息”，但评测难以判定”承认时间敏感”是否到位
3. 多模态混合：图文混合的 RAG（如医疗影像 + 病历），现有 ragas 只测文本，丢失图像信号
4. 隐式假设差异：用户问”我应该退货吗”，答案”应该 / 不应该”取决于用户隐式假设（金额？是否拆封？），评测难以覆盖
5. 创造性需求：用户希望 RAG 返回”基于这些资料，给我一个有创意的方案”——Faithfulness 高反而约束了创意

工程修法：

• 单 RAG 评测不够：上层用户体验评测（满意度 / NPS）作为补充
• 领域专家 review：高合规领域的”边界 case” 必须人工
• 持续 hard case mining：把生产里 RAG 评测覆盖不到的失败 case 反哺到下一版评测集

这些局限不是 RAG 评测体系的缺陷——是任何评测体系都不可能 100% 自动化的客观现实。承认局限 + 配套补充手段，比”试图完美自动化”务实得多。

13.7.17 一个团队的真实迭代节奏

把全章方法落到一个具体团队的迭代节奏（中等团队 / 客服 RAG 应用 / 月活 10 万用户）：

Week 1: PR merge -> CI 子集评测 (50 题) -> 通过即可上 staging
Week 1: 周二全集评测 -> 200 题黄金 + 100 题对抗 -> 监控指标
Week 1: 在线评测 1% 采样 Faithfulness -> 实时曲线
Week 1: 周五 hard case mining -> 把生产低分 trace 整理成 5-10 条新样例
Month: 月初元评测 -> 验证 grader 自身仍在 calibration 范围
Month: 月末复盘 -> 讨论本月失败 case 模式, 决定下月调优方向
Quarter: 季度刷新黄金集 -> 替换过时样例, 加入新业务场景
Year: 年度安全审计 -> 全套红队 + 合规检查 + Quality Gate review

这套节奏覆盖了从 PR 到年度的所有时间尺度。每条都对应本书一章 / 一节的方法。坚持执行 12 个月以上，团队会拥有一套自维持的 RAG 评测体系——不再依赖任何单个工程师的英雄时刻。

13.7.18 RAG 评测的”端到端 vs 分层”两种哲学

RAG 评测有两种核心哲学，影响整个评测体系的设计：

哲学 A：端到端评测（黑盒）

• 输入 query → 输出 answer，只看最终回答质量
• 优点：贴近用户体验、直接反映业务价值
• 缺点：失败时不知道是 retriever 还是 generator 的问题

哲学 B：分层评测（白盒）

• 拆 retriever / generator 各自评测
• 优点：失败定位精确、调参方向清晰
• 缺点：可能过度优化局部、忽略整体协同

工业实操：先做分层评测，理解每层贡献；再做端到端评测，验证整体质量。两者不是二选一，是按时间顺序的组合。

具体节奏：

• 第 1-3 月：以分层评测为主，调通 retriever / generator
• 第 4-6 月：加端到端评测，看真实业务表现
• 第 6 月之后：双轨并行，分层定位 + 端到端验证

这种”分层 + 端到端”的双视角，让 RAG 系统既能精确调试又能整体把控。是 RAG 评测体系成熟度的标志。

13.7.19 一个被遗漏的 RAG 评测维度：Citation 质量

很多 RAG 系统会在回答中显式引用 context 片段（如”根据政策第 3 条…”）。这种”citation”的质量是一个独立的评测维度——常被忽略：

• Citation 准确性：引用的内容是否真在 context 中
• Citation 完整性：所有需要 cite 的事实陈述都附了 citation
• Citation 易读性：用户能根据 citation 快速找到原文
• Citation 不滥用：不该 cite 的地方（如常识陈述）没有冗余 citation

判分实现：用 LLM-as-Judge 评估”answer 中的每条 citation 是否真的对应 context 中的内容”。或者用规则判分：parse citation 标记、提取被 cite 的 context 片段、做 substring match。

工业实务：高合规场景（医疗 / 法律 / 金融）必须做 citation 评测。普通场景如果产品提供 citation 功能，也建议做这一维度——citation 是用户 trust 的关键信号，错误 citation 比无 citation 更损害信任。

13.7.20 RAG 评测的”用户语义”维度：从工程指标到业务感受

回顾本章所有指标——Faithfulness / Recall / Precision / Relevance / Correctness / Citation——它们都是工程视角。但用户感受到的是另一些维度：

• 回答有用吗：能解决我的问题吗（不只是 faithful）
• 回答清晰吗：表述是否易懂
• 回答合适吗：长度 / 语气 / 格式与场景匹配
• 回答有礼貌吗：客服场景的关键

这些”业务感受”维度需要单独评测。修法：

1. 在评测集每条样例里加 expected_user_satisfaction 字段（1-5 分人工标注）
2. 用 LLM-judge 评估”answer 让用户满意吗”——参考人工标注做 calibration
3. 把这个 satisfaction 分作为辅助指标，与 Faithfulness 一起报告

发现一个常见现象：Faithfulness = 0.95 但 Satisfaction = 0.6 ——技术上”忠实”但用户体验差。例子：“你问的退货政策，请阅读官网政策第 3 章”——技术 faithful 但用户体验糟糕（用户希望直接得到答案）。

工程意义：RAG 评测体系最终要把工程指标和用户体验对齐。光看工程指标会优化出”机械正确但用户讨厌”的答案。这是 RAG 评测从”技术达标”走向”产品满意”的最后一公里。

13.7.21 RAG 评测的”健壮性”维度

除了主流的 Faithfulness / Recall / Relevance 等指标，工业 RAG 评测还应该包含健壮性维度——系统在异常输入下的表现。

测试矩阵：

每种异常的评测样例约 5-10 条。整套健壮性测试集约 50-100 条。每月跑一次，确保系统在生产中遇到异常输入时仍然合理。

工业实务：很多 RAG 系统在”理想 query”上 95% 通过，在异常输入上崩溃 30%。这种”理想环境表现良好但鲁棒性差”的状态是上线后事故的主要来源——本节方法学是工程上必须有的”防御性评测”。

13.7.22 RAG 评测体系的”成熟度自检”

读完本章后，团队可以用 7 项自检 RAG 评测体系成熟度：

□ 1. 有黄金集 ≥ 200 题且每月更新
□ 2. 跑双层指标 (retriever + generator) 而非合并
□ 3. 主指标 Faithfulness / Recall 阈值明确
□ 4. CI 集成，PR 评测 < 10 分钟
□ 5. 在线 trace 接入 + 1% 采样判分
□ 6. 每周 hard case mining
□ 7. 元评测季度跑一次

7 项全过 = 工业级 RAG 评测体系。3-5 项过 = 起步阶段。0-2 项过 = 急需建设。

工程团队的实操：贴在 wiki 上每季度 review 一次。这种”自检清单”让 RAG 评测从”主观感觉”变成”客观 checklist”——更容易跨团队对齐 / 跨季度追踪。

13.7.23 RAG 评测的”演化曲线”

回顾 RAG 评测从 2023 → 2026 的演化，给出”成熟度曲线”：

2023: 原始 RAG 评测——只看"answer 看起来对不对"
2023 末: ragas 出现, 引入 Faithfulness / Recall 等术语
2024 初: 工业团队开始接 LangSmith / Langfuse trace
2024 中: CI 集成 + 多指标体系成熟
2024 末: 元评测 / 红队 / GraphRAG 评测兴起
2025 初: 多模态 RAG 评测 + Agent-RAG 融合
2025 末: cost-aware 评测 + Pareto 前沿决策
2026 初: 自动化 hard case mining + 持续评测

每年都有新维度被加入。工程团队不必”一上来就做最完整的版本”——按业务规模选 3-5 项核心评测维度先跑起来，逐年扩展。

读完本章后，团队可以在评测体系成熟度曲线上定位自己——发现自己处于哪个时期、下一个该补什么、远期目标是什么。这种”地图视角”让 RAG 评测建设有了清晰路径。

13.7.24 一个工程现实：RAG 评测从来不能”完美”

最后一个工程现实——RAG 评测体系永远在”够用 vs 完美”之间。

任何评测体系都覆盖不了 100% 的失败模式。即使做了：

• 500 条黄金集
• 双层指标（retriever × generator）
• 在线 1% 采样
• 月度元评测
• 季度 hard case mining
• CI Quality Gate

仍然会有”没覆盖到的”用户场景偶尔失败。这是 LLM 时代的工程现实——不是评测体系不够好，是问题本身没有完美解。

工程团队的姿态：

1. 接受”100% 不可能”
2. 用资源覆盖 80-90% 主流场景
3. 剩下 10-20% 靠在线观测 + 人工抽查 + 用户反馈
4. 持续迭代，每月覆盖度+1-2pp

这种”够用且持续改进”的姿态比”追求完美”务实。完美主义是评测体系建设的最大障碍——很多团队卡在”还不够好不能上线”的纠结里，反而错过了”先 60 分上线 + 持续打磨到 90 分”的机会。

读完 RAG 评测一章，希望读者带走的是工程务实主义——评测是工具不是教条，60 分够用就先上、剩下慢慢补。

13.7.25 一份 RAG 评测的”年度成熟度阶梯”

把全章方法学整合成”团队 RAG 评测体系按年度演进”的阶梯：

Year 1 (新搭): 50 题黄金集 + Faithfulness + 离线评测 → 60 分
Year 2 (扩展): 200 题 + 双层指标 + 在线 trace → 75 分
Year 3 (成熟): 500 题 + 元评测 + CI Gate + 红队 → 85 分
Year 4 (规模): 2000 题 + GraphRAG / 多模态 + Agent 集成 → 90 分
Year 5+ (平台): SLA 化 + 跨业务 + 自动化 mining → 95 分

每升 1 阶段约 10pp 成熟度提升，需要 6-12 个月。完整 5 年路径让 RAG 评测体系从”够用”走向”标杆”。

工程团队的判断：定位自己处于哪个阶段、明确下个阶段的目标、预留资源。这种”长期视角”让 RAG 评测投入有了具体路线图。

13.7.26 RAG 评测的”产品-工程”协同

最后讨论一个组织维度——RAG 评测的产品-工程协同。

工业团队最常见的失败模式：

• 工程师专注 Faithfulness / Recall 等技术指标
• 产品团队专注用户满意度 / 转化率等业务指标
• 两套指标各自跟踪，不交流

结果：工程指标涨但业务指标降，团队相互指责。

修法：每月一次”产品-工程评测对齐会议”——

• 看工程指标曲线（Faithfulness / Recall）
• 看业务指标曲线（满意度 / 转化）
• 找两者的关联关系
• 调整工程优化方向

这种”双指标对齐”避免工程师”为了刷工程指标牺牲业务”的局部优化。读完本章希望读者带走的最后一个认知：评测最终服务业务，不只服务工程。

13.7.27 RAG 评测的”成本曲线”

工业 RAG 评测的成本随评测深度演化：

每升一档约 +¥2-5 万。这种”边际成本”判断让团队能精确决定”投入到什么深度”——不是越深越好、是匹配业务规模。

工程实务：把”评测投入档位”作为团队年度预算的明确决策项。每年 review 一次：当前档位是否匹配业务规模、要不要升档。这种”按业务规模匹配评测深度”的思路是工程务实主义的具体应用。

13.7.28 RAG 评测体系建设的”投入产出陷阱”

最后讨论一个工程现实——RAG 评测投入与产出的非线性陷阱。

陷阱 1：早期投入（0 → 60 分）边际产出最大

• 50 题黄金集 + Faithfulness → 拦下 40-60% 的明显失败
• ROI 极高、是必须做的部分

陷阱 2：中期投入（60 → 80 分）边际产出递减

• 双层评测 + 在线 trace → 多拦 15-20% 失败
• ROI 仍正向但需要明确投入价值

陷阱 3：高级投入（80 → 95 分）边际产出极低

• 元评测 / 红队 / 多模态 → 多拦 5-10% 失败
• 需要权衡：“最后 10pp 是否值得花 50% 工程时间”

关键认知：80-90 分是”工程务实主义”的甜蜜点。追求 95+ 的代价是指数增长的。

工程团队的判断：根据业务敏感度选择”投入终点”。普通业务到 80 分够用、高合规到 90 分必要、极敏感才需要 95+。这种”按需投入”避免”为了完美而过度投入”的常见错误。

13.7.29 RAG 评测的”工程价值层级”

最后给一个超出技术的观察——RAG 评测体系给团队的工程价值有 4 层：

Layer 4: 商业价值
  → 减少事故、提升用户体验、加速迭代
Layer 3: 团队价值
  → 团队对质量的共识、跨职能协作语言
Layer 2: 系统价值
  → RAG 系统的可靠性 / 可预测性 / 可改进
Layer 1: 工具价值
  → 知道当前指标、看到失败 case

绝大部分团队只看到 Layer 1（工具价值）。但真正的红利在 Layer 3-4——团队共识 + 商业价值。

读完本章希望读者带走的最高视角：RAG 评测建设是组织级的工程投入，价值远超”我们能跑出多少 metric”。这种价值视角让评测体系建设获得长期支持，不被短期成本压力放弃。

工业团队的实务：每年向管理层汇报评测体系时，强调 Layer 3-4 价值（团队共识 / 减少事故）而非只罗列 Layer 1 的指标数字。这种”价值翻译”能力让评测工程师的工作获得应有的组织重视。

13.7.30 RAG 评测的”读完誓言”

整章方法学覆盖 RAG 评测从入门到精通。读完后给读者一份”誓言”——

我（读者）在做 RAG 系统时——

• 不会跳过评测体系而追求功能堆叠
• 不会只做端到端而忽视分层评测
• 不会让 Faithfulness 失真损害 Helpfulness
• 不会让评测成本失控
• 不会忘记产品-工程协同
• 会持续 hard case mining
• 会承认 RAG 评测的局限并配合人工
• 会给团队留下清晰的评测体系传承

这份誓言看起来朴素，实际是 RAG 评测体系长期成功的关键。承诺写下来 + 团队公开 + 季度自查——能让承诺落到实处。

读完本章希望读者带走的最强承诺：评测不是技术问题，是工程纪律。承诺这种纪律，带头执行，带动团队——是 RAG 工程师的最高职业素养。

13.7.31 一份完整的 RAG retriever 调参评测脚本

整合本章方法学，给一份”对 retriever 做参数扫描评测”的完整 Python 脚本：

# retriever_grid_search.py
import json
from itertools import product
from ragas import evaluate
from ragas.metrics import ContextRecall, ContextPrecision
from datasets import Dataset

def grid_search_retriever(
    golden_set_path: str,
    chunk_sizes: list[int] = [256, 512, 1024],
    top_ks: list[int] = [3, 5, 10],
    embedding_models: list[str] = [
        "text-embedding-3-small",
        "text-embedding-3-large",
        "BAAI/bge-m3",
    ],
):
    """对 retriever 做 chunk_size × top_k × embedding 三轴扫描"""
    golden = [json.loads(l) for l in open(golden_set_path)]

    results = []
    for chunk_size, top_k, embed_model in product(chunk_sizes, top_ks, embedding_models):
        # 构建 retriever
        retriever = build_retriever(
            chunk_size=chunk_size,
            top_k=top_k,
            embedding_model=embed_model,
        )

        # 跑评测
        rows = []
        for sample in golden:
            contexts = retriever.retrieve(sample["query"])
            rows.append({
                "user_input": sample["query"],
                "retrieved_contexts": contexts,
                "reference": sample["expected"],
            })

        ds = Dataset.from_list(rows)
        result = evaluate(
            dataset=ds,
            metrics=[ContextRecall(), ContextPrecision()],
        )

        results.append({
            "chunk_size": chunk_size,
            "top_k": top_k,
            "embedding_model": embed_model,
            "context_recall": result["context_recall"],
            "context_precision": result["context_precision"],
            "f1": 2 * result["context_recall"] * result["context_precision"]
                  / (result["context_recall"] + result["context_precision"]),
        })

    return sorted(results, key=lambda x: -x["f1"])  # 按 F1 降序

# 主流程
if __name__ == "__main__":
    results = grid_search_retriever("data/golden/v1.jsonl")
    print("Top 5 configurations:")
    for r in results[:5]:
        print(f"  chunk={r['chunk_size']} k={r['top_k']} model={r['embedding_model']}: "
              f"R={r['context_recall']:.3f} P={r['context_precision']:.3f} F1={r['f1']:.3f}")

不到 50 行代码完成 retriever 三轴扫描（27 种组合 × 200 题 = 5400 次评测，约 30-60 分钟跑完）。输出按 F1 排序，团队直接取 top 配置上线。

工业实务：每次大改动（如换 embedding 模型 / chunk 重切）都跑一次 grid_search。这种”数据驱动调参”比”工程师拍脑袋”靠谱得多——比起花一周凭直觉调，1 小时 grid search 跑出客观最优。

读完本章希望读者带走的最具体行动：今天就跑一次 retriever grid search。1 小时投入，可能改善你 RAG 系统 5-10pp 的 Recall——这是评测体系最具体的 ROI。

13.7.32 一份完整的 RAG 端到端评测 + Grafana 推送脚本

整合本章方法学，给一份”RAG 端到端评测 + 自动推送指标到 Grafana”的完整脚本：

# rag_eval_full.py
import asyncio
from datasets import Dataset
from prometheus_client import Gauge, push_to_gateway, CollectorRegistry
from ragas import evaluate
from ragas.metrics import (
    Faithfulness, ContextRecall, ContextPrecision,
    AnswerRelevancy, AnswerCorrectness, NoiseSensitivity,
)

class RAGEvalRunner:
    """端到端 RAG 评测 + 监控推送"""
    THRESHOLDS = {
        "faithfulness": 0.85,
        "context_recall": 0.90,
        "context_precision": 0.70,
        "answer_relevance": 0.80,
        "answer_correctness": 0.80,
        "noise_sensitivity": 0.30,  # 越低越好
    }

    def __init__(self, rag_pipeline, golden_set_path: str):
        self.pipeline = rag_pipeline
        self.golden = self._load_jsonl(golden_set_path)
        self.registry = CollectorRegistry()
        self.gauges = {
            metric: Gauge(f"rag_{metric}", f"RAG {metric}", registry=self.registry)
            for metric in self.THRESHOLDS
        }

    async def run(self) -> dict:
        rows = []
        for sample in self.golden:
            contexts = await self.pipeline.retrieve(sample["query"])
            answer = await self.pipeline.generate(sample["query"], contexts)
            rows.append({
                "user_input": sample["query"],
                "retrieved_contexts": contexts,
                "response": answer,
                "reference": sample.get("expected", ""),
            })

        ds = Dataset.from_list(rows)
        result = evaluate(
            dataset=ds,
            metrics=[
                Faithfulness(), ContextRecall(), ContextPrecision(),
                AnswerRelevancy(), AnswerCorrectness(), NoiseSensitivity(),
            ],
        )

        # 推送到 Prometheus
        scores = result.to_pandas().mean(numeric_only=True).to_dict()
        for metric, value in scores.items():
            if metric in self.gauges:
                self.gauges[metric].set(value)
        push_to_gateway(
            "http://prom-pushgateway.internal:9091",
            job="rag_eval",
            registry=self.registry,
        )

        # 阈值检查
        failures = []
        for metric, threshold in self.THRESHOLDS.items():
            actual = scores.get(metric)
            if actual is None:
                continue
            if metric == "noise_sensitivity":
                if actual > threshold:
                    failures.append((metric, actual, threshold))
            else:
                if actual < threshold:
                    failures.append((metric, actual, threshold))

        return {
            "scores": scores,
            "failures": failures,
            "passed": len(failures) == 0,
        }

    def _load_jsonl(self, path):
        import json
        return [json.loads(l) for l in open(path)]


# 主流程：CI 触发或周度 cron
if __name__ == "__main__":
    from my_app import rag_pipeline
    runner = RAGEvalRunner(rag_pipeline, "data/golden/v1.jsonl")
    result = asyncio.run(runner.run())
    print(result)

    if not result["passed"]:
        for metric, actual, threshold in result["failures"]:
            print(f"FAIL: {metric}={actual:.3f} (threshold {threshold})")
        exit(1)

约 80 行代码完成 RAG 端到端评测的完整闭环：

• 端到端跑 retriever + generator
• 6 个核心 metric 计算（ragas）
• 阈值检查（含 noise_sensitivity 反向阈值）
• 自动推送 Prometheus（接 Grafana dashboard）
• CI 退出码（不达标阻止上线）

工业实务：把这份脚本接进 CI 第 2 层（merge 后跑全集），与第 4 章 §4.8.26 Grafana dashboard 配套使用。这是 RAG 评测体系工程化的”开箱即用”完整方案。

13.7.33 一份 RAG 失败诊断工具的完整实现

整合本章方法学，给一份”RAG 失败 case 自动诊断”工具：

# rag_diagnostic.py
from dataclasses import dataclass
from enum import Enum

class FailureMode(Enum):
    RETRIEVER_MISS = "retriever 没召回相关 chunk"
    RETRIEVER_NOISE = "retriever 召回太多无关 chunk"
    GENERATOR_HALLUCINATE = "generator 编造 context 没说的"
    GENERATOR_PARTIAL = "generator 只用了部分 context"
    GENERATOR_OFFTOPIC = "generator 答非所问"
    UNKNOWN = "需要人工 review"


@dataclass
class RAGTrace:
    query: str
    retrieved_contexts: list[str]
    response: str
    expected: str = ""
    scores: dict = None  # ragas 跑出的 metric


class RAGDiagnostic:
    """自动诊断 RAG 失败 case 属于哪类失败模式"""

    def diagnose(self, trace: RAGTrace) -> FailureMode:
        s = trace.scores or {}
        recall = s.get("context_recall", 0)
        precision = s.get("context_precision", 0)
        faithfulness = s.get("faithfulness", 0)
        relevance = s.get("answer_relevance", 0)

        # 决策树
        if recall < 0.5:
            return FailureMode.RETRIEVER_MISS
        if precision < 0.4:
            return FailureMode.RETRIEVER_NOISE
        if faithfulness < 0.5 and recall >= 0.7:
            return FailureMode.GENERATOR_HALLUCINATE
        if faithfulness >= 0.8 and recall >= 0.8 and not self._covers_all(trace):
            return FailureMode.GENERATOR_PARTIAL
        if relevance < 0.5:
            return FailureMode.GENERATOR_OFFTOPIC
        return FailureMode.UNKNOWN

    def _covers_all(self, trace: RAGTrace) -> bool:
        """简化判断：response 是否覆盖了 expected 的关键内容"""
        if not trace.expected:
            return True  # 无法判断
        expected_keywords = set(trace.expected.split())
        response_keywords = set(trace.response.split())
        coverage = len(expected_keywords & response_keywords) / len(expected_keywords)
        return coverage >= 0.6

    def suggest_fix(self, mode: FailureMode) -> list[str]:
        """根据失败模式给修法建议"""
        suggestions = {
            FailureMode.RETRIEVER_MISS: [
                "扩大 chunk overlap (256 → 100 token)",
                "提高 top_k (3 → 10)",
                "尝试更强 embedding (text-embedding-3-large 或 BGE-M3)",
                "做 query 重写 (HyDE / multi-query)",
            ],
            FailureMode.RETRIEVER_NOISE: [
                "加 reranker (BAAI/bge-reranker-v2)",
                "降低 chunk_size (1024 → 512)",
                "加更严格 metadata filter",
            ],
            FailureMode.GENERATOR_HALLUCINATE: [
                "在 prompt 加 'Use ONLY the provided context'",
                "升级到更强 model (mini → 4o)",
                "在 generation 后做 fact-check 二次验证",
            ],
            FailureMode.GENERATOR_PARTIAL: [
                "prompt 加 'List all relevant facts from context'",
                "增加 max_tokens",
                "在 prompt 给 reference 答案 demo",
            ],
            FailureMode.GENERATOR_OFFTOPIC: [
                "system prompt 加'Answer the user's question directly'",
                "前置一步 query 理解",
                "Generator 模型不稳, 升级",
            ],
            FailureMode.UNKNOWN: [
                "人工 review trace",
                "可能是评测集本身有问题",
            ],
        }
        return suggestions.get(mode, [])


# 使用
diagnostic = RAGDiagnostic()
for trace in failed_traces:
    mode = diagnostic.diagnose(trace)
    print(f"\n{trace.query}")
    print(f"  失败模式: {mode.value}")
    print(f"  建议修法:")
    for fix in diagnostic.suggest_fix(mode):
        print(f"    - {fix}")

约 90 行代码完成 RAG 失败诊断的完整自动化：

• 6 种失败模式分类（retriever miss / noise / generator hallucinate / partial / offtopic / unknown）
• 基于 metric 的自动决策树
• 每种失败模式的具体修法建议（共 18 条）

工业实务：每周对失败 case 跑一次 diagnostic，按失败模式分组归纳——“本周失败 60% 是 retriever miss、30% 是 generator hallucinate”——比”凭感觉调”靠谱得多。

13.7.34 RAG 评测的”Embedding 模型选型”实验设计模板

RAG 系统里 embedding 模型的选择对 retriever 性能影响最大——但工程团队往往凭”听说 BGE 不错”就拍板，缺乏量化证据。下面是一份可直接套用的对照实验模板，覆盖中文 RAG 最常用的 4 类 embedding：

import json
import time
from dataclasses import dataclass
from typing import Callable

@dataclass
class EmbeddingExperimentResult:
    model_id: str
    dim: int
    recall_at_5: float
    mrr: float
    ndcg_at_10: float
    qps: float
    cost_per_1m_tokens_usd: float
    storage_mb_per_1m_chunks: float

class EmbeddingComparator:
    """对 N 个 embedding 在同一份 RAG 评测集上做严格对照"""

    def __init__(self, eval_set: list[dict], chunks: list[dict]):
        self.eval_set = eval_set
        self.chunks = chunks
        self.results: list[EmbeddingExperimentResult] = []

    def _index(self, encode_fn: Callable[[list[str]], list[list[float]]]):
        texts = [c["text"] for c in self.chunks]
        return encode_fn(texts)

    def _retrieve(self, query_vec, chunk_vecs, top_k: int) -> list[int]:
        import numpy as np
        chunk_arr = np.array(chunk_vecs)
        scores = chunk_arr @ np.array(query_vec)
        return np.argsort(-scores)[:top_k].tolist()

    def evaluate(self, model_id: str, dim: int,
                 encode_fn: Callable[[list[str]], list[list[float]]],
                 cost_per_1m_tokens: float):
        chunk_vecs = self._index(encode_fn)
        recall_hits, rr_sum, ndcg_sum = 0, 0.0, 0.0
        start = time.monotonic()
        for query in self.eval_set:
            qv = encode_fn([query["question"]])[0]
            top10 = self._retrieve(qv, chunk_vecs, top_k=10)
            top5 = top10[:5]
            gold = set(query["gold_chunk_ids"])
            if any(self.chunks[i]["id"] in gold for i in top5):
                recall_hits += 1
            for rank, idx in enumerate(top10, start=1):
                if self.chunks[idx]["id"] in gold:
                    rr_sum += 1.0 / rank
                    ndcg_sum += 1.0 / (rank ** 0.5)
                    break
        elapsed = time.monotonic() - start
        n = len(self.eval_set)
        self.results.append(EmbeddingExperimentResult(
            model_id=model_id, dim=dim,
            recall_at_5=recall_hits / n,
            mrr=rr_sum / n,
            ndcg_at_10=ndcg_sum / n,
            qps=n / max(elapsed, 0.001),
            cost_per_1m_tokens_usd=cost_per_1m_tokens,
            storage_mb_per_1m_chunks=(dim * 4 * 1_000_000) / (1024 * 1024),
        ))

    def export_table(self) -> str:
        header = "| model | dim | recall@5 | mrr | ndcg@10 | qps | $/1M | MB/1M |"
        sep = "|---|---|---|---|---|---|---|---|"
        rows = [f"| {r.model_id} | {r.dim} | {r.recall_at_5:.3f} | "
                f"{r.mrr:.3f} | {r.ndcg_at_10:.3f} | {r.qps:.1f} | "
                f"${r.cost_per_1m_tokens_usd} | {r.storage_mb_per_1m_chunks:.0f} |"
                for r in self.results]
        return "\n".join([header, sep, *rows])

flowchart LR
  E[评测集 100-500 题] --> EX[EmbeddingComparator]
  C[语料 chunks] --> EX
  EX -->|encode_fn| E1[BGE-large-zh]
  EX -->|encode_fn| E2[bce-embedding]
  EX -->|encode_fn| E3[OpenAI text-embedding-3]
  EX -->|encode_fn| E4[Cohere embed-multilingual]
  E1 --> M[recall/mrr/ndcg/qps/cost/storage]
  E2 --> M
  E3 --> M
  E4 --> M
  M --> D[选型决策表]

  style D fill:#e8f5e9

实验设计要点：

• 同一份评测集 + 同一份语料——只换 embedding，控制变量
• 6 个维度：recall@5（核心）/ MRR / nDCG@10 / QPS（吞吐）/ 单位 token 成本 / 单位规模存储
• 存储估算 = dim × 4 bytes × N——dim 1024 比 dim 768 多 33% 存储，必须算
• 不该看：单点对比——必须跑同一份至少 100 题的评测集才有统计意义

工业实务的选型经验：

• BGE-large-zh-v1.5 在中文 RAG 通常 recall@5 = 0.78-0.85（开源免费）
• bce-embedding-base 中英混排场景常胜（来自有道）
• OpenAI text-embedding-3-large recall 高但成本高 6 倍
• Cohere embed-multilingual 国际化场景最稳

工程实务：把这份脚本作为团队”换 embedding”的标准动作——任何”切到新 embedding”的提议都先跑这套对比，结果进 PR 描述。半年下来团队就有了私有的 embedding 选型 trace。

13.7.35 RAG 评测的”chunk size / overlap”敏感度网格

embedding 之外，影响 retriever 性能最大的两个参数是 chunk_size（每段切多大）和 chunk_overlap（相邻段的重叠）。下面是一份网格搜索脚本，可在 1 小时内得出”我们语料的最优切分参数”。

import json
from itertools import product
from dataclasses import dataclass
from typing import Callable

@dataclass
class ChunkingResult:
    chunk_size: int
    overlap: int
    total_chunks: int
    avg_chunk_chars: int
    recall_at_5: float
    context_precision: float
    context_recall: float
    storage_mb_estimate: float
    score: float

class ChunkingGridSearch:
    """对一份语料 + 评测集做 chunk_size × overlap 网格搜索"""

    def __init__(self, raw_corpus: list[dict], eval_set: list[dict],
                 splitter_fn: Callable, embed_fn: Callable):
        self.corpus = raw_corpus
        self.eval_set = eval_set
        self.splitter = splitter_fn
        self.embed = embed_fn

    def _split_corpus(self, chunk_size: int, overlap: int) -> list[dict]:
        chunks = []
        for doc in self.corpus:
            for i, frag in enumerate(self.splitter(doc["text"],
                                                   chunk_size, overlap)):
                chunks.append({
                    "id": f"{doc['id']}-{i}",
                    "doc_id": doc["id"],
                    "text": frag,
                })
        return chunks

    def _evaluate(self, chunks: list[dict]) -> tuple[float, float, float]:
        chunk_vecs = self.embed([c["text"] for c in chunks])
        recall_hits, prec_sum, rec_sum = 0, 0.0, 0.0

        for query in self.eval_set:
            qv = self.embed([query["question"]])[0]
            scores = self._cosine(qv, chunk_vecs)
            top5 = sorted(range(len(scores)),
                          key=lambda i: -scores[i])[:5]
            top_doc_ids = {chunks[i]["doc_id"] for i in top5}
            gold_docs = set(query["gold_doc_ids"])

            if top_doc_ids & gold_docs:
                recall_hits += 1
            if top_doc_ids:
                prec_sum += len(top_doc_ids & gold_docs) / len(top_doc_ids)
            if gold_docs:
                rec_sum += len(top_doc_ids & gold_docs) / len(gold_docs)

        n = len(self.eval_set)
        return (recall_hits / n, prec_sum / n, rec_sum / n)

    def _cosine(self, q, vecs):
        import math
        def norm(v): return math.sqrt(sum(x * x for x in v))
        qn = norm(q)
        return [sum(qx * vx for qx, vx in zip(q, vec)) /
                (qn * norm(vec) + 1e-9) for vec in vecs]

    def run(self, chunk_sizes: list[int],
            overlaps: list[int]) -> list[ChunkingResult]:
        results = []
        for cs, ov in product(chunk_sizes, overlaps):
            if ov >= cs:
                continue
            chunks = self._split_corpus(cs, ov)
            recall5, prec, rec = self._evaluate(chunks)
            avg_chars = sum(len(c["text"]) for c in chunks) / len(chunks)
            score = 0.4 * recall5 + 0.3 * prec + 0.3 * rec
            results.append(ChunkingResult(
                chunk_size=cs,
                overlap=ov,
                total_chunks=len(chunks),
                avg_chunk_chars=int(avg_chars),
                recall_at_5=round(recall5, 3),
                context_precision=round(prec, 3),
                context_recall=round(rec, 3),
                storage_mb_estimate=len(chunks) * 768 * 4 / 1024 / 1024,
                score=round(score, 3),
            ))
        return sorted(results, key=lambda r: -r.score)

flowchart LR
  C[语料库 raw_corpus] --> G[ChunkingGridSearch]
  E[评测集 100-500 题] --> G
  G -->|cs=128 ov=0| R1[ChunkingResult]
  G -->|cs=256 ov=32| R2[ChunkingResult]
  G -->|cs=512 ov=64| R3[ChunkingResult]
  G -->|cs=1024 ov=128| R4[ChunkingResult]
  R1 --> SC[score = 0.4 recall + 0.3 prec + 0.3 rec]
  R2 --> SC
  R3 --> SC
  R4 --> SC
  SC --> BEST[排序后选最高 score]

  style BEST fill:#e8f5e9

工程实务的 4 个关键经验：

• 不必跑 cs ≥ overlap 之外的组合——overlap > size 无意义
• 典型搜索网格：cs ∈ [128, 256, 512, 1024] × overlap ∈ [0, 32, 64, 128, 256] = 16 个组合
• score 的权重选择：0.4 × recall + 0.3 × prec + 0.3 × rec——recall 略重于 prec，因为 RAG 业务里召不回的代价大于 noise
• 复跑 3 次取中位数——embedding 推理有少量随机性

工业经验中常见结论：

• 短文档（FAQ / 客服日志，平均 < 500 字）：cs=256/ov=32 通常最优
• 长文档（PDF 报告 / 法律条文）：cs=512/ov=64 更稳
• 表格 / 代码：cs=1024/ov=0 + 边界对齐切分

把 ChunkingGridSearch 挂到 PR check：任何”调整 chunk_size”的代码改动都自动跑该网格——结果与 baseline 比对差异，不达标卡 PR。这是 RAG 系统”参数演化”的工程化沉淀。

13.7.36 一份”RAG 引文质量”评测 —— 答案的每一句话都该带出处

合规场景（医疗 / 法律 / 金融 / 政府）的 RAG 不仅要 Faithfulness 高，还要每一句话都标明引文（citation）。这与 ChatGPT、Perplexity 等公开产品的”带 [^1] 标记”形态完全一致。下面是一份针对引文质量的专项评测：

import re
from dataclasses import dataclass
from typing import Iterable

@dataclass
class CitationCheck:
    answer: str
    sentences: list[str]
    cited_sentences: int
    citation_density: float
    invalid_refs: list[str]
    fabricated_refs: list[str]
    coverage_score: float

class RAGCitationEvaluator:
    """评测 RAG 答案中引文的覆盖率、准确性、虚假引用"""

    CITATION_RE = re.compile(r"\[\^(\w+)\]|\[(\d+)\]|\(来源: ([^)]+)\)")

    def __init__(self, valid_doc_ids: set[str]):
        self.valid_doc_ids = valid_doc_ids

    def _split_sentences(self, text: str) -> list[str]:
        text = text.replace("\n", " ")
        parts = re.split(r"(?<=[。.!?！？])\s+", text)
        return [p.strip() for p in parts if p.strip() and len(p) > 5]

    def _extract_refs(self, sentence: str) -> list[str]:
        refs = []
        for m in self.CITATION_RE.finditer(sentence):
            refs.append(next((g for g in m.groups() if g), ""))
        return refs

    def _is_valid_ref(self, ref: str) -> bool:
        return ref in self.valid_doc_ids

    def evaluate(self, answer: str,
                 retrieved_doc_ids: list[str]) -> CitationCheck:
        sentences = self._split_sentences(answer)
        cited_count = 0
        all_refs, invalid_refs, fabricated = [], [], []

        for sent in sentences:
            refs = self._extract_refs(sent)
            if refs:
                cited_count += 1
                for r in refs:
                    all_refs.append(r)
                    if not self._is_valid_ref(r):
                        invalid_refs.append(r)
                    elif r not in retrieved_doc_ids:
                        fabricated.append(r)

        density = cited_count / max(len(sentences), 1)
        coverage = (1.0 if not all_refs else
                    1 - (len(invalid_refs) + len(fabricated)) / len(all_refs))
        return CitationCheck(
            answer=answer,
            sentences=sentences,
            cited_sentences=cited_count,
            citation_density=round(density, 3),
            invalid_refs=invalid_refs,
            fabricated_refs=fabricated,
            coverage_score=round(coverage, 3),
        )

    def aggregate(self, checks: Iterable[CitationCheck]) -> dict:
        checks = list(checks)
        n = len(checks)
        return {
            "n": n,
            "avg_density": sum(c.citation_density for c in checks) / max(n, 1),
            "avg_coverage": sum(c.coverage_score for c in checks) / max(n, 1),
            "fabrication_rate": sum(1 for c in checks if c.fabricated_refs) / max(n, 1),
            "invalid_ref_rate": sum(1 for c in checks if c.invalid_refs) / max(n, 1),
        }

flowchart LR
  A[RAG 答案] --> S[切句]
  S --> E[逐句抽 citation]
  E --> R{引文?}
  R -->|有| V{有效 doc_id?}
  R -->|无| NO[句子未引]
  V -->|否| INV[invalid_ref]
  V -->|是| RT{在 retrieved 中?}
  RT -->|否| FAB[fabricated 编造引用]
  RT -->|是| OK[合法引文]
  INV --> SC[coverage 降分]
  FAB --> SC
  OK --> SC
  SC --> AGG[聚合 fabrication_rate]

  style FAB fill:#ffebee
  style INV fill:#fff3e0
  style OK fill:#e8f5e9

3 类常见失败模式：

工程实务的 4 条上线红线（合规 RAG 场景）：

• avg_density ≥ 0.7：至少 70% 的句子都该有引文
• avg_coverage ≥ 0.95：95% 引文都该有效（指向 retrieval 命中文档）
• fabrication_rate < 1%：绝对禁线
• invalid_ref_rate < 2%：可容忍极少的 prompt 工程导致的格式错乱

研究背景：

• Liu et al. 2023 “Evaluating Verifiability in Generative Search Engines” arXiv:2304.09848 系统给出 citation 评测方法
• Perplexity AI 在 2024-Q3 工程博客披露其内部”citation hallucination rate”是上线 gate
• Anthropic Claude 3 Model Card §6 提到 “citations as a built-in capability”

合规场景上线 RAG 时把 §13.7.34 embedding + §13.7.35 chunking + §13.7.36 citation 三个评测串联——retriever / generator / 引用 三层全过才放行。这是医疗、法律、金融 RAG 系统从原型到合规生产的标准路径。

13.7.37 RAG 评测的”复杂查询能力分级”——简单 vs 多跳 vs 推理

公开 benchmark 喜欢用 single-hop 题（“今年北京的 GDP 是多少？“）评测 RAG，但生产里最难的是 multi-hop / 推理类查询（“北京 GDP 5 年增长是上海的几倍？”）。下面是一份覆盖 4 类难度的评测设计：

import asyncio
from dataclasses import dataclass
from typing import Callable, Awaitable
from collections import defaultdict

@dataclass
class DifficultyProbe:
    probe_id: str
    level: str
    query: str
    required_chunk_ids: list[str]
    expected_facts: list[str]
    expected_reasoning_steps: int

@dataclass
class DifficultyResult:
    probe_id: str
    level: str
    retriever_recall: float
    facts_grounded: int
    reasoning_steps_present: bool
    fully_correct: bool

class RAGDifficultyEvaluator:
    """RAG 系统在 4 个难度等级上的能力评估"""

    def __init__(self, retriever: Callable, generator: Callable[[str, list], Awaitable[str]]):
        self.retriever = retriever
        self.generator = generator

    def _check_facts(self, response: str, expected_facts: list[str]) -> int:
        return sum(1 for f in expected_facts if f.lower() in response.lower())

    def _detect_steps(self, response: str, min_steps: int) -> bool:
        markers = ["首先", "其次", "然后", "因此", "因为", "假设", "推断",
                   "step 1", "step 2", "first", "then", "therefore"]
        hits = sum(1 for m in markers if m in response.lower())
        return hits >= min_steps

    async def evaluate(self, probe: DifficultyProbe) -> DifficultyResult:
        retrieved = self.retriever(probe.query, top_k=5)
        retrieved_ids = {c["id"] for c in retrieved}
        gold = set(probe.required_chunk_ids)
        recall = len(retrieved_ids & gold) / max(len(gold), 1)

        response = await self.generator(probe.query, retrieved)
        grounded = self._check_facts(response, probe.expected_facts)
        steps_ok = self._detect_steps(response, probe.expected_reasoning_steps)

        fully = (recall >= 0.8 and
                 grounded >= len(probe.expected_facts) * 0.8 and
                 steps_ok)

        return DifficultyResult(
            probe_id=probe.probe_id,
            level=probe.level,
            retriever_recall=round(recall, 3),
            facts_grounded=grounded,
            reasoning_steps_present=steps_ok,
            fully_correct=fully,
        )

    async def evaluate_all(self, probes: list[DifficultyProbe]) -> dict:
        results = await asyncio.gather(*(self.evaluate(p) for p in probes))
        by_level = defaultdict(list)
        for r in results:
            by_level[r.level].append(r)
        return {
            "by_level": {lvl: {
                "n": len(rows),
                "fully_correct_rate": sum(r.fully_correct for r in rows) / len(rows),
                "avg_retriever_recall": sum(r.retriever_recall for r in rows) / len(rows),
            } for lvl, rows in by_level.items()},
            "overall_correct_rate": sum(r.fully_correct for r in results) / len(results),
        }

flowchart LR
  P[probes 100 题<br/>L1~L4 各 25] --> E[Evaluator]
  E --> R{retriever_recall ≥ 0.8?}
  R -->|否| FAIL[retriever 不足]
  R -->|是| G{facts grounded ≥ 80%?}
  G -->|否| GEN[generator 漏要点]
  G -->|是| ST{reasoning_steps?}
  ST -->|缺失| RES[reasoning 弱]
  ST -->|完整| OK[fully_correct]
  OK --> AGG[按 L1~L4 聚合]
  FAIL --> AGG
  GEN --> AGG
  RES --> AGG

  style OK fill:#e8f5e9
  style FAIL fill:#ffebee

工程实务的 4 条阶梯式期待：

具体例子：

• 用户问 “中国 2024 GDP 增速比 2023 高多少？” → L2 multi-hop
• retrieved 命中 2024 + 2023 两条 chunk
• LLM 答 “2024 增速 5.0%、2023 增速 5.2%、差 -0.2pp”
• 评测：recall ≥ 80% ✅、3 个 fact 全 grounded ✅、step “因此差 -0.2”识别 ✅ → fully_correct

如果 LLM 错答 “差 0.3pp”（数值错），即使 recall 高、step 完整也判 ❌ —— 这正是该 evaluator 的关键价值：捕捉 generator 对 retrieved context 的推理失败，而不只是查 retriever 是否拿到。

研究背景：

• HotpotQA（Yang et al. arXiv:1809.09600）是 multi-hop QA 的经典 benchmark
• MuSiQue（Trivedi et al. arXiv:2108.00573）专门评测 2-4 跳推理
• ragas TestsetGenerator 的 MultiHopAbstractQuerySynthesizer（§11.7.43）能自动合成 L2-L3 难度题

读者把这套 evaluator 接入团队 RAG 评测——每周看 L1~L4 4 个分数——能看清”我们的 RAG 在哪类查询上仍弱”。生产里大量 L3-L4 类查询失败往往才是用户骂”AI 像智障”的真实原因。

13.7.38 RAG “Reranker”贡献度评测——为何要花钱加 cross-encoder

很多 RAG 系统在 embedding retriever 之上加一层 reranker（如 Cohere Rerank、bge-reranker-large）—— 但加了 reranker 真的提升 recall@K 吗？需要量化测算后才能判断”$X 成本是否值得”。下面是一份对照评测：

import asyncio
from dataclasses import dataclass
from typing import Iterable, Callable, Awaitable

@dataclass
class RerankerComparisonResult:
    config: str
    recall_at_5: float
    mrr: float
    avg_relevance_score_at_top1: float
    avg_latency_ms: float
    cost_per_query_usd: float
    score_per_dollar: float

class RerankerEvaluator:
    """对比 retriever vs retriever + reranker 的端到端贡献"""

    def __init__(self, retriever: Callable, reranker_fn: Callable | None,
                 reranker_cost_per_call_usd: float = 0.001):
        self.retriever = retriever
        self.reranker = reranker_fn
        self.reranker_cost = reranker_cost_per_call_usd

    async def _retrieve_only(self, query: str, top_k: int = 5) -> list[dict]:
        return self.retriever(query, top_k=top_k)

    async def _retrieve_then_rerank(self, query: str,
                                    initial_k: int = 30,
                                    final_k: int = 5) -> list[dict]:
        candidates = self.retriever(query, top_k=initial_k)
        reranked = await self.reranker(query, candidates)
        return reranked[:final_k]

    def _recall_at_k(self, retrieved: list[dict],
                      gold_ids: set[str]) -> float:
        retrieved_ids = {c["id"] for c in retrieved}
        return 1.0 if (retrieved_ids & gold_ids) else 0.0

    def _mrr(self, retrieved: list[dict], gold_ids: set[str]) -> float:
        for rank, doc in enumerate(retrieved, start=1):
            if doc["id"] in gold_ids:
                return 1.0 / rank
        return 0.0

    async def evaluate(self, eval_set: list[dict],
                        config: str = "no_reranker") -> RerankerComparisonResult:
        import time
        recall_sum = mrr_sum = top1_score_sum = 0
        latency_sum = 0
        n = len(eval_set)
        for query in eval_set:
            t0 = time.monotonic()
            if config == "no_reranker":
                retrieved = await self._retrieve_only(
                    query["question"], top_k=5)
            else:
                retrieved = await self._retrieve_then_rerank(
                    query["question"], initial_k=30, final_k=5)
            latency_sum += (time.monotonic() - t0) * 1000
            gold = set(query["gold_chunk_ids"])
            recall_sum += self._recall_at_k(retrieved, gold)
            mrr_sum += self._mrr(retrieved, gold)
            top1_score_sum += retrieved[0].get("score", 0) if retrieved else 0

        cost = (self.reranker_cost if config != "no_reranker" else 0) * n
        return RerankerComparisonResult(
            config=config,
            recall_at_5=round(recall_sum / n, 3),
            mrr=round(mrr_sum / n, 3),
            avg_relevance_score_at_top1=round(top1_score_sum / n, 3),
            avg_latency_ms=round(latency_sum / n, 1),
            cost_per_query_usd=round(cost / n, 5),
            score_per_dollar=round(
                (recall_sum / n) /
                max(cost / n + 0.0005, 1e-6), 0),
        )

flowchart LR
  Q[query] --> EM[embedding 召回 top 30]
  EM --> A1[no_reranker: 直接 top 5]
  EM --> RR[reranker top 30 → top 5]

  A1 --> EV1[eval no_reranker]
  RR --> EV2[eval with_reranker]

  EV1 --> CMP[recall / MRR / latency / cost 对比]
  EV2 --> CMP
  CMP --> DEC{score_per_dollar 涨?}
  DEC -->|是| KEEP[保留 reranker]
  DEC -->|否| DROP[砍 reranker]

  style KEEP fill:#e8f5e9
  style DROP fill:#ffebee

工程实务的 4 条 reranker 决策准则：

具体例子：

洞察：bge-reranker-large 性价比是 Cohere 的 2.4x。除非 score 差距 > 5pp，否则用 bge。这种数字化决策替代”凭直觉觉得 Cohere 更好”。

工程实务：每月跑一次 reranker 评测——市场上 reranker 库每月有新选项（jina / mixedbread / cohere-3 …），用本评测做选型 sanity check。

研究背景：

• Cohere 在 2024-Q3 公布 Rerank-3.5 数据：在 BEIR benchmark recall@10 涨 + 12pp
• bge-reranker-large（智源）作为开源重排第一梯队
• ColBERT v2（Khattab et al. arXiv:2112.01488）是 late-interaction reranker 的经典

读者把这套 RerankerEvaluator 与 §13.7.34 embedding 评测、§13.7.35 chunking 评测、§13.7.36 citation 评测组合——RAG 系统每个环节都有量化决策依据。

13.7.39 RAG 评测的”端到端 vs 分层”两条决策路径整合

§13.7.18 提到双层与端到端两种哲学，但生产团队最常问的问题是”我什么时候该看哪种”。下面给出整合性的决策路径——每种 case 都有明确指引：

flowchart TB
  S[评测意图?] --> Q1{是否 debug?}
  Q1 -->|是| LAYER[分层评测<br/>retriever / generator 单独看]
  Q1 -->|否，是 release gate| Q2{是 PR check?}
  Q2 -->|是 PR| E2E[端到端 + answer correctness]
  Q2 -->|否，是模型选型| Q3{对比维度?}
  Q3 -->|"看综合能力"| FULL[端到端 + 多 metric]
  Q3 -->|"看具体短板"| LAYER

  Q1 -->|否| Q4{用户骂?}
  Q4 -->|是，看哪环出问题| LAYER
  Q4 -->|否，但成本担忧| COST[Cost analyzer + 分层]

  LAYER --> A1[ragas Faithfulness + Recall + Precision]
  E2E --> A2[Answer Correctness + Citation]
  FULL --> A3[全套 metrics × hard subset]
  COST --> A4[per-step cost breakdown]

  style LAYER fill:#e3f2fd
  style E2E fill:#e8f5e9
  style FULL fill:#fff3e0

from dataclasses import dataclass

@dataclass
class EvalDecision:
    intent: str
    primary_layer: str
    primary_metrics: list[str]
    sample_size: int
    estimated_cost_usd: float
    estimated_time_min: int

class RAGEvalRouter:
    """根据评测意图自动选择评测路径"""

    DECISION_TABLE = {
        ("debug", "any"): {
            "layer": "layered",
            "metrics": ["faithfulness", "context_recall",
                          "context_precision"],
            "n": 50, "cost_per_q": 0.012, "min": 8,
        },
        ("release_gate", "pr"): {
            "layer": "end_to_end",
            "metrics": ["answer_correctness", "citation_coverage"],
            "n": 200, "cost_per_q": 0.018, "min": 25,
        },
        ("model_selection", "comprehensive"): {
            "layer": "full_battery",
            "metrics": ["faithfulness", "answer_relevance",
                          "context_recall", "answer_correctness",
                          "latency_p99", "cost_per_query"],
            "n": 500, "cost_per_q": 0.025, "min": 60,
        },
        ("model_selection", "specific_weakness"): {
            "layer": "layered",
            "metrics": ["context_recall", "context_precision"],
            "n": 200, "cost_per_q": 0.010, "min": 20,
        },
        ("user_complaint", "any"): {
            "layer": "layered",
            "metrics": ["context_recall", "faithfulness"],
            "n": 30, "cost_per_q": 0.012, "min": 5,
        },
        ("cost_review", "any"): {
            "layer": "layered_with_cost",
            "metrics": ["cost_per_query", "context_precision"],
            "n": 100, "cost_per_q": 0.010, "min": 12,
        },
    }

    def route(self, intent: str, context: str = "any") -> EvalDecision:
        config = self.DECISION_TABLE.get((intent, context),
                                          self.DECISION_TABLE.get((intent, "any"),
                                                                    self.DECISION_TABLE[("debug", "any")]))
        return EvalDecision(
            intent=intent,
            primary_layer=config["layer"],
            primary_metrics=config["metrics"],
            sample_size=config["n"],
            estimated_cost_usd=round(config["n"] * config["cost_per_q"], 2),
            estimated_time_min=config["min"],
        )

工程实务的 4 条路由原则：

具体例子：某团队的”模型选型周”路由 4 次评测：

• Day 1：路由”PR check + 端到端” → 跑 200 题 / 25 分钟 / $3.6
• Day 2：路由”comprehensive 模型对比” → 跑 500 题 × 3 模型 / 60 分钟 × 3 / $37.5
• Day 3：路由”specific_weakness 长尾”→ 跑 200 题 hard subset / 20 分钟 / $2
• Day 4：路由”cost_review” → 100 题 / 12 分钟 / $1

总成本 ~$45，时间总计约 4 小时（异步并发）。比”全量跑”省 60%。

研究背景：

• ragas 在 2024-Q4 论文 §6 公开了”4 类典型评测意图与对应工具组合”
• LangChain 的 RAGAS integration 文档把”评测意图 → 工具”作为推荐使用模式
• DBRX 团队公开过他们 RAG 评测路由的内部决策树

把 RAGEvalRouter 接入团队的 evals CLI——任何”我要评测”的需求都先经路由器给出”该跑哪种 / 多少 / 多久 / 多少钱”，把模糊需求转成具体计划。这是 RAG 评测从”凭经验跑”到”工程化路由”的最后一步。

13.7.40 RAG 评测的”假阳性 / 假阴性”分析——避免被指标骗

RAG 评测的 confusion matrix 比单纯 accuracy 重要——尤其在合规场景。“假阳性”（faithfulness 高但实际错）的代价远高于”假阴性”（faithfulness 低但实际对）。下面是一份双向错误分析：

import asyncio
from dataclasses import dataclass
from collections import Counter
from typing import Iterable

@dataclass
class ConfusionMatrixResult:
    metric_name: str
    true_positive: int      # eval 判对 + 实际对
    false_positive: int     # eval 判对 + 实际错（最危险）
    true_negative: int      # eval 判错 + 实际错
    false_negative: int     # eval 判错 + 实际对（无害但浪费）
    precision: float
    recall: float
    f1: float
    fp_rate: float
    fn_rate: float

class RAGConfusionMatrixAnalyzer:
    """跑 RAG 指标 vs 人工 ground-truth 的 confusion matrix"""

    def analyze(self, metric_results: list[dict],
                 human_judgments: list[dict],
                 metric_threshold: float = 0.7) -> ConfusionMatrixResult:
        assert len(metric_results) == len(human_judgments)
        tp = fp = tn = fn = 0
        for m, h in zip(metric_results, human_judgments):
            metric_pass = m["score"] >= metric_threshold
            human_pass = h["correct"]
            if metric_pass and human_pass:
                tp += 1
            elif metric_pass and not human_pass:
                fp += 1   # 危险：评测说对但实际错
            elif not metric_pass and human_pass:
                fn += 1   # 浪费：评测说错但实际对
            else:
                tn += 1

        precision = tp / max(tp + fp, 1)
        recall = tp / max(tp + fn, 1)
        f1 = 2 * precision * recall / max(precision + recall, 1e-9)
        fp_rate = fp / max(fp + tn, 1)
        fn_rate = fn / max(fn + tp, 1)

        return ConfusionMatrixResult(
            metric_name=metric_results[0].get("metric_name", "unknown"),
            true_positive=tp,
            false_positive=fp,
            true_negative=tn,
            false_negative=fn,
            precision=round(precision, 3),
            recall=round(recall, 3),
            f1=round(f1, 3),
            fp_rate=round(fp_rate, 3),
            fn_rate=round(fn_rate, 3),
        )

    def severity_score(self, cm: ConfusionMatrixResult,
                        scenario: str = "general") -> str:
        """根据场景评估 fp 与 fn 的严重程度"""
        weights = {
            "general": (1.0, 1.0),
            "medical": (10.0, 1.0),     # 假阳性致命
            "customer_service": (2.0, 0.5),
            "education": (3.0, 1.0),
            "internal_qa": (0.5, 0.5),
        }
        fp_w, fn_w = weights.get(scenario, weights["general"])
        loss = fp_w * cm.fp_rate + fn_w * cm.fn_rate
        if loss < 0.05:
            return "excellent"
        if loss < 0.10:
            return "acceptable"
        if loss < 0.20:
            return "concerning"
        return "unsafe"

flowchart LR
  M[evaluator 输出] --> CM[confusion matrix]
  H[人工 ground-truth] --> CM

  CM --> TP[TP eval 对 + 实际对]
  CM --> FP[FP eval 对 + 实际错<br/>**最危险**]
  CM --> TN[TN eval 错 + 实际错]
  CM --> FN[FN eval 错 + 实际对<br/>无害浪费]

  FP --> SEV{场景权重}
  FN --> SEV
  SEV -->|医疗| HI[fp 权重 ×10]
  SEV -->|客服| MID[fp 权重 ×2]
  SEV -->|内部 QA| LO[等权重]

  HI --> S[severity 评分]
  MID --> S
  LO --> S

  style FP fill:#ffebee
  style HI fill:#ffebee

工程实务的 4 类场景的 fp/fn 权重：

具体例子：医疗 RAG 跑 1000 题，Faithfulness 与人工 anchor 对比：

precision=0.954, recall=0.941, f1=0.948——看似漂亮。

但 severity_score(scenario=“medical”)：fp_rate=14.9%, fn_rate=5.9% → loss = 10×0.149 + 1×0.059 = 1.55 → “unsafe” 状态！

诊断：35 个 FP 即”评测说 grounded 但实际错答”——医疗场景下每条都是潜在事故。修法：metric_threshold 从 0.70 提到 0.85，FP 降到 8 个，loss 回到 acceptable。

研究背景：

• Wallach et al. 2014 “Evaluating Test Suites of LLMs” 系统化了 LLM 评测的 fp/fn 分析框架
• BeIR 2021 paper 的 evaluation chapter 强调 “context retrieval 失误 = retrieval 假阳性”
• 医疗 ML 公约（FDA AI/ML SaMD Action Plan 2021）把 fp 权重列为合规审核要点

读者不要满足于”f1 = 0.95”——必须分场景看 fp/fn 单独的 rate。这是评测工程师”医生级别”的细致——一个数字背后藏着不同场景下的不同风险。

13.7.41 一份”RAG 索引版本兼容”评测——索引升级后旧评测还能比吗？

RAG 系统升级 embedding / 重新索引后，旧评测分数与新评测不直接可比——不同 embedding 的 retrieval 结果不同。下面给出”索引版本兼容评测”工程方案：

import hashlib
from dataclasses import dataclass
from typing import Iterable

@dataclass
class IndexCompatibilityReport:
    old_index_id: str
    new_index_id: str
    sample_count: int
    same_top1_pct: float          # top-1 chunk 相同的比例
    same_top5_overlap_avg: float  # top-5 集合 jaccard 平均
    score_correlation: float      # 旧分数 vs 新分数的相关性
    can_compare_directly: bool

class IndexCompatibilityChecker:
    """检查两个 RAG 索引版本之间的评测可比性"""

    def __init__(self, old_retriever, new_retriever):
        self.old = old_retriever
        self.new = new_retriever

    def _jaccard(self, a: set, b: set) -> float:
        return len(a & b) / max(len(a | b), 1)

    def assess(self, eval_set: list[dict],
                threshold_top1: float = 0.7,
                threshold_jaccard: float = 0.6,
                threshold_corr: float = 0.85) -> IndexCompatibilityReport:
        same_top1 = 0
        jaccards = []
        old_scores, new_scores = [], []

        for query in eval_set:
            old_top5 = self.old(query["question"], top_k=5)
            new_top5 = self.new(query["question"], top_k=5)

            old_ids = {c["id"] for c in old_top5}
            new_ids = {c["id"] for c in new_top5}

            if old_top5 and new_top5 and old_top5[0]["id"] == new_top5[0]["id"]:
                same_top1 += 1

            jaccards.append(self._jaccard(old_ids, new_ids))

            # 假设有 score 字段
            old_scores.append(query.get("old_score", 0))
            new_scores.append(query.get("new_score", 0))

        n = len(eval_set)
        same_top1_pct = same_top1 / max(n, 1)
        avg_jaccard = sum(jaccards) / max(n, 1)
        corr = self._correlation(old_scores, new_scores)

        can_compare = (
            same_top1_pct >= threshold_top1 and
            avg_jaccard >= threshold_jaccard and
            corr >= threshold_corr
        )

        return IndexCompatibilityReport(
            old_index_id=self._hash_index(self.old),
            new_index_id=self._hash_index(self.new),
            sample_count=n,
            same_top1_pct=round(same_top1_pct, 3),
            same_top5_overlap_avg=round(avg_jaccard, 3),
            score_correlation=round(corr, 3),
            can_compare_directly=can_compare,
        )

    def _hash_index(self, retriever) -> str:
        return hashlib.md5(str(retriever).encode()).hexdigest()[:12]

    def _correlation(self, x: list[float], y: list[float]) -> float:
        if len(x) < 2:
            return 0.0
        mean_x = sum(x) / len(x)
        mean_y = sum(y) / len(y)
        num = sum((xi - mean_x) * (yi - mean_y) for xi, yi in zip(x, y))
        den_x = (sum((xi - mean_x) ** 2 for xi in x)) ** 0.5
        den_y = (sum((yi - mean_y) ** 2 for yi in y)) ** 0.5
        return num / (den_x * den_y) if (den_x > 0 and den_y > 0) else 0.0

flowchart LR
  E[评测集] --> O[旧索引 retrieve]
  E --> N[新索引 retrieve]

  O --> T1[old top-5]
  N --> T2[new top-5]

  T1 --> CMP[同 top1?]
  T2 --> CMP
  T1 --> JC[jaccard top5]
  T2 --> JC

  CMP --> RP[CompatReport]
  JC --> RP
  RP --> Q{can_compare_directly?}
  Q -->|是| OK[直接比较新旧分数]
  Q -->|否| WARN[需对照运行 + 重建 baseline]

  style OK fill:#e8f5e9
  style WARN fill:#ffebee

工程实务的 4 条索引升级评测准则：

具体例子：从 BGE-large 升级到 OpenAI text-embedding-3：

判定：can_compare = false → 不能直接 say “新指标 +5pp 是进步”！

修法：

• 重建新索引的 baseline（跑 200 题人工 anchor 重新校准）
• 对照运行 4-8 周——同时跑新旧两套评测
• 对照期满后再切换，避免”看似进步实则换标尺”

3 类常见误判场景：

研究背景：

• Information Retrieval 经典书 Manning 2008 §8.4 讨论 “evaluation under different ranking”
• BeIR (Thakur et al. arXiv:2104.08663) 强调”benchmark 切换需要重新校准”
• HuggingFace 的 MTEB benchmark 明确要求 “新版 embedding 必须对照公开基准”

读者升级任何 RAG 组件前必跑 IndexCompatibilityChecker——避免用同一份”显然不可比”的数字给主管汇报”我们进步了”。这是 RAG 评测里”统计学正确性”的工程化保障。

13.7.42 RAG 评测的”业务领域权重”——把通用指标转成业务可消费

§13.7.40 给了 fp/fn 场景权重，本节给出更细粒度的”按业务领域加权”——同样 Faithfulness 0.85 在不同业务里意味不同：

from dataclasses import dataclass
from typing import Iterable

@dataclass
class BusinessWeightedScore:
    metric: str
    raw_score: float
    domain_weight: float
    weighted_score: float
    business_grade: str
    business_meaning: str

class RAGBusinessWeightCalculator:
    """把通用 RAG 指标转成业务可消费的"业务分""""

    DOMAIN_WEIGHT_MATRIX = {
        # (业务域, 指标) → weight + meaning
        ("medical", "faithfulness"): {
            "weight": 1.5,
            "meaning": "医疗回答必须 grounded - 编造可致死",
            "min_for_acceptable": 0.92,
        },
        ("medical", "context_recall"): {
            "weight": 1.4,
            "meaning": "漏检索关键医学文献 = 漏诊",
            "min_for_acceptable": 0.90,
        },
        ("legal", "citation_quality"): {
            "weight": 1.6,
            "meaning": "法律必须有引文 - 法庭呈交",
            "min_for_acceptable": 0.95,
        },
        ("customer_service", "answer_relevance"): {
            "weight": 1.3,
            "meaning": "客服回答跑题 = 用户骂",
            "min_for_acceptable": 0.85,
        },
        ("customer_service", "latency"): {
            "weight": 1.2,
            "meaning": "客服 > 3s 用户跳出",
            "min_for_acceptable": 2000,
        },
        ("internal_kb", "faithfulness"): {
            "weight": 0.8,
            "meaning": "内部 KB 错点不致命",
            "min_for_acceptable": 0.75,
        },
    }

    def calculate(self, domain: str, metric: str,
                   raw_score: float) -> BusinessWeightedScore:
        config = self.DOMAIN_WEIGHT_MATRIX.get(
            (domain, metric),
            {"weight": 1.0, "meaning": "通用", "min_for_acceptable": 0.85},
        )

        weighted = raw_score * config["weight"]
        min_acceptable = config["min_for_acceptable"]

        # 业务等级判定（针对越高越好的指标）
        if isinstance(min_acceptable, float):
            if raw_score >= min_acceptable * 1.05:
                grade = "excellent"
            elif raw_score >= min_acceptable:
                grade = "acceptable"
            elif raw_score >= min_acceptable * 0.9:
                grade = "warning"
            else:
                grade = "critical"
        else:   # latency 类
            if raw_score <= min_acceptable * 0.8:
                grade = "excellent"
            elif raw_score <= min_acceptable:
                grade = "acceptable"
            elif raw_score <= min_acceptable * 1.2:
                grade = "warning"
            else:
                grade = "critical"

        return BusinessWeightedScore(
            metric=metric,
            raw_score=raw_score,
            domain_weight=config["weight"],
            weighted_score=round(weighted, 3),
            business_grade=grade,
            business_meaning=config["meaning"],
        )

    def report_for_executive(self, domain: str,
                              metric_scores: dict) -> dict:
        """给主管的 1 页业务报告——不讲技术，只讲业务影响"""
        results = {m: self.calculate(domain, m, s)
                   for m, s in metric_scores.items()}
        critical = [m for m, r in results.items()
                    if r.business_grade == "critical"]
        return {
            "domain": domain,
            "overall": "GO" if not critical else "STOP",
            "critical_issues": critical,
            "business_summary": "; ".join(
                f"{m}: {r.business_grade} ({r.business_meaning})"
                for m, r in results.items()
            ),
        }

flowchart LR
  M[原始 metric: 0.85] --> W[业务权重 matrix]
  D[业务域: medical] --> W
  W --> WGT[加权 score]
  WGT --> Q{业务等级}
  Q -->|≥ acceptable × 1.05| EX[excellent]
  Q -->|≥ acceptable| OK[acceptable]
  Q -->|≥ acceptable × 0.9| WARN[warning]
  Q -->|< 0.9 acceptable| CRIT[critical]

  CRIT --> EXEC[exec report: STOP]
  EX --> EXEC
  OK --> EXEC

  style CRIT fill:#ffebee
  style EX fill:#e8f5e9

工程实务的 4 类典型业务权重模式：

具体例子：医疗 chatbot 跑指标：

exec 报告：“1 个 warning（faithfulness）→ 该 case 可上线但需密切监控，下次迭代必须修到 0.92+”

3 类常见业务权重错误：

研究背景：

• ML 评测的 “domain-specific scorecard” 是 healthcare AI 主流（如 Mayo Clinic 公开方法）
• DataDog APM 的 “service-tier-aware threshold” 是同思路
• Anthropic 的”per-deployment safety bar”概念

读者把 RAGBusinessWeightCalculator 接到 dashboard——主管看到的不是冷冰冰的 0.85，而是”warning，编造未越线但接近”。这是评测信号”业务化翻译”的工程化体现。

13.7.43 RAG 评测的”chunk attribution”溯源——为什么这个回答有这个 chunk

RAG 系统最难调试的失败模式之一：无法追溯”这条答案中的某个 fact 来自哪个 chunk”。下面给出 chunk attribution 评测：

import asyncio
from dataclasses import dataclass
from typing import Iterable, Callable, Awaitable

@dataclass
class ChunkAttribution:
    fact_in_answer: str
    likely_source_chunk_id: str | None
    similarity_score: float
    explanation_quality: str    # "perfect" / "weak" / "fabricated"

@dataclass
class AttributionAuditResult:
    answer_id: str
    total_facts: int
    attributable_facts: int
    fabricated_facts: int
    weak_attribution: int
    attribution_rate: float
    fabrication_rate: float

class RAGAttributionAnalyzer:
    """对每条 RAG 回答进行 chunk attribution 溯源"""

    PERFECT_THRESHOLD = 0.85
    WEAK_THRESHOLD = 0.50

    def __init__(self, embedder: Callable[[str], list[float]],
                 fact_extractor: Callable[[str], Awaitable[list[str]]]):
        self.embedder = embedder
        self.extract_facts = fact_extractor

    def _cosine_similarity(self, v1: list[float],
                            v2: list[float]) -> float:
        import math
        dot = sum(a * b for a, b in zip(v1, v2))
        norm1 = math.sqrt(sum(a * a for a in v1))
        norm2 = math.sqrt(sum(b * b for b in v2))
        return dot / (norm1 * norm2 + 1e-9)

    async def attribute_answer(self, answer: str,
                                retrieved_chunks: list[dict]) -> AttributionAuditResult:
        facts = await self.extract_facts(answer)
        chunks_with_emb = [
            {**c, "emb": self.embedder(c["text"])}
            for c in retrieved_chunks
        ]

        attribute_count = 0
        weak_count = 0
        fabricated_count = 0

        for fact in facts:
            fact_emb = self.embedder(fact)
            best_match_score = 0
            best_chunk = None
            for c in chunks_with_emb:
                sim = self._cosine_similarity(fact_emb, c["emb"])
                if sim > best_match_score:
                    best_match_score = sim
                    best_chunk = c

            if best_match_score >= self.PERFECT_THRESHOLD:
                attribute_count += 1
            elif best_match_score >= self.WEAK_THRESHOLD:
                weak_count += 1
            else:
                fabricated_count += 1

        n = len(facts)
        return AttributionAuditResult(
            answer_id="?",
            total_facts=n,
            attributable_facts=attribute_count,
            fabricated_facts=fabricated_count,
            weak_attribution=weak_count,
            attribution_rate=round(attribute_count / max(n, 1), 3),
            fabrication_rate=round(fabricated_count / max(n, 1), 3),
        )

flowchart LR
  A[RAG 回答] --> F[fact_extractor 抽事实]
  C[retrieved chunks] --> EM[embedding 索引]

  F --> M[逐 fact 找最佳 chunk]
  EM --> M

  M --> Q{similarity?}
  Q -->|"≥ 0.85"| AT[attributed]
  Q -->|"0.5-0.85"| WK[weak]
  Q -->|"< 0.5"| FB[fabricated]

  AT --> RPT[Attribution Report]
  WK --> RPT
  FB --> RPT
  RPT --> EXEC["每答案 5 fact: 4 attr + 1 weak<br/>+ 0 fab → 健康"]

  style FB fill:#ffebee
  style AT fill:#e8f5e9

工程实务的 4 条 attribution 阈值：

具体例子：医疗 RAG 100 条回答 audit：

诊断：22% facts 只是 weak attribution → retrieved chunks 与 answer 仅”相关”非”直接 grounded”。修法：reranker 调高 + 加 citation 强制。

3 类 attribution 工程价值：

研究背景：

• “Attribution-based Evaluation” (Bohnet et al. 2022) 是这套思路的学术起源
• Perplexity AI 把 attribution 作为产品核心 feature
• §13.7.36 citation 评测是 attribution 的”显式版本”——本节是隐式的

读者把 RAGAttributionAnalyzer 接入合规场景 RAG——任何 fabrication 都能精确定位到具体 chunk 错位。这是 RAG 评测里”知其然又知其所以然”的工程化能力。

13.7.44 RAG 评测的”问题分类”——按 query 类型分维度看分数

同样 Faithfulness=0.85 在不同 query 类型背后含义不同——事实查询失败 vs 总结性查询失败的代价完全不同。下面给出按 query 类型分维度评测：

from dataclasses import dataclass
from enum import Enum
from typing import Iterable, Awaitable, Callable
from collections import defaultdict

class QueryType(Enum):
    FACTUAL = "factual"           # "巴黎是哪国首都"
    SUMMARIZATION = "summarization"  # "总结这份政策"
    COMPARISON = "comparison"      # "X 和 Y 的差别"
    HOW_TO = "how_to"              # "如何申请退款"
    OPINION = "opinion"            # "你觉得 X 怎么样"
    REASONING = "reasoning"        # "如果...那么..."

@dataclass
class CategorizedRAGEvalResult:
    by_query_type: dict[str, dict[str, float]]   # type → {metric → score}
    weakest_type: str
    strongest_type: str
    overall_balance: float        # 类型间方差，越低越均衡

class CategorizedRAGEvaluator:
    """按 query 类型分维度跑 RAG 评测"""

    QUERY_KEYWORDS = {
        QueryType.FACTUAL: ["是什么", "在哪", "多少", "what is", "where"],
        QueryType.SUMMARIZATION: ["总结", "概括", "summarize", "TLDR"],
        QueryType.COMPARISON: ["比较", "差别", "vs", "compare"],
        QueryType.HOW_TO: ["如何", "怎么", "how to", "步骤"],
        QueryType.OPINION: ["你觉得", "建议", "推荐", "recommend"],
        QueryType.REASONING: ["为什么", "如果", "why", "if"],
    }

    def classify(self, query: str) -> QueryType:
        q = query.lower()
        for qtype, keywords in self.QUERY_KEYWORDS.items():
            if any(kw in q for kw in keywords):
                return qtype
        return QueryType.FACTUAL

    async def evaluate(self, eval_dataset: list[dict],
                        metric_runners: dict[str, Callable]
                        ) -> CategorizedRAGEvalResult:
        # group by query type
        by_type = defaultdict(list)
        for item in eval_dataset:
            qtype = self.classify(item["query"])
            by_type[qtype.value].append(item)

        # 每类型跑各 metric
        results = {}
        for qtype, items in by_type.items():
            type_scores = {}
            for metric_name, runner in metric_runners.items():
                scores = []
                for item in items:
                    s = await runner(item)
                    scores.append(s)
                type_scores[metric_name] = sum(scores) / max(len(scores), 1)
            results[qtype] = type_scores

        # 找最弱最强类型（用 Faithfulness 排）
        sorted_types = sorted(
            results.items(),
            key=lambda x: x[1].get("faithfulness", 0)
        )
        weakest = sorted_types[0][0] if sorted_types else "unknown"
        strongest = sorted_types[-1][0] if sorted_types else "unknown"

        # 类型间方差
        f_scores = [v.get("faithfulness", 0) for v in results.values()]
        if len(f_scores) > 1:
            mean_f = sum(f_scores) / len(f_scores)
            variance = sum((s - mean_f) ** 2 for s in f_scores) / len(f_scores)
            balance = max(0, 1 - variance ** 0.5)
        else:
            balance = 1.0

        return CategorizedRAGEvalResult(
            by_query_type=results,
            weakest_type=weakest,
            strongest_type=strongest,
            overall_balance=round(balance, 3),
        )

flowchart LR
  D[评测集] --> CL[query 分类]
  CL --> Q1[FACTUAL]
  CL --> Q2[SUMMARIZATION]
  CL --> Q3[COMPARISON]
  CL --> Q4[HOW_TO]
  CL --> Q5[OPINION]
  CL --> Q6[REASONING]

  Q1 --> M[per-type 跑 metric]
  Q2 --> M
  Q3 --> M
  Q4 --> M
  Q5 --> M
  Q6 --> M

  M --> R[6 类 × 4 metric 矩阵]
  R --> WK[找 weakest 类型]
  R --> ST[找 strongest 类型]
  R --> B[balance score]

  style WK fill:#ffebee
  style ST fill:#e8f5e9

工程实务的 4 类典型 query 分布：