beautifulsoups.dev
Published on

LangChain Tool Calling 무한루프 차단 7단계

Authors

서론: 왜 Tool Calling은 쉽게 무한루프에 빠질까

LangChain의 tool calling은 모델이 외부 도구를 호출해 정보를 보강하고 최종 답을 만드는 강력한 방식입니다. 하지만 운영 환경에서는 다음과 같은 형태의 무한루프가 자주 발생합니다.

  • 모델이 같은 도구를 반복 호출하며 결론을 내리지 못함
  • 도구 에러를 복구하려다 재시도만 계속함
  • 도구 결과가 애매하거나 스키마가 불명확해 모델이 추가 호출을 유도함
  • 상태 저장이 꼬여서 이미 처리한 요청을 다시 처리함

이 글은 원인 분석보다 한 단계 더 나아가, “어떤 가드를 어디에 두면 루프를 구조적으로 차단할 수 있는지”를 7단계 체크리스트로 제공합니다. 무한루프는 애플리케이션 레벨에서 보면 일종의 “재시도 폭주”이기도 하므로, 다른 영역의 루프 문제 해결 접근도 참고가 됩니다. 예를 들어 nginx HTTPS 강제 리다이렉트 무한루프 해결 글에서처럼, 루프는 보통 “조건이 충족되지 않는데도 같은 경로로 되돌아가는 구조”에서 생깁니다.

1단계: 런타임 하드 리밋 설정 (max iterations, timeout)

가장 먼저 해야 할 일은 “절대 무한히 돌지 못하게” 상한선을 거는 것입니다. 이 단계는 근본 원인 해결이 아니라 안전장치지만, 운영 장애를 막는 데는 필수입니다.

핵심은 두 가지입니다.

  • 반복 횟수 제한: 에이전트 루프의 최대 tool 호출 횟수
  • 시간 제한: 전체 요청 처리 시간의 데드라인

아래는 LangChain의 에이전트 실행에 반복 제한을 거는 예시입니다. 버전마다 API가 조금씩 다를 수 있으니, 개념은 max_iterations 와 실행 타임아웃을 함께 거는 것으로 이해하면 됩니다.

import asyncio

async def run_with_timeout(agent_executor, inputs, timeout_s=20):
    return await asyncio.wait_for(agent_executor.ainvoke(inputs), timeout=timeout_s)

# agent_executor 구성 시 반복 제한을 둔다 (예: max_iterations=6)
# agent_executor = AgentExecutor(..., max_iterations=6)

result = asyncio.run(run_with_timeout(agent_executor, {"input": "..."}, timeout_s=20))
print(result)

운영 팁

  • 반복 제한은 너무 낮게 잡으면 정상 요청도 중간에 끊깁니다. 먼저 관측값을 모은 뒤, 평균 호출 횟수의 상위 백분위 기준으로 설정하세요.
  • 시간 제한은 upstream 타임아웃과 정렬해야 합니다. 예를 들어 API Gateway가 29초 제한이면 애플리케이션은 25초 내로 끊는 식입니다.

2단계: 도구 스키마를 “모델이 오해할 수 없게” 좁히기

무한루프의 흔한 원인은 도구 입력 스키마가 넓고 모호해서, 모델이 계속 “다른 파라미터로 다시 호출해보자”라고 판단하는 것입니다.

다음 원칙을 적용하면 루프 확률이 급격히 줄어듭니다.

  • 필수 입력은 필수로, 선택 입력은 최소화
  • enum, minLength, pattern 등 제약을 적극 사용
  • 한 도구가 너무 많은 일을 하지 않게 쪼개기
  • 에러 케이스를 스키마로 표현하기보다, 도구 결과에 명시적인 상태 필드를 포함

예시로, 검색 도구에 query 만 받게 하고, 범위나 정렬 같은 옵션을 마구 열어두지 않는 편이 모델의 탐색 폭주를 줄입니다.

from pydantic import BaseModel, Field
from typing import Literal, Optional

class SearchInput(BaseModel):
    query: str = Field(..., min_length=2, description="검색 질의")
    top_k: int = Field(5, ge=1, le=10, description="최대 결과 수")
    mode: Literal["web", "docs"] = Field("docs", description="검색 대상")

class SearchOutput(BaseModel):
    status: Literal["ok", "no_result", "error"]
    items: list[dict]
    message: Optional[str] = None

도구 결과에 status 를 넣는 이유는, 모델이 “결과가 없다”를 오류로 착각하고 재시도 루프에 들어가는 것을 막기 위해서입니다.

3단계: 도구 실패를 “재시도”가 아니라 “분기”로 설계하기

도구가 실패했을 때 모델이 같은 호출을 반복하는 패턴은, 시스템이 실패를 복구할 다른 경로를 제공하지 않아서 생깁니다.

