在强化学习中，环境（Environment）是智能体（Agent）进行学习和互动的场所，它定义了状态空间、动作空间以及奖励机制。Env Wrapper（环境包装器）提供了一种方便的机制来增强或修改原始环境的功能，而不需要改变环境本身的代码。
Env Wrapper主要有以下的几个特性
-
预处理和归一化：
- 为了提高学习效率和稳定性，很多时候需要对环境的状态进行预处理，例如归一化处理，使得所有的输入特征都处在同一量级上，以便于智能体更好地学习。
- Env Wrapper可以在不修改原始环境代码的情况下，添加这些预处理步骤。
-
奖励形状改变（Reward Shaping）：
- 有时原始环境的奖励信号太稀疏或不利于学习，Env Wrapper可以用来修改奖励信号，使之更加符合特定学习任务的需求。
-
动作和状态空间修改：
- 如果需要对动作空间进行离散化或者对状态空间进行降维，Env Wrapper允许用户实现这样的变换。
- 这对于将复杂环境简化以适应低容量智能体或算法尤其有用。
-
实验一致性和复现性：
- 当研究者想要分享他们的实验设置或与其他研究者进行比较时，使用Env Wrapper可以确保所有人都在相同的环境变种上进行实验。
-
代码的模块化和复用：
- 通过Env Wrappers，可以创建可复用的模块，对不同的环境应用相同的处理逻辑。
- 这样可以避免代码冗余，提高开发效率。
-
易于维护和扩展：
- 当需要更新环境处理逻辑时，只需修改对应的Env Wrapper，而不需要重新触碰环境本身的实现。
例如，若要为一个环境添加噪声处理或者状态的标准化处理，可以创建一个Env Wrapper来实现这些功能。这样，当需要在不同的环境中实施同样的噪声或标准化处理时，只需要将环境通过这个Env Wrapper传递即可，而不需要为每个环境单独编写处理代码。
DI-engine 提供了大量已经定义好的、通用的 Env Wrapper
DI-engine 继承和扩展了 OpenAI Gym 的设计，提供了一系列通用的 Env Wrapper，以便用户可以轻松地根据自己的需求来修改环境。以下是这些 Env Wrapper 的简介：
-
NoopResetEnv：- 在环境启动时随机执行一定数量的无操作（no-op）步骤，然后再重置环境。这有助于提供更多样化的初始状态。
-
MaxAndSkipEnv：- 在连续的几个帧中执行相同的动作，并从中选取两个连续帧的最大值作为代表，这样做可以减少环境内的非必要变化，有助于稳定训练。
-
WarpFrame：- 将输入的图像帧缩放到固定的大小（例如 84x84）并转换为灰度图像，这是许多深度强化学习工作中的常见做法。
-
ScaledFloatFrame：- 将图像帧的像素值从整数转换为浮点数，并将其标准化到 0 到 1 的范围内。
-
ClipRewardEnv：- 将奖励剪裁为 -1, 0, 或 +1，根据奖励的正负性。这样做有利于限制梯度更新的规模，有时能够增加训练的稳定性。
-
FrameStack：- 将连续的几个帧堆叠在一起作为网络的输入，以提供给智能体关于动作的时间连续性信息。
-
ObsTransposeWrapper：- 重新排列观测数组的维度，这在处理图像输入时特别有用，如将图像数据从 HxWxC 调整为 CxHxW。
-
RunningMeanStd：- 计算和更新环境观测值的均值和标准差，通常用于环境状态的归一化处理。
-
ObsNormEnv：- 使用实时更新的均值和标准差归一化观测值，以保持输入值的分布稳定。
-
RewardNormEnv：- 使用实时更新的均值和标准差归一化奖励值，帮助稳定训练过程。
-
RamWrapper：- 将环境的 RAM 状态转换为类似图像的格式，这对于处理非图像的原始状态表示特别有用。
-
EpisodicLifeEnv：- 当智能体“死亡”时，即使游戏没有真正结束，也会结束当前的 episode。这可以让智能体更快地学习有关失败的信息。
-
FireResetEnv：- 在某些游戏中，环境重置后需要执行一个 “fire” 动作来开始游戏。这个 Wrapper 自动处理这个动作。
二、如何使用 Env Wrapper
下一个问题是我们如何给环境包裹上 Env Wrapper。最简单的一种方法就是手动地显式对环境进行包裹：
from ding.envs.env_wrappers.env_wrappers import NoopResetWrapper
env = gym.make(env_id) # 'PongNoFrameskip-v4'
env = NoopResetWrapper(env, noop_max = 30)
env = MaxAndSkipWrapper(env, skip = 4)
我们也可以尝试在gym中使用FrameStack环境包裹器
import gym
from gym.wrappers import FrameStack
# 创建并包裹环境
env_id = 'CartPole-v1'
env = gym.make(env_id)
env = FrameStack(env, num_stack=4)
# 使用包裹后的环境运行一些episode
for episode in range(2):
obs = env.reset()
done = False
step = 0
while not done:
action = env.action_space.sample() # 随机选择动作
obs, reward, done, info = env.step(action)
print(f"Episode: {episode}, Step: {step}, Observation Shape: {obs.shape}, Reward: {reward}, Done: {done}")
step += 1
print(f"Episode {episode} finished after {step} steps.")
# 关闭环境
env.close()
以下是在 Python 中非显式的使用 Env Wrapper 的具体示例：
import gym
from ditk import logging
from ding.model.template.qac_dist import QACDIST
from ding.policy import D4PGPolicy
from ding.envs import DingEnvWrapper, BaseEnvManagerV2
from ding.data import DequeBuffer
from ding.data.buffer.middleware import PriorityExperienceReplay
from ding.config import compile_config
from ding.framework import task
from ding.framework.context import OnlineRLContext
from ding.framework.middleware import OffPolicyLearner, StepCollector, interaction_evaluator, data_pusher, \
CkptSaver, nstep_reward_enhancer
from ding.utils import set_pkg_seed
from dizoo.classic_control.pendulum.envs.pendulum_env import PendulumEnv
from dizoo.classic_control.pendulum.config.pendulum_d4pg_config import main_config, create_config
def main():
logging.getLogger().setLevel(logging.INFO)
cfg = compile_config(main_config, create_cfg=create_config, auto=True)
with task.start(async_mode=False, ctx=OnlineRLContext()):
collector_env = BaseEnvManagerV2(
env_fn=[lambda: PendulumEnv(cfg.env) for _ in range(cfg.env.collector_env_num)], cfg=cfg.env.manager
)
evaluator_env = BaseEnvManagerV2(
env_fn=[lambda: PendulumEnv(cfg.env) for _ in range(cfg.env.evaluator_env_num)], cfg=cfg.env.manager
)
set_pkg_seed(cfg.seed, use_cuda=cfg.policy.cuda)
model = QACDIST(**cfg.policy.model)
buffer_ = DequeBuffer(size=cfg.policy.other.replay_buffer.replay_buffer_size)
buffer_.use(PriorityExperienceReplay(buffer_, IS_weight=True))
policy = D4PGPolicy(cfg.policy, model=model)
task.use(interaction_evaluator(cfg, policy.eval_mode, evaluator_env))
task.use(
StepCollector(cfg, policy.collect_mode, collector_env, random_collect_size=cfg.policy.random_collect_size)
)
task.use(nstep_reward_enhancer(cfg))
task.use(data_pusher(cfg, buffer_))
task.use(OffPolicyLearner(cfg, policy.learn_mode, buffer_))
task.use(CkptSaver(policy, cfg.exp_name, train_freq=100))
task.run()
if __name__ == "__main__":
main()
这段代码是一个使用DI-engine框架和dizoo环境库设置的深度确定性策略梯度（D4PG）算法的强化学习训练流程的例子。代码的目的是训练一个智能体在Pendulum（倒立摆）环境中做出决策。
在这段代码中，环境包裹器（Env Wrapper）的使用是非显式的。PendulumEnv 可能已经在它的构造函数内部使用了环境包裹器，但这不是在这段具体代码中直接声明的。如果需要对环境进行额外的包装（如修改观察空间或奖励结构），这将在 PendulumEnv 类的实现中完成，或者通过传递给 PendulumEnv 的 cfg.env 配置参数来实现。
这里我们挑战到PendulumEnv 类的定义中：
class PendulumEnv(BaseEnv):
def __init__(self, cfg: dict) -> None:
self._cfg = cfg
self._act_scale = cfg.act_scale
self._env = gym.make('Pendulum-v1')
self._init_flag = False
self._replay_path = None
if 'continuous' in cfg.keys():
self._continuous = cfg.continuous
else:
self._continuous = True
self._observation_space = gym.spaces.Box(
low=np.array([-1.0, -1.0, -8.0]), high=np.array([1.0, 1.0, 8.0]), shape=(3, ), dtype=np.float32
)
if self._continuous:
self._action_space = gym.spaces.Box(low=-2.0, high=2.0, shape=(1, ), dtype=np.float32)
else:
self._discrete_action_num = 11
self._action_space = gym.spaces.Discrete(self._discrete_action_num)
self._action_space.seed(0) # default seed
self._reward_space = gym.spaces.Box(
low=-1 * (3.14 * 3.14 + 0.1 * 8 * 8 + 0.001 * 2 * 2), high=0.0, shape=(1, ), dtype=np.float32
)
def reset(self) -> np.ndarray:
if not self._init_flag:
self._env = gym.make('Pendulum-v1')
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))
)
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)
self._action_space.seed(self._seed + np_seed)
elif hasattr(self, '_seed'):
self._env.seed(self._seed)
self._action_space.seed(self._seed)
obs = self._env.reset()
obs = to_ndarray(obs).astype(np.float32)
self._eval_episode_return = 0.
return obs
def close(self) -> None:
if self._init_flag:
self._env.close()
self._init_flag = False
def seed(self, seed: int, dynamic_seed: bool = True) -> None:
self._seed = seed
self._dynamic_seed = dynamic_seed
np.random.seed(self._seed)
def step(self, action: np.ndarray) -> BaseEnvTimestep:
assert isinstance(action, np.ndarray), type(action)
# if require discrete env, convert actions to [-1 ~ 1] float actions
if not self._continuous:
action = (action / (self._discrete_action_num - 1)) * 2 - 1
# scale into [-2, 2]
if self._act_scale:
action = affine_transform(action, min_val=self._env.action_space.low, max_val=self._env.action_space.high)
obs, rew, done, info = self._env.step(action)
self._eval_episode_return += rew
obs = to_ndarray(obs).astype(np.float32)
# wrapped to be transfered to a array with shape (1,)
rew = to_ndarray([rew]).astype(np.float32)
if done:
info['eval_episode_return'] = self._eval_episode_return
return BaseEnvTimestep(obs, rew, done, info)
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 random_action(self) -> np.ndarray:
# consider discrete
if self._continuous:
random_action = self.action_space.sample().astype(np.float32)
else:
random_action = self.action_space.sample()
random_action = to_ndarray([random_action], dtype=np.int64)
return random_action
@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
def __repr__(self) -> str:
return "DI-engine Pendulum Env({})".format(self._cfg.env_id)
这段代码是一个名为 PendulumEnv 的类定义，它继承自 BaseEnv（DI-engine中的基础环境类）。这个自定义环境是对 OpenAI Gym 的 Pendulum-v1 环境的封装和扩展。下面我们将逐句分析这个类的定义：
初始化方法 (init)
def __init__(self, cfg: dict) -> None:
- 这个初始化方法接受一个配置字典
cfg，用于设置环境。
self._env = gym.make('Pendulum-v1')
- 利用
gym.make方法创建了一个原始的Pendulum-v1环境。
self._init_flag = False
- 初始标志设置为
False，表示还未进行额外的初始化步骤。
self._observation_space = gym.spaces.Box(...)
self._action_space = gym.spaces.Box(...) or gym.spaces.Discrete(...)
- 重定义观测空间和动作空间。如果环境是连续的（
self._continuous == True），则使用Box空间；如果不是连续的，则使用Discrete空间。
重置方法 (reset)
def reset(self) -> np.ndarray:
reset方法用于将环境重置到初始状态，并返回初始观测。
if self._replay_path is not None:
self._env = gym.wrappers.RecordVideo(...)
- 如果指定了回放路径 (
self._replay_path)，那么这里实际上使用了一个 Gym 环境包裹器RecordVideo来包装原始环境，以便录制回放视频。
self._env.seed(self._seed + np_seed)
self._action_space.seed(self._seed + np_seed)
- 在这里，如果设置了动态种子，环境和动作空间的种子会被设置为一个基于原始种子的新值。
步骤方法 (step)
def step(self, action: np.ndarray) -> BaseEnvTimestep:
step方法接受一个动作action，执行一步交互，并返回新的观测、奖励、是否结束的标志以及额外信息。
if not self._continuous:
action = (action / (self._discrete_action_num - 1)) * 2 - 1
- 如果环境是离散的，动作被转换成一个连续的范围。
if self._act_scale:
action = affine_transform(action, min_val=self._env.action_space.low, max_val=self._env.action_space.high)
- 如果设置了动作缩放（
self._act_scale），动作会被缩放到环境的有效动作范围内。
其它方法
close方法用于关闭环境。seed方法用于设置环境的随机种子。enable_save_replay方法用于启用回放录制功能。random_action方法返回一个随机动作。
环境包裹器的使用
在 PendulumEnv 类中，环境包裹器的应用体现在 reset 方法中对 Gym 环境的包装。具体来说，就是使用 RecordVideo 包装器来录制视频。这个包装器是 Gym 提供的一个标准包装器，用于在环境执行过程中录制视频。以下是相关代码段：
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))
)
在这里，RecordVideo 包装器被应用于内部的 _env 属性，它是通过 gym.make('Pendulum-v1') 创建的原始 Pendulum 环境。包装器的作用是在每个 episode 开始时触发视频录制，将视频保存在指定的 self._replay_path 路径下。
这表明 PendulumEnv 类并没有直接继承或修改 Gym 环境的基本行为，而是选择性地在环境实例上应用了一个包装器。这种使用方式是“非显式”的，因为它不是通过像 DingEnvWrapper 这样的外部环境包装器类来实现的，而是直接在环境类的方法内部根据条件应用的。这种方法的优点是可以根据配置文件中的参数动态地决定是否应用特定的包装器，而不需要改变环境类本身的定义。
评论（0）
您还未登录，请登录后发表或查看评论