memory 長期會話記憶

FileChatMessageHistory類實(shí)現(xiàn)，核心思路：

• 基于文件存儲會話記錄，以session_id為文件名，不同session_id有不同文件存儲消息

繼承BaseChatMessageHistory實(shí)現(xiàn)如下3個(gè)方法：

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

# 導(dǎo)入必要的庫和模塊
import json  # 用于JSON序列化和反序列化
import os    # 用于文件和目錄操作
from typing import Sequence  # 用于類型注解

# 從LangChain核心模塊導(dǎo)入必要的組件
from langchain_core.messages import (
    messages_from_dict,  # 將字典列表轉(zhuǎn)換為消息對象列表
    message_to_dict,     # 將消息對象轉(zhuǎn)換為字典
    BaseMessage         # 消息基類
)
from langchain_core.chat_history import BaseChatMessageHistory  # 聊天歷史基類
from langchain_core.runnables.history import RunnableWithMessageHistory  # 帶消息歷史的可運(yùn)行對象
from langchain_community.chat_models.tongyi import ChatTongyi  # 通義千問模型
from langchain_core.output_parsers import StrOutputParser  # 字符串輸出解析器
from langchain_core.prompts import MessagesPlaceholder, ChatPromptTemplate  # 提示模板相關(guān)


# 消息轉(zhuǎn)換函數(shù)說明：
# message_to_dict：單個(gè)消息對象（BaseMessage類實(shí)例）→ 字典
# messages_from_dict：[字典、字典...] → [消息、消息...]
# AIMessage、HumanMessage、SystemMessage都是BaseMessage的子類


