beautifulsoups.dev
Published on

LangChain 에이전트 무한루프·툴콜 폭주 차단법

Authors

서버에 올린 LangChain 에이전트가 어느 순간부터 끝나지 않거나, 같은 툴을 초당 수십 번 호출하는 상황을 겪으면 비용과 레이턴시가 동시에 터집니다. 특히 검색·브라우징·DB 조회 같은 툴이 붙어 있을수록 에이전트는 “조금만 더 확인”을 반복하며 종료 조건을 놓치기 쉽습니다.

이 글은 무한루프와 툴콜 폭주를 원인별로 분해하고, LangChain에서 실제로 걸어둘 수 있는 차단 장치를 코드 중심으로 정리합니다. 스트리밍에서 중복 응답이나 토큰 폭주까지 함께 나타나는 케이스는 아래 글도 같이 보면 디버깅 시간이 크게 줄어듭니다.

왜 무한루프가 생기나: 5가지 전형 패턴

1) 종료 조건이 모델에게만 맡겨져 있음

에이전트가 Final Answer 를 내야 종료되는데, 프롬프트가 애매하면 모델은 “추가 조사”를 선택합니다. 특히 “정확히 확인해” 같은 지시가 강하면 툴을 계속 부릅니다.

2) 툴 결과가 비결정적이거나 노이즈가 큼

검색 결과가 매 호출마다 달라지거나, 페이지가 랜덤으로 바뀌거나, 시간 의존 데이터면 “방금 결과가 이상하니 다시”가 반복됩니다.

3) 툴 에러가 재시도로 변환됨

툴이 429 나 타임아웃을 내면 모델은 “잠시 후 다시 시도”를 선택합니다. 재시도 정책이 없다면 모델이 직접 재시도를 구현해 버립니다.

4) 상태가 누적되지 않음

이전 단계에서 무엇을 확인했고 무엇이 확정인지가 메모리나 상태에 남지 않으면, 매번 같은 질문을 같은 툴로 확인합니다.

5) 관찰값이 너무 길어 컨텍스트가 오염됨

관찰(툴 결과)을 그대로 대량 주입하면, 중요한 단서가 묻혀 모델이 결론 대신 “다시 검색”으로 회귀합니다.

1차 방어선: 스텝 제한과 예산(토큰·시간·툴콜)

가장 먼저 해야 할 일은 “에이전트가 망가져도 시스템은 멈춘다”를 보장하는 것입니다. 즉, 하드 리밋이 필요합니다.

실행 스텝 제한

LangChain의 에이전트 실행에는 보통 max_iterations 같은 제한이 있습니다. 버전에 따라 이름이 다를 수 있으니, 사용하는 러너의 옵션을 확인하세요.

# Python 예시 (개념 코드)
from langchain.agents import AgentExecutor

executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=8,          # 스텝 제한
    max_execution_time=20,     # 초 단위 시간 제한(지원되는 경우)
    verbose=True,
)

result = executor.invoke({"input": "..."})

툴콜 횟수 예산과 툴별 쿼터

스텝 제한만으로는 부족합니다. 한 스텝에서 툴을 여러 번 호출하거나, 특정 툴만 반복하는 “편향 폭주”가 있기 때문입니다. 그래서 툴별 쿼터가 효과적입니다.

import time
from collections import defaultdict

class ToolBudget:
    def __init__(self, total_calls=10, per_tool_calls=4, wall_clock_s=20):
        self.total_calls = total_calls
        self.per_tool_calls = per_tool_calls
        self.wall_clock_s = wall_clock_s
        self.started = time.time()
        self.total = 0
        self.per_tool = defaultdict(int)

    def check(self, tool_name: str):
        if time.time() - self.started > self.wall_clock_s:
            raise RuntimeError("Tool budget exceeded: wall clock limit")
        if self.total >= self.total_calls:
            raise RuntimeError("Tool budget exceeded: total tool calls")
        if self.per_tool[tool_name] >= self.per_tool_calls:
            raise RuntimeError(f"Tool budget exceeded: {tool_name}")

    def consume(self, tool_name: str):
        self.total += 1
        self.per_tool[tool_name] += 1

budget = ToolBudget(total_calls=12, per_tool_calls=5, wall_clock_s=25)

# 툴 래퍼
async def guarded_tool_call(tool, tool_name: str, *args, **kwargs):
    budget.check(tool_name)
    budget.consume(tool_name)
    return await tool.ainvoke(*args, **kwargs)

