0
0

深度强化学习-TD3算法原理与代码

引言

Twin Delayed Deep Deterministic policy gradient (TD3)是由Scott Fujimoto等人在Deep Deterministic Policy Gradient (DDPG)算法上改进得到的一种用于解决连续控制问题的在线(on-line)异策(off-policy)式深度强化学习算法。本质上，TD3算法就是将Double Q-Learning算法的思想融入到DDPG算法中。前面我们已经分别介绍过DDPG算法和Double DQN算法的原理并进行了代码实现，有兴趣的小伙伴可以先去看一下，之后再来看本文应该就能很容易理解。本文就带领大家了解一下TD3算法的具体原理，并采用Pytorch进行实现，论文和代码的链接见下方。

论文：http://proceedings.mlr.press/v80/fujimoto18a/fujimoto18a.pdf

代码：https://github.com/indigoLovee/TD3

1 TD3算法简介

之前我们在讲Double DQN算法时就曾分析过Deep Q-Learning (DQN)算法存在高估问题，而DDPG算法是从DQN算法进化得到，因此它也存在一样的问题。为此，TD3算法就很自然地被提出，主要解决DDPG算法的高估问题。

TD3算法也是Actor-Critic (AC)框架下的一种确定性深度强化学习算法，它结合了深度确定性策略梯度算法和双重Q学习，在许多连续控制任务上都取得了不错的表现。

2 TD3算法原理

TD3算法在DDPG算法的基础上，提出了三个关键技术：

（1）双重网络 (Double network)：采用两套Critic网络，计算目标值时取二者中的较小值，从而抑制网络过估计问题。

（2）目标策略平滑正则化 (Target policy smoothing regularization)：计算目标值时，在下一个状态的动作上加入扰动，从而使得价值评估更准确。

（3）延迟更新 (Delayed update)：Critic网络更新多次后，再更新Actor网络，从而保证Actor网络的训练更加稳定。

2.1 双重网络

TD3算法中包括六个网络，分别是Actor网络\mu \left ( \cdot \mid \theta^{\mu} \right )，Critic1网络Q_{1}\left ( \cdot \mid \theta^{Q_{1}} \right )，Critic2网络Q_{2}\left ( \cdot \mid \theta^{Q_{2}} \right )，Target Actor网络\mu^{'} \left ( \cdot \mid \theta^{\mu^{'}} \right )，Target Critic1网络Q^{'}_{1}\left ( \cdot \mid \theta^{Q^{'}_{1}} \right )​​​​​​​，Target Critic2网络Q^{'}_{2}\left ( \cdot \mid \theta^{Q^{'}_{2}} \right )。相较于DDPG算法，TD3算法多了一套Critic网络，这就是双重网络的由来。本节首先分析网络过估计的成因，然后引入双重网络，最后介绍算法的更新过程。

2.1.1 网络过估计的成因

DQN算法的高估主要来源于两个方面：自举 (Bootstrapping)和最大化，DDPG算法也是如此。之前我们在讲DDPG算法时曾强调，如果高估是均匀的，对于智能体最终的决策不会带来影响；如果是非均匀的，对于智能体最终的决策会带来显著影响。然而实际上网络的高估通常是非均匀的，这里简单分析一下原因。

在更新Critic网络时，假设从经验池中采样的数据为\left ( s, a, r, s^{'}, done \right )。首先我们会计算目标y

y=r+\gamma Q^{'}\left ( s^{'},a^{'} \mid \theta ^{Q^{'}} \right )

由于网络高估，因此

Q^{'}\left ( s^{'},a^{'} \mid \theta ^{Q^{'}} \right )\geqslant Q^{\ast }\left ( s^{'},a^{'} \right )

其中， Q^{\ast }\left ( s^{'},a^{'} \right )表示状态动作对\left ( s^{'},a^{'} \right )真实的最优状态动作价值。

接着我们会让Q\left ( s, a \mid \theta ^{Q} \right )逼近y，从而使得Q\left ( s, a \mid \theta ^{Q} \right )出现过估计，即

Q\left ( s,a \mid \theta ^{Q} \right )\geqslant Q^{\ast }\left ( s,a\right )

