ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

AI Agent记忆重建:从向量检索到动态生成,突破上下文窗口限制

AI Agent记忆重建:从向量检索到动态生成,突破上下文窗口限制 最近在开发一个基于大模型的智能助手项目时遇到了一个棘手的问题随着对话轮次增加模型经常“忘记”之前讨论过的关键信息比如用户偏好的编程语言、项目背景等。简单地增加上下文窗口长度不仅成本飙升效果也有限。这让我开始深入思考AI Agent的记忆机制难道仅仅是“记住更多”吗一次偶然的机会我在GitHub上看到了一个关于“记忆重建”的开源项目讨论其核心观点“记忆是重建出来的不是回放出来的”让我豁然开朗。这不仅仅是哲学思辨更是当前构建高效、长程AI Agent必须面对的技术现实。本文将围绕这一核心观点结合GitHub上的相关开源项目与前沿讨论为你系统拆解AI Agent记忆机制的原理、主流实现方案并提供一套从理论到实战的完整指南。无论你是正在探索Agent开发的初学者还是寻求优化现有智能体长期记忆的资深开发者都能从中获得可直接落地的思路与代码。1. 背景与核心概念为什么我们需要“重建”记忆在深入技术细节之前我们首先要理解传统“回放式”记忆的局限以及“重建式”记忆为何成为必然。1.1 传统记忆的瓶颈上下文窗口与“回放”困境当前大多数基于大语言模型LLM的AI Agent其记忆依赖于模型的上下文窗口Context Window。你可以将上下文窗口想象成一个固定大小的“短期记忆白板”。新的对话内容被写在白板上当内容写满时最早的信息就会被擦除遗忘。这种方式本质上是“回放”PlaybackAgent通过将历史对话的原始文本或经过简单总结的文本再次输入给模型来“回忆”过去。这带来了几个核心问题容量限制即使上下文窗口扩展到128K甚至更长对于长期、复杂的任务如持续数周的项目协作依然不够。信息冗余与噪声原始对话记录包含大量无关细节全部回放会干扰模型对当前任务的专注。成本高昂处理超长上下文需要巨大的计算和存储开销。缺乏结构化与推理单纯的文本回放无法让Agent主动提取、关联和推理出高层次的用户意图、偏好和事实知识。当你的Agent回答“你之前喜欢用Python”时它可能只是在回放看到过的句子而非真正“理解”了你的编程偏好。1.2 新一代记忆范式重建Reconstruction“记忆是重建出来的”这一观点借鉴了认知科学。人类的记忆并非对过去的精确录像而是基于关键线索和现有认知框架在每次回忆时动态构建的一个新版本。对应到AI Agent“重建式记忆”意味着存储的不是原始对话而是提炼后的“记忆线索”如实体人、物、概念、关系、事件、用户偏好等结构化或半结构化数据。回忆是一个动态生成过程当需要用到记忆时Agent根据当前查询和存储的线索实时地、有针对性地“重建”出最相关的记忆内容而非吐出整段历史。记忆可更新与整合新的信息可以与旧记忆融合修正或强化原有的认知。例如Agent在与你的十次聊天中逐步提取并存储了{“用户偏好”: {“编程语言”: “Python”, “框架”: “FastAPI”}, “当前项目”: “智能客服系统”}这样的结构化信息。当你第11次问“用哪个框架写后端接口比较好”时Agent无需翻阅前十次聊天记录直接根据存储的“偏好”和“项目”线索就能重建出“推荐使用FastAPI因为它与你熟悉的Python栈匹配且适合API开发”的回答。1.3 核心组件记忆Memory与Agent工作流Workflow要实现记忆重建离不开几个关键技术的组合记忆Memory负责信息的存储、提取、更新。可分为短期记忆会话内和长期记忆跨会话。检索增强生成RAG为记忆检索提供了核心技术。将记忆线索存入向量数据库通过语义搜索召回相关片段。工具调用Tool CallingAgent通过调用“记忆存储”、“记忆查询”等工具来管理记忆。工作流Workflow如 LangGraph、LangChain 提供的框架用于编排Agent的推理、行动和记忆管理步骤形成闭环。接下来我们将从环境准备开始搭建一个具备“重建式记忆”能力的AI Agent原型。2. 环境准备与版本说明我们将使用 Python 作为开发语言并依托当前最流行的 LangChain 和 LangGraph 框架来构建Agent。同时会使用 Chroma 作为向量数据库来存储记忆线索。环境要求操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)Python 版本3.10 或 3.11 (推荐 3.11兼容性最佳)包管理工具pip核心依赖库及版本以下版本经过测试能保证较好的兼容性。请务必在虚拟环境中安装。# 创建并激活虚拟环境 (可选但推荐) python -m venv agent_memory_env source agent_memory_env/bin/activate # Linux/macOS # agent_memory_env\Scripts\activate # Windows # 安装核心依赖 pip install langchain0.1.0 pip install langchain-community0.0.10 pip install langgraph0.0.26 pip install chromadb0.4.22 pip install tiktoken0.5.2 # 用于Token计数 pip install python-dotenv1.0.0 # 用于管理API密钥 # 安装大模型接口这里以OpenAI为例你也可以使用其他兼容OpenAI API的模型 pip install openai1.12.0重要提示LangChain 生态版本迭代较快若遇到兼容性问题可尝试锁定上述版本。本文的重点是概念与流程代码逻辑在不同版本间具有参考价值。IDE 选择任何你熟悉的代码编辑器均可如 VS Code、PyCharm。项目结构预览my_agent_project/ ├── .env # 存储API密钥等敏感信息 ├── main.py # 主程序入口 ├── memory_core.py # 记忆系统的核心类定义 ├── agent_graph.py # 基于LangGraph的Agent工作流定义 └── utils.py # 工具函数3. 核心原理与组件拆解在动手编码前我们需要深入理解“重建式记忆”系统的几个核心组件是如何工作的。3.1 记忆的存储从非结构化对话到结构化线索记忆存储的目标是将流水账式的对话转化为易于检索和推理的表示。常见方法有总结提炼Summarization 定期如每5轮对话或按主题让模型对近期对话进行摘要将摘要存入长期记忆。# 伪代码逻辑 def summarize_conversation(conversation_history): prompt f“” 请将以下对话总结成一段简洁的摘要突出关键决策、用户偏好和重要事实。 对话历史{conversation_history} 摘要 “” # 调用LLM生成摘要 summary llm.invoke(prompt) return summary结构化提取Structured Extraction 利用LLM的函数调用或Pydantic输出解析能力从对话中提取预设类别的信息。from pydantic import BaseModel, Field from langchain.output_parsers import PydanticOutputParser class UserPreference(BaseModel): programming_language: str Field(description“用户偏好的编程语言”) tech_stack: list[str] Field(description“用户熟悉的技术栈”) project_interest: str Field(description“用户当前感兴趣的项目类型”) # 定义解析器 parser PydanticOutputParser(pydantic_objectUserPreference) # 构建Prompt让LLM从对话中提取信息 # ... 提取后的UserPreference对象即可作为记忆线索存储向量化嵌入Vector Embedding 将对话片段或总结的文本通过嵌入模型如 text-embedding-3-small转换为向量存入向量数据库。这是实现语义检索的基础。3.2 记忆的检索基于语义的线索召回当Agent需要“回忆”时它面对的查询是当前的问题或情境。我们需要从记忆库中找到最相关的线索。向量检索Vector Search 将当前查询也转换为向量在向量数据库中进行相似度搜索如余弦相似度返回最相似的K条记忆片段。# 伪代码使用Chroma进行检索 from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings embeddings OpenAIEmbeddings(model“text-embedding-3-small”) vectorstore Chroma(embedding_functionembeddings, persist_directory“./mem_db”) # 检索 relevant_memories vectorstore.similarity_search(query“用户喜欢用什么语言编程”, k3)混合检索Hybrid Search 结合向量搜索语义匹配和关键词搜索精确匹配提高召回准确率。一些高级向量数据库如 Weaviate, Qdrant原生支持。3.3 记忆的重建从线索到上下文检索到的记忆线索是碎片化的。重建步骤负责将这些线索结合当前查询整合成一段连贯、有用的背景信息再送入LLM的上下文。def reconstruct_memory(retrieved_clues, current_query): 重建记忆将检索到的线索整合成一段自然的叙述。 clues_text “\n”.join([f“- {clue.page_content}” for clue in retrieved_clues]) reconstruction_prompt f“” 你是一个AI助手正在回忆与当前对话相关的过去信息。 以下是从你记忆中检索到的相关线索 {clues_text} 当前用户的问题是{current_query} 请根据这些线索组织一段简洁、连贯的背景叙述直接回答用户问题或补充上下文。不要提及“根据记忆线索”这类元话语。 重建的记忆背景 “” reconstructed_context llm.invoke(reconstruction_prompt) return reconstructed_context这个过程是“重建”精髓所在最终的记忆文本是即时生成的、针对性的而非原封不动的回放。3.4 Agent工作流集成LangGraph 的角色LangGraph 允许我们以“图”的形式定义Agent的思维和行动流程。记忆的存储、检索、重建可以作为图中的节点Node通过条件边Edge来决定流程走向。一个典型的工作流循环可能是接收用户输入-检索相关记忆-重建记忆上下文-结合上下文思考并决定行动-执行行动或存储新记忆-返回结果。4. 完整实战构建一个具备“重建式记忆”的对话Agent现在我们将把上述理论付诸实践构建一个简单的对话Agent它能记住用户的偏好并在后续对话中运用。4.1 项目初始化与配置首先创建项目目录和文件。mkdir my_memory_agent cd my_memory_agent touch .env main.py memory_core.py agent_graph.py utils.py在.env文件中配置你的 OpenAI API 密钥# .env OPENAI_API_KEYsk-your-openai-api-key-here在utils.py中编写环境变量加载和LLM初始化函数# utils.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI, OpenAIEmbeddings load_dotenv() # 加载 .env 文件中的环境变量 def get_llm(model_name“gpt-3.5-turbo”, temperature0.1): 获取聊天模型实例温度调低使输出更稳定。 return ChatOpenAI(modelmodel_name, temperaturetemperature, api_keyos.getenv(“OPENAI_API_KEY”)) def get_embeddings(model“text-embedding-3-small”): 获取嵌入模型实例。 return OpenAIEmbeddings(modelmodel, api_keyos.getenv(“OPENAI_API_KEY”))4.2 实现核心记忆系统在memory_core.py中我们定义MemorySystem类。# memory_core.py from langchain.vectorstores import Chroma from langchain.schema import Document from langchain.text_splitter import RecursiveCharacterTextSplitter from utils import get_embeddings, get_llm import json class MemorySystem: def __init__(self, persist_directory“./chroma_memory_db”): self.embeddings get_embeddings() self.vectorstore Chroma( embedding_functionself.embeddings, persist_directorypersist_directory ) self.llm get_llm() self.text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) # 用于存储简单键值对记忆如用户偏好 self.key_value_memory {} def store_conversation(self, conversation_text: str, metadata: dict None): 存储一段对话文本到向量数据库。 # 对长文本进行分块 texts self.text_splitter.split_text(conversation_text) docs [Document(page_contenttext, metadatametadata or {}) for text in texts] self.vectorstore.add_documents(docs) print(f“已存储 {len(docs)} 个对话片段到记忆库。”) def extract_and_store_preferences(self, user_input: str, system_observation: str): 从单轮交互中尝试提取用户偏好结构化记忆。 prompt f“” 根据以下用户输入和助理的观察提取或更新用户的长期偏好信息。 用户输入{user_input} 助理观察{system_observation} 请以JSON格式输出只包含以下字段如果无法推断则留空 - “programming_language”: 偏好的编程语言 - “favorite_topic”: 感兴趣的话题 - “project_goal”: 提到的项目目标 示例输出{{“programming_language”: “Python”, “favorite_topic”: “AI Agent”, “project_goal”: “构建一个智能助手”}} JSON: “” response self.llm.invoke(prompt) try: pref json.loads(response.content) # 更新键值对记忆 for key, value in pref.items(): if value: # 只更新非空值 self.key_value_memory[key] value print(f“偏好记忆更新{self.key_value_memory}”) except json.JSONDecodeError: print(“未能解析偏好信息。”) def retrieve_memories(self, query: str, k: int 3): 检索与查询相关的记忆片段。 return self.vectorstore.similarity_search(query, kk) def reconstruct_context(self, query: str, retrieved_docs): 根据检索到的文档重建记忆上下文。 if not retrieved_docs: return “没有相关的过去记忆。” docs_content “\n”.join([doc.page_content for doc in retrieved_docs]) # 加入键值对记忆 kv_memory_str json.dumps(self.key_value_memory, ensure_asciiFalse) if self.key_value_memory else “无” reconstruction_prompt f“” 你正在协助处理一次对话。以下是从历史中检索到的记忆片段以及已知的用户偏好 【历史记忆片段】 {docs_content} 【已知用户偏好】 {kv_memory_str} 请根据以上信息针对当前用户问题“{query}”生成一段简洁、连贯的背景叙述用于帮助助理更好地回答问题。直接叙述事实不要用“根据记忆”这样的开头。 生成的背景叙述 “” context self.llm.invoke(reconstruction_prompt) return context.content def get_memory_for_agent(self, query: str): 供Agent调用的主记忆接口检索并重建。 retrieved self.retrieve_memories(query) context self.reconstruct_context(query, retrieved) return context4.3 定义Agent工作流LangGraph在agent_graph.py中我们定义一个具有记忆-思考-行动循环的Agent。# agent_graph.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage from memory_core import MemorySystem from utils import get_llm # 定义状态结构描述图中每个节点共享的数据 class AgentState(TypedDict): messages: Annotated[List, operator.add] # 对话消息历史 current_query: str # 当前用户问题 memory_context: str # 重建后的记忆背景 response: str # Agent的最终响应 class MemoryAgent: def __init__(self): self.llm get_llm() self.memory_system MemorySystem() self.graph self._create_graph() def _create_graph(self): workflow StateGraph(AgentState) # 定义节点Node workflow.add_node(“process_input”, self._process_input) workflow.add_node(“retrieve_memory”, self._retrieve_memory) workflow.add_node(“generate_response”, self._generate_response) workflow.add_node(“update_memory”, self._update_memory) # 设置入口点 workflow.set_entry_point(“process_input”) # 添加边Edge workflow.add_edge(“process_input”, “retrieve_memory”) workflow.add_edge(“retrieve_memory”, “generate_response”) workflow.add_edge(“generate_response”, “update_memory”) workflow.add_edge(“update_memory”, END) return workflow.compile() def _process_input(self, state: AgentState) - AgentState: 节点1处理输入提取当前查询。 # 从消息历史中取出最新的一条用户消息 last_message state[“messages”][-1] if isinstance(last_message, HumanMessage): state[“current_query”] last_message.content else: state[“current_query”] “” return state def _retrieve_memory(self, state: AgentState) - AgentState: 节点2检索并重建记忆。 query state[“current_query”] memory_context self.memory_system.get_memory_for_agent(query) state[“memory_context”] memory_context print(f“【记忆重建】\n{memory_context}”) # 打印看看重建结果 return state def _generate_response(self, state: AgentState) - AgentState: 节点3结合记忆生成回复。 query state[“current_query”] memory_context state[“memory_context”] messages_history state[“messages”] # 构建给LLM的Prompt注入重建的记忆 prompt f“” 你是一个有帮助的AI助手。以下是你回忆起的与当前对话相关的背景信息 {memory_context} 当前的对话历史如下 {messages_history} 请基于以上背景和历史回答用户的最后一个问题。 用户最后的问题{query} 助手回答 “” response self.llm.invoke(prompt) state[“response”] response.content return state def _update_memory(self, state: AgentState) - AgentState: 节点4根据本轮对话更新记忆系统。 # 将本轮完整的QA作为一个片段存储到向量记忆 last_human_msg state[“messages”][-1].content last_ai_msg state[“response”] conversation_snippet f“用户{last_human_msg}\n助手{last_ai_msg}” self.memory_system.store_conversation(conversation_snippet, metadata{“turn”: len(state[“messages”])}) # 尝试提取用户偏好结构化记忆 self.memory_system.extract_and_store_preferences(last_human_msg, last_ai_msg) # 将助手的回复添加到消息历史以便下一轮使用 state[“messages”].append(AIMessage(contentstate[“response”])) return state def invoke(self, user_input: str, chat_history: list None) - str: 运行Agent的主方法。 if chat_history is None: chat_history [] # 初始化状态 initial_state: AgentState { “messages”: chat_history [HumanMessage(contentuser_input)], “current_query”: “”, “memory_context”: “”, “response”: “” } # 执行图 final_state self.graph.invoke(initial_state) return final_state[“response”]4.4 运行与验证最后在main.py中创建交互循环。# main.py from agent_graph import MemoryAgent def main(): print(“初始化具备‘重建式记忆’的AI Agent...”) agent MemoryAgent() chat_history [] # 用于维护LangChain格式的消息历史 print(“\nAgent已就绪。输入 ‘quit’ 退出对话。”) while True: user_input input(“\n你 “) if user_input.lower() ‘quit’: print(“再见”) break # 调用Agent response agent.invoke(user_input, chat_history) print(f“助手 {response}”) # 更新本地历史Agent内部已更新其状态 # 这里简化处理实际应将HumanMessage和AIMessage都加入chat_history from langchain_core.messages import HumanMessage, AIMessage chat_history.append(HumanMessage(contentuser_input)) chat_history.append(AIMessage(contentresponse)) if __name__ “__main__”: main()4.5 运行示例与结果说明运行程序python main.py开始多轮对话。第一轮对话你 我喜欢用Python编程最近对机器学习很感兴趣。 助手 很高兴知道你偏好Python并对机器学习感兴趣。Python在机器学习领域有非常丰富的生态像TensorFlow、PyTorch、scikit-learn这些库都是基于Python的。你有什么具体的机器学习项目想法吗此时记忆系统会存储这段对话并尝试从中提取结构化偏好{“programming_language”: “Python”, “favorite_topic”: “机器学习”}。第二轮对话相隔几轮后讨论其他话题之后你 我之前说的那个项目用什么语言实现比较好 助手 根据我们之前的交流你提到过偏好使用Python编程并且对机器学习感兴趣。因此对于你的项目尤其是如果它涉及机器学习方面使用Python会是一个非常好的选择因为它有强大的库支持和活跃的社区。关键观察助手并没有直接回放第一轮的原始对话“我喜欢用Python编程...”而是基于重建的记忆背景由记忆系统动态生成给出了回答。如果你查看打印的【记忆重建】日志可能会看到类似“用户偏好使用Python并对机器学习感兴趣”这样的生成文本。这就是“重建”在起作用——它生成了针对当前问题“用什么语言”的最相关背景。5. 常见问题与排查思路在实现和使用此类记忆系统时你可能会遇到以下问题问题现象可能原因排查思路与解决方案记忆检索不相关1. 嵌入模型不适合领域。2. 文本分块策略不合理太大或太小。3. 查询与存储的文本语义不匹配。1. 尝试不同的嵌入模型如text-embedding-3-large或开源模型。2. 调整chunk_size和chunk_overlap对于对话较小的块如200-500字符可能更有效。3. 在检索前用LLM对查询进行重写或扩展使其更接近记忆存储时的表述。记忆重建内容错误或幻觉1. 检索到的线索本身噪声大。2. 重建提示词Prompt设计不佳。3. LLM在生成时偏离线索。1. 提高检索精度增加k值或使用更优的检索器如MutiQueryRetriever。2. 优化重建提示词明确要求“严格基于以下线索”、“不得编造”。3. 使用温度temperature更低的模型或在提示词中加入“如果线索中未提及请回答‘我不知道’”。向量数据库存储失败1. 持久化目录权限问题。2. Chroma 客户端版本与服务端不兼容如果使用客户端/服务器模式。3. 文档嵌入失败。1. 检查persist_directory的读写权限。2. 确保使用的chromadb版本与运行环境兼容。对于简单项目优先使用本地持久化模式。3. 检查网络连接和嵌入模型API是否正常。添加异常处理逻辑。Agent响应速度慢1. 检索和重建步骤增加了延迟。2. 向量数据库未使用索引或规模过大。3. LLM调用耗时。1. 对于简单记忆可引入缓存机制对相同或相似查询缓存重建结果。2. 确保向量数据库建立了有效索引HNSW。对于生产环境考虑专业的向量数据库如 Pinecone, Weaviate。3. 使用响应更快的LLM如 GPT-3.5-Turbo或对记忆查询与响应生成进行异步处理。结构化提取不准1. 提取提示词定义模糊。2. LLM未按要求输出JSON。1. 使用PydanticOutputParser或 OpenAI 的 JSON Mode 来强制结构化输出。2. 在提示词中提供更清晰的示例Few-Shot。6. 最佳实践与工程建议将“重建式记忆”投入实际项目需要考虑更多工程细节。6.1 记忆的粒度与分层不要将所有记忆混为一谈。建议设计分层记忆系统短期记忆/工作记忆保存在上下文窗口内的最近几次交互。用于维持对话连贯性。长期记忆存储到向量数据库或传统数据库中的核心信息。可进一步分为情景记忆具体的对话事件片段本文示例主要实现这种。语义记忆提炼出的通用知识和用户偏好通过结构化提取实现。程序性记忆Agent学会的操作流程或工具使用模式。6.2 记忆的更新与遗忘机制记忆不是只增不减的。更新当新信息与旧记忆冲突时应有机制决定是覆盖、修正还是保留多版本。例如用户说“我现在更喜欢Go了”则应更新programming_language偏好。遗忘/压缩定期对记忆进行总结、去重和重要性排序。可以设定一个时间窗口或容量上限将低频、次要的记忆进行压缩用一条总结性记忆替代多条细节记忆或直接归档。6.3 提示词工程优化记忆系统的效果严重依赖提示词。存储提示词指导LLM如何从对话中提取有价值信息。要明确提取的字段和格式。重建提示词指导LLM如何将线索组织成自然语言。要强调“基于线索”、“简洁”、“针对当前问题”。在系统提示词System Prompt中定义Agent角色明确告诉Agent它拥有记忆能力并应如何利用记忆。例如“你是一个能记住过往对话的助手。在回答时你会参考之前了解到的关于用户的偏好和信息。”6.4 性能与可观测性异步操作记忆的存储尤其是向量化入库可以设计为异步操作不阻塞主响应流程。日志与监控记录关键操作如“存储了哪些记忆”、“检索到了哪些线索”、“重建的背景是什么”。这有助于调试和优化系统。评估设计简单的测试用例检查Agent在涉及历史信息的问题上是否回答正确。这是持续改进的基础。6.5 安全与隐私敏感信息处理记忆系统会存储用户对话。必须考虑数据加密、匿名化以及用户数据的删除机制“被遗忘权”。记忆污染防止恶意输入污染Agent的记忆。例如用户说“我的密码是123456”这类信息不应被存储或应在存储前进行过滤。“记忆是重建出来的”这一理念为我们突破大模型有限上下文窗口、构建真正实用且智能的AI Agent提供了清晰的技术路径。通过本文的讲解和实战你已经掌握了从原理到实现的核心步骤从对话中提取线索、利用向量数据库进行语义检索、动态重建记忆上下文并将其集成到Agent的工作流中。这不仅仅是技术的堆砌更是对Agent认知架构的一次升级。下一步你可以探索更复杂的记忆结构如图记忆、尝试不同的检索策略如混合检索或者将这套系统应用到具体的垂直场景如编程助手、游戏NPC或个性化学习伴侣中。记住一个好的记忆系统应该让Agent变得更“聪明”而非更“笨重”。
返回列表