强化学习 14 —— TD3 算法详解与tensorflow 2.0 实现

在强化学习领域，深度确定性策略梯度（Deep Deterministic Policy Gradient, DDPG）算法曾因其能够处理连续动作空间问题而广受关注。然而，DDPG在实际应用中常面临高估偏差（overestimation bias）和策略过拟合的问题。为了克服这些挑战，Scott Fujimoto等人在2018年提出了双延迟深度确定性策略梯度（Twin Delayed Deep Deterministic Policy Gradient, TD3）算法。本文将详细解析TD3算法的核心思想、优势，并通过TensorFlow 2.0框架提供完整的实现指南。

一、TD3算法的核心思想与优势

1.1 核心思想

TD3算法在DDPG的基础上引入了三项关键改进：

1. 双Q网络（Twin Q-Networks）：使用两个独立的Q网络（Q1和Q2）来估计目标值，并取两者中的较小值作为最终目标，以减少高估偏差。
2. 延迟策略更新（Delayed Policy Updates）：策略网络的更新频率低于Q网络，通常每更新两次Q网络后更新一次策略网络，以增强策略的稳定性。
3. 目标策略平滑（Target Policy Smoothing）：在目标Q网络计算目标值时，对目标动作添加一定的噪声，使Q值估计更加平滑，减少过拟合。

1.2 优势

• 减少高估偏差：通过双Q网络机制，TD3有效缓解了DDPG中的高估问题，提高了值函数估计的准确性。
• 增强策略稳定性：延迟策略更新和目标策略平滑机制共同作用，使策略网络的学习更加稳定，减少了策略振荡。
• 适用于复杂环境：TD3在连续动作空间的任务中表现出色，能够处理高维状态空间和复杂动态环境。

二、TD3算法的实现步骤

2.1 初始化网络

• Actor网络（策略网络）：输入状态，输出动作。
• Critic网络（Q网络）：输入状态和动作，输出Q值。TD3使用两个Critic网络（Q1和Q2）。
• 目标Actor网络和目标Critic网络：用于计算目标值，参数通过软更新（soft update）方式从主网络复制。

2.2 经验回放与采样

• 使用经验回放缓冲区（Replay Buffer）存储历史经验（状态、动作、奖励、下一状态、终止标志）。
• 每次训练时，从缓冲区中随机采样一批经验用于网络更新。

2.3 Q网络更新

1. 计算目标Q值：
   • 使用目标Actor网络生成目标动作，并添加噪声。
   • 使用目标Critic网络（Q1和Q2）计算两个目标Q值，取较小值作为最终目标。
   • 公式：$y = r + \gamma \cdot \min(Q_1’(s’, a’), Q_2’(s’, a’))$，其中$a’ = \pi’(s’) + \epsilon$，$\epsilon \sim \mathcal{N}(0, \sigma)$。
2. 更新Q网络：
   • 使用均方误差（MSE）损失函数更新Q1和Q2网络。

2.4 策略网络更新

• 延迟更新策略网络：通常每更新两次Q网络后更新一次策略网络。
• 更新策略网络时，使用当前Critic网络（Q1）的梯度来指导策略网络的学习。
• 公式：$\nabla\theta J(\theta) = \mathbb{E}{s \sim \mathcal{D}} [\nablaa Q_1(s, a)|{a=\pi(s)} \nabla_\theta \pi(s)]$。

2.5 目标网络软更新

• 定期将主网络的参数以较小的比例（如$\tau=0.005$）复制到目标网络。
• 公式：$\theta’ \leftarrow \tau \theta + (1-\tau) \theta’$。

三、TensorFlow 2.0实现指南

3.1 环境准备

首先，安装必要的库：

pip install tensorflow gym numpy matplotlib

3.2 网络定义

使用TensorFlow 2.0的tf.keras模块定义Actor和Critic网络：

import tensorflow as tf
from tensorflow.keras.layers import Dense
class Actor(tf.keras.Model):
    def __init__(self, state_dim, action_dim, max_action):
        super(Actor, self).__init__()
        self.l1 = Dense(256, activation='relu')
        self.l2 = Dense(256, activation='relu')
        self.l3 = Dense(action_dim, activation='tanh')
        self.max_action = max_action
    def call(self, state):
        x = tf.keras.activations.relu(self.l1(state))
        x = tf.keras.activations.relu(self.l2(x))
        x = self.max_action * self.l3(x)
        return x
class Critic(tf.keras.Model):
    def __init__(self, state_dim, action_dim):
        super(Critic, self).__init__()
        # Q1架构
        self.l1 = Dense(256, activation='relu')
        self.l2 = Dense(256, activation='relu')
        self.l3 = Dense(1)
        # Q2架构
        self.l4 = Dense(256, activation='relu')
        self.l5 = Dense(256, activation='relu')
        self.l6 = Dense(1)
    def call(self, state, action):
        xu = tf.concat([state, action], axis=1)
        # Q1
        x1 = tf.keras.activations.relu(self.l1(xu))
        x1 = tf.keras.activations.relu(self.l2(x1))
        x1 = self.l3(x1)
        # Q2
        x2 = tf.keras.activations.relu(self.l4(xu))
        x2 = tf.keras.activations.relu(self.l5(x2))
        x2 = self.l6(x2)
        return x1, x2

