langchain学习--会话记忆
临时记忆
理解LangChain的临时记忆机制
LangChain的临时记忆指模型在单次对话或短时间内保留上下文信息的能力，通常通过缓存或短期存储实现。其核心是维护对话状态，确保在多轮交互中连贯响应。
实现临时记忆的关键方法
对话历史缓存
将用户与模型的交互记录存储在内存或轻量级数据库（如Redis）中，每次请求时加载最近的N条记录。例如：
from collections import deque chat_history = deque(maxlen=5) # 保留最近5轮对话 def add_to_history(user_input, bot_response): chat_history.append({"user": user_input, "bot": bot_response})上下文窗口管理
通过限制Token数量或轮次控制记忆范围，避免过度消耗资源。例如设定最大上下文长度为2048个Token，超出时自动裁剪早期内容。
会话标识绑定
为每个用户或会话分配唯一ID，关联独立的记忆存储。可通过如下方式实现：
session_memories = {} def get_memory(session_id): return session_memories.get(session_id, [])临时记忆的优化策略
- 重要性标记：对关键信息（如用户偏好）加权存储，延长其保留时间。
- 自动摘要：对长对话生成摘要，替代原始内容以减少存储压力。
- 超时清理：设置不活跃会话的过期时间，定期清除陈旧数据。
临时记忆使用介绍
使用LangChain中的RunnableWithMessageHistory
构造函数：
""" RunnableWithMessageHistory( runnable, # 必填：要被包装的链/模型 get_session_history, # 必填：获取历史记录的函数 input_messages_key, # 可选：输入消息在 prompt 中的变量名 history_messages_key, # 可选：历史消息在 prompt 中的变量名 history_factory_config # 可选：历史消息工厂配置 ) """InMemoryChatMessageHistory 是 LangChain 提供的一个内存级别的对话历史存储类，用于在程序运行期间临时保存对话记录。
class InMemoryChatMessageHistory(BaseChatMessageHistory): # 属性 messages: List[BaseMessage] # 存储所有消息的列表 # 方法 def add_message(self, message: BaseMessage) -> None def add_user_message(self, message: str) -> None def add_ai_message(self, message: str) -> None def clear(self) -> NoneRunnableWithMessageHistory简单示例：
from langchain_community.chat_models.tongyi import ChatTongyi from dotenv import load_dotenv from langchain_core.prompts import PromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.chat_history import InMemoryChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory load_dotenv() model = ChatTongyi(model="qwen-plus") def print_prompt(full_prompt): print("="*20,full_prompt,"="*20) return full_prompt # 创建一个提示词 prompt=ChatPromptTemplate.from_messages( [ ("system", "你需要根据用户提供给你的对话历史来回答问题。对话历史：{chat_history}，用户输入：{input}，请给出简单回答"), MessagesPlaceholder("chat_history"), ("human", "{input}") ] ) #MessagesPlaceholder 是 LangChain 中用于在 ChatPromptTemplate 中动态插入消息列表的占位符。 # 创建一个基础运行链 base_chain=prompt|print_prompt|model|StrOutputParser() # 创建一个存储对话历史的字典 chat_history_store={} 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] #使用RunnableWithMessageHistory conversation_chain=RunnableWithMessageHistory( base_chain,#被附加历史消息的Runnable，通常是chain get_history,#获取历史消息的函数 input_messages_key="input",#输入消息的key history_messages_key="chat_history"#历史消息的key ) if __name__ == '__main__': #配置当前会话ID session_config={"configurable":{"session_id":"user_001"}} print(conversation_chain.invoke({"input":"我现在有3个苹果"},session_config)) print(conversation_chain.invoke({"input":"你有7个苹果"},session_config)) print(conversation_chain.invoke({"input":"给我三个你还剩几个"},session_config))关于会话id配置介绍：
LangChain的配置分为多个层级：
config = {
"tags": ["important", "test"], # 标签（用于追踪）
"metadata": {"author": "小明"}, # 元数据
"callbacks": [my_callback], # 回调函数
"configurable": { # 可配置参数（传递给组件）
"session_id": "user_001",
"model_name": "qwen-plus"
},
"run_name": "my_chain_run" # 运行名称
}
顶层字段（tags, metadata, callbacks）：LangChain框架自己使用
configurable内部：开发者自定义的参数，传递给具体的组件（如get_history函数）
长期会话记忆--文件存储
需要自己编写相关类（该类需要继承BaseChatMessageHistory）
示例如下：
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 序列 all_message=list(self.messages) # 获取当前所有消息 all_message.extend(messages) # 添加新消息 # 将数据同步写入文件中 # 类对象写入文件为一堆二进制（进行序列化：将BaseMessage转化为字典，在使用json去写入） dicts=[message_to_dict(message) for message in all_message] # 将字典写入文件 with open(self.file_path,"w",encoding="utf-8") as f: json.dump(dicts,f, ensure_ascii=False, indent=2) """ ensure_ascii=False 允许直接写入中文字符 indent=2 格式化输出，便于阅读美化 JSON 结构 """ @property #通过装饰器来变成成员属性 def messages(self) -> list[BaseMessage]: # 读取文件--将dict转换为BaseMessage try: with open(self.file_path,"r",encoding="utf-8") as f: dicts=json.load(f) return messages_from_dict(dicts) except FileNotFoundError: return [] def clear(self) -> None: #文件清空 with open(self.file_path,"w",encoding="utf-8") as f: json.dump([],f, ensure_ascii=False, indent=2)然后仅需修改临时记忆的get_history即可
def get_history(session_id): return FileChatMessageHistory(session_id,"./chat_history")