企业官网建设流程全解析

IsaacLab Direct Workflow 强化学习环境开发实战：以 Cartpole 直接工作流任务为例

【免费下载链接】IsaacLabUnified framework for robot learning with multi-physics/renderer support项目地址: https://gitcode.com/GitHub_Trending/is/IsaacLab

本篇技术指南聚焦 IsaacLab 中的Direct Workflow（直接工作流）强化学习环境开发方式，以仓库自带的isaaclab_tasks.direct.cartpole倒立摆平衡任务为完整示例，讲解如何通过继承DirectRLEnv基类、直接手写场景创建、动作应用、重置、奖励与观测函数，快速构建一个可训练的 RL 环境。读完本文，你将掌握 DirectRLEnv 的六个核心 API 的职责与调用时机、任务配置类的组织方式，并能在本地启动端到端训练，同时理解域随机化与动作/观测噪声在直接工作流中的配置方法。

为什么需要 Direct Workflow？

在 IsaacLab 中，除了使用配置类驱动的模块化环境 ManagerBasedRLEnv 之外，还提供了isaaclab.envs.DirectRLEnv类，允许开发者在环境脚本中获得更直接的编程控制：

• 奖励与观测直接写在任务脚本中：不再通过 RewardManager、ObservationManager 等管理器类进行配置化组装，而是把完整的奖励函数、观测函数直接实现为环境类的方法；
• 更低的抽象层级：无需在多个配置文件之间跳转查找逻辑，任务代码的所有关键部分集中在一个文件中，易于定位和理解；
• 更灵活的实现手段：可以直接利用 PyTorch JIT 等特性对奖励函数做脚本化编译，获得更好的数值与执行效率。

从源码角度看，DirectRLEnv 继承自gymnasium.Env，是一个向量化环境（is_vector_env = True），但刻意不继承gym.vector.VectorEnv，以避免引入等待/异步等本框架不需要的机制，转而由各 RL 库自己的 wrapper 来包装该环境。基类通过抽象方法强制要求子类实现五个核心 API（见 direct_rl_env.py）：

任务配置类：CartpoleEnvCfg

与 manager 工作流类似，直接工作流同样使用@configclass定义一个任务配置类，但基类换成了envs.DirectRLEnvCfg。由于直接工作流不使用 Action/Observation Manager，配置类必须显式声明环境的动作维度与观测维度：

@configclass class CartpoleEnvCfg(DirectRLEnvCfg): ... action_space = 1 observation_space = 4 state_space = 0

其中state_space用于非对称 Actor-Critic 架构（critic 专用状态），本任务不使用因此置 0。配置类也可以自由定义任务特定属性，例如奖励项缩放系数与重置条件阈值（完整定义见 cartpole_env_cfg.py）：

@configclass class CartpoleEnvCfg(DirectRLEnvCfg): # env decimation = 2 episode_length_s = 5.0 action_scale = 100.0 # [N] action_space = 1 observation_space = 4 state_space = 0 # simulation sim: SimulationCfg = SimulationCfg(dt=1 / 120, render_interval=decimation, physics=CartpolePhysicsCfg()) # robot robot_cfg: ArticulationCfg = CARTPOLE_CFG.replace(prim_path="/World/envs/env_.*/Robot") cart_dof_name = "slider_to_cart" pole_dof_name = "cart_to_pole" # scene scene: InteractiveSceneCfg = InteractiveSceneCfg( num_envs=4096, env_spacing=4.0, replicate_physics=True, clone_in_fabric=True ) # reset max_cart_pos = 3.0 # the cart is reset if it exceeds that position [m] initial_pole_angle_range = [-0.25, 0.25] # the range in which the pole angle is sampled from on reset [rad] # reward scales rew_scale_alive = 1.0 rew_scale_terminated = -2.0 rew_scale_pole_pos = -1.0 rew_scale_cart_vel = -0.01 rew_scale_pole_vel = -0.005

这些参数的含义与影响：

