如何评测并改进 RAG 应用

在本指南中，你将学习如何用 Ragas 评测并迭代改进一个 RAG（Retrieval-Augmented Generation）应用。

你将完成什么

• 设置评测数据集
• 建立测量 RAG 表现的指标
• 构建可复用的评测流水线
• 分析错误并系统地改进你的 RAG 应用
• 学习如何利用 Ragas 做 RAG 评测

设置并运行 RAG 系统

我们构建了一个简单的 RAG 系统，它从 Hugging Face documentation 数据集 中检索相关文档，并用 LLM 生成答案。该数据集包含许多 Hugging Face 包的文档页面，以 markdown 存储，为测试 RAG 能力提供了丰富的知识库。

完整实现可在这里获取：ragas_examples/improve_rag/

flowchart LR
    A[User Query] --> B[Retrieve Documents<br/>BM25]
    B --> C[Generate Response<br/>OpenAI]
    C --> D[Return Answer]

要运行它，安装依赖：

uv pip install "ragas-examples[improverag]"

然后运行 RAG 应用：

import os
import asyncio
from openai import AsyncOpenAI
from ragas_examples.improve_rag.rag import RAG, BM25Retriever

# Set up OpenAI client
os.environ["OPENAI_API_KEY"] = "<your_key>"
openai_client = AsyncOpenAI()

# Create retriever and RAG system
retriever = BM25Retriever()
rag = RAG(openai_client, retriever)

# Query the system
question = "What architecture is the `tokenizers-linux-x64-musl` binary designed for?"
result = asyncio.run(rag.query(question))
print(f"Answer: {result['answer']}")

输出

Answer: It's built for the x86_64 architecture (specifically the x86_64-unknown-linux-musl target — 64-bit Linux with musl libc).

理解 RAG 实现

上面的代码使用一个简单的 RAG 类，演示核心 RAG 模式。它的工作方式如下：

# examples/ragas_examples/improve_rag/rag.py
from typing import Any, Dict, Optional
from openai import AsyncOpenAI

class RAG:
    """Simple RAG system for document retrieval and answer generation."""

    def __init__(self, llm_client: AsyncOpenAI, retriever: BM25Retriever, system_prompt=None, model="gpt-4o-mini", default_k=3):
        self.llm_client = llm_client
        self.retriever = retriever
        self.model = model
        self.default_k = default_k
        self.system_prompt = system_prompt or "Answer only based on documents. Be concise.\n\nQuestion: {query}\nDocuments:\n{context}\nAnswer:"

    async def query(self, question: str, top_k: Optional[int] = None) -> Dict[str, Any]:
        """Query the RAG system."""
        if top_k is None:
            top_k = self.default_k

        return await self._naive_query(question, top_k)

    async def _naive_query(self, question: str, top_k: int) -> Dict[str, Any]:
        """Handle naive RAG: retrieve once, then generate."""
        # 1. Retrieve documents using BM25
        docs = self.retriever.retrieve(question, top_k)

        if not docs:
            return {"answer": "No relevant documents found.", "retrieved_documents": [], "num_retrieved": 0}

        # 2. Build context from retrieved documents
        context = "\n\n".join([f"Document {i}:\n{doc.page_content}" for i, doc in enumerate(docs, 1)])
        prompt = self.system_prompt.format(query=question, context=context)

        # 3. Generate response using OpenAI with retrieved context
        response = await self.llm_client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )

        return {
            "answer": response.choices[0].message.content.strip(),
            "retrieved_documents": [{"content": doc.page_content, "metadata": doc.metadata, "document_id": i} for i, doc in enumerate(docs)],
            "num_retrieved": len(docs)
        }

这展示了本质的 RAG 模式：检索相关文档 → 注入 prompt → 生成答案。

创建评测数据集

我们将使用 huggingface_doc_qa_eval，这是一个关于 Hugging Face 文档的问答数据集。

下面是数据集中的几行示例：

评测脚本从这里下载数据集，并转换成 Ragas Dataset 格式：

# examples/ragas_examples/improve_rag/evals.py
import urllib.request
from pathlib import Path
from ragas import Dataset
import pandas as pd

