LangChain 对话记忆 -- Checkpointer

默认情况下，每次 agent.invoke() 都是独立的——Agent 不记得之前聊过什么。Checkpointer（检查点保存器）让 Agent 能够记住对话历史，实现真正的多轮对话。

没有 Checkpointer 的问题

先看看没有 Checkpointer 时的情况：

实例

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
system_prompt="你是菜鸟教程 RUNOOB 的助手。",
)

# 第一轮
result1 = agent.invoke({
"messages": [HumanMessage(content="我叫小明")]
})
print(f"第一轮: {result1['messages'][-1].content}")

# 第二轮——Agent 不记得第一轮的内容！
result2 = agent.invoke({
"messages": [HumanMessage(content="我叫什么名字？")]
})
print(f"第二轮: {result2['messages'][-1].content}")

运行结果：

第一轮: 你好小明！很高兴认识你。
第二轮: 抱歉，我没有你的信息，不知道你叫什么名字。

使用 Checkpointer 记住对话

添加 Checkpointer 后，同一 thread_id 下的对话会自动关联：

实例

from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

# 创建一个内存 Checkpointer
checkpointer = InMemorySaver()

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
checkpointer=checkpointer, # 传入 Checkpointer
system_prompt="你是菜鸟教程 RUNOOB 的助手。",
)

# 使用 thread_id 来标识对话线程
config = {"configurable": {"thread_id": "user-001"}}

# 第一轮
result1 = agent.invoke(
{"messages": [HumanMessage(content="我叫小明，我在学 Python")]},
config=config,
)
print(f"第一轮: {result1['messages'][-1].content}")

# 第二轮——使用相同的 thread_id，Agent 记住了！
result2 = agent.invoke(
{"messages": [HumanMessage(content="我叫什么名字？我在学什么？")]},
config=config,
)
print(f"第二轮: {result2['messages'][-1].content}")

运行结果：

第一轮: 你好小明！Python 是一门很好的入门语言，有什么需要帮助的吗？
第二轮: 你叫小明，你正在学习 Python。有什么具体问题我可以帮你吗？

thread_id 是关键。同一个 thread_id 下的对话是连续的，不同 thread_id 之间的对话完全隔离。这让你可以用一个 Agent 实例同时服务多个用户。

Checkpointer 的工作原理

Checkpointer 在每次 Agent 执行后自动保存状态快照（checkpoint）。下一次使用相同 thread_id 调用时，自动从最近的 checkpoint 恢复状态。

具体工作流程：

调用 agent.invoke()，传入 config（含 thread_id）
Agent 检查是否有该 thread_id 的 checkpoint
如果有，加载历史消息，追加新消息后继续
执行完成后，自动保存新的 checkpoint

实例

from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

checkpointer = InMemorySaver()
agent = create_agent(
model=init_chat_model("deepseek:deepseek-v4-flash", temperature=0),
checkpointer=checkpointer,
system_prompt="你是菜鸟教程 RUNOOB 的助手。",
)

config = {"configurable": {"thread_id": "demo-001"}}

# 模拟多轮对话
questions = [
"我叫小明",
"我在学 Python",
"帮我总结一下关于我的信息",
]

for i, q in enumerate(questions, 1):
result = agent.invoke(
{"messages": [HumanMessage(content=q)]},
config=config,
)

# 查看 checkpoint 状态
state = agent.get_state(config)
print(f"\n第 {i} 轮后:")
print(f" 消息数: {len(state.values.get('messages', []))}")
print(f" 下一步: {state.next}")
print(f" 回复: {result['messages'][-1].content[:80]}...")

运行结果：

第 1 轮后:
  消息数: 2
  下一步: ()
  回复: 你好小明！很高兴认识你。

第 2 轮后:
  消息数: 4
  下一步: ()
  回复: Python 是一门非常流行的编程语言，简单易学。

第 3 轮后:
  消息数: 6
  下一步: ()
  回复: 根据我们的对话，你的信息如下：你叫小明，正在学习 Python。

Checkpointer 类型

类型	存储位置	持久化	适用场景
InMemorySaver	内存	否（重启丢失）	开发调试、测试
SqliteSaver	SQLite 文件	是	单机部署、小规模应用
PostgresSaver	PostgreSQL	是	生产环境、多实例共享

SqliteSaver 示例

实例

from langgraph.checkpoint.sqlite import SqliteSaver

# SQLite 持久化——重启后数据不丢失
checkpointer = SqliteSaver.from_conn_string("conversations.db")

agent = create_agent(
model="deepseek:deepseek-v4-flash",
checkpointer=checkpointer,
)

# 用法和 InMemorySaver 完全相同
config = {"configurable": {"thread_id": "user-001"}}
result = agent.invoke(
{"messages": [{"role": "user", "content": "你好"}]},
config=config,
)

# 重启程序后，相同 thread_id 的对话依然存在

管理对话线程

查看对话状态

实例

# 查看对话状态
state = agent.get_state(config)
print(f"下一步: {state.next}") # () 表示空闲
print(f"消息数: {len(state.values.get('messages', []))}")

# 查看对话历史
for msg in state.values.get("messages", []):
print(f" [{msg.type}] {str(msg.content)[:60]}")

创建新线程

实例

# 不同的 thread_id = 不同的独立对话
config_alice = {"configurable": {"thread_id": "alice"}}
config_bob = {"configurable": {"thread_id": "bob"}}

# Alice 的对话
agent.invoke(
{"messages": [HumanMessage(content="我是 Alice")]},
config=config_alice,
)

# Bob 的对话——完全独立，不知道 Alice 说了什么
agent.invoke(
{"messages": [HumanMessage(content="我是 Bob")]},
config=config_bob,
)

# 验证隔离性
alice_state = agent.get_state(config_alice)
bob_state = agent.get_state(config_bob)
print(f"Alice 对话消息数: {len(alice_state.values['messages'])}")
print(f"Bob 对话消息数: {len(bob_state.values['messages'])}")

更新状态——手动修改对话

有时你需要手动修改对话状态，比如清空对话、插入系统消息等：

实例

from langchain.messages import SystemMessage

# 更新状态——插入一条系统消息
agent.update_state(
config,
{
"messages": [
SystemMessage(content="（用户升级到了 VIP 会员）")
]
}
)

# 之后的对话会包含这条插入的消息

update_state() 的参数会通过 add_messages reducer 处理（对 messages 字段而言），所以新消息会追加而不是覆盖。如果需要覆盖整个状态，需要使用不同的方法。

返回顶部