• decimation = 2：每个环境步（policy 步）内执行 2 次物理仿真步。DirectRLEnvCfg.decimation定义为"每个 policy dt 内、以 sim dt 执行的控制更新次数"，环境步长由sim.dt * decimation计算得出（见 direct_rl_env_cfg.py）。本任务sim.dt = 1/120，因此环境步长约为 16.7 ms；
• episode_length_s = 5.0：回合时长。max_episode_length = ceil(episode_length_s / (sim.dt * decimation))，即5.0 / (1/120 * 2) = 300个环境步（见 direct_rl_env.py）；
• action_scale = 100.0：作用于_pre_physics_step，把策略输出（归一化范围）缩放到小车施加的力（牛顿）；
• initial_pole_angle_range = [-0.25, 0.25]：重置时杆的初始倾角在±0.25π弧度区间内均匀采样（注意_reset_idx中会乘以math.pi）；
• max_cart_pos = 3.0：小车位移超过该阈值（米）即触发回合终止。

此外，配置类基类 DirectRLEnvCfg 还提供了seed、is_finite_horizon、observation_space/action_space（支持 Gymnasium 空间或 Python 原生类型）、observation_noise_model、action_noise_model、events、num_rerenders_on_reset、video_recorder等通用字段，为环境行为提供统一入口。

环境类骨架：继承 DirectRLEnv

新建环境时，需要定义一个继承DirectRLEnv的子类，并在__init__中通过super().__init__完成基类初始化（包括 SimulationContext 创建、场景搭建、事件管理器初始化、Gym 空间配置等）：

class CartpoleEnv(DirectRLEnv): cfg: CartpoleEnvCfg def __init__(self, cfg: CartpoleEnvCfg, render_mode: str | None = None, **kwargs): super().__init__(cfg, render_mode, **kwargs) self._cart_dof_idx, _ = self.cartpole.find_joints(self.cfg.cart_dof_name) self._pole_dof_idx, _ = self.cartpole.find_joints(self.cfg.pole_dof_name) self.action_scale = self.cfg.action_scale self.joint_pos = self.cartpole.data.joint_pos.torch self.joint_vel = self.cartpole.data.joint_vel.torch

初始化阶段解析出小车与杆对应的关节索引（find_joints），缓存关节位置/速度张量引用，供后续各 API 直接使用。类变量（如self.actions、self.joint_pos）在整个类中共享，可在动作应用、重置、奖励、观测等所有函数中访问。

场景创建：_setup_scene

与 manager 工作流由框架代为完成场景创建不同，直接工作流把场景搭建的灵活性完全交给开发者。_setup_scene中通常需要完成：创建 Actor 资产、克隆环境、过滤环境间碰撞、把 Actor 注册进场景，以及补充地面与灯光等附加元素：

def _setup_scene(self): self.cartpole = Articulation(self.cfg.robot_cfg) # add ground plane spawn_ground_plane(prim_path="/World/ground", cfg=GroundPlaneCfg()) # clone and replicate self.scene.clone_environments(copy_from_source=False) # we need to explicitly filter collisions for CPU simulation if self.device == "cpu": self.scene.filter_collisions(global_prim_paths=[]) # add articulation to scene self.scene.articulations["cartpole"] = self.cartpole # add lights light_cfg = sim_utils.DomeLightCfg(intensity=2000.0, color=(0.75, 0.75, 0.75)) light_cfg.func("/World/Light", light_cfg)

关键点说明：

• 使用Articulation(self.cfg.robot_cfg)实例化机器人资产，其prim_path已通过CARTPOLE_CFG.replace(prim_path="/World/envs/env_.*/Robot")配置为可被批量克隆的模式；
• self.scene.clone_environments(copy_from_source=False)负责把单个环境实例克隆为num_envs（4096）个并行环境；
• CPU 仿真下需显式调用self.scene.filter_collisions(global_prim_paths=[])过滤环境之间的碰撞；
• 最后把创建的 Articulation 注册进self.scene.articulations字典（key 为"cartpole"），并添加穹顶灯光供渲染。

_setup_scene会在基类_init_sim中创建InteractiveScene(self.cfg.scene)之后被调用（见 direct_rl_env.py）。

