IQLLoss¶
- class torchrl.objectives.IQLLoss(*args, **kwargs)[源代码]¶
TorchRL 中 IQL 损失的实现。
在“使用隐式 Q 学习的离线强化学习”中提出 https://arxiv.org/abs/2110.06169
- 参数:
actor_network (ProbabilisticActor) – 随机 actor
qvalue_network (TensorDictModule) –
Q(s, a) 参数模型 如果提供 qvalue_network 的单个实例,它将被复制
num_qvalue_nets
次。如果传递模块列表,则它们的权重将被堆叠,除非它们共享相同的标识(在这种情况下,原始权重将被扩展)。警告
当传递参数列表时,它**不会**与策略参数进行比较,并且所有参数都将被视为未绑定。
value_network (TensorDictModule, 可选) – V(s) 参数模型。
- 关键字参数:
num_qvalue_nets (整数, 可选) – 使用的 Q 值网络数量。默认为
2
。loss_function (str, 可选) – 用于值函数损失的损失函数。默认为 “smooth_l1”。
temperature (float, 可选) – 逆温度 (beta)。对于较小的超参数值,目标的行为类似于行为克隆,而对于较大的值,它试图恢复 Q 函数的最大值。
expectile (float, 可选) – 期望值 \(\tau\)。较大的 \(\tau\) 值对于需要动态规划(“建立”)的 antmaze 任务至关重要。
priority_key (str, 可选) – [已弃用,请改用 .set_keys(priority_key=priority_key) ] tensordict 密钥,用于写入优先级(用于优先回放缓冲区使用)。默认为 “td_error”。
separate_losses (bool, 可选) – 如果
True
,则策略和评论家之间共享的参数将仅在策略损失上进行训练。默认为False
,即梯度对策略和评论家损失的共享参数进行反向传播。reduction (str, 可选) – 指定应用于输出的缩减:
"none"
|"mean"
|"sum"
。"none"
:不应用缩减,"mean"
:输出的总和将除以输出中的元素数量,"sum"
:输出将求和。默认值:"mean"
。
示例
>>> import torch >>> from torch import nn >>> from torchrl.data import BoundedTensorSpec >>> from torchrl.modules.distributions import NormalParamExtractor, TanhNormal >>> from torchrl.modules.tensordict_module.actors import ProbabilisticActor, ValueOperator >>> from torchrl.modules.tensordict_module.common import SafeModule >>> from torchrl.objectives.iql import IQLLoss >>> from tensordict import TensorDict >>> n_act, n_obs = 4, 3 >>> spec = BoundedTensorSpec(-torch.ones(n_act), torch.ones(n_act), (n_act,)) >>> net = nn.Sequential(nn.Linear(n_obs, 2 * n_act), NormalParamExtractor()) >>> module = SafeModule(net, in_keys=["observation"], out_keys=["loc", "scale"]) >>> actor = ProbabilisticActor( ... module=module, ... in_keys=["loc", "scale"], ... spec=spec, ... distribution_class=TanhNormal) >>> class QValueClass(nn.Module): ... def __init__(self): ... super().__init__() ... self.linear = nn.Linear(n_obs + n_act, 1) ... def forward(self, obs, act): ... return self.linear(torch.cat([obs, act], -1)) >>> qvalue = SafeModule( ... QValueClass(), ... in_keys=["observation", "action"], ... out_keys=["state_action_value"], ... ) >>> value = SafeModule( ... nn.Linear(n_obs, 1), ... in_keys=["observation"], ... out_keys=["state_value"], ... ) >>> loss = IQLLoss(actor, qvalue, value) >>> batch = [2, ] >>> action = spec.rand(batch) >>> data = TensorDict({ ... "observation": torch.randn(*batch, n_obs), ... "action": action, ... ("next", "done"): torch.zeros(*batch, 1, dtype=torch.bool), ... ("next", "terminated"): torch.zeros(*batch, 1, dtype=torch.bool), ... ("next", "reward"): torch.randn(*batch, 1), ... ("next", "observation"): torch.randn(*batch, n_obs), ... }, batch) >>> loss(data) TensorDict( fields={ entropy: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False), loss_actor: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False), loss_qvalue: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False), loss_value: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False)}, batch_size=torch.Size([]), device=None, is_shared=False)
此类也与非 tensordict 基于的模块兼容,并且无需使用任何与 tensordict 相关的原语即可使用。在这种情况下,期望的关键字参数为:
["action", "next_reward", "next_done", "next_terminated"]
+ actor、value 和 qvalue 网络的 in_keys 返回值为按以下顺序排列的张量元组:["loss_actor", "loss_qvalue", "loss_value", "entropy"]
。示例
>>> import torch >>> from torch import nn >>> from torchrl.data import BoundedTensorSpec >>> from torchrl.modules.distributions import NormalParamExtractor, TanhNormal >>> from torchrl.modules.tensordict_module.actors import ProbabilisticActor, ValueOperator >>> from torchrl.modules.tensordict_module.common import SafeModule >>> from torchrl.objectives.iql import IQLLoss >>> _ = torch.manual_seed(42) >>> n_act, n_obs = 4, 3 >>> spec = BoundedTensorSpec(-torch.ones(n_act), torch.ones(n_act), (n_act,)) >>> net = nn.Sequential(nn.Linear(n_obs, 2 * n_act), NormalParamExtractor()) >>> module = SafeModule(net, in_keys=["observation"], out_keys=["loc", "scale"]) >>> actor = ProbabilisticActor( ... module=module, ... in_keys=["loc", "scale"], ... spec=spec, ... distribution_class=TanhNormal) >>> class QValueClass(nn.Module): ... def __init__(self): ... super().__init__() ... self.linear = nn.Linear(n_obs + n_act, 1) ... def forward(self, obs, act): ... return self.linear(torch.cat([obs, act], -1)) >>> qvalue = SafeModule( ... QValueClass(), ... in_keys=["observation", "action"], ... out_keys=["state_action_value"], ... ) >>> value = SafeModule( ... nn.Linear(n_obs, 1), ... in_keys=["observation"], ... out_keys=["state_value"], ... ) >>> loss = IQLLoss(actor, qvalue, value) >>> batch = [2, ] >>> action = spec.rand(batch) >>> loss_actor, loss_qvalue, loss_value, entropy = loss( ... observation=torch.randn(*batch, n_obs), ... action=action, ... next_done=torch.zeros(*batch, 1, dtype=torch.bool), ... next_terminated=torch.zeros(*batch, 1, dtype=torch.bool), ... next_observation=torch.zeros(*batch, n_obs), ... next_reward=torch.randn(*batch, 1)) >>> loss_actor.backward()
还可以使用
IQLLoss.select_out_keys()
方法过滤输出密钥。示例
>>> _ = loss.select_out_keys('loss_actor', 'loss_qvalue') >>> loss_actor, loss_qvalue = loss( ... observation=torch.randn(*batch, n_obs), ... action=action, ... next_done=torch.zeros(*batch, 1, dtype=torch.bool), ... next_terminated=torch.zeros(*batch, 1, dtype=torch.bool), ... next_observation=torch.zeros(*batch, n_obs), ... next_reward=torch.randn(*batch, 1)) >>> loss_actor.backward()
- forward(tensordict: TensorDictBase) TensorDictBase [源代码]¶
旨在读取输入 TensorDict 并返回另一个包含名为“loss*”的损失密钥的 tensordict。
然后,将损失分成其组成部分可以由训练器在整个训练过程中记录各种损失值。输出 tensordict 中的其他标量也将被记录。
- 参数:
tensordict – 一个包含计算损失所需值的输入 tensordict。
- 返回:
一个新的没有批次维度的 tensordict,其中包含各种损失标量,这些标量将被命名为“loss*”。这些损失必须使用此名称返回,因为它们将在反向传播之前由训练器读取。
- make_value_estimator(value_type: Optional[ValueEstimators] = None, **hyperparams)[源代码]¶
值函数构造函数。
如果需要非默认值函数,则必须使用此方法构建它。
- 参数:
value_type (ValueEstimators) – 一个
ValueEstimators
枚举类型,指示要使用的值函数。如果没有提供,则将使用default_value_estimator
属性中存储的默认值。生成的估值器类将在self.value_type
中注册,允许进行未来的改进。**hyperparams** – 用于值函数的超参数。如果未提供,则将使用
default_value_kwargs()
指示的值。
示例
>>> from torchrl.objectives import DQNLoss >>> # initialize the DQN loss >>> actor = torch.nn.Linear(3, 4) >>> dqn_loss = DQNLoss(actor, action_space="one-hot") >>> # updating the parameters of the default value estimator >>> dqn_loss.make_value_estimator(gamma=0.9) >>> dqn_loss.make_value_estimator( ... ValueEstimators.TD1, ... gamma=0.9) >>> # if we want to change the gamma value >>> dqn_loss.make_value_estimator(dqn_loss.value_type, gamma=0.9)