LangGraph 并行执行与异步优化：Superstep 机制、Send API 与实战避坑

1. 为什么单靠链式编排迟早会撞墙

如果你用 LangGraph 搭过稍微复杂一点的 Agent，大概率经历过这个阶段：一开始节点串成一条线，跑得挺顺；等到工具调用多起来、检索分支多起来，整张图的耗时就开始线性膨胀。一个节点等一个节点，明明彼此之间毫无依赖，却非要排队执行，最后用户等一个回答要十几秒。

这个问题的本质不是 LangGraph 慢，而是编排模型没选对。LangGraph 的执行引擎本身是基于 Pregel 的 BSP（Bulk Synchronous Parallel）模型，也就是按“超步”（Superstep）推进的。理解这一点，是理解并行执行和异步优化的钥匙。很多人把 LangGraph 当成一个普通的 DAG 调度器来用，结果既没吃到它的并行红利，也没搞明白为什么加了async反而更慢。

这篇内容我打算把 LangGraph 的并行执行和异步优化讲透。核心围绕四件事：Superstep 的推进机制、Send API 做动态扇出、异步节点与同步节点的混用陷阱、以及并行执行外部命令这类真实场景的落地。适合已经写过几个 LangGraph 图、但一遇到性能瓶颈就只会加机器的人。看完你应该能自己判断：哪些节点该并行、哪些必须串行、异步到底加在哪一层。

先说结论，省得你带着疑问往下读：LangGraph 的并行不是靠你手动开线程池实现的，而是靠图结构本身——同一 Superstep 内没有依赖关系的节点会被自动并行调度；而 Send API 解决的是“运行时才知道要扇出多少个分支”的动态并行问题。异步优化则是另一条线，它优化的是单个节点内部的 IO 等待，和并行是两个维度，混为一谈是新手最常见的误区。

2. Superstep 机制：并行执行的底层逻辑

2.1 什么是 Superstep，为什么它决定了并行度

Pregel 模型里，计算是一轮一轮推进的。每一轮叫一个 Superstep，规则很简单：同一 Superstep 内的所有节点并行执行，全部执行完后，把消息（状态更新）汇总，再进入下一个 Superstep。这就像公司开周会，同一批议题可以分给不同小组同时做，但必须等所有小组都交作业了，才能开下一轮会。

放到 LangGraph 里，你定义的那些节点和边，会被编译成一张执行计划。引擎从入口节点开始，凡是当前能触达、且互相没有依赖的节点，全部塞进同一个 Superstep 并行跑。举个最直观的例子：

from langgraph.graph import StateGraph, START, END def node_a(state): return {"a": 1} def node_b(state): return {"b": 2} def node_c(state): return {"c": state["a"] + state["b"]} builder = StateGraph(dict) builder.add_node("a", node_a) builder.add_node("b", node_b) builder.add_node("c", node_c) builder.add_edge(START, "a") builder.add_edge(START, "b") builder.add_edge("a", "c") builder.add_edge("b", "c") builder.add_edge("c", END) graph = builder.compile()

这里a和b都从 START 出发，互相不依赖，所以它们落在同一个 Superstep，会被并行调度。c依赖a和b的结果，必须等这一轮结束才执行。整个图是 2 个 Superstep，而不是 3 个串行步骤。

关键点在于：并行度是由图的拓扑结构决定的，不是由你写没写 async 决定的。哪怕node_a和node_b都是同步函数，只要它们在同一 Superstep 且没有依赖，LangGraph 也会想办法并行推进。这一点很多人没意识到，白白浪费了图结构带来的天然并行。

2.2 状态合并：并行带来的隐藏成本

并行不是免费的。同一 Superstep 里多个节点同时写状态，就涉及**状态合并（reducer）**的问题。默认情况下，LangGraph 对同一个 key 的并发写入会报错或者覆盖，你必须显式声明 reducer。

from typing import Annotated from operator import add class State(dict): results: Annotated[list, add]

这里用operator.add作为 reducer，意味着多个节点往results里 append 时，会被合并成一个列表，而不是互相覆盖。我踩过的坑是：早期没定义 reducer，两个并行节点都写results，结果只有一个节点的数据活下来，排查了半天才发现是合并策略的问题。

