【 LangChain v1.2 入门系列教程】【五】记忆管理，让 Agent 记住对话

系列文章目录

【 LangChain v1.2 入门系列教程】【一】开篇入门 | 从零开始，跑通你的第一个 AI Agent
【 LangChain v1.2 入门系列教程】【二】消息类型与提示词工程
【 LangChain v1.2 入门系列教程】【三】工具（Tools）开发，让 Agent 连接外部世界
【 LangChain v1.2 入门系列教程】【四】结构化输出，让 Agent 返回可预测的结构
【 LangChain v1.2 入门系列教程】【五】记忆管理，让 Agent 记住对话
【 LangChain v1.2 入门系列教程】【六】流式输出， 让 Agent 告别“想好了再说”
【 LangChain v1.2 入门系列教程】【七】中间件——给 Agent 装上"外挂"

前言

在前几篇文章中，我们学会了如何构建基础的 Agent，给 Agent 绑定工具实现复杂能力。但你可能已经发现：每次与 Agent 对话，它都像是“失忆”了一样，完全不记得你上一句说了什么。这篇教程，我们就从 0 到 1，带你彻底搞懂 LangChain v1.2 的记忆管理，看完就能写出能记住对话、支持多用户隔离、可直接生产落地的 Agent。

一、langchain v1.2版本重构记忆机制

在 v1.2 版本中，LangChain 对记忆机制进行了重大革新，下放了 LangGraph框架 Checkpointer 的状态管理机制。将对话历史直接作为 Agent 状态（State）的一部分，通过检查点（Checkpoint）自动持久化。这彻底告别了旧版冗余的 Memory 类设计，更加统一和强大，使用上也更加简单方便。

简单理解：Agent 的状态（包括对话历史）现在是一个“自动保存的游戏进度”，随时可以存档和读档。

二、记忆类型分类：短期 vs 长期

在 LangChain 的体系中，记忆分为两大核心类型：

• 短期记忆：也叫会话级记忆，作用范围是 ** 单个对话线程（Thread）** 内，对应我们常说的单场对话的聊天记录。比如你和 AI 的一个聊天窗口，从开启到结束的所有交互，都属于短期记忆的管理范围。

• 长期记忆：跨会话、跨线程的持久化记忆，比如记住“用户张三喜欢素食”，无论开启多少次新对话，Agent 都知道，通常会结合向量数据库、外部存储实现。

三、Checkpoint 检查点机制

Checkpointer是记忆的 “存储硬盘”，负责把 Agent 的每一步运行状态（包括对话历史、自定义字段）持久化保存。它会在 Agent 每一次调用、每一步工具执行完成后，自动保存状态快照；在 Agent 每次执行前，自动读取最新状态并恢复记忆。
LangChain 提供了多种开箱即用的 Checkpointer 实现：测试环境用InMemorySaver（内存级）（对应短期记忆），生产环境用PostgresSaver（数据库级）（对应长期记忆）等。

四、快速上手：让 Agent 记住你的名字

接下来让我们用一个简单示例，演示下短期记忆的实现：

fromlanggraph.checkpoint.memoryimportInMemorySaver# 初始化内存级记忆引擎checkpointer=InMemorySaver()agent=create_agent(model=llm,checkpointer=checkpointer#绑定记忆引擎，开启记忆能力)#配置会话：指定thread_id，标识唯一对话config={"configurable":{"thread_id":"1"}}# 第一轮：告诉名字response1=agent.invoke({"messages":[HumanMessage('你好，我叫小明')]},config=config)print(f"第一轮回答：{response1["messages"][-1].content}")## 第二轮：询问名字response2=agent.invoke({"messages":[HumanMessage('我叫什么名字')]},config=config)print(f"第二轮回答：{response2["messages"][-1].content}")

运行这段代码，你会发现，第二轮对话中，AI 完美记住了你第一轮输入的名字，一个带记忆的 Agent 就这么简单实现了！

说明：

• InMemorySaver：将对话状态保存在 Python 内存中，进程重启后数据丢失（也称作短期记忆），适合开发和测试环境。
• thread_id：会话的唯一标识符，相同 thread_id 的调用会共享同一份对话历史。不同用户必须使用不同的 thread_id，否则会出现"串台"现象
• 必须给invoke方法传入带thread_id的 config，否则 Agent 无法找到对应的记忆
• 所有对话历史，都会自动存在 AgentState 的messages字段中，无需手动维护

五、thread_id：实现多用户会话隔离的核心

