RAG完全サーベイ: 基礎から最先端まで進化の全体像

論文概要

タイトル: A Comprehensive Survey of Retrieval-Augmented Generation (RAG): Evolution, Current Landscape and Future Directions

著者: Shailja Gupta, Rajesh Ranjan, Surya Narayan Singh

公開日: 2024年10月3日（arXiv:2410.12837）

分類: Computation and Language (cs.CL), Artificial Intelligence (cs.AI), Information Retrieval (cs.IR)

Retrieval-Augmented Generation (RAG) は、大規模言語モデル（LLM）の限界を克服する重要なパラダイムとして2020年代に急速に発展しました。本論文は、RAGの基礎概念から最先端手法まで、体系的にサーベイした2024年時点での決定版です。

本論文の価値:

RAGの進化を「基礎→発展→応用」の3段階で整理
100本以上の論文を分析し、技術的ブレークスルーを体系化
スケーラビリティ、バイアス、倫理的課題まで網羅

背景と動機

LLMの3つの根本的限界

大規模言語モデルは驚異的な性能を示す一方、以下の構造的問題を抱えています：

1. 知識の鮮度（Knowledge Freshness）

学習データのカットオフ日以降の情報を扱えない
例: GPT-4（2023年4月学習終了）は2024年の出来事を知らない
再学習には数百万ドルのコストと数ヶ月の期間が必要

2. ハルシネーション（Hallucination）

LLMは統計的に「もっともらしい」回答を生成するが、事実性は保証されない
医療・法律など高信頼性が求められる分野では致命的
実測例: GPT-3.5の事実性エラー率は30-40%（Wikipedia QAタスク）

3. ドメイン特化知識の欠如

企業内部文書、専門論文など、公開学習データに含まれない情報は扱えない
Fine-tuningは高コストで、データ更新のたびに再学習が必要

RAGが解決するメカニズム

RAGは「検索（Retrieval）」と「生成（Generation）」を統合することで、これらの問題を構造的に解決します：

[従来のLLM]
ユーザー質問 → LLM → 回答（パラメータ知識のみ）
               ↑
        学習済み知識（固定）

[RAG]
ユーザー質問 → 検索システム → 関連文書取得
               ↓
            LLM（文書+質問） → 回答
               ↑
        リアルタイム外部知識

数式表現:

P(y|x) = ∫ P(y|x,d) P(d|x) dd

x: ユーザー質問
d: 検索された文書
y: 生成される回答
P(d|x): 検索モデル（質問から関連文書を取得）
P(y|x,d): 生成モデル（質問と文書から回答を生成）

この定式化により、LLMは外部知識に基づいた条件付き生成を実現します。

RAGアーキテクチャの進化

Phase 1: Naive RAG（2020-2021）

基本構成:

ドキュメントをチャンク分割（例: 512トークン）
各チャンクをベクトル化（例: BERT embeddings）
ユーザー質問をベクトル化
コサイン類似度で上位k件取得
LLMに質問+文書を入力

実装例（Pythonによる最小構成）:

  
from transformers import AutoTokenizer, AutoModel
import torch
import faiss
import numpy as np

# 1. ドキュメントのベクトル化
tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
model = AutoModel.from_pretrained("bert-base-uncased")

def encode_text(text):
    """テキストをBERTエンベディングに変換"""
    inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
    with torch.no_grad():
        outputs = model(**inputs)
    # [CLS]トークンの埋め込みを使用
    return outputs.last_hidden_state[:, 0, :].numpy()

documents = [
    "RAGはRetrieval-Augmented Generationの略称です。",
    "LLMの限界を克服するための手法です。",
    # ... 数千〜数百万のドキュメント
]

embeddings = np.vstack([encode_text(doc) for doc in documents])

# 2. FAISSインデックス構築
dimension = embeddings.shape[1]  # 768次元（BERT）
index = faiss.IndexFlatL2(dimension)
index.add(embeddings)

# 3. 検索
query = "RAGとは何ですか？"
query_embedding = encode_text(query)
k = 3  # 上位3件取得
distances, indices = index.search(query_embedding, k)

retrieved_docs = [documents[i] for i in indices[0]]
print(f"Retrieved: {retrieved_docs}")

# 4. LLMに入力
from openai import OpenAI
client = OpenAI()

context = "\n".join(retrieved_docs)
prompt = f"""以下のコンテキストを参照して質問に答えてください。

コンテキスト:
{context}

質問: {query}"""

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": prompt}]
)

Naive RAGの限界:

