SafeSequential

class torchrl.modules.tensordict_module.SafeSequential(*args, **kwargs)

一個安全的 TensorDictModule 序列。

與 nn.Sequence 類似，後者透過讀取和寫入單個 tensor 的一系列對映來傳遞 tensor，此模組將透過查詢每個輸入模組來讀取和寫入 tensordict。當使用函式式模組呼叫 TensorDictSequential 例項時，引數列表（和 buffer）預計會被連線成一個列表。

引數:

• modules (TensorDictModule 可迭代物件) – 按順序執行的 TensorDictModule 例項的有序序列。

• partial_tolerant (布林值, 可選) – 如果為 True，則輸入 tensordict 可能缺少某些輸入鍵。在這種情況下，只有根據現有鍵能夠執行的模組才會被執行。此外，如果輸入 tensordict 是一個 tensordict 的惰性堆疊（lazy stack），並且 partial_tolerant 為 True，並且該堆疊不包含所需的鍵，那麼 SafeSequential 將掃描子 tensordict 以查詢包含所需鍵的 tensordict（如果存在）。

TensorDictSequence 支援函式式、模組化和 vmap 編碼：.. 標題:: 示例

>>> import torch
>>> from tensordict import TensorDict
>>> from torchrl.data import Composite, Unbounded
>>> from torchrl.modules import TanhNormal, SafeSequential, TensorDictModule, NormalParamExtractor
>>> from torchrl.modules.tensordict_module import SafeProbabilisticModule
>>> td = TensorDict({"input": torch.randn(3, 4)}, [3,])
>>> spec1 = Composite(hidden=Unbounded(4), loc=None, scale=None)
>>> net1 = nn.Sequential(torch.nn.Linear(4, 8), NormalParamExtractor())
>>> module1 = TensorDictModule(net1, in_keys=["input"], out_keys=["loc", "scale"])
>>> td_module1 = SafeProbabilisticModule(
...     module=module1,
...     spec=spec1,
...     in_keys=["loc", "scale"],
...     out_keys=["hidden"],
...     distribution_class=TanhNormal,
...     return_log_prob=True,
... )
>>> spec2 = Unbounded(8)
>>> module2 = torch.nn.Linear(4, 8)
>>> td_module2 = TensorDictModule(
...    module=module2,
...    spec=spec2,
...    in_keys=["hidden"],
...    out_keys=["output"],
...    )
>>> td_module = SafeSequential(td_module1, td_module2)
>>> params = TensorDict.from_module(td_module)
>>> with params.to_module(td_module):
...     td_module(td)
>>> print(td)
TensorDict(
    fields={
        hidden: Tensor(torch.Size([3, 4]), dtype=torch.float32),
        input: Tensor(torch.Size([3, 4]), dtype=torch.float32),
        loc: Tensor(torch.Size([3, 4]), dtype=torch.float32),
        output: Tensor(torch.Size([3, 8]), dtype=torch.float32),
        sample_log_prob: Tensor(torch.Size([3, 1]), dtype=torch.float32),
        scale: Tensor(torch.Size([3, 4]), dtype=torch.float32)},
    batch_size=torch.Size([3]),
    device=None,
    is_shared=False)
>>> # The module spec aggregates all the input specs:
>>> print(td_module.spec)
Composite(
    hidden: UnboundedContinuous(
        shape=torch.Size([4]), space=None, device=cpu, dtype=torch.float32, domain=continuous),
    loc: None,
    scale: None,
    output: UnboundedContinuous(
        shape=torch.Size([8]), space=None, device=cpu, dtype=torch.float32, domain=continuous))

在 vmap 情況下

>>> from torch import vmap
>>> params = params.expand(4, *params.shape)
>>> td_vmap = vmap(td_module, (None, 0))(td, params)
>>> print(td_vmap)
TensorDict(
    fields={
        hidden: Tensor(torch.Size([4, 3, 4]), dtype=torch.float32),
        input: Tensor(torch.Size([4, 3, 4]), dtype=torch.float32),
        loc: Tensor(torch.Size([4, 3, 4]), dtype=torch.float32),
        output: Tensor(torch.Size([4, 3, 8]), dtype=torch.float32),
        sample_log_prob: Tensor(torch.Size([4, 3, 1]), dtype=torch.float32),
        scale: Tensor(torch.Size([4, 3, 4]), dtype=torch.float32)},
    batch_size=torch.Size([4, 3]),
    device=None,
    is_shared=False)