记忆系统

为什么 Agent 需要记忆 LLM 本身是无状态的——每次调用都是独立的。对话历史塞进 Context Window 是最简单的"记忆"，但有致命缺陷：token 成本线性增长、上下文长度有上限、无法跨会话持久化。一个真正有用的 Agent 需要像人一样拥有多层次记忆系统。 四种记忆类型 记忆类型 类比人类 生命周期 存储介质 示例 工作记忆 工作台上的文件 当前任务 Redis/内存 当前对话上下文、临时变量 情景记忆 个人经历 天~月 向量数据库 “用户上周问了XX问题” 语义记忆 知识事实 长期 向量数据库 + 图数据库 “用户是Java开发者” 程序记忆 肌肉记忆 长期 规则库/技能库 “调用API的固定流程” 架构设计 ┌──────────────────────────────────────────────────┐ │ Agent 记忆系统 │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ 工作记忆 │ │ 情景记忆 │ │ 语义记忆 │ │ │ │ (Redis) │ │(Qdrant) │ │(Qdrant) │ │ │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ │ │ │ │ └──────────────┼──────────────┘ │ │ ▼ │ │ ┌──────────────┐ │ │ │ 记忆管理器 │ │ │ │ (Memory Mgr) │ │ │ └──────┬───────┘ │ │ ▼ │ │ ┌──────────────┐ │ │ │ 程序记忆 │ │ │ │ (规则库) │ │ │ └──────────────┘ │ └──────────────────────────────────────────────────┘ 核心实现 1. 记忆数据结构 from dataclasses import dataclass, field from datetime import datetime from enum import Enum import uuid class MemoryType(Enum): WORKING = "working" EPISODIC = "episodic" SEMANTIC = "semantic" PROCEDURAL = "procedural" @dataclass class Memory: id: str = field(default_factory=lambda: str(uuid.uuid4())) type: MemoryType = MemoryType.EPISODIC content: str = "" embedding: list[float] = field(default_factory=list) metadata: dict = field(default_factory=dict) importance: float = 0.5 # 0-1, 影响遗忘策略 created_at: datetime = field(default_factory=datetime.now) last_accessed: datetime = field(default_factory=datetime.now) access_count: int = 0 decay_factor: float = 1.0 # 随时间衰减 2. 记忆写入 import redis import json from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct class MemoryStore: def __init__(self): self.redis = redis.Redis(host='localhost', port=6379, db=0) self.qdrant = QdrantClient(host='localhost', port=6333) self._init_collections() def _init_collections(self): """初始化向量集合""" for collection in ["episodic", "semantic"]: if not self.qdrant.collection_exists(collection): self.qdrant.create_collection( collection_name=collection, vectors_config=VectorParams(size=1024, distance=Distance.COSINE) ) async def write(self, memory: Memory): """写入记忆""" if memory.type == MemoryType.WORKING: # 工作记忆写入 Redis，设置过期时间 key = f"working:{memory.id}" self.redis.setex( key, 3600, # 1小时过期 json.dumps({ "content": memory.content, "metadata": memory.metadata }) ) elif memory.type in (MemoryType.EPISODIC, MemoryType.SEMANTIC): # 情景/语义记忆写入向量库 self.qdrant.upsert( collection_name=memory.type.value, points=[PointStruct( id=memory.id, vector=memory.embedding, payload={ "content": memory.content, "metadata": memory.metadata, "importance": memory.importance, "created_at": memory.created_at.isoformat(), "last_accessed": memory.last_accessed.isoformat(), "access_count": memory.access_count } )] ) async def search(self, query_vector, memory_type: MemoryType, top_k=5): """检索记忆""" if memory_type == MemoryType.WORKING: # 工作记忆：返回所有未过期的 keys = self.redis.keys("working:*") results = [] for key in keys: data = json.loads(self.redis.get(key)) results.append(data) return results # 向量检索 hits = self.qdrant.search( collection_name=memory_type.value, query_vector=query_vector, limit=top_k, score_threshold=0.7 ) return [hit.payload for hit in hits] 3. 记忆检索与排序 class MemoryRetriever: """记忆检索器：综合相关性、重要性、时效性""" def __init__(self, store: MemoryStore): self.store = store async def retrieve(self, query: str, query_vector: list, context: dict = None) -> list[dict]: # 并行检索三种记忆 import asyncio tasks = [ self.store.search(query_vector, MemoryType.WORKING), self.store.search(query_vector, MemoryType.EPISODIC, top_k=5), self.store.search(query_vector, MemoryType.SEMANTIC, top_k=5), ] working, episodic, semantic = await asyncio.gather(*tasks) # 融合排序 all_memories = [] for mem in episodic + semantic: score = self._compute_score(mem, query_vector) all_memories.append((mem, score)) all_memories.sort(key=lambda x: x[1], reverse=True) # 返回 top-K + 工作记忆 return { "working": working, "long_term": [m[0] for m in all_memories[:5]] } def _compute_score(self, mem, query_vector): """综合评分 = 相关性 × 时效性 × 重要性 × 访问频率""" import numpy as np from datetime import datetime, timedelta # 相关性（向量相似度，假设payload中存了） relevance = mem.get("score", 0.5) # 时效性衰减 created = datetime.fromisoformat(mem["created_at"]) days_old = (datetime.now() - created).days recency = np.exp(-days_old / 30) # 30天半衰期 # 重要性 importance = mem.get("importance", 0.5) # 访问频率（越常访问越重要） access_factor = np.log1p(mem.get("access_count", 0)) return relevance * (0.3 + 0.4 * recency) * importance * (1 + 0.1 * access_factor) 4. 遗忘机制 class ForgettingManager: """遗忘机制：模拟人类遗忘曲线""" async def decay_memories(self): """定期执行，降低旧记忆的重要性""" from datetime import datetime, timedelta cutoff = datetime.now() - timedelta(days=7) # 在向量库中批量更新 self.store.qdrant.set_payload( collection_name="episodic", payload={"decay_factor": 0.8}, # 每次衰减20% filters={ "must": [ {"field": "last_accessed", "range": {"lt": cutoff.isoformat()}} ] } ) async def cleanup_memories(self): """删除重要性低于阈值的记忆""" self.store.qdrant.delete( collection_name="episodic", points_selector={ "filter": { "must": [ {"field": "importance", "range": {"lt": 0.1}}, {"field": "decay_factor", "range": {"lt": 0.3}} ] } } ) 实战建议 重要性评分策略：不要让 Agent 自己打分（LLM 评分不稳定）。用规则：用户明确要求记住的 → 1.0，任务关键信息 → 0.8，普通对话 → 0.3。 ...

