Antigravity AgentKit 2.0 実行時エラー完全解決ガイド — tool_call失敗・無限ループ・コンテキスト超過の診断と修正
AgentKit 2.0でエージェントを本番環境に投入すると、開発時には見えなかったエラーが顔を出します。ツール呼び出しの失敗、想定外のループ、コンテキストの突然のリセット——これらは設計上の問題か、実装の落とし穴に起因するものがほとんどです。
AgentKit 2.0のアーキテクチャを理解してからデバッグを始める
効率的なデバッグのために、AgentKit 2.0の実行モデルを整理しておきましょう。
from antigravity.agentkit import Agent, Tool, AgentRunner
# AgentKit 2.0の基本構造
agent = Agent(
name="task_agent",
model="antigravity-pro",
tools=[tool1, tool2, tool3],
instructions="あなたはタスクを実行するエージェントです",
max_iterations=20, # 無限ループ防止の必須設定
)
runner = AgentRunner(agent=agent)AgentKit 2.0のエージェントは「LLMが判断 → ツール実行 → 結果を受け取りLLMが次の判断」というサイクルを繰り返します。エラーは主にこのサイクルの各ステップで起きます。
エラーパターン①：tool_callが失敗・ツールが見つからない
症状: ToolNotFoundError: Tool 'analyze_data' not registered や ToolExecutionError: unexpected argument 'format'.
from antigravity.agentkit import Agent, Tool
from antigravity.agentkit.exceptions import ToolNotFoundError, ToolExecutionError
import logging
logger = logging.getLogger(__name__)
# ❌ よくある間違い：ツール名の不一致
def analyze_data_function(data: str, output_format: str = "json") -> dict:
# ツール関数の実装
return {"result": "processed"}
tool_wrong = Tool(
name="analyze", # ← エージェントが "analyze_data" と呼び出すと見つからない
function=analyze_data_function
)
# ✅ 正しい実装：ツール名とdescriptionを明確に
def analyze_data(data: str, output_format: str = "json") -> dict:
"""
データを分析して構造化された結果を返す。
Args:
data: 分析するデータ（JSON文字列またはテキスト）
output_format: 出力形式 ("json" または "text")
Returns:
分析結果の辞書
"""
try:
processed = process_data(data, output_format)
return {"status": "success", "result": processed}
except ValueError as e:
# ツール内でのエラーは明確なエラー情報を返す
return {"status": "error", "error": str(e), "input_received": data[:100]}
tool_correct = Tool(
name="analyze_data", # エージェントが呼び出す名前と一致させる
function=analyze_data,
description="データを分析します。テキストやJSON形式の入力を受け付け、構造化された結果を返します"
)
def build_agent_with_error_handling() -> Agent:
"""エラーハンドリング付きのエージェント構築"""
tools = [tool_correct]
# ツールが正しく登録されているか確認
registered_names = [t.name for t in tools]
logger.info(f"登録ツール: {registered_names}")
return Agent(
name="robust_agent",
model="antigravity-pro",
tools=tools,
instructions=f"""
あなたはデータ分析エージェントです。
利用可能なツール: {', '.join(registered_names)}
ツールの実行に失敗した場合は、エラーメッセージを確認して引数を修正して再試行してください。
""",
max_iterations=15
)エラーパターン②：エージェントの無限ループ
症状: エージェントが同じツールを何度も繰り返し呼び出し、タスクが完了しません。
from antigravity.agentkit import Agent, AgentRunner
from antigravity.agentkit.callbacks import BaseCallback
from datetime import datetime
from collections import Counter
class LoopDetectionCallback(BaseCallback):
"""無限ループを検出して介入するコールバック"""
def __init__(self, max_same_tool_calls: int = 3, time_window_seconds: int = 60):
self.tool_call_history = []
self.max_same_tool_calls = max_same_tool_calls
self.time_window = time_window_seconds
def on_tool_start(self, tool_name: str, tool_input: dict, **kwargs):
"""ツール呼び出し前に呼ばれる"""
now = datetime.now().timestamp()
# 時間窓内の呼び出し履歴を追跡
self.tool_call_history.append({
"tool": tool_name,
"input_hash": hash(str(sorted(tool_input.items()))),
"timestamp": now
})
# 古い履歴を削除
self.tool_call_history = [
h for h in self.tool_call_history
if now - h["timestamp"] < self.time_window
]
# 同じツール×同じ引数の呼び出しが繰り返されていないか確認
recent_calls = [
h for h in self.tool_call_history
if h["tool"] == tool_name and h["input_hash"] == hash(str(sorted(tool_input.items())))
]
if len(recent_calls) >= self.max_same_tool_calls:
raise RuntimeError(
f"無限ループ検出: {tool_name}が同じ引数で{len(recent_calls)}回呼ばれました。"
f"タスクの完了条件を見直してください。"
)
def on_agent_iteration(self, iteration: int, messages: list, **kwargs):
"""各イテレーション完了時に呼ばれる"""
if iteration % 5 == 0: # 5イテレーションごとにログ
recent_tools = [h["tool"] for h in self.tool_call_history[-10:]]
logger.info(f"Iteration {iteration}. Recent tools: {Counter(recent_tools)}")
# エージェントの終了条件を明確に設定する
agent = Agent(
name="loop_safe_agent",
model="antigravity-pro",
tools=tools,
instructions="""
タスクを完了したら、必ず "TASK_COMPLETE: [結果の要約]" という形式で最終回答してください。
同じ操作を3回以上繰り返すことなく、効率的に目標を達成してください。
情報が不足している場合は、ユーザーに確認を求める前に一度推論で補完を試みてください。
""",
max_iterations=20,
callbacks=[LoopDetectionCallback(max_same_tool_calls=3)]
)ループを引き起こしやすいinstruction（悪い例）と改善例:
# ❌ 終了条件が曖昧なinstruction
bad_instructions = """
情報を収集して分析してください。
必要に応じてツールを使ってください。
"""
# ✅ 終了条件が明確なinstruction
good_instructions = """
以下の手順でタスクを完了してください:
1. search_webツールで情報を1〜2回収集する
2. analyze_dataツールで結果を分析する（最大1回）
3. 分析結果を基に最終回答を生成する
手順が完了したら "完了: [要約]" で回答を終えてください。
追加のデータ収集が必要と感じても、3回以内にとどめてください。
"""エラーパターン③：コンテキストウィンドウの超過
症状: ContextWindowExceededError または長いタスクの途中でエージェントが「以前の情報を忘れた」ように振る舞う。
from antigravity.agentkit import Agent
from antigravity.agentkit.memory import SlidingWindowMemory, SummaryMemory
# ❌ デフォルトのメモリ設定では長期タスクでコンテキストが溢れる
# ✅ スライディングウィンドウメモリで最新の会話のみ保持
agent_sliding = Agent(
name="long_task_agent",
model="antigravity-pro",
tools=tools,
memory=SlidingWindowMemory(
max_messages=30, # 最新30メッセージのみ保持
preserve_system=True # システムメッセージは常に保持
)
)
# ✅ 要約メモリで重要情報を圧縮して保持
agent_summary = Agent(
name="summarizing_agent",
model="antigravity-pro",
tools=tools,
memory=SummaryMemory(
summary_interval=10, # 10メッセージごとに要約
max_summary_length=500 # 要約の最大文字数
)
)
# コンテキスト使用量を監視する
class ContextMonitorCallback(BaseCallback):
def on_llm_response(self, response, **kwargs):
usage = response.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
total_tokens = usage.get("total_tokens", 0)
# コンテキスト使用率を計算（モデルの上限値に応じて調整）
context_limit = 200_000 # antigravity-proの場合
usage_percent = (prompt_tokens / context_limit) * 100
if usage_percent > 70:
logger.warning(
f"コンテキスト使用率: {usage_percent:.1f}% "
f"({prompt_tokens}/{context_limit} tokens). "
"メモリの圧縮またはタスクの分割を検討してください。"
)
# 長大なタスクを複数エージェントに分割する
async def process_large_task(task: str, data_chunks: list) -> dict:
"""大きなタスクをチャンクに分割して複数エージェントで処理する"""
results = []
for i, chunk in enumerate(data_chunks):
# 各チャンクを独立したエージェントで処理
chunk_agent = Agent(
name=f"chunk_agent_{i}",
model="antigravity-pro",
tools=tools,
max_iterations=10
)
runner = AgentRunner(agent=chunk_agent)
result = await runner.run(
f"以下のデータチャンク（{i+1}/{len(data_chunks)}）を処理してください:\n{chunk}"
)
results.append(result)
# 集約エージェントで結果をまとめる
aggregator = Agent(
name="aggregator",
model="antigravity-pro",
tools=[], # ツール不要の集約専用エージェント
max_iterations=5
)
agg_runner = AgentRunner(agent=aggregator)
return await agg_runner.run(
f"以下の{len(results)}つのチャンク処理結果を統合して最終回答を作成してください:\n" +
"\n---\n".join([str(r) for r in results])
)エラーパターン④：並列エージェントの同期エラー
症状: 複数のエージェントを並列実行すると、一部のエージェントが別エージェントの出力を正しく受け取れない、または順序が乱れます。
import asyncio
from antigravity.agentkit import Agent, AgentRunner
from typing import Optional
class ParallelAgentOrchestrator:
"""並列エージェント実行のエラーハンドリング付きオーケストレーター"""
def __init__(self, agents: list[Agent], timeout: float = 120.0):
self.agents = agents
self.timeout = timeout
self.results = {}
self.errors = {}
async def run_agent_safely(self, agent: Agent, task: str) -> Optional[dict]:
"""タイムアウトとエラーハンドリング付きでエージェントを実行する"""
try:
runner = AgentRunner(agent=agent)
result = await asyncio.wait_for(
runner.run(task),
timeout=self.timeout
)
return {"agent": agent.name, "status": "success", "result": result}
except asyncio.TimeoutError:
logger.error(f"エージェント {agent.name} がタイムアウト ({self.timeout}秒)")
return {"agent": agent.name, "status": "timeout", "result": None}
except Exception as e:
logger.error(f"エージェント {agent.name} でエラー: {e}")
return {"agent": agent.name, "status": "error", "error": str(e), "result": None}
async def run_all(self, tasks: dict[str, str]) -> dict:
"""全エージェントを並列実行し、失敗したものを記録する"""
if len(tasks) != len(self.agents):
raise ValueError(f"エージェント数({len(self.agents)})とタスク数({len(tasks)})が一致しません")
# 全エージェントを並列実行
coroutines = [
self.run_agent_safely(agent, tasks[agent.name])
for agent in self.agents
if agent.name in tasks
]
results = await asyncio.gather(*coroutines, return_exceptions=False)
# 成功・失敗を分類
successful = [r for r in results if r and r["status"] == "success"]
failed = [r for r in results if r and r["status"] != "success"]
if failed:
logger.warning(
f"並列実行: {len(successful)}件成功, {len(failed)}件失敗\n"
+ "\n".join([f" - {r['agent']}: {r['status']}" for r in failed])
)
return {
"successful": successful,
"failed": failed,
"total": len(results)
}
# 使用例
async def multi_agent_analysis(document: str) -> dict:
"""ドキュメントを複数のエージェントで並列分析する"""
agents = [
Agent(name="sentiment_agent", model="antigravity-pro", tools=[sentiment_tool], max_iterations=5),
Agent(name="summary_agent", model="antigravity-pro", tools=[summary_tool], max_iterations=5),
Agent(name="keyword_agent", model="antigravity-pro", tools=[keyword_tool], max_iterations=5),
]
orchestrator = ParallelAgentOrchestrator(agents=agents, timeout=60.0)
tasks = {
"sentiment_agent": f"次のドキュメントのセンチメント分析を行ってください:\n{document[:5000]}",
"summary_agent": f"次のドキュメントを3行で要約してください:\n{document[:5000]}",
"keyword_agent": f"次のドキュメントの重要キーワードを10個抽出してください:\n{document[:5000]}",
}
return await orchestrator.run_all(tasks)エラーパターン⑤：ツール間の依存関係エラーと循環参照
症状: ツールAがツールBの結果を必要とし、ツールBがツールAを呼び出すような循環参照が発生します。
from typing import Optional
from functools import wraps
# ツール依存グラフの定義と検証
class ToolDependencyGraph:
"""ツール依存関係を管理し、循環参照を検出する"""
def __init__(self):
self.dependencies = {} # ツール名 -> 依存ツール名のセット
def add_dependency(self, tool: str, depends_on: list[str]):
if tool not in self.dependencies:
self.dependencies[tool] = set()
self.dependencies[tool].update(depends_on)
def check_cycles(self) -> list[list[str]]:
"""深さ優先探索で循環参照を検出する"""
visited = set()
path = []
cycles = []
def dfs(node):
if node in path:
cycle_start = path.index(node)
cycles.append(path[cycle_start:] + [node])
return
if node in visited:
return
visited.add(node)
path.append(node)
for dep in self.dependencies.get(node, set()):
dfs(dep)
path.pop()
for tool in self.dependencies:
if tool not in visited:
dfs(tool)
return cycles
def validate(self):
cycles = self.check_cycles()
if cycles:
cycle_str = " → ".join(cycles[0])
raise ValueError(f"ツール依存関係に循環参照があります: {cycle_str}")
return True
# ツール実行のメモ化（同じ引数での重複実行を防ぐ）
def memoize_tool(ttl_seconds: int = 300):
"""ツール関数の結果をキャッシュするデコレーター"""
cache = {}
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
import time
cache_key = (args, tuple(sorted(kwargs.items())))
if cache_key in cache:
result, timestamp = cache[cache_key]
if time.time() - timestamp < ttl_seconds:
logger.debug(f"キャッシュヒット: {func.__name__}({cache_key})")
return result
result = func(*args, **kwargs)
cache[cache_key] = (result, time.time())
return result
return wrapper
return decorator
@memoize_tool(ttl_seconds=60)
def fetch_user_data(user_id: str) -> dict:
"""ユーザーデータを取得（60秒間キャッシュ）"""
# 実際のAPI呼び出し
return api_client.get_user(user_id)AgentKit 2.0ランタイムのデバッグとロギング
import logging
from antigravity.agentkit.callbacks import BaseCallback
# 構造化ロギングの設定
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s [%(name)s] %(levelname)s: %(message)s'
)
class DetailedExecutionLogger(BaseCallback):
"""AgentKit 2.0の実行を詳細にログ記録するコールバック"""
def __init__(self, log_level: str = "INFO"):
self.logger = logging.getLogger("agentkit.execution")
self.iteration = 0
self.tool_calls = []
def on_agent_start(self, agent_name: str, task: str, **kwargs):
self.logger.info(f"=== エージェント開始: {agent_name} ===")
self.logger.info(f"タスク: {task[:200]}...")
def on_llm_start(self, messages: list, **kwargs):
self.iteration += 1
self.logger.debug(f"[Iteration {self.iteration}] LLM呼び出し: {len(messages)}メッセージ")
def on_llm_response(self, response, **kwargs):
content = response.get("content", "")
tool_calls = response.get("tool_calls", [])
if tool_calls:
for tc in tool_calls:
self.logger.info(f"ツール呼び出し: {tc['name']}({tc.get('arguments', {})})")
self.tool_calls.append(tc['name'])
else:
self.logger.info(f"最終回答生成: {content[:200]}...")
def on_tool_end(self, tool_name: str, result, **kwargs):
result_preview = str(result)[:100]
self.logger.debug(f"ツール結果: {tool_name} → {result_preview}")
def on_agent_end(self, result, **kwargs):
from collections import Counter
tool_summary = Counter(self.tool_calls)
self.logger.info(f"=== エージェント完了 ===")
self.logger.info(f"総イテレーション: {self.iteration}")
self.logger.info(f"ツール使用回数: {dict(tool_summary)}")
# エラー時のリカバリー設計
class AgentWithFallback:
"""エラー発生時にフォールバック戦略を実行するエージェントラッパー"""
def __init__(self, primary_agent: Agent, fallback_agent: Agent):
self.primary = primary_agent
self.fallback = fallback_agent
async def run(self, task: str, max_retries: int = 2) -> dict:
for attempt in range(max_retries):
try:
runner = AgentRunner(
agent=self.primary,
callbacks=[DetailedExecutionLogger()]
)
result = await runner.run(task)
return {"agent": "primary", "result": result}
except RuntimeError as e:
if "無限ループ" in str(e) or "max_iterations" in str(e):
logger.warning(f"プライマリエージェント失敗（試行{attempt+1}）: {e}")
if attempt == max_retries - 1:
# フォールバックエージェントに切り替え
logger.info("フォールバックエージェントに切り替えます")
fallback_runner = AgentRunner(agent=self.fallback)
result = await fallback_runner.run(task)
return {"agent": "fallback", "result": result}
else:
raise
raise RuntimeError(f"{max_retries}回の試行が全て失敗しました")全体を振り返って
AgentKit 2.0のランタイムエラーの主なパターンと解決策:
- tool_call失敗 → ツール名の一致確認、引数型の検証、エラー時の明確なフィードバック返却
- 無限ループ →
max_iterations必須設定、LoopDetectionCallback、終了条件を明確にしたinstruction - コンテキスト超過 →
SlidingWindowMemory/SummaryMemory、タスク分割と集約エージェント設計 - 並列同期エラー → タイムアウト付き並列実行、失敗エージェントの記録とリカバリー設計
- ツール依存循環参照 →
ToolDependencyGraphで事前検証、メモ化で重複実行を防止
本番投入前にDetailedExecutionLoggerを活用して実行パターンを可視化し、ループや過剰なツール呼び出しが起きていないかを確認する習慣をつけることが、安定稼働への近道です。
AgentKit 2.0の本番運用パターンについてさらに深めたい方には、Antigravity マルチエージェント本番運用ガイドもあわせてご覧ください。
個人開発12年とアーティスト活動から見るエージェント設計
線引きするときの3つの判断軸
- 失敗時の影響が金銭やユーザー体験にどれだけ波及するか
- 復旧オペレーションが明文化されているか
- 観測ログから人間が再現できる粒度に整っているか
まとめ — 検証ステップの推奨
- 観測メトリクスとアラートを設置
- 限定ロールアウトで本番検証