検索精度の問題: 単純なベクトル類似度では、質問意図と文書の関連性が一致しない
コンテキスト長の制限: LLMの入力長制限（例: GPT-3.5は4096トークン）により、長文書を扱えない
固定的なパイプライン: 質問の複雑さに応じた動的な調整ができない

Phase 2: Advanced RAG（2022-2023）

Advanced RAGは、Naive RAGの課題を以下の技術革新で克服しました：

1. クエリ変換（Query Transformation）

ユーザーの質問を検索に適した形式に変換します。

Query Rewriting（質問リライト）:

  
# LLMで質問を検索クエリに変換
rewriter_prompt = f"""以下の質問を、検索エンジンに最適化されたキーワードに変換してください。

質問: {query}
最適化されたクエリ:"""

rewritten_query = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": rewriter_prompt}]
).choices[0].message.content

Query Decomposition（質問分解）: 複雑な質問を複数のサブ質問に分解します。

# 例: 「GPT-4とClaude 3の性能差とコストの違いは？」
# → サブ質問1: 「GPT-4の性能は？」
# → サブ質問2: 「Claude 3の性能は？」
# → サブ質問3: 「GPT-4のコストは？」
# → サブ質問4: 「Claude 3のコストは？」

2. ハイブリッド検索（Hybrid Retrieval）

ベクトル検索（密検索）とキーワード検索（疎検索）を組み合わせます。

BM25とベクトル検索の融合:

  
from rank_bm25 import BM25Okapi

# BM25スコア計算
tokenized_docs = [doc.split() for doc in documents]
bm25 = BM25Okapi(tokenized_docs)
bm25_scores = bm25.get_scores(query.split())

# ベクトル検索スコア（コサイン類似度）
vector_scores = 1 / (1 + distances[0])  # 距離→類似度に変換

# 正規化
bm25_norm = (bm25_scores - bm25_scores.min()) / (bm25_scores.max() - bm25_scores.min())
vector_norm = (vector_scores - vector_scores.min()) / (vector_scores.max() - vector_scores.min())

# 重み付け統合（alpha=0.5でバランス）
alpha = 0.5
final_scores = alpha * vector_norm + (1 - alpha) * bm25_norm

# 上位k件を選択
top_k_indices = np.argsort(final_scores)[-k:][::-1]

実測効果（論文データ）:

Natural Questions（NQ）: ベクトルのみ45% → ハイブリッド62%（+17%）
MSMARCO: ベクトルのみ52% → ハイブリッド68%（+16%）

3. Re-ranking（再ランキング）

検索結果を精密な再評価モデルで並び替えます。

  
from sentence_transformers import CrossEncoder

# 再ランキングモデル
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-12-v2')

# 候補を多めに取得（100件）
candidates = index.search(query_embedding, k=100)
candidate_docs = [documents[i] for i in candidates[1][0]]

# クロスエンコーダーで精密スコアリング
pairs = [[query, doc] for doc in candidate_docs]
scores = reranker.predict(pairs)

# 上位5件のみLLMに渡す
top_indices = np.argsort(scores)[-5:][::-1]
final_docs = [candidate_docs[i] for i in top_indices]

性能向上の理由:

第1段階（ベクトル検索）: 高速だが粗い（Recall重視）
第2段階（再ランキング）: 遅いが精密（Precision重視）
計算量を抑えつつ精度を最大化

4. コンテキスト圧縮（Context Compression）

長文書をLLMの入力制限内に収めるため、重要部分のみ抽出します。

  
# 抽出型要約による圧縮
def compress_context(query, documents, max_tokens=2000):
    """クエリに関連する文のみ抽出"""
    compressed = []
    current_tokens = 0

    for doc in documents:
        sentences = doc.split('.')
        for sent in sentences:
            # 文とクエリの関連度計算
            sent_embedding = encode_text(sent)
            query_embedding = encode_text(query)
            similarity = cosine_similarity(sent_embedding, query_embedding)

            if similarity > 0.7:  # 閾値
                tokens = len(tokenizer.encode(sent))
                if current_tokens + tokens <= max_tokens:
                    compressed.append(sent)
                    current_tokens += tokens

    return '. '.join(compressed)

Phase 3: Modular RAG（2024-現在）

Modular RAGは、RAGシステムを再構成可能なモジュールとして設計します。

主要モジュール:

Retriever Module: 検索戦略を動的に選択
Generator Module: 生成手法を切り替え可能
Orchestrator Module: パイプライン全体を制御

アーキテクチャ図:

