エージェントを一度動かすところまでは、実のところそれほど難しくありません。難しいのは、ツール呼び出しが失敗したとき・状態が途中で壊れたとき・想定外の順序で処理が走ったときに、それでも止まらないようにすることです。
Gemini 2.5 Pro でマルチステップの自動化を組む前提で、ツール呼び出し・状態管理・オーケストレーションの設計を順に積み上げます。デモを動かすための構成ではなく、動かし続けるための構成です。
エージェントシステム設計の基本原則
リアクティブ vs プランニングエージェント
リアクティブエージェントは、各ステップで環境の状態を観察し即座に行動します。シンプルで予測可能ですが、複雑なタスクには向きません。
プランニングエージェントは、全体計画を立てた上で段階的に実行します。Gemini 2.5 Proの拡張推論能力により、より複雑な問題解決が可能です。
大規模プロジェクト管理やリサーチタスクなど、複数の相互依存するステップが必要な場合はプランニングエージェントを推奨します。
ツール定義スキーマの設計
Gemini APIでのツール定義には、JSONスキーマの厳密な型付けが必須です：
import json
import google.generativeai as genai
# ツール定義の例
tools = [
{
"name": "search_web",
"description": "ウェブから関連情報を検索して返す",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "検索クエリ"
},
"num_results": {
"type": "integer",
"description": "返す結果の数（デフォルト5）",
"default": 5
}
},
"required": ["query"]
}
},
{
"name": "fetch_url",
"description": "指定されたURLからコンテンツを取得する",
"parameters": {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "取得するURL"
}
},
"required": ["url"]
}
}
]
model = genai.GenerativeModel(
"gemini-2.5-pro",
tools=tools
)ツール説明は明確かつ簡潔に。不明確な説明はモデルの誤った呼び出しを招きます。
並列ツール呼び出しの実装
Gemini 2.5 Proは複数ツールの同時呼び出しをサポートします：
def execute_parallel_tools(user_query):
"""並列ツール呼び出しの実行例"""
messages = [
{"role": "user", "content": user_query}
]
# 最初の推論ステップ
response = model.generate_content(messages)
# ツール呼び出しの解析
tool_calls = []
for part in response.content.parts:
if part.function_call:
tool_calls.append(part.function_call)
print(f"並列ツール呼び出し: {len(tool_calls)}個")
# ツール実行結果の集約
tool_results = []
for call in tool_calls:
result = execute_tool(call.name, call.args)
tool_results.append({
"tool_name": call.name,
"result": result
})
# モデルに結果を送信
messages.append({"role": "model", "content": response.content})
messages.append({
"role": "user",
"content": [
{
"function_response": {
"name": r["tool_name"],
"response": r["result"]
}
}
for r in tool_results
]
})
# 最終応答の生成
final_response = model.generate_content(messages)
return final_response.text
def execute_tool(tool_name, arguments):
"""ツール実行のシミュレーション"""
if tool_name == "search_web":
return f"検索結果: {arguments['query']}に関する情報"
elif tool_name == "fetch_url":
return f"URL {arguments['url']} からのコンテンツ取得済"
return "不明なツール"マルチステップ推論ループの実装
エージェントのコア部分はツール呼び出しと結果統合のループです：
class ResearchAgent:
def __init__(self, model):
self.model = model
self.max_iterations = 10
self.conversation_history = []
def run(self, task):
"""マルチステップ推論ループ"""
self.conversation_history = [
{"role": "user", "content": task}
]
for iteration in range(self.max_iterations):
print(f"\n=== イテレーション {iteration + 1} ===")
# モデルから次のアクションを取得
response = self.model.generate_content(
self.conversation_history
)
# 終了判定
if self._should_stop(response):
print(f"タスク完了: {response.text}")
return response.text
# ツール呼び出しの処理
self.conversation_history.append({
"role": "model",
"content": response.content
})
tool_calls = [
part.function_call
for part in response.content.parts
if part.function_call
]
if not tool_calls:
# ツール呼び出しなし = 推論完了
return response.text
# ツール実行と結果追加
results = self._execute_tools(tool_calls)
self.conversation_history.append({
"role": "user",
"content": results
})
return "最大イテレーション数に達しました"
def _should_stop(self, response):
"""終了条件のチェック"""
return not any(
part.function_call for part in response.content.parts
)
def _execute_tools(self, tool_calls):
"""複数のツール呼び出しを実行"""
results = []
for call in tool_calls:
result = execute_tool(call.name, call.args)
results.append({
"function_response": {
"name": call.name,
"response": result
}
})
return resultsエージェント状態管理
本番環境では、エージェントの状態を永続化する必要があります：
from datetime import datetime
import json
class AgentStateManager:
def __init__(self, storage_backend="memory"):
self.backend = storage_backend
self.sessions = {}
def create_session(self, user_id, task):
"""新しいセッションを作成"""
session_id = f"{user_id}_{datetime.now().isoformat()}"
session_state = {
"session_id": session_id,
"user_id": user_id,
"task": task,
"created_at": datetime.now().isoformat(),
"conversation_history": [],
"tool_call_log": [],
"status": "running"
}
self.sessions[session_id] = session_state
self._persist_state(session_id, session_state)
return session_id
def update_conversation(self, session_id, role, content):
"""会話履歴を更新"""
if session_id not in self.sessions:
raise ValueError(f"セッション {session_id} が見つかりません")
self.sessions[session_id]["conversation_history"].append({
"role": role,
"content": content,
"timestamp": datetime.now().isoformat()
})
self._persist_state(session_id, self.sessions[session_id])
def log_tool_call(self, session_id, tool_name, args, result):
"""ツール呼び出しを記録"""
self.sessions[session_id]["tool_call_log"].append({
"tool_name": tool_name,
"arguments": args,
"result": result,
"timestamp": datetime.now().isoformat()
})
self._persist_state(session_id, self.sessions[session_id])
def _persist_state(self, session_id, state):
"""状態を永続化（バックエンド別実装）"""
if self.backend == "memory":
self.sessions[session_id] = state
elif self.backend == "firestore":
# Firestoreへの保存実装
pass
elif self.backend == "redis":
# Redisへの保存実装
passエラーリカバリーパターン
本番環境では堅牢なエラーハンドリングが必須です：
import time
from typing import Optional
class ResilientAgent:
def __init__(self, model, max_retries=3):
self.model = model
self.max_retries = max_retries
def call_tool_with_retry(self, tool_name, arguments):
"""指数バックオフによるリトライ付きツール呼び出し"""
for attempt in range(self.max_retries):
try:
result = execute_tool(tool_name, arguments)
return result
except Exception as e:
if attempt < self.max_retries - 1:
# 指数バックオフ: 2^attempt 秒待機
wait_time = 2 ** attempt
print(f"リトライ {attempt + 1}/{self.max_retries} - {wait_time}秒待機")
time.sleep(wait_time)
else:
# 最後のリトライで失敗 → グレースフルデグラデーション
return self._graceful_fallback(tool_name, arguments, e)
def _graceful_fallback(self, tool_name, arguments, error):
"""ツール失敗時のフォールバック処理"""
print(f"ツール {tool_name} 失敗: {str(error)}")
if tool_name == "search_web":
return "キャッシュされた検索結果（オフライン）"
elif tool_name == "fetch_url":
return "ドキュメント取得失敗 - 別の情報源を試してください"
return f"{tool_name} は現在利用できません"人間参加型ループ（HITL）の実装
重要な判断には人間承認を挿入：
class HumanInTheLoopAgent:
def __init__(self, model, approval_handler):
self.model = model
self.approval_handler = approval_handler
def run_with_approval(self, task, approval_required_actions=None):
"""人間承認が必要なアクションを含むエージェント実行"""
if approval_required_actions is None:
approval_required_actions = ["delete", "modify_financial", "send_email"]
messages = [{"role": "user", "content": task}]
while True:
response = self.model.generate_content(messages)
# 終了判定
if not any(part.function_call for part in response.content.parts):
return response.text
messages.append({"role": "model", "content": response.content})
# ツール呼び出しのフィルタリングと承認
tool_calls = [
part.function_call for part in response.content.parts
if part.function_call
]
approval_needed = [
call for call in tool_calls
if any(action in call.name for action in approval_required_actions)
]
if approval_needed:
print(f"\n承認が必要なアクション:\n")
for call in approval_needed:
print(f" - {call.name}: {call.args}")
# 人間の判断を待機
if not self.approval_handler.get_approval():
return "ユーザーがアクションを拒否しました"
# ツール実行
results = []
for call in tool_calls:
result = execute_tool(call.name, call.args)
results.append({
"function_response": {
"name": call.name,
"response": result
}
})
messages.append({"role": "user", "content": results})リサーチエージェント実装例
実際のユースケースとしてリサーチエージェントを実装：
class ResearchReportAgent:
def __init__(self, model):
self.model = model
self.report_sections = {}
def generate_report(self, topic, num_sources=5):
"""トピックに関するリサーチレポートを生成"""
task = f"""
次のトピックについて詳細なリサーチレポートを作成してください:
'{topic}'
以下の構成で5つ以上の信頼できるソースから情報を収集してください:
1. 概要
2. 主な発見
3. 業界への影響
4. 今後の展望
5. 推奨事項
各セクションで具体的な引用とソースを含めてください。
"""
research_agent = ResearchAgent(self.model)
final_report = research_agent.run(task)
return final_reportエージェントの評価とテスト
本番前の品質保証は重要です：
import json
class AgentEvaluator:
def __init__(self):
self.test_cases = []
self.results = []
def add_test_case(self, task, expected_behaviors):
"""テストケースを追加"""
self.test_cases.append({
"task": task,
"expected_behaviors": expected_behaviors
})
def evaluate_agent(self, agent, test_case):
"""エージェントのテスト実行と評価"""
result = agent.run(test_case["task"])
evaluation = {
"task": test_case["task"],
"result": result,
"metrics": {
"tool_calls_made": len(agent.conversation_history),
"total_iterations": agent.iteration_count,
"success": self._check_success(result, test_case["expected_behaviors"]),
"execution_time": agent.execution_time
}
}
self.results.append(evaluation)
return evaluation
def _check_success(self, result, expected_behaviors):
"""成功判定ロジック"""
for behavior in expected_behaviors:
if behavior not in result:
return False
return True
def generate_report(self):
"""評価レポートを生成"""
success_rate = sum(
1 for r in self.results if r["metrics"]["success"]
) / len(self.results)
return {
"total_tests": len(self.results),
"success_rate": success_rate,
"average_iterations": sum(
r["metrics"]["total_iterations"] for r in self.results
) / len(self.results),
"detailed_results": self.results
}コストと遅延の最適化
class OptimizedAgentPipeline:
def __init__(self):
self.model_routing = {
"complex_reasoning": "gemini-2.5-pro",
"simple_lookup": "gemini-1.5-flash",
"text_generation": "gemini-1.5-flash"
}
def route_task(self, task_type, content):
"""タスク種別に応じた最適なモデルを選択"""
model_name = self.model_routing.get(task_type, "gemini-2.5-pro")
return genai.GenerativeModel(model_name)
def batch_tool_results(self, tool_calls):
"""複数ツール結果をバッチ化"""
batched = {}
for call in tool_calls:
if call.name not in batched:
batched[call.name] = []
batched[call.name].append(call.args)
return batched本番環境での考慮事項
レート制限とタイムアウト
from ratelimit import limits, sleep_and_retry
class ProductionAgent:
@sleep_and_retry
@limits(calls=100, period=60) # 1分間に100回まで
def call_api(self, request):
return self.model.generate_content(request)
def run_with_timeout(self, task, timeout_seconds=300):
"""タイムアウト付きエージェント実行"""
import signal
def timeout_handler(signum, frame):
raise TimeoutError("エージェント実行がタイムアウト")
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(timeout_seconds)
try:
result = self.run(task)
signal.alarm(0) # タイマーをキャンセル
return result
except TimeoutError:
return "実行時間がタイムアウト制限を超過"可観測性とログ
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ObservableAgent:
def run_with_logging(self, task):
"""完全なログ記録付きエージェント実行"""
agent_id = datetime.now().isoformat()
logger.info(f"エージェント {agent_id} 開始: {task}")
try:
result = self.agent.run(task)
logger.info(f"エージェント {agent_id} 成功")
return result
except Exception as e:
logger.error(f"エージェント {agent_id} 失敗: {str(e)}")
raiseエージェント設計チェックリスト
本番デプロイ前の確認項目：
- [ ] ツール定義の型が正確か確認
- [ ] 並列ツール呼び出しのテスト実施
- [ ] マルチステップループが正常に終了するか確認
- [ ] エラーリカバリーのテスト実施
- [ ] 重要な判定フローにHITLを組み込み
- [ ] 状態管理の永続化先を決定
- [ ] レート制限とタイムアウト設定を実装
- [ ] ログ出力とモニタリングを設定
- [ ] コスト推定と予算管理を実施
- [ ] 本番前にA/Bテストを実施
- [ ] ロールバック計画を準備
- [ ] 24時間のサポート体制を確認
まとめ
Gemini 2.5 Proを使用した本番品質のエージェントシステムは、適切な設計原則、堅牢なエラーハンドリング、人間参加型ループの統合により実現できます。上記のパターンと実装例を参考に、信頼性と保守性の高いAIエージェントを構築してください。
個人開発の視点で — エージェントは「人の介在点」を先に置く
個人開発で Dolice Labs の運用をエージェントに任せている私自身、本番でいちばん効いたのは賢いオーケストレーションより「どこで人間が止められるか」を最初に設計しておくことでした。完全自律に振り切ると、ズレたときの被害が一気に広がります。
私は、課金や公開のように取り返しのつかない操作の手前には必ず確認のゲートを置き、状態とログをそこに集約しています。エージェントの自律性は魅力的ですが、止め方を持っているからこそ安心して任せられる、というのが実感です。
ツール呼び出しも、失敗やタイムアウトを前提に組むようにしています。私は各ツールに上限回数とタイムアウトを設定し、超えたら諦めて上位に報告させます。エージェントが「無限に粘って黙り込む」状態を作らないことが、本番で運用を続けられるかどうかの分かれ目だと考えています。