def download_and_save_dataset() -> Path:
    dataset_path = Path("datasets/hf_doc_qa_eval.csv")
    dataset_path.parent.mkdir(exist_ok=True)

    if not dataset_path.exists():
        github_url = "https://raw.githubusercontent.com/vibrantlabsai/ragas/main/examples/ragas_examples/improve_rag/datasets/hf_doc_qa_eval.csv"
        urllib.request.urlretrieve(github_url, dataset_path)

    return dataset_path

def create_ragas_dataset(dataset_path: Path) -> Dataset:
    dataset = Dataset(name="hf_doc_qa_eval", backend="local/csv", root_dir=".")
    df = pd.read_csv(dataset_path)

    for _, row in df.iterrows():
        dataset.append({"question": row["question"], "expected_answer": row["expected_answer"]})

    dataset.save()
    return dataset

了解更多关于使用 datasets 的信息，见 Core Concepts - Datasets。

为 RAG 评测设置指标

现在我们已经准备好评测数据集，需要指标来测量 RAG 表现。从简单、聚焦的指标开始，直接测量你的核心用例。关于指标的更多信息见 Core Concepts - Metrics。

这里我们使用一个 correctness 离散指标，评测 RAG 回复是否包含期望答案中的关键信息，并且基于所提供的上下文事实上准确。

# examples/ragas_examples/improve_rag/evals.py
from ragas.metrics import DiscreteMetric

# Define correctness metric
correctness_metric = DiscreteMetric(
    name="correctness",
    prompt="""Compare the model response to the expected answer and determine if it's correct.

Consider the response correct if it:
1. Contains the key information from the expected answer
2. Is factually accurate based on the provided context
3. Adequately addresses the question asked

Return 'pass' if the response is correct, 'fail' if it's incorrect.

Question: {question}
Expected Answer: {expected_answer}
Model Response: {response}

Evaluation:""",
    allowed_values=["pass", "fail"],
)

现在我们有了评测指标，需要在数据集上系统地运行它。这就是 Ragas experiments 发挥作用的地方。

创建评测实验

实验函数在每个数据样本上运行你的 RAG 系统，并用我们的 correctness 指标评测回复。关于 experimentation 的更多信息见 Core Concepts - Experimentation。

实验函数接收包含 question、expected context 和 expected answer 的数据集行，然后：

1. 用问题查询 RAG 系统
2. 用 correctness 指标评测回复
3. 返回包括分数和 reason 的详细结果

# examples/ragas_examples/improve_rag/evals.py
import asyncio
from typing import Dict, Any
from ragas import experiment

@experiment()
async def evaluate_rag(row: Dict[str, Any], rag: RAG, llm) -> Dict[str, Any]:
    """
    Run RAG evaluation on a single row.

    Args:
        row: Dictionary containing question and expected_answer
        rag: Pre-initialized RAG instance
        llm: Pre-initialized LLM client for evaluation

    Returns:
        Dictionary with evaluation results
    """
    question = row["question"]

    # Query the RAG system
    rag_response = await rag.query(question, top_k=4)
    model_response = rag_response.get("answer", "")

    # Evaluate correctness asynchronously
    score = await correctness_metric.ascore(
        question=question,
        expected_answer=row["expected_answer"],
        response=model_response,
        llm=llm
    )

    # Return evaluation results
    result = {
        **row,
        "model_response": model_response,
        "correctness_score": score.value,
        "correctness_reason": score.reason,
        "mlflow_trace_id": rag_response.get("mlflow_trace_id", "N/A"),  # MLflow trace ID for debugging (explained later)
        "retrieved_documents": [
            doc.get("content", "")[:200] + "..." if len(doc.get("content", "")) > 200 else doc.get("content", "")
            for doc in rag_response.get("retrieved_documents", [])
        ]
    }

    return result

有了数据集、指标和实验函数，我们现在可以评测 RAG 系统的表现。

运行初始 RAG 实验

启动 MLflow 服务器

在运行评测之前，你必须启动 MLflow 服务器。RAG 系统会自动把 traces 记录到 MLFlow，用于调试和分析：

# Start MLflow server (required - in a separate terminal)
uv run mlflow ui --backend-store-uri sqlite:///mlflow.db --port 5000

MLflow UI 将在 http://127.0.0.1:5000 可用。

