LangGraph Agent Memory 详细解读

📚 概述

本文档详细解读 LangGraph 中的 Agent Memory（智能体记忆） 机制。这是构建多轮对话系统的核心技术，使得 AI 智能体能够记住之前的对话内容，实现真正的上下文连续性。

🎯 核心概念

什么是 Agent Memory？

Agent Memory 是 LangGraph 提供的持久化机制，允许智能体在多次交互之间保持状态。就像人类的记忆一样，它能让 AI：

记住历史对话：回忆之前说过的话
理解上下文引用：理解"那个"、"它"等指代词
保持连续性：在中断后继续对话
维护会话状态：跟踪任务进度和数据

为什么需要 Memory？

没有 Memory 的问题：

python

# 第一轮对话
user: "3 + 4 等于多少？"
agent: "等于 7"

# 第二轮对话（新的图执行）
user: "把它乘以 2"
agent: "把什么乘以 2？" ❌ 记不住 7

有 Memory 的效果：

python

# 第一轮对话
user: "3 + 4 等于多少？"
agent: "等于 7"

# 第二轮对话（继续之前的状态）
user: "把它乘以 2"
agent: "7 × 2 = 14" ✅ 记住了 7

🔧 技术实现详解

1. 问题根源：状态的临时性

在 LangGraph 中，状态默认是临时的，只在单次图执行期间存在。

python

# 每次调用 invoke() 都是全新的执行
react_graph.invoke({"messages": [msg1]})  # 执行 1：有 msg1
react_graph.invoke({"messages": [msg2]})  # 执行 2：只有 msg2，忘记 msg1

问题示意图：

执行 1                    执行 2
┌─────────────┐          ┌─────────────┐
│ State       │          │ State       │
│ msg: [msg1] │          │ msg: [msg2] │  ← 完全独立，互不相关
└─────────────┘          └─────────────┘

2. 解决方案：Checkpointer

LangGraph 提供 Checkpointer（检查点器） 机制，在每个步骤后自动保存状态。

核心原理

python

from langgraph.checkpoint.memory import MemorySaver

# 创建内存检查点器
memory = MemorySaver()

# 编译图时注入检查点器
graph_with_memory = builder.compile(checkpointer=memory)

工作流程：

用户输入 → [节点 1] → 💾 保存状态 → [节点 2] → 💾 保存状态 → 输出
                 ↓                          ↓
            Checkpoint 1               Checkpoint 2

每个节点执行后，当前状态都会被保存到检查点。

3. Thread：会话线程

为了管理多个独立的对话，LangGraph 使用 Thread（线程） 概念。

Thread ID 的作用

python

# 配置：指定线程 ID
config = {"configurable": {"thread_id": "1"}}

# 第一轮对话
graph.invoke({"messages": [msg1]}, config)  # 保存到 thread_id=1

# 第二轮对话（同一线程）
graph.invoke({"messages": [msg2]}, config)  # 继续 thread_id=1 的状态

Thread 示意图：

Thread ID: 1                    Thread ID: 2
┌─────────────────┐            ┌─────────────────┐
│ Checkpoint 1    │            │ Checkpoint 1    │
│ Checkpoint 2    │            │ Checkpoint 2    │
│ Checkpoint 3    │            └─────────────────┘
└─────────────────┘
  ↑ 用户 A 的对话                 ↑ 用户 B 的对话（独立）

每个 thread_id 维护独立的对话历史，互不干扰。

📖 完整代码实现

1. 环境设置

python

# 安装依赖
%pip install --quiet -U langchain_openai langchain_core langgraph

# 设置 API Key
import os, getpass

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

_set_env("OPENAI_API_KEY")
_set_env("LANGSMITH_API_KEY")

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "langchain-academy"

说明：

OPENAI_API_KEY：调用 GPT 模型
LANGSMITH_API_KEY：启用 LangSmith 追踪（可视化执行流程）

2. 定义工具

python

from langchain_openai import ChatOpenAI

def add(a: int, b: int) -> int:
    """Adds a and b.

    Args:
        a: first int
        b: second int
    """
    return a + b

