LangChain之Memory(记忆)核心组件.2026年新版讲解,超详细
摘要: LangChain 的记忆组件经历重大重构,旧版 ConversationBufferMemory 等类已被淘汰。新版采用透明化设计,将记忆拆解为三个核心部件: MessagesPlaceholder:在 Prompt 中预留历史记录插槽 BaseChatMessageHistory:存储聊天记录的“数据库” RunnableWithMessageHistory:自动管理历史记录的包装器
在 LangChain 的发展史上,Memory(记忆)组件经历过一次巨大的重构。
如果你看网上的老教程,大多会教你用 ConversationBufferMemory、ConversationChain 等带有 Memory 字眼的类。请立刻把它们忘掉! 在最新的 LangChain(v0.2/v0.3 及以后)中,这些老式的黑盒组件正在被淘汰。
现代 LangChain 的记忆哲学是:
大模型本身是无状态的(像鱼只有7秒记忆),所谓的“记忆”,本质上就是每次提问时,把以前的聊天记录(Message List)重新拼接在 Prompt 里传给大模型。
因此,最新的框架将记忆拆解成了更透明、更贴合 LCEL 管道的三个核心部件:
MessagesPlaceholder(占位符):在 Prompt 中留个空位。BaseChatMessageHistory(历史记录库):专门负责存取聊天记录的“数据库”。RunnableWithMessageHistory(历史拦截器):一个极其优雅的包装器,自动帮你把记录塞进 Prompt,并把新回复存进数据库。
接下来,我将带你从零组装现代版的“带记忆的链”。
第一部分:核心组装 —— 打造现代记忆链(三步走)
我们直接用代码说话,看看现代记忆系统是如何运转的。
第一步:在 Prompt 中留出“记忆插槽”
你要告诉模型,历史记录应该插在哪个位置。通常插在 System 指令之后,当前用户输入之前。
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# 1. 定义带有“历史记录插槽”的 Prompt
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个幽默的 AI 助手。"),
# 【核心1】: 这是一个占位符,名字叫 "chat_history" (你可以随便起)
# 它会在运行时,自动被替换成一长串的对话记录列表。
MessagesPlaceholder(variable_name="chat_history"),
("human", "{question}"),
])
第二步:准备“历史记录数据库”
在实际生产中,你的记录可能存在 Redis 或 MySQL 里。在测试时,我们用 LangChain 提供的 InMemoryChatMessageHistory(内存存储),并用一个字典来管理不同用户的会话。
from langchain_core.chat_history import InMemoryChatMessageHistory
# 用一个全局字典来模拟数据库,存放不同用户的对话记录
# 格式: {"session_id_1": InMemoryChatMessageHistory, "session_id_2": ...}
store = {}
# 这是一个获取历史记录的函数,它必须接收 session_id 作为参数
def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
# 如果这个用户(session_id)还没聊过天,就给他建一个空的历史记录本
if session_id not in store:
store[session_id] = InMemoryChatMessageHistory()
return store[session_id]
第三步:使用 RunnableWithMessageHistory 包装你的链
这是现代 LangChain 记忆系统的灵魂。它就像一个拦截器,包裹在你的 LCEL 链外面。
from langchain_openai import ChatOpenAI
from langchain_core.runnables.history import RunnableWithMessageHistory
# 基础链 (LCEL)
model = ChatOpenAI(model="gpt-3.5-turbo")
base_chain = prompt | model
# 【核心2】: 给基础链套上“记忆外壳”
chain_with_history = RunnableWithMessageHistory(
runnable=base_chain, # 你的原始链
get_session_history=get_session_history, # 你的“数据库查询函数”
input_messages_key="question", # 告诉它,用户的新问题是哪个变量?
history_messages_key="chat_history", # 告诉它,历史记录要塞给 Prompt 里的哪个变量?
)
第二部分:见证奇迹 —— 多用户会话管理 (session_id)
刚才我们已经把带有记忆的链组装好了。现在我们来调用它!
注意现代记忆链的调用方式:
你不能只传 question 了,你必须在 config(配置项)里传入 session_id。这就完美解决了张三和李四同时和你聊天时,记忆串线的问题。
完整演示代码:
# --- 模拟张三来聊天 (session_id = "zhangsan_123") ---
print("=== 张三的对话 ===")
config_zhangsan = {"configurable": {"session_id": "zhangsan_123"}}
# 第一轮
res1 = chain_with_history.invoke(
{"question": "你好,我叫张三,我最喜欢吃苹果。"},
config=config_zhangsan
)
print(f"AI: {res1.content}")
# 第二轮 (测试记忆)
res2 = chain_with_history.invoke(
{"question": "你还记得我叫什么吗?我最喜欢吃什么?"},
config=config_zhangsan
)
print(f"AI: {res2.content}")
# 预期输出: AI: 记得呀,你叫张三,最喜欢吃苹果!
# --- 模拟李四来聊天 (session_id = "lisi_456") ---
print("\n=== 李四的对话 ===")
config_lisi = {"configurable": {"session_id": "lisi_456"}}
res3 = chain_with_history.invoke(
{"question": "我叫李四,你好。刚才那个叫张三的人喜欢吃什么?"},
config=config_lisi
)
print(f"AI: {res3.content}")
# 预期输出: AI: 你好李四!我不知道张三喜欢吃什么,这是我们第一次聊天哦。(因为记忆被 session_id 严格隔离了)
原理解析:当 invoke 发生时,RunnableWithMessageHistory 在后台偷偷做了什么?
- 拦截到调用,提取出
session_id="zhangsan_123"。 - 调用
get_session_history("zhangsan_123"),从字典里拿到了张三的历史 Message 列表。 - 把这个列表塞进输入字典
{"question": "...", "chat_history": [历史列表]}。 - 把组装好的大字典扔进你的
base_chain里执行。 - 拿到大模型的回复后,自动把
HumanMessage和AIMessage追加保存进张三的历史记录本里。 - 最后把结果返回给你。
可见新技术下,执行过程和书写代码时非常清晰!一气呵成!
第三部分:高阶实战 —— 记忆爆满怎么办?(修剪记忆 trim_messages)
只要你不断聊天,历史记录就会越来越长,最后不仅会消耗巨量 Token(极贵),还会超出模型的最大上下文窗口(报错)。
在老版本中,有 ConversationTokenBufferMemory 这种东西。在现代版本中,LangChain 提供了一个极其灵活的通用函数:trim_messages(修剪消息)。
它可以放在你的 LCEL 链条中,像个“滤网”一样,在传给大模型之前,把太老的消息丢掉,只保留最近的 N 条,或者最近的 N 个 Token。
代码实战:在链中加入“记忆修剪器”
from langchain_core.messages import trim_messages
from langchain_core.runnables import RunnablePassthrough
# 1. 定义一个“修剪器”
# 规则:保留最后 65 个 token(或者可以按条数保留,比如 max_tokens=10 且 token_counter是计算条数)
trimmer = trim_messages(
max_tokens=65, # 最大限制
strategy="last", # 策略:保留最新的(丢弃最老的)
token_counter=model, # 按照模型的分词器来精确计算 token (非常智能!)
include_system=True, # 绝对不能把 System 设定给剪了!
allow_partial=False, # 不允许截断半句话
)
# 2. 改造基础链,把 trimmer 加进去
# 现在的流程: Prompt 生成消息列表 -> Trimmer 修剪消息列表 -> Model
base_chain_with_trimmer = (
RunnablePassthrough.assign(
# 在传入 Prompt 之前,如果你想先对拿到的 chat_history 进行修剪,也可以在这里做
# 但更优雅的方式是修剪 Prompt 生成出来的完整 Messages 列表
)
)
# 更加优雅的 LCEL 组装法:
base_chain_trimmed = prompt | trimmer | model
# 3. 再次套上记忆外壳
chain_with_history_and_trim = RunnableWithMessageHistory(
runnable=base_chain_trimmed,
get_session_history=get_session_history,
input_messages_key="question",
history_messages_key="chat_history",
)
# 现在,这个链不论你聊多长,它永远只会把最近的几十个 token 发给大模型!永远不会爆掉!
第四部分:拓展 —— 如果想长期记住(持久化数据库)
我们刚才用的 store = {} 和 InMemoryChatMessageHistory 是存在内存里的,程序一重启就全丢了。
在现代 LangChain 中,你只需要换掉 get_session_history 函数里的返回值,就能实现各种数据库的接入,而无需修改任何链的代码!
- Redis:
from langchain_redis import RedisChatMessageHistory - PostgreSQL:
from langchain_postgres import PostgresChatMessageHistory - MongoDB:
from langchain_mongodb import MongoDBChatMessageHistory
例如换成 Redis 记忆:
# 假设你想用 Redis 存记忆,只需要改这一个函数:
from langchain_redis import RedisChatMessageHistory
def get_redis_history(session_id: str):
return RedisChatMessageHistory(
session_id=session_id,
redis_url="redis://localhost:6379/0"
)
# 传给 wrapper 即可
chain_with_redis = RunnableWithMessageHistory(
base_chain,
get_session_history=get_redis_history, # 换成了 Redis 版本的函数!
# ... 省略其他参数
)
这就是解耦带来的巨大好处!
总结与思维升华
如果你把老版本的 Memory 比作一辆“不可拆卸、焊死了的燃油车”,那么最新的 RunnableWithMessageHistory 就是一辆“模块化的智能电动车”。
现代记忆框架的三大黄金法则:
- 记忆就是消息列表:在 Prompt 中用
MessagesPlaceholder留出位置。 - 状态与逻辑解耦:数据库(Redis/内存)只管存取,
RunnableWithMessageHistory只管拦截和注入,你的业务链(Prompt+Model)无需关心记忆是怎么来的。 - 认准
session_id:在invoke的config参数中传递session_id,这是区分多租户/多用户对话的唯一标准。
掌握了这一节,你就真正做到了“框架级”的大模型应用开发。接下来,不管用户怎么闲聊,你的机器人都能对答如流且“过目不忘”了!
更多推荐


所有评论(0)