오케스트레이션과 멀티 에이전트: 수퍼바이저와 HITL 승인 게이트

'AI 에이전트 엔지니어링(AI Agent Engineering)' 시리즈의 4편입니다. 수퍼바이저, 서브에이전트 핸드오프, HITL 승인 게이트, 상태 그래프를 다룹니다.

---

1. 단일 에이전트의 한계와 그래프 오케스트레이션의 필요성

초기 PoC에서는 모든 도구와 지시를 한 에이전트에 넣기도 합니다. 도구와 작업 범위가 늘어나면 컨텍스트 크기와 도구 선택을 관리하기 어려워질 수 있습니다.

단일 에이전트의 문제 연결은 **모든 전문 지식 프롬프트 비대화(System Instruction Overhead) → 30개 이상의 파편화된 도구 스키마(Tool Schema Bloat) → 거대한 단일 컨텍스트 루프(Context Contamination) → 환각 및 지시 이탈(Instruction Drift & Compounding Errors)**입니다.

그래프 오케스트레이션은 다음 연결을 분리합니다.

1.1 단일 에이전트의 병목 현상

모델의 성능 외에도 컨텍스트에 섞인 정보와 선택할 도구의 수가 실행에 영향을 줍니다.

단일 에이전트 루프에서 발생한 단 한 번의 잘못된 도구 입력이나 파싱 실수가 전체 컨텍스트 히스토리에 전파되면, 후속 추론 단계 전체가 오염되는 컴파운딩 오류(Compounding Errors)가 발생합니다.

1.2 절차적 워크플로우 vs 그래프 기반 상태 머신

이러한 한계를 극복하기 위해 에이전트 아키텍처는 단순 선형 워크플로우(DAG)에서 상태 그래프(Dynamic State Graph)로 진화하고 있습니다.

에이전틱 애플리케이션에서는 분기(Branching)뿐만 아니라 실패 시 이전 상태로 되돌아가는 순환(Looping)과 외부 승인을 기다리는 인터럽트(Interrupt)가 빈번히 발생하므로, 무순환 그래프(DAG) 대신 순환 상태 머신(State Graph)을 기본 아키텍처로 선택해야 합니다.

---

2. 수퍼바이저(Supervisor) 패턴과 동적 핸드오프(Dynamic Handoff)

수퍼바이저(Supervisor)는 전체 목표에 따라 서브에이전트에게 작업을 맡깁니다. 결과를 모은 뒤 다음 작업을 정합니다.

2.1 중앙 수퍼바이저 vs P2P Swarm 네트워크 패턴

서브에이전트 간 제어권을 넘기는 패턴은 크게 중앙 라우터 방식과 자율 핸드오프 방식으로 나뉩니다.

P2P Swarm 아키텍처는 제어권 위임이 자유로운 반면 서브에이전트끼리 서로를 무한 호출하는 핑퐁 루프(Infinite Handoff Loop)가 발생하기 쉽습니다. 반드시 전역 카운터나 루프 브레이커(Loop Breaker)를 탑재해야 합니다.

2.2 상태 전달 전략: Full Transcript vs State Reduction

서브에이전트로 제어권을 이전할 때, 대화 전체 히스토리를 넘기면 단일 에이전트의 컨텍스트 오염 문제가 그대로 재발합니다. 프로덕션 에이전트 엔진은 상태 요약 및 축소(State Reduction) 방식을 채택합니다.

Copy1# 상태 축소(State Reduction) 메커니즘 예시2class GlobalState(BaseModel):3    task_id: str4    original_goal: str5    plan_steps: list[str]6    research_summary: Optional[str] = None7    generated_code: Optional[str] = None8    approval_status: str = "PENDING"9    messages: list[BaseMessage]  # 전역 대화 목록

수퍼바이저가 DeveloperAgent를 호출할 때는 GlobalState.messages 전체가 아니라, original_goal과 research_summary만을 정제하여 프롬프트로 주입함으로써 서브에이전트의 집중도를 최대로 끌어올립니다.

서브에이전트에 격리된 컨텍스트(Scoped Context)를 넘겨주고 작업 결과는 구조화된 상태 갱신값(State Mutation)으로 리턴받아 병합하는 방식이 토큰 비용 절감과 정밀도 향상의 핵심입니다.

---

3. Human-in-the-Loop (HITL) 인터럽트와 승인 게이트

에이전트가 외부 API 호출, DB 레코드 삭제, 금융 결제, 또는 프로덕션 서버 배포와 같이 가역적이지 않은(Irreversible) 고위험 액션을 수행할 때는 반드시 HITL(Human-in-the-Loop) 승인 게이트가 개입해야 합니다.

