---
title: "오케스트레이션과 멀티 에이전트: 수퍼바이저와 HITL 승인 게이트"
slug: "ai-agent-engineering-04-orchestration"
canonicalUrl: "https://moonshotnotes.com/posts/ai-agent-engineering-04-orchestration/"
sourceUrl: "https://moonshotnotes.com/posts/ai-agent-engineering-04-orchestration/"
markdownUrl: "https://moonshotnotes.com/agent/posts/ai-agent-engineering-04-orchestration.md"
language: "ko"
category: "AI Agent"
updatedAt: "2026-07-26"
agentTokenEstimate: 5724
---

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

단일 에이전트의 한계를 극복하는 그래프 기반 오케스트레이션, 수퍼바이저 라우팅, 전문 서브에이전트 간 동적 핸드오프, 그리고 안전한 프로덕션 배포를 위한 HITL(Human-in-the-Loop) 승인 게이트와 트랜잭션 상태 관리 기법을 상세히 다룹니다.

## Agent metadata

- Source: https://moonshotnotes.com/posts/ai-agent-engineering-04-orchestration/
- Markdown: https://moonshotnotes.com/agent/posts/ai-agent-engineering-04-orchestration.md
- Language: ko
- Category: AI Agent
- Tags: AI Agent, Multi Agent, Orchestration, Supervisor Pattern, Human In The Loop, State Graph, Python
- Updated: 2026-07-26
- Estimated tokens: 5724

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

---

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

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

```mermaid
graph TD
 subgraph SingleAgent["단일 에이전트 (Monolithic Agent)"]
 direction TB
 S_System["모든 전문 지식 프롬프트 비대화<br/>(System Instruction Overhead)"]
 S_Tools["30개 이상의 파편화된 도구 스키마<br/>(Tool Schema Bloat)"]
 S_Loop["거대한 단일 컨텍스트 루프<br/>(Context Contamination)"]
 S_Fail["환각 및 지시 이탈<br/>(Instruction Drift & Compounding Errors)"]

 S_System --> S_Tools --> S_Loop --> S_Fail
 end

 subgraph GraphOrchestration["그래프 오케스트레이션 (Multi-Agent System)"]
 direction TB
 G_Sup["Supervisor Router<br/>(중앙 상태 제어 노드)"]
 G_Res["Research Agent<br/>(독립 검색 컨텍스트)"]
 G_Dev["Developer Agent<br/>(코드 전용 격리 도구)"]
 G_HITL["HITL Gate<br/>(비동기 인터럽트)"]
 G_Ops["Deploy Agent<br/>(격리된 실행 환경)"]

 G_Sup --> |"상태 축소 (State Reduction)" |G_Res
 G_Res --> |"리포트 반환" |G_Sup
 G_Sup --> |"격리 프롬프트 전달" |G_Dev
 G_Dev --> |"패치 검증 요청" |G_HITL
 G_HITL --> |"승인 완료 (Approved)" |G_Ops
 end

 style SingleAgent fill:#2d1b1b,stroke:#f85149,stroke-width:1px
 style GraphOrchestration fill:#1b2d23,stroke:#3fb950,stroke-width:1px
```

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

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

| 병목 유형 | 발생 원인 (Root Cause) | 프로덕션 임팩트 | 그래프 오케스트레이션 해결 방안 |
| :--- | :--- | :--- | :--- |
| 컨텍스트 오염<br/>*(Context Contamination)* | 무분별하게 누적되는 대화 히스토리 및 도구의 Large Payload | LLM 주의력 분산으로 인한 지시 이탈(Instruction Drift) 발생 | 서브에이전트별 독립 스코프 컨텍스트 이관 및 결과 요약 합성 |
| 도구 비대화<br/>*(Tool Schema Overhead)* | 단일 프롬프트에 30개 이상의 OpenAPI 도구 스키마 직접 주입 | 토큰 비용 폭증 및 유사 도구 간 오호출(Mis-selection) 확률 폭증 | 전문 노드별 도구 스키마 격리(Max –개) 배치 |
| 오류 복구 불능<br/>*(Error Compounding)* | 초기 추론 실수가 루프 컨텍스트에 포함되어 계속 재인용 | 잘못된 전제를 바탕으로 환각을 반복하는 컴파운딩 오류에 매몰 | 체크포인트 롤백(Rollback) 및 조건부 에지 기반 Fallback |

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

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

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

