PPOLoss¶
- class torchrl.objectives.PPOLoss(*args, **kwargs)[源代码]¶
父 PPO 损失类。
PPO(近端策略优化)是一种无模型、在线 RL 算法,它利用记录的(批量的)轨迹来执行多个优化步骤,同时主动防止更新后的策略偏离其原始参数配置太多。
PPO 损失有不同的形式,具体取决于约束优化的实现方式:ClipPPOLoss 和 KLPENPPOLoss。与子类不同,此类不实现任何正则化,因此应谨慎使用。
有关 PPO 的更多详细信息,请参阅:“近端策略优化算法”,https://arxiv.org/abs/1707.06347
- 参数:
actor_network (ProbabilisticTensorDictSequential) – 策略运算符。
critic_network (ValueOperator) – 价值运算符。
- 关键字参数:
entropy_bonus (bool, optional) – 如果
True
,则会在损失中添加熵奖励,以支持探索性策略。samples_mc_entropy (int, optional) – 如果从策略运算符检索的分布没有熵的闭式公式,则将使用蒙特卡罗估计。
samples_mc_entropy
将控制用于计算此估计的样本数量。默认为1
。entropy_coef (scalar, optional) – 计算总损失时的熵乘数。默认为
0.01
。critic_coef (scalar, optional) – 计算总损失时的评论家损失乘数。默认为
1.0
。将critic_coef
设置为None
以从前向输出中排除价值损失。loss_critic_type (str, optional) – 价值差异的损失函数。可以是 “l1”、“l2” 或 “smooth_l1” 之一。默认为
"smooth_l1"
。normalize_advantage (bool, optional) – 如果
True
,则优势将在使用前进行归一化。默认为False
。separate_losses (bool, optional) – 如果
True
,策略和评论家之间的共享参数将仅在策略损失上进行训练。默认为False
,即梯度会传播到策略损失和评论家损失的共享参数。advantage_key (str, optional) – [已弃用,请改用 set_keys(advantage_key=advantage_key)] 输入 tensordict 键,优势预计将写入此处。默认为
"advantage"
。value_target_key (str, optional) – [已弃用,请改用 set_keys(value_target_key=value_target_key) 代替] 输入 tensordict 键,目标状态值预计将写入此处。默认为
"value_target"
。value_key (str, optional) – [已弃用,请改用 set_keys(value_key) 代替] 输入 tensordict 键,状态值预计将写入此处。默认为
"state_value"
。functional (bool, optional) – 模块是否应被函数化。函数化允许元 RL 等功能,但使得无法使用分布式模型(DDP、FSDP 等),并且会带来少量成本。默认为
True
。reduction (str, optional) – 指定应用于输出的缩减:
"none"
|"mean"
|"sum"
。"none"
:不应用缩减,"mean"
:输出的总和将除以输出中元素的数量,"sum"
:输出将被求和。默认值:"mean"
。clip_value (float, optional) – 如果提供,它将用于计算相对于输入 tensordict 价值估计的裁剪版本的值预测,并使用它来计算价值损失。裁剪的目的是限制极端价值预测的影响,帮助稳定训练并防止大型更新。但是,如果价值估计是由当前版本的价值估计器完成的,则它将不起作用。默认为
None
。
注意
优势(通常为 GAE)可以由损失函数或在训练循环中计算。后者通常是首选选项,但这取决于用户选择哪个选项。如果优势键(默认为
"advantage
)在输入 tensordict 中不存在,则优势将由forward()
方法计算。>>> ppo_loss = PPOLoss(actor, critic) >>> advantage = GAE(critic) >>> data = next(datacollector) >>> losses = ppo_loss(data) >>> # equivalent >>> advantage(data) >>> losses = ppo_loss(data)
可以使用
make_value_estimator()
构建自定义优势模块。默认值为GAE
,其超参数由default_value_kwargs()
指示。>>> ppo_loss = PPOLoss(actor, critic) >>> ppo_loss.make_value_estimator(ValueEstimators.TDLambda) >>> data = next(datacollector) >>> losses = ppo_loss(data)
注意
如果 actor 和价值函数共享参数,则可以通过仅将价值网络的头部传递给 PPO 损失模块来避免多次调用公共模块
>>> common = SomeModule(in_keys=["observation"], out_keys=["hidden"]) >>> actor_head = SomeActor(in_keys=["hidden"]) >>> value_head = SomeValue(in_keys=["hidden"]) >>> # first option, with 2 calls on the common module >>> model = ActorValueOperator(common, actor_head, value_head) >>> loss_module = PPOLoss(model.get_policy_operator(), model.get_value_operator()) >>> # second option, with a single call to the common module >>> loss_module = PPOLoss(ProbabilisticTensorDictSequential(model, actor_head), value_head)
无论是否激活 separate_losses,这都将起作用。
示例
>>> import torch >>> from torch import nn >>> from torchrl.data.tensor_specs 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.ppo import PPOLoss >>> from tensordict import TensorDict >>> n_act, n_obs = 4, 3 >>> spec = Bounded(-torch.ones(n_act), torch.ones(n_act), (n_act,)) >>> base_layer = nn.Linear(n_obs, 5) >>> net = nn.Sequential(base_layer, nn.Linear(5, 2 * n_act), NormalParamExtractor()) >>> module = SafeModule(net, in_keys=["observation"], out_keys=["loc", "scale"]) >>> actor = ProbabilisticActor( ... module=module, ... distribution_class=TanhNormal, ... in_keys=["loc", "scale"], ... spec=spec) >>> module = nn.Sequential(base_layer, nn.Linear(5, 1)) >>> value = ValueOperator( ... module=module, ... in_keys=["observation"]) >>> loss = PPOLoss(actor, value) >>> batch = [2, ] >>> action = spec.rand(batch) >>> data = TensorDict({"observation": torch.randn(*batch, n_obs), ... "action": action, ... "sample_log_prob": torch.randn_like(action[..., 1]), ... ("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", "sample_log_prob", "next_reward", "next_done", "next_terminated"]
+ actor 和价值网络的 in_keys。返回值是张量的元组,顺序如下:["loss_objective"]
+ 如果设置了 entropy_bonus,则为["entropy", "loss_entropy"]
+ 如果 critic_coef 不是None
,则为"loss_critic"
。也可以使用PPOLoss.select_out_keys()
方法过滤输出键。示例
>>> import torch >>> from torch import nn >>> from torchrl.data.tensor_specs 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.ppo import PPOLoss >>> n_act, n_obs = 4, 3 >>> spec = Bounded(-torch.ones(n_act), torch.ones(n_act), (n_act,)) >>> base_layer = nn.Linear(n_obs, 5) >>> net = nn.Sequential(base_layer, nn.Linear(5, 2 * n_act), NormalParamExtractor()) >>> module = SafeModule(net, in_keys=["observation"], out_keys=["loc", "scale"]) >>> actor = ProbabilisticActor( ... module=module, ... distribution_class=TanhNormal, ... in_keys=["loc", "scale"], ... spec=spec) >>> module = nn.Sequential(base_layer, nn.Linear(5, 1)) >>> value = ValueOperator( ... module=module, ... in_keys=["observation"]) >>> loss = PPOLoss(actor, value) >>> loss.set_keys(sample_log_prob="sampleLogProb") >>> _ = loss.select_out_keys("loss_objective") >>> batch = [2, ] >>> action = spec.rand(batch) >>> loss_objective = loss( ... observation=torch.randn(*batch, n_obs), ... action=action, ... sampleLogProb=torch.randn_like(action[..., 1]) / 10, ... 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_objective.backward()
注意
关于与非 tensordict 基于的模块的兼容性,有一个例外。如果 actor 网络是概率性的并且使用
CompositeDistribution
,则此类必须与 tensordict 一起使用,并且不能作为独立于 tensordict 的模块运行。这是因为复合动作空间本质上依赖于 tensordict 提供的结构化数据表示来处理其动作。- forward(tensordict: TensorDictBase = None) TensorDictBase [源代码]¶
它旨在读取输入 TensorDict 并返回另一个具有名为 “loss*” 的损失键的 tensordict。
然后,训练器可以使用将其损失分解为组件,以便在整个训练过程中记录各种损失值。输出 tensordict 中存在的其他标量也将被记录。
- 参数:
tensordict – 一个输入 tensordict,其中包含计算损失所需的值。
- 返回:
一个新的 tensordict,没有批次维度,其中包含各种损失标量,这些标量将被命名为 “loss*”。损失必须使用此名称返回,这一点至关重要,因为训练器将在反向传播之前读取它们。
- property functional¶
模块是否是函数式的。
除非经过专门设计使其不具有函数式,否则所有损失都是函数式的。
- loss_critic(tensordict: TensorDictBase) Tensor [源代码]¶
如果
critic_coef
不是None
,则返回乘以critic_coef
的评论家损失。
- 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)