关于人工智能:使用ActorCritic的DDPG强化学习算法控制双关节机械臂

在本文中，咱们将介绍在 Reacher 环境中训练智能代理管制双关节机械臂，这是一种应用 Unity ML-Agents 工具包开发的基于 Unity 的模拟程序。咱们的指标是高精度的达到指标地位，所以这里咱们能够应用专为间断状态和动作空间设计的最先进的 Deep Deterministic Policy Gradient (DDPG) 算法。

事实世界的应用程序

机械臂在制造业、生产设施、空间摸索和搜救口头中施展着关键作用。管制机械臂的高精度和灵活性是十分重要的。通过采纳强化学习技术，能够使这些机器人零碎实时学习和调整其行为，从而进步性能和灵活性。强化学习的提高不仅有助于咱们对人工智能的了解，而且有可能彻底改变行业并对社会产生有意义的影响。

而 Reacher 是一种机械臂模拟器，罕用于控制算法的开发和测试。它提供了一个虚拟环境，模仿了机械臂的物理个性和静止法则，使得开发者能够在不须要理论硬件的状况下进行控制算法的钻研和试验。

Reacher 的环境次要由以下几个局部组成：

1. 机械臂：Reacher 模仿了一个双关节机械臂，包含一个固定基座和两个可动关节。开发者能够通过管制机械臂的两个关节来扭转机械臂的姿势和地位。
2. 指标点：在机械臂的静止范畴内，Reacher 提供了一个指标点，指标点的地位是随机生成的。开发者的工作是管制机械臂，使得机械臂的末端可能接触到指标点。
3. 物理引擎：Reacher 应用物理引擎来模仿机械臂的物理个性和静止法则。开发者能够通过调整物理引擎的参数来模仿不同的物理环境。
4. 视觉界面：Reacher 提供了一个可视化的界面，能够显示机械臂和指标点的地位，以及机械臂的姿势和静止轨迹。开发者能够通过视觉界面来调试和优化控制算法。

Reacher 模拟器是一个十分实用的工具，能够帮忙开发者在不须要理论硬件的状况下，疾速测试和优化控制算法。

模仿环境

Reacher 应用 Unity ML-Agents 工具包构建，咱们的代理能够管制双关节机械臂。指标是疏导手臂朝向指标地位并尽可能长时间地放弃其在指标区域内的地位。该环境具备 20 个同步代理，每个代理独立运行，这有助于在训练期间无效地收集教训。

状态和动作空间

理解状态和动作空间对于设计无效的强化学习算法至关重要。在 Reacher 环境中，状态空间由 33 个连续变量组成，这些变量提供无关机械臂的信息，例如其地位、旋转、速度和角速度。动作空间也是间断的，四个变量对应于施加在机械臂两个关节上的扭矩。每个动作变量都是一个介于 -1 和 1 之间的实数。

工作类型和胜利规范

Reacher 工作被认为是片段式的，每个片段都蕴含固定数量的工夫步长。代理的指标是在这些步骤中最大化其总处分。手臂末端执行器放弃在指标地位的每一步都会取得 +0.1 的处分。当代理在间断 100 次操作中的均匀得分达到 30 分或以上时，就认为胜利。

理解了环境，上面咱们将探讨 DDPG 算法、它的实现，以及它如何无效地解决这种环境中的间断管制问题。

间断管制的算法抉择：DDPG

当波及到像 Reacher 问题这样的间断管制工作时，算法的抉择对于实现最佳性能至关重要。在这个我的项目中，咱们抉择了 DDPG 算法，因为这是一种专门设计用于解决间断状态和动作空间的 actor-critic 办法。

DDPG 算法通过联合两个神经网络，联合了基于策略和基于值的办法的劣势: 行动者网络 (Actor network) 决定给定以后状态下的最佳行为，批评家网络 (Critic network) 预计状态 - 行为值函数(Q-function)。这两种网络都有指标网络，通过在更新过程中提供一个固定的指标来稳固学习过程。

通过应用 Critic 网络预计 q 函数，应用 Actor 网络确定最优行为，DDPG 算法无效地交融了策略梯度办法和 DQN 的长处。这种混合办法容许代理在间断管制环境中无效地学习。

import random
from collections import deque
import torch
import torch.nn as nn
import numpy as np

