RAG Evaluation Frameworks
概述
RAG(Retrieval-Augmented Generation)系统的评估是确保生成答案质量的关键环节。与传统信息检索系统不同,RAG 需要同时评估检索质量和生成质量两个维度。检索质量决定模型能否获取正确的上下文,而生成质量决定模型能否基于上下文给出准确、相关且流畅的回答。
在 2026 年的生产环境中,RAG 评估已形成一套成熟的框架体系,主要包括 RAGAS、Trulens 等专用评估工具,以及 BLEU、ROUGE、MRR 等传统指标的扩展应用。这些框架相互补充,共同支撑 RAG 系统的质量保障。
RAGAS 评估框架
RAGAS(RAG Assessment)是目前最广泛使用的 RAG 系统专用评估框架,由 Exploding Gradients 团队开发。该框架从多个维度对 RAG 系统进行量化评估,核心指标包括 Faithfulness、Answer Relevancy、Context Precision 和 Context Recall。1
Faithfulness(忠实度)
Faithfulness 衡量答案是否忠实于检索到的上下文,避免模型引入外部知识或产生幻觉。其计算流程如下:
- 从答案中提取所有原子陈述(Atomic Statements)
- 判断每个陈述是否可归因于上下文
- 计算归因陈述的比例
低忠实度的典型表现:
| 表现类型 | 示例 |
|---|---|
| 信息补充 | 上下文提到「RAG 是检索增强生成」,答案补充「由 OpenAI 于 2023 年提出」 |
| 过度推理 | 上下文仅提到「公司营收增长」,答案推断「这意味着市场份额提升」 |
| 语义偏移 | 上下文说「效果有限」,答案理解为「效果显著」 |
提升方法:
- 在 Prompt 中明确约束:「仅基于以下上下文回答,不要添加外部知识」
- 使用引用标注(Citation),让模型明确标注每个陈述的来源
- 引入自检机制,让模型验证自己的回答是否与上下文一致
from ragas.metrics import faithfulness
from ragas import evaluate
from datasets import Dataset
eval_data = {
"user_input": ["什么是 RAG?"],
"retrieved_contexts": [["RAG = Retrieval Augmented Generation,是一种结合检索和生成的技术"]],
"response": ["RAG 即检索增强生成,它结合了信息检索与大语言模型生成能力。"],
"reference": ["RAG = Retrieval Augmented Generation"]
}
dataset = Dataset.from_dict(eval_data)
result = evaluate(dataset, metrics=[faithfulness])
print(result)Answer Relevancy(答案相关性)
Answer Relevancy 评估答案是否真正回答了用户问题,而非仅仅在表面上相关。RAGAS 采用反向问题生成的方式来评估该指标:
- 根据答案生成多个重新表述的问题
- 计算原始问题与生成问题的语义相似度
- 取最高相似度作为答案相关性得分
低相关性的典型场景:
- 答非所问:问题问原因,答案讲结果
- 信息冗余:包含大量与问题无关的细节
- 不完整:只回答了问题的部分子问题
from ragas.metrics import answer_relevancy
metric = answer_relevancy(
batch_size=4,
model="gpt-4" # 用于生成重述问题
)
result = metric.score(
user_input="如何优化 RAG 的检索效果?",
response="可以通过混合检索、查询扩展和重排序来优化检索效果。",
retrieved_contexts=[...]
)Context Precision(上下文精确度)
Context Precision 评估检索结果中相关文档的排名质量。理想情况下,所有相关文档应该排在最前面。计算公式:
其中 表示第 个文档是否相关(1 或 0)。
优化方向:
| 方法 | 说明 | 收益 |
|---|---|---|
| 改进 Embedding | 使用领域适配的 Embedding 模型 | MRR 提升 10-20% |
| 混合检索 | 结合向量检索与 BM25 | 召回率提升 15% |
| 重排序 | 使用 Cross-Encoder 进行二次排序 | Precision@5 提升 25% |
Context Recall(上下文召回率)
Context Recall 衡量检索系统是否找到了所有与问题相关的文档。该指标需要参考答案(Ground Truth)来判断哪些文档应该被检索到:
挑战:在实际应用中,很难枚举「所有相关文档」,通常采用人工标注或参考答案来近似。
Context Entity Recall(上下文实体召回率)
Context Entity Recall 是 RAGAS 新增的指标,专注于评估答案中引用的实体是否都能在上下文中找到:
from ragas.metrics import context_entity_recall
metric = context_entity_recall()
result = metric.score(
user_input="2024 年奥运会在哪里举办?",
response="2024 年奥运会在法国巴黎举办。",
retrieved_contexts=[["2024年夏季奥运会将在巴黎举办,这是巴黎第二次举办奥运会"]]
)Trulens 评估平台
Trulens 是由 TruEra 团队开发的高级评估平台,专注于 LLM 应用的可观测性和深度分析。相比 RAGAS,Trulens 提供更细粒度的反馈和更强大的溯源能力。2
核心架构
Trulens 的核心概念包括:
| 概念 | 说明 |
|---|---|
| TruCustomApp | 将任何 RAG 应用包装为可评估对象 |
| Feedback Function | 可定制的评估函数 |
| Trace | 完整的执行链路追踪 |
| Session | 会话级别的上下文管理 |
安装与基础用法
pip install trulens-evalfrom trulens_eval import TruCustomApp, Feedback
from trulens_eval.feedback import Groundedness, Relevance
from trulens_eval.feedback.provider.openai import OpenAI as TruOpenAI
import openai
# 初始化 Provider
openai_provider = TruOpenAI()
# 定义评估反馈
groundedness = Feedback(
openai_provider.groundedness_measure_with_cot_reasons,
name="Groundedness"
).on(context=..., response=...)
relevance = Feedback(
openai_provider.relevance,
name="Response Relevance"
).on(context=..., response=...).on_default()
# 创建 TruCustomApp
tru_app = TruCustomApp(
my_rag_app,
app_name="Production RAG",
app_version="v1.0",
feedbacks=[groundedness, relevance]
)
# 运行评估
with tru_app as recording:
response = my_rag_app.query("如何提高 RAG 的检索精度?")
# 获取评估结果
for feedback, result in recording.get_feedbacks():
print(f"{feedback.name}: {result.result}")内置反馈函数
Trulens 提供了丰富的内置反馈函数:
| 反馈函数 | 说明 | 适用场景 |
|---|---|---|
groundedness_measure_with_cot_reasons | 带推理过程的忠实度评估 | 详细诊断幻觉原因 |
relevance | 答案与问题的相关性 | 快速相关性检查 |
sentiment | 情感分析 | 客服对话评估 |
language_match | 语言一致性 | 多语言 RAG 系统 |
moderation | 内容安全检测 | 敏感内容过滤 |
高级用法:自定义反馈函数
from trulens_eval import Feedback
from trulens_eval.feedback.provider.base import Provider
class CustomProvider(Provider):
def __init__(self):
super().__init__()
def completeness_score(self, response: str, context: str) -> float:
"""评估答案的完整性"""
required_aspects = ["是什么", "为什么", "怎么做"]
covered = sum(1 for aspect in required_aspects
if aspect in response)
return covered / len(required_aspects)
custom_provider = CustomProvider()
completeness = Feedback(
custom_provider.completeness_score,
name="Answer Completeness"
).on(response=..., context=...)可视化与调试
Trulens 提供强大的可视化工具,用于分析评估结果:
from trulens_eval import TruSession
session = TruSession()
session.get_leaderboard()
# 获取特定会话的详细追踪
session.get_session(session_id="xxx").get_records_and_feedback(
app_id=0,
record_id=0
)RAG 评估关键指标详解
检索质量评估
检索质量是 RAG 系统的基础,决定了生成阶段的上限。以下是检索评估的核心指标。
Hit Rate(命中率)
Hit Rate 衡量在前 K 个检索结果中包含相关文档的比例:
def hit_rate_at_k(retrieved_docs, relevant_docs, k=10):
"""计算 Hit Rate@K"""
hits = sum(
1 for ret, rel in zip(retrieved_docs, relevant_docs)
if len(set(ret[:k]) & set(rel)) > 0
)
return hits / len(retrieved_docs)MRR(平均倒数排名)
MRR 衡量相关文档的平均排名质量。如果第一个结果就命中,得分为 1;第二个命中得分为 0.5:
其中 是第一个相关文档的排名位置。
def mean_reciprocal_rank(retrieved_docs, relevant_docs):
"""计算 MRR"""
reciprocal_ranks = []
for ret, rel in zip(retrieved_docs, relevant_docs):
for i, doc in enumerate(ret, 1):
if doc in rel:
reciprocal_ranks.append(1 / i)
break
else:
reciprocal_ranks.append(0)
return sum(reciprocal_ranks) / len(reciprocal_ranks)Recall@K
Recall@K 衡量在前 K 个检索结果中,相关文档的召回比例:
指标对比:
| 指标 | 关注点 | 适用场景 |
|---|---|---|
| Hit Rate@K | 是否命中 | 只需一个相关结果即可的场景 |
| MRR | 首次命中位置 | 强调首个结果质量的场景 |
| Recall@K | 召回完整性 | 需要全面信息的场景 |
生成质量评估
生成质量评估关注 LLM 输出的准确性、相关性和流畅性。
BLEU(Bilingual Evaluation Understudy)
BLEU 是最早广泛使用的文本生成评估指标,通过 n-gram 重合度来衡量生成文本与参考答案的相似度:
其中 BP 是简短惩罚因子, 是 n-gram 精度。
优点:
- 计算简单高效
- 与人类判断有一定相关性
- 广泛使用,便于比较
局限性:
- 无法评估语义相似性
- 对词序变化敏感
- 不适合开放式生成
from sacrebleu import sentence_bleu
hypothesis = "RAG combines retrieval with generation"
reference = "RAG is retrieval-augmented generation"
bleu = sentence_bleu(hypothesis, [reference])
print(f"BLEU: {bleu.score}") # BLEU: 47.52ROUGE(Recall-Oriented Understudy for Gisting Evaluation)
ROUGE 是一组基于召回率的评估指标,常用的包括:
| 指标 | 说明 | 计算方式 |
|---|---|---|
| ROUGE-N | N-gram 召回率 | 重叠 N-gram 数 / 参考 N-gram 数 |
| ROUGE-L | 最长公共子序列 | LCS 长度 / 参考长度 |
| ROUGE-S | Skip-bigram 召回率 | Skip-bigram 重叠数 |
from rouge import Rouge
hypothesis = "RAG combines retrieval with generation for better answers"
reference = "RAG uses retrieval-augmented generation"
rouge = Rouge()
scores = rouge.get_scores(hypothesis, reference)
print(scores)
# [{'rouge-1': {'f': 0.52, 'p': 0.46, 'r': 0.60}, ...}]BERTScore
BERTScore 利用预训练语言模型(如 BERT)计算生成文本与参考答案的语义相似度:
其中
优势:
- 捕捉语义相似性,不受词汇表层差异影响
- 支持同义词和释义
- 在中文场景表现优于 BLEU/ROUGE
from bert_score import score
cands = ["RAG 是一种结合检索和生成的技术"]
refs = ["RAG 即检索增强生成"]
P, R, F1 = score(cands, refs, lang="zh", verbose=True)
print(f"BERTScore F1: {F1.item():.3f}") # BERTScore F1: 0.892端到端评估策略
端到端评估从用户视角评估整个 RAG 系统的表现,关注最终答案的整体质量。
人工评估维度
人工评估仍是 RAG 评估的金标准,主要维度包括:
| 维度 | 评估问题 | 评分标准 |
|---|---|---|
| 准确性 | 答案是否正确? | 1-5 分 |
| 相关性 | 答案是否针对问题? | 1-5 分 |
| 完整性 | 答案是否全面? | 1-5 分 |
| 流畅性 | 表达是否自然流畅? | 1-5 分 |
| 引用准确性 | 引用是否正确标注? | 有/无 |
LLM-as-Judge 评估
使用强大的 LLM(如 GPT-4)作为评估器,可以接近人类判断的质量:
from openai import OpenAI
client = OpenAI()
def llm_judge_evaluate(question: str, answer: str, context: str) -> dict:
"""使用 GPT-4 作为评估器"""
prompt = f"""你是一个专业的 RAG 系统评估专家。请评估以下问答系统的质量。
问题: {question}
上下文:
{context}
答案:
{answer}
请从以下维度打分(1-5分),并给出简短理由:
1. 答案准确性
2. 答案相关性
3. 上下文利用率
4. 幻觉程度(越低分越好)
输出格式:
{{"accuracy": X, "relevance": X, "context_utilization": X, "hallucination": X, "reasons": "..."}}"""
response = client.chat.completions.create(
model="gpt-4-turbo",
messages=[{"role": "user", "content": prompt}]
)
return parse_llm_response(response.choices[0].message.content)注意事项:
- 位置偏差:评估者倾向于给靠前或靠后的答案更高分
- 顺序效应:成对比较时,第二个选项可能获得更高分
- 自我偏好:评估者可能偏好与自己风格相似的答案
评估实践代码示例
完整的 RAGAS 评估流程
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
context_entity_recall
)
from ragas.run_config import RunConfig
from datasets import Dataset
import pandas as pd
# 准备评估数据集
eval_data = [
{
"user_input": "RAG 的核心原理是什么?",
"retrieved_contexts": [
"RAG = Retrieval Augmented Generation",
"RAG 结合了信息检索与大语言模型生成"
],
"response": "RAG 即检索增强生成,它将信息检索与大语言模型生成相结合,通过先检索相关文档,再基于检索结果生成答案。",
"reference": "RAG 是检索增强生成"
},
{
"user_input": "如何评估 RAG 系统?",
"retrieved_contexts": [
"RAGAS 是常用的 RAG 评估框架",
"核心指标包括 Faithfulness、Answer Relevancy 等"
],
"response": "RAG 系统评估通常使用 RAGAS 框架,主要评估指标包括忠实度、答案相关性、上下文精确度和上下文召回率。",
"reference": "RAG 评估框架 RAGAS"
}
]
# 转换为 Dataset 格式
dataset = Dataset.from_pandas(pd.DataFrame(eval_data))
# 配置评估参数
run_config = RunConfig(
max_workers=4,
timeout=300
)
# 执行评估
result = evaluate(
dataset,
metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall,
context_entity_recall
],
run_config=run_config
)
# 输出结果
print(result)
# {
# 'faithfulness': 0.95,
# 'answer_relevancy': 0.89,
# 'context_precision': 0.82,
# 'context_recall': 0.88,
# 'context_entity_recall': 0.91
# }
# 导出详细报告
result.to_pandas()混合指标评估系统
import numpy as np
from dataclasses import dataclass
from typing import List, Dict
from ragas.metrics import faithfulness, answer_relevancy
from sklearn.metrics import ndcg_score
@dataclass
class RAGEvalResult:
"""RAG 评估结果"""
query_id: str
retrieval_metrics: Dict[str, float]
generation_metrics: Dict[str, float]
end_to_end_score: float
def is_acceptable(self, thresholds: Dict[str, float]) -> bool:
"""判断评估结果是否达到阈值"""
all_metrics = {**self.retrieval_metrics, **self.generation_metrics}
return all(
all_metrics.get(k, 0) >= v
for k, v in thresholds.items()
)
def comprehensive_evaluation(
queries: List[str],
rag_pipeline,
ground_truth: Dict[str, List[str]]
) -> List[RAGEvalResult]:
"""综合评估 RAG 系统"""
results = []
for query in queries:
# 1. 检索阶段
retrieved = rag_pipeline.retrieve(query, top_k=10)
retrieved_ids = [doc.id for doc in retrieved]
relevant_ids = ground_truth.get(query, [])
# 2. 计算检索指标
retrieval_metrics = {
"hit_rate_at_5": hit_rate_at_k([retrieved_ids], [relevant_ids], k=5),
"mrr": mean_reciprocal_rank([retrieved_ids], [relevant_ids]),
"ndcg@10": calculate_ndcg(retrieved_ids, relevant_ids, k=10)
}
# 3. 生成阶段
response = rag_pipeline.generate(query, retrieved)
# 4. 计算生成指标
generation_metrics = {
"faithfulness": faithfulness_score(response, retrieved),
"answer_relevancy": answer_relevancy_score(query, response)
}
# 5. 综合评分
end_to_end = 0.4 * retrieval_metrics["mrr"] + \
0.3 * generation_metrics["faithfulness"] + \
0.3 * generation_metrics["answer_relevancy"]
results.append(RAGEvalResult(
query_id=query,
retrieval_metrics=retrieval_metrics,
generation_metrics=generation_metrics,
end_to_end_score=end_to_end
))
return results
def calculate_ndcg(retrieved: List[str], relevant: List[str], k: int) -> float:
"""计算 NDCG@K"""
relevance_scores = [1 if doc_id in relevant else 0 for doc_id in retrieved[:k]]
if sum(relevance_scores) == 0:
return 0.0
return ndcg_score([relevance_scores], [relevant])持续监控流水线
from datetime import datetime, timedelta
import pandas as pd
import logging
class RAGMonitoringPipeline:
"""RAG 系统持续监控流水线"""
def __init__(self, rag_app, eval_samples: List[Dict]):
self.rag_app = rag_app
self.eval_samples = eval_samples
self.history = []
self.thresholds = {
"faithfulness": 0.85,
"answer_relevancy": 0.80,
"mrr": 0.75
}
def run_monitoring(self) -> Dict:
"""执行监控评估"""
eval_data = []
for sample in self.eval_samples:
response = self.rag_app.query(sample["user_input"])
eval_data.append({
"user_input": sample["user_input"],
"retrieved_contexts": response.contexts,
"response": response.text,
"reference": sample.get("reference", "")
})
# 使用 RAGAS 评估
dataset = Dataset.from_pandas(pd.DataFrame(eval_data))
results = evaluate(dataset, metrics=[
faithfulness, answer_relevancy
])
# 计算检索指标
retrieval_scores = self._evaluate_retrieval(eval_data)
# 生成报告
report = {
"timestamp": datetime.now(),
"generation_metrics": results,
"retrieval_metrics": retrieval_scores,
"alert_triggered": self._check_thresholds(results, retrieval_scores)
}
self.history.append(report)
return report
def _evaluate_retrieval(self, eval_data: List[Dict]) -> Dict:
"""评估检索质量"""
mrr_scores = []
hit_rates = []
for sample in eval_data:
# 简化实现,实际需要 ground truth
mrr_scores.append(0.8) # 占位
hit_rates.append(0.9)
return {
"mrr_mean": np.mean(mrr_scores),
"hit_rate_mean": np.mean(hit_rates)
}
def _check_thresholds(self, gen_metrics: Dict, ret_metrics: Dict) -> bool:
"""检查是否触发告警"""
return (
gen_metrics.get("faithfulness", 0) < self.thresholds["faithfulness"] or
gen_metrics.get("answer_relevancy", 0) < self.thresholds["answer_relevancy"] or
ret_metrics.get("mrr_mean", 0) < self.thresholds["mrr"]
)
def get_drift_detection(self) -> Dict:
"""检测质量漂移"""
if len(self.history) < 7:
return {"status": "insufficient_data"}
recent = self.history[-7:]
previous = self.history[-14:-7] if len(self.history) >= 14 else self.history[:-7]
recent_avg = np.mean([h["generation_metrics"]["faithfulness"]
for h in recent])
previous_avg = np.mean([h["generation_metrics"]["faithfulness"]
for h in previous])
drift = recent_avg - previous_avg
return {
"status": "drift_detected" if abs(drift) > 0.05 else "stable",
"recent_avg": recent_avg,
"previous_avg": previous_avg,
"drift_delta": drift
}
# 使用示例
monitoring = RAGMonitoringPipeline(my_rag_app, eval_samples=eval_set)
report = monitoring.run_monitoring()
if report["alert_triggered"]:
logging.warning(f"RAG 质量告警: {report}")
drift_report = monitoring.get_drift_detection()
print(f"漂移检测: {drift_report}")生产级评估最佳实践
分层评估策略
┌─────────────────────────────────────────────────────────────┐
│ 快速冒烟测试 │
│ 每次提交触发,< 1分钟,低成本 │
│ 检查基本功能、API 可用性 │
├─────────────────────────────────────────────────────────────┤
│ 回归测试 │
│ 每日运行,5-10分钟,中等成本 │
│ RAGAS 核心指标、检索基线 │
├─────────────────────────────────────────────────────────────┤
│ 深度评估 │
│ 每周运行,30-60分钟,高成本 │
│ LLM-as-Judge、人工抽检 │
├─────────────────────────────────────────────────────────────┤
│ 性能基准 │
│ 每月运行,全面评估 │
│ A/B 测试、新模型对比 │
└─────────────────────────────────────────────────────────────┘
阈值设置建议
| 指标 | 生产阈值 | 说明 |
|---|---|---|
| Faithfulness | ≥ 0.85 | 低于此值可能产生严重幻觉 |
| Answer Relevancy | ≥ 0.80 | 确保答案真正回答问题 |
| Context Precision | ≥ 0.75 | 检索结果需足够精准 |
| MRR | ≥ 0.70 | 相关文档排名需靠前 |
| Hit Rate@5 | ≥ 0.85 | 绝大多数查询需命中 |
评估数据集管理
from datasets import load_dataset, concatenate_datasets
class EvalDatasetManager:
"""评估数据集管理器"""
def __init__(self, dataset_path: str):
self.dataset_path = dataset_path
self.current_version = self._get_latest_version()
def load_eval_set(self, split: str = "test") -> Dataset:
"""加载评估集"""
return load_dataset(
self.dataset_path,
split=split,
revision=self.current_version
)
def add_samples(self, samples: List[Dict]):
"""添加新评估样本"""
new_data = Dataset.from_list(samples)
# 合并并去重
existing = self.load_eval_set()
combined = concatenate_datasets([existing, new_data])
combined = combined.unique()
combined.save_to_disk(self.dataset_path)
def get_diverse_subset(self, n: int, seed: int = 42) -> Dataset:
"""获取多样化子集,覆盖不同主题和难度"""
full = self.load_eval_set()
# 按主题分层采样
return full.shuffle(seed=seed).select(range(n))