Supervisor、Router 与 Handoff

多Agent协作的组织方式，决定问题交给谁、子结果交回哪里，以及最后由谁向用户负责。Supervisor是主管：把子任务交给专家，再收回结果汇总。Router是分派器：按问题类别选择处理者。Handoff是移交：把当前对话的控制权交给另一位处理者，让它接着服务用户。

同样是“找退款专家”，三种结构也可能完全不同。判断时先跟踪最后一次答复和下一轮对话，别只数框图上有几个头像。客服群里多了三个人，并不自动产生一个负责人。

用一条退款咨询看控制权

客户问“退款什么时候到账”。如果问题已经属于明确类别，分派器把它交给退款专家，专家直接答复。这是一次路由。

客户又问“订单没发货，能不能取消，钱会退到哪里”。主管可以让配送专家查订单条件，让退款专家说明退款渠道，再整合成一段答复。专家交回的是子结果，最后的解释责任仍在主管。

假如接待员已经确认客户要办理退款，它把后续对话移交给退款专家。客户补充订单号时，接手者继续处理，不必再从接待员绕一圈。移交既包括负责人变化，也包括所需上下文的交接。

后面的程序用固定答复观察这些线路。trace是调用记录，记录程序实际经过的角色；它并不是模型的内心思考。billing表示退款专家，delivery表示配送专家，intake表示接待员。

保存完整程序并运行

在自己的学习目录打开PowerShell，创建一个新的子目录：

mkdir l6-patterns
cd l6-patterns

把下面完整代码保存为 patterns_demo.py。固定数据和运行入口都已包含，文件名保留 .py 后缀。

"""观察三种协作结构中的调用与控制权。"""
ANSWERS = {"billing": "退款沿原支付渠道退回", "delivery": "未发货订单可申请取消"}

def route(topic):
    # 路由只选择处理者，未知主题交回人工。
    return {"退款": "billing", "配送": "delivery"}.get(topic, "human")

def router(topic):
    specialist = route(topic)
    answer = ANSWERS.get(specialist, "请人工确认问题类别")
    return answer, [f"router→{specialist}", f"{specialist}→客户"]

def supervisor(topics):
    # 专家返回子结果，由主管保留最后答复的责任。
    results, trace = [], []
    for topic in topics:
        specialist = route(topic)
        trace.append(f"supervisor→{specialist}→supervisor")
        results.append(ANSWERS.get(specialist, "请人工确认问题类别"))
    return "；".join(results), trace + ["supervisor→客户"]

def handoff(active, topic):
    if active not in {"intake", "billing", "delivery", "human"}:
        raise ValueError("未知负责人")
    # 首次接待可以移交；专家接手后，后续对话仍由它负责。
    target = route(topic) if active == "intake" else active
    answer = ANSWERS.get(target, "请人工确认问题类别")
    trace = [f"{active}→{target}：移交"] if active != target else []
    return target, answer, trace + [f"{target}→客户"]

def main():
    answer, trace = router("退款")
    print("Router：" + " / ".join(trace))
    print("答复：" + answer)
    answer, trace = supervisor(["退款", "配送"])
    print("Supervisor：" + " / ".join(trace))
    print("答复：" + answer)
    active, _, trace = handoff("intake", "退款")
    print("Handoff：" + " / ".join(trace))
    active, answer, trace = handoff(active, "补充订单号")
    print("后续对话：" + " / ".join(trace))
    print("当前负责人：" + active)

if __name__ == "__main__":
    main()

保存后执行 dir，确认能看到 patterns_demo.py，再运行：

ls
python -X utf8 patterns_demo.py

程序输出：

Router：router→billing / billing→客户
答复：退款沿原支付渠道退回
Supervisor：supervisor→billing→supervisor / supervisor→delivery→supervisor / supervisor→客户
答复：退款沿原支付渠道退回；未发货订单可申请取消
Handoff：intake→billing：移交 / billing→客户
后续对话：billing→客户
当前负责人：billing

沿着结果找负责人

Router的记录只有“分派器到退款专家、退款专家到客户”，所以这次只有一个专家提供答复。把主题分类做好，可以让不同领域使用各自的资料和工具；分类不确定时，程序转human，即人工处理者。

