LangGraph 并行与 map-reduce

第 11 篇的 CRAG 图把检索、打分、改写串成一条纠错链，单线程问答够用；任务一旦变成"一批文档并行摘要、二十条查询并行检索"，串行就是灾难。第 4 篇亮过一次并行——条件边返回节点名列表——但它有个隐藏前提：列表里的节点名是代码里写死的，所有支路还共享同一份黑板。数量到运行期才知道、每路只需要自己那份输入的活，它接不住。第 4 篇埋的伏笔——路由函数返回 Send 对象——正是为此准备，本篇是《LangGraph 从入门到实战》第 12 篇，把它正面兑现。

版本口径与全系列一致：langgraph 1.2.x，PyPI 独立发布，Send 的规则以 Graph API 文档的 Edges 一节为准。全文按"静态并行的局限 → Send 基础用法 → map-reduce 骨架 → 与 Command 对照 → 并发写风险 → 性能账 → 多文档摘要实战 → 总结"展开。

1、静态并行的局限

已经学过两种并行。其一，普通边扇出：同一个源节点 add_edge 到多个目标，第 3 篇复现冲突时两条 START 出边就是这么搭的；其二，条件边路由函数返回节点名列表，第 4 篇的风控双线。两者共享同一条官方规则：列表里的目标节点作为下一个超步并行执行。它们也共享同一个天花板——扇出的路数在建图期就写死了：节点名是代码里的字面量，写几个就是几路；每一路拿到的又都是同一份共享 State，各路要自己去黑板里认领属于自己的那份数据。

两档并行的分界如下：

一句话立住分界：数量已知、路线固定，节点名列表够用；数量跟着数据走、每路各干各的活，就得 Send。官方文档把后者明确划给 map-reduce 设计模式——首个节点产出一批对象，下游节点要逐个作用到所有对象上，对象数事前未知，每路要的输入还各不相同。

2、Send 基础用法

先把定义钉死。Send 是 langgraph.types 里的一个类，官方口径只吃两个参数：目标节点名、传给该节点的那份 State。用法也只有一个官方位置：在条件边的路由函数里返回 Send 实例的列表，返回多少条 Send，下一超步就并行调度多少个目标节点实例。最小示例如下，两个主题各产一条口号：

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

class State(TypedDict):
    topics: list
    slogans: Annotated[list, operator.add]   # 汇合各实例的增量

def slogan_worker(s: dict) -> dict:          # 只认 Send 给它的输入
    return {"slogans": [f"{s['topic']}：一键直达"]}

def fan_out(state: State) -> list:           # 路由：一份主题一条 Send
    return [Send("slogan_worker", {"topic": t})
            for t in state["topics"]]

builder = StateGraph(State)
builder.add_node("slogan_worker", slogan_worker)
builder.add_conditional_edges(START, fan_out)   # 条件入口：进图即扇出
builder.add_edge("slogan_worker", END)
r = builder.compile().invoke({"topics": ["网盘", "相册"], "slogans": []})
print(r["slogans"])

输出如下：

['网盘：一键直达', '相册：一键直达']

解读，五个要点连读。其一，条件边挂在 START 上做条件入口（第 4 篇提过的用法），图一启动就按输入数据扇出；挂在普通节点后面同理。其二，slogan_worker 的入参类型写的是 dict 而非全图 State——它收到的就是 Send 第二参那份独立输入，不是共享黑板，topics 它根本看不见。其三，同一个节点名被两条 Send 指向，下一超步就是两个并行实例；这正是与节点名列表的本质差异——列表激活的是几个固定节点，Send 激活的是一个节点的多份实例。其四，worker 返回的增量照常走合并规则写回主黑板，operator.add 把两条口号汇成一个列表。其五，topics 塞二十个主题，代码一个字不用改。

3、map-reduce 骨架

Send 的主场是 map-reduce：map 阶段把一批数据逐份派给同一个 worker 节点并行处理，reduce 阶段把所有成果汇成一份结果。官方实战指南的示例（给每个主题各生成一个笑话再汇总）就是这个形状，抽象成拓扑如下：

解读：三个角色各司其职。路由函数是 map 的发动机，读数据列表、逐份产 Send；worker 同超步并行，每实例只处理自己那份输入、只返回增量；汇总节点是 reduce 的收口。注意汇总节点与 worker 之间是普通固定边——同超步的全部实例结束，超步才翻页，汇总节点恰好只跑一次。完整小例（三份报告并行打分再汇总平均，导入同第 2 节）如下：

