LangChain 에이전트 무한루프·툴난사 차단법

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

agent = create_openai_tools_agent(
    llm=llm,
    tools=[],
    prompt=None,  # 예시 단순화
)

executor = AgentExecutor(
    agent=agent,
    tools=[],
    verbose=True,
    max_iterations=8,
    handle_parsing_errors=True,
)

result = executor.invoke({"input": "요약해줘"})
print(result)

import time
from langchain.agents import AgentExecutor

class TimeoutAgentExecutor(AgentExecutor):
    def __init__(self, *args, max_wall_time_s=20, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_wall_time_s = max_wall_time_s

    def invoke(self, input, config=None, **kwargs):
        start = time.time()
        while True:
            if time.time() - start > self.max_wall_time_s:
                return {
                    "output": "시간 예산을 초과해 중단했습니다. 필요한 정보를 더 구체적으로 알려주세요.",
                    "stopped": True,
                }
            return super().invoke(input, config=config, **kwargs)

import json
import hashlib
import time
from collections import deque
from langchain.tools import BaseTool

class AntiLoopTool(BaseTool):
    def __init__(
        self,
        tool: BaseTool,
        window=6,
        max_same_calls=2,
        cooldown_s=0,
    ):
        super().__init__(name=tool.name, description=tool.description)
        self._tool = tool
        self._recent = deque(maxlen=window)
        self._max_same_calls = max_same_calls
        self._cooldown_s = cooldown_s
        self._last_call_ts = 0.0

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

    def _run(self, tool_input, **kwargs):
        now = time.time()
        if self._cooldown_s and (now - self._last_call_ts) < self._cooldown_s:
            return "툴 호출 쿨다운 중입니다. 다른 접근을 시도하세요."

        sig = self._sig(tool_input)
        same = sum(1 for s in self._recent if s == sig)
        if same >= self._max_same_calls:
            return (
                "동일한 툴 호출이 반복되어 차단했습니다. "
                "입력을 바꾸거나, 다른 툴/추론으로 전환하세요."
            )

        self._recent.append(sig)
        self._last_call_ts = now
        return self._tool.run(tool_input, **kwargs)

from collections import defaultdict
from langchain.tools import BaseTool

class QuotaTool(BaseTool):
    def __init__(self, tool: BaseTool, limits: dict, counter: dict):
        super().__init__(name=tool.name, description=tool.description)
        self._tool = tool
        self._limits = limits
        self._counter = counter

    def _run(self, tool_input, **kwargs):
        self._counter[self.name] += 1
        limit = self._limits.get(self.name)
        if limit is not None and self._counter[self.name] > limit:
            return (
                f"툴 `{self.name}` 호출 한도({limit})를 초과했습니다. "
                "현재 정보로 결론을 내리거나 사용자에게 추가 질문을 하세요."
            )
        return self._tool.run(tool_input, **kwargs)

# 사용 예
limits = {"web_search": 2, "sql_query": 3}
counts = defaultdict(int)

# tools = [QuotaTool(t, limits, counts) for t in tools]

- You have a limited tool budget.
- Do not call the same tool with the same arguments more than once.
- If you cannot make progress after two tool calls, ask a clarifying question and stop.
- Prefer answering with available information; only use tools when necessary.

from langchain.callbacks.base import BaseCallbackHandler

class ToolAuditCallback(BaseCallbackHandler):
    def __init__(self):
        self.events = []

    def on_tool_start(self, serialized, input_str, **kwargs):
        name = serialized.get("name") if isinstance(serialized, dict) else str(serialized)
        self.events.append({"event": "tool_start", "tool": name, "input": input_str})

    def on_tool_end(self, output, **kwargs):
        self.events.append({"event": "tool_end", "output_preview": str(output)[:200]})

# executor.invoke(..., config={"callbacks": [ToolAuditCallback()]})

from collections import defaultdict

counts = defaultdict(int)
limits = {"web_search": 2, "sql_query": 3}

wrapped_tools = []
for t in tools:
    t1 = AntiLoopTool(t, window=6, max_same_calls=2, cooldown_s=0)
    t2 = QuotaTool(t1, limits=limits, counter=counts)
    wrapped_tools.append(t2)

executor = AgentExecutor(
    agent=agent,
    tools=wrapped_tools,
    max_iterations=10,
    verbose=True,
)