LangGraph实战②——Parallelization与Diamond拓扑

上一节结尾留了个扣子：链的软肋是延迟串行叠加，四步链每步2秒用户就等8秒。今天就来解这个扣子——用一个"分析→多语言并行翻译→合并"的Diamond拓扑，把Send API扇出、Reducer扇入、单路失败隔离这三块硬骨头一次啃完，最后附上串行vs并行的实测对照数据。

本文导航

• Diamond：把"队形"从纵队变成雁阵
• 完整代码
• 控制台输出
• 扇出的三块技术积木
• 什么时候不该并行
• 串行 vs 并行：完整对照表

Diamond：把"队形"从纵队变成雁阵

先统一概念。所谓Diamond（菱形）拓扑，指的是这种形状：一个节点展开成N个并行分支，最后再汇聚到一个合并节点——展开、收口，画出来就是个菱形。

业务场景设定：把一段产品介绍同时翻成英语、日语、斯瓦希里语（对，最后一个是我故意埋的雷，专门用来演示"一路失败不拖垮全局"）。

有人会问：为什么不写三个翻译节点、三条边？三个语言可以，三十个呢？每个用户自定义语言组合呢？静态连边把并行度焊死在图装配期，而Send把并行度留给运行时——Send("translate", payload)想发几份发几份，翻译节点只写一个。这就是Part 2说的"动态分发边"的实战价值。

完整代码

# translate_diamond.py —— LangGraph实战②：Diamond拓扑多语言并行翻译# 依赖：pip install "langgraph>=0.4" Python 3.10+fromtypingimportTypedDict,Annotatedimportoperatorimporttimefromlanggraph.graphimportStateGraph,START,ENDtry:fromlanggraph.typesimportSendexceptImportError:fromlanggraph.constantsimportSend SRC="Graph 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,total=False):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而非全量状态"""lang=payload["lang"]time.sleep(0.3)# 模拟真实LLM延迟（IO等待）iflangnotinTRANSLATIONS:print(f"[translate:{lang}] ❌ 不支持该语言，登记失败（不抛异常）")return{"translations":[{"lang":lang,"ok":False,"text":None}],"log":[f"translate:{lang}:fail"]}print(f"[translate:{lang}] ✅ 翻译完成")return{"translations":[{"lang":lang,"ok":True,"text":TRANSLATIONS[lang]}],"log":[f"translate:{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"]]# ========== 装配：菱形 ==========builder=StateGraph(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)graph=builder.compile()if__name__=="__main__":print("===== 串行基线：循环顺序翻译 =====")t0=time.perf_counter()forlanginLANGS:time.sleep(0.3)# 同样的模拟延迟print(f" serial:{lang}done")serial_cost=time.perf_counter()-t0print(f">>> 串行耗时{serial_cost:.3f}s\n")print("===== Diamond并行版 =====")t1=time.perf_counter()result=graph.invoke({"src":SRC,"langs":LANGS})par_cost=time.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 是编排多智能体的工程学：节点是职责，边是协作。

三个观察点，比代码本身更重要：

1. 加速比2.94x，接近理论值3x。三路延迟都是0.3s，并行后总耗时≈最慢那一路（0.308s）而不是三路之和（0.905s）。翻译成生产语言：3路各5秒的LLM调用，用户感知从15秒变成5秒；
2. sw路失败了，en和ja照常交付。失败被封装成{"ok": False}随流转运，在merge节点统一兜底（回退原文）。这就是"失败隔离"——一路翻车不掀桌；
3. merge只执行了一次。三个translate实例虽然并行，但都被框架归入同一个超步（superstep），merge等所有分支到齐才开始。这是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并行去折腾多进程，不值。

什么时候不该并行

并行不是银弹，三种情况要刹住车：

第二行展开说一句，这是并行化后最常见的生产事故：本地测试3路并发美滋滋，上线流量一上来100路扇出，供应商直接限流，全部超时重试，雪崩。我的经验值：对单一供应商，并发扇出别超过你限额的1/3，留出重试余量。

串行 vs 并行：完整对照表

注意"省时间不省钱"这行：并行化优化的是时延，不是成本——该烧的token一个不少。所以决策公式很简单：用户在等（交互场景）→ 并行；后台批处理无人等 → 串行更稳。当然，后台任务适度并行能提高吞吐（单位时间处理量），那是另一个维度的收益。

小结

1. Diamond拓扑 = Send扇出（运行时定并行度）+ 共同后继节点收口（框架做同步屏障）；
2. 实测：3路0.3s任务，串行0.905s vs 并行0.308s，加速比2.94x，接近理论上限；
3. 失败隔离三步法：Worker内不抛异常、失败转ok:False标记、merge节点统一兜底；
4. 并行省时延不省成本；扇出并发要给供应商限流留余量（≤限额1/3是我的经验值）。

下节预告

并行搞定了"同一件事的N个副本"，但现实里更多的问题是"不同的事该派给谁"——用户一句话进来，是找售后、找技术还是找投诉专线？第48节：动态路由与条件边，我们搭一个意图识别→专家分派的客服路由系统，并对比"规则路由 vs LLM路由"两条技术路线的延迟、成本和可维护性。

互动区

动手题 + 讨论题：

1. 把LANGS改成10种语言再跑一次，加速比是多少？把某一路的time.sleep改成1.0s，总耗时变成多少？——答案是"最慢那一路"，跑完你就对"并行耗时=max"有体感了；
2. 你的业务里有哪些"其实无依赖却在串行"的步骤？评论区揪出来，说不定下期我就拿它当案例。

觉得加速比2.94x值一个赞的，别客气。