# 第一章 LangGraph 入门：StateGraph 与带记忆的 ReAct 循环


>"LangChain 负责单次调用，LangGraph 负责把调用串成可循环、可记忆的工作流。"


## 一、问题


用 LangChain 写 Agent 时，一个常见的困境是：**链（Chain）只能线性执行**。用户提问 -> 模型思考 -> 调用工具 -> 返回结果，流程结束。如果模型发现"工具返回的结果不够，还需要再查一次"，线性链路就断了。你必须手动把结果粘贴回 prompt，再发起新一轮请求。

另一个困境是**状态管理**。多轮对话的历史存在哪里？跨轮次的消息怎么传递？用全局变量容易乱，用局部变量进程一重启就丢失。想让 Agent 记住"你刚才问的是上海"，需要写大量胶水代码来拼接消息列表。

LangGraph 的出现正是为了解决这两个问题：**循环（Cycles）** 与 **持久化状态（Persistent State）**。

## 二、解决方案


LangGraph 的核心思路是"状态图"（State Graph）：把整个 Agent 的运行抽象成一张图，图上有节点、有边，还有一块在所有节点之间流转的"公共黑板"——状态。

```
+--------+      +------------+      +------------------+
| START  | ---> |   agent    | ---> | should_continue  |
|        |      | (模型调用)  |      |   (条件判断)      |
+--------+      +------+-----+      +---------+--------+
                       ^                      |
                       |                      |
                       |        +-------------+-------------+
                       |        | 返回 "tools" |  返回 END    |
                       |        |              |              |
                       |        v              v              v
                       |   +--------+    +--------+
                       |   | tools  |    |  END   |
                       |   |(工具)  |    +--------+
                       |   +---+----+
                       |       |
                       +-------+
                       (回到 agent)
```
这张图的运行逻辑非常清晰：


1. 从 `START` 进入 `agent` 节点，模型接收用户提问并思考。

2. `agent` 节点完成后，进入 `should_continue` 条件判断：
	- 如果模型想调用工具，流向 `tools` 节点；
	- 如果模型直接给出答案，流向 `END`，结束运行。

3. `tools` 节点执行完毕后，通过一条普通边固定回到 `agent` 节点，让模型再次审视结果。

4. 循环往复，直到模型不再调用工具为止。


与此同时，**检查点（Checkpoint）**会把每一轮结束后的完整状态保存下来。只要使用相同的 `thread_id`，下一次调用就会带着之前的记忆继续。

## 三、工作原理


### 1. 定义工具：`@tool` 与 `ToolNode`


```
from langchain.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def search(query: str) -> str:
    """模拟一个搜索工具"""
    if "上海" in query.lower() or "Shanghai" in query.lower():
        return "现在30°，有雾"
    return "现在温度35度"

tools = [search]
tool_node = ToolNode(tools)
```
`@tool` 是 LangChain 的装饰器，作用是给一个普通 Python 函数打上标记，让它能被大语言模型识别为可调用的工具。装饰器会自动读取函数的参数类型、文档字符串，生成模型所需的工具描述（Tool Schema）。你不需要手写 JSON 格式的工具定义。

`ToolNode` 是 LangGraph 提供的预置节点。你只需要把工具列表传给它，它内部会自动完成"解析参数 -> 查找对应函数 -> 执行 -> 包装结果"的完整流程。你不需要自己写 `for block in response.content` 那种分发逻辑。

### 2. 绑定工具到模型


```
from langchain.chat_models import ChatOpenAI

model = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)
```
`bind_tools` 的作用是把工具"注册"进模型实例。注册之后，模型在生成回复时就知道自己拥有 `search` 这个工具。当它判断当前问题需要外部信息（比如实时天气）时，就会输出带有 `tool_calls` 的特殊消息结构，而不是直接给出一个编造的答案。

### 3. 定义条件路由：`should_continue`


```
from typing import Literal
from langgraph.graph import END

def should_continue(state: MessagesState) -> Literal["tools", END]:
    messages = state['messages']
    last_message = messages[-1]

    # 如果 LLM 调用了工具，则转到 tools 节点
    if last_message.tool_calls:
        return "tools"
    # 否则停止
    return END
```
这是 LangGraph **条件边（Conditional Edge）**的核心机制。`should_continue` 是一个普通的 Python 函数，但它接收的参数是整个图的当前 `state`。函数读取最后一条消息，检查模型是否发起了工具调用：


