LangGraph 子图与多代理

QuibblerAgentQuibblerAgent 2026-11-04 约 23 分钟 2 次阅读

前六篇把一张图的能力推到了头:State 定黑板语义,条件边管路由,create_agent 一行拿到成品代理,interrupt 又把人工审批接进流程。图越写越大是迟早的事——几十个节点挤在一张图里、几个团队同时在改、同一段流程在三个项目里各有一份拷贝,都是单图撑不住的信号。第 1 篇预告的“多代理”超能力正是为此准备:一张图可以作为节点嵌进另一张图。本篇是第 7 篇,把子图与多代理一次讲完。

版本口径与全系列一致:langgraph 1.2.x,PyPI 独立发布,子图机制以官方子图文档为准,多代理生态以官方多代理文档为准。全文按“拆分信号 → 子图即节点 → 状态转换层 → 拓扑三式 → Command 路由 → 生态现状 → 审稿流水线实战 → 总结”展开。


1、什么时候该拆子图

先把定义钉死,官方一句话:子图就是作为另一个图的节点来使用的图——封装概念在 LangGraph 里的落地。一个编译好的图,整个挂在父图的一个节点位上;父图看它,与看一个普通节点函数没有区别。

什么时候该动手拆?官方给过三个动机:搭建多代理系统、跨图复用节点、团队各自独立开发。翻译成工程信号更直观,对照如下:

信号 典型表现 拆子图的收益
单图过大 节点几十个,draw_mermaid 一屏放不下 分层组装,每层一眼读懂
多人协作 谁改一处,全图跟着回归 接口收窄,各团队各管一张图
流程复用 同一套审校流程到处复制粘贴 一处定义,多处挂载

三个信号出现任意一个,就可以考虑拆;三个全中,基本必须拆。也要看到代价:拆开后跨层调试要跳图,状态要在两套黑板之间搬运——能一张图讲清楚的流程,不必为拆而拆。

2、子图即节点

接入子图的第一种方式最直接:编译图本身就是合法节点,add_node 直接收。前提只有一条——父图与子图的 State 键名兼容,官方叫 shared state keys:两张图的黑板至少共享一部分键,运行时自动把父图黑板交给子图,子图跑完再把增量写回来。

先造一张两节点的润色子图,与父图共用同一个 State,代码如下:

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

class State(TypedDict):
    article: str
    notes: Annotated[list, operator.add]   # 父子图共享的键

def polish(s: State) -> dict:              # 子图节点一:润色
    return {"article": s["article"] + "(已润色)"}

def fact_check(s: State) -> dict:          # 子图节点二:查证
    return {"notes": ["事实核查通过"]}

sub = StateGraph(State)                    # 子图与父图同一个 State
sub.add_node("polish", polish)
sub.add_node("fact_check", fact_check)
sub.add_edge(START, "polish")
sub.add_edge("polish", "fact_check")
sub.add_edge("fact_check", END)
polish_graph = sub.compile()               # 编译完就是一个合法节点

解读:State 里 article 与 notes 两个键,父图子图同名同义——这就是“兼容”的全部要求,直接传同一个类是最省事的形态。子图内部怎么拐弯,父图既不知道也不关心;在父图眼里,它就是进去一次、出来一次的普通节点。父图这边照常五步搭图:

builder = StateGraph(State)
builder.add_node("polish_graph", polish_graph)    # 编译图直接登记
builder.add_node("publish",
                 lambda s: {"notes": ["已发布"]})
builder.add_edge(START, "polish_graph")
builder.add_edge("polish_graph", "publish")
builder.add_edge("publish", END)
result = builder.compile().invoke(
    {"article": "多代理初稿", "notes": []})
print(result["notes"])

输出如下:

['事实核查通过', '已发布']

解读:polish_graph 这一个节点内部实际跑了润色与查证两个超步,notes 里那条“事实核查通过”正是子图节点写回的——状态双向自动共享,父图零转换代码。再兑现一段前情:第 5 篇讲过 create_agent 返回的就是编译好的图(见官方 Agents 文档),所以 builder.add_node(“researcher”, create_agent(model=“openai:gpt-6”, tools=...)) 一行就能把成品代理挂进父图,前提仍是父图黑板有键兼容的 messages 字段。想在 stream 里看到子图内部的事件,调用时加 subgraphs=True 即可。