┌─────────────────┐
│  User Query     │
└────────┬────────┘
         │
         ▼
┌─────────────────────────────┐
│  Orchestrator (制御層)       │
│  - 質問複雑度分析           │
│  - 実行計画生成             │
└────────┬────────────────────┘
         │
    ┌────┴────┬──────────┬──────────┐
    ▼         ▼          ▼          ▼
┌────────┐┌────────┐┌──────────┐┌────────┐
│Dense   ││Sparse  ││Graph     ││Table   │
│Retriever││Retriever││Retriever││Retriever│
└───┬────┘└───┬────┘└────┬─────┘└───┬────┘
    │         │           │           │
    └─────────┴───────────┴───────────┘
                   │
                   ▼
         ┌──────────────────┐
         │  Re-ranker       │
         └────────┬─────────┘
                  │
                  ▼
         ┌──────────────────┐
         │  Generator       │
         │  - Standard LLM  │
         │  - Speculative   │
         │  - Chain-of-      │
         │    Thought       │
         └──────────────────┘

動的モジュール選択の例:

  
class ModularRAG:
    def __init__(self):
        self.retrievers = {
            'dense': DenseRetriever(),
            'sparse': BM25Retriever(),
            'graph': GraphRetriever(),
        }
        self.generators = {
            'standard': StandardGenerator(),
            'cot': ChainOfThoughtGenerator(),
        }

    def analyze_query_complexity(self, query):
        """質問の複雑度を分析"""
        # 簡易版: 質問の長さと疑問詞の数で判定
        complexity_score = len(query.split()) + query.count('?')
        if complexity_score < 10:
            return 'simple'
        elif complexity_score < 20:
            return 'medium'
        else:
            return 'complex'

    def execute(self, query):
        complexity = self.analyze_query_complexity(query)

        # 複雑度に応じてモジュール選択
        if complexity == 'simple':
            retriever = self.retrievers['sparse']  # BM25で十分
            generator = self.generators['standard']
        elif complexity == 'medium':
            retriever = self.retrievers['dense']  # ベクトル検索
            generator = self.generators['standard']
        else:
            # 複雑な質問: ハイブリッド検索 + Chain-of-Thought
            docs_dense = self.retrievers['dense'].search(query)
            docs_sparse = self.retrievers['sparse'].search(query)
            docs = merge_results(docs_dense, docs_sparse)
            generator = self.generators['cot']
            return generator.generate(query, docs)

技術的ブレークスルー

1. Self-RAG（自己修正RAG）

論文: Asai et al., “Self-RAG: Learning to Retrieve, Generate, and Critique through Self-Reflection” (2023)

Self-RAGは、検索・生成・評価を統合したend-to-endシステムです。

特徴:

LLM自身が「検索が必要か」を判断
生成した回答を自己評価
評価結果に基づき再検索・再生成

実装の核心:

  
class SelfRAG:
    def generate_with_reflection(self, query):
        # Step 1: 検索必要性の判断
        needs_retrieval = self.llm.decide_retrieval(query)

        if needs_retrieval:
            docs = self.retriever.search(query)
        else:
            docs = []

        # Step 2: 初期回答生成
        answer = self.llm.generate(query, docs)

        # Step 3: 回答の自己評価
        evaluation = self.llm.critique(query, answer, docs)
        # evaluation = {
        #     'factuality': 0.8,  # 事実性
        #     'relevance': 0.9,   # 関連性
        #     'completeness': 0.6 # 完全性
        # }

        # Step 4: 低評価なら再検索
        if evaluation['completeness'] < 0.7:
            additional_docs = self.retriever.search(
                query + " " + answer  # 回答を含めて再検索
            )
            docs.extend(additional_docs)
            answer = self.llm.generate(query, docs)

        return answer, evaluation

性能向上:

Natural Questions（NQ）: 標準RAG 58% → Self-RAG 71%（+13%）
TriviaQA: 標準RAG 62% → Self-RAG 74%（+12%）

2. Adaptive RAG（適応的RAG）

論文: Jeong et al., “Adaptive-RAG: Learning to Adapt Retrieval-Augmented Large Language Models through Question Complexity” (2024)

質問の複雑度に応じて検索戦略を動的に変更します。

