LangGraph 是一个用于构建有状态 Agent 的编排框架。
它主要解决一个问题:当 Agent 的执行过程开始变复杂以后,怎样把状态和流程管理清楚。
在 LangGraph 中,一个 Agent 的运行过程通常会拆成 State、Node 和 Edge。
State 保存任务执行过程中需要使用的数据,Node 负责完成某一步工作,Edge 决定完成这一步之后继续执行哪里。
模型调用、Tool 执行、条件判断、人工确认,都可以放进这套结构中。
如果只是一个很简单的 Agent,其实没有必要直接使用 LangGraph。一个普通的 Agent Loop 已经能够完成。
但真实项目中的 Agent 往往会更复杂。
任务可能需要分成多个步骤,中途产生的数据需要保存,某一步可能需要等待人工确认,执行失败以后还希望从原来的位置继续。
这时我们关心的问题就不再只是:
模型下一步调用哪个 Tool?
还会变成:
任务现在执行到哪一步了?
当前已经有哪些数据?
下一步应该执行什么?
如果暂停了,之后怎样继续?
如果中途失败,哪些步骤不用重新执行?
LangGraph 主要处理的就是这些问题。
我们可以先这样理解:普通 Agent Loop 主要解决“下一步做什么”,LangGraph 进一步解决“整个任务怎样运行”。
这也是 LangGraph 为什么要引入 State、Node 和 Edge 的原因。它不是为了把一个简单程序变复杂,而是把原本散落在循环、条件判断和变量中的执行逻辑,整理成一套更清楚的结构。
后面我们会看到,Checkpoint、Interrupt、Persistence、Durable Execution 等能力,也都是建立在这套结构之上的。
一、Agent Loop 的能力局限
最常见的 Agent Loop,其流程如下:
模型拿到用户输入以后,判断是否需要调用 Tool。如果需要,就执行 Tool,再把 Tool Result 交给模型继续处理。直到模型不再请求 Tool,最后返回结果。
整个过程执行路径很明确,持续时间也相对较短:
Model → Tool → Model
针对简单的任务,普通 Agent Loop 已完全够用。
然而,随着需求持续不断地增加进来,问题会逐渐暴露出来。
例如,我们准备做一个研究 Agent。
第一版可能只有三个步骤:
搜索资料 → 整理资料 → 生成报告
很快就会遇到一个问题:搜索一次,不一定能找到足够的资料。
于是流程变成:
搜索资料
↓
检查资料是否充分
├─ 不够 → 继续搜索
└─ 足够 → 生成报告
再往后,需求可能继续增加。
例如:
搜索失败 → 重试
资料不足 → 更换关键词继续搜索
报告生成 → 等待人工审核
审核不通过 → 返回修改
执行重要操作 → 先让用户确认
这些逻辑仍然可以放进一个普通循环。
例如:
while True:
response = model.invoke(messages)
if response.tool_calls:
...
continue
if need_more_research:
...
continue
if need_review:
...
continue
if should_retry:
...
continue
if finished:
break
这样写并没有问题。
但随着逻辑越来越多,代码会出现另一个麻烦:任务的数据和执行流程开始耦合在一起。
例如 messages 中可能同时保存:
用户输入
模型回复
Tool Result
搜索结果
中间判断
除此之外,程序里还可能出现:
retry_count
review_status
current_step
search_round
approved
这些变量有的在函数里,有的在数据库里,有的放在消息上下文中。
代码依然可以运行,但我们会越来越难以弄清楚:
这个任务现在究竟进行到哪一步了?
这就是 LangGraph 真正体现其价值的地方。
它希望把这些原本分散的数据和流程,整理成明确的 State 和 Graph,正如在本文开头所提到的。
二、为什么会用到“图”
LangGraph 中的 “Graph” 意为“图”,那么,为什么要用“图”来构建 Agent 呢?
很多程序的执行流程都很简单。
例如:
A → B → C → D
执行完 A,再执行 B,然后 C、D,一路向前即可。
数据处理、持续集成 Pipeline 等任务经常适合这种结构。
但 Agent 有一个明显特点:它很容易重复执行前面的步骤。
例如 Tool Calling:
Model
↓
Tool
↓
Model
↓
Tool
↓
Model
模型调用一次 Tool 以后,并不一定结束。
它可能根据 Tool Result 再决定调用另一个 Tool。
再比如一个旅行规划 Agent:
收集偏好
↓
生成行程
↓
目的地冲突
↓
调整路线
↓
再次确认
这里同样存在循环。
所以 Agent 的流程通常不只是:
A → B → C
还可能出现:
A → B → C
↑ ↓
└───┘
同时还可能有多个分支:
┌→ retry
check_result ┤
├→ review
└→ finish
这时候,用 Graph 来描述流程就比较自然了。
Graph 可以明确表示:
有哪些步骤
步骤之间是什么关系
什么情况下走哪条路径
什么情况下重新执行前面的步骤
所以 LangGraph 中的 “Graph”,并非为了画图而存在。它真正代表的,是 Agent 的执行结构。
三、LangGraph 在 Agent 中的职责
LangGraph 可以理解为一个用于组织和运行 Agent 工作流的框架。
这里可以把它负责的事情分成两部分。
第一部分是:流程怎样走。
例如:
先搜索
再检查
资料不足继续搜索
资料充分开始生成
生成之后进入审核
第二部分是:任务运行过程中怎样保存状态。
例如:
现在搜索了几次
已经找到哪些资料
报告有没有生成
用户是否已经审核
任务当前停在哪一步
这两个问题在简单 Agent 中通常不明显。因为任务可能几秒钟就结束了。但任务一旦变长,它们就会越来越重要。
在 LangGraph 中,一个 Node 中可以调用模型:
调用 Model
也可以调用 Tool:
执行 Tool
还可以只是普通程序:
查询数据库
检查参数
调用 API
保存文件
判断结果
等待人工确认
所以更准确地说,LangGraph 管理的是整个 Agent 任务怎样一步一步执行。
当前 LangChain Agent 与 LangGraph,大致可以这样理解:
应用
│
┌────────┴────────┐
│ │
LangChain Agent 自定义 Workflow
│ │
└────────┬────────┘
│
LangGraph
│
Model / Tool / Database
LangChain Agent 已经封装好了最常见的 Model + Tool Calling Loop。
如果只是标准 Agent,直接使用 LangChain Agent 通常就够了。
实际上,目前 LangChain 的 create_agent 本身也是建立在 LangGraph 之上的。
也就是说,使用 LangChain Agent 时,底层已经使用了 LangGraph,只是开发者不需要自己定义 StateGraph。
四、State、Node 和 Edge
理解 LangGraph,第一步就是理解三个概念:
State
Node
Edge
它们分别关注三个不同的问题。
State:现在有哪些数据?
Node:这一步做什么?
Edge:下一步去哪?
State:保存任务数据
State 表示当前任务拥有的数据。
例如:
from typing_extensions import TypedDict
class State(TypedDict):
topic: str
result: str
retry_count: int
这里记录了三个信息:当前任务主题、当前执行结果、已经重试几次。
实际 Agent 中还可能保存:
messages
search_results
review_status
approved
不同项目的 State 会完全不同。
任务执行过程中需要持续使用的数据,可以集中放在 State 中。Node 执行时读取 State。执行结束之后,再把产生的新数据写回 State。
整个过程如下:
当前 State
↓
执行 Node
↓
更新 State
↓
执行下一个 Node
Node:完成一步工作
Node 表示 Graph 中的一步操作。
例如:
搜索资料
调用模型
执行 Tool
检查结果
人工审核
保存数据
Node 在 Python 中通常就是普通函数:
def research(state: State):
...
return {
"result": "..."
}
它做的事情很直接:
读取 State
↓
完成任务
↓
返回更新
例如 research Node 可以读取 topic,然后搜索相关资料。
搜索完成以后返回:
{
"result": "..."
}
这样下一步 Node 就可以继续使用这个结果。
Edge:决定下一步
Edge 表示 Node 之间怎样连接。
最简单的是固定执行:
START → research → write → END
也就是:
开始
↓
搜索
↓
撰写
↓
结束
也可以根据条件选择不同路径:
┌→ retry
check_result ┤
└→ finish
如果结果不合格,就进入 retry。
如果结果已经满足要求,就进入 finish。
所以 Edge 本质上解决的是:当前步骤完成以后,下一步执行哪里?
综上,State 保存数据,Node 处理数据,Edge 连接步骤,这便是 LangGraph 最基础的模型。
五、State 是怎样更新的
这里还有一个很重要的细节,Node 通常不需要返回完整 State。
例如当前 State 是:
{
"topic": "LangGraph",
"result": "",
"retry_count": 0,
}
现在有一个 Node 只负责生成 result:
def generate_result(state: State):
return {
"result": "LangGraph is an orchestration framework."
}
这个 Node 并没有返回:
topic
retry_count
它只返回自己修改的部分。
LangGraph 会把它合并到原来的 State 中,这叫做 Partial State Update,也即 Node 只需要返回自己负责更新的字段。
但如果字段本身是一个列表,就会出现一个问题。
例如:
steps: list[str]
当前值是:
["search"]
Node 返回:
["write"]
最终应该变成:
["write"]
还是:
["search", "write"]
这取决于 State 的更新规则。
LangGraph 使用 Reducer 来处理这件事。
Reducer 可以理解成:当一个字段有旧值,又收到新值时,应该怎样合并。
如果没有特殊配置,很多字段默认就是直接覆盖:
旧值:["search"]
新值:["write"]
结果:["write"]
如果我们希望列表追加,可以这样定义:
from operator import add
from typing import Annotated
from typing_extensions import TypedDict
class State(TypedDict):
steps: Annotated[list[str], add]
这时:
旧值:["search"]
新值:["write"]
结果:["search", "write"]
一开始学 LangGraph 时,并不需要马上掌握复杂 Reducer。先记住一个区别即可:
普通字段:通常直接更新
需要累积的字段:可以定义合并规则
消息列表就是很典型的例子。
Agent 每执行一步,都可能产生一条新 Message。
这些 Message 通常不是覆盖旧消息,而是继续追加。
后面讲 State 和 MessagesState 时,我们再详细展开。
六、写一个最简单的 StateGraph
刚开始学习 LangGraph,为了简单,可以不加入 LLM 和 Tool。
我们先看一个普通示例。这样更容易看清 LangGraph 自己到底做了什么。
安装:
pip install -U langgraph
先定义 State:
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
text: str
length: int
然后定义第一个 Node:
def normalize_text(state: State):
return {
"text": state["text"].strip()
}
它负责去掉文本两边的空格。
第二个 Node:
def count_length(state: State):
return {
"length": len(state["text"])
}
它负责计算文本长度。
接下来创建 Graph:
builder = StateGraph(State)
builder.add_node("normalize_text", normalize_text)
builder.add_node("count_length", count_length)
然后定义执行顺序:
builder.add_edge(START, "normalize_text")
builder.add_edge("normalize_text", "count_length")
builder.add_edge("count_length", END)
整张 Graph 就是:
START
↓
normalize_text
↓
count_length
↓
END
最后编译:
graph = builder.compile()
执行:
result = graph.invoke({
"text": " LangGraph ",
"length": 0,
})
print(result)
结果:
{'text': 'LangGraph', 'length': 9}
这个例子并未引入 LLM,只使用了两个普通函数。
这里真正要观察的是 State 怎样变化。
最开始:
{
"text": " LangGraph ",
"length": 0
}
经过 normalize_text:
{
"text": "LangGraph",
"length": 0
}
再经过 count_length:
{
"text": "LangGraph",
"length": 9
}
所以执行过程其实是:
读取 State
↓
执行 Node
↓
返回部分更新
↓
得到新的 State
↓
执行下一个 Node
第二个 Node 读取到的,已经是第一个 Node 更新之后的数据。
这就是 StateGraph 最基本的运行方式。
七、什么情况下值得使用 LangGraph
LangGraph 并不是所有 Agent 项目的默认选择。
如果只是:
用户 → Model → 返回结果
直接调用模型即可。
如果只是常见的:
Model → Tool → Model
并且标准 Agent Loop 已经能够清楚完成任务,直接使用 LangChain Agent 通常更简单。
实际项目中,主要看执行过程是否已经变复杂。
例如任务开始出现多个明确步骤:
搜索 → 分析 → 生成 → 审核
或者根据结果走不同路径:
检查通过 → 结束
检查失败 → 重新生成
又或者任务需要暂停以后继续:
生成方案
↓
等待用户确认
↓
继续执行
如果中途失败后还希望保留已经完成的结果,或者一个任务会跨越多次 HTTP 请求,那么执行状态就需要被单独管理。
这时程序里通常也会逐渐出现:
current_step
retry_count
review_status
pending
resume
越来越多代码开始处理同一个问题:
任务之前执行到了哪里?
这通常就是引入 LangGraph 的信号。
需要注意,Tool 多并不等于一定需要 LangGraph。
一个 Agent 即使有十几个 Tool,只要每次调用很快结束,标准 LangChain Agent 仍然可能非常合适。
反过来,一个 Workflow 即使只有一次模型调用:
生成内容
↓
等待人工审核
↓
继续处理
只要它需要长期保存状态、暂停和恢复,就已经很适合 LangGraph。
因此判断时,可以重点看三个方面:
状态是否越来越多
执行路径是否越来越复杂
任务是否越来越难一次完成
如果这些问题逐渐增多,LangGraph 的价值也就开始明显。
总结
本文主要解决了一个问题:为什么一个已经能够运行的 Agent,还需要 LangGraph。原因并不在于 Tool 数量,而在于任务开始出现明确状态、分支、循环、暂停和恢复需求。
LangGraph 用 State 保存数据,用 Node 表示步骤,用 Edge 描述路径,再由 Graph 组织整个执行过程。理解这几个基本概念后,后面的 Checkpoint、Interrupt、Persistence 和 Durable Execution 就有了统一的理解基础。
社区讨论
参与讨论
有问题或想法?欢迎继续讨论。