我要投稿

LangChain---Memory：为AI应用赋予“记忆”

发布日期：2025-07-25 14:14:38 浏览次数： 1554

作者：AI云枢

微信搜一搜，关注“AI云枢”

【本期目标】

理解为什么LLM需要记忆，以及记忆在多轮对话中的作用。
掌握 ChatMessageHistory 作为底层消息存储机制。
学习 LangChain 中几种常用记忆类型（ConversationBufferMemory, ConversationBufferWindowMemory, ConversationSummaryMemory, ConversationTokenBufferMemory）的工作原理和适用场景。
重点掌握如何使用 RunnableWithMessageHistory 将记忆功能无缝集成到任何LCEL链条中。
通过实际案例，构建一个具有上下文理解能力的聊天机器人。

引言：为什么AI会“失忆”？

想象一下你和一个刚认识的朋友聊天，你问了几个问题，他每次都只能回答你当前的问题，完全不记得你之前说过什么，这会让你感觉非常糟糕。

LLM（大语言模型）天生就是“失忆”的。每次你向LLM发送一个请求（invoke），它都是一个独立的、无状态的请求。LLM不会自动记住你上一次的对话内容。这就是所谓的**“无状态性”**。

在构建聊天机器人、智能客服、或者任何需要“上下文”的应用时，这种无状态性是一个巨大的挑战。用户希望机器人能记住他们之前说过的话，从而进行连贯的对话。

LangChain 的 Memory模块就是用来解决这个问题的。它负责：

存储和管理对话历史。
在每次LLM调用前，将相关的对话历史注入到提示（Prompt）中。
根据需要（如历史过长），对对话历史进行摘要或裁剪。

核心概念：ChatMessageHistory——对话的基石

在 LangChain 中，对话历史被表示为一系列消息（Messages）。ChatMessageHistory 是所有记忆类型的基础，它提供了一个简单的接口来存储和获取消息。

消息类型包括：

HumanMessage：用户发送的消息。
AIMessage：AI模型（LLM）生成的消息。
SystemMessage：系统级别的指令或角色设定。
FunctionMessage / ToolMessage：工具调用相关的消息（后续Agent章节会涉及）。

from langchain_core.messages import HumanMessage, AIMessagefrom langchain_core.chat_history import BaseChatMessageHistory# 简单的消息历史存储class InMemoryChatMessageHistory(BaseChatMessageHistory):    def __init__(self):        self.messages = []    def add_message(self, message):        self.messages.append(message)    def clear(self):        self.messages = []# 实例化并添加消息history = InMemoryChatMessageHistory()history.add_message(HumanMessage(content="你好！"))history.add_message(AIMessage(content="你好！有什么可以帮助你的吗？"))history.add_message(HumanMessage(content="你是谁？"))print("--- ChatMessageHistory 示例 ---")for msg in history.messages:    print(f"{msg.type}: {msg.content}")print("\n")

小结：ChatMessageHistory 是存储和管理对话消息的底层接口。

第一部分：常见的记忆类型及其工作原理

LangChain 提供了多种开箱即用的记忆类型，每种都有其独特的优势和适用场景。它们都继承自 BaseMemory。

要点：这些记忆类型在内部都维护一个 ChatMessageHistory 对象，并负责将其内容格式化后注入到LLM的提示中。

ConversationBufferMemory：缓冲记忆

原理：最简单的记忆类型，它会完整地保存所有对话历史。

优点：简单直接，保留所有上下文。

缺点：对话轮次增多时，消耗的Token会越来越多，可能导致LLM上下文溢出或费用过高。

适用场景：对话历史不长，或需要完整回顾所有对话的场景。

from langchain.memory import ConversationBufferMemorybuffer_memory = ConversationBufferMemory()buffer_memory.save_context({"input": "我的名字是小明。"}, {"output": "好的，小明。很高兴认识你！"})buffer_memory.save_context({"input": "你知道我叫什么吗？"}, {"output": "我知道，你叫小明。"})# load_memory_variables 会返回一个字典，其中包含对话历史# 默认键是 "history"print("--- ConversationBufferMemory 示例 ---")print(buffer_memory.load_memory_variables({}))# 你会看到 history 字段包含了所有对话，格式为字符串# output: {'history': 'Human: 我的名字是小明。\nAI: 好的，小明。很高兴认识你！\nHuman: 你知道我叫什么吗？\nAI: 我知道，你叫小明。'}print("\n")

ConversationBufferWindowMemory：窗口记忆

原理：只保留最近 k 轮的对话历史。当对话轮次超过 k 时，最旧的对话会被丢弃。
优点：有效控制Token使用，避免上下文溢出。
缺点：可能会丢失早期但重要的上下文。
适用场景：多数聊天场景，LLM只需要近期上下文就能进行连贯对话。

