Stack¶
- 类 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)[源代码]¶
堆叠张量(tensor)和 tensordict。
沿新维度连接一系列张量或 tensordict。
in_keys
下的 tensordict 或张量必须具有相同的形状。此转换仅将输入堆叠为一个输出 key。将多组输入 key 堆叠到不同的输出 key 需要多个转换。
此转换对于包含多个智能体、其 specs 在不同 key 下相同的情况非常有用。可以将智能体的 specs 和 tensordict 在一个共享 key 下堆叠在一起,以便运行多智能体强化学习 (MARL) 算法,这些算法需要观察、奖励等的张量包含所有智能体的批量数据。
- 参数:
in_keys (NestedKey 序列) – 要堆叠的 key。
out_key (NestedKey) – 结果堆叠条目的 key。
in_key_inv (NestedKey, 可选) – 在
inv()
调用期间用于取消堆叠的 key。默认为None
。out_keys_inv (NestedKey 序列, 可选) –
inv()
调用后结果取消堆叠条目的 key。默认为None
。dim (int, 可选) – 要插入的维度。默认为
-1
。allow_positive_dim (bool, 可选) – 如果为
True
,则接受正维度。默认为False
,即不允许非负维度。
- 关键字参数:
del_keys (bool, 可选) – 如果为
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(tensordict: TensorDictBase) TensorDictBase ¶
读取输入的 tensordict,并对选定的 key 应用转换。
- transform_done_spec(done_spec: TensorSpec) TensorSpec [源代码]¶
转换 done spec,使其结果 spec 与转换映射匹配。
- 参数:
done_spec (TensorSpec) – 转换前的 spec
- 返回:
转换后预期的 spec
- transform_input_spec(input_spec: TensorSpec) TensorSpec [源代码]¶
转换 input spec,使其结果 spec 与转换映射匹配。
- 参数:
input_spec (TensorSpec) – 转换前的 spec
- 返回:
转换后预期的 spec
- transform_observation_spec(observation_spec: TensorSpec) TensorSpec [源代码]¶
转换 observation spec,使其结果 spec 与转换映射匹配。
- 参数:
observation_spec (TensorSpec) – 转换前的 spec
- 返回:
转换后预期的 spec
- transform_reward_spec(reward_spec: TensorSpec) TensorSpec [源代码]¶
转换 reward spec,使其结果 spec 与转换映射匹配。
- 参数:
reward_spec (TensorSpec) – 转换前的 spec
- 返回:
转换后预期的 spec