| 구분 | 결정론적 워크플로우<br/>*(Deterministic Workflow)* | 그래프 기반 상태 머신<br/>*(Dynamic State Graph)* |
| :--- | :--- | :--- |
| 제어 흐름 | 하드코딩된 Python 순차 실행 (`A -> B -> C`) | 조건부 에지(Conditional Edge) 기반 자율 라우팅 |
| 순환 및 복구 | 루프 및 동적 재시도(Retry/Fallback) 구현 복잡 | 순환 그래프(Cyclic Graph)를 통한 자율 상태 재진입 |
| 상태 불변성 | 함수 파라미터 간 단순 변수 전달 | Pydantic 스키마 기반 전역 상태 저장 및 스냅샷 관리 |
| 인터럽트 지원 | 동기식 대기로 서버 스레드 블로킹 발생 | 비동기 인터럽트(Pause) 및 체크포인트 복원(Resume) |
| 오류 복구력 | 예외 발생 시 전체 프로세스 강제 종료 | 노드 단위 격리 및 롤백 스냅샷 복구 지원 |

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

---

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

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

```mermaid
flowchart TD
 User["User"] --> |1. 과업 요청 ('결제 API 리팩토링 및 배포') |Sup["Sup"]
 Sup["Sup"] --> |2. 핸드오프 (의존성 검색 컨텍스트 스코프) |Res["Res"]
 Res["Res"] --> |3. 코드 분석 리포트 전달 (상태 축소) |Sup["Sup"]
 Sup["Sup"] --> |4. 핸드오프 (리포트 + Dev 전용 도구) |Dev["Dev"]
 Dev["Dev"] --> |5. 패치 파일 생성 완료 |Sup["Sup"]
 Sup["Sup"] --> |6. 배포 승인 인터럽트 발행 (State Checkpointed) |Gate["Gate"]
```

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

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

| 아키텍처 패턴 | 제어 흐름 구조 | 핵심 장점 | 위험 요소 & 단점 | 추천 적용 도메인 |
| :--- | :--- | :--- | :--- | :--- |
| 중앙 수퍼바이저<br/>*(Centralized Router)* | 중앙 라우터 노드가 상태를 검토하고 서브에이전트로 분기 | • 명확한 가시성 및 추적성<br/>• 중앙 집중식 보안/권한 제어 | • 수퍼바이저 프롬프트 성능 병목<br/>• 라우팅 효율에 따른 홉(Hop) 증가 | 엔터프라이즈 업무 자동화,<br/>HITL 승인이 필수적인 시스템 |
| P2P Swarm<br/>*(Dynamic Handoff)* | 서브에이전트가 직접 `handoff_to_agent()` 도구 호출 | • 빠른 실행 속도 (라우터 홉 감소)<br/>• 유연한 동적 자율 협업 | • 무한 순환 호출(Infinite Loop) 위험<br/>• 전체 상태 흐름 추적의 어려움 | 탐색적 연구, 데이터 분석,<br/>자율적 핑퐁이 필요한 창의적 작업 |

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

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

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

| 전달 전략 | 전달 데이터 스코프 | 토큰 효율성 | 모델 주의력 | 오염 방지 능력 |
| :--- | :--- | :--- | :--- | :--- |
| Full Transcript | 전체 대화 히스토리 및 이전 도구 실행 결과 일체 | 토큰 소모 극대화 (비용 폭증) | 불필요 패치에 주의력 분산 | 초기 오류가 계속 전파됨 |
| State Reduction | 정제된 Pydantic State (필수 목표 + 요약 결과만 전달) | 최고 수준의 토큰 절감 | 작업 필수 맥락에 100% 집중 | 서브에이전트 간 완벽 격리 |

