LANGGRAPH MODULE 4

1 什么是 Map-Reduce？两个阶段的本质

Map-Reduce 是一种经典的并行计算模式，在大数据领域（如 Hadoop、Spark）中广泛使用。LangGraph 将这个思想引入到 LLM 工作流中，用于将一个任务分解为多个子任务并行处理，再汇聚结果。

它由两个阶段组成：

Map（映射/扇出）阶段：把一个大任务拆分成若干个小的、相互独立的子任务，并发地分发给多个 Worker 节点去处理。
Reduce（归约/聚合）阶段：收集所有 Worker 节点的输出，将它们合并成一个最终结果。

Map-Reduce 两阶段示意

输入话题

topic = "animals"

↓

generate_topics

生成子话题列表

Send API 动态扇出 ↓ ↓ ↓

Send("generate_joke", mammals)

↓

Worker 1

mammals 笑话

↓

Send("generate_joke", reptiles)

↓

Worker 2

reptiles 笑话

↓

Send("generate_joke", birds)

↓

Worker 3

birds 笑话

↓

Reducer 自动合并 jokes 列表

best_joke

从所有笑话中选最佳

↓

END

核心价值

Map-Reduce 模式在 LLM 工作流中最大的价值是处理动态数量的并行任务。你事先不知道要处理多少个子任务（比如 LLM 会根据话题生成 3 个或 5 个子话题），Map-Reduce 可以在运行时动态决定 Worker 数量，并等所有 Worker 都完成后再聚合。

与 Module 4-1 普通并行的本质区别

Module 4-1 学过用 add_edge 将多个节点并行执行，那是静态并行——编译时就固定了有哪些并行分支。Map-Reduce 是动态并行——分支数量在运行时根据数据决定，可以是 1 个、3 个或 100 个 Worker，完全弹性。

2 静态并行边的局限：为什么不够用

在 Module 4-1（并行化）中，你已经学会了用普通边实现并行：

# 静态并行：编译时固定有 3 个并行节点
graph.add_edge(START, "node_a")
graph.add_edge(START, "node_b")
graph.add_edge(START, "node_c")  # 永远是这 3 个，不多不少

这种方式的问题在于：并行的分支数量在图编译时就必须确定，它们被"硬编码"进了图的结构里。

特性	静态并行边（Module 4-1）	Map-Reduce / Send API（本节）
Worker 数量	编译时固定，无法改变	运行时动态决定，弹性伸缩
决定时机	Compile time（图构建阶段）	Runtime（图执行阶段）
典型场景	固定数量的并行子任务（如同时调用 3 个 API）	对 LLM 生成的列表并行处理（数量不确定）
每个 Worker 的输入	共享同一个 State	每个 Worker 拥有自己独立的子 State
结果收集	所有节点写回同一个 OverallState 字段	通过 Reducer 自动将各 Worker 输出合并到列表

核心矛盾

如果你想对 LLM 生成的子话题列表（可能是 3 条，也可能是 5 条）每一条都并行地调用 LLM 生成内容，静态并行边无法做到——因为列表的长度在编译时未知，只有运行时才能知道。这就是 Send API 出现的原因。

一个说明问题的具体例子

假设你让 LLM 根据话题 "animals" 生成子话题，LLM 返回了 ["mammals", "reptiles", "birds"]。下一次输入话题 "programming"，LLM 可能返回 ["Python", "Rust", "Go", "TypeScript", "Java"]（5 个）。

你不可能在写代码时用静态并行边来处理这种情况——因为你不知道要写几条 add_edge。唯一的解法就是在运行时动态生成 Worker，这正是 Send API 的工作。

3 Send API：动态派发工作的核心机制

Send 是什么？

Send 是 LangGraph 提供的一个特殊对象，用于在运行时动态地向某个节点派发一个任务，同时携带该任务专属的状态数据。

from langgraph.types import Send