- 如果有 `tool_calls`，返回字符串 `"tools"`，LangGraph 就会把流程导向名为 `tools` 的节点；

- 如果没有，返回内置常量 `END`，LangGraph 就会终止图的运行。


`Literal["tools", END]` 是 Python 的类型提示，它的作用不仅是给 IDE 看，也让 LangGraph 在编译时就能校验：你的返回值必须对应图中真实存在的节点名，或者是 `END`。

### 4. 定义 Agent 节点：`call_model`


```
def call_model(state: MessagesState):
    messages = state['messages']
    response = model.invoke(messages)
    # 返回字典，LangGraph 会自动合并到状态中
    return {"messages": [response]}
```
`call_model` 是一个**节点函数**。在 LangGraph 中，节点函数必须遵循一个约定：


- **输入**：接收当前的 `state`（这里是 `MessagesState`）；

- **输出**：返回一个字典，字典的键对应状态中的字段名。


这里返回的 `{"messages": [response]}` 看起来只包含了一条新消息，但 LangGraph 不会直接覆盖原有列表。因为 `MessagesState` 内部自带了**归约器（Reducer）**，它的默认行为是"追加"（append）。所以你不需要手动把新消息塞进旧列表，LangGraph 会自动帮你合并。这解决了 LangChain 中常见的"消息历史管理"问题。

### 5. 构建状态图：`StateGraph(MessagesState)`


```
from langgraph.graph import StateGraph, MessagesState

workflow = StateGraph(MessagesState)
```
`StateGraph` 是 LangGraph 的图构建器，相当于一张"空白画布"。构造函数里传入的 `MessagesState` 是**状态模式（State Schema）**，它规定了这张图上所有节点共享的数据结构。

`MessagesState` 是 LangGraph 内置的一个 `TypedDict`，核心字段只有 `messages: list`，但它已经预配置好了追加归约器。如果你需要自定义字段（比如加上 `user_name`、`session_id`），可以继承 `TypedDict` 自己定义状态模式。

### 6. 注册节点与边


```
from langgraph.graph import START

workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# 设置入口点：图从 START 标记流向 agent 节点
workflow.add_edge(START, "agent")

# 添加条件边：从 agent 节点出发，根据 should_continue 的返回值决定去向
workflow.add_conditional_edges(
    "agent",
    should_continue
)

# 普通边：tools 执行完后固定回到 agent，形成循环
workflow.add_edge("tools", "agent")
```
这一步把"空白画布"真正画成流程图：


- `add_node(name, func)`：注册节点。第一个参数是节点在图中的唯一标识，第二个参数是实际执行的函数或对象。这里 `"agent"` 绑定的是 `call_model` 函数，`"tools"` 绑定的是 `tool_node` 对象。

- `add_edge(START, "agent")`：设置图的入口。`START` 是 LangGraph 内置的特殊标记，表示"图开始运行时的第一个节点"。老版本 API 使用 `set_entry_point("agent")`，新版本推荐显式写为边，语义更清晰。

- `add_conditional_edges(source, condition)`：从 `source` 节点出发，根据 `condition` 函数的返回值动态选择下一跳。这是 LangGraph 支持**循环**的关键：普通边只能走一次，条件边可以在运行时根据状态反复选择不同路径。

- `add_edge("tools", "agent")`：工具执行完后，固定回到 `agent` 节点。这条边没有条件，因此它每次都会执行，形成 `agent -> tools -> agent` 的闭环。


### 7. 编译与持久化：`InMemorySaver`


```
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
app = workflow.compile(checkpointer=checkpointer)
```
`InMemorySaver` 是一个**检查点保存器（Checkpointer）**，你可以把它想象成一个带索引的仓库管理员。它内部其实是一个字典结构：`{thread_id: [checkpoint_0, checkpoint_1, ...]}`。刚创建时这个字典是空的——它不会预先为 `42` 分配任何空间。

