RAG Retrieval 評価指標：Recall / MRR / NDCG の計算方法和

RAG（Retrieval-Augmented Generation）システムを構築する際のリトリーバーの性能評価は、検索品質を確保する上で極めて重要です。本稿では、HolySheep AI を活用した実践的な評価指標の計算方法を詳しく解説します。

HolySheep vs 公式API vs 他リレーサービス比較表

比較項目	HolySheep AI	公式API	他リレー
為替レート	¥1 = $1	¥7.3 = $1	¥4-6 = $1
GPT-4.1 出力コスト	$8/MTok	$8/MTok	$10-15/MTok
Claude Sonnet 4.5	$15/MTok	$15/MTok	$18-25/MTok
DeepSeek V3.2	$0.42/MTok	$0.42/MTok	$0.8-1.2/MTok
レイテンシ	<50ms	100-300ms	80-200ms
支払い方法	WeChat Pay / Alipay対応	国際カードのみ	限定的
無料クレジット	登録時付与	$5〜$18	稀
base_url	api.holysheep.ai/v1	api.openai.com/v1	各自異なる

RAG評価指標の基礎知識

RAGシステムにおけるリトリーバーの性能評価には、主に3つの指標が使用されます。これらは情報検索の分野で標準的に用いられる指標であり、検索引擎の品質を定量的に把握できます。

1. Recall@K（再現率）

RelevantなドキュメントがTop-Kの中に含まれている割合を測定します。全関連ドキュメントをどれだけの確率で取得できるかを表します。

Recall@K = ( Relevantドキュメント ∧ Top-K内ドキュメント ) / Relevantドキュメント総数

2. MRR（Mean Reciprocal Rank）

各クエリにおける最初 Relevant ドキュメントの順位の逆数の平均を計算します。最初の結果の品質を重視する場合に有効です。

MRR = (1/N) * Σ(1/rank_i)

rank_i: i番目のクエリで最初に見つかったRelevantドキュメントの順位
RelevantドキュメントがTop-Kにない場合は0

3. NDCG@K（Normalized Discounted Cumulative Gain）

relevance の程度を考慮した評価指標です。上位の結果ほど重要視し、 relevance スコアに応じた重み付けを行います。

DCG@K = Σ(i=1 to K) [rel_i / log2(i+1)]

NDCG@K = DCG@K / IDCG@K
IDCG@K: 完璧にソートされた場合のDCG

実践的な実装コード

以下では、HolySheep AI のEmbedding APIを活用したRAG評価システムの実装例を示します。

評価指標計算モジュール

import numpy as np
from typing import List, Dict, Tuple
import requests

