【收藏版】LangGraph智能体开发核心三板斧:3种工作流模式(附完整代码+实战解析)
将一个复杂任务拆解为多个简单、有依赖关系的子步骤,让大模型按顺序完成每一步,前一步的输出作为后一步的输入,形成串行处理的工作流。为了避免流程"跑偏",提示链模式会在步骤之间加入"质量检查站"——也就是门控(Gate)节点。门控会校验上一步输出是否符合预设条件,再决定流程是继续执行、转向修正分支,还是直接终止,从源头提升输出的可控性。# 1. 定义图状态:存储主题、各阶段生成的笑话内容topic:
本文系统拆解LangGraph智能体开发的三大核心工作流模式:提示链模式通过任务分步拆解+门控校验双管齐下提升输出精准度;路由模式依托智能分类实现任务专业化分流,适配多场景需求;并行模式通过多子任务同步处理,大幅提升复杂任务的处理效率。每种模式均配套可直接运行的代码案例,结合新手易懂的实战场景,帮助开发者快速搭建高可用、易维护的智能体工作流,为大模型应用落地提供可复用的架构模板。
一、LangGraph设计模式:大模型智能体的"架构蓝图"
1.1 LangGraph设计模式总览(建议收藏)
大模型智能体的能力边界持续拓展,面对的业务任务也从简单问答转向复杂多步骤处理。就像盖房子需要施工图、写项目需要架构图一样,想要打造稳定、高效的大模型智能体系统,一套成熟的LangGraph设计模式是核心支撑。
从本篇开始,我会从零拆解LangGraph框架下智能体开发的核心架构模式,覆盖从基础工作流编排到高阶多智能体协同的全流程,不仅讲理论,更结合实战代码,帮新手和进阶开发者建立一套可复用、可扩展的智能体开发方法论。

1.2 LangGraph工作流模式:单智能体的核心逻辑
首先要掌握的是LangGraph的工作流模式(单智能体模式) ——这里的"单智能体"并非指只用一个大模型,而是系统目标唯一,所有决策由一个核心智能体统一把控。一个基础的单智能体架构包含四大核心:以DeepSeek、Qwen3等大模型为"核心大脑",支持工具调用能力,具备记忆存储与检索模块,同时通过精准的系统提示词承接用户输入。

而LangGraph工作流模式,正是在这个基础架构上,加入任务规划和多步调用机制,让智能体能够拆解复杂任务、按规则执行,解决单次大模型调用"力不从心"的问题。常见的工作流模式有提示链、路由、并行化、协调器-工作者、评估器-优化器等,本篇先聚焦新手最易上手、落地性最强的三种:提示链模式、路由模式、并行化模式。
二、提示链模式:复杂任务的"分步拆解神器"
2.1 提示链模式定义:把大任务拆成"小积木"
提示链是LangGraph最基础也最常用的工作流模式,核心逻辑是:将一个复杂任务拆解为多个简单、有依赖关系的子步骤,让大模型按顺序完成每一步,前一步的输出作为后一步的输入,形成串行处理的工作流。
为了避免流程"跑偏",提示链模式会在步骤之间加入"质量检查站"——也就是门控(Gate) 节点。门控会校验上一步输出是否符合预设条件,再决定流程是继续执行、转向修正分支,还是直接终止,从源头提升输出的可控性。

