在上一篇文章中，我们总览了legged_robot.py代码架构。本文，我们将深入引擎的心脏——强化学习核心循环 (Core RL Loop)。

无论你使用的是 PPO、SAC 还是其他 RL 算法，算法端最终都需要一个标准的接口来与物理环境进行交互。在基于 Isaac Gym 等 GPU 并行加速的框架中，这个交互过程被高度优化并封装在了以下几个核心函数中。接下来，我们将按执行顺序逐一拆解。

---

1. 核心步进驱动：step(actions)

这是环境提供给算法的最顶层接口。每当神经网络输出了一组动作（Actions），就会调用这个函数。

• 输入处理与裁剪：网络输出的动作通常会被限制在一定范围内（例如 [-1, 1]），step 函数首先会接收这些动作并进行必要的裁剪，防止异常值导致物理引擎崩溃。

• 控制频率与物理步进 (Decimation)：这是 step 中非常关键的概念。通常神经网络的推理频率（如 50Hz）会低于物理引擎的仿真频率（如 200Hz）。因此，step 会包含一个内部循环，根据设定的 decimation（控制频率衰减系数，例如 4），将同一个动作转换为关节力矩后，在物理引擎中连续步进 4 次。

• 返回标准 RL 接口数据：步进完成后，它会收集并返回强化学习所需的标准数据包：下一步观测值 (obs)、特权观测（privileged_obs，仅限非对称 Actor-Critic 训练）、奖励 (rewards) 以及回合结束标志 (dones)。

    def step(self, actions):
        """ Apply actions, simulate, call self.post_physics_step()

        Args:
            actions (torch.Tensor): Tensor of shape (num_envs, num_actions_per_env)
        """
        clip_actions = self.cfg.normalization.clip_actions#获取限制动作的阈值
        self.actions = torch.clip(actions, -clip_actions, clip_actions).to(self.device)#限制并转换到指定设备
        # step physics and render each frame
        self.render()#更新图形界面（GUI）的显示内容。
        for _ in range(self.cfg.control.decimation):
            # 将动作转换为关节力矩 (PD 控制器)
            self.torques = self._compute_torques(self.actions).view(self.torques.shape)
            # 将力矩下发给底层 Isaac Gym 张量
            self.gym.set_dof_actuation_force_tensor(self.sim, gymtorch.unwrap_tensor(self.torques))
            self.gym.simulate(self.sim)# 步进一次物理仿真
            if self.device == 'cpu':
                self.gym.fetch_results(self.sim, True)
            self.gym.refresh_dof_state_tensor(self.sim)# 刷新关节状态张量
        self.post_physics_step()# 物理仿真后的状态更新、奖励计算等

        # return clipped obs, clipped states (None), rewards, dones and infos
        # 裁剪观测值，防止异常值干扰网络
        clip_obs = self.cfg.normalization.clip_observations
        self.obs_buf = torch.clip(self.obs_buf, -clip_obs, clip_obs)
        if self.privileged_obs_buf is not None:
            self.privileged_obs_buf = torch.clip(self.privileged_obs_buf, -clip_obs, clip_obs)
        return self.obs_buf, self.privileged_obs_buf, self.rew_buf, self.reset_buf, self.extras

---

2. 承上启下的桥梁：post_physics_step()

当 step 函数驱动底层的物理引擎完成指定的仿真步数后，就轮到 post_physics_step() 出场了。它起到了承上启下的作用，将物理世界的数据转化为强化学习所需的状态评估。

• 状态同步：从物理引擎底层显存中拉取并更新机器人的最新状态，比如根节点（躯干）的位置、速度、四肢关节的角度和角速度。

