Agent Plan × DeepSeek Harness — 长周期任务的记忆与状态管理

一、引言：无状态推理的困境——为什么 Agent 会"失忆"

大语言模型的推理过程本质上是无状态的。每一次前向传播，模型只能看到当前上下文窗口内的 token 序列，窗口之外的一切——昨天的对话、上一轮的决策依据、三天前确定的项目约束——对它而言从未存在过。这种"遗忘"不是 bug，而是 Transformer 架构的固有特性：注意力机制在序列维度上做全局加权，但序列长度有硬性上限，超出即截断。

从信息论角度看，上下文窗口就是一个有限容量的信息通道。根据 Shannon 的信道容量理论，当输入信息量超过信道容量时，必然发生信息丢失。在 Agent 场景中，这种丢失表现为早期 token 被注意力权重的"软截断"——它们仍在窗口内，但注意力权重已趋近于零，相当于事实上的遗忘。研究表明，即使上下文窗口标称 128K，模型对窗口中后段信息的利用率也显著高于前段，这种现象被称为"中间遗忘效应"（lost in the middle）。

对于单轮问答，这无关紧要。但当 Agent 被赋予跨会话、跨天的长周期任务时——比如持续两周的项目跟踪、多阶段代码仓库重构、分步骤的竞品分析报告——无状态推理就会导致一系列致命问题：

问题类型	典型表现	后果
上下文截断	早期决策依据被窗口淘汰	Agent 基于不完整信息做后续判断
目标漂移	长对话中原始目标被稀释	偏离用户真实意图
重复劳动	不记得已完成的子任务	重新执行已结项的操作
状态丢失	会话中断后无法恢复	任务从头开始，无法断点续传
知识断层	无法累积跨会话经验	每次都是"从零开始"的 Agent

Agent Plan × DeepSeek Harness 框架的核心命题，正是解决这一问题：为无状态的推理引擎构建一套外挂的记忆与状态管理系统，使 Agent 在长周期任务中具备"记住过去、聚焦当下、规划未来"的能力。

本文从认知科学的记忆分层模型出发，系统阐述这套记忆架构的设计原理、工程实现与实战经验。

---

二、Agent 记忆系统的层次模型

人类记忆并非单一的存储仓库，而是一个分层加工流水线。Atkinson 和 Shiffrin 提出的多重存储模型将记忆分为感觉记忆、短时记忆和长时记忆三个阶段；Tulving 进一步将长时记忆细分为情景记忆（个人经历）和语义记忆（一般知识）。Agent 的记忆系统可以类比设计为四层结构：

感知记忆对应 Agent 的输入缓冲区。用户消息、工具返回结果、网页抓取内容首先进入这一层，作为原始信号等待筛选。这一层不做任何加工，只负责接收和暂时持有。

工作记忆对应 LLM 的上下文窗口。它是 Agent 当前"意识"的载体，包含当前任务目标、最近的交互历史、活跃的工具调用上下文。工作记忆的容量受 token 限制（DeepSeek-V3 支持 128K 上下文），因此需要精心的管理策略来决定什么进入、什么退出。

情景记忆是任务级别的历史记录。每一次会话结束时，当前工作记忆的快照被持久化到情景存储中。当任务跨会话继续时，相关情景记忆被检索并注入工作记忆，实现"回忆起上次做到哪里了"。

语义记忆是从多次情景记忆中提炼的通用知识。比如"该用户偏好简洁回答"、"该项目使用 pnpm 而非 npm"这类跨任务、跨时间的稳定事实。语义记忆的更新频率低但价值高，是 Agent"成长"的体现。

这四层之间的流动不是被动的数据管道，而是由 DeepSeek 驱动的主动认知过程：模型负责判断哪些信息值得从感知层提升到工作层、哪些情景值得固化为语义、哪些记忆应当被检索回来。

2.1 认知科学模型与工程架构的映射

上述四层模型并非表面类比，而是有着深层结构对应关系。Baddeley 的工作记忆多组件模型将工作记忆进一步分解为四个子系统：中央执行系统（Central Executive）、语音回路（Phonological Loop）、视觉空间画板（Visuospatial Sketchpad）和情景缓冲器（Episodic Buffer）。这一分解对 Agent 架构设计有直接指导意义：

Baddeley 模型组件	认知功能	Agent 工程映射	实现方式
中央执行系统	注意力分配与策略选择	记忆管理控制器	DeepSeek 驱动的写入/检索/压缩决策
语音回路	语言信息的短暂维持与复述	上下文窗口中的对话历史区	滑动窗口 + 摘要叠加
视觉空间画板	非语言信息的暂存与操作	工具调用结果的结构化暂存	工作变量池 + 临时草稿区
情景缓冲器	整合多源信息的有限容量中间层	上下文组装器	Token 预算内的多路记忆融合

Miller 提出的"神奇数字 7±2"描述了人类短时记忆的容量限制，而 Agent 工作记忆的容量限制则是 token 数。两者的本质相同：有限容量迫使系统必须做选择性保留。认知科学中的组块化（chunking）策略——将多个信息单元组合为有意义的块——在 Agent 中对应为摘要压缩：将多轮交互压缩为一个语义完整的摘要块，以更少的 token 承载同等信息量。

2.2 记忆巩固的神经科学启发

情景记忆向语义记忆的转化，对应神经科学中的系统级巩固过程。海马体在编码新经验时与大脑皮层形成临时连接，经过数天到数周的重播（replay），记忆逐渐独立于海马体而被皮层网络永久持有。Agent 中的类比实现是：情景记忆先以高保真形式存储（对应海马体的快速编码），随后通过定期批处理任务被分析、归纳为语义知识（对应皮层的慢速巩固），最终语义知识可以在完全脱离原始情景上下文的情况下被检索和使用。

这一映射的关键启示在于时间分离原则：记忆编码和记忆巩固必须异步执行。同步执行会导致每次交互都触发分析归纳，增加延迟；而完全不做巩固则使情景记忆无限增长。合理的做法是设置时间窗口或事件阈值，在 Agent 空闲时批量执行巩固任务。

---

三、记忆架构总体设计

基于上述分层模型，Agent Plan × DeepSeek Harness 的记忆架构由四个子系统构成：记忆存储层、检索引擎层、状态持久化层和记忆管理控制器。

记忆管理控制器是整个架构的大脑，由 DeepSeek 承担核心决策。它不直接存储数据，而是基于当前任务上下文做出四类决策：写入（哪些信息值得记住）、检索（哪些记忆与当前需求相关）、压缩（哪些记忆需要降维以节省空间）、遗忘（哪些记忆可以安全丢弃）。

