第 7 课：确定性流程与多 Agent——图编排和咨询台团队

学完本课你能：读懂并用 GraphBuilder 写一条含人工审批节点的确定性流程；搭一个 supervisor 分诊团队并验证路由正确；说清单 Agent / 图 / 团队各自该用在哪｜预计耗时 60-90 分钟｜前置：第 6 课｜难度 ★★★

这是最硬核的一课，但你会发现图里的每个零件都是老朋友：图节点里的 agent 跑的是普通 Agent（前六课的主角），step 是普通 Python 函数（第 4 课你就在写），人工节点用的是审批（第 4 课你配过）。新东西只有"把这些零件按图的结构组织起来"这一层。

先交代为什么需要这层。到第 6 课为止，小图的一切行为都是"模型决策"：调不调工具、答不答、怎么答。这对大部分场景够用——但有类场景不行：顺序和规则本身不能出错的事。图书馆的逾期处理就是：查逾期 → 分级 → 严重的要人工确认才能冻结 → 算滞纳金 → 通知。"严重逾期必须先有人确认才能冻结权限"是规章，不是建议。你把这套流程写进 prompt 里"请务必按以下步骤……"，模型大概率遵守——但第 0 课就说过，大概率在涉及权限和金钱的地方不够格。图编排把这些规则从"请求"变成"结构"：流程是图，条件是分支，人工节点是闸，模型只在需要理解语言的节点出场。

1. 学习目标

用 GraphBuilder 七件套（entry/step/human/when/edge/build/@graph）写一条流程
跑通含人工节点的流程：interrupt 暂停 → 人工给值 → resume 续跑
搭 supervisor 团队（调度 Agent + 专家成员），验证路由正确性
对"单 Agent / 图 / 团队"三选一有自己的判断框架

2. 概念讲解：把"流程"从 Prompt 挪进代码

图编排的零件表（对照 app/graphs/refund_flow.py 看一遍再往下读）：

DSL	干什么	对应你已会的东西
`builder.entry(name)`	指定入口节点	—
`builder.step(name, fn)`	确定性步骤：`fn(state: dict) -> dict`，返回值合并进 state	第 4 课的工具函数同款形态
`builder.agent(name, agent_key)`	节点 = 跑一个 Agent	前六课的 Agent
`builder.human(name, prompt=...)`	人工节点：流程暂停等人给值，值写进 `state[name]`	第 4 课的审批（同一个 interrupt 机制）
`builder.when(src, cond, then=..., otherwise=...)`	条件分支：`cond(state)` 真假走两条边	第 4 课的风险定级逻辑
`builder.edge(a, b)`	普通连线	—
`@graph(name=...)`	注册进 Registry，工作台流程图页可见	`@tool` 的同款

一句话：step 是代码、agent 是模型、human 是人、when 是分支——一张图把三种执行者按业务流程编排起来，谁在哪步出场写得死死的。

两个容易踩的结构规则：① 同一节点不能同时有普通边和条件边（出边要么确定要么二选一，不能三心二意）；② 汇点自动接 END，不用写 edge(x, END)。

跑一条图：图没有独立的 HTTP 运行入口（/v1/graphs 是只读结构视图），跑图用 GraphRunner——测试、脚本、或把它包成 Agent 的工具（flow_as_tool，延伸阅读里点到）。本课用一个小脚本直跑，恰好让你零距离看到 interrupt/resume 的原始形态。

多 Agent 团队是另一个方向的问题。当小图既要办事务（查借阅、挂失）又要答规章，一个 Agent 全扛的下场是人设越来越长、互相干扰。团队做法：一个 supervisor（调度员）读用户请求，输出一个 JSON 决策——{"action": "delegate", "agent": "my-lib-rules", "input": "..."} 或 {"action": "answer", ...}——框架据此把请求转给对应成员，成员的结果回到 supervisor 汇总答复。

