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)