# Send 的语法
Send(
    node_name,  # str：目标节点的名称
    state       # dict：传给该节点的状态（可以不是 OverallState 的子集！）
)

关键特性：

node_name：必须是图中已注册的节点名（字符串）。
state：传递给该 Worker 节点的状态字典。这个字典不需要与 OverallState 对齐——Worker 节点可以有自己的私有 State 结构（JokeState）。
多个 Send 对象会被 LangGraph 并行执行，就像同时启动多个独立的 Worker 线程。

Send 如何被触发？——条件边返回 Send 列表

Send 对象并不直接调用，而是从条件边（conditional edge）的路由函数中返回。当路由函数返回一个 Send 对象的列表时，LangGraph 会将列表中的每个 Send 并行地分发出去：

from langgraph.types import Send

# 路由函数：读取 subjects 列表，为每个 subject 创建一个 Send 对象
def continue_to_jokes(state: OverallState):
    # 返回一个 Send 对象的列表
    # LangGraph 看到 Send 列表，就知道要并行分发
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]
    #         ↑ 目标节点名       ↑ 每个 Worker 独享的子 State

假设 state["subjects"] 是 ["mammals", "reptiles", "birds"]，上面的列表推导式会生成：

[
    Send("generate_joke", {"subject": "mammals"}),   # Worker 1
    Send("generate_joke", {"subject": "reptiles"}),  # Worker 2
    Send("generate_joke", {"subject": "birds"}),     # Worker 3
]

关键理解：Send 不是普通的"下一步"

普通条件边的路由函数返回一个字符串（节点名），表示"下一步去哪个节点"。而当路由函数返回 Send 对象（或 Send 列表）时，LangGraph 理解为"同时启动多个并行任务"，每个任务有自己独立的状态数据。这是 Send API 最精妙的设计。

Worker 节点可以使用私有 State

这是 Send API 另一个强大之处：Worker 节点（generate_joke）不需要使用 OverallState，它可以定义自己的私有状态结构：

# Worker 节点的私有 State——只有 subject 一个字段
class JokeState(TypedDict):
    subject: str

# Worker 节点使用 JokeState 而不是 OverallState
def generate_joke(state: JokeState):
    prompt = joke_prompt.format(subject=state["subject"])  # 读私有 subject
    response = model.with_structured_output(Joke).invoke(prompt)
    # 但返回写入 OverallState 的 jokes 字段！
    return {"jokes": [response.joke]}

Send 派发时传入的 {"subject": s} 就是 JokeState 的内容，LangGraph 会用它来填充 Worker 节点的输入 State。Worker 的输出（{"jokes": [...]}）则会被写回 OverallState 的对应字段，由 Reducer 合并。

4 State 设计：Annotated Reducer 如何聚合结果

多个 Worker 同时写入同一字段，会冲突吗？

这是一个非常重要的问题。在 Map-Reduce 中，3 个 Worker 节点各自生成一条笑话，都需要写入 OverallState 的 jokes 字段。如果使用默认的"覆盖"语义，只有最后一个写入的 Worker 的结果会保留！

解决方案是使用 Annotated + Reducer 函数，将字段的写入语义从"覆盖"改为"追加合并"：

import operator
from typing import Annotated
from typing_extensions import TypedDict
from pydantic import BaseModel

# 结构化输出的 Pydantic 模型
class Subjects(BaseModel):
    subjects: list[str]   # LLM 输出的子话题列表

class BestJoke(BaseModel):
    id: int               # 最佳笑话的索引

class OverallState(TypedDict):
    topic: str                                    # 输入：用户给定的大话题
    subjects: list                                # Map 节点生成的子话题列表
    jokes: Annotated[list, operator.add]          # ← 关键！Reducer 追加合并
    best_selected_joke: str                       # Reduce 节点选出的最佳笑话

Annotated[list, operator.add] 的含义

