万字长文拆解LangGraph：状态机模式在复杂Agent工作流中的应用

关键词：LangGraph、大语言模型Agent、有限状态机、工作流编排、多Agent协作、LLM应用开发、全局状态管理

摘要：随着大语言模型应用从单轮对话、简单RAG向复杂多Agent系统演进，传统的顺序链（Chain）模式已经无法满足循环、回退、多节点同步等复杂流程需求。本文从生活场景出发，用通俗易懂的语言拆解LangGraph的核心原理——有限状态机模式，从概念、原理、数学模型、项目实战到最佳实践，完整讲解如何用LangGraph构建可控、可扩展、可维护的复杂Agent工作流。读完本文你将能独立完成多Agent协作系统的设计与开发，解决当前LLM应用开发中流程不可控、状态不同步的核心痛点。

---

背景介绍

为什么我们需要LangGraph？

相信很多做过LLM应用开发的同学都遇到过这样的痛点：

1. 用LangChain的顺序链做需求评审机器人，产品提完需求后，开发评审发现需求有问题，想要退回给产品修改，可顺序链是单向的，根本做不到回退；
2. 做多Agent写作系统，大纲Agent写完大纲，内容Agent写内容，校对Agent发现内容不符合大纲要求，想要让内容Agent重写，可传统Chain没有循环能力，只能全流程重新跑一遍；
3. 做客服机器人，用户问了一个技术问题，分类节点分到技术线，技术Agent解答后用户不满意，想要转人工，转完人工还要同步之前的对话历史给客服，可每个Chain的上下文是独立的，同步状态非常麻烦。

这些问题的本质是：复杂任务的流程不是线性的，而是有循环、回退、分支、中断的状态转移过程，传统的线性链模式天生不适合处理这类问题。

LangGraph就是为了解决这个问题诞生的：它把有限状态机（FSM）的思想和LLM Agent结合，让开发者可以自定义任意的流程跳转规则，内置全局状态同步，天然支持循环、回退、多Agent协作，是当前构建复杂Agent系统的最佳方案。

目的和范围

本文将从0到1完整讲解LangGraph的核心原理与落地实践：

• 理解有限状态机的核心思想，以及为什么它适合编排Agent工作流
• 掌握LangGraph的三大核心要素：节点、边、全局状态的设计与使用
• 实战开发一个多Agent需求评审系统，包含流程编排、状态持久化、人机中断等生产级特性
• 掌握LangGraph的最佳实践，避免常见的死循环、状态覆盖、性能问题

本文不要求你有深厚的Agent开发经验，只要有基础的Python和LangChain使用经验就能看懂。

预期读者

• LLM应用开发工程师、AI产品经理
• 想要构建复杂多Agent系统的开发者
• 对LangChain生态感兴趣的技术人员

术语表

核心术语定义

1. 有限状态机（FSM）：一种计算模型，包含有限个状态，在不同的输入下会按照 predefined 的规则在不同状态之间转移
2. LangGraph：LangChain团队推出的基于状态机的Agent工作流编排框架
3. 节点（Node）：LangGraph中的执行单元，通常是一个Agent或者一个工具调用逻辑
4. 边（Edge）：LangGraph中的流转规则，定义了节点之间的跳转关系
5. 全局状态（State）：LangGraph中所有节点共享的数据存储，用来同步流程中的所有信息
6. Agent工作流：多个Agent按照一定的规则协作完成复杂任务的流程

缩略词列表

缩略词	全称	含义
LLM	Large Language Model	大语言模型
FSM	Finite State Machine	有限状态机
RAG	Retrieval Augmented Generation	检索增强生成
DAG	Directed Acyclic Graph	有向无环图

---

核心概念与联系

故事引入：奶茶店的工作流就是完美的状态机

我们先来想象一个你每天都可能接触的场景：奶茶店的运营流程。

