)
★ FEATURED ARTICLE ILImitation LearningEAIEmbodied Artificial IntelligenceGymnasium 是一个 强化学习Reinforcement Learning, RL的标准环境接口库。它是 OpenAI Gym 的社区维护和升级版本由 Farama Foundation 主导开发广泛用于 RL 研究、教学和工程实践中。一、Gymnasium 简介定位通用强化学习环境库。作用提供标准化的 env.reset()、env.step(action)、env.render() 等接口使算法开发者可以无缝切换不同任务环境如 CartPole、Atari、MuJoCo 等。特点完全兼容 OpenAI Gym 的 API支持类型提示、错误检查活跃社区维护持续更新支持自定义环境注册与主流 RL 框架如 Stable-Baselines3、RLlib、CleanRL无缝集成。✅ 简单说Gymnasium OpenAI Gym 的“官方继承者”但 不是数据库而是 Python 软件库。二、Gymnasium 与 Isaac Gym / Isaac Sim 的对比它们都属于强化学习仿真工具生态但定位、性能和用途有显著区别特性GymnasiumIsaac GymIsaac Sim开发者社区Farama FoundationNVIDIANVIDIA底层物理引擎多种Box2D、MuJoCo 等PhysXGPU 加速PhysX Omniverse高保真运行设备CPU普通电脑即可GPU 必需NVIDIAGPU 必需高端显卡主要用途通用 RL 任务CartPole、Atari、简单机器人大规模并行机器人 RL 训练高保真机器人仿真 AI训练 传感器建模API 风格标准 Gym 接口兼容 Gym 风格但为向量化环境基于 Omniverse通过 Isaac Lab 提供 RL 接口是否仍在活跃开发✅ 是⚠️ 已停止更新被 Isaac Sim/Isaac Lab 取代✅ 是主力产品三、Gymnasium优点和不足Isaac Gym 曾是 Gymnasium 的“高性能替代品” 它模仿 Gym 的 API但专为 GPU 并行仿真设计适合训练成百上千个机器人实例。 例如用 Isaac Gym 同时训练 10,000 个机械臂速度比 MuJoCo Gym 快数百倍。 Isaac Sim 是 Isaac Gym 的“继任者” Isaac Sim 基于 NVIDIA Omniverse支持更真实的渲染、传感器摄像头、LiDAR、多机器人协作等。 强化学习功能现在由 Isaac Lab集成在 Isaac Sim 中提供取代了 Isaac Gym。 官方已逐步将 Isaac Gym 的能力迁移到 Isaac Sim Isaac Lab 架构中。 Gymnasium 与 Isaac 系列无直接依赖 Gymnasium 是通用 RL 接口标准 Isaac Gym/Sim 是具体仿真平台 你可以用 Gymnasium 接口包装 Isaac Gym 环境社区有尝试但官方不直接耦合。四、适用场景初学者 / 通用 RL 任务 → 用 Gymnasium 高性能机器人 RL旧项目 → 用 Isaac Gym仅限 Linux NVIDIA GPU 前沿机器人仿真 多模态感知 部署闭环 → 用 Isaac Sim Isaac Lab五、使用举例环境准备Mujoco Gymnasium资源准备动作采集文件csv 机器人模型xml[A]、构建训练环境# motion_trans.pyfrom dance_envimportDanceEnv from stable_baselines3importPPO from stable_baselines3.common.monitorimportMonitorimportosimportloggingimporttime# 固定配置 MODEL_SAVE_DIR./trained_robot_modelLOG_DIR./training_logsos.makedirs(MODEL_SAVE_DIR,exist_okTrue)os.makedirs(LOG_DIR,exist_okTrue)logging.basicConfig(levellogging.INFO,format%(asctime)s - %(levelname)s - %(message)s,handlers[logging.FileHandler(os.path.join(LOG_DIR,training.log)), logging.StreamHandler()])loggerlogging.getLogger(__name__)def create_robot_env():envDanceEnv(xml_pathrobot.xml,ref_csv_pathWalk.csv,max_episode_steps500,render_modehuman,domain_randomizationFalse)envMonitor(env, LOG_DIR)returnenv# 早停回调函数 from stable_baselines3.common.callbacksimportBaseCallback class EarlyStopCallback(BaseCallback): def __init__(self, target_reward,verbose0): super().__init__(verbose)self.target_rewardtarget_reward def _on_step(self):ifself.locals[rewards].mean()self.target_reward: self.logger.info(f奖励达到 {self.target_reward}当前平均奖励{self.locals[rewards].mean():.3f}提前终止训练)returnFalsereturnTrue def main(): try: logger.info(开始创建机器人环境...)envcreate_robot_env()obs, infoenv.reset(seed42)logger.info(f环境创建成功观测空间形状: {obs.shape})logger.info(f动作空间维度: {env.action_space.shape})logger.info(初始化PPO神经网络模型...)# 临时调小训练步数快速观察可视化效果TRAIN_TIMESTEPS50000# 从1000000改为10000后续可按需调大modelPPO(MlpPolicy, env,learning_rate3e-4,n_steps2048,batch_size64,gamma0.99,verbose1,tensorboard_logLOG_DIR,devicecpu,policy_kwargsdict(net_archdict(pi[512,512],vf[512,512])))logger.info(PPO模型初始化成功开始训练...)callbackEarlyStopCallback(target_reward-100.0,verbose1)model.learn(total_timestepsTRAIN_TIMESTEPS,callbackcallback,tb_log_namerobot_walk_ppo,progress_barTrue# 显示训练进度条)# 模型保存model_save_pathos.path.join(MODEL_SAVE_DIR,robot_walk_ppo_model.zip)model.save(model_save_path)logger.info(f训练完成模型已保存至: {model_save_path})# 模型验证添加渲染延迟logger.info(开始验证训练好的模型...)obs, infoenv.reset(seed42)total_reward0.0episode_count0forstepinrange(500): action, _statesmodel.predict(obs,deterministicTrue)obs, reward, terminated, truncated, infoenv.step(action)total_rewardreward time.sleep(0.05)ifstep %1000: logger.info(f验证步骤 {step}, 即时奖励: {reward:.3f}, 累计奖励: {total_reward:.3f})ifterminated or truncated: logger.info(f验证终止步骤: {step}, 原因: {达到最大步数 if terminated else 机器人倒地})obs, infoenv.reset()total_reward0.0episode_count1time.sleep(0.5)logger.info(模型验证完成)except Exception as e: logger.error(f训练过程发生错误错误信息: {str(e)},exc_infoTrue)raise e finally: try: env.close()logger.info(环境已安全关闭)except: passif__name____main__:main()[B]、数据载录文件# dance_env.pyimportgymnasium as gym from gymnasiumimportspacesimportnumpy as npimportpandas as pdimportmujoco class DanceEnv(gym.Env): metadata{render_modes:[human],render_fps:50}def __init__(self, xml_path: str, ref_csv_path: str, max_episode_steps: int1000, render_mode: strNone, domain_randomization: boolFalse,): super().__init__()self.xml_pathxml_path raw_trajpd.read_csv(ref_csv_path).values.astype(np.float32)self.ref_pelvisraw_traj[:, :7]# [T, 7] 骨盆位姿self.ref_joints_rawraw_traj[:,7:]# [T, 28] 原始CSV关节角度csv_joint_names[# 左侧下肢严格匹配MuJoCo actuator顺序yaw→roll→pitch→knee→ankle pitch→ankle rolldof_left_hip_yaw_link,dof_left_hip_roll_link,dof_left_hip_pitch_link,dof_left_knee_link,dof_left_ankle_pitch_link,dof_left_ankle_roll_link,# 右侧下肢严格匹配MuJoCo actuator顺序yaw→roll→pitch→knee→ankle pitch→ankle rolldof_right_hip_yaw_link,dof_right_hip_roll_link,dof_right_hip_pitch_link,dof_right_knee_link,dof_right_ankle_pitch_link,dof_right_ankle_roll_link,# 左侧上肢严格匹配MuJoCo actuator顺序pitch→roll→yaw→elbow→wrist roll→wrist pitch→wrist yawdof_left_shoulder_pitch_link,dof_left_shoulder_roll_link,dof_left_shoulder_yaw_link,dof_left_elbow_link,dof_left_wrist_roll_link,dof_left_wrist_pitch_link,dof_left_wrist_yaw_link,# 右侧上肢严格匹配MuJoCo actuator顺序pitch→roll→yaw→elbow→wrist roll→wrist pitch→wrist yawdof_right_shoulder_pitch_link,dof_right_shoulder_roll_link,dof_right_shoulder_yaw_link,dof_right_elbow_link,dof_right_wrist_roll_link,dof_right_wrist_pitch_link,dof_right_wrist_yaw_link,# 躯干严格匹配MuJoCo actuator顺序roll→yawdof_waist_roll_link,dof_waist_yaw_link]# 2. 加载模型并获取MuJoCo受控关节名称按actuator顺序self.modelmujoco.MjModel.from_xml_path(xml_path)self.datamujoco.MjData(self.model)controlled_joint_idsself.model.actuator_trnid[:,0]self.controlled_joint_idscontrolled_joint_ids# 获取MuJoCo关节名称列表mujoco_joint_names[self.model.joint(jid).nameforjidincontrolled_joint_ids]print(MuJoCo受控关节顺序, mujoco_joint_names)csv_joint_names_clean[]fornameincsv_joint_names:# 步骤1移除dof_前缀name_without_dofname.replace(dof_,)# 步骤2移除_link后缀name_without_dof_linkname_without_dof.replace(_link,)# 步骤3补充_joint后缀匹配MuJoCo关节名name_matching_mujocof{name_without_dof_link}_jointcsv_joint_names_clean.append(name_matching_mujoco)# 3. 建立CSV关节到MuJoCo关节的索引映射对齐顺序csv_to_mujoco_idx[]formujoco_joint_nameinmujoco_joint_names:ifmujoco_joint_nameincsv_joint_names_clean: csv_idxcsv_joint_names_clean.index(mujoco_joint_name)csv_to_mujoco_idx.append(csv_idx)else: raise ValueError(fMuJoCo关节{mujoco_joint_name}未在CSV中找到对应项)# 4. 重新排列参考关节角度匹配MuJoCo关节顺序self.ref_jointsself.ref_joints_raw[:, csv_to_mujoco_idx]# [T, 28] 对齐后的关节角度assert self.ref_joints.shape[1]28, fExpected 28 joint columns, got {self.ref_joints.shape[1]}self.T_refraw_traj.shape[0]self.n_jointsself.ref_joints.shape[1]assert self.model.nuself.n_joints, fActuator count {self.model.nu} ! CSV joints {self.n_joints}self.joint_rangesnp.array([self.model.jnt_range[j_id][1]- self.model.jnt_range[j_id][0]forj_idincontrolled_joint_ids],dtypenp.float32)self.joint_minsnp.array([self.model.jnt_range[j_id][0]forj_idincontrolled_joint_ids],dtypenp.float32)self.controlled_joint_qpos_adrsself.model.jnt_qposadr[controlled_joint_ids]self.controlled_joint_dof_adrsself.model.jnt_dofadr[controlled_joint_ids]self.nominal_massself.model.body_mass.copy()self.nominal_frictionself.model.geom_friction[:,0].copy()self.nominal_kpnp.array([200.0]*12 [100.0]*16,dtypenp.float32)# 下肢关节kp200躯干/上肢100self.nominal_kdnp.array([10.0]*12 [5.0]*16,dtypenp.float32)# 下肢关节kd10躯干/上肢5self.kpself.nominal_kp.copy()self.kdself.nominal_kd.copy()self.obs_dimself.model.nq self.model.nv self.n_joints 1self.observation_spacespaces.Box(low-np.inf,highnp.inf,shape(self.obs_dim,),dtypenp.float32)self.action_spacespaces.Box(low-1.0,high1.0,shape(self.n_joints,),dtypenp.float32)self.max_episode_stepsmax_episode_steps self.render_moderender_mode self.domain_randomizationdomain_randomization self.viewerNone self.timestep0ifself.domain_randomization: self._apply_domain_randomization()print(Controlled joint names:,[self.model.joint(jid).nameforjidincontrolled_joint_ids])def _apply_domain_randomization(self): rngnp.random.default_rng()# Massforiinrange(1, self.model.nbody): self.model.body_mass[i]self.nominal_mass[i]* rng.uniform(0.8,1.2)# Frictionforiinrange(self.model.ngeom): self.model.geom_friction[i][0]self.nominal_friction[i]* rng.uniform(0.7,1.3)# Actuator gainsself.kpself.nominal_kp * rng.uniform(0.85,1.15,sizeself.n_joints)self.kdself.nominal_kd * rng.uniform(0.85,1.15,sizeself.n_joints)# Gravityself.model.opt.gravity[2]-9.81* rng.uniform(0.95,1.05)def _get_obs(self): phasefloat(self.timestep % self.T_ref)/ self.T_ref target_idxself.timestep % self.T_ref target_qself.ref_joints[target_idx]obsnp.concatenate([self.data.qpos.astype(np.float32),# shape: (35,) → 7(pelvis)28(joints)self.data.qvel.astype(np.float32),# shape: (34,)target_q,# shape: (28,)[phase]])returnobs def _pd_control(self, action): assert action.shape(self.n_joints,), fAction shape {action.shape} ! ({self.n_joints},)assert len(self.joint_mins)self.n_joints target_qself.joint_mins (action 1.0)*0.5* self.joint_ranges joint_maxsself.joint_mins self.joint_ranges target_qnp.clip(target_q, self.joint_mins, joint_maxs)q_nowself.data.qpos[self.controlled_joint_qpos_adrs]qdot_nowself.data.qvel[self.controlled_joint_dof_adrs]torqueself.kp *(target_q - q_now)- self.kd * qdot_now torquenp.clip(torque, -200.0,200.0)# 从-100→100调整为-200→200returntorque def _compute_reward(self, target_q): q_nowself.data.qpos[self.controlled_joint_qpos_adrs]track_errnp.linalg.norm(target_q - q_now,ord2)effortnp.sum(np.square(self.data.ctrl))uprightself.data.qpos[2]# 骨盆z高度com_heightself.data.xipos[1,2]# 骨盆质心高度# 放宽倾斜阈值从0.5调整为0.8降低误判pelvis_rollself.data.qpos[3]pelvis_pitchself.data.qpos[4]tilt_penalty-50.0if(np.abs(pelvis_roll)0.8or np.abs(pelvis_pitch)0.8)else0.0# 放宽高度阈值从0.4调整为0.6避免初始姿态直接触发倒地fall_penalty-200.0ifcom_height0.6else0.0reward-0.5* track_err -0.001* effort 5.0* upright tilt_penalty fall_penalty# 同步放宽倒地标记阈值self.is_fallen(com_height0.6)or(np.abs(pelvis_roll)0.8)or(np.abs(pelvis_pitch)0.8)returnreward def step(self, action): torqueself._pd_control(action)# 限制控制力矩防止力矩过大导致关节失控torquenp.clip(torque, -100.0,100.0)self.data.ctrl[:]torque mujoco.mj_step(self.model, self.data)target_idxself.timestep % self.T_ref target_qself.ref_joints[target_idx]rewardself._compute_reward(target_q)self.timestep1terminatedself.timestepself.max_episode_steps truncatedself.is_fallen obsself._get_obs()ifself.render_modehuman:self._render_frame()returnobs, reward, terminated, truncated,{}def reset(self,seedNone,optionsNone): super().reset(seedseed)self.timestep0mujoco.mj_resetData(self.model, self.data)ifself.domain_randomization: self._apply_domain_randomization()# 初始姿态使用参考轨迹第0帧新增小幅随机扰动增强鲁棒性initial_pelvisself.ref_pelvis[0]initial_jointsself.ref_joints[0]rngnp.random.default_rng(seed)initial_pelvisrng.normal(0,0.001,sizeinitial_pelvis.shape)initial_jointsrng.normal(0,0.001,sizeinitial_joints.shape)self.data.qpos[:7]initial_pelvis self.data.qpos[7:]initial_joints self.data.qvel[:]0.0mujoco.mj_forward(self.model, self.data)self.is_fallenFalse obsself._get_obs()returnobs,{}def _render_frame(self):ifself.viewer is None: from mujocoimportviewer try: self.viewerviewer.launch_passive(self.model, self.data)self.viewer._loop_rateself.metadata[render_fps]except Exception as e: print(f启动Viewer失败: {e})self.render_modeNonereturnelse: try:ifself.viewer.is_running(): self.viewer.sync()importtimetime.sleep(1/self.metadata[render_fps])# 按50fps添加固定延迟else: self.viewer.close()self.viewerNone self.render_modeNone except Exception as e: print(fViewer同步失败: {e})self.viewerNone self.render_modeNone def close(self):ifself.viewer is not None: try:ifself.viewer.is_running(): self.viewer.close()except Exception as e: print(f关闭Viewer失败: {e})finally: self.viewerNone[C]、训练部署文件from dance_envimportDanceEnv from stable_baselines3importPPOimportlogging# 配置日志logging.basicConfig(levellogging.INFO,format%(asctime)s - %(levelname)s - %(message)s)loggerlogging.getLogger(__name__)def main(): MODEL_PATH./trained_robot_model/robot_walk_ppo_model.ziptry:# 1. 加载训练好的神经网络模型logger.info(f开始加载模型: {MODEL_PATH})modelPPO.load(MODEL_PATH)logger.info(模型加载成功)# 2. 创建部署环境开启可视化logger.info(创建部署环境...)envDanceEnv(xml_pathrobot.xml,ref_csv_pathWalk.csv,max_episode_steps500,render_modehuman,# 开启可视化domain_randomizationFalse)obs, infoenv.reset(seed42)logger.info(部署环境创建成功开始运行模型...)# 3. 运行模型total_reward0.0forstepinrange(500): action, _statesmodel.predict(obs,deterministicTrue)obs, reward, terminated, truncated, infoenv.step(action)total_rewardrewardifstep %1000: logger.info(f部署步骤 {step}, 即时奖励: {reward:.3f}, 累计奖励: {total_reward:.3f})ifterminated or truncated: logger.info(f部署运行终止步骤: {step}, 原因: {达到最大步数 if terminated else 机器人倒地})obs, infoenv.reset()total_reward0.0logger.info(模型部署运行完成)except Exception as e: logger.error(f模型部署失败错误信息: {str(e)},exc_infoTrue)finally: env.close()logger.info(部署环境已关闭)if__name____main__:main()六、结束语基于动作采集文件(csv、pkl或者npz)开展基于环境仿真(mujoco、Animate或者Isaac_sim) RL的部署策略的训练是sim2Sim和sim2Real的桥接环节必要且重要。实现方法较多最新的是Isaac sim的方案。
阅读完成 · 觉得有帮助？
