DecisionTransformerInferenceWrapper

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

用於決策Transformer的推理動作包裝器。

專門為決策Transformer設計的包裝器，它將掩蓋輸入 tensordict 序列到推理上下文。輸出將是一個 TensorDict，其鍵與輸入相同，但僅包含預測動作序列中的最後一個動作和最後一個回報目標。

此模組建立並返回 tensordict 的修改副本，即它**不會**就地修改 tensordict。

注意

如果動作、觀察或回報目標鍵不是標準的，應使用方法 set_tensor_keys()，例如：

>>> dt_inference_wrapper.set_tensor_keys(action="foo", observation="bar", return_to_go="baz")

in_keys 是觀察、動作和回報目標鍵。out-keys 與 in-keys 匹配，並額外包含來自策略的任何其他輸出鍵（例如，分佈引數或隱藏值）。

引數:

policy (TensorDictModule) – 接收觀察併產生動作值的策略模組

關鍵字引數:

• inference_context (int) – 上下文中不會被掩蓋的先前動作的數量。例如，對於形狀為 [batch_size, context, obs_dim]，其中 context=20 且 inference_context=5 的觀察輸入，上下文的前 15 個條目將被掩蓋。預設為 5。

• spec (Optional[TensorSpec], optional) – 輸入 TensorDict 的規範。如果為 None，將從策略模組推斷。

• device (torch.device, optional) – 如果提供，指定緩衝區/specs 將被放置的裝置。

示例

>>> import torch
>>> from tensordict import TensorDict
>>> from tensordict.nn import TensorDictModule
>>> from torchrl.modules import (
...      ProbabilisticActor,
...      TanhDelta,
...      DTActor,
...      DecisionTransformerInferenceWrapper,
...  )
>>> dtactor = DTActor(state_dim=4, action_dim=2,
...             transformer_config=DTActor.default_config()
... )
>>> actor_module = TensorDictModule(
...         dtactor,
...         in_keys=["observation", "action", "return_to_go"],
...         out_keys=["param"])
>>> dist_class = TanhDelta
>>> dist_kwargs = {
...     "low": -1.0,
...     "high": 1.0,
... }
>>> actor = ProbabilisticActor(
...     in_keys=["param"],
...     out_keys=["action"],
...     module=actor_module,
...     distribution_class=dist_class,
...     distribution_kwargs=dist_kwargs)
>>> inference_actor = DecisionTransformerInferenceWrapper(actor)
>>> sequence_length = 20
>>> td = TensorDict({"observation": torch.randn(1, sequence_length, 4),
...                 "action": torch.randn(1, sequence_length, 2),
...                 "return_to_go": torch.randn(1, sequence_length, 1)}, [1,])
>>> result = inference_actor(td)
>>> print(result)
TensorDict(
    fields={
        action: Tensor(shape=torch.Size([1, 2]), device=cpu, dtype=torch.float32, is_shared=False),
        observation: Tensor(shape=torch.Size([1, 20, 4]), device=cpu, dtype=torch.float32, is_shared=False),
        param: Tensor(shape=torch.Size([1, 20, 2]), device=cpu, dtype=torch.float32, is_shared=False),
        return_to_go: Tensor(shape=torch.Size([1, 1]), device=cpu, dtype=torch.float32, is_shared=False)},
    batch_size=torch.Size([1]),
    device=None,
    is_shared=False)

forward(tensordict: TensorDictBase = None) → TensorDictBase

定義每次呼叫時執行的計算。

應被所有子類覆蓋。

注意

儘管需要在本函式中定義前向傳播的實現，但應在之後呼叫 Module 例項而不是直接呼叫此函式，因為前者會處理註冊的鉤子，而後者會靜默地忽略它們。

mask_context(tensordict: TensorDictBase) → TensorDictBase

掩蓋輸入序列的上下文。

set_tensor_keys(**kwargs)

設定模組的輸入鍵。

關鍵字引數:

• observation (NestedKey, optional) – 觀察鍵。

• action (NestedKey, optional) – 動作鍵（網路的輸入）。

• return_to_go (NestedKey, optional) – 回報目標鍵。

• out_action (NestedKey, optional) – 動作鍵（網路的輸出）。