Agentic RAG：AIに自律検索させる

受動的RAGから能動的エージェントへ

従来のRAGは受動的な動作モードです：ユーザーが質問→文書を検索→回答を生成。このモードは単純なQ&Aシナリオでは良好ですが、多段階の推論、文書横断的な統合、動的な意思決定を必要とする複雑な問題に直面すると、受動的RAGでは対応が難しくなります。例えば、ユーザーが「A社とB社のAI分野における特許ポートフォリオを比較してください」と尋ねた場合、受動的RAGはA社またはB社の部分的な情報しか検索できず、自動的に複数回の検索と比較分析を行うことはできません。Agentic RAGはまさにこのような複雑な検索シナリオを解決するために生まれました。

Agentic RAGの核となる考え方は、エージェントの自律的な意思決定能力をRAGシステムに導入することです。エージェントは受動的に単一の検索を実行するのではなく、人間の研究者のように動作します：問題を分析→検索計画を立案→検索を実行→検索結果を評価→追加検索が必要か判断→すべての情報を統合→最終回答を生成。このループは、エージェントがユーザーの質問に答えるのに十分な情報を収集したと判断するまで繰り返すことができます。

Agentic RAGのコアアーキテクチャ

Agentic RAGアーキテクチャは4つのコアコンポーネントで構成されています：プランナー（Planner）はユーザーの質問を分析し、検索戦略を立案します——複雑な質問を複数のサブ質問に分解し、各サブ質問がどのタイプの情報を検索する必要があるかを決定します。レトリーバー（Retriever）は実際の検索操作を実行し、ベクトル検索、キーワード検索、構造化クエリなど複数の検索方法をサポートします。評価器（Evaluator）は検索結果の品質を評価します——情報がユーザーの質問に答えるのに十分か、検索戦略を調整する必要があるか、追加検索が必要か。シンセサイザー（Synthesizer）はすべての検索情報を一貫した回答に統合します。

from openai import OpenAI
import json

client = OpenAI(api_key="your-deepseek-api-key", base_url="https://api.deepseek.com")

class AgenticRAG:
    def __init__(self, vector_db, max_iterations=5):
        self.vector_db = vector_db
        self.max_iterations = max_iterations
        self.search_history = []

    def plan(self, question):
        """プランナー：問題を分析し、検索計画を立案"""
        prompt = f"""あなたは検索戦略立案の専門家です。以下の質問を分析し、検索計画を立案してください。

ユーザーの質問：{question}

検索計画をJSON形式で出力してください：
{{
  "sub_questions": [
    {{"query": "サブ質問1の検索語", "source": "knowledge_base/web/internal_docs"}},
    {{"query": "サブ質問2の検索語", "source": "knowledge_base/web/internal_docs"}}
  ],
  "reasoning": "このように分割した理由"
}}"""
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role":"user","content":prompt}],
            temperature=0.2
        )
        return json.loads(response.choices[0].message.content)

    def retrieve(self, query, source="knowledge_base"):
        """レトリーバー：単一の検索を実行"""
        # 本番環境では、異なるソースにルーティング
        results = self.vector_db.search(query, top_k=5)
        return [r["content"] for r in results]

    def evaluate(self, question, collected_info):
        """評価器：情報が十分かどうかを判断"""
        prompt = f"""現在収集された情報がユーザーの質問に答えるのに十分かどうかを評価してください。

ユーザーの質問：{question}
収集された情報：
{json.dumps(collected_info, ensure_ascii=False)}

JSON出力：
{{
  "is_sufficient": true/false,
  "missing_info": ["さらに必要な情報"],
  "next_queries": ["推奨される次の検索語"],
  "confidence": 0.0-1.0
}}"""
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role":"user","content":prompt}],
            temperature=0.1
        )
        return json.loads(response.choices[0].message.content)

    def synthesize(self, question, all_info):
        """シンセサイザー：すべての情報を統合して回答を生成"""
        prompt = f"""以下の情報に基づいて、ユーザーの質問に回答してください。

ユーザーの質問：{question}

収集された情報：
{json.dumps(all_info, ensure_ascii=False, indent=2)}

要件：
1. すべての情報を統合して完全な回答を提供
2. 情報の出所を明記（どの情報がどの検索から得られたか）
3. 情報間に矛盾がある場合は、矛盾を指摘し分析を提供
4. 一部の側面で情報が不足している場合は、その旨を明確に述べる"""
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role":"user","content":prompt}],
            temperature=0.3
        )
        return response.choices[0].message.content

    def run(self, question):
        print(f"ユーザーの質問：{question}\n")
        plan = self.plan(question)
        print(f"検索計画：{len(plan['sub_questions'])} 個のサブ質問")

        all_info = {}
        for i, sq in enumerate(plan["sub_questions"]):
            results =