杂项包装器

常用包装器

class gymnasium.wrappers.TimeLimit(env: Env, max_episode_steps: int)[source]

通过在超过最大步数时截断环境来限制环境的步数。

如果环境本身未定义截断,这是发出截断信号的唯一位置。至关重要的是,这不同于源自基础环境作为 MDP 一部分的 terminated 信号。不存在矢量包装器。

使用 TimeLimit 包装器的示例
>>> from gymnasium.wrappers import TimeLimit
>>> from gymnasium.envs.classic_control import CartPoleEnv
>>> spec = gym.spec("CartPole-v1")
>>> spec.max_episode_steps
500
>>> env = gym.make("CartPole-v1")
>>> env  # TimeLimit is included within the environment stack
<TimeLimit<OrderEnforcing<PassiveEnvChecker<CartPoleEnv<CartPole-v1>>>>>
>>> env.spec  
EnvSpec(id='CartPole-v1', ..., max_episode_steps=500, ...)
>>> env = gym.make("CartPole-v1", max_episode_steps=3)
>>> env.spec  
EnvSpec(id='CartPole-v1', ..., max_episode_steps=3, ...)
>>> env = TimeLimit(CartPoleEnv(), max_episode_steps=10)
>>> env
<TimeLimit<CartPoleEnv instance>>
TimeLimit 确定剧集步骤的示例
>>> env = gym.make("CartPole-v1", max_episode_steps=3)
>>> _ = env.reset(seed=123)
>>> _ = env.action_space.seed(123)
>>> _, _, terminated, truncated, _ = env.step(env.action_space.sample())
>>> terminated, truncated
(False, False)
>>> _, _, terminated, truncated, _ = env.step(env.action_space.sample())
>>> terminated, truncated
(False, False)
>>> _, _, terminated, truncated, _ = env.step(env.action_space.sample())
>>> terminated, truncated
(False, True)
更改日志
  • v0.10.6 - 最初添加

  • v0.25.0 - 随着 step API 的更新,终止和截断信号将分别返回。

参数:
  • env – 要应用包装器的环境

  • max_episode_steps – 剧集被截断后的环境步数 (elapsed >= max_episode_steps)

class gymnasium.wrappers.RecordVideo(env: gym.Env[ObsType, ActType], video_folder: str, episode_trigger: Callable[[int], bool] | None = None, step_trigger: Callable[[int], bool] | None = None, video_length: int = 0, name_prefix: str = 'rl-video', fps: int | None = None, disable_logger: bool = True)[source]

使用环境的渲染函数录制环境剧集的视频。

通常,你只希望间歇性地录制剧集,比如每百个剧集或每千个环境步骤录制一次。为此,你可以指定 episode_triggerstep_trigger。它们应该是返回布尔值的函数,表示是否应该在当前剧集或步骤分别开始录制。

episode_trigger 应该在录制应该开始的剧集返回 Truestep_trigger 应该在录制应该开始的第 n 个环境步骤返回 True,其中 n 对所有之前的剧集求和。如果既没有传递 episode_trigger 也没有传递 step_trigger,则将使用默认的 episode_trigger,即 capped_cubic_video_schedule()。此函数将在每个 3 的幂次方的剧集开始视频,直到 1000,然后每 1000 个剧集开始视频。默认情况下,录制将在调用重置时停止。但是,你也可以通过传递 video_length 的严格正值来创建固定长度的录制(可能跨越多个剧集)。

不存在此包装器的矢量版本。

