logo

开发者热搜

深度解析TD3算法：原理、实现与TensorFlow 2.0实战指南

作者：很酷cat2025.09.18 17:44浏览量：1

简介：本文深入解析强化学习中的TD3算法，详细阐述其核心思想、算法流程及与DDPG的对比优势，并给出基于TensorFlow 2.0的完整实现代码，帮助读者掌握TD3算法的原理与实践。

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

一、引言

强化学习（Reinforcement Learning, RL）作为机器学习的重要分支，近年来在机器人控制、游戏AI、自动驾驶等领域取得了显著进展。其中，深度确定性策略梯度（Deep Deterministic Policy Gradient, DDPG）算法因其能够处理连续动作空间的问题而备受关注。然而，DDPG算法在实际应用中存在过估计（Overestimation）问题，导致策略性能下降。为了解决这一问题，TD3（Twin Delayed Deep Deterministic Policy Gradient）算法应运而生，它通过引入双批评家网络（Twin Critic Networks）、延迟策略更新（Delayed Policy Update）和目标策略平滑正则化（Target Policy Smoothing Regularization）等技术，有效缓解了过估计问题，提升了算法的稳定性和性能。

本文将详细解析TD3算法的核心思想、算法流程，并给出基于TensorFlow 2.0的完整实现代码，帮助读者深入理解TD3算法的原理与实践。

二、TD3算法核心思想

1. 双批评家网络

DDPG算法中，单个批评家网络（Critic Network）用于估计状态-动作对的Q值。然而，由于神经网络本身的近似误差和训练过程中的噪声，单个批评家网络容易产生过估计问题，即估计的Q值高于真实值。TD3算法通过引入双批评家网络，即两个独立的批评家网络分别估计Q值，并取两者中的较小值作为目标Q值，从而有效缓解了过估计问题。

2. 延迟策略更新

在DDPG算法中，策略网络（Actor Network）和批评家网络同时更新。然而，这种同步更新方式可能导致策略网络基于过估计的Q值进行更新，进而影响策略的性能。TD3算法采用延迟策略更新的方式，即先更新批评家网络，待批评家网络稳定后再更新策略网络，从而减少了过估计对策略更新的影响。

3. 目标策略平滑正则化

为了进一步减少过估计，TD3算法在目标Q值的计算中引入了目标策略平滑正则化。具体来说，对目标动作添加一定的噪声，然后计算多个噪声动作对应的Q值，并取平均值作为目标Q值。这种方法使得目标Q值对动作噪声更加鲁棒，从而减少了过估计。

三、TD3算法流程

TD3算法的流程如下：

  1. 初始化：初始化策略网络、两个批评家网络及其目标网络，初始化经验回放缓冲区。
  2. 交互：智能体与环境交互，收集状态、动作、奖励、下一状态等信息，并存入经验回放缓冲区。
  3. 采样：从经验回放缓冲区中随机采样一批数据。
  4. 计算目标Q值：
    • 对目标动作添加噪声，得到噪声动作。
    • 使用两个目标批评家网络分别计算噪声动作对应的Q值。
    • 取两个Q值中的较小值作为目标Q值。
  5. 更新批评家网络：使用均方误差损失函数更新两个批评家网络。
  6. 延迟更新策略网络：每隔一定步数，使用策略梯度更新策略网络。
  7. 更新目标网络：使用软更新（Soft Update）方式更新目标批评家网络和目标策略网络。

四、TensorFlow 2.0实现

