Agent记忆层级机制：从上下文管理到Python实现

Agent 类应用遇到的最典型问题不是模型能力不够而是上下文管理能力不够。对话一长前面的指令、用户偏好、中间结论都可能被截断或者在注意力机制里被稀释。Prime Agent 的记忆层级机制就是为了应对这个问题而设计的它不把记忆看成一条无限增长的文本列表而是按生命周期和重要程度分成多个层级让 Agent 在工作时只加载当前任务真正需要的部分同时把高价值信息持久化保存按需召回。这篇文章会从记忆层级的概念讲起拆解写入、检索、更新、压缩的完整链路然后给出一个可以运行的最小 Python 实现最后补充参数调优、常见问题排查和生产环境落地建议。1. 记忆层级机制是什么先理解 Agent 为什么需要分层记忆1.1 上下文窗口不是无限存储大语言模型有固定的上下文窗口窗口越大能容纳的 token 越多但这不意味着把历史全部塞进 Prompt 就是最优方案。对话记录一旦变长输入 token 数量会持续增长推理耗时、响应延迟和成本都会跟着上升。更关键的是模型对一整段超长上下文的利用并不是均匀的中间位置的历史信息容易被忽略系统越早给出的约束也可能被后面的大量内容覆盖。Agent 场景比普通对话更复杂。系统提示里要放角色定义和工具说明执行过程中还会产生工具调用结果、中间状态、用户反馈、错误日志等。即使上下文窗口很大真正留给“记忆”的空间也有限。如果把所有历史都平铺在一起Agent 很容易忘记任务早期的关键约束或者把已经废弃的中间步骤当成当前事实。Prime Agent 的记忆层级机制解决的正是这个问题。它把记忆从单一文本流拆成不同层级的存储单元在每次推理前只选择当前任务相关的部分注入上下文。这样既控制输入长度也提高历史信息的信噪比。1.2 把记忆分成工作台、便签和档案柜一个直观的类比是办公桌上的三样东西工作台正在处理的任务、临时变量、当前步骤的执行结果。任务一旦结束这些内容就应该清空否则会干扰下一个任务。便签最近几天或几个会话内需要反复参考的信息比如用户偏好的表达风格、最近讨论过的方案结论。档案柜跨会话、跨项目的长期事实比如用户身份、业务规则、历史项目背景。这些内容不能全部常驻上下文但可以在需要时检索出来。在 Prime Agent 这类 Agent 系统中常见的层级设计方案如下层级存储内容生命周期典型实现working memory当前任务步骤、临时变量、正在处理的对象任务结束即清理内存字典或短生命周期缓存short-term memory最近几轮对话、近期用户偏好、短时结论会话级或数小时内存列表、Redis、带时间戳的数据库long-term memory用户画像、业务规则、项目历史、领域知识数天到数月甚至更久向量数据库、关系型数据库、对象存储summary memory对多条原始记忆的压缩摘要不定期更新摘要文本、结构化知识卡片不同项目中层级名称可能不一样但核心思想一致越靠近工作台的数据越快、越容易修改也越容易被淘汰越靠近档案柜的数据越稳定但必须在需要时通过检索才能回到上下文。1.3 记忆的生命周期不只是写入和读取如果把记忆机制只实现成“把文本存进数组再按相似度取回”它会在真实场景里快速失效。原因是记忆会过期、会重复、会相互矛盾也可能会随着时间失去价值。一套完整的记忆机制需要覆盖五个阶段采集从对话、工具结果、用户反馈中识别值得记忆的信息。编码把原始文本转成统一结构生成向量和元数据。存储按层级写入对应存储介质。检索根据当前任务和查询条件召回相关记忆。更新与遗忘调整记忆优先级清理过期内容压缩低价值信息。理解这五步就能明白为什么简单的“历史消息列表”不是记忆系统。真正的记忆系统是有取舍的什么该记、什么不该记、什么时候压缩、什么时候删除。Prime Agent 的层级机制本质上就是把这种取舍用工程方式固化下来。2. 核心链路拆解记忆从写入到召回的完整过程2.1 写入链路不是每句对话都值得记忆最容易犯的错误是“把每一轮对话都存起来”。这样短期记忆会迅速膨胀检索时也会被大量无关内容干扰。真实系统应该在写入前做判断。一条会话文本是否值得写入记忆通常可以从下面几个维度判断是否包含用户明确偏好或约束。例如“以后报告都用中文”属于必须长期保留的信息。是否包含任务相关事实。例如“数据库连接超时时间调整为 30 秒”需要进入当前任务或短期记忆。是否包含可以复用的结论。例如“该接口返回结构已经确认”可以进入短期记忆。是否是纯粹的寒暄或临时表达。例如“好的”“谢谢”一般不需要写入。写入链路的一种简单实现是规则过滤搭配重要性打分。规则先判断信息类型例如包含“偏好”“不要”“必须”等关键词时提高重要性然后再由程序或模型给出 0 到 1 的重要性分数。一条标准记忆条目在存储层大致长这样{ id: mem_9f2c1a, content: 用户偏好使用简洁的中文报告, layer: short_term, created_at: 1736000000, last_access_at: 1736000000, access_count: 0, importance: 0.7, source: conversation, metadata: { session_id: sess_abc, user_id: u_123 } }content是最终要注入上下文的文本layer决定这条记忆的存储层级importance决定它在压缩和淘汰时的优先级metadata用来做用户隔离、会话过滤、来源追溯。如果缺少这些元数据后续检索时只能做纯文本相似度匹配很难支持“只看当前用户”“只看最近一天”这类精确条件。2.2 编码与存储文本、向量、元数据三者分开记忆条目不是只存一个字符串。为了兼顾可读性、语义检索和条件过滤应该把文本、向量、元数据拆开存储原始文本用于展示和注入 Prompt。向量由嵌入模型生成用于语义相似度检索。元数据用于过滤和排序例如时间范围、用户 ID、来源类型、重要度。如果项目规模小可以用内存列表配合 NumPy 计算余弦相似度如果记忆量达到几十万条建议引入专用向量数据库并保留一个关系型表存元数据。向量数据库负责“找语义相近”关系型数据库负责“按条件过滤”两者配合才能获得高精度召回。这里要注意向量检索解决的是“语义相近”不等同于“业务正确”。例如用户问“上次那个订单后来怎么样了”向量检索可能召回一段关于“订单状态”的对话但如果不结合会话 ID 和订单号过滤结果仍然不可用。所以存储设计一定要把元数据作为一等公民。2.3 检索链路相关性、层级、时间衰减的加权检索不是简单地把用户问题拿去向量库搜索一次。在 Prime Agent 这类系统中完整的检索链路可以拆成四步查询改写。用户可能问“上次说的那个方案”需要先改写成具体的检索意图例如“项目方案 A 的关键结论”。条件过滤。根据当前会话 ID、用户 ID、时间范围、允许检索的层级把候选集缩小。语义召回。用改写后的查询向量计算相似度取前 N 个候选。重排。把相关性分数、重要度、时间衰减融合为一个最终分数。重排阶段可以使用一个平衡公式具体权重根据业务调整final_score 0.6 * similarity 0.3 * importance 0.1 * recency其中recency可以设计为recency 1 / (1 (now - last_access_at) / 86400)这样一条记忆如果长期没有被访问即使相关性较高也会排在前面但如果一条记忆刚被访问且重要度很高则更容易被选中。2.4 更新与遗忘让记忆随着使用“变热”或“变冷”记忆系统不能只增不改。每次检索到一条记忆都应该更新它的last_access_at和access_count。这些字段决定了记忆的“热度”。热度越高越不容易被压缩或淘汰热度越低越可能在容量不足时被合并成摘要。长期记忆阶段仍然会增长。用户画像会变化旧项目的结论可能不再适用历史对话摘要也可能和最新事实冲突。因此系统需要定期执行清理任务删除来源已经失效的记忆。把低重要度且长时间未访问的长期记忆标记为待删除或归档。把多条短期记忆合并成一条长期摘要。对互相矛盾的记忆做冲突处理保留更新时间更晚或置信度更高的一条。更新和遗忘策略的核心目标是控制存储增长同时让重要信息尽可能长久保留。如果没有这层机制记忆库会变成一座只进不出的垃圾山。3. 最小可运行实现用 Python 写一个记忆管理器3.1 环境准备与依赖下面的示例使用sentence-transformers生成嵌入向量使用numpy计算余弦相似度不依赖外部向量数据库。安装命令pip install sentence-transformers numpy首次运行MemoryStore时会下载嵌入模型因此需要机器可以访问模型下载地址。如果生产环境网络受限可以先将模型文件下载到本地再通过SentenceTransformer(/本地/模型目录)加载。本示例使用的模型是paraphrase-multilingual-MiniLM-L12-v2它对中文和英文都能处理模型体积适中。实际项目可以根据语种和速度要求选择其他模型但要注意一旦切换模型历史向量与新的向量可能不在同一语义空间必须重建索引。3.2 定义记忆条目和层级用Enum定义三个基础层级用MemoryItem定义记忆条目的统一结构import time import uuid from dataclasses import dataclass, field from enum import Enum class MemoryLayer(str, Enum): WORKING working SHORT_TERM short_term LONG_TERM long_term dataclass class MemoryItem: memory_id: str content: str layer: MemoryLayer created_at: float last_access_at: float access_count: int 0 importance: float 0.5 source: str conversation metadata: dict field(default_factorydict)memory_id用 UUID 生成避免多实例写入时冲突。created_at用于生命周期判断last_access_at和access_count用于热度排序importance用于压缩和淘汰时的保留优先级。3.3 实现轻量级向量存储MemoryStore负责生成向量、保存条目、按余弦相似度检索import numpy as np from sentence_transformers import SentenceTransformer class MemoryStore: def __init__(self, model_name: str paraphrase-multilingual-MiniLM-L12-v2): self.model SentenceTransformer(model_name) self.items [] def _embed(self, text: str) - np.ndarray: return self.model.encode(text, normalize_embeddingsTrue) def add(self, item: MemoryItem): vector self._embed(item.content) self.items.append({item: item, vector: vector}) def remove_by_ids(self, ids: set): self.items [ entry for entry in self.items if entry[item].memory_id not in ids ] def search(self, query: str, top_k: int 10): query_vec self._embed(query) scored [] for entry in self.items: score float(np.dot(query_vec, entry[vector])) scored.append((score, entry[item])) scored.sort(keylambda x: x[0], reverseTrue) return scored[:top_k]这里使用了归一化后的向量计算点积也就是余弦相似度。search返回一个(score, item)列表后续由MemoryManager再做加权排序。如果记忆量变大建议把items替换成 SQLite 表或专用向量数据库检索逻辑改成 SQL 加向量索引。3.4 实现 MemoryManager 的写入、检索、压缩MemoryManager负责协调各层级的写入、读取和自动压缩。核心逻辑如下class MemoryManager: def __init__(self, store: MemoryStore, short_term_limit: int 20): self.store store self.short_term_limit short_term_limit def write( self, content: str, layer: MemoryLayer, importance: float 0.5, source: str conversation, metadata: dict | None None, ): item MemoryItem( memory_iduuid.uuid4().hex, contentcontent, layerlayer, created_attime.time(), last_access_attime.time(), importanceimportance, sourcesource, metadatametadata or {}, ) self.store.add(item) if layer MemoryLayer.SHORT_TERM: self._maybe_compact_short_term() return item def read(self, query: str, top_k: int 5, layer_filter: MemoryLayer | None None): results self.store.search(query, top_ktop_k * 3) if layer_filter is not None: results [ (score, item) for score, item in results if item.layer layer_filter ] now time.time() def final_score(pair): score, item pair age_penalty 1.0 / (1.0 (now - item.last_access_at) / 86400.0) return score * 0.6 item.importance * 0.3 age_penalty * 0.1 results.sort(keyfinal_score, reverseTrue) return results[:top_k] def touch(self, memory_id: str): for entry in self.store.items: if entry[item].memory_id memory_id: entry[item].last_access_at time.time() entry[item].access_count 1 return def _maybe_compact_short_term(self): short_entries [ entry for entry in self.store.items if entry[item].layer MemoryLayer.SHORT_TERM ] if len(short_entries) self.short_term_limit: return short_entries.sort( keylambda entry: (entry[item].access_count, entry[item].last_access_at) ) victim_entries short_entries[: len(short_entries) - self.short_term_limit] victim_ids {entry[item].memory_id for entry in victim_entries} combined \n.join(entry[item].content for entry in victim_entries) summary self._simple_summary(combined) summary_item MemoryItem( memory_iduuid.uuid4().hex, contentsummary, layerMemoryLayer.LONG_TERM, created_atmin(entry[item].created_at for entry in victim_entries), last_access_attime.time(), importancemax(entry[item].importance for entry in victim_entries), sourcecompacted_summary, ) self.store.add(summary_item) self.store.remove_by_ids(victim_ids) def _simple_summary(self, text: str) - str: lines [line.strip() for line in text.splitlines() if line.strip()] seen [] for line in lines: if line not in seen: seen.append(line) if len(seen) 5: return .join(seen) return .join(seen[:5]) f共{len(lines)}条记录这里把短期记忆容量设置为short_term_limit。当短期记忆条数超过限制时系统按访问次数和时间排序挑出最不活跃的若干条合并成一条长期摘要。示例里的摘要函数只是去重拼接生产环境可以换成 LLM 摘要。3.5 模拟对话写入与召回验证用最小用例验证完整链路store MemoryStore() manager MemoryManager(store, short_term_limit3) manager.write(用户偏好使用简洁的中文报告, MemoryLayer.SHORT_TERM, importance0.7) manager.write(项目名称是 Prime Agent, MemoryLayer.SHORT_TERM, importance0.8) manager.write(用户偏好图形化展示结果, MemoryLayer.SHORT_TERM, importance0.6) # 写入第四条后短期记忆超过 limit会触发压缩 manager.write(数据库连接超时时间调整为30秒, MemoryLayer.SHORT_TERM, importance0.4) results manager.read(用户对报告格式有什么要求, top_k5) for score, item in results: print(f{score:.3f} {item.layer.value} {item.content})测试输出类似下面的结果实际分数会因为嵌入模型和运行时间不同而变化0.823 short_term 用户偏好使用简洁的中文报告 0.691 long_term 用户偏好使用简洁的中文报告项目名称是 Prime Agent用户偏好图形化展示结果共3条记录 0.614 short_term 用户偏好图形化展示结果验证记忆系统是否正常可以检查以下几点写入“用户偏好使用简洁的中文报告”后用“报告风格”查询能召回这条记录。写入“项目名称是 Prime Agent”后用“项目名称”查询能得到包含该信息的记忆。短期记忆超过限制后能生成一条source为compacted_summary的长期摘要。摘要内容也能通过语义检索被召回且不会被当成普通短期记忆。4. 关键参数与调优策略4.1 记忆层级参数速查表参数含义参考值调大影响调小影响short_term_limit短期记忆最多保留条数10-30短期上下文更多但检索噪声增加更早触发压缩可能丢失细节top_k每次检索召回条数3-8召回更多候选但可能引入无关内容上下文更精简但可能漏关键信息importance记忆重要度0-1高重要度记忆更不容易被淘汰低重要度记忆很快被压缩或删除age_penalty时间衰减在最终排序中的权重0.1旧记忆更难被召回旧记忆更容易抢占上下文compaction_interval压缩操作的最小时间间隔60 秒压缩频率低短期记忆峰值高压缩频繁额外开销变大这些参数不是固定标准。不同业务、不同模型、不同上下文长度下最优配置差异很大。建议先在一个小规模测试集上跑几组实验记录记忆命中率和任务完成质量再决定参数。4.2 top_k 和重要性权重的选择逻辑top_k是每次检索注入上下文的记忆条数。它不是越多越好。如果top_k太大Agent 会在推理时读到大量相似但无用的信息反而干扰决策。如果太小真正关键的历史事实可能漏掉。一个常见的开始值是 5。如果任务经常需要用户历史偏好可以适当提高到 8如果任务偏向单次执行业务可以降到 3。调参时要观察召回内容里有多少条被最终 Prompt 真正使用。这个比例就是“上下文利用率”。重要性权重的作用是解决“语义相似但不重要”的问题。比如用户问“之前提到过的权限问题”系统可能同时召回一段闲聊里的“权限”和一段正式方案里的“权限设计”记录。正式方案记录的重要性应该更高所以最终排序会把它排到前面。4.3 过期时间与清理周期记忆不能只靠压缩还需要过期清理。设计上可以区分两个维度短期记忆主要是“最近 N 条”和“最近 N 小时”两种策略。长期记忆按“重要度”和“最后访问时间”定期清理。示例清理任务可以是每 10 分钟删除超过 7 天未访问且importance 0.3的长期记忆每 60 分钟把超过 24 小时的短期记忆合并成摘要。具体阈值取决于业务容忍度。注意不要在大模型推理的关键链路上同步执行清理任务建议用独立定时任务或后台队列处理。4.4 调优前先看数据分布不要盲调参数调优最怕的是脱离数据。建议先统计以下指标每轮对话平均产生多少条记忆。短期记忆达到上限的时间间隔。检索结果的 Top 1 是否经常就是任务需要的记忆。压缩后的摘要是否仍能被后续查询召回。这些指标决定了记忆层级机制是否健康。如果短期记忆总是不到 10 条就冷却就不需要把short_term_limit设成 50如果长期记忆里 70% 都是三个月前且从未被访问的记录应该优先增加清理策略而不是扩大容量。5. 常见问题与排查路径5.1 检索结果不相关现象查询“用户喜欢什么颜色”返回的是项目时间表。可能原因查询改写不够明确向量搜索只匹配了表面词。缺少元数据过滤不同用户、不同会话的记忆混在一起。top_k设置过大召回了太多相似度很低的候选。嵌入模型选择不合适对业务术语表达不好。检查方式打印候选记忆的相似度分数看 Top 1 和 Top 5 的分数差距。检查查询改写后的文本是否准确表达了检索意图。检查记忆条目中的metadata是否包含当前用户和会话 ID。处理建议在检索前增加用户、会话、时间范围过滤。提高importance在最终分数中的权重。适当降低top_k并加入相似度最低阈值例如低于 0.35 的结果直接丢弃。5.2 写入长期记忆后短期记忆仍然膨胀现象长期记忆库增长正常但短期记忆条数一直超过限制。可能原因写入时没有正确分类所有内容都写入SHORT_TERM。压缩函数没有被调用或者short_term_limit判断条件写错。压缩完成后又把摘要写回了短期记忆造成循环膨胀。检查方式打印每条记忆的layer字段确认写入层级。在_maybe_compact_short_term入口加日志确认每次写入后是否触发。检查压缩后的摘要写入层是LONG_TERM还是SHORT_TERM。处理建议在write方法之前加信息分类层根据信息类型自动决定目标层级。压缩摘要统一写入LONG_TERM。设置压缩冷却时间避免频繁触发后短暂膨胀又压缩。5.3 摘要压缩丢失关键信息现象短期记忆被合并成摘要后Agent 后续回答问题缺少关键细节。可能原因示例里的_simple_summary只是拼接去重没有理解语义。被压缩的短期记忆里包含高重要度信息但没有被单独保留。摘要长度限制太短无法容纳多条事实。处理建议生产环境用 LLM 生成摘要并要求按“事实条目”输出而不是自然语言段落。压缩前判断重要度importance高于某个阈值的记忆不参与压缩继续留在短期记忆。摘要里保留结构化字段例如“用户偏好”“用户约束”“任务事实”分组。5.4 并发写入顺序错乱或数据竞争现象多个请求同时写入记忆时短期记忆压缩出现吞掉新写入内容的情况。可能原因MemoryStore.items是普通 Python 列表多线程同时修改时没有加锁。压缩逻辑读取和删除之间没有原子性保证。多实例部署时每台机器的内存不共享同一用户记忆被分散到不同实例。检查方式在写入和压缩方法里打印线程名和时间戳观察是否出现并发交叉。检查部署方式确认是否多实例无状态运行。处理建议单进程内使用threading.Lock保护写入、删除、压缩操作。多实例场景下把记忆存储切换到 Redis、SQLite 或 PostgreSQL避免内存不一致。每次更新记忆时带上版本号或更新时间冲突时以新区间记录为准。5.5 embedding 模型切换导致历史向量无法检索现象切换嵌入模型后查询“用户名”检索不到之前写入的“用户姓名”相关记忆。可能原因新旧模型生成的向量维度不同或者虽然维度相同但语义空间不兼容。历史记忆向量没有重新生成仍然使用旧模型。向量数据库索引未重建。处理建议模型切换是迁移任务必须先离线重建全部记忆的向量。双跑期间同时保留旧向量和新向量避免服务中断。代码里固定模型名称和版本禁止线上随机切换。6. 生产环境落地建议与扩展方向6.1 从最小示例到生产系统需要补齐的模块内存版实现的意义是讲解核心逻辑离生产系统还有一段距离。一个可用的生产级记忆层级系统至少需要补齐以下能力持久化把记忆存到 SQLite、PostgreSQL、Redis 或专用向量数据库而不是进程内存。服务化提供独立的记忆读写接口让 Agent 主流程通过 API 调用。一致性写入、更新、删除操作需要事务和锁保护。可观测记录记忆写入量、检索次数、压缩频率、命中率。权限隔离按用户 ID 或租户 ID 隔离记忆防止跨用户泄露。定时清理用后台任务定期删除过期记忆、重建摘要、更新热度和重要度。安全脱敏用户输入中可能包含敏感信息写入长期记忆前要经过脱敏处理。6.2 记忆层级和 Agent 编排如何配合记忆机制不会单独工作它应该被接入到 Agent 的完整执行链路中。规划阶段读取长期记忆获取业务规则和历史项目背景执行阶段写入工作记忆保存每个步骤的中间结果会话切换时把有价值的中间结论写回短期记忆任务结束后由后台任务把短期记忆压缩到长期记忆。这种配合方式可以让 Agent 做到“当前任务结束后不会完全遗忘上一次对话的经验”同时避免大量历史信息一直占据上下文窗口。6.3 评估记忆质量三个可落地指标记忆系统上线后需要持续评估。建议关注以下三个指标指标含义计算思路目标召回准确率检索到的记忆是否确实有用人工或规则标注 Top 5 结果中有用占比越高越好上下文利用率实际被 Agent 使用的记忆条数与召回条数之比在 Prompt 构造日志中记录哪些记忆被最终引用越高越好遗忘合理性被压缩或删除的记忆是否真的不重要定期抽样查看清理日志判断是否有高价值信息丢失误删率越低越好这三个指标可以帮助团队判断记忆层级机制是否真正缓解了上下文问题而不是仅仅多了一个存储组件。回到 Prime Agent 的记忆层级机制核心价值不是“存得多”而是“在正确的时间把正确的记忆放到正确的层级里”。理解这个概念之后可以先在小项目里实现一个简单的三层记忆管理器不断观察检索命中率和上下文的信噪比。对于新手而言最有价值的练习不是抄代码而是给系统加入一套日志统计每一轮对话中哪些记忆被召回、哪些被压缩、哪些被遗漏再用这些数据反向调整层级划分和参数。