如何在 LightZero 中自定义您的环境？

在使用 LightZero 进行强化学习研究或应用时，您可能需要创建一个自定义环境。创建自定义环境可以更好地适应特定问题或任务，使强化学习算法能够在这些特定环境中有效训练。

对于LightZero中的典型环境，请参考atari_lightzero_env.py。LightZero的环境设计很大程度上基于DI-engine中的BaseEnv类。在创建自定义环境时，我们遵循与DI-engine中类似的基本步骤。

与BaseEnv的主要区别

在 LightZero 中，有许多棋盘游戏环境。由于玩家交替行动和合法移动集合的变化，棋盘游戏环境中的观察状态不仅应包括棋盘信息，还应包括动作掩码和当前玩家信息。因此，在 LightZero 中，obs 不再是像在 DI-engine 中那样的数组，而是一个字典。字典中的观察键对应于 DI-engine 中的 obs，此外，字典还包含诸如 action_mask 和 to_play 等信息。为了代码兼容性，LightZero 还要求环境返回包含 action_mask、to_play 和类似信息的 obs，即使对于非棋盘游戏环境也是如此。

在具体实现中，这些差异主要体现在以下几个方面：

在 reset() 方法中，LightZeroEnv 返回一个字典 lightzero_obs_dict = {’observation’: obs, ‘action_mask’: action_mask, ‘to_play’: -1}。
- 对于非棋盘游戏环境
  - 关于 to_play 的设置：由于非棋盘游戏环境通常只有一个玩家，to_play 被设置为 -1。（在我们的算法中，我们根据这个值判断是执行单玩家算法逻辑（to_play=-1），还是多玩家算法逻辑（to_play=N）。）
  - 关于设置 action_mask：
    - 离散动作空间：action_mask= np.ones(self.env.action_space.n, ‘int8’) 是一个由1组成的numpy数组，表示所有动作都是合法动作。
    - 连续动作空间：action_mask= None，特殊的 None 表示环境是一个连续动作空间。
- 对于棋盘游戏环境：为了便于后续的MCTS过程，lightzero_obs_dict可能还包括诸如棋盘信息board和当前玩家索引current_player_index等变量。
在步骤方法中，返回 BaseEnvTimestep(lightzero_obs_dict, rew, done, info)，其中 lightzero_obs_dict 包含更新后的观察结果。

基本步骤

以下是创建自定义 LightZero 环境的基本步骤：

1. Create the Environment Class

首先，你需要创建一个新的环境类，该类继承自 DI-engine 中的 BaseEnv 类。例如：

from ding.envs import BaseEnv

2. init Method


在您的自定义环境类中，您需要定义一个初始化方法 init。在此方法中，您需要设置环境的一些基本属性，例如观察空间、动作空间、奖励空间等。例如：

def __init__(self, cfg=None):
    self.cfg = cfg
    self._init_flag = False
    # set other properties...

3. Reset Method


reset 方法用于将环境重置为初始状态。此方法应返回环境的初始观察结果。例如：

def reset(self):
    # reset the environment...
    obs = self._env.reset()
    # get the action_mask according to the legal action
    ...
    lightzero_obs_dict = {'observation': obs, 'action_mask': action_mask, 'to_play': -1}
    return lightzero_obs_dict

4. Step Method


step 方法接受一个动作作为输入，执行该动作，并返回一个包含新观察、奖励、是否完成以及其他信息的元组。例如：

def step(self, action):
  # The core original env step.
    obs, rew, done, info = self.env.step(action)
    
    if self.cfg.continuous:
        action_mask = None
    else:
        # get the action_mask according to the legal action
        action_mask = np.ones(self.env.action_space.n, 'int8')
    
    lightzero_obs_dict = {'observation': obs, 'action_mask': action_mask, 'to_play': -1}
    
    self._eval_episode_return += rew
    if done:
        info['eval_episode_return'] = self._eval_episode_return
    
    return BaseEnvTimestep(lightzero_obs_dict, rew, done, info)

5. Observation Space and Action Space


在自定义环境中，你需要为观察空间和动作空间提供属性。这些属性是 gym.Space 对象，用于描述观察和动作的形状和类型。例如：

@property
defobservation_space(self):
    return self.env.observation_space

@property
def action_space(self):
    return self.env.action_space

6. Render Method


