深夜2時15分の504 Gateway Timeout：従来のRAGが限界を迎えるとき
午前2時15分、ワーカークラスタから504 Gateway Timeoutが頻発し、PagerDutyのアラートが鳴り響きました。ログを追跡すると、いつもの原因が判明しました。マーケティング部門が競合20社の数値データ集約を求めるあまりに広範なプロンプトを投入していたのです。単一パイプラインのRAG構成が、膨大なWebページの生データを1回のLLM呼び出しに詰め込もうとした結果、コンテキストは128kトークンを超過し、並行スクレイピングスレッドも軒並みタイムアウトしていました。
本番環境の実績が示すように、基本的なRAGやZero-shot LLMでは本格的なリサーチ（Deep Research）の課題に対応できません。複雑な問いに対して、一般的なパイプラインはベクトル検索を1〜2回実行しただけで回答を生成してしまいます。その結果、データ不足からハルシネーション（幻覚）を起こすか、実用に耐えない浅薄なレポートを出力することになります。
Open Deep Researchモデルは、この課題を根本から解決します。LLMに即座の回答を強制するのではなく、プロセスを自律ループ（Autonomous Loop）へと分解します。
- 計画立案（Planning）：大きな問いをサブ課題に分解します（例：「AIチップの比較」というテーマを、アーキテクチャ、価格、推論性能の観点に分割）。
- 再帰的検索（Recursive Search）：検索エンジン経由でWebを照会し、内容を精読したうえで、データ不足を検知すると自動的に深掘り用クエリを生成します。
- 統合・ファクトチェック（Synthesis & Fact-checking）：複数ソースを相互検証し、ノイズを除去したうえで、参照元URL付きの標準Markdownレポートを出力します。
Pythonで自前のパイプラインを構築することで、実行フローを完全に制御できます。トークン消費のハードリミット設定やメモリリークの抑止が容易になり、高額なプロプライエタリAPIに依存する必要もなくなります。
実行環境のセットアップ
Ubuntu 22.04 / 24.04サーバーでは、システムパッケージとの競合を避けるため、専用の仮想環境に依存関係を隔離することを推奨します。
1. 仮想環境（Virtual Environment）の構築
# 作業ディレクトリの作成とPython 3.11+仮想環境のアクティベーション
mkdir -p /opt/open-deep-research && cd /opt/open-deep-research
python3 -m venv venv
source venv/bin/activate
# 基本ビルドツールのアップグレード
pip install --upgrade pip setuptools wheel2. 必要なライブラリのインストール
ここでは、APIキー不要でWeb検索を実行できるduckduckgo-search、不要なスクリプトタグや広告、ナビゲーションバーを除去してクリーンなテキストを抽出するtrafilatura、そしてOpenAI、OpenRouter、またはセルフホストのvLLM/Ollamaと互換性の高いopenai SDKを採用します。
pip install openai duckduckgo-search trafilatura pydantic tenacity richDeep Research Agentの実装：モジュール別の詳細
無限ループによるクォータ枯渇を防ぐため、ソースコードは堅牢なWebスクレイピングモジュール、ループ制御エージェント、テスト実行スクリプトの3つに明確に分離します。
1. 安全な検索＆テキスト抽出モジュール
tools.pyは検索とテキストクローリングを担当します。一時的なネットワーク障害やレートリミットを回避するため、Exponential Backoffによるリトライ処理を組み込んでいます。
import trafilatura
from duckduckgo_search import DDGS
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=6))
def search_web(query: str, max_results: int = 3) -> list[dict]:
"""レートリミット対策のリトライ機能を備えたDuckDuckGo Web検索"""
results = []
with DDGS() as ddgs:
raw_results = ddgs.text(query, max_results=max_results)
if raw_results:
for item in raw_results:
results.append({
"title": item.get("title", ""),
"url": item.get("href", ""),
"snippet": item.get("body", "")
})
return results
def fetch_clean_content(url: str, max_chars: int = 4000) -> str:
"""Webページを取得して本文テキストを抽出し、コンテキスト溢れを防ぐため文字数を制限する"""
try:
downloaded = trafilatura.fetch_url(url, timeout=10)
if not downloaded:
return ""
text = trafilatura.extract(downloaded, include_links=False, include_comments=False)
return text[:max_chars] if text else ""
except Exception as err:
return f"{url} の読み込みエラー: {err}"2. 再帰ループを制御するAgentモジュール
agent.pyはオーケストレーターとしての役割を果たします。ドキュメントを読み込むたびに、モデルは自ら「現在のデータで十分な数値と論拠が揃っているか？ 不足している場合、次に追加検索すべきキーワードは何か？」を自己評価します。
import os
import json
from openai import OpenAI
from tools import search_web, fetch_clean_content
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY", "sk-mock-key"))
class DeepResearchAgent:
def __init__(self, max_iterations: int = 3):
self.max_iterations = max_iterations
self.collected_evidence = []
def plan_next_step(self, topic: str, history: str) -> dict:
"""収集済みデータを評価し、次のクエリを決定する"""
prompt = f"""あなたは専門的なディープリサーチAIです。調査テーマ: "{topic}"
これまでに収集した情報:
{history}
分析を行い、以下のJSON形式に厳密に従って回答してください:
{{
"is_sufficient": true/false,
"reason": "情報が十分である理由、または不足しているデータの簡潔な説明",
"next_query": "情報不足の場合に実行する次の検索クエリ（十分な場合は空文字）"
}}"""
response = client.chat.completions.create(
model="gpt-4o-mini",
response_format={"type": "json_object"},
messages=[{"role": "user", "content": prompt}],
temperature=0.2
)
return json.loads(response.choices[0].message.content)
def run(self, topic: str) -> str:
print(f"[*] リサーチ開始: {topic}")
current_query = topic
for iteration in range(1, self.max_iterations + 1):
print(f"\n--- ループ {iteration}/{self.max_iterations}: '{current_query}' を検索中 ---")
search_hits = search_web(current_query, max_results=2)
for hit in search_hits:
url = hit["url"]
print(f" -> クロール中: {url}")
page_text = fetch_clean_content(url)
if page_text:
self.collected_evidence.append({
"source": url,
"query": current_query,
"snippet": page_text
})
# 直近の収集エビデンスの要約をAgentに提示して検討させる
evidence_summary = "\n".join([
f"- 情報源 [{e['source']}]: {e['snippet'][:250]}..."
for e in self.collected_evidence[-4:]
])
decision = self.plan_next_step(topic, evidence_summary)
if decision.get("is_sufficient") or iteration == self.max_iterations:
print(f"[*] 終了条件に到達: {decision.get('reason')}")
break
current_query = decision.get("next_query", topic)
return self.generate_report(topic)
def generate_report(self, topic: str) -> str:
"""収集したすべてのエビデンスを統合してテクニカルレポートを生成する"""
print("\n[*] 最終レポートを作成中...")
raw_data = "\n\n".join([
f"### 情報源: {e['source']}\n{e['snippet']}"
for e in self.collected_evidence
])
system_prompt = "あなたはテクニカルアナリストです。提供された引用情報のみに基づき、Markdown形式で詳細な統合レポートを作成してください。各論点には必ず対応する参照元URLを付記してください。"
user_prompt = f"テーマ: {topic}\n\n調査データ:\n{raw_data}"
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.3
)
return response.choices[0].message.content3. テスト実行用スクリプト
main.pyは入力テーマを受け取り、ワーカーを起動してMarkdownファイルに結果を出力します：
import os
from agent import DeepResearchAgent
def main():
if not os.getenv("OPENAI_API_KEY"):
raise ValueError("環境変数 OPENAI_API_KEY が設定されていません！")
topic = "AIエージェント向け外部メモリのアーキテクチャ：レイテンシとトークンコストにおけるMemGPTと従来型Vector DBの比較評価"
agent = DeepResearchAgent(max_iterations=3)
report = agent.run(topic)
output_file = "deep_research_report.md"
with open(output_file, "w", encoding="utf-8") as f:
f.write(report)
print(f"\n[✓] 完了！レポートを保存しました: {output_file}")
if __name__ == "__main__":
main()本番環境での運用：コストとサーバーリソースのコントロール
ローカル環境でスムーズに動作しても、本番環境で安定稼働するとは限りません。毎日数百件のタスクをワーカーに処理させる場合、以下の3つの安全策を講じる必要があります：
1. 実際の実行フローの検証
スクリプトを実行し、ターミナルで詳細ログを確認します：
export OPENAI_API_KEY="sk-your-actual-api-key"
python3 main.pyターミナルには、Agentの思考プロセスがステップごとに表示されます。たとえば、反復1でMemGPTの定義を読み取った後、反復2でレイテンシのベンチマーク表が不足していると認識し、自動的にMemGPT vs Pinecone latency benchmarkというクエリを生成して追加検索を行い、最終的にdeep_research_report.mdを出力します。
2. 3つの障害点（Failure Points）とその対策
- 反復回数の上限設定（Iteration Cap）：必ず
max_iterations = 3（最大でも4）を設定してください。1回のループで約3,000〜5,000トークンを消費します。曖昧なテーマに対してエージェントの自律判断に任せきりにすると、15〜20回もループしてしまい、わずか1時間で10〜15ドルのAPI費用を浪費するリスクがあります。 - テキストバッファのサイズ制限：URLごとに
max_chars = 4000（約1,000トークン）の上限を設けます。一部のニュースサイトには大量のスクリプトが埋め込まれており、Wikipediaのページは2万語を超えることもあります。そうしたページを制限なしで3つクロールするだけで、最初の反復で128kのコンテキストウィンドウが溢れてしまいます。 - Systemdによるリソース制限：
nohupやtmuxによるバックグラウンド手動実行は避けてください。同一VPS上で稼働するデータベースのクラッシュを防ぐため、メモリ上限を設定したsystemdサービスとして構成します：
# /etc/systemd/system/deep-research-worker.service
[Unit]
Description=Deep Research Agent Worker
After=network.target
[Service]
Type=simple
User=ubuntu
WorkingDirectory=/opt/open-deep-research
ExecStart=/opt/open-deep-research/venv/bin/python main.py
Restart=on-failure
RestartSec=10s
MemoryMax=1.5G
[Install]
WantedBy=multi-user.targetサービスを有効化した後は、systemctl status deep-research-workerでプロセスの稼働状況を確認し、journalctl -u deep-research-worker -fでリアルタイムログを監視します。これにより、検索エンジンによるBotブロックやCAPTCHAの発生を早期に検知できます。
