摘要：并行化（Parallelization）让智能体中的多个独立子任务同时执行，而不是一个接一个排队，从而显著缩短总耗时。本文说明并行化的动机、典型应用场景，并重点介绍在 LangGraph 中如何用 Send API + 状态图 实现「多节点扇出并行、reducer 汇聚、再合成」的流程。配套示例为多源研究：对同一主题拆成多个子课题，每个子课题在独立节点内完成「生成检索词 → 模拟检索 → 总结」多步，是体现图级并行的典型案例。

关键词：并行化；Parallelization；并发执行；LangGraph；Send API；扇出；reducer；多节点并行；Agentic Design Patterns

源代码链接：Langgragh 3. Parallelization 源代码。

---

1 为什么需要「并行化」？

在前两章中，我们接触了顺序链（Prompt Chaining）和路由（Routing）：前者是固定步骤依次执行，后者是根据条件选择不同路径。但很多复杂任务里，存在多件互不依赖的事可以同时做——若仍按顺序执行，就会白白浪费等待时间。

并行化（Parallelization） 就是在智能体中让多个组件（如多次 LLM 调用、多路工具调用甚至多个子智能体）并发执行。不依赖前一步输出的任务可以同时启动，等它们都完成后再进入依赖它们的步骤（如汇总、综合），从而在保证正确性的前提下缩短整体耗时。

💡 理解要点：并行化 = 能同时做的就同时做；特别适合涉及外部 I/O（API、数据库、多源检索）的场景，因为等待时间可以重叠。

---

2 并行化在解决什么问题？

想象一个「研究某主题并写报告」的智能体：

• 顺序做法：先查资料 A → 总结 A → 再查资料 B → 总结 B → 最后综合。总时间 ≈ 查 A + 总结 A + 查 B + 总结 B + 综合。
• 并行做法：同时查资料 A 和 B → 两者都完成后，同时总结 A 和 B（若彼此独立）→ 再综合。总时间 ≈ max(查 A, 查 B) + max(总结 A, 总结 B) + 综合。

当「查资料」「总结」涉及网络或外部服务时，并行能明显减少总延迟。同理：多 API 调用、多数据源拉取、多维度分析（情感、关键词、分类同时做）等，都可以用并行化优化。

🔍 实际例子：就像同时问三个人「北京天气」「上海天气」「广州天气」——三个人同时查、同时回，你只等最慢的那一个，而不是等三个人依次排队回答。

---

3 典型应用场景

场景	说明	并行收益
信息搜集与调研	同时从新闻、股票、社交、数据库等多源拉取信息	快速得到多维度视图
数据处理与分析	对同一批反馈同时做情感分析、关键词抽取、分类、紧急度识别	多维度分析一次完成
多 API / 多工具	旅行规划时同时查机票、酒店、当地活动、餐厅	更快拼出完整方案
多部分内容生成	营销邮件同时生成标题、正文、配图建议、按钮文案	组装更快
校验与验证	同时校验邮箱格式、手机号、地址、敏感词	更快给出校验结果
多模态处理	同一输入中的文本与图像同时做分析	更快融合多模态结论

💡 理解要点：只要子任务之间没有强依赖（不依赖对方本轮的输出），就适合用并行化；有依赖的步骤仍需在汇聚点之后顺序或再分支处理。

---

4 LangGraph 中的并行化：Send API 与图级扇出

LangGraph 用状态图（State Graph） 描述流程。实现并行化时，常见做法有两种：

1. 单节点内并行：一个节点内部使用 LCEL 的 RunnableParallel（或异步并发）同时执行多条链，把多条结果写回状态，再交给下一节点。这样「并行」发生在一个节点内，图结构仍是线性。
2. 图级扇出（Send API）：由条件边返回多个 Send(node_name, state)，框架会同时启动多个目标节点，每个节点收到各自的状态；各节点返回的状态更新通过 reducer（如 operator.add 对列表做拼接）合并，再进入后续步骤或结束。这是「多节点真正并行」的典型写法。

图级扇出具体是怎么工作的？ 可以拆成四步理解：

• 扇出（fan-out）：从图上的「一个点」分出「多条边」。比如 dispatcher 跑完后，不是沿着一条边走到下一个节点，而是由条件边函数返回一个 Send 列表，例如 [Send("research", {query, subtopic: "A"}), Send("research", {query, subtopic: "B"}), Send("research", {query, subtopic: "C"})]。每一份 Send("节点名", 状态) 相当于说：「请用这份状态去跑一次这个节点」。
• 并行执行：LangGraph 看到这一列表后，会同时启动多份目标节点（这里是 3 份 research），每份拿到的状态不同（各自一个 subtopic）。所以是「图上多条分支真正并行」，而不是在一个节点里用 RunnableParallel。
• 各写各的结果：每个 research 节点跑完后，返回例如 {"research_results": [我这一条结果]}。多条分支会往状态的同一个键（research_results）里写，若没有合并规则，后写的会覆盖先写的。
• reducer 汇聚：给该键加上 reducer（如 Annotated[list, operator.add]）后，框架不会覆盖，而是按规则合并（这里就是把多个列表拼成一个）。于是最终状态里 research_results 就是「所有分支结果拼成的大列表」。

🔍 类比：就像主管把同一任务拆成三份，同时派给三个同事（三个 research 节点），每人拿到的子任务不同；三人各自交回一份报告，最后用 reducer 把三份报告装订成一本（一个列表），再交给下一步或结束。

