AutoGPT 메모리 폭주·루프 반복 막는 6가지 가드레일

import time
from dataclasses import dataclass

@dataclass
class Budget:
    max_steps: int = 20
    max_seconds: int = 60
    max_input_tokens: int = 6000
    max_output_tokens: int = 1500

class BudgetExceeded(Exception):
    pass

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

    def on_step(self, input_tokens: int, output_tokens: int):
        self.steps += 1
        if self.steps > self.budget.max_steps:
            raise BudgetExceeded("step budget exceeded")
        if time.time() - self.started_at > self.budget.max_seconds:
            raise BudgetExceeded("time budget exceeded")
        if input_tokens > self.budget.max_input_tokens:
            raise BudgetExceeded("input token budget exceeded")
        if output_tokens > self.budget.max_output_tokens:
            raise BudgetExceeded("output token budget exceeded")

from enum import Enum, auto

class State(Enum):
    PLAN = auto()
    ACT = auto()
    OBSERVE = auto()
    FINALIZE = auto()
    ABORT = auto()

ALLOWED = {
    State.PLAN: {State.ACT, State.FINALIZE, State.ABORT},
    State.ACT: {State.OBSERVE, State.ABORT},
    State.OBSERVE: {State.PLAN, State.ACT, State.FINALIZE, State.ABORT},
    State.FINALIZE: set(),
    State.ABORT: set(),
}

def transition(cur: State, nxt: State) -> State:
    if nxt not in ALLOWED[cur]:
        return State.ABORT
    return nxt

import hashlib
from collections import deque, Counter

class LoopDetected(Exception):
    pass

class ToolLoopGuard:
    def __init__(self, window=12, max_repeat=3):
        self.window = window
        self.max_repeat = max_repeat
        self.recent = deque(maxlen=window)

    def _sig(self, tool_name: str, args: dict) -> str:
        raw = tool_name + "|" + repr(sorted(args.items()))
        return hashlib.sha256(raw.encode("utf-8")).hexdigest()

    def on_tool_call(self, tool_name: str, args: dict):
        sig = self._sig(tool_name, args)
        self.recent.append(sig)
        counts = Counter(self.recent)
        if counts[sig] >= self.max_repeat:
            raise LoopDetected(f"tool call repeated: {tool_name}")

def compress_observation(obs: str, max_chars: int = 800) -> str:
    # 1차로 로컬 규칙 기반 축약(비용 0)
    obs = obs.strip().replace("\n\n", "\n")
    if len(obs) <= max_chars:
        return obs
    return obs[:max_chars] + "..."

from pydantic import BaseModel, HttpUrl, ValidationError, Field

class WebGetArgs(BaseModel):
    url: HttpUrl
    timeout_sec: int = Field(ge=1, le=20, default=10)

class ToolPolicyViolation(Exception):
    pass

def guarded_web_get(args: dict):
    try:
        parsed = WebGetArgs(**args)
    except ValidationError as e:
        raise ToolPolicyViolation(str(e))

    # 예: 내부망/메타데이터 차단 등
    if "169.254.169.254" in str(parsed.url):
        raise ToolPolicyViolation("blocked url")

    return do_http_get(str(parsed.url), timeout=parsed.timeout_sec)

import json

def log_event(event: dict):
    # 운영에서는 구조화 로그(JSON)로 쌓아야 검색이 됩니다.
    print(json.dumps(event, ensure_ascii=False))

# 예시
log_event({
    "step": 7,
    "state": "ACT",
    "tool": "web_get",
    "input_tokens": 5120,
    "output_tokens": 320,
    "memory_messages": 18,
    "reason": "ok"
})

def run_agent(llm, tools, initial_state):
    budget = BudgetGuard(Budget(max_steps=18, max_seconds=45))
    loop_guard = ToolLoopGuard(window=10, max_repeat=3)

    state = State.PLAN
    memory = []  # 단기 메모리(요약된 메시지)

    for _ in range(10_000):
        # 입력 토큰 추정은 구현에 맞게 교체
        input_tokens = estimate_tokens(memory)

        if state in (State.FINALIZE, State.ABORT):
            break

        action = llm.decide(state=state.name, memory=memory)
        # action 예: {"next_state": "ACT", "tool": "web_get", "args": {...}}

        next_state = State[action.get("next_state", "ABORT")]
        state = transition(state, next_state)

        if state == State.ACT:
            tool = action["tool"]
            args = action.get("args", {})

            loop_guard.on_tool_call(tool, args)
            result = tools[tool](args)

            obs = compress_observation(str(result))
            memory.append({"role": "observation", "content": obs})

            output_tokens = estimate_tokens([{"content": obs}])
            budget.on_step(input_tokens=input_tokens, output_tokens=output_tokens)

            state = State.OBSERVE

        elif state == State.PLAN:
            plan = action.get("plan", "")
            memory.append({"role": "plan", "content": compress_observation(plan, 500)})
            budget.on_step(input_tokens=input_tokens, output_tokens=estimate_tokens([{ "content": plan }]))

        else:
            # FINALIZE/ABORT는 상위에서 처리
            budget.on_step(input_tokens=input_tokens, output_tokens=0)

    return state.name, memory