长程智能体如何不“失忆”？Recuris双记忆机制全解析

长程智能体做多了之后，会发现一个很典型的失败模式：任务在 5 步以内能完成得很好，一旦超过 10 步，模型就开始“失忆”，要么忘记最初的目标，要么把上一步的错误结果当成正确前提继续向下做，最后整条任务链崩塌。网上方案很多，但大多只解决“单步决策”，并没有解决“如何让智能体记住自己做到哪了、下次怎么更少犯错”。Recuris 双记忆机制就是针对这个问题设计的一套解决方案，本文会从原理、设计、代码到评测，完整拆解这套机制的落地思路。

Recuris 可以理解为一套面向长程任务设计的智能体系统方案，其核心创新是双记忆机制：把智能体的记忆拆成工作记忆和情景记忆两个通道。工作记忆负责管理当前任务的状态，防止目标漂移；情景记忆负责沉淀历史任务的经验，避免重复试错。两类记忆协同工作，能明显提升长程智能体的成功率。

如果你正在开发 Agent、做网页自动操作、多轮工具调用，或者研究长程任务规划，这篇文章会比较有用。下面从长程智能体为什么难讲起，逐步拆解双记忆机制。

1. 长程智能体为什么这么难做

1.1 什么是长程智能体

长程智能体，英文常写为 Long-horizon Agent，指的是需要执行多步骤、长时间跨度任务的智能体系统。典型场景包括：

• 网页自动操作：打开页面、填写表单、点击按钮、翻页、提交数据。
• 数据分析流水线：读取文件、清洗数据、建模、输出报告。
• 智能客服解决复杂问题：多轮询问、查询订单、协调多个系统。
• 机器人任务序列：导航、抓取、放置、再回到起点。

这类任务的共同点是：单步动作不难，难的是把几十步动作串联起来，并且保证每一步都依赖正确的历史状态。

1.2 长程任务的三个核心挑战

长程智能体成功率不高，根源集中在三个问题上。

第一个问题是上下文爆炸。每一步决策都要带上历史信息，但大模型上下文窗口有限。当轨迹越来越长，要么强行截断导致历史丢失，要么全部塞进去导致输入过长、成本过高。

第二个问题是目标漂移。模型在中间步骤看到大量中间状态后，容易被当前局部信息带偏，忘记最初的用户意图。最常见的表现是：任务做到一半，模型开始“自由发挥”，做了一些不在目标范围内的事情。

第三个问题是错误累积。长程任务中，一步出错往往会污染后续所有步骤。前一步状态错了，下一步基于错误状态推理，后面的步骤全部失去意义，而且错误很难定位。

这三个问题互相叠加，导致任务越长，成功率下降越明显。

1.3 Recuris 的解决思路

Recuris 没有把精力放在“让单步推理更强”上，而是把问题拆成记忆问题：当前任务状态记不住，那就用工作记忆管理；历史经验用不起来，那就用情景记忆沉淀。

工作记忆解决目标漂移和上下文爆炸，情景记忆解决错误累积和重复试错。两条通道组合起来，正好覆盖长程任务最薄弱的环节。

2. 双记忆机制总体设计

2.1 工作记忆与情景记忆的分工

工作记忆，类似人类脑子里的“草稿纸”。它保存的信息包括：当前任务目标、已经完成的子任务、最近几步的执行轨迹、以及被压缩后的历史摘要。特点是非常轻量、更新频繁、任务结束即清理。

情景记忆，类似人类的“经验库”。它保存的是历史任务的完整经验：任务类型、任务目标、执行轨迹、最终状态、失败原因。特点是数据量更大、需要检索、长期保留。

两者的差异可以用表格概括：

2.2 为什么需要两条记忆通道

一个很自然的疑问是：为什么不直接把所有历史信息放进一个大记忆池，而要做两个通道？

原因在于两类记忆的读写模式完全不同。工作记忆是高频写、高频读，每一步决策都要看；情景记忆是低频写、按需读，只有遇到相似任务时才用到。如果把两者混在一起，工作记忆会把历史任务的噪声带进当前决策，情景记忆也会因为频繁被无关上下文干扰而失去参考价值。

分开之后，工作记忆只负责“当前我在哪”，情景记忆只负责“过去类似任务怎么做”，职责清晰，提示词也更容易控制。

2.3 双记忆机制的运行流程

整个流程可以用六个阶段概括：

1. 初始化：把任务目标写入工作记忆，并在情景记忆中检索相似历史经验。
2. 决策：将工作记忆里的目标、摘要、最近轨迹，和情景记忆返回的相关经验，一起组装成提示词，交给大模型。
3. 执行：大模型输出动作，Agent 调用工具或环境接口执行该动作。
4. 观察：拿到动作执行结果，写入工作记忆。
5. 更新：如果轨迹过长，触发工作记忆压缩，把旧步骤生成摘要，释放空间。
6. 沉淀：任务成功或失败后，将整条轨迹写入情景记忆，供后续任务参考。

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.py

memory.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(maxlen=window_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(key=lambda x: x[0], reverse=True) 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(maxlen=window_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( model="gpt-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_k=2 ) 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_size=4) em = EpisodicMemory() # 预先放一条历史经验 em.add_episode({ "task_type": "网页表单填写", "goal": "注册账号", "trajectory": "打开注册页 -> 填写邮箱 -> 设置密码 -> 提交表单", "status": "success", }) agent = RecurisAgent( working_memory=wm, episodic_memory=em, llm_fn=StepMockLLM(), max_steps=10, ) executor = DemoActionExecutor() success = agent.run( task_goal="注册账号", task_type="网页表单填写", action_fn=executor.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 情景记忆库