ding.model¶

常见¶

请参考 ding/model/common 了解更多详情。

创建模型¶

ding.model.create_model(cfg: EasyDict) → Module[source]¶

Overview:

根据给定的EasyDict类型的cfg创建一个神经网络模型。

Arguments:

cfg: (EasyDict): 用户的模型配置。键 import_name 用于导入模块，键 type 用于指示模型。

Returns:

(torch.nn.Module): 创建的神经网络模型。

Examples:

>>> cfg = EasyDict({
>>>     'import_names': ['ding.model.template.q_learning'],
>>>     'type': 'dqn',
>>>     'obs_shape': 4,
>>>     'action_shape': 2,
>>> })
>>> model = create_model(cfg)

提示

此方法不会修改cfg，它会深拷贝cfg然后进行修改。

ConvEncoder¶

class ding.model.ConvEncoder(obs_shape: SequenceType, hidden_size_list: SequenceType = [32, 64, 64, 128], activation: Module | None = ReLU(), kernel_size: SequenceType = [8, 4, 3], stride: SequenceType = [4, 2, 1], padding: SequenceType | None = None, layer_norm: bool | None = False, norm_type: str | None = None)[源代码]¶

Overview:: 卷积编码器用于编码二维图像观测。
Interfaces:: __init__, forward.

__init__(obs_shape: SequenceType, hidden_size_list: SequenceType = [32, 64, 64, 128], activation: Module | None = ReLU(), kernel_size: SequenceType = [8, 4, 3], stride: SequenceType = [4, 2, 1], padding: SequenceType | None = None, layer_norm: bool | None = False, norm_type: str | None = None) → None[源代码]¶

Overview:

根据提供的参数初始化Convolution Encoder。

Arguments:

obs_shape (SequenceType): in_channel 的序列，加上一个或多个 input size。
hidden_size_list (SequenceType): 后续卷积层和最终全连接层的 hidden_size 序列。
激活函数 (nn.Module): 在卷积 layers 和 ResBlock 中使用的激活函数类型。默认是 nn.ReLU()。
kernel_size (SequenceType): 后续卷积层的 kernel_size 序列。
步幅 (SequenceType): 后续卷积层的stride序列。
填充 (SequenceType): 为每个卷积层的输入添加的四个边的填充。详情请参见 nn.Conv2d。默认值为 None。
layer_norm (bool): 是否使用 DreamerLayerNorm，这是在 DreamerV3 中提出的一种特殊技巧。
norm_type (str): 使用的归一化类型。详情请参见 ding.torch_utils.network.ResBlock。默认值为 None。

forward(x: Tensor) → Tensor[source]¶

Overview:

返回环境2D图像观测的1D嵌入张量。

Arguments:

x (torch.Tensor): 环境的原始2D观测。

Returns:

输出 (torch.Tensor): 输出嵌入张量。

Shapes:

x : \((B, C, H, W)\), 其中 B 是批量大小, C 是通道数, H 是高度, W 是宽度.
输出：\((B, N)\)，其中 N = hidden_size_list[-1] 。

Examples:

>>> conv = ConvEncoder(
>>>    obs_shape=(4, 84, 84),
>>>    hidden_size_list=[32, 64, 64, 128],
>>>    activation=nn.ReLU(),
>>>    kernel_size=[8, 4, 3],
>>>    stride=[4, 2, 1],
>>>    padding=None,
>>>    layer_norm=False,
>>>    norm_type=None
>>> )
>>> x = torch.randn(1, 4, 84, 84)
>>> output = conv(x)

FCEncoder¶

class ding.model.FCEncoder(obs_shape: int, hidden_size_list: SequenceType, res_block: bool = False, activation: Module | None = ReLU(), norm_type: str | None = None, dropout: float | None = None)[source]¶

Overview:: 全连接编码器用于编码一维输入变量。
Interfaces:: __init__, forward.

__init__(obs_shape: int, hidden_size_list: SequenceType, res_block: bool = False, activation: Module | None = ReLU(), norm_type: str | None = None, dropout: float | None = None) → None[来源]¶

Overview:

根据参数初始化FC编码器。

Arguments:

obs_shape (int): 观测形状。
hidden_size_list (SequenceType): 后续全连接层的 hidden_size 序列。
res_block (bool): 是否使用 res_block。默认是 False。
activation (nn.Module): 在 ResFCBlock 中使用的激活类型。默认是 nn.ReLU()。
norm_type (str): 使用的归一化类型。有关更多详细信息，请参见 ding.torch_utils.network.ResFCBlock。默认值为 None。
dropout (float): dropout层的丢弃率。如果为None，则默认没有dropout层。

forward(x: Tensor) → Tensor[source]¶

Overview:

返回环境观察的输出嵌入张量。

Arguments:

x (torch.Tensor): 环境的原始观察值。

Returns:

输出 (torch.Tensor): 输出嵌入张量。

Shapes:

x : \((B, M)\), 其中 M = obs_shape.
输出：\((B, N)\)，其中 N = hidden_size_list[-1]。

Examples:

>>> fc = FCEncoder(
>>>    obs_shape=4,
>>>    hidden_size_list=[32, 64, 64, 128],
>>>    activation=nn.ReLU(),
>>>    norm_type=None,
>>>    dropout=None
>>> )
>>> x = torch.randn(1, 4)
>>> output = fc(x)

IMPALAConvEncoder¶

class ding.model.IMPALAConvEncoder(obs_shape: SequenceType, channels: SequenceType = (16, 32, 32), outsize: int = 256, scale_ob: float = 255.0, nblock: int = 2, final_relu: bool = True, **kwargs)[source]¶

Overview:: IMPALA CNN编码器，用于IMPALA算法。 IMPALA：具有重要性加权演员-学习者架构的可扩展分布式深度强化学习，https://arxiv.org/pdf/1802.01561.pdf,
Interface:: __init__, forward, output_shape.

__init__(obs_shape: SequenceType, channels: SequenceType = (16, 32, 32), outsize: int = 256, scale_ob: float = 255.0, nblock: int = 2, final_relu: bool = True, **kwargs) → None[源代码]¶

Overview:

根据参数初始化IMPALA CNN编码器。

Arguments:

obs_shape (SequenceType): 二维图像观测形状。
通道 (SequenceType): 一系列impala cnn块的通道数。序列中的每个元素是一个impala cnn块的输出通道数。
outsize (int): 最终线性层的输出大小，即一维嵌入向量的维度。
scale_ob (float): 输入观测的缩放比例，用于归一化输入观测，例如将原始图像观测除以255.0。
nblock (int): 每个块中的残差块数量。
final_relu (bool): 是否在编码器的最终输出中使用ReLU激活函数。
kwargs (Dict[str, Any]): IMPALACnnDownStack的其他参数。

离散头¶

class ding.model.DiscreteHead(hidden_size: int, output_size: int, layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, dropout: float | None = None, noise: bool | None = False)[source]¶

Overview:: DiscreteHead 用于生成离散动作的 logit 或 Q 值 logit，通常用于离散动作空间的 q-learning 算法或 actor-critic 算法。
Interfaces:: __init__, forward.

__init__(hidden_size: int, output_size: int, layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, dropout: float | None = None, noise: bool | None = False) → None[源代码]¶

Overview:

根据提供的参数初始化DiscreteHead层。

Arguments:

hidden_size (int): 连接到 DiscreteHead 的 MLP 的 hidden_size。
output_size (int): 输出的数量。
层数 (int): 网络中用于计算Q值输出的层数。
activation (nn.Module): 在MLP中使用的激活函数类型。如果为None，则默认将激活函数设置为nn.ReLU()。默认值为None。
norm_type (str): 使用的归一化类型。有关更多详细信息，请参见 ding.torch_utils.network.fc_block。默认值为 None。
dropout (float): 丢弃率，默认设置为None。
噪声 (bool): 是否在Q网络的MLP中使用NoiseLinearLayer作为layer_fn。默认值为False。

forward(x: Tensor) → Dict[来源]¶

Overview:

使用编码的嵌入张量运行MLP，并使用DiscreteHead返回预测字典。

Arguments:

x (torch.Tensor): 包含输入嵌入的张量。

Returns:

输出 (Dict): 包含关键字 logit (torch.Tensor) 的字典。

Shapes:

x: \((B, N)\), 其中 B = batch_size 和 N = hidden_size.
logit: \((B, M)\), 其中 M = output_size.

Examples:

>>> head = DiscreteHead(64, 64)
>>> inputs = torch.randn(4, 64)
>>> outputs = head(inputs)
>>> assert isinstance(outputs, dict) and outputs['logit'].shape == torch.Size([4, 64])

DistributionHead¶

class ding.model.DistributionHead(hidden_size: int, output_size: int, layer_num: int = 1, n_atom: int = 51, v_min: float = -10, v_max: float = 10, activation: Module | None = ReLU(), norm_type: str | None = None, noise: bool | None = False, eps: float | None = 1e-06)[source]¶

Overview:: DistributionHead 用于生成 Q 值的分布。该模块用于 C51 算法。
Interfaces:: __init__, forward.

__init__(hidden_size: int, output_size: int, layer_num: int = 1, n_atom: int = 51, v_min: float = -10, v_max: float = 10, activation: Module | None = ReLU(), norm_type: str | None = None, noise: bool | None = False, eps: float | None = 1e-06) → None[源代码]¶

Overview:

根据提供的参数初始化DistributionHead层。

Arguments:

hidden_size (int): 连接到 DistributionHead 的 MLP 的 hidden_size。
output_size (int): 输出的数量。
层数 (int): 网络中用于计算Q值分布的层数。
n_atom (int): 原子数量（离散支持）。默认值为 51。
v_min (int): 原子的最小值。默认值为 -10。
v_max (int): 原子的最大值。默认值为 10。
activation (nn.Module): 在MLP中使用的激活函数类型。如果为None，则默认将激活函数设置为nn.ReLU()。默认值为None。
norm_type (str): 使用的归一化类型。有关更多详细信息，请参见 ding.torch_utils.network.fc_block。默认值为 None。
噪声 (bool): 是否在Q网络的MLP中使用NoiseLinearLayer作为layer_fn。默认值为False。
eps (float): 用于数值稳定性的小常数。

forward(x: Tensor) → Dict[source]¶

Overview:

使用编码的嵌入张量运行MLP，并使用DistributionHead返回预测字典。

Arguments:

x (torch.Tensor): 包含输入嵌入的张量。

Returns:

输出 (Dict): 包含关键字 logit (torch.Tensor) 和 distribution (torch.Tensor) 的字典。

Shapes:

x: \((B, N)\), 其中 B = batch_size 和 N = hidden_size.
logit: \((B, M)\), 其中 M = output_size.
分布: \((B, M, n_atom)\).

Examples:

>>> head = DistributionHead(64, 64)
>>> inputs = torch.randn(4, 64)
>>> outputs = head(inputs)
>>> assert isinstance(outputs, dict)
>>> assert outputs['logit'].shape == torch.Size([4, 64])
>>> # default n_atom is 51
>>> assert outputs['distribution'].shape == torch.Size([4, 64, 51])

彩虹头¶

class ding.model.RainbowHead(hidden_size: int, output_size: int, layer_num: int = 1, n_atom: int = 51, v_min: float = -10, v_max: float = 10, activation: Module | None = ReLU(), norm_type: str | None = None, noise: bool | None = True, eps: float | None = 1e-06)[source]¶

Overview:: RainbowHead 用于生成 Q 值的分布。该模块用于 Rainbow DQN。
Interfaces:: __init__, forward.

__init__(hidden_size: int, output_size: int, layer_num: int = 1, n_atom: int = 51, v_min: float = -10, v_max: float = 10, activation: Module | None = ReLU(), norm_type: str | None = None, noise: bool | None = True, eps: float | None = 1e-06) → None[source]¶

Overview:

根据提供的参数初始化RainbowHead层。

Arguments:

hidden_size (int): 连接到 RainbowHead 的 MLP 的 hidden_size。
output_size (int): 输出的数量。
层数 (int): 网络中用于计算Q值输出的层数。
n_atom (int): 原子数量（离散支持）。默认值为 51。
v_min (int): 原子的最小值。默认值为 -10。
v_max (int): 原子的最大值。默认值为 10。
activation (nn.Module): 在MLP中使用的激活函数类型。如果为None，则默认将激活函数设置为nn.ReLU()。默认值为None。
norm_type (str): 使用的归一化类型。有关更多详细信息，请参见 ding.torch_utils.network.fc_block。默认值为 None。
噪声 (bool): 是否在Q网络的MLP中使用NoiseLinearLayer作为layer_fn。默认值为False。
eps (float): 用于数值稳定性的小常数。

forward(x: Tensor) → Dict[source]¶

Overview:

使用编码的嵌入张量运行MLP与RainbowHead并返回预测字典。

Arguments:

x (torch.Tensor): 包含输入嵌入的张量。

Returns:

输出 (Dict): 包含关键字 logit (torch.Tensor) 和 distribution (torch.Tensor) 的字典。

Shapes:

x: \((B, N)\), 其中 B = batch_size 和 N = hidden_size.
logit: \((B, M)\), 其中 M = output_size.
分布: \((B, M, n_atom)\).

Examples:

>>> head = RainbowHead(64, 64)
>>> inputs = torch.randn(4, 64)
>>> outputs = head(inputs)
>>> assert isinstance(outputs, dict)
>>> assert outputs['logit'].shape == torch.Size([4, 64])
>>> # default n_atom is 51
>>> assert outputs['distribution'].shape == torch.Size([4, 64, 51])

QRDQNHead¶

class ding.model.QRDQNHead(hidden_size: int, output_size: int, layer_num: int = 1, num_quantiles: int = 32, activation: Module | None = ReLU(), norm_type: str | None = None, noise: bool | None = False)[source]¶

Overview:: QRDQNHead（分位数回归DQN）用于输出动作分位数。
Interfaces:: __init__, forward.

__init__(hidden_size: int, output_size: int, layer_num: int = 1, num_quantiles: int = 32, activation: Module | None = ReLU(), norm_type: str | None = None, noise: bool | None = False) → None[源代码]¶

Overview:

根据提供的参数初始化QRDQNHead层。

Arguments:

hidden_size (int): 连接到 QRDQNHead 的 MLP 的 hidden_size。
output_size (int): 输出的数量。
层数 (int): 网络中用于计算Q值输出的层数。
num_quantiles (int): 分位数的数量。默认值为 32。
activation (nn.Module): 在MLP中使用的激活函数类型。如果为None，则默认将激活函数设置为nn.ReLU()。默认值为None。
norm_type (str): 使用的归一化类型。有关更多详细信息，请参见 ding.torch_utils.network.fc_block。默认值为 None。
噪声 (bool): 是否在Q网络的MLP中使用NoiseLinearLayer作为layer_fn。默认值为False。

forward(x: Tensor) → Dict[来源]¶

Overview:

使用编码的嵌入张量运行MLP与QRDQNHead并返回预测字典。

Arguments:

x (torch.Tensor): 包含输入嵌入的张量。

Returns:

输出 (Dict): 包含关键字 logit (torch.Tensor), q (torch.Tensor), 和 tau (torch.Tensor) 的字典。

Shapes:

x: \((B, N)\), 其中 B = batch_size 和 N = hidden_size.
logit: \((B, M)\), 其中 M = output_size.
q: \((B, M, num_quantiles)\).
tau: \((B, M, 1)\).

Examples:

>>> head = QRDQNHead(64, 64)
>>> inputs = torch.randn(4, 64)
>>> outputs = head(inputs)
>>> assert isinstance(outputs, dict)
>>> assert outputs['logit'].shape == torch.Size([4, 64])
>>> # default num_quantiles is 32
>>> assert outputs['q'].shape == torch.Size([4, 64, 32])
>>> assert outputs['tau'].shape == torch.Size([4, 32, 1])

分位数头¶

class ding.model.QuantileHead(hidden_size: int, output_size: int, layer_num: int = 1, num_quantiles: int = 32, quantile_embedding_size: int = 128, beta_function_type: str | None = 'uniform', activation: Module | None = ReLU(), norm_type: str | None = None, noise: bool | None = False)[源代码]¶

Overview:: QuantileHead 用于输出动作分位数。该模块用于IQN。
Interfaces:: __init__, forward, quantile_net.

注意

QuantileHead 和 QRDQNHead 的区别在于，QuantileHead 将状态-动作分位数函数建模为从状态-动作和某些基础分布的样本的映射，而 QRDQNHead 通过狄拉克函数的均匀混合来近似随机回报。

__init__(hidden_size: int, output_size: int, layer_num: int = 1, num_quantiles: int = 32, quantile_embedding_size: int = 128, beta_function_type: str | None = 'uniform', activation: Module | None = ReLU(), norm_type: str | None = None, noise: bool | None = False) → None[源代码]¶

Overview:

根据提供的参数初始化QuantileHead层。

Arguments:

hidden_size (int): 连接到 QuantileHead 的 MLP 的 hidden_size。
output_size (int): 输出的数量。
层数 (int): 网络中用于计算Q值输出的层数。
num_quantiles (int): 分位数的数量。
quantile_embedding_size (int): 分位数的嵌入大小。
beta_function_type (str): beta函数的类型。更多详情请参见 ding.rl_utils.beta_function.py。默认值为 uniform。
activation (nn.Module): 在MLP中使用的激活函数类型。如果为None，则默认将激活函数设置为nn.ReLU()。默认值为None。
norm_type (str): 使用的归一化类型。有关更多详细信息，请参见 ding.torch_utils.network.fc_block。默认值为 None。
噪声 (bool): 是否在Q网络的MLP中使用NoiseLinearLayer作为layer_fn。默认值为False。

forward(x: Tensor, num_quantiles: int | None = None) → Dict[来源]¶

Overview:

使用编码的嵌入张量运行MLP，并使用QuantileHead返回预测字典。

Arguments:

x (torch.Tensor): 包含输入嵌入的张量。

Returns:

输出 (Dict): 包含关键字 logit (torch.Tensor), q (torch.Tensor), 和 quantiles (torch.Tensor) 的字典。

Shapes:

x: \((B, N)\), 其中 B = batch_size 和 N = hidden_size.
logit: \((B, M)\), 其中 M = output_size.
q: \((num_quantiles, B, M)\).
分位数: \((quantile_embedding_size, 1)\).

Examples:

>>> head = QuantileHead(64, 64)
>>> inputs = torch.randn(4, 64)
>>> outputs = head(inputs)
>>> assert isinstance(outputs, dict)
>>> assert outputs['logit'].shape == torch.Size([4, 64])
>>> # default num_quantiles is 32
>>> assert outputs['q'].shape == torch.Size([32, 4, 64])
>>> assert outputs['quantiles'].shape == torch.Size([128, 1])

quantile_net(quantiles: Tensor) → Tensor[来源]¶

Overview:

确定性参数函数经过训练以重新参数化来自基础分布的样本。通过Q学习的重复贝尔曼更新迭代，估计出最优动作值函数。

Arguments:

x (torch.Tensor): 参数样本的编码嵌入张量。

Returns:

quantile_net (torch.Tensor): 重参数化后的分位数网络输出张量。

Shapes:

quantile_net \((quantile_embedding_size, M)\), 其中 M = output_size.

Examples:

>>> head = QuantileHead(64, 64)
>>> quantiles = torch.randn(128,1)
>>> qn_output = head.quantile_net(quantiles)
>>> assert isinstance(qn_output, torch.Tensor)
>>> # default quantile_embedding_size: int = 128,
>>> assert qn_output.shape == torch.Size([128, 64])

FQFHead¶

class ding.model.FQFHead(hidden_size: int, output_size: int, layer_num: int = 1, num_quantiles: int = 32, quantile_embedding_size: int = 128, activation: Module | None = ReLU(), norm_type: str | None = None, noise: bool | None = False)[source]¶

Overview:: FQFHead 用于输出动作分位数。该模块用于FQF。
Interfaces:: __init__, forward, quantile_net.

注意

FQFHead的实现基于论文https://arxiv.org/abs/1911.02140。 FQFHead与QuantileHead的区别在于，在FQF中，估计了N个可调整的分位数值以近似分位数函数。回报的分布通过N个Dirac函数的加权混合来近似。而在IQN中，状态-动作分位数函数被建模为从状态-动作和某些基础分布的样本的映射。

__init__(hidden_size: int, output_size: int, layer_num: int = 1, num_quantiles: int = 32, quantile_embedding_size: int = 128, activation: Module | None = ReLU(), norm_type: str | None = None, noise: bool | None = False) → None[源代码]¶

Overview:

根据提供的参数初始化FQFHead层。

Arguments:

hidden_size (int): 连接到 FQFHead 的 MLP 的 hidden_size。
output_size (int): 输出的数量。
层数 (int): 网络中用于计算Q值输出的层数。
num_quantiles (int): 分位数的数量。
quantile_embedding_size (int): 分位数的嵌入大小。
activation (nn.Module): 在MLP中使用的激活函数类型。如果为None，则默认将激活函数设置为nn.ReLU()。默认值为None。
norm_type (str): 使用的归一化类型。有关更多详细信息，请参见 ding.torch_utils.network.fc_block。默认值为 None。
噪声 (bool): 是否在Q网络的MLP中使用NoiseLinearLayer作为layer_fn。默认值为False。

forward(x: Tensor, num_quantiles: int | None = None) → Dict[source]¶

Overview:

使用编码的嵌入张量通过FQFHead运行MLP并返回预测字典。

Arguments:

x (torch.Tensor): 包含输入嵌入的张量。

Returns:

输出 (Dict): 包含关键词 logit (torch.Tensor), q (torch.Tensor), quantiles (torch.Tensor), quantiles_hats (torch.Tensor), q_tau_i (torch.Tensor), entropies (torch.Tensor) 的字典。

Shapes:

x: \((B, N)\), 其中 B = batch_size 和 N = hidden_size.
logit: \((B, M)\), 其中 M = output_size.
q: \((B, num_quantiles, M)\).
分位数: \((B, num_quantiles + 1)\).
quantiles_hats: \((B, num_quantiles)\).
q_tau_i: \((B, num_quantiles - 1, M)\).
熵值: \((B, 1)\).

Examples:

>>> head = FQFHead(64, 64)
>>> inputs = torch.randn(4, 64)
>>> outputs = head(inputs)
>>> assert isinstance(outputs, dict)
>>> assert outputs['logit'].shape == torch.Size([4, 64])
>>> # default num_quantiles is 32
>>> assert outputs['q'].shape == torch.Size([4, 32, 64])
>>> assert outputs['quantiles'].shape == torch.Size([4, 33])
>>> assert outputs['quantiles_hats'].shape == torch.Size([4, 32])
>>> assert outputs['q_tau_i'].shape == torch.Size([4, 31, 64])
>>> assert outputs['quantiles'].shape == torch.Size([4, 1])

quantile_net(quantiles: Tensor) → Tensor[来源]¶

Overview:

确定性参数函数经过训练，用于重新参数化来自quantiles_proposal网络的样本。通过Q-learning的重复Bellman更新迭代，估计最优动作值函数。

Arguments:

x (torch.Tensor): 参数样本的编码嵌入张量。

Returns:

quantile_net (torch.Tensor): 重参数化后的分位数网络输出张量。

Examples:

>>> head = FQFHead(64, 64)
>>> quantiles = torch.randn(4,32)
>>> qn_output = head.quantile_net(quantiles)
>>> assert isinstance(qn_output, torch.Tensor)
>>> # default quantile_embedding_size: int = 128,
>>> assert qn_output.shape == torch.Size([4, 32, 64])

决斗头¶

class ding.model.DuelingHead(hidden_size: int, output_size: int, layer_num: int = 1, a_layer_num: int | None = None, v_layer_num: int | None = None, activation: Module | None = ReLU(), norm_type: str | None = None, dropout: float | None = None, noise: bool | None = False)[source]¶

Overview:: DuelingHead 用于输出离散动作的logit。该模块用于Dueling DQN。
Interfaces:: __init__, forward.

Overview:

根据提供的参数初始化DuelingHead层。

Arguments:

hidden_size (int): 连接到 DuelingHead 的 MLP 的 hidden_size。
output_size (int): 输出的数量。
a_layer_num (int): 用于计算动作输出的网络中的层数。
v_layer_num (int): 网络中用于计算值输出的层数。
activation (nn.Module): 在MLP中使用的激活函数类型。如果为None，则默认将激活函数设置为nn.ReLU()。默认值为None。
norm_type (str): 使用的归一化类型。有关更多详细信息，请参见 ding.torch_utils.network.fc_block。默认值为 None。
dropout (float): dropout层的丢弃率。默认值为 None。
噪声 (bool): 是否在Q网络的MLP中使用NoiseLinearLayer作为layer_fn。默认值为False。

forward(x: Tensor) → Dict[来源]¶

Overview:

使用编码的嵌入张量运行带有DuelingHead的MLP并返回预测字典。

Arguments:

x (torch.Tensor): 包含输入嵌入的张量。

Returns:

输出 (Dict): 包含关键字 logit (torch.Tensor) 的字典。

Shapes:

x: \((B, N)\), 其中 B = batch_size 和 N = hidden_size.
logit: \((B, M)\), 其中 M = output_size.

Examples:

>>> head = DuelingHead(64, 64)
>>> inputs = torch.randn(4, 64)
>>> outputs = head(inputs)
>>> assert isinstance(outputs, dict)
>>> assert outputs['logit'].shape == torch.Size([4, 64])

随机对决头¶

class ding.model.StochasticDuelingHead(hidden_size: int, action_shape: int, layer_num: int = 1, a_layer_num: int | None = None, v_layer_num: int | None = None, activation: Module | None = ReLU(), norm_type: str | None = None, noise: bool | None = False, last_tanh: bool | None = True)[来源]¶

