我在接手一个智能客服Agent的改造时卡了很久用户只是问了一句“帮我对比A、B、C三款产品的功能、价格和用户评价”系统却要串行地查产品库、查价格库、查评论库再到最后汇总回答整条链路跑下来差不多6到7秒。用户早就没耐心了负责人也只给了一句话能不能把这几步同时跑。那时候我正好在看LangGraph因为它把Agent工作流建模成图天然支持把没有依赖关系的节点并发执行。顺着这个思路我把链路改造成了并行化结构延迟直接砍到2秒左右。这篇文章就是基于这次改造加上后续几个项目里的实践把“基于LangGraph的AI Agent并行化设计模式”这件事彻底讲透内容包括并行化的底层原理、四种常见设计模式、完整可抄的代码以及一堆只有真跑过才会踩到的坑。适合正在用LangGraph做Agent编排、或者准备把串行链路改成并行的开发者和架构师。1. 为什么召回链路慢Agent并行化要解决的三个核心问题先说个扎心的事实很多Agent慢不是模型推理慢而是调用链路被写成了“一条道走到黑”。每个环节都要等上一步完全结束才轮到下一步外部API的网络往返时间被一个接一个地累加整体延迟自然又高又飘。并行化的直接价值就是把那些互相不依赖的步骤从“排队”改成“同时进行”。1.1 串行执行的时间账我给你算一笔挺直观的账。假设一个Agent要完成四件事查产品库、查价格库、查评论库、生成最终对比总结。串行执行时每次外部API调用往返耗时约1.2秒四步就是4.8秒加上中间状态整理、LLM处理总耗时6秒以上。如果前三步并行执行总耗时就是“最慢的那一步耗时”加上汇总环节的耗时通常2.5秒到3秒就结束。你不需要改任何模型不需要换更强的基础设施仅仅把执行顺序从串行改成并行就白赚了一半以上的延迟收益。这在多工具调用的场景里尤其明显。现在的Agent经常要同时检索网页、查知识库、翻数据库、看图理解这些都是独立的数据源彼此没有依赖关系。把它们串起来跑是一种典型的“用时间换简单”——代码是好写了但用户在等结果。而并行化本质上是“用空间换时间”让多个独立子任务共享同一个时间窗口。1.2 LangGraph的并行化底座图执行引擎与超级步LangGraph能做好并行化不是因为它有什么魔法而是因为它本身就把Agent流程当成一张有向无环图DAG来执行。图里的每个节点是一个逻辑单元节点之间通过边表示依赖关系。执行引擎每次会找出“当前所有入边都已完成的节点”把这些节点放到同一个超级步super-step里一起调度。这里的关键点在于如果两个节点都满足执行条件而且它们之间没有数据依赖LangGraph就会把它们作为同一批任务并发执行。做法上你可以用add_edge让一个上游节点扇出到多个下游节点也可以让多个上游节点的结果汇聚到一个下游节点。并行度是框架自动管理的不需要你自己写线程池或者维护异步任务队列。还有一个容易被忽略的点LangGraph的节点函数既支持同步也支持异步而并行节点内部如果各自调用了LLM或外部API那它们的耗时天然就是叠加的时长而不是排队时长。也就是说并行化不仅仅发生在“不同节点之间”你还可以在单个节点内部用await同时发起多个外部请求。所以它是“节点级并行 节点内并发”两层收益。1.3 并行化的适用边界不过必须先泼一盆冷水不是所有链路都适合并行化。如果两个任务之间有强数据依赖比如第二个任务的输入依赖第一个任务的输出那无论如何都不能同时跑。举个例子你要先对用户的消息做意图识别再根据意图决定调用哪个工具那“意图识别”和“工具调用”就是典型的串行关系。并行化只能发生在“意图识别完之后的多个工具之间”。还有一种情况要小心多个并行分支都在修改同一个全局状态而且没有定义好状态合并规则那很容易出现结果互相覆盖、数据丢失的问题。并行化不是无脑并发它要求你先想清楚每个分支产生什么数据、怎么合并。这也是为什么这篇文章后半部分会花大量篇幅讲状态Reducer和数据合并。2. 四种并行化设计模式从简单到进阶的完整拆解我把实际项目里用得最多、查资料时也最常被问到的并行化模式总结成了四种扇出/扇入、动态路由并行Worker、映射/归约、编排者/工作者。它们之间的区别主要在于“分支是固定的还是动态的”以及“谁来决定拆分成多少个子任务”。2.1 扇出/扇入固定分支的多路并发扇出/扇入是最基础的并行模式适合“分支数量和结构在写代码时就已经确定”的场景。所谓扇出就是上游节点同时触发多个下游节点扇入则是这些下游节点把结果汇聚到同一个汇总节点。我做一个客服Agent时就是这么干的用户的问题先进意图识别节点然后扇出到产品信息查询、库存查询、物流查询三个并行节点最后扇入到汇总节点拼装答案。三个查询节点互相独立总耗时等于查询最慢的那个节点。用LangGraph的写法核心就三行add_edgebuilder.add_edge(intent_node, product_query) builder.add_edge(intent_node, stock_query) builder.add_edge(intent_node, logistics_query) builder.add_edge(product_query, summary_node) builder.add_edge(stock_query, summary_node) builder.add_edge(logistics_query, summary_node)这个模式最大的优点就是简单、可预期、好调试。缺点则是不够灵活如果查询的数据源数量是动态变化的比如这次要查三个产品下次要查十个产品你就没法在写代码的时候把所有分支都画出来。这时候需要的是映射/归约模式。2.2 动态路由并行Worker先规划再分发动态路由并行Worker的核心思路是在流程里加一个“路由节点”先让这个节点分析当前输入并生成一份任务清单然后通过条件边把任务动态分发给不同的Worker节点。和扇出/扇入的固定结构相比这种模式里的分支数量不是写死的而是由路由节点运行时的输出决定的。我在一个文档处理Agent里用过这种模式。上游节点先识别输入文档的类型如果是PDF就发给PDF解析Worker是Word就发给Word解析Worker是图片就发给OCR Worker。这些Worker互相独立可以同时执行。实现方式并不复杂路由节点返回一个“下一步应该走哪条边”的值然后用add_conditional_edges绑定路由函数。builder.add_conditional_edges( router_node, lambda state: route_by_doc_type(state[doc_type]), {pdf: pdf_worker, word: word_worker, image: ocr_worker} )需要注意的是这种模式的并行度受限于“同一时刻有多少种任务需要执行”。如果路由结果最终只指向一个Worker那它就是串行的只有多个Worker同时被触发才算真正并行。所以实际项目里我通常用它处理“分类后各分支互不干扰”的场景而把真正需要大量同质子任务的情况交给映射/归约。2.3 映射/归约用Send API做动态并行如果说前两种模式还是“图结构层面的并行”那映射/归约就是“数据驱动的并行”。这种模式的核心诉求是我有N个相互独立的同类型任务需要把它们逐个处理完再汇总结果。这里的N在写代码时可能完全不知道只有运行到某个节点看到数据之后才知道。LangGraph专门为这种场景提供了SendAPI。它的作用和普通边的区别在于普通边的执行逻辑是“一个上游节点触发一个下游节点”而Send可以让你在单个节点内部同时发起多个下游任务的执行请求。每个Send都会独立地创建一份下游状态快照并作为独立分支去跑下游子图。我做用户评论情感分析时就拆成了“先把所有评论切成小块再用一个节点动态生成N个分析任务发给分析Worker最后用汇总节点做归约”。伪代码结构大致是这样from langgraph.types import Send def dispatch_comments(state): comments state[comments] return [Send(analyze_worker, {comment_chunk: chunk}) for chunk in split_comments(comments)] builder.add_conditional_edges(split_node, dispatch_comments, [analyze_worker]) builder.add_edge(analyze_worker, reduce_node)这里的关键点是Send创建的每个并行分支都会把各自的执行结果“追加”到共享状态里而不是覆盖。这是Map-Reduce能高效工作的前提。后面我会专门讲状态Reducer怎么写。2.4 编排者/工作者Agent自己拆任务编排者/工作者模式是并行化里最“智能”的一种有一个编排者Agent负责理解高层目标、拆解子任务、把子任务分配给多个Worker Agent去执行最后再把所有结果整合成最终输出。这种模式适合任务本身没有固定拆法需要依赖LLM来动态规划的场景。我用过一个调研Agent它就是先由主Agent把“调研某个主题”拆成“查背景”“找案例”“整理数据”三个子任务然后三个Worker并行执行各自的检索和分析最后主Agent再综合所有结果生成报告。在实现上这种模式通常是前面的路由/分发节点由LLM驱动Worker仍然是普通函数或子Agent。最大的工程挑战在于怎么编排状态主Agent的规划结果、每个Worker的结果、最终整合结果都需要定义清楚命名空间和合并规则。如果所有Worker都把结果写进同一个list[str]字段确实简单但最终汇总的时候会分不清哪段内容来自哪个子任务。我个人的做法是给每个子任务分配一个任务ID并把Worker结果按任务ID存成字典结构汇总节点再按ID去取数。下面用一张表对比这四种模式方便你选型模式分支数量谁决定拆法典型场景实现成本扇出/扇入固定写代码时确定多数据源并发查询低动态路由并行Worker有限动态路由节点文档类型分流处理中映射/归约高度动态数据内容N条数据批量处理中高编排者/工作者动态编排Agent复杂任务智能拆解高3. 实操把一个串行客服链路改造成混合并行链路理论部分说完了下面进入实战。我会用一个完整案例把“扇出/扇入”和“映射/归约”两种模式串在同一个Agent链路里这样你能直观看到它们各自承担什么职责也方便直接改改用到自己项目里。3.1 改造目标与状态设计模拟场景是一个电商客服Agent。用户输入“帮我对比产品A、B、C三款产品的功能、价格和用户评价并总结哪款更适合家庭使用。”原来的串行链路是先查产品A资料 → 再查产品B资料 → 再查产品C资料 → 再查三款产品的评论 → 最后LLM总结。一共5个步骤后一步都在等前一步。改造目标是把“查A、查B、查C、查评论”这四个互不依赖的查询全部并行执行。同时评论数据可能很多所以评论查询节点拿到数据之后还要用一个动态并行的情感分析子流程把一坨评论切成几块并行分析。状态设计是这类项目里最重要的一步。我建议把所有并行分支的输出都设计成“可追加的列表”结构。因为LangGraph里多个并行分支在执行完成时都会去更新同一个状态如果字段是普通的字符串或字典后写入的值就会覆盖先写入的值。只有把字段声明为Annotated[list, operator.add]才能让多个分支的写入操作自动累积成一个列表。下面是我常用的状态定义from typing import Annotated, TypedDict import operator class AgentState(TypedDict): query: str products: list[dict] product_details: Annotated[list[str], operator.add] comments: Annotated[list[str], operator.add] comment_sentiments: Annotated[list[dict], operator.add] final_answer: str这里的product_details和comments都会由多个并行分支写入所以都加上了operator.add这个Reducer。这样A、B、C三个查询分支分别写入自己的结果后最终product_details里就会有三个元素分别对应三款产品的资料。3.2 扇出扇入的落地代码我先搭一个最简化的LangGraph图包含四个节点意图识别节点、并行查询A节点、并行查询B节点、并行查询C节点。意图识别完成后统一扇出到三个查询节点三个查询节点再全部汇聚到最终总结节点。from langgraph.graph import StateGraph, START, END async def intent_node(state: AgentState) - dict: query state[query] # 这里用LLM做意图识别返回需要查询的产品列表 products extract_product_names(query) return {products: products} async def query_product_a(state: AgentState) - dict: info await fetch_product_detail(A) return {product_details: [info]} async def query_product_b(state: AgentState) - dict: info await fetch_product_detail(B) return {product_details: [info]} async def query_product_c(state: AgentState) - dict: info await fetch_product_detail(C) return {product_details: [info]} async def summary_node(state: AgentState) - dict: details state[product_details] # 这里调用LLM基于details生成对比总结 answer await generate_comparison(details) return {final_answer: answer} builder StateGraph(AgentState) builder.add_node(intent, intent_node) builder.add_node(query_a, query_product_a) builder.add_node(query_b, query_product_b) builder.add_node(query_c, query_product_c) builder.add_node(summary, summary_node) builder.add_edge(START, intent) builder.add_edge(intent, query_a) builder.add_edge(intent, query_b) builder.add_edge(intent, query_c) builder.add_edge(query_a, summary) builder.add_edge(query_b, summary) builder.add_edge(query_c, summary) builder.add_edge(summary, END) graph builder.compile()注意query_product_a返回的字典里product_details是一个只有一个元素的列表而不是直接返回字符串。这么写是因为Reducer的operator.add只会把两个列表拼起来如果你返回字符串Reducer执行时就会直接报错或者覆盖。这是LangGraph并行化编码里的第一道坎我见过不少新手在这里翻车。实际跑起来三个查询节点会在同一个超级步里并发执行。如果你在日志里观察执行过程会看到“query_a、query_b、query_c”三个节点几乎同时开始然后大约在同一个时间点先后返回。3.3 用Send API实现评论情感的动态并行分析接下来是更进阶的部分。评论查询节点返回的评论可能是20条也可能是200条。如果每次都把全部评论塞给一个大模型去分析既费token又容易丢细节如果串行一条条分析又非常慢。正确做法是用Send API做动态并行。先看代码from langgraph.types import Send async def fetch_comments_node(state: AgentState) - dict: comments await fetch_all_comments(state[query]) return {comments: comments} def split_comments_to_chunks(state: AgentState): comments state[comments] chunk_size 5 chunks [comments[i:i chunk_size] for i in range(0, len(comments), chunk_size)] return [ Send(analyze_chunk, {comment_sentiments: chunk}) for chunk in chunks ] async def analyze_chunk_node(state: AgentState) - dict: chunk state[comment_sentiments] results [] for comment in chunk: sentiment await analyze_sentiment(comment) results.append({comment: comment, sentiment: sentiment}) return {comment_sentiments: results} async def reduce_sentiment_node(state: AgentState) - dict: all_results state[comment_sentiments] positive sum(1 for r in all_results if r[sentiment] positive) negative sum(1 for r in all_results if r[sentiment] negative) # 把统计结果并入最终总结 return {comment_stats: {positive: positive, negative: negative}} builder.add_node(fetch_comments, fetch_comments_node) builder.add_node(analyze_chunk, analyze_chunk_node) builder.add_node(reduce_sentiment, reduce_sentiment_node) builder.add_edge(fetch_comments, split_comments) builder.add_conditional_edges( split_comments, split_comments_to_chunks, [analyze_chunk] ) builder.add_edge(analyze_chunk, reduce_sentiment) builder.add_edge(reduce_sentiment, summary)这里split_comments_to_chunks函数返回的不是一个字符串类型的路由结果而是一个Send对象列表。每个Send都会把对应的chunk作为独立分支启动一次analyze_chunk_node的执行。框架会自动为每个Send创建一个独立的状态快照并并发运行。这里有个细节值得记住Send构造函数里的第二个参数是传给下游节点的“局部状态”不需要包含全量AgentState里的所有字段。你在下游节点里只会看到这个局部状态。这种设计其实是刻意的它强制你显式声明每个子任务依赖哪些数据避免并行分支拖着一大堆无关状态到处跑。所有analyze_chunk分支的结果都会通过Annotated[list, operator.add]追加到comment_sentiments里。等到reduce_sentiment_node执行时已经汇总了全部分支结果。也就是说reduce_sentiment_node天然就是一个Map-Reduce里的Reduce阶段。3.4 并发限流信号量包装外部API动态并行看着很爽但有个隐藏炸弹Send可以一次性创建几百个分支意味着它会同时发出几百个外部API请求。你的服务不一定会被打爆但公共API的限流策略大概率会触发429错误。我在一次处理3000条短信数据的批处理任务里就踩过这个坑。解决办法是在节点内部用信号量Semaphore控制并发数量。LangGraph的节点本身是异步函数所以可以直接用asyncio.Semaphore做并发限流。import asyncio semaphore asyncio.Semaphore(5) async def analyze_chunk_node(state: AgentState) - dict: chunk state[comment_sentiments] results [] async def limited_analyze(comment): async with semaphore: return await analyze_sentiment(comment) results await asyncio.gather(*[limited_analyze(c) for c in chunk]) return {comment_sentiments: results}这样即使你动态派发了100个子任务每个子任务内部同一时间最多只有5个请求在飞。实验结果非常明显限流错误从几百次降到了0次整体耗时才多了不到20%。所以动态并行一定要配合并发限流这不是可选项是必选项。3.5 完整链路的技术要点汇总把整个案例串起来改造后的客服Agent链路是用户输入进入意图识别节点输出需要对比的产品列表同时触发评论拉取。产品A、B、C三个查询节点在同一个超级步并发执行。评论拉取节点拿到原始评论后通过Send API动态切分并并行分析情感。所有查询和情感分析结果扇入到总结节点由LLM生成最终对比结论。整个链路的延迟从串行的6秒多降到了并行的2.5秒左右而且评论越多并行分析的优势越明显。这个改造方案里我只用了扇出/扇入和映射/归约两种模式已经能覆盖绝大多数业务场景。4. 常见问题与排查技巧实录并行化改造过程中真正花时间的往往不是写并行代码而是排查那些“并行以后才会出现的神奇问题”。我来整理几个我实际踩过的坑以及对应的排查思路。4.1 并行结果互相覆盖的状态陷阱最典型的现象是多个并行分支都执行了但最终状态里只剩下其中一个分支的结果。这时候十有八九是状态字段没有配置Reducer。数据写入同一个字段时默认行为是把旧值替换成新值。并行分支一个接一个完成后完成的就把先完成的覆盖了。排查方法很简单打印每个节点返回前状态里该字段的值看是不是在执行过程中被重置过。修复方式就是给字段加Annotated[list, operator.add]。记住并行分支写同一个字段时这个字段必须设计成“可累积”的数组结构。4.2 动态分支数量失控与限流另一个高频问题是动态并行分支数量太大外部服务开始报错。可能是限流也可能是下游数据库连接数被打满。排查时先看日志里有没有大量HTTP 429或者连接超时同时统计每个超级步里实际创建了多少个分支。解决思路有三层第一在分发函数里限制每个Send的子集大小比如每个子任务只处理5条数据第二在节点内部加Semaphore控制并发请求数第三给整个图执行设置超时或最大并发步数。这三层可以叠加使用我一般会同时设第二层和第三层。4.3 并行链路调试和可观测性并行化以后节点执行顺序不再是一条直线日志也变成了一团乱麻。排查问题的难度会明显上升。所以我强烈建议在关键节点里写上明确的日志标注节点名和状态字段的长度。还可以用LangGraph提供的状态历史功能在每次图执行结束后查看每一个超级步里执行了哪些节点、每个节点产出了什么状态变更。这里有个实用技巧给每个并行分支返回的数据加上“来源标记”。比如查询产品A的分支在返回时可以写成{product_details: [{product: A, info: info}]}这样汇总节点就知道每条数据来自哪个分支排查问题时一眼就能定位。这个习惯在并行规模变大以后收益特别高。4.4 故障恢复与checkpoint建议并行分支如果有一个失败整个图通常都会失败。但有些场景下你只希望失败分支重试其他结果保留。LangGraph的checkpoint机制帮了大忙它会在每个超级步结束时保存状态快照图失败后你可以从最近一个成功的checkpoint恢复执行。实际项目中我建议给重要的并行链路开启checkpoint并且专门写一个异常处理节点捕获单个Worker的失败把错误信息作为一个特殊结果写进状态而不是让整个流程崩掉。这样即使某个数据源挂了汇总节点也能基于已有结果生成一个“部分成功”的答复而不是直接给用户返回报错。下面把高频问题整理成一张速查表现象可能原因排查方向解决办法并行结果只剩最后一个状态字段没有Reducer看状态定义和节点返回值加Annotated[list, operator.add]分支数量爆炸导致限流没有控制并发度检查分支数和外部API错误加Semaphore、限制子任务粒度某个分支挂了整个图失败缺少异常隔离看日志定位失败节点增加catch或fallback节点日志顺序混乱难排查缺少来源标记看状态变更快照返回数据里加来源标记并行后耗时没下降节点存在隐藏依赖观察超级步调度检查是否真并发、是否有公共瓶颈最后再说点我的个人体会。LangGraph的并行化设计并不是越复杂越好。扇出/扇入能解决80%的延迟问题而且它足够直观、容易排查。映射/归约适合处理真正数据集驱动的动态并行场景。至于编排者/工作者它上限很高但也意味着你要面对更复杂的状态管理和更不确定的LLM规划质量。如果你刚开始接触这块我建议不要一上来就上最复杂的模式。先挑一个延迟最高的串行链路用扇出/扇入改造一遍把状态Reducer和并行调试这套基本功练熟再去碰Send API和动态并行。每次改造都先画出节点依赖图明确哪些节点真的没有依赖关系哪些只是表面上看起来没有。这个习惯比我见过任何花哨的工具都管用。