여기서 중요한 것은 “재시도 정책을 모델에게 맡기지 말고, 애플리케이션이 소유”하는 것입니다.

  • HTTP 429 나 타임아웃은 지수 백오프 후 제한된 횟수만 재시도
  • 입력 오류 400 계열은 재시도 금지, 즉시 사용자에게 필요한 입력을 요청
  • 인증 오류 401 은 세션 갱신 흐름으로 분기

코드 레벨에서는 도구 함수 내부에서 예외를 그대로 던지기보다, 예외를 분류해 구조화된 결과로 반환하는 편이 루프를 줄입니다.

import time
import random

def safe_call_tool(tool_fn, *args, **kwargs):
    max_retry = 2
    for attempt in range(max_retry + 1):
        try:
            data = tool_fn(*args, **kwargs)
            return {"status": "ok", "data": data}
        except TimeoutError:
            if attempt == max_retry:
                return {"status": "error", "error_type": "timeout", "message": "tool timeout"}
            time.sleep((2 ** attempt) + random.random())
        except ValueError as e:
            # 입력 오류는 재시도하지 않는다
            return {"status": "error", "error_type": "bad_request", "message": str(e)}
        except Exception as e:
            return {"status": "error", "error_type": "unknown", "message": str(e)}

이렇게 하면 모델은 “같은 도구를 다시 호출”하기보다, 반환된 error_type 에 따라 다음 행동을 선택할 여지가 생깁니다.

4단계: 동일 호출 중복 방지 (idempotency key, 캐시, 디듀프)

무한루프의 본질이 “같은 입력으로 같은 도구를 계속 호출”하는 것이라면, 가장 확실한 차단법은 중복 호출을 시스템이 감지해 끊는 것입니다.

추천 패턴

  • tool name + normalized args 를 해시해서 idempotency key 생성
  • 동일 키가 일정 시간 내 재등장하면 결과를 캐시 반환하거나 차단
  • 분산 환경이면 Redis 같은 외부 저장소로 디듀프

이 접근은 분산 트랜잭션에서 중복 실행을 막는 패턴과도 유사합니다. 보다 일반적인 디듀프 사고방식은 Saga 보상 트랜잭션 중복 실행 방지 패턴 도 함께 참고할 만합니다.

간단한 프로세스 내 디듀프 예시입니다.

import json
import hashlib
from cachetools import TTLCache

seen = TTLCache(maxsize=2048, ttl=60)  # 60초 동안 동일 호출 디듀프

def make_key(tool_name: str, args: dict) -> str:
    normalized = json.dumps(args, sort_keys=True, ensure_ascii=False)
    raw = f"{tool_name}:{normalized}".encode("utf-8")
    return hashlib.sha256(raw).hexdigest()

def call_tool_dedup(tool_name, tool_fn, args: dict):
    key = make_key(tool_name, args)

    if key in seen:
        return {
            "status": "blocked",
            "reason": "duplicate_tool_call",
            "cached": True,
            "data": seen[key],
        }

    out = tool_fn(**args)
    seen[key] = out
    return {"status": "ok", "cached": False, "data": out}

운영에서는 blocked 를 반환할지, 캐시를 반환할지 선택해야 합니다.

  • 캐시 반환은 사용자 경험이 좋지만, 잘못된 결과를 고착시킬 수 있습니다.
  • 차단은 루프를 즉시 끊지만, 모델이 대체 경로를 찾지 못하면 답변 품질이 떨어질 수 있습니다.

대부분은 “캐시 반환 + 모델에게 이미 처리된 호출임을 명확히 알리는 메시지”가 더 안정적입니다.

5단계: 상태 머신으로 에이전트 흐름을 제한하기

자유도가 높은 에이전트는 편하지만, 운영에서는 상태 머신이 훨씬 안전합니다. 즉, “지금 단계에서 허용되는 도구만 호출 가능”하도록 게이트를 두는 방식입니다.

예시 단계

  • COLLECT_REQUIREMENTS: 사용자 요구 수집, 도구 호출 금지
  • FETCH_CONTEXT: 검색 도구만 허용
  • PLAN: 계획 수립, 도구 호출 금지
  • EXECUTE: 특정 실행 도구만 허용
  • FINALIZE: 최종 답변, 도구 호출 금지

간단한 게이트 예시입니다.

ALLOWED_TOOLS = {
    "COLLECT_REQUIREMENTS": set(),
    "FETCH_CONTEXT": {"search"},
    "PLAN": set(),
    "EXECUTE": {"db_query", "http_call"},
    "FINALIZE": set(),
}

def guard_tool_call(state: str, tool_name: str):
    allowed = ALLOWED_TOOLS.get(state, set())
    if tool_name not in allowed:
        raise RuntimeError(f"tool_not_allowed_in_state: state={state}, tool={tool_name}")