1. 你走到前台，点单员接待你，问你要喝什么、糖度冰度、要不要加小料，确认后给你一个取餐码，把订单信息同步给后厨——这就是点单节点，订单信息就是全局状态的一部分；
2. 如果你点了加珍珠、椰果，订单就会流转到加料员那里，加料员按照订单要求加完小料，更新订单状态，再流转到制茶员——这就是条件边：有小料就走加料节点，没有就直接去制茶节点；
3. 制茶员做完奶茶，品管尝了一口发现糖放多了，就让制茶员重新做——这就是循环边：不符合要求就回到上一个节点重新执行；
4. 如果奶茶符合要求，就放到取餐台，叫号给你——这就是结束节点；
5. 如果制茶的时候发现你点的草莓卖完了，点单员会找到你，问你要不要换芒果口味，你同意后更新订单，再重新走流程——这就是回退边和人机中断。

整个奶茶店的工作流就是一个典型的有限状态机：每个环节是一个状态，流转规则是边，所有员工共享订单信息（全局状态），整个流程完全可控，不会出现“点单员突然去做奶茶、制茶员突然去接待客户”的混乱情况。

LangGraph做的事情，就是把这个奶茶店的流程搬到了代码里：你可以自己定义每个岗位（节点）的工作内容，定义岗位之间的流转规则（边），所有岗位共享订单信息（全局状态），最终实现一个完全可控的复杂Agent工作流。

核心概念解释（小学生都能懂）

核心概念一：有限状态机

有限状态机就像我们玩的闯关游戏：

• 游戏有很多关卡（状态），比如第一关、第二关、Boss关、通关、失败；
• 你只有打完第一关才能去第二关（流转规则）；
• 如果你打Boss关输了，就会回到第二关重新打（循环/回退）；
• 你要么通关，要么失败，没有中间的模糊状态（终止状态）。

所有有明确规则的流程都可以用状态机来描述：电梯的运行、红绿灯的切换、外卖的配送流程，都是状态机的实际应用。

核心概念二：LangGraph的三大核心要素

1. 节点（Node）：就像奶茶店的每个员工，每个节点只做一件事：点单员只接点单，加料员只加小料，制茶员只做奶茶。对应到代码里，节点就是一个Python函数，输入是当前的全局状态，输出是状态的更新内容。
2. 边（Edge）：就像奶茶店的传菜铃，告诉员工做完自己的活之后把订单传给谁。边分为两种：普通边就是做完A直接去B，条件边就是做完A之后看情况，满足条件去B，不满足去C。
3. 全局状态（State）：就像奶茶店的订单，所有员工都能看到，上面写着用户点了什么、加了什么料、付了多少钱、当前走到哪一步了。每个员工做完自己的活之后，就把自己的工作内容更新到订单上，下一个员工拿到的就是最新的订单信息。

核心概念三：Agent工作流

Agent工作流就像公司里的项目组：产品、开发、测试、UI都是独立的Agent，大家按照项目流程协作，产品出需求，开发写代码，测试测bug，UI做设计，中间有问题就退回给上一个角色修改，最终完成项目。LangGraph就是这个项目组的项目经理，负责安排每个人的工作，同步项目进度，确保整个流程按规则走。

核心概念之间的关系

我们可以用一个形象的比喻来描述三者的关系：

• 有限状态机是LangGraph的骨架，定义了整个工作流的流转规则，保证流程不会乱走；
• 节点是LangGraph的肌肉，是实际执行任务的单元，每个节点完成具体的工作；
• 全局状态是LangGraph的血液，在各个节点之间流转，同步所有的信息，让所有节点拿到的数据都是一致的。

概念一和概念二的关系：状态机是LangGraph的底层逻辑

LangGraph的所有节点和边都是按照状态机的规则设计的：你只能从入口节点开始，按照边的规则跳转，不能随便跳到没有关联的节点，就像你不能在奶茶店点单之后直接去拿做好的奶茶，必须走加料、制茶的流程。

概念二和概念三的关系：LangGraph是Agent工作流的编排工具

多Agent协作的最大痛点就是状态不同步和流程不可控：之前我们做多个Agent，要手动把A Agent的输出传给B Agent，还要处理B Agent不满意要A重写的逻辑，非常麻烦。用LangGraph的话，所有Agent的输出都会更新到全局状态，所有Agent都能拿到最新的信息，跳转规则用边定义就好，不用写一堆if else。