```python
# 상태 축소(State Reduction) 메커니즘 예시
class GlobalState(BaseModel):
    task_id: str
    original_goal: str
    plan_steps: list[str]
    research_summary: Optional[str] = None
    generated_code: Optional[str] = None
    approval_status: str = "PENDING"
    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 상태 전환 다이어그램:

```mermaid
stateDiagram-v2
 [*] --> Idle: 그래프 세션 생성

 state SupervisorRouting {
  Idle --> Analyzing: 과업 분해 및 라우팅
  Analyzing --> SubAgentExecution: 서브에이전트 제어권 이관
  SubAgentExecution --> EvaluatingRisk: 결과 수집 및 위험도 평가
 }

 EvaluatingRisk --> ExecutingAction: 저위험 액션 (Safe Action)
 EvaluatingRisk --> InterruptedPending: 고위험 액션 감지 (High-Risk)

 state InterruptedPending {
  [*] --> CheckpointSaved: State 스냅샷 DB 영속화
  CheckpointSaved --> WaitingHumanInput: 웹 서버 스레드 비차단 (Pause)
 }

 state HumanDecision {
  WaitingHumanInput --> Approved: 관리자 승인 (Approve)
  WaitingHumanInput --> Rejected: 관리자 거절 + 수정 피드백 (Reject)
  WaitingHumanInput --> TimedOut: 타임아웃 / 세션 취소 (Timeout)
 }

 Approved --> ExecutingAction: Checkpoint 복원 후 노드 실행
 Rejected --> RePlanning: 피드백 주입 후 Developer 재작업
 RePlanning --> Analyzing: 수퍼바이저 재라우팅
 TimedOut --> RollbackState: 이전 안전 스냅샷으로 롤백
 RollbackState --> [*]: 세션 안전 종료

 ExecutingAction --> Completed: 과업 최종 완료
 Completed --> [*]
```

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

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

| 실행 단계 | 상태 전환 | Checkpoint 저장 내용 | 시스템 스레드 동작 |
| :--- | :--- | :--- | :--- |
| 1. 위험 감지 | `NONE` → `PENDING` | 현재 AgentState snapshot을 DB에 저장 | 위험 노드 실행 직전 비동기 중단 |
| 2. 비동기 대기 | `PENDING` 유지 | DB 스냅샷 저장 상태 유지 (`is_interrupted=True`) | HTTP 서버 응답 반환 후 스레드 해제 |
| 3. 인간 결정 주입 | `PENDING` → `APPROVED` / `REJECTED` | 승인 상태 및 feedback 속성 업데이트 | Webhook / API 요청 수신 시 트리거 |
| 4. 그래프 재개 | `APPROVED` → `NONE` | 복원된 스냅샷 기반 다음 노드 실행 | 비동기 스트림 재연결 및 실행 완수 |

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

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

---

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

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

### 4.1 Async Checkpoint 상태 영속화 시퀀스

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

```mermaid
flowchart TD
 Engine["Engine"] --> |1. save(AgentState) 호출 |Checkpointer["Checkpointer"]
 Checkpointer["Checkpointer"] --> |2. HSET thread_id:tx_9901 state_json |DB["DB"]
 DB["DB"] --> |3. OK (Save Confirmed) |Checkpointer["Checkpointer"]
 Engine["Engine"] --> |4. Stream Event [INTERRUPT] (스레드 비동기 종료) |Client["Client"]
 Admin["Admin"] --> |5. Click 'Approve & Deploy' (Feedback attached) |Client["Client"]
 Client["Client"] --> |6. POST /api/v1/resume (thread_id='tx_9901', approved=True) |Engine["Engine"]
 Engine["Engine"] --> |7. load('tx_9901') |Checkpointer["Checkpointer"]
 Checkpointer["Checkpointer"] --> |8. HGET thread_id:tx_9901 |DB["DB"]
 DB["DB"] --> |9. Raw AgentState JSON Data |Checkpointer["Checkpointer"]
 Checkpointer["Checkpointer"] --> |10. AgentState 객체 역직렬화 복원 |Engine["Engine"]
 Engine["Engine"] --> |11. Mutate State (approval_status=APPROVED) |Engine["Engine"]
 Engine["Engine"] --> |12. Deployer Node 실행 (Resume Execution) |Node["Node"]
 Node["Node"] --> |13. Execution Success |Engine["Engine"]
 Engine["Engine"] --> |14. save(FinalState) |Checkpointer["Checkpointer"]
 Engine["Engine"] --> |15. Stream Event [NODE_END] (과업 완수) |Client["Client"]
```

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

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

| 스트림 이벤트 유형 | 발생 시점 | 주요 페이로드 구성 | Front-end UX 활용 |
| :--- | :--- | :--- | :--- |
| `NODE_START` | 새 상태 노드 진입 시 | `node_name`, `history_len`, `timestamp` | 진행 상황 프로그레스 바 및 에이전트 변경 알림 |
| `NODE_END` | 상태 노드 연산 완료 시 | `node_name`, `next_node`, `state_diff` | 단계 완수 체크마크 표시 및 다음 단계 안내 |
| `INTERRUPT` | HITL Gate 감지 시 | `patch_to_review`, `approval_status` | 사용자 승인 Modal 팝업 및 피드백 입력창 노출 |
| `STREAM_CHUNK` | LLM 토큰 생성 시 | `delta_text`, `agent_id` | 실시간 타이핑 텍스트 모나코 에디터 렌더링 |

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

---

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

다음은 Pydantic v2와 Python `asyncio`를 활용하여 외부 의존성 없이 상태 그래프, 수퍼바이저 동적 라우팅, HITL 인터럽트 게이트, 체크포인트 메모리 스토어, 비동기 이벤트 스트리밍을 완전히 구현한 프로덕션 레벨의 파이썬 예제 코드입니다.

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

```python
"""
Production-grade Multi-Agent State Graph Orchestrator with HITL Gate
Filename: agent_orchestrator.py
"""

