如何将您自己的环境迁移到DI-engine

DI-zoo 为用户提供了大量常用的强化学习环境（支持的环境），但在许多研究和工程场景中，用户仍然需要自己实现一个环境，并希望快速将其迁移到 DI-engine 以满足 DI-engine 的相关规范。因此，在本节中，我们将逐步介绍如何执行上述迁移，以满足 DI-engine 的环境基类 BaseEnv 的规范，从而使其能够轻松应用于训练管道中。

以下介绍将从基础和高级开始。基础描述了必须实现的功能以及您应注意的细节；高级描述了一些扩展功能。

然后会介绍DingEnvWrapper，它是一个“工具”，可以快速将ClassicControl、Box2d、Atari、Mujoco、GymHybrid等简单环境转换为符合BaseEnv的环境。最后还有一个问答环节。

基础

本节描述了用户必须满足的规范约束，以及在迁移环境时必须实现的功能。

如果你想在DI-engine中使用环境，你需要实现一个继承自BaseEnv的子类环境，例如YourEnv。YourEnv与你自己的环境之间的关系是一种组合关系，也就是说，在一个YourEnv实例中，会有一个用户原生环境的实例（例如，一个gym类型的环境）。

强化学习环境有一些大多数环境都实现的常见主要接口，例如 reset(), step(), seed() 等。在 DI-engine 中，BaseEnv 将进一步封装这些接口。在大多数情况下，将以 Atari 为例进行说明。具体代码请参考 Atari Env 和 Atari Env Wrapper

1. __init__()

   通常，环境可以在__init__方法中实例化，但在DI-engine中，为了方便支持像EnvManager这样的“环境向量化”模块，环境实例通常使用延迟初始化机制，即__init__方法不会初始化真正的原始环境实例，而只是设置相关的参数配置。当第一次调用reset方法时，实际的环境初始化才会发生。

   以Atari为例。__init__ 不会实例化环境，它只是设置参数配置值 self._cfg，并将变量 self._init_flag 初始化为 False（表示环境尚未实例化）。

   class AtariEnv(BaseEnv):
   
      def __init__(self, cfg: dict) -> None:
         self._cfg = cfg
         self._init_flag = False
   

2. seed()

   seed 用于设置环境中的随机种子。环境中有两种类型的随机种子需要设置，一种是原始环境的随机种子，另一种是各种环境转换中的库种子（例如 random , np.random 等）。

   对于第二种类型，随机库种子的设置相对简单，直接在环境的seed方法中设置。

   对于第一种类型，原始环境的种子仅在seed方法中分配，但并未真正设置；真正的设置在调用环境的reset方法内部，具体的原始环境在设置之前reset。

   class AtariEnv(BaseEnv):
   
      def seed(self, seed: int, dynamic_seed: bool = True) -> None:
         self._seed = seed
         self._dynamic_seed = dynamic_seed
         np.random.seed(self._seed)
   

   对于原始环境的种子，DI-engine 有静态种子和动态种子的概念。

   静态种子用于测试环境（evaluator_env）中，以确保所有回合的随机种子相同，即在reset时仅使用固定的静态种子值self._seed。需要在seed方法中手动将dynamic_seed参数传递为False。

   动态种子用于训练环境（collector_env），尝试使每一集的随机种子不同，即在reset时，将使用随机数生成器100 * np.random.randint(1, 1000)（但此随机数生成器的种子由环境的seed方法固定，因此保证了实验的可重复性）。你需要在seed中手动传入dynamic_seed参数为True（或者可以不传入，因为默认参数是True）。

3. reset()

   DI-engine 的 Lazy Init 初始化方法已在 __init__ 方法中引入，即在 第一次调用 reset 方法时执行实际的环境初始化。

   reset 方法会根据 self._init_flag 判断实际环境是否需要实例化（如果为 False，则会实例化；否则，已经实例化并可以直接使用），并设置随机种子，然后调用原始环境的 reset 方法以获取初始状态下的观测值 obs，并将其转换为 np.ndarray 数据格式（将在第4部分详细解释），并初始化 self._eval_episode_return 的值（将在第5部分详细解释），在 Atari 中，self._eval_episode_return 指的是整个 episode 中获得的真实奖励的累积和，用于评估代理在此环境中的表现，不用于训练。

   class AtariEnv(BaseEnv):
   
      def __init__(self, cfg: dict) -> None:
         self._cfg = cfg
         self._init_flag = False
   
      def reset(self) -> np.ndarray:
         if not self._init_flag:
            self._env = self._make_env(only_info=False)
            self._init_flag = True
         if hasattr(self, '_seed') and hasattr(self, '_dynamic_seed') and self._dynamic_seed:
            np_seed = 100 * np.random.randint(1, 1000)
            self._env.seed(self._seed + np_seed)
         elif hasattr(self, '_seed'):
            self._env.seed(self._seed)
         obs = self._env.reset()
         obs = to_ndarray(obs)
         self._eval_episode_return = 0.
         return obs
   

