LLM Agentic RAGワークフロー実践：自律型検索拡張生成アーキテクチャ設計

AI与大数据

概要

  • Agentic RAGは2026年のRAGの究極形態：「検索・生成」から「計画・検索・推論・検証」のクローズドループへ、精度40%以上向上
  • 4つのコア能力：自律検索計画、マルチステップ推論チェーン、ツール呼び出し拡張、自己反省検証
  • 3つのAgentic RAGアーキテクチャ：単一エージェントルーティング、マルチエージェント協調、階層型エージェントオーケストレーション、それぞれ最適なユースケースあり
  • 本番運用の鍵：検索品質評価、ハルシネーション検出、コスト制御、3つの指標はすべて不可欠
  • 本記事ではLangGraph + Agentic RAGの完全実装と本番デプロイソリューションを提供

目次

Agentic RAG：RAGの究極形態

従来のRAG vs Agentic RAG

項目 従来のRAG Agentic RAG
検索方式 単一クエリ マルチラウンド自律検索
推論深度 1ステップ生成 マルチステップ推論チェーン
ツール使用 検索のみ 検索＋計算＋API
自己修正 なし 反省＋リトライ
複雑クエリ 弱い 強い
精度 60-70% 85-95%

Agentic RAGの進化ロードマップ

┌──────────────────────────────────────────────────────────────┐ │ RAG進化ロードマップ │ │ │ │ RAG 1.0 (2023) │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Query → 検索 → 生成 │ │ │ │ シンプルパイプライン、反省なし、計画なし │ │ │ └──────────────────────────────────────────────────────┘ │ │ ↓ │ │ RAG 2.0 (2024) │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Query → リライト → ハイブリッド検索 → リランク → 生成│ │ │ │ クエリ拡張＋マルチパス検索＋リランキング │ │ │ └──────────────────────────────────────────────────────┘ │ │ ↓ │ │ Agentic RAG (2025-2026) │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Query → 計画 → 検索 → 推論 → 検証 → (ループ) → 生成 │ │ │ │ 自律計画＋マルチステップ推論＋自己反省＋ツール呼び出し│ │ │ └──────────────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────────────┘

2026年Agentic RAGフレームワーク比較

フレームワーク 言語 主な特徴 Agent対応 本番対応
LangGraph Python グラフオーケストレーション＋状態マシン 強い はい
CrewAI Python マルチエージェント協調 強い 中
AutoGen Python マルチエージェント会話 強い 中
LlamaIndex Python RAGエコシステム最良 中 はい
Haystack Python パイプライン型オーケストレーション 中 はい
Dify Python/Go ビジュアル＋ローコード 中 はい

4つのコア能力

能力1：自律検索計画

`python from pydantic import BaseModel from typing import List, Optional

class RetrievalPlan(BaseModel): queries: List[str] tools: List[str] priority: int max_iterations: int

class RetrievalPlanner: def init(self, llm): self.llm = llm

def plan(self, question: str) -> RetrievalPlan:
    prompt = f"""以下の問題を分析し、検索計画を作成してください：

問題：{question}

出力内容：

  1. 検索が必要なサブ問題のリスト（優先度順）
  2. 各サブ問題に必要な検索ツール
  3. 最大検索ラウンド数

フォーマット：

  • サブ問題1 [ツール: vector_search]

  • サブ問題2 [ツール: web_search]

  • ..."""

      response = self.llm.invoke(prompt)
  return self._parse_plan(response)

    def _parse_plan(self, response: str) -> RetrievalPlan: queries = [] tools = []

      for line in response.strip().split("\n"):
      if "[" in line and "]" in line:
          query = line.split("[")[0].strip("- ").strip()
          tool = line.split("[")[1].split("]")[0].replace("ツール:", "").strip()
          queries.append(query)
          tools.append(tool)
  
  return RetrievalPlan(
      queries=queries,
      tools=tools,
      priority=1,
      max_iterations=3,
  )

`

能力2：マルチステップ推論チェーン

