Agent: 叮咚～【每日 AI 学习笔记】Day 10 已送达。

今天是 Day 10：构建 RAG 自动化测试流水线。

Day 9 我们把评测指标与 Ground Truth 补齐了；Day 10 的目标是：把评测变成流水线，让 RAG 像普通服务一样“每次变更必回归”。

---

每日 AI 学习笔记 | Day 10: 构建 RAG 自动化测试流水线（CI 回归 + 指标门禁）

0. 核心结论（TL;DR）

1. RAG 的回归点不仅是代码：更常见的变化来自文档、切片策略、embedding 模型、索引参数、过滤条件。
2. 流水线要同时测两条链路：
   • Retrieval Gate：Recall@K / MRR / 空召回率
   • Answer Gate：答案与标准答案的相似度（ROUGE/Embedding Sim）、以及“是否忠实于证据”（Faithfulness）
3. 不要追求一次到位的满分评测：先把 P0 题集跑起来并设红线，再逐步扩展维度。

---

1. 一条“可落地”的 RAG 测试流水线长什么样？

下面是一条典型的流水线（你可以映射到 GitHub Actions / Jenkins / 内部流水线）：

1. 准备数据

• 拉取文档（或固定版本快照）
• 运行解析与切片

2. 构建索引（Build Index）

• 生成 embeddings
• 写入向量库（或构建检索索引）

3. 运行评测（Run Evaluation）

• Retrieval：对 ground truth 跑检索评测
• Answer：对 ground truth 跑生成评测

4. 生成报告（Report）

• 输出 JSON/HTML/Markdown 报告
• 记录指标与趋势（便于发现“缓慢退化”）

5. 门禁（Gate）

• 指标低于阈值 → fail pipeline
• 允许白名单/人工审批（可选）

---

2. Answer 评测：怎么自动对比“回答 vs 标准答案”？

2.1 三种常用对比方式（从稳到强）

1. 字符串重合度（ROUGE / F1）

• 优点：快、确定性强
• 缺点：同义改写会被误判为不相似

2. 语义相似度（Embedding Similarity）

• 优点：对同义改写鲁棒
• 缺点：阈值需要寻优；embedding 模型升级会影响分数

3. LLM-as-a-Judge / RAGAS Faithfulness

• 优点：更接近“人类验收”
• 缺点：成本高、需要固定评审模型版本与温度，避免不可复现

QA 建议：Day 10 先落地 1)+2)，确保“快且稳定”；3) 作为增强项按需加入。

---

3. Python 实践：自动比对回答与标准答案（ROUGE + Embedding）

下面给出一个“可跑进 CI 的最小评测脚本骨架”。

• 输入：ground_truth.jsonl
• 输出：report.json（包含每条用例的 retrieval 命中情况、答案分数、失败原因）

# eval_rag_pipeline.py

import json
from dataclasses import dataclass

# 可选：如果你不想引入第三方库，先用最简单的 token F1

def token_f1(a: str, b: str) -> float:
    ta = [x for x in a.lower().split() if x]
    tb = [x for x in b.lower().split() if x]
    if not ta or not tb:
        return 0.0
    sa, sb = set(ta), set(tb)
    inter = len(sa & sb)
    p = inter / len(sa)
    r = inter / len(sb)
    return 0.0 if (p + r) == 0 else 2 * p * r / (p + r)


@dataclass
class Case:
    id: str
    query: str
    expected_answer: str
    expected_evidence_ids: list[str]


def load_cases(path: str) -> list[Case]:
    out = []
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            o = json.loads(line)
            out.append(Case(
                id=o["id"],
                query=o["query"],
                expected_answer=o.get("expected_answer", ""),
                expected_evidence_ids=o.get("expected_evidence_ids", []),
            ))
    return out