`workflow.compile(checkpointer=checkpointer)` 的作用是**把仓库管理员注册进图引擎**。这行代码执行后，图引擎就知道"运行前后应该找这个对象存取状态"，但此时仓库里仍然一条记录都没有。`thread_id` 是在每次调用 `invoke()` 时才作为"钥匙"传递进去的。

LangGraph 在每次 `invoke` 时会自动触发三个动作：


1. **运行前**：调用 `checkpointer.get(config)` → 查询 "thread_id=42 的最新检查点"

2. **运行中**：每完成一个节点，调用 `checkpointer.put(config, state)` → 保存中间快照

3. **运行后**：调用 `checkpointer.put(config, final_state)` → 保存最终状态


因此，`InMemorySaver` 的职责不是"持有所有线程的状态"，而是"在收到请求时帮你存、在收到请求时帮你取"。状态的生命周期完全由你传入的 `thread_id` 决定。

### 8. 执行与跨轮次记忆


```
from langchain.messages import HumanMessage

# 第一次调用：询问上海天气
final_state = app.invoke(
    {"messages": [HumanMessage(content="上海的天气怎么样？")]},
    config={"configurable": {"thread_id": 42}}
)
result = final_state['messages'][-1].content
print(result)

# 第二次调用：同一个 thread_id，测试记忆
final_state = app.invoke(
    {"messages": [HumanMessage(content="我问的哪一个城市？")]},
    config={"configurable": {"thread_id": 42}}
)
result = final_state['messages'][-1].content
print(result)
```
`app.invoke()` 触发图的完整运行。`config` 参数里的 `thread_id` 是**记忆键（Memory Key）**。它的作用相当于一把钥匙——你每次调用时把钥匙交给 `InMemorySaver`，它凭这把钥匙去仓库里找对应的历史快照。

**第一次调用（thread_id=42）**

```
你传入 config={"configurable": {"thread_id": 42}}
         │
         ▼
LangGraph 问 InMemorySaver: "42 号仓库有货吗？"
         │
         ▼
InMemorySaver 查字典 → 没有记录 → 返回 None
         │
         ▼
图从零开始运行: agent → tools → agent → END
         │
         ▼
运行结束，LangGraph 把最终 state 交给 InMemorySaver
         │
         ▼
InMemorySaver 存入字典: {42: [checkpoint_0]}
```
**第二次调用（thread_id=42）**

```
你传入 config={"configurable": {"thread_id": 42}}  (同一把钥匙)
         │
         ▼
LangGraph 问 InMemorySaver: "42 号仓库有货吗？"
         │
         ▼
InMemorySaver 查字典 → 有！返回 checkpoint_0
         │
         ▼
LangGraph 把 checkpoint_0 的消息列表作为初始 state
         │
         ▼
你的新消息 "我问的哪一个城市？" 被追加到列表末尾
         │
         ▼
图继续运行: agent → END (模型看到历史里有上海，直接回答)
         │
         ▼
运行结束，更新后的 state 再次存入 {42: [checkpoint_0, checkpoint_1]}
```
两次调用使用同一个 `thread_id`，效果就像"挂断电话后重新拨通同一个分机"——对方还记得你们刚才聊到哪里。如果把第二次的 `thread_id` 改成 `43`，`InMemorySaver` 会查到空记录，模型就会回答"我不确定你指的是哪个城市"，或者重新调用搜索工具。

这就是 LangGraph 解决"跨轮次记忆"问题的核心机制：**状态不是存在 Python 变量里，而是存在图的外部仓库里，用 thread_id 作为索引键**。

## 四、核心组件一览


|组件|类/函数/常量|作用|
|:-:|:-:|:-:|
|工具装饰器|`@tool`|把 Python 函数标记为 LLM 可调用的工具，自动生成工具描述|
|工具节点|`ToolNode`|预置节点，自动完成工具调用的参数解析、分发与执行|
|状态模式|`MessagesState`|规定图中流转的数据结构，内置消息列表与追加归约器|
|图构建器|`StateGraph`|注册节点、连接边、构建完整流程图|
|条件路由函数|`should_continue`|根据运行时状态（如 `tool_calls`）决定下一跳目标|
|检查点|`InMemorySaver`|在内存中持久化状态，实现跨轮次记忆恢复|
|编译输出|`compile()`|把静态图编译成可运行的 `Runnable` 对象|
|起始标记|`START`|表示图的起始点，用于 `add_edge(START, node)`|
|终止标记|`END`|表示图的终止点，条件函数返回它时流程结束|