3、状态不兼容时的转换层

第二种方式应对另一类现实:子图黑板与父图没有共享键。老模块、外部团队的图、领域差异大的两段流程,schema 对不上很常见。官方口径明确:此时子图不能直接登记,要包在一个节点函数里调用,进出时转换 State——这个包装函数就是转换层。

以一张“广告清理”子图为例,它只认 text 与 clean 两个键,与父图的 article 对不上,代码如下:

class EditState(TypedDict):
    text: str          # 子图黑板:与父图无共享键
    clean: str

def strip_ads(s: EditState) -> dict:
    return {"clean": s["text"].replace("广告:", "").strip()}

sub = StateGraph(EditState)
sub.add_node("strip_ads", strip_ads)
sub.add_edge(START, "strip_ads")
sub.add_edge("strip_ads", END)
strip_graph = sub.compile()

def ad_stripper(s: State) -> dict:        # 包装节点:状态转换层
    result = strip_graph.invoke({"text": s["article"]})
    return {"article": result["clean"]}   # 以父图认的键写回

解读:转换层干两件事。进去时翻译:把父图黑板的 article 映射成子图要的 text;出来时再翻译:从子图结果里取 clean,以父图认的 article 键写回。父图把 ad_stripper 当普通节点登记即可,子图从此是父图眼里的黑盒,内部状态不再自动共享——这正是适配器的经典取舍:丢掉共享的便利,换来彻底解耦,异构的图不改一行就能复用。

4、多代理拓扑三式

子图解决“怎么装进去”,多代理解决“怎么组起来”:把多个代理各自的图当作节点,再用边与路由组织协作。经典拓扑就三式,先看图再下表。

4.1、supervisor 中央调度

第一式是中央调度,拓扑如下:

flowchart LR S([START]) --> C[调度者] C -->|派活| W1[写手] C -->|派活| W2[查证] W1 -->|交差| C W2 -->|交差| C C -->|完工| E([END])

解读:调度者是唯一的协调中枢——看黑板决定派谁干活,worker 干完把结果交回,彼此互不认识。分工明确的流程最合适;代价是调度者既是瓶颈也是单点。

4.2、handoff 交接棒

第二式是交接棒,没有常驻调度者,拓扑如下:

flowchart LR S([START]) --> A[接待] A -->|移交| B[技术专家] B -->|移交| C[售后] C --> E([END])

解读:控制权像接力棒——接待员处理完自己的段落,连同全部上下文移交给技术专家,专家再交给售后,一条链走到底。每棒都带着完整前情,适合分阶段推进的线性流程。

4.3、swarm 全互联

第三式是全互联,任何代理都能把控制权移交给任何其他代理,组网形态如下:

flowchart LR A[代理一] -->|移交| B[代理二] B -->|移交| C[代理三] C -->|移交| A A -->|移交| C

解读:无中心、全互联,最灵活也最容易绕圈——没有裁判,靠每个代理自己的移交判断兜底,动态开放的探索型任务才值得上这一式。

三种拓扑对比如下:

维度 supervisor handoff swarm
谁决定去向 中央调度者 当前代理指定后继 当前代理任选
代理间关系 互不相识 链式相邻 全互联
上下文归属 汇聚在调度者 顺流携带 随移交漂移
适用场景 分工明确 分阶段线性 动态开放

三式并不互斥:大系统里 supervisor 管总体,某一段内部跑 handoff,完全可以叠加。

5、Command 动态指定下一跳

第 4 篇讲条件边时留过一句:真需要“边改状态边路由”,官方另备 Command 原语——本篇兑现。第 6 篇已经用过 Command(resume=...) 恢复中断,它还有另一副面孔:节点直接返回 Command(goto=..., update=...),一个返回值同时携带状态增量与下一跳。

一个 supervisor 式主编节点如下:

from langgraph.types import Command

def chief(s: State) -> Command:
    if not s["notes"]:                     # 还没人干过活:派写稿
        return Command(goto="writer",
                       update={"stage": "写稿"})
    if len(s["notes"]) == 1:               # 写完:派审校
        return Command(goto="reviewer",
                       update={"stage": "审校"})
    return Command(goto=END, update={"stage": "收工"})

