LangChain 에이전트 무한 루프 끊는 실전 디버깅

from __future__ import annotations

import json
import time
from typing import Any, Dict, List, Optional

from langchain_core.callbacks import BaseCallbackHandler


def _safe_preview(obj: Any, limit: int = 400) -> str:
    try:
        s = obj if isinstance(obj, str) else json.dumps(obj, ensure_ascii=False)
    except Exception:
        s = repr(obj)
    return s[:limit]


class AgentTraceCallback(BaseCallbackHandler):
    def __init__(self) -> None:
        self.events: List[Dict[str, Any]] = []

    def on_tool_start(
        self,
        serialized: Dict[str, Any],
        input_str: str,
        **kwargs: Any,
    ) -> None:
        self.events.append({
            "ts": time.time(),
            "type": "tool_start",
            "tool": serialized.get("name"),
            "input": _safe_preview(input_str),
        })

    def on_tool_end(self, output: Any, **kwargs: Any) -> None:
        self.events.append({
            "ts": time.time(),
            "type": "tool_end",
            "output": _safe_preview(output),
        })

    def on_agent_finish(self, finish: Any, **kwargs: Any) -> None:
        self.events.append({
            "ts": time.time(),
            "type": "agent_finish",
            "output": _safe_preview(getattr(finish, "return_values", finish)),
        })

    def dump(self) -> str:
        return json.dumps(self.events, ensure_ascii=False, indent=2)

# 예시: 에이전트 실행기 생성 시 반복 제한과 early stopping 설정
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=8,
    early_stopping_method="generate",
    verbose=False,
)

import asyncio

async def run_with_timeout(coro, timeout_s: float):
    return await asyncio.wait_for(coro, timeout=timeout_s)

# 사용 예
# result = await run_with_timeout(agent_executor.ainvoke({"input": q}), 20.0)

from collections import deque
from dataclasses import dataclass
from typing import Callable, Deque, Tuple


@dataclass
class LoopGuard:
    window: int = 4
    max_same: int = 3

    def __post_init__(self):
        self.recent: Deque[Tuple[str, str]] = deque(maxlen=self.window)

    def check(self, tool_name: str, tool_input: str) -> None:
        key = (tool_name, tool_input)
        self.recent.append(key)
        same = sum(1 for x in self.recent if x == key)
        if same >= self.max_same:
            raise RuntimeError(
                "Detected repeated tool call loop: "
                f"tool={tool_name}, count_in_window={same}"
            )


def wrap_tool(tool_name: str, tool_fn: Callable[[str], str], guard: LoopGuard):
    def _wrapped(inp: str) -> str:
        guard.check(tool_name, inp)
        return tool_fn(inp)
    return _wrapped

def search_tool(query: str) -> str:
    docs = do_search(query)
    if not docs:
        return json.dumps({
            "ok": False,
            "reason": "no_results",
            "hint": "Try broader keywords or ask user for constraints",
        }, ensure_ascii=False)

    return json.dumps({
        "ok": True,
        "count": len(docs),
        "top": docs[:3],
    }, ensure_ascii=False)

- You must not call the same tool with the same input more than 2 times.
- If observations do not change after retries, stop and ask the user for missing info.
- If you cannot complete within 6 tool calls, provide the best partial answer and list next questions.

import random
import time


def call_api_with_backoff(fn, max_tries: int = 4, base: float = 0.5):
    for i in range(max_tries):
        try:
            return fn()
        except Exception as e:
            # 마지막이면 구조화된 실패를 던지거나 반환
            if i == max_tries - 1:
                raise
            sleep_s = base * (2 ** i) + random.random() * 0.2
            time.sleep(sleep_s)

def safe_http_tool(url: str) -> str:
    try:
        return fetch(url)
    except Exception as e:
        return json.dumps({
            "ok": False,
            "reason": "network_error",
            "error": type(e).__name__,
            "hint": "Do not retry immediately. Ask user to confirm URL or try later.",
        }, ensure_ascii=False)