PPOLoss

class torchrl.objectives.PPOLoss(*args, **kwargs)

一个父 PPO loss 类。

PPO（Proximal Policy Optimization，近端策略优化）是一种无模型、在线强化学习算法，它利用记录的（一批）轨迹来执行多个优化步骤，同时主动防止更新后的策略偏离其原始参数配置太远。

PPO loss 有不同的实现方式，取决于约束优化方法的实现：ClipPPOLoss 和 KLPENPPOLoss。与它的子类不同，这个类不实现任何正则化，因此应谨慎使用。

关于 PPO 的更多细节，请参考：“Proximal Policy Optimization Algorithms”，https://arxiv.org/abs/1707.06347

参数:

• actor_network (ProbabilisticTensorDictSequential) – 策略算子。通常，这是一个 ProbabilisticTensorDictSequential 的子类，它接收观察作为输入，并输出动作（或动作）以及其对数概率值。

• critic_network (ValueOperator) – 值算子。critic 通常接收观察作为输入，并在输出键中返回一个标量值（默认为 state_value）。

注意

虽然此 loss 模块不强制执行任何特定的模型模式（训练/评估），但强烈建议在 RL 训练期间将模型保持在评估模式下，以确保确定性行为。当有效样本数量（ESS）下降或显著增加时，通常会观察到由于训练/评估模式不匹配导致的学习失败（请参阅下方的注意事项）。

注意

PPO loss 公开了一些额外的指标，可用于监控训练过程

• Clip fraction 是 PPO loss 中被裁剪权重的数量的比例（即被裁剪的权重数量与总权重数量的比率）。

• 有效样本数量（ESS）是批量中有效样本数量的度量，计算方法是重要性权重平方和的倒数。值为 1 表示所有重要性权重都等于 1（即样本权重相等）。任何低于 1 的值都表示样本的权重不相等，ESS 是有效样本数量的度量。如果该值显着下降或增加，通常表明模型配置存在问题（例如，训练/评估模式不匹配，或策略更新幅度过大）。

关键字参数:

• entropy_bonus (bool, optional) – 如果为 True，将向损失添加熵奖励，以鼓励探索性策略。

• samples_mc_entropy (int, optional) – 如果从策略算子中检索到的分布没有熵的封闭式公式，将使用蒙特卡洛估计。 samples_mc_entropy 将控制计算此估计所需的样本数量。默认为 1。

• entropy_coeff –

  标量 | 映射[嵌套键, 标量], 可选）：计算总 loss 时熵的乘数。 * **标量**：一个值应用于所有动作头的总熵。 * **映射** {head_name: coeff} 为每个动作头的熵提供单独的系数。默认为 0.01。

  有关详细的使用示例和故障排除，请参阅 ppo_entropy_coefficients。

• log_explained_variance (bool, optional) – 如果为 True，将计算 critic 预测与值目标之间的解释方差，并记录为 "explained_variance"。这有助于在训练期间监控 critic 的质量。最佳得分为 1.0，值越低越差。默认为 True。

• critic_coeff (scalar, optional) – 计算总 loss 时 critic loss 的乘数。默认为 1.0。将 critic_coeff 设置为 None 以将值 loss 从 forward 输出中排除。

• loss_critic_type (str, optional) – 值差异的 loss 函数。可以是 "l1"、"l2" 或 "smooth_l1" 之一。默认为 "smooth_l1"。

• normalize_advantage (bool, optional) – 如果为 True，则在使用之前将对优势（advantage）进行归一化。默认为 False。

• normalize_advantage_exclude_dims (Tuple[int], optional) – 在优势标准化中要排除的维度。负维度是有效的。这在多智能体（或多目标）设置中很有用，因为智能体（或目标）维度可以从缩减中排除。默认值：()。

• separate_losses (bool, 可选) – 如果为 True，则策略和评估器之间的共享参数将仅针对策略损失进行训练。默认为 False，即梯度将传播到策略和评估器损失的共享参数。

• advantage_key (str, optional) – [已弃用，请改用 set_keys(advantage_key=advantage_key) ] 预期在输入 tensordict 中写入优势的 tensordict 键。默认为 "advantage"。

• value_target_key (str, optional) – [已弃用，请改用 set_keys(value_target_key=value_target_key) ] 预期在输入 tensordict 中写入目标状态值的 tensordict 键。默认为 "value_target"。

• value_key (str, optional) – [已弃用，请改用 set_keys(value_key) ] 预期在输入 tensordict 中写入状态值的 tensordict 键。默认为 "state_value"。

• functional (bool, optional) – 模块是否应被函数化。函数化允许 meta-RL 等功能，但使得无法使用分布式模型（DDP、FSDP 等），并且会带来一点开销。默认为 True。

• reduction (str, optional) – 指定应用于输出的约简："none" | "mean" | "sum"。"none"：不应用约简，"mean"：输出的总和将除以输出中的元素数量，"sum"：将对输出进行求和。默认为 "mean"。