注意：reducer 的选择直接决定并行结果是否正确。列表累加用add，字典合并要自己写 merge 函数，计数用add也行但要小心重复。别默认“并行写同一个 key 没事”，这是并行场景下最容易翻车的地方。

2.3 并行度的实际边界

理论上同一 Superstep 的节点都能并行，但实际并行度受几个因素限制。第一是执行器：LangGraph 默认用线程池跑同步节点，池子大小有限；第二是外部资源：你并行调 10 个 API，但对方限流 5 QPS，那并行度再高也没用；第三是状态合并开销：并行节点越多，合并时的数据量越大，reducer 写得不好会成为新瓶颈。

我的经验是，单个 Superstep 里并行 5 到 10 个节点是比较舒服的区间。超过这个数，要么拆成多个 Superstep 分批，要么改用 Send API 做受控扇出。盲目追求“全并行”往往适得其反，尤其是节点之间有隐式资源竞争的时候。

3. Send API：运行时动态扇出的正确姿势

3.1 静态边 vs 动态扇出，差在哪

前面说的并行，前提是你在编译期就知道有几个分支。但真实场景经常是：用户上传了一份文档，你要按段落切分，每段独立处理，段落数是运行时才知道的。这时候静态边就无能为力了——你不可能为“未知数量的段落”预先画好边。

Send API 就是干这个的。它允许你在一个节点里，运行时动态生成一批并行的下游任务，每个任务带着自己的私有状态，最后再汇总。

from langgraph.types import Send def dispatch(state): return [ Send("process_chunk", {"chunk": c, "idx": i}) for i, c in enumerate(state["chunks"]) ] builder.add_conditional_edges("split", dispatch, ["process_chunk"])

dispatch节点返回一个Send列表，每个Send指定目标节点和该分支的私有输入。LangGraph 会把这些 Send 全部塞进下一个 Superstep 并行执行。这就是所谓的map-reduce 模式：dispatch 是 map 的发起，process_chunk 是 map 的执行，后续再 reduce 汇总。

3.2 Send 的私有状态与汇总

Send 传的第二个参数是该分支独有的状态，它不会污染全局 state。每个process_chunk只看到自己那份chunk和idx，处理完把结果写回一个带 reducer 的全局 key。

class State(dict): chunks: list summaries: Annotated[list, add] def process_chunk(state): chunk = state["chunk"] summary = llm_summarize(chunk) return {"summaries": [summary]}

这里summaries用add做 reducer，所有并行分支的结果会自动拼成一个列表。整个流程就是：split 切分 → dispatch 扇出 N 个 Send → N 个 process_chunk 并行 → 结果自动汇总。段落数从 3 变成 300，代码一行不用改，这就是动态扇出的价值。

3.3 Send 的常见误用

我见过最多的误用是在 Send 里传整个 state。有人图省事，Send("process", state)，结果每个分支都拿到全量状态，既浪费内存，又容易在合并时产生冲突。Send 的设计意图就是只传该分支需要的最小数据，保持分支隔离。

另一个坑是在 Send 目标节点里又发起 Send，形成嵌套扇出。这本身合法，但 Superstep 会一层层展开，如果层数深、分支多，Superstep 数量会爆炸，调试起来非常痛苦。我的建议是嵌套不超过两层，再深就该考虑用队列或者外部任务系统了。

4. 异步优化：别把 async 当万能药

4.1 异步解决的是 IO 等待，不是并行度

这是最需要澄清的一点。并行（parallelism）和异步（async）是两个维度。并行是“多个任务同时跑”，异步是“单个任务在等待 IO 时不阻塞”。LangGraph 的 Superstep 已经帮你做了并行调度，你加 async 优化的是每个节点内部的等待时间。

举个例子：一个节点要调 3 次 LLM API，每次 2 秒。如果同步串行写，这个节点耗时 6 秒；如果节点内部用asyncio.gather并发这 3 次调用，耗时降到 2 秒。注意，这里优化的是节点内部，节点本身还是占一个 Superstep 的位置。

import asyncio async def node_with_multi_calls(state): results = await asyncio.gather( call_llm(state["q1"]), call_llm(state["q2"]), call_llm(state["q3"]), ) return {"answers": list(results)}

