LangGraph并行执行与异步优化：Superstep机制、Send API及RAG多路召回实战

1. 为什么单靠链式调用撑不住复杂场景

刚接触 LangGraph 那会儿，我习惯性地把每个节点都写成串行——A 跑完跑 B，B 跑完跑 C。简单问答链路没问题，可一旦碰上"同时查三个数据源再汇总"这种需求，整个图就跑成了线性队列，原本能并行的事情硬生生拖成了串行，延迟直接翻倍。后来我把这套并行执行和异步优化的机制啃了一遍，才发现 LangGraph 在这块的設計其实相当克制但够用，关键在于你得理解它底层的调度模型。

LangGraph 的核心执行单位叫Superstep（超步），这个概念借鉴自分布式计算里的 BSP 模型。简单说，一个 Superstep 就是"一轮并行执行"——在这一轮里，所有被激活的节点同时开跑，全部跑完后系统做一次状态合并，然后进入下一轮。你可以把它想象成接力赛里的"棒次"：同一棒次内所有选手同时出发，等所有人都到交接区了，才统一交棒进入下一棒。这个模型决定了 LangGraph 的并行不是"谁先好谁先走"，而是"同轮齐步走"。

理解这一点非常关键，因为它直接解释了很多新手困惑的现象：为什么我明明让两个节点并行，结果一个快一个慢，快的那个却要等慢的？答案就在 Superstep 的同步屏障机制里——同一轮内，系统会等所有节点执行完毕才推进状态。这不是缺陷，而是为了保证状态一致性做的取舍。如果你需要"谁先完成谁先触发下游"，那得用别的手段，比如Send API做动态分发，或者干脆把逻辑拆到不同的图里。

那 LangGraph 到底解决了什么问题？我的总结是三点：第一，把"哪些节点可以同时跑"这件事从手写调度里解放出来，你只需要在图上连边，框架自动识别并行分支；第二，异步节点让 I/O 密集型任务不再阻塞事件循环，比如同时调三个外部接口，用async def写节点，整体耗时约等于最慢那个而不是三者之和；第三，Send API 支持运行时动态生成并行任务，处理"根据上一步结果决定要并行处理几条数据"这类场景特别顺手。

这套东西适合谁？如果你正在用 LangGraph 搭多智能体协作、RAG 多路召回、批量数据处理这类流程，并且已经感觉到串行执行拖了后腿，那这篇内容就是给你准备的。下面我会从设计思路、核心机制、实操落地到踩坑排查，把并行与异步这两块讲透。

2. 并行执行的整体设计与调度逻辑

2.1 并行分支是怎么被识别出来的

LangGraph 判断并行的依据非常朴素：看一个节点的出边是否指向多个下游节点。如果你从节点 A 连了两条边分别到 B 和 C，那 B 和 C 就会被放进同一个 Superstep 里并行执行。这里没有复杂的依赖分析，就是纯粹的图结构决定。

我一开始以为它会做什么智能的依赖推断，后来读源码才明白，LangGraph 的调度器就是按"当前 Superstep 里所有就绪节点"来批量执行的。所谓"就绪"，指的是该节点的所有入边来源都已在之前的 Superstep 中完成。这个规则简单但有效，你只要把图画对，并行就是自然发生的。

举个我实际用过的例子：一个内容审核流程，需要同时做"敏感词检测"和"情感倾向分析"，两个结果都出来后综合判定。图结构就是入口节点连两条边分别到这两个检测节点，两个检测节点再各自连边到汇总节点。运行时，两个检测节点自动进入同一 Superstep 并行跑，汇总节点等它们都完成才启动。

