本文介绍了LangGraph智能体开发的两种高阶设计模式：协调器-工作者模式和评估器-优化器模式。协调器-工作者模式通过动态任务分解与委派，处理复杂不确定的子任务；评估器-优化器模式通过生成-反馈-优化的迭代循环，持续提升输出质量。文章提供了详细的代码示例和实现方法，帮助开发者掌握这些模式在复杂AI应用中的实际应用。

---

前言

一、协调器-工作者模式

1.1 协调器-工作者模式定义

协调器-工作者模式专为子任务需求实现未知，需要在执行期间动态确定的复杂任务而设计。在此模式中，大模型扮演“协调器”的核心角色，负责对初始任务进行规划、分解，将其拆解为一系列更小、更易管理的子任务，随后将这些子任务委派给作为“工作者”的大模型去执行。当所有工作者完成各自分配的任务后，协调器将汇总并整合它们的输出，最终形成一个连贯、完整的最终结果。

协调器-工作者模式与简单的并行化工作流模式最主要的区别在于任务的动态性：并行化工作流通常适用于子任务数量与逻辑都提前预知并设计好的场景；而协调器-工作者模式则更加灵活，它依赖大模型根据任务的具体情况和上下文，动态地生成需要执行的子任务分支，实现了一种“边规划、边执行”的自适应过程。

协调器-工作者工作流特别适用于子任务难以或无法预先确定的复杂场景，典型案例如下：

1. 复杂编码任务

   当要求实现一个复杂功能或重构大型代码库时，大模型（协调器）会首先分析需求，动态生成修改规划，识别出需要变更的多个文件和代码模块，随后将各个具体的代码修改任务委派给配备了代码编辑工具的多个工作者模型并行执行。

2. 深度信息检索与研究任务

   在需要从多来源搜集、分析并综合信息的复杂搜索中，协调器大模型可以动态决定需要检索的相关信息来源（如不同的网站、数据库），并将每个来源的信息收集与分析工作委派给不同的工作者模型。这正是诸如DeepResearch等深度研究功能的常见实现模式。在OpenAI推出的DeepResearch功能中，其工作流就是先由一个大模型生成研究规划，然后调用多个模型并行查找资料、分别撰写报告的不同部分，最后再将这些部分合并统一。关于此模式的具体实现，笔者在解析LangChain官方OpenDeepResearch项目的文章LangGraph实战项目：从零手搓DeepResearch（四）——OpenDeepResearch源码解析与本地部署 中已有详细阐述，其中的多智能体工作流正是协调器-工作者模式的典范，感兴趣大家可以阅读笔者文章深入学习。

LangGraph为实现协调器-工作者工作流提供了强大支持，其核心机制在于 Send API。该API允许在运行时动态创建工作者节点，使协调器节点能够根据规划，灵活地将不同子任务分派给相应的工作者。每个工作者独立运行，维护自己的状态，它们的输出最终会被收集到协调器可访问的共享状态中。这种设计使得协调器能够轻松汇总所有工作者的成果，并将其合成为最终输出。

1.2 协调器-工作者模式代码实战

理论描述或许不够直观，接下来笔者将通过一个构建“专题报告生成智能体”的完整代码实例，带领大家具体学习如何在 LangGraph 中实现协调器-工作者模式。该智能体的工作流程是：首先由协调器分析主题并生成报告的不同章节规划，然后将每个章节的撰写任务分配给独立的“工作者”，最后汇总所有章节形成完整报告。

1. 引入相关依赖：在项目文件夹下新建.env文件，填入你的DEEPSEEK_API_KEY。

import operator
from typing import Annotated, List, TypedDict
from langchain.messages import SystemMessage, HumanMessage
from langchain_deepseek import ChatDeepSeek
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from dotenv import load_dotenv
from pydantic import BaseModel, Field
load_dotenv()
llm = ChatDeepSeek(
model="deepseek-chat",
)

2. 定义结构化的输出：主要用来定义报告的章节。此处使用Pydantic模型进行约束，通过Field为不同字段提供描述，可以更有效地帮助大模型规划器planner = llm.with_structured_output(Sections) 理解并生成结构化的规划结果，确保输出格式的稳定性。