from langchain.memory import ConversationBufferWindowMemory# 只保留最近1轮对话 (即只保留当前轮和上一轮)window_memory = ConversationBufferWindowMemory(k=1)window_memory.save_context({"input": "第一句话。"}, {"output": "这是第一句话的回复。"})window_memory.save_context({"input": "第二句话。"}, {"output": "这是第二句话的回复。"})window_memory.save_context({"input": "第三句话。"}, {"output": "这是第三句话的回复。"})print("--- ConversationBufferWindowMemory (k=1) 示例 ---")print(window_memory.load_memory_variables({}))# output: {'history': 'Human: 第三句话。\nAI: 这是第三句话的回复。'}# 注意，只有最后一句对话被保留了print("\n")

ConversationSummaryMemory：摘要记忆

原理：使用LLM来定期总结旧的对话历史，然后只将摘要和最新的几轮对话提供给LLM。
优点：在保留长对话主要上下文的同时，有效控制Token使用。
缺点：需要额外的LLM调用来生成摘要，会增加延迟和成本；摘要可能会丢失细节。
适用场景：长对话、复杂对话，需要LLM理解长期上下文但又不想传输完整历史。

from langchain.memory import ConversationSummaryMemoryfrom langchain_openai import ChatOpenAIfrom dotenv import load_dotenv; load_dotenv() # 确保加载了OPENAI_API_KEYllm_for_summary = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # 用于摘要的LLMsummary_memory = ConversationSummaryMemory(llm=llm_for_summary)summary_memory.save_context({"input": "我最喜欢的颜色是蓝色。"}, {"output": "蓝色很棒！"})summary_memory.save_context({"input": "你知道天空为什么是蓝色的吗？"}, {"output": "天空看起来是蓝色是因为瑞利散射。"})summary_memory.save_context({"input": "那大海呢？"}, {"output": "大海的蓝色则是由水的固有性质以及对光线的吸收和散射造成的。"})print("--- ConversationSummaryMemory 示例 ---")print(summary_memory.load_memory_variables({}))# 你会看到 history 字段中包含了LLM生成的摘要和最新的对话# 摘要可能类似："用户提到他最喜欢的颜色是蓝色，询问天空和大海为何呈现蓝色，AI解释了瑞利散射和水的吸收散射。"print("\n")

ConversationTokenBufferMemory：Token缓冲记忆

原理：类似于窗口记忆，但它不是按轮次，而是根据对话历史的**Token数量**来控制。当Token总数超过设定阈值时，最旧的对话会被移除，直到Token数达标。
优点：更精确地控制Token使用，直接面向LLM的上下文限制。
缺点：同样可能丢失旧的细节。
适用场景：对Token使用非常敏感的场景，或者对话轮次长度不固定。

from langchain.memory import ConversationTokenBufferMemoryfrom langchain_openai import ChatOpenAIfrom langchain.callbacks import get_openai_callback # 用于查看token使用# 1. 创建自定义的 Chat Model 类，本地是vllm部署的模型，计算token方面需要重写方法class VllmChatModel(ChatOpenAI):
    def __init__(self, *args: Any, **kwargs: Any) -> None:        super().__init__(*args, **kwargs)        self._tokenizer = tiktoken.get_encoding("cl100k_base")    def get_num_tokens_from_messages(self, messages: List[BaseMessage]) -> int:        """根据消息列表计算 token 数量。"""        text = ""        for message in messages:            text += message.content        return len(self._tokenizer.encode(text))
    @property    def _llm_type(self) -> str:        return "vllm-chat-model"
llm_token = VllmChatModel(    model=os.environ.get("OPENAI_MODEL"),    temperature=0.9,    base_url=os.environ.get("OPENAI_BASE_URL"),    openai_api_key=os.environ.get("OPENAI_API_KEY"),)# 设置最大Token数为 50 （实际会略有浮动，因为是估算）token_memory = ConversationTokenBufferMemory(llm=llm_token , max_token_limit=50)token_memory.save_context({"input": "第一句非常长的对话内容，包含了多个词语，这样才能体现Token的限制作用。"}, {"output": "第一句回复。"})token_memory.save_context({"input": "第二句相对较短的对话内容。"}, {"output": "第二句回复。"})token_memory.save_context({"input": "第三句简短的话。"}, {"output": "第三句回复。"})print("--- ConversationTokenBufferMemory 示例 ---")print(token_memory.load_memory_variables({}))# 你会发现，可能只有最后几句话被保留了，因为前几句话的Token数已经超出了50的限制# print(cb) # 如果是LLM调用，可以看到Token数print("\n")

