멀티에이전트 오케스트레이션 실전 가이드

멀티에이전트는 공짜가 아니다. 호출이 늘면 토큰 비용·지연·실패 지점이 모두 곱으로 증가한다. 그래서 첫 번째 질문은 항상 "정말 필요한가"여야 한다.

멀티에이전트가 정당화되는 신호

• 이질적인 책임이 한 작업에 섞여 있다. 예: 코드를 생성하는 일과 그 코드를 적대적으로 비판하는 일은 같은 마인드셋으로 동시에 잘할 수 없다. 생성자는 "되게 만들기"에, 비평가는 "깨뜨리기"에 최적화돼야 한다.
• 검증이 작성과 독립적이어야 한다. 같은 모델이 자기 출력을 검토하면 동일한 오해를 반복한다(self-consistency bias). 분리된 검수자가 실제 결함률을 낮춘다.
• 병렬화로 지연을 줄일 수 있다. 보안/성능/과설계/비즈니스 4개 관점 리뷰는 서로 의존이 없으므로 동시에 돌리면 1/4 시간에 끝난다.
• 도구 권한을 격리해야 한다. 리서치 에이전트엔 웹 검색만, 배포 에이전트엔 배포 권한만 주는 식으로 최소 권한 원칙을 적용하려면 에이전트를 쪼개야 한다.

단일 에이전트로 충분한 경우 (멀티에이전트 안티패턴)

• 작업이 본질적으로 순차적이고 단일 책임이다 ("이 함수에 타입 힌트 추가").
• 서브태스크 간 컨텍스트 공유가 너무 빈번해서, 분리하면 핸드오프 오버헤드가 작업 자체보다 커진다.
• 한 번의 잘 짜인 프롬프트 + 도구 호출로 끝난다.

베스트 프랙티스: "에이전트 수"가 아니라 "분리된 책임 수"로 설계하라. 책임이 2개면 에이전트도 2개다. 책임이 1개인데 에이전트를 5개로 쪼개면 핸드오프 지옥만 만든다. 의심스러우면 단일 에이전트 + 도구로 시작하고, 품질이 무너지는 구체적 지점이 보일 때 그 경계에서만 분리하라.

흔한 함정: "에이전트를 많이 쓰면 똑똑해진다"는 착각. 실제로는 에이전트가 늘수록 오케스트레이션 버그(컨텍스트 유실, 무한 루프, 모순된 출력 병합)가 늘고, 디버깅 표면이 넓어진다. 멀티에이전트는 복잡도를 추가하는 결정이지 제거하는 결정이 아니다. 그 복잡도가 품질 이득보다 작을 때만 정당하다.

에이전트들을 어떻게 연결하느냐가 시스템의 성격을 결정한다. 대표 토폴로지 4가지와 적합한 상황을 정리한다.

(1) Orchestrator–Worker (중앙 제어)

하나의 오케스트레이터가 작업을 쪼개 워커들에게 분배하고 결과를 모은다. 가장 흔하고 디버깅이 쉽다. 제어 흐름이 한 곳에 집중되므로 종료 조건·예산·라우팅을 통제하기 좋다.

                ┌──────────────┐
         ┌──────│ Orchestrator │──────┐
         │      └──────────────┘      │
         ▼              ▼             ▼
   ┌─────────┐   ┌──────────┐   ┌──────────┐
   │ Worker A│   │ Worker B │   │ Worker C │   (병렬, 무상태)
   └─────────┘   └──────────┘   └──────────┘
         └──────────────┬─────────────┘
                        ▼
                ┌──────────────┐
                │  Aggregator  │  (결과 통합)
                └──────────────┘

(2) Pipeline / Sequential (파이프라인)

에이전트가 컨베이어 벨트처럼 직렬 연결된다. 각 단계가 앞 단계의 출력을 입력으로 받는다. Plan → Work → Review → Compound 같은 흐름이 전형. 단계마다 책임이 명확하고 핸드오프 계약을 강제하기 좋다. 단점은 병렬화 불가 → 지연이 합산된다.

(3) Debate / Adversarial (토론형)

2개 이상 에이전트가 같은 문제에 대해 상반된 입장을 내고, 심판(judge) 에이전트가 수렴시킨다. 신규 서비스 기획의 "발산 → 교차공격 → 수렴" 같은 의사결정에 강하다. 단일 모델의 확증편향을 깨는 데 효과적이지만 비용이 가장 비싸다.

(4) Hierarchical (계층형)

오케스트레이터가 또 다른 서브-오케스트레이터를 워커로 둔다. 대규모 작업(예: 멀티 레포 마이그레이션)에서 책임을 트리로 쪼갠다. 강력하지만 계층마다 컨텍스트가 희석되므로 3계층을 넘기면 정보 손실이 심해진다.

선택 기준 표

상황	권장 토폴로지
독립적 서브태스크 병렬 처리	Orchestrator–Worker
단계별 변환 + 게이트 강제	Pipeline
모호한 의사결정·트레이드오프 평가	Debate
초대형 작업, 재귀적 분해	Hierarchical (얕게)

베스트 프랙티스: 토폴로지를 섞어 써라. 실전에서는 "파이프라인의 한 단계가 내부적으로 orchestrator–worker"인 경우가 흔하다. 예: Plan(단일) → Implement(단일) → Review(병렬 4개 워커 + aggregator) → Compound(단일).

함정: 디베이트 토폴로지에서 심판 없이 에이전트끼리 합의시키려 하면 무한 핑퐁에 빠지거나 둘 다 양보해 평균적인 무난한 답으로 수렴한다(groupthink). 반드시 독립된 judge를 두고 max round를 박아라.