3.1 HITL Approval Gate 상태 전환 다이어그램:

원본에서 WaitingHumanInput은 InterruptedPending과 HumanDecision 양쪽 범위에 나타납니다. 표에서도 이 중첩 표기를 유지합니다.

3.2 비동기 인터럽트 및 4단계 실행 라이프사이클

HITL을 성공적으로 구현하려면 그래프 실행 엔진이 동기식 대기가 아닌 비동기 중단(Pause)과 재개(Resume) 라이프사이클을 지원해야 합니다.

HITL 인터럽트 구현에서 핵심은 웹 서버 스레드가 비동기로 동작하여 리소스를 블로킹하지 않아야 하며, 무기한 대기 상태에서도 DB에 저장된 체크포인트를 통해 서버 재시작 후 안전하게 복구할 수 있어야 합니다.

HITL Resume 엔드포인트는 악의적인 사용자의 상태 조작 공격 대상이 될 수 있습니다. 반드시 RBAC(Role-Based Access Control) 인가 검증과 수정 불가능한 감사 로그(Audit Log)를 남겨야 합니다.

---

4. 파이썬 트랜잭션 상태 관리와 실시간 스트리밍 인터페이스

멀티 에이전트 상태 머신은 복수의 비동기 노드가 동시 다발적으로 상태를 읽고 쓰는 환경이므로, 트랜잭션 격리와 실시간 스트리밍(Streaming) 구조가 필수적입니다.

4.1 Async Checkpoint 상태 영속화 시퀀스

체크포인터는 그래프의 각 노드 실행 직전/직후의 상태를 영속화하여 비동기 중단 및 장애 발생 시의 복구를 보장합니다.

4.2 이벤트 기반 스트리밍 (Event-driven Streaming)

사용자 경험(UX) 측면에서 멀티 에이전트의 길고 복잡한 연산 과정을 투명하게 보여주기 위해 4가지 레벨의 스트리밍 이벤트를 제공해야 합니다.

상태 변경의 불변성(Immutability)을 보장하기 위해 노드 간 이동 시에는 기존 State를 직접 수정하기보다 복사본을 만들어 갱신(State Mutation) 후 저장하는 트랜잭션 패턴을 사용해야 합니다.

---

5. 프로덕션 레벨 파이썬 구현: 수퍼바이저 그래프 & HITL 승인 엔진

다음 Python asyncio 예시는 Pydantic v2를 사용해 상태 그래프, 수퍼바이저 라우팅, HITL 게이트, 메모리 체크포인트, 이벤트 스트리밍을 연결합니다.

본 코드는 서드파티 프레임워크 없이 Python 순수 코드로 작성되어 LangGraph, AutoGen, CrewAI 등 하부 엔진의 인터럽트 및 스냅샷 복원 동작 원리를 직관적으로 이해할 수 있도록 설계되었습니다.

