16 · 强化学习接口 ★

ai2work

182人浏览 · 2026-09-13 10:30:44

ai2work · 2026-09-13 10:30:44 发布

这一章要解决什么问题：第 15 章拿到了 Physics，但强化学习要的不是「一个仿真句柄」，
而是一个标准环境——能 reset()、能 step(action)、能告诉算法
「动作长什么样、观测长什么样」。

dm_control 通过 dm_control.rl.control 提供了这套接口，
本章把它拆开讲清楚，并复盘本书在这一步踩过的所有坑。

配套代码：[code/ch16_dm_control_rl.py]

"""
第 16 章配套代码：强化学习接口（dm_control.rl.control + dm_env）。

运行：
    D:\\Environment\\dm_control_env\\python.exe ch16_dm_control_rl.py
    D:\\Environment\\dm_control_env\\python.exe ch16_dm_control_rl.py --gui   # 顺便开一次窗口

内容：
  16.1 为什么需要标准环境接口
  16.2 dm_env 的四个概念：Environment / TimeStep / StepType / specs
  16.3 Task 协议的九个方法（含调用顺序）
  16.4 完整例子：末端到达 Task
  16.5 control.Environment 的构造参数 + 两条硬约束
  16.6 观测 / 动作空间设计
  16.7 跑完整 episode：随机策略 vs 手搓策略
  16.8 🔥 本书踩坑实录（每一条都有实测报错）
  16.10 动手练答案
"""
import os
import sys
import inspect
from collections import OrderedDict
import numpy as np
import mujoco
import dm_env
from dm_control import mjcf
from dm_control.rl import control
from dm_env import specs

np.set_printoptions(precision=5, suppress=True)

HERE = os.path.dirname(os.path.abspath(__file__))
MJCF = os.path.abspath(
    os.path.join(HERE, "..", "..", "models",
                 "cx4_a601c_simulation.xml"))

JN = [f"joint_j{i}" for i in range(1, 7)]        # 6 个关节名
AN = [f"pos_j{i}" for i in range(1, 7)]          # 6 个执行器名
HOME = np.array([0.0, 0.2, 0.6, 0.0, 0.3, 0.0])
CONTROL_DT = 0.02


def banner(t):
    print("\n" + "=" * 72)
    print(t)
    print("=" * 72)


# ============================================================
# 16.1 / 16.2 dm_env 的四个概念
# ============================================================
banner("16.1 / 16.2  dm_env 的四个概念")

print("""
  dm_env（DeepMind 的环境接口标准）只规定了 4 件事：

  ① Environment      环境本身：reset() / step()
  ② TimeStep         每一步的返回值：(step_type, reward, discount, observation)
  ③ StepType         FIRST / MID / LAST  —— 首帧 / 中间帧 / 末帧
  ④ specs            观测与动作的"形状说明书"

  为什么要有标准？
    因为有了它，同一套 RL 算法代码可以跑在 dm_control / gym / 自定义环境上，
    只要大家都返回 TimeStep、都提供 xxx_spec()。
""")

print("  StepType 枚举:")
for st in dm_env.StepType:
    print(f"    {st.name:<6} = {st.value}")
print(f"\n  TimeStep 字段 = {dm_env.TimeStep._fields}")
print(f"  specs 可用类型 = {[s for s in dir(specs) if s[0].isupper()]}")

# ============================================================
# 16.3 / 16.4 完整 Task
# ============================================================
banner("16.3 / 16.4  Task 协议 + 完整例子：末端到达")

ABSTRACT = sorted(control.Task.__abstractmethods__)
print("""
  control.Task 里有 9 个方法。怎么知道哪些必须实现？最靠谱的是看抽象方法集合：

    sorted(control.Task.__abstractmethods__)
""")
print(f"    -> {ABSTRACT}")
print(f"       共 {len(ABSTRACT)} 个，【必须】实现\n")

print("""
    方法                       是否必须   什么时候被调用
    ─────────────────────────────────────────────────────────
    initialize_episode(physics)  必须★   reset() 时，重置状态
    before_step(action, physics) 必须★   step() 里，把 action 写进 physics
    action_spec(physics)         必须★   环境询问动作空间
    get_observation(physics)     必须★   组装观测
    get_reward(physics)          必须★   计算奖励
    observation_spec(physics)    可选    不写也行！基类从 get_observation 自动推断
    after_step(physics)          可选    物理步进之后
    get_termination(physics)     可选    返回 None=继续 / float=终止
    step_spec(physics)           可选    基类默认抛 NotImplementedError

  ⚠️ 参数顺序！before_step 是 (action, physics)，不是 (physics, action)。
  ⚠️ 本书踩过的坑：最初看报错文本被截断成 '... get_re...' 就以为只有 4 个方法，
     漏掉了 initialize_episode，还误以为 observation_spec 是必须的。
     两个结论都错了 —— 永远用 __abstractmethods__ 查，别数报错文本。
""")

print("  实测 before_step 签名:",
      inspect.signature(control.Task.before_step))
print("  实测 Environment 签名:",
      inspect.signature(control.Environment.__init__))

mjcf_model = mjcf.from_path(MJCF)

# 实测：缺 observation_spec 能不能跑
print("\n  【实测】缺 observation_spec 会怎样？")


class NoObsSpec(control.Task):
    def initialize_episode(self, physics): pass

    def before_step(self, action, physics): pass

    def action_spec(self, physics):
        return specs.BoundedArray((1,), np.float64, -1.0, 1.0, "a")

    def get_observation(self, physics):
        return OrderedDict(q=np.zeros(3))

    def get_reward(self, physics): return 0.0


try:
    _e = control.Environment(mjcf.Physics.from_mjcf_model(mjcf_model),
                             NoObsSpec(), control_timestep=0.02, time_limit=0.1)
    _ts = _e.reset()
    print(f"    ✅ 能构造。env.observation_spec() = "
          f"{list(_e.observation_spec().keys())}  (自动从 get_observation 推断)")
    print(f"       reset 观测 = {list(_ts.observation.keys())}")
except Exception as ex:
    print(f"    ❌ {type(ex).__name__}: {str(ex)[:70]}")

# 实测：缺 initialize_episode 会怎样
print("\n  【实测】缺 initialize_episode 会怎样？")


class NoInit(control.Task):
    def before_step(self, action, physics): pass

    def action_spec(self, physics):
        return specs.BoundedArray((1,), np.float64, -1.0, 1.0, "a")

    def observation_spec(self, physics):
        return OrderedDict(q=specs.Array((3,), np.float64, "q"))

    def get_observation(self, physics): return OrderedDict(q=np.zeros(3))

    def get_reward(self, physics): return 0.0


try:
    control.Environment(mjcf.Physics.from_mjcf_model(mjcf_model), NoInit(),
                        control_timestep=0.02, time_limit=0.1)
    print("    （能构造 —— 与预期不符，请检查）")
except TypeError as ex:
    print(f"    ❌ TypeError: {ex}")
    print("       -> initialize_episode 是抽象方法，必须实现")


class ReachTask(control.Task):
    """任务：让末端执行器尽量靠近一个目标点。"""

    def __init__(self, target, tol=0.05):
        self._target = np.asarray(target, dtype=float)
        self._tol = float(tol)
        self._ee_site = mjcf_model.find("site", "end_effector")
        self.n_reset = 0

    # ---- 生命周期 ----
    def initialize_episode(self, physics):
        self.n_reset += 1
        physics.named.data.qpos[JN] = HOME
        physics.named.data.ctrl[AN] = HOME
        physics.named.data.ctrl["pos_gripper_left"] = 0.03
        physics.named.data.ctrl["pos_gripper_right"] = 0.03
        physics.forward()

    def before_step(self, action, physics):
        """★ 把 RL 给出的 action 写进 physics。参数顺序 (action, physics)。"""
        physics.named.data.ctrl[AN] = action

    def after_step(self, physics):
        pass

    # ---- 空间声明 ----
    def action_spec(self, physics):
        # ⚠️ 参数顺序: BoundedArray(shape, dtype, minimum, maximum, name)
        return specs.BoundedArray(
            shape=(6,), dtype=np.float64,
            minimum=-3.0, maximum=3.0, name="joint_target")

    def observation_spec(self, physics):
        return OrderedDict([
            ("joints", specs.Array((6,), np.float64, "joint_pos")),
            ("ee", specs.Array((3,), np.float64, "ee_pos")),
            ("target", specs.Array((3,), np.float64, "target_pos")),
        ])

    # ---- 数据 ----
    def get_observation(self, physics):
        return OrderedDict(
            joints=physics.named.data.qpos[JN].copy(),
            ee=physics.bind(self._ee_site).xpos.copy(),
            target=self._target.copy(),
        )

    def get_reward(self, physics):
        d = np.linalg.norm(physics.bind(self._ee_site).xpos - self._target)
        return float(-d)

    def get_termination(self, physics):
        d = np.linalg.norm(physics.bind(self._ee_site).xpos - self._target)
        if d < self._tol:
            return 0.0        # 成功终止，discount=0
        return None           # 继续


physics = mjcf.Physics.from_mjcf_model(mjcf_model)
TARGET = np.array([-0.30, 0.30, 0.10])
task = ReachTask(TARGET)
env = control.Environment(physics, task, control_timestep=CONTROL_DT,
                          time_limit=4.0)

print(f"\n  type(env) = {type(env).__name__}   "
      f"是 dm_env.Environment? {isinstance(env, dm_env.Environment)}")
print(f"  env.physics is physics ? {env.physics is physics}")

print("\n  action_spec:")
a = env.action_spec()
print(f"    {a}")
print(f"    shape={a.shape}  dtype={a.dtype}  "
      f"minimum={a.minimum}  maximum={a.maximum}")
print("\n  observation_spec:")
for k, v in env.observation_spec().items():
    print(f"    {k:<8} shape={v.shape}  dtype={v.dtype}")

# ============================================================
# 16.5 构造参数的两条硬约束
# ============================================================
banner("16.5  control.Environment 的构造参数 + 两条硬约束")

print("  ⚠️ 约束①: n_sub_steps 与 control_timestep 【互斥】")
try:
    control.Environment(mjcf.Physics.from_mjcf_model(mjcf_model), ReachTask(TARGET),
                        control_timestep=0.02, n_sub_steps=10, time_limit=1.0)
except ValueError as e:
    print(f"    ValueError: {e}")

print("\n  ⚠️ 约束②: control_timestep 必须是【仿真步长的整数倍】")
try:
    control.Environment(mjcf.Physics.from_mjcf_model(mjcf_model), ReachTask(TARGET),
                        control_timestep=0.021, time_limit=1.0)
except ValueError as e:
    print(f"    ValueError: {e}")
print(f"    （本项目 timestep = {physics.model.opt.timestep}，"
      f"0.02 / 0.002 = 10 ✅）")

print("\n  用 control_timestep 指定（按时间，推荐）:")


def step_advance(**kw):
    p = mjcf.Physics.from_mjcf_model(mjcf_model)
    e = control.Environment(p, ReachTask(TARGET), **kw)
    e.reset()
    t0 = p.data.time
    e.step(np.zeros(6))
    return p.data.time - t0


for ct in (0.01, 0.02, 0.05):
    dt = step_advance(control_timestep=ct, time_limit=1.0)
    print(f"    control_timestep={ct:<5} -> 一个 step 推进 {dt:.4f} s "
          f"(= {dt/0.002:.0f} 个物理步)")

print("\n  用 n_sub_steps 指定（按步数）:")
for nss in (1, 5, 10, 20):
    dt = step_advance(n_sub_steps=nss, time_limit=1.0)
    print(f"    n_sub_steps={nss:<5}       -> 一个 step 推进 {dt:.4f} s "
          f"(= {dt/0.002:.0f} 个物理步)")

# ============================================================
# 16.6 空间设计
# ============================================================
banner("16.6  观测 / 动作空间设计")

print("  specs 常用类型：")
print("    specs.Array(shape, dtype, name)                     无界连续量")
print("    specs.BoundedArray(shape, dtype, min, max, name)    有界连续量")
print("    specs.DiscreteArray(num_values, dtype, name)        离散动作")
print()
print("  ⚠️ BoundedArray 的参数顺序是 (shape, dtype, minimum, maximum, name)")
print("     位置写错不会报错，只是上下界全乱 —— 很隐蔽")

