摘要记忆系统是AI Agent实现持续学习和个性化服务的核心基础设施。本文系统梳理Agent记忆的四大类型——感官记忆、短期记忆、长期记忆与情节记忆并提供完整的工程实现方案。## 一、为什么Agent需要记忆系统大语言模型本质上是无状态的每次推理相互独立无法感知上下文历史。这使得复杂的多轮任务、个性化交互以及跨会话的经验积累成为难题。记忆系统的引入让Agent能够-跨轮次保持上下文避免重复询问已知信息-积累领域经验将成功/失败的工具调用结果保存为可复用知识-实现个性化记住用户偏好、历史决策和工作习惯-支持长程任务将数小时或数天的任务拆解成可恢复的子步骤## 二、四种记忆类型详解### 2.1 感官记忆Sensory Memory定义原始输入的即时缓冲保留时间极短单次推理内。工程实现pythonclass SensoryBuffer: 原始输入预处理与临时存储 def __init__(self, max_tokens: int 2048): self.buffer [] self.max_tokens max_tokens def ingest(self, inputs: list[dict]) - list[dict]: 过滤并截断输入到允许长度 filtered [i for i in inputs if i.get(relevance_score, 1.0) 0.3] return self._truncate_to_token_limit(filtered)适用场景多模态输入预处理、噪声过滤、输入压缩。### 2.2 短期记忆Short-Term / Working Memory定义当前任务的活跃上下文对应LLM的上下文窗口Context Window。工程实现pythonfrom collections import dequefrom typing import Optionalimport tiktokenclass WorkingMemory: def __init__(self, model: str gpt-4o, max_tokens: int 16000): self.messages: deque deque() self.encoder tiktoken.encoding_for_model(model) self.max_tokens max_tokens def add(self, role: str, content: str): self.messages.append({role: role, content: content}) self._evict_oldest() def _count_tokens(self, messages) - int: return sum(len(self.encoder.encode(m[content])) for m in messages) def _evict_oldest(self): while self._count_tokens(list(self.messages)) self.max_tokens: self.messages.popleft() def get_context(self) - list[dict]: return list(self.messages)关键挑战上下文窗口的LRU淘汰策略、重要消息的钉住pinning机制。### 2.3 长期记忆Long-Term Memory定义跨会话持久化的知识存储通常基于向量数据库实现语义检索。工程实现pythonfrom qdrant_client import QdrantClientfrom qdrant_client.models import Distance, VectorParams, PointStructimport uuidclass LongTermMemory: def __init__(self, collection: str agent_memory): self.client QdrantClient(urlhttp://localhost:6333) self.collection collection self._ensure_collection() def _ensure_collection(self): existing [c.name for c in self.client.get_collections().collections] if self.collection not in existing: self.client.create_collection( collection_nameself.collection, vectors_configVectorParams(size1536, distanceDistance.COSINE) ) def store(self, content: str, embedding: list[float], metadata: dict {}): self.client.upsert( collection_nameself.collection, points[PointStruct( idstr(uuid.uuid4()), vectorembedding, payload{content: content, **metadata} )] ) def recall(self, query_embedding: list[float], top_k: int 5) - list[dict]: results self.client.search( collection_nameself.collection, query_vectorquery_embedding, limittop_k ) return [{content: r.payload[content], score: r.score} for r in results]最佳实践- 使用元数据过滤时间范围、话题标签缩小召回范围- 存储时对内容分块避免超长文档噪声- 定期做记忆整合Memory Consolidation合并相似条目### 2.4 情节记忆Episodic Memory定义以完整事件或任务为单位的记忆保存做了什么、结果如何的完整轨迹。工程实现pythonimport jsonfrom datetime import datetimefrom pathlib import Pathclass EpisodicMemory: def __init__(self, storage_dir: str ./episodic_memory): self.storage_dir Path(storage_dir) self.storage_dir.mkdir(exist_okTrue) def record_episode(self, task: str, steps: list[dict], outcome: str, success: bool): episode { task: task, steps: steps, outcome: outcome, success: success, timestamp: datetime.now().isoformat(), duration_seconds: sum(s.get(duration, 0) for s in steps) } episode_id f{datetime.now().strftime(%Y%m%d_%H%M%S)}_{task[:20]} with open(self.storage_dir / f{episode_id}.json, w, encodingutf-8) as f: json.dump(episode, f, ensure_asciiFalse, indent2) def get_similar_episodes(self, task: str, limit: int 3) - list[dict]: 简单关键词匹配生产环境建议用向量检索 episodes [] for f in sorted(self.storage_dir.glob(*.json), reverseTrue)[:50]: ep json.loads(f.read_text(encodingutf-8)) if any(kw in ep[task] for kw in task.split()[:5]): episodes.append(ep) if len(episodes) limit: break return episodes## 三、四种记忆的集成架构用户输入 │ ▼[感官记忆] ──→ 过滤 压缩 ──→ [短期记忆/上下文窗口] │ ┌────────────┼────────────┐ ▼ ▼ ▼ [长期记忆检索] [情节记忆检索] [工具调用] │ │ │ └────────────┴────────────┘ │ 组合上下文 │ LLM推理 │ ┌─────────┴─────────┐ ▼ ▼ [回答用户] [更新记忆存储]## 四、生产环境关键考量### 4.1 记忆隔离不同用户的记忆必须严格隔离推荐以user_id作为向量数据库的 namespace 或 collection 前缀。### 4.2 记忆遗忘策略无限增长的记忆会降低检索质量。可实现基于访问频率的 TTL 机制pythondef decay_memories(self, days_threshold: int 90): 删除90天未访问的低相关记忆 cutoff datetime.now().timestamp() - days_threshold * 86400 self.client.delete( collection_nameself.collection, points_selectorFilter( must[FieldCondition( keylast_accessed, rangeRange(ltcutoff) )] ) )### 4.3 记忆一致性当长期记忆与短期记忆中存在矛盾信息时优先以短期记忆更新信息为准并触发长期记忆更新。## 五、总结| 记忆类型 | 持续时间 | 容量 | 访问速度 | 技术实现 ||---------|---------|------|---------|---------|| 感官记忆 | 单次推理 | 小 | 极快 | 内存缓冲 || 短期记忆 | 单次会话 | 中 | 快 | Context Window || 长期记忆 | 永久 | 大 | 中 | 向量数据库 || 情节记忆 | 永久 | 中 | 中 | 结构化JSON向量 |构建完善的Agent记忆系统是让AI从单次问答进化为持续成长的智能助手的关键工程投入。建议从短期记忆的上下文管理开始逐步引入长期记忆最终形成完整的四层记忆架构。