class RAGEvaluator:
    """
    RAG Retrieval 評価指標計算クラス
    Recall@K, MRR@K, NDCG@K を計算
    """
    
    HOLYSHEEP_API_URL = "https://api.holysheep.ai/v1/embeddings"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    def get_embedding(self, text: str, model: str = "text-embedding-3-large") -> List[float]:
        """HolySheep AIからEmbeddingを取得（<50msレイテンシ）"""
        response = requests.post(
            self.HOLYSHEEP_API_URL,
            headers=self.headers,
            json={"input": text, "model": model}
        )
        response.raise_for_status()
        return response.json()["data"][0]["embedding"]
    
    def cosine_similarity(self, a: List[float], b: List[float]) -> float:
        """コサイン類似度の計算"""
        a = np.array(a)
        b = np.array(b)
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
    
    def retrieve_top_k(self, query: str, corpus: List[str], k: int = 10) -> List[Tuple[int, float]]:
        """クエリに基づきTop-Kドキュメントを取得"""
        query_emb = self.get_embedding(query)
        similarities = []
        
        for idx, doc in enumerate(corpus):
            doc_emb = self.get_embedding(doc)
            sim = self.cosine_similarity(query_emb, doc_emb)
            similarities.append((idx, sim))
        
        # 類似度降順でソート
        similarities.sort(key=lambda x: x[1], reverse=True)
        return similarities[:k]
    
    def calculate_recall(self, retrieved: List[int], relevant: List[int], k: int) -> float:
        """Recall@K の計算"""
        retrieved_k = set(retrieved[:k])
        relevant_set = set(relevant)
        
        if len(relevant_set) == 0:
            return 0.0
        
        return len(retrieved_k & relevant_set) / len(relevant_set)
    
    def calculate_mrr(self, retrieved: List[int], relevant: List[int], k: int) -> float:
        """MRR@K の計算"""
        for i, doc_idx in enumerate(retrieved[:k]):
            if doc_idx in relevant:
                return 1.0 / (i + 1)
        return 0.0
    
    def calculate_dcg(self, retrieved: List[int], relevance_scores: Dict[int, float], k: int) -> float:
        """DCG@K の計算"""
        dcg = 0.0
        for i, doc_idx in enumerate(retrieved[:k]):
            rel = relevance_scores.get(doc_idx, 0.0)
            dcg += rel / np.log2(i + 2)  # i+2 because log2(1)=0
        return dcg
    
    def calculate_ndcg(self, retrieved: List[int], relevance_scores: Dict[int, float], k: int) -> float:
        """NDCG@K の計算"""
        dcg = self.calculate_dcg(retrieved, relevance_scores, k)
        
        # IDCG: Relevantドキュメントを完璧にソートした場合
        sorted_relevance = sorted(relevance_scores.values(), reverse=True)
        idcg = sum(rel / np.log2(i + 2) for i, rel in enumerate(sorted_relevance[:k]))
        
        if idcg == 0:
            return 0.0
        
        return dcg / idcg
    
    def evaluate_retrieval(
        self,
        queries: List[str],
        corpus: List[str],
        relevant_docs: List[List[int]],
        relevance_scores: List[Dict[int, float]],
        k: int = 10
    ) -> Dict[str, float]:
        """全評価指標を一括計算"""
        recalls, mrrs, ndcgs = [], [], []
        
        for query, relevant, scores in zip(queries, relevant_docs, relevance_scores):
            retrieved = self.retrieve_top_k(query, corpus, k)
            retrieved_indices = [idx for idx, _ in retrieved]
            
            recalls.append(self.calculate_recall(retrieved_indices, relevant, k))
            mrrs.append(self.calculate_mrr(retrieved_indices, relevant, k))
            ndcgs.append(self.calculate_ndcg(retrieved_indices, scores, k))
        
        return {
            f"Recall@{k}": np.mean(recalls),
            f"MRR@{k}": np.mean(mrrs),
            f"NDCG@{k}": np.mean(ndcgs)
        }


使用例
if __name__ == "__main__":
    API_KEY = "YOUR_HOLYSHEEP_API_KEY"
    evaluator = RAGEvaluator(API_KEY)
    
    # テストデータ
    corpus = [
        "DeepSeekは中国のAI企業です",
        "GPT-4はOpenAIの開発した大規模言語モデルです",
        "ClaudeはAnthropic社のAIアシスタントです",
        "HolySheep AIは高性能なAPIリレーサービスを提供しています",
        "RAGは検索拡張生成の略称です"
    ]
    
    queries = ["AI企業について", "言語モデルの種類"]
    relevant_docs = [[0, 3], [1, 2]]  # 各クエリの関連ドキュメントINDEX
    relevance_scores = [
        {0: 1.0, 3: 0.8},  # クエリ1: 重み付き relevance
        {1: 1.0, 2: 0.9}   # クエリ2: 重み付き relevance
    ]
    
    results = evaluator.evaluate_retrieval(queries, corpus, relevant_docs, relevance_scores, k=3)
    print("評価結果:", results)

バッチ評価ユーティリティ

import json
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor

class BatchRAGEvaluator:
    """
    大量データに対する効率的なRAG評価
    HolySheep APIの<50msレイテンシを活かした高速処理
    """
    
    def __init__(self, api_key: str, max_workers: int = 10):
        self.api_key = api_key
        self.max_workers = max_workers
        self.evaluator = RAGEvaluator(api_key)
    
    def evaluate_dataset(
        self,
        dataset_path: str,
        output_path: str,
        k_values: List[int] = [5, 10, 20]
    ) -> Dict:
        """データセットファイルから一括評価を実行"""
        
        with open(dataset_path, 'r', encoding='utf-8') as f:
            dataset = json.load(f)
        
        all_results = {f"Recall@{k}": [] for k in k_values}
        all_results.update({f"MRR@{k}": [] for k in k_values})
        all_results.update({f"NDCG@{k}": [] for k in k_values})
        
        corpus = dataset["corpus"]
        
        def process_query(q_data):
            query = q_data["query"]
            relevant = q_data["relevant_docs"]
            scores = q_data.get("relevance_scores", {i: 1.0 for i in relevant})
            
            retrieved = self.evaluator.retrieve_top_k(query, corpus, max(k_values))
            retrieved_indices = [idx for idx, _ in retrieved]
            
            result = {}
            for k in k_values:
                result[f"Recall@{k}"] = self.evaluator.calculate_recall(
                    retrieved_indices, relevant, k
                )
                result[f"MRR@{k}"] = self.evaluator.calculate_mrr(
                    retrieved_indices, relevant, k
                )
                result[f"NDCG@{k}"] = self.evaluator.calculate_ndcg(
                    retrieved_indices, scores, k
                )
            return result
        
        # 並列処理で高速化
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            results = list(executor.map(process_query, dataset["queries"]))
        
        # 結果を集計
        for r in results:
            for key, value in r.items():
                all_results[key].append(value)
        
        # 平均値を計算
        summary = {
            key: round(np.mean(values), 4) 
            for key, values in all_results.items()
        }
        
        # 詳細結果を保存
        output = {
            "timestamp": datetime.now().isoformat(),
            "summary": summary,
            "detailed_results": all_results
        }
        
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(output, f, indent=2, ensure_ascii=False)
        
        return summary


サンプルデータセット生成
def generate_sample_dataset() -> dict:
    """評価用のサンプルデータセットを生成"""
    
    corpus = [
        "機械学習は人工智能の一分野です",
        "自然言語処理はテキストデータを扱う技術です",
        "ベクトルデータベースはEmbeddingを保存します",
        "RAGはRetrieval-Augmented Generationの略です",
        "TransformerはAttention機構を使用したモデルです",
        "Fine-tuningは事前学習済みモデルを微調整する方法です",
        "Embeddingはテキストをベクトル表現に変換することです",
        "HolySheep AIはapi.holysheep.aiを通じてAPIを提供します",
        "DeepSeek V3.2は$0.42/MTokという低コストモデルです",
        "ClaudeはAnthropic社の安全性重視AIです"
    ]
    
    queries = [
        {
            "query": "Embeddingとは何か？",
            "relevant_docs": [1, 6],
            "relevance_scores": {1: 0.9, 6: 1.0}
        },
        {
            "query": "RAGシステムの構築方法",
            "relevant_docs": [0, 2, 3],
            "relevance_scores": {0: 0.7, 2: 0.8, 3: 1.0}
        },
        {
            "query": "低コストなAIモデル",
            "relevant_docs": [7, 8],
            "relevance_scores": {7: 0.8, 8: 1.0}
        }
    ]
    
    return {"corpus": corpus, "queries": queries}