定义奖励：_get_rewards 与 torch.jit

奖励函数在_get_rewards(self)API 中实现，并以返回值为奖励缓冲。本示例中，奖励计算被抽取为一个带@torch.jit.script装饰器的独立函数，用于对数值计算做脚本化编译：

def _get_rewards(self) -> torch.Tensor: total_reward = compute_rewards( self.cfg.rew_scale_alive, self.cfg.rew_scale_terminated, self.cfg.rew_scale_pole_pos, self.cfg.rew_scale_cart_vel, self.cfg.rew_scale_pole_vel, self.joint_pos[:, self._pole_dof_idx[0]], self.joint_vel[:, self._pole_dof_idx[0]], self.joint_pos[:, self._cart_dof_idx[0]], self.joint_vel[:, self._cart_dof_idx[0]], self.reset_terminated, ) return total_reward @torch.jit.script def compute_rewards( rew_scale_alive: float, rew_scale_terminated: float, rew_scale_pole_pos: float, rew_scale_cart_vel: float, rew_scale_pole_vel: float, pole_pos: torch.Tensor, pole_vel: torch.Tensor, cart_pos: torch.Tensor, cart_vel: torch.Tensor, reset_terminated: torch.Tensor, ): rew_alive = rew_scale_alive * (1.0 - reset_terminated.float()) rew_termination = rew_scale_terminated * reset_terminated.float() rew_pole_pos = rew_scale_pole_pos * torch.sum(torch.square(pole_pos), dim=-1) rew_cart_vel = rew_scale_cart_vel * torch.sum(torch.abs(cart_vel), dim=-1) rew_pole_vel = rew_scale_pole_vel * torch.sum(torch.abs(pole_vel), dim=-1) total_reward = rew_alive + rew_termination + rew_pole_pos + rew_cart_vel + rew_pole_vel return total_reward

奖励设计解读（逐项与配置中的缩放系数对应）：

• 存活奖励rew_alive：每个未终止环境每步获得+1.0，鼓励策略让杆保持平衡更久；
• 终止惩罚rew_termination：环境终止（杆倒下或小车越界）时施加-2.0；
• 杆位置惩罚rew_pole_pos：-1.0 × pole_pos²，杆偏离竖直方向越远惩罚越大，是该任务的主要学习信号；
• 小车速度惩罚rew_cart_vel：-0.01 × |cart_vel|，抑制小车来回高速滑动；
• 杆角速度惩罚rew_pole_vel：-0.005 × |pole_vel|，抑制杆的摆动。

完整实现见 cartpole_env.py。在基类step()中，_get_rewards()在物理步进完成后被调用一次，产出形状为(num_envs,)的奖励缓冲（见 direct_rl_env.py）。

定义观测：_get_observations

观测缓冲在_get_observations(self)中构造。该 API 结束时必须返回一个字典：以policy为 key、完整观测缓冲为 value；对于非对称策略，还需包含critickey 与状态缓冲：

def _get_observations(self) -> dict: obs = torch.cat( ( self.joint_pos[:, self._pole_dof_idx[0]].unsqueeze(dim=1), self.joint_vel[:, self._pole_dof_idx[0]].unsqueeze(dim=1), self.joint_pos[:, self._cart_dof_idx[0]].unsqueeze(dim=1), self.joint_vel[:, self._cart_dof_idx[0]].unsqueeze(dim=1), ), dim=-1, ) observations = {"policy": obs} return observations

观测向量由 4 个标量拼接而成：杆的角度、杆的角速度、小车位置、小车速度（与配置中的observation_space = 4一致）。在基类step()末尾，self.obs_buf = self._get_observations()会被执行，若配置了观测噪声模型，则对obs_buf["policy"]施加噪声；state 空间不施加噪声（见 direct_rl_env.py）。

计算 done 与执行重置：_get_dones 与 _reset_idx

终止判定 _get_dones

_get_dones(self)负责填充 done 缓冲，判断哪些环境需要重置、哪些环境到达回合长度上限，并以(terminated, time_out)布尔张量元组形式返回：

