22. LangChain组件——memory长期会话记忆,科技·AI·新纪元,前沿与探索,鱼C论坛

糖逗发表于昨天 08:38

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

1. 长期会话记忆实现的核心思想步骤？

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) -> 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 =
   # 将数据写入文件
   with open(self.file_path, "w", encoding="utf-8") as f:
         json.dump(new_messages, f)

@property    # @property装饰器将messages方法变成成员属性用
def messages(self) -> List:
   # 当前文件内： 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技术从大模型提示词到实战项目】https://www.bilibili.com/video/BV1yjz5BLEoY/?p=38&share_source=copy_web&vd_source=792a2cb63a1882bff4ed856eadc41a71]https://www.bilibili.com/video/BV1yjz5BLEoY/?p=38&share_source=copy_web&vd_source=792a2cb63a1882bff4ed856eadc41a71

FishC 发表于昨天 08:39

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

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

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

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

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

【代码实战填空】

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

1. 填空1（FileChatMessageHistory类的初始化方法）：
   self.storage_path = storage_path # 使用传入的存储路径

2. 填空2（add_messages方法的转换）：
   new_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 小助理，如未能正确解答您的问题，请继续追问。

不二如是 发表于昨天 09:36

学习

糖逗发表于昨天 09:44

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

{:13_450:}

不二如是 发表于昨天 09:50

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

加油！

页: [1]

鱼C论坛's Archiver

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