在上面示例代码中，config 里的 thread_id 是记忆管理的核心。它就像聊天软件的“会话窗口ID”。

• 不同 thread_id：会话完全隔离。用户 A 和用户 B 的对话互不干扰。
• 相同 thread_id：Agent 会加载该 ID 下的所有历史状态，实现连续对话。

我们用代码演示，两个不同用户使用不同的 thread_id，对话记忆完全互不影响：

# 初始化内存级记忆引擎checkpointer=InMemorySaver()agent=create_agent(model=llm,checkpointer=checkpointer#绑定记忆引擎，开启记忆能力)#小明的会话idconfig_xiaoming={"configurable":{"thread_id":"1"}}#小红的会话idconfig_xiaohone={"configurable":{"thread_id":"2"}}# 小明的对话print('—————— 以下是小明对话 ——————')agent.invoke({"messages":[HumanMessage('你好，我叫小明，我的工号是10001')]},config=config_xiaoming)response1=agent.invoke({"messages":[HumanMessage('我的工号是多少？')]},config=config_xiaoming)print(f"小明的AI回复：{response1["messages"][-1].content}")# 小明的对话print('—————— 以下是小红对话 ——————')agent.invoke({"messages":[HumanMessage('你好，我叫小红，我的工号是20002')]},config=config_xiaohone)response2=agent.invoke({"messages":[HumanMessage('我的工号是多少？')]},config=config_xiaohone)print(f"小红的AI回复：{response2["messages"][-1].content}")# 验证会话隔离：小红的会话里，无法获取小明的信息print("—————— 验证会话隔离 ——————")response3=agent.invoke({"messages":[HumanMessage('你知道小明的工号是多少吗？')]},config=config_xiaohone)print(f"验证会话隔离回复：{response3["messages"][-1].content}")

运行后你会发现，小明和小红的会话记忆完全独立，AI 在小红的会话中，完全无法获取小明的信息，完美实现了多用户的记忆隔离。

ps:在生产环境中建议使用「用户 ID + 会话 ID」作为 thread_id，确保每个用户的每个对话都有全局唯一标识，从根本上避免记忆串扰。

六、记忆的读取、更新与重置：完全掌控对话记忆

除了自动的记忆读写，我们还可以手动对记忆进行操作，满足各类业务场景的定制化需求。

1.手动读取记忆，查看完整历史对话

方法1：通过checkpointer

# 初始化内存级记忆引擎checkpointer=InMemorySaver()agent=create_agent(model=llm,checkpointer=checkpointer#绑定记忆引擎，开启记忆能力)#配置会话：指定thread_id，标识唯一对话config={"configurable":{"thread_id":"1"}}#第一轮对话agent.invoke({"messages":[HumanMessage('你好，我叫小明')]},config=config)#第二轮对话response=agent.invoke({"messages":[HumanMessage('我叫什么名字')]},config=config)checkpointer_tuple=checkpointer.get_tuple(config=config)if(checkpointer_tuple):#获取对话历史messages=checkpointer_tuple.checkpoint["channel_values"]["messages"]print(messages,'messages')

方法2：通过agent

# 初始化内存级记忆引擎checkpointer=InMemorySaver()agent=create_agent(model=llm,checkpointer=checkpointer)# 绑定记忆引擎，开启记忆能力# 配置会话：指定thread_id，标识唯一对话config={"configurable":{"thread_id":"1",},}agent.invoke({"messages":[HumanMessage("你好，我叫小明")]},config=config)agent.invoke({"messages":[HumanMessage("我叫什么名字")]},config=config)#读取当前状态state=agent.get_state(config)#获取对话历史messages=state.values['messages']print(messages,'messages')

2.手动更新记忆，修改或删除对话历史

插入消息：

checkpointer=InMemorySaver()agent=create_agent(model=llm,checkpointer=checkpointer)# 绑定记忆引擎，开启记忆能力# 配置会话：指定thread_id，标识唯一对话config={"configurable":{"thread_id":"1",},}agent.invoke({"messages":[HumanMessage("你好，我叫小明")]},config=config)agent.invoke({"messages":[HumanMessage("我叫什么名字")]},config=config)agent.update_state(config,{"messages":[AIMessage(content="[系统备注] 这是一条插入的系统消息")]}#末尾插入)

删除某条消息