class State(TypedDict):
    docs: list
    scores: Annotated[list, operator.add]   # 各实例增量汇合
    report: str

def score_worker(s: dict) -> dict:          # map：每份文档一个实例
    return {"scores": [len(s["doc"])]}      # 演示口径：按长度打分

def fan_out(state: State) -> list:
    return [Send("score_worker", {"doc": d})
            for d in state["docs"]]

def summarize(state: State) -> dict:        # reduce：读汇合后的列表
    avg = sum(state["scores"]) // len(state["scores"])
    return {"report": f"共 {len(state['scores'])} 份，平均 {avg} 字"}

建图连边与运行如下：

builder = StateGraph(State)
builder.add_node("score_worker", score_worker)
builder.add_node("summarize", summarize)
builder.add_conditional_edges(START, fan_out)
builder.add_edge("score_worker", "summarize")   # 全部完工才进汇总
builder.add_edge("summarize", END)
print(builder.compile().invoke(
    {"docs": ["春", "夏至", "秋分露凝"], "scores": []})["report"])

输出如下：

共 3 份，平均 2 字

解读：三个实例各贡献一个长度进 scores，operator.add 在超步边界把它们汇成完整列表，summarize 进场时读到的已经是三份成果之和——reduce 的一半由 reducer 自动完成，另一半才是汇总节点的活。想亲眼看并行实况，用第 10 篇的 stream 配 updates 模式回放，同一超步里会看到三条同名的 score_worker 片段次第落账。

4、与 Command 对照

第 7 篇讲过 Command 的 goto 接受节点名列表，一次并行派活给多个编辑；本篇的 Send 又是一路并行。三个机制放在一起，选型就不纠结了，对比如下：

解读：三问定位。问路数固不固定——固定就直接写列表；问派活的人是不是节点自己、要不要捎状态增量——是就用 Command，第 7 篇 supervisor 派活带指令正是它；问路数是不是跟着数据走、每路是不是各干一份活——是就 Send。三者也不互斥：一张图里 map 段用 Send 扇出，汇总之后的去向交给条件边或 Command，各管一段。第 7 篇引过的官方告诫在这里同样成立：同一个节点的出边机制别混用，固定边、条件边、Command 选一种。

5、并发写风险

第 3 篇把规则钉过一次：同一个超步里多个节点写同一个没有 reducer 的字段，抛 InvalidUpdateError。map-reduce 是这条规则的高压线场景——N 个 worker 实例几乎必然写同一个汇总键。把第 2 节整段拷进全新脚本、只把 State 摘掉 Annotated 复跑，翻车现场如下：

class State(TypedDict):            # 摘掉 reducer 的版本
    topics: list
    slogans: list                  # 没有 reducer：并行写同键

# 其余节点与建图与第 2 节一字不差，重跑同一个 invoke

输出如下：

InvalidUpdateError: At key 'slogans': Can receive only one value per step.
Use an Annotated key to handle multiple values.

解读：两个实例同一超步各带回一份 slogans 更新，覆盖语义下框架无法回答留哪份，宁可报错不肯悄悄丢数据——与第 3 篇串讲的是同一个错。解法还是那两条，但主次分明：给汇总键挂 reducer 是 map-reduce 的默认姿势，列表成果用 operator.add，计数字段用同款累加，去重这类语义第 3 篇的自定义函数六行解决；让各实例写不同的键理论上可行，路数不定时下游拼装会写得很难看，不作首选。

路数开得再多，黑板只有一个——并行写同键，先想 reducer。

6、并行的性能账

并行不是免费的，但账算得清。模型调用是 IO 型负载，绝大多数时间在等网络返回，同一超步的 N 个 worker 各等各的，墙钟时间接近最慢一路而不是各路之和。以单次调用约 3 秒做量级示意（示意口径，非实测基准）：

解读：三条边界划清楚。其一，省的是墙钟，省不了成本——token 一个不少，并发只是把等待叠在一起。其二，上限来自服务端限流：路数一多，模型服务的并发配额会以 429 顶回来；稳妥做法是分批，把扇出从 START 挪到一个分派节点后面，一次只产十来条 Send，汇总节点看黑板里还有剩余就沿第 4 篇的回边再扇下一批。其三，CPU 密集任务别指望同比例加速，解释器与线程调度都要打折，IO 型才是并行的主场。20 份从约 60 秒压到 10 秒之内，已经是量级差异。

