长程智能体long-horizon agent是当前 AI 应用里最难做扎实的部分之一。复杂任务往往需要模型在多个步骤之间保留大量中间状态例如记住用户偏好、前一步生成的 ID、已经处理过的文件列表。Recuris 可以理解为 recursive 的变体核心思路是用双记忆机制解决长程任务中常见的“做着做着就忘了”的问题工作记忆负责保存当前任务上下文长期记忆负责保存重要事实和可复用决策二者按统一调度协作。这篇文章围绕 Recuris 的设计思路拆解双记忆机制的核心原理并用一个可以本地运行的 Python 示例验证它对长程任务成功率的影响最后给出生产环境落地的排查思路和工程化建议。整个过程不需要大模型 API适合作为理解记忆机制的入门实验。1. 先弄清长程智能体为什么会在任务中途“失忆”1.1 长程任务的技术定义和典型特征长程任务不是“步骤多”这么简单。它通常包含三个可量化特征任务状态依赖链条长后续决策要依赖前面若干步的输出。环境中没有统一的全局状态智能体必须自己维护上下文。任务周期超过单次推理的上限。典型场景包括多轮对话客服、自动撰写研究报告、批量处理文件并生成汇总、在复杂 UI 上完成跨页面操作。一个任务之所以难往往不是“某一步不会做”而是“第一步的输出在第十步被忘掉了”。解决方案可以从提示词里塞上下文、扩大上下文窗口也可以设计记忆分层。Recuris 走的是后者。1.2 记忆缺失导致任务失败的具体表现从实际运行日志里看长程任务失败通常呈现下面几种模式状态覆盖新信息把旧信息挤出了上下文后续步骤读不到早期结果于是重复执行或乱猜。检索模糊虽然信息还在上下文里但模型没有在需要的时候“想起来”表现为答非所问。上下文污染为了保留更多历史把大量无关片段塞进提示词结果模型被噪声干扰主任务反而做错。一致性问题同一实体在不同步骤中被写成不同名称后续步骤无法对齐。这些失败模式看起来是“模型能力不够”但很多情况下是记忆管理不到位。换句话说模型不是不知道怎么做而是没有在正确的时间拿到正确的信息。1.3 从上下文窗口看记忆瓶颈现在的主流大模型有固定上下文窗口。窗口不是不能用它的瓶颈体现在三方面成本、噪声和衰减。窗口越长单次请求的 token 成本越高推理时延也会上升。更重要的是把全部历史堆进窗口模型需要从很长的序列中定位关键信息这比从“最近几步加选中的长期记忆”中查找要难得多。所以单纯调大窗口并不能等价于拥有更好的记忆。1.4 常见单记忆方案的局限只靠对话历史列表本质上是一条不断变长的线性记录。它的问题是没有“遗忘”和“重点”的概念。所有信息都被同等看待那么关键信息就容易被淹没。Recuris 的双记忆机制把记忆拆成两层用工作记忆应对即时上下文用长期记忆应对跨步骤依赖目的就是让系统在需要时主动决定“这一条该记在哪里”。2. 双记忆机制的设计工作记忆与长期记忆如何协同2.1 工作记忆容量有限服务当前任务工作记忆相当于智能体在当前任务里“手边”的信息。它的容量通常很小比如 3 到 10 个条目。优点是读取快、没有延迟缺点是容量有限需要淘汰策略。Recuris 里工作记忆的淘汰策略综合两条规则最近使用优先不常用的旧条目先淘汰。重要性优先重要条目不会被普通条目覆盖而是先写入长期记忆。这正是双记忆机制和简单“先进先出”队列的区别不是无脑覆盖而是在覆盖前判断是否值得长期保存。2.2 长期记忆容量大按需召回长期记忆存的是“值得保留的中间结论”。它不参与每一步的即时推理只在智能体需要时按查询召回。Recuris 的长期记忆在实现上是一个可检索的存储可以是向量数据库也可以像本文示例一样使用 JSON 文件加关键词检索。召回时返回与当前查询最相关的 top-k 条避免把整个记忆库塞进上下文。2.3 双记忆机制的四个核心环节写入每产生一条新记忆先进入工作记忆。归档当工作记忆快满或者条目重要性超过阈值把条目写入长期记忆。召回当前步骤需要某个信息时先查工作记忆如果没有再用查询词检索长期记忆。更新召回命中的条目会被“重新置顶”同时记录访问次数方便后续淘汰判断。这四个环节共同构成一个记忆闭环。只设计存储结构不算双记忆机制必须把写入、归档、召回、更新串联起来才能避免“存了但用不上”的问题。2.4 为什么“双记忆”能提升成功率长期记忆让关键事实可以跨过上下文窗口的容量限制。工作记忆的容量限制被长期记忆补上工作记忆的速度优势又避免了每次都要从长期存储检索。组合起来的直接效果是即时上下文更干净跨步骤依赖更稳。这里有一个容易被忽略的细节重要性评估很关键。如果所有信息都写入长期记忆长期记忆会变成第二个“垃圾堆”。Recuris 的策略是只对高重要性数据做归档低重要性数据随工作记忆淘汰。这样召回时得到的结果才足够精确。3. 环境准备和最小实验设计3.1 运行环境示例使用 Python 3.8 及以上版本不需要第三方库。建议在虚拟环境中运行项目文件结构如下recuris-demo/ ├── memory.py └── agent.py其中memory.py负责记忆数据结构和工作记忆、长期记忆、双记忆调度核心agent.py负责模拟智能体、生成实验任务和统计成功率。3.2 实验任务定义为了让结果可复现不使用真实大模型而是定义一个“事实依赖型”任务。每个任务是一串原子步骤步骤之间有一个关键记忆跨度第start步产生一条事实例如user_uid u_10086。第end步需要这条事实。中间步骤会不断产生干扰信息占满工作记忆。只有短期记忆的智能体当跨度超过工作记忆容量时会遗忘关键事实使用 Recuris 双记忆机制的智能体会在事实产生时按重要性归档到长期记忆在需要时召回。3.3 成功率如何定义一次任务成功必须同时满足两步关键事实没有被破坏或错误覆盖。消费步骤能够从记忆系统中检索到正确值。统计时对同组任务执行多次得到成功率成功次数 / 总任务数 * 100%。3.4 对照设计基线只有工作记忆容量 3。Recuris工作记忆容量 3 加长期记忆重要性阈值 0.6。对每个跨度 2、4、6、8、10 各生成 5 个随机任务最终对比这两个组的成功率。这个设计为的是直接观察“跨度增大后双记忆是否仍然能保持高成功率”。4. 实现一个可运行的 Recuris 双记忆模块4.1 先定义记忆条目结构记忆不能直接存成纯字符串。Recuris 把记忆设计成结构化条目包含内容、检索键、重要度和访问信息。实现如下# memory.py import json import os import time from dataclasses import dataclass, field dataclass class MemoryItem: content: str key: str importance: float 0.5 created_at: float field(default_factorytime.time) last_access_at: float field(default_factorytime.time) access_count: int 0字段含义content记忆内容例如user_uid u_10086。key用于检索的键例如user_uid。importance重要度取值 0 到 1。created_at、last_access_at、access_count用于淘汰和更新策略。把记忆拆成结构化字段后续索引、检索、序列化都会方便很多。4.2 实现工作记忆工作记忆容量有限核心是淘汰策略。这里使用“访问次数少且重要度低”的条目先淘汰class WorkingMemory: def __init__(self, capacity: int 3): self.capacity capacity self.items [] def add(self, item: MemoryItem): self.items.append(item) if len(self.items) self.capacity: self._evict() def get(self, key: str): for item in self.items: if item.key key: item.access_count 1 item.last_access_at time.time() return item return None def _evict(self): self.items.sort(keylambda x: (x.access_count, x.importance)) self.items.pop(0)这里为了展示核心逻辑省略了淘汰时把重要条目归档的联动。真正的 Recuris 调度会在上层完成归档避免重要信息直接丢失。4.3 实现长期记忆长期记忆在示例中使用 JSON 文件存储检索方式使用关键词包含匹配。生产环境可以替换成向量检索但核心召回逻辑不变class LongTermMemory: def __init__(self, storage_path: str long_term_memory.json): self.storage_path storage_path self.items [] self._load() def store(self, item: MemoryItem): self.items.append(item) self._save() def search(self, key: str, top_k: int 2): result [] for item in self.items: if key in item.key or key in item.content: result.append(item) return result[:top_k] def _save(self): data [{ content: i.content, key: i.key, importance: i.importance, created_at: i.created_at, last_access_at: i.last_access_at, access_count: i.access_count } for i in self.items] with open(self.storage_path, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2) def _load(self): if os.path.exists(self.storage_path): with open(self.storage_path, r, encodingutf-8) as f: data json.load(f) for d in data: self.items.append(MemoryItem(**d))关键词匹配足够说明“按需召回”的价值。真实项目中如果数据量大、相似语义多建议换成 embedding 加相似度检索否则召回精度很容易变成瓶颈。4.4 实现双记忆调度核心RecurisMemory把工作记忆和长期记忆串起来暴露给智能体两个方法remember和recallclass RecurisMemory: def __init__(self, working_capacity: int 3, importance_threshold: float 0.6): self.working WorkingMemory(capacityworking_capacity) self.long_term LongTermMemory() self.importance_threshold importance_threshold def remember(self, key: str, content: str, importance: float 0.5): item MemoryItem(keykey, contentcontent, importanceimportance) self.working.add(item) if importance self.importance_threshold: self.long_term.store(item) def recall(self, key: str, top_k: int 2): item self.working.get(key) if item: return item matches self.long_term.search(key, top_ktop_k) if matches: best matches[0] self.working.add(best) return best return None这个实现的关键点在于高重要性条目在写入时就直接归档到长期记忆。这样即使工作记忆随后被噪声占满关键事实也已经在长期记忆里存在。为了做基线对照还需要一个不包含长期记忆的版本class BaselineMemory: def __init__(self, working_capacity: int 3): self.working WorkingMemory(capacityworking_capacity) def remember(self, key: str, content: str, importance: float 0.5): item MemoryItem(keykey, contentcontent, importanceimportance) self.working.add(item) def recall(self, key: str, top_k: int 2): return self.working.get(key)4.5 模拟智能体消费记忆模拟智能体按照任务步骤顺序执行。在产生关键事实时调用remember在需要关键事实时调用recall中间步骤写入干扰信息# agent.py import random from memory import ( MemoryItem, WorkingMemory, LongTermMemory, RecurisMemory, BaselineMemory, ) class StepAgent: def __init__(self, memory): self.memory memory def run_task(self, task): for step in range(1, task[end] 1): if step task[start]: self.memory.remember( keytask[fact_key], contenttask[value], importance0.9 ) elif step task[end]: item self.memory.recall(task[fact_key]) return item is not None and item.content task[value] else: noise_key fnoise_{step} self.memory.remember( keynoise_key, contentftmp_{step}, importance0.1 ) return False注意中间干扰步骤的重要性只有 0.1低于归档阈值 0.6因此不会写入长期记忆。这正是双记忆机制保护关键事实的方式。5. 运行实验并验证双记忆机制的效果5.1 生成任务和运行脚本任务生成需要覆盖不同记忆跨度。这里固定随机种子让结果可以重复def build_tasks(): tasks [] random.seed(42) spans [2, 4, 6, 8, 10] for span in spans: for _ in range(5): start 1 end start span random.randint(0, 3) tasks.append({ start: start, end: end, fact_key: critical_fact, value: fv_{start}_{end}, }) return tasks def evaluate(memory_factory, tasks): success 0 for task in tasks: memory memory_factory() agent StepAgent(memory) if agent.run_task(task): success 1 return success, len(tasks), success / len(tasks) * 100 if __name__ __main__: tasks build_tasks() baseline_success, total, baseline_rate evaluate( lambda: BaselineMemory(working_capacity3), tasks ) recuris_success, _, recuris_rate evaluate( lambda: RecurisMemory(working_capacity3, importance_threshold0.6), tasks ) print(fbaseline: {baseline_success}/{total} {baseline_rate:.2f}%) print(frecuris: {recuris_success}/{total} {recuris_rate:.2f}%)运行命令python agent.py预期输出格式baseline: 2/25 8.00% recuris: 25/25 100.00%具体比例取决于随机任务分布。从趋势上看跨度越大基线成功率越低而 Recuris 双记忆机制几乎不受跨度影响。5.2 结果解读成功率的差距主要来自跨度大于工作记忆容量的任务。基线智能体在中间塞入太多噪声后关键事实被淘汰Recuris 在写入时把高重要性条目归档到长期记忆消费时能按 key 召回所以跨度不再是瓶颈。这个结果说明提升长程任务成功率不一定要无限扩大上下文关键是让重要记忆突破“最近几步”的限制。5.3 从日志看记忆变化可以在WorkingMemory.add和LongTermMemory.store中加入打印观察关键事实被写入长期记忆的时间点以及消费步骤从长期记忆命中时工作记忆里有哪些条目。这种日志就是排错时最需要的证据。例如在LongTermMemory.store中打印print(f[store] key{item.key}, importance{item.importance})在RecurisMemory.recall命中长期记忆时打印print(f[recall] from long_term key{key})通过日志可以确认关键事实没有被丢弃只是在需要时从另一个存储层取回来了。5.4 参数调整对结果的影响工作记忆容量从 3 调到 8基线成功率会上升因为可以容纳更长跨度Recuris 成功率保持高位。重要度阈值调到 0.95关键事实 importance 为 0.9 时不会被归档双记忆会退化成接近基线说明阈值选择很关键。检索 top_k 过小如果长期记忆里存在多个相似条目可能取不到正确那条需要结合更精确的 key 设计。注意不要只验证程序能启动还要验证输入、输出、异常分支和日志是否符合预期。记忆机制的效果必须通过“是否在正确步骤拿到正确信息”来判断。6. 生产环境落地时的核心问题排查6.1 检索不到关键记忆现象任务执行到中后段recall返回 None。检查顺序先看长期记忆数据文件里是否真的写入了该 key。再看查询 key 与存储 key 是否一致。最后看检索函数是否只支持完全包含匹配。解决统一 key 生成规则例如统一用entity:attribute格式检索逻辑改成标准化的关键词抽取必要时建立索引或向量检索。6.2 检索结果太杂导致上下文污染现象召回了很多相似条目智能体被无关信息干扰。检查查看 top-k 设置查看相似度阈值查看长期记忆中是否写入过多低价值条目。解决提高写入长期记忆的重要度阈值召回后增加相关性过滤每次召回只注入 top-k 条并且对重复内容做去重。6.3 长期记忆写入慢、文件越来越大现象任务量上来后JSON 读写越来越慢。检查看存储文件大小和写入频率。解决学习环境用 JSON 没问题生产环境应换用向量数据库或专用记忆服务对长期记忆做定期压缩和失效清理避免无限增长。6.4 记忆一致性和数据格式问题现象同一条事实在不同步骤中 key 不一致导致重复存储和召回失败。检查对照任务日志中的 key 书写方式。解决定义统一的事件 schema例如{ type: entity_update, entity: user, attribute: uid, value: u_10086 }在写入和查询时都使用同一个序列化函数避免手写 key 造成的不一致。常见问题排查表现象可能原因检查方式解决建议召回不到关键事实key 不一致或未写入长期记忆查看长期记忆文件和查询 key统一 key 规则增加命中日志长期记忆膨胀低价值条目大量写入统计 importance 分布提高归档阈值增加压缩任务工作记忆淘汰重要条目重要性阈值设置不合理检查淘汰日志重要度高于阈值时先归档再淘汰7. 工程化建议与扩展方向7.1 学习环境与生产环境的差异维度学习环境生产环境长期记忆存储JSON 文件向量数据库或专业记忆服务检索方式关键词包含匹配embedding 加相似度检索记忆写入直接同步写入异步写入、幂等去重淘汰策略简单排序分层策略加监控指标日志集中打印结构化和可检索回滚删除文件即可需要版本控制和快照7.2 关键参数速查表参数建议值调大影响调小影响working_capacity3 到 10即时上下文更大但噪声可能增加淘汰变频繁长跨度任务易失败importance_threshold0.6 到 0.8长期记忆更少检索更干净但可能漏存长期记忆膨胀召回噪声增加top_k2 到 5召回信息更多上下文变长可能漏掉关键结果7.3 可复用的实现清单记忆条目必须有结构化字段而不是纯字符串。写入时先评估重要性高于阈值先归档。召回先查工作记忆再查长期记忆命中后提升访问热度。淘汰前检查是否已归档避免重要信息直接丢失。所有读写操作记录日志确保任务失败时可以复盘记忆链路。7.4 下一步扩展方向Recuris 的双记忆机制是记忆管理的一种基础形态实际项目中可以继续扩展短期记忆可以加入摘要压缩长期记忆可以加入时间衰减和定期遗忘跨会话场景可以增加用户级记忆隔离与真实大模型集成时可以把recall的结果渲染成提示词片段让模型在推理时只看到相关记忆。如果把示例代码接入真实模型最重要的不是换一个更大的上下文窗口而是先建立可观测的记忆链路。只要能在日志中回答“这条记忆是什么时候写入的、为什么被保留、在哪个步骤被召回”长程任务的成功率就可以被系统性地优化。