# 1. 定义结构化的输出格式，用于规划报告的章节
class Section(BaseModel):
name: str = Field(
description='报告章节的名称'
)
description: str = Field(
description='本章节中涵盖的主要主题和概念的简要概述'
)
class Sections(BaseModel):
sections: List[Section] = Field(
description='报告的章节'
)
planner = llm.with_structured_output(Sections)

3. 定义状态：这里定义了两个状态类，一个是协调器的全局状态 (State)，另一个是工作者的任务状态 (WorkerState)。它们通过一些共享的键（如completed_sections）来实现数据传递与汇总。

# 2. 定义协调器的状态
class State(TypedDict):
topic: str
sections: List[Section]
completed_sections: Annotated[list, operator.add]
final_report: str
# 3. 定义工作者状态
class WorkerState(TypedDict):
section: Section
completed_sections: Annotated[list, operator.add]

4. 定义图中的节点：需要定义三类节点：协调器节点（生成章节规划）、工作者节点（撰写具体章节内容）、以及归并节点（合并所有章节）。

# 4. 定义图中的节点:
def orchestrator(state: State):
'''
协调器节点，使用结构化输出生成报告计划
'''
topic = state['topic']
# 首先使用planner生成结构化输出报告
report_sections = planner.invoke(
[
SystemMessage(content='请根据用户输入的主题生成报告计划。'),
HumanMessage(content=f'这是报告主题:{topic}')
]
)
return {
'sections': report_sections.sections
}
def llm_call(state: WorkerState):
'''
工作者节点：根据分配的章节详细信息描述生成报告的章节内容
'''
section_name = state['section'].name
section = llm.invoke(
[
SystemMessage(content='根据提供的章节的名称和描述编写报告章节，每个章节中不包含序言，使用markdown格式。200字以内'),
HumanMessage(content=f'这是章节的名称: {section_name}')
]
)
return {
'completed_sections': [section.content]
}
def synthesizer(state: State):
'''
将各个章节的输出合称为完整的报告
'''
completed_sections = state['completed_sections']
completed_report_sections = "\n\n".join(completed_sections)
return {
'final_report': completed_report_sections
}

5. 定义边和图：这是实现动态工作流的关键步骤。特别注意assign_workers 函数，它利用 Send API，将 state['sections'] 列表中的每一个章节 (Section) 对象，分别发送给名为 'llm_call' 的工作者节点。同时，State 与 WorkerState 之间的共同状态键（如 completed_sections）会自动传递。

# 5. 定义条件边函数，将工作者动态分配对应的计划章节
def assign_workers(state: State):
'''
使用send API 将工作者分配给计划中的每个章节，以实现动态工作者创建
'''
return [Send('llm_call', {'section': s}) for s in state['sections']]
orchestrator_worker_builder = StateGraph(State)
orchestrator_worker_builder.add_node('orchestrator', orchestrator)
orchestrator_worker_builder.add_node('llm_call', llm_call)
orchestrator_worker_builder.add_node('synthesizer', synthesizer)
orchestrator_worker_builder.add_edge(START, 'orchestrator')
orchestrator_worker_builder.add_conditional_edges(
'orchestrator',
assign_workers,
['llm_call']
)
orchestrator_worker_builder.add_edge('llm_call', 'synthesizer')
orchestrator_worker_builder.add_edge('synthesizer', END)
orchestrator_worker = orchestrator_worker_builder.compile()

6. 运行测试：运行智能体，可以看到成功生成了关于“LLM缩放定律”的报告。当然，这个示例还可以进一步精细化，例如控制章节的生成顺序、加入更复杂的审核逻辑等。这些优化就留给大家练习实现啦。

result = orchestrator_worker.invoke({
'topic': '创建关于LLM缩放定律的报告'
})
print(result['final_report'])

二、评估器-优化器模式

2.1 评估器-优化器模式定义

评估器-优化器模式直观地模拟了人类通过“生成-反馈-修订”进行迭代改进的工作方式，旨在通过多轮循环不断提升输出结果的质量。在此模式中，核心包含两个角色：生成器与评估器。生成器负责创建初始响应或作品；评估器则负责对该输出进行审视、分析，并提供具体、可操作的反馈意见。随后，此反馈被送回给生成器，指导其在下一次迭代中优化输出。这一过程可以循环进行，直至产出令人满意的结果，或达到预设的迭代次数上限。