에이전트 하나하나는 "좁고 깊은 전문가"로 정의한다. 좋은 역할 정의는 세 가지를 명시한다: (1) 단일 책임, (2) 입출력 계약, (3) 허용 도구.

각 에이전트는 별도 파일로 선언해 두는 게 유지보수에 유리하다. 아래는 코드 리뷰 시스템의 "보안 검수자" 정의 예시다.

# security-critic.md

## 역할
변경된 diff에서 보안 취약점만 찾는다. 스타일·성능·가독성은 다루지 않는다.

## 입력 계약
- `diff`: unified diff 문자열
- `context`: 변경 파일들의 주변 코드 (호출부 포함)

## 출력 계약 (엄격한 JSON)
{
  "verdict": "pass" | "revision_needed",
  "findings": [
    { "severity": "P0"|"P1"|"P2", "file": str, "line": int,
      "issue": str, "exploit": str, "fix": str }
  ]
}

## 허용 도구
- read_file (읽기 전용)
- grep
## 금지
- 파일 쓰기, 명령 실행, 네트워크 접근

## 검수 렌즈 (적대적)
- 인증/인가 우회: 모든 mutation 라우트에 권한 체크가 있는가
- 입력 검증: 길이/타입 오버플로, SQL/명령 인젝션
- 시크릿: 로그·에러 메시지·응답 본문에 PII/키 노출
- 멱등성·재진입: 결제·환불 경로의 중복 실행

시스템 프롬프트 작성의 핵심 원칙:

• 한 가지만 시킨다. "보안 그리고 성능"을 한 에이전트에 주면 둘 다 얕아진다.
• 출력 스키마를 강제한다. 다음 단계(aggregator/orchestrator)가 파싱할 수 있도록 JSON 계약을 못박는다. 자연어 산문은 핸드오프에서 깨진다.
• 렌즈(lens)를 명시한다. "심사관이 뭘로 리젝할까" 같은 적대적 관점을 프롬프트에 직접 넣으면 검수 깊이가 달라진다.

도구 권한 격리 (최소 권한)

# 에이전트별 도구 화이트리스트 — 코드로 강제
AGENT_TOOLS = {
    "researcher":      {"web_search", "read_file"},        # 읽기만
    "maker":           {"read_file", "write_file", "run_tests"},
    "security_critic": {"read_file", "grep"},              # 쓰기 불가
    "deployer":        {"deploy"},                          # 배포만
}

def call_tool(agent: str, tool: str, **kwargs):
    if tool not in AGENT_TOOLS[agent]:
        raise PermissionError(f"{agent}는 {tool} 권한 없음")
    return TOOL_REGISTRY[tool](**kwargs)

베스트 프랙티스: 검수자(Critic)에게는 절대 쓰기 권한을 주지 마라. 검수자가 직접 코드를 고치기 시작하면 "검수"와 "작성"의 경계가 무너지고, 자기가 고친 걸 자기가 통과시키는 self-approval 루프가 생긴다. 검수자는 판정과 지적만, 수정은 Maker에게 핸드오프한다.

함정: 한 에이전트의 시스템 프롬프트에 "필요하면 다른 역할도 수행하라"는 식의 탈출구를 넣는 것. 이러면 역할 경계가 즉시 붕괴한다. 경계는 프롬프트가 아니라 도구 권한 코드로 강제해야 새어나가지 않는다.

핸드오프는 멀티에이전트 시스템에서 가장 많은 버그가 사는 곳이다. 에이전트 A가 알던 정보를 B가 모르거나, A의 산문 출력을 B가 잘못 파싱하거나, 둘이 서로에게 제어권을 떠넘기며 무한 루프를 도는 일이 전부 여기서 생긴다.

핵심 원칙: 핸드오프는 명시적 계약이다. 전체 대화 히스토리를 통째로 넘기지 마라.

나쁜 핸드오프 (전체 컨텍스트 덤프)

# 안티패턴: 앞 에이전트의 전체 메시지 히스토리를 그대로 전달
result_b = agent_b.run(messages=agent_a.full_history)  # 토큰 폭증 + 노이즈

앞 에이전트의 시행착오·도구 호출 로그·중간 산출물이 전부 따라가서, 다음 에이전트는 노이즈에 파묻히고 토큰은 폭증한다.

좋은 핸드오프 (구조화된 산출물만)

from pydantic import BaseModel

class PlanHandoff(BaseModel):
    """Planner -> Maker 로 넘기는 계약. 이것만 전달한다."""
    goal: str
    files_to_change: list[str]
    acceptance_criteria: list[str]   # Maker가 충족해야 할 체크리스트
    constraints: list[str]            # 건드리면 안 되는 것
    open_questions: list[str]         # 불확실한 부분 명시

def plan_to_make(plan: PlanHandoff) -> MakeResult:
    # Maker는 plan 객체만 받는다. Planner의 사고 과정은 안 받는다.
    return maker.run(
        system=MAKER_PROMPT,
        input=plan.model_dump_json(),
    )

핸드오프 계약이 반드시 담아야 할 것

1. 목표(goal): 다음 에이전트가 무엇을 달성해야 하는가.
2. 수용 기준(acceptance criteria): 완료를 어떻게 판정하는가. 검증 루프의 기준점이 된다.
3. 제약(constraints): 건드리면 안 되는 것 ("기존 API 시그니처 유지").
4. 불확실성(open questions): 앞 에이전트가 확신 못 한 부분. 이걸 빼면 다음 에이전트가 잘못된 가정을 진실로 받아들인다.