解读:chief 节点没有配路由函数,判断与写黑板合在同一个返回值里——update 照第 2 篇的部分更新语义合并,goto 指定下一个节点,END 也能作为去向收束全图。goto 还接受节点名列表,一次并行派活给多个编辑;子图内部想直接跳父图的节点,再加 graph=Command.PARENT。它与条件边的关系整理如下:

维度 条件边 Command
判断位置 节点外的路由函数 节点函数的返回值
同时写状态 不能,纯函数只读 能,update 随行携带
声明位置 add_conditional_edges 节点函数体
典型用途 规则路由 代理间派活与移交

选型口径:路由是纯判断、节点不想背路由职责,用条件边;去向依赖节点内部刚算出的结果,或要顺手捎一条消息给下一个代理,用 Command。多代理场景九成是后者——supervisor 派活要带指令,handoff 移交要带上下文。

6、多代理生态现状

生态层面先把事实交代清楚:langgraph-supervisor 与 langgraph-swarm 两个库在 0.x 时代很流行,但它们建立在已废弃的 create_react_agent 之上,2025 年下半年起基本没有实质更新,与第 5 篇的 create_agent 新 API 并不兼容;两个库的官方参考页与swarm 参考页仍在线,但当前的官方多代理文档已经不再推荐这条路径。截至写作时的现状如下:

对象 状态 路径
langgraph-supervisor 停更 自建调度图
langgraph-swarm 停更 自建移交图
Subagents 模式 官方文档主推 调度代理把子代理当工具调
Handoffs 模式 官方文档主推 代理经工具调用移交控制权
Deep Agents 官方内置推荐 更高层 harness 开箱即用

官方文档当前把多代理归纳为五种模式:Subagents、Handoffs、Skills、Router、Custom workflow。与第 4 节对应:supervisor 对应 Subagents(中心化路由),handoff 与 swarm 对应 Handoffs(控制权转移的机制名)。想少写代码,官方推荐 Deep Agents 这个高层 harness,自带子代理、规划与上下文管理;想在图层面完全掌控,就走本篇的路:create_agent 造代理,Command 组网,LangGraph 兜底。

官方文档开篇先泼冷水:不是每个复杂任务都需要多代理——工具得当、提示词到位的单代理,常常拿到相近结果。

这句提醒值得贴在显示器上:先问一句单代理够不够,再决定组网。

7、审稿流水线实战

机制凑齐,合体做一个真东西:审稿流水线——主编依次派活给内容编辑与技术编辑,意见汇合后收工;技术编辑背后藏着一张自有 schema 的校对子图,用转换层接入。导入沿用前文(TypedDict、Annotated、operator、StateGraph、START、END、Command),校对子图先来:

class ProofState(TypedDict):
    text: str            # 子图自己的黑板
    issues: list

def check_terms(s: ProofState) -> dict:
    return {"issues": ["术语不一致"] if "AI" in s["text"]
            else []}

def check_length(s: ProofState) -> dict:
    return {"issues": s["issues"] + (["篇幅过短"]
            if len(s["text"]) < 50 else [])}

pb = StateGraph(ProofState)
pb.add_node("check_terms", check_terms)
pb.add_node("check_length", check_length)
pb.add_edge(START, "check_terms")
pb.add_edge("check_terms", "check_length")
pb.add_edge("check_length", END)
proof_graph = pb.compile()               # 校对子图完工

解读:ProofState 只有 text 与 issues 两个键,与外层黑板无共享键——这是刻意为之,逼出第 3 节的转换层。两个检查节点串成直线,编译完就是一块能独立测试的零件。外层黑板与全部节点如下:

class ReviewState(TypedDict):
    draft: str
    stage: int
    notes: Annotated[list, operator.add]   # 意见汇合

def tech_edit(s: ReviewState) -> Command:
    r = proof_graph.invoke({"text": s["draft"], "issues": []})
    note = "技术编辑:" + ";".join(r["issues"] or ["无问题"])
    return Command(goto="chief", update={"notes": [note]})

def content_edit(s: ReviewState) -> Command:
    return Command(goto="chief",
                   update={"notes": ["内容编辑:标题可更抓人"]})