def main():
    cases = load_cases("ground_truth.jsonl")

    report = {"cases": [], "summary": {}}

    hit_cnt = 0
    f1_sum = 0.0

    for c in cases:
        # 1) 检索
        # topk = retriever.search(c.query, top_k=5)  # -> [{chunk_id, score, text}, ...]
        topk = [{"chunk_id": "doc1#chunk_0", "score": 0.88, "text": "..."}]  # mock

        got_ids = [x["chunk_id"] for x in topk]
        expected = set(c.expected_evidence_ids)
        retrieval_hit = any(i in expected for i in got_ids)
        hit_cnt += 1 if retrieval_hit else 0

        # 2) 生成（示例：真实场景替换为 LLM 调用）
        # answer = llm.generate(query=c.query, contexts=[x["text"] for x in topk])
        answer = "必须返回 error_message，并携带 trace_id"  # mock

        # 3) 答案对比（token F1 先兜底）
        f1 = token_f1(answer, c.expected_answer)
        f1_sum += f1

        report["cases"].append({
            "id": c.id,
            "query": c.query,
            "retrieval_topk_ids": got_ids,
            "retrieval_hit": retrieval_hit,
            "answer": answer,
            "expected_answer": c.expected_answer,
            "token_f1": f1,
        })

    n = max(1, len(cases))
    report["summary"] = {
        "retrieval_recall_at_5": hit_cnt / n,
        "avg_token_f1": f1_sum / n,
    }

    with open("report.json", "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)

    print("Summary:", report["summary"])

    # 4) 质量门禁示例
    assert report["summary"]["retrieval_recall_at_5"] >= 0.85, "Recall@5 regression"
    assert report["summary"]["avg_token_f1"] >= 0.70, "Answer overlap regression"


if __name__ == "__main__":
    main()

提示：一开始用 token F1 没问题。后续你可以把 token_f1 替换为：

• ROUGE-L（更贴近摘要式答案）
• Embedding Similarity（更抗同义改写）

---

4. Go（Ginkgo）兜底：把“流水线门禁”写成测试

当你希望 CI 在 Go 侧统一做门禁（或做二次断言）时，可以把 report.json 当成测试输入：

var _ = Describe("RAG Pipeline Gate", func() {
    It("should pass retrieval and answer thresholds", func() {
        rep := LoadReport("report.json") // 伪代码

        Expect(rep.Summary.RetrievalRecallAt5).To(BeNumerically(">=", 0.85))
        Expect(rep.Summary.AvgTokenF1).To(BeNumerically(">=", 0.70))

        // 可选：对关键用例做强约束（P0 永不退化）
        p0 := rep.FindCase("GT-001")
        Expect(p0.RetrievalHit).To(BeTrue())
        Expect(p0.TokenF1).To(BeNumerically(">=", 0.80))
    })
})

---

5. QA Checklist：把 RAG 回归做“可维护”

为了避免“评测脚本越写越乱”，建议你从 Day 10 起把下面这些固化成规范：

1. 固定版本：

• embedding 模型版本
• prompt 版本
• 检索参数（TopK、filters、rerank 开关）

2. 固定输入快照：

• 文档来源（hash / 版本号）
• 切片策略（chunk_size/overlap/分隔符）

3. 可解释失败：

• 检索没命中？（MISS）
• 命中了但答案不对？（GEN_FAIL）
• 命中了但引用不一致？（CITATION_FAIL）

---

6. 课后小思考

1. 你认为 RAG 的“变更触发回归”规则应该怎么定？（例如：文档更新、embedding 升级、索引参数变更、prompt 变更）
2. 如果 Recall@5 没退化，但 avg_token_f1 退化了，你会优先怀疑哪里？你会加什么日志来定位？
3. 你会如何定义 P0 用例（永不退化）与 P1 用例（允许波动），以减少误报与维护成本？

---

到这里，RAG 这条线我们已经具备“搭建 + 评测 + 流水线门禁”的闭环能力了。

下一次 Day 11：进入 Agent 世界，拆解 Profile / Memory / Planning / Action，把“会做事的系统”也纳入可测试范畴。