概念一和概念三的关系：状态机是复杂Agent工作流的可控保障

很多人做Agent的时候都会遇到“Agent乱跑”的问题：本来让它先查资料再写内容，结果它直接瞎编内容。有了状态机之后，你可以规定必须先走查资料节点，拿到资料之后才能走写内容节点，Agent完全没有乱跑的可能，流程100%可控。

核心概念原理和架构的文本示意图

┌───────────────────────────────────────────────────────────┐
│                     全局状态存储                          │
│  [订单信息、用户偏好、对话历史、当前流程状态、各节点输出]   │
└─────────────┬───────────────────┬───────────────────┬─────┘
              ▼                   ▼                   ▼
┌────────────────────┐  ┌────────────────────┐  ┌────────────────────┐
│   节点1：点单Agent  │  │  节点2：加料Agent  │  │  节点3：制茶Agent  │
│  逻辑：确认用户需求  │  │  逻辑：添加小料     │  │  逻辑：制作奶茶     │
└────────────────────┘  └────────────────────┘  └────────────────────┘
              │                   │                   │
              ▼                   ▼                   ▼
┌───────────────────────────────────────────────────────────┐
│                     边（流转规则）                        │
│ 普通边：点单→加料→制茶  条件边：有没有小料？是否合格？    │
└───────────────────────────────────────────────────────────┘

核心概念对比表

我们对比一下常见的工作流编排方案的差异，就能更清楚LangGraph的优势：

特性	LangChain顺序链	LangGraph	Airflow DAG
底层模式	线性执行	有限状态机	有向无环图
支持循环/回退	❌ 不支持	✅ 完全支持	❌ 不支持
全局状态管理	❌ 无内置，需手动传参	✅ 内置自动同步	❌ 需用XCom手动传递
执行延迟	低（实时）	低（实时）	高（离线调度）
适用场景	简单RAG、单轮对话	复杂Agent、多Agent协作、人机交互	离线数据任务、定时任务
流程可控性	低，只能按顺序走	高，可自定义任意跳转规则	中，只能走DAG路径

ER实体关系图

Mermaid 核心流程图

---

核心算法原理 & 具体操作步骤

LangGraph的核心算法逻辑非常简单，总结下来就是三步：取状态→跑节点→跳边，无限循环直到走到终止节点。

状态更新算法

LangGraph的状态更新遵循「部分更新+合并算子」的规则，不会全量覆盖状态，避免数据丢失：

1. 每个节点执行完成后，只返回需要更新的部分状态字段，不需要返回全量状态；
2. 对于普通字段，新值会直接替换旧值；
3. 对于用Annotated标记的字段，会用指定的合并算子处理新旧值，比如operator.add就是把新列表拼到旧列表后面，不会覆盖。

我们用Python代码演示状态的定义和更新：

from typing import TypedDict, Annotated, List
import operator
from langchain_core.messages import BaseMessage

# 定义奶茶店订单状态
class OrderState(TypedDict):
    # 普通字段：新值直接替换旧值
    customer_name: str
    total_price: float
    order_status: str
    # 带合并算子的字段：新列表会拼到旧列表后面
    items: Annotated[List[str], operator.add]
    add_ons: Annotated[List[str], operator.add]
    messages: Annotated[List[BaseMessage], operator.add]

# 初始状态
initial_state = OrderState(
    customer_name="张三",
    total_price=0,
    order_status="init",
    items=["珍珠奶茶"],
    add_ons=[],
    messages=[]
)

# 点单节点执行后返回的部分更新
node_output = {
    "total_price": 12,
    "order_status": "order_taken",
    "add_ons": ["椰果"],
    "messages": [{"role": "assistant", "content": "已确认订单，12元"}]
}

# LangGraph自动合并状态：普通字段替换，带算子的字段累加
updated_state = {**initial_state, **node_output}
# 合并后的add_ons是["椰果"]，messages多了一条，total_price变成12

流转调度算法

