记忆检索与更新，Agent什么时候该回忆什么时候该遗忘

记忆检索与更新，Agent什么时候该回忆什么时候该遗忘

记忆系统做大了以后，会遇到一个矛盾。记的东西越多，检索越慢，噪音越多。记的东西太少，又不够用。

关键不在于记多少，在于该记的记住了、该忘的忘掉了、需要的时候能找到。

这一篇我们讲记忆的检索和更新策略。怎么高效地回忆，怎么合理地遗忘，怎么保持记忆库的整洁。

记忆检索策略

记忆存在向量数据库里，检索的时候怎么做。

最简单的方式是，每次用户提问，都去记忆库里搜一遍。但这样有两个问题。一是每次都搜，浪费时间和Token。二是搜回来的记忆不一定有用，反而干扰Agent的回答。

更好的做法是，判断一下当前问题是否需要回忆。

什么情况需要回忆。

用户提到了之前聊过的内容。“上次你说的那个方案”、“我们之前讨论的”。这类明确的回溯信号，需要回忆。

用户的问题涉及个人偏好或历史决策。“我适合学什么”、“帮我推荐一个框架”。这类问题需要结合用户画像和历史记忆来回答。

Agent的回答跟上下文不连贯。可能是丢了之前的上下文，需要从记忆里找回来。

什么情况不需要回忆。

通用知识问题。“Python怎么写冒泡排序”。这种问题不需要个人记忆，大模型自己就能回答。

全新的话题。用户开始聊一个之前没聊过的话题，旧记忆没有相关内容，不用回忆。

classSmartRecall:"""智能回忆策略"""def__init__(self,llm,memory_store):self.llm=llm self.memory=memory_store self.judge_prompt="""判断以下用户问题是否需要从历史记忆中检索相关信息。 判断标准： - 需要检索：提到之前的内容、涉及个人偏好、需要历史决策信息 - 不需要检索：通用知识问题、全新话题、纯技术问题 用户问题：{question} 是否需要检索记忆？只回答"是"或"否"。"""defmaybe_recall(self,question):"""根据问题判断是否需要回忆"""# 先用简单规则过滤recall_keywords=["上次","之前","我记得","我们讨论","你说的","我的偏好","帮我推荐"]needs_recall=any(kwinquestionforkwinrecall_keywords)ifnotneeds_recall:# 规则没命中，用大模型判断response=self.llm.invoke(self.judge_prompt.format(question=question))needs_recall="是"inresponse.contentifneeds_recall:memories=self.memory.recall(question,k=3)returnmemorieselse:return[]

这个智能回忆策略能减少不必要的检索，提升响应速度。

检索结果排序

记忆检索回来以后，怎么排序。

向量相似度是一个维度，但不是唯一的。一条记忆的价值，除了相似度，还要看时效性、重要程度、使用频率。

时效性。近期的记忆通常比旧记忆更有参考价值。用户一个月前说在学Python，现在可能已经在学Agent了。时间越久，权重越低。

重要程度。用户的核心信息比临时信息重要。用户的职业和技能水平，比某次闲聊的内容重要得多。可以给不同类型的记忆打上重要程度标签。

使用频率。被频繁检索到的记忆，说明它经常有用，应该排在前面。很久没被检索到的记忆，可能已经不太相关了。

fromdatetimeimportdatetime,timedeltaclassWeightedRecall:"""加权回忆"""def__init__(self,memory_store):self.memory=memory_store self.access_count={}# 记录每条记忆的访问次数defrecall(self,query,k=5):"""加权检索记忆"""# 先多检索一些raw_results=self.memory.similarity_search_with_score(query,k=k*2)weighted=[]fordoc,distanceinraw_results:# 相似度分数（距离越小越相似）similarity=1/(1+distance)# 时间衰减timestamp=doc.metadata.get("timestamp","")iftimestamp:created=datetime.fromisoformat(timestamp)age_days=(datetime.now()-created).days time_weight=max(0.3,1-age_days/180)# 半年衰减到0.3else:time_weight=0.5# 重要程度importance=doc.metadata.get("importance",0.5)# 使用频率content_hash=hash(doc.page_content)freq=self.access_count.get(content_hash,0)freq_weight=min(1.0,0.5+freq*0.1)# 综合分数final_score=similarity*0.5+time_weight*0.2+importance*0.2+freq_weight*0.1weighted.append((doc,final_score))# 排序并取前k个weighted.sort(key=lambdax:x[1],reverse=True)# 更新访问次数fordoc,_inweighted[:k]:content_hash=hash(doc.page_content)self.access_count[content_hash]=self.access_count.get(content_hash,0)+1return[docfordoc,_inweighted[:k]]

