AutoGPT 툴 호출 무한루프 끊는 Stop조건 설계

from dataclasses import dataclass
import time

@dataclass
class Budget:
    max_steps: int = 30
    max_tool_calls: int = 20
    max_wall_time_ms: int = 60_000

class BudgetGuard:
    def __init__(self, budget: Budget):
        self.budget = budget
        self.start = time.time()
        self.steps = 0
        self.tool_calls = 0

    def on_step(self):
        self.steps += 1
        if self.steps > self.budget.max_steps:
            return False, "stop: max_steps exceeded"
        if (time.time() - self.start) * 1000 > self.budget.max_wall_time_ms:
            return False, "stop: max_wall_time exceeded"
        return True, None

    def on_tool_call(self):
        self.tool_calls += 1
        if self.tool_calls > self.budget.max_tool_calls:
            return False, "stop: max_tool_calls exceeded"
        return True, None

import json
import hashlib
from collections import deque, Counter

def canonicalize(obj):
    return json.dumps(obj, sort_keys=True, ensure_ascii=False)

def call_signature(tool_name: str, args: dict) -> str:
    raw = tool_name + "|" + canonicalize(args)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

class LoopDetector:
    def __init__(self, window=10, max_same=3):
        self.window = window
        self.max_same = max_same
        self.sig_window = deque(maxlen=window)

    def observe_call(self, tool_name, args):
        sig = call_signature(tool_name, args)
        self.sig_window.append(sig)
        c = Counter(self.sig_window)
        if c[sig] >= self.max_same:
            return False, f"stop: repeated tool call signature x{c[sig]}"
        return True, None

import random
import time

TRANSIENT = {"timeout", "rate_limit", "overloaded"}
PERMANENT = {"invalid_auth", "not_found", "bad_request"}

class RetryPolicy:
    def __init__(self, max_retries=4, base_delay=0.5):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.fail_count = 0

    def on_failure(self, error_code: str):
        if error_code in PERMANENT:
            return False, f"stop: permanent error {error_code}"

        self.fail_count += 1
        if self.fail_count > self.max_retries:
            return False, f"stop: retries exceeded for {error_code}"

        if error_code in TRANSIENT:
            delay = self.base_delay * (2 ** (self.fail_count - 1))
            delay = delay * (0.5 + random.random())  # jitter
            time.sleep(delay)
            return True, f"retrying after {delay:.2f}s"

        # 알 수 없는 오류는 보수적으로 중단하거나 휴먼 핸드오프
        return False, f"stop: unknown error {error_code}"

from enum import Enum

class Phase(str, Enum):
    DISCOVER = "discover"
    PLAN = "plan"
    EXECUTE = "execute"
    VERIFY = "verify"
    DONE = "done"
    FAIL = "fail"

class ProgressTracker:
    def __init__(self):
        self.phase = Phase.DISCOVER
        self.no_progress_steps = 0
        self.max_no_progress = 5
        self.last_artifact_hash = None

    def artifact_hash(self, text: str) -> str:
        import hashlib
        return hashlib.md5(text.strip().encode("utf-8")).hexdigest()

    def observe_artifact(self, artifact_text: str):
        h = self.artifact_hash(artifact_text)
        if h == self.last_artifact_hash:
            self.no_progress_steps += 1
        else:
            self.no_progress_steps = 0
        self.last_artifact_hash = h

        if self.no_progress_steps >= self.max_no_progress:
            return False, "stop: no progress (artifact unchanged)"
        return True, None

import json

REQUIRED_KEYS = {"final_answer", "confidence", "citations"}

def validate_final(output_text: str):
    try:
        obj = json.loads(output_text)
    except Exception as e:
        return False, f"invalid_json: {e}"

    missing = REQUIRED_KEYS - set(obj.keys())
    if missing:
        return False, f"missing_keys: {sorted(list(missing))}"

    if not isinstance(obj["citations"], list):
        return False, "citations_must_be_list"

    return True, None

import uuid

def make_idempotency_key(task_id: str, tool_name: str, args: dict) -> str:
    # 같은 task_id와 같은 args면 같은 키가 나오도록 설계
    import hashlib, json
    raw = task_id + "|" + tool_name + "|" + json.dumps(args, sort_keys=True)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

# 예: 결제 툴 호출 시
# headers = {"Idempotency-Key": make_idempotency_key(task_id, "charge", args)}

class StopController:
    def __init__(self, budget_guard, loop_detector, progress_tracker, retry_policy):
        self.budget = budget_guard
        self.loop = loop_detector
        self.progress = progress_tracker
        self.retry = retry_policy

    def before_step(self):
        return self.budget.on_step()

    def before_tool(self, tool_name, args):
        ok, reason = self.budget.on_tool_call()
        if not ok:
            return ok, reason
        return self.loop.observe_call(tool_name, args)

    def after_tool_result(self, artifact_text: str):
        return self.progress.observe_artifact(artifact_text)

    def on_tool_error(self, error_code: str):
        return self.retry.on_failure(error_code)