✍️ 필사 모드: 프로덕션 에이전트 설계 패턴 2026 — Supervisor·CodeAct·Plan-Execute·Self-RAG·Handoff·Subagent를 실제로 조합하는 법

                    ┌────────────┐
   user task ────►  │ Supervisor │
                    └─────┬──────┘
                          │ route by intent / state
              ┌───────────┼───────────┐
              ▼           ▼           ▼
       ┌──────────┐ ┌──────────┐ ┌──────────┐
       │ Research │ │ Coder    │ │ Reviewer │
       │ agent    │ │ agent    │ │ agent    │
       └────┬─────┘ └────┬─────┘ └────┬─────┘
            └──────┬─────┴──────┬─────┘
                   ▼            ▼
              return to Supervisor (loop)

# supervisor.py — LangGraph 스타일의 최소 supervisor
from typing import Literal, TypedDict
from langgraph.graph import StateGraph, END

class State(TypedDict):
    task: str
    messages: list
    next: Literal["research", "code", "review", "done"]
    hop: int

def supervisor(state: State) -> dict:
    if state["hop"] >= 6:
        return {"next": "done"}
    prompt = f"Task: {state['task']}\nHistory: {state['messages'][-3:]}\nWho works next? research/code/review/done"
    decision = llm(prompt)  # 강제로 enum 중 하나만 리턴
    return {"next": decision, "hop": state["hop"] + 1}

def specialist(name):
    def run(state: State) -> dict:
        result = run_agent(name, state["task"], state["messages"])
        # 핵심: raw output을 통째로 넣지 말고 요약본만
        return {"messages": state["messages"] + [{"agent": name, "summary": summarize(result)}]}
    return run

g = StateGraph(State)
g.add_node("supervisor", supervisor)
g.add_node("research", specialist("research"))
g.add_node("code", specialist("code"))
g.add_node("review", specialist("review"))
g.set_entry_point("supervisor")
g.add_conditional_edges("supervisor", lambda s: s["next"],
    {"research": "research", "code": "code", "review": "review", "done": END})
for s in ["research", "code", "review"]:
    g.add_edge(s, "supervisor")
graph = g.compile()

[기존 ReAct + JSON tool calls]
Action: {"tool": "search", "args": {"q": "kubernetes pod oom"}}
Observation: [50개 결과]
Action: {"tool": "search", "args": {"q": "kubernetes pod oom limits"}}
Observation: [30개 결과]
Action: {"tool": "summarize", "args": {"text": "..."}}
→ 3 LLM 호출, 3 라운드 트립

[CodeAct]
Action:
```python
results_a = search("kubernetes pod oom")
results_b = search("kubernetes pod oom limits")
merged = list(set(results_a + results_b))[:20]
print(summarize("\n".join(r.snippet for r in merged)))

### 언제 빛나는가

- **여러 툴을 조합해야 하는 태스크.** 데이터 fetch → 변환 → 집계 → 차트 같은 파이프라인이 한 블록에 들어간다.
- **반복이 자연스러운 태스크** — 100개 파일을 처리해야 하는데 JSON tool call로는 100번 라운드 트립이 필요하다. CodeAct는 `for f in files:` 한 번.
- **수치 연산이 섞이는 태스크.** LLM이 직접 산수하지 않고 `numpy` 한 줄로 끝낸다.
- **상태가 누적되는 태스크** — 변수를 통해 이전 스텝의 결과를 자연스럽게 다음 스텝에서 쓸 수 있다.

### 어떻게 무너지는가

- **샌드박스가 새는 순간 보안이 끝난다.** Python 코드를 실행한다는 것은 임의 코드 실행이라는 뜻이다. → 해법: Docker, Firecracker, E2B, gVisor, 혹은 적어도 Python의 `restricted exec` + 시스템콜 필터.
- **`import os; os.system('rm -rf /')` 한 줄로 모든 게 끝난다.** LLM이 이런 코드를 쓰지 않는다는 보장은 없다. → 해법: allowlist 기반 import + 네트워크 격리 + ephemeral 파일시스템.
- **무한 루프.** `while True: search("x")` 한 번이면 토큰 청구서가 폭발한다. → 해법: 실행 타임아웃 + 출력 토큰 캡.
- **디버깅이 어렵다.** 한 코드 블록이 7개 툴을 호출하고 5번째에서 실패하면, 모델은 "왜 실패했는지"를 다시 파악해야 한다. → 해법: 실행 trace를 모델에게 그대로 돌려준다 (line-by-line stdout).

### 최소 스케치

```python
# codeact_executor.py — 격리된 환경에서 모델의 코드를 실행
import subprocess, json, textwrap