from actor_critic import Actor, Critic

class ReplayBuffer:
    def __init__(self, buffer_size, batch_size):
        self.memory = deque(maxlen=buffer_size)
        self.batch_size = batch_size

    def add(self, state, action, reward, next_state, done):
        self.memory.append((state, action, reward, next_state, done))

    def sample(self):
        batch = random.sample(self.memory, self.batch_size)
        states, actions, rewards, next_states, dones = zip(*batch)
        return states, actions, rewards, next_states, dones

    def __len__(self):
        return len(self.memory)


class DDPG:
    def __init__(self, state_dim, action_dim, hidden_dim, buffer_size, batch_size, actor_lr, critic_lr, tau, gamma):
        self.actor = Actor(state_dim, hidden_dim, action_dim, actor_lr)
        self.actor_target = Actor(state_dim, hidden_dim, action_dim, actor_lr)
        self.critic = Critic(state_dim, action_dim, hidden_dim, critic_lr)
        self.critic_target = Critic(state_dim, action_dim, hidden_dim, critic_lr)

        self.memory = ReplayBuffer(buffer_size, batch_size)
        self.batch_size = batch_size
        self.tau = tau
        self.gamma = gamma

        self._update_target_networks(tau=1)  # initialize target networks

    def act(self, state, noise=0.0):
        state = torch.tensor(state, dtype=torch.float32).unsqueeze(0)
        action = self.actor(state).detach().numpy()[0]
        return np.clip(action + noise, -1, 1)

    def store_transition(self, state, action, reward, next_state, done):
        self.memory.add(state, action, reward, next_state, done)

    def learn(self):
        if len(self.memory) < self.batch_size:
            return

        states, actions, rewards, next_states, dones = self.memory.sample()

        states = torch.tensor(states, dtype=torch.float32)
        actions = torch.tensor(actions, dtype=torch.float32)
        rewards = torch.tensor(rewards, dtype=torch.float32).unsqueeze(1)
        next_states = torch.tensor(next_states, dtype=torch.float32)
        dones = torch.tensor(dones, dtype=torch.float32).unsqueeze(1)

        # Update Critic
        self.critic.optimizer.zero_grad()

        with torch.no_grad():
            next_actions = self.actor_target(next_states)
            target_q_values = self.critic_target(next_states, next_actions)
            target_q_values = rewards + (1 - dones) * self.gamma * target_q_values

        current_q_values = self.critic(states, actions)
        critic_loss = nn.MSELoss()(current_q_values, target_q_values)

        critic_loss.backward()
        self.critic.optimizer.step()

        # Update Actor
        self.actor.optimizer.zero_grad()

        actor_loss = -self.critic(states, self.actor(states)).mean()
        actor_loss.backward()
        self.actor.optimizer.step()

        # Update target networks
        self._update_target_networks()

    def _update_target_networks(self, tau=None):
        if tau is None:
            tau = self.tau

        for target_param, param in zip(self.actor_target.parameters(), self.actor.parameters()):
            target_param.data.copy_(tau * param.data + (1 - tau) * target_param.data)

        for target_param, param in zip(self.critic_target.parameters(), self.critic.parameters()):
            target_param.data.copy_(tau * param.data + (1 - tau) * target_param.data)

下面的代码还应用了 Replay Buffer，这能够进步学习效率和稳定性。Replay Buffer 实质上是一种存储固定数量的过来教训或过渡的内存数据结构，由状态、动作、处分、下一状态和实现信息组成。应用它的次要长处是使代理可能突破间断教训之间的相关性，从而缩小无害的工夫相关性的影响。

通过从缓冲区中抽取随机的小批量教训，代理能够从一组不同的转换中学习，这有助于稳固和概括学习过程。Replay Buffer 还能够让代理屡次重用过来的教训，从而进步数据效率并促成从与环境的无限交互中更无效地学习。

DDPG 算法是一个很好的抉择，因为它可能无效地解决间断的动作空间，这是这个环境的一个要害方面。该算法的设计容许无效地利用多个代理收集的并行教训，从而实现更快的学习和更好的收敛。就像下面介绍的 Reacher 能够同时运行 20 个代理，所以咱们能够应用这 20 个代理进行分享教训，个体学习，进步学习速度。

实现了算法，上面咱们将介绍、超参数抉择和训练过程。