print("\n  每个维度不同范围（真实机器人几乎都是这样）:")
lo = np.array([-3.0, -2.0, -2.5, -3.0, -2.0, -3.0])
hi = np.array([3.0, 2.0, 2.5, 3.0, 2.0, 3.0])
s = specs.BoundedArray((6,), np.float64, lo, hi, "per_joint")
print(f"    minimum = {s.minimum}")
print(f"    maximum = {s.maximum}")

print("\n  ⚠️ action 越界时 dm_env 不强制校验：")
p2 = mjcf.Physics.from_mjcf_model(mjcf_model)
e2 = control.Environment(p2, ReachTask(TARGET), control_timestep=CONTROL_DT,
                         time_limit=1.0)
e2.reset()
try:
    e2.step(np.full(6, 99.0))
    print("    step([99]*6) -> 没报错（需要你自己 clip）")
except Exception as e:
    print(f"    {type(e).__name__}: {str(e)[:60]}")
try:
    e2.step(np.zeros(3))
except Exception as e:
    print(f"    step(shape=(3,)) -> {type(e).__name__}: {str(e)[:60]}")

print("\n  flat_observation=True：把字典拼成一个向量（喂给 MLP 时用）")
for fo in (False, True):
    p3 = mjcf.Physics.from_mjcf_model(mjcf_model)
    e3 = control.Environment(p3, ReachTask(TARGET), control_timestep=CONTROL_DT,
                             time_limit=1.0, flat_observation=fo)
    obs = e3.reset().observation
    keys = list(obs.keys())
    shape = np.shape(obs["observations"]) if fo else None
    print(f"    flat={str(fo):<5} -> keys={keys}"
          f"{'' if shape is None else '  observations.shape=' + str(shape)}")

# ============================================================
# 16.7 跑完整 episode
# ============================================================
banner("16.7  跑完整 episode")

print("  ① 先看 reset() 返回的 TimeStep：")
ts = env.reset()
print(f"    step_type = {ts.step_type}  (FIRST={dm_env.StepType.FIRST})")
print(f"    reward    = {ts.reward}")
print(f"    discount  = {ts.discount}")
print(f"    observation = {{{', '.join(f'{k}: {np.round(v,4)}' for k, v in ts.observation.items())}}}")

print("\n  ② 随机策略 vs 手搓策略，各跑一个 episode：")


def run_episode(policy, seed=0, max_steps=300):
    rng = np.random.default_rng(seed)
    ts = env.reset()
    total, n = 0.0, 0
    dists = []
    while ts.step_type != dm_env.StepType.LAST and n < max_steps:
        act = policy(ts, rng)
        ts = env.step(act)
        total += (ts.reward or 0.0)
        dists.append(-(ts.reward or 0.0))
        n += 1
    return total, n, (dists[-1] if dists else float("nan")), ts.step_type


def rand_policy(ts, rng):
    return rng.uniform(-1.0, 1.0, size=6)


def hold_policy(ts, rng):
    """一个很笨但稳定的策略：始终命令同一个关节角。"""
    return np.array([0.0, 0.6, 0.9, 0.0, 0.4, 0.0])


for name, pol in [("随机策略", rand_policy), ("固定关节角", hold_policy)]:
    tot, n, d, st = run_episode(pol)
    print(f"    {name:<10}: {n:>3} 步  累计奖励={tot:>9.3f}  "
          f"末端最终距目标={d:.4f} m  结束于 {st}")

print("\n  ③ 一个真正会收敛的策略：用雅可比做 IK（复习第 6 章的 DLS）")
print("     ⚠️ 小结教训：随便手调关节角多半不收敛，还是得算雅可比。")

_EE_ID = mujoco.mj_name2id(physics.model.ptr, mujoco.mjtObj.mjOBJ_SITE,
                           "end_effector")


def ik_policy(ts, rng, step_size=0.5, lam=0.05):
    """每个控制周期做一步阻尼最小二乘 IK，把末端往目标拉。"""
    m, d = physics.model.ptr, physics.data.ptr
    # ⚠️ mujoco 3.x 的 mj_jacSite 要 (3, nv) 的二维数组，不是扁平的 3*nv
    jacp = np.zeros((3, m.nv))
    jacr = np.zeros((3, m.nv))
    mujoco.mj_jacSite(m, d, jacp, jacr, _EE_ID)
    J = jacp[:, :6]                                   # 只驱动 6 个臂关节
    err = ts.observation["target"] - ts.observation["ee"]
    dq = J.T @ np.linalg.solve(J @ J.T + lam ** 2 * np.eye(3), err)
    q = np.asarray(ts.observation["joints"]) + step_size * dq
    return np.clip(q, -3.0, 3.0)


print(f"    {'step':>5} {'距目标(m)':>10} {'reward':>10} {'step_type':>10}")
ts = env.reset()
for i in range(300):
    ts = env.step(ik_policy(ts, None))
    if i % 20 == 0 or ts.step_type == dm_env.StepType.LAST:
        print(f"    {i:>5} {-(ts.reward or 0):>10.4f} {ts.reward:>10.4f} "
              f"{str(ts.step_type):>10}")
    if ts.step_type == dm_env.StepType.LAST:
        break
print(f"    -> 收敛到 {-(ts.reward or 0):.4f} m，"
      f"{'✅ 触发成功终止' if ts.step_type == dm_env.StepType.LAST and ts.discount == 0.0 else '超时结束'}")

tot, n, d, st = run_episode(lambda ts, rng: ik_policy(ts, rng))
print(f"    整轮统计: {n} 步, 累计奖励={tot:.3f}, 最终距离={d:.4f} m, {st}")

print("\n  ④ get_termination 各种返回值对照:")
print("    None  -> 继续（step_type=MID,  discount=1.0）")
print("    0.0   -> 成功终止（step_type=LAST, discount=0.0）")
print("    1.0   -> 超时截断（step_type=LAST, discount=1.0，可 bootstrapping）")

# ============================================================
# 16.8 🔥 踩坑实录
# ============================================================
banner("16.8  🔥 本书踩坑实录（每条都有实测报错）")

print("【坑 1】包装类与 mujoco.viewer.launch_passive 不兼容")
p4 = mjcf.Physics.from_mjcf_model(mjcf_model)
import mujoco.viewer  # noqa: F401
print(f"    type(physics.model) = {type(p4.model)}")
try:
    mujoco.viewer.launch_passive(p4.model, p4.data)
    print("    没报错")
except Exception as e:
    msg = str(e).replace("\n", " ")
    print(f"    ❌ launch_passive(physics.model, physics.data)")
    print(f"       -> {type(e).__name__}: {msg[:100]}")
print(f"    type(physics.model.ptr) = {type(p4.model.ptr)}")
try:
    v = mujoco.viewer.launch_passive(p4.model.ptr, p4.data.ptr)
    print(f"    ✅ launch_passive(physics.model.ptr, physics.data.ptr) -> 成功")
    v.close()
    print("       窗口已关闭（本机有 GUI）")
except Exception as e:
    print(f"    (无 GUI 环境) {type(e).__name__}: {str(e)[:70]}")

print("\n【坑 2】mujoco 3.x 的对象枚举搬到了 mjtObj 下")
try:
    mujoco.mjOBJ_SITE
    print("    mujoco.mjOBJ_SITE 存在")
except AttributeError as e:
    print(f"    ❌ mujoco.mjOBJ_SITE -> AttributeError: {str(e)[:60]}")
print(f"    ✅ mujoco.mjtObj.mjOBJ_SITE = {mujoco.mjtObj.mjOBJ_SITE}")
SENSOR_TYPES = {int(v): k for k, v in vars(mujoco.mjtSensor).items()
                if k.startswith("mjSENS_")}
print(f"    ✅ 传感器类型反查: mjtSensor 里 mjSENS_FRAMEPOS = "
      f"{mujoco.mjtSensor.mjSENS_FRAMEPOS}")

print("\n【坑 3】before_step 的参数顺序是 (action, physics)")
print(f"    实测签名: {inspect.signature(control.Task.before_step)}")
print("    网上的旧教程常写成 (physics, action)，结果 action 变成 physics、")
print("    一调用就 AttributeError。")

print("\n【坑 4/5】见 16.5：control_timestep 必须整数倍，且与 n_sub_steps 互斥")

print("\n【坑 6】空 Task 会报抽象方法未实现（注意是 5 个，别被截断骗了）")
class Empty(control.Task):
    pass
try:
    control.Environment(mjcf.Physics.from_mjcf_model(mjcf_model), Empty(),
                        control_timestep=CONTROL_DT, time_limit=0.1)
except TypeError as e:
    print(f"    TypeError: {e}")
    print(f"\n    完整清单: {ABSTRACT}")

print("\n【坑 7】step_spec() 基类默认抛 NotImplementedError")
try:
    env.step_spec()
except NotImplementedError:
    print("    NotImplementedError（不用它就不用管）")

# ============================================================
# 16.10 动手练
# ============================================================
banner("16.10 动手练 参考答案")
print(f"练习1 TimeStep  : reset -> FIRST(reward=None)，step -> MID，"
      f"终止 -> LAST")
print(f"练习2 Task 协议 : 5 个抽象方法 = {ABSTRACT}")
print(f"                 observation_spec 是可选的（会自动推断）")
print(f"练习3 时间关系  : control_timestep={CONTROL_DT} / "
      f"timestep={physics.model.opt.timestep} = "
      f"{CONTROL_DT/physics.model.opt.timestep:.0f} 个物理步/控制步")
print(f"练习4 空间设计  : BoundedArray(shape, dtype, minimum, maximum, name)")
print(f"练习5 跑 episode: 见 16.7 的对比（随机 vs 固定 vs 渐进）")
print(f"练习6 踩坑      : 传 .ptr 给 launch_passive；枚举用 mjtObj")

# 可选：开一次窗口
if "--gui" in sys.argv:
    print("\n[--gui] 打开交互窗口 3 秒后自动关闭 ...")
    try:
        import time
        v = mujoco.viewer.launch_passive(physics.model.ptr, physics.data.ptr)
        ts = env.reset()
        t_end = time.time() + 3.0
        while time.time() < t_end:
            ts = env.step(hold_policy(ts, None))
            v.sync()
        v.close()
        print("  窗口已关闭")
    except Exception as e:
        print(f"  跳过: {type(e).__name__}: {str(e)[:80]}")

print("\n第 16 章示例代码运行完毕。")
print("下一章：composer —— 把机器人和场景拆成可复用的模块。")

📌 本章所有数字与报错信息都在作者机器上实测得到
（mujoco 3.11.0 / dm-control 1.0.44 / dm-env 1.6）。

🎯 学习目标

学完本章，你将能够：

理解 dm_env 标准的四个核心概念——Environment / TimeStep / StepType / specs，说清为什么需要标准化环境接口。
实现一个完整的 control.Task——掌握 5 个必须实现的抽象方法（initialize_episode / before_step / action_spec / get_observation / get_reward），理解 4 个可选方法的作用。
设计观测空间和动作空间——用 specs.Array / specs.BoundedArray 声明空间，理解每维不同范围的设置方法，知道 dm_env 不校验动作越界需要自己 clip。
跑通完整的 RL 交互循环——理解 reset() → step(action) → 奖励/观测 → 策略决策的闭环，能对比随机策略、固定策略和 IK 策略的收敛效果。
规避本章的关键陷阱——before_step 参数顺序（action 在前）、control_timestep 必须是仿真步长整数倍且与 n_sub_steps 互斥、包装类与 launch_passive 不兼容（要用 .ptr）、mujoco 3.x 的枚举搬到了 mjtObj 下。
理解奖励函数和终止条件的设计——掌握 get_reward 和 get_termination 的返回值约定（None/0.0/1.0 分别代表继续/成功/超时），理解 discount 因子对算法的意义。

📖 前置知识：本章需要第 15 章的 dm_control 基础知识（Physics 句柄、named 命名访问、.ptr 原生对象）。如果你还没读第 15 章，请先回去读完。

16.1 为什么需要标准环境接口

直观理解：RL 交互循环是什么？

想象你在教一个小孩玩"把积木放到指定位置"的游戏：

