LangChain 도구호출 무한루프 차단·비용 절감

import time
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a careful assistant. Use tools only when necessary. Stop when you have enough information."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

# tools = [...]  # 실제 도구 리스트
agent = create_openai_tools_agent(llm, tools=[], prompt=prompt)

executor = AgentExecutor(
    agent=agent,
    tools=[],
    verbose=True,
    max_iterations=6,
    max_execution_time=15,  # seconds
    early_stopping_method="force",
)

result = executor.invoke({"input": "오늘 환율 알려줘"})
print(result["output"])

import json
import hashlib
from collections import Counter, deque

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

    def fingerprint(self, tool_name: str, tool_args: dict) -> str:
        payload = json.dumps({"tool": tool_name, "args": tool_args}, sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(payload.encode("utf-8")).hexdigest()

    def check(self, tool_name: str, tool_args: dict):
        fp = self.fingerprint(tool_name, tool_args)
        self.recent.append(fp)
        counts = Counter(self.recent)
        if counts[fp] > self.max_repeat:
            raise RuntimeError("Blocked: repeated identical tool call detected")

loop_guard = ToolLoopGuard(window=8, max_repeat=2)

# 도구 실행 래퍼 예시

def guarded_tool_call(tool, tool_name: str, tool_args: dict):
    loop_guard.check(tool_name, tool_args)
    return tool.invoke(tool_args)

import time
import random

class ToolError(Exception):
    pass


def call_with_retry(fn, *, max_attempts=3, base_sleep=0.5):
    for attempt in range(1, max_attempts + 1):
        try:
            return fn()
        except ToolError as e:
            if attempt == max_attempts:
                raise
            sleep = base_sleep * (2 ** (attempt - 1)) + random.random() * 0.1
            time.sleep(sleep)

# 예: 실제 도구 호출을 fn 으로 감싼다
# result = call_with_retry(lambda: my_tool.invoke(args))

import time

class SimpleTTLCache:
    def __init__(self, ttl_seconds=300):
        self.ttl = ttl_seconds
        self.store = {}

    def get(self, key):
        item = self.store.get(key)
        if not item:
            return None
        value, expires_at = item
        if time.time() > expires_at:
            self.store.pop(key, None)
            return None
        return value

    def set(self, key, value):
        self.store[key] = (value, time.time() + self.ttl)

cache = SimpleTTLCache(ttl_seconds=120)


def cached_tool_call(tool, tool_name: str, tool_args: dict, guard: ToolLoopGuard):
    guard.check(tool_name, tool_args)
    fp = guard.fingerprint(tool_name, tool_args)
    cached = cache.get(fp)
    if cached is not None:
        return cached
    out = tool.invoke(tool_args)
    cache.set(fp, out)
    return out

MAX_CHARS = 1500


def normalize_search_result(raw: dict) -> dict:
    # raw 가 어떤 형태든, 모델에게는 일정한 형태로 준다
    items = raw.get("items", [])
    normalized = []
    for it in items[:5]:
        normalized.append({
            "title": (it.get("title") or "")[:120],
            "url": (it.get("url") or "")[:300],
            "snippet": (it.get("snippet") or "")[:300],
        })

    payload = {"items": normalized}
    text = json.dumps(payload, ensure_ascii=False)
    if len(text) > MAX_CHARS:
        payload = {"items": normalized[:3]}
    return payload

Tool policy:
1) Call at most 2 tools per user request.
2) Never call the same tool with identical arguments more than once.
3) If a tool fails, do not retry via the model. Explain the failure and ask a clarification or suggest alternatives.
4) After receiving sufficient info, respond to the user without further tool calls.