运行初始 RAG 实验

现在让我们运行完整评测流水线，获取 RAG 系统的基线表现指标：

# Import required components
import asyncio
from datetime import datetime
from ragas_examples.improve_rag.evals import (
    evaluate_rag,
    download_and_save_dataset,
    create_ragas_dataset,
    get_openai_client,
    get_llm_client
)
from ragas_examples.improve_rag.rag import RAG, BM25Retriever

async def run_evaluation():
    # Download and prepare dataset
    dataset_path = download_and_save_dataset()
    dataset = create_ragas_dataset(dataset_path)

    # Initialize RAG components
    openai_client = get_openai_client()
    retriever = BM25Retriever()
    rag = RAG(llm_client=openai_client, retriever=retriever, model="gpt-5-mini", mode="naive")
    llm = get_llm_client()

    # Run evaluation experiment
    exp_name = f"{datetime.now().strftime('%Y%m%d-%H%M%S')}_naiverag"
    results = await evaluate_rag.arun(
        dataset, 
        name=exp_name,
        rag=rag,
        llm=llm
    )

    # Print results
    if results:
        pass_count = sum(1 for result in results if result.get("correctness_score") == "pass")
        total_count = len(results)
        pass_rate = (pass_count / total_count) * 100 if total_count > 0 else 0
        print(f"Results: {pass_count}/{total_count} passed ({pass_rate:.1f}%)")

    return results

# Run the evaluation
results = await run_evaluation()
print(results)

这会下载数据集、初始化 BM25 retriever、在每个样本上运行评测实验，并把详细结果作为 CSV 文件保存到 experiments/ 目录以供分析。

输出

Results: 43/66 passed (65.2%)
Evaluation completed successfully!

Detailed results:
Experiment(name=20250924-212541_naiverag,  len=66)

以 65.2% 的通过率为基线。experiments/ 中的详细结果 CSV 现在包含我们进行错误分析和系统改进所需的全部数据。

在 MLflow 中查看 traces

实验结果 CSV 为每次评测包含 mlflow_trace_id 和 mlflow_trace_url，让你可以分析详细的执行 traces。traces 帮助你准确理解失败发生在哪里——是在检索、生成还是评测步骤。

RAG 系统会自动把 traces 记录到（先前启动的）MLflow 服务器，你可以在 http://127.0.0.1:5000 查看它们。

这让你可以：

1. 在 CSV 中分析结果：查看回复、指标分数和 reasons
2. 用 traces 深入分析：点击结果中的 mlflow_trace_url，直接跳转到 MLflow UI 中该次评测的详细执行 trace

提示：点击 Trace URL 进行调试

每次评测结果都包含 mlflow_trace_url——一个可点击的直达 MLflow UI 中 trace 的链接。无需手动导航或复制 trace IDs。只需点击，直接跳到详细执行 trace！

分析错误与失败模式

运行评测后，检查 experiments/ 目录中的结果 CSV 文件，识别失败用例中的模式。每一行都包含 mlflow_trace_id/mlflow_trace_url——以便在 MLflow UI 中查看详细执行 traces。标注每个失败用例以理解模式，从而改进我们的应用。

对我们评测中实际失败模式的分析：

在我们的例子中，核心问题是 检索失败——BM25 retriever 没有找到包含答案的文档。模型正确地遵循指令，在文档不包含信息时说出来，但检索到的是错误文档。

文档检索差的例子 BM25 retriever 未能检索到包含答案的相关文档：

基于这一分析，我们可以看到检索是主要瓶颈。让我们实施针对性改进。

改进 RAG 应用

把检索识别为主要瓶颈后，我们可以通过两种方式改进系统：

传统方法 聚焦更好的切分、混合搜索或向量 embeddings。然而，由于我们的 BM25 检索用单次查询持续错过相关文档，我们将转而探索 agentic 方法。

Agentic RAG 让 AI 迭代地打磨其搜索策略——尝试多个搜索词，并决定何时已找到足够上下文，而不是依赖一次静态查询。

Agentic RAG 实现

flowchart LR
    A[User Query] --> B[AI Agent<br/>OpenAI]
    B --> C[BM25 Tool]
    C --> B
    B --> D[Final Answer]