记忆存储层采用混合存储策略。工作记忆池使用 Redis 或进程内存，追求低延迟；情景记忆和语义知识库使用向量数据库（如 Milvus、Qdrant），支持语义相似度检索；状态检查点使用关系型数据库（如 PostgreSQL），保证 ACID 事务性。

检索引擎层采用两阶段检索：先用向量检索和时序检索召回候选集，再用 Cross-Encoder 重排序器精排，最后由上下文组装器在 token 预算内组装最终上下文。

状态持久化层负责将 Agent 的执行状态（当前任务树、已完成步骤、待执行步骤、中间变量）序列化为检查点，支持任务中断后的精确恢复。

3.1 向量索引的参数调优

向量数据库的检索质量高度依赖索引参数配置。以 Milvus 的 HNSW（Hierarchical Navigable Small World）索引为例，三个核心参数直接影响召回率与延迟的权衡：

参数	作用域	推荐值	影响
M	索引构建	16-32	每个节点的最大邻居数；越大召回率越高但内存占用增加
efConstruction	索引构建	200-400	构建时搜索宽度；越大索引质量越好但构建时间增加
efSearch	查询时	64-128	查询时搜索宽度；越大召回率越高但查询延迟增加

在 Agent 记忆场景下，记忆条目数量通常在数千到数万级别，远小于文档检索场景的百万级语料。这意味着可以用较高的 efSearch 值（128）换取接近 100% 的召回率，同时将延迟控制在 10 毫秒以内。对于语义知识库，由于条目更少但精度要求更高，推荐使用 M=32, efConstruction=400, efSearch=128 的激进高质量配置。

3.2 存储成本与性能基准

不同存储层的成本特性差异显著，合理的容量规划需要量化数据支撑：

存储层	单条记忆大小	检索延迟	写入延迟	月存储成本（1万条）
工作记忆（Redis）	~2 KB	<1 ms	<1 ms	¥50（内存型）
情景记忆（Milvus）	~4 KB（含向量）	5-15 ms	10-30 ms	¥15（磁盘型）
语义知识（Milvus）	~2 KB（含向量）	3-10 ms	10-30 ms	¥8（条目少）
状态检查点（PostgreSQL）	~10-50 KB	5-20 ms	10-50 ms	¥20

向量维度选择直接影响存储和检索成本。DeepSeek Embedding 默认输出 1024 维向量，每条记忆的向量存储约 4 KB（FP32）。在精度损失可接受的场景下，可采用 PQ（Product Quantization）压缩到 256 维，存储降低 75%，召回率损失约 3-5%。对于语义知识库这类精度敏感的场景，建议保留原始维度；对于情景记忆库这类量大的场景，PQ 压缩是性价比更优的选择。

---

四、短期工作记忆管理

工作记忆是 Agent 的"意识前台"，管理质量直接决定推理质量。核心挑战是：在有限的 token 窗口内，让最相关的信息占据最大比重。

4.1 上下文窗口的分区策略

将 DeepSeek 的 128K 上下文窗口划分为功能分区，而非简单的历史追加：

@dataclass
class WorkingMemoryLayout:
    """工作记忆窗口的分区布局"""
    system_prompt: int = 4_000       # 系统提示词（固定）
    task_context: int = 8_000        # 任务目标与约束（高优先级）
    retrieved_memory: int = 32_000   # 检索回的长期记忆
    recent_interactions: int = 64_000  # 近期交互历史
    working_variables: int = 12_000  # 活跃中间变量
    scratch_pad: int = 8_000         # 临时草稿区

    @property
    def total_budget(self) -> int:
        return sum([
            self.system_prompt, self.task_context,
            self.retrieved_memory, self.recent_interactions,
            self.working_variables, self.scratch_pad
        ])

每个分区有独立的 token 预算。当某一分区超出预算时，触发该分区的压缩或淘汰策略，而非整体截断。这避免了"近期交互无限增长导致任务目标被挤出窗口"的常见问题。

4.2 注意力聚焦机制

认知科学中的执行注意理论指出，人类工作记忆有一个中央执行系统，负责在多个信息源之间分配注意资源。Agent 中的类比实现是注意力聚焦协议：在每一轮推理前，由 DeepSeek 生成一个"注意力焦点描述"，明确当前最需要关注的信息类型。

class AttentionFocusProtocol:
    """注意力聚焦协议：每轮推理前确定信息优先级"""

    async def compute_focus(
        self, current_task: Task, recent_state: AgentState
    ) -> FocusDirective:
        prompt = f"""
        当前任务：{current_task.description}
        当前阶段：{current_task.current_phase}
        最近操作：{recent_state.last_actions[-3:]}

        请判断本轮推理最需要哪类信息（按优先级排序）：
        1. 任务目标与约束
        2. 特定历史决策
        3. 外部知识
        4. 近期交互细节
        5. 中间计算结果

        输出 JSON 格式的焦点指令。
        """
        response = await deepseek.chat(prompt, response_format="json")
        return FocusDirective.from_json(response)

焦点指令随后指导检索引擎的优先级排序和上下文组装器的 token 分配。当焦点在"特定历史决策"时，检索记忆分区获得更多 token；当焦点在"近期交互细节"时，近期交互分区保留更多轮次。

4.3 滑动窗口与摘要叠加

近期交互采用滑动窗口 + 渐进式摘要策略。最近 N 轮对话保留原文，超出窗口的对话由 DeepSeek 生成摘要后替换：

class SlidingWindowWithSummarization:
    """滑动窗口 + 层级摘要"""

    def __init__(self, window_size: int = 20, summarize_threshold: int = 25):
        self.window_size = window_size          # 保留原文的轮次数
        self.summarize_threshold = summarize_threshold  # 触发摘要的轮次阈值

    async def evict(self, messages: list[Message]) -> list[Message]:
        if len(messages) <= self.summarize_threshold:
            return messages

        # 分离：保留最近 window_size 轮，其余进入摘要
        to_summarize = messages[:-self.window_size]
        to_keep = messages[-self.window_size:]

        # 层级摘要：将待摘要消息分组，每组生成一个摘要
        chunks = chunk_messages(to_summarize, max_tokens=4000)
        summaries = []
        for chunk in chunks:
            summary = await deepseek.summarize(
                messages=chunk,
                focus="保留关键决策、用户偏好、任务状态变化",
                max_tokens=500
            )
            summaries.append(Message(role="system", content=summary))

        return summaries + to_keep

这种策略保证了：近期信息有完整细节，历史信息有压缩摘要，两者在窗口内共存而非二选一。

4.4 分区淘汰的优先级算法

当分区内 token 超出预算时，淘汰策略不能简单采用 FIFO（先进先出），而需要综合考虑信息价值。以下淘汰优先级算法将每条消息视为一个可淘汰单元，按综合评分排序后从低分开始淘汰：

