2026年になって、RAG（Retrieval-Augmented Generation）の世界がだいぶ変わってきました。正直なところ、もう「ベクトル検索してプロンプトに突っ込むだけ」では全然通用しないんですよね。取得した文書がクエリと噛み合っていなかったり、ハルシネーションが混ざったり──プロダクション環境だとこういう問題が本当に致命的になります。
そこで注目されているのがAgentic RAGというアプローチです。
AIエージェントが「そもそも検索が必要か？」「取得した文書は使い物になるか？」「クエリを書き換えたほうがいいんじゃないか？」といった判断を自律的に行い、自己修正ループを回しながら最適な回答を導き出す。従来のRAGとは根本的に設計思想が違います。
この記事では、LangGraph 0.3系（2026年3月時点の最新安定版）を使って、Adaptive RAG・Corrective RAG・Self-Reflective RAGという3つのパターンを統合した実践的なAgentic RAGシステムをPythonで構築していきます。日本語の記事だと概念紹介で終わっているものが多かったので、動的クエリルーティングからハルシネーション検出まで、一気通貫で実装できるガイドにまとめました。
従来のRAGの限界とAgentic RAGの必要性
まず、なぜ従来のRAGでは不十分なのかを整理しておきましょう。
標準的なRAGパイプラインって、「入力 → ベクトル検索 → LLMに投げる → 回答」という一方向のフローなんですよね。シンプルで分かりやすい反面、構造的な弱点がいくつかあります。
- ワンショット検索の脆さ：一度の検索で関連文書が取得できなかった場合、リトライの仕組みがない
- 文書の品質チェックなし：取得した文書がクエリに対して本当に関連しているかを検証しない
- ルーティング不在：すべてのクエリに対して同じ検索戦略を適用してしまう（たとえば時事ネタにベクトル検索をかけても意味がない）
- ハルシネーションの検出なし：生成された回答が取得文書に基づいているかを確認しない
Agentic RAGは、これらの問題を状態機械（State Machine）として設計することで解決します。LangGraphの循環グラフを使えば、「検索 → 評価 → 必要なら再検索 → 生成 → 検証」というループが自然に表現できます。
Agentic RAGの3つの主要パターン
LangGraphで実装できるAgentic RAGには、主に3つのアーキテクチャパターンがあります。それぞれの特徴を押さえておきましょう。
Adaptive RAG（適応型RAG）
クエリの種類に応じて検索戦略を動的に切り替えるパターンです。シンプルな質問ならLLMの内部知識で回答し、専門的な質問ならベクトルストアを検索し、最新情報が必要ならWeb検索にルーティングする。いわば「賢い振り分け係」ですね。
Corrective RAG（修正型RAG）
取得した文書を生成前に評価するパターンです。文書の関連性スコアが低い場合、クエリを書き換えて再検索するか、Web検索にフォールバックします。個人的には、この文書グレーディングの仕組みが一番効果を実感しやすいパターンだと思います。
Self-Reflective RAG（自己反省型RAG）
生成した回答を自己評価するパターンです。回答がハルシネーションを含んでいないか、ユーザーの質問にちゃんと答えているかをチェックし、不合格なら追加コンテキストで再生成します。
今回の実装では、この3つを全部統合します。これが最も堅牢な構成です。
環境構築とインストール
さて、さっそく実装に入りましょう。まずは必要なライブラリのインストールから。Python 3.10以上が必要です。
pip install langgraph langchain langchain-openai langchain-community \
chromadb tavily-python pydanticAPIキーを環境変数に設定します。
export OPENAI_API_KEY="sk-..."
export TAVILY_API_KEY="tvly-..."TavilyはWeb検索用のAPIで、Adaptive RAGのルーティング先として使います。無料プランでも月1,000リクエストまで利用できるので、試すだけなら十分です。
ステップ1：ベクトルストアの構築
最初のステップとして、ChromaDBでローカルのベクトルストアを構築します。ここではAIエージェントに関する技術文書をインデックスする例を見ていきましょう。
from langchain_community.document_loaders import WebBaseLoader
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
# ドキュメントの読み込み（例：技術ブログ記事）
urls = [
"https://lilianweng.github.io/posts/2023-06-23-agent/",
"https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
"https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",
]
docs = []
for url in urls:
loader = WebBaseLoader(url)
docs.extend(loader.load())
# テキスト分割（チャンクサイズとオーバーラップの調整が重要）
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=100,
)
splits = text_splitter.split_documents(docs)
# ChromaDBにベクトルインデックスを構築
vectorstore = Chroma.from_documents(
documents=splits,
embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
collection_name="agentic-rag-docs",
)
# Retrieverの作成（上位4件を取得）
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})chunk_size=500とchunk_overlap=100は、まずはこのあたりから始めてみるのがおすすめです。実際のプロジェクトでは文書の性質に応じて調整が必要になります。日本語文書を扱う場合は、形態素解析との組み合わせも検討してみてください（これについてはFAQでも触れています）。
ステップ2：グラフの状態定義
LangGraphの核となるStateを定義します。グラフ内のすべてのノードがこの状態を共有して、更新していく仕組みです。
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from langchain_core.documents import Document
class AgenticRAGState(TypedDict):
"""Agentic RAGグラフの共有状態"""
question: str # ユーザーの元の質問
generation: str # LLMが生成した回答
documents: list[Document] # 取得した文書リスト
query_rewrite_count: int # クエリ書き換え回数（無限ループ防止）
route: Literal["vectorstore", "web_search", "direct"] # ルーティング先ここで地味に大事なのがquery_rewrite_countです。これは自己修正ループの安全弁で、これがないと文書が見つからない場合に無限リトライし続ける危険があります。プロダクション環境では絶対に入れておくべきフィールドです。
ステップ3：クエリルーターの実装
Adaptive RAGの要となるクエリルーターを作ります。LLMがクエリの内容を分析して、最適な検索戦略を選ぶ部分ですね。
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
class RouteDecision(BaseModel):
"""クエリルーティングの判断結果"""
datasource: Literal["vectorstore", "web_search", "direct"] = Field(
description="クエリに最適なデータソース。"
"vectorstore: 技術文書に関する質問、"
"web_search: 最新情報や時事ネタ、"
"direct: LLMの知識で直接回答可能な一般的な質問"
)
reasoning: str = Field(description="ルーティング判断の理由")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
router_llm = llm.with_structured_output(RouteDecision)
ROUTER_PROMPT = """あなたはクエリルーターです。
ユーザーの質問を分析し、最適な情報源を選択してください。
判断基準：
- vectorstore: AIエージェント、プロンプトエンジニアリング、LLM攻撃に関する技術的な質問
- web_search: 最新ニュース、現在の日付・時刻、リアルタイム情報が必要な質問
- direct: 一般常識や基本的な概念に関する質問（検索不要）
ユーザーの質問: {question}"""
def route_question(state: AgenticRAGState) -> AgenticRAGState:
"""クエリを分析し、最適なデータソースにルーティングする"""
question = state["question"]
result = router_llm.invoke(ROUTER_PROMPT.format(question=question))
return {**state, "route": result.datasource}ルーターにgpt-4o-miniを使っている点がポイントです。ルーティングは分類タスクとしては比較的シンプルなので、安価なモデルで問題ありません。コスト最適化の鉄則は「判断系は軽量モデル、生成系は高性能モデル」です。これだけで結構コストが変わってきます。
ステップ4：文書グレーダーの実装
ここからがCorrective RAGの核心部分です。取得した文書がクエリに対して本当に関連しているかどうか、LLMに評価させます。
class GradeDocument(BaseModel):
"""文書の関連性評価"""
is_relevant: bool = Field(
description="文書がクエリに関連しているかどうか"
)
confidence: float = Field(
description="判断の確信度（0.0〜1.0）"
)
grader_llm = llm.with_structured_output(GradeDocument)
GRADER_PROMPT = """あなたは文書の関連性を評価する採点者です。
以下の文書がユーザーの質問に対して関連する情報を含んでいるかを判断してください。
完全一致は不要です。質問に答えるための手がかりが含まれていれば関連ありと判断してください。
質問: {question}
文書: {document}"""
def grade_documents(state: AgenticRAGState) -> AgenticRAGState:
"""取得した文書の関連性を評価し、関連文書のみを残す"""
question = state["question"]
documents = state["documents"]
relevant_docs = []
for doc in documents:
grade = grader_llm.invoke(
GRADER_PROMPT.format(
question=question,
document=doc.page_content
)
)
if grade.is_relevant:
relevant_docs.append(doc)
return {**state, "documents": relevant_docs}このグレーダーがAgentic RAGの「目」にあたる部分です。
従来のRAGだと取得した文書をそのままLLMに渡していましたが、ここで関連性の低い文書をフィルタリングすることで、ハルシネーションのリスクをかなり減らせます。実際にこれを入れるか入れないかで回答の質が目に見えて変わるので、ぜひ試してみてください。
ステップ5：クエリ書き換えと検索ノード
グレーディングの結果、関連文書が足りなかった場合にクエリを書き換えるノードと、実際の検索を行うノードを実装します。
from langchain_community.tools.tavily_search import TavilySearchResults
web_search_tool = TavilySearchResults(max_results=3)
def retrieve_from_vectorstore(state: AgenticRAGState) -> AgenticRAGState:
"""ベクトルストアから文書を取得する"""
question = state["question"]
documents = retriever.invoke(question)
return {**state, "documents": documents}
def web_search(state: AgenticRAGState) -> AgenticRAGState:
"""Web検索を実行する"""
question = state["question"]
results = web_search_tool.invoke({"query": question})
web_docs = [
Document(
page_content=r["content"],
metadata={"source": r["url"]}
)
for r in results
]
return {**state, "documents": web_docs}
def rewrite_query(state: AgenticRAGState) -> AgenticRAGState:
"""クエリを書き換えてより良い検索結果を得る"""
question = state["question"]
count = state.get("query_rewrite_count", 0)
rewrite_prompt = f"""以下の質問をより検索に適した形に書き換えてください。
元の意図は保ったまま、より具体的で検索エンジンに適した表現にしてください。
元の質問: {question}
書き換えた質問のみを出力してください。"""
rewritten = llm.invoke(rewrite_prompt)
return {
**state,
"question": rewritten.content,
"query_rewrite_count": count + 1,
}クエリ書き換えは地味な機能に見えますが、検索結果の質に大きく影響します。ユーザーの曖昧な質問を、ベクトル検索に適した具体的な表現に変換してくれるわけです。
ステップ6：回答生成とハルシネーション検出
いよいよSelf-Reflective RAGの核心に入ります。回答を生成して、さらにその回答が取得文書にちゃんと基づいているかを検証するパートです。
class HallucinationCheck(BaseModel):
"""ハルシネーション検出の結果"""
is_grounded: bool = Field(
description="回答が提供された文書に基づいているかどうか"
)
is_useful: bool = Field(
description="回答がユーザーの質問に実際に答えているかどうか"
)
hallucination_llm = llm.with_structured_output(HallucinationCheck)
GENERATE_PROMPT = """以下の文書を参考にして、ユーザーの質問に回答してください。
文書に記載されている情報のみを使用し、推測や補完は行わないでください。
文書:
{context}
質問: {question}"""
HALLUCINATION_PROMPT = """以下の回答が、提供された文書の内容に基づいているかを評価してください。
文書:
{documents}
回答: {generation}
質問: {question}"""
def generate_answer(state: AgenticRAGState) -> AgenticRAGState:
"""取得した文書に基づいて回答を生成する"""
question = state["question"]
documents = state["documents"]
context = "\n\n".join(doc.page_content for doc in documents)
generation_llm = ChatOpenAI(model="gpt-4o", temperature=0)
response = generation_llm.invoke(
GENERATE_PROMPT.format(context=context, question=question)
)
return {**state, "generation": response.content}
def check_hallucination(state: AgenticRAGState) -> AgenticRAGState:
"""生成された回答のハルシネーションをチェックする"""
documents = state["documents"]
generation = state["generation"]
question = state["question"]
doc_texts = "\n\n".join(doc.page_content for doc in documents)
result = hallucination_llm.invoke(
HALLUCINATION_PROMPT.format(
documents=doc_texts,
generation=generation,
question=question,
)
)
return {**state, "hallucination_check": result}ここでもモデルの使い分けがキモになります。回答生成にはgpt-4o、グレーディングやハルシネーションチェックにはgpt-4o-mini。生成品質は最終出力に直結するので、ここだけは高性能モデルに投資する価値があります。
ステップ7：グラフの組み立てと条件付きエッジ
さて、ここが一番ワクワクする部分です。すべてのノードをLangGraphのStateGraphで接続していきます。
from langgraph.graph import StateGraph, END, START
def decide_after_routing(state: AgenticRAGState) -> str:
"""ルーティング結果に基づいて次のノードを決定する"""
route = state["route"]
if route == "web_search":
return "web_search"
elif route == "vectorstore":
return "retrieve"
else:
return "generate"
def decide_after_grading(state: AgenticRAGState) -> str:
"""文書グレーディング結果に基づいて次のアクションを決定する"""
documents = state["documents"]
rewrite_count = state.get("query_rewrite_count", 0)
if len(documents) == 0 and rewrite_count < 2:
# 関連文書なし → クエリ書き換えてリトライ（最大2回）
return "rewrite"
elif len(documents) == 0 and rewrite_count >= 2:
# リトライ上限到達 → Web検索にフォールバック
return "web_search"
else:
# 関連文書あり → 回答生成へ
return "generate"
def decide_after_hallucination_check(state: AgenticRAGState) -> str:
"""ハルシネーションチェック結果に基づいて判断する"""
check = state.get("hallucination_check")
if check and check.is_grounded and check.is_useful:
return "end"
else:
return "generate"
# グラフの構築
workflow = StateGraph(AgenticRAGState)
# ノードの追加
workflow.add_node("route", route_question)
workflow.add_node("retrieve", retrieve_from_vectorstore)
workflow.add_node("web_search", web_search)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("rewrite", rewrite_query)
workflow.add_node("generate", generate_answer)
workflow.add_node("check_hallucination", check_hallucination)
# エッジの定義
workflow.add_edge(START, "route")
# ルーティング後の条件分岐
workflow.add_conditional_edges(
"route",
decide_after_routing,
{
"web_search": "web_search",
"retrieve": "retrieve",
"generate": "generate",
}
)
# 検索後 → グレーディング
workflow.add_edge("retrieve", "grade_documents")
workflow.add_edge("web_search", "grade_documents")
# グレーディング後の条件分岐
workflow.add_conditional_edges(
"grade_documents",
decide_after_grading,
{
"rewrite": "rewrite",
"web_search": "web_search",
"generate": "generate",
}
)
# クエリ書き換え後 → 再検索
workflow.add_edge("rewrite", "retrieve")
# 生成後 → ハルシネーションチェック
workflow.add_edge("generate", "check_hallucination")
# ハルシネーションチェック後の条件分岐
workflow.add_conditional_edges(
"check_hallucination",
decide_after_hallucination_check,
{
"end": END,
"generate": "generate",
}
)
# グラフのコンパイル
app = workflow.compile()このグラフの全体像を図にすると、こんなフローになります。
START → route ─┬─→ retrieve → grade_documents ─┬─→ generate → check_hallucination ─┬─→ END
│ │ │
├─→ web_search ──→ grade_documents ├─→ rewrite → retrieve（ループ） └─→ generate（再生成）
│ │
└─→ generate（直接回答） └─→ web_search（フォールバック）条件分岐のポイントはdecide_after_gradingです。関連文書がゼロだった場合、まずクエリを書き換えてリトライし、それでもダメならWeb検索にフォールバックする。この「段階的エスカレーション」がAgentic RAGの強みです。
ステップ8：実行とテスト
構築したAgentic RAGグラフを実際に動かしてみましょう。
# ベクトルストア向けのクエリ
result = app.invoke({
"question": "AIエージェントのメモリアーキテクチャについて説明してください",
"generation": "",
"documents": [],
"query_rewrite_count": 0,
"route": "",
})
print("回答:", result["generation"])
print("ルーティング先:", result["route"])
print("参照文書数:", len(result["documents"]))
print("---")
# Web検索向けのクエリ
result = app.invoke({
"question": "2026年3月のAI業界の最新ニュースは？",
"generation": "",
"documents": [],
"query_rewrite_count": 0,
"route": "",
})
print("回答:", result["generation"])最初のクエリは「AIエージェントのメモリアーキテクチャ」という技術的な質問なので、ベクトルストアにルーティングされるはずです。二番目は最新ニュースなので、Web検索に向かいます。ルーターがちゃんと振り分けてくれているか、ここで確認しておきましょう。
プロダクション環境での考慮事項
動くものができたら、次はプロダクション環境への移行です。ここからが本番ですね。押さえておくべきポイントをいくつか紹介します。
コスト最適化
繰り返しになりますが、判断系ノード（ルーター、グレーダー、ハルシネーションチェック）にはgpt-4o-mini、生成ノードにはgpt-4oを使い分けるのが基本です。それと、グレーディングの呼び出し回数は文書数に比例するので、Retrieverのk値を必要以上に大きくしないことも意外と大事だったりします。
レイテンシの管理
自己修正ループは、当然ながらレイテンシを増やします。ユーザー体験を損なわないために、クエリ書き換えの最大回数を制限し（上の実装では2回にしています）、LangGraphのストリーミング機能を活用して中間結果をリアルタイム表示するのがおすすめです。
チェックポイントの活用
LangGraphのMemorySaverやSqliteSaverを使えば、グラフの実行状態を永続化できます。長時間かかるタスクや、途中でユーザーの承認が必要なワークフローで重宝します。
from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
# スレッドIDを指定して実行（状態が永続化される）
config = {"configurable": {"thread_id": "user-session-123"}}
result = app.invoke(initial_state, config=config)Human-in-the-Loop
高リスクな判断の前にユーザー承認を挟みたい場合は、interrupt_beforeが使えます。指定したノードの実行前にグラフが一時停止してくれます。
# 回答生成前にユーザー承認を求める
app = workflow.compile(
checkpointer=memory,
interrupt_before=["generate"]
)LangSmithによる可観測性
プロダクション環境では、各ノードの実行結果やLLMの入出力、レイテンシの監視が欠かせません。LangSmithと連携すれば、エージェントの動作をトレースしてパフォーマンスのボトルネックを特定できます。正直、デバッグのしやすさが段違いなので、本番運用するなら導入を強くおすすめします。
よくある質問（FAQ）
Agentic RAGと通常のRAGの違いは何ですか？
通常のRAGは「検索→生成」のワンショット処理です。一方、Agentic RAGではAIエージェントが自律的に検索戦略を選択し、取得文書の品質を評価し、必要に応じてクエリを書き換えたり検索先を変更したりします。この自己修正ループがあることで、回答の正確性と信頼性が大幅に向上します。
LangGraphとLangChainの違いは何ですか？
LangChainは一方向のチェーン（DAG）でLLMアプリケーションを構築するフレームワークです。LangGraphはその拡張で、循環グラフ──つまりループや条件分岐──をサポートします。Agentic RAGのような自己修正ループが必要なケースでは、LangGraphが必須になります。
Agentic RAGのコストはどのくらい増えますか？
自己修正ループが発生する分、通常のRAGよりLLM呼び出し回数は増えます。ただ、判断系ノードに安価なモデルを使い、生成ノードだけ高性能モデルにすることで、コスト増はかなり抑えられます。典型的なケースで、通常のRAGの1.5〜3倍程度のトークン消費になるイメージです。
OpenAI以外のモデルでも使えますか？
もちろん使えます。LangGraphはモデル非依存のフレームワークです。この記事ではOpenAIモデルを使いましたが、langchain-anthropicでClaude、langchain-google-genaiでGemini、Ollama経由でローカルモデルなど、好きなLLMに差し替え可能です。
日本語文書でもAgentic RAGは有効ですか？
はい、有効です。ただし日本語文書の場合、チャンキング戦略には少し注意が必要です。RecursiveCharacterTextSplitterはスペース区切りを前提としているため、日本語ではMeCabやSudachiなどの形態素解析器と組み合わせたカスタムスプリッターの導入を検討してみてください。埋め込みモデルも多言語対応のもの（OpenAIのtext-embedding-3-smallは日本語対応済みです）を選ぶことが重要です。