AI_Agent记忆系统设计与实现

AI Agent记忆系统设计与实现：从短期到长期记忆的完整方案

让你的Agent拥有"记忆"，实现真正的智能对话

前言

你有没有遇到过这种情况：跟AI助手聊了半天，换了个话题再回来，它就完全忘了之前聊过什么？这就是记忆缺失的问题。

一个真正的AI Agent需要三种记忆：

• 工作记忆：当前对话的上下文
• 短期记忆：最近几次对话的摘要
• 长期记忆：用户偏好、历史知识、重要事实

本文将手把手教你设计和实现一个完整的Agent记忆系统。

你将学到：

• 记忆系统的架构设计
• 对话历史管理策略
• 长期记忆存储与检索
• 记忆压缩与摘要技术
• 个性化记忆应用

一、记忆系统架构

1.1 三层记忆模型

fromdataclassesimportdataclassfromtypingimportList,Dict,Optionalfromdatetimeimportdatetimeimportjson@dataclassclassMemory:"""记忆单元"""content:str# 记忆内容memory_type:str# 类型：working/short/longtimestamp:datetime# 创建时间importance:float# 重要性评分 0-1access_count:int# 访问次数metadata:Dict# 元数据classMemorySystem:"""三层记忆系统"""def__init__(self,max_working=20,max_short=100,max_long=10000):self.working_memory:List[Memory]=[]# 工作记忆（当前对话）self.short_term:List[Memory]=[]# 短期记忆（最近对话）self.long_term:List[Memory]=[]# 长期记忆（持久化）self.max_working=max_working self.max_short=max_short self.max_long=max_longdefadd_memory(self,content:str,memory_type:str,importance:float=0.5,metadata:dict=None):"""添加记忆"""memory=Memory(content=content,memory_type=memory_type,timestamp=datetime.now(),importance=importance,access_count=0,metadata=metadataor{})ifmemory_type=="working":self.working_memory.append(memory)iflen(self.working_memory)>self.max_working:self._consolidate_working()elifmemory_type=="short":self.short_term.append(memory)iflen(self.short_term)>self.max_short:self._consolidate_short()elifmemory_type=="long":self.long_term.append(memory)iflen(self.long_term)>self.max_long:self._forget_long()def_consolidate_working(self):"""工作记忆整合到短期记忆"""# 将旧的工作记忆转为摘要存入短期记忆old_memories=self.working_memory[:10]summary=self._summarize_memories(old_memories)self.add_memory(content=summary,memory_type="short",importance=0.6,metadata={"type":"conversation_summary"})self.working_memory=self.working_memory[10:]def_consolidate_short(self):"""短期记忆整合到长期记忆"""# 找出高重要性的短期记忆important=[mforminself.short_termifm.importance>0.7]formeminimportant:self.add_memory(content=mem.content,memory_type="long",importance=mem.importance,metadata=mem.metadata)# 清理低重要性的短期记忆self.short_term=[mforminself.short_termifm.importance>0.5]def_forget_long(self):"""长期记忆遗忘策略"""# 按重要性和访问频率排序self.long_term.sort(key=lambdam:m.importance*(1+m.access_count*0.1),reverse=True)# 保留前80%keep_count=int(self.max_long*0.8)self.long_term=self.long_term[:keep_count]def_summarize_memories(self,memories:List[Memory])->str:"""将多条记忆压缩成摘要"""# 实际项目中会调用LLM生成摘要contents=[m.contentforminmemories]return"对话摘要："+"；".join(contents[:3])+"..."defretrieve(self,query:str,top_k:int=5)->List[Memory]:"""检索相关记忆"""all_memories=self.working_memory+self.short_term+self.long_term# 简单的关键词匹配（实际用向量检索）scored=[]formeminall_memories:score=self._calculate_relevance(query,mem)scored.append((score,mem))scored.sort(key=lambdax:x[0],reverse=True)return[memfor_,meminscored[:top_k]]def_calculate_relevance(self,query:str,memory:Memory)->float:"""计算相关性分数"""# 简单实现：关键词匹配 + 时间衰减 + 重要性keyword_score=sum(1forwordinqueryifwordinmemory.content)time_decay=1.0/(1+(datetime.now()-memory.timestamp).seconds/3600)returnkeyword_score*time_decay*memory.importance

1.2 记忆生命周期

classMemoryLifecycle:"""记忆生命周期管理"""STAGES={"encoding":"编码：将信息转化为可存储的形式","storage":"存储：保存到相应的记忆层","consolidation":"整合：短期→长期的转化","retrieval":"检索：根据需要提取记忆","forgetting":"遗忘：删除不重要的记忆"}defprocess_memory(self,content:str,context:dict)->Memory:"""处理一条新记忆"""# 1. 编码：提取关键信息encoded=self.encode(content,context)# 2. 评估重要性importance=self.evaluate_importance(encoded)# 3. 决定存储位置memory_type=self.decide_storage(importance,context)# 4. 存储memory=self.store(encoded,memory_type,importance)returnmemorydefencode(self,content:str,context:dict)->dict:"""编码：提取关键信息"""return{"content":content,"entities":self.extract_entities(content),"intent":self.detect_intent(content),"sentiment":self.analyze_sentiment(content),"context":context}defevaluate_importance(self,encoded:dict)->float:"""评估重要性"""score=0.5# 包含实体信息更重要ifencoded["entities"]:score+=0.2# 用户偏好很重要if"偏好"inencoded["content"]or"喜欢"inencoded["content"]:score+=0.3# 情绪强烈的内容更重要ifabs(encoded["sentiment"])>0.5:score+=0.1returnmin(score,1.0)defdecide_storage(self,importance:float,context:dict)->str:"""决定存储位置"""ifimportance>0.8:return"long"elifimportance>0.4:return"short"else:return"working"

二、对话历史管理

2.1 滑动窗口策略

fromcollectionsimportdequefromtypingimportTupleclassConversationWindow:"""滑动窗口管理对话历史"""def__init__(self,max_messages:int=20):self.messages=deque(maxlen=max_messages)self.summary=""defadd_message(self,role:str