render 方法用于向用户展示游戏的玩法。对于实现了 render 方法的环境，用户可以选择是否在执行 step 函数期间调用 render 来在每一步渲染游戏状态。

def render(self, mode: str = 'image_savefile_mode') -> None:
        """
        Overview:
            Renders the game environment.
        Arguments:
            - mode (:obj:`str`): The rendering mode. Options are 
            'state_realtime_mode', 
            'image_realtime_mode', 
            or 'image_savefile_mode'.
        """
        # In 'state_realtime_mode' mode, print the current game board for rendering.
        if mode == "state_realtime_mode":
            ...
        # In other two modes, use a screen for rendering. 
        # Draw the screen.
        ...
        if mode == "image_realtime_mode":
            # Render the picture to user's window.
            ...
        elif mode == "image_savefile_mode":
            # Save the picture to frames.
            ...
            self.frames.append(self.screen)
        return None

在渲染函数中，有三种不同的模式可用：

在 state_realtime_mode 中，render 直接打印当前状态。
在 image_realtime_mode 中，渲染使用图形资源来渲染环境状态，创建一个视觉界面并在实时窗口中显示。
在 image_savefile_mode 中，渲染器将渲染的图像保存在 self.frames 中，并在游戏结束时使用 save_render_output 将它们转换为文件。

在运行时，render 使用的模式取决于 self.render_mode 的值。如果 self.render_mode 设置为 None，环境将不会调用 render 方法。

7. Other Methods


根据需求，您可能还需要定义其他方法，例如 close（用于关闭环境并执行清理操作）等。

8. Register the Environment


最后，你需要使用 ENV_REGISTRY.register 装饰器来注册你的新环境，以便它可以在配置文件中使用。例如：

from ding.utils import ENV_REGISTRY

@ENV_REGISTRY.register('my_custom_env')
class MyCustomEnv(BaseEnv):
    # ...

一旦环境被注册，您可以在配置文件的 create_config 部分指定创建相应环境：

create_config = dict(
    env=dict(
        type='my_custom_env',
        import_names=['zoo.board_games.my_custom_env.envs.my_custom_env'],
    ),
    ...
)

在配置中，类型应设置为已注册的环境名称，而 import_names 应设置为环境包的位置。

创建自定义环境可能需要对特定任务和强化学习有深入的理解。在实现自定义环境时，您可能需要进行实验和调整，以使环境能够有效支持强化学习训练。

棋盘游戏环境的特殊方法

以下是创建自定义桌游环境在 LightZero 中的额外步骤：

LightZero 的棋盘游戏环境有三种不同的模式：self_play_mode、play_with_bot_mode 和 eval_mode。以下是对这些模式的解释：
- self_play_mode: 在此模式下，环境遵循棋盘游戏的经典设置。每次调用step函数时，都会根据提供的动作在环境中放置一个棋子。当游戏决定胜负时，返回+1的奖励。在游戏未决定胜负的所有其他时间步长中，奖励为0。
- play_with_bot_mode: 在此模式下，每次调用 step 函数都会根据提供的动作在环境中放置一个棋子，然后由机器人根据该动作生成一个动作并放置一个棋子。换句话说，代理作为玩家 1 进行游戏，机器人作为玩家 2 与代理对战。游戏结束时，如果代理获胜，则返回 +1 的奖励。如果机器人获胜，则返回 -1 的奖励。如果是平局，则奖励为 0。在游戏未决定的任何其他时间步中，奖励为 0。
- eval_mode: 此模式用于评估当前代理的水平。有两种评估方法：机器人评估和人类评估。在机器人评估中，类似于play_with_bot_mode，机器人作为玩家2与代理对战，并根据结果计算代理的胜率。在人类评估中，用户作为玩家2，通过在命令行中输入动作与代理进行交互。
在每种模式下，游戏结束时，会记录玩家1视角下的eval_episode_return信息（如果玩家1获胜，eval_episode_return为1；如果玩家1失败，则为-1；如果是平局，则为0），并在最后的时间步中记录。
在棋盘游戏环境中，随着游戏的进行，可用的动作可能会减少。因此，有必要实现 legal_action 方法。该方法可用于验证玩家提供的动作，并在 MCTS 过程中生成子节点。以 Connect4 环境为例，该方法检查游戏板上每一列是否已满，并返回一个列表。列表中的值在可以进行移动的列中为 1，在其他位置为 0。