Agent（智能体） = 小孩——他看着积木的位置（观测），决定手怎么动（动作）。
Environment（环境） = 游戏场地——积木会因为重力下落、撞到桌子会弹开（物理仿真）。
Observation（观测） = 小孩眼睛看到的——积木在哪、手在哪、目标在哪。
Action（动作） = 小孩手的运动指令——往左移一点、往上抬一点。
Reward（奖励） = 大人给的反馈——积木离目标近了就夸一句（正奖励），远了就摇头（负奖励）。
Episode（回合） = 一局游戏——从开始（积木在初始位置）到结束（积木到位或时间到）。

交互循环：小孩看一眼 → 决定动手 → 环境变化 → 再看一眼 → 再决定……如此循环，直到游戏结束。

Agent-Environment 交互循环图

                    ┌──────────────────────────────────────┐
                    │            Agent（智能体）             │
                    │  策略 π(observation) → action         │
                    │  （神经网络 / 规则 / IK / 随机）        │
                    └───────────────┬──────────────────────┘
                                    │ action (动作)
                                    │ 例：[0.1, 0.3, -0.2, ...]
                                    ▼
┌─────────────────────────────────────────────────────────────────┐
│                     Environment（环境）                           │
│                                                                   │
│  ┌─────────────┐    ┌──────────────┐    ┌──────────────────┐  │
│  │ before_step │ →  │ physics.step │ →  │ get_observation  │  │
│  │ (写控制量)   │    │ (物理推进N步) │    │ (组装观测字典)    │  │
│  └─────────────┘    └──────────────┘    └──────────────────┘  │
│                                          │                        │
│  ┌─────────────┐    ┌──────────────┐    │                        │
│  │ get_termin. │ ←  │  get_reward  │ ←  ┘                        │
│  │ (是否终止)   │    │  (计算奖励)    │                             │
│  └─────────────┘    └──────────────┘                              │
│         │                                                           │
│         ▼                                                           │
│  TimeStep(step_type, reward, discount, observation)               │
└─────────────────────────────┬─────────────────────────────────────┘
                              │
        ┌─────────────────────┼─────────────────────┐
        │ observation         │ reward              │ step_type
        │ 观测字典             │ 标量奖励             │ FIRST/MID/LAST
        ▼                     ▼                     ▼
  给 Agent 决策用         告诉 Agent 好不好      标记 episode 状态
  例：{joints, ee, target}  例：-0.15（距离越近越大）

💡 关键洞察：RL 的本质就是这个闭环——Agent 不知道环境内部怎么工作（物理引擎、碰撞检测），
它只通过「观测」了解环境、通过「动作」影响环境、通过「奖励」知道自己做得好不好。
这就是为什么需要标准接口：只要环境返回标准的 TimeStep，任何 Agent 都能接入。

没有标准时，每个人写的 RL 循环都不一样：

# 张三的写法
obs = env.reset()
while not done:
    obs, r, done, info = env.step(a)

# 李四的写法
state = sim.reset()
while True:
    state, r, terminal = sim.advance(a)

于是算法代码没法复用。DeepMind 的 dm_env 就是为了解决这个：
它规定「环境长什么样」，不管底下是 MuJoCo、Unity 还是真实机器人。

dm_control 的 control.Environment 就是 dm_env 标准的一个实现：

env = control.Environment(physics, task, control_timestep=0.02, time_limit=4.0)
time_step = env.reset()
time_step = env.step(action)

16.2 dm_env 的四个概念

直观理解：四个概念分别是什么？

概念	生活类比	一句话定义
Environment	游戏机本体	提供 `reset()` 和 `step(action)` 两个按钮
TimeStep	游戏画面的一帧	每一步返回的数据包：(当前状态, 得分, 折扣, 画面)
StepType	游戏状态指示灯	这一帧是「刚开始」「进行中」还是「结束了」
specs	游戏手柄说明书	告诉 Agent：观测长什么样、动作长什么样、范围是多少

为什么需要 specs？ 想象你买了一个游戏手柄，但没有说明书——你不知道左边摇杆控制什么、
右边按钮有几个、摇杆能推多大角度。specs 就是这份说明书，它让 Agent 在真正开始玩之前
就知道「我应该输出什么形状的动作」「我会收到什么形状的观测」。

Episode 时序图

时间轴 →
  t=0        t=1        t=2        ...      t=T
  │          │          │                   │
  ▼          ▼          ▼                   ▼
┌──────┐  ┌──────┐  ┌──────┐           ┌──────┐
│FIRST │  │ MID  │  │ MID  │    ...    │ LAST │
│reset │  │ step │  │ step │           │ step │
│      │  │      │  │      │           │      │
│obs   │  │obs   │  │obs   │           │obs   │
│reward│  │reward│  │reward│           │reward│
│=None │  │=r1   │  │=r2   │           │=rT   │
│disc  │  │disc  │  │disc  │           │disc  │
│=None │  │=1.0  │  │=1.0  │           │=0/1  │
└──────┘  └──────┘  └──────┘           └──────┘
  │          │          │                   │
  └──────────┴──────────┴───────────────────┘
              一个完整的 episode
         （从 reset 到终止/超时）

累计奖励 = r1 + r2 + ... + rT
（注意：FIRST 帧的 reward 是 None，不计入）

⚠️ 特别注意 FIRST 帧：reset() 返回的 TimeStep 中，reward=None、discount=None。
这是因为游戏还没开始，没有任何动作被执行，所以没有奖励。
写 RL 循环时如果直接 total += ts.reward，会在第一步就 TypeError。
正确写法：total += (ts.reward or 0.0)。

dm_env 只规定了 4 件事：

概念	是什么
Environment	环境本身：`reset()` / `step()`
TimeStep	每一步的返回值：`(step_type, reward, discount, observation)`
StepType	`FIRST` / `MID` / `LAST`
specs	观测与动作的「形状说明书」

实测：

StepType 枚举:
  FIRST  = 0     首帧（reset 返回）
  MID    = 1     中间帧（普通的 step）
  LAST   = 2     末帧（episode 结束）

TimeStep 字段 = ('step_type', 'reward', 'discount', 'observation')
specs 可用类型 = ['Array', 'BoundedArray', 'DiscreteArray', 'Optional', 'StringArray']

`TimeStep` 四个字段的含义

TimeStep(step_type, reward, discount, observation)

字段	说明
`step_type`	FIRST / MID / LAST
`reward`	这一步的奖励；`reset()` 时是 `None`
`discount`	折扣因子；`FIRST` 时是 `None`
`observation`	观测（字典或数组）

⚠️ reset() 返回的 TimeStep 里 reward=None、discount=None。
写 RL 循环时别直接 total += ts.reward，会 TypeError。
正确写法：total += (ts.reward or 0.0)。

16.3 `Task` 协议：九个方法

直观理解：Task 是什么？

想象你是一个游乐场的设计师，要设计一个"套圈"游戏：

Environment = 游乐场的通用运营框架——负责开门、计时、叫号。
Task = 你设计的具体游戏规则——圈有多大、目标在哪、怎么计分、什么时候游戏结束。

control.Task 就是一个"游戏规则模板"，你继承它并填写具体规则，Environment 就会按照你的规则来运营游戏。

为什么需要 Task 协议？ 因为 Environment 不需要知道具体游戏是什么——它只需要知道
“什么时候重置”、“动作怎么传给物理”、“观测怎么组装”、“奖励怎么算”、“什么时候结束”。
这些通用逻辑由 Environment 实现，具体规则由你的 Task 子类实现。这就是模板方法模式（Template Method Pattern）。

Task 方法调用时序图

env.reset()
    │
    ├─→ task.initialize_episode(physics)
    │       做什么：重置关节到 HOME 位姿、随机化物体位置、清零计数器
    │       类比：游戏开始前，把道具摆回初始位置
    │
    ├─→ task.get_observation(physics)
    │       做什么：组装观测字典（关节角、末端位置、目标位置）
    │       类比：给玩家看一眼当前局面
    │
    └─→ 返回 TimeStep(FIRST, None, None, obs)
            reward=None（还没做任何动作，没有奖励）
            discount=None（FIRST 帧没有折扣）

env.step(action)
    │
    ├─→ task.before_step(action, physics)
    │       做什么：把 action 写进 physics.data.ctrl（发送控制指令）
    │       类比：玩家投出圈，工作人员记录投掷力度
    │       ⚠️ 参数顺序：(action, physics)，action 在前！
    │
    ├─→ physics.step(n_sub_steps)
    │       做什么：物理引擎推进 N 步（N = control_timestep / timestep）
    │       类比：圈在空中飞，物理引擎计算轨迹
    │
    ├─→ task.after_step(physics)
    │       做什么：物理步进后的钩子（通常不用，可用于记录中间状态）
    │
    ├─→ task.get_observation(physics)
    │       做什么：组装新的观测字典
    │       类比：圈落地后，给玩家看新的局面
    │
    ├─→ task.get_reward(physics)
    │       做什么：计算这一步的奖励（标量）
    │       类比：工作人员打分——圈离目标越近分越高
    │
    ├─→ task.get_termination(physics)
    │       做什么：判断是否终止（None=继续, 0.0=成功, 1.0=超时）
    │       类比：判断游戏是否结束——套中了？时间到了？
    │
    └─→ 返回 TimeStep(MID or LAST, reward, discount, obs)
            如果继续：step_type=MID, discount=1.0
            如果成功终止：step_type=LAST, discount=0.0
            如果超时：step_type=LAST, discount=1.0

💡 记住这个时序图：它解释了 Task 的 9 个方法分别在什么时候被调用、做什么用。
你只需要实现 5 个必须的方法，其余 4 个可选方法有默认实现。

control.Task 里一共 9 个方法。哪些必须实现？实测最靠谱：

sorted(control.Task.__abstractmethods__)
# ['action_spec', 'before_step', 'get_observation', 'get_reward', 'initialize_episode']

5 个抽象方法必须实现。完整清单：

方法	是否必须	什么时候被调用
`initialize_episode(physics)`	必须★	`reset()` 时，重置状态
`before_step(action, physics)`	必须★	`step()` 里，把 action 写进 physics
`action_spec(physics)`	必须★	环境询问动作空间
`get_observation(physics)`	必须★	组装观测
`get_reward(physics)`	必须★	计算奖励
`observation_spec(physics)`	可选	不写也行 —— 基类会从 `get_observation` 的返回值自动推断
`after_step(physics)`	可选	物理步进之后
`get_termination(physics)`	可选	返回 `None`=继续 / `float`=终止
`step_spec(physics)`	可选	基类默认抛 `NotImplementedError`

实测签名：

before_step(self, action, physics)          ← 注意参数顺序！
Environment(self, physics, task, time_limit=inf, control_timestep=None,
            n_sub_steps=None, flat_observation=False, legacy_step=True)

⚠️ 本书在这里写错过一次：最初把 initialize_episode 标成「建议实现」、
把 observation_spec 标成「必须」—— 两个都标反了。
原因是看到报错文本被截断成 ... get_re... 就以为结束了。

实测结论：

缺 initialize_episode → TypeError: Can't instantiate abstract class NoInit with abstract method initialize_episode
缺 observation_spec → 完全没问题，env.observation_spec() 会自动推断出
dict_keys(['q'])（键名取自 get_observation 返回的字典）

调用顺序

env.reset()
  └─ task.initialize_episode(physics)        重置状态
  └─ task.get_observation(physics)           组装首个观测
  └─ 返回 TimeStep(FIRST, None, None, obs)

env.step(action)
  └─ task.before_step(action, physics)       写控制量
  └─ physics.step() × n_sub_steps            物理推进
  └─ task.after_step(physics)
  └─ task.get_observation(physics)           新观测
  └─ task.get_reward(physics)                奖励
  └─ task.get_termination(physics)           是否终止 / 超时
  └─ 返回 TimeStep(MID 或 LAST, reward, discount, obs)

⚠️ `before_step` 的参数顺序是 `(action, physics)`

网上不少旧教程写成 (physics, action)。写反了之后 action 变量里装的是 physics，
一调用就 AttributeError。实测签名就是 (self, action, physics)。

空 Task 会明确报缺哪些方法

class Empty(control.Task): pass
# TypeError: Can't instantiate abstract class Empty with abstract methods
#            action_spec, before_step, get_observation, get_reward,
#            initialize_episode

