22. LangChain组件——memory长期会话记忆

糖逗 · 发表于 3 小时前

马上注册，结交更多好友，享用更多功能^_^

您需要登录才可以下载或查看，没有账号？立即注册

x

1. 长期会话记忆实现的核心思想步骤？
下载 (41).png

2. 代码实战

import os, json
from typing import Sequence
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import message_to_dict, messages_from_dict, BaseMessage
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistory
from typing import List
# message_to_dict：单个消息对象（BaseMessage类实例） -> 字典
# messages_from_dict：[字典、字典...] -> [消息、消息...]
# AIMessage、HumanMessage、SystemMessage 都是BaseMessage的子类
class FileChatMessageHistory(BaseChatMessageHistory):
def __init__(self, session_id, storage_path):
self.session_id = session_id # 会话id
self.storage_path = r'E:\AI大模型RAG与智能体开发\P3_LangChainRAG开发\chat_history' # 不同会话id的存储文件，所在的文件夹路径 <b>（按情况改成自己的文件存储地址）</b>
# 完整的文件路径
self.file_path = os.path.join(self.storage_path, self.session_id)
print(self.file_path)
# 确保文件夹是存在的
os.makedirs(os.path.dirname(self.file_path), exist_ok=True)
def add_messages(self, messages: Sequence[BaseMessage]) -> None:
# Sequence序列类似list、tuple
all_messages = list(self.messages) # 已有的消息列表
all_messages.extend(messages) # 新的和已有的融合成一个list
# 将数据同步写入到本地文件中
# 类对象写入文件 -> 一堆二进制
# 为了方便，可以将BaseMessage消息转为字典（借助json模块以json字符串写入文件）
# 官方message_to_dict：单个消息对象（BaseMessage类实例） -> 字典
# new_messages = []
# for message in all_messages:
# d = message_to_dict(message)
# new_messages.append(d)
new_messages = [message_to_dict(message) for message in all_messages]
# 将数据写入文件
with open(self.file_path, "w", encoding="utf-8") as f:
json.dump(new_messages, f)
@property # @property装饰器将messages方法变成成员属性用
def messages(self) -> List[BaseMessage]:
# 当前文件内： list[字典]
try:
with open(self.file_path, "r", encoding="utf-8") as f:
messages_data = json.load(f) # 返回值就是：list[字典]
return messages_from_dict(messages_data)
except FileNotFoundError:
return []
def clear(self) -> None:
with open(self.file_path, "w", encoding="utf-8") as f:
json.dump([], f)
model = ChatTongyi(model="qwen3-max")
# prompt = PromptTemplate.from_template(
# "你需要根据会话历史回应用户问题。对话历史：{chat_history}，用户提问：{input}，请回答"
# )
prompt = ChatPromptTemplate.from_messages(
[
("system", "你需要根据会话历史回应用户问题。对话历史："),
MessagesPlaceholder("chat_history"),
("human", "请回答如下问题：{input}")
]
)
str_parser = StrOutputParser()
def print_prompt(full_prompt):
print("="*20, full_prompt.to_string(), "="*20)
return full_prompt
base_chain = prompt | print_prompt | model | str_parser
def get_history(session_id):
return FileChatMessageHistory(session_id, "./chat_history")
# 创建一个新的链，对原有链增强功能：自动附加历史消息
conversation_chain = RunnableWithMessageHistory(
base_chain, # 被增强的原有chain
get_history, # 通过会话id获取InMemoryChatMessageHistory类对象
input_messages_key="input", # 表示用户输入在模板中的占位符
history_messages_key="chat_history" # 表示用户输入在模板中的占位符
)
if __name__ == '__main__':
# 固定格式，添加LangChain的配置，为当前程序配置所属的session_id
session_config = {
"configurable": {
"session_id": "user_001"
}
}
res = conversation_chain.invoke({"input": "小明有2个猫"}, session_config)
print("第1次执行：", res)
res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
print("第2次执行：", res)
res = conversation_chain.invoke({"input": "总共有几个宠物"}, session_config)
print("第3次执行：", res)

复制代码

学习视频：【黑马程序员大模型RAG与Agent智能体项目实战教程，基于主流的LangChain技术从大模型提示词到实战项目】[url=https://www.bilibili.com/video/B ... 2bff4ed856eadc41a71]https://www.bilibili.com/video/B ... 2bff4ed856eadc41a71[/url]

FishC · 发表于 3 小时前

【长期会话记忆实现的核心思想步骤】

长期会话记忆的核心思想是通过持久化存储来保存对话历史，主要分为三个步骤：

1. 历史消息的存储
- 将每次对话的消息（用户输入和AI回复）保存到外部存储（如文件、数据库）
- 每个会话有独立的存储标识（如session_id）

2. 历史消息的读取
- 每次新对话时，从存储中读取该会话的历史消息
- 将历史消息作为上下文输入给模型

3. 历史消息的管理
- 管理消息的添加、清理和更新
- 处理消息的序列化和反序列化（转换格式）

【代码实战填空】

以下是需要填写的代码部分：

1. 填空1（FileChatMessageHistory类的初始化方法）：

self.storage_path = storage_path # 使用传入的存储路径

复制代码

2. 填空2（add_messages方法的转换）：

new_messages = [message_to_dict(message) for message in all_messages]

复制代码

3. 填空3（messages方法的异常处理）：

return [] # 返回空列表

复制代码

4. 填空4（clear方法的清空操作）：

json.dump([], f) # 写入空列表

复制代码

5. 填空5（创建prompt模板）：

[
("system", "你需要根据会话历史回应用户问题。"),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}")
]

复制代码

【完整代码说明】

这个示例展示了如何实现基于文件存储的长期会话记忆：

工作原理：
1. 每个用户会话（session_id）对应一个独立的文件
2. 每次对话时：
- 从文件读取历史消息
- 将历史消息和当前输入一起传给AI模型
- 将新的对话消息追加到文件中

关键组件：
- FileChatMessageHistory：自定义的存储类，继承自BaseChatMessageHistory
- RunnableWithMessageHistory：自动管理消息历史的链
- message_to_dict 和 messages_from_dict：消息格式转换工具

运行效果：
程序会保存对话历史，因此在第三次提问“总共有几个宠物”时，AI能基于前两次对话（2只猫+1只狗）正确回答“总共3个宠物”。

以上回复来自 FishC AI 小助理，如未能正确解答您的问题，请继续追问。

不二如是 · 发表于 2 小时前

学习

糖逗 · 发表于 2 小时前

不二如是发表于 2026-3-8 09:36
学习

不二如是 · 发表于 2 小时前

糖逗发表于 2026-3-8 09:44

加油！

账号		自动登录	找回密码
密码			立即注册

[AI工作流] 22. LangChain组件——memory长期会话记忆

马上注册，结交更多好友，享用更多功能^_^

评分

本帖被以下淘专辑推荐: