生命不息，折腾不止。上一课你的 Agent 已经能「一个老板带一队专家」了，但专家还是一个一个上、活还是一个一个干。今天教你把同一批互不相干的活拆开同时干，再把结果收回汇总，全程不用改任何一个专家的内部代码。

一、第四课留下的尾巴：专家其实还在「排队」

第四课用 Supervisor 派活，看着是一队专家协同，实际执行时每个时间步只让一个专家上场：老板派活 → 甲专家干完交回 → 老板再派乙。可现实中一大批活彼此根本没依赖——翻译十篇文档、给五百条用户评论分类、并行搜索几个信源——这么串行干就是纯浪费时间。

LangGraph 给「同一类活、批量干」的场景备了两个家伙：Send 负责把一份活扇出成 N 份同时派出去，reducer 负责把 N 份结果收回来合并。合起来就是经典的 map-reduce 模式：map 拆开并行干，reduce 收回来汇总。

二、Send：一条返回值，扇出 N 个分身

Send 的用法一句话：在一个「条件边」函数里，返回一串 Send(节点名, 状态参数)，LangGraph 就会在同一时间步把这些节点全部并行跑一遍。 每个 Send 只塞那个分支自己需要的那一点点状态，不用塞整份大状态。

先看最小例子（官方文档原样，能直接跑）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from typing_extensions import TypedDict, Annotated
import operator

class OverallState(TypedDict):
    subjects: list[str]                          # 待处理的 N 个主题
    jokes: Annotated[list[str], operator.add]     # 并行结果累加的地方

def continue_to_jokes(state: OverallState):
    # 每个 subject 派一个独立任务，各带各的 subject
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

builder = StateGraph(OverallState)
builder.add_node(
    "generate_joke",
    lambda state: {"jokes": [f"Joke about {state['subject']}"]},
)
builder.add_conditional_edges(START, continue_to_jokes, ["generate_joke"])
builder.add_edge("generate_joke", END)
graph = builder.compile()

graph.invoke({"subjects": ["cats", "dogs"]})
# {'subjects': ['cats', 'dogs'], 'jokes': ['Joke about cats', 'Joke about dogs']}

三个关键点拆开讲：

  1. Send(节点名, 参数字典)：第一个参数 node 是要并行的那个节点（generate_joke），第二个参数 arg 是只给这个分支用的参数。这是精髓——每个并行分支只拿自己那份，不拿全量状态，上下文不冗余；
  2. 挂在 add_conditional_edges 上扇出：把返回 Send 列表的函数写在条件边里，调度器会把这些 Send 全收集起来，在同一时间步并行执行，跑完才进入下一步。注意这里的第二参数 ["generate_joke"] 是声明这些 Send 可能落到的目标节点；
  3. Annotated[list, operator.add] 这个 reducer 绝对不能省：并行分支同时往 jokes 里写结果，没有 reducer 就是「后写的覆盖先写的」，最后只会剩最后跑完那个分支的一条结果。加了 operator.add，LangGraph 才会把每个分支的返回追加合并进同一个列表。

导入路径提醒：新版用 from langgraph.types import Send，老一点的环境里 from langgraph.constants import Send 也常见，装最新版照前面这个写即可。

三、实战：长文档 map-reduce，拆分并行到汇总

拿个真实场景练手：你有一批长文档（或一篇几万字长文切成 N 段），要逐段总结再合成一篇总报告。串行跑一篇篇等；用 Send 拆开，一次全上。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class State(TypedDict):
    docs: list[str]                                # N 篇/段待总结的文档
    summaries: Annotated[list[str], operator.add]  # 并行摘要的结果池
    report: str

# map：把每篇文档扇出一个并行任务
def fan_out(state: State) -> list[Send]:
    return [Send("summarize_one", {"doc": d}) for d in state["docs"]]

# worker：每个并行分支只处理自己那一篇
def summarize_one(state):
    summary = llm.invoke(f"用一句话总结下面的内容：\n{state['doc']}")
    return {"summaries": [summary.content]}

# reduce：所有分支都跑完才轮到它——汇总成报告
def aggregate(state: State) -> State:
    joined = "\n\n".join(f"{i+1}. {s}" for i, s in enumerate(state["summaries"]))
    report = llm.invoke(f"把下面这些摘要合成一篇总报告：\n{joined}")
    return {"report": report.content}

builder = StateGraph(State)
builder.add_node("summarize_one", summarize_one)
builder.add_node("aggregate", aggregate)
builder.add_conditional_edges(START, fan_out, ["summarize_one"])
builder.add_edge("summarize_one", "aggregate")
builder.add_edge("aggregate", END)
graph = builder.compile()

跑的时候：

1
2
result = graph.invoke({"docs": ["段落A", "段落B", "段落C", "段落D"]})
print(result["report"])

A、B、C、D 四段会同时开跑，全部完成后 aggregate 才执行——LangGraph 自动帮你做了「汇合点」（fan-in）：所有并行分支到齐才继续往下走，一行 join 逻辑都不用你自己写。

