雜項封裝器

常用封裝器

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 張量應移動到的裝置