目录1.Memory临时会话记忆2.Memory长期会话记忆1.Memory临时会话记忆from langchain_community.chat_models import ChatTongyi from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import PromptTemplate from langchain_core.output_parsers import JsonOutputParser from langchain_core.runnables import RunnableWithMessageHistory from langchain_core.chat_history import InMemoryChatMessageHistory model ChatTongyi(modelqwen-max) prompt PromptTemplate.from_template( 你需要根据会话历史回应用户问题。对话历史{chat_history}用户问题{input}请给出简短的回答。 ) str_parser StrOutputParser() base_chain prompt | model | str_parser store {} def get_history(session_id): if session_id not in store: store[session_id] InMemoryChatMessageHistory() return store[session_id] #创建一个新的链对base_chain进行包装添加会话历史功能 conversation_chain RunnableWithMessageHistory( base_chain, # 基础链 get_history,# 获取会话历史的函数 input_messages_keyinput,#表示用户输入在模板中的占位符 history_messages_keychat_history#表示用户输入在模板中的占位符 ) if __name__ __main__: #固定格式添加langchain的配置为当前程序配置所属的session_id session_config { configurable: { session_id: user001 } } res conversation_chain.invoke(input{input:小明有2只猫},configsession_config) print(第一次调用结果,res) res conversation_chain.invoke(input{input:小刚有1只狗},configsession_config) print(第二次调用结果,res) res conversation_chain.invoke(input{input:总共有几个宠物},configsession_config) print(第三次调用结果,res)2.Memory长期会话记忆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_okTrue) def add_messages(self,messages:Sequence[BaseMessage]) - None: #Sequence:序列类型这里表示消息列表 all_messages list(self.messages) #已有的消息列表 all_messages.extend(messages) #新的和已有的融合成一个列表 #将所有消息写入文件 #类对象写入文件-一堆二进制 #为了方便将BaseMessage对象转为字典借助json模块以json字符串写入文件 #官方message_to_dict函数 new_messages [] for msg in all_messages: d message_to_dict(msg) new_messages.append(d) #将字典列表写入文件 with open(self.file_path,w,encodingutf-8) as f: json.dump(new_messages,f) property #装饰器将messages方法转换为成员属性 def messages(self) - list[BaseMessage]: #当前文件内list[字典] try: with open(self.file_path,r,encodingutf-8) as f: messages_data json.load(f) return messages_from_dict(messages_data) except FileNotFoundError: return [] def clear(self) - None: with open(self.file_path,w,encodingutf-8) as f: json.dump([],f)def get_history(session_id): return FileChatMessageHistory(session_id,storage_path./chat_history)