Overview:: 论文ACER (arxiv 1611.01224) 提出了Stochastic Dueling Network。也就是说，在连续动作空间中的决斗网络架构。
Interfaces:: __init__, forward.

Overview:

根据提供的参数初始化Stochastic DuelingHead层。

Arguments:

hidden_size (int): 连接到 StochasticDuelingHead 的 MLP 的 hidden_size。
action_shape (int): 连续动作形状的数量，通常为整数值。
层数 (int): 网络中用于计算动作和值输出的默认层数。
a_layer_num (int): 用于计算动作输出的网络层数。默认值为 layer_num。
v_layer_num (int): 网络中用于计算值输出的层数。默认值为 layer_num。
activation (nn.Module): 在MLP中使用的激活函数类型。如果为None，则默认将激活函数设置为nn.ReLU()。默认值为None。
norm_type (str): 使用的归一化类型。有关更多详细信息，请参见 ding.torch_utils.network.fc_block。默认值为 None。
噪声 (bool): 是否在Q网络的MLP中使用NoiseLinearLayer作为layer_fn。默认值为False。
last_tanh (bool): 如果为 True，则对动作应用 tanh。默认值为 True。

forward(s: Tensor, a: Tensor, mu: Tensor, sigma: Tensor, sample_size: int = 10) → Dict[str, Tensor][来源]¶

Overview:

使用编码的嵌入张量运行MLP，并使用StochasticDuelingHead返回预测字典。

Arguments:

s (torch.Tensor): 包含输入嵌入的张量。
a (torch.Tensor): 原始的连续行为动作。
mu (torch.Tensor): 当前时间步actor head的mu高斯重参数化输出。
sigma (torch.Tensor): 当前时间步actor head的sigma高斯重参数化输出。
样本大小 (int): 计算Q值时用于连续动作的样本数量。

Returns:

输出 (Dict): 包含关键词 q_value (torch.Tensor) 和 v_value (torch.Tensor) 的字典。

Shapes:

s: \((B, N)\), 其中 B = batch_size 和 N = hidden_size.
a: \((B, A)\), 其中 A = action_size.
mu: \((B, A)\).
sigma: \((B, A)\).
q_value: \((B, 1)\).
v_value: \((B, 1)\).

Examples:

>>> head = StochasticDuelingHead(64, 64)
>>> inputs = torch.randn(4, 64)
>>> a = torch.randn(4, 64)
>>> mu = torch.randn(4, 64)
>>> sigma = torch.ones(4, 64)
>>> outputs = head(inputs, a, mu, sigma)
>>> assert isinstance(outputs, dict)
>>> assert outputs['q_value'].shape == torch.Size([4, 1])
>>> assert outputs['v_value'].shape == torch.Size([4, 1])

分支头¶

class ding.model.BranchingHead(hidden_size: int, num_branches: int = 0, action_bins_per_branch: int = 2, layer_num: int = 1, a_layer_num: int | None = None, v_layer_num: int | None = None, norm_type: str | None = None, activation: Module | None = ReLU(), noise: bool | None = False)[source]¶

Overview:: BranchingHead 用于生成具有不同分支的Q值。该模块用于分支DQN。
Interfaces:: __init__, forward.

__init__(hidden_size: int, num_branches: int = 0, action_bins_per_branch: int = 2, layer_num: int = 1, a_layer_num: int | None = None, v_layer_num: int | None = None, norm_type: str | None = None, activation: Module | None = ReLU(), noise: bool | None = False) → None[源代码]¶

Overview:

根据提供的参数初始化BranchingHead层。这个头通过允许每个单独动作的一定程度的独立性，实现了网络输出数量与自由度数量的线性增加。因此，这个头适用于高维动作空间。

Arguments:

hidden_size (int): 连接到 BranchingHead 的 MLP 的 hidden_size。
num_branches (int): 分支的数量，等同于动作的维度。
action_bins_per_branch (:obj:int): 每个维度中的动作箱数量。
层数 (int): 网络中用于计算优势和值输出的层数。
a_layer_num (int): 用于计算优势输出的网络层数。
v_layer_num (int): 网络中用于计算值输出的层数。
output_size (int): 输出的数量。
norm_type (str): 使用的归一化类型。有关更多详细信息，请参见 ding.torch_utils.network.fc_block。默认值为 None。
activation (nn.Module): 在MLP中使用的激活函数类型。如果为None，则默认将激活函数设置为nn.ReLU()。默认值为None。
噪声 (bool): 是否在Q网络的MLP中使用NoiseLinearLayer作为layer_fn。默认值为False。

forward(x: Tensor) → Dict[源代码]¶

Overview:

使用编码的嵌入张量运行MLP，并使用BranchingHead返回预测字典。

Arguments:

x (torch.Tensor): 包含输入嵌入的张量。

Returns:

输出 (Dict): 包含关键字 logit (torch.Tensor) 的字典。

Shapes:

x: \((B, N)\), 其中 B = batch_size 和 N = hidden_size.
logit: \((B, M)\), 其中 M = output_size.

Examples:

>>> head = BranchingHead(64, 5, 2)
>>> inputs = torch.randn(4, 64)
>>> outputs = head(inputs)
>>> assert isinstance(outputs, dict) and outputs['logit'].shape == torch.Size([4, 5, 2])

回归头¶

class ding.model.RegressionHead(input_size: int, output_size: int, layer_num: int = 2, final_tanh: bool | None = False, activation: Module | None = ReLU(), norm_type: str | None = None, hidden_size: int | None = None)[source]¶

Overview:: RegressionHead 用于回归连续变量。该模块用于生成连续动作的Q值（DDPG评论家），或状态值（A2C/PPO），或直接预测连续动作（DDPG演员）。
Interfaces:: __init__, forward.

__init__(input_size: int, output_size: int, layer_num: int = 2, final_tanh: bool | None = False, activation: Module | None = ReLU(), norm_type: str | None = None, hidden_size: int | None = None) → None[源代码]¶

Overview:

根据提供的参数初始化RegressionHead层。

Arguments:

hidden_size (int): 连接到 RegressionHead 的 MLP 的 hidden_size。
output_size (int): 输出的数量。
层数 (int): 网络中用于计算Q值输出的层数。
final_tanh (bool): 如果为 True，则对输出应用 tanh。默认为 False。
activation (nn.Module): 在MLP中使用的激活函数类型。如果为None，则默认将激活函数设置为nn.ReLU()。默认值为None。
norm_type (str): 使用的归一化类型。有关更多详细信息，请参见 ding.torch_utils.network.fc_block。默认值为 None。

forward(x: Tensor) → Dict[source]¶

Overview:

使用编码的嵌入张量运行MLP，并使用RegressionHead返回预测字典。

Arguments:

x (torch.Tensor): 包含输入嵌入的张量。

Returns:

输出 (Dict): 包含关键字 pred (torch.Tensor) 的字典。

Shapes:

x: \((B, N)\), 其中 B = batch_size 和 N = hidden_size.
预测: \((B, M)\), 其中 M = output_size.

Examples:

>>> head = RegressionHead(64, 64)
>>> inputs = torch.randn(4, 64)
>>> outputs = head(inputs)
>>> assert isinstance(outputs, dict)
>>> assert outputs['pred'].shape == torch.Size([4, 64])

重参数化头¶

class ding.model.ReparameterizationHead(input_size: int, output_size: int, layer_num: int = 2, sigma_type: str | None = None, fixed_sigma_value: float | None = 1.0, activation: Module | None = ReLU(), norm_type: str | None = None, bound_type: str | None = None, hidden_size: int | None = None)[源代码]¶

Overview:: ReparameterizationHead 用于生成连续变量的高斯分布，该分布由 mu 和 sigma 参数化。该模块通常用于随机策略中，例如 PPO 和 SAC。
Interfaces:: __init__, forward.

Overview:

根据提供的参数初始化ReparameterizationHead层。

Arguments:

hidden_size (int): 连接到 ReparameterizationHead 的 MLP 的 hidden_size。
output_size (int): 输出的数量。
层数 (int): 网络中用于计算Q值输出的层数。
sigma_type (str): 使用的Sigma类型。在 ['fixed', 'independent', 'conditioned'] 中选择。默认是 None。
fixed_sigma_value (float): 当选择 fixed 类型时，张量 output['sigma'] 将填充此输入值。默认值为 None。
activation (nn.Module): 在MLP中使用的激活函数类型。如果为None，则默认将激活函数设置为nn.ReLU()。默认值为None。
norm_type (str): 使用的归一化类型。有关更多详细信息，请参见 ding.torch_utils.network.fc_block。默认值为 None。
bound_type (str): 应用于输出 mu 的边界类型。在 ['tanh', None] 中选择。默认值为 None。

forward(x: Tensor) → Dict[source]¶

Overview:

使用编码的嵌入张量通过ReparameterizationHead运行MLP并返回预测字典。

Arguments:

x (torch.Tensor): 包含输入嵌入的张量。

Returns:

输出 (Dict): 包含关键字 mu (torch.Tensor) 和 sigma (torch.Tensor) 的字典。

Shapes:

x: \((B, N)\), 其中 B = batch_size 和 N = hidden_size.
mu: \((B, M)\), 其中 M = output_size.
sigma: \((B, M)\).

Examples:

>>> head =  ReparameterizationHead(64, 64, sigma_type='fixed')
>>> inputs = torch.randn(4, 64)
>>> outputs = head(inputs)
>>> assert isinstance(outputs, dict)
>>> assert outputs['mu'].shape == torch.Size([4, 64])
>>> assert outputs['sigma'].shape == torch.Size([4, 64])

AttentionPolicyHead¶

class ding.model.AttentionPolicyHead[source]¶

Overview:: 交叉注意力类型的离散动作策略头，通常用于变量离散动作空间。
Interfaces:: __init__, forward.

__init__() → None[源代码]¶: 初始化内部模块状态，由nn.Module和ScriptModule共享。

forward(key: Tensor, query: Tensor) → Tensor[source]¶

Overview:

使用类似注意力的机制来结合键和查询张量，以输出离散动作的logit。

Arguments:

键 (torch.Tensor): 包含键嵌入的张量。
查询 (torch.Tensor): 包含查询嵌入的张量。

Returns:

logit (torch.Tensor): 包含输出离散动作logit的张量。

Shapes:

键: \((B, N, K)\), 其中 B = batch_size, N = possible discrete action choices 和 K = hidden_size.
查询: \((B, K)\).
logit: \((B, N)\).

Examples:

>>> head = AttentionPolicyHead()
>>> key = torch.randn(4, 5, 64)
>>> query = torch.randn(4, 64)
>>> logit = head(key, query)
>>> assert logit.shape == torch.Size([4, 5])

注意

在这一部分，我们假设key和query张量都已归一化。

多头¶

class ding.model.MultiHead(head_cls: type, hidden_size: int, output_size_list: SequenceType, **head_kwargs)[source]¶

Overview:: MultiHead 用于生成多个相似的结果。例如，我们可以结合 Distribution 和 MultiHead 来生成多离散动作空间的 logit。
Interfaces:: __init__, forward.

__init__(head_cls: type, hidden_size: int, output_size_list: SequenceType, **head_kwargs) → None[源代码]¶

Overview:

根据提供的参数初始化MultiHead层。

Arguments:

head_cls (type): 头部的类别，从[DuelingHead, DistributionHead, ‘’QuatileHead’’, …]中选择。
hidden_size (int): 连接到 Head 的 MLP 的 hidden_size。
output_size_list (int): 多离散动作的output_size序列，例如 [2, 3, 5]。
head_kwargs: (dict): 包含特定类参数的字典。

forward(x: Tensor) → Dict[source]¶

Overview:

使用编码的嵌入张量运行带有MultiHead的MLP并返回预测字典。

Arguments:

x (torch.Tensor): 包含输入嵌入的张量。

Returns:

输出 (Dict): 包含关键字 logit (torch.Tensor) 的字典，对应于每个 output 的 logit，每个 logit 可以通过 ['logit'][i] 访问。

Shapes:

x: \((B, N)\), 其中 B = batch_size 和 N = hidden_size.
logit: \((B, Mi)\), 其中 Mi = output_size 对应于输出 i.

Examples:

>>> head = MultiHead(DuelingHead, 64, [2, 3, 5], v_layer_num=2)
>>> inputs = torch.randn(4, 64)
>>> outputs = head(inputs)
>>> assert isinstance(outputs, dict)
>>> # output_size_list is [2, 3, 5] as set
>>> # Therefore each dim of logit is as follows
>>> outputs['logit'][0].shape
>>> torch.Size([4, 2])
>>> outputs['logit'][1].shape
>>> torch.Size([4, 3])
>>> outputs['logit'][2].shape
>>> torch.Size([4, 5])

独立正态分布¶

ding.model.independent_normal_dist(logits: List | Dict) → Distribution[source]¶

Overview:

将不同类型的logit转换为独立的正态分布。

Arguments:

logits (Union[List, Dict]): 要转换的logits。

Returns:

dist (torch.distributions.Distribution): 转换后的正态分布。

Examples:

>>> logits = [torch.randn(4, 5), torch.ones(4, 5)]
>>> dist = independent_normal_dist(logits)
>>> assert isinstance(dist, torch.distributions.Independent)
>>> assert isinstance(dist.base_dist, torch.distributions.Normal)
>>> assert dist.base_dist.loc.shape == torch.Size([4, 5])
>>> assert dist.base_dist.scale.shape == torch.Size([4, 5])

Raises:

TypeError: 如果logits的类型不是list或dict。

模板¶

请参考 ding/model/template 获取更多详细信息。

DQN¶

class ding.model.DQN(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], dueling: bool = True, head_hidden_size: int | None = None, head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, dropout: float | None = None, init_bias: float | None = None)[source]¶

Overview:: 深度Q网络（DQN）算法的神经网络结构和计算图，这是最经典的基于值的离散动作强化学习算法。DQN由两部分组成：encoder和head。encoder用于从各种观察中提取特征，而head用于计算每个动作维度的Q值。
Interfaces:: __init__, forward.

注意

当前的 DQN 支持两种类型的编码器：FCEncoder 和 ConvEncoder，两种类型的头部：DiscreteHead 和 DuelingHead。你可以通过继承这个类来自定义你自己的编码器或头部。

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], dueling: bool = True, head_hidden_size: int | None = None, head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, dropout: float | None = None, init_bias: float | None = None) → None[源代码]¶

Overview:

根据相应的输入参数初始化DQN（编码器+头部）模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测空间形状，例如 8 或 [4, 84, 84]。
action_shape (Union[int, SequenceType]): 动作空间形状，例如 6 或 [2, 3, 3]。
encoder_hidden_size_list (SequenceType): 传递给Encoder的hidden_size集合，最后一个元素必须匹配head_hidden_size。
决斗 (Optional[bool]): 是否选择 DuelingHead 或 DiscreteHead (默认)。
head_hidden_size (Optional[int]): 头部网络的hidden_size，默认为None，然后它将被设置为encoder_hidden_size_list的最后一个元素。
head_layer_num (int): 用于计算Q值输出的头部网络中的层数。
activation (Optional[nn.Module]): 网络中的激活函数类型，如果 None 则默认设置为 nn.ReLU()。
norm_type (Optional[str]): 网络中的归一化类型，详情请参见 ding.torch_utils.fc_block。你可以选择 [‘BN’, ‘IN’, ‘SyncBN’, ‘LN’] 中的一个。
dropout (Optional[float]): dropout层的dropout率。如果None，则默认禁用dropout层。
init_bias (Optional[float]): 头部网络中最后一层偏置的初始值。

forward(x: Tensor) → Dict[来源]¶

Overview:

DQN 前向计算图，输入观察张量以预测 q_value。

Arguments:

x (torch.Tensor): 输入的观测张量数据。

Returns:

输出 (Dict): DQN前向的输出，包括q_value。

ReturnsKeys:

logit (torch.Tensor): 每个可能动作维度的离散Q值输出。

Shapes:

x (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是 obs_shape
logit (torch.Tensor): \((B, M)\), 其中 B 是批量大小，M 是 action_shape

Examples:

>>> model = DQN(32, 6)  # arguments: 'obs_shape' and 'action_shape'
>>> inputs = torch.randn(4, 32)
>>> outputs = model(inputs)
>>> assert isinstance(outputs, dict) and outputs['logit'].shape == torch.Size([4, 6])

注意

为了保持一致性和兼容性，我们将与动作选择相关的网络所有输出命名为logit。

C51DQN¶

Overview:: C51DQN的神经网络结构和计算图，结合了分布强化学习和DQN。您可以参考https://arxiv.org/pdf/1707.06887.pdf了解更多详情。C51DQN由encoder和head组成。encoder用于提取观察的特征，head用于计算Q值的分布。
Interfaces:: __init__, forward

注意

当前的C51DQN支持两种类型的编码器：FCEncoder 和 ConvEncoder。

Overview:

根据相应的输入参数初始化C51模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测空间形状，例如 8 或 [4, 84, 84]。
action_shape (Union[int, SequenceType]): 动作空间形状，例如 6 或 [2, 3, 3]。
encoder_hidden_size_list (SequenceType): 传递给Encoder的hidden_size集合，最后一个元素必须匹配head_hidden_size。
head_hidden_size (Optional[int]): 头部网络的hidden_size，默认为None，然后它将被设置为encoder_hidden_size_list的最后一个元素。
head_layer_num (int): 用于计算Q值输出的头部网络中的层数。
activation (Optional[nn.Module]): 网络中的激活函数类型，如果 None 则默认设置为 nn.ReLU()。
norm_type (Optional[str]): 网络中的归一化类型，详情请参见 ding.torch_utils.fc_block。你可以选择 [‘BN’, ‘IN’, ‘SyncBN’, ‘LN’] 中的一个。
v_min (Optional[float]): 分布支持的最小值，与特定环境的值（奖励的折现总和）尺度相关。默认为-10。
v_max (Optional[float]): 分布支持的最大值，与特定环境的值（奖励的折现总和）规模相关。默认值为10。
n_atom (Optional[int]): 预测分布中的原子数量，论文中默认值为51，你也可以尝试其他值，例如301。

forward(x: Tensor) → Dict[来源]¶

Overview:

C51DQN 前向计算图，输入观测张量以预测 q_value 及其分布。

Arguments:

x (torch.Tensor): 输入的观测张量数据。

Returns:

输出 (Dict): DQN前向传播的输出，包括q_value和分布。

ReturnsKeys:

logit (torch.Tensor): 每个可能动作维度的离散Q值输出。
分布 (torch.Tensor): Q值离散化分布，即每个均匀间隔的原子Q值的概率，例如将[-10, 10]分成51个均匀间隔。

Shapes:

x (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是 head_hidden_size。
logit (torch.Tensor): \((B, M)\), 其中 M 是 action_shape。
分布(torch.Tensor): \((B, M, P)\), 其中 P 是 n_atom。

Examples:

>>> model = C51DQN(128, 64)  # arguments: 'obs_shape' and 'action_shape'
>>> inputs = torch.randn(4, 128)
>>> outputs = model(inputs)
>>> assert isinstance(outputs, dict)
>>> # default head_hidden_size: int = 64,
>>> assert outputs['logit'].shape == torch.Size([4, 64])
>>> # default n_atom: int = 51
>>> assert outputs['distribution'].shape == torch.Size([4, 64, 51])

注意

为了保持一致性和兼容性，我们将与动作选择相关的网络所有输出命名为logit。

注意

为了方便起见，我们建议原子的数量应为奇数，这样中间原子的值正好是Q值。

QRDQN¶

class ding.model.QRDQN(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], head_hidden_size: int | None = None, head_layer_num: int = 1, num_quantiles: int = 32, activation: Module | None = ReLU(), norm_type: str | None = None)[source]¶

Overview:: QRDQN的神经网络结构和计算图，结合了分布式强化学习和DQN。您可以参考《使用分位数回归的分布式强化学习》https://arxiv.org/pdf/1710.10044.pdf了解更多详情。
Interfaces:: __init__, forward

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], head_hidden_size: int | None = None, head_layer_num: int = 1, num_quantiles: int = 32, activation: Module | None = ReLU(), norm_type: str | None = None) → None[source]¶

Overview:

根据输入参数初始化QRDQN模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测的空间。
action_shape (Union[int, SequenceType]): 动作的空间。
encoder_hidden_size_list (SequenceType): 传递给 Encoder 的 hidden_size 的集合
head_hidden_size (Optional[int]): 传递给 Head 的 hidden_size。
head_layer_num (int): 网络中用于计算Q值输出的层数
num_quantiles (int): 预测分布中的分位数数量。
activation (Optional[nn.Module]):
在MLP中使用的激活函数类型，在layer_fn之后，如果None则默认设置为nn.ReLU()
norm_type (Optional[str]):
使用的归一化类型，详情请参见 ding.torch_utils.fc_block

forward(x: Tensor) → Dict[来源]¶

Overview:

使用观测张量来预测QRDQN的输出。通过QRDQN的MLPs前向设置进行参数更新。

Arguments:

x (torch.Tensor):
编码后的嵌入张量，形状为 (B, N=hidden_size)。

Returns:

outputs (Dict):
使用编码器和头部运行。返回结果预测字典。

ReturnsKeys:

logit (torch.Tensor): 与输入 x 大小相同的Logit张量。
q (torch.Tensor): Q 值张量，大小为 (B, N, num_quantiles)
tau (torch.Tensor): tau 张量的大小为 (B, N, 1)

Shapes:

x (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是 head_hidden_size。
logit (torch.FloatTensor): \((B, M)\), 其中 M 是动作形状。
tau (torch.Tensor): \((B, M, 1)\)

Examples:

>>> model = QRDQN(64, 64)
>>> inputs = torch.randn(4, 64)
>>> outputs = model(inputs)
>>> assert isinstance(outputs, dict)
>>> assert outputs['logit'].shape == torch.Size([4, 64])
>>> # default num_quantiles : int = 32
>>> assert outputs['q'].shape == torch.Size([4, 64, 32])
>>> assert outputs['tau'].shape == torch.Size([4, 32, 1])

IQN¶

class ding.model.IQN(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], head_hidden_size: int | None = None, head_layer_num: int = 1, num_quantiles: int = 32, quantile_embedding_size: int = 128, activation: Module | None = ReLU(), norm_type: str | None = None)[source]¶

Overview:: IQN的神经网络结构和计算图，结合了分布强化学习和DQN。您可以参考论文《隐式分位数网络用于分布强化学习》https://arxiv.org/pdf/1806.06923.pdf了解更多详情。
Interfaces:: __init__, forward

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], head_hidden_size: int | None = None, head_layer_num: int = 1, num_quantiles: int = 32, quantile_embedding_size: int = 128, activation: Module | None = ReLU(), norm_type: str | None = None) → None[source]¶

Overview:

根据输入参数初始化IQN模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测空间形状。
action_shape (Union[int, SequenceType]): 动作空间形状。
encoder_hidden_size_list (SequenceType): 传递给 Encoder 的 hidden_size 的集合
head_hidden_size (Optional[int]): 传递给 Head 的 hidden_size。
head_layer_num (int): 网络中用于计算Q值输出的层数
num_quantiles (int): 预测分布中的分位数数量。
activation (Optional[nn.Module]):
在MLP中使用的激活函数类型，在layer_fn之后，如果None则默认设置为nn.ReLU()
norm_type (Optional[str]):
使用的归一化类型，详情请参见ding.torch_utils.fc_block。

forward(x: Tensor) → Dict[来源]¶

Overview:

使用编码的嵌入张量来预测IQN的输出。通过IQN的MLPs前向设置更新参数。

Arguments:

x (torch.Tensor):
编码后的嵌入张量，形状为 (B, N=hidden_size)。

Returns:

outputs (Dict):
使用编码器和头部运行。返回结果预测字典。

ReturnsKeys:

logit (torch.Tensor): 与输入 x 大小相同的Logit张量。
q (torch.Tensor): Q 值张量，大小为 (num_quantiles, N, B)
分位数 (torch.Tensor): 大小为 (quantile_embedding_size, 1) 的分位数张量

Shapes:

x (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是 head_hidden_size。
logit (torch.FloatTensor): \((B, M)\), 其中 M 是动作形状
分位数 (torch.Tensor): \((P, 1)\), 其中 P 是分位数嵌入大小。

Examples:

>>> model = IQN(64, 64) # arguments: 'obs_shape' and 'action_shape'
>>> inputs = torch.randn(4, 64)
>>> outputs = model(inputs)
>>> assert isinstance(outputs, dict)
>>> assert outputs['logit'].shape == torch.Size([4, 64])
>>> # default num_quantiles: int = 32
>>> assert outputs['q'].shape == torch.Size([32, 4, 64]
>>> # default quantile_embedding_size: int = 128
>>> assert outputs['quantiles'].shape == torch.Size([128, 1])

FQF¶

class ding.model.FQF(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], head_hidden_size: int | None = None, head_layer_num: int = 1, num_quantiles: int = 32, quantile_embedding_size: int = 128, activation: Module | None = ReLU(), norm_type: str | None = None)[source]¶

Overview:: FQF的神经网络结构和计算图，结合了分布强化学习和DQN。您可以参考论文《完全参数化的分位数函数用于分布强化学习》https://arxiv.org/pdf/1911.02140.pdf了解更多详情。
Interface:: __init__, forward

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], head_hidden_size: int | None = None, head_layer_num: int = 1, num_quantiles: int = 32, quantile_embedding_size: int = 128, activation: Module | None = ReLU(), norm_type: str | None = None) → None[源代码]¶

Overview:

根据输入参数初始化FQF模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测空间形状。
action_shape (Union[int, SequenceType]): 动作空间形状。
encoder_hidden_size_list (SequenceType): 传递给 Encoder 的 hidden_size 的集合
head_hidden_size (Optional[int]): 传递给 Head 的 hidden_size。
head_layer_num (int): 网络中用于计算Q值输出的层数
num_quantiles (int): 预测分布中的分位数数量。
activation (Optional[nn.Module]):
在MLP中使用的激活函数类型，在layer_fn之后，如果None则默认设置为nn.ReLU()
norm_type (Optional[str]):
使用的归一化类型，详情请参见ding.torch_utils.fc_block。

