Agentic RAG実践：自律型推論検索Agentを構築する5つのコアパターン
AI与大数据
従来のRAGではもう足りない
念入りに構築したRAGシステムに、ユーザーが「RAGとファインチューニング、どちらが私のユースケースに適している？」と質問したところ、RAGの基礎的な紹介とファインチューニングの定義が山ほど返ってきた——全く質問に答えていない。これは検索精度の問題ではなく、従来のRAGには推論能力がないのが本質です。「聞かれたことを検索する」ことしかできず、多段推論や複数知識源の統合、自己修正が必要な複雑な問題にはお手上げです。
2026年、Agentic RAGが突破口となっています。受動的な「検索→生成」パイプラインではなく、Agentが自律的に判断する：検索すべきか？どこを検索すべきか？結果は十分か？別の戦略で再検索すべきか？この「自律的意志決定＋反復的改善」のパラダイムこそが、複雑な問題への正しいアプローチです。
コア概念早見表
| 概念 | 説明 | 典型的な実装 |
|---|---|---|
| Agentic RAG | Agentが自律的に検索戦略を決定するRAGパラダイム | LangGraph StateGraph、LlamaIndex Agent |
| 自律検索 | Agentが質問に基づき検索の要否と内容を自律決定 | ReActループ、ツール呼び出し |
| マルチホップ推論 | 複雑な質問を複数のサブ質問に分解して段階的に検索 | 質問分解、チェーン検索 |
| 自己反思 | Agentが検索結果の品質を評価し、再検索の要否を判断 | ドキュメント採点、クエリ書き換え |
| ツール呼び出し | Agentが外部ツール（検索、計算、データベース）を呼び出して能力を拡張 | Function Calling、Tool Use |
| LangGraph | ステートフルなAgentワークフローを構築するフレームワーク | StateGraph、条件付きエッジ、チェックポイント |
| ReActパターン | 推論（Reasoning）と行動（Acting）を交互に繰り返す | Thought-Action-Observationループ |
問題分析：従来のRAGの5つの限界
| # | 限界 | 具体的な表れ | 影響 |
|---|---|---|---|
| 1 | 単回検索 | ベクトル検索を1回しか実行できず、反復最適化が不可能 | 検索が不完全、重要情報の見落とし |
| 2 | 推論能力なし | 複雑な質問をサブ質問に分解できない | マルチホップ問題の回答品質が低い |
| 3 | 自己修正不可 | 無関係なドキュメントを検索しても識別・再試行できない | 幻覚が深刻、的外れな回答 |
| 4 | 計画性の欠如 | 検索戦略がなく、何を先に検索すべきか決定できない | 非効率、リソースの無駄 |
| 5 | マルチホップ問題への対応不可 | 「AとB、どちらがCのシナリオに適している？」という質問に完全に無力 | ユーザー体験が極めて悪く、離脱率が高い |
従来のRAGの根本的な問題：ステートレスなパイプラインであり、知能を持つAgentではない。Agentic RAGのコアとなる転換は、「パイプライン」から「Agent」へ——検索プロセスに自律的意志決定、反復的改善、自己修正の能力を持たせることです。
パターン1：ルーターエージェント — インテリジェントな振り分け
最もシンプルなAgentic RAGパターン：質問タイプに基づいて、異なるナレッジベースや検索戦略にクエリをルーティング。統一ベクトル検索よりはるかに精度が高い。
from enum import Enum
from typing import Optional
class QueryType(Enum):
TECHNICAL = "technical"
BUSINESS = "business"
COMPARISON = "comparison"
TUTORIAL = "tutorial"
class RouterAgent:
def __init__(self, llm, retrievers: dict[str, object]):
self.llm = llm
self.retrievers = retrievers
def classify(self, question: str) -> QueryType:
prompt = f"""Classify this question into one of:
- technical: code/API/implementation questions
- business: pricing/ROI/strategy questions
- comparison: comparing two or more options
- tutorial: how-to/step-by-step guides
Question: {question}
Type:"""
result = self.llm.invoke(prompt).strip().lower()
return QueryType(result)
def route(self, question: str) -> list[dict]:
query_type = self.classify(question)
retriever = self.retrievers.get(query_type.value)
if not retriever:
retriever = self.retrievers["technical"]
docs = retriever.similarity_search(question, k=5)
return [{"content": d.page_content, "source": d.metadata.get("source", "")} for d in docs]
retrievers = {
"technical": tech_vector_store,
"business": biz_vector_store,
"comparison": compare_vector_store,
"tutorial": tutorial_vector_store,
}
router = RouterAgent(llm, retrievers)
result = router.route("TiDBとCockroachDBの分散トランザクションにおける違いは？")
ルーターエージェントはナレッジベースの分類が明確なシナリオに最適。実装はシンプルだが、効果は即座に現れます。
パターン2：マルチホップ推論エージェント
複雑な質問には分解が必要。例えば「TiDBとCockroachDBの分散トランザクションの違いを比較し、パフォーマンスベンチマークデータを提示」には、各データベースの情報を別々に検索し、ベンチマークデータを検索し、最後に統合して回答する必要があります。
from langchain.agents import create_react_agent
tools = [
search_knowledge_base,
search_web,
calculate_metrics,
query_database,
]
agent = create_react_agent(llm, tools, prompt)
result = agent.invoke({
"input": "TiDBとCockroachDBの分散トランザクションの違いを比較し、ベンチマークデータを提示"
})
よりきめ細かいマルチホップ実装：
from dataclasses import dataclass
from typing import Optional
@dataclass
class SubQuestion:
question: str
answer: Optional[str] = None
sources: list[str] = None
class MultiHopAgent:
def __init__(self, llm, retriever):
self.llm = llm
self.retriever = retriever
def decompose(self, question: str) -> list[SubQuestion]:
prompt = f"""Break down this complex question into 2-4 sub-questions.
Each sub-question should be independently answerable.
Question: {question}
Sub-questions (one per line):"""
result = self.llm.invoke(prompt)
lines = [l.strip() for l in result.strip().split("\n") if l.strip()]
return [SubQuestion(question=line) for line in lines]
def answer_sub_question(self, sub_q: SubQuestion) -> SubQuestion:
docs = self.retriever.similarity_search(sub_q.question, k=3)
context = "\n".join([d.page_content for d in docs])
answer = self.llm.invoke(
f"Based on:\n{context}\n\nAnswer: {sub_q.question}"
)
sub_q.answer = answer
sub_q.sources = [d.metadata.get("source", "") for d in docs]
return sub_q
def synthesize(self, question: str, sub_answers: list[SubQuestion]) -> str:
evidence = "\n".join([f"- {sq.question}: {sq.answer}" for sq in sub_answers])
return self.llm.invoke(
f"Based on the following evidence, answer the original question.\n\n"
f"Original: {question}\n\nEvidence:\n{evidence}\n\nAnswer:"
)
def run(self, question: str) -> dict:
sub_questions = self.decompose(question)
answered = [self.answer_sub_question(sq) for sq in sub_questions]
final = self.synthesize(question, answered)
return {
"answer": final,
"sub_questions": [
{"q": sq.question, "a": sq.answer} for sq in answered
],
}
agent = MultiHopAgent(llm, vector_store)
result = agent.run("RAGとファインチューニング、どちらが適しているか？コスト、効果、データ量の3つの観点から分析")
パターン3：自己反思検索エージェント
Agentはドキュメントを検索した後、自ら品質を評価します——品質が不十分なら、クエリを書き直して再検索。これこそがAgentic RAGのコア能力：自己修正。
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class ReflectionState:
question: str
documents: list = field(default_factory=list)
filtered_docs: list = field(default_factory=list)
answer: str = ""
iterations: int = 0
max_iterations: int = 3
need_more_search: bool = False
class SelfReflectiveAgent:
def __init__(self, llm, retriever, min_relevant: int = 2):
self.llm = llm
self.retriever = retriever
self.min_relevant = min_relevant
def retrieve(self, state: ReflectionState) -> dict:
docs = self.retriever.similarity_search(state.question, k=5)
return {"documents": docs}
def grade_documents(self, state: ReflectionState) -> dict:
filtered = []
for doc in state.documents:
score = self.llm.invoke(
f"Is this document relevant to '{state.question}'? "
f"Answer yes or no:\n{doc.page_content[:500]}"
)
if "yes" in score.lower():
filtered.append(doc)
return {
"filtered_docs": filtered,
"need_more_search": len(filtered) < self.min_relevant,
}
def rewrite_query(self, state: ReflectionState) -> dict:
better = self.llm.invoke(
f"Rewrite this question for better search results: {state.question}"
)
return {"question": better, "iterations": state.iterations + 1}
def generate(self, state: ReflectionState) -> dict:
context = "\n".join([d.page_content for d in state.filtered_docs])
answer = self.llm.invoke(
f"Based on:\n{context}\n\nAnswer: {state.question}"
)
return {"answer": answer}
def run(self, question: str) -> dict:
state = ReflectionState(question=question)
self.retrieve(state)
while state.iterations < state.max_iterations:
self.grade_documents(state)
if not state.need_more_search:
break
self.rewrite_query(state)
self.retrieve(state)
self.generate(state)
return {"answer": state.answer, "iterations": state.iterations}
agent = SelfReflectiveAgent(llm, vector_store)
result = agent.run("プロダクション環境でAgentic RAGをデプロイするには？")
パターン4：ツール拡張エージェント
Agentはドキュメントを検索するだけでなく、検索エンジン、計算機、データベースなどの外部ツールを呼び出すことができます。これによりAgentの能力の境界が大幅に拡張されます。
from typing import Annotated
def search_web(query: str) -> str:
"""Search the web for up-to-date information."""
results = web_search_engine.search(query, max_results=3)
return "\n".join([r.snippet for r in results])
def query_database(sql: str) -> str:
"""Execute SQL query against the product database."""
rows = db.execute(sql)
return str(rows[:20])
def calculate(expression: str) -> str:
"""Safely evaluate a mathematical expression."""
allowed = set("0123456789+-*/.() ")
if not all(c in allowed for c in expression):
return "Error: invalid expression"
return str(eval(expression))
def search_knowledge_base(query: str) -> str:
"""Search internal knowledge base for documentation."""
docs = vector_store.similarity_search(query, k=3)
return "\n".join([d.page_content for d in docs])
tools = [search_web, query_database, calculate, search_knowledge_base]
tool_descriptions = "\n".join(
[f"- {t.__name__}: {t.__doc__}" for t in tools]
)
class ToolAugmentedAgent:
def __init__(self, llm, tools: list):
self.llm = llm
self.tools = {t.__name__: t for t in tools}
def run(self, question: str, max_steps: int = 5) -> dict:
messages = [{"role": "user", "content": question}]
steps = []
for _ in range(max_steps):
response = self.llm.invoke(
f"Available tools:\n{tool_descriptions}\n\n"
f"To use a tool, respond: USE: tool_name(arg)\n"
f"To give final answer, respond: ANSWER: your answer\n\n"
f"Question: {messages[-1]['content']}"
)
if response.startswith("ANSWER:"):
return {"answer": response[7:].strip(), "steps": steps}
if response.startswith("USE:"):
tool_call = response[4:].strip()
tool_name = tool_call.split("(")[0]
tool_arg = tool_call.split("(")[1].rstrip(")")
tool_fn = self.tools.get(tool_name)
if tool_fn:
observation = tool_fn(tool_arg)
steps.append({
"tool": tool_name,
"arg": tool_arg,
"result": observation[:200],
})
messages.append({
"role": "assistant",
"content": f"Used {tool_name}, got: {observation}",
})
return {"answer": "Max steps reached", "steps": steps}
agent = ToolAugmentedAgent(llm, tools)
result = agent.run("製品の月間アクティブユーザー数は？競合と比較してどう？")
パターン5：LangGraphワークフローオーケストレーション
最も完全なAgentic RAG実装——LangGraphのStateGraphを使って、検索→評価→書き直し→生成の完全なループを定義。条件分岐とチェックポイントをサポート。
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator
class AgentState(TypedDict):
question: str
documents: Annotated[list, operator.add]
answer: str
iterations: int
need_more_search: bool
def retrieve(state: AgentState) -> dict:
docs = vector_store.similarity_search(state["question"], k=5)
return {"documents": docs}
def grade_documents(state: AgentState) -> dict:
question = state["question"]
docs = state["documents"]
filtered = []
for doc in docs:
score = llm.invoke(f"Is this doc relevant to '{question}'? Answer yes/no: {doc.page_content}")
if "yes" in score.lower():
filtered.append(doc)
return {"documents": filtered, "need_more_search": len(filtered) < 2}
def generate(state: AgentState) -> dict:
context = "\n".join([d.page_content for d in state["documents"]])
answer = llm.invoke(f"Based on:\n{context}\n\nAnswer: {state['question']}")
return {"answer": answer}
def decide_to_search(state: AgentState) -> str:
if state["need_more_search"] and state["iterations"] < 3:
return "rewrite"
return "generate"
def rewrite_query(state: AgentState) -> dict:
better_query = llm.invoke(f"Rewrite this question for better search: {state['question']}")
return {"question": better_query, "iterations": state["iterations"] + 1}
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("rewrite", rewrite_query)
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade")
workflow.add_conditional_edges("grade", decide_to_search, {
"rewrite": "rewrite",
"generate": "generate",
})
workflow.add_edge("rewrite", "retrieve")
workflow.add_edge("generate", END)
app = workflow.compile()
result = app.invoke({"question": "RAGとファインチューニング、どちらが私のユースケースに適している？", "iterations": 0})
LangGraphの利点：ワークフローの可視化、チェックポイント復旧のサポート、柔軟な条件分岐、デバッグの容易さ。プロダクション環境での第一選択です。
落とし穴ガイド：5つのよくある罠
| # | 罠 | 表れ | 解決策 |
|---|---|---|---|
| 1 | 無限ループ | Agentがクエリを書き直し続け、永遠に満足しない | max_iterationsのハードリミットを設定（3-5回） |
| 2 | 過剰検索 | すべての質問で複数回の検索がトリガーされ、レイテンシが急増 | 単純な質問はファストパス、複雑な質問のみAgenticに |
| 3 | ツール幻覚 | Agentが存在しないツール名をでっち上げて呼び出す | ホワイトリストでツール名を厳格に検証 |
| 4 | コンテキスト爆発 | 複数回の検索ドキュメントがすべてプロンプトに詰め込まれる | 各ラウンドでスコア上位Top-Kのドキュメントのみ保持 |
| 5 | 採点バイアス | LLMのドキュメント関連性スコアが不安定 | 構造化出力（JSON Schema）で採点形式を制約 |
トラブルシューティング：10のよくあるエラー
| # | エラーメッセージ | 原因 | 解決策 |
|---|---|---|---|
| 1 | GraphRecursionError: Recursion limit reached |
AgentループがLangGraphのデフォルトステップ制限を超過 | recursion_limitパラメータを設定、または条件付きエッジのロジックを最適化 |
| 2 | KeyError: 'documents' |
Stateに期待されるフィールドが欠落 | ノードの戻り値にすべての必須フィールドが含まれているか確認 |
| 3 | ToolCallingError: Unknown tool |
Agentが未登録のツールを呼び出した | ツールリストを確認、ツール名と説明が一致しているか確認 |
| 4 | TokenLimitExceeded |
複数回の検索で蓄積されたドキュメントがコンテキストウィンドウを超過 | gradeノードでドキュメントをフィルタリング、generateに渡すドキュメント数を制限 |
| 5 | ValidationError: State schema mismatch |
Stateの型定義と実際のデータが一致しない | TypedDictのフィールド型、Annotatedの使用法が正しいか確認 |
| 6 | TimeoutError: LLM invocation timed out |
採点/生成ステップでLLMの応答が遅い | timeoutパラメータを設定、非同期呼び出しを検討 |
| 7 | ImportError: No module named 'langgraph' |
LangGraphがインストールされていない | pip install langgraph langchain-core |
| 8 | JSONDecodeError in structured output |
LLMが非標準JSONを返却 | with_structured_output()で構造化出力を強制 |
| 9 | RateLimitError: Too many requests |
複数回の検索でAPIレート制限に抵触 | exponential backoffリトライを追加、またはローカルモデルに切替 |
| 10 | AttributeError: 'NoneType' has no attribute |
ノードがNoneを返却 | 各ノードがdictを返すことを保証、Noneを返さない |
高度な最適化テクニック
1. ハイブリッドルーティング：単純な質問のファストパス
すべての質問にAgenticフローが必要なわけではありません。分類器を追加し、単純な質問はNaive RAG、複雑な質問のみAgentic RAGに振り分けることで、平均レイテンシを60%削減。
class HybridRouter:
def __init__(self, llm, naive_rag, agentic_rag):
self.llm = llm
self.naive_rag = naive_rag
self.agentic_rag = agentic_rag
def is_simple(self, question: str) -> bool:
result = self.llm.invoke(
f"Is this a simple factual question? Answer yes/no: {question}"
)
return "yes" in result.lower()
def run(self, question: str) -> dict:
if self.is_simple(question):
return self.naive_rag.query(question)
return self.agentic_rag.invoke({"question": question, "iterations": 0})
2. チェックポイント復旧：長時間ワークフローの状態保持
LangGraphはチェックポイントをサポート——Agentは中断後、前回の状態から再開可能。
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"question": "技術選定を分析して", "iterations": 0}, config=config)
3. 並列検索：複数のナレッジソースを同時に検索
import asyncio
async def parallel_retrieve(question: str, stores: list) -> list:
tasks = [store.asimilarity_search(question, k=3) for store in stores]
results = await asyncio.gather(*tasks)
return [doc for batch in results for doc in batch]
4. ホットクエリのキャッシュ
from functools import lru_cache
@lru_cache(maxsize=1000)
def cached_retrieve(query_hash: str, question: str) -> list:
return vector_store.similarity_search(question, k=5)
5. オブザーバビリティ：Agentの各決定ステップをトレース
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agentic_rag")
def traced_node(func):
def wrapper(state):
logger.info(f"Node: {func.__name__}, Input: {str(state)[:200]}")
result = func(state)
logger.info(f"Node: {func.__name__}, Output: {str(result)[:200]}")
return result
return wrapper
比較分析：Agentic RAG vs Naive RAG vs Advanced RAG vs Modular RAG
| 次元 | Naive RAG | Advanced RAG | Modular RAG | Agentic RAG |
|---|---|---|---|---|
| 検索戦略 | 単回ベクトル検索 | クエリ書き直し＋ハイブリッド検索 | プラグイン可能な検索モジュール | Agentが自律的に検索を決定 |
| 推論能力 | ✗ | ✗ | ✗ | ✓ マルチホップ推論 |
| 自己修正 | ✗ | △ リランキング | ✗ | ✓ 反思＋書き直し |
| ツール呼び出し | ✗ | ✗ | △ | ✓ マルチツール連携 |
| ワークフロー | リニアパイプライン | リニア＋前処理 | モジュラーパイプライン | ステートフルグラフ |
| 実装複雑度 | ★☆☆ | ★★☆ | ★★★ | ★★★★ |
| レイテンシ | ~1s | ~2s | ~2s | ~3-8s |
| 精度 | 60-70% | 75-85% | 80-88% | 85-95% |
| 適用シナリオ | シンプルなFAQ | 標準的なQ&A | カスタムシナリオ | 複雑なマルチホップ問題 |
★が多いほど実装複雑度が高い；✓対応 △部分対応 ✗非対応；レイテンシと精度は参考値
オンラインツール推奨
- JSONフォーマッター — Agent Stateと検索結果のJSON構造をフォーマット、デバッグに必須
- ハッシュ計算 — ドキュメント重複排除とクエリキャッシュのMD5/SHAハッシュ値を計算
- Curl→コード変換 — LLM APIデバッグのcurlコマンドをPythonコードに変換
まとめと展望
"The future of RAG is not better retrieval — it's smarter agents that know when and how to retrieve."
"RAGの未来はより良い検索ではなく、いつ検索すべきか、どう検索すべきかを知るより賢いAgentである。"
Agentic RAGは「実験的アーキテクチャ」から「プロダクション標準」へと移行しています。2026年の主要トレンド：
- ネイティブAgentサポート：LangGraph、LlamaIndexなどのフレームワークがAgentic RAGを第一級市民として扱う
- マルチAgent協調：複数の専門Agentが協力して複雑な検索タスクを完了
- 適応型戦略：Agentが質問の難易度に応じて検索深度を自動選択
- 説明性の強化：すべての検索決定に明確な推論チェーンが追跡可能
- コスト最適化：インテリジェントルーティング＋キャッシュで、Agentic RAGのコストをNaive RAGに近づける
Agentic RAGの選択原則：ルーターエージェントから始め、段階的にアップグレード。まずルーティングでナレッジベースの振り分けを解決し、次に自己反思で検索品質を向上させ、最後にLangGraphで完全なワークフローを編成。最初から最も複雑なものを構築せず——まずAgentを動かし、それから賢くする。
参考リンク
ブラウザローカルツールを無料で試す →
#Agentic RAG#Agent RAG#自主检索Agent#LLM Agent推理#RAG Agent工作流#2026#LangGraph Agentic RAG