你有没有遇到过这样的场景刚跟一个AI助手聊得挺好让它帮你分析一份文档、规划一个项目或者写一段代码。聊到一半你出去接了个电话或者刷新了一下页面回来再问它“刚才我们说到哪了”它却一脸茫然仿佛失忆了一般。或者更常见的是在处理一个长文档时你明确告诉它“请参考第三段的数据”但它给出的回答却完全忽略了上下文只基于最后几句话进行判断。这不是AI笨而是当前绝大多数AI应用尤其是基于大语言模型LLM的Agent普遍存在的一个核心痛点上下文丢失。它们就像金鱼只有7秒的记忆对话一长任务一复杂之前的关键信息就“蒸发”了。网上充斥着各种“企业级Agent记忆系统”的教程和框架很多听起来高大上但要么是云里雾里的理论要么是封装得严严实实的黑盒你只知道调用一个save_memory()的API却不知道数据到底存哪了、怎么存的、为什么有时候存了也没用。更让人头疼的是当你真正想把一个能记住事的Agent应用到自己的业务里——比如一个能记住用户偏好的客服机器人一个能追踪项目全生命周期的协作助手——你会发现从“玩具Demo”到“可用系统”之间隔着一道名为“记忆工程化”的鸿沟。今天我们不谈那些遥不可及的“下一代架构”就从最实际的问题出发一个真正能用的Agent记忆系统到底该怎么从零开始搭建我们将彻底拆解“长短记忆”的核心原理绕过那些华而不实的抽象概念手把手完成一套可落地的代码实现。你会发现解决上下文丢失的关键不在于用了多牛的算法而在于你是否理解数据在Agent内部的流转逻辑并为之设计了一套简洁、可靠、可扩展的存储与检索机制。1. 先别被“记忆”这个词唬住它本质是数据的存与取很多人一听到“Agent记忆系统”脑海里立刻浮现出科幻电影里具有人格化记忆的AI。但在工程实践里我们需要立刻祛魅Agent的“记忆”在99%的场景下指的就是“将对话历史、工具调用结果、用户偏好等结构化或非结构化数据进行持久化存储并在后续需要时高效、准确地检索出来”。它的核心挑战可以归结为两个存什么不是所有对话流水账都值得存。存太多垃圾信息淹没有效信息存太少关键上下文丢失。怎么取当Agent需要“回忆”时如何从海量存储中快速找到最相关的几条信息这比数据库的精确查询要难因为用户的问题可能是模糊的、概括性的。目前主流的解决方案是借鉴人类记忆的二分法短期记忆Short-term Memory和长期记忆Long-term Memory。短期记忆可以理解为Agent当前的“工作台”或“思考缓存”。它容量有限但存取速度极快通常就是LLM模型本身的上下文窗口Context Window。例如你最近说的10句话模型能直接“看到”并用于生成回答。它的“丢失”是物理性的一旦上下文窗口滑动移出模型就无法再直接感知到那些信息。长期记忆这就是我们要构建的“外部记忆体”。它是一个独立于模型上下文之外的存储系统数据库、向量库、文件等容量理论上无限但检索需要时间和计算资源。它的目标是保存那些跨越单次会话、对长期交互有价值的信息。很多教程的问题在于只教你怎么搭建一个孤立的“长期记忆”数据库却没有讲清楚短期记忆如何与长期记忆协同工作。一个健壮的记忆系统必须有一套清晰的策略来决定何时将短期记忆中的重要内容“固化”到长期记忆当Agent需要时又如何从长期记忆中“提取”最相关的片段并精准地“加载”回短期记忆即模型的当前上下文中接下来我们就围绕“存-取-用”这个核心流程开始我们的实战。2. 搭建记忆系统的核心三要素存储、编码与检索在写第一行代码之前我们必须为记忆系统选定三个核心组件这决定了系统的能力和复杂度上限。2.1 存储后端记忆放在哪里这是最基础的一层。根据数据形式和访问模式我们有几种选择存储类型典型代表适合存储的数据优点缺点适用场景结构化存储SQLite, PostgreSQL, MySQL用户ID、会话ID、时间戳、明确的键值对如用户偏好{“theme”: “dark”}、工具调用记录。精确查询快事务支持好结构清晰。无法直接处理模糊的自然语言查询。存储元数据、会话列表、用户属性、确定的配置信息。向量数据库Chroma, Pinecone, Weaviate, Qdrant对话文本、文档片段、总结性内容等需要语义搜索的信息。支持相似度检索能理解语义找到“意思相近”的记忆。需要额外的嵌入Embedding步骤有精度损失和延迟。存储非结构化的对话历史、知识片段用于基于内容的回忆。文件系统本地文件对象存储S3完整的对话日志、生成的文档、图像等大型二进制数据。简单容量大适合存“原始档案”。检索效率低需要自己建立索引。归档完整会话存储最终产出物。实战建议对于大多数入门和中级场景一个“SQLite Chroma”的组合就非常强大。SQLite负责存结构化元数据谁、什么时候、什么会话Chroma负责存文本记忆片段并提供语义搜索能力。两者都非常轻量无需复杂部署。2.2 编码器如何把文本变成可存储、可搜索的形式这是长期记忆的“魔法”所在。我们不能直接把一大段文本扔进数据库那样无法检索。我们需要一个编码器Embedding Model将文本转换为一个高维空间中的向量一组数字。语义相近的文本其向量在空间中的距离也更近。选择编码器你可以使用OpenAI的text-embedding-ada-002或者开源模型如BGE-M3、text2vec。对于本地部署all-MiniLM-L6-v2是一个在速度和效果上平衡得很好的选择。编码什么不是所有文本都值得编码。通常我们会提取对话中的“用户消息”、“AI回复”或者两者的组合有时还会加上时间戳和会话ID作为元数据一起存储。2.3 检索器如何找到最相关的记忆当Agent需要“回忆”时例如用户问“我之前提到过的那本书叫什么”检索器开始工作。查询编码首先将用户的当前问题或AI需要回忆的线索用同样的编码器转换为一个查询向量。相似度搜索在向量数据库中计算查询向量与所有存储记忆向量之间的相似度常用余弦相似度。返回Top-K返回相似度最高的K条记忆片段例如最相关的3条。这里有一个关键陷阱单纯的相似度搜索可能返回一堆语义相关但时间错乱或来源混杂的信息。因此高级的记忆系统会在检索时加入过滤器Filter比如只检索某个用户的记忆或者某个特定会话中的记忆或者最近一周的记忆。这就是为什么我们需要SQLite来存储元数据——它可以帮助我们做高效的预过滤。理解了这三个要素我们就可以开始设计系统的数据流了。3. 从零手撸一个简易可用的记忆系统我们用一个具体的例子来贯穿始终构建一个“学习伙伴”Agent它能记住你和它讨论过的所有技术概念并在后续对话中主动关联或应要求回忆。我们将使用Python主要依赖langchain用于简化流程和chromadb向量数据库。当然我们会解释每一步背后的原理让你即使不用langchain也能自己实现。3.1 环境准备与数据模型定义首先安装基础包pip install langchain langchain-community chromadb sentence-transformers我们定义两个核心的数据模型对应两种记忆# memory_models.py from pydantic import BaseModel from datetime import datetime from typing import Optional, Dict, Any class ShortTermMemory(BaseModel): 短期记忆即当前的对话上下文列表 messages: list[Dict[str, Any]] # 通常格式: [{role: user, content: ...}, ...] class LongTermMemoryItem(BaseModel): 长期记忆中的一个条目 id: str user_id: str # 用户标识 session_id: str # 会话标识 content: str # 需要被记住的文本内容 embedding: Optional[list[float]] None # 内容的向量表示 metadata: Dict[str, Any] # 其他元数据如来源消息类型、时间戳、实体标签等 created_at: datetime class Config: arbitrary_types_allowed True3.2 构建记忆存储库Memory Store这是系统的中枢负责所有存储和检索逻辑。# memory_store.py import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer from typing import List, Dict, Any, Optional import uuid from datetime import datetime from .memory_models import LongTermMemoryItem class AgentMemoryStore: def __init__(self, persist_directory: str ./chroma_db): # 初始化向量数据库客户端 self.chroma_client chromadb.PersistentClient( pathpersist_directory, settingsSettings(anonymized_telemetryFalse) ) # 获取或创建一个集合Collection相当于一个命名空间 self.collection self.chroma_client.get_or_create_collection( nameagent_memories, metadata{hnsw:space: cosine} # 使用余弦相似度 ) # 初始化本地嵌入模型轻量级 self.embedder SentenceTransformer(all-MiniLM-L6-v2) def _generate_embedding(self, text: str) - List[float]: 生成文本的向量嵌入 return self.embedder.encode(text).tolist() def save_memory( self, user_id: str, session_id: str, content: str, metadata: Optional[Dict[str, Any]] None ) - str: 将一段内容保存为长期记忆 memory_id str(uuid.uuid4()) embedding self._generate_embedding(content) # 准备元数据 full_metadata { user_id: user_id, session_id: session_id, created_at: datetime.now().isoformat(), **(metadata or {}) } # 存入向量数据库 self.collection.add( documents[content], embeddings[embedding], metadatas[full_metadata], ids[memory_id] ) print(f[Memory Saved] ID: {memory_id}, Content: {content[:50]}...) return memory_id def search_memories( self, query: str, user_id: Optional[str] None, session_id: Optional[str] None, limit: int 3 ) - List[LongTermMemoryItem]: 根据查询文本检索相关记忆可过滤用户和会话 query_embedding self._generate_embedding(query) # 构建过滤条件 where_filter {} if user_id: where_filter[user_id] user_id if session_id: where_filter[session_id] session_id # 执行相似度搜索 results self.collection.query( query_embeddings[query_embedding], n_resultslimit, wherewhere_filter # 应用过滤器 ) memories [] # results 结构: {ids: [[...]], documents: [[...]], metadatas: [[...]], ...} if results[ids][0]: for i in range(len(results[ids][0])): mem_id results[ids][0][i] doc results[documents][0][i] meta results[metadatas][0][i] memories.append( LongTermMemoryItem( idmem_id, user_idmeta.get(user_id, ), session_idmeta.get(session_id, ), contentdoc, embeddingresults[embeddings][0][i] if results[embeddings] else None, metadatameta, created_atdatetime.fromisoformat(meta.get(created_at)) ) ) return memories def get_conversation_context(self, user_id: str, session_id: str, recent_turns: int 10) - List[Dict]: 获取最近N轮对话作为短期记忆简化版实际可能从消息队列或DB获取 # 这里为简化我们模拟从向量库中按时间获取最近的消息 # 实际生产环境短期记忆可能直接保存在内存或Redis中 all_memories self.search_memories( query, # 空查询通过过滤器获取所有 user_iduser_id, session_idsession_id, limitrecent_turns * 2 # 假设一轮对话产生两条记忆 ) # 按时间排序并返回最近的内容 sorted_memories sorted(all_memories, keylambda x: x.created_at, reverseTrue) context_messages [] for mem in sorted_memories[:recent_turns]: # 根据元数据判断是用户还是AI说的话 role mem.metadata.get(role, user) context_messages.append({role: role, content: mem.content}) return context_messages[::-1] # 反转让最早的在最前3.3 设计记忆策略何时存何时取有了存储库我们需要定义记忆系统的“大脑”——策略。这是区分玩具和工具的关键。# memory_strategy.py from typing import List, Dict, Any from .memory_store import AgentMemoryStore class MemoryManagementStrategy: def __init__(self, memory_store: AgentMemoryStore): self.store memory_store def decide_what_to_save(self, message: Dict[str, Any]) - bool: 决定当前这条消息是否值得存入长期记忆。 这是一个策略核心你可以根据业务定制。 content message.get(content, ) role message.get(role, ) # 策略1只保存用户的关键输入和AI的重要结论 if role user: # 用户消息如果较长或包含特定关键词如“记住”、“重要”则保存 if len(content) 50 or any(keyword in content.lower() for keyword in [记住, 重要, note, important]): return True elif role assistant: # AI消息如果包含总结、定义、步骤通常有标题或编号则保存 if 总结来说 in content or 定义为 in content or 步骤如下 in content: return True # 策略2也可以基于更复杂的NLP模型来判断信息密度 return False def format_memory_content(self, message: Dict[str, Any]) - str: 格式化要存储的内容可以提炼、总结而不是存原始消息 raw_content message.get(content, ) role message.get(role, ) # 简单示例给内容加个前缀标签 return f[{role.upper()}] {raw_content} def retrieve_relevant_context( self, current_query: str, user_id: str, session_id: str, short_term_memory: List[Dict] ) - Dict[str, Any]: 为当前查询组装完整的上下文。 返回一个字典包含短期记忆 从长期记忆中检索到的相关记忆。 # 1. 短期记忆最近对话直接使用 final_context_messages short_term_memory.copy() # 2. 从长期记忆中检索与当前查询最相关的部分 relevant_long_term_memories self.store.search_memories( querycurrent_query, user_iduser_id, session_idsession_id, limit2 # 不宜过多避免上下文爆炸 ) # 3. 如果检索到了长期记忆将它们作为“系统提示”或“参考信息”插入上下文 if relevant_long_term_memories: memory_text \n.join([f- {mem.content} for mem in relevant_long_term_memories]) # 将回忆起的记忆放在系统提示中或作为一条特殊消息 memory_prompt { role: system, content: f以下是你在本次或历史对话中提及的相关信息请参考\n{memory_text} } # 插入到上下文的最前面在系统提示之后在最近对话之前 final_context_messages.insert(0, memory_prompt) return { enhanced_messages: final_context_messages, retrieved_memories: relevant_long_term_memories }3.4 将记忆系统接入Agent工作流最后我们创建一个简单的Agent类将记忆系统融入对话循环。# agent_with_memory.py from .memory_store import AgentMemoryStore from .memory_strategy import MemoryManagementStrategy from typing import Dict, Any import json class ConversationalAgent: def __init__(self, user_id: str default_user): self.user_id user_id self.session_id fsession_{user_id}_{hash(str(id(self)))} # 简化生成会话ID self.memory_store AgentMemoryStore() self.memory_strategy MemoryManagementStrategy(self.memory_store) # 模拟的短期记忆缓冲区实际可能用更复杂的数据结构 self.short_term_buffer [] self.max_short_term_turns 6 # 短期记忆保留最近3轮对话每轮userassistant def process_user_input(self, user_input: str) - str: 处理用户输入的核心流程 # 1. 构建当前短期记忆最近几轮 recent_context self.memory_store.get_conversation_context( self.user_id, self.session_id, recent_turnsself.max_short_term_turns ) # 合并内存中的短期记忆和当前缓冲区如果有 self.short_term_buffer recent_context self.short_term_buffer[-4:] # 保持缓冲区不过大 current_short_term self.short_term_buffer[-self.max_short_term_turns:] # 2. 检索长期记忆增强上下文 enhanced_context self.memory_strategy.retrieve_relevant_context( current_queryuser_input, user_idself.user_id, session_idself.session_id, short_term_memorycurrent_short_term ) full_messages enhanced_context[enhanced_messages] # 加入最新的用户输入 full_messages.append({role: user, content: user_input}) # 3. 调用LLM生成回复这里用模拟函数代替真实API调用 ai_response self._call_llm_simulate(full_messages) # 4. 决定是否将本轮对话存入长期记忆 user_message_for_memory {role: user, content: user_input} ai_message_for_memory {role: assistant, content: ai_response} if self.memory_strategy.decide_what_to_save(user_message_for_memory): self.memory_store.save_memory( user_idself.user_id, session_idself.session_id, contentself.memory_strategy.format_memory_content(user_message_for_memory), metadata{role: user, turn: len(self.short_term_buffer)//2} ) if self.memory_strategy.decide_what_to_save(ai_message_for_memory): self.memory_store.save_memory( user_idself.user_id, session_idself.session_id, contentself.memory_strategy.format_memory_content(ai_message_for_memory), metadata{role: assistant, turn: len(self.short_term_buffer)//2} ) # 5. 更新短期记忆缓冲区 self.short_term_buffer.append(user_message_for_memory) self.short_term_buffer.append(ai_message_for_memory) return ai_response def _call_llm_simulate(self, messages: List[Dict]) - str: 模拟LLM调用实际应替换为OpenAI、DeepSeek等API调用 # 这里简单模拟一个能“看到”上下文的回复 context_str json.dumps(messages, ensure_asciiFalse, indent2) # 模拟响应在实际中这里会是 model.chat(messages) 的调用 return f[模拟AI响应] 我已收到你的消息。当前上下文中有 {len(messages)} 条消息其中包含长期记忆检索结果。3.5 运行一个完整的对话示例# main.py from agent_with_memory import ConversationalAgent def main(): agent ConversationalAgent(user_idalice) # 第一轮对话教授Agent一个概念 print(用户: 什么是RAG) response1 agent.process_user_input(什么是RAG) print(fAI: {response1}) # 模拟AI给出了一个定义我们的策略可能会将这个定义存入长期记忆 # 用户输入一个需要记忆的陈述 print(\n用户: 记住RAG的核心是检索外部知识并增强生成。) response2 agent.process_user_input(记住RAG的核心是检索外部知识并增强生成。) print(fAI: {response2}) # 这条包含“记住”关键词会被策略判定为需要保存 # 过几轮其他对话后... agent.process_user_input(今天的天气真好。) agent.process_user_input(我们来聊聊机器学习。) # 关键测试询问一个之前教过的概念 print(\n用户: 我之前跟你说的RAG它的核心是什么来着) response3 agent.process_user_input(我之前跟你说的RAG它的核心是什么来着) print(fAI: {response3}) # 理想情况下AI的响应应该能体现出它“回忆”起了长期记忆中的内容。 # 因为“RAG的核心”这个查询会触发向量检索找到之前存储的那条记忆。 if __name__ __main__: main()运行这个示例你会看到系统如何工作当用户说“记住RAG的核心是...”时这条信息被编码并存入向量数据库。当用户后来模糊地问“RAG的核心是什么来着”时系统会将这个问题编码成查询向量在数据库中搜索到最相关的记忆片段并将其作为上下文的一部分提供给LLM从而生成一个基于记忆的准确回答。4. 从“能跑通”到“能用”必须解决的工程化问题上面的代码是一个高度简化的原型它演示了核心流程。但要投入实际使用无论是作为个人项目还是企业级应用你必须考虑以下问题4.1 记忆的粒度与摘要问题存每一条消息那数据库会爆炸且噪音极大。存整个会话摘要又会丢失细节。解决方案采用分层记忆。对话轮次记忆对于重要的Q-A对直接存储如我们上面的策略。会话摘要记忆在一个会话结束时或达到一定轮次用LLM生成一个本次会话的摘要例如“用户Alice在本轮会话中主要探讨了RAG技术明确了其核心是检索与生成的结合并讨论了向量数据库的选择。”然后将这个摘要也存入向量库。当用户问“我们上次聊了什么”时优先检索会话摘要。实体/事实记忆使用NER命名实体识别或LLM提取对话中出现的具体实体如“项目Alpha”、“客户张三”、“Python 3.11”、事实如“部署环境要求Ubuntu 22.04”和结论将它们作为结构化或半结构化的记忆点单独存储便于精确查询。4.2 记忆的更新、合并与遗忘问题用户说“我的喜好是咖啡”过一会儿又说“其实我更喜欢茶”。系统是存储两条矛盾记忆还是更新前一条解决方案版本化为同一主题的记忆保留多个版本并附加时间戳和置信度。冲突检测与解决在保存新记忆时检索是否有高度相关的旧记忆。如果存在可以触发一个解决流程例如询问用户“你之前说过喜欢咖啡现在改为茶了吗”或者根据时间戳自动覆盖。记忆衰减为记忆设置“强度”或“新鲜度”字段随着时间推移或未被访问而衰减。定期清理强度低于阈值的记忆。4.3 检索质量与上下文管理问题检索出的3条记忆可能彼此冗余或者与当前短期记忆高度重复挤占了宝贵的上下文窗口。解决方案检索后重排序Rerank使用更精细的交叉编码器Cross-Encoder对初步检索结果进行重排序选出最相关、最不冗余的几条。记忆去重在注入上下文前对检索到的记忆和现有短期记忆进行去重。动态上下文窗口根据查询复杂度动态决定检索多少条记忆以及为它们分配多少上下文令牌Token。4.4 性能、扩展性与安全性能向量检索在大规模数据下可能变慢。需要考虑缓存热点记忆、对记忆进行聚类索引、使用更高效的向量数据库。扩展性当用户量、对话量激增时记忆存储需要支持分库分表、分布式向量检索。安全与隐私这是企业级应用的生命线。数据隔离必须严格保证用户A无法检索到用户B的记忆。我们的代码中通过user_id过滤实现了这一点。敏感信息过滤在记忆保存前应有过滤层防止密码、密钥、个人身份信息PII被存入。记忆删除权必须提供用户查看和删除其个人记忆的接口以满足数据合规要求如GDPR。5. 总结记忆系统的价值不在“记住”而在“用对”搭建一个Agent记忆系统技术实现只是第一步。真正的价值来自于你如何定义“什么值得记忆”以及“如何利用记忆”。它不是一个炫技的组件而是一个需要精心设计策略的业务逻辑层。对于大多数项目我的建议是不要一开始就追求一个完美、复杂的记忆系统。应该遵循以下路径从最简单的键值对开始先实现“记住用户的名字”、“记住上次选择的语言”。这能立刻带来体验提升。引入向量检索解决模糊查询当需要“根据聊天内容找相关历史”时引入我们上面实现的简易向量记忆库。定制你的记忆策略根据你的Agent的专属领域是客服编程助手创意伙伴设计decide_what_to_save函数。一个客服机器人可能需要记住用户的投诉历史和产品型号一个编程助手则需要记住项目结构和技术栈。关注工程化与可靠性在策略稳定后再着手解决数据持久化、检索性能、缓存、监控和清理机制。最终一个优秀的记忆系统会让你的Agent从“每次对话都是初见”的陌生人变成“了解你过去服务你当下”的合作伙伴。这个转变才是Agent技术从演示走向实用的关键一步。现在你可以基于上面的框架和代码开始为你自己的Agent赋予“记忆”了。先从一个小而具体的记忆功能做起观察它如何改变交互体验然后再逐步迭代。