Annotated[list, operator.add] 告诉 LangGraph：当多个节点同时写入 jokes 字段时，不要覆盖，而是使用 operator.add（即列表拼接 +）将所有写入的列表合并成一个大列表。这就是"Reducer"的作用。

State 字段的生命周期

topic: "animals"

subjects: []

jokes: []

best_selected_joke: ""

初始输入

→

topic: "animals"

subjects: ["mammals","reptiles","birds"]

jokes: []

best_selected_joke: ""

generate_topics 之后

jokes: []

3 个 Worker 并行执行...

各自写入 ["笑话X"]

Worker 执行中

→

jokes: ["Why mammals...", "Why alligators...", "Why birds..."]

Reducer (operator.add) 自动合并

所有 Worker 完成后

为什么 Worker 写入 jokes 时要包裹成列表？

def generate_joke(state: JokeState):
    ...
    return {"jokes": [response.joke]}  # 注意：是 [joke]，而不是 joke
    # 因为 jokes 字段的类型是 list
    # Reducer operator.add 对 list 做的是列表拼接：
    #   [] + ["joke1"] + ["joke2"] + ["joke3"] = ["joke1","joke2","joke3"]

每个 Worker 返回一个只含一个元素的列表。Reducer operator.add 将这些单元素列表拼接起来，得到所有笑话的完整列表。这是一个非常精妙的设计——保持 Worker 节点的输出格式一致，让 Reducer 自动处理合并。

5 完整示例：笑话生成与评选系统

Notebook 中的示例是一个完整的 Map-Reduce 应用：给定一个大话题，自动生成多个子话题，对每个子话题并行地让 LLM 生成一则笑话，最后由 LLM 从所有笑话中评选出最佳的一则。

第一步：准备 LLM 和提示词

from langchain_deepseek import ChatDeepSeek

# 三个提示词模板
subjects_prompt = """Generate a list of 3 sub-topics that are all related to this overall topic: {topic}."""
# → 根据大话题生成 3 个子话题（LLM 决定具体是哪 3 个）

joke_prompt = """Generate a joke about {subject}"""
# → 根据子话题生成一则笑话

best_joke_prompt = """Below are a bunch of jokes about {topic}. Select the best one!
Return the ID of the best one, starting 0 as the ID for the first joke.
Jokes: \n\n  {jokes}"""
# → 从笑话列表中选出最佳（返回序号）

# 使用 DeepSeek 模型（temperature=0 确保输出稳定）
model = ChatDeepSeek(model="deepseek-v4-pro", temperature=0)

第二步：定义所有 State 和 Pydantic 模型

import operator
from typing import Annotated
from typing_extensions import TypedDict
from pydantic import BaseModel

# LLM 结构化输出的类型定义
class Subjects(BaseModel):
    subjects: list[str]   # 子话题列表

class Joke(BaseModel):
    joke: str             # 单则笑话文本

class BestJoke(BaseModel):
    id: int               # 最佳笑话的索引（0-based）

# 整图共享的 Overall State
class OverallState(TypedDict):
    topic: str                           # 用户输入的大话题
    subjects: list                       # generate_topics 写入，默认覆盖
    jokes: Annotated[list, operator.add] # Worker 各自追加，Reducer 合并
    best_selected_joke: str              # best_joke 节点写入最终结果

# Worker 节点专用的私有 State（只有 subject 一个字段）
class JokeState(TypedDict):
    subject: str

第三步：定义三个节点函数

### 节点 1：generate_topics（Map 的"前置"节点）
# 读取用户输入的大话题，让 LLM 生成子话题列表
def generate_topics(state: OverallState):
    prompt = subjects_prompt.format(topic=state["topic"])
    response = model.with_structured_output(Subjects).invoke(prompt)
    return {"subjects": response.subjects}
    # 写入 subjects 字段，触发后续的 Send 扇出