💡 这个报错很有用——它直接告诉你要实现哪 5 个方法。
⚠️ 但别像本书一样被截断的文本坑了：完整的 5 个里最后一个是 initialize_episode，
很容易在 get_reward, 后面就被截掉。
最稳妥的查法：sorted(control.Task.__abstractmethods__)。

16.4 完整例子：末端到达 Task

from collections import OrderedDict
import numpy as np
from dm_control.rl import control
from dm_env import specs


class ReachTask(control.Task):
    """让末端执行器尽量靠近一个目标点。"""

    def __init__(self, target, tol=0.05):
        self._target = np.asarray(target, dtype=float)
        self._tol = tol
        self._ee_site = mjcf_model.find("site", "end_effector")

    # ---------- 生命周期 ----------
    def initialize_episode(self, physics):
        physics.named.data.qpos[JN] = HOME          # 用名字重置
        physics.named.data.ctrl[AN] = HOME
        physics.forward()

    def before_step(self, action, physics):
        physics.named.data.ctrl[AN] = action        # action -> 控制量

    def after_step(self, physics):
        pass

    # ---------- 空间声明 ----------
    def action_spec(self, physics):
        return specs.BoundedArray(shape=(6,), dtype=np.float64,
                                  minimum=-3.0, maximum=3.0,
                                  name="joint_target")

    def observation_spec(self, physics):
        return OrderedDict([
            ("joints", specs.Array((6,), np.float64, "joint_pos")),
            ("ee", specs.Array((3,), np.float64, "ee_pos")),
            ("target", specs.Array((3,), np.float64, "target_pos")),
        ])

    # ---------- 数据 ----------
    def get_observation(self, physics):
        return OrderedDict(
            joints=physics.named.data.qpos[JN].copy(),
            ee=physics.bind(self._ee_site).xpos.copy(),
            target=self._target.copy())

    def get_reward(self, physics):
        d = np.linalg.norm(physics.bind(self._ee_site).xpos - self._target)
        return float(-d)                            # 距离越近奖励越高

    def get_termination(self, physics):
        d = np.linalg.norm(physics.bind(self._ee_site).xpos - self._target)
        return 0.0 if d < self._tol else None

逐行讲解：ReachTask 完整实现

下面对 ReachTask 的每个方法做逐行讲解，说明输入、输出、内部逻辑和设计理由。

class ReachTask(control.Task):
    """让末端执行器尽量靠近一个目标点。"""

    def __init__(self, target, tol=0.05):
        # 构造函数：保存任务参数
        # 输入：
        #   target - 目标点坐标 [x, y, z]，单位米
        #   tol - 成功容忍距离，单位米（默认 0.05m = 5cm）
        # 内部做了什么：
        #   1. 把 target 转成 numpy 数组（方便后续计算）
        #   2. 保存容忍距离
        #   3. 预先查找末端执行器的 site 元素（避免每次 step 都 find）
        self._target = np.asarray(target, dtype=float)
        self._tol = tol
        self._ee_site = mjcf_model.find("site", "end_effector")
        # 为什么预先 find？因为 find 有查找开销，在 __init__ 里找一次
        # 比在每个 step 的 get_observation 里都找一次高效得多。

    # ---------- 生命周期 ----------
    def initialize_episode(self, physics):
        # 每个 episode 开始时被调用（对应 env.reset()）
        # 输入：physics - 仿真句柄
        # 做什么：把机械臂重置到 HOME 位姿
        physics.named.data.qpos[JN] = HOME          # 用名字重置关节角
        physics.named.data.ctrl[AN] = HOME          # 控制指令也设为 HOME（避免阶跃）
        physics.forward()                             # 必须 forward，否则正向运动学不更新
        # 为什么 ctrl 也要设？因为如果只设 qpos 不设 ctrl，PD 控制器会从
        # 旧的 ctrl 值突然跳到新位置，产生很大的初始力矩。设成一样的值更平滑。

    def before_step(self, action, physics):
        # 每个 step 开始时被调用（对应 env.step(action)）
        # 输入：
        #   action - Agent 给出的动作，形状 (6,)，对应 6 个关节的目标角度
        #   physics - 仿真句柄
        # ⚠️ 参数顺序是 (action, physics)，不是 (physics, action)！
        # 做什么：把 action 写进控制量数组
        physics.named.data.ctrl[AN] = action        # action -> 控制量
        # 这里没有 clip，因为假设 Agent 输出的动作已经在 action_spec 范围内。
        # 但实际上 dm_env 不校验越界，安全起见应该 np.clip（见 16.6 节）。

    def after_step(self, physics):
        # 物理步进之后被调用（可选，默认什么都不做）
        # 用途：记录中间状态、做一些后处理
        pass

    # ---------- 空间声明 ----------
    def action_spec(self, physics):
        # 声明动作空间：告诉 Agent 动作长什么样、范围是多少
        # 输入：physics（通常不用，但接口要求传入）
        # 输出：specs.BoundedArray 对象
        return specs.BoundedArray(
            shape=(6,),           # 6 个关节，每个关节一个目标角度
            dtype=np.float64,     # 双精度浮点数
            minimum=-3.0,         # 最小角度 -3.0 弧度（约 -172°）
            maximum=3.0,          # 最大角度 +3.0 弧度（约 +172°）
            name="joint_target")  # 名字，用于调试
        # 为什么是 ±3.0 弧度？因为我们的 6 轴机械臂关节范围最大是 ±180°（J1），
        # ±3.0 rad ≈ ±172°，留了一点余量。实际项目中应该用每个关节的真实范围。

    def observation_spec(self, physics):
        # 声明观测空间（可选——不写的话基类会从 get_observation 自动推断）
        # 输出：OrderedDict，每个键对应一个观测项的 spec
        return OrderedDict([
            ("joints", specs.Array((6,), np.float64, "joint_pos")),
            # 6 个关节的当前角度，无界（关节角本身有物理范围，但 spec 不限制）
            ("ee", specs.Array((3,), np.float64, "ee_pos")),
            # 末端执行器的世界坐标 [x, y, z]
            ("target", specs.Array((3,), np.float64, "target_pos")),
            # 目标点的世界坐标 [x, y, z]
        ])

    # ---------- 数据 ----------
    def get_observation(self, physics):
        # 组装观测：把物理状态打包成 Agent 能理解的字典
        # 输入：physics
        # 输出：OrderedDict，键名必须与 observation_spec 一致
        return OrderedDict(
            joints=physics.named.data.qpos[JN].copy(),
            # 6 个关节角。为什么 .copy()？因为 named 视图是共享内存的，
            # 如果不 copy，后续 physics.step() 会修改这个数组，导致 Agent
            # 拿到的观测被意外篡改。
            ee=physics.bind(self._ee_site).xpos.copy(),
            # 末端执行器位置。用 bind 而不是 named，因为手里有 site 元素对象。
            target=self._target.copy(),
            # 目标位置。也要 copy，避免外部修改影响内部状态。
        )

    def get_reward(self, physics):
        # 计算奖励：告诉 Agent 这一步做得好不好
        # 输入：physics
        # 输出：float 标量
        d = np.linalg.norm(physics.bind(self._ee_site).xpos - self._target)
        # 计算末端到目标的欧氏距离（L2 范数）
        # d = sqrt((ex-tx)^2 + (ey-ty)^2 + (ez-tz)^2)
        return float(-d)
        # 奖励 = -距离。距离越近，奖励越大（越接近 0）；距离越远，奖励越小（越负）。
        # 这是最朴素的"密集奖励"（dense reward）——每一步都有梯度信号。
        # 缺点：Agent 可能学会"慢慢靠近"而不是"快速到达"，因为每一步的奖励差异不大。

    def get_termination(self, physics):
        # 判断是否终止：None=继续, 0.0=成功终止, 1.0=超时截断
        d = np.linalg.norm(physics.bind(self._ee_site).xpos - self._target)
        return 0.0 if d < self._tol else None
        # 如果距离小于容忍值（5cm），返回 0.0 表示成功终止。
        # 否则返回 None 表示继续。
        # 注意：超时截断（返回 1.0）由 Environment 的 time_limit 自动处理，
        # 不需要在 Task 里写。

💡 奖励函数设计推导：
我们选择 reward = -distance 而不是 reward = 1/distance 或 reward = exp(-distance)，原因是：

简单：线性函数，梯度恒定，Agent 每靠近 1cm 奖励增加 0.01，信号清晰。
无奇异点：1/distance 在 distance=0 时无穷大，不稳定。
范围可控：距离最大约 1m，奖励范围约 [-1, 0]，不会出现极端值。
在我们的机械臂项目中：末端工作空间约 0.6m，奖励范围约 [-0.6, 0]，
一个 episode（200 步）的累计奖励范围约 [-120, 0]，与实测数据一致。

装配：

env = control.Environment(physics, task, control_timestep=0.02, time_limit=4.0)

实测：

type(env) = Environment   是 dm_env.Environment? True
env.physics is physics ? True

action_spec: BoundedArray(shape=(6,), dtype=float64,
                          name='joint_target', minimum=-3.0, maximum=3.0)
observation_spec:
  joints   shape=(6,)  dtype=float64
  ee       shape=(3,)  dtype=float64
  target   shape=(3,)  dtype=float64

💡 观测里的 ee 用 physics.bind(site).xpos 取的，
与第 14 章的 sensordata['ee_pos'] 实测一致 —— 两种写法都行。

16.5 `control.Environment` 的构造参数 + 两条硬约束

control.Environment(physics, task, time_limit=inf, control_timestep=None,
                    n_sub_steps=None, flat_observation=False, legacy_step=True)

⚠️ 约束①：`n_sub_steps` 与 `control_timestep` 互斥

control.Environment(physics, task, control_timestep=0.02, n_sub_steps=10, ...)
# ValueError: Both n_sub_steps and control_timestep were supplied.

⚠️ 约束②：`control_timestep` 必须是仿真步长的整数倍

control.Environment(physics, task, control_timestep=0.021, ...)
# ValueError: Control timestep (0.021) must be an integer multiple of
#             physics timestep (0.002)

本项目 timestep = 0.002，0.02 / 0.002 = 10 ✅

魔法数字来源解释

数字	值	来源	为什么选这个值
`timestep`	0.002 s	模型 XML 的 `<option timestep="0.002"/>`	MuJoCo 推荐的仿真步长，2ms 足以稳定模拟刚体碰撞和关节动力学
`control_timestep`	0.02 s	代码中手动设置	20ms = 50Hz 控制频率，与真实工业机械臂的控制周期一致（Epson CX4 系列默认 500Hz~1kHz，但 50Hz 对 RL 训练足够）
`n_sub_steps`	10	`control_timestep / timestep = 0.02 / 0.002`	每个控制周期内物理引擎推进 10 步，保证物理精度
`time_limit`	4.0 s	代码中手动设置	4 秒 = 200 个控制步，足够机械臂从 HOME 位姿到达工作空间内任意目标点
`action range`	±3.0 rad	`action_spec` 中设置	约 ±172°，覆盖 J1（±180°）的大部分范围，留余量避免极限位置

💡 控制频率 vs 仿真步长（复习第 12 章）：

仿真步长（timestep）决定物理精度——越小越精确，但计算越慢。
控制周期（control_timestep）决定 Agent 多久做一次决策——越小响应越快，但需要更频繁的神经网络推理。
两者的比值（n_sub_steps）决定每个控制步内物理推进多少步。
在我们的项目中：2ms 仿真 + 20ms 控制 = 10 倍物理过采样，这是机器人仿真的标准配置。

两种指定方式（实测）

方式	一个 step 推进
`control_timestep=0.01`	0.0100 s（5 个物理步）
`control_timestep=0.02`	0.0200 s（10 个物理步）
`control_timestep=0.05`	0.0500 s（25 个物理步）
`n_sub_steps=1`	0.0020 s（1 个物理步）
`n_sub_steps=5`	0.0100 s（5 个物理步）
`n_sub_steps=10`	0.0200 s（10 个物理步）
`n_sub_steps=20`	0.0400 s（20 个物理步）

💡 推荐 control_timestep（按时间思考，跟第 12 章讲的「控制周期 vs 仿真步长」一致）。
n_sub_steps 适合你想精确定「一步 = 几个物理步」时。

16.6 观测 / 动作空间设计

观测 / 动作 / 奖励数据流图

