跳转至

第17讲：RAG 回归验收与入库质量

上一讲：文档入库与索引链路 下一讲：测试与接口验收

企业路线

本讲以 LangSmith Evaluation 作为企业主线。项目本地只保留 source 推断准确率、场景隔离率、FAQ 直出准确率、Prompt Profile 命中率、表格行召回等领域指标；Trace、Annotation、Dataset 和 Evaluation 统一交给 LangSmith。

本讲边界

第 17 讲回答“知识和答案质量如何评估”。它关注入库质量、检索效果、性能基线和领域指标。第 18 讲会回答“代码和接口如何验收”，第 19 讲会回答“上线后如何观测、压测和扩容”。

本讲目标

  • 理解 RAG 系统的RAG 回归与入库质量全景
  • 掌握入库质量、检索评测、性能基线的三层保障体系
  • 理解验收（Gate）机制的设计思路
  • 理解 Bad Case 如何从 LangSmith trace/annotation 沉淀为回归样本
  • 能手算 Recall@K、MRR、关键词覆盖率

本讲地图

本图对应本讲功能闭环，展示从输入到本讲交付物的主干路径。节点与主项目代码文件和函数保持一致，后续章节消费的能力只作为交付边界出现。

图 1：第 17 讲功能闭环地图

flowchart TD
    C17_REPORT["质量报告<br/>build_ingestion_quality_report()"]
    C17_FAQ_READ["FAQ 读取<br/>read_faq_records()"]
    C17_FAQ_ANALYZE["FAQ 检查<br/>analyze_faq_csv()"]
    C17_CONFLICT["冲突检测<br/>detect_faq_document_conflicts()"]
    C17_META["资料类型判断<br/>is_table_metadata()"]
    C17_CHUNK["Chunk 质量<br/>analyze_chunk_quality()"]
    C17_DEMO["命令行验证<br/>main()"]
    C17_TEST{{"回归测试<br/>QualityReportChapter17Test"}}
    C17_REPORT --> C17_FAQ_READ
    C17_FAQ_READ --> C17_FAQ_ANALYZE
    C17_FAQ_ANALYZE --> C17_CONFLICT
    C17_META --> C17_CONFLICT
    C17_REPORT --> C17_CHUNK
    C17_CONFLICT --> C17_DEMO
    C17_CHUNK --> C17_DEMO
    C17_DEMO --> C17_TEST
    style C17_REPORT fill:#F8FAFC,stroke:#64748B,stroke-width:2px
    style C17_FAQ_READ fill:#DBEAFE,stroke:#2563EB,stroke-width:2px
    style C17_FAQ_ANALYZE fill:#F5F3FF,stroke:#7C3AED,stroke-width:2px
    style C17_CONFLICT fill:#FEF3C7,stroke:#D97706,stroke-width:2px
    style C17_META fill:#DBEAFE,stroke:#2563EB,stroke-width:2px
    style C17_CHUNK fill:#F5F3FF,stroke:#7C3AED,stroke-width:2px
    style C17_DEMO fill:#FEF3C7,stroke:#D97706,stroke-width:2px
    style C17_TEST fill:#DCFCE7,stroke:#16A34A,stroke-width:2px

节点与代码对齐

节点 对齐文件 函数/对象 本章职责
质量报告 qa_core/quality/ingestion.py build_ingestion_quality_report() 扫描候选文件并汇总入库质量问题。
FAQ 读取 qa_core/quality/faq.py read_faq_records() 读取 FAQ CSV 为检查记录。
FAQ 检查 qa_core/quality/faq.py analyze_faq_csv() 检查必填项、重复问题和非法 source。
冲突检测 qa_core/quality/conflicts.py detect_faq_document_conflicts() 识别 FAQ 与文档中的数字、极性和关键词冲突。
资料类型判断 qa_core/document_metadata.py is_table_metadata() 区分普通 chunk 与表格行，避免误报。
Chunk 质量 qa_core/quality/chunk.py analyze_chunk_quality() 检查 chunk 长度、空白和 metadata。
命令行验证 scripts/demo_quality_report.py main() 输出一份可阅读的质量报告。
回归测试 tests/test_quality_report.py QualityReportChapter17Test 锁住质量报告结构和关键告警。

第一部分：前置知识 — 为什么 RAG 需要系统化评测

1.1 RAG 评测的挑战

传统软件测试通常是二元的（通过/失败）。但 RAG 系统的输出是自然语言文本，不能简单地用 assertEqual(expected, actual) 来判断。

1
2
3
4
5
6
7
8
9
问题："入职流程有哪些步骤"

预期行为：
  ✅ 召回了正确的文档片段（检索质量）
  ✅ 答案包含了流程的完整步骤（完整性）
  ✅ 答案基于提供的资料而非幻觉（忠实性）
  ✅ 来源引用正确（可溯源性）

❌ 这些指标不能用一个简单的 test case 覆盖

1.2 三层保障体系

flowchart TD
    subgraph L1["第一层：入库质量"]
        L1A["文件解析成功率"]
        L1B["低质量 chunk 比例"]
        L1C["FAQ 空值/重复率"]
        L1D["FAQ/正文冲突检测"]
    end

    subgraph L2["第二层：检索评测"]
        L2A["Recall@K 召回率"]
        L2B["MRR 平均倒数排名"]
        L2C["关键词覆盖率"]
        L2D["场景隔离准确率"]
    end

    subgraph L3["第三层：性能基线"]
        L3A["首 token 耗时"]
        L3B["总耗时 P50/P95"]
        L3C["各阶段耗时分布"]
    end

    L1 --> L2 --> L3

    subgraph Gates["回归验收体系"]
        G1["入库质量检查"]
        G2["RAG 回归验收（分组）"]
        G3["追问回归验收"]
        G4["性能回归验收"]
        G5["接口验收"]
    end

    L1 --> G1
    L2 --> G2
    L2 --> G3
    L3 --> G4
    G1 --> G5
    G2 --> G5
    G3 --> G5
    G4 --> G5

    style L1 fill:#EFF6FF,stroke:#3B82F6,stroke-width:2px
    style L2 fill:#ECFDF5,stroke:#059669,stroke-width:2px
    style L3 fill:#FFFBEB,stroke:#D97706,stroke-width:2px
    style Gates fill:#FEF2F2,stroke:#DC2626,stroke-width:2px

第二部分：入库质量报告

2.1 检查项

1
2
python scripts/check_ingestion_quality_gate.py \
    --scenario enterprise_knowledge

生成报告覆盖以下维度：

文件解析检查： - 哪些文件解析失败（PDF 损坏、编码错误） - 哪些文件类型不被支持 - 哪些文件为空（没有任何有效文本）

Chunk 质量检查： - 低质量 chunk：字符数过少（<50 字符）或噪声占比过高 - 重复 chunk：内容高度相似的 chunk 对