def _get_dones(self) -> tuple[torch.Tensor, torch.Tensor]: self.joint_pos = self.cartpole.data.joint_pos.torch self.joint_vel = self.cartpole.data.joint_vel.torch time_out = self.episode_length_buf >= self.max_episode_length - 1 out_of_bounds = torch.any(torch.abs(self.joint_pos[:, self._cart_dof_idx]) > self.cfg.max_cart_pos, dim=1) out_of_bounds = out_of_bounds | torch.any(torch.abs(self.joint_pos[:, self._pole_dof_idx]) > math.pi / 2, dim=1) return out_of_bounds, time_out

判定逻辑包含两类终止条件：

• 越界终止（terminated）：小车位移绝对值超过max_cart_pos（3.0 m），或杆倾角绝对值超过π/2（杆倒下）；
• 超时截断（time_out）：episode_length_buf达到max_episode_length - 1，即回合步数上限。

在基类step()中，这两个布尔张量被写入self.reset_terminated与self.reset_time_outs，并合并为reset_buf，用于筛选需要重置的环境索引（见 direct_rl_env.py）。

状态重置 _reset_idx

_reset_idx(self, env_ids)对指定环境执行重置，直接将新状态写入仿真：

def _reset_idx(self, env_ids: Sequence[int] | None): if env_ids is None: env_ids = self.cartpole._ALL_INDICES # Log survival success rate before resetting survived = self.reset_time_outs[env_ids].float() self.extras.setdefault("log", {})["Metrics/success_rate"] = survived.mean().item() super()._reset_idx(env_ids) joint_pos = self.cartpole.data.default_joint_pos.torch[env_ids].clone() joint_pos[:, self._pole_dof_idx] += sample_uniform( self.cfg.initial_pole_angle_range[0] * math.pi, self.cfg.initial_pole_angle_range[1] * math.pi, joint_pos[:, self._pole_dof_idx].shape, joint_pos.device, ) joint_vel = self.cartpole.data.default_joint_vel.torch[env_ids].clone() default_root_pose = self.cartpole.data.default_root_pose.torch[env_ids].clone() default_root_pose[:, :3] += self.scene.env_origins[env_ids] default_root_vel = self.cartpole.data.default_root_vel.torch[env_ids].clone() self.joint_pos[env_ids] = joint_pos self.joint_vel[env_ids] = joint_vel self.cartpole.write_root_pose_to_sim_index(root_pose=default_root_pose, env_ids=env_ids) self.cartpole.write_root_velocity_to_sim_index(root_velocity=default_root_vel, env_ids=env_ids) self.cartpole.write_joint_position_to_sim_index(position=joint_pos, env_ids=env_ids) self.cartpole.write_joint_velocity_to_sim_index(position=joint_vel, env_ids=env_ids)

重置流程要点：

• 若env_ids为None（例如首次reset()全量重置），回退到全部环境索引；
• 重置前用reset_time_outs计算"存活成功率"（以超时结束视为存活）并写入extras["log"]，便于训练日志记录指标；
• 调用super()._reset_idx(env_ids)触发场景级重置与 reset 模式事件；
• 从default_joint_pos恢复默认关节位置，并在杆关节上叠加sample_uniform采样的初始倾角扰动（范围[-0.25π, 0.25π]）；
• 根位姿叠加scene.env_origins偏移以匹配各克隆环境的原点；
• 通过write_*_to_sim_index系列 API 把位姿、速度、关节状态直接写回物理仿真。

应用动作：_pre_physics_step 与 _apply_action

直接工作流提供两个分工明确的动作 API：

_pre_physics_step(self, actions)：每个 RL 步只调用一次，位于物理步进之前，接收策略输出的动作张量，用于预处理并把数据缓存在类变量中：

def _pre_physics_step(self, actions: torch.Tensor) -> None: self.actions = self.action_scale * actions.clone()

这里将归一化的策略输出乘以action_scale（100.0 N）得到实际施加的力。