评估器-优化器模式尤其适用于对输出质量有较高要求、且需要精细打磨的任务场景。以文学翻译为例：生成器的初版翻译可能准确传达了字面意思，但在语言风格、文化隐喻或情感韵味上有所缺失。经过专门设定的评估器能够识别这些不足，并提供诸如“建议使用更地道的俚语”或“此处宜考虑文化背景进行意译”等反馈。生成器据此修订，从而在后续版本中产出更优秀的译文。由此可见，评估器-优化器模式的有效性依赖于两个关键因素：首先生成器必须具备根据反馈持续改进输出的能力；其次评估器必须能够提供精准、有用且具指导性的反馈，而不仅仅是简单的优劣判断。

2.2 评估器-优化器代码实战

接下来，笔者将通过一个生动且易于理解的示例—— “优化笑话生成” ，来具体演示如何构建评估器-优化器工作流。笔者将创建一个能够不断自我评估并改进，直至生成一个“足够好笑”笑话的智能体。

1. 引入相关依赖：

from typing import TypedDict, Literal
from langchain_deepseek import ChatDeepSeek
from langgraph.graph import StateGraph, START, END
from dotenv import load_dotenv
from pydantic import BaseModel, Field
load_dotenv()
llm = ChatDeepSeek(
model="deepseek-chat",
)

2. 定义评估器结构和评估模型：笔者使用Pydantic模型来严格定义评估器的输出格式，确保其每次都能提供结构化的反馈（包含“是否好笑”的判定和改进建议）。

# 1. 定义评估器的结构
class Feedback(BaseModel):
grade: Literal['funny', 'not funny'] = Field(
description='判断笑话是否有趣'
)
feedback: str = Field(
description='如果笑话不好笑，提供改进它的反馈'
)
evaluator = llm.with_structured_output(Feedback)

3. 定义状态：状态用于在工作流的不同节点间传递和共享信息，主要包括笑话的主题、当前版本的内容、评估反馈以及最终评判结果。

# 2. 定义状态
class State(TypedDict):
topic: str
joke: str
feedback: str
funny_or_not: str

4. 定义节点：这是工作流的核心，包含生成器节点和评估器节点。生成器节点会检查状态中是否存在历史反馈，若有则将其融入新的生成指令中，从而实现基于反馈的优化。

# 3. 定义节点
def llm_call_generator(state: State):
'''
生成器节点，llm生成笑话，可能会结合之前评估器的反馈
'''
topic = state['topic']
if state.get('feedback'):
feedback = state['feedback']
msg = llm.invoke(f'请写一个关于{topic}的笑话，但是要考虑反馈:{feedback}')
else:
msg = llm.invoke(f'写一个关于{topic}的笑话')
return {
'joke': msg.content
}
def llm_call_evaluator(state: State):
'''
评估生成笑话
'''
joke = state['joke']
grade = evaluator.invoke(f'评估笑话{joke}是否好笑,如果不好笑给出修改建议')
return {
'funny_or_not': grade.grade,
'feedback': grade.feedback
}

5. 定义条件边和图：通过条件边将生成器节点和评估器节点连接成一个可循环的图。关键逻辑在于：评估器对笑话进行评判后，如果结果为‘not funny’，工作流将带着反馈返回到生成器节点，开启新一轮的优化迭代；如果结果为‘funny’，则流程终止，输出最终笑话。

# 定义条件边
def route_joke(state: State):
if state['funny_or_not'] == 'funny':
return 'Accepted'
elif state['funny_or_not'] == 'not funny':
return 'Rejected'
optimizer_builder = StateGraph(State)
optimizer_builder.add_node('llm_call_generator', llm_call_generator)
optimizer_builder.add_node('llm_call_evaluator', llm_call_evaluator)
optimizer_builder.add_edge(START, 'llm_call_generator')
optimizer_builder.add_edge('llm_call_generator', 'llm_call_evaluator')
optimizer_builder.add_conditional_edges(
'llm_call_evaluator',
route_joke,
{
'Accepted': END,
'Rejected': 'llm_call_generator'
}
)
optimizer_workflow = optimizer_builder.compile()