FAQ 质量检查： - question 或 answer 为空的记录 - 完全相同的 FAQ 对（重复录入） - source 不在 valid_sources 白名单中的 FAQ

2.2 FAQ/正文冲突检测

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
# qa_core/quality/conflicts.py

def detect_faq_document_conflicts(faq_documents, doc_chunks):
    """检测 FAQ 标准答案与文档正文的矛盾。"""

    import jieba

    conflicts = []
    for faq_doc in faq_documents:
        faq_question = faq_doc.page_content
        faq_answer = faq_doc.metadata.get("answer", "")

        # 用 jieba 搜索分词提取 FAQ 问题中的关键概念
        faq_keywords = set(jieba.cut_for_search(faq_question))

        for chunk in doc_chunks:
            chunk_text = chunk.page_content
            chunk_keywords = set(jieba.cut_for_search(chunk_text))

            # 如果 FAQ 和文档都讨论了同一个概念
            common_keywords = faq_keywords & chunk_keywords
            if len(common_keywords) >= 3:
                # 比较判断是否有矛盾信息
                # （简化逻辑，实际会做更复杂的语义比较）
                similarity = compute_conflict_score(faq_answer, chunk_text)
                if similarity > 0.8:  # 疑似矛盾
                    conflicts.append({
                        "faq_question": faq_question,
                        "faq_answer": faq_answer,
                        "conflicting_chunk": chunk_text[:200],
                        "common_keywords": list(common_keywords),
                    })

    return conflicts

为什么用 jieba.cut_for_search 而不是简单正则：

cut_for_search 是 jieba 的搜索模式分词，会同时输出原词和更细粒度的子词。例如"管理员密码重置"会被分为 ["管理员", "管理", "密码", "重置"]，这样"用户密码修改"也能匹配到"密码"这个公共关键词。

2.3 入库质量检查

1
2
python scripts/check_ingestion_quality_gate.py \
    --report reports/ingestion/enterprise_knowledge_phase1_gate_check.json

这里要区分两个概念：

概念 职责 对应代码
入库质量报告 记录本次候选版本有哪些质量事实，例如失败文件、空文件、重复 FAQ、低质量 chunk build_ingestion_quality_report()
入库质量门禁 根据阈值判断候选版本能不能继续激活 evaluate_report_against_gate()

真实代码默认采用严格门禁，下面这些问题默认都要求为 0：

条件 阈值
文件解析失败 max_failed_files = 0
未支持文件或不在 source 白名单的文件 max_unsupported_files = 0
空文件 max_empty_files = 0
低质量 chunk max_low_quality_issues = 0
重复 chunk max_duplicate_chunks = 0
FAQ question 为空 max_empty_faq_questions = 0
FAQ answer 为空 max_empty_faq_answers = 0
FAQ 问题重复 max_duplicate_faq_questions = 0
FAQ source 非法 max_invalid_faq_sources = 0
FAQ/正文潜在冲突 max_faq_document_conflicts = 0

验收不通过时，不激活新版本。这样可以确保线上知识库始终是经过质量验证的。真实企业项目可以在资料治理早期临时放宽某个阈值，但必须通过命令行显式传入，例如 --max-duplicate-chunks 3；不要在代码里悄悄吞掉质量问题。

这部分和第 14 章的关系是：第 14 章负责说明“为什么门禁失败不能切换 active 指针”，第 17 章负责说明“门禁根据哪些质量事实做判断”。

第三部分：检索评测

3.1 评测数据集格式

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
// eval_sets/multi_scenario_smoke.json
[
    {
        "scenario_id": "enterprise_knowledge",
        "query": "入职流程有哪些步骤",
        "expected_source": "hr",
        "expected_hit_type": "rag",
        "expected_keywords": ["入职", "流程", "步骤", "材料", "合同"],
        "min_expected_sources": 2
    },
    {
        "scenario_id": "enterprise_knowledge",
        "query": "忘记密码怎么办",
        "expected_source": "it",
        "expected_hit_type": "faq_direct",
        "expected_keywords": ["密码", "重置", "邮箱", "手机"],
        "min_expected_sources": 1
    }
]

3.2 评测指标

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
# 以下 RAGEvaluationMetrics 为简化伪代码，实际评测逻辑分布在
# scripts/evaluate_core_chain.py 和 scripts/eval_common.py 中，不存在该独立类

class RAGEvaluationMetrics:
    def compute(self, test_cases, actual_results):
        metrics = {}

        # Recall@K：期望的关键词在召回的文档中出现了多少
        metrics["recall_at_k"] = sum(
            self._keyword_recall(tc, result)
            for tc, result in zip(test_cases, actual_results)
        ) / len(test_cases)

        # MRR：正确答案在召回列表中的排名的倒数平均值
        metrics["mrr"] = sum(
            1.0 / self._first_relevant_rank(tc, result)
            for tc, result in zip(test_cases, actual_results)
        ) / len(test_cases)

        # 关键词覆盖率
        metrics["avg_keyword_coverage"] = sum(
            self._keyword_coverage(tc, result)
            for tc, result in zip(test_cases, actual_results)
        ) / len(test_cases)

        # hit_type 准确率
        metrics["hit_type_accuracy"] = sum(
            1.0 if tc["expected_hit_type"] == result.get("hit_type")
            else 0.0
            for tc, result in zip(test_cases, actual_results)
        ) / len(test_cases)

        # source 推断准确率
        metrics["source_inference_accuracy"] = sum(
            1.0 if tc["expected_source"] == result.get("source_filter")
            else 0.0
            for tc, result in zip(test_cases, actual_results)
        ) / len(test_cases)

        # 场景隔离准确率
        metrics["scenario_isolation_accuracy"] = ...

        metrics["errors"] = sum(
            1 for r in actual_results if r.get("error")
        )

        return metrics

3.3 分组验收

关键设计：回归验收不只是看全局平均值，而是按场景、source、hit_type 分组检查。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# scripts/check_evaluation_gate.py

def check_evaluation_gate(report):
    """按维度分组检查 RAG 回归验收。"""
    failures = []

    # 全局验收
    global_metrics = report["metrics"]
    if global_metrics["recall_at_k"] < 0.85:
        failures.append(f"全局 Recall@K {global_metrics['recall_at_k']} < 0.85")

    # 按场景分组验收 ← 防止某个场景退化被全局均值掩盖
    for scenario_id, scenario_metrics in report["by_scenario"].items():
        if scenario_metrics["recall_at_k"] < 0.80:
            failures.append(
                f"场景 {scenario_id} Recall@K {scenario_metrics['recall_at_k']} < 0.80"
            )

    # 按 source 分组验收
    for source, source_metrics in report["by_source"].items():
        if source_metrics["recall_at_k"] < 0.75:
            failures.append(f"分类 {source} 召回率不达标")

    return len(failures) == 0, failures

3.4 RAGAS 补充评测

