️ LangGraph 工作流引擎底层笔记 (2026版)一、 LangGraph 的定位如果说 LangChain 的create_agent是组装好的自动挡汽车给它工具让它自己跑那么 LangGraph 就是一套工业级流水线/状态机框架。它赋予开发者极其精细的控制力让你能清晰地绘制出一张包含条件分支、循环重试、多智能体交互的流程图。实际上create_agent底层100%也是由一张简化的 LangGraph 图构成的。二、 图的三大核心组件 (State, Node, Edge)1. State (状态数据)State 是这张图生命周期内的全局共享内存承载了数据从开始到结束的所有流转变更。它必须被定义为一个TypedDict。fromtypingimportTypedDict,Annotatedfromlanggraph.graphimportadd_messagesclassCustomerState(TypedDict):# 标注了 add_messages 的字段如果有多个节点写入会执行 追加操作 而不是 覆盖操作messages:Annotated[list,add_messages]# 普通字段后一个节点写入会覆盖前一个节点的数据category:str# 记录分类技术/退款/闲聊quality_score:float# 记录文章打分2. Node (业务节点)Node 是定义了当前这步具体要干什么的普通 Python 函数。标准模式接收完整的 State 数据 - 执行逻辑 - 返回一个只包含需要更新字段的字典。defclassify_intent(state:CustomerState)-dict:user_msgstate[messages][-1].contentreturn{category:tech}# 通知图只更新 category 字段3. Edge (边流向)用来定义节点间的流转顺序规则。普通边 (Edge)固定写死A - B必定走。入口/出口指明这张图从哪里START进入走向END。条件边 (Conditional Edge)一个路由函数根据当前 State 的内容动态跳转。三、 五大核心路由模式架构LangGraph 极其灵活几乎能拟合人类处理任务的所有架构模式。模式一线性流水线 (Pipeline)最简单的结构提取数据 - 翻译成中文 - 总结要点 - END。模式二条件分发 (Decision Tree)用户的请求进来后经过一个大脑裁判 Node 根据业务逻辑路由给后面的专项处理 Node典型的多意图客服分流。模式三反射与自我纠错循环 (Reflection/Self-Critique Loop)这是保证 AI 最终输出质量最靠谱的生产级策略。START - 撰写节点 - 评估节点 - 判断边若不够好走向 - 修改建议节点 - 回流撰写节点构成 Loop模式四并行聚合 (Parallel Map-Reduce)同时派发给不同源抓取最终汇聚总结。模式五多 Agent 主系统 (Supervisor Delegation)START - 经理 Agent - 派单写手与研发 - 逐个回收审查 - END。四、 完整实战代码示例 (客服自动工单分流图)这个示例纯手工搓一张 LangGraph它接收一段客诉文本第一步使用大模型对意图进行提取第二步通过 Conditional Edge 判断进入技术支持节点或普通安抚节点。fromtypingimportTypedDict,Annotated,Literalfromlanggraph.graphimportStateGraph,START,END,add_messagesfromlangchain_core.messagesimportHumanMessage,AIMessagefromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4o)# 1. 定义图共享的全局数据结构 State classTicketState(TypedDict):messages:Annotated[list,add_messages]intent:str# 记录用户的意图分类 (tech 或 other)response:str# 最终要回复给用户的文本# 2. 定义纯函数的执行节点 (Nodes) defintent_analyzer(state:TicketState)-dict:裁判节点分析最后一条用户消息意图user_msgstate[messages][-1].content# 强制让大模型以特定词汇返回分类resultllm.invoke([{role:system,content:将客诉分为两类涉及软件报错、闪退、硬件故障返回 tech其余的退款、辱骂、闲聊返回 other。只输出英文单字。},{role:user,content:user_msg}])categoryresult.content.strip().lower()return{intent:category}# 写入状态defhandle_tech_issue(state:TicketState)-dict:专职技术客服节点responsellm.invoke([{role:system,content:你是高级运维工程师请给出详细的排障命令或方案。},{role:user,content:state[messages][-1].content}])return{response:response.content}# 写入最终回复给状态defhandle_other_issue(state:TicketState)-dict:普通在线客服节点responsellm.invoke([{role:system,content:你是情感安抚客服请用官方且柔和的口吻先道歉然后记录工单。},{role:user,content:state[messages][-1].content}])return{response:response.content}# 写入最终回复给状态# 3. 定义图内的条件动态路牌 (Conditional Edge) defroute_to_handler(state:TicketState)-Literal[handle_tech_issue,handle_other_issue]:ifstate[intent]tech:returnhandle_tech_issuereturnhandle_other_issue# 4. 组装整张图 (Build Graph) graphStateGraph(TicketState)# 先把定义好的三个动作挂载进图网络graph.add_node(intent_analyzer,intent_analyzer)graph.add_node(handle_tech_issue,handle_tech_issue)graph.add_node(handle_other_issue,handle_other_issue)# 告诉图开始节点是意图分析graph.add_edge(START,intent_analyzer)# 从意图分析节点开始加一个条件判断路牌接下来的走向graph.add_conditional_edges(intent_analyzer,route_to_handler)# 无论最后走了哪个子处理客服完事后全部拉回大门 END 结束本次执行graph.add_edge(handle_tech_issue,END)graph.add_edge(handle_other_issue,END)# 编译成可运行应用appgraph.compile()# 5. 测试运行 defsubmit_ticket(query:str):print(f\n[用户工单]:{query})resultapp.invoke({messages:[HumanMessage(contentquery)]})print(f- 判决流向: 被分发到了 [{result[intent]}] 客服)print(f- 客服回复:{result[response]})# 测试用例 1 会走 handle_other_issue 节点submit_ticket(你们怎么搞的昨天买的为什么现在发货还没消息我要投诉)# 测试用例 2 会走 handle_tech_issue 节点submit_ticket(我本地的 mysql 启动一直报 error 1045 access denied 怎么整。)
LangGraph 工作流引擎底层笔记
️ LangGraph 工作流引擎底层笔记 (2026版)一、 LangGraph 的定位如果说 LangChain 的create_agent是组装好的自动挡汽车给它工具让它自己跑那么 LangGraph 就是一套工业级流水线/状态机框架。它赋予开发者极其精细的控制力让你能清晰地绘制出一张包含条件分支、循环重试、多智能体交互的流程图。实际上create_agent底层100%也是由一张简化的 LangGraph 图构成的。二、 图的三大核心组件 (State, Node, Edge)1. State (状态数据)State 是这张图生命周期内的全局共享内存承载了数据从开始到结束的所有流转变更。它必须被定义为一个TypedDict。fromtypingimportTypedDict,Annotatedfromlanggraph.graphimportadd_messagesclassCustomerState(TypedDict):# 标注了 add_messages 的字段如果有多个节点写入会执行 追加操作 而不是 覆盖操作messages:Annotated[list,add_messages]# 普通字段后一个节点写入会覆盖前一个节点的数据category:str# 记录分类技术/退款/闲聊quality_score:float# 记录文章打分2. Node (业务节点)Node 是定义了当前这步具体要干什么的普通 Python 函数。标准模式接收完整的 State 数据 - 执行逻辑 - 返回一个只包含需要更新字段的字典。defclassify_intent(state:CustomerState)-dict:user_msgstate[messages][-1].contentreturn{category:tech}# 通知图只更新 category 字段3. Edge (边流向)用来定义节点间的流转顺序规则。普通边 (Edge)固定写死A - B必定走。入口/出口指明这张图从哪里START进入走向END。条件边 (Conditional Edge)一个路由函数根据当前 State 的内容动态跳转。三、 五大核心路由模式架构LangGraph 极其灵活几乎能拟合人类处理任务的所有架构模式。模式一线性流水线 (Pipeline)最简单的结构提取数据 - 翻译成中文 - 总结要点 - END。模式二条件分发 (Decision Tree)用户的请求进来后经过一个大脑裁判 Node 根据业务逻辑路由给后面的专项处理 Node典型的多意图客服分流。模式三反射与自我纠错循环 (Reflection/Self-Critique Loop)这是保证 AI 最终输出质量最靠谱的生产级策略。START - 撰写节点 - 评估节点 - 判断边若不够好走向 - 修改建议节点 - 回流撰写节点构成 Loop模式四并行聚合 (Parallel Map-Reduce)同时派发给不同源抓取最终汇聚总结。模式五多 Agent 主系统 (Supervisor Delegation)START - 经理 Agent - 派单写手与研发 - 逐个回收审查 - END。四、 完整实战代码示例 (客服自动工单分流图)这个示例纯手工搓一张 LangGraph它接收一段客诉文本第一步使用大模型对意图进行提取第二步通过 Conditional Edge 判断进入技术支持节点或普通安抚节点。fromtypingimportTypedDict,Annotated,Literalfromlanggraph.graphimportStateGraph,START,END,add_messagesfromlangchain_core.messagesimportHumanMessage,AIMessagefromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4o)# 1. 定义图共享的全局数据结构 State classTicketState(TypedDict):messages:Annotated[list,add_messages]intent:str# 记录用户的意图分类 (tech 或 other)response:str# 最终要回复给用户的文本# 2. 定义纯函数的执行节点 (Nodes) defintent_analyzer(state:TicketState)-dict:裁判节点分析最后一条用户消息意图user_msgstate[messages][-1].content# 强制让大模型以特定词汇返回分类resultllm.invoke([{role:system,content:将客诉分为两类涉及软件报错、闪退、硬件故障返回 tech其余的退款、辱骂、闲聊返回 other。只输出英文单字。},{role:user,content:user_msg}])categoryresult.content.strip().lower()return{intent:category}# 写入状态defhandle_tech_issue(state:TicketState)-dict:专职技术客服节点responsellm.invoke([{role:system,content:你是高级运维工程师请给出详细的排障命令或方案。},{role:user,content:state[messages][-1].content}])return{response:response.content}# 写入最终回复给状态defhandle_other_issue(state:TicketState)-dict:普通在线客服节点responsellm.invoke([{role:system,content:你是情感安抚客服请用官方且柔和的口吻先道歉然后记录工单。},{role:user,content:state[messages][-1].content}])return{response:response.content}# 写入最终回复给状态# 3. 定义图内的条件动态路牌 (Conditional Edge) defroute_to_handler(state:TicketState)-Literal[handle_tech_issue,handle_other_issue]:ifstate[intent]tech:returnhandle_tech_issuereturnhandle_other_issue# 4. 组装整张图 (Build Graph) graphStateGraph(TicketState)# 先把定义好的三个动作挂载进图网络graph.add_node(intent_analyzer,intent_analyzer)graph.add_node(handle_tech_issue,handle_tech_issue)graph.add_node(handle_other_issue,handle_other_issue)# 告诉图开始节点是意图分析graph.add_edge(START,intent_analyzer)# 从意图分析节点开始加一个条件判断路牌接下来的走向graph.add_conditional_edges(intent_analyzer,route_to_handler)# 无论最后走了哪个子处理客服完事后全部拉回大门 END 结束本次执行graph.add_edge(handle_tech_issue,END)graph.add_edge(handle_other_issue,END)# 编译成可运行应用appgraph.compile()# 5. 测试运行 defsubmit_ticket(query:str):print(f\n[用户工单]:{query})resultapp.invoke({messages:[HumanMessage(contentquery)]})print(f- 判决流向: 被分发到了 [{result[intent]}] 客服)print(f- 客服回复:{result[response]})# 测试用例 1 会走 handle_other_issue 节点submit_ticket(你们怎么搞的昨天买的为什么现在发货还没消息我要投诉)# 测试用例 2 会走 handle_tech_issue 节点submit_ticket(我本地的 mysql 启动一直报 error 1045 access denied 怎么整。)
