在本教程第 1 部分中,我们克隆了存储库,安装了它,并设置了适当的源来首次运行训练。这应该使我们能够在 ROS 仿真中对基于 TD3 架构的神经网络进行全面训练。在这一部分中,我们将详细介绍TD3网络代码本身。

神经网络代码位于 python 文件中:

train_velodyne_td3.py 

位于:

~<PATH_TO_FOLDER>/DRL-robot-navigation/TD3

此处的命名约定代表:

  • train — 用于训练的 Python 文件
  • Velodyne — 用于获取周围环境的传感器
  • TD3 — 网络架构

TD3网络实施

TD3 是一种类似于 DDPG 的演员-评论家类型的网络。这意味着有一个“演员”网络来计算一个要执行的动作,还有一个“批评者”网络,它估计这个动作有多好。简单来说,TD3架构是DDPG架构的扩展,用于解决Q值高估的问题。它通过在循环中引入第二个评论家网络并从产生较低 Q 值估计的输出中选择输出来实现这一点。(同样,可以在此处获得数学和算法背景概述。因此,我们需要创建一个参与者网络,它将环境状态作为机器人要采取的输入和输出行动。此外,我们需要创建两个批评者网络,它们将环境状态以及来自参与者网络的动作作为输入,并输出此状态-动作对的估计值。

演员网络

要创建一个 actor 模型,我们只需要创建一个解码器,它将状态输入映射到操作,如下所示:

class Actor(nn.Module):
    def __init__(self, state_dim, action_dim):
        super(Actor, self).__init__()

        self.layer_1 = nn.Linear(state_dim, 800)
        self.layer_2 = nn.Linear(800, 600)
        self.layer_3 = nn.Linear(600, action_dim)
        self.tanh = nn.Tanh()

    def forward(self, s):
        s = F.relu(self.layer_1(s))
        s = F.relu(self.layer_2(s))
        a = self.tanh(self.layer_3(s))
        return a

我们创建一个 actor 类,它由 3 个线性、全连接的层组成。在类初始化中,我们定义了将在前向传递中使用的层。第一层的输入将是一个单维向量,将环境状态表示为state_dim。我们将用 800 个参数嵌入此状态。由于这些层将按顺序调用,因此我们必须确保上一层中的参数数量与下一层中的参数数相同。因此,下一层输入需要相同数量的 800 个参数。然后,我们将它们映射到 600 个参数。在最后一层中,我们将 600 个参数直接映射到我们的机器人action_dim可控动作的数量。在我们的例子中,我们将控制一个地面移动机器人,该机器人将具有可控的线性和角速度。由于我们的机器人的最大和最小可能速度有限,因此神经网络的输出需要限制在最小-最大范围内。我们可以通过定义 Tanh 激活函数来实现这一点。Tanh 函数将输出限制在 -1、1 范围内。

在前向传递中,我们定义序列、如何调用层以及如何通过它传播值。首先,我们调用第一个定义的层,对其应用 ReLU 激活,然后对第二个层执行相同的操作。之后,我们调用第三个线性层并应用 Tanh 函数强制输出在 -1、1 范围内。这个简单的参与者网络的输出将成为我们机器人的线速度和角速度的基础。

评论家网络

同样,我们将创建一个评论家网络。如前所述,我们需要 2 个批评家来减轻 Q 值高估。幸运的是,这两个批评家总是同时工作,所以他们可以被一个批评家阶层来定义。

class Critic(nn.Module):
    def __init__(self, state_dim, action_dim):
        super(Critic, self).__init__()

        self.layer_1 = nn.Linear(state_dim, 800)
        self.layer_2_s = nn.Linear(800, 600)
        self.layer_2_a = nn.Linear(action_dim, 600)
        self.layer_3 = nn.Linear(600, 1)

        self.layer_4 = nn.Linear(state_dim, 800)
        self.layer_5_s = nn.Linear(800, 600)
        self.layer_5_a = nn.Linear(action_dim, 600)
        self.layer_6 = nn.Linear(600, 1)

    def forward(self, s, a):
        s1 = F.relu(self.layer_1(s))
        self.layer_2_s(s1)
        self.layer_2_a(a)
        s11 = torch.mm(s1, self.layer_2_s.weight.data.t())
        s12 = torch.mm(a, self.layer_2_a.weight.data.t())
        s1 = F.relu(s11 + s12 + self.layer_2_a.bias.data)
        q1 = self.layer_3(s1)

        s2 = F.relu(self.layer_4(s))
        self.layer_5_s(s2)
        self.layer_5_a(a)
        s21 = torch.mm(s2, self.layer_5_s.weight.data.t())
        s22 = torch.mm(a, self.layer_5_a.weight.data.t())
        s2 = F.relu(s21 + s22 + self.layer_5_a.bias.data)
        q2 = self.layer_6(s2)
        return q1, q2

在这里,Q 值的解码器与我们在 actor 类中看到的解码器非常相似。然而,由于我们需要评估的不是状态,而是行动者对状态的反应,因此我们不仅需要状态信息,还需要行动者网络的行动作为我们批评网络的输入。有多种方法可以将状态操作参数定义为输入。例如,最简单的形式是简单地连接(或只是附加)表示状态的一维向量末尾的动作值。然后,我们可以使用在actor网络中看到的相同形式的解码器。但是,我们将使用受此处描述的方法启发的略有不同的方法。从本质上讲,我们将状态和行动视为批评家网络的不同输入。用单线性嵌入状态信息,分别传递第二层的状态嵌入值和动作值,将它们组合在一起,得到最后一层的值估计。由于我们将使用单个类创建两个批评家网络,因此我们需要在初始化中定义这些层两次。第 1、2、3 层属于第一个批评家,第 4、5、6 层属于第二个批评家。请注意,2_a 层和 4_a 层使用输入大小作为action_dim,因为这将是嵌入操作的第一个图层。

在前向传递中,我们定义传播将如何发生。首先,只有状态输入通过第一层,并对其应用 ReLU 激活。然后,它通过第二层。同时,操作值也通过具有相同输出大小的单个层传递。状态嵌入和操作乘以相应第二层的权重。每次乘法将输出具有 600 个参数的相同大小的张量。然后可以对这些张量求和,并且还可以将第二个动作层的偏差添加到它们中。(这种将动作输入注入神经网络的方法在实践中效果很好)之后,我们使用最后一层将这个组合状态-动作对嵌入映射到 Q 值估计中。请注意,这里没有激活函数,因为 Q 值没有最小值或最大值,因此不需要上限。作为输出,我们将得到两个评论家网络的两个 Q 值。

TD3网络

最后,我们可以实现完整的TD3网络,这将是我们的演员和评论家网络的结合。我们可以定义一个 TD3 类,并在其中创建我们的演员和评论家。


class TD3(object):
    def __init__(self, state_dim, action_dim, max_action):
        # Initialize the Actor network
        self.actor = Actor(state_dim, action_dim).to(device)
        self.actor_target = Actor(state_dim, action_dim).to(device)
        self.actor_target.load_state_dict(self.actor.state_dict())
        self.actor_optimizer = torch.optim.Adam(self.actor.parameters())

        # Initialize the Critic networks
        self.critic = Critic(state_dim, action_dim).to(device)
        self.critic_target = Critic(state_dim, action_dim).to(device)
        self.critic_target.load_state_dict(self.critic.state_dict())
        self.critic_optimizer = torch.optim.Adam(self.critic.parameters())

        self.max_action = max_action
        self.writer = SummaryWriter()
        self.iter_count = 0

在TD3网络中,我们将实现一种称为软更新的东西。每次我们优化网络参数时,我们只会选择一个数据子集,称为批处理。这意味着每个网络的优化参数将针对这些样本进行高度调整。然而,这并不意味着我们将找到所有可能情况的最佳参数。事实上,如果我们直接使用这些优化的参数,我们的学习将非常不稳定,因为每次网络参数都会针对不同的集合进行优化。这就是为什么我们将为每个网络设置一个控制或目标网络的原因。在每个训练周期中,我们将优化我们的基础网络并获得新的参数。然后,我们将从基础网络中注入少量这些优化参数,并重置基础网络,使其具有与新更新的目标网络相同的参数。这样,在每次更新时,目标网络都会在(希望)最佳策略的总体方向上略微移动,同时保持训练过程的稳定。这就是为什么在 TD3 类的初始化中不仅需要调用我们的 actor 和 critic 类,还需要创建actor_targetcritic_target调用。之后,我们加载可能已经为每个类保存的任何参数,并为每个类设置 Adam 优化器。接下来,我们设置最大可能的操作值,创建一个编写器来记录我们的数据以进行 Tensorboard 可视化,并在 iter_count 中设置一个训练迭代计数器。

初始化后,我们可以在 TD3 类中定义一些额外的帮助程序函数。


def get_action(self, state):
        # Function to get the action from the actor
        state = torch.Tensor(state.reshape(1, -1)).to(device)
        return self.actor(state).cpu().data.numpy().flatten()

def save(self, filename, directory):
        torch.save(self.actor.state_dict(), "%s/%s_actor.pth" % (directory, filename))
        torch.save(self.critic.state_dict(), "%s/%s_critic.pth" % (directory, filename))

def load(self, filename, directory):
    self.actor.load_state_dict(
        torch.load("%s/%s_actor.pth" % (directory, filename))
    )
    self.critic.load_state_dict(
        torch.load("%s/%s_critic.pth" % (directory, filename))
    )

给定环境状态,get_action函数将获得单个操作。对于神经网络中的处理,状态张量将被放在设备(cpu 或 cuda)上,但为了稍后在 ROS 模拟器中使用,我们要求它专门放在 cpu 上,变成一个 NumPy 数组,并作为单个维度返回。

调用 save 和 load 函数只是保存或加载 actor 和 critic 网络参数。

也许,我们必须在类中定义的最重要的部分是训练函数。这将是我们 TD3 实现的关键,并将指定如何执行机器人运动的训练。首先,让我们定义函数并命名输入变量。

def train(
        self,
        replay_buffer,
        iterations,
        batch_size=100,
        discount=0.99,
        tau=0.005,
        policy_noise=0.2,
        noise_clip=0.5,
        policy_freq=2,
    ):
        av_Q = 0
        max_Q = -inf
        av_loss = 0

replay_buffer将定义从哪里获取用于训练的样本,迭代将定义在此函数调用中运行多少次训练,batch_size定义每次迭代中要采样的样本数量,折扣是贝尔曼方程中的折扣因子(从理论上讲),tau 是我们将在软更新中向目标网络添加多少基本网络参数。policy_noise 是要添加到操作中以启用探索的噪声的标准偏差,noise_clip是用于裁剪噪声的最小值和最大值,policy_freq是更新 actor 网络参数的频率。此外,定义变量以存储平均和最大 Q 值以及平均损失,以便在 Tensorbord 中显示。

接下来,我们通过运行迭次数的网络优化来开始训练过程。

for it in range(iterations):
            # sample a batch from the replay buffer
            (
                batch_states,
                batch_actions,
                batch_rewards,
                batch_dones,
                batch_next_states,
            ) = replay_buffer.sample_batch(batch_size)
            state = torch.Tensor(batch_states).to(device)
            next_state = torch.Tensor(batch_next_states).to(device)
            action = torch.Tensor(batch_actions).to(device)
            reward = torch.Tensor(batch_rewards).to(device)
            done = torch.Tensor(batch_dones).to(device)

            # Obtain the estimated action from the next state by using the actor-target
            next_action = self.actor_target(next_state)

            # Add noise to the action
            noise = torch.Tensor(batch_actions).data.normal_(0, policy_noise).to(device)
            noise = noise.clamp(-noise_clip, noise_clip)
            next_action = (next_action + noise).clamp(-self.max_action, self.max_action)

我们从存储在回放缓冲区中的记录的机器人运动体验中抽取一批样本。然后,我们计算批处理中每个状态的下一个可能操作,并向其添加一些噪声。


            # Calculate the Q values from the critic-target network for the next state-action pair
            target_Q1, target_Q2 = self.critic_target(next_state, next_action)

            # Select the minimal Q value from the 2 calculated values
            target_Q = torch.min(target_Q1, target_Q2)
            av_Q += torch.mean(target_Q)
            max_Q = max(max_Q, torch.max(target_Q))
            # Calculate the final Q value from the target network parameters by using Bellman equation
            target_Q = reward + ((1 - done) * discount * target_Q).detach()

            # Get the Q values of the basis networks with the current parameters
            current_Q1, current_Q2 = self.critic(state, action)

            # Calculate the loss between the current Q value and the target Q value
            loss = F.mse_loss(current_Q1, target_Q) + F.mse_loss(current_Q2, target_Q)

            # Perform the gradient descent
            self.critic_optimizer.zero_grad()
            loss.backward()
            self.critic_optimizer.step()

然后,使用目标网络估计两个评论家网络的下一个状态-动作对的可能 Q 值,并为批处理中的每个样本选择两个输出的最小值。从最小值出发,我们使用 Bellman 方程获得目标 Q 值。随后,我们得到了基本批评家网络的Q值。通过计算基网标注值和目标 Q 值之间的均方误差,我们可以估计基网的偏差。这里的诀窍是,使用基础网络,我们估算当前状态奖励 + 折扣所有未来状态奖励。但目标 Q 值使用已知的当前状态奖励,并使用对未来奖励的估计。因此,目标值应该更接近实际估计值,并且根据它计算损失应该推动网络朝着正确的优化方向发展。然后简单地从损失中的梯度优化评论家网络。

同时,我们还收集了av_Qmax_Q的平均值和最大Q值。

            if it % policy_freq == 0:
                # Maximize the actor output value by performing gradient descent on negative Q values
                # (essentially perform gradient ascent)
                actor_grad, _ = self.critic(state, self.actor(state))
                actor_grad = -actor_grad.mean()
                self.actor_optimizer.zero_grad()
                actor_grad.backward()
                self.actor_optimizer.step()

                # Use soft update to update the actor-target network parameters by
                # infusing small amount of current parameters
                for param, target_param in zip(
                    self.actor.parameters(), self.actor_target.parameters()
                ):
                    target_param.data.copy_(
                        tau * param.data + (1 - tau) * target_param.data
                    )
                # Use soft update to update the critic-target network parameters by infusing
                # small amount of current parameters
                for param, target_param in zip(
                    self.critic.parameters(), self.critic_target.parameters()
                ):
                    target_param.data.copy_(
                        tau * param.data + (1 - tau) * target_param.data
                    )

            av_loss += loss

Twin Delayed DDPG (TD3) 中的延迟部分接下来发挥作用。从本质上讲,如果与评论家网络相比,我们会延迟 actor 网络的更新,并且通过每policy_freq迭代一次执行此步骤来延迟网络参数的软更新。这应该可以提高网络稳定性,并提供更好的参与者参数更新(因为批评者会更精确)。在 actor 更新期间,我们计算 state-action 对的 Q 值。我们优化网络参数的方向,这将为我们提供尽可能高的 Q 值。这可能看起来很奇怪,起初我们使用固定的 actor 参数来获取 actor 网络丢失,然后使用相同的 critic 来获取 actor 更新的 Q 值。虽然这种方法可能不稳定,但每个网络的策略不断来回“拉动”最终会导致正确的方向,我们获得了(希望)最优的参与者和批评者网络。

然后,我们只需使用软更新稍微更新两个网络参数即可。

最后,只需记录我们的 Q 和损失值变量,以便在 Tensorboard 中显示。

这应该创建一个可用于学习机器人运动的 TD3 神经网络。在下一部分(第 3 部分)中,我们将了解如何在移动机器人设置中实际调用此网络。

Logo

北京人形旗下天工造物具身智能开源社区,聚焦具身天工与慧思开物两大平台

更多推荐