Map-Reduce and Send API
Dynamic Parallel Fanout · Result Aggregation

1What is Map-Reduce? The nature of the two stages

Map-Reduce is a classic parallel computing model widely used in the field of big data (such as Hadoop, Spark). LangGraph introduces this idea into the LLM workflow forDecompose a task into multiple subtasks for parallel processing, and then aggregate the results。

It consists of two stages:

• Map (mapping/fanout) phase: Split a large task into several small, independent subtasks, and distribute them to multiple Worker nodes concurrently for processing.
• Reduce (reduce/aggregate) phase: Collect the output of all Worker nodes and combine them into a final result.

Essential differences from Module 4-1 ordinary parallelism

Module 4-1 Learned and usedadd_edgeExecute multiple nodes in parallel, that isstatic parallelism——Which parallel branches are fixed at compile time. Map-Reduce isdynamic parallelism——The number of branches is determined based on data at runtime, and can be 1, 3 or 100 Workers, fully flexible.

2Limitations of static parallel edges: why they are not enough

In Module 4-1 (Parallelization), you learned to implement parallelism using ordinary edges:

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

The problem with this approach is:The number of parallel branches must be determined during graph compilation, they are "hard-coded" into the structure of the graph.

A concrete example to illustrate the problem

Suppose you ask LLM to generate subtopics based on the topic "animals", and LLM returns["mammals", "reptiles", "birds"]. Next time you enter the topic "programming", LLM may return["Python", "Rust", "Go", "TypeScript", "Java"](5 pieces).

youImpossibleUse static parallel edges to handle this situation when writing code - because you don't know how many you have to writeadd_edge. The only solution is to dynamically generate workers at runtime, which is what the Send API does.

3Send API: the core mechanism for dynamically dispatching work

What is Send?

Sendis a special object provided by LangGraph for use at runtimeDynamically dispatch a task to a node, and also carries status data exclusive to the task.

from langgraph.types import Send

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

Key features:

• node_name: Must be a registered node name (string) in the graph.
• state: The state dictionary passed to this Worker node.This dictionary does not need to be aligned with OverallState——Worker nodes can have their own private State structure (JokeState).
• multipleSendObjects will be executed in parallel by LangGraph, just like starting multiple independent Worker threads at the same time.

How is Send triggered? ——The conditional edge returns the Send list

SendThe object is not called directly, butReturned from the routing function of a conditional edge. When the routing function returns aSendwhen a list of objects, LangGraph willSendDistributed in parallel:

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

hypothesisstate["subjects"]Yes["mammals", "reptiles", "birds"], the above list comprehension will generate:

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

Worker nodes can use private State

This is another powerful thing about the Send API: Worker nodes (generate_joke) does not need to be usedOverallState, which can define its own private state structure:

# 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 passed in when dispatching{"subject": s}that isJokeStateThe content of LangGraph will be used to populate the input State of the Worker node. The output of the Worker ({"jokes": [...]}) will be written backOverallStateThe corresponding fields are merged by reducer.

4State Design: How Annotated reducer Aggregates Results

If multiple workers write to the same field at the same time, will there be a conflict?

This is a very important question. In Map-Reduce, each of the three Worker nodes generates a joke and all need to be written.OverallStateofjokesfield. If you use the default "overwrite" semantics, only the results of the last written Worker will be retained!

The solution is to useAnnotated+ reducer function, changes the write semantics of the field from "overwrite" to "append merge":

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 节点选出的最佳笑话

State field life cycle

Why does Worker wrap jokes into lists when writing them?

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

Each Worker returns a list containing only one element. reduceroperator.addConcatenate these single-element lists to get a complete list of all jokes. This is a very subtle design - keep the output format of the Worker node consistent and let the reducer handle the merging automatically.

5Complete example: Joke generation and selection system

The example in the notebook is a complete Map-Reduce application: given a large topic, multiple subtopics are automatically generated, and LLM generates a joke for each subtopic in parallel, and finally the LLM selects the best one from all the jokes.

Step One: Prepare LLM and Prompt Words

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)

Step 2: Define all State and Pydantic models

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

Step 3: Define three node functions

### 节点 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]}

Step 4: Define the routing function (Send trigger point)

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

6Line-by-line analysis: graph construction and special usage of conditional edges

Complete graph construction code

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()

The complete topology of the graph structure

generate_joke → waiting mechanism for best_joke

There is an ordinary edge in the graphadd_edge("generate_joke", "best_joke"). when multiplegenerate_jokeWhen Worker instances run in parallel, LangGraph willWait for all Workers to completeonly afterbest_jokenode. This is LangGraph's built-in "Fan-in" synchronization mechanism, which does not require developers to manually handle waiting logic.

7Execution Process Tracking: From Topic to Best Joke

Run chart

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

Actual output (from 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!"}}

Breaking down the execution process step by step

• 1

  invoke call, OverallState initialization

  app.stream({"topic": "animals"})Trigger graph execution, OverallState initialization:topic="animals", other fields are empty. Picture from START →generate_topicsStart.

• 2

  generate_topics execution

  The node calls LLM and generates 3 subtopics["mammals", "reptiles", "birds"], writeOverallState.subjects。

• 3

  continue_to_jokes routing function is called

  Conditional edge triggercontinue_to_jokes, which readsstate["subjects"](3 elements), using list comprehensions to create 3Sendobject and return.

• 4

  3 workers started in parallel

  LangGraph received 3SendObject, start 3 at the same timegenerate_jokeNode instances, each with differentJokeState(subjects are mammals/reptiles/birds respectively).

• 5

  reducer collects results

  Each of the 3 Workers returns{"jokes": ["笑话X"]}。Reducer operator.addTo concatenate 3 single-element lists into a list containing 3 jokes, writeOverallState.jokes。

• 6

  best_joke node execution (Reduce phase)

  After all Workers are completed,best_jokeNode reads completejokesList, call LLM selection, return the index of the best joke, writebest_selected_joke。

• 7

  Reach END and return to the final State

  The graph is executed, and finallyOverallStateContains complete data for topic, subjects, all jokes and best_selected_joke.

State final value

8Summary of application scenarios and design patterns

What scenarios is Map-Reduce suitable for?

Summary of Send API design points

• A

  Routing function returns Send list, not string

  Ordinary conditional edges return node name strings. When using Send, the routing function returnsList[Send], each Send object represents a Worker instance.

• B

  Worker nodes can have private State

  The dictionary passed in by Send does not need to be consistent with OverallState, and the Worker node can define an independent JokeState. LangGraph will populate the Worker's input State with the dictionary from Send.

• C

  Aggregation fields must use reducer

  The OverallState field that receives Worker output must useAnnotated[list, operator.add](or other reducer), otherwise concurrent writes will cause results to be lost.

• D

  Worker wraps it into a list when writing

  Worker node returns a single element list{"jokes": [joke]}instead of string{"jokes": joke}, to matchoperator.addlist concatenation semantics.

• E

  add_conditional_edges needs to declare legal targets

  When using Send,add_conditional_edgesThe third parameter needs to be passed in a list declaring the node names that Send may be routed to, allowing LangGraph to verify the graph structure at compile time.

Map-Reduce vs other parallel methods

Contact with Module 4-4

The next section (4-4) will build a complete multi-agent research assistant, which will comprehensively use the Map-Reduce model in this section: first let LLM decompose the research problem into multiple subqueries, let multiple Worker Agents collect information in parallel, and finally aggregate it into a complete research report. Map-Reduce is the core parallel engine of this research assistant.