LangGraph 并行执行
Fan-out · Fan-in · Reducer

1 什么是并行执行？为什么需要它？

在 LangGraph 的前三个模块里，我们学习的图都是线性的：节点 A 执行完，节点 B 开始，节点 B 执行完，节点 C 开始。这种"单车道"执行方式简单可预测，但有一个天然瓶颈：每个步骤都必须等待上一个步骤完成。

现实场景中，很多任务是相互独立的，可以同时进行。比如：

• 同时从 Wikipedia 和 Web 搜索引擎检索资料，然后让 LLM 综合两份资料回答问题
• 同时对一篇文章做情感分析和关键词提取，然后合并结果
• 同时调用多个外部 API，把结果聚合

如果这些独立任务串行执行，总耗时 = 任务 1 时间 + 任务 2 时间 + …；而如果并行执行，总耗时 ≈ 最慢那个任务的时间。当每个子任务都需要等待 LLM 或网络 IO 时，并行化带来的提速效果非常显著。

Fan-out 和 Fan-in 是什么意思？

这两个术语来自电路设计，在 LangGraph 中含义如下：

2 线性图 vs. 并行图：从顺序执行到 Fan-out / Fan-in

先看线性图：逐个覆盖 state

notebook 首先用一个线性图做基准对比。每个节点把自己的值覆盖掉 state 字段（注意：这里没有 Reducer）：

from typing import Any, List
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    # 注意：没有 Reducer，默认行为是"覆盖"
    state: List[str]

class ReturnNodeValue:
    def __init__(self, node_secret: str):
        self._value = node_secret

    def __call__(self, state: State) -> Any:
        print(f"Adding {self._value} to {state['state']}")
        return {"state": [self._value]}  # 直接覆盖，不追加

# 构建线性图：a → b → c → d
builder = StateGraph(State)
builder.add_node("a", ReturnNodeValue("I'm A"))
builder.add_node("b", ReturnNodeValue("I'm B"))
builder.add_node("c", ReturnNodeValue("I'm C"))
builder.add_node("d", ReturnNodeValue("I'm D"))

builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", "c")
builder.add_edge("c", "d")
builder.add_edge("d", END)
graph = builder.compile()

运行结果如下。因为每个节点都覆盖上一个节点的值，最终 state 里只剩 D 的值：

改成并行图：Fan-out 到 b 和 c，Fan-in 到 d

现在把 b 和 c 改为并行执行。关键改变只有边的连接方式——从 a 同时连向 b 和 c，然后 b 和 c 都连向 d：

# Fan-out：a → b 且 a → c（两条边同时从 a 出发）
builder.add_edge("a", "b")
builder.add_edge("a", "c")

# Fan-in：b → d 且 c → d（两条边汇聚到 d）
builder.add_edge("b", "d")
builder.add_edge("c", "d")

3 并行冲突：同一字段被多个节点同时写入

并行图如果直接用没有 Reducer 的 State，会报错。这是 LangGraph 的一个重要安全机制，理解它非常关键。

为什么会冲突？

在并行步骤里，b 和 c 在同一个"步骤"（step）内同时执行，它们各自都要向 state 字段写入一个值。没有 Reducer 时，LangGraph 不知道该保留哪个——两个写入互相冲突，所以它直接抛出异常，而不是悄悄丢掉其中一个。

from langgraph.errors import InvalidUpdateError

try:
    graph.invoke({"state": []})
except InvalidUpdateError as e:
    print(f"An error occurred: {e}")

错误的根本原因：LangGraph 的步骤（Step）概念

LangGraph 把图的执行划分为一个个"步骤"。在并行场景下，同一步骤里可能有多个节点并发运行。当这些节点都要更新同一个 State 字段时，LangGraph 需要一个明确的规则来决定如何合并这些更新——这个规则就是 Reducer。

4 Reducer：并行输出的安全合并机制

Reducer 是附加在 State 字段上的一个合并函数。当同一步骤内多个节点都对同一字段写入值时，LangGraph 会用 Reducer 函数把所有写入值合并成一个最终值。

用 operator.add 作为 Reducer

Python 标准库的 operator.add 应用在列表上等价于列表拼接（list1 + list2）。把它作为 Reducer，所有并行节点的输出列表会被依次拼接：

import operator
from typing import Annotated

