62-LangGraph有状态AI工作流-状态图-条件边-审批Agent实现

文章目录【62.PythonAI】LangGraph实战用状态图构建有状态的AI工作流导入语1 ~ 为什么需要状态图1.1 LangChain Chain 的局限1.2 真实业务需要什么2 ~ LangGraph 核心概念2.1 四大要素2.2 State 的定义2.3 状态图构建五步法3 ~ 完整案例文档审批 Agent3.1 节点函数实现3.2 完整工作流图3.3 执行工作流4 ~ 高级特性4.1 人工审核节点4.2 并行执行5 ~ LangGraph vs LangChain Chain思考 总结结尾【62.PythonAI】LangGraph实战用状态图构建有状态的AI工作流文章简介本文系统讲解LangGraph——LangChain生态中专门用于构建有状态AI工作流的框架。文章从DAG不够用了的痛点出发引出状态图StateGraph的核心概念节点Node、边Edge、条件边Conditional Edge、循环节点的设计哲学。通过一个完整的文档审批Agent案例逐行拆解状态定义、节点函数编写、条件路由和人工审核节点的实现并配有Mermaid状态机图展示Flow转换过程。适合已经用过LangChain做简单链式调用、想要构建多步骤、有分支和循环的复杂AI工作流的开发者。 个人主页源码骑士❄专栏传送门《Android开发基础》《python基础课程》⭐️热衷从源码视角拆解技术底层原理将复杂架构讲得通俗易懂 源码骑士的简介5年Android Framework系统开发经验曾主导多项系统级性能优化专项技术栈覆盖Android系统全链路Binder/Handler/AMS/WMS/启动流程及Java后端全家桶Spring MyBatis Redis Oracle累计产出原创技术文章100篇文章以流程图为特色被读者评价为看一篇胜过啃一周源码导入语LangChain 的 Chain 让你能把几个步骤串起来——先检索、再总结、最后翻译一条直线走到底。但真实的业务场景很少有一条直线这么简单。用户提交一个文档你需要先判断文档类型 → 如果是合同就走法律审核流程 → 审核不通过退回修改 → 修改后重新审核 → 最终归档。这里面有分支、有循环、有状态保持——这已经不是一条直线能搞定的事了。这就是 LangGraph 要解决的问题把AI工作流从一次性的直线调用升级为有状态的、带分支和循环的图执行。看完这篇文章你会用 LangGraph 画出自己的第一个状态图工作流。1 ~ 为什么需要状态图1.1 LangChain Chain 的局限LangChain Chain 的执行模式 输入 →[步骤1]→[步骤2]→[步骤3]→ 输出 ↑ 直线没有回头路 ↑Chain 本质是一个有向无环图DAG。数据从入口流向出口每步只能走一次不能回头、不能循环。1.2 真实业务需要什么合同报告通过不通过接收文档文档类型?法律审核内容审核审核通过?归档退回修改这个流程中G → C是一个循环——被驳回的文档要重新进入审核。Chain 做不到这一点LangGraph 可以。2 ~ LangGraph 核心概念2.1 四大要素概念说明类比State状态定义工作流在每一步之间传递的数据结构快递包裹上的运单信息Node节点执行具体逻辑的函数快递中转站Edge边固定路由——从节点A直接到节点B固定运输路线Conditional Edge条件边根据状态决定下一步去哪根据目的地选择不同路线2.2 State 的定义State 是整个工作流共享的数据对象每个节点都可以读写它fromtypingimportTypedDict,ListclassApprovalState(TypedDict):document:str# 原始文档内容doc_type:str# 文档类型合同/报告/其他review_result:str# 审核结果通过/不通过review_comment:str# 审核意见revision_count:int# 修改次数final_status:str# 最终状态2.3 状态图构建五步法fromlanggraph.graphimportStateGraph,END# 1. 创建状态图workflowStateGraph(ApprovalState)# 2. 添加节点每个节点是一个函数workflow.add_node(classify,classify_document)# 分类workflow.add_node(legal_review,legal_review)# 法律审核workflow.add_node(content_review,content_review)# 内容审核workflow.add_node(revise,revise_document)# 修改# 3. 设置入口workflow.set_entry_point(classify)# 4. 添加条件边workflow.add_conditional_edges(classify,route_by_type,# 路由函数{合同:legal_review,报告:content_review,})workflow.add_conditional_edges(legal_review,decide_next,{通过:END,不通过:revise,})# 5. 添加固定边workflow.add_edge(revise,legal_review)# 循环回去workflow.add_edge(content_review,END)# 编译appworkflow.compile()3 ~ 完整案例文档审批 Agent3.1 节点函数实现fromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4,temperature0)defclassify_document(state:ApprovalState)-ApprovalState:分类节点判断文档类型promptf判断以下文档属于什么类型合同/报告/其他\n{state[document]}resultllm.invoke(prompt).content state[doc_type]result.strip()state[revision_count]0returnstatedeflegal_review(state:ApprovalState)-ApprovalState:法律审核节点promptf你是法务专员审核这份合同是否合规。 请给出通过或不通过并附审核意见。 合同内容{state[document]}修改次数{state[revision_count]}resultllm.invoke(prompt).contentif通过inresult:state[review_result]通过state[final_status]审核通过else:state[review_result]不通过state[review_comment]resultreturnstatedefrevise_document(state:ApprovalState)-ApprovalState:修改节点根据审核意见修改文档promptf根据以下审核意见修改合同 原合同{state[document]}审核意见{state[review_comment]}请输出修改后的合同全文。 resultllm.invoke(prompt).content state[document]result state[revision_count]1returnstatedefroute_by_type(state:ApprovalState)-str:根据文档类型路由returnstate[doc_type]defdecide_next(state:ApprovalState)-str:审核后决定下一步ifstate[review_result]通过:return通过# 最多修改3次超过就强制通过ifstate[revision_count]3:state[final_status]超过最大修改次数强制通过return通过return不通过3.2 完整工作流图合同报告通过不通过开始分类节点法律审核内容审核结束-归档修改节点3.3 执行工作流# 编译并执行appworkflow.compile()resultapp.invoke({document:甲方与乙方就软件开发项目达成以下协议...,})print(f审核结果{result[final_status]})print(f修改次数{result[revision_count]})print(f审核意见{result[review_comment]})4 ~ 高级特性4.1 人工审核节点不是所有审核都可以交给AI。LangGraph 支持在流程中插入interrupt断点让人工介入defhuman_review(state:ApprovalState)-ApprovalState:等待人工审核的节点设置断点# 这个节点什么也不做等待外部的 update_state 调用returnstate# 在编译时设置断点appworkflow.compile(interrupt_before[human_review]# 执行到此节点前暂停)# 执行到断点处resultapp.invoke({document:...})# 人工审核后手动推进app.update_state(config,{review_result:通过})resultapp.invoke(None,config)# 继续执行4.2 并行执行多个审核节点同时进行最后汇总结果fromlanggraph.graphimportStateGraph# 并行审核workflow.add_node(legal_check,legal_check)workflow.add_node(compliance_check,compliance_check)# 两个审核并行执行workflow.add_edge(classify,legal_check)workflow.add_edge(classify,compliance_check)# 汇聚节点两个都完成后才执行workflow.add_node(merge_results,merge)workflow.add_edge(legal_check,merge_results)workflow.add_edge(compliance_check,merge_results)5 ~ LangGraph vs LangChain Chain对比维度ChainLangGraph执行模式直线/顺序图支持分支和循环状态管理无数据流过即丢有State贯穿始终断点恢复不支持支持interrupt update_state复杂度天花板低高学习曲线平缓稍陡适合场景简单流水线复杂业务流程思考 总结LangGraph 解决的是 Chain 解决不了的状态问题当业务逻辑需要循环驳回修改、分支不同文档类型走不同流程、断点人工审核Chain 就力不从心了。State 是核心定义好 State 的数据结构就定义好了工作流中能传递什么、能记住什么。State 设计得不好后面所有节点都会跟着受限。条件边是灵活性的来源没有条件边的工作流就是一条死路。有了条件边同一个输入可以根据不同状态走出完全不同的执行路径。interrupt 让人机协作成为可能不是所有决策都适合AI来做。LangGraph 的断点机制让你在AI流程中保留人工把关的环节。先画图再写代码用 Mermaid 画出状态图确认流程逻辑没问题之后再翻译成 LangGraph 的 add_node 和 add_edge 调用。这比对着代码硬想要清晰得多。LangGraph 把 AI 工作流从问答机器人升级到了业务流程引擎。如果你的 AI 应用需要处理多步骤的、有状态的业务逻辑LangGraph 是当前最成熟的选择。结尾各位小伙伴本文的内容到这里就全部结束了源码骑士在这里再次感谢您的阅读源码骑士 — Android Framework 全栈开发关注跟博主一起从源码视角深耕底层原理见证每一次成长❤️点赞让优质内容被更多人看见让知识传递更有力量⭐收藏把核心知识点存好在需要时随时查、随时用评论分享你的经验或疑问评论区一起交流避坑一键四连不要忘记给博主一键四连哦️寄语技术之路难免有困惑但同行的人会让前进更有方向结语Chain 是单行道LangGraph 是立交桥——能转弯、能掉头、能并线。把本文的审批 Agent 案例在你的项目里跑通一遍状态图的思想你就掌握了。不要忘记给博主一键四连哦