算法学习 Agent 的上下文记忆设计：别让每次对话都从零开始

算法学习 Agent 的上下文记忆设计：别让每次对话都从零开始

一、每次打开 Agent 都要重新自我介绍

使用 AI 辅助学习算法时，一个很让人沮丧的体验是：你昨天刚和 Agent 讨论过动态规划的「最优子结构」概念，今天打开对话，它又开始问你"你了解 DP 吗？"。

这不是模型的问题，是上下文记忆机制的问题。聊天类 Agent 的默认行为是：每次对话独立，不携带任何历史信息。但对于算法学习这种渐进式、有依赖关系的场景，上下文记忆是刚需——你不知道学习者昨天卡在了哪个概念上，就无法给出连贯的指导。

这篇文章讨论算法学习 Agent 的上下文记忆应该如何设计，包括记忆的存储结构、检索策略和遗忘机制。

二、记忆的分层架构

flowchart TD subgraph 短期记忆 A[当前对话历史] B[本轮讨论的概念] C[最近的纠错记录] end subgraph 长期记忆 D[知识掌握图谱] E[典型错误模式] F[学习进度与节奏偏好] end subgraph 工作记忆 G[当前题目信息] H[候选算法列表] I[中间推导状态] end 短期记忆 -->|摘要压缩后| 长期记忆 长期记忆 -->|按需检索| 工作记忆 工作记忆 -->|推理结束后清理| 短期记忆

三种记忆的定位不同：

• 短期记忆：保留最近几轮对话的完整上下文，不压缩。
• 长期记忆：结构化的知识状态，持久存储，按需检索。
• 工作记忆：当前推理任务需要的临时信息，任务完成后释放。

三、实现：记忆管理器

