之前在训练强化学习策略时，我遇到过一种非常典型的现象：策略网络结构没问题，奖励函数也调了很多轮，采样步数甚至拉到了百万级，但智能体就是练不上去，甚至越训越差。后来把采集到的轨迹翻出来一条一条排查，才发现问题出在“学习素材”上——大量样本来自那些已经无法挽回的状态。智能体在危险区域附近反复试探，得到的不是有效反馈，而是一堆把价值估计越带越偏的噪声。
这个问题的本质，是策略学错了东西。强化学习里我们经常讨论 policy gradient、Q-learning 这些更新方式，却很少认真回答一个问题：策略到底应该从哪些数据里学习？今天我们围绕一个研究方向展开：Recoverability-aware Rollout Intervention Learning，也就是“可恢复性感知的 Rollout 干预学习”。它关注的是如何在采样阶段就筛选、干预、优化策略的学习来源，而不是等到损失函数计算时再做补救。
这篇文章适合两类读者：一类是刚接触强化学习，想搞清楚 rollout、intervention 这些概念的学生；另一类是在做决策控制、机器人导航、自动驾驶仿真等方向，被“采样效率低”“训练不收敛”困扰的开发者。读完你会理解可恢复性的定义与计算思路，并拿到一套可以运行的 Python 演示代码，直接观察干预前后样本质量的变化。
1. 先从两个关键概念说起：Rollout 与 Intervention
1.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, size=6): 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, horizon=20, num_trials=30): """ 通过多次模拟估计状态 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, horizon=20, num_trials=30): """ 计算状态空间中所有合法状态的可恢复性，返回字典。 """ 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_steps=50): """ 基础 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, threshold=0.3, intervention_mode='reset', max_steps=50, expert_policy=None): """ 可恢复性感知的 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(size=6) policy = make_random_policy(env) print(">> Step 1: 计算状态空间可恢复性...") recover_map = compute_recoverability_map( env, policy, horizon=20, num_trials=50 ) 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_steps=50)) print(">> Step 3: recoverability-aware rollout 采集 1000 条轨迹...") aware_samples = [] for _ in range(1000): aware_samples.extend(recoverability_aware_rollout( env, policy, recover_map, threshold=0.3, intervention_mode='reset', max_steps=50 )) 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(f"Recoverability-aware 样本数={len(aware_samples)}, " f"到达目标占比={aware_goal:.4f}, 进入危险区占比={aware_danger:.4f}")