示例 - 运行环境 50 个剧集,从第 0 个剧集开始每 10 个剧集保存一次视频
>>> import os
>>> import gymnasium as gym
>>> env = gym.make("LunarLander-v3", render_mode="rgb_array")
>>> trigger = lambda t: t % 10 == 0
>>> env = RecordVideo(env, video_folder="./save_videos1", episode_trigger=trigger, disable_logger=True)
>>> for i in range(50):
...     termination, truncation = False, False
...     _ = env.reset(seed=123)
...     while not (termination or truncation):
...         obs, rew, termination, truncation, info = env.step(env.action_space.sample())
...
>>> env.close()
>>> len(os.listdir("./save_videos1"))
5
示例 - 运行环境 5 个剧集,每 200 个步骤开始录制一次,确保每个视频有 100 帧
>>> import os
>>> import gymnasium as gym
>>> env = gym.make("LunarLander-v3", render_mode="rgb_array")
>>> trigger = lambda t: t % 200 == 0
>>> env = RecordVideo(env, video_folder="./save_videos2", step_trigger=trigger, video_length=100, disable_logger=True)
>>> for i in range(5):
...     termination, truncation = False, False
...     _ = env.reset(seed=123)
...     _ = env.action_space.seed(123)
...     while not (termination or truncation):
...         obs, rew, termination, truncation, info = env.step(env.action_space.sample())
...
>>> env.close()
>>> len(os.listdir("./save_videos2"))
2
示例 - 运行 3 个剧集,录制所有内容,但以 1000 帧为块
>>> import os
>>> import gymnasium as gym
>>> env = gym.make("LunarLander-v3", render_mode="rgb_array")
>>> env = RecordVideo(env, video_folder="./save_videos3", video_length=1000, disable_logger=True)
>>> for i in range(3):
...     termination, truncation = False, False
...     _ = env.reset(seed=123)
...     while not (termination or truncation):
...         obs, rew, termination, truncation, info = env.step(env.action_space.sample())
...
>>> env.close()
>>> len(os.listdir("./save_videos3"))
2
更改日志
  • v0.25.0 - 最初添加以替换 wrappers.monitoring.VideoRecorder

参数:
  • env – 将要被包装的环境

  • video_folder (str) – 将存储录制的文件夹

  • episode_trigger – 接受整数并返回 True 的函数,当且仅当应该在此剧集开始录制时

  • step_trigger – 接受整数并返回 True 的函数,当且仅当应该在此步骤开始录制时

  • video_length (int) – 录制剧集的长度。如果为 0,则录制整个剧集。否则,将捕获指定长度的片段

  • name_prefix (str) – 将预先附加到录制的文件名

  • fps (int) – 视频的每秒帧数。为环境提供自定义视频 fps,如果为 None,则使用环境元数据 render_fps 键(如果存在),否则使用默认值 30。

  • disable_logger (bool) – 是否禁用 moviepy 日志记录器,默认情况下禁用

class gymnasium.wrappers.RecordEpisodeStatistics(env: Env[ObsType, ActType], buffer_length: int = 100, stats_key: str = 'episode')[source]

此包装器将跟踪累积奖励和剧集长度。

在剧集结束时,剧集的统计数据将使用键episode添加到info中。如果使用向量化环境,还会使用键_episode,它指示相应索引处的环境是否具有剧集统计数据。包装器的向量版本存在,gymnasium.wrappers.vector.RecordEpisodeStatistics

剧集完成之后,info将看起来像这样

>>> info = {
...     "episode": {
...         "r": "<cumulative reward>",
...         "l": "<episode length>",
...         "t": "<elapsed time since beginning of episode>"
...     },
... }

对于向量化环境,输出将采用以下形式

>>> infos = {
...     "episode": {
...         "r": "<array of cumulative reward>",
...         "l": "<array of episode length>",
...         "t": "<array of elapsed time since beginning of episode>"
...     },
...     "_episode": "<boolean array of length num-envs>"
... }

此外,最新的奖励和剧集长度存储在缓冲区中,可以通过wrapped_env.return_queuewrapped_env.length_queue分别访问。

变量:
  • time_queue (*) – 最后deque_size个剧集的时间长度

  • return_queue (*) – 最后deque_size个剧集的累积奖励

  • length_queue (*) – 最后deque_size个剧集的长度

更改日志
参数:
  • env (Env) – 要应用包装器的环境

  • buffer_length – 缓冲区return_queuelength_queuetime_queue的大小

  • stats_key – 剧集统计数据的 info 键

class gymnasium.wrappers.AtariPreprocessing(env: gym.Env, noop_max: int = 30, frame_skip: int = 4, screen_size: int | tuple[int, int] = 84, terminal_on_life_loss: bool = False, grayscale_obs: bool = True, grayscale_newaxis: bool = False, scale_obs: bool = False)[source]

实现了 Atari 环境的常用预处理技术(不包括帧堆叠)。

对于帧堆叠,请使用gymnasium.wrappers.FrameStackObservation。包装器没有向量版本

此类遵循 Machado 等人(2018 年)的指南,“重温街机学习环境:通用智能体的评估协议和开放问题”。