┌─────────────────────────────────────────────────────────────────────┐
│                        Agent（策略网络）                              │
│                                                                       │
│  输入: observation (观测字典)                                        │
│    {                                                                  │
│      "joints": shape=(6,)   ← 6个关节当前角度                       │
│      "ee":     shape=(3,)   ← 末端执行器世界坐标                    │
│      "target": shape=(3,)   ← 目标点世界坐标                        │
│    }                                                                  │
│    总维度 = 6 + 3 + 3 = 12                                           │
│                                                                       │
│  输出: action (动作向量)                                              │
│    shape=(6,)  ← 6个关节的目标角度                                   │
│    range=[-3.0, +3.0] rad                                            │
│                                                                       │
└───────────────┬───────────────────────────────────┬─────────────────┘
                │ action                              │ reward (标量)
                │ [a1,a2,a3,a4,a5,a6]               │ r = -||ee - target||
                ▼                                     │
┌───────────────────────────────────────────────────▼─────────────────┐
│                        Environment（环境）                             │
│                                                                       │
│  before_step(action):                                                 │
│    physics.named.data.ctrl[AN] = action  ← 动作→控制量              │
│                                                                       │
│  physics.step(n_sub_steps=10):                                       │
│    物理推进 10 步（每步 0.002s，共 0.02s）                          │
│                                                                       │
│  get_observation(physics):                                            │
│    joints = qpos[0:6]      ← 关节角                                  │
│    ee     = site_xpos[ee]  ← 末端位置                                │
│    target = [固定值]        ← 目标位置                                │
│                                                                       │
│  get_reward(physics):                                                 │
│    d = ||ee - target||     ← 欧氏距离                                │
│    reward = -d              ← 距离越近奖励越高                       │
│                                                                       │
│  get_termination(physics):                                            │
│    if d < 0.05: return 0.0  ← 成功终止（5cm以内）                   │
│    else: return None       ← 继续                                    │
│                                                                       │
└───────────────────────────────────────────────────────────────────────┘

💡 数据流的关键洞察：

观测是环境的"成绩单"——它告诉 Agent 当前状态，但不直接告诉 Agent “该怎么做”。
动作是 Agent 的"答卷"——Agent 根据观测决定动作，环境执行动作后状态改变。
奖励是环境的"评分"——它告诉 Agent 刚才的动作好不好，但不告诉 Agent “怎么才能更好”。
Agent 不知道环境内部——它不知道物理引擎怎么算、碰撞怎么检测，它只通过这三个通道与环境交互。

这就是强化学习的核心：Agent 在"黑箱"环境中，通过试错（trial and error）学会最大化累计奖励。

`specs` 常用类型

specs.Array(shape, dtype, name)                    # 无界连续量
specs.BoundedArray(shape, dtype, minimum, maximum, name)   # 有界连续量
specs.DiscreteArray(num_values, dtype, name)       # 离散动作

⚠️ BoundedArray 的参数顺序是 (shape, dtype, minimum, maximum, name)。
位置写错不会报错，只是上下界全乱 —— 很隐蔽。

每个维度不同范围（真实机器人几乎都是这样）

lo = np.array([-3.0, -2.0, -2.5, -3.0, -2.0, -3.0])
hi = np.array([ 3.0,  2.0,  2.5,  3.0,  2.0,  3.0])
specs.BoundedArray((6,), np.float64, lo, hi, "per_joint")

minimum / maximum 可以是数组。

⚠️ `action` 越界时 dm_env 不强制校验

实测：

step([99]*6)          -> 没报错（需要你自己 clip）
step(np.zeros(3))     -> ValueError: shape mismatch: value array of shape (3,)
                                     could not be broadcast to ...

形状错会报，数值越界不会报。做 RL 时记得在 before_step 里 np.clip。

`flat_observation=True`：拼成一个向量

flat=False -> keys=['joints', 'ee', 'target']
flat=True  -> keys=['observations']  observations.shape=(12,)

喂给 MLP 时用 flat=True 更方便；但字典形式更好调试。

16.7 跑完整 episode

`reset()` 返回什么

step_type = FIRST(0)
reward    = None
discount  = None
observation = {joints: [0. 0.2 0.6 0. 0.3 0.],
               ee: [-0.2343  0.      0.9051],
               target: [-0.3  0.3  0.1]}

三种策略对比（实测）

策略	步数	累计奖励	末端最终距目标
随机（均匀 −1~1）	200	−131.314	0.6667 m
固定关节角	200	−196.578	0.9854 m
雅可比 IK（DLS）	34	−9.761	0.0474 m ✅

⚠️ 本书的一个教训：我最初写了个「手调关节角」的策略当作「会收敛的示例」，
实测跑完发现末端离目标越来越远（最终 0.99 m）—— 是发散的。
换成第 6 章学过的阻尼最小二乘（DLS）后，34 步就收敛到 0.047 m 并触发成功终止。

别把「看起来在动」当成「在收敛」，永远用数字验证。

IK 策略逐行讲解（复习第 6 章 DLS）

这个策略不依赖学习，而是用**雅可比矩阵 + 阻尼最小二乘（DLS）**直接计算关节增量。
它是验证环境是否正确的"黄金标准"——如果 IK 策略都不能收敛，说明环境有 bug。

def ik_policy(time_step, rng, step_size=0.5, lam=0.05):
    # 输入：
    #   time_step - 当前的 TimeStep（包含 observation）
    #   rng - 随机数生成器（这个策略不用，但接口要求传入）
    #   step_size - 步长系数（0.5 = 每次走误差的一半，保守但稳定）
    #   lam - 阻尼系数（DLS 的关键参数，避免奇异点附近关节速度爆炸）
    # 输出：
    #   shape=(6,) 的动作向量（6个关节的目标角度）

    m, d = physics.model.ptr, physics.data.ptr
    # 拿原生对象。因为 mj_jacSite 是原生 API，只接受原生 MjModel/MjData。
    # 这是第 15 章讲的 .ptr 陷阱的实际应用。

    jacp = np.zeros((3, m.nv))
    jacr = np.zeros((3, m.nv))
    # 分配雅可比矩阵空间。
    # ⚠️ mujoco 3.x 要求 (3, nv) 二维数组，不是扁平的 3*nv。
    # jacp = 位置雅可比（3行：x,y,z；nv列：每个关节的贡献）
    # jacr = 旋转雅可比（这个策略不用，但 mj_jacSite 要求同时传入）

    mujoco.mj_jacSite(m, d, jacp, jacr, ee_site_id)
    # 计算末端执行器 site 的雅可比矩阵。
    # 雅可比的物理意义：J * dq = dx（关节微小变化 dq 导致末端微小位移 dx）
    # 所以 J 的第 i 列表示：第 i 个关节转 1 弧度，末端会移动多少。

    J = jacp[:, :6]
    # 只取前 6 列（对应 6 个臂关节）。
    # 为什么？因为我们的模型有 8 个 actuator（6个臂关节 + 2个夹爪），
    # nv=14（6个铰链 + 1个freejoint占6维 + 2个夹爪），但我们只驱动前 6 个臂关节。
    # 夹爪和自由物体不在这个任务的控制范围内。

    err = time_step.observation["target"] - time_step.observation["ee"]
    # 计算位置误差：目标位置 - 当前末端位置。
    # err 是 shape=(3,) 的向量，指向从末端到目标的方向。

    dq = J.T @ np.linalg.solve(J @ J.T + lam**2 * np.eye(3), err)
    # ⭐ 阻尼最小二乘（DLS）的核心公式。
    # 推导过程：
    #   我们想找 dq 使得 J * dq ≈ err（关节变化导致末端移动 err）
    #   普通最小二乘：dq = J^T * (J * J^T)^{-1} * err
    #   但当 J * J^T 接近奇异（末端在奇异位形）时，逆矩阵会爆炸。
    #   DLS 的改进：在 J * J^T 上加一个阻尼项 lam^2 * I：
    #     dq = J^T * (J * J^T + lam^2 * I)^{-1} * err
    #   这样即使 J * J^T 奇异，(J * J^T + lam^2 * I) 也一定可逆。
    #   lam 越大，阻尼越强，运动越慢但越稳定；lam 越小，收敛越快但可能震荡。
    #   lam=0.05 是经验值，在我们的机械臂上表现稳定。

    q = np.asarray(time_step.observation["joints"]) + step_size * dq
    # 新的关节目标角 = 当前关节角 + 步长 * 关节增量。
    # step_size=0.5 意味着每次只走目标增量的一半，避免超调和震荡。
    # 这是一种"保守控制"——虽然收敛慢一点，但不会因为一步走太大而发散。

    return np.clip(q, -3.0, 3.0)
    # 最后裁剪到动作范围 [-3.0, +3.0]。
    # 虽然 action_spec 声明了范围，但 dm_env 不强制校验，所以必须自己 clip。
    # 这是 16.6 节讲的"动作越界不报错"陷阱的防御措施。

💡 为什么 IK 策略 34 步就能收敛，而随机策略 200 步还在乱晃？
因为 IK 策略利用了模型知识（雅可比矩阵告诉它"动哪个关节、动多少"），
而随机策略完全是盲目的。这就是强化学习要解决的问题：
让 Agent 通过试错自己学会类似 IK 的策略，而不需要人工推导雅可比。
但在训练初期，IK 策略是验证环境正确性的最佳工具。

IK 策略的核心（复习第 6 章）：

def ik_policy(time_step, rng, step_size=0.5, lam=0.05):
    m, d = physics.model.ptr, physics.data.ptr           # 原生对象
    # ⚠️ mujoco 3.x 的 mj_jacSite 要 (3, nv) 二维数组，不是扁平的 3*nv
    jacp = np.zeros((3, m.nv))
    jacr = np.zeros((3, m.nv))
    mujoco.mj_jacSite(m, d, jacp, jacr, ee_site_id)
    J = jacp[:, :6]                                      # 只驱动 6 个臂关节
    err = time_step.observation["target"] - time_step.observation["ee"]
    dq = J.T @ np.linalg.solve(J @ J.T + lam**2 * np.eye(3), err)
    q = np.asarray(time_step.observation["joints"]) + step_size * dq
    return np.clip(q, -3.0, 3.0)

`get_termination` 返回值对照

返回	效果
`None`	继续（`step_type=MID`, `discount=1.0`）
`0.0`	成功终止（`LAST`, `discount=0.0`）
`1.0`	超时截断（`LAST`, `discount=1.0`，可 bootstrapping）

💡 这个设计很讲究：
discount=0.0 告诉算法「episode 真的结束了，别再往回传播价值」；
discount=1.0 表示「只是超时，后面本来还能继续，可以继续传播」。

16.8 🔥 本书踩坑实录

坑 1：包装类与 `launch_passive` 不兼容

第 15 章讲过 physics.model 是包装类。后果在这里：

❌ launch_passive(physics.model, physics.data)
   -> ValueError: `model` is not a mujoco.MjModel:
      got <dm_control.mujoco.wrapper.core.MjModel object at 0x...>

✅ launch_passive(physics.model.ptr, physics.data.ptr) -> 成功

💡 想一边跑 RL 一边看画面，正确姿势：

with mujoco.viewer.launch_passive(physics.model.ptr, physics.data.ptr) as v:
    ts = env.reset()
    while not ts.last():
        ts = env.step(policy(ts))
        v.sync()

坑 2：mujoco 3.x 的对象枚举搬到了 `mjtObj` 下

mujoco.mjOBJ_SITE          # ❌ AttributeError: module 'mujoco' has no attribute 'mjOBJ_SITE'
mujoco.mjtObj.mjOBJ_SITE   # ✅

同样地，传感器类型要用 mujoco.mjtSensor.mjSENS_FRAMEPOS 反查
（第 11 章猜编号猜错过一次，第 14 章又强调了一遍）。

坑 3：`before_step` 的参数顺序

def before_step(self, action, physics):    # ✅ 实测签名

不是 (physics, action)。

坑 4 / 5：`control_timestep` 要整数倍，且与 `n_sub_steps` 互斥

见 16.5，两条都有明确的 ValueError。

坑 6：空 Task 报抽象方法未实现（5 个，不是 4 个）

TypeError: Can't instantiate abstract class Empty with abstract methods
           action_spec, before_step, get_observation, get_reward,
           initialize_episode

