【AI開発指南書：第4回】Agentic RAGの実践：Self-RAGとクエリ自律再生成・自己評価ループ

【次世代AI開発：第4回】Agentic RAGの実践：Self-RAGとクエリ自律再生成・自己評価ループ

【新連載：MCP・高度RAG・自律エージェントの現場設計論】

• 前回（第3回）：【次世代AI開発：第3回】PromptからContextへ：長文LLM時代の Context Engineering の極意

※本連載は、MCP（Model Context Protocol）、Context Engineering、GraphRAG、Self-Healing Agent、ローカルSLMなどの最先端技術スタックを用い、実務プロダクションで真に耐えうる次世代AIシステムを構築する重厚なハンズオン連載です。

---

1. イントロダクション（本記事のねらいと到達目標）

前回は、長文LLM時代において「Lost in the Middle 現象」を回避し、トークン配置とコストを最適化する「Context Engineering」の4層レイヤー設計を解説しました。

しかし、どれほどプロンプトのコンテキスト配置を工夫したとしても、RAG (検索拡張生成) の根本となる「最初のドキュメント検索」で間違った情報や無関係なノイズを取得してしまった場合、LLMは間違った回答を出力するか、ハルシネーション（嘘の生成）を起こしてしまいます。

従来の Naive RAG（単純な一発検索・一発生成） では、ユーザーの質問が抽象的であったり、社内DBの検索クエリとして不適切だった場合に対処できません。これに対し、現在エンタープライズの現場で採用が進んでいるのが、AI自身が検索結果の適合性を判定し、クエリを自律修正して再検索・自己検証を行う 「Agentic RAG（Self-RAG / Corrective RAG）」 の手法です。

第4回となる今回は、学術論文で提唱された Self-RAG および Corrective RAG (CRAG) の基本理論を解剖します。そして、Python と Pydantic の型定義を用いて「クエリ再生成 ➡ 検索結果の評価 ➡ ハルシネーション自己判定」を行う自動修正ループを完全実装することを目標とします。

前提条件と動作検証環境（Prerequisites）

• Pythonバージョン: Python 3.10 以上
• 動作確認済み主要ライブラリ:
  • openai>=1.30.0（OpenAI API クライアント）
  • pydantic>=2.7.0（Structured Outputs 用型定義）
• 動作確認環境: macOS / Linux / Windows (PowerShell)

---

2. 【理論・背景】Naive RAGの限界と Agentic RAG の必然性

従来の一般的な RAG パイプライン（Naive RAG）は、「ユーザー質問 ➡ ベクトル検索 ➡ LLM生成」という一方向の固定プロセスで動作します。

【従来の Naive RAG (一方向の固定フロー)】
[ユーザーの質問] ---> [ベクトル検索] ---> [取得ドキュメント (ノイズ混入)] ---> [LLM] ---> [嘘の回答 (失敗)]

この固定パイプラインには、実務プロダクションにおいて以下の 3大崩壊パターン が存在します。

1. 検索クエリのミスマッチ: ユーザーの質問が口語的・抽象的すぎて、社内文書の専門用語キーワードとマッチしない。
2. ノイズ情報の盲信（Blind Trust）: 検索システムがスコアの低い無関係なドキュメントを返してきた場合でも、LLMがそれを信じ込んで間違った回答を生成する。
3. ハルシネーションの未検証: LLMが検索ドキュメントに書かれていない事実を勝手に捏造して回答する。

学術的解法：Self-RAG と Corrective RAG (CRAG)

これらの課題を解決するため、近年のトップAIカンファレンスで相次いで発表されたのが、セルフチェック構造を持つRAGモデルです。

• Self-RAG (Self-Reflective RAG / Asai et al., ICLR 2024):
  • LLM自身に「今、検索が必要か？」「取得した文書は質問に関連しているか？」「生成した回答は事実に基づいているか？」を自身で自己反射（Self-Reflection）評価させるフレームワーク。
• Corrective RAG (CRAG / Yan et al., 2024):
  • 検索されたドキュメントの品質をスコアリングし、適合度が低い場合（INCORRECT）はクエリを自動修正（Query Rewriting）して再検索やウェブ検索へフォールバックする補正ループ。

RAG パイプラインの世代進化比較