from langgraph.graph import StateGraph, START, END from typing import TypedDict class State(TypedDict): text: str sensitive_result: str sentiment_result: str final: str def check_sensitive(state: State): # 模拟敏感词检测 return {"sensitive_result": "pass"} def check_sentiment(state: State): # 模拟情感分析 return {"sentiment_result": "neutral"} def aggregate(state: State): return {"final": f"{state['sensitive_result']}-{state['sentiment_result']}"} builder = StateGraph(State) builder.add_node("sensitive", check_sensitive) builder.add_node("sentiment", check_sentiment) builder.add_node("aggregate", aggregate) builder.add_edge(START, "sensitive") builder.add_edge(START, "sentiment") builder.add_edge("sensitive", "aggregate") builder.add_edge("sentiment", "aggregate") builder.add_edge("aggregate", END) graph = builder.compile()

这段代码里，sensitive和sentiment就是天然的并行分支。你不需要写任何"并行"关键字，框架自己会识别。

2.2 Superstep 的同步屏障与状态合并

前面提到 Superstep 是"同轮齐步走"，这里展开说说状态合并的细节。每个 Superstep 结束时，LangGraph 会把这一轮所有节点返回的状态更新合并进全局 State。合并规则取决于你定义的 reducer——如果某个字段没指定 reducer，默认是"后写覆盖"；如果指定了operator.add之类的 reducer，就会做累积。

这个机制有个容易踩的坑：并行节点如果同时写同一个字段，且没定义 reducer，结果是不确定的。因为两个节点在同一 Superstep 里都返回了该字段的更新，合并顺序取决于内部调度，你没法保证谁先谁后。我早期就吃过这个亏，两个并行节点都往messages里塞内容，结果偶尔丢数据。后来给字段加了Annotated[list, operator.add]才稳定。

from typing import Annotated import operator class State(TypedDict): text: str # 用 reducer 保证并行写入时累积而非覆盖 logs: Annotated[list, operator.add]

提示：只要你的图里存在并行分支，就要仔细检查这些分支是否会写同一个 State 字段。会写的话，务必配 reducer，否则就是随机丢数据的隐患。

2.3 并行不等于无序：执行顺序的可控性

有人会担心并行导致结果不可预测。实际上 LangGraph 保证了同一 Superstep 内的节点是并行无序的，但 Superstep 之间是严格有序的。也就是说，跨轮次的执行顺序完全确定，只有同轮内的相对顺序不保证。这个特性让流程整体可控——你只要把有依赖关系的节点放在不同轮次，就能保证正确的执行顺序。

我通常的做法是：把"必须先后执行"的逻辑用边明确串起来，把"互不依赖"的逻辑放在同一层并行。这样既拿到了并行的性能收益，又不用担心顺序问题。判断标准很简单——如果节点 B 需要读节点 A 写的字段，那 A 和 B 就不能在同一 Superstep，必须用边连起来。

3. 异步优化的核心机制与实操要点

3.1 同步节点与异步节点的本质区别

LangGraph 支持节点函数写成def或async def。这两者的区别不是"快慢"，而是"阻塞与否"。同步节点在执行时会占用一个线程，如果里面做了网络请求、文件读写这类 I/O 操作，线程就干等着。异步节点用await把等待时间让出去，事件循环可以在这段时间调度其他任务。

我实测过一个场景：三个节点各自调用一个耗时 1 秒的外部接口。全用同步写法，总耗时约 3 秒；全改成异步，总耗时约 1 秒出头。差距就是这么直接。但要注意，异步的收益只在 I/O 密集型任务上明显，如果是纯 CPU 计算（比如大矩阵运算），异步反而因为事件循环开销略慢，这时候该用多进程而不是异步。

import asyncio import httpx async def fetch_data(state: State): async with httpx.AsyncClient() as client: # 三个请求并发发出 results = await asyncio.gather( client.get("https://api.example.com/a"), client.get("https://api.example.com/b"), client.get("https://api.example.com/c"), ) return {"data": [r.json() for r in results]}

这段代码里，asyncio.gather让三个请求真正并发，而不是顺序等待。这是异步优化最常用的手法。

3.2 异步节点的调度与事件循环管理

