LightZero 中如何自定义环境?

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

  • 一个典型的 LightZero 中的环境,请参考 atari_lightzero_env.py 。LightZero的环境设计大致基于DI-engine的BaseEnv类。在创建自定义环境时,我们遵循了与DI-engine相似的基本步骤。以下是 DI-engine 中创建自定义环境的文档

    • https://di-engine-docs.readthedocs.io/zh_CN/latest/04_best_practice/ding_env_zh.html

与 BaseEnv 的主要差异

在 LightZero 中,有很多棋类环境。棋类环境由于存在玩家交替执行动作,合法动作在变化的情况,所以环境的观测状态除了棋面信息,还应包含动作掩码,当前玩家等信息。因此,LightZero 中的 obs 不再像 DI-engine 中那样是一个数组,而是一个字典。字典中的 'observation' 对应于 DI-engine 中的 obs,此外字典中还包含了 'action_mask''to_play' 等信息。为了代码的兼容性,对于非棋类环境,LightZero 同样要求环境返回的 obs 包含'action_mask''to_play' 等信息。

在具体的方法实现中,这种差异主要体现在下面几点:

  • reset 方法中,LightZeroEnv 返回的是一个字典 lightzero_obs_dict = {'observation': obs, 'action_mask': action_mask, 'to_play': -1}

    • 对于非棋类环境

      • to_play 的设置:由于非棋类环境一般只有一个玩家,因此设置 to_play =-1 。(我们在算法中根据该值,判断执行单player的算法逻辑 (to_play =-1) ,还是多player的算法逻辑 (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 和当前玩家 curren_player_index 等变量。

  • step 方法中,返回的是 BaseEnvTimestep(lightzero_obs_dict, rew, done, info) ,其中的 lightzero_obs_dict 包含了更新后的观察结果。

基本步骤

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

1. 创建环境类

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

from ding.envs import BaseEnv

class MyCustomEnv(BaseEnv):
    pass

2. __init__方法

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

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

3. Reset 方法

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 方法

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. 观察空间和动作空间

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

@property
def observation_space(self):
    return self._observation_space

@property
def action_space(self):
    return self._action_space
    
@property
def legal_actions(self):
    # get the actual legal actions
    return np.arange(self._action_space.n)

6. render 方法

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

render 中,有三种不同的模式。

  • state_realtime_mode 下,render 会直接打印当前状态。

  • image_realtime_mode 下, render 会根据一些图形素材将环境状态渲染出来,形成可视化的界面,并弹出实时的窗口展示。

  • image_savefile_mode 下, render 会将渲染的图像保存在 self.frames 中,并在对局结束时通过 save_render_output 将其转化为文件保存下来。 在运行时, render 所采取的模式取决于 self.render_mode 的取值。当 self.render_mode 取值为 None 时,环境不会调用 render 方法。

7. 其他方法

根据需要,可能还需要定义其他方法,例如 close (用于关闭环境并进行清理)等。

8. 注册环境

最后,需要使用 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'],
    ),
    ...
)

其中 type 要设定为所注册的环境名, import_names 则设置为环境包的位置。

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

棋类环境的特殊方法

以下是创建自定义 LightZero 棋类环境的额外步骤:

  1. LightZero中的棋类环境有三种不同的模式: self_play_mode , play_with_bot_mode , eval_mode 。这三种模式的说明如下:

    • self_play_mode:该模式下,采取棋类环境的经典设置,每调用一次 step 函数,会根据传入的动作在环境中落子一次。在分出胜负的时间步,会返回+1的 reward 。在没有分出胜负的所有时间步, reward 均为0。

    • play_with_bot_mode:该模式下,每调用一次 step 函数,会根据传入的动作在环境中落子一次,随后调用环境中的 bot 产生一个动作,并根据 bot 的动作再落子一次。也就是说, agent 扮演了1号玩家的角色,而 bot 扮演了2号玩家的角色和 agent 对抗。在对局结束时,如果 agent 胜利,则返回+1的 reward ,如果 bot 胜利,则返回-1的 reward ,平局则 reward 为0。在其余没有分出胜负的时间步, reward 均为0。

    • eval_mode:该模式用于评估当前的 agent 的水平。具体有 bot 和 human 两种评估方法。采取 bot 评估时,和 play_with_bot_mode 中一样,会让 bot 扮演2号玩家和 agent 对抗,并根据结果计算 agent 的胜率。采取 human 模式时,则让用户扮演2号玩家,在命令行输入动作和 agent 对打。

    每种模式下,在棋局结束后,都会从1号玩家的视角记录本局的 eval_episode_return 信息(如果1号玩家赢了,则 eval_episode_return 为1,如果输了为-1,平局为0),并记录在最后一个时间步中。

  2. 在棋类环境中,随着对局的推进,可以采取的动作会不断变少,因此还需要实现 legal_action 方法。该方法可以用于检验玩家输入的动作是否合法,以及在 MCTS 过程中根据合法动作生成子节点。以 Connect4 环境为例,该方法会检查棋盘中的每一列是否下满,然后返回一个列表。该列表在可以落子的列取值为1,其余位置取值为0。

def legal_actions(self) -> List[int]:
        return [i for i in range(7) if self.board[i] == 0]
  1. LightZero的棋类环境中,还需要实现一些动作生成方法,例如 bot_actionrandom_action 。其中 bot_action 会根据 self.bot_action_type 的值调取相应种类的 bot ,通过 bot 中预实现的算法生成一个动作。而 random_action 则会从当前的合法动作列表中随机选取一个动作返回。 bot_action 用于实现环境的 play_with_bot_mode ,而 random_action 则会在 agent 和 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 所需要的环境格式。在初始化实例时,会传入一个原始环境,这个原始环境通过父类 gym.Wrapper 被初始化,这使得实例可以调用原始环境中的 rendercloseseed 等方法。在此基础上, LightZeroEnvWrapper 类重写了 stepreset 方法,将其输出封装成符合 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 环境,通过 LightZeroEnvWrapper 包装成 LightZero 所需要的环境格式。 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 方法,即可使用上述包装后的 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 算法中的内在奖励权重等超参数。