`python class MultiStepReasoner: def init(self, llm, max_steps=5): self.llm = llm self.max_steps = max_steps

def reason(self, question: str, context: List[str]) -> dict:
    steps = []
    current_question = question
    accumulated_context = list(context)
    
    for step in range(self.max_steps):
        prompt = f"""以下のコンテキストに基づき、段階的に推論して問題に答えてください。

収集済みコンテキスト： {chr(10).join(accumulated_context)}

現在の問題：{current_question}

出力内容：

  1. 既存情報に基づく推論ステップ

  2. 追加情報が必要か（はい/いいえ）

  3. 必要な場合、次の検索クエリは何か

  4. 現在の推論結論"""

         response = self.llm.invoke(prompt)
     step_result = self._parse_step(response)
     steps.append(step_result)
     
     if not step_result["need_more_info"]:
         break
     
     current_question = step_result["next_query"]
     new_context = self._retrieve(current_question)
     accumulated_context.extend(new_context)
 
 return {
     "steps": steps,
     "final_answer": steps[-1]["conclusion"],
     "total_steps": len(steps),
 }

    def _parse_step(self, response: str) -> dict: lines = response.strip().split("\n") return { "reasoning": lines[0] if len(lines) > 0 else "", "need_more_info": "はい" in (lines[1] if len(lines) > 1 else ""), "next_query": lines[2] if len(lines) > 2 else "", "conclusion": lines[3] if len(lines) > 3 else "", } `

能力3：ツール呼び出し拡張

`python from typing import Callable, Dict, Any

class ToolRegistry: def init(self): self.tools: Dict[str, Callable] = {} self.tool_descriptions: Dict[str, str] = {}

def register(self, name: str, description: str, func: Callable):
    self.tools[name] = func
    self.tool_descriptions[name] = description

def execute(self, name: str, **kwargs) -> Any:
    if name not in self.tools:
        raise ValueError(f"Tool {name} not found")
    return self.tools[name](**kwargs)

def get_descriptions(self) -> str:
    return "\n".join([
        f"- {name}: {desc}" 
        for name, desc in self.tool_descriptions.items()
    ])

def setup_tools(registry: ToolRegistry): registry.register( "vector_search", "ナレッジベースで関連ドキュメントを検索", lambda query, top_k=5: vector_search_impl(query, top_k), ) registry.register( "web_search", "インターネットで最新情報を検索", lambda query, num_results=5: web_search_impl(query, num_results), ) registry.register( "calculator", "数学計算を実行", lambda expression: eval(expression), ) registry.register( "sql_query", "データベースをクエリ", lambda sql: sql_query_impl(sql), ) registry.register( "api_call", "外部APIを呼び出し", lambda url, params: api_call_impl(url, params), ) `

能力4：自己反省検証

`python class SelfReflector: def init(self, llm): self.llm = llm

def verify(self, question: str, answer: str, context: List[str]) -> dict:
    prompt = f"""以下の回答の正確性を検証してください：

問題：{question} 回答：{answer} 参照コンテキスト： {chr(10).join(context)}

評価項目：

  1. 回答はコンテキストに完全に基づいているか？（grounded: はい/いいえ）

  2. 回答は問題に完全に答えているか？（complete: はい/いいえ）

  3. 回答にハルシネーション内容はあるか？（hallucination: はい/いいえ）

  4. 信頼度スコア (0-1)

  5. 改善提案"""

     response = self.llm.invoke(prompt)
 result = self._parse_verification(response)
 
 return result

    def _parse_verification(self, response: str) -> dict: lines = response.strip().split("\n") grounded = "はい" in (lines[0] if len(lines) > 0 else "") complete = "はい" in (lines[1] if len(lines) > 1 else "") hallucination = "はい" in (lines[2] if len(lines) > 2 else "")

     return {
     "grounded": grounded,
     "complete": complete,
     "has_hallucination": hallucination,
     "confidence": 0.8,
     "needs_retry": not grounded or not complete or hallucination,
 }

`

3つのAgentic RAGアーキテクチャ

アーキテクチャ比較

┌──────────────────────────────────────────────────────────────┐ │ 3つのAgentic RAGアーキテクチャ │ │ │ │ アーキテクチャ1: 単一エージェントルーティング │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Query → Router Agent → [検索|計算|API] → 生成 │ │ │ │ 適用：シンプルなシナリオ、迅速なデプロイ │ │ │ └──────────────────────────────────────────────────────┘ │ │ │ │ アーキテクチャ2: マルチエージェント協調 │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Query → Planner → [Retriever|Reasoner|Verifier] │ │ │ │ ← 協調ループ → │ │ │ │ 適用：複雑なクエリ、高精度 │ │ │ └──────────────────────────────────────────────────────┘ │ │ │ │ アーキテクチャ3: 階層型エージェントオーケストレーション │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Orchestrator → [Sub-Orchestrator → [Workers]] │ │ │ │ 適用：エンタープライズ、マルチテナント │ │ │ └──────────────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────────────┘