提示链模式的核心优势(新手必记):
- 降低认知负荷:将"写一篇营销文案"这类大任务,拆成"确定主题→生成初稿→校验内容→优化润色"等小步骤,大模型单次处理更聚焦;
- 提升输出可靠性:多步骤精细化处理+中间门控校验,比单次调用大模型的结果更符合预期,减少"答非所问";
- 易调试易维护:每个步骤独立,出问题时能快速定位是哪个环节的问题,新手也能轻松排查。
典型适用场景:
适合能清晰拆分出顺序子任务的场景,比如:生成小众语言营销文案(中文初稿→门控校验→目标语言翻译)、撰写技术文档(大纲生成→内容填充→格式校验→最终润色)等。
2.2 提示链模式代码示例:手把手实现"智能笑话生成器"
光讲概念太抽象,我们用"生成优质笑话"这个趣味场景,手把手写一个提示链工作流:先生成基础笑话→门控检查是否达标→不达标则优化→最后润色,全程代码可直接复制运行。
前置准备:
在项目目录新建.env文件,填入你的DEEPSEEK_API_KEY(没有的话去DeepSeek官网申请,新手也能快速搞定)。
from typing import TypedDict
from langchain_deepseek import ChatDeepSeek
from langgraph.graph import END, START, StateGraph
from dotenv import load_dotenv
# 0. 配置大模型(新手注意:确保安装依赖 pip install langchain-deepseek python-dotenv langgraph)
load_dotenv() # 加载.env文件中的API密钥
llm = ChatDeepSeek(
model="deepseek-chat", # 指定调用的模型版本
temperature=0.7 # 新增:调整随机性,新手可先固定0.7
)
步骤1:定义工作流状态(记录每一步的输出)
# 1. 定义图状态:存储主题、各阶段生成的笑话内容
class State(TypedDict):
topic: str # 笑话主题(用户输入)
joke: str # 初始生成的笑话
improved_joke: str # 优化后的笑话
final_joke: str # 最终润色的笑话
步骤2:定义核心节点函数(每个步骤的具体逻辑)
# 2. 定义节点函数:完成各步骤的核心逻辑
def generate_joke(state: State):
'''第一步:根据主题生成初始简短笑话'''
topic = state['topic']
# 新增:优化提示词,让生成的笑话更聚焦
prompt = f"写一个简短、易懂的关于{topic}的笑话,控制在30字以内"
msg = llm.invoke(prompt)
return {'joke': msg.content}
def check_punchline(state: State):
'''门控检查:判断笑话是否包含反问/感叹语气(提升趣味性)'''
joke = state['joke']
if '?' in joke or '?' in joke or '!' in joke or '!' in joke:
return "Pass" # 达标,直接结束
return "Fail" # 不达标,进入优化环节
def improve_joke(state: State):
'''第二步:优化不达标的笑话(添加文字游戏)'''
joke = state['joke']
prompt = f"给这个笑话添加简单的文字游戏,提升趣味性,保留原主题:{joke}"
msg = llm.invoke(prompt)
return {'improved_joke': msg.content}
def polish_joke(state: State):
'''第三步:最终润色(添加转折)'''
# 兼容逻辑:如果有优化后的笑话则用优化版,否则用初始版
joke_to_polish = state.get('improved_joke', state['joke'])
prompt = f"为这个笑话添加一个出人意料但不生硬的转折:{joke_to_polish}"
msg = llm.invoke(prompt)
return {'final_joke': msg.content}
步骤3:构建工作流并运行
# 3. 构建提示链工作流
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()
# 测试:生成关于"小猫"的笑话
result = chain.invoke({'topic': '小猫'})
# 输出结果(新手友好:清晰展示每一步)
print("=== 提示链模式运行结果 ===")
print(f"初始笑话:{result['joke']}")
if 'improved_joke' in result:
print(f"优化后笑话:{result['improved_joke']}")
print(f"最终笑话:{result['final_joke']}")

新手小贴士:运行后如果初始笑话包含"?“或”!“,会直接结束;否则会自动进入优化和润色环节,这就是提示链+门控的核心价值——让流程按规则走,而非全靠大模型"自由发挥”。
三、路由模式:智能体的"任务分流器"
3.1 路由模式概念:让不同任务找对"专属处理通道"
路由模式的核心是智能分类+定向处理:先通过"路由器节点"对用户输入的任务类型进行精准分类,再根据分类结果,将任务导向对应的专业处理节点。在LangGraph中,这一逻辑通过"条件边"实现——路由器节点输出分类结果,条件边根据结果匹配下游节点。
路由模式的关键是分类的准确性:如果分类错了,任务就会进入错误的处理分支,最终输出必然不符合预期。因此新手在实现时,建议先用结构化输出约束大模型的分类结果,避免分类混乱。