### 节点 2：generate_joke（Map 阶段的 Worker 节点）
# 接受 JokeState（私有 State），生成一则笑话写回 OverallState.jokes
def generate_joke(state: JokeState):   # ← 注意：使用 JokeState，不是 OverallState
    prompt = joke_prompt.format(subject=state["subject"])
    response = model.with_structured_output(Joke).invoke(prompt)
    return {"jokes": [response.joke]}    # ← 单元素列表，由 Reducer 合并

### 节点 3：best_joke（Reduce 节点）
# 所有 Worker 完成后，从合并后的 jokes 列表中选最佳
def best_joke(state: OverallState):
    jokes = "\n\n".join(state["jokes"])  # 将所有笑话拼成一段文本
    prompt = best_joke_prompt.format(topic=state["topic"], jokes=jokes)
    response = model.with_structured_output(BestJoke).invoke(prompt)
    return {"best_selected_joke": state["jokes"][response.id]}

第四步：定义路由函数（Send 的触发点）

from langgraph.types import Send

# 路由函数：从 subjects 列表中为每个 subject 创建一个 Send 对象
def continue_to_jokes(state: OverallState):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]
    # 返回 Send 列表 → LangGraph 并行执行所有 generate_joke Worker

6 逐行解析：图的构建与条件边的特殊用法

完整的图构建代码

from langgraph.graph import END, StateGraph, START

# ① 以 OverallState 为蓝图创建图
graph = StateGraph(OverallState)

# ② 注册所有节点
graph.add_node("generate_topics", generate_topics)  # Map 前置节点
graph.add_node("generate_joke", generate_joke)        # Worker 节点（会被多次并行实例化）
graph.add_node("best_joke", best_joke)                # Reduce 节点

# ③ START → generate_topics：固定边，图从 generate_topics 开始
graph.add_edge(START, "generate_topics")

# ④ 核心！条件边：generate_topics → [Send, Send, ...] → generate_joke
graph.add_conditional_edges(
    "generate_topics",     # 源节点
    continue_to_jokes,      # 路由函数（返回 Send 列表）
    ["generate_joke"]       # 可能的目标节点列表（告诉图 generate_joke 是合法目标）
)

# ⑤ generate_joke → best_joke：所有 Worker 完成后汇聚到 best_joke
graph.add_edge("generate_joke", "best_joke")

# ⑥ best_joke → END
graph.add_edge("best_joke", END)

# ⑦ 编译
app = graph.compile()

add_conditional_edges 的第三个参数

在使用 Send API 时，add_conditional_edges 需要第三个参数：一个字符串列表，声明路由函数可能路由到哪些节点。这是为了让 LangGraph 在编译时知道图的结构（即使实际分发数量是运行时决定的）。在这里写 ["generate_joke"] 表示"所有 Send 的目标都是 generate_joke 节点"。

图结构的完整拓扑

Map-Reduce 图的节点与边关系

__start__

↓ add_edge

generate_topics

生成 subjects 列表

add_conditional_edges(continue_to_jokes) → Send 列表

Send(mammals)

↓

generate_joke

JokeState {subject: mammals}

↓

Send(reptiles)

↓

generate_joke

JokeState {subject: reptiles}

↓

Send(birds)

↓

generate_joke

JokeState {subject: birds}

↓

Reducer operator.add 合并所有 jokes

best_joke

Reduce：评选最佳

↓ add_edge

__end__

generate_joke → best_joke 的等待机制

图中有一条普通边 add_edge("generate_joke", "best_joke")。当多个 generate_joke Worker 实例并行运行时，LangGraph 会等待所有 Worker 都完成后，才触发 best_joke 节点。这是 LangGraph 内置的"扇入（Fan-in）"同步机制，不需要开发者手动处理等待逻辑。

扇出（Fan-out）与扇入（Fan-in）

扇出：Send API 将一个节点的工作动态分发给多个并行 Worker——这是 Map 阶段。扇入：LangGraph 自动等待所有 Worker 完成，Reducer 合并结果，然后触发下游节点——这是 Reduce 阶段。两者配合，构成完整的 Map-Reduce 生命周期。