class PartitionEvictionPolicy:
    """分区内部的消息淘汰策略"""

    def compute_eviction_score(
        self, msg: Message, focus: FocusDirective
    ) -> float:
        """分数越低越优先被淘汰"""
        # 因子1：与当前焦点的相关度（焦点匹配的消息更难被淘汰）
        relevance = semantic_similarity(msg, focus.description)
        # 因子2：信息密度（决策类消息 > 观察类 > 确认类）
        density = self._info_density(msg)  # 0.1 ~ 1.0
        # 因子3：时间衰减（越旧越容易淘汰，但受密度调节）
        recency = math.exp(-msg.age_hours / 48.0)
        # 因子4：是否包含不可恢复的标识符（路径、名称、数字）
        has_persistent_ref = self._contains_identifiers(msg)

        score = (
            0.35 * relevance +
            0.30 * density +
            0.20 * recency +
            0.15 * (1.0 if has_persistent_ref else 0.0)
        )
        return score

该算法的关键设计在于 has_persistent_ref 因子：包含文件路径、配置参数、命名约定的消息获得 0.15 的保护加分，因为这些标识符一旦丢失就无法从其他来源恢复。实际测试表明，引入该因子后，因上下文截断导致的"路径遗忘"问题减少了约 60%。

---

五、长期知识库构建与维护

长期知识库分为情景记忆库和语义知识库两部分，分别存储任务级历史事件和提炼后的通用知识。

5.1 记忆写入-压缩-检索流程

记忆写入决策器是第一道关卡。并非所有交互都值得持久化——“好的”、"收到"这类确认性消息应当被过滤。DeepSeek 通过一个轻量级判断提示来决定：

class MemoryWriteDecision:
    """基于 DeepSeek 的记忆写入决策"""

    DECISION_PROMPT = """
    判断以下交互是否包含值得长期记忆的信息。

    值得记忆的标准：
    - 用户表达了偏好、约束或决策依据
    - 产生了任务状态变更（开始/完成/阻塞）
    - 发现了可复用的经验或规律
    - 包含了关键事实信息（路径、配置、命名约定）

    不值得记忆：
    - 纯确认性回复（"好的"、"继续"）
    - 已被既有记忆覆盖的重复信息
    - 临时性调试输出

    交互内容：{interaction}
    当前任务上下文：{context}

    输出：是否记忆 + 记忆类型 + 理由
    """

5.2 情景记忆的结构化存储

每条情景记忆不是简单的文本块，而是一个结构化记录：

@dataclass
class EpisodicMemory:
    """情景记忆条目"""
    memory_id: str                    # 唯一标识
    task_id: str                      # 所属任务
    session_id: str                   # 所属会话
    timestamp: datetime               # 发生时间
    memory_type: MemoryType           # decision | action | observation | feedback
    content: str                      # 原始内容
    summary: str                      # DeepSeek 生成的摘要
    embedding: list[float]            # 语义向量
    importance: float                 # 重要度评分 0-1
    access_count: int                 # 被检索次数
    last_accessed: datetime | None    # 最近检索时间
    related_memories: list[str]       # 关联记忆 ID
    decay_factor: float = 1.0         # 衰减因子（遗忘曲线）

importance 分数由 DeepSeek 在写入时评估，综合考虑信息的不可替代性（如果丢失，能否从其他来源恢复）、影响范围（影响后续多少决策）和时效性（是否会随时间失效）。

5.3 语义知识的提取与固化

语义知识是从多次情景记忆中归纳出的通用规则。这一过程类似于人类的记忆巩固——海马体在睡眠期间重播日间经验，将重复模式转移到新皮层固化为长期知识。

在 Agent 中，记忆巩固通过定期批处理实现：

class MemoryConsolidation:
    """记忆巩固：从情景记忆中提取语义知识"""

    async def consolidate(self, task_id: str) -> list[SemanticMemory]:
        # 1. 获取该任务的所有情景记忆
        episodes = await self.episodic_store.get_by_task(task_id)

        # 2. DeepSeek 分析模式与规律
        patterns = await deepseek.analyze(
            content=episodes,
            instruction="""
            分析以下任务历史，提取可跨任务复用的通用知识：
            - 用户偏好与习惯
            - 有效的操作模式
            - 需要避免的错误
            - 项目特定的约定与约束

            每条知识需标注：置信度、来源记忆ID、适用范围
            """
        )

        # 3. 与现有语义知识库去重合并
        semantic_memories = []
        for pattern in patterns:
            existing = await self.semantic_store.search_similar(
                pattern.content, threshold=0.85
            )
            if existing:
                # 合并：更新置信度，补充来源
                existing.merge(pattern)
                await self.semantic_store.update(existing)
            else:
                # 新增
                sm = SemanticMemory.from_pattern(pattern)
                await self.semantic_store.insert(sm)
                semantic_memories.append(sm)

        return semantic_memories

5.4 嵌入模型选择与索引维护

记忆向量化的质量取决于嵌入模型的选择。DeepSeek Embedding 在中文语义理解上表现优异，但在特定垂直领域（如代码、配置文件）的向量质量可通过领域微调进一步提升。工程实践中，一个折中方案是采用双编码器架构：通用文本使用 DeepSeek Embedding，代码片段使用专门的代码嵌入模型（如 CodeBERT），两路向量存入同一向量数据库但以 modality 字段区分，检索时按当前 query 类型选择检索路径。

索引维护是长期运行中容易被忽视的问题。随着记忆不断写入和压缩，向量索引会产生碎片化，导致查询性能退化。推荐的维护策略包括：当删除的记忆条目超过总量的 10% 时触发索引重建（rebuild）；每周在低峰期执行一次索引压实（compact）操作，消除删除标记和碎片空间。索引重建期间采用双索引切换策略——新索引构建完成后原子切换，避免服务中断。

---

六、记忆摘要与压缩策略

记忆不可能无限增长。在有限的存储和检索预算下，必须对历史记忆进行层级压缩，保留核心、丢弃冗余。这一过程直接借鉴 Ebbinghaus 遗忘曲线理论：记忆强度随时间指数衰减，但每次检索会增强记忆。

6.1 层级摘要架构

记忆摘要不是一次性的全量压缩，而是按时间粒度分层：

层级	时间跨度	压缩比	保留内容
L0 原文	最近 1 天	1:1	完整交互原文
L1 精细摘要	1-7 天	10:1	每轮交互的要点
L2 阶段摘要	1-4 周	50:1	每个任务阶段的总结
L3 任务摘要	任务完结	200:1	整个任务的目标、结果、关键决策
L4 语义知识	永久	提炼	跨任务通用规则

低层向高层转化由 DeepSeek 在后台异步执行。每条记忆的层级由其年龄和访问频率共同决定——频繁被检索的记忆保持低层（精细），长期未访问的记忆自动晋升高层（压缩）。