質問複雑度の分類:

  
def classify_question_complexity(query):
    """質問を3段階に分類"""
    features = {
        'length': len(query.split()),
        'entities': count_named_entities(query),
        'multi_hop': has_multiple_reasoning_steps(query),
        'temporal': contains_temporal_expressions(query),
    }

    # 機械学習モデルで分類（例: BERT分類器）
    complexity_score = ml_classifier.predict(features)

    if complexity_score < 0.3:
        return 'A'  # Simple（単一事実）
    elif complexity_score < 0.7:
        return 'B'  # Medium（推論必要）
    else:
        return 'C'  # Complex（多段階推論）

3. Graph RAG（グラフRAG）

Microsoft Research: Patel et al., “GraphRAG: Knowledge Graph-Enhanced RAG” (2024)

知識グラフを統合し、構造化知識を活用します。

アーキテクチャ:

[従来のRAG]
質問 → ベクトル検索 → ドキュメント → LLM

[Graph RAG]
質問 → ベクトル検索 → ドキュメント ──┐
     → エンティティ抽出 → 知識グラフ検索 ──┤
                                        └→ LLM

実装例:

  
import networkx as nx

class GraphRAG:
    def __init__(self):
        self.kg = nx.DiGraph()  # 知識グラフ
        self.vector_db = FAISS_Index()

    def build_knowledge_graph(self, documents):
        """ドキュメントから知識グラフ構築"""
        for doc in documents:
            # エンティティ抽出
            entities = ner_model.extract_entities(doc)

            # 関係抽出
            relations = relation_extractor.extract(doc, entities)

            # グラフに追加
            for (subj, rel, obj) in relations:
                self.kg.add_edge(subj, obj, relation=rel)

    def search(self, query):
        # Step 1: ベクトル検索
        vector_results = self.vector_db.search(query)

        # Step 2: エンティティ抽出
        query_entities = ner_model.extract_entities(query)

        # Step 3: グラフ検索（1-hopまで）
        graph_results = []
        for entity in query_entities:
            if entity in self.kg:
                neighbors = list(self.kg.neighbors(entity))
                for neighbor in neighbors:
                    edge_data = self.kg.get_edge_data(entity, neighbor)
                    graph_results.append({
                        'triplet': (entity, edge_data['relation'], neighbor),
                        'source': 'knowledge_graph'
                    })

        # Step 4: 統合
        return {
            'vector_docs': vector_results,
            'graph_triplets': graph_results
        }

性能向上（Microsoft実測データ）:

多段階推論タスク（HotpotQA）: 標準RAG 45% → Graph RAG 68%（+23%）
エンティティ中心のQA（WebQuestions）: 標準RAG 52% → Graph RAG 71%（+19%）

ドメイン応用

1. 質問応答（QA）

医療QA（MedQA）の事例:

  
class MedicalRAG:
    def __init__(self):
        # 医学文献データベース（PubMed、Uptodate等）
        self.retriever = DenseRetriever(
            index_path="pubmed_embeddings.index"
        )
        # 医療特化LLM
        self.generator = MedicalLLM(model="meditron-70b")

    def answer_medical_query(self, query):
        # Step 1: エビデンス検索
        evidence = self.retriever.search(
            query,
            filters={'publication_year': '>= 2020'}  # 最新論文のみ
        )

        # Step 2: 回答生成（Citation付き）
        prompt = f"""以下の医学文献を参照して、質問に答えてください。
必ず引用（[1], [2]等）を付けてください。

文献:
{format_citations(evidence)}

質問: {query}"""

        answer = self.generator.generate(prompt)

        # Step 3: 安全性チェック
        safety_check = self.verify_medical_safety(answer)
        if not safety_check['is_safe']:
            return "この質問には医療従事者への相談を推奨します。"

        return answer

2. 要約（Summarization）

長文書要約の実装:

  
class RAGSummarizer:
    def summarize_long_document(self, document, max_summary_length=500):
        # Step 1: ドキュメントをチャンク分割
        chunks = split_into_chunks(document, chunk_size=1000)

        # Step 2: 各チャンクの要約
        chunk_summaries = []
        for chunk in chunks:
            summary = self.llm.generate(
                f"以下を100文字で要約:\n{chunk}"
            )
            chunk_summaries.append(summary)

        # Step 3: 要約の要約（Hierarchical Summarization）
        if len(chunk_summaries) > 10:
            # 再帰的に要約
            combined = '\n'.join(chunk_summaries)
            return self.summarize_long_document(combined, max_summary_length)
        else:
            # 最終要約
            final_summary = self.llm.generate(
                f"以下を{max_summary_length}文字で統合要約:\n" +
                '\n'.join(chunk_summaries)
            )
            return final_summary

3. 対話システム（Conversational AI）

マルチターン対話RAG:

  
class ConversationalRAG:
    def __init__(self):
        self.retriever = DenseRetriever()
        self.conversation_history = []

    def chat(self, user_message):
        # Step 1: 対話履歴を考慮したクエリ生成
        contextualized_query = self.rewrite_with_history(
            user_message,
            self.conversation_history
        )

        # Step 2: 検索
        docs = self.retriever.search(contextualized_query)

        # Step 3: 対話履歴を含めた生成
        messages = [
            {"role": "system", "content": "以下の文書を参照して回答してください。"},
            *self.conversation_history,  # 過去の対話
            {"role": "user", "content": f"文書: {docs}\n質問: {user_message}"}
        ]

        response = self.llm.chat(messages)

        # Step 4: 履歴更新
        self.conversation_history.append(
            {"role": "user", "content": user_message}
        )
        self.conversation_history.append(
            {"role": "assistant", "content": response}
        )

        return response

    def rewrite_with_history(self, current_query, history):
        """対話履歴を参照してクエリを書き換え"""
        if not history:
            return current_query

        # 直近3ターンのみ参照
        recent_history = history[-6:]  # user + assistant × 3

        rewrite_prompt = f"""対話履歴を参照して、現在の質問を独立した検索クエリに変換してください。

対話履歴:
{format_history(recent_history)}

現在の質問: {current_query}
検索クエリ:"""

        return self.llm.generate(rewrite_prompt)

# 使用例
rag = ConversationalRAG()
print(rag.chat("RAGとは何ですか？"))
# → "Retrieval-Augmented Generationの略で..."

print(rag.chat("それの利点は？"))
# 内部でクエリ書き換え: "RAGの利点は？"
# → "RAGの利点は、LLMの知識を外部文書で補完できることです..."

実装における課題

1. スケーラビリティ

課題: 数億ドキュメント規模での検索レイテンシ

解決策: 階層的インデックス（HNSW）

  
import hnswlib

# HNSWインデックス構築
dimension = 768
num_elements = 10_000_000

index = hnswlib.Index(space='cosine', dim=dimension)
index.init_index(
    max_elements=num_elements,
    ef_construction=200,  # 構築時の探索範囲
    M=48  # グラフの接続数
)

# ベクトル追加
index.add_items(embeddings, ids)

# 検索時のパラメータ調整
index.set_ef(100)  # 検索時の探索範囲
neighbors, distances = index.knn_query(query_embedding, k=10)

性能比較（1億ドキュメント、1000次元ベクトル）: | 手法 | 検索時間 | Recall@10 | メモリ使用量 | |——|———-|———–|————-| | Flat（全件探索） | 8.5秒 | 100% | 400GB | | IVF（k-means分割） | 120ms | 95% | 420GB | | HNSW | 35ms | 98% | 450GB |

2. バイアス

課題: 検索結果のバイアスがLLM出力に伝播

実測例（性別バイアス）:

質問: "優れたソフトウェアエンジニアの特徴は？"

バイアス検索結果（男性中心）:
- "彼は論理的思考力が高い"
- "彼の技術スキルは..."

LLM出力:
- "優れたエンジニアは彼の論理的思考力..."（性別バイアスを継承）

軽減策: Debiasing Retrieval

  
class DebiasedRetriever:
    def search_with_fairness(self, query, k=10):
        # Step 1: 通常の検索（候補を多めに取得）
        candidates = self.retriever.search(query, k=k*10)

        # Step 2: バイアス検出
        bias_scores = []
        for doc in candidates:
            # 性別バイアススコア
            gender_bias = count_gendered_words(doc) / len(doc.split())
            bias_scores.append(gender_bias)

        # Step 3: バイアス正規化
        # スコアを均等に分散させる
        diverse_indices = self.diversify_selection(
            bias_scores,
            k=k,
            diversity_weight=0.3
        )

        return [candidates[i] for i in diverse_indices]

    def diversify_selection(self, scores, k, diversity_weight):
        """Maximal Marginal Relevance (MMR) による多様性確保"""
        selected = []
        remaining = list(range(len(scores)))

        # 最初は最高スコアを選択
        first = np.argmax(scores)
        selected.append(first)
        remaining.remove(first)

        while len(selected) < k:
            mmr_scores = []
            for i in remaining:
                relevance = scores[i]
                # 既選択との類似度（バイアス観点）
                diversity = min([
                    abs(scores[i] - scores[j]) for j in selected
                ])
                mmr = diversity_weight * relevance + (1 - diversity_weight) * diversity
                mmr_scores.append(mmr)

            next_idx = remaining[np.argmax(mmr_scores)]
            selected.append(next_idx)
            remaining.remove(next_idx)

        return selected

