之前在训练强化学习策略时我遇到过一种非常典型的现象策略网络结构没问题奖励函数也调了很多轮采样步数甚至拉到了百万级但智能体就是练不上去甚至越训越差。后来把采集到的轨迹翻出来一条一条排查才发现问题出在“学习素材”上——大量样本来自那些已经无法挽回的状态。智能体在危险区域附近反复试探得到的不是有效反馈而是一堆把价值估计越带越偏的噪声。这个问题的本质是策略学错了东西。强化学习里我们经常讨论 policy gradient、Q-learning 这些更新方式却很少认真回答一个问题策略到底应该从哪些数据里学习今天我们围绕一个研究方向展开Recoverability-aware Rollout Intervention Learning也就是“可恢复性感知的 Rollout 干预学习”。它关注的是如何在采样阶段就筛选、干预、优化策略的学习来源而不是等到损失函数计算时再做补救。这篇文章适合两类读者一类是刚接触强化学习想搞清楚 rollout、intervention 这些概念的学生另一类是在做决策控制、机器人导航、自动驾驶仿真等方向被“采样效率低”“训练不收敛”困扰的开发者。读完你会理解可恢复性的定义与计算思路并拿到一套可以运行的 Python 演示代码直接观察干预前后样本质量的变化。1. 先从两个关键概念说起Rollout 与 Intervention1.1 策略到底在“学习什么”强化学习的训练循环看起来很简单策略与环境交互收集数据更新参数再交互。但“收集数据”这四个字背后隐藏着巨大的差异。同样是 10000 条样本如果其中有 8000 条来自智能体已经陷入死胡同的状态那么这 8000 条样本不仅无法告诉策略“应该怎么做”反而会强化“错误路径也有价值”的假象。更准确地说策略学习的是一个从状态到动作的映射。这个映射的质量取决于训练数据的覆盖范围和质量。如果数据集中在低价值、不可恢复的状态区域策略就会在这些区域浪费大量表达能力。这也是为什么我们首先要思考“优化策略学习什么”而不是只盯着优化器怎么更新。1.2 Rollout策略的采样与试错Rollout 是强化学习中的常用术语指的是让策略从某个初始状态开始与环境交互若干步直到到达终止状态或达到最大步数限制。每执行一次 rollout我们会得到一条轨迹轨迹由一系列 (状态, 动作, 奖励, 下一状态) 组成。在 Python 里rollout 函数通常是训练框架中最基础的一个工具函数。理解它的写法很关键因为后面所有的干预逻辑都要嵌在这个采样循环里。最简单的 rollout 函数长这样从环境重置开始循环中选择动作、执行动作、记录元组直到回合结束。需要强调的是普通 rollout 是“盲采”的它不会判断当前状态是否有意义也不会判断这条轨迹对策略更新是正向帮助还是负向干扰。它把所有样本一视同仁地交给学习器。当环境简单时这没问题一旦环境存在大量不可恢复状态盲采就会严重拖慢训练。1.3 Intervention在采样过程中人为介入Intervention 的中文常翻译为“干预”。在强化学习里干预指的是在采样过程中引入外部信号用来改变智能体的行为或学习方向。最常见的干预形式有重置干预当智能体进入危险状态时强制终止当前回合把智能体重置到安全起点。动作干预当智能体面临高风险决策时用专家策略或规则策略代替当前策略输出动作。样本干预对采样到的样本进行加权或过滤低质量样本不参与后续学习。干预并不是强化学习的“外挂”而是很多实用系统中不可或缺的组件。比如机器人训练时工程师会设置安全边界一旦机械臂接近碰撞区域就立刻停止或接管控制。这就是一种典型的干预。它的目的不是替代策略而是保证策略不在错误方向上浪费时间。1.4 两者的结合点Recoverability把 Rollout 和 Intervention 放在一起看很自然会冒出一个问题什么时候该干预一种简单粗暴的做法是设置固定阈值比如“每 10 步干预一次”或“进入红色区域就重置”。但这忽视了状态本身的差异。有些状态虽然距离危险区域很近但只要有足够步数就能绕出来而有些状态看起来离目标不远实际上四周都是死路根本无法恢复。这里就引出了 Recoverability也就是“可恢复性”。可恢复性描述的是从某个状态出发策略是否还有机会在有限步数内回到安全区域或到达目标。如果一个状态的可恢复性很低那么策略从这个状态学到的样本大概率是噪声反之如果可恢复性很高即使当前奖励不理想这些样本也有学习价值。Recoverability-aware就是“可恢复性感知”的意思。它要求采样过程动态评估每个状态的可恢复性并据此决定是否干预、如何干预。这就是本文标题的核心逻辑优化策略学习的数据来源而不是盲目增加采样量。2. Recoverability-aware 的核心思想2.1 什么是可恢复性先给出一个比较直观的理解可恢复性是一个概率值表示从状态 s 出发按照某个策略通常是当前策略或随机策略执行在给定的步数预算内能够到达安全区域而不进入危险区域的概率。形式化地说对于状态 s、策略 π、步数预算 T可恢复性 R(s) 可以写作R(s) P(从 s 出发在 T 步内到达目标且不进入危险区域 | 策略 π)这个定义有几个关键点可恢复性依赖于策略 π。同一个状态对随机策略来说可能很难恢复但对一个训练有素的策略来说很容易恢复。因此在训练初期和训练后期同一个状态的可恢复性可能差异很大。可恢复性受步数预算 T 的影响。T 越大状态的可恢复性通常越高T 越小评估越严格。可恢复性不同于“距离目标的远近”。一个距离目标 2 步但被障碍物包围的状态可恢复性可能比距离目标 10 步但路径畅通的状态更低。在工程实现中我们通常用蒙特卡洛模拟来估计 R(s)从状态 s 出发重复执行多次 rollout统计成功到达目标的次数占比。这个估计方法简单、直观代价是需要额外计算量但相比训练不收敛带来的损失这笔开销通常可以接受。2.2 传统 rollout 存在的问题传统的 rollout 采样有三个容易被忽视的问题第一个问题是样本污染。当智能体进入不可恢复状态后它接下来的动作几乎都是随机的、无意义的挣扎。这些挣扎产生的样本会被当作正常经验存入回放缓冲区参与价值函数更新导致 Q 值估计被污染。第二个问题是探索浪费。智能体把大量时间花在反复进入死胡同、反复碰壁上却没有办法利用这些失败经验改善策略。因为它根本没有足够的“成功到达目标”的样本来学习正确路径。第三个问题是干预滞后。很多系统只在“已经进入危险区域”时才触发干预但此时样本已经产生了错误的影响已经进入了学习器。可恢复性感知的思路是提前预测在状态即将变得不可恢复之前就进行干预。2.3 可恢复性感知的学习优化思路理解了问题解决方案就清晰了。整个流程可以拆成三步第一步评估。在每次 rollout 开始前或在 rollout 进行中对当前状态计算可恢复性 R(s)。计算可以基于真实环境模拟也可以基于学习到的环境模型或动力学模型。第二步决策。设定一个阈值 α。当 R(s) α 时认为当前状态不可恢复或难以恢复需要触发干预当 R(s) ≥ α 时认为状态安全继续正常采样。第三步干预。根据具体任务选择干预方式重置终止当前轨迹将智能体重置到起点或某个安全状态。专家介入将动作替换为专家策略的输出把智能体引导回高可恢复性区域。样本过滤给低可恢复性样本打上低权重标记在更新损失函数时降低它们的贡献。这种思路的关键好处是策略不再被动地接收所有数据而是主动选择“值得学习”的数据。它优化的是学习材料的质量而非学习算法的公式。2.4 与常见 RL 方法的区别与联系有人可能会问这和 PPO、DQN 里的经验回放有什么关系和模仿学习里的专家数据又有什么区别经验回放解决的是样本利用率问题它让样本可以被多次使用但不区分样本来源是否可靠。可恢复性感知解决的是样本质量问题它在样本进入回放缓冲区之前就进行了筛选和干预。行为克隆和 GAIL 这类模仿学习方法依赖专家数据它们假设专家数据是高质量的。而可恢复性感知不依赖专家数据它只需要环境本身的信息就可以判断哪些状态值得探索哪些状态应该回避。从更广的角度看这个思路和连续学习、自监督强化学习也有联系。比如在连续学习场景中旧任务中学到的策略可能在新任务中产生不可恢复的状态这时候就需要根据可恢复性做干预避免灾难性遗忘。不过本文先聚焦最基础的采样干预机制其他扩展放在文章末尾讨论。3. 实验环境与场景设计3.1 环境与版本说明为了让读者能直接运行我选用一个非常轻量的演示环境Python 编写的网格世界。它不依赖 Gym、PyTorch 等大型框架只用到 Python 标准库和可选的 NumPy。本文示例代码在以下环境中验证过Python 3.9 及以上无需安装第三方库如果安装 NumPy 也可以但演示代码不强制使用操作系统Windows / macOS / Linux 均可版本需要根据你的项目实际情况调整。如果你使用其他 Python 版本代码兼容性应该没有问题但建议尽量保持 3.8 以上方便使用类型注解和 f-string 等语法特性。3.2 网格世界场景设计我们设计一个 6×6 的网格世界包含以下元素S起点位于左上角 (0, 0)。G终点位于右下角 (5, 5)。X障碍物不可通行。D危险区域进入即视为不可恢复回合立即结束并给予较大负奖励。在这个场景里普通随机策略很容易进入危险区域或者在障碍物附近兜圈子。通过对比普通 rollout 和可恢复性感知 rollout 采集到的样本质量我们可以直观看到干预的效果。地图布局示意如下S . . . . . . X . . . . . . . X . . . X . . D . . . . . . . . . D . . G实际代码中用集合保存障碍物和危险区域坐标渲染函数会打印出当前智能体位置。3.3 项目文件结构整个演示项目放在一个名为rl_recoverability_demo的目录中文件结构如下rl_recoverability_demo/ ├── env.py # 网格世界环境 ├── recoverability.py # 可恢复性评估模块 ├── rollouts.py # rollout 采样与干预逻辑 ├── train.py # 对比实验主程序 └── README.md # 说明文档可选接下来我们按模块逐个实现。4. 核心代码实现与拆解4.1 环境模块 env.py先定义地图边界环境模块负责维护状态转移、奖励和终止条件。这里要注意的是我们把“危险区域”和“障碍物”分开处理障碍物不可通行智能体撞上后会留在原地危险区域可以进入但进入后回合直接终止。这种设计是为了模拟真实场景中“不可恢复”的状态分支。# -*- coding: utf-8 -*- # 文件路径rl_recoverability_demo/env.py 简易网格世界环境用于演示 Recoverability-aware Rollout Intervention Learning。 地图元素说明 S : 起点 G : 终点 X : 障碍物 D : 危险区域 class GridWorldEnv: def __init__(self, size6): self.size size self.start (0, 0) self.goal (size - 1, size - 1) self.obstacles {(1, 1), (2, 3), (3, 1)} self.danger {(4, 2), (5, 3), (2, 4)} # 动作空间0上1下2左3右 self.action_space [0, 1, 2, 3] self.action_map { 0: (-1, 0), 1: (1, 0), 2: (0, -1), 3: (0, 1), } self.state None self.step_count 0 self.max_steps 50 def reset(self): 重置环境到起点。 self.state self.start self.step_count 0 return self.state def is_valid(self, pos): 判断位置是否合法在地图内且不是障碍物。 r, c pos if not (0 r self.size and 0 c self.size): return False if pos in self.obstacles: return False return True def step(self, action): 执行一步动作。 返回 (next_state, reward, done)。 r, c self.state dr, dc self.action_map[action] next_state (r dr, c dc) # 撞墙或障碍物时留在原地 if not self.is_valid(next_state): next_state self.state self.state next_state self.step_count 1 if next_state in self.danger: reward -10.0 done True elif next_state self.goal: reward 10.0 done True else: reward -0.1 done False # 超过最大步数视为回合截断 truncated self.step_count self.max_steps return next_state, reward, done or truncated def render(self): 在控制台打印当前地图。 grid [[. for _ in range(self.size)] for _ in range(self.size)] for r, c in self.obstacles: grid[r][c] X for r, c in self.danger: grid[r][c] D sr, sc self.start gr, gc self.goal grid[sr][sc] S grid[gr][gc] G if self.state: sr, sc self.state grid[sr][sc] * for row in grid: print( .join(row))这段代码有几个细节值得说明is_valid方法统一处理越界和障碍物判断避免在两个地方重复写逻辑。危险区域不放在is_valid的排除范围内因为我们要允许智能体进入然后触发终止这样才能模拟“不可恢复”的代价。truncated和done合并返回方便后续 rollout 循环统一处理。4.2 可恢复性评估模块 recoverability.py可恢复性评估是整个方法的核心。我们采用蒙特卡洛模拟的方式对每个状态使用给定的策略这里用随机策略从该状态出发做多次模拟统计在步数预算内到达目标的概率。# -*- coding: utf-8 -*- # 文件路径rl_recoverability_demo/recoverability.py 可恢复性评估模块。 可恢复性定义 给定状态 s 和策略 πR(s) 表示从 s 出发、按照策略 π 执行 在有限步数内到达目标区域且不进入危险区域的概率。 import random def make_random_policy(env): 创建一个随机策略函数。 策略接口policy(state) - action def policy(state): return random.choice(env.action_space) return policy def is_state_recoverable(env, state, policy, horizon20, num_trials30): 通过多次模拟估计状态 state 的可恢复性。 参数 env : 环境对象 state : 待评估状态 policy : 策略函数输入 state输出 action horizon : 步数预算 num_trials : 模拟次数 返回 0~1 之间的概率值。 success 0 for _ in range(num_trials): current state env.state current env.step_count 0 reached False for _ in range(horizon): action policy(current) next_state, _, done env.step(action) current next_state if current in env.danger: break if current env.goal: reached True break if done: break if reached: success 1 # 恢复环境状态避免污染后续实验 env.state state env.step_count 0 return success / max(num_trials, 1) def compute_recoverability_map(env, policy, horizon20, num_trials30): 计算状态空间中所有合法状态的可恢复性返回字典。 recover_map {} for r in range(env.size): for c in range(env.size): state (r, c) if not env.is_valid(state) or state in env.danger: recover_map[state] 0.0 continue recover_map[state] is_state_recoverable( env, state, policy, horizon, num_trials ) return recover_map这里的实现有几个容易踩坑的地方我提前说明每次模拟前必须重置env.step_count否则环境内部的步数计数会累积导致truncated提前触发。模拟结束后要把env.state恢复为传入的状态防止影响后续计算。虽然评估函数内部不依赖外部状态但保持环境干净是好的工程习惯。num_trials不能太小否则概率估计方差太大。演示中取 30~50 次实际项目可以根据计算资源调整。4.3 Rollout 与干预模块 rollouts.py这一节对应热搜词“python 中 rollout 函数的用法”。我们实现两个函数普通rollout和recoverability_aware_rollout。前者是所有强化学习采样循环的基础后者在前者基础上加入可恢复性判断和干预逻辑。# -*- coding: utf-8 -*- # 文件路径rl_recoverability_demo/rollouts.py Rollout 采样与干预逻辑。 核心思路 1. 用当前策略执行 rollout记录轨迹 2. 对轨迹中的每个状态评估可恢复性 3. 当可恢复性低于阈值时按指定模式干预 4. 返回轨迹供上层策略学习使用。 import random def rollout(env, policy, max_steps50): 基础 rollout 函数让策略从起点出发与环境交互采集一条轨迹。 参数 env : 环境对象 policy : 策略函数输入 state输出 action max_steps : 单条轨迹最大步数 返回 轨迹列表每个元素是六元组 (state, action, reward, next_state, done, valid) valid 表示该样本是否被判定为有效学习样本。 state env.reset() trajectory [] done False step 0 while not done and step max_steps: action policy(state) next_state, reward, done env.step(action) # 普通 rollout 不筛样本全部标记为有效 trajectory.append((state, action, reward, next_state, done, True)) state next_state step 1 return trajectory def recoverability_aware_rollout(env, policy, recover_map, threshold0.3, intervention_modereset, max_steps50, expert_policyNone): 可恢复性感知的 rollout 采样。 参数 env : 环境对象 policy : 当前策略 recover_map : 状态到可恢复性的映射 threshold : 可恢复性阈值低于该值触发干预 intervention_mode : 干预模式支持 reset / mask / expert max_steps : 最大步数 expert_policy : 专家策略仅在 expert 模式下使用 返回 轨迹列表元素结构与 rollout 函数一致。 state env.reset() trajectory [] done False step 0 while not done and step max_steps: recover_value recover_map.get(state, 0.0) if recover_value threshold: # ---------- 干预分支 ---------- if intervention_mode reset: # 中止当前轨迹低可恢复性样本不进入学习器 break elif intervention_mode mask: # 样本仍采集但 valid 标记为 False供上层过滤 action policy(state) next_state, reward, done env.step(action) trajectory.append((state, action, reward, next_state, done, False)) state next_state step 1 continue elif intervention_mode expert: # 使用专家策略介入引导智能体离开低可恢复性区域 if expert_policy is None: expert_policy lambda s: random.choice(env.action_space) action expert_policy(state) next_state, reward, done env.step(action) trajectory.append((state, action, reward, next_state, done, True)) state next_state step 1 continue else: # 未知模式保守处理直接中止 break else: # ---------- 正常采样分支 ---------- action policy(state) next_state, reward, done env.step(action) trajectory.append((state, action, reward, next_state, done, True)) state next_state step 1 return trajectoryrollout函数本身很简单但它定义了所有高级采样逻辑的骨架。你可以在此基础上加入 epsilon-greedy、随机噪声、状态归一化等操作核心循环结构不会变。recoverability_aware_rollout的关键在于决策不是在“已经进入危险区”之后而是在“即将进入低可恢复性状态”之前。recover_map.get(state, 0.0)的默认值设为 0.0意味着如果某个状态不在评估表中我们宁可保守地认为它不可恢复也不冒险继续探索。干预模式的选择需要结合任务特点。reset模式最简单但会显著减少样本数量mask模式保留样本但标记无效适合配合带权重的损失函数使用expert模式能维持样本数量但依赖专家策略的质量实际项目中专家策略可以是规则、模型预测控制或者人工遥控。4.4 对比主程序 train.py为了直观展示可恢复性感知采样的价值我们写一个对比实验分别用普通 rollout 和 recoverability-aware rollout 各采集 1000 条轨迹然后统计样本中“到达目标”和“进入危险区域”的比例。前者反映样本的最终收益后者反映样本的安全性。# -*- coding: utf-8 -*- # 文件路径rl_recoverability_demo/train.py 对比实验普通 rollout 与 recoverability-aware rollout 的样本质量。 import random from env import GridWorldEnv from rollouts import rollout, recoverability_aware_rollout from recoverability import compute_recoverability_map, make_random_policy def sample_quality(samples, env): 统计样本到达目标与进入危险区域的比例。 if not samples: return 0.0, 0.0 goal_hits sum(1 for s in samples if s[3] env.goal) danger_hits sum(1 for s in samples if s[3] in env.danger) return goal_hits / len(samples), danger_hits / len(samples) def run_demo(): env GridWorldEnv(size6) policy make_random_policy(env) print( Step 1: 计算状态空间可恢复性...) recover_map compute_recoverability_map( env, policy, horizon20, num_trials50 ) high_recover sum(1 for v in recover_map.values() if v 0.5) print(f 可恢复性大于 0.5 的状态数: {high_recover}/{len(recover_map)}) print(\n Step 2: 普通 rollout 采集 1000 条轨迹...) normal_samples [] for _ in range(1000): normal_samples.extend(rollout(env, policy, max_steps50)) print( Step 3: recoverability-aware rollout 采集 1000 条轨迹...) aware_samples [] for _ in range(1000): aware_samples.extend(recoverability_aware_rollout( env, policy, recover_map, threshold0.3, intervention_modereset, max_steps50 )) normal_goal, normal_danger sample_quality(normal_samples, env) aware_goal, aware_danger sample_quality(aware_samples, env) print(\n 对比结果 ) print(f普通 rollout 样本数{len(normal_samples)}, f到达目标占比{normal_goal:.4f}, 进入危险区占比{normal_danger:.4f}) print(fRecoverability-aware 样本数{len(aware_samples)}, f到达目标占比{aware_goal:.4f}, 进入危险区占比{aware_danger:.4f}) 强化学习中的可恢复性感知Rollout干预：优化策略学习的采样质量
之前在训练强化学习策略时我遇到过一种非常典型的现象策略网络结构没问题奖励函数也调了很多轮采样步数甚至拉到了百万级但智能体就是练不上去甚至越训越差。后来把采集到的轨迹翻出来一条一条排查才发现问题出在“学习素材”上——大量样本来自那些已经无法挽回的状态。智能体在危险区域附近反复试探得到的不是有效反馈而是一堆把价值估计越带越偏的噪声。这个问题的本质是策略学错了东西。强化学习里我们经常讨论 policy gradient、Q-learning 这些更新方式却很少认真回答一个问题策略到底应该从哪些数据里学习今天我们围绕一个研究方向展开Recoverability-aware Rollout Intervention Learning也就是“可恢复性感知的 Rollout 干预学习”。它关注的是如何在采样阶段就筛选、干预、优化策略的学习来源而不是等到损失函数计算时再做补救。这篇文章适合两类读者一类是刚接触强化学习想搞清楚 rollout、intervention 这些概念的学生另一类是在做决策控制、机器人导航、自动驾驶仿真等方向被“采样效率低”“训练不收敛”困扰的开发者。读完你会理解可恢复性的定义与计算思路并拿到一套可以运行的 Python 演示代码直接观察干预前后样本质量的变化。1. 先从两个关键概念说起Rollout 与 Intervention1.1 策略到底在“学习什么”强化学习的训练循环看起来很简单策略与环境交互收集数据更新参数再交互。但“收集数据”这四个字背后隐藏着巨大的差异。同样是 10000 条样本如果其中有 8000 条来自智能体已经陷入死胡同的状态那么这 8000 条样本不仅无法告诉策略“应该怎么做”反而会强化“错误路径也有价值”的假象。更准确地说策略学习的是一个从状态到动作的映射。这个映射的质量取决于训练数据的覆盖范围和质量。如果数据集中在低价值、不可恢复的状态区域策略就会在这些区域浪费大量表达能力。这也是为什么我们首先要思考“优化策略学习什么”而不是只盯着优化器怎么更新。1.2 Rollout策略的采样与试错Rollout 是强化学习中的常用术语指的是让策略从某个初始状态开始与环境交互若干步直到到达终止状态或达到最大步数限制。每执行一次 rollout我们会得到一条轨迹轨迹由一系列 (状态, 动作, 奖励, 下一状态) 组成。在 Python 里rollout 函数通常是训练框架中最基础的一个工具函数。理解它的写法很关键因为后面所有的干预逻辑都要嵌在这个采样循环里。最简单的 rollout 函数长这样从环境重置开始循环中选择动作、执行动作、记录元组直到回合结束。需要强调的是普通 rollout 是“盲采”的它不会判断当前状态是否有意义也不会判断这条轨迹对策略更新是正向帮助还是负向干扰。它把所有样本一视同仁地交给学习器。当环境简单时这没问题一旦环境存在大量不可恢复状态盲采就会严重拖慢训练。1.3 Intervention在采样过程中人为介入Intervention 的中文常翻译为“干预”。在强化学习里干预指的是在采样过程中引入外部信号用来改变智能体的行为或学习方向。最常见的干预形式有重置干预当智能体进入危险状态时强制终止当前回合把智能体重置到安全起点。动作干预当智能体面临高风险决策时用专家策略或规则策略代替当前策略输出动作。样本干预对采样到的样本进行加权或过滤低质量样本不参与后续学习。干预并不是强化学习的“外挂”而是很多实用系统中不可或缺的组件。比如机器人训练时工程师会设置安全边界一旦机械臂接近碰撞区域就立刻停止或接管控制。这就是一种典型的干预。它的目的不是替代策略而是保证策略不在错误方向上浪费时间。1.4 两者的结合点Recoverability把 Rollout 和 Intervention 放在一起看很自然会冒出一个问题什么时候该干预一种简单粗暴的做法是设置固定阈值比如“每 10 步干预一次”或“进入红色区域就重置”。但这忽视了状态本身的差异。有些状态虽然距离危险区域很近但只要有足够步数就能绕出来而有些状态看起来离目标不远实际上四周都是死路根本无法恢复。这里就引出了 Recoverability也就是“可恢复性”。可恢复性描述的是从某个状态出发策略是否还有机会在有限步数内回到安全区域或到达目标。如果一个状态的可恢复性很低那么策略从这个状态学到的样本大概率是噪声反之如果可恢复性很高即使当前奖励不理想这些样本也有学习价值。Recoverability-aware就是“可恢复性感知”的意思。它要求采样过程动态评估每个状态的可恢复性并据此决定是否干预、如何干预。这就是本文标题的核心逻辑优化策略学习的数据来源而不是盲目增加采样量。2. Recoverability-aware 的核心思想2.1 什么是可恢复性先给出一个比较直观的理解可恢复性是一个概率值表示从状态 s 出发按照某个策略通常是当前策略或随机策略执行在给定的步数预算内能够到达安全区域而不进入危险区域的概率。形式化地说对于状态 s、策略 π、步数预算 T可恢复性 R(s) 可以写作R(s) P(从 s 出发在 T 步内到达目标且不进入危险区域 | 策略 π)这个定义有几个关键点可恢复性依赖于策略 π。同一个状态对随机策略来说可能很难恢复但对一个训练有素的策略来说很容易恢复。因此在训练初期和训练后期同一个状态的可恢复性可能差异很大。可恢复性受步数预算 T 的影响。T 越大状态的可恢复性通常越高T 越小评估越严格。可恢复性不同于“距离目标的远近”。一个距离目标 2 步但被障碍物包围的状态可恢复性可能比距离目标 10 步但路径畅通的状态更低。在工程实现中我们通常用蒙特卡洛模拟来估计 R(s)从状态 s 出发重复执行多次 rollout统计成功到达目标的次数占比。这个估计方法简单、直观代价是需要额外计算量但相比训练不收敛带来的损失这笔开销通常可以接受。2.2 传统 rollout 存在的问题传统的 rollout 采样有三个容易被忽视的问题第一个问题是样本污染。当智能体进入不可恢复状态后它接下来的动作几乎都是随机的、无意义的挣扎。这些挣扎产生的样本会被当作正常经验存入回放缓冲区参与价值函数更新导致 Q 值估计被污染。第二个问题是探索浪费。智能体把大量时间花在反复进入死胡同、反复碰壁上却没有办法利用这些失败经验改善策略。因为它根本没有足够的“成功到达目标”的样本来学习正确路径。第三个问题是干预滞后。很多系统只在“已经进入危险区域”时才触发干预但此时样本已经产生了错误的影响已经进入了学习器。可恢复性感知的思路是提前预测在状态即将变得不可恢复之前就进行干预。2.3 可恢复性感知的学习优化思路理解了问题解决方案就清晰了。整个流程可以拆成三步第一步评估。在每次 rollout 开始前或在 rollout 进行中对当前状态计算可恢复性 R(s)。计算可以基于真实环境模拟也可以基于学习到的环境模型或动力学模型。第二步决策。设定一个阈值 α。当 R(s) α 时认为当前状态不可恢复或难以恢复需要触发干预当 R(s) ≥ α 时认为状态安全继续正常采样。第三步干预。根据具体任务选择干预方式重置终止当前轨迹将智能体重置到起点或某个安全状态。专家介入将动作替换为专家策略的输出把智能体引导回高可恢复性区域。样本过滤给低可恢复性样本打上低权重标记在更新损失函数时降低它们的贡献。这种思路的关键好处是策略不再被动地接收所有数据而是主动选择“值得学习”的数据。它优化的是学习材料的质量而非学习算法的公式。2.4 与常见 RL 方法的区别与联系有人可能会问这和 PPO、DQN 里的经验回放有什么关系和模仿学习里的专家数据又有什么区别经验回放解决的是样本利用率问题它让样本可以被多次使用但不区分样本来源是否可靠。可恢复性感知解决的是样本质量问题它在样本进入回放缓冲区之前就进行了筛选和干预。行为克隆和 GAIL 这类模仿学习方法依赖专家数据它们假设专家数据是高质量的。而可恢复性感知不依赖专家数据它只需要环境本身的信息就可以判断哪些状态值得探索哪些状态应该回避。从更广的角度看这个思路和连续学习、自监督强化学习也有联系。比如在连续学习场景中旧任务中学到的策略可能在新任务中产生不可恢复的状态这时候就需要根据可恢复性做干预避免灾难性遗忘。不过本文先聚焦最基础的采样干预机制其他扩展放在文章末尾讨论。3. 实验环境与场景设计3.1 环境与版本说明为了让读者能直接运行我选用一个非常轻量的演示环境Python 编写的网格世界。它不依赖 Gym、PyTorch 等大型框架只用到 Python 标准库和可选的 NumPy。本文示例代码在以下环境中验证过Python 3.9 及以上无需安装第三方库如果安装 NumPy 也可以但演示代码不强制使用操作系统Windows / macOS / Linux 均可版本需要根据你的项目实际情况调整。如果你使用其他 Python 版本代码兼容性应该没有问题但建议尽量保持 3.8 以上方便使用类型注解和 f-string 等语法特性。3.2 网格世界场景设计我们设计一个 6×6 的网格世界包含以下元素S起点位于左上角 (0, 0)。G终点位于右下角 (5, 5)。X障碍物不可通行。D危险区域进入即视为不可恢复回合立即结束并给予较大负奖励。在这个场景里普通随机策略很容易进入危险区域或者在障碍物附近兜圈子。通过对比普通 rollout 和可恢复性感知 rollout 采集到的样本质量我们可以直观看到干预的效果。地图布局示意如下S . . . . . . X . . . . . . . X . . . X . . D . . . . . . . . . D . . G实际代码中用集合保存障碍物和危险区域坐标渲染函数会打印出当前智能体位置。3.3 项目文件结构整个演示项目放在一个名为rl_recoverability_demo的目录中文件结构如下rl_recoverability_demo/ ├── env.py # 网格世界环境 ├── recoverability.py # 可恢复性评估模块 ├── rollouts.py # rollout 采样与干预逻辑 ├── train.py # 对比实验主程序 └── README.md # 说明文档可选接下来我们按模块逐个实现。4. 核心代码实现与拆解4.1 环境模块 env.py先定义地图边界环境模块负责维护状态转移、奖励和终止条件。这里要注意的是我们把“危险区域”和“障碍物”分开处理障碍物不可通行智能体撞上后会留在原地危险区域可以进入但进入后回合直接终止。这种设计是为了模拟真实场景中“不可恢复”的状态分支。# -*- coding: utf-8 -*- # 文件路径rl_recoverability_demo/env.py 简易网格世界环境用于演示 Recoverability-aware Rollout Intervention Learning。 地图元素说明 S : 起点 G : 终点 X : 障碍物 D : 危险区域 class GridWorldEnv: def __init__(self, size6): self.size size self.start (0, 0) self.goal (size - 1, size - 1) self.obstacles {(1, 1), (2, 3), (3, 1)} self.danger {(4, 2), (5, 3), (2, 4)} # 动作空间0上1下2左3右 self.action_space [0, 1, 2, 3] self.action_map { 0: (-1, 0), 1: (1, 0), 2: (0, -1), 3: (0, 1), } self.state None self.step_count 0 self.max_steps 50 def reset(self): 重置环境到起点。 self.state self.start self.step_count 0 return self.state def is_valid(self, pos): 判断位置是否合法在地图内且不是障碍物。 r, c pos if not (0 r self.size and 0 c self.size): return False if pos in self.obstacles: return False return True def step(self, action): 执行一步动作。 返回 (next_state, reward, done)。 r, c self.state dr, dc self.action_map[action] next_state (r dr, c dc) # 撞墙或障碍物时留在原地 if not self.is_valid(next_state): next_state self.state self.state next_state self.step_count 1 if next_state in self.danger: reward -10.0 done True elif next_state self.goal: reward 10.0 done True else: reward -0.1 done False # 超过最大步数视为回合截断 truncated self.step_count self.max_steps return next_state, reward, done or truncated def render(self): 在控制台打印当前地图。 grid [[. for _ in range(self.size)] for _ in range(self.size)] for r, c in self.obstacles: grid[r][c] X for r, c in self.danger: grid[r][c] D sr, sc self.start gr, gc self.goal grid[sr][sc] S grid[gr][gc] G if self.state: sr, sc self.state grid[sr][sc] * for row in grid: print( .join(row))这段代码有几个细节值得说明is_valid方法统一处理越界和障碍物判断避免在两个地方重复写逻辑。危险区域不放在is_valid的排除范围内因为我们要允许智能体进入然后触发终止这样才能模拟“不可恢复”的代价。truncated和done合并返回方便后续 rollout 循环统一处理。4.2 可恢复性评估模块 recoverability.py可恢复性评估是整个方法的核心。我们采用蒙特卡洛模拟的方式对每个状态使用给定的策略这里用随机策略从该状态出发做多次模拟统计在步数预算内到达目标的概率。# -*- coding: utf-8 -*- # 文件路径rl_recoverability_demo/recoverability.py 可恢复性评估模块。 可恢复性定义 给定状态 s 和策略 πR(s) 表示从 s 出发、按照策略 π 执行 在有限步数内到达目标区域且不进入危险区域的概率。 import random def make_random_policy(env): 创建一个随机策略函数。 策略接口policy(state) - action def policy(state): return random.choice(env.action_space) return policy def is_state_recoverable(env, state, policy, horizon20, num_trials30): 通过多次模拟估计状态 state 的可恢复性。 参数 env : 环境对象 state : 待评估状态 policy : 策略函数输入 state输出 action horizon : 步数预算 num_trials : 模拟次数 返回 0~1 之间的概率值。 success 0 for _ in range(num_trials): current state env.state current env.step_count 0 reached False for _ in range(horizon): action policy(current) next_state, _, done env.step(action) current next_state if current in env.danger: break if current env.goal: reached True break if done: break if reached: success 1 # 恢复环境状态避免污染后续实验 env.state state env.step_count 0 return success / max(num_trials, 1) def compute_recoverability_map(env, policy, horizon20, num_trials30): 计算状态空间中所有合法状态的可恢复性返回字典。 recover_map {} for r in range(env.size): for c in range(env.size): state (r, c) if not env.is_valid(state) or state in env.danger: recover_map[state] 0.0 continue recover_map[state] is_state_recoverable( env, state, policy, horizon, num_trials ) return recover_map这里的实现有几个容易踩坑的地方我提前说明每次模拟前必须重置env.step_count否则环境内部的步数计数会累积导致truncated提前触发。模拟结束后要把env.state恢复为传入的状态防止影响后续计算。虽然评估函数内部不依赖外部状态但保持环境干净是好的工程习惯。num_trials不能太小否则概率估计方差太大。演示中取 30~50 次实际项目可以根据计算资源调整。4.3 Rollout 与干预模块 rollouts.py这一节对应热搜词“python 中 rollout 函数的用法”。我们实现两个函数普通rollout和recoverability_aware_rollout。前者是所有强化学习采样循环的基础后者在前者基础上加入可恢复性判断和干预逻辑。# -*- coding: utf-8 -*- # 文件路径rl_recoverability_demo/rollouts.py Rollout 采样与干预逻辑。 核心思路 1. 用当前策略执行 rollout记录轨迹 2. 对轨迹中的每个状态评估可恢复性 3. 当可恢复性低于阈值时按指定模式干预 4. 返回轨迹供上层策略学习使用。 import random def rollout(env, policy, max_steps50): 基础 rollout 函数让策略从起点出发与环境交互采集一条轨迹。 参数 env : 环境对象 policy : 策略函数输入 state输出 action max_steps : 单条轨迹最大步数 返回 轨迹列表每个元素是六元组 (state, action, reward, next_state, done, valid) valid 表示该样本是否被判定为有效学习样本。 state env.reset() trajectory [] done False step 0 while not done and step max_steps: action policy(state) next_state, reward, done env.step(action) # 普通 rollout 不筛样本全部标记为有效 trajectory.append((state, action, reward, next_state, done, True)) state next_state step 1 return trajectory def recoverability_aware_rollout(env, policy, recover_map, threshold0.3, intervention_modereset, max_steps50, expert_policyNone): 可恢复性感知的 rollout 采样。 参数 env : 环境对象 policy : 当前策略 recover_map : 状态到可恢复性的映射 threshold : 可恢复性阈值低于该值触发干预 intervention_mode : 干预模式支持 reset / mask / expert max_steps : 最大步数 expert_policy : 专家策略仅在 expert 模式下使用 返回 轨迹列表元素结构与 rollout 函数一致。 state env.reset() trajectory [] done False step 0 while not done and step max_steps: recover_value recover_map.get(state, 0.0) if recover_value threshold: # ---------- 干预分支 ---------- if intervention_mode reset: # 中止当前轨迹低可恢复性样本不进入学习器 break elif intervention_mode mask: # 样本仍采集但 valid 标记为 False供上层过滤 action policy(state) next_state, reward, done env.step(action) trajectory.append((state, action, reward, next_state, done, False)) state next_state step 1 continue elif intervention_mode expert: # 使用专家策略介入引导智能体离开低可恢复性区域 if expert_policy is None: expert_policy lambda s: random.choice(env.action_space) action expert_policy(state) next_state, reward, done env.step(action) trajectory.append((state, action, reward, next_state, done, True)) state next_state step 1 continue else: # 未知模式保守处理直接中止 break else: # ---------- 正常采样分支 ---------- action policy(state) next_state, reward, done env.step(action) trajectory.append((state, action, reward, next_state, done, True)) state next_state step 1 return trajectoryrollout函数本身很简单但它定义了所有高级采样逻辑的骨架。你可以在此基础上加入 epsilon-greedy、随机噪声、状态归一化等操作核心循环结构不会变。recoverability_aware_rollout的关键在于决策不是在“已经进入危险区”之后而是在“即将进入低可恢复性状态”之前。recover_map.get(state, 0.0)的默认值设为 0.0意味着如果某个状态不在评估表中我们宁可保守地认为它不可恢复也不冒险继续探索。干预模式的选择需要结合任务特点。reset模式最简单但会显著减少样本数量mask模式保留样本但标记无效适合配合带权重的损失函数使用expert模式能维持样本数量但依赖专家策略的质量实际项目中专家策略可以是规则、模型预测控制或者人工遥控。4.4 对比主程序 train.py为了直观展示可恢复性感知采样的价值我们写一个对比实验分别用普通 rollout 和 recoverability-aware rollout 各采集 1000 条轨迹然后统计样本中“到达目标”和“进入危险区域”的比例。前者反映样本的最终收益后者反映样本的安全性。# -*- coding: utf-8 -*- # 文件路径rl_recoverability_demo/train.py 对比实验普通 rollout 与 recoverability-aware rollout 的样本质量。 import random from env import GridWorldEnv from rollouts import rollout, recoverability_aware_rollout from recoverability import compute_recoverability_map, make_random_policy def sample_quality(samples, env): 统计样本到达目标与进入危险区域的比例。 if not samples: return 0.0, 0.0 goal_hits sum(1 for s in samples if s[3] env.goal) danger_hits sum(1 for s in samples if s[3] in env.danger) return goal_hits / len(samples), danger_hits / len(samples) def run_demo(): env GridWorldEnv(size6) policy make_random_policy(env) print( Step 1: 计算状态空间可恢复性...) recover_map compute_recoverability_map( env, policy, horizon20, num_trials50 ) high_recover sum(1 for v in recover_map.values() if v 0.5) print(f 可恢复性大于 0.5 的状态数: {high_recover}/{len(recover_map)}) print(\n Step 2: 普通 rollout 采集 1000 条轨迹...) normal_samples [] for _ in range(1000): normal_samples.extend(rollout(env, policy, max_steps50)) print( Step 3: recoverability-aware rollout 采集 1000 条轨迹...) aware_samples [] for _ in range(1000): aware_samples.extend(recoverability_aware_rollout( env, policy, recover_map, threshold0.3, intervention_modereset, max_steps50 )) normal_goal, normal_danger sample_quality(normal_samples, env) aware_goal, aware_danger sample_quality(aware_samples, env) print(\n 对比结果 ) print(f普通 rollout 样本数{len(normal_samples)}, f到达目标占比{normal_goal:.4f}, 进入危险区占比{normal_danger:.4f}) print(fRecoverability-aware 样本数{len(aware_samples)}, f到达目标占比{aware_goal:.4f}, 进入危险区占比{aware_danger:.4f})