记忆检索与更新Agent什么时候该回忆什么时候该遗忘记忆系统做大了以后会遇到一个矛盾。记的东西越多检索越慢噪音越多。记的东西太少又不够用。关键不在于记多少在于该记的记住了、该忘的忘掉了、需要的时候能找到。这一篇我们讲记忆的检索和更新策略。怎么高效地回忆怎么合理地遗忘怎么保持记忆库的整洁。记忆检索策略记忆存在向量数据库里检索的时候怎么做。最简单的方式是每次用户提问都去记忆库里搜一遍。但这样有两个问题。一是每次都搜浪费时间和Token。二是搜回来的记忆不一定有用反而干扰Agent的回答。更好的做法是判断一下当前问题是否需要回忆。什么情况需要回忆。用户提到了之前聊过的内容。“上次你说的那个方案”、“我们之前讨论的”。这类明确的回溯信号需要回忆。用户的问题涉及个人偏好或历史决策。“我适合学什么”、“帮我推荐一个框架”。这类问题需要结合用户画像和历史记忆来回答。Agent的回答跟上下文不连贯。可能是丢了之前的上下文需要从记忆里找回来。什么情况不需要回忆。通用知识问题。“Python怎么写冒泡排序”。这种问题不需要个人记忆大模型自己就能回答。全新的话题。用户开始聊一个之前没聊过的话题旧记忆没有相关内容不用回忆。classSmartRecall:智能回忆策略def__init__(self,llm,memory_store):self.llmllm self.memorymemory_store self.judge_prompt判断以下用户问题是否需要从历史记忆中检索相关信息。 判断标准 - 需要检索提到之前的内容、涉及个人偏好、需要历史决策信息 - 不需要检索通用知识问题、全新话题、纯技术问题 用户问题{question} 是否需要检索记忆只回答是或否。defmaybe_recall(self,question):根据问题判断是否需要回忆# 先用简单规则过滤recall_keywords[上次,之前,我记得,我们讨论,你说的,我的偏好,帮我推荐]needs_recallany(kwinquestionforkwinrecall_keywords)ifnotneeds_recall:# 规则没命中用大模型判断responseself.llm.invoke(self.judge_prompt.format(questionquestion))needs_recall是inresponse.contentifneeds_recall:memoriesself.memory.recall(question,k3)returnmemorieselse:return[]这个智能回忆策略能减少不必要的检索提升响应速度。检索结果排序记忆检索回来以后怎么排序。向量相似度是一个维度但不是唯一的。一条记忆的价值除了相似度还要看时效性、重要程度、使用频率。时效性。近期的记忆通常比旧记忆更有参考价值。用户一个月前说在学Python现在可能已经在学Agent了。时间越久权重越低。重要程度。用户的核心信息比临时信息重要。用户的职业和技能水平比某次闲聊的内容重要得多。可以给不同类型的记忆打上重要程度标签。使用频率。被频繁检索到的记忆说明它经常有用应该排在前面。很久没被检索到的记忆可能已经不太相关了。fromdatetimeimportdatetime,timedeltaclassWeightedRecall:加权回忆def__init__(self,memory_store):self.memorymemory_store self.access_count{}# 记录每条记忆的访问次数defrecall(self,query,k5):加权检索记忆# 先多检索一些raw_resultsself.memory.similarity_search_with_score(query,kk*2)weighted[]fordoc,distanceinraw_results:# 相似度分数距离越小越相似similarity1/(1distance)# 时间衰减timestampdoc.metadata.get(timestamp,)iftimestamp:createddatetime.fromisoformat(timestamp)age_days(datetime.now()-created).days time_weightmax(0.3,1-age_days/180)# 半年衰减到0.3else:time_weight0.5# 重要程度importancedoc.metadata.get(importance,0.5)# 使用频率content_hashhash(doc.page_content)freqself.access_count.get(content_hash,0)freq_weightmin(1.0,0.5freq*0.1)# 综合分数final_scoresimilarity*0.5time_weight*0.2importance*0.2freq_weight*0.1weighted.append((doc,final_score))# 排序并取前k个weighted.sort(keylambdax:x[1],reverseTrue)# 更新访问次数fordoc,_inweighted[:k]:content_hashhash(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.memorymemory_store self.llmllmdefcleanup_expired(self,days180):清理过期的记忆cutoffdatetime.now()-timedelta(daysdays)# 获取所有记忆all_docsself.memory.get()deleted0fordocinall_docs:timestampdoc.metadata.get(timestamp,)iftimestamp:createddatetime.fromisoformat(timestamp)ifcreatedcutoff:# 超过阈值的记忆检查重要程度importancedoc.metadata.get(importance,0.5)ifimportance0.7:# 不太重要的直接删self.memory.delete(doc.metadata.get(id))deleted1print(f清理了{deleted}条过期记忆)defmerge_similar(self):合并相似的记忆all_docsself.memory.get()# 找出内容相似的记忆对fori,doc1inenumerate(all_docs):fordoc2inall_docs[i1:]:# 用简单的文本相似度判断ifself._is_similar(doc1.page_content,doc2.page_content):# 用大模型合并mergedself._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):用大模型合并两条记忆promptf请合并以下两条记忆保留所有重要信息去掉重复内容。 只输出合并后的内容不要其他文字。 记忆1{mem1}记忆2{mem2}responseself.llm.invoke(prompt)returnresponse.content记忆的定期维护记忆系统需要定期维护就像定期整理电脑文件一样。建议设置一个定期任务比如每周或每月跑一次。维护内容包括清理过期记忆、合并重复记忆、更新用户画像、统计记忆使用情况。defweekly_maintenance(memory_store,llm):每周记忆维护print(开始每周记忆维护...)# 1. 清理过期记忆maintenanceMemoryMaintenance(memory_store,llm)maintenance.cleanup_expired(days90)# 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()ifv0),}print(f记忆总数{stats[total_memories]})print(f平均访问次数{stats[avg_access_count]:.1f})print(f从未被访问的记忆数{stats[never_accessed]})print(维护完成)一个实际的记忆管理策略给一个实用的记忆管理策略可以直接参考。记忆分层。把记忆分成三层。热记忆是最近7天的检索权重最高。温记忆是7到90天的正常权重。冷记忆是90天以上的降低权重。按需回忆。用智能回忆策略不是每次都检索记忆。有回溯信号才检索。定期维护。每周清理过期记忆每月合并相似记忆。保持记忆库的整洁。重要程度标记。存入记忆时标记重要程度。核心信息高权重临时信息低权重。遗忘时先忘低权重的。下一篇是记忆篇的最后一篇。我们做一个完整的记忆系统实战项目把短期记忆、长期记忆、用户画像整合在一起。 记忆检索与更新Agent什么时候该回忆什么时候该遗忘记忆系统做大了以后会遇到一个矛盾。记的东西越多检索越慢噪音越多。记的东西太少又不够用。关键不在于记多少在于该记的记住了、该忘的忘掉了、需要的时候能找到。这一篇我们讲记忆的检索和更新策略。怎么高效地回忆怎么合理地遗忘怎么保持记忆库的整洁。记忆检索策略记忆存在向量数据库里检索的时候怎么做。最简单的方式是每次用户提问都去记忆库里搜一遍。但这样有两个问题。一是每次都搜浪费时间和Token。二是搜回来的记忆不一定有用反而干扰Agent的回答。更好的做法是判断一下当前问题是否需要回忆。什么情况需要回忆。用户提到了之前聊过的内容。“上次你说的那个方案”、“我们之前讨论的”。这类明确的回溯信号需要回忆。用户的问题涉及个人偏好或历史决策。“我适合学什么”、“帮我推荐一个框架”。这类问题需要结合用户画像和历史记忆来回答。Agent的回答跟上下文不连贯。可能是丢了之前的上下文需要从记忆里找回来。什么情况不需要回忆。通用知识问题。“Python怎么写冒泡排序”。这种问题不需要个人记忆大模型自己就能回答。全新的话题。用户开始聊一个之前没聊过的话题旧记忆没有相关内容不用回忆。classSmartRecall:智能回忆策略def__init__(self,llm,memory_store):self.llmllm self.memorymemory_store self.judge_prompt判断以下用户问题是否需要从历史记忆中检索相关信息。 判断标准 - 需要检索提到之前的内容、涉及个人偏好、需要历史决策信息 - 不需要检索通用知识问题、全新话题、纯技术问题 用户问题{question} 是否需要检索记忆只回答是或否。defmaybe_recall(self,question):根据问题判断是否需要回忆# 先用简单规则过滤recall_keywords[上次,之前,我记得,我们讨论,你说的,我的偏好,帮我推荐]needs_recallany(kwinquestionforkwinrecall_keywords)ifnotneeds_recall:# 规则没命中用大模型判断responseself.llm.invoke(self.judge_prompt.format(questionquestion))needs_recall是inresponse.contentifneeds_recall:memoriesself.memory.recall(question,k3)returnmemorieselse:return[]这个智能回忆策略能减少不必要的检索提升响应速度。检索结果排序记忆检索回来以后怎么排序。向量相似度是一个维度但不是唯一的。一条记忆的价值除了相似度还要看时效性、重要程度、使用频率。时效性。近期的记忆通常比旧记忆更有参考价值。用户一个月前说在学Python现在可能已经在学Agent了。时间越久权重越低。重要程度。用户的核心信息比临时信息重要。用户的职业和技能水平比某次闲聊的内容重要得多。可以给不同类型的记忆打上重要程度标签。使用频率。被频繁检索到的记忆说明它经常有用应该排在前面。很久没被检索到的记忆可能已经不太相关了。fromdatetimeimportdatetime,timedeltaclassWeightedRecall:加权回忆def__init__(self,memory_store):self.memorymemory_store self.access_count{}# 记录每条记忆的访问次数defrecall(self,query,k5):加权检索记忆# 先多检索一些raw_resultsself.memory.similarity_search_with_score(query,kk*2)weighted[]fordoc,distanceinraw_results:# 相似度分数距离越小越相似similarity1/(1distance)# 时间衰减timestampdoc.metadata.get(timestamp,)iftimestamp:createddatetime.fromisoformat(timestamp)age_days(datetime.now()-created).days time_weightmax(0.3,1-age_days/180)# 半年衰减到0.3else:time_weight0.5# 重要程度importancedoc.metadata.get(importance,0.5)# 使用频率content_hashhash(doc.page_content)freqself.access_count.get(content_hash,0)freq_weightmin(1.0,0.5freq*0.1)# 综合分数final_scoresimilarity*0.5time_weight*0.2importance*0.2freq_weight*0.1weighted.append((doc,final_score))# 排序并取前k个weighted.sort(keylambdax:x[1],reverseTrue)# 更新访问次数fordoc,_inweighted[:k]:content_hashhash(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.memorymemory_store self.llmllmdefcleanup_expired(self,days180):清理过期的记忆cutoffdatetime.now()-timedelta(daysdays)# 获取所有记忆all_docsself.memory.get()deleted0fordocinall_docs:timestampdoc.metadata.get(timestamp,)iftimestamp:createddatetime.fromisoformat(timestamp)ifcreatedcutoff:# 超过阈值的记忆检查重要程度importancedoc.metadata.get(importance,0.5)ifimportance0.7:# 不太重要的直接删self.memory.delete(doc.metadata.get(id))deleted1print(f清理了{deleted}条过期记忆)defmerge_similar(self):合并相似的记忆all_docsself.memory.get()# 找出内容相似的记忆对fori,doc1inenumerate(all_docs):fordoc2inall_docs[i1:]:# 用简单的文本相似度判断ifself._is_similar(doc1.page_content,doc2.page_content):# 用大模型合并mergedself._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):用大模型合并两条记忆promptf请合并以下两条记忆保留所有重要信息去掉重复内容。 只输出合并后的内容不要其他文字。 记忆1{mem1}记忆2{mem2}responseself.llm.invoke(prompt)returnresponse.content记忆的定期维护记忆系统需要定期维护就像定期整理电脑文件一样。建议设置一个定期任务比如每周或每月跑一次。维护内容包括清理过期记忆、合并重复记忆、更新用户画像、统计记忆使用情况。defweekly_maintenance(memory_store,llm):每周记忆维护print(开始每周记忆维护...)# 1. 清理过期记忆maintenanceMemoryMaintenance(memory_store,llm)maintenance.cleanup_expired(days90)# 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()ifv0),}print(f记忆总数{stats[total_memories]})print(f平均访问次数{stats[avg_access_count]:.1f})print(f从未被访问的记忆数{stats[never_accessed]})print(维护完成)一个实际的记忆管理策略给一个实用的记忆管理策略可以直接参考。记忆分层。把记忆分成三层。热记忆是最近7天的检索权重最高。温记忆是7到90天的正常权重。冷记忆是90天以上的降低权重。按需回忆。用智能回忆策略不是每次都检索记忆。有回溯信号才检索。定期维护。每周清理过期记忆每月合并相似记忆。保持记忆库的整洁。重要程度标记。存入记忆时标记重要程度。核心信息高权重临时信息低权重。遗忘时先忘低权重的。下一篇是记忆篇的最后一篇。我们做一个完整的记忆系统实战项目把短期记忆、长期记忆、用户画像整合在一起。 
