Gemini Deep Research Agent は、複数ステップの調査を自律的に遂行できるエージェントです。競合分析、市場リサーチ、学術調査など、深い思考が必要なタスクを自動化します。
Deep Research Agent とは
Deep Research Agent は Gemini 3.1 Pro をベースとしたエージェントで、以下のような多段階プロセスを自動で遂行します:
- 計画立案 — どのような情報が必要かを判断
- 複数キーワード検索 — 多角的に情報を収集
- 情報統合 — 複数ソースを組み合わせて理解
- 分析と推論 — 根拠に基づいた結論を導出
- レポート生成 — 読みやすい形式で結果をまとめる
従来の「キーワード1つ→結果リスト」という検索では得られない、複合的で洞察に満ちた結果が得られます。
Interactions API の役割
多くの開発者が Gemini API と聞くと generate_content エンドポイントを思い浮かべますが、Deep Research Agent は Interactions API という別のAPI体系を使います。この違いは重要です。
- generate_content — リアルタイム応答を返す同期型
- Interactions API — 非同期で長時間かかるタスクを処理(ポーリング方式)
Deep Research では調査時間が数分~1時間かかることもあります。Interactions API を使うと、リクエストを投げて、その後ステータスを確認する「ファイア・アンド・フォーゲット」型の処理が可能になります。
セットアップと基本フロー
Google AI Studio から簡単に API キーを取得できます。Google AI Studio にアクセス、左サイドバーの「API keys」をクリック、「Create API key」をクリック、生成されたキーを環境変数に設定します。
pip install google-genai
export GOOGLE_API_KEY="YOUR_API_KEY"Deep Research Agent の使用フロー
ステップ1-2: リサーチタスク作成と確認
リクエストを Interactions API に送信します。background=true パラメータが重要で、これにより非同期処理モードになります。
import google.generativeai as genai
import time
genai.configure(api_key="YOUR_API_KEY")
def create_research_task(query: str, audience: str = "general") -> str:
"""リサーチタスクを作成してインタラクションIDを返す"""
client = genai.Client()
interaction = client.aistudio.create_interaction(
model="gemini-3-1-pro",
request={
"system_instruction": f"Target audience: {audience}",
"contents": {
"role": "user",
"parts": [{"text": f"Research: {query}"}]
}
},
background=True # 非同期モード
)
return interaction.nameステータス確認はポーリングで定期的に行います。
def poll_research_status(interaction_id: str) -> dict:
"""リサーチのステータスをポーリング"""
client = genai.Client()
interaction = client.aistudio.get_interaction(name=interaction_id)
if interaction.state == "PROCESSING":
return {"status": "processing"}
elif interaction.state == "COMPLETED":
return {"status": "completed", "result": interaction.response.text}
elif interaction.state == "FAILED":
return {"status": "failed", "error": interaction.error_message}
return {"status": "unknown"}
def wait_for_completion(interaction_id: str, max_wait: int = 3600) -> dict:
"""リサーチ完了を待機(最大60分)"""
elapsed = 0
while elapsed < max_wait:
status = poll_research_status(interaction_id)
if status["status"] == "completed":
return {"success": True, "result": status["result"]}
elif status["status"] == "failed":
return {"success": False, "error": status["error"]}
time.sleep(10)
elapsed += 10
return {"success": False, "error": "Timeout"}実装例:競合分析レポート自動生成
import json
from datetime import datetime
def competitive_analysis(company_name: str) -> dict:
"""競合企業の自動分析"""
query = f"""
Analyze {company_name}:
1. Business model and revenue streams
2. Main products and target markets
3. Market position and competitors
4. Financial performance
5. R&D focus and innovation
Provide citations for all claims.
"""
interaction_id = create_research_task(
query,
audience="business analysts"
)
result = wait_for_completion(interaction_id)
if result["success"]:
return {
"company": company_name,
"timestamp": datetime.now().isoformat(),
"content": result["result"]
}
else:
return {"error": result["error"]}
# 使用例
analysis = competitive_analysis("OpenAI")
print(json.dumps(analysis, indent=2, ensure_ascii=False))プロンプトベストプラクティス
1. ターゲットオーディエンスの明示
query = """
Impact of AI on healthcare
Target audience: Hospital administrators
(Use practical language, focus on ROI and implementation)
"""2. 判断基準の明確化
query = """
Best cloud storage alternatives to Google Drive
Decision criteria:
- Security compliance (GDPR, SOC 2)
- Pricing transparency
- API documentation
"""3. ソースタイプの指定
query = """
EU cryptocurrency regulations
Prefer: Official government sources, published research
Avoid: Social media, opinion blogs
"""Deep Research と他のリサーチ手段の比較
手動調査
- 速度: 遅い
- 深さ: 高
- コスト: 高い時間コスト
- 用途: 限定的な質問
従来の検索
- 速度: 速い
- 深さ: 中程度
- コスト: 低
- 用途: 簡単な事実確認
Deep Research Agent
- 速度: 中程度
- 深さ: 高
- コスト: 低(API従量課金)
- 用途: 複合的な分析
リサーチアナリスト
- 速度: 遅い
- 深さ: 最高
- コスト: 非常に高い
- 用途: 重要な意思決定
Deep Research Agent は「手動調査ほどではないが、検索より深い分析が必要」というユースケースで最も価値があります。
本番運用とまとめ
リトライ実装ではexponentialバックオフを使います:
def research_with_retry(query: str, max_retries: int = 2) -> dict:
for attempt in range(max_retries):
try:
interaction_id = create_research_task(query)
result = wait_for_completion(interaction_id)
if result["success"]:
return result
except Exception as e:
print(f"Attempt {attempt + 1} failed: {e}")
return {"success": False, "error": "Max retries exceeded"}キャッシング戦略では24時間TTLで重複リサーチを防ぎます:
import hashlib
cache = {}
def cached_research(query: str, cache_ttl_hours: int = 24) -> dict:
query_hash = hashlib.md5(query.encode()).hexdigest()
if query_hash in cache:
cached_result, timestamp = cache[query_hash]
from datetime import datetime, timedelta
if datetime.now() - timestamp < timedelta(hours=cache_ttl_hours):
return cached_result
result = competitive_analysis(query)
cache[query_hash] = (result, datetime.now())
return resultGemini Deep Research Agent は、Web検索から分析・レポート生成まで複雑なリサーチタスクを自動化できる強力なツールです。Interactions APIの非同期特性を活かし、キューイングやキャッシュを組み合わせることで、スケーラブルなリサーチシステムが構築できます。市場分析、競合調査、トレンド研究など、ビジネスに必要な情報活動の効率化に、ぜひ活用してみてください。