作者WangQiaomei版本1.02026/3/16一、核心概念解析1.1 Orchestrator-Worker 模式是什么Orchestrator-Worker编排器 - 工作者是 LangGraph 中极具灵活性的工作流模式核心逻辑是 **“先规划、再执行、最后汇总”**编排器OrchestratorAI 根据输入动态拆分任务比如写报告时自动规划章节数量而非预先固定任务列表工作者Worker按编排器的规划逐个执行子任务比如为每个章节生成内容合成器Synthesizer汇总所有子任务结果输出最终产物比如整合所有章节成完整报告1.2 与 Routing 模式的核心区别表格模式核心特点适用场景Routing任务固定如选 “写故事 / 写笑话 / 写诗” 其一仅做选择执行任务类型明确、数量固定的场景Orchestrator-Worker任务数量不固定由 AI 动态决定拆分数量报告撰写、多文件代码更新、未知数量子任务的场景1.3 实现目标输入任意报告主题如 “人工智能的发展历史”程序自动完成AI 智能规划报告章节2-3 个为每个章节生成针对性内容自动汇总所有章节输出格式规整的完整报告1.4 工作流流程图plaintext用户输入: 写一篇关于人工智能的报告 ↓ ┌─────────────────────┐ │ 编排器 (Orchestrator) │ ← AI 规划需要写3个章节 │ 拆分任务 │ └─────────────────────┘ ↓ ┌───┬───┬───┐ │ │ │ │ ↓ ↓ ↓ ↓ 章节1 章节2 章节3 ← 工作者(Worker)分别执行 │ │ │ │ └───┴───┴───┘ ↓ 汇总结果合成器流程执行链路START → orchestrator → worker → synthesizer → END二、环境准备2.1 安装依赖库执行以下命令安装所需 Python 库建议在虚拟环境中操作bash运行# 升级pip python.exe -m pip install --upgrade pip # 安装核心依赖 pip install --upgrade langchain langchain-core langchain-openai langgraph pydantic2.2 大模型初始化说明本文示例使用通用 OpenAI 兼容接口的大模型你可根据实际使用的模型如 OpenAI GPT-4、智谱 AI、通义千问等调整初始化参数核心代码框架完全通用python运行from langchain_openai import ChatOpenAI # 通用大模型初始化模板按需替换参数 llm ChatOpenAI( model你使用的模型名称, # 如gpt-4o、glm-4、qwen-plus等 api_key你的API密钥, # 替换为自己的API Key base_url模型接口地址 # 非OpenAI官方模型需填写对应base_url )三、分步实现详解步骤 1定义结构化输出模板Schema核心目的让 AI 返回固定格式的结构化数据而非自由文本方便后续程序直接解析使用。1.1 定义单个章节结构python运行from pydantic import BaseModel, Field from typing import List # 单个章节的结构定义 class Section(BaseModel): name: str Field(descriptionName for this section of the report.) # 章节名称 description: str Field(descriptionBrief overview of the main topics and concepts to be covered in this section.) # 章节描述作用强制 AI 返回的每个章节必须包含 “名称” 和 “描述” 两个字段避免格式混乱。1.2 定义多章节列表结构python运行# 多个章节的列表结构 class Sections(BaseModel): sections: List[Section] Field(descriptionSections of the report.) # 章节列表作用让 AI 返回的章节规划以列表形式呈现方便后续遍历执行。1.3 绑定结构化输出python运行# 将结构化模板绑定到LLM确保输出符合定义 planner llm.with_structured_output(Sections)为什么需要结构化非结构化输出AI 可能返回 “我建议分三章写第一章讲起源...”需手动解析字符串易出错结构化输出AI 直接返回Sections(sections[Section(name早期发展, ...), ...])可直接用for循环遍历代码更健壮步骤 2定义工作流状态StateState 是 LangGraph 的核心用于存储工作流执行过程中所有数据相当于 “全局数据容器”python运行from typing_extensions import TypedDict from typing import Annotated, List import operator class State(TypedDict): topic: str # 用户输入的报告主题 sections: List[Section] # 编排器规划的章节列表 completed_sections: Annotated[List[str], operator.add] # 已完成的章节内容累加存储 final_report: str # 最终生成的完整报告operator.add确保多轮执行中completed_sections能累加存储所有章节内容而非覆盖步骤 3实现核心节点函数3.1 编排器Orchestrator- 规划章节python运行from langchain.messages import HumanMessage, SystemMessage def orchestrator(state: State): 编排器节点根据用户输入的主题规划报告章节 :param state: 工作流状态包含topic字段 :return: 更新后的状态添加sections字段 # 调用LLM生成章节规划 report_sections planner.invoke( [ # 系统提示限定生成2-3个章节 SystemMessage(contentGenerate a plan for a report. Return a list of 2-3 sections.), # 用户输入传入报告主题 HumanMessage(contentfTopic: {state[topic]}), ] ) # 返回规划的章节列表更新状态 return {sections: report_sections.sections}3.2 工作者Worker- 生成章节内容python运行def worker(state: State): 工作者节点遍历章节列表为每个章节生成内容 :param state: 工作流状态包含sections字段 :return: 更新后的状态添加completed_sections字段 completed [] # 遍历编排器规划的每个章节 for section in state[sections]: # 为当前章节生成内容 result llm.invoke( [ SystemMessage(contentWrite a short paragraph (2-3 sentences) for the given section.), HumanMessage(contentfSection: {section.name}\nDescription: {section.description}), ] ) # 按Markdown格式存储章节内容 completed.append(f## {section.name}\n{result.content}\n) # 返回已完成的章节内容更新状态 return {completed_sections: completed}3.3 合成器Synthesizer- 汇总最终报告python运行def synthesizer(state: State): 合成器节点汇总所有章节内容生成最终报告 :param state: 工作流状态包含completed_sections字段 :return: 更新后的状态添加final_report字段 # 拼接最终报告标题所有章节 final_report f# Report: {state[topic]}\n\n final_report \n.join(state[completed_sections]) # 返回最终报告更新状态 return {final_report: final_report}步骤 4构建并编译工作流python运行from langgraph.graph import StateGraph, START, END # 1. 初始化状态图传入State定义 builder StateGraph(State) # 2. 添加节点将函数绑定为工作流节点 builder.add_node(orchestrator, orchestrator) # 编排器节点 builder.add_node(worker, worker) # 工作者节点 builder.add_node(synthesizer, synthesizer) # 合成器节点 # 3. 添加执行边定义节点执行顺序 builder.add_edge(START, orchestrator) # 开始 → 编排器 builder.add_edge(orchestrator, worker) # 编排器 → 工作者 builder.add_edge(worker, synthesizer) # 工作者 → 合成器 builder.add_edge(synthesizer, END) # 合成器 → 结束 # 4. 编译工作流生成可执行的chain workflow builder.compile()步骤 5执行工作流并输出结果python运行def safe_print(text): 兼容GBK编码的打印函数避免特殊字符导致的打印错误 :param text: 要打印的文本 print(text.encode(gbk, errorsignore).decode(gbk), end, flushTrue) if __name__ __main__: # 执行工作流传入初始状态主题空的已完成章节列表 state workflow.invoke({topic: 人工智能的发展历史, completed_sections: []}) # 输出最终报告 safe_print(state[final_report])四、完整可运行源码python运行# -*- coding: utf-8 -*- langchain_test6_orchestrator-worker ~~~~~~~~~~~~ :copyright: (c) 2026 :authors: WangQiaomei :version: 1.0 of 2026/3/16 import warnings # 1. 导入核心库 from typing import Annotated, List import operator # 2. 定义结构化输出模板 class Section(BaseModel): name: str Field(descriptionName for this section of the report.) description: str Field(descriptionBrief overview of the main topics and concepts to be covered in this section.) class Sections(BaseModel): sections: List[Section] Field(descriptionSections of the report.) # 绑定结构化输出 planner llm.with_structured_output(Sections) # 3. 定义工作流状态 class State(TypedDict): topic: str # 用户输入的主题 sections: List[Section] # 编排器规划的章节列表 completed_sections: Annotated[List[str], operator.add] # 已完成的章节内容 final_report: str # 最终报告 # 4. 实现节点函数 def orchestrator(state: State): 编排器分析主题规划报告章节 report_sections planner.invoke( [ SystemMessage(contentGenerate a plan for a report. Return a list of 2-3 sections.), HumanMessage(contentfTopic: {state[topic]}), ] ) return {sections: report_sections.sections} def worker(state: State): 工作者为每个章节生成内容 completed [] for section in state[sections]: result llm.invoke( [ SystemMessage(contentWrite a short paragraph (2-3 sentences) for the given section.), HumanMessage(contentfSection: {section.name}\nDescription: {section.description}), ] ) completed.append(f## {section.name}\n{result.content}\n) return {completed_sections: completed} def synthesizer(state: State): 合成器汇总所有章节为最终报告 final_report f# Report: {state[topic]}\n\n final_report \n.join(state[completed_sections]) return {final_report: final_report} # 5. 构建工作流 builder StateGraph(State) # 添加节点 builder.add_node(orchestrator, orchestrator) builder.add_node(worker, worker) builder.add_node(synthesizer, synthesizer) # 添加执行边 builder.add_edge(START, orchestrator) builder.add_edge(orchestrator, worker) builder.add_edge(worker, synthesizer) builder.add_edge(synthesizer, END) # 编译工作流 workflow builder.compile() if __name__ __main__: # 执行工作流 state workflow.invoke({topic: 人工智能的发展历史, completed_sections: []}) # 输出结果 safe_print(state[final_report])五、预期输出示例plaintext# Report: 人工智能的发展历史 ## 早期发展 人工智能的概念起源于20世纪50年代1956年达特茅斯会议正式提出了人工智能这一术语早期研究聚焦于逻辑推理和符号系统如通用问题求解器等。 ## 机器学习时代 21世纪初机器学习开始快速发展随着数据量的增长和计算能力的提升支持向量机、决策树等算法逐步落地让AI从理论走向实际应用。 ## 深度学习革命 2010年代深度学习带来了重大突破卷积神经网络CNN、循环神经网络RNN等模型的出现推动AI在计算机视觉、自然语言处理等领域实现跨越式发展。六、关键知识点总结Orchestrator-Worker 核心AI 动态拆分任务数量而非固定任务列表适配报告撰写、多文件处理等灵活场景结构化输出的价值通过 Pydantic 定义 Schema让 AI 返回可直接解析的结构化数据避免自由文本解析的繁琐与错误LangGraph 状态管理State 作为全局数据容器通过operator.add实现数据累加保障工作流数据流转的完整性。