杂项 Wrappers¶

通用 Wrappers¶

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 确定 episode 步骤的示例

>>> 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 – episode 在此环境步数之后被截断 (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]¶

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

通常，你只想间歇性地录制 episodes，例如每百个 episode 或每千个环境步。为此，你可以指定 episode_trigger 或 step_trigger。它们应该是返回布尔值的函数，指示是否应在当前 episode 或步骤开始录制。

episode_trigger 应在应开始录制的 episode 时返回 True。 step_trigger 应在应开始录制的第 n 个环境步骤时返回 True，其中 n 对所有先前的 episodes 求和。如果未传递 episode_trigger 和 step_trigger，则将采用默认的 episode_trigger，即 capped_cubic_video_schedule()。此函数在每个为 3 的幂次的 episode 开始视频，直到 1000，然后每 1000 个 episodes 开始一个视频。默认情况下，一旦调用 reset，录制将停止。但是，你也可以通过为 video_length 传递严格的正值来创建固定长度的录制（可能跨越多个 episodes）。

不存在包装器的向量版本。

示例 - 运行环境 50 个 episodes，并从第 0 个 episode 开始每 10 个 episodes 保存视频

>>> 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 个 episodes，每 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 个 episodes，录制所有内容，但以 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 的函数，当且仅当应在此 episode 开始录制时
step_trigger – 接受整数并返回 True 的函数，当且仅当应在此步骤开始录制时
video_length (int) – 录制的 episodes 的长度。如果为 0，则录制整个 episodes。否则，捕获指定长度的片段
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 长度。

在一个 episode 结束时，episode 的统计信息将使用键 episode 添加到 info 中。如果使用向量化环境，则还会使用键 _episode，该键指示各个索引处的 env 是否具有 episode 统计信息。存在包装器的向量版本，gymnasium.wrappers.vector.RecordEpisodeStatistics。

在一个 episode 完成后，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>"
... }

此外，最近的奖励和 episode 长度存储在缓冲区中，可以通过 wrapped_env.return_queue 和 wrapped_env.length_queue 分别访问。

变量:

time_queue (*) – 最近 deque_size 个 episodes 的时间长度
return_queue (*) – 最近 deque_size 个 episodes 的累积奖励
length_queue (*) – 最近 deque_size 个 episodes 的长度

变更日志

v0.15.4 - 首次添加
v1.0.0 - 删除了向量环境支持（参见 gymnasium.wrappers.vector.RecordEpisodeStatistics）并添加了属性 time_queue

参数:

env (Env) – 要应用包装器的环境
buffer_length – 缓冲区 return_queue、length_queue 和 time_queue 的大小
stats_key – episode 统计信息的 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) 的指南，“重新审视 Arcade Learning Environment：通用智能体的评估协议和开放问题”。

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

Noop 重置：通过在重置时执行随机数量的 no-ops 来获得初始状态，默认最大 30 个 no-ops。
帧跳过：步骤之间跳过的帧数，默认为 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) – 对于 No-op 重置，重置时采取的最大 no-ops 操作数，要关闭，请设置为 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 Wrapper 的内存优化优势。

Raises:

DependencyNotInstalled – opencv-python 包未安装
ValueError – 在原始 env 中禁用帧跳过

不常用 Wrappers¶

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

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

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

变更日志

v0.24.0 - 最初添加为 AutoResetWrapper
v1.0.0 - 重命名为 Autoreset，自动重置顺序更改为在环境终止或截断后的步骤中重置。因此，“final_observation” 和 “final_info” 被删除。

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

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

一个被动包装器，围绕 step、reset 和 render 函数，以检查它们是否遵循 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 - Bugs 已全部修复
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。仅当环境未原生实现 human-rendering 时（即 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 之前调用 step 或 render，将产生错误。

不存在包装器的向量版本。

示例

>>> 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 - 为 render 函数添加了顺序强制执行

参数:

env – 要包装的环境
disable_render_order_enforcing – 是否禁用 render 顺序强制执行

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

返回 episode 运行步骤数的帧列表。 >>> 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

收集所有 episodes 的所有帧，在调用 render 时不清除它们 >>> 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) – 如果为 true，则在调用 meth:render 后清除收集帧。默认值为 True。
reset_clean (bool) – 如果为 true，则在调用 meth:reset 时清除收集帧。默认值为 True。

数据转换 Wrappers¶

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 Tensors 交互。

动作必须作为 PyTorch Tensors 提供，观测将作为 PyTorch Tensors 返回。存在包装器的向量版本，gymnasium.wrappers.vector.JaxToTorch。

注意

对于 rendered，这作为 NumPy 数组而不是 pytorch Tensor 返回。

示例

>>> 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 Tensor 返回。

示例

>>> 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 – 要包装的基于 NumPy 的环境
device – Torch 张量应移动到的设备