rag-evaluation

RAG の評価とテスト

重要な指標を測定します。検索品質、回答の忠実性、エンドツーエンドのパフォーマンスです。

主な指標の概要

指標	測定内容	良好なスコア
コンテキストの適合率	検索されたドキュメントは関連性がありますか？	>0.8
コンテキストの再現率	関連するドキュメントをすべて取得できましたか？	>0.7
忠実性	回答はコンテキストに基づいていますか？	>0.9
回答の関連性	回答は質問に応えていますか？	>0.8
回答の正確性	回答は事実に基づいて正しいですか？	>0.8

RAGAS フレームワーク

RAGAS（Retrieval Augmented Generation Assessment）は、RAG 評価の標準です。

インストールとセットアップ

pip install ragas langchain_openai

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
    answer_correctness
)
from datasets import Dataset

評価データセットの作成

def create_eval_dataset(test_cases: list[dict]) -> Dataset:
    """
    Create evaluation dataset from test cases.

    Each test case needs:
    - question: The query
    - answer: Generated answer from your RAG
    - contexts: Retrieved documents (list of strings)
    - ground_truth: Expected correct answer (for some metrics)
    """
    return Dataset.from_dict({
        "question": [tc["question"] for tc in test_cases],
        "answer": [tc["answer"] for tc in test_cases],
        "contexts": [tc["contexts"] for tc in test_cases],
        "ground_truth": [tc.get("ground_truth", "") for tc in test_cases]
    })

# Example test cases
test_cases = [
    {
        "question": "What is the return policy?",
        "answer": "Items can be returned within 30 days with receipt.",
        "contexts": [
            "Our return policy allows returns within 30 days of purchase. A receipt is required.",
            "Refunds are processed within 5-7 business days."
        ],
        "ground_truth": "30-day return policy with receipt required"
    },
    # Add more test cases...
]

eval_dataset = create_eval_dataset(test_cases)

評価の実行

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)

# Run evaluation
results = evaluate(
    eval_dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall
    ],
    llm=ChatOpenAI(model="gpt-4"),
    embeddings=OpenAIEmbeddings()
)

# View results
print(results)
# {'faithfulness': 0.92, 'answer_relevancy': 0.88, ...}

# Detailed per-question results
df = results.to_pandas()
print(df)

各指標について

忠実性（回答は根拠に基づいていますか？）

回答が検索されたコンテキストから推測できるかを測定します。

from ragas.metrics import faithfulness

# Faithfulness checks:
# 1. Extracts claims from the answer
# 2. Verifies each claim against context
# 3. Score = supported claims / total claims

# Example:
# Answer: "The product costs $99 and ships free"
# Context: "Product price: $99. Standard shipping: $5"
# Faithfulness: 0.5 (only price claim is supported)

コンテキストの適合率（検索されたドキュメントは関連性がありますか？）

検索されたドキュメントが実際に質問と関連しているかを測定します。

from ragas.metrics import context_precision

# High precision: Retrieved docs are all relevant
# Low precision: Retrieved docs contain irrelevant noise

# Scored by checking if each context chunk is needed
# to answer the question

コンテキストの再現率（必要な情報をすべて取得できましたか？）

回答に必要な情報がすべて検索されたコンテキストに含まれているかを測定します。

from ragas.metrics import context_recall

# Requires ground_truth answer
# Checks if context contains info needed for ground truth

# High recall: Context has everything needed
# Low recall: Missing relevant documents

回答の関連性（質問に答えていますか？）

回答が元の質問にどの程度応えているかを測定します。

from ragas.metrics import answer_relevancy

# Generates questions from the answer
# Compares generated questions to original
# Higher similarity = more relevant answer

カスタム評価指標

検索ヒット率

def calculate_hit_rate(
    test_cases: list[dict],
    retriever,
    k: int = 5
) -> float:
    """
    Calculate retrieval hit rate.
    Hit = relevant doc in top-k results.
    """
    hits = 0
    for tc in test_cases:
        results = retriever.invoke(tc["question"])[:k]
        retrieved_texts = [r.page_content for r in results]

        # Check if any relevant doc was retrieved
        for relevant_doc in tc["relevant_docs"]:
            if any(relevant_doc in text for text in retrieved_texts):
                hits += 1
                break

    return hits / len(test_cases)

平均逆順位（MRR）

def calculate_mrr(
    test_cases: list[dict],
    retriever
) -> float:
    """
    Mean Reciprocal Rank.
    Higher = relevant docs appear earlier in results.
    """
    reciprocal_ranks = []

    for tc in test_cases:
        results = retriever.invoke(tc["question"])
        retrieved_texts = [r.page_content for r in results]

        # Find rank of first relevant doc
        for rank, text in enumerate(retrieved_texts, 1):
            if any(rel in text for rel in tc["relevant_docs"]):
                reciprocal_ranks.append(1 / rank)
                break
        else:
            reciprocal_ranks.append(0)

    return sum(reciprocal_ranks) / len(reciprocal_ranks)

レイテンシの追跡

import time
from dataclasses import dataclass