以下是基于TensorFlow 2.0的TD3算法实现代码：

  1. import tensorflow as tf
  2. import numpy as np
  3. import gym
  4. from collections import deque
  5. import random
  7. # 环境设置
  8. env = gym.make('Pendulum-v0')
  9. state_size = env.observation_space.shape[0]
  10. action_size = env.action_space.shape[0]
  11. action_bound = env.action_space.high[0]
  13. # 超参数设置
  14. BUFFER_SIZE = 10000
  15. BATCH_SIZE = 64
  16. GAMMA = 0.99
  17. TAU = 0.005
  18. LR_ACTOR = 0.001
  19. LR_CRITIC = 0.002
  20. EXPLORATION_NOISE = 0.1
  21. POLICY_NOISE = 0.2
  22. NOISE_CLIP = 0.5
  23. POLICY_UPDATE_FREQUENCY = 2
  25. # 经验回放缓冲区
  26. class ReplayBuffer:
  27.     def __init__(self, buffer_size):
  28.         self.buffer = deque(maxlen=buffer_size)
  30.     def add(self, state, action, reward, next_state, done):
  31.         self.buffer.append((state, action, reward, next_state, done))
  33.     def sample(self, batch_size):
  34.         batch = random.sample(self.buffer, batch_size)
  35.         state, action, reward, next_state, done = map(np.array, zip(*batch))
  36.         return state, action, reward, next_state, done
  38.     def size(self):
  39.         return len(self.buffer)
  41. # 策略网络
  42. class Actor(tf.keras.Model):
  43.     def __init__(self, state_size, action_size, action_bound):
  44.         super(Actor, self).__init__()
  45.         self.dense1 = tf.keras.layers.Dense(256, activation='relu')
  46.         self.dense2 = tf.keras.layers.Dense(256, activation='relu')
  47.         self.dense3 = tf.keras.layers.Dense(action_size, activation='tanh')
  48.         self.action_bound = action_bound
  50.     def call(self, state):
  51.         x = self.dense1(state)
  52.         x = self.dense2(x)
  53.         x = self.dense3(x) * self.action_bound
  54.         return x
  56. # 批评家网络
  57. class Critic(tf.keras.Model):
  58.     def __init__(self, state_size, action_size):
  59.         super(Critic, self).__init__()
  60.         self.dense1 = tf.keras.layers.Dense(256, activation='relu')
  61.         self.dense2 = tf.keras.layers.Dense(256, activation='relu')
  62.         self.dense3 = tf.keras.layers.Dense(1)
  64.     def call(self, state, action):
  65.         x = tf.concat([state, action], axis=-1)
  66.         x = self.dense1(x)
  67.         x = self.dense2(x)
  68.         x = self.dense3(x)
  69.         return x
  71. # TD3算法类
  72. class TD3:
  73.     def __init__(self, state_size, action_size, action_bound):
  74.         self.state_size = state_size
  75.         self.action_size = action_size
  76.         self.action_bound = action_bound
  78.         self.actor = Actor(state_size, action_size, action_bound)
  79.         self.actor_target = Actor(state_size, action_size, action_bound)
  80.         self.actor_target.set_weights(self.actor.get_weights())
  82.         self.critic1 = Critic(state_size, action_size)
  83.         self.critic2 = Critic(state_size, action_size)
  84.         self.critic1_target = Critic(state_size, action_size)
  85.         self.critic2_target = Critic(state_size, action_size)
  86.         self.critic1_target.set_weights(self.critic1.get_weights())
  87.         self.critic2_target.set_weights(self.critic2.get_weights())
  89.         self.actor_optimizer = tf.keras.optimizers.Adam(LR_ACTOR)
  90.         self.critic1_optimizer = tf.keras.optimizers.Adam(LR_CRITIC)
  91.         self.critic2_optimizer = tf.keras.optimizers.Adam(LR_CRITIC)
  93.         self.replay_buffer = ReplayBuffer(BUFFER_SIZE)
  94.         self.policy_update_counter = 0
  96.     def act(self, state, add_noise=True):
  97.         state = tf.convert_to_tensor([state], dtype=tf.float32)
  98.         action = self.actor(state).numpy()[0]
  99.         if add_noise:
  100.             action += np.random.normal(0, EXPLORATION_NOISE, size=self.action_size)
  101.             action = np.clip(action, -self.action_bound, self.action_bound)
  102.         return action
  104.     def learn(self, state, action, reward, next_state, done):
  105.         self.replay_buffer.add(state, action, reward, next_state, done)
  107.         if self.replay_buffer.size() < BATCH_SIZE:
  108.             return
  110.         state, action, reward, next_state, done = self.replay_buffer.sample(BATCH_SIZE)
  111.         state = tf.convert_to_tensor(state, dtype=tf.float32)
  112.         action = tf.convert_to_tensor(action, dtype=tf.float32)
  113.         reward = tf.convert_to_tensor(reward, dtype=tf.float32)
  114.         next_state = tf.convert_to_tensor(next_state, dtype=tf.float32)
  115.         done = tf.convert_to_tensor(done, dtype=tf.float32)
  117.         # 计算目标Q值
  118.         next_action = self.actor_target(next_state)
  119.         noise = tf.random.normal(tf.shape(next_action), 0, POLICY_NOISE)
  120.         noise = tf.clip_by_value(noise, -NOISE_CLIP, NOISE_CLIP)
  121.         next_action = tf.clip_by_value(next_action + noise, -self.action_bound, self.action_bound)
  123.         target_q1 = self.critic1_target(next_state, next_action)
  124.         target_q2 = self.critic2_target(next_state, next_action)
  125.         target_q = tf.minimum(target_q1, target_q2)
  126.         target_q = reward + (1 - done) * GAMMA * target_q
  128.         # 更新批评家网络
  129.         with tf.GradientTape() as tape:
  130.             current_q1 = self.critic1(state, action)
  131.             current_q2 = self.critic2(state, action)
  132.             critic1_loss = tf.reduce_mean(tf.square(target_q - current_q1))
  133.             critic2_loss = tf.reduce_mean(tf.square(target_q - current_q2))
  135.         critic1_grads = tape.gradient(critic1_loss, self.critic1.trainable_variables)
  136.         critic2_grads = tape.gradient(critic2_loss, self.critic2.trainable_variables)
  137.         self.critic1_optimizer.apply_gradients(zip(critic1_grads, self.critic1.trainable_variables))
  138.         self.critic2_optimizer.apply_gradients(zip(critic2_grads, self.critic2.trainable_variables))
  140.         # 延迟更新策略网络
  141.         self.policy_update_counter += 1
  142.         if self.policy_update_counter % POLICY_UPDATE_FREQUENCY == 0:
  143.             with tf.GradientTape() as tape:
  144.                 new_action = self.actor(state)
  145.                 actor_loss = -self.critic1(state, new_action)  # 只使用critic1计算策略梯度
  146.                 actor_loss = tf.reduce_mean(actor_loss)
  148.             actor_grads = tape.gradient(actor_loss, self.actor.trainable_variables)
  149.             self.actor_optimizer.apply_gradients(zip(actor_grads, self.actor.trainable_variables))
  151.             # 软更新目标网络
  152.             self.soft_update(self.critic1, self.critic1_target, TAU)
  153.             self.soft_update(self.critic2, self.critic2_target, TAU)
  154.             self.soft_update(self.actor, self.actor_target, TAU)
  156.     def soft_update(self, model, model_target, tau):
  157.         for var, var_target in zip(model.trainable_variables, model_target.trainable_variables):
  158.             var_target.assign(tau * var + (1 - tau) * var_target)
  160. # 训练过程
  161. def train_td3(env, td3, episodes):
  162.     total_rewards = []
  163.     for episode in range(episodes):
  164.         state = env.reset()
  165.         episode_reward = 0
  166.         while True:
  167.             action = td3.act(state)
  168.             next_state, reward, done, _ = env.step(action)
  169.             td3.learn(state, action, reward, next_state, done)
  170.             state = next_state
  171.             episode_reward += reward
  172.             if done:
  173.                 break
  174.         total_rewards.append(episode_reward)
  175.         if (episode + 1) % 10 == 0:
  176.             print(f'Episode {episode + 1}, Average Reward: {np.mean(total_rewards[-10:])}')
  177.     return total_rewards
  179. # 主函数
  180. if __name__ == '__main__':
  181.     td3 = TD3(state_size, action_size, action_bound)
  182.     total_rewards = train_td3(env, td3, 1000)

五、总结与展望

TD3算法通过引入双批评家网络、延迟策略更新和目标策略平滑正则化等技术，有效缓解了DDPG算法中的过估计问题，提升了算法的稳定性和性能。本文详细解析了TD3算法的核心思想、算法流程，并给出了基于TensorFlow 2.0的完整实现代码。未来，随着深度学习和强化学习技术的不断发展，TD3算法及其变种有望在更多复杂场景中发挥重要作用。

相关文章推荐

发表评论

最热文章

关于作者

  • 被阅读数
  • 被赞数
  • 被收藏数
活动
咨询