注意 supervisor 的产出是结构化决策，不是自然语言——路由是数据不是文章，所以它的指令里那个"只输出一个 JSON 对象"的输出格式段是刚需（这也正是本课参考产出里 supervisor YAML 的核心段落，照着写）。防失控有三道：max_delegations（委派上限，防两个 Agent 互相踢皮球死循环）、成员必须在登记表里、路由失败降级为直接答复。

什么时候用哪个（先给个起步框架，练习 C 会逼你自己长出更细的判断）：

单 Agent：默认选项。一个场景、一套工具、人设写得清楚——90% 的需求到这为止。
图编排：流程有不可出错的顺序/分支/人工环节（钱、权限、合规）。
团队：多个异质场景挤在一个 Agent 里互相干扰（事务 vs 知识 vs …），且量级值得为之付额外延迟和 token。

3. 动手：主线步骤

步骤 1：写逾期处理图

新建 app/graphs/my_library_flow.py（参考产出是终版；关键是理解结构——四个节点、一条条件边）：

"""逾期处理流水线：查逾期 → 分级 → 严重逾期人工确认 → 处罚 → 通知。"""

from typing import Any
from datetime import date

from lubanagent.extension import default_registry, graph
from lubanagent.graph import GraphBuilder

SEVERE_OVERDUE_DAYS = 30  # 严重逾期线：超过则冻结借阅权（需人工确认）


def _load_overdue(state: dict[str, Any]) -> dict[str, Any]:
    """按学号查借阅记录，算出逾期最重的一册（复用第 4 课的工具，不绕过治理）。"""
    tool = default_registry.get_tool("my_lib_lookup")
    records = tool.handler(student_id=state["student_id"])["records"]
    today = date(2026, 9, 9)  # 教程演示：固定"今天"，真实系统用 date.today()
    worst = max(records, key=lambda r: (today - date.fromisoformat(r["due"])).days)
    overdue_days = (today - date.fromisoformat(worst["due"])).days
    return {"worst_title": worst["title"], "overdue_days": overdue_days}


def _needs_confirm(state: dict[str, Any]) -> bool:
    return state["overdue_days"] > SEVERE_OVERDUE_DAYS


def _apply_penalty(state: dict[str, Any]) -> dict[str, Any]:
    days = state["overdue_days"]
    return {"penalty": round(days * 0.1, 1), "frozen": days > SEVERE_OVERDUE_DAYS}


def _notify(state: dict[str, Any]) -> dict[str, Any]:
    if state.get("frozen"):
        return {"message": f"《{state['worst_title']}》逾期 {state['overdue_days']} 天，"
                            f"滞纳金 {state['penalty']} 元，借阅权限已冻结（缴清后恢复）。"}
    return {"message": f"《{state['worst_title']}》逾期 {state['overdue_days']} 天，"
                        f"滞纳金 {state['penalty']} 元，请尽快归还。"}


def _build() -> "GraphBuilder":
    builder = GraphBuilder("my_library_flow", description="逾期处理：查逾期→分级→严重确认→处罚→通知")
    builder.entry("load_overdue")
    builder.step("load_overdue", _load_overdue)
    builder.human("confirm_freeze", prompt=f"逾期超过 {SEVERE_OVERDUE_DAYS} 天将冻结借阅权限，需人工确认")
    builder.step("apply_penalty", _apply_penalty)
    builder.step("notify", _notify)
    # entry 节点只出条件边：严重逾期走人工确认，轻度直走处罚（同一节点不能混两种边）
    builder.when("load_overdue", _needs_confirm, then="confirm_freeze", otherwise="apply_penalty")
    builder.edge("confirm_freeze", "apply_penalty")
    builder.edge("apply_penalty", "notify")
    return builder


@graph(name="my_library_flow", description="逾期处理：查逾期→分级→严重确认→处罚→通知")
def my_library_flow_builder() -> Any:
    return _build()

对照概念表读一遍结构：查逾期（step，复用第 4 课工具）→ when 按逾期天数分流 → 严重的进 human（冻结前必须有人确认）→ 处罚（step，规则是代码算的，不是模型说的）→ 通知（step）。