3. 倫理的課題

プライバシー: ユーザー質問からの機密情報漏洩

対策: Query Sanitization

  
import re
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

class PrivacyPreservingRAG:
    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

    def sanitize_query(self, query):
        """個人情報を匿名化"""
        # PII検出
        results = self.analyzer.analyze(
            text=query,
            language='ja',
            entities=[
                "PERSON",  # 人名
                "EMAIL_ADDRESS",
                "PHONE_NUMBER",
                "CREDIT_CARD",
                "IBAN_CODE",
            ]
        )

        # 匿名化
        anonymized = self.anonymizer.anonymize(
            text=query,
            analyzer_results=results
        )

        return anonymized.text

    def search_with_privacy(self, query):
        # Step 1: クエリの匿名化
        safe_query = self.sanitize_query(query)

        # Step 2: 検索
        docs = self.retriever.search(safe_query)

        # Step 3: 検索結果の匿名化
        safe_docs = [self.sanitize_document(doc) for doc in docs]

        return safe_docs

# 使用例
rag = PrivacyPreservingRAG()
query = "太郎さん（taro@example.com）の注文履歴は？"
safe_query = rag.sanitize_query(query)
# → "<PERSON>さん（<EMAIL_ADDRESS>）の注文履歴は？"

評価指標

検索品質（Retrieval Quality）

Recall@k: 正解が上位k件に含まれる確率

  
def recall_at_k(retrieved_docs, ground_truth_docs, k):
    """
    retrieved_docs: 検索結果のドキュメントID list
    ground_truth_docs: 正解ドキュメントID set
    """
    top_k = set(retrieved_docs[:k])
    return len(top_k & ground_truth_docs) / len(ground_truth_docs)

# 例
retrieved = [101, 205, 308, 412, 501]  # 検索結果
ground_truth = {205, 412}  # 正解

print(recall_at_k(retrieved, ground_truth, k=3))  # 1件/2件 = 0.5
print(recall_at_k(retrieved, ground_truth, k=5))  # 2件/2件 = 1.0

MRR（Mean Reciprocal Rank）: 最初の正解が何位に出現するか

  
def mean_reciprocal_rank(retrieved_list, ground_truth_list):
    """
    retrieved_list: [query1_results, query2_results, ...]
    ground_truth_list: [query1_answer_set, query2_answer_set, ...]
    """
    reciprocal_ranks = []
    for retrieved, ground_truth in zip(retrieved_list, ground_truth_list):
        for rank, doc_id in enumerate(retrieved, start=1):
            if doc_id in ground_truth:
                reciprocal_ranks.append(1.0 / rank)
                break
        else:
            reciprocal_ranks.append(0.0)

    return np.mean(reciprocal_ranks)

# 例
retrieved_list = [
    [101, 205, 308],  # query1: 正解は2位
    [412, 501, 607],  # query2: 正解は1位
    [708, 809, 910],  # query3: 正解なし
]
ground_truth_list = [
    {205},
    {412},
    {999},  # 検索結果に含まれない
]

print(mean_reciprocal_rank(retrieved_list, ground_truth_list))
# (1/2 + 1/1 + 0) / 3 = 0.5

生成品質（Generation Quality）

Faithfulness（忠実性）: 生成回答が検索文書に忠実か

  
from sentence_transformers import SentenceTransformer, util

def compute_faithfulness(answer, retrieved_docs):
    """
    回答の各文が文書に含まれる情報か検証
    """
    model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

    # 回答を文に分割
    answer_sentences = answer.split('。')

    # 文書を結合
    context = ' '.join(retrieved_docs)

    # 各文の忠実性スコア
    faithfulness_scores = []
    for sent in answer_sentences:
        if not sent.strip():
            continue

        # 文と文書の意味的類似度
        sent_embedding = model.encode(sent, convert_to_tensor=True)
        context_embedding = model.encode(context, convert_to_tensor=True)
        similarity = util.cos_sim(sent_embedding, context_embedding).item()

        faithfulness_scores.append(similarity)

    return np.mean(faithfulness_scores)

# 使用例
answer = "RAGはRetrieval-Augmented Generationの略称です。LLMの知識を外部文書で補完します。"
docs = ["RAG (Retrieval-Augmented Generation) combines retrieval with LLMs."]

print(compute_faithfulness(answer, docs))  # 0.85（高い忠実性）