这里 llm 就是前几课你已经接好的模型客户端（ChatOpenAI 之类，指向任意 OpenAI 兼容接口即可）。想省点接入成本，用中转站 ai.aklibk.com 这种国内直连、免绑卡、人民币按量付费的也行，接口格式跟官方一致，代码一行不用改。

看完这段你可能想问：上一课的 Supervisor 里，那几个专家能不能也并行？能，但专家之间往往有依赖（先调研再计算），不是纯独立活；文档总结这种「互不相干、同类活批量」才是 Send 的主场。判断标准就一条：这批任务之间有没有依赖，没依赖就上 Send。

四、子图：把每个专家打包成可复用模块

并行能跑了，代码也越写越长。等专家逻辑多起来，每个专家内部还自带好几个节点（比如「调研专家」内部 = 搜索 + 筛选 + 写小结三步），全摊在一个大图里，节点名会撞、状态互相踩，维护起来想死。

这时候上子图（Subgraph）：把一套完整逻辑单独建一个 StateGraph 编译好，然后当成一个节点塞进父图。官方给了两种接法，看状态 key 是否一致：

  • 父子共享 key → 直接把编译好的子图传给 add_node，零胶水；
  • 父子 key 不同 → 外面包一层函数，先把父状态转成子图输入，跑完再转回来。

先说最省事的第一种，官方例子（能直接跑）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
from langgraph.graph import START, StateGraph
from typing import TypedDict

# —— 子图：一套独立小逻辑 ——
class SubgraphState(TypedDict):
    foo: str   # 这个 key 与父图共用
    bar: str   # 子图独有

def subgraph_node_1(state: SubgraphState):
    return {"bar": "bar"}

def subgraph_node_2(state: SubgraphState):
    return {"foo": state["foo"] + state["bar"]}  # 用到子图独有的 bar

sub_builder = StateGraph(SubgraphState)
sub_builder.add_node(subgraph_node_1)
sub_builder.add_node(subgraph_node_2)
sub_builder.add_edge(START, "subgraph_node_1")
sub_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = sub_builder.compile()

# —— 父图 ——
class ParentState(TypedDict):
    foo: str

def node_1(state: ParentState):
    return {"foo": "hi! " + state["foo"]}

builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)   # 子图直接当节点，共享 foo
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()

graph.stream({"foo": "foo"})
# {'node_1': {'foo': 'hi! foo'}}
# {'node_2': {'foo': 'hi! foobar'}}

看到了没，父图一个 add_node("node_2", subgraph) 就把整套子逻辑接进来了，子图内部那俩节点对父图完全透明。这正是「可复用模块」的意义：同一个「调研专家子图」，A 项目当节点用，B 项目也能复用，还各自带独立状态、独立检查点。

想 debug 子图内部的状态，stream 时要加 subgraphs=True，否则默认只看父图层。

五、避坑清单

  1. 忘加 reducer：items: list[str] 没写 Annotated[..., operator.add]，并行结果互相覆盖，最后「怎么只剩一条」——这是 Send 最高频的翻车点；
  2. 把 Send 存进 state：Send 必须从节点/条件边的返回值发出去，它不是普通数据，别塞进 state 存着；
  3. Send 参数塞全量状态：每个分支只带自己需要的那点参数，别把整份大状态复制 N 份，上下文和开销一起爆炸；
  4. 有依赖的任务硬上 Send：任务之间有先后依赖（乙的输入是甲的输出），拆并行就是乱跑。无依赖才并行；
  5. 在节点里手动 for 循环调子图：想要多次子图运行，别在一个节点里命令式循环 invoke 子图——checkpoint 会撞 MULTIPLE_SUBGRAPHS 命名冲突，官方论坛的明确建议是直接改用 Send 扇出；
  6. 无脑开 N 个分支：fan-out 成本随分支数线性涨，几十个还好，上千个就该在代码层限制并发宽度（cap），别指望框架替你兜底。

六、小结

到这一课，你的 LangGraph 技能树是这样：

  • 第一课：会画图（State / Node / Edge）；
  • 第二课：会调工具、断点续跑、人审、长期记忆；
  • 第三课：能打包成 API 对外服务；
  • 第四课：会带队（Supervisor 派活）；
  • 第五课：会并行——Send 扇出 + map-reduce 汇总 + 子图模块化。

把「专家子图 + Supervisor 调度 + Send 并行 + 长期记忆 + API 打包」拼起来，已经是一套能上生产的多智能体骨架了。Send 解决的是「无依赖的活别傻等」，子图解决的是「逻辑一改到处改」，这俩是让 Agent 系统从「能跑」迈向「能规模化跑」的关键一步。

生命不息，折腾不止。第五课先到这。下一篇是 LangGraph 实战的收官课：把第四课挖下的坑填上——用 Command 精确接管控制权，一个 Agent 干到一半，怎么带着现场状态、带着一句「任务描述」把控制权连锅端转交给下一个 Agent（goto + graph=Command.PARENT），把 Supervisor 那套工具式 handoff 升级成能精确传神的徒手 handoff。学完这篇，多智能体的三件事——派活（Supervisor）、并行（Send）、交接（Command）——你就全齐了。