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

import time
from dataclasses import dataclass

@dataclass
class Budget:
    max_steps: int = 30
    max_wall_time_sec: int = 120

class BudgetExceeded(Exception):
    pass

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

    def tick(self):
        self.steps += 1
        if self.steps > self.budget.max_steps:
            raise BudgetExceeded(f"max_steps exceeded: {self.steps}")
        if time.time() - self.start > self.budget.max_wall_time_sec:
            raise BudgetExceeded("max_wall_time_sec exceeded")

# loop
budget_guard = BudgetGuard(Budget(max_steps=20, max_wall_time_sec=60))
while True:
    budget_guard.tick()
    # plan -> act -> observe

from collections import deque

class ContextManager:
    def __init__(self, max_messages=12):
        self.window = deque(maxlen=max_messages)
        self.summary = ""  # long-term summary

    def add(self, role, content):
        self.window.append({"role": role, "content": content})

    def should_summarize(self):
        return len(self.window) >= self.window.maxlen

    def summarize(self, llm):
        # window 내용을 요약하고, summary에 누적한 뒤 window를 줄인다
        prompt = (
            "다음 대화를 핵심 사실/결정/남은 할일 중심으로 10줄 이내로 요약해줘.\n\n"
            f"기존 요약:\n{self.summary}\n\n"
            f"새 대화:\n{self.window}"
        )
        new_summary = llm(prompt)
        self.summary = new_summary
        self.window.clear()
        self.window.append({"role": "system", "content": f"요약 메모리:\n{self.summary}"})

    def build_messages(self):
        return list(self.window)

import hashlib
from collections import Counter, deque

class LoopDetected(Exception):
    pass

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

    def _sig(self, action: dict) -> str:
        # action 예: {"tool": "web_search", "args": {"q": "..."}}
        raw = str(action).encode("utf-8")
        return hashlib.sha256(raw).hexdigest()[:16]

    def check(self, action: dict):
        sig = self._sig(action)
        self.recent.append(sig)
        c = Counter(self.recent)
        if c[sig] >= self.max_same:
            raise LoopDetected(f"repeated action detected: sig={sig}")

repeat_guard = RepeatGuard(window=12, max_same=3)

# before executing tool
repeat_guard.check(next_action)

import time

class CircuitOpen(Exception):
    pass

class CircuitBreaker:
    def __init__(self, failure_threshold=5, cool_down_sec=30):
        self.failure_threshold = failure_threshold
        self.cool_down_sec = cool_down_sec
        self.failures = 0
        self.open_until = 0

    def before_call(self):
        if time.time() < self.open_until:
            raise CircuitOpen("circuit is open")

    def on_success(self):
        self.failures = 0

    def on_failure(self):
        self.failures += 1
        if self.failures >= self.failure_threshold:
            self.open_until = time.time() + self.cool_down_sec

breaker = CircuitBreaker()

def call_tool(tool_fn):
    breaker.before_call()
    try:
        res = tool_fn()
        breaker.on_success()
        return res
    except Exception:
        breaker.on_failure()
        raise

docker run --rm \
  --cpus=1 \
  --memory=2g \
  --pids-limit=256 \
  --read-only \
  --tmpfs /tmp:rw,size=256m \
  -e AGENT_MAX_STEPS=30 \
  my-autogpt-agent:latest

import json

def validate_final(result: dict) -> None:
    required = ["status", "deliverable"]
    for k in required:
        if k not in result:
            raise ValueError(f"missing key: {k}")
    if result["status"] not in ["done", "need_input", "failed"]:
        raise ValueError("invalid status")

# LLM이 반드시 JSON만 출력하도록 유도했다고 가정
raw = '{"status":"done","deliverable":"...","acceptance_checks":["..."],"notes":"..."}'
result = json.loads(raw)
validate_final(result)

def run_agent(llm, tools, objective):
    budget = BudgetGuard(Budget(max_steps=25, max_wall_time_sec=90))
    repeat = RepeatGuard(window=12, max_same=3)
    breaker = CircuitBreaker(failure_threshold=4, cool_down_sec=20)
    ctx = ContextManager(max_messages=10)

    ctx.add("system", "너는 안전 가드레일을 준수하는 에이전트다.")
    ctx.add("user", objective)

    while True:
        budget.tick()

        if ctx.should_summarize():
            ctx.summarize(llm)

        action = plan_next_action(llm, ctx.build_messages())  # dict
        if action.get("type") == "final":
            validate_final(action["result"])
            return action["result"]

        repeat.check(action)

        breaker.before_call()
        try:
            obs = tools[action["tool"]](**action.get("args", {}))
            breaker.on_success()
        except Exception as e:
            breaker.on_failure()
            obs = f"tool_error: {type(e).__name__}: {e}"

        ctx.add("assistant", f"action: {action}")
        ctx.add("tool", f"observation: {obs}")