불확실성을 명시적으로 전달하지 않으면, 단계마다 가정이 사실로 굳어져 파이프라인 끝에서 완전히 틀린 결과가 나온다. "명시/없음/잘못됨"의 3상태를 구분하라 — undefined(검증 안 함)와 false(검증 후 없음)는 다른 의미다.

제어형 핸드오프 (에이전트가 다음 에이전트를 호출)

OpenAI Agents SDK 등은 handoff를 도구처럼 노출한다. 이때도 데이터 계약은 동일하게 강제해야 한다.

from agents import Agent, handoff

triage = Agent(
    name="Triage",
    instructions="버그면 dev_agent로, 질문이면 qa_agent로 핸드오프.",
    handoffs=[handoff(dev_agent), handoff(qa_agent)],
)
# 라우팅 책임만 가진 얇은 에이전트가 분류 후 넘긴다

함정: "양방향 자유 핸드오프". A↔B가 서로를 자유롭게 호출할 수 있으면 무한 핑퐁(서로 "네가 해"를 반복)에 빠진다. 핸드오프는 가능하면 단방향 + 종료 조건으로 설계하고, 양방향이 꼭 필요하면 오케스트레이터를 거치게 해서 라운드 카운터로 끊어라.

멀티에이전트의 진짜 가치는 여기서 나온다. 작성자와 검수자를 분리하면 품질이 측정 가능하게 올라간다. 핵심 패턴은 Maker → Critic → Integrator다.

• Maker: 산출물을 만든다 (코드, 문서, 설계).
• Critic: 적대적으로 검수한다. Maker와 다른 인스턴스/프롬프트여야 한다. 통과/반려 판정 + 구체적 지적을 낸다.
• Integrator: Critic의 지적을 반영해 수정하거나, 여러 Critic의 상충 의견을 조정한다.

철칙: Maker는 자기 작업을 검수하지 않는다. 같은 컨텍스트를 가진 모델은 자기가 만들 때 가졌던 오해를 검수할 때도 똑같이 가진다. "이거 맞아?"라고 자문하면 "응 맞아"가 나온다. Critic은 별도 프롬프트로, 산출물만 보고, 작성 과정은 모른 채 검수해야 객관성이 생긴다.

def verification_loop(task: PlanHandoff, max_rounds: int = 3):
    artifact = maker.run(task)

    for round_num in range(max_rounds):
        # Critic은 artifact만 본다 — maker의 reasoning은 안 준다
        review = critic.run(
            system=CRITIC_PROMPT,
            input={
                "artifact": artifact,
                "acceptance_criteria": task.acceptance_criteria,
            },
        )

        if review["verdict"] == "pass":
            return artifact, review  # 게이트 통과

        # 반려 -> Maker가 지적사항 반영 (별도 에이전트가 fix)
        artifact = maker.run(
            system=MAKER_PROMPT,
            input={
                "previous": artifact,
                "required_fixes": review["findings"],
            },
        )

    # max_rounds 소진 -> 수렴 실패. 사람에게 에스컬레이션
    raise ConvergenceError(
        f"{max_rounds}회 내 수렴 실패", last_review=review
    )

병렬 다관점 검수 (Orchestrator–Worker + Critic)

복잡한 산출물은 여러 렌즈로 동시에 검수한다. 각 렌즈가 독립 에이전트다.

import asyncio

CRITICS = ["security_critic", "performance_critic",
           "overdesign_critic", "business_critic"]

async def multi_lens_review(artifact, criteria):
    reviews = await asyncio.gather(*[
        run_critic(name, artifact, criteria) for name in CRITICS
    ])
    # 통합: P0가 하나라도 있으면 반려
    all_findings = [f for r in reviews for f in r["findings"]]
    has_blocker = any(f["severity"] == "P0" for f in all_findings)
    return {
        "verdict": "revision_needed" if has_blocker else "pass",
        "findings": sorted(all_findings, key=lambda f: f["severity"]),
    }

수용 기준을 검수의 기준점으로: Critic에게 "좋은지 봐줘"라고 하면 주관적 잡담이 나온다. 4단계에서 만든 acceptance_criteria를 Critic에 함께 넘기면, 검수가 체크리스트 대조로 바뀌어 재현 가능해진다.

베스트 프랙티스: 심각도(severity)로 게이트를 정의하라. P0(머지 차단)/P1(곧 수정)/P2(개선)로 나누고, "P0이 0건일 때만 통과"처럼 기계적으로 판정 가능한 종료 조건을 둔다. "검수자가 만족하면 통과"는 종료 조건이 아니다 — 영원히 트집 잡을 수 있다.

함정 1: Critic이 너무 관대하다. LLM은 기본적으로 협조적이라 "대체로 좋아 보입니다"로 통과시키는 경향이 있다. 프롬프트에 "통과시키는 게 기본이 아니다. 결함을 찾는 게 임무다. 결함이 없다고 증명할 때만 pass"라고 적대적 디폴트를 박아라.

함정 2: 무한 수렴 루프. Critic이 매 라운드 새 트집을 잡아 절대 통과 안 되는 경우. max_rounds를 반드시 두고, 소진 시 사람에게 에스컬레이션한다. 또한 Critic에게 "이전 라운드에서 지적 안 한 새 항목은 P2 이하만 허용" 같은 제약을 줘서 골대 이동(moving goalposts)을 막는다.

오케스트레이터는 시스템의 제어 평면이다. 라우팅, 상태 관리, 종료 조건을 한 곳에 모은다. 에이전트들이 자유롭게 서로를 호출하게 두지 말고, 오케스트레이터가 다음 누구를 부를지 결정하게 하면 흐름이 통제 가능해진다.

가장 견고한 구현은 명시적 상태 머신이다. "지금 어느 단계인가"를 데이터로 들고, 각 단계의 출력에 따라 다음 단계를 결정한다.

