结合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 评估流程建议

  1. 分层评估策略

    • 日常监控:使用RAGAS + LangSmith(采样5-10%)
    • 版本迭代:全量RAGAS评估 + 问题样本RAGChecker诊断
    • 深度优化:对发现的问题模块使用RAGChecker专项分析
  2. 成本控制

    • RAGChecker使用更强大的模型(GPT-4),成本较高,建议只对问题样本使用
    • RAGAS可以使用GPT-3.5-Turbo进行日常评估
  3. 指标解读重点

    • 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评估)展开说明吗?

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