LangGraph 在运行异步图时，内部会维护一个事件循环。如果你在同步环境里调用异步图，需要用asyncio.run()或graph.ainvoke()。这里有个我踩过的坑：不要在异步节点里调用asyncio.run()，因为事件循环已经在跑了，再嵌套一个会直接报错。正确做法是用await调用协程，或者用asyncio.create_task创建任务。

另一个细节是，LangGraph 的异步执行对节点函数的签名有要求——必须是async def，且返回状态更新字典。如果你混用同步和异步节点，框架也能处理，但同步节点会阻塞事件循环，所以能异步就全异步，别混着来。

3.3 用 Send API 做动态并行分发

Send API 是我觉得 LangGraph 里最被低估的功能。它解决的是"运行时才知道要并行几路"的问题。比如你有一个列表，长度不定，需要为每个元素并行跑一个子任务，这时候静态的图结构就不够用了，得用 Send 动态生成。

from langgraph.types import Send def dispatch(state: State): # 根据输入动态生成并行任务 return [Send("process_item", {"item": item}) for item in state["items"]] def process_item(state: dict): # 处理单个元素 return {"results": [state["item"] * 2]} builder.add_node("dispatch", dispatch) builder.add_node("process_item", process_item) builder.add_conditional_edges("dispatch", lambda s: s)

Send的第一个参数是目标节点名，第二个是传给该节点的状态。每个 Send 都会生成一个独立的并行执行实例，它们在同一 Superstep 里跑。这个模式在批量处理、多路召回、map-reduce 类场景里特别好用。

注意：Send 生成的任务数量如果很大（比如上千个），要考虑资源限制。LangGraph 默认没有并发上限，任务太多可能把内存或连接池打爆。我一般会分批，或者用信号量控制并发数。

4. 完整实操：搭一个并行异步的 RAG 召回流程

4.1 场景定义与图结构设计

假设我们要做一个多路召回的 RAG 系统：用户提问后，同时从"向量库检索""关键词检索""知识图谱查询"三个来源召回文档，然后合并去重，最后交给大模型生成答案。三个召回源互不依赖，天然适合并行；而且它们都是 I/O 密集型（查数据库、调接口），适合异步。

图结构设计如下：入口节点接收问题，然后分三条边到三个召回节点，三个召回节点都连到合并节点，合并节点连到生成节点。三个召回节点用async def写，内部并发调用各自的数据源。

4.2 关键代码实现与参数说明

import asyncio from typing import Annotated, TypedDict import operator from langgraph.graph import StateGraph, START, END class RAGState(TypedDict): question: str vector_docs: list keyword_docs: list kg_docs: list merged: Annotated[list, operator.add] answer: str async def vector_search(state: RAGState): await asyncio.sleep(0.5) # 模拟向量库查询 return {"vector_docs": ["v1", "v2"]} async def keyword_search(state: RAGState): await asyncio.sleep(0.3) # 模拟关键词检索 return {"keyword_docs": ["k1", "k2"]} async def kg_query(state: RAGState): await asyncio.sleep(0.4) # 模拟图谱查询 return {"kg_docs": ["g1"]} def merge(state: RAGState): all_docs = state["vector_docs"] + state["keyword_docs"] + state["kg_docs"] # 去重 unique = list(dict.fromkeys(all_docs)) return {"merged": unique} def generate(state: RAGState): return {"answer": f"基于{len(state['merged'])}篇文档生成答案"} builder = StateGraph(RAGState) builder.add_node("vector", vector_search) builder.add_node("keyword", keyword_search) builder.add_node("kg", kg_query) builder.add_node("merge", merge) builder.add_node("generate", generate) builder.add_edge(START, "vector") builder.add_edge(START, "keyword") builder.add_edge(START, "kg") builder.add_edge("vector", "merge") builder.add_edge("keyword", "merge") builder.add_edge("kg", "merge") builder.add_edge("merge", "generate") builder.add_edge("generate", END) graph = builder.compile() # 异步调用 result = await graph.ainvoke({"question": "什么是 LangGraph"})