from dataclasses import dataclass, field
from enum import Enum

class Stage(Enum):
    PLAN = "plan"
    IMPLEMENT = "implement"
    REVIEW = "review"
    REVISE = "revise"
    DONE = "done"
    ESCALATE = "escalate"   # 사람 개입

@dataclass
class OrchestratorState:
    stage: Stage = Stage.PLAN
    artifact: dict | None = None
    review: dict | None = None
    round: int = 0
    max_rounds: int = 3
    cost_usd: float = 0.0
    budget_usd: float = 5.0
    history: list = field(default_factory=list)

def step(state: OrchestratorState) -> OrchestratorState:
    # 예산 가드 — 어느 단계에서든 우선 검사
    if state.cost_usd >= state.budget_usd:
        state.stage = Stage.ESCALATE
        return state

    if state.stage == Stage.PLAN:
        plan = planner.run(...)
        state.artifact = plan
        state.stage = Stage.IMPLEMENT

    elif state.stage == Stage.IMPLEMENT:
        state.artifact = maker.run(state.artifact)
        state.stage = Stage.REVIEW

    elif state.stage == Stage.REVIEW:
        state.review = run_multi_lens_review(state.artifact)
        state.stage = (Stage.DONE if state.review["verdict"] == "pass"
                       else Stage.REVISE)

    elif state.stage == Stage.REVISE:
        state.round += 1
        if state.round >= state.max_rounds:
            state.stage = Stage.ESCALATE   # 수렴 실패
        else:
            state.artifact = maker.run({
                "previous": state.artifact,
                "fixes": state.review["findings"],
            })
            state.stage = Stage.REVIEW

    state.cost_usd += last_call_cost()
    return state

def orchestrate(state: OrchestratorState) -> OrchestratorState:
    while state.stage not in (Stage.DONE, Stage.ESCALATE):
        state = step(state)
    return state

이 구조의 장점:

• 모든 전이가 한눈에 보인다. 디버깅 시 "어느 단계에서 막혔나"가 명확하다.
• 종료 조건이 명시적이다. DONE / ESCALATE 둘 중 하나로만 끝난다. 무한 루프가 구조적으로 불가능하다 (round 카운터 + 예산 가드).
• 재시작 가능하다. state를 직렬화해 저장하면, 크래시 후 그 지점부터 재개할 수 있다 (durable execution).

LangGraph로 같은 구조 표현하기 (그래프 기반 오케스트레이션 라이브러리를 쓴다면):

from langgraph.graph import StateGraph, END

graph = StateGraph(OrchestratorState)
graph.add_node("plan", plan_node)
graph.add_node("implement", implement_node)
graph.add_node("review", review_node)
graph.add_node("revise", revise_node)

graph.add_conditional_edges(
    "review",
    lambda s: "done" if s.review["verdict"] == "pass"
              else ("escalate" if s.round >= s.max_rounds else "revise"),
    {"done": END, "revise": "revise", "escalate": END},
)
graph.add_edge("revise", "review")
app = graph.compile()

베스트 프랙티스: 종료 조건을 세 종류 모두 두라 — (1) 성공(DONE), (2) 라운드 소진(ESCALATE), (3) 예산 초과(ESCALATE). 하나라도 빠지면 폭주 시나리오가 생긴다. 특히 예산 가드는 단계 진입 전에 검사해야 초과 직전에 멈춘다.

함정: 종료 조건을 LLM의 판단("작업이 끝난 것 같으면 멈춰")에 맡기는 것. LLM은 "한 번 더 개선할 수 있을 것 같다"며 영원히 안 멈추거나, 반대로 절반만 하고 "완료"를 선언한다. 종료는 항상 오케스트레이터 코드의 결정론적 조건으로 통제하라.

에이전트들이 작업하려면 일부 상태를 공유해야 한다. 하지만 전부 공유하면 컨텍스트가 오염되고, 아무것도 공유 안 하면 핸드오프가 불가능하다. 경계 설계가 관건이다.

3계층 메모리 모델

계층	범위	예시	공유 여부
Scratchpad	에이전트 1개의 1회 실행	도구 호출 중간 결과, 추론 과정	격리 (절대 공유 X)
Handoff payload	에이전트 간 1회 전달	4절의 PlanHandoff	명시적 전달만
Shared store	세션/작업 전체	결정 기록, 산출물 버전, 비용 누적	구조화해서 공유

핵심: Scratchpad(추론 과정)는 절대 다음 에이전트로 새어 나가면 안 된다. Maker가 코드를 짜며 시도한 5번의 실패는 Critic에게 노이즈일 뿐이다. Critic은 최종 산출물만 봐야 객관적으로 검수한다.

공유 스토어 구현 (구조화 키-값)

class SharedStore:
    """작업 전체에 걸친 구조화 상태. 산문 덤프가 아니라 키-값."""
    def __init__(self):
        self._data: dict = {
            "decisions": [],   # 확정된 의사결정 (불변 로그)
            "artifacts": {},   # 버전별 산출물
            "open_questions": [],
            "cost_usd": 0.0,
        }

    def record_decision(self, who: str, what: str, why: str):
        self._data["decisions"].append({
            "agent": who, "decision": what,
            "rationale": why, "ts": now(),
        })

    def view_for(self, agent: str) -> dict:
        """에이전트별로 필요한 슬라이스만 노출 (최소 공개)."""
        if agent == "critic":
            # critic은 결정 로그를 볼 필요 없음 — 산출물만
            return {"artifacts": self._data["artifacts"]}
        return self._data

