RAGAS完全ガイド：RAGシステムをどのように定量的に評価するか

RAGパイプライン全体：

[ユーザーの質問]
        |
[検索ステップ] <- Context Precision、Context Recallで評価
        |
[生成ステップ] <- Faithfulness、Answer Relevancyで評価
        |
[最終回答]

質問: 「この製品のバッテリー寿命はどのくらいですか？」
コンテキスト: 「バッテリーは最大12時間持続します。充電時間は2時間です。」
回答: 「バッテリーは12時間持続し、防水機能もあります。」

Claim分析：
- 「バッテリーは12時間持続」→ コンテキストで支持される ○
- 「防水機能もある」→ コンテキストにない ×

Faithfulness = 1/2 = 0.5（低い！ハルシネーション発生）

元の質問: 「バッテリー寿命はどのくらいですか？」
生成された回答: 「この製品はプレミアム素材で作られており、様々な色があります。」

Synthetic質問の生成：
- 「この製品はどんな素材で作られていますか？」
- 「どんな色がありますか？」

元の質問との類似度: 非常に低い
Answer Relevancy ≈ 0.1（低い！回答が質問を無視している）

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

# 評価データセットの準備
eval_data = {
    "question": [
        "この製品のバッテリー寿命はどのくらいですか？",
        "返品ポリシーはどうなっていますか？",
        "配送にはどのくらいかかりますか？"
    ],
    "answer": [
        "バッテリーは最大12時間持続します。",
        "ご購入から30日以内に返品申請が可能です。",
        "通常配送は3〜5営業日かかります。"
    ],
    "contexts": [
        [
            "バッテリーは最大12時間持続します。充電時間は2時間です。",
            "製品サイズは15cm x 10cm x 2cmです。"
        ],
        [
            "お客様はご購入から30日以内に返品を申請できます。",
            "返金は営業日5〜7日以内に処理されます。"
        ],
        [
            "通常配送：3〜5営業日、速達配送：1〜2営業日",
            "3,000円未満のご注文には送料500円が加算されます。"
        ]
    ],
    "ground_truth": [
        "バッテリー寿命は最大12時間で、充電時間は2時間です。",
        "返品はご購入から30日以内に申請可能で、5〜7営業日以内に処理されます。",
        "通常配送は3〜5営業日、速達配送は1〜2営業日かかります。"
    ]
}

dataset = Dataset.from_dict(eval_data)

# 評価の実行
results = evaluate(
    dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
    ]
)

print(results)
# 出力例：
# {
#   'faithfulness': 0.95,
#   'answer_relevancy': 0.88,
#   'context_precision': 0.82,
#   'context_recall': 0.91
# }

df = results.to_pandas()
print(df.to_string())

from ragas.llms import LangchainLLMWrapper
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from ragas.embeddings import LangchainEmbeddingsWrapper

# コスト削減のためgpt-4o-miniを使用
judge_llm = LangchainLLMWrapper(
    ChatOpenAI(model="gpt-4o-mini", temperature=0)
)

embeddings = LangchainEmbeddingsWrapper(OpenAIEmbeddings())

results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    llm=judge_llm,
    embeddings=embeddings,
)

# evaluate_rag.py - CI/CDで実行される評価スクリプト

import json
import sys
from datetime import datetime
from pathlib import Path
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset


def load_test_dataset(path: str) -> Dataset:
    """保存されたテストデータセットを読み込む"""
    with open(path, 'r', encoding='utf-8') as f:
        data = json.load(f)
    return Dataset.from_dict(data)


def run_rag_on_testset(testset: Dataset, rag_chain) -> Dataset:
    """テストセットの各質問をRAGで実行してanswersとcontextsを収集"""
    answers = []
    contexts_list = []

    for question in testset["question"]:
        result = rag_chain.invoke({"query": question})
        answers.append(result["result"])
        contexts_list.append([
            doc.page_content
            for doc in result["source_documents"]
        ])

    return testset.add_column("answer", answers).add_column("contexts", contexts_list)


def evaluate_and_gate(
    dataset: Dataset,
    thresholds: dict,
    save_path: str = None
) -> bool:
    """
    評価を実行し、閾値に基づいて合格/不合格を判定。
    """
    results = evaluate(
        dataset,
        metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    )

    print("\n" + "="*60)
    print(f"RAG評価結果 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("="*60)

    all_passed = True
    for metric_name, threshold in thresholds.items():
        score = results[metric_name]
        passed = score >= threshold
        status = "PASS" if passed else "FAIL"
        print(f"{metric_name:25s}: {score:.3f} (閾値: {threshold}) [{status}]")
        if not passed:
            all_passed = False

    if save_path:
        result_data = {
            "timestamp": datetime.now().isoformat(),
            "scores": {k: float(v) for k, v in results.items()},
            "passed": all_passed
        }
        Path(save_path).parent.mkdir(parents=True, exist_ok=True)
        with open(save_path, 'a') as f:
            f.write(json.dumps(result_data) + "\n")

    return all_passed


if __name__ == "__main__":
    testset = load_test_dataset("tests/rag_testset.json")

    thresholds = {
        "faithfulness": 0.80,
        "answer_relevancy": 0.75,
        "context_precision": 0.70,
        "context_recall": 0.75,
    }

    passed = evaluate_and_gate(
        testset,
        thresholds=thresholds,
        save_path="results/rag_eval_history.jsonl"
    )

    sys.exit(0 if passed else 1)  # CI/CDで閾値未満ならビルドを失敗させる

# .github/workflows/rag-evaluation.yml
name: RAG Quality Gate

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  evaluate-rag:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: pip install ragas langchain openai datasets
      - name: Run RAG Evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: python evaluate_rag.py

from ragas.testset.generator import TestsetGenerator
from ragas.testset.evolutions import simple, reasoning, multi_context
from langchain_community.document_loaders import DirectoryLoader

# ドキュメントの読み込み
loader = DirectoryLoader("./docs", glob="**/*.md")
documents = loader.load()

# テストセットジェネレーターの初期化
generator = TestsetGenerator.with_openai()

# テストセットの生成（質問タイプを指定可能）
testset = generator.generate_with_langchain_docs(
    documents,
    test_size=50,
    distributions={
        simple: 0.5,        # シンプルな事実質問 50%
        reasoning: 0.25,    # 推論が必要な質問 25%
        multi_context: 0.25 # 複数ドキュメントが必要な質問 25%
    }
)

# ファイルに保存
testset_df = testset.to_pandas()
testset_df.to_json("tests/rag_testset.json", orient='records', force_ascii=False)

print(f"生成された質問数: {len(testset_df)}")
print("\nサンプル質問:")
print(testset_df[['question', 'ground_truth']].head())

1. ベースラインを測定（RAGASを実行）
        |
2. 最も低いスコアの指標を特定
        |
3. その指標を改善（チャンキング/検索/プロンプトを修正）
        |
4. 再度測定して改善を確認
        |
5. CI/CDにQuality Gateを追加
        |
6. コード変更のたびに自動評価