6. 运行测试：以“贾乃亮与pg one”为主题运行这个工作流。经过数轮生成-评估-优化的循环后，最终产出的笑话相较于直接一次性生成的结果提升还是很明显滴。

# 运行测试
result = optimizer_workflow.invoke({
'topic': '贾乃亮与pg one'
})
print(result['joke'])

以上就是今天的全部内容，完整代码大家可以关注笔者的同名微信公众号 大模型真好玩，并私信: LangChain智能体开发免费获取。

三、总结

本文分享了LangGraph单智能体开发的两种高阶模式：协调器-工作者模式通过动态任务分解与委派，擅长处理复杂、不确定的子任务场景；评估器-优化器模式则通过生成-反馈-优化的迭代循环，持续提升输出质量。两者均显著增强了智能体处理复杂任务的能力，为开发更强大的AI应用提供了核心思路。

​最后

我在一线科技企业深耕十二载，见证过太多因技术更迭而跃迁的案例。那些率先拥抱 AI 的同事，早已在效率与薪资上形成代际优势，我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。

我整理出这套 AI 大模型突围资料包：

• ✅AI大模型学习路线图
• ✅Agent行业报告
• ✅100集大模型视频教程
• ✅大模型书籍PDF
• ✅DeepSeek教程
• ✅AI产品经理入门资料

完整的大模型学习和面试资料已经上传带到CSDN的官方了，有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇
​​

为什么说现在普通人就业/升职加薪的首选是AI大模型？

人工智能技术的爆发式增长，正以不可逆转之势重塑就业市场版图。从DeepSeek等国产大模型引发的科技圈热议，到全国两会关于AI产业发展的政策聚焦，再到招聘会上排起的长队，AI的热度已从技术领域渗透到就业市场的每一个角落。

智联招聘的最新数据给出了最直观的印证：2025年2月，AI领域求职人数同比增幅突破200% ，远超其他行业平均水平；整个人工智能行业的求职增速达到33.4%，位居各行业榜首，其中人工智能工程师岗位的求职热度更是飙升69.6%。

AI产业的快速扩张，也让人才供需矛盾愈发突出。麦肯锡报告明确预测，到2030年中国AI专业人才需求将达600万人，人才缺口可能高达400万人，这一缺口不仅存在于核心技术领域，更蔓延至产业应用的各个环节。

​​

资料包有什么？

①从入门到精通的全套视频教程⑤⑥

包含提示词工程、RAG、Agent等技术点

② AI大模型学习路线图（还有视频解说）

全过程AI大模型学习路线

③学习电子书籍和技术文档

市面上的大模型书籍确实太多了，这些是我精选出来的

④各大厂大模型面试题目详解

⑤ 这些资料真的有用吗?

这份资料由我和鲁为民博士共同整理，鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位，在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利，同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。

所有的视频教程由智泊AI老师录制，且资料与智泊AI共享，相互补充。这份学习大礼包应该算是现在最全面的大模型学习资料了。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

智泊AI始终秉持着“让每个人平等享受到优质教育资源”的育人理念‌，通过动态追踪大模型开发、数据标注伦理等前沿技术趋势‌，构建起"前沿课程+智能实训+精准就业"的高效培养体系。

课堂上不光教理论，还带着学员做了十多个真实项目。学员要亲自上手搞数据清洗、模型调优这些硬核操作，把课本知识变成真本事‌！

​​​​

如果说你是以下人群中的其中一类，都可以来智泊AI学习人工智能，找到高薪工作，一次小小的“投资”换来的是终身受益！

应届毕业生‌：无工作经验但想要系统学习AI大模型技术，期待通过实战项目掌握核心技术。

零基础转型‌：非技术背景但关注AI应用场景，计划通过低代码工具实现“AI+行业”跨界‌。

业务赋能 ‌突破瓶颈：传统开发者（Java/前端等）学习Transformer架构与LangChain框架，向AI全栈工程师转型‌。

👉获取方式：

😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓**

​