왜 키-값 구조인가: 공유 메모리를 "지금까지의 대화를 요약한 긴 문자열"로 만들면, 매 핸드오프마다 그 문자열이 다시 요약되며 정보가 손실되고(lossy compression), 토큰은 누적된다. 구조화 키-값은 필요한 필드만 정확히 꺼내 쓸 수 있어 손실이 없다.

메모리 주입의 피드백 루프 위험

장기 운영 시스템에서 "이전 세션 로그를 다음 세션에 자동 주입"하는 패턴은 위험하다. 출력이 입력으로 되먹임되면서 특정 패턴(편향, 정책 위반 어휘, 잘못된 가정)이 증폭될 수 있다. 자동 주입을 한다면:

• 본문이 아니라 메타데이터만 주입한다 (타임스탬프, 출처, 요약 한 줄). 전체 본문을 통째로 주입하면 노이즈와 위험이 누적된다.
• 격리 폴더를 둔다. 민감하거나 오염된 로그는 자동 주입 대상에서 제외한다.
• 킬 스위치를 둔다. 환경변수 하나로 자동 주입을 끌 수 있게 한다.

베스트 프랙티스: "이 정보를 이 에이전트가 반드시 알아야 하는가"를 매 공유마다 물어라. 답이 "있으면 좋고"면 공유하지 마라. 컨텍스트는 많을수록 좋은 게 아니라, 관련성 높을수록 좋다. 무관한 컨텍스트는 정확도를 떨어뜨린다.

함정: 가변 공유 상태에 여러 에이전트가 동시에 쓰기. 병렬 워커가 같은 store를 동시에 수정하면 race condition이 생긴다. 병렬 단계에서는 각 워커가 자기 결과만 반환하고, 통합(merge)은 단일 aggregator가 순차적으로 하게 하라. 공유 스토어 쓰기는 단일 지점으로 직렬화한다.

에이전트가 자율적으로 행동할수록 "하면 안 되는 일"을 막는 게 중요해진다. 가드레일은 프롬프트 권고가 아니라 오케스트레이터가 강제하는 코드여야 한다. 프롬프트로만 "외부 이메일 보내지 마"라고 하면, 충분히 설득력 있는 컨텍스트가 주어졌을 때 모델은 어긴다.

가드레일 3계층

1. 입력 가드 (pre-flight): 에이전트 실행 전 입력 검사. 예: 프롬프트 인젝션 패턴, 정책 위반 요청 차단.
2. 행동 가드 (in-flight): 도구 호출 시점 권한·정책 검사. 예: 금지 도구 차단, 승인 필요 작업 게이트.
3. 출력 가드 (post-flight): 결과 반환 전 검사. 예: PII 노출, 톤, 사실성.

정책을 선언적으로 분리해 두면 (예: YAML) 코드와 정책이 섞이지 않는다.

# guard.yaml — 선언적 정책
forbidden:                      # 절대 금지 (어떤 컨텍스트에서도)
  - external_email_send
  - data_delete
  - payment_execute
  - public_channel_post

require_approval:               # 사람 승인 게이트
  - external_customer_content
  - bulk_update_over: 10

always_allow:
  - internal_dm
  - read_notion
  - local_file_write
  - web_search

import yaml

POLICY = yaml.safe_load(open("guard.yaml"))

class GuardrailViolation(Exception): ...
class ApprovalRequired(Exception): ...

def enforce_action(action: str, payload: dict, ctx: dict):
    if action in POLICY["forbidden"]:
        raise GuardrailViolation(f"{action}: 정책상 금지")

    for rule in POLICY["require_approval"]:
        if isinstance(rule, str) and action == rule:
            raise ApprovalRequired(action)
        if isinstance(rule, dict) and "bulk_update_over" in rule:
            if payload.get("count", 0) > rule["bulk_update_over"]:
                raise ApprovalRequired(f"{action}: 대량 변경 승인 필요")

    # 통과한 것만 실행
    return TOOL_REGISTRY[action](**payload)

예산 가드 (비용 폭주 방지)

멀티에이전트는 호출이 곱으로 늘어 비용이 통제를 벗어나기 쉽다. 토큰 비용을 누적 추적하고 임계치에서 경고/중단한다.

class BudgetGuard:
    def __init__(self, hard_limit_usd: float, warn_at: float):
        self.spent = 0.0
        self.hard = hard_limit_usd
        self.warn = warn_at

    def charge(self, usd: float):
        self.spent += usd
        if self.spent >= self.hard:
            raise GuardrailViolation(
                f"예산 초과: ${self.spent:.2f} / ${self.hard}")
        if self.spent >= self.warn:
            log.warning(f"예산 경고: ${self.spent:.2f}")

중요 원칙: 가드레일은 완화 불가, 강화만 가능. 글로벌 정책을 하위 작업에서 약화할 수 있게 하면, 충분히 그럴듯한 이유와 함께 결국 우회된다. 하위 컨텍스트는 글로벌 가드를 더 엄격하게만 만들 수 있어야 한다.

베스트 프랙티스: "승인 필요" 작업은 예외를 던져 흐름을 멈추고 사람에게 에스컬레이션하는 구조로 만들어라. 모델이 "승인받았다고 가정하고 진행"하지 못하게, 승인 토큰이 실제로 주입되기 전엔 도구 자체가 실행되지 않아야 한다.

함정: 가드레일을 시스템 프롬프트에만 적어두는 것. "너는 절대 X를 하면 안 돼"는 권고지 강제가 아니다. 프롬프트 인젝션이나 충분히 긴 대화 후의 컨텍스트 드리프트로 무력화된다. 진짜 가드는 도구 실행 직전 코드 검사에 있어야 한다. 프롬프트 가드와 코드 가드를 둘 다 두되(defense in depth), 신뢰는 코드 쪽에만 둔다.