def chief(s: ReviewState) -> Command:
    nxt = {1: "content_edit", 2: "tech_edit"}.get(s["stage"] + 1)
    if nxt:
        return Command(goto=nxt, update={"stage": s["stage"] + 1})
    return Command(goto=END, update={"stage": s["stage"] + 1})

解读:三件事。其一,chief 用 stage 记派活进度,每个节点都用 Command 给出下一跳与增量,全图没有一条 add_conditional_edges;其二,notes 挂 operator.add,两位编辑的意见在第 3 篇的合并语义下汇合;其三,tech_edit 就是转换层——进去时把 draft 翻译成 text,出来时把 issues 包装成意见写回 notes。建图运行如下:

builder = StateGraph(ReviewState)
builder.add_node("chief", chief)
builder.add_node("content_edit", content_edit)
builder.add_node("tech_edit", tech_edit)   # 转换层包着子图
builder.add_edge(START, "chief")           # 入口直接交给主编
graph = builder.compile()

r = graph.invoke({"draft": "LangGraph 子图与多代理",
                  "stage": 0, "notes": []})
print(r["notes"])

输出如下:

['内容编辑:标题可更抓人', '技术编辑:篇幅过短']

解读:三次派活、两条意见——主编按站调度,编辑各干一段,子图藏在技术编辑背后干细活,supervisor 拓扑与转换层在同一张图里各就各位。想看子图内部的事件,stream 加 subgraphs=True;把派活进度换成模型结构化输出、两个检查节点换成真校验 API,图结构一字不改。派活循环也是循环,recursion_limit 兜底依旧在场(1.0.6 起默认 1000)。

8、总结

子图把封装带回图世界:State 键名兼容时,编译图直接 add_node,状态双向自动共享;schema 对不上就包一层转换函数,进出翻译、彻底解耦。多代理的经典拓扑三式——supervisor 中央调度、handoff 交接棒、swarm 全互联——落地全靠 Command(goto=..., update=...):一个返回值同时携带增量与去向,派活带指令、移交带上下文。生态上 langgraph-supervisor 与 langgraph-swarm 已停更,官方口径收敛为 Subagents、Handoffs 等模式与 Deep Agents,自建路径就是本篇这套手艺。

练习两道。练习一:给流水线加第三个编辑“标题编辑”——chief 的派活序列加一站,notes 里汇出第三条意见。练习二:把 supervisor 式改成 handoff 式——去掉 chief 的循环调度,content_edit 干完直接 Command(goto=“tech_edit”) 移交,跑通后用 stream 对比两种拓扑的黑板轨迹。对于想继续深入的读者,第 8 篇把 checkpointer 的完整机制展开——子图与多代理挂上持久化,才真正具备生产形态。

相关推荐

获取系统SDK版本、判断手机ROM
精选Code

获取系统SDK版本、判断手机ROM

Build获取系统SDK版本Android中部分API的使用,需要在特定的SDK版本之后才能使用,因此在兼容老版本SDK的时候,经常需要判断API的版本。各种Android版本的对应关系参考《Android各版本对应的SDK版本》判断手机ROM有时候需要判断手机系统的ROM,检测ROM是MIUI、EMUI还是Flyme,可以使用getprop命令,去系统build.prop文件(关于build.p

4.2k
AndroidStudio中各种中文乱码问题
精选Code

AndroidStudio中各种中文乱码问题

AndroidStudio中各种中文乱码问题1、编译Java错误信息乱码在出现这个Annotation processors must be explicitly declared now...问题的时候,正好也发现AndroidStudio Build Output错误信息都乱码。经常遇到各种问题,习惯了,现在遇到的问题,都是以后的答案。 1.1、修改项目build.gradle(无效)在整个p

2.8k
PydanticAI 详解:类型安全的 Python AI SDK
精选AI

PydanticAI 详解:类型安全的 Python AI SDK

PydanticAI 详解:类型安全的 Python AI SDK写 LLM 应用最疼的两件事:模型返回的 JSON 不合预期、换一家模型商要改一串代码。pydantic/pydantic-ai(PydanticAI)用 Pydantic 团队的老本行回答了第一件——输出用 BaseModel 定义、运行前先校验,错误从运行时提前到编码时;第二件则用"模型即字符串"回答:换个厂商只改一个参数。如今

75