4. step()

   step 方法负责接收当前时间步的 action，然后给出当前时间步的 reward 和下一个时间步的 obs。在 DI-engine 中，你还需要给出：当前回合是否结束的标志 done（这里要求 done 必须是 bool 类型，而不是 np.bool），以及其他信息以字典 info 的形式（其中至少包括键 self._eval_episode_return）。

   在获取reward、obs、done、info等数据后，需要对其进行处理并转换为np.ndarray格式以符合DI-engine规范。self._eval_episode_return将在每个时间步累积当前步骤获得的实际奖励，并在一个回合结束时（done == True）返回累积值。

   最后，将上述四个数据放入定义为namedtuple的BaseEnvTimestep中并返回（定义为：BaseEnvTimestep = namedtuple('BaseEnvTimestep', ['obs', 'reward', 'done ', 'info'])）

   from ding.envs import BaseEnvTimestep
   
   class AtariEnv(BaseEnv):
   
      def step(self, action: np.ndarray) -> BaseEnvTimestep:
         assert isinstance(action, np.ndarray), type(action)
         action = action.item()
         obs, rew, done, info = self._env.step(action)
         self._eval_episode_return += rew
         obs = to_ndarray(obs)
         rew = to_ndarray([rew])  # Transformed to an array with shape (1, )
         if done:
            info['eval_episode_return'] = self._eval_episode_return
         return BaseEnvTimestep(obs, rew, done, info)
   

5. self._eval_episode_return

   在Atari环境中，self._eval_episode_return 指的是一个episode中所有奖励的累积和，self._eval_episode_return 的数据类型必须是python原生类型，而不是 np.array。

   • 在reset方法中，将当前的self._eval_episode_return设置为0；

   • 在step方法中，将每个时间步获得的实际奖励添加到self._eval_episode_return中。

   • 在step方法中，如果当前回合已经结束（done == True），则添加到info字典并返回：info['eval_episode_return'] = self._eval_episode_return

   然而，其他环境可能不需要self._eval_episode_return的总和。例如，在smac中，需要当前回合的胜率，因此有必要修改第二步step方法中的简单累加。相反，我们应该记录游戏情况，并最终在回合结束时返回计算出的胜率。

6. 数据规范

   DI-engine 要求环境中每个方法的输入和输出数据必须为 np.ndarray 格式，并且数据类型必须为 np.int64（整数）、np.float32（浮点数）或 np.uint8（图像）。包括：

   • obs 由 reset 方法返回

   • action 由 step 方法接收

   • obs 由 step 方法返回

   • reward 由 step 方法返回，这里还要求 reward 必须是一维的，而不是零维的，例如，Atari 会将零维扩展为一维 rew = to_ndarray([rew])

   • done 由 step 方法返回的类型必须是 bool，而不是 np.bool

高级

1. 环境预处理包装器

   如果在强化学习训练中要使用许多环境，需要进行一些预处理以达到增加随机性、数据归一化和便于训练的目的。这些预处理以包装器的形式实现（有关包装器的介绍，请参阅这里）。

   每个用于环境预处理的包装器都是gym.Wrapper的子类。例如，NoopResetEnv用于在每集开始时执行随机数量的无操作动作。这是一种增加随机性的手段。它的使用方式如下：

   env = gym.make('Pong-v4')
   env = NoopResetEnv(env)
   

   由于reset方法在NoopResetEnv中实现，当调用env.reset()时，NoopResetEnv中的相应逻辑将被执行。

   以下环境包装器已在DI-engine中实现：（位于ding/envs/env_wrappers/env_wrappers.py）

   • NoopResetEnv: 在每一集开始时执行随机数量的无操作动作

   • MaxAndSkipEnv: 返回几帧中的最大值，可以视为一种时间步上的最大池化

   • WarpFrame: 使用cv2库的cvtColor将原始图像转换为颜色代码，并将其调整为特定长度和宽度的图像（通常为84x84）

   • ScaledFloatFrame: 将观察值归一化到区间 [0, 1]（保持数据类型为 np.float32）

   • ClipRewardEnv: 通过一个符号函数将奖励传递到 {+1, 0, -1}

   • FrameStack: 将一定数量（通常为4）的帧堆叠在一起作为新的观察结果，可用于处理POMDP情况，例如，通过单帧信息无法知道运动的速度方向

   • ObsTransposeWrapper: 将观察值转置以将通道放在第一个维度

   • ObsNormEnv: 使用 RunningMeanStd 来对滑动窗口的观察值进行归一化

   • RewardNormEnv: 使用 RunningMeanStd 通过滑动窗口来标准化奖励

   • RamWrapper: 将ram环境包装成类似图像的环境

   • EpisodicLifeEnv: 处理内置多条生命的游戏环境（例如Qbert），并将每条生命视为一个独立的回合。

   • FireResetEnv: 在环境重置后立即执行动作1（开火）

   • GymHybridDictActionWrapper: 将Gym-Hybrid的原始gym.spaces.Tuple动作空间转换为gym.spaces.Dict

   如果上述包装器不能满足您的需求，您也可以自定义包装器。

   值得一提的是，每个包装器不仅必须完成相应观察/动作/奖励值的更改，还必须相应地修改其空间（当且仅当形状、数据类型等被修改时），该方法将在下一节中详细描述。