三个召回节点并行执行，总耗时约等于最慢的那个（0.5 秒），而不是三者之和（1.2 秒）。这就是并行加异步带来的实际收益。

4.3 执行耗时对比与验证方法

为了验证优化效果，我习惯在节点里打时间戳。下面是我实测的一组数据（模拟环境）：

验证方法很简单，在入口和出口各记一个time.time()，打印差值。如果并行生效，耗时应该接近最慢分支而不是总和。如果发现耗时还是总和，八成是图结构连错了——检查是不是不小心把三个召回节点串起来了。

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

5.1 并行节点结果丢失或覆盖

这是最高频的问题。症状是：明明两个并行节点都返回了数据，最终 State 里只剩一个。原因几乎都是没配 reducer。LangGraph 默认对未标注 reducer 的字段采用覆盖策略，并行写入时后写的把先写的盖掉了。

解决办法是给会被并行写入的字段加Annotated[类型, reducer]。列表用operator.add，字典可以用自定义合并函数，数字用operator.add或max看需求。我现在的习惯是：只要字段可能被多个节点写，一律加 reducer，哪怕当前是串行，将来改成并行也不会出问题。

5.2 异步节点报 "event loop is already running"

这个错误通常出现在你试图在异步节点里再启动一个事件循环。比如在async def节点里调用了asyncio.run()。事件循环是单例的，不能嵌套。正确做法是用await直接调用协程，或者用asyncio.create_task把任务挂到当前循环上。

还有一种情况是在 Jupyter Notebook 里跑异步图。Notebook 本身已经有一个事件循环，你直接await graph.ainvoke()就行，别用asyncio.run()。如果非要用，得用nest_asyncio打补丁，但我不推荐，容易出玄学问题。

5.3 Send API 任务数过多导致资源耗尽

用 Send 动态分发时，如果一次生成几百上千个任务，可能会把数据库连接池、HTTP 连接池或者内存打爆。我遇到过一次，Send 生成了 2000 个任务，结果 HTTP 客户端连接数超限，大量请求失败。

解决办法有两个：一是分批处理，把大列表切成小块，每块跑完再跑下一块；二是用信号量控制并发，在节点内部用asyncio.Semaphore限制同时执行的任务数。

sem = asyncio.Semaphore(50) # 最多 50 个并发 async def process_item(state: dict): async with sem: # 实际处理逻辑 return {"results": [state["item"]]}

5.4 并行分支里有一个失败怎么办

LangGraph 默认的行为是：如果某个节点抛异常，整个图执行会中断。这在并行场景下意味着一个分支失败会拖垮全部。如果你希望"部分失败不影响整体"，需要在节点内部做异常捕获，返回一个标记失败的状态，然后在合并节点里处理。

我的做法是在每个并行节点里包一层 try-except，失败时返回{"error": str(e), "docs": []}，合并节点检查 error 字段决定是否降级。这样单路召回失败不会导致整个 RAG 流程挂掉。

5.5 常见问题速查表

6. 性能调优的几个实战心得

6.1 并发数的甜点区在哪里

并发不是越高越好。我实测下来，对于外部 API 调用类任务，并发数在 20 到 50 之间通常是最优的。再往上，收益递减，因为瓶颈转移到了网络带宽或对方服务的限流。对于数据库查询，并发数受连接池大小限制，一般设成连接池的 80% 左右比较稳妥。

判断甜点区的方法很简单：逐步提高并发数，观察总耗时。当耗时不再明显下降，甚至开始上升时，就说明过了甜点区。我一般会写个小脚本跑几组不同并发数的对比，找出当前环境的最优值。

6.2 状态设计对性能的隐性影响

State 字段的设计会间接影响性能。如果 State 里塞了大对象（比如几 MB 的文档内容），每个 Superstep 的状态合并都要复制这些数据，开销不小。我的建议是：State 里只放必要的小数据，大内容用引用或 ID 代替，需要时再去取。

