Supervisor（主管）模式就是对上面问题的经典答案：设一个"中央主管"，它负责接任务、判断该派给哪个"下属"去做、收集结果，下属之间不直接对话，全部经过主管。

LangChain 官方出过一个 langgraph-supervisor 库。但本框架默认不用它，选择完全自实现。理由直接写在源码的模块 docstring 里——dispatch.py:11-13：

"""supervisor_node · 自实 LLM dispatch · zero deps on `langgraph-supervisor` lib
...
revise 2026-05-22:
- LLM default 走 toolkit get_llm("haiku") · 不强 import langchain_anthropic
- 完全自实 · 不依赖 langgraph-supervisor lib(v0.1 不稳 + venv 未装)
"""

逐句读："v0.1 不稳"——那个库当时还是 0.1 版本，行为不稳定；"venv 未装"——环境里根本没装它，硬依赖会让整个包 import 就炸。所以框架的选择是：用最原始的 LangGraph StateGraph 自己拼一个 supervisor，只有 100 多行，行为完全可控。

要读懂图内 supervisor，先看它读写的"黑板"（回忆 Day 03 的 State）。它定义在 state.py:24，只有三个字段：

class SupervisorState(TypedDict, total=False):          # state.py:24
    # LangGraph messages reducer · 累积对话
    messages: Annotated[list[Any], add_messages]        # state.py:34
    # supervisor 调度历史 · ["rca", "trace", "rca"] 形式
    dispatch_history: list[str]                          # state.py:37
    # 终止时填 · None 表示仍在 dispatch loop
    final_answer: str | None                             # state.py:39

total=False 表示这三个字段都是可选的（黑板一开始可以是空的，节点逐步往上写）。source 里还有一段很务实的兜底（state.py:15-21）：如果环境没装 langgraph，就退化成一个只做 append 的 add_messages 假实现，保证这个 schema"光 import 也不炸"。

代码在 packages/ai-trust-toolkit-starter-supervisor/.../dispatch.py。它的活就一件：看当前黑板，让便宜 LLM 决定"下一个调谁"或"给最终答案"。我们分四段读。

① 主管给 LLM 的"任务书"（dispatch.py:29）

def _build_supervisor_system_prompt(specialist_names: list[str]) -> str:
    names_str = ", ".join(f'"{n}"' for n in specialist_names)
    return (
        "你是一个 supervisor 节点 · ... 输出 JSON。可选 specialist:[" + names_str + "]。\n"
        'JSON 输出格式(只输出 JSON · 不附其他文本):\n'
        '  - 继续调度 specialist:{"next": "<specialist_name>"}\n'
        '  - 终止 + 给最终答案:{"final_answer": "..."}\n'
        "调度策略:\n"
        "- 若 dispatch_history 已含同一 specialist 多次且无新信息 · 终止 final_answer")

大白话：主管严格约束 LLM 只能输出两种 JSON 之一——要么 {"next":"trace"}（继续，调这个专家），要么 {"final_answer":"..."}（够了，收尾）。它还把"同一个专家反复调又没新信息就该收尾"这条软规则写进了提示词，从源头减少打转。

② 防死循环的硬闸门（dispatch.py:100-115）

dispatch_history = state.get("dispatch_history", [])
# 防无限循环 · max_iter 兜底
if len(dispatch_history) >= max_iter:                 # dispatch.py:103
    log.warning("supervisor max_iter=%d reached · 强制 final_answer", max_iter)
    return {
        "final_answer": f"达到最大调度次数(max_iter={max_iter})· 强制终止 · ...",
        "next": END_MARKER,                            # dispatch.py:26 END_MARKER="__END__"
    }

这是今天第一个关键的边界处理：上面提示词那条软规则是"劝"LLM 别打转，但 LLM 不一定听话。所以这里有一道硬闸门：调度历史长度一旦到 max_iter（默认 10），不管 LLM 想怎样，直接强制填 final_answer 收尾。"劝 + 硬拦"双保险，是所有 LLM 循环控制的标准姿势。