其中， Q^{\ast }\left ( s,a\right )表示状态动作对\left ( s,a \right )真实的最优状态动作价值。

每次采样状态动作对\left ( s,a \right )来对Critic网络进行更新时，就会让网络高估\left ( s,a \right )的状态动作价值，而\left ( s,a \right )在经验池中的频率显然是不均匀的。如果出现的频率越高，那么高估就越严重。因此，网络的高估是非均匀的，而非均匀的高估对智能体的决策有害，因此我们需要避免网络高估。

2.1.2 双重网络的引入

DDPG算法采用目标网络解决了自举问题，有兴趣的小伙伴可以看一下我的那篇博文，里面详细分析了自举问题带来的危害，以及目标网络是如何解决自举问题的，这里就不再赘述了。但是，除了自举以外，最大化也是造成过估计的重要原因，因此要想彻底解决网络过估计，还需要解决最大化问题。这里简单分析一下为什么最大化会造成网络过估计。

假设x_{1},x_{2},\cdots ,x_{n}为观测到的真实值，在其中加入均值为0的随机噪声，得到Q_{1},Q_{2},\cdots,Q_{n}。由于噪声的均值为0，因此满足

E\left [ mean_{i}\left ( Q_{i} \right ) \right ]=mean_{i}\left ( x_{i} \right )

但是随机噪声会让最大值变大，即

E\left [ max_{i}\left ( Q_{i} \right ) \right ]\geqslant max_{i}\left ( x_{i} \right )

同理， 随机噪声会让最小值变小，即

E\left [ min_{i}\left ( Q_{i} \right ) \right ]\leqslant min_{i}\left ( x_{i} \right )

这三个公式都可以被证明出来，这里就不给大家证明了。

回到DQN算法的高估问题，假设每个状态动作对的真实状态动作价值为x\left ( s,a_{1} \right ),\cdots,x\left ( s,a_{n} \right )。Q网络的估计会存在一定噪声，不妨假设是无偏估计，那么估计出的状态动作价值为Q\left ( s,a_{1} \mid \theta^{Q} \right ), \cdots, Q\left ( s,a_{n} \mid \theta^{Q} \right )。由于噪声的均值为0，因此满足

mean_{a}\left ( x\left ( s,a \right ) \right )=mean_{a}\left ( Q\left ( s,a\mid \theta^{Q} \right ) \right )

q=max_{a}Q\left ( s,a \mid \theta^{Q} \right )是一种典型的高估，即

q\geqslant max_{a}\left ( x\left ( s,a \right ) \right )

我们在计算目标值y时，会执行q_{t+1}=max_{a}Q\left ( s_{t+1}, a \mid \theta^{Q} \right )，由于q_{t+1}高估，因此目标值

y=r+\gamma q_{t+1} 也会高估。网络更新时我们会将Q\left ( s,a\mid \theta^{Q} \right )逼近y，由于y高估，因此Q\left ( s,a\mid \theta^{Q} \right )就会出现高估。

总结起来就是，最大化操作会使得网络的估计值大于真实值，从而造成网络过估计。

双重网络是解决最大化问题的有效方法。在TD3算法中，作者引入了两套相同网络架构的Critic网络。计算目标值时，会利用二者间的较小值来估计下一个状态动作对\left ( s^{'},a^{'} \right )的状态动作价值，即

y=r+\gamma min_{i=1,2}Q_{i}^{'}\left ( s^{'},a^{'} \mid \theta_{i}^{Q^{'}} \right )

从而可以有效避免最大化问题带来的高估。这时可能会有小伙伴比较疑惑，取两个网络之间的较小值会不会不太稳妥？如果用多个网络，然后取它们中的最小值会不会更好呢？其实有实验证明采用两个网络就可以了，多个网络不会带来明显的性能提升。

2.2 目标策略平滑正则化

确定性策略存在一个问题：它会过度拟合以缩小价值估计中的峰值。当更新Critic网络时，使用确定性策略的学习目标极易受到函数逼近误差的影响，从而导致目标估计的方差大，估计值不准确。这种诱导方差可以通过正则化来减少，因此作者模仿SARSA的学习更新，引入了一种深度价值学习的正则化策略——目标策略平滑。

