当前位置：首页 > news >正文

记忆系统实战，构建一个越聊越懂你的个性化助手

news 2026/8/7 13:14:19

记忆系统实战，构建一个越聊越懂你的个性化助手

记忆篇讲了很多理论和组件。这一篇我们把它们整合起来，做一个完整的个性化助手。

这个助手有短期记忆、长期记忆、用户画像，能记住你是谁、你喜欢什么、之前聊过什么。越用越懂你。

项目目标

做一个个性化AI助手，具备以下能力。

记住当前对话的上下文，支持多轮对话。记住用户的基本信息和偏好，跨对话保持记忆。能智能回忆相关信息，不需要每次都翻全部记忆。对话结束时自动提取重要信息存入长期记忆。支持多用户，每个用户有独立的记忆和画像。

系统架构

整个系统分四层。

对话层。管理当前对话的上下文，处理用户输入和Agent输出。

记忆层。短期记忆管理当前对话，长期记忆管理跨对话信息，用户画像管理用户偏好。

智能层。智能回忆判断、信息提取、记忆维护。

存储层。向量数据库存长期记忆，JSON文件存用户画像。

完整代码实现

fromdatetimeimportdatetimefrompathlibimportPathimportjsonfromlangchain_openaiimportChatOpenAI,OpenAIEmbeddingsfromlangchain_community.vectorstoresimportChromafromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.messagesimportSystemMessage,HumanMessage,AIMessageclassShortTermMemory:"""短期记忆：管理当前对话"""def__init__(self,max_messages=20):self.messages=[]self.max_messages=max_messagesdefadd_user_message(self,content):self.messages.append(HumanMessage(content=content))self._trim()defadd_ai_message(self,content):self.messages.append(AIMessage(content=content))self._trim()def_trim(self):iflen(self.messages)>self.max_messages:self.messages=self.messages[-self.max_messages:]defget_messages(self):returnself.messagesdefget_conversation_text(self):lines=[]formsginself.messages:role="用户"ifisinstance(msg,HumanMessage)else"助手"lines.append(f"{role}:{msg.content}")return"\n".join(lines)defclear(self):self.messages=[]classLongTermMemory:"""长期记忆：跨对话的持久记忆"""def__init__(self,db_path,user_id):self.embeddings=OpenAIEmbeddings()self.store=Chroma(persist_directory=f"{db_path}/user_{user_id}",embedding_function=self.embeddings,)self.access_count={}defsave(self,content,importance=0.5):"""保存一条记忆"""metadata={"timestamp":datetime.now().isoformat(),"importance":importance,}self.store.add_texts(texts=[content],metadatas=[metadata])self.store.persist()defrecall(self,query,k=3):"""回忆相关记忆"""results=self.store.similarity_search(query,k=k)# 更新访问计数fordocinresults:key=hash(doc.page_content)self.access_count[key]=self.access_count.get(key,0)+1return[doc.page_contentfordocinresults]defcount(self):returnself.store._collection.count()classUserProfile:"""用户画像"""def__init__(self,storage_dir,user_id,llm):self.storage_dir=Path(storage_dir)self.storage_dir.mkdir(exist_ok=True)self.user_id=user_id self.llm=llm self.filepath=self.storage_dir/f"{user_id}_profile.json"self.profile=self._load()def_load(self):ifself.filepath.exists():withopen(self.filepath,"r",encoding="utf-8")asf:returnjson.load(f)return{"basic_info":{},"preferences":{},"knowledge_level":{},"topics":[],"last_updated":"",}defsave(self):self.profile["last_updated"]=datetime.now().isoformat()withopen(self.filepath,"w",encoding="utf-8")asf:json.dump(self.profile,f,ensure_ascii=False,indent=2)defupdate_from_conversation(self,conversation_text):"""从对话更新画像"""prompt=f"""请分析对话，提取用户信息并更新画像。 当前画像：{json.dumps(self.profile,ensure_ascii=False,indent=2)}对话：{conversation_text}请输出更新后的完整画像，JSON格式。只输出JSON。"""response=self.llm.invoke(prompt)try:new_profile=json.loads(response.content)self.profile=new_profile self.save()returnTrueexceptjson.JSONDecodeError:returnFalsedefget_summary(self):"""获取画像摘要"""parts=[]ifself.profile.get("basic_info"):info=self.profile["basic_info"]parts.append(f"用户信息：{json.dumps(info,ensure_ascii=False)}")ifself.profile.get("preferences"):prefs=self.profile["preferences"]parts.append(f"偏好：{json.dumps(prefs,ensure_ascii=False)}")ifself.profile.get("knowledge_level"):levels=self.profile["knowledge_level"]parts.append(f"知识水平：{json.dumps(levels,ensure_ascii=False)}")return"\n".join(parts)ifpartselse"暂无用户画像"classPersonalAssistant:"""个性化助手：整合短期记忆、长期记忆和用户画像"""def__init__(self,user_id="default",storage_dir="./assistant_data"):self.llm=ChatOpenAI(model="gpt-3.5-turbo",temperature=0)self.user_id=user_id# 初始化三个记忆组件self.short_memory=ShortTermMemory(max_messages=20)self.long_memory=LongTermMemory(db_path=f"{storage_dir}/memory",user_id=user_id,)self.profile=UserProfile(storage_dir=f"{storage_dir}/profiles",user_id=user_id,llm=self.llm,)self.system_prompt="""你是一个个性化AI助手。请根据用户画像和历史记忆来提供个性化服务。 用户画像： {user_profile} 相关记忆： {memories} 注意事项： 1. 根据用户画像调整回答风格和技术选型 2. 参考历史记忆中的信息，保持对话连续性 3. 如果画像或记忆中没有相关信息，正常回答即可 4. 不要提及"画像"或"记忆"这些内部机制"""defchat(self,user_input):"""用户对话"""# 1. 智能回忆memories=self._smart_recall(user_input)memory_text="\n".join(f"-{m}"forminmemories)ifmemorieselse"无"# 2. 获取用户画像profile_text=self.profile.get_summary()# 3. 构建系统提示system_content=self.system_prompt.format(user_profile=profile_text,memories=memory_text,)# 4. 构建消息列表messages=[SystemMessage(content=system_content)]messages.extend(self.short_memory.get_messages())messages.append(HumanMessage(content=user_input))# 5. 调用大模型response=self.llm.invoke(messages)# 6. 更新短期记忆self.short_memory.add_user_message(user_input)self.short_memory.add_ai_message(response.content)returnresponse.contentdef_smart_recall(self,query):"""智能回忆：判断是否需要检索记忆"""# 简单的规则判断recall_signals=["上次","之前","我记得","我们聊过","你说过","我的"]needs_recall=any(siginqueryforsiginrecall_signals)ifneeds_recall:returnself.long_memory.recall(query,k=3)return[]defend_session(self):"""会话结束时，提取重要信息并更新画像"""conv_text=self.short_memory.get_conversation_text()ifnotconv_text:return# 1. 提取重要信息存入长期记忆extract_prompt=f"""请从以下对话中提取值得长期记住的重要信息。 只提取重要的事实、偏好、决策。每条一行。如果没有，返回"无"。 对话：{conv_text}"""result=self.llm.invoke(extract_prompt)ifresult.content.strip()!="无":lines=[l.strip()forlinresult.content.split("\n")ifl.strip()]forlineinlines:clean=line.split(". ",1)[-1]ifline[0].isdigit()elseline self.long_memory.save(clean,importance=0.7)# 2. 更新用户画像self.profile.update_from_conversation(conv_text)# 3. 清空短期记忆self.short_memory.clear()print(f"会话已保存。长期记忆共{self.long_memory.count()}条。")# ============ 使用示例 ============if__name__=="__main__":# 创建助手assistant=PersonalAssistant(user_id="zhangsan",storage_dir="./assistant_data",)# 第一次会话print("=== 第一次会话 ===")print(assistant.chat("你好，我叫张三，是个Python开发者，有5年经验"))print(assistant.chat("我最近在学AI Agent开发，用的LangChain"))print(assistant.chat("我偏好简洁的代码风格，不喜欢太多注释"))assistant.end_session()# 第二次会话（新的对话）print("\n=== 第二次会话 ===")print(assistant.chat("你好"))# 助手应该知道你是张三print(assistant.chat("我想做一个Agent项目，你有什么建议"))# 助手会根据你的Python和LangChain背景给建议assistant.end_session()# 第三次会话print("\n=== 第三次会话 ===")print(assistant.chat("上次你建议我做什么来着"))# 助手能回忆起上次对话的内容assistant.end_session()