アーキテクチャ選定

シナリオ 推奨アーキテクチャ 複雑度 精度 レイテンシ
FAQ/簡単なクエリ 単一エージェントルーティング 低 80% 2s
リサーチ分析 マルチエージェント協調 中 92% 8s
エンタープライズナレッジベース 階層型エージェント 高 95% 5s
リアルタイム意思決定 単一エージェント＋キャッシュ 低 85% 1s

LangGraphでAgentic RAGを実装

完全実装

`python from langgraph.graph import StateGraph, END from typing import TypedDict, List, Dict, Any, Annotated import operator

class AgentState(TypedDict): question: str plan: Dict[str, Any] documents: List[str] reasoning_steps: List[Dict] answer: str verification: Dict[str, Any] iteration: int max_iterations: int

class AgenticRAGGraph: def init(self, llm, retriever, reflector): self.llm = llm self.retriever = retriever self.reflector = reflector self.graph = self._build_graph()

def _build_graph(self):
    workflow = StateGraph(AgentState)
    
    workflow.add_node("plan", self._plan_node)
    workflow.add_node("retrieve", self._retrieve_node)
    workflow.add_node("reason", self._reason_node)
    workflow.add_node("generate", self._generate_node)
    workflow.add_node("verify", self._verify_node)
    
    workflow.set_entry_point("plan")
    workflow.add_edge("plan", "retrieve")
    workflow.add_edge("retrieve", "reason")
    workflow.add_edge("reason", "generate")
    workflow.add_edge("generate", "verify")
    
    workflow.add_conditional_edges(
        "verify",
        self._should_retry,
        {
            "retry": "retrieve",
            "finish": END,
        },
    )
    
    return workflow.compile()

def _plan_node(self, state: AgentState) -> AgentState:
    planner = RetrievalPlanner(self.llm)
    plan = planner.plan(state["question"])
    state["plan"] = plan.dict()
    state["iteration"] = state.get("iteration", 0) + 1
    return state

def _retrieve_node(self, state: AgentState) -> AgentState:
    documents = []
    plan = state["plan"]
    
    for query in plan.get("queries", [state["question"]]):
        docs = self.retriever.search(query, top_k=5)
        documents.extend([d.page_content for d in docs])
    
    state["documents"] = list(set(documents))
    return state

def _reason_node(self, state: AgentState) -> AgentState:
    reasoner = MultiStepReasoner(self.llm)
    result = reasoner.reason(state["question"], state["documents"])
    state["reasoning_steps"] = result["steps"]
    return state

def _generate_node(self, state: AgentState) -> AgentState:
    context = "\n".join(state["documents"])
    reasoning = "\n".join([
        f"ステップ{i+1}: {s['conclusion']}" 
        for i, s in enumerate(state["reasoning_steps"])
    ])
    
    prompt = f"""以下の情報に基づいて問題に答えてください。

コンテキスト： {context}

推論プロセス： {reasoning}

問題：{state["question"]}

完全で正確な回答を提供してください："""

    state["answer"] = self.llm.invoke(prompt)
    return state

def _verify_node(self, state: AgentState) -> AgentState:
    result = self.reflector.verify(
        state["question"], state["answer"], state["documents"]
    )
    state["verification"] = result
    return state

def _should_retry(self, state: AgentState) -> str:
    if (
        state["verification"]["needs_retry"]
        and state["iteration"] < state.get("max_iterations", 3)
    ):
        return "retry"
    return "finish"

def run(self, question: str, max_iterations: int = 3) -> dict:
    initial_state = {
        "question": question,
        "plan": {},
        "documents": [],
        "reasoning_steps": [],
        "answer": "",
        "verification": {},
        "iteration": 0,
        "max_iterations": max_iterations,
    }
    
    result = self.graph.invoke(initial_state)
    return result

`

検索品質とハルシネーション検出

検索品質評価

指標 計算方法 目標値
リコール(Recall) 関連文書 / 全関連文書 >90%
精度(Precision) 関連文書 / 取得文書 >80%
MRR 正解の順位の逆数の平均 >0.7
NDCG@10 正規化割引累積利得 >0.8