为什么 Agent 需要记忆 LLM 是无状态的——每次调用都是全新的。但真正的 Agent 需要记住： 5 分钟前用户说了什么（工作记忆） 昨天的对话中用户提到了什么偏好（情景记忆） 用户公司的 API 文档（语义记忆） 上周犯的错误，不要再犯（反思记忆） 记忆类型与人类类比 记忆类型 人类类比 Agent 实现 持续时间 工作记忆 短期记忆 上下文窗口 单次对话 情景记忆 个人经历 对话历史存储 天-月 语义记忆 知识事实 向量知识库 永久 程序记忆 技能习惯 工具使用模式 永久 反思记忆 经验教训 总结+规则 永久 工作记忆管理 上下文窗口策略 class WorkingMemory: def __init__(self, max_tokens=8000): self.max_tokens = max_tokens self.messages = [] def add(self, message): self.messages.append(message) self._compress_if_needed() def _compress_if_needed(self): while self.token_count() > self.max_tokens: if len(self.messages) <= 2: break # 取出最早的几条消息做摘要 old_messages = self.messages[:3] summary = self.summarize(old_messages) self.messages = [ {"role": "system", "content": f"之前的对话摘要：{summary}"} ] + self.messages[3:] 注意力机制 class AttentionManager: """管理上下文中各部分的"重要性"""" def __init__(self): self.attention_scores = {} def update(self, message_id, content, user_query): """计算每条消息对当前问题的相关性""" self.attention_scores[message_id] = { "recency": self.recency_score(message_id), "relevance": self.relevance_score(content, user_query), "importance": self.importance_score(content), } def get_context(self, max_tokens=4000): """按注意力分数排序，选择最重要的上下文""" ranked = sorted( self.attention_scores.items(), key=lambda x: x[1]["recency"] * 0.2 + x[1]["relevance"] * 0.5 + x[1]["importance"] * 0.3, reverse=True ) context = [] token_count = 0 for msg_id, _ in ranked: msg = self.get_message(msg_id) token_count += count_tokens(msg) if token_count > max_tokens: break context.append(msg) return context 情景记忆：对话历史 存储结构 from dataclasses import dataclass from datetime import datetime @dataclass class EpisodicMemory: id: str session_id: str timestamp: datetime user_input: str agent_response: str tools_used: list[str] outcome: str # success / failure / partial embedding: list[float] # 用于检索 metadata: dict # 用户信息、环境等 检索策略 class EpisodicMemoryStore: async def recall(self, query, user_id, limit=5): """回忆与当前情况相关的历史对话""" # 时间衰减权重：越近的记忆权重越高 recency_weight = 0.3 # 语义相似度 query_emb = self.embedder.embed(query) results = await self.db.search( user_id=user_id, embedding=query_emb, limit=limit * 3, # 多取一些做重排 ) # 综合排序 for r in results: days_ago = (datetime.now() - r.timestamp).days r.score = r.similarity * (1 - recency_weight) + \ math.exp(-days_ago / 30) * recency_weight results.sort(key=lambda x: x.score, reverse=True) return results[:limit] 记忆巩固 class MemoryConsolidation: """将情景记忆转化为语义记忆（类似人类睡眠时的记忆巩固）""" async def consolidate(self, user_id): # 1. 提取最近7天的对话 recent = await self.episodic_store.get_recent(user_id, days=7) # 2. 用 LLM 提取可复用的知识 knowledge = await self.llm.extract( f"从以下对话中提取可复用的知识和模式：\n{recent}" ) # 3. 存入语义记忆 for item in knowledge: await self.semantic_store.add( content=item.content, embedding=self.embedder.embed(item.content), source="consolidated", confidence=item.confidence, ) # 4. 标记已巩固的情景记忆 await self.episodic_store.mark_consolidated(recent) 语义记忆：知识库 分层知识存储 class SemanticMemory: """三层知识存储""" def __init__(self): # 第一层：快速事实（Key-Value） self.facts = RedisStore() # "用户公司" → "ABC科技" # 第二层：文档知识（向量检索） self.documents = VectorStore() # API文档、产品手册 # 第三层：深层知识（知识图谱） self.graph = GraphStore() # 实体关系 知识更新 class KnowledgeUpdater: async def learn(self, new_info, source="conversation"): """学习新知识""" # 1. 检查是否与已有知识冲突 conflicts = await self.find_conflicts(new_info) if conflicts: # 2. 冲突解决：用更新的信息覆盖 for conflict in conflicts: if conflict.confidence < new_info.confidence: await self.update(conflict.id, new_info) else: # 不确定，存为"待确认" await self.flag_for_review(new_info, conflict) else: # 3. 新知识，直接存入 await self.semantic_store.add(new_info) 反思记忆：经验学习 class ReflectionSystem: """Agent 从失败中学习""" async def reflect(self, task, attempt, outcome): if outcome.success: return # 成功不需要反思 # 分析失败原因 analysis = await self.llm.analyze( task=task, actions=attempt.actions, error=outcome.error, prompt="分析为什么这次任务失败了，如何避免" ) # 提取规则 rules = self.extract_rules(analysis) # 存入反思记忆 for rule in rules: await self.reflection_store.add({ "situation": rule.situation, "mistake": rule.mistake, "better_action": rule.better_action, "confidence": rule.confidence, }) async def consult(self, task): """执行任务前，查阅反思记忆""" relevant_mistakes = await self.reflection_store.search(task) if relevant_mistakes: return f""" 注意：根据历史经验，类似任务有以下陷阱： {format_mistakes(relevant_mistakes)} 请避免这些错误。 """ return None 记忆架构实战 class AgentMemory: """完整的 Agent 记忆系统""" def __init__(self): self.working = WorkingMemory(max_tokens=8000) self.episodic = EpisodicMemoryStore() self.semantic = SemanticMemory() self.reflection = ReflectionSystem() async def prepare_context(self, query, user_id): """为 LLM 调用准备完整上下文""" context_parts = [] # 1. 工作记忆（当前对话） context_parts.append(self.working.get_recent(n=6)) # 2. 相关情景记忆 episodes = await self.episodic.recall(query, user_id, limit=3) if episodes: context_parts.append(f"历史相关对话：{format_episodes(episodes)}") # 3. 语义知识 knowledge = await self.semantic.search(query, limit=5) if knowledge: context_parts.append(f"相关知识：{format_knowledge(knowledge)}") # 4. 反思提醒 warning = await self.reflection.consult(query) if warning: context_parts.append(warning) return "\n\n".join(context_parts) 记忆数据库选型 记忆类型 推荐数据库 理由 工作记忆 Redis 低延迟，自动过期 情景记忆 PostgreSQL + pgvector 事务+向量检索 语义记忆 Qdrant / Milvus 高性能向量检索 反思记忆 MongoDB 灵活 Schema 知识图谱 Neo4j / NebulaGraph 图查询 记忆隐私 class MemoryPrivacy: """记忆系统中的隐私保护""" PII_PATTERNS = [ (r'\d{11}', 'PHONE'), # 手机号 (r'\d{18}', 'ID_CARD'), # 身份证 (r'[\w.]+@[\w.]+', 'EMAIL'), # 邮箱 ] def sanitize(self, memory): """存储前脱敏""" for pattern, pii_type in self.PII_PATTERNS: memory = re.sub(pattern, f'[{pii_type}]', memory) return memory def user_forget(self, user_id): """用户要求删除所有记忆""" await self.episodic.delete_by_user(user_id) await self.semantic.delete_by_user(user_id) await self.reflection.delete_by_user(user_id) # 生成删除确认报告 return {"deleted": True, "user_id": user_id} 结论 记忆是 Agent 从"工具"变成"伙伴"的关键。没有记忆的 Agent 每次都是陌生人，有了记忆它才能： ...

