A2CLoss¶
- class torchrl.objectives.A2CLoss(*args, **kwargs)[源代码]¶
TorchRL 中 A2C 损失的实现。
A2C(优势演员评论)是一种无模型在线 RL 算法,它使用 n 步的并行展开来更新策略,依靠 REINFORCE 估计器来计算梯度。它还在目标函数中添加了一个熵项以改进探索。
有关 A2C 的更多详细信息,请参阅:“深度强化学习的异步方法”,https://arxiv.org/abs/1602.01783v2
- 参数:
actor_network (ProbabilisticTensorDictSequential) – 策略算子。
critic_network (ValueOperator) – 值算子。
entropy_bonus (bool) – 如果为
True
,则会将熵奖励添加到损失中以偏好探索性策略。samples_mc_entropy (int) – 如果从策略算子检索到的分布没有熵的封闭形式公式,则将使用蒙特卡罗估计。
samples_mc_entropy
将控制用于计算此估计值的样本数。默认为1
。entropy_coef (float) – 熵损失的权重。
critic_coef (float) – 评论家损失的权重。
loss_critic_type (str) – 值差异的损失函数。可以是“l1”、“l2”或“smooth_l1”之一。默认为
"smooth_l1"
。separate_losses (bool, optional) – 如果为
True
,则策略和评论家之间共享的参数将仅在策略损失上进行训练。默认为False
,即梯度会针对策略和评论家损失传播到共享参数。advantage_key (str) – [已弃用,请改用 set_keys(advantage_key=advantage_key) ] 预期写入优势的输入 tensordict 密钥。默认值:“advantage”
value_target_key (str) – [已弃用,请改用 set_keys() ] 预期写入目标状态值的输入 tensordict 密钥。默认为
"value_target"
。functional (bool, optional) – 模块是否应功能化。功能化允许元 RL 等功能,但会使使用分布式模型 (DDP、FSDP 等) 变得不可能,并且会带来一些成本。默认为
True
。reduction (str, optional) – 指定要应用于输出的约简:
"none"
|"mean"
|"sum"
。"none"
:不应用约简,"mean"
:输出的总和将除以输出中的元素数,"sum"
:输出将求和。默认值:"mean"
。clip_value (float, optional) – 如果提供,它将用于计算相对于输入值估计值的价值预测的剪辑版本,并使用它来计算价值损失。剪辑的目的是限制极端价值预测的影响,帮助稳定训练并防止出现大型更新。但是,如果价值估计是由当前版本的价值估计器完成的,它将不会有任何影响。默认为
None
。
示例
>>> 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.a2c import A2CLoss >>> 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) >>> module = nn.Linear(n_obs, 1) >>> value = ValueOperator( ... module=module, ... in_keys=["observation"]) >>> loss = A2CLoss(actor, value, loss_critic_type="l2") >>> 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_critic: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False), loss_entropy: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False), loss_objective: 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"]
+ 演员和评论家的 in_keys。返回值是一个张量元组,其顺序如下:["loss_objective"]
+ 如果 critic_coef 不为 None 则为["loss_critic"]
+ 如果 entropy_bonus 为 True 且 critic_coef 不为 None 则为["entropy", "loss_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.a2c import A2CLoss >>> _ = 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) >>> module = nn.Linear(n_obs, 1) >>> value = ValueOperator( ... module=module, ... in_keys=["observation"]) >>> loss = A2CLoss(actor, value, loss_critic_type="l2") >>> batch = [2, ] >>> loss_obj, loss_critic, entropy, loss_entropy = loss( ... observation = torch.randn(*batch, n_obs), ... action = spec.rand(batch), ... 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)) >>> loss_obj.backward()
还可以使用
SACLoss.select_out_keys()
方法过滤输出密钥。示例
>>> loss.select_out_keys('loss_objective', 'loss_critic') >>> loss_obj, loss_critic = loss( ... observation = torch.randn(*batch, n_obs), ... action = spec.rand(batch), ... 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)) >>> loss_obj.backward()
- forward(tensordict: TensorDictBase) TensorDictBase [源代码]¶
它旨在读取输入 TensorDict 并返回另一个包含名为“loss*”的损失密钥的 tensordict。
然后,将损失分成其组成部分可用于训练器在整个训练过程中记录各种损失值。输出 tensordict 中存在的其他标量也将被记录。
- 参数:
tensordict – 一个包含计算损失所需值的输入 tensordict。
- 返回值:
一个新的不包含批次维度的 tensordict,其中包含各种损失标量,这些标量将被命名为“loss*”。必须以这个名称返回损失,因为训练器将在反向传播之前读取它们。
- property functional¶
模块是否为函数式。
除非专门设计为非函数式,否则所有损失都是函数式。
- 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)