小结：选择合适的记忆类型是平衡上下文长度、信息保留和成本的关键。以上方法更适合入门学习，因为大部分在新版中已经迁移。

第二部分：LCEL整合：RunnableWithMessageHistory——现代记忆管理

在 LangChain 的最新版本中，将记忆功能与LCEL链条集成，最推荐且最方便的方式是使用 RunnableWithMessageHistory。

RunnableWithMessageHistory 是一个 Runnable 包装器。它能够将任何无状态的LCEL链条转化为有状态的链条，自动处理对话历史的加载、保存和注入。

核心思想：**get_session_history** 回调函数

RunnableWithMessageHistory 需要一个 get_session_history 函数。
这个函数接收一个 session_id（通常是每个用户或每个对话的唯一标识符）。
它应该返回一个 BaseChatMessageHistory 实例（例如我们之前演示的 InMemoryChatMessageHistory 或 ConversationBufferMemory 内部的 history 对象）。
意味着可以灵活地存储对话历史，无论是内存、Redis、数据库还是其他持久化存储。

【实践：构建一个具有记忆的LCEL聊天机器人】

使用 ConversationBufferWindowMemory 为聊天机器人赋予记忆。

from dotenv import load_dotenvimport osfrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.runnables.history import RunnableWithMessageHistoryfrom langchain_core.messages import HumanMessage, AIMessagefrom langchain.memory import ConversationBufferWindowMemoryfrom langchain_core.chat_history import BaseChatMessageHistoryload_dotenv()llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)# 1. 定义一个用于存储会话历史的字典 (模拟持久化存储，实际应用中会是数据库/Redis)# key 是 session_id, value 是 BaseChatMessageHistory 实例store = {}# 2. 实现 get_session_history 函数# 这个函数会根据 session_id 返回对应的 ChatMessageHistory 对象def get_session_history(session_id: str) -> BaseChatMessageHistory:    if session_id not in store:        # 这里我们使用 ConversationBufferWindowMemory，只保留最近3轮对话        # 你也可以根据需要替换为 ConversationBufferMemory 或其他记忆类型        store[session_id] = ConversationBufferWindowMemory(            k=3, # 保持3轮记忆            return_messages=True, # 以消息对象列表形式返回，适合MessagesPlaceholder            output_key="output", input_key="input" # 定义输入输出的key        ).chat_memory # 获取底层的 ChatMessageHistory 对象    return store[session_id]# 3. 定义一个普通的 LCEL 链条 (不带记忆)# 注意 PromptTemplate 中使用 MessagesPlaceholder 来占位历史消息prompt = ChatPromptTemplate.from_messages([    ("system", "你是一个友好的AI助手，请根据上下文进行对话。"),    MessagesPlaceholder(variable_name="history"), # 这是记忆会填充的地方    ("user", "{input}") # 用户当前输入])# 核心业务逻辑链 (这是我们希望包装成有记忆的链条)# input -> prompt (加入history和input) -> LLM -> output parserbase_conversation_chain = prompt | llm | StrOutputParser()# 4. 使用 RunnableWithMessageHistory 包装你的 LCEL 链条with_message_history_chain = RunnableWithMessageHistory(    base_conversation_chain, # 你要包装的 LCEL 链条    get_session_history, # 会话历史获取函数    input_messages_key="input", # 链条的输入中，哪个键是用户消息    history_messages_key="history", # 提示模板中，哪个键是历史消息的占位符    output_messages_key="output" # 定义输出的key，以便save_context能够工作)# 5. 进行多轮对话 (需要传入 config 参数，其中包含 session_id)print("--- 具有记忆的聊天机器人示例 ---")session_id_user1 = "user_123"# 第一轮对话response1 = with_message_history_chain.invoke(    {"input": "我的名字是张三。"},    config={"configurable": {"session_id": session_id_user1}})print(f"用户1 (轮1): 我的名字是张三。")print(f"AI (轮1): {response1}")# 第二轮对话 (同一个 session_id)response2 = with_message_history_chain.invoke(    {"input": "你知道我叫什么吗？"},    config={"configurable": {"session_id": session_id_user1}})print(f"用户1 (轮2): 你知道我叫什么吗？")print(f"AI (轮2): {response2}") # AI应该能回答出“张三”# 第三轮对话 (同一个 session_id)response3 = with_message_history_chain.invoke(    {"input": "我喜欢吃苹果，不喜欢吃香蕉。"},    config={"configurable": {"session_id": session_id_user1}})print(f"用户1 (轮3): 我喜欢吃苹果，不喜欢吃香蕉。")print(f"AI (轮3): {response3}")# 第四轮对话 (同一个 session_id)response4 = with_message_history_chain.invoke(    {"input": "我刚才说了我喜欢什么水果？"},    config={"configurable": {"session_id": session_id_user1}})print(f"用户1 (轮4): 我刚才说了我喜欢什么水果？")print(f"AI (轮4): {response4}") # AI应该能回答出“苹果”# 切换到另一个用户 (不同的 session_id)session_id_user2 = "user_456"response_user2_1 = with_message_history_chain.invoke(    {"input": "你好，我是李四。"},    config={"configurable": {"session_id": session_id_user2}})print(f"\n用户2 (轮1): 你好，我是李四。")print(f"AI (轮1): {response_user2_1}")response_user2_2 = with_message_history_chain.invoke(    {"input": "我喜欢什么水果？"}, # 故意问这个，看AI是否会混淆    config={"configurable": {"session_id": session_id_user2}})print(f"用户2 (轮2): 我喜欢什么水果？")print(f"AI (轮2): {response_user2_2}") # AI应该不知道，因为这是新会话

