LangChain에서 OpenAI 툴콜 무한루프 끊기

from typing import Any, Dict


def normalize_tool_result(data: Any = None, *, ok: bool = True,
                          error: str | None = None,
                          retryable: bool = False) -> Dict[str, Any]:
    return {
        "ok": ok,
        "data": data,
        "error": error,
        "retryable": retryable,
    }


def search_tool(query: str) -> Dict[str, Any]:
    try:
        # ... 실제 검색 로직
        results = [
            {"title": "...", "url": "...", "snippet": "..."}
        ]
        if not results:
            return normalize_tool_result(
                ok=True,
                data={"results": [], "note": "no_results"},
            )
        return normalize_tool_result(ok=True, data={"results": results})
    except TimeoutError:
        return normalize_tool_result(ok=False, error="timeout", retryable=True)
    except Exception as e:
        return normalize_tool_result(ok=False, error=str(e), retryable=False)

import hashlib
import json
from typing import Any, Dict, Tuple


class ToolCallDeduper:
    def __init__(self, max_calls: int = 8):
        self.max_calls = max_calls
        self.seen: dict[str, int] = {}

    def _fingerprint(self, tool_name: str, tool_args: Dict[str, Any]) -> str:
        payload = json.dumps(tool_args, sort_keys=True, ensure_ascii=False)
        raw = f"{tool_name}:{payload}".encode("utf-8")
        return hashlib.sha256(raw).hexdigest()

    def check(self, tool_name: str, tool_args: Dict[str, Any]) -> Tuple[bool, str]:
        fp = self._fingerprint(tool_name, tool_args)
        self.seen[fp] = self.seen.get(fp, 0) + 1

        # 총 호출 상한
        total = sum(self.seen.values())
        if total > self.max_calls:
            return False, "tool_call_limit_exceeded"

        # 동일 호출 반복 상한
        if self.seen[fp] >= 3:
            return False, "duplicate_tool_call_detected"

        return True, "ok"

너는 필요한 경우에만 도구를 호출한다.
- 같은 도구를 같은 인자로 2회 이상 반복 호출하지 않는다.
- 도구 결과가 ok=true 이고 data.results 가 비어 있으면 추가 검색을 하지 말고
  사용자가 선택할 수 있는 다음 액션을 제안하며 답변을 종료한다.
- 도구가 ok=false 이면 retryable=true 인 경우에만 1회 재시도하고,
  그 외에는 실패 원인과 대안을 제시하며 답변을 종료한다.

from langchain.agents import AgentExecutor

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=6,
    # 일부 버전에서는 max_execution_time 지원
    # max_execution_time=20,
    handle_parsing_errors=True,
)

from langchain_core.messages import ToolMessage


def tool_guarded_call(tool, tool_name: str, tool_args: dict, deduper: ToolCallDeduper):
    allowed, reason = deduper.check(tool_name, tool_args)
    if not allowed:
        # 모델이 이해하기 쉬운 형태로 종료 힌트를 제공
        return normalize_tool_result(
            ok=False,
            error=f"blocked:{reason}",
            retryable=False,
        )

    return tool.invoke(tool_args)

import time


def with_retry_once(fn, *, backoff_sec: float = 0.8):
    try:
        return fn()
    except TimeoutError:
        time.sleep(backoff_sec)
        return fn()


def flaky_tool(x: str) -> dict:
    def _do():
        # 외부 API 호출
        return {"value": x}

    try:
        data = with_retry_once(_do)
        return normalize_tool_result(ok=True, data=data)
    except Exception as e:
        return normalize_tool_result(ok=False, error=str(e), retryable=False)

from langchain.agents import AgentExecutor


def run_agent_with_guards(agent, tools, user_input: str):
    deduper = ToolCallDeduper(max_calls=8)

    # 여기서는 개념 예시로 tools 를 래핑했다고 가정
    guarded_tools = []
    for t in tools:
        tool_name = t.name

        def _make_guarded(tool_obj, name):
            def _invoke(args):
                return tool_guarded_call(tool_obj, name, args, deduper)
            tool_obj.invoke = _invoke
            return tool_obj

        guarded_tools.append(_make_guarded(t, tool_name))

    executor = AgentExecutor(
        agent=agent,
        tools=guarded_tools,
        verbose=True,
        max_iterations=6,
        handle_parsing_errors=True,
    )

    return executor.invoke({"input": user_input})