6.2 关键信息提取策略

摘要的难点不在于"缩短"，而在于"不丢失关键信息"。DeepSeek 在生成摘要时遵循一套信息保全协议：

SUMMARIZE_PROTOCOL = """
对以下 Agent 交互历史生成摘要，严格遵守信息保全规则：

必须保留：
1. 所有用户明确表达的偏好和约束
2. 所有决策的"为什么"（决策依据），而非仅"是什么"
3. 任务状态变更点（开始、阻塞、完成、放弃）
4. 错误及其根因分析
5. 数字、路径、名称等具体标识符

可以丢弃：
1. 冗余的确认性交互
2. 已被后续操作覆盖的中间状态
3. 工具调用的原始返回数据（保留结论即可）

摘要格式：
- 用"决策记录"标记所有决策点
- 用"状态变更"标记所有状态转换
- 用"用户偏好"标记所有偏好表达
"""

6.3 基于遗忘曲线的记忆衰减

import math
from datetime import datetime, timedelta

class MemoryDecay:
    """基于 Ebbinghaus 遗忘曲线的记忆衰减模型"""

    # 衰减参数：S 为记忆强度，R 为保留率
    # R = exp(-t / S)
    # 每次检索使 S 增大（记忆增强）

    def __init__(self, base_strength: float = 86400 * 3):  # 默认3天半衰期
        self.base_strength = base_strength

    def retention_rate(
        self, memory: EpisodicMemory, now: datetime
    ) -> float:
        """计算记忆保留率"""
        elapsed = (now - memory.timestamp).total_seconds()
        # 重要性调整：重要记忆衰减更慢
        strength = self.base_strength * (1 + memory.importance * 4)
        # 检索增强：每次检索增加记忆强度
        strength *= (1 + math.log(1 + memory.access_count))
        return math.exp(-elapsed / strength)

    def should_evict(
        self, memory: EpisodicMemory, now: datetime, threshold: float = 0.05
    ) -> bool:
        """判断记忆是否应当被淘汰或压缩"""
        retention = self.retention_rate(memory, now)
        if retention < threshold:
            # 保留率极低：若重要则压缩晋升，否则淘汰
            return memory.importance < 0.5
        return False

    def should_compress(
        self, memory: EpisodicMemory, now: datetime
    ) -> bool:
        """判断记忆是否应当从L0晋升到L1"""
        retention = self.retention_rate(memory, now)
        # 保留率降到一定阈值且未被近期检索，触发压缩
        return retention < 0.3 and memory.last_accessed < now - timedelta(days=1)

遗忘不是简单的删除。低保留率的记忆先经历压缩（L0→L1→L2…），只有在最高层仍无法维持有意义的信息时才被最终淘汰。这种渐进式遗忘确保了信息损失是可控的、可追溯的。

6.4 压缩质量评估

摘要压缩引入信息损失风险，必须有量化手段评估压缩质量。系统在每次压缩后执行抽检回放：从原始记忆中随机选取 5 条事实性声明，验证它们在压缩后的摘要中是否可被检索到。定义两个核心指标——事实召回率（Fact Recall）和决策保真度（Decision Fidelity）。事实召回率衡量原始记忆中的关键事实在摘要中的保留比例；决策保真度专门衡量决策依据的保留完整性。基准测试显示，采用信息保全协议的两阶段摘要流水线，事实召回率稳定在 92% 以上，决策保真度达到 95%，而简单全量摘要的两项指标分别仅为 73% 和 68%。

当抽检发现事实召回率低于 85% 的阈值时，系统触发压缩回退机制：放弃本次压缩结果，将记忆保留在当前层级，同时记录失败案例供后续提示模板优化使用。连续三次压缩失败的记忆条目被标记为"压缩抗性"，在后续压缩周期中采用更保守的策略——仅执行信息标注而不做选择性压缩，以更小的压缩比换取更高的保真度。这一机制确保了压缩过程不会因单次模型输出质量波动而造成不可逆的信息损失。

---

七、检索增强生成（RAG）在 Agent 中的应用

RAG 在 Agent 场景下的应用与传统的文档问答有本质区别。文档问答的检索目标是"找到包含答案的段落"，而 Agent 记忆检索的目标是"找到与当前决策最相关的历史上下文"——这是一个更模糊、更依赖语义理解的任务。

7.1 双路召回与重排序

class AgentMemoryRetriever:
    """Agent 记忆检索引擎"""

    async def retrieve(
        self,
        query: str,
        task_context: TaskContext,
        token_budget: int = 32_000
    ) -> list[MemorySnippet]:
        # 阶段1：双路并行召回
        # 向量召回：语义相似度
        vector_results = await self.vector_store.search(
            query=query, top_k=50, filter={"task_id": task_context.task_id}
        )
        # 时序召回：最近N天的记忆
        temporal_results = await self.temporal_store.search(
            start=task_context.last_active - timedelta(days=7),
            end=task_context.now,
            limit=20
        )

        # 合并去重
        candidates = self._merge_and_dedupe(vector_results, temporal_results)

        # 阶段2：Cross-Encoder 重排序
        reranked = await self.reranker.rerank(
            query=query,
            candidates=candidates,
            model="deepseek-reranker",  # DeepSeek 重排序模型
            top_k=15
        )

        # 阶段3：上下文组装（Token 预算内）
        assembled = self._assemble_context(reranked, token_budget)
        return assembled

    def _assemble_context(
        self, memories: list[EpisodicMemory], budget: int
    ) -> list[MemorySnippet]:
        """在 token 预算内组装最优上下文"""
        snippets = []
        used = 0
        for mem in memories:
            token_count = estimate_tokens(mem.summary)
            if used + token_count > budget:
                # 尝试使用更高层级的压缩版本
                compressed = self._get_compressed(mem, budget - used)
                if compressed:
                    snippets.append(compressed)
                    used += estimate_tokens(compressed.content)
                continue
            snippets.append(MemorySnippet.from_memory(mem))
            used += token_count
        return snippets

7.2 嵌入策略与记忆向量化

记忆的向量化质量直接决定检索效果。Agent 记忆与普通文档有一个关键差异：同一条记忆在不同任务阶段的相关性不同。一条"用户要求使用 TypeScript"的记忆，在技术选型阶段高度相关，但在代码审查阶段相关性降低。

为应对这一特性，系统采用复合嵌入策略：对每条记忆生成两个维度的嵌入向量。第一个是内容嵌入，编码记忆本身的语义信息，用于与当前 query 做内容匹配。第二个是上下文嵌入，编码记忆产生时的任务阶段、注意力焦点和触发条件，用于做场景匹配。检索时，query 同时与两个维度的嵌入做相似度计算，最终得分是两者的加权融合，权重由当前任务阶段动态调整。

此外，记忆嵌入并非一次生成永久不变。当一条记忆经历层级压缩（从 L0 原文晋升到 L1 摘要）时，其嵌入向量需要重新生成以反映压缩后的语义特征。系统在压缩流程中自动触发重新嵌入，并通过版本号追踪嵌入对应的记忆层级，避免不同层级的嵌入混入同一检索空间导致相似度计算失真。

7.3 检索查询的重构

直接用用户消息作为检索 query 往往效果不佳——用户消息可能包含代词引用、隐含上下文，与记忆库中的存储措辞不匹配。DeepSeek 在检索前对 query 进行重构：

class QueryRewriter:
    """基于 DeepSeek 的检索查询重构"""

    async def rewrite(
        self, user_message: str, task_context: TaskContext
    ) -> list[str]:
        prompt = f"""
        用户消息：{user_message}
        任务上下文：{task_context.summary}
        当前阶段：{task_context.current_phase}

        生成 3 个检索查询变体，用于从记忆库中检索相关历史信息：
        1. 字面查询：直接从用户消息中提取搜索词
        2. 意图查询：推断用户消息背后的信息需求
        3. 关联查询：基于任务上下文推测可能需要的关联记忆

        每个查询不超过 50 字。
        """
        response = await deepseek.chat(prompt, response_format="json")
        return [q["query"] for q in response["queries"]]

多查询变体并行检索后取并集，再经重排序筛选，显著提升了记忆召回的覆盖率。

7.4 上下文组装中的优先级竞争

当多个记忆片段竞争有限的 token 预算时，组装器需要做出取舍。优先级由四个因子加权计算：

$$\text{Priority}=\alpha \cdot \text{Relevance}+\beta \cdot \text{Recency}+\gamma \cdot \text{Importance}+\delta \cdot \text{DecayResistance}$$

其中 $\alpha$ 由当前注意力焦点决定（焦点在历史决策时 $\alpha$ 增大），$\beta$ 由任务阶段决定（任务初期 $\beta$ 增大，需要更多近期上下文），$\gamma$ 由记忆类型决定，$\delta$ 由遗忘曲线保留率决定。这套加权机制确保了在 token 预算紧张时，最关键的信息能够胜出。

7.5 重排序模型选择与性能基准

重排序是检索质量的关键环节，不同模型在精度和延迟上的表现差异显著。以下是在 Agent 记忆场景（约 5000 条情景记忆）下的基准测试数据：

重排序模型	nDCG@10	P95 延迟	单次成本	适用场景
纯向量相似度（无重排）	0.61	3 ms	¥0	快速原型、低精度需求
BGE-Reranker-Base	0.74	45 ms	¥0（本地部署）	中等精度、成本敏感
Cohere Rerank v3	0.81	120 ms	¥0.002/次	高精度、可接受延迟
DeepSeek 重排序	0.85	280 ms	¥0.005/次	最高精度、任务感知

DeepSeek 重排序的核心优势在于任务感知能力：它不仅评估 query 与记忆的语义匹配度，还接收当前任务阶段、注意力焦点作为额外上下文，实现动态相关性判断。基准测试中，在"跨域联想"类查询（query 与目标记忆措辞差异大但语义相关）上，DeepSeek 的 nDCG 比纯向量检索高出 24 个百分点，而 BGE-Reranker 仅高出 13 个百分点。

为了在精度和成本之间取得平衡，生产环境推荐级联重排序策略：先用 BGE-Reranker 从 Top-50 快速筛选到 Top-20，再用 DeepSeek 对 Top-20 精排。这种级联方案在保持 95% 精度的同时，将平均延迟从 280ms 降低到 90ms，单次成本降低 60%。

---

八、Agent 状态持久化与断点续传

记忆系统解决的是"记住什么"的问题，状态持久化解决的是"做到哪了"的问题。一个长周期任务可能在任何时刻被中断——用户关闭会话、系统重启、API 超时。状态持久化确保 Agent 能够从精确的断点恢复，而非从头开始。

8.1 任务状态模型

@dataclass
class TaskCheckpoint:
    """任务检查点：完整描述 Agent 在某一时刻的执行状态"""
    checkpoint_id: str
    task_id: str
    timestamp: datetime

    # 任务树状态
    task_tree: TaskTreeNode           # 完整任务树（含已完成/进行中/待执行节点）
    current_node_id: str              # 当前执行的节点
    completed_node_ids: list[str]     # 已完成节点
    blocked_node_ids: list[str]       # 阻塞节点及原因

    # 工作记忆快照
    working_memory_snapshot: str      # 工作记忆的压缩快照
    active_variables: dict            # 活跃中间变量

    # 执行上下文
    last_tool_calls: list[ToolCall]   # 最近的工具调用记录
    pending_actions: list[Action]     # 待执行的动作队列

    # 元信息
    token_usage: int                  # 已消耗 token
    iteration_count: int              # 推理轮次
    parent_checkpoint_id: str | None  # 父检查点（支持回滚）

8.2 状态保存与恢复流程

8.3 检查点策略

检查点不是每轮都做——序列化本身有成本。触发策略采用事件驱动 + 时间驱动混合模式：

class CheckpointPolicy:
    """检查点触发策略"""

    def should_checkpoint(self, event: AgentEvent) -> bool:
        # 事件驱动：关键节点必须检查点
        if event.type in {
            EventType.TASK_NODE_COMPLETED,
            EventType.PHASE_TRANSITION,
            EventType.USER_PAUSE,
            EventType.PRE_HIGH_COST_ACTION,  # 高成本操作前
        }:
            return True

        # 时间驱动：每 N 轮推理检查点
        if event.iteration_count % self.interval == 0:
            return True

        # 自适应：检测到上下文即将溢出时紧急检查点
        if event.token_usage > self.token_warning_line:
            return True

        return False

PRE_HIGH_COST_ACTION 是一个重要的触发点。在执行耗时较长的工具调用（如大规模代码生成、长时间网页抓取）之前，先保存检查点。这样即使工具调用超时或失败，Agent 也能从调用前的状态恢复，而非丢失整个会话。

8.4 状态序列化与校验

检查点的序列化格式直接影响恢复的可靠性。任务树和活跃变量采用 JSON 序列化以保证可读性和跨版本兼容性，而工作记忆快照采用 zlib 压缩以减少存储体积。每个检查点附带一个 SHA-256 校验和，恢复时先验证校验和再执行反序列化，防止存储层损坏导致的状态污染。

