35.RAG开发-22-Memory长期会话记忆

memory 长期会话记忆

FileChatMessageHistory类实现，核心思路：

• 基于文件存储会话记录，以session_id为文件名，不同session_id有不同文件存储消息

继承BaseChatMessageHistory实现如下3个方法：

• add_messages:同步模式，添加消息
• messages:同步模式，获取消息
• clear: 同步模式，清除消息

# 导入必要的库和模块
import json  # 用于JSON序列化和反序列化
import os    # 用于文件和目录操作
from typing import Sequence  # 用于类型注解

# 从LangChain核心模块导入必要的组件
from langchain_core.messages import (
    messages_from_dict,  # 将字典列表转换为消息对象列表
    message_to_dict,     # 将消息对象转换为字典
    BaseMessage         # 消息基类
)
from langchain_core.chat_history import BaseChatMessageHistory  # 聊天历史基类
from langchain_core.runnables.history import RunnableWithMessageHistory  # 带消息历史的可运行对象
from langchain_community.chat_models.tongyi import ChatTongyi  # 通义千问模型
from langchain_core.output_parsers import StrOutputParser  # 字符串输出解析器
from langchain_core.prompts import MessagesPlaceholder, ChatPromptTemplate  # 提示模板相关


# 消息转换函数说明：
# message_to_dict：单个消息对象（BaseMessage类实例）→ 字典
# messages_from_dict：[字典、字典...] → [消息、消息...]
# AIMessage、HumanMessage、SystemMessage都是BaseMessage的子类


class FileChatMessageHistory(BaseChatMessageHistory):
    """
    基于文件的聊天消息历史记录类
    
    该类继承自BaseChatMessageHistory，实现了基于文件系统的消息历史存储
    每个会话的消息历史会被存储到对应的文件中，支持消息的添加、获取和清空操作
    """
    
    def __init__(self, storage_path: str, session_id: str) -> None:
        """
        初始化FileChatMessageHistory实例
        
        Args:
            storage_path: 存储路径，用于存放会话历史文件
            session_id: 会话ID，用于唯一标识一个会话
        """
        self.storage_path = storage_path
        self.session_id = session_id

        # 构建文件路径：存储路径 + 会话ID
        self.file_path = os.path.join(self.storage_path, self.session_id)

        # 确保存储路径存在，如果不存在则创建
        os.makedirs(os.path.dirname(self.file_path), exist_ok=True)

    @property
    def messages(self) -> list[BaseMessage]:
        """
        获取当前会话的所有消息列表
        
        Returns:
            list[BaseMessage]: 消息对象列表，如果文件不存在则返回空列表
        """
        try:
            # 打开文件并读取消息数据
            with open(
                self.file_path,
                "r",
                encoding="utf-8"
            ) as f:
                messages_data = json.load(f)
                # 将字典列表转换为消息对象列表
                return messages_from_dict(messages_data)
        except FileNotFoundError:
            # 如果文件不存在，返回空列表
            return []

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        """
        添加消息到历史记录
        
        Args:
            messages: 要添加的消息序列
        """
        # 获取当前消息列表
        all_messages = list(self.messages)
        # 添加新消息
        all_messages.extend(messages)

        # 序列化消息：将消息对象列表转换为字典列表
        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)

    def clear(self) -> None:
        """
        清空消息历史记录
        """
        # 写入空列表，清空文件内容
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump([], f)


def print_prompt(full_prompt):
    """
    打印完整的提示信息
    
    Args:
        full_prompt: 完整的提示对象
    
    Returns:
        full_prompt: 原样返回提示对象，用于链式调用
    """
    # 打印提示信息，前后用等号包围以增强可读性
    print("="*20, full_prompt.to_string(), "="*20)
    return full_prompt


# 初始化通义千问模型
model = ChatTongyi(model="qwen3-max")

# 定义提示模板，包含对话历史和用户输入
# 注释掉的是使用PromptTemplate的方式
# prompt = PromptTemplate.from_template(
#     "你需要根据对话历史回应用户问题，对话历史：{chat_history}\n用户当前输入：{input}, 请给出回应"
# )

# 使用ChatPromptTemplate构建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你需要根据对话历史回应用户问题，对话历史："),
    MessagesPlaceholder(variable_name="chat_history"),  # 消息历史占位符
    ("human", "请回答如下问题：{input}"),  # 用户输入占位符
])

# 构建基础链：提示模板 → 打印提示 → 模型 → 输出解析
base_chain = prompt | print_prompt | model | StrOutputParser()

# 会话历史存储字典（当前未使用，保留以备扩展）
chat_history_store = {}


def get_history(session_id):
    """
    获取指定会话ID的历史记录对象
    
    Args:
        session_id: 会话ID
    
    Returns:
        FileChatMessageHistory: 会话历史对象
    """
    return FileChatMessageHistory("./chat_history", session_id)


# 创建带有消息历史功能的对话链
conversation_chain = RunnableWithMessageHistory(
    base_chain,          # 基础链对象
    get_history,         # 获取历史会话的函数
    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": "小明有一只猫"}, config=session_config)
    # print("第一次执行：", res)
    
    # # 测试对话：询问小刚的狗
    # res = conversation_chain.invoke({"input": "小刚有两只狗"}, config=session_config)
    # print("第二次执行：", res)
    
    # 测试对话：基于历史询问总共有几只宠物
    res = conversation_chain.invoke({"input": "共有几只宠物？"}, config=session_config)
    print("第三次执行：", res)

==================== System: 你需要根据对话历史回应用户问题，对话历史：
Human: 请回答如下问题：小明有一只猫 ====================
第一次执行： 小明有一只猫。请问您想了解关于这只猫的更多信息，还是有其他问题需要解答呢？
==================== System: 你需要根据对话历史回应用户问题，对话历史：
Human: 小明有一只猫
AI: 小明有一只猫。请问您想了解关于这只猫的更多信息，还是有其他问题需要解答呢？
Human: 请回答如下问题：小刚有两只狗 ====================
第二次执行： 小刚有两只狗。  
如果您有进一步的问题，比如关于小刚的狗、小明的猫，或者他们之间的关系等，请随时告诉我！

==================== System: 你需要根据对话历史回应用户问题，对话历史：
Human: 小明有一只猫
AI: 小明有一只猫。请问您想了解关于这只猫的更多信息，还是有其他问题需要解答呢？
Human: 小刚有两只狗
AI: 小刚有两只狗。
如果您有进一步的问题，比如关于小刚的狗、小明的猫，或者他们之间的关系等，请随时告诉我！
Human: 请回答如下问题：共有几只宠物？ ====================
第三次执行： 小明有1只猫，小刚有2只狗。  
所以他们一共有：
1 + 2 = **3只宠物**。

实现自定义 FileChatMessageHistory：会话历史持久化到本地文件

核心内容

• 通过自定义 FileChatMessageHistory 类，将聊天会话历史持久化到本地 JSON 文件。
• 解决程序重启后历史记录丢失的问题。

一、核心概念对比

聊天历史存储方式对比

核心区别：

• 临时记忆：速度快但易失，适合开发调试。
• 长期记忆：数据持久化，适合用户对话历史的保存。

二、代码实现：自定义 FileChatMessageHistory

1. 导入必要的模块

import json
import os
from typing import List, Sequence
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, messages_from_dict, messages_to_dict

说明：

• BaseChatMessageHistory：所有聊天历史记录类的基类。
• messages_from_dict / messages_to_dict：消息序列化与反序列化的工具函数。

2. 实现自定义类

class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, file_path: str):
        self.file_path = file_path

    @property
    def messages(self) -> List[BaseMessage]:
        """获取历史消息列表"""
        # 如果文件不存在，返回空列表
        if not os.path.exists(self.file_path):
            return []
        # 读取文件内容
        with open(self.file_path, 'r', encoding='utf-8') as f:
            messages_data = json.load(f)
        # 将字典列表转换为消息对象列表
        return messages_from_dict(messages_data)

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        """添加消息到历史记录"""
        # 获取现有消息
        existing_messages = self.messages
        # 合并新消息
        all_messages = existing_messages + list(messages)
        # 将消息对象列表转换为可序列化的字典列表
        messages_data = messages_to_dict(all_messages)
        # 写入文件
        with open(self.file_path, 'w', encoding='utf-8') as f:
            json.dump(messages_data, f, ensure_ascii=False, indent=4)

    def clear(self) -> None:
        """清空历史记录"""
        # 直接写入空列表即可
        with open(self.file_path, 'w', encoding='utf-8') as f:
            json.dump([], f)