用 sorted(control.Task.__abstractmethods__) 查最准：
['action_spec', 'before_step', 'get_observation', 'get_reward', 'initialize_episode']

反过来，observation_spec 不在里面 —— 不写也能跑，基类会自动推断。

坑 7：`step_spec()` 基类默认抛 `NotImplementedError`

用不到就不用管。

坑 8：`mujoco.mj_jacSite` 要 `(3, nv)` 二维数组

jacp = np.zeros(3 * m.nv)      # ❌ TypeError: jacp should be of shape (3, nv)
jacp = np.zeros((3, m.nv))     # ✅

16.9 常见错误速查

报错 / 现象	原因	解决
`TypeError: 'NoneType' + float`	用了 `reset()` 的 `reward`（是 None）	`ts.reward or 0.0`
`ValueError: Both n_sub_steps and control_timestep`	两个都传了	只传一个
`must be an integer multiple of physics timestep`	控制周期不是仿真步长整数倍	改成整数倍，如 0.02 / 0.002
`ValueError: 'model' is not a mujoco.MjModel`	传了包装类	传 `physics.model.ptr`
`AttributeError: module 'mujoco' has no attribute 'mjOBJ_SITE'`	枚举位置变了	`mujoco.mjtObj.mjOBJ_SITE`
`before_step` 里 AttributeError	参数顺序写反	`(action, physics)`
`TypeError: Can't instantiate abstract class`	Task 方法没实现	补 5 个抽象方法
action 越界没报错、机器人乱飞	dm_env 不校验数值	在 `before_step` 里 `np.clip`
`TypeError: jacp should be of shape (3, nv)`	雅可比数组形状	用 `np.zeros((3, m.nv))`
策略看着在动但不收敛	没用数值验证	打印距离曲线，别只看画面

16.10 动手练

练习详解与答案提示

练习 1：TimeStep

目标：理解 reset() 和 step() 返回的 TimeStep 结构
关键步骤：
1. ts = env.reset() → 打印 ts.step_type（应为 FIRST）、ts.reward（应为 None）、ts.discount（应为 None）
2. ts = env.step(np.zeros(6)) → 打印 ts.step_type（应为 MID）、ts.reward（应为浮点数）、ts.discount（应为 1.0）
常见错误：直接 total += ts.reward 在 reset 后会 TypeError，因为 reward 是 None

练习 2：Task 协议

目标：掌握 5 个必须实现的抽象方法，理解 observation_spec 是可选的
关键步骤：
1. 写一个最小 Task，只实现 5 个抽象方法 → 能构造 Environment
2. 删掉 initialize_episode → TypeError: Can't instantiate abstract class
3. 不写 observation_spec → 完全没问题，基类从 get_observation 自动推断
验证方法：sorted(control.Task.__abstractmethods__) 查看必须实现的方法列表

练习 3：时间关系

目标：理解 control_timestep 与 timestep 的关系
关键步骤：
1. 用 control_timestep=0.01 跑一步 → physics.data.time 增加 0.01（5 个物理步）
2. 用 control_timestep=0.02 跑一步 → 增加 0.02（10 个物理步）
3. 用 control_timestep=0.05 跑一步 → 增加 0.05（25 个物理步）
公式：物理步数 = control_timestep / timestep = 0.02 / 0.002 = 10

练习 4：空间设计

目标：掌握每维不同范围的 BoundedArray，理解 dm_env 不校验越界
关键步骤：
1. lo = np.array([-3.0, -2.0, -2.5, -3.0, -2.0, -3.0])
2. hi = np.array([3.0, 2.0, 2.5, 3.0, 2.0, 3.0])
3. specs.BoundedArray((6,), np.float64, lo, hi, "per_joint")
4. 用 np.full(6, 99.0) 试探 → 不报错（dm_env 不校验数值越界）
防御措施：在 before_step 里 np.clip(action, lo, hi)

练习 5：跑 episode

目标：对比三种策略的收敛效果
关键步骤：
1. 随机策略：rng.uniform(-1.0, 1.0, size=6) → 200 步，累计奖励 -131.3，最终距离 0.667m
2. 固定关节角：[0.0, 0.6, 0.9, 0.0, 0.4, 0.0] → 200 步，累计奖励 -196.6，最终距离 0.985m
3. IK 策略（DLS）：雅可比 + 阻尼最小二乘 → 34 步，累计奖励 -9.8，最终距离 0.047m ✅
关键教训：别把"看起来在动"当成"在收敛"，永远用数字验证

练习 6：踩坑复现

目标：验证包装类与 launch_passive 的不兼容
关键步骤：
1. mujoco.viewer.launch_passive(physics.model, physics.data) → ValueError: model is not a mujoco.MjModel
2. mujoco.viewer.launch_passive(physics.model.ptr, physics.data.ptr) → 成功
根因：physics.model 是 dm_control 的包装类，不是原生 mujoco.MjModel

16.11 进阶实战：完整取放仿真的工程模式

前面讲了 Task / Environment / specs 的基础 API。但在真实项目中（如
scripts/dm_control_smooth_simulation.py），
一个完整的仿真还需要解决几个工程问题：策略如何维护状态、抓取如何实现、
dm_control 的 Physics 怎么和原生 viewer 配合、怎么录屏和无头验证。
本节把这些模式一次讲清。

16.11.1 Policy 与 Task 分离：为什么需要 Policy 类

在 16.4 的 ReachTask 中，策略是一个函数 ik_policy(time_step, rng)。
但在取放任务中，策略需要维护状态（当前轨迹进度、是否携带物体、夹爪位置），
函数式写法就不够了。

标准做法是把策略封装成可调用类，与 Task 分离：

┌──────────────┐     action      ┌──────────────┐
│  Policy 类    │ ──────────────→ │  Task 类     │
│ (生成动作)    │                 │ (环境接口)    │
│              │ ←── observation ─┤              │
│ - trajectory │                 │ - before_step │
│ - ik solver  │                 │ - get_obs     │
│ - weld_id    │                 │ - get_reward  │
└──────────────┘                 └──────────────┘

Task 只做一件事：把 action 写进 physics、组装观测、算奖励。
Policy 做决策：根据当前物理状态和轨迹进度，算出下一个 action。

class SmoothPolicy:
    """把「当前轨迹点 → IK → 关节目标」算成 dm_control 的 action。"""

    def __init__(self, ik, trajectory, weld_id):
        self.ik = ik
        self.traj = trajectory
        self.weld_id = weld_id

    def __call__(self, physics):
        # 1. 读当前关节角（作为 IK 的初始猜测）
        live_q = physics.data.qpos[:6].copy()
        # 2. 轨迹推进一个控制周期，返回目标末端位置 + 夹爪指令
        target_pos, gripper_cmd = self.traj.sample(CONTROL_DT)
        # 3. IK 求解关节角
        q, ok, err = self.ik.solve(target_pos, initial_qpos=live_q)
        # 4. 数值保护：NaN/Inf 时回退到当前姿态
        if not np.all(np.isfinite(q)):
            q = live_q
        # 5. 组装 action（6 关节 + 2 夹爪）
        action = np.zeros(8, dtype=np.float32)
        action[:6] = q.astype(np.float32)
        action[6] = action[7] = float(gripper_cmd)
        # 6. 抓取约束：夹爪闭合 → 激活 weld；张开 → 释放
        if self.weld_id >= 0:
            physics.data.eq_active[self.weld_id] = 1 if gripper_cmd > 0.02 else 0
        return action

对应的 Task 就非常薄——只负责接口：

class PickPlaceTask(control.Task):
    def __init__(self, policy, ee_id, obj_id):
        self.policy = policy
        self.ee_id = ee_id
        self.obj_id = obj_id

    def initialize_episode(self, physics):
        physics.data.qpos[:6] = INITIAL_POSE
        physics.data.ctrl[:6] = INITIAL_POSE
        physics.data.qpos[6] = physics.data.qpos[7] = GRIPPER_OPEN
        physics.data.ctrl[6] = physics.data.ctrl[7] = GRIPPER_OPEN
        physics.forward()
        self.policy.traj.reset()          # ⭐ 重置策略状态

    def before_step(self, action, physics):
        physics.data.ctrl[:] = action     # 只做一件事：写控制量

    def after_step(self, physics):
        pass

    def action_spec(self, physics):
        return specs.BoundedArray(shape=(8,), dtype=np.float32,
                                  minimum=-np.inf, maximum=np.inf, name="action")

    def get_observation(self, physics):
        return OrderedDict(
            ee_position=physics.named.data.site_xpos["end_effector"].copy(),
            object_position=physics.named.data.site_xpos["object_center"].copy(),
            gripper=np.array([physics.data.qpos[6], physics.data.qpos[7]],
                             dtype=np.float32),
        )

    def get_reward(self, physics):
        return 0.0    # 脚本驱动的任务不需要奖励

    def get_termination(self, physics):
        return None   # 由外层循环依据 policy.traj.done 结束

💡 关键设计决策：

轨迹状态放在 Policy 里，不在 Task 里——因为轨迹是"决策逻辑"，不是"环境规则"。
initialize_episode 里调用 policy.traj.reset()——确保每个 episode 轨迹从头开始。
get_termination 返回 None——脚本驱动的任务由外层 while not policy.traj.done 控制结束，
不依赖 Environment 的 time_limit。
action_spec 用 -inf/+inf——因为 action 是 IK 算出来的关节角，理论上无界（实际有物理限制）。
观测用 physics.named.data.site_xpos["end_effector"]——按名字读 site 的世界坐标，
比 bind(site).xpos 更直接（不需要预先 find 元素）。

16.11.2 用 weld 约束实现抓取

在 MuJoCo 中，抓取有两种实现方式：

方式	原理	优点	缺点
接触力抓取	靠夹爪 geom 与物体的接触摩擦力夹住	物理真实	不稳定，容易滑掉
weld 约束抓取	夹爪闭合时激活 equality 约束，把物体"焊"在末端	稳定可控	不完全物理真实

仿真中最常用的是 weld 约束抓取。dm_control 中操作方法：

# 1. 在 XML 中定义 equality 约束（name="grasp_weld"）
# <equality>
#   <weld name="grasp_weld" body1="target_object" body2="gripper_mount"/>
# </equality>

# 2. 查约束 id（用原生模型查，因为 dm_control 的 named 不直接支持 equality）
weld_id = mujoco.mj_name2id(raw_model, mujoco.mjtObj.mjOBJ_EQUALITY, "grasp_weld")

# 3. 在 Policy 中激活/释放约束
if gripper_cmd > 0.02:              # 夹爪闭合
    physics.data.eq_active[weld_id] = 1    # 激活 weld → 物体被焊住
else:                                  # 夹爪张开
    physics.data.eq_active[weld_id] = 0    # 释放 weld → 物体自由

💡 注意：eq_active 是布尔数组（实测 dtype=bool），赋 0 / 1 或
True / False 均可。

为什么用原生模型查 id？ dm_control 的 named 命名访问主要覆盖
qpos / qvel / ctrl / xpos / site_xpos / sensordata 等数据字段，
不直接覆盖 equality 约束的 id 查询。所以用
mujoco.mj_name2id(raw_model, mjtObj.mjOBJ_EQUALITY, name) 查最可靠。
同一 XML 编译出的模型，equality id 顺序一致，可以直接用于
physics.data.eq_active[idx]。

16.11.3 双模型模式：dm_control Physics + 原生 MjModel

在真实项目中，你经常需要同时使用 dm_control 的便利功能（named、Task）和
原生 MuJoCo 的功能（IK 求解器、viewer、name2id 查询）。但 mujoco.viewer
和一些原生 C 函数只接受原生 MjModel/MjData，不接受 dm_control 的包装类。

解决方案：双模型模式——同时维护一份 dm_control Physics 和一份原生 MjModel，
两者从同一个 XML 编译，id 完全对应。

from dm_control import mjcf
import mujoco

# ① dm_control 端：用于 Task / Environment / named 访问
root = mjcf.from_path(str(SIM_MJCF))
physics = mjcf.Physics.from_mjcf_model(root)

# ② 原生端：用于 IK / name2id / viewer / Renderer
raw_model = mujoco.MjModel.from_xml_path(str(SIM_MJCF))

