【62.Python+AI】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 四大要素

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. 创建状态图workflow=StateGraph(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)# 编译app=workflow.compile()

3 ~> 完整案例：文档审批 Agent

3.1 节点函数实现

fromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model="gpt-4",temperature=0)defclassify_document(state:ApprovalState)->ApprovalState:"""分类节点：判断文档类型"""prompt=f"判断以下文档属于什么类型（合同/报告/其他）：\n{state['document']}"result=llm.invoke(prompt).content state["doc_type"]=result.strip()state["revision_count"]=0returnstatedeflegal_review(state:ApprovalState)->ApprovalState:"""法律审核节点"""prompt=f"""你是法务专员，审核这份合同是否合规。 请给出"通过"或"不通过"，并附审核意见。 合同内容：{state['document']}修改次数：{state['revision_count']}"""result=llm.invoke(prompt).contentif"通过"inresult:state["review_result"]="通过"state["final_status"]="审核通过"else:state["review_result"]="不通过"state["review_comment"]=resultreturnstatedefrevise_document(state:ApprovalState)->ApprovalState:"""修改节点：根据审核意见修改文档"""prompt=f"""根据以下审核意见修改合同： 原合同：{state['document']}审核意见：{state['review_comment']}请输出修改后的合同全文。 """result=llm.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 执行工作流

# 编译并执行app=workflow.compile()result=app.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# 在编译时设置断点app=workflow.compile(interrupt_before=["human_review"]# 执行到此节点前暂停)# 执行到断点处result=app.invoke({"document":"..."})# 人工审核后，手动推进app.update_state(config,{"review_result":"通过"})result=app.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

思考 && 总结

1. LangGraph 解决的是 Chain 解决不了的"状态问题"：当业务逻辑需要循环（驳回修改）、分支（不同文档类型走不同流程）、断点（人工审核），Chain 就力不从心了。
2. State 是核心：定义好 State 的数据结构，就定义好了工作流中能传递什么、能记住什么。State 设计得不好，后面所有节点都会跟着受限。
3. 条件边是灵活性的来源：没有条件边的工作流就是一条死路。有了条件边，同一个输入可以根据不同状态走出完全不同的执行路径。
4. interrupt 让人机协作成为可能：不是所有决策都适合AI来做。LangGraph 的断点机制让你在AI流程中保留人工把关的环节。
5. 先画图再写代码：用 Mermaid 画出状态图，确认流程逻辑没问题之后，再翻译成 LangGraph 的 add_node 和 add_edge 调用。这比对着代码硬想要清晰得多。

LangGraph 把 AI 工作流从"问答机器人"升级到了"业务流程引擎"。如果你的 AI 应用需要处理多步骤的、有状态的业务逻辑，LangGraph 是当前最成熟的选择。

结尾

各位小伙伴，本文的内容到这里就全部结束了，源码骑士在这里再次感谢您的阅读！

源码骑士 — Android Framework & 全栈开发

👀关注：跟博主一起从源码视角深耕底层原理，见证每一次成长

❤️点赞：让优质内容被更多人看见，让知识传递更有力量

⭐收藏：把核心知识点存好，在需要时随时查、随时用

💬评论：分享你的经验或疑问，评论区一起交流避坑

🔄一键四连：不要忘记给博主"一键四连"哦！

🗡️寄语：技术之路难免有困惑，但同行的人会让前进更有方向

结语：Chain 是单行道，LangGraph 是立交桥——能转弯、能掉头、能并线。把本文的审批 Agent 案例在你的项目里跑通一遍，状态图的思想你就掌握了。不要忘记给博主"一键四连"哦！