知识库评估体系：怎么知道你的RAG系统好不好
这篇文章解决什么问题
RAG系统搭好了，怎么知道它效果好不好？用户说”不好用”，具体哪里不好？回答错了，是检索的问题还是生成的问题？这篇文章教你建立一套完整的评估体系，用数据说话。
前言：评估是个技术活
先说个让我头疼的问题。
搭好一个RAG系统后，老板问：“效果怎么样？”
我说：“挺好的。” 老板问：“怎么证明？” 我说：“我测试了几个case，感觉还行。” 老板说：“几个case能说明什么？”
…当场社死。
后来我学会了系统化评估，用数据说话，老板终于不问了（可能是因为数据太难看）。
一、为什么需要评估
1.1 评估的价值
| 阶段 | 没有评估 | 有评估 |
|---|---|---|
| 开发时 | 不知道改了什么 | 量化改进效果 |
| 上线后 | 用户说不好用但不知道哪里不好 | 精确定位问题 |
| 迭代时 | 瞎试，浪费时间 | 数据驱动决策 |
1.2 评估的三个维度
RAG系统有三个核心环节：
检索 → 重排 → 生成
↓ ↓ ↓
检索 排序 生成
质量 质量 质量
每个环节都需要评估：
- 检索评估：找到的内容对不对？
- 重排评估：排序是否合理？
- 生成评估：回答质量怎么样？
1.3 评估的难点
难点1：什么是”好”
- 检索到的文档”相关”怎么定义？
- 生成的答案”正确”怎么判断？
难点2：标注成本高
- 需要人工标注大量测试数据
- 不同人标注可能有分歧
难点3：多维度评估
- 精确率、召回率、答案质量…
- 多个指标可能相互矛盾
二、检索评估指标
2.1 Precision@K
定义：前K个结果中有多少是相关的
Precision@5 = 相关文档数 / 5
def precision_at_k(retrieved_docs, relevant_docs, k):
"""
计算Precision@K
参数：
- retrieved_docs: 检索到的文档列表
- relevant_docs: 真正相关的文档列表
- k: 取前k个结果
"""
retrieved_k = retrieved_docs[:k]
relevant_set = set(relevant_docs)
num_relevant = len([d for d in retrieved_k if d in relevant_set])
return num_relevant / k
# 示例
retrieved = ["doc1", "doc2", "doc3", "doc4", "doc5"]
relevant = ["doc1", "doc3", "doc6"]
precision_5 = precision_at_k(retrieved, relevant, 5)
print(f"Precision@5: {precision_5}") # 0.4 (5个里2个相关)2.2 Recall@K
定义：所有相关文档中，有多少被找出来了
Recall@5 = 找到的相关文档数 / 所有相关文档数
def recall_at_k(retrieved_docs, relevant_docs, k):
"""
计算Recall@K
"""
retrieved_k = set(retrieved_docs[:k])
relevant_set = set(relevant_docs)
num_found = len(retrieved_k & relevant_set)
total_relevant = len(relevant_set)
if total_relevant == 0:
return 0.0
return num_found / total_relevant
# 示例
recall_5 = recall_at_k(retrieved, relevant, 5)
print(f"Recall@5: {recall_5}") # 0.667 (3个相关里找到2个)2.3 MRR（Mean Reciprocal Rank）
定义：第一个相关文档出现位置的倒数，取平均
def mean_reciprocal_rank(retrieved_docs, relevant_docs):
"""
计算MRR
"""
for i, doc in enumerate(retrieved_docs):
if doc in relevant_docs:
return 1.0 / (i + 1) # 位置从1开始
return 0.0
# 示例
retrieved = ["doc1", "doc4", "doc2", "doc3", "doc5"]
relevant = ["doc2"]
mrr = mean_reciprocal_rank(retrieved, relevant)
print(f"MRR: {mrr}") # 0.333 (doc2在第3位)2.4 NDCG（Normalized Discounted Cumulative Gain）
定义：考虑相关度等级的排序质量指标
def dcg_at_k(relevance_scores, k):
"""
计算DCG@K
relevance_scores: 相关度评分列表（按检索结果排序）
"""
dcg = 0.0
for i, rel in enumerate(relevance_scores[:k]):
dcg += rel / math.log2(i + 2) # i+2因为位置从1开始，log2(1)=0
return dcg
def ndcg_at_k(retrieved_docs, relevance_labels, k):
"""
计算NDCG@K
relevance_labels: 每个文档的相关度（0-5分）
"""
# 计算DCG
dcg = dcg_at_k(relevance_labels, k)
# 计算IDCG（理想排序）
ideal_labels = sorted(relevance_labels, reverse=True)
idcg = dcg_at_k(ideal_labels, k)
# NDCG = DCG / IDCG
if idcg == 0:
return 0.0
return dcg / idcg
# 示例
relevance = [3, 1, 4, 0, 2] # doc1:3分, doc2:1分, doc3:4分...
ndcg_5 = ndcg_at_k(None, relevance, 5)
print(f"NDCG@5: {ndcg_5:.4f}")2.5 MAP（Mean Average Precision）
定义：每个查询的平均精确率的均值
def average_precision(retrieved_docs, relevant_docs):
"""
计算单个查询的AP
"""
num_relevant = 0
sum_precision = 0.0
for i, doc in enumerate(retrieved_docs):
if doc in relevant_docs:
num_relevant += 1
precision_at_i = num_relevant / (i + 1)
sum_precision += precision_at_i
if num_relevant == 0:
return 0.0
return sum_precision / num_relevant
def mean_average_precision(queries_results):
"""
计算MAP
queries_results: [{"retrieved": [...], "relevant": [...]}, ...]
"""
aps = []
for result in queries_results:
ap = average_precision(result["retrieved"], result["relevant"])
aps.append(ap)
return sum(aps) / len(aps)
# 示例
queries = [
{"retrieved": ["d1", "d2", "d3"], "relevant": ["d1", "d4"]},
{"retrieved": ["d5", "d1", "d2"], "relevant": ["d1", "d2"]},
]
map_score = mean_average_precision(queries)
print(f"MAP: {map_score:.4f}")三、生成评估指标
3.1 BLEU
定义：基于n-gram重叠的评估指标，主要用于机器翻译
from collections import Counter
import math
def bleu_score(candidate, reference, max_n=4):
"""
简化的BLEU分数计算
"""
candidate_tokens = candidate.split()
reference_tokens = reference.split()
# 惩罚短句
brevity_penalty = min(1.0, len(candidate_tokens) / len(reference_tokens))
if len(candidate_tokens) < len(reference_tokens):
bp = math.exp(1 - len(reference_tokens) / len(candidate_tokens))
# 计算各阶n-gram精确率
precisions = []
for n in range(1, max_n + 1):
c_ngrams = Counter([tuple(candidate_tokens[i:i+n]) for i in range(len(candidate_tokens)-n+1)])
r_ngrams = Counter([tuple(reference_tokens[i:i+n]) for i in range(len(reference_tokens)-n+1)])
matches = sum((c_ngrams & r_ngrams).values())
total = sum(c_ngrams.values())
if total == 0:
precisions.append(0)
else:
precisions.append(matches / total)
# 几何平均
if precisions:
geo_mean = math.exp(sum(math.log(p) if p > 0 else 0 for p in precisions) / len(precisions))
else:
geo_mean = 0
return brevity_penalty * geo_mean
# 示例
candidate = "the cat sat on the mat"
reference = "the cat is on the mat"
bleu = bleu_score(candidate, reference)
print(f"BLEU: {bleu:.4f}")3.2 ROUGE
定义：基于召回率的评估指标，主要用于文本摘要
def rouge_l(candidate, reference):
"""
ROUGE-L：最长公共子序列
"""
candidate_tokens = candidate.split()
reference_tokens = reference.split()
# LCS长度
m, n = len(candidate_tokens), len(reference_tokens)
dp = [[0] * (n + 1) for _ in range(m + 1)]
for i in range(1, m + 1):
for j in range(1, n + 1):
if candidate_tokens[i-1] == reference_tokens[j-1]:
dp[i][j] = dp[i-1][j-1] + 1
else:
dp[i][j] = max(dp[i-1][j], dp[i][j-1])
lcs_length = dp[m][n]
# ROUGE-L = LCS / |reference|
recall = lcs_length / n if n > 0 else 0
precision = lcs_length / m if m > 0 else 0
f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0
return {"precision": precision, "recall": recall, "f1": f1}
# 示例
candidate = "the cat sat on the mat"
reference = "the cat is on the mat"
rouge = rouge_l(candidate, reference)
print(f"ROUGE-L F1: {rouge['f1']:.4f}")3.3 答案相关性（Answer Relevance）
定义：生成的答案和问题相关程度
async def answer_relevance(question, answer, llm):
"""
评估答案相关性
用LLM打分（1-5分）
"""
prompt = f"""
请评估以下问答的质量：
问题：{question}
答案：{answer}
请从相关性角度评分（1-5分）：
1分：答案和问题完全不相关
2分：答案和问题部分相关，但偏离主题
3分：答案和问题相关，但不完整
4分：答案和问题相关，内容完整
5分：答案和问题高度相关，完全切题
请只返回数字分数：
"""
score_text = await llm.generate(prompt)
try:
score = int(score_text.strip())
return score / 5.0 # 归一化到0-1
except:
return 0.0
# 使用
score = await answer_relevance(
"什么是机器学习？",
"机器学习是人工智能的一个分支，让计算机从数据中学习。",
llm
)
print(f"答案相关性: {score:.2f}")3.4 答案 faithfulness（忠实度）
定义：答案是否忠实于检索到的上下文
async def faithfulness(context, answer, llm):
"""
评估答案对上下文的忠实度
"""
prompt = f"""
请评估以下答案是否忠实于给定的上下文：
上下文：{context}
答案：{answer}
请检查：
1. 答案中的事实是否都能在上下文中找到依据？
2. 答案是否有添加上下文中没有的信息？
评分标准（1-5分）：
1分：答案严重偏离上下文，包含大量错误信息
2分：答案部分偏离上下文，包含一些错误信息
3分：答案基本忠实，但有少量偏差
4分：答案完全忠实于上下文
5分：答案完全忠实，且充分利用了上下文信息
请只返回数字分数：
"""
score_text = await llm.generate(prompt)
try:
score = int(score_text.strip())
return score / 5.0
except:
return 0.0四、RAGAS评估框架
4.1 RAGAS是什么
RAGAS（Retrieval-Augmented Generation Assessment）是一个专门用于评估RAG系统的框架：
- Context Precision：检索质量
- Answer Faithfulness：忠实度
- Answer Relevance：相关性
4.2 安装和使用
# 安装
# pip install ragas
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
)
# 准备测试数据
test_data = {
"user_input": [
"什么是机器学习？",
"深度学习和机器学习有什么区别？",
],
"retrieved_contexts": [
["机器学习是AI的一个分支...", "ML让计算机从数据中学习..."],
["深度学习使用神经网络...", "深度学习是ML的子领域..."],
],
"response": [
"机器学习是人工智能的一个分支...",
"深度学习是机器学习的子领域，使用神经网络...",
],
"reference": [
"机器学习是AI的一个分支，使计算机具有学习能力",
"深度学习是ML的子领域，以神经网络为基础",
]
}
# 评估
result = evaluate(test_data, metrics=[faithfulness, answer_relevancy, context_precision])
print(result)4.3 自定义RAGAS指标
from ragas.metrics.base import MetricWithLLM
from pydantic import Field
class CustomFaithfulness(MetricWithLLM):
"""
自定义的忠实度指标
"""
name = "custom_faithfulness"
async def _score(self, row):
context = row["retrieved_contexts"]
answer = row["response"]
# 用LLM评估
prompt = f"评估答案是否忠实于上下文..."
score = await self.llm.generate(prompt)
return float(score)
# 使用自定义指标
custom_metric = CustomFaithfulness(llm=your_llm)
result = evaluate(test_data, metrics=[custom_metric])五、TruLens评估框架
5.1 TruLens是什么
TruLens是另一个RAG评估框架，特别擅长：
- 过程追踪
- 中间结果评估
- 可视化分析
5.2 安装和使用
# 安装
# pip install trulens
from trulens_eval import TruChain, Feedback, Select
from trulens_eval.feedback import Groundedness
# 定义评估函数
feedback_functions = [
Feedback(
provider.contextual_relevance,
name="Context Relevance"
),
Feedback(
provider.answer_relevance,
question=Select.RecordInput,
response=Select.RecordOutput
),
Feedback(
groundedness.groundedness,
name="Groundedness"
)
]
# 添加到Chain
tru = TruChain(
chain=your_rag_chain,
app_id="my_rag_app",
feedbacks=feedback_functions
)
# 运行评估
with tru:
response = your_rag_chain.invoke("什么是机器学习？")
# 查看评估结果
tru.get_leaderboard()5.3 评估结果分析
# 获取详细评估结果
records = tru.get_records(app_id="my_rag_app")
for record in records:
print(f"问题: {record['input']}")
print(f"答案: {record['output']}")
print(f"Context Relevance: {record['feedback_context_relevance']:.2f}")
print(f"Answer Relevance: {record['feedback_answer_relevance']:.2f}")
print(f"Groundedness: {record['feedback_groundedness']:.2f}")
print("---")六、端到端评估流程
6.1 评估数据集准备
class EvaluationDataset:
"""
评估数据集管理
"""
def __init__(self):
self.data = []
def add(self, question, ground_truth_context, ground_truth_answer, metadata=None):
"""添加评估样本"""
self.data.append({
"question": question,
"ground_truth_context": ground_truth_context,
"ground_truth_answer": ground_truth_answer,
"metadata": metadata or {}
})
def save(self, path):
"""保存评估数据集"""
with open(path, 'w') as f:
json.dump(self.data, f, ensure_ascii=False, indent=2)
def load(self, path):
"""加载评估数据集"""
with open(path, 'r') as f:
self.data = json.load(f)
def get(self):
"""获取数据集"""
return self.data
# 示例：创建评估数据集
dataset = EvaluationDataset()
dataset.add(
question="深度学习的三要素是什么？",
ground_truth_context=[
"深度学习的三要素是：数据、模型、算力。",
"数据是深度学习的基础...",
],
ground_truth_answer="深度学习的三要素是数据、模型和算力。"
)
dataset.add(
question="Transformer为什么比RNN好？",
ground_truth_context=[
"Transformer使用自注意力机制...",
"RNN存在梯度消失问题...",
],
ground_truth_answer="Transformer相比RNN的优势在于..."
)
dataset.save("eval_dataset.json")6.2 完整评估管道
class RAGEvaluator:
"""
RAG系统完整评估器
"""
def __init__(self, rag_system, eval_dataset):
self.rag = rag_system
self.dataset = eval_dataset.get()
async def evaluate_retrieval(self):
"""
评估检索质量
"""
results = []
for item in self.dataset:
# 执行检索
retrieved = await self.rag.retrieve(item["question"])
retrieved_ids = [doc.id for doc in retrieved]
# 评估指标
relevant = item["ground_truth_context"] # 简化处理
metrics = {
"question": item["question"],
"precision@5": precision_at_k(retrieved_ids, relevant, 5),
"recall@5": recall_at_k(retrieved_ids, relevant, 5),
"mrr": mean_reciprocal_rank(retrieved_ids, relevant),
}
results.append(metrics)
# 汇总
summary = self._summarize(results)
return summary
async def evaluate_generation(self):
"""
评估生成质量
"""
results = []
for item in self.dataset:
# 执行问答
response = await self.rag.answer(item["question"])
# 用RAGAS评估
eval_result = await evaluate_single(
question=item["question"],
answer=response,
context=item["ground_truth_context"],
reference=item["ground_truth_answer"]
)
results.append(eval_result)
summary = self._summarize(results)
return summary
async def evaluate_full(self):
"""
端到端评估
"""
retrieval_result = await self.evaluate_retrieval()
generation_result = await self.evaluate_generation()
return {
"retrieval": retrieval_result,
"generation": generation_result
}
def _summarize(self, results):
"""汇总结果"""
if not results:
return {}
summary = {}
for key in results[0].keys():
if key != "question":
values = [r[key] for r in results if key in r]
summary[key] = {
"mean": sum(values) / len(values),
"min": min(values),
"max": max(values)
}
return summary6.3 运行评估
async def run_evaluation():
"""
运行完整评估
"""
# 加载评估数据
dataset = EvaluationDataset()
dataset.load("eval_dataset.json")
# 初始化评估器
evaluator = RAGEvaluator(rag_system=your_rag, eval_dataset=dataset)
# 运行评估
print("开始评估检索质量...")
retrieval_metrics = await evaluator.evaluate_retrieval()
print("开始评估生成质量...")
generation_metrics = await evaluator.evaluate_generation()
# 打印结果
print("\n" + "="*50)
print("评估结果")
print("="*50)
print("\n【检索指标】")
for metric, values in retrieval_metrics.items():
print(f" {metric}: {values['mean']:.4f}")
print("\n【生成指标】")
for metric, values in generation_metrics.items():
print(f" {metric}: {values['mean']:.4f}")
# 运行
asyncio.run(run_evaluation())七、A/B测试
7.1 什么是A/B测试
A/B测试就是比较两个版本的系统，看哪个更好：
用户流量
↓
随机分配
↓
A组 → 系统A → 收集反馈
B组 → 系统B → 收集反馈
↓
比较两组结果
7.2 A/B测试实现
import random
from collections import defaultdict
class ABTest:
"""
A/B测试
"""
def __init__(self, variants):
"""
variants: {"A": 系统A, "B": 系统B}
"""
self.variants = variants
self.results = {name: [] for name in variants.keys()}
def get_variant(self, user_id):
"""
根据用户ID决定走哪个版本
"""
# 简单的哈希分配
hash_value = hash(user_id) % 100
if hash_value < 50:
return "A"
else:
return "B"
def record_result(self, variant, metrics):
"""记录结果"""
self.results[variant].append(metrics)
def analyze(self):
"""分析结果"""
analysis = {}
for variant, results in self.results.items():
if not results:
continue
# 计算各指标均值
metrics_summary = defaultdict(list)
for r in results:
for k, v in r.items():
metrics_summary[k].append(v)
analysis[variant] = {
k: sum(v) / len(v)
for k, v in metrics_summary.items()
}
return analysis
# 使用示例
ab_test = ABTest({"A": system_v1, "B": system_v2})
# 用户请求
user_id = "user_123"
variant = ab_test.get_variant(user_id)
# 调用对应版本
if variant == "A":
result = await system_v1.query(question)
else:
result = await system_v2.query(question)
# 记录反馈
ab_test.record_result(variant, {"relevance": 0.8, "latency": 0.5})
# 分析
analysis = ab_test.analyze()
print(analysis)八、评估最佳实践
8.1 评估数据质量
原则1：数据要真实
# ❌ 不好：人工编造的数据
test_cases = [
{"question": "什么是AI？", "answer": "AI是人工智能..."}
]
# ✅ 好：从真实用户日志中采样
test_cases = sample_from_real_logs(user_queries, n=100)原则2：覆盖多种场景
# 不同类型的问题都要覆盖
test_cases = [
# 事实类
{"question": "XXX是什么？", ...},
# 解释类
{"question": "为什么XXX？", ...},
# 操作类
{"question": "如何XXX？", ...},
# 对比类
{"question": "XXX和YYY有什么区别？", ...},
]原则3：定期更新
# 定期从生产环境采样新数据
new_cases = sample_from_production(days=7)
test_dataset.update(new_cases)8.2 评估指标选择
| 场景 | 推荐指标 |
|---|---|
| 检索优化 | Precision@K, Recall@K |
| 排序优化 | NDCG, MRR |
| 回答质量 | Faithfulness, Relevance |
| 整体评估 | RAGAS综合评分 |
8.3 评估频率
# 持续评估
class ContinuousEvaluator:
"""持续评估器"""
def __init__(self, rag_system, eval_dataset):
self.rag = rag_system
self.eval_dataset = eval_dataset
async def daily_evaluation(self):
"""每日评估"""
# 每天跑一次完整评估
results = await self.evaluate()
# 检查是否有退化
if results["score"] < self.baseline - 0.05:
self.alert("性能下降！")
return results
async def online_evaluation(self):
"""在线评估"""
# 实时采样用户请求
while True:
sample = await sample_user_request()
result = await self.rag.query(sample)
# 快速评估
quality = await quick_evaluate(sample, result)
self.log(quality)
await asyncio.sleep(60) # 每分钟采样一次九、常见问题
9.1 评估结果和用户反馈不一致
可能原因：
- 评估数据集不代表性
- 用户场景和测试场景不同
- 评估指标没有覆盖用户关心的点
解决方案：
- 重新审视评估数据集
- 增加用户调研
- 调整评估指标
9.2 指标都很高但用户说不好用
可能原因：
- 指标只衡量了部分质量
- 用户期望比”正确”更高
解决方案：
- 增加答案可读性评估
- 增加答案完整性评估
- 增加答案实用性评估
9.3 怎么确定评估阈值
建议：
# 参考值（可根据业务调整）
THRESHOLDS = {
"precision@5": 0.8, # 前5个至少80%相关
"recall@5": 0.9, # 90%的相关文档要找到
"faithfulness": 0.9, # 答案90%要忠实于上下文
"relevance": 0.7, # 答案70%要和问题相关
}十、总结
RAG评估的三个层面：
- 检索评估：Precision、Recall、MRR、NDCG
- 生成评估：Faithfulness、Relevance、BLEU
- 端到端评估：RAGAS、TruLens
评估的最佳实践：
- 数据质量第一：真实、多样、覆盖全
- 指标要匹配业务：不同场景用不同指标
- 持续评估：不只是上线前，上线后也要持续监控
记住：没有完美的评估，只有不断完善的评估。定期审视评估体系，让它和业务一起成长。
相关主题
更新记录
- 2026-04-24：改写完成，语言风格优化
- 增加RAGAS和TruLens框架详解