def multiply(a: int, b: int) -> int:
    """Multiply a and b.

    Args:
        a: first int
        b: second int
    """
    return a * b

def divide(a: int, b: int) -> float:
    """Divide a and b.

    Args:
        a: first int
        b: second int
    """
    return a / b

# 工具列表
tools = [add, multiply, divide]

# 创建 LLM 并绑定工具
llm = ChatOpenAI(model="gpt-4o")
llm_with_tools = llm.bind_tools(tools)

Python 知识点：Docstring 的重要性

python

def add(a: int, b: int) -> int:
    """Adds a and b.  # ← LLM 会读取这段描述

    Args:
        a: first int   # ← 参数说明帮助 LLM 理解如何调用
        b: second int
    """
    return a + b

LLM 通过 Docstring 理解工具功能
清晰的描述能提高工具调用的准确性
遵循 Google 风格的 Docstring 规范

3. 定义状态和节点

python

from langgraph.graph import MessagesState
from langchain_core.messages import HumanMessage, SystemMessage

# 系统消息
sys_msg = SystemMessage(
    content="You are a helpful assistant tasked with performing arithmetic on a set of inputs."
)

# 定义助手节点
def assistant(state: MessagesState):
    return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}

LangGraph 知识点：MessagesState

MessagesState 是 LangGraph 预定义的状态类，专门用于消息管理：

python

class MessagesState(TypedDict):
    messages: Annotated[list, add_messages]
    #          ^^^^^^^^^^^^^^^^^^^^^^^^^^
    #          自动处理消息列表的追加和去重

特性：

自动追加：新消息会自动添加到列表末尾
智能去重：相同 ID 的消息会被更新而非重复
便捷访问：直接通过 state["messages"] 访问

4. 构建图（无 Memory）

python

from langgraph.graph import START, StateGraph
from langgraph.prebuilt import tools_condition, ToolNode
from IPython.display import Image, display

# 创建图构建器
builder = StateGraph(MessagesState)

# 添加节点
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))

# 添加边
builder.add_edge(START, "assistant")

# 条件边：决定是否调用工具
builder.add_conditional_edges(
    "assistant",
    tools_condition,  # 如果 assistant 返回工具调用 → tools 节点
                      # 否则 → END
)

builder.add_edge("tools", "assistant")

# 编译（无 Memory）
react_graph = builder.compile()

# 可视化
display(Image(react_graph.get_graph(xray=True).draw_mermaid_png()))

LangGraph 知识点：tools_condition

tools_condition 是 LangGraph 预定义的条件函数：

python