2. 三个空间属性 observation/action/reward space

   如果你想根据环境的维度自动创建一个神经网络，或者在EnvManager中使用shared_memory技术来加速环境返回的大张量数据的传输，你需要让环境支持提供属性observation_space action_space reward_space。

   注意

   为了代码的可扩展性，我们强烈建议实现这三个空间属性。

   这里的空间都是gym.spaces.Space子类的实例，最常用的gym.spaces.Space包括Discrete Box Tuple Dict等。shape和dtype需要在空间中给出。在原始的gym环境中，大多数都会支持observation_space、action_space和reward_range。在DI-engine中，reward_range也被扩展为reward_space，以便这三者保持一致。

   例如，以下是cartpole的三个属性：

   class CartpoleEnv(BaseEnv):
   
      def __init__(self, cfg: dict = {}) -> None:
         self._observation_space = gym.spaces.Box(
               low=np.array([-4.8, float("-inf"), -0.42, float("-inf")]),
               high=np.array([4.8, float("inf"), 0.42, float("inf")]),
               shape=(4, ),
               dtype=np.float32
         )
         self._action_space = gym.spaces.Discrete(2)
         self._reward_space = gym.spaces.Box(low=0.0, high=1.0, shape=(1, ), dtype=np.float32)
   
      @property
      def observation_space(self) -> gym.spaces.Space:
         return self._observation_space
   
      @property
      def action_space(self) -> gym.spaces.Space:
         return self._action_space
   
      @property
      def reward_space(self) -> gym.spaces.Space:
         return self._reward_space
   

   由于cartpole不使用任何包装器，其三个空间是固定的。然而，如果像Atari这样的环境被多个包装器装饰，每次包装器包装原始环境后，都需要修改相应的空间。例如，Atari将使用ScaledFloatFrameWrapper将观察值归一化到区间[0, 1]，然后相应地修改其observation_space：

   class ScaledFloatFrameWrapper(gym.ObservationWrapper):
   
      def __init__(self, env):
         # ...
         self.observation_space = gym.spaces.Box(low=0., high=1., shape=env.observation_space.shape, dtype=np.float32)
   

3. enable_save_replay()

   DI-engine 不需要实现 render 方法。如果你想完成可视化，我们建议实现 enable_save_replay 方法来保存游戏视频。

   此方法在reset方法之前和seed方法之后调用，其中指定了记录存储的路径。需要注意的是，此方法不直接存储视频，而只是设置一个标志，用于确定是否保存视频。实际存储视频的代码和逻辑需要由您自己实现。（因为可能会打开多个环境，并且每个环境运行多个片段，所以需要在文件名中进行区分）

   这里给出了一个DI-engine中的示例。reset 方法使用了 gym 提供的装饰器来封装环境，使其具有存储游戏视频的功能，如代码所示：

   class AtariEnv(BaseEnv):
   
      def enable_save_replay(self, replay_path: Optional[str] = None) -> None:
         if replay_path is None:
            replay_path = './video'
         self._replay_path = replay_path
   
      def reset():
         # ...
         if self._replay_path is not None:
            self._env = gym.wrappers.RecordVideo(
               self._env,
               video_folder=self._replay_path,
               episode_trigger=lambda episode_id: True,
               name_prefix='rl-video-{}'.format(id(self))
            )
         # ...
   

   在实际使用中，调用这些方法的顺序应该是：

   atari_env = AtariEnv(easydict_cfg)
   atari_env.seed(413)
   atari_env.enable_save_replay('./replay_video')
   obs = atari_env.reset()
   # ...
   