ALLOWED_IMPORTS = {"json", "re", "math", "datetime", "tools"}

def execute_codeact(code: str, tools: dict, timeout_s: int = 30) -> str:
    # tools는 모델이 호출할 수 있는 함수 dict — 컨테이너 안에 미리 주입
    wrapper = textwrap.dedent(f"""
        import json, sys
        from tools import {", ".join(tools.keys())}
        try:
        {textwrap.indent(code, "            ")}
        except Exception as e:
            print(f"__ERROR__: {{type(e).__name__}}: {{e}}", file=sys.stderr)
    """)
    result = subprocess.run(
        ["docker", "run", "--rm", "-i", "--network=none",
         "--memory=512m", "--cpus=1.0", "codeact-sandbox:latest", "python", "-"],
        input=wrapper.encode(),
        capture_output=True,
        timeout=timeout_s,
    )
    stdout = result.stdout.decode()[-4000:]  # 출력 캡
    stderr = result.stderr.decode()[-1000:]
    return f"STDOUT:\n{stdout}\n\nSTDERR:\n{stderr}"

# 모델 루프
while not done:
    code = llm_extract_codeact(messages)  # ```python ... ``` 블록만 뽑기
    obs = execute_codeact(code, tools)
    messages.append({"role": "user", "content": f"Execution result:\n{obs}"})

[Plan phase]                       [Execute phase]
┌──────────────┐                   ┌──────────────┐
│  Planner LLM │  →  step list  →  │ Executor LLM │  →  results
│  (one shot)  │                   │  (per step)  │
└──────────────┘                   └──────┬───────┘
                                          │
                  optional re-plan  ◄─────┘
                  (if step fails)

# plan_execute.py
def plan(task: str, tools_schema: dict) -> list[dict]:
    prompt = f"""You will be given a task. Decompose it into 3-7 ordered steps.
Each step: {{"id": int, "goal": str, "tool": str | null, "depends_on": [int]}}
Available tools: {list(tools_schema.keys())}
Task: {task}
Return a JSON array."""
    return json.loads(llm(prompt, temperature=0))

def execute_step(step: dict, prior_results: dict) -> dict:
    context = "\n".join(f"step{i}: {prior_results[i]}" for i in step["depends_on"])
    return run_executor_agent(step["goal"], context, tool=step["tool"])

def plan_execute(task: str, tools_schema: dict, max_replans: int = 2):
    steps = plan(task, tools_schema)
    results, replans = {}, 0
    i = 0
    while i < len(steps):
        try:
            results[steps[i]["id"]] = execute_step(steps[i], results)
            i += 1
        except UnexpectedObservation as e:
            if replans >= max_replans: raise
            steps = plan(f"{task}\nSo far: {results}\nProblem: {e}", tools_schema)
            replans += 1
            i = 0  # 새 plan으로 다시
    return results

                        ┌─────────────┐
   query  ───────────►  │ Route LLM   │
                        └──────┬──────┘
                  retrieve?    │     direct?
              ┌────────────────┴────────────────┐
              ▼                                 ▼
        ┌──────────┐                      ┌──────────┐
        │ Retrieve │                      │ Answer   │
        └────┬─────┘                      │ directly │
             ▼                            └──────────┘
        ┌──────────┐
        │ Grade    │  ──→  irrelevant  ──→  rewrite query, retry
        │ docs LLM │  ──→  partial     ──→  web search supplement
        └────┬─────┘  ──→  relevant    ──→  answer
             ▼
        ┌──────────┐
        │ Answer   │  ──→  hallucinated?  ──→  retry
        │ + check  │  ──→  grounded      ──→  return
        └──────────┘

# self_rag.py
def adaptive_rag(query: str, max_hops: int = 3):
    route = llm_classify(query, choices=["direct", "retrieve"])
    if route == "direct":
        return llm_answer(query)

    for hop in range(max_hops):
        docs = retrieve(query, k=5)
        grade = llm_grade(query, docs)  # {"relevant", "partial", "irrelevant"}
        if grade == "relevant":
            answer = llm_answer_with_context(query, docs)
            if llm_check_grounding(answer, docs) == "grounded":
                return answer
            # 환각이면 재시도
            continue
        if grade == "partial":
            web_docs = web_search(query, k=3)
            return llm_answer_with_context(query, docs + web_docs)
        # irrelevant → rewrite
        query = llm_rewrite(query, hint=f"prior docs were off-topic: {summarize(docs)}")
    return llm_answer_with_context(query, docs)  # 마지막 시도

     ┌──────────────────────────────────────┐
     │                                      │
     ▼                                      │
  Actor  ──► Action ──► Env ──► Reward      │
  (LLM)                                     │
     ▲                                      │
     │                                      │
  Evaluator (success/fail판단)              │
     │                                      │
     ▼                                      │
  Self-Reflection LLM ──► "why I failed" ───┘
                          (memory buffer)

