1. 为什么单靠链式调用撑不住复杂场景刚接触 LangGraph 那会儿我习惯性地把每个节点都写成串行——A 跑完跑 BB 跑完跑 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 执行耗时对比与验证方法为了验证优化效果我习惯在节点里打时间戳。下面是我实测的一组数据模拟环境执行方式三路召回总耗时说明全串行同步约 1.2 秒0.50.30.4并行同步约 0.5 秒取最慢但线程阻塞并行异步约 0.5 秒取最慢且不阻塞事件循环并行异步高并发约 0.5 秒并发数增加时优势更明显验证方法很简单在入口和出口各记一个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 常见问题速查表问题现象可能原因排查方向解决手段并行结果只剩一个字段无 reducer检查 State 定义加 Annotated reducer耗时等于各分支之和图结构连成串行检查边的连接改为从同一节点分叉event loop 报错嵌套事件循环检查是否用了 asyncio.run改用 awaitSend 任务失败率高并发过高看连接池/内存分批或加信号量单分支失败全图挂无异常处理检查节点内 try节点内捕获异常异步没提速任务是 CPU 密集分析任务类型改用多进程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 或者拆图来实现。最后分享一个我常用的小技巧调试并行流程时把每个节点的执行时间打印出来画成时间轴。如果看到本该并行的节点在时间轴上首尾相接而不是重叠那基本就是图连错了或者异步没生效。这个可视化方法比看日志快得多一眼就能定位问题。