Supervisor记录了两次往返：专家结果回主管，再由主管答复客户。汇总不能只是把两段文字贴在一起。订单专家说“已发货”，退款专家说“未发货可取消”时，主管还要判断条件是否成立。例子使用两条相容的固定答复，所以以分号拼接；实际业务应合并结构化事实，再检查冲突与缺失。

Handoff输出的当前负责人是billing。下一轮“补充订单号”不是已知主题，程序仍由billing回复，说明接手者保留对话控制权。它没有重新调用route。这里的移交规则只有“首次接待可以转交”；若还要允许退款专家再转技术专家，就需要定义合法目标、移交次数和转回规则。

图中主管有返回线路，而移交线路继续向右。这是最值得记住的差别：委派子任务之后收回结果，与把接下来的对话交给别人，是两种责任安排。

什么任务适合哪条路线

结构可以组合。主管内部可能先路由，再把一个需要连续问答的流程交出去。图上的名称不会自动保证实现满足职责；代码中谁保存active、谁读取子结果、谁调用下一步，才决定真实行为。

共享完整聊天记录看似省事，却可能把无关客户资料也交过去。移交包通常需要目标、已确认事实、待补信息和允许动作。旧处理者是否还可写状态也要明确，避免“我以为你接了，你以为我还在处理”。

改一个位置，观察负责人是否变化

把main中的第一处router("退款")改为router("改密码")，运行python -X utf8 patterns_demo.py。前两行会变成路由到human，并答复“请人工确认问题类别”；Supervisor和Handoff部分保持原来的固定数据。

再把首次handoff("intake", "退款")中的“退款”改为“配送”，运行同一条命令。最后一行变成“当前负责人：delivery”，后续对话也由delivery处理。若只修改第二次调用的主题，负责人仍是billing，这说明“新输入内容变化”和“明确移交控制权”不是同一件事。

完成对照后，把首次router的主题恢复为“退款”，首次handoff也恢复为“退款”，保存并重跑。前两行重新交给billing，末行负责人也是billing，再用配套检查核对原始规则。

出现往返移交时，查看active变化和转交理由，而不是增加角色数量。主管成了瓶颈时，再看它是否串行等待所有专家、是否反复发送完整上下文；有证据证明瓶颈之后才考虑并行和上下文裁剪。

参考阅读：LangChain多Agent模式，对照不同模式的控制流与上下文安排；Handoffs重点看active agent状态如何影响后续对话。

运行配套检查

把下面测试完整保存为 test_patterns_demo.py，与主程序放在同一目录。5项检查分别覆盖单次分派、未知类别、主管汇总、移交后的连续处理和非法负责人。

"""检查分派、汇总和移交后的责任归属。"""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import patterns_demo as demo

class PatternTests(unittest.TestCase):
    def test_router_calls_one_specialist(self):
        answer, trace = demo.router("退款")
        self.assertEqual(trace, ["router→billing", "billing→客户"])
        self.assertIn("原支付渠道", answer)

    def test_unknown_routes_to_human(self):
        self.assertEqual(demo.route("改密码"), "human")
        self.assertIn("人工", demo.router("改密码")[0])

    def test_supervisor_receives_both_results(self):
        answer, trace = demo.supervisor(["退款", "配送"])
        self.assertIn("退款", answer)
        self.assertIn("未发货", answer)
        self.assertEqual(trace[-1], "supervisor→客户")
        self.assertEqual(len(trace), 3)

    def test_handoff_keeps_active_owner(self):
        active, _, _ = demo.handoff("intake", "退款")
        active, _, trace = demo.handoff(active, "补充订单号")
        self.assertEqual(active, "billing")
        self.assertEqual(trace, ["billing→客户"])

    def test_invalid_owner_rejected(self):
        with self.assertRaises(ValueError):
            demo.handoff("administrator", "退款")

if __name__ == "__main__":
    unittest.main()

python -X utf8 -m unittest -v test_patterns_demo.py

测试名称后应显示 ok，末尾显示 Ran 5 tests 和 OK。若找不到模块，先用 dir 确认两个文件保存在同一目录。