核心内容摘要
SolidWorks动画进阶：用配合关系实现变速直线运动（2023版技巧）
好的遵照您的要求我将基于随机种子1769652000066所激发的独特视角为您撰写一篇关于 LangChain 记忆 API 的深度技术文章。
本文将超越简单的“记住上一条消息”深入探讨记忆的持久化、检索与结构化并构建一个新颖的“对话式搜索引擎”案例。
超越对话气泡深入探索 LangChain 记忆 API 的持久化、检索与工程化实践在构建基于大语言模型LLM的对话系统时“记忆”是赋予其连续性与智能感的核心组件。
大多数入门教程止步于ConversationBufferMemory—— 一个简单的对话列表缓存。
然而在真实的生产环境或复杂应用中这远远不够。
记忆系统需要解决几个关键问题如何海量持久化如何从冗长历史中精准检索相关片段如何结构化记忆以便于推理本文将深入 LangChain 的记忆MemoryAPI超越基础用法聚焦于记忆的持久化存储、向量化检索以及自定义结构化记忆这三个高阶主题。
我们将通过构建一个“对话式搜索引擎”的案例来演示如何设计一个实用、高效且可扩展的记忆系统。
记忆的本质从状态缓存到知识库在 LangChain 的语境下记忆本质上是一个用于在链Chain或智能体Agent多次调用间持久化状态的机制。
这个状态通常是对话历史但也可以是任何需要跨步骤保留的信息。
1 记忆模块的核心接口所有记忆类都继承自BaseMemory其核心方法是load_memory_variables(inputs: Dict[str, Any]) - Dict[str, Any]: 根据当前输入加载相关的记忆变量如{“history”: “用户之前说: …”}。
save_context(inputs: Dict[str, Any], outputs: Dict[str, Any]) - None: 将本次交互的输入和输出保存到记忆中。
clear() - None: 清空记忆。
理解这个接口至关重要因为它揭示了记忆模块的工作流在链执行前通过load_memory_variables将记忆注入到本次执行的输入中执行后通过save_context将本轮交互保存。
2 常见内存类型速览ConversationBufferMemory: 最简单的列表式缓存会无差别地存储所有对话容易导致提示词Prompt膨胀。
ConversationBufferWindowMemory: 只保留最近 K 轮对话的滑动窗口内存解决了无限增长问题但会丢失早期重要信息。
ConversationSummaryMemory: 使用 LLM 定期
总结历史对话用摘要作为记忆。
节省 token但存在信息失真和 summarization 成本。
ConversationSummaryBufferMemory: 结合了窗口内存和摘要内存保留最近对话的原文更早的则用摘要替代。
这些内置方案对于简单场景有效但面对长周期、多话题、需精准回溯的复杂对话时就显得力不从心。
我们需要更强大的武器。
持久化记忆让对话跨越重启ConversationBufferMemory等内存是进程内的应用重启即消失。
生产环境需要将记忆持久化到数据库。
LangChain 提供了与多种数据库集成的能力。
1 使用Redis持久化对话历史Redis 以其高性能和丰富的数据结构成为存储会话数据的理想选择。
langchain.memory中提供了RedisChatMessageHistory来专门处理消息历史的存储。
import os from langchain.memory import ConversationBufferMemory from langchain_community.chat_message_histories import RedisChatMessageHistory from langchain_openai import ChatOpenAI from langchain.chains import ConversationChain # 配置 Redis 连接。
session_id 是关键用于区分不同会话。
# 这里我们使用用户提供的随机种子作为 session_id 的一部分确保唯一性。
SESSION_ID ftech_discussion_{1769652000066} redis_url redis://localhost:6379/0 #
创建基于 Redis 的消息历史存储 message_history RedisChatMessageHistory( session_idSESSION_ID, urlredis_url, ) #
创建内存并绑定到持久化的消息历史 memory ConversationBufferMemory( chat_memorymessage_history, # 关键指定自定义的 chat_memory memory_keychat_history, return_messagesTrue # 返回 Message 对象列表更适合 ChatModels ) #
在链中使用 llm ChatOpenAI(modelgpt-4, temperature
0.
conversation ConversationChain( llmllm, memorymemory, # 使用我们配置了持久化存储的内存 verboseTrue ) print(f当前会话 ID: {SESSION_ID}) # 第一次运行历史为空 response1 conversation.predict(input我对LangChain的记忆API的持久化功能很感兴趣。
) print(fAI: {response1}) # 即使程序重启只要使用相同的 SESSION_ID历史对话就能恢复。
# 模拟重启后重新连接 print(\n--- 模拟应用重启 ---\n) message_history_restarted RedisChatMessageHistory(session_idSESSION_ID, urlredis_url) memory_restarted ConversationBufferMemory(chat_memorymessage_history_restarted, memory_keychat_history, return_messagesTrue) conversation_restarted ConversationChain(llmllm, memorymemory_restarted, verboseTrue) # 此时直接提问AI 应该能关联之前的上下文 response2 conversation_restarted.predict(input你能否详细解释一下刚才提到的‘绑定’是如何工作的) print(fAI (重启后): {response2})通过RedisChatMessageHistory我们实现了记忆的持久化。
但这里存储的是完整的、线性的对话记录。
当对话轮数成百上千时直接将整个历史塞入 Prompt 是不可行的。
我们需要“检索”而非“全量加载”。
智能检索记忆向量搜索与ConversationKGMemory解决长对话记忆问题的核心思想是并非所有历史都与当前问题相关。
我们只需要检索出最相关的片段。
这自然引向了向量数据库Vector Store和知识图谱Knowledge Graph。
1 使用ConversationVectorStoreMemory进行语义检索ConversationVectorStoreMemory将每一轮对话都转换为向量Embedding并存储到向量数据库如Chroma, FAISS中。
当需要加载记忆时它使用当前输入的向量去数据库中搜索语义最相似的过往对话片段。
from langchain.memory import ConversationVectorStoreMemory from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_community.vectorstores import Chroma from langchain.chains import ConversationChain # 初始化 Embedding 模型和向量库 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore Chroma( collection_namechat_memory_demo, embedding_functionembeddings, persist_directory./chroma_db_memory # 持久化到磁盘 ) # 创建基于向量存储的记忆 vector_memory ConversationVectorStoreMemory( vectorstorevectorstore, memory_keyrelevant_history, input_keyinput, # 搜索时返回的最相关对话片段数 k3, # 返回形式Human/AI 对话对 return_docsTrue ) llm ChatOpenAI(modelgpt-4, temperature
0.
conversation ConversationChain( llmllm, memoryvector_memory, verboseTrue ) # 模拟一段多话题的对话 topics [ Python的asyncio库中如何正确使用gather和wait, 帮我写一个FastAPI的中间件用于记录请求耗时。
, 再回到asyncio如果任务抛出了异常gather会怎么处理 ] for topic in topics: print(f\n[用户]: {topic}) resp conversation.predict(inputtopic) print(f[AI]: {resp[:100]}...) # 截断输出 # 关键验证询问一个需要联系历史中特定部分的问题 query 关于记录请求耗时的中间件你刚才给出的示例中时间单位是什么 print(f\n[用户] (检索测试): {query}) # 此时memory.load_memory_variables 会自动用query去向量库搜索 # 最相关的将是讨论FastAPI的那轮对话而不是最近的asyncio异常处理。
final_resp conversation.predict(inputquery) print(f[AI] (应基于FastAPI历史回答): {final_resp})这种方法的优势在于跨话题的精准关联。
无论目标信息藏在多久以前只要语义相关就能被检索出来。
但它可能丢失对话的时序逻辑性。
2 探索ConversationKGMemory基于知识图谱的结构化记忆对于需要逻辑推理、关系推断的对话知识图谱KG是更好的记忆结构。
ConversationKGMemory会使用 LLM 从对话中提取实体和关系并将其存储为三元组头实体关系尾实体。
from langchain.memory import ConversationKGMemory from langchain_openai import ChatOpenAI from langchain.chains import ConversationChain from langchain.prompts.prompt import PromptTemplate # 定义专门用于知识图谱记忆的提示词模板 kg_prompt PromptTemplate( input_variables[history, input], template你是一个知识图谱提取器。
基于以下对话历史和最新输入识别出新的实体、关系并以三元组形式输出。
如果没有任何新信息输出“无”。
格式示例主题实体, 关系, 对象实体 对话历史: {history} 最新输入: {input} 提取的新知识三元组: ) llm ChatOpenAI(modelgpt-4, temperature
kg_memory ConversationKGMemory( llmllm, # 知识图谱的记忆key memory_keyknowledge_triplets, # 用于提取三元组的提示词 entity_extraction_promptkg_prompt, # 在生成回复时根据当前输入从KG中找出相关实体路径并注入提示词 k2 ) conversation ConversationChain( llmChatOpenAI(modelgpt-4, temperature
0.
, # 可以用一个更有创造力的LLM来生成最终回复 memorykg_memory, verboseTrue ) # 进行一段富有实体和关系的对话 dialogue [ 苏轼是北宋著名的文学家。
, 他的父亲叫苏洵弟弟是苏辙他们三人合称‘三苏’。
, 苏轼的代表作有《念奴娇·赤壁怀古》和《水调歌头·明月几时有》。
] for line in dialogue: resp conversation.predict(inputline) print(f用户: {line}) print(fAI: {resp}\n---) # 查询记忆我们直接访问内存的内部知识图谱 print(\n--- 当前知识图谱内容 ---) graph kg_memory.kg.get_triples() for triple in graph: print(triple) # 现在问一个需要推理的问题 question 苏轼的弟弟有什么文学成就吗 print(f\n用户 (KG推理测试): {question}) # AI在生成回答时load_memory_variables会检索到 (苏轼, 弟弟, 苏辙) 这条边 # 并结合LLM自身的知识来生成关于“苏辙”的回复。
final_answer conversation.predict(inputquestion) print(fAI: {final_answer})ConversationKGMemory将非结构化的对话文本转化为了结构化的知识网络。
这使得系统能够进行简单的关系推理如A的弟弟是B那么B的哥哥就是A并且记忆的表示更加紧凑和语义化。
综合实践构建对话式搜索引擎记忆系统现在我们将结合以上技术构建一个新颖的“对话式搜索引擎”记忆系统。
其需求是持久化所有用户搜索和交互历史必须永久保存。
智能检索用户当前问题可能与其数月前的某个旧查询高度相关系统应能自动找回那段历史上下文。
会话隔离不同用户的会话完全独立。
记忆摘要对于非常长的会话在检索到相关片段后能动态生成一个聚焦于当前问题的精简摘要以节省 Token。
我们将使用Redis存储原始对话消息使用Chroma向量库存储对话的嵌入以实现语义检索并在最终组装 Prompt 前引入一个“动态摘要”步骤。
import hashlib from typing import List, Dict, Any from langchain.schema import BaseMessage, HumanMessage, AIMessage from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.chat_message_histories import RedisChatMessageHistory from langchain_community.vectorstores import Chroma from langchain.memory import ChatMessageHistory from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.runnables import RunnablePassthrough, RunnableLambda class HybridSearchMemorySystem: 一个混合记忆系统结合了
Redis持久化存储
向量检索用于跨历史语义搜索
动态上下文摘要用于压缩过长历史 def __init__(self, session_id: str, redis_url: str, embedding_model, llm): self.session_id session_id self.llm llm #
持久化存储所有原始消息 self.raw_history RedisChatMessageHistory( session_idsession_id, urlredis_url ) #
向量存储用于检索。
为每个会话创建独立的collection。
collection_name fsession_{hashlib.md5(session_id.encode()).hexdigest()[:8]} self.vectorstore Chroma( collection_namecollection_name, embedding_functionembedding_model, persist_directoryf./chroma_hybrid_{collection_name} ) def _format_message_for_index(self, message: BaseMessage) - str: 将消息对象格式化为用于创建向量索引的文本。
role Human if isinstance(message, HumanMessage) else AI return f{role}: {message.content} def save_interaction(self, human_input: str, ai_response: str): 保存一次交互到原始历史和向量库。
# 保存到 Redis self.raw_history.add_user_message(human_input) self.raw_history.add_ai_message(ai_response) # 获取刚添加的两条消息 all_messages self.raw_history.messages last_two all_messages[-2:] # 为这两条消息创建索引文本 texts_to_index [self._format_message_for_index(msg) for msg in last_two] # 元数据记录时间戳和消息类型 metadatas [{type: msg.type, session: self.session_id} for msg in last_two] # 添加到向量库 self.vectorstore.add_texts(textstexts_to_index, metadatasmetadatas) def _retrieve_relevant_memories(self, current_query: str, k: int
- List[str]: 从向量库中检索与当前查询最相关的历史对话片段。
docs self.vectorstore.similarity_search(current_query, kk) return [doc.page_content for doc in docs] def _generate_dynamic_summary(self, query: str, relevant_memories: List[str]) - str: 根据当前查询和相关记忆片段动态生成一个聚焦的摘要。
if not relevant_memories: return 无相关历史对话。
memories_text \n---\n.join(relevant_memories) prompt