LangGraph实战②——Parallelization与Diamond拓扑上一节结尾留了个扣子链的软肋是延迟串行叠加四步链每步2秒用户就等8秒。今天就来解这个扣子——用一个分析→多语言并行翻译→合并的Diamond拓扑把Send API扇出、Reducer扇入、单路失败隔离这三块硬骨头一次啃完最后附上串行vs并行的实测对照数据。本文导航Diamond把队形从纵队变成雁阵完整代码控制台输出扇出的三块技术积木什么时候不该并行串行 vs 并行完整对照表Diamond把队形从纵队变成雁阵先统一概念。所谓Diamond菱形拓扑指的是这种形状一个节点展开成N个并行分支最后再汇聚到一个合并节点——展开、收口画出来就是个菱形。Send 扇出一变三Send 扇出Send 扇出失败也照样抵达merge以失败标记的形式STARTanalyze分析确定源文本与目标语言translate:en英语翻译translate:ja日语翻译translate:sw斯瓦希里语翻译 ❌故意失败merge扇入合并结果失败兜底END业务场景设定把一段产品介绍同时翻成英语、日语、斯瓦希里语对最后一个是我故意埋的雷专门用来演示一路失败不拖垮全局。有人会问为什么不写三个翻译节点、三条边三个语言可以三十个呢每个用户自定义语言组合呢静态连边把并行度焊死在图装配期而Send把并行度留给运行时——Send(translate, payload)想发几份发几份翻译节点只写一个。这就是Part 2说的动态分发边的实战价值。完整代码# translate_diamond.py —— LangGraph实战②Diamond拓扑多语言并行翻译# 依赖pip install langgraph0.4 Python 3.10fromtypingimportTypedDict,Annotatedimportoperatorimporttimefromlanggraph.graphimportStateGraph,START,ENDtry:fromlanggraph.typesimportSendexceptImportError:fromlanggraph.constantsimportSend SRCGraph Engineering 是编排多智能体的工程学节点是职责边是协作。LANGS[en,ja,sw]# sw不支持 → 演示单路失败隔离TRANSLATIONS{# 模拟LLM翻译结果真实项目换成模型调用en:Graph Engineering is the discipline of orchestrating multi-agent systems: nodes are responsibilities, edges are collaboration.,ja:Graph Engineeringとはマルチエージェントを編成する工学であるノードは責務、エッジは協働である。,}# 共享状态扇入靠Reducer classTState(TypedDict,totalFalse):src:str# 源文本langs:list# 目标语言运行时数据translations:Annotated[list[dict],operator.add]# 扇入并行结果追加合并final:dict# 合并后的交付物log:Annotated[list[str],operator.add]# 节点1个分析 1个翻译Worker 1个合并 defanalyze(state:TState):print(f[analyze ] 源文本{len(state[src])}字目标语言:{state[langs]})return{log:[analyze]}deftranslate(payload:dict):翻译Worker每个Send实例独立执行一份。注意入参是Send的payload而非全量状态langpayload[lang]time.sleep(0.3)# 模拟真实LLM延迟IO等待iflangnotinTRANSLATIONS:print(f[translate:{lang}] ❌ 不支持该语言登记失败不抛异常)return{translations:[{lang:lang,ok:False,text:None}],log:[ftranslate:{lang}:fail]}print(f[translate:{lang}] ✅ 翻译完成)return{translations:[{lang:lang,ok:True,text:TRANSLATIONS[lang]}],log:[ftranslate:{lang}:ok]}defmerge(state:TState):扇入节点合并所有并行结果失败语言回退原文交付final,failed{},[]fortinstate[translations]:final[t[lang]]t[text]ift[ok]elsef[回退原文]{state[src]}ifnott[ok]:failed.append(t[lang])print(f[merge ] 成功{len(state[translations])-len(failed)}路 f失败{len(failed)}路{( - str(failed))iffailedelse})return{final:final,log:[merge]}# 扇出Send动态分发 deffan_out(state:TState):运行时按langs列表扇出发几份、发给谁装配图时一无所知return[Send(translate,{lang:lang,src:state[src]})forlanginstate[langs]]# 装配菱形 builderStateGraph(TState)builder.add_node(analyze,analyze)builder.add_node(translate,translate)builder.add_node(merge,merge)builder.add_edge(START,analyze)builder.add_conditional_edges(analyze,fan_out)# 上半菱形扇出builder.add_edge(translate,merge)# 下半菱形扇入builder.add_edge(merge,END)graphbuilder.compile()if__name____main__:print( 串行基线循环顺序翻译 )t0time.perf_counter()forlanginLANGS:time.sleep(0.3)# 同样的模拟延迟print(f serial:{lang}done)serial_costtime.perf_counter()-t0print(f 串行耗时{serial_cost:.3f}s\n)print( Diamond并行版 )t1time.perf_counter()resultgraph.invoke({src:SRC,langs:LANGS})par_costtime.perf_counter()-t1print(f 并行耗时{par_cost:.3f}s 加速比{serial_cost/par_cost:.2f}x\n)print( 最终交付物 )forlang,textinresult[final].items():print(f[{lang}]{text[:46]}...)控制台输出python translate_diamond.py并行分支的打印顺序每次可能不同这是并行的正常现象 串行基线循环顺序翻译 serial:en done serial:ja done serial:sw done 串行耗时 0.905s Diamond并行版 [analyze ] 源文本37字目标语言: [en, ja, sw] [translate:ja] ✅ 翻译完成 [translate:en] ✅ 翻译完成 [translate:sw] ❌ 不支持该语言登记失败不抛异常 [merge ] 成功2路 失败1路 - [sw] 并行耗时 0.308s 加速比 2.94x 最终交付物 [en] Graph Engineering is the discipline of orchestr... [ja] Graph Engineeringとはマルチエージェントを編成する工学で... [sw] [回退原文] Graph Engineering 是编排多智能体的工程学节点是职责边是协作。三个观察点比代码本身更重要加速比2.94x接近理论值3x。三路延迟都是0.3s并行后总耗时≈最慢那一路0.308s而不是三路之和0.905s。翻译成生产语言3路各5秒的LLM调用用户感知从15秒变成5秒sw路失败了en和ja照常交付。失败被封装成{ok: False}随流转运在merge节点统一兜底回退原文。这就是失败隔离——一路翻车不掀桌merge只执行了一次。三个translate实例虽然并行但都被框架归入同一个超步superstepmerge等所有分支到齐才开始。这是LangGraph替你做的同步屏障手写多线程代码时最容易漏的就是这个。扇出的三块技术积木积木一Send的payload是局部视图。注意translate的入参不是全量状态而是Send时给的{lang: ..., src: ...}小字典。这有两层含义Worker拿到的是够用就好的最小输入省token、防越权Worker的返回值才会被Reducer合并回主状态。入参走Send专线回程走Reducer大巴——理解了这句话Send API你就毕业了。积木二扇入必须靠Reducer。translations标注了Annotated[list, operator.add]三个并行Worker都往这个键追加框架自动合并。如果把Reducer去掉LangGraph会直接抛InvalidUpdateError——因为两个节点同时写一个没有合并规则的键框架不知道听谁的。这个报错每个LangGraph新手都会撞上一次撞过就懂了。积木三并行的是等待不是计算。Python的GIL决定了CPU密集型任务多线程没用但LLM调用是IO等待网络往返time.sleep和真实API调用都会释放GIL所以线程级并行就能拿到接近N倍的加速。别为了Agent并行去折腾多进程不值。什么时候不该并行并行不是银弹三种情况要刹住车场景为什么不能并行怎么办步骤间有数据依赖B需要A的输出老实串行或重新设计切分会撞供应商限流100路并发直接触发RPM限制全被429加信号量限并发如Semaphore(10)或分批扇出结果有顺序要求并行完成顺序不可控在merge节点按key重排而不是赌完成顺序第二行展开说一句这是并行化后最常见的生产事故本地测试3路并发美滋滋上线流量一上来100路扇出供应商直接限流全部超时重试雪崩。我的经验值对单一供应商并发扇出别超过你限额的1/3留出重试余量。串行 vs 并行完整对照表维度串行链Diamond并行延迟Σ各步延迟0.905smax各路延迟0.308s单路失败影响后续全部阻塞只影响该路merge兜底token成本相同相同并行省时间不省钱实现复杂度低中要处理Reducer、失败、乱序适用有依赖的推理步骤无依赖的同类任务批量处理注意省时间不省钱这行并行化优化的是时延不是成本——该烧的token一个不少。所以决策公式很简单用户在等交互场景→ 并行后台批处理无人等 → 串行更稳。当然后台任务适度并行能提高吞吐单位时间处理量那是另一个维度的收益。小结Diamond拓扑 Send扇出运行时定并行度 共同后继节点收口框架做同步屏障实测3路0.3s任务串行0.905s vs 并行0.308s加速比2.94x接近理论上限失败隔离三步法Worker内不抛异常、失败转ok:False标记、merge节点统一兜底并行省时延不省成本扇出并发要给供应商限流留余量≤限额1/3是我的经验值。下节预告并行搞定了同一件事的N个副本但现实里更多的问题是不同的事该派给谁——用户一句话进来是找售后、找技术还是找投诉专线第48节动态路由与条件边我们搭一个意图识别→专家分派的客服路由系统并对比规则路由 vs LLM路由两条技术路线的延迟、成本和可维护性。互动区动手题 讨论题把LANGS改成10种语言再跑一次加速比是多少把某一路的time.sleep改成1.0s总耗时变成多少——答案是最慢那一路跑完你就对并行耗时max有体感了你的业务里有哪些其实无依赖却在串行的步骤评论区揪出来说不定下期我就拿它当案例。觉得加速比2.94x值一个赞的别客气。 
长河君 · 2026/10/29 10:50:55 · 阅读 1.2k Graph-Engineering-第10章第47节-LangGraph实战②-Parallelization与Diamond拓扑
LangGraph实战②——Parallelization与Diamond拓扑上一节结尾留了个扣子链的软肋是延迟串行叠加四步链每步2秒用户就等8秒。今天就来解这个扣子——用一个分析→多语言并行翻译→合并的Diamond拓扑把Send API扇出、Reducer扇入、单路失败隔离这三块硬骨头一次啃完最后附上串行vs并行的实测对照数据。本文导航Diamond把队形从纵队变成雁阵完整代码控制台输出扇出的三块技术积木什么时候不该并行串行 vs 并行完整对照表Diamond把队形从纵队变成雁阵先统一概念。所谓Diamond菱形拓扑指的是这种形状一个节点展开成N个并行分支最后再汇聚到一个合并节点——展开、收口画出来就是个菱形。Send 扇出一变三Send 扇出Send 扇出失败也照样抵达merge以失败标记的形式STARTanalyze分析确定源文本与目标语言translate:en英语翻译translate:ja日语翻译translate:sw斯瓦希里语翻译 ❌故意失败merge扇入合并结果失败兜底END业务场景设定把一段产品介绍同时翻成英语、日语、斯瓦希里语对最后一个是我故意埋的雷专门用来演示一路失败不拖垮全局。有人会问为什么不写三个翻译节点、三条边三个语言可以三十个呢每个用户自定义语言组合呢静态连边把并行度焊死在图装配期而Send把并行度留给运行时——Send(translate, payload)想发几份发几份翻译节点只写一个。这就是Part 2说的动态分发边的实战价值。完整代码# translate_diamond.py —— LangGraph实战②Diamond拓扑多语言并行翻译# 依赖pip install langgraph0.4 Python 3.10fromtypingimportTypedDict,Annotatedimportoperatorimporttimefromlanggraph.graphimportStateGraph,START,ENDtry:fromlanggraph.typesimportSendexceptImportError:fromlanggraph.constantsimportSend SRCGraph Engineering 是编排多智能体的工程学节点是职责边是协作。LANGS[en,ja,sw]# sw不支持 → 演示单路失败隔离TRANSLATIONS{# 模拟LLM翻译结果真实项目换成模型调用en:Graph Engineering is the discipline of orchestrating multi-agent systems: nodes are responsibilities, edges are collaboration.,ja:Graph Engineeringとはマルチエージェントを編成する工学であるノードは責務、エッジは協働である。,}# 共享状态扇入靠Reducer classTState(TypedDict,totalFalse):src:str# 源文本langs:list# 目标语言运行时数据translations:Annotated[list[dict],operator.add]# 扇入并行结果追加合并final:dict# 合并后的交付物log:Annotated[list[str],operator.add]# 节点1个分析 1个翻译Worker 1个合并 defanalyze(state:TState):print(f[analyze ] 源文本{len(state[src])}字目标语言:{state[langs]})return{log:[analyze]}deftranslate(payload:dict):翻译Worker每个Send实例独立执行一份。注意入参是Send的payload而非全量状态langpayload[lang]time.sleep(0.3)# 模拟真实LLM延迟IO等待iflangnotinTRANSLATIONS:print(f[translate:{lang}] ❌ 不支持该语言登记失败不抛异常)return{translations:[{lang:lang,ok:False,text:None}],log:[ftranslate:{lang}:fail]}print(f[translate:{lang}] ✅ 翻译完成)return{translations:[{lang:lang,ok:True,text:TRANSLATIONS[lang]}],log:[ftranslate:{lang}:ok]}defmerge(state:TState):扇入节点合并所有并行结果失败语言回退原文交付final,failed{},[]fortinstate[translations]:final[t[lang]]t[text]ift[ok]elsef[回退原文]{state[src]}ifnott[ok]:failed.append(t[lang])print(f[merge ] 成功{len(state[translations])-len(failed)}路 f失败{len(failed)}路{( - str(failed))iffailedelse})return{final:final,log:[merge]}# 扇出Send动态分发 deffan_out(state:TState):运行时按langs列表扇出发几份、发给谁装配图时一无所知return[Send(translate,{lang:lang,src:state[src]})forlanginstate[langs]]# 装配菱形 builderStateGraph(TState)builder.add_node(analyze,analyze)builder.add_node(translate,translate)builder.add_node(merge,merge)builder.add_edge(START,analyze)builder.add_conditional_edges(analyze,fan_out)# 上半菱形扇出builder.add_edge(translate,merge)# 下半菱形扇入builder.add_edge(merge,END)graphbuilder.compile()if__name____main__:print( 串行基线循环顺序翻译 )t0time.perf_counter()forlanginLANGS:time.sleep(0.3)# 同样的模拟延迟print(f serial:{lang}done)serial_costtime.perf_counter()-t0print(f 串行耗时{serial_cost:.3f}s\n)print( Diamond并行版 )t1time.perf_counter()resultgraph.invoke({src:SRC,langs:LANGS})par_costtime.perf_counter()-t1print(f 并行耗时{par_cost:.3f}s 加速比{serial_cost/par_cost:.2f}x\n)print( 最终交付物 )forlang,textinresult[final].items():print(f[{lang}]{text[:46]}...)控制台输出python translate_diamond.py并行分支的打印顺序每次可能不同这是并行的正常现象 串行基线循环顺序翻译 serial:en done serial:ja done serial:sw done 串行耗时 0.905s Diamond并行版 [analyze ] 源文本37字目标语言: [en, ja, sw] [translate:ja] ✅ 翻译完成 [translate:en] ✅ 翻译完成 [translate:sw] ❌ 不支持该语言登记失败不抛异常 [merge ] 成功2路 失败1路 - [sw] 并行耗时 0.308s 加速比 2.94x 最终交付物 [en] Graph Engineering is the discipline of orchestr... [ja] Graph Engineeringとはマルチエージェントを編成する工学で... [sw] [回退原文] Graph Engineering 是编排多智能体的工程学节点是职责边是协作。三个观察点比代码本身更重要加速比2.94x接近理论值3x。三路延迟都是0.3s并行后总耗时≈最慢那一路0.308s而不是三路之和0.905s。翻译成生产语言3路各5秒的LLM调用用户感知从15秒变成5秒sw路失败了en和ja照常交付。失败被封装成{ok: False}随流转运在merge节点统一兜底回退原文。这就是失败隔离——一路翻车不掀桌merge只执行了一次。三个translate实例虽然并行但都被框架归入同一个超步superstepmerge等所有分支到齐才开始。这是LangGraph替你做的同步屏障手写多线程代码时最容易漏的就是这个。扇出的三块技术积木积木一Send的payload是局部视图。注意translate的入参不是全量状态而是Send时给的{lang: ..., src: ...}小字典。这有两层含义Worker拿到的是够用就好的最小输入省token、防越权Worker的返回值才会被Reducer合并回主状态。入参走Send专线回程走Reducer大巴——理解了这句话Send API你就毕业了。积木二扇入必须靠Reducer。translations标注了Annotated[list, operator.add]三个并行Worker都往这个键追加框架自动合并。如果把Reducer去掉LangGraph会直接抛InvalidUpdateError——因为两个节点同时写一个没有合并规则的键框架不知道听谁的。这个报错每个LangGraph新手都会撞上一次撞过就懂了。积木三并行的是等待不是计算。Python的GIL决定了CPU密集型任务多线程没用但LLM调用是IO等待网络往返time.sleep和真实API调用都会释放GIL所以线程级并行就能拿到接近N倍的加速。别为了Agent并行去折腾多进程不值。什么时候不该并行并行不是银弹三种情况要刹住车场景为什么不能并行怎么办步骤间有数据依赖B需要A的输出老实串行或重新设计切分会撞供应商限流100路并发直接触发RPM限制全被429加信号量限并发如Semaphore(10)或分批扇出结果有顺序要求并行完成顺序不可控在merge节点按key重排而不是赌完成顺序第二行展开说一句这是并行化后最常见的生产事故本地测试3路并发美滋滋上线流量一上来100路扇出供应商直接限流全部超时重试雪崩。我的经验值对单一供应商并发扇出别超过你限额的1/3留出重试余量。串行 vs 并行完整对照表维度串行链Diamond并行延迟Σ各步延迟0.905smax各路延迟0.308s单路失败影响后续全部阻塞只影响该路merge兜底token成本相同相同并行省时间不省钱实现复杂度低中要处理Reducer、失败、乱序适用有依赖的推理步骤无依赖的同类任务批量处理注意省时间不省钱这行并行化优化的是时延不是成本——该烧的token一个不少。所以决策公式很简单用户在等交互场景→ 并行后台批处理无人等 → 串行更稳。当然后台任务适度并行能提高吞吐单位时间处理量那是另一个维度的收益。小结Diamond拓扑 Send扇出运行时定并行度 共同后继节点收口框架做同步屏障实测3路0.3s任务串行0.905s vs 并行0.308s加速比2.94x接近理论上限失败隔离三步法Worker内不抛异常、失败转ok:False标记、merge节点统一兜底并行省时延不省成本扇出并发要给供应商限流留余量≤限额1/3是我的经验值。下节预告并行搞定了同一件事的N个副本但现实里更多的问题是不同的事该派给谁——用户一句话进来是找售后、找技术还是找投诉专线第48节动态路由与条件边我们搭一个意图识别→专家分派的客服路由系统并对比规则路由 vs LLM路由两条技术路线的延迟、成本和可维护性。互动区动手题 讨论题把LANGS改成10种语言再跑一次加速比是多少把某一路的time.sleep改成1.0s总耗时变成多少——答案是最慢那一路跑完你就对并行耗时max有体感了你的业务里有哪些其实无依赖却在串行的步骤评论区揪出来说不定下期我就拿它当案例。觉得加速比2.94x值一个赞的别客气。