_apply_action(self)：每个 RL 步内被调用decimation（此处为 2）次，每次物理步进前调用一次，适用于需要逐物理步应用动作的场景：

def _apply_action(self) -> None: self.cartpole.set_joint_effort_target_index(target=self.actions, joint_ids=self._cart_dof_idx)

该 API 通过关节索引把力目标写入小车关节（slider_to_cart）。从基类step()的物理循环可以看到二者的调用时机（见 direct_rl_env.py）：_pre_physics_step先执行一次，随后进入for _ in range(self.cfg.decimation)循环，在每次sim.step()前调用_apply_action()，实现"策略低频决策、物理高频执行"的标准 RL 采样节奏。

启动训练：代码执行

使用以下命令运行直接工作流 Cartpole 的 RL 训练（以 rl_games 库为例）：

./isaaclab.sh train --rl_library rl_games --task=Isaac-Cartpole-Direct-v0

几点说明：

• 所有直接工作流任务在任务名中都带有-Direct后缀，用于与同一任务的 manager 工作流实现相区分（例如Isaac-Cartpole-v0对应 manager 工作流版本）；
• Isaac-Cartpole-Direct-v0通过gymnasium.register注册，入口点为cartpole_env.py:CartpoleEnv，配置入口为cartpole_env_cfg.py:CartpoleEnvCfg，同时为 rl_games、rsl_rl、skrl、sb3 分别绑定了训练配置（见init.py 及同目录下的 agents 子目录）；
• 训练配置文件位于 agents/rl_games_ppo_cfg.yaml 等路径下；更多训练参数与多库切换方式可参考 configuring_rl_training.rst 与 run_rl_training.rst。

域随机化（Domain Randomization）

直接工作流的域随机化同样通过isaaclab.utils.configclass定义配置类，由若干managers.EventTermCfg（即EventTerm）变量组成：

