Orchestration and Multi-Agent Systems: Supervisor Routing and HITL Approval Gates

This post is Episode 4 of the 'AI Agent Engineering' series. To resolve context contamination and tool bloat in single-agent runtimes, we deeply explore supervisor orchestration, dynamic sub-agent handoffs, Human-in-the-Loop (HITL) approval gates, and transactional State Graph management techniques.

---

1. Limitations of Single Agents and the Need for Graph Orchestration

When introducing LLM-based agents into production systems, early PoC stages often adopt an approach where all tools and instructions are crammed into a single monolithic agent prompt. However, as system complexity increases, a single agent inevitably encounters severe structural limitations.

준비

1. 1subgraph
2. 2direction
3. 3System instruction bloat across all domains
   (System Instruction Overhead)
4. 430+ fragmented tool schemas
   (Tool Schema Bloat)
5. 5Massive monolithic context loop
   (Context Contamination)
6. 6Hallucinations & instruction drift
   (Instruction Drift & Compounding Errors)
7. 7end

반복 실행

1. 1Supervisor Router
   (Central State Control Node)
2. 2Research Agent
   (Isolated Search Context)

"Return Report"

정리

1. 1Developer Agent
   (Code-Dedicated Tool Scope)
2. 2HITL Gate
   (Async Interrupt)
3. 3Deploy Agent
   (Isolated Execution Environment)
4. 4style

1.1 Bottlenecks of Single-Agent Architectures

The failure of single-agent architectures at production scale is not caused by insufficient model capability, but rather by structural context contamination and tool search space explosion.

When a single faulty tool input or parsing error occurs in a single-agent loop and propagates into the full context history, all subsequent reasoning steps become corrupted—leading to severe compounding errors.

1.2 Procedural Workflows vs. Graph-Based State Machines

To overcome these limitations, agent architectures are evolving from simple linear workflows (DAGs) into dynamic state graphs.

In agentic applications, looping to revert to a previous state upon failure and interrupts to await external approval occur frequently alongside simple branching. Therefore, a cyclic state graph—rather than a directed acyclic graph (DAG)—must be chosen as the core architecture.

---

2. Supervisor Pattern and Dynamic Handoff

The most widely used core pattern in multi-agent orchestration is the Supervisor architecture. The supervisor observes the overall goal, delegates control to the appropriate sub-agent, collects results, and determines the next step.

준비

1. 1User

반복 실행

1. 1Supervisor
2. 2Research Agent

"3. Code Analysis Report (State Reduction)"

정리

1. 1Developer Agent
2. 2HITL Gate

2.1 Centralized Supervisor vs. P2P Swarm Network Pattern

Patterns for passing control among sub-agents are broadly divided into centralized router approaches and autonomous handoff approaches.

While P2P Swarm architectures offer freedom in delegating control, sub-agents can easily fall into infinite handoff loops by repeatedly calling each other. A global counter or loop breaker must be incorporated.

2.2 State Transfer Strategy: Full Transcript vs. State Reduction

When transferring control to a sub-agent, passing the full conversation history reintroduces the single-agent context contamination problem. Production agent engines adopt state reduction techniques.

Copy1# Example of State Reduction Mechanism2class 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]  # Global message list

When the supervisor invokes DeveloperAgent, instead of supplying the entire GlobalState.messages, it extracts and injects only original_goal and research_summary into the prompt, maximizing the sub-agent's focus.

Passing a scoped context to a sub-agent and receiving task results back as structured state mutations to merge into global state is key to reducing token costs and boosting precision.

---

3. Human-in-the-Loop (HITL) Interrupts and Approval Gates

When an agent performs irreversible high-risk actions—such as calling external APIs, deleting DB records, executing financial transactions, or deploying to production servers—a Human-in-the-Loop (HITL) approval gate must intervene.

3.1 HITL Approval Gate State Transition Diagram

준비

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 Async Interrupt and 4-Phase Execution Lifecycle

To successfully implement HITL, the graph execution engine must support an asynchronous pause and resume lifecycle rather than synchronous blocking wait.

The core of HITL interrupt implementation is that web server threads must operate asynchronously without blocking resources, and even during indefinite waiting periods, the state must be safely recoverable from DB checkpoints after server restarts.

The HITL Resume endpoint can become a target for state tampering attacks by malicious users. Mandatory RBAC (Role-Based Access Control) authorization checks and immutable audit logging must be enforced.

---

4. Python Transactional State Management and Real-Time Streaming Interface

Because a multi-agent state machine is an environment where multiple asynchronous nodes concurrently read and write state, transactional isolation and real-time streaming architectures are essential.

4.1 Async Checkpoint State Persistence Sequence

The checkpointer persists state immediately before and after each node execution, guaranteeing recovery during async pauses or system failures.

준비

1. 1Engine

반복 실행

1. 1Checkpointer
2. 2DB

"3. OK (Save Confirmed)"

정리

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

4.2 Event-Driven Streaming

To provide transparency into the multi-agent execution process from a UX perspective, four levels of streaming events should be provided.

To guarantee state immutability, transitions between nodes should create and modify a copy of state (State Mutation) before saving, adhering to a transactional state pattern.

---

5. Production-Level Python Implementation: Supervisor Graph & HITL Approval Engine

Below is a production-level Python code example utilizing Pydantic v2 and Python asyncio to build a state graph, supervisor dynamic router, HITL interrupt gate, checkpointer memory store, and async event streaming—without external dependencies.

Written in pure Python without third-party frameworks, this code is designed to provide an intuitive understanding of the underlying interrupt and snapshot restoration mechanics found in engines like LangGraph, AutoGen, and 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("=== [Phase 1: 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=== [Phase 2: 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=== [Phase 3: 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 Architecture & Sequence Diagrams

6.1 Supervisor & Sub-agent Graph Orchestration

State graph diagram visualizing the state node transitions and supervisor control structure across the entire system.

준비

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 Interrupt & Async Resume Sequence

Sequence diagram demonstrating the async interrupt and state restoration (Resume) process between external web/application interfaces and the agent graph engine.

준비

1. 1Engine
2. 2DB

반복 실행

1. 1API
2. 2Inspector

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

정리

1. 1Node

---

7. Multi-Agent Orchestration Operational Checklist

Checklist to validate before deploying supervisor and multi-agent orchestration into production environments.

Deploying multi-agent orchestration to production servers without checkpoint persistence and infinite loop prevention (Loop Breaker) risks incurring excessive LLM API costs and facing out-of-control system execution.

---

Conclusion

Going beyond single-agent limits, graph-based multi-agent orchestration has established itself as the standard architecture for solving complex enterprise domain problems. The key is not merely increasing the number of agents, but designing a clear global state schema, sophisticated supervisor-based routing, isolated sub-agent contexts, and safety-guaranteeing HITL approval gates.

In Episode 5, we will deeply examine Model Context Protocol (MCP), the standard for agent scalability and tool ecosystems, as well as Agent-to-Agent (A2A) communication architectures.