【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 四大要素

概念	说明	类比
State（状态）	定义工作流在每一步之间传递的数据结构	快递包裹上的运单信息
Node（节点）	执行具体逻辑的函数	快递中转站
Edge（边）	固定路由——从节点A直接到节点B	固定运输路线
Conditional Edge（条件边）	根据状态决定下一步去哪	根据目的地选择不同路线

2.2 State 的定义

State 是整个工作流共享的数据对象，每个节点都可以读写它：

from typing import TypedDict, List

class ApprovalState(TypedDict):
    document: str           # 原始文档内容
    doc_type: str           # 文档类型：合同/报告/其他
    review_result: str      # 审核结果：通过/不通过
    review_comment: str     # 审核意见
    revision_count: int     # 修改次数
    final_status: str       # 最终状态

2.3 状态图构建五步法

from langgraph.graph import StateGraph, 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 节点函数实现

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4", temperature=0)

def classify_document(state: ApprovalState) -> ApprovalState:
    """分类节点：判断文档类型"""
    prompt = f"判断以下文档属于什么类型（合同/报告/其他）：\n{state['document']}"
    result = llm.invoke(prompt).content
    
    state["doc_type"] = result.strip()
    state["revision_count"] = 0
    return state


def legal_review(state: ApprovalState) -> ApprovalState:
    """法律审核节点"""
    prompt = f"""你是法务专员，审核这份合同是否合规。
    请给出"通过"或"不通过"，并附审核意见。
    
    合同内容：{state['document']}
    修改次数：{state['revision_count']}
    """
    result = llm.invoke(prompt).content
    
    if "通过" in result:
        state["review_result"] = "通过"
        state["final_status"] = "审核通过"
    else:
        state["review_result"] = "不通过"
    
    state["review_comment"] = result
    return state


def revise_document(state: ApprovalState) -> ApprovalState:
    """修改节点：根据审核意见修改文档"""
    prompt = f"""根据以下审核意见修改合同：
    
    原合同：{state['document']}
    审核意见：{state['review_comment']}
    
    请输出修改后的合同全文。
    """
    result = llm.invoke(prompt).content
    
    state["document"] = result
    state["revision_count"] += 1
    return state


def route_by_type(state: ApprovalState) -> str:
    """根据文档类型路由"""
    return state["doc_type"]


def decide_next(state: ApprovalState) -> str:
    """审核后决定下一步"""
    if state["review_result"] == "通过":
        return "通过"
    
    # 最多修改3次，超过就强制通过
    if state["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 断点，让人工介入：

def human_review(state: ApprovalState) -> ApprovalState:
    """等待人工审核的节点（设置断点）"""
    # 这个节点什么也不做，等待外部的 update_state 调用
    return state


# 在编译时设置断点
app = workflow.compile(
    interrupt_before=["human_review"]   # 执行到此节点前暂停
)

# 执行到断点处
result = app.invoke({"document": "..."})

# 人工审核后，手动推进
app.update_state(config, {"review_result": "通过"})
result = app.invoke(None, config)   # 继续执行

4.2 并行执行

多个审核节点同时进行，最后汇总结果：

from langgraph.graph import StateGraph

# 并行审核
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

对比维度	Chain	LangGraph
执行模式	直线/顺序	图（支持分支和循环）
状态管理	无（数据流过即丢）	有（State贯穿始终）
断点恢复	不支持	支持（interrupt + update_state）
复杂度天花板	低	高
学习曲线	平缓	稍陡
适合场景	简单流水线	复杂业务流程

---

思考 && 总结

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 案例在你的项目里跑通一遍，状态图的思想你就掌握了。不要忘记给博主"一键四连"哦！