def tools_condition(state: MessagesState) -> Literal["tools", "__end__"]:
    """检查最后一条消息是否包含工具调用"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"  # 有工具调用 → 进入 tools 节点
    return "__end__"    # 无工具调用 → 结束

ToolNode 的作用：

python

ToolNode(tools)

自动执行工具调用
将工具输出包装成 ToolMessage
添加到消息列表中

5. 测试无 Memory 的问题

第一轮对话

python

messages = [HumanMessage(content="Add 3 and 4.")]
messages = react_graph.invoke({"messages": messages})

for m in messages['messages']:
    m.pretty_print()

输出：

================================ Human Message =================================
Add 3 and 4.

================================== Ai Message ==================================
Tool Calls:
  add (call_zZ4JPASfUinchT8wOqg9hCZO)
  Args:
    a: 3
    b: 4

================================= Tool Message =================================
Name: add
7

================================== Ai Message ==================================
The sum of 3 and 4 is 7.

第二轮对话（问题出现）

python

messages = [HumanMessage(content="Multiply that by 2.")]
messages = react_graph.invoke({"messages": messages})

for m in messages['messages']:
    m.pretty_print()

输出：

================================ Human Message =================================
Multiply that by 2.

================================== Ai Message ==================================
Tool Calls:
  multiply (call_prnkuG7OYQtbrtVQmH2d3Nl7)
  Args:
    a: 2  # ❌ 错误！应该是 7
    b: 2

================================= Tool Message =================================
Name: multiply
4

================================== Ai Message ==================================
The result of multiplying 2 by 2 is 4.  # ❌ 应该是 14

问题分析：

LLM 不知道"那个"（that）指的是什么
因为每次 invoke() 都是全新的执行，没有历史上下文
LLM 只看到 "Multiply that by 2."，无法理解 "that" 的含义

6. 添加 Memory

python

from langgraph.checkpoint.memory import MemorySaver

# 创建内存检查点器
memory = MemorySaver()

# 重新编译图（带 Memory）
react_graph_memory = builder.compile(checkpointer=memory)

MemorySaver 详解：

特性	说明
类型	内存键值存储
持久化	仅在进程运行期间
用途	开发、测试、演示
生产替代	PostgresSaver, RedisSaver 等

为什么不用于生产？

python

memory = MemorySaver()  # 数据存在内存中

# 问题：
# 1. 进程重启 → 数据丢失
# 2. 多进程部署 → 数据不共享
# 3. 无法持久化存储

生产环境推荐：

python

from langgraph.checkpoint.postgres import PostgresSaver

# 连接数据库
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/db"
)

graph = builder.compile(checkpointer=checkpointer)

7. 使用 Thread ID

python

# 配置线程 ID
config = {"configurable": {"thread_id": "1"}}

# 第一轮对话
messages = [HumanMessage(content="Add 3 and 4.")]
messages = react_graph_memory.invoke({"messages": messages}, config)

for m in messages['messages']:
    m.pretty_print()

输出：

================================ Human Message =================================
Add 3 and 4.

================================== Ai Message ==================================
Tool Calls:
  add (call_MSupVAgej4PShIZs7NXOE6En)
  Args:
    a: 3
    b: 4

================================= Tool Message =================================
Name: add
7

================================== Ai Message ==================================
The sum of 3 and 4 is 7.

检查点保存示意：

Thread ID: 1
├─ Checkpoint 0: {"messages": [HumanMessage("Add 3 and 4.")]}
├─ Checkpoint 1: {"messages": [..., AIMessage(tool_calls=[...])]}
├─ Checkpoint 2: {"messages": [..., ToolMessage("7")]}
└─ Checkpoint 3: {"messages": [..., AIMessage("The sum is 7.")]}

8. 继续对话（Memory 生效）

python

# 同一线程，新消息
messages = [HumanMessage(content="Multiply that by 2.")]
messages = react_graph_memory.invoke({"messages": messages}, config)

for m in messages['messages']:
    m.pretty_print()

输出：

================================ Human Message =================================
Add 3 and 4.
                                   ↑ 历史消息（从检查点加载）

================================== Ai Message ==================================
Tool Calls:
  add (...)
  Args:
    a: 3
    b: 4

================================= Tool Message =================================
Name: add
7

================================== Ai Message ==================================
The sum of 3 and 4 is 7.

================================ Human Message =================================
Multiply that by 2.
                   ↑ 新消息（追加到历史）

================================== Ai Message ==================================
Tool Calls:
  multiply (call_fWN7lnSZZm82tAg7RGeuWusO)
  Args:
    a: 7  # ✅ 正确！记住了之前的 7
    b: 2

================================= Tool Message =================================
Name: multiply
14

================================== Ai Message ==================================
The result of multiplying 7 by 2 is 14.  # ✅ 正确答案

关键观察：

历史消息自动加载：输出包含了第一轮对话的所有消息
上下文理解：LLM 知道 "that" 指的是 7
正确计算：7 × 2 = 14

🎓 核心知识点总结

LangGraph 特有概念

1. Checkpointer（检查点器）

作用： 在每个节点执行后自动保存状态

python

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

保存时机：

节点执行前    节点执行       节点执行后
    ↓          ↓              ↓
 加载状态   → 处理数据   →  💾 保存状态

2. Thread（线程）

作用： 隔离不同用户/会话的状态

python

# 用户 A
config_a = {"configurable": {"thread_id": "user_a"}}
graph.invoke(input, config_a)

# 用户 B（完全独立）
config_b = {"configurable": {"thread_id": "user_b"}}
graph.invoke(input, config_b)

Thread 层级结构：

Checkpointer（全局）
├─ Thread: user_a
│  ├─ Checkpoint 1
│  ├─ Checkpoint 2
│  └─ Checkpoint 3
├─ Thread: user_b
│  ├─ Checkpoint 1
│  └─ Checkpoint 2
└─ Thread: user_c
   └─ Checkpoint 1

3. Config（配置）

python

config = {
    "configurable": {
        "thread_id": "1",           # 必需：线程标识
        "checkpoint_id": "...",     # 可选：特定检查点
        "checkpoint_ns": "..."      # 可选：检查点命名空间
    }
}

configurable 的作用：

传递运行时配置
不影响状态结构
用于检查点管理

Python 特有知识点

1. Type Hints（类型提示）

python

def add(a: int, b: int) -> int:
    #      ^^^^^  ^^^^^    ^^^
    #      参数类型        返回类型
    return a + b

优势：

IDE 支持：自动补全、类型检查
文档作用：明确函数签名
LangChain 集成：自动转换为工具模式

2. Docstring（文档字符串）

python

def multiply(a: int, b: int) -> int:
    """Multiply a and b.  # 功能描述

    Args:                 # 参数说明
        a: first int
        b: second int
    """
    return a * b

LangChain 如何使用 Docstring？

python

# LangChain 自动解析函数签名
tools = [multiply]

# 生成工具模式（OpenAI Function Calling 格式）
{
    "name": "multiply",
    "description": "Multiply a and b.",  # ← 来自 Docstring
    "parameters": {
        "type": "object",
        "properties": {
            "a": {"type": "integer", "description": "first int"},
            "b": {"type": "integer", "description": "second int"}
        }
    }
}

3. Literal（字面量类型）

python

from typing import Literal

def route(state) -> Literal["tools", "__end__"]:
    #                  ^^^^^^^^^^^^^^^^^^^^^^^^
    #                  返回值只能是这两个字符串之一
    if condition:
        return "tools"
    return "__end__"

作用：

限制返回值范围
IDE 可以检查拼写错误
提高代码可靠性

💡 最佳实践

1. 何时使用 Memory？

✅ 适用场景：

多轮对话系统（聊天机器人）
需要上下文的任务（"继续"、"那个"等指代）
长期任务追踪（任务进度、用户偏好）
调试和审计（查看历史执行）

❌ 不适用场景：

无状态 API（每次请求独立）
批处理任务（无需记忆）
一次性查询（问答系统的单次问答）

2. Thread ID 设计原则

原则 1：按用户隔离

python

# ✅ 好的设计
config = {"configurable": {"thread_id": f"user_{user_id}"}}

# ❌ 不好的设计
config = {"configurable": {"thread_id": "global"}}  # 所有用户共享

原则 2：按会话隔离

python

# ✅ 更细粒度的隔离
import uuid

session_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": f"{user_id}_{session_id}"}}

原则 3：有意义的命名

python

# ✅ 清晰的命名
thread_id = f"support_ticket_{ticket_id}"
thread_id = f"order_processing_{order_id}"

# ❌ 无意义的命名
thread_id = "123"
thread_id = "abc"

3. Checkpointer 选择

场景	推荐 Checkpointer	理由
开发/测试	`MemorySaver`	简单、快速
生产（小规模）	`SqliteSaver`	持久化、单机部署
生产（分布式）	`PostgresSaver`	支持并发、可扩展
高性能	`RedisSaver`	低延迟、高吞吐

示例：生产环境配置

python

from langgraph.checkpoint.postgres import PostgresSaver

# 数据库连接
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/langchain_db"
)

# 编译图
graph = builder.compile(checkpointer=checkpointer)

# 使用（与 MemorySaver 完全相同）
config = {"configurable": {"thread_id": "user_123"}}
graph.invoke(input, config)

🚀 进阶技巧

1. 检查点历史查询

python

# 获取线程的所有检查点
config = {"configurable": {"thread_id": "1"}}

# 列出所有检查点
checkpoints = list(graph.checkpointer.list(config))

for checkpoint in checkpoints:
    print(f"Checkpoint ID: {checkpoint.id}")
    print(f"State: {checkpoint.state}")

2. 回溯到特定检查点

python

# 获取特定检查点的状态
config = {
    "configurable": {
        "thread_id": "1",
        "checkpoint_id": "abc123"  # 指定检查点
    }
}

# 从这个检查点继续执行
graph.invoke(new_input, config)

3. 清理旧检查点

python

# 删除线程的所有检查点
config = {"configurable": {"thread_id": "1"}}
graph.checkpointer.delete_thread(config)

4. 条件性保存

python

# 只在特定条件下保存检查点
def conditional_checkpoint(state):
    if state.get("important"):
        # 保存
        return state
    else:
        # 跳过保存（通过自定义逻辑）
        pass

📊 Memory vs No Memory 对比

特性	无 Memory	有 Memory
状态持久化	❌	✅
多轮对话	❌	✅
上下文引用	❌	✅
调试便利性	❌	✅
性能开销	低	中（取决于 Checkpointer）
存储需求	无	有
适用场景	单次查询	对话系统

🎯 实际应用案例

案例 1：客服机器人

python

# 用户开始对话
config = {"configurable": {"thread_id": f"customer_{customer_id}"}}

# 第 1 轮
user: "我的订单号是 12345"
agent: "好的，我看到您的订单 12345 了"

# 第 2 轮（几分钟后）
user: "那个订单发货了吗？"
agent: "订单 12345 已于今天上午发货"  # ✅ 记住了订单号

案例 2：多步骤任务

python

# 数据分析任务
config = {"configurable": {"thread_id": f"analysis_{task_id}"}}

# 步骤 1
user: "加载 sales.csv"
agent: "已加载 1000 条销售记录"

# 步骤 2
user: "计算总收入"
agent: "总收入：$50,000"  # ✅ 记住了加载的数据

# 步骤 3
user: "生成图表"
agent: "已生成收入趋势图"  # ✅ 基于之前的计算

案例 3：用户偏好学习

python

# 长期用户交互
config = {"configurable": {"thread_id": f"user_{user_id}"}}

# 第 1 次交互
user: "我喜欢简洁的回答"
agent: "好的，我会简洁回复"

# 第 100 次交互（数天后）
user: "解释量子计算"
agent: "简单说：用量子态并行计算"  # ✅ 记住了用户偏好

🔍 常见问题

Q1: MemorySaver 和数据库 Checkpointer 有什么区别？

MemorySaver：

数据存在 Python 进程内存中
进程重启 → 数据丢失
适合开发和测试

PostgresSaver：

数据存在数据库中
持久化存储
支持分布式部署
适合生产环境

Q2: 如何实现多用户隔离？

使用不同的 thread_id：

python

# 用户 A
config_a = {"configurable": {"thread_id": "user_123"}}
graph.invoke(input_a, config_a)

# 用户 B（完全独立）
config_b = {"configurable": {"thread_id": "user_456"}}
graph.invoke(input_b, config_b)

Q3: 检查点会占用多少存储空间？

取决于：

消息数量和大小
检查点保留策略
状态复杂度

优化建议：

python

# 1. 定期清理旧线程
if thread_inactive_for_days(30):
    checkpointer.delete_thread(config)

# 2. 限制消息历史长度
def trim_messages(state):
    # 只保留最近 50 条消息
    state["messages"] = state["messages"][-50:]
    return state

Q4: 可以在不同图之间共享 Memory 吗？

可以，只要使用相同的 checkpointer 和 thread_id：

python

# 共享检查点器
shared_memory = MemorySaver()

graph_1 = builder_1.compile(checkpointer=shared_memory)
graph_2 = builder_2.compile(checkpointer=shared_memory)

# 使用相同的 thread_id
config = {"configurable": {"thread_id": "shared_thread"}}

graph_1.invoke(input_1, config)  # 写入状态
graph_2.invoke(input_2, config)  # 读取相同状态

注意： 两个图的状态结构必须兼容！

📖 扩展阅读

🎬 总结

Agent Memory 是 LangGraph 构建多轮对话系统的基础能力：

Checkpointer 自动保存每个步骤的状态
Thread ID 隔离不同用户/会话的数据
Config 传递运行时配置，控制检查点行为

通过这些机制，我们可以构建真正具有"记忆"的智能体，实现自然、连贯的多轮交互！

核心要点：

开发用 MemorySaver，生产用 PostgresSaver
每个用户/会话使用唯一的 thread_id
定期清理无用的检查点数据
结合 LangSmith 追踪，调试 Memory 行为

LangGraph Agent Memory 详细解读 ​

📚 概述 ​

🎯 核心概念 ​

什么是 Agent Memory？ ​

为什么需要 Memory？ ​

🔧 技术实现详解 ​

1. 问题根源：状态的临时性 ​

2. 解决方案：Checkpointer ​

核心原理 ​

3. Thread：会话线程 ​

Thread ID 的作用 ​

📖 完整代码实现 ​

1. 环境设置 ​

2. 定义工具 ​

3. 定义状态和节点 ​

4. 构建图（无 Memory） ​

5. 测试无 Memory 的问题 ​

第一轮对话 ​

第二轮对话（问题出现） ​

6. 添加 Memory ​

7. 使用 Thread ID ​

8. 继续对话（Memory 生效） ​

🎓 核心知识点总结 ​

LangGraph 特有概念 ​

1. Checkpointer（检查点器） ​

2. Thread（线程） ​

3. Config（配置） ​

Python 特有知识点 ​

1. Type Hints（类型提示） ​

2. Docstring（文档字符串） ​

3. Literal（字面量类型） ​

💡 最佳实践 ​

1. 何时使用 Memory？ ​

2. Thread ID 设计原则 ​

原则 1：按用户隔离 ​

原则 2：按会话隔离 ​

原则 3：有意义的命名 ​

3. Checkpointer 选择 ​

🚀 进阶技巧 ​

1. 检查点历史查询 ​

2. 回溯到特定检查点 ​

3. 清理旧检查点 ​

4. 条件性保存 ​

📊 Memory vs No Memory 对比 ​

🎯 实际应用案例 ​

案例 1：客服机器人 ​

案例 2：多步骤任务 ​

案例 3：用户偏好学习 ​

🔍 常见问题 ​

Q1: MemorySaver 和数据库 Checkpointer 有什么区别？ ​

Q2: 如何实现多用户隔离？ ​

Q3: 检查点会占用多少存储空间？ ​

Q4: 可以在不同图之间共享 Memory 吗？ ​

📖 扩展阅读 ​

🎬 总结 ​

LangGraph Agent Memory 详细解读

📚 概述

🎯 核心概念

什么是 Agent Memory？

为什么需要 Memory？

🔧 技术实现详解

1. 问题根源：状态的临时性

2. 解决方案：Checkpointer

核心原理

3. Thread：会话线程

Thread ID 的作用

📖 完整代码实现

1. 环境设置

2. 定义工具

3. 定义状态和节点

4. 构建图（无 Memory）

5. 测试无 Memory 的问题

第一轮对话

第二轮对话（问题出现）

6. 添加 Memory

7. 使用 Thread ID

8. 继续对话（Memory 生效）

🎓 核心知识点总结

LangGraph 特有概念

1. Checkpointer（检查点器）

2. Thread（线程）

3. Config（配置）

Python 特有知识点

1. Type Hints（类型提示）

2. Docstring（文档字符串）

3. Literal（字面量类型）

💡 最佳实践

1. 何时使用 Memory？

2. Thread ID 设计原则

原则 1：按用户隔离

原则 2：按会话隔离

原则 3：有意义的命名

3. Checkpointer 选择

🚀 进阶技巧

1. 检查点历史查询

2. 回溯到特定检查点

3. 清理旧检查点

4. 条件性保存

📊 Memory vs No Memory 对比

🎯 实际应用案例

案例 1：客服机器人

案例 2：多步骤任务

案例 3：用户偏好学习

🔍 常见问题

Q1: MemorySaver 和数据库 Checkpointer 有什么区别？

Q2: 如何实现多用户隔离？

Q3: 检查点会占用多少存储空间？

Q4: 可以在不同图之间共享 Memory 吗？

📖 扩展阅读

🎬 总结