✅ 检查点：bash scripts/dev.sh restart 后，工作台流程图页出现 my_library_flow，点开看到四个节点（step 蓝 / human 橙）和一条虚线双出边的条件分支。也可用 API 看：curl http://localhost:8000/v1/graphs/my_library_flow。

步骤 2：跑轻度逾期——直走全流程

图用 GraphRunner 跑。新建脚本 run_flow.py（放仓库根目录，跑完可删；这是看图运行原始形态的最短路径）：

"""跑逾期处理图：轻度直走 / 重度人工确认。"""
import asyncio
import sys

sys.path.insert(0, ".")
from lubanagent.config import Settings
from lubanagent.extension import default_registry, discover
from lubanagent.graph import GraphRunner
from lubanagent.persistence import Database

from app.graphs.my_library_flow import _build

discover(Settings.load().extension_packages)


async def main():
    db = Database(Settings.load().database_url)
    runner = GraphRunner(db=db, registry=default_registry)
    flow = _build().build()

    r1 = await runner.run(flow, {"student_id": "2023055678"})
    print("轻度逾期:", r1.status, r1.output.get("message"))

    await db.close()


asyncio.run(main())

（学生 2023055678 是步骤 3 会加进工具数据的轻度逾期读者。）

先给 app/tools/my_library/lookup.py 的 _BORROWS 换成含两个读者的版本（参考产出 04 有终版）：

_BORROWS: dict[str, list[dict]] = {
    "2023011234": [
        {"title": "三体", "due": "2026-08-30", "renewals_left": 1},
        {"title": "机器学习系统设计", "due": "2026-08-01", "renewals_left": 0},
    ],
    "2023055678": [
        {"title": "计算机网络", "due": "2026-09-05", "renewals_left": 1},
    ],
}

重启服务（工具数据变了），然后：

.venv/bin/python run_flow.py

✅ 你应该看到：轻度逾期: completed 《计算机网络》逾期 4 天，滞纳金 0.4 元，请尽快归还。——四步直通，没碰人工节点（4 天 < 30 天，when 走了 otherwise 边）。滞纳金 0.4 是代码算的（4 × 0.1），不是模型"觉得"的。

步骤 3：跑重度逾期——interrupt 暂停

往脚本里加第二段（重启前先加好）：

    r2 = await runner.run(flow, {"student_id": "2023011234"})
    print("重度逾期:", r2.status, r2.interrupt)

    r3 = await runner.resume(flow, r2.thread_id, True)
    print("人工确认后:", r3.status, r3.output.get("message"))

再跑一遍。

✅ 你应该看到三行——

轻度逾期: completed 《计算机网络》逾期 4 天，滞纳金 0.4 元，请尽快归还。
重度逾期: interrupted {'node': 'confirm_freeze', 'prompt': '逾期超过 30 天将冻结借阅权限，需人工确认', 'state': {..., 'overdue_days': 39}}
人工确认后: completed 《机器学习系统设计》逾期 39 天，滞纳金 3.9 元，借阅权限已冻结（缴清后恢复）。

逐行读：重度逾期的 Run 状态是 interrupted——流程在 confirm_freeze 节点停住，interrupt 里带着节点名、提示语和当时的 state 快照；resume(flow, thread_id, True) 把人工决定（True=确认冻结）续进去，流程走完。在"确认"发生之前，冻结在代码层面就不可能发生——这就是把流程从 Prompt 挪进代码的全部意义。把 resume 的第三个参数换成 False 试试，看流程怎么走（习题里有一问）。

顺带看一眼工作台 Traces 页：两次图 Run 各自留了完整现场（每个节点的执行都有记录）——图 Run 与 Agent Run 同库同治理，第 0 课的承诺在这里兑现。

步骤 4：写 supervisor 和规章专家

团队需要新角色。新建 app/agents/my_lib_triage.yaml（supervisor，输出 JSON 决策是核心，完整版见参考产出）：

