LangGraph 进阶实战：并行分支与 Annotated 状态聚合

在简单的线性工作流中，一个节点执行完成后通常只会进入下一个节点。但真实业务往往需要把一个任务拆成多个分支同时处理。例如并行查询多个数据源、同时调用多个工具，或让多个 Agent 分别完成不同任务。

使用者痛点：很多开发者在设计并行工作流时会遇到以下困惑：

• 不清楚如何让同一个节点同时触发多个后续节点；其实，
• 并行节点写入同一状态字段时出现冲突报错；
• 对 Annotated 与 reducer 的作用模糊，导致状态聚合不符合预期；
• 不明白 thread_id 与检查点之间的关系，导致持久化失效。

这里通过一个可直接运行的 LangGraph 示例，演示以下能力：

• 使用多条普通边创建并行分支；
• 使用 Annotated 为状态字段绑定 reducer；
• 使用 operator.add 聚合多个节点返回的列表；
• 编译并执行 StateGraph；
• 将图结构导出为 PNG 图片；
• 理解 thread_id 与 checkpointer 的关系。

整个示例不调用大模型。不需要 API Key，适合用来理解 LangGraph 的状态更新机制。

. 最终流程结构

这篇文章建立的图可以表示为：

START -> step1 -> step2 -> step3 -> END
-> step4 -> END

step2 有两条出边，所以执行完 step2 后step3 和 step4 都会被调度执行。 这是一种 fan‑out结构，而不是“二选一”的条件分支。

实际生成的流程图：

. 运行环境

Pain point: 环境依赖不一致常导致代码在本地跑不通。 不过，下面给出已验证可运行的环境配置。

Python 3.10+
langgraph
typing-extensions

建议为项目创建独立虚拟环境：

python -m venv .venv
source .venv/bin/activate # macOS / Linux
# Windows PowerShell:
# .venv\Scripts\Activate.ps1
pip install -U langgraph typing-extensions

. 为什么需要 reducer

Pain point: 当多个并行节点写入同一字段时如果没有明确的合并规则，会抛出“并发更新冲突”。按理说，LangGraph 提供了 Annotated` 来解决这个问题。

import operator
from typing import Annotated,Any
from typing_extensions import TypedDict
class State:
aggregate: Annotated。operator.add]

• : 表示 alertate `aggregate` 是一个列表。
• : 新值写入时使用加法进行合并。

Pain point: 如果省略了 reducer。两个并行节点都尝试覆盖同一字段，LangGraph 将报错。使用 reducer 后每个节点只需返回自己的增量结果，框架负责自动拼接。怎么说呢，

. 编写节点

step1

def step1:
print
return {"aggregate": } # 增量列表

step2

def step2:
print
return {"aggregate": }

step3 & step4

Pain point: 很多人误以为控制台打印顺序代表业务执行顺序。并行任务的日志顺序是非确定性的，只要状态聚合正确即可。如果业务必须严格有序，请改为线性链或添加显式依赖边。

. 注册节点 & 建立图对象

from langgraph.graph import END,START。StateGraph

graph_builder = StateGraph

graphbuilder.addnode.addnode.addnode.add_node

. 使用多条普通边创建并行分支

# 线性部分
graphbuilder.addedge
graphbuilder.addedge

graphbuilder.addedge

Pain point: 如果想实现条件路由而误用了普通边，会导致所有目标都被执行。老实说，此时应改用 add_conditional_edges.

. 编译工作流

graph = graph_builder.compile

The compiled graph.invoke。.stream,还有检查点功能。

. 生成流程图

pngbytes = graph.getgraph.drawmermaidpng
with open as f:
f.write

If PNG generation fails。check network connectivity or configure a local Mermaid renderer.

. 执行图 & thread_id 与检查点的关系

res = graph.invoke(
{"aggregate": },{"configurable": {"thread_id": ""}} # 当前示例未配置 checkpointer → 不持久化
)
print

Pain point: 很多人把 `thread_id` 当成必填项，却忘记在编译阶段提供 `checkpointer`。没有检查点，`thread_id` 只能用于路由上下文。而不会自动保存历史状态，下面给出带内存检查点的完整写法：

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver graph = graph_builder.compile

res = graph.invoke( {"aggregate": },{"configurable": {"thread_id": "demo-run-01"}} )

。

---