forward(x: Tensor) → Dict[源代码]¶

Overview:

使用编码的嵌入张量来预测FQF的输出。参数更新使用FQF的MLPs前向设置。

Arguments:

x (torch.Tensor):
编码后的嵌入张量，形状为 (B, N=hidden_size)。

Returns:

输出 (Dict): 包含关键词 logit (torch.Tensor), q (torch.Tensor), quantiles (torch.Tensor), quantiles_hats (torch.Tensor), q_tau_i (torch.Tensor), entropies (torch.Tensor) 的字典。

Shapes:

x: \((B, N)\), 其中 B 是批量大小，N 是 head_hidden_size。
logit: \((B, M)\), 其中 M 是动作形状。
q: \((B, num_quantiles, M)\).
分位数: \((B, num_quantiles + 1)\).
quantiles_hats: \((B, num_quantiles)\).
q_tau_i: \((B, num_quantiles - 1, M)\).
熵值: \((B, 1)\).

Examples:

>>> model = FQF(64, 64) # arguments: 'obs_shape' and 'action_shape'
>>> inputs = torch.randn(4, 64)
>>> outputs = model(inputs)
>>> assert isinstance(outputs, dict)
>>> assert outputs['logit'].shape == torch.Size([4, 64])
>>> # default num_quantiles: int = 32
>>> assert outputs['q'].shape == torch.Size([4, 32, 64])
>>> assert outputs['quantiles'].shape == torch.Size([4, 33])
>>> assert outputs['quantiles_hats'].shape == torch.Size([4, 32])
>>> assert outputs['q_tau_i'].shape == torch.Size([4, 31, 64])
>>> assert outputs['quantiles'].shape == torch.Size([4, 1])

BDQ¶

class ding.model.BDQ(obs_shape: int | SequenceType, num_branches: int = 0, action_bins_per_branch: int = 2, layer_num: int = 3, a_layer_num: int | None = None, v_layer_num: int | None = None, encoder_hidden_size_list: SequenceType = [128, 128, 64], head_hidden_size: int | None = None, norm_type: Module | None = None, activation: Module | None = ReLU())[source]¶

__init__(obs_shape: int | SequenceType, num_branches: int = 0, action_bins_per_branch: int = 2, layer_num: int = 3, a_layer_num: int | None = None, v_layer_num: int | None = None, encoder_hidden_size_list: SequenceType = [128, 128, 64], head_hidden_size: int | None = None, norm_type: Module | None = None, activation: Module | None = ReLU()) → None[来源]¶

Overview:

根据输入参数初始化BDQ（编码器+头部）模型。参考论文《深度强化学习的动作分支架构》<https://arxiv.org/pdf/1711.08946>

Arguments:

obs_shape (Union[int, SequenceType]): 观测空间形状，例如 8 或 [4, 84, 84]。
num_branches (int): 分支的数量，相当于动作维度，例如在mujoco的halfcheetah环境中为6。
action_bins_per_branch (int): 每个维度中的动作数量。
层数 (int): 网络中用于计算优势和值输出的层数。
a_layer_num (int): 用于计算优势输出的网络层数。
v_layer_num (int): 网络中用于计算值输出的层数。
encoder_hidden_size_list (SequenceType): 传递给Encoder的hidden_size集合，最后一个元素必须匹配head_hidden_size。
head_hidden_size (Optional[int]): 头部网络的hidden_size。
norm_type (Optional[str]): 网络中的归一化类型，更多详情请参见 ding.torch_utils.fc_block。
activation (Optional[nn.Module]): 网络中的激活函数类型如果 None 则默认设置为 nn.ReLU()

forward(x: Tensor) → Dict[源代码]¶

Overview:

BDQ 前向计算图，输入观察张量以预测 q_value。

Arguments:

x (torch.Tensor): 观测输入

Returns:

输出 (Dict): BDQ 前向输出，例如 q_value。

ReturnsKeys:

logit (torch.Tensor): 每个动作维度的离散Q值输出。

Shapes:

x (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是 obs_shape
logit (torch.FloatTensor): \((B, M)\), where B is batch size and M is
num_branches * action_bins_per_branch

Examples:

>>> model = BDQ(8, 5, 2)  # arguments: 'obs_shape', 'num_branches' and 'action_bins_per_branch'.
>>> inputs = torch.randn(4, 8)
>>> outputs = model(inputs)
>>> assert isinstance(outputs, dict) and outputs['logit'].shape == torch.Size([4, 5, 2])

彩虹DQN¶

Overview:: RainbowDQN的神经网络结构和计算图，结合了分布式强化学习和DQN。您可以参考论文《Rainbow: Combining Improvements in Deep Reinforcement Learning》https://arxiv.org/pdf/1710.02298.pdf了解更多详情。
Interfaces:: __init__, forward

注意

RainbowDQN 默认包含决斗架构。

Overview:

根据参数初始化彩虹模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测空间形状。
action_shape (Union[int, SequenceType]): 动作空间形状。
encoder_hidden_size_list (SequenceType): 传递给 Encoder 的 hidden_size 的集合
head_hidden_size (Optional[int]): 传递给 Head 的 hidden_size。
head_layer_num (int): 网络中用于计算Q值输出的层数
activation (Optional[nn.Module]): 在MLP中使用的激活函数类型，位于layer_fn之后，如果为None，则默认设置为nn.ReLU()
norm_type (Optional[str]): 使用的归一化类型，详情请参见 ding.torch_utils.fc_block
n_atom (Optional[int]): 预测分布中的原子数量。

forward(x: Tensor) → Dict[来源]

Overview:

使用观测张量来预测Rainbow输出。参数更新使用Rainbow的MLPs前向设置。

Arguments:

x (torch.Tensor):
编码后的嵌入张量，形状为 (B, N=hidden_size)。

Returns:

outputs (Dict):
使用RainbowHead设置运行MLP并返回结果预测字典。

ReturnsKeys:

logit (torch.Tensor): 与输入 x 大小相同的Logit张量。
分布 (torch.Tensor): 大小为 (B, N, n_atom) 的分布张量

Shapes:

x (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是 head_hidden_size。
logit (torch.FloatTensor): \((B, M)\), 其中 M 是动作形状。
分布(torch.FloatTensor): \((B, M, P)\), 其中 P 是 n_atom。

Examples:

>>> model = RainbowDQN(64, 64) # arguments: 'obs_shape' and 'action_shape'
>>> inputs = torch.randn(4, 64)
>>> outputs = model(inputs)
>>> assert isinstance(outputs, dict)
>>> assert outputs['logit'].shape == torch.Size([4, 64])
>>> # default n_atom: int =51
>>> assert outputs['distribution'].shape == torch.Size([4, 64, 51])

DRQN¶

class ding.model.DRQN(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], dueling: bool = True, head_hidden_size: int | None = None, head_layer_num: int = 1, lstm_type: str | None = 'normal', activation: Module | None = ReLU(), norm_type: str | None = None, res_link: bool = False)[源代码]¶

Overview:: DRQN（DQN + RNN = DRQN）算法的神经网络结构和计算图，这是用于序列数据和部分可观测环境的最常见的DQN变体。DRQN由三部分组成：encoder、head和rnn。encoder用于从各种观测中提取特征，rnn用于处理序列观测和其他数据，head用于计算每个动作维度的Q值。
Interfaces:: __init__, forward.

注意

当前的 DRQN 支持两种类型的编码器：FCEncoder 和 ConvEncoder，两种类型的头：DiscreteHead 和 DuelingHead，三种类型的 rnn：normal (LSTM with LayerNorm)、pytorch 和 gru。您可以通过继承此类来自定义自己的编码器、rnn 或头。

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], dueling: bool = True, head_hidden_size: int | None = None, head_layer_num: int = 1, lstm_type: str | None = 'normal', activation: Module | None = ReLU(), norm_type: str | None = None, res_link: bool = False) → None[source]¶

Overview:

根据相应的输入参数初始化DRQN模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测空间形状，例如 8 或 [4, 84, 84]。
action_shape (Union[int, SequenceType]): 动作空间形状，例如 6 或 [2, 3, 3]。
encoder_hidden_size_list (SequenceType): 传递给Encoder的hidden_size集合，最后一个元素必须匹配head_hidden_size。
决斗 (Optional[bool]): 是否选择 DuelingHead 或 DiscreteHead (默认)。
head_hidden_size (Optional[int]): 头部网络的hidden_size，默认为None，然后它将被设置为encoder_hidden_size_list的最后一个元素。
head_layer_num (int): 用于计算Q值输出的头部网络中的层数。
lstm_type (Optional[str]): RNN模块的类型，目前支持 ['normal', 'pytorch', 'gru']。
activation (Optional[nn.Module]): 网络中的激活函数类型，如果 None 则默认设置为 nn.ReLU()。
norm_type (Optional[str]): 网络中的归一化类型，详情请参见 ding.torch_utils.fc_block。你可以选择 [‘BN’, ‘IN’, ‘SyncBN’, ‘LN’] 中的一个。
res_link (bool): 是否启用残差链接，即单帧数据与序列数据之间的跳跃连接，默认为 False。

forward(inputs: Dict, inference: bool = False, saved_state_timesteps: list | None = None) → Dict[source]¶

Overview:

DRQN 前向计算图，输入观测张量以预测 q_value。

Arguments:

输入 (torch.Tensor): 输入数据的字典，包括观察值和先前的RNN状态。
推理: (:obj:’bool’): 是否启用推理前向模式，如果为True，我们展开一个时间步的转换，否则，我们展开整个序列的转换。
saved_state_timesteps: (:obj:’Optional[list]’): 当推理为False时，我们展开序列转换，然后我们将使用此列表来指示如何保存和返回隐藏状态。

ArgumentsKeys:

obs (torch.Tensor): 原始观测张量。
prev_state (list): 之前的RNN状态张量，其结构取决于lstm_type。

Returns:

输出 (Dict): DRQN前向的输出，包括logit（q_value）和下一个状态。

ReturnsKeys:

logit (torch.Tensor): 每个可能动作维度的离散Q值输出。
next_state (list): 下一个RNN状态张量，其结构取决于 lstm_type。

Shapes:

obs (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是 obs_shape
logit (torch.Tensor): \((B, M)\), 其中 B 是批量大小，M 是 action_shape

Examples:

>>> # Init input's Keys:
>>> prev_state = [[torch.randn(1, 1, 64) for __ in range(2)] for _ in range(4)] # B=4
>>> obs = torch.randn(4,64)
>>> model = DRQN(64, 64) # arguments: 'obs_shape' and 'action_shape'
>>> outputs = model({'obs': inputs, 'prev_state': prev_state}, inference=True)
>>> # Check outputs's Keys
>>> assert isinstance(outputs, dict)
>>> assert outputs['logit'].shape == (4, 64)
>>> assert len(outputs['next_state']) == 4
>>> assert all([len(t) == 2 for t in outputs['next_state']])
>>> assert all([t[0].shape == (1, 1, 64) for t in outputs['next_state']])

GTrXLDQN¶

class ding.model.GTrXLDQN(obs_shape: int | SequenceType, action_shape: int | SequenceType, head_layer_num: int = 1, att_head_dim: int = 16, hidden_size: int = 16, att_head_num: int = 2, att_mlp_num: int = 2, att_layer_num: int = 3, memory_len: int = 64, activation: Module | None = ReLU(), head_norm_type: str | None = None, dropout: float = 0.0, gru_gating: bool = True, gru_bias: float = 2.0, dueling: bool = True, encoder_hidden_size_list: SequenceType = [128, 128, 256], encoder_norm_type: str | None = None)[source]¶

Overview:: Gated Transformer-XL DQN算法的神经网络结构和计算图，这是DRQN的增强版本，使用Transformer-XL来提高长期序列建模能力。GTrXL-DQN由三部分组成：encoder、head和core。encoder用于从各种观察中提取特征，core用于处理序列观察和其他数据，head用于计算每个动作维度的Q值。
Interfaces:: __init__, forward, reset_memory, get_memory .

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType, head_layer_num: int = 1, att_head_dim: int = 16, hidden_size: int = 16, att_head_num: int = 2, att_mlp_num: int = 2, att_layer_num: int = 3, memory_len: int = 64, activation: Module | None = ReLU(), head_norm_type: str | None = None, dropout: float = 0.0, gru_gating: bool = True, gru_bias: float = 2.0, dueling: bool = True, encoder_hidden_size_list: SequenceType = [128, 128, 256], encoder_norm_type: str | None = None) → None[source]¶

Overview:: 根据相应的输入参数初始化GTrXLDQN模型。

提示

您可以参考ding.torch_utils.network.gtrxl中的GTrXl类以获取有关输入参数的更多详细信息。

Arguments:

obs_shape (Union[int, SequenceType]): 由Transformer使用。观测的空间。
action_shape (:obj:Union[int, SequenceType]): 由Head使用。动作的空间。
head_layer_num (int): 由Head使用。层数。
att_head_dim (int): 用于Transformer。
隐藏大小 (int): 由Transformer和Head使用。
att_head_num (int): 由Transformer使用。
att_mlp_num (int): 由Transformer使用。
att_layer_num (int): 用于Transformer。
memory_len (int): 由Transformer使用。
activation (Optional[nn.Module]): 由Transformer和Head使用。如果None，则默认设置为nn.ReLU()。
head_norm_type (Optional[str]): 由Head使用。使用的归一化类型，更多详情请参见 ding.torch_utils.fc_block。
dropout (bool): 由Transformer使用。
gru_gating (bool): 由Transformer使用。
gru_bias (float): 由Transformer使用。
决斗 (bool): 由Head使用。使头部进行决斗。
encoder_hidden_size_list(SequenceType): 由编码器使用。如果使用自定义卷积编码器，则为hidden_size的集合。
encoder_norm_type (Optional[str]): 由编码器使用。指定使用的归一化类型，更多详情请参见 ding.torch_utils.fc_block。

forward(x: Tensor) → Dict[源代码]¶

Overview:

让输入张量依次通过GTrXl和Head。

Arguments:

x (torch.Tensor): 输入张量的形状为 (seq_len, bs, obs_shape)。

Returns:

输出 (Dict): 使用 GTrXL 和 DiscreteHead 设置运行并返回结果预测字典。

ReturnKeys:

logit (torch.Tensor): 每个动作维度的离散Q值输出，形状为 (B, action_space)。
memory (torch.Tensor): 大小为 (bs x layer_num+1 x memory_len x embedding_dim) 的内存张量。
transformer_out (torch.Tensor): 转换器的输出张量，大小与输入 x 相同。

Examples:

>>> # Init input's Keys:
>>> obs_dim, seq_len, bs, action_dim = 128, 64, 32, 4
>>> obs = torch.rand(seq_len, bs, obs_dim)
>>> model = GTrXLDQN(obs_dim, action_dim)
>>> outputs = model(obs)
>>> assert isinstance(outputs, dict)

get_memory() → Tensor | None[source]¶

Overview:

返回GTrXL的内存。

Returns:

内存: (Optional[torch.Tensor]): 输出内存，如果内存尚未初始化则为None，其形状为(layer_num, memory_len, bs, embedding_dim)。

reset_memory(batch_size: int | None = None, state: Tensor | None = None) → None[source]¶

Overview:

清除或重置GTrXL的内存。

Arguments:

batch_size (Optional[int]): 训练批次中的样本数量。
状态 (Optional[torch.Tensor]): 输入的内存数据，其形状为 (layer_num, memory_len, bs, embedding_dim)。

PG¶

class ding.model.PG(obs_shape: int | SequenceType, action_shape: int | SequenceType, action_space: str = 'discrete', encoder_hidden_size_list: SequenceType = [128, 128, 64], head_hidden_size: int | None = None, head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None)[source]¶

Overview:: 与策略梯度（Policy Gradient, PG）相关的算法的神经网络和计算图（https://proceedings.neurips.cc/paper/1999/file/464d828b85b0bed98e80ade0a5c43b0f-Paper.pdf）。PG模型由两部分组成：编码器和头部。编码器用于从各种观察中提取特征。头部用于预测相应的动作对数。
Interface:: __init__, forward.

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType, action_space: str = 'discrete', encoder_hidden_size_list: SequenceType = [128, 128, 64], head_hidden_size: int | None = None, head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None) → None[source]¶

Overview:

根据相应的输入参数初始化PG模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测空间形状，例如 8 或 [4, 84, 84]。
action_shape (Union[int, SequenceType]): 动作空间形状，例如 6 或 [2, 3, 3]。
action_space (str): 不同类型的动作空间，包括 [‘discrete’, ‘continuous’]，然后会实例化相应的头部，包括 DiscreteHead 和 ReparameterizationHead。
encoder_hidden_size_list (SequenceType): 传递给Encoder的hidden_size集合，最后一个元素必须匹配head_hidden_size。
head_hidden_size (Optional[int]): head 网络的 hidden_size，默认为 None，它必须与 encoder_hidden_size_list 的最后一个元素匹配。
head_layer_num (int): 用于计算动作的head网络中的层数。
activation (Optional[nn.Module]): 网络中的激活函数类型，如果 None 则默认设置为 nn.ReLU()。
norm_type (Optional[str]): 网络中的归一化类型，详情请参见 ding.torch_utils.fc_block。你可以选择 [‘BN’, ‘IN’, ‘SyncBN’, ‘LN’] 中的一个。

Examples:

>>> model = PG((4, 84, 84), 5)
>>> inputs = torch.randn(8, 4, 84, 84)
>>> outputs = model(inputs)
>>> assert isinstance(outputs, dict)
>>> assert outputs['logit'].shape == (8, 5)
>>> assert outputs['dist'].sample().shape == (8, )

forward(x: Tensor) → Dict[source]¶

Overview:

PG前向计算图，输入观测张量以预测策略分布。

Arguments:

x (torch.Tensor): 输入的观测张量数据。

Returns:

输出 (torch.distributions): 输出策略分布。如果动作空间是离散的，输出是分类分布；如果动作空间是连续的，输出是正态分布。

VAC¶

class ding.model.VAC(obs_shape: int | SequenceType, action_shape: int | SequenceType | EasyDict, action_space: str = 'discrete', share_encoder: bool = True, encoder_hidden_size_list: SequenceType = [128, 128, 64], actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, sigma_type: str | None = 'independent', fixed_sigma_value: int | None = 0.3, bound_type: str | None = None, encoder: Module | None = None, impala_cnn_encoder: bool = False)[source]¶

Overview:: 与（状态）值演员-评论家（VAC）相关的算法的神经网络和计算图，例如A2C/PPO/IMPALA。该模型现在支持离散、连续和混合动作空间。VAC由四部分组成：actor_encoder、critic_encoder、actor_head和critic_head。编码器用于从各种观察中提取特征。头部用于预测相应的值或动作逻辑。在高维观察空间（如2D图像）中，我们通常为actor_encoder和critic_encoder使用共享编码器。在低维观察空间（如1D向量）中，我们通常使用不同的编码器。
Interfaces:: __init__, forward, compute_actor, compute_critic, compute_actor_critic.

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType | EasyDict, action_space: str = 'discrete', share_encoder: bool = True, encoder_hidden_size_list: SequenceType = [128, 128, 64], actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, sigma_type: str | None = 'independent', fixed_sigma_value: int | None = 0.3, bound_type: str | None = None, encoder: Module | None = None, impala_cnn_encoder: bool = False) → None[source]¶

Overview:

根据相应的输入参数初始化VAC模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测空间形状，例如 8 或 [4, 84, 84]。
action_shape (Union[int, SequenceType]): 动作空间形状，例如 6 或 [2, 3, 3]。
action_space (str): 不同类型的动作空间，包括 ['discrete', 'continuous', 'hybrid']，然后会实例化相应的头部，包括 DiscreteHead, ReparameterizationHead, 和混合头部。
share_encoder (bool): 是否在actor和decoder之间共享观察编码器。
encoder_hidden_size_list (SequenceType): 传递给Encoder的hidden_size集合，最后一个元素用作actor_head和critic_head的输入大小。
actor_head_hidden_size (Optional[int]): actor_head 网络的 hidden_size，默认为 64，它是 actor_head 网络最后一层的隐藏大小。
actor_head_layer_num (int): 用于计算动作的actor_head网络中的层数。
critic_head_hidden_size (Optional[int]): critic_head 网络的 hidden_size，默认为 64，它是 critic_head 网络最后一层的隐藏大小。
critic_head_layer_num (int): 用于critic_head网络中的层数。
activation (Optional[nn.Module]): 网络中的激活函数类型，如果 None 则默认设置为 nn.ReLU()。
norm_type (Optional[str]): 网络中的归一化类型，详情请参见 ding.torch_utils.fc_block。你可以选择 [‘BN’, ‘IN’, ‘SyncBN’, ‘LN’] 中的一个。
sigma_type (Optional[str]): 连续动作空间中的sigma类型，详情请参见 ding.torch_utils.network.dreamer.ReparameterizationHead，在A2C/PPO中，默认值为 independent，表示状态独立的sigma参数。
fixed_sigma_value (Optional[int]): 如果 sigma_type 是 fixed，则使用此值作为 sigma。
bound_type (Optional[str]): 连续动作空间中动作绑定方法的类型，默认为 None，表示没有绑定。
编码器 (Optional[torch.nn.Module]): 编码器模块，默认为 None，您可以定义自己的编码器模块并将其传递给VAC以处理不同的观察空间。
impala_cnn_encoder (bool): 是否使用IMPALA CNN编码器，默认为 False。

compute_actor(x: Tensor) → Dict[source]¶

Overview:

VAC 前向计算图用于演员部分，输入观察张量以预测动作逻辑值。

Arguments:

x (torch.Tensor): 输入的观测张量数据。

Returns:

输出 (Dict): VAC前向计算图的输出字典，包括 logit。

ReturnsKeys:

logit (torch.Tensor): 预测的动作logit张量，对于离散动作空间，它将是相同维度的可能动作选择的实值范围张量，对于连续动作空间，它将是高斯分布的mu和sigma，mu和sigma的数量与连续动作的数量相同。混合动作空间是离散和连续动作空间的一种组合，因此logit将是一个包含action_type和action_args的字典。

Shapes:

