在 LangChain 的发展史上,Memory(记忆)组件经历过一次巨大的重构
如果你看网上的老教程,大多会教你用 ConversationBufferMemoryConversationChain 等带有 Memory 字眼的类。请立刻把它们忘掉! 在最新的 LangChain(v0.2/v0.3 及以后)中,这些老式的黑盒组件正在被淘汰。
在这里插入图片描述

现代 LangChain 的记忆哲学是:
大模型本身是无状态的(像鱼只有7秒记忆),所谓的“记忆”,本质上就是每次提问时,把以前的聊天记录(Message List)重新拼接在 Prompt 里传给大模型。

因此,最新的框架将记忆拆解成了更透明、更贴合 LCEL 管道的三个核心部件:

  1. MessagesPlaceholder (占位符):在 Prompt 中留个空位。
  2. BaseChatMessageHistory (历史记录库):专门负责存取聊天记录的“数据库”。
  3. 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 在后台偷偷做了什么?

  1. 拦截到调用,提取出 session_id="zhangsan_123"
  2. 调用 get_session_history("zhangsan_123"),从字典里拿到了张三的历史 Message 列表。
  3. 把这个列表塞进输入字典 {"question": "...", "chat_history": [历史列表]}
  4. 把组装好的大字典扔进你的 base_chain 里执行。
  5. 拿到大模型的回复后,自动把 HumanMessageAIMessage 追加保存进张三的历史记录本里。
  6. 最后把结果返回给你。

可见新技术下,执行过程和书写代码时非常清晰!一气呵成!


第三部分:高阶实战 —— 记忆爆满怎么办?(修剪记忆 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 函数里的返回值,就能实现各种数据库的接入,而无需修改任何链的代码

  • Redisfrom langchain_redis import RedisChatMessageHistory
  • PostgreSQLfrom langchain_postgres import PostgresChatMessageHistory
  • MongoDBfrom 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 就是一辆“模块化的智能电动车”。

现代记忆框架的三大黄金法则:

  1. 记忆就是消息列表:在 Prompt 中用 MessagesPlaceholder 留出位置。
  2. 状态与逻辑解耦:数据库(Redis/内存)只管存取,RunnableWithMessageHistory 只管拦截和注入,你的业务链(Prompt+Model)无需关心记忆是怎么来的。
  3. 认准 session_id:在 invokeconfig 参数中传递 session_id,这是区分多租户/多用户对话的唯一标准。

掌握了这一节,你就真正做到了“框架级”的大模型应用开发。接下来,不管用户怎么闲聊,你的机器人都能对答如流且“过目不忘”了!

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