RAGとマルチエージェントにおける高度なループ設計
AIエージェントの適用領域が広がるにつれ、単一のエージェントがツールを呼び出す単純なループだけでは対応できない複雑なユースケースが増えています。特に、不正確な検索結果を修正しながら回答を生成する RAG（検索拡張生成）の進化系ループ や、複数の役割を持つAI同士が交渉・協力する マルチエージェントループ は、現在の実用的なAIシステム設計の最前線です。
本章では、Corrective RAG や Self-RAG に代表される「検索・評価・生成の検証ループ」、複数エージェント間の「合意形成（コンセンサス）ループ」、セキュアな「サンドボックス実行ループ」、そして「タスクの再帰的分解と並列化」について、具体的なコードと Mermaid 図を交えて技術的に解説します。
5.1 検索と生成の検証・修正ループ（Corrective RAG, Self-RAG）
従来の単純な RAG（Naive RAG）は、「ユーザーの質問 ➔ ベクトル検索 ➔ 検索されたドキュメントをLLMに入力して回答生成」という一本道のプロセスでした。しかし、このアプローチには「検索結果にゴミ（ノイズ）が含まれているとハルシネーションを起こす」「検索結果に必要な情報が足りない場合に嘘をつく」という大きな弱点があります。
これを解決するのが、検索ドキュメントの品質判定をループ内に組み込んだ Corrective RAG (CRAG) や Self-RAG の設計です。
Self-RAG / CRAG ループの主要なステップ
- 評価 (Grading): 検索で得られた各ドキュメントが、ユーザーの質問に対して本当に「関連しているか（Relevant/Irrelevant）」をLLMで二値判定します。
- 外部補強 (Web Search / Fallback): 関連ドキュメントがゼロ、またはスコアが閾値未満の場合は、元の知識ベース（ベクトルDB）だけでは回答不能と判断し、Web検索API等を呼び出すか、またはクエリをリライトして再検索します。
- 自己検証 (Self-Reflection): 生成された回答が、参照したドキュメントの内容と矛盾していないか（Groundedness）、および質問への直接的な回答になっているか（Answer Relevance）をLLM自身に判定させ、NGであれば検索・生成のループをやり直します。
PythonによるSelf-RAGループの実装例
以下は、検索結果の品質判定とクエリリライト（再構成）を組み込んだRAGリトライループの実装コードです。
from typing import List, Dict, Any, Literal
# =====================================================================
# モック用のデータ・API定義
# =====================================================================
VECTOR_DATABASE = {
"AGYエージェントの最大ループ数": "AGYエージェントの規定設定は存在しません。（※実際は第2章で解説した通り、プログラム側で設定します）",
"LangGraphのrecursion_limit": "LangGraphにおけるデフォルトのrecursion_limitは25です。"
}
def mock_vector_search(query: str) -> List[str]:
"""擬似的なベクトル検索"""
results = []
for key, val in VECTOR_DATABASE.items():
if any(word in query for word in ["recursion_limit", "最大", "制限"]):
results.append(val)
return results
def mock_web_search(query: str) -> str:
"""検索DBに情報がない場合のWeb検索フォールバック"""
print(f" [CRAG] Web検索を実行中: '{query}'")
if "recursion_limit" in query:
return "最新のLangGraph仕様では、recursion_limitはコンパイル時または実行時に変更可能です。"
return "関連するWebページが見つかりませんでした。"
# =====================================================================
# LLM（監査・生成）のモック
# =====================================================================
class SelfRAGLLM:
def __init__(self):
self.rewrite_count = 0
def grade_document(self, query: str, document: str) -> Literal["relevant", "irrelevant"]:
"""ドキュメントがユーザーのクエリに関連しているかを判定する"""
if "recursion_limit" in query and "recursion_limit" in document:
return "relevant"
return "irrelevant"
def rewrite_query(self, query: str) -> str:
"""より適切な検索結果を得るためにクエリをリライトする"""
self.rewrite_count += 1
print(f" [LLM] クエリをリライトします (回数: {self.rewrite_count})")
return "LangGraph recursion_limit 仕様 変更方法"
def generate_answer(self, query: str, documents: List[str]) -> str:
"""ドキュメントを元に回答を生成する"""
context = "\n".join(documents)
return f"提供されたドキュメントに基づくと、LangGraphのデフォルトのrecursion_limitは25ですが、これは変更可能です。({context})"
# =====================================================================
# Self-RAG 制御パイプライン
# =====================================================================
def run_self_rag(query: str, max_attempts: int = 3) -> str:
llm = SelfRAGLLM()
current_query = query
retrieved_docs = []
for attempt in range(1, max_attempts + 1):
print(f"\n--- 試行 {attempt}/{max_attempts} ---")
# 1. 検索実行
raw_docs = mock_vector_search(current_query)
print(f" [Search] {len(raw_docs)} 件のドキュメントを検索しました。")
# 2. ドキュメントの評価 (Grading)
relevant_docs = []
for doc in raw_docs:
grade = llm.grade_document(current_query, doc)
if grade == "relevant":
relevant_docs.append(doc)
print(f" [Grading] 適合ドキュメント数: {len(relevant_docs)}/{len(raw_docs)}")
# 3. 判定とリライト判定
if not relevant_docs:
print(" [CRAG] 適合するドキュメントがないため、Web検索で補強します。")
web_result = mock_web_search(current_query)
# Web検索結果を再評価し、適合する場合のみ追加する
if llm.grade_document(current_query, web_result) == "relevant":
relevant_docs.append(web_result)
# 4. 生成と自己検証（簡易化のため、適合ドキュメントがある場合は回答生成へ進む）
if relevant_docs:
answer = llm.generate_answer(query, relevant_docs)
print(" [Generate] 回答を生成しました。")
return answer
# 適合データが得られない場合はクエリをリライトして再検索
current_query = llm.rewrite_query(current_query)
return "エラー: 指定回数内に十分な情報を見つけられず、回答を生成できませんでした。"
if __name__ == "__main__":
# 初期クエリ（少し曖昧な検索キーワード）
result = run_self_rag("LangGraphのループ上限を変更するには？")
print("\n【最終生成回答】\n", result)
5.2 複数エージェント間の協調・交渉・コンセンサスループ
単一のLLMにすべての役割（プログラミング、コードレビュー、テスト、ドキュメント作成）を押し付けると、コンテキストが極めて複雑になり、出力精度が著しく低下します。
これを解決するため、「特定タスクに特化した複数のエージェントを自律的に対話させる（マルチエージェント）」 設計パターンが有効です。
特に、「提案者」と「批評者」のディスカッション（対話・コンセンサス）ループを構築することで、成果物のクオリティを自律的に極限まで高めることができます。
マルチエージェントによる議論・合意のシーケンス
以下は、「Writer（ライター）」 が書いたブログ記事案を、「Editor（編集者）」 がレビューし、編集者の承認（GOサイン）が出るまで原稿修正を繰り返す協調ループのシーケンスです。
Pythonによるマルチエージェントコンセンサスループの実装
以下は、2つのエージェントのクラスを定義し、お互いのメッセージを送り合って合意に達する対話ループの実装例です。
class WriterAgent:
def __init__(self):
self.draft_version = 0
def generate_draft(self, topic: str, feedback: str = None) -> str:
self.draft_version += 1
if not feedback:
return f"【草稿 v{self.draft_version}】テーマ: '{topic}'。AIエージェントは自律的に動くシステムです。終わり。"
else:
return (
f"【草稿 v{self.draft_version}】テーマ: '{topic}'。\n"
f"AIエージェントは、LLMを脳として動き、ツールを利用し、自律的に思考ループを回すシステムです。\n"
f"例えば、第2章で紹介したReActパターンが代表例です。(編集フィードバック: '{feedback}' を反映)"
)
class EditorAgent:
def evaluate_draft(self, draft: str) -> Dict[str, Any]:
"""ドラフトを評価し、修正が必要か、または承認できるかを判定する"""
print(f"\n[Editor] 草稿をレビュー中...")
# 簡易的な評価基準（具体例やReActというキーワードがあるか）
if "ReAct" in draft or "例えば" in draft:
return {
"decision": "approve",
"feedback": "具体例が追加され、エンジニアにとって分かりやすい内容になりました。承認します。"
}
else:
return {
"decision": "reject",
"feedback": "説明が抽象的すぎます。具体的な設計パターン（例: ReActなど）への言及や具体例を追加してください。"
}
# =====================================================================
# 対話ループの実行
# =====================================================================
def run_consensus_loop(topic: str, max_turns: int = 4):
writer = WriterAgent()
editor = EditorAgent()
current_feedback = None
print(f"=== マルチエージェント会議開始: テーマ '{topic}' ===")
for turn in range(1, max_turns + 1):
print(f"\n--- ターン {turn} ---")
# ライターが執筆（フィードバックがあればそれを反映）
draft = writer.generate_draft(topic, current_feedback)
print(f"[Writer] 成果物を提出しました:\n{draft}")
# 編集者が査読
evaluation = editor.evaluate_draft(draft)
print(f"[Editor] 判定: {evaluation['decision'].upper()}")
print(f"[Editor] フィードバック: {evaluation['feedback']}")
if evaluation["decision"] == "approve":
print("\n=== 合意成立: 成果物が承認されました ===")
return draft
# 却下された場合は、フィードバックを次のターンのライター入力に渡す
current_feedback = evaluation["feedback"]
print("\n=== 警告: 最大対話ターン数に達しましたが、合意に至りませんでした ===")
return draft
if __name__ == "__main__":
final_article = run_consensus_loop("自律型エージェントの基本設計")
5.3 ツール利用（Tool Use）とサンドボックス実行のセキュアループ
AIエージェントに「コード実行（Python REPLなど）」や「シェルコマンド実行」のツールを渡すと、エージェントは自らプログラムを書いてバグを自己修正しながらタスクを処理できるようになり、問題解決能力が劇的に向上します。しかし、これは同時に**極めて高いセキュリティリスク（悪意あるコードの実行、無限ループによるホストリソースの枯渇、システムファイルの改ざん・漏洩）**を生み出します。
したがって、エージェントがコードを実行するループは、ホスト環境から完全に隔離された 「セキュアサンドボックス（Secure Sandbox）」 内に限定しなければなりません。
[ エージェント (ホスト環境) ]
│
▼ (コード生成: "rm -rf /" または無限ループコード等)
┌───────────────┐
│ ゲートウエイ │ (タイムアウト制限・リソース制限・通信制限)
└───────┬───────┘
▼ (コンテナやWasmへのコード流し込み)
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃ セキュアサンドボックス ┃ (Dockerコンテナ / gVisor / Wasm / MicroVM)
┃ ┃
┃ [実行環境] コード実行 ┃ ➔ ファイルシステム等は使い捨て
┗━━━━━━━━━━━━━┳━━━━━━━━━━━━━━┛
▼ (標準出力・標準エラーを回収)
[ エージェント (ホスト環境) ] ➔ エラーメッセージを見てコードを自動修復
セキュアサンドボックスループの設計原則
-
使い捨ての環境 (Ephemeral Environment):
- 実行ごとに新しくコンテナやマイクロVM（Firecracker等）を立ち上げ、実行完了後に即座に破棄（クリーンアップ）します。
-
ネットワークの遮断 (Network Isolation):
- 特段の理由がない限り、サンドボックス内のインターネット接続を遮断し、エージェントがコード実行を通じて外部にデータを送信（C2サーバーへのデータ漏洩など）するのを防ぎます。
-
ハードリソース制限 (Resource Constraints):
- CPU使用率、メモリ上限、ディスク書き込み量をcgroupsなどで厳格に制限し、AIが生成した無限ループコードによってサーバーがダウンするのを防ぎます。
-
自己修復リトライループの融合:
- サンドボックスの
stderr(標準エラー出力) とexit_codeをエージェントに返し、「Pythonエラーが発生しました。修正して再実行してください」というループ（第2章で解説した自己修復）を組み合わせることで、安全かつ自律的なコーディング能力を実現します。
- サンドボックスの
5.4 タスクの再帰的分解（Recursive Decomposition）と並列ループ処理
1つの巨大なタスクを線形な（一直線の）ループで処理するのには限界があります。
複雑な開発プロジェクトや大規模な調査タスクを処理する場合、エージェントはタスクを 再帰的に分解（Decomposition） して複数の子タスクを作り、それらを 並列で処理（Map-Reduce） し、最後に結果を結合（Join）するツリー型のループ処理を行います。
タスクの再帰的分解と並列実行の構造
[ 親タスク: 技術レポートの作成 ]
│
┌──────────────────┼──────────────────┐
▼ (分解) ▼ (分解) ▼ (分解)
[ 子タスク 1 ] [ 子タスク 2 ] [ 子タスク 3 ]
(フロントエンド) (バックエンド) (インフラ)
│ │ │
(並列実行) (並列実行) (並列実行)
▼ ▼ ▼
[ 実行結果 1 ] [ 実行結果 2 ] [ 実行結果 3 ]
└──────────────────┬──────────────────┘
▼ (結合・集約)
[ 最終レポートの統合と検証 ]
並列ループ処理（Map-Reduce）の設計パターン
- Decomposer（分解器）: LLMが大目標（例: 「3つの競合他社のWebサイトの料金プランを比較する」）を受け取り、それを独立して実行可能な子タスク（「A社の料金調査」「B社の料金調査」「C社の料金調査」）のリストに分解します。
-
Parallel Dispatcher（並列ディスパッチャー）: 分解された子タスクの配列に対して、非同期処理（Pythonの
asyncio.gatherや LangGraphのSendオブジェクト）を利用して、それぞれのエージェントインスタンスを同時に起動します。 - Reducer / Synthesizer（集約器）: すべての並列処理ノードの完了を待ち受け、集まった結果を取りまとめて1つの最終レポートを作成します。この際、足りない情報や矛盾があれば、特定の競合他社に対してのみピンポイントで「再調査（リトライ）ループ」を発生させます。
マルチエージェントと並列分散処理を組み合わせることで、単一エージェントでは数十分かかる処理を、安全かつ極めて短時間で完遂することが可能になります。