RAG 评估体系：用 RAGAS 量化你的检索质量
·
RAG 评估体系：用 RAGAS 量化你的检索质量
本文是我学习 RAG 评估体系过程中的总结与梳理。在写完《RAG 文档分块策略完全指南》、《向量数据库选型与实践》和《RAG 检索优化：从基础到高级的 10 种策略》之后，我的 RAG 系统已经具备了完整的能力——文档切好了、向量库选好了、检索也优化了。但一个关键问题始终悬而未决：我怎么知道这些优化到底有没有效果？ 这篇文章就是我对这个问题的系统回答——从 RAGAS 的八大核心指标到评估流程自动化，从评估数据集构建到持续监控方案，把 RAG 系统的质量量化、可视化、可追踪。
声明：本文为作者在学习过程中的总结与梳理，仅供学习参考。由于作者水平有限，文中可能存在表述不准确或遗漏之处，欢迎读者提出指正与交流。
目录
- 引言：RAG 系统的"最后一公里"——评估
- RAGAS 是什么：我的理解
- RAGAS 八大核心指标详解
- 评估数据集构建
- 评估流程自动化
- 持续监控方案
- 实战：完整评估流水线
- 常见问题与避坑指南
- 总结与学习建议
1. 引言：RAG 系统的"最后一公里"——评估
1.1 从"能跑"到"跑得好"
回顾一下我构建 RAG 系统的完整历程：
Phase 4: RAG 系统构建
文档分块策略 → 向量数据库选型 → 检索优化策略 → 评估体系
(第7篇) (第8篇) (第9篇) (第10篇)
✅ ✅ ✅ ← 本文
前三篇文章分别解决了"怎么切"、“存哪里”、"怎么查"的问题。但学完之后我发现，还有一个更根本的问题没有解决：我怎么判断这些优化到底有没有效果？
举个例子：我在检索优化那篇文章里实现了混合检索、Reranker、HyDE 等策略。但当我实际部署之后，面对的是这样的困惑：
场景一：改了分块大小
之前 chunk_size=500，现在改成 chunk_size=800
→ 检索结果变了，但变好了还是变差了？不知道。
场景二：加了 Reranker
之前只用语义检索，现在加了 Cross-Encoder 重排序
→ 答案看起来更准了，但有没有引入新的问题？不知道。
场景三：换了 Embedding 模型
之前用 text-embedding-3-small，现在换成 bge-m3
→ 成本降低了，但检索质量有没有下降？不知道。
这些"不知道"让我意识到：没有评估的优化是盲目的。你可能花了很多时间调参数、加策略，但最终效果可能还不如不改。
1.2 为什么 RAG 评估这么难
刚开始我以为评估 RAG 系统很简单——找几个问题问问，看看答案对不对就行了。但很快我就发现事情没那么简单：
传统 NLP 任务的评估：
分类任务 → 准确率（Accuracy）
翻译任务 → BLEU 分数
摘要任务 → ROUGE 分数
→ 都有明确的 ground truth，算就行了
RAG 系统的评估：
问题："公司年假政策是什么？"
答案："根据公司规定，员工入职满1年享有5天年假..."
→ 怎么判断这个答案好不好？
→ 检索到的文档对不对？
→ 答案有没有编造不存在的信息？
→ 这些维度怎么量化？
RAG 评估的难点在于它是多维度的：
- 检索维度：检索到的文档是否相关？有没有漏掉关键文档？
- 生成维度：生成的答案是否准确？有没有幻觉？
- 端到端维度：整个流程下来，用户是否得到了满意的答案？
这三个维度相互关联但又各自独立。检索好不代表生成好（检索到了但 LLM 没用对），生成好不代表检索好（LLM 用自己的知识补上了检索的不足）。
1.3 RAGAS 的出现
在我为评估问题头疼的时候，发现了 RAGAS（Retrieval Augmented Generation Assessment）——一个专门为 RAG 系统设计的评估框架。它的核心思路是：把 RAG 评估拆解成多个可量化的指标，每个指标聚焦一个维度，最终形成一个完整的评估画像。
RAGAS 的论文发表于 2023 年，目前在 GitHub 上有 10K+ stars，已经成为 RAG 评估领域的事实标准。它的设计理念我概括为三点：
- 不需要人工标注 ground truth：大部分指标只需要问题、检索上下文和生成答案，不需要"标准答案"
- 用 LLM 做评判：利用 LLM 的理解能力来判断答案质量，而不是简单的字符串匹配
- 指标相互独立：每个指标衡量一个维度，组合起来形成完整画像
2. RAGAS 是什么：我的理解
2.1 核心架构
RAGAS 的评估架构可以概括为这张图：
RAGAS 评估架构
输入 评估指标 输出
──── ────────── ────
question ──────────────┬──→ Faithfulness ──────→ 忠实度分数
│
answer ────────────────┤──→ Answer Relevancy ──→ 相关性分数
│
contexts ──────────────┤──→ Context Precision ─→ 精确度分数
│
ground_truth(可选) ────┤──→ Context Recall ────→ 召回率分数
│
├──→ Context Relevancy ─→ 上下文相关性
│
├──→ Entity Recall ─────→ 实体召回率
│
├──→ Semantic Similarity → 语义相似度
│
└──→ Answer Correctness ─→ 正确性分数
2.2 指标分类
我把 RAGAS 的八大指标分为三类：
| 类别 | 指标 | 评估对象 | 是否需要 ground truth |
|---|---|---|---|
| 生成质量 | Faithfulness | 答案是否忠于检索到的上下文 | 否 |
| Answer Relevancy | 答案是否与问题相关 | 否 | |
| Answer Correctness | 答案是否事实正确 | 是 | |
| Answer Semantic Similarity | 答案与标准答案的语义相似度 | 是 | |
| 检索质量 | Context Precision | 检索到的文档是否都相关 | 否 |
| Context Recall | 是否检索到了所有相关文档 | 是 | |
| Context Relevancy | 检索到的文档与问题的相关性 | 否 | |
| Context Entity Recall | 检索到的文档覆盖了多少关键实体 | 是 |
2.3 安装与基础使用
pip install ragas datasets langchain-openai
版本说明：RAGAS 目前主要版本为 0.1.x 和 0.2.x，两者 API 有较大差异。0.1.x 使用
ragas.evaluate()函数式 API，0.2.x 引入了EvaluationDataset和evaluate()的新用法。本文代码基于 ragas 0.1.x（推荐用于生产环境，生态更成熟），如果你使用的是 0.2.x+，部分 API 需要调整，建议参考 RAGAS 官方迁移指南。安装时建议锁定版本：pip install ragas==0.1.20。
# ragas_basic.py —— RAGAS 基础使用示例
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
context_relevancy,
)
from datasets import Dataset
# 准备评估数据
eval_data = {
"question": [
"公司年假政策是什么？",
"如何申请差旅报销？",
"2025年Q3的营收是多少？",
],
"answer": [
"员工入职满1年享有5天年假，满3年享有10天年假，满5年享有15天年假。",
"差旅报销需在出差结束后7个工作日内提交申请，附上所有票据。",
"2025年Q3公司营收达到15亿元，同比增长23%。",
],
"contexts": [
["员工入职满1年享有5天年假，满3年享有10天年假..."],
["差旅报销需在出差结束后7个工作日内提交报销申请..."],
["2025年第三季度公司营收达到15亿元，较去年同期增长23%..."],
],
"ground_truth": [
"入职满1年5天年假，满3年10天，满5年15天。",
"出差后7个工作日内提交报销申请并附票据。",
"2025年Q3营收15亿元，同比增长23%。",
],
}
dataset = Dataset.from_dict(eval_data)
# 运行评估
result = evaluate(
dataset,
metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall,
context_relevancy,
],
)
print(result)
输出示例：
{
'faithfulness': 0.85,
'answer_relevancy': 0.92,
'context_precision': 0.78,
'context_recall': 0.82,
'context_relevancy': 0.75,
}
3. RAGAS 八大核心指标详解
3.1 Faithfulness（忠实度）
是什么
Faithfulness 衡量的是：生成的答案中有多少内容是可以从检索到的上下文中推导出来的。换句话说，它检测 LLM 有没有"编造"信息（幻觉）。
这是我最看重的指标，因为 RAG 的核心价值就是让 LLM 基于外部知识回答，如果答案中的信息不在检索到的文档里，那 RAG 就失去了意义。
计算原理
RAGAS 的 Faithfulness 计算分为两步：
第一步：从答案中提取"声明"（claims）
答案："员工入职满1年享有5天年假，满3年享有10天年假。"
→ 声明1："员工入职满1年享有5天年假"
→ 声明2："员工入职满3年享有10天年假"
第二步：逐一判断每个声明是否能从上下文中推导出来
声明1 + 上下文 → LLM 判断 → "Yes"（上下文中确实有这句话）
声明2 + 上下文 → LLM 判断 → "Yes"（上下文中确实有这句话）
Faithfulness = 能从上下文推导的声明数 / 总声明数
= 2 / 2 = 1.0
代码示例
# faithfulness_demo.py —— Faithfulness 指标详解
from ragas.metrics import faithfulness
from ragas import evaluate
from datasets import Dataset
# 场景一：高忠实度（答案完全来自上下文）
data_high = Dataset.from_dict({
"question": ["年假怎么算？"],
"answer": ["入职满1年有5天年假，满3年有10天。"],
"contexts": [["员工入职满1年享有5天年假，满3年享有10天年假，满5年享有15天年假。"]],
})
result_high = evaluate(data_high, metrics=[faithfulness])
print(f"高忠实度场景: {result_high['faithfulness']}") # 预期: ~0.9-1.0
# 场景二：低忠实度（答案包含上下文中没有的信息）
data_low = Dataset.from_dict({
"question": ["年假怎么算？"],
"answer": ["入职满1年有5天年假，满3年有10天，而且年假可以累积到第二年使用。"],
"contexts": [["员工入职满1年享有5天年假，满3年享有10天年假，满5年享有15天年假。"]],
# 注意：上下文中没有提到"年假可以累积"
})
result_low = evaluate(data_low, metrics=[faithfulness])
print(f"低忠实度场景: {result_low['faithfulness']}") # 预期: ~0.5-0.7
我的理解
Faithfulness 是 RAG 系统的"底线指标"。如果这个分数低，说明 LLM 在大量编造信息，RAG 形同虚设。在实际项目中，我一般要求 Faithfulness 不低于 0.85。
提升 Faithfulness 的方法：
- 优化检索质量，确保检索到的文档包含足够的信息
- 在 Prompt 中强调"只基于提供的文档回答，不要编造"
- 使用更低的 temperature（0-0.3），减少 LLM 的"创造性"
3.2 Answer Relevancy（答案相关性）
是什么
Answer Relevancy 衡量的是：生成的答案与用户问题的相关程度。它检测 LLM 有没有"答非所问"。
这个指标和 Faithfulness 的区别在于：Faithfulness 关注"答案是否来自上下文"，Answer Relevancy 关注"答案是否回应了问题"。一个答案可以完全忠实于上下文（Faithfulness 高），但完全没有回答用户的问题（Answer Relevancy 低）。
计算原理
第一步：基于答案，用 LLM 反向生成可能的问题
答案："员工入职满1年享有5天年假"
→ 反向生成的问题：
- "年假有多少天？"
- "入职多久有年假？"
- "年假政策是什么？"
第二步：计算原始问题和反向生成问题的语义相似度
原始问题："公司年假政策是什么？"
反向问题1："年假有多少天？" → 相似度 0.75
反向问题2："入职多久有年假？" → 相似度 0.68
反向问题3："年假政策是什么？" → 相似度 0.92
Answer Relevancy = mean([0.75, 0.68, 0.92]) = 0.78
代码示例
# answer_relevancy_demo.py —— Answer Relevancy 指标详解
from ragas.metrics import answer_relevancy
from ragas import evaluate
from datasets import Dataset
# 场景一：高相关性（答案直接回应问题）
data_high = Dataset.from_dict({
"question": ["年假有多少天？"],
"answer": ["入职满1年有5天年假，满3年有10天，满5年有15天。"],
"contexts": [["员工入职满1年享有5天年假..."]],
})
result_high = evaluate(data_high, metrics=[answer_relevancy])
print(f"高相关性场景: {result_high['answer_relevancy']}")
# 场景二：低相关性（答案偏离问题）
data_low = Dataset.from_dict({
"question": ["年假有多少天？"],
"answer": ["公司非常重视员工的工作与生活平衡，提供了完善的福利体系。"],
"contexts": [["公司重视员工福利..."]],
})
result_low = evaluate(data_low, metrics=[answer_relevancy])
print(f"低相关性场景: {result_low['answer_relevancy']}")
我的理解
Answer Relevancy 是"用户体验指标"。即使答案完全正确（Faithfulness 高），如果它没有直接回应用户的问题，用户也会觉得"这个 AI 不太聪明的样子"。
提升 Answer Relevancy 的方法：
- 优化 Prompt，明确要求 LLM 直接回答问题
- 使用更好的 LLM（gpt-4o 比 gpt-4o-mini 更能抓住问题核心）
- 检索优化：如果检索到的文档不相关，LLM 很难给出相关答案
3.3 Context Precision（上下文精确度）
是什么
Context Precision 衡量的是：检索到的文档中，有多少是真正与问题相关的。它关注的是检索结果的"信噪比"——检索到的 Top-K 文档中，相关文档的比例。
这个指标对应传统信息检索中的 Precision@K，但 RAGAS 用 LLM 来判断相关性，比简单的关键词匹配更准确。
计算原理
问题："公司年假政策是什么？"
检索到的 Top-5 文档：
1. "员工入职满1年享有5天年假..." → LLM 判断 → 相关 ✓
2. "公司实行弹性工作制..." → LLM 判断 → 不相关 ✗
3. "年假需提前一周申请..." → LLM 判断 → 相关 ✓
4. "差旅报销需在7个工作日内..." → LLM 判断 → 不相关 ✗
5. "未使用的年假可累积..." → LLM 判断 → 相关 ✓
Context Precision = 3/5 = 0.60
RAGAS 的 Context Precision 还考虑了排序位置——排在前面的相关文档权重更高。其核心思想是：用户更关注排名靠前的结果，所以前面的相关文档应该获得更高的"奖励"：
Context Precision@K = Σ(k=1 to K) ( Precision@k × v_k ) / 总相关文档数
其中：
- Precision@k = (前k个文档中相关文档的数量) / k
- v_k ∈ {0, 1}，表示第k个位置的文档是否相关
- 总相关文档数 = Σ(k=1 to K) v_k
举例说明：假设检索到5个文档，相关性为 [相关, 不相关, 相关, 相关, 不相关]，即 v = [1, 0, 1, 1, 0]：
k=1: Precision@1 = 1/1 = 1.0, v_1=1 → 贡献 1.0×1 = 1.0
k=2: Precision@2 = 1/2 = 0.5, v_2=0 → 贡献 0.5×0 = 0
k=3: Precision@3 = 2/3 ≈ 0.67, v_3=1 → 贡献 0.67×1 ≈ 0.67
k=4: Precision@4 = 3/4 = 0.75, v_4=1 → 贡献 0.75×1 = 0.75
k=5: Precision@5 = 3/5 = 0.6, v_5=0 → 贡献 0.6×0 = 0
Context Precision@5 = (1.0 + 0 + 0.67 + 0.75 + 0) / 3 = 2.42/3 ≈ 0.807
注意：如果所有相关文档都排在最前面（v = [1, 1, 1, 0, 0]），Context Precision 会接近 1.0；如果相关文档排在后面（v = [0, 0, 1, 1, 1]），分数会显著降低。这就是"排序位置加权"的效果。
代码示例
# context_precision_demo.py —— Context Precision 指标详解
from ragas.metrics import context_precision
from ragas import evaluate
from datasets import Dataset
# 场景一：高精确度（检索结果都很相关）
data_high = Dataset.from_dict({
"question": ["年假政策"],
"answer": ["入职满1年有5天年假。"],
"contexts": [[
"员工入职满1年享有5天年假，满3年享有10天年假。",
"年假需提前一周向直属领导申请。",
"未使用的年假可累积至下一年度，最多累积15天。",
]],
})
result_high = evaluate(data_high, metrics=[context_precision])
print(f"高精确度场景: {result_high['context_precision']}")
# 场景二：低精确度（检索结果混入不相关文档）
data_low = Dataset.from_dict({
"question": ["年假政策"],
"answer": ["入职满1年有5天年假。"],
"contexts": [[
"员工入职满1年享有5天年假。",
"公司实行弹性工作制，核心工作时间为10:00至16:00。",
"差旅报销需在出差结束后7个工作日内提交。",
]],
})
result_low = evaluate(data_low, metrics=[context_precision])
print(f"低精确度场景: {result_low['context_precision']}")
我的理解
Context Precision 直接影响 Faithfulness——如果检索到的文档大部分不相关，LLM 要么基于不相关的文档编造答案（Faithfulness 低），要么拒绝回答（Answer Relevancy 低）。
提升 Context Precision 的方法：
- 使用 Reranker 对检索结果重排序
- 提高相似度阈值，过滤低分文档
- 使用混合检索减少噪音
3.4 Context Recall（上下文召回率）
是什么
Context Recall 衡量的是：ground truth（标准答案）中的信息，有多少能在检索到的文档中找到。它关注的是检索有没有"漏掉"关键信息。
这个指标需要 ground truth，因为它需要知道"正确答案应该包含哪些信息"。
计算原理
第一步：从 ground truth 中提取关键信息
ground_truth: "入职满1年5天年假，满3年10天，满5年15天。"
→ 信息1："入职满1年享有5天年假"
→ 信息2："入职满3年享有10天年假"
→ 信息3："入职满5年享有15天年假"
第二步：逐一判断每条信息是否能在检索到的文档中找到
信息1 + 检索文档 → LLM 判断 → "能找到" ✓
信息2 + 检索文档 → LLM 判断 → "能找到" ✓
信息3 + 检索文档 → LLM 判断 → "找不到" ✗（检索文档中没有提到满5年的情况）
Context Recall = 2/3 = 0.67
代码示例
# context_recall_demo.py —— Context Recall 指标详解
from ragas.metrics import context_recall
from ragas import evaluate
from datasets import Dataset
# 场景一：高召回率（检索文档覆盖了所有关键信息）
data_high = Dataset.from_dict({
"question": ["年假政策"],
"answer": ["入职满1年5天，满3年10天，满5年15天。"],
"contexts": [[
"员工入职满1年享有5天年假，满3年享有10天年假，满5年享有15天年假，满10年享有20天年假。",
]],
"ground_truth": ["入职满1年5天年假，满3年10天，满5年15天。"],
})
result_high = evaluate(data_high, metrics=[context_recall])
print(f"高召回率场景: {result_high['context_recall']}")
# 场景二：低召回率（检索文档遗漏了关键信息）
data_low = Dataset.from_dict({
"question": ["年假政策"],
"answer": ["入职满1年5天，满3年10天。"],
"contexts": [[
"员工入职满1年享有5天年假，满3年享有10天年假。",
]],
"ground_truth": ["入职满1年5天年假，满3年10天，满5年15天。"],
# 注意：ground truth 中有"满5年15天"，但检索文档中没有
})
result_low = evaluate(data_low, metrics=[context_recall])
print(f"低召回率场景: {result_low['context_recall']}")
我的理解
Context Recall 和 Context Precision 是一对需要平衡的指标。提高 Recall 通常意味着降低 Precision（检索更多文档，其中包含更多噪音），反之亦然。
在实际项目中，我通常优先保证 Recall——宁可多检索一些文档然后靠 Reranker 精排，也不要漏掉关键信息。因为漏掉的信息 LLM 永远看不到，而多检索的噪音可以通过后续处理过滤。
3.5 Context Relevancy（上下文相关性）
是什么
Context Relevancy 衡量的是：检索到的文档整体上与问题的相关程度。它和 Context Precision 的区别在于：Precision 是逐文档判断"这个文档是否相关"，Relevancy 是整体判断"这批文档和问题有多相关"。
计算原理
第一步：从检索文档中提取与问题相关的关键句子
检索文档 + 问题 → LLM 提取 → 相关句子列表
第二步：计算相关句子占比
Context Relevancy = 相关句子数 / 总句子数
代码示例
# context_relevancy_demo.py —— Context Relevancy 指标详解
from ragas.metrics import context_relevancy
from ragas import evaluate
from datasets import Dataset
data = Dataset.from_dict({
"question": ["如何申请年假？"],
"answer": ["年假需提前一周向直属领导提交申请。"],
"contexts": [[
"员工入职满1年享有5天年假，满3年享有10天年假。年假需提前一周向直属领导提交申请。",
"公司实行弹性工作制，核心工作时间为上午10:00至下午4:00。",
"差旅报销需在出差结束后7个工作日内提交报销申请。",
]],
})
result = evaluate(data, metrics=[context_relevancy])
print(f"上下文相关性: {result['context_relevancy']}")
3.6 Context Entity Recall（实体召回率）
是什么
Context Entity Recall 衡量的是：ground truth 中提到的关键实体（人名、地名、数字、日期等），有多少在检索到的文档中出现了。
这个指标特别适合评估 RAG 系统对事实性信息的覆盖能力。
计算原理
ground_truth: "2025年Q3公司营收15亿元，CEO张明在财报会议上宣布。"
→ 实体：2025年Q3、15亿元、张明、财报会议
检索文档中出现的实体：2025年Q3、15亿元、财报会议
→ 缺少：张明
Context Entity Recall = 3/4 = 0.75
3.7 Answer Semantic Similarity（答案语义相似度）
是什么
Answer Semantic Similarity 衡量的是：生成的答案与 ground truth 在语义上的相似程度。
它和传统的 BLEU/ROUGE 等 n-gram 匹配指标不同——它用 Embedding 模型计算语义相似度，所以"5天年假"和"享有五个工作日的带薪年假"能匹配上。
代码示例
# semantic_similarity_demo.py —— 语义相似度指标详解
from ragas.metrics import answer_semantic_similarity
from ragas import evaluate
from datasets import Dataset
# 场景一：高语义相似度（表述不同但意思相同）
data_high = Dataset.from_dict({
"question": ["年假政策"],
"answer": ["入职满一年可以享受五个工作日的带薪年假。"],
"contexts": [["员工入职满1年享有5天年假。"]],
"ground_truth": ["入职满1年享有5天年假。"],
})
result_high = evaluate(data_high, metrics=[answer_semantic_similarity])
print(f"高语义相似度: {result_high['answer_semantic_similarity']}")
# 场景二：低语义相似度（答案错误）
data_low = Dataset.from_dict({
"question": ["年假政策"],
"answer": ["入职满一年可以享受十个工作日的带薪年假。"],
"contexts": [["员工入职满1年享有5天年假。"]],
"ground_truth": ["入职满1年享有5天年假。"],
})
result_low = evaluate(data_low, metrics=[answer_semantic_similarity])
print(f"低语义相似度: {result_low['answer_semantic_similarity']}")
3.8 Answer Correctness（答案正确性）
是什么
Answer Correctness 是 RAGAS 中最"严格"的指标——它综合了语义相似度和事实正确性两个维度，需要 ground truth。
计算原理
Answer Correctness = w1 × 语义相似度 + w2 × 事实正确性
其中：
- 语义相似度：答案与 ground truth 的 Embedding 余弦相似度
- 事实正确性：用 LLM 判断答案中的陈述是否与 ground truth 一致
- w1、w2：权重（默认各 0.5）
代码示例
# answer_correctness_demo.py —— Answer Correctness 指标详解
from ragas.metrics import answer_correctness
from ragas import evaluate
from datasets import Dataset
data = Dataset.from_dict({
"question": ["年假政策"],
"answer": ["入职满1年有5天年假，满3年有10天，满5年有15天。"],
"contexts": [["员工入职满1年享有5天年假，满3年享有10天年假，满5年享有15天年假。"]],
"ground_truth": ["入职满1年5天年假，满3年10天，满5年15天。"],
})
result = evaluate(data, metrics=[answer_correctness])
print(f"答案正确性: {result['answer_correctness']}")
3.9 八大指标速查表
| 指标 | 评估对象 | 需要 GT | 核心问题 | 低分意味着 |
|---|---|---|---|---|
| Faithfulness | 生成质量 | 否 | 答案有没有编造？ | LLM 在大量幻觉 |
| Answer Relevancy | 生成质量 | 否 | 答案是否回应了问题？ | LLM 答非所问 |
| Answer Correctness | 生成质量 | 是 | 答案事实是否正确？ | 答案包含错误信息 |
| Answer Semantic Similarity | 生成质量 | 是 | 答案和标准答案像不像？ | 答案偏离预期 |
| Context Precision | 检索质量 | 否 | 检索到的文档是否都相关？ | 检索结果噪音大 |
| Context Recall | 检索质量 | 是 | 有没有漏掉关键文档？ | 检索遗漏了重要信息 |
| Context Relevancy | 检索质量 | 否 | 检索文档整体是否相关？ | 检索方向偏了 |
| Context Entity Recall | 检索质量 | 是 | 关键实体有没有覆盖？ | 实体信息缺失 |
4. 评估数据集构建
4.1 为什么需要专门的评估数据集
刚开始做评估时，我随手写了几个问题来测。但很快发现这样不行：
- 不系统：想到什么问什么，覆盖不全
- 不可复现：每次测的问题不一样，无法对比
- 没有难度分级：简单问题和复杂问题混在一起，看不出系统的真实水平
一个好的评估数据集应该具备以下特征：
评估数据集的要求
1. 覆盖全面
├── 覆盖所有文档类型（制度、报告、技术文档...）
├── 覆盖所有查询类型（事实查询、推理查询、比较查询...）
└── 覆盖不同难度（简单检索、多跳推理、模糊查询...）
2. 标注准确
├── ground_truth 经过人工验证
└── 相关文档标注准确
3. 规模适中
├── 太少（<20条）：统计不显著
└── 太多（>200条）：评估太慢、成本太高
4. 可复现
└── 固定数据集，每次评估用同一批数据
4.2 评估数据集构建代码
# eval_dataset_builder.py —— 评估数据集构建器
from dataclasses import dataclass, field
from typing import List, Optional, Dict
import json
@dataclass
class EvalSample:
"""单条评估样本"""
question: str # 用户问题
ground_truth: str # 标准答案
reference_contexts: Optional[List[str]] = None # 参考上下文（可选）
difficulty: str = "medium" # 难度：easy/medium/hard
category: str = "general" # 类别：factual/reasoning/comparison
metadata: Dict = field(default_factory=dict) # 额外元数据
class EvalDatasetBuilder:
"""评估数据集构建器"""
def __init__(self):
self.samples: List[EvalSample] = []
def add_sample(
self,
question: str,
ground_truth: str,
difficulty: str = "medium",
category: str = "general",
reference_contexts: Optional[List[str]] = None,
**metadata,
):
"""添加一条评估样本"""
sample = EvalSample(
question=question,
ground_truth=ground_truth,
reference_contexts=reference_contexts,
difficulty=difficulty,
category=category,
metadata=metadata,
)
self.samples.append(sample)
def add_batch(self, samples: List[dict]):
"""批量添加评估样本"""
for s in samples:
self.add_sample(**s)
def to_ragas_dataset(self):
"""转换为 RAGAS 需要的 Dataset 格式"""
from datasets import Dataset
data = {
"question": [],
"ground_truth": [],
}
for sample in self.samples:
data["question"].append(sample.question)
data["ground_truth"].append(sample.ground_truth)
return Dataset.from_dict(data)
def save(self, filepath: str):
"""保存评估数据集到 JSON 文件"""
samples_dict = []
for s in self.samples:
samples_dict.append({
"question": s.question,
"ground_truth": s.ground_truth,
"reference_contexts": s.reference_contexts,
"difficulty": s.difficulty,
"category": s.category,
"metadata": s.metadata,
})
with open(filepath, "w", encoding="utf-8") as f:
json.dump(samples_dict, f, ensure_ascii=False, indent=2)
print(f"评估数据集已保存: {filepath} ({len(self.samples)} 条)")
@classmethod
def load(cls, filepath: str) -> "EvalDatasetBuilder":
"""从 JSON 文件加载评估数据集"""
with open(filepath, "r", encoding="utf-8") as f:
samples_dict = json.load(f)
builder = cls()
builder.add_batch(samples_dict)
print(f"评估数据集已加载: {filepath} ({len(builder.samples)} 条)")
return builder
def get_statistics(self) -> dict:
"""获取数据集统计信息"""
stats = {
"total": len(self.samples),
"by_difficulty": {},
"by_category": {},
}
for s in self.samples:
stats["by_difficulty"][s.difficulty] = stats["by_difficulty"].get(s.difficulty, 0) + 1
stats["by_category"][s.category] = stats["by_category"].get(s.category, 0) + 1
return stats
def print_statistics(self):
"""打印数据集统计信息"""
stats = self.get_statistics()
print("=" * 50)
print("评估数据集统计")
print("=" * 50)
print(f"总样本数: {stats['total']}")
print(f"\n按难度分布:")
for diff, count in stats["by_difficulty"].items():
print(f" {diff}: {count} 条")
print(f"\n按类别分布:")
for cat, count in stats["by_category"].items():
print(f" {cat}: {count} 条")
# ============================================================
# 构建示例评估数据集
# ============================================================
if __name__ == "__main__":
builder = EvalDatasetBuilder()
# 添加评估样本
builder.add_sample(
question="公司年假政策是什么？",
ground_truth="员工入职满1年享有5天年假，满3年享有10天年假，满5年享有15天年假。",
difficulty="easy",
category="factual",
)
builder.add_sample(
question="如何申请差旅报销？",
ground_truth="差旅报销需在出差结束后7个工作日内提交报销申请，并附上所有票据。",
difficulty="easy",
category="factual",
)
builder.add_sample(
question="入职3年的员工比入职1年的员工多多少天年假？",
ground_truth="入职3年享有10天年假，入职1年享有5天年假，多5天。",
difficulty="medium",
category="reasoning",
)
builder.add_sample(
question="年假和调休在使用上有什么区别？",
ground_truth="年假是法定带薪假期，调休是因加班产生的补休。年假需提前一周申请，调休需在加班后一个月内使用。",
difficulty="hard",
category="comparison",
)
builder.add_sample(
question="公司为员工缴纳哪些社保？",
ground_truth="公司为所有正式员工缴纳五险一金，包括养老保险、医疗保险、失业保险、工伤保险、生育保险和住房公积金。试用期员工从入职当月开始缴纳。",
difficulty="easy",
category="factual",
)
builder.add_sample(
question="如果我在出差期间生病了，医疗费用怎么报销？",
ground_truth="出差期间因病就医的费用，可凭医院出具的诊断证明和费用清单，按公司医疗保险规定报销。需在返回公司后7个工作日内提交。",
difficulty="hard",
category="reasoning",
)
builder.print_statistics()
builder.save("./eval_datasets/company_policy_eval.json")
5. 评估流程自动化
5.1 为什么需要自动化
手动评估的问题是：
- 太慢：跑一次评估要手动执行 RAG 流程、收集结果、计算指标
- 容易出错：手动操作容易遗漏步骤
- 不可复现：每次手动操作的方式可能不同
- 无法持续：不可能每天都手动评估一次
自动化评估流水线的目标：一键运行，自动收集所有指标，生成评估报告。
5.2 自动化评估流水线设计
自动化评估流水线
┌─────────────────────────────────────────────────────────┐
│ 评估流水线 │
│ │
│ 1. 加载评估数据集 │
│ │ │
│ ▼ │
│ 2. 遍历每个评估样本 │
│ │ │
│ ├──→ RAG Pipeline → 生成 answer + contexts │
│ │ │
│ ▼ │
│ 3. 收集所有结果 │
│ │ │
│ ▼ │
│ 4. RAGAS 评估 → 计算所有指标 │
│ │ │
│ ▼ │
│ 5. 生成评估报告 │
│ ├── 整体指标汇总 │
│ ├── 按难度/类别分组 │
│ ├── 与历史基线对比 │
│ └── 问题样本标记 │
│ │
└─────────────────────────────────────────────────────────┘
5.3 核心代码
# eval_pipeline.py —— 自动化评估流水线
import json
import time
from typing import List, Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
context_relevancy,
answer_correctness,
answer_semantic_similarity,
)
@dataclass
class EvalReport:
"""评估报告"""
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
config: Dict = field(default_factory=dict)
overall_metrics: Dict = field(default_factory=dict)
by_difficulty: Dict = field(default_factory=dict)
by_category: Dict = field(default_factory=dict)
per_sample: List[Dict] = field(default_factory=list)
comparison_with_baseline: Optional[Dict] = None
low_score_samples: List[Dict] = field(default_factory=list)
class RAGEvalPipeline:
"""RAG 评估自动化流水线"""
def __init__(
self,
rag_pipeline, # RAG 流水线实例
eval_dataset_path: str, # 评估数据集路径
metrics: Optional[List] = None,
baseline_path: Optional[str] = None, # 历史基线报告路径
):
self.rag = rag_pipeline
self.eval_dataset_path = eval_dataset_path
self.baseline_path = baseline_path
# 默认评估指标
self.metrics = metrics or [
faithfulness,
answer_relevancy,
context_precision,
context_recall,
context_relevancy,
]
# 加载评估数据集
with open(eval_dataset_path, "r", encoding="utf-8") as f:
self.eval_samples = json.load(f)
print(f"评估流水线初始化完成: {len(self.eval_samples)} 条评估样本")
def run(self) -> EvalReport:
"""运行完整评估流水线"""
print("=" * 60)
print("RAG 评估流水线开始运行")
print("=" * 60)
report = EvalReport(
config={
"eval_dataset": self.eval_dataset_path,
"num_samples": len(self.eval_samples),
"metrics": [m.name for m in self.metrics],
}
)
# 第一步：遍历评估样本，运行 RAG
print("\n[1/4] 运行 RAG Pipeline...")
results = []
for i, sample in enumerate(self.eval_samples):
print(f" 处理样本 {i+1}/{len(self.eval_samples)}: {sample['question'][:50]}...")
try:
rag_result = self.rag.query(sample["question"])
results.append({
"question": sample["question"],
"answer": rag_result.get("answer", ""),
"contexts": rag_result.get("contexts", []),
"ground_truth": sample["ground_truth"],
"difficulty": sample.get("difficulty", "medium"),
"category": sample.get("category", "general"),
})
except Exception as e:
print(f" 错误: {e}")
results.append({
"question": sample["question"],
"answer": f"[ERROR: {str(e)}]",
"contexts": [],
"ground_truth": sample["ground_truth"],
"difficulty": sample.get("difficulty", "medium"),
"category": sample.get("category", "general"),
})
# 第二步：构建 RAGAS Dataset
print("\n[2/4] 构建评估数据集...")
ragas_data = {
"question": [r["question"] for r in results],
"answer": [r["answer"] for r in results],
"contexts": [r["contexts"] for r in results],
"ground_truth": [r["ground_truth"] for r in results],
}
dataset = Dataset.from_dict(ragas_data)
# 第三步：运行 RAGAS 评估
print("\n[3/4] 运行 RAGAS 评估...")
start_time = time.time()
eval_result = evaluate(dataset, metrics=self.metrics)
elapsed = time.time() - start_time
print(f" 评估完成，耗时: {elapsed:.2f} 秒")
# 收集整体指标
report.overall_metrics = {k: float(v) for k, v in eval_result.items()}
# 提取逐样本分数
per_sample_scores = []
try:
for sample in eval_result.scores:
per_sample_scores.append({k: float(v) for k, v in sample.items()})
except Exception:
for _ in range(len(results)):
per_sample_scores.append(dict(report.overall_metrics))
# 第四步：分组统计
print("\n[4/4] 生成评估报告...")
self._compute_group_metrics(report, results, per_sample_scores)
# 标记低分样本
self._flag_low_score_samples(report, results, per_sample_scores)
# 与基线对比
if self.baseline_path:
report.comparison_with_baseline = self._compare_with_baseline(report)
print("\n评估流水线运行完成!")
return report
def _compute_group_metrics(self, report: EvalReport, results: List[Dict], per_sample_scores: List[Dict]):
"""按难度和类别分组统计"""
# 按难度分组
difficulty_groups = {}
for i, r in enumerate(results):
diff = r.get("difficulty", "medium")
if diff not in difficulty_groups:
difficulty_groups[diff] = []
difficulty_groups[diff].append(i)
for diff, indices in difficulty_groups.items():
group_metrics = {}
metric_names = list(per_sample_scores[0].keys()) if per_sample_scores else []
for metric_name in metric_names:
scores = [per_sample_scores[i].get(metric_name, 0) for i in indices]
group_metrics[metric_name] = sum(scores) / len(scores) if scores else 0
report.by_difficulty[diff] = group_metrics
# 按类别分组
category_groups = {}
for i, r in enumerate(results):
cat = r.get("category", "general")
if cat not in category_groups:
category_groups[cat] = []
category_groups[cat].append(i)
for cat, indices in category_groups.items():
group_metrics = {}
metric_names = list(per_sample_scores[0].keys()) if per_sample_scores else []
for metric_name in metric_names:
scores = [per_sample_scores[i].get(metric_name, 0) for i in indices]
group_metrics[metric_name] = sum(scores) / len(scores) if scores else 0
report.by_category[cat] = group_metrics
# 逐样本记录
for i, r in enumerate(results):
sample_metrics = per_sample_scores[i] if i < len(per_sample_scores) else {}
report.per_sample.append({
"question": r["question"],
"answer": r["answer"],
"metrics": sample_metrics,
"difficulty": r.get("difficulty", "medium"),
"category": r.get("category", "general"),
})
def _flag_low_score_samples(self, report: EvalReport, results: List[Dict], per_sample_scores: List[Dict]):
"""标记低分样本（Faithfulness < 0.5 或 Answer Relevancy < 0.5）"""
for i, r in enumerate(results):
if i >= len(per_sample_scores):
continue
sample_scores = per_sample_scores[i]
is_low = False
reasons = []
for metric_name, score in sample_scores.items():
if metric_name == "faithfulness" and score < 0.5:
is_low = True
reasons.append(f"Faithfulness={score:.2f}")
elif metric_name == "answer_relevancy" and score < 0.5:
is_low = True
reasons.append(f"AnswerRelevancy={score:.2f}")
if is_low:
report.low_score_samples.append({
"question": r["question"],
"answer": r["answer"][:200],
"reasons": reasons,
"difficulty": r.get("difficulty", "medium"),
})
def _compare_with_baseline(self, report: EvalReport) -> Dict:
"""与历史基线对比"""
try:
with open(self.baseline_path, "r", encoding="utf-8") as f:
baseline = json.load(f)
except FileNotFoundError:
return {"error": f"基线文件不存在: {self.baseline_path}"}
comparison = {}
baseline_metrics = baseline.get("overall_metrics", {})
for metric, current_score in report.overall_metrics.items():
baseline_score = baseline_metrics.get(metric)
if baseline_score is not None:
delta = current_score - baseline_score
comparison[metric] = {
"current": round(current_score, 4),
"baseline": round(baseline_score, 4),
"delta": round(delta, 4),
"trend": "improved" if delta > 0.01 else ("declined" if delta < -0.01 else "stable"),
}
return comparison
def save_report(self, report: EvalReport, filepath: str):
"""保存评估报告"""
report_dict = {
"timestamp": report.timestamp,
"config": report.config,
"overall_metrics": report.overall_metrics,
"by_difficulty": report.by_difficulty,
"by_category": report.by_category,
"low_score_samples": report.low_score_samples,
"comparison_with_baseline": report.comparison_with_baseline,
}
with open(filepath, "w", encoding="utf-8") as f:
json.dump(report_dict, f, ensure_ascii=False, indent=2)
print(f"评估报告已保存: {filepath}")
def print_report(self, report: EvalReport):
"""打印评估报告摘要"""
print("\n" + "=" * 60)
print("RAG 评估报告")
print("=" * 60)
print(f"评估时间: {report.timestamp}")
print(f"评估样本数: {report.config.get('num_samples', 'N/A')}")
print(f"\n--- 整体指标 ---")
for metric, score in report.overall_metrics.items():
bar = self._score_bar(score)
print(f" {metric:<25s}: {score:.4f} {bar}")
if report.by_difficulty:
print(f"\n--- 按难度分组 ---")
for diff, metrics in report.by_difficulty.items():
print(f" [{diff}]")
for m, s in metrics.items():
print(f" {m}: {s:.4f}")
if report.comparison_with_baseline:
print(f"\n--- 与基线对比 ---")
for metric, comp in report.comparison_with_baseline.items():
delta_str = f"+{comp['delta']:.4f}" if comp['delta'] > 0 else f"{comp['delta']:.4f}"
trend_icon = "↑" if comp['trend'] == 'improved' else ("↓" if comp['trend'] == 'declined' else "→")
print(f" {metric}: {comp['baseline']:.4f} → {comp['current']:.4f} ({delta_str}) {trend_icon}")
if report.low_score_samples:
print(f"\n--- 低分样本 ({len(report.low_score_samples)} 条) ---")
for i, sample in enumerate(report.low_score_samples[:5]):
print(f" {i+1}. {sample['question'][:50]}...")
print(f" 原因: {', '.join(sample['reasons'])}")
@staticmethod
def _score_bar(score: float, width: int = 20) -> str:
"""生成分数可视化条"""
filled = int(score * width)
return "[" + "#" * filled + "-" * (width - filled) + "]"
6. 持续监控方案
6.1 为什么需要持续监控
一次性的评估只能告诉你"当前系统怎么样"，但无法回答：
- 换了 Embedding 模型之后，质量有没有下降？
- 新增了一批文档之后，检索效果有没有变差？
- 用户问的问题类型变了，系统还能不能 hold 住？
持续监控的目标：让 RAG 系统的质量变化可追踪、可回溯、可预警。
6.2 监控架构设计
持续监控架构
┌─────────────────────────────────────────────────────────┐
│ 监控数据流 │
│ │
│ 生产环境 监控系统 │
│ ┌──────────┐ ┌──────────┐ │
│ │ RAG │──→ 日志记录 ──→ │ 指标计算 │ │
│ │ Pipeline │ (每次查询) │ │ │
│ └──────────┘ │ - 延迟 │ │
│ │ - Token │ │
│ 定时评估 │ - 检索数 │ │
│ ┌──────────┐ │ │ │
│ │ 评估流水线│──→ 评估报告 ──→ ├──────────┤ │
│ │ (每天/周) │ │ 质量趋势 │ │
│ └──────────┘ │ │ │
│ │ - 各指标 │ │
│ 异常检测 │ 变化 │ │
│ ┌──────────┐ │ │ │
│ │ 阈值告警 │←── 指标异常 ──← ├──────────┤ │
│ └──────────┘ │ 告警通知 │ │
│ └──────────┘ │
└─────────────────────────────────────────────────────────┘
6.3 监控指标设计
# monitor.py —— RAG 系统持续监控
import json
import time
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from dataclasses import dataclass, field
from collections import defaultdict
@dataclass
class QueryLog:
"""单次查询日志"""
timestamp: str
question: str
answer: str
contexts: List[str]
latency_ms: float
token_usage: Dict
retrieval_count: int
user_feedback: Optional[str] = None # positive/negative/neutral
@dataclass
class AlertRule:
"""告警规则"""
metric: str
threshold: float
direction: str # "above" 或 "below"
severity: str = "warning" # warning/critical
class RAGMonitor:
"""RAG 系统持续监控器"""
def __init__(self, alert_rules: Optional[List[AlertRule]] = None):
self.query_logs: List[QueryLog] = []
self.eval_reports: List[Dict] = []
self.alerts: List[Dict] = []
# 默认告警规则
self.alert_rules = alert_rules or [
AlertRule(metric="faithfulness", threshold=0.7, direction="below", severity="critical"),
AlertRule(metric="answer_relevancy", threshold=0.7, direction="below", severity="warning"),
AlertRule(metric="context_precision", threshold=0.5, direction="below", severity="warning"),
AlertRule(metric="avg_latency_ms", threshold=5000, direction="above", severity="warning"),
]
def log_query(self, question: str, answer: str, contexts: List[str],
latency_ms: float, token_usage: Dict, retrieval_count: int,
user_feedback: Optional[str] = None):
"""记录一次查询"""
log = QueryLog(
timestamp=datetime.now().isoformat(),
question=question,
answer=answer,
contexts=contexts,
latency_ms=latency_ms,
token_usage=token_usage,
retrieval_count=retrieval_count,
user_feedback=user_feedback,
)
self.query_logs.append(log)
def add_eval_report(self, report: Dict):
"""添加一次评估报告"""
self.eval_reports.append({
"timestamp": datetime.now().isoformat(),
"metrics": report.get("overall_metrics", {}),
})
# 检查告警规则
self._check_alerts(report)
def _check_alerts(self, report: Dict):
"""检查是否触发告警"""
metrics = report.get("overall_metrics", {})
for rule in self.alert_rules:
current_value = metrics.get(rule.metric)
if current_value is None:
continue
triggered = False
if rule.direction == "below" and current_value < rule.threshold:
triggered = True
elif rule.direction == "above" and current_value > rule.threshold:
triggered = True
if triggered:
alert = {
"timestamp": datetime.now().isoformat(),
"metric": rule.metric,
"current_value": current_value,
"threshold": rule.threshold,
"direction": rule.direction,
"severity": rule.severity,
"message": f"[{rule.severity.upper()}] {rule.metric}={current_value:.4f} "
f"(阈值: {rule.direction} {rule.threshold})",
}
self.alerts.append(alert)
print(f" [告警] {alert['message']}")
def get_metrics_summary(self, window_hours: int = 24) -> Dict:
"""获取最近 N 小时的指标摘要"""
cutoff = datetime.now() - timedelta(hours=window_hours)
recent_logs = [
log for log in self.query_logs
if datetime.fromisoformat(log.timestamp) > cutoff
]
if not recent_logs:
return {"message": f"过去 {window_hours} 小时内无查询记录"}
latencies = [log.latency_ms for log in recent_logs]
total_tokens = sum(
log.token_usage.get("total_tokens", 0) for log in recent_logs
)
positive_feedback = sum(
1 for log in recent_logs if log.user_feedback == "positive"
)
negative_feedback = sum(
1 for log in recent_logs if log.user_feedback == "negative"
)
return {
"window_hours": window_hours,
"total_queries": len(recent_logs),
"avg_latency_ms": sum(latencies) / len(latencies),
"p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)],
"p99_latency_ms": sorted(latencies)[int(len(latencies) * 0.99)],
"total_tokens": total_tokens,
"avg_tokens_per_query": total_tokens / len(recent_logs),
"positive_feedback_rate": positive_feedback / len(recent_logs) if recent_logs else 0,
"negative_feedback_rate": negative_feedback / len(recent_logs) if recent_logs else 0,
"active_alerts": len([a for a in self.alerts if a["severity"] == "critical"]),
}
def get_quality_trend(self) -> Dict:
"""获取质量趋势（基于历史评估报告）"""
if len(self.eval_reports) < 2:
return {"message": "评估报告不足2份，无法计算趋势"}
trend = {}
latest = self.eval_reports[-1]["metrics"]
previous = self.eval_reports[-2]["metrics"]
for metric in latest:
if metric in previous:
delta = latest[metric] - previous[metric]
trend[metric] = {
"latest": round(latest[metric], 4),
"previous": round(previous[metric], 4),
"delta": round(delta, 4),
"direction": "up" if delta > 0 else "down",
}
return trend
def print_dashboard(self):
"""打印监控仪表盘"""
summary = self.get_metrics_summary(window_hours=24)
trend = self.get_quality_trend()
print("\n" + "=" * 60)
print("RAG 系统监控仪表盘")
print("=" * 60)
print(f"\n--- 运行指标 (过去24小时) ---")
for key, value in summary.items():
if isinstance(value, float):
print(f" {key}: {value:.2f}")
else:
print(f" {key}: {value}")
if trend and "message" not in trend:
print(f"\n--- 质量趋势 ---")
for metric, data in trend.items():
direction = "↑" if data["direction"] == "up" else "↓"
print(f" {metric}: {data['previous']:.4f} → {data['latest']:.4f} "
f"({data['delta']:+.4f}) {direction}")
if self.alerts:
print(f"\n--- 活跃告警 ({len(self.alerts)} 条) ---")
for alert in self.alerts[-5:]:
print(f" [{alert['severity']}] {alert['message']}")
7. 实战：完整评估流水线
7.1 端到端示例
# run_evaluation.py —— 完整评估流水线运行示例
"""
RAG 系统评估流水线 —— 端到端运行脚本
使用方法:
python run_evaluation.py --eval-dataset ./eval_datasets/company_policy_eval.json
流程:
1. 加载评估数据集
2. 初始化 RAG Pipeline
3. 遍历评估样本，收集结果
4. 运行 RAGAS 评估
5. 生成评估报告
6. 与历史基线对比
7. 输出报告摘要
"""
import argparse
import os
from datetime import datetime
def main():
parser = argparse.ArgumentParser(description="RAG 系统评估流水线")
parser.add_argument(
"--eval-dataset",
type=str,
default="./eval_datasets/company_policy_eval.json",
help="评估数据集 JSON 文件路径",
)
parser.add_argument(
"--baseline",
type=str,
default=None,
help="历史基线报告路径（用于对比）",
)
parser.add_argument(
"--output-dir",
type=str,
default="./eval_reports",
help="评估报告输出目录",
)
parser.add_argument(
"--rag-config",
type=str,
default="./config.py",
help="RAG 系统配置文件路径",
)
args = parser.parse_args()
# 确保输出目录存在
os.makedirs(args.output_dir, exist_ok=True)
print("=" * 60)
print("RAG 系统评估流水线")
print("=" * 60)
print(f"评估数据集: {args.eval_dataset}")
print(f"基线报告: {args.baseline or '无'}")
print(f"输出目录: {args.output_dir}")
# 初始化 RAG Pipeline
from rag_pipeline import RAGPipeline
rag = RAGPipeline.from_config(args.rag_config)
# 初始化评估流水线
pipeline = RAGEvalPipeline(
rag_pipeline=rag,
eval_dataset_path=args.eval_dataset,
baseline_path=args.baseline,
)
# 运行评估
report = pipeline.run()
# 保存报告
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
report_path = os.path.join(args.output_dir, f"eval_report_{timestamp}.json")
pipeline.save_report(report, report_path)
# 打印报告摘要
pipeline.print_report(report)
# 更新监控器
from monitor import RAGMonitor
monitor = RAGMonitor()
monitor.add_eval_report({
"overall_metrics": report.overall_metrics,
})
monitor.print_dashboard()
print(f"\n完整报告已保存至: {report_path}")
if __name__ == "__main__":
main()
8. 常见问题与避坑指南
8.1 评估成本问题
RAGAS 的评估依赖 LLM 来判断——Faithfulness 需要 LLM 判断声明是否可推导，Answer Relevancy 需要 LLM 反向生成问题。这意味着评估本身也在消耗 Token，而且消耗量比你想象的要大。
以 50 条评估样本、5 个指标（Faithfulness + Answer Relevancy + Context Precision + Context Recall + Context Relevancy）为例，每个指标内部会触发多次 LLM 调用：
- Faithfulness：先将答案拆解为多个"声明"，再逐条判断每个声明是否能从上下文中推导出来。每条样本约 3-8 次 LLM 调用。
- Answer Relevancy：根据答案反向生成 3-5 个问题，再计算语义相似度。每条样本约 3-5 次 LLM 调用。
- Context Precision：逐条判断每个检索文档是否与问题相关。每条样本约 K 次 LLM 调用（K 为检索文档数）。
- Context Recall：将 ground truth 拆解为句子，逐句判断是否能从检索上下文中找到依据。
- Context Relevancy：将检索上下文拆解为句子，逐句判断是否与问题相关。
综合来看，一次完整评估大约消耗 200万-500万 tokens（取决于答案长度和检索文档数），而不是直觉上的 50万-100万。具体成本取决于使用的模型：
| 模型 | 输入价格（$/1M tokens） | 输出价格（$/1M tokens） | 50样本×5指标估算成本 |
|---|---|---|---|
| gpt-4o | $2.50 | $10.00 | $8-20 |
| gpt-4o-mini | $0.15 | $0.60 | $0.5-1.5 |
| deepseek-chat | ¥1.00 | ¥2.00 | ¥3-8 |
省钱建议：
- 日常评估用 gpt-4o-mini 或 deepseek-chat，成本极低且判断准确度足够
- 只在版本发布前的最终评估使用 gpt-4o，确保最高准确度
- 日常监控只跑核心指标（Faithfulness + Answer Relevancy），完整指标每周跑一次
- 评估样本控制在 30-50 条即可获得统计显著性，不需要上百条
- 如果使用本地部署的 LLM（如 vLLM 部署的 Qwen 系列），评估成本几乎为零
8.2 评估结果的解读
一个常见的误区是只看整体分数。比如 Faithfulness=0.85，看起来不错，但可能掩盖了问题：
整体 Faithfulness=0.85
按难度分组：
easy: 0.95 ← 简单问题很好
medium: 0.82 ← 中等难度还行
hard: 0.55 ← 复杂问题很差！
结论：系统在处理复杂推理问题时存在严重幻觉
正确做法：一定要按难度、类别分组看，找出系统的薄弱环节。
8.3 ground truth 的编写
Context Recall 和 Answer Correctness 需要 ground truth。编写 ground truth 时注意：
- 不要太简略：“5天” 不如 “入职满1年享有5天年假”
- 包含关键数字和实体：这些是 Context Entity Recall 的评估依据
- 覆盖答案的所有要点：如果答案应该包含 3 个要点，ground truth 也要包含 3 个
8.4 评估频率建议
| 场景 | 评估频率 | 指标范围 |
|---|---|---|
| 开发调试 | 每次改动后 | 全部指标 |
| 日常监控 | 每天 1 次 | Faithfulness + Answer Relevancy |
| 周报 | 每周 1 次 | 全部指标 + 趋势分析 |
| 版本发布 | 发布前 | 全部指标 + 与上一版本对比 |
8.5 不要过度依赖自动化评估
RAGAS 虽然强大，但它毕竟是"用 LLM 评估 LLM"——LLM 的判断也可能出错。建议：
- 定期人工抽查 10-20% 的评估样本，验证 LLM 判断的准确性
- 关注用户反馈（点赞/点踩），作为评估的补充信号
- 对于关键业务场景（法律、医疗），人工评估不可替代
9. 总结与学习建议
9.1 Phase 4 回顾
这篇文章是 Phase 4：RAG 系统构建的最后一篇。回顾整个 Phase 4 的学习历程：
Phase 4: RAG 系统构建 —— 完整知识体系
文档分块策略 → 怎么把文档切成合适的块
向量数据库选型 → 文档块存到哪里、怎么检索
检索优化策略 → 怎么检索才能更准
评估体系 → 怎么量化检索和生成的质量
四篇文章形成了一个完整的闭环：切分 → 存储 → 检索 → 评估。每一篇都解决一个具体的问题，合在一起就是一个生产级 RAG 系统的完整知识体系。
9.2 核心收获
通过这篇文章的学习，我掌握了：
- RAGAS 八大指标的含义、计算原理和使用场景
- 评估数据集的构建方法和质量标准
- 评估流水线自动化的设计和实现
- 持续监控的架构设计和告警机制
- 评估结果的解读方法——不只看整体分数，要看分组、看趋势
9.3 学习建议
如果你也在学习 RAG 评估，我的建议是：
- 先理解指标含义：不要急着跑代码，先搞清楚每个指标衡量什么、为什么重要
- 从小数据集开始：先准备 10-20 条评估样本，跑通流程，再逐步扩充
- 建立基线：第一次评估的结果就是你的基线，后续所有优化都要和基线对比
- 关注低分样本：不要只看整体分数，低分样本才是系统真正的短板
- 评估和优化形成闭环：评估 → 发现问题 → 优化 → 再评估 → 确认改进
本文完成于 2026-05-13，是我学习 RAG 评估体系过程中的总结与梳理。
脑启社区是一个专注类脑智能领域的开发者社区。欢迎加入社区，共建类脑智能生态。社区为开发者提供了丰富的开源类脑工具软件、类脑算法模型及数据集、类脑知识库、类脑技术培训课程以及类脑应用案例等资源。
更多推荐
7
0- 0
已为社区贡献1条内容
所有评论(0)