③ 便宜 LLM 出决策，异步优先（dispatch.py:117-146）

if llm is None:
    from ai_trust_toolkit.llm import get_llm
    actual_llm = get_llm("haiku")            # 默认用便宜的 Haiku 做调度 · dispatch.py:121
specialist_names = sorted(specialists.keys())      # 排序 → 提示词稳定 · 可缓存
...
ainvoke = getattr(actual_llm, "ainvoke", None)
if ainvoke is not None and callable(ainvoke):
    response = await ainvoke(full_messages)   # 有异步就走异步 · dispatch.py:144
else:
    response = actual_llm.invoke(full_messages)

两个细节：（1）默认 get_llm("haiku")——调度只是"选下一步"，不需要贵模型，用便宜的 Haiku 省钱（这是全框架的成本纪律，Day 11 细讲）。（2）sorted(specialists.keys()) 把专家名排序，保证同一批专家生成的提示词字节完全一致——这对prompt 缓存命中很重要。

④ 解析结果 + 三种出路（dispatch.py:158-183）

parsed = _parse_dispatch_response(content)
# 优先看 final_answer
if "final_answer" in parsed and parsed["final_answer"]:
    return {"final_answer": str(parsed["final_answer"]), "next": END_MARKER}
next_name = parsed.get("next")
if next_name in specialists:
    return {"next": next_name}                        # 正常：派给这个专家
# LLM 给了 next 但不在名单 · 视为终止（不硬塞一个不存在的专家）
log.warning("supervisor LLM 返 next=%r 不在 specialists ...", next_name)
return {"final_answer": f"supervisor 选择无效 specialist={next_name!r} ...", "next": END_MARKER}

三条出路：有 final_answer → 收尾；next 在名单里 → 派专家；LLM 幻觉出一个不存在的专家名 → 不冒险执行，直接当收尾处理（第 3 条又是一处边界防御）。

光有 supervisor_node 还不是一张图。auto_config.py:37 的 auto_configure_supervisor() 一行把整张图连好：

def auto_configure_supervisor(app, *, specialists, llm=None,
                              enable_critic=True, max_iter=10):
    if not enable_critic:                              # auto_config.py:65 红线
        raise ValueError("R-STARTER-SUPERVISOR-2 红线 · critic 不可关 ...")
    if not specialists:
        raise ValueError("specialists dict 不能为空 · 至少 1 个 specialist")

    builder = StateGraph(SupervisorState)
    builder.add_node("supervisor", _supervisor)        # 主管节点
    for name, runnable in specialists.items():
        builder.add_node(name, runnable)
        builder.add_edge(name, "supervisor")           # 专家跑完 → 回主管
    builder.add_edge(START, "supervisor")              # 入口
    # 主管出口是条件边：路由到某专家，或到 END
    conditional_map = {name: name for name in specialists}
    conditional_map[END_MARKER] = END                  # __END__ → LangGraph END
    builder.add_conditional_edges("supervisor", route_after_supervisor, conditional_map)
    return builder.compile()

逐行翻译：建一张以 SupervisorState 为黑板的图 → 加主管节点 → 每个专家加成节点，并且专家跑完统一回到主管（add_edge(name,"supervisor")，这就是"下属不互相对话、全走主管"的代码化身）→ 入口连到主管 → 主管的出口是一条条件边，route_after_supervisor 返回的字符串（专家名或 __END__）在 conditional_map 里查到下一站。

那个条件路由函数极短（dispatch.py:186）：

def route_after_supervisor(state) -> str:
    next_name = state.get("next")
    if next_name is None or next_name == END_MARKER:
        return END_MARKER            # 走向 END
    return str(next_name)            # 走向某个专家

换到第二个"工头"——大厅导览员。代码在 apps/supervisor-router-agent/supervisor_router/core.py。它没有 LangGraph 图，就是一个函数 route(task, tier)。先看它的产物数据结构 RouteDecision（core.py:37）：

