第 08 章：Domain Randomization——从仿真鲁棒到真机部署
本章定位：前面三章建立了 observation/action 接口（Ch05）、reward/termination/curriculum 设计（Ch06）和 PPO 训练管线（Ch07）——这些让你能在仿真中训练出一个行为正确的策略。但仿真中训练好的策略直接部署到真机上几乎一定会失败——因为仿真器的物理参数（摩擦系数、质量、电机增益、传感器噪声）与真实世界不完全匹配。Domain Randomization（DR）是弥补这一 sim-to-real gap 的核心工程手段：通过在训练时随机化环境参数，让策略在一个参数分布上优化，而非在一个固定参数点上优化。如果真实世界的参数落在训练分布覆盖的范围内，策略就有望在真机上表现良好。
前置依赖：Ch05（Observation 与 Action 设计）、Ch06（Reward, Curriculum, Termination）、Ch07（PPO 训练管线）
关键文献：Tobin et al. 2017（视觉 DR）、Peng et al. 2018（动力学 DR）、OpenAI 2019（ADR / Rubik's Cube）、Rucker & Wensing 2022（Pseudo-inertia）、He et al. 2025（ASAP delta action model）
参考项目：🔧 mjlab EventManager · 🔧 Isaac Lab EventTermCfg · ✅ ASAP（
github.com/LeCAR-Lab/ASAP，~1.8k Stars，RSS 2025）
前置自测
📋 答不出 ≥ 2 题 → 先回前置章节复习
| 问题 | 检查目的 |
|---|---|
| mjlab 的 EventManager 支持哪几种触发模式？每种适合什么场景？ | 检查 Ch04 manager 架构理解 |
reward 的 scale_by_dt 和 DR 的关系是什么？DR 改变 decimation 后 reward scale 会变吗？ |
检查 Ch06 dt 缩放理解 |
| 如果只修改质量不修改惯量张量，这个刚体在物理上是否一致？ | 检查刚体动力学基础 |
| DR 改变环境参数后，PPO 的 advantage 估计会受什么影响？ | 检查 Ch07 GAE 理解 |
| 如果 friction DR 范围设为 [0.01, 5.0]，策略会学到什么行为？ | 检查 DR 范围选择直觉 |
| 仿真中训练效果很好但真机上失败，最可能的原因是什么？ | 检查 sim-to-real gap 概念 |
本章难度为 ⭐⭐⭐。DR 的理论很简单（在参数分布上训练），但工程实现充满陷阱——从 CUDA Graph 静默失效到物理不一致的惯性参数，每个陷阱都可能让你在完全没有 DR 的情况下误以为策略是鲁棒的。
本章目标
学完本章后，你应该能够：
- 解释 DR 的鲁棒优化视角——为什么在参数分布上训练可以提升 sim-to-real 迁移
- 配置 mjlab 和 Isaac Lab 中完整的 DR 方案——startup/reset/interval/step 四种模式各用于什么
- 实现 物理一致的惯性随机化——理解 pseudo-inertia 参数化为什么比单独改 mass 更合理
- 设计 分阶段 DR 策略——从无 DR baseline 到完整 DR 的递进方案
- 诊断 DR 不生效的常见原因——从配置层到 CUDA Graph 的四层排查
- 评估 DR 策略的鲁棒性——在固定参数网格上系统测试
- 理解 ASAP delta action model 的动机——当传统 DR 不够时的替代方案
本章路线图 ⭐
8.1 理论基础（为什么需要 DR）→ 8.2 EventManager 四种模式（何时触发）→ 8.3 DR 函数映射表（随机化什么）→ 8.4 pseudo-inertia（物理一致性）→ 8.5 外力扰动（push/impulse）→ 8.6 分阶段 DR 策略（怎样递进）→ 8.7 评估方法论（怎样验证）→ 8.8 ASAP 精读（超越 DR）→ 8.9 观测噪声与延迟（传感器 DR）→ 8.10 自定义 Event（扩展框架）→ 8.11 诊断工具箱（一站式诊断）→ 8.12 源码路线（深入探索）。
前置依赖与本章定位 ⭐
本章在全书中的定位是"从仿真到真机的第一道桥"。Ch05-Ch07 建立了在仿真中训练策略的完整能力，本章让这个策略不仅在仿真标称参数下工作，而是在一个参数范围内都工作。Ch09（Teacher-Student 蒸馏）将在此基础上解决另一个 sim-to-real 问题——部署时缺少 privileged 信息。Ch23（Sim2Real 部署全链路）将整合 DR + 蒸馏 + 执行器建模 + 部署工程，完成完整的 sim-to-real 管线。
DR 和 Ch06 的 Curriculum Learning 有交互但本质不同。Curriculum 改变任务难度（命令范围、地形复杂度）但物理规律不变。DR 改变物理规律（摩擦、质量、延迟）但任务目标不变。两者可以同时使用，但不要混淆——"逐步扩大摩擦范围"是 DR，不是 curriculum。
8.1 DR 的理论基础与鲁棒优化视角 ⭐⭐
这一节解决什么问题：为什么在参数分布上训练可以提升 sim-to-real 迁移？DR 的数学本质是什么？
从固定 MDP 到参数化 MDP ⭐⭐
标准 RL 在一个固定的 MDP \(\mathcal{M} = (S, A, P, R, \gamma)\) 上训练。但真实世界的 MDP 参数 \(\theta\)（摩擦、质量、电机增益等）是未知的——仿真器使用的标称参数 \(\theta_0\) 只是一个估计。DR 的核心思想是：不要在 \(\theta_0\) 上训练，而是在一个参数分布 \(p(\theta)\) 上训练。
\[J_{\text{DR}}(\pi) = \mathbb{E}_{\theta \sim p(\theta)}\left[J(\pi, \theta)\right] = \int J(\pi, \theta) \, p(\theta) \, d\theta\]
其中 \(J(\pi, \theta) = \mathbb{E}_\pi\left[\sum_t \gamma^t r_t \mid \theta\right]\) 是策略 \(\pi\) 在参数 \(\theta\) 下的期望回报。DR 优化的是平均性能，而非某个特定参数下的性能。
这和保险的精算逻辑有深层类比。保险公司不为某一个特定客户优化保费（那样要么赔钱要么客户付不起），而是在客户群体的风险分布上优化——让平均利润最大化。DR 策略不为某一个特定摩擦系数优化（那样换个地面就摔倒），而是在摩擦分布上优化——让平均 tracking 性能最大化。
DR 不是万能的。DR 优化的是期望（平均情况），不是最坏情况。如果参数分布 \(p(\theta)\) 的某些区域特别难（如极低摩擦 \(\mu = 0.1\)），DR 策略可能选择"放弃"这些区域来提高平均性能——这在数学上是最优的，但在部署时如果遇到这些参数就会失败。这就是为什么 DR 范围的选择如此重要——范围应该反映真实的物理不确定性，而不是盲目扩大。
反事实推理：如果把 DR 目标从期望改为最小值 \(\min_\theta J(\pi, \theta)\)？这就变成了鲁棒优化（minimax），策略会为最坏情况优化——结果通常是极度保守的行为（站着不动是对最坏摩擦的"最优"响应）。工程实践中，DR 的"平均优化"比鲁棒优化的"最坏优化"更实用。
DR 的四个维度 ⭐⭐
DR 不是一个单一的旋钮——它沿四个维度展开，每个维度对应真实世界中不同类型的不确定性：
| 维度 | 随机化内容 | 影响什么 | 典型参数 |
|---|---|---|---|
| 动力学 | 质量/惯量/COM/摩擦/接触刚度 | 系统对力的响应 | body_mass, geom_friction, dof_damping |
| 执行器 | 电机增益/力矩上限/响应延迟/齿轮间隙 | 动作的有效幅度和带宽 | pd_gains, effort_limits, dof_armature |
| 感知 | 编码器偏置/IMU 噪声/通信延迟 | 策略接收到的信息质量 | encoder_bias, actor corruption, obs noise |
| 环境 | 外部扰动/风力/负载变化 | 任务本身的难度 | push velocity, external wrench |
这个分类框架的工程价值在于：每个 DR 项都应该能追溯到具体的物理不确定性来源。如果一个随机化项找不到真实的部署对应物（如"每秒改变一次质量"——物理上不可能），它只能作为压力测试，不能作为 sim-to-real 的证据。
Automatic Domain Randomization (ADR) ⭐⭐⭐
OpenAI (2019) 提出的 ADR 自动调节 DR 范围：从无随机化开始，当策略性能超过上阈值时扩大某个参数的范围，当性能低于下阈值时收缩。ADR 在 Shadow Hand 解魔方的任务中展示了 sim-to-real 迁移，且训练出的 LSTM 策略的 hidden state 自发编码了环境参数（类似 meta-learning）。
ADR 的工程挑战：(1) 各参数独立扩展可能产生不可训练的参数组合（如同时低摩擦 + 低电机力矩）；(2) 范围扩大后需要重新收集 on-policy 数据——PPO 的旧数据可能不再代表新分布；(3) 超参数（阈值、步长 Δ）本身需要调优。
在当前实践中，ADR 主要用于灵巧操作（Isaac Lab 2.3 DexSuite 集成了 ADR）。对于 locomotion，手动设置 DR 范围 + 分阶段引入仍然是主流方法——因为 locomotion 的参数不确定性范围通常可以从工程规格书中获得，不需要自动搜索。
delta action model 的动机 ⭐⭐
当 DR 不够时怎么办？ASAP（He et al., RSS 2025）提出了一种替代思路：不是把仿真器的参数分布做得更宽，而是训练一个残差模型来修正仿真器和真实世界的差异。这个 delta action model 在 8.8 节详细讲解。
本质洞察：DR 和 delta action model 解决 sim-to-real gap 的方式完全不同。DR 说"我不知道真实参数是什么，所以我在一大堆可能的参数上训练"。delta model 说"我知道仿真和真实有差距，让我用真机数据学一个修正项"。DR 不需要真机数据但可能过于保守，delta model 需要真机数据但更精确。两者不互斥——ASAP 第一阶段在仿真中训练 motion-tracking base policy，通常也会配合合理 DR（但注意其公开代码的训练命令默认
+domain_rand=NO_domain_rand，是否叠加 DR 取决于具体配置，并非论文强制步骤）。
⚠️ 常见陷阱
⚠️ 编程陷阱：DR 配置写了但没生效。 从配置到物理生效需要经过四层（配置→管理→字段→仿真），任何一层断裂都会导致 DR 静默失效——训练正常运行，日志中 reward 也在上升，但策略实际上是在无 DR 的环境中训练的。自检方法：训练开始后打印已扩展的 model fields 列表。
💡 概念误区：DR 范围越大越鲁棒。 过大的 DR 范围会把原本可学的任务变成不可学的鲁棒优化问题。在 [0.01, 5.0] 的摩擦范围下，极低摩擦可能让任何步态都无法稳定——策略被迫学习极度保守的行为。DR 范围应该反映真实的物理不确定性，不是"越大越好"。
🧠 思维陷阱：把 DR 等同于数据增强。 图像增强改变的是输入的表示形式，不改变任务本身。DR 改变的是环境动力学——它改变了状态转移函数 \(p(s'|s,a;\theta)\)。在低摩擦环境中，相同的动作会产生完全不同的接触力和滑移。DR 本质上是在训练策略同时解一族不同的控制问题。
练习
- [推导题] 证明当 DR 分布 \(p(\theta) = \delta(\theta - \theta_0)\)（退化为点分布）时，\(J_{\text{DR}}\) 退化为标准 MDP 目标 \(J(\pi, \theta_0)\)。
- [分析题] 如果 Go1 的标称质量是 12.5 kg，装配误差在 [-5%, +8%]，负载最多 2 kg，计算
body_mass的合理 DR 范围。 - [思考题] 为什么 DR 通常选择均匀分布而不是正态分布？提示：考虑尾部行为和边界覆盖。
上节建立了 DR 的理论动机。但理论不告诉你"什么时候触发随机化"——这由 EventManager 的模式决定。模式选错会导致从训练不稳定到物理不一致的各种问题。
8.2 EventManager 四种模式 ⭐⭐⭐
这一节解决什么问题：四种事件触发模式各适合什么场景？选错模式的后果是什么？
模式总览 ⭐⭐
mjlab 和 Isaac Lab 的 EventManager 支持四种（Isaac Lab 多一种 prestartup）触发模式。每种模式对应不同的物理语义：
| 模式 | 触发时机 | 典型用途 | 适合随机化什么 |
|---|---|---|---|
startup |
环境初始化后一次 | 硬件之间的固定差异 | 质量、惯量基线 |
reset |
episode 重置时 | 每集可能变化的参数 | 摩擦、PD 增益、COM 偏移 |
interval |
随机间隔触发 | 稀疏外部扰动 | push velocity, 重力扰动 |
step |
每个 env step | 需要精确时序的外力 | impulse 生命周期管理 |
Isaac Lab 额外有 prestartup 模式——在仿真启动前的 USD 层面执行，用于 randomize_rigid_body_scale 等必须在 PhysX 初始化前完成的操作。mjlab 不需要此模式因为 MuJoCo 没有 USD 中间层。
双框架配置对照 ⭐⭐⭐
以下是 Go1 velocity task 中典型 DR 事件的双框架配置：
# ============= mjlab 事件配置 =============
# src/mjlab/tasks/velocity/velocity_env_cfg.py
events = {
# startup: 模拟硬件差异（每个 env 不同但 episode 内不变）
"foot_friction": EventTermCfg(
func=mdp.randomize_geom_friction,
mode="startup",
params={
"entity_cfg": SceneEntityCfg("robot", geom_names=".*foot.*"),
"friction_range": (0.5, 1.25),
"operation": "scale",
},
),
"base_mass": EventTermCfg(
func=mdp.randomize_body_mass,
mode="startup",
params={
"entity_cfg": SceneEntityCfg("robot", body_names="base"),
"mass_distribution_params": (-1.0, 3.0),
"operation": "add",
},
),
"encoder_bias": EventTermCfg(
func=mdp.add_body_offset,
mode="startup",
params={
"entity_cfg": SceneEntityCfg("robot"),
"offset_range": (-0.05, 0.05),
},
),
# reset: 每集环境变化
"reset_robot_state": EventTermCfg(
func=mdp.reset_root_state_uniform,
mode="reset",
params={
"pose_range": {"x": (-0.5, 0.5), "y": (-0.5, 0.5), "yaw": (-3.14, 3.14)},
"velocity_range": {
"x": (-0.5, 0.5), "y": (-0.5, 0.5), "z": (-0.5, 0.5),
"roll": (-0.5, 0.5), "pitch": (-0.5, 0.5), "yaw": (-0.5, 0.5),
},
},
),
"reset_joints": EventTermCfg(
func=mdp.reset_joints_by_offset,
mode="reset",
params={
"position_range": (-0.5, 0.5),
"velocity_range": (-0.5, 0.5),
},
),
# interval: 稀疏扰动
"push_robot": EventTermCfg(
func=mdp.push_by_setting_velocity,
mode="interval",
interval_range_s=(10.0, 15.0),
params={
"velocity_range": {
"x": (-1.0, 1.0), "y": (-1.0, 1.0),
"roll": (-0.25, 0.25), "pitch": (-0.25, 0.25),
},
},
),
}
# ============= Isaac Lab 事件配置 =============
# source/isaaclab_tasks/.../velocity/velocity_env_cfg.py
@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, # PhysX 材质 bucket 限制
},
)
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",
},
)
push_robot = EventTerm(
func=mdp.push_by_setting_velocity,
mode="interval",
is_global_time=True,
interval_range_s=(10.0, 15.0),
params={
"velocity_range": {"x": (-1.0, 1.0), "y": (-1.0, 1.0)},
},
)
关键差异：
| 维度 | mjlab | Isaac Lab |
|---|---|---|
| 摩擦随机化 | 直接修改 geom_friction 字段 |
通过 randomize_rigid_body_material + num_buckets |
| PD 增益 | 修改 actuator_kp / actuator_kd |
randomize_actuator_gains + log_uniform |
| 材质限制 | 无限制（MuJoCo 无 bucket 概念） | PhysX 限 64,000 unique materials → 必须用 bucket |
| prestartup | 不需要（无 USD 层） | USD-level 操作必须用 prestartup |
| 字段扩展 | expand_model_fields() + CUDA Graph 重 capture |
PhysX 内部管理 |
模式选择决策树 ⭐⭐
这个参数在真实世界中多久变化一次？
├── 出厂后不变（硬件公差、装配误差）
│ └── startup 模式
│ 例：base_mass, encoder_bias, link_inertia
├── 每个任务/场景可能不同
│ └── reset 模式
│ 例：ground_friction, PD_gains, initial_pose
├── 任务中偶尔突发
│ └── interval 模式
│ 例：push（被人撞了一下）, wind_gust, gravity_tilt
└── 每一步都在变（连续扰动）
└── step 模式（或 actor corruption in obs）
例：sensor_noise（通常通过 obs corruption 实现更高效）
反事实推理：如果把应该用 startup 的质量随机化放到 interval 模式会怎样？每隔几秒机器人的质量就会突然改变——这在物理上违反了质量守恒。策略会被迫学习一种非物理的"质量跳变"响应，部署到真机上不会遇到这种情况，训练出的应对能力完全浪费。更糟的是，质量变化可能触发 set_const 级别的派生常量重算，导致训练吞吐量大幅下降。
EventManager 内部调度机制 ⭐⭐
# EventManager.apply() 的核心逻辑（简化）
def apply(self, mode: str, env_ids: torch.Tensor | None = None):
"""按模式执行所有对应的事件。"""
for name, (func, cfg) in self._mode_term_cfgs[mode].items():
if mode == "interval":
# interval 模式使用 per-env 计时器
env_ids_to_apply = self._check_interval_timer(name, env_ids)
if env_ids_to_apply is None or len(env_ids_to_apply) == 0:
continue
func.func(self._env, env_ids_to_apply, **cfg.params)
elif mode == "reset":
# reset 模式只对刚被 reset 的 env 执行
func.func(self._env, env_ids, **cfg.params)
elif mode == "startup":
# startup 模式在初始化时对所有 env 执行一次
all_env_ids = torch.arange(self._env.num_envs, device=self._env.device)
func.func(self._env, all_env_ids, **cfg.params)
elif mode == "step":
# step 模式每步对所有 env 执行
all_env_ids = torch.arange(self._env.num_envs, device=self._env.device)
func.func(self._env, all_env_ids, **cfg.params)
interval 计时器的实现。interval 模式使用 per-env 计时器避免所有环境同步触发——如果 4096 个环境在同一步都收到 push，会导致 batch 内的 transition 高度相关，GAE 的 advantage 估计方差暴增。per-env 计时器让每个 env 独立触发 push，保持 batch 的多样性。
# interval 计时器逻辑
def _check_interval_timer(self, term_name, env_ids):
"""检查哪些 env 应该在当前步触发 interval 事件。"""
lo, hi = self._term_cfgs[term_name].interval_range_s
# 每个 env 的下一次触发时间
self._interval_timers[term_name] -= self._env.step_dt
# 找到计时器到期的 env
expired = self._interval_timers[term_name] <= 0
env_ids_to_apply = expired.nonzero().squeeze(-1)
# 重置到期 env 的计时器（重新采样间隔）
if len(env_ids_to_apply) > 0:
# 必须指定 device，否则 CPU 张量赋值给 CUDA 张量会触发 device mismatch
new_intervals = torch.rand(len(env_ids_to_apply), device=self._env.device) * (hi - lo) + lo
self._interval_timers[term_name][env_ids_to_apply] = new_intervals
return env_ids_to_apply
⚠️ 常见陷阱
⚠️ 编程陷阱：interval 事件的 is_global_time=True 导致同步触发。 Isaac Lab 的 is_global_time=True 让所有 env 共享计时器——所有 env 同时收到 push。对 locomotion 来说这通常不是期望行为。推荐 is_global_time=False（默认），让每个 env 独立计时。
💡 概念误区：认为 reset 模式 = startup 模式重复执行。 startup 在环境创建后只触发一次，之后所有 episode 都使用相同的参数。reset 在每次 episode 重置时重新采样。startup 模拟"每台机器人出厂不同但使用中不变"，reset 模拟"每次任务环境可能不同"。
🧠 思维陷阱：认为"事件配置写了就一定生效"。 从配置到物理生效需要经过四层。自检方法：训练开始后打印 sim.expanded_fields（mjlab）或检查 PhysX material count（Isaac Lab），确认目标字段确实被随机化了。
练习
- [分类题] 把以下 DR 项分配到正确的模式：(a) 足底摩擦系数 (b) base mass (c) 外部推力 (d) encoder bias (e) 初始关节位置 (f) 重力方向微扰。说明每个的物理理由。
- [源码题] 在 mjlab velocity task 的 cfg 中找到所有事件配置。画出事件触发时间线（哪些在 startup、哪些在每次 reset、哪些在 episode 中间）。
- [调试题] 训练有 DR 和无 DR 的效果完全一样。列出至少 3 个可能的原因和排查步骤。
上节讲了"什么时候触发"。接下来的核心问题是"随机化什么"——每个物理参数对应哪个 API，两个框架有什么差异。
8.3 DR 函数映射表与双框架配置 ⭐⭐
这一节解决什么问题：双框架中可随机化的物理参数有哪些？API 怎么调用？参数范围怎么选？
完整 DR 函数对照表 ⭐⭐⭐
| 物理参数 | mjlab 函数 | Isaac Lab 函数 | 推荐模式 | 典型范围 |
|---|---|---|---|---|
| 刚体质量 | randomize_body_mass |
randomize_rigid_body_mass |
startup/reset | add: (-1, 3) kg |
| 地面摩擦 | randomize_geom_friction |
randomize_rigid_body_material |
startup/reset | scale: (0.5, 1.25) |
| PD 刚度 | randomize_actuator_gains |
randomize_actuator_gains |
reset | log_uniform: (0.75, 1.5) |
| PD 阻尼 | randomize_actuator_gains |
randomize_actuator_gains |
reset | log_uniform: (0.3, 3.0) |
| 关节 armature | randomize_dof_armature |
(via joint properties) | startup | scale: (0.9, 1.1) |
| COM 偏移 | randomize_body_com |
randomize_rigid_body_mass (recompute_inertia) |
startup | add: (-2cm, 2cm) |
| 弹性系数 | randomize_geom_restitution |
(via randomize_rigid_body_material) |
startup | (0.0, 0.5) |
| 外部推力 | push_by_setting_velocity |
push_by_setting_velocity |
interval | (-1, 1) m/s |
| 脉冲力 | apply_body_impulse |
apply_external_force_torque |
step | (0, 50) N |
| 重力方向 | randomize_gravity |
randomize_physics_scene_gravity |
interval | ±5° tilt |
| 初始姿态 | reset_root_state_uniform |
reset_root_state_uniform |
reset | xyz, rpy 范围 |
| 初始关节 | reset_joints_by_offset |
reset_joints_by_offset |
reset | ±0.5 rad |
| 编码器偏置 | add_body_offset |
(via obs corruption) | startup | ±0.05 rad |
| 惯量张量 | pseudo_inertia |
(手动实现) | startup | alpha: (0.8, 1.2) |
采样分布的选择 ⭐⭐
DR 函数通常支持三种采样分布。选择哪种分布取决于参数的物理特性：
均匀分布（uniform）：适用于参数范围明确且均匀覆盖的情况。摩擦系数、初始姿态偏移通常用均匀分布。
对数均匀分布（log_uniform）：适用于参数跨越多个数量级的情况。PD 增益的阻尼系数可能从 0.3 到 3.0——在线性空间中，0.3-1.0 占范围的 30% 但在对数空间中占 50%。log_uniform 确保低值和高值都有足够的采样概率。
# 对数均匀采样的实现
import torch
def log_uniform(low: float, high: float, size: tuple) -> torch.Tensor:
"""在对数空间均匀采样。"""
log_low = torch.log(torch.tensor(low))
log_high = torch.log(torch.tensor(high))
return torch.exp(torch.rand(size) * (log_high - log_low) + log_low)
# 示例：PD 阻尼从 0.3 到 3.0
damping_samples = log_uniform(0.3, 3.0, (4096, 12))
# 线性空间中：约 50% 的样本在 [0.3, 1.0]，50% 在 [1.0, 3.0]
# 这比 uniform(0.3, 3.0) 更好——uniform 只有 26% 的样本在 [0.3, 1.0]
高斯分布（gaussian）：适用于参数集中在标称值附近但有长尾的情况。观测噪声通常用高斯分布。注意需要 clip 防止极端值。
operation 模式 ⭐⭐
DR 函数的 operation 参数控制如何应用采样值：
| operation | 含义 | 公式 | 适用场景 |
|---|---|---|---|
"add" |
在标称值上加偏移 | \(\theta_{\text{new}} = \theta_0 + \Delta\) | 质量负载、编码器偏置 |
"scale" |
按比例缩放标称值 | \(\theta_{\text{new}} = \theta_0 \times s\) | 摩擦、PD 增益 |
"abs" |
直接设置绝对值 | \(\theta_{\text{new}} = v\) | 需要精确控制范围的参数 |
选择原则：如果参数的不确定性与标称值成比例（如"摩擦系数可能偏离标称值 ±30%"），用 scale。如果不确定性是绝对量（如"负载 ±2 kg"），用 add。
自定义 DR 函数的编写模式 ⭐⭐
# mjlab 自定义 DR 函数模板
def my_custom_randomization(
env: ManagerBasedRLEnv,
env_ids: torch.Tensor,
entity_cfg: SceneEntityCfg,
range_params: tuple[float, float],
operation: str = "scale",
):
"""自定义 DR 函数必须遵循的签名。
Args:
env: 环境引用
env_ids: 需要随机化的 env indices [N]
entity_cfg: 目标实体配置（机器人、关节等）
range_params: (low, high) 采样范围
operation: "add" | "scale" | "abs"
"""
# 1. 获取目标实体
entity = env.scene[entity_cfg.name]
# 2. 采样随机值
low, high = range_params
num_envs = len(env_ids)
random_values = torch.rand(num_envs, device=env.device) * (high - low) + low
# 3. 获取当前参数值
current = entity.data.some_parameter[env_ids] # 取决于具体参数
# 4. 应用 operation
if operation == "add":
new_values = current + random_values
elif operation == "scale":
new_values = current * random_values
else: # abs
new_values = random_values
# 5. 写回（通过 mjlab 的 model field API）
entity.write_some_parameter_to_sim(new_values, env_ids)
Isaac Lab 的自定义 DR 函数签名略有不同——通过 asset_cfg 而非 entity_cfg 访问：
# Isaac Lab 自定义 DR 函数模板
def my_custom_randomization_isaac(
env: ManagerBasedRLEnv,
env_ids: torch.Tensor,
asset_cfg: SceneEntityCfg,
distribution_params: tuple[float, float],
operation: str = "scale",
distribution: str = "uniform",
):
"""Isaac Lab DR 函数签名。"""
asset = env.scene[asset_cfg.name]
if distribution == "uniform":
values = torch.rand(len(env_ids), device=env.device) * (
distribution_params[1] - distribution_params[0]
) + distribution_params[0]
elif distribution == "log_uniform":
values = log_uniform(distribution_params[0], distribution_params[1],
(len(env_ids),))
elif distribution == "gaussian":
mean, std = distribution_params
values = torch.randn(len(env_ids), device=env.device) * std + mean
# 应用 operation 到资产...
DR 配置验证工具 ⭐⭐⭐
以下工具在训练开始前验证 DR 配置是否正确——包括字段扩展、entity 匹配和数值范围：
# dr_verify.py — DR 配置验证工具
import torch
def verify_dr_config(env, verbose=True):
"""验证 DR 配置的完整性和正确性。"""
em = env.event_manager
issues = []
print("DR Configuration Verification")
print("=" * 60)
# 1. 检查各模式的事件数量
for mode in ["startup", "reset", "interval", "step"]:
terms = em._mode_term_cfgs.get(mode, {})
count = len(terms)
print(f"\n [{mode:>8s}] {count} events:")
for name, (func, cfg) in terms.items():
print(f" - {name}: func={func.func.__name__}")
if hasattr(cfg, 'params'):
for k, v in cfg.params.items():
if 'range' in k.lower() or 'distribution' in k.lower():
print(f" {k} = {v}")
# 2. 检查 expanded fields（mjlab 特有）
if hasattr(env.sim, 'expanded_fields'):
fields = env.sim.expanded_fields
print(f"\n Expanded model fields: {fields}")
if not fields:
issues.append("⚠️ No expanded fields — DR may not be writing to sim!")
# 3. 检查 entity pattern 匹配
for mode, terms in em._mode_term_cfgs.items():
for name, (func, cfg) in terms.items():
if hasattr(cfg, 'params') and 'entity_cfg' in cfg.params:
entity_cfg = cfg.params['entity_cfg']
# 检查 entity 是否存在于 scene 中
try:
entity = env.scene[entity_cfg.name]
# 检查 body/geom/joint pattern 是否匹配到内容
if hasattr(entity_cfg, 'body_names') and entity_cfg.body_names:
matched = entity.find_bodies(entity_cfg.body_names)
if len(matched) == 0:
issues.append(
f"⚠️ {name}: body pattern '{entity_cfg.body_names}' "
f"matched 0 bodies!")
else:
print(f" {name}: matched {len(matched)} bodies")
except KeyError:
issues.append(f"🔴 {name}: entity '{entity_cfg.name}' not in scene!")
# 4. 范围合理性检查
for mode, terms in em._mode_term_cfgs.items():
for name, (func, cfg) in terms.items():
if hasattr(cfg, 'params'):
for k, v in cfg.params.items():
if isinstance(v, (tuple, list)) and len(v) == 2:
lo, hi = v
if isinstance(lo, (int, float)) and isinstance(hi, (int, float)):
if lo >= hi:
issues.append(f"⚠️ {name}.{k}: range [{lo}, {hi}] is empty!")
if 'mass' in k.lower() and lo < 0 and 'add' not in str(cfg.params.get('operation', '')):
issues.append(f"⚠️ {name}.{k}: negative mass range!")
# 报告
print(f"\n{'='*60}")
if issues:
print(f"Found {len(issues)} issues:")
for issue in issues:
print(f" {issue}")
else:
print("✅ All DR checks passed!")
return len(issues) == 0
# 使用：在训练前调用
# verify_dr_config(env)
DR 生效性测试——before/after 对比 ⭐⭐
# dr_effectiveness_test.py — 验证 DR 是否真正影响了物理仿真
def test_dr_effectiveness(env, param_name="geom_friction", num_envs_to_check=5):
"""验证 DR 是否产生了 per-env 不同的物理参数。"""
print(f"\nDR Effectiveness Test: {param_name}")
print("=" * 50)
# Reset 触发 DR
env.reset()
# 检查不同 env 的参数值是否不同
# （具体实现取决于框架和参数类型）
if hasattr(env.sim, 'model'):
model = env.sim.model
if hasattr(model, param_name):
field = getattr(model, param_name)
if field.dim() >= 2: # per-env field
for i in range(min(num_envs_to_check, env.num_envs)):
print(f" env[{i}]: {field[i].cpu().numpy()[:4]}...")
# 检查 env 间的差异
if field.shape[0] > 1:
variance = field.var(dim=0).mean().item()
print(f" Cross-env variance: {variance:.6f}")
if variance < 1e-10:
print(f" ⚠️ All envs have SAME {param_name} — DR not working!")
else:
print(f" ✅ Envs have different {param_name} — DR is active")
else:
print(f" ⚠️ {param_name} is shared (not per-env) — not expanded!")
# 功能测试：在不同摩擦下同一 action 应产生不同 next state
obs = env.reset()
actions = torch.zeros(env.num_envs, env.action_manager.total_action_dim,
device=env.device)
obs_after, _, _, _, _ = env.step(actions)
# 如果 DR 生效，不同 env 的 next state 应该不同
actor_obs = obs_after.get("actor", obs_after.get("policy"))
env_variance = actor_obs.var(dim=0).mean().item()
print(f" Obs variance across envs (same action): {env_variance:.6f}")
if env_variance < 1e-8:
print(f" ⚠️ Same action → same obs in all envs — DR may not be effective!")
完整的 DR Checklist ⭐⭐
在训练完成后、进入 Ch09 之前，逐项验证：
[ ] 1. Phase 0 clean baseline tracking > 0.7
[ ] 2. Phase 4 full DR tracking > 0.55（允许比 Phase 0 低 15-25%）
[ ] 3. 摩擦网格评估：最低摩擦处 tracking > 0.3, fall < 30%
[ ] 4. 质量网格评估：±3 kg 处 tracking > 0.4
[ ] 5. Push recovery：1 m/s push 后 2 秒内恢复
[ ] 6. 分布外测试：范围外 ±20% 参数下平缓退化（非突然崩溃）
[ ] 7. verify_dr_config() 无 ⚠️ 或 🔴
[ ] 8. test_dr_effectiveness() 确认 per-env 参数不同
[ ] 9. TensorBoard 中 tracking 和 penalty 的 DR 开关 A/B 对比合理
⚠️ 常见陷阱
⚠️ 编程陷阱：entity pattern 空匹配导致 DR 静默失效。 用正则 "foot.*" 匹配足端 geom，但 MJCF 中名字是 left_foot_1、right_foot_2——pattern 空匹配，摩擦随机化从未被应用。自检：在小 env 中先打印 pattern 匹配的 entity ids，确认非空。
⚠️ 编程陷阱：Isaac Lab 的 num_buckets 设置不当。 PhysX 限制每个场景 64,000 个 unique materials。如果 num_buckets=None（无 bucket），4096 envs × 若干 body = 可能超限导致崩溃。推荐设 num_buckets=250。
练习
- [编程题] 在 mjlab 中为 Go1 velocity task 添加一个自定义 DR：随机化
dof_damping，使用 log_uniform 分布，范围 (0.5, 2.0)，operation="scale"，mode="reset"。 - [计算题]
damping_distribution_params=(0.3, 3.0)+distribution="log_uniform"+operation="scale"。标称阻尼 \(d_0 = 1.0\)。计算随机化后阻尼的范围和中位数。 - [跨框架题] 对比 mjlab 和 Isaac Lab 中 foot friction DR 的配置差异。两者实际效果等价吗？
8.4 物理一致的惯性随机化：pseudo-inertia ⭐⭐⭐
这一节解决什么问题：为什么单独随机化 mass 是不够的？如何实现物理一致的刚体惯性参数随机化？
问题：朴素质量随机化的物理不一致 ⭐⭐⭐
一个刚体的惯性特征由 10 个参数完整描述：质量 \(m\)（1 个）、质心位置 \(\mathbf{c}\)（3 个）、惯量张量 \(\mathbf{I}\)（6 个独立参数，\(\mathbf{I}\) 是对称正定矩阵）。这 10 个参数之间有物理约束——不是所有 \((m, \mathbf{c}, \mathbf{I})\) 组合都对应一个真实存在的刚体。
如果只随机化 \(m\) 而不动 \(\mathbf{I}\) 和 \(\mathbf{c}\)，得到的刚体物理上不一致——质量翻倍但惯量不变，意味着平动响应变了但旋转响应没变。这就像一辆车质量翻倍但转弯半径不变——物理上不可能。MuJoCo 在这种参数下不会报错但行为可能不自然，PhysX 可能静默接受或产生微妙的接触异常。
Pseudo-Inertia 参数化（Rucker & Wensing 2022） ⭐⭐⭐
解决方案是用 pseudo-inertia 矩阵 \(\Sigma \in \mathbb{R}^{4 \times 4}\) 统一表示所有 10 个参数：
\[\Sigma = \begin{pmatrix} \frac{1}{2}\text{tr}(\mathbf{I})\mathbf{I}_3 - \mathbf{I} + m \mathbf{c}\mathbf{c}^T & m\mathbf{c} \\ m\mathbf{c}^T & m \end{pmatrix}\]
关键性质：\(\Sigma\) 对称正定 \(\Leftrightarrow\) \((m, \mathbf{c}, \mathbf{I})\) 物理一致。因此，只要在对称正定矩阵空间中随机采样，就保证了物理一致性。
Cholesky 分解方法：\(\Sigma = LL^T\)，\(L\) 是下三角矩阵（10 个独立参数）。在 \(L\) 的元素上做随机扰动，通过 \(LL^T\) 恢复正定矩阵，再提取 \((m, \mathbf{c}, \mathbf{I})\)。
# pseudo-inertia 随机化的核心逻辑（简化自 mjlab 实现）
import torch
def randomize_pseudo_inertia(
nominal_mass: torch.Tensor, # [num_envs, num_bodies]
nominal_com: torch.Tensor, # [num_envs, num_bodies, 3]
nominal_inertia: torch.Tensor, # [num_envs, num_bodies, 3, 3]
alpha_range: tuple[float, float], # 全局密度缩放范围
env_ids: torch.Tensor,
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
"""物理一致的惯性参数随机化。
Returns:
new_mass, new_com, new_inertia — 保证物理一致
"""
num_envs = len(env_ids)
num_bodies = nominal_mass.shape[1]
results_mass = nominal_mass[env_ids].clone()
results_com = nominal_com[env_ids].clone()
results_inertia = nominal_inertia[env_ids].clone()
for b in range(num_bodies):
m = nominal_mass[env_ids, b] # [N]
c = nominal_com[env_ids, b] # [N, 3]
I = nominal_inertia[env_ids, b] # [N, 3, 3]
# 构造 pseudo-inertia 矩阵 [N, 4, 4]
sigma = torch.zeros(num_envs, 4, 4, device=m.device)
sigma[:, :3, :3] = (0.5 * torch.diagonal(I, dim1=1, dim2=2).sum(1, keepdim=True).unsqueeze(-1)
* torch.eye(3, device=m.device).unsqueeze(0) - I
+ m.unsqueeze(-1).unsqueeze(-1) * torch.bmm(c.unsqueeze(2), c.unsqueeze(1)))
sigma[:, :3, 3] = m.unsqueeze(-1) * c
sigma[:, 3, :3] = m.unsqueeze(-1) * c
sigma[:, 3, 3] = m
# Cholesky 分解
L = torch.linalg.cholesky(sigma) # [N, 4, 4]
# 全局密度缩放（alpha）
alpha = torch.rand(num_envs, 1, 1, device=m.device) * (alpha_range[1] - alpha_range[0]) + alpha_range[0]
L_perturbed = L * torch.sqrt(alpha)
# 恢复正定矩阵
sigma_new = torch.bmm(L_perturbed, L_perturbed.transpose(1, 2))
# 提取物理参数
eye = torch.eye(3, device=m.device).unsqueeze(0)
m_new = sigma_new[:, 3, 3]
c_new = sigma_new[:, :3, 3] / m_new.unsqueeze(-1)
Sigma_new = sigma_new[:, :3, :3] # 关于原点的二阶矩 Σ_origin
# 逆映射 I = tr(Σ)·I₃ − Σ；再用平行轴定理从原点移回质心：减去 m(‖c‖²I₃ − ccᵀ)
I_origin = (torch.diagonal(Sigma_new, dim1=1, dim2=2).sum(1, keepdim=True).unsqueeze(-1) * eye
- Sigma_new)
I_recovered = I_origin - m_new.unsqueeze(-1).unsqueeze(-1) * (
c_new.square().sum(1, keepdim=True).unsqueeze(-1) * eye
- torch.bmm(c_new.unsqueeze(2), c_new.unsqueeze(1)))
results_mass[:, b] = m_new
results_com[:, b] = c_new
results_inertia[:, b] = I_recovered
return results_mass, results_com, results_inertia
expand_model_fields 机制 ⭐⭐⭐
mjlab 的 DR 需要 per-env 存储来支持每个环境独立的物理参数。默认情况下，MuJoCo Warp 的模型参数是所有环境共享的（节省显存）。expand_model_fields() 负责把共享字段扩展为 per-env 数组：
# expand_model_fields 的工程语义
# 扩展前：body_mass shape = [num_bodies] (所有 env 共享)
# 扩展后：body_mass shape = [num_envs, num_bodies] (每个 env 独立)
# 这在 EventManager 初始化时自动完成：
# 1. EventManager 扫描所有事件配置
# 2. 找到所有需要随机化的 model 字段
# 3. 调用 sim.expand_model_fields() 扩展这些字段
# 4. 重建 CUDA capture graph（关键！）
expand_model_fields 的完整实现逻辑（简化自 sim/randomization.py）：
# expand_model_fields 核心实现
def expand_model_fields(sim, field_names):
"""把共享 model 字段扩展为 per-env 数组。"""
for field_name in field_names:
original = getattr(sim.model, field_name)
if field_name in sim._expanded_fields:
continue # 已经扩展过了
# 创建 per-env 版本
if original.ndim == 1:
expanded = original.unsqueeze(0).repeat(sim.num_worlds, 1)
elif original.ndim == 2:
expanded = original.unsqueeze(0).repeat(sim.num_worlds, 1, 1)
else:
raise ValueError(f"Unsupported ndim for {field_name}: {original.ndim}")
# 替换 model 字段
setattr(sim.model, field_name, expanded)
sim._expanded_fields.add(field_name)
# 保存默认值快照（用于 scale/add 操作的基准）
sim._default_snapshot[field_name] = expanded.clone()
# 必须在所有字段扩展后重新 capture Graph
sim._graph_captured = False
print(f"Expanded fields: {field_names}, total: {sim._expanded_fields}")
CUDA Graph 与字段扩展的时序约束。MuJoCo Warp 使用 CUDA Graph capture 来优化仿真性能——把一系列 GPU kernel 的调用顺序记录下来，后续 replay 时跳过 CPU 端的调度开销。expand_model_fields() 替换了 model 的 GPU 数组（从共享到 per-env），必须在 CUDA Graph capture 之前完成。
mjlab 的初始化时序保证了这一点：
1. env.__init__()
2. EventManager.__init__() → 收集所有 event 需要的字段名
3. sim.expand_model_fields(all_needed_fields) ← 字段扩展
4. startup events 执行（写入 per-env 初始值）
5. sim.capture_graph() ← CUDA Graph capture
6. 训练循环开始（Graph replay 正确读取 per-env 数组）
步骤 3 必须在步骤 5 之前。如果在 capture 之后扩展字段，Graph 失效：
# ❌ 错误：在 capture 后扩展
sim.capture_graph() # Graph 记录了 body_mass 的旧指针
sim.expand_model_fields(["body_mass"]) # 替换了指针
# Graph replay 读旧指针 → DR 值被忽略
# ✅ 正确：mjlab 自动在 capture 前扩展
# 用户只需在 EventTermCfg 中声明需要的字段
# EventManager 自动收集并在 init 时扩展
DR 核心写入引擎 ⭐⭐
当 DR 事件触发时，值如何写入 per-env 数组？以 randomize_body_mass 为例：
# DR 写入核心逻辑（简化自 dr/_core.py）
def _randomize_model_field(
sim, field_name, env_ids,
distribution_params, operation, distribution
):
"""DR 核心引擎：采样值并写入 per-env model 字段。"""
# 1. 获取 per-env 数组
field = getattr(sim.model, field_name)
assert field.ndim >= 2, f"Field {field_name} not expanded! Run expand first."
# 2. 获取默认值（scale/add 操作的基准）
default = sim._default_snapshot[field_name]
# 3. 采样
lo, hi = distribution_params
shape = (len(env_ids),) + field.shape[1:]
if distribution == "uniform":
samples = torch.rand(shape, device=sim.device) * (hi - lo) + lo
elif distribution == "log_uniform":
log_lo = torch.log(torch.tensor(lo, device=sim.device))
log_hi = torch.log(torch.tensor(hi, device=sim.device))
samples = torch.exp(
torch.rand(shape, device=sim.device) * (log_hi - log_lo) + log_lo
)
elif distribution == "gaussian":
mean, std = lo, hi
samples = torch.randn(shape, device=sim.device) * std + mean
# 4. 应用操作
if operation == "scale":
new_values = default[env_ids] * samples
elif operation == "add":
new_values = default[env_ids] + samples
elif operation == "abs":
new_values = samples
# 5. 关键：原地写入（不替换指针！CUDA Graph 安全）
field[env_ids] = new_values
# 注意：field[env_ids] = ... 是 tensor 的 __setitem__
# 它修改数组内容，不改变 GPU 指针地址
# CUDA Graph replay 仍然有效
# 6. 按需重算派生量
if field_name in {"body_mass", "body_ipos", "body_inertia"}:
sim.recompute_constants(env_ids)
步骤 5 的原地写入是整个 DR 链路中最关键的工程约束。field[env_ids] = new_values 修改数组内容但不改变 GPU 指针地址，CUDA Graph replay 仍然读取同一块内存——只是内容已更新。如果不小心写成 field = new_values（替换整个 tensor），指针变了，Graph 失效。
per-env 存储的显存分析 ⭐⭐
# 显存估算脚本
def estimate_dr_memory(num_envs, num_bodies, expanded_fields):
"""估算 DR per-env 存储的额外显存。"""
field_sizes = {
"body_mass": num_bodies * 1, # [num_bodies] float32
"body_inertia": num_bodies * 6, # [num_bodies, 6]
"body_ipos": num_bodies * 3, # [num_bodies, 3]
"geom_friction": num_bodies * 3, # [num_geoms, 3]
"dof_damping": num_bodies * 1, # [num_dofs] float32
"dof_armature": num_bodies * 1, # [num_dofs]
}
total_bytes = 0
for field in expanded_fields:
if field in field_sizes:
field_bytes = num_envs * field_sizes[field] * 4 # float32
total_bytes += field_bytes
print(f" {field}: {field_bytes / 1024:.1f} KB")
total_mb = total_bytes / (1024 * 1024)
print(f" Total DR memory: {total_mb:.1f} MB")
return total_mb
# Go1: 12 bodies, 4096 envs, 扩展 mass + inertia + friction
# estimate_dr_memory(4096, 12, ["body_mass", "body_inertia", "geom_friction"])
# → 约 2-3 MB，可忽略
RecomputeLevel 与性能影响 ⭐⭐
MuJoCo 的模型中，许多量是从基础参数派生的。修改基础参数后，可能需要调用 recompute_constants() 来更新派生量：
| 等级 | 含义 | 典型字段 | 性能影响 |
|---|---|---|---|
none |
无需重算 | geom_friction, dof_damping |
几乎为零 |
set_const_0 |
重算 armature、qpos0 | dof_armature, qpos0 |
中等 |
set_const |
完整重算所有派生常量 | body_mass, body_ipos |
较大 |
工程含义：不要在 step 或高频 interval 模式中触发 set_const 级别的字段修改。质量随机化（set_const 级别）应放在 startup 或 reset 模式——每 episode 只触发一次，性能影响可接受。
性能测量方法：
# 测量 DR 的性能开销
import time
def measure_dr_overhead(env, num_steps=1000):
"""测量有/无 DR 的训练吞吐量差异。"""
actions = torch.zeros(env.num_envs, env.action_manager.total_action_dim,
device=env.device)
# 热身
for _ in range(50):
env.step(actions)
# 测量
start = time.perf_counter()
for _ in range(num_steps):
env.step(actions)
elapsed = time.perf_counter() - start
steps_per_sec = num_steps * env.num_envs / elapsed
print(f" Steps/s: {steps_per_sec:.0f}")
return steps_per_sec
# 使用：
# 1. 先关闭所有 DR，测量 baseline steps/s
# 2. 打开 DR，测量新的 steps/s
# 3. 如果下降 >30%，检查是否有 set_const 字段在高频模式
⚠️ 常见陷阱
⚠️ 编程陷阱：pseudo-inertia 的 alpha 范围过大导致 NaN。 Cholesky 分解要求输入矩阵正定。如果 alpha 范围太极端（如 0.01 到 100），数值精度可能导致分解失败。推荐从 (0.9, 1.1) 开始，逐步扩大到 (0.7, 1.5)。
练习
- [计算题] Go1 标称 base 质量 12.5 kg。使用 pseudo-inertia 的 alpha=(0.8, 1.2)。计算随机化后质量的范围。解释为什么这比直接
add(-2.5, 2.5)更好。 - [编程题] 实现一个简单的 pseudo-inertia 验证脚本：给定一组 (m, c, I) 参数，检查它们是否物理一致（pseudo-inertia 矩阵正定）。
8.5 外力扰动 ⭐⭐
这一节解决什么问题：如何在训练中模拟外部推力？频率和强度怎么选？
Push vs Impulse ⭐⭐
两种外力扰动模式：
Push（速度注入）：直接设置 base 的线/角速度。物理上等价于瞬间施加一个冲量。简单高效，是 locomotion 训练的标准选择。
# push_by_setting_velocity 的核心逻辑
def push_by_setting_velocity(env, env_ids, velocity_range):
"""通过直接设置 base velocity 模拟外部推力。"""
robot = env.scene.robot
# 采样随机速度
vel_range = velocity_range
random_vel = torch.zeros(len(env_ids), 6, device=env.device)
for i, key in enumerate(["x", "y", "z", "roll", "pitch", "yaw"]):
if key in vel_range:
lo, hi = vel_range[key]
random_vel[:, i] = torch.rand(len(env_ids), device=env.device) * (hi - lo) + lo
# 直接设置 base velocity
robot.write_root_velocity_to_sim(random_vel, env_ids)
Impulse（力/力矩注入）：在指定 body 上施加外力，持续若干步后清零。更物理真实，但需要管理力的生命周期——通常需要 step 模式。
# apply_body_impulse 的核心逻辑（简化）
def apply_body_impulse(env, env_ids, body_name, force_range, duration_steps=5):
"""施加脉冲力，持续 duration_steps 步后自动清零。"""
robot = env.scene.robot
body_idx = robot.find_bodies(body_name)
# 采样随机力
random_force = torch.zeros(len(env_ids), 6, device=env.device)
# ... 从 force_range 采样
# 施加外力（需要在后续步骤中清零）
robot.set_external_force_and_torque(random_force, body_idx, env_ids)
# 注册一个 step 事件在 duration_steps 后清零
# 或使用 step 模式的计数器管理
频率和强度的选择 ⭐⭐
| 参数 | 推荐值 | 理由 |
|---|---|---|
| Push 间隔 | 10-15 秒 | 让策略有恢复时间 |
| Push 线速度 | ±1.0 m/s | 约等于中等力度人推一下 |
| Push 角速度 | ±0.25 rad/s | 较小的旋转扰动 |
| Impulse 力 | 0-50 N | 依赖机器人质量 |
| Impulse 持续 | 3-10 步 | 太短策略感觉不到，太长变成持续力 |
分阶段引入 push 的原因：如果策略还不会走路就施加 push，策略在每次被推后都会摔倒——reward 信号几乎全是负的，策略无法从中学到"怎样抵抗推力"。正确做法是先在无 push 环境中训练到基本稳定，再引入 push。这与 Ch06 的 curriculum 思想一致——"先学基础再加难度"。
⚠️ 常见陷阱
⚠️ 编程陷阱：push 过早引入导致策略学到"不动"。 如果 push 在训练第一步就激活，策略发现"站着不动被推倒 → 摔倒 penalty"和"走路时被推倒 → 更大的 penalty"——不动反而更安全。
💡 概念误区：认为 impulse 比 push 更"真实"。 物理上 impulse 确实更精确（通过力而非直接修改速度），但 push_by_setting_velocity 在工程上更常用——因为它不需要管理力的生命周期，且与 MuJoCo 和 PhysX 的不同力模型无关。大多数 locomotion 论文（Rudin 2022, extreme-parkour, Walk-These-Ways）都使用 push_by_setting_velocity。
Push 强度选择的经验法则 ⭐⭐
push velocity 的选择需要与机器人质量和步态速度匹配：
| 机器人 | 质量 | 推荐 push velocity | 等效冲量 | 说明 |
|---|---|---|---|---|
| Go1 | ~12 kg | ±1.0 m/s | ~12 N·s | 相当于中等力度手推 |
| Go2 | ~15 kg | ±1.0 m/s | ~15 N·s | 与 Go1 类似 |
| ANYmal-C | ~50 kg | ±0.5 m/s | ~25 N·s | 大型四足需要更大绝对冲量 |
| G1 (人形) | ~35 kg | ±0.3 m/s | ~10.5 N·s | 人形平衡更脆弱，推力要小 |
| H1 (人形) | ~47 kg | ±0.3 m/s | ~14 N·s | 同上 |
经验法则：push velocity 应该使策略在训练充分后有 >80% 的概率在 1 秒内恢复平衡。如果恢复率 <50%，push 太强；如果恢复率 >98%，push 太弱。
Push 频率与训练阶段的交互 ⭐⭐
push 的 interval_range_s 需要与训练阶段配合：
# 早期训练（iteration < 500）：长间隔，给策略学习基本步态的时间
events["push_early"] = EventTermCfg(
func=mdp.push_by_setting_velocity,
mode="interval",
interval_range_s=(20.0, 30.0), # 20-30 秒推一次
params={
"asset_cfg": SceneEntityCfg("robot"),
"velocity_range": {"x": (-0.5, 0.5), "y": (-0.5, 0.5)},
},
)
# 后期训练（iteration > 500）：短间隔，增加挑战
events["push_late"] = EventTermCfg(
func=mdp.push_by_setting_velocity,
mode="interval",
interval_range_s=(8.0, 12.0), # 8-12 秒推一次
params={
"asset_cfg": SceneEntityCfg("robot"),
"velocity_range": {"x": (-1.0, 1.0), "y": (-1.0, 1.0)},
},
)
实现 push curriculum 需要在 CurriculumManager 中根据 iteration 切换 push 配置。或者更简单地：Phase 0-3 不配 push，Phase 4 开始 resume 训练时加入 push（与下节六阶段方案一致）。
练习
- [设计题] 为一个 12.5 kg 的四足机器人设计 push 配置。计算 1 m/s 的 push velocity 等价于多大的冲量力（假设作用时间 0.02 s = 一个 policy step）。
- [实验题] 对比以下三种 push 配置的训练效果：(a) 无 push (b) 从第 0 步开始 push (c) 从第 500 iteration 开始 push。
上节覆盖了"随机化什么"（摩擦/质量/增益/惯量/外力/噪声/延迟）。接下来的核心问题是"按什么顺序随机化"——分阶段 DR 策略。
8.6 分阶段 DR 策略 ⭐⭐⭐
这一节解决什么问题：如何系统地、可调试地引入 DR？
为什么要分阶段 ⭐⭐
一次性打开所有 DR 的问题：训练失败时无法归因。是摩擦太低？质量偏差太大？编码器偏置太强？还是它们的组合？如果同时变化 10 个参数，每个失败视频可能对应无数种根因组合。
分阶段引入的核心原则是单变量递进——每个阶段只增加一类 DR，验证通过后再加下一类。这样每个阶段的失败都能清晰归因。
六阶段方案 ⭐⭐⭐
Phase 0: Clean Baseline（无 DR）
├── 目标：验证 reward/obs/action 设计正确
├── 检查：tracking reward > 0.7, play video 正常
└── 如果失败：问题在 Ch05/Ch06/Ch07，不是 DR
Phase 1: 动力学随机化（startup）
├── 添加：foot_friction, base_mass, com_offset
├── 目标：策略适应不同硬件参数
├── 检查：tracking reward > 0.6（允许略降）
└── 如果失败：收窄范围，逐项排查
Phase 2: 执行器随机化（reset）
├── 添加：PD gains, joint damping, armature
├── 目标：策略适应不同执行器特性
├── 检查：action std 不要过大
└── 如果失败：检查 PD gain 范围是否过大
Phase 3: 感知随机化（startup/step）
├── 添加：encoder_bias, obs noise/corruption
├── 目标：策略适应传感器不确定性
├── 检查：tracking 在有噪声时不崩溃
└── 如果失败：噪声幅度太大
Phase 4: 外力扰动（interval）
├── 添加：push_velocity, gravity_tilt
├── 目标：策略能从扰动中恢复
├── 检查：被推后 1-2 秒内恢复正常步态
└── 如果失败：push 太强或策略 baseline 不够稳
Phase 5: 鲁棒性验证
├── 在固定参数网格上评估（见 8.7 节）
├── 在极端参数组合上压力测试
└── 通过 → 准备 sim-to-real（Ch23）
各阶段的完整 CLI 命令 ⭐⭐
# ============= Phase 0: Clean Baseline =============
uv run train Mjlab-Velocity-Flat-Unitree-Go1 \
--env.scene.num-envs 4096 --agent.max-iterations 300 \
--agent.run-name phase0_clean --agent.seed 42 \
--env.events.foot-friction None \
--env.events.base-mass None \
--env.events.push-robot None
# 验证：tracking > 0.7, play video 正常
# ============= Phase 1: 动力学随机化 =============
uv run train Mjlab-Velocity-Flat-Unitree-Go1 \
--env.scene.num-envs 4096 --agent.max-iterations 500 \
--agent.run-name phase1_dynamics --agent.seed 42 \
--agent.resume True --agent.load-run phase0_clean
# foot_friction 和 base_mass 使用默认配置（已在 cfg 中）
# push 仍然关闭
# ============= Phase 2: 执行器随机化 =============
# 在 cfg 中添加 PD gain randomization
uv run train Mjlab-Velocity-Flat-Unitree-Go1 \
--env.scene.num-envs 4096 --agent.max-iterations 800 \
--agent.run-name phase2_actuator --agent.seed 42 \
--agent.resume True --agent.load-run phase1_dynamics
# ============= Phase 3: 感知随机化 =============
# 在 cfg 中添加 encoder_bias 和 obs noise
uv run train Mjlab-Velocity-Flat-Unitree-Go1 \
--env.scene.num-envs 4096 --agent.max-iterations 1000 \
--agent.run-name phase3_perception --agent.seed 42 \
--agent.resume True --agent.load-run phase2_actuator
# ============= Phase 4: 外力扰动 =============
uv run train Mjlab-Velocity-Flat-Unitree-Go1 \
--env.scene.num-envs 4096 --agent.max-iterations 1500 \
--agent.run-name phase4_push --agent.seed 42 \
--agent.resume True --agent.load-run phase3_perception
# 此时 push 事件自动从 cfg 中激活
# ============= Phase 5: 鲁棒性验证 =============
# 在不同摩擦下 play
for FRICTION in 0.3 0.5 0.8 1.0 1.5; do
uv run play Mjlab-Velocity-Flat-Unitree-Go1 \
--agent.load-run phase4_push --num-envs 4 \
--env.events.foot-friction.params.friction-range "($FRICTION, $FRICTION)"
done
Isaac Lab 等价命令：
# Phase 0
python scripts/reinforcement_learning/rsl_rl/train.py --task Isaac-Velocity-Flat-Anymal-C-v0 \
--num_envs 4096 --max_iterations 300 --seed 42
# Phase 1-4: 通过修改 cfg 文件或 hydra overrides
# Isaac Lab 的事件配置在 @configclass 中，需要修改 Python 文件
# Phase 5: Play
python scripts/reinforcement_learning/rsl_rl/play.py --task Isaac-Velocity-Flat-Anymal-C-v0
各阶段的诊断脚本 ⭐⭐⭐
# phase_diagnostic.py — 各阶段专用诊断
import torch
def diagnose_phase0(env, policy, num_steps=200):
"""Phase 0 诊断：验证 clean baseline 质量。"""
obs = env.reset()
actor_obs = obs.get("actor", obs.get("policy"))
tracking_rewards = []
fall_count = 0
total_steps = 0
for _ in range(num_steps):
with torch.inference_mode():
actions = policy(actor_obs)
obs_dict, rewards, terminated, truncated, extras = env.step(actions)
actor_obs = obs_dict.get("actor", obs_dict.get("policy"))
total_steps += 1
if terminated.any():
fall_count += terminated.sum().item()
fall_rate = fall_count / (total_steps * env.num_envs)
print(f"Phase 0 Diagnostic:")
print(f" Fall rate: {fall_rate:.1%}")
if fall_rate > 0.05:
print(f" ⚠️ Fall rate too high for Phase 0!")
print(f" → 问题在 Ch05/Ch06/Ch07，不是 DR")
print(f" → 先修复 baseline 再加 DR")
return False
print(f" ✅ Clean baseline healthy, ready for Phase 1")
return True
def diagnose_phase1(env, policy, num_steps=200):
"""Phase 1 诊断：检查动力学 DR 对训练的影响。"""
obs = env.reset()
actor_obs = obs.get("actor", obs.get("policy"))
# 检查 DR 是否生效
em = env.event_manager
active_events = {name: cfg for name, (_, cfg) in em._mode_term_cfgs.get("startup", {}).items()}
print(f"Phase 1 Diagnostic:")
print(f" Active startup events: {list(active_events.keys())}")
if not active_events:
print(f" ⚠️ No startup events found! DR may not be configured.")
return False
# 检查 expanded fields
if hasattr(env.sim, 'expanded_fields'):
print(f" Expanded fields: {env.sim.expanded_fields}")
# 运行短 rollout 检查 tracking 下降幅度
for _ in range(num_steps):
with torch.inference_mode():
actions = policy(actor_obs)
obs_dict, rewards, _, _, _ = env.step(actions)
actor_obs = obs_dict.get("actor", obs_dict.get("policy"))
print(f" ✅ Phase 1 DR active, check TensorBoard for tracking regression")
return True
def diagnose_phase4_push(env, policy, num_steps=500):
"""Phase 4 诊断：检查 push recovery 能力。"""
obs = env.reset()
actor_obs = obs.get("actor", obs.get("policy"))
push_count = 0
recovery_count = 0
fall_after_push = 0
for step in range(num_steps):
with torch.inference_mode():
actions = policy(actor_obs)
obs_dict, rewards, terminated, truncated, extras = env.step(actions)
actor_obs = obs_dict.get("actor", obs_dict.get("policy"))
# 检查 push 事件是否触发（通过 velocity 突变检测）
base_vel = env.scene.robot.data.root_lin_vel_b[:, :2]
vel_mag = base_vel.norm(dim=1)
if terminated.any():
fall_after_push += terminated.sum().item()
print(f"Phase 4 Diagnostic:")
print(f" Falls in {num_steps} steps: {fall_after_push}")
if fall_after_push > num_steps * env.num_envs * 0.1:
print(f" ⚠️ High fall rate after push!")
print(f" → 减小 push velocity range")
print(f" → 或延后 push 引入时机")
else:
print(f" ✅ Push recovery acceptable")
分阶段 DR 实验日志模板 ⭐⭐
# Phased DR Experiment Log
## Setup
- Robot: Go1 / G1 / ANYmal-C
- Framework: mjlab / Isaac Lab
- GPU: ________________
- Date: ________________
## Phase Results
| Phase | DR Items | Iterations | tracking | fall_rate | wall_time |
|:-----:|:---------|:----------:|:--------:|:---------:|:---------:|
| 0 | None | 300 | 0.78 | 2% | 3 min |
| 1 | friction, mass | 500 | 0.72 | 5% | 5 min |
| 2 | + PD gains | 800 | 0.68 | 8% | 8 min |
| 3 | + encoder_bias | 1000 | 0.65 | 10% | 10 min |
| 4 | + push | 1500 | 0.62 | 12% | 15 min |
## Robustness Grid
| Friction | tracking | fall_rate |
|:--------:|:--------:|:---------:|
| 0.3 | ____ | ____ |
| 0.5 | ____ | ____ |
| 0.8 | ____ | ____ |
| 1.0 | ____ | ____ |
| 1.5 | ____ | ____ |
## Analysis
- Phase 0→1 tracking 下降幅度: ____% (可接受: <15%)
- Phase 1→2 下降: ____% (可接受: <10%)
- 最弱参数区域: ________________
- 是否需要收窄范围: ________________
## Decision
- [ ] 准备进入 Ch09 (Teacher-Student)
- [ ] 需要调整 DR 范围
- [ ] 需要增加训练 iteration
⚠️ 常见陷阱
🧠 思维陷阱：跳过 Phase 0 直接从 Phase 4 开始。 如果不先验证 clean baseline，DR 引入的任何问题都无法归因——你不知道是 reward 设计错了还是 DR 范围太大。Phase 0 的投资回报最高。
练习
- [设计题] 为 G1 人形机器人设计完整的六阶段 DR 方案。G1 有 29 个自由度（vs Go1 的 12 个），需要额外考虑什么？
- [分析题] Phase 1 中低摩擦下策略很差。列出 3 种修复方向及其优劣。
8.7 DR 评估方法论 ⭐⭐
这一节解决什么问题：训练完成后如何系统地验证策略的鲁棒性？
固定参数网格评估 ⭐⭐
训练时 DR 在分布上随机采样——但评估时应该在固定的参数网格上系统测试。这样能发现策略在哪些参数区域表现好、哪些区域弱。
# dr_evaluation.py — 完整的 DR 鲁棒性评估脚本
import torch
import numpy as np
from collections import defaultdict
class DRRobustnessEvaluator:
"""系统化评估策略在不同物理参数下的鲁棒性。"""
def __init__(self, env, policy):
self.env = env
self.policy = policy
self.results = defaultdict(dict)
def evaluate_single_param(self, param_name, param_values,
num_episodes=50, max_steps=1000):
"""在单个参数的固定值网格上评估。"""
print(f"\n{'='*50}")
print(f"Evaluating: {param_name}")
print(f"{'='*50}")
for value in param_values:
# 固定参数（关闭该参数的随机化）
self._set_fixed_param(param_name, value)
# 运行评估
metrics = self._run_episodes(num_episodes, max_steps)
self.results[param_name][value] = metrics
print(f" {param_name}={value:.3f}: "
f"tracking={metrics['tracking']:.3f}, "
f"fall_rate={metrics['fall_rate']:.0%}, "
f"avg_speed={metrics['avg_speed']:.2f} m/s")
return self.results[param_name]
def _run_episodes(self, num_episodes, max_steps):
"""运行指定数量的 episode 并收集 metrics。"""
tracking_rewards = []
fall_count = 0
speeds = []
obs = self.env.reset()
actor_obs = obs.get("actor", obs.get("policy"))
episode_count = 0
step_count = 0
current_tracking = 0.0
while episode_count < num_episodes and step_count < max_steps * num_episodes:
with torch.inference_mode():
actions = self.policy(actor_obs)
obs_dict, rewards, terminated, truncated, extras = self.env.step(actions)
actor_obs = obs_dict.get("actor", obs_dict.get("policy"))
step_count += 1
# 收集速度数据
base_vel = self.env.scene.robot.data.root_lin_vel_b[:, :2]
speeds.append(base_vel.norm(dim=1).mean().item())
# 检查 episode 结束
done = terminated | truncated
if done.any():
episode_count += done.sum().item()
# terminated 形如 [num_envs]，直接 sum 统计本步真正摔倒（true termination）的 env 数；
# 旧写法 terminated.any(dim=0).sum() 会塌缩成标量，每步最多 +1，严重低估摔倒数
fall_count += terminated.sum().item()
return {
"tracking": np.mean(speeds[-100:]) if speeds else 0,
"fall_rate": fall_count / max(episode_count, 1),
"avg_speed": np.mean(speeds) if speeds else 0,
"num_episodes": episode_count,
}
def _set_fixed_param(self, param_name, value):
"""固定某个物理参数到指定值（关闭随机化）。"""
# 这里的实现取决于具体框架
# mjlab: 直接修改 model field
# Isaac Lab: 通过 EventManager 设置固定值
pass
def evaluate_friction_grid(self):
"""标准摩擦评估网格。"""
return self.evaluate_single_param(
"geom_friction",
[0.2, 0.4, 0.6, 0.8, 1.0, 1.2, 1.5, 2.0],
)
def evaluate_mass_grid(self):
"""标准质量评估网格。"""
return self.evaluate_single_param(
"body_mass_offset",
[-3.0, -1.5, 0.0, 1.5, 3.0, 5.0],
)
def evaluate_push_recovery(self, push_velocities, recovery_steps=100):
"""评估策略的推力恢复能力。"""
print(f"\nPush Recovery Evaluation")
print(f"{'='*50}")
for push_vel in push_velocities:
obs = self.env.reset()
actor_obs = obs.get("actor", obs.get("policy"))
# 等待稳定
for _ in range(50):
with torch.inference_mode():
actions = self.policy(actor_obs)
obs_dict, _, _, _, _ = self.env.step(actions)
actor_obs = obs_dict.get("actor", obs_dict.get("policy"))
# 施加 push
push = torch.zeros(self.env.num_envs, 6, device=self.env.device)
push[:, 0] = push_vel # x 方向 push
self.env.scene.robot.write_root_velocity_to_sim(push)
# 测量恢复时间
recovered = False
for step in range(recovery_steps):
with torch.inference_mode():
actions = self.policy(actor_obs)
obs_dict, _, terminated, _, _ = self.env.step(actions)
actor_obs = obs_dict.get("actor", obs_dict.get("policy"))
if terminated.any():
print(f" push={push_vel:.1f} m/s: FELL at step {step}")
break
# 检查是否恢复（速度回到命令附近）
base_vel = self.env.scene.robot.data.root_lin_vel_b[:, :2]
vel_error = base_vel.norm(dim=1).mean().item()
if vel_error < 0.3 and step > 10:
print(f" push={push_vel:.1f} m/s: recovered at step {step} "
f"({step * 0.02:.2f}s)")
recovered = True
break
if not recovered and step == recovery_steps - 1:
print(f" push={push_vel:.1f} m/s: did not recover in {recovery_steps} steps")
def generate_report(self):
"""生成完整的鲁棒性评估报告。"""
report = []
report.append("DR Robustness Evaluation Report")
report.append("=" * 60)
for param_name, values_dict in self.results.items():
report.append(f"\n--- {param_name} ---")
for value, metrics in sorted(values_dict.items()):
status = "✅" if metrics['fall_rate'] < 0.1 else "⚠️" if metrics['fall_rate'] < 0.3 else "🔴"
report.append(f" {status} {value:.3f}: "
f"tracking={metrics['tracking']:.3f}, "
f"fall_rate={metrics['fall_rate']:.0%}")
report_text = "\n".join(report)
print(report_text)
return report_text
# 使用示例
# evaluator = DRRobustnessEvaluator(env, trained_policy)
# evaluator.evaluate_friction_grid()
# evaluator.evaluate_mass_grid()
# evaluator.evaluate_push_recovery([0.5, 1.0, 1.5, 2.0, 3.0])
# evaluator.generate_report()
评估的三个层次 ⭐⭐
| 层次 | 含义 | 测试方法 | 判据 |
|---|---|---|---|
| 分布内 | 训练 DR 范围内 | 在范围内均匀采样网格 | tracking > 0.6, fall < 10% |
| 边界 | DR 范围的极值 | 专门测试最低/最高值 | tracking > 0.4, fall < 20% |
| 分布外 | 超出 DR 范围 ±20% | 测试范围外参数 | 平缓退化而非突然崩溃 |
分布外测试为什么重要？ 真实世界的参数不保证落在你的 DR 范围内。了解策略在分布外的退化行为（是平缓下降还是突然崩溃）对部署决策非常有价值。如果策略在摩擦 0.5 处表现良好但在 0.45 处突然摔倒，说明策略没有学到真正的鲁棒性——它只是记住了 [0.5, 1.25] 范围内的最优行为。
DR 效果的 A/B 对比脚本 ⭐⭐
# dr_ab_test.py — 对比有/无 DR 策略的鲁棒性
def compare_dr_policies(env, policy_with_dr, policy_no_dr,
param_name, param_values):
"""A/B 对比有/无 DR 策略在不同参数下的表现。"""
eval_dr = DRRobustnessEvaluator(env, policy_with_dr)
eval_no_dr = DRRobustnessEvaluator(env, policy_no_dr)
results_dr = eval_dr.evaluate_single_param(param_name, param_values)
results_no_dr = eval_no_dr.evaluate_single_param(param_name, param_values)
print(f"\n{'='*60}")
print(f"A/B Comparison: {param_name}")
print(f"{'='*60}")
print(f"{'Value':>8s} | {'No DR tracking':>15s} | {'DR tracking':>12s} | {'Δ':>8s}")
print(f"{'-'*50}")
for value in param_values:
t_no_dr = results_no_dr.get(value, {}).get('tracking', 0)
t_dr = results_dr.get(value, {}).get('tracking', 0)
delta = t_dr - t_no_dr
indicator = "↑" if delta > 0.05 else "↓" if delta < -0.05 else "≈"
print(f" {value:6.2f} | {t_no_dr:13.3f} | {t_dr:10.3f} | {delta:+6.3f} {indicator}")
DR 与 Curriculum 的交互分析 ⭐⭐
DR 和 Curriculum Learning（Ch06）都会改变训练任务——它们可能互相干扰。
| 交互模式 | 效果 | 建议 |
|---|---|---|
| DR 范围突然扩大 + Curriculum 同时升级 | 双重难度跳变，策略可能崩溃 | 不要同时扩大 DR 和 Curriculum |
| DR 范围固定 + Curriculum 逐步升级 | 正常：在固定参数分布上逐步学更难任务 | 推荐默认组合 |
| DR 逐步扩大 + Curriculum 固定 | 正常：在固定任务上逐步适应更宽参数 | 适合分阶段 DR |
| 低摩擦 DR + 粗糙地形 Curriculum | 极度困难：低摩擦 + 粗糙地形几乎不可学 | 先学地形再加摩擦 DR |
本质洞察：DR 和 Curriculum 都是在改变训练数据的分布——但 DR 改变的是物理参数分布（让策略更鲁棒），Curriculum 改变的是任务难度分布（让策略更有能力）。它们是正交的维度，不应该同时大幅变化。类似于健身训练中不应该同时增加重量和增加组数——每次只改变一个训练变量才能评估效果。
8.8 ASAP Delta Action Model 精读 ⭐⭐⭐
这一节解决什么问题：当传统 DR 不够时的替代方案——用真机数据学习一个残差模型来修正 sim-to-real gap。
动机：DR 的局限性 ⭐⭐
DR 通过扩大训练分布来覆盖真实参数，但这有两个根本限制：(1) 你必须知道哪些参数需要随机化——如果真实世界的不确定性来源不在你的随机化列表中（如传动间隙、电机非线性、接触模型差异），DR 无法覆盖；(2) 过大的 DR 范围导致策略过于保守——ASAP 论文明确指出 DR 是 agile motion 的瓶颈。
ASAP 五阶段管线 ⭐⭐⭐
Stage A: Pretrain（仿真中训练 base policy）
└─ 使用 DR 在 Isaac Gym / mjlab 中训练 motion tracking policy
DR 参数：标准 locomotion DR（摩擦、质量、PD 增益）
Stage B: Real Rollout（真机采集数据）
└─ 部署 Stage A 策略到 Unitree G1
使用 MoCap 系统记录 (s_t, a_t, s_{t+1})_real
Stage C: Train δ（训练 delta action model）
└─ 在仿真中 replay 相同的 a_t
最小化 ||s_{t+1}^sim - s_{t+1}^real||
得到 δ(s, a) = a_corrected - a_original
Stage D: Finetune（用 delta 修正仿真器后继续 RL 训练）
└─ 冻结 δ，在仿真中插入 a_effective = a_policy + δ(s, a)
用 PPO 继续训练（仿真器现在更接近真实世界）
Stage E: Deploy（直接部署，不需要 δ）
└─ 只部署 Stage D 的策略，不部署 delta model
因为 delta 已经通过 finetune 内化到策略中
delta model 的网络结构和训练非常简单——关键是数据采集和对齐。
# ASAP delta action model 的核心思想（概念示意，非官方实现——官方用 open-loop RL/PPO，详见本节末尾说明）
import torch
import torch.nn as nn
class DeltaActionModel(nn.Module):
"""学习仿真和真实世界之间的 action 修正。
输入：state + action 拼接
输出：action 的修正量 δa
"""
def __init__(self, state_dim: int, action_dim: int, hidden_dims=(256, 128)):
super().__init__()
input_dim = state_dim + action_dim
layers = []
prev_dim = input_dim
for h in hidden_dims:
layers.extend([nn.Linear(prev_dim, h), nn.ELU()])
prev_dim = h
layers.append(nn.Linear(prev_dim, action_dim))
self.mlp = nn.Sequential(*layers)
def forward(self, state, action):
"""返回 action 修正量。"""
x = torch.cat([state, action], dim=-1)
return self.mlp(x)
class DeltaTrainer:
"""训练 delta action model 的管线。"""
def __init__(self, delta_model, sim_env, lr=1e-3):
self.delta = delta_model
self.sim = sim_env
self.optimizer = torch.optim.Adam(delta_model.parameters(), lr=lr)
def train_from_real_data(self, real_trajectories, num_epochs=100, batch_size=256):
"""从真机采集的轨迹数据训练 delta model。
Args:
real_trajectories: list of (s_t, a_t, s_{t+1}_real) tuples
"""
# 把真机数据转换为 tensor
states = torch.stack([t[0] for t in real_trajectories])
actions = torch.stack([t[1] for t in real_trajectories])
next_states_real = torch.stack([t[2] for t in real_trajectories])
for epoch in range(num_epochs):
# 随机打乱
indices = torch.randperm(len(states))
total_loss = 0.0
for start in range(0, len(states), batch_size):
idx = indices[start:start + batch_size]
s = states[idx]
a = actions[idx]
s_next_real = next_states_real[idx]
# 计算 delta
delta_a = self.delta(s, a)
# 在仿真中 replay a + delta_a
corrected_action = a + delta_a
s_next_sim = self.sim.simulate_step(s, corrected_action)
# Loss: 让仿真的 next state 匹配真实的 next state
loss = nn.functional.mse_loss(s_next_sim, s_next_real)
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
total_loss += loss.item()
if (epoch + 1) % 10 == 0:
avg_loss = total_loss / (len(states) / batch_size)
print(f"Epoch {epoch+1}: loss={avg_loss:.6f}")
def create_corrected_env(self):
"""创建用 delta 修正后的仿真环境（Stage D 用）。"""
# 在 env.step() 中插入 delta correction
# a_effective = a_policy + delta(s, a_policy)
pass
# Stage C 的工作流
# delta = DeltaActionModel(state_dim=48, action_dim=12)
# trainer = DeltaTrainer(delta, sim_env)
# trainer.train_from_real_data(real_data, num_epochs=200)
# torch.save(delta.state_dict(), "delta_model.pt")
ASAP 代码仓库结构（LeCAR-Lab/ASAP，截至 2026-06 核对官方仓库）：
LeCAR-Lab/ASAP/
├── humanoidverse/ # 核心：sim-agnostic 训练框架（基于 HumanoidVerse）
│ ├── train_agent.py # 统一训练入口（含 motion tracking 与 delta action 训练）
│ ├── envs/ # 环境定义（如 motion_tracking/）
│ ├── agents/ # RL agent（PPO 等）
│ └── config/ # 各阶段 Hydra 配置（如 algo/ppo_train_delta_a.yaml）
├── isaac_utils/ # Isaac 仿真工具
├── sim2real/
│ └── rl_policy/
│ └── listener_deltaa.py # 真机部署/数据采集
└── scripts/ # 数据处理与可视化脚本
⚠️ 注意：官方仓库没有根目录
asap/，也没有delta_model.py/train_delta.py这两个文件——delta action model 复用同一个humanoidverse/train_agent.py入口、靠+exp=...配置切换。delta model 的训练方式是 open-loop RL（PPO）（reward 最小化 sim/real 状态误差并带 action norm 惩罚），随后再 closed-loop 微调 policy；它不是对仿真 step 直接反传的监督式 MSE。本节上面的DeltaTrainer/simulate_step代码仅为概念示意，不对应官方实现，ASAP 也不使用 mjlab（支持 IsaacGym/IsaacSim/Genesis）。
与 DR 的互补关系 ⭐⭐
ASAP 和 DR 不是替代关系——它们互补：
| 方法 | 需要真机数据？ | 策略保守性 | 适用场景 |
|---|---|---|---|
| 只用 DR | 否 | 中-高 | 大多数 locomotion 任务 |
| DR + System ID | 少量（参数辨识） | 低-中 | 需要精确跟踪 |
| DR + ASAP delta | 中等（轨迹采集） | 低 | agile motion, 高精度 |
| 只用 delta（无 DR） | 大量 | 低但泛化差 | 不推荐 |
对大多数项目来说，只用 DR 就够了（Ch23 的 sim-to-real 案例）。只有在追求 agile 动作（如跳跃、翻滚、快速转向）时，DR 的保守性才成为瓶颈——此时考虑 ASAP。
最新替代方案：SPI-Active ⭐⭐
SPI-Active（Sobanbabu et al., arXiv 2505.14266, 2025）提出了另一种思路：不是用 delta model 修正仿真器，而是用大规模并行采样进行系统辨识——直接找到最接近真实世界的仿真参数，然后在这些精确参数上训练。SPI-Active 在 Unitree Go2 和 G1 上展示了比 DR 更精确的 velocity tracking 和 attitude tracking。这种方法需要的真机数据比 ASAP 更少，但需要能在仿真中高速并行搜索参数——恰好是 mjlab / Isaac Lab + GPU 仿真的优势场景。
⚠️ 常见陷阱
🧠 思维陷阱：认为"有了 delta model 就不需要 DR 了"。 delta model 修正的是 DR 无法覆盖的系统性偏差，二者通常互补：base policy 若完全不做任何鲁棒化，可能在真机上表现太差，连采集有意义的 real rollout 数据都做不到。（注意：ASAP 公开代码的训练命令默认 NO_domain_rand，因此"Stage A 一定用 DR"是配置相关的，不要当成论文硬性要求。）
练习
- [分析题] ASAP 的 Stage E 为什么不部署 delta model？如果部署了会有什么问题？
- [设计题] 如果你没有 MoCap 系统，如何修改 ASAP 的 Stage B？有哪些替代方案？
8.9 观测噪声与延迟随机化 ⭐⭐
这一节解决什么问题：除物理参数外，传感器噪声和通信延迟也是 sim-to-real gap 的重要来源。
观测噪声配置 ⭐⭐
真实传感器有噪声——IMU 有漂移、编码器有量化误差。如果策略只在 clean observation 上训练，部署时遇到噪声可能行为失控。mjlab 和 Isaac Lab 的 ObservationManager 都支持 per-term noise injection：
# ============= mjlab obs noise 配置 =============
observations = {
"actor": {
"joint_pos_rel": ObservationTermCfg(
func=mdp.joint_pos_rel,
noise=GaussianNoiseCfg(mean=0.0, std=0.01), # ±0.01 rad
),
"joint_vel": ObservationTermCfg(
func=mdp.joint_vel,
noise=GaussianNoiseCfg(mean=0.0, std=0.05), # ±0.05 rad/s
),
"base_ang_vel": ObservationTermCfg(
func=mdp.base_ang_vel,
noise=GaussianNoiseCfg(mean=0.0, std=0.02),
),
"projected_gravity": ObservationTermCfg(
func=mdp.projected_gravity,
noise=GaussianNoiseCfg(mean=0.0, std=0.01),
),
},
}
# ============= Isaac Lab obs noise 配置 =============
@configclass
class ObservationsCfg:
@configclass
class PolicyCfg:
joint_pos = ObsTerm(
func=mdp.joint_pos_rel,
noise=GaussianNoise(mean=0.0, std=0.01),
)
joint_vel = ObsTerm(
func=mdp.joint_vel,
noise=GaussianNoise(mean=0.0, std=0.05),
)
base_ang_vel = ObsTerm(
func=mdp.base_ang_vel,
noise=GaussianNoise(mean=0.0, std=0.02),
)
噪声标准差参考值：
| 传感器 | 典型噪声 | 推荐 std | 来源 |
|---|---|---|---|
| 关节编码器 | ±0.5° | 0.01 rad | Unitree Go1 spec |
| 关节速度（数值微分） | ±2-5% | 0.05 rad/s | 经验值 |
| IMU 角速度 | ±0.5-2°/s | 0.02 rad/s | MPU-6050 spec |
| IMU 加速度 | ±0.05 m/s² | 0.05 m/s² | 经验值 |
| projected gravity | ±1-3% | 0.01 | 从 IMU 派生 |
| 力传感器 | ±2-5% | 按比例 | 传感器 spec |
关键工程点：obs noise 只加在 actor obs 上——critic 保持 clean observation。这利用了 Ch05 中 asymmetric actor-critic 的设计——critic 看到的 clean obs 让 value function 更准确，而 actor 被迫学会在噪声条件下做出正确决策。
观测延迟模拟 ⭐⭐
真实控制系统的传感器有不同延迟——关节编码器 ~1ms，IMU ~3ms，深度相机 ~33ms。延迟意味着策略收到的不是当前状态，而是过去某一步的状态。
# mjlab obs delay 配置
observations = {
"actor": {
"joint_pos_rel": ObservationTermCfg(
func=mdp.joint_pos_rel,
noise=GaussianNoiseCfg(mean=0.0, std=0.01),
delay=UniformIntDelayCfg(min_steps=0, max_steps=2), # 0-2 步延迟
),
"depth_image": ObservationTermCfg(
func=mdp.depth_camera,
delay=UniformIntDelayCfg(min_steps=1, max_steps=3), # 相机延迟更大
),
},
}
延迟通过 history buffer 实现——ObservationManager 维护每个 term 的最近 N 步值，delay 从 buffer 中选取偏移。延迟步数在每个 episode reset 时重采样（reset 模式语义）。
编码器偏置——固定偏移量 ⭐⭐
装配误差导致的编码器偏置不是每步随机噪声，而是每 episode 固定的偏移：
# 编码器偏置：reset 模式
events = {
"encoder_bias": EventTermCfg(
func=mdp.randomize_joint_position_bias,
mode="reset",
params={"bias_range": (-0.02, 0.02)}, # ±0.02 rad ≈ ±1.1°
),
}
这个偏置在整个 episode 内保持不变——策略需要学会在关节角度有系统性偏差的情况下也能正确控制。它与 obs noise（每步随机）和 obs delay（时间偏移）是三个正交的 DR 维度。
Action 延迟模拟 ⭐⭐
策略输出到电机执行之间的通信延迟（1-5 ms）可以通过 action delay buffer 模拟：
# Action 延迟模拟
class ActionDelayBuffer:
"""在 env 和 wrapper 之间插入 action 延迟。"""
def __init__(self, env, max_delay_steps=2):
self.env = env
self.max_delay = max_delay_steps
self.buffer = []
def step(self, actions):
self.buffer.append(actions.clone())
# 当前使用的 action 是 N 步前的
if len(self.buffer) > self.max_delay:
delayed_actions = self.buffer.pop(0)
else:
delayed_actions = torch.zeros_like(actions)
return self.env.step(delayed_actions)
def reset(self):
self.buffer.clear()
return self.env.reset()
⚠️ 常见陷阱
⚠️ 编程陷阱：obs noise 加在 critic obs 上导致 value loss 不降。 critic 需要 clean 信号来准确估计 value。给 critic 加噪声等于给 PPO 的 value target 加噪声——advantage 质量下降，训练不稳定。
💡 概念误区：认为"obs noise 越大策略越鲁棒"。 过大的 noise 会让策略无法区分信号和噪声——它可能学到"忽略所有 obs 输入"的行为。noise std 应该反映真实传感器的实际噪声水平。
练习
- [设计题] 人形机器人的头部 RGB 相机帧率 30 FPS，控制频率 50 Hz。延迟应该是多少个 control step？
- [编程题] 在 mjlab 中配置 obs noise，训练 200 iteration。对比有/无 noise 的 tracking reward 和 action std。
8.10 自定义 DR Event 的编写模式 ⭐⭐
这一节解决什么问题：内置 DR 函数不够用时，如何编写自定义 event？
mjlab 自定义 event 示例 ⭐⭐
# 自定义 DR event：模拟随机负载
# src/my_task/mdp/events.py
import torch
def random_payload(
env, asset_cfg,
payload_mass_range=(0.0, 3.0),
com_offset_range=(-0.05, 0.05),
):
"""在 base 上模拟随机负载。"""
robot = env.scene[asset_cfg.name]
num_envs = env.num_envs
# 采样负载质量
payload = torch.rand(num_envs, device=env.device) \
* (payload_mass_range[1] - payload_mass_range[0]) \
+ payload_mass_range[0]
# 获取当前 base 质量并添加负载
base_mass = env.sim.model.body_mass[:, 0] # per-env
env.sim.model.body_mass[:, 0] = base_mass + payload # 原地写入
# 采样质心偏移
com_offset = torch.rand(num_envs, 3, device=env.device) \
* (com_offset_range[1] - com_offset_range[0]) \
+ com_offset_range[0]
env.sim.model.body_ipos[:, 0, :3] += com_offset
# 重算派生常量
env.sim.recompute_constants()
# 在 cfg 中注册
events = {
"payload": EventTermCfg(
func=random_payload,
mode="reset",
requires_model_fields=["body_mass", "body_ipos"],
params={
"asset_cfg": SceneEntityCfg("robot"),
"payload_mass_range": (0.0, 3.0),
"com_offset_range": (-0.05, 0.05),
},
),
}
Isaac Lab 自定义 event ⭐⭐
# Isaac Lab 自定义 event
def random_payload(
env: ManagerBasedRLEnv,
asset_cfg: SceneEntityCfg,
payload_mass_range: tuple[float, float],
):
"""模拟随机负载。"""
asset = env.scene[asset_cfg.name]
num_envs = env.num_envs
payload = torch.rand(num_envs, device=env.device) \
* (payload_mass_range[1] - payload_mass_range[0]) \
+ payload_mass_range[0]
body_masses = asset.root_physx_view.get_body_masses()
body_masses[:, 0] += payload
asset.root_physx_view.set_body_masses(body_masses)
# 注册
random_payload_event = EventTerm(
func=random_payload,
mode="reset",
params={
"asset_cfg": SceneEntityCfg("robot"),
"payload_mass_range": (0.0, 3.0),
},
)
编写自定义 event 的三条规则 ⭐⭐
规则一：event 只修改物理参数，不修改 obs/reward/done。DR 的语义是"改变环境"。
规则二：如果修改了 body_mass 等派生量源字段，必须调用 recompute_constants()。
规则三：在 mjlab 中，声明 requires_model_fields 让 EventManager 自动扩展字段。否则写入 per-env 数组时会因为字段未扩展而报错或影响所有 env。
DR 事件的测试方法 ⭐⭐
# 测试自定义 DR event
def test_dr_event(env, event_name, num_tests=5):
"""验证 DR event 是否正确修改了不同 env 的参数。"""
# 记录修改前的值
before = env.sim.model.body_mass[:num_tests, 0].clone()
# 触发 reset（会调用 reset 模式的事件）
env.reset()
# 记录修改后的值
after = env.sim.model.body_mass[:num_tests, 0].clone()
print(f"DR Event '{event_name}' test:")
for i in range(num_tests):
changed = "✅ CHANGED" if before[i] != after[i] else "⚠️ SAME"
print(f" env[{i}]: {before[i]:.4f} → {after[i]:.4f} {changed}")
⚠️ 常见陷阱
⚠️ 编程陷阱：自定义 event 中使用 field = new_tensor（替换引用）而非 field[:] = new_tensor（原地写入）。 替换引用会导致 CUDA Graph 失效。始终使用索引赋值。
8.11 DR 综合诊断工具箱 ⭐⭐
这一节解决什么问题：提供一站式 DR 诊断脚本。
DR 健康检查脚本 ⭐⭐
# dr_health_check.py — 一站式 DR 诊断
import torch
def dr_health_check(env, num_checks=5):
"""综合诊断 DR 配置是否正确。"""
print("=" * 60)
print(" DR Health Check")
print("=" * 60)
# Check 1: 字段扩展
expanded = getattr(env.sim, '_expanded_fields', set())
print(f"\n[1] Expanded fields: {len(expanded)}")
for field in sorted(expanded):
shape = getattr(env.sim.model, field).shape
print(f" {field}: shape={list(shape)}")
if not expanded:
print(" ⚠️ No fields expanded — DR may not be active!")
# Check 2: per-env 参数差异
print(f"\n[2] Per-env parameter variation:")
for field in list(expanded)[:5]:
data = getattr(env.sim.model, field)
if data.ndim >= 2 and data.shape[0] >= 2:
std_across_envs = data.float().std(dim=0).mean().item()
print(f" {field}: cross-env std={std_across_envs:.6f}")
if std_across_envs < 1e-8:
print(f" ⚠️ Zero variation — DR values may be identical!")
# Check 3: event mode 分布
print(f"\n[3] Event mode distribution:")
em = env.event_manager
mode_counts = {}
for name, term in em.active_terms.items():
mode = term.cfg.mode if hasattr(term.cfg, 'mode') else 'unknown'
mode_counts[mode] = mode_counts.get(mode, 0) + 1
print(f" [{mode:>8s}] {name}")
print(f" Totals: {mode_counts}")
# Check 4: entity pattern 匹配
print(f"\n[4] Entity pattern matching:")
for name, term in em.active_terms.items():
cfg = term.cfg
if hasattr(cfg, 'params') and 'asset_cfg' in cfg.params:
asset_cfg = cfg.params['asset_cfg']
if hasattr(asset_cfg, 'body_names') and asset_cfg.body_names:
import re
robot = env.scene[asset_cfg.name]
pattern = asset_cfg.body_names
matched = [n for n in robot.body_names
if re.match(pattern, n)]
status = "✅" if matched else "⚠️ EMPTY"
print(f" {name}: '{pattern}' → {len(matched)} matches {status}")
# Check 5: 性能影响估算
print(f"\n[5] Performance impact estimate:")
set_const_events = []
for name, term in em.active_terms.items():
# 检查事件是否触发 set_const 重算
# 这里简化为检查字段名
if hasattr(term.cfg, 'params'):
for key in term.cfg.params:
if 'mass' in str(key) or 'inertia' in str(key):
mode = term.cfg.mode if hasattr(term.cfg, 'mode') else '?'
if mode in ('step', 'interval'):
set_const_events.append((name, mode))
if set_const_events:
print(f" ⚠️ High-recompute events in frequent modes:")
for name, mode in set_const_events:
print(f" {name} in mode '{mode}' — consider moving to startup/reset")
else:
print(f" ✅ No high-recompute events in frequent modes")
print(f"\n{'=' * 60}")
print(" DR Health Check Complete")
print(f"{'=' * 60}")
# 使用：训练前运行
# dr_health_check(env)
三档鲁棒性评估脚本 ⭐⭐⭐
# dr_robustness_eval.py — 三档参数评估
import torch
def evaluate_robustness_3level(env, policy, param_configs):
"""在低/中/高三档参数下评估策略鲁棒性。
Args:
env: 环境
policy: 推理策略
param_configs: dict 如 {
"low_friction": {"geom_friction": 0.3},
"nominal": {"geom_friction": 0.7},
"high_friction": {"geom_friction": 1.5},
}
"""
results = {}
for config_name, params in param_configs.items():
print(f"\n--- Evaluating: {config_name} ---")
# 设置固定参数
for field, value in params.items():
field_data = getattr(env.sim.model, field)
if field_data.ndim >= 2:
field_data[:] = value # 所有 env 设为相同值
# 运行 10 个 episode
ep_rewards = []
ep_lengths = []
fell_count = 0
total_eps = 0
obs = env.reset()["actor"]
for step in range(5000):
with torch.no_grad():
actions = policy(obs)
obs_dict, rewards, terminated, truncated, extras = env.step(actions)
obs = obs_dict["actor"]
done = terminated | truncated
if done.any():
total_eps += done.sum().item()
fell_count += terminated.sum().item()
fell_rate = fell_count / max(total_eps, 1)
results[config_name] = {
"total_episodes": total_eps,
"fell_rate": fell_rate,
}
status = "✅" if fell_rate < 0.2 else "⚠️" if fell_rate < 0.5 else "🔴"
print(f" Episodes: {total_eps}, Fell rate: {fell_rate:.1%} {status}")
# 总结
print(f"\n{'=' * 50}")
print("Robustness Summary:")
for name, r in results.items():
print(f" {name}: fell_rate={r['fell_rate']:.1%}")
return results
# 使用示例
# configs = {
# "low_friction": {"geom_friction": torch.tensor([0.3, 0.005, 0.001])},
# "nominal": {"geom_friction": torch.tensor([0.7, 0.005, 0.001])},
# "high_friction": {"geom_friction": torch.tensor([1.5, 0.005, 0.001])},
# }
# evaluate_robustness_3level(env, policy, configs)
DR 对比实验脚本 ⭐⭐
# DR ablation 实验
export LOG_DIR="/tmp/mjlab/dr_ablation"
# 1. No DR baseline
uv run train Mjlab-Velocity-Flat-Unitree-Go1 \
--env.scene.num-envs 4096 --agent.max-iterations 500 \
--agent.run-name no_dr --agent.seed 42 \
--agent.logger tensorboard --agent.log-dir $LOG_DIR
# 需要注释掉 events 配置中的所有 DR
# 2. 只有 friction DR
uv run train Mjlab-Velocity-Flat-Unitree-Go1 \
--env.scene.num-envs 4096 --agent.max-iterations 500 \
--agent.run-name friction_only --agent.seed 42 \
--agent.logger tensorboard --agent.log-dir $LOG_DIR
# 3. 完整 DR
uv run train Mjlab-Velocity-Flat-Unitree-Go1 \
--env.scene.num-envs 4096 --agent.max-iterations 500 \
--agent.run-name full_dr --agent.seed 42 \
--agent.logger tensorboard --agent.log-dir $LOG_DIR
# 4. TensorBoard 对比
tensorboard --logdir $LOG_DIR
8.12 源码阅读路线 ⭐⭐
路线 A：mjlab DR 主链
src/mjlab/tasks/velocity/velocity_env_cfg.py→ events 配置src/mjlab/managers/event_manager.py→ 四种模式调度src/mjlab/envs/mdp/events.py→ push_by_setting_velocity, apply_body_impulsesrc/mjlab/envs/mdp/dr/_core.py→_randomize_model_field()src/mjlab/envs/mdp/dr/body.py→ pseudo_inertiasrc/mjlab/sim/randomization.py→ expand_model_fields, CUDA Graph 重 capture
路线 A 的核心阅读目标：理解从 task cfg 中的 events dict 到物理仿真中 per-env 参数变化的完整链路。每一层做了什么转换？字段在哪里被扩展？CUDA Graph 在哪里被重新 capture？
# 在源码中验证 DR 链路的诊断脚本
def trace_dr_pipeline(env, event_name="foot_friction"):
"""追踪一个 DR 事件从配置到物理生效的完整链路。"""
em = env.event_manager
# Layer 1: 配置层
print("Layer 1: Configuration")
for mode, terms in em._mode_term_cfgs.items():
for name, (func, cfg) in terms.items():
if event_name in name:
print(f" Found '{name}' in mode '{mode}'")
print(f" Func: {func.func.__name__}")
if hasattr(cfg, 'params'):
print(f" Params: {cfg.params}")
# Layer 2: 管理层
print("\nLayer 2: Management")
print(f" EventManager has {sum(len(t) for t in em._mode_term_cfgs.values())} total terms")
# Layer 3: 字段层
print("\nLayer 3: Field expansion")
if hasattr(env.sim, 'expanded_fields'):
print(f" Expanded fields: {env.sim.expanded_fields}")
# Layer 4: 仿真层
print("\nLayer 4: Simulation")
if hasattr(env.sim, 'model'):
model = env.sim.model
if hasattr(model, 'geom_friction'):
shape = model.geom_friction.shape
print(f" geom_friction shape: {shape}")
is_per_env = shape[0] == env.num_envs if len(shape) >= 2 else False
print(f" Per-env: {is_per_env}")
路线 B：Isaac Lab DR 对等路径
source/isaaclab_tasks/.../velocity/velocity_env_cfg.py→ EventCfgsource/isaaclab/managers/event_manager.py→ 五种模式调度（含 prestartup）source/isaaclab/envs/mdp/events.py→ DR 函数库- 对比
num_buckets参数（PhysX 特有）和randomize_rigid_body_material的实现
Isaac Lab 的 DR 函数库比 mjlab 更完整（Isaac Lab 积累了更长时间的社区贡献）。但 mjlab 的 pseudo-inertia 支持是独有优势——Isaac Lab 要实现等价功能需要手动编写。
路线 C：ASAP delta model
LeCAR-Lab/ASAP/humanoidverse/train_agent.py→ 统一训练入口（motion tracking 与 delta action）LeCAR-Lab/ASAP/humanoidverse/envs/motion_tracking/→ motion tracking 环境LeCAR-Lab/ASAP/humanoidverse/config/algo/ppo_train_delta_a.yaml→ delta action 的 PPO 配置（+exp=切换）LeCAR-Lab/ASAP/sim2real/rl_policy/listener_deltaa.py→ 真机部署/数据采集
ASAP 的代码结构值得阅读——它展示了如何把 delta action model 接入现有 RL 训练管线。注意 delta model 通过 open-loop RL（PPO）训练、再 closed-loop 微调 policy，复用统一的 train_agent.py 入口，而不是独立的 delta_model.py/train_delta.py。
路线 D：对比不同 DR 策略的论文实现
- extreme-parkour（Cheng et al., ICRA 2024）：aggressive friction DR + terrain curriculum
- walk-these-ways（Margolis & Agrawal, RSS 2023）：per-gait DR 配置
- TienKung-Lab（Open-X-Humanoid）：humanoid-specific DR + AMP reward 交互
阅读原则与 Ch05-Ch07 一致：先 task config → 再 manager → 最后底层实现。
📋 DR 设计审查 Checklist
使用方法：在完成新任务的 DR 设计或修改现有 DR 配置后，逐项检查。
A. 配置完整性
- [ ] 所有 DR 项的 mode 选择有物理理由
- [ ] entity pattern 经过匹配验证（非空）
- [ ] 参数范围反映真实物理不确定性
- [ ]
verify_dr_config()无警告
B. 物理一致性
- [ ] 质量/惯量使用 pseudo-inertia（或至少 recompute_inertia=True）
- [ ] 摩擦范围物理合理（>0）
- [ ] PD 增益范围不会导致不稳定
- [ ] 没有在 step 模式中触发 set_const 级别的字段修改
C. 训练效果
- [ ] Phase 0 clean baseline tracking > 0.7
- [ ] Phase 4 full DR tracking 下降 < 25%
- [ ]
test_dr_effectiveness()确认 per-env 参数不同 - [ ] TensorBoard 中 reward 方差增加在可接受范围
D. 鲁棒性验证
- [ ] 摩擦网格评估通过（最低摩擦 fall < 30%）
- [ ] Push recovery 测试通过（1 m/s 后 2 秒内恢复）
- [ ] 分布外 ±20% 参数下平缓退化
本章小结
| 知识点 | 核心要点 | 难度 |
|---|---|---|
| DR 鲁棒优化本质 | 在参数分布上优化期望性能 | ⭐⭐ |
| 四种事件模式 | startup/reset/interval/step 各有物理语义 | ⭐⭐⭐ |
| DR 函数映射 | 双框架 API 对照 + 采样分布选择 | ⭐⭐ |
| pseudo-inertia | Cholesky 参数化保证物理一致 | ⭐⭐⭐ |
| expand_model_fields | per-env 存储 + CUDA Graph 重 capture | ⭐⭐⭐ |
| 外力扰动 | push(速度注入) vs impulse(力注入) | ⭐⭐ |
| 分阶段策略 | 单变量递进，可归因 | ⭐⭐⭐ |
| 鲁棒性评估 | 固定参数网格，分布内+边界+分布外 | ⭐⭐ |
| ASAP delta model | 用真机数据修正 DR 无法覆盖的偏差 | ⭐⭐⭐ |
本章建立了两个关键认知：
从"固定参数训练"到"参数分布训练"。 DR 不是让仿真更逼真——而是让策略更鲁棒。真实世界的参数只是 DR 分布中的一个点。
从"一次性全开 DR"到"分阶段可归因引入"。 DR 的工程挑战不是"写配置"——而是"诊断问题"。分阶段引入让每个失败都能清晰归因到具体的 DR 项。
累积项目：本章新增模块
本章为累积项目新增"Domain Randomization 工程化"模块。你现在应该能够：
- 为 velocity task 配置完整的 DR 方案（在 mjlab 和 Isaac Lab 中）
- 执行分阶段 DR 训练并记录各阶段的 tracking 变化
- 使用固定参数网格评估策略鲁棒性
- 诊断 DR 不生效的常见原因
- 理解 pseudo-inertia 参数化的物理动机
累积项目检查点 ⭐⭐
完整验证命令序列：
# ============= 1. Phase 0: Clean baseline =============
uv run train Mjlab-Velocity-Flat-Unitree-Go1 \
--env.scene.num-envs 4096 --agent.max-iterations 300 \
--agent.run-name ch08_phase0 --agent.seed 42
# 验证 Phase 0
uv run play Mjlab-Velocity-Flat-Unitree-Go1 \
--agent.load-run ch08_phase0 --num-envs 4 --viewer viser
# ============= 2. Phase 1-4: 分阶段 DR =============
# Phase 1: 动力学 DR
uv run train Mjlab-Velocity-Flat-Unitree-Go1 \
--env.scene.num-envs 4096 --agent.max-iterations 500 \
--agent.run-name ch08_phase1 --agent.seed 42 \
--agent.resume True --agent.load-run ch08_phase0
# Phase 4: 完整 DR + push
uv run train Mjlab-Velocity-Flat-Unitree-Go1 \
--env.scene.num-envs 4096 --agent.max-iterations 1500 \
--agent.run-name ch08_phase4 --agent.seed 42 \
--agent.resume True --agent.load-run ch08_phase1
# ============= 3. 鲁棒性评估 =============
# 在不同摩擦下 play（手动观察）
for FRICTION in 0.3 0.5 0.8 1.0 1.5; do
echo "Testing friction=$FRICTION"
uv run play Mjlab-Velocity-Flat-Unitree-Go1 \
--agent.load-run ch08_phase4 --num-envs 4 \
--env.events.foot-friction.params.friction-range "($FRICTION, $FRICTION)" &
sleep 10 && kill %1 # 每个摩擦观察 10 秒
done
# ============= 4. TensorBoard 对比 =============
tensorboard --logdir /tmp/mjlab/logs/
# 对比 ch08_phase0 和 ch08_phase4 的 tracking reward 和 fall rate
Isaac Lab 等价验证：
# Phase 0
python scripts/reinforcement_learning/rsl_rl/train.py --task Isaac-Velocity-Flat-Anymal-C-v0 \
--num_envs 4096 --max_iterations 300 --seed 42
# Phase 1-4 需要修改 cfg 文件
# Play 验证
python scripts/reinforcement_learning/rsl_rl/play.py --task Isaac-Velocity-Flat-Anymal-C-v0
| 项目 | 本章对应任务 |
|---|---|
| 项目 A：四足速度跟踪 | 添加 foot friction + base mass + push DR |
| 项目 C：机械臂操作 | 添加 cube mass/friction DR + fingertip friction |
| 后期动态目标项目 | 添加球/拍 mass/friction + 风力扰动（Ch20-Ch22） |
与前置章节的连接 ⭐
Ch06 → Ch08 的耦合点：DR 改变环境参数分布 → reward 的期望值和方差都会变化 → PPO 的 advantage 估计质量受影响。如果 DR 范围太大，不同参数下的 reward scale 差异可能导致 advantage 估计方差暴增——此时可能需要增大 num_steps_per_env（Ch07）来改善 GAE 估计质量。
Ch07 → Ch08 的耦合点：DR 导致的 reward 方差增大可能触发 adaptive KL schedule 频繁降 LR → 训练变慢。如果 Phase 1 引入 DR 后 LR 大幅下降，检查 reward 分项是否有 DR 相关的 spike。
Ch08 → Ch09 的预告：Teacher-Student 蒸馏将使用 DR 训练的 teacher 策略。teacher 的鲁棒性直接影响 student 的蒸馏质量——如果 teacher 在某些 DR 参数下行为不好，student 从这些不好的行为中学到的也是不好的。
下一章（Ch09）将在 DR 的基础上解决另一个 sim-to-real 问题——部署时缺少 privileged 信息。Teacher-Student 蒸馏将利用 DR 训练的 teacher policy 来训练一个只使用部署可得信号的 student policy。
附录：DR 参数范围参考表
以下是各项目中常用的 DR 参数范围汇总，供配置参考：
四足 Locomotion 标准 DR 范围
| 参数 | legged_gym | Isaac Lab ANYmal | mjlab Go1 | 物理理由 |
|---|---|---|---|---|
| foot friction (scale) | (0.5, 1.25) | (0.7, 1.3) | (0.5, 1.25) | 不同地面材质 |
| base mass (add, kg) | (-1, 3) | (-1, 3) | (-1, 3) | 负载变化 |
| COM x/y/z (add, m) | (-0.02, 0.02) | (-0.02, 0.02) | (-0.02, 0.02) | 装配误差 |
| Kp (scale, log_uniform) | (0.75, 1.5) | (0.75, 1.5) | (0.75, 1.5) | 电机增益不确定 |
| Kd (scale, log_uniform) | (0.3, 3.0) | (0.3, 3.0) | (0.3, 3.0) | 阻尼不确定 |
| push lin vel (m/s) | (-1, 1) | (-1, 1) | (-1, 1) | 中等推力 |
| push ang vel (rad/s) | (-0.25, 0.25) | (-0.25, 0.25) | (-0.25, 0.25) | 轻度旋转 |
| push interval (s) | (10, 15) | (10, 15) | (10, 15) | 给恢复时间 |
| encoder bias (rad) | (-0.05, 0.05) | N/A | (-0.05, 0.05) | 传感器零点漂移 |
| restitution | (0.0, 0.5) | (0.0, 0.5) | (0.0, 0.5) | 接触弹性 |
Manipulation 标准 DR 范围
| 参数 | 范围 | 物理理由 |
|---|---|---|
| object mass (scale) | (0.5, 2.0) | 不同物体重量 |
| object friction (scale) | (0.5, 1.5) | 不同表面材质 |
| fingertip friction (scale) | (0.3, 2.0) | 指腹磨损/污染 |
| table height (add, m) | (-0.02, 0.02) | 放置误差 |
| object initial pose | 工作空间内随机 | 泛化到不同位置 |
人形 Locomotion 额外 DR 范围
| 参数 | 范围 | 物理理由 |
|---|---|---|
| trunk mass (add, kg) | (-2, 5) | 背包/工具负载 |
| ankle friction (scale) | (0.5, 1.5) | 鞋底材质差异 |
| gravity tilt (deg) | (-5, 5) | 斜坡近似 |
| motor strength (scale) | (0.8, 1.2) | 电机老化/温度影响 |
| link length (scale) | (0.98, 1.02) | 制造公差 |
DR 范围选择的三条原则
原则一：从物理不确定性出发。 查阅机器人的工程规格书，获得制造公差、传感器精度、负载范围等信息。DR 范围应反映这些真实的不确定性。
原则二：从小范围开始逐步扩大。 先用 ±10% 的范围训练，确认策略稳定后扩大到 ±30%。一次性设太大的范围可能导致无法训练。
原则三：关注 training tracking 的退化幅度。 引入 DR 后 tracking reward 下降是正常的（策略需要在更多参数上平均优化）。但如果下降超过 25%，说明范围可能太大或需要更多训练 iteration。
附录：DR 与 PPO 的数值交互分析
DR 不仅改变了任务——它还改变了 PPO 看到的数据分布。理解这种交互对正确配置训练超参数至关重要。
Reward 方差增大 ⭐⭐
无 DR 时，同一个策略在所有 4096 个环境中面对相同的物理参数——reward 的方差主要来自初始状态和 action 噪声。加 DR 后，不同环境有不同的摩擦、质量、PD 增益——reward 的方差增大，因为策略在低摩擦环境中的表现可能比高摩擦环境差很多。
方差增大对 GAE 的影响：GAE 使用 \(\delta_t = r_t + \gamma V(s_{t+1}) - V(s_t)\) 作为 TD error。如果 reward 方差增大，\(\delta_t\) 的方差也增大 → advantage 估计的噪声增大 → 策略梯度的方差增大 → 训练可能不稳定。
缓解方法：(1) 增大 num_steps_per_env（更长的 rollout 让 GAE 有更多步来平均噪声）；(2) 增大 num_envs（更多环境 → 更大 batch → 梯度方差更小）；(3) 使用 advantage normalization（RSL-RL 已默认启用）。
# 监控 DR 导致的 reward 方差变化
def monitor_reward_variance(env, policy, num_steps=100):
"""对比有/无 DR 时的 reward 方差。"""
obs = env.reset()
actor_obs = obs.get("actor", obs.get("policy"))
rewards_per_env = torch.zeros(env.num_envs, num_steps, device=env.device)
for step in range(num_steps):
with torch.inference_mode():
actions = policy(actor_obs)
obs_dict, rewards, _, _, _ = env.step(actions)
actor_obs = obs_dict.get("actor", obs_dict.get("policy"))
rewards_per_env[:, step] = rewards
# 计算 per-env 的 episode reward
env_rewards = rewards_per_env.sum(dim=1) # [num_envs]
mean_reward = env_rewards.mean().item()
std_reward = env_rewards.std().item()
cv = std_reward / abs(mean_reward) if abs(mean_reward) > 1e-8 else float('inf')
print(f"Reward statistics ({num_steps} steps):")
print(f" Mean: {mean_reward:.3f}")
print(f" Std: {std_reward:.3f}")
print(f" CV: {cv:.2f}")
if cv > 0.5:
print(f" ⚠️ High coefficient of variation — DR may be causing too much reward variance")
print(f" → Consider: narrowing DR ranges, increasing num_steps_per_env, or increasing num_envs")
else:
print(f" ✅ Reward variance acceptable")
# 使用：在 Phase 0 和 Phase 4 后分别运行，对比 CV 变化
Value Function 学习难度增大 ⭐⭐
DR 让 value function 的学习更难——因为 critic 需要在不同物理参数下准确估计未来 reward。如果 critic obs 不包含 DR 参数信息（如摩擦系数），critic 在面对同一个 state 但不同摩擦时需要预测一个"平均 value"——这个平均值对任何具体摩擦都不够精确。
这就是为什么 Ch05 的 asymmetric actor-critic 设计在 DR 训练中特别有价值——如果 critic obs 包含当前 DR 参数（作为 privileged information），critic 可以为每个参数下的 state 学到精确的 value。Actor 不需要看到 DR 参数（因为部署时不可得），但 critic 看到它们能让 advantage 估计更准确。
本质洞察：DR 训练中给 critic 提供 DR 参数作为 privileged information，是 DR + asymmetric AC + Teacher-Student 三者协同的关键接口。DR 提供鲁棒性，asymmetric AC 让 critic 在 DR 下更准确，Teacher-Student 把鲁棒性迁移到只用部署可得信号的 student 上。这三者是 sim-to-real 管线的核心三角。
延伸阅读
| 资料 | 难度 | 内容 |
|---|---|---|
| Tobin et al. 2017, "Domain Randomization for Sim-to-Real" | ⭐⭐ | 视觉 DR 的开山之作 |
| Peng et al. 2018, "Sim-to-Real with Dynamics Randomization" | ⭐⭐ | 动力学 DR |
| OpenAI 2019, "Solving Rubik's Cube" (arXiv 1910.07113) | ⭐⭐⭐ | ADR + Shadow Hand |
| Rucker & Wensing 2022, "Smooth Parameterization of Rigid-Body Inertia" | ⭐⭐⭐ | pseudo-inertia 数学基础 |
| He et al. 2025, "ASAP" (arXiv 2502.01143, RSS 2025) | ⭐⭐⭐ | delta action model |
| Sobanbabu et al. 2025, "SPI-Active" (arXiv 2505.14266) | ⭐⭐⭐ | Sampling-based System ID |
| Muratore et al. 2022, "Robot Learning from Randomized Simulations" | ⭐⭐⭐⭐ | DR 理论分析 |
🔧 故障排查手册
| 症状 | 可能原因 | 排查步骤 | 相关小节 |
|---|---|---|---|
| DR 配置生效但策略行为不变 | CUDA Graph 持旧指针 | 1.打印 expanded_fields 2.检查 graph capture 3.对比 no-DR 训练 | 8.4 |
| 低摩擦下转向打滑 | friction 范围未覆盖 | 1.做三档摩擦评估 2.检查 geom pattern 3.记录 slip velocity | 8.7 |
| pseudo_inertia NaN | alpha 范围过大 | 1.收窄 alpha 到 (0.9, 1.1) 2.检查正定性 3.逐步扩大 | 8.4 |
| push 训练摔倒率 >90% | 扰动太早或太强 | 1.验证 clean baseline 2.减小 push 幅度 3.延后引入 | 8.5, 8.6 |
| steps/s 下降 >30% | 高频 set_const 重算 | 1.统计 recompute 次数/step 2.检查 event mode 3.移至 reset | 8.4 |
| entity pattern 空匹配 | 正则表达式不匹配 MJCF 名字 | 1.打印 entity names 2.修正 pattern 3.在小 env 验证 | 8.3 |
| Isaac Lab material 超限 | num_buckets 未设或过大 | 1.检查 num_buckets 设置 2.设为 250 | 8.3 |
| DR 后 reward 大幅下降 | 范围太大 | 1.逐项收窄范围 2.增加 curriculum 3.检查分项 reward | 8.6 |
| 仿真中 DR 鲁棒但真机失败 | DR 未覆盖真实不确定性源 | 1.系统辨识 2.考虑 ASAP delta model 3.检查执行器模型 | 8.8 |