LangChain Agent 무한루프·비용폭탄 차단 7가지

import asyncio
from langchain.agents import AgentExecutor

# agent + tools + llm은 이미 구성되어 있다고 가정
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=8,          # 논리적 상한
    return_intermediate_steps=True,
)

async def run_with_timeout(user_input: str):
    # 물리적 상한(예: 20초)
    return await asyncio.wait_for(
        executor.ainvoke({"input": user_input}),
        timeout=20,
    )

result = asyncio.run(run_with_timeout("지난달 매출 추이 분석해줘"))
print(result["output"])

from langchain.callbacks.base import BaseCallbackHandler

class BudgetGuard(BaseCallbackHandler):
    def __init__(self, max_total_tokens: int, max_tool_calls: int):
        self.max_total_tokens = max_total_tokens
        self.max_tool_calls = max_tool_calls
        self.total_tokens = 0
        self.tool_calls = 0

    def on_llm_end(self, response, **kwargs):
        # 프로바이더에 따라 token usage 접근 방식이 다릅니다.
        usage = getattr(response, "llm_output", {}) or {}
        token_usage = usage.get("token_usage") or {}
        self.total_tokens += int(token_usage.get("total_tokens") or 0)
        if self.total_tokens >= self.max_total_tokens:
            raise RuntimeError("Budget exceeded: total token limit")

    def on_tool_start(self, serialized, input_str, **kwargs):
        self.tool_calls += 1
        if self.tool_calls >= self.max_tool_calls:
            raise RuntimeError("Budget exceeded: tool call limit")

# executor 생성 시 callbacks에 주입
budget_guard = BudgetGuard(max_total_tokens=6000, max_tool_calls=6)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[budget_guard],
    max_iterations=10,
)

import hashlib
from langchain.callbacks.base import BaseCallbackHandler

class LoopDetector(BaseCallbackHandler):
    def __init__(self, max_same_action: int = 3):
        self.max_same_action = max_same_action
        self.counts = {}

    def on_tool_start(self, serialized, input_str, **kwargs):
        tool_name = serialized.get("name")
        key_raw = f"{tool_name}:{input_str.strip()}"
        key = hashlib.sha256(key_raw.encode("utf-8")).hexdigest()
        self.counts[key] = self.counts.get(key, 0) + 1
        if self.counts[key] >= self.max_same_action:
            raise RuntimeError("Loop detected: repeated tool call")

loop_detector = LoopDetector(max_same_action=3)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[loop_detector],
    max_iterations=12,
)

from pydantic import BaseModel, Field
from langchain.tools import tool

class SearchInput(BaseModel):
    query: str = Field(..., min_length=2, max_length=120)
    top_k: int = Field(5, ge=1, le=5)

@tool(args_schema=SearchInput)
def web_search(query: str, top_k: int = 5) -> str:
    # 반드시 상위 top_k만, 그리고 각 결과는 짧게
    results = my_search_impl(query, top_k=top_k)
    compact = []
    for r in results:
        compact.append(f"- {r['title']} | {r['url']} | {r['snippet'][:160]}")
    return "\n".join(compact)

규칙:
1) 툴 호출은 최대 4회까지만 한다.
2) 동일한 툴을 동일 목적의 입력으로 2회 이상 반복 호출하지 않는다.
3) 4회 안에 답이 불가능하면, 필요한 추가 정보 질문 1개만 하고 종료한다.
4) 최종 답변에는 (a) 결론 (b) 근거 (c) 불확실한 점 을 포함한다.

import time
import random

def backoff_sleep(attempt: int):
    base = min(2 ** attempt, 30)
    jitter = random.uniform(0, 1)
    time.sleep(base + jitter)

def call_llm_with_retry(prompt: str, max_retry: int = 3):
    for attempt in range(max_retry + 1):
        try:
            return llm.invoke(prompt)
        except Exception as e:
            # 여기서 429/timeout만 선별하는 것이 중요
            if attempt == max_retry:
                raise
            backoff_sleep(attempt)

import json
import time

start = time.time()
res = executor.invoke({"input": "경쟁사 가격 정책 요약"})

duration_ms = int((time.time() - start) * 1000)
log = {
    "trace_id": "req_123",
    "duration_ms": duration_ms,
    "output": res.get("output"),
    "intermediate_steps": [
        {
            "action": str(step[0]),
            "observation_len": len(str(step[1])) if step[1] is not None else 0,
        }
        for step in (res.get("intermediate_steps") or [])
    ],
}
print(json.dumps(log, ensure_ascii=False))