第 7 章 动态并行：Send 与 MapReduce

本章目标：掌握 Send——LangGraph 中"运行时才知道并行几路"的扇出机制。学完本章，你能实现标准的 MapReduce 工作流：N 份输入并行处理、结果自动归并，而 N 在编译期完全未知。

7.1 需求升级：批次评估

业务又演进了。之前所有版本都在评估一份简历，现在产品要求：

HR 一次提交一批简历（数量不定，可能 3 份也可能 30 份），系统并行评估每一份，最后产出一份批次汇总报告。

这个需求里藏着一个新维度的难题：并行的路数在编译期未知。

回顾一下目前掌握的并行手段——条件边返回多标签、Command 的 goto 列表——它们有一个共同的天花板：目标节点集合是图结构的一部分，编译时就固定了。你可以写出"tech_review 和 background_check 两路并行"，因为这两路在写代码时就存在于你的脑子里。但"30 份简历、每份一路"——你不可能在图里预先注册 30 条分支，更不知道下一批是 30 份还是 3 份。

需要的是把"执行几路、每路处理什么"从图结构里拿出来，变成运行时数据。

7.2 Send：把执行计划变成数据

Send 就是这个"执行计划的原子"：一个 Send 描述"去执行某节点，并喂给它这份数据"。

from langgraph.types import Send

Send("evaluate", {"job": job, "resume": resume_text})
#      ^目标节点    ^这份实例的私有载荷

路由函数不再返回标签，而是返回 Send 列表——列表多长，就扇出几路；每个 Send 带自己的一份载荷。列表本身在运行时根据状态计算，于是"并行几路"彻底变成数据问题。

Send 通常挂在条件边上（不需要映射表——Send 自带目标节点）：

def fan_out(state) -> list[Send]:
    return [Send("evaluate", {"resume": r}) for r in state["resumes"]]

builder.add_conditional_edges("prepare", fan_out)

也可以作为 Command(goto=[...]) 的元素由节点直接返回——两种入口，同一种语义。本章统一用条件边入口。

7.3 完整 demo：批次简历评估

下面是完整的批次评估图。prepare 模拟"接收一批简历"，evaluate 是被 Send 扇出的评估节点，summarize 归并出批次报告：

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

# ---- 全局状态：整批共享 ----
class BatchState(TypedDict):
    job: str
    resumes: list[str]
    reviews: Annotated[list[str], operator.add]   # 各份评估结论：追加
    total_score: Annotated[int, operator.add]     # 各份得分：累加

# ---- 私有载荷：单个评估实例拿到的输入 ----
class EvalPayload(TypedDict):
    job: str
    resume: str

def prepare(state: BatchState) -> dict:
    # 真实场景：这里会做简历清洗/去重等预处理，演示中直接放行
    return {}

def fan_out(state: BatchState) -> list[Send]:
    # 关键：扇出计划按运行时的简历数量计算
    return [
        Send("evaluate", {"job": state["job"], "resume": r})
        for r in state["resumes"]
    ]

def evaluate(payload: EvalPayload) -> dict:
    # 注意：这里的入参是 Send 的私有载荷，不是全局 BatchState！
    name = payload["resume"].split("：")[0]
    length = len(payload["resume"])
    score = min(100, 40 + length % 50)            # 演示用伪评分
    review = f"{name} 得分 {score}（依据岗位 {payload['job']}）"
    return {"reviews": [review], "total_score": score}

def summarize(state: BatchState) -> dict:
    n = len(state["resumes"])
    avg = state["total_score"] // max(1, n)
    return {"summary": f"共评估 {n} 份简历，平均分 {avg}"}

builder = StateGraph(BatchState)
builder.add_node("prepare", prepare)
builder.add_node("evaluate", evaluate)
builder.add_node("summarize", summarize)

builder.add_edge(START, "prepare")
builder.add_conditional_edges("prepare", fan_out)   # Send 扇出：无需映射表
builder.add_edge("evaluate", "summarize")
builder.add_edge("summarize", END)
graph = builder.compile()

batch = {
    "job": "Python 后端工程师",
    "resumes": [
        "李思远：六年电商后端经验，主导订单中台重构",
        "王敏：四年金融系统开发，熟悉高并发交易链路",
        "赵子昂：八年基础架构，Kubernetes 与服务治理",
    ],
}
result = graph.invoke(batch)
print(result["summary"])    # 共评估 3 份简历，平均分 63
print(result["reviews"])    # 三条评估结论，顺序归并

换一批 5 份简历再 invoke，扇出路数自动变成 5——图结构一行不改。这就是"执行计划是数据"的含义。

7.4 Send 的两条关键语义

语义一：目标节点收到的是私有载荷，不是全局状态。

