Agent 记忆架构设计：短期/长期/情景记忆
为什么 Agent 需要记忆 没有记忆的 Agent 就像金鱼——每次对话都从零开始。你告诉它你的名字，下一轮它就忘了。你纠正过它的错误，它下次照犯。 记忆系统让 Agent 能够： 记住用户偏好和历史交互 从过去经验中学习 保持跨会话的上下文一致性 在长任务中不丢失关键信息 记忆类型体系 人类认知科学启发 记忆类型 持续时间 容量 Agent 对应 存储方案 工作记忆 秒-分钟 4±2 项 当前对话上下文 LLM Context Window 短期记忆 分钟-小时 有限 当前会话摘要 内存/Redis 长期记忆 天-永久 近无限 用户偏好、知识 向量库 + 关系库 情景记忆 事件级 大 具体交互记录 时序向量库 架构总览 ┌──────────────────────────────────────────────┐ │ Agent 主循环 │ │ │ │ ┌─────────┐ ┌─────────┐ ┌──────────────┐ │ │ │ 工作记忆 │ │ 短期记忆 │ │ 长期记忆 │ │ │ │ Context │←→│ Session │←→│ Persistent │ │ │ │ Window │ │ Summary │ │ Store │ │ │ └─────────┘ └─────────┘ └──────────────┘ │ │ │ │ │ │ │ │ │ ┌────────┼────────┐ │ │ │ │ │ │ │ │ │ │ │ ┌──┴──┐ ┌───┴───┐ ┌──┴──┐ │ │ │ │语义 │ │情景 │ │程序 │ │ │ │ │记忆 │ │记忆 │ │记忆 │ │ │ │ │向量 │ │时序 │ │图 │ │ │ │ └─────┘ └───────┘ └─────┘ │ │ │ │ ┌────┴────────────┴────────────────────────┐│ │ │ 记忆管理器 ││ │ │ 写入 → 压缩 → 检索 → 遗忘 → 巩固 ││ │ └────────────────────────────────────────────┘│ └──────────────────────────────────────────────┘ 工作记忆：上下文窗口管理 from dataclasses import dataclass, field from typing import List, Dict @dataclass class WorkingMemory: """工作记忆: 管理 LLM 上下文窗口""" max_tokens: int = 128000 reserved_for_response: int = 4096 messages: List[dict] = field(default_factory=list) def add(self, role: str, content: str): self.messages.append({"role": role, "content": content}) self._evict_if_needed() def add_system(self, content: str): """系统消息放在最前面，不被驱逐""" self.messages.insert(0, {"role": "system", "content": content}) def _evict_if_needed(self): """当超过 token 限制时，驱逐最旧的对话""" while self._count_tokens() > self.max_tokens - self.reserved_for_response: if len(self.messages) <= 1: break # 找到第一个非 system 消息并移除 for i, msg in enumerate(self.messages): if msg["role"] != "system": evicted = self.messages.pop(i) # 触发压缩: 被驱逐的消息需要被压缩到短期记忆 self._on_evict(evicted) break def _count_tokens(self) -> int: import tiktoken enc = tiktoken.encoding_for_model("gpt-4o") return sum(len(enc.encode(m["content"])) for m in self.messages) def _on_evict(self, evicted_msg: dict): """被驱逐的消息回调，交给短期记忆处理""" # 由 MemoryManager 注册 pass def get_context(self) -> List[dict]: """获取当前上下文""" return self.messages.copy() 短期记忆：会话级摘要 class ShortTermMemory: """短期记忆: 当前会话的压缩摘要""" def __init__(self): self.summary = "" self.key_points: List[str] = [] self.token_budget = 2000 # 摘要不超 2000 token async def compress(self, messages: List[dict]) -> str: """将多轮对话压缩为摘要""" conversation = "\n".join( f"{m['role']}: {m['content'][:200]}" for m in messages ) prompt = f"""请将以下对话压缩为简洁摘要，保留: 1. 用户的核心需求 2. 已确定的关键事实 3. 未解决的问题 4. 用户的明确偏好 对话: {conversation} 摘要:""" response = await llm_call(prompt, max_tokens=self.token_budget) self.summary = response return response async def extract_key_points(self, messages: List[dict]) -> List[str]: """提取关键信息点""" prompt = f"""从以下对话中提取关键信息点，每点一行: {chr(10).join(m['content'][:200] for m in messages)} 关键信息:""" response = await llm_call(prompt, max_tokens=500) self.key_points = response.strip().split('\n') return self.key_points def to_context(self) -> str: """转换为可注入上下文的文本""" parts = [] if self.summary: parts.append(f"之前的对话摘要:\n{self.summary}") if self.key_points: parts.append(f"关键信息:\n" + "\n".join(f"- {p}" for p in self.key_points)) return "\n\n".join(parts) 长期记忆：持久化存储 语义记忆（向量库） class SemanticMemory: """语义记忆: 存储事实和知识，向量检索""" def __init__(self, vector_store, embed_model): self.store = vector_store self.embed = embed_model async def remember(self, content: str, metadata: dict = None): """记住一条信息""" embedding = await self.embed(content) await self.store.upsert({ "id": self._generate_id(content), "values": embedding, "metadata": { "content": content, "type": "semantic", "created_at": datetime.now().isoformat(), **(metadata or {}) } }) async def recall(self, query: str, top_k: int = 5) -> List[dict]: """检索相关记忆""" query_embedding = await self.embed(query) results = await self.store.query( vector=query_embedding, top_k=top_k, filter={"type": {"$eq": "semantic"}} ) return [ { "content": r.metadata["content"], "score": r.score, "metadata": r.metadata } for r in results.matches ] async def forget(self, content_id: str): """主动遗忘""" await self.store.delete(ids=[content_id]) 情景记忆（时序事件） class EpisodicMemory: """情景记忆: 记录具体交互事件，支持时间检索""" def __init__(self, db): self.db = db # PostgreSQL + pgvector async def record(self, event: dict): """记录一个交互事件""" await self.db.execute( """ INSERT INTO episodic_memory (event_id, user_id, timestamp, input, output, outcome, context_snapshot, embedding) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) """, event["id"], event["user_id"], datetime.now(), event["input"], event["output"], event["outcome"], json.dumps(event.get("context", {})), event.get("embedding") ) async def recall_by_time(self, user_id: str, start: datetime, end: datetime) -> List[dict]: """按时间范围回忆""" rows = await self.db.fetch( """SELECT * FROM episodic_memory WHERE user_id = $1 AND timestamp BETWEEN $2 AND $3 ORDER BY timestamp DESC""", user_id, start, end ) return [dict(r) for r in rows] async def recall_similar(self, user_id: str, query_embedding: list, top_k: int = 5) -> List[dict]: """检索相似的历史事件""" rows = await self.db.fetch( """SELECT *, embedding <=> $1 as distance FROM episodic_memory WHERE user_id = $2 ORDER BY embedding <=> $1 LIMIT $3""", query_embedding, user_id, top_k ) return [dict(r) for r in rows] 程序记忆（知识图谱） class ProceduralMemory: """程序记忆: 存储技能、流程、因果关系（图结构）""" def __init__(self, graph_db): self.graph = graph_db # Neo4j async def learn_procedure(self, name: str, steps: List[str], triggers: List[str]): """学习一个新流程""" await self.graph.execute( """ CREATE (p:Procedure {name: $name}) WITH p UNWIND $steps as step_text CREATE (s:Step {text: step_text}) CREATE (p)-[:HAS_STEP]->(s) WITH p UNWIND $triggers as trigger CREATE (t:Trigger {text: trigger}) CREATE (t)-[:ACTIVATES]->(p) """, name=name, steps=steps, triggers=triggers ) async def recall_procedure(self, context: str) -> List[dict]: """根据上下文检索相关流程""" result = await self.graph.execute( """ MATCH (t:Trigger)-[:ACTIVATES]->(p:Procedure)-[:HAS_STEP]->(s:Step) WHERE $context CONTAINS t.text RETURN p.name as procedure, collect(s.text) as steps """, context=context ) return [dict(r) for r in result] 记忆管理器：统一调度 class MemoryManager: """统一记忆管理器""" def __init__(self): self.working = WorkingMemory() self.short_term = ShortTermMemory() self.semantic = SemanticMemory(...) self.episodic = EpisodicMemory(...) self.procedural = ProceduralMemory(...) self.recall_count = 0 async def think(self, user_input: str, user_id: str) -> str: """带记忆的思考过程""" # 1. 从长期记忆检索相关信息 semantic_results = await self.semantic.recall(user_input, top_k=3) episodic_results = await self.episodic.recall_similar( user_id, await self.embed(user_input), top_k=3 ) procedures = await self.procedural.recall_procedure(user_input) # 2. 组装上下文 context_parts = [] if self.short_term.summary: context_parts.append(self.short_term.to_context()) if semantic_results: context_parts.append("相关知识:\n" + "\n".join(f"- {r['content']}" for r in semantic_results)) if episodic_results: context_parts.append("历史交互:\n" + "\n".join(f"- {r['input']} → {r['output']}" for r in episodic_results)) if procedures: context_parts.append("可用流程:\n" + "\n".join(f"- {p['procedure']}: {p['steps']}" for p in procedures)) context = "\n\n".join(context_parts) # 3. 注入工作记忆 if context: self.working.add("system", f"记忆上下文:\n{context[:4000]}") self.working.add("user", user_input) # 4. 生成回复 response = await llm_call(self.working.get_context()) self.working.add("assistant", response) # 5. 异步写入记忆 asyncio.create_task(self._consolidate(user_input, response, user_id)) return response async def _consolidate(self, input: str, output: str, user_id: str): """记忆巩固: 筛选重要信息写入长期记忆""" # 判断是否值得记住 importance = await self._assess_importance(input, output) if importance > 0.7: # 高重要性: 写入语义记忆 + 情景记忆 await self.semantic.remember( f"用户说了: {input}\n助手回答: {output}", metadata={"importance": importance, "user_id": user_id} ) # 总是记录情景记忆 await self.episodic.record({ "id": str(uuid4()), "user_id": user_id, "input": input, "output": output, "outcome": "success", "embedding": await self.embed(input), }) async def _assess_importance(self, input: str, output: str) -> float: """评估信息重要性 0-1""" prompt = f"""评估以下交互的信息重要性(0-1): 输入: {input[:200]} 输出: {output[:200]} 评分标准: - 0.9+: 用户明确要求记住/涉及核心偏好 - 0.7-0.8: 包含重要事实或决策 - 0.5-0.6: 有一定参考价值 - 0.3-0.4: 闲聊但含少量信息 - 0.0-0.2: 纯闲聊无信息 只输出数字:""" return float(await llm_call(prompt, max_tokens=10)) 遗忘机制 class ForgettingMechanism: """遗忘机制: 防止记忆膨胀""" async def decay(self, user_id: str): """基于时间的记忆衰减""" await self.db.execute( """ UPDATE semantic_memory SET importance = importance * 0.95 -- 每天衰减5% WHERE user_id = $1 AND last_accessed < NOW() - INTERVAL '1 day' """, user_id ) # 删除低重要性记忆 await self.db.execute( """ DELETE FROM semantic_memory WHERE user_id = $1 AND importance < 0.1 """, user_id ) async def deduplicate(self, user_id: str): """去重: 合并语义相似的记忆""" # 找到相似度 > 0.9 的记忆对 duplicates = await self.db.fetch( """ SELECT a.id as id_a, b.id as id_b, a.content as content_a, b.content as content_b FROM semantic_memory a JOIN semantic_memory b ON a.id < b.id WHERE a.user_id = $1 AND b.user_id = $1 AND a.embedding <=> b.embedding < 0.1 """, user_id ) for dup in duplicates: # 合并: 保留更详细的，删除另一个 longer = dup['content_a'] if len(dup['content_a']) > len(dup['content_b']) else dup['content_b'] shorter_id = dup['id_b'] if len(dup['content_a']) > len(dup['content_b']) else dup['id_a'] await self.db.execute("DELETE FROM semantic_memory WHERE id = $1", shorter_id) async def compress_old_memories(self, user_id: str, days: int = 30): """压缩旧记忆: 把多条相关记忆合并为一条摘要""" old_memories = await self.db.fetch( """ SELECT content FROM semantic_memory WHERE user_id = $1 AND created_at < NOW() - INTERVAL '$2 days' ORDER BY created_at """, user_id, days ) if len(old_memories) < 10: return # 用 LLM 合并 memories_text = "\n".join(m['content'] for m in old_memories) summary = await llm_call( f"将以下记忆合并为简洁摘要，保留关键信息:\n{memories_text}" ) # 删除旧记忆，写入摘要 await self.db.execute( "DELETE FROM semantic_memory WHERE user_id = $1 AND created_at < NOW() - INTERVAL '$2 days'", user_id, days ) await self.semantic.remember(summary, metadata={"compressed": True}) MemGPT / Letta 架构参考 MemGPT (现 Letta) 的核心思想是把操作系统的虚拟内存概念引入 LLM： ...
