PyTorch 深度学习实战(15):Twin Delayed DDPG (TD3) 算法

server/2025/3/17 21:56:42/

在上一篇文章中,我们介绍了 Deep Deterministic Policy Gradient (DDPG) 算法,并使用它解决了 Pendulum 问题。本文将深入探讨 Twin Delayed DDPG (TD3) 算法,这是一种改进的 DDPG 算法,能够有效解决 DDPG 中的过估计问题。我们将使用 PyTorch 实现 TD3 算法,并应用于经典的 Pendulum 问题。


一、TD3 算法基础

TD3 是 DDPG 的改进版本,通过引入以下三个关键技术来解决 DDPG 中的过估计问题:

  1. 双重 Critic 网络

    使用两个 Critic 网络来估计 Q 值,从而减少过估计问题。
  2. 延迟更新

    延迟 Actor 网络的更新,确保 Critic 网络更稳定地收敛。
  3. 目标策略平滑

    在目标动作中加入噪声,从而减少 Critic 网络的过拟合。

1. TD3 的核心思想

  • 双重 Critic 网络

    • 使用两个 Critic 网络来估计 Q 值,取两者中的较小值作为目标 Q 值,从而减少过估计。

  • 延迟更新

    • 每更新 Critic 网络多次,才更新一次 Actor 网络,确保 Critic 网络更稳定地收敛。

  • 目标策略平滑

    • 在目标动作中加入噪声,从而减少 Critic 网络的过拟合。

2. TD3 的优势

  • 减少过估计

    • 通过双重 Critic 网络和目标策略平滑,TD3 能够有效减少 Q 值的过估计。

  • 训练稳定

    • 延迟更新策略确保 Critic 网络更稳定地收敛。

  • 适用于连续动作空间

    • TD3 能够直接输出连续动作,适用于机器人控制、自动驾驶等任务。

3. TD3 的算法流程

  1. 使用当前策略采样一批数据。

  2. 使用目标网络计算目标 Q 值。

  3. 更新 Critic 网络以最小化 Q 值的误差。

  4. 延迟更新 Actor 网络以最大化 Q 值。

  5. 更新目标网络。

  6. 重复上述过程,直到策略收敛。


二、Pendulum 问题实战

我们将使用 PyTorch 实现 TD3 算法,并应用于 Pendulum 问题。目标是控制摆杆使其保持直立。

1. 问题描述

Pendulum 环境的状态空间包括摆杆的角度和角速度。动作空间是一个连续的扭矩值,范围在 −2,2 之间。智能体每保持摆杆直立一步,就会获得一个负的奖励,目标是最大化累积奖励。

2. 实现步骤

  1. 安装并导入必要的库。

  2. 定义 Actor 网络和 Critic 网络。

  3. 定义 TD3 训练过程。

  4. 测试模型并评估性能。

3. 代码实现

以下是完整的代码实现:

import gym
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
import numpy as np
import random
from collections import deque
import matplotlib.pyplot as plt
​
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False
​
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"使用设备: {device}")
​
env = gym.make('Pendulum-v1')
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.shape[0]
max_action = float(env.action_space.high[0])
​
SEED = 42
torch.manual_seed(SEED)
np.random.seed(SEED)
random.seed(SEED)
​
​
# 改进的 Actor 网络(增加层归一化)
class Actor(nn.Module):def __init__(self, state_dim, action_dim, max_action):super(Actor, self).__init__()self.l1 = nn.Linear(state_dim, 256)self.ln1 = nn.LayerNorm(256)self.l2 = nn.Linear(256, 256)self.ln2 = nn.LayerNorm(256)self.l3 = nn.Linear(256, action_dim)self.max_action = max_action
​def forward(self, x):x = F.relu(self.ln1(self.l1(x)))x = F.relu(self.ln2(self.l2(x)))x = torch.tanh(self.l3(x)) * self.max_actionreturn x
​
​
# 改进的 Critic 网络(增加层归一化)
class Critic(nn.Module):def __init__(self, state_dim, action_dim):super(Critic, self).__init__()self.l1 = nn.Linear(state_dim + action_dim, 256)self.ln1 = nn.LayerNorm(256)self.l2 = nn.Linear(256, 256)self.ln2 = nn.LayerNorm(256)self.l3 = nn.Linear(256, 1)
​def forward(self, x, u):x = F.relu(self.ln1(self.l1(torch.cat([x, u], 1))))x = F.relu(self.ln2(self.l2(x)))x = self.l3(x)return x
​
​
class TD3:def __init__(self, state_dim, action_dim, max_action):self.actor = Actor(state_dim, action_dim, max_action).to(device)self.actor_target = Actor(state_dim, action_dim, max_action).to(device)self.actor_target.load_state_dict(self.actor.state_dict())self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=3e-4)
​self.critic1 = Critic(state_dim, action_dim).to(device)self.critic2 = Critic(state_dim, action_dim).to(device)self.critic1_target = Critic(state_dim, action_dim).to(device)self.critic2_target = Critic(state_dim, action_dim).to(device)self.critic1_target.load_state_dict(self.critic1.state_dict())self.critic2_target.load_state_dict(self.critic2.state_dict())self.critic1_optimizer = optim.Adam(self.critic1.parameters(), lr=3e-4)self.critic2_optimizer = optim.Adam(self.critic2.parameters(), lr=3e-4)
​self.max_action = max_actionself.replay_buffer = deque(maxlen=1000000)self.batch_size = 256self.gamma = 0.99self.tau = 0.005self.policy_noise = 0.2self.noise_clip = 0.5self.policy_freq = 2self.total_it = 0self.exploration_noise = 0.1  # 新增探索噪声
​def select_action(self, state, add_noise=True):state = torch.FloatTensor(state.reshape(1, -1)).to(device)action = self.actor(state).cpu().data.numpy().flatten()if add_noise:noise = np.random.normal(0, self.exploration_noise, size=action_dim)action = (action + noise).clip(-self.max_action, self.max_action)return action
​def train(self):if len(self.replay_buffer) < self.batch_size:return
​self.total_it += 1
​batch = random.sample(self.replay_buffer, self.batch_size)state = torch.FloatTensor(np.array([t[0] for t in batch])).to(device)action = torch.FloatTensor(np.array([t[1] for t in batch])).to(device)reward = torch.FloatTensor(np.array([t[2] for t in batch])).reshape(-1, 1).to(device) / 10.0  # 奖励缩放next_state = torch.FloatTensor(np.array([t[3] for t in batch])).to(device)done = torch.FloatTensor(np.array([t[4] for t in batch])).reshape(-1, 1).to(device)
​with torch.no_grad():noise = (torch.randn_like(action) * self.policy_noise).clamp(-self.noise_clip, self.noise_clip)next_action = (self.actor_target(next_state) + noise).clamp(-self.max_action, self.max_action)target_Q1 = self.critic1_target(next_state, next_action)target_Q2 = self.critic2_target(next_state, next_action)target_Q = torch.min(target_Q1, target_Q2)target_Q = reward + (1 - done) * self.gamma * target_Q
​current_Q1 = self.critic1(state, action)current_Q2 = self.critic2(state, action)critic1_loss = F.mse_loss(current_Q1, target_Q)critic2_loss = F.mse_loss(current_Q2, target_Q)self.critic1_optimizer.zero_grad()critic1_loss.backward()torch.nn.utils.clip_grad_norm_(self.critic1.parameters(), 1.0)  # 梯度裁剪self.critic1_optimizer.step()self.critic2_optimizer.zero_grad()critic2_loss.backward()torch.nn.utils.clip_grad_norm_(self.critic2.parameters(), 1.0)self.critic2_optimizer.step()
​if self.total_it % self.policy_freq == 0:actor_loss = -self.critic1(state, self.actor(state)).mean()self.actor_optimizer.zero_grad()actor_loss.backward()torch.nn.utils.clip_grad_norm_(self.actor.parameters(), 1.0)self.actor_optimizer.step()
​for param, target_param in zip(self.critic1.parameters(), self.critic1_target.parameters()):target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)for param, target_param in zip(self.critic2.parameters(), self.critic2_target.parameters()):target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)for param, target_param in zip(self.actor.parameters(), self.actor_target.parameters()):target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)
​def save(self, filename):torch.save(self.actor.state_dict(), filename + "_actor.pth")
​
​
def train_td3(env, agent, episodes=2000, early_stop_threshold=-150):rewards_history = []moving_avg = []best_avg = -np.inf
​for ep in range(episodes):state,_ = env.reset()episode_reward = 0done = Falsestep = 0
​while not done:# 线性衰减探索噪声if ep < 300:agent.exploration_noise = max(0.5 * (1 - ep / 300), 0.1)else:agent.exploration_noise = 0.1
​action = agent.select_action(state, add_noise=(ep < 100))  # 前100轮强制探索next_state, reward, done, _, _ = env.step(action)agent.replay_buffer.append((state, action, reward, next_state, done))state = next_stateepisode_reward += rewardagent.train()step += 1
​rewards_history.append(episode_reward)current_avg = np.mean(rewards_history[-50:])moving_avg.append(current_avg)
​if current_avg > best_avg:best_avg = current_avgagent.save("td3_pendulum_best")
​if (ep + 1) % 50 == 0:print(f"Episode: {ep + 1}, Avg Reward: {current_avg:.2f}")
​# 早停机制if current_avg >= early_stop_threshold:print(f"早停触发,平均奖励达到 {current_avg:.2f}")break
​return moving_avg, rewards_history
​
​
# 训练并可视化
td3_agent = TD3(state_dim, action_dim, max_action)
moving_avg, rewards_history = train_td3(env, td3_agent, episodes=2000)
​
# 可视化结果
plt.figure(figsize=(12, 6))
plt.plot(rewards_history, alpha=0.6, label='single round reward')
plt.plot(moving_avg, 'r-', linewidth=2, label='moving average (50 rounds)')
plt.xlabel('episodes')
plt.ylabel('reward')
plt.title('TD3 training performance on Pendulum-v1')
plt.legend()
plt.grid(True)
plt.show()