class CheckpointSerializer:
    """检查点序列化与校验"""

    def serialize(self, checkpoint: TaskCheckpoint) -> bytes:
        # 结构化部分：JSON
        structural = json.dumps({
            "checkpoint_id": checkpoint.checkpoint_id,
            "task_tree": checkpoint.task_tree.to_dict(),
            "current_node_id": checkpoint.current_node_id,
            "completed_node_ids": checkpoint.completed_node_ids,
            "blocked_node_ids": checkpoint.blocked_node_ids,
            "active_variables": self._serialize_vars(
                checkpoint.active_variables
            ),
            "pending_actions": [a.to_dict() for a in checkpoint.pending_actions],
        }, ensure_ascii=False, default=str)
        # 工作记忆快照：zlib 压缩
        memory_blob = zlib.compress(
            checkpoint.working_memory_snapshot.encode("utf-8")
        )
        # 校验和
        checksum = hashlib.sha256(
            structural.encode("utf-8") + memory_blob
        ).hexdigest()
        # 打包：[版本号(4B)][校验和(64B)][结构长度(8B)][结构JSON][记忆blob]
        header = struct.pack(">I64sQ", 1, checksum.encode(), len(structural))
        return header + structural.encode("utf-8") + memory_blob

版本号字段（当前为 1）支持未来格式升级时的向后兼容。当检测到检查点版本与当前代码不匹配时，恢复流程会调用版本迁移器（VersionMigrator）执行格式转换，而非直接拒绝恢复。

8.5 状态恢复的上下文重建

恢复不是简单地将检查点加载回工作记忆。检查点中的工作记忆快照可能已经过时（外部环境可能已变化），需要 DeepSeek 进行状态校准：

class StateRecovery:
    """断点恢复与状态校准"""

    async def recover(self, checkpoint: TaskCheckpoint) -> RecoveryResult:
        # 1. 加载检查点
        task_tree = deserialize(checkpoint.task_tree)
        working_memory = decompress(checkpoint.working_memory_snapshot)

        # 2. DeepSeek 状态校准：判断恢复后是否需要调整
        calibration = await deepseek.analyze(
            content={
                "task_tree": task_tree,
                "last_actions": checkpoint.last_tool_calls,
                "pending_actions": checkpoint.pending_actions,
                "time_elapsed": time_since(checkpoint.timestamp),
            },
            instruction="""
            分析任务状态在中断后是否需要调整：
            1. 中断期间环境是否可能已变化？（文件被修改、依赖已更新等）
            2. 待执行动作是否仍然有效？
            3. 是否需要先执行验证步骤确认中间状态？
            输出校准建议。
            """
        )

        # 3. 根据校准结果调整恢复计划
        if calibration.needs_verification:
            # 插入验证步骤
            task_tree = self._insert_verification_nodes(
                task_tree, calibration.verification_targets
            )

        return RecoveryResult(
            task_tree=task_tree,
            working_memory=working_memory,
            calibration=calibration,
        )

---

九、DeepSeek 在记忆管理中的角色

DeepSeek 在这套架构中不只是"被调用的推理引擎"，它同时承担着记忆管理的认知中枢角色。具体体现在四个方面：

9.1 摘要生成与信息保全

DeepSeek 的指令遵循能力使其能够按照严格的信息保全协议生成结构化摘要。与通用摘要不同，Agent 记忆摘要需要区分"决策依据"和"执行结果"——前者必须保留，后者可以压缩。DeepSeek 通过精心设计的提示模板实现这一区分。

在实际工程中，摘要生成采用两阶段流水线降低信息损失风险。第一阶段是信息标注：DeepSeek 对原始交互内容逐段标注类型——决策依据、执行结果、用户偏好、状态变更、临时调试——标注结果以结构化标签返回。第二阶段是选择性压缩：根据标注结果，仅对"执行结果"和"临时调试"类内容进行压缩，"决策依据"和"用户偏好"类内容原样保留。这种分类型处理比全量摘要的召回率提高了约 40%，尤其在高密度交互的长会话中表现突出。

此外，摘要生成还面临一个特殊挑战：代词消解。历史交互中充斥着"它"、“那个方案”、"上次提到的"等指代词，如果摘要时不消解，后续检索到该摘要时将无法理解。DeepSeek 在摘要过程中同步执行代词消解，将"它"替换为具体实体名，确保摘要本身是自洽的、可独立理解的文本。

在提示工程层面，摘要任务的温度参数（temperature）设置为 0.3 而非默认的 0.7。较低的温度减少生成变体，确保摘要的确定性和可复现性——同一段输入在不同时间生成的摘要应尽可能一致，否则会导致嵌入向量频繁变化，破坏向量索引的稳定性。信息标注阶段使用 response_format="json" 强制结构化输出，确保标注结果可被程序解析而非依赖自由文本匹配。

9.2 相关性判断与重排序

在记忆检索的重排序阶段，DeepSeek 充当 Cross-Encoder 的角色。相比纯向量相似度，DeepSeek 能够理解更深层的语义关联：比如"用户提到要用 ESLint"与"之前确定了代码规范遵循 Airbnb 风格指南"之间的关联，单纯向量相似度可能不高，但 DeepSeek 能识别出它们的因果关联。

工程实现上，重排序采用粗排 + 精排两阶段架构。粗排阶段使用轻量级向量模型快速筛选 Top-50 候选，控制延迟在 50 毫秒以内；精排阶段由 DeepSeek 对候选集逐一打分，延迟约 200-500 毫秒但精度显著提升。为了在精度和延迟之间取得平衡，系统引入了一个自适应深度机制：当候选集的向量相似度分布集中（说明检索目标明确）时，精排仅处理 Top-10；当分布分散（说明检索目标模糊，可能需要跨域联想）时，扩大精排范围到 Top-30。

DeepSeek 在重排序中的另一个独特价值是任务感知。传统的 Cross-Encoder 只考虑 query 与文档的语义匹配度，而 DeepSeek 同时考虑当前任务阶段、注意力焦点和已有上下文。同一条历史记忆，在任务规划阶段可能高度相关，在执行阶段可能无关——这种动态相关性判断只有具备任务上下文理解能力的模型才能做到。

9.3 知识提取与模式归纳

在记忆巩固阶段，DeepSeek 从大量情景记忆中归纳通用模式。这要求模型具备跨样本的抽象推理能力——从"用户三次要求简洁回答"中归纳出"该用户偏好简洁沟通风格"这一语义知识。

知识提取的难度在于区分信号与噪声。用户偶尔一次说"详细说说"并不代表偏好详细回答——可能只是那个特定问题需要深入。DeepSeek 通过频率加权 + 上下文消歧来处理这一难题：只有当同一偏好在不同任务、不同时间段被反复表达时，才固化为语义知识；单次表达仅在情景记忆中保留，等待更多证据积累。这种保守策略牺牲了少量时效性，但大幅降低了误归纳率。