評価項目	Naive RAG (第1世代)	Advanced RAG (第2世代)	Agentic RAG (第3世代 / 本手法)
フロー構造	完全固定（一方向に実行）	リランカーや前処理の追加	動的自律ループ（条件分岐・再試行）
検索失敗時の挙動	そのまま不適切な回答を出力	類似度スコア閾値でエラー	クエリを自動でリライトして再検索
ドキュメント評価	なし（検索結果を盲信）	ベクトル類似度スコアのみ	LLMによる文脈適合度の意味論的評価
ハルシネーション対策	なし	プロンプト指示のみ	生成結果と根拠文書の自己検証ループ

---

3. 【アーキテクチャ解剖】Self-RAG / CRAG 融合パイプライン

本記事で構築する Agentic RAG システムは、クエリの自律再生成、ドキュメントの適合度評価、ハルシネーション自己判定の3つの防衛線を組み合わせた、以下のステート循環パイプラインで動作します。

パイプラインを構成する4つのエージェント役割

1. Query Rewriter（クエリ最適化）: ユーザーの曖昧な質問を、データベースの専門用語に合致するよう最適な検索クエリへ変換・言い換えします。
2. Document Evaluator（検索結果査読）: 取得したドキュメント群が、質問に答えるために十分かつ適切かを CORRECT / INCORRECT で評価します。
3. Generator（回答生成）: 適合判定をクリアしたドキュメントのみをコンテキストに投入し、回答を作成します。
4. Hallucination Grader（ハルシネーション自己検証）: 生成された回答のすべての主張が、投入されたドキュメント内の事実に由来しているかを自律判定し、捏造があれば生成をリトライさせます。

---

4. 【完全実装】Self-RAG ＆ Corrective RAG の自律ループ実装

それでは、Python と OpenAI の Structured Outputs（Pydantic による型安全出力） を活用して、自己評価と自動再試行（Self-Correction）を行う完全な Agentic RAG システム agentic_rag_system.py を実装しましょう。

新規ファイル agentic_rag_system.py を作成し、以下のコードを記述します。

# 動作確認済みライブラリバージョン: openai>=1.30.0, pydantic>=2.7.0
import os
import sys
from typing import List, Literal, Optional
from pydantic import BaseModel, Field
from openai import OpenAI

# 1. Pydantic による構造化評価モデルの定義

class DocumentGrade(BaseModel):
    """検索されたドキュメントがユーザーの質問に関連しているかの評価"""
    binary_score: Literal["yes", "no"] = Field(
        description="ドキュメントが質問に関連している場合は 'yes'、無関係な場合は 'no'"
    )
    explanation: str = Field(description="評価の根拠・理由")


class HallucinationGrade(BaseModel):
    """生成された回答が検索ドキュメントの事実に由来しているかの評価"""
    binary_score: Literal["yes", "no"] = Field(
        description="回答がドキュメントの事実に完全に裏付けられている場合は 'yes'、捏造がある場合は 'no'"
    )
    explanation: str = Field(description="評価の根拠・理由")


class RewrittenQuery(BaseModel):
    """最適化・言い換えされた検索クエリ"""
    optimized_query: str = Field(description="データベース検索に最適化された新しい検索クエリ")


# =====================================================================
# Agentic RAG コアエンジンの実装
# =====================================================================

