AI–LangChain会话记忆

1.Memory临时会话记忆

  在大模型的使用中，免不了要封装历史记录的功能，而除了自行维护历史消息外，也可以借助LangChain内置的历史记录附加功能，LangChain提供了History功能，帮助模型在有历史记忆的情况下进行回答

• RunnableWithMessageHistory

  在原有链的基础上创建带有历史记录功能的新链

• InMemoryChatMessageHistory

  为历史记录提供内存存储（临时使用的需求时）

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import MessagesPlaceholder , ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_models.tongyi import ChatTongyi

model = ChatTongyi(model="qwen-max")

# prompt = PromptTemplate.from_template(
#     "你需要根据会话历史回应用户问题。对话历史：{chat_history},用户问题：{input},请回答"
# )
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你需要根据会话历史回应用户问题，对话历史：{chat_history}"),
    MessagesPlaceholder("chat_history"),
    ("human", "请回答如下问题:{input}"),
])


str_parser = StrOutputParser()

def print_prompt(full_prompt):
    print("="*20 ,full_prompt, "="*20)
    return full_prompt

base_chain = chat_prompt | print_prompt | model | str_parser

store = {} #key就是session_id, value就是InMemoryChatMessageHistory对象  

#通过会话id获取InMemoryChatMessageHistory对象
def get_history(session_id):
    # 如果store中不存在session_id, 则创建一个新的InMemoryChatMessageHistory对象
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

#通过RunnableWithMessageHistory获取一个新的带有历史记录功能的会话chain
conversation_chain = RunnableWithMessageHistory(
    base_chain, # 要添加历史记录功能的chain
    get_history, # 通过会话id获取InMemoryChatMessageHistory对象
    input_messages_key="input",   # 输入消息的key
    history_messages_key="chat_history", # 历史记录的key
)

if __name__ == "__main__":
    #固定格式，添加LangChain的配置，为当前程序配置所属的session_id
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }


    res = conversation_chain.invoke({"input": "小明有2只猫"},session_config)
    print("第一次执行:",res)  

    res = conversation_chain.invoke({"input": "小刚有1只狗"},session_config)
    print("第二次执行:",res)  

    res = conversation_chain.invoke({"input": "他们有多少只动物"},session_config)
    print("第三次执行:",res)  

  

2.Memory长期会话记忆

  在使用大模型的时候，短期记忆肯定是不够的，这时候我们就需要了解长期会话记忆，在LangChain的官方注释中给出了相关实现的指南，并给出了基于文件的历史消息存储实例代码，我们可以自行实现一个基于json格式和本地文件的会话数据保存

  FileChatMessageHistory类实现，核心思路是基于文件存储会话记录，以session_id为文件并，不同session_id有不同文件存储消息

#官方memory长期会话记忆模版
import json,os
from typing import Sequence
from langchain_core.messages import messages_to_dict,messages_from_dict, BaseMessage,message_to_dict
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import MessagesPlaceholder , ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_models.tongyi import ChatTongyi

class FileChatMessageHistory(BaseChatMessageHistory):
    # 初始化会话id和存储路径
    def __init__(self, session_id, storage_path):
        self.session_id = session_id
        self.storage_path = storage_path
        self.file_path = os.path.join(self.storage_path, f"{self.session_id}.json")

         #确保存储路径存在
        os.makedirs(os.path.dirname(self.file_path), exist_ok=True)

    # 添加消息到会话历史
    def add_messages(self, message: Sequence[BaseMessage]) -> None: 
        # Sequence序列 类型list tuple
        all_messages = list(self.messages) #已有的消息列表
        all_messages.extend(message)                    # 合并新消息

        # 将数据同步写入到本地文件中
        # 类对象写入文件 -> 一堆二进制
        # 为了方便，可以将BaseMessage消息转为字典（借助json模块以json字符串写入文件）
        # 官方message_to_dict: 单个消息对象 (BaseMessage类实例) -> 字典
        new_messages = [message_to_dict(message) for message in all_messages]
        #new_messages = messages_to_dict(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:
                message_data = json.load(f)
                return messages_from_dict(message_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="qwen-max")

# prompt = PromptTemplate.from_template(
#     "你需要根据会话历史回应用户问题。对话历史：{chat_history},用户问题：{input},请回答"
# )
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你需要根据会话历史回应用户问题，对话历史：{chat_history}"),
    MessagesPlaceholder("chat_history"),
    ("human", "请回答如下问题:{input}"),
])


str_parser = StrOutputParser()

def print_prompt(full_prompt):
    print("="*20 ,full_prompt, "="*20)
    return full_prompt

base_chain = chat_prompt | print_prompt | model | str_parser


#通过会话id获取InMemoryChatMessageHistory对象
def get_history(session_id):
    return FileChatMessageHistory(session_id, "./chat_history")

#通过RunnableWithMessageHistory获取一个新的带有历史记录功能的会话chain
conversation_chain = RunnableWithMessageHistory(
    base_chain, # 要添加历史记录功能的chain
    get_history, # 通过会话id获取InMemoryChatMessageHistory对象
    input_messages_key="input",   # 输入消息的key
    history_messages_key="chat_history", # 历史记录的key
)

if __name__ == "__main__":
    #固定格式，添加LangChain的配置，为当前程序配置所属的session_id
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }


    # res = conversation_chain.invoke({"input": "小明有2只猫"},session_config)
    # print("第一次执行:",res)  

    # res = conversation_chain.invoke({"input": "小刚有1只狗"},session_config)
    # print("第二次执行:",res)  

    res = conversation_chain.invoke({"input": "他们有多少只动物"},session_config)
    print("第三次执行:",res)