协调者 - 工作者模式Orchestrator-Workers概念好我们接下来继续来看第4种工作模式。第4种工作模式呢它叫协调者工作者模式。什么是协调者和工作者模式呢跟大家讲解这个模式我们需要结合实际当中的例子方便大家更好的去理解。首先我们要清楚协调者工作者模式本身就是为了处理复杂任务而诞生的一种模式。什么叫做复杂任务举个例子现在我们要完成一项大的复杂任务需要把这个整体任务进行拆解拆分出A 任务、B 任务、C 任务、D 任务等一系列子任务把所有拆分出来的子任务全部完成之后整体的复杂大任务也就顺利完成了。那在这个流程里面协调者充当什么样的角色协调者的核心作用就是拆分复杂任务。比如协调者把一项大任务拆分成三个子任务而工作者就是负责执行这些子任务的执行角色。简单来说协调者拆分完任务之后就会把第一个子任务分配给第一位工作者第二个子任务分配给第二位工作者第三个子任务分配给第三位工作者完成任务派发。等到所有工作者把自己接手的子任务全部处理完毕整体的复杂任务就算全部完成。有不少同志可能会疑惑协调者工作者模式和我们之前学习的并行化模式看起来十分相似接下来我们就把两种模式放在一起清晰区分二者的差异。先回顾并行化模式并行化模式流程很简单由开始节点发起直接同时调动多个节点执行对应任务比如市场分析、竞品分析、技术分析所有任务同步执行最后统一进入汇总报告环节收尾本质就是固定的任务一、任务二、任务三同步运行。而我们现在学习的协调者工作者模式同样也是由工作者去执行一个个独立子任务表面上看起来同样是多任务同步处理那二者真正的区别到底在哪里我们拆分概念就能一目了然并行化模式只有工作执行这一部分从搭建工作流图表的一开始我们就明确知道需要处理哪些具体任务直接提前把固定任务分配给对应节点执行最后做结果汇总即可。但是协调者工作者模式比并行化模式多了一个核心角色协调者这也是二者最核心的区别。这里有一个重中之重的知识点协调者在流程运行之前提前并不知道最终要拆分出多少个子任务。任务数量只有在工作流正式运行阶段接收到用户输入内容、流程正式启动之后协调者才会现场完成任务拆分确定最终的子任务数量。反观并行化模式在我们搭建流程图、编写代码构建工作流的阶段就已经提前确定好了所有要执行的任务。我们换一个生活化的企业工作场景能让大家理解得更加透彻。我们把场景设定成一家互联网公司公司内部储备了各类岗位人员也就是人员备选池里面包含前端开发、后端开发、测试人员、运维人员、产品人员、项目经理、项目对接负责人等等每一个人员都具备对应的专属工作能力。日常工作当中公司会接到不同类型的项目需求任务第一种任务需求老板下达指令要求团队开发一款全新的APP 软件。这个任务最先交由项目经理接手项目经理就相当于我们工作流当中的协调者。项目经理接到开发 APP 这个整体大任务之后不会直接开工而是先对整体项目进行拆解拆分比如拆分成页面设计、前端开发、后端接口开发、功能测试等多个子任务。拆分完成之后作为协调者的项目经理就会从公司的人员备选池当中挑选匹配对应工作能力的人员承接子任务把页面设计任务交给产品人员开发任务交给开发人员测试任务交给测试人员。而人员备选池当中没有被选中的工作人员暂时处于待命状态不会参与本次项目流程。整个人员挑选、任务派发的流程全部都是在接到项目任务、项目正式启动运行之后才确定的并不是提前就定好谁负责什么工作。第二种任务需求公司需要和外部其他企业完成项目商务对接。同样由协调者统筹安排结合本次任务的工作性质协调者就会直接挑选商务对接负责人、项目旁听陪同人员参与本次工作不再安排开发、设计类工作人员参与。从这两个实际场景就能总结出来协调者会根据不同的实际任务类型在流程运行的过程当中动态挑选对应的工作人员、动态划分任务执行路径。所有人员分配、任务划分、执行路径确定全部依托实时任务而定而非提前固定设定。等到所有承接子任务的工作人员完成自身工作之后所有工作成果统一汇总整合最终完成整体项目交付这就是完整的协调者工作者运行逻辑。总结核心区别并行化模式搭建工作流阶段就确定所有任务固定节点同步执行全程流程固定不变协调者工作者模式搭建工作流阶段仅搭建基础框架运行阶段由协调者动态拆分任务、动态分配工作者、动态确定执行路径灵活性极强。依托这个运行逻辑协调者工作者模式能够实现流程高度动态化只要我们提前在备选池当中录入各类功能的工作节点后续不管接到什么样的复杂任务协调者都能自主拆分任务、自主匹配对应工作节点完成执行。梳理完核心概念之后我们明确这套模式当中三大核心身份节点第一类协调者节点核心职责是接收整体大任务分析任务内容自主拆分出对应的子任务清单第二类工作者节点统一存入工作者备选池当中数量不固定可根据业务需求自由新增删减专门负责承接协调者派发的各类子任务独立完成对应工作内容第三类合成器汇总节点等到所有工作者完成全部子任务之后统一收集整合所有工作者产出的结果整理汇总成为最终完整成果完成整个工作流收尾。除此之外还有一种灵活应用场景如果多项子任务的工作内容高度一致我们不需要设置多个同类型工作者节点只需要设置单个通用工作者节点即可。协调者可以把多个同类子任务分批次依次派发至同一个工作者节点当中执行同一个工作节点重复多次运行完成多项同类任务处理进一步精简工作流结构。综上协调者 - 工作者模式可以理解为一个大脑协调者分配任务多个工人工作者执行最后合成最终结果。例如当我们要处理几百页的技术文档可以让协调者拆分文档接着安排多个工作者并行处理不同章节最终汇总结果。如下图所示协调者 - 工作者 模式和 并行化 模式都涉及同时执行多个任务但它们的核心区别在于任务分配方式并行化任务在设计时就确定所有任务同时开始。协调者 - 工作者任务在运行时由协调者动态分配。举个例子【案例文档翻译】并行化已知要翻译 3 个固定章节同时翻译协调者 - 工作者先分析文档结构发现需要翻译 5 个章节动态分配【案例数据分析】并行化同时计算平均值、最大值、最小值固定指标协调者 - 工作者先分析数据特征决定需要计算哪些统计指标模式实践理清所有概念和应用场景之后接下来我们开始动手编写代码落地实现协调者工作者模式。首先第一步进行全局状态 State 设计合理规划状态当中需要存储的所有数据任务主题也就是用户输入的整体大任务内容作为整个工作流的启动输入子任务清单由协调者拆分任务之后生成用来存储所有拆分完毕的子任务内容也可以称之为工作计划工作者成果列表专门用来存储每一个工作者完成子任务之后产出的结果这里需要采用追加写入的存储方式依次收纳所有成果方便后续统一汇总最终整合结果由汇总节点整理所有工作成果之后生成的完整最终输出内容。class State(TypedDict): topic: str sections: list # 协调者生成的计划 completed_sections: Annotated[list, operator.add] # 工作者完成的结果 final_report: str从状态设计当中我们也能看出这套模式同时还融入了提示链模式的特点前一个节点产出的数据内容会作为后一个节点运行的输入依据节点之间依靠状态数据完成联动传递。状态定义完成之后第二步开始逐个定义功能节点第一个核心节点协调者节点协调者节点的核心逻辑十分明确接收用户传入的任务主题调用大语言模型借助结构化输出的方式让大模型自主对整体任务进行拆解自动生成标准化的子任务大纲与子任务描述最终把拆分完成的所有子任务整理为任务清单更新存入全局状态当中。这里需要重点注意拆分出来的子任务数量是不固定的只有代码运行调用大模型之后我们才能确定最终拆分出多少个子任务完全贴合协调者动态拆分任务的核心逻辑。# 定义数据结构-结构化输出 # 借助结构化输出的方式让大模型自主对整体任务进行拆解自动生成标准化的子任务大纲与子任务描述 class Section(BaseModel): name: str description: str # 最终把拆分完成的所有子任务整理为任务清单更新存入全局状态当中 class Sections(BaseModel): sections: List[Section] # 创建规划器 model init_chat_model(gpt-4o-mini) planner model.with_structured_output(Sections) # 协调者节点 - 制定计划 def orchestrator(state: State): 协调者分析任务并制定执行计划 report_sections planner.invoke( f为主题{state[topic]}制定报告大纲包含3个章节 ) return {sections: report_sections.sections}第二个核心节点工作者节点我们本次实操设定统一工作者节点所有文本内容生成类的子任务全部交由这一个工作者节点执行。工作者节点运行时会接收协调者派发的单个具体子任务内容读取子任务对应的名称、工作要求、内容描述再调用大语言模型按照任务要求独立完成内容创作、内容编写等对应工作完成工作之后把自身产出的工作成果以追加的方式存入全局状态的成果列表当中。如果后续业务需要新增不同类型的工作内容我们只需要新增对应的独立工作者节点存入备选池即可拓展性极强。# 工作者节点 - 执行具体任务 def llm_call(state: State): 工作者根据分配的任务生成内容 section state[section] # 从协调者接收的任务 result model.invoke( f编写报告章节: {section.name}, 内容要求: {section.description} ) return {completed_sections: [result.content]} # 结果会自动合并第三个核心节点汇总合成器节点汇总节点不需要执行复杂逻辑只需要读取全局状态当中已经收集齐全的所有工作者成果列表按照规范的格式把零散的成果内容拼接整合梳理成为一篇完整、通顺、格式统一的最终成品内容最后更新为全局状态当中的最终结果完成整体工作流程。# 汇总节点 def synthesizer(state: State): 汇总所有工作者的成果 completed_sections state[completed_sections] final_report \n\n.join(completed_sections) return {final_report: final_report}所有功能节点全部定义完毕之后整个工作流搭建的重中之重就是配置节点与节点之间的流向边尤其是实现协调者动态派发任务的条件边逻辑。首先配置基础固定边工作流起始节点 START 直接连通协调者节点代表流程启动之后最先进入任务拆分环节所有工作者节点执行完毕之后统一连通汇总节点汇总节点处理完成之后直接连通结束节点 END。最核心的动态任务派发逻辑依靠条件边实现同时这里需要用到 LangGraph 当中一个关键特性Send 对象。Send 对象一共包含两个核心传入参数第一个参数指定需要接收任务的工作者节点名称第二个参数需要传递给对应工作者节点的专属状态数据也就是协调者拆分出来的具体单个子任务信息。在协调者完成任务拆分、生成子任务清单之后我们编写任务分配函数遍历整份子任务清单遍历过程当中为每一个子任务都单独构建一个对应的 Send 对象明确任务需要派发至哪一个工作节点同时把具体的子任务内容搭载在 Send 对象当中完成传递。和以往普通条件边固定返回单一节点名称不同本次任务分配函数最终会批量返回多个 Send 对象组成的列表。工作流识别到返回内容为 Send 对象列表之后就会在运行阶段按照每一个 Send 对象当中指定的节点和传递的任务数据自动完成动态任务派发同时支持多个任务同步派发让多个工作任务并行执行。简单来说我们在搭建工作流图表、编写基础代码的时候无法提前确定后续具体的执行路径和任务分配方式所有的路径选择、任务派发、节点调用全部都是在代码实际运行过程当中依靠 Send 对象动态生成、动态执行。同时工作者节点能够顺利接收专属子任务内容的原因也迎刃而解工作者节点当中读取到的独立子任务数据并不是从全局公共状态当中直接获取而是协调者通过 Send 对象单独定向传递过来的专属状态数据实现了任务精准下发。# 任务分配函数 - 关键部分! def assign_workers(state: State): 为每个任务创建工作者 worker_tasks [] for section in state[sections]: worker_tasks.append( Send(llm_call, {section: section}) # 发送任务给工作者 ) return worker_tasks # 构建工作流 builder StateGraph(State) builder.add_node(orchestrator, orchestrator) builder.add_node(llm_call, llm_call) builder.add_node(synthesizer, synthesizer) builder.add_edge(START, orchestrator) # 关键: 协调者后创建多个工作者 builder.add_conditional_edges( orchestrator, assign_workers, [llm_call] # 创建的工作者都指向llm_call节点 ) # 所有工作者完成后汇总 builder.add_edge(llm_call, synthesizer) builder.add_edge(synthesizer, END) worker builder.compile()完成所有节点定义、流向边配置之后整体协调者工作者模式的代码框架就全部搭建完成。我们设定实操测试场景传入任务主题中国近代史让协调者自主拆分编写文章所需的章节大纲拆分完成之后动态把每一个章节编写任务派发至工作者节点多个章节内容同步并行编写所有章节内容全部编写完成之后由汇总节点整合所有章节内容最终生成一篇完整的中国近代史完整文章。response worker.invoke({topic: 中国近代史}) print(response)代码运行过程当中我们可以打印日志清晰查看执行流程首先协调者执行任务拆分自主确定拆分章节数量随后多个编写任务同步下发工作者批量并行生成章节内容所有内容生成完毕之后汇总节点统一整合排版最终输出完整成品内容。{ topic: 中国近代史, sections: [ { name: 第一章近代史概述, description: 介绍中国近代史的背景与重要性涵盖社会、经济、文化等多个方面的变化。 }, { name: 第二章鸦片战争与列强侵华, description: 探讨鸦片战争对中国的影响以及随后的列强侵华与民族觉醒进程。 }, { name: 第三章辛亥革命与新文化运动, description: 分析辛亥革命及其后果以及新文化运动对中国现代化的推动作用。 } ], completed_sections: [ # 第一章: ..., # 第二章: ..., # 第三章: ... ], final_report: # 第一章: ...\n\n# 第二章: ...\n\n# 第三章: ... }通过实际代码运行测试我们能够直观发现协调者拆分出来的子任务数量随机不固定完全由大模型结合任务主题自主判定完美实现了运行阶段动态拆分任务、动态分配工作的核心特性。到这里协调者工作者模式的核心概念、和并行化模式的核心差异、实际应用场景以及完整代码实操逻辑就全部讲解完毕了。大家一定要牢牢记住这套模式的核心精髓依靠协调者实现运行时动态拆分任务、动态分配执行节点摆脱固定流程的限制适配各类复杂多变的大型任务处理场景这也是它和并行化模式最本质的区别。对于我们要实现的案例其完整代码如下from langchain.chat_models import init_chat_model from langgraph.constants import START, END from langgraph.graph import StateGraph from langgraph.types import Send from typing import Annotated, TypedDict, List import operator from pydantic import BaseModel class State(TypedDict): topic: str sections: list # 协调者生成的计划 completed_sections: Annotated[list, operator.add] # 工作者完成的结果 final_report: str # 定义数据结构-结构化输出 class Section(BaseModel): name: str description: str class Sections(BaseModel): sections: List[Section] # 创建规划器 model init_chat_model(gpt-4o-mini) planner model.with_structured_output(Sections) # 协调者节点 - 制定计划 def orchestrator(state: State): 协调者分析任务并制定执行计划 report_sections planner.invoke( f为主题{state[topic]}制定报告大纲包含3个章节 ) return {sections: report_sections.sections} # 工作者节点 - 执行具体任务 def llm_call(state: State): 工作者根据分配的任务生成内容 section state[section] # 从协调者接收的任务 result model.invoke( f编写报告章节: {section.name}, 内容要求: {section.description} ) return {completed_sections: [result.content]} # 结果会自动合并 # 汇总节点 def synthesizer(state: State): 汇总所有工作者的成果 completed_sections state[completed_sections] final_report \n\n.join(completed_sections) return {final_report: final_report} # 任务分配函数 - 关键部分! def assign_workers(state: State): 为每个任务创建工作者 worker_tasks [] for section in state[sections]: worker_tasks.append( Send(llm_call, {section: section}) # 发送任务给工作者 ) return worker_tasks # 构建工作流 builder StateGraph(State) builder.add_node(orchestrator, orchestrator) builder.add_node(llm_call, llm_call) builder.add_node(synthesizer, synthesizer) builder.add_edge(START, orchestrator) # 关键: 协调者后创建多个工作者 builder.add_conditional_edges( orchestrator, assign_workers, [llm_call] # 创建的工作者都指向llm_call节点 ) # 所有工作者完成后汇总 builder.add_edge(llm_call, synthesizer) builder.add_edge(synthesizer, END) worker builder.compile() response worker.invoke({topic: 中国近代史}) print(response)评估器 - 优化器模式Evaluator-optimizer概念我们继续来讲最后一个工作流的模式也叫做评估器优化器模式。其实这个评估器和优化器的模式我们之前编写案例、搭建RAG系统的时候就已经使用过相关逻辑也早就写过了。接下来我们详细梳理评估优化器模式完整的运行流程。简单来说整套流程逻辑十分清晰首先由程序生成对应的内容内容生成完成之后交由评估器专门负责完成质量审核工作。如果审核判定内容质量合格整个流程直接结束一旦判定质量达不到标准就带着对应的问题反馈重新进入生成环节再次产出内容。新一轮内容生成完毕后依旧再次走质量评估流程依旧遵循合格结束、不合格重写的规则以此不断循环往复。所以评估器优化器模式的核心逻辑就是对不达标的内容进行迭代重写、持续优化调整一直到最终产出的结果完全满足预设的质量要求为止这就是完整的评估优化器模式。结合我们之前做过的RAG 检索问答案例就能快速对应上这套逻辑在系统当中我们先检索获取相关文档内容紧接着对检索出来的文档结果进行质量审核判断检索到的内容能不能匹配上用户提出的问题、能不能支撑作答。审核合格就直接进入生成答案的环节如果审核判定检索内容不符合需求系统就会执行重写问题的策略优化用户原始提问之后再重新执行文档检索操作。我们这套RAG系统的运行链路和评估器优化器模式的运行逻辑完全一致。大家也要清楚评估器优化器模式不局限于固定两个节点的组合形式我们可以根据实际业务需求自由拓展LangGraph流程图。只要整体流程能够实现内容生成 质量评估这两大核心动作满足内容不达标就携带反馈重新生成、反复循环校验优化的逻辑都可以归类为评估器优化器模式。直白划分两个核心角色评估器的核心作用就是做质量检测、结果判定优化器对应的就是二次内容生成环节并且生成过程中可以带入上一轮评估给出的问题反馈针对性调整优化内容。因为这套模式对应的实操代码我们之前已经完整实现过这里就不再重复编写代码。我们重点来讲它在实际项目和日常应用当中的落地场景。第一个常用场景就是内容质量优化、代码生成与迭代改进。比如我们调用大语言模型自动编写程序代码代码生成完成之后我们可以运行测试用例、校验代码语法规范、排查逻辑漏洞等多种方式完成质量评估。如果评估发现代码存在问题无法正常运行就整理好具体的问题反馈信息交给模型重新编写代码。这里有一个关键要点二次生成内容时一定要携带评估反馈明确告知上一轮内容存在的缺陷这样新一轮生成才能规避同类问题优化效果才会更好。除了代码优化之外还有十分生活化的落地场景比如用户信息完整度收集场景。举个例子用户前往银行办理银行卡业务办理业务需要收集齐全身份证信息、手机号等各类必备资料我们可以把这套业务流程套入评估器优化器模式当中。我们可以设置两个核心流程节点第一个节点负责统一收集用户提交的各类资料信息第二个节点充当评估检测节点专门校验收集到的信息是否齐全合规。举个实际场景举例用户只提供了手机号遗漏了身份证证件信息。流程运行时信息收集节点录入用户现有资料经过评估检测节点判定后发现缺失核心身份证信息此时流程不会直接结束而是给出明确的缺失反馈提醒用户补齐对应的资料。用户按照反馈补齐身份证信息之后再次进入信息核验环节所有资料全部齐全核验通过整个业务流程才算正式结束。在这个场景里评估检测给出的缺失提示就是迭代优化的核心反馈依靠反馈完成信息补全完美契合评估优化的循环逻辑。除此之外我们还要拓展一个重要知识点流程里的质量评估环节实现方式并不单一。一方面我们可以依靠代码逻辑、大模型自主判定完成自动化质量检测比如让大模型自主排查代码漏洞、判断文本内容合理性另一方面LangGraph也支持人工介入评估的模式把质量审核的权限交给使用者。机器初次生成内容之后由人工手动判定内容是否合格人工给出不通过的评价以及具体修改意见程序再携带人工反馈重新生成内容两种评估方式都能适配评估器优化器的整体架构。总而言之不管是自动化智能评估还是人工手动参与评估只要整体业务流程遵循生成内容 — 评估校验 — 不合格携反馈重生成 — 合格终止流程这套循环逻辑就都属于评估器优化器模式的应用范畴。综上评估器 - 优化器模式的流程是先进行生成内容再进行评估质量如果需要改进就重新生成不断改进直到满足质量标准经典的场景就是内容质量优化、代码生成和改进等。评估器 - 优化器流程图如下所示模式实践对于评估器 - 优化器模式快速上手篇章的【案例 3 - 智能的文档问答系统】中实际上已经实现过了该案例能够根据检索结果的质量动态调整查询策略。逻辑如下图所示评估器 - 优化器的核心在于质量检测相关代码如下def grade_documents(state: MessagesState): 评估检索结果质量 # 评估逻辑... if score yes: return generate_answer # 质量达标生成答案 else: return rewrite_question # 质量不够优化问题重新检索值得再说的是除了可以让 LLM 当作评估器在 LangGraph 中还支持人机交互式的评估即将质量评估由人工来完成。这部分能力敬请期待到这里为止我们日常开发当中最常用的五种LangGraph工作流设计模式就全部讲解完毕了。大家掌握这五类模式之后再回头翻看我们前期做的各类实战案例就能发现之前编写的案例早就已经融合使用了这些经典设计思路。最后给大家做一个简单的场景选型总结方便大家后续开发直接套用如果需要并行执行多项任务并且所有任务在搭建流程之初就已经确定好直接选用并行化模式如果需要多任务并行处理但任务数量、任务具体内容无法提前确定需要流程运行之后动态拆分分配就选用协调者工作者模式日常最简单直白的线性流程按照固定顺序一步步执行前一步输出作为后一步输入直接使用提示链模式需要按照用户输入内容自动分流、划分不同业务执行链路选用路由模式需要对产出内容反复校验、迭代优化、循环调整直至达标直接使用评估器优化器模式。后续我们在实战开发、编写业务代码的过程中遇到对应的业务场景我们还会再次对应回顾对应的工作流模式加深大家的理解与使用熟练度。