멀티에이전트 시스템이 잘못됐을 때, "어느 에이전트가 어떤 입력으로 무엇을 출력했는지"를 볼 수 없으면 디버깅이 불가능하다. 단일 에이전트는 메시지 히스토리 하나만 보면 되지만, 멀티에이전트는 여러 실행이 인과로 얽혀 있어 트레이싱이 필수다.

계층적 트레이스 구조

하나의 작업(trace)이 여러 에이전트 실행(span)으로 구성되고, 각 span이 도구 호출(sub-span)을 가진다. OpenTelemetry의 trace/span 모델을 그대로 적용할 수 있다.

import time, json, uuid, contextvars

current_trace = contextvars.ContextVar("trace_id")

def traced(agent_name: str):
    """에이전트 실행을 span으로 기록하는 데코레이터."""
    def deco(fn):
        def wrapper(*args, **kwargs):
            span_id = str(uuid.uuid4())[:8]
            t0 = time.time()
            log_event({
                "trace": current_trace.get("-"),
                "span": span_id, "agent": agent_name,
                "event": "start", "input": redact(kwargs),
            })
            try:
                out = fn(*args, **kwargs)
                log_event({
                    "trace": current_trace.get("-"),
                    "span": span_id, "agent": agent_name,
                    "event": "end", "ms": int((time.time()-t0)*1000),
                    "output": redact(out),
                    "tokens": last_token_usage(),
                })
                return out
            except Exception as e:
                log_event({"span": span_id, "agent": agent_name,
                           "event": "error", "err": str(e)})
                raise
        return wrapper
    return deco

로그에 반드시 담아야 할 것

• trace_id / span_id / parent_span: 인과 관계 재구성용.
• 에이전트 이름 + 단계(stage): "어느 역할이 언제".
• 입력/출력 (민감정보 redact): 시크릿·PII는 [REDACTED]로 마스킹. 로그가 유출되어도 키가 안 새게.
• 토큰 사용량 + 비용: 어느 에이전트가 비용을 먹는지.
• verdict / round: 검증 루프가 몇 바퀴 돌았는지.

디버깅 시나리오별 무엇을 보는가

증상	봐야 할 것
결과가 엉뚱함	핸드오프 payload — 컨텍스트가 제대로 전달됐나
무한 루프/안 끝남	round 카운터 + verdict 추이 — 골대 이동 여부
비용 폭증	span별 토큰 — 어느 에이전트가 컨텍스트를 통째 받았나
가끔만 실패	error span + 그 직전 입력 — 특정 입력 패턴

평가(eval) 연동: 트레이스를 저장해 두면 회귀 테스트셋이 된다. 잘 된 실행과 망한 실행을 케이스로 모아, 시스템 프롬프트나 라우팅을 바꿀 때 "이전에 통과하던 게 깨지지 않는지" 자동 검증할 수 있다.

베스트 프랙티스: 트레이스를 처음부터 넣어라. "나중에 추가"하면 이미 프로덕션에서 재현 안 되는 버그를 만나 후회한다. 최소한 "trace_id + agent + input/output(redacted) + tokens" JSON 한 줄씩이라도 첫날부터 남겨라.

함정: 로그에 전체 프롬프트/응답을 redact 없이 남기는 것. 멀티에이전트는 컨텍스트를 여기저기 전달하므로 시크릿이 예상 못 한 span에 묻어 들어간다. redact를 로깅 함수 자체에 박아, 빼먹을 수 없게 하라. 또 다른 함정은 trace_id를 전파 안 해서(contextvars 미사용) 병렬 워커들의 로그가 섞이는 것 — 누가 누군지 구분 불가능해진다.

멀티에이전트는 본질적으로 호출이 많아 비용·지연이 단일 에이전트보다 크다. 하지만 잘 설계하면 각 작업에 맞는 크기의 모델과 병렬화로 단일 거대 모델 호출보다 싸고 빠르게 만들 수 있다.

모델 라우팅 (작업 복잡도 ↔ 모델 크기)

모든 에이전트에 최고 성능 모델을 쓸 필요는 없다. 분류·라우팅·단순 추출은 작고 빠른 모델로, 깊은 추론·코드 생성·적대적 검수는 강한 모델로 보낸다.

MODEL_BY_ROLE = {
    "triage":          "small-fast",   # 분류만 — 작은 모델로 충분
    "researcher":      "small-fast",   # 검색 + 추출
    "planner":         "large",        # 깊은 추론 필요
    "maker":           "large",        # 코드 생성
    "security_critic": "large",        # 적대적 검수 — 절대 아끼지 말 것
    "formatter":       "small-fast",   # 출력 포맷팅
}

검수자(Critic)에서 모델 크기를 아끼지 마라. 검수 깊이가 시스템 전체 품질의 상한이다. Maker에 강한 모델, Critic에 약한 모델을 쓰면 약한 검수자가 강한 작성자의 결함을 놓친다.

프롬프트 캐싱 (반복되는 컨텍스트 재사용)

같은 시스템 프롬프트·도구 정의·공유 컨텍스트를 여러 에이전트가 반복 사용하면, 프롬프트 캐싱으로 입력 토큰 비용을 크게 줄일 수 있다. 변하지 않는 부분(시스템 프롬프트, 도구 스키마, 큰 참조 문서)을 프롬프트 앞쪽에 고정 배치하고, 변하는 부분(이번 작업 입력)을 뒤에 두면 캐시 적중률이 올라간다.

