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

본 포스트는 'AI 에이전트 엔지니어링(AI Agent Engineering)' 시리즈의 4편입니다. 단일 에이전트 런타임의 컨텍스트 왜곡과 도구 비대화 문제를 해결하기 위해, 수퍼바이저 오케스트레이션, 동적 서브에이전트 핸드오프, HITL(Human-in-the-Loop) 승인 게이트, 그리고 트랜잭션 그래프 상태(State Graph) 관리 기법을 깊이 있게 탐구합니다.

---

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

LLM 기반 에이전트를 프로덕션 시스템에 도입할 때 초기 PoC 단계에서는 모든 도구(Tool)와 지시사항(Instruction)을 단일 에이전트(Single Monolithic Agent) 프롬프트에 몰아넣는 방식을 취하곤 합니다. 하지만 시스템의 복잡도가 증가함에 따라 단일 에이전트는 필연적으로 구조적 한계에 부딪히게 됩니다.

준비

1. 1subgraph
2. 2direction
3. 3모든 전문 지식 프롬프트 비대화
   (System Instruction Overhead)
4. 430개 이상의 파편화된 도구 스키마
   (Tool Schema Bloat)
5. 5거대한 단일 컨텍스트 루프
   (Context Contamination)
6. 6환각 및 지시 이탈
   (Instruction Drift & Compounding Errors)
7. 7end

반복 실행

1. 1Supervisor Router
   (중앙 상태 제어 노드)
2. 2Research Agent
   (독립 검색 컨텍스트)

"리포트 반환"

정리

1. 1Developer Agent
   (코드 전용 격리 도구)
2. 2HITL Gate
   (비동기 인터럽트)
3. 3Deploy Agent
   (격리된 실행 환경)
4. 4style

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

단일 에이전트 아키텍처가 프로덕션 스케일에서 실패하는 원인은 단순한 모델 성능 부족이 아니라 구조적 컨텍스트 오염과 도구 탐색 공간 폭발에 있습니다.

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

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

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

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

---

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

멀티 에이전트 오케스트레이션에서 가장 널리 쓰이는 핵심 패턴은 수퍼바이저(Supervisor) 아키텍처입니다. 수퍼바이저는 전체 목표를 관조하며 적절한 서브에이전트에게 제어권을 위임하고, 결과를 수집하여 다음 작업을 판단합니다.

준비

1. 1User

반복 실행

1. 1Sup
2. 2Res

3. 코드 분석 리포트 전달 (상태 축소)

정리

1. 1Dev
2. 2Gate

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 상태 전환 다이어그램:

준비

1. 1state
2. 2Idle
3. 3Analyzing
4. 4SubAgentExecution
5. 5EvaluatingRisk

반복 실행

1. 1ExecutingAction
2. 2InterruptedPending
3. 3CheckpointSaved
4. 4WaitingHumanInput
5. 5Approved

실패 시 다시 실행 단계로 돌아갑니다.

정리

1. 1Rejected
2. 2TimedOut
3. 3RePlanning
4. 4RollbackState
5. 5Completed

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

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

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

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

---

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

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

4.1 Async Checkpoint 상태 영속화 시퀀스

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

준비

1. 1Engine

반복 실행

1. 1Checkpointer
2. 2DB

3. OK (Save Confirmed)

정리

1. 1Client
2. 2Admin
3. 3Node

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

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

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

---

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