name: my-lib-triage
description: 图书馆咨询台 supervisor——判断请求类型并委派成员或直接答复
instruction: |
  # 角色
  你是图书馆咨询台调度「小台」，负责判断读者请求类型，
  决定自己答复还是委派给团队成员。你不解决具体问题，只做准确分流。

  # 判断规则（按优先级）
  1. 借阅记录查询、校园卡挂失、逾期处理 → 委派 my-lib-assistant（借阅事务专员）。
  2. 规章制度咨询（借期/续借/滞纳金/开放时间）→ 委派 my-lib-rules（规章专家）。
  3. 问候、闲聊、感谢 → 自己简短答复，并提示你能转接图书馆业务。
  4. 请求不明确 → 追问一句确认意图，再分流。

  # 委派要点
  - 转述请求时带上全部关键信息（学号、书名、诉求原话），让成员不用再问一遍。

  # 输出格式（必须严格遵守）
  只输出一个 JSON 对象，不要输出其他内容，不要用 markdown 代码块包裹：
  - 自己回答：{"action": "answer", "answer": "<最终答复>"}
  - 委派成员：{"action": "delegate", "agent": "<成员 key>", "input": "<转述请求>"}

  # 边界
  - 一次只做一个动作（answer 或 delegate 二选一），不连环委派。

新建 app/agents/my_lib_rules.yaml（规章专家：纯 RAG + 守门，knowledge 段照抄第 5 课——完整版见参考产出）。

导入：

.venv/bin/luban init

✅ 你应该看到：my-lib-rules@1、my-lib-triage@1 两行新导入。

步骤 5：组队

新建 app/multiagents/my_library_team.py：

"""图书馆咨询台团队：triage 调度 + 借阅事务/规章专家两成员。"""

from lubanagent.multiagent import Team, TeamBuilder


def build_my_library_team() -> Team:
    builder = TeamBuilder(
        "my-library-team", description="图书馆咨询台：triage 调度 + 事务/规章成员"
    )
    builder.supervisor("my-lib-triage")
    builder.agent("my-lib-assistant", description="借阅记录查询、校园卡挂失、逾期处理")
    builder.agent("my-lib-rules", description="借阅规章、续借规则、滞纳金、开放时间咨询")
    builder.route(max_delegations=4)
    return builder.build()

结构就这么多：调度是谁（supervisor）、成员有谁和各自管什么（description 就是给调度看的路由提示）、防失控上限（max_delegations）。

步骤 6：给团队开一个 HTTP 入口

团队的运行入口是用户领地自定义路由（框架不预置团队端点——app/api/teams.py 里内置客服的 /v1/teams/cs-team/runs 就是这个模式的示范）。在 app/api/teams.py 末尾照抄模式加一段：

# 用户领地自定义路由模式（照抄上面 cs-team 的写法加自己的团队端点）：
@router.post("/my-library-team/runs", response_model=TeamRunResponse)
async def run_my_library_team(body: TeamRunRequest, ctx: CtxDep) -> TeamRunResponse:
    """触发图书馆咨询台团队（教程第 7 课：用户领地自定义团队路由示范）。"""
    from app.multiagents.my_library_team import build_my_library_team

    team = build_my_library_team()
    runner = TeamRunner(db=ctx.db, runtime=ctx.runtime, registry=ctx.registry)
    thread_id = body.thread_id or uuid.uuid4().hex
    result = await runner.run(team, body.input, thread_id=thread_id)
    return TeamRunResponse(
        status=result.status,
        output=result.output,
        run_id=result.run_id,
        thread_id=result.thread_id,
    )

重启服务，然后测路由：

curl -X POST http://localhost:8000/v1/teams/my-library-team/runs \
  -H "Content-Type: application/json" \
  -d '{"input": "我是学号 2023011234，帮我查下借的书"}'