LangGraph的调度逻辑是一个无限循环，直到走到END节点：

def run_graph(graph, initial_state, config):
    # 1. 初始化状态，从持久化存储加载历史状态（如果有）
    current_state = load_state(config["thread_id"]) or initial_state
    current_node = graph.entry_node
    
    while True:
        # 2. 执行当前节点的逻辑，拿到部分更新
        node_output = graph.nodes[current_node].execute(current_state)
        # 3. 合并更新全局状态
        current_state = merge_state(current_state, node_output)
        # 4. 持久化状态到存储
        save_state(config["thread_id"], current_state)
        # 5. 根据边的规则找下一个节点
        next_node = get_next_node(graph.edges, current_node, current_state)
        # 6. 如果下一个节点是终止节点，结束运行
        if next_node == END:
            break
        # 7. 如果设置了中断，暂停运行，等待用户唤醒
        if next_node in graph.interrupt_nodes:
            save_interrupt(config["thread_id"], next_node)
            break
        current_node = next_node
    return current_state

核心操作步骤

用LangGraph开发工作流只需要五步：

1. 定义全局状态：确定哪些数据需要在节点之间共享，给需要累加的字段加合并算子；
2. 实现节点逻辑：每个节点写一个Python函数，输入是当前状态，输出是部分更新；
3. 定义流转边：添加普通边和条件边，定义节点之间的跳转规则；
4. 编译图：设置入口节点和终止节点，编译成可运行的应用；
5. 运行图：传入初始状态，流式获取每个节点的输出。

---

数学模型和公式 & 详细讲解

LangGraph的底层是标准的有限状态机，我们可以用数学公式严格定义：

有限状态机五元组

LangGraph的工作流可以表示为五元组：
$$G=(Q,\Sigma ,\delta ,q_0,F)$$
其中：

• $Q$：有限状态集合，对应所有节点执行后的状态集合；
• $\Sigma$：输入字母表，对应用户输入、LLM输出、工具调用结果、人工修改的状态等所有外部输入；
• $\delta$：状态转移函数，对应LangGraph中的边（普通边+条件边），输入是当前状态和当前输入，输出是下一个状态：$\delta :Q\times \Sigma \to Q$；
• $q_0$：初始状态，对应工作流的入口节点的初始状态；
• $F$：终止状态集合，对应END节点的状态集合。

状态更新公式

每个节点执行后的状态更新可以表示为：
$$S_{t+1}=S_t\oplus \Delta S_t$$
其中：

• $S_t$：t时刻的全局状态；
• $\Delta S_t$：t时刻节点执行后返回的部分更新内容；
• $\oplus$：状态合并算子，对于普通字段是替换操作，对于Annotated标记的字段是用户指定的算子（比如累加、求最大值等）。

循环终止条件

为了避免死循环，我们通常会给工作流加最大循环次数限制，终止条件为：
$$Terminate=(q_t\in F)\vee (coun{t}_t>MaxCount)$$
其中$coun{t}_t$是当前循环的次数，$MaxCount$是预先设置的最大循环次数，避免工作流无限运行。

举例说明

我们用奶茶店的例子代入公式：

1. 初始状态 q 0 = { c u s t o m e r : 张三 , i t e m s : [ 珍珠奶茶 ] , p r i c e : 0 , s t a t u s : i n i t } q_0 = \{customer: 张三, items: [珍珠奶茶], price:0, status: init\} q0​={customer:张三,items:[珍珠奶茶],price:0,status:init}；
2. 输入 Σ 0 = { 用户确认加椰果 } \Sigma_0 = \{用户确认加椰果\} Σ0​={用户确认加椰果}，转移函数$\delta (q_0,{\Sigma }_0)=点单节点$；
3. 点单节点执行后返回 Δ S 0 = { p r i c e : 12 , a d d o n s : [ 椰果 ] , s t a t u s : o r d e r t a k e n } \Delta S_0 = \{price:12, add_ons: [椰果], status: order_taken\} ΔS0​={price:12,addo​ns:[椰果],status:ordert​aken}；
4. 合并后的状态 S 1 = S 0 ⊕ Δ S 0 = { c u s t o m e r : 张三 , i t e m s : [ 珍珠奶茶 ] , a d d o n s : [ 椰果 ] , p r i c e : 12 , s t a t u s : o r d e r t a k e n } S_1 = S_0 \oplus \Delta S_0 = \{customer:张三, items:[珍珠奶茶], add_ons:[椰果], price:12, status:order_taken\} S1​=S0​⊕ΔS0​={customer:张三,items:[珍珠奶茶],addo​ns:[椰果],price:12,status:ordert​aken}；
5. 转移函数$\delta (S_1,\varnothing )=加料节点$，继续执行。