这种方法主要强调：类似的行动应该具有类似的价值。虽然函数近似隐式地实现了这一点，但可以通过修改训练过程显示地强调类似动作之间的关系。具体的实现是利用目标动作周围的区域来计算目标值，从而有利于平滑估计值

y=r+E\left [ Q^{'}\left ( s^{'}, \mu^{'}\left ( s^{'} \mid \theta^{'} \right ) +\epsilon \mid \theta^{Q^{'}} \right ) \right ]

在实际操作时，我们可以通过向目标动作中添加少量随机噪声，并在小批量中求平均值，来近似动作的期望。因此，上式可以修改为

y=r+\gamma Q^{'}\left ( s^{'}, \mu^{'}\left ( s^{'} \mid \theta^{'} \right ) + \epsilon \mid \theta^{Q^{'}} \right )

\epsilon \sim clip\left ( N\left ( 0,\sigma \right ),-c,c \right )

其中，我们添加的噪声是服从正态分布的，并且对采样的噪声做了裁剪，以保持目标接近原始动作。直观的说，采用这种方法得出的策略往往更加安全，因为它们为抵抗干扰的动作提供了更高的价值。说了这么多可能不是特别容易理解，不妨来看两张图。

假设上图为Critic网络估计的Q值曲面。这里我们直接采用Q\left ( s^{'},a^{'} \right )来估计Q\left ( s,a \right )，因此方差会很大，不利于网络训练。

这次我们采用状态动作对\left ( s^{'},a^{'} \right )的邻域来估计Q\left ( s,a \right )，从而可以极大地降低方差，提高目标值估计的准确性，保证网络训练过程的稳定。

2.3 延迟更新

这里的延迟更新指的是Actor网络的延迟更新，即Critic网络更新多次之后再对Actor网络进行更新。这个想法其实是非常直观的，因为Actor网络是通过最大化累积期望回报来更新的，它需要利用Critic网络来进行评估。如果Critic网络非常不稳定，那么Actor网络自然也会出现震荡。

因此，我们可以让Critic网络的更新频率高于Actor网络，即等待Critic网络更加稳定之后再来帮助Actor网络更新。

3 TD3算法更新过程