import asyncio
from enum import Enum
from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, List, Optional
from pydantic import BaseModel, Field

# -----------------------------------------------------------------------
# 1. State Models & Types
# -----------------------------------------------------------------------

class ApprovalStatus(str, Enum):
    NONE = "NONE"
    PENDING = "PENDING"
    APPROVED = "APPROVED"
    REJECTED = "REJECTED"

class AgentState(BaseModel):
    thread_id: str
    user_query: str
    current_node: str = "supervisor"
    next_node: Optional[str] = None

    # Domain specific data
    research_notes: List[str] = Field(default_factory=list)
    generated_patch: Optional[str] = None

    # Control & HITL Flags
    is_interrupted: bool = False
    approval_status: ApprovalStatus = ApprovalStatus.NONE
    hitl_feedback: Optional[str] = None
    execution_history: List[str] = Field(default_factory=list)

    def log(self, step: str) -> None:
        self.execution_history.append(step)

class StreamEvent(BaseModel):
    event_type: str  # "NODE_START", "NODE_END", "INTERRUPT", "STREAM_CHUNK"
    node_name: str
    payload: Dict[str, Any]

# -----------------------------------------------------------------------
# 2. State Checkpointer & Persistence Store
# -----------------------------------------------------------------------

class AsyncCheckpointer:
    def __init__(self) -> None:
        self._store: Dict[str, AgentState] = {}

    async def save(self, state: AgentState) -> None:
        # Deep copy simulated via model_copy
        self._store[state.thread_id] = state.model_copy(deep=True)

    async def load(self, thread_id: str) -> Optional[AgentState]:
        state = self._store.get(thread_id)
        if state:
            return state.model_copy(deep=True)
        return None

# -----------------------------------------------------------------------
# 3. Subagent Nodes & Logic
# -----------------------------------------------------------------------

async def supervisor_node(state: AgentState) -> AgentState:
    state.log("Supervisor: Assessing task status...")

    if not state.research_notes:
        state.next_node = "researcher"
    elif state.generated_patch is None:
        state.next_node = "developer"
    elif state.approval_status == ApprovalStatus.NONE:
        state.next_node = "hitl_gate"
    elif state.approval_status == ApprovalStatus.APPROVED:
        state.next_node = "deployer"
    elif state.approval_status == ApprovalStatus.REJECTED:
        state.next_node = "developer"  # Re-plan/Re-code on rejection
    else:
        state.next_node = "end"

    return state

async def researcher_node(state: AgentState) -> AgentState:
    state.log("Researcher: Gathering system requirements and codebase context...")
    await asyncio.sleep(0.3)  # Async API work simulation

    state.research_notes.append("Found legacy payment gateway route requiring OAuth2 update.")
    state.research_notes.append("Dependency check: Auth SDK v2.4 compatibility confirmed.")
    state.next_node = "supervisor"
    return state

async def developer_node(state: AgentState) -> AgentState:
    state.log("Developer: Drafting code patch...")
    await asyncio.sleep(0.4)

    if state.approval_status == ApprovalStatus.REJECTED:
        # Refine code based on HITL feedback
        state.generated_patch = f"PATCH v2 (Refactored based on feedback: {state.hitl_feedback})"
        state.approval_status = ApprovalStatus.NONE  # Reset for re-approval
    else:
        state.generated_patch = "PATCH v1: Fix OAuth2 Authorization Header formatting in payment_client.py"

    state.next_node = "supervisor"
    return state

async def hitl_approval_gate_node(state: AgentState) -> AgentState:
    state.log("HITL Gate: Checking safety boundary for code deployment...")

    if state.approval_status == ApprovalStatus.NONE:
        state.log("HITL Gate: High-risk action detected (Deploy)! Interrupting graph execution.")
        state.is_interrupted = True
        state.approval_status = ApprovalStatus.PENDING
        state.next_node = "supervisor"
    elif state.approval_status in (ApprovalStatus.APPROVED, ApprovalStatus.REJECTED):
        state.is_interrupted = False
        state.next_node = "supervisor"

    return state