from dataclasses import dataclass, field from typing import Optional import time import json from collections import OrderedDict @dataclass class KnowledgeNode: """知识图谱节点：记录学习者在某个知识点上的掌握状态 设计思路： 每个知识点独立记录，包含掌握程度、最近练习时间、 错误历史等信息。知识点之间可以建立前置依赖关系。 """ concept_id: str # 知识点唯一标识，如 "dp.state_transfer" concept_name: str # 知识点名称，如 "状态转移方程" mastery_level: float # 掌握程度 0.0-1.0 last_practice_at: float # 最近练习时间戳 error_count: int = 0 # 累计错误次数 # 前置知识点：如状态转移依赖最优子结构 prerequisites: list[str] = field(default_factory=list) class ContextMemoryManager: """上下文记忆管理器 三种记忆的统一管理入口。 短期记忆用滑动窗口保留最近 K 轮对话。 长期记忆以知识图谱形式持久化存储。 工作记忆按任务生命周期动态创建和释放。 """ # 短期记忆保留的最近对话轮数 SHORT_TERM_WINDOW = 10 def __init__(self, user_id: str, storage_path: str = "./memory_store"): self.user_id = user_id self.storage_path = storage_path # 短期记忆：用 OrderedDict 实现 FIFO 淘汰 self._short_term: OrderedDict[int, str] = OrderedDict() # 长期记忆：核心是知识图谱 self._knowledge_graph: dict[str, KnowledgeNode] = {} # 工作记忆：临时存储，随任务创建和销毁 self._working_memory: dict[str, any] = {} # 加载持久化的长期记忆 self._load_long_term() def add_conversation(self, turn_id: int, message: str) -> None: """记录一轮对话到短期记忆""" self._short_term[turn_id] = message # 淘汰最旧的记录 while len(self._short_term) > self.SHORT_TERM_WINDOW: oldest_key = next(iter(self._short_term)) removed = self._short_term.popitem(last=False) # 淘汰前做一次摘要，写入长期记忆 self._summarize_and_persist(removed) def update_knowledge( self, concept_id: str, mastery_delta: float, prerequisites: Optional[list[str]] = None, ) -> None: """更新知识图谱中的掌握程度 Args: mastery_delta: 掌握程度变化，正值进步，负值表示遗忘或出错 """ if concept_id not in self._knowledge_graph: self._knowledge_graph[concept_id] = KnowledgeNode( concept_id=concept_id, concept_name=concept_id, mastery_level=0.3, # 初始掌握程度 last_practice_at=time.time(), ) node = self._knowledge_graph[concept_id] # 使用指数移动平均更新掌握程度，避免单次波动过大 alpha = 0.3 # 平滑系数 node.mastery_level = max(0.0, min(1.0, node.mastery_level + alpha * mastery_delta )) node.last_practice_at = time.time() if mastery_delta < 0: node.error_count += 1 if prerequisites: node.prerequisites = prerequisites self._save_long_term() def get_context_for_prompt(self, concept_id: Optional[str] = None) -> str: """为下一轮 LLM 调用构建上下文提示 组合短期记忆 + 相关长期记忆，生成一个信息丰富的 prompt。 """ parts = [] # 短期记忆：最近几轮对话原文 if self._short_term: recent = list(self._short_term.values())[-5:] parts.append("【近期对话】\n" + "\n".join(recent)) # 长期记忆：当前知识点的掌握状态 if concept_id and concept_id in self._knowledge_graph: node = self._knowledge_graph[concept_id] parts.append( f"【学习状态】{node.concept_name}: " f"掌握程度 {node.mastery_level:.0%}, " f"累计错误 {node.error_count} 次" ) # 附上未掌握的前置知识点 unfinished_prereqs = [ p for p in node.prerequisites if self._knowledge_graph.get(p, KnowledgeNode("", "", 0)).mastery_level < 0.6 ] if unfinished_prereqs: parts.append(f"【前置知识薄弱】{', '.join(unfinished_prereqs)}") return "\n\n".join(parts) if parts else "" def create_workspace(self, task_id: str, initial_state: dict) -> None: """为当前任务创建工作记忆空间""" self._working_memory[task_id] = initial_state def get_workspace(self, task_id: str) -> dict: """获取工作记忆空间""" return self._working_memory.get(task_id, {}) def release_workspace(self, task_id: str) -> None: """释放工作记忆""" self._working_memory.pop(task_id, None) def _summarize_and_persist(self, message: str) -> None: """将短期记忆中的对话压缩写入长期记忆""" # 简化实现：提取关键词和结论 # 生产环境应使用轻量 LLM 做摘要 keywords = self._extract_keywords(message) for kw in keywords: self.update_knowledge(kw, mastery_delta=0.02) def _extract_keywords(self, text: str) -> list[str]: """简易关键词提取（实际应使用 NLP 或 LLM）""" return [] # 占位 def _save_long_term(self) -> None: """持久化长期记忆""" pass # 实际上通过文件或数据库存储 def _load_long_term(self) -> None: """加载长期记忆""" pass

四、边界分析与权衡

4.1 记忆精度 vs 存储成本

保留每轮对话的完整原文，精度最高但存储成本大。对历史对话做摘要压缩，存储成本小但可能丢失细节。折中方案是：近期对话保留原文，远期对话保留摘要。

4.2 知识图谱的更新策略

当前使用指数移动平均更新掌握程度。这种方式的优点是平滑、抗噪声，缺点是反应慢——如果你突然"顿悟"了一个知识点，EMA 需要多轮才能追上真实水平。可以引入突变检测机制：如果连续 3 次正确率 100%，跳跃式提升掌握程度。

4.3 跨会话的记忆一致性

当用户同时打开多个设备与 Agent 对话时，记忆的同步一致性问题会出现。解决方案是在后端维护单一记忆源，前端只做展示。这涉及到分布式一致性的老问题。

4.4 遗忘机制的必要性

不是所有记忆都值得保留。三个月前的错误模式可能已经不再适用。遗忘机制有两种策略：时间衰减（越久远的记忆权重越低）和容量淘汰（记忆总数超限时淘汰最不重要的）。对于算法学习场景，时间衰减更合理。

五、总结

上下文记忆设计的核心矛盾是：信息量越大，检索越精准，但存储和计算成本越高。分层记忆架构（短期-长期-工作）是平衡这个矛盾的常用方案。对于算法学习 Agent 来说，知识图谱是最核心的长期记忆载体——它不仅记录"学了什么"，还记录"掌握到什么程度"以及"还有什么前置知识未掌握"。这种结构化的记忆，是 Agent 从「对答如流」升级为「因材施教」的关键。