AI Agent 记忆与上下文管理从短期记忆到长期知识的工程实现一、Agent 的金鱼记忆每次对话都从零开始大多数 AI Agent 每次对话都从零开始——不记得用户上次问了什么不知道之前做过什么决策不理解上下文的延续性。这种金鱼记忆严重限制了 Agent 的实用性用户每次都要重复背景信息Agent 无法积累经验跨会话的复杂任务无法执行。Agent 记忆系统的核心挑战是在有限的上下文窗口内管理无限的信息。短期记忆对话上下文受 Token 限制长期记忆知识库需要高效检索工作记忆当前任务状态需要实时更新。三类记忆协同让 Agent 从金鱼进化为有经验的助手。二、Agent 记忆架构graph TB subgraph 短期记忆 A[对话历史br/最近 N 轮对话] B[上下文窗口br/4K-128K Token] end subgraph 工作记忆 C[当前任务状态br/目标/进度/待办] D[工具调用结果br/API 返回/代码输出] end subgraph 长期记忆 E[向量数据库br/语义检索] F[知识图谱br/实体关系] G[用户偏好br/个性化配置] end A -- H[记忆编排器br/选择压缩注入] C -- H E -- H F -- H G -- H H -- I[LLM 推理br/带记忆上下文]三类记忆通过记忆编排器统一管理选择最相关的记忆片段、压缩冗余信息、注入 LLM 上下文窗口。编排器的目标是在有限的 Token 预算内提供最有用的记忆。三、实现3.1 短期记忆管理from dataclasses import dataclass from typing import List, Optional dataclass class Message: 对话消息 role: str # user/assistant/system content: str token_count: int 0 class ShortTermMemory: 短期记忆管理对话上下文窗口 def __init__(self, max_tokens: int 4096): self.max_tokens max_tokens self.messages: List[Message] [] self.current_tokens 0 def add(self, message: Message) - None: 添加消息 self.messages.append(message) self.current_tokens message.token_count # 超出窗口时压缩旧消息 if self.current_tokens self.max_tokens: self._compress() def get_context( self, max_tokens: int None ) - List[Message]: 获取当前上下文 budget max_tokens or self.max_tokens result [] used 0 # 从最新消息开始倒序填充 for msg in reversed(self.messages): if used msg.token_count budget: break result.insert(0, msg) used msg.token_count return result def _compress(self) - None: 压缩旧消息保留首尾摘要中间 if len(self.messages) 4: return # 保留系统消息 system_msgs [ m for m in self.messages if m.role system ] # 保留最近 2 轮对话 recent self.messages[-4:] # 中间消息生成摘要 middle self.messages[len(system_msgs):-4] if middle: summary self._summarize(middle) summary_msg Message( rolesystem, contentf[对话摘要] {summary}, token_countlen(summary) // 4, ) self.messages system_msgs [summary_msg] recent else: self.messages system_msgs recent self.current_tokens sum( m.token_count for m in self.messages ) def _summarize(self, messages: List[Message]) - str: 摘要中间对话简化实现 # 实际实现中调用 LLM 生成摘要 key_points [] for m in messages[-6:]: # 最近 6 条 key_points.append( f{m.role}: {m.content[:100]} ) return ; .join(key_points)3.2 长期记忆检索from dataclasses import dataclass from typing import List dataclass class MemoryEntry: 长期记忆条目 id: str content: str embedding: List[float] None metadata: dict None timestamp: float 0.0 access_count: int 0 class LongTermMemory: 长期记忆向量数据库 知识图谱 def __init__(self, embedding_fnNone): self.embedding_fn embedding_fn self.memories: List[MemoryEntry] [] self.entity_graph {} # 简化知识图谱 def store(self, entry: MemoryEntry) - None: 存储记忆 if self.embedding_fn and not entry.embedding: entry.embedding self.embedding_fn(entry.content) self.memories.append(entry) # 提取实体关系简化 self._extract_entities(entry) def retrieve( self, query: str, top_k: int 5, min_similarity: float 0.7, ) - List[MemoryEntry]: 检索相关记忆 if not self.embedding_fn: return self.memories[:top_k] query_embedding self.embedding_fn(query) # 计算相似度 scored [] for entry in self.memories: if entry.embedding: sim self._cosine_similarity( query_embedding, entry.embedding ) if sim min_similarity: scored.append((entry, sim)) # 按相似度排序 scored.sort(keylambda x: x[1], reverseTrue) # 更新访问计数 results [] for entry, score in scored[:top_k]: entry.access_count 1 results.append(entry) return results def _cosine_similarity( self, a: List[float], b: List[float] ) - float: 余弦相似度 import numpy as np a_arr np.array(a) b_arr np.array(b) norm_a np.linalg.norm(a_arr) norm_b np.linalg.norm(b_arr) if norm_a 0 or norm_b 0: return 0.0 return float(np.dot(a_arr, b_arr) / (norm_a * norm_b)) def _extract_entities(self, entry: MemoryEntry) - None: 提取实体关系简化 # 实际实现中使用 NER 模型 pass3.3 记忆编排器class MemoryOrchestrator: 记忆编排器选择压缩注入 def __init__( self, short_term: ShortTermMemory, long_term: LongTermMemory, context_budget: int 4096, ): self.short_term short_term self.long_term long_term self.context_budget context_budget def build_context( self, current_query: str ) - List[Message]: 构建完整的记忆上下文 budget self.context_budget context [] # 1. 系统提示固定占用 system_msg Message( rolesystem, content你是一个有帮助的 AI 助手。, token_count20, ) context.append(system_msg) budget - system_msg.token_count # 2. 长期记忆检索占用 20% 预算 lt_budget int(budget * 0.2) relevant_memories self.long_term.retrieve( current_query, top_k3, ) if relevant_memories: memory_text [相关记忆]\n \n.join( f- {m.content[:200]} for m in relevant_memories ) memory_msg Message( rolesystem, contentmemory_text, token_countlen(memory_text) // 4, ) if memory_msg.token_count lt_budget: context.append(memory_msg) budget - memory_msg.token_count # 3. 短期记忆占用剩余预算 st_messages self.short_term.get_context( max_tokensbudget ) context.extend(st_messages) return context四、Agent 记忆的 Trade-offs 分析上下文窗口 vs. 检索精度将更多记忆注入上下文窗口LLM 能看到更多信息但信息过多会分散注意力降低推理质量。建议长期记忆只注入 Top 3-5 条最相关的短期记忆保留最近 2-4 轮对话。记忆压缩的信息损失对话摘要会丢失细节——用户提到的具体数值、时间、人名可能在摘要中遗漏。解决方案是关键信息提取——在摘要前先提取关键实体和数值确保摘要保留核心事实。长期记忆的更新与遗忘长期记忆不是只增不减——过时的信息需要更新错误的信息需要删除。建议设置记忆 TTLTime-To-Live超过 TTL 的记忆自动降权或删除。高频访问的记忆权重更高低频访问的记忆逐渐遗忘。隐私与安全长期记忆存储用户的对话历史可能包含敏感信息密码、地址、财务数据。存储前必须脱敏检索时需要权限控制。企业场景建议本地部署向量数据库不使用云端服务。五、总结Agent 记忆系统的核心是在有限的上下文窗口内管理无限的信息。短期记忆管理对话上下文长期记忆提供知识检索工作记忆跟踪任务状态。记忆编排器统一管理三类记忆在 Token 预算内选择最有用的记忆注入 LLM。落地建议先实现短期记忆管理对话压缩再接入长期记忆向量数据库检索最后实现记忆编排器。长期记忆使用 Milvus 或 Qdrant 作为向量数据库对话压缩使用 LLM 生成摘要。监控记忆命中率和推理质量持续优化检索策略和压缩算法。