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 衡量答案是否忠实于检索到的上下文,避免模型引入外部知识或产生幻觉。其计算流程如下:

  1. 从答案中提取所有原子陈述(Atomic Statements)
  2. 判断每个陈述是否可归因于上下文
  3. 计算归因陈述的比例

低忠实度的典型表现

表现类型示例
信息补充上下文提到「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 采用反向问题生成的方式来评估该指标:

  1. 根据答案生成多个重新表述的问题
  2. 计算原始问题与生成问题的语义相似度
  3. 取最高相似度作为答案相关性得分

低相关性的典型场景

  • 答非所问:问题问原因,答案讲结果
  • 信息冗余:包含大量与问题无关的细节
  • 不完整:只回答了问题的部分子问题
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-eval
from 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.52

ROUGE(Recall-Oriented Understudy for Gisting Evaluation)

ROUGE 是一组基于召回率的评估指标,常用的包括:

指标说明计算方式
ROUGE-NN-gram 召回率重叠 N-gram 数 / 参考 N-gram 数
ROUGE-L最长公共子序列LCS 长度 / 参考长度
ROUGE-SSkip-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))

参考资料

Footnotes

  1. RAGAS: Evaluation Framework for RAG

  2. TruLens: Evaluable Deep Learning Applications