@dataclass
class RAGMetrics:
    retrieval_latency_ms: float
    generation_latency_ms: float
    total_latency_ms: float
    num_docs_retrieved: int
    num_tokens_generated: int

def measure_rag_performance(rag_chain, query: str) -> RAGMetrics:
    """Measure RAG pipeline performance."""

    # Measure retrieval
    start = time.perf_counter()
    docs = rag_chain.retriever.invoke(query)
    retrieval_time = (time.perf_counter() - start) * 1000

    # Measure generation
    start = time.perf_counter()
    response = rag_chain.generate(query, docs)
    generation_time = (time.perf_counter() - start) * 1000

    return RAGMetrics(
        retrieval_latency_ms=retrieval_time,
        generation_latency_ms=generation_time,
        total_latency_ms=retrieval_time + generation_time,
        num_docs_retrieved=len(docs),
        num_tokens_generated=len(response.split())
    )

テストスイートの構築

class RAGTestSuite:
    """Comprehensive RAG test suite."""

    def __init__(self, rag_chain, test_cases: list[dict]):
        self.rag = rag_chain
        self.test_cases = test_cases

    def run_all(self) -> dict:
        """Run all evaluation metrics."""
        results = {
            "ragas_metrics": self._run_ragas(),
            "retrieval_metrics": self._run_retrieval_metrics(),
            "latency_metrics": self._run_latency_metrics(),
            "failure_analysis": self._analyze_failures()
        }
        return results

    def _run_ragas(self) -> dict:
        # Generate answers for test cases
        eval_data = []
        for tc in self.test_cases:
            answer = self.rag.query(tc["question"])
            contexts = self.rag.retriever.invoke(tc["question"])
            eval_data.append({
                "question": tc["question"],
                "answer": answer,
                "contexts": [c.page_content for c in contexts],
                "ground_truth": tc.get("expected_answer", "")
            })

        dataset = Dataset.from_dict({
            "question": [d["question"] for d in eval_data],
            "answer": [d["answer"] for d in eval_data],
            "contexts": [d["contexts"] for d in eval_data],
            "ground_truth": [d["ground_truth"] for d in eval_data]
        })

        return evaluate(dataset, metrics=[
            faithfulness, answer_relevancy,
            context_precision, context_recall
        ])

    def _run_retrieval_metrics(self) -> dict:
        return {
            "hit_rate@5": calculate_hit_rate(self.test_cases, self.rag.retriever, k=5),
            "hit_rate@10": calculate_hit_rate(self.test_cases, self.rag.retriever, k=10),
            "mrr": calculate_mrr(self.test_cases, self.rag.retriever)
        }

    def _run_latency_metrics(self) -> dict:
        latencies = []
        for tc in self.test_cases[:10]:  # Sample for latency
            metrics = measure_rag_performance(self.rag, tc["question"])
            latencies.append(metrics)

        return {
            "avg_total_ms": sum(l.total_latency_ms for l in latencies) / len(latencies),
            "avg_retrieval_ms": sum(l.retrieval_latency_ms for l in latencies) / len(latencies),
            "p95_total_ms": sorted([l.total_latency_ms for l in latencies])[int(len(latencies) * 0.95)]
        }

    def _analyze_failures(self) -> list[dict]:
        """Identify and categorize failures."""
        failures = []
        for tc in self.test_cases:
            answer = self.rag.query(tc["question"])
            if tc.get("expected_answer") and tc["expected_answer"].lower() not in answer.lower():
                failures.append({
                    "question": tc["question"],
                    "expected": tc["expected_answer"],
                    "actual": answer,
                    "failure_type": self._classify_failure(tc, answer)
                })
        return failures

    def _classify_failure(self, tc: dict, answer: str) -> str:
        """Classify failure type."""
        contexts = self.rag.retriever.invoke(tc["question"])
        context_text = " ".join([c.page_content for c in contexts])

        if tc.get("expected_answer") and tc["expected_answer"] in context_text:
            return "generation_failure"  # Right docs, wrong answer
        else:
            return "retrieval_failure"  # Wrong docs retrieved

継続的な評価

# In production, log and monitor
import logging

logger = logging.getLogger("rag_eval")

def monitored_rag_query(rag, query: str) -> dict:
    """RAG query with monitoring."""
    start = time.perf_counter()

    # Get results
    docs = rag.retriever.invoke(query)
    answer = rag.generate(query, docs)

    # Log metrics
    logger.info({
        "query": query,
        "num_docs": len(docs),
        "latency_ms": (time.perf_counter() - start) * 1000,
        "answer_length": len(answer)
    })

    return {"answer": answer, "sources": docs}

ベストプラクティス

1. ゴールデンデータセットを構築する - 人手で検証した Q&A ペアを 50～100 件用意します
2. エッジケースをテストする - 結果が空の場合、曖昧なクエリ、対象領域外のクエリをテストします
3. CI/CD で自動化する - 変更のたびに評価を実行します
4. 経時的に追跡する - 指標の推移を監視します
5. クエリの種類ごとに分類する - ユースケースに応じて異なる指標を使用します