logit (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是 action_shape

Examples:

>>> model = VAC(64, 64)
>>> inputs = torch.randn(4, 64)
>>> actor_outputs = model(inputs,'compute_actor')
>>> assert actor_outputs['logit'].shape == torch.Size([4, 64])

compute_actor_critic(x: Tensor) → Dict[来源]¶

Overview:

VAC 前向计算图，包括演员和评论家部分，输入观察张量以预测动作 logit 和状态值。

Arguments:

x (torch.Tensor): 输入的观测张量数据。

Returns:

输出 (Dict): VAC前向计算图的输出字典，包括演员和评论家，包含 logit 和 value。

ReturnsKeys:

logit (torch.Tensor): 预测的动作logit张量，对于离散动作空间，它将是相同维度的可能动作选择的实值范围张量，对于连续动作空间，它将是高斯分布的mu和sigma，mu和sigma的数量与连续动作的数量相同。混合动作空间是离散和连续动作空间的一种组合，因此logit将是一个包含action_type和action_args的字典。
值 (torch.Tensor): 预测的状态值张量。

Shapes:

logit (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是 action_shape
值 (torch.Tensor): \((B, )\), 其中 B 是批量大小，(B, 1) 被压缩为 (B, )。

Examples:

>>> model = VAC(64, 64)
>>> inputs = torch.randn(4, 64)
>>> outputs = model(inputs,'compute_actor_critic')
>>> assert critic_outputs['value'].shape == torch.Size([4])
>>> assert outputs['logit'].shape == torch.Size([4, 64])

注意

compute_actor_critic 接口旨在在共享编码器时节省计算，并返回组合的字典输出。

compute_critic(x: Tensor) → Dict[source]¶

Overview:

VAC 前向计算图用于评论家部分，输入观察张量以预测状态值。

Arguments:

x (torch.Tensor): 输入的观测张量数据。

Returns:

输出 (Dict): VAC前向计算图的输出字典，包括 value。

ReturnsKeys:

值 (torch.Tensor): 预测的状态值张量。

Shapes:

值 (torch.Tensor): \((B, )\), 其中 B 是批量大小，(B, 1) 被压缩为 (B, )。

Examples:

>>> model = VAC(64, 64)
>>> inputs = torch.randn(4, 64)
>>> critic_outputs = model(inputs,'compute_critic')
>>> assert critic_outputs['value'].shape == torch.Size([4])

forward(x: Tensor, mode: str) → Dict[source]¶

Overview:

VAC前向计算图，输入观测张量以预测状态值或动作逻辑。不同的mode将使用不同的网络模块进行前向传播，以获得不同的输出并节省计算。

Arguments:

x (torch.Tensor): 输入的观测张量数据。
模式 (str): 前向模式，所有模式都在这个类的开头定义。

Returns:

输出 (Dict): VAC前向计算图的输出字典，其键值因不同的mode而异。

Examples (Actor):

>>> model = VAC(64, 128)
>>> inputs = torch.randn(4, 64)
>>> actor_outputs = model(inputs,'compute_actor')
>>> assert actor_outputs['logit'].shape == torch.Size([4, 128])

Examples (Critic):

>>> model = VAC(64, 64)
>>> inputs = torch.randn(4, 64)
>>> critic_outputs = model(inputs,'compute_critic')
>>> assert actor_outputs['logit'].shape == torch.Size([4, 64])

Examples (Actor-Critic):

>>> model = VAC(64, 64)
>>> inputs = torch.randn(4, 64)
>>> outputs = model(inputs,'compute_actor_critic')
>>> assert critic_outputs['value'].shape == torch.Size([4])
>>> assert outputs['logit'].shape == torch.Size([4, 64])

DREAMERVAC¶

class ding.model.DREAMERVAC(action_shape: int | SequenceType | EasyDict, dyn_stoch=32, dyn_deter=512, dyn_discrete=32, actor_layers=2, value_layers=2, units=512, act='SiLU', norm='LayerNorm', actor_dist='normal', actor_init_std=1.0, actor_min_std=0.1, actor_max_std=1.0, actor_temp=0.1, action_unimix_ratio=0.01)[source]¶

Overview:: DreamerV3（状态）值演员-评论家（VAC）的神经网络和计算图。该模型现在支持离散和连续的动作空间。
Interfaces:: __init__, forward.

__init__(action_shape: int | SequenceType | EasyDict, dyn_stoch=32, dyn_deter=512, dyn_discrete=32, actor_layers=2, value_layers=2, units=512, act='SiLU', norm='LayerNorm', actor_dist='normal', actor_init_std=1.0, actor_min_std=0.1, actor_max_std=1.0, actor_temp=0.1, action_unimix_ratio=0.01) → None[source]¶

Overview:

根据参数初始化DREAMERVAC模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测空间形状，例如 8 或 [4, 84, 84]。
action_shape (Union[int, SequenceType]): 动作空间形状，例如 6 或 [2, 3, 3]。

MAVAC¶

class ding.model.MAVAC(agent_obs_shape: int | SequenceType, global_obs_shape: int | SequenceType, action_shape: int | SequenceType, agent_num: int, actor_head_hidden_size: int = 256, actor_head_layer_num: int = 2, critic_head_hidden_size: int = 512, critic_head_layer_num: int = 1, action_space: str = 'discrete', activation: Module | None = ReLU(), norm_type: str | None = None, sigma_type: str | None = 'independent', bound_type: str | None = None, encoder: Tuple[Module, Module] | None = None)[source]¶

Overview:: 与多智能体相关的（状态）价值演员-评论家（VAC）算法的神经网络和计算图，例如MAPPO(https://arxiv.org/abs/2103.01955)。该模型现在支持离散和连续动作空间。MAVAC由四部分组成：actor_encoder、critic_encoder、actor_head和critic_head。编码器用于从各种观察中提取特征。头部用于预测相应的价值或动作逻辑。
Interfaces:: __init__, forward, compute_actor, compute_critic, compute_actor_critic.

__init__(agent_obs_shape: int | SequenceType, global_obs_shape: int | SequenceType, action_shape: int | SequenceType, agent_num: int, actor_head_hidden_size: int = 256, actor_head_layer_num: int = 2, critic_head_hidden_size: int = 512, critic_head_layer_num: int = 1, action_space: str = 'discrete', activation: Module | None = ReLU(), norm_type: str | None = None, sigma_type: str | None = 'independent', bound_type: str | None = None, encoder: Tuple[Module, Module] | None = None) → None[source]¶

Overview:

根据参数初始化MAVAC模型。

Arguments:

agent_obs_shape (Union[int, SequenceType]): 单个代理的观察空间，例如 8 或 [4, 84, 84]。
global_obs_shape (Union[int, SequenceType]): 全局观测的空间，例如 8 或 [4, 84, 84]。
action_shape (Union[int, SequenceType]): 单个代理的动作空间形状，例如 6 或 [2, 3, 3]。
agent_num (int): 此参数暂时保留。此参数可能需要在后续模型更改时使用。
actor_head_hidden_size (Optional[int]): actor_head 网络的 hidden_size，默认值为 256，它必须与 agent_obs_shape 的最后一个元素匹配。
actor_head_layer_num (int): 用于计算动作的actor_head网络中的层数。
critic_head_hidden_size (Optional[int]): critic_head 网络的 hidden_size，默认值为 512，它必须与 global_obs_shape 的最后一个元素匹配。
critic_head_layer_num (int): 用于计算评论者神经网络Q值输出的网络层数。
action_space (Union[int, SequenceType]): 不同类型的动作空间，包括 ['discrete', 'continuous']，然后会实例化相应的头部，包括 DiscreteHead 和 ReparameterizationHead。
activation (Optional[nn.Module]): 在MLP中使用的激活函数类型，位于layer_fn之后，如果为None，则默认设置为nn.ReLU()。
norm_type (Optional[str]): 网络中的归一化类型，详情请参见 ding.torch_utils.fc_block。你可以选择 [‘BN’, ‘IN’, ‘SyncBN’, ‘LN’] 中的一个。
sigma_type (Optional[str]): 连续动作空间中的sigma类型，详情请参见 ding.torch_utils.network.dreamer.ReparameterizationHead，在MAPPO中，默认值为 independent，表示状态独立的sigma参数。
bound_type (Optional[str]): 连续动作空间中动作绑定方法的类型，默认为 None，表示没有绑定。
编码器 (Optional[Tuple[torch.nn.Module, torch.nn.Module]]): 编码器模块列表，默认为 None，你可以定义自己的演员和评论家编码器模块并将其传递给 MAVAC 来处理不同的观察空间。

compute_actor(x: Dict) → Dict[source]¶

Overview:

MAVAC 前向计算图用于演员部分，使用代理观察张量 x 预测动作逻辑值。

Arguments:

x (Dict): Input data dict with keys [‘agent_state’, ‘action_mask’(optional)].
- agent_state: (torch.Tensor): 每个代理的本地状态(obs)。
- action_mask(可选): (torch.Tensor): 当 action_space 是离散的时，需要提供 action_mask 来屏蔽非法动作。

Returns:

输出 (Dict): 前向计算图的输出字典，包括 logit。

ReturnsKeys:

logit (torch.Tensor): 预测的动作logit张量，对于离散动作空间，它将是相同维度的实值范围张量，表示可能的动作选择，而对于连续动作空间，它将是高斯分布的mu和sigma，mu和sigma的数量与连续动作的数量相同。

Shapes:

logit (torch.FloatTensor): \((B, M, N)\), 其中 B 是批量大小，N 是 action_shape，M 是 agent_num。

Examples:

>>> model = MAVAC(agent_obs_shape=64, global_obs_shape=128, action_shape=14)
>>> inputs = {
        'agent_state': torch.randn(10, 8, 64),
        'global_state': torch.randn(10, 8, 128),
        'action_mask': torch.randint(0, 2, size=(10, 8, 14))
    }
>>> actor_outputs = model(inputs,'compute_actor')
>>> assert actor_outputs['logit'].shape == torch.Size([10, 8, 14])

compute_actor_critic(x: Dict) → Dict[source]¶

Overview:

MAVAC前向计算图，包括演员和评论家部分，输入观察以预测动作logit和状态值。

Arguments:

x (Dict): 输入字典包含 agent_state, global_state 和其他相关信息。

Returns:

输出 (Dict): MAVAC前向计算图的输出字典，包括演员和评论家，包含 logit 和 value。

ReturnsKeys:

logit (torch.Tensor): Logit编码张量，与输入x的大小相同。
值 (torch.Tensor): Q值张量，大小与批量大小相同。

Shapes:

logit (torch.FloatTensor): \((B, M, N)\), 其中 B 是批量大小，N 是 action_shape，M 是 agent_num。
值 (torch.FloatTensor): \((B, M)\), 其中 B 是批量大小，M 是 agent_num。

Examples:

>>> model = MAVAC(64, 64)
>>> inputs = {
        'agent_state': torch.randn(10, 8, 64),
        'global_state': torch.randn(10, 8, 128),
        'action_mask': torch.randint(0, 2, size=(10, 8, 14))
    }
>>> outputs = model(inputs,'compute_actor_critic')
>>> assert outputs['value'].shape == torch.Size([10, 8])
>>> assert outputs['logit'].shape == torch.Size([10, 8, 14])

compute_critic(x: Dict) → Dict[source]¶

Overview:

MAVAC 前向计算图用于评论部分。使用全局观察张量 x 预测状态值。

Arguments:

x (Dict): Input data dict with keys [‘global_state’].
- global_state: (torch.Tensor): 全局状态(obs).

Returns:

输出 (Dict): MAVAC 前向计算图的输出字典，包括 value。

ReturnsKeys:

值 (torch.Tensor): 预测的状态值张量。

Shapes:

值 (torch.FloatTensor): \((B, M)\), 其中 B 是批量大小，M 是 agent_num。

Examples:

>>> model = MAVAC(agent_obs_shape=64, global_obs_shape=128, action_shape=14)
>>> inputs = {
        'agent_state': torch.randn(10, 8, 64),
        'global_state': torch.randn(10, 8, 128),
        'action_mask': torch.randint(0, 2, size=(10, 8, 14))
    }
>>> critic_outputs = model(inputs,'compute_critic')
>>> assert critic_outputs['value'].shape == torch.Size([10, 8])

forward(inputs: Tensor | Dict, mode: str) → Dict[source]¶

Overview:

MAVAC前向计算图，输入观测张量以预测状态值或动作逻辑。mode包括compute_actor、compute_critic、compute_actor_critic。不同的mode将通过不同的网络模块进行前向传播，以获得不同的输出并节省计算。

Arguments:

输入 (Dict): 输入字典，包括观察和相关信息，其键值因不同的模式而异。
模式 (str): 前向模式，所有模式都在这个类的开头定义。

Returns:

输出 (Dict): MAVAC前向计算图的输出字典，其键值因不同的mode而异。

Examples (Actor):

>>> model = MAVAC(agent_obs_shape=64, global_obs_shape=128, action_shape=14)
>>> inputs = {
        'agent_state': torch.randn(10, 8, 64),
        'global_state': torch.randn(10, 8, 128),
        'action_mask': torch.randint(0, 2, size=(10, 8, 14))
    }
>>> actor_outputs = model(inputs,'compute_actor')
>>> assert actor_outputs['logit'].shape == torch.Size([10, 8, 14])

Examples (Critic):

>>> model = MAVAC(agent_obs_shape=64, global_obs_shape=128, action_shape=14)
>>> inputs = {
        'agent_state': torch.randn(10, 8, 64),
        'global_state': torch.randn(10, 8, 128),
        'action_mask': torch.randint(0, 2, size=(10, 8, 14))
    }
>>> critic_outputs = model(inputs,'compute_critic')
>>> assert actor_outputs['value'].shape == torch.Size([10, 8])

Examples (Actor-Critic):

>>> model = MAVAC(64, 64)
>>> inputs = {
        'agent_state': torch.randn(10, 8, 64),
        'global_state': torch.randn(10, 8, 128),
        'action_mask': torch.randint(0, 2, size=(10, 8, 14))
    }
>>> outputs = model(inputs,'compute_actor_critic')
>>> assert outputs['value'].shape == torch.Size([10, 8, 14])
>>> assert outputs['logit'].shape == torch.Size([10, 8])

ContinuousQAC¶

class ding.model.ContinuousQAC(obs_shape: int | SequenceType, action_shape: int | SequenceType | EasyDict, action_space: str, twin_critic: bool = False, actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, encoder_hidden_size_list: SequenceType | None = None, share_encoder: bool | None = False)[source]¶

Overview:: 与Q值演员-评论家（QAC）相关的算法的神经网络和计算图，例如DDPG/TD3/SAC。该模型现在支持连续和混合动作空间。ContinuousQAC由四部分组成：actor_encoder、critic_encoder、actor_head和critic_head。编码器用于从各种观察中提取特征。头部用于预测相应的Q值或动作逻辑。在高维观察空间（如2D图像）中，我们通常为actor_encoder和critic_encoder使用共享编码器。在低维观察空间（如1D向量）中，我们通常使用不同的编码器。
Interfaces:: __init__, forward, compute_actor, compute_critic

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType | EasyDict, action_space: str, twin_critic: bool = False, actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, encoder_hidden_size_list: SequenceType | None = None, share_encoder: bool | None = False) → None[源代码]¶

Overview:

根据输入参数初始化ContinuousQAC模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测值的形状，例如 128, (156, )。
action_shape (Union[int, SequenceType, EasyDict]): 动作的形状，例如 4, (3, ), EasyDict({‘action_type_shape’: 3, ‘action_args_shape’: 4}).
action_space (str): 动作空间的类型，包括 [regression, reparameterization, hybrid], regression 用于 DDPG/TD3, reparameterization 用于 SAC 和 hybrid 用于 PADDPG.
twin_critic (bool): 是否使用双评论家，这是TD3中的一个技巧。
actor_head_hidden_size (Optional[int]): 传递给actor head的hidden_size。
actor_head_layer_num (int): 用于计算动作的actor网络中的层数。
critic_head_hidden_size (Optional[int]): 传递给critic头的hidden_size。
critic_head_layer_num (int): 用于计算Q值的评论家网络中的层数。
activation (Optional[nn.Module]): 在MLP中每个FC层后使用的激活函数类型，如果None则默认设置为nn.ReLU()。
norm_type (Optional[str]): 网络层（FC, Conv）后的归一化类型，详情请参见 ding.torch_utils.network。
encoder_hidden_size_list (SequenceType): 传递给Encoder的hidden_size集合，最后一个元素必须匹配head_hidden_size，此参数仅用于图像观察。
share_encoder (Optional[bool]): 是否在actor和critic之间共享编码器。

compute_actor(obs: Tensor) → Dict[str, Tensor | Dict[str, Tensor]][source]¶

Overview:

QAC 前向计算图用于演员部分，输入观察张量以预测动作或动作对数。

Arguments:

x (torch.Tensor): 输入的观测张量数据。

Returns:

输出 (Dict[str, Union[torch.Tensor, Dict[str, torch.Tensor]]]): 根据动作空间变化的Actor输出字典：regression, reparameterization, hybrid.

ReturnsKeys (regression):

动作 (torch.Tensor): 与 action_shape 大小相同的连续动作，通常在 DDPG/TD3 中使用。

ReturnsKeys (reparameterization):

logit (Dict[str, torch.Tensor]): 预测的重新参数化动作logit，通常在SAC中使用。它是一个包含两个张量的列表：mu 和 sigma。前者是高斯分布的均值，后者是高斯分布的标准差。

ReturnsKeys (hybrid):

logit (torch.Tensor): 预测的离散动作类型的logit，它将与action_type_shape具有相同的维度，即所有可能的离散动作类型。
action_args (torch.Tensor): 连续动作参数，大小与 action_args_shape 相同。

Shapes:

obs (torch.Tensor): \((B, N0)\), B 是批量大小，N0 对应于 obs_shape。
动作 (torch.Tensor): \((B, N1)\), B 是批量大小，N1 对应于 action_shape。
logit.mu (torch.Tensor): \((B, N1)\), B 是批量大小，N1 对应于 action_shape。
logit.sigma (torch.Tensor): \((B, N1)\), B 是批量大小。
logit (torch.Tensor): \((B, N2)\), B 是批量大小，N2 对应于 action_shape.action_type_shape。
action_args (torch.Tensor): \((B, N3)\), B 是批量大小，N3 对应于 action_shape.action_args_shape。

Examples:

>>> # Regression mode
>>> model = ContinuousQAC(64, 6, 'regression')
>>> obs = torch.randn(4, 64)
>>> actor_outputs = model(obs,'compute_actor')
>>> assert actor_outputs['action'].shape == torch.Size([4, 6])
>>> # Reparameterization Mode
>>> model = ContinuousQAC(64, 6, 'reparameterization')
>>> obs = torch.randn(4, 64)
>>> actor_outputs = model(obs,'compute_actor')
>>> assert actor_outputs['logit'][0].shape == torch.Size([4, 6])  # mu
>>> actor_outputs['logit'][1].shape == torch.Size([4, 6]) # sigma

compute_critic(inputs: Dict[str, Tensor]) → Dict[str, Tensor][来源]¶

Overview:

QAC前向计算图用于评论家部分，输入观察和动作张量以预测Q值。

Arguments:

输入 (Dict[str, torch.Tensor]): 输入数据的字典，包括 obs 和 action 张量，在混合动作空间中还包含 logit 和 action_args 张量。

ArgumentsKeys:

obs: (torch.Tensor): 观测张量数据，现在支持一批一维向量数据。
动作 (Union[torch.Tensor, Dict]): 与 action_shape 大小相同的连续动作。
logit (torch.Tensor): 离散动作的logit，仅在混合动作空间中存在。
action_args (torch.Tensor): 连续动作参数，仅在混合动作空间中存在。

Returns:

输出 (Dict[str, torch.Tensor]): QAC前向计算图的输出字典，包括 q_value。

ReturnKeys:

q_value (torch.Tensor): Q值张量，大小与批量大小相同。

Shapes:

obs (torch.Tensor): \((B, N1)\), 其中 B 是批量大小，N1 是 obs_shape。
logit (torch.Tensor): \((B, N2)\), B 是批量大小，N2 对应于 action_shape.action_type_shape。
action_args (torch.Tensor): \((B, N3)\), B 是批量大小，N3 对应于 action_shape.action_args_shape。
动作 (torch.Tensor): \((B, N4)\), 其中 B 是批量大小，N4 是 action_shape。
q_value (torch.Tensor): \((B, )\), 其中 B 是批量大小。

Examples:

>>> inputs = {'obs': torch.randn(4, 8), 'action': torch.randn(4, 1)}
>>> model = ContinuousQAC(obs_shape=(8, ),action_shape=1, action_space='regression')
>>> assert model(inputs, mode='compute_critic')['q_value'].shape == (4, )  # q value

forward(inputs: Tensor | Dict[str, Tensor], mode: str) → Dict[str, Tensor][来源]¶

Overview:

QAC前向计算图，输入观测张量以预测Q值或动作逻辑值。不同的mode将使用不同的网络模块进行前向传播，以获得不同的输出并节省计算。

Arguments:

输入 (Union[torch.Tensor, Dict[str, torch.Tensor]]): 用于前向计算图的输入数据，对于 compute_actor，它是观察张量，对于 compute_critic，它是包含观察和动作张量的字典数据。
模式 (str): 前向模式，所有模式都在这个类的开头定义。

Returns:

输出 (Dict[str, torch.Tensor]): QAC前向计算图的输出字典，其键值在不同的前向模式中有所不同。

Examples (Actor):

>>> # Regression mode
>>> model = ContinuousQAC(64, 6, 'regression')
>>> obs = torch.randn(4, 64)
>>> actor_outputs = model(obs,'compute_actor')
>>> assert actor_outputs['action'].shape == torch.Size([4, 6])
>>> # Reparameterization Mode
>>> model = ContinuousQAC(64, 6, 'reparameterization')
>>> obs = torch.randn(4, 64)
>>> actor_outputs = model(obs,'compute_actor')
>>> assert actor_outputs['logit'][0].shape == torch.Size([4, 6])  # mu
>>> actor_outputs['logit'][1].shape == torch.Size([4, 6]) # sigma

Examples (Critic):

>>> inputs = {'obs': torch.randn(4, 8), 'action': torch.randn(4, 1)}
>>> model = ContinuousQAC(obs_shape=(8, ),action_shape=1, action_space='regression')
>>> assert model(inputs, mode='compute_critic')['q_value'].shape == (4, )  # q value

离散QAC¶

class ding.model.DiscreteQAC(obs_shape: int | SequenceType, action_shape: int | SequenceType, twin_critic: bool = False, actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, encoder_hidden_size_list: SequenceType | None = None, share_encoder: bool | None = False)[source]¶

Overview:: 与离散动作Q值演员-评论家（QAC）相关的算法的神经网络和计算图，例如DiscreteSAC。该模型目前仅支持离散动作空间。DiscreteQAC由四个部分组成：actor_encoder、critic_encoder、actor_head和critic_head。编码器用于从各种观察中提取特征。头部用于预测相应的Q值或动作逻辑。在高维观察空间（如2D图像）中，我们通常为actor_encoder和critic_encoder使用共享编码器。在低维观察空间（如1D向量）中，我们通常使用不同的编码器。
Interfaces:: __init__, forward, compute_actor, compute_critic

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType, twin_critic: bool = False, actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, encoder_hidden_size_list: SequenceType | None = None, share_encoder: bool | None = False) → None[源代码]¶

Overview:

根据输入参数初始化DiscreteQAC模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测值的形状，例如 128, (156, )。
action_shape (Union[int, SequenceType, EasyDict]): 动作的形状，例如 4, (3, )。
twin_critic (bool): 是否使用双评论家。
actor_head_hidden_size (Optional[int]): 传递给actor head的hidden_size。
actor_head_layer_num (int): 用于计算动作的actor网络中的层数。
critic_head_hidden_size (Optional[int]): 传递给critic头的hidden_size。
critic_head_layer_num (int): 用于计算Q值的评论家网络中的层数。
activation (Optional[nn.Module]): 在MLP中每个FC层后使用的激活函数类型，如果None则默认设置为nn.ReLU()。
norm_type (Optional[str]): 网络层（FC, Conv）后的归一化类型，详情请参见 ding.torch_utils.network。
encoder_hidden_size_list (SequenceType): 传递给Encoder的hidden_size集合，最后一个元素必须匹配head_hidden_size，此参数仅用于图像观察。
share_encoder (Optional[bool]): 是否在actor和critic之间共享编码器。

compute_actor(inputs: Tensor) → Dict[str, Tensor][source]¶

Overview:

QAC 前向计算图用于演员部分，输入观察张量以预测动作或动作对数。

Arguments:

输入 (torch.Tensor): 输入的观测张量数据。

Returns:

输出 (Dict[str, torch.Tensor]): QAC前向计算图的输出字典，包括离散动作的logit。

ReturnsKeys:

logit (torch.Tensor): 预测的离散动作类型的logit，它将与action_shape具有相同的维度，即所有可能的离散动作选择。

Shapes:

输入 (torch.Tensor): \((B, N0)\), B 是批量大小，N0 对应于 obs_shape。
logit (torch.Tensor): \((B, N2)\), B 是批量大小，N2 对应于 action_shape。

Examples:

>>> model = DiscreteQAC(64, 6)
>>> obs = torch.randn(4, 64)
>>> actor_outputs = model(obs,'compute_actor')
>>> assert actor_outputs['logit'].shape == torch.Size([4, 6])

compute_critic(inputs: Tensor) → Dict[str, Tensor][source]¶

Overview:

QAC 前向计算图用于评论家部分，输入观察以预测每个可能的离散动作选择的 Q 值。

Arguments:

输入 (torch.Tensor): 输入的观测张量数据。

Returns:

输出 (Dict[str, torch.Tensor]): QAC前向计算图的输出字典，包括每个可能离散动作选择的q_value。

ReturnKeys:

q_value (torch.Tensor): 每个可能的离散动作选择的预测Q值，它将与action_shape具有相同的维度，并用于计算损失。

Shapes:

obs (torch.Tensor): \((B, N1)\), 其中 B 是批量大小，N1 是 obs_shape。
q_value (torch.Tensor): \((B, N2)\), 其中 B 是批量大小，N2 是 action_shape。

Examples:

>>> model = DiscreteQAC(64, 6, twin_critic=False)
>>> obs = torch.randn(4, 64)
>>> actor_outputs = model(obs,'compute_critic')
>>> assert actor_outputs['q_value'].shape == torch.Size([4, 6])

forward(inputs: Tensor, mode: str) → Dict[str, Tensor][来源]¶

Overview:

QAC前向计算图，输入观测张量以预测Q值或动作逻辑值。不同的mode将使用不同的网络模块进行前向传播，以获得不同的输出并节省计算。

Arguments:

输入 (torch.Tensor): 输入的观测张量数据。
模式 (str): 前向模式，所有模式都在这个类的开头定义。

Returns:

输出 (Dict[str, torch.Tensor]): QAC前向计算图的输出字典，其键值在不同的前向模式中有所不同。

Examples (Actor):

>>> model = DiscreteQAC(64, 6)
>>> obs = torch.randn(4, 64)
>>> actor_outputs = model(obs,'compute_actor')
>>> assert actor_outputs['logit'].shape == torch.Size([4, 6])

Examples(Critic):

>>> model = DiscreteQAC(64, 6, twin_critic=False)
>>> obs = torch.randn(4, 64)
>>> actor_outputs = model(obs,'compute_critic')
>>> assert actor_outputs['q_value'].shape == torch.Size([4, 6])

ContinuousMAQAC¶

class ding.model.ContinuousMAQAC(agent_obs_shape: int | SequenceType, global_obs_shape: int | SequenceType, action_shape: int | SequenceType | EasyDict, action_space: str, twin_critic: bool = False, actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None)[source]¶

Overview:: 与连续动作多智能体Q值演员-评论家（MAQAC）模型相关的算法的神经网络和计算图。该模型由演员和评论家组成，其中演员是一个MLP网络，评论家也是一个MLP网络。演员网络用于预测动作概率分布，评论家网络用于预测状态-动作对的Q值。
Interfaces:: __init__, forward, compute_actor, compute_critic

__init__(agent_obs_shape: int | SequenceType, global_obs_shape: int | SequenceType, action_shape: int | SequenceType | EasyDict, action_space: str, twin_critic: bool = False, actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None) → None[source]¶

Overview:

根据参数初始化QAC模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测的空间。
action_shape (Union[int, SequenceType, EasyDict]): 动作的空间，例如 4, (3, )
action_space (str): 选择 regression 还是 reparameterization。
twin_critic (bool): 是否包含双评论家。
actor_head_hidden_size (Optional[int]): 传递给actor-nn的Head的hidden_size。
actor_head_layer_num (int): 用于网络中计算Q值输出的层数，用于actor的神经网络。
critic_head_hidden_size (Optional[int]): 传递给critic-nn的Head的hidden_size。
critic_head_layer_num (int): 用于计算Q值输出的网络层数，用于critic的神经网络。
activation (Optional[nn.Module]): 在MLP中使用的激活函数类型，位于layer_fn之后，如果为None，则默认设置为nn.ReLU()
norm_type (Optional[str]): 使用的归一化类型，详情请参见 ding.torch_utils.fc_block。

compute_actor(inputs: Dict) → Dict[source]¶

Overview:

使用观察张量来预测动作的logits。

Arguments:

inputs (Dict[str, torch.Tensor]): The input dict tensor data, has keys:
- agent_state (torch.Tensor): 代理的观察张量数据，形状为 \((B, A, N0)\)，其中 B 是批量大小，A 是代理数量。N0 对应于 agent_obs_shape。

Returns:

输出 (Dict): 网络前向的输出。

ReturnKeys (action_space == 'regression'):

动作 (torch.Tensor): 与 action_shape 大小相同的动作张量。

ReturnKeys (action_space == 'reparameterization'):

logit (list): 2个元素，每个元素的形状为 \((B, A, N3)\)，其中 B 是批量大小，A 是代理数量。N3 对应于 action_shape。

Examples:

>>> B = 32
>>> agent_obs_shape = 216
>>> global_obs_shape = 264
>>> agent_num = 8
>>> action_shape = 14
>>> act_space = 'reparameterization'  # 'regression'
>>> data = {
>>>     'agent_state': torch.randn(B, agent_num, agent_obs_shape),
>>> }
>>> model = ContinuousMAQAC(agent_obs_shape, global_obs_shape, action_shape, act_space, twin_critic=False)
>>> if action_space == 'regression':
>>>     action = model.compute_actor(data)['action']
>>> elif action_space == 'reparameterization':
>>>     (mu, sigma) = model.compute_actor(data)['logit']

compute_critic(inputs: Dict) → Dict[source]¶

Overview:

使用观察张量和动作张量来预测Q值。

Arguments:

inputs (Dict[str, torch.Tensor]): The input dict tensor data, has keys:
- obs (Dict[str, torch.Tensor]): The input dict tensor data, has keys:
  
  agent_state (torch.Tensor): 代理的观察张量数据，形状为 \((B, A, N0)\)，其中 B 是批量大小，A 是代理数量。N0 对应于 agent_obs_shape。
  
  global_state (torch.Tensor): 全局观测张量数据，形状为 \((B, A, N1)\)，其中 B 是批量大小，A 是代理数量。N1 对应于 global_obs_shape。
  
  action_mask (torch.Tensor): 动作掩码张量数据，形状为 \((B, A, N2)\)，其中 B 是批量大小，A 是代理数量。N2 对应于 action_shape。
- action (torch.Tensor): 动作张量数据，形状为 \((B, A, N3)\)，其中 B 是批量大小，A 是代理数量。N3 对应于 action_shape。

Returns:

输出 (Dict): 网络前向的输出。

ReturnKeys (twin_critic=True):

q_value (list): 2个元素，每个元素的形状为 \((B, A)\)，其中B是批量大小，A是代理数量。

ReturnKeys (twin_critic=False):

q_value (torch.Tensor): \((B, A)\), 其中 B 是批量大小，A 是代理数量。

Examples:

>>> B = 32
>>> agent_obs_shape = 216
>>> global_obs_shape = 264
>>> agent_num = 8
>>> action_shape = 14
>>> act_space = 'reparameterization'  # 'regression'
>>> data = {
>>>     'obs': {
>>>         'agent_state': torch.randn(B, agent_num, agent_obs_shape),
>>>         'global_state': torch.randn(B, agent_num, global_obs_shape),
>>>         'action_mask': torch.randint(0, 2, size=(B, agent_num, action_shape))
>>>     },
>>>     'action': torch.randn(B, agent_num, squeeze(action_shape))
>>> }
>>> model = ContinuousMAQAC(agent_obs_shape, global_obs_shape, action_shape, act_space, twin_critic=False)
>>> value = model.compute_critic(data)['q_value']

forward(inputs: Tensor | Dict, mode: str) → Dict[来源]¶

Overview:

使用观察和动作张量在compute_actor或compute_critic模式下预测输出。

Arguments:

inputs (Dict[str, torch.Tensor]): The input dict tensor data, has keys:
- obs (Dict[str, torch.Tensor]): The input dict tensor data, has keys:
  
  agent_state (torch.Tensor): 代理的观察张量数据，形状为 \((B, A, N0)\)，其中 B 是批量大小，A 是代理数量。N0 对应于 agent_obs_shape。
  
  global_state (torch.Tensor): 全局观测张量数据，形状为 \((B, A, N1)\)，其中 B 是批量大小，A 是代理数量。N1 对应于 global_obs_shape。
  
  action_mask (torch.Tensor): 动作掩码张量数据，形状为 \((B, A, N2)\)，其中 B 是批量大小，A 是代理数量。N2 对应于 action_shape。
- action (torch.Tensor): 动作张量数据，形状为 \((B, A, N3)\)，其中 B 是批量大小，A 是代理数量。N3 对应于 action_shape。
模式 (str): 前向模式的名称。

Returns:

输出 (Dict): 网络前向的输出，其键值将根据不同的 mode, twin_critic, action_space 而有所不同。

Examples:

>>> B = 32
>>> agent_obs_shape = 216
>>> global_obs_shape = 264
>>> agent_num = 8
>>> action_shape = 14
>>> act_space = 'reparameterization'  # regression
>>> data = {
>>>     'obs': {
>>>         'agent_state': torch.randn(B, agent_num, agent_obs_shape),
>>>         'global_state': torch.randn(B, agent_num, global_obs_shape),
>>>         'action_mask': torch.randint(0, 2, size=(B, agent_num, action_shape))
>>>     },
>>>     'action': torch.randn(B, agent_num, squeeze(action_shape))
>>> }
>>> model = ContinuousMAQAC(agent_obs_shape, global_obs_shape, action_shape, act_space, twin_critic=False)
>>> if action_space == 'regression':
>>>     action = model(data['obs'], mode='compute_actor')['action']
>>> elif action_space == 'reparameterization':
>>>     (mu, sigma) = model(data['obs'], mode='compute_actor')['logit']
>>> value = model(data, mode='compute_critic')['q_value']

离散MAQAC¶

class ding.model.DiscreteMAQAC(agent_obs_shape: int | SequenceType, global_obs_shape: int | SequenceType, action_shape: int | SequenceType, twin_critic: bool = False, actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None)[源代码]¶

Overview:: 与离散动作多智能体Q值演员-评论家（MAQAC）模型相关的算法的神经网络和计算图。该模型由演员和评论家组成，其中演员是一个MLP网络，评论家也是一个MLP网络。演员网络用于预测动作概率分布，评论家网络用于预测状态-动作对的Q值。
Interfaces:: __init__, forward, compute_actor, compute_critic

__init__(agent_obs_shape: int | SequenceType, global_obs_shape: int | SequenceType, action_shape: int | SequenceType, twin_critic: bool = False, actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None) → None[source]¶

Overview:

根据参数初始化DiscreteMAQAC模型。

Arguments:

agent_obs_shape (Union[int, SequenceType]): 代理的观察空间。
global_obs_shape (Union[int, SequenceType]): 全局观测的空间。
obs_shape (Union[int, SequenceType]): 观测的空间。
action_shape (Union[int, SequenceType]): 动作的空间。
twin_critic (bool): 是否包含双评论家。
actor_head_hidden_size (Optional[int]): 传递给actor-nn的Head的hidden_size。
actor_head_layer_num (int): 用于网络中计算Q值输出的层数，用于actor的神经网络。
critic_head_hidden_size (Optional[int]): 传递给critic-nn的Head的hidden_size。
critic_head_layer_num (int): 用于计算Q值输出的网络层数，用于critic的神经网络。
activation (Optional[nn.Module]): 在MLP中使用的激活函数类型，位于layer_fn之后，如果为None，则默认设置为nn.ReLU()
norm_type (Optional[str]): 使用的归一化类型，详情请参见 ding.torch_utils.fc_block。

compute_actor(inputs: Dict) → Dict[source]¶

Overview:

使用观察张量来预测动作的logits。

Arguments:

inputs (Dict[str, torch.Tensor]): The input dict tensor data, has keys:
- obs (Dict[str, torch.Tensor]): The input dict tensor data, has keys:
  
  agent_state (torch.Tensor): 代理的观察张量数据，形状为 \((B, A, N0)\)，其中 B 是批量大小，A 是代理数量。N0 对应于 agent_obs_shape。
  
  global_state (torch.Tensor): 全局观测张量数据，形状为 \((B, A, N1)\)，其中 B 是批量大小，A 是代理数量。N1 对应于 global_obs_shape。
  
  action_mask (torch.Tensor): 动作掩码张量数据，形状为 \((B, A, N2)\)，其中 B 是批量大小，A 是代理数量。N2 对应于 action_shape。

Returns:

output (Dict[str, torch.Tensor]): The output dict of DiscreteMAQAC forward computation graph, whose key-values vary in different forward modes.
- logit (torch.Tensor): 动作的输出logit（实值范围），其形状为 \((B, A, N2)\)，其中N2对应于 action_shape。
- action_mask (torch.Tensor): 与 action_shape 大小相同的动作掩码张量。

Examples:

>>> B = 32
>>> agent_obs_shape = 216
>>> global_obs_shape = 264
>>> agent_num = 8
>>> action_shape = 14
>>> data = {
>>>     'obs': {
>>>         'agent_state': torch.randn(B, agent_num, agent_obs_shape),
>>>         'global_state': torch.randn(B, agent_num, global_obs_shape),
>>>         'action_mask': torch.randint(0, 2, size=(B, agent_num, action_shape))
>>>     }
>>> }
>>> model = DiscreteMAQAC(agent_obs_shape, global_obs_shape, action_shape, twin_critic=True)
>>> logit = model.compute_actor(data)['logit']

compute_critic(inputs: Dict) → Dict[source]¶

Overview:

使用观测张量来预测Q值。

Arguments:

inputs (Dict[str, torch.Tensor]): The input dict tensor data, has keys:
- obs (Dict[str, torch.Tensor]): The input dict tensor data, has keys:
  
  agent_state (torch.Tensor): 代理的观察张量数据，形状为 \((B, A, N0)\)，其中 B 是批量大小，A 是代理数量。N0 对应于 agent_obs_shape。
  
  global_state (torch.Tensor): 全局观测张量数据，形状为 \((B, A, N1)\)，其中 B 是批量大小，A 是代理数量。N1 对应于 global_obs_shape。
  
  action_mask (torch.Tensor): 动作掩码张量数据，形状为 \((B, A, N2)\)，其中 B 是批量大小，A 是代理数量。N2 对应于 action_shape。

Returns:

output (Dict[str, torch.Tensor]): The output dict of DiscreteMAQAC forward computation graph, whose key-values vary in different values of twin_critic.
- q_value (list): 如果 twin_critic=True, q_value 应该包含2个元素，每个元素的形状为 \((B, A, N2)\), 其中 B 是批量大小，A 是代理数量。N2 对应于 action_shape。否则，q_value 应该是 torch.Tensor。

Examples:

>>> B = 32
>>> agent_obs_shape = 216
>>> global_obs_shape = 264
>>> agent_num = 8
>>> action_shape = 14
>>> data = {
>>>     'obs': {
>>>         'agent_state': torch.randn(B, agent_num, agent_obs_shape),
>>>         'global_state': torch.randn(B, agent_num, global_obs_shape),
>>>         'action_mask': torch.randint(0, 2, size=(B, agent_num, action_shape))
>>>     }
>>> }
>>> model = DiscreteMAQAC(agent_obs_shape, global_obs_shape, action_shape, twin_critic=True)
>>> value = model.compute_critic(data)['q_value']

forward(inputs: Tensor | Dict, mode: str) → Dict[source]¶

Overview:

使用观察张量来预测输出，使用compute_actor或compute_critic模式。

Arguments:

inputs (Dict[str, torch.Tensor]): The input dict tensor data, has keys:
- obs (Dict[str, torch.Tensor]): The input dict tensor data, has keys:
  
  agent_state (torch.Tensor): 代理的观察张量数据，形状为 \((B, A, N0)\)，其中 B 是批量大小，A 是代理数量。N0 对应于 agent_obs_shape。
  
  global_state (torch.Tensor): 全局观测张量数据，形状为 \((B, A, N1)\)，其中 B 是批量大小，A 是代理数量。N1 对应于 global_obs_shape。
  
  action_mask (torch.Tensor): 动作掩码张量数据，形状为 \((B, A, N2)\)，其中 B 是批量大小，A 是代理数量。N2 对应于 action_shape。
模式 (str): 前向模式，所有模式都在这个类的开头定义。

Returns:

输出 (Dict[str, torch.Tensor]): DiscreteMAQAC前向计算图的输出字典，其键值在不同的前向模式中有所不同。

Examples:

>>> B = 32
>>> agent_obs_shape = 216
>>> global_obs_shape = 264
>>> agent_num = 8
>>> action_shape = 14
>>> data = {
>>>     'obs': {
>>>         'agent_state': torch.randn(B, agent_num, agent_obs_shape),
>>>         'global_state': torch.randn(B, agent_num, global_obs_shape),
>>>         'action_mask': torch.randint(0, 2, size=(B, agent_num, action_shape))
>>>     }
>>> }
>>> model = DiscreteMAQAC(agent_obs_shape, global_obs_shape, action_shape, twin_critic=True)
>>> logit = model(data, mode='compute_actor')['logit']
>>> value = model(data, mode='compute_critic')['q_value']

QACDIST¶

class ding.model.QACDIST(obs_shape: int | SequenceType, action_shape: int | SequenceType, action_space: str = 'regression', critic_head_type: str = 'categorical', actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, v_min: float | None = -10, v_max: float | None = 10, n_atom: int | None = 51)[源代码]¶

Overview:: 具有分布Q值的QAC模型。
Interfaces:: __init__, forward, compute_actor, compute_critic

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType, action_space: str = 'regression', critic_head_type: str = 'categorical', actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, v_min: float | None = -10, v_max: float | None = 10, n_atom: int | None = 51) → None[source]¶