メイン実行
if __name__ == "__main__":
    API_KEY = "YOUR_HOLYSHEEP_API_KEY"
    
    # データセット生成
    dataset = generate_sample_dataset()
    
    with open("rag_eval_dataset.json", 'w', encoding='utf-8') as f:
        json.dump(dataset, f, indent=2, ensure_ascii=False)
    
    # バッチ評価実行
    batch_evaluator = BatchRAGEvaluator(API_KEY, max_workers=5)
    results = batch_evaluator.evaluate_dataset(
        "rag_eval_dataset.json",
        "rag_eval_results.json",
        k_values=[3, 5, 10]
    )
    
    print("=" * 50)
    print("RAG Evaluation Results")
    print("=" * 50)
    for metric, value in results.items():
        print(f"{metric}: {value:.4f}")
    print("=" * 50)

評価指標の結果解釈とベストプラクティス

Recall@K ≥ 0.8: リトリーバーが関連ドキュメントの大部分を捕捉できている状態
MRR@K ≥ 0.6: 最初の結果にRelevantドキュメントが来る確率が高い
NDCG@K ≥ 0.7: relevance の質も加味した良好な検索品質

指標選択の指針

タスクに応じて評価指標を使い分けることが重要です。事実確認重視のQAシステムでは MRR を、包括的な情報検索では NDCG を、医療・法務などの高リスク領域では Recall を重視すべきです。

HolySheep AI でのコスト最適化

RAG評価を大量実行する場合、APIコストが重要な因子となります。HolySheep AI では ¥1=$1 の為替レートを提供しており、特にDeepSeek V3.2 ($0.42/MTok) を利用することで、評価処理全体のコストを大幅に削減できます。

# コスト比較計算
def calculate_evaluation_cost(
    num_queries: int,
    num_corpus_docs: int,
    embedding_model: str = "text-embedding-3-large",
    avg_text_length: int = 500
) -> Dict[str, float]:
    """
    RAG評価のAPIコストを見積もり
    """
    # トークン估算（簡易）
    tokens_per_doc = avg_text_length // 4  # 約1トークン=4文字
    total_query_tokens = num_queries * tokens_per_doc * 2  # クエリ+応答
    total_corpus_tokens = num_corpus_docs * tokens_per_doc
    
    # HolySheep AI の料金
    HOLYSHEEP_PRICING = {
        "text-embedding-3-large": 0.00013,  # $0.00013 / token
        "text-embedding-3-small": 0.00002,   # $0.00002 / token
    }
    
    # 公式API pricing（比較用）
    OPENAI_PRICING = {
        "text-embedding-3-large": 0.000195,  # $0.000195 / token
    }
    
    price_per_token = HOLYSHEEP_PRICING[embedding_model]
    openai_price = OPENAI_PRICING["text-embedding-3-large"]
    
    # 総コスト計算
    total_tokens = total_query_tokens + total_corpus_tokens
    
    holysheep_cost = total_tokens * price_per_token
    openai_cost = total_tokens * openai_price
    
    # ¥1=$1 で計算（HolySheep）、¥7.3=$1（公式）
    holysheep_yen = holysheep_cost  # ドル即为円
    openai_yen = openai_cost * 7.3
    
    return {
        "total_tokens": total_tokens,
        "holysheep_cost_usd": round(holysheep_cost, 4),
        "openai_cost_usd": round(openai_cost, 4),
        "holysheep_cost_jpy": round(holysheep_yen, 2),
        "openai_cost_jpy": round(openai_yen, 2),
        "savings_percent": round((1 - holysheep_cost / openai_cost) * 100, 1)
    }


使用例
if __name__ == "__main__":
    # 100クエリ、1000ドキュメントの評価
    costs = calculate_evaluation_cost(
        num_queries=100,
        num_corpus_docs=1000,
        embedding_model="text-embedding-3-large"
    )
    
    print("=" * 60)
    print("RAG Evaluation Cost Estimation")
    print("=" * 60)
    print(f"総トークン数: {costs['total_tokens']:,} tokens")
    print("-" * 60)
    print(f"HolySheep AI: ${costs['holysheep_cost_usd']} (¥{costs['holysheep_cost_jpy']})")
    print(f"公式API:      ${costs['openai_cost_usd']} (¥{costs['openai_cost_jpy']})")
    print("-" * 60)
    print(f"コスト削減:   {costs['savings_percent']}%")
    print("=" * 60)

