历史会话记录功能的实现

代码实践

创建名为file_history_store.py的文件，将langchain中的长期会话记忆的类实现复制过来

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


def get_history(session_id):
    return FileChatMessageHistory(session_id, "./chat_history")


class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, session_id, storage_path):
        self.session_id = session_id              # 会话id
        self.storage_path = storage_path          # 不同会话id的存储文件，所在的文件夹路径
        # 完整的文件路径
        self.file_path = os.path.join(self.storage_path, self.session_id)

        # 确保文件夹是存在的
        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, ensure_ascii=False,indent=4)

    @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)

回到rag.py,在原来基础上修改：

from langchain_core.documents import Document
from langchain_core.runnables import RunnablePassthrough, RunnableWithMessageHistory, RunnableLambda
from vector_stores import VectorStoresService
from langchain_community.embeddings import DashScopeEmbeddings
import os
from dotenv import load_dotenv
import config_data as config
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models import ChatTongyi
from langchain_core.output_parsers import StrOutputParser
from file_history_store import get_history

load_dotenv()


def print_prompt(prompt):
    print("="*20)
    print(prompt.to_string())
    print('='*20)
    return prompt

class RagService(object):

    def __init__(self):

        self.vector_service = VectorStoresService(
            embedding=DashScopeEmbeddings(model=config.embedding_model_name,dashscope_api_key=os.getenv("LLM_API_KEY"))
        )

        self.prompt_template = ChatPromptTemplate.from_messages(
            [
                ("system", "以我提供的已知参考资料为主, 简洁和专业的回答用户问题。参考资料:{context}。"),
                ("system", "用户的对话历史记录如下："),
                MessagesPlaceholder("history"),
                ("user", "用户提问: {input}")
            ]
        )

        self.chat_model = ChatTongyi(model=config.chat_model_name,api_key=os.getenv("LLM_API_KEY"))

        self.chain = self.__get_chain()


    def __get_chain(self):
        """获取最终的执行链"""
        retriever = self.vector_service.get_retriever()

        def format_document(docs: list[Document]):
            if not docs:
                return "无相关参考资料"
            
            formatted_str = ""
            for doc in docs:
                formatted_str += f"文档片段：{doc.page_content}\n文档元数据：{doc.metadata}"

            return formatted_str

        def format_for_retriever(value:dict) -> str:
            return value["input"]

        def format_for_prompt_template(value):
            # {'input': {'input': '我身高170厘米，尺码推荐', 'history': []}, 'context': "文档片段：}
            # {input, context, history}
            new_value = {}
            new_value["input"] = value["input"]["input"]
            new_value["context"] = value["context"]
            new_value["history"] = value["input"]["history"]

            return new_value

        chain = (
            {
                "input": RunnablePassthrough(),
                "context": RunnableLambda(format_for_retriever) | retriever | format_document
            } | RunnableLambda(format_for_prompt_template) | self.prompt_template | print_prompt | self.chat_model | StrOutputParser()
        )

        conversation_chain = RunnableWithMessageHistory(
            chain,
            get_history,
            input_messages_key="input",
            history_messages_key="history"
        )

        return conversation_chain

if __name__ == "__main__":
    # session id配置
    session_config = {
        "configurable":{
            "session_id": "user_001",
        }
    }
    # 增强链需要的输入是字典
    # res = RagService().chain.invoke({"input": "我的体重120斤，身高172，尺码推荐"}, session_config)
    res = RagService().chain.invoke({"input": "春天穿什么衣服"}, session_config)
    print(res)