✍️ 필사 모드: 本番エージェント設計パターン 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": "..."}}
→ LLM 呼び出し 3 回、ラウンドトリップ 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 → 変換 → 集計 → グラフ化が 1 ブロックに収まる。
- **反復が自然な** タスク。100 ファイルの処理を JSON tool call でやると 100 回のラウンドトリップが必要。CodeAct なら `for f in files:` 一行。
- **算術が混ざる** タスク。LLM 自身が計算せず、`numpy` 一行で終わる。
- **状態が積み上がる** タスク。変数を介して前ステップの結果が次ステップに自然に流れる。

### どう崩れるか

- **サンドボックスが漏れた瞬間、セキュリティは終わる。** Python を実行するということは任意コード実行ということ。対策：Docker、Firecracker、E2B、gVisor、最低でも Python の `restricted exec` + syscall フィルタ。
- **`import os; os.system('rm -rf /')` の一行で全てが終わる。** LLM がそれを書かない保証はない。対策：import の allowlist + ネットワーク隔離 + ephemeral filesystem。
- **無限ループ。** `while True: search("x")` 一発でトークン請求書が爆発する。対策：実行タイムアウト + 出力トークン上限。
- **デバッグが難しい。** 1 ブロックが 7 ツールを呼び、5 番目で失敗すると、モデルは「何が起きたか」を再構築する必要がある。対策：実行トレースをそのままモデルに返す（行ごとの 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)  # status: success|fail, plus reason
        if evaluation["status"] == "success":
            return result
        reflection = reflect_llm(task=task, attempt=result, why_failed=evaluation["reason"])
        # 重複除去
        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 分類器、正規表現、もしくは Haiku のような安い LLM
│ classifier  │
└──┬─────┬────┘
   │     │
   ▼     ▼
agent_a  agent_b  ... agent_n
（各自が自分のシステムプロンプトとツールを持つ）

# 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)  # フォールバック
    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,  # 親とは別のシステムプロンプト
        "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+Refl    | （ベンチマークここで止まる）
実本番推奨              Routing| Plan-Exec+    | Supervisor+Subagent+
                       +Self-  | Hooks+        | CodeAct+Hooks+Reflexion
                        RAG    | Subagent      |