7 执行流程追踪：从话题到最佳笑话

运行图

# 以流式方式运行，实时观察每个节点的输出
for s in app.stream({"topic": "animals"}):
    print(s)

实际输出（来自 Notebook）

# 第一步：generate_topics 节点执行，生成 3 个子话题
{'generate_topics': {'subjects': ['mammals', 'reptiles', 'birds']}}

# 第二步：3 个 generate_joke Worker 并行执行（顺序可能不同）
{'generate_joke': {'jokes': ["Why don't mammals ever get lost? Because they always follow their 'instincts'!"]}}
{'generate_joke': {'jokes': ["Why don't alligators like fast food? Because they can't catch it!"]}}
{'generate_joke': {'jokes': ["Why do birds fly south for the winter? Because it's too far to walk!"]}}

# 第三步：best_joke 节点执行，选出最佳
{'best_joke': {'best_selected_joke': "Why don't alligators like fast food? Because they can't catch it!"}}

逐步拆解执行过程

1

invoke 调用，OverallState 初始化

app.stream({"topic": "animals"}) 触发图执行，OverallState 初始化：topic="animals"，其他字段为空。图从 START → generate_topics 开始。
2

generate_topics 执行

节点调用 LLM，生成 3 个子话题 ["mammals", "reptiles", "birds"]，写入 OverallState.subjects。
3

continue_to_jokes 路由函数被调用

条件边触发 continue_to_jokes，它读取 state["subjects"]（3 个元素），用列表推导式创建 3 个 Send 对象并返回。
4

3 个 Worker 并行启动

LangGraph 接收到 3 个 Send 对象，同时启动 3 个 generate_joke 节点实例，各自带着不同的 JokeState（subject 分别为 mammals/reptiles/birds）。
5

Reducer 收集结果

3 个 Worker 各自返回 {"jokes": ["笑话X"]}。Reducer operator.add 将 3 个单元素列表拼接为包含 3 条笑话的列表，写入 OverallState.jokes。
6

best_joke 节点执行（Reduce 阶段）

所有 Worker 完成后，best_joke 节点读取完整的 jokes 列表，调用 LLM 评选，返回最佳笑话的索引，写入 best_selected_joke。
7

到达 END，返回最终 State

图执行完毕，最终 OverallState 中包含了 topic、subjects、所有 jokes 以及 best_selected_joke 的完整数据。

State 最终值

topic: "animals"

subjects: ["mammals", "reptiles", "birds"]

jokes: [

"Why don't mammals ever get lost? Because they always follow their 'instincts'!",

"Why don't alligators like fast food? Because they can't catch it!",

"Why do birds fly south for the winter? Because it's too far to walk!"

]

best_selected_joke: "Why don't alligators like fast food? Because they can't catch it!"

最终 OverallState（图执行完毕）

8 应用场景与设计模式总结

Map-Reduce 适合哪些场景？

场景	Map 阶段（扇出）	Reduce 阶段（聚合）
研究助手	LLM 生成多个搜索查询，并行搜索	合并所有搜索结果，综合生成报告
文档分析	将长文档分段，并行提取每段的关键信息	汇总所有段落的提取结果，生成摘要
多角度评估	从不同维度（安全性、性能、可读性）并行评估代码	合并各维度评分，给出综合建议
内容生成	为多个目标受众（技术/产品/市场）并行生成内容	评选最适合的版本
数据验证	对数据集的每条记录并行调用 LLM 验证	统计验证结果，生成质量报告

Send API 设计要点总结

A

路由函数返回 Send 列表，不是字符串

普通条件边返回节点名字符串。使用 Send 时，路由函数返回 List[Send]，每个 Send 对象代表一个 Worker 实例。
B

Worker 节点可以有私有 State