fromlangchain.messagesimportRemoveMessage agent.update_state(config,{"messages":[RemoveMessage(id='019d5dfa5e1668c0b673312e2761a7d'),# 通过message id]})

清空所有消息：

fromlangchain.messagesimportRemoveMessagefromlanggraph.graph.messageimportREMOVE_ALL_MESSAGES agent.update_state(config,{"messages":[RemoveMessage(id=REMOVE_ALL_MESSAGES),# 清空所有]})

清空后追加消息：

agent.update_state(config,{"messages":[RemoveMessage(id=REMOVE_ALL_MESSAGES),# 清空所有HumanMessage("新内容")# 追加到末尾！]})

修改某条消息：

new_messages=[]#重建消息列表formsginmessages:if(msg.content=='你好，我叫小明'):#替换新消息new_msg=type(msg)(content="这是替换后的消息",id=msg.id)new_messages.append(new_msg)else:new_messages.append(msg)agent.update_state(config,{"messages":[RemoveMessage(id=REMOVE_ALL_MESSAGES),#清除所有*new_messages#更新]})

ps：update_state 的 "messages" 字段不是"赋值覆盖"，而是 "reducer 累加处理" —— 必须用 REMOVE_ALL_MESSAGES 清空后，再按目标顺序展开完整列表，才能实现"原地修改"效果

七、生产级方案：使用 PostgreSQL 持久化记忆

上面我们用的InMemorySaver是内存级存储，程序重启后记忆会全部丢失，仅适用于测试和本地开发。生产环境中，我们需要使用数据库，实现记忆的持久化存储。

LangChain 官方推荐的生产级方案是PostgresSaver，基于 PostgreSQL 数据库实现，支持高并发、数据持久化、分布式部署，是生产环境的首选

1.安装PostgreSQL （以window为例）

（1）官网下载

https://www.postgresql.org/

（2）安装

pgAdmin4(图形化管理工具)按需勾选，其他全选

一直“next“后安装

最后点击"finish"完成

（3）添加环境变量：

打开“系统属性“-“环境变量“，在系统变量“Path“”新增环境变量

安装目录\18\bin

（4）验证安装是否成功：

打开终端（Windows 用 CMD 或 PowerShell），输入：

psql--version

如图显示出版本号，证明安装成功

（5）登录数据库

在终端输入：

psql-Upostgres

输入密码，如下图所示连接成功：

（6）创建数据库

在 postgres=# 提示符下输入：

CREATE DATABASE mydb;

将 mydb 替换为你想要的数据库名称

查看数据库列表：

\l

发现mydb已在数据库列表中，数据库安装和创建工作完成。

2.安装依赖

pipenvinstall"psycopg[binary]"langgraph-checkpoint-postgres

3.代码实现

from langgraph.checkpoint.postgresimportPostgresSaver llm=ChatOpenAI(model="Qwen/Qwen3-8B",base_url="https://api.siliconflow.cn/v1",api_key=API_KEY,)DB_URI="postgresql://用户名:密码@数据库地址:端口/数据库名"#例如postgresql://postgres:123456@localhost:5432/mydbwith PostgresSaver.from_conn_string(DB_URI)as checkpointer:# 首次运行时自动创建必要的表checkpointer.setup()config={"configurable":{"thread_id":"1"}}# 创建agentagent=create_agent(model=llm,system_prompt="你是一个智能助手，请根据用户提问作答",checkpointer=checkpointer,)agent.invoke({"messages":[HumanMessage("你好，我叫小明")]},config=config)agent.invoke({"messages":[HumanMessage("我叫什么名字")]},config=config)

运行这段代码后，所有的对话记忆都会持久化到 PostgreSQL 数据库中。哪怕你重启程序、关闭服务器，只要重新连接同一个数据库，使用同一个thread_id，就能完全恢复之前的对话记忆，实现跨会话、跨进程的记忆续传。

ps:对于生产环境，建议在数据库层面设置数据保留策略，定期清理旧会话数据，既保护隐私又节省存储成本

总结

LangChain v1.2 将记忆管理统一到 LangGraph 的 Checkpointer 机制中，对话历史成为 Agent 状态的一部分，彻底告别了旧版繁琐的 Memory 类设计。通过 InMemorySaver 可快速实现短期记忆（会话级），而 thread_id 则是多用户隔离的核心钥匙——不同 ID 实现完全独立的对话空间。更进一步，开发者可以手动读取、更新甚至重置记忆，满足各种定制化需求。生产环境中，切换到 PostgresSaver 即可获得持久化、高可用的长期记忆能力。掌握这套记忆体系，你就能轻松构建出真正“有记性”、可落地的智能 Agent。下一篇，我们将继续探讨“流式输出”功能。