Graph-Engineering-第10章第49节-LangGraph实战④-Orchestrator-Worker模式

LangGraph实战④——Orchestrator-Worker模式

第47节的Diamond把"固定3路"并行化了，第48节的路由器解决了"五选一"。但还有一类更野的需求：任务清单本身就是动态的——“研究一下LangGraph的生产落地”，拆几个子课题？每个子课题派几个Worker？规划完成之前，鬼才知道。这一节用Orchestrator-Worker模式做一个完整的研究报告生成系统，顺便解锁一个新武器：自定义Reducer。

本文导航

• 从"派活"到"排兵布阵"
• 完整代码
• 控制台输出
• 本节新武器：自定义Reducer
• Orchestrator的设计要点：拆多细？
• 结果收集的三个细节

从"派活"到"排兵布阵"

先把三个模式的关系摆清楚，不然容易混：

Orchestrator-Worker的核心分工：

• Orchestrator（编排者）：拿到目标，先"排兵布阵"——把大任务拆成子任务清单。拆几个、拆成什么样，取决于目标内容，是运行时决策；
• Worker（干活的）：每人认领一个子任务，独立完成，交回结果；
• Collector（收结果的）：等Worker全部归来，质检、汇总、组装成品。

这个模式为什么重要？因为它是"Manager-员工"这个人类组织最成熟协作方式的图拓扑化。Anthropic和OpenAI的官方多智能体构建指南里都不约而同把它列为生产首选——理由很实在：任务分解质量可控（Orchestrator专注规划）、执行可并行（Worker互不依赖）、失败可定位（哪个子任务挂了一目了然）。

我们的案例：研究报告生成系统。输入一个主题，输出一份四章报告，其中一章质量不达标会被自动补强。

看仔细：Worker数量在装配图时是未知的，plan输出几个section，Send就扇出几份；collect→polish→collect的环负责质量收敛。这就是"动态任务分配+结果收集"的完整闭环。

完整代码

# research_system.py —— LangGraph实战④：Orchestrator-Worker研究报告生成系统# 依赖：pip install "langgraph>=0.4" Python 3.10+fromtypingimportTypedDict,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,total=False):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做任务分解"""topic=state["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":[f"plan:{len(sections)}节"]}deffan_out(state:ReportState):"""动态分派：sections是运行时数据，几份Send完全由规划结果决定"""return[Send("research",{"section":s,"topic":state["topic"]})forsinstate["sections"]]# ========== Worker：独立研究一个子课题 ==========defresearch(payload:dict):sec,topic=payload["section"],payload["topic"]time.sleep(0.2)# 模拟：检索→阅读→写作（IO等待）score=0.55ifsec=="成本与风险"else0.90# 故意留一个薄弱章节draft=f"【{sec}】围绕'{topic}'的{sec}分析：3个数据点、2个案例、1个风险清单。"print(f"[research] '{sec}' 完稿 score={score:.2f}")return{"reports":{sec:{"draft":draft,"score":score}},"log":[f"research:{sec[:6]}"]}# ========== Collector：质检 + 决定是否收敛 ==========defcollect(state:ReportState):weak=[sfors,rinstate["reports"].items()ifr["score"]<0.8]r=state.get("round",0)+1print(f"[collect ] 第{r}轮质检：{len(state['reports'])}节完稿，"f"薄弱={weakifweakelse'无'}")return{"weak":weak,"round":r,"log":[f"collect:r{r}"]}defpolish(state:ReportState):"""补强薄弱章节：自定义Reducer按key覆盖，精准更新而非追加"""updates={}forsecinstate["weak"]:old=state["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"]]report=f"#{state['topic']}研究报告\n\n"+"\n\n".join(parts)avg=sum(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:# 有薄弱且有轮次预算 → 继续打磨return"polish"return"assemble"# ========== 装配 ==========builder=StateGraph(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)graph=builder.compile()if__name__=="__main__":importtimeas_t t0=_t.perf_counter()result=graph.invoke({"topic":"LangGraph"},config={"recursion_limit":50})cost=_t.perf_counter()-t0print(f"\n>>> 全程耗时{cost:.2f}s（4个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并发模型' 完稿 score=0.90 [research] '框架定位与核心抽象' 完稿 score=0.90 [research] '生产落地案例' 完稿 score=0.90 [research] '成本与风险' 完稿 score=0.55 [collect ] 第1轮质检：4节完稿，薄弱=['成本与风险'] [polish ] '成本与风险' 0.55 → 0.95 [collect ] 第2轮质检：4节完稿，薄弱=无 [assemble] 报告完成：4节 平均分0.94 全文287字 >>> 全程耗时 0.41s（4个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看着平淡，藏着三个生产细节：

1. 质检要有量化标准。score < 0.8这个阈值来自哪？不能拍脑袋。真实做法：用另一个LLM按rubric打分（完整性/准确性/相关性各占权重），阈值通过人工抽检校准——我在项目里的做法是每周抽20份报告人工复核，错分率超5%就调阈值或换打分模型；
2. 轮次预算防打磨失控。round < 3和第43节的重试预算同源：质量环可能永远收敛不了（模型能力天花板），必须给"差不多就收"的出口。收尾时如果还有薄弱章节，标记给人工，而不是无限循环；
3. 按规划顺序组装，不赌完成顺序。assemble遍历的是state["sections"]而不是reports的key——并行Worker的完成顺序是随机的，dict插入顺序也就不可靠，成品的章节顺序必须回到规划者的意图上来。这是第47节"结果有顺序要求时在merge重排"的完整落地。

小结

1. Orchestrator-Worker = 动态规划（plan拆任务）+ 动态分派（Send按需扇出）+ 质量收敛环（collect→polish→collect）；
2. 自定义Reducer是业务合并语义的声明：追加（append）适合互异结果，按键覆盖（upsert）适合同key新版本；
3. 拆分粒度经验值：一个Worker=一个可独立阅读的章节，单Worker 30_{90秒，总数3}10个，再多就上两级编排；
4. 编排用强模型、执行用性价比模型，是第一省钱杠杆；质检阈值要人工抽检校准，质量环必配轮次预算。

下节预告

五连实战的收官之战：Evaluator-Optimizer与完整项目。把前四节的链、并行、路由、编排全部装进一个系统——"写→评→改→再评"的收敛循环驱动质量迭代，配上正经的logging日志和性能分析。那是第10章的期末考试，也是你从"会拼拓扑"到"能交付项目"的毕业证。

互动区

1. 动手题：给plan的主题字典加一个你熟悉的领域（比如"供应链优化"），自定义4个子课题，跑通整个"规划→并行→质检→补强→组装"流程，评论区贴你的执行链；
2. 思考题：如果researchWorker有5%概率直接抛异常（比如检索超时），你会怎么改？提示：回顾第47节的失败隔离三步法。

追到这里的同学，你们已经完成80%了，最后一节别掉队。