Google ADKは強力です。でも、使えない場面は少なくありません。
社内のセキュリティポリシーでサードパーティフレームワーク禁止、既存のPythonコードベースに組み込みたい、ADKの抽象化が邪魔でデバッグが難しい——そういった理由で「Gemini APIを直接使ってエージェントを自作したい」という相談を受けることがあります。
私自身、個人開発のプロダクトにAIエージェントを組み込む際、ADKの初期化オーバーヘッドとサイズが気になって生APIを選んだ経験があります。結果、コードの見通しが良くなり、問題が起きたときの原因特定も早くなりました。
ここではGemini APIのFunction CallingとContent APIを組み合わせて、本番で動くカスタムエージェントループを一から設計・実装する方法を詳しく解説します。ADKの内部で何が行われているかを理解すると、ADKをより上手く使えるようにもなります。
エージェントループの基本構造
エージェントとは「モデルが自律的にツールを選び、実行し、結果を見て次の判断をする」サイクルを繰り返すシステムです。最小構成はこうなります。
# agent_core.py
import google.generativeai as genai
from typing import Callable, Any
import json
genai.configure(api_key="YOUR_GEMINI_API_KEY")
class AgentLoop:
"""最小構成のGemini APIエージェントループ"""
def __init__(
self,
model_name: str = "gemini-2.5-pro",
tools: list[dict] = None,
tool_functions: dict[str, Callable] = None,
max_iterations: int = 10, # 無限ループ防止
):
self.model = genai.GenerativeModel(
model_name=model_name,
tools=tools or [],
)
self.tool_functions = tool_functions or {}
self.max_iterations = max_iterations
def run(self, user_message: str) -> str:
"""エージェントループを実行して最終回答を返す"""
chat = self.model.start_chat()
response = chat.send_message(user_message)
for iteration in range(self.max_iterations):
# 1. モデルがツール呼び出しを要求しているか確認
if not response.candidates[0].content.parts:
break
tool_calls = [
part for part in response.candidates[0].content.parts
if hasattr(part, "function_call") and part.function_call.name
]
# 2. ツール呼び出しがなければ最終回答
if not tool_calls:
return response.text
# 3. ツールを実行して結果を返す
tool_results = []
for call in tool_calls:
result = self._execute_tool(
call.function_call.name,
dict(call.function_call.args)
)
tool_results.append({
"function_response": {
"name": call.function_call.name,
"response": {"result": str(result)},
}
})
# 4. ツール結果をモデルに送り次のステップへ
response = chat.send_message(tool_results)
# max_iterations到達時は最後のテキスト回答を返す
return response.text if hasattr(response, "text") else "エージェントが最大反復回数に達しました"
def _execute_tool(self, name: str, args: dict) -> Any:
"""ツール関数を安全に実行する"""
if name not in self.tool_functions:
return f"エラー: ツール '{name}' が見つかりません"
try:
return self.tool_functions[name](**args)
except Exception as e:
return f"ツール実行エラー: {str(e)}"このコアの動作を理解するのが、すべての出発点です。max_iterations で無限ループを防いでいる点と、ツール実行を try/except で囲んでいる点に注目してください。本番では必ずこの2つが必要です。
Function Calling の基礎から確認したい場合はこちら
ツール定義の設計パターン
ツールの定義方法がエージェントの品質を大きく左右します。モデルはツールのdescriptionを読んで「いつ・どう使うか」を判断するため、説明文が曖昧だとツール選択が不安定になります。
# tools.py
# ツール定義とツール関数を一元管理する
# --- ツール定義（Gemini APIに渡すスキーマ） ---
TOOL_DEFINITIONS = [
{
"function_declarations": [
{
"name": "search_documents",
"description": (
"社内ドキュメントをキーワード検索します。"
"ユーザーが特定の情報を探しているとき、"
"または質問に答えるために参照情報が必要なときに使います。"
"一般的な知識の質問には使いません。"
),
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "検索クエリ。具体的なキーワードを含めると精度が上がります。",
},
"max_results": {
"type": "integer",
"description": "返す結果の最大数。デフォルト5、最大20。",
"default": 5,
},
},
"required": ["query"],
},
},
{
"name": "execute_python_code",
"description": (
"Pythonコードを安全なサンドボックスで実行します。"
"計算、データ変換、リスト操作など、"
"確実な処理が必要なときにのみ使います。"
"ファイルシステムやネットワークへのアクセスはできません。"
),
"parameters": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "実行するPythonコード。標準ライブラリのみ使用可能。",
},
},
"required": ["code"],
},
},
]
}
]
# --- ツール実装 ---
import subprocess
import tempfile
import os
def search_documents(query: str, max_results: int = 5) -> list[dict]:
"""実際の検索ロジック（例: Elasticsearch, Pinecone等）"""
# ここに実際のベクター検索やキーワード検索を実装する
# サンプルとして固定データを返す
return [
{"id": "doc-1", "title": "APIガイド", "snippet": f"'{query}'に関連するドキュメント..."}
][:max_results]
def execute_python_code(code: str) -> str:
"""Pythonコードを制限付きサンドボックスで実行する"""
# 危険なimportをブロック
blocked_modules = ["os", "subprocess", "socket", "requests", "http"]
for module in blocked_modules:
if f"import {module}" in code or f"from {module}" in code:
return f"エラー: '{module}' モジュールの使用は許可されていません"
with tempfile.NamedTemporaryFile(suffix=".py", mode="w", delete=False) as f:
f.write(code)
temp_path = f.name
try:
result = subprocess.run(
["python3", temp_path],
capture_output=True,
text=True,
timeout=10, # 10秒タイムアウト
)
return result.stdout if result.returncode == 0 else f"実行エラー: {result.stderr}"
except subprocess.TimeoutExpired:
return "エラー: コード実行がタイムアウトしました（10秒）"
finally:
os.unlink(temp_path)
# ツール関数マッピング
TOOL_FUNCTIONS = {
"search_documents": search_documents,
"execute_python_code": execute_python_code,
}ツール定義で特に重要なポイントは description の書き方です。「いつ使うか」「いつ使わないか」を両方書くと、モデルが不要なツール呼び出しをしなくなります。execute_python_code の説明に「ファイルシステムやネットワークへのアクセスはできません」と書いているのは、モデルにセキュリティ上の制約を先に伝えておくためです。
会話メモリの設計
標準のChatセッションは会話履歴を自動で管理しますが、本番では「長い会話でトークンが超過する」「セッションを跨いで記憶を持続させたい」という問題が必ず出てきます。
# memory.py
from google.generativeai.types import content_types
import json
from typing import Optional
from dataclasses import dataclass, field
import time
@dataclass
class ConversationTurn:
role: str # "user" or "model"
content: str
timestamp: float = field(default_factory=time.time)
token_count: Optional[int] = None
class SlidingWindowMemory:
"""
スライディングウィンドウ方式の会話メモリ。
古い会話を削除してトークン上限を防ぐ。
"""
def __init__(
self,
max_turns: int = 20, # 保持する会話ターン数
summarize_threshold: int = 15, # この数を超えたら古い会話を要約
model_name: str = "gemini-2.5-pro",
):
self.max_turns = max_turns
self.summarize_threshold = summarize_threshold
self.model_name = model_name
self.turns: list[ConversationTurn] = []
self.summary: Optional[str] = None
def add_turn(self, role: str, content: str):
self.turns.append(ConversationTurn(role=role, content=content))
# 閾値を超えたら古い会話を要約して圧縮
if len(self.turns) >= self.summarize_threshold:
self._compress_old_turns()
def _compress_old_turns(self):
"""
古い会話（最初の半分）を要約して保存。
最新のターンは詳細を維持する。
"""
import google.generativeai as genai
# 要約対象は古いほうの半分
compress_count = len(self.turns) // 2
turns_to_compress = self.turns[:compress_count]
# 会話を文字列に変換
conversation_text = "\n".join([
f"{t.role}: {t.content}" for t in turns_to_compress
])
model = genai.GenerativeModel(self.model_name)
summary_prompt = (
f"以下の会話を3〜5文で簡潔に要約してください。"
f"重要な事実・決定・コンテキストを保持してください：\n\n{conversation_text}"
)
try:
response = model.generate_content(summary_prompt)
new_summary = response.text
# 既存の要約がある場合は連結
if self.summary:
self.summary = f"{self.summary}\n\n[追加の要約] {new_summary}"
else:
self.summary = new_summary
# 要約済みのターンを削除
self.turns = self.turns[compress_count:]
except Exception as e:
# 要約失敗時は古いターンをそのまま削除（データ損失より安全性を優先）
self.turns = self.turns[compress_count:]
def get_history_for_api(self) -> list[dict]:
"""Gemini API の chat history 形式に変換する"""
history = []
# 要約がある場合はシステムメッセージとして先頭に追加
if self.summary:
history.append({
"role": "user",
"parts": [f"[会話の要約] {self.summary}"],
})
history.append({
"role": "model",
"parts": ["了解しました。これまでの会話の文脈を把握した上で対応します。"],
})
# 残りのターンを追加
for turn in self.turns:
history.append({
"role": turn.role,
"parts": [turn.content],
})
return history
def to_json(self) -> str:
"""永続化のためのシリアライズ"""
return json.dumps({
"summary": self.summary,
"turns": [
{"role": t.role, "content": t.content, "timestamp": t.timestamp}
for t in self.turns
],
}, ensure_ascii=False)
@classmethod
def from_json(cls, json_str: str) -> "SlidingWindowMemory":
"""JSONから復元"""
data = json.loads(json_str)
memory = cls()
memory.summary = data.get("summary")
memory.turns = [
ConversationTurn(**t) for t in data.get("turns", [])
]
return memoryこのメモリ設計で重要なのは「要約に失敗しても動作を継続する」フォールバック設計です。要約APIが遅延しても、本体のエージェント処理が止まらないようにしています。
並列エージェント実行
複雑なタスクを並列で処理すると、実行時間を大幅に短縮できます。たとえば「複数のWebサイトを調査して比較する」タスクは、順番に実行するより並列で実行したほうが明らかに速くなります。
# parallel_agent.py
import asyncio
import google.generativeai as genai
from typing import Any
import time
class ParallelAgentOrchestrator:
"""
複数のサブエージェントを並列実行するオーケストレーター。
各サブタスクを独立したエージェントに委任する。
"""
def __init__(self, model_name: str = "gemini-2.5-flash"):
# 並列実行にはコストが低いFlashモデルを推奨
self.model_name = model_name
async def run_subagent(
self,
task_name: str,
task_prompt: str,
tools: list[dict],
tool_functions: dict,
) -> dict[str, Any]:
"""単一サブエージェントを非同期で実行する"""
start_time = time.time()
try:
model = genai.GenerativeModel(
model_name=self.model_name,
tools=tools,
)
chat = model.start_chat()
response = chat.send_message(task_prompt)
# ツール呼び出しループ（最大5回）
for _ in range(5):
tool_calls = [
part for part in response.candidates[0].content.parts
if hasattr(part, "function_call") and part.function_call.name
]
if not tool_calls:
break
tool_results = []
for call in tool_calls:
func = tool_functions.get(call.function_call.name)
if func:
result = func(**dict(call.function_call.args))
else:
result = f"ツール未定義: {call.function_call.name}"
tool_results.append({
"function_response": {
"name": call.function_call.name,
"response": {"result": str(result)},
}
})
response = chat.send_message(tool_results)
return {
"task_name": task_name,
"result": response.text,
"elapsed_seconds": time.time() - start_time,
"status": "success",
}
except Exception as e:
return {
"task_name": task_name,
"result": None,
"elapsed_seconds": time.time() - start_time,
"status": "error",
"error": str(e),
}
async def run_parallel(
self,
subtasks: list[dict],
) -> list[dict]:
"""
複数サブタスクを並列実行する。
subtasks形式:
[
{
"name": "task_name",
"prompt": "タスクの内容",
"tools": [...],
"tool_functions": {...},
},
...
]
"""
coroutines = [
self.run_subagent(
task_name=task["name"],
task_prompt=task["prompt"],
tools=task.get("tools", []),
tool_functions=task.get("tool_functions", {}),
)
for task in subtasks
]
# asyncio.gather で並列実行（例外が出ても他タスクは継続）
results = await asyncio.gather(*coroutines, return_exceptions=False)
return list(results)
# 使用例
async def example_parallel_research():
orchestrator = ParallelAgentOrchestrator()
# 3つの競合製品を並列調査
subtasks = [
{
"name": "product_a_research",
"prompt": "製品Aの価格・機能・レビューを調査してまとめてください",
"tools": TOOL_DEFINITIONS,
"tool_functions": TOOL_FUNCTIONS,
},
{
"name": "product_b_research",
"prompt": "製品Bの価格・機能・レビューを調査してまとめてください",
"tools": TOOL_DEFINITIONS,
"tool_functions": TOOL_FUNCTIONS,
},
{
"name": "product_c_research",
"prompt": "製品Cの価格・機能・レビューを調査してまとめてください",
"tools": TOOL_DEFINITIONS,
"tool_functions": TOOL_FUNCTIONS,
},
]
results = await orchestrator.run_parallel(subtasks)
# 結果を集約するプロセスエージェントに渡す
all_research = "\n\n".join([
f"【{r['task_name']}】\n{r['result']}"
for r in results
if r["status"] == "success"
])
return all_research
if __name__ == "__main__":
result = asyncio.run(example_parallel_research())
print(result)並列実行で気をつけるのは レート制限です。3つのエージェントを同時に走らせると、APIリクエストが同時に増えます。Geminiの無料枠では秒間リクエスト数の上限に引っかかることがあるため、並列数の上限（asyncio.Semaphore）を設けるのが実用的です。
並列Function Callingのパターンはこちらも参照
エラー回復とリトライ設計
本番でエージェントを運用すると、さまざまな理由でAPIが失敗します。503エラー、レート制限、タイムアウト——そのたびにユーザーにエラーを見せるわけにはいきません。
# retry.py
import google.generativeai as genai
from google.api_core import exceptions as google_exceptions
import time
import random
from typing import Callable, TypeVar, Any
from functools import wraps
T = TypeVar("T")
class AgentRetryPolicy:
"""
指数バックオフ + ジッター付きリトライポリシー。
本番で安定稼働するエージェントに必須。
"""
def __init__(
self,
max_retries: int = 3,
base_delay_seconds: float = 1.0,
max_delay_seconds: float = 60.0,
retryable_exceptions: tuple = None,
):
self.max_retries = max_retries
self.base_delay = base_delay_seconds
self.max_delay = max_delay_seconds
self.retryable_exceptions = retryable_exceptions or (
google_exceptions.ResourceExhausted, # 429: レート制限
google_exceptions.ServiceUnavailable, # 503: サーバーエラー
google_exceptions.DeadlineExceeded, # タイムアウト
ConnectionError,
TimeoutError,
)
def execute_with_retry(self, func: Callable[..., T], *args, **kwargs) -> T:
"""
関数をリトライポリシーに従って実行する。
期待する出力（成功時）:
>>> policy = AgentRetryPolicy(max_retries=3)
>>> result = policy.execute_with_retry(my_api_call, "input")
>>> print(result) # APIの正常レスポンス
期待する出力（全試行失敗時）:
>>> # 3回失敗後に最後の例外を raise する
"""
last_exception = None
for attempt in range(self.max_retries + 1):
try:
return func(*args, **kwargs)
except self.retryable_exceptions as e:
last_exception = e
if attempt == self.max_retries:
break
# 指数バックオフ（2^attempt * base_delay）+ ランダムジッター
delay = min(
self.base_delay * (2 ** attempt) + random.uniform(0, 1),
self.max_delay,
)
print(f"⚠️ 試行 {attempt + 1}/{self.max_retries} 失敗: {type(e).__name__}. "
f"{delay:.1f}秒後にリトライします...")
time.sleep(delay)
except Exception as e:
# リトライ対象外の例外は即座に raise
raise
raise last_exception
# デコレーター版（既存の関数に簡単に適用できる）
def with_retry(max_retries: int = 3, base_delay: float = 1.0):
def decorator(func: Callable) -> Callable:
policy = AgentRetryPolicy(
max_retries=max_retries,
base_delay_seconds=base_delay,
)
@wraps(func)
def wrapper(*args, **kwargs):
return policy.execute_with_retry(func, *args, **kwargs)
return wrapper
return decorator
# 使用例
class ResilientAgent:
"""リトライポリシーを組み込んだ堅牢なエージェント"""
def __init__(self):
self.model = genai.GenerativeModel("gemini-2.5-pro")
self.retry_policy = AgentRetryPolicy(max_retries=3)
@with_retry(max_retries=3, base_delay=1.0)
def send_message_with_retry(self, chat, message):
"""リトライ付きメッセージ送信"""
return chat.send_message(message)
def run(self, prompt: str) -> str:
chat = self.model.start_chat()
# リトライ機能付きでAPI呼び出し
response = self.retry_policy.execute_with_retry(
chat.send_message, prompt
)
return response.textなぜジッターが重要か、少し説明します。複数のエージェントが同時に失敗してリトライすると、全員が同じタイミングでAPIを叩いて再びエラーになるサンダーリング・ハード問題が起きます。random.uniform(0, 1) を足すだけで、各エージェントのリトライタイミングがばらけて衝突を避けられます。
エラーハンドリングとレート制限の詳細パターン
よくある落とし穴と対処法
ADKなしでエージェントを実装すると、必ず踏む地雷があります。私が実際に経験したものを中心にまとめます。
落とし穴1: ツール結果のフォーマットエラー
ツール結果を返すとき、function_response の形式が微妙に間違っているとAPIが無視したり、モデルが混乱したりします。
# ❌ 間違い: 結果を文字列で直接渡す
tool_result = "検索結果: 3件見つかりました"
# ❌ 間違い: responseキーがない
tool_result = {
"function_response": {
"name": "search_documents",
"content": "検索結果", # ← "response" キーが必要
}
}
# ✅ 正しい形式
tool_result = {
"function_response": {
"name": "search_documents", # ← ツール名と完全一致
"response": { # ← "response" キーが必須
"result": "検索結果の内容", # ← 任意のキーで値を返す
},
}
}落とし穴2: ツール名の大文字小文字
ツール定義に書いた名前と、function_call.name で返ってくる名前は大文字小文字まで完全一致している必要があります。search_Documents と search_documents は別物として扱われます。
# ツール定義
{"name": "search_documents", ...}
# 呼び出し時の確認
tool_name = part.function_call.name
# ↑ "search_documents" と完全一致するか確認してから実行する
if tool_name in self.tool_functions:
result = self.tool_functions[tool_name](**args)
else:
result = f"未定義のツール: {tool_name}" # サイレントに失敗させない落とし穴3: 空のParts確認漏れ
# ❌ 危険: parts が空の場合に IndexError
first_part = response.candidates[0].content.parts[0]
# ✅ 安全な確認
parts = response.candidates[0].content.parts if response.candidates else []
tool_calls = [p for p in parts if hasattr(p, "function_call") and p.function_call.name]
if not tool_calls:
# テキスト回答を返す
return response.text or "回答がありませんでした"落とし穴4: Proモデルをすべてに使う
エージェントのすべてのステップでGemini 2.5 Proを使うとコストが急増します。私が好んでいる方針は、ルーティング・要約・ツール結果の評価にはFlashを使い、最終的な回答生成にだけProを使う設計です。
# コスト最適化: 役割によってモデルを使い分ける
ORCHESTRATOR_MODEL = "gemini-2.5-pro" # 最終判断・複雑な推論
SUBAGENT_MODEL = "gemini-2.5-flash" # ツール実行・中間処理
SUMMARIZE_MODEL = "gemini-2.5-flash" # 要約・分類ADKとカスタム実装の選択基準
最後に、どちらを選ぶべきかの判断軸を共有します。
ADKが向いている場面:
- チームでエージェントを開発するとき（標準化が重要）
- Google Cloudの他サービスと深く統合するとき
- 実装スピードを最優先にするとき
- エージェントの設計経験が浅いとき
カスタム実装が向いている場面:
- 既存のPythonコードベースへの統合
- フレームワークの更新による予期しない挙動変更を避けたいとき
- デバッグのしやすさを優先するとき
- Cloudflare WorkersやLambdaなど軽量環境へのデプロイ
- ADKに含まれない独自の最適化（キャッシュ戦略など）が必要なとき
どちらが「正解」かよりも、自分のプロジェクトの要件に合っているかを基準に選ぶのが現実的です。
本番エージェントのモニタリング設計はこちら Google ADKを使うマルチエージェント設計はこちら
締めくくり
今日からできる一歩は、最初のコアループ（AgentLoop クラス）をコピーして、自分のプロジェクトのツール1つを繋いでみることです。エラーが出ても、ツール結果のフォーマットを見直すとたいてい解決します。
エージェントを自作すると、「モデルがどこで判断に迷っているか」がログに明確に現れます。その観察から、System Instructionのチューニングやツール定義の改善が具体的に見えてくるはずです。ADKも内部では同じことをやっています。一度自作してみると、ADKを使うときの理解度も上がります。
さらに深く