class AgenticRAGEngine:
    def __init__(self, model_name: str = "gpt-4o"):
        self.client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", "dummy_key"))
        self.model_name = model_name
        
        # 疑似ナレッジベース（本番ではベクトルDBに置き換え）
        self.knowledge_base = {
            "mcp": "Model Context Protocol (MCP) はAnthropicが開発したJSON-RPC 2.0に基づく標準規格で、LLMと外部DBやツールを安全に接続します。",
            "rag": "Agentic RAGはSelf-RAGやCorrective RAGを含み、LLMが検索結果の評価やクエリのリライトを自律的にループ実行する高度なRAG構造です。",
            "foundry": "Azure AI Foundryはエンタープライズ向けのAIモデル開発・評価・監視プラットフォームです。"
        }

    def simulate_retrieval(self, query: str) -> List[str]:
        """クエリに基づいてナレッジベースからドキュメントを疑似検索します"""
        print(f"🔍 [Retriever] 検索を実行中: クエリ = '{query}'")
        results = []
        for key, text in self.knowledge_base.items():
            if key in query.lower() or any(w in text for w in query.split()):
                results.append(text)
        
        if not results:
            # 該当がない場合は無関係なドキュメントを返してテスト
            results.append("一般的なお知らせ: 明日は社内システムの定期メンテナンスが予定されています。")
        return results

    def rewrite_query(self, original_query: str, last_explanation: str = "") -> str:
        """1. Query Rewriter: 検索に適したクエリへ自律的に言い換え再生成します"""
        print("🔄 [Query Rewriter] クエリを最適化・再生成中...")
        prompt = f"""あなたは検索クエリの専門家です。以下のユーザー質問を、データベース検索で的確なドキュメントがヒットするように専門用語を含めた最適な検索キーワードへ書き換えてください。

【元の質問】: {original_query}
【前回の不適合理由】: {last_explanation if last_explanation else '初回検索'}
"""
        response = self.client.beta.chat.completions.parse(
            model=self.model_name,
            messages=[{"role": "user", "content": prompt}],
            response_format=RewrittenQuery,
            temperature=0.1
        )
        new_query = response.choices[0].message.parsed.optimized_query
        print(f"✨ [Query Rewriter] 最適化クエリ: '{new_query}'")
        return new_query

    def grade_documents(self, query: str, documents: List[str]) -> bool:
        """2. Document Evaluator: 検索結果が質問に対して関連しているかを厳格評価します"""
        print("🧐 [Document Evaluator] 検索ドキュメントの適合度を査読中...")
        combined_docs = "\n\n".join(documents)
        prompt = f"""以下のドキュメント群が、ユーザーの質問に答えるために十分に関連しているか評価してください。

【ユーザー質問】: {query}
【検索ドキュメント】:
{combined_docs}
"""
        response = self.client.beta.chat.completions.parse(
            model=self.model_name,
            messages=[{"role": "user", "content": prompt}],
            response_format=DocumentGrade,
            temperature=0.0
        )
        grade = response.choices[0].message.parsed
        print(f"📊 [Document Evaluator] 判定結果: {grade.binary_score.upper()} (理由: {grade.explanation})")
        return grade.binary_score == "yes"

    def grade_hallucination(self, documents: List[str], generated_answer: str) -> bool:
        """3. Hallucination Grader: 生成された回答が事実に裏付けられているかを自己検証します"""
        print("🛡️ [Hallucination Grader] ハルシネーション（事実捏造）を検証中...")
        combined_docs = "\n\n".join(documents)
        prompt = f"""以下の回答が、与えられた参照ドキュメントの事実のみに基づいているか厳格に判定してください。ドキュメントに書かれていない事実が捏造されている場合は 'no' にしてください。

【参照ドキュメント】:
{combined_docs}

【生成された回答】:
{generated_answer}
"""
        response = self.client.beta.chat.completions.parse(
            model=self.model_name,
            messages=[{"role": "user", "content": prompt}],
            response_format=HallucinationGrade,
            temperature=0.0
        )
        grade = response.choices[0].message.parsed
        print(f"⚖️ [Hallucination Grader] 判定結果: {grade.binary_score.upper()} (理由: {grade.explanation})")
        return grade.binary_score == "yes"

    def run_agentic_rag(self, user_query: str, max_retries: int = 2) -> str:
        """Agentic RAG の全自律ループを統合実行します"""
        print(f"\n🚀 === Agentic RAG パイプライン開始: 『{user_query}』 ===")
        current_query = user_query
        retries = 0
        last_explanation = ""

        while retries <= max_retries:
            print(f"\n--- [試行 #{retries + 1} / 最大 {max_retries + 1} 回] ---")
            
            # Step A: 初回以外はクエリを書き換え
            if retries > 0:
                current_query = self.rewrite_query(user_query, last_explanation)
            
            # Step B: ドキュメント検索
            docs = self.simulate_retrieval(current_query)
            
            # Step C: 検索結果の適合性評価 (Corrective RAG)
            is_relevant = self.grade_documents(user_query, docs)
            
            if not is_relevant:
                print("⚠️ 検索ドキュメントの適合度が低いため、自律再検索へループします。")
                last_explanation = "検索結果がユーザー質問と無関係でした。"
                retries += 1
                continue
                
            # Step D: 回答生成
            print("📝 [Generator] 適合ドキュメントに基づき回答を生成中...")
            gen_prompt = f"以下の参照ドキュメントに基づき、質問に正確に答えてください。\n\n参照ドキュメント:\n" + "\n".join(docs) + f"\n\n質問: {user_query}"
            gen_res = self.client.chat.completions.create(
                model=self.model_name,
                messages=[{"role": "user", "content": gen_prompt}],
                temperature=0.2
            )
            candidate_answer = gen_res.choices[0].message.content

            # Step E: ハルシネーションの自己判定 (Self-RAG)
            is_grounded = self.grade_hallucination(docs, candidate_answer)
            
            if is_grounded:
                print("✅ [成功] ハルシネーション検証をクリアしました。最終回答を出力します。")
                return candidate_answer
            else:
                print("⚠️ ハルシネーションが検知されたため、回答生成をリトライします。")
                retries += 1
                
        return "申し訳ありません。十分な根拠のある正確な情報を見つけられなかったため、安全のため回答を差し控えます。"


