SACLoss¶
- 类 torchrl.objectives.SACLoss(*args, **kwargs)[源代码]¶
SAC 损失函数的 TorchRL 实现。
论文 “Soft Actor-Critic: Off-Policy Maximum Entropy Deep Reinforcement Learning with a Stochastic Actor” https://arxiv.org/abs/1801.01290 和 “Soft Actor-Critic Algorithms and Applications” https://arxiv.org/abs/1812.05905 中提出。
- 参数:
actor_network (ProbabilisticActor) – 随机 actor
qvalue_network (TensorDictModule) –
Q(s, a) 参数化模型。此模块通常输出一个
"state_action_value"
条目。如果提供单个 qvalue_network 实例,它将被复制num_qvalue_nets
次。如果传递模块列表,除非它们共享相同的身份(在这种情况下,原始参数将被扩展),否则它们的参数将被堆叠。警告
当传递参数列表时,它将__不__与策略参数进行比较,所有参数将被视为非绑定的。
value_network (TensorDictModule, 可选) –
V(s) 参数化模型。此模块通常输出一个
"state_value"
条目。注意
如果未提供,则假定使用 SAC 的第二个版本,该版本仅需要 Q-Value 网络。
- 关键字参数:
num_qvalue_nets (integer, 可选) – 使用的 Q-Value 网络数量。默认为
2
。loss_function (str, 可选) – 用于值函数损失的损失函数。默认为 “smooth_l1”。
alpha_init (
float
, 可选) – 初始熵乘数。默认为 1.0。min_alpha (
float
, 可选) – alpha 的最小值。默认为 None (无最小值)。max_alpha (
float
, 可选) – alpha 的最大值。默认为 None (无最大值)。action_spec (TensorSpec, 可选) – 动作张量规范。如果未提供且目标熵为
"auto"
,则将从 actor 获取。fixed_alpha (bool, 可选) – 如果为
True
,则 alpha 将固定为其初始值。否则,alpha 将被优化以匹配 'target_entropy' 值。默认为False
。target_entropy (
float
或 str, 可选) – 随机策略的目标熵。默认为 "auto",此时目标熵计算为-prod(n_actions)
。delay_actor (bool, 可选) – 是否将目标 actor 网络与用于数据收集的 actor 网络分开。默认为
False
。delay_qvalue (bool, 可选) – 是否将目标 Q 值网络与用于数据收集的 Q 值网络分开。默认为
True
。delay_value (bool, 可选) – 是否将目标值网络与用于数据收集的值网络分开。默认为
True
。priority_key (str, 可选) – [已弃用,请改用 .set_keys(priority_key=priority_key) 代替] Tensordict 中用于写入优先级(用于优先回放缓冲区)的键。默认为
"td_error"
。separate_losses (bool, 可选) – 如果为
True
,则策略和 critic 之间共享的参数将仅在策略损失上进行训练。默认为False
,即梯度会传播到策略和 critic 损失的共享参数。reduction (str, 可选) – 指定应用于输出的归约方式:
"none"
|"mean"
|"sum"
。"none"
:不应用归约,"mean"
:输出的总和将除以输出中的元素数量,"sum"
:输出将求和。默认为:"mean"
。skip_done_states (bool, 可选) – 用于值计算的 actor 网络是否只应在有效的、未终止的下一个状态上运行。如果为
True
,则假定 done 状态可以广播到数据的形状,并且屏蔽数据会产生有效的数据结构。除此之外,在 MARL 设置或使用 RNN 时,这可能不成立。默认为False
。
示例
>>> 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.sac import SACLoss >>> 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 ValueClass(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)) >>> module = ValueClass() >>> qvalue = ValueOperator( ... module=module, ... in_keys=['observation', 'action']) >>> module = nn.Linear(n_obs, 1) >>> value = ValueOperator( ... module=module, ... in_keys=["observation"]) >>> loss = SACLoss(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={ alpha: Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, is_shared=False), 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_alpha: 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。返回值为按以下顺序排列的张量元组:["loss_actor", "loss_qvalue", "loss_alpha", "alpha", "entropy"]
+"loss_value"
(如果使用第一个版本)。示例
>>> 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.sac import SACLoss >>> _ = 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 ValueClass(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)) >>> module = ValueClass() >>> qvalue = ValueOperator( ... module=module, ... in_keys=['observation', 'action']) >>> module = nn.Linear(n_obs, 1) >>> value = ValueOperator( ... module=module, ... in_keys=["observation"]) >>> loss = SACLoss(actor, qvalue, value) >>> batch = [2, ] >>> action = spec.rand(batch) >>> 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()
输出键也可以使用
SACLoss.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 [源代码]¶
它旨在读取输入的 TensorDict,并返回另一个包含名为“loss*”的损失键的 tensordict。
然后,训练器可以使用其组件中的损失拆分来记录训练过程中的各种损失值。输出 tensordict 中存在的其他标量也将被记录。
- 参数:
tensordict – 包含计算损失所需值的输入 tensordict。
- 返回:
一个不含批量维度的新 tensordict,包含各种损失标量,这些标量将被命名为“loss*”。损失必须以此名称返回,因为训练器会在反向传播之前读取它们,这一点至关重要。
- load_state_dict(state_dict: Mapping[str, Any], strict: bool = True, assign: bool = False)¶
从
state_dict
中复制参数和缓冲区到此模块及其子模块。如果
strict
为True
,则state_dict
的键必须与此模块的state_dict()
函数返回的键完全匹配。警告
如果
assign
为True
,则优化器必须在调用load_state_dict
之后创建,除非get_swap_module_params_on_conversion()
为True
。- 参数:
state_dict (dict) – 包含参数和持久化缓冲区的字典。
strict (bool, 可选) – 是否严格要求
state_dict
中的键与此模块的state_dict()
函数返回的键匹配。默认为:True
assign (bool, 可选) – 设置为
False
时,保留当前模块中张量的属性;设置为True
时,保留 state dict 中张量的属性。唯一的例外是Default: ``False`
的requires_grad
字段。
- 返回:
- missing_keys 是一个字符串列表,包含任何预期但
未在提供的
state_dict
中找到的键。
- unexpected_keys 是一个字符串列表,包含未
被此模块预期但存在于提供的
state_dict
中的键。
- 返回类型:
带有
missing_keys
和unexpected_keys
字段的NamedTuple
注意
如果参数或缓冲区注册为
None
且其对应的键存在于state_dict
中,则load_state_dict()
将引发RuntimeError
。
- 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)
- state_dict(*args, destination=None, prefix='', keep_vars=False)¶
返回一个字典,其中包含对模块整体状态的引用。
参数和持久缓冲区(例如,运行平均值)都包含在内。键是对应的参数和缓冲区名称。设置为
None
的参数和缓冲区不包含在内。注意
返回的对象是一个浅拷贝。它包含对模块参数和缓冲区的引用。
警告
当前
state_dict()
也按顺序接受destination
、prefix
和keep_vars
的位置参数。但是,此用法已被弃用,未来版本将强制使用关键字参数。警告
请避免使用参数
destination
,因为它并非为最终用户设计。- 参数:
destination (dict, 可选) – 如果提供,模块的状态将更新到该字典中并返回同一对象。否则,将创建一个
OrderedDict
并返回。默认为:None
。prefix (str, 可选) – 添加到参数和缓冲区名称前的字符串前缀,用于构成 state_dict 中的键。默认为:
''
。keep_vars (bool, 可选) – 默认情况下,state dict 中返回的
Tensor
与 autograd 分离。如果设置为True
,则不会执行分离。默认为:False
。
- 返回:
一个包含模块整体状态的字典
- 返回类型:
dict
示例
>>> # xdoctest: +SKIP("undefined vars") >>> module.state_dict().keys() ['bias', 'weight']