一、LangGraph设计模式
1.1 LangGraph设计模式总览
随着大模型智能体能力的不断增强，其需要处理的任务也日趋复杂。如同城市建设需要规划蓝图、软件开发离不开架构设计一样，构建强大且稳健的人工智能体系统，同样需要精心设计的架构模式作为支撑。
从本期开始，笔者将系统介绍在 LangGraph 框架下构建各类智能体系统的关键架构模式。这些模式将涵盖从基础的工作流编排，到更高级的多智能体协同设计，为大家开发智能体提供一套可复用、可扩展的方法论。
1.2 LangGraph工作流模式
笔者首先介绍LangGraph 的工作流模式，它常被称为单智能体模式。需要明确的是，“单智能体”并非指仅使用一个大模型，而是指系统目标单一，且所有决策由一个统一的智能体核心来完成。一个基础的单智能体通常由以下部分构成：大模型（如 DeepSeek、Qwen3）作为“大脑”，可调用各种工具，具备记忆存储与检索能力，并通过系统提示词来响应用户输入。
而LangGraph 的工作流模式，正是在此基础上，通过引入规划与多步调用机制等，来更高效、更可靠地完成复杂用户任务。常见的工作流模式包括提示链模式、路由模式、并行化模式、协调器-工作者模式、评估器-优化器模式等。本期笔者先从几个相对基础的模式开始分享：提示链模式、路由模式和并行化模式。
二、提示链模式
2.1 提示链模式定义
提示链模式是最基础的工作流模式之一，其核心思想是将复杂任务分解为一系列更简单、相互关联的步骤。在该模式中，大模型被多次顺序调用，且前一次调用的输出常作为后续调用的输入，从而形成一个多阶段、串行处理的工作流。
为了确保流程的质量和方向，提示链模式在步骤之间通常会引入程序化的检查点，称为“门控”（Gate）。门控作为质量审查节点，能够校验上一步的输出是否满足特定条件，从而决定流程是继续向前、转向特定分支还是提前结束。
提示链模式的核心优势在于：
分解复杂度
将庞杂任务拆分为更小、更易管理的子任务，降低单次处理的认知负荷。
提升准确性与可控性
通过多步骤的精细化和中间检查，能够获得比单次大模型调用更可靠、更符合预期的输出。
该模式特别适用于那些能够被清晰划分为顺序子任务的场景。例如，要生成一个小众语言的营销文案，可以将其拆解为：第一步由大模型生成高质量的英文/中文原稿，第二步再由大模型将其翻译为目标语言。在第一步之后，还可以加入门控检查，确保原稿的调性、长度或关键信息点符合要求，从而更好地控制最终结果的质量。
2.2 提示链模式代码示例
单纯的概念描述不够直观，接下来笔者通过一个具体的代码实例来理解。本例将模拟一个多步骤生成优质笑话的流程：首先生成简短笑话，然后根据检查结果决定是否增强趣味性，最后进行润色转折。
1. 引入相关依赖：在项目文件夹下新建.env文件，填入你的DEEPSEEK_API_KEY。
from typing import TypedDict from langchain_deepseek import ChatDeepSeek from langgraph.graph import END, START, StateGraph from dotenv import load_dotenv #0. 配置模型 load_dotenv() llm = ChatDeepSeek( model="deepseek-chat", )2. 定义状态：这是一个三阶段流程，因此状态需要记录初始主题、各阶段生成的笑话
# 1. 定义图状态 class State(TypedDict): topic: str joke: str improved_joke: str final_joke: str3. 定义节点函数：包括三个大模型调用节点和一个门控检查节点。
# 2. 定义节点函数 def generate_joke(state: State): ''' 第一个大模型调用，根据主题生成初始笑话 ''' topic = state['topic'] msg = llm.invoke(f'写一个关于{topic}的简短笑话') return { 'joke': msg.content } def check_punchline(state: State): ''' 模拟门控函数——笑话中是否包含?或! ''' joke = state['joke'] if '?' in joke or '？' in joke: return "Fail" # 未能通过门控检查, 需要继续增强 return "Pass" def improve_joke(state: State): ''' 第二个大模型调用，通过添加文字游戏改进笑话 ''' joke = state['joke'] msg = llm.invoke(f'通过添加文字游戏使笑话更有趣，当前笑话是: {joke}') return { 'improved_joke': msg.content } def polish_joke(state: State): ''' 第三个大模型调用，最终润色笑话，添加令人惊讶的转折 ''' improved_joke = state['improved_joke'] msg = llm.invoke(f'为这个笑话添加一个令人惊讶的转折: {improved_joke}') return { 'final_joke': msg.content }4. 定义边并构建图：通过条件边来实现门控逻辑
#2. 定义边和图 workflow = StateGraph(State) workflow.add_node('generate_joke', generate_joke) workflow.add_node('improve_joke', improve_joke) workflow.add_node('polish_joke', polish_joke) workflow.add_edge(START, 'generate_joke') workflow.add_conditional_edges('generate_joke', check_punchline, { 'Fail': 'improve_joke', 'Pass': END }) workflow.add_edge('improve_joke', 'polish_joke') workflow.add_edge('polish_joke', END) chain = workflow.compile()5. 测试运行：当生成的简短笑话包含‘?’时，无法通过门控检查，流程会进入增强和润色环节。虽然笔者一直觉得大模型生成的笑话不好笑，但从最终表现形式上看，笑话已经从简短的一小段扩充到丰富的内容，这正是****提示链模式通过顺序分解任务来提升输出效果的体现。
#3. 测试运行 state = chain.invoke({'topic': '小猫'}) print('初始笑话:') print(state['joke']) if 'improved_joke' in state: print('改进后笑话:') print(state['improved_joke']) print('最终笑话:') print(state['final_joke'])三、路由模式
3.1 路由模式概念
路由模式旨在通过对输入请求进行智能分类，并将其定向到专门的下游处理节点，从而高效应对多样化的任务需求。在 LangGraph 的实现中，这通常涉及定义一个路由器节点（负责分类决策）和一组下游节点（每个节点专精于处理某一特定类型的任务）。工作流会使用条件边，根据路由器节点的分类结果，将任务流转导向最合适的处理分支。
路由模式的核心思想是引入一个分流步骤。该步骤可以由大模型驱动，也可以由更传统的分类模型支持。整个工作流的成功在很大程度上取决于这个分类步骤的准确性：如果输入被错误分类，就会被引导至不恰当的处理路径，从而导致输出结果不正确或不满足预期。
路由模式非常适合处理同一大类任务下包含多种不同处理逻辑的场景。以下是两个典型适用例子：
客户服务分流
在客户服务应用中，路由可用于区分不同类型的客户查询。例如，一般咨询、退款请求和技术支持工单可以被路由到不同的专用下游流程。一般咨询可能触发简单的FAQ检索；退款请求则可能启动一个需要查询订单历史并调用支付接口的工作流；而技术支持则会被导向包含详细故障排查步骤的专业流程。
资源与成本优化
路由还可用于优化系统的响应速度与计算成本。例如，可以根据请求的复杂度或紧急性进行分类：将简单、常见的查询路由到更轻量、快速且成本低廉的模型（如 Qwen3-8B），而将复杂、特殊的请求定向给能力更强但资源消耗也更大的模型（如 DeepSeek-671B）。
3.2 路由模式代码示例
单纯概念描述不直观，笔者通过一个具体的代码实例来理解。本例是一个内容生成器，根据用户输入决定是输出笑话、故事还是诗歌。
1. 引入依赖：（同样需要配置好.env文件中的DEEPSEEK_API_KEY）
from typing import TypedDict, Literal from langchain_deepseek import ChatDeepSeek from langgraph.graph import StateGraph, START, END from langchain.messages import HumanMessage, SystemMessage from dotenv import load_dotenv load_dotenv() llm = ChatDeepSeek( model="deepseek-chat", )2. 定义状态：状态需要记录用户输入、路由决策以及最终输出
class State(TypedDict): input: str decision: str output: str3. 定义节点函数：包括三个专精于不同创作类型的下游节点，以及一个负责分类的路由器节点
# 1. 定义节点 def generate_story(state: State): ''' 写故事节点 ''' print('进入写故事处理逻辑') result = llm.invoke(state['input']) return { 'output': result.content } def generate_joke(state: State): ''' 写笑话节点 ''' print('进入写笑话处理逻辑') result = llm.invoke(state['input']) return { 'output': result.content } def generate_poetry(state: State): ''' 写诗歌节点 ''' print('进入写诗歌处理逻辑') result = llm.invoke(state['input']) return { 'output': result.content } class Classification(TypedDict): response_format: Literal['story', 'joke', 'poetry'] def llm_call_router(state: State): ''' 使用结构化输出将输入路由到适当的节点 ''' structed_llm = llm.with_structured_output(Classification) input_content = state['input'] response = structed_llm.invoke([ SystemMessage(content=''' 你是一个分类路由，根据用户的输入进行分类，分类结果是story, joke, poetry三者中的一种 '''), HumanMessage(content=input_content) ]) return { 'decision': response['response_format'] }4. 定义边并构建图：利用条件边实现基于决策的分支路由
# 定义条件边函数 def route_decision(state: State): if state['decision'] == 'story': return 'llm_story' elif state['decision'] == 'joke': return 'llm_joke' elif state['decision'] == 'poetry': return 'llm_poetry' # 2. 定义边和图 router_builder = StateGraph(State) router_builder.add_node('llm_story', generate_story) router_builder.add_node('llm_joke', generate_joke) router_builder.add_node('llm_poetry', generate_poetry) router_builder.add_node('llm_call_router', llm_call_router) router_builder.add_edge(START, 'llm_call_router') router_builder.add_conditional_edges( 'llm_call_router', route_decision, { 'llm_story': 'llm_story', 'llm_joke': 'llm_joke', 'llm_poetry': 'llm_poetry' } ) router_builder.add_edge('llm_story', END) router_builder.add_edge('llm_joke', END) router_builder.add_edge('llm_poetry', END) workflow = router_builder.compile()5. 运行测试：输入“给我写一个关于苍井空的笑话”。如图所示，路由器成功将请求分类为joke，并将其引导至生成笑话的节点进行处理与输出（不得不说，当前大模型在内容生成上仍趋于保守）
result = workflow.invoke({ 'input': '给我写一个关于苍井空的笑话' }) print(result['output'])四、并行模式
4.1 并行模式概念
并行模式旨在充分利用大模型的能力，同时处理一个任务中不同方面或同一任务的不同变体。与顺序执行的提示链模式不同，并行模式允许多个大模型调用同时进行，最后将各自的输出进行聚合，从而显著提升处理效率。
根据任务目标的不同，并行模式的主要聚合方式可分为两类：分段与投票。
分段
适用于将一个复杂任务拆分为多个独立的子任务。这些子任务可以并行执行，每个由一个独立的大模型调用处理，最终将结果组合成完整输出。例如，要全面评估一段内容，可以并行调用多个模型，分别评估其事实准确性、逻辑连贯性和文风匹配度，最后聚合这些维度的评分，得到一个综合评估报告。
投票
适用于通过共识机制提升结果的准确性。即对同一任务并行执行多次，每次使用略有不同的提示词或配置，然后根据多数输出或某种共识来决定最终结果。例如，在审查一段代码是否存在安全漏洞时，可以用三个并行的模型，分别从“内存安全”、“注入攻击”和“逻辑缺陷”三个角度进行审查。如果其中两个模型都认为存在漏洞，则可以相对可靠地判定该代码有风险。
4.2 并行模式
单纯概念描述不直观，笔者通过一个具体的代码实例来理解。该任务是通过用户输入的主题，生成一个包含故事、笑话和诗歌的合集。
1. 引入依赖:（同样需要配置好.env文件中的DEEPSEEK_API_KEY）
from langchain_deepseek import ChatDeepSeek from langgraph.graph import StateGraph, START, END from dotenv import load_dotenv load_dotenv() llm = ChatDeepSeek( model="deepseek-chat", )2. 定义工作流的状态：围绕一个主题，并行生成笑话、故事和诗歌，最后聚合为合集
class State(TypedDict): topic: str joke: str story: str poetry: str combined_output: str3. 定义节点函数：除了生成笑话、故事、诗歌三个****并行执行的节点外，还需要一个聚合节点将三者的输出整合。
def generate_joke(state: State): ''' 生成笑话的节点 ''' topic = state['topic'] msg = llm.invoke(f'写一个关于{topic}的笑话') return { 'joke': msg.content } def generate_story(state: State): ''' 生成故事的节点 ''' topic = state['topic'] msg = llm.invoke(f'写一个关于{topic}的故事') return { 'story': msg.content } def generate_poetry(state: State): ''' 生成诗歌的节点 ''' topic = state['topic'] msg = llm.invoke(f'写一个关于{topic}的诗歌') return { 'poetry': msg.content } def aggregator(state: State): ''' 聚合笑话、故事、诗歌的节点 ''' topic = state['topic'] joke = state['joke'] story = state['story'] poetry = state['poetry'] combined = f'这是一个关于 {topic} 的故事、笑话和诗歌的合集\n\n' combined += f'故事\n {story}\n\n' combined += f'笑话\n {joke}\n\n' combined += f'诗歌\n {poetry}\n\n' return { 'combined_output': combined }4. 定义边和图：注意，这里三个生成节点都从START同时开始，体现了并行性。
parallel_builder = StateGraph(State) parallel_builder.add_node('generate_joke', generate_joke) parallel_builder.add_node('generate_story', generate_story) parallel_builder.add_node('generate_poetry', generate_poetry) parallel_builder.add_node('aggregator', aggregator) parallel_builder.add_edge(START, 'generate_joke') parallel_builder.add_edge(START, 'generate_story') parallel_builder.add_edge(START, 'generate_poetry') parallel_builder.add_edge('generate_joke', 'aggregator') parallel_builder.add_edge('generate_story', 'aggregator') parallel_builder.add_edge('generate_poetry', 'aggregator') parallel_builder.add_edge('aggregator', END) workflow = parallel_builder.compile()5. 运行测试：使用“pgone 与 李小璐”作为主题进行测试，结果如下（deepseek 连这个故事也不能讲嘛）
state = workflow.invoke({ 'topic': 'pgone 与 李小璐' }) print(state['combined_output'])以上就是今天的全部内容，完整代码大家可以关注笔者的微信公众号大模型真好玩，并私信:LangChain智能体开发免费获取。
五、总结
本文分享了LangGraph智能体开发的设计模式：工作流模式和多智能体模式，同时详细讲解了三种核心工作流模式。提示链模式通过顺序分解任务与门控检查提升准确性；路由模式借助智能分类实现专业化任务分流；并行模式同时处理多个子任务以提升效率。这些模式为构建复杂、可靠的智能体系统提供了可复用的基础架构。当然今天的内容仅仅是开胃小菜，下一期内容笔者将分享更加灵活的设计模式：协调器-工作者模式和评估器-优化器模式，大家敬请期待。
如何学习大模型 AI ？
由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。
但是具体到个人，只能说是：
“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。
这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。
我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。
我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。
第一阶段（10天）：初阶应用
该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。
- 大模型 AI 能干什么？
- 大模型是怎样获得「智能」的？
- 用好 AI 的核心心法
- 大模型应用业务架构
- 大模型应用技术架构
- 代码示例：向 GPT-3.5 灌入新知识
- 提示工程的意义和核心思想
- Prompt 典型构成
- 指令调优方法论
- 思维链和思维树
- Prompt 攻击和防范
- …
第二阶段（30天）：高阶应用
该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。
- 为什么要做 RAG
- 搭建一个简单的 ChatPDF
- 检索的基础概念
- 什么是向量表示（Embeddings）
- 向量数据库与向量检索
- 基于向量检索的 RAG
- 搭建 RAG 系统的扩展知识
- 混合检索与 RAG-Fusion 简介
- 向量模型本地部署
- …
第三阶段（30天）：模型训练
恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。
到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？
- 为什么要做 RAG
- 什么是模型
- 什么是模型训练
- 求解器 & 损失函数简介
- 小实验2：手写一个简单的神经网络并训练它
- 什么是训练/预训练/微调/轻量化微调
- Transformer结构简介
- 轻量化微调
- 实验数据集的构建
- …
第四阶段（20天）：商业闭环
对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。
- 硬件选型
- 带你了解全球大模型
- 使用国产大模型服务
- 搭建 OpenAI 代理
- 热身：基于阿里云 PAI 部署 Stable Diffusion
- 在本地计算机运行大模型
- 大模型的私有化部署
- 基于 vLLM 部署大模型
- 案例：如何优雅地在阿里云私有部署开源大模型
- 部署一套开源 LLM 项目
- 内容安全
- 互联网信息服务算法备案
- …
学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。
如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。
)