Answer Relevance（回答関連性）: 質問に対する回答の適切さ

  
def answer_relevance(query, answer):
    """質問と回答の意味的類似度"""
    model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

    query_embedding = model.encode(query, convert_to_tensor=True)
    answer_embedding = model.encode(answer, convert_to_tensor=True)

    relevance = util.cos_sim(query_embedding, answer_embedding).item()
    return relevance

End-to-End評価

RAGAS（Retrieval-Augmented Generation Assessment）

RAGASは、検索・生成の両方を統合評価するフレームワークです。

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

# 評価データ
eval_data = {
    'question': ["RAGとは何ですか？"],
    'answer': ["Retrieval-Augmented Generationの略称で..."],
    'contexts': [["RAG combines retrieval with generation..."]],
    'ground_truth': ["Retrieval-Augmented Generation (RAG) is a technique..."]
}

# 評価実行
result = evaluate(
    eval_data,
    metrics=[
        faithfulness,        # 忠実性
        answer_relevancy,    # 回答関連性
        context_precision,   # コンテキスト精度
        context_recall,      # コンテキスト再現率
    ]
)

print(result)
# {
#   'faithfulness': 0.92,
#   'answer_relevancy': 0.88,
#   'context_precision': 0.85,
#   'context_recall': 0.90
# }

今後の研究方向

1. Multimodal RAG（マルチモーダルRAG）

テキスト・画像・音声を統合した検索生成システム。

アーキテクチャ:

  
class MultimodalRAG:
    def __init__(self):
        self.text_retriever = DenseRetriever()
        self.image_retriever = CLIPRetriever()
        self.audio_retriever = WhisperRetriever()
        self.multimodal_llm = GPT4Vision()

    def search_multimodal(self, query, modalities=['text', 'image']):
        results = {}

        if 'text' in modalities:
            results['text'] = self.text_retriever.search(query)

        if 'image' in modalities:
            # CLIPで画像検索
            results['image'] = self.image_retriever.search(query)

        if 'audio' in modalities:
            results['audio'] = self.audio_retriever.search(query)

        return results

    def generate_with_multimodal_context(self, query, results):
        # マルチモーダルLLMに入力
        prompt = {
            'text': query,
            'images': results.get('image', []),
            'audio_transcripts': results.get('audio', []),
            'context': results.get('text', [])
        }

        return self.multimodal_llm.generate(prompt)

2. Real-time RAG（リアルタイムRAG）

ストリーミングデータ（ニュース、SNS等）を即座に反映するシステム。

課題:

インデックス更新のレイテンシ（ベクトル化に数秒）
古い情報との優先度調整

解決策: Incremental Indexing

  
class RealtimeRAG:
    def __init__(self):
        self.index = OnlineIndex()  # オンライン学習対応
        self.stream_buffer = []

    def ingest_stream(self, new_documents):
        """ストリーミングデータを即座にインデックス追加"""
        self.stream_buffer.extend(new_documents)

        # バッファが一定量に達したら一括更新
        if len(self.stream_buffer) >= 100:
            self.batch_update()

    def batch_update(self):
        """バッチでインデックス更新"""
        embeddings = encode_batch(self.stream_buffer)
        self.index.add(embeddings)
        self.stream_buffer = []

    def search_with_freshness(self, query, freshness_weight=0.5):
        """鮮度を考慮した検索"""
        results = self.index.search(query)

        # 鮮度スコアを加算
        for i, doc in enumerate(results):
            age_days = (datetime.now() - doc['timestamp']).days
            freshness_score = 1 / (1 + age_days)  # 新しいほど高スコア

            # 関連度と鮮度の重み付け統合
            results[i]['score'] = (
                (1 - freshness_weight) * doc['relevance_score'] +
                freshness_weight * freshness_score
            )

        # 再ソート
        results.sort(key=lambda x: x['score'], reverse=True)
        return results

3. Federated RAG（連合RAG）

複数組織が知識を共有せずにRAGを実現する、プライバシー保護型システム。