class State(TypedDict):
    # Annotated 的第二个参数是 Reducer 函数
    # operator.add 作用于列表时 = 列表拼接（追加）
    state: Annotated[list, operator.add]

# 其余代码（节点定义、边连接）与之前完全相同
builder = StateGraph(State)
builder.add_node("a", ReturnNodeValue("I'm A"))
builder.add_node("b", ReturnNodeValue("I'm B"))
builder.add_node("c", ReturnNodeValue("I'm C"))
builder.add_node("d", ReturnNodeValue("I'm D"))

# Fan-out + Fan-in
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", "d")
builder.add_edge("c", "d")
builder.add_edge("d", END)
graph = builder.compile()

现在运行成功：

Reducer 的工作原理图解

5 不等长支路：LangGraph 如何等待所有并行路径完成

实际场景中，并行的多条支路往往长度不同——比如一条支路只有一个节点，另一条支路有两个节点串联。LangGraph 如何处理这种情况？

构建不等长并行图

在这个例子中，从 a Fan-out 出两条支路：支路一是 b → b2（两个节点串联），支路二是 c（只有一个节点）。两条支路都 Fan-in 到 d：

builder = StateGraph(State)

builder.add_node("a",  ReturnNodeValue("I'm A"))
builder.add_node("b",  ReturnNodeValue("I'm B"))
builder.add_node("b2", ReturnNodeValue("I'm B2"))
builder.add_node("c",  ReturnNodeValue("I'm C"))
builder.add_node("d",  ReturnNodeValue("I'm D"))

builder.add_edge(START, "a")
builder.add_edge("a", "b")    # 支路一：a → b → b2
builder.add_edge("a", "c")    # 支路二：a → c
builder.add_edge("b", "b2")

# 列表语法：等 b2 和 c 都完成后，才触发 d
builder.add_edge(["b2", "c"], "d")
builder.add_edge("d", END)
graph = builder.compile()

执行结果：

关键观察：LangGraph 的"步骤"调度机制

从输出中可以看出：

• 步骤 1：a 执行
• 步骤 2：b 和 c 同时执行（因为它们都是 a 的直接后继）
• 步骤 3：b2 执行（b 完成后触发），同时 c 已完成等待
• 步骤 4：d 执行——只有当 b2 和 c 都完成后才触发

6 自定义 Reducer：控制并行输出的合并顺序

用 operator.add 合并并行输出时，各个节点写入的顺序由 LangGraph 内部调度决定，不可预测。有时候我们希望合并后的结果有确定的顺序，这就需要自定义 Reducer。

问题描述

上一节的输出 ["I'm A", "I'm B", "I'm C", "I'm B2", "I'm D"] 中，C 在 B2 前面，但我们可能希望结果按字母顺序排序。

编写自定义 Reducer 函数

def sorting_reducer(left, right):
    """合并两个列表并按字母顺序排序"""
    # 确保 left 和 right 都是列表
    if not isinstance(left, list):
        left = [left]
    if not isinstance(right, list):
        right = [right]
    # 合并后排序
    return sorted(left + right, reverse=False)

class State(TypedDict):
    # 使用自定义 Reducer
    state: Annotated[list, sorting_reducer]

其他代码（节点、边的定义）与上一节相同，只需把 State 定义里的 Reducer 替换即可：

注意最终结果 ["I'm A", "I'm B", "I'm B2", "I'm C", "I'm D"] 按字母顺序排列了——即使在执行过程中 C 比 B 先被打印，排序 Reducer 确保了最终结果的有序性。

Reducer 函数的签名约定

7 真实案例：并行搜索 Wikipedia + Web 后用 LLM 汇总回答

前面的例子都用简单的字符串来演示并行机制。现在用一个真实的 AI 场景来展示并行执行的实际价值：同时从 Wikipedia 和 Web 搜索引擎检索资料，再由 LLM 综合两份资料回答用户问题。

State 定义

import operator
from typing import Annotated
from typing_extensions import TypedDict

class State(TypedDict):
    question: str              # 用户问题（不需要 Reducer，只写一次）
    answer:   str              # LLM 的最终回答
    context:  Annotated[list, operator.add]  # 检索到的资料（需要 Reducer 追加）

context 字段会被 search_web 和 search_wikipedia 两个并行节点同时写入，所以必须加 operator.add Reducer。question 和 answer 只会被一个节点写入，不需要 Reducer。

三个节点的完整实现