---

项目实战：多Agent需求评审系统开发

我们来开发一个生产级的多Agent需求评审系统：产品提需求后，依次经过产品经理、开发工程师、测试工程师、UI设计师评审，任何一个环节有问题就退回给产品修改，最多修改3次，超过3次就终止流程，支持人工介入修改需求，支持状态持久化。

开发环境搭建

首先安装依赖包：

pip install langgraph langchain langchain-openai python-dotenv redis

然后创建.env文件配置OpenAI密钥和Redis连接信息：

OPENAI_API_KEY=你的OpenAI密钥
REDIS_HOST=localhost
REDIS_PORT=6379

系统架构设计

我们的系统分为四层：

1. 接入层：提供HTTP接口，支持提交需求、查询评审进度、人工修改需求；
2. 编排层：LangGraph工作流，负责节点调度、状态同步、流程跳转；
3. 执行层：四个Agent节点（产品、开发、测试、UI），调用LLM完成评审逻辑；
4. 存储层：Redis做状态持久化，MySQL存储评审历史数据；
5. 监控层：LangSmith跟踪每个节点的输入输出，方便调试。

系统接口设计

接口地址	请求方式	功能描述	参数	返回值
/api/review/start	POST	提交需求开始评审	requirement: 需求内容	review_id: 评审ID
/api/review/status/{review_id}	GET	查询评审进度	review_id	当前状态、所有评审意见
/api/review/update	POST	人工修改需求	review_id: 评审ID, new_requirement: 新需求内容	状态更新结果
/api/review/resume/{review_id}	POST	唤醒中断的流程	review_id	继续运行结果

核心源代码实现

1. 导入依赖和初始化配置

import os
import operator
from typing import TypedDict, Annotated, List
from dotenv import load_dotenv
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.redis import RedisCheckpointSaver
import redis

load_dotenv()
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 初始化Redis持久化
redis_client = redis.Redis(
    host=os.getenv("REDIS_HOST"),
    port=int(os.getenv("REDIS_PORT")),
    db=0,
    decode_responses=False
)
checkpointer = RedisCheckpointSaver(redis_client)

2. 定义全局状态

class ReviewState(TypedDict):
    # 对话历史：累加
    messages: Annotated[List[BaseMessage], operator.add]
    # 需求文档内容：替换
    requirement_doc: str
    # 各角色评审意见：替换
    product_opinion: str
    dev_opinion: str
    test_opinion: str
    ui_opinion: str
    # 评审状态：init/product_review/dev_review/test_review/ui_review/approved/rejected
    review_status: str
    # 修改次数：替换
    revision_count: int

3. 实现各个Agent节点逻辑

# 产品评审Agent：检查需求完整性
product_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是资深互联网产品经理，负责评审需求的完整性、合理性和用户价值。如果需求清晰合理请输出「产品评审通过」，否则输出具体的修改意见。"),
    ("user", "需求文档：{requirement}\n 历史修改意见：{messages}")
])
product_chain = product_prompt | llm

def product_review_node(state: ReviewState) -> dict:
    res = product_chain.invoke({
        "requirement": state["requirement_doc"],
        "messages": state["messages"]
    })
    opinion = res.content
    next_status = "dev_review" if "产品评审通过" in opinion else "rejected"
    return {
        "product_opinion": opinion,
        "review_status": next_status,
        "revision_count": state["revision_count"] + (1 if next_status == "rejected" else 0),
        "messages": [AIMessage(content=f"【产品评审】{opinion}")]
    }

