在强化学习中,需要根据决策问题和策略选择合适的神经网络。DI-engine中,神经网络模型可以通过两种方式指定:
-
使用配置文件中的cfg.policy.model自动生成默认模型。这种方式下,可以在配置文件中指定神经网络的类型(MLP、CNN等)以及超参数(隐层大小、激活函数等),DI-engine会根据这些配置自动构建神经网络模型。这种方式简单易用,适用于常见的标准网络结构。
-
自定义模型实例并传入Policy。这种方式下,需要用户自己定义Tensorflow/Pytorch模型类,实现前向传播等接口,然后将实例传入Policy中。这种方式灵活度高,用户可以自由设计任意结构的神经网络。但是需要用户比较熟悉网络定义和 Tensorflow/Pytorch接口。
（注：在强化学习中，策略（Policy）是指智能体（Agent）决策的规则。策略是从状态（State）到动作（Action）的映射，它定义了在给定的状态下，智能体应该采取什么动作。策略可以是确定性的（Deterministic）也可以是随机性的（Stochastic）。）
以上两种方式都会在Policy中封装为neural_net属性,策略学习会通过这个网络完成状态的 embedding 以及动作的选择。这套机制和接口为用户提供了必要的灵活性,可以根据具体问题和需求配置各种神经网络模型。
这是红色的文字Policy 默认使用的模型是什么
DI-engine 中已经实现的 policy,默认使用 default_model 方法中表明的神经网络模型,例如在 SACPolicy 中:
@POLICY_REGISTRY.register('sac')
class SACPolicy(Policy):
...
def default_model(self) -> Tuple[str, List[str]]:
if self._cfg.multi_agent:
return 'maqac_continuous', ['ding.model.template.maqac']
else:
return 'qac', ['ding.model.template.qac']
...
在这段代码中，我们看到的是一个名为 DI-engine 的强化学习框架中的一个策略（Policy）类的一部分。具体来说，它定义了一个使用Soft Actor-Critic, SAC 算法的策略类。这个段落描述了如何在这个框架内设置和使用策略相关的神经网络模型。
让我们逐步解释这段代码：
-
@POLICY_REGISTRY.register('sac')是一个装饰器，它将SACPolicy类注册到一个名为POLICY_REGISTRY的注册器中，并且用'sac'作为这个策略的标识符。这样的注册机制允许框架能够根据名字轻松地查找和实例化策略。 -
class SACPolicy(Policy):表明SACPolicy是从更一般的Policy类派生的，它是一个具体的策略实现，使用了 SAC 算法。 -
default_model是SACPolicy类的一个方法，它定义了该策略默认使用的模型。这个方法返回两个元素的元组：'maqac_continuous'或'qac'：这是在模型注册器中注册的模型名字。根据配置是否是多智能体（multi_agent），它返回不同的模型名。['ding.model.template.maqac']或['ding.model.template.qac']：这是包含模型类的文件路径的列表。这个路径告诉 DI-engine 在哪里可以找到定义模型的代码。
-
当使用配置文件时，DI-engine 的入口文件将使用
cfg.policy.model中的参数（比如obs_shape,action_shape）来实例化提供的模型类。这个过程是自动化的，意味着用户定义好配置，DI-engine 将负责根据这些配置创建并初始化模型。 -
模型类会根据传入的参数构造适当的神经网络。例如，如果传入的
obs_shape参数表明观测是一个图像，则模型可能会使用卷积层来处理输入；如果观测是一个向量，则可能使用全连接层。
简而言之，这段代码展示了 DI-engine 如何灵活地处理不同类型的策略和模型，以及如何通过配置文件来方便地自定义和实例化这些策略和模型。这种设计允许研究者和开发者能够轻松试验不同的算法和模型架构，而无需直接修改代码。
如何自定义神经网络模型
在 DI-engine 强化学习框架中，每个策略（如 SACPolicy）通常有一个关联的默认模型（通过 default_model 方法指定），这个默认模型是为特定类型的任务设计的。例如，原始的 qac 模型可能是为处理具有一维观测空间的环境设计的，即观测是一个向量。
但是！如果任务是在一个模拟器（如 dmc2gym，一个DeepMind Control Suite到OpenAI Gym接口的适配器）上运行，并且任务是 cartpole-swingup，而且你希望使用观测为像素的输入（即观测是一个图像），那么默认的 qac 模型不足以处理这样的高维度和多通道的输入。在这种情况下，观测空间的形状是 (3, height, width)，其中 3 表示图像的颜色通道数（RGB），height 和 width 分别表示图像的高度和宽度。
在 dmc2gym 文档中，from_pixel 参数设定为 True 意味着环境将提供像素级的观测，而 channels_first 设定为 True 表明图像的通道维度是第一维（这是PyTorch等深度学习库通常采用的格式）。
面对这样的情况，如果你想要使用 SAC 算法处理像素级的观测，你需要自定义一个能够处理这种高维观测的模型。所以我们创建一个新的模型类，该类在内部使用卷积神经网络（CNN）来处理输入的图像数据，并适当地修改网络架构以适应任务的特定要求。
自定义模型完成后，可以将这个模型应用到 SACPolicy 中，替换原本的 qac 模型。涉及到以下几个步骤：
- 实现一个新的模型类，它继承自某个基础模型类，并覆盖必要的方法以支持像素级输入。
- 在策略配置中指定你的自定义模型，以便 DI-engine 使用你提供的模型而不是默认模型。
- 确保你的自定义模型注册到 DI-engine 的模型注册器中，这样框架可以识别和使用它。
自定义 model 基本步骤
1. 明确环境 (env) 和策略 (policy)
首先，需要确定你的强化学习任务的具体环境和任务。例如，我们选择 dmc2gym 环境中的 cartpole-swingup 任务，并且决定观测将以像素数据的形式提供，我们的观测空间是一个图像，其形状为 (3, height, width)。下面我们使用 SAC 算法来进行学习。
在这里，from_pixel = True 表明环境将提供基于像素的观测，channels_first = True 表明图像数据的通道维度在前，这通常是深度学习库（如 PyTorch）的标准格式。
2. 查阅策略中的 default_model 是否适用
接下来，需要检查选择的策略是否具有适用于任务的默认模型。这可以通过查看策略的文档或直接查阅源代码来完成。以 DI-engine 中的 SAC 策略为例，可以查看 SACPolicy 类中的 default_model 方法来了解默认模型：
@POLICY_REGISTRY.register('sac')
class SACPolicy(Policy):
...
def default_model(self) -> Tuple[str, List[str]]:
if self._cfg.multi_agent:
return 'maqac_continuous', ['ding.model.template.maqac']
else:
return 'qac', ['ding.model.template.qac']
...
如果进一步看一下 ding.model.template.qac 中的 QAC 模型，咱们可能会发现它仅支持一维的观测空间，而不支持像 (3, height, width) 这样的图像形状。这意味着对于我们的 cartpole-swingup 任务，需要创建一个自定义模型来处理像素级的观测。
3. 自定义模型 (custom_model) 实现
自定义模型的实现需要遵循一些基本的原则，以确保与 DI-engine 框架的兼容性。
a. 实现功能
自定义模型需要实现默认模型中的所有公共方法。包括：
__init__: 构造函数，对模型的各个部分进行初始化。forward: 定义模型如何从输入到输出的前向传递。compute_actor: 计算策略网络的输出，即给定观测值时的动作。compute_critic: 计算价值网络的输出，即动作的价值。
b. 保持返回值类型一致
自定义模型的方法需要保证与原始默认模型的返回值类型一致，以便于替换使用。
c. 利用已实现的 encoder 和 head
在 ding/model/common 下有多种 encoder 和 head 的实现，可用于构建不同部分的模型：
- Encoder: 负责对输入数据进行编码，使其适合后续的处理。例如，
ConvEncoder用于处理图像观测输入，FCEncoder用于处理一维观测输入。
- Head: 负责对编码后的数据进行处理，输出策略所需信息或辅助强化学习过程。有各种 head 可供选择，如
DiscreteHead、DuelingHead、RegressionHead等。
d. 自定义模型实现示例
例如这里需要自定义针对 sac+dmc2gym+cartpole-swingup 任务的 model ,我们把新的 custom_model 实现为 QACPixel 类
这里对照 QAC已经实现的方法, QACPixel需要实现 init , forward,以及 compute_actor和 compute_critic。
@MODEL_REGISTRY.register('qac')
class QAC(nn.Module):
...
def __init__(self, ...) -> None:
...
def forward(self, ...) -> Dict[str, torch.Tensor]:
...
def compute_actor(self, obs: torch.Tensor) -> Dict[str, Union[torch.Tensor, Dict[str, torch.Tensor]]]:
...
def compute_critic(self, inputs: Dict[str, torch.Tensor]) -> Dict[str, torch.Tensor]:
...
让我们一起来分析一下：
该代码段是一个Python类的声明，它定义了一个名为 QAC 的神经网络模型，用于在深度强化学习中作为 actor-critic 方法的一部分。此模型继承自 nn.Module，后者是PyTorch深度学习框架中的一个基础类，用于构建新的神经网络模型。
让我们逐步解析代码：
-
装饰器
@MODEL_REGISTRY.register('qac'):- 这是一个装饰器，用于将
QAC类注册到一个名为MODEL_REGISTRY的注册表中。 'qac'是模型的注册名，当在深度学习框架中需要创建一个QAC实例时，可以通过这个名字来引用。- 这种注册机制通常用于插件式架构，允许用户扩展框架功能而无需修改框架本身的代码。它使得框架能够动态地识别和使用新的模型。
- 这是一个装饰器，用于将
-
类定义
class QAC(nn.Module):- 这行代码定义了一个名为
QAC的类，它继承自nn.Module。 - 在PyTorch中，
nn.Module是所有神经网络模块的基类，你的自定义模型需要继承自这个基类。
- 这行代码定义了一个名为
-
构造函数
def __init__(self, ...) -> None:__init__是Python类的构造函数，当创建类的新实例时会被调用。...表示省略了代码中的参数列表，这些参数通常用于初始化模型的层和其他组件。-> None表示这个方法不返回任何值。
-
前向传播
def forward(self, ...) -> Dict[str, torch.Tensor]:forward方法定义了模型的前向传播逻辑，即如何根据输入数据计算输出。...表示省略了输入参数，这些参数是模型的输入数据。-> Dict[str, torch.Tensor]表示该方法返回一个字典，其键是字符串类型，值是torch.Tensor类型。
-
计算 actor 输出
def compute_actor(self, obs: torch.Tensor) -> Dict[str, Union[torch.Tensor, Dict[str, torch.Tensor]]]:compute_actor方法根据观测值obs计算并返回策略网络（actor）的输出。- 输入
obs是一个torch.Tensor，代表从环境中获取的观测值。 - 返回类型是一个字典，其中包含
torch.Tensor或更复杂的字典结构，这取决于策略网络输出的具体需求。
-
计算 critic 输出
def compute_critic(self, inputs: Dict[str, torch.Tensor]) -> Dict[str, torch.Tensor]:compute_critic方法根据一组输入计算并返回价值网络（critic）的输出。- 输入
inputs是一个字典，其值是torch.Tensor类型。 - 返回一个字典，其值也是
torch.Tensor类型，这通常包含了对每个动作值的估计，用于强化学习算法中的价值估计。
-
针对图像输入, QACPixel主要需要修改的是 init中对 self.actor和 self.critic的定义。 可以看到 QAC中 self.actor和 self.critic的 encoder 都只是一层 nn.Linear
@MODEL_REGISTRY.register('qac') class QAC(nn.Module): ... def __init__(self, ...) -> None: ... self.actor = nn.Sequential( nn.Linear(obs_shape, actor_head_hidden_size), activation, ReparameterizationHead( ... ) ) ... self.critic = nn.Sequential( nn.Linear(critic_input_size, critic_head_hidden_size), activation, RegressionHead( ... ) )
对于代码片段的分析如下：
-
类定义和注册：
@MODEL_REGISTRY.register('qac')是一个装饰器，它在MODEL_REGISTRY中以 ‘qac’ 的名字注册了QAC类。这样做可以基于注册名轻松实例化模型，并支持在更大框架内模型的即插即用功能。
-
类继承：
QAC类继承自nn.Module，这是PyTorch的所有神经网络模块的基类。继承使得QAC能够使用nn.Module提供的所有功能。
-
初始化方法（
__init__方法）：- 在
__init__方法中，actor 和 critic 网络被定义为nn.Sequential的实例。 nn.Sequential容器初始化了一系列将按照添加的顺序应用于输入数据的层。
- 在
-
Actor 网络：
- 对于 actor，
nn.Linear(obs_shape, actor_head_hidden_size)定义了一个全连接的线性层，将观测输入映射到大小为actor_head_hidden_size的隐藏层。 activation指的是激活函数，通常是非线性函数，如ReLU、Tanh等。ReparameterizationHead可能是一个自定义的模块或函数，它应用了某些重参数化技术，这在变分自编码器（VAEs）等生成模型中常见。
- 对于 actor，
-
Critic 网络：
- 对于 critic，
nn.Linear(critic_input_size, critic_head_hidden_size)同样定义了一个全连接的线性层，将输入映射到隐藏层。 activation同样表示激活函数。RegressionHead可能是一个用于回归任务的自定义模块或函数，用于输出价值函数的估计。
- 对于 critic，
-
我们通过定义 encoder_cls 指定 encoder 的类型,加入 ConvEncoder,并且因为需要对 obs 进行encode 后和 action 进行拼接, 将 self.critic分为 self.critic_encoder和 self.critic_head两部分
@MODEL_REGISTRY.register('qac_pixel') class QACPixel(nn.Module): def __init__(self, ...) -> None: ... encoder_cls = ConvEncoder ... self.actor = nn.Sequential( encoder_cls(obs_shape, encoder_hidden_size_list, activation=activation, norm_type=norm_type), ReparameterizationHead( ... ) ) ... self.critic_encoder = global_encoder_cls(obs_shape, encoder_hidden_size_list, activation=activation, norm_type=norm_type) self.critic_head = RegressionHead( ... ) self.critic = nn.ModuleList([self.critic_encoder, self.critic_head])这个类是为处理像素级观测数据（例如图像）的强化学习模型而设计的。它采用了一个编码器-解码器（encoder-decoder）架构，特别是针对 actor 和 critic 的网络结构进行了调整，以适应图像输入。
以下是对代码片段的分析：
-
类定义和注册:
@MODEL_REGISTRY.register('qac_pixel')是一个装饰器，用于在MODEL_REGISTRY中注册QACPixel类。这个注册允许后续通过 ‘qac_pixel’ 名称来引用或创建QACPixel类的实例。
-
类初始化:
class QACPixel(nn.Module)定义了一个名为QACPixel的类，继承自 PyTorch 中的nn.Module。def __init__(self, ...)是类的构造函数，用于初始化类的实例。
-
Actor 网络:
encoder_cls = ConvEncoder指定使用ConvEncoder作为 actor 的编码器类。ConvEncoder类可能是一个专门为处理图像设计的卷积神经网络编码器。self.actor是一个nn.Sequential容器，包含了编码器层和一个ReparameterizationHead。编码器层通过encoder_cls创建，用于从观测数据中提取特征。ReparameterizationHead可能是负责生成动作的模块，该模块可能涉及到概率分布的参数化表示。
-
Critic 网络:
self.critic_encoder通过类似global_encoder_cls（这里似乎是一个变量名错误，应该是encoder_cls）的编码器实例化，用于处理观测数据obs。self.critic_head是RegressionHead的实例，用于价值估计。self.critic是由self.critic_encoder和self.critic_head组成的nn.ModuleList。这里将编码器和头部分开是为了方便处理，因为对观测数据进行编码后，需要将编码后的特征与动作action拼接在一起再进行处理。
-
后续步骤:
- 需要对
compute_actor和compute_critic进行修改，这两个负责计算 actor 和 critic 的输出。对于图像输入，需要考虑到新的编码器结构和可能的维度变化。
- 需要对
-
如何应用自定义模型
新 pipeline : 直接定义model,作为参数传入 policy 进行初始化,如:
... from ding.model.template.qac import QACPixel ... model = QACPixel(**cfg.policy.model) policy = SACPolicy(cfg.policy, model=model) ...旧pipeline
将定义好的 model 作为参数传入 serial_pipeline, 传入的 model 将在 serial_pipeline通过 create_policy 被调用。或者跟上述新 pipeline 一样,作为参数传入 policy 。
def serial_pipeline( input_cfg: Union[str, Tuple[dict, dict]], seed: int = 0, env_setting: Optional[List[Any]] = None, model: Optional[torch.nn.Module] = None, max_train_iter: Optional[int] = int(1e10), max_env_step: Optional[int] = int(1e10), ) -> 'Policy': ... policy = create_policy(cfg.policy, model=model, enable_field=['learn', 'collect', 'eval', 'command'])-
测试自定义 model
编写新的 model 测试，一般而言，首先需要构造 obs action等输入，传入 model ，验证输出的维度、类型的正确性。其次如果涉及神经网络，需要验证 model 是否可微。 如对于我们编写的新模型 QACPixel编写测试，首先构造维度为 (B, channel, height, width)（B = batch_size）的 obs和维度为 (B, action_shape)的 obs，传入 QACPixel的 actor和 critic得到输出. 检查输出的 q, mu, sigma的维度是否正确，以及相应的 actor和 critic model 是否可微：class TestQACPiexl: def test_qacpixel(self, action_shape, twin): inputs = {'obs': torch.randn(B, 3, 100, 100), 'action': torch.randn(B, squeeze(action_shape))} model = QACPixel( obs_shape=(3,100,100 ), action_shape=action_shape, ... ) ... q = model(inputs, mode='compute_critic')['q_value'] if twin: is_differentiable(q[0].sum(), model.critic[0]) is_differentiable(q[1].sum(), model.critic[1]) else: is_differentiable(q.sum(), model.critic_head) (mu, sigma) = model(inputs['obs'], mode='compute_actor')['logit'] assert mu.shape == (B, *action_shape) assert sigma.shape == (B, *action_shape) is_differentiable(mu.sum() + sigma.sum(), model.actor)
-
评论（0）
您还未登录，请登录后发表或查看评论