AI Agent记忆系统工程：从短期记忆到长期知识的完整架构

为什么记忆是Agent工程化的核心难题在2026年构建一个能在单次对话中完成复杂任务的AI Agent已经相对成熟——LangGraph、AutoGen等框架提供了完善的工具链。但当我们试图构建一个能够跨会话学习、记住用户偏好、积累领域知识的AI应用时挑战才真正开始。人类的工作记忆和长期记忆是两套完全不同的系统- 工作记忆短期有限容量用于当前任务- 长期记忆几乎无限容量通过整合形成持久知识LLM的上下文窗口就是工作记忆的模拟——但它有硬性的token上限。如何在这个限制内构建类人的记忆体系是AI Agent工程化最核心的挑战之一。—## 一、四层记忆架构设计### 1.1 记忆层次模型┌─────────────────────────────────────────────────────┐│ Layer 4: 语义记忆 (Semantic Memory) ││ 存储领域知识、事实、概念 ││ 实现向量数据库 知识图谱 ││ 特点高度结构化支持精确检索 │├─────────────────────────────────────────────────────┤│ Layer 3: 情节记忆 (Episodic Memory) ││ 存储历史交互事件、成功/失败经验 ││ 实现时序数据库 向量索引 ││ 特点按时间组织支持回溯 │├─────────────────────────────────────────────────────┤│ Layer 2: 程序记忆 (Procedural Memory) ││ 存储工作流程、决策规则、操作模式 ││ 实现结构化存储JSON/YAML 代码 ││ 特点高度可执行直接影响Agent行为 │├─────────────────────────────────────────────────────┤│ Layer 1: 工作记忆 (Working Memory) ││ 存储当前会话上下文、活跃任务状态 ││ 实现LLM上下文窗口 短期KV存储 ││ 特点高速读写会话结束后清空 │└─────────────────────────────────────────────────────┘### 1.2 Python实现框架pythonfrom abc import ABC, abstractmethodfrom dataclasses import dataclass, fieldfrom datetime import datetimefrom typing import List, Optional, Dict, Anyimport jsondataclassclass Memory: 记忆单元基础数据结构 id: str content: str memory_type: str # semantic, episodic, procedural, working created_at: datetime last_accessed: datetime access_count: int 0 importance: float 0.5 # 0-1影响记忆保留优先级 metadata: Dict[str, Any] field(default_factorydict) embedding: Optional[List[float]] Noneclass MemoryLayer(ABC): 记忆层抽象基类 abstractmethod async def store(self, content: str, metadata: dict None) - str: 存储记忆返回记忆ID pass abstractmethod async def retrieve(self, query: str, top_k: int 5) - List[Memory]: 检索相关记忆 pass abstractmethod async def update(self, memory_id: str, updates: dict) - bool: 更新记忆 pass abstractmethod async def forget(self, memory_id: str) - bool: 遗忘删除记忆 passclass WorkingMemory(MemoryLayer): 工作记忆管理当前会话的活跃上下文 def __init__(self, max_tokens: int 8000): self.max_tokens max_tokens self.active_memories: List[Memory] [] self.current_task: Optional[str] None self.active_tools: List[str] [] async def store(self, content: str, metadata: dict None) - str: memory Memory( idself._generate_id(), contentcontent, memory_typeworking, created_atdatetime.now(), last_accesseddatetime.now(), metadatametadata or {} ) self.active_memories.append(memory) # 超过容量时触发压缩 await self._compress_if_needed() return memory.id async def retrieve(self, query: str, top_k: int 5) - List[Memory]: 工作记忆检索返回最近和最相关的记忆 # 简单实现按时间倒序返回 return sorted(self.active_memories, keylambda m: m.last_accessed, reverseTrue)[:top_k] async def _compress_if_needed(self): 当工作记忆超限时进行压缩 total_tokens sum( estimate_tokens(m.content) for m in self.active_memories ) if total_tokens self.max_tokens: return # 策略将早期记忆摘要压缩 old_memories self.active_memories[:len(self.active_memories)//2] summary await self._summarize(old_memories) # 用摘要替换原始记忆 summary_memory Memory( idself._generate_id(), contentf[会话摘要] {summary}, memory_typeworking, created_atold_memories[0].created_at, last_accesseddatetime.now(), importance0.8 ) self.active_memories [summary_memory] \ self.active_memories[len(old_memories):] async def _summarize(self, memories: List[Memory]) - str: 使用LLM对记忆列表生成摘要 content \n.join(m.content for m in memories) return await call_llm( f请将以下内容压缩为200字以内的摘要保留关键信息\n{content} ) def to_context_string(self) - str: 将工作记忆转换为可注入上下文的字符串 if not self.active_memories: return parts [] for m in self.active_memories: parts.append(m.content) return \n.join(parts)class EpisodicMemory(MemoryLayer): 情节记忆存储历史交互事件和经验 def __init__(self, vector_db, time_db): self.vector_db vector_db self.time_db time_db async def store(self, content: str, metadata: dict None) - str: 存储一个情节记忆 memory Memory( idself._generate_id(), contentcontent, memory_typeepisodic, created_atdatetime.now(), last_accesseddatetime.now(), importancemetadata.get(importance, 0.5) if metadata else 0.5, metadatametadata or {} ) # 生成embedding并存储到向量库 memory.embedding await embed(content) await self.vector_db.insert(memory) # 同时存储到时序库支持时间范围查询 await self.time_db.insert(memory) return memory.id async def retrieve(self, query: str, top_k: int 5, time_range: tuple None) - List[Memory]: 按语义相似度检索情节记忆 query_embedding await embed(query) results await self.vector_db.search( query_embedding, top_ktop_k * 2, # 多检索一些再过滤 filterself._build_time_filter(time_range) ) # 重排序综合相关性和时间新鲜度 reranked self._rerank_by_recency_and_relevance(results, query_embedding) return reranked[:top_k] def _rerank_by_recency_and_relevance(self, results, query_embedding) - List[Memory]: 综合相关性和时间新鲜度重排序 now datetime.now() for memory in results: # 时间衰减因子7天内的记忆权重更高 days_old (now - memory.created_at).days recency_factor max(0.1, 1 - days_old / 30) # 综合分相关性 * 0.7 新鲜度 * 0.3 memory.combined_score ( memory.relevance_score * 0.7 recency_factor * 0.3 ) return sorted(results, keylambda m: m.combined_score, reverseTrue)class SemanticMemory(MemoryLayer): 语义记忆存储领域知识和长期事实 def __init__(self, vector_db, knowledge_graphNone): self.vector_db vector_db self.kg knowledge_graph # 可选知识图谱增强 async def store(self, content: str, metadata: dict None) - str: 存储语义知识 # 提取知识的关键实体和关系用于知识图谱 if self.kg and metadata and metadata.get(extract_entities): entities await self._extract_entities(content) await self.kg.add_entities(entities) memory Memory( idself._generate_id(), contentcontent, memory_typesemantic, created_atdatetime.now(), last_accesseddatetime.now(), importancemetadata.get(importance, 0.7) if metadata else 0.7, metadatametadata or {} ) memory.embedding await embed(content) await self.vector_db.insert(memory) return memory.id async def update_importance(self, memory_id: str, new_importance: float): 更新记忆重要性基于访问频率和用户反馈 await self.update(memory_id, {importance: new_importance})—## 二、记忆整合器协调多层记忆pythonclass MemoryOrchestrator: 记忆整合器在Agent决策时协调多层记忆的检索和注入 def __init__(self, working: WorkingMemory, episodic: EpisodicMemory, semantic: SemanticMemory, procedural: ProceduralMemory): self.working working self.episodic episodic self.semantic semantic self.procedural procedural async def build_context_for_query(self, query: str, token_budget: int 4000) - str: 为给定查询构建最优化的记忆上下文 按优先级注入不同层次的记忆 context_parts [] tokens_used 0 # 优先级1工作记忆当前会话上下文 working_context self.working.to_context_string() working_tokens estimate_tokens(working_context) if working_tokens token_budget * 0.4: # 最多用40%给工作记忆 context_parts.append((working, working_context)) tokens_used working_tokens remaining token_budget - tokens_used # 优先级2程序记忆相关操作规则 procedures await self.procedural.retrieve(query, top_k3) for proc in procedures: proc_tokens estimate_tokens(proc.content) if tokens_used proc_tokens token_budget * 0.6: context_parts.append((procedural, proc.content)) tokens_used proc_tokens # 优先级3语义记忆领域知识 semantic_memories await self.semantic.retrieve(query, top_k5) for mem in semantic_memories: mem_tokens estimate_tokens(mem.content) if tokens_used mem_tokens token_budget * 0.85: context_parts.append((semantic, mem.content)) tokens_used mem_tokens # 优先级4情节记忆历史经验 episodic_memories await self.episodic.retrieve(query, top_k3) for mem in episodic_memories: mem_tokens estimate_tokens(mem.content) if tokens_used mem_tokens token_budget: context_parts.append((episodic, mem.content)) tokens_used mem_tokens return self._format_context(context_parts) def _format_context(self, parts: list) - str: 格式化多层记忆为结构化上下文 sections { working: [], procedural: [], semantic: [], episodic: [] } for mem_type, content in parts: sections[mem_type].append(content) formatted [] if sections[working]: formatted.append([当前会话]\n \n.join(sections[working])) if sections[procedural]: formatted.append([操作规则]\n \n.join(sections[procedural])) if sections[semantic]: formatted.append([背景知识]\n \n.join(sections[semantic])) if sections[episodic]: formatted.append([历史经验]\n \n.join(sections[episodic])) return \n\n.join(formatted) async def consolidate_session(self, session_id: str): 会话结束时整合记忆将工作记忆中的重要信息 升级到长期记忆 working_memories await self.working.retrieve(, top_k100) # 识别值得长期保留的记忆 for memory in working_memories: if memory.importance 0.7: # 事实性知识 → 语义记忆 if self._is_factual(memory.content): await self.semantic.store( memory.content, metadata{source: fsession:{session_id}} ) # 经验性知识 → 情节记忆 else: await self.episodic.store( memory.content, metadata{session_id: session_id} ) # 清空工作记忆 self.working.active_memories []—## 三、记忆遗忘机制防止知识库膨胀pythonclass MemoryGarbageCollector: 记忆垃圾回收器防止记忆库无限膨胀 async def run_gc(self, memory_layer: MemoryLayer, strategy: str lru): 执行记忆垃圾回收 策略 - lru: 最近最少使用类似CPU缓存替换 - importance: 按重要性淘汰低分记忆 - hybrid: 综合访问频率、重要性、时间 all_memories await memory_layer.list_all() if len(all_memories) self.max_memories: return scores self._score_memories(all_memories, strategy) # 淘汰分数最低的记忆 to_forget sorted(scores, keylambda x: x[1]) count_to_remove len(all_memories) - self.max_memories for memory, score in to_forget[:count_to_remove]: await memory_layer.forget(memory.id) def _score_memories(self, memories: List[Memory], strategy: str) - List[tuple]: now datetime.now() scores [] for memory in memories: if strategy lru: days_since_access (now - memory.last_accessed).days score -days_since_access # 越久未访问分越低 elif strategy importance: score memory.importance elif strategy hybrid: days_since_access (now - memory.last_accessed).days recency_score max(0, 1 - days_since_access / 90) score (memory.importance * 0.5 recency_score * 0.3 min(memory.access_count / 10, 1.0) * 0.2) scores.append((memory, score)) return scores—## 四、实践总结Agent记忆系统的设计原则1.分层设计不同类型的知识用不同存储机制不要试图用一个大的向量库搞定一切2.记忆注入要精准每次推理时只注入当前查询真正需要的记忆而不是全量注入3.重要性评分是关键记忆系统的核心是如何判断什么值得记住这通常需要多个信号明确反馈隐式行为4.遗忘是必要的没有遗忘机制的记忆系统终将爆炸——无论是成本还是检索质量都会恶化5.记忆整合要异步会话结束后的记忆整合consolidation应该异步进行不阻塞用户交互Agent记忆系统是当前AI应用工程化中最值得深入投入的方向之一。它不只是技术问题更是认知科学和工程设计的交叉地带。