class FileChatMessageHistory(BaseChatMessageHistory):
    """
    基于文件的聊天消息歷史記錄類
    
    該類繼承自BaseChatMessageHistory，實(shí)現(xiàn)了基于文件系統(tǒng)的消息歷史存儲
    每個(gè)會話的消息歷史會被存儲到對應(yīng)的文件中，支持消息的添加、獲取和清空操作
    """
    
    def __init__(self, storage_path: str, session_id: str) -> None:
        """
        初始化FileChatMessageHistory實(shí)例
        
        Args:
            storage_path: 存儲路徑，用于存放會話歷史文件
            session_id: 會話ID，用于唯一標(biāo)識一個(gè)會話
        """
        self.storage_path = storage_path
        self.session_id = session_id

        # 構(gòu)建文件路徑：存儲路徑 + 會話ID
        self.file_path = os.path.join(self.storage_path, self.session_id)

        # 確保存儲路徑存在，如果不存在則創(chuàng)建
        os.makedirs(os.path.dirname(self.file_path), exist_ok=True)

    @property
    def messages(self) -> list[BaseMessage]:
        """
        獲取當(dāng)前會話的所有消息列表
        
        Returns:
            list[BaseMessage]: 消息對象列表，如果文件不存在則返回空列表
        """
        try:
            # 打開文件并讀取消息數(shù)據(jù)
            with open(
                self.file_path,
                "r",
                encoding="utf-8"
            ) as f:
                messages_data = json.load(f)
                # 將字典列表轉(zhuǎn)換為消息對象列表
                return messages_from_dict(messages_data)
        except FileNotFoundError:
            # 如果文件不存在，返回空列表
            return []

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        """
        添加消息到歷史記錄
        
        Args:
            messages: 要添加的消息序列
        """
        # 獲取當(dāng)前消息列表
        all_messages = list(self.messages)
        # 添加新消息
        all_messages.extend(messages)

        # 序列化消息：將消息對象列表轉(zhuǎn)換為字典列表
        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:
        """
        清空消息歷史記錄
        """
        # 寫入空列表，清空文件內(nèi)容
        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: 原樣返回提示對象，用于鏈?zhǔn)秸{(diào)用
    """
    # 打印提示信息，前后用等號包圍以增強(qiáng)可讀性
    print("="*20, full_prompt.to_string(), "="*20)
    return full_prompt


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

# 定義提示模板，包含對話歷史和用戶輸入
# 注釋掉的是使用PromptTemplate的方式
# prompt = PromptTemplate.from_template(
#     "你需要根據(jù)對話歷史回應(yīng)用戶問題，對話歷史：{chat_history}\n用戶當(dāng)前輸入：{input}, 請給出回應(yīng)"
# )

# 使用ChatPromptTemplate構(gòu)建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你需要根據(jù)對話歷史回應(yīng)用戶問題，對話歷史："),
    MessagesPlaceholder(variable_name="chat_history"),  # 消息歷史占位符
    ("human", "請回答如下問題：{input}"),  # 用戶輸入占位符
])

# 構(gòu)建基礎(chǔ)鏈：提示模板 → 打印提示 → 模型 → 輸出解析
base_chain = prompt | print_prompt | model | StrOutputParser()

# 會話歷史存儲字典（當(dāng)前未使用，保留以備擴(kuò)展）
chat_history_store = {}


def get_history(session_id):
    """
    獲取指定會話ID的歷史記錄對象
    
    Args:
        session_id: 會話ID
    
    Returns:
        FileChatMessageHistory: 會話歷史對象
    """
    return FileChatMessageHistory("./chat_history", session_id)


# 創(chuàng)建帶有消息歷史功能的對話鏈
conversation_chain = RunnableWithMessageHistory(
    base_chain,          # 基礎(chǔ)鏈對象
    get_history,         # 獲取歷史會話的函數(shù)
    input_messages_key="input",       # 輸入消息的鍵名
    history_messages_key="chat_history"  # 歷史消息的鍵名
)


if __name__ == "__main__":
    """
    主函數(shù)，測試對話鏈的會話記憶功能
    """
    # 固定格式，添加LangChain的配置，為當(dāng)前程序配置所屬的session_id
    session_config = {"configurable": {"session_id": "user_001"}}
    
    # # 測試對話：詢問小明的貓
    # res = conversation_chain.invoke({"input": "小明有一只貓"}, config=session_config)
    # print("第一次執(zhí)行：", res)
    
    # # 測試對話：詢問小剛的狗
    # res = conversation_chain.invoke({"input": "小剛有兩只狗"}, config=session_config)
    # print("第二次執(zhí)行：", res)
    
    # 測試對話：基于歷史詢問總共有幾只寵物
    res = conversation_chain.invoke({"input": "共有幾只寵物？"}, config=session_config)
    print("第三次執(zhí)行：", res)

==================== System: 你需要根據(jù)對話歷史回應(yīng)用戶問題，對話歷史：
Human: 請回答如下問題：小明有一只貓 ====================
第一次執(zhí)行： 小明有一只貓。請問您想了解關(guān)于這只貓的更多信息，還是有其他問題需要解答呢？
==================== System: 你需要根據(jù)對話歷史回應(yīng)用戶問題，對話歷史：
Human: 小明有一只貓
AI: 小明有一只貓。請問您想了解關(guān)于這只貓的更多信息，還是有其他問題需要解答呢？
Human: 請回答如下問題：小剛有兩只狗 ====================
第二次執(zhí)行： 小剛有兩只狗。  
如果您有進(jìn)一步的問題，比如關(guān)于小剛的狗、小明的貓，或者他們之間的關(guān)系等，請隨時(shí)告訴我！

==================== System: 你需要根據(jù)對話歷史回應(yīng)用戶問題，對話歷史：
Human: 小明有一只貓
AI: 小明有一只貓。請問您想了解關(guān)于這只貓的更多信息，還是有其他問題需要解答呢？
Human: 小剛有兩只狗
AI: 小剛有兩只狗。
如果您有進(jìn)一步的問題，比如關(guān)于小剛的狗、小明的貓，或者他們之間的關(guān)系等，請隨時(shí)告訴我！
Human: 請回答如下問題：共有幾只寵物？ ====================
第三次執(zhí)行： 小明有1只貓，小剛有2只狗。  
所以他們一共有：
1 + 2 = **3只寵物**。

實(shí)現(xiàn)自定義 FileChatMessageHistory：會話歷史持久化到本地文件

核心內(nèi)容

• 通過自定義 FileChatMessageHistory 類，將聊天會話歷史持久化到本地 JSON 文件。
• 解決程序重啟后歷史記錄丟失的問題。

一、核心概念對比

聊天歷史存儲方式對比

核心區(qū)別：

• 臨時(shí)記憶：速度快但易失，適合開發(fā)調(diào)試。
• 長期記憶：數(shù)據(jù)持久化，適合用戶對話歷史的保存。

二、代碼實(shí)現(xiàn)：自定義 FileChatMessageHistory

1. 導(dǎo)入必要的模塊

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：消息序列化與反序列化的工具函數(shù)。

2. 實(shí)現(xiàn)自定義類

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 []
        # 讀取文件內(nèi)容
        with open(self.file_path, 'r', encoding='utf-8') as f:
            messages_data = json.load(f)
        # 將字典列表轉(zhuǎn)換為消息對象列表
        return messages_from_dict(messages_data)

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        """添加消息到歷史記錄"""
        # 獲取現(xiàn)有消息
        existing_messages = self.messages
        # 合并新消息
        all_messages = existing_messages + list(messages)
        # 將消息對象列表轉(zhuǎn)換為可序列化的字典列表
        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)

關(guān)鍵點(diǎn)解析：

• messages 屬性：使用 @property 裝飾器，訪問時(shí)自動觸發(fā)讀取文件邏輯。
• add_messages 方法：必須實(shí)現(xiàn)，RunnableWithMessageHistory 會自動調(diào)用此方法保存新對話。
• 序列化邏輯：
  • 存儲時(shí)：BaseMessage 對象 → dict → JSON 字符串。
  • 讀取時(shí)：JSON 字符串 → dict → BaseMessage 對象。

三、代碼實(shí)現(xiàn)：歷史記錄管理函數(shù)

1. 定義存儲路徑與緩存字典

# 定義歷史記錄存儲目錄
history_dir = "chat_history"
os.makedirs(history_dir, exist_ok=True)

# 用于緩存歷史記錄實(shí)例的字典
history_cache = {}

2. 定義獲取歷史記錄的函數(shù)

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    """
    根據(jù) session_id 獲取對應(yīng)的歷史記錄實(shí)例
    """
    # 構(gòu)建文件路徑：chat_history/{session_id}.json
    file_path = os.path.join(history_dir, f"{session_id}.json")
    # 如果該 session_id 的實(shí)例未創(chuàng)建，則創(chuàng)建并緩存
    if session_id not in history_cache:
        history_cache[session_id] = FileChatMessageHistory(file_path)
    return history_cache[session_id]

說明：

• session_id：會話唯一標(biāo)識符（如用戶ID、會話ID）。
• history_cache：緩存字典，避免重復(fù)創(chuàng)建對象。

四、代碼實(shí)現(xiàn)：構(gòu)建帶長期記憶的對話鏈

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", "你是一個(gè)樂于助人的AI助手。"),
    MessagesPlaceholder(variable_name="history"),  # 歷史消息占位符
    ("human="{input}")                                  # 用戶輸入占位符
])

關(guān)鍵點(diǎn)：

• MessagesPlaceholder：必須使用此占位符，LangChain 會自動填充歷史消息。
• 變量名：history 和 input 需保持一致。

2. 構(gòu)建基礎(chǔ)鏈與包裝鏈

# 構(gòu)建基礎(chǔ)鏈（提示詞 → 模型）
base_chain = prompt | model

# 包裝為帶歷史記錄的鏈
chain_with_history = RunnableWithMessageHistory(
    base_chain,
    get_session_history,         # 歷史記錄獲取函數(shù)
    input_messages_key="input",  # 輸入消息的鍵名
    history_messages_key="history"  # 歷史消息的鍵名
)

參數(shù)詳解：

五、代碼實(shí)現(xiàn)：測試長期記憶效果

1. 第一輪對話

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

# 發(fā)起對話
response1 = chain_with_history.invoke(
    {"input": "你好，我是小明，我喜歡打籃球。"},
    config=config
)
print(f"AI回復(fù): {response1.content}")

后臺動作：

1. LangChain 調(diào)用 get_session_history("user_001")
2. 讀取文件 chat_history/user_001.json（此時(shí)為空）
3. 將用戶輸入填入模板，調(diào)用模型
4. 將用戶消息和 AI 回復(fù)寫入文件

2. 第二輪對話（驗(yàn)證記憶）

response2 = chain_with_history.invoke(
    {"input": "我剛才說我喜歡什么運(yùn)動？"},
    config=config
)
print(f"AI回復(fù): {response2.content}")
# 輸出: 你剛才說你喜歡打籃球。

3. 驗(yàn)證程序重啟后的記憶

# 模擬程序重啟：清空緩存，重新創(chuàng)建對象
history_cache.clear()

# 再次提問，依然能記住之前的對話
response3 = chain_with_history.invoke(
    {"input": "我是誰？"},
    config=config
)
print(f"AI回復(fù): {response3.content}")
# 輸出: 你是小明。

六、總結(jié)