よくあるエラーと対処法

エラー1：API Key認証エラー (401 Unauthorized)

# ❌ 誤ったbase_urlの使用
response = requests.post(
    "https://api.openai.com/v1/embeddings",  # 絶対に使用禁止
    headers={"Authorization": f"Bearer {api_key}"},
    json={"input": text, "model": "text-embedding-3-large"}
)

✅ 正しい実装（HolySheep AI）
response = requests.post(
    "https://api.holysheep.ai/v1/embeddings",  # 正しいエンドポイント
    headers={"Authorization": f"Bearer {api_key}"},
    json={"input": text, "model": "text-embedding-3-large"}
)

解決方法：base_url を必ず https://api.holysheep.ai/v1 に設定してください。API Key は HolySheep AI ダッシュボードから取得できます。

エラー2：Rate Limit Exceeded (429)

import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session_with_retry(max_retries: int = 3) -> requests.Session:
    """リトライ機能付きのセッションを作成"""
    session = requests.Session()
    
    retry_strategy = Retry(
        total=max_retries,
        back
関連リソース
📚 AI API 記事一覧
💰 料金を見る
📖 開発者ドキュメント
🚀 無料登録
関連記事
Cohere Rerank API を RAG パイプラインに組み込む 完全ガイド
RAG 幻觉控制：引用溯源与答案可信度验证
Apple MLX フレームワークで Mac ローカルに大規模言語モデルを導入する方法

RAG Retrieval 評価指標：Recall / MRR / NDCG の計算方法和

HolySheep vs 公式API vs 他リレーサービス比較表

RAG評価指標の基礎知識

1. Recall@K（再現率）

2. MRR（Mean Reciprocal Rank）

rank_i: i番目のクエリで最初に見つかったRelevantドキュメントの順位

`RelevantドキュメントがTop-Kにない場合は0`

3. NDCG@K（Normalized Discounted Cumulative Gain）

`IDCG@K: 完璧にソートされた場合のDCG`

実践的な実装コード

評価指標計算モジュール

使用例

バッチ評価ユーティリティ

サンプルデータセット生成

メイン実行

評価指標の結果解釈とベストプラクティス

指標選択の指針

HolySheep AI でのコスト最適化

使用例

よくあるエラーと対処法

エラー1：API Key認証エラー (401 Unauthorized)

✅ 正しい実装（HolySheep AI）

エラー2：Rate Limit Exceeded (429)

関連リソース

関連記事

HolySheep vs 公式API vs 他リレーサービス 比較表

RAG評価指標の基礎知識

1. Recall@K（再現率）

2. MRR（Mean Reciprocal Rank）

rank_i: i番目のクエリで最初に見つかったRelevantドキュメントの順位

RelevantドキュメントがTop-Kにない場合は0

3. NDCG@K（Normalized Discounted Cumulative Gain）

IDCG@K: 完璧にソートされた場合のDCG

実践的な実装コード

評価指標計算モジュール

使用例

バッチ評価ユーティリティ

サンプルデータセット生成

メイン実行

評価指標の結果解釈とベストプラクティス

指標選択の指針

HolySheep AI でのコスト最適化

使用例

よくあるエラーと対処法

エラー1：API Key認証エラー (401 Unauthorized)

✅ 正しい実装（HolySheep AI）

エラー2：Rate Limit Exceeded (429)

関連リソース

関連記事

🔥 HolySheep AIを使ってみる

HolySheep vs 公式API vs 他リレーサービス比較表

`RelevantドキュメントがTop-Kにない場合は0`

`IDCG@K: 完璧にソートされた場合のDCG`