本项目的主评测不是 RAGAS，而是面向企业 RAG 主链路的工程回归门禁。原因是企业项目不能只判断答案文本是否“看起来合理”，还必须验证：

  • 是否召回到预期来源：Recall@K
  • 预期来源排名是否靠前：MRR
  • 答案是否覆盖关键事实：keyword_coverage
  • FAQ 直出、RAG 生成、边界提示等路径是否正确：hit_type_accuracy
  • source 自动推断是否正确：source_inference_accuracy
  • Prompt Profile 路由是否正确：prompt_profile_accuracy
  • 多场景隔离是否正确：scenario_isolation_accuracy
  • 是否出现错误和明显超时

为什么不直接把 RAGAS 作为主评测

RAGAS 是很好的 RAG 语义质量评估工具，但它默认关注的是“问题、答案、上下文、参考答案”之间的语义关系。KnowForge 的主评测目标更偏企业工程回归，很多关键指标不是 RAGAS 默认能直接判断的。

企业级验收问题 RAGAS 默认是否能直接判断 本项目主评测如何判断
是否召回到预期业务来源 部分能，需要额外改造样本和上下文标注 expected_source_contains + Recall@K + MRR
FAQ 是否应该高置信直出 不能直接判断 expected_hit_type=faq_direct + faq_direct_accuracy
问题是否应该进入 RAG 生成 不能直接判断 hit_type_accuracy
是否识别到 source 选错边界 不能直接判断 source_boundary / source_inference_accuracy
是否命中正确 Prompt Profile 不能直接判断 expected_prompt_profile + prompt_profile_accuracy
是否遵守多场景隔离 不能直接判断 scenario_isolation_accuracy
是否只查当前 active kb_version 不能直接判断 工程评测报告记录 kb_version 和检索诊断
是否遵守 DataScope 权限过滤 不能直接判断 评测样本传入 tenant_id / dataset_id / visibility / user_role
是否出现接口错误或依赖异常 不能作为主指标 errors / error_rate
响应耗时是否可接受 不能作为主指标 avg_elapsed_ms / 性能基线门禁

因此，如果把 RAGAS 直接作为主评测，会出现三个问题：

  1. 会漏掉企业级链路指标：FAQ 直出、source 边界、Prompt Profile、多场景隔离、DataScope、active 版本等都不是 RAGAS 的默认评价对象。
  2. 会把工程问题误看成语义问题：例如召回 source 错了、版本过滤错了、权限过滤错了，RAGAS 可能只看到“答案和上下文是否一致”，但无法指出是哪条工程链路退化。
  3. 不适合做唯一 CI 门禁：RAGAS 依赖 LLM-as-judge，成本更高、速度更慢、结果有一定波动；本项目需要一个稳定、可解释、可失败的工程回归门禁。

所以本项目的定位是：

1
2
3
4
5
6
7
主评测：自研工程回归门禁
  负责判断主链路有没有退化：
  Recall@K / MRR / hit_type / source 推断 / Prompt Profile / 场景隔离 / DataScope / 错误率 / 耗时

补充评测：RAGAS
  负责分析答案语义质量：
  faithfulness / answer_relevancy / context_relevance / groundedness

一句话总结：RAGAS 适合回答“答案语义质量怎么样”，本项目自研门禁负责回答“企业 RAG 主链路是否稳定正确”。两者互补，不是替代关系。

RAGAS 适合补充回答语义质量，例如：

  • faithfulness：答案是否忠实于召回上下文
  • answer_relevancy：答案是否回应了用户问题
  • context_relevance：召回上下文是否和问题相关
  • response_groundedness：答案是否能被上下文支撑

所以本项目采用两层评测：

层级 工具 作用 是否主门禁
工程回归门禁 evaluate_core_chain.py + check_evaluation_gate.py 验证召回、路由、场景隔离、Prompt、错误率和耗时 是
追问专项门禁 evaluate_followup_chain.py + check_followup_gate.py 验证多轮追问、改写和历史上下文 是
RAGAS 补充分析 evaluate_ragas_quality.py 评估忠实度、答案相关性等语义质量 否

典型执行顺序：

1
2
3
python scripts/evaluate_core_chain.py --dataset eval_sets/multi_scenario_smoke.json --limit 20 --output reports/evaluation/core_chain_latest.json
python scripts/check_evaluation_gate.py --report reports/evaluation/core_chain_latest.json
python scripts/evaluate_ragas_quality.py --report reports/evaluation/core_chain_latest.json --limit 10 --output reports/evaluation/core_chain_latest_ragas.json

evaluate_ragas_quality.py 会读取工程评测报告中的完整答案和检索上下文，生成独立的 RAGAS 报告。它不会替代 check_evaluation_gate.py，因为 RAGAS 无法直接判断本项目最关键的企业级指标，例如 active kb_version 是否正确、DataScope 是否隔离、source boundary 是否识别、FAQ 是否高置信直出。

第三部分补充：评测指标手算示例

上面的代码展示了指标的计算公式。为了能真正理解 MRR，需要用具体检索排序样例解释它的含义。以下用本项目的真实评测数据说明。

3.4 Recall@K 手算示例

Recall@K 衡量的是：在召回的 Top-K 个文档中，有多少期望的关键词被覆盖了。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
以测试样本为例：
  查询："入职流程有哪些步骤"
  期望关键词：["入职", "流程", "步骤", "材料", "合同"]

召回结果（Top-5 文档片段）：
  [1] "入职流程包括以下步骤：1. 提交个人材料..." → 命中：入职, 流程, 步骤, 材料 ✅
  [2] "新员工入职当天需要携带身份证、学历证书..." → 命中：入职 ✅
  [3] "劳动合同应在入职后一个月内签订..." → 命中：合同 ✅
  [4] "培训安排将在入职第二周进行..." → 命中：入职 ✅
  [5] "员工福利包括五险一金、带薪年假..." → 命中：无 ❌

已覆盖的关键词：{"入职", "流程", "步骤", "材料", "合同"} → 5/5 = 1.0

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
def recall_at_k(expected_keywords, retrieved_docs, k=5):
    """计算单条测试样本的 Recall@K。"""
    # 取前 K 个文档
    top_k_docs = retrieved_docs[:k]

    # 合并所有召回文档的文本
    combined_text = " ".join(doc.page_content for doc in top_k_docs)

    # 统计被覆盖的关键词
    covered = {kw for kw in expected_keywords if kw in combined_text}

    return len(covered) / len(expected_keywords)

# 手算验证
expected = ["入职", "流程", "步骤", "材料", "合同"]
recalled_docs = [...]  # 上面 5 个文档
print(recall_at_k(expected, recalled_docs, k=5))  # 5/5 = 1.0

# 如果 k=2（只看前 2 个文档）
print(recall_at_k(expected, recalled_docs, k=2))  # 4/5 = 0.8
# 因为"合同"在第 3 个文档才出现

K 值的选择：

K 值 含义 本项目使用
K=1 只看第 1 个召回结果 太严格
K=3 看前 3 个 中等
K=5 看前 5 个 本项目使用
K=10 看前 10 个 较宽松