async def deployer_node(state: AgentState) -> AgentState:
    state.log("Deployer: Applying patch to production environment...")
    await asyncio.sleep(0.5)
    state.log(f"Deployer: Successfully deployed -> {state.generated_patch}")
    state.next_node = "end"
    return state

# -----------------------------------------------------------------------
# 4. State Graph Orchestrator Engine
# -----------------------------------------------------------------------

class StateGraphEngine:
    def __init__(self, checkpointer: AsyncCheckpointer) -> None:
        self.nodes: Dict[str, Callable[[AgentState], Awaitable[AgentState]]] = {}
        self.checkpointer = checkpointer

    def add_node(self, name: str, func: Callable[[AgentState], Awaitable[AgentState]]) -> None:
        self.nodes[name] = func

    async def run_stream(self, thread_id: str, initial_query: Optional[str] = None) -> AsyncGenerator[StreamEvent, None]:
        # Load or initialize state
        state = await self.checkpointer.load(thread_id)
        if state is None:
            if not initial_query:
                raise ValueError("Initial query required for new thread.")
            state = AgentState(thread_id=thread_id, user_query=initial_query)

        while state.current_node != "end":
            current_name = state.current_node
            node_func = self.nodes.get(current_name)

            if not node_func:
                raise KeyError(f"Node '{current_name}' is not registered in state graph.")

            yield StreamEvent(
                event_type="NODE_START",
                node_name=current_name,
                payload={"history_len": len(state.execution_history)}
            )

            # Execute Node
            state = await node_func(state)

            # Check for HITL Interrupt
            if state.is_interrupted:
                await self.checkpointer.save(state)
                yield StreamEvent(
                    event_type="INTERRUPT",
                    node_name=current_name,
                    payload={
                        "message": "Graph paused for human approval.",
                        "patch_to_review": state.generated_patch,
                        "approval_status": state.approval_status.value
                    }
                )
                return  # Exit stream, waiting for external resume trigger

            yield StreamEvent(
                event_type="NODE_END",
                node_name=current_name,
                payload={"next_node": state.next_node}
            )

            state.current_node = state.next_node
            await self.checkpointer.save(state)

    async def resume_with_approval(self, thread_id: str, approved: bool, feedback: Optional[str] = None) -> AsyncGenerator[StreamEvent, None]:
        state = await self.checkpointer.load(thread_id)
        if not state or not state.is_interrupted:
            raise RuntimeError(f"Thread '{thread_id}' is not in an interrupted state.")

        # Inject Human Decision
        state.approval_status = ApprovalStatus.APPROVED if approved else ApprovalStatus.REJECTED
        state.hitl_feedback = feedback
        state.is_interrupted = False
        state.log(f"Human Decision Injected: Approved={approved}, Feedback={feedback}")

        await self.checkpointer.save(state)

        # Resume execution stream
        async for event in self.run_stream(thread_id):
            yield event

# -----------------------------------------------------------------------
# 5. Verification & Execution Walkthrough
# -----------------------------------------------------------------------

async def main() -> None:
    checkpointer = AsyncCheckpointer()
    engine = StateGraphEngine(checkpointer)

    # Register Graph Nodes
    engine.add_node("supervisor", supervisor_node)
    engine.add_node("researcher", researcher_node)
    engine.add_node("developer", developer_node)
    engine.add_node("hitl_gate", hitl_approval_gate_node)
    engine.add_node("deployer", deployer_node)

    thread_id = "tx_session_9901"
    print("=== [1 Phase: Execution until HITL Interrupt] ===")
    async for event in engine.run_stream(thread_id, initial_query="Update payment API route"):
        print(f"[{event.event_type}] Node: {event.node_name} -> {event.payload}")

    # Inspect state at interrupt point
    paused_state = await checkpointer.load(thread_id)
    print(f"\n[System Status] Graph Interrupted? {paused_state.is_interrupted}")
    print(f"[Review Target] Proposed Code: {paused_state.generated_patch}")

    print("\n=== [2 Phase: Simulated Human Rejection & Feedback Loop] ===")
    async for event in engine.resume_with_approval(thread_id, approved=False, feedback="Use Bearer prefix for Auth header"):
        print(f"[{event.event_type}] Node: {event.node_name} -> {event.payload}")

    print("\n=== [3 Phase: Final Human Approval & Deployment] ===")
    async for event in engine.resume_with_approval(thread_id, approved=True):
        print(f"[{event.event_type}] Node: {event.node_name} -> {event.payload}")

    final_state = await checkpointer.load(thread_id)
    print("\n=== [Final Execution Log] ===")
    for idx, log_entry in enumerate(final_state.execution_history, 1):
        print(f"{idx}. {log_entry}")