Federated Learning for RAG:

  
class FederatedRAG:
    def __init__(self, client_retrievers):
        """
        client_retrievers: 各組織の検索システム
        """
        self.clients = client_retrievers
        self.global_aggregator = GlobalAggregator()

    def federated_search(self, query):
        # Step 1: 各クライアントで検索（ローカル）
        local_results = []
        for client in self.clients:
            # クライアントは生データを送信せず、スコアのみ共有
            results = client.search_local(query)
            local_results.append({
                'client_id': client.id,
                'doc_scores': results['scores'],  # ドキュメントID + スコア
            })

        # Step 2: グローバル集約
        aggregated_scores = self.global_aggregator.aggregate(local_results)

        # Step 3: 上位ドキュメントのみ要求
        top_doc_ids = aggregated_scores[:10]

        final_docs = []
        for client in self.clients:
            # 該当クライアントのみドキュメント本文を取得
            client_docs = client.fetch_documents(
                [doc_id for doc_id in top_doc_ids if doc_id.startswith(client.id)]
            )
            final_docs.extend(client_docs)

        return final_docs

実運用への応用

エンタープライズRAGの実装例

  
class EnterpriseRAG:
    def __init__(self):
        # 複数データソースを統合
        self.retrievers = {
            'confluence': ConfluenceRetriever(),
            'slack': SlackRetriever(),
            'gdrive': GoogleDriveRetriever(),
            'code': GitHubRetriever(),
        }

        self.llm = OpenAI(model="gpt-4")

        # アクセス制御
        self.rbac = RoleBasedAccessControl()

    def search_enterprise(self, query, user_id):
        """ユーザー権限に基づいた検索"""
        # Step 1: ユーザーの権限確認
        accessible_sources = self.rbac.get_accessible_sources(user_id)

        # Step 2: 権限のあるソースのみ検索
        all_results = []
        for source_name, retriever in self.retrievers.items():
            if source_name in accessible_sources:
                results = retriever.search(query)
                all_results.extend(results)

        # Step 3: 重複排除・統合
        deduplicated = self.deduplicate_results(all_results)

        # Step 4: 再ランキング
        reranked = self.rerank(query, deduplicated)

        return reranked[:10]

    def generate_with_sources(self, query, user_id):
        """ソース表示付き回答生成"""
        docs = self.search_enterprise(query, user_id)

        # ソース情報を含めたプロンプト
        sources_text = "\n\n".join([
            f"[Source {i+1}: {doc['source']}]\n{doc['content']}"
            for i, doc in enumerate(docs)
        ])

        prompt = f"""以下の社内文書を参照して質問に答えてください。
必ず情報源を [Source X] の形式で引用してください。

{sources_text}

質問: {query}"""

        answer = self.llm.generate(prompt)

        # 引用元リンクを追加
        for i, doc in enumerate(docs):
            answer += f"\n[Source {i+1}]: {doc['url']}"

        return answer

まとめ

本論文「A Comprehensive Survey of Retrieval-Augmented Generation (RAG)」は、RAGの発展を体系的に整理した決定版サーベイです。

主要な洞察:

RAGの進化は3段階: Naive RAG → Advanced RAG → Modular RAG
技術革新の方向性:
- 検索精度向上（ハイブリッド検索、再ランキング）
- 適応的システム（質問複雑度に応じた動的調整）
- マルチモーダル・リアルタイム対応
実装課題: スケーラビリティ、バイアス、倫理的配慮が今後の焦点
評価の重要性: Recall@k、Faithfulness、RAGASによる総合評価

実装者へのアドバイス:

プロトタイプ: Naive RAGで迅速に検証（Chroma + OpenAI API）
本番移行: ハイブリッド検索 + 再ランキングで精度向上
スケール: HNSWインデックス + Qdrantでインフラ対応
評価: 最低50件の評価セットで定量測定を実施

RAGは2026年現在も急速に発展しており、Self-RAG、Graph RAG、Multimodal RAGなど、最先端手法が続々と登場しています。本サーベイを起点に、最新論文を追跡することで、実運用レベルのシステム構築が可能です。

参考文献

Gupta, S., Ranjan, R., & Singh, S. N. (2024). “A Comprehensive Survey of Retrieval-Augmented Generation (RAG): Evolution, Current Landscape and Future Directions.” arXiv:2410.12837.
Lewis, P., et al. (2020). “Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks.” NeurIPS.
Asai, A., et al. (2023). “Self-RAG: Learning to Retrieve, Generate, and Critique through Self-Reflection.” arXiv:2310.11511.
Jeong, S., et al. (2024). “Adaptive-RAG: Learning to Adapt Retrieval-Augmented Large Language Models through Question Complexity.” arXiv:2403.14403.

関連するZenn記事: 2026年版：RAG検索システムの実装と本番運用ガイドでは、RAGの基本実装から本番運用までの実践的な知見を解説しています。本論文解説と併せてお読みください。

✍️ RAG完全サーベイ: 基礎から最先端まで進化の全体像