路由模式的典型应用场景(收藏备用):
- 多场景客服分流:将用户咨询分为"订单查询"“退款申请”“技术问题”,分别导向FAQ检索、订单接口调用、故障排查流程;
- 模型资源优化:简单问题(如天气查询)路由到轻量模型(Qwen3-8B),复杂问题(如代码生成)路由到高性能模型(DeepSeek-67B),兼顾效率与成本;
- 内容生成分流:用户输入"写故事/写笑话/写诗歌",自动导向对应的生成逻辑,避免大模型"一锅烩"。
3.2 路由模式代码示例:打造智能内容生成分流器
我们实现一个"内容生成路由系统":用户输入任意创作需求,系统先分类(故事/笑话/诗歌),再调用对应的生成逻辑,新手可直接复用这个框架做自己的分流系统。
前置准备:
同样确保.env文件中有DEEPSEEK_API_KEY,并安装依赖:pip install langchain-deepseek langgraph python-dotenv。
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
# 0. 初始化模型(新手注意:添加超时配置,避免调用失败)
load_dotenv()
llm = ChatDeepSeek(
model="deepseek-chat",
timeout=30 # 新增:超时时间,新手调试更友好
)
步骤1:定义状态和节点函数
# 1. 定义工作流状态
class State(TypedDict):
input: str # 用户输入的创作需求
decision: str # 路由器的分类结果(story/joke/poetry)
output: str # 最终生成的内容
# 2. 定义专业处理节点(每个节点只做一件事)
def generate_story(state: State):
'''故事生成节点:专注处理故事创作需求'''
print("[路由日志] 进入故事生成流程")
prompt = f"根据需求写一篇简短的故事:{state['input']}"
result = llm.invoke(prompt)
return {'output': result.content}
def generate_joke(state: State):
'''笑话生成节点:专注处理笑话创作需求'''
print("[路由日志] 进入笑话生成流程")
prompt = f"根据需求写一个轻松的笑话:{state['input']}"
result = llm.invoke(prompt)
return {'output': result.content}
def generate_poetry(state: State):
'''诗歌生成节点:专注处理诗歌创作需求'''
print("[路由日志] 进入诗歌生成流程")
prompt = f"根据需求写一首简短的诗歌:{state['input']}"
result = llm.invoke(prompt)
return {'output': result.content}
# 3. 定义路由器节点(核心:结构化分类)
class Classification(TypedDict):
# 约束输出只能是story/joke/poetry,避免分类混乱
response_format: Literal['story', 'joke', 'poetry']
def llm_router(state: State):
'''智能路由器:精准分类用户需求'''
# 结构化输出:强制大模型按指定格式返回分类结果
structured_llm = llm.with_structured_output(Classification)
system_prompt = """你是一个精准的内容分类器,仅需根据用户输入判断创作类型,
输出只能是:story(故事)、joke(笑话)、poetry(诗歌)三者中的一个,不要添加任何额外内容。"""
response = structured_llm.invoke([
SystemMessage(content=system_prompt),
HumanMessage(content=state['input'])
])
return {'decision': response['response_format']}
步骤2:构建路由工作流并运行
# 4. 定义路由分支函数
def route_by_decision(state: State):
'''根据分类结果匹配下游节点'''
decision_map = {
'story': 'llm_story',
'joke': 'llm_joke',
'poetry': 'llm_poetry'
}
return decision_map.get(state['decision'], 'llm_joke') # 兜底:默认走笑话分支
# 5. 构建工作流
router_graph = StateGraph(State)
# 添加节点
router_graph.add_node('llm_story', generate_story)
router_graph.add_node('llm_joke', generate_joke)
router_graph.add_node('llm_poetry', generate_poetry)
router_graph.add_node('llm_router', llm_router)
# 定义流程
router_graph.add_edge(START, 'llm_router') # 开始→路由器
# 核心:条件边实现路由分流
router_graph.add_conditional_edges(
'llm_router', # 从路由器节点出发
route_by_decision, # 路由分支函数
{
'llm_story': 'llm_story',
'llm_joke': 'llm_joke',
'llm_poetry': 'llm_poetry'
}
)
# 各处理节点→结束
router_graph.add_edge('llm_story', END)
router_graph.add_edge('llm_joke', END)
router_graph.add_edge('llm_poetry', END)
# 编译并测试
workflow = router_graph.compile()
# 测试输入:可替换成"写一首关于春天的诗"、"写一个关于程序员的故事"等
test_input = "给我写一个关于小猫的笑话"
result = workflow.invoke({'input': test_input})
# 输出结果
print("=== 路由模式运行结果 ===")
print(f"用户需求:{test_input}")
print(f"分类结果:{result['decision']}")
print(f"生成内容:{result['output']}")