7、多文档摘要实战

机制凑齐，合成一个真东西：多文档并行摘要——N 段文字各自出一份摘要与关键词，汇总成一篇简报。摘要与关键词用简化规则演示，换真模型只改 worker 一个函数体。State 与 worker 如下：

import operator
from collections import Counter
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send

class State(TypedDict):
    docs: list
    briefs: Annotated[list, operator.add]     # 摘要汇合
    keywords: Annotated[list, operator.add]   # 关键词汇合
    report: str

def digest_worker(s: dict) -> dict:           # 每份文档一个实例
    doc = s["doc"]
    brief = doc[:12] + "……"                  # 演示口径：截断当摘要
    words = [w for w in doc.split() if len(w) > 2]
    top = Counter(words).most_common(2)       # 演示口径：高频当关键词
    return {"briefs": [brief], "keywords": [w for w, _ in top]}

解读：worker 只认 Send 递进来的那份 doc，产出一条摘要加两个关键词，两个键都等着汇合。路由与汇总节点如下：

def fan_out(state: State) -> list:
    return [Send("digest_worker", {"doc": d})
            for d in state["docs"]]

def make_report(state: State) -> dict:
    freq = Counter(state["keywords"])
    hot = "、".join(f"{w}×{n}" for w, n in freq.most_common(3))
    lines = [f"- {b}" for b in state["briefs"]]
    return {"report": "简报：\n" + "\n".join(lines)
            + f"\n高频词：{hot}"}

建图连边与运行如下：

builder = StateGraph(State)
builder.add_node("digest_worker", digest_worker)
builder.add_node("make_report", make_report)
builder.add_conditional_edges(START, fan_out)      # 按文档数扇出
builder.add_edge("digest_worker", "make_report")   # 全部完工再汇总
builder.add_edge("make_report", END)
graph = builder.compile()
r = graph.invoke({"docs": [
    "langgraph lets agents loop until done",
    "reducer merges parallel updates safely",
    "send fans out one worker per item"],
    "briefs": [], "keywords": []})
print(r["report"])

输出如下：

简报：
- langgraph le……
- reducer merg……
- send fans ou……
高频词：langgraph×1、lets×1、reducer×1

解读：四件事。其一，三个 digest_worker 实例同超步并行，briefs 与 keywords 两键各靠 operator.add 汇合，第 5 节的规矩全部在场。其二，汇合顺序不保证与输入同序——各实例完成先后不定，reducer 按完成次序拼接；摘要列表要跟输入对齐，就在 Send 的输入里带序号，make_report 排序后再用。其三，换真实现只改 digest_worker：截断换成一次模型摘要调用、高频词换成结构化输出，图结构与路由一字不动。其四，想盯执行实况，第 10 篇的 stream 配 updates 能看到同超步三段同名片段。

8、总结

并行的两档分界记牢：路数固定、各路共享黑板，条件边返回节点名列表就够；路数由数据决定、每路各带一份输入，用 Send——from langgraph.types import Send，路由函数里返回 Send 列表，返回几条下一超步就并行几个实例。map-reduce 骨架三件套：Send 扇出、worker 只写增量、汇总键挂 reducer；worker 到汇总节点之间一条固定边，管住"全部完工才收口"。并发写同键是唯一高压线，reducer 挂上，冲突变汇合。性能账上 IO 型负载收益最大，成本不省、限流要防、分批靠回边。

练习两道。练习一：给实战图补顺序——fan_out 改用 enumerate 产序号，Send 输入带上 i，make_report 按 i 排序 briefs 再进简报，验证输出与输入同序。练习二：给扇出加上限——把条件边从 START 挪到一个 dispatch 节点后面，fan_out 一次最多产 3 条 Send，make_report 判断 docs 还有剩余就沿回边回 dispatch 再扇一批，跑 7 份文档观察分批轨迹。对于手头正有"一批条目逐个过模型"任务的读者，建议直接用本篇骨架替换 for 循环：路由按数据产 Send，汇总键挂 operator.add，worker 换成真调用——当轮就能看到墙钟时间的量级变化。