3.5 MRR 手算示例

MRR（Mean Reciprocal Rank） 衡量的是：第一个真正相关的文档排在召回列表的第几位。

MRR = (1/排名₁ + 1/排名₂ + ... + 1/排名n) / n

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
假设有 3 个测试查询：

查询 1："入职流程有哪些步骤"
  召回结果：[doc_A(0.92), doc_B(0.85), doc_C(0.78), ...]
  第一个相关文档是 doc_A，排名第 1 位
  → Reciprocal Rank = 1/1 = 1.0

查询 2："VPN 连不上怎么办"
  召回结果：[doc_X(0.78), doc_Y(0.75), doc_Z(0.71), ...]
  前两个都不相关（虽然分数高，但内容不匹配）
  第一个相关文档是 doc_Z，排名第 3 位
  → Reciprocal Rank = 1/3 ≈ 0.333

查询 3："员工报销需要准备哪些材料"
  召回结果：[doc_M(0.95), doc_N(0.82), ...]
  第一个相关文档是 doc_M，排名第 1 位
  → Reciprocal Rank = 1/1 = 1.0

MRR = (1.0 + 0.333 + 1.0) / 3 ≈ 0.778

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
def mean_reciprocal_rank(test_cases, search_fn):
    """计算 MRR。"""
    reciprocal_ranks = []

    for tc in test_cases:
        results = search_fn(tc["query"])  # 执行检索
        relevant_doc_id = tc["relevant_doc_id"]

        # 找第一个相关文档的排名
        rank = None
        for i, result in enumerate(results, start=1):
            if result.id == relevant_doc_id:
                rank = i
                break

        if rank is not None:
            reciprocal_ranks.append(1.0 / rank)
        else:
            reciprocal_ranks.append(0.0)  # 没找到 = 0

    return sum(reciprocal_ranks) / len(reciprocal_ranks)

# 手算验证
print(f"MRR = {mean_reciprocal_rank(test_cases, search_fn):.3f}")
# MRR = 0.778

MRR 的直观理解：

1
2
3
4
5
MRR = 1.0  → 每个查询的第一个结果是相关的           → 完美
MRR = 0.9  → 第一个相关结果平均排在第 1.1 位          → 本项目水平
MRR = 0.5  → 第一个相关结果平均排在第 2 位            → 合格
MRR = 0.2  → 第一个相关结果平均排在第 5 位            → 需要改进
MRR = 0.05 → 几乎找不到相关结果                        → 严重问题

3.6 关键词覆盖率手算示例

关键词覆盖率 衡量的是：期望关键词中有多少出现在了召回的文档片段里。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
查询："跨境贸易中 HS 编码归类争议怎么处理"
期望关键词：["HS编码", "归类", "海关", "争议", "行政复议", "预裁定"]

召回文档片段（合并后）：
  "HS 编码归类争议可通过以下途径解决：1. 向海关申请预裁定
   2. 如对归类决定有异议可申请行政复议 3. 必要时走行政诉讼流程"

逐个检查关键词是否出现在文本中：
  "HS编码"     → 出现了 "HS 编码"   → ✅ 覆盖
  "归类"       → 出现了 "归类"       → ✅ 覆盖
  "海关"       → 出现了 "海关"       → ✅ 覆盖
  "争议"       → 出现了 "争议"       → ✅ 覆盖
  "行政复议"   → 出现了 "行政复议"   → ✅ 覆盖
  "预裁定"     → 出现了 "预裁定"     → ✅ 覆盖

关键词覆盖率 = 6/6 = 1.0

1
2
3
4
5
def keyword_coverage(expected_keywords, retrieved_docs):
    """计算关键词覆盖率。"""
    combined_text = " ".join(doc.page_content for doc in retrieved_docs)
    covered = sum(1 for kw in expected_keywords if kw in combined_text)
    return covered / len(expected_keywords)

3.7 一个完整评测样本长什么样

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
{
    "scenario_id": "engineering_project_qa",
    "query": "隐蔽工程验收需要哪些资料",
    "expected_source": "quality",
    "expected_hit_type": "rag",
    "expected_keywords": [
        "隐蔽工程",
        "验收",
        "质量验收报告",
        "隐蔽工程验收记录",
        "材料检测报告",
        "功能性试验报告"
    ],
    "expected_prompt_profile": "knowledge_answer",
    "min_expected_sources": 3,
    "relevant_doc_id": "engineering_project_qa/doc_chunk_quality_042",
    "notes": "期望从 quality 分类召回，覆盖至少 3 个关键词，使用 knowledge_answer 模板"
}

一个好的评测样本需要： 1. expected_source：验证 source 推断是否正确 2. expected_keywords：至少 4-6 个具体关键词，不是模糊描述 3. expected_hit_type：验证 FAQ 直出 vs 文档 RAG 的判断是否正确 4. min_expected_sources：验证是否跨 source 串库 5. relevant_doc_id（可选）：用于精确计算 MRR 6. notes：解释为什么期望这些值，帮助其他人理解评测意图

第四部分：Bad Case 沉淀

4.1 先看一个具体 Bad Case

与第 19 讲的边界

本部分是 Bad Case 质量闭环的主位置：重点讲“问题如何被标注、沉淀为 Dataset、进入 Evaluation，并最终影响发布 Gate”。第 19 讲只讲线上 Trace 如何发现问题、定位问题和把问题交接到这里，不再重复完整评测闭环。

Bad Case 不是一句“答案不对”，而是一条能复现、能标注、能再次评测的问题样本。先看一个具体例子。

用户提问：

1
VPN 客户端版本、账号锁定、公网 IP 这些排查项分别应该怎么处理？

系统回答：

1
可以先重启 VPN 客户端，确认网络正常；如果仍然无法连接，请提交 IT 工单。

这个回答看起来不算错，但它没有分别回答三个排查项：

用户问到的点 期望回答 当前回答是否覆盖
VPN 客户端版本 确认是否为 IT 发布的最新版，旧版本需重新安装 否
账号锁定 检查账号是否过期、锁定或权限被回收 否
公网 IP 判断当前公网 IP 是否在允许范围或是否被安全策略拦截 否

所以这个问题应该被沉淀为 Bad Case。它的价值不是“记录一次失败”，而是保证后续改 Prompt、改检索策略、重建知识库后，这个问题必须被重新验证。

4.2 这条 Bad Case 在 Trace 里怎么看

LangSmith Trace 中会看到类似信息：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
{
  "scenario_id": "enterprise_knowledge",
  "kb_version": "kb_enterprise_knowledge_20260620_082630_4c1df17a",
  "intent": "KNOWLEDGE_QUERY",
  "question_category": "troubleshooting",
  "prompt_profile": "troubleshooting_steps",
  "hit_type": "rag",
  "sources_count": 2,
  "top_source_score": 0.63,
  "slowest_stage": "llm_generation",
  "stage_timings_ms": {
    "doc_retrieval": 840,
    "rerank": 390,
    "llm_generation": 5200
  }
}