TD3算法的更新过程与DDPG算法的更新过程差别不大，主要区别在于目标值的计算方式（2.1.2节已经给出）。其中Actor网络通过最大化累积期望回报来更新（确定性策略梯度），Critic1和Critic2网络都是通过最小化评估值与目标值之间的误差来更新（MSE），所有的目标网络都采用软更新的方式来更新（Exponential Moving Average, EMA）。在训练阶段，我们从Replay Buffer中采样一个批次 (Batch size) 的数据，假设采样到的一条数据为\left ( s,a,r,s^{'}, done \right )，所有网络的更新过程如下。

Critic1和Critic2网络更新过程：利用Target Actor网络计算出状态s^{'}下的动作

a^{'}=\mu ^{'}\left ( s^{'} \mid \theta^{\mu^{'}} \right )

然后基于目标策略平滑正则化，再目标动作a^{'}上加入噪声

a^{'}=a^{'}+\epsilon

\epsilon \sim clip\left ( N\left ( 0,\sigma \right ),-c,c \right )

接着基于双重网络的思想，计算目标值

y=r+\gamma min_{i=1,2}Q_{i}^{'}\left ( s^{'},a^{'} \mid \theta_{i}^{Q^{'}} \right )

最后利用梯度下降算法最小化评估值和目标值之间的误差L_{c_{i}}，从而对Critic1和Critic2网络中的参数进行更新

L_{c_{i}}=\left ( Q_{i}\left ( s,a \mid \theta^{Q_{i}} \right )-y \right )^{2} \left ( i=1,2 \right )

Actor网络更新过程：(在Ctitic1和Critic2网络更新d 步之后，启动Actor网络更新) 利用Actor网络计算出状态s下的动作

a_{new}=\mu \left ( s \mid \theta^{\mu} \right )

这里需要注意：计算出动作后不需要加入噪声，因为这里是希望Actor网络能够朝着最大值方向更新，加入噪声没有任何意义。然后利用Critic1或者Critic2网络来计算状态动作对\left ( s,a_{new} \right )的评估值，这里我们假定使用Critic1网络

q_{new}=Q_{1}\left ( s,a_{new} \mid \theta^{Q_{1}} \right )

最后采用梯度上升算法最大化q_{new}，从而完成对Actor网络的更新。

注：这里我们之所以可以使用Critic1和Critic2两者中的任何一个来计算Q值，我觉得主要是因为Actor网络的目的就在于最大化累积期望回报，没有必要使用最小值。

目标网络的更新过程：采用软更新方式对目标网络进行更新。引入一个学习率（或者成为动量）\tau，将旧的目标网络参数和新的对应网络参数做加权平均，然后赋值给目标网络

\theta^{Q_{i}^{'}}=\tau \theta^{Q_{i}}+\left ( 1- \tau \right )\theta^{Q_{i}^{'}}(i=1,2)

\theta^{\mu^{'}}=\tau \theta^{\mu}+\left ( 1-\tau \right ) \theta^{\mu^{'}}

学习率（动量）\tau \in \left ( 0,1 \right )，通常取值0.005。

4 TD3算法伪代码

5 PyTorch代码实现

Replay Buffer的代码实现（buffer.py）:

import numpy as np

class ReplayBuffer:
    def __init__(self, max_size, state_dim, action_dim, batch_size):
        self.mem_size = max_size
        self.batch_size = batch_size
        self.mem_cnt = 0

        self.state_memory = np.zeros((max_size, state_dim))
        self.action_memory = np.zeros((max_size, action_dim))
        self.reward_memory = np.zeros((max_size, ))
        self.next_state_memory = np.zeros((max_size, state_dim))
        self.terminal_memory = np.zeros((max_size, ), dtype=np.bool)

    def store_transition(self, state, action, reward, state_, done):
        mem_idx = self.mem_cnt % self.mem_size

        self.state_memory[mem_idx] = state
        self.action_memory[mem_idx] = action
        self.reward_memory[mem_idx] = reward
        self.next_state_memory[mem_idx] = state_
        self.terminal_memory[mem_idx] = done

        self.mem_cnt += 1

    def sample_buffer(self):
        mem_len = min(self.mem_cnt, self.mem_size)
        batch = np.random.choice(mem_len, self.batch_size, replace=False)

        states = self.state_memory[batch]
        actions = self.action_memory[batch]
        rewards = self.reward_memory[batch]
        states_ = self.next_state_memory[batch]
        terminals = self.terminal_memory[batch]

        return states, actions, rewards, states_, terminals

    def ready(self):
        return self.mem_cnt >= self.batch_size

Actor和Critic网络的代码实现（networks.py）:

import torch as T
import torch.nn as nn
import torch.optim as optim

device = T.device("cuda:0" if T.cuda.is_available() else "cpu")

class ActorNetwork(nn.Module):
    def __init__(self, alpha, state_dim, action_dim, fc1_dim, fc2_dim):
        super(ActorNetwork, self).__init__()
        self.fc1 = nn.Linear(state_dim, fc1_dim)
        self.ln1 = nn.LayerNorm(fc1_dim)
        self.fc2 = nn.Linear(fc1_dim, fc2_dim)
        self.ln2 = nn.LayerNorm(fc2_dim)
        self.action = nn.Linear(fc2_dim, action_dim)

        self.optimizer = optim.Adam(self.parameters(), lr=alpha)
        self.to(device)

    def forward(self, state):
        x = T.relu(self.ln1(self.fc1(state)))
        x = T.relu(self.ln2(self.fc2(x)))
        action = T.tanh(self.action(x))

        return action

    def save_checkpoint(self, checkpoint_file):
        T.save(self.state_dict(), checkpoint_file, _use_new_zipfile_serialization=False)

    def load_checkpoint(self, checkpoint_file):
        self.load_state_dict(T.load(checkpoint_file))

class CriticNetwork(nn.Module):
    def __init__(self, beta, state_dim, action_dim, fc1_dim, fc2_dim):
        super(CriticNetwork, self).__init__()
        self.fc1 = nn.Linear(state_dim+action_dim, fc1_dim)
        self.ln1 = nn.LayerNorm(fc1_dim)
        self.fc2 = nn.Linear(fc1_dim, fc2_dim)
        self.ln2 = nn.LayerNorm(fc2_dim)
        self.q = nn.Linear(fc2_dim, 1)

        self.optimizer = optim.Adam(self.parameters(), lr=beta)
        self.to(device)

    def forward(self, state, action):
        x = T.cat([state, action], dim=-1)
        x = T.relu(self.ln1(self.fc1(x)))
        x = T.relu(self.ln2(self.fc2(x)))
        q = self.q(x)

        return q

    def save_checkpoint(self, checkpoint_file):
        T.save(self.state_dict(), checkpoint_file, _use_new_zipfile_serialization=False)

    def load_checkpoint(self, checkpoint_file):
        self.load_state_dict(T.load(checkpoint_file))

TD3算法的代码实现（TD3.py）:

import torch as T
import torch.nn.functional as F
import numpy as np
from networks import ActorNetwork, CriticNetwork
from buffer import ReplayBuffer

device = T.device("cuda:0" if T.cuda.is_available() else "cpu")

class TD3:
    def __init__(self, alpha, beta, state_dim, action_dim, actor_fc1_dim, actor_fc2_dim,
                 critic_fc1_dim, critic_fc2_dim, ckpt_dir, gamma=0.99, tau=0.005, action_noise=0.1,
                 policy_noise=0.2, policy_noise_clip=0.5, delay_time=2, max_size=1000000,
                 batch_size=256):
        self.gamma = gamma
        self.tau = tau
        self.action_noise = action_noise
        self.policy_noise = policy_noise
        self.policy_noise_clip = policy_noise_clip
        self.delay_time = delay_time
        self.update_time = 0
        self.checkpoint_dir = ckpt_dir

        self.actor = ActorNetwork(alpha=alpha, state_dim=state_dim, action_dim=action_dim,
                                  fc1_dim=actor_fc1_dim, fc2_dim=actor_fc2_dim)
        self.critic1 = CriticNetwork(beta=beta, state_dim=state_dim, action_dim=action_dim,
                                     fc1_dim=critic_fc1_dim, fc2_dim=critic_fc2_dim)
        self.critic2 = CriticNetwork(beta=beta, state_dim=state_dim, action_dim=action_dim,
                                     fc1_dim=critic_fc1_dim, fc2_dim=critic_fc2_dim)

        self.target_actor = ActorNetwork(alpha=alpha, state_dim=state_dim, action_dim=action_dim,
                                         fc1_dim=actor_fc1_dim, fc2_dim=actor_fc2_dim)
        self.target_critic1 = CriticNetwork(beta=beta, state_dim=state_dim, action_dim=action_dim,
                                            fc1_dim=critic_fc1_dim, fc2_dim=critic_fc2_dim)
        self.target_critic2 = CriticNetwork(beta=beta, state_dim=state_dim, action_dim=action_dim,
                                            fc1_dim=critic_fc1_dim, fc2_dim=critic_fc2_dim)

        self.memory = ReplayBuffer(max_size=max_size, state_dim=state_dim, action_dim=action_dim,
                                   batch_size=batch_size)

        self.update_network_parameters(tau=1.0)

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

        for actor_params, target_actor_params in zip(self.actor.parameters(),
                                                     self.target_actor.parameters()):
            target_actor_params.data.copy_(tau * actor_params + (1 - tau) * target_actor_params)

        for critic1_params, target_critic1_params in zip(self.critic1.parameters(),
                                                         self.target_critic1.parameters()):
            target_critic1_params.data.copy_(tau * critic1_params + (1 - tau) * target_critic1_params)

        for critic2_params, target_critic2_params in zip(self.critic2.parameters(),
                                                         self.target_critic2.parameters()):
            target_critic2_params.data.copy_(tau * critic2_params + (1 - tau) * target_critic2_params)

    def remember(self, state, action, reward, state_, done):
        self.memory.store_transition(state, action, reward, state_, done)

    def choose_action(self, observation, train=True):
        self.actor.eval()
        state = T.tensor([observation], dtype=T.float).to(device)
        action = self.actor.forward(state)

        if train:
            # exploration noise
            noise = T.tensor(np.random.normal(loc=0.0, scale=self.action_noise),
                             dtype=T.float).to(device)
            action = T.clamp(action+noise, -1, 1)
        self.actor.train()

        return action.squeeze().detach().cpu().numpy()

    def learn(self):
        if not self.memory.ready():
            return

        states, actions, rewards, states_, terminals = self.memory.sample_buffer()
        states_tensor = T.tensor(states, dtype=T.float).to(device)
        actions_tensor = T.tensor(actions, dtype=T.float).to(device)
        rewards_tensor = T.tensor(rewards, dtype=T.float).to(device)
        next_states_tensor = T.tensor(states_, dtype=T.float).to(device)
        terminals_tensor = T.tensor(terminals).to(device)

        with T.no_grad():
            next_actions_tensor = self.target_actor.forward(next_states_tensor)
            action_noise = T.tensor(np.random.normal(loc=0.0, scale=self.policy_noise),
                                    dtype=T.float).to(device)
            # smooth noise
            action_noise = T.clamp(action_noise, -self.policy_noise_clip, self.policy_noise_clip)
            next_actions_tensor = T.clamp(next_actions_tensor+action_noise, -1, 1)
            q1_ = self.target_critic1.forward(next_states_tensor, next_actions_tensor).view(-1)
            q2_ = self.target_critic2.forward(next_states_tensor, next_actions_tensor).view(-1)
            q1_[terminals_tensor] = 0.0
            q2_[terminals_tensor] = 0.0
            critic_val = T.min(q1_, q2_)
            target = rewards_tensor + self.gamma * critic_val
        q1 = self.critic1.forward(states_tensor, actions_tensor).view(-1)
        q2 = self.critic2.forward(states_tensor, actions_tensor).view(-1)

        critic1_loss = F.mse_loss(q1, target.detach())
        critic2_loss = F.mse_loss(q2, target.detach())
        critic_loss = critic1_loss + critic2_loss
        self.critic1.optimizer.zero_grad()
        self.critic2.optimizer.zero_grad()
        critic_loss.backward()
        self.critic1.optimizer.step()
        self.critic2.optimizer.step()

        self.update_time += 1
        if self.update_time % self.delay_time != 0:
            return

        new_actions_tensor = self.actor.forward(states_tensor)
        q1 = self.critic1.forward(states_tensor, new_actions_tensor)
        actor_loss = -T.mean(q1)
        self.actor.optimizer.zero_grad()
        actor_loss.backward()
        self.actor.optimizer.step()

        self.update_network_parameters()

    def save_models(self, episode):
        self.actor.save_checkpoint(self.checkpoint_dir + 'Actor/TD3_actor_{}.pth'.format(episode))
        print('Saving actor network successfully!')
        self.target_actor.save_checkpoint(self.checkpoint_dir +
                                          'Target_actor/TD3_target_actor_{}.pth'.format(episode))
        print('Saving target_actor network successfully!')
        self.critic1.save_checkpoint(self.checkpoint_dir + 'Critic1/TD3_critic1_{}.pth'.format(episode))
        print('Saving critic1 network successfully!')
        self.target_critic1.save_checkpoint(self.checkpoint_dir +
                                            'Target_critic1/TD3_target_critic1_{}.pth'.format(episode))
        print('Saving target critic1 network successfully!')
        self.critic2.save_checkpoint(self.checkpoint_dir + 'Critic2/TD3_critic2_{}.pth'.format(episode))
        print('Saving critic2 network successfully!')
        self.target_critic2.save_checkpoint(self.checkpoint_dir +
                                            'Target_critic2/TD3_target_critic2_{}.pth'.format(episode))
        print('Saving target critic2 network successfully!')

    def load_models(self, episode):
        self.actor.load_checkpoint(self.checkpoint_dir + 'Actor/TD3_actor_{}.pth'.format(episode))
        print('Loading actor network successfully!')
        self.target_actor.load_checkpoint(self.checkpoint_dir +
                                          'Target_actor/TD3_target_actor_{}.pth'.format(episode))
        print('Loading target_actor network successfully!')
        self.critic1.load_checkpoint(self.checkpoint_dir + 'Critic1/TD3_critic1_{}.pth'.format(episode))
        print('Loading critic1 network successfully!')
        self.target_critic1.load_checkpoint(self.checkpoint_dir +
                                            'Target_critic1/TD3_target_critic1_{}.pth'.format(episode))
        print('Loading target critic1 network successfully!')
        self.critic2.load_checkpoint(self.checkpoint_dir + 'Critic2/TD3_critic2_{}.pth'.format(episode))
        print('Loading critic2 network successfully!')
        self.target_critic2.load_checkpoint(self.checkpoint_dir +
                                            'Target_critic2/TD3_target_critic2_{}.pth'.format(episode))
        print('Loading target critic2 network successfully!')

算法仿真环境是gym库中的LunarLanderContinuous-v2环境，因此需要先配置好gym库。进入Aanconda中对应的Python环境中，执行下面的指令

pip install gym

但是，这样安装的gym库只包括少量的内置环境，如算法环境、简单文字游戏环境和经典控制环境，无法使用LunarLanderContinuous-v2。因此还要安装一些其他依赖项，具体可以参照这篇blog: AttributeError: module ‘gym.envs.box2d‘ has no attribute ‘LunarLander‘解决办法。如果已经配置好环境，那请忽略这一段。

训练脚本（train.py）：

import gym
import numpy as np
import argparse
from TD3 import TD3
from utils import create_directory, plot_learning_curve, scale_action

parser = argparse.ArgumentParser()
parser.add_argument('--max_episodes', type=int, default=1000)
parser.add_argument('--ckpt_dir', type=str, default='./checkpoints/TD3/')
parser.add_argument('--figure_file', type=str, default='./output_images/reward.png')

args = parser.parse_args()

def main():
    env = gym.make('LunarLanderContinuous-v2')
    agent = TD3(alpha=0.0003, beta=0.0003, state_dim=env.observation_space.shape[0],
                action_dim=env.action_space.shape[0], actor_fc1_dim=400, actor_fc2_dim=300,
                critic_fc1_dim=400, critic_fc2_dim=300, ckpt_dir=args.ckpt_dir, gamma=0.99,
                tau=0.005, action_noise=0.1, policy_noise=0.2, policy_noise_clip=0.5,
                delay_time=2, max_size=1000000, batch_size=256)
    create_directory(path=args.ckpt_dir, sub_path_list=['Actor', 'Critic1', 'Critic2', 'Target_actor',
                                                        'Target_critic1', 'Target_critic2'])

    total_reward_history = []
    avg_reward_history = []
    for episode in range(args.max_episodes):
        total_reward = 0
        done = False
        observation = env.reset()
        while not done:
            action = agent.choose_action(observation, train=True)
            action_ = scale_action(action, low=env.action_space.low, high=env.action_space.high)
            observation_, reward, done, info = env.step(action_)
            agent.remember(observation, action, reward, observation_, done)
            agent.learn()
            total_reward += reward
            observation = observation_
        total_reward_history.append(total_reward)
        avg_reward = np.mean(total_reward_history[-100:])
        avg_reward_history.append(avg_reward)
        print('Ep: {} Reward: {} AvgReward: {}'.format(episode+1, total_reward, avg_reward))

        if (episode + 1) % 200 == 0:
            agent.save_models(episode+1)

    episodes = [i+1 for i in range(args.max_episodes)]
    plot_learning_curve(episodes, avg_reward_history, title='AvgReward', ylabel='reward',
                        figure_file=args.figure_file)

if __name__ == '__main__':
    main()

训练脚本中有三个参数，max_episodes表示训练幕数，checkpoint_dir表示训练权重保存路径，figure_file表示训练结果的保存路径（其实是一张累积奖励曲线图），按照默认设置即可。

训练时还会用到画图函数和创建文件夹函数，它们都被放在utils.py脚本中：

import os
import numpy as np
import matplotlib.pyplot as plt

def create_directory(path: str, sub_path_list: list):
    for sub_path in sub_path_list:
        if not os.path.exists(path + sub_path):
            os.makedirs(path + sub_path, exist_ok=True)
            print('Path: {} create successfully!'.format(path + sub_path))
        else:
            print('Path: {} is already existence!'.format(path + sub_path))

def plot_learning_curve(episodes, records, title, ylabel, figure_file):
    plt.figure()
    plt.plot(episodes, records, color='b', linestyle='-')
    plt.title(title)
    plt.xlabel('episode')
    plt.ylabel(ylabel)

    plt.show()
    plt.savefig(figure_file)

def scale_action(action, low, high):
    action = np.clip(action, -1, 1)
    weight = (high - low) / 2
    bias = (high + low) / 2
    action_ = action * weight + bias

    return action_

另外我们还提供了测试代码，主要用于测试训练效果以及观察环境的动态渲染 (test.py)：

import gym
import imageio
import argparse
from TD3 import TD3
from utils import scale_action

parser = argparse.ArgumentParser()
parser.add_argument('--ckpt_dir', type=str, default='./checkpoints/TD3/')
parser.add_argument('--figure_file', type=str, default='./output_images/LunarLander.gif')
parser.add_argument('--fps', type=int, default=30)
parser.add_argument('--render', type=bool, default=True)
parser.add_argument('--save_video', type=bool, default=True)

args = parser.parse_args()

def main():
    env = gym.make('LunarLanderContinuous-v2')
    agent = TD3(alpha=0.0003, beta=0.0003, state_dim=env.observation_space.shape[0],
                action_dim=env.action_space.shape[0], actor_fc1_dim=400, actor_fc2_dim=300,
                critic_fc1_dim=400, critic_fc2_dim=300, ckpt_dir=args.ckpt_dir, gamma=0.99,
                tau=0.005, action_noise=0.1, policy_noise=0.2, policy_noise_clip=0.5,
                delay_time=2, max_size=1000000, batch_size=256)
    agent.load_models(1000)
    video = imageio.get_writer(args.figure_file, fps=args.fps)

    done = False
    observation = env.reset()
    while not done:
        if args.render:
            env.render()
        action = agent.choose_action(observation, train=True)
        action_ = scale_action(action, low=env.action_space.low, high=env.action_space.high)
        observation_, reward, done, info = env.step(action_)
        observation = observation_
        if args.save_video:
            video.append_data(env.render(mode='rgb_array'))

if __name__ == '__main__':
    main()

测试脚本中包括五个参数，filename表示环境动态图的保存路径，checkpoint_dir表示加载的权重路径，save_video表示是否要保存动态图，fps表示动态图的帧率，rander表示是否开启环境渲染。大家只需要调整save_video和rander这两个参数，其余保持默认即可。

6 实验结果

通过平均奖励曲线可以看出，大概迭代到400步左右时算法趋于收敛。相较于DDPG算法，TD3算法的性能有了明显提升。

这是测试效果图，智能体能够很好地完成降落任务，整个过程非常平稳！

7 结论

本文主要讲解了TD3算法中的相关技术细节，并进行了代码实现。TD3算法是一种有效解决连续控制问题的深度强化学习算法，也是我非常喜欢用的算法之一，希望各位小伙伴能够理解并掌握这个算法。

以上如果有出现错误的地方，欢迎各位怒斥！

标签： 深度学习 人工智能 python
本文转载自: https://blog.csdn.net/weixin_46133643/article/details/124509577
版权归原作者 indigo love 所有， 如有侵权，请联系我们删除。
登录后发布评论

“深度强化学习-TD3算法原理与代码”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

落实算法安全主体责任：全面保障算法合规与安全

Hadoop生态圈框架部署（四）- Hadoop完全分布式部署

毕设项目 基于大数据的b站数据分析

【前端】Bootstrap：栅格系统 (Grid System)

docker部署rustdesk

【PostgreSQL】实战篇——用户管理、角色和权限控制的高级用法及技巧

AI时代：弯道超车的新思维与实践路径

文章导航