Agent 记忆系统 2026：让 AI 真正「记住」 2026 年，Agent 记忆系统已经从「概念验证」进化到「生产级架构」。本文系统梳理当前最成熟的 Agent 记忆方案，从架构设计到落地实践。 记忆系统的三层架构 现代 Agent 记忆系统通常分为三层： 第一层：工作记忆（Working Memory） 当前会话的上下文，通常存在 LLM 的上下文窗口中 容量：4K-2M tokens（取决于模型） 特点：高速、容量有限、会话结束即丢失 第二层：情景记忆（Episodic Memory） 历史会话的关键信息，通过向量检索召回 存储：向量数据库（Chroma/Pinecone/Milvus） 容量：无限（受存储成本限制） 特点：语义检索、持久化、跨会话 第三层：长期记忆（Long-term Memory） 结构化的知识和用户偏好，通常以知识图谱形式存储 存储：Neo4j/TigerGraph + 向量数据库 特点：精确检索、关系推理、持续更新 2026 年主流方案对比 方案 记忆类型 检索方式 适用场景 代表项目 MemGPT 架构 分层记忆 函数调用调度 长文档处理 MemGPT/OpenClaw Vector-Only 向量检索 语义相似度 简单问答 Dify/RAGFlow Hybrid Memory 向量+图谱 向量+图查询 复杂推理 Hermes Agent Database Memory 结构化存储 SQL+向量 企业应用 Microsoft Copilot File-based Memory 文件系统 全文搜索 个人 Agent OpenClaw 工程实践：构建一个生产级记忆系统 Step 1：记忆提取 ...