멀티 에이전트 환경에서 안전한 Tool 실행 패턴

왜 멀티 에이전트 Tool 실행이 위험한가

단일 에이전트와 달리 오케스트레이터가 서브에이전트에게 tool 실행을 위임하면 실패 전파 경로가 기하급수적으로 늘어난다. 대표적인 사고 시나리오:

• 서브에이전트 A가 DB write tool을 무한 재시도 → 커넥션 풀 고갈
• tool 결과가 너무 커서 오케스트레이터 컨텍스트 초과 → 전체 태스크 실패
• 악의적 입력이 tool 인자로 전달되는 prompt injection

수치로 보면, tool 결과를 그대로 컨텍스트에 누적할 경우 10회 루프만으로 claude-opus-4-5의 200k 토큰 한도의 **30~40%**를 소진하는 케이스가 흔하다.

격리 실행 패턴 (SDK 예제)

import anthropic
import asyncio
from typing import Any

client = anthropic.Anthropic()

def execute_tool_safely(tool_name: str, tool_input: dict, timeout: float = 10.0) -> Any:
    """타임아웃 + 결과 크기 제한이 적용된 tool 실행기"""
    import concurrent.futures, json

    def _run():
        # 실제 tool 라우터 호출 (예: DB 조회, API 호출)
        result = tool_router(tool_name, tool_input)
        serialized = json.dumps(result, ensure_ascii=False)
        if len(serialized) > 8000:  # 토큰 폭발 방지: ~2k 토큰 상한
            return {"truncated": True, "preview": serialized[:8000]}
        return result

    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(_run)
        try:
            return future.result(timeout=timeout)
        except concurrent.futures.TimeoutError:
            return {"error": f"tool '{tool_name}' timed out after {timeout}s"}
        except Exception as e:
            return {"error": str(e)}

def run_agent_loop(user_task: str, max_iterations: int = 5) -> str:
    messages = [{"role": "user", "content": user_task}]
    tools = [{"name": "search_db", "description": "DB 검색",
               "input_schema": {"type": "object",
                                "properties": {"query": {"type": "string"}},
                                "required": ["query"]}}]
    for i in range(max_iterations):
        response = client.messages.create(
            model="claude-opus-4-5", max_tokens=1024,
            tools=tools, messages=messages
        )
        if response.stop_reason == "end_turn":
            return next(b.text for b in response.content if hasattr(b, 'text'))
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                result = execute_tool_safely(block.name, block.input)
                tool_results.append({"type": "tool_result",
                                      "tool_use_id": block.id,
                                      "content": str(result)})
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})
    return "max_iterations reached"

트레이드오프와 실패 모드

트레이드오프

• 결과 크기 8000자 상한: 대용량 데이터 분석 태스크에선 정보 손실 발생 → 페이지네이션 tool로 설계 변경 필요.
• max_iterations 하드 캡: 복잡한 태스크가 중간에 잘릴 수 있음 → 태스크 유형별로 다른 상한 적용.
• 동기 ThreadPoolExecutor: 고동시성 환경에선 asyncio + asyncio.wait_for로 교체 권장.

실패 모드

1. tool 결과에 포함된 사용자 데이터가 다음 프롬프트를 조작하는 간접 prompt injection → tool 결과를 XML 태그로 래핑해 모델에 맥락 분리 신호 제공.
2. 서브에이전트가 같은 tool을 다른 인자로 반복 호출 → 호출 히스토리를 집계해 동일 tool 5회 초과 시 강제 종료.
3. stop_reason == "tool_use" 인데 content에 tool_use 블록이 없는 엣지 케이스 → 방어 코드 필수.

운영 체크리스트

• [ ] 모든 tool 실행에 타임아웃(I/O bound 10s, CPU bound 30s 기준) 적용
• [ ] tool 결과 크기를 토큰 환산 8k 이하로 강제 truncate
• [ ] max_iterations를 태스크 유형별로 파라미터화, 기본값 5
• [ ] tool 호출 로그(이름, 인자 해시, 소요시간, 결과 크기)를 구조화 로그로 저장
• [ ] 동일 tool 반복 호출 이상 탐지 알림 설정
• [ ] prompt injection 대응: tool 결과를 <tool_result> 태그로 감싸 컨텍스트 분리