from langchain_core.messages import HumanMessage, SystemMessage
from langchain_community.document_loaders import WikipediaLoader
from langchain_tavily import TavilySearch
from langchain_deepseek import ChatDeepSeek

llm = ChatDeepSeek(model="deepseek-v4-pro", temperature=0)

# ──── 节点 1：Web 搜索 ────
def search_web(state):
    """从 Web 搜索引擎检索资料"""
    tavily_search = TavilySearch(max_results=3)
    data = tavily_search.invoke({"query": state['question']})
    search_docs = data.get("results", data)

    # 格式化为带 URL 的文档字符串
    formatted = "\n\n---\n\n".join([
        f'<Document href="{doc["url"]}">\n{doc["content"]}\n</Document>'
        for doc in search_docs
    ])
    return {"context": [formatted]}  # 返回列表，Reducer 会追加

# ──── 节点 2：Wikipedia 搜索 ────
def search_wikipedia(state):
    """从 Wikipedia 检索资料"""
    search_docs = WikipediaLoader(
        query=state['question'], load_max_docs=2
    ).load()

    # 格式化为带来源的文档字符串
    formatted = "\n\n---\n\n".join([
        f'<Document source="{doc.metadata["source"]}">\n{doc.page_content}\n</Document>'
        for doc in search_docs
    ])
    return {"context": [formatted]}  # 返回列表，Reducer 会追加

# ──── 节点 3：LLM 综合回答 ────
def generate_answer(state):
    """基于检索到的上下文，让 LLM 回答问题"""
    context  = state["context"]   # 已合并的两份资料
    question = state["question"]

    answer_template = "Answer the question {question} using this context: {context}"
    instructions = answer_template.format(question=question, context=context)

    answer = llm.invoke([
        SystemMessage(content=instructions),
        HumanMessage(content="Answer the question.")
    ])
    return {"answer": answer}

构建并行图

builder = StateGraph(State)

builder.add_node("search_web",       search_web)
builder.add_node("search_wikipedia", search_wikipedia)
builder.add_node("generate_answer",  generate_answer)

# 从 START 同时 Fan-out 到两个搜索节点（并行）
builder.add_edge(START, "search_wikipedia")
builder.add_edge(START, "search_web")

# 两个搜索节点都完成后，Fan-in 到 generate_answer
builder.add_edge("search_wikipedia", "generate_answer")
builder.add_edge("search_web",       "generate_answer")
builder.add_edge("generate_answer",  END)

graph = builder.compile()

运行示例

result = graph.invoke({"question": "How were Nvidia's Q2 2025 earnings"})
print(result['answer'].content)

这个案例体现的核心价值

• 1

  并行加速

  Web 搜索和 Wikipedia 搜索同时进行，总耗时约等于两者中较慢的那个，而不是两者之和。

• 2

  自动合并

  两个搜索节点的结果通过 operator.add Reducer 自动拼接进 context 列表，generate_answer 节点收到的是完整的两份资料，无需手动合并。

• 3

  信息质量更高

  LLM 基于 Web 的时效性内容 + Wikipedia 的权威知识库两个来源综合回答，比单一来源更全面准确。

8 LangGraph Studio 部署与 API 调用

构建好的并行图除了在 notebook 里直接 invoke，还可以部署到 LangGraph API Server，通过 HTTP 接口或 SDK 异步调用。

本地启动 LangGraph Dev Server

在 module-4/studio/ 目录下执行：

# 在终端里运行（不是 Python 代码）
langgraph dev

在浏览器打开 Studio UI 地址，可以可视化地看到图结构、实时执行状态、每个节点的输入/输出，这在调试复杂并行图时非常有用。

通过 SDK 异步流式调用

from langgraph_sdk import get_client

client = get_client(url="http://127.0.0.1:2024")

# 创建新线程（每个线程有独立的状态）
thread = await client.threads.create()

input_question = {"question": "How were Nvidia Q2 2025 earnings?"}

# 流式订阅执行过程，stream_mode="values" 表示每个步骤后推送完整 State
async for event in client.runs.stream(
    thread["thread_id"],
    assistant_id="parallelization",  # 对应 studio 里的图名称
    input=input_question,
    stream_mode="values"
):
    # 检查 answer 字段是否已经出现
    if event.data is not None:
        answer = event.data.get('answer', None)
        if answer:
            print(answer['content'])

为什么要用 API Server 而不是直接 invoke？

本节知识全景总结