记忆遗忘机制

记忆不是越多越好。记太多旧的、过时的、不相关的记忆，会干扰检索质量。需要有遗忘机制。

什么时候该遗忘。

信息过时了。用户之前说"在学Python"，现在已经是Python高手了。旧的"在学Python"这条记忆就应该被更新或删除。

信息重复了。同一条信息被存了多次。保留一条就行，其余删除。

信息不再相关了。用户换工作了、换技术栈了、兴趣转移了。跟旧状态相关的记忆不再有用。

长期没被检索到。一条记忆如果半年都没被检索到过，大概率不太重要。可以考虑清理。

遗忘的几种方式。

硬删除。直接从数据库里删掉。简单粗暴，不可恢复。适合明确无用的信息。

软删除。标记为"已过期"或"低优先级"，检索时降低权重或跳过。保留数据但不参与检索。适合不确定是否该删的信息。

合并压缩。把多条相关的旧记忆合并成一条摘要。信息保留但体积变小。适合同一主题的多条记忆。

classMemoryMaintenance:"""记忆维护和遗忘"""def__init__(self,memory_store,llm):self.memory=memory_store self.llm=llmdefcleanup_expired(self,days=180):"""清理过期的记忆"""cutoff=datetime.now()-timedelta(days=days)# 获取所有记忆all_docs=self.memory.get()deleted=0fordocinall_docs:timestamp=doc.metadata.get("timestamp","")iftimestamp:created=datetime.fromisoformat(timestamp)ifcreated<cutoff:# 超过阈值的记忆，检查重要程度importance=doc.metadata.get("importance",0.5)ifimportance<0.7:# 不太重要的直接删self.memory.delete(doc.metadata.get("id"))deleted+=1print(f"清理了{deleted}条过期记忆")defmerge_similar(self):"""合并相似的记忆"""all_docs=self.memory.get()# 找出内容相似的记忆对fori,doc1inenumerate(all_docs):fordoc2inall_docs[i+1:]:# 用简单的文本相似度判断ifself._is_similar(doc1.page_content,doc2.page_content):# 用大模型合并merged=self._merge_memories(doc1.page_content,doc2.page_content)# 删除旧的两条，存入合并后的self.memory.delete(doc1.metadata.get("id"))self.memory.delete(doc2.metadata.get("id"))self.memory.add_text(merged)print("记忆合并完成")def_merge_memories(self,mem1,mem2):"""用大模型合并两条记忆"""prompt=f"""请合并以下两条记忆，保留所有重要信息，去掉重复内容。 只输出合并后的内容，不要其他文字。 记忆1：{mem1}记忆2：{mem2}"""response=self.llm.invoke(prompt)returnresponse.content

记忆的定期维护

记忆系统需要定期维护，就像定期整理电脑文件一样。

建议设置一个定期任务，比如每周或每月跑一次。

维护内容包括清理过期记忆、合并重复记忆、更新用户画像、统计记忆使用情况。

defweekly_maintenance(memory_store,llm):"""每周记忆维护"""print("开始每周记忆维护...")# 1. 清理过期记忆maintenance=MemoryMaintenance(memory_store,llm)maintenance.cleanup_expired(days=90)# 2. 合并相似记忆maintenance.merge_similar()# 3. 统计记忆使用情况stats={"total_memories":memory_store.count(),"avg_access_count":sum(maintenance.access_count.values())/max(len(maintenance.access_count),1),"never_accessed":sum(1forvinmaintenance.access_count.values()ifv==0),}print(f"记忆总数：{stats['total_memories']}")print(f"平均访问次数：{stats['avg_access_count']:.1f}")print(f"从未被访问的记忆数：{stats['never_accessed']}")print("维护完成")

一个实际的记忆管理策略

给一个实用的记忆管理策略，可以直接参考。

记忆分层。把记忆分成三层。热记忆是最近7天的，检索权重最高。温记忆是7到90天的，正常权重。冷记忆是90天以上的，降低权重。

按需回忆。用智能回忆策略，不是每次都检索记忆。有回溯信号才检索。

定期维护。每周清理过期记忆，每月合并相似记忆。保持记忆库的整洁。

重要程度标记。存入记忆时标记重要程度。核心信息高权重，临时信息低权重。遗忘时先忘低权重的。

下一篇是记忆篇的最后一篇。我们做一个完整的记忆系统实战项目，把短期记忆、长期记忆、用户画像整合在一起。