如本章前面所述,随着交流长度增加,有效管理对话历史变得有难度。基本的缓冲区记忆最终会超出上下文窗口,而总结式记忆可能会丢失重要细节。向量存储记忆提供了一种有力的替代方案,它将过去的交流以嵌入形式存储在向量数据库中,并在生成新回复时语义化地检索最相关的部分。这种方法使得模型能够从可能非常长的历史中回忆起相关信息,即使这些信息最近没有被提及。在本实战部分,我们将使用FAISS(一个用于高效相似性搜索的常用库)以及OpenAI的嵌入模型,来实现VectorStoreRetrieverMemory。前置条件与设置首先,请确保您已安装所需的库。我们需要langchain,其具体集成包(langchain-openai),一个向量存储实现(faiss-cpu或faiss-gpu),以及用于文本处理的tiktoken。pip install langchain langchain-openai faiss-cpu tiktoken您还需要在环境中配置OpenAI API密钥,通常命名为OPENAI_API_KEY。现在,我们来导入所需的组件:import os from langchain_openai import OpenAIEmbeddings, OpenAI from langchain.memory import VectorStoreRetrieverMemory from langchain.chains import ConversationChain from langchain.vectorstores import FAISS from langchain.prompts import PromptTemplate # 确保您的OPENAI_API_KEY已在环境变量中设置 # 示例:os.environ["OPENAI_API_KEY"] = "您的API密钥" # 检查API密钥是否可用 if not os.getenv("OPENAI_API_KEY"): raise ValueError("OPENAI_API_KEY环境变量未设置。") 使用FAISS实现向量存储记忆核心思想是使用向量存储来保存对话历史。对话的每一轮(输入和输出)都将被嵌入并存储。在生成下一个回复时,我们将使用当前输入来查询向量存储以获取相关的过往交流。**初始化组件:**我们需要一个嵌入模型和一个空的FAISS向量存储。向量存储需要嵌入模型和一个index名称(此处仅作标签)。# 1. 初始化嵌入模型 embedding_model = OpenAIEmbeddings() # 2. 初始化一个空的FAISS向量存储 # 维度取决于嵌入模型(OpenAIEmbeddings使用1536) embedding_size = 1536 index = FAISS.from_texts(["_initial_"], embedding_model, metadatas=[{"hnsw:space": "ip"}]) # 对OpenAI嵌入使用内积空间 *注意:*我们使用一个虚拟文本_initial_来初始化FAISS,因为它不能通过from_texts完全空地初始化。这个初始条目不会对检索产生明显影响。我们在元数据中指定"hnsw:space": "ip"(内积),这通常推荐用于OpenAI嵌入,尽管余弦相似度是默认设置且效果也不错。**创建检索器:**记忆模块不直接与向量存储交互;它使用LangChain的Retriever。我们从FAISS索引创建检索器。search_kwargs={'k': 2}参数指示检索器根据与当前输入的语义相似性,获取前2个最相关的文档(对话片段)。# 3. 创建检索器 # 我们将检索最相关的2个对话片段 retriever = index.as_retriever(search_kwargs=dict(k=2))选择合适的k值很重要。较大的k值会带来更多上下文,但会增加token使用量并可能引入不相关信息的风险。较小的k值更简洁,但可能遗漏有用上下文。通常需要进行实验。**实例化VectorStoreRetrieverMemory:**现在,我们创建记忆对象本身,传入检索器。# 4. 实例化记忆模块 memory = VectorStoreRetrieverMemory(retriever=retriever, memory_key="history")memory_key="history"指定了在提示中保存检索到的上下文的变量名。与对话链结合我们将此记忆功能整合到标准的ConversationChain中。我们需要一个LLM和一个提示模板,其中包含history变量(由我们的记忆模块管理)和input变量(用户当前消息)。# 5. 初始化LLM llm = OpenAI(temperature=0) # 使用确定性设置以保证可预测性 # 6. 定义提示模板 # 注意"{history}"变量,它将由VectorStoreRetrieverMemory填充 _DEFAULT_TEMPLATE = """以下是人类与AI之间的一段友好对话。AI健谈并提供许多来自其上下文的具体细节。如果AI不知道问题的答案,它会如实告知。 过往对话的相关片段: {history} (如果这些信息不相关,您无需使用它们) 当前对话: 人类:{input} AI:""" PROMPT = PromptTemplate( input_variables=["history", "input"], template=_DEFAULT_TEMPLATE ) # 7. 创建对话链 conversation_with_vectorstore_memory = ConversationChain( llm=llm, prompt=PROMPT, memory=memory, verbose=True # 设置为True可查看内部步骤 )运行对话现在,我们来模拟一段对话。请注意,记忆模块如何自动保存输入/输出并为后续轮次检索相关历史。# 第一次交互 response = conversation_with_vectorstore_memory.predict(input="My favorite programming language is Python because it's versatile.") print(response) # 第二次交互 - 不相关 response = conversation_with_vectorstore_memory.predict(input="The weather today is sunny.") print(response) # 第三次交互 - 隐式地回溯到第一次陈述 response = conversation_with_vectorstore_memory.predict(input="Why did I mention I liked Python?") print(response)如果您以verbose=True运行,您将看到第三次交互的类似(简化)输出:> 进入新的ConversationChain链... 格式化后的提示: 以下是人类与AI之间的一段友好对话。... 过往对话的相关片段: 人类:我最喜欢的编程语言是Python,因为它多才多艺。 AI:太棒了!Python确实以其多用性、可读性和丰富的库而闻名。它被用于Web开发、数据科学、AI、脚本等许多方面。 (如果这些信息不相关,您无需使用它们) 当前对话: 人类:我为什么说我喜欢Python? AI: > 链完成。 您提到您喜欢Python是因为它的多用性。请注意,VectorStoreRetrieverMemory如何根据第三个输入(“我为什么说我喜欢Python?”)的语义内容检索到第一次交互,从而填充了“过往对话的相关片段:”部分。第二次关于天气的不相关交互很可能没有被检索到(或排名较低),因为它在语义上不相似。向量存储记忆的工作原理:检索流程在链中使用VectorStoreRetrieverMemory时的过程可以如下所示:digraph G { rankdir=LR; node [shape=box, style=rounded, fontname="Arial", fontsize=10, margin="0.2,0.1"]; edge [fontname="Arial", fontsize=9]; UserInput [label="用户输入"]; SaveMemory [label="保存输入/输出\n(嵌入并存储到向量数据库)"]; RetrieveMemory [label="检索相关历史\n(使用当前输入查询向量数据库)"]; FormatPrompt [label="格式化提示\n(注入检索到的历史)"]; LLM [label="LLM调用"]; Output [label="生成回复"]; FinalOutput [label="最终输出", shape=ellipse]; UserInput -> SaveMemory [label=" 生成回复后"]; UserInput -> RetrieveMemory [label=" 提示格式化前"]; RetrieveMemory -> FormatPrompt; UserInput -> FormatPrompt; FormatPrompt -> LLM; LLM -> Output; Output -> SaveMemory; Output -> FinalOutput; }流程图:说明在对话链中使用向量存储记忆的步骤。用户输入在提示格式化之前触发检索,而输入/输出对在回复生成后保存。调优与持久化检索参数(k):as_retriever(search_kwargs=dict(k=k))中的k值是一个主要的调优参数。增加k值会提供更多上下文,但会增加提示大小和成本。减少k值可以节省token,但可能遗漏相关信息。您还可以尝试其他search_type选项,例如"mmr"(最大边际相关性),以平衡检索文档的相关性和多样性。**持久化:**我们示例中的FAISS索引是内存中的,脚本结束后会丢失。对于生产用途,您通常会希望实现持久化。您可以将FAISS索引保存到本地并从本地加载:# 保存索引 index.save_local("my_faiss_index") # 稍后加载索引(需要嵌入模型) loaded_index = FAISS.load_local("my_faiss_index", embedding_model, allow_dangerous_deserialization=True) retriever = loaded_index.as_retriever(search_kwargs=dict(k=2)) memory = VectorStoreRetrieverMemory(retriever=retriever, memory_key="history") # ... 使用此记忆重新创建链*安全注意:*如果索引文件来自不可信来源,加载用save_local保存的FAISS索引可能存在安全风险,因此需要allow_dangerous_deserialization=True标志。对于与可能不可信数据交互的生产系统,请考虑更安全的序列化方法或托管向量数据库服务。或者,使用前面讨论过的基于云的向量存储(Pinecone、Weaviate等),它们会自动处理持久化和扩展。完整示例脚本以下是结合了所有步骤的完整脚本:import os from langchain_openai import OpenAIEmbeddings, OpenAI from langchain.memory import VectorStoreRetrieverMemory from langchain.chains import ConversationChain from langchain.vectorstores import FAISS from langchain.prompts import PromptTemplate # 确保您的OPENAI_API_KEY已设置 if not os.getenv("OPENAI_API_KEY"): raise ValueError("OPENAI_API_KEY environment variable not set.") # 1. 初始化嵌入 embedding_model = OpenAIEmbeddings() # 2. 初始化FAISS向量存储 # 使用一个小技巧通过from_texts初始化,因为它需要一些文本 try: # 尝试加载(如果存在) index = FAISS.load_local("my_faiss_index", embedding_model, allow_dangerous_deserialization=True) print("已加载现有FAISS索引。") except Exception: print("正在创建新的FAISS索引。") # embedding_size = 1536 # 通常从嵌入中推断 index = FAISS.from_texts(["_initial_"], embedding_model, metadatas=[{"hnsw:space": "ip"}]) # 3. 创建检索器(检索最相关的2个片段) retriever = index.as_retriever(search_kwargs=dict(k=2)) # 4. 实例化记忆 memory = VectorStoreRetrieverMemory(retriever=retriever, memory_key="history") # 5. 初始化LLM llm = OpenAI(temperature=0) # 6. 定义提示模板 _DEFAULT_TEMPLATE = """以下是人类与AI之间的一段友好对话。AI健谈并提供许多来自其上下文的具体细节。如果AI不知道问题的答案,它会如实告知。 过往对话的相关片段: {history} (如果这些信息不相关,您无需使用它们) 当前对话: 人类:{input} AI:""" PROMPT = PromptTemplate( input_variables=["history", "input"], template=_DEFAULT_TEMPLATE ) # 7. 创建对话链 conversation_with_vectorstore_memory = ConversationChain( llm=llm, prompt=PROMPT, memory=memory, verbose=False # 设置为True可查看详细日志 ) # --- 运行对话 --- print("开始对话(输入'quit'退出):") while True: user_input = input("人类:") if user_input.lower() == 'quit': break response = conversation_with_vectorstore_memory.predict(input=user_input) print(f"AI:{response}") # --- 退出前保存索引 --- try: index.save_local("my_faiss_index") print("已保存FAISS索引。") except Exception as e: print(f"保存FAISS索引时出错:{e}") print("对话结束。") 注意事项**嵌入成本:**每个保存的输入/输出对都会产生生成其嵌入的成本。这在长时间对话中可能会累积。**检索相关性:**记忆的质量很大程度上取决于检索步骤的有效性。糟糕的检索(由于k值不佳、嵌入模型弱或历史记录嘈杂)会导致不相关的上下文被提供给LLM。重新排序或查询转换等方法(在第4章中讨论)有时会有帮助。**上下文大小:**虽然向量存储记忆有助于选择相关历史,但检索到的片段仍需与当前输入和提示指令一起,适应LLM的上下文窗口。本次实战演示了如何实现VectorStoreRetrieverMemory,为会话应用中保持长期、语义相关的上下文提供了一种有力机制。通过将历史存储在向量存储中,您可以克服简单缓冲区的限制,并实现在长时间内更连贯、更丰富的交互。请记住根据您的应用需求调整检索参数并考虑持久化策略。