# ③ 两者 id 完全对应（同一 XML 编译结果一致）
ee_id = mujoco.mj_name2id(raw_model, mujoco.mjtObj.mjOBJ_SITE, "end_effector")
# 这个 ee_id 可以直接用于 physics.data.site_xpos[ee_id]

┌─────────────────────────────────────────────────────┐
│              同一个 XML 文件                           │
│         cx4_a601c_simulation.xml                     │
└──────────────────────┬──────────────────────────────┘
                       │ 编译
          ┌────────────┴────────────┐
          ▼                         ▼
┌─────────────────┐       ┌─────────────────┐
│ dm_control      │       │ 原生 mujoco     │
│ Physics         │       │ MjModel/MjData  │
│                 │       │                 │
│ ✅ named 访问   │       │ ✅ mj_name2id   │
│ ✅ Task/Env     │       │ ✅ mj_jacSite   │
│ ✅ bind()       │       │ ✅ viewer        │
│ ❌ viewer       │       │ ✅ Renderer      │
│ ❌ 原生 C 函数   │       │ ❌ named 访问    │
└─────────────────┘       └─────────────────┘
          │                         │
          └───────────┬─────────────┘
                      ▼
              id 完全对应
        （joint/site/equality 顺序一致）

💡 什么时候需要双模型？

需要用 mujoco.viewer.launch_passive 看画面（只接受原生对象）
需要用 mujoco.mj_jacSite 等原生 C 函数算雅可比（IK 用）
需要用 mujoco.mj_name2id 查 equality/site 的 id
需要用 mujoco.Renderer 离屏渲染（录屏用）

什么时候不需要？ 如果只用 dm_control 的 physics.render() 离屏渲染、
不用 viewer、不用原生 IK，那么单模型（dm_control Physics + .ptr）就够了。
第 15 章讲的 .ptr 方法适用于简单场景，双模型适用于复杂项目。

16.11.4 viewer 渲染同步模式

mujoco.viewer.launch_passive 只接受原生 MjModel/MjData，但仿真由
dm_control 的 env.step() 驱动。解决方案：维护一份原生 render_data，
每帧把 dm_control 的仿真状态复制过去并 mj_forward。

# ① 创建原生渲染副本
render_data = mujoco.MjData(raw_model)

def sync_render():
    """把 dm_control 的仿真状态同步到原生渲染副本。"""
    render_data.qpos[:] = physics.data.qpos[:]
    render_data.qvel[:] = physics.data.qvel[:]
    render_data.ctrl[:] = physics.data.ctrl[:]
    # equality 约束状态也要同步（抓取 weld 的激活状态）
    ea = physics.data.eq_active
    if ea is not None:
        n = min(render_data.eq_active.shape[0], ea.shape[0])
        render_data.eq_active[:n] = ea[:n]
    mujoco.mj_forward(raw_model, render_data)

# ② 主循环
env.reset()
sync_render()

with mujoco.viewer.launch_passive(raw_model, render_data) as viewer:
    # 设置相机（Z-up 约定：azimuth=90 相机在 -Y 侧 = 工作区正面；
    # elevation 是【视线】仰角，负值=相机在上方俯视，正值会钻到地下）
    viewer.cam.type = mujoco.mjtCamera.mjCAMERA_FREE
    viewer.cam.azimuth = 90.0
    viewer.cam.elevation = -20.0
    viewer.cam.distance = 2.6
    viewer.cam.lookat = [-0.10, 0.0, 0.28]
    viewer.sync()

    while viewer.is_running() and step_count < max_steps and not policy.traj.done:
        action = policy(physics)       # Policy 生成 action
        env.step(action)                # dm_control 推进仿真
        sync_render()                   # ⭐ 同步到渲染副本
        viewer.sync()                   # 刷新画面
        step_count += 1

⚠️ 为什么不直接用 physics.data.ptr 给 viewer？
理论上 physics.data.ptr 是原生 MjData，可以直接传给 viewer（见 16.8 坑 1 的简单示例）。
但在复杂项目中不推荐，原因：

env.step() 内部会多次调用 physics.step()，中间状态可能不一致。
viewer 在另一个线程运行，直接共享 physics.data 可能有竞态条件。
维护独立的 render_data 可以在仿真和渲染之间做解耦（比如降帧率渲染）。

简单场景（单线程、不需要降帧率）可以直接用 physics.model.ptr, physics.data.ptr，
见 16.8 坑 1 的示例。

16.11.5 平滑控制插值：梯形速度剖面 + smoothstep

取放任务中，机械臂的运动平滑度直接影响观感和物理稳定性。
dm_control_smooth_simulation.py 用了两种插值技术：

① 梯形速度剖面（Trapezoidal Velocity Profile）

末端执行器在笛卡尔空间沿直线移动，速度曲线呈梯形：

加速段：从 0 匀加速到巡航速度 v
巡航段：以恒定速度 v 移动
减速段：从 v 匀减速到 0

速度 v
│     ┌───────┐
│    /         \
│   /           \
│  /             \
└────────────────── 时间 t
  加速  巡航   减速

如果距离太短（2 * d_acc > d），退化为三角形速度剖面（无巡航段）。

def _trap_fraction(shape, tau):
    """返回 tau 时刻走过的弧长占段长 d 的比例 f∈[0,1]。
    shape 为 ('trap', t_acc, t_cruise, v, d_acc, d) 或 ('tri', T, d)。
    速度连续（加减速段与巡航段衔接处速度相等），加速度分段恒定，
    是工业机械臂最常用、观感最「流畅」的规划方式。
    """
    if shape[0] == "trap":
        _, t_acc, t_cruise, v, d_acc, d = shape
        a = v / t_acc
        if tau <= t_acc:                    # 加速段: s = ½aτ²
            s = 0.5 * a * tau * tau
        elif tau <= t_acc + t_cruise:       # 巡航段: s = d_acc + v(τ-t_acc)
            s = d_acc + v * (tau - t_acc)
        else:                                # 减速段
            td = tau - t_acc - t_cruise
            s = d_acc + v * t_cruise + (v * td - 0.5 * a * td * td)
        return float(np.clip(s / d, 0.0, 1.0))
    else:  # 三角形剖面（距离太短时退化）
        _, T, d = shape
        t_peak = T / 2.0
        a = d / (t_peak * t_peak)
        if tau <= t_peak:
            s = 0.5 * a * tau * tau
        else:
            td = tau - t_peak
            s = (0.5 * a * t_peak * t_peak) + (a * t_peak * td - 0.5 * a * td * td)
        return float(np.clip(s / d, 0.0, 1.0))

② smoothstep 夹爪过渡

夹爪开合不用瞬间置位，而是用 smoothstep 平滑过渡：

u = np.clip(tau / dwell_time, 0.0, 1.0)
ss = 3.0 * u * u - 2.0 * u * u * u   # smoothstep: S 形曲线
g = grip_from + (grip_to - grip_from) * ss

夹爪开度
│         ______
│        /
│       /
│      /
│ ____/
└────────────── 时间
  平滑 S 形过渡

💡 为什么不用五次多项式？ 传统的"固定步数五次多项式"在每段起停时速度为 0，
多段拼接时会有明显的"一顿一顿"感。梯形速度剖面在巡航段保持恒定速度，
拐角由位置控制器自然圆滑，整体观感更流畅。这也是工业机械臂最常用的规划方式。

与 dm_control 的集成方式：

class CartesianTrajectory:
    """笛卡尔空间平滑轨迹：匀速巡航 + 取放驻留。
    每个 move 段用梯形速度剖面沿直线插值；
    每个 dwell 段保持位置，夹爪用 smoothstep 在 grip_from->grip_to 间平滑过渡。
    """

    def __init__(self, start, waypoints, ee_speed, ee_accel, grip_time):
        self.ee_speed = ee_speed      # 末端巡航速度 (m/s)
        self.ee_accel = ee_accel      # 末端最大加速度 (m/s²)
        self.grip_time = grip_time    # 取/放料驻留时间 (s)
        self.phases = []               # 阶段列表
        self.cum = [0.0]               # 累计时间
        # ... 构建 phases（每个 waypoint 生成 move 段 + 可选 dwell 段）...

    def sample(self, dt):
        """推进 dt 秒，返回当前末端位置 (3,) 与夹爪指令 (标量)。"""
        # 定位当前 phase，根据 phase 类型计算位置和夹爪
        ...

# 在 Policy 中使用
target_pos, gripper_cmd = self.traj.sample(CONTROL_DT)
q, ok, err = self.ik.solve(target_pos, initial_qpos=live_q)

16.11.6 无头验证与录屏

无头模式（Headless）：不需要显示器，用于 CI 验证和批量测试。

def run_headless(env, policy, physics, max_steps=2000):
    env.reset()
    ee_positions, gripper_cmds = [], []
    step = 0
    while step < max_steps and not policy.traj.done:
        action = policy(physics)
        env.step(action)
        ee_positions.append(physics.named.data.site_xpos["end_effector"].copy())
        gripper_cmds.append(float(action[6]))
        step += 1
    return np.array(ee_positions), np.array(gripper_cmds)

💡 无头模式的典型用途：

回归测试：修改代码后跑一次，验证末端轨迹是否符合预期
参数扫描：批量测试不同 ee_speed / ee_accel 下的运动时间
CI/CD：在没有显示器的服务器上验证仿真是否能正常运行

录屏模式：用 mujoco.Renderer 离屏渲染 + imageio 写 MP4。

import imageio

# ① 准备 Renderer 和 writer
renderer = mujoco.Renderer(raw_model, height=720, width=1280)
rec_cam = mujoco.MjvCamera()
rec_writer = imageio.get_writer(output_path, fps=30, codec="libx264",
                                 macro_block_size=None,
                                 output_params=["-crf", "20"])

def capture():
    # 跟随用户相机（从 viewer 复制相机参数）
    rec_cam.type = viewer.cam.type
    rec_cam.azimuth = viewer.cam.azimuth
    rec_cam.elevation = viewer.cam.elevation
    rec_cam.distance = viewer.cam.distance
    rec_cam.lookat[:] = viewer.cam.lookat[:]
    # 离屏渲染一帧
    renderer.update_scene(render_data, camera=rec_cam)
    frame = renderer.render()
    rec_writer.append_data(frame)

# ② 主循环中每帧调用 capture()
# ③ 结束后关闭 writer
rec_writer.close()

⚠️ 注意：mujoco.Renderer 接受的是原生 MjModel，不是 dm_control 的包装类。
这也是双模型模式的另一个用例——录屏需要原生模型来创建 Renderer。

-crf 20 是 x264 的恒定质量因子（0=无损，51=最差），20 是视觉无损的常用值。
macro_block_size=None 避免 imageio 对分辨率做 16 对齐裁剪。

16.11.7 完整环境构建：build_smooth 模式

把上面所有模式组合起来，就是一个完整的 dm_control 仿真环境构建函数：

def build_smooth(max_steps=2000, ee_speed=0.18, ee_accel=0.9, grip_time=0.6):
    # ① dm_control 端
    root = mjcf.from_path(str(SIM_MJCF))
    physics = mjcf.Physics.from_mjcf_model(root)

    # ② 原生端（双模型模式）
    raw_model = mujoco.MjModel.from_xml_path(str(SIM_MJCF))
    ee_id = mujoco.mj_name2id(raw_model, mujoco.mjtObj.mjOBJ_SITE, "end_effector")
    obj_id = mujoco.mj_name2id(raw_model, mujoco.mjtObj.mjOBJ_SITE, "object_center")
    weld_id = mujoco.mj_name2id(raw_model, mujoco.mjtObj.mjOBJ_EQUALITY, "grasp_weld")

    # ③ 计算初始姿态对应的末端位置（用原生模型算，不污染 dm_control physics）
    tmp = mujoco.MjData(raw_model)
    tmp.qpos[:6] = INITIAL_POSE
    mujoco.mj_forward(raw_model, tmp)
    home_ee = tmp.site_xpos[ee_id].copy()

    # ④ 构建轨迹（接近取料点 → 取料(驻留+夹爪闭合) → 上升 → 接近放料点 → ...）
    waypoints = [
        above_pick,
        {"pos": PICK_POS, "dwell": grip_time,
         "grip_from": GRIPPER_OPEN, "grip_to": GRIPPER_CLOSED},
        above_pick,
        above_place,
        {"pos": PLACE_POS, "dwell": grip_time,
         "grip_from": GRIPPER_CLOSED, "grip_to": GRIPPER_OPEN},
        above_place,
    ]
    traj = CartesianTrajectory(start=home_ee, waypoints=waypoints,
                               ee_speed=ee_speed, ee_accel=ee_accel,
                               grip_time=grip_time)

    # ⑤ IK 求解器（用原生模型 + 独立 scratch data，不污染实时仿真）
    ik_data = mujoco.MjData(raw_model)
    ik = IKSolver(raw_model, ik_data)

    # ⑥ Policy + Task + Environment
    policy = SmoothPolicy(ik, traj, weld_id)
    task = PickPlaceTask(policy, ee_id, obj_id)
    env = control.Environment(physics=physics, task=task,
                              control_timestep=CONTROL_DT)

    return env, policy, physics, raw_model