代码解析：

store** 字典：** 模拟了一个会话历史的外部存储。在实际生产环境中，这会是 Redis、数据库（如MongoDB, PostgreSQL with pgvector）、或者其他会话管理服务。
get_session_history(session_id)**：** 这是 RunnableWithMessageHistory 的关键。每次调用都会根据 session_id 获取或创建一个 ChatMessageHistory 对象。我们在这里选择了 ConversationBufferWindowMemory 来保持最近3轮的记忆。
MessagesPlaceholder(variable_name="history")**：** 在 ChatPromptTemplate 中，这个占位符是专门用来插入对话历史的。RunnableWithMessageHistory 会自动将 ChatMessageHistory 中的消息格式化后填充到这个位置。
RunnableWithMessageHistory 的参数：

base_conversation_chain：要赋予记忆的无状态 LCEL 链。
get_session_history：上面定义的获取历史的函数。
input_messages_key：指定 base_conversation_chain 的输入字典中，哪个键代表用户的当前消息。
history_messages_key：指定 prompt 中，哪个占位符用来接收历史消息。
output_messages_key：指定 base_conversation_chain 的输出中，哪个键是AI的回复，以便记忆可以保存。

小结：RunnableWithMessageHistory 是 LangChain 在 LCEL 时代处理多轮对话的推荐方式。它提供了一个干净的接口，将记忆逻辑与核心业务逻辑分离，让你能够专注于链条本身的功能。

第三部分：记忆与RAG链的结合

RunnableWithMessageHistory 的强大之处在于，它不仅仅能包装简单的聊天链，**它可以包装任何 LCEL 链条**，包括我们之前构建的 RAG 链！

这意味着，你可以将一个**无状态的RAG链**（输入用户问题，输出答案）包装起来，使其具备**多轮对话记忆**的能力。例如，用户可以先问“LangChain是什么？”，然后接着问“它的调试工具叫什么？”，RAG系统就能根据上下文理解第二个问题中的“它”指的是LangChain，并返回准确的答案。

如何在RAG链中引入记忆：

历史感知检索：在检索前，LLM会先查看对话历史，根据历史和当前问题，重新生成或优化检索查询。例如，用户问“它的调试工具叫什么？”LLM会知道“它”指的是上文中提到的“LangChain”。
上下文感知生成：在答案生成阶段，LLM不仅会收到检索到的文档，还会收到对话历史，以便生成更连贯和个性化的回答。

我们将在后续的RAG高级章节中，更详细地讲解如何构建一个完整的、具备记忆功能的RAG聊天机器人。

本期小结

在本期教程中，已经掌握了为AI应用赋予“记忆”的关键技能：

理解了LLM无状态性的挑战，以及记忆的重要性。
学习了 ChatMessageHistory 作为消息存储的基础。
深入了解了 ConversationBufferMemory、ConversationBufferWindowMemory、ConversationSummaryMemory 和 ConversationTokenBufferMemory 这四种记忆类型的特点和适用场景。
最重要的是，学会使用 RunnableWithMessageHistory ，轻松地为任何LCEL链条添加多轮对话能力。

现在，AI应用将不再“失忆”，可以与用户进行连贯、有上下文的对话。在下一期教程中，我们将迈向 LangChain 的另一大核心功能——**Agents（智能代理）**。我们将学习如何让LLM不再仅仅是回答问题，而是能够自主思考、规划，并调用外部工具来完成更复杂的任务！敬请期待！

53AI，企业落地大模型首选服务商

产品：场景落地咨询+大模型应用平台+行业解决方案

承诺：免费场景POC验证，效果验证后签署服务协议。零风险落地应用大模型，已交付160+中大型企业