对一个示例查询运行 Agentic RAG 应用：

# Switch to agentic mode
rag_agentic = RAG(openai_client, retriever, mode="agentic")

question = "What architecture is the `tokenizers-linux-x64-musl` binary designed for?"
result = await rag_agentic.query(question)
print(f"Answer: {result['answer']}")

输出

Answer: It targets x86_64 — i.e. the x86_64-unknown-linux-musl target triple.

理解 Agentic RAG 实现

Agentic RAG 模式使用 OpenAI Agents SDK 创建一个带有 BM25 检索工具的 AI agent：

# Key components from the RAG class when mode="agentic"
from agents import Agent, Runner, function_tool

def _setup_agent(self):
    """Setup agent for agentic mode."""
    @function_tool
    def retrieve(query: str) -> str:
        """Search documents using BM25 retriever for a given query."""
        docs = self.retriever.retrieve(query, self.default_k)
        if not docs:
            return "No documents found."
        return "\n\n".join([f"Doc {i}: {doc.page_content}" for i, doc in enumerate(docs, 1)])

    self._agent = Agent(
        name="RAG Assistant",
        model=self.model,
        instructions="Use short keywords to search. Try 2-3 different searches. Only answer based on documents. Be concise.",
        tools=[retrieve]
    )

async def _agentic_query(self, question: str, top_k: int) -> Dict[str, Any]:
    """Handle agentic mode: agent controls retrieval strategy."""
    result = await Runner.run(self._agent, input=question)
    print(result.answer)

与 naive 模式的单次检索调用不同，agent 自主决定何时以及如何搜索——尝试多种关键词组合，直到找到足够上下文。

再次运行实验并比较结果

现在让我们评测 agentic RAG 方法：

# Import required components
import asyncio
from datetime import datetime
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

from ragas_examples.improve_rag.evals import (
    evaluate_rag,
    download_and_save_dataset, 
    create_ragas_dataset,
    get_openai_client,
    get_llm_client
)
from ragas_examples.improve_rag.rag import RAG, BM25Retriever

async def run_agentic_evaluation():
    # Download and prepare dataset
    dataset_path = download_and_save_dataset()
    dataset = create_ragas_dataset(dataset_path)

    # Initialize RAG components with agentic mode
    openai_client = get_openai_client()
    retriever = BM25Retriever()
    rag = RAG(llm_client=openai_client, retriever=retriever, model="gpt-5-mini", mode="agentic")
    llm = get_llm_client()

    # Run evaluation experiment
    exp_name = f"{datetime.now().strftime('%Y%m%d-%H%M%S')}_agenticrag"
    results = await evaluate_rag.arun(
        dataset, 
        name=exp_name,
        rag=rag,
        llm=llm
    )

    # Print results
    if results:
        pass_count = sum(1 for result in results if result.get("correctness_score") == "pass")
        total_count = len(results)
        pass_rate = (pass_count / total_count) * 100 if total_count > 0 else 0
        print(f"Results: {pass_count}/{total_count} passed ({pass_rate:.1f}%)")

    return results

# Run the agentic evaluation
results = await run_agentic_evaluation()
print("\nDetailed results:")
print(results)

Agentic RAG 评测输出

Results: 58/66 passed (87.9%)

很好！我们实现了显著改进，从 65.2%（naive）到 87.9%（agentic）——agentic RAG 方法带来了 22.7 个百分点的提升！

表现比较

agentic RAG 方法相对 naive RAG 基线显示出很大改进：

把这个循环应用到你的 RAG 系统

遵循这个系统方法来改进任何 RAG 系统：

1. 创建评测数据集：使用来自系统的真实查询，或用 LLM 生成合成数据。
2. 定义指标：选择与用例对齐的简单指标。保持聚焦。
3. 运行基线评测：测量当前表现并分析错误模式，以识别系统性失败。
4. 实施针对性改进：基于错误分析，改进检索（切分、混合搜索）、生成（prompts、模型），或尝试 agentic 方法。
5. 比较并迭代：对照基线测试改进。一次只改一件事，直到准确率满足业务要求。

Ragas 框架自动处理编排和结果汇总，让你聚焦于分析和改进，而不是构建评测基础设施。