LangChain框架中的记忆

《AI Agent智能体开发实践+玩转FastGPT 像搭积木一样构建智能体 LLM大语言模型AI Agent开发 智能体性能优化调试部署实施方法书籍 AIAgent智能体开发实践 无规格》【摘要 书评 试读】- 京东图书

本节将详细介绍LangChain框架中的记忆（Memory）系统，包括概念、接口实现和实际应用。

8.4.1 记忆的概念

在LangChain框架中，记忆是一种机制，用于存储和检索对话历史或其他上下文信息，使得模型能够“记住”之前的交互内容。这对于构建有状态的聊天机器人尤为重要，因为它允许对话具有连贯性和上下文感知能力。LangChain提供了多种记忆类型，包括：

1. ConversationBufferMemory：存储完整对话历史。
2. ConversationBufferWindowMemory：存储最近N条消息。
3. ConversationSummaryMemory：存储对话摘要。
4. EntityMemory：基于实体提取的记忆。
5. VectorStoreRetrieverMemory：使用向量数据库存储和检索记忆。

记忆组件通常与ChatOpenAI等聊天模型结合使用，通过memory_key参数将对话历史注入模型调用中。

8.4.2 BaseChatMessageHistory接口及其子类

BaseChatMessageHistory是LangChain中用于管理消息历史的抽象基类，定义了消息存储和检索的标准接口。其主要方法包括：

1. add_user_message(message:str)：添加用户消息。
2. add_ai_message(message:str)：添加AI回复。
3. clear()：清空消息历史。
4. messages：获取所有消息（属性）。

其常见子类说明如下。

1）内存存储实现

1. ChatMessageHistory：内存中的存储消息（临时存储）。

2）持久化存储实现

1. FileChatMessageHistory：基于文件存储（如JSON、CSV）。
2. RedisChatMessageHistory：使用Redis存储。
3. MongoDBChatMessageHistory：使用MongoDB存储。
4. SQLChatMessageHistory：使用SQL数据库（如SQLite、PostgreSQL）。
5. CassandraChatMessageHistory：使用Cassandra存储。

这些实现允许消息历史在会话间持久化，或在分布式系统中共享。

8.4.3 RunnableWithMessageHistory

RunnableWithMessageHistory是LangChain中用于构建带记忆功能的可运行组件的工具类。它允许将消息历史与链或模型结合，自动管理对话上下文的注入和更新。其核心功能包括：

1. 自动从用户输入和模型输出中提取消息。
2. 将历史消息格式化为模型可接受的输入格式。
3. 支持多种记忆类型和消息历史存储方式。

【示例8.10】RunnableWithMessageHistory（LangChain+Qwen）。

#登录阿里云百炼平台，创建 API-KEY 并替换代码中的 "你的API密钥" #请确保你安装了最新版的 LangChain 相关包 #pip install -U langchain langchain-community langchain-core dashscope from langchain_community.chat_models import ChatTongyi from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.runnables import RunnablePassthrough, RunnableWithMessageHistory from langchain_community.chat_message_histories import ChatMessageHistory import os # 设置你的 DashScope API Key（通义千问） os.environ["DASHSCOPE_API_KEY"] = "你的API密钥" # 替换为你的实际密钥 # 创建聊天模型（使用 Qwen） chat = ChatTongyi(model="qwen-max") # 可选: qwen-turbo, qwen-plus, qwen-max # 创建提示模板，包含历史消息占位符 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个有用的助手。"), MessagesPlaceholder(variable_name="history"), ("human", "{input}") ]) # 创建链 chain = prompt | chat # 包装为带有历史记录的链 chain_with_history = RunnableWithMessageHistory( chain, lambda session_id: ChatMessageHistory(), # 历史记录工厂函数 input_messages_key="input", history_messages_key="history" ) # 使用链 response = chain_with_history.invoke( {"input": "你好！"}, config={"configurable": {"session_id": "user123"}} ) print(response.content)

输出：

你好！有什么我能帮助你的吗？

8.4.4 基于LangChain的聊天机器人

使用LangChain构建聊天机器人时，记忆组件是关键一环。以下是一个完整示例，展示如何结合RunnableWithMessageHistory和ConversationBufferMemory构建一个简单的聊天机器人。

【示例8.11】基于LangChain的聊天机器人（适配LangChain最新版+Qwen）。

#依赖安装（确保版本兼容） #pip install --upgrade langchain langchain-community langchain-core dashscope #登录阿里云DashScope控制台，创建API Key并替换代码中的"your_dashscope_api_key" from langchain_community.chat_models import ChatTongyi from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_community.chat_message_histories import ChatMessageHistory from langchain_core.runnables import RunnableWithMessageHistory # ✅ 修正导入路径 import os # 设置 DashScope API Key（请替换为你自己的） api_key=os.getenv("DASHSCOPE_API_KEY") # 1. 创建聊天模型（使用 Qwen） chat_model = ChatTongyi( model="qwen-plus", # 可选: qwen-turbo, qwen-plus, qwen-max api_key=api_key, temperature=0.7 ) # 2. 创建提示模板 prompt_template = ChatPromptTemplate.from_messages([ ("system", "你是一个友好的AI助手，名字叫小智。"), MessagesPlaceholder(variable_name="chat_history"), ("human", "{user_input}") ]) # 3. 创建处理链 chain = prompt_template | chat_model # 4. 创建历史记录存储字典 store = {} def get_session_history(session_id: str) -> ChatMessageHistory: if session_id not in store: store[session_id] = ChatMessageHistory() return store[session_id] # 5. 创建带有历史记录的链 chatbot = RunnableWithMessageHistory( chain, get_session_history, input_messages_key="user_input", history_messages_key="chat_history" ) # 6. 模拟对话 def chat(session_id, message): response = chatbot.invoke( {"user_input": message}, config={"configurable": {"session_id": session_id}} ) return response.content # 测试对话 session = "user_123" print("AI:", chat(session, "你好！")) print("AI:", chat(session, "你叫什么名字？")) print("AI:", chat(session, "我们刚才聊了什么？"))

输出：

AI: 你好呀！有什么我可以帮你的吗？😊 AI: 我叫小智，很高兴认识你！🌟 有什么问题或者需要帮助的吗？ AI: 我们刚刚聊到我是小智，一个友好且乐于助人的AI助手！🌟 你还问我我们之前聊了什么，现在你看到的就是我们的最新对话啦！有什么其他想知道的吗？😊

这个实现完整展示了LangChain的对话记忆功能，能够跨多个交互回合保持上下文一致性。