三、代码解析

  1. Actor 和 Critic 网络

    • Actor 网络输出连续动作,通过 tanh 函数将动作限制在 −max_action,max_action 范围内。

    • Critic 网络输出状态-动作对的 Q 值。

  2. TD3 训练过程

    • 使用当前策略采样一批数据。

    • 使用目标网络计算目标 Q 值。

    • 更新 Critic 网络以最小化 Q 值的误差。

    • 延迟更新 Actor 网络以最大化 Q 值。

    • 更新目标网络。

  3. 训练过程

    • 在训练过程中,每 50 个 episode 打印一次平均奖励。

    • 训练结束后,绘制训练过程中的总奖励曲线。


四、运行结果

运行上述代码后,你将看到以下输出:

  • 训练过程中每 50 个 episode 打印一次平均奖励。

  • 训练结束后,绘制训练过程中的总奖励曲线。


五、总结

本文介绍了 TD3 算法的基本原理,并使用 PyTorch 实现了一个简单的 TD3 模型来解决 Pendulum 问题。通过这个例子,我们学习了如何使用 TD3 算法进行连续动作空间的策略优化。

在下一篇文章中,我们将探讨更高级的强化学习算法,如 Soft Actor-Critic (SAC)。敬请期待!

代码实例说明

  • 本文代码可以直接在 Jupyter Notebook 或 Python 脚本中运行。

  • 如果你有 GPU,代码会自动检测并使用 GPU 加速。

希望这篇文章能帮助你更好地理解 TD3 算法!如果有任何问题,欢迎在评论区留言讨论。


http://www.ppmy.cn/server/175791.html

相关文章

vue3:八、登录界面实现-忘记密码

该文章实现登录界面的忘记密码功能&#xff0c;点击忘记密码文本&#xff0c;打开dialog对话框 一、页面效果 加入忘记密码&#xff0c;在记住密码的同一行中&#xff0c;实现flex-between 二、对话框实现 1、新建组件页面 2、引入dialog组件到组件页面 参考路径 Dialog 对…

《Operating System Concepts》阅读笔记:p272-p285

《Operating System Concepts》学习第 27 天&#xff0c;p272-p285 总结&#xff0c;总计 14 页。 一、技术总结 1.semaphore A semaphore S is an integer variable that, apart from initialization, is accessed only through two standard atomic operations: wait() an…

【人工智能】人工智能安全(AI Security)

人工智能安全&#xff08;AI Security&#xff09; 是指保障人工智能系统免受各种攻击、滥用和错误操作的措施与技术。随着人工智能的广泛应用&#xff0c;AI的安全性问题变得越来越重要。AI安全不仅关注系统本身的稳定性与安全性&#xff0c;还涉及到如何确保AI的决策和行为是…

五种最新优化算法(ALA、AE、DOA、GOA、OX)求解多个无人机协同路径规划(可以自定义无人机数量及起始点),MATLAB代码

一、算法简介 &#xff08;一&#xff09;阿尔法进化&#xff08;Alpha Evolution&#xff0c;AE&#xff09;算法 阿尔法进化&#xff08;Alpha Evolution&#xff0c;AE&#xff09;算法是2024年提出的一种新型进化算法&#xff0c;其核心在于通过自适应基向量和随机步长的…

群体智能优化算法-䲟鱼优化算法 (Remora Optimization Algorithm, ROA,含Matlab源代码)

摘要 䲟鱼优化算法&#xff08;Remora Optimization Algorithm&#xff0c;ROA&#xff09;是一种基于䲟鱼在海洋中寄生与捕食者间交互关系而提出的元启发式算法。通过模拟䲟鱼在宿主附近进行寄生、吸附和随机机动等行为&#xff0c;ROA 在全局与局部搜索之间取得平衡。本文提…

Windows安全日志Defender 的配置被修改5007

事件 ID 5007 的含义 当 Windows Defender 的配置&#xff08;如实时保护、扫描策略、排除项等&#xff09;被手动或通过策略修改时&#xff0c;系统会记录此事件。可能的原因包括&#xff1a; 手动修改&#xff1a;用户或管理员通过界面更改了 Defender 设置。组策略或脚本&…

Blender-MCP服务源码2-依赖分析

Blender-MCP服务源码2-依赖分析 有个大佬做了一个Blender-MCP源码&#xff0c;第一次提交代码是【2025年3月7号】今天是【2025年月15日】也就是刚过去一周的时间&#xff0c;所以想从0开始学习这个代码&#xff0c;了解一下大佬们的开发思路 1-核心知识点 from mcp.server.fas…

基于asp.net实现的连锁餐厅收银系统[包运行成功+永久免费答疑辅导]

基于ASP.NET实现的连锁餐厅收银系统背景&#xff0c;可以从以下几个方面进行阐述&#xff1a; 一、技术背景 ASP.NET框架的普及与优势&#xff1a; ASP.NET是微软开发的一种用于构建Web应用程序的框架&#xff0c;它基于.NET Framework&#xff0c;提供了丰富的类库和开发工具…