[ 캐시 대상: 시스템 프롬프트 + 도구 정의 + 참조 문서 ]  <- 고정, 재사용
[ 캐시 미스: 이번 핸드오프 payload ]                    <- 매번 변함

병렬화로 지연 줄이기

의존성 없는 단계는 반드시 병렬로. 4개 관점 리뷰를 직렬로 돌리면 4배 느리다.

import asyncio

async def parallel_stage(artifact):
    # 서로 의존 없는 검수들 — 동시 실행
    results = await asyncio.gather(
        security_critic(artifact),
        performance_critic(artifact),
        business_critic(artifact),
        return_exceptions=True,   # 하나 실패가 전체를 막지 않게
    )
    return [r for r in results if not isinstance(r, Exception)]

조기 종료 (early exit)

비용을 가장 크게 아끼는 건 불필요한 호출을 안 하는 것이다. Triage에서 "이건 단순 작업"으로 분류되면 전체 파이프라인 대신 단일 에이전트로 직행시킨다. 검증 1라운드에서 P0이 없으면 추가 라운드를 돌지 않는다.

비용 비교 직관

전략	상대 비용	비고
모든 에이전트 = 최고 모델, 직렬	100% (기준)	느리고 비쌈
역할별 모델 라우팅	~40–60%	단순 역할에 작은 모델
+ 프롬프트 캐싱	~25–40%	반복 컨텍스트 캐시
+ 병렬화	비용 동일, 지연 ↓	벽시계 시간만 단축
+ 조기 종료	~15–30%	쉬운 케이스 우회

(수치는 작업 특성·모델·캐시 적중률에 따라 크게 달라지므로 방향으로만 받아들이고, 실제 트레이스의 토큰 로그로 측정하라.)

베스트 프랙티스: 최적화하기 전에 9절의 트레이스로 측정하라. 대개 비용의 80%는 한두 에이전트(주로 전체 컨텍스트를 통째 받는 곳)에서 나온다. 추측으로 모든 곳을 깎지 말고, 토큰 로그에서 가장 비싼 span부터 줄여라.

함정: 비용을 아끼려고 검수 라운드 수나 검수자 모델을 줄였다가 결함이 새는 것. 검수에서 아낀 돈은 프로덕션 인시던트로 몇 배 돌아온다. 비용은 작성·포맷팅·라우팅 쪽에서 아끼고, 검증 쪽은 충분히 투자하라.

멀티에이전트는 실패 지점이 곱으로 많다. 도구 타임아웃, API 레이트 리밋, MCP 연결 끊김, 검증 수렴 실패, 모순된 출력... 각각에 대해 결정론적 복구 전략이 필요하다. "알아서 잘 되겠지"는 프로덕션에서 통하지 않는다.

실패 유형별 복구 정책 (선언적)

# recovery.yaml
tool_error:
  strategy: retry
  max_attempts: 1
  fallback: report_to_user

rate_limit:
  strategy: backoff
  wait_seconds: 30
  max_attempts: 3

mcp_connection_lost:
  strategy: reconnect_then_notify

convergence_failure:        # 검증 max_rounds 소진
  strategy: escalate_to_human

budget_exceeded:
  strategy: halt_and_report   # 절대 계속 진행 X

지수 백오프 재시도 (레이트 리밋·일시 오류)

import time, random

def with_retry(fn, max_attempts=3, base=1.0):
    for attempt in range(max_attempts):
        try:
            return fn()
        except RateLimitError:
            if attempt == max_attempts - 1:
                raise
            # 지수 백오프 + 지터
            sleep = base * (2 ** attempt) + random.uniform(0, 0.5)
            time.sleep(sleep)
        except GuardrailViolation:
            raise   # 정책 위반은 재시도하지 않는다 — 즉시 중단

주의: 모든 예외를 재시도하면 안 된다. 가드레일 위반·검증 반려 같은 논리적 실패는 재시도해도 같은 결과다. 재시도는 일시적 인프라 실패(타임아웃, 레이트 리밋, 연결 끊김)에만 적용한다.

서킷 브레이커 (반복 실패 시 차단)

특정 도구나 에이전트가 연속 실패하면, 계속 때리지 말고 회로를 열어 빠르게 실패시킨다.

class CircuitBreaker:
    def __init__(self, threshold=3, cooldown=60):
        self.fails = 0
        self.threshold = threshold
        self.open_until = 0

    def call(self, fn):
        if time.time() < self.open_until:
            raise CircuitOpen("회로 열림 — 빠른 실패")
        try:
            r = fn()
            self.fails = 0   # 성공 시 리셋
            return r
        except Exception:
            self.fails += 1
            if self.fails >= self.threshold:
                self.open_until = time.time() + 60
            raise

휴먼 인 더 루프 (HITL) 에스컬레이션

자율 시스템이라도 반드시 사람을 부르는 지점을 명시해야 한다:

• 수렴 실패: 검증 루프가 max_rounds 안에 통과 못 함.
• 승인 필요 작업: 가드레일이 require_approval로 막은 작업.
• 모순된 검수 결과: 두 Critic이 정반대 판정 → 사람이 중재.
• 예산 초과: 한도 도달.

에스컬레이션 시에는 사람이 판단할 수 있는 충분한 컨텍스트를 함께 넘긴다 — 마지막 산출물, 검수 결과, 막힌 이유, 가능한 선택지. "실패했습니다"만 던지면 사람도 디버깅을 처음부터 해야 한다.

def escalate(state, reason):
    notify_human({
        "reason": reason,
        "stage": state.stage.value,
        "last_artifact": state.artifact,
        "last_review": state.review,
        "rounds_used": state.round,
        "options": suggest_next_actions(state),  # 사람이 고를 선택지
    })