def legal_actions(self) -> List[int]:
    return [i for i in range(7) if self.board[i] == 0]

在LightZero的棋盘游戏环境中，需要实现额外的动作生成方法，例如bot_action和random_action。bot_action方法根据self.bot_action_type的值检索相应类型的bot，并使用bot中预先实现的算法生成动作。另一方面，random_action从当前合法动作列表中选择一个随机动作。bot_action用于play_with_bot_mode中，以实现与bot的交互，而random_action在代理和bot选择动作时以一定概率被调用，以增加游戏样本的随机性。

def bot_action(self) -> int:
    if np.random.rand() < self.prob_random_action_in_bot:
        return self.random_action()
    else:
        if self.bot_action_type == 'rule':
            return self.rule_bot.get_rule_bot_action(self.board, self._current_player)
        elif self.bot_action_type == 'mcts':
            return self.mcts_bot.get_actions(self.board, player_index=self.current_player_index)

LightZeroEnvWrapper

我们在 lzero/envs/wrappers 目录下提供了一个 LightZeroEnvWrapper。它将 classic_control 和 box2d 环境包装成 LightZero 所需的格式。在初始化过程中，一个原始环境被传递给 LightZeroEnvWrapper 实例，该实例使用父类 gym.Wrapper 进行初始化。这使得实例能够调用原始环境中的方法，如 render、close 和 seed。基于此，LightZeroEnvWrapper 类重写了 step 和 reset 方法，将它们的输出包装成符合 LightZero 要求的字典 lightzero_obs_dict。因此，包装后的环境实例满足了 LightZero 自定义环境的要求。

class LightZeroEnvWrapper(gym.Wrapper):
    # overview comments
    def __init__(self, env: gym.Env, cfg: EasyDict) -> None:
        # overview comments
        super().__init__(env)
        ...

具体来说，使用以下函数将gym环境包装成LightZero所需的格式，使用LightZeroEnvWrapper。get_wrappered_env函数返回一个匿名函数，每次调用时都会生成一个DingEnvWrapper实例。该实例将LightZeroEnvWrapper作为匿名函数，并在内部将原始环境包装成LightZero所需的格式。

def get_wrappered_env(wrapper_cfg: EasyDict, env_id: str):
    # overview comments
    ...
    if wrapper_cfg.manually_discretization:
        return lambda: DingEnvWrapper(
            gym.make(env_id),
            cfg={
                'env_wrapper': [
                    lambda env: ActionDiscretizationEnvWrapper(env, wrapper_cfg), lambda env:
                    LightZeroEnvWrapper(env, wrapper_cfg)
                ]
            }
        )
    else:
        return lambda: DingEnvWrapper(
            gym.make(env_id), cfg={'env_wrapper': [lambda env: LightZeroEnvWrapper(env, wrapper_cfg)]}
        )

然后在算法的主入口点调用 train_muzero_with_gym_env 方法，您可以使用包装的环境进行训练：

if __name__ == "__main__":
    """
    Overview:
        The ``train_muzero_with_gym_env`` entry means that the environment used in the training process is generated by wrapping the original gym environment with LightZeroEnvWrapper.
        Users can refer to lzero/envs/wrappers for more details.
    """
    from lzero.entry import train_muzero_with_gym_env
    train_muzero_with_gym_env([main_config, create_config], seed=0, max_env_step=max_env_step)

考虑因素

状态表示：考虑如何将环境状态表示为观察空间。对于简单的环境，您可以直接使用低维连续状态；对于复杂的环境，您可能需要使用图像或其他高维离散状态。
预处理观察空间：根据观察空间的类型，对输入数据执行适当的预处理操作，例如缩放、裁剪、灰度化、归一化等。预处理可以减少输入数据的维度并加速学习过程。
奖励设计：设计一个与目标一致的合理奖励函数。例如，尝试将环境给出的外在奖励归一化到 [0, 1]。通过归一化环境给出的外在奖励，你可以更好地确定内在奖励和其他超参数在 RND 算法中的权重。

如何在 LightZero 中自定义您的环境？

与BaseEnv的主要区别

基本步骤

1. Create the Environment Class

2. init Method

3. Reset Method

4. Step Method

5. Observation Space and Action Space

6. Render Method

7. Other Methods

8. Register the Environment