Send 传入的字典不需要与 OverallState 一致，Worker 节点可以定义独立的 JokeState。LangGraph 会用 Send 中的字典填充 Worker 的输入 State。
C

聚合字段必须使用 Reducer

接收 Worker 输出的 OverallState 字段必须使用 Annotated[list, operator.add]（或其他 Reducer），否则并发写入会导致结果丢失。
D

Worker 写入时包裹成列表

Worker 节点返回单元素列表 {"jokes": [joke]} 而非字符串 {"jokes": joke}，以配合 operator.add 的列表拼接语义。
E

add_conditional_edges 需要声明合法目标

使用 Send 时，add_conditional_edges 的第三个参数需要传入一个列表，声明 Send 可能路由到的节点名称，让 LangGraph 在编译时验证图结构。

Map-Reduce vs 其他并行方式

方式	并行数量	Worker 输入	适用场景
静态并行边（Module 4-1）	编译时固定	共享完整 OverallState	固定数量的并行子任务
Send API（本节）	运行时动态决定	每个 Worker 独享私有子 State	对动态列表中每个元素并行处理
子图（Module 4-2）	固定（子图实例数）	共享 OverallState 的子集	复杂子工作流的封装与复用

本节核心结论

Send API 是 LangGraph 实现动态 Map-Reduce 模式的关键机制。它允许一个路由函数在运行时根据数据动态创建任意数量的并行 Worker，每个 Worker 拥有独立的私有 State，最终通过 Reducer 将所有结果合并回 OverallState。这种设计让 LangGraph 能够优雅地处理"对 LLM 生成的动态列表进行并行处理"这类在 AI 应用中极为常见的需求。

与 Module 4-4 的联系

下一节（4-4）将构建一个完整的多 Agent 研究助手，它会综合运用本节的 Map-Reduce 模式：先让 LLM 将研究问题分解为多个子查询，并行地让多个 Worker Agent 搜集信息，最后聚合成完整的研究报告。Map-Reduce 是这个研究助手的核心并行引擎。

Send API 动态并行 Map-Reduce Reducer Annotated operator.add 扇出 Fan-out 扇入 Fan-in 私有 Worker State 条件边路由

1 什么是 Map-Reduce？两个阶段的本质

Map-Reduce 两阶段示意

与 Module 4-1 普通并行的本质区别

2 静态并行边的局限：为什么不够用

一个说明问题的具体例子

3 Send API：动态派发工作的核心机制

Send 是什么？

Send 如何被触发？——条件边返回 Send 列表

Worker 节点可以使用私有 State

4 State 设计：Annotated Reducer 如何聚合结果

多个 Worker 同时写入同一字段，会冲突吗？

State 字段的生命周期

为什么 Worker 写入 jokes 时要包裹成列表？

5 完整示例：笑话生成与评选系统

第一步：准备 LLM 和提示词

第二步：定义所有 State 和 Pydantic 模型

第三步：定义三个节点函数

第四步：定义路由函数（Send 的触发点）

6 逐行解析：图的构建与条件边的特殊用法

完整的图构建代码

图结构的完整拓扑

Map-Reduce 图的节点与边关系

generate_joke → best_joke 的等待机制

7 执行流程追踪：从话题到最佳笑话

运行图

实际输出（来自 Notebook）

逐步拆解执行过程

invoke 调用，OverallState 初始化

generate_topics 执行

continue_to_jokes 路由函数被调用

3 个 Worker 并行启动

Reducer 收集结果

best_joke 节点执行（Reduce 阶段）

到达 END，返回最终 State

State 最终值

8 应用场景与设计模式总结

Map-Reduce 适合哪些场景？

Send API 设计要点总结

路由函数返回 Send 列表，不是字符串

Worker 节点可以有私有 State

聚合字段必须使用 Reducer

Worker 写入时包裹成列表

add_conditional_edges 需要声明合法目标

Map-Reduce vs 其他并行方式

与 Module 4-4 的联系