2026 收藏版｜LangGraph 智能体三大核心工作流，程序员零基础上手大模型开发

本篇全面剖析 2026 主流 LangGraph 智能体三类经典工作流架构依托任务拆分校验、智能任务分发、多任务并行处理三种思路全方位提升大模型智能体运行精度与处理效率。每类模式均搭配可直接运行的实战代码案例贴合新手学习场景帮助编程从业者快速搭建稳定易维护的 Agent 工作体系产出可复用的大模型落地架构模板。一、LangGraph设计模式大模型智能体的架构蓝图1.1 LangGraph设计模式总览建议收藏大模型智能体的能力边界持续拓展面对的业务任务也从简单问答转向复杂多步骤处理。就像盖房子需要施工图、写项目需要架构图一样想要打造稳定、高效的大模型智能体系统一套成熟的LangGraph设计模式是核心支撑。从本篇开始我会从零拆解LangGraph框架下智能体开发的核心架构模式覆盖从基础工作流编排到高阶多智能体协同的全流程不仅讲理论更结合实战代码帮新手和进阶开发者建立一套可复用、可扩展的智能体开发方法论。1.2 LangGraph工作流模式单智能体的核心逻辑首先要掌握的是LangGraph的工作流模式单智能体模式——这里的单智能体并非指只用一个大模型而是系统目标唯一所有决策由一个核心智能体统一把控。一个基础的单智能体架构包含四大核心以DeepSeek、Qwen3等大模型为核心大脑支持工具调用能力具备记忆存储与检索模块同时通过精准的系统提示词承接用户输入。而LangGraph工作流模式正是在这个基础架构上加入任务规划和多步调用机制让智能体能够拆解复杂任务、按规则执行解决单次大模型调用力不从心的问题。常见的工作流模式有提示链、路由、并行化、协调器-工作者、评估器-优化器等本篇先聚焦新手最易上手、落地性最强的三种提示链模式、路由模式、并行化模式。二、提示链模式复杂任务的分步拆解神器2.1 提示链模式定义把大任务拆成小积木提示链是LangGraph最基础也最常用的工作流模式核心逻辑是将一个复杂任务拆解为多个简单、有依赖关系的子步骤让大模型按顺序完成每一步前一步的输出作为后一步的输入形成串行处理的工作流。为了避免流程跑偏提示链模式会在步骤之间加入质量检查站——也就是门控Gate节点。门控会校验上一步输出是否符合预设条件再决定流程是继续执行、转向修正分支还是直接终止从源头提升输出的可控性。提示链模式的核心优势新手必记降低认知负荷将写一篇营销文案这类大任务拆成确定主题→生成初稿→校验内容→优化润色等小步骤大模型单次处理更聚焦提升输出可靠性多步骤精细化处理中间门控校验比单次调用大模型的结果更符合预期减少答非所问易调试易维护每个步骤独立出问题时能快速定位是哪个环节的问题新手也能轻松排查。典型适用场景适合能清晰拆分出顺序子任务的场景比如生成小众语言营销文案中文初稿→门控校验→目标语言翻译、撰写技术文档大纲生成→内容填充→格式校验→最终润色等。2.2 提示链模式代码示例手把手实现智能笑话生成器光讲概念太抽象我们用生成优质笑话这个趣味场景手把手写一个提示链工作流先生成基础笑话→门控检查是否达标→不达标则优化→最后润色全程代码可直接复制运行。前置准备在项目目录新建.env文件填入你的DEEPSEEK_API_KEY没有的话去DeepSeek官网申请新手也能快速搞定。fromtypingimportTypedDictfromlangchain_deepseekimportChatDeepSeekfromlanggraph.graphimportEND,START,StateGraphfromdotenvimportload_dotenv# 0. 配置大模型新手注意确保安装依赖 pip install langchain-deepseek python-dotenv langgraphload_dotenv()# 加载.env文件中的API密钥llmChatDeepSeek(modeldeepseek-chat,# 指定调用的模型版本temperature0.7# 新增调整随机性新手可先固定0.7)步骤1定义工作流状态记录每一步的输出# 1. 定义图状态存储主题、各阶段生成的笑话内容classState(TypedDict):topic:str# 笑话主题用户输入joke:str# 初始生成的笑话improved_joke:str# 优化后的笑话final_joke:str# 最终润色的笑话步骤2定义核心节点函数每个步骤的具体逻辑# 2. 定义节点函数完成各步骤的核心逻辑defgenerate_joke(state:State):第一步根据主题生成初始简短笑话topicstate[topic]# 新增优化提示词让生成的笑话更聚焦promptf写一个简短、易懂的关于{topic}的笑话控制在30字以内msgllm.invoke(prompt)return{joke:msg.content}defcheck_punchline(state:State):门控检查判断笑话是否包含反问/感叹语气提升趣味性jokestate[joke]if?injokeorinjokeor!injokeorinjoke:returnPass# 达标直接结束returnFail# 不达标进入优化环节defimprove_joke(state:State):第二步优化不达标的笑话添加文字游戏jokestate[joke]promptf给这个笑话添加简单的文字游戏提升趣味性保留原主题{joke}msgllm.invoke(prompt)return{improved_joke:msg.content}defpolish_joke(state:State):第三步最终润色添加转折# 兼容逻辑如果有优化后的笑话则用优化版否则用初始版joke_to_polishstate.get(improved_joke,state[joke])promptf为这个笑话添加一个出人意料但不生硬的转折{joke_to_polish}msgllm.invoke(prompt)return{final_joke:msg.content}步骤3构建工作流并运行# 3. 构建提示链工作流workflowStateGraph(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)# 润色→结束# 编译并运行chainworkflow.compile()# 测试生成关于小猫的笑话resultchain.invoke({topic:小猫})# 输出结果新手友好清晰展示每一步print( 提示链模式运行结果 )print(f初始笑话{result[joke]})ifimproved_jokeinresult: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。fromtypingimportTypedDict,Literalfromlangchain_deepseekimportChatDeepSeekfromlanggraph.graphimportStateGraph,START,ENDfromlangchain.messagesimportHumanMessage,SystemMessagefromdotenvimportload_dotenv# 0. 初始化模型新手注意添加超时配置避免调用失败load_dotenv()llmChatDeepSeek(modeldeepseek-chat,timeout30# 新增超时时间新手调试更友好)步骤1定义状态和节点函数# 1. 定义工作流状态classState(TypedDict):input:str# 用户输入的创作需求decision:str# 路由器的分类结果story/joke/poetryoutput:str# 最终生成的内容# 2. 定义专业处理节点每个节点只做一件事defgenerate_story(state:State):故事生成节点专注处理故事创作需求print([路由日志] 进入故事生成流程)promptf根据需求写一篇简短的故事{state[input]}resultllm.invoke(prompt)return{output:result.content}defgenerate_joke(state:State):笑话生成节点专注处理笑话创作需求print([路由日志] 进入笑话生成流程)promptf根据需求写一个轻松的笑话{state[input]}resultllm.invoke(prompt)return{output:result.content}defgenerate_poetry(state:State):诗歌生成节点专注处理诗歌创作需求print([路由日志] 进入诗歌生成流程)promptf根据需求写一首简短的诗歌{state[input]}resultllm.invoke(prompt)return{output:result.content}# 3. 定义路由器节点核心结构化分类classClassification(TypedDict):# 约束输出只能是story/joke/poetry避免分类混乱response_format:Literal[story,joke,poetry]defllm_router(state:State):智能路由器精准分类用户需求# 结构化输出强制大模型按指定格式返回分类结果structured_llmllm.with_structured_output(Classification)system_prompt你是一个精准的内容分类器仅需根据用户输入判断创作类型 输出只能是story故事、joke笑话、poetry诗歌三者中的一个不要添加任何额外内容。responsestructured_llm.invoke([SystemMessage(contentsystem_prompt),HumanMessage(contentstate[input])])return{decision:response[response_format]}步骤2构建路由工作流并运行# 4. 定义路由分支函数defroute_by_decision(state:State):根据分类结果匹配下游节点decision_map{story:llm_story,joke:llm_joke,poetry:llm_poetry}returndecision_map.get(state[decision],llm_joke)# 兜底默认走笑话分支# 5. 构建工作流router_graphStateGraph(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)# 编译并测试workflowrouter_graph.compile()# 测试输入可替换成写一首关于春天的诗、写一个关于程序员的故事等test_input给我写一个关于小猫的笑话resultworkflow.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 并行模式代码示例一键生成主题内容合集我们实现一个主题内容生成器输入一个主题并行生成故事、笑话、诗歌最后自动聚合为合集新手可基于这个框架扩展为多维度内容创作工具。fromtypingimportTypedDictfromlangchain_deepseekimportChatDeepSeekfromlanggraph.graphimportStateGraph,START,ENDfromdotenvimportload_dotenv# 0. 初始化模型load_dotenv()llmChatDeepSeek(modeldeepseek-chat,temperature0.8# 提高随机性让创作内容更丰富)步骤1定义状态和核心节点# 1. 定义工作流状态classState(TypedDict):topic:str# 创作主题joke:str# 并行生成的笑话story:str# 并行生成的故事poetry:str# 并行生成的诗歌combined_output:str# 聚合后的最终合集# 2. 定义并行生成节点三个节点同时运行defgenerate_joke(state:State):笑话生成节点并行promptf写一个轻松有趣的关于{state[topic]}的笑话控制在50字以内return{joke:llm.invoke(prompt).content}defgenerate_story(state:State):故事生成节点并行promptf写一个简短温馨的关于{state[topic]}的小故事控制在100字以内return{story:llm.invoke(prompt).content}defgenerate_poetry(state:State):诗歌生成节点并行promptf写一首关于{state[topic]}的短诗句式简洁有画面感return{poetry:llm.invoke(prompt).content}# 3. 定义聚合节点整合所有并行结果defaggregate_content(state:State):聚合节点将笑话、故事、诗歌整合成合集combinedf ###{state[topic]}主题内容合集 #### 小故事{state[story]}#### 趣味笑话{state[joke]}#### 短诗{state[poetry]}return{combined_output:combined.strip()}步骤2构建并行工作流并运行# 4. 构建并行工作流parallel_graphStateGraph(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)# 编译并运行workflowparallel_graph.compile()# 测试生成关于小猫的内容合集resultworkflow.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大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的技术人员这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。以上全套大模型资料如何领取