下面这个示例采用图级扇出：用户输入主题后，先经 dispatcher 确定若干子课题，再通过 Send 同时启动多个 research 节点；每个 research 节点内部是多步流程 (生成检索词 → 模拟检索 → 总结)；多分支结果通过 reducer 汇聚，合成报告可在图外一次 LLM 调用完成。

---

5 配套代码结构概览

实际例子：本示例中，主题「sustainable technology」被拆成三个子课题（renewable energy、electric vehicles、carbon capture），三个 research 节点并行执行，每个节点 2 次 LLM + 1 次模拟 I/O；结果汇聚后，再一次性合成报告。

代码建议与 README 对照阅读。

5.1 状态定义与 reducer

图的状态用 TypedDict 定义；多分支写入同一字段时需用 reducer 合并，此处用 operator.add 对列表做拼接：

# 摘自 demo_codes/parallel_graph.py
from typing import Annotated
import operator

class ParallelState(TypedDict):
    query: str
    subtopics: list
    subtopic: str
    research_results: Annotated[list, operator.add]  # 多分支结果由 reducer 合并

5.2 Workflow

整体流程：START → dispatcher → [Send × N] → research（×N 并行）→ END。汇聚后的 research_results 在图外交给合成链生成最终报告：

workflow = StateGraph(ParallelState)

workflow.add_node("dispatcher", node_dispatcher)
workflow.add_node("research", node_research)

workflow.add_edge(START, "dispatcher")
workflow.add_conditional_edges("dispatcher", fan_out_to_research)
workflow.add_edge("research", END)

我们将图可视化出来：

（图里看不到三条并行线，是因为可视化器对「条件边返回多个 Send」的展示不完整）

5.3 Dispatcher

用户输入主题后，先经 dispatcher 确定若干子课题。

在当前代码中，我们给定了三个子主题。当然，我们也可以通过大模型来确定，或者做一个前端让用户输入来进行填充：

def node_dispatcher(state: ParallelState) -> dict:
    """
    根据用户主题确定要并行研究的子课题列表。
    此处写死三路子课题以突出并行结构；实际可由 LLM 根据 query 动态生成。
    """
    return {
        "subtopics": [
            "renewable energy",
            "electric vehicles",
            "carbon capture",
        ]
    }

5.4 扇出：条件边返回多个 Send

从 dispatcher 出来后，条件边根据 subtopics 返回多个 Send("research", {...})，每个分支携带 query 与一个 subtopic：

# 摘自 demo_codes/parallel_graph.py
from langgraph.types import Send

def fan_out_to_research(state: ParallelState) -> list[Send]:
    return [
        Send("research", {"query": state["query"], "subtopic": st})
        for st in state.get("subtopics") or []
    ]

# 图中：dispatcher 后接条件边
workflow.add_conditional_edges("dispatcher", fan_out_to_research)

💡 理解要点：Send 的第二个参数是该分支传给目标节点的状态；多个 Send 指向同一节点时，该节点会被并发执行多次，每次收到不同状态。

5.3 research 节点

每个 research 节点内：先根据 query + subtopic 生成检索词（1 次 LLM），再模拟检索（可替换为真实 API/DB），最后对「文档」做摘要（1 次 LLM）。两步 LLM + 一步 I/O：

# 摘自 demo_codes/parallel_graph.py（片段）
def node_research(state: ParallelState) -> dict:
    query, subtopic = state.get("query"), state.get("subtopic")
    search_query = chain_generate_search_query.invoke({"query": query, "subtopic": subtopic})
    document = simulated_search(subtopic, search_query)
    summary = chain_summarize_doc.invoke({"document": document})
    return {"research_results": [{"subtopic": subtopic, "search_query": search_query, "summary": summary}]}

各分支返回的 research_results 为单元素列表，经 reducer 拼接成完整列表。

5.5 完整代码与运行

完整图构建、模拟检索、链定义与运行方式见 demo_codes/parallel_graph.py 与 demo_codes/main.py。运行方式：在 demo_codes 目录下安装依赖、配置 .env 中的 API Key 后执行 python main.py，即可跑通「dispatcher → 多 research 并行 → 汇聚 → 图外合成」流程。

---

6 如何运行示例

1. 进入 demo_codes 目录，创建并激活虚拟环境，安装依赖：

   pip install -r requirements.txt
   

2. 在目录下配置 .env（如 OPENAI_API_KEY 或 DASHSCOPE_API_KEY，可选 BASE_URL、MODEL）。
3. 运行：

   python main.py
   

4. 程序会对示例主题执行并行图，打印各分支的检索词与摘要，以及图外合成后的报告。

---

7 小结与延伸

• 并行化 通过同时执行互不依赖的子任务减少总耗时，尤其适合多 API、多数据源、多步研究的场景。
• 在 LangGraph 中，Send API + 条件边 可实现图级扇出：多节点真正并行、各收各的状态；用 reducer（如 Annotated[list, operator.add]）汇聚多分支结果。
• 本示例刻意设计为「每分支多步（生成检索词 → I/O → 总结）」，以说明何时必须用图与并行，而非单 prompt 出 JSON。

若要将子课题改为由 LLM 动态生成，可在 dispatcher 节点内增加一次 LLM 调用，根据 query 输出 subtopics 列表，再交给 fan_out_to_research；若某分支需依赖其他分支结果，则需拆成多阶段图（先并行 A/B，再根据结果并行 C/D），在汇聚点用状态传递数据。