LangChain 에이전트 무한루프·비용폭탄 9가지 차단법

from langchain.agents import AgentExecutor

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=8,  # 최저선
    early_stopping_method="force",  # 더 이상 진행 불가 시 강제 종료
    verbose=True,
)

import asyncio

async def run_agent_with_timeout(agent_executor, inputs, timeout_sec=20):
    return await asyncio.wait_for(
        agent_executor.ainvoke(inputs),
        timeout=timeout_sec,
    )

class BudgetExceeded(Exception):
    pass

class TokenBudget:
    def __init__(self, max_total_tokens: int):
        self.max_total_tokens = max_total_tokens
        self.total_tokens = 0

    def add(self, prompt_tokens: int, completion_tokens: int):
        self.total_tokens += (prompt_tokens + completion_tokens)
        if self.total_tokens > self.max_total_tokens:
            raise BudgetExceeded(f"token budget exceeded: {self.total_tokens}")

import json
import hashlib
from collections import deque

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

    def _sig(self, tool_name: str, tool_input) -> str:
        raw = json.dumps({"tool": tool_name, "input": tool_input}, sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(raw.encode("utf-8")).hexdigest()

    def check(self, tool_name: str, tool_input):
        sig = self._sig(tool_name, tool_input)
        self.recent.append(sig)
        if self.recent.count(sig) >= self.max_same:
            raise RuntimeError("repeat detected: same tool call repeated")

import time

class ToolGate:
    def __init__(self, per_tool_quota=None, cooldown_sec=0.0):
        self.per_tool_quota = per_tool_quota or {}
        self.cooldown_sec = cooldown_sec
        self.count = {}
        self.last_called_at = {}

    def allow(self, tool_name: str):
        now = time.time()
        if tool_name in self.last_called_at:
            if now - self.last_called_at[tool_name] < self.cooldown_sec:
                raise RuntimeError("tool cooldown")
        self.last_called_at[tool_name] = now

        self.count[tool_name] = self.count.get(tool_name, 0) + 1
        quota = self.per_tool_quota.get(tool_name)
        if quota is not None and self.count[tool_name] > quota:
            raise RuntimeError("tool quota exceeded")

def tool_error(retryable: bool, reason: str, hint: str = ""):
    return {
        "ok": False,
        "retryable": retryable,
        "reason": reason,
        "hint": hint,
    }

# 예: 인증 실패는 재시도 불가
return tool_error(False, "AUTH_FAILED", "API 키 또는 권한을 확인하세요")

# 예: 429는 재시도 가능
return tool_error(True, "RATE_LIMIT", "지수 백오프로 재시도하세요")

시스템 규약:
- 목표를 달성했거나 추가 정보가 필요하면 즉시 종료한다.
- 추가 정보가 필요할 때는 질문을 최대 3개만 하고, 그 외의 추측으로 진행하지 않는다.
- 툴 호출은 전체 실행 동안 최대 5회까지만 허용한다.
- 최종 답변은 반드시 FINAL 섹션에만 작성한다.

# 개념 예시: 최근 N개 메시지만 유지
class SlidingWindowMemory:
    def __init__(self, max_messages=12):
        self.max_messages = max_messages
        self.messages = []

    def add(self, role, content):
        self.messages.append({"role": role, "content": content})
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages:]

import time

class Trace:
    def __init__(self):
        self.events = []
        self.started = time.time()

    def log(self, kind: str, data: dict):
        self.events.append({
            "t": time.time(),
            "kind": kind,
            "data": data,
        })

    def summary(self):
        return {
            "elapsed_sec": round(time.time() - self.started, 3),
            "events": self.events,
        }