✅ 你应该看到：返回 status: completed，output 的 messages 里能看到完整路由过程——supervisor 决定委派 my-lib-assistant（转述里带着学号）、成员调 my_lib_lookup 拿到真实借阅记录作答、supervisor 汇总回复。再问一个规章问题（书可以续借几次？），确认这次被路由到 my-lib-rules。

一个值得知道的幕后：这套团队机制在 0.1.0 上其实跑不通——supervisor 指令里的 JSON 样例会撞上框架的变量渲染模板直接报错（已修复，登记为风险 R13）。你此刻的顺利，是踩平了坑之后的顺利。

4. 完成自检清单

流程图页看到 my_library_flow 的节点拓扑（step/human 着色、条件分支虚线）
轻度逾期直走 completed，滞纳金由代码算出
重度逾期 interrupted 在 confirm_freeze，resume(True) 后 completed 且冻结
resume(False) 的行为亲手看过（练习里有问）
团队两种问题分别路由到正确的成员，supervisor 转述带全上下文

5. 常见坑

图结构报错（如"同一节点不能同时有普通边与条件边"）。 build() 会做拓扑校验（入口存在/边引用的节点存在/无孤立节点），报错信息会指明问题。最常见的就是给 entry 又加了普通边又加了 when——条件边自带两条出路，不需要再连。

step 函数里用 run_until_complete。 step 可以直接是 async 函数（async def + await），不要在同步函数里套 run_until_complete——图跑在事件循环里，嵌套跑另一个循环必炸。

团队 Run 报「子 Run 未成功: xxx → failed」。 先查 supervisor 的指令是否真的强制了 JSON 输出格式（"只输出一个 JSON 对象"那段不能省）；解析失败会降级为直接答复，但子 Run 失败通常是别的问题——去 Traces 页看那个子 Run 的现场。

路由总是给同一个成员。 supervisor 的判断规则段写糊了，或成员 description 没有区分度。修判断规则，别修框架。

改了团队/图没生效。 图和团队是 Python 模块，重启服务（discovery）；Agent YAML 则照旧 init。

6. 练习

练习 A（模仿）：给图加一个分支——逾期天数 ≤ 3 天的"宽限"路径（不收滞纳金，只提醒）。这需要动 when 的条件函数和处罚/通知逻辑，试试保持图结构不动、只改函数内容能不能做到。（能——这正是图结构的稳定性价值：规则微调不 reshape 流程。）

练习 B（变式）：resume(flow, thread_id, False) 后流程走成什么样？在 state 里加个字段让"拒绝确认"的路径可观察（比如 notify 的文案区分"已撤销冻结"）。想清楚业务语义：人工拒绝了冻结，流程应该怎么收尾？

练习 C（综合，本课答辩题）：同一句需求——"学生申请把逾期的书续借"——分别考虑三种实现：单 Agent（人设+工具）、图（步骤编排）、团队（分诊+专家）。写下你的三选一和 100 字理由。判断要点回看 §2 的起步框架，但要用你自己的话。写完对照第 0 课的表格——你现在的判断框架比第 0 课粗看一眼时精细了多少？

7. 小结与下一课预告

你集齐了全部三种执行形态：单 Agent（第 2-5 课）、图编排（确定性流程 + 人工节点）、团队（supervisor 分诊 + 专家成员）——并且知道它们不是"高级程度"的递进关系，而是适用场景的平行选择。你也亲历了 interrupt/resume 的原始形态和用户领地自定义路由的接线方法。

最后一课：把这套东西带出你的笔记本。Docker 五服务全栈部署、Langfuse 观测接线、以及这个基座最独特的承诺的实操验证——跟着上游升级而不丢你的改动。毕业设计清单也在那等着你。

延伸阅读：用户手册 §12（图编排——含流程作 Tool 的 flow_as_tool）、§13（多 Agent 协作）｜参考产出：reference/07-library-flow-and-team.py、reference/07-triage-and-rules.yaml ｜对照范本：app/graphs/refund_flow.py、app/multiagents/cs_team.py、app/agents/cs_triage.yaml