はじめに
検索拡張生成（Retrieval-Augmented Generation）は、シンプルなアイデアから始まりました：言語モデルのパラメトリックメモリだけに頼るのではなく、外部のナレッジベースから関連するドキュメントを検索し、プロンプトのコンテキストに含めるというものです。このナイーブRAGパターン、つまりエンベッド-検索-生成で構成されるパイプラインは、構造化されたドキュメントコレクションに対するシンプルな質問応答で驚くほどうまく機能しました。しかし、チームがより複雑なユースケースのためにRAGをプロダクションに導入すると、その限界が痛いほど明らかになりました。複数のドキュメントにまたがる推論を必要とするクエリ、明確化が必要な曖昧な質問、異質なコンテンツタイプを持つナレッジベースのすべてが、検索してから生成するパイプラインの脆弱性を露呈しました。
エージェンティックRAGは、検索システムのアーキテクチャにおける根本的な転換を表しています。各ステップが直線的に次のステップに入力される固定パイプラインの代わりに、エージェンティックRAGは言語モデルに検索戦略を計画し、検索結果の品質を評価し、初回の検索が失敗した場合にクエリを再構成し、最終回答を生成するのに十分な情報があるかを判断する能力を与えます。モデルは検索されたコンテキストの受動的な消費者ではなく、検索プロセスの能動的な参加者となります。
このガイドでは、エージェンティックRAGシステムを構築するためのアーキテクチャパターン、実装フレームワーク、評価手法、プロダクションにおける考慮事項を扱います。エンタープライズのナレッジベース、カスタマーサポートシステム、技術ドキュメントプラットフォームでこれらのパターンが大規模に検証された実際のデプロイメントに基づいています。
ナイーブRAGからエージェンティックRAGへ：何が変わったか
ナイーブRAGは決定論的な3段階のパイプラインに従います：ユーザークエリがエンベッドされ、ベクトル類似度検索が最も類似したtop-kのドキュメントチャンクを検索し、それらのチャンクがLLMが回答を生成するために使用するプロンプトに連結されます。このパイプラインにはエージェンティックRAGが対処する3つの根本的な弱点があります。
第一に、ナイーブRAGはユーザーのクエリがすでに検索に適した形で構成されていると仮定します。実際には、ユーザークエリはしばしば曖昧であったり、多面的であったり、インデックスされたドキュメントの語彙と一致しない用語を使用したりします。デプロイメントの失敗について尋ねるユーザーは、エラーハンドリング、インフラストラクチャ設定、CI/CDパイプラインに関するドキュメントが必要かもしれませんが、単一のベクトル検索ではこれらの側面の1つだけしか返さない場合があります。
第二に、ナイーブRAGには品質ゲートがありません。検索されたドキュメントが無関係、古い、または不十分であっても、パイプラインはそのまま進行し、LLMは質の低いコンテキストから回答を生成します。システムが検索の失敗を認識して再試行するメカニズムがありません。
第三に、ナイーブRAGはすべてのクエリを同一に扱います。事実検索の質問、複雑な分析的質問、複数のソースからの合成を必要とする質問のすべてが同じ検索-生成パイプラインを通過します。エージェンティックRAGは、クエリの特性に基づいて異なる検索および生成戦略を選択する条件分岐ロジックを導入します。
ナイーブからエージェンティックRAGへの移行は、3つの機能の追加を伴います：クエリ計画（複雑なクエリをサブクエリに分解）、検索評価（検索されたコンテキストが十分かどうかの評価）、反復的改善（結果が不十分な場合のクエリの再構成と再検索）。これらの機能により、静的なパイプラインが動的で自己修正するシステムに変換されます。
コアアーキテクチャパターン
エージェンティックRAGシステムは、少数の組み合わせ可能なパターンから構築されます。これらのパターンを理解することで、特定のユースケースに合わせたシステムを設計できます。
ルーティング
最もシンプルなエージェンティックパターンは、クエリ分類に基づいてクエリを異なる検索バックエンドにルーティングします。ルーターエージェントが受信クエリを分析し、最も適切なナレッジソースに振り分けます：
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
router_prompt = ChatPromptTemplate.from_messages([
("system", """Classify the user query into one of these categories:
- technical_docs: Questions about API usage, configuration, or code
- policy: Questions about company policies, procedures, or compliance
- support: Questions about troubleshooting or known issues
- general: General questions that don't fit other categories
Respond with ONLY the category name."""),
("human", "{query}")
])
router_chain = router_prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()
ルーティングは、ナレッジベースが異なるインデキシング戦略を持つ複数のドメインにまたがる場合に有用です。技術ドキュメントはコード対応のチャンキングとエンベディングでインデックスされ、ポリシードキュメントはメタデータフィルター付きのセマンティックチャンキングを使用するかもしれません。
クエリ分解
複雑な質問は、単一の検索では一緒に出現しない複数のドキュメントチャンクからの情報を必要とすることがよくあります。クエリ分解は複雑なクエリを独立したサブクエリに分割し、各々について検索し、結果を合成します：
decomposition_prompt = ChatPromptTemplate.from_messages([
("system", """Break the following complex question into 2-4 simpler
sub-questions that, when answered together, provide a complete answer
to the original question. Return as a JSON array of strings."""),
("human", "{query}")
])
# Example: "How does our API rate limiting compare to competitors
# and what are customers saying about it?"
# Decomposes to:
# ["What are our current API rate limiting policies?",
# "What rate limiting do our main competitors use?",
# "What customer feedback have we received about rate limiting?"]
自己修正
自己修正は、エージェンティックRAGをナイーブRAGと最も区別するパターンです。検索後、評価ステップが検索されたドキュメントが関連性があり十分かどうかを評価します。十分でない場合、システムはクエリを再構成して再試行します：
grading_prompt = ChatPromptTemplate.from_messages([
("system", """You are a retrieval quality grader. Given a user question
and a set of retrieved documents, determine:
1. Are the documents relevant to the question? (yes/no)
2. Do the documents contain sufficient information to answer? (yes/no)
3. If insufficient, suggest a reformulated query.
Respond as JSON with keys: relevant, sufficient, reformulated_query"""),
("human", "Question: {query}\n\nDocuments: {documents}")
])
このパターンは通常、エレガントな失敗メッセージにフォールバックする前に2〜3回の検索試行を許可します。再構成されたクエリは、セマンティック類似度検索からキーワードベースの検索への切り替え、範囲の拡大、特定のメタデータフィールドのターゲティングなどを行います。
ステートフルエージェントオーケストレーションのためのLangGraph
LangGraphは、LangChainのよりシンプルなチェーンベースの抽象化では表現できない明示的な状態管理、条件付きルーティング、サイクルサポートを提供するため、エージェンティックRAGシステム構築の標準フレームワークとして台頭しました。
LangGraphベースのエージェンティックRAGシステムは、ノードが処理ステップを、エッジがステップ間の遷移を表す状態グラフとして定義されます。条件付きエッジにより、グラフは中間結果に基づいて分岐できます：
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
from operator import add
class AgentState(TypedDict):
query: str
sub_queries: List[str]
documents: List[dict]
generation: str
retry_count: int
retrieval_grade: str
def route_query(state: AgentState) -> AgentState:
"""Classify and route the incoming query."""
query = state["query"]
classification = router_chain.invoke({"query": query})
state["route"] = classification
return state
def retrieve(state: AgentState) -> AgentState:
"""Retrieve documents based on query and route."""
query = state["query"]
docs = retriever.invoke(query)
state["documents"] = docs
return state
def grade_documents(state: AgentState) -> AgentState:
"""Grade retrieved documents for relevance and sufficiency."""
grade = grading_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["retrieval_grade"] = grade["sufficient"]
state["retry_count"] = state.get("retry_count", 0) + 1
if not grade["sufficient"] and grade.get("reformulated_query"):
state["query"] = grade["reformulated_query"]
return state
def generate(state: AgentState) -> AgentState:
"""Generate final answer from retrieved context."""
answer = generation_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["generation"] = answer
return state
def should_retry(state: AgentState) -> str:
"""Decide whether to retry retrieval or proceed to generation."""
if state["retrieval_grade"] == "yes":
return "generate"
if state["retry_count"] >= 3:
return "generate" # Give up and generate with what we have
return "retrieve"
# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("route", route_query)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade", grade_documents)
workflow.add_node("generate", generate)
workflow.set_entry_point("route")
workflow.add_edge("route", "retrieve")
workflow.add_edge("retrieve", "grade")
workflow.add_conditional_edges("grade", should_retry, {
"retrieve": "retrieve",
"generate": "generate"
})
workflow.add_edge("generate", END)
app = workflow.compile()
グラフ構造により、制御フローが明示的でデバッグ可能になります。グラフを可視化し、各ノードを通じた実行をトレースし、任意のクエリに対してシステムが特定のパスを取った理由を正確に理解できます。この透明性は、障害を診断し動作を説明する必要があるプロダクションシステムにとって不可欠です。
LangGraphは、複数のインデックスにわたる並列検索、エージェントがユーザー確認のために一時停止するヒューマン・イン・ザ・ループのチェックポイント、会話間で永続する状態など、より高度なパターンもサポートしています。
検索戦略
エージェンティックRAGシステムの検索レイヤーは、通常、単一のベクトルストアルックアップよりも洗練されています。プロダクションシステムは複数の検索戦略を組み合わせます。
ハイブリッド検索
ハイブリッド検索は、密なベクトル検索と疎なキーワード検索を組み合わせ、エンベディングのセマンティックな理解とBM25キーワードマッチングの精度の両方を提供します：
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
from langchain_chroma import Chroma
vectorstore = Chroma(
collection_name="documents",
embedding_function=embedding_model,
)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 10
hybrid_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.6, 0.4],
)
リランキング
初回検索後、クロスエンコーダーリランカーがバイエンコーダーの類似度よりもはるかに高い精度で各ドキュメントをクエリに対してスコアリングします。計算コストは高いですが、精度を劇的に向上させます：
from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
reranker = CohereRerank(model="rerank-v3.5", top_n=5)
compression_retriever = ContextualCompressionRetriever(
base_compressor=reranker,
base_retriever=hybrid_retriever,
)
マルチインデックス検索
プロダクションのナレッジベースは、異なるスキーマ、エンベディングモデル、ドキュメントタイプを持つ複数のインデックスにまたがることがよくあります。エージェンティックシステムは複数のインデックスを並列にクエリし、結果をマージできます：
async def multi_index_retrieve(query: str, indexes: list) -> list:
"""Retrieve from multiple indexes in parallel."""
import asyncio
async def retrieve_from_index(index, query):
return await index.aretrieve(query)
tasks = [retrieve_from_index(idx, query) for idx in indexes]
results = await asyncio.gather(*tasks)
# Flatten and deduplicate
all_docs = []
seen_ids = set()
for result_set in results:
for doc in result_set:
if doc.metadata["id"] not in seen_ids:
all_docs.append(doc)
seen_ids.add(doc.metadata["id"])
return all_docs
RAGASとDeepEvalによる評価
エージェンティックRAGシステムの評価には、複数の品質次元の測定が必要です。RAGASとDeepEvalは最も広く採用されている2つの評価フレームワークで、それぞれ異なる強みを持っています。
RAGASは、すべてのテストクエリに対するグラウンドトゥルースの回答を必要とせずにRAGの品質を評価するリファレンスフリーのメトリクスセットを提供します：
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
)
from datasets import Dataset
eval_dataset = Dataset.from_dict({
"question": questions,
"answer": generated_answers,
"contexts": retrieved_contexts,
"ground_truth": reference_answers,
})
results = evaluate(
dataset=eval_dataset,
metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall,
],
)
print(results)
忠実性（Faithfulness）は、生成された回答が検索されたコンテキストに裏付けられているかを測定し、ハルシネーションを検出します。回答関連性は、回答が質問に対応しているかを測定します。コンテキスト精度は、検索されたドキュメントが関連しているかを測定し、コンテキスト再現率は、検索が必要なすべての情報を捕捉したかを測定します。
DeepEvalは、エージェンティックシステムに特に関連する追加メトリクスで評価を拡張します：
from deepeval import evaluate
from deepeval.metrics import (
FaithfulnessMetric,
AnswerRelevancyMetric,
ContextualRelevancyMetric,
HallucinationMetric,
)
from deepeval.test_case import LLMTestCase
test_case = LLMTestCase(
input="How do I configure rate limiting?",
actual_output=generated_answer,
expected_output=reference_answer,
retrieval_context=retrieved_chunks,
)
metrics = [
FaithfulnessMetric(threshold=0.8),
AnswerRelevancyMetric(threshold=0.7),
ContextualRelevancyMetric(threshold=0.7),
HallucinationMetric(threshold=0.5),
]
evaluate(test_cases=[test_case], metrics=metrics)
エージェンティックRAGに特化して、検索効率も測定する必要があります：システムが平均で何回の検索ラウンドを必要とするか、クエリの何パーセントが再構成を必要とするか、リトライの試行に伴い回答品質がどのように劣化するか。これらのメトリクスはエージェンティックループに固有のものであり、標準的なRAG評価フレームワークではカバーされていません。
プロダクションパターン
エージェンティックRAGをプロトタイプからプロダクションに移行するには、開発中には浮上しない信頼性、安全性、運用上の懸念に対処する必要があります。
ガードレール
すべてのプロダクションRAGシステムには、LLMが有害な、トピック外の、または事実に基づかない応答を生成することを防ぐガードレールが必要です：
from guardrails import Guard
from guardrails.hub import ToxicLanguage, CompetitorCheck
guard = Guard().use_many(
ToxicLanguage(on_fail="exception"),
CompetitorCheck(
competitors=["competitor_a", "competitor_b"],
on_fail="fix"
),
)
raw_response = generation_chain.invoke({"query": query, "documents": docs})
validated_response = guard.validate(raw_response)
フォールバックチェーン
エージェンティックループが最大リトライ後も十分なコンテキストを見つけられない場合、信頼性の低い回答を生成するのではなく、エレガントな劣化が必要です：
def generate_with_fallback(state: AgentState) -> AgentState:
if state["retrieval_grade"] != "yes" and state["retry_count"] >= 3:
state["generation"] = (
"I wasn't able to find sufficient information in our knowledge base "
"to fully answer your question. Here's what I found:\n\n"
f"{partial_answer_from_context(state['documents'])}\n\n"
"For a complete answer, I'd recommend contacting the support team."
)
state["confidence"] = "low"
else:
state["generation"] = generation_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["confidence"] = "high"
return state
ヒューマン・イン・ザ・ループ
高リスクなアプリケーションでは、LangGraphはエージェントが実行を一時停止して人間の承認を待つ割り込みポイントをサポートしています：
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
# Add interrupt before generation for sensitive queries
workflow.add_node("human_review", lambda state: state)
workflow.add_conditional_edges(
"grade",
lambda state: "human_review" if state.get("sensitive") else "generate",
{"human_review": "human_review", "generate": "generate"}
)
app = workflow.compile(
checkpointer=checkpointer,
interrupt_before=["human_review"]
)
LangSmithとPhoenixによるオブザーバビリティ
オブザーバビリティは、プロダクションのエージェンティックシステムでは妥協できません。LLM駆動の制御フローの非決定的な性質は、あらゆる可能な実行パスを予測またはテストできないことを意味します。プロダクションでシステムが何をしているかを理解するには、包括的なトレーシングが必要です。
LangSmithはLangGraphアプリケーション向けのエンドツーエンドのトレーシングを提供します：
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "agentic-rag-production"
# All LangGraph invocations are automatically traced
result = app.invoke({"query": "How do I configure rate limiting?"})
各トレースは、グラフを通じた完全な実行パスを示します：どのノードが訪問されたか、各ステップでのLLMの入出力、検索結果、評価判断、トークン数。これは、エージェントが予期しないパスを取ったり、品質の低い回答を生成したりするケースのデバッグに不可欠です。
Arize Phoenixは、検索品質モニタリングに焦点を当てたオープンソースの代替手段を提供します：
import phoenix as px
from phoenix.trace.langchain import LangChainInstrumentor
px.launch_app()
LangChainInstrumentor().instrument()
# Traces are now collected in Phoenix
result = app.invoke({"query": user_query})
Phoenixは、時間の経過に伴う検索品質のモニタリングに特に有用です。エンベディングのドリフト、検索関連性の分布を追跡し、検索品質が低下した場合にアラートを出すことができます。これは多くの場合、ナレッジベースがエンベディングモデルの学習分布からドリフトしており、再インデキシングが必要であることを示しています。
プロダクションでモニタリングすべき主要メトリクス：
# Custom metrics to track
metrics = {
"avg_retrieval_rounds": [],
"reformulation_rate": [],
"fallback_rate": [],
"latency_p50_ms": [],
"latency_p99_ms": [],
"token_cost_per_query": [],
"faithfulness_score": [],
}
コストとレイテンシの最適化
エージェンティックRAGシステムは、クエリごとに複数のLLM呼び出しを行うため、本質的にナイーブRAGよりもコストが高くレイテンシも大きくなります。品質を犠牲にせずにコストとレイテンシを最適化するには、慎重なアーキテクチャ上の判断が必要です。
ルーティングと評価にはより小さなモデルを使用します。ルーターとドキュメント評価器には、フロンティアモデルの完全な推論能力は不要です。GPT-4o-miniやClaude 3.5 Haikuは、コストとレイテンシのわずかな部分でこれらのタスクを処理できます：
# Use a small, fast model for routing and grading
routing_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
grading_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# Use a larger model only for final generation
generation_model = ChatOpenAI(model="gpt-4o", temperature=0.1)
検索結果を積極的にキャッシュします。多くのユーザークエリは、同じ基本的な質問のバリエーションです。類似度閾値内でクエリをマッチングするセマンティックキャッシュは、冗長な検索とLLM呼び出しを排除できます：
from langchain_community.cache import RedisSemanticCache
set_llm_cache(RedisSemanticCache(
redis_url="redis://localhost:6379",
embedding=embedding_model,
score_threshold=0.95,
))
各ステップに積極的なタイムアウトを設定して、制御不能なレイテンシを防止します。単一の遅いLLM呼び出しがリクエスト全体のタイムアウトを引き起こすべきではありません：
from asyncio import wait_for, TimeoutError
async def retrieve_with_timeout(state, timeout_seconds=5):
try:
return await wait_for(retrieve(state), timeout=timeout_seconds)
except TimeoutError:
state["documents"] = []
state["retrieval_grade"] = "no"
return state
実例のケーススタディとアンチパターン
エージェンティックRAGのデプロイメントで最も一般的な障害パターンは、無限リトライループです。検索試行のハードキャップと明確なフォールバック動作がなければ、エージェントは無限に再構成を繰り返し、トークンとレイテンシを浪費する可能性があります。常に明示的なリトライ制限を設定し、再構成率を測定してください。クエリの20%以上が再構成を必要とする場合、問題はエージェントのロジックではなく、検索レイヤー（不適切なチャンキング、誤ったエンベディングモデル、古いインデックス）にある可能性が高いです。
もう1つの頻繁なアンチパターンは過度なルーティングです。チームは時に、単一のよく設計されたハイブリッドリトリーバーの方がパフォーマンスが良いにもかかわらず、数十の専門化されたインデックスを持つ複雑なルーティンググラフを構築することがあります。要件を満たす最もシンプルなアーキテクチャから始め、よりシンプルなアプローチが不十分であることを示すメトリクスがある場合にのみ複雑さを追加してください。
コンテキストウィンドウの詰め込みは3番目のアンチパターンです。20のドキュメントチャンクを検索してすべてをプロンプトに詰め込むことはトークンを浪費し、LLMが関連性の低いコンテキストの海の中から関連情報を特定するのに苦労するため、実際には回答品質を低下させる可能性があります。少数の高品質なチャンク（通常3〜5個）へのリランキングは、評価ベンチマークでより大きなコンテキストウィンドウを一貫して上回ります。
最後に、評価駆動の過度な最適化に注意してください。RAGASスコアのみを最適化するチームは、ベンチマークでは良いスコアを出すが、ユーザーが役に立たないと感じる過度に慎重で曖昧な回答を生成するシステムを構築する可能性があります。定量的な評価と実際のユーザー会話の定性的なレビューのバランスを取ってください。
未来：マルチエージェントRAGシステム
エージェンティックRAGの次のフロンティアは、専門化されたエージェントが複雑な検索と合成タスクで協力するマルチエージェントシステムです。単一のエージェントが検索-生成パイプライン全体を処理する代わりに、異なる専門性を持つエージェントのチームをデプロイします。
リサーチエージェントがクエリ分解と大規模ドキュメントコレクション全体にわたるマルチホップ検索を処理します。ファクトチェックエージェントが、生成された主張をソースドキュメントに対して検証します。合成エージェントが複数のサブクエリからの発見を一貫性のある、よく構造化された回答にまとめます。エディターエージェントが最終出力の明瞭さ、正確さ、トーンをレビューします。
from langgraph.graph import StateGraph
# Multi-agent RAG architecture
multi_agent = StateGraph(MultiAgentState)
multi_agent.add_node("planner", planner_agent)
multi_agent.add_node("researcher", researcher_agent)
multi_agent.add_node("fact_checker", fact_check_agent)
multi_agent.add_node("synthesizer", synthesis_agent)
multi_agent.set_entry_point("planner")
multi_agent.add_edge("planner", "researcher")
multi_agent.add_edge("researcher", "fact_checker")
multi_agent.add_conditional_edges(
"fact_checker",
lambda s: "researcher" if s["needs_more_research"] else "synthesizer",
)
multi_agent.add_edge("synthesizer", END)
このアーキテクチャはクエリあたりのコストは高くなりますが、シングルエージェントシステムが苦手とする複雑なリサーチ質問を処理できます。設計上の主要な課題は、エージェント間の明確なインターフェースを定義し、エージェント間通信のオーバーヘッドが専門化のメリットを打ち消さないようにすることです。
ツールエコシステムは、マルチエージェントパターンをネイティブにサポートする方向に収束しています。LangGraphのサブグラフ合成、LlamaIndexのエージェントオーケストレーションレイヤー、CrewAIはすべて、マルチエージェントRAGシステムを構築するためのプリミティブを提供しています。これらのフレームワークが成熟しコストが低下し続けるにつれ、マルチエージェントRAGはより広範なプロダクションアプリケーションで実用的になるでしょう。このガイドで扱った原則 — 明示的な状態管理、検索品質評価、エレガントなフォールバック、包括的なオブザーバビリティ — は、シングルエージェントをデプロイする場合でも専門化されたエージェントのチームをデプロイする場合でも、引き続き不可欠です。