LangGraph原理深度解析:构建Agent最适合的框架
时间:2026-06-11 17:06
LangGraph将Agent运行过程抽象为有向图,通过State、Node、Edge和Checkpoint等核心概念实现复杂控制流。State作为全局不可变数据,使用Reducer处理并发更新;Node是无副作用的计算单元;Edge显式定义流转,条件边实现分支与循环。Checkpoint持久化每步快照,支持多轮对话与断点恢复,并提供人工中断原语。该框架不预
LangGraph 原理深度解析:为什么它是目前最适合构建 Agent 的框架
说个具体场景。假设你要搭建一个 AI Agent,它需要:
- 接收用户问题
- 判断是否需要查资料
- 如果需要,调用搜索工具
- 把搜索结果交给 LLM 继续推理
- 如果还不够,继续搜索;否则输出答案

最朴素的写法就是一堆嵌套的 `if/else` 加 `while` 循环。可一旦逻辑稍微复杂一点,代码就很容易腐烂——状态传来传去、循环条件难以维护、想加一个步骤要改好几个地方。
LangGraph 解决的正是这个难题。它将“Agent 的运行过程”抽象成一张有向图,用图的语言来描述流程,让复杂的控制逻辑变得清晰、可维护、可扩展。
### 核心抽象:图、状态、节点、边
LangGraph 的整个设计围绕四个核心概念展开。
**State(状态)**
State 是贯穿整个图的共享数据结构,用 Python 的 `TypedDict` 定义:
```python
from typing import TypedDict, Annotated
import operator
class AgentState(TypedDict):
messages: Annotated[list, operator.add] # 对话历史
current_plan: str # 当前计划
iteration_count: int # 已迭代次数
```
State 有几个关键特性:
首先是不可变更新——节点函数不直接修改 state,而是返回一个“diff”(增量),LangGraph 负责把这个 diff 合并进当前 state。这个思路与 Redux 一致,能让状态变更变得可追溯。
其次是 Reducer 机制:`Annotated[list, operator.add]` 告诉 LangGraph,当多个节点同时更新 `messages` 字段时,用 `operator.add`(列表追加)而不是覆盖。这是处理并发更新冲突的方式——每个字段都可以有自己的合并策略。
```python
# 节点只需返回要更新的字段,不需要返回完整 state
def my_node(state: AgentState) -> dict:
return {"messages": [new_message]}
# LangGraph 负责 merge
```
**Node(节点)**
节点是图中的计算单元,实质上就是一个 Python 函数:
```python
def call_llm(state: AgentState) -> dict:
response = model.invoke(state["messages"])
return {"messages": [response]}
```
节点可以是任何操作:LLM 调用、工具执行、数据转换、外部 API 请求。LangGraph 对节点内部的实现没有任何约束。
LangGraph 还内置了一些常用节点,最常用的是 `ToolNode`,它会自动解析 LLM 输出的 `tool_calls`,执行对应的工具,然后把结果包装成 `ToolMessage` 写回 state:
```python
from langgraph.prebuilt import ToolNode
tool_node = ToolNode(tools=[search, calculator])
```
**Edge(边)**
边定义了节点之间的流转关系,分两种:
固定边:A 执行完永远去 B。
```python
builder.add_edge("tools", "agent")
```
条件边:根据当前 state 决定下一步去哪里,这是 LangGraph 实现分支和循环的关键。
```python
def should_continue(state: AgentState) -> str:
last_msg = state["messages"][-1]
if last_msg.tool_calls:
return "tools" # 去执行工具
return END # 结束
builder.add_conditional_edges("agent", should_continue)
```
`should_continue` 函数接收当前 state,返回一个字符串,LangGraph 根据这个字符串决定下一个节点。这个函数就是整个 Agent 的“大脑决策”——完全由你控制,不依赖任何框架的黑魔法。
**Graph(图)**
图是上述三个元素的容器,负责编排和执行:
```python
from langgraph.graph import StateGraph, END
builder = StateGraph(AgentState)
builder.add_node("agent", call_llm)
builder.add_node("tools", tool_node)
builder.set_entry_point("agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")
graph = builder.compile()
```
`compile()` 做的事情包括:验证图的合法性(有没有死路、节点是否都定义)、生成执行计划、挂载 checkpointer。编译后的 `graph` 对象是不可变的,可以在多线程环境下安全复用。
### 执行机制:图是怎么跑起来的
调用 `graph.invoke(input)` 后,LangGraph 的执行流程如下:
1. 初始化 state:将 input 合并进初始 state
2. 从 entry_point 节点开始
3. 执行当前节点函数,拿到返回的 diff
4. 将 diff 通过 reducer 合并进 state(生成新 state 快照)
5. 根据出边(固定边或条件边)确定下一个节点
6. 重复步骤 3-5,直到到达 END
7. 返回最终 state
每一步执行后,LangGraph 都会生成一个新的 state 快照。如果配置了 checkpointer,这个快照会被持久化——这是多轮对话和断点恢复的基础。
这个执行模型本质上是一个事件循环,每次循环处理一个节点。循环本身(即 Agent 的“思考-行动”循环)是通过图的结构来表达的,而不是通过代码里的 `while True`。
### Checkpoint:状态持久化的原理
Checkpoint 是 LangGraph 最被低估的特性之一。它的核心思想非常简单:把每一步执行后的完整 state 快照存储下来。
```python
from langgraph.checkpoint.memory import MemorySa ver
memory = MemorySa ver()
graph = builder.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "session_001"}}
graph.invoke({"messages": [HumanMessage("你好")]}, config=config)
```
`thread_id` 是会话的唯一标识。每次 `invoke` 时,LangGraph 会:
1. 用 `thread_id` 从 checkpointer 里查找历史快照
2. 如果有,恢复到上次结束时的 state
3. 执行完成后,把新的 state 快照存进 checkpointer
这意味着多轮对话的记忆不是靠手动维护消息列表实现的,而是 checkpointer 自动帮你完成。换成 SQLite 做持久化,进程重启后对话历史依然还在:
```python
from langgraph.checkpoint.sqlite import SqliteSa ver
memory = SqliteSa ver.from_conn_string("./agent.db")
```
Checkpoint 还支持时间旅行——你可以获取任意历史快照,从那个时间点重新执行,这对调试复杂 Agent 非常有用。
### Human-in-the-Loop:中断与恢复
LangGraph 原生支持在指定节点前暂停,等待人工干预后再继续:
```python
graph = builder.compile(
checkpointer=memory,
interrupt_before=["execute_action"] # 执行这个节点前暂停
)
# 第一次调用:执行到 execute_action 前暂停
graph.invoke(input, config=config)
# 人工审核,确认后继续执行
graph.invoke(None, config=config) # 传 None 表示从暂停点继续
```
实现原理是:到达中断节点时,LangGraph 保存当前 state 快照,然后抛出一个特殊的中断信号。下次 `invoke` 时,从快照恢复,跳过已执行的节点,从中断点继续。这个机制对需要人工审核的高风险操作场景——比如执行数据库删除、发送邮件——非常关键。
### 与其他框架的本质区别
很多人问:LangGraph 和 LangChain 的 `AgentExecutor`、AutoGen 有什么本质区别?
`AgentExecutor` 是一个封装好的黑盒,它内置了 ReAct 的执行逻辑,你只需要提供工具列表。好处是上手快,坏处是控制权在框架手里——想改循环逻辑、加条件分支、自定义状态,都很别扭。
`AutoGen` 的核心抽象是“多个 Agent 互相对话”,适合角色分工明确的多 Agent 协作场景,但对单个 Agent 内部的控制流支持有限。
而 LangGraph 的定位是控制流引擎——它不预置 Agent 的默认行为,只提供图的描述能力和执行机制。你来定义流程,它负责执行。这种设计在简单场景下比 `AgentExecutor` 啰嗦,但在复杂场景下优势明显:任何你能想到的流程,都可以用图来描述。
### 总结
LangGraph 的设计哲学可以用一句话概括:将 Agent 的控制流从代码中解放出来,用图来表达复杂流程。
| 概念 | 作用 |
|------|------|
| State | 全局共享的不可变数据,reducer 处理并发更新 |
| Node | 无副作用的计算单元,只负责处理和返回 diff |
| Edge | 控制流的显式描述,条件边实现分支和循环 |
| Checkpoint | 每步快照持久化,支持多轮对话与断点恢复 |
| Interrupt | 人工介入的原语,高风险操作不可或缺 |
理解了这五个概念,LangGraph 就没有黑魔法了——它就是一个带状态管理的图执行引擎,只不过这个图恰好特别适合描述 Agent 的行为。
**参考资料**
- LangGraph 官方文档
- LangGraph 源码
- LangGraph Academy