@configclass class EventCfg: robot_physics_material = EventTerm( func=mdp.randomize_rigid_body_material, mode="reset", params={ "asset_cfg": SceneEntityCfg("robot", body_names=".*"), "static_friction_range": (0.7, 1.3), "dynamic_friction_range": (1.0, 1.0), "restitution_range": (1.0, 1.0), "num_buckets": 250, }, ) robot_joint_stiffness_and_damping = EventTerm( func=mdp.randomize_actuator_gains, mode="reset", params={ "asset_cfg": SceneEntityCfg("robot", joint_names=".*"), "stiffness_distribution_params": (0.75, 1.5), "damping_distribution_params": (0.3, 3.0), "operation": "scale", "distribution": "log_uniform", }, ) reset_gravity = EventTerm( func=mdp.randomize_physics_scene_gravity, mode="interval", is_global_time=True, interval_range_s=(36.0, 36.0), # time_s = num_steps * (decimation * dt) params={ "gravity_distribution_params": ([0.0, 0.0, 0.0], [0.0, 0.0, 0.4]), "operation": "add", "distribution": "gaussian", }, )

每个EventTerm对象属于managers.EventTermCfg类，其关键字段：

• func：随机化时调用的函数。可在envs.mdp.events模块（见 events.py）中找到全部可用事件函数，本示例用到mdp.randomize_rigid_body_material（刚体物理材质）、mdp.randomize_actuator_gains（执行器刚度/阻尼增益）、mdp.randomize_physics_scene_gravity（物理场景重力）；
• mode：触发时机，可取startup（仿真启动时）、reset（环境重置时）、interval（按固定时间间隔）；
• params：传递给func的参数字典，包含随机化分布与作用对象。

其中"asset_cfg": SceneEntityCfg("robot", body_names=".*")参数通过正则表达式指定作用对象："robot"为场景中 Actor 的名称，body_names/joint_names为正则表达式，匹配到的刚体或关节将被施加随机化。各事件的设计意图：

• 物理材质随机化：静摩擦系数在(0.7, 1.3)区间随机，模拟不同接触表面；
• 执行器增益随机化：以log_uniform分布、scale操作缩放刚度（0.75~1.5）与阻尼（0.3~3.0），增强对执行器不确定性的鲁棒性；
• 重力随机化：以interval模式每 36 秒全局触发一次，向重力矢量叠加高斯噪声（operation="add"），interval_range_s的时间换算关系为time_s = num_steps * (decimation * dt)。

配置完成后，把该configclass挂载到任务基础配置类的events变量上即可生效：

@configclass class MyTaskConfig: events: EventCfg = EventCfg()

基类在初始化时会创建EventManager(self.cfg.events, self)，并按模式自动应用prestartup、startup事件；reset模式事件在_reset_idx中应用，interval模式事件在每次step()后以step_dt为间隔应用（见 direct_rl_env.py）。

动作与观测噪声

动作与观测噪声同样通过utils.configclass模块配置，挂载到任务配置类的action_noise_model与observation_noise_model变量上：

@configclass class MyTaskConfig: # at every time-step add gaussian noise + bias. The bias is a gaussian sampled at reset action_noise_model: NoiseModelWithAdditiveBiasCfg = NoiseModelWithAdditiveBiasCfg( noise_cfg=GaussianNoiseCfg(mean=0.0, std=0.05, operation="add"), bias_noise_cfg=GaussianNoiseCfg(mean=0.0, std=0.015, operation="abs"), ) # at every time-step add gaussian noise + bias. The bias is a gaussian sampled at reset observation_noise_model: NoiseModelWithAdditiveBiasCfg = NoiseModelWithAdditiveBiasCfg( noise_cfg=GaussianNoiseCfg(mean=0.0, std=0.002, operation="add"), bias_noise_cfg=GaussianNoiseCfg(mean=0.0, std=0.0001, operation="abs"), )

utils.noise.NoiseModelWithAdditiveBiasCfg（定义见 noise_cfg.py）可同时建模两种噪声：

• noise_cfg：指定每个步长对所有环境采样的高斯分布（mean、std、operation="add"），该噪声每一步都会被加到对应的动作/观测缓冲上，属于不相关（逐步独立）噪声；
• bias_noise_cfg：指定相关性噪声的高斯分布，该偏置在环境重置时采样一次（operation="abs"表示取其绝对值），随后在整个回合内保持不变，直到下一次重置才重新采样，用于模拟传感器/执行器的系统性偏差。

基类中对应的应用逻辑为：step()入口处action = self._action_noise_model(action)，观测则在计算完成后通过self.obs_buf["policy"] = self._observation_noise_model(self.obs_buf["policy"])施加；两个噪声模型都会在_reset_idx中对被重置的环境调用reset(env_ids)重新采样（见 direct_rl_env.py）。

如果只需要逐步噪声，可直接使用utils.noise.GaussianNoiseCfg（定义见 noise_cfg.py）指定一个加性高斯分布：

@configclass class MyTaskConfig: action_noise_model: GaussianNoiseCfg = GaussianNoiseCfg(mean=0.0, std=0.05, operation="add")

小结与下一步

本教程展示了如何通过直接工作流创建强化学习任务环境：继承DirectRLEnv，在_setup_scene中搭建场景，并实现_pre_physics_step/_apply_action（动作）、_get_dones/_reset_idx（终止与重置）、_get_rewards（奖励）、_get_observations（观测）等全部核心 API，最终通过./isaaclab.sh train --rl_library rl_games --task=Isaac-Cartpole-Direct-v0启动训练。

虽然可以手动实例化DirectRLEnv子类，但为每个任务编写专用脚本不具备可扩展性，因此 IsaacLab 借助gymnasium.make以 Gym 接口创建环境——这正是下一篇教程 register_rl_env_gym.rst 的主题。此外，如果你希望修改现有直接工作流任务的行为（例如调整奖励系数或重置逻辑），可以参考 modify_direct_rl_env.rst；若想对比 manager 工作流的写法差异，可阅读 create_manager_rl_env.rst。

【免费下载链接】IsaacLabUnified framework for robot learning with multi-physics/renderer support项目地址: https://gitcode.com/GitHub_Trending/is/IsaacLab