## 五、试一试


在运行前，请确保已配置好 OpenAI 的 API Key 环境变量，或在项目根目录放置 `.env` 文件：

```
OPENAI_API_KEY=sk-...
```
执行脚本：

```
cd demo-01
python langgraph-01.py
```
你可以尝试修改 `HumanMessage` 的内容，观察循环的执行：


1. `"上海的天气怎么样？"` —— 模型会调用 `search` 工具，拿到结果后再组织语言回答。注意观察控制台输出，第一次调用会走 `agent -> tools -> agent` 的循环。

2. `"我问的哪一个城市？"` —— 不调用工具，直接基于记忆回答"上海"。这是因为 `thread_id=42` 把上一轮的状态带了进来。

3. `"北京呢？"` —— 模型可能会再次调用 `search`，查询北京的天气，并再次回到 `agent` 组织答案。


### 完整代码


```
from typing import Literal
from langchain.tools import tool
from langgraph.prebuilt import ToolNode
from langchain.chat_models import ChatOpenAI
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import MessagesState, END, StateGraph, START


@tool
def search(query: str) -> str:
    """模拟一个搜索工具"""
    if "上海" in query.lower() or "Shanghai" in query.lower():
        return "现在30°，有雾"
    return "现在温度35度"


# 将工具放入列表
tools = [search]

# 创建工具节点
tool_node = ToolNode(tools)

# 初始化模型和工具，定义并绑定工具到模型
model = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)


# 定义函数，决定是否要继续执行
def should_continue(state: MessagesState) -> Literal["tools", END]:
    messages = state['messages']
    last_message = messages[-1]

    # 如果 LLM 调用了工具，则转到 Tools 节点
    if last_message.tool_calls:
        return "tools"
    # 否则停止
    return END


# 定义调用模型的函数
def call_model(state: MessagesState):
    messages = state['messages']
    response = model.invoke(messages)
    # 返回列表，LangGraph 会自动合并到现有状态中
    return {"messages": [response]}


# 用状态初始化图，定义一个新的状态图
workflow = StateGraph(MessagesState)

# 定义图节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# 定义入口点和图边
workflow.add_edge(START, "agent")

# 添加条件边
workflow.add_conditional_edges(
    "agent",
    should_continue
)

# 添加从 tools 到 agent 的普通边
workflow.add_edge("tools", "agent")

# 初始化内存以在图运行之间持久化状态
checkpointer = InMemorySaver()

# 编译图，这将其编译成一个 LangChain 可运行对象
app = workflow.compile(checkpointer=checkpointer)

# 执行图：第一次调用
final_state = app.invoke(
    {"messages": [HumanMessage(content="上海的天气怎么样？")]},
    config={"configurable": {"thread_id": 42}}
)

# 拿到最后一条消息
result = final_state['messages'][-1].content
print(result)

# 执行图：第二次调用，同一个 thread_id，测试记忆能力
final_state = app.invoke(
    {"messages": [HumanMessage(content="我问的哪一个城市？")]},
    config={"configurable": {"thread_id": 42}}
)

result = final_state['messages'][-1].content
print(result)

# 将生成的图片保存到文件夹
graph_png = app.get_graph().draw_mermaid_png()

with open("langgraph_hello.png", "wb") as f:
    f.write(graph_png)
```
## 六、其他


### 可视化：把图画出来


LangGraph 内置了 Mermaid 图表生成功能，只需要两行代码就能导出流程图：

```
graph_png = app.get_graph().draw_mermaid_png()

with open("langgraph_hello.png", "wb") as f:
    f.write(graph_png)
```
生成的 PNG 会清晰展示 `agent`、`tools`、`__start__`、`__end__` 四个节点，以及它们之间的条件边和普通边。`__start__` 对应 `START`，`__end__` 对应 `END`。这对于调试复杂的多节点图非常有帮助——你可以一眼看出循环在哪里，条件分支走向哪里。