这条 Trace 给出的判断不是简单“模型答错了”，而是：

  • hit_type=rag：它确实进入了 RAG 生成，不是 FAQ 直出问题。
  • prompt_profile=troubleshooting_steps：Prompt 档位基本正确。
  • sources_count=2：召回到了资料，但证据可能不完整。
  • top_source_score=0.63：召回置信度一般，需要检查是否缺少更细的排障资料。
  • 回答漏掉三个子问题：更像“上下文覆盖不足或生成未按子问题展开”。

因此它应进入质量闭环，而不是只作为一次线上投诉处理。

4.3 企业中如何处理 Bad Case

企业路线下，优先在 LangSmith 中完成 trace 查看、人工标注和 dataset 沉淀：

flowchart LR
    T["📊 LangSmith Trace<br/>线上问答和业务元数据"] --> A["🔎 过滤器<br/>错误/无来源/低分/<br/>超时/信息不足"]

    A --> H["🏷️ Annotation<br/>填写 expected hit_type/source/keywords"]

    H --> D["📦 Dataset<br/>沉淀真实线上回归样本"]

    D --> E["🧪 Evaluation<br/>运行领域评测指标"]

    E --> EV["🧪 下次评测<br/>之前失败的问题必须通过"]

    EV -.->|"新的 Bad Case"| T

    style T fill:#EFF6FF,stroke:#2563EB,stroke-width:2px
    style A fill:#FFFBEB,stroke:#D97706,stroke-width:2px
    style H fill:#ECFDF5,stroke:#059669,stroke-width:2px
    style EV fill:#FEF2F2,stroke:#DC2626,stroke-width:2px

4.4 自动识别

Bad Case 的入口通常不是人工逐条翻聊天记录，而是先用 Trace metadata 过滤出“疑似异常样本”：

过滤条件 说明 适合优先复核的问题
error=true 运行时异常 代码、依赖、模型服务问题
hit_type=insufficient_context 系统认为资料不足 需要判断是资料真缺失，还是检索没召回
sources_count=0 没有任何引用来源 重点排查 active 版本、DataScope、source_filter
top_source_score < 0.65 召回置信度偏低 重点排查查询改写、切分、embedding、BM25
prompt_profile != expected 模板档位不符合问题类型 重点排查问题分类和 Prompt Profile 路由
first_token_ms 或 elapsed_ms 超阈值 性能异常 进入性能基线排查，不一定是质量 Bad Case

对于上面的 VPN 示例，可以用下面的过滤条件批量找类似样本：

1
2
3
4
scenario_id = enterprise_knowledge
question_category = troubleshooting
hit_type = rag
top_source_score < 0.7

4.5 人工复核怎么填

人工复核不是写一句“回答不完整”，而是把业务期望结构化。以 VPN 示例为例，Annotation 可以这样填：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
{
  "is_correct": false,
  "issue_type": "missing_sub_questions",
  "priority": "high",
  "expected_hit_type": "rag",
  "expected_source": "it",
  "expected_prompt_profile": "troubleshooting_steps",
  "expected_keywords": [
    "客户端版本",
    "账号锁定",
    "公网 IP",
    "IT 工单",
    "截图"
  ],
  "grading_notes": "答案必须分别说明客户端版本、账号锁定、公网 IP 三个排查项，不能只给泛泛重启建议。"
}

这些字段会直接影响后续评测：

标注字段 后续如何使用
expected_hit_type 判断是否走了正确路径，避免应该 RAG 的问题被误判为信息不足
expected_source 判断是否召回 IT 分类资料，避免跨 source 串库
expected_prompt_profile 判断是否使用排障步骤模板
expected_keywords 判断答案是否覆盖关键事实
grading_notes 给人工复核和 LLM-as-judge 提供判分依据

4.6 提升为评测样本

复核完成后，将 Trace 加入 LangSmith Dataset，形成真实线上问题回归集。Dataset 不建议混成一个大池子，而是按场景或问题类型拆分：

Dataset 放什么样本 示例
enterprise_it_troubleshooting_cases IT 排障类问题 VPN、账号锁定、工单、权限回收
enterprise_finance_reimbursement_cases 财务报销类问题 发票、预算、审批、付款材料
enterprise_hr_onboarding_cases HR 入职类问题 入职材料、合同、异地办理
multi_turn_followup_cases 多轮追问 “那审批呢”“材料呢”“谁负责”

加入 Dataset 后，这条样本就不再只是一次线上记录，而是变成以后每次版本变更都要验证的“质量资产”。

4.7 LangSmith Evaluation 如何落到本项目

LangSmith Evaluation 不能只理解成“平台会自动评估”，还要结合网页端页面一起看。网页端负责让 Trace、标注、Dataset、Experiment 变得可见；本项目字段负责让这些页面能解释企业 RAG 主链路为什么正确或退化。

官方 Evaluation 主流程是：创建 Dataset、定义 Evaluator、运行 Experiment、分析结果；线上闭环还会把失败的生产 Trace 加入 Dataset，再用离线 Experiment 验证修复效果。对应到本项目，就是把 scenario_id、kb_version、hit_type、source_filter、prompt_profile、召回分数和耗时写入 Trace metadata，然后在 LangSmith 页面上完成复核和沉淀。

参考页面：

4.7.1 网页端先看哪些页面

LangSmith 页面 主要看什么 对应本项目字段 判断问题
Traces / Runs 单次问答链路、输入输出、子步骤、metadata、耗时 scenario_id、kb_version、session_id、hit_type、elapsed_ms 这次请求到底走了哪条链路
Run detail 检索、重排、Prompt、LLM 生成等子步骤 sources_count、top_source_score、prompt_profile、stage_timings_ms 是检索问题、Prompt 问题，还是生成问题
Feedback / Annotation 人工复核结论和问题类型 expected_hit_type、expected_source、expected_keywords 这条 Trace 应该如何被判分
Annotation Queues 批量分发待复核样本 过滤条件、队列名称、rubric 多条疑似 Bad Case 如何集中复核
Datasets 固化后的回归样本 query、history、expected fields、metadata 这条问题以后是否每次都要重测
Experiments 一批样本的评测结果 evaluator scores、outputs、run traces 某次版本改动是否整体变好
Experiment Compare 两个版本横向对比 prompt version、model version、retrieval params 新版本是否出现回归
本地 Gate 报告 是否允许发布 check_evaluation_gate.py 输出 是否给出确定性发布结论

这一节的重点不是记住 LangSmith 的某个按钮位置，而是记住“页面在看什么”。UI 入口可能调整，但 Trace、Feedback、Dataset、Experiment 这几个对象不会变。

4.7.2 本项目字段如何映射到页面

