VecNormV2¶
- class torchrl.envs.transforms.VecNormV2(in_keys: Sequence[NestedKey], out_keys: Sequence[NestedKey] | None = None, *, lock: mp.Lock = None, stateful: bool = True, decay: float = 0.9999, eps: float = 0.0001, shared_data: TensorDictBase | None = None, reduce_batch_dims: bool = False)[原始碼]¶
一個用於在強化學習環境中對向量化觀測和獎勵進行歸一化的類。
VecNormV2 可以以有狀態或無狀態模式執行。在有狀態模式下，它會維護內部統計資料（均值和方差）來歸一化輸入。在無狀態模式下，它需要提供外部統計資料來進行歸一化。
注意
此類旨在成為
VecNorm的幾乎即插即用替代品。它不應直接例項化，而應透過 new_api=True 關鍵字引數使用VecNorm變換來例項化。在 v0.10 中，VecNorm變換預設將切換到新 API。- 有狀態 vs. 無狀態
有狀態模式 (stateful=True)
維護用於歸一化的內部統計資料（loc、var、count）。
除非被凍結，否則會更新每個呼叫傳入的統計資料。
state_dict 返回當前的統計資料。
load_state_dict 使用提供的狀態更新內部統計資料。
無狀態模式 (stateful=False)
需要提供外部統計資料來進行歸一化。
不維護或更新內部統計資料。
state_dict 返回一個空字典。
load_state_dict 不影響內部狀態。
- 引數:
in_keys (Sequence[NestedKey]) – 要歸一化的資料的輸入鍵。
out_keys (Sequence[NestedKey] | None) – 歸一化資料的輸出鍵。如果未提供，則預設為 in_keys。
lock (mp.Lock, optional) – 用於執行緒安全的鎖。
stateful (bool, optional) – VecNorm 是否有狀態。此變換的無狀態版本需要資料在輸入/輸出 tensordicts 中傳遞。預設為 True。
decay (float, optional) – 用於更新統計資料的衰減率。預設為 0.9999。如果使用 decay=1，則歸一化統計資料具有無限記憶（每個項的權重相同）。較低的值更重視近期資料而非舊資料。
eps (float, optional) – 一個小值，用於防止除以零。預設為 1e-4。
shared_data (TensorDictBase | None, optional) – 用於初始化的共享資料。預設為 None。
reduce_batch_dims (bool, optional) – 如果為 True，則在更新統計資料之前對資料進行平均以減小批次維度。這在接收到批次的樣本時很有用，因為它允許移動平均值在整個批次上計算，而不是在單個元素上。請注意，此選項僅在有狀態模式（stateful=True）下受支援。預設為 False。
- 變數:
stateful (bool) – 指示 VecNormV2 是有狀態還是無狀態。
lock (mp.Lock) – 一個多程序鎖，用於在更新統計資料時確保執行緒安全。
decay (float) – 更新統計資料的衰減率。
eps (float) – 在歸一化期間防止除以零的小值。
frozen (bool) – 指示 VecNormV2 是否已凍結，防止更新統計資料。
_cast_int_to_float (bool) – 指示是否將整數輸入轉換為浮點數。
- loc()¶
返回用於歸一化的位置（均值）。
- scale()¶
返回用於歸一化的尺度（標準差）。
- standard_normal()¶
指示歸一化是否遵循標準正態分佈。
狀態字典行為
在有狀態模式下，state_dict 返回一個包含當前 loc、var 和 count 的字典。這些可用於在程序之間共享張量（此方法由
VecNorm自動呼叫，以在程序之間共享 VecNorm 狀態）。在無狀態模式下，state_dict 返回一個空字典，因為不維護任何內部狀態。
載入狀態字典行為
在有狀態模式下，load_state_dict 使用提供的狀態更新內部的 loc、var 和 count。
在無狀態模式下，load_state_dict 不修改任何內部狀態，因為沒有需要更新的內部狀態。
另請參閱
VecNorm是此變換的第一個版本。示例
>>> import torch >>> from torchrl.envs import EnvCreator, GymEnv, ParallelEnv, SerialEnv, VecNormV2 >>> >>> torch.manual_seed(0) >>> env = GymEnv("Pendulum-v1") >>> env_trsf = env.append_transform( >>> VecNormV2(in_keys=["observation", "reward"], out_keys=["observation_norm", "reward_norm"]) >>> ) >>> r = env_trsf.rollout(10) >>> print("Unnormalized rewards", r["next", "reward"]) Unnormalized rewards tensor([[ -1.7967], [ -2.1238], [ -2.5911], [ -3.5275], [ -4.8585], [ -6.5028], [ -8.2505], [-10.3169], [-12.1332], [-13.1235]]) >>> print("Normalized rewards", r["next", "reward_norm"]) Normalized rewards tensor([[-1.6596e-04], [-8.3072e-02], [-1.9170e-01], [-3.9255e-01], [-5.9131e-01], [-7.4671e-01], [-8.3760e-01], [-9.2058e-01], [-9.3484e-01], [-8.6185e-01]]) >>> # Aggregate values when using batched envs >>> env = SerialEnv(2, [lambda: GymEnv("Pendulum-v1")] * 2) >>> env_trsf = env.append_transform( >>> VecNormV2( >>> in_keys=["observation", "reward"], >>> out_keys=["observation_norm", "reward_norm"], >>> # Use reduce_batch_dims=True to aggregate values across batch elements >>> reduce_batch_dims=True, ) >>> ) >>> r = env_trsf.rollout(10) >>> print("Unnormalized rewards", r["next", "reward"]) Unnormalized rewards tensor([[[-0.1456], [-0.1862], [-0.2053], [-0.2605], [-0.4046], [-0.5185], [-0.8023], [-1.1364], [-1.6183], [-2.5406]],
- [[-0.0920],
[-0.1492], [-0.2702], [-0.3917], [-0.5001], [-0.7947], [-1.0160], [-1.3347], [-1.9082], [-2.9679]]])
>>> print("Normalized rewards", r["next", "reward_norm"]) Normalized rewards tensor([[[-0.2199], [-0.2918], [-0.1668], [-0.2083], [-0.4981], [-0.5046], [-0.7950], [-0.9791], [-1.1484], [-1.4182]],
- [[ 0.2201],
[-0.0403], [-0.5206], [-0.7791], [-0.8282], [-1.2306], [-1.2279], [-1.2907], [-1.4929], [-1.7793]]])
>>> print("Loc / scale", env_trsf.transform.loc["reward"], env_trsf.transform.scale["reward"]) Loc / scale tensor([-0.8626]) tensor([1.1832]) >>> >>> # Share values between workers >>> def make_env(): ... env = GymEnv("Pendulum-v1") ... env_trsf = env.append_transform( ... VecNormV2(in_keys=["observation", "reward"], out_keys=["observation_norm", "reward_norm"]) ... ) ... return env_trsf ... ... >>> if __name__ == "__main__": ... # EnvCreator will share the loc/scale vals ... make_env = EnvCreator(make_env) ... # Create a local env to track the loc/scale ... local_env = make_env() ... env = ParallelEnv(2, [make_env] * 2) ... r = env.rollout(10) ... # Non-zero loc and scale testify that the sub-envs share their summary stats with us ... print("Remotely updated loc / scale", local_env.transform.loc["reward"], local_env.transform.scale["reward"]) Remotely updated loc / scale tensor([-0.4307]) tensor([0.9613]) ... env.close()
- freeze() VecNormV2[原始碼]¶
凍結 VecNorm，避免在呼叫時更新統計資料。
參見
unfreeze()。
- get_extra_state() OrderedDict[原始碼]¶
返回要包含在模組 state_dict 中的任何額外狀態。
如果您的模組需要儲存額外狀態，請實現此函式以及相應的
set_extra_state()。此函式在構建模組的 state_dict() 時被呼叫。注意，為了保證 state_dict 的序列化工作正常，額外狀態應該是可被 pickle 的。我們僅為 Tensors 的序列化提供向後相容性保證；其他物件的序列化形式若發生變化，可能導致向後相容性中斷。
- 返回:
要儲存在模組 state_dict 中的任何額外狀態
- 返回型別:
物件
- property loc¶
返回一個包含用於仿射變換的 loc 的 TensorDict。
- property scale¶
返回一個包含用於仿射變換的 scale 的 TensorDict。
- set_extra_state(state: OrderedDict) None[原始碼]¶
設定載入的 state_dict 中包含的額外狀態。
此函式從
load_state_dict()呼叫，以處理 state_dict 中找到的任何額外狀態。如果您的模組需要在其 state_dict 中儲存額外狀態，請實現此函式和相應的get_extra_state()。- 引數:
state (dict) – 來自 state_dict 的額外狀態
- property standard_normal¶
由 loc 和 scale 給出的仿射變換是否遵循標準正態方程。
類似於
ObservationNorm的 standard_normal 屬性。始終返回
True。
- transform_observation_spec(observation_spec: Composite) Composite[原始碼]¶
轉換觀察規範，使結果規範與轉換對映匹配。
- 引數:
observation_spec (TensorSpec) – 轉換前的規範
- 返回:
轉換後的預期規範
- transform_output_spec(output_spec: Composite) Composite[原始碼]¶
轉換輸出規範，使結果規範與轉換對映匹配。
此方法通常應保持不變。更改應透過
transform_observation_spec()、transform_reward_spec()和transform_full_done_spec()實現。 :param output_spec: 變換前的規範 :type output_spec: TensorSpec- 返回:
轉換後的預期規範
- transform_reward_spec(reward_spec: Composite, observation_spec) Composite[原始碼]¶
轉換獎勵的 spec，使其與變換對映匹配。
- 引數:
reward_spec (TensorSpec) – 變換前的 spec
- 返回:
轉換後的預期規範
