没有记忆的 Agent 每次对话都从零开始。通过 InMemorySaver + thread_id,让 Agent 记住每个用户的对话历史。
默认情况下,create_agent 创建的 Agent没有记忆。每次 agent.invoke() 都是完全独立的:Agent 不知道之前说过什么,也不知道用户叫什么名字。
agent = create_agent(deepseek_model())
# 第一次对话:告诉 Agent 我的名字和爱好
response1 = agent.invoke({
"messages": [HumanMessage(content="Hello my name is Seán and my favourite colour is green")]
})
print(response1['messages'][-1].content)
# 第二次对话(独立的 invoke):问 Agent 我的爱好
response2 = agent.invoke({
"messages": [HumanMessage(content="What's my favourite colour?")]
})
print(response2['messages'][-1].content)
第二次调用时,Agent 完全不记得第一次对话——它不知道用户叫 Seán,也不知道最爱绿色。每次 invoke 都是一个全新的、无上下文的对话。
LangGraph 的记忆机制叫做 Checkpointer(检查点)。它的工作原理很简单:
agent.invoke() 执行后,Checkpointer 把完整的 State(包括所有消息历史)保存到存储里thread_id 调用时,Checkpointer 先从存储里加载之前的 StateCheckpointer 不会让 LLM 本身记住什么——LLM 每次调用都是无状态的。Checkpointer 做的是:把历史消息作为上下文一起传给 LLM,让 LLM 能"看到"之前说过什么。本质上是把历史消息放进了这次调用的上下文窗口。
| Checkpointer 类型 | 存储位置 | 适用场景 |
|---|---|---|
InMemorySaver |
Python 进程内存 | 开发/测试,进程重启后消失 |
SqliteSaver |
SQLite 文件 | 本地持久化,单机生产环境 |
PostgresSaver |
PostgreSQL 数据库 | 多实例、高并发生产环境 |
| LangGraph Cloud | 托管云存储 | 完全托管的云端部署 |
在 create_agent 里传入 checkpointer=InMemorySaver(),即可为 Agent 开启记忆功能:
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
agent = create_agent(
deepseek_model(),
checkpointer=InMemorySaver(), # 开启内存记忆
)
光有 checkpointer 还不够——还需要通过 config 参数传入 thread_id,告诉 Agent 这次对话属于哪个"线程"(即哪个用户)。
thread_id 是对话的唯一标识符。同一个 thread_id 的多次 invoke 共享消息历史;不同 thread_id 的对话完全隔离,互不干扰。
# config 字典:指定这次 invoke 属于哪个线程
config = {"configurable": {"thread_id": "1"}}
# 第一次调用:告知名字和爱好
response1 = agent.invoke(
{"messages": [HumanMessage(content="Hello my name is Seán and my favourite colour is green")]},
config # 传入 config
)
# 第二次调用:同一个 config(同一个 thread_id)
response2 = agent.invoke(
{"messages": [HumanMessage(content="What's my favourite colour?")]},
config # 同一个 config = 同一个对话历史
)
print(response2['messages'][-1].content)
在实际应用中,thread_id 通常是用户 ID 或会话 ID,例如 "user_12345" 或 "session_abc123"。这样不同用户各自有独立的对话历史,互不干扰。同一个用户的多次对话(同一个 session)用相同的 thread_id,就能持续记住上下文。
下面是完整的两轮对话代码,展示记忆在实际多轮对话中的工作效果:
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from pprint import pprint
# 创建有记忆的 Agent
agent = create_agent(
deepseek_model(),
checkpointer=InMemorySaver()
)
config = {"configurable": {"thread_id": "1"}}
# 轮次 1:建立上下文
response1 = agent.invoke(
{"messages": [{"role": "user", "content": "Hello my name is Seán and my favourite colour is green"}]},
config
)
# 轮次 2:利用上下文
response2 = agent.invoke(
{"messages": [{"role": "user", "content": "What's my favourite colour?"}]},
config
)
# 查看第二次 invoke 的完整消息历史
for msg in response2["messages"]:
pprint(msg.content)
print()
注意消息历史包含了两次 invoke 的所有消息(共 4 条)。第二次调用时,Agent 看到了第一次对话里 Seán 提到的绿色,所以能正确回答。
使用有记忆的 Agent 时,每次 invoke 只需要传当前轮次的新消息,不需要手动拼接历史消息。Checkpointer 会自动把历史加载进来。如果你手动把历史消息也传进去,消息就会重复出现两次,导致行为异常。
| 策略 | 原理 | 适用场景 |
|---|---|---|
| InMemorySaver | 全消息历史存内存 | 开发测试、Demo |
| SqliteSaver / PostgresSaver | 全消息历史存数据库 | 持久化需求、中小型应用 |
| 消息摘要 | 定期把历史消息压缩为摘要,节省上下文空间 | 长期对话,上下文有限的模型 |
| 滑动窗口 | 只保留最近 N 条消息 | 只需要短期上下文的场景 |
| 向量数据库记忆 | 历史对话嵌入存向量库,检索相关历史 | 超长期对话,智能记忆检索 |
checkpointer=InMemorySaver()config = {"configurable": {"thread_id": "1"}}