长程智能体做多了之后会发现一个很典型的失败模式任务在 5 步以内能完成得很好一旦超过 10 步模型就开始“失忆”要么忘记最初的目标要么把上一步的错误结果当成正确前提继续向下做最后整条任务链崩塌。网上方案很多但大多只解决“单步决策”并没有解决“如何让智能体记住自己做到哪了、下次怎么更少犯错”。Recuris 双记忆机制就是针对这个问题设计的一套解决方案本文会从原理、设计、代码到评测完整拆解这套机制的落地思路。Recuris 可以理解为一套面向长程任务设计的智能体系统方案其核心创新是双记忆机制把智能体的记忆拆成工作记忆和情景记忆两个通道。工作记忆负责管理当前任务的状态防止目标漂移情景记忆负责沉淀历史任务的经验避免重复试错。两类记忆协同工作能明显提升长程智能体的成功率。如果你正在开发 Agent、做网页自动操作、多轮工具调用或者研究长程任务规划这篇文章会比较有用。下面从长程智能体为什么难讲起逐步拆解双记忆机制。1. 长程智能体为什么这么难做1.1 什么是长程智能体长程智能体英文常写为 Long-horizon Agent指的是需要执行多步骤、长时间跨度任务的智能体系统。典型场景包括网页自动操作打开页面、填写表单、点击按钮、翻页、提交数据。数据分析流水线读取文件、清洗数据、建模、输出报告。智能客服解决复杂问题多轮询问、查询订单、协调多个系统。机器人任务序列导航、抓取、放置、再回到起点。这类任务的共同点是单步动作不难难的是把几十步动作串联起来并且保证每一步都依赖正确的历史状态。1.2 长程任务的三个核心挑战长程智能体成功率不高根源集中在三个问题上。第一个问题是上下文爆炸。每一步决策都要带上历史信息但大模型上下文窗口有限。当轨迹越来越长要么强行截断导致历史丢失要么全部塞进去导致输入过长、成本过高。第二个问题是目标漂移。模型在中间步骤看到大量中间状态后容易被当前局部信息带偏忘记最初的用户意图。最常见的表现是任务做到一半模型开始“自由发挥”做了一些不在目标范围内的事情。第三个问题是错误累积。长程任务中一步出错往往会污染后续所有步骤。前一步状态错了下一步基于错误状态推理后面的步骤全部失去意义而且错误很难定位。这三个问题互相叠加导致任务越长成功率下降越明显。1.3 Recuris 的解决思路Recuris 没有把精力放在“让单步推理更强”上而是把问题拆成记忆问题当前任务状态记不住那就用工作记忆管理历史经验用不起来那就用情景记忆沉淀。工作记忆解决目标漂移和上下文爆炸情景记忆解决错误累积和重复试错。两条通道组合起来正好覆盖长程任务最薄弱的环节。2. 双记忆机制总体设计2.1 工作记忆与情景记忆的分工工作记忆类似人类脑子里的“草稿纸”。它保存的信息包括当前任务目标、已经完成的子任务、最近几步的执行轨迹、以及被压缩后的历史摘要。特点是非常轻量、更新频繁、任务结束即清理。情景记忆类似人类的“经验库”。它保存的是历史任务的完整经验任务类型、任务目标、执行轨迹、最终状态、失败原因。特点是数据量更大、需要检索、长期保留。两者的差异可以用表格概括维度工作记忆情景记忆核心职责管理当前任务状态沉淀历史任务经验更新频率每一步都更新任务结束或关键节点更新容量小固定窗口大持续增长读取方式直接读取无需检索相似度召回生命周期任务结束即清理长期保留核心目标防止目标漂移避免重复试错2.2 为什么需要两条记忆通道一个很自然的疑问是为什么不直接把所有历史信息放进一个大记忆池而要做两个通道原因在于两类记忆的读写模式完全不同。工作记忆是高频写、高频读每一步决策都要看情景记忆是低频写、按需读只有遇到相似任务时才用到。如果把两者混在一起工作记忆会把历史任务的噪声带进当前决策情景记忆也会因为频繁被无关上下文干扰而失去参考价值。分开之后工作记忆只负责“当前我在哪”情景记忆只负责“过去类似任务怎么做”职责清晰提示词也更容易控制。2.3 双记忆机制的运行流程整个流程可以用六个阶段概括初始化把任务目标写入工作记忆并在情景记忆中检索相似历史经验。决策将工作记忆里的目标、摘要、最近轨迹和情景记忆返回的相关经验一起组装成提示词交给大模型。执行大模型输出动作Agent 调用工具或环境接口执行该动作。观察拿到动作执行结果写入工作记忆。更新如果轨迹过长触发工作记忆压缩把旧步骤生成摘要释放空间。沉淀任务成功或失败后将整条轨迹写入情景记忆供后续任务参考。3. 双记忆机制核心原理拆解3.1 工作记忆如何防止目标漂移工作记忆的内容不是随意堆叠而是结构化维护。首先是任务目标这一项在初始化时写入之后不允许覆盖。这是防止目标漂移最关键的设计。很多 Agent 失败就是因为原始目标被中间结果不断覆盖最终变成了另一个任务。Recuris 的做法是把目标放在提示词固定位置并且每次决策都带上。其次是最近轨迹一般保留最近 5 到 10 步。轨迹太短模型缺少局部上下文轨迹太长又容易发生注意力分散。设定一个固定窗口超出窗口的旧轨迹进入压缩流程。压缩策略也很重要不是简单丢弃旧步骤而是由大模型把旧轨迹整理成结构化摘要。例如“已完成打开注册页 - 填写邮箱当前阶段填写资料”。这样模型在决策时既能获取全局进度又不需要阅读全部原始轨迹。工作记忆的作用本质上是对输入做信息过滤目标字段负责对齐方向摘要字段负责全局进展最近轨迹负责局部状态。三者各司其职模型每一步看到的信息都是精炼且可控的。3.2 情景记忆如何实现经验复用情景记忆要解决的是“类似任务我上次怎么做的”。它存储的每一条经验通常会包含任务类型、任务目标、执行轨迹、结果状态和失败原因。检索方式上Recuris 的典型实现是把任务类型和目标文本转成向量通过余弦相似度召回相近经验。生产环境中可以用向量数据库做索引也可以先用标签过滤再计算相似度减少无关经验进入提示词。这里有两个关键参数需要关注。一个是 top_k也就是一次最多召回几条经验。取值过大会把低相关经验带进来过小则可能漏掉有效经验一般取 2 到 5 条比较合适。另一个是相似度阈值 threshold。低于阈值的经验应该直接丢弃而不是硬塞给模型。没有阈值过滤的召回大概率会引入干扰信息反而降低决策质量。情景记忆还需要做经验分级。成功经验和失败经验不能混为一谈。成功经验可以作为正面示范失败经验则应该在提示词中明确标注“此路径不可行”避免模型把错误路线当成正确路线复用。3.3 双记忆如何协同决策在 Recuris 中两类记忆最终会拼进同一条提示词但位置和格式是固定的。工作记忆内容放在前面包括任务目标、历史摘要、最近轨迹情景记忆内容放在后面以“相关经验”的形式出现。这样模型先看当前状态再看历史经验最后输出动作。协同的关键在于避免记忆污染。情景记忆检索频率不宜过高否则每步都会把大量历史经验塞进上下文干扰当前判断。比较合理的做法是任务开始时检索一次任务进入新阶段或发生失败时再检索一次。通过这种“低频检索、高频更新”的策略工作记忆保证短期方向稳定情景记忆提供长期经验参考两者互不干扰。4. 核心代码实现Recuris 双记忆机制简化原型下面用一个可运行的 Python 简化原型演示双记忆机制的数据流。演示环境为 Python 3.9代码只依赖标准库重点说明记忆结构、更新逻辑和决策循环。4.1 项目结构recuris-demo/ ├── memory.py └── agent.pymemory.py 负责定义工作记忆和情景记忆agent.py 负责组装智能体主循环。4.2 编写 memory.py先看文件路径recuris-demo/memory.py。# 文件路径recuris-demo/memory.py from collections import deque from typing import List, Dict, Optional class WorkingMemory: 工作记忆负责保存当前任务的目标、最近轨迹和压缩摘要。 def __init__(self, window_size: int 6): self.window_size window_size self.goal: str self.subtasks: List[str] [] self.recent_steps: deque deque(maxlenwindow_size) self.summary: str def init_task(self, goal: str, subtasks: Optional[List[str]] None) - None: self.goal goal self.subtasks subtasks or [] self.recent_steps.clear() self.summary def add_step(self, step_description: str, result: str) - None: self.recent_steps.append({ step: step_description, result: result }) # 窗口写满后触发压缩 if len(self.recent_steps) self.window_size: self._compress() def _compress(self) - None: old_steps list(self.recent_steps) compressed_parts [ f{s[step]} - {s[result]} for s in old_steps ] new_summary .join(compressed_parts) self.summary f已完成轨迹摘要{new_summary} self.recent_steps.clear() def to_prompt_section(self) - str: recent .join( f{s[step]} - {s[result]} for s in self.recent_steps ) or 暂无 return ( f任务目标{self.goal}\n f历史摘要{self.summary or 暂无}\n f最近轨迹{recent} ) class EpisodicMemory: 情景记忆负责沉淀历史任务经验并按照相似度召回。 def __init__(self): self.episodes: List[Dict] [] def add_episode(self, episode: Dict) - None: self.episodes.append(episode) def retrieve( self, query: str, top_k: int 2, threshold: float 0.25 ) - List[Dict]: scored [] for ep in self.episodes: score self._similarity(query, ep.get(task_type, )) if score threshold: scored.append((score, ep)) scored.sort(keylambda x: x[0], reverseTrue) return [ep for _, ep in scored[:top_k]] staticmethod def _similarity(query: str, task_type: str) - float: query_words set(query.lower().split()) type_words set(task_type.lower().split()) if not query_words or not type_words: return 0.0 union query_words | type_words inter query_words type_words return round(len(inter) / len(union), 4)这段代码有几个地方需要注意。WorkingMemory 中deque(maxlenwindow_size)会在插入新元素时自动丢弃最旧的元素天然实现了滑动窗口。当窗口写满时触发_compress()把旧步骤合并成摘要这是控制上下文长度的关键。EpisodicMemory 中的_similarity使用了 Jaccard 相似度即两个文本集合的交集除以并集。这个方法只能用于演示实际项目中遇到中文长文本建议替换为向量检索通过 embedding 模型把文本转成向量再做余弦相似度计算。4.3 编写 agent.py再看文件路径recuris-demo/agent.py。# 文件路径recuris-demo/agent.py from memory import WorkingMemory, EpisodicMemory from typing import Callable, Optional, List def default_llm(prompt: str) - str: 真实项目中替换为大模型调用。 示例 resp openai.chat.completions.create( modelgpt-4o, messages[{role: user, content: prompt}] ) return resp.choices[0].message.content.strip() 请以你实际使用的 SDK 文档为准。 raise NotImplementedError(请替换为实际大模型调用) class RecurisAgent: def __init__( self, working_memory: WorkingMemory, episodic_memory: EpisodicMemory, llm_fn: Callable[[str], str] default_llm, max_steps: int 15, ): self.wm working_memory self.em episodic_memory self.llm_fn llm_fn self.max_steps max_steps def run( self, task_goal: str, task_type: str, action_fn: Callable[[str], str], subtasks: Optional[List[str]] None, ) - bool: # 1. 初始化工作记忆 self.wm.init_task(task_goal, subtasks) # 2. 从情景记忆召回相似经验 related_experiences self.em.retrieve( f{task_goal} {task_type}, top_k2 ) for step_index in range(1, self.max_steps 1): # 3. 组装提示词 prompt self._build_prompt(related_experiences) # 4. 大模型输出动作 action self.llm_fn(prompt) print(f[第 {step_index} 步] 模型动作{action}) if action.strip().upper() FINISH: self._save_experience(task_goal, task_type, success) print(任务完成已写入情景记忆。) return True # 5. 执行动作并观察结果 result action_fn(action) print(f[执行结果] {result}) # 6. 更新工作记忆 self.wm.add_step(action, result) # 失败时快速终止并记录失败经验 if error in result.lower() or 失败 in result: self._save_experience(task_goal, task_type, failed) print(任务失败失败经验已写入情景记忆。) return False print(达到最大步数任务未完成。) return False def _build_prompt(self, related_experiences) - str: wm_section self.wm.to_prompt_section() exp_section 相关经验 if related_experiences: exp_sections [] for exp in related_experiences: status exp.get(status, unknown) goal exp.get(goal, ) trajectory exp.get(trajectory, ) exp_sections.append( f[{status}] {goal}{trajectory} ) exp_section | .join(exp_sections) else: exp_section 无 return ( f{wm_section}\n f{exp_section}\n\n f请根据当前状态和参考经验输出下一步动作。 f如果认为任务已经完成请只输出 FINISH。 ) def _save_experience(self, task_goal, task_type, status) - None: self.em.add_episode({ task_type: task_type, goal: task_goal, trajectory: list(self.wm.recent_steps), summary: self.wm.summary, status: status, }) if __name__ __main__: # 模拟环境只用来演示记忆数据流 class DemoActionExecutor: def __init__(self): self.executed_times 0 def execute(self, action: str) - str: self.executed_times 1 if action 提交表单: return success提交成功页面跳转到完成页 return ok页面响应正常可以继续下一步 class StepMockLLM: def __init__(self): self.actions [填写邮箱, 设置密码, 提交表单] def __call__(self, prompt: str) - str: if not self.actions: return FINISH return self.actions.pop(0) wm WorkingMemory(window_size4) em EpisodicMemory() # 预先放一条历史经验 em.add_episode({ task_type: 网页表单填写, goal: 注册账号, trajectory: 打开注册页 - 填写邮箱 - 设置密码 - 提交表单, status: success, }) agent RecurisAgent( working_memorywm, episodic_memoryem, llm_fnStepMockLLM(), max_steps10, ) executor DemoActionExecutor() success agent.run( task_goal注册账号, task_type网页表单填写, action_fnexecutor.execute, subtasks[填写邮箱, 设置密码, 提交表单], ) print(最终结果:, success) print(情景记忆库条数:, len(em.episodes))在run方法中整个流程正好对应前面说的六个阶段。初始化时写入目标并检索情景记忆循环中组装 prompt、调用模型、执行动作、更新工作记忆结束后把结果写入情景记忆。_build_prompt中把工作记忆放在提示词上半部分情景记忆放在下半部分用“相关经验”标注让模型明确区分“当前状态”和“历史参考”。4.4 运行与验证在 recuris-demo 目录下执行python agent.py运行后的预期输出大致如下[第 1 步] 模型动作填写邮箱 [执行结果] ok页面响应正常可以继续下一步 [第 2 步] 模型动作设置密码 [执行结果] ok页面响应正常可以继续下一步 [第 3 步] 模型动作提交表单 [执行结果] success提交成功页面跳转到完成页 [第 4 步] 模型动作FINISH 任务完成已写入情景记忆。 最终结果: True 情景记忆库