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 棋类环境的额外步骤：
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），并记录在最后一个时间步中。在棋类环境中，随着对局的推进，可以采取的动作会不断变少，因此还需要实现
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，而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 被初始化，这使得实例可以调用原始环境中的 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 环境，通过 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 算法中的内在奖励权重等超参数。