if __name__ == "__main__":
    asyncio.run(main())
```

---

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

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

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

```mermaid
graph TD
 Start([Start Task]) --> Supervisor[Node: Supervisor Router]

 Supervisor --> |Research Needed |Researcher[Node: Researcher Sub-agent]
 Researcher --> |Return Notes |Supervisor

 Supervisor --> |Code Needed |Developer[Node: Developer Sub-agent]
 Developer --> |Return Patch |Supervisor

 Supervisor --> |Approval Required |HITLGate{Node: HITL Approval Gate}

 HITLGate --> |Pending / Interrupt |CheckpointStore[(Checkpointer Store)]
 HITLGate --> |Approved |Deployer[Node: Deployer Sub-agent]
 HITLGate --> |Rejected w/ Feedback |Developer

 Deployer --> End([Task Finished])

 classDef supervisorStyle fill:#2b3a4a,stroke:#4a90e2,color:#fff,stroke-width:2px;
 classDef gateStyle fill:#4a2b2b,stroke:#e24a4a,color:#fff,stroke-width:2px;
 class Supervisor supervisorStyle;
 class HITLGate gateStyle;
```

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

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

```mermaid
flowchart TD
 Engine["Engine"] --> |1. HITL Gate 노드 진입 (위험 액션 감지) |Engine["Engine"]
 Engine["Engine"] --> |2. AgentState 저장 (is_interrupted=True, thread_id='tx_123') |DB["DB"]
 Engine["Engine"] --> |3. Stream Event: INTERRUPT (응답 종료 및 연결 해제) |API["API"]
 API["API"] --> |4. UI 알림: '배포 승인 대기 중 (Patch v1)' |Inspector["Inspector"]
 Inspector["Inspector"] --> |5. POST /api/v1/agent/resume (Approved=True) |API["API"]
 API["API"] --> |6. engine.resume_with_approval('tx_123', approved=True) |Engine["Engine"]
 Engine["Engine"] --> |7. Load AgentState('tx_123') |DB["DB"]
 DB["DB"] --> |8. State 복원 완료 |Engine["Engine"]
 Engine["Engine"] --> |9. approval_status = APPROVED 갱신 |Engine["Engine"]
 Engine["Engine"] --> |10. Deployer Node 실행 |Node["Node"]
 Node["Node"] --> |11. 배포 성공 완료 |Engine["Engine"]
 Engine["Engine"] --> |12. Stream Event: NODE_END (최종 완료) |API["API"]
```

---

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

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

| 영역 | 항목 | 검증 내용 | 심각도 |
| :--- | :--- | :--- | :--- |
| 아키텍처 | 서브에이전트 단일 책임 | 각 서브에이전트의 역할과 보유 도구가 명확히 격리되어 있는가? (도구 10개 이하 권장) | High |
| 상태 관리 | 상태 축소 (State Reduction) | 수퍼바이저 및 서브에이전트 이동 시 불필요한 전체 대화 히스토리를 요약/제거하는가? | Critical |
| 상태 관리 | 체크포인트 영속성 | 프로세스 재부팅 시에도 대기 중인 인터럽트 상태를 DB에서 안전하게 복원할 수 있는가? | Critical |
| HITL 보안 | 인터럽트 롤백 보장 | 승인 거부(Rejected) 시 이전 안전 노드로 복귀하여 재작업(Re-plan)을 수행하는 흐름이 존재하는가? | Critical |
| HITL 보안 | 인가(Authorization) 게이트 | 승인 신호(Resume Trigger)를 보내는 주체의 권한이 RBAC 시스템과 연동되어 검증되는가? | Critical |
| 운영 / Observability | 무한 루프 감지 (Loop Breaker) | 서브에이전트 간 핸드오프 횟수가 최대 이터레이션(e.g., Max 15 steps)을 초과할 때 강제 종료되는가? | High |
| 운영 / Observability | 이벤트 스트리밍 추적 | Node Transition, Tool Call, LLM Streaming이 OpenTelemetry/LangSmith 트레이스와 동기화되는가? | Medium |

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

---

## 마무리하며

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