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
- 返回:
转换后的预期规范