flowchart TD
    APP["KnowForge 问答请求<br/>query / history / source_filter"] --> TRACE["LangSmith Trace / Run<br/>完整链路和 metadata"]
    TRACE --> DETAIL["Run detail<br/>检索 / Prompt / LLM / 耗时"]
    DETAIL --> ANNO["Feedback / Annotation<br/>人工标注 expected_*"]
    ANNO --> DATASET["Dataset<br/>固化为回归样本"]
    DATASET --> EXP["Experiment<br/>批量运行评测"]
    EXP --> COMPARE["Experiment Compare<br/>对比版本退化"]
    COMPARE --> GATE["本地 Gate<br/>确定是否允许发布"]

    TRACE -.-> M1["metadata<br/>scenario_id / kb_version / hit_type"]
    DETAIL -.-> M2["diagnostics<br/>sources_count / top_source_score / prompt_profile"]
    ANNO -.-> M3["expectation<br/>expected_source / expected_keywords"]
    EXP -.-> M4["scores<br/>hit_type_accuracy / keyword_coverage / error_rate"]

    style APP fill:#EFF6FF,stroke:#2563EB,stroke-width:2px
    style TRACE fill:#F8FAFC,stroke:#64748B,stroke-width:2px
    style DETAIL fill:#DBEAFE,stroke:#2563EB,stroke-width:2px
    style ANNO fill:#FFFBEB,stroke:#D97706,stroke-width:2px
    style DATASET fill:#ECFDF5,stroke:#059669,stroke-width:2px
    style EXP fill:#F5F3FF,stroke:#7C3AED,stroke-width:2px
    style COMPARE fill:#F5F3FF,stroke:#7C3AED,stroke-width:2px
    style GATE fill:#FEF2F2,stroke:#DC2626,stroke-width:2px

可以把页面信息理解成三层：

层级 页面里看到的内容 本项目如何使用
输入输出层 用户问题、历史消息、最终答案、引用来源 判断现象是否符合预期
链路诊断层 意图、路由、召回、重排、Prompt、LLM 子步骤 判断问题发生在哪个节点
验收字段层 expected_*、score、pass/fail、metadata 判断能不能进入 Gate

4.7.3 用 VPN Bad Case 走一遍网页端

仍然使用前面的 VPN 示例：

1
VPN 客户端版本、账号锁定、公网 IP 这些排查项分别应该怎么处理？

第 1 步：在 Traces 页面定位样本。

可以先用 metadata 过滤出疑似样本：

1
2
3
4
scenario_id = enterprise_knowledge
question_category = troubleshooting
hit_type = rag
top_source_score < 0.7

打开 Trace 后，先看四件事：

查看项 如果异常 说明
hit_type 不是 rag 路由或上下文不足判断可能有问题
kb_version 不是当前 active 版本 版本过滤或部署环境可能不一致
sources_count 等于 0 检索没有给 LLM 提供资料
top_source_score 明显偏低 召回质量不足，可能需要补资料或改写查询

第 2 步：进入 Run detail 看子步骤。

这一页不要只看最终答案，要展开检索和生成相关的子 Run：

子步骤 重点看 常见结论
FAQ retrieval FAQ 最高分、是否高置信直出 判断是否被 FAQ 短路
Doc retrieval 命中文档、source、score、chunk 内容 判断证据是否覆盖问题
Rerank 重排前后排序是否变化 判断相关证据是否被排到前面
Prompt build 使用的 Prompt Profile、上下文长度 判断模板和上下文是否合理
LLM generation 最终回答、引用编号、耗时 判断是否生成阶段漏答

对于 VPN 示例，如果检索命中了 IT 资料，但答案没有覆盖“客户端版本、账号锁定、公网 IP”，问题更可能在上下文覆盖或 Prompt 约束上，而不是场景路由。

第 3 步：在 Feedback / Annotation 中写清楚期望。

人工复核不是简单写“错了”，而是把期望写成可评测字段：

标注字段 含义 示例
is_correct 本次回答是否合格 false
issue_type 问题类型 missing_sub_questions
expected_hit_type 期望命中路径 rag
expected_source 应召回的业务分类 it
expected_prompt_profile 应使用的 Prompt 档位 troubleshooting_steps
expected_keywords 答案必须覆盖的关键词 ["客户端版本", "账号锁定", "公网 IP", "IT 工单", "截图"]

第 4 步：加入 Dataset。

加入 Dataset 时建议保留输入、期望字段和关键 metadata：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
{
  "inputs": {
    "query": "VPN 客户端版本、账号锁定、公网 IP 这些排查项分别应该怎么处理？",
    "history": [],
    "source_filter": "it"
  },
  "outputs": {
    "expected_hit_type": "rag",
    "expected_source": "it",
    "expected_prompt_profile": "troubleshooting_steps",
    "expected_keywords": [
      "客户端版本",
      "账号锁定",
      "公网 IP",
      "IT 工单",
      "截图"
    ]
  },
  "metadata": {
    "scenario_id": "enterprise_knowledge",
    "case_type": "troubleshooting",
    "priority": "high"
  }
}

Dataset 可以放在 enterprise_it_troubleshooting_cases，也可以放入更大的 enterprise_knowledge_regression，但必须能按 scenario_id 和 case_type 筛选。

第 5 步：运行 Experiment。

Experiment 不是只看一个分数，而是看一批样本在同一个版本上的执行结果。对本项目来说，Evaluator 至少分三类：

Evaluator 类型 判断什么 本项目对应指标
规则型 evaluator 路径和字段是否命中 expected_source、expected_hit_type、expected_prompt_profile
文本型 evaluator 答案是否覆盖关键事实 keyword_coverage、引用完整性
LLM-as-judge evaluator 复杂答案质量 完整性、准确性、是否越界

本项目本地脚本已经实现了规则型和领域指标型检查；LangSmith Evaluation 更适合承接 trace 样本、人工标注、实验对比和长期趋势。

第 6 步：对比 Experiment。

每次改动都应该形成新的 Experiment 名称，例如：

  • rerank_threshold_v2
  • prompt_profile_guard_update
  • bge_m3_rebuild_202606
  • cross_border_query_rewrite_v3

对比时不要只看平均分，要看分组：

  • 按 scenario_id 看是否某个场景退化。
  • 按 hit_type 看 FAQ 直出和文档 RAG 是否分别稳定。
  • 按 source_filter 看是否某个分类召回变差。
  • 按 prompt_profile 看高风险问题是否仍然走正确模板。

第 7 步：本地 Gate 复核。 LangSmith Evaluation 给出趋势和样本级结果，本地 gate 负责把结果变成“能不能发布”的工程约束。推荐口径是：

1
2
LangSmith Evaluation 负责发现和解释退化；
本地 Gate 负责给出是否允许发布的确定性结论。

也就是说，网页端用于观察、复核、沉淀和对比；本地脚本用于 CI 或发布流程中的确定性拦截。二者不是替代关系，而是互补关系。

4.8 这条 Bad Case 如何影响 Gate

