LangChain组件 Memory（记忆）
LangChain 的 Memory（记忆） 组件是解决大语言模型（LLM）“无状态” 问题的核心模块。它通过存储和管理对话历史，让模型能够记住之前的对话内容，从而实现连贯的多轮交互。
临时会话记忆
如果想要封装历史记录，除了自行维护历史消息外，也可以借助LangChain内置的历史记录附加功能
LangChain提供了History功能，帮助模型在有历史记忆的情况下回答
- 基于
RunnableWithMessageHistory在原有链的基础上创建带有历史记录功能的新链（新Runnable实例） - 基于
InMemoryChatMessageHistory为历史记录提供内存存储（临时用）
RunnableWithMessageHistory是LangChain内Runnable接口的实现，主要用于：
- 创建一个带有历史记忆功能的Runnable实例（链）
它在创建的时候需要提供一个BaseChatMessageHistory的具体实现（用来存储历史消息）
InMemoryChatMessageHistory可以实现在内存中存储历史
额外地，如果想要在invoke或stream执行链的同时，将提示词print出来，可以在链中加入自定义函数实现。
注意：函数的输入应原封不动返回回去，避免破坏原有业务
以下是基于该临时记忆实现的多轮对话功能示例：
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistorymodel = ChatTongyi(model="deepseek-v3.2")chat_template = ChatPromptTemplate.from_messages([("system", "你是一个话少的AI， 只会简短回答问题， 不会主动发起对话"),("ai", "你好"),MessagesPlaceholder(variable_name="chat_history"),("user", "{input}"),]
)chat_history_store = {} # 存放多个会话ID所对应的历史记录def get_history(session_id):if session_id not in chat_history_store:chat_history_store[session_id] = InMemoryChatMessageHistory()return chat_history_store[session_id]base_chain = chat_template | model | StrOutputParser()# 通过RunnableWithMessageHistory获取一个新的带有历史记录功能的Chain
conversation_chain = RunnableWithMessageHistory(base_chain, # 被附加历史消息的Runnable， 通常是Chainget_history, # 获取历史消息的函数， 需要根据会话ID返回对应的历史记录对象input_messages_key="input", # 声明用户输入消息在模板中的占位符history_messages_key="chat_history" # 声明历史消息在模板中的占位符
)if __name__ == "__main__":session_id = "session_1" # 模拟一个会话IDsession_config = {"configurable": {"session_id": session_id}}while True:user_input = input("用户输入：")if user_input.lower() == "exit":breakresponse = conversation_chain.invoke({"input": user_input}, session_config)print("模型回复：", response)
长期会话记忆
使用InMemoryChatMessageHistory仅可以在内存中存储会话记忆，一旦程序退出，记忆丢失。
InMemoryChatMessageHistory类继承自BaseChatMessageHistory，要实现长期会话记忆，我们可以自行实现一个基于Json格式和本地文件的会话数据保存
FileChatMessageHistory类实现，核心思路：
- 基于文件存储会话记录，以session_id为文件名，不同session_id有不同文件存储消息
继承BaseChatMessageHistory实现如下3个方法：
- add_messages：同步模式，添加消息
- message：同步模式，获取消息
- clear：同步模式，清除消息
from typing import Sequence
from uu import encodefrom langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory, BaseChatMessageHistoryimport json, os
from langchain_core.messages import messages_from_dict, message_to_dict, BaseMessageclass FileChatMessageHistory(BaseChatMessageHistory):storage_path: strsession_id: strdef __init__(self, session_id: str, storage_path: str):self.storage_path = storage_pathself.session_id = session_id# 完整路径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)@property # @property装饰器将方法转换为属性，使得调用时不需要加括号 可以当成属性一样访问def messages(self) -> list[BaseMessage]:try:with open(self.file_path, "r", encoding="utf-8") as f:# 当前内容 list[字典]messages_data = json.load(f)# 转换为 list[BaseMessage]return messages_from_dict(messages_data)except FileNotFoundError:return []def add_messages(self, messages: Sequence[BaseMessage]) -> None:# Sequence 序列 包括 list tuple 等等 这里转为list 方便后续操作all_messages = list(self.messages)# 追加新的消息all_messages.extend(messages)# 将BaseMessage对象转换为可序列化的字典列表serialized = [message_to_dict(message) for message in all_messages]# 将数据写入文件with open(self.file_path, "w", encoding="utf-8") as f:json.dump(serialized, f)def clear(self) -> None:with open(self.file_path, "w", encoding="utf-8") as f:json.dump([], f)model = ChatTongyi(model="qwen-plus")chat_template = ChatPromptTemplate.from_messages([("system", "你是一个话少的AI， 只会简短回答问题， 不会主动发起对话"),("ai", "你好"),MessagesPlaceholder(variable_name="chat_history"),("user", "{input}"),]
)def get_history(session_id):return FileChatMessageHistory(session_id, "./chat_history")base_chain = chat_template | model | StrOutputParser()# 通过RunnableWithMessageHistory获取一个新的带有历史记录功能的Chain
conversation_chain = RunnableWithMessageHistory(base_chain, # 被附加历史消息的Runnable， 通常是Chainget_history, # 获取历史消息的函数， 需要根据会话ID返回对应的历史记录对象input_messages_key="input", # 声明用户输入消息在模板中的占位符history_messages_key="chat_history" # 声明历史消息在模板中的占位符
)if __name__ == "__main__":session_id = "session_1" # 模拟一个会话IDsession_config = {"configurable": {"session_id": session_id}}while True:user_input = input("用户输入：")if user_input.lower() == "exit":breakresponse = conversation_chain.invoke({"input": user_input}, session_config)print("模型回复：", response)