• clip_value (float, optional) – 如果提供，将用于计算值预测的裁剪版本，相对于输入 tensordict 的值估计，并使用它来计算值 loss。裁剪的目的是限制极端值预测的影响，有助于稳定训练并防止大幅更新。但是，如果值估计是由当前版本的值估计器完成的，则此参数将不起作用。默认为 None。

• device (torch.device, 可选) –

  缓冲区的设备。默认为 None。

  注意

  策略/ critic 的参数和缓冲区不会被转换为该设备，以确保存储与传递给其他组件（如数据收集器）的存储匹配。

注意

优势（通常是 GAE）可以由 loss 函数或在训练循环中计算。后者通常更受欢迎，但用户可以自行选择首选选项。如果输入 tensordict 中不存在优势键（默认为 "advantage"），则优势将由 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 和 value function 共享参数，可以通过仅将 value network 的 head 传递给 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,
...         "action_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),
        explained_variance: Tensor(shape=torch.Size([1]), device=cpu, dtype=torch.float32, is_shared=False),
        kl_approx: 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 和 value 网络的 in_keys。返回值是一个元组，按以下顺序排列：["loss_objective"] + ["entropy", "loss_entropy"]（如果设置了 entropy_bonus）+ "loss_critic"（如果 critic_coeff 不是 None）。还可以使用 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()

简单的熵系数示例:

>>> # Scalar entropy coefficient (default behavior)
>>> loss = PPOLoss(actor, critic, entropy_coeff=0.01)
>>>
>>> # Per-head entropy coefficients (for composite action spaces)
>>> entropy_coeff = {
...     ("agent0", "action_log_prob"): 0.01,  # Low exploration
...     ("agent1", "action_log_prob"): 0.05,  # High exploration
... }
>>> loss = PPOLoss(actor, critic, entropy_coeff=entropy_coeff)

注意

在与非 tensordict 模块兼容性方面有一个例外。如果 actor 网络是概率性的并且使用 CompositeDistribution，则此类必须与 tensordicts 一起使用，并且不能作为独立的 tensordict 模块运行。这是因为复合动作空间本质上依赖于 tensordicts 提供的数据结构化表示来处理其动作。

注意

熵奖励和系数管理

熵奖励通过将策略的负熵添加到 loss 来鼓励探索。可以通过两种方式进行配置：

标量系数（默认）：对所有动作头使用单个系数

>>> loss = PPOLoss(actor, critic, entropy_coeff=0.01)

每头系数：对不同的动作组件使用不同的系数

>>> # For a robot with movement and gripper actions
>>> entropy_coeff = {
...     ("agent0", "action_log_prob"): 0.01,  # Movement: low exploration
...     ("agent1", "action_log_prob"): 0.05,  # Gripper: high exploration
... }
>>> loss = PPOLoss(actor, critic, entropy_coeff=entropy_coeff)

键要求：在使用每头系数时，必须提供每个动作头对数概率的完整嵌套键路径（例如，(“agent0”, “action_log_prob”)）。

监控熵 loss:

在使用复合动作空间时，loss 输出包括：- “entropy”：所有动作头的总熵（用于日志记录）- “composite_entropy”：每个动作头的单独熵值- “loss_entropy”：加权的熵 loss 项

示例输出

>>> result = loss(data)
>>> print(result["entropy"])           # Total entropy: 2.34
>>> print(result["composite_entropy"]) # Per-head: {"movement": 1.2, "gripper": 1.14}
>>> print(result["loss_entropy"])      # Weighted loss: -0.0234

常见问题:

KeyError: “Missing entropy coeff for head ‘head_name’”:

• 确保为所有动作头提供系数

• 使用完整的嵌套键：(“head_name”, “action_log_prob”)

• 检查您的动作空间结构是否与系数映射匹配

熵计算错误:

• 在创建策略之前调用 set_composite_lp_aggregate(False).set()

• 验证您的动作空间是否使用 CompositeDistribution

default_keys

别名：_AcceptedKeys

forward(tensordict: TensorDictBase = None) → TensorDictBase

它旨在读取一个输入的 TensorDict 并返回另一个包含名为“loss*”的损失键的 tensordict。

将损失分解为其组成部分可以被训练器用于在训练过程中记录各种损失值。输出 tensordict 中存在的其他标量也将被记录。

参数:

tensordict – 一个输入的 tensordict，包含计算损失所需的值。

返回:

一个没有批处理维度的新 tensordict，其中包含各种损失标量，这些标量将被命名为“loss*”。重要的是，损失必须以这个名称返回，因为它们将在反向传播之前被训练器读取。

property functional

模块是否功能化。

除非经过专门设计使其不具有功能性，否则所有损失都具有功能性。

loss_critic(tensordict: TensorDictBase) → tuple[torch.Tensor | TensorDict, ...]

返回 critic loss 乘以 critic_coeff，如果它不是 None。

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)