把 VPN 示例加入 Dataset 后，下一次运行 Evaluation 时，这条样本会变成一条明确的验收用例。失败结果可以长成这样：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
{
  "case_id": "enterprise_it_vpn_sub_questions_001",
  "query": "VPN 客户端版本、账号锁定、公网 IP 这些排查项分别应该怎么处理？",
  "expected_hit_type": "rag",
  "expected_source": "it",
  "expected_prompt_profile": "troubleshooting_steps",
  "expected_keywords": [
    "客户端版本",
    "账号锁定",
    "公网 IP",
    "IT 工单",
    "截图"
  ],
  "actual_hit_type": "rag",
  "actual_source_hit": true,
  "actual_prompt_profile": "troubleshooting_steps",
  "keyword_coverage": 0.4,
  "passed": false,
  "failures": [
    "missing_keywords: 客户端版本, 账号锁定, 公网 IP"
  ]
}

这份结果说明：检索路径、业务分类和 Prompt 档位都没错，但答案没有覆盖关键排查项。此时 Gate 不应该放行，因为同一个线上问题仍然会复发。

检查点 失败含义 应该修哪里
actual_hit_type != expected_hit_type 命中路径错了 查意图分类、直出规则、上下文不足判断
actual_source_hit = false 没召回到正确业务资料 查 source 推断、Milvus 过滤、资料入库
actual_prompt_profile != expected_prompt_profile Prompt 档位错了 查问题类别识别和 Profile 选择
keyword_coverage 低于阈值 答案漏掉关键事实 查上下文覆盖、Prompt 约束或资料内容

修复后，这条样本的结果应该变成：

1
2
3
4
5
6
7
8
{
  "case_id": "enterprise_it_vpn_sub_questions_001",
  "actual_hit_type": "rag",
  "actual_source_hit": true,
  "actual_prompt_profile": "troubleshooting_steps",
  "keyword_coverage": 1.0,
  "passed": true
}

把这条链路说完整后，Bad Case 沉淀就不再是抽象概念，而是一个可执行的质量机制：

1
2
3
4
5
6
线上问题
  -> Trace 定位
  -> Annotation 写清楚期望
  -> Dataset 固化样本
  -> Evaluation 重跑验证
  -> Gate 决定是否允许发布

需要抓住三句话：

  1. Bad Case 不是聊天记录，而是可复现、可标注、可验收的质量资产。
  2. Annotation 把“感觉不满意”变成结构化期望字段。
  3. Gate 让同类问题不会在下一次版本变更中悄悄复发。

第五部分：回归验收体系汇总

5.1 全部回归验收

1
2
3
4
5
6
7
8
9
# 检查顺序
1. 项目守护检查 (check_project_guardrails.py)
2. 编译检查 (Python 语法)
3. 单元测试 (python -m pytest tests -q)
4. 入库质量检查 (check_ingestion_quality_gate.py)
5. RAG 回归验收 (check_evaluation_gate.py)
6. 追问回归验收 (check_followup_gate.py)
7. 性能回归验收 (check_performance_gate.py)
8. API 合同验收 (api_e2e_smoke.py)

5.2 接口验收

1
2
python scripts/api_e2e_smoke.py --base-url http://127.0.0.1:8000
python scripts/acceptance_smoke.py --base-url http://127.0.0.1:8000

验证管理接口、问答页面和 WebSocket 流式事件是否可用。

5.3 评测趋势

状态页只保留回归报告入口；历次评测对比优先在 LangSmith Experiments 中查看：

1
2
3
4
5
                Recall@K    MRR    关键词覆盖  Source 推断  场景隔离
2026-05-01 v1:   1.000     0.900    0.933      1.000       1.000
2026-05-07 v2:   1.000     0.920    0.945      1.000       1.000
2026-05-14 v3:   0.980 ⚠   0.910    0.940      0.980 ⚠     1.000
                                  ↑ 需要排查 v3 的退化原因

第六部分：核心评测结果

本项目已完成最终验收，核心指标如下：

指标 值 说明
errors 0 零错误
recall_at_k 1.0 期望关键词全部被召回
mrr 0.9 正确答案平均排在第 1.1 位
avg_keyword_coverage 0.933 93.3% 的关键词出现在召回文档中
hit_type_accuracy 1.0 命中类型判断全部正确
source_inference_accuracy 1.0 业务分类推断全部正确
prompt_profile_accuracy 1.0 Prompt 模板选择全部正确
faq_direct_accuracy 1.0 FAQ 直出全部准确
scenario_isolation_accuracy 1.0 无跨场景数据泄露
avg_total_ms 3444 平均总耗时 3.4 秒
p95_total_ms 12810 P95 耗时 12.8 秒
avg_first_token_ms 2479 平均首 token 耗时 2.5 秒

本讲实践闭环

项目 内容
本讲类型 工程治理
实践产物 入库质量报告、RAG Evaluation、质量门禁和 Bad Case 闭环
是否进入最终项目 是
验收方式 运行评测/门禁脚本，生成报告并能识别退化
后续落点 第 18 讲转成自动化测试，第 19 讲纳入生产观测

通过标准：系统质量不是凭感觉判断，而是通过指标、报告和门禁证明。

本讲从 0 到 1 实现闭环

这一讲把“我感觉效果还行”改成“指标和报告证明可以上线”。实现顺序如下：

flowchart TD
    Data["入库后的知识库版本"] --> IngestionReport["入库质量报告<br/>空文件/重复 FAQ/低质量 chunk"]
    Data --> Eval["核心链路评测<br/>Recall@K / MRR / 耗时 / 隔离"]
    IngestionReport --> Gate{"质量门禁"}
    Eval --> Gate
    Gate -->|"通过"| Activate["允许激活或发布"]
    Gate -->|"失败"| Block["阻断发布<br/>输出失败项"]
    Block --> Fix["修复资料/策略/Prompt"]
    Fix --> Data
  1. 先做入库质量检查，发现空文件、重复 FAQ、无效 source、低质量 chunk。
  2. 再准备回归评测集，覆盖 8 个场景、FAQ、文档、追问、高风险类别。
  3. 然后运行核心链路评测，统计 Recall@K、MRR、首 token、总耗时、场景隔离准确率。
  4. 最后执行质量门禁，指标低于阈值就拒绝激活或发布。

实现完成后，相关代码结构应该是下面这张图：

flowchart LR
    subgraph Quality["qa_core/quality"]
        Ingestion["ingestion.py<br/>入库质量汇总"]
        FAQ["faq.py<br/>FAQ 空值/重复检查"]
        Chunk["chunk.py<br/>低质量 chunk 检查"]
        Conflict["conflicts.py<br/>FAQ/文档冲突"]
    end

    subgraph Scripts["scripts"]
        Eval["evaluate_core_chain.py<br/>核心链路评测"]
        EvalGate["check_evaluation_gate.py<br/>评测门禁"]
        IngestGate["check_ingestion_quality_gate.py<br/>入库门禁"]
        PerfGate["check_performance_gate.py<br/>性能门禁"]
    end

    Quality --> IngestGate
    Eval --> EvalGate
    Eval --> PerfGate

