第 13 章 · 综合实战二：多智能体工单路由

把第 8、9 章的子图与第 3 章的条件路由组合成“监督者 + 两个领域智能体”：工单先由监督者按关键词路由，再由账单或工程子图处理，最后回到监督者收尾。全程不依赖模型。

机制与设计选择

监督者不直接干活，只写 routing 字段；两个 worker 节点各自调用一个编译好的子图并只取回自己关心的结果，处理完把 routing 置为 done，让监督者收尾后结束。三个工单覆盖账单、工程两条路径，断言把队列结论与主题标记都锁定。

示例

python

import operator
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END

class BillingState(TypedDict):
    subject: str
    finding: str

def billing_validate(state: BillingState):
    return {"finding": "billing queue: " + ("valid" if state["subject"].strip() else "missing detail")}

def billing_close(state: BillingState):
    return {"subject": state["subject"] + " [processed]"}

billing = (
    StateGraph(BillingState)
    .add_node("validate", billing_validate)
    .add_node("close", billing_close)
    .add_edge(START, "validate")
    .add_edge("validate", "close")
    .add_edge("close", END)
    .compile()
)

class EngState(TypedDict):
    subject: str
    finding: str

def eng_triage(state: EngState):
    severity = "high" if len(state["subject"].split()) > 25 else "low"
    return {"finding": "engineering queue: severity=" + severity}

def eng_assign(state: EngState):
    return {"subject": state["subject"] + " [assigned]"}

eng = (
    StateGraph(EngState)
    .add_node("triage", eng_triage)
    .add_node("assign", eng_assign)
    .add_edge(START, "triage")
    .add_edge("triage", "assign")
    .add_edge("assign", END)
    .compile()
)

class TicketState(TypedDict):
    subject: str
    routing: str
    notes: Annotated[list[str], operator.add]

def supervise(state: TicketState):
    if state["routing"] in ("billing", "engineering", "done"):
        return {}
    text = state["subject"].lower()
    routing = "billing" if ("invoice" in text or "refund" in text) else "engineering"
    return {"routing": routing}

def run_billing(state: TicketState):
    result = billing.invoke({"subject": state["subject"], "finding": ""})
    return {"subject": result["subject"], "notes": [result["finding"]], "routing": "done"}

def run_engineering(state: TicketState):
    result = eng.invoke({"subject": state["subject"], "finding": ""})
    return {"subject": result["subject"], "notes": [result["finding"]], "routing": "done"}

supervisor = (
    StateGraph(TicketState)
    .add_node("supervise", supervise)
    .add_node("run_billing", run_billing)
    .add_node("run_engineering", run_engineering)
    .add_edge(START, "supervise")
    .add_conditional_edges(
        "supervise",
        lambda s: s["routing"],
        {"billing": "run_billing", "engineering": "run_engineering", "done": END},
    )
    .add_edge("run_billing", "supervise")
    .add_edge("run_engineering", "supervise")
    .compile()
)

for subject, expected in [
    ("Refund for invoice #1042", "billing queue: valid"),
    ("Crash when saving a long file, please help", "engineering queue: severity=low"),
]:
    result = supervisor.invoke({"subject": subject, "routing": "", "notes": []})
    assert result["notes"] == [expected], result
    assert result["subject"].endswith("[processed]") or result["subject"].endswith("[assigned]")
    print(result["subject"], "|", result["notes"])
print("routing graph ok")

运行与验收

直接运行：应打印两行工单结果与 routing graph ok。第二个工单主题较长（6 个词但未超 25 词阈值），因此判定为 low severity，验证了确定性的优先级规则而不是随机行为。

容易踩的坑

worker 节点把 routing 置为 done 让监督者判定收尾，是为了避免“worker 直接连 END”导致另一类工单无法再走监督逻辑；如果多个 worker 并存，务必让每个 worker 都回到监督者。不要把子图内部键直接摊进父状态，ch12 已经示范过映射方式。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 监督者负责什么？

2. worker 完成后去哪？

3. 子图结果如何进入父状态？

4. 本例需要模型吗？

动手实践

增加第三个“安全”队列并改路由关键词。
给工程子图加一条条件边：high severity 走加急分支。
用参数化测试覆盖 25 词阈值两侧。

官方参考

核验日期：2026-09-17，示例已在本机离线运行验证。

第 13 章 · 综合实战二：多智能体工单路由 ​

机制与设计选择 ​

示例 ​

运行与验收 ​

容易踩的坑 ​