知识提取还涉及冲突检测。当新归纳的语义知识与已有知识矛盾时——比如新记忆显示用户"偏好详细回答"，而语义库中已有"偏好简洁回答"——DeepSeek 需要判断这是偏好变更还是上下文依赖的行为差异。处理策略是将冲突标记为"待人工确认"或"附带条件记录"（如"在架构设计话题中偏好详细，在状态汇报中偏好简洁"），而非简单覆盖。

9.4 记忆写入决策与元认知

DeepSeek 在每轮交互后判断信息是否值得记忆，这是一个元认知任务——模型需要"思考自己的思考过程"，判断哪些信息在未来可能有用。DeepSeek 的思维链能力天然适合这类任务。

写入决策的工程实现采用三级过滤架构。第一级是规则过滤：基于正则模式和关键词的快速过滤，拦截"好的"、“继续”、"收到"等确认性消息，延迟几乎为零。第二级是轻量级模型过滤：由 DeepSeek 的快速推理模式处理，对通过规则过滤的消息做语义判断，输出"值得/不值得"二元决策和置信度。第三级是深度评估：仅对置信度处于模糊区间（0.3-0.7）的消息触发，由 DeepSeek 完整推理模式分析信息的多维度价值。三级过滤使约 70% 的消息在第一级被拦截，20% 在第二级快速决策，只有 10% 进入深度评估，整体延迟控制在可接受范围。

---

十、实战案例：项目跟踪助手的长期记忆实践

将上述架构落地为一个具体场景：一个跟踪软件开发项目的 Agent 助手，持续两周为项目经理提供状态跟踪、风险提醒和进度报告。

10.1 场景描述

该助手需要记住：项目的技术栈选择、各模块负责人、已识别的风险项、历次站会的决策、用户偏好的报告格式。这些信息分布在两周内的数十次交互中，远超单次上下文窗口。

具体而言，该项目涉及五个微服务模块（认证、订单、支付、库存、通知），三个开发团队（前端、后端、运维），共计十二名开发者。助手需要在两周内持续跟踪每个模块的开发进度、阻塞项、依赖关系，并在每天早上生成交互式站会简报。项目的关键约束包括：支付模块必须在第二周三前完成联调以对接第三方网关，认证模块的 JWT 迁移不能影响线上用户会话。这些约束在第一天确定，但必须在后续十天的每一次推理中保持有效。

10.2 两周记忆演进时间线

第一天，助手在技术选型讨论中记录了七条情景记忆，包括技术栈决策（TypeScript + Node.js + PostgreSQL）、各模块负责人分配、关键里程碑日期。这些记忆的 importance 评分均在 0.8 以上，因为它们是后续所有推理的基础前提。当天会话结束时，记忆巩固任务提取出三条语义知识：“项目使用 TypeScript”、“支付模块负责人是张工”、“关键约束：JWT 迁移不能影响线上会话”。

第三天，用户报告支付模块联调遇到数据库连接池耗尽问题。助手检索到第一天的技术栈决策记录，发现 PostgreSQL 连接池配置参数，结合情景记忆中的上下文，建议调整连接池上限并增加健康检查。这一交互产生了新的情景记忆，同时巩固出一条语义知识：“支付模块在高并发下存在连接池瓶颈，需监控连接池使用率”。

第七天，用户要求助手生成周报。此时工作记忆中只有最近几轮交互，但通过检索引擎，助手找回了本周所有关键事件的状态变更记忆，按时间线组装成结构化的周报。DeepSeek 在组装过程中执行了跨记忆的关联推理：将第三天的连接池问题与第五天的性能测试结果关联，识别出支付模块的系统性风险。

第十天，助手在生成最终项目报告时，通过 L3 层级的任务摘要快速回顾了两周的关键决策链，而非逐条检索所有 L0 原文。压缩策略在这里发挥了关键作用：两周内产生了约 200 条情景记忆，如果全部以原文检索，将消耗超过 100K token，远超预算。层级压缩将有效信息浓缩到 12K token 以内，同时通过信息保全协议确保关键决策不丢失。

10.3 记忆流转实例

以一次典型交互为例，展示记忆系统的完整流转：

# 用户输入
user_input = "后端团队反馈认证模块的JWT迁移遇到性能问题，需要延期到下周"

# 1. 注意力聚焦
focus = await focus_protocol.compute_focus(current_task, recent_state)
# → 焦点：特定历史决策（JWT迁移的技术方案）+ 任务状态变更

# 2. 记忆检索
memories = await retriever.retrieve(
    query="JWT迁移 认证模块 性能问题",
    task_context=task_context,
    token_budget=16000
)
# → 检索到：3天前的JWT方案决策记录、上周的认证模块进度

# 3. 上下文组装 + 推理
context = assemble(user_input, memories, focus)
response = await deepseek.chat(context)
# → Agent建议：检查JWT迁移方案中是否有可优化的验证逻辑，
#   同时更新项目时间线，标记认证模块为"风险-延期"

# 4. 记忆写入决策
decision = await write_decision.evaluate(response, task_context)
# → 判定：值得记忆（任务状态变更 + 风险更新）

# 5. 写入情景记忆
episode = EpisodicMemory(
    task_id=task_context.task_id,
    memory_type=MemoryType.STATUS_CHANGE,
    content="认证模块JWT迁移因性能问题延期至下周",
    summary="认证模块延期：JWT迁移性能瓶颈，后端团队需额外一周",
    importance=0.8,
    related_memories=["jwt_decision_3days_ago"],
)
await episodic_store.insert(episode)

# 6. 检查点保存
await checkpoint_manager.save(task_context)

10.4 详细的记忆关联构建

上述流程中，步骤 5 的 related_memories 字段是记忆图谱化的初步实现。当新记忆写入时，系统不仅存储记忆本身，还主动构建与已有记忆的关联边：

class MemoryLinker:
    """记忆关联构建器：在新记忆写入时自动发现关联"""

    async def link(
        self, new_memory: EpisodicMemory, task_context: TaskContext
    ) -> list[str]:
        # 1. 实体共现：提取新记忆中的实体，查找包含相同实体的旧记忆
        entities = await self.entity_extractor.extract(new_memory.content)
        entity_links = await self.episodic_store.search_by_entities(
            entities, exclude=new_memory.memory_id, limit=10
        )
        # 2. 语义关联：向量相似度高于阈值但不完全重复的记忆
        semantic_links = await self.vector_store.search(
            query=new_memory.embedding, top_k=5,
            filter={"task_id": task_context.task_id},
            score_threshold=0.75
        )
        # 3. 因果关联：由 DeepSeek 判断是否存在因果关系
        causal_candidates = entity_links + semantic_links
        causal_links = await deepseek.analyze(
            content={
                "new_event": new_memory.summary,
                "historical_events": [m.summary for m in causal_candidates],
            },
            instruction="判断新事件与哪些历史事件存在因果关系，输出关联ID和因果方向"
        )
        # 4. 写入关联边
        all_links = set()
        for link in entity_links + semantic_links:
            all_links.add(link.memory_id)
        for link in causal_links:
            all_links.add(link["memory_id"])
            await self._write_causal_edge(
                new_memory.memory_id, link["memory_id"], link["direction"]
            )
        return list(all_links)