4.2 同步节点和异步节点混用的陷阱

LangGraph 允许图里同时有同步节点和异步节点，但混用有讲究。如果图里有异步节点，整个图的执行会走异步路径；同步节点会被包一层丢到线程池里跑。这意味着：同步节点里的阻塞调用（比如requests.get）会占用线程池的线程，线程池满了，其他同步节点就得排队。

我实测过一个反直觉的现象：把几个同步节点改成 async 之后，整体反而变慢了。原因是这些节点内部用的是同步的 HTTP 库，改成 async 函数但里面还是阻塞调用，等于既没享受到异步的并发，又增加了事件循环的调度开销。异步要生效，整条调用链都得是异步的——用httpx.AsyncClient而不是requests，用async版的 SDK 而不是同步版。

4.3 异步节点的并发控制

异步节点内部用asyncio.gather并发调用时，如果不加限制，几百个并发请求打出去，很容易触发对方限流或者把自己的连接池打满。正确做法是用Semaphore控制并发数。

sem = asyncio.Semaphore(10) async def limited_call(q): async with sem: return await call_llm(q) async def node(state): results = await asyncio.gather(*[limited_call(q) for q in state["queries"]]) return {"answers": list(results)}

这个Semaphore(10)就是并发闸门，保证同时最多 10 个请求在飞。具体设多少，取决于对方的限流策略和你自己的资源。我的经验值是：外部 API 一般设 5 到 20，本地 CPU 密集任务设成 CPU 核数。

5. 并行执行外部命令的实战落地

5.1 场景拆解：为什么要在图里跑命令

热词里提到“langgraph 并行执行 linux 命令”，这其实是个很典型的场景：Agent 需要批量执行一批独立的 shell 命令（比如对多个仓库跑测试、对多台机器做健康检查、批量处理文件），这些命令彼此独立，天然适合并行。用 LangGraph 编排的好处是，命令的执行结果能回流到状态里，供后续节点做决策。

核心思路是：用 Send API 把命令列表扇出成 N 个并行分支，每个分支执行一条命令，结果汇总。

5.2 完整实现

import asyncio from langgraph.graph import StateGraph, START, END from langgraph.types import Send from typing import Annotated from operator import add class State(dict): commands: list outputs: Annotated[list, add] async def run_command(state): cmd = state["cmd"] proc = await asyncio.create_subprocess_shell( cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, stderr = await proc.communicate() return { "outputs": [{ "cmd": cmd, "code": proc.returncode, "stdout": stdout.decode(), "stderr": stderr.decode(), }] } def dispatch(state): return [Send("run_command", {"cmd": c}) for c in state["commands"]] builder = StateGraph(State) builder.add_node("run_command", run_command) builder.add_conditional_edges(START, dispatch, ["run_command"]) builder.add_edge("run_command", END) graph = builder.compile()

这里用asyncio.create_subprocess_shell而不是subprocess.run，是因为前者是异步的，多个命令能真正并发跑，而不是一个跑完再跑下一个。communicate()会等命令结束并收集输出，整个过程不阻塞事件循环。

5.3 并发数与超时控制

批量跑命令最大的风险是资源打爆和卡死。几十个命令同时起进程，CPU 和内存瞬间拉满；某条命令挂住不返回，整个图就卡在那。所以必须加并发限制和超时。

sem = asyncio.Semaphore(8) async def run_command(state): cmd = state["cmd"] async with sem: proc = await asyncio.create_subprocess_shell( cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) try: stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60) except asyncio.TimeoutError: proc.kill() return {"outputs": [{"cmd": cmd, "code": -1, "stderr": "timeout"}]} return {"outputs": [{"cmd": cmd, "code": proc.returncode, "stdout": stdout.decode(), "stderr": stderr.decode()}]}

Semaphore(8)控制同时最多 8 个命令在跑，wait_for加 60 秒超时，超时就 kill 掉进程并记录。这两道防线缺一不可。我踩过的坑是没加超时，一条ping命令因为网络问题挂了几分钟，整个图一直不返回，排查时还以为是 LangGraph 卡了。

注意：create_subprocess_shell会经过 shell 解析，如果命令字符串来自不可信输入，存在注入风险。生产环境建议用create_subprocess_exec传参数列表，或者对输入做严格校验。

