IQLLoss¶
- class torchrl.objectives.IQLLoss(*args, **kwargs)[source]¶
TorchRL 实现的 IQL 损失函数。
出自论文 “Offline Reinforcement Learning with Implicit Q-Learning” https://arxiv.org/abs/2110.06169
- 参数:
actor_network (ProbabilisticActor) – 随机 actor
qvalue_network (TensorDictModule) –
Q(s, a) 参数模型。如果提供了一个 qvalue_network 实例,它将被复制
num_qvalue_nets
次。如果传入模块列表,它们的参数将被堆叠,除非它们共享相同的身份(在这种情况下,原始参数将被展开)。警告
当传入参数列表时,它们将__不会__与 policy 参数进行比较,并且所有参数将被视为未绑定。
value_network (TensorDictModule, 可选) – V(s) 参数模型。
- 关键字参数:
num_qvalue_nets (integer, 可选) – 使用的 Q 值网络数量。默认为
2
。loss_function (str, 可选) – 用于 value function loss 的损失函数。默认为 “smooth_l1”。
temperature (
float
, 可选) – 逆温度 (beta)。对于较小的超参数值,目标函数表现得类似于行为克隆,而对于较大的值,它试图恢复 Q 函数的最大值。expectile (
float
, 可选) – expectile \(\tau\)。 对于需要动态规划(“stichting”)的 antmaze 任务,较大的 \(\tau\) 值至关重要。priority_key (str, 可选) – [已弃用,请改用 .set_keys(priority_key=priority_key)] tensordict 中用于写入优先级(用于优先级回放缓冲区)的键。默认为 “td_error”。
separate_losses (bool, 可选) – 如果为
True
,policy 和 critic 之间的共享参数将仅通过 policy loss 进行训练。默认为False
,即梯度会传播到 policy 和 critic loss 的共享参数。reduction (str, 可选) – 指定应用于输出的 reduction 类型:
"none"
|"mean"
|"sum"
。"none"
:不应用 reduction。"mean"
:输出的总和将除以输出中的元素数量。"sum"
:输出将被求和。默认为"mean"
。
示例
>>> import torch >>> from torch import nn >>> from torchrl.data import Bounded >>> 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 = Bounded(-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。返回值是按以下顺序排列的 tensor 元组:["loss_actor", "loss_qvalue", "loss_value", "entropy"]
。示例
>>> import torch >>> from torch import nn >>> from torchrl.data import Bounded >>> 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 = Bounded(-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()
- default_keys¶
别名
_AcceptedKeys
- forward(tensordict: TensorDictBase = None) TensorDictBase [source]¶
它旨在读取输入的 TensorDict 并返回另一个包含名为“loss*”的损失键的 tensordict。
然后,训练器可以使用将损失拆分为其组成部分的功能,在整个训练过程中记录各种损失值。输出 tensordict 中存在的其他标量也将被记录。
- 参数:
tensordict – 包含计算损失所需值的输入 tensordict。
- 返回值:
一个新的不含批次维度的 tensordict,其中包含各种名为“loss*”的损失标量。损失必须以此名称返回,这一点至关重要,因为训练器会在反向传播之前读取它们。
- make_value_estimator(value_type: Optional[ValueEstimators] = None, **hyperparams)[source]¶
值函数构造器。
如果需要非默认值函数,必须使用此方法构建。
- 参数:
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)