新手重点:路由器节点用了
with_structured_output约束输出格式,这是避免分类出错的关键——新手如果直接让大模型自由输出,很容易出现"故事/笑话"混用的情况,结构化输出能从源头解决这个问题。
四、并行模式:提升效率的"多线程处理术"
4.1 并行模式概念:让多个子任务"同时跑"
并行模式的核心是多节点同步执行+结果聚合:将一个复杂任务拆分为多个独立的子任务,让这些子任务在LangGraph中同时运行(而非串行),最后通过聚合节点将所有子任务的输出整合为最终结果,大幅缩短整体处理时间。
根据业务目标不同,并行模式的聚合方式主要分两类(新手必懂):
- 分段聚合:将一个任务拆成多个独立维度,每个维度并行处理,最后拼接结果(如:内容评估=事实准确性+逻辑连贯性+文风匹配度);
- 投票聚合:同一任务并行运行多次(不同提示词/模型),按多数结果决策(如:代码漏洞检测=多模型投票判定是否有风险)。

4.2 并行模式代码示例:一键生成"主题内容合集"
我们实现一个"主题内容生成器":输入一个主题,并行生成故事、笑话、诗歌,最后自动聚合为合集,新手可基于这个框架扩展为"多维度内容创作工具"。
from typing import TypedDict
from langchain_deepseek import ChatDeepSeek
from langgraph.graph import StateGraph, START, END
from dotenv import load_dotenv
# 0. 初始化模型
load_dotenv()
llm = ChatDeepSeek(
model="deepseek-chat",
temperature=0.8 # 提高随机性,让创作内容更丰富
)
步骤1:定义状态和核心节点
# 1. 定义工作流状态
class State(TypedDict):
topic: str # 创作主题
joke: str # 并行生成的笑话
story: str # 并行生成的故事
poetry: str # 并行生成的诗歌
combined_output: str # 聚合后的最终合集
# 2. 定义并行生成节点(三个节点同时运行)
def generate_joke(state: State):
'''笑话生成节点(并行)'''
prompt = f"写一个轻松有趣的关于{state['topic']}的笑话,控制在50字以内"
return {'joke': llm.invoke(prompt).content}
def generate_story(state: State):
'''故事生成节点(并行)'''
prompt = f"写一个简短温馨的关于{state['topic']}的小故事,控制在100字以内"
return {'story': llm.invoke(prompt).content}
def generate_poetry(state: State):
'''诗歌生成节点(并行)'''
prompt = f"写一首关于{state['topic']}的短诗,句式简洁,有画面感"
return {'poetry': llm.invoke(prompt).content}
# 3. 定义聚合节点(整合所有并行结果)
def aggregate_content(state: State):
'''聚合节点:将笑话、故事、诗歌整合成合集'''
combined = f"""
### {state['topic']} 主题内容合集
#### 小故事
{state['story']}
#### 趣味笑话
{state['joke']}
#### 短诗
{state['poetry']}
"""
return {'combined_output': combined.strip()}
步骤2:构建并行工作流并运行
# 4. 构建并行工作流
parallel_graph = StateGraph(State)
# 添加节点
parallel_graph.add_node('generate_joke', generate_joke)
parallel_graph.add_node('generate_story', generate_story)
parallel_graph.add_node('generate_poetry', generate_poetry)
parallel_graph.add_node('aggregate_content', aggregate_content)
# 核心:并行执行(三个生成节点都从START出发)
parallel_graph.add_edge(START, 'generate_joke')
parallel_graph.add_edge(START, 'generate_story')
parallel_graph.add_edge(START, 'generate_poetry')
# 所有生成节点完成后→聚合节点
parallel_graph.add_edge('generate_joke', 'aggregate_content')
parallel_graph.add_edge('generate_story', 'aggregate_content')
parallel_graph.add_edge('generate_poetry', 'aggregate_content')
# 聚合节点→结束
parallel_graph.add_edge('aggregate_content', END)
# 编译并运行
workflow = parallel_graph.compile()
# 测试:生成关于"小猫"的内容合集
result = workflow.invoke({'topic': '小猫'})
# 输出结果
print("=== 并行模式运行结果 ===")
print(result['combined_output'])

