【收藏】LangGraph工作流深度解析:5种编排模式+子图架构,轻松构建稳定可控的AI系统

发布时间:2026/8/27 18:12:49
【收藏】LangGraph工作流深度解析:5种编排模式+子图架构,轻松构建稳定可控的AI系统 01 前言构建智能系统主要有两种哲学工作流 (Workflows) 与 智能体 (Agents)。1、Workflows 拥有预设的代码路径设计严谨强调执行的顺序和确定性。2、Agents 更加动态由 LLM 自主决定执行流程和工具的使用强调灵活性。文章将深度聚焦于 Workflow工作流模式这是构建复杂 AI 系统的基石。02 编排模式构建生产级应用官方提供了以下五种经过实战检验的架构蓝图。这些模式并非互斥可以在一个复杂系统中组合使用。1、提示词链式调用 (Prompt Chaining)这是最基础的模式。将一个复杂的任务拆解为多个简单的线性步骤。适用场景必须严格遵守先后顺序的任务例如信息提取 - 敏感词过滤 - 格式化输出。2、并行化 (Parallelization)让多个节点同时处理任务最后由一个汇聚节点Aggregator合并结果。适用场景多维度的文档分析、多模型投票机制Model Ensemble。3、路由 (Routing)基于输入的语义意图将控制权引导至不同的下游路径。本质上是一个意图识别或分类任务。适用场景客服系统售前/售后/投诉、多模态任务分发。4、编排者-工作者 (Orchestrator-Worker)这是经典的 Map-Reduce 思想在 Agent 领域的投射。编排者负责拆解任务并动态生成 PlanWorker 负责并行执行最后由编排者汇总。通过 LangGraph 的SendAPI可以动态创建 Worker 节点无需预定义具体的 Worker 数量。适用场景多章节报告生成、批量文档处理。5、评估者-优化者 (Evaluator-Optimizer)这是实现“Agent 自治”的关键。一个节点生成结果另一个节点进行评估如果未通过则带回反馈意见进行重试。适合场景高质量翻译、代码迭代优化。如下是五种模式的对比图为了更直观地理解评估者-优化者模式接下来用一个生成笑话的例子并配置一个自我审查机制确保生成的笑话既简短又好笑。# 安装依赖 # pip install langgraph langchain-openai # langgraph 核心模式评估者-优化者 (Evaluator-Optimizer) # https://docs.langchain.com/oss/python/langgraph/workflows-agents from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage # 0. 配置 DeepSeek 客户端 # 2026年背景DeepSeek API 依然保持极高的性价比 llm ChatOpenAI( modeldeepseek-chat, openai_api_keysk-..., # 请替换为你的 DeepSeek API Key openai_api_basehttps://api.deepseek.com, temperature0.7 ) # 1. 定义状态 (State) # State 是图流转过程中的“上下文” classJokeState(TypedDict): topic: str # 用户输入的主题 joke: str # 生成的笑话 critique: str # 评估意见 iteration_count: int # 当前重试次数 # 2. 定义节点逻辑 (Nodes) defgenerate_joke(state: JokeState): 生成器节点负责创作或修改笑话 topic state[topic] iteration state.get(iteration_count, 0) critique state.get(critique) if critique: print(f\n--- 第 {iteration} 次重试根据意见优化:{critique} ---) prompt f你讲了一个关于{topic}的笑话但收到了以下反馈{critique}。\n请根据反馈改进这个笑话。 else: print(f\n--- 初次生成 ---) prompt f讲一个关于 {topic} 的幽默笑话要求简短有力。 # 调用 DeepSeek response llm.invoke([HumanMessage(contentprompt)]) return { joke: response.content, iteration_count: iteration 1 } defevaluate_joke(state: JokeState): 评估器节点负责质量检查 joke state[joke] # 这里演示简单的规则评估生产环境可以换成另一个 LLM 进行评分 # 规则必须包含哈哈且长度不能超过 100 字 iflen(joke) 100: return {critique: 笑话太长了请缩减到 50 字以内。} elif哈哈notin joke and嘿嘿notin joke: # 为了演示循环强制要求笑话里要有笑声词 return {critique: 笑话不够生动请在结尾加上哈哈或嘿嘿。} return {critique: PASS} # 3. 定义条件边逻辑 (Edges) defcheck_quality(state: JokeState) - Literal[end, retry]: 路由逻辑决定是结束还是重试 critique state[critique] count state[iteration_count] # 退出条件评估通过 OR 超过 3 次重试防止死循环 if critique PASSor count 3: if count 3: print(--- 达到最大重试次数强制结束 ---) returnend returnretry # 4. 构建图 (Graph Construction) workflow StateGraph(JokeState) # 添加节点 workflow.add_node(generator, generate_joke) workflow.add_node(evaluator, evaluate_joke) # 添加边控制流 workflow.add_edge(START, generator) # 启动 - 生成 workflow.add_edge(generator, evaluator) # 生成 - 评估 # 添加条件边 # 从 evaluator 出发根据 check_quality 的返回值路由到不同节点 workflow.add_conditional_edges( evaluator, check_quality, { end: END, retry: generator # 闭环回流 } ) # 编译图 app workflow.compile() # 5. 执行工作流 if __name__ __main__: # 初始化输入 initial_input {topic: 程序员与AI, iteration_count: 0} print(f开始执行 Workflow主题{initial_input[topic]}) # invoke 触发执行 final_state app.invoke(initial_input) print(\n * 30) print(f最终结果:\n{final_state[joke]}) print(f总共迭代次数: {final_state[iteration_count]}) print( * 30)示例先定义OpenAI兼容API使用DeepSeek模型并指定相关参数然后定义State状态记录执行过程中的状态包括用户输入随后定义生成器和评估器节点以及条件边逻辑再通过StateGraph构建workflow指定节点和边编译后通过invoke并传入参数进行调用。示例输出效果如下所示最后省略了最终的输出主要看下过程开始执行 Workflow主题程序员与AI --- 初次生成 --- --- 第 1 次重试根据意见优化:笑话不够生动请在结尾加上哈哈或嘿嘿。 --- --- 第 2 次重试根据意见优化:笑话太长了请缩减到 50 字以内。 --- --- 达到最大重试次数强制结束 --- 最终结果: xxxxxx03 进阶-子图Subgraphs当我们的 Graph 变得越来越复杂节点数量超过 20 个时维护一个单体的大图Monolithic Graph就是一场噩梦。就像微服务架构拆分单体应用一样LangGraph 提供了Subgraphs子图机制来实现模块化。1、为什么需要子图呢状态隔离子图可以拥有自己独立的 State Schema只处理局部逻辑避免父图的 Global State 过于臃肿。团队协作不同的开发小组可以负责不同的子图例如一个组做“搜索Agent”一个组做“写作Agent”最后集成到一个父图中。分形设计子图本身也可以包含子图理论上支持无限层级嵌套。2、架构设计在 LangGraph 中每一个编译好的Graph都可以直接作为另一个 Graph 的节点。例如下面的示例用户指令发给“项目经理父图”项目经理将其指派给“编码专家子图”或“测试专家子图”。3、父图与子图有两种集成方式直接添加为节点 如果父子图共享状态键如共享 messages可以直接将编译后的子图添加到父图中。作为节点函数调用在父图节点中通过 subgraph.invoke() 显式调用需手动转换状态。更推荐此方式使得父子图解耦。# 代码示例片段 # 方式一直接添加需要状态Key一致 # joke_subgraph是编译后的子图 # main_builder.add_node(joke_maker, joke_subgraph) # 方式二使用包装函数推荐解耦更强 # 演示通过 invoke 调用子图这样可以灵活处理输入输出映射 def invoke_joke_graph(state: MainState): # 1. 准备子图输入 sub_input {topic: state[user_query]} # 2. 调用子图 sub_result joke_subgraph.invoke(sub_input) # 3. 映射回父图状态 return { content: sub_result[joke], logs: [执行子图完毕]4、使用上的几点说明如果父图定义了checkpointer实现记忆那么默认会自动传递到子图中除非重新定义子图checkpointer。如果共享同一个checkpointer子图中断了可以通过graph.get_state(config, subgraphsTrue)指定subgraphsTrue查看子图状态。如果需要子图的内容也是流式输出的可以在调用父图stream的时候指定subgraphsTrue。04 总结通过Workflow 模式如评估者-优化者我们解决了 LLM 输出不稳定的问题通过Subgraphs子图我们解决了复杂系统代码难维护的问题。掌握 LangGraph本质上是掌握了一种用代码定义 AI 思考路径的语言。通过将如 DeepSeek 强大的推理能力装入严谨的图结构中可以构建出既具备创造力又安全可控的智能体。最后唠两句为什么AI大模型成为越来越多程序员转行就业、升职加薪的首选很简单这些岗位缺人且高薪智联招聘的最新数据给出了最直观的印证2025年2月AI领域求职人数同比增幅突破200% 远超其他行业平均水平整个人工智能行业的求职增速达到33.4%位居各行业榜首其中人工智能工程师岗位的求职热度更是飙升69.6%。AI产业的快速扩张也让人才供需矛盾愈发突出。麦肯锡报告明确预测到2030年中国AI专业人才需求将达600万人人才缺口可能高达400万人这一缺口不仅存在于核心技术领域更蔓延至产业应用的各个环节。那0基础普通人如何学习大模型 深耕科技一线十二载亲历技术浪潮变迁。我见证那些率先拥抱AI的同行如何建立起效率与薪资的代际优势。如今我将积累的大模型面试真题、独家资料、技术报告与实战路线系统整理分享于此为你扫清学习困惑共赴AI时代新程。我整理出这套 AI 大模型突围资料包【允许白嫖】✅从入门到精通的全套视频教程✅AI大模型学习路线图0基础到项目实战仅需90天✅大模型书籍与技术文档PDF✅各大厂大模型面试题目详解✅640套AI大模型报告合集✅大模型入门实战训练这份完整版的大模型 AI 学习和面试资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】①从入门到精通的全套视频教程包含提示词工程、RAG、Agent等技术点② AI大模型学习路线图0基础到项目实战仅需90天全过程AI大模型学习路线③学习电子书籍和技术文档市面上的大模型书籍确实太多了这些是我精选出来的④各大厂大模型面试题目详解⑤640套AI大模型报告合集⑥大模型入门实战训练如果说你是以下人群中的其中一类都可以来智泊AI学习人工智能找到高薪工作一次小小的“投资”换来的是终身受益应届毕业生‌无工作经验但想要系统学习AI大模型技术期待通过实战项目掌握核心技术。零基础转型‌非技术背景但关注AI应用场景计划通过低代码工具实现“AI行业”跨界‌。业务赋能 ‌突破瓶颈传统开发者Java/前端等学习Transformer架构与LangChain框架向AI全栈工程师转型‌。获取方式有需要的小伙伴可以保存图片到wx扫描二v码免费领取【保证100%免费】