• 核心逻辑调度：它是一个“调度中心”，会依次调用检查终止条件、计算奖励、重置环境和计算下一步观测值的函数。这种模块化的设计使得代码非常易于维护和扩展。

    def post_physics_step(self):
        """ check terminations, compute observations and rewards
            calls self._post_physics_step_callback() for common computations 
            calls self._draw_debug_vis() if needed
        """
        self.gym.refresh_actor_root_state_tensor(self.sim)# 刷新角色根节点状态张量，获取最新的位姿和速度信息
        self.gym.refresh_net_contact_force_tensor(self.sim)# 刷新网络接触力张量

        self.episode_length_buf += 1# 记录每个环境存活的步数
        self.common_step_counter += 1# 记录总的仿真步数

        # prepare quantities
        self.base_quat[:] = self.root_states[:, 3:7]# 获取每个环境中机器人的基座四元数（旋转信息）
        self.base_lin_vel[:] = quat_rotate_inverse(self.base_quat, self.root_states[:, 7:10])# 将世界坐标系下的线速度旋转到机身局部坐标系下
        self.base_ang_vel[:] = quat_rotate_inverse(self.base_quat, self.root_states[:, 10:13])# 将世界坐标系下的角速度旋转到机身局部坐标系下
        self.projected_gravity[:] = quat_rotate_inverse(self.base_quat, self.gravity_vec)# 计算机身感知的重力向量（投影重力，常用于状态观测）

        self._post_physics_step_callback()

        # compute observations, rewards, resets, ...
        self.check_termination()
        self.compute_reward()
        env_ids = self.reset_buf.nonzero(as_tuple=False).flatten()
        self.reset_idx(env_ids)
        self.compute_observations() # in some cases a simulation step might be required to refresh some obs (for example body positions)

        self.last_actions[:] = self.actions[:]# 更新上一步的动作记录
        self.last_dof_vel[:] = self.dof_vel[:]# 更新上一步的关节速度记录
        self.last_root_vel[:] = self.root_states[:, 7:13]# 更新上一步的根节点速度记录

        if self.viewer and self.enable_viewer_sync and self.debug_viz:# 如果启用了调试可视化，则绘制调试信息
            self._draw_debug_vis()

    def _post_physics_step_callback(self):
        """ Callback called before computing terminations, rewards, and observations
            Default behaviour: Compute ang vel command based on target and heading, compute measured terrain heights and randomly push robots
        """
        # 每隔一定时间（resampling_time）重新采样一次命令，得到需要更新命令的环境 ID 列表
        env_ids = (self.episode_length_buf % int(self.cfg.commands.resampling_time / self.dt)==0).nonzero(as_tuple=False).flatten()
        self._resample_commands(env_ids)# 根据环境 ID 列表重新采样命令（线速度和角速度或航向），并更新命令张量
        if self.cfg.commands.heading_command:# 如果配置中使用了航向命令，则计算当前航向与目标航向之间的误差，并将其作为角速度命令的一部分
            forward = quat_apply(self.base_quat, self.forward_vec)
            heading = torch.atan2(forward[:, 1], forward[:, 0])
            self.commands[:, 2] = torch.clip(0.5*wrap_to_pi(self.commands[:, 3] - heading), -1., 1.)

        if self.cfg.terrain.measure_heights:# 如果配置中要求测量地形高度，则获取当前地形高度
            self.measured_heights = self._get_heights()
        if self.cfg.domain_rand.push_robots and  (self.common_step_counter % self.cfg.domain_rand.push_interval == 0):# 如果配置中要求随机推动机器人，则执行推动操作
            self._push_robots()

---

3. 生死判官：check_termination()

强化学习是一个不断试错的过程，而 check_termination() 负责判断当前这个回合 (Episode) 是否应该结束。

• 失败触发 (Termination)：检查机器人是否处于不可恢复的状态。最常见的是躯干触地（通过检查 Base 刚体与其他物体或地面的接触力）。一旦触发，当前环境的终止标志就会被置为 True。

• 超时截断 (Truncation)：除了摔倒，机器人可能在原地发呆或者走得很慢。为了提高训练效率，环境通常会设置一个最大步数（例如 20 秒）。如果达到了最大步数，即便机器人没有摔倒，也会强制结束回合。