来源：真实代码调用点，见 qa_core/quality/。

1
2
3
4
5
6
7
def build_ingestion_quality_report(data_dir, scenario):
    return {
        "empty_files": check_empty_files(data_dir),
        "duplicate_faq": check_duplicate_faq(scenario),
        "invalid_sources": check_invalid_sources(scenario),
        "low_quality_chunks": check_low_quality_chunks(data_dir),
    }

RAG 评测不是只看答案文本，还要看检索命中、场景隔离和耗时。否则模型恰好蒙对，也会掩盖检索退化。

来源：真实代码逻辑压缩版，对应 scripts/evaluate_core_chain.py::run_case()。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def run_case(service, item, index, args):
    runtime = EvalCaseRuntime.from_item(item, index, args, session_prefix="eval")

    # 先跑 debug_retrieval，把“召回是否命中”和“最终生成是否答对”拆开看。
    try:
        debug_payload = service.debug_retrieval(
            runtime.question,
            runtime.source_filter,
            runtime.session_id,
            **runtime.service_kwargs(),
        )
    except Exception as exc:
        debug_payload = {"error": str(exc)}

    # 再跑正式 stream_query，收集 token、end 事件、hit_type、sources 和 retrieval 诊断。
    answer = ""
    for event in service.stream_query(
        runtime.question,
        runtime.source_filter,
        runtime.session_id,
        **runtime.service_kwargs(),
    ):
        if event["type"] == "token":
            answer += event.get("token", "")
        elif event["type"] == "end":
            hit_type = event.get("hit_type", "")
            sources = event.get("sources", [])
            retrieval = event.get("retrieval") or {}

    debug_sources = list(debug_payload.get("faq_sources") or []) + list(debug_payload.get("doc_sources") or [])
    rank = find_expected_source_rank(debug_sources or sources, expected_sources, prefer_table=prefer_table)
    coverage = keyword_coverage(answer, expected_keywords)
    return build_case_metrics(rank=rank, coverage=coverage, hit_type=hit_type, retrieval=retrieval)

设计解释：先 debug_retrieval() 再 stream_query() 是为了分离两类问题：如果 debug 阶段没召回预期来源，问题在入库、过滤、query variants 或阈值；如果 debug 命中但最终答案不对，问题更可能在 Prompt 或模型生成。

质量门禁要做成脚本，而不是人工看报告。这样入库脚本和 CI 都能复用同一套规则。

来源：真实代码调用点，见 scripts/check_ingestion_quality_gate.py、scripts/check_evaluation_gate.py、scripts/check_performance_gate.py。

1
2
3
4
5
if report["recall_at_5"] < min_recall_at_5:
    raise SystemExit("quality gate failed: recall_at_5 too low")

if report["scenario_isolation_accuracy"] < 1.0:
    raise SystemExit("quality gate failed: scenario leakage")

验收时至少跑一次核心链路评测和门禁检查，确认报告能落盘、阈值能阻断退化。

来源：命令行验收，对应 scripts/evaluate_core_chain.py。

1
2
3
python scripts/evaluate_core_chain.py --scenario enterprise_knowledge
python scripts/check_evaluation_gate.py --report reports/latest_eval_summary.json
python scripts/check_ingestion_quality_gate.py --report reports/latest_ingestion_quality.json

闭环验证重点：

验证项 验证方式 期望结果
入库质量 生成质量报告 能发现空文件、重复 FAQ、低质量 chunk
检索指标 运行核心链路评测 输出 Recall@K、MRR 等指标
Debug/生成分离 查看单 case 报告 能区分召回失败和生成失败
场景隔离 评测跨场景问题 不出现跨场景召回
性能基线 检查耗时报告 首 token 和总耗时有统计
门禁阻断 设置严格阈值 不达标时脚本失败

验收重点：系统质量要能被报告证明，退化时 gate 应拒绝通过；不要靠主观体验判断是否上线。

重点掌握

优先级 内容 原因
★★★ 必会 三层保障体系：入库质量（资料健康）→ 检索评测（策略有效）→ 性能基线（响应合理） RAG 系统RAG 回归与入库质量的全景图
★★★ 必会 Recall@K 手算：期望关键词在前 K 个召回文档中的覆盖率 RAG 检索评估核心指标
★★★ 必会 MRR（Mean Reciprocal Rank）手算：第一个相关文档在召回列表中排名的倒数平均值 衡量相关证据排序位置
★★★ 必会 分组验收设计：按场景 / source / hit_type 分组检查，防止局部退化被全局均值掩盖 回归验收体系的关键设计，避免"平均主义"陷阱
★★ 理解 入库质量检查的检查项：文件解析失败率、低质量 chunk 比例、FAQ 空值/重复率、FAQ/文档冲突 知识库上线前的质量把关
★★ 理解 Bad Case 沉淀流程：LangSmith Trace → Annotation → Dataset → Evaluation 持续改进的完整闭环
★★ 理解 质量检查总览：项目守护、单元测试、入库质量、评测、追问、性能和 API 合同 变更前后的质量证明
★ 了解 关键词覆盖率的手算方法和评测数据集 JSON 格式 了解评测数据的结构
★ 了解 LangSmith Experiments 和本地领域指标如何配合 企业落地时用成熟平台承接评测趋势

本讲小结

  • 三层保障：入库质量（资料健康）→ 检索评测（策略有效）→ 性能基线（响应合理）
  • 分组验收防止局部退化被全局均值掩盖：按场景、source、hit_type 分别检查
  • Bad Case 沉淀：LangSmith Trace → Annotation → Dataset → Evaluation
  • 回归验收体系是可执行的工程约束，不是建议性文档 —— 验收不通过时阻止继续发布
  • 评测数据全部保存在 reports/ 和 eval_sets/ 中，可版本管理、可历史对比

阶段小结：到第 17 讲你已经具备的能力

学习到第 17 讲，你应该已经掌握了以下能力：

  1. RAG 系统架构设计：从检索到生成的完整链路，不是 Demo 而是企业级工程
  2. 分层检索策略：FAQ 优先 + 文档补充，混合检索 + 动态阈值
  3. Prompt 工程：按问题类别选择模板，确定性的 Profile 选择
  4. 知识库治理：多版本管理、数据隔离、增量入库
  5. RAG 回归与入库质量：入库质量检查、LangSmith Evaluation、领域指标回归验收、Bad Case 沉淀
  6. 工程化思维：入口极薄、模块拆分、拥抱生态、不走简化旁路

这些能力不仅适用于本项目，也适用于任何需要构建 RAG 系统的场景。后续第 18 讲会把这些质量目标转成可运行的测试与接口验收，第 19 讲会进一步讲线上观测、生产部署和容量评估。