evaluate 的入参是 EvalPayload 而非 BatchState——它看不到 resumes 列表，也看不到其他实例写了什么。这不是缺陷而是设计：每个并行实例是独立的 map 任务，天然隔离。推论很重要：节点需要什么，Send 的载荷里就必须带什么。fan_out 里把 job 逐份打包进载荷，正是因为 evaluate 读不到全局的 job 字段。漏带依赖是 Send 场景第一高频 bug（症状：节点内 KeyError）。

语义二：节点返回的补丁，写回的是全局状态。

evaluate 返回 {"reviews": [...], "total_score": ...}，合并进的是 BatchState 的通道。3 个实例在同一个超步各写一次，operator.add 把三份补丁归并——这正是第 3 章 reducer 与第 5 章超步合并规则在此处的会师：

另一个必然推论：被 Send 触发的节点会写的那几个全局字段，必须有合并语义的 reducer。如果 total_score 用默认覆盖语义，同一超步内 N 次写入会让引擎直接抛 InvalidUpdateError（第 5 章解释过：默认通道在一个超步内只接受一次写入）。在 Send 场景下，reducer 不是"锦上添花的正确性保障"，而是能跑起来的硬前提。

7.5 MapReduce 视角

把 7.3 节的图抽象一下，就是教科书式的 MapReduce：

Send 场景里所有实例在同一个超步完成（它们由同一次 fan_out 触发），因此 evaluate → summarize 这条汇合边天然满足"所有入边同超步触发"——summarize 恰好执行一次。第 5 章的汇合陷阱在这里不会发生，除非你让不同的 Send 在不同超步触发（比如扇出前还有别的节点在跑）。保持扇出"一次性、同超步"，是 MapReduce 图结构清晰的保证。

顺带一个常见误解：Send 不是"把一个节点变多份"的 hack，它和条件边多选、goto 列表是同一族机制的不同表达力层级——静态扇出（编译期定路数） → 动态选择（运行时挑分支） → 动态扇出（运行时定路数）。三者可以组合：比如先条件边判断"这批简历要不要走深度评估"，需要时再 Send 扇出。

7.6 载荷是独立的，别共享可变对象

fan_out 生成载荷时有个隐蔽纪律：每个 Send 的载荷应当是独立对象，尤其避免多个 Send 共享同一个可变容器。

# 危险写法：三个 Send 共享同一个 list 对象
shared = []
sends = [Send("evaluate", {"history": shared}) for _ in range(3)]
# 任一实例修改 history，其余实例跟着遭殃——并行隔离被击穿

# 正确写法：逐份构造
sends = [Send("evaluate", {"history": []}) for _ in range(3)]

多数场景你只是把不可变的字符串/数字放进载荷，天然安全；一旦载荷里出现 list/dict，就问自己一句："会不会有多个实例拿到同一个对象？"

常见坑与排错

1. 节点里读不到全局状态（KeyError）。Send 触发的节点只拿载荷。凡是节点用到的字段——岗位要求、配置、上下文——必须在 fan_out 时逐份塞进载荷。排查时打印节点入参，和 Send 构造处对照。

2. InvalidUpdateError: Can receive only one value per step。N 个实例同超步写默认通道。给被写字段加 Annotated[T, operator.add] 或自定义 reducer。这个报错在 Send 场景出现得非常频繁，值得记住原文。

3. 以为 evaluate 实例之间能互相看到数据。同超步语义（第 5 章）在 Send 下同样生效：实例们共享的只是"超步起始快照"，而且它们读的是各自的载荷，连快照都不共享。实例间通信的唯一途径是"写全局状态 → 下一超步的节点读"。

4. 扇出数量失控。30 份简历就 30 个实例，每个实例里再调一次 LLM——token 与并发压力按 N 倍放大。生产上要配合批次切分（每批最多 K 份）与第 16 章的并发/容错策略。

5. Send 目标节点名拼错。和 Command 的 goto 一样是运行时数据，编译期不校验。节点名用常量。

6. 在 fan_out 里做重活。fan_out 是路由函数，第 4 章定下的纪律同样适用：保持廉价、哑、可重放。需要 LLM 参与的"智能分批"应该在前置节点里做完、把结果写进状态，fan_out 只做纯粹的循环展开。

动手练习

1. 给 7.3 节的图加"批次上限"：超过 5 份时只评估前 5 份，summarize 的报告里注明"本批仅处理前 5 份"。只改 fan_out 和 summarize，思考：被丢弃的简历信息要不要进状态？

2. 把评分逻辑换成真实 LLM 调用：evaluate 内部用 ChatOpenAI + 结构化输出给单份简历打分（提示词里用 payload["job"] 和 payload["resume"]）。用 3 份真实简历跑通，观察 3 个实例的流式事件是否落在同一超步。

3. 进阶：实现两层扇出——外层 Send 按"岗位"扇出，每个岗位实例内部再对"该岗位的简历列表"二次 Send 扇出。注意全局字段与私有载荷的边界在哪一层，reducer 需要怎么设计才能让两层结果都正确归并。

本章速查