💡 构建顺序的逻辑：

先建 dm_control Physics（仿真主体）
再建原生模型（辅助工具：IK / name2id / viewer / Renderer）
用原生模型计算初始末端位置（不污染 dm_control physics）
构建轨迹（依赖初始末端位置）
构建 IK 求解器（用原生模型 + 独立 scratch data）
最后组装 Policy → Task → Environment

IK 用独立 scratch data 的原因：IK 求解过程中会修改 MjData（设置关节角、调用 mj_forward），
如果用 physics.data 会污染实时仿真状态。所以 IK 专用一份 ik_data = mujoco.MjData(raw_model)。

16.12 小结

本章知识地图

强化学习接口
├── dm_env 标准（4个概念）
│   ├── Environment：reset() / step()
│   ├── TimeStep：(step_type, reward, discount, observation)
│   │   └── ⚠️ reset() 时 reward=None, discount=None
│   ├── StepType：FIRST / MID / LAST
│   └── specs：Array / BoundedArray / DiscreteArray
│
├── control.Task（9个方法，5个必须）
│   ├── 必须：initialize_episode / before_step / action_spec
│   │       get_observation / get_reward
│   ├── 可选：observation_spec（自动推断）/ after_step
│   │       get_termination / step_spec
│   └── ⚠️ before_step 参数顺序：(action, physics)
│
├── control.Environment
│   ├── 参数：physics, task, time_limit, control_timestep
│   ├── ⚠️ control_timestep 必须是 timestep 整数倍
│   ├── ⚠️ control_timestep 与 n_sub_steps 互斥
│   └── flat_observation：字典 vs 扁平向量
│
├── 空间设计
│   ├── action_spec：BoundedArray(shape, dtype, min, max, name)
│   ├── observation_spec：OrderedDict（可选，自动推断）
│   └── ⚠️ 动作越界不校验，要自己 clip
│
├── 踩坑实录
│   ├── 包装类 → .ptr 才能给原生 API
│   ├── mujoco 3.x 枚举 → mjtObj.mjOBJ_SITE
│   ├── mj_jacSite → (3, nv) 二维数组
│   └── 别把"在动"当成"在收敛"
│
└── 进阶实战（完整取放仿真）
    ├── Policy 类与 Task 分离（维护轨迹状态）
    ├── weld 约束抓取（eq_active 激活/释放）
    ├── 双模型模式（dm_control Physics + 原生 MjModel）
    ├── viewer 渲染同步（render_data + sync_render）
    ├── 平滑控制插值（梯形速度剖面 + smoothstep）
    ├── 无头验证 + 录屏（imageio + mujoco.Renderer）
    └── build_smooth 完整环境构建模式

在 6 轴机械臂项目中的应用

知识点	项目中的具体用途
`ReachTask`	末端到达任务的基础框架，后续抓取任务在此基础上扩展
`initialize_episode`	每个 episode 开始时重置机械臂到 HOME 位姿、随机化物体位置
`before_step`	将 RL 算法输出的关节目标角写入 `ctrl` 数组
`get_observation`	组装观测：6个关节角 + 末端位置 + 目标位置（共12维）
`get_reward`	奖励 = -末端到目标距离（密集奖励，提供每步梯度）
`get_termination`	距离 < 5cm 时成功终止，4秒超时自动截断
`control_timestep=0.02`	50Hz 控制频率，与真实工业机械臂一致
IK 策略（DLS）	验证环境正确性的黄金标准，也作为 RL 训练的初始策略参考
Policy 类模式	取放任务中维护轨迹进度、携带状态、夹爪位置，与 Task 分离
weld 约束抓取	夹爪闭合时激活 equality 约束，把物体焊在末端，稳定可控
双模型模式	dm_control Physics 跑仿真 + 原生 MjModel 做 IK/name2id/viewer/Renderer
viewer 同步	维护 render_data 副本，每帧同步 dm_control 状态后用 viewer 渲染
梯形速度剖面	笛卡尔空间匀速巡航 + 加减速段，工业机械臂最流畅的规划方式
无头验证	不需要显示器，用于 CI 回归测试和参数扫描

扩展阅读方向

dm_env 官方文档：https://github.com/google-deepmind/dm_env — 标准接口的完整定义
奖励函数设计：Reward Shaping（Ng et al. 1999）— 如何设计奖励函数加速收敛而不改变最优策略
稀疏奖励 vs 密集奖励：稀疏奖励（只有成功时给1）更难训练但更接近真实场景；密集奖励（每步给距离）更容易训练但可能导致"刷分"行为
下一章预告：第 17 章将引入 composer——把机器人与场景拆成可复用的模块（Entity / Arena / Task），让场景组装变成代码拼乐高，为后续复杂抓取任务打下基础。
TimeStep：跑一次 reset() 和几次 step()，打印每步的
step_type / reward / discount，确认 FIRST 时 reward 是 None。
Task 协议：写一个最小 Task（只实现 5 个抽象方法），
确认能构造出 Environment；再删掉 initialize_episode，看报什么错；
然后试着不写 observation_spec，确认它其实可选。
时间关系：用 control_timestep=0.01 / 0.02 / 0.05 各跑一步，
打印 physics.data.time 的变化，验证「物理步数 = 控制周期 / 仿真步长」。
空间设计：把 action_spec 改成每维不同范围，
再用 np.full(6, 99.0) 试探，确认越界不会被拦截。
跑 episode：实现随机策略 / 固定关节角 / 雅可比 IK 三种策略，
对比累计奖励与最终距离（参考答案里 IK 34 步收敛到 0.047 m）。
踩坑复现：故意把 physics.model（不是 .ptr）传给
launch_passive，确认报错信息与你预期一致。

参考答案见 code/ch16_dm_control_rl.py。

16.13 小结

dm_env 规定 4 件事：Environment / TimeStep / StepType / specs。
TimeStep = (step_type, reward, discount, observation)；
⚠️ reset() 时 reward=None、discount=None。
🔥 Task 的 5 个抽象方法：initialize_episode / before_step /
action_spec / get_observation / get_reward。
（observation_spec 不是抽象方法，不写会自动推断。）
⚠️ before_step(action, physics) —— 参数顺序是 action 在前。
🔥 control_timestep 必须是仿真步长整数倍，且与 n_sub_steps 互斥。
BoundedArray(shape, dtype, minimum, maximum, name)；
数值越界 dm_env 不拦，要自己 clip。
🔥 physics.model 是包装类，传给原生 API 要用 .ptr
（launch_passive 会明确报 not a mujoco.MjModel）。
🔥 对象枚举在 mujoco.mjtObj 下，mujoco.mjOBJ_SITE 已不存在。
get_termination 返回 None/0.0/1.0 分别代表继续 / 成功 / 超时。
💡 别把「在动」当成「在收敛」 —— 用数字验证策略。

下一章：composer —— 把机器人与场景拆成可复用的模块（Entity / Arena / Task）。

上一章：15 · dm_control 入门｜下一章：17 · composer 进阶

DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

基于蚁群优化算法的直流电机模糊PID控制（Matlab实现）

直流电机因其调速性能优异、响应速度快等特点，在工业控制、电动汽车、机器人等领域有着广泛应用。然而，传统PID控制存在参数固定、难以应对非线性及环境扰动等问题。模糊PID控制虽通过引入模糊逻辑实现了参数自适应调整，但在复杂场景下仍面临规则库构建与参数优化的难题。本文将蚁群优化算法（Ant Colony Optimization, ACO）引入直流电机模糊PID控制系统中，利用其全局寻优能力对模糊PI

DAMO开发者矩阵

基于蚁群优化算法的直流电机模糊PID控制（Matlab实现）

DAMO开发者矩阵

从生产课长到PLC工程师：一个 9 年逆袭者的近30万年薪真相

DAMO开发者矩阵

所有评论(0)

查看更多评论

ai2work

@m0_64703412

已为社区贡献10条内容

16 · 强化学习接口 ★

ai2work

🎯 学习目标

16.1 为什么需要标准环境接口

直观理解：RL 交互循环是什么？

Agent-Environment 交互循环图

16.2 dm_env 的四个概念

直观理解：四个概念分别是什么？

Episode 时序图

TimeStep 四个字段的含义

16.3 Task 协议：九个方法

直观理解：Task 是什么？

Task 方法调用时序图

调用顺序

⚠️ before_step 的参数顺序是 (action, physics)

空 Task 会明确报缺哪些方法

16.4 完整例子：末端到达 Task

逐行讲解：ReachTask 完整实现

16.5 control.Environment 的构造参数 + 两条硬约束

⚠️ 约束①：n_sub_steps 与 control_timestep 互斥

⚠️ 约束②：control_timestep 必须是仿真步长的整数倍

魔法数字来源解释

两种指定方式（实测）

16.6 观测 / 动作空间设计

观测 / 动作 / 奖励 数据流图

specs 常用类型

每个维度不同范围（真实机器人几乎都是这样）

⚠️ action 越界时 dm_env 不强制校验

flat_observation=True：拼成一个向量

16.7 跑完整 episode

reset() 返回什么

三种策略对比（实测）

IK 策略逐行讲解（复习第 6 章 DLS）

get_termination 返回值对照

16.8 🔥 本书踩坑实录

坑 1：包装类与 launch_passive 不兼容

坑 2：mujoco 3.x 的对象枚举搬到了 mjtObj 下

坑 3：before_step 的参数顺序

坑 4 / 5：control_timestep 要整数倍，且与 n_sub_steps 互斥

坑 6：空 Task 报抽象方法未实现（5 个，不是 4 个）

坑 7：step_spec() 基类默认抛 NotImplementedError

坑 8：mujoco.mj_jacSite 要 (3, nv) 二维数组

16.9 常见错误速查

16.10 动手练

练习详解与答案提示

16.11 进阶实战：完整取放仿真的工程模式

16.11.1 Policy 与 Task 分离：为什么需要 Policy 类

16.11.2 用 weld 约束实现抓取

16.11.3 双模型模式：dm_control Physics + 原生 MjModel

16.11.4 viewer 渲染同步模式

16.11.5 平滑控制插值：梯形速度剖面 + smoothstep

16.11.6 无头验证与录屏

16.11.7 完整环境构建：build_smooth 模式

16.12 小结

本章知识地图

在 6 轴机械臂项目中的应用

扩展阅读方向

16.13 小结

所有评论(0)

温馨提示：您尚未绑定手机号

ai2work

`TimeStep` 四个字段的含义

16.3 `Task` 协议：九个方法

⚠️ `before_step` 的参数顺序是 `(action, physics)`

16.5 `control.Environment` 的构造参数 + 两条硬约束

⚠️ 约束①：`n_sub_steps` 与 `control_timestep` 互斥

⚠️ 约束②：`control_timestep` 必须是仿真步长的整数倍

观测 / 动作 / 奖励数据流图

`specs` 常用类型

⚠️ `action` 越界时 dm_env 不强制校验

`flat_observation=True`：拼成一个向量

`reset()` 返回什么

`get_termination` 返回值对照

坑 1：包装类与 `launch_passive` 不兼容

坑 2：mujoco 3.x 的对象枚举搬到了 `mjtObj` 下

坑 3：`before_step` 的参数顺序

坑 4 / 5：`control_timestep` 要整数倍，且与 `n_sub_steps` 互斥

坑 7：`step_spec()` 基类默认抛 `NotImplementedError`

坑 8：`mujoco.mj_jacSite` 要 `(3, nv)` 二维数组