新手小贴士:并行模式的关键是"多个节点同时从START出发",LangGraph会自动处理多节点的并行执行,无需手动写多线程——这也是框架的优势,让新手不用关注底层并发逻辑,只需聚焦业务流程。
总结
- 提示链模式:核心是"分步拆解+门控校验",适合需要多步骤精细化处理的任务,通过串行执行提升输出准确性;
- 路由模式:核心是"智能分类+定向处理",依托结构化输出实现精准分流,适配多场景、多类型的任务处理需求;
- 并行模式:核心是"多节点同步执行+结果聚合",通过并行处理大幅提升复杂任务的效率,新手可优先掌握分段聚合的实现方式。
这三种模式是LangGraph智能体开发的基础,所有代码均经过实测可直接运行,建议新手先跑通案例,再结合自己的业务场景修改提示词和节点逻辑,快速落地大模型智能体应用。
如何学习大模型 AI ?
由于新岗位的生产效率,要优于被取代岗位的生产效率,所以实际上整个社会的生产效率是提升的。
但是具体到个人,只能说是:
“最先掌握AI的人,将会比较晚掌握AI的人有竞争优势”。
这句话,放在计算机、互联网、移动互联网的开局时期,都是一样的道理。
我在一线科技企业深耕十二载,见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事,早已在效率与薪资上形成代际优势,我意识到有很多经验和知识值得分享给大家,也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。我们整理出这套 AI 大模型突围资料包:
- ✅ 从零到一的 AI 学习路径图
- ✅ 大模型调优实战手册(附医疗/金融等大厂真实案例)
- ✅ 百度/阿里专家闭门录播课
- ✅ 大模型当下最新行业报告
- ✅ 真实大厂面试真题
- ✅ 2025 最新岗位需求图谱
所有资料 ⚡️ ,朋友们如果有需要 《AI大模型入门+进阶学习资源包》,下方扫码获取~
① 全套AI大模型应用开发视频教程
(包含提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点)
② 大模型系统化学习路线
作为学习AI大模型技术的新手,方向至关重要。 正确的学习路线可以为你节省时间,少走弯路;方向不对,努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划,带你从零基础入门到精通!
③ 大模型学习书籍&文档
学习AI大模型离不开书籍文档,我精选了一系列大模型技术的书籍和学习文档(电子版),它们由领域内的顶尖专家撰写,内容全面、深入、详尽,为你学习大模型提供坚实的理论基础。
④ AI大模型最新行业报告
2025最新行业报告,针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估,以了解哪些行业更适合引入大模型的技术和应用,以及在哪些方面可以发挥大模型的优势。
⑤ 大模型项目实战&配套源码
学以致用,在项目实战中检验和巩固你所学到的知识,同时为你找工作就业和职业发展打下坚实的基础。
⑥ 大模型大厂面试真题
面试不仅是技术的较量,更需要充分的准备。在你已经掌握了大模型技术之后,就需要开始准备面试,我精心整理了一份大模型面试题库,涵盖当前面试中可能遇到的各种技术问题,让你在面试中游刃有余。

以上资料如何领取?

为什么大家都在学大模型?
最近科技巨头英特尔宣布裁员2万人,传统岗位不断缩减,但AI相关技术岗疯狂扩招,有3-5年经验,大厂薪资就能给到50K*20薪!

不出1年,“有AI项目经验”将成为投递简历的门槛。
风口之下,与其像“温水煮青蛙”一样坐等被行业淘汰,不如先人一步,掌握AI大模型原理+应用技术+项目实操经验,“顺风”翻盘!

这些资料真的有用吗?
这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理,现任上海殷泊信息科技CEO,其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证,服务航天科工、国家电网等1000+企业,以第一作者在IEEE Transactions发表论文50+篇,获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。
资料内容涵盖了从入门到进阶的各类视频教程和实战项目,无论你是小白还是有些技术基础的技术人员,这份资料都绝对能帮助你提升薪资待遇,转行大模型岗位。

以上全套大模型资料如何领取?

更多推荐



所有评论(0)