### 从内存到持久化


本例使用的是 `InMemorySaver`，状态只保存在内存中，进程重启后丢失。在生产环境中，你可以替换为 `SqliteSaver` 或 `RedisSaver`，把检查点写入数据库或缓存，实现真正的长期记忆。

```
# 示例：使用 SQLite 持久化
from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string(":memory:") as checkpointer:
    app = workflow.compile(checkpointer=checkpointer)
    # ... invoke ...
```
检查点机制是 LangGraph 的灵魂。它让 Agent 从一个"无状态的函数调用链"升级为一个"有状态、可中断、可恢复"的长期会话系统。这也是 LangGraph 相比纯 LangChain 最核心的优势所在。

### 验证记忆机制


如果你还是不确定 `thread_id` 到底起什么作用，可以用下面这段代码亲自验证：

```
# 实验 1：相同 thread_id，应该有记忆
app.invoke(
    {"messages": [HumanMessage(content="记住我的名字叫 Alice")]},
    config={"configurable": {"thread_id": "alice_session"}}
)

result = app.invoke(
    {"messages": [HumanMessage(content="我叫什么名字？")]},
    config={"configurable": {"thread_id": "alice_session"}}  # 同一把钥匙
)
print(result['messages'][-1].content)  # 预期输出: "Alice" 或包含 Alice

# 实验 2：不同 thread_id，应该没有记忆
result = app.invoke(
    {"messages": [HumanMessage(content="我叫什么名字？")]},
    config={"configurable": {"thread_id": "bob_session"}}  # 另一把钥匙
)
print(result['messages'][-1].content)  # 预期输出: 模型表示不知道，或者反问你是谁
```
**验证结论**：


|实验|thread_id|历史状态|模型能否回答"我叫什么"|
|:-:|:-:|:-:|:-:|
|实验 1|`alice_session`|有（第一轮已存入）|✅ 能，基于记忆|
|实验 2|`bob_session`|无（从未运行过）|❌ 不能，从零开始|

`thread_id` 就是一个字符串或数字，你可以用用户 ID、会话 ID、甚至是随机 UUID。LangGraph 对它没有任何语义要求——唯一的要求是：**你想让两次调用共享状态，就传相同的 `thread_id`；你想让它们彼此隔离，就传不同的 `thread_id`**。

"LangChain 负责单次调用，LangGraph 负责把调用串成可循环、可记忆的工作流。"

【LangGraph新手村系列】（1）LangGraph 入门：StateGraph 与带记忆的 ReAct 循环

介绍 LangGraph 核心思想：用 StateGraph 把单次 LLM 调用串成可循环的 ReAct 工作流。通过节点、边与公共状态黑板，实现模型思考、工具调用、条件跳转的闭环，并引入检查点让 Agent 拥有跨轮次记忆。

人工智能

编程语言

开发工具

LangGraph通过状态图(StateGraph)实现Agent工作流的循环执行与状态持久化，解决LangChain线性执行的局限性。核心功能包括工具节点(ToolNode)、条件路由(should_continue)和检查点机制(InMemorySaver)，支持多轮对话记忆和工具调用循环，提升AI代理的交互能力。

数据结构

数据库

路由

SQLite

Python

Agent

JSON

2026采购季 | AI焕新·智启新局

tencentdb-catalog

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云AI代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

功能1上新10个字符

功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符。

功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符。

功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符

功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符

功能4上新

文章&问答评论现已支持表情

全新交互，全新视觉，新增快捷键、悬浮工具栏、高亮块等功能并同时优化现有功能，全面提升创作效率和体验

社区富文本编辑器全新改版！诚邀体验～ 

精选全网热门MCP server，让你的AI更好用 🚀

💥开发者 MCP广场重磅上线！

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【LangGraph新手村系列】（1）LangGraph 入门：StateGraph 与带记忆的 ReAct 循环-腾讯云开发者社区-腾讯云

【LangGraph新手村系列】（1）LangGraph 入门：StateGraph 与带记忆的 ReAct 循环

【LangGraph新手村系列】（1）LangGraph 入门：StateGraph 与带记忆的 ReAct 循环

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