Stack

class torchrl.envs.transforms.Stack(in_keys: Sequence[NestedKey], out_key: NestedKey, in_key_inv: NestedKey | None = None, out_keys_inv: Sequence[NestedKey] | None = None, dim: int = - 1, allow_positive_dim: bool = False, *, del_keys: bool = True)

堆叠张量和 tensordicts。

沿新维度连接一系列张量或 tensordicts。 in_keys 下的 tensordicts 或张量必须具有相同的形状。

此转换仅将输入堆叠到一个输出键中。将多个输入键组堆叠到不同的输出键需要多个转换。

对于在不同键下具有相同规范的多个代理的环境，此转换很有用。代理的规范和 tensordicts 可以堆叠在共享键下，以便运行期望观测值、奖励等张量包含所有代理的批量数据的 MARL 算法。

参数:

• in_keys (Sequence[NestedKey]) – 要堆叠的键。

• out_key (NestedKey) – 结果堆叠条目的键。

• in_key_inv (NestedKey, optional) – 在调用 inv() 时要解堆叠的键。默认为 None。

• out_keys_inv (Sequence[NestedKey], optional) – 在调用 inv() 后结果解堆叠条目的键。默认为 None。

• dim (int, optional) – 要插入的维度。默认为 -1。

• allow_positive_dim (bool, optional) – 如果为 True，则接受正维度。默认为 False，即不允许非负维度。

关键字参数:

del_keys (bool, optional) – 如果为 True，则输入值将在堆叠后被删除。默认为 True。

示例

>>> import torch
>>> from tensordict import TensorDict
>>> from torchrl.envs import Stack
>>> td = TensorDict({"key1": torch.zeros(3), "key2": torch.ones(3)}, [])
>>> td
TensorDict(
    fields={
        key1: Tensor(shape=torch.Size([3]), device=cpu, dtype=torch.float32, is_shared=False),
        key2: Tensor(shape=torch.Size([3]), device=cpu, dtype=torch.float32, is_shared=False)},
    batch_size=torch.Size([]),
    device=None,
    is_shared=False)
>>> transform = Stack(in_keys=["key1", "key2"], out_key="out", dim=-2)
>>> transform(td)
TensorDict(
    fields={
        out: Tensor(shape=torch.Size([2, 3]), device=cpu, dtype=torch.float32, is_shared=False)},
    batch_size=torch.Size([]),
    device=None,
    is_shared=False)
>>> td["out"]
tensor([[0., 0., 0.],
        [1., 1., 1.]])

>>> agent_0 = TensorDict({"obs": torch.rand(4, 5), "reward": torch.zeros(1)})
>>> agent_1 = TensorDict({"obs": torch.rand(4, 5), "reward": torch.zeros(1)})
>>> td = TensorDict({"agent_0": agent_0, "agent_1": agent_1})
>>> transform = Stack(in_keys=["agent_0", "agent_1"], out_key="agents")
>>> transform(td)
TensorDict(
    fields={
        agents: TensorDict(
            fields={
                obs: Tensor(shape=torch.Size([2, 4, 5]), device=cpu, dtype=torch.float32, is_shared=False),
                reward: Tensor(shape=torch.Size([2, 1]), device=cpu, dtype=torch.float32, is_shared=False)},
            batch_size=torch.Size([2]),
            device=None,
            is_shared=False)},
    batch_size=torch.Size([]),
    device=None,
    is_shared=False)

forward(next_tensordict: TensorDictBase) → TensorDictBase

读取输入 tensordict，并对选定的键应用转换。

默认情况下，此方法

• 直接调用 _apply_transform()。

• 不调用 _step() 或 _call()。

此方法不会在 env.step 的任何点被调用。但是，它在 sample() 中被调用。

注意

forward 还使用 dispatch 与常规关键字参数配合使用，以将参数名称强制转换为键。

示例

>>> class TransformThatMeasuresBytes(Transform):
...     '''Measures the number of bytes in the tensordict, and writes it under `"bytes"`.'''
...     def __init__(self):
...         super().__init__(in_keys=[], out_keys=["bytes"])
...
...     def forward(self, tensordict: TensorDictBase) -> TensorDictBase:
...         bytes_in_td = tensordict.bytes()
...         tensordict["bytes"] = bytes
...         return tensordict
>>> t = TransformThatMeasuresBytes()
>>> env = env.append_transform(t) # works within envs
>>> t(TensorDict(a=0))  # Works offline too.

transform_done_spec(done_spec: TensorSpec) → TensorSpec

变换 done spec，使结果 spec 与变换映射匹配。

参数:

done_spec (TensorSpec) – 变换前的 spec

返回:

转换后的预期规范

transform_input_spec(input_spec: TensorSpec) → TensorSpec

转换输入规范，使结果规范与转换映射匹配。

参数:

input_spec (TensorSpec) – 转换前的规范

返回:

转换后的预期规范

transform_observation_spec(observation_spec: TensorSpec) → TensorSpec

转换观察规范，使结果规范与转换映射匹配。

参数:

observation_spec (TensorSpec) – 转换前的规范

返回:

转换后的预期规范

transform_reward_spec(reward_spec: TensorSpec) → TensorSpec

转换奖励的 spec，使其与变换映射匹配。

参数:

reward_spec (TensorSpec) – 变换前的 spec

返回:

转换后的预期规范