Copy1"""2Production-grade Multi-Agent State Graph Orchestrator with HITL Gate3Filename: agent_orchestrator.py4"""5 6import asyncio7from enum import Enum8from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, List, Optional9from pydantic import BaseModel, Field10 11# -----------------------------------------------------------------------12# 1. State Models & Types13# -----------------------------------------------------------------------14 15class ApprovalStatus(str, Enum):16    NONE = "NONE"17    PENDING = "PENDING"18    APPROVED = "APPROVED"19    REJECTED = "REJECTED"20 21class AgentState(BaseModel):22    thread_id: str23    user_query: str24    current_node: str = "supervisor"25    next_node: Optional[str] = None26 27    # Domain specific data28    research_notes: List[str] = Field(default_factory=list)29    generated_patch: Optional[str] = None30 31    # Control & HITL Flags32    is_interrupted: bool = False33    approval_status: ApprovalStatus = ApprovalStatus.NONE34    hitl_feedback: Optional[str] = None35    execution_history: List[str] = Field(default_factory=list)36 37    def log(self, step: str) -> None:38        self.execution_history.append(step)39 40class StreamEvent(BaseModel):41    event_type: str  # "NODE_START", "NODE_END", "INTERRUPT", "STREAM_CHUNK"42    node_name: str43    payload: Dict[str, Any]44 45# -----------------------------------------------------------------------46# 2. State Checkpointer & Persistence Store47# -----------------------------------------------------------------------48 49class AsyncCheckpointer:50    def __init__(self) -> None:51        self._store: Dict[str, AgentState] = {}52 53    async def save(self, state: AgentState) -> None:54        # Deep copy simulated via model_copy55        self._store[state.thread_id] = state.model_copy(deep=True)56 57    async def load(self, thread_id: str) -> Optional[AgentState]:58        state = self._store.get(thread_id)59        if state:60            return state.model_copy(deep=True)61        return None62 63# -----------------------------------------------------------------------64# 3. Subagent Nodes & Logic65# -----------------------------------------------------------------------66 67async def supervisor_node(state: AgentState) -> AgentState:68    state.log("Supervisor: Assessing task status...")69 70    if not state.research_notes:71        state.next_node = "researcher"72    elif state.generated_patch is None:73        state.next_node = "developer"74    elif state.approval_status == ApprovalStatus.NONE:75        state.next_node = "hitl_gate"76    elif state.approval_status == ApprovalStatus.APPROVED:77        state.next_node = "deployer"78    elif state.approval_status == ApprovalStatus.REJECTED:79        state.next_node = "developer"  # Re-plan/Re-code on rejection80    else:81        state.next_node = "end"82 83    return state84 85async def researcher_node(state: AgentState) -> AgentState:86    state.log("Researcher: Gathering system requirements and codebase context...")87    await asyncio.sleep(0.3)  # Async API work simulation88 89    state.research_notes.append("Found legacy payment gateway route requiring OAuth2 update.")90    state.research_notes.append("Dependency check: Auth SDK v2.4 compatibility confirmed.")91    state.next_node = "supervisor"92    return state93 94async def developer_node(state: AgentState) -> AgentState:95    state.log("Developer: Drafting code patch...")96    await asyncio.sleep(0.4)97 98    if state.approval_status == ApprovalStatus.REJECTED:99        # Refine code based on HITL feedback100        state.generated_patch = f"PATCH v2 (Refactored based on feedback: {state.hitl_feedback})"101        state.approval_status = ApprovalStatus.NONE  # Reset for re-approval102    else:103        state.generated_patch = "PATCH v1: Fix OAuth2 Authorization Header formatting in payment_client.py"104 105    state.next_node = "supervisor"106    return state107 108async def hitl_approval_gate_node(state: AgentState) -> AgentState:109    state.log("HITL Gate: Checking safety boundary for code deployment...")110 111    if state.approval_status == ApprovalStatus.NONE:112        state.log("HITL Gate: High-risk action detected (Deploy)! Interrupting graph execution.")113        state.is_interrupted = True114        state.approval_status = ApprovalStatus.PENDING115        state.next_node = "supervisor"116    elif state.approval_status in (ApprovalStatus.APPROVED, ApprovalStatus.REJECTED):117        state.is_interrupted = False118        state.next_node = "supervisor"119 120    return state121 122async def deployer_node(state: AgentState) -> AgentState:123    state.log("Deployer: Applying patch to production environment...")124    await asyncio.sleep(0.5)125    state.log(f"Deployer: Successfully deployed -> {state.generated_patch}")126    state.next_node = "end"127    return state128 129# -----------------------------------------------------------------------130# 4. State Graph Orchestrator Engine131# -----------------------------------------------------------------------132 133class StateGraphEngine:134    def __init__(self, checkpointer: AsyncCheckpointer) -> None:135        self.nodes: Dict[str, Callable[[AgentState], Awaitable[AgentState]]] = {}136        self.checkpointer = checkpointer137 138    def add_node(self, name: str, func: Callable[[AgentState], Awaitable[AgentState]]) -> None:139        self.nodes[name] = func140 141    async def run_stream(142        self, thread_id: str, initial_query: Optional[str] = None143    ) -> AsyncGenerator[StreamEvent, None]:144        # Load or initialize state145        state = await self.checkpointer.load(thread_id)146        if state is None:147            if not initial_query:148                raise ValueError("Initial query required for new thread.")149            state = AgentState(thread_id=thread_id, user_query=initial_query)150 151        while state.current_node != "end":152            current_name = state.current_node153            node_func = self.nodes.get(current_name)154 155            if not node_func:156                raise KeyError(f"Node '{current_name}' is not registered in state graph.")157 158            yield StreamEvent(159                event_type="NODE_START",160                node_name=current_name,161                payload={"history_len": len(state.execution_history)}162            )163 164            # Execute Node165            state = await node_func(state)166 167            # Check for HITL Interrupt168            if state.is_interrupted:169                await self.checkpointer.save(state)170                yield StreamEvent(171                    event_type="INTERRUPT",172                    node_name=current_name,173                    payload={174                        "message": "Graph paused for human approval.",175                        "patch_to_review": state.generated_patch,176                        "approval_status": state.approval_status.value177                    }178                )179                return  # Exit stream, waiting for external resume trigger180 181            yield StreamEvent(182                event_type="NODE_END",183                node_name=current_name,184                payload={"next_node": state.next_node}185            )186 187            state.current_node = state.next_node188            await self.checkpointer.save(state)189 190    async def resume_with_approval(191        self, thread_id: str, approved: bool, feedback: Optional[str] = None192    ) -> AsyncGenerator[StreamEvent, None]:193        state = await self.checkpointer.load(thread_id)194        if not state or not state.is_interrupted:195            raise RuntimeError(f"Thread '{thread_id}' is not in an interrupted state.")196 197        # Inject Human Decision198        state.approval_status = ApprovalStatus.APPROVED if approved else ApprovalStatus.REJECTED199        state.hitl_feedback = feedback200        state.is_interrupted = False201        state.log(f"Human Decision Injected: Approved={approved}, Feedback={feedback}")202 203        await self.checkpointer.save(state)204 205        # Resume execution stream206        async for event in self.run_stream(thread_id):207            yield event208 209# -----------------------------------------------------------------------210# 5. Verification & Execution Walkthrough211# -----------------------------------------------------------------------212 213async def main() -> None:214    checkpointer = AsyncCheckpointer()215    engine = StateGraphEngine(checkpointer)216 217    # Register Graph Nodes218    engine.add_node("supervisor", supervisor_node)219    engine.add_node("researcher", researcher_node)220    engine.add_node("developer", developer_node)221    engine.add_node("hitl_gate", hitl_approval_gate_node)222    engine.add_node("deployer", deployer_node)223 224    thread_id = "tx_session_9901"225    print("=== [1 Phase: Execution until HITL Interrupt] ===")226    async for event in engine.run_stream(thread_id, initial_query="Update payment API route"):227        print(f"[{event.event_type}] Node: {event.node_name} -> {event.payload}")228 229    # Inspect state at interrupt point230    paused_state = await checkpointer.load(thread_id)231    print(f"\n[System Status] Graph Interrupted? {paused_state.is_interrupted}")232    print(f"[Review Target] Proposed Code: {paused_state.generated_patch}")233 234    print("\n=== [2 Phase: Simulated Human Rejection & Feedback Loop] ===")235    async for event in engine.resume_with_approval(236        thread_id, approved=False, feedback="Use Bearer prefix for Auth header"237    ):238        print(f"[{event.event_type}] Node: {event.node_name} -> {event.payload}")239 240    print("\n=== [3 Phase: Final Human Approval & Deployment] ===")241    async for event in engine.resume_with_approval(thread_id, approved=True):242        print(f"[{event.event_type}] Node: {event.node_name} -> {event.payload}")243 244    final_state = await checkpointer.load(thread_id)245    print("\n=== [Final Execution Log] ===")246    for idx, log_entry in enumerate(final_state.execution_history, 1):247        print(f"{idx}. {log_entry}")248 249if __name__ == "__main__":250    asyncio.run(main())