5.4 结果汇总与后续决策

所有命令跑完后，outputs里是每条命令的返回码、stdout、stderr。后续节点可以基于这些结果做决策：比如统计失败数、提取关键输出、决定是否重试。这就是把“并行执行”和“状态驱动决策”结合起来的价值——不是单纯跑完就完事，而是让结果回流到图里继续流转。

def summarize(state): failed = [o for o in state["outputs"] if o["code"] != 0] return {"failed_count": len(failed), "failed_cmds": [o["cmd"] for o in failed]}

6. 常见问题与排查技巧实录

6.1 并行没生效，节点还是串行跑

最常见的排查路径：先确认这些节点是不是在同一个 Superstep。如果它们之间有边相连，那就是串行，别指望并行。用graph.get_graph().draw_mermaid()或者打印执行计划，看清楚拓扑。其次确认状态 key 有没有定义 reducer，没有 reducer 的并发写会导致引擎保守处理甚至报错。

6.2 异步节点报 “no running event loop”

这通常是因为在同步上下文里调用了异步节点，或者图本身是用invoke而不是ainvoke跑的。异步图必须用await graph.ainvoke(...)，用invoke会走同步路径，异步节点里的await就找不到事件循环。这个错误信息很直白，但新手容易忽略调用方式。

6.3 Send 扇出后结果丢失

八成是 reducer 没配对。Send 的每个分支写回全局 key 时，如果那个 key 没有 reducer，后写的会覆盖先写的，看起来就像“只有最后一个分支的结果”。检查Annotated[list, add]这类声明有没有漏。

6.4 并行命令把机器跑挂

并发数没控制。默认asyncio.gather会一次性把所有任务都发出去，命令多了直接打爆。加Semaphore，并且根据机器规格调整。CPU 密集的命令，并发数别超过核数；IO 密集的可以适当放宽。

6.5 独家避坑心得

第一，先画图再写代码。LangGraph 的并行完全由拓扑决定，动手前把 Superstep 划分想清楚，比写完再调快得多。第二，异步和并行分开优化。先确认图结构并行到位了，再考虑节点内部异步，顺序反了会互相干扰判断。第三，给每个并行分支加可观测性。并行执行出问题时，日志是交错的，最好给每个分支带上唯一标识（比如 Send 里的 idx），排查时能对上号。第四，超时和并发限制是并行场景的标配，不是可选项，任何对外部资源的并行调用都要有这两道闸。

7. 性能调优的取舍与实测体会

调优这件事，最忌讳的是“看到 async 就加、看到并行就上”。我实际做下来，LangGraph 的性能优化有个清晰的优先级：先优化图结构（减少不必要的串行依赖），再优化 Superstep 划分（把能并的都并了），最后才优化节点内部（异步、并发控制）。顺序错了，前面省下的时间会被后面的瓶颈吃掉。

举个实测对比。一个文档处理图，原本是 10 个段落串行处理，每段 1.5 秒，总耗时 15 秒。改成 Send 扇出并行后，10 个分支同时跑，理论耗时 1.5 秒，但实测是 3 秒左右——多出来的时间是状态合并和调度开销。再把节点内部改成异步调 LLM，耗时降到 2 秒出头。最后加Semaphore(5)限制并发，耗时回到 3 秒，但稳定性大幅提升，不再触发限流。你看，最快的配置不一定是最稳的，生产环境要的是稳中求快。

还有个容易被忽略的点：Superstep 之间的同步屏障是有成本的。每个 Superstep 结束都要等所有节点完成，如果某个节点特别慢（长尾），整个 Superstep 都被它拖住。这时候可以考虑把慢节点拆出来单独成一个 Superstep，或者给它单独的超时和降级策略。并行不是银弹，长尾问题在并行系统里反而更突出。

最后分享一个我常用的判断方法：拿一张纸，把图的每个节点画出来，标上预估耗时，然后按 Superstep 分组。如果某个 Superstep 里有一个节点耗时远超其他，那它就是瓶颈，要么优化它，要么把它挪走。这个方法土，但比任何 profiling 工具都直观，尤其适合在动手写代码之前做方案设计。