LangGraph实战④——Orchestrator-Worker模式第47节的Diamond把固定3路并行化了第48节的路由器解决了五选一。但还有一类更野的需求任务清单本身就是动态的——“研究一下LangGraph的生产落地”拆几个子课题每个子课题派几个Worker规划完成之前鬼才知道。这一节用Orchestrator-Worker模式做一个完整的研究报告生成系统顺便解锁一个新武器自定义Reducer。本文导航从派活到排兵布阵完整代码控制台输出本节新武器自定义ReducerOrchestrator的设计要点拆多细结果收集的三个细节从派活到排兵布阵先把三个模式的关系摆清楚不然容易混模式分派结构并行度典型问题Diamond第47节固定扇出N路编排期确定“翻成这5种语言”Router第48节多选一1“这句话该谁接”Orchestrator-Worker本节动态规划动态分派运行时确定“研究这个主题”Orchestrator-Worker的核心分工Orchestrator编排者拿到目标先排兵布阵——把大任务拆成子任务清单。拆几个、拆成什么样取决于目标内容是运行时决策Worker干活的每人认领一个子任务独立完成交回结果Collector收结果的等Worker全部归来质检、汇总、组装成品。这个模式为什么重要因为它是Manager-员工这个人类组织最成熟协作方式的图拓扑化。Anthropic和OpenAI的官方多智能体构建指南里都不约而同把它列为生产首选——理由很实在任务分解质量可控Orchestrator专注规划、执行可并行Worker互不依赖、失败可定位哪个子任务挂了一目了然。我们的案例研究报告生成系统。输入一个主题输出一份四章报告其中一章质量不达标会被自动补强。Send扇出几份看规划结果Send扇出有薄弱且有轮次预算环形回流全部达标STARTplanOrchestrator动态拆解子课题research:子课题1research:子课题2…按sections数量动态展开…collect质检找出薄弱章节polish补强薄弱章节assemble按规划顺序组装报告END看仔细Worker数量在装配图时是未知的plan输出几个sectionSend就扇出几份collect→polish→collect的环负责质量收敛。这就是动态任务分配结果收集的完整闭环。完整代码# research_system.py —— LangGraph实战④Orchestrator-Worker研究报告生成系统# 依赖pip install langgraph0.4 Python 3.10fromtypingimportTypedDict,Annotatedimportoperatorimporttimefromlanggraph.graphimportStateGraph,START,ENDtry:fromlanggraph.typesimportSendexceptImportError:fromlanggraph.constantsimportSend# 自定义Reducer按章节名合并后写覆盖defmerge_by_section(old:dict,new:dict)-dict:Worker并行交稿、polish补强都要更新同一章节 —— 按key覆盖比追加更合理return{**old,**new}classReportState(TypedDict,totalFalse):topic:str# 研究主题输入sections:list# Orchestrator动态拆出的子课题reports:Annotated[dict,merge_by_section]# 章节 - {draft, score}weak:list# 质检揪出的薄弱章节round:int# 质检轮次防死循环report:str# 最终报告log:Annotated[list[str],operator.add]# Orchestrator动态规划 defplan(state:ReportState):编排者按主题内容决定拆几个子课题。真实项目LLM做任务分解topicstate[topic]sections{LangGraph:[框架定位与核心抽象,Send API并发模型,生产落地案例,成本与风险],}.get(topic,[行业现状,技术方案对比,成本与风险,落地路线图])print(f[plan ] 主题{topic} → 拆出{len(sections)}个子课题)fori,sinenumerate(sections,1):print(f{i}.{s})return{sections:sections,log:[fplan:{len(sections)}节]}deffan_out(state:ReportState):动态分派sections是运行时数据几份Send完全由规划结果决定return[Send(research,{section:s,topic:state[topic]})forsinstate[sections]]# Worker独立研究一个子课题 defresearch(payload:dict):sec,topicpayload[section],payload[topic]time.sleep(0.2)# 模拟检索→阅读→写作IO等待score0.55ifsec成本与风险else0.90# 故意留一个薄弱章节draftf【{sec}】围绕{topic}的{sec}分析3个数据点、2个案例、1个风险清单。print(f[research] {sec} 完稿 score{score:.2f})return{reports:{sec:{draft:draft,score:score}},log:[fresearch:{sec[:6]}]}# Collector质检 决定是否收敛 defcollect(state:ReportState):weak[sfors,rinstate[reports].items()ifr[score]0.8]rstate.get(round,0)1print(f[collect ] 第{r}轮质检{len(state[reports])}节完稿f薄弱{weakifweakelse无})return{weak:weak,round:r,log:[fcollect:r{r}]}defpolish(state:ReportState):补强薄弱章节自定义Reducer按key覆盖精准更新而非追加updates{}forsecinstate[weak]:oldstate[reports][sec]updates[sec]{draft:old[draft]补强量化成本测算表3条缓解措施,score:min(0.99,old[score]0.40)}print(f[polish ] {sec}{old[score]:.2f}→{updates[sec][score]:.2f})return{reports:updates,log:[polish]}defassemble(state:ReportState):组装按规划顺序sections重排不赌Worker完成顺序parts[state[reports][s][draft]forsinstate[sections]]reportf#{state[topic]}研究报告\n\n\n\n.join(parts)avgsum(r[score]forrinstate[reports].values())/len(state[reports])print(f[assemble] 报告完成{len(parts)}节 平均分{avg:.2f}全文{len(report)}字)return{report:report,log:[assemble]}defafter_collect(state:ReportState):ifstate[weak]andstate.get(round,0)3:# 有薄弱且有轮次预算 → 继续打磨returnpolishreturnassemble# 装配 builderStateGraph(ReportState)builder.add_node(plan,plan)builder.add_node(research,research)builder.add_node(collect,collect)builder.add_node(polish,polish)builder.add_node(assemble,assemble)builder.add_edge(START,plan)builder.add_conditional_edges(plan,fan_out)# 动态扇出几份运行时定builder.add_edge(research,collect)# Worker全部归来后进质检builder.add_conditional_edges(collect,after_collect)builder.add_edge(polish,collect)# 质量环补强后重新质检builder.add_edge(assemble,END)graphbuilder.compile()if__name____main__:importtimeas_t t0_t.perf_counter()resultgraph.invoke({topic:LangGraph},config{recursion_limit:50})cost_t.perf_counter()-t0print(f\n 全程耗时{cost:.2f}s4个Worker并行串行约需{cost*4:.2f}s)print(f 执行链:{ - .join(result[log])})控制台输出python research_system.py并行Worker的打印顺序可能不同[plan ] 主题LangGraph → 拆出4个子课题 1. 框架定位与核心抽象 2. Send API并发模型 3. 生产落地案例 4. 成本与风险 [research] Send API并发模型 完稿 score0.90 [research] 框架定位与核心抽象 完稿 score0.90 [research] 生产落地案例 完稿 score0.90 [research] 成本与风险 完稿 score0.55 [collect ] 第1轮质检4节完稿薄弱[成本与风险] [polish ] 成本与风险 0.55 → 0.95 [collect ] 第2轮质检4节完稿薄弱无 [assemble] 报告完成4节 平均分0.94 全文287字 全程耗时 0.41s4个Worker并行串行约需1.64s 执行链: plan:4节 - research:框架定位与 - research:Send A - research:生产落地 - research:成本与风 - collect:r1 - polish - collect:r2 - assemble注意耗时那行4个Worker各需0.2s全程只花了0.41s——并行执行 一轮补强环串行做同样的事至少1.64s起步。更重要的是执行链呈现的完整故事规划→四路并行→质检发现短板→补强→复检通过→组装。这就是一张会自我修正的生产线。本节新武器自定义Reducer第47节我们用的是operator.add列表追加本节的reports换成了自定义的merge_by_section。为什么要换因为两个场景的合并语义不同翻译场景每路产出互不相同的新结果 → 追加append研究场景Worker初稿和polish补强稿是同一章节的新旧版本→ 覆盖upsert。如果这里还用追加reports会变成同一章节多版本共存的浆糊assemble还得自己挑最新版。自定义Reducer把按key覆盖下沉为状态层的合并规则节点代码保持干净。设计启示Reducer不是框架给的固定配件是你业务合并语义的声明式表达——想清楚两路并行写同一个键时业务上谁说了算Reducer就是那句话的代码翻译。顺带一个坑merge_by_section里我写的是{**old, **new}新值覆盖。如果你要旧值优先比如防止补强反而降级把两个dict换个位置就行。Reducer里这种一字符的语义差别值得在code review时重点盯。Orchestrator的设计要点拆多细排兵布阵听着爽但拆多细是个真实的工程难题。拆太粗单Worker任务太重失败重试成本高拆太细编排开销和合并复杂度上升还会出现章节之间各说各话。我的经验边界一个Worker的产出 报告里一个可独立阅读的章节本节的做法单Worker耗时控制在30~90秒LLM场景超了说明该继续拆子任务数量级在3~10个。超过10个说明该做两级编排Orchestrator下面再挂Sub-Orchestrator而不是一张大平图。还有个容易被忽略的点Orchestrator自己是单点它的输出质量决定全局下限。所以生产上plan节点要用你手里最强的模型规划是智力密集型Worker可以用性价比模型执行是劳动密集型。这种编排贵、执行贱的模型混布是多智能体系统省钱的第一杠杆——第51节讲成本优化时会给出具体配比数据。结果收集的三个细节collect和assemble看着平淡藏着三个生产细节质检要有量化标准。score 0.8这个阈值来自哪不能拍脑袋。真实做法用另一个LLM按rubric打分完整性/准确性/相关性各占权重阈值通过人工抽检校准——我在项目里的做法是每周抽20份报告人工复核错分率超5%就调阈值或换打分模型轮次预算防打磨失控。round 3和第43节的重试预算同源质量环可能永远收敛不了模型能力天花板必须给差不多就收的出口。收尾时如果还有薄弱章节标记给人工而不是无限循环按规划顺序组装不赌完成顺序。assemble遍历的是state[sections]而不是reports的key——并行Worker的完成顺序是随机的dict插入顺序也就不可靠成品的章节顺序必须回到规划者的意图上来。这是第47节结果有顺序要求时在merge重排的完整落地。小结Orchestrator-Worker 动态规划plan拆任务 动态分派Send按需扇出 质量收敛环collect→polish→collect自定义Reducer是业务合并语义的声明追加append适合互异结果按键覆盖upsert适合同key新版本拆分粒度经验值一个Worker一个可独立阅读的章节单Worker 3090秒总数310个再多就上两级编排编排用强模型、执行用性价比模型是第一省钱杠杆质检阈值要人工抽检校准质量环必配轮次预算。下节预告五连实战的收官之战Evaluator-Optimizer与完整项目。把前四节的链、并行、路由、编排全部装进一个系统——写→评→改→再评的收敛循环驱动质量迭代配上正经的logging日志和性能分析。那是第10章的期末考试也是你从会拼拓扑到能交付项目的毕业证。互动区动手题给plan的主题字典加一个你熟悉的领域比如供应链优化自定义4个子课题跑通整个规划→并行→质检→补强→组装流程评论区贴你的执行链思考题如果researchWorker有5%概率直接抛异常比如检索超时你会怎么改提示回顾第47节的失败隔离三步法。追到这里的同学你们已经完成80%了最后一节别掉队。 
长河君 · 2026/10/29 10:50:55 · 阅读 1.2k Graph-Engineering-第10章第49节-LangGraph实战④-Orchestrator-Worker模式
LangGraph实战④——Orchestrator-Worker模式第47节的Diamond把固定3路并行化了第48节的路由器解决了五选一。但还有一类更野的需求任务清单本身就是动态的——“研究一下LangGraph的生产落地”拆几个子课题每个子课题派几个Worker规划完成之前鬼才知道。这一节用Orchestrator-Worker模式做一个完整的研究报告生成系统顺便解锁一个新武器自定义Reducer。本文导航从派活到排兵布阵完整代码控制台输出本节新武器自定义ReducerOrchestrator的设计要点拆多细结果收集的三个细节从派活到排兵布阵先把三个模式的关系摆清楚不然容易混模式分派结构并行度典型问题Diamond第47节固定扇出N路编排期确定“翻成这5种语言”Router第48节多选一1“这句话该谁接”Orchestrator-Worker本节动态规划动态分派运行时确定“研究这个主题”Orchestrator-Worker的核心分工Orchestrator编排者拿到目标先排兵布阵——把大任务拆成子任务清单。拆几个、拆成什么样取决于目标内容是运行时决策Worker干活的每人认领一个子任务独立完成交回结果Collector收结果的等Worker全部归来质检、汇总、组装成品。这个模式为什么重要因为它是Manager-员工这个人类组织最成熟协作方式的图拓扑化。Anthropic和OpenAI的官方多智能体构建指南里都不约而同把它列为生产首选——理由很实在任务分解质量可控Orchestrator专注规划、执行可并行Worker互不依赖、失败可定位哪个子任务挂了一目了然。我们的案例研究报告生成系统。输入一个主题输出一份四章报告其中一章质量不达标会被自动补强。Send扇出几份看规划结果Send扇出有薄弱且有轮次预算环形回流全部达标STARTplanOrchestrator动态拆解子课题research:子课题1research:子课题2…按sections数量动态展开…collect质检找出薄弱章节polish补强薄弱章节assemble按规划顺序组装报告END看仔细Worker数量在装配图时是未知的plan输出几个sectionSend就扇出几份collect→polish→collect的环负责质量收敛。这就是动态任务分配结果收集的完整闭环。完整代码# research_system.py —— LangGraph实战④Orchestrator-Worker研究报告生成系统# 依赖pip install langgraph0.4 Python 3.10fromtypingimportTypedDict,Annotatedimportoperatorimporttimefromlanggraph.graphimportStateGraph,START,ENDtry:fromlanggraph.typesimportSendexceptImportError:fromlanggraph.constantsimportSend# 自定义Reducer按章节名合并后写覆盖defmerge_by_section(old:dict,new:dict)-dict:Worker并行交稿、polish补强都要更新同一章节 —— 按key覆盖比追加更合理return{**old,**new}classReportState(TypedDict,totalFalse):topic:str# 研究主题输入sections:list# Orchestrator动态拆出的子课题reports:Annotated[dict,merge_by_section]# 章节 - {draft, score}weak:list# 质检揪出的薄弱章节round:int# 质检轮次防死循环report:str# 最终报告log:Annotated[list[str],operator.add]# Orchestrator动态规划 defplan(state:ReportState):编排者按主题内容决定拆几个子课题。真实项目LLM做任务分解topicstate[topic]sections{LangGraph:[框架定位与核心抽象,Send API并发模型,生产落地案例,成本与风险],}.get(topic,[行业现状,技术方案对比,成本与风险,落地路线图])print(f[plan ] 主题{topic} → 拆出{len(sections)}个子课题)fori,sinenumerate(sections,1):print(f{i}.{s})return{sections:sections,log:[fplan:{len(sections)}节]}deffan_out(state:ReportState):动态分派sections是运行时数据几份Send完全由规划结果决定return[Send(research,{section:s,topic:state[topic]})forsinstate[sections]]# Worker独立研究一个子课题 defresearch(payload:dict):sec,topicpayload[section],payload[topic]time.sleep(0.2)# 模拟检索→阅读→写作IO等待score0.55ifsec成本与风险else0.90# 故意留一个薄弱章节draftf【{sec}】围绕{topic}的{sec}分析3个数据点、2个案例、1个风险清单。print(f[research] {sec} 完稿 score{score:.2f})return{reports:{sec:{draft:draft,score:score}},log:[fresearch:{sec[:6]}]}# Collector质检 决定是否收敛 defcollect(state:ReportState):weak[sfors,rinstate[reports].items()ifr[score]0.8]rstate.get(round,0)1print(f[collect ] 第{r}轮质检{len(state[reports])}节完稿f薄弱{weakifweakelse无})return{weak:weak,round:r,log:[fcollect:r{r}]}defpolish(state:ReportState):补强薄弱章节自定义Reducer按key覆盖精准更新而非追加updates{}forsecinstate[weak]:oldstate[reports][sec]updates[sec]{draft:old[draft]补强量化成本测算表3条缓解措施,score:min(0.99,old[score]0.40)}print(f[polish ] {sec}{old[score]:.2f}→{updates[sec][score]:.2f})return{reports:updates,log:[polish]}defassemble(state:ReportState):组装按规划顺序sections重排不赌Worker完成顺序parts[state[reports][s][draft]forsinstate[sections]]reportf#{state[topic]}研究报告\n\n\n\n.join(parts)avgsum(r[score]forrinstate[reports].values())/len(state[reports])print(f[assemble] 报告完成{len(parts)}节 平均分{avg:.2f}全文{len(report)}字)return{report:report,log:[assemble]}defafter_collect(state:ReportState):ifstate[weak]andstate.get(round,0)3:# 有薄弱且有轮次预算 → 继续打磨returnpolishreturnassemble# 装配 builderStateGraph(ReportState)builder.add_node(plan,plan)builder.add_node(research,research)builder.add_node(collect,collect)builder.add_node(polish,polish)builder.add_node(assemble,assemble)builder.add_edge(START,plan)builder.add_conditional_edges(plan,fan_out)# 动态扇出几份运行时定builder.add_edge(research,collect)# Worker全部归来后进质检builder.add_conditional_edges(collect,after_collect)builder.add_edge(polish,collect)# 质量环补强后重新质检builder.add_edge(assemble,END)graphbuilder.compile()if__name____main__:importtimeas_t t0_t.perf_counter()resultgraph.invoke({topic:LangGraph},config{recursion_limit:50})cost_t.perf_counter()-t0print(f\n 全程耗时{cost:.2f}s4个Worker并行串行约需{cost*4:.2f}s)print(f 执行链:{ - .join(result[log])})控制台输出python research_system.py并行Worker的打印顺序可能不同[plan ] 主题LangGraph → 拆出4个子课题 1. 框架定位与核心抽象 2. Send API并发模型 3. 生产落地案例 4. 成本与风险 [research] Send API并发模型 完稿 score0.90 [research] 框架定位与核心抽象 完稿 score0.90 [research] 生产落地案例 完稿 score0.90 [research] 成本与风险 完稿 score0.55 [collect ] 第1轮质检4节完稿薄弱[成本与风险] [polish ] 成本与风险 0.55 → 0.95 [collect ] 第2轮质检4节完稿薄弱无 [assemble] 报告完成4节 平均分0.94 全文287字 全程耗时 0.41s4个Worker并行串行约需1.64s 执行链: plan:4节 - research:框架定位与 - research:Send A - research:生产落地 - research:成本与风 - collect:r1 - polish - collect:r2 - assemble注意耗时那行4个Worker各需0.2s全程只花了0.41s——并行执行 一轮补强环串行做同样的事至少1.64s起步。更重要的是执行链呈现的完整故事规划→四路并行→质检发现短板→补强→复检通过→组装。这就是一张会自我修正的生产线。本节新武器自定义Reducer第47节我们用的是operator.add列表追加本节的reports换成了自定义的merge_by_section。为什么要换因为两个场景的合并语义不同翻译场景每路产出互不相同的新结果 → 追加append研究场景Worker初稿和polish补强稿是同一章节的新旧版本→ 覆盖upsert。如果这里还用追加reports会变成同一章节多版本共存的浆糊assemble还得自己挑最新版。自定义Reducer把按key覆盖下沉为状态层的合并规则节点代码保持干净。设计启示Reducer不是框架给的固定配件是你业务合并语义的声明式表达——想清楚两路并行写同一个键时业务上谁说了算Reducer就是那句话的代码翻译。顺带一个坑merge_by_section里我写的是{**old, **new}新值覆盖。如果你要旧值优先比如防止补强反而降级把两个dict换个位置就行。Reducer里这种一字符的语义差别值得在code review时重点盯。Orchestrator的设计要点拆多细排兵布阵听着爽但拆多细是个真实的工程难题。拆太粗单Worker任务太重失败重试成本高拆太细编排开销和合并复杂度上升还会出现章节之间各说各话。我的经验边界一个Worker的产出 报告里一个可独立阅读的章节本节的做法单Worker耗时控制在30~90秒LLM场景超了说明该继续拆子任务数量级在3~10个。超过10个说明该做两级编排Orchestrator下面再挂Sub-Orchestrator而不是一张大平图。还有个容易被忽略的点Orchestrator自己是单点它的输出质量决定全局下限。所以生产上plan节点要用你手里最强的模型规划是智力密集型Worker可以用性价比模型执行是劳动密集型。这种编排贵、执行贱的模型混布是多智能体系统省钱的第一杠杆——第51节讲成本优化时会给出具体配比数据。结果收集的三个细节collect和assemble看着平淡藏着三个生产细节质检要有量化标准。score 0.8这个阈值来自哪不能拍脑袋。真实做法用另一个LLM按rubric打分完整性/准确性/相关性各占权重阈值通过人工抽检校准——我在项目里的做法是每周抽20份报告人工复核错分率超5%就调阈值或换打分模型轮次预算防打磨失控。round 3和第43节的重试预算同源质量环可能永远收敛不了模型能力天花板必须给差不多就收的出口。收尾时如果还有薄弱章节标记给人工而不是无限循环按规划顺序组装不赌完成顺序。assemble遍历的是state[sections]而不是reports的key——并行Worker的完成顺序是随机的dict插入顺序也就不可靠成品的章节顺序必须回到规划者的意图上来。这是第47节结果有顺序要求时在merge重排的完整落地。小结Orchestrator-Worker 动态规划plan拆任务 动态分派Send按需扇出 质量收敛环collect→polish→collect自定义Reducer是业务合并语义的声明追加append适合互异结果按键覆盖upsert适合同key新版本拆分粒度经验值一个Worker一个可独立阅读的章节单Worker 3090秒总数310个再多就上两级编排编排用强模型、执行用性价比模型是第一省钱杠杆质检阈值要人工抽检校准质量环必配轮次预算。下节预告五连实战的收官之战Evaluator-Optimizer与完整项目。把前四节的链、并行、路由、编排全部装进一个系统——写→评→改→再评的收敛循环驱动质量迭代配上正经的logging日志和性能分析。那是第10章的期末考试也是你从会拼拓扑到能交付项目的毕业证。互动区动手题给plan的主题字典加一个你熟悉的领域比如供应链优化自定义4个子课题跑通整个规划→并行→质检→补强→组装流程评论区贴你的执行链思考题如果researchWorker有5%概率直接抛异常比如检索超时你会怎么改提示回顾第47节的失败隔离三步法。追到这里的同学你们已经完成80%了最后一节别掉队。