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

멀티 에이전트 Tool 실행의 핵심 위험

단일 에이전트와 달리 멀티 에이전트 구조에서는 서브 에이전트가 오케스트레이터 모르게 도구를 연쇄 호출할 수 있다. 실제 장애 패턴:

• 권한 확대: 읽기 에이전트가 쓰기 tool을 호출하도록 프롬프트 인젝션
• 무한 루프: tool 결과가 다시 tool 호출을 유발하는 재귀 (관측된 최대 47회 루프)
• 부분 실패: 3단계 중 2단계 완료 후 실패 시 롤백 없이 상태 불일치

오케스트레이터 게이트 패턴

해결책은 중앙 오케스트레이터가 모든 tool 실행을 중계하는 구조다. 서브 에이전트는 tool을 직접 실행하지 않고 "tool 실행 요청"을 오케스트레이터에 반환한다. 오케스트레이터는 ① 권한 검사 → ② 위험도 분류(읽기/쓰기/파괴) → ③ 실행 또는 거부 → ④ 결과 반환 순서로 처리한다.

import anthropic
from enum import Enum

client = anthropic.Anthropic()

class RiskLevel(Enum):
    READ = "read"
    WRITE = "write"
    DESTRUCTIVE = "destructive"

TOOL_RISK_MAP = {
    "search_web": RiskLevel.READ,
    "write_file": RiskLevel.WRITE,
    "delete_record": RiskLevel.DESTRUCTIVE,
}

def orchestrate(task: str, max_iterations: int = 10) -> str:
    messages = [{"role": "user", "content": task}]
    tools = [
        {"name": "search_web", "description": "웹 검색",
         "input_schema": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}},
        {"name": "write_file", "description": "파일 저장",
         "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}},
    ]
    for iteration in range(max_iterations):  # 루프 상한
        response = client.messages.create(
            model="claude-opus-4-5", max_tokens=4096, tools=tools, messages=messages
        )
        if response.stop_reason == "end_turn":
            return response.content[0].text
        tool_results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            risk = TOOL_RISK_MAP.get(block.name, RiskLevel.DESTRUCTIVE)
            if risk == RiskLevel.DESTRUCTIVE:
                tool_results.append({"type": "tool_result", "tool_use_id": block.id,
                                      "content": "DENIED: destructive operation requires human approval"})
                continue
            # 실제 tool 실행 (여기선 mock)
            result = f"[{block.name} 실행 결과: ok, input={block.input}]"
            tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": result})
        messages += [{"role": "assistant", "content": response.content},
                     {"role": "user", "content": tool_results}]
    return "MAX_ITERATIONS_EXCEEDED"

트레이드오프와 실패 모드

오버헤드: 모든 tool 호출이 오케스트레이터를 거치므로 레이턴시가 평균 +80~120ms 증가한다. 고속 파이프라인에서는 READ 등급 tool은 서브 에이전트 직접 실행을 허용하는 2-tier 정책이 현실적이다.

부분 실패 대응: 다단계 쓰기 작업은 각 단계를 이벤트 로그에 기록하고 실패 시 역순 보상 트랜잭션을 실행한다. Saga 패턴을 tool 결과 메시지로 구현 가능하다.

운영 체크리스트

• [ ] 모든 tool에 RiskLevel 명시적 분류, 미분류 tool은 DESTRUCTIVE 기본값
• [ ] max_iterations 환경변수화, 프로덕션 권장값: 15 이하
• [ ] tool 호출마다 (tool_name, input_hash, timestamp, agent_id) 로그 기록
• [ ] DESTRUCTIVE 거부 이벤트 → PagerDuty 알림 연동
• [ ] 주간 tool 호출 분포 리뷰: 특정 tool 비율 > 40% 시 에이전트 목표 재검토
• [ ] 스테이징 환경에서 프롬프트 인젝션 red-team 테스트 월 1회 실시