摘要
记忆系统是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(url="http://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_name=self.collection, vectors_config=VectorParams(size=1536, distance=Distance.COSINE) ) def store(self, content: str, embedding: list[float], metadata: dict = {}): self.client.upsert( collection_name=self.collection, points=[PointStruct( id=str(uuid.uuid4()), vector=embedding, payload={"content": content, **metadata} )] ) def recall(self, query_embedding: list[float], top_k: int = 5) -> list[dict]: results = self.client.search( collection_name=self.collection, query_vector=query_embedding, limit=top_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_ok=True) 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", encoding="utf-8") as f: json.dump(episode, f, ensure_ascii=False, indent=2) def get_similar_episodes(self, task: str, limit: int = 3) -> list[dict]: """简单关键词匹配（生产环境建议用向量检索）""" episodes = [] for f in sorted(self.storage_dir.glob("*.json"), reverse=True)[:50]: ep = json.loads(f.read_text(encoding="utf-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_name=self.collection, points_selector=Filter( must=[FieldCondition( key="last_accessed", range=Range(lt=cutoff) )] ) )### 4.3 记忆一致性当长期记忆与短期记忆中存在矛盾信息时，优先以短期记忆（更新信息）为准，并触发长期记忆更新。## 五、总结| 记忆类型 | 持续时间 | 容量 | 访问速度 | 技术实现 ||---------|---------|------|---------|---------|| 感官记忆 | 单次推理 | 小 | 极快 | 内存缓冲 || 短期记忆 | 单次会话 | 中 | 快 | Context Window || 长期记忆 | 永久 | 大 | 中 | 向量数据库 || 情节记忆 | 永久 | 中 | 中 | 结构化JSON+向量 |构建完善的Agent记忆系统，是让AI从"单次问答"进化为"持续成长的智能助手"的关键工程投入。建议从短期记忆的上下文管理开始，逐步引入长期记忆，最终形成完整的四层记忆架构。