DDPG 算法在 Reacher 环境中工作

为了更好地了解算法在环境中的有效性，咱们须要认真钻研学习过程中波及的要害组件和步骤。

网络架构

DDPG 算法采纳两个神经网络，Actor 和 Critic。两个网络都蕴含两个暗藏层，每个暗藏层蕴含 400 个节点。暗藏层应用 ReLU (Rectified Linear Unit)激活函数，而 Actor 网络的输入层应用 tanh 激活函数产生范畴为 - 1 到 1 的动作。Critic 网络的输入层没有激活函数，因为它间接预计 q 函数。

以下是网络的代码:

import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim

class Actor(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim, learning_rate=1e-4):
        super(Actor, self).__init__()

        self.fc1 = nn.Linear(input_dim, hidden_dim)
        self.fc2 = nn.Linear(hidden_dim, hidden_dim)
        self.fc3 = nn.Linear(hidden_dim, output_dim)

        self.tanh = nn.Tanh()

        self.optimizer = optim.Adam(self.parameters(), lr=learning_rate)

    def forward(self, state):
        x = torch.relu(self.fc1(state))
        x = torch.relu(self.fc2(x))
        x = self.tanh(self.fc3(x))
        return x

class Critic(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim, learning_rate=1e-4):
        super(Critic, self).__init__()

        self.fc1 = nn.Linear(state_dim, hidden_dim)
        self.fc2 = nn.Linear(hidden_dim + action_dim, hidden_dim)
        self.fc3 = nn.Linear(hidden_dim, 1)

        self.optimizer = optim.Adam(self.parameters(), lr=learning_rate)

    def forward(self, state, action):
        x = torch.relu(self.fc1(state))
        x = torch.relu(self.fc2(torch.cat([x, action], dim=1)))
        x = self.fc3(x)
        return x

超参数抉择

抉择的超参数对于高效学习至关重要。在这个我的项目中，咱们的 Replay Buffer 大小为 200,000，批大小为 256。演员 Actor 的学习率为 5e-4，Critic 的学习率为 1e-3，soft update 参数 (tau) 为 5e-3，gamma 为 0.995。最初还退出了动作噪声，初始噪声标度为 0.5，噪声衰减率为 0.998。

训练过程

训练过程波及两个网络之间的继续交互，并且 20 个平行代理共享雷同的网络，模型会从所有代理收集的教训中个体学习。这种设置放慢了学习过程，进步了效率。

from collections import deque
import numpy as np
import torch

from ddpg import DDPG

def train_ddpg(env, agent, episodes, max_steps, num_agents, noise_scale=0.1, noise_decay=0.99):
    scores_window = deque(maxlen=100)
    scores = []

    for episode in range(1, episodes + 1):
        env_info = env.reset(train_mode=True)[brain_name]
        states = env_info.vector_observations
        agent_scores = np.zeros(num_agents)

        for step in range(max_steps):
            actions = agent.act(states, noise_scale)
            env_info = env.step(actions)[brain_name]
            next_states = env_info.vector_observations
            rewards = env_info.rewards
            dones = env_info.local_done

            for i in range(num_agents):
                agent.store_transition(states[i], actions[i], rewards[i], next_states[i], dones[i])
            agent.learn()

            states = next_states
            agent_scores += rewards
            noise_scale *= noise_decay

            if np.any(dones):
                break

        avg_score = np.mean(agent_scores)
        scores_window.append(avg_score)
        scores.append(avg_score)

        if episode % 10 == 0:
            print(f"Episode: {episode}, Score: {avg_score:.2f}, Avg Score: {np.mean(scores_window):.2f}")

        # Saving trained Networks
        torch.save(agent.actor.state_dict(), "actor_final.pth")
        torch.save(agent.critic.state_dict(), "critic_final.pth")

    return scores

if __name__ == "__main__":
    env = UnityEnvironment(file_name='Reacher_20.app')
    brain_name = env.brain_names[0]
    brain = env.brains[brain_name]

    state_dim = 33
    action_dim = brain.vector_action_space_size

    num_agents = 20
    
    # Hyperparameter suggestions
    hidden_dim = 400
    batch_size = 256
    actor_lr = 5e-4
    critic_lr = 1e-3
    tau = 5e-3
    gamma = 0.995
    noise_scale = 0.5
    noise_decay = 0.998

    agent = DDPG(state_dim, action_dim, hidden_dim=hidden_dim, buffer_size=200000, batch_size=batch_size,
                 actor_lr=actor_lr, critic_lr=critic_lr, tau=tau, gamma=gamma)

    episodes = 200
    max_steps = 1000

    scores = train_ddpg(env, agent, episodes, max_steps, num_agents, noise_scale=0.2, noise_decay=0.995)