---

6. Mermaid 아키텍처 & 시퀀스 다이어그램

6.1 수퍼바이저 & 서브에이전트 그래프 오케스트레이션

시스템 전체의 상태 노드 전환과 수퍼바이저 제어 구조를 시각화한 상태 그래프 다이어그램입니다.

Start Task → Node: Supervisor Router에서 시작합니다.

대기 중 Checkpointer Store에서 실행으로 향하는 연결은 원본에 없습니다. 재개 시퀀스는 다음 절에서 별도로 읽습니다.

6.2 HITL 인터럽트 및 비동기 재개(Resume) 시퀀스

외부 웹/애플리케이션 인터페이스와 에이전트 그래프 엔진 간의 비동기 인터럽트 및 상태 복원(Resume) 과정을 나타낸 시퀀스 다이어그램입니다.

---

7. 멀티 에이전트 오케스트레이션 운영 체크리스트

프로덕션 환경에 수퍼바이저 및 멀티 에이전트 오케스트레이션을 배치하기 전에 반드시 검증해야 할 체크리스트입니다.

프로덕션 멀티 에이전트 오케스트레이션 구축 시 체크포인트 영속성과 무한 루프 방지(Loop Breaker)가 구비되지 않은 상태로 실서버 배포를 진행할 경우, 과도한 LLM API 호출 비용 발생 및 제어 불능 상태에 직면할 위험이 높습니다.

---

마무리하며

에이전트 사이에 작업을 나누려면 전역 상태, 라우팅 규칙, 전달할 컨텍스트를 먼저 정해야 합니다. 승인 대기 상태를 저장하고 재개하는 과정도 같은 실행 흐름 안에서 관리합니다.