• 注意区分：在高级的 RL 算法中，因“失败”导致的结束和因“超时”导致的结束，在计算价值函数（Value Function）的自举（Bootstrapping）时处理方式是不同的。

    def check_termination(self):
        """ Check if environments need to be reset
        """
        self.reset_buf = torch.any(torch.norm(self.contact_forces[:, self.termination_contact_indices, :], dim=-1) > 1., dim=1)# 如果接触力超过阈值，则标记环境需要重置
        self.time_out_buf = self.episode_length_buf > self.max_episode_length # 如果超出最大步数，则标记环境需要重置
        self.reset_buf |= self.time_out_buf # 将超时标记合并到重置标记中

---

4. 目标指引灯：compute_reward()

如果说动作是引擎，那么奖励就是方向盘。compute_reward() 直接决定了机器人最终会学会什么样的行为。

• 权重相加求和：该函数会遍历配置文件中所有设定的奖励/惩罚项（例如：前进速度奖励、存活奖励、动作平滑惩罚、能量消耗惩罚等），并根据设定的权重系数将它们相加，得到一个总的标量奖励。

• 负奖励截断：为了防止机器人因为惩罚过大而倾向于“立刻自杀”（快速结束回合以停止累积负奖励），有时会在计算总和后对奖励的下限进行裁剪（例如通过 torch.max 保证总奖励不小于 0）。

• 数据记录 (Logging)：这个函数通常还会负责记录每一项具体奖励的数值累加，方便后续输出到 TensorBoard 中，供我们分析哪一项惩罚过大导致了步态变形。

    def compute_reward(self):
        """ Compute rewards
            Calls each reward function which had a non-zero scale (processed in self._prepare_reward_function())
            adds each terms to the episode sums and to the total reward
        """
        self.rew_buf[:] = 0.# 初始化奖励缓冲区
        for i in range(len(self.reward_functions)):# 遍历所有奖励函数
            name = self.reward_names[i]# 获取奖励函数名称
            rew = self.reward_functions[i]() * self.reward_scales[name]# 计算当前奖励
            self.rew_buf += rew# 更新总奖励
            self.episode_sums[name] += rew# 更新当前回合的奖励总和
        if self.cfg.rewards.only_positive_rewards:# 仅保留正奖励
            self.rew_buf[:] = torch.clip(self.rew_buf[:], min=0.)# 将负奖励裁剪为0
        # add termination reward after clipping
        if "termination" in self.reward_scales:# 如果有结束奖励
            rew = self._reward_termination() * self.reward_scales["termination"]# 计算结束奖励
            self.rew_buf += rew# 更新总奖励
            self.episode_sums["termination"] += rew# 更新当前回合的结束奖励总和

---

5. 机器人的眼睛与感知：compute_observations()

感知世界是做出决策的前提。compute_observations() 负责将环境中乱七八糟的物理量，打包成神经网络可以理解的“一维数组”。

• 本体感受感知 (Proprioception)：收集机器人自身的内部状态，包括底座的线速度/角速度、重力向量在机器人坐标系下的投影、当前的指令目标、各关节的位置偏差和速度，以及上一帧输出的动作。

• 外部环境感知 (Exteroception)：如果启用了地形感知（如平移网格 Terrain），函数还会将机器人脚底周围的高度图数据加入观测向量中，让机器人“看”到前方的坑洼或台阶。