另外，reducer 的选择也有讲究。operator.add对列表是拼接，如果列表很大，每次合并都创建新列表，开销累积起来很可观。这种场景可以考虑用可变对象加原地修改，但要小心并发安全。

6.3 什么时候该拆图而不是硬并行

不是所有能并行的都该并行。如果一个流程里并行分支特别多，图会变得很难维护，调试也麻烦。我的经验是：如果一个子流程有 5 个以上节点且内部依赖复杂，就把它拆成独立的子图，主图只负责调度子图。这样每个子图内部可以有自己的并行和异步策略，主图保持清爽。

LangGraph 支持子图嵌套，子图可以像普通节点一样被调用。这个模式在大型多智能体系统里特别有用，每个智能体是一个子图，主图负责协调它们之间的并行和依赖。

6.4 监控与可观测性

并行异步流程出问题时，最难的是定位是哪个分支、哪个环节出的错。我强烈建议在节点里加日志，记录节点名、开始时间、结束时间、返回状态的关键字段。LangGraph 本身有回调机制，可以挂 LangSmith 之类的追踪工具，但即使不用外部工具，自己打日志也能解决大部分排查需求。

我通常会在每个节点入口和出口各打一条日志，格式统一成[节点名] 事件 时间戳 关键信息。这样出问题时，把日志按时间排序，就能还原出完整的执行时序，哪个分支慢、哪个分支失败一目了然。

7. 从串行到并行异步的迁移路径

如果你手上已经有一个跑得好好的串行 LangGraph 项目，想迁移到并行异步，我的建议是分三步走，别一步到位。

第一步，先识别可并行点。把图里所有节点列出来，画出依赖关系。凡是互不依赖、不需要读对方输出的节点，就是候选并行点。这一步只做分析，不改代码。

第二步，先改图结构，再改异步。把可并行节点的边改成从同一上游分叉，跑一遍确认结果正确。这一步先不碰异步，用同步节点验证并行逻辑没问题。很多人一上来就同步异步一起改，出了问题分不清是并行连错了还是异步写错了。

第三步，逐个把 I/O 节点改成异步。从最耗时的节点开始，改成async def，用await调用外部服务。每改一个跑一次测试，确认没问题再改下一个。全部改完后，用ainvoke替代invoke调用整个图。

这个路径的好处是每一步都可验证、可回滚。我迁移过一个十几个节点的流程，按这个节奏走，全程没出过大问题。踩过的唯一坑是某个节点内部用了同步的数据库驱动，改成异步后报错，换成异步驱动就好了。所以迁移前最好确认一下依赖库是否支持异步。

提示：迁移过程中保留一份串行版本的图作为对照，用同样的输入跑两个版本，对比输出是否一致。这是验证迁移正确性最直接的办法。

8. 写在最后的一点个人体会

并行和异步这两块，刚上手时容易陷入"为了并行而并行"的误区。我见过有人把本来就有依赖关系的节点硬拆成并行，结果为了同步状态写了一堆额外逻辑，反而更复杂。判断标准其实很朴素：两个节点之间有没有数据依赖？没有才并行，有就串行。异步同理，I/O 密集才异步，CPU 密集用多进程。

另外，LangGraph 的 Superstep 模型决定了它的并行是"轮次同步"的，这跟某些框架的"流水线并行"不一样。理解这个差异，你就能预判哪些场景适合它、哪些不适合。需要严格同步屏障的场景（比如多路结果必须全部到齐才能继续），LangGraph 的模型天然契合；需要"谁先好谁先走"的流式场景，就得靠 Send 或者拆图来实现。

最后分享一个我常用的小技巧：调试并行流程时，把每个节点的执行时间打印出来，画成时间轴。如果看到本该并行的节点在时间轴上首尾相接而不是重叠，那基本就是图连错了或者异步没生效。这个可视化方法比看日志快得多，一眼就能定位问题。