AutoGPT 툴 호출 무한루프 차단 - ReAct+가드레일

from dataclasses import dataclass
import time

@dataclass
class Budget:
    max_steps: int = 12
    max_tool_calls: int = 20
    max_same_tool_calls: int = 6
    deadline_ms: int = 25_000

class BudgetExceeded(Exception):
    pass

class BudgetGuard:
    def __init__(self, budget: Budget):
        self.budget = budget
        self.started = time.time()
        self.steps = 0
        self.tool_calls = 0
        self.tool_counts = {}

    def tick_step(self):
        self.steps += 1
        self._check_deadline()
        if self.steps > self.budget.max_steps:
            raise BudgetExceeded("max_steps exceeded")

    def tick_tool(self, tool_name: str):
        self.tool_calls += 1
        self.tool_counts[tool_name] = self.tool_counts.get(tool_name, 0) + 1
        self._check_deadline()
        if self.tool_calls > self.budget.max_tool_calls:
            raise BudgetExceeded("max_tool_calls exceeded")
        if self.tool_counts[tool_name] > self.budget.max_same_tool_calls:
            raise BudgetExceeded(f"too many calls to tool: `{tool_name}`")

    def _check_deadline(self):
        elapsed_ms = int((time.time() - self.started) * 1000)
        if elapsed_ms > self.budget.deadline_ms:
            raise BudgetExceeded("deadline exceeded")

from enum import Enum

class State(str, Enum):
    PLAN = "plan"
    FETCH = "fetch"
    SYNTH = "synth"
    VERIFY = "verify"
    DONE = "done"

ALLOWED_TOOLS = {
    State.PLAN: set(),
    State.FETCH: {"web_search", "http_get"},
    State.SYNTH: set(),
    State.VERIFY: {"verifier"},
    State.DONE: set(),
}

def ensure_tool_allowed(state: State, tool_name: str):
    if tool_name not in ALLOWED_TOOLS[state]:
        raise ValueError(f"tool `{tool_name}` not allowed in state `{state}`")

import json
import hashlib

def fingerprint(tool_name: str, args: dict) -> str:
    normalized = json.dumps(args, sort_keys=True, ensure_ascii=False)
    raw = f"{tool_name}:{normalized}".encode("utf-8")
    return hashlib.sha256(raw).hexdigest()

class DedupGuard:
    def __init__(self, window: int = 10):
        self.window = window
        self.recent = []
        self.set_recent = set()

    def check_and_remember(self, fp: str):
        if fp in self.set_recent:
            raise RuntimeError("duplicate tool call detected")
        self.recent.append(fp)
        self.set_recent.add(fp)
        if len(self.recent) > self.window:
            old = self.recent.pop(0)
            self.set_recent.remove(old)

역할: 너는 에이전트 실행 결과를 검증하는 검증자다.
입력: goal, current_answer, evidence_summaries, tool_call_stats
출력: 아래 JSON만 출력
- decision: "done" | "need_more"
- reason: 2문장 이내
- next_action: 필요 시 다음에 할 1가지 액션(툴명과 핵심 인자)
체크리스트:
1) goal의 필수 산출물이 모두 포함되었나
2) 근거가 최소 2개 이상이며 서로 독립적인가
3) 최근 5회 툴 호출이 중복 또는 무진전이었나
4) 비용/시간 예산을 초과할 위험이 있나

def run_agent(goal: str, llm, tools: dict):
    budget = BudgetGuard(Budget())
    dedup = DedupGuard(window=12)
    state = State.PLAN

    context = {"goal": goal, "notes": [], "evidence": []}

    while state != State.DONE:
        budget.tick_step()

        if state == State.PLAN:
            plan = llm.plan(goal)
            context["notes"].append(plan)
            state = State.FETCH
            continue

        if state == State.FETCH:
            action = llm.next_tool_action(context)  # {tool_name, args}
            tool_name, args = action["tool_name"], action["args"]

            ensure_tool_allowed(state, tool_name)
            budget.tick_tool(tool_name)

            fp = fingerprint(tool_name, args)
            dedup.check_and_remember(fp)

            result = tools[tool_name](**args)
            classified = classify_result(result)  # RETRYABLE/RATE_LIMITED/FATAL/OK/NO_PROGRESS

            if classified.kind == "FATAL":
                return {"status": "failed", "error": classified.message}

            if classified.kind in {"RETRYABLE", "RATE_LIMITED"}:
                handle_backoff(classified)
                continue

            if classified.kind == "NO_PROGRESS":
                state = State.VERIFY
                continue

            context["evidence"].append(classified.data)

            if len(context["evidence"]) >= 3:
                state = State.SYNTH
            continue

        if state == State.SYNTH:
            answer = llm.summarize(context)
            context["current_answer"] = answer
            state = State.VERIFY
            continue

        if state == State.VERIFY:
            verdict = tools["verifier"](
                goal=context["goal"],
                current_answer=context.get("current_answer", ""),
                evidence_summaries=context["evidence"],
                tool_call_stats=budget.tool_counts,
            )
            if verdict["decision"] == "done":
                state = State.DONE
            else:
                # 검증자가 제안한 다음 액션을 제한적으로 반영
                state = State.FETCH
            continue

    return {"status": "ok", "answer": context.get("current_answer", "")}