用子图和 Supervisor 隔离职责

本章目标

把 Worker 封装成独立子图，由父图负责选择，而不是把全部逻辑塞进一个巨大节点。

完整代码：examples/subgraph_supervisor.py

builder.add_node("supervisor", classify)
builder.add_node("order_agent", build_worker("order"))
builder.add_node("refund_agent", build_worker("refund"))
builder.add_edge(START, "supervisor")
builder.add_conditional_edges(
    "supervisor",
    lambda state: state["intent"],
    {"order": "order_agent", "refund": "refund_agent"},
)

运行：

python examples/subgraph_supervisor.py

预期：

退款 Worker：需要先提交审批。

子图适合隔离以下边界：

• 独立状态或上下文。
• 独立工具权限。
• 独立测试与质量目标。
• 需要单独持久化策略的长期 Worker。

多 Agent 不等于多个模型。分类可以使用规则或低成本模型，关键决策可以使用结构化输出，确定性校验应继续用普通代码。没有职责隔离需求时，不要为了“多 Agent”增加路由和 token 成本。

显式投影父图与子图状态

父图和子图碰巧使用相同字段名时，可以直接把编译后的子图作为节点。状态契约不同或涉及敏感字段时，应增加适配节点，只传 Worker 需要的数据：

class RefundWorkerInput(TypedDict):
    ticket_id: str
    order_status: str
    policy: str


async def call_refund_worker(state: SupervisorState) -> dict[str, str]:
    worker_input: RefundWorkerInput = {
        "ticket_id": state["ticket_id"],
        "order_status": state["order_result"],
        "policy": state["kb_result"],
    }
    result = await refund_worker.ainvoke(worker_input)
    return {"proposal": result["proposal"]}

这样可以避免把完整用户画像、其他租户数据或父图内部控制字段暴露给 Worker，也能在边界处验证输出。

工具权限属于服务端配置

Supervisor 选择 Worker 不等于授权 Worker。每个 Worker 的工具集合、凭证、网络访问和最大金额应由服务端静态配置或策略服务决定，不能让模型通过输出工具名称获得新权限。

一个安全的职责拆分示例：

• 分类 Worker：只读文本，不访问订单系统。
• 订单 Worker：只读订单状态，不具备退款权限。
• 退款方案 Worker：生成结构化方案，不执行资金动作。
• 执行节点：普通代码调用退款服务，并要求审批事实与幂等键。

子图的故障与持久化边界

父图应决定 Worker 超时后是重试、降级还是转人工。不要让每个 Worker 无限重试。子图是否独立持久化取决于生命周期：只服务于一次父图运行的短子图可继承父图执行上下文；需要被单独恢复、查询或扩缩容的长期 Worker 应拥有自己的线程标识和状态所有权。

测试时不要只断言最终回答字符串，还应断言：路由选择、传入 Worker 的字段集合、禁止工具没有被调用、Worker 异常进入预定边界、多个 Worker 的结果以确定方式合并。

本章验收

• 能写出父图到子图的输入投影和输出校验。
• 每个 Worker 有明确的数据、工具和凭证最小权限。
• Worker 超时、非法输出和不可用都有父图级处理策略。
• 能解释什么时候应该用普通节点而不是增加一个“Agent”。