4. 为训练环境和测试环境使用不同的配置

   用于训练的环境（collector_env）和用于测试的环境（evaluator_env）可能使用不同的配置项。您可以在环境中实现一个静态方法，以实现对不同环境配置项的自定义配置。以Atari为例：

   class AtariEnv(BaseEnv):
   
      @staticmethod
      def create_collector_env_cfg(cfg: dict) -> List[dict]:
         collector_env_num = cfg.pop('collector_env_num')
         cfg = copy.deepcopy(cfg)
         cfg.is_train = True
         return [cfg for _ in range(collector_env_num)]
   
      @staticmethod
      def create_evaluator_env_cfg(cfg: dict) -> List[dict]:
         evaluator_env_num = cfg.pop('evaluator_env_num')
         cfg = copy.deepcopy(cfg)
         cfg.is_train = False
         return [cfg for _ in range(evaluator_env_num)]
   

   在实际使用中，原始配置项 cfg 可以转换为获取训练和测试的两个版本的配置项：

   # env_fn is an env class
   collector_env_cfg = env_fn.create_collector_env_cfg(cfg)
   evaluator_env_cfg = env_fn.create_evaluator_env_cfg(cfg)
   

   设置cfg.is_train项将相应地使用不同的装饰器。例如，如果cfg.is_train == True，将使用奖励的符号函数映射到{+1, 0, -1}以促进训练，如果cfg.is_train == False，则原始奖励值将保持不变，这便于在测试期间评估代理的性能。

5. random_action()

   一些离策略算法希望在训练开始前使用随机策略收集一些数据来填充缓冲区，并完成缓冲区的初始化。对于这样的需求，DI-engine 鼓励实现 random_action 方法。

   由于环境已经实现了action_space，你可以直接调用gym中提供的Space.sample()方法来随机选择动作。但需要注意的是，由于DI-engine要求所有返回的动作都必须是np.ndarray格式，因此可能需要进行一些必要的dtype转换。int和dict类型通过to_ndarray函数转换为np.ndarray类型，如下代码所示：

   def random_action(self) -> np.ndarray:
      random_action = self.action_space.sample()
      if isinstance(random_action, np.ndarray):
            pass
      elif isinstance(random_action, int):
            random_action = to_ndarray([random_action], dtype=np.int64)
      elif isinstance(random_action, dict):
            random_action = to_ndarray(random_action)
      else:
            raise TypeError(
               '`random_action` should be either int/np.ndarray or dict of int/np.ndarray, but get {}: {}'.format(
                  type(random_action), random_action
               )
            )
      return random_action
   

6. default_config()

   如果一个环境有一些默认或常用的配置项，你可以考虑将类变量 config 设置为 默认配置（为了方便外部访问，你也可以实现类方法 default_config，它返回 config）。如下代码所示：

   在运行实验时，会为此实验配置一个用户配置文件，例如dizoo/mujoco/config/ant_ddpg_config.py。在用户配置文件中，您可以省略这部分键值对，并通过deep_merge_dicts将默认配置与用户配置合并（请记住在此处使用默认配置作为第一个参数，用户配置作为第二个参数以确保用户配置具有更高的优先级）。如下代码所示：

   class MujocoEnv(BaseEnv):
   
      @classmethod
      def default_config(cls: type) -> EasyDict:
         cfg = EasyDict(copy.deepcopy(cls.config))
         cfg.cfg_type = cls.__name__ + 'Dict'
         return cfg
   
      config = dict(
         use_act_scale=False,
         delay_reward_step=0,
      )
   
      def __init__(self, cfg) -> None:
         self._cfg = deep_merge_dicts(self.config, cfg)
   

7. 环境实现正确性检查

   我们提供了一套检查工具，用于用户实现的环境以检查：

   • 观察/动作/奖励的数据类型

   • 重置/步骤方法

   • 在两个相邻时间步的观察中是否存在不合理的相同引用（即应使用deepcopy以避免相同引用）

   检查工具的实现位于ding/envs/env/env_implementation_check.py。 关于检查工具的使用方法，请参考ding/envs/env/tests/test_env_implementation_check.py中的test_an_implemented_env。

DingEnvWrapper

DingEnvWrapper 可以快速将诸如 ClassicControl、Box2d、Atari、Mujoco、GymHybrid 等简单环境转换为符合 BaseEnv 的环境。

注意：DingEnvWrapper的具体实现可以在ding/envs/env/ding_env_wrapper.py中找到，此外，你可以查看示例以获取更多信息。

问答

1. MARL 环境应该如何迁移？

   你可以参考Competitive RL

   • 如果环境支持单代理、双代理甚至多代理，请考虑不同的模式分类

   • 在多代理环境中，动作和观察的数量与代理的数量相匹配，但奖励和完成情况不一定相同。有必要明确奖励的定义

   • 注意原始环境如何要求将动作和观察结果结合起来（元组、列表、字典、堆叠数组等）

2. 混合动作空间的环境应该如何迁移？

   你可以参考 Gym-Hybrid

   • 在Gym-Hybrid中，一些离散动作（加速、转向）需要给出相应的一维连续参数来表示加速度和旋转角度，因此类似的环境需要关注其动作空间的定义