1. 先搞清楚企业级 Agent 记忆系统到底要解决什么如果你正在开发或使用 AI Agent大概率遇到过这个问题对话进行到一半Agent 好像“失忆”了不记得之前提过的关键信息或者处理长文档、多轮复杂任务时上下文窗口一满前面的内容就被无情丢弃。这本质上是上下文丢失是当前基于大语言模型的 Agent 在走向实用化过程中最核心的痛点之一。所谓的“企业级 Agent 记忆系统”目标就是解决这个痛点。它不是一个单一的模型或工具而是一套工程化的记忆管理策略和存储架构。核心是把 Agent 运行过程中产生的海量、杂乱的交互信息用户指令、中间思考、工具调用结果、最终输出等进行有效的筛选、压缩、存储和高效检索让 Agent 在需要时能“回想”起相关的历史从而做出更连贯、更准确的决策。和网上很多只讲概念的文章不同我们这次要拆解的是能代码落地的方案。你会看到一个实用的记忆系统关键在于区分短期记忆和长期记忆并设计好它们之间的流转机制。短期记忆就像工作内存处理当前会话的即时信息长期记忆则像知识库或数据库存储经过提炼的、未来可能用到的关键知识。很多项目失败不是因为模型不够强而是记忆流转的管道没设计好。所以这篇文章适合两类人一是正在为自家 Agent 健忘而头疼的开发者二是想深入理解 Agent 架构、超越简单 Prompt 工程的学习者。我们不会空谈理论而是会从原理拆解开始一步步走到可运行的代码告诉你每个设计选择背后的“为什么”以及在实际部署时最容易踩的坑。2. 拆解长短记忆的核心原理与设计选择在动手写代码之前必须把原理吃透。很多教程一上来就扔给你一个向量数据库说这就是记忆系统这太片面了。一个完整的企业级记忆系统至少包含四个核心部分我们逐一拆解。2.1 短期记忆不仅仅是对话历史短期记忆的核心任务是维持当前任务或会话的连贯性。它通常直接受限于大模型本身的上下文窗口长度。很多人直接把完整的对话历史扔进上下文这在小规模测试时没问题但一旦对话轮次增多或单次输入内容很长很快就会触及窗口上限。更工程化的做法是动态的短期记忆管理原始记录完整、按序存储当前会话的所有消息用户输入、Agent 思考、工具调用、系统输出。摘要压缩当对话轮次或内容长度达到某个阈值时触发摘要过程。使用大模型将之前的多轮对话压缩成一段精炼的摘要保留核心事实、决策和用户意图。上下文组装下一次请求模型时不再发送全部原始历史而是发送“本轮问题 最新摘要 最近几轮原始对话”。这样能在有限的上下文窗口内携带更长时间跨度的信息。为什么这么做因为大模型处理长文本的成本时间、算力是指数级增长的。摘要压缩是用一次小的计算开销换取后续多轮交互的顺畅和低成本。这是平衡效果与效率的关键。2.2 长期记忆知识库与向量检索的深度结合长期记忆用于存储超越单次会话的、需要持久化并支持未来检索的信息。它不应该是对话历史的简单备份而应该是经过结构化处理的知识点。常见的实现是“向量数据库 元数据”的方案信息提取与分块从 Agent 的交互中识别出值得长期存储的信息。例如用户明确说“记住我喜欢喝美式咖啡”或者 Agent 通过工具查询到了某个产品的价格政策。将这些信息从原始对话中剥离出来处理成清晰的文本片段分块。向量化嵌入使用嵌入模型如 text-embedding-3-small将文本块转化为高维向量。这个向量代表了该文本的语义。存储与索引将向量和对应的原始文本以及关键的元数据如时间戳、来源会话、信息类型等存入向量数据库如 Chroma, Pinecone, Weaviate。检索当 Agent 处理新任务时将当前问题或上下文也转化为向量在向量数据库中进行相似性搜索找出最相关的几条长期记忆作为补充上下文提供给大模型。关键设计选择存储时机是每轮对话后都尝试存储还是仅在识别到特定类型信息如用户偏好、事实结论时才存储后者更精准成本更低。更新与失效记忆不是一成不变的。如何更新过时的信息例如用户说“我改喝拿铁了”简单的方案是新增一条记录并标记旧记录失效复杂点可以做逻辑关联。元数据过滤检索时除了语义相似往往还需要用元数据过滤。比如只检索某个特定用户的记忆或某个时间点之后的记忆。这能极大提升检索精度。2.3 记忆的写入与读取流程理解了长短记忆是什么还要设计它们如何协同工作。这是一个典型的读写流程写入流程记忆形成当前轮次交互完成 - 信息进入短期记忆原始队列 - (定期或触发) 对短期记忆进行摘要压缩 - 压缩摘要更新短期记忆池 - (判断是否为有价值长期知识) - 是 - 提取、分块、向量化 - 存入长期记忆向量库读取流程记忆唤起新请求到来 - 从短期记忆池中组装上下文摘要最近记录 - 将新请求向量化在长期记忆中检索相关条目 - 将“短期记忆上下文” “检索到的长期记忆” “新请求”一并提交给大模型 - 模型生成基于完整“记忆”的回复2.4 与热门框架如 LangChain, LlamaIndex的关系你可能会听到 LangChain 的ConversationSummaryMemory或 LlamaIndex 的索引工具。它们是优秀的组件和抽象层提供了记忆机制的实现模板。但在企业级场景中你很少能直接使用它们的默认配置。你需要做的是理解其原理它们是如何实现摘要、如何与向量库交互的。定制化改造默认的摘要提示词可能不适合你的业务领域默认的向量检索策略可能召回不相关的信息。你需要根据业务逻辑调整记忆的提取规则、摘要的格式、检索的相似度阈值和元数据过滤条件。关注性能与稳定性生产环境要求高并发、低延迟。你需要考虑向量检索的延迟、摘要生成的耗时以及整个记忆系统的错误处理如检索失败时是降级为空记忆还是重试。3. 手把手代码落地从零搭建记忆系统理论说再多不如跑通一行代码。下面我们用一个相对清晰的例子搭建一个具备长短记忆功能的 Agent 系统。我们会使用 Python并选择一些主流且轻量的库。环境准备建议使用 Python 3.9。我们将主要用到openai(或兼容 OpenAI API 的库)、chromadb(向量数据库)、langchain(用于提供一些基础组件和思路但我们会侧重核心逻辑)。先安装基础包pip install openai chromadb langchain langchain-openai。3.1 定义数据结构与存储层首先定义记忆的基本单元。from datetime import datetime from typing import Dict, Any, List, Optional from pydantic import BaseModel class MemoryItem(BaseModel): 记忆项基类 id: str content: str # 记忆的文本内容 metadata: Dict[str, Any] # 元数据如创建时间、会话ID、用户ID、类型等 embedding: Optional[List[float]] None # 向量嵌入 class ShortTermMemory: 短期记忆管理 def __init__(self, max_raw_entries: int 10): self.raw_memories: List[MemoryItem] [] # 原始记忆队列 self.summary: str # 当前会话摘要 self.max_raw_entries max_raw_entries def add_raw_memory(self, content: str, metadata: Dict): 添加一条原始记忆 item MemoryItem( idfshort_{datetime.utcnow().isoformat()}, contentcontent, metadatametadata ) self.raw_memories.append(item) # 如果原始记忆过多触发摘要压缩 if len(self.raw_memories) self.max_raw_entries: self._summarize_memories() def _summarize_memories(self): 调用大模型生成摘要简化示例 # 这里应该调用LLM例如将self.raw_memories中的content拼接请求生成摘要 # 为简化我们模拟一个过程 print([短期记忆] 触发摘要压缩...) # 假设调用LLM后得到了新的摘要 new_summary f摘要更新于{datetime.utcnow()} 包含约{len(self.raw_memories)}条交互的核心信息。 self.summary new_summary # 摘要后可以清空或保留部分最新原始记忆 self.raw_memories self.raw_memories[-5:] # 保留最后5条原始记录 def get_context_for_llm(self) - str: 组装给LLM的上下文 recent_raw \n.join([m.content for m in self.raw_memories[-3:]]) # 取最近3条原始记录 return f会话摘要{self.summary}\n\n最近对话\n{recent_raw}3.2 实现长期记忆与向量检索接下来实现长期记忆的存储和检索。import chromadb from chromadb.config import Settings class LongTermMemory: 长期记忆管理基于ChromaDB def __init__(self, persist_directory: str ./chroma_memory): self.client chromadb.PersistentClient( pathpersist_directory, settingsSettings(anonymized_telemetryFalse) ) # 创建一个集合collection来存储记忆 self.collection self.client.get_or_create_collection(nameagent_long_term_memory) # 需要嵌入模型这里假设使用OpenAI的嵌入API self.embedding_model text-embedding-3-small # 指定模型 def _get_embedding(self, text: str) - List[float]: 获取文本的向量嵌入需要替换为实际的嵌入调用 # 示例调用OpenAI Embedding API # from openai import OpenAI # client OpenAI(api_keyyour-key) # response client.embeddings.create(modelself.embedding_model, inputtext) # return response.data[0].embedding # 为演示返回一个模拟向量 return [0.1] * 1536 # 假设维度是1536 def store_memory(self, memory_item: MemoryItem): 存储一条长期记忆 embedding self._get_embedding(memory_item.content) self.collection.add( documents[memory_item.content], metadatas[memory_item.metadata], embeddings[embedding], ids[memory_item.id] ) print(f[长期记忆] 已存储记忆{memory_item.id}) def search_memories(self, query: str, filter_metadata: Optional[Dict] None, top_k: int 3) - List[MemoryItem]: 检索相关长期记忆 query_embedding self._get_embedding(query) results self.collection.query( query_embeddings[query_embedding], n_resultstop_k, wherefilter_metadata # 使用元数据过滤 ) memories [] if results[documents]: for i in range(len(results[documents][0])): item MemoryItem( idresults[ids][0][i], contentresults[documents][0][i], metadataresults[metadatas][0][i], embeddingresults[embeddings][0][i] if results[embeddings] else None ) memories.append(item) return memories3.3 构建整合记忆的 Agent 执行循环现在我们将长短记忆整合到一个简单的 Agent 循环中。class AgentWithMemory: 具备记忆能力的Agent def __init__(self): self.short_memory ShortTermMemory() self.long_memory LongTermMemory() # 假设的LLM客户端 self.llm_model gpt-3.5-turbo def _call_llm(self, prompt: str) - str: 调用大模型需要替换为实际调用 # 示例调用OpenAI Chat API # from openai import OpenAI # client OpenAI(api_keyyour-key) # response client.chat.completions.create(modelself.llm_model, messages[{role: user, content: prompt}]) # return response.choices[0].message.content return f模拟LLM对以下输入的回复\n{prompt} def _extract_long_term_info(self, conversation_turn: str) - Optional[MemoryItem]: 判断当前对话轮次中是否有需要长期记忆的信息简单规则示例 # 这是一个非常简单的启发式规则。生产环境可能需要更复杂的NLP或规则引擎。 keywords [记住, 我喜欢, 我的偏好是, 重要信息] for kw in keywords: if kw in conversation_turn: # 提取出关键句子或信息 return MemoryItem( idflong_{datetime.utcnow().isoformat()}, contentconversation_turn, metadata{type: user_preference, extracted_at: datetime.utcnow().isoformat()} ) return None def run_turn(self, user_input: str, user_id: str default_user): 运行一轮对话 # 1. 读取记忆组装短期上下文 检索长期记忆 short_context self.short_memory.get_context_for_llm() long_memories self.long_memory.search_memories( queryuser_input, filter_metadata{user_id: user_id} # 只检索该用户的记忆 ) long_context \n.join([m.content for m in long_memories]) if long_memories else 无相关长期记忆。 # 2. 构造最终Prompt full_prompt f 你是一个有帮助的助手请根据以下记忆和当前问题回答。 【历史会话摘要与近期对话】 {short_context} 【来自长期记忆的相关信息】 {long_context} 【当前用户问题】 {user_input} 请回答 # 3. 调用LLM获取回复 llm_response self._call_llm(full_prompt) # 4. 写入记忆 # 4.1 短期记忆记录本轮交互 self.short_memory.add_raw_memory(f用户{user_input}, {role: user, user_id: user_id}) self.short_memory.add_raw_memory(f助手{llm_response}, {role: assistant, user_id: user_id}) # 4.2 长期记忆判断是否需要存储 long_term_item self._extract_long_term_info(user_input) if long_term_item: long_term_item.metadata[user_id] user_id self.long_memory.store_memory(long_term_item) return llm_response # 运行一个简单示例 if __name__ __main__: agent AgentWithMemory() print(Agent 记忆系统启动...\n) responses [] responses.append(agent.run_turn(你好我是小明。)) responses.append(agent.run_turn(我喜欢打篮球和编程。)) responses.append(agent.run_turn(记住我咖啡只喝美式不加糖。)) responses.append(agent.run_turn(我之前和你提过我的爱好吗)) responses.append(agent.run_turn(那我喜欢的咖啡是什么)) for i, resp in enumerate(responses): print(f轮次 {i1} 回复: {resp[:100]}...) # 打印前100字符这段代码勾勒出了一个最小可运行的记忆系统骨架。它包含了短期记忆的滚动摘要、长期记忆的向量化存储与检索以及在每轮对话中动态组装上下文的流程。4. 从 Demo 到企业级必须解决的工程化问题上面的代码能跑通一个概念验证但离“企业级”还差很远。企业级意味着稳定、高效、可扩展、易维护。以下几个问题是升级路上必须面对的。4.1 记忆的提取与摘要质量这是记忆系统的灵魂。垃圾进垃圾出。问题我们之前用的_extract_long_term_info函数只是简单关键词匹配这在实际中远远不够。用户说“我好像更倾向于方案A了”这种隐含的偏好如何提取解决方案微调分类器训练一个小的文本分类模型判断一段文本是否包含“可记忆”的信息如用户偏好、事实陈述、任务结果等。使用更强大的LLM进行提取在将信息存入长期记忆前先用一个LLM调用对其进行重构和提炼。例如将“用户说‘我咖啡只喝美式不加糖。’” 提炼成结构化的{entity: 用户偏好, subject: 咖啡, value: 美式不加糖}。这能极大提升后续检索的准确性。设计领域特定的摘要提示词短期记忆的摘要不能是泛泛而谈。对于客服Agent摘要要突出用户问题和解决方案对于编程助手摘要要突出代码上下文和错误信息。你需要为你的场景定制提示词。4.2 检索的准确性与效率向量检索不是万能的语义相似不代表相关。问题用户问“之前提到的那个方案”向量检索可能找不到因为“那个方案”没有明确的语义向量。解决方案混合检索结合向量检索语义相似和关键词检索如 BM25。先用关键词缩小范围再用向量排序。丰富的元数据为每条记忆打上丰富的标签user_id,session_id,timestamp,entity_type,topic等。检索时先通过元数据进行硬过滤如user_id‘小明’ AND entity_type‘咖啡偏好’再在结果集内做向量相似度排序。这能精准命中目标。检索后重排序检索出 Top K 个结果后再用一个轻量级模型或规则对它们进行相关性重排序确保最相关的排在最前。缓存对频繁查询的长期记忆进行缓存避免每次都对向量数据库进行全量检索。4.3 系统的性能与可观测性线上服务不能接受秒级的延迟。问题每一轮对话都要进行向量嵌入计算和数据库查询延迟可能很高。解决方案异步处理记忆的存储尤其是长期记忆的提取、向量化、存储可以做成异步任务不阻塞主对话流程。用户说完Agent 先基于现有记忆回复后台再慢慢处理记忆存储。批处理将多轮对话的记忆写入请求批量处理减少对向量数据库的写入次数。监控与日志必须对记忆系统的关键指标进行监控短期记忆队列长度、摘要生成耗时、向量检索耗时/召回率、长期记忆存储失败率。这些日志是排查“Agent 又失忆了”这类问题的第一手资料。4.4 记忆的更新、冲突与遗忘记忆不是只增不减的。问题用户说“我不喜欢美式了现在喝拿铁”。如何更新旧的“美式”记忆如果两个记忆冲突怎么办无用的记忆如何清理解决方案逻辑关联与版本管理为同一实体的记忆建立关联。新记忆存入时标记它“覆盖”或“修正”了哪条旧记忆。检索时优先返回最新版本。设置记忆强度与衰减为记忆引入“强度”或“新鲜度”概念。每次被成功检索并利用强度增加随时间流逝强度衰减。强度低于阈值的记忆可以被归档或删除模拟遗忘。人工审核与干预提供后台界面允许管理员查看、编辑或删除 Agent 的记忆。这对于纠正错误记忆、清理测试数据至关重要。5. 实战避坑指南与排查清单最后分享几个从 Demo 走向生产环境时一定会踩的坑和排查思路。5.1 坑点一Agent 表现时好时坏感觉“记忆错乱”可能原因检索结果过多或过杂长期记忆检索返回了太多不相关条目污染了 LLM 的上下文。或者检索结果排序错误最相关的没排在最前面。短期记忆摘要质量差摘要丢失了关键细节或者包含了误导性信息。记忆冲突长期记忆中存在多条相互矛盾的信息LLM 不知该信哪条。排查步骤打印调试在开发阶段将每一轮组装给 LLM 的完整上下文短期长期打印出来或记入日志。肉眼检查提供给模型的信息是否准确、干净。调整检索参数降低top_k比如从 5 降到 2提高相似度得分阈值增加元数据过滤条件。优化摘要提示词在摘要提示词中明确要求“请提取关于[用户偏好]、[任务状态]、[关键决策]的信息忽略寒暄和无关细节。”实施记忆去重与冲突检测在存储长期记忆前检查是否有语义高度相似或直接冲突的旧记忆并采取合并或标记策略。5.2 坑点二系统响应速度越来越慢可能原因短期记忆膨胀原始记忆队列没有有效压缩导致组装上下文时文本过长。向量数据库性能下降记忆条目累积到百万级没有做索引优化或分区。嵌入模型调用延迟每次检索都要实时调用嵌入 API网络或模型延迟成为瓶颈。排查步骤监控短期记忆长度设定告警当原始记忆条数或字符数超过阈值时触发强制摘要。向量数据库优化对向量索引类型如 HNSW的参数进行调优按时间或用户对集合进行分区。嵌入缓存对常见的查询文本和记忆文本的嵌入向量进行本地缓存。可以使用LRU Cache。考虑轻量级本地嵌入模型对于非核心场景可以使用all-MiniLM-L6-v2这类本地模型虽然效果略逊但延迟极低且无网络开销。5.3 坑点三长期记忆似乎根本没被用到可能原因提取规则太严_extract_long_term_info逻辑太苛刻导致几乎没有信息被存入长期记忆。检索 query 构建不佳直接用用户当前问题作为检索 query可能太短或太模糊无法匹配到已存储的记忆。元数据过滤过强比如误用了错误的user_id进行过滤。排查步骤检查长期记忆库直接查询向量数据库看看里面到底存了什么。确保有数据。优化 query 扩展对用户当前问题用 LLM 生成几个相关的搜索关键词或改写句用这些扩展后的 query 去检索。放宽提取规则初期可以设置一个“学习模式”将更多交互内容存入长期记忆即使可能包含噪音然后通过分析检索和使用的日志来迭代优化提取规则。记录检索日志记录每一次检索的 query、返回的记忆 ID 和得分。分析低分或空返回的原因。最后的核心建议不要试图一开始就设计一个完美的、大而全的记忆系统。最好的路径是从简单开始逐步迭代。先实现一个基础的、能工作的版本就像我们上面的代码把它接入你的 Agent 跑起来。然后通过真实的用户交互数据去观察和分析记忆系统在哪里出了问题是存错了还是没存是检索不到还是检索错了。用数据驱动你去优化提取策略、摘要提示词和检索参数。这样构建出来的记忆系统才是真正能解决你业务中“上下文丢失”痛点的系统。 企业级AI Agent记忆系统：从原理到代码落地的工程实践
1. 先搞清楚企业级 Agent 记忆系统到底要解决什么如果你正在开发或使用 AI Agent大概率遇到过这个问题对话进行到一半Agent 好像“失忆”了不记得之前提过的关键信息或者处理长文档、多轮复杂任务时上下文窗口一满前面的内容就被无情丢弃。这本质上是上下文丢失是当前基于大语言模型的 Agent 在走向实用化过程中最核心的痛点之一。所谓的“企业级 Agent 记忆系统”目标就是解决这个痛点。它不是一个单一的模型或工具而是一套工程化的记忆管理策略和存储架构。核心是把 Agent 运行过程中产生的海量、杂乱的交互信息用户指令、中间思考、工具调用结果、最终输出等进行有效的筛选、压缩、存储和高效检索让 Agent 在需要时能“回想”起相关的历史从而做出更连贯、更准确的决策。和网上很多只讲概念的文章不同我们这次要拆解的是能代码落地的方案。你会看到一个实用的记忆系统关键在于区分短期记忆和长期记忆并设计好它们之间的流转机制。短期记忆就像工作内存处理当前会话的即时信息长期记忆则像知识库或数据库存储经过提炼的、未来可能用到的关键知识。很多项目失败不是因为模型不够强而是记忆流转的管道没设计好。所以这篇文章适合两类人一是正在为自家 Agent 健忘而头疼的开发者二是想深入理解 Agent 架构、超越简单 Prompt 工程的学习者。我们不会空谈理论而是会从原理拆解开始一步步走到可运行的代码告诉你每个设计选择背后的“为什么”以及在实际部署时最容易踩的坑。2. 拆解长短记忆的核心原理与设计选择在动手写代码之前必须把原理吃透。很多教程一上来就扔给你一个向量数据库说这就是记忆系统这太片面了。一个完整的企业级记忆系统至少包含四个核心部分我们逐一拆解。2.1 短期记忆不仅仅是对话历史短期记忆的核心任务是维持当前任务或会话的连贯性。它通常直接受限于大模型本身的上下文窗口长度。很多人直接把完整的对话历史扔进上下文这在小规模测试时没问题但一旦对话轮次增多或单次输入内容很长很快就会触及窗口上限。更工程化的做法是动态的短期记忆管理原始记录完整、按序存储当前会话的所有消息用户输入、Agent 思考、工具调用、系统输出。摘要压缩当对话轮次或内容长度达到某个阈值时触发摘要过程。使用大模型将之前的多轮对话压缩成一段精炼的摘要保留核心事实、决策和用户意图。上下文组装下一次请求模型时不再发送全部原始历史而是发送“本轮问题 最新摘要 最近几轮原始对话”。这样能在有限的上下文窗口内携带更长时间跨度的信息。为什么这么做因为大模型处理长文本的成本时间、算力是指数级增长的。摘要压缩是用一次小的计算开销换取后续多轮交互的顺畅和低成本。这是平衡效果与效率的关键。2.2 长期记忆知识库与向量检索的深度结合长期记忆用于存储超越单次会话的、需要持久化并支持未来检索的信息。它不应该是对话历史的简单备份而应该是经过结构化处理的知识点。常见的实现是“向量数据库 元数据”的方案信息提取与分块从 Agent 的交互中识别出值得长期存储的信息。例如用户明确说“记住我喜欢喝美式咖啡”或者 Agent 通过工具查询到了某个产品的价格政策。将这些信息从原始对话中剥离出来处理成清晰的文本片段分块。向量化嵌入使用嵌入模型如 text-embedding-3-small将文本块转化为高维向量。这个向量代表了该文本的语义。存储与索引将向量和对应的原始文本以及关键的元数据如时间戳、来源会话、信息类型等存入向量数据库如 Chroma, Pinecone, Weaviate。检索当 Agent 处理新任务时将当前问题或上下文也转化为向量在向量数据库中进行相似性搜索找出最相关的几条长期记忆作为补充上下文提供给大模型。关键设计选择存储时机是每轮对话后都尝试存储还是仅在识别到特定类型信息如用户偏好、事实结论时才存储后者更精准成本更低。更新与失效记忆不是一成不变的。如何更新过时的信息例如用户说“我改喝拿铁了”简单的方案是新增一条记录并标记旧记录失效复杂点可以做逻辑关联。元数据过滤检索时除了语义相似往往还需要用元数据过滤。比如只检索某个特定用户的记忆或某个时间点之后的记忆。这能极大提升检索精度。2.3 记忆的写入与读取流程理解了长短记忆是什么还要设计它们如何协同工作。这是一个典型的读写流程写入流程记忆形成当前轮次交互完成 - 信息进入短期记忆原始队列 - (定期或触发) 对短期记忆进行摘要压缩 - 压缩摘要更新短期记忆池 - (判断是否为有价值长期知识) - 是 - 提取、分块、向量化 - 存入长期记忆向量库读取流程记忆唤起新请求到来 - 从短期记忆池中组装上下文摘要最近记录 - 将新请求向量化在长期记忆中检索相关条目 - 将“短期记忆上下文” “检索到的长期记忆” “新请求”一并提交给大模型 - 模型生成基于完整“记忆”的回复2.4 与热门框架如 LangChain, LlamaIndex的关系你可能会听到 LangChain 的ConversationSummaryMemory或 LlamaIndex 的索引工具。它们是优秀的组件和抽象层提供了记忆机制的实现模板。但在企业级场景中你很少能直接使用它们的默认配置。你需要做的是理解其原理它们是如何实现摘要、如何与向量库交互的。定制化改造默认的摘要提示词可能不适合你的业务领域默认的向量检索策略可能召回不相关的信息。你需要根据业务逻辑调整记忆的提取规则、摘要的格式、检索的相似度阈值和元数据过滤条件。关注性能与稳定性生产环境要求高并发、低延迟。你需要考虑向量检索的延迟、摘要生成的耗时以及整个记忆系统的错误处理如检索失败时是降级为空记忆还是重试。3. 手把手代码落地从零搭建记忆系统理论说再多不如跑通一行代码。下面我们用一个相对清晰的例子搭建一个具备长短记忆功能的 Agent 系统。我们会使用 Python并选择一些主流且轻量的库。环境准备建议使用 Python 3.9。我们将主要用到openai(或兼容 OpenAI API 的库)、chromadb(向量数据库)、langchain(用于提供一些基础组件和思路但我们会侧重核心逻辑)。先安装基础包pip install openai chromadb langchain langchain-openai。3.1 定义数据结构与存储层首先定义记忆的基本单元。from datetime import datetime from typing import Dict, Any, List, Optional from pydantic import BaseModel class MemoryItem(BaseModel): 记忆项基类 id: str content: str # 记忆的文本内容 metadata: Dict[str, Any] # 元数据如创建时间、会话ID、用户ID、类型等 embedding: Optional[List[float]] None # 向量嵌入 class ShortTermMemory: 短期记忆管理 def __init__(self, max_raw_entries: int 10): self.raw_memories: List[MemoryItem] [] # 原始记忆队列 self.summary: str # 当前会话摘要 self.max_raw_entries max_raw_entries def add_raw_memory(self, content: str, metadata: Dict): 添加一条原始记忆 item MemoryItem( idfshort_{datetime.utcnow().isoformat()}, contentcontent, metadatametadata ) self.raw_memories.append(item) # 如果原始记忆过多触发摘要压缩 if len(self.raw_memories) self.max_raw_entries: self._summarize_memories() def _summarize_memories(self): 调用大模型生成摘要简化示例 # 这里应该调用LLM例如将self.raw_memories中的content拼接请求生成摘要 # 为简化我们模拟一个过程 print([短期记忆] 触发摘要压缩...) # 假设调用LLM后得到了新的摘要 new_summary f摘要更新于{datetime.utcnow()} 包含约{len(self.raw_memories)}条交互的核心信息。 self.summary new_summary # 摘要后可以清空或保留部分最新原始记忆 self.raw_memories self.raw_memories[-5:] # 保留最后5条原始记录 def get_context_for_llm(self) - str: 组装给LLM的上下文 recent_raw \n.join([m.content for m in self.raw_memories[-3:]]) # 取最近3条原始记录 return f会话摘要{self.summary}\n\n最近对话\n{recent_raw}3.2 实现长期记忆与向量检索接下来实现长期记忆的存储和检索。import chromadb from chromadb.config import Settings class LongTermMemory: 长期记忆管理基于ChromaDB def __init__(self, persist_directory: str ./chroma_memory): self.client chromadb.PersistentClient( pathpersist_directory, settingsSettings(anonymized_telemetryFalse) ) # 创建一个集合collection来存储记忆 self.collection self.client.get_or_create_collection(nameagent_long_term_memory) # 需要嵌入模型这里假设使用OpenAI的嵌入API self.embedding_model text-embedding-3-small # 指定模型 def _get_embedding(self, text: str) - List[float]: 获取文本的向量嵌入需要替换为实际的嵌入调用 # 示例调用OpenAI Embedding API # from openai import OpenAI # client OpenAI(api_keyyour-key) # response client.embeddings.create(modelself.embedding_model, inputtext) # return response.data[0].embedding # 为演示返回一个模拟向量 return [0.1] * 1536 # 假设维度是1536 def store_memory(self, memory_item: MemoryItem): 存储一条长期记忆 embedding self._get_embedding(memory_item.content) self.collection.add( documents[memory_item.content], metadatas[memory_item.metadata], embeddings[embedding], ids[memory_item.id] ) print(f[长期记忆] 已存储记忆{memory_item.id}) def search_memories(self, query: str, filter_metadata: Optional[Dict] None, top_k: int 3) - List[MemoryItem]: 检索相关长期记忆 query_embedding self._get_embedding(query) results self.collection.query( query_embeddings[query_embedding], n_resultstop_k, wherefilter_metadata # 使用元数据过滤 ) memories [] if results[documents]: for i in range(len(results[documents][0])): item MemoryItem( idresults[ids][0][i], contentresults[documents][0][i], metadataresults[metadatas][0][i], embeddingresults[embeddings][0][i] if results[embeddings] else None ) memories.append(item) return memories3.3 构建整合记忆的 Agent 执行循环现在我们将长短记忆整合到一个简单的 Agent 循环中。class AgentWithMemory: 具备记忆能力的Agent def __init__(self): self.short_memory ShortTermMemory() self.long_memory LongTermMemory() # 假设的LLM客户端 self.llm_model gpt-3.5-turbo def _call_llm(self, prompt: str) - str: 调用大模型需要替换为实际调用 # 示例调用OpenAI Chat API # from openai import OpenAI # client OpenAI(api_keyyour-key) # response client.chat.completions.create(modelself.llm_model, messages[{role: user, content: prompt}]) # return response.choices[0].message.content return f模拟LLM对以下输入的回复\n{prompt} def _extract_long_term_info(self, conversation_turn: str) - Optional[MemoryItem]: 判断当前对话轮次中是否有需要长期记忆的信息简单规则示例 # 这是一个非常简单的启发式规则。生产环境可能需要更复杂的NLP或规则引擎。 keywords [记住, 我喜欢, 我的偏好是, 重要信息] for kw in keywords: if kw in conversation_turn: # 提取出关键句子或信息 return MemoryItem( idflong_{datetime.utcnow().isoformat()}, contentconversation_turn, metadata{type: user_preference, extracted_at: datetime.utcnow().isoformat()} ) return None def run_turn(self, user_input: str, user_id: str default_user): 运行一轮对话 # 1. 读取记忆组装短期上下文 检索长期记忆 short_context self.short_memory.get_context_for_llm() long_memories self.long_memory.search_memories( queryuser_input, filter_metadata{user_id: user_id} # 只检索该用户的记忆 ) long_context \n.join([m.content for m in long_memories]) if long_memories else 无相关长期记忆。 # 2. 构造最终Prompt full_prompt f 你是一个有帮助的助手请根据以下记忆和当前问题回答。 【历史会话摘要与近期对话】 {short_context} 【来自长期记忆的相关信息】 {long_context} 【当前用户问题】 {user_input} 请回答 # 3. 调用LLM获取回复 llm_response self._call_llm(full_prompt) # 4. 写入记忆 # 4.1 短期记忆记录本轮交互 self.short_memory.add_raw_memory(f用户{user_input}, {role: user, user_id: user_id}) self.short_memory.add_raw_memory(f助手{llm_response}, {role: assistant, user_id: user_id}) # 4.2 长期记忆判断是否需要存储 long_term_item self._extract_long_term_info(user_input) if long_term_item: long_term_item.metadata[user_id] user_id self.long_memory.store_memory(long_term_item) return llm_response # 运行一个简单示例 if __name__ __main__: agent AgentWithMemory() print(Agent 记忆系统启动...\n) responses [] responses.append(agent.run_turn(你好我是小明。)) responses.append(agent.run_turn(我喜欢打篮球和编程。)) responses.append(agent.run_turn(记住我咖啡只喝美式不加糖。)) responses.append(agent.run_turn(我之前和你提过我的爱好吗)) responses.append(agent.run_turn(那我喜欢的咖啡是什么)) for i, resp in enumerate(responses): print(f轮次 {i1} 回复: {resp[:100]}...) # 打印前100字符这段代码勾勒出了一个最小可运行的记忆系统骨架。它包含了短期记忆的滚动摘要、长期记忆的向量化存储与检索以及在每轮对话中动态组装上下文的流程。4. 从 Demo 到企业级必须解决的工程化问题上面的代码能跑通一个概念验证但离“企业级”还差很远。企业级意味着稳定、高效、可扩展、易维护。以下几个问题是升级路上必须面对的。4.1 记忆的提取与摘要质量这是记忆系统的灵魂。垃圾进垃圾出。问题我们之前用的_extract_long_term_info函数只是简单关键词匹配这在实际中远远不够。用户说“我好像更倾向于方案A了”这种隐含的偏好如何提取解决方案微调分类器训练一个小的文本分类模型判断一段文本是否包含“可记忆”的信息如用户偏好、事实陈述、任务结果等。使用更强大的LLM进行提取在将信息存入长期记忆前先用一个LLM调用对其进行重构和提炼。例如将“用户说‘我咖啡只喝美式不加糖。’” 提炼成结构化的{entity: 用户偏好, subject: 咖啡, value: 美式不加糖}。这能极大提升后续检索的准确性。设计领域特定的摘要提示词短期记忆的摘要不能是泛泛而谈。对于客服Agent摘要要突出用户问题和解决方案对于编程助手摘要要突出代码上下文和错误信息。你需要为你的场景定制提示词。4.2 检索的准确性与效率向量检索不是万能的语义相似不代表相关。问题用户问“之前提到的那个方案”向量检索可能找不到因为“那个方案”没有明确的语义向量。解决方案混合检索结合向量检索语义相似和关键词检索如 BM25。先用关键词缩小范围再用向量排序。丰富的元数据为每条记忆打上丰富的标签user_id,session_id,timestamp,entity_type,topic等。检索时先通过元数据进行硬过滤如user_id‘小明’ AND entity_type‘咖啡偏好’再在结果集内做向量相似度排序。这能精准命中目标。检索后重排序检索出 Top K 个结果后再用一个轻量级模型或规则对它们进行相关性重排序确保最相关的排在最前。缓存对频繁查询的长期记忆进行缓存避免每次都对向量数据库进行全量检索。4.3 系统的性能与可观测性线上服务不能接受秒级的延迟。问题每一轮对话都要进行向量嵌入计算和数据库查询延迟可能很高。解决方案异步处理记忆的存储尤其是长期记忆的提取、向量化、存储可以做成异步任务不阻塞主对话流程。用户说完Agent 先基于现有记忆回复后台再慢慢处理记忆存储。批处理将多轮对话的记忆写入请求批量处理减少对向量数据库的写入次数。监控与日志必须对记忆系统的关键指标进行监控短期记忆队列长度、摘要生成耗时、向量检索耗时/召回率、长期记忆存储失败率。这些日志是排查“Agent 又失忆了”这类问题的第一手资料。4.4 记忆的更新、冲突与遗忘记忆不是只增不减的。问题用户说“我不喜欢美式了现在喝拿铁”。如何更新旧的“美式”记忆如果两个记忆冲突怎么办无用的记忆如何清理解决方案逻辑关联与版本管理为同一实体的记忆建立关联。新记忆存入时标记它“覆盖”或“修正”了哪条旧记忆。检索时优先返回最新版本。设置记忆强度与衰减为记忆引入“强度”或“新鲜度”概念。每次被成功检索并利用强度增加随时间流逝强度衰减。强度低于阈值的记忆可以被归档或删除模拟遗忘。人工审核与干预提供后台界面允许管理员查看、编辑或删除 Agent 的记忆。这对于纠正错误记忆、清理测试数据至关重要。5. 实战避坑指南与排查清单最后分享几个从 Demo 走向生产环境时一定会踩的坑和排查思路。5.1 坑点一Agent 表现时好时坏感觉“记忆错乱”可能原因检索结果过多或过杂长期记忆检索返回了太多不相关条目污染了 LLM 的上下文。或者检索结果排序错误最相关的没排在最前面。短期记忆摘要质量差摘要丢失了关键细节或者包含了误导性信息。记忆冲突长期记忆中存在多条相互矛盾的信息LLM 不知该信哪条。排查步骤打印调试在开发阶段将每一轮组装给 LLM 的完整上下文短期长期打印出来或记入日志。肉眼检查提供给模型的信息是否准确、干净。调整检索参数降低top_k比如从 5 降到 2提高相似度得分阈值增加元数据过滤条件。优化摘要提示词在摘要提示词中明确要求“请提取关于[用户偏好]、[任务状态]、[关键决策]的信息忽略寒暄和无关细节。”实施记忆去重与冲突检测在存储长期记忆前检查是否有语义高度相似或直接冲突的旧记忆并采取合并或标记策略。5.2 坑点二系统响应速度越来越慢可能原因短期记忆膨胀原始记忆队列没有有效压缩导致组装上下文时文本过长。向量数据库性能下降记忆条目累积到百万级没有做索引优化或分区。嵌入模型调用延迟每次检索都要实时调用嵌入 API网络或模型延迟成为瓶颈。排查步骤监控短期记忆长度设定告警当原始记忆条数或字符数超过阈值时触发强制摘要。向量数据库优化对向量索引类型如 HNSW的参数进行调优按时间或用户对集合进行分区。嵌入缓存对常见的查询文本和记忆文本的嵌入向量进行本地缓存。可以使用LRU Cache。考虑轻量级本地嵌入模型对于非核心场景可以使用all-MiniLM-L6-v2这类本地模型虽然效果略逊但延迟极低且无网络开销。5.3 坑点三长期记忆似乎根本没被用到可能原因提取规则太严_extract_long_term_info逻辑太苛刻导致几乎没有信息被存入长期记忆。检索 query 构建不佳直接用用户当前问题作为检索 query可能太短或太模糊无法匹配到已存储的记忆。元数据过滤过强比如误用了错误的user_id进行过滤。排查步骤检查长期记忆库直接查询向量数据库看看里面到底存了什么。确保有数据。优化 query 扩展对用户当前问题用 LLM 生成几个相关的搜索关键词或改写句用这些扩展后的 query 去检索。放宽提取规则初期可以设置一个“学习模式”将更多交互内容存入长期记忆即使可能包含噪音然后通过分析检索和使用的日志来迭代优化提取规则。记录检索日志记录每一次检索的 query、返回的记忆 ID 和得分。分析低分或空返回的原因。最后的核心建议不要试图一开始就设计一个完美的、大而全的记忆系统。最好的路径是从简单开始逐步迭代。先实现一个基础的、能工作的版本就像我们上面的代码把它接入你的 Agent 跑起来。然后通过真实的用户交互数据去观察和分析记忆系统在哪里出了问题是存错了还是没存是检索不到还是检索错了。用数据驱动你去优化提取策略、摘要提示词和检索参数。这样构建出来的记忆系统才是真正能解决你业务中“上下文丢失”痛点的系统。