다음은 Pydantic v2와 Python asyncio를 활용하여 외부 의존성 없이 상태 그래프, 수퍼바이저 동적 라우팅, 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(self, thread_id: str, initial_query: Optional[str] = None) -> AsyncGenerator[StreamEvent, None]:142        # Load or initialize state143        state = await self.checkpointer.load(thread_id)144        if state is None:145            if not initial_query:146                raise ValueError("Initial query required for new thread.")147            state = AgentState(thread_id=thread_id, user_query=initial_query)148 149        while state.current_node != "end":150            current_name = state.current_node151            node_func = self.nodes.get(current_name)152 153            if not node_func:154                raise KeyError(f"Node '{current_name}' is not registered in state graph.")155 156            yield StreamEvent(157                event_type="NODE_START",158                node_name=current_name,159                payload={"history_len": len(state.execution_history)}160            )161 162            # Execute Node163            state = await node_func(state)164 165            # Check for HITL Interrupt166            if state.is_interrupted:167                await self.checkpointer.save(state)168                yield StreamEvent(169                    event_type="INTERRUPT",170                    node_name=current_name,171                    payload={172                        "message": "Graph paused for human approval.",173                        "patch_to_review": state.generated_patch,174                        "approval_status": state.approval_status.value175                    }176                )177                return  # Exit stream, waiting for external resume trigger178 179            yield StreamEvent(180                event_type="NODE_END",181                node_name=current_name,182                payload={"next_node": state.next_node}183            )184 185            state.current_node = state.next_node186            await self.checkpointer.save(state)187 188    async def resume_with_approval(self, thread_id: str, approved: bool, feedback: Optional[str] = None) -> AsyncGenerator[StreamEvent, None]:189        state = await self.checkpointer.load(thread_id)190        if not state or not state.is_interrupted:191            raise RuntimeError(f"Thread '{thread_id}' is not in an interrupted state.")192 193        # Inject Human Decision194        state.approval_status = ApprovalStatus.APPROVED if approved else ApprovalStatus.REJECTED195        state.hitl_feedback = feedback196        state.is_interrupted = False197        state.log(f"Human Decision Injected: Approved={approved}, Feedback={feedback}")198 199        await self.checkpointer.save(state)200 201        # Resume execution stream202        async for event in self.run_stream(thread_id):203            yield event204 205# -----------------------------------------------------------------------206# 5. Verification & Execution Walkthrough207# -----------------------------------------------------------------------208 209async def main() -> None:210    checkpointer = AsyncCheckpointer()211    engine = StateGraphEngine(checkpointer)212 213    # Register Graph Nodes214    engine.add_node("supervisor", supervisor_node)215    engine.add_node("researcher", researcher_node)216    engine.add_node("developer", developer_node)217    engine.add_node("hitl_gate", hitl_approval_gate_node)218    engine.add_node("deployer", deployer_node)219 220    thread_id = "tx_session_9901"221    print("=== [1 Phase: Execution until HITL Interrupt] ===")222    async for event in engine.run_stream(thread_id, initial_query="Update payment API route"):223        print(f"[{event.event_type}] Node: {event.node_name} -> {event.payload}")224 225    # Inspect state at interrupt point226    paused_state = await checkpointer.load(thread_id)227    print(f"\n[System Status] Graph Interrupted? {paused_state.is_interrupted}")228    print(f"[Review Target] Proposed Code: {paused_state.generated_patch}")229 230    print("\n=== [2 Phase: Simulated Human Rejection & Feedback Loop] ===")231    async for event in engine.resume_with_approval(thread_id, approved=False, feedback="Use Bearer prefix for Auth header"):232        print(f"[{event.event_type}] Node: {event.node_name} -> {event.payload}")233 234    print("\n=== [3 Phase: Final Human Approval & Deployment] ===")235    async for event in engine.resume_with_approval(thread_id, approved=True):236        print(f"[{event.event_type}] Node: {event.node_name} -> {event.payload}")237 238    final_state = await checkpointer.load(thread_id)239    print("\n=== [Final Execution Log] ===")240    for idx, log_entry in enumerate(final_state.execution_history, 1):241        print(f"{idx}. {log_entry}")242 243if __name__ == "__main__":244    asyncio.run(main())

---

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

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

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

준비

1. 1Start

반복 실행

1. 1Supervisor
2. 2Researcher

Return Notes

정리

1. 1Developer
2. 2Node: HITL Approval Gate
3. 3CheckpointStore
4. 4Deployer
5. 5End
6. 6classDef
7. 7class

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

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

준비

1. 1Engine
2. 2DB

반복 실행

1. 1API
2. 2Inspector

5. POST /api/v1/agent/resume (Approved=True)

정리

1. 1Node

---

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

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

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

---

마무리하며

단일 에이전트의 한계를 넘어선 그래프 기반 멀티 에이전트 오케스트레이션은 복잡한 엔터프라이즈 도메인 문제를 해결하기 위한 표준 아키텍처로 자리잡았습니다. 핵심은 단지 에이전트의 개수를 늘리는 것이 아니라, 명확한 전역 상태 스키마, 수퍼바이저 기반의 정교한 라우팅, 격리된 서브에이전트 컨텍스트, 그리고 안전성을 보장하는 HITL 인터럽트 게이트를 설계하는 것입니다. 다음 5편에서는 에이전트의 확장성과 도구 생태계 표준인 와 에이전트 간 통신 표준인 A2A(Agent-to-Agent) 아키텍처를 상세히 다룹니다.