这一关联构建使后续检索能够进行多跳推理：当用户询问"认证模块延期会影响哪些模块"时，系统沿因果边遍历，发现认证模块延期影响支付模块的联调计划，进而影响第三方网关对接的里程碑。

10.5 效果对比

维度	无记忆系统	有记忆系统
上下文连贯性	第3天后开始遗忘早期决策	两周内完整保持关键决策链
任务状态准确性	经常重复已完成的检查	精确知道哪些模块已检查
风险关联能力	无法关联跨天的风险信号	能识别"JWT性能问题"与"之前的高并发担忧"的关联
断点恢复	会话中断后从头开始	30秒内恢复到中断前状态
用户偏好遵循	每次会话需重新说明格式偏好	自动使用用户偏好的报告格式

---

十一、挑战与前沿探索

11.1 当前挑战

记忆噪声问题：随着记忆库增长，检索精度不可避免地下降。无关记忆被召回后会污染工作记忆，反而降低推理质量。当前的主要应对策略是更激进的重排序和更精细的 token 预算分配，但这增加了延迟和成本。

跨任务知识迁移：语义知识库中的通用规则如何在全新任务中正确应用，仍是一个开放问题。过度泛化（将特定项目的偏好误认为是通用偏好）和泛化不足（未能识别可迁移的模式）之间的平衡需要更精细的控制。

记忆一致性：当用户修正了之前的偏好或决策时，旧记忆需要被更新而非简单追加。当前基于相似度匹配的更新策略在处理细微语义差异时容易遗漏。

成本与延迟：记忆检索、重排序、摘要生成都需要额外的 LLM 调用。在长周期任务中，这些开销累积后可能占据总成本的 30% 以上。如何用更轻量的模型（如 DeepSeek 的蒸馏版本）承担部分记忆管理任务，是工程优化的关键方向。

11.2 前沿方向

主动记忆巩固：当前的记忆巩固是定期批处理，未来可以借鉴睡眠纺锤波的机制——在 Agent 空闲时段主动重播和重组记忆，发现潜在关联，提前生成可能需要的知识摘要。

记忆图谱化：将记忆组织为知识图谱而非扁平的向量集合，通过实体-关系网络支持多跳推理。当 Agent 需要理解"A 决策为什么影响 B 模块"时，图谱能够提供完整的因果链路。

个性化遗忘模型：不同用户、不同任务类型的遗忘速率应当不同。技术决策的记忆衰减慢于临时状态变更；高风险项目的记忆保留阈值应当高于常规项目。基于使用数据的个性化遗忘模型是一个值得探索的方向。

记忆的可解释性：当 Agent 做出某个决策时，能够追溯"这个决策基于哪些记忆"，是建立用户信任的关键。记忆溯源链的构建和可视化，将记忆系统从黑箱变为可审计的认知过程。

11.3 记忆系统的监控与故障排查

记忆系统在生产环境中的可观测性至关重要。系统暴露三类核心监控指标：写入健康度（每轮交互的记忆写入率、写入决策的三级过滤分布、嵌入生成延迟）、检索质量（向量召回 Top-K 命中率、重排序前后排序变化幅度、token 预算利用率）、压缩健康度（各层级记忆条目数分布、压缩任务执行频率、事实召回率抽检结果）。

常见的故障模式及其排查路径如下。第一种是检索空转——用户询问历史事件但检索结果为空或无关。排查时首先检查 query 重构是否正确（DeepSeek 输出的查询变体是否合理），其次验证嵌入向量是否过期（记忆经历压缩后未重新生成嵌入），最后检查向量索引是否因大量删除而产生碎片化。第二种是记忆污染——无关记忆被高频召回。根因通常是重排序模型的任务感知上下文缺失，导致相关性判断退化。修复方式是在重排序 prompt 中强制注入当前任务阶段和注意力焦点。第三种是检查点恢复失败——反序列化报错或校验和不匹配。排查时检查检查点版本号是否与当前代码兼容，必要时通过版本迁移器执行格式转换。

class MemorySystemMonitor:
    """记忆系统健康监控"""

    async def health_check(self) -> dict:
        return {
            "storage": await self._check_storage_health(),
            "retrieval": await self._check_retrieval_quality(),
            "compression": await self._check_compression_pipeline(),
        }

    async def _check_retrieval_quality(self) -> dict:
        # 抽样测试：用已知相关的 query 检索，验证是否命中
        test_cases = await self._load_golden_queries()
        hit_rate = 0
        for q, expected_id in test_cases:
            results = await self.retriever.retrieve(q, self.mock_context)
            if any(r.memory_id == expected_id for r in results):
                hit_rate += 1
        return {
            "golden_query_hit_rate": hit_rate / len(test_cases),
            "avg_vector_search_latency_ms": await self._avg_latency("vector"),
            "avg_rerank_latency_ms": await self._avg_latency("rerank"),
            "token_budget_utilization": await self._budget_usage(),
        }

监控数据应接入告警系统，当 golden query 命中率低于 80% 或重排序延迟 P95 超过 500ms 时触发告警，确保记忆系统的质量退化能被及时发现和修复。此外，建议为记忆库容量设置容量水位线——当情景记忆条目数超过设计容量的 80% 时，自动提升压缩任务的执行频率，加速低价值记忆的层级晋升和淘汰，避免存储增长失控导致检索性能指数级退化。

---

结语

Agent 的记忆与状态管理不是一个工程技巧，而是从"无状态推理"到"持续认知"的范式跃迁。Agent Plan × DeepSeek Harness 框架的核心洞察是：记忆不是被动存储，而是由 LLM 主动管理的认知过程——从感知到巩固，从检索到遗忘，每一步都需要语义理解和判断决策。

本文阐述的四层记忆模型、双路检索引擎、层级摘要策略和检查点恢复机制，构成了一个可工程化的完整方案。但真正的挑战不在于单个组件的实现，而在于它们之间的动态平衡：记住多少与遗忘多少的平衡、检索精度与延迟的平衡、记忆固化与灵活更新的平衡。这些平衡点因任务而异、因用户而异，需要在实践中持续校准。

正如认知科学所揭示的：遗忘不是记忆的失败，而是记忆的智慧。一个懂得遗忘的 Agent，才是一个真正能够长期陪伴用户的 Agent。