AIエージェントのツール呼び出しループを止める：5層ガードの実装

import copy
import json

class Guard:
    def __init__(self, now, verify, max_models=8, max_tools=6,
                 duration=60, repeat_limit=2, stall_limit=3):
        self.now, self.verify = now, verify
        self.deadline = now() + duration
        self.max_models, self.max_tools = max_models, max_tools
        self.repeat_limit, self.stall_limit = repeat_limit, stall_limit
        self.models = self.tools = self.progress = self.stalled = 0
        self.repeats, self.operations = {}, {}
        self.ready_at = now()
        self.terminal = None

    def end(self, reason, state="PARTIAL"):
        if self.terminal is None:
            self.terminal = {"state": state, "reason": reason,
                             "model_calls": self.models,
                             "tool_calls": self.tools,
                             "progress": self.progress}
        return dict(self.terminal)

    def boundary(self):
        if self.terminal is not None:
            return dict(self.terminal)
        if self.now() >= self.deadline:
            return self.end("deadline")
        return None

    def model_gate(self):
        stopped = self.boundary()
        if stopped:
            return stopped
        if self.now() < self.ready_at:
            return {"state": "WAIT", "until": self.ready_at}
        if self.models >= self.max_models:
            return self.end("model_budget")
        self.models += 1
        return {"state": "ALLOW_MODEL"}

    def dispatch(self, call, execute):
        stopped = self.boundary()
        if stopped:
            return stopped
        if self.now() < self.ready_at:
            return {"state": "WAIT", "until": self.ready_at}
        try:
            name, args, write = call["tool"], call["args"], call["write"]
            if not isinstance(name, str) or not isinstance(args, dict):
                raise ValueError("invalid call")
            if type(write) is not bool:
                raise ValueError("invalid write flag")
            fingerprint = json.dumps([name, args], sort_keys=True,
                                     ensure_ascii=False, allow_nan=False)
            key = call.get("operation_key") if write else None
            if write and (not isinstance(key, str) or not key):
                raise ValueError("missing business operation key")
        except (KeyError, TypeError, ValueError):
            return self.end("invalid_call", "FAILED")

        if key in self.operations:
            record = self.operations[key]
            if record["fingerprint"] != fingerprint:
                return self.end("operation_input_conflict", "NEEDS_HUMAN")
            if record["state"] != "COMMITTED":
                return self.end("commit_unknown", "NEEDS_HUMAN")
            return {"state": "CACHED", "result": copy.deepcopy(record["result"])}
        if self.tools >= self.max_tools:
            return self.end("tool_budget")
        if self.repeats.get(fingerprint, 0) >= self.repeat_limit:
            return self.end("same_call_without_progress")
        self.tools += 1
        self.repeats[fingerprint] = self.repeats.get(fingerprint, 0) + 1
        if write:
            self.operations[key] = {"fingerprint": fingerprint, "state": "UNKNOWN"}
        try:
            result = self.verify(call, execute(call))
            if result["kind"] not in {"SUCCESS", "PENDING", "RETRYABLE", "BLOCKED", "FATAL"}:
                raise ValueError("unknown result")
            if type(result["progress"]) is not int or result["progress"] < self.progress:
                raise ValueError("invalid progress")
            if type(result["done"]) is not bool:
                raise ValueError("invalid completion")
            wait = result.get("wait", 0)
            if type(wait) not in (int, float) or not 0 <= wait <= 30:
                raise ValueError("invalid wait")
            if result["done"] and result["kind"] != "SUCCESS":
                raise ValueError("invalid completion kind")
            if write and (result["kind"] != "SUCCESS" or result.get("committed") is not True):
                return self.end("commit_unknown", "NEEDS_HUMAN")
        except Exception:
            return self.end("commit_unknown" if write else "unverified_result", "NEEDS_HUMAN")
        if write:
            self.operations[key] = {"fingerprint": fingerprint, "state": "COMMITTED",
                                    "result": copy.deepcopy(result)}
        if result["progress"] > self.progress:
            self.progress, self.stalled = result["progress"], 0
            self.repeats.clear()
        else:
            self.stalled += 1
        if result["kind"] in {"BLOCKED", "FATAL"}:
            return self.end(result["kind"].lower(), "NEEDS_HUMAN")
        if result["done"]:
            return self.end("verified_task_complete", "COMPLETED")
        if self.stalled >= self.stall_limit:
            return self.end("no_progress")
        self.ready_at = self.now() + wait
        return {"state": "CONTINUE", "result": result}

t = [0]
guard = Guard(lambda: t[0], lambda call, raw: raw)
proposals = iter([{"tool": "read", "args": {"page": 1}, "write": False}] * 4)
executed = [0]

def fake_execute(call):
    executed[0] += 1
    return {"kind": "RETRYABLE", "progress": 0, "done": False, "wait": 1}

def run():
    while True:
        gate = guard.model_gate()
        if gate["state"] == "WAIT":
            t[0] = gate["until"]
            continue
        if gate["state"] != "ALLOW_MODEL":
            return gate
        call = next(proposals, None)  # 模擬提案。モデルは呼びません。
        if call is None:
            return guard.end("no_more_proposals")
        decision = guard.dispatch(call, fake_execute)
        if decision["state"] not in {"CONTINUE", "CACHED", "WAIT"}:
            return decision

report = run()
assert report["reason"] == "same_call_without_progress"
assert executed[0] == 2
print(report)