이렇게 예산을 코드로 강제하면, 모델이 아무리 “다시 해보자”를 말해도 시스템이 차단합니다.

2차 방어선: 멱등성(중복 툴콜 제거)과 캐시

무한루프의 상당수는 동일 입력으로 동일 툴을 반복 호출하는 형태입니다. 이때는 멱등 키를 만들어 “같은 요청은 한 번만” 처리하면 폭주가 즉시 줄어듭니다.

툴 입력 해시로 멱등 키 만들기

import json
import hashlib

class IdempotencyCache:
    def __init__(self):
        self.store = {}

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

    def get(self, k: str):
        return self.store.get(k)

    def set(self, k: str, v):
        self.store[k] = v

cache = IdempotencyCache()

async def cached_tool_call(tool, tool_name: str, tool_input: dict):
    k = cache.key(tool_name, tool_input)
    hit = cache.get(k)
    if hit is not None:
        return hit
    out = await tool.ainvoke(tool_input)
    cache.set(k, out)
    return out

검색 툴, RAG retriever, DB 조회처럼 “같은 쿼리면 같은 결과”인 툴에 특히 유효합니다. 멱등성 설계 관점은 이벤트 처리에서도 동일하게 중요합니다.

캐시가 오히려 루프를 강화하는 경우

캐시가 “빈 결과”를 계속 돌려주면 에이전트는 “결과가 없으니 다시 검색”을 반복할 수 있습니다. 이때는 캐시 자체보다 종료 조건을 강화해야 합니다.

  • “N회 검색했는데도 근거가 없으면 불가능으로 결론” 같은 정책
  • “빈 결과면 쿼리 확장 1회까지만 허용” 같은 룰

3차 방어선: 상태 머신으로 에이전트의 선택지를 줄이기

에이전트가 자유롭게 툴을 고르면 편하지만, 운영 단계에서는 상태 머신이 훨씬 안전합니다.

예를 들어 “요약 생성” 에이전트라면 상태를 다음처럼 제한합니다.

  1. PLAN : 필요한 정보 항목을 최대 3개로 계획
  2. FETCH : 각 항목당 툴 호출 1회
  3. SYNTHESIS : 툴 호출 금지, 답변 생성만
  4. FINAL : 종료

이렇게 하면 SYNTHESIS 단계에서 툴을 부르는 순간 바로 차단할 수 있습니다.

from enum import Enum

class Phase(str, Enum):
    PLAN = "plan"
    FETCH = "fetch"
    SYNTHESIS = "synthesis"
    FINAL = "final"

class PhaseGuard:
    def __init__(self):
        self.phase = Phase.PLAN
        self.fetch_calls = 0

    def allow_tool(self) -> bool:
        return self.phase == Phase.FETCH

    def next(self):
        if self.phase == Phase.PLAN:
            self.phase = Phase.FETCH
        elif self.phase == Phase.FETCH:
            self.phase = Phase.SYNTHESIS
        elif self.phase == Phase.SYNTHESIS:
            self.phase = Phase.FINAL

pg = PhaseGuard()

def assert_tool_allowed():
    if not pg.allow_tool():
        raise RuntimeError("Tool call blocked: not in FETCH phase")

핵심은 “모델이 알아서 잘하겠지”가 아니라, 툴 호출 가능 구간을 시스템이 강제하는 것입니다.

4차 방어선: 서킷 브레이커와 백오프(특히 429, 타임아웃)

툴콜 폭주는 종종 외부 API의 429 또는 네트워크 타임아웃에서 시작됩니다. 이때 모델은 재시도를 학습적으로 선택할 수 있으므로, 재시도를 모델에게 맡기지 말고 서킷 브레이커를 두세요.

import time

class CircuitBreaker:
    def __init__(self, fail_threshold=3, cool_down_s=15):
        self.fail_threshold = fail_threshold
        self.cool_down_s = cool_down_s
        self.fail_count = 0
        self.opened_at = None

    def before(self):
        if self.opened_at is None:
            return
        if time.time() - self.opened_at < self.cool_down_s:
            raise RuntimeError("Circuit open: tool temporarily disabled")
        # half-open
        self.opened_at = None
        self.fail_count = 0

    def success(self):
        self.fail_count = 0

    def fail(self):
        self.fail_count += 1
        if self.fail_count >= self.fail_threshold:
            self.opened_at = time.time()