ハルシネーション検出方法

`python class HallucinationDetector: def init(self, llm, embedder): self.llm = llm self.embedder = embedder

def detect(self, answer: str, context: List[str]) -> dict:
    claim_score = self._claim_verification(answer, context)
    consistency_score = self._self_consistency(answer)
    similarity_score = self._context_similarity(answer, context)
    
    overall = (
        0.4 * claim_score
        + 0.3 * consistency_score
        + 0.3 * similarity_score
    )
    
    return {
        "hallucination_risk": 1 - overall,
        "claim_verification": claim_score,
        "self_consistency": consistency_score,
        "context_similarity": similarity_score,
        "is_reliable": overall > 0.7,
    }

def _claim_verification(self, answer: str, context: List[str]) -> float:
    prompt = f"""回答内の各主張がコンテキストに裏付けられているか検証してください。

コンテキスト：{chr(10).join(context)} 回答：{answer}

各主張にラベル付け：支持/不支持/判断不能"""

    response = self.llm.invoke(prompt)
    supported = response.count("支持")
    total = supported + response.count("不支持") + response.count("判断不能")
    return supported / max(total, 1)

def _self_consistency(self, answer: str) -> float:
    variations = []
    for _ in range(3):
        var = self.llm.invoke(f"別の言い方で言い換えてください：{answer}")
        variations.append(var)
    
    embeddings = self.embedder.embed(variations + [answer])
    similarities = [
        cosine_similarity(embeddings[-1], emb) for emb in embeddings[:-1]
    ]
    return sum(similarities) / len(similarities)

def _context_similarity(self, answer: str, context: List[str]) -> float:
    answer_emb = self.embedder.embed([answer])[0]
    context_emb = self.embedder.embed(context)
    max_sim = max(
        cosine_similarity(answer_emb, emb) for emb in context_emb
    )
    return max_sim

`

本番デプロイとコスト制御

コスト最適化戦略

戦略 コスト削減 精度への影響
検索結果キャッシュ 40-60% なし
小型モデルルーティング 30-50% <2%
バッチ推論 20-30% なし
コンテキスト圧縮 15-25% <1%
動的イテレーション制御 10-20% <3%

動的イテレーション制御

`python class DynamicIterationController: def init(self, max_iterations=3, confidence_threshold=0.8): self.max_iterations = max_iterations self.confidence_threshold = confidence_threshold

def should_continue(self, state: dict) -> bool:
    if state["iteration"] >= self.max_iterations:
        return False
    
    if state.get("verification", {}).get("confidence", 0) >= self.confidence_threshold:
        return False
    
    if state["iteration"] > 1:
        prev_conf = state.get("prev_confidence", 0)
        curr_conf = state.get("verification", {}).get("confidence", 0)
        if curr_conf - prev_conf < 0.05:
            return False
    
    return True

`

Agentic RAGパフォーマンスベンチマーク

指標 従来のRAG Agentic RAG 向上
簡単なクエリの精度 85% 92% +7%
複雑なクエリの精度 55% 88% +33%
ハルシネーション率 15% 5% -67%
平均レイテンシ 2s 5s -60%
平均トークン消費 500 2000 -75%

まとめと関連情報

重要ポイントの振り返り

  1. Agentic RAGはRAGの究極形態：計画・検索・推論・検証のクローズドループで精度40%以上向上
  2. 4つのコア能力：自律計画、マルチステップ推論、ツール呼び出し、自己反省
  3. 3つのアーキテクチャ：単一エージェントルーティング（シンプル）、マルチエージェント協調（複雑）、階層型エージェント（エンタープライズ）
  4. 本番運用の鍵：検索品質評価＋ハルシネーション検出＋コスト制御

Agentic RAGソリューション推奨

シナリオ 推奨ソリューション 期待効果
クイック検証 単一エージェント＋LangGraph 精度85%以上
本番デプロイ マルチエージェント＋キャッシュ＋量子化 精度90%以上
エンタープライズ 階層型エージェント＋モニタリング 精度95%以上

RAGデータのフォーマット変換が必要ですか？JSON→YAMLツールとテキスト差分比較で検索データを素早く処理できます。

関連記事

ブラウザローカルツールを無料で試す →

#Agentic RAG#智能RAG工作流#LLM Agent#RAG管道#自主检索增强#2026