结合RAGChecker和RAGAS进行RAG系统评估
·
结合RAGChecker和RAGAS进行RAG系统评估,是目前最强大的组合方案——RAGAS提供快速、全面的核心指标,RAGChecker则深入诊断检索和生成模块的具体问题,两者互补形成完整的评估闭环。
一、核心定位对比
| 维度 | RAGAS | RAGChecker |
|---|---|---|
| 定位 | 通用RAG评估框架,行业事实标准 | 精细化诊断框架,亚马逊和上海交大联合研发 |
| 评估粒度 | 组件级别(检索器、生成器) | 声明级别(claim-level)细粒度诊断 |
| 核心优势 | 易用性高、指标全面、社区支持好 | 诊断性强、与人类判断相关性高(61.93% vs RAGAS 48.31%) |
| 适用场景 | 快速迭代、生产监控、整体性能评估 | 深度问题定位、性能瓶颈分析、精度优化 |
二、环境配置
2.1 安装依赖
# 安装RAGAS
pip install ragas datasets
# 安装RAGChecker
pip install ragchecker
python -m spacy download en_core_web_sm # RAGChecker依赖
# 可选:LangSmith用于追踪
pip install langchain langchain-openai langsmith
2.2 配置环境变量
import os
# OpenAI配置(两者都需要)
os.environ["OPENAI_API_KEY"] = "sk-xxx"
# LangSmith配置(可选,用于追踪)
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls_xxx"
os.environ["LANGCHAIN_PROJECT"] = "rag-evaluation"
三、数据准备
两个框架需要的数据格式略有不同,我们需要准备一个统一的数据结构,然后分别转换。
3.1 原始数据格式
# 准备评估数据集
eval_data = [
{
"query_id": "1",
"query": "What is RAGChecker?",
"gt_answer": "RAGChecker is an advanced automatic evaluation framework designed to assess and diagnose Retrieval-Augmented Generation (RAG) systems. It provides a comprehensive suite of metrics and tools for in-depth analysis of RAG performance.",
"response": "RAGChecker is an evaluation framework for RAG systems that provides detailed metrics for diagnosis.",
"retrieved_contexts": [
"RAGChecker is an advanced automatic evaluation framework designed to assess and diagnose Retrieval-Augmented Generation (RAG) systems.",
"It provides a comprehensive suite of metrics and tools for in-depth analysis of RAG performance."
]
},
# 更多样本...
]
3.2 转换为RAGAS格式
from datasets import Dataset
def convert_to_ragas_format(data):
"""转换为RAGAS所需的数据集格式"""
return Dataset.from_dict({
"question": [item["query"] for item in data],
"answer": [item["response"] for item in data],
"contexts": [item["retrieved_contexts"] for item in data],
"ground_truth": [item["gt_answer"] for item in data]
})
ragas_dataset = convert_to_ragas_format(eval_data)
3.3 转换为RAGChecker格式
import json
from ragchecker import RAGResults, RAGChecker
from ragchecker.metrics import all_metrics
def convert_to_ragchecker_format(data):
"""转换为RAGChecker所需的JSON格式"""
results = []
for item in data:
results.append({
"query_id": item["query_id"],
"query": item["query"],
"gt_answer": item["gt_answer"],
"response": item["response"],
"retrieved_contexts": item["retrieved_contexts"]
})
# 保存为临时文件或直接使用
return {"results": results}
# 创建RAGResults对象
rag_results = RAGResults.from_dict(convert_to_ragchecker_format(eval_data))
四、执行评估:RAGAS先行,RAGChecker深入
4.1 第一阶段:RAGAS快速评估
from ragas import evaluate
from ragas.metrics import (
faithfulness, # 忠实度:答案是否基于上下文
answer_relevancy, # 答案相关性
context_precision, # 上下文精确率
context_recall, # 上下文召回率
answer_correctness # 答案正确性
)
# 执行RAGAS评估
ragas_result = evaluate(
dataset=ragas_dataset,
metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall,
answer_correctness
]
)
print("RAGAS评估结果:")
print(ragas_result)
# 输出示例:
# {
# 'faithfulness': 0.85,
# 'answer_relevancy': 0.92,
# 'context_precision': 0.78,
# 'context_recall': 0.81,
# 'answer_correctness': 0.79
# }
4.2 第二阶段:RAGChecker深度诊断
from ragchecker import RAGChecker
# 初始化RAGChecker评估器
evaluator = RAGChecker(
extractor_name="gpt-4", # 用于提取声明的模型
checker_name="gpt-4", # 用于检查声明的模型
batch_size_extractor=32,
batch_size_checker=32
)
# 执行RAGChecker评估
evaluator.evaluate(rag_results, all_metrics)
# 查看详细结果
print("\nRAGChecker评估结果:")
print(rag_results)
"""
输出示例:
RAGResults(
Metrics:
{
"overall_metrics": {
"precision": 76.4,
"recall": 62.5,
"f1": 68.3
},
"retriever_metrics": {
"claim_recall": 61.4, # 检索召回:检索覆盖了多少真实信息
"context_precision": 87.5 # 上下文精度:检索结果中有多少相关
},
"generator_metrics": {
"context_utilization": 87.5, # 上下文利用率
"noise_sensitivity_in_relevant": 19.1, # 相关噪声敏感度
"noise_sensitivity_in_irrelevant": 0.0, # 不相关噪声敏感度
"hallucination": 4.5, # 幻觉率
"self_knowledge": 27.3, # 自知识依赖
"faithfulness": 68.2 # 忠实度
}
}
)
"""
五、结果分析与解读
5.1 指标对应关系
| RAGAS指标 | RAGChecker对应指标 | 解读 |
|---|---|---|
| faithfulness | generator_metrics.faithfulness | 答案是否基于检索上下文 |
| context_precision | retriever_metrics.context_precision | 检索结果的相关性 |
| context_recall | retriever_metrics.claim_recall | 检索覆盖的真实信息比例 |
| - | generator_metrics.hallucination | 幻觉比例(RAGAS无直接对应) |
| - | generator_metrics.noise_sensitivity | 对噪声的容忍度(RAGAS无直接对应) |
5.2 诊断流程示例
def diagnose_rag_performance(ragas_result, ragchecker_result):
"""结合两个框架的结果进行诊断"""
diagnosis = {
"summary": {},
"issues": [],
"recommendations": []
}
# 1. 整体判断
if ragas_result["faithfulness"] < 0.7:
diagnosis["issues"].append("忠实度低:答案经常脱离检索上下文")
# 使用RAGChecker深入诊断
if ragchecker_result.generator_metrics.hallucination > 10:
diagnosis["recommendations"].append(
"幻觉率过高,建议:1) 优化prompt强调基于上下文 2) 考虑使用更小的temperature"
)
# 2. 检索问题诊断
if ragas_result["context_precision"] < 0.6:
diagnosis["issues"].append("上下文精确率低:检索结果包含大量噪声")
if ragchecker_result.retriever_metrics.context_precision < 70:
diagnosis["recommendations"].append(
"检索器引入过多噪声,建议:1) 优化embedding模型 2) 调整chunk大小 3) 增加rerank"
)
# 3. 检索覆盖率诊断
if ragas_result["context_recall"] < 0.6:
diagnosis["issues"].append("上下文召回率低:关键信息未被检索到")
if ragchecker_result.retriever_metrics.claim_recall < 60:
diagnosis["recommendations"].append(
"检索遗漏关键信息,建议:1) 增加检索chunk数量 2) 使用混合检索策略"
)
# 4. 生成器噪声敏感度分析
if ragchecker_result.generator_metrics.noise_sensitivity_in_relevant > 20:
diagnosis["recommendations"].append(
"生成器对相关噪声敏感,建议优化prompt或尝试更大规模的生成模型"
)
return diagnosis
# 执行诊断
diagnosis = diagnose_rag_performance(ragas_result, rag_results)
print(json.dumps(diagnosis, indent=2))
六、与LangSmith集成实现全链路追踪
6.1 配置LangSmith回调
from langchain.callbacks.tracers import LangChainTracer
from langchain.callbacks.manager import CallbackManager
# 初始化LangSmith tracer
tracer = LangChainTracer(project_name="rag-evaluation")
# 在RAGAS评估中启用追踪
from ragas import evaluate
from ragas.integrations.langsmith import LangSmithTracer
# RAGAS支持通过callback传递追踪
result = evaluate(
dataset=ragas_dataset,
metrics=[faithfulness, answer_relevancy],
callbacks=[tracer] # 将评估过程记录到LangSmith
)
6.2 统一的数据流架构
import asyncio
from datetime import datetime
class UnifiedRAGEvaluator:
"""统一的RAG评估器,整合RAGAS、RAGChecker和LangSmith"""
def __init__(self, openai_api_key, langsmith_api_key=None):
self.openai_api_key = openai_api_key
self.langsmith_api_key = langsmith_api_key
# 初始化评估器
self.ragas_metrics = [
faithfulness, answer_relevancy,
context_precision, context_recall, answer_correctness
]
# RAGChecker评估器(延迟初始化)
self.ragchecker_evaluator = None
def _init_ragchecker(self):
"""延迟初始化RAGChecker"""
if not self.ragchecker_evaluator:
from ragchecker import RAGChecker
self.ragchecker_evaluator = RAGChecker(
extractor_name="gpt-4",
checker_name="gpt-4",
batch_size_extractor=32,
batch_size_checker=32
)
async def evaluate(self, eval_data, sample_rate=1.0):
"""
执行完整评估
Args:
eval_data: 评估数据列表
sample_rate: 采样率,用于成本控制
"""
# 采样
import random
sampled_data = [d for d in eval_data if random.random() < sample_rate]
print(f"开始评估 {len(sampled_data)} 个样本...")
# 阶段1: RAGAS快速评估
print("阶段1: RAGAS评估中...")
ragas_dataset = self._to_ragas_format(sampled_data)
ragas_result = await evaluate(
dataset=ragas_dataset,
metrics=self.ragas_metrics,
llm=self._get_evaluator_llm()
)
# 阶段2: RAGChecker深度诊断(对发现问题样本进行)
print("阶段2: RAGChecker深度诊断中...")
self._init_ragchecker()
# 识别问题样本
problem_indices = self._identify_problem_samples(ragas_result, threshold=0.7)
deep_diagnosis = {}
if problem_indices:
problem_data = [sampled_data[i] for i in problem_indices]
ragchecker_results = self._run_ragchecker(problem_data)
deep_diagnosis = self._analyze_ragchecker_results(ragchecker_results)
# 生成综合报告
report = self._generate_report(
ragas_result=ragas_result,
deep_diagnosis=deep_diagnosis,
sample_count=len(sampled_data),
problem_count=len(problem_indices)
)
# 可选:上传到LangSmith
if self.langsmith_api_key:
self._upload_to_langsmith(report, sampled_data)
return report
def _identify_problem_samples(self, ragas_result, threshold=0.7):
"""识别需要深度诊断的问题样本"""
problem_indices = []
for i, scores in enumerate(ragas_result.scores):
if scores["faithfulness"] < threshold:
problem_indices.append(i)
return problem_indices
def _run_ragchecker(self, data):
"""执行RAGChecker深度诊断"""
rag_results = RAGResults.from_dict(
self._to_ragchecker_format(data)
)
self.ragchecker_evaluator.evaluate(rag_results, all_metrics)
return rag_results
def _generate_report(self, ragas_result, deep_diagnosis, sample_count, problem_count):
"""生成综合评估报告"""
return {
"timestamp": datetime.now().isoformat(),
"summary": {
"sample_count": sample_count,
"problem_count": problem_count,
"problem_rate": problem_count / sample_count if sample_count > 0 else 0
},
"ragas_scores": ragas_result,
"deep_diagnosis": deep_diagnosis,
"recommendations": self._generate_recommendations(ragas_result, deep_diagnosis)
}
def _generate_recommendations(self, ragas_result, deep_diagnosis):
"""生成优化建议"""
recommendations = []
# 基于RAGAS结果
if ragas_result["context_precision"] < 0.6:
recommendations.append("检索精度低:考虑优化embedding模型或增加rerank")
if ragas_result["context_recall"] < 0.6:
recommendations.append("检索覆盖率低:考虑增加chunk数量或使用混合检索")
if ragas_result["faithfulness"] < 0.7:
recommendations.append("忠实度低:优化prompt强调基于上下文")
# 基于RAGChecker深度诊断
if deep_diagnosis.get("hallucination_rate", 0) > 10:
recommendations.append("幻觉率高:考虑使用更小的temperature或更严格的prompt")
return recommendations
# 辅助方法
def _to_ragas_format(self, data):
from datasets import Dataset
return Dataset.from_dict({
"question": [d["query"] for d in data],
"answer": [d["response"] for d in data],
"contexts": [d["retrieved_contexts"] for d in data],
"ground_truth": [d["gt_answer"] for d in data]
})
def _to_ragchecker_format(self, data):
results = []
for d in data:
results.append({
"query_id": d["query_id"],
"query": d["query"],
"gt_answer": d["gt_answer"],
"response": d["response"],
"retrieved_contexts": d["retrieved_contexts"]
})
return {"results": results}
def _get_evaluator_llm(self):
from langchain_openai import ChatOpenAI
return ChatOpenAI(model="gpt-4", temperature=0)
def _analyze_ragchecker_results(self, results):
"""解析RAGChecker结果"""
return {
"hallucination_rate": results.generator_metrics.hallucination,
"noise_sensitivity": results.generator_metrics.noise_sensitivity_in_relevant,
"context_utilization": results.generator_metrics.context_utilization,
"claim_recall": results.retriever_metrics.claim_recall,
"context_precision": results.retriever_metrics.context_precision
}
def _upload_to_langsmith(self, report, data):
"""上传评估结果到LangSmith"""
import langsmith
client = langsmith.Client(api_key=self.langsmith_api_key)
# 创建运行记录
client.create_run(
name="rag-evaluation",
inputs={"sample_count": len(data)},
outputs=report,
run_type="chain",
extra={"metadata": {"framework": "ragas+ragchecker"}}
)
6.3 使用示例
# 初始化评估器
evaluator = UnifiedRAGEvaluator(
openai_api_key="sk-xxx",
langsmith_api_key="ls_xxx"
)
# 准备评估数据
eval_data = [
{
"query_id": "1",
"query": "What is RAGChecker?",
"gt_answer": "RAGChecker is an evaluation framework...",
"response": "RAGChecker is a framework for RAG evaluation.",
"retrieved_contexts": ["RAGChecker is an evaluation framework..."]
},
# 更多数据...
]
# 执行评估(采样10%控制成本)
report = await evaluator.evaluate(eval_data, sample_rate=0.1)
# 查看报告
print(json.dumps(report, indent=2, ensure_ascii=False))
七、最佳实践总结
7.1 评估流程建议
-
分层评估策略:
- 日常监控:使用RAGAS + LangSmith(采样5-10%)
- 版本迭代:全量RAGAS评估 + 问题样本RAGChecker诊断
- 深度优化:对发现的问题模块使用RAGChecker专项分析
-
成本控制:
- RAGChecker使用更强大的模型(GPT-4),成本较高,建议只对问题样本使用
- RAGAS可以使用GPT-3.5-Turbo进行日常评估
-
指标解读重点:
- RAGAS:快速发现问题(忠实度低?召回率低?)
- RAGChecker:深入诊断原因(幻觉率高?噪声敏感?)
7.2 典型问题诊断路径
| 现象 | RAGAS指标 | RAGChecker诊断 | 优化建议 |
|---|---|---|---|
| 答案跑题 | answer_relevancy低 | - | 优化prompt,明确任务 |
| 答案编造 | faithfulness低 | hallucination高 | 强调基于上下文,降低temperature |
| 检索噪声大 | context_precision低 | context_precision低 | 增加rerank,优化embedding |
| 信息遗漏 | context_recall低 | claim_recall低 | 增加chunk数,使用混合检索 |
| 对噪声敏感 | - | noise_sensitivity高 | 使用更大模型,优化prompt |
这种组合方案能帮你建立从"发现问题"到"诊断原因"再到"优化改进"的完整评估闭环。需要我针对某个具体场景(比如特定领域的RAG评估)展开说明吗?
更多推荐
所有评论(0)