• 噪声注入：为了弥合仿真与现实的差距（Sim-to-Real），通常会在这个环节为观测数据加上一定的随机噪声（Noise），强制网络学习到更加鲁棒的策略。

    def compute_observations(self):
        """ Computes observations
        """
        self.obs_buf = torch.cat((  self.base_lin_vel * self.obs_scales.lin_vel,# 线速度
                                    self.base_ang_vel  * self.obs_scales.ang_vel,# 角速度
                                    self.projected_gravity,# 重力
                                    self.commands[:, :3] * self.commands_scale,# 线性命令
                                    (self.dof_pos - self.default_dof_pos) * self.obs_scales.dof_pos,# DOF位置
                                    self.dof_vel * self.obs_scales.dof_vel,# DOF速度
                                    self.actions
                                    ),dim=-1)# 动作
        # add perceptive inputs if not blind
        if self.cfg.terrain.measure_heights:# 如果需要测量高度
            heights = torch.clip(self.root_states[:, 2].unsqueeze(1) - 0.5 - self.measured_heights, -1, 1.) * self.obs_scales.height_measurements# 高度测量
            self.obs_buf = torch.cat((self.obs_buf, heights), dim=-1)# 添加高度测量
        # add noise if needed
        if self.add_noise:# 如果需要添加噪声
            self.obs_buf += (2 * torch.rand_like(self.obs_buf) - 1) * self.noise_scale_vec# 添加噪声

---

6. 并行世界的轮回：reset_idx(env_ids)

在单线程强化学习中，环境重置通常使用 reset()。但在 GPU 并行环境中，我们通常有成千上万个机器人同时在跑。reset_idx(env_ids) 的精髓在于它的参数 env_ids——它只重置那些挂掉的（或者超时的）环境，而不会打断其他表现良好的环境。

• 状态重置：将指定 ID 的环境中的机器人瞬移回起始位置，重置关节角度和速度为默认姿态。

• 重采样与课程更新：为重置的机器人重新生成随机的速度和朝向指令。同时，如果启用了地形课程学习，系统会根据该机器人在上一个回合的移动距离，决定将它传送到更难的地形（升级）还是更简单的地形（降级）。