cb = CircuitBreaker(fail_threshold=2, cool_down_s=10)

async def resilient_tool_call(tool, tool_input: dict):
    cb.before()
    try:
        out = await tool.ainvoke(tool_input)
        cb.success()
        return out
    except Exception:
        cb.fail()
        raise

이 패턴은 에이전트뿐 아니라 RPC 전반에서 동일하게 통합니다. 데드라인과 재시도 설계는 아래 글도 참고할 만합니다.

프롬프트 레벨 가드레일: “더 조사”를 정책으로 제한

코드 가드레일이 1순위지만, 프롬프트도 같이 정리해야 루프 확률이 낮아집니다.

필수 포함 문구 예시

  • “툴 호출은 최대 N 번까지만 허용된다. 초과하면 현재 정보로 결론을 내린다.”
  • “같은 툴을 같은 입력으로 두 번 호출하지 않는다.”
  • “근거가 부족하면 부족하다고 명시하고, 추가로 필요한 정보 2개만 제시한 뒤 종료한다.”

안티패턴

  • “확실해질 때까지 반복해서 검증해라”
  • “가능한 모든 자료를 찾아라”

운영 환경에서는 “완벽”보다 “예측 가능한 종료”가 더 중요합니다.

관찰값(툴 결과) 다이어트: 컨텍스트 오염 방지

툴 결과를 그대로 넣으면 모델이 핵심을 놓치고 다시 툴을 부르는 루프가 자주 생깁니다. 다음 중 하나를 적용하세요.

  1. 툴 결과를 1차 요약해서 주입
  2. 상위 K 개만 주입
  3. 구조화된 필드만 추출해 주입
def compress_observation(raw: str, limit_chars: int = 1200) -> str:
    raw = raw.strip()
    if len(raw) <= limit_chars:
        return raw
    return raw[:limit_chars] + "\n[truncated]"

또는 아예 툴 자체가 title, url, snippet 같은 스키마를 반환하도록 고정하면 효과가 큽니다.

운영 체크리스트: 폭주를 “탐지”하고 “격리”하기

차단 장치만으로는 부족합니다. 폭주가 발생했을 때 원인을 빨리 찾으려면 관측 가능성이 필요합니다.

필수 로깅

  • trace_id 단위로 스텝 수, 툴명, 툴 입력 해시, 응답 크기
  • 툴 에러 코드(429, 타임아웃 등)와 재시도 여부
  • 종료 사유: 정상 종료, 스텝 초과, 예산 초과, 서킷 오픈

격리 전략

  • 특정 사용자 세션에서만 폭주하면 세션별 레이트 리밋
  • 특정 툴에서만 폭주하면 툴 단위로 강제 쿨다운
  • 모델 변경이나 프롬프트 변경 직후면 점진 롤아웃

웹 앱에서 렌더링 폭주를 캐시와 재검증 정책으로 제어하듯, 에이전트도 “실행 폭주”를 정책으로 제어해야 합니다.

최소 구현 템플릿: 예산 + 멱등 + 상태 + 서킷 브레이커

아래는 실무에서 바로 붙이기 쉬운 조합입니다.

  • 스텝 제한: max_iterations
  • 툴 예산: 총량 및 툴별 쿼터
  • 멱등 캐시: 동일 입력 중복 제거
  • 상태 머신: 툴 허용 구간 제한
  • 서킷 브레이커: 실패 연쇄 차단

이 5가지를 넣으면 “모델이 이상해져도 시스템이 안전하게 종료”하는 형태가 됩니다.

마무리

LangChain 에이전트의 무한루프와 툴콜 폭주는 모델의 문제가 아니라 제어면(control plane) 이 없는 실행 구조에서 발생하는 경우가 많습니다. 운영 환경에서는 다음 우선순위로 적용하는 것이 안전합니다.

  1. 하드 리밋: 스텝, 시간, 툴콜 예산
  2. 툴 안전장치: 멱등성, 캐시, 툴별 쿼터
  3. 실패 제어: 서킷 브레이커, 백오프, 데드라인
  4. 구조 제어: 상태 머신으로 툴 호출 구간 제한
  5. 관측 가능성: 종료 사유와 툴 호출 패턴 로깅

이 순서대로 적용하면 “가끔 똑똑하지만 가끔 폭주하는 에이전트”를 “항상 예측 가능하게 종료하는 에이전트”로 바꿀 수 있습니다.

Discuss on Twitter • View on GitHub