베스트 프랙티스: 복구의 디폴트는 "안전하게 멈추고 사람에게 보고"여야 한다. 모호할 때 추측으로 진행시키지 마라. 특히 되돌릴 수 없는 행동(배포, 결제, 외부 발송) 직전의 실패는 항상 중단 + 에스컬레이션이다. 자동 복구는 멱등하고 되돌릴 수 있는 작업에만 적용한다.

함정 1: 부분 실패 후 일관성 깨짐. 병렬 워커 3개 중 1개가 실패했는데 나머지 2개 결과로 통합을 강행하면, 불완전한 데이터를 완전한 것처럼 다음 단계로 넘긴다. 통합 전에 "필수 입력이 다 모였나"를 검사하고, 빠졌으면 격하(degrade)하거나 멈춰라.

함정 2: 무한 재시도. 백오프 없는 재시도나 max_attempts 미설정은 레이트 리밋을 더 악화시키거나 비용을 태운다. 모든 재시도 루프에 상한과 백오프를 박아라.

지금까지의 조각을 하나로 묶은, 실제로 돌아가는 코드 리뷰 오케스트레이션의 골격이다. Plan → Implement → 병렬 다관점 Review → Revise 루프 → 게이트 통과 또는 에스컬레이션.

import asyncio

async def run_code_review_pipeline(pr_diff: str, context: dict):
    store = SharedStore()
    budget = BudgetGuard(hard_limit_usd=5.0, warn_at=3.0)
    state = OrchestratorState(max_rounds=3, budget_usd=5.0)

    # 1) PLAN — 무엇을 검수할지 계획 (작은 모델)
    plan = await planner.run(model="large", input={
        "diff": pr_diff, "context": context,
    })
    store.record_decision("planner", plan["goal"], plan["rationale"])

    artifact = pr_diff
    for round_num in range(state.max_rounds):
        budget.charge(estimate_round_cost())

        # 2) REVIEW — 4관점 병렬, 검수자는 large 모델
        reviews = await asyncio.gather(
            security_critic.run(artifact, plan["acceptance_criteria"]),
            performance_critic.run(artifact, plan["acceptance_criteria"]),
            overdesign_critic.run(artifact, plan["acceptance_criteria"]),
            business_critic.run(artifact, plan["acceptance_criteria"]),
            return_exceptions=True,
        )
        findings = aggregate(reviews)   # P0/P1/P2 통합·정렬

        # 3) GATE — 결정론적 종료 조건
        if not any(f["severity"] == "P0" for f in findings):
            return {"verdict": "pass", "findings": findings,
                    "rounds": round_num + 1, "cost": budget.spent}

        # 4) REVISE — Maker가 수정 (Critic은 수정 안 함)
        artifact = await maker.run(model="large", input={
            "previous": artifact,
            "required_fixes": [f for f in findings
                               if f["severity"] in ("P0", "P1")],
        })

    # 5) 수렴 실패 -> 사람에게 컨텍스트와 함께 에스컬레이션
    escalate(state, reason="3라운드 내 P0 해소 실패")
    return {"verdict": "escalated", "findings": findings}

이 예제가 가이드의 모든 원칙을 어떻게 구현하는지:

• 역할 분담(3절): planner/maker/4 critics가 각각 단일 책임.
• 핸드오프(4절): acceptance_criteria를 계약으로 전달, Maker엔 required_fixes만.
• 검증 루프(5절): Critic이 Maker와 분리, 자기검증 없음.
• 오케스트레이터(6절): 결정론적 게이트 + max_rounds 종료.
• 병렬화(10절): 4관점 동시 실행.
• 가드레일·복구(8·11절): budget guard + escalation.

프로덕션 투입 전 체크리스트

• 종료 조건이 3종(성공/라운드소진/예산) 모두 있는가. LLM 판단에 종료를 맡기지 않았는가.
• Maker가 자기 작업을 검수하지 않는가. Critic은 별도 프롬프트·인스턴스인가.
• Critic에 쓰기 권한이 없는가. 도구 권한을 코드로 화이트리스트했는가.
• 핸드오프가 구조화 계약인가. 전체 히스토리를 덤프하지 않는가. 불확실성(open questions)을 전달하는가.
• Scratchpad가 격리되는가. 추론 과정이 다음 에이전트로 새지 않는가.
• 가드레일이 프롬프트가 아니라 코드로 강제되는가. forbidden 작업이 도구 실행 직전 차단되는가.
• 예산 가드가 단계 진입 전 검사되는가.
• 트레이스 로깅(trace_id/agent/input/output(redacted)/tokens)이 첫날부터 있는가.
• 재시도가 일시 실패에만 적용되고 상한·백오프가 있는가. 논리적 실패는 재시도 안 하는가.
• HITL 에스컬레이션이 충분한 컨텍스트와 함께 사람을 부르는가.
• 되돌릴 수 없는 행동(배포/결제/외부발송) 전에 승인 게이트가 있는가.
• 회귀 평가셋이 트레이스로 축적되고 있는가. 프롬프트 변경 시 검증하는가.

마지막 원칙 — Compound Engineering: 잘 설계된 멀티에이전트 시스템은 매 실행이 다음 실행을 더 쉽게 만든다. 검증 루프에서 나온 반려 사유를 재사용 가능한 규칙으로 추출해 축적하고("이 함정은 이렇게 잡는다"), 실패한 트레이스를 회귀셋에 넣어라. 시간이 지날수록 Critic은 과거의 함정을 자동으로 거르고, 같은 실수가 두 번 통과하지 않는다. 오케스트레이션의 진짜 복리는 학습이 시스템에 영구히 남는 구조에서 나온다.