训练过程中的关键步骤如下所示：

初始化网络：代理应用随机权重初始化共享的 Actor 和 Critic 网络及其各自的指标网络。指标网络在更新期间提供稳固的学习指标。

• 与环境交互：每个代理应用共享的 Actor 网络，通过依据其以后状态抉择动作来与环境交互。为了激励摸索，在训练的初始阶段还将噪声项增加到动作中。采取行动后，每个代理都会察看由此产生的处分和下一个状态。
• 存储教训：每个代理将察看到的教训（状态、动作、处分、next_state）存储在共享重放缓冲区中。该缓冲区蕴含固定数量的近期教训，这样每个代理可能从所有代理收集的各种转换中学习。
• 从教训中学习：定期从共享重放缓冲区中抽取一批教训。通过最小化预测 Q 值和指标 Q 值之间的均方误差，应用采样教训来更新共享 Critic 网络。
• 更新 Actor 网络：共享 Actor 网络应用策略梯度进行更新，策略梯度是通过采纳共享 Critic 网络对于所选动作的输入梯度来计算的。共享 Actor 网络学习抉择最大化预期 Q 值的动作。
• 更新指标网络：共享的 Actor 和 Critic 指标网络应用以后和指标网络权重的混合进行软更新。这确保了稳固的学习过程。

后果展现

咱们的 agent 应用 DDPG 算法胜利地学会了在 Racher 环境下管制双关节机械臂。在整个训练过程中，咱们依据所有 20 个代理的均匀得分来监控代理的体现。随着智能体摸索环境和收集教训，其预测处分最大化最佳行为的能力显著进步。

能够看到代理在工作中体现出了显著的熟练程度，均匀得分超过了解决环境所需的阈值(30+)，尽管代理的体现在整个训练过程中有所不同，但总体趋势呈上升趋势，表明学习过程是胜利的。

该图显示了 20 个代理的均匀得分:

能够看到咱们实现的 DDPG 算法，无效地解决了 Racher 环境的问题。代理可能调整本人的行为，并在工作中达到预期的性能。

下一步工作

本我的项目中的超参数是依据文献和实证测试的倡议组合抉择的。还能够通过零碎超参数调优的进一步优化可能会带来更好的性能。

多 agent 并行训练: 在这个我的项目中，咱们应用 20 个 agent 同时收集教训。应用更多代理对整个学习过程的影响可能会导致更快的收敛或进步性能。

批归一化: 为了进一步加强学习过程，在神经网络架构中实现批归一化是值得摸索的。通过在训练过程中对每一层的输出特色进行归一化，批归一化能够帮忙缩小外部协变量移位，减速学习，并潜在地进步泛化。将批处理归一化退出到 Actor 和 Critic 网络可能会导致更稳固和无效的训练，然而这个须要进一步测试。

本文残缺代码：

https://avoid.overfit.cn/post/54829204a5c74f0bb2b3a686c5fe079f

援用：

1. Lillicrap, T. P., Hunt, J. J., Pritzel, A., Heess, N., Erez, T., Tassa, Y., Silver, D., & Wierstra, D. (2015). Continuous control with deep reinforcement learning.
2. Sutton, R. S., & Barto, A. G. (2018). Reinforcement learning: An introduction. MIT press.
3. Mnih, V., Kavukcuoglu, K., Silver, D., Rusu, A. A., Veness, J., Bellemare, M. G., … & Hassabis, D. (2015). Human-level control through deep reinforcement learning. Nature, 518(7540), 529–533.
4. Udacity Deep Reinforcement Learning Nanodegree.
5. Barth-Maron, G., Hoffman, M. W., Budden, D., Dabney, W., Horgan, D., TB, D., & Lillicrap, T. (2018). Distributed Distributional Deterministic Policy Gradients. arXiv preprint arXiv:1804.08617.

作者：Gabriel Cassimiro