• 清空历史记忆：清空与该环境相关的各项缓冲区（Buffers），如历史动作、历史观测等，确保新的回合是一个全新的开始。

    def reset_idx(self, env_ids):
        """ Reset some environments.
            Calls self._reset_dofs(env_ids), self._reset_root_states(env_ids), and self._resample_commands(env_ids)
            [Optional] calls self._update_terrain_curriculum(env_ids), self.update_command_curriculum(env_ids) and
            Logs episode info
            Resets some buffers

        Args:
            env_ids (list[int]): List of environment ids which must be reset
        """
        if len(env_ids) == 0:# 如果没有环境需要重置，则直接返回
            return
        # update curriculum
        if self.cfg.terrain.curriculum:# 如果启用了地形课程学习，则根据当前环境的表现更新地形难度等级
            self._update_terrain_curriculum(env_ids)
        # avoid updating command curriculum at each step since the maximum command is common to all envs
        if self.cfg.commands.curriculum and (self.common_step_counter % self.max_episode_length==0):# 如果启用了命令课程学习，则根据当前环境的表现更新命令难度等级
            self.update_command_curriculum(env_ids)
        
        # reset robot states
        self._reset_dofs(env_ids)
        self._reset_root_states(env_ids)

        self._resample_commands(env_ids)

        # reset buffers
        self.last_actions[env_ids] = 0.# 将上一步动作缓冲区中需要重置的环境对应的动作值设为0
        self.last_dof_vel[env_ids] = 0.# 将上一步关节速度缓冲区中需要重置的环境对应的速度值设为0
        self.feet_air_time[env_ids] = 0.# 将脚部空中时间缓冲区中需要重置的环境对应的时间值设为0
        self.episode_length_buf[env_ids] = 0# 将回合长度缓冲区中需要重置的环境对应的长度值设为0
        self.reset_buf[env_ids] = 1# 将重置缓冲区中需要重置的环境对应的标记值设为1
        # fill extras
        self.extras["episode"] = {}# 在额外信息字典中添加一个 "episode" 键，用于存储当前回合的统计信息
        for key in self.episode_sums.keys():# 遍历所有回合奖励统计信息
            self.extras["episode"]['rew_' + key] = torch.mean(self.episode_sums[key][env_ids]) / self.max_episode_length_s # 计算当前回合的平均奖励
            self.episode_sums[key][env_ids] = 0. # 重置当前回合的奖励统计信息
        # log additional curriculum info
        if self.cfg.terrain.curriculum:# 如果启用了地形课程学习，则在额外信息字典中记录当前回合的平均地形难度等级
            self.extras["episode"]["terrain_level"] = torch.mean(self.terrain_levels.float())
        if self.cfg.commands.curriculum:# 如果启用了命令课程学习，则在额外信息字典中记录当前回合的最大命令值
            self.extras["episode"]["max_command_x"] = self.command_ranges["lin_vel_x"][1]
        # send timeout info to the algorithm
        if self.cfg.env.send_timeouts:# 如果启用了超时发送，则在额外信息字典中记录当前回合的超时信息
            self.extras["time_outs"] = self.time_out_buf

    def _update_terrain_curriculum(self, env_ids):
        """ Implements the game-inspired curriculum.

        Args:
            env_ids (List[int]): ids of environments being reset
        """
        # Implement Terrain curriculum
        if not self.init_done:# 如果初始化尚未完成，则不进行地形课程更新
            # don't change on initial reset
            return
        distance = torch.norm(self.root_states[env_ids, :2] - self.env_origins[env_ids, :2], dim=1) # 计算机器人当前位置与环境原点之间的距离
        # robots that walked far enough progress to harder terains
        move_up = distance > self.terrain.env_length / 2# 如果机器人走过的距离超过环境长度的一半，则标记为需要提升地形难度
        # robots that walked less than half of their required distance go to simpler terrains
        move_down = (distance < torch.norm(self.commands[env_ids, :2], dim=1)*self.max_episode_length_s*0.5) * ~move_up# 如果机器人走过的距离少于其所需距离的一半，则标记为需要降低地形难度
        self.terrain_levels[env_ids] += 1 * move_up - 1 * move_down# 根据需要提升或降低地形难度的标记，更新环境的地形难度等级（增加1或减少1）
        # Robots that solve the last level are sent to a random one
        self.terrain_levels[env_ids] = torch.where(self.terrain_levels[env_ids]>=self.max_terrain_level,# 如果地形难度等级超过最大值，则随机分配一个新的地形难度等级
                                                   torch.randint_like(self.terrain_levels[env_ids], self.max_terrain_level),# 随机生成一个新的地形难度等级
                                                   torch.clip(self.terrain_levels[env_ids], 0)) # (the minumum level is zero)
        self.env_origins[env_ids] = self.terrain_origins[self.terrain_levels[env_ids], self.terrain_types[env_ids]]# 更新环境原点位置

    def update_command_curriculum(self, env_ids):
        """ Implements a curriculum of increasing commands

        Args:
            env_ids (List[int]): ids of environments being reset
        """
        # If the tracking reward is above 80% of the maximum, increase the range of commands
        if torch.mean(self.episode_sums["tracking_lin_vel"][env_ids]) / self.max_episode_length > 0.8 * self.reward_scales["tracking_lin_vel"]:# 如果线速度跟踪奖励的平均值超过最大值的80%，则增加命令范围
            self.command_ranges["lin_vel_x"][0] = np.clip(self.command_ranges["lin_vel_x"][0] - 0.5, -self.cfg.commands.max_curriculum, 0.) # 将线速度命令范围的下限减少0.5
            self.command_ranges["lin_vel_x"][1] = np.clip(self.command_ranges["lin_vel_x"][1] + 0.5, 0., self.cfg.commands.max_curriculum) # 将线速度命令范围的上限增加0.5

---

总结

这六个函数共同编织了强化学习中的马尔可夫链条。step 接收决策并推进时间，post_physics_step 承接物理反馈，compute_observations 观察现状，compute_reward 评估好坏，check_termination 和 reset_idx 负责结束与重生。理解了这个循环，你就真正掌握了 RL 环境代码的命脉。