1. 从“健忘”到“有记忆”为什么智能体需要记忆系统在智能体开发的前几篇实战中我们搭建了感知、决策和行动的基础框架。一个能听会说、能调用工具的智能体已经初具雏形。但如果你和它多聊几句或者让它处理一个稍长的任务链很快就会发现一个致命问题它像个“金鱼”只有七秒记忆。你刚告诉它“我叫张三喜欢喝冰美式”下一句问“给我推荐一杯咖啡”它可能就会推荐拿铁或卡布奇诺完全忘了你上一秒的偏好。这种“健忘”让智能体无法进行连贯的对话也无法执行需要上下文依赖的复杂任务。记忆系统就是解决这个问题的核心组件它让智能体从“单次对话模型”进化为“持续交互的智能体”。简单来说记忆系统就是智能体的“大脑皮层”负责信息的存储、索引和提取。没有它每次交互都是孤立的有了它智能体才能认识你、理解任务背景、积累经验甚至形成独特的“个性”。在智能体开发中记忆系统绝非一个可选的“加分项”而是实现实用化、人格化智能体的基石。本次实战我们将深入一个智能体记忆系统的设计与实现从零开始构建一个具备短期、长期和工作记忆的完整记忆模块并解决其中最棘手的几个问题记忆如何高效存储如何在海量记忆中快速找到相关的以及如何防止记忆被无关信息污染2. 记忆系统的三层架构设计短期、长期与工作记忆人的记忆是分层的智能体的记忆系统也应如此。直接用一个列表或数据库存储所有对话历史是最简陋的做法这会导致检索效率低下和记忆干扰。一个健壮的记忆系统通常借鉴认知心理学设计为三层结构短期记忆、长期记忆和工作记忆。2.1 短期记忆对话的即时上下文短期记忆或称对话记忆保存当前对话轮次中的信息。它的容量有限生命周期短通常只持续一次对话会话。在技术实现上这往往就是传递给大语言模型LLM的上下文窗口Context Window内的内容。核心实现要点数据结构通常是一个固定长度的队列FIFO。当新的对话轮次用户输入智能体回复加入时如果队列已满则自动移除最旧的轮次。容量管理容量不是指“条数”而是“Token数”。你需要根据所用LLM的上下文长度来设定。例如如果使用上下文窗口为128K的模型你可能需要为短期记忆预留4K-8K Token剩下的空间留给系统指令、工具描述和本次查询。内容格式化每条记忆不应是原始文本而应被结构化为一个对象至少包含roleuser/assistant、content和timestamp。更高级的可以包含type陈述、问题、指令等。from collections import deque import tiktoken # 用于计算Token的库 class ShortTermMemory: def __init__(self, max_token_limit4000): self.memory deque() self.max_token_limit max_token_limit self.current_tokens 0 self.encoder tiktoken.get_encoding(cl100k_base) # 适配GPT-4等模型 def add_interaction(self, role, content): 添加一次交互到短期记忆 interaction {role: role, content: content, timestamp: time.time()} token_count len(self.encoder.encode(content)) # 检查并确保加入后不超过限制 while self.current_tokens token_count self.max_token_limit and self.memory: removed self.memory.popleft() self.current_tokens - len(self.encoder.encode(removed[content])) self.memory.append(interaction) self.current_tokens token_count def get_context(self): 获取格式化后的上下文用于拼接到LLM提示词中 return [{role: m[role], content: m[content]} for m in self.memory]注意tiktoken是OpenAI官方的Token计算库但如果你使用非OpenAI系列模型需要找到对应模型的Tokenizer。短期记忆的溢出管理策略也可以是“总结压缩”即当快满时让LLM自动对最早期的几条记忆进行摘要然后用摘要替换原始内容但这会引入额外的计算开销。2.2 长期记忆智能体的知识库与经验库长期记忆是智能体的持久化存储用于保存跨越对话会话的重要信息例如用户画像、关键事实、执行任务的经验教训等。它的特点是容量大、保存时间长但检索是关键挑战。设计核心向量数据库Vector Database纯文本匹配检索在长期记忆中效率低下。例如用户说“我上次说的那家意大利餐厅”关键词匹配可能失效。向量数据库通过将文本转换为高维向量嵌入并计算向量间的相似度如余弦相似度来检索可以实现语义级别的模糊匹配。实现步骤记忆生成当判断一条信息需要长期保存时例如用户明确说“记住这个”或智能体自行判断为关键信息将其转换为文本片段。向量化使用嵌入模型如text-embedding-3-small将该文本转换为一个固定长度的向量。存储将向量、原始文本以及元数据如来源对话ID、时间戳、信息类型标签一起存入向量数据库。检索当需要从长期记忆中回忆时将当前查询或上下文也转换为向量然后在向量数据库中搜索最相似的K个记忆片段。# 示例使用ChromaDB轻量级向量数据库 import chromadb from chromadb.config import Settings from openai import OpenAI # 假设使用OpenAI Embedding模型 class LongTermMemory: def __init__(self, persist_directory./chroma_db): self.client chromadb.PersistentClient(pathpersist_directory) self.collection self.client.get_or_create_collection(nameagent_memories) self.embed_client OpenAI() # 初始化Embedding客户端 def _get_embedding(self, text): response self.embed_client.embeddings.create(modeltext-embedding-3-small, inputtext) return response.data[0].embedding def store_memory(self, text, metadata{}): 存储一条长期记忆 embedding self._get_embedding(text) # ChromaDB会自动生成ID这里我们用时间戳哈希生成唯一ID memory_id fmem_{int(time.time())}_{hash(text) % 10000:04d} self.collection.add( embeddings[embedding], documents[text], metadatas[metadata], ids[memory_id] ) def retrieve_memories(self, query_text, n_results3): 根据查询检索相关记忆 query_embedding self._get_embedding(query_text) results self.collection.query( query_embeddings[query_embedding], n_resultsn_results ) # results 包含 documents, metadatas, distances 等 return results[documents][0] if results[documents] else []实操心得元数据metadata的设计至关重要。除了时间戳建议加上entity实体如“用户偏好”、“餐厅信息”、importance重要性评分由LLM生成等字段。这样在检索时可以结合向量相似度和元数据过滤得到更精准的结果。例如可以优先检索与当前对话实体相关且重要性高的记忆。2.3 工作记忆任务执行的“草稿纸”工作记忆是智能体在执行当前任务时的临时信息暂存区。它不同于短期记忆的对话历史而是专注于一个任务目标的上下文。例如在完成“订机票-订酒店-租车”的复杂任务时工作记忆会暂存已查询到的航班信息、酒店选项等供后续步骤决策使用。实现模式工作记忆通常是一个结构化的、任务粒度的字典或对象。它在任务开始时被初始化在任务执行过程中被不断更新在任务完成后被清理或归档到长期记忆。class WorkingMemory: def __init__(self, task_id): self.task_id task_id self.data { goal: , # 任务目标 steps: [], # 已完成的步骤 current_step: , # 当前步骤 context: {}, # 任务相关上下文如收集到的参数 intermediate_results: {} # 中间结果如查询到的航班列表 } def update(self, key, value): 更新工作记忆中的特定字段 if key in self.data: self.data[key] value else: self.data[context][key] value def get(self, key): 获取工作记忆中的值 return self.data.get(key, self.data[context].get(key))三层记忆各司其职短期记忆维持对话流畅长期记忆形成持久认知工作记忆保障复杂任务执行。它们之间需要协同工作这引出了下一个核心问题记忆的流动与调度。3. 记忆的流动、检索与更新策略记忆不是静态存储的需要在三层之间智能地流动。同时如何从海量长期记忆中精准检索以及如何更新或遗忘记忆是工程上的难点。3.1 记忆的写入什么该记记在哪里并非所有信息都值得记忆。无差别存储会导致记忆库充满噪音。我们需要一个“记忆写入控制器”。策略建议规则触发定义明确规则。例如用户说“记住我喜欢喝茶”或“我的生日是7月10日”这类包含明确指令或关键个人事实的信息直接触发写入长期记忆。LLM判断更灵活的方式是让LLM充当“记忆过滤器”。在每轮对话后将对话片段交给一个轻量级LLM或使用主模型的一个特定提示让其判断是否值得存入长期记忆提取关键事实、用户偏好、承诺等信息类型是什么用户属性、事实知识、待办事项等重要性分数是多少1-10分 基于LLM的输出决定是否存储及如何打标签。def should_save_to_long_term(conversation_snippet, llm_client): prompt f 请分析以下对话片段判断是否包含值得智能体长期记住的重要信息。 对话{conversation_snippet} 请按JSON格式输出{{save: true/false, type: preference/fact/todo, key_info: 提取的关键信息, importance: 分数}}。 response llm_client.chat.completions.create( modelgpt-4o-mini, # 使用小模型降低成本 messages[{role: user, content: prompt}], response_format{type: json_object} ) decision json.loads(response.choices[0].message.content) return decision3.2 记忆的检索在正确的时间想起正确的事检索是记忆系统的灵魂。核心是相关性和时效性。混合检索策略向量检索语义相似性如上文所述是基础。它能找到语义上相关的内容。元数据过滤结合向量检索的结果用元数据如entity、recency进行二次筛选。例如当对话主题是“咖啡”时可以过滤entity包含“咖啡”或“饮品”的记忆。时间衰减加权更近的记忆通常更相关。可以在计算最终相关性分数时加入一个时间衰减因子让近期记忆的排名提升。LLM重排序Rerank对于初步检索到的Top N条记忆可以再用LLM根据当前对话的完整上下文对其相关性进行精细排序和去重。这是效果最好但成本也最高的方法。def retrieve_relevant_memories(current_query, lt_memory, st_memory, n5): 综合检索相关记忆 # 1. 从长期记忆进行向量检索 raw_memories lt_memory.retrieve_memories(current_query, n_resultsn*2) # 2. 构建包含短期记忆的完整上下文 full_context \n.join([f{m[role]}: {m[content]} for m in st_memory.get_context()[-5:]]) # 取最近5轮 # 3. (可选)LLM重排序与摘要 if raw_memories: prompt f 当前对话上下文 {full_context} 用户最新查询{current_query} 以下是从知识库中找到的一些可能相关的记忆片段 {chr(10).join([f{i1}. {mem} for i, mem in enumerate(raw_memories)])} 请根据当前对话选出最相关的1-3条记忆并简要说明理由。如果都不相关请回答“无”。 # 调用LLM进行判断... # 根据LLM输出选择最终的记忆 return filtered_memories3.3 记忆的更新与遗忘保持记忆的鲜活与准确信息会过时用户偏好会改变。一个只会增加不会修改的记忆系统最终会充满矛盾。更新机制冲突检测当要存入的新记忆与已有记忆在语义上高度相似但内容矛盾时例如旧记忆“用户喜欢狗”新记忆“用户对狗毛过敏”触发更新流程。LLM仲裁将冲突的记忆对和当前上下文交给LLM判断哪个信息更可能正确、更新或是可以合并。执行操作根据LLM的判断执行替换、合并或添加时间戳注释例如“截至2023年喜欢狗2024年起因过敏不再养狗”。遗忘压缩机制基于重要性的清理定期扫描长期记忆对重要性评分极低且很久未被访问的记忆进行归档或删除。总结性压缩对于关于同一主题的多次琐碎记忆例如多次提到“喜欢某家店的拿铁”可以定期让LLM生成一个总结性的记忆条目并删除原始琐碎条目。踩坑实录在早期版本中我们只实现了记忆的添加导致当用户说“我其实不喜欢吃辣了”之后智能体仍然会推荐川菜因为旧的“喜欢吃辣”记忆还在且被检索到。实现更新逻辑后我们让LLM判断“不喜欢吃辣了”是对“喜欢吃辣”的否定更新从而用新记忆替换了旧记忆的content字段并在元数据中记录了更新时间彻底解决了这个问题。4. 将记忆整合进智能体推理循环记忆系统不是孤立的模块它必须深度嵌入智能体的核心推理循环感知→思考→行动中。4.1 在感知阶段注入记忆在将用户输入Query提交给LLM进行核心思考规划、工具调用判断之前先进行记忆检索。class AgentWithMemory: def __init__(self): self.st_memory ShortTermMemory() self.lt_memory LongTermMemory() self.working_memory None def perceive(self, user_input): 感知阶段整合记忆 # 1. 检索相关长期记忆 relevant_memories retrieve_relevant_memories(user_input, self.lt_memory, self.st_memory) # 2. 获取短期记忆上下文 short_term_context self.st_memory.get_context() # 3. 获取工作记忆如果存在活跃任务 working_context self.working_memory.data if self.working_memory else {} # 4. 构建增强的提示词 enhanced_prompt self._construct_prompt(user_input, short_term_context, relevant_memories, working_context) return enhanced_prompt def _construct_prompt(self, query, short_term, long_term, working): 构建包含所有上下文的系统提示词 system_message f你是一个有帮助的智能体拥有以下记忆和上下文 关于用户的长期记忆 {chr(10).join(long_term) if long_term else 暂无相关长期记忆。} 当前任务的工作记忆 {json.dumps(working, indent2, ensure_asciiFalse) if working else 无活跃任务。} 最近的对话历史短期记忆 {chr(10).join([f{m[role]}: {m[content]} for m in short_term])} 请基于以上信息回应用户的最新输入{query} return system_message4.2 在思考与行动后更新记忆在LLM生成回复或执行工具调用后需要根据本轮交互的结果更新各个记忆层。def cycle(self, user_input): 智能体的完整交互循环 # 感知 prompt self.perceive(user_input) # 思考与行动调用LLM可能涉及工具调用 llm_response, used_tools_result self.think_and_act(prompt) # 记忆更新阶段 # 1. 更新短期记忆 self.st_memory.add_interaction(user, user_input) self.st_memory.add_interaction(assistant, llm_response) # 2. 判断是否存入长期记忆 conversation_snippet fUser: {user_input}\nAssistant: {llm_response} save_decision should_save_to_long_term(conversation_snippet, llm_client) if save_decision.get(save): self.lt_memory.store_memory( textsave_decision[key_info], metadata{type: save_decision[type], importance: save_decision[importance]} ) # 3. 更新工作记忆如果任务有进展 if self.working_memory and used_tools_result: self.working_memory.update(intermediate_results, used_tools_result) return llm_response4.3 工作记忆驱动任务分解对于复杂任务工作记忆是任务分解Chain-of-Thought和执行的协调中心。LLM根据任务目标存储在working_memory[“goal”]和已有上下文working_memory[“context”]规划下一步更新工作记忆然后执行。5. 实战中的性能优化与常见陷阱实现一个可用的记忆系统是一回事实现一个高效、稳定的记忆系统是另一回事。以下是几个关键的优化点和避坑指南。5.1 向量检索的精度与效率平衡问题直接使用用户当前查询语句进行向量检索可能不够准确特别是查询很短或很模糊时。解决方案查询扩展。在检索前先用LLM对当前查询和对话上下文进行简要分析生成一个更丰富、包含潜在语义的搜索查询语句。def expand_query_for_retrieval(user_query, conversation_context): prompt f基于以下对话历史和最新问题生成一个更全面、用于在知识库中搜索相关信息的查询语句。 历史 {conversation_context} 最新问题{user_query} 生成的搜索查询 expanded_query llm_client.chat.completions.create(...) # 调用小模型 return expanded_query # 然后用 expanded_query 去检索长期记忆5.2 记忆token消耗与成本控制问题记忆内容尤其是长期记忆检索结果会占用大量LLM上下文Token显著增加API调用成本。解决方案记忆摘要在将长期记忆插入上下文前如果记忆文本过长让LLM先对其进行摘要。选择性注入不是所有检索到的记忆都放入上下文。可以设定一个相关性分数阈值或只选择Top 1-2条最相关的记忆。分层提示将最核心的指令和最近几轮对话放在提示词最前面将检索到的记忆放在靠后位置。有些模型对靠前的内容关注度更高。5.3 记忆幻觉与冲突解决问题LLM可能会混淆记忆来源甚至“脑补”出不存在于记忆库中的细节幻觉。或者记忆库中存在多条相互矛盾的信息。解决方案明确引用来源在向LLM提供记忆时明确标注来源例如“[记忆-2024-01-01]用户说喜欢蓝色”。这能一定程度上减少幻觉。设置置信度为每条记忆附加一个置信度分数基于来源可靠性、确认次数等。在出现冲突时优先采用置信度高的记忆。主动询问当冲突无法通过规则解决时最稳妥的方式是让智能体向用户主动确认。“我记得您之前提过喜欢狗但刚才又提到对狗毛过敏请问您目前的偏好是怎样的”5.4 长期记忆的初始化与冷启动问题新智能体长期记忆为空无法进行个性化服务影响初期体验。解决方案预制知识根据智能体的定位预置一些通用知识或领域常识。例如一个咖啡推荐机器人可以预置常见咖啡种类、口味特点等记忆。主动引导在初次交互时智能体可以主动询问一些关键信息来填充记忆例如“为了更好的为您服务可以告诉我您喜欢的咖啡口味吗例如偏苦/偏酸加奶/不加奶”。从交互中快速学习设计机制让智能体在头几次交互中以更高的频率和更低的阈值将信息存入长期记忆快速构建用户画像。构建记忆系统的过程是让智能体真正“活”起来的关键一步。它从本质上改变了智能体的交互模式从一次性的问答变成了持续的、有历史感的陪伴与合作。这套系统目前运行在我们的多个智能体项目中显著提升了任务完成率和用户满意度。当然记忆系统本身也是一个持续学习和优化的过程下一步我们正在探索如何让记忆之间产生更复杂的关联形成真正的“知识图谱”但那将是另一个更深入的话题了。 智能体记忆系统实战：三层架构设计与向量检索优化
1. 从“健忘”到“有记忆”为什么智能体需要记忆系统在智能体开发的前几篇实战中我们搭建了感知、决策和行动的基础框架。一个能听会说、能调用工具的智能体已经初具雏形。但如果你和它多聊几句或者让它处理一个稍长的任务链很快就会发现一个致命问题它像个“金鱼”只有七秒记忆。你刚告诉它“我叫张三喜欢喝冰美式”下一句问“给我推荐一杯咖啡”它可能就会推荐拿铁或卡布奇诺完全忘了你上一秒的偏好。这种“健忘”让智能体无法进行连贯的对话也无法执行需要上下文依赖的复杂任务。记忆系统就是解决这个问题的核心组件它让智能体从“单次对话模型”进化为“持续交互的智能体”。简单来说记忆系统就是智能体的“大脑皮层”负责信息的存储、索引和提取。没有它每次交互都是孤立的有了它智能体才能认识你、理解任务背景、积累经验甚至形成独特的“个性”。在智能体开发中记忆系统绝非一个可选的“加分项”而是实现实用化、人格化智能体的基石。本次实战我们将深入一个智能体记忆系统的设计与实现从零开始构建一个具备短期、长期和工作记忆的完整记忆模块并解决其中最棘手的几个问题记忆如何高效存储如何在海量记忆中快速找到相关的以及如何防止记忆被无关信息污染2. 记忆系统的三层架构设计短期、长期与工作记忆人的记忆是分层的智能体的记忆系统也应如此。直接用一个列表或数据库存储所有对话历史是最简陋的做法这会导致检索效率低下和记忆干扰。一个健壮的记忆系统通常借鉴认知心理学设计为三层结构短期记忆、长期记忆和工作记忆。2.1 短期记忆对话的即时上下文短期记忆或称对话记忆保存当前对话轮次中的信息。它的容量有限生命周期短通常只持续一次对话会话。在技术实现上这往往就是传递给大语言模型LLM的上下文窗口Context Window内的内容。核心实现要点数据结构通常是一个固定长度的队列FIFO。当新的对话轮次用户输入智能体回复加入时如果队列已满则自动移除最旧的轮次。容量管理容量不是指“条数”而是“Token数”。你需要根据所用LLM的上下文长度来设定。例如如果使用上下文窗口为128K的模型你可能需要为短期记忆预留4K-8K Token剩下的空间留给系统指令、工具描述和本次查询。内容格式化每条记忆不应是原始文本而应被结构化为一个对象至少包含roleuser/assistant、content和timestamp。更高级的可以包含type陈述、问题、指令等。from collections import deque import tiktoken # 用于计算Token的库 class ShortTermMemory: def __init__(self, max_token_limit4000): self.memory deque() self.max_token_limit max_token_limit self.current_tokens 0 self.encoder tiktoken.get_encoding(cl100k_base) # 适配GPT-4等模型 def add_interaction(self, role, content): 添加一次交互到短期记忆 interaction {role: role, content: content, timestamp: time.time()} token_count len(self.encoder.encode(content)) # 检查并确保加入后不超过限制 while self.current_tokens token_count self.max_token_limit and self.memory: removed self.memory.popleft() self.current_tokens - len(self.encoder.encode(removed[content])) self.memory.append(interaction) self.current_tokens token_count def get_context(self): 获取格式化后的上下文用于拼接到LLM提示词中 return [{role: m[role], content: m[content]} for m in self.memory]注意tiktoken是OpenAI官方的Token计算库但如果你使用非OpenAI系列模型需要找到对应模型的Tokenizer。短期记忆的溢出管理策略也可以是“总结压缩”即当快满时让LLM自动对最早期的几条记忆进行摘要然后用摘要替换原始内容但这会引入额外的计算开销。2.2 长期记忆智能体的知识库与经验库长期记忆是智能体的持久化存储用于保存跨越对话会话的重要信息例如用户画像、关键事实、执行任务的经验教训等。它的特点是容量大、保存时间长但检索是关键挑战。设计核心向量数据库Vector Database纯文本匹配检索在长期记忆中效率低下。例如用户说“我上次说的那家意大利餐厅”关键词匹配可能失效。向量数据库通过将文本转换为高维向量嵌入并计算向量间的相似度如余弦相似度来检索可以实现语义级别的模糊匹配。实现步骤记忆生成当判断一条信息需要长期保存时例如用户明确说“记住这个”或智能体自行判断为关键信息将其转换为文本片段。向量化使用嵌入模型如text-embedding-3-small将该文本转换为一个固定长度的向量。存储将向量、原始文本以及元数据如来源对话ID、时间戳、信息类型标签一起存入向量数据库。检索当需要从长期记忆中回忆时将当前查询或上下文也转换为向量然后在向量数据库中搜索最相似的K个记忆片段。# 示例使用ChromaDB轻量级向量数据库 import chromadb from chromadb.config import Settings from openai import OpenAI # 假设使用OpenAI Embedding模型 class LongTermMemory: def __init__(self, persist_directory./chroma_db): self.client chromadb.PersistentClient(pathpersist_directory) self.collection self.client.get_or_create_collection(nameagent_memories) self.embed_client OpenAI() # 初始化Embedding客户端 def _get_embedding(self, text): response self.embed_client.embeddings.create(modeltext-embedding-3-small, inputtext) return response.data[0].embedding def store_memory(self, text, metadata{}): 存储一条长期记忆 embedding self._get_embedding(text) # ChromaDB会自动生成ID这里我们用时间戳哈希生成唯一ID memory_id fmem_{int(time.time())}_{hash(text) % 10000:04d} self.collection.add( embeddings[embedding], documents[text], metadatas[metadata], ids[memory_id] ) def retrieve_memories(self, query_text, n_results3): 根据查询检索相关记忆 query_embedding self._get_embedding(query_text) results self.collection.query( query_embeddings[query_embedding], n_resultsn_results ) # results 包含 documents, metadatas, distances 等 return results[documents][0] if results[documents] else []实操心得元数据metadata的设计至关重要。除了时间戳建议加上entity实体如“用户偏好”、“餐厅信息”、importance重要性评分由LLM生成等字段。这样在检索时可以结合向量相似度和元数据过滤得到更精准的结果。例如可以优先检索与当前对话实体相关且重要性高的记忆。2.3 工作记忆任务执行的“草稿纸”工作记忆是智能体在执行当前任务时的临时信息暂存区。它不同于短期记忆的对话历史而是专注于一个任务目标的上下文。例如在完成“订机票-订酒店-租车”的复杂任务时工作记忆会暂存已查询到的航班信息、酒店选项等供后续步骤决策使用。实现模式工作记忆通常是一个结构化的、任务粒度的字典或对象。它在任务开始时被初始化在任务执行过程中被不断更新在任务完成后被清理或归档到长期记忆。class WorkingMemory: def __init__(self, task_id): self.task_id task_id self.data { goal: , # 任务目标 steps: [], # 已完成的步骤 current_step: , # 当前步骤 context: {}, # 任务相关上下文如收集到的参数 intermediate_results: {} # 中间结果如查询到的航班列表 } def update(self, key, value): 更新工作记忆中的特定字段 if key in self.data: self.data[key] value else: self.data[context][key] value def get(self, key): 获取工作记忆中的值 return self.data.get(key, self.data[context].get(key))三层记忆各司其职短期记忆维持对话流畅长期记忆形成持久认知工作记忆保障复杂任务执行。它们之间需要协同工作这引出了下一个核心问题记忆的流动与调度。3. 记忆的流动、检索与更新策略记忆不是静态存储的需要在三层之间智能地流动。同时如何从海量长期记忆中精准检索以及如何更新或遗忘记忆是工程上的难点。3.1 记忆的写入什么该记记在哪里并非所有信息都值得记忆。无差别存储会导致记忆库充满噪音。我们需要一个“记忆写入控制器”。策略建议规则触发定义明确规则。例如用户说“记住我喜欢喝茶”或“我的生日是7月10日”这类包含明确指令或关键个人事实的信息直接触发写入长期记忆。LLM判断更灵活的方式是让LLM充当“记忆过滤器”。在每轮对话后将对话片段交给一个轻量级LLM或使用主模型的一个特定提示让其判断是否值得存入长期记忆提取关键事实、用户偏好、承诺等信息类型是什么用户属性、事实知识、待办事项等重要性分数是多少1-10分 基于LLM的输出决定是否存储及如何打标签。def should_save_to_long_term(conversation_snippet, llm_client): prompt f 请分析以下对话片段判断是否包含值得智能体长期记住的重要信息。 对话{conversation_snippet} 请按JSON格式输出{{save: true/false, type: preference/fact/todo, key_info: 提取的关键信息, importance: 分数}}。 response llm_client.chat.completions.create( modelgpt-4o-mini, # 使用小模型降低成本 messages[{role: user, content: prompt}], response_format{type: json_object} ) decision json.loads(response.choices[0].message.content) return decision3.2 记忆的检索在正确的时间想起正确的事检索是记忆系统的灵魂。核心是相关性和时效性。混合检索策略向量检索语义相似性如上文所述是基础。它能找到语义上相关的内容。元数据过滤结合向量检索的结果用元数据如entity、recency进行二次筛选。例如当对话主题是“咖啡”时可以过滤entity包含“咖啡”或“饮品”的记忆。时间衰减加权更近的记忆通常更相关。可以在计算最终相关性分数时加入一个时间衰减因子让近期记忆的排名提升。LLM重排序Rerank对于初步检索到的Top N条记忆可以再用LLM根据当前对话的完整上下文对其相关性进行精细排序和去重。这是效果最好但成本也最高的方法。def retrieve_relevant_memories(current_query, lt_memory, st_memory, n5): 综合检索相关记忆 # 1. 从长期记忆进行向量检索 raw_memories lt_memory.retrieve_memories(current_query, n_resultsn*2) # 2. 构建包含短期记忆的完整上下文 full_context \n.join([f{m[role]}: {m[content]} for m in st_memory.get_context()[-5:]]) # 取最近5轮 # 3. (可选)LLM重排序与摘要 if raw_memories: prompt f 当前对话上下文 {full_context} 用户最新查询{current_query} 以下是从知识库中找到的一些可能相关的记忆片段 {chr(10).join([f{i1}. {mem} for i, mem in enumerate(raw_memories)])} 请根据当前对话选出最相关的1-3条记忆并简要说明理由。如果都不相关请回答“无”。 # 调用LLM进行判断... # 根据LLM输出选择最终的记忆 return filtered_memories3.3 记忆的更新与遗忘保持记忆的鲜活与准确信息会过时用户偏好会改变。一个只会增加不会修改的记忆系统最终会充满矛盾。更新机制冲突检测当要存入的新记忆与已有记忆在语义上高度相似但内容矛盾时例如旧记忆“用户喜欢狗”新记忆“用户对狗毛过敏”触发更新流程。LLM仲裁将冲突的记忆对和当前上下文交给LLM判断哪个信息更可能正确、更新或是可以合并。执行操作根据LLM的判断执行替换、合并或添加时间戳注释例如“截至2023年喜欢狗2024年起因过敏不再养狗”。遗忘压缩机制基于重要性的清理定期扫描长期记忆对重要性评分极低且很久未被访问的记忆进行归档或删除。总结性压缩对于关于同一主题的多次琐碎记忆例如多次提到“喜欢某家店的拿铁”可以定期让LLM生成一个总结性的记忆条目并删除原始琐碎条目。踩坑实录在早期版本中我们只实现了记忆的添加导致当用户说“我其实不喜欢吃辣了”之后智能体仍然会推荐川菜因为旧的“喜欢吃辣”记忆还在且被检索到。实现更新逻辑后我们让LLM判断“不喜欢吃辣了”是对“喜欢吃辣”的否定更新从而用新记忆替换了旧记忆的content字段并在元数据中记录了更新时间彻底解决了这个问题。4. 将记忆整合进智能体推理循环记忆系统不是孤立的模块它必须深度嵌入智能体的核心推理循环感知→思考→行动中。4.1 在感知阶段注入记忆在将用户输入Query提交给LLM进行核心思考规划、工具调用判断之前先进行记忆检索。class AgentWithMemory: def __init__(self): self.st_memory ShortTermMemory() self.lt_memory LongTermMemory() self.working_memory None def perceive(self, user_input): 感知阶段整合记忆 # 1. 检索相关长期记忆 relevant_memories retrieve_relevant_memories(user_input, self.lt_memory, self.st_memory) # 2. 获取短期记忆上下文 short_term_context self.st_memory.get_context() # 3. 获取工作记忆如果存在活跃任务 working_context self.working_memory.data if self.working_memory else {} # 4. 构建增强的提示词 enhanced_prompt self._construct_prompt(user_input, short_term_context, relevant_memories, working_context) return enhanced_prompt def _construct_prompt(self, query, short_term, long_term, working): 构建包含所有上下文的系统提示词 system_message f你是一个有帮助的智能体拥有以下记忆和上下文 关于用户的长期记忆 {chr(10).join(long_term) if long_term else 暂无相关长期记忆。} 当前任务的工作记忆 {json.dumps(working, indent2, ensure_asciiFalse) if working else 无活跃任务。} 最近的对话历史短期记忆 {chr(10).join([f{m[role]}: {m[content]} for m in short_term])} 请基于以上信息回应用户的最新输入{query} return system_message4.2 在思考与行动后更新记忆在LLM生成回复或执行工具调用后需要根据本轮交互的结果更新各个记忆层。def cycle(self, user_input): 智能体的完整交互循环 # 感知 prompt self.perceive(user_input) # 思考与行动调用LLM可能涉及工具调用 llm_response, used_tools_result self.think_and_act(prompt) # 记忆更新阶段 # 1. 更新短期记忆 self.st_memory.add_interaction(user, user_input) self.st_memory.add_interaction(assistant, llm_response) # 2. 判断是否存入长期记忆 conversation_snippet fUser: {user_input}\nAssistant: {llm_response} save_decision should_save_to_long_term(conversation_snippet, llm_client) if save_decision.get(save): self.lt_memory.store_memory( textsave_decision[key_info], metadata{type: save_decision[type], importance: save_decision[importance]} ) # 3. 更新工作记忆如果任务有进展 if self.working_memory and used_tools_result: self.working_memory.update(intermediate_results, used_tools_result) return llm_response4.3 工作记忆驱动任务分解对于复杂任务工作记忆是任务分解Chain-of-Thought和执行的协调中心。LLM根据任务目标存储在working_memory[“goal”]和已有上下文working_memory[“context”]规划下一步更新工作记忆然后执行。5. 实战中的性能优化与常见陷阱实现一个可用的记忆系统是一回事实现一个高效、稳定的记忆系统是另一回事。以下是几个关键的优化点和避坑指南。5.1 向量检索的精度与效率平衡问题直接使用用户当前查询语句进行向量检索可能不够准确特别是查询很短或很模糊时。解决方案查询扩展。在检索前先用LLM对当前查询和对话上下文进行简要分析生成一个更丰富、包含潜在语义的搜索查询语句。def expand_query_for_retrieval(user_query, conversation_context): prompt f基于以下对话历史和最新问题生成一个更全面、用于在知识库中搜索相关信息的查询语句。 历史 {conversation_context} 最新问题{user_query} 生成的搜索查询 expanded_query llm_client.chat.completions.create(...) # 调用小模型 return expanded_query # 然后用 expanded_query 去检索长期记忆5.2 记忆token消耗与成本控制问题记忆内容尤其是长期记忆检索结果会占用大量LLM上下文Token显著增加API调用成本。解决方案记忆摘要在将长期记忆插入上下文前如果记忆文本过长让LLM先对其进行摘要。选择性注入不是所有检索到的记忆都放入上下文。可以设定一个相关性分数阈值或只选择Top 1-2条最相关的记忆。分层提示将最核心的指令和最近几轮对话放在提示词最前面将检索到的记忆放在靠后位置。有些模型对靠前的内容关注度更高。5.3 记忆幻觉与冲突解决问题LLM可能会混淆记忆来源甚至“脑补”出不存在于记忆库中的细节幻觉。或者记忆库中存在多条相互矛盾的信息。解决方案明确引用来源在向LLM提供记忆时明确标注来源例如“[记忆-2024-01-01]用户说喜欢蓝色”。这能一定程度上减少幻觉。设置置信度为每条记忆附加一个置信度分数基于来源可靠性、确认次数等。在出现冲突时优先采用置信度高的记忆。主动询问当冲突无法通过规则解决时最稳妥的方式是让智能体向用户主动确认。“我记得您之前提过喜欢狗但刚才又提到对狗毛过敏请问您目前的偏好是怎样的”5.4 长期记忆的初始化与冷启动问题新智能体长期记忆为空无法进行个性化服务影响初期体验。解决方案预制知识根据智能体的定位预置一些通用知识或领域常识。例如一个咖啡推荐机器人可以预置常见咖啡种类、口味特点等记忆。主动引导在初次交互时智能体可以主动询问一些关键信息来填充记忆例如“为了更好的为您服务可以告诉我您喜欢的咖啡口味吗例如偏苦/偏酸加奶/不加奶”。从交互中快速学习设计机制让智能体在头几次交互中以更高的频率和更低的阈值将信息存入长期记忆快速构建用户画像。构建记忆系统的过程是让智能体真正“活”起来的关键一步。它从本质上改变了智能体的交互模式从一次性的问答变成了持续的、有历史感的陪伴与合作。这套系统目前运行在我们的多个智能体项目中显著提升了任务完成率和用户满意度。当然记忆系统本身也是一个持续学习和优化的过程下一步我们正在探索如何让记忆之间产生更复杂的关联形成真正的“知识图谱”但那将是另一个更深入的话题了。