langchain学习--会话记忆

临时记忆理解LangChain的临时记忆机制LangChain的临时记忆指模型在单次对话或短时间内保留上下文信息的能力通常通过缓存或短期存储实现。其核心是维护对话状态确保在多轮交互中连贯响应。实现临时记忆的关键方法对话历史缓存将用户与模型的交互记录存储在内存或轻量级数据库如Redis中每次请求时加载最近的N条记录。例如from collections import deque chat_history deque(maxlen5) # 保留最近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(modelqwen-plus) def print_prompt(full_prompt): print(*20,full_prompt,*20) return full_prompt # 创建一个提示词 promptChatPromptTemplate.from_messages( [ (system, 你需要根据用户提供给你的对话历史来回答问题。对话历史{chat_history}用户输入{input}请给出简单回答), MessagesPlaceholder(chat_history), (human, {input}) ] ) #MessagesPlaceholder 是 LangChain 中用于在 ChatPromptTemplate 中动态插入消息列表的占位符。 # 创建一个基础运行链 base_chainprompt|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_chainRunnableWithMessageHistory( base_chain,#被附加历史消息的Runnable通常是chain get_history,#获取历史消息的函数 input_messages_keyinput,#输入消息的key history_messages_keychat_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, callbacksLangChain框架自己使用configurable内部开发者自定义的参数传递给具体的组件如get_history函数长期会话记忆--文件存储需要自己编写相关类该类需要继承BaseChatMessageHistory示例如下class FileChatMessageHistory(BaseChatMessageHistory): def __init__(self,session_id,storage_path): self.session_idsession_id # 会话ID self.storage_pathstorage_path # 不同会话id储存的文件夹路径 self.file_pathos.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_messagelist(self.messages) # 获取当前所有消息 all_message.extend(messages) # 添加新消息 # 将数据同步写入文件中 # 类对象写入文件为一堆二进制进行序列化将BaseMessage转化为字典在使用json去写入 dicts[message_to_dict(message) for message in all_message] # 将字典写入文件 with open(self.file_path,w,encodingutf-8) as f: json.dump(dicts,f, ensure_asciiFalse, indent2) ensure_asciiFalse 允许直接写入中文字符 indent2 格式化输出便于阅读美化 JSON 结构 property #通过装饰器来变成成员属性 def messages(self) - list[BaseMessage]: # 读取文件--将dict转换为BaseMessage try: with open(self.file_path,r,encodingutf-8) as f: dictsjson.load(f) return messages_from_dict(dicts) except FileNotFoundError: return [] def clear(self) - None: #文件清空 with open(self.file_path,w,encodingutf-8) as f: json.dump([],f, ensure_asciiFalse, indent2)然后仅需修改临时记忆的get_history即可def get_history(session_id): return FileChatMessageHistory(session_id,./chat_history)