@dataclass(frozen=True)                     # core.py:37 · 冻结 = 不可变
class RouteDecision:
    agent_id: str | None                    # 选中的 agent · None = 不确定
    confidence: float                        # 置信度 0~1
    args_hint: dict[str, str]                # 建议参数
    reasoning: str                           # 为什么这么选
    suggested_command: str                   # 建议的 shell 命令
    fallback_alternatives: list[str]         # 备选 agent
    token_usage: dict | None = None          # 花了多少 token / 钱（可观测）
    tier_used: str = DEFAULT_TIER
    latency_ms: int = 0

再看主函数 route()（core.py:53）的五步骨架（已裁剪 try/except 细节）：

def route(task, *, tier=None) -> RouteDecision:
    if not task or not task.strip():                    # core.py:59 空任务早返回
        return RouteDecision(agent_id=None, confidence=0.0, reasoning="task 空 · 给个具体任务", ...)
    registry = list_agents()                             # ① 拿 agent 名单 · core.py:74
    if not registry:
        return _unsure("registry 空 ...", ...)
    try:
        from ai_trust_toolkit.llm import get_llm
        llm = get_llm(tier=actual_tier)                  # ② 便宜模型 · core.py:86
    except Exception as e:
        return _unsure(f"LLM 不可用 ... {e}", ...)        #    AC-1.5 挂了不抛
    system, user = _render_prompts(task, registry)       # ③ 渲染 prompt · core.py:93
    response = llm.invoke([SystemMessage(...), HumanMessage(...)])  # 调 LLM
    parsed = _extract_json(raw)                          # ④ 正则抓 JSON · core.py:115
    if parsed is None:
        return _unsure("LLM 输出非 JSON · skip", ...)
    decision = _build_decision(parsed, registry, ...)    # ⑤ 组装 RouteDecision
    decision = _critic_validate(decision, registry)      #    再 critic 校验
    return decision

route() 拿到 LLM 给的 RouteDecision 后不直接信，要过一道 _critic_validate（core.py:278）。这是路由器的"防幻觉"：

def _critic_validate(decision, registry) -> RouteDecision:
    if decision.agent_id is None:
        return decision                                  # 已经是 unsure · 直接返
    valid_ids = {a.agent_id for a in registry}
    if decision.agent_id not in valid_ids:               # ① LLM 编了个不存在的 agent
        log.warning("LLM 选 %r 不在 registry · fallback unsure", decision.agent_id)
        return RouteDecision(agent_id=None, confidence=0.0,
            reasoning=f"LLM 选 {decision.agent_id} 不在 registry · fallback unsure", ...)
    min_conf = float(os.getenv("PORTAL_SUPERVISOR_MIN_CONFIDENCE", "0.5"))
    if decision.confidence < min_conf:                   # ② 置信度太低
        return RouteDecision(agent_id=None, confidence=decision.confidence,
            reasoning=f"confidence {decision.confidence:.2f} < {min_conf} · 不自信 · 工程师手挑",
            fallback_alternatives=[decision.agent_id] + decision.fallback_alternatives, ...)
    return decision                                      # 通过

两道校验：① LLM幻觉出一个不在名单里的 agent → 退回 unsure；② 置信度低于阈值（默认 0.5） → 也退回 unsure。注意第②处的细节：低置信时它把 LLM 原本想选的 agent 放进 fallback_alternatives 的第一位，而不是丢掉——这样工程师看到"不确定，但我猜是 risk-reviewer"，可以手动挑回。

再看那个反复出现的 _unsure（core.py:130）——它是所有失败路径的统一出口：

def _unsure(reason, tier, start, latency=None) -> RouteDecision:
    return RouteDecision(agent_id=None, confidence=0.0, args_hint={},
        reasoning=reason, suggested_command="", fallback_alternatives=[],
        tier_used=tier, latency_ms=latency if latency is not None else ...)

最后一块拼图是 scope guard（范围守卫）：core.py:1-15 的模块 docstring 明确写"不真执行下游 agent(scope guard)"。route() 只吐"该找谁 + 命令怎么写"，绝不替你真的去调。

