Antigravity のエージェントが暴走・ループする原因と止め方

agent: search_docs(query="auth flow")
  -> []
agent: search_docs(query="auth flow")   ← 同じ引数で再試行
  -> []
agent: search_docs(query="auth flow")   ← 無限に続く

def search_docs(query: str) -> dict:
    """ドキュメント検索。結果の状態を明示的に返す。"""
    results = vector_search(query)
    if not results:
        return {
            "status": "empty",
            "query": query,
            "suggestion": "Try a more specific or different keyword.",
            "results": [],
        }
    return {
        "status": "ok",
        "results": [{"title": r.title, "excerpt": r.excerpt[:500]} for r in results],
    }

from collections import deque
from typing import Any
 
class ToolCallGuard:
    """直近の呼び出しを記憶し、同一呼び出しの繰り返しを検出する。"""
 
    def __init__(self, window: int = 5, max_repeats: int = 2):
        self.window = window
        self.max_repeats = max_repeats
        self.recent = deque(maxlen=window)
 
    def check(self, tool_name: str, args: dict) -> bool:
        """同一呼び出しの繰り返し検出。繰り返しなら False を返す。"""
        key = (tool_name, json.dumps(args, sort_keys=True))
        count = sum(1 for k in self.recent if k == key)
        if count >= self.max_repeats:
            return False
        self.recent.append(key)
        return True
 
# 使用例
guard = ToolCallGuard()
def safe_call(tool_name, args, tools):
    if not guard.check(tool_name, args):
        return {"status": "blocked",
                "reason": f"Tool {tool_name} was called with identical args {args} too many times. Please try a different approach."}
    return tools[tool_name](**args)

class AgentContext:
    """エージェントの実行文脈に深さと呼び出し履歴を持たせる。"""
 
    def __init__(self, max_depth: int = 3, call_chain: tuple = ()):
        self.max_depth = max_depth
        self.call_chain = call_chain
 
    def can_invoke(self, agent_type: str) -> tuple[bool, str]:
        if len(self.call_chain) >= self.max_depth:
            return False, f"max agent depth {self.max_depth} reached"
        if agent_type in self.call_chain:
            return False, f"recursion detected ({agent_type} already in chain)"
        return True, ""
 
    def child(self, agent_type: str) -> "AgentContext":
        return AgentContext(self.max_depth, self.call_chain + (agent_type,))

class PlanController:
    """計画と実行のサイクルを管理し、書き直し上限を守る。"""
 
    def __init__(self, max_replans: int = 2):
        self.max_replans = max_replans
        self.replan_count = 0
        self.failure_history: list[dict] = []
 
    def can_replan(self) -> bool:
        return self.replan_count < self.max_replans
 
    def record_failure(self, step: str, error: str):
        self.failure_history.append({"step": step, "error": error})
 
    def replan_prompt(self) -> str:
        """書き直し上限に達したら「部分達成で終わる」ことを促す指示を返す。"""
        if not self.can_replan():
            return ("書き直し上限に達しました。現時点で完了している成果物を元に、"
                    "未完了部分を明示した上で最終報告を作成してください。")
        self.replan_count += 1
        past = "\n".join(f"- {h['step']}: {h['error']}" for h in self.failure_history)
        return f"以下の失敗を踏まえて計画を改訂してください:\n{past}"

def build_agent_prompt(task: str, acceptance_criteria: list[str]) -> str:
    """受入基準を明示した上でエージェントプロンプトを組み立てる。"""
    criteria_str = "\n".join(f"- {c}" for c in acceptance_criteria)
    return f"""タスク: {task}
 
このタスクは次の条件をすべて満たしたときに完了です:
{criteria_str}
 
重要:
- 完了条件を満たしたら直ちに report_complete ツールで終了してください
- すべてを完璧にする必要はありません。上記条件を満たせば十分です
- 15 ステップを超えても完了していなければ、現状を report_partial で報告して終了してください
"""