DecisionTransformerInferenceWrapper

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

Decision Transformer 的推理动作包装器。

一个专门为 Decision 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，并添加策略中的任何其他 out_key（例如，分布的参数或隐藏值）。

参数:

policy (TensorDictModule) – 接收观测并产生动作值的策略模块

关键字参数:

• inference_context (int) – 将在上下文中不被屏蔽的先前动作的数量。例如，对于形状为 [batch_size, context, obs_dim] 且 context=20 和 inference_context=5 的观测输入，上下文的前 15 个条目将被屏蔽。默认为 5。

• spec (Optional[TensorSpec]) – 输入 TensorDict 的规格。如果为 None，将从策略模块推断。

• device (torch.device, optional) – 如果提供，则指定缓冲区/规格放置的设备。

示例

>>> 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) – 动作键（从网络输出）。