这个思想在框架里到处都是——Day 15 会讲的 spec-executor 只往 agent/* 分支提交、绝不碰 main。把"决策"和"执行"分开，是可信 Agent 系统的一条通用安全原则。

有时一个任务要好几个 agent 接力完成。框架用两个东西支持，且依然遵守 scope guard——只建议、不自动执行。

HandoffSignal（交接信号 · handoff.py:14）

@dataclass(frozen=True)                   # handoff.py:14
class HandoffSignal:
    next_agent_hint: str | None           # canonical agent_id · None = 链终止
    context_for_next: dict | None         # 传给下一棒的上下文
    chain_id: str | None                  # 整条链共享的 ULID
    reason: str                           # 给 supervisor 看的理由

逐字段：一个 agent 干完，可以在返回结果上挂这个信号——next_agent_hint 是"建议下一个谁接手"（None 表示这条链到此结束）；context_for_next 是传给下一棒的数据；reason 是理由，比如"verify fail · 需 prod pod 兜底"。但这同样只是建议，由 supervisor 决定要不要真交接（又是 scope guard）。

chain_id 值得单独说：它由 new_chain_id()（handoff.py:31）生成，而这个函数直接复用了 envelope 的 new_case_id()（同一套 ULID 生成器）。为什么？docstring 说得清楚：让 chain_id 和门户里 invocation 的 case_id 格式一致，串库时无需额外校验。一个 26 字符、按时间可排序的 ID，把一条接力链的多次调用串起来。

route_chain（多步规划 · core.py:322）

route_chain(task, max_steps=5) 让 LLM 一次性规划出 2-5 步的 agent 调用链。它的循环体（core.py:392-402）藏着两处边界防御：

decisions = []
for step in steps_raw[:max_steps]:            # AC-3.4：超出 max_steps 直接截断
    if not isinstance(step, dict):
        break                                 # AC-3.3：半途冒出非法项 → 截断，保留前几步
    d = _build_decision_from_step(step, registry, ...)
    d = _critic_validate(d, registry)         # 每一步都过 critic
    if d.agent_id is None:
        break                                 # AC-3.3：某步选了名单外 agent → 截断
    decisions.append(d)
return decisions

大白话：规划不是"要么全成功要么全失败"。若第 3 步 LLM 选了个不存在的 agent，它不会丢掉前 2 步，而是"截断"——保留已经合法的前缀，把断掉的地方给你看。这比"全砍掉重来"友好得多。另外 max_steps==1 时（core.py:341）它直接退化成单步 route()，保持向后兼容。

依然是只规划、不执行——打印出计划 + 共享 chain_id 给你看。上面这条链其实就是 Day 15/20 要讲的平台"自我开发"闭环。

✋ 动手：跟着真源码读一遍

# ① 图内 supervisor 的三段核心：任务书 / max_iter / 三种出路
sed -n '29,42p;100,115p;158,194p' packages/ai-trust-toolkit-starter-supervisor/src/*/dispatch.py

# ② 三字段黑板 + 一行组装成图（含 enable_critic 红线）
sed -n '24,41p' packages/ai-trust-toolkit-starter-supervisor/src/*/state.py
sed -n '37,124p' packages/ai-trust-toolkit-starter-supervisor/src/*/auto_config.py

# ③ 跨 agent router 的 RouteDecision / route() 五步 / critic 校验 / 优雅降级
sed -n '37,146p;278,319p' apps/supervisor-router-agent/supervisor_router/core.py

# ④ handoff 数据结构 + route_chain 的截断逻辑
sed -n '14,39p' packages/ai-trust-toolkit/src/ai_trust_toolkit/handoff.py
sed -n '322,402p' apps/supervisor-router-agent/supervisor_router/core.py

# ⑤（若已启用）用 CLI 试试路由决策
uv run supervisor-router "帮我看下这个 PR 的上线风险" --chain