# =====================================================================
# 動作デモ
# =====================================================================
if __name__ == "__main__":
    if "OPENAI_API_KEY" not in os.environ:
        print("💡 NOTE: 実際に動かす場合は OPENAI_API_KEY の環境変数を設定してください。")
        sys.exit(0)
        
    engine = AgenticRAGEngine(model_name="gpt-4o")
    
    # 抽象的な質問（クエリ書き換えと再検索ループのテスト）
    user_input = "Agentic RAGの自律ループの仕組みについて詳しく教えて"
    final_result = engine.run_agentic_rag(user_input)
    
    print("\n==================================================")
    print("                 最終出力結果                      ")
    print("==================================================")
    print(final_result)

---

5. コードの行別・ロジック詳細解説

実装した AgenticRAGEngine のコアロジックを解説します。

1. Pydantic と response_format による型安全な評価

• DocumentGrade や HallucinationGrade クラスを定義し、OpenAIの client.beta.chat.completions.parse を使用しています。
• LLMは自然言語ではなく {"binary_score": "yes", "explanation": "..."} という完全な JSON / Pydantic オブジェクトを返却するため、grade.binary_score == "yes" という条件分岐をコード内で100%安全に実行できます。

2. while ループによる自律リトライ機構

• run_agentic_rag メソッドでは、while retries <= max_retries: によるループを構成しています。
• 検索適合度（grade_documents）が no であった場合、continue により rewrite_query へ戻り、クエリを言い換えて自動再検索を実行します。

3. ハルシネーション判定（Self-RAG）の2重防衛線

• 生成された回答案 candidate_answer に対し、grade_hallucination メソッドで「元の参照ドキュメントに書かれていない事実」が含まれていないかをチェックします。
• 捏造が発覚した場合は安易にユーザーへ出力せず、リトライまたは安全なフォールバックメッセージを出力します。

---

6. プロダクション導入・セキュリティ・パフォーマンス最適化

Agentic RAG を商用環境へ投入する際の最適化ノウハウを解説します。

1. 無限ループの回避とコストキャップ

• クエリの自動リライトやリトライを無限に行うと、APIコストの急増やレスポンス遅延の原因になります。
• max_retries（推奨: 2〜3回）を設定し、上限に達した場合は「正確な情報が見つかりませんでした」と回答を安全に辞退する安全弁（Guardrail） を必ず設けてください。

2. 評価エージェントの軽量モデル化（SLM活用）

• ドキュメント評価（grade_documents）やハルシネーションチェック（grade_hallucination）は、二値判定（yes/no）の比較的シンプルで高速なタスクです。
• この評価エージェント部分のみ、コストが1/10以下の軽量モデル（GPT-4o-mini や Claude 3.5 Haiku）へ委譲することで、システム全体の応答速度を高速化し、運用コストを劇的に削減できます。

---

7. まとめと次回の展望

今回は、従来の一方向な Naive RAG の限界を突破し、自律的なクエリ書き換え、検索結果の適合度評価、ハルシネーション自己判定を行う Agentic RAG（Self-RAG / Corrective RAG） のアーキテクチャと完全な Python 実装を解説しました。

検索と評価の自律ループを回すことで、ハルシネーションを極限まで抑え込み、高い回答精度を実現できることがご理解いただけたかと思います。

しかし、RAGが取り扱うデータが「単なるテキスト文章」ではなく、「組織図」「複雑な製品仕様間の依存関係」「エンティティ同士の関連性」である場合、通常のベクトル検索（Chunk分割）ではドキュメント全体の広域な全体像を回答することができません。

次回、第5回。
ベクトル検索とナレッジグラフ（知識グラフ）を融合させ、複雑な関係性を捉える次世代RAG、**『GraphRAG超入門：ナレッジグラフ×ベクトル検索で全体関係性を捉えるハイブリッドRAG』**に進みます。

ドキュメント全体の構造をグラフとして理解し、真の広域回答を可能にする GraphRAG の極意へ進みましょう。