杂项封装器

常用封装器

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 - 随着步进 API 的更新,终止和截断信号是分开返回的。

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

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

class gymnasium.wrappers.RecordVideo(env: 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, gc_trigger: Callable[[int], bool] | None = lambda episode: ...)[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 日志记录器,默认为禁用

  • gc_trigger – 接受整数并返回 True(当且仅当应在此剧集之后执行垃圾回收时)的函数

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 – 剧集统计信息的信息键

class gymnasium.wrappers.AtariPreprocessing(env: 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 环境

  • 空操作重置 (Noop Reset):通过在重置时执行随机次数的空操作来获取初始状态,默认最多 30 次空操作。

  • 帧跳过 (Frame skipping):步进之间跳过的帧数,默认为 4。

  • 最大池化 (Max-pooling):对帧跳过中最近的两个观测进行池化。

  • 生命损失时的终止信号:当智能体在环境中失去一条生命时,环境终止。

    默认关闭。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) – 如果为 Truegrayscale_obs=True,则在灰度观测中添加一个通道轴,使其变为 3 维。

  • scale_obs (bool) – 如果为 True,则返回归一化到 [0,1) 范围内的观测。它还会限制 FrameStack Wrapper 的内存优化优势。

引发:
  • 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。

此封装器在创建时自动应用,可以通过 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) – 如果为 true,则在调用 meth:render 后清除收集的帧。默认值为 True

  • reset_clean (bool) – 如果为 true,则在调用 meth:reset 时清除收集的帧。默认值为 True

数据转换封装器

class gymnasium.wrappers.ArrayConversion(env: Env, env_xp: ModuleType, target_xp: ModuleType, env_device: Any | None = None, target_device: Any | None = None)[source]

封装一个与 Array API 兼容的环境,使其可以与另一个 Array API 框架进行交互。

流行的 Array API 框架包括 numpytorchjax.numpycupy 等。使用此封装器,您可以将环境的输出转换为其中任何一个框架。反之,如果可能且无需移动数据或设备传输,动作将自动映射回环境框架。

存在该封装器的向量版本,gymnasium.wrappers.vector.ArrayConversion

示例

>>> import torch                                                
>>> import jax.numpy as jnp                                     
>>> import gymnasium as gym                                     
>>> env = gym.make("JaxEnv-vx")                                 
>>> env = ArrayConversion(env, env_xp=jnp, target_xp=torch)     
>>> 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.2.0 - 初次添加

参数:
  • env – 要封装的 Array API 兼容环境

  • env_xp – 环境所基于的 Array API 框架

  • target_xp – 要转换到的 Array API 框架

  • env_device – 环境所在的设备

  • target_device – 数组应返回的设备

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: Env, device: str | 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 – PyTorch 张量应移动到的设备

class gymnasium.wrappers.NumpyToTorch(env: Env, device: str | 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 – 要封装的基于 NumPy 的环境

  • device – PyTorch 张量应移动到的设备