이 구조의 장점은 루프가 생겨도 “동일 상태에서 반복 호출”이 아니라 “상태 전이가 막혀서 즉시 실패”하게 만들 수 있다는 점입니다. 실패는 관측 가능하고, 수정 가능하지만, 무한루프는 비용을 태우며 조용히 장애를 만듭니다.

6단계: 종료 조건을 프롬프트가 아니라 “검증기”로 강제하기

많은 구현이 “이제 최종 답변을 해” 같은 프롬프트 지시로 종료를 유도합니다. 하지만 모델은 확률적으로 행동하므로, 종료 조건은 코드로 검증해야 합니다.

추천하는 종료 검증기

  • 최종 답변이 특정 스키마를 만족하는지
  • 필수 필드가 채워졌는지
  • 금지된 표현이나 미해결 TODO가 남아있는지
  • tool 결과를 인용해야 하는 정책이 있다면, 인용이 포함됐는지

예를 들어 “최종 답변은 JSON이어야 한다” 같은 요구가 있다면, 파싱 가능한지 검사하고 실패 시에만 제한적으로 재시도합니다.

import json

def validate_final_answer(text: str) -> dict:
    try:
        obj = json.loads(text)
    except Exception:
        return {"ok": False, "reason": "final_not_json"}

    if "answer" not in obj or not isinstance(obj["answer"], str) or not obj["answer"].strip():
        return {"ok": False, "reason": "missing_answer"}

    return {"ok": True, "data": obj}

중요한 포인트는 “검증 실패 시 무한 재시도”가 아니라, 재시도 횟수와 수정 지시를 짧고 명확하게 제한하는 것입니다.

7단계: 관측성과 포렌식 준비 (trace, tool call graph, 알람)

루프를 차단하는 마지막 단계는, 루프가 발생했을 때 “재현 가능하게 기록”하는 것입니다. 특히 에이전트는 입력, 중간 추론, 도구 결과가 얽혀 있어서 로그가 부실하면 원인 규명이 불가능합니다.

최소 관측 항목

  • request id, user id, session id
  • 각 tool call의 name, args hash, latency, status
  • 동일 tool call 반복 횟수
  • 에이전트 반복 횟수, 종료 사유 max_iterations, timeout, blocked_duplicate 등

또한 인프라 관점에서 “어디서 지연이 생겨 루프처럼 보였는지” 추적할 필요도 있습니다. 네트워크 경로 문제로 도구가 계속 타임아웃 나면, 모델은 정보를 얻지 못해 반복 호출을 시도할 수 있습니다. 이런 경우에는 AWS VPC Reachability Analyzer로 502 추적하기 같은 접근이 실제로 도움이 됩니다.

간단한 tool call 로깅 예시입니다.

import time

def instrumented_tool_call(logger, tool_name, args, fn):
    t0 = time.time()
    try:
        out = fn(**args)
        logger.info("tool_call", extra={
            "tool": tool_name,
            "latency_ms": int((time.time() - t0) * 1000),
            "status": "ok",
        })
        return out
    except Exception as e:
        logger.warning("tool_call_failed", extra={
            "tool": tool_name,
            "latency_ms": int((time.time() - t0) * 1000),
            "status": "error",
            "error": str(e),
        })
        raise

알람 설계

  • 특정 시간 창에서 max_iterations 종료 비율이 임계치 초과
  • 동일 tool name의 호출 횟수 급증
  • tool timeout 비율 급증

이런 신호는 “모델이 이상하다”가 아니라 “도구 품질, 스키마, 네트워크, 인증” 중 하나가 깨졌다는 운영 시그널일 가능성이 큽니다.

종합 체크리스트: 7단계를 적용하는 권장 순서

현장에서 바로 적용할 수 있도록 우선순위를 정리하면 다음 순서가 효율적입니다.

  1. 반복 횟수 제한과 타임아웃으로 하드 리밋
  2. 도구 스키마를 좁히고 status 같은 명시적 결과 필드 추가
  3. 에러를 분류하고 재시도 정책을 모델이 아닌 코드가 소유
  4. 동일 호출 디듀프와 캐시로 중복 호출 차단
  5. 상태 머신으로 단계별 허용 도구 제한
  6. 최종 응답 검증기로 종료 조건을 강제
  7. trace와 알람으로 재발 방지용 데이터 축적

무한루프는 한 가지 원인으로만 생기지 않습니다. 모델의 불확실성, 도구의 불안정성, 스키마의 모호함, 상태관리 결함이 결합될 때 폭발합니다. 따라서 “프롬프트를 조금 더 잘 쓰자”로 해결하려고 하기보다, 위 7단계처럼 시스템 차원에서 루프가 구조적으로 불가능해지도록 설계하는 것이 가장 비용 대비 효과가 좋습니다.

Discuss on Twitter • View on GitHub