具体来说,以下预处理阶段适用于 Atari 环境

  • 无操作重置:通过在重置时执行随机数量的无操作操作来获取初始状态,默认最大 30 个无操作操作。

  • 帧跳过:在步长之间跳过的帧数,默认值为 4。

  • 最大池化:在最近两次来自帧跳过的观察结果上进行池化。

  • 生命值损失时的终止信号:当代理在环境中损失生命时,环境将终止。

    默认情况下已关闭。Machado 等人(2018 年)不建议这样做。

  • 调整为方形图像:默认情况下将 Atari 环境的原始观察形状从 210x180 调整为 84x84。

  • 灰度观察:使观察结果变为灰度,默认情况下已启用。

  • 灰度新轴:扩展观察结果的最后一个通道,使得图像为 3 维,默认情况下未启用。

  • 缩放观察:是否将观察结果缩放到 [0, 1) 或 [0, 255) 之间,默认情况下未缩放。

示例

>>> import gymnasium as gym
>>> import ale_py
>>> gym.register_envs(ale_py)
>>> env = gym.make("ALE/Pong-v5", frameskip=1)
>>> env = AtariPreprocessing(
...     env,
...     noop_max=10, frame_skip=4, terminal_on_life_loss=True,
...     screen_size=84, grayscale_obs=False, grayscale_newaxis=False
... )
更改日志
  • 在 gym v0.12.2 中添加(gym #1455)

参数:
  • env (Env) – 要应用预处理的环境

  • noop_max (int) – 对于无操作重置,重置时采取的最大无操作操作数,要关闭,请设置为 0。

  • frame_skip (int) – 影响代理体验游戏的频率,代理观察结果之间的新观察结果之间的帧数。

  • screen_size (int | tuple[int, int]) – 调整 Atari 帧的大小。

  • terminal_on_life_loss (bool) – 如果为 True,则step()只要生命值损失就返回terminated=True

  • grayscale_obs (bool) – 如果为 True,则返回灰度观察结果,否则返回 RGB 观察结果。

  • grayscale_newaxis (bool) – 如果为 True 且 grayscale_obs=True,则会向灰度观察结果添加一个通道轴,使其成为 3 维。

  • scale_obs (bool) – 如果为 True,则返回范围在 [0,1) 内的归一化观察结果。它还限制了 FrameStack 包装器的内存优化优势。

引发:
  • DependencyNotInstalled – 未安装 opencv-python 包

  • ValueError – 在原始环境中禁用帧跳过

不常见包装器

class gymnasium.wrappers.Autoreset(env: Env)[source]

当达到终止或截断状态时,包装环境会自动重置。

这遵循向量自动重置 API,其中在剧集终止或截断后的步骤中,环境将重置。

更改日志
  • v0.24.0 - 最初添加为 AutoResetWrapper

  • v1.0.0 - 重命名为 Autoreset,并且自动重置顺序已更改为在环境终止或截断后的步骤中重置。因此,“final_observation”“final_info” 已被删除。

参数:

env (gym.Env) – 要应用包装器的环境

class gymnasium.wrappers.PassiveEnvChecker(env: Env[ObsType, ActType])[source]

一个被动包装器,它围绕着stepresetrender函数,以检查它们是否遵循 Gymnasium 的 API。

此包装器在 make 期间会自动应用,可以通过 disable_env_checker 禁用。包装器没有向量版本。

示例

>>> import gymnasium as gym
>>> env = gym.make("CartPole-v1")
>>> env
<TimeLimit<OrderEnforcing<PassiveEnvChecker<CartPoleEnv<CartPole-v1>>>>>
>>> env = gym.make("CartPole-v1", disable_env_checker=True)
>>> env
<TimeLimit<OrderEnforcing<CartPoleEnv<CartPole-v1>>>>
更改日志
  • v0.24.1 - 最初添加,但在某些方面存在缺陷

  • v0.25.0 - 错误已修复

  • v0.29.0 - 删除了对 Box 观察空间和动作空间的无限边界以及不规则边界形状的警告

使用环境初始化包装器,运行观察空间和动作空间测试。

class gymnasium.wrappers.HumanRendering(env: Env[ObsType, ActType])[source]

允许对支持“rgb_array”渲染的环境进行类似人类的渲染。

此包装器在您已实现一个可以生成 RGB 图像的环境但尚未实现任何代码将图像渲染到屏幕上时特别有用。如果您想在您的环境中使用此包装器,请记住在环境的元数据中指定 "render_fps"

包装环境的render_mode必须是'rgb_array''rgb_array_list'

不存在此包装器的矢量版本。

示例

>>> import gymnasium as gym
>>> from gymnasium.wrappers import HumanRendering
>>> env = gym.make("LunarLander-v3", render_mode="rgb_array")
>>> wrapped = HumanRendering(env)
>>> obs, _ = wrapped.reset()     # This will start rendering to the screen

包装器也可以在实例化环境时直接应用,只需将render_mode="human"传递给make。只有在环境未原生实现人类渲染(即render_mode不包含"human")时,才会应用包装器。

>>> env = gym.make("phys2d/CartPole-v1", render_mode="human")      # CartPoleJax-v1 doesn't implement human-rendering natively
>>> obs, _ = env.reset()     # This will start rendering to the screen

警告:如果基础环境使用 render_mode="rgb_array_list",其(即 *基础环境*)的渲染方法将始终返回一个空列表。

>>> env = gym.make("LunarLander-v3", render_mode="rgb_array_list")
>>> wrapped = HumanRendering(env)
>>> obs, _ = wrapped.reset()
>>> env.render() # env.render() will always return an empty list!
[]
更改日志
  • v0.25.0 - 首次添加

参数:

env – 被包装的环境

class gymnasium.wrappers.OrderEnforcing(env: Env[ObsType, ActType], disable_render_order_enforcing: bool = False)[source]

如果在 reset 之前调用了 steprender,将产生错误。

不存在此包装器的矢量版本。

示例

>>> import gymnasium as gym
>>> from gymnasium.wrappers import OrderEnforcing
>>> env = gym.make("CartPole-v1", render_mode="human")
>>> env = OrderEnforcing(env)
>>> env.step(0)
Traceback (most recent call last):
    ...
gymnasium.error.ResetNeeded: Cannot call env.step() before calling env.reset()
>>> env.render()
Traceback (most recent call last):
    ...
gymnasium.error.ResetNeeded: Cannot call `env.render()` before calling `env.reset()`, if this is an intended action, set `disable_render_order_enforcing=True` on the OrderEnforcer wrapper.
>>> _ = env.reset()
>>> env.render()
>>> _ = env.step(0)
>>> env.close()
更改日志
  • v0.22.0 - 首次添加

  • v0.24.0 - 为渲染函数添加了顺序强制

参数:
  • env – 要包装的环境

  • disable_render_order_enforcing – 是否禁用渲染顺序强制

class gymnasium.wrappers.RenderCollection(env: Env[ObsType, ActType], pop_frames: bool = True, reset_clean: bool = True)[source]

收集环境的渲染帧,例如 render 返回一个 list[RenderedFrame]

不存在此包装器的矢量版本。

示例

返回 render 未被调用的步数的帧列表。>>> import gymnasium as gym >>> env = gym.make(“LunarLander-v3”, render_mode=”rgb_array”) >>> env = RenderCollection(env) >>> _ = env.reset(seed=123) >>> for _ in range(5): … _ = env.step(env.action_space.sample()) … >>> frames = env.render() >>> len(frames) 6

>>> frames = env.render()
>>> len(frames)
0

返回该回合运行的步数的帧列表。>>> import gymnasium as gym >>> env = gym.make(“LunarLander-v3”, render_mode=”rgb_array”) >>> env = RenderCollection(env, pop_frames=False) >>> _ = env.reset(seed=123) >>> for _ in range(5): … _ = env.step(env.action_space.sample()) … >>> frames = env.render() >>> len(frames) 6

>>> frames = env.render()
>>> len(frames)
6

收集所有回合的所有帧,在调用渲染时不清除它们。>>> import gymnasium as gym >>> env = gym.make(“LunarLander-v3”, render_mode=”rgb_array”) >>> env = RenderCollection(env, pop_frames=False, reset_clean=False) >>> _ = env.reset(seed=123) >>> for _ in range(5): … _ = env.step(env.action_space.sample()) … >>> _ = env.reset(seed=123) >>> for _ in range(5): … _ = env.step(env.action_space.sample()) … >>> frames = env.render() >>> len(frames) 12

>>> frames = env.render()
>>> len(frames)
12
更改日志
  • v0.26.2 - 首次添加

参数:
  • env – 被包装的环境

  • pop_frames (bool) – 如果为真,则在调用 meth:render 后清除集合帧。默认值为 True

  • reset_clean (bool) – 如果为真,则在调用 meth:reset 后清除集合帧。默认值为 True

数据转换包装器

class gymnasium.wrappers.JaxToNumpy(env: Env[ObsType, ActType])[source]

包装基于 Jax 的环境,使其能够与 NumPy 数组交互。

动作必须以 numpy 数组的形式提供,并且观察结果将以 numpy 数组的形式返回。包装器的向量版本存在于 gymnasium.wrappers.vector.JaxToNumpy 中。

笔记

Jax 到 NumPy 和 NumPy 到 Jax 的转换不能保证往返(jax -> numpy -> jax)反之亦然。原因是 jax 不支持非数组值,因此 numpy int_32(5) -> DeviceArray([5], dtype=jnp.int23)

示例

>>> import gymnasium as gym                                     
>>> env = gym.make("JaxEnv-vx")                                 
>>> env = JaxToNumpy(env)                                       
>>> obs, _ = env.reset(seed=123)                                
>>> type(obs)                                                   
<class 'numpy.ndarray'>
>>> action = env.action_space.sample()                          
>>> obs, reward, terminated, truncated, info = env.step(action) 
>>> type(obs)                                                   
<class 'numpy.ndarray'>
>>> type(reward)                                                
<class 'float'>
>>> type(terminated)                                            
<class 'bool'>
>>> type(truncated)                                             
<class 'bool'>
更改日志
  • v1.0.0 - 首次添加

参数:

env – 要包装的 jax 环境

class gymnasium.wrappers.JaxToTorch(env: gym.Env, device: Device | None = None)[source]

包装基于 Jax 的环境,使其能够与 PyTorch 张量交互。

动作必须以 PyTorch 张量形式提供,并且观察结果将以 PyTorch 张量形式返回。包装器的向量版本存在于 gymnasium.wrappers.vector.JaxToTorch 中。

注意

对于 rendered,它将以 NumPy 数组形式返回,而不是 pytorch 张量。

示例

>>> import torch                                                
>>> import gymnasium as gym                                     
>>> env = gym.make("JaxEnv-vx")                                 
>>> env = JaxtoTorch(env)                                       
>>> obs, _ = env.reset(seed=123)                                
>>> type(obs)                                                   
<class 'torch.Tensor'>
>>> action = torch.tensor(env.action_space.sample())            
>>> obs, reward, terminated, truncated, info = env.step(action) 
>>> type(obs)                                                   
<class 'torch.Tensor'>
>>> type(reward)                                                
<class 'float'>
>>> type(terminated)                                            
<class 'bool'>
>>> type(truncated)                                             
<class 'bool'>
更改日志
  • v1.0.0 - 首次添加

参数:
  • env – 要包装的基于 Jax 的环境

  • device – 应该将 torch 张量移动到的设备

class gymnasium.wrappers.NumpyToTorch(env: gym.Env, device: Device | None = None)[source]

包装基于 NumPy 的环境,使其能够与 PyTorch 张量交互。

动作必须以 PyTorch 张量形式提供,并且观察结果将以 PyTorch 张量形式返回。包装器的向量版本存在于 gymnasium.wrappers.vector.NumpyToTorch 中。

注意

对于 rendered,它将以 NumPy 数组形式返回,而不是 pytorch 张量。

示例

>>> import torch
>>> import gymnasium as gym
>>> env = gym.make("CartPole-v1")
>>> env = NumpyToTorch(env)
>>> obs, _ = env.reset(seed=123)
>>> type(obs)
<class 'torch.Tensor'>
>>> action = torch.tensor(env.action_space.sample())
>>> obs, reward, terminated, truncated, info = env.step(action)
>>> type(obs)
<class 'torch.Tensor'>
>>> type(reward)
<class 'float'>
>>> type(terminated)
<class 'bool'>
>>> type(truncated)
<class 'bool'>
更改日志
  • v1.0.0 - 首次添加

参数:
  • env – 要包装的基于 Jax 的环境

  • device – 应该将 torch 张量移动到的设备