Overview:

根据参数初始化QAC分布模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测的空间。
action_shape (Union[int, SequenceType]): 动作的空间。
action_space (str): 选择 regression 还是 reparameterization。
critic_head_type (str): 仅 categorical。
actor_head_hidden_size (Optional[int]): 传递给actor-nn的Head的hidden_size。
actor_head_layer_num (int):
网络中用于计算演员神经网络Q值输出的层数。
critic_head_hidden_size (Optional[int]): 传递给critic-nn的Head的hidden_size。
critic_head_layer_num (int):
网络中用于计算评论者神经网络Q值输出的层数。
activation (Optional[nn.Module]):
在MLP中使用的激活函数类型，在layer_fn之后，如果None则默认设置为nn.ReLU()
norm_type (Optional[str]):
使用的归一化类型，详情请参见ding.torch_utils.fc_block。
v_min (int): 最小原子的值
v_max (int): 最大原子的值
n_atom (int): 支持中的原子数量

compute_actor(inputs: Tensor) → Dict[source]¶

Overview:

使用编码的嵌入张量来预测输出。使用'compute_actor'模式执行参数更新使用编码的嵌入张量来预测输出。

Arguments:

inputs (torch.Tensor):
编码后的嵌入张量，由给定的hidden_size决定，即(B, N=hidden_size)。 hidden_size = actor_head_hidden_size
模式 (str): 前向模式的名称。

Returns:

输出 (Dict): 前向传递编码器和头部的输出。

ReturnsKeys (either):

动作 (torch.Tensor): 与 action_shape 大小相同的连续动作张量。
logit (torch.Tensor):
Logit 张量编码 mu 和 sigma，两者的大小与输入 x 相同。

Shapes:

输入 (torch.Tensor): \((B, N0)\), B 是批量大小，N0 对应于 hidden_size
动作 (torch.Tensor): \((B, N0)\)
logit (list): 2个元素，mu和sigma，每个的形状为\((B, N0)\)。
q_value (torch.FloatTensor): \((B, )\), B 是批量大小。

Examples:

>>> # Regression mode
>>> model = QACDIST(64, 64, 'regression')
>>> inputs = torch.randn(4, 64)
>>> actor_outputs = model(inputs,'compute_actor')
>>> assert actor_outputs['action'].shape == torch.Size([4, 64])
>>> # Reparameterization Mode
>>> model = QACDIST(64, 64, 'reparameterization')
>>> inputs = torch.randn(4, 64)
>>> actor_outputs = model(inputs,'compute_actor')
>>> actor_outputs['logit'][0].shape # mu
>>> torch.Size([4, 64])
>>> actor_outputs['logit'][1].shape # sigma
>>> torch.Size([4, 64])

compute_critic(inputs: Dict) → Dict[来源]¶

Overview:

使用'compute_critic'模式执行参数更新使用编码的嵌入张量来预测输出。

Arguments:

obs, action 编码的张量。
模式 (str): 前向模式的名称。

Returns:

输出 (Dict): Q值输出和分布。

ReturnKeys:

q_value (torch.Tensor): Q值张量，大小与批量大小相同。
分布 (torch.Tensor): Q值分布张量。

Shapes:

obs (torch.Tensor): \((B, N1)\), 其中 B 是批量大小，N1 是 obs_shape
动作 (torch.Tensor): \((B, N2)\), 其中 B 是批量大小，N2 是``action_shape``
q_value (torch.FloatTensor): \((B, N2)\), 其中 B 是批量大小，N2 是 action_shape
分布 (torch.FloatTensor): \((B, 1, N3)\), 其中 B 是批量大小，N3 是 num_atom

Examples:

>>> # Categorical mode
>>> inputs = {'obs': torch.randn(4,N), 'action': torch.randn(4,1)}
>>> model = QACDIST(obs_shape=(N, ),action_shape=1,action_space='regression',             ...                 critic_head_type='categorical', n_atoms=51)
>>> q_value = model(inputs, mode='compute_critic') # q value
>>> assert q_value['q_value'].shape == torch.Size([4, 1])
>>> assert q_value['distribution'].shape == torch.Size([4, 1, 51])

forward(inputs: Tensor | Dict, mode: str) → Dict[来源]¶

Overview:

使用观察和动作张量来预测输出。参数更新使用QACDIST的MLPs前向设置。

Arguments:

Forward with 'compute_actor':

inputs (torch.Tensor):
编码的嵌入张量，由给定的hidden_size决定，即(B, N=hidden_size)。是actor_head_hidden_size还是critic_head_hidden_size取决于mode。

Forward with 'compute_critic', inputs (Dict) Necessary Keys:

obs, action 编码的张量。

模式 (str): 前向模式的名称。

Returns:

输出 (Dict): 网络前向的输出。
Forward with 'compute_actor', Necessary Keys (either):
动作 (torch.Tensor): 与输入 x 大小相同的动作张量。

logit (torch.Tensor):
Logit 张量编码 mu 和 sigma，两者的大小与输入 x 相同。
Forward with 'compute_critic', Necessary Keys:
q_value (torch.Tensor): Q值张量，大小与批量大小相同。

分布 (torch.Tensor): Q值分布张量。

Actor Shapes:

输入 (torch.Tensor): \((B, N0)\), B 是批量大小，N0 对应于 hidden_size
动作 (torch.Tensor): \((B, N0)\)
q_value (torch.FloatTensor): \((B, )\), 其中 B 是批量大小。

Critic Shapes:

obs (torch.Tensor): \((B, N1)\), 其中 B 是批量大小，N1 是 obs_shape
动作 (torch.Tensor): \((B, N2)\), 其中 B 是批量大小，N2 是``action_shape``
q_value (torch.FloatTensor): \((B, N2)\), 其中 B 是批量大小，N2 是 action_shape
分布 (torch.FloatTensor): \((B, 1, N3)\), 其中 B 是批量大小，N3 是 num_atom

Actor Examples:

>>> # Regression mode
>>> model = QACDIST(64, 64, 'regression')
>>> inputs = torch.randn(4, 64)
>>> actor_outputs = model(inputs,'compute_actor')
>>> assert actor_outputs['action'].shape == torch.Size([4, 64])
>>> # Reparameterization Mode
>>> model = QACDIST(64, 64, 'reparameterization')
>>> inputs = torch.randn(4, 64)
>>> actor_outputs = model(inputs,'compute_actor')
>>> actor_outputs['logit'][0].shape # mu
>>> torch.Size([4, 64])
>>> actor_outputs['logit'][1].shape # sigma
>>> torch.Size([4, 64])

Critic Examples:

>>> # Categorical mode
>>> inputs = {'obs': torch.randn(4,N), 'action': torch.randn(4,1)}
>>> model = QACDIST(obs_shape=(N, ),action_shape=1,action_space='regression',             ...                 critic_head_type='categorical', n_atoms=51)
>>> q_value = model(inputs, mode='compute_critic') # q value
>>> assert q_value['q_value'].shape == torch.Size([4, 1])
>>> assert q_value['distribution'].shape == torch.Size([4, 1, 51])

离散边界条件¶

class ding.model.DiscreteBC(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], dueling: bool = True, head_hidden_size: int | None = None, head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, strides: list | None = None)[source]¶

Overview:: 离散BC网络。
Interfaces:: __init__, forward

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], dueling: bool = True, head_hidden_size: int | None = None, head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, strides: list | None = None) → None[源代码]¶

Overview:

根据输入参数初始化离散BC（编码器 + 头部）模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测空间形状，例如 8 或 [4, 84, 84]。
action_shape (Union[int, SequenceType]): 动作空间形状，例如 6 或 [2, 3, 3]。
encoder_hidden_size_list (SequenceType): 传递给Encoder的hidden_size集合，最后一个元素必须匹配head_hidden_size。
决斗 (dueling): 是否选择 DuelingHead 或 DiscreteHead（默认）。
head_hidden_size (Optional[int]): 头部网络的hidden_size。
head_layer_num (int): 用于计算Q值输出的头部网络中的层数
activation (Optional[nn.Module]): 网络中的激活函数类型，如果 None 则默认设置为 nn.ReLU()。
norm_type (Optional[str]): 网络中的归一化类型，更多详情请参见 ding.torch_utils.fc_block。
步幅 (Optional[list]): 每个卷积层的步幅，例如 [2, 2, 2]。此参数的长度应与 encoder_hidden_size_list 相同。

forward(x: Tensor) → Dict[来源]¶

Overview:

离散BC前向计算图，输入观测张量以预测q_value。

Arguments:

x (torch.Tensor): 观测输入

Returns:

输出 (Dict): DiscreteBC 前向输出，例如 q_value。

ReturnsKeys:

logit (torch.Tensor): 每个动作维度的离散Q值输出。

Shapes:

x (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是 obs_shape
logit (torch.FloatTensor): \((B, M)\), 其中 B 是批量大小，M 是 action_shape

Examples:

>>> model = DiscreteBC(32, 6)  # arguments: 'obs_shape' and 'action_shape'
>>> inputs = torch.randn(4, 32)
>>> outputs = model(inputs)
>>> assert isinstance(outputs, dict) and outputs['logit'].shape == torch.Size([4, 6])

连续边界条件¶

class ding.model.ContinuousBC(obs_shape: int | SequenceType, action_shape: int | SequenceType | EasyDict, action_space: str, actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None)[source]¶

Overview:: ContinuousBC网络。
Interfaces:: __init__, forward

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType | EasyDict, action_space: str, actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None) → None[源代码]¶

Overview:

根据输入参数初始化ContinuousBC模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测值的形状，例如 128, (156, )。
action_shape (Union[int, SequenceType, EasyDict]): 动作的形状，例如 4, (3, ), EasyDict({‘action_type_shape’: 3, ‘action_args_shape’: 4}).
action_space (str): 动作空间的类型，包括 [regression, reparameterization]。
actor_head_hidden_size (Optional[int]): 传递给actor head的hidden_size。
actor_head_layer_num (int): 用于计算演员头Q值输出的网络层数。
activation (Optional[nn.Module]): 在MLP中每个FC层后使用的激活函数类型，如果None则默认设置为nn.ReLU()。
norm_type (Optional[str]): 网络层（FC, Conv）后的归一化类型，详情请参见 ding.torch_utils.network。

forward(inputs: Tensor | Dict[str, Tensor]) → Dict[来源]¶

Overview:

ContinuousBC 的唯一执行（前向）方法。

Arguments:

输入 (torch.Tensor): 观测数据，默认为张量。

Returns:

输出 (Dict): 输出字典数据，包括不同动作空间中的不同键值对。

ReturnsKeys:

动作 (torch.Tensor): 演员网络的动作输出，形状为 \((B, action_shape)\)。
logit (List[torch.Tensor]): 演员网络的重新参数化动作输出，形状为 \((B, action_shape)\)。

Shapes:

输入 (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是 obs_shape
动作 (torch.FloatTensor): \((B, M)\), 其中 B 是批量大小，M 是 action_shape
logit (List[torch.FloatTensor]): \((B, M)\), 其中 B 是批量大小，M 是 action_shape

Examples (Regression):

>>> model = ContinuousBC(32, 6, action_space='regression')
>>> inputs = torch.randn(4, 32)
>>> outputs = model(inputs)
>>> assert isinstance(outputs, dict) and outputs['action'].shape == torch.Size([4, 6])

Examples (Reparameterization):

>>> model = ContinuousBC(32, 6, action_space='reparameterization')
>>> inputs = torch.randn(4, 32)
>>> outputs = model(inputs)
>>> assert isinstance(outputs, dict) and outputs['logit'][0].shape == torch.Size([4, 6])
>>> assert outputs['logit'][1].shape == torch.Size([4, 6])

PDQN¶

class ding.model.PDQN(obs_shape: int | SequenceType, action_shape: EasyDict, encoder_hidden_size_list: SequenceType = [128, 128, 64], dueling: bool = True, head_hidden_size: int | None = None, head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, multi_pass: bool | None = False, action_mask: list | None = None)[源代码]¶

Overview:: PDQN（https://arxiv.org/abs/1810.06394v1）和MPDQN（https://arxiv.org/abs/1905.04388）算法的神经网络和计算图，用于参数化动作空间。该模型支持具有离散action_type和连续action_arg的参数化动作空间。原则上，PDQN由x网络（连续动作参数网络）和Q网络（离散动作类型网络）组成。但为了简化，代码被分为encoder和actor_head，它们分别包含上述两个网络的编码器和头部。
Interface:: __init__, forward, compute_discrete, compute_continuous.

__init__(obs_shape: int | SequenceType, action_shape: EasyDict, encoder_hidden_size_list: SequenceType = [128, 128, 64], dueling: bool = True, head_hidden_size: int | None = None, head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, multi_pass: bool | None = False, action_mask: list | None = None) → None[源代码]¶

Overview:

根据输入参数初始化PDQN（编码器 + 头部）模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测空间形状，例如 8 或 [4, 84, 84]。
action_shape (EasyDict): 动作空间形状以字典类型表示，例如 EasyDict({'action_type_shape': 3, 'action_args_shape': 5})。
encoder_hidden_size_list (SequenceType): 传递给Encoder的hidden_size集合，最后一个元素必须匹配head_hidden_size。
决斗 (dueling): 是否选择 DuelingHead 或 DiscreteHead（默认）。
head_hidden_size (Optional[int]): 头部网络的hidden_size。
head_layer_num (int): 用于计算Q值输出的头部网络中的层数。
activation (Optional[nn.Module]): 网络中的激活函数类型，如果 None 则默认设置为 nn.ReLU()。
norm_type (Optional[str]): 网络中的归一化类型，更多详情请参见 ding.torch_utils.fc_block。
multi_pass (Optional[bool]): 是否使用多通道版本。
action_mask: (Optional[list]): 一个动作掩码，指示动作参数如何与每个离散动作相关联。例如，如果有3个离散动作，4个连续动作参数，第一个离散动作与第一个连续动作参数相关联，第二个离散动作与第二个连续动作参数相关联，第三个离散动作与剩下的2个动作参数相关联，那么动作掩码将类似于：[[1,0,0,0],[0,1,0,0],[0,0,1,1]]，形状为3*4。

compute_continuous(inputs: Tensor) → Dict[source]¶

Overview:

使用观察张量来预测连续动作参数。

Arguments:

输入 (torch.Tensor): 观测输入。

Returns:

outputs (Dict): A dict with key ‘action_args’.
- ‘action_args’ (torch.Tensor): 连续动作参数。

Shapes:

输入 (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是 obs_shape。
action_args (torch.Tensor): \((B, M)\), 其中 M 是 action_args_shape.

Examples:

>>> act_shape = EasyDict({'action_type_shape': (3, ), 'action_args_shape': (5, )})
>>> model = PDQN(4, act_shape)
>>> inputs = torch.randn(64, 4)
>>> outputs = model.forward(inputs, mode='compute_continuous')
>>> assert outputs['action_args'].shape == torch.Size([64, 5])

compute_discrete(inputs: Dict | EasyDict) → Dict[source]¶

Overview:

使用观测张量和连续动作参数来预测离散动作类型。

Arguments:

inputs (Union[Dict, EasyDict]): A dict with keys ‘state’, ‘action_args’.
- 状态 (torch.Tensor): 观测输入。
- action_args (torch.Tensor): 动作参数用于与观察结果连接，并作为离散动作类型网络的输入。

Returns:

outputs (Dict): A dict with keys ‘logit’, ‘action_args’.
- ‘logit’: 每个离散动作的logit值。
- ‘action_args’：连续动作参数（与inputs[‘action_args’]相同）供后续使用。

Examples:

>>> act_shape = EasyDict({'action_type_shape': (3, ), 'action_args_shape': (5, )})
>>> model = PDQN(4, act_shape)
>>> inputs = {'state': torch.randn(64, 4), 'action_args': torch.randn(64, 5)}
>>> outputs = model.forward(inputs, mode='compute_discrete')
>>> assert outputs['logit'].shape == torch.Size([64, 3])
>>> assert outputs['action_args'].shape == torch.Size([64, 5])

forward(inputs: Tensor | Dict | EasyDict, mode: str) → Dict[source]¶

Overview:

PDQN前向计算图，输入观测张量以预测离散动作的q_value和连续动作参数的值。

Arguments:

输入 (Union[torch.Tensor, Dict, EasyDict]): 输入包括观察值和其他根据模式的信息。
模式 (str): 前向模式的名称。

Shapes:

输入 (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是 obs_shape。

决策变换器¶

class ding.model.DecisionTransformer(state_dim: int | SequenceType, act_dim: int, n_blocks: int, h_dim: int, context_len: int, n_heads: int, drop_p: float, max_timestep: int = 4096, state_encoder: Module | None = None, continuous: bool = False)[源代码]¶

Overview:: 决策变换器的实现。
Interfaces:: __init__, forward, configure_optimizers

__init__(state_dim: int | SequenceType, act_dim: int, n_blocks: int, h_dim: int, context_len: int, n_heads: int, drop_p: float, max_timestep: int = 4096, state_encoder: Module | None = None, continuous: bool = False)[源代码]¶

Overview:

根据输入参数初始化DecisionTransformer模型。

Arguments:

obs_shape (Union[int, SequenceType]): 状态的维度，例如 128 或 (4, 84, 84)。
act_dim (int): 动作的维度，例如6。
n_blocks (int): 决策变换器中变压器块的数量，例如3。
h_dim (int): 隐藏层的维度，例如128。
context_len (int): 注意力的最大上下文长度，例如6。
n_heads (int): 计算注意力时的头数，例如8。
drop_p (float): 丢弃层的丢弃率，例如 0.1。
max_timestep (int): 总序列的最大长度，默认为4096。
state_encoder (Optional[nn.Module]): 用于预处理给定输入的编码器。如果设置为None，原始状态将被直接推入transformer。
continuous (bool): 动作空间是否为连续的，默认为 False。

forward(timesteps: Tensor, states: Tensor, actions: Tensor, returns_to_go: Tensor, tar: int | None = None) → Tuple[Tensor, Tensor, Tensor][source]¶

Overview:

决策变换器的前向计算图，输入一个序列张量并返回一个具有相同形状的张量。

Arguments:

时间步长 (torch.Tensor): 输入序列的时间步长。
状态 (torch.Tensor): 状态的序列。
动作 (torch.Tensor): 动作序列。
returns_to_go (torch.Tensor): 返回序列。
tar (Optional[int]): 是否预测动作，无论索引如何。

Returns:

输出 (Tuple[torch.Tensor, torch.Tensor, torch.Tensor]): 输出包含三个张量，它们分别是预测的状态、预测的动作和预测的返回目标。

Examples:

>>> B, T = 4, 6
>>> state_dim = 3
>>> act_dim = 2
>>> DT_model = DecisionTransformer(                state_dim=state_dim,                act_dim=act_dim,                n_blocks=3,                h_dim=8,                context_len=T,                n_heads=2,                drop_p=0.1,            )
>>> timesteps = torch.randint(0, 100, [B, 3 * T - 1, 1], dtype=torch.long)  # B x T
>>> states = torch.randn([B, T, state_dim])  # B x T x state_dim
>>> actions = torch.randint(0, act_dim, [B, T, 1])
>>> action_target = torch.randint(0, act_dim, [B, T, 1])
>>> returns_to_go_sample = torch.tensor([1, 0.8, 0.6, 0.4, 0.2, 0.]).repeat([B, 1]).unsqueeze(-1).float()
>>> traj_mask = torch.ones([B, T], dtype=torch.long)  # B x T
>>> actions = actions.squeeze(-1)
>>> state_preds, action_preds, return_preds = DT_model.forward(                timesteps=timesteps, states=states, actions=actions, returns_to_go=returns_to_go            )
>>> assert state_preds.shape == torch.Size([B, T, state_dim])
>>> assert return_preds.shape == torch.Size([B, T, 1])
>>> assert action_preds.shape == torch.Size([B, T, act_dim])

语言转换器¶

class ding.model.LanguageTransformer(model_name: str = 'bert-base-uncased', add_linear: bool = False, embedding_size: int = 128, freeze_encoder: bool = True, hidden_dim: int = 768, norm_embedding: bool = False)[源代码]¶

Overview:: LanguageTransformer网络。下载一个预训练的语言模型并在其上添加头部。在默认情况下，我们使用BERT模型作为文本编码器，其双向特性非常适合获取整个句子的嵌入。
Interfaces:: __init__, forward

__init__(model_name: str = 'bert-base-uncased', add_linear: bool = False, embedding_size: int = 128, freeze_encoder: bool = True, hidden_dim: int = 768, norm_embedding: bool = False) → None[source]¶

Overview:

根据输入参数初始化LanguageTransformer模型。

Arguments:

model_name (str): huggingface中的基础语言模型名称，例如“bert-base-uncased”。
add_linear (bool): 是否在语言模型的顶部添加一个线性层，默认为 False。
embedding_size (int): 添加的线性层的嵌入大小，例如128。
freeze_encoder (bool): 是否在训练时冻结编码器语言模型，默认为 True。
hidden_dim (int): 编码模型（例如 BERT）的嵌入维度。此值应与您使用的模型相对应。对于 bert-base-uncased，此值为 768。
norm_embedding (bool): 是否对嵌入向量进行归一化。默认为 False。

forward(train_samples: List[str], candidate_samples: List[str] | None = None, mode: str = 'compute_actor') → Dict[source]¶

Overview:

LanguageTransformer 前向计算图，输入两个字符串列表并预测它们的匹配分数。不同的 mode 将使用不同的网络模块进行前向传播，以获得不同的输出。

Arguments:

train_samples (List[str]): 一个字符串列表。
candidate_samples (Optional[List[str]]): 用于计算匹配分数的其他字符串列表。
- 模式 (str): 前向模式，所有模式都在这个类的开头定义。

Returns:

输出 (Dict): 输出字典数据，包括匹配分数的logit和相应的torch.distributions.Categorical对象。

Examples:

>>> test_pids = [1]
>>> cand_pids = [0, 2, 4]
>>> problems = [                 "This is problem 0", "This is the first question", "Second problem is here", "Another problem",                 "This is the last problem"             ]
>>> ctxt_list = [problems[pid] for pid in test_pids]
>>> cands_list = [problems[pid] for pid in cand_pids]
>>> model = LanguageTransformer(model_name="bert-base-uncased", add_linear=True, embedding_size=256)
>>> scores = model(ctxt_list, cands_list)
>>> assert scores.shape == (1, 3)

混音器¶

class ding.model.Mixer(agent_num: int, state_dim: int, mixing_embed_dim: int, hypernet_embed: int = 64, activation: Module = ReLU())[source]¶

Overview:: QMIX中的混合网络，将每个代理的独立q_value混合成一个总的q_value。混合网络的权重（但不包括偏置）被限制为非负，并由单独的超网络生成。每个超网络以全局状态s作为输入，并生成混合网络一层的权重。
Interface:: __init__, forward.

__init__(agent_num: int, state_dim: int, mixing_embed_dim: int, hypernet_embed: int = 64, activation: Module = ReLU())[source]¶

Overview:

根据参数初始化QMIX中提出的混合器网络。每个超网络由线性层组成，后跟一个绝对激活函数，以确保混合器网络的权重是非负的。

Arguments:

agent_num (int): 代理的数量，例如8。
state_dim(int): 全局观察状态的维度，例如16。
mixing_embed_dim (int): 混合状态嵌入的维度，例如128。
hypernet_embed (int): 超网络嵌入的维度，默认为64。
activation (nn.Module): 网络中的激活函数，默认为 nn.ReLU()。

forward(agent_qs, states)[source]¶

Overview:

pymarl混合器网络的前向计算图。将每个代理的独立q_value混合成一个总的q_value，权重由超网络根据全局states生成。

Arguments:

agent_qs (torch.FloatTensor): 每个代理的独立q_value。
状态 (torch.FloatTensor): 全局状态的嵌入向量。

Returns:

q_tot (torch.FloatTensor): 总的混合q值。

Shapes:

agent_qs (torch.FloatTensor): \((B, N)\), 其中 B 是批量大小，N 是代理数量。
状态 (torch.FloatTensor): \((B, M)\), 其中 M 是嵌入大小。
q_tot (torch.FloatTensor): \((B, )\).

QMix¶

class ding.model.QMix(agent_num: int, obs_shape: int, global_obs_shape: int | List[int], action_shape: int, hidden_size_list: list, mixer: bool = True, lstm_type: str = 'gru', activation: Module = ReLU(), dueling: bool = False)[source]¶

Overview:: 与QMIX相关的算法的神经网络和计算图(https://arxiv.org/abs/1803.11485)。QMIX由两部分组成：代理Q网络和混合器（可选）。QMIX论文提到所有代理共享本地Q网络参数，因此这里只初始化一个Q网络。然后根据mixer设置使用求和或混合器网络处理本地Q以获得全局Q。
Interface:: __init__, forward.

__init__(agent_num: int, obs_shape: int, global_obs_shape: int | List[int], action_shape: int, hidden_size_list: list, mixer: bool = True, lstm_type: str = 'gru', activation: Module = ReLU(), dueling: bool = False) → None[源代码]¶

Overview:

根据参数初始化QMIX神经网络，即代理Q网络和混合器。

Arguments:

agent_num (int): 代理的数量，例如8。
obs_shape (int): 每个代理的观察状态的维度，例如8或[4, 84, 84]。
global_obs_shape (int): 全局观测状态的维度，例如8或[4, 84, 84]。
action_shape (int): 动作形状的维度，例如6或[2, 3, 3]。
hidden_size_list (list): q_network 的隐藏层大小列表，最后一个元素必须与 mixer 的 mixing_embed_dim 匹配。
mixer (bool): 是否使用mixer网络，默认为True。如果为False，则最终的本地Q将被添加以获得全局Q。
lstm_type (str): q_network 中 RNN 模块的类型，目前支持 ['normal', 'pytorch', 'gru']，默认为 gru。
activation (nn.Module): 在MLP中使用的激活函数类型，位于layer_fn之后，如果为None，则默认设置为nn.ReLU()。
决斗 (bool): 是否选择 DuelingHead (True) 或 DiscreteHead (False), 默认为 False。

forward(data: dict, single_step: bool = True) → dict[source]¶

Overview:

QMIX 前向计算图，输入字典包括时间序列观察和相关数据，用于预测总 q_value 和每个代理的 q_value。

Arguments:

data (dict): Input data dict with keys [‘obs’, ‘prev_state’, ‘action’].
- agent_state (torch.Tensor): 每个代理的时间序列局部观测数据。
- global_state (torch.Tensor): 时间序列全局观测数据。
- prev_state (list): 用于 q_network 的先前rnn状态。
- action (torch.Tensor 或 None): 每个代理在函数外部给出的动作。如果 action 为 None，则使用 argmax q_value 索引作为动作来计算 agent_q_act。
single_step (bool): 是否单步前进，如果是，则在前进前添加时间步维度，并在前进后移除它。

Returns:

ret (dict): 输出数据字典，包含键 [total_q, logit, next_state].

ReturnsKeys:

total_q (torch.Tensor): 总q值，这是混合器网络的结果。
agent_q (torch.Tensor): 每个代理的q值。
next_state (list): q_network 的下一个 rnn 状态。

Shapes:

agent_state (torch.Tensor): \((T, B, A, N)\), 其中 T 是时间步，B 是批次大小，A 是代理数量，N 是观测形状。
全局状态 (torch.Tensor): \((T, B, M)\), 其中 M 是全局观测形状。
prev_state (list): 数学:(B, A), 一个长度为B的列表，每个元素是一个长度为A的列表。
动作 (torch.Tensor): \((T, B, A)\).
total_q (torch.Tensor): \((T, B)\).
agent_q (torch.Tensor): \((T, B, A, P)\), 其中 P 是动作形状。
next_state (list): 数学:(B, A), 一个长度为B的列表，每个元素是一个长度为A的列表。

COMA¶

class ding.model.COMA(agent_num: int, obs_shape: Dict, action_shape: int | SequenceType, actor_hidden_size_list: SequenceType)[source]¶

Overview:

COMA算法的网络，属于QAC类型的演员-评论家。

Interface:

__init__, forward

Properties:

模式 (list): 前向模式的列表，包括 compute_actor 和 compute_critic

__init__(agent_num: int, obs_shape: Dict, action_shape: int | SequenceType, actor_hidden_size_list: SequenceType) → None[source]¶

Overview:

初始化COMA网络

Arguments:

agent_num (int): 代理的数量
obs_shape (Dict): 观察信息，包括 agent_state 和 global_state
action_shape (Union[int, SequenceType]): 动作形状的维度
actor_hidden_size_list (SequenceType): 隐藏大小的列表

forward(inputs: Dict, mode: str) → Dict[source]¶

Overview:

COMA网络的前向计算图

Arguments:

输入 (dict): 输入数据字典，包含键 ['obs', 'prev_state', 'action']
agent_state (torch.Tensor): 每个代理的本地状态(obs)
global_state (torch.Tensor): 全局状态(obs)
动作 (torch.Tensor): 被掩码的动作

ArgumentsKeys:

必要的: obs { agent_state, global_state, action_mask }, action, prev_state

ReturnsKeys:

necessary:
- compute_critic: q_value
- compute_actor: logit, next_state, action_mask

Shapes:

obs (dict): agent_state: \((T, B, A, N, D)\), action_mask: \((T, B, A, N, A)\)
prev_state (list): \([[[h, c] for _ in range(A)] for _ in range(B)]\)
logit (torch.Tensor): \((T, B, A, N, A)\)
next_state (list): \([[[h, c] for _ in range(A)] for _ in range(B)]\)
action_mask (torch.Tensor): \((T, B, A, N, A)\)
q_value (torch.Tensor): \((T, B, A, N, A)\)

Examples:

>>> agent_num, bs, T = 4, 3, 8
>>> agent_num, bs, T = 4, 3, 8
>>> obs_dim, global_obs_dim, action_dim = 32, 32 * 4, 9
>>> coma_model = COMA(
>>>     agent_num=agent_num,
>>>     obs_shape=dict(agent_state=(obs_dim, ), global_state=(global_obs_dim, )),
>>>     action_shape=action_dim,
>>>     actor_hidden_size_list=[128, 64],
>>> )
>>> prev_state = [[None for _ in range(agent_num)] for _ in range(bs)]
>>> data = {
>>>     'obs': {
>>>         'agent_state': torch.randn(T, bs, agent_num, obs_dim),
>>>         'action_mask': None,
>>>     },
>>>     'prev_state': prev_state,
>>> }
>>> output = coma_model(data, mode='compute_actor')
>>> data= {
>>>     'obs': {
>>>         'agent_state': torch.randn(T, bs, agent_num, obs_dim),
>>>         'global_state': torch.randn(T, bs, global_obs_dim),
>>>     },
>>>     'action': torch.randint(0, action_dim, size=(T, bs, agent_num)),
>>> }
>>> output = coma_model(data, mode='compute_critic')

QTran¶

class ding.model.QTran(agent_num: int, obs_shape: int, global_obs_shape: int, action_shape: int, hidden_size_list: list, embedding_size: int, lstm_type: str = 'gru', dueling: bool = False)[源代码]¶

Overview:: QTRAN 网络
Interface:: __init__, forward

__init__(agent_num: int, obs_shape: int, global_obs_shape: int, action_shape: int, hidden_size_list: list, embedding_size: int, lstm_type: str = 'gru', dueling: bool = False) → None[源代码]¶

Overview:

初始化QTRAN网络

Arguments:

agent_num (int): 代理的数量
obs_shape (int): 每个代理的观察状态的维度
global_obs_shape (int): 全局观测状态的维度
action_shape (int): 动作形状的维度
hidden_size_list (list): 隐藏层大小的列表
embedding_size (int): 嵌入的维度
lstm_type (str): 使用lstm或gru，默认为gru
决斗 (bool): 是否使用决斗头，默认为 False。

forward(data: dict, single_step: bool = True) → dict[source]¶

Overview:

qtran网络的前向计算图

Arguments:

data (dict): input data dict with keys [‘obs’, ‘prev_state’, ‘action’]
- agent_state (torch.Tensor): 每个代理的本地状态(obs)
- global_state (torch.Tensor): 全局状态(obs)
- prev_state (list): 之前的rnn状态
- 动作 (torch.Tensor 或 None): 如果动作为 None，则使用 argmax q_value 索引作为动作来计算 agent_q_act
single_step (bool): 是否单步前进，如果是，则在前进前添加时间步维度并在前进后移除它

Return:

ret (dict): output data dict with keys [‘total_q’, ‘logit’, ‘next_state’]
- total_q (torch.Tensor): 总q值，这是混合器网络的结果
- agent_q (torch.Tensor): 每个代理的q值
- next_state (list): 下一个RNN状态

Shapes:

agent_state (torch.Tensor): \((T, B, A, N)\), 其中 T 是时间步长，B 是批量大小，A 是代理数量，N 是观测形状
global_state (torch.Tensor): \((T, B, M)\), 其中 M 是 global_obs_shape
prev_state (list): 数学:(B, A), 一个长度为B的列表，每个元素是一个长度为A的列表
动作 (torch.Tensor): \((T, B, A)\)
total_q (torch.Tensor): \((T, B)\)
agent_q (torch.Tensor): \((T, B, A, P)\), 其中 P 是动作形状
next_state (list): 数学:(B, A), 一个长度为B的列表，每个元素是一个长度为A的列表

WQMix¶

class ding.model.WQMix(agent_num: int, obs_shape: int, global_obs_shape: int, action_shape: int, hidden_size_list: list, lstm_type: str = 'gru', dueling: bool = False)[源代码]¶

Overview:: WQMIX (https://arxiv.org/abs/2006.10800) 网络，包含两个组件：1) Q_tot，与QMIX网络相同，由代理Q网络和混合网络组成。2) 一个不受限制的联合动作Q_star，由代理Q网络和mixer_star网络组成。QMIX论文提到所有代理共享本地Q网络参数，因此在Q_tot或Q_star中只初始化一个Q网络。
Interface:: __init__, forward.

__init__(agent_num: int, obs_shape: int, global_obs_shape: int, action_shape: int, hidden_size_list: list, lstm_type: str = 'gru', dueling: bool = False) → None[source]¶

Overview:

根据参数初始化WQMIX神经网络，即代理Q网络和混合器，Q_star网络和mixer_star。

Arguments:

agent_num (int): 代理的数量，例如8。
obs_shape (int): 每个代理的观察状态的维度，例如8。
global_obs_shape (int): 全局观测状态的维度，例如8。
action_shape (int): 动作形状的维度，例如6。
hidden_size_list (list): q_network 的隐藏层大小列表，最后一个元素必须与 mixer 的 mixing_embed_dim 匹配。
lstm_type (str): q_network 中 RNN 模块的类型，目前支持 ['normal', 'pytorch', 'gru']，默认为 gru。
决斗 (bool): 是否选择 DuelingHead (True) 或 DiscreteHead (False), 默认为 False。

forward(data: dict, single_step: bool = True, q_star: bool = False) → dict[source]¶

Overview:

QMIX网络的前向计算图。输入字典包括时间序列观察数据和相关数据，用于预测总q_value和每个代理的q_value。根据q_star参数决定是计算Q_tot还是Q_star。

Arguments:

data (dict): Input data dict with keys [‘obs’, ‘prev_state’, ‘action’].
- agent_state (torch.Tensor): 每个代理的时间序列局部观测数据。
- global_state (torch.Tensor): 时间序列全局观测数据。
- prev_state (list): 用于 q_network 或 _q_network_star 的先前rnn状态。
- 动作 (torch.Tensor 或 None): 如果动作为 None，则使用 argmax q_value 索引作为动作来计算 agent_q_act。
single_step (bool): 是否单步前进，如果是，则在前进前添加时间步维度，并在前进后移除它。
Q_star (bool): 是否使用Q_star网络进行前向传播。如果为True，则使用Q_star网络，其中代理网络具有与Q网络相同的架构但不共享参数，混合网络是一个具有3个256维隐藏层的前馈网络；如果为False，则使用Q网络，与Qmix论文中的Q网络相同。

Returns:

ret (dict): 输出数据字典，包含键 [total_q, logit, next_state].
total_q (torch.Tensor): 总q值，这是混合器网络的结果。
agent_q (torch.Tensor): 每个代理的q值。
next_state (list): 下一个RNN状态。

Shapes:

agent_state (torch.Tensor): \((T, B, A, N)\), 其中 T 是时间步，B 是批次大小，A 是代理数量，N 是观测形状。
全局状态 (torch.Tensor): \((T, B, M)\), 其中 M 是全局观测形状。
prev_state (list): 数学公式:(T, B, A), 一个长度为B的列表，每个元素是一个长度为A的列表。
动作 (torch.Tensor): \((T, B, A)\).
total_q (torch.Tensor): \((T, B)\).
agent_q (torch.Tensor): \((T, B, A, P)\), 其中 P 是动作形状。
next_state (list): 数学:(T, B, A), 一个长度为B的列表，每个元素是一个长度为A的列表。

PPG¶

class ding.model.PPG(obs_shape: int | SequenceType, action_shape: int | SequenceType, action_space: str = 'discrete', share_encoder: bool = True, encoder_hidden_size_list: SequenceType = [128, 128, 64], actor_head_hidden_size: int = 64, actor_head_layer_num: int = 2, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, impala_cnn_encoder: bool = False)[source]¶

Overview:: 来自论文Phasic Policy Gradient的Phasic Policy Gradient (PPG)模型 https://arxiv.org/abs/2009.04416 该模块包含VAC模块和一个辅助评论模块。
Interfaces:: forward, compute_actor, compute_critic, compute_actor_critic

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType, action_space: str = 'discrete', share_encoder: bool = True, encoder_hidden_size_list: SequenceType = [128, 128, 64], actor_head_hidden_size: int = 64, actor_head_layer_num: int = 2, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, impala_cnn_encoder: bool = False) → None[源代码]¶

Overview:

根据输入参数初始化PPG模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测值的形状，例如 128, (156, )。
action_shape (Union[int, SequenceType]): 动作的形状，例如 4, (3, )。
action_space (str): 动作空间类型，例如‘discrete’、‘continuous’。
share_encoder (bool): 是否共享编码器。
encoder_hidden_size_list (SequenceType): 编码器的隐藏大小列表。
actor_head_hidden_size (int): 传递给actor头的hidden_size。
actor_head_layer_num (int): 用于计算演员头Q值输出的网络层数。
critic_head_hidden_size (int): 传递给critic头的hidden_size。
critic_head_layer_num (int): 用于计算Q值输出的网络层数，用于critic head。
activation (Optional[nn.Module]): 在MLP中每个FC层后使用的激活函数类型，如果None则默认设置为nn.ReLU()。
norm_type (Optional[str]): 网络层（FC, Conv）后的归一化类型，详情请参见 ding.torch_utils.network。
impala_cnn_encoder (bool): 是否使用impala cnn编码器。

compute_actor(x: Tensor) → Dict[source]¶

Overview:

使用actor计算动作的logits。

Arguments:

x (torch.Tensor): 输入的观测张量数据。

Returns:

输出 (Dict): 包含动作逻辑的输出数据。

ReturnsKeys:

logit (torch.Tensor): 预测的动作logit张量，对于离散动作空间，它将是相同维度的可能动作选择的实值范围张量，对于连续动作空间，它将是高斯分布的mu和sigma，mu和sigma的数量与连续动作的数量相同。混合动作空间是离散和连续动作空间的一种组合，因此logit将是一个包含action_type和action_args的字典。

Shapes:

x (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是输入特征大小。
输出 (Dict): logit: \((B, A)\), 其中 B 是批量大小，A 是动作空间大小。

compute_actor_critic(x: Tensor) → Dict[来源]¶

Overview:

使用actor和critic来计算动作的logits和值。

Arguments:

x (torch.Tensor): 输入的观测张量数据。

Returns:

输出 (Dict): PPG前向计算图的输出字典，包括演员和评论家，包含 logit 和 value。

ReturnsKeys:

logit (torch.Tensor): 预测的动作logit张量，对于离散动作空间，它将是相同维度的可能动作选择的实值范围张量，对于连续动作空间，它将是高斯分布的mu和sigma，mu和sigma的数量与连续动作的数量相同。混合动作空间是离散和连续动作空间的一种组合，因此logit将是一个包含action_type和action_args的字典。
值 (torch.Tensor): 预测的状态值张量。

Shapes:

x (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是输入特征大小。
输出 (Dict): value: \((B, 1)\), 其中 B 是批量大小。
输出 (Dict): logit: \((B, A)\), 其中 B 是批量大小，A 是动作空间大小。

注意

compute_actor_critic 接口旨在在共享编码器时节省计算资源。

compute_critic(x: Tensor) → Dict[source]¶

Overview:

使用critic来计算值。

Arguments:

x (torch.Tensor): 输入的观测张量数据。

Returns:

输出 (Dict): VAC前向计算图的输出字典，包括 value。

ReturnsKeys:

必要的: value

Shapes:

x (torch.Tensor): \((B, N)\), 其中 B 是批量大小，N 是输入特征大小。
显示的内容为\((B, 1)\)，其中B是批量大小。

forward(inputs: Tensor | Dict, mode: str) → Dict[源代码]¶

Overview:

根据模式计算动作的logits或值，模式可以是compute_actor、compute_critic或compute_actor_critic。

Arguments:

x (torch.Tensor): 输入的观测张量数据。
模式 (str): 前向模式，所有模式都在这个类的开头定义。

Returns:

输出 (Dict): PPG前向计算图的输出字典，其键值因不同的mode而异。

过程克隆广度优先搜索¶

class ding.model.ProcedureCloningBFS(obs_shape: SequenceType, action_shape: int, encoder_hidden_size_list: SequenceType = [128, 128, 256, 256])[source]¶

Overview:: 在程序克隆（PC）中引入的神经网络用于处理三维观测。给定一个输入，该模型将执行多个3x3卷积，并输出一个与输入高度和宽度相同的特征图。输出的通道数将是action_shape。
Interfaces:: __init__, forward.

__init__(obs_shape: SequenceType, action_shape: int, encoder_hidden_size_list: SequenceType = [128, 128, 256, 256])[source]¶

Overview:

根据提供的参数初始化 BFSConvolution Encoder。

Arguments:

obs_shape (SequenceType): in_channel 的序列，加上一个或多个 input size，例如 [4, 84, 84]。
action_dim (int): 动作空间的形状，例如 6。
cnn_hidden_list (SequenceType): 每个块的CNN通道维度，例如 [128, 128, 256, 256]。

forward(x: Tensor) → Dict[source]¶

Overview:

计算图。给定一个三维观测值，此函数将返回一个具有相同高度和宽度的张量。输出的通道数将是action_shape。

Arguments:

x (torch.Tensor): 输入的观测张量数据。

Returns:

输出 (Dict): 模型前向计算图的输出字典，仅包含一个键 logit。

Examples:

>>> model = ProcedureCloningBFS([3, 16, 16], 4)
>>> inputs = torch.randn(16, 16, 3).unsqueeze(0)
>>> outputs = model(inputs)
>>> assert outputs['logit'].shape == torch.Size([16, 16, 4])

过程克隆MCTS¶

class ding.model.ProcedureCloningMCTS(obs_shape: SequenceType, action_dim: int, cnn_hidden_list: SequenceType = [128, 128, 256, 256, 256], cnn_activation: Module = ReLU(), cnn_kernel_size: SequenceType = [3, 3, 3, 3, 3], cnn_stride: SequenceType = [1, 1, 1, 1, 1], cnn_padding: SequenceType = [1, 1, 1, 1, 1], mlp_hidden_list: SequenceType = [256, 256], mlp_activation: Module = ReLU(), att_heads: int = 8, att_hidden: int = 128, n_att: int = 4, n_feedforward: int = 2, feedforward_hidden: int = 256, drop_p: float = 0.5, max_T: int = 17)[源代码]¶

Overview:: 与过程克隆（PC）相关的算法神经网络。
Interfaces:: __init__, forward.

__init__(obs_shape: SequenceType, action_dim: int, cnn_hidden_list: SequenceType = [128, 128, 256, 256, 256], cnn_activation: Module = ReLU(), cnn_kernel_size: SequenceType = [3, 3, 3, 3, 3], cnn_stride: SequenceType = [1, 1, 1, 1, 1], cnn_padding: SequenceType = [1, 1, 1, 1, 1], mlp_hidden_list: SequenceType = [256, 256], mlp_activation: Module = ReLU(), att_heads: int = 8, att_hidden: int = 128, n_att: int = 4, n_feedforward: int = 2, feedforward_hidden: int = 256, drop_p: float = 0.5, max_T: int = 17) → None[source]¶

Overview:

根据相应的输入参数初始化MCTS过程克隆模型。

Arguments:

obs_shape (SequenceType): 观测空间形状，例如 [4, 84, 84]。
action_dim (int): 动作空间的形状，例如6。
cnn_hidden_list (SequenceType): 每个块的CNN通道维度，例如 [128, 128, 256, 256, 256].
cnn_activation (nn.Module): cnn块的激活函数，例如 nn.ReLU()。
cnn_kernel_size (SequenceType): 每个cnn块的核大小，例如 [3, 3, 3, 3, 3]。
cnn_stride (SequenceType): 每个cnn块的步幅，例如 [1, 1, 1, 1, 1]。
cnn_padding (SequenceType): 每个cnn块的填充，例如 [1, 1, 1, 1, 1]。
mlp_hidden_list (SequenceType): 此参数的最后一个维度必须与 cnn_hidden_list 的最后一个维度匹配，例如 [256, 256]。
mlp_activation (nn.Module): MLP层的激活函数，例如 nn.ReLU()。
att_heads (int): 变压器中的注意力头数量，例如8。
att_hidden (int): transformer中的注意力维度数量，例如128。
n_att (int): transformer中的注意力块数量，例如4。
n_feedforward (int): 变压器中的前馈层数，例如2。
drop_p (float): 注意力的丢弃率，例如 0.5。
max_T (int): 程序克隆的序列长度，例如17。

forward(states: Tensor, goals: Tensor, actions: Tensor) → Tuple[Tensor, Tensor][source]¶

Overview:

ProcedureCloningMCTS 前向计算图，输入状态张量和目标张量，计算预测的状态和动作。

Arguments:

状态 (torch.Tensor): 当前时间的观察。
目标 (torch.Tensor): 一段时间后的目标观测值。
actions (torch.Tensor): 期间执行的动作。

Returns:

输出 (Tuple[torch.Tensor, torch.Tensor]): 预测的状态和动作。

Examples:

>>> inputs = {                 'states': torch.randn(2, 3, 64, 64),                 'goals': torch.randn(2, 3, 64, 64),                 'actions': torch.randn(2, 15, 9)             }
>>> model = ProcedureCloningMCTS(obs_shape=(3, 64, 64), action_dim=9)
>>> goal_preds, action_preds = model(inputs['states'], inputs['goals'], inputs['actions'])
>>> assert goal_preds.shape == (2, 256)
>>> assert action_preds.shape == (2, 16, 9)

ACER¶

class ding.model.ACER(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None)[source]¶

Overview:: 算法模型ACER（带有经验回放的演员评论家）带有经验回放的样本高效演员评论家。 https://arxiv.org/abs/1611.01224
Interfaces:: __init__, forward, compute_actor, compute_critic

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None) → None[源代码]¶

Overview:

根据参数初始化ACER模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测的空间。
action_shape (Union[int, SequenceType]): 动作的空间。
actor_head_hidden_size (Optional[int]): 传递给actor-nn的Head的hidden_size。
actor_head_layer_num (int):
网络中用于计算演员神经网络Q值输出的层数。
critic_head_hidden_size (Optional[int]): 传递给critic-nn的Head的hidden_size。
critic_head_layer_num (int):
网络中用于计算评论者神经网络Q值输出的层数。
activation (Optional[nn.Module]):
在MLP中使用的激活函数类型，在layer_fn之后，如果None则默认设置为nn.ReLU()
norm_type (Optional[str]):
使用的归一化类型，详情请参见ding.torch_utils.fc_block。

compute_actor(inputs: Tensor) → Dict[source]¶

Overview:

使用编码的嵌入张量来预测输出。使用compute_actor模式执行参数更新使用编码的嵌入张量来预测输出。

Arguments:

inputs (torch.Tensor):
编码后的嵌入张量，由给定的hidden_size决定，即(B, N=hidden_size)。 hidden_size = actor_head_hidden_size
模式 (str): 前向模式的名称。

Returns:

输出 (Dict): 前向传递编码器和头部的输出。

ReturnsKeys (either):

logit (torch.FloatTensor): \((B, N1)\), 其中 B 是批量大小，N1 是 action_shape

Shapes:

输入 (torch.Tensor): \((B, N0)\), B 是批量大小，N0 对应于 hidden_size
logit (torch.FloatTensor): \((B, N1)\), 其中 B 是批量大小，N1 是 action_shape

Examples:

>>> # Regression mode
>>> model = ACER(64, 64)
>>> inputs = torch.randn(4, 64)
>>> actor_outputs = model(inputs,'compute_actor')
>>> assert actor_outputs['logit'].shape == torch.Size([4, 64])

compute_critic(inputs: Tensor) → Dict[源代码]¶

Overview:

使用compute_critic模式执行参数更新使用编码的嵌入张量来预测输出。

Arguments:

obs, action 编码的张量。
模式 (str): 前向模式的名称。

Returns:

输出 (Dict): Q值输出。

ReturnKeys:

q_value (torch.Tensor): Q值张量，大小与批量大小相同。

Shapes:

obs (torch.Tensor): \((B, N1)\), 其中 B 是批量大小，N1 是 obs_shape
q_value (torch.FloatTensor): \((B, N2)\), 其中 B 是批量大小，N2 是 action_shape。

Examples:

>>> inputs =torch.randn(4, N)
>>> model = ACER(obs_shape=(N, ),action_shape=5)
>>> model(inputs, mode='compute_critic')['q_value']

forward(inputs: Tensor | Dict, mode: str) → Dict[source]¶

Overview:

使用观察来预测输出。参数更新使用ACER的MLPs前向设置。

Arguments:

模式 (str): 前向模式的名称。

Returns:

输出 (Dict): 网络前向的输出。

Shapes (Actor):

obs (torch.Tensor): \((B, N1)\), 其中 B 是批量大小，N1 是 obs_shape
logit (torch.FloatTensor): \((B, N2)\), 其中 B 是批量大小，N2 是 action_shape

Shapes (Critic):

输入 (torch.Tensor): \((B, N1)\), B 是批量大小，N1 对应于 obs_shape
q_value (torch.FloatTensor): \((B, N2)\), 其中 B 是批量大小，N2 是 action_shape

NGU¶

class ding.model.NGU(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], collector_env_num: int | None = 1, dueling: bool = True, head_hidden_size: int | None = None, head_layer_num: int = 1, lstm_type: str | None = 'normal', activation: Module | None = ReLU(), norm_type: str | None = None)[source]¶

Overview:: NGU策略的循环Q模型（https://arxiv.org/pdf/2002.06038.pdf），修改自q_leaning.py中的DRQN类。原始论文中提到的实现是“适应使用在卷积神经网络之后带有LSTM层的决斗网络架构的R2D2代理”。NGU网络包括编码器、LSTM核心（rnn）和头部。
Interface:: __init__, forward.

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType, encoder_hidden_size_list: SequenceType = [128, 128, 64], collector_env_num: int | None = 1, dueling: bool = True, head_hidden_size: int | None = None, head_layer_num: int = 1, lstm_type: str | None = 'normal', activation: Module | None = ReLU(), norm_type: str | None = None) → None[源代码]¶

Overview:

根据参数初始化NGU的DRQN模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测空间，例如 8 或 [4, 84, 84]。
action_shape (Union[int, SequenceType]): 动作的空间，例如 6 或 [2, 3, 3]。
encoder_hidden_size_list (SequenceType): 传递给 Encoder 的 hidden_size 集合。
collector_env_num (Optional[int]): 用于同时收集数据的环境数量。
决斗 (bool): 是否选择 DuelingHead (True) 或 DiscreteHead (False)，默认为 True。
head_hidden_size (Optional[int]): 传递给 Head 的 hidden_size，应与 encoder_hidden_size_list 的最后一个元素匹配。
head_layer_num (int): 头部网络中的层数。
lstm_type (Optional[str]): RNN单元的版本，目前支持 ['normal', 'pytorch', 'hpc', 'gru']，默认是 'normal'。
activation (Optional[nn.Module]):
在MLP中使用的激活函数类型，在layer_fn之后，如果None则默认设置为nn.ReLU()。
norm_type (Optional[str]):
使用的归一化类型，更多详情请参见ding.torch_utils.fc_block。

forward(inputs: Dict, inference: bool = False, saved_state_timesteps: list | None = None) → Dict[来源]¶

Overview:

NGU R2D2网络的前向计算图。输入观测值、前一个动作、前一个外部奖励，以预测NGU Q输出。使用NGU的多层感知器前向设置进行参数更新。

Arguments:

inputs (Dict):
- obs (torch.Tensor): 编码的观测值。
- prev_state (list): 前一个状态的大小为 (B, N) 的张量。
推理: (:obj:’bool’): 如果推理为True，我们展开一个时间步的转换，如果推理为False，我们展开序列转换。
saved_state_timesteps: (:obj:’Optional[list]’): 当推理为False时，我们会展开序列转换，然后我们会在列表saved_state_timesteps中列出的时间步长保存rnn隐藏状态。

Returns:

outputs (Dict):
使用DRQN设置运行MLP并返回结果预测字典。

ReturnsKeys:

logit (torch.Tensor): 与输入 obs 大小相同的Logit张量。
next_state (list): 下一个状态的大小为 (B, N) 的张量。

Shapes:

obs (torch.Tensor): \((B, N=obs_space)\), 其中 B 是批量大小。
prev_state(torch.FloatTensor list): \([(B, N)]\).
logit (torch.FloatTensor): \((B, N)\).
next_state(torch.FloatTensor list): \([(B, N)]\).

BCQ¶

class ding.model.BCQ(obs_shape: int | SequenceType, action_shape: int | SequenceType | EasyDict, actor_head_hidden_size: List = [400, 300], critic_head_hidden_size: List = [400, 300], activation: Module | None = ReLU(), vae_hidden_dims: List = [750, 750], phi: float = 0.05)[source]¶

Overview:: BCQ（批量约束深度Q学习）模型。无需探索的离线策略深度强化学习。 https://arxiv.org/abs/1812.02900
Interface:: forward, compute_actor, compute_critic, compute_vae, compute_eval
Property:: mode

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType | EasyDict, actor_head_hidden_size: List = [400, 300], critic_head_hidden_size: List = [400, 300], activation: Module | None = ReLU(), vae_hidden_dims: List = [750, 750], phi: float = 0.05) → None[source]¶

Overview:

初始化神经网络，即代理Q网络和演员。

Arguments:

obs_shape (int): 观测状态的维度
action_shape (int): 动作形状的维度
actor_hidden_size (list): actor的隐藏大小列表
critic_hidden_size (:obj:’list’): 评论者隐藏层大小的列表
activation (nn.Module): 网络中的激活函数，默认为 nn.ReLU()。
vae_hidden_dims (list): vae的隐藏层大小列表

compute_actor(inputs: Dict[str, Tensor]) → Dict[str, Tensor | Dict[str, Tensor]][source]¶

Overview:

使用演员网络来计算动作。

Arguments:

输入 (Dict): 输入字典数据，包括观测和动作张量。

Returns:

输出 (Dict): 包含关键字 action (torch.Tensor) 的字典。

Shapes:

输入 (Dict): \((B, N, D)\), 其中 B 是批量大小，N 是样本数量，D 是输入维度。
输出 (Dict): \((B, N)\).

Examples:

>>> inputs = {'obs': torch.randn(4, 32), 'action': torch.randn(4, 6)}
>>> model = BCQ(32, 6)
>>> outputs = model.compute_actor(inputs)

compute_critic(inputs: Dict[str, Tensor]) → Dict[str, Tensor][source]¶

Overview:

使用评论家网络来计算q值。

Arguments:

输入 (Dict): 输入字典数据，包括观测和动作张量。

Returns:

输出 (Dict): 包含关键字 q_value (torch.Tensor) 的字典。

Shapes:

输入 (Dict): \((B, N, D)\), 其中 B 是批量大小，N 是样本数量，D 是输入维度。
输出 (Dict): \((B, N)\).

Examples:

>>> inputs = {'obs': torch.randn(4, 32), 'action': torch.randn(4, 6)}
>>> model = BCQ(32, 6)
>>> outputs = model.compute_critic(inputs)

compute_eval(inputs: Dict[str, Tensor]) → Dict[str, Tensor][source]¶

Overview:

使用演员网络来计算动作。

Arguments:

输入 (Dict): 输入字典数据，包括观测和动作张量。

Returns:

输出 (Dict): 包含关键字 action (torch.Tensor) 的字典。

Shapes:

输入 (Dict): \((B, N, D)\), 其中 B 是批量大小，N 是样本数量，D 是输入维度。
输出 (Dict): \((B, N)\).

Examples:

>>> inputs = {'obs': torch.randn(4, 32), 'action': torch.randn(4, 6)}
>>> model = BCQ(32, 6)
>>> outputs = model.compute_eval(inputs)

compute_vae(inputs: Dict[str, Tensor]) → Dict[str, Tensor][source]¶

Overview:

使用vae网络来计算动作。

Arguments:

输入 (Dict): 输入字典数据，包括观测和动作张量。

Returns:

输出 (Dict): 包含关键词 recons_action (torch.Tensor), prediction_residual (torch.Tensor), input (torch.Tensor), mu (torch.Tensor), log_var (torch.Tensor) 和 z (torch.Tensor) 的字典。

Shapes:

输入 (Dict): \((B, N, D)\), 其中 B 是批量大小，N 是样本数量，D 是输入维度。
输出 (Dict): \((B, N)\).

Examples:

>>> inputs = {'obs': torch.randn(4, 32), 'action': torch.randn(4, 6)}
>>> model = BCQ(32, 6)
>>> outputs = model.compute_vae(inputs)

forward(inputs: Dict[str, Tensor], mode: str) → Dict[str, Tensor][source]¶

Overview:

BCQ方法的独特执行（前向）方法，可以指示不同的模式来实现不同的计算图，包括BCQ中的compute_actor和compute_critic。

Mode compute_actor:

Arguments:

输入 (Dict): 输入字典数据，包括观测和动作张量。

Returns:

输出 (Dict): 输出字典数据，包括动作张量。

Mode compute_critic:

Arguments:

输入 (Dict): 输入字典数据，包括观测和动作张量。

Returns:

输出 (Dict): 输出字典数据，包括q_value张量。

Mode compute_vae:

Arguments:

输入 (Dict): 输入字典数据，包括观测和动作张量。

Returns:

输出 (Dict): 包含关键词 recons_action (torch.Tensor), prediction_residual (torch.Tensor), input (torch.Tensor), mu (torch.Tensor), log_var (torch.Tensor) 和 z (torch.Tensor) 的字典。

Mode compute_eval:

Arguments:

输入 (Dict): 输入字典数据，包括观测和动作张量。

Returns:

输出 (Dict): 输出字典数据，包括动作张量。

Examples:

>>> inputs = {'obs': torch.randn(4, 32), 'action': torch.randn(4, 6)}
>>> model = BCQ(32, 6)
>>> outputs = model(inputs, mode='compute_actor')
>>> outputs = model(inputs, mode='compute_critic')
>>> outputs = model(inputs, mode='compute_vae')
>>> outputs = model(inputs, mode='compute_eval')

注意

对于具体示例，可以参考compute_actor和compute_critic的API文档。

EDAC¶

class ding.model.EDAC(obs_shape: int | SequenceType, action_shape: int | SequenceType | EasyDict, ensemble_num: int = 2, actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, **kwargs)[source]¶

Overview:: 在EDAC中使用的带有集成机制的Q值Actor-Critic网络。
Interfaces:: __init__, forward, compute_actor, compute_critic

__init__(obs_shape: int | SequenceType, action_shape: int | SequenceType | EasyDict, ensemble_num: int = 2, actor_head_hidden_size: int = 64, actor_head_layer_num: int = 1, critic_head_hidden_size: int = 64, critic_head_layer_num: int = 1, activation: Module | None = ReLU(), norm_type: str | None = None, **kwargs) → None[source]¶

Overview:

根据输入参数初始化EDAC模型。

Arguments:

obs_shape (Union[int, SequenceType]): 观测值的形状，例如 128, (156, )。
action_shape (Union[int, SequenceType, EasyDict]): 动作的形状，例如 4, (3, ), EasyDict({‘action_type_shape’: 3, ‘action_args_shape’: 4}).
ensemble_num (int): Q-net 数量。
actor_head_hidden_size (Optional[int]): 传递给actor head的hidden_size。
actor_head_layer_num (int): 用于计算演员头Q值输出的网络层数。
critic_head_hidden_size (Optional[int]): 传递给critic头的hidden_size。
critic_head_layer_num (int): 用于计算Q值输出的网络层数，用于critic head。
activation (Optional[nn.Module]): 在MLP中每个FC层后使用的激活函数类型，如果None则默认设置为nn.ReLU()。
norm_type (Optional[str]): 网络层（FC, Conv）后的归一化类型，详情请参见 ding.torch_utils.network。

compute_actor(obs: Tensor) → Dict[str, Tensor | Dict[str, Tensor]][source]¶

Overview:

compute_actor模式的前向计算图，使用观察张量来生成actor输出，例如action、logit等。

Arguments:

obs (torch.Tensor): 观测张量数据，现在支持一批一维向量数据，即 (B, obs_shape)。

Returns:

输出 (Dict[str, Union[torch.Tensor, Dict[str, torch.Tensor]]]): 根据动作空间变化的Actor输出: reparameterization.

ReturnsKeys (either):

logit (Dict[str, torch.Tensor]): Reparameterization logit, usually in SAC.
- mu (torch.Tensor): 参数化高斯分布的均值。
- sigma (torch.Tensor): 参数化高斯分布的标准差。

Shapes:

obs (torch.Tensor): \((B, N0)\), B 是批量大小，N0 对应于 obs_shape。
动作 (torch.Tensor): \((B, N1)\), B 是批量大小，N1 对应于 action_shape。
logit.mu (torch.Tensor): \((B, N1)\), B 是批量大小，N1 对应于 action_shape。
logit.sigma (torch.Tensor): \((B, N1)\), B 是批量大小。
logit (torch.Tensor): \((B, N2)\), B 是批量大小，N2 对应于 action_shape.action_type_shape。
action_args (torch.Tensor): \((B, N3)\), B 是批量大小，N3 对应于 action_shape.action_args_shape。

Examples:

>>> model = EDAC(64, 64,)
>>> obs = torch.randn(4, 64)
>>> actor_outputs = model(obs,'compute_actor')
>>> assert actor_outputs['logit'][0].shape == torch.Size([4, 64])  # mu
>>> actor_outputs['logit'][1].shape == torch.Size([4, 64]) # sigma

compute_critic(inputs: Dict[str, Tensor]) → Dict[str, Tensor][source]¶

Overview:

计算评论家模式的前向计算图，使用观察和动作张量来生成评论家输出，例如 q_value。

Arguments:

输入 (Dict[str, torch.Tensor]): 输入数据的字典结构，包括 obs 和 action 张量

Returns:

输出 (Dict[str, torch.Tensor]): 评论者输出，例如 q_value。

ArgumentsKeys:

obs: (torch.Tensor): 观测张量数据，现在支持一批一维向量数据。
动作 (Union[torch.Tensor, Dict]): 与 action_shape 大小相同的连续动作。

ReturnKeys:

q_value (torch.Tensor): Q值张量，大小与批量大小相同。

Shapes:

obs (torch.Tensor): \((B, N1)\) 或 ‘(Ensemble_num, B, N1)’，其中 B 是批量大小，N1 是 obs_shape。
动作 (torch.Tensor): \((B, N2)\) 或 '(Ensemble_num, B, N2)'，其中 B 是批量大小，N4 是 action_shape。
q_value (torch.Tensor): \((Ensemble_num, B)\), 其中 B 是批量大小。

Examples:

>>> inputs = {'obs': torch.randn(4, 8), 'action': torch.randn(4, 1)}
>>> model = EDAC(obs_shape=(8, ),action_shape=1)
>>> model(inputs, mode='compute_critic')['q_value']  # q value
... tensor([0.0773, 0.1639, 0.0917, 0.0370], grad_fn=<SqueezeBackward1>)

forward(inputs: Tensor | Dict[str, Tensor], mode: str) → Dict[str, Tensor][source]¶

Overview:

EDAC方法的独特执行（前向）方法，可以指示不同的模式来实现不同的计算图，包括EDAC中的compute_actor和compute_critic。

Mode compute_actor:

Arguments:

输入 (torch.Tensor): 观测数据，默认为张量。

Returns:

输出 (Dict): 输出字典数据，包括不同动作空间中的不同键值对。

Mode compute_critic:

Arguments:

输入 (Dict): 输入字典数据，包括观测和动作张量。

Returns:

输出 (Dict): 输出字典数据，包括q_value张量。

注意

对于具体示例，可以参考compute_actor和compute_critic的API文档。

EBM¶

class ding.model.EBM(obs_shape: int, action_shape: int, hidden_size: int = 512, hidden_layer_num: int = 4, **kwargs)[source]¶

Overview:: 基于能量的模型。
Interface:: __init__, forward

__init__(obs_shape: int, action_shape: int, hidden_size: int = 512, hidden_layer_num: int = 4, **kwargs)[source]¶

Overview:

初始化EBM。

Arguments:

obs_shape (int): 观测形状。
动作形状 (int): 动作形状。
hidden_size (int): 隐藏层大小。
hidden_layer_num (int): 隐藏层的数量。

forward(obs, action)[source]¶

Overview:

EBM的前向计算图。

Arguments:

obs (torch.Tensor): 形状为 (B, N, O) 的观测值。
动作 (torch.Tensor): 形状为 (B, N, A) 的动作。

Returns:

pred (torch.Tensor): 形状为 (B, N) 的能量。

Examples:

>>> obs = torch.randn(2, 3, 4)
>>> action = torch.randn(2, 3, 5)
>>> ebm = EBM(4, 5)
>>> pred = ebm(obs, action)

自回归EBM¶

class ding.model.AutoregressiveEBM(obs_shape: int, action_shape: int, hidden_size: int = 512, hidden_layer_num: int = 4)[source]¶

Overview:: 自回归能量模型。
Interface:: __init__, forward

__init__(obs_shape: int, action_shape: int, hidden_size: int = 512, hidden_layer_num: int = 4)[source]¶

Overview:

初始化自回归EBM。

Arguments:

obs_shape (int): 观测形状。
动作形状 (int): 动作形状。
hidden_size (int): 隐藏层大小。
hidden_layer_num (int): 隐藏层的数量。

forward(obs, action)[来源]¶

Overview:

自回归EBM的前向计算图。

Arguments:

obs (torch.Tensor): 形状为 (B, N, O) 的观测值。
动作 (torch.Tensor): 形状为 (B, N, A) 的动作。

Returns:

pred (torch.Tensor): 形状为 (B, N, A) 的能量。

Examples:

>>> obs = torch.randn(2, 3, 4)
>>> action = torch.randn(2, 3, 5)
>>> arebm = AutoregressiveEBM(4, 5)
>>> pred = arebm(obs, action)

变分自编码器¶

class ding.model.VanillaVAE(action_shape: int, obs_shape: int, latent_size: int, hidden_dims: List = [256, 256], **kwargs)[source]¶

Overview:: 实现用于动作重建的Vanilla变分自编码器。
Interfaces:: __init__, encode, decode, decode_with_obs, reparameterize, forward, loss_function .

__init__(action_shape: int, obs_shape: int, latent_size: int, hidden_dims: List = [256, 256], **kwargs) → None[来源]¶: 初始化内部模块状态，由nn.Module和ScriptModule共享。

decode(z: Tensor, obs_encoding: Tensor) → Dict[str, Any][source]¶

Overview:

将给定的潜在动作和obs_encoding映射到原始动作空间。

Arguments:

z (torch.Tensor): 采样的潜在动作
obs_encoding (torch.Tensor): 观测编码

Returns:

输出 (Dict): DQN 前向输出，例如 q_value。

ReturnsKeys:

reconstruction_action (torch.Tensor): 重建动作。
预测残差 (torch.Tensor): 预测残差。

Shapes:

z (torch.Tensor): \((B, L)\), 其中 B 是批量大小，L 是 latent_size
obs_encoding (torch.Tensor): \((B, H)\), 其中 B 是批量大小，H 是 hidden dim

decode_with_obs(z: Tensor, obs: Tensor) → Dict[str, Any][source]¶

Overview:

将给定的潜在动作和观测映射到原始动作空间。使用self.encode_obs_head(obs)方法获取观测编码。

Arguments:

z (torch.Tensor): 采样的潜在动作
obs (torch.Tensor): 观测值

Returns:

输出 (Dict): DQN 前向输出，例如 q_value。

ReturnsKeys:

reconstruction_action (torch.Tensor): 由VAE重建的动作。
预测残差 (torch.Tensor): 由VAE预测的观测值。

Shapes:

z (torch.Tensor): \((B, L)\), 其中 B 是批量大小，L 是 latent_size
obs (torch.Tensor): \((B, O)\), 其中 B 是批量大小，O 是 obs_shape

encode(input: Dict[str, Tensor]) → Dict[str, Any][source]¶

Overview:

通过编码器网络对输入进行编码并返回潜在代码。

Arguments:

输入 (Dict): 包含关键字 obs (torch.Tensor) 和 action (torch.Tensor) 的字典，分别表示观察值和代理的动作。

Returns:

输出 (Dict): 包含关键字 mu (torch.Tensor), log_var (torch.Tensor) 和 obs_encoding (torch.Tensor) 的字典，表示潜在编码。

Shapes:

obs (torch.Tensor): \((B, O)\), 其中 B 是批量大小，O 是 observation dim。
动作 (torch.Tensor): \((B, A)\), 其中 B 是批量大小，A 是 action dim。
mu (torch.Tensor): \((B, L)\), 其中 B 是批量大小，L 是 latent size。
log_var (torch.Tensor): \((B, L)\), 其中 B 是批量大小，L 是 latent size。
obs_encoding (torch.Tensor): \((B, H)\), 其中 B 是批量大小，H 是 hidden dim。

forward(input: Dict[str, Tensor], **kwargs) → dict[来源]¶

Overview:

对输入进行编码，重新参数化 mu 和 log_var，解码 obs_encoding。

Argumens:

输入 (Dict): 包含关键字 obs (torch.Tensor) 和 action (torch.Tensor) 的字典，分别表示观察值和代理的动作。

Returns:

输出 (Dict): 包含关键词 recons_action (torch.Tensor), prediction_residual (torch.Tensor), input (torch.Tensor), mu (torch.Tensor), log_var (torch.Tensor) 和 z (torch.Tensor) 的字典。

Shapes:

recons_action (torch.Tensor): \((B, A)\), 其中 B 是批量大小，A 是 action dim。
预测残差 (torch.Tensor): \((B, O)\), 其中 B 是批量大小，O 是 observation dim。
mu (torch.Tensor): \((B, L)\), 其中 B 是批量大小，L 是 latent size。
log_var (torch.Tensor): \((B, L)\), 其中 B 是批量大小，L 是 latent size。
z (torch.Tensor): \((B, L)\), 其中 B 是批量大小，L 是 latent_size

loss_function(args: Dict[str, Tensor], **kwargs) → Dict[str, Tensor][source]¶

Overview:

计算VAE损失函数。

Arguments:

args (Dict[str, Tensor]): 包含关键词 recons_action, prediction_residual original_action, mu, log_var 和 true_residual 的字典。
kwargs (Dict): 包含关键字 kld_weight 和 predict_weight 的字典。

Returns:

输出 (Dict[str, Tensor]): 包含不同 loss 结果的字典，包括 loss, reconstruction_loss, kld_loss, predict_loss.

Shapes:

recons_action (torch.Tensor): \((B, A)\), 其中 B 是批量大小，A 是 action dim。
预测残差 (torch.Tensor): \((B, O)\), 其中 B 是批量大小，O 是 observation dim。
original_action (torch.Tensor): \((B, A)\), 其中 B 是批量大小，A 是 action dim。
mu (torch.Tensor): \((B, L)\), 其中 B 是批量大小，L 是 latent size。
log_var (torch.Tensor): \((B, L)\), 其中 B 是批量大小，L 是 latent size。
true_residual (torch.Tensor): \((B, O)\), 其中 B 是批量大小，O 是 observation dim。

reparameterize(mu: Tensor, logvar: Tensor) → Tensor[来源]¶

Overview:

从N(0,1)中采样N(mu, var)的重参数化技巧。

Arguments:

mu (torch.Tensor): 潜在高斯的均值
logvar (torch.Tensor): 潜在高斯分布的标准差

Shapes:

mu (torch.Tensor): \((B, L)\), 其中 B 是批量大小，L 是 latnet_size
logvar (torch.Tensor): \((B, L)\), 其中 B 是批量大小，L 是 latnet_size

包装器¶

请参考 ding/model/wrapper 获取更多详细信息。

IModelWrapper¶

class ding.model.IModelWrapper(model: Module)[源代码]¶

Overview:: 模型包装器的基本接口类。模型包装器是torch.nn.Module模型的包装类，用于为包装的模型添加一些额外的操作，例如为基于RNN的模型维护隐藏状态，为离散动作空间选择argmax动作等。
Interfaces:: __init__, __getattr__, info, reset, forward.

__getattr__(key: str) → Any[source]¶

Overview:

获取torch.nn.Module模型的原始属性，例如模型中定义的变量和方法。

Arguments:

键 (str): 要查询的字符串键。

Returns:

ret (Any): 查询的属性。

__init__(model: Module) → None[源代码]¶

Overview:: 在模型包装器中初始化模型和其他必要的成员变量。

forward(*args, **kwargs) → Any[源代码]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

info(attr_name: str) → str[source]¶

Overview:

获取指定的attr_name的一些字符串信息，这些信息用于调试包装器。此方法将递归搜索指定的attr_name。

Arguments:

attr_name (str): 查询信息的字符串键。

Returns:

info_string (str): 指定 attr_name 的信息字符串。

reset(data_id: List[int] | None = None, **kwargs) → None[source]¶

Overview

基本接口，重置模型包装器中的一些状态变量，例如RNN的隐藏状态。这里我们不做任何操作，只是实现这个接口方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

Arguments:

data_id (List[int]): 要重置的数据ID列表。如果为None，则重置所有数据。实际上，模型包装器通常需要为每个数据轨迹维护一些状态变量，因此我们保留这个data_id参数来重置指定数据的状态变量。

模型包装¶

ding.model.model_wrap(model: Module | IModelWrapper, wrapper_name: str | None = None, **kwargs)[source]¶

Overview:

使用指定的包装器包装模型并返回包装后的模型。

Arguments:

模型 (Any): 要包装的模型。
wrapper_name (str): 要使用的包装器的名称。

注意

包装器的参数应作为kwargs传入。

register_wrapper¶

ding.model.register_wrapper(name: str, wrapper_type: type) → None[源代码]¶

Overview:

注册新的包装器到wrapper_name_map。当用户实现一个新的包装器时，他们必须调用此函数来完成注册。然后该包装器可以通过model_wrap调用。

Arguments:

名称 (str): 要注册的新包装器的名称。
wrapper_type (type): 需要在 wrapper_name_map 中添加包装类。此参数应为 IModelWrapper 的子类。

基础模型包装器¶

class ding.model.wrapper.model_wrappers.BaseModelWrapper(model: Module)[source]¶

Overview:: 模型包装器的占位符类。此类用于包装模型而不进行任何额外操作，包括一个空的reset方法和一个直接调用包装模型前向的forward方法。为了保持模型包装器接口的一致性，我们在DI-engine策略的实现中使用此类来包装模型而不进行特定操作。

forward(*args, **kwargs) → Any¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

reset(data_id: List[int] | None = None, **kwargs) → None¶

Overview

基本接口，重置模型包装器中的一些状态变量，例如RNN的隐藏状态。这里我们不做任何操作，只是实现这个接口方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

Arguments:

data_id (List[int]): 要重置的数据ID列表。如果为None，则重置所有数据。实际上，模型包装器通常需要为每个数据轨迹维护一些状态变量，因此我们保留这个data_id参数来重置指定数据的状态变量。

ArgmaxSampleWrapper¶

class ding.model.wrapper.model_wrappers.ArgmaxSampleWrapper(model: Module)[源代码]¶

Overview:: 用于帮助模型采样argmax动作。
Interfaces:: forward.

forward(*args, **kwargs)[source]¶

Overview:: 使用模型前向计算图，并使用输出logit贪婪地选择最大动作（argmax）。

多项式样本包装器¶

class ding.model.wrapper.model_wrappers.MultinomialSampleWrapper(model: Module)[源代码]¶

Overview:: 用于帮助模型从输出['logits']中获取相应的动作。
Interfaces:: forward.

forward(*args, **kwargs)[来源]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

EpsGreedySampleWrapper¶

class ding.model.wrapper.model_wrappers.EpsGreedySampleWrapper(model: Module)[source]¶

Overview:: 在collector_model中使用的Epsilon贪婪采样器，用于帮助平衡探索和利用。 eps的类型可以根据不同的算法而变化，例如： - float（即python原生标量）：适用于几乎正常的情况 - Dict[str, float]：适用于算法NGU
Interfaces:: forward.

forward(*args, **kwargs)[source]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

EpsGreedyMultinomialSampleWrapper¶

class ding.model.wrapper.model_wrappers.EpsGreedyMultinomialSampleWrapper(model: Module)[source]¶

Overview:: Epsilon贪婪采样器与多项式采样结合使用于collector_model中，以帮助平衡探索和利用。
Interfaces:: forward.

forward(*args, **kwargs)[源代码]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

确定性样本包装器¶

class ding.model.wrapper.model_wrappers.DeterministicSampleWrapper(model: Module)[source]¶

Overview:: 在eval_model中使用的确定性采样器（直接使用mu）。
Interfaces:: 前进

forward(*args, **kwargs)[来源]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

ReparamSampleWrapper¶

class ding.model.wrapper.model_wrappers.ReparamSampleWrapper(model: Module)[源代码]¶

Overview:: 在collector_model中使用的重新参数化高斯采样器。
Interfaces:: 前进

forward(*args, **kwargs)[source]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

CombinationArgmaxSampleWrapper¶

class ding.model.wrapper.model_wrappers.CombinationArgmaxSampleWrapper(model: Module)[source]¶

Overview:: 用于帮助模型采样组合argmax动作。
Interfaces:: forward.

forward(shot_number, *args, **kwargs)[source]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

组合多项式样本包装器¶

class ding.model.wrapper.model_wrappers.CombinationMultinomialSampleWrapper(model: Module)[源代码]¶

Overview:: 用于帮助模型对组合多项动作进行采样。
Interfaces:: forward.

forward(shot_number, *args, **kwargs)[来源]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

混合Argmax样本包装器¶

class ding.model.wrapper.model_wrappers.HybridArgmaxSampleWrapper(model: Module)[源代码]¶

Overview:: 用于帮助模型在混合动作空间中采样argmax动作，即{'action_type': 离散, 'action_args', 连续}
Interfaces:: forward.

forward(*args, **kwargs)[source]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

混合Epsilon贪婪样本包装器¶

class ding.model.wrapper.model_wrappers.HybridEpsGreedySampleWrapper(model: Module)[source]¶

Overview:: 在collector_model中使用的Epsilon贪婪采样器，用于帮助平衡探索和利用。在混合动作空间中，即{'action_type': 离散, 'action_args': 连续}
Interfaces:: forward.

forward(*args, **kwargs)[source]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

混合Epsilon贪婪多项式采样包装器¶

class ding.model.wrapper.model_wrappers.HybridEpsGreedyMultinomialSampleWrapper(model: Module)[源代码]¶

Overview:: Epsilon贪婪采样器与多项式采样器结合使用于collector_model中，以帮助平衡探索和利用。在混合动作空间中，即{'action_type': 离散, 'action_args': 连续}
Interfaces:: forward.

forward(*args, **kwargs)[source]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

混合重参数多项式采样包装器¶

class ding.model.wrapper.model_wrappers.HybridReparamMultinomialSampleWrapper(model: Module)[source]¶

Overview:: 重参数化采样器与多项式采样器结合使用于collector_model中，以帮助平衡探索和利用。在混合动作空间中，即{'action_type': 离散, 'action_args': 连续}
Interfaces:: 前进

forward(*args, **kwargs)[source]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

混合确定性Argmax样本包装器¶

class ding.model.wrapper.model_wrappers.HybridDeterministicArgmaxSampleWrapper(model: Module)[source]¶

Overview:: 确定性采样器与argmax采样结合用于eval_model。在混合动作空间中，即{'action_type': 离散, 'action_args': 连续}
Interfaces:: 前进

forward(*args, **kwargs)[来源]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

ActionNoiseWrapper¶

class ding.model.wrapper.model_wrappers.ActionNoiseWrapper(model: Any, noise_type: str = 'gauss', noise_kwargs: dict = {}, noise_range: dict | None = None, action_range: dict | None = {'max': 1, 'min': -1})[源代码]¶

Overview:

向收集器的动作输出添加噪声；在添加噪声后对生成的噪声和动作进行剪辑。

Interfaces:

__init__, forward.

Arguments:

模型 (Any): 包装的模型类。应包含 forward 方法。
噪声类型 (str): 应该生成的噪声类型，支持 [‘gauss’, ‘ou’]。
noise_kwargs (dict): 应该在噪声初始化中使用的关键字参数。取决于 noise_type。
噪声范围 (Optional[dict]): 噪声的范围，用于裁剪。
action_range (Optional[dict]): 动作范围 + 噪声，用于裁剪，默认裁剪到 [-1, 1]。

__init__(model: Any, noise_type: str = 'gauss', noise_kwargs: dict = {}, noise_range: dict | None = None, action_range: dict | None = {'max': 1, 'min': -1}) → None[source]¶

Overview:: 在模型包装器中初始化模型和其他必要的成员变量。

forward(*args, **kwargs)[source]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

目标网络包装器¶

class ding.model.wrapper.model_wrappers.TargetNetworkWrapper(model: Any, update_type: str, update_kwargs: dict)[源代码]¶

Overview:: 维护和更新目标网络
Interfaces:: 更新，重置

__init__(model: Any, update_type: str, update_kwargs: dict)[源代码]¶

Overview:: 在模型包装器中初始化模型和其他必要的成员变量。

forward(*args, **kwargs) → Any¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

隐藏状态包装器¶

class ding.model.wrapper.model_wrappers.HiddenStateWrapper(model: ~typing.Any, state_num: int, save_prev_state: bool = False, init_fn: ~typing.Callable = <function HiddenStateWrapper.<lambda>>)[source]¶

Overview:: 维护基于RNN模型的隐藏状态。批次中的每个样本都有其自己的状态。
Interfaces:: __init__, reset, forward.

__init__(model: ~typing.Any, state_num: int, save_prev_state: bool = False, init_fn: ~typing.Callable = <function HiddenStateWrapper.<lambda>>) → None[source]¶

Overview:

维护基于RNN模型的隐藏状态。批次中的每个样本都有其自己的状态。初始化维护状态和状态函数；然后用自动保存的数据['prev_state']输入包装model.forward方法，并创建model.reset方法。

Arguments:

model(Any): 包装的模型类，应包含forward方法。
state_num (int): 要处理的状态数量。
save_prev_state (bool): 是否在输出中包含前一个状态。
init_fn (Callable): 用于在初始化和重置时初始化每个隐藏状态的函数，默认返回隐藏状态为None。

注意

此助手必须处理包含部分样本的实际批次，例如：6个状态数为8的样本。
此助手必须处理单个样本状态重置。

forward(data, **kwargs)[源代码]¶

Overview:: 基本接口，调用封装模型的forward方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

reset(*args, **kwargs)[源代码]¶

Overview

基本接口，重置模型包装器中的一些状态变量，例如RNN的隐藏状态。这里我们不做任何操作，只是实现这个接口方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

Arguments:

data_id (List[int]): 要重置的数据ID列表。如果为None，则重置所有数据。实际上，模型包装器通常需要为每个数据轨迹维护一些状态变量，因此我们保留这个data_id参数来重置指定数据的状态变量。

TransformerInputWrapper¶

class ding.model.wrapper.model_wrappers.TransformerInputWrapper(model: ~typing.Any, seq_len: int, init_fn: ~typing.Callable = <function TransformerInputWrapper.<lambda>>)[源代码]¶

__init__(model: ~typing.Any, seq_len: int, init_fn: ~typing.Callable = <function TransformerInputWrapper.<lambda>>) → None[source]¶

Overview:

给定N为Transformer模型接收到的序列长度，保持最后N-1个输入观察值。这样我们可以在每一步提供Transformer计算其输出所需的所有观察值。我们需要这样做，因为一些方法如‘collect’和‘evaluate’每步只提供一个观察值，并且没有过去观察值的记忆，但Transformer需要一个N个观察值的序列。包装方法forward将输入观察值保存在长度为N的FIFO内存中，方法reset将重置内存。通过调用方法reset_input，空的内存空间将用‘init_fn’或零初始化。由于不同的环境可能在不同的步骤终止，方法reset_memory_entry只初始化批次大小中特定环境的内存。

Arguments:

模型 (Any): 包装的模型类，应包含 forward 方法。
seq_len (int): 要记住的过去观察值的数量。
init_fn (Callable): 用于在初始化和重置时初始化每个内存位置的函数。

forward(input_obs: Tensor, only_last_logit: bool = True, data_id: List | None = None, **kwargs) → Dict[str, Tensor][source]¶

Arguments:

input_obs (torch.Tensor): 输入观测值，无序列形状: (bs, *obs_shape).
only_last_logit (bool): 如果为True，'logit'仅包含与当前观察对应的输出（形状：bs, embedding_dim），否则logit的形状为(seq_len, bs, embedding_dim)。
data_id (List): 当前正在运行的环境的id。内存更新和logits返回仅对这些环境有效。如果为None，则认为所有环境都在运行。

Returns:

包含存储在内存中的输入序列‘input_seq’和变压器输出‘logit’的字典。

reset(*args, **kwargs)[source]¶

Overview

基本接口，重置模型包装器中的一些状态变量，例如RNN的隐藏状态。这里我们不做任何操作，只是实现这个接口方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

Arguments:

data_id (List[int]): 要重置的数据ID列表。如果为None，则重置所有数据。实际上，模型包装器通常需要为每个数据轨迹维护一些状态变量，因此我们保留这个data_id参数来重置指定数据的状态变量。

TransformerSegmentWrapper¶

class ding.model.wrapper.model_wrappers.TransformerSegmentWrapper(model: Any, seq_len: int)[源代码]¶

__init__(model: Any, seq_len: int) → None[source]¶

Overview:

给定T为轨迹的长度，N为Transformer模型接收的序列长度，将T分割成N个元素的序列，并逐个前向传播每个序列。如果T % N != 0，最后一个序列将用零填充。通常在Transformer训练阶段使用。

Arguments:

模型 (Any): 包装的模型类，应包含 forward 方法。
seq_len (int): N, 序列的长度。

forward(obs: Tensor, **kwargs) → Dict[str, Tensor][source]¶

Arguments:

数据 (dict): 字典类型的数据，至少包括 [‘main_obs’, ‘target_obs’, ‘action’, ‘reward’, ‘done’, ‘weight’]

Returns:

包含每个序列模型输出的字典列表。

TransformerMemoryWrapper¶

class ding.model.wrapper.model_wrappers.TransformerMemoryWrapper(model: Any, batch_size: int)[source]¶

__init__(model: Any, batch_size: int) → None[source]¶

Overview:

Stores a copy of the Transformer memory in order to be reused across different phases. To make it more: 清楚起见，假设训练管道分为三个阶段：评估、收集、学习。包装器的目标是保持每个阶段结束时内存的内容，并在再次执行相同阶段时重复使用它。通过这种方式，它防止了不同阶段相互干扰内存。

Arguments:

模型 (Any): 包装的模型类，应包含 forward 方法。
batch_size (int): 内存批处理大小。

forward(*args, **kwargs) → Dict[str, Tensor][来源]¶

Arguments:

数据 (dict): 字典类型的数据，至少包括 [‘main_obs’, ‘target_obs’, ‘action’, ‘reward’, ‘done’, ‘weight’]

Returns:

前向方法的输出。

reset(*args, **kwargs)[source]¶

Overview

基本接口，重置模型包装器中的一些状态变量，例如RNN的隐藏状态。这里我们不做任何操作，只是实现这个接口方法。其他派生的模型包装器可以重写此方法以添加一些额外的操作。

Arguments:

data_id (List[int]): 要重置的数据ID列表。如果为None，则重置所有数据。实际上，模型包装器通常需要为每个数据轨迹维护一些状态变量，因此我们保留这个data_id参数来重置指定数据的状态变量。