# 开发评审Agent：检查技术可行性
dev_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是资深后端开发工程师，负责评审需求的技术可行性、评估开发周期、指出技术风险。如果技术上可以实现请输出「开发评审通过」，否则输出具体的修改意见。"),
    ("user", "需求文档：{requirement}\n 产品评审意见：{product_opinion}\n 历史修改意见：{messages}")
])
dev_chain = dev_prompt | llm

def dev_review_node(state: ReviewState) -> dict:
    res = dev_chain.invoke({
        "requirement": state["requirement_doc"],
        "product_opinion": state["product_opinion"],
        "messages": state["messages"]
    })
    opinion = res.content
    next_status = "test_review" if "开发评审通过" in opinion else "rejected"
    return {
        "dev_opinion": opinion,
        "review_status": next_status,
        "revision_count": state["revision_count"] + (1 if next_status == "rejected" else 0),
        "messages": [AIMessage(content=f"【开发评审】{opinion}")]
    }

# 测试评审Agent：检查可测试性
test_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是资深测试工程师，负责评审需求的可测试性，指出测试边界和测试风险。如果可以测试请输出「测试评审通过」，否则输出具体的修改意见。"),
    ("user", "需求文档：{requirement}\n 开发评审意见：{dev_opinion}\n 历史修改意见：{messages}")
])
test_chain = test_prompt | llm

def test_review_node(state: ReviewState) -> dict:
    res = test_chain.invoke({
        "requirement": state["requirement_doc"],
        "dev_opinion": state["dev_opinion"],
        "messages": state["messages"]
    })
    opinion = res.content
    next_status = "ui_review" if "测试评审通过" in opinion else "rejected"
    return {
        "test_opinion": opinion,
        "review_status": next_status,
        "revision_count": state["revision_count"] + (1 if next_status == "rejected" else 0),
        "messages": [AIMessage(content=f"【测试评审】{opinion}")]
    }

# UI评审Agent：检查设计可行性
ui_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是资深UI设计师，负责评审需求的设计可行性，指出交互和视觉风险。如果可以设计请输出「UI评审通过」，否则输出具体的修改意见。"),
    ("user", "需求文档：{requirement}\n 测试评审意见：{test_opinion}\n 历史修改意见：{messages}")
])
ui_chain = ui_prompt | llm

def ui_review_node(state: ReviewState) -> dict:
    res = ui_chain.invoke({
        "requirement": state["requirement_doc"],
        "test_opinion": state["test_opinion"],
        "messages": state["messages"]
    })
    opinion = res.content
    next_status = "approved" if "UI评审通过" in opinion else "rejected"
    return {
        "ui_opinion": opinion,
        "review_status": next_status,
        "revision_count": state["revision_count"] + (1 if next_status == "rejected" else 0),
        "messages": [AIMessage(content=f"【UI评审】{opinion}")]
    }

4. 定义流转边和路由逻辑

def router(state: ReviewState) -> str:
    # 超过3次修改直接终止
    if state["revision_count"] >= 3:
        return "end"
    # 评审通过就结束
    if state["review_status"] == "approved":
        return "end"
    # 被驳回就回到产品节点修改
    if state["review_status"] == "rejected":
        return "product_review"
    # 否则走下一个节点
    return state["review_status"]

5. 构建和编译工作流

# 初始化图
workflow = StateGraph(ReviewState)
# 添加节点
workflow.add_node("product_review", product_review_node)
workflow.add_node("dev_review", dev_review_node)
workflow.add_node("test_review", test_review_node)
workflow.add_node("ui_review", ui_review_node)
# 设置入口节点
workflow.set_entry_point("product_review")
# 添加条件边
workflow.add_conditional_edges(
    "product_review",
    router,
    {
        "dev_review": "dev_review",
        "product_review": "product_review",
        "end": END
    }
)
workflow.add_conditional_edges(
    "dev_review",
    router,
    {
        "test_review": "test_review",
        "product_review": "product_review",
        "end": END
    }
)
workflow.add_conditional_edges(
    "test_review",
    router,
    {
        "ui_review": "ui_review",
        "product_review": "product_review",
        "end": END
    }
)
workflow.add_conditional_edges(
    "ui_review",
    router,
    {
        "end": END,
        "product_review": "product_review"
    }
)
# 编译图，设置开发节点前中断（需要人工确认技术方案）
app = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["dev_review"]
)

