AI 开发基础第6篇Memory - 让 Agent 拥有记忆适合读者已读完第5篇Skills/MCP想让Agent记住之前的交互预计阅读时间30分钟前言没有记忆的Agent每次对话开始LLM都是白纸一张。它不记得你昨天说了什么不记得你喜欢什么不记得上次任务做到哪里。第1次对话 用户我是素食主义者 Agent好的我记住了 第2次对话新会话 用户帮我推荐北京好吃的餐厅 Agent推荐全聚德烤鸭、东来顺涮羊肉...它记住了吗没有。第2次对话是新会话LLM的上下文窗口里没有任何第1次对话的信息。Memory机制就是解决这个问题。一、理解记忆的三个层次1.1 上下文窗口 ≠ 记忆很多人混淆了两个概念上下文窗口记忆生命周期单次对话跨对话持久化容量128K-200K Token理论无限成本每次都重新发送只发送需要的部分控制LLM自动处理开发者管理上下文窗口是工作记忆Memory是长期记忆。1.2 三层记忆架构层次名称存储位置生命周期L1对话历史Working Memorymessages数组单次对话L2短期记忆Session MemoryRedis/文件会话级几小时到几天L3长期记忆Long-term Memory数据库/向量库永久用户输入 ↓ [L1] 对话历史本次对话的前几轮 ↓ [L2] 短期记忆本次会话的关键信息摘要 ↓ [L3] 长期记忆用户画像、历史偏好、知识库 ↓ LLM处理 → 回复二、L1对话历史管理2.1 最简单的方式全部发送messages[{role:user,content:我是素食主义者},{role:assistant,content:好的记住了},{role:user,content:推荐北京餐厅},{role:assistant,content:推荐素心餐厅...},{role:user,content:再推荐几个},]问题对话越来越长Token越来越多成本越来越高最终超出上下文窗口。2.2 Token预算控制deftrim_messages(messages:list,max_tokens:int8000):按Token预算裁剪消息保留系统提示和最近的对话importtiktoken enctiktoken.encoding_for_model(gpt-4o-mini)# 保留系统提示system_msgs[mforminmessagesifm[role]system]other_msgs[mforminmessagesifm[role]!system]# 从最新的消息开始保留直到Token预算用完kept[]total_tokenssum(len(enc.encode(m[content]))forminsystem_msgs)formsginreversed(other_msgs):msg_tokenslen(enc.encode(msg[content]))iftotal_tokensmsg_tokensmax_tokens:breakkept.insert(0,msg)total_tokensmsg_tokensreturnsystem_msgskept2.3 滑动窗口 摘要更聪明的做法旧的消息不是直接丢弃而是压缩成摘要。asyncdefsummarize_messages(messages:list)-str:将旧消息压缩成摘要ifnotmessages:returnold_content\n.join([f{用户ifm[role]userelse助手}:{m[content]}forminmessages])summary_promptf将以下对话历史压缩成一段简洁的摘要保留关键信息 1. 用户提到的偏好和需求 2. 已完成和未完成的任务 3. 重要的决策和结论 对话历史{old_content}摘要responseawaitllm.chat(summary_prompt)returnresponseasyncdefmanaged_messages(messages:list,max_recent:int10):管理消息旧消息摘要 最近N轮完整保留iflen(messages)max_recent:returnmessages old_msgsmessages[:-max_recent]recent_msgsmessages[-max_recent:]summaryawaitsummarize_messages(old_msgs)return[{role:system,content:f之前的对话摘要{summary}},*recent_msgs]效果10轮前的对话被压缩成一段摘要几百Token最近10轮完整保留。既省Token又不错过重要信息。三、L2短期记忆3.1 用Redis存储会话记忆importjsonimportredisfromdatetimeimporttimedeltaclassSessionMemory:短期会话记忆Redis实现def__init__(self,redis_url:strredis://localhost:6379):self.clientredis.from_url(redis_url)self.ttltimedelta(hours24)# 24小时过期defsave_key_info(self,session_id:str,key_info:dict):保存会话关键信息keyfsession:{session_id}:key_infoself.client.setex(key,self.ttl,json.dumps(key_info,ensure_asciiFalse))defget_key_info(self,session_id:str)-dict:获取会话关键信息keyfsession:{session_id}:key_infodataself.client.get(key)returnjson.loads(data)ifdataelse{}defappend_task(self,session_id:str,task:dict):追加任务记录keyfsession:{session_id}:taskstasksself.get_tasks(session_id)tasks.append(task)self.client.setex(key,self.ttl,json.dumps(tasks,ensure_asciiFalse))defget_tasks(self,session_id:str)-list:获取任务列表keyfsession:{session_id}:tasksdataself.client.get(key)returnjson.loads(data)ifdataelse[]# 使用memorySessionMemory()# 用户说我是素食主义者后memory.save_key_info(session_001,{diet:素食,city:北京,preferences:[川菜,火锅],})# 下次对话时把关键信息注入到system promptkey_infomemory.get_key_info(session_001)system_promptf用户信息{json.dumps(key_info,ensure_asciiFalse)}3.2 真实项目经验在智能行程规划项目中用户可能分多次提供信息第1次城市、日期第2次预算、人数第3次特殊需求带小孩、无障碍用Redis保存这些信息下次对话自动加载用户不需要重复说。踩坑Redis连接池要配好不然并发高了会报错TTL设置太短1小时导致用户中途去吃个饭回来信息就没了。后来改成24小时四、L3长期记忆4.1 用户画像长期记忆最常见的形式记住用户的偏好和特征。importsqlite3classUserProfile:用户画像SQLite实现def__init__(self,db_path:struser_profiles.db):self.connsqlite3.connect(db_path)self._init_db()def_init_db(self):self.conn.execute( CREATE TABLE IF NOT EXISTS user_profile ( user_id TEXT PRIMARY KEY, preferences TEXT, -- JSON: {diet: 素食, city: 北京} history TEXT, -- JSON: 最近10次任务摘要 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) )self.conn.commit()defupdate_preference(self,user_id:str,key:str,value:str):更新用户偏好profileself.get_profile(user_id)ifnotprofile:profile{preferences:{},history:[]}profile[preferences][key]value self.conn.execute(INSERT OR REPLACE INTO user_profile (user_id, preferences, history) VALUES (?, ?, ?),(user_id,json.dumps(profile[preferences]),json.dumps(profile[history])))self.conn.commit()defget_profile(self,user_id:str)-dict:获取用户画像rowself.conn.execute(SELECT preferences, history FROM user_profile WHERE user_id ?,(user_id,)).fetchone()ifnotrow:returnNonereturn{preferences:json.loads(row[0])ifrow[0]else{},history:json.loads(row[1])ifrow[1]else[],}# 使用profile_dbUserProfile()# 每次对话结束后提取关键偏好defextract_preferences(user_input:str,user_id:str):从用户输入中提取偏好简单规则实际可用LLM提取preference_rules{素食:diet,不辣:spicy_level,预算:budget_range,}forkeyword,pref_keyinpreference_rules.items():ifkeywordinuser_input:profile_db.update_preference(user_id,pref_key,keyword)4.2 向量检索RAG当长期记忆数据量大时不能每次都全部发给LLM。用向量数据库检索相关记忆。importnumpyasnpfromnumpy.linalgimportnormclassSimpleVectorMemory:简单向量记忆小规模可用生产环境用Chroma/FAISSdef__init__(self,embedding_func):self.memories[]# [{text: ..., embedding: [...], meta: {...}}]self.embedding_funcembedding_funcasyncdefadd(self,text:str,meta:dictNone):添加记忆embeddingawaitself.embedding_func(text)self.memories.append({text:text,embedding:embedding,meta:metaor{},})asyncdefsearch(self,query:str,top_k:int5)-list:检索相关记忆query_embeddingawaitself.embedding_func(query)scored[]formeminself.memories:similaritynp.dot(query_embedding,mem[embedding])/(norm(query_embedding)*norm(mem[embedding]))scored.append((similarity,mem))scored.sort(reverseTrue)return[mfor_,minscored[:top_k]]# 使用asyncdefbuild_memory_context(query:str,user_id:str)-str:构建带记忆的上下文# 1. 用户画像profileprofile_db.get_profile(user_id)# 2. 向量检索相关记忆related_memoriesawaitvector_memory.search(query)contextifprofileandprofile[preferences]:contextf用户偏好{json.dumps(profile[preferences])}\nifrelated_memories:context相关历史信息\nformeminrelated_memories:contextf-{mem[text]}\nreturncontextRAGRetrieval-Augmented Generation就是这个流程用户问题 → 向量检索相关记忆 → 把记忆注入Prompt → LLM生成回答五、Memory与RAG的关系5.1 它们不是同一个东西MemoryRAG本质记住信息检索信息数据来源用户交互产生的信息外部知识库文档、网页目的个性化记住你增强知识知道更多存储用户级别的偏好/历史领域级别的文档/知识类比Memory 你的日记本记录你自己的事RAG 你的参考书查外部资料5.2 它们可以结合asyncdefbuild_full_context(user_id:str,query:str)-str:构建完整上下文用户画像 对话历史 外部知识parts[]# L1: 对话历史最近N轮recentget_recent_messages(user_id,n10)ifrecent:parts.append(f最近对话\n{format_messages(recent)})# L2: 会话摘要summaryget_session_summary(user_id)ifsummary:parts.append(f会话摘要{summary})# L3: 用户画像profileprofile_db.get_profile(user_id)ifprofile:parts.append(f用户画像{json.dumps(profile[preferences])})# RAG: 外部知识检索knowledgeawaitknowledge_base.search(query,top_k3)ifknowledge:parts.append(相关知识\n\n.join(knowledge))return\n\n.join(parts)六、本章总结你学到了什么三层记忆架构L1对话历史、L2短期记忆Redis、L3长期记忆数据库/向量库上下文管理滑动窗口 摘要旧消息不丢Token不超标用户画像用SQLite/JSON存储用户偏好跨会话保持个性化RAG向量检索相关记忆/知识注入Prompt增强回答质量Memory vs RAGMemory记住你RAG知道更多两者结合效果最好关键公式完整上下文 对话历史 会话摘要 用户画像 外部知识 Memory 个性化记住你的偏好 RAG 增强知识检索外部资料下一篇预告第7篇Subagent 与 Multi-Agent - 分而治之多智能体协作参考资料MemGPT论文https://arxiv.org/abs/2310.08560RAG原论文https://arxiv.org/abs/2005.11401LangChain Memory文档https://python.langchain.com/docs/how_to/memory/Chroma向量数据库https://www.trychroma.com/上一篇第5篇 Skills 与 MCP下一篇第7篇 Subagent 与 Multi-Agent