关键点解析：

• messages 属性：使用 @property 装饰器，访问时自动触发读取文件逻辑。
• add_messages 方法：必须实现，RunnableWithMessageHistory 会自动调用此方法保存新对话。
• 序列化逻辑：
  • 存储时：BaseMessage 对象 → dict → JSON 字符串。
  • 读取时：JSON 字符串 → dict → BaseMessage 对象。

三、代码实现：历史记录管理函数

1. 定义存储路径与缓存字典

# 定义历史记录存储目录
history_dir = "chat_history"
os.makedirs(history_dir, exist_ok=True)

# 用于缓存历史记录实例的字典
history_cache = {}

2. 定义获取历史记录的函数

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    """
    根据 session_id 获取对应的历史记录实例
    """
    # 构建文件路径：chat_history/{session_id}.json
    file_path = os.path.join(history_dir, f"{session_id}.json")
    # 如果该 session_id 的实例未创建，则创建并缓存
    if session_id not in history_cache:
        history_cache[session_id] = FileChatMessageHistory(file_path)
    return history_cache[session_id]

说明：

• session_id：会话唯一标识符（如用户ID、会话ID）。
• history_cache：缓存字典，避免重复创建对象。

四、代码实现：构建带长期记忆的对话链

1. 初始化模型与提示词模板

from langchain_community.chat_models import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory

# 初始化模型
model = ChatTongyi(model="qwen-plus")

# 定义提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个乐于助人的AI助手。"),
    MessagesPlaceholder(variable_name="history"),  # 历史消息占位符
    ("human="{input}")                                  # 用户输入占位符
])

关键点：

• MessagesPlaceholder：必须使用此占位符，LangChain 会自动填充历史消息。
• 变量名：history 和 input 需保持一致。

2. 构建基础链与包装链

# 构建基础链（提示词 → 模型）
base_chain = prompt | model

# 包装为带历史记录的链
chain_with_history = RunnableWithMessageHistory(
    base_chain,
    get_session_history,         # 历史记录获取函数
    input_messages_key="input",  # 输入消息的键名
    history_messages_key="history"  # 历史消息的键名
)

参数详解：

五、代码实现：测试长期记忆效果

1. 第一轮对话

# 配置 session_id
config = {"configurable": {"session_id": "user_001"}}

# 发起对话
response1 = chain_with_history.invoke(
    {"input": "你好，我是小明，我喜欢打篮球。"},
    config=config
)
print(f"AI回复: {response1.content}")

后台动作：

1. LangChain 调用 get_session_history("user_001")
2. 读取文件 chat_history/user_001.json（此时为空）
3. 将用户输入填入模板，调用模型
4. 将用户消息和 AI 回复写入文件

2. 第二轮对话（验证记忆）

response2 = chain_with_history.invoke(
    {"input": "我刚才说我喜欢什么运动？"},
    config=config
)
print(f"AI回复: {response2.content}")
# 输出: 你刚才说你喜欢打篮球。

3. 验证程序重启后的记忆

# 模拟程序重启：清空缓存，重新创建对象
history_cache.clear()

# 再次提问，依然能记住之前的对话
response3 = chain_with_history.invoke(
    {"input": "我是谁？"},
    config=config
)
print(f"AI回复: {response3.content}")
# 输出: 你是小明。

六、总结