6. 运行工作流

if __name__ == "__main__":
    # 初始状态
    initial_state = {
        "requirement_doc": "做一个用户签到功能，每天签到得积分，连续签到7天得额外奖励，积分可以换优惠券",
        "review_status": "init",
        "revision_count": 0,
        "messages": []
    }
    # 配置：thread_id用来区分不同的评审任务
    config = {"configurable": {"thread_id": "review_123"}}
    # 流式运行
    for output in app.stream(initial_state, config=config):
        for node_name, node_output in output.items():
            print(f"===== 节点 {node_name} 运行完成 =====")
            print(f"当前评审状态：{node_output['review_status']}")
            print(f"输出：{node_output['messages'][-1].content}")
            print(f"修改次数：{node_output['revision_count']}")
            print("-"*60)
    
    # 模拟人工修改需求后继续运行
    print("\n===== 人工修改需求后继续运行 =====")
    # 更新状态
    app.update_state(config, {"requirement_doc": "做一个用户签到功能，每天签到得10积分，连续签到7天得100额外奖励，1000积分可以换10元优惠券"})
    # 继续运行
    for output in app.stream(None, config=config):
        for node_name, node_output in output.items():
            print(f"===== 节点 {node_name} 运行完成 =====")
            print(f"当前评审状态：{node_output['review_status']}")
            print(f"输出：{node_output['messages'][-1].content}")
            print(f"修改次数：{node_output['revision_count']}")
            print("-"*60)

代码解读与分析

1. 状态持久化：我们用RedisCheckpointSaver把状态存在Redis里，即使程序重启，只要用同一个thread_id就可以继续之前的流程；
2. 人机中断：我们设置了interrupt_before=["dev_review"]，流程走到开发节点之前会暂停，我们可以人工修改需求之后再继续运行，非常适合需要人工介入的场景；
3. 防死循环：我们设置了最多修改3次，超过就终止流程，避免无限循环；
4. 状态合并：messages字段用了operator.add合并，所有评审意见都会累加，不会丢失历史信息。

---

实际应用场景

LangGraph的状态机模式几乎可以覆盖所有复杂LLM应用场景：

1. 多Agent内容创作：大纲Agent→内容Agent→校对Agent→排版Agent→发布Agent，中间任何环节不满意都可以回退修改，适合写小说、写方案、做PPT等场景；
2. 企业内部审批流：报销申请→部门经理审批→财务审批→打款，有问题退回修改，支持多级审批，适合OA、CRM等系统的AI自动化审批；
3. 智能客服系统：用户咨询→意图分类→答疑Agent→如果解决不了转人工→满意度调查，全程同步对话历史，不用用户重复描述问题；
4. 自动数据分析平台：用户提交分析需求→取数Agent→数据清洗Agent→分析Agent→报表生成Agent，如果数据不对就重新取数，自动完成全流程数据分析；
5. 代码自动开发系统：需求评审→代码生成→代码评审→测试用例生成→运行测试，有BUG就回到代码生成节点修改，实现端到端的代码自动开发。

---

工具和资源推荐

1. 官方文档：LangGraph官方文档，最全的教程和API参考；
2. 官方示例库：LangGraph Examples，有几十种现成的工作流示例，包括多Agent协作、RAG、工具调用等；
3. LangSmith：LangChain官方的调试监控工具，可以跟踪每个节点的输入输出，排查问题非常方便；
4. LangServe：可以一键把LangGraph工作流部署成HTTP接口，不用自己写服务端代码；
5. 社区资源：LangChain Discord社区，有很多开发者分享LangGraph的最佳实践和问题解决方案。

---

未来发展趋势与挑战

发展历史

时间	阶段	核心痛点	解决方案
2022年之前	LLM应用1.0	只能单轮对话，无法处理复杂任务	基于Prompt工程的单Agent
2022-2023年	LLM应用2.0	单Agent能力有限，无法处理多步骤任务	LangChain顺序链、Sequential
2023年中	LLM应用3.0	顺序链无法支持循环、回退、多Agent状态同步	基于状态机的LangGraph发布
2024年及以后	LLM应用4.0	复杂多Agent系统的编排、调度、容错	可视化编排、自动生成状态机、分布式LangGraph

未来发展趋势

1. 低代码可视化编排：未来会有更多拖拽式的LangGraph编排工具，不用写代码就能搭建复杂Agent工作流，产品经理也能上手；
2. 自动状态机生成：输入需求描述，大模型自动生成LangGraph的节点、边和状态定义，开发者只需要微调即可；
3. 分布式状态支持：支持超大状态的分片存储和跨节点同步，适合成百上千个Agent协作的超复杂系统；
4. 智能容错机制：节点运行失败自动重试、自动降级，自动处理LLM输出异常的情况，提高系统稳定性。

面临的挑战

1. 状态爆炸问题：复杂工作流的状态字段会越来越多，如何高效管理和合并状态是一个挑战；
2. 多Agent冲突解决：多个Agent的意见不一致的时候，如何自动仲裁，需要更智能的路由规则；
3. 可解释性问题：用户需要知道工作流当前走到哪一步，为什么这么跳转，需要更友好的流程可视化；
4. 性能优化：节点多了之后运行延迟会升高，如何并行执行不相关的节点，提高运行效率。

---

总结：学到了什么？

核心概念回顾

1. 有限状态机：是LangGraph的底层核心，所有流程都是按照 predefined 的规则在状态之间跳转，保证流程100%可控；
2. LangGraph三大核心：节点是执行单元，边是流转规则，全局状态是所有节点共享的数据，三者配合实现复杂工作流；
3. Agent工作流：就是把多个Agent作为节点，用状态机来编排，解决传统线性链无法处理的循环、回退、状态同步问题。

概念关系回顾

• 状态机是骨架，定义了流程的跳转规则；
• 节点是肌肉，完成具体的执行逻辑；
• 全局状态是血液，在节点之间同步所有信息；
• 三者配合，就能构建出可控、可扩展、可维护的复杂Agent系统。

---

思考题：动动小脑筋

1. 你所在的行业有哪些场景可以用LangGraph的状态机模式来实现？请写出至少3个场景的节点、边和状态设计；
2. 如果要做一个自动写毕业论文的Agent工作流，你会怎么设计节点、边和状态？怎么避免生成的内容重复、逻辑不通？
3. 怎么避免LangGraph的工作流出现死循环？请写出至少3种解决方案。

---

附录：常见问题与解答

Q1：LangGraph和LangChain的Chain有什么区别？

A：Chain是线性的，只能顺序执行，不支持循环、回退、分支，没有全局状态，适合简单场景；LangGraph是基于状态机的，支持任意跳转、循环、回退，内置全局状态同步，适合复杂Agent场景。

Q2：LangGraph的状态存在哪里？生产环境怎么持久化？

A：默认存在内存里，生产环境可以用官方提供的RedisCheckpointSaver、PostgresCheckpointSaver把状态存在外部存储里，也可以自己实现CheckpointSaver对接其他存储。

Q3：LangGraph支持多租户吗？

A：支持，每个租户的任务用不同的thread_id区分，状态完全隔离，同一个工作流可以同时处理多个租户的任务。

Q4：LangGraph可以和其他LLM框架配合使用吗？

A：可以，LangGraph不绑定任何LLM框架，节点逻辑可以用任意框架实现，比如LlamaIndex、AutoGPT等，只要返回状态更新即可。

---

扩展阅读 & 参考资料

1. LangGraph官方文档
2. LangChain官方博客：LangGraph发布介绍
3. 有限状态机维基百科
4. 论文：AutoGPT的工作流编排模式
5. LangGraph生产级最佳实践