3.3 TD3算法实现

import numpy as np
import gym
from collections import deque
class TD3:
    def __init__(self, state_dim, action_dim, max_action):
        self.actor = Actor(state_dim, action_dim, max_action)
        self.actor_target = Actor(state_dim, action_dim, max_action)
        self.actor_target.set_weights(self.actor.get_weights())
        self.critic = Critic(state_dim, action_dim)
        self.critic_target = Critic(state_dim, action_dim)
        self.critic_target.set_weights(self.critic.get_weights())
        self.max_action = max_action
        self.tau = 0.005
        self.gamma = 0.99
        self.policy_noise = 0.2
        self.noise_clip = 0.5
        self.policy_freq = 2
        self.optimizer = tf.keras.optimizers.Adam(learning_rate=3e-4)
        self.replay_buffer = deque(maxlen=1e6)
    def select_action(self, state):
        state = tf.convert_to_tensor([state], dtype=tf.float32)
        return self.actor(state).numpy()[0]
    def train(self, batch_size=100):
        if len(self.replay_buffer) < batch_size:
            return
        batch = np.random.choice(len(self.replay_buffer), batch_size, replace=False)
        states = np.array([self.replay_buffer[i][0] for i in batch])
        actions = np.array([self.replay_buffer[i][1] for i in batch])
        rewards = np.array([self.replay_buffer[i][2] for i in batch])
        next_states = np.array([self.replay_buffer[i][3] for i in batch])
        dones = np.array([self.replay_buffer[i][4] for i in batch])
        states = tf.convert_to_tensor(states, dtype=tf.float32)
        actions = tf.convert_to_tensor(actions, dtype=tf.float32)
        rewards = tf.convert_to_tensor(rewards, dtype=tf.float32).reshape(-1, 1)
        next_states = tf.convert_to_tensor(next_states, dtype=tf.float32)
        dones = tf.convert_to_tensor(dones, dtype=tf.float32).reshape(-1, 1)
        # 计算目标Q值
        noise = tf.random.normal(tf.shape(actions), 0, self.policy_noise)
        noise = tf.clip_by_value(noise, -self.noise_clip, self.noise_clip)
        next_actions = self.actor_target(next_states) + noise
        next_actions = tf.clip_by_value(next_actions, -self.max_action, self.max_action)
        target_Q1, target_Q2 = self.critic_target(next_states, next_actions)
        target_Q = tf.minimum(target_Q1, target_Q2)
        target_Q = rewards + (1 - dones) * self.gamma * target_Q
        # 更新Critic网络
        with tf.GradientTape() as tape:
            current_Q1, current_Q2 = self.critic(states, actions)
            critic_loss = tf.reduce_mean((current_Q1 - target_Q)**2 + (current_Q2 - target_Q)**2)
        critic_grads = tape.gradient(critic_loss, self.critic.trainable_variables)
        self.optimizer.apply_gradients(zip(critic_grads, self.critic.trainable_variables))
        # 延迟更新Actor网络
        if self.train_step % self.policy_freq == 0:
            with tf.GradientTape() as tape:
                actor_actions = self.actor(states)
                actor_Q1, _ = self.critic(states, actor_actions)
                actor_loss = -tf.reduce_mean(actor_Q1)
            actor_grads = tape.gradient(actor_loss, self.actor.trainable_variables)
            self.optimizer.apply_gradients(zip(actor_grads, self.actor.trainable_variables))
            # 软更新目标网络
            for var, target_var in zip(self.actor.trainable_variables, self.actor_target.trainable_variables):
                target_var.assign(self.tau * var + (1 - self.tau) * target_var)
            for var, target_var in zip(self.critic.trainable_variables, self.critic_target.trainable_variables):
                target_var.assign(self.tau * var + (1 - self.tau) * target_var)
        self.train_step += 1

3.4 训练与评估

env = gym.make('Pendulum-v0')
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.shape[0]
max_action = float(env.action_space.high[0])
td3 = TD3(state_dim, action_dim, max_action)
episodes = 1000
for ep in range(episodes):
    state = env.reset()
    episode_reward = 0
    while True:
        action = td3.select_action(state)
        next_state, reward, done, _ = env.step(action)
        td3.replay_buffer.append((state, action, reward, next_state, done))
        state = next_state
        episode_reward += reward
        td3.train()
        if done:
            break
    print(f'Episode {ep}, Reward: {episode_reward}')

四、总结与展望

TD3算法通过双Q网络、延迟策略更新和目标策略平滑三项关键改进，有效解决了DDPG中的高估偏差和策略过拟合问题。本文通过TensorFlow 2.0框架提供了完整的TD3算法实现，包括网络定义、经验回放、Q网络更新、策略网络更新和目标网络软更新等核心步骤。未来，可以进一步探索TD3算法在更复杂环境中的应用，如多智能体强化学习、部分可观测环境等。同时，结合其他先进技术（如注意力机制、图神经网络等），有望进一步提升TD3算法的性能和适用性。