# reflexion.py
def reflexion_loop(task: str, max_trials: int = 5):
    reflections = []
    for trial in range(max_trials):
        sys_prompt = base_prompt + "\n\nPast reflections:\n" + "\n".join(reflections[-3:])
        result = actor_llm(task, system=sys_prompt)
        evaluation = evaluator(result, task)  # {"success", "fail", reason}
        if evaluation["status"] == "success":
            return result
        reflection = reflect_llm(task=task, attempt=result, why_failed=evaluation["reason"])
        # 핵심: 중복 reflection 제거
        if reflection not in reflections:
            reflections.append(reflection)
        else:
            # 같은 깨달음 반복 → 강제 변형
            reflection = reflect_llm(task=task, attempt=result, hint="try a completely different approach")
            reflections.append(reflection)
    return result  # 마지막 시도 반환

[Supervisor]                       [Handoff]
   ┌─────────┐                        A
   │   Sup   │                        │ handoff_to(B)
   └──┬──┬──┬┘                        ▼
      ▼  ▼  ▼                         B
      A  B  C                         │ handoff_to(C)
                                      ▼
중앙집중 결정                          C
"누가 다음에?"                        분산 결정
                                      "다음은 누구한테 넘길지 내가 안다"

# swarm.py — OpenAI Swarm 스타일
def make_agent(name, instructions, tools, handoffs):
    return {
        "name": name,
        "instructions": instructions,
        "tools": tools + [{"name": f"transfer_to_{h.name}", "fn": lambda h=h: h} for h in handoffs],
    }

triage = make_agent("triage",
    "Route user to billing or tech_support based on intent.",
    tools=[], handoffs=[billing, tech])

def run(agent, user_msg, history):
    while True:
        history.append({"role": "user", "content": user_msg})
        out = llm(agent["instructions"], history)
        if out.tool_call and out.tool_call.name.startswith("transfer_to_"):
            agent = out.tool_call.fn()  # handoff!
            continue
        return out, history

user msg
    │
    ▼
┌─────────────┐
│ Intent      │   ← 작은 BERT 분류기, 혹은 regex 룰, 혹은 Haiku 같은 싼 LLM
│ classifier  │
└──┬─────┬────┘
   │     │
   ▼     ▼
agent_a  agent_b  ... agent_n
(각자 자기 system prompt, 자기 도구만)

# routing.py
INTENT_AGENTS = {
    "billing": billing_agent,
    "shipping": shipping_agent,
    "account": account_agent,
    "general": general_agent,
}

def classify(text: str) -> tuple[str, float]:
    # 작은 분류기 — bert-base finetune 또는 Haiku zero-shot
    logits = small_classifier(text)
    intent, conf = top1(logits)
    return intent, conf

def route(user_msg: str):
    intent, conf = classify(user_msg)
    if conf < 0.7 or intent == "uncertain":
        return general_agent(user_msg)  # fallback
    return INTENT_AGENTS[intent](user_msg)

[main agent context]
  ┌────────────────────────────────────────────────┐
  │ user task, system prompt, history (240k tok)   │
  │                                                │
  │   ── spawn(subagent, "summarize 18 PRs") ──►   │
  │                                                │
  │            [subagent context]                  │
  │            ┌───────────────────┐               │
  │            │ fresh window      │               │
  │            │ subtask + tools   │               │
  │            │ (returns: 400 tok)│               │
  │            └─────────┬─────────┘               │
  │                      │                         │
  │   ◄────── result summary (400 tok) ──          │
  │                                                │
  │ continues main task (context grew by 400 tok)  │
  └────────────────────────────────────────────────┘

# subagent.py
def spawn_subagent(parent_state, subtask: str, tools: list,
                   max_tokens: int = 8000, depth: int = 0):
    if depth >= 2:
        raise RuntimeError("subagent depth limit reached")
    sub_context = {
        "system": SUB_AGENT_SYSTEM,  # 부모와 다른 system prompt
        "messages": [{"role": "user", "content": subtask}],
    }
    while not done(sub_context) and tokens(sub_context) < max_tokens:
        sub_context = step(sub_context, tools, depth=depth + 1)
    summary = final_summary(sub_context)  # 자식이 자기 결과를 짧게 요약
    return summary  # 부모는 이것만 받는다

# 부모 루프 안에서
if parent_decides_to_delegate():
    summary = spawn_subagent(state, "summarize the 18 open PRs in the repo", tools=[git_tool, gh_tool])
    state["messages"].append({"role": "tool", "name": "subagent", "content": summary})

       ┌──────────────────────────────────────┐
       │              Agent loop              │
       │                                      │
 user  │  ┌──── UserPromptSubmit hook ───┐    │
 msg ──┼─►│ check policy, strip secrets   │   │
       │  └──┬────────────────────────────┘   │
       │     ▼                                │
       │  ┌──── LLM step ──────────┐          │
       │  │ propose tool call      │          │
       │  └──┬─────────────────────┘          │
       │     ▼                                │
       │  ┌── PreToolUse hook ────┐           │
       │  │ allow? deny? mutate?  │           │
       │  └──┬────────────────────┘           │
       │     ▼                                │
       │  ┌── Tool execution ─────┐           │
       │  └──┬────────────────────┘           │
       │     ▼                                │
       │  ┌── PostToolUse hook ───┐           │
       │  │ redact, audit         │           │
       │  └──┬────────────────────┘           │
       │     ▼                                │
       │  loop                                │
       └──────────────────────────────────────┘

# hooks.py
def pre_tool_hook(tool: str, args: dict, state) -> tuple[bool, str | None]:
    if tool == "bash":
        cmd = args.get("command", "")
        if "rm -rf" in cmd or "curl http" in cmd:
            return False, "blocked: dangerous command pattern"
        if "$SECRET" in cmd or re.search(r"sk-[a-zA-Z0-9]{20,}", cmd):
            return False, "blocked: secret leak detected"
    if tool == "write_file":
        if not args["path"].startswith(state["allowed_root"]):
            return False, f"blocked: outside allowed root {state['allowed_root']}"
    return True, None

def agent_step(state):
    proposal = llm_propose_action(state)
    allowed, reason = pre_tool_hook(proposal.tool, proposal.args, state)
    if not allowed:
        state["messages"].append({"role": "tool", "content": f"BLOCKED: {reason}"})
        return state
    result = execute(proposal)
    audit_log(proposal, result)  # post hook
    state["messages"].append({"role": "tool", "content": result})
    return state

[user task]
    │
    ▼
┌────────────────────────────────────────────────┐
│  Main agent (loop) — Plan-Execute              │
│   - Reads task, builds a 3-7 step plan         │
│   - Per step:                                  │
│      ┌─── PreToolUse hook ────┐                │
│      │ permission + policy    │                │
│      └─────────┬──────────────┘                │
│                ▼                               │
│      ┌─── Tool / CodeAct exec ────┐            │
│      └─────────┬──────────────────┘            │
│                ▼                               │
│      ┌─── PostToolUse hook ───┐                │
│      │ audit, redact          │                │
│      └─────────┬──────────────┘                │
│                ▼                               │
│      need deep exploration? ──► spawn subagent │
│                                  (isolated ctx)│
│                ▼                               │
│   - On step failure: Reflexion-style retry     │
└────────────────────────────────────────────────┘

[user msg]
    │
    ▼
┌──── Specialist routing (intent classifier) ────┐
│       ▼          ▼         ▼          ▼        │
│   billing    shipping   account    general     │
│      │          │         │           │        │
└──────┼──────────┼─────────┼───────────┼────────┘
       ▼          ▼         ▼           ▼
     [each specialist: Self-RAG over internal KB]
       │
       ▼
     can resolve? ──── yes ──► reply
       │
       no
       ▼
     Handoff to human OR another specialist

[user question over CSV / DB]
    │
    ▼
Supervisor (Manus-style)
    │
    ├─► Researcher subagent  (subagent isolation)
    │     └─ Self-RAG over schemas/docs
    │
    ├─► Analyst subagent
    │     └─ CodeAct (writes pandas / SQL in Python blocks)
    │     └─ executes in sandbox with Hooks (network=none, fs scoped)
    │
    └─► Reporter subagent
          └─ writes markdown summary

[user goal: "book a flight"]
    │
    ▼
Plan phase: high-level plan (find site → search → fill form → confirm)
    │
    ▼
Execute phase: action loop
   - screenshot
   - decide click/type/key
   - PreActionHook: dangerous-zone? (banking, settings) → halt + ask user
   - execute
   - PostActionHook: screenshot diff
   - on unexpected screen → Reflexion-style replan

                       단순  | 중간복잡 | 매우복잡
입문서                  ReAct만 | ReAct+Reflection | (벤치마크 거기서 멈춤)
실제 프로덕션 추천      Routing+ | Plan-Exec+    | Supervisor+Subagent+
                       Self-RAG | Hooks+Subagent| CodeAct+Hooks+Reflexion