【强化学习】双延迟深度确定性策略梯度算法(TD3)详解

【强化学习】双延迟深度确定性策略梯度算法(TD3)详解
        📢本篇文章是博主强化学习(RL)领域学习时,用于个人学习、研究或者欣赏使用,并基于博主对相关等领域的一些理解而记录的学习摘录和笔记,若有不当和侵权之处,指出后将会立即改正,还望谅解。文章分类在👉强化学习专栏:

       【强化学习】- 【单智能体强化学习】(11)---《双延迟深度确定性策略梯度算法(TD3)详解》

双延迟深度确定性策略梯度算法(TD3)详解

目录

一、TD3算法的背景

二、TD3的背景

1.TD3的理论背景

2.DDPG的局限性

三、TD3算法的核心思想

1.双Critic网络(Twin Critics)

2.延迟更新(Delayed Policy Updates)

3.目标策略平滑(Target Policy Smoothing)

四、TD3算法详细讲解

1. Actor-Critic 框架的核心

(1) 策略梯度

(2) 价值评估 (Critic)

2. TD3 的关键改进

(1) 双Critic网络

(2) 延迟Actor更新

(3) 目标动作平滑

3. 完整TD3算法流程

4. 数学细节解析

(1) Critic损失函数

(2) Actor策略梯度

(3) 延迟更新的效果

[Python] TD3算法的实现

TD3的简易版核心实现: 

 TD3算法的详细代码见下

[Notice]  主要模块及功能

五、TD3的优势

六、总结


一、TD3算法的背景

        双延迟深度确定性策略梯度算法,TD3(Twin Delayed Deep Deterministic Policy Gradient)是强化学习中专为解决连续动作空间问题设计的一种算法。TD3算法的提出是在深度确定性策略梯度(DDPG)算法的基础上改进而来,用于解决强化学习训练中存在的一些关键挑战。


二、TD3的背景

1.TD3的理论背景

        TD3的提出基于以下几个强化学习的理论与技术发展:

Actor-Critic架构

        Actor网络负责生成动作,Critic网络负责评估动作的价值(Q值)。这种架构使得算法能够高效地解决高维连续动作问题。

        Actor更新目标是最大化Critic网络的Q值,而Critic网络优化目标是最小化Q值预测误差。

确定性策略梯度(Deterministic Policy Gradient, DPG)

        DPG是强化学习中一种适用于连续动作空间的策略梯度方法,TD3继承了DPG的优势,即通过学习一个确定性策略直接生成动作。

双Q学习(Double Q-Learning)

        TD3借鉴了双Q学习的思想,使用两个独立的Critic网络来降低Q值估计的偏差。

经验回放池(Replay Buffer)

        TD3通过从经验回放池中采样数据训练网络,打破数据相关性,提高了学习效率。

2.DDPG的局限性

        TD3算法由Fujimoto等人在2018年提出,对深度确定性策略梯度(Deep Deterministic Policy Gradient, DDPG)算法的改进。DDPG是一种结合策略(Actor)和价值函数(Critic)的强化学习方法,可以在连续动作空间中表现出色。然而,DDPG存在以下问题:

这些问题会使训练结果不够鲁棒,甚至使算法在复杂任务中失败。

  • Q值过估计问题:Critic网络在训练时容易高估Q值,从而导致策略网络(Actor)学习不稳定。
  • 策略噪声问题:由于策略直接输出确定性动作,在训练时容易陷入局部最优解。
  • 训练不稳定性:Critic网络和Actor网络同时训练时,相互影响可能导致训练震荡。

了解决上述问题,TD3通过以下三点创新改进了DDPG:


三、TD3算法的核心思想

        TD3在DDPG的基础上提出了三项关键改进:

1.双Critic网络(Twin Critics)

        动机:DDPG中的Critic网络在估计Q值时存在系统性的高估问题。

        方法:TD3使用两个独立的Critic网络计算Q值,取两者的最小值来作为目标Q值。

        效果:有效减少了Q值的高估偏差(Overestimation Bias)。

2.延迟更新(Delayed Policy Updates)

        动机:在DDPG中,Critic网络和Actor网络同时更新,可能导致Actor策略在不稳定的Q值估计上进行优化。

        方法:TD3降低Actor和目标网络的更新频率,通常在Critic更新两次后才更新Actor。

        效果:降低了Actor网络的更新频率,从而提高了策略的稳定性。

3.目标策略平滑(Target Policy Smoothing)

        动机:DDPG中的目标策略直接输出确定性动作,容易对极端动作过拟合。TD3通过在目标策略中加入高斯噪声,对动作进行“平滑”。

        方法:在目标值计算中,对动作加入噪声并裁剪到一定范围。

        效果:提高了算法对噪声和目标值波动的鲁棒性。


四、TD3算法详细讲解

        TD3(Twin Delayed Deep Deterministic Policy Gradient)适用于连续动作空间问题,主要基于Actor-Critic框架和深度确定性策略梯度(DDPG)。以下是TD3的数学基础与推导。

1. Actor-Critic 框架的核心

        Actor-Critic方法的核心在于将策略学习(Actor)与价值评估(Critic)结合。Actor负责生成动作,Critic负责评估当前策略的表现。Actor网络优化目标是通过Critic网络的反馈提高策略质量。

(1) 策略梯度

        Actor通过最大化累计奖励学习最优策略:

\nabla_\phi J(\pi_\phi) = \mathbb{E}{s \sim \rho^\pi} \left[ \nabla\phi \pi_\phi(s) \nabla_a Q^\pi(s, a) \big|{a=\pi\phi(s)} \right]

其中:

\rho^\pi

是由策略生成的状态分布。

Q^\pi(s, a)

是Critic估计的动作值函数。

\pi_\phi(s)

是Actor的策略函数。

(2) 价值评估 (Critic)

        Critic通过最小化时间差分(Temporal Difference, TD)误差,学习状态-动作值函数:

L(\theta) = \mathbb{E}{(s, a, r, s') \sim \mathcal{D}} \left[ \big( Q\theta(s, a) - y \big)^2 \right]

其中目标值 ( y ) 定义为:

y = r + \gamma Q_{\theta'}(s', \pi_{\phi'}(s'))
\mathcal{D}

是经验回放池中采样的数据。

\theta'

\phi'

是Critic和Actor的目标网络参数。

2. TD3 的关键改进

        TD3在DDPG的基础上,针对Q值过估计和策略训练不稳定问题,提出了三项核心改进。

(1) 双Critic网络

        TD3引入两个Critic网络

Q_{\theta_1}

Q_{\theta_2}

,通过取最小值来降低Q值的高估偏差:

y = r + \gamma \min \big( Q_{\theta_1'}(s', \pi_{\phi'}(s')), Q_{\theta_2'}(s', \pi_{\phi'}(s')) \big)
  • 目标是防止策略在训练中受到错误Q值估计的误导。
(2) 延迟Actor更新

        为了避免Actor网络频繁更新导致策略不稳定,TD3在Critic更新  n 次后才更新Actor一次(通常 n=2 )。Actor的优化目标为:

L(\phi) = -\mathbb{E}{s \sim \mathcal{D}} \big[ Q{\theta_1}(s, \pi_\phi(s)) \big]

Critic网络训练稳定后,Actor的策略梯度才会更加准确。

L(\phi) = -\mathbb{E}{s} \big[ Q{\theta_1}(s, \pi_\phi(s)) \big]
(3) 目标动作平滑

        在计算目标值 y  时,对动作加入高斯噪声

\epsilon \sim \mathcal{N}(0, \sigma)

并进行裁剪,防止策略过拟合到极端动作:

a' = \pi_{\phi'}(s') + \text{clip}(\epsilon, -c, c)
  • 这样可以让目标Q值更加平滑,增强策略的鲁棒性。

3. 完整TD3算法流程

伪代码如下:

  1. 初始化Actor和Critic网络及其对应的目标网络;
  2. 重复以上步骤直到收敛。

目标网络软更新

\theta_i' \leftarrow \tau \theta_i + (1 - \tau) \theta_i' ] [ \phi' \leftarrow \tau \phi + (1 - \tau) \phi'

延迟更新Actor:每隔 d 步,更新Actor策略:

L(\phi) = -\mathbb{E}{s} \big[ Q{\theta_1}(s, \pi_\phi(s)) \big]

更新Critic:通过最小化损失函数更新

Q_{\theta_1}

Q_{\theta_2}

L(\theta_i) = \mathbb{E}{(s, a, r, s')} \big[ (Q{\theta_i}(s, a) - y)^2 \big]

\mathcal{D}

中随机抽取一个批量数据

(s, a, r, s')

与环境交互,使用当前策略

\pi_\phi

执行动作

a_t

,存储

(s_t, a_t, r_t, s_{t+1})

构建经验回放池

\mathcal{D}

,用于存储交互数据;


4. 数学细节解析

(1) Critic损失函数

        TD3使用两个Critic网络,损失函数为:

L(\theta) = \mathbb{E} \big[ (Q_{\theta}(s, a) - y)^2 \big]

        其中目标值:

y = r + \gamma \min \big( Q_{\theta_1'}(s', a'), Q_{\theta_2'}(s', a') \big)
(2) Actor策略梯度

        Actor通过最大化Critic网络的输出优化策略:

\nabla_\phi J(\phi) = \mathbb{E}{s \sim \rho^\pi} \big[ \nabla_a Q{\theta_1}(s, a) \big|{a=\pi\phi(s)} \nabla_\phi \pi_\phi(s) \big]
(3) 延迟更新的效果

        延迟更新使Actor网络只在Critic网络收敛后才更新,减少了Actor网络梯度被不准确Q值引导的风险,从而提高了稳定性。


[Python] TD3算法的实现

TD3的简易版核心实现: 

"""《TD3算法的代码》 时间:2024.12 作者:不去幼儿园 """ import torch import torch.nn as nn import torch.optim as optim import numpy as np import gym from collections import deque import random # Actor Network class Actor(nn.Module): def __init__(self, state_dim, action_dim, max_action): super(Actor, self).__init__() self.layer1 = nn.Linear(state_dim, 256) self.layer2 = nn.Linear(256, 256) self.layer3 = nn.Linear(256, action_dim) self.max_action = max_action def forward(self, x): x = torch.relu(self.layer1(x)) x = torch.relu(self.layer2(x)) x = self.max_action * torch.tanh(self.layer3(x)) return x # Critic Network class Critic(nn.Module): def __init__(self, state_dim, action_dim): super(Critic, self).__init__() self.layer1 = nn.Linear(state_dim + action_dim, 256) self.layer2 = nn.Linear(256, 256) self.layer3 = nn.Linear(256, 1) def forward(self, x, u): x = torch.cat([x, u], 1) x = torch.relu(self.layer1(x)) x = torch.relu(self.layer2(x)) return self.layer3(x) # TD3 Algorithm 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_optimizer = optim.Adam(self.actor.parameters(), lr=1e-3) 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.critic_optimizer = optim.Adam( list(self.critic1.parameters()) + list(self.critic2.parameters()), lr=1e-3 ) self.max_action = max_action self.replay_buffer = deque(maxlen=1000000) def update(self, batch_size=100, gamma=0.99, tau=0.005, policy_noise=0.2, noise_clip=0.5, delay=2): # Implementation of TD3 training logic here pass def select_action(self, state): state = torch.FloatTensor(state.reshape(1, -1)).to(device) return self.actor(state).cpu().data.numpy().flatten() 
 项目代码我已经放入GitCode里面,可以通过下面链接跳转:🔥

【强化学习】--- TD3算法 

后续相关单智能体强化学习算法也会不断在【强化学习】项目里更新,如果该项目对你有所帮助,请帮我点一个星星✨✨✨✨✨,鼓励分享,十分感谢!!!

若是下面代码复现困难或者有问题,也欢迎评论区留言

 TD3算法的详细代码见下

 参数配置

import argparse # 用于解析命令行参数的库 from collections import namedtuple # 提供轻量级的数据结构 from itertools import count # 无限循环计数器 import os, sys, random # 操作系统、系统操作及随机库 import numpy as np # 用于数组操作和科学计算的库 import gym # OpenAI Gym库,用于构建强化学习环境 import torch # 深度学习框架 PyTorch import torch.nn as nn # 神经网络模块 import torch.nn.functional as F # 提供激活函数和其他功能 import torch.optim as optim # 优化器模块,用于梯度下降 from torch.distributions import Normal # 正态分布,用于策略采样 from tensorboardX import SummaryWriter # 用于记录训练日志 # 如果GPU可用,则使用CUDA,否则使用CPU device = 'cuda' if torch.cuda.is_available() else 'cpu' # 创建一个命令行参数解析器 parser = argparse.ArgumentParser() # 添加脚本运行时的参数 parser.add_argument('--mode', default='train', type=str) # 模式:训练('train')或测试('test') parser.add_argument("--env_name", default="Pendulum-v0") # OpenAI Gym环境名称 parser.add_argument('--tau', default=0.005, type=float) # 目标网络的软更新系数 parser.add_argument('--target_update_interval', default=1, type=int) # 目标网络更新间隔 parser.add_argument('--iteration', default=5, type=int) # 迭代次数 # 学习相关参数 parser.add_argument('--learning_rate', default=3e-4, type=float) # 学习率 parser.add_argument('--gamma', default=0.99, type=int) # 折扣因子,用于奖励的衰减 parser.add_argument('--capacity', default=50000, type=int) # 经验回放缓冲区大小 parser.add_argument('--num_iteration', default=100000, type=int) # 总训练迭代次数 parser.add_argument('--batch_size', default=100, type=int) # 批量大小 parser.add_argument('--seed', default=1, type=int) # 随机种子,确保结果可复现 # 可选参数 parser.add_argument('--num_hidden_layers', default=2, type=int) # 神经网络的隐藏层数 parser.add_argument('--sample_frequency', default=256, type=int) # 采样频率 parser.add_argument('--activation', default='Relu', type=str) # 激活函数类型 parser.add_argument('--render', default=False, type=bool) # 是否显示渲染的界面 parser.add_argument('--log_interval', default=50, type=int) # 日志记录间隔 parser.add_argument('--load', default=False, type=bool) # 是否加载模型 parser.add_argument('--render_interval', default=100, type=int) # 渲染间隔 parser.add_argument('--policy_noise', default=0.2, type=float) # 策略噪声 parser.add_argument('--noise_clip', default=0.5, type=float) # 噪声裁剪范围 parser.add_argument('--policy_delay', default=2, type=int) # 策略更新延迟 parser.add_argument('--exploration_noise', default=0.1, type=float) # 探索噪声 parser.add_argument('--max_episode', default=2000, type=int) # 最大训练轮数 parser.add_argument('--print_log', default=5, type=int) # 打印日志的间隔 args = parser.parse_args() # 解析命令行参数 # 设置随机种子以保证结果可复现 # env.seed(args.seed) # 环境随机种子 # torch.manual_seed(args.seed) # PyTorch随机种子 # np.random.seed(args.seed) # Numpy随机种子 # 获取当前脚本文件名 script_name = os.path.basename(__file__) # 创建Gym环境 env = gym.make(args.env_name) # 提取环境的状态空间和动作空间的维度 state_dim = env.observation_space.shape[0] # 状态空间的维度 action_dim = env.action_space.shape[0] # 动作空间的维度 max_action = float(env.action_space.high[0]) # 动作空间的最大值 min_Val = torch.tensor(1e-7).float().to(device) # 防止数值错误的最小值 # 设置保存模型和日志的目录 directory = './exp' + script_name + args.env_name + './' 

经验回放缓冲区

# 定义经验回放缓冲区类 class Replay_buffer(): ''' 用于存储经验数据: (state, next_state, action, reward, done) ''' def __init__(self, max_size=args.capacity): self.storage = [] # 存储经验的列表 self.max_size = max_size # 最大存储容量 self.ptr = 0 # 指针,用于循环覆盖旧数据 def push(self, data): # 如果缓冲区已满,则覆盖旧数据 if len(self.storage) == self.max_size: self.storage[int(self.ptr)] = data # 覆盖旧数据 self.ptr = (self.ptr + 1) % self.max_size # 更新指针位置 else: self.storage.append(data) # 添加新数据 def sample(self, batch_size): # 随机从缓冲区中抽取一个批次的数据 ind = np.random.randint(0, len(self.storage), size=batch_size) x, y, u, r, d = [], [], [], [], [] # 用于存储抽样结果 # 遍历随机索引并提取对应的数据 for i in ind: X, Y, U, R, D = self.storage[i] x.append(np.array(X, copy=False)) y.append(np.array(Y, copy=False)) u.append(np.array(U, copy=False)) r.append(np.array(R, copy=False)) d.append(np.array(D, copy=False)) # 返回转换为NumPy数组的采样结果 return np.array(x), np.array(y), np.array(u), np.array(r).reshape(-1, 1), np.array(d).reshape(-1, 1)

网络配置

# 定义Actor(策略网络)类 class Actor(nn.Module): def __init__(self, state_dim, action_dim, max_action): super(Actor, self).__init__() # 定义三层全连接层,输入是状态,输出是动作 self.fc1 = nn.Linear(state_dim, 400) # 第一层全连接,400个神经元 self.fc2 = nn.Linear(400, 300) # 第二层全连接,300个神经元 self.fc3 = nn.Linear(300, action_dim) # 输出层,输出维度为动作维度 self.max_action = max_action # 动作的最大值,用于约束输出 def forward(self, state): # 前向传播函数 a = F.relu(self.fc1(state)) # 第一层激活函数ReLU a = F.relu(self.fc2(a)) # 第二层激活函数ReLU a = torch.tanh(self.fc3(a)) * self.max_action # 输出层使用tanh激活并乘以最大动作值 return a # 返回动作值 # 定义Critic(值网络)类 class Critic(nn.Module): def __init__(self, state_dim, action_dim): super(Critic, self).__init__() # 定义三层全连接层,输入是状态和动作的拼接,输出是Q值 self.fc1 = nn.Linear(state_dim + action_dim, 400) # 第一层全连接 self.fc2 = nn.Linear(400, 300) # 第二层全连接 self.fc3 = nn.Linear(300, 1) # 输出层,输出一个Q值 def forward(self, state, action): # 将状态和动作拼接在一起作为输入 state_action = torch.cat([state, action], 1) q = F.relu(self.fc1(state_action)) # 第一层激活函数ReLU q = F.relu(self.fc2(q)) # 第二层激活函数ReLU q = self.fc3(q) # 输出层 return q # 返回Q值

算法逻辑

# 定义TD3算法类 class TD3(): def __init__(self, state_dim, action_dim, max_action): # 初始化Actor网络和目标Actor网络 self.actor = Actor(state_dim, action_dim, max_action).to(device) self.actor_target = Actor(state_dim, action_dim, max_action).to(device) # 初始化两个Critic网络及其目标网络 self.critic_1 = Critic(state_dim, action_dim).to(device) self.critic_1_target = Critic(state_dim, action_dim).to(device) self.critic_2 = Critic(state_dim, action_dim).to(device) self.critic_2_target = Critic(state_dim, action_dim).to(device) # 定义优化器 self.actor_optimizer = optim.Adam(self.actor.parameters()) # Actor网络的优化器 self.critic_1_optimizer = optim.Adam(self.critic_1.parameters()) # Critic 1网络的优化器 self.critic_2_optimizer = optim.Adam(self.critic_2.parameters()) # Critic 2网络的优化器 # 将目标网络的参数初始化为与主网络相同 self.actor_target.load_state_dict(self.actor.state_dict()) self.critic_1_target.load_state_dict(self.critic_1.state_dict()) self.critic_2_target.load_state_dict(self.critic_2.state_dict()) self.max_action = max_action # 最大动作值 self.memory = Replay_buffer(args.capacity) # 初始化经验回放缓冲区 self.writer = SummaryWriter(directory) # 初始化TensorBoard记录器 self.num_critic_update_iteration = 0 # Critic更新次数计数 self.num_actor_update_iteration = 0 # Actor更新次数计数 self.num_training = 0 # 总训练次数计数 def select_action(self, state): # 根据当前策略选择动作 state = torch.tensor(state.reshape(1, -1)).float().to(device) # 将状态转换为张量 return self.actor(state).cpu().data.numpy().flatten() # 通过Actor网络生成动作 def update(self, num_iteration): # 更新网络,训练过程 if self.num_training % 500 == 0: # 每500次训练打印日志 print("====================================") print("model has been trained for {} times...".format(self.num_training)) print("====================================") for i in range(num_iteration): # 迭代更新 x, y, u, r, d = self.memory.sample(args.batch_size) # 从经验回放缓冲区中采样 state = torch.FloatTensor(x).to(device) # 转换采样的状态为张量 action = torch.FloatTensor(u).to(device) # 转换采样的动作为张量 next_state = torch.FloatTensor(y).to(device) # 转换采样的下一状态为张量 done = torch.FloatTensor(d).to(device) # 转换采样的完成标志为张量 reward = torch.FloatTensor(r).to(device) # 转换采样的奖励为张量 # 选择目标动作,并加入噪声 noise = torch.ones_like(action).data.normal_(0, args.policy_noise).to(device) noise = noise.clamp(-args.noise_clip, args.noise_clip) # 裁剪噪声 next_action = (self.actor_target(next_state) + noise).clamp(-self.max_action, self.max_action) # 计算目标Q值 target_Q1 = self.critic_1_target(next_state, next_action) target_Q2 = self.critic_2_target(next_state, next_action) target_Q = torch.min(target_Q1, target_Q2) # 取两个Critic网络中最小的Q值 target_Q = reward + ((1 - done) * args.gamma * target_Q).detach() # Bellman公式计算目标值 # 优化Critic 1网络 current_Q1 = self.critic_1(state, action) # 当前Q值 loss_Q1 = F.mse_loss(current_Q1, target_Q) # 均方误差损失 self.critic_1_optimizer.zero_grad() # 清除梯度 loss_Q1.backward() # 反向传播 self.critic_1_optimizer.step() # 更新参数 # 优化Critic 2网络 current_Q2 = self.critic_2(state, action) loss_Q2 = F.mse_loss(current_Q2, target_Q) self.critic_2_optimizer.zero_grad() loss_Q2.backward() self.critic_2_optimizer.step() # 延迟更新策略网络和目标网络 if i % args.policy_delay == 0: # 计算Actor损失 actor_loss = - self.critic_1(state, self.actor(state)).mean() # 最大化Q值 self.actor_optimizer.zero_grad() # 清除梯度 actor_loss.backward() # 反向传播 self.actor_optimizer.step() # 更新Actor网络 # 更新目标网络参数 for param, target_param in zip(self.actor.parameters(), self.actor_target.parameters()): target_param.data.copy_((1 - args.tau) * target_param.data + args.tau * param.data) for param, target_param in zip(self.critic_1.parameters(), self.critic_1_target.parameters()): target_param.data.copy_((1 - args.tau) * target_param.data + args.tau * param.data) for param, target_param in zip(self.critic_2.parameters(), self.critic_2_target.parameters()): target_param.data.copy_((1 - args.tau) * target_param.data + args.tau * param.data) self.num_actor_update_iteration += 1 # 更新计数 self.num_critic_update_iteration += 1 # Critic更新计数 self.num_training += 1 # 总训练次数计数

模型保存与加载

 def save(self): # 保存模型的参数到指定目录 torch.save(self.actor.state_dict(), directory + 'actor.pth') # 保存Actor网络 torch.save(self.actor_target.state_dict(), directory + 'actor_target.pth') # 保存目标Actor网络 torch.save(self.critic_1.state_dict(), directory + 'critic_1.pth') # 保存Critic 1网络 torch.save(self.critic_1_target.state_dict(), directory + 'critic_1_target.pth') # 保存目标Critic 1网络 torch.save(self.critic_2.state_dict(), directory + 'critic_2.pth') # 保存Critic 2网络 torch.save(self.critic_2_target.state_dict(), directory + 'critic_2_target.pth') # 保存目标Critic 2网络 print("====================================") print("Model has been saved...") # 打印保存完成日志 print("====================================") def load(self): # 加载保存的模型参数 self.actor.load_state_dict(torch.load(directory + 'actor.pth')) # 加载Actor网络 self.actor_target.load_state_dict(torch.load(directory + 'actor_target.pth')) # 加载目标Actor网络 self.critic_1.load_state_dict(torch.load(directory + 'critic_1.pth')) # 加载Critic 1网络 self.critic_1_target.load_state_dict(torch.load(directory + 'critic_1_target.pth')) # 加载目标Critic 1网络 self.critic_2.load_state_dict(torch.load(directory + 'critic_2.pth')) # 加载Critic 2网络 self.critic_2_target.load_state_dict(torch.load(directory + 'critic_2_target.pth')) # 加载目标Critic 2网络 print("====================================") print("Model has been loaded...") # 打印加载完成日志 print("====================================")

主程序入口

 # 主程序入口 if __name__ == '__main__': # 初始化TD3智能体 agent = TD3(state_dim, action_dim, max_action) ep_r = 0 # 累计奖励初始化为0 if args.mode == 'test': # 如果模式为测试 agent.load() # 加载模型 for i in range(args.iteration): # 测试运行指定迭代次数 state = env.reset() # 环境重置,获取初始状态 for t in count(): # 无限循环,直到完成或达到限制 action = agent.select_action(state) # 使用智能体选择动作 next_state, reward, done, info = env.step(np.float32(action)) # 执行动作,获取下一状态和奖励 ep_r += reward # 累计奖励 env.render() # 渲染环境(显示界面) if done or t == 2000: # 如果完成或者达到最大步数 print("Ep_i \t{}, the ep_r is \t{:0.2f}, the step is \t{}".format(i, ep_r, t)) # 打印日志 break # 结束当前测试 state = next_state # 更新当前状态 elif args.mode == 'train': # 如果模式为训练 print("====================================") print("Collection Experience...") # 收集经验 print("====================================") if args.load: agent.load() # 如果需要,加载模型 for i in range(args.num_iteration): # 训练运行指定迭代次数 state = env.reset() # 环境重置 for t in range(2000): # 每次训练的最大步数 # 使用智能体选择动作,并加入探索噪声 action = agent.select_action(state) action = action + np.random.normal(0, args.exploration_noise, size=env.action_space.shape[0]) action = action.clip(env.action_space.low, env.action_space.high) # 限制动作范围 next_state, reward, done, info = env.step(action) # 执行动作 ep_r += reward # 累计奖励 # 如果需要渲染且达到渲染间隔,显示环境 if args.render and i >= args.render_interval: env.render() # 将当前状态、下一状态、动作、奖励、完成标志存入经验回放缓冲区 agent.memory.push((state, next_state, action, reward, np.float(done))) # 打印内存大小日志 if (i + 1) % 10 == 0: print('Episode {}, The memory size is {} '.format(i, len(agent.memory.storage))) # 如果经验回放缓冲区已满,则更新模型 if len(agent.memory.storage) >= args.capacity - 1: agent.update(10) # 每次更新10步 state = next_state # 更新当前状态 if done or t == args.max_episode - 1: # 如果完成或达到最大轮数 agent.writer.add_scalar('ep_r', ep_r, global_step=i) # 记录奖励到日志 if i % args.print_log == 0: print("Ep_i \t{}, the ep_r is \t{:0.2f}, the step is \t{}".format(i, ep_r, t)) # 打印奖励日志 ep_r = 0 # 重置累计奖励 break # 跳出当前循环 # 每隔一定间隔保存模型 if i % args.log_interval == 0: agent.save() else: raise NameError("mode wrong!!!") # 如果模式错误,抛出异常


[Notice]  主要模块及功能

  1. 命令行参数解析
    使用 argparse 模块定义和解析运行时的参数配置,例如模式选择(训练/测试)、环境名称、学习率、经验缓冲区大小等。
  2. 环境初始化
    使用 OpenAI Gym 创建强化学习环境,并提取状态空间和动作空间的维度,以及动作的上下限。
  3. 经验回放缓冲区
    定义 Replay_buffer 类,用于存储状态、动作、奖励、下一状态等经验数据,并支持随机采样,为批量训练提供数据。
  4. 神经网络定义
    • Actor:生成动作的策略网络,使用 ReLU 和 Tanh 激活函数。
    • Critic:评估动作的价值(Q值)的网络,输入状态和动作的拼接数据,输出Q值。
  5. TD3算法逻辑
    • 目标网络(Target Network):用于稳定训练,参数以软更新方式与主网络保持同步。
    • 延迟更新策略(Delayed Policy Update):降低Actor的更新频率,确保Critic训练充分,避免不稳定。
    • 双Critic网络:缓解Q值的高估偏差问题,通过取两个Critic网络输出的最小值作为目标Q值。
  6. 训练和测试

                在训练模式下,智能体通过环境交互收集经验并更新网络权重。

                在测试模式下,使用训练好的模型直接与环境交互,评估性能。

    7.模型保存与加载

                支持保存和加载模型权重,便于训练中断后继续,或在测试中复用。

​# 环境配置 Python 3.11.5 torch 2.1.0 torchvision 0.16.0 gym 0.26.2
        由于博文主要为了介绍相关算法的原理和应用的方法,缺乏对于实际效果的关注,算法可能在上述环境中的效果不佳或者无法运行,一是算法不适配上述环境,二是算法未调参和优化,三是没有呈现完整的代码,四是等等。上述代码用于了解和学习算法足够了,但若是想直接将上面代码应用于实际项目中,还需要进行修改。

五、TD3的优势

  1. 降低Q值高估偏差:双Critic网络的最小值策略有效减少了偏差。
  2. 增强训练稳定性:延迟更新减少了网络间的干扰。
  3. 适应复杂环境:目标动作平滑提高了鲁棒性。

六、总结

        TD3不仅改进了DDPG的不足,还为强化学习的稳定性研究提供了重要的理论和实践参考。其成功之处在于:

  • 克服了Q值过估计问题,使得训练过程更加稳定;
  • 提升了策略更新的鲁棒性,能更高效地探索动作空间。

        作为一个里程碑式的算法,TD3推动了连续动作空间强化学习的发展,为后续算法(如SAC、PPO等)提供了宝贵的启发。

参考文献:Addressing Function Approximation Error in Actor-Critic Methods

 更多强化学习文章,请前往:【强化学习(RL)】专栏

        博客都是给自己看的笔记,如有误导深表抱歉。文章若有不当和不正确之处,还望理解与指出。由于部分文字、图片等来源于互联网,无法核实真实出处,如涉及相关争议,请联系博主删除。如有错误、疑问和侵权,欢迎评论留言联系作者,或者添加VX:Rainbook_2,联系作者。✨

Read more

《C++ 基础进阶:内存开辟规则、类型转换原理与 IO 流高效使用》

《C++ 基础进阶:内存开辟规则、类型转换原理与 IO 流高效使用》

前引:在 C++ 编程中,内存管理是程序稳定性与性能的基石,而类型转换与 IO 流则是数据处理和交互的核心工具。栈与堆作为内存分配的两大核心区域,其开辟方式直接决定了变量的生命周期、访问效率及内存安全 —— 错误的分配策略可能导致内存泄漏、野指针或栈溢出等致命问题。与此同时,类型转换的合理性关乎类型系统的严谨性,不当转换易引发数据截断、逻辑错误;IO 流作为数据输入输出的桥梁,其正确使用则直接影响程序与外部设备(如控制台、文件)交互的可靠性! 目录 【一】内存完美开辟 (1)栈和堆的本质区别 (2)如何只在栈上开辟空间 (3)如何只在堆上开辟空间 【二】C++的四种类型转换 (1)static_cast (2)reinterpret_cast (3)const_cast (4)dynamic_cast 【三】operator类型转换 (1)

By Ne0inhk
华为OD技术面八股文真题_C++_3

华为OD技术面八股文真题_C++_3

文章目录 * 变量的声明和定义的区别 * 内存泄露是什么意思?怎么避免内存泄露 * 怎么排查内存泄漏,遇到内存泄漏情况,一般怎么解决 * 说一下define和const的区别 * define和typedef的区别 * 宏函数和内联函数的区别 * 类和结构体的区别 * 结构体(struct)和联合体(union)差别 * 静态库和动态库区别 * 介绍一下C++的编译过程 变量的声明和定义的区别 * 变量的声明是告诉编译器变量的名称和类型,不分配存储空间; * 变量的定义会为变量分配存储空间并建立实体。 * 一个变量可以在多个地方声明,但只能在一个地方定义。 使用 extern 修饰的变量通常是声明,表示该变量在其它文件中定义,但 如果 extern 变量带初始化,则该语句仍然属于定义。 内存泄露是什么意思?怎么避免内存泄露 内存泄漏是指程序在动态申请内存后,后续失去对该内存的控制,导致这块内存无法被释放,从而造成内存资源浪费的现象。内存被申请了,却释放不了。 内存泄漏的危害如下: 1. 程序内存占用不断增大,导致系统可用内存减少,性能下

By Ne0inhk

C++:实现DLL注入(附带源码)

一、项目背景详细介绍 在 Windows 平台下,DLL(Dynamic Link Library)注入是一项非常经典且重要的技术。它的本质是: 让目标进程在不知情或未主动加载的情况下,加载并执行我们指定的 DLL DLL 注入在多个领域都有实际应用: 1. 常见应用场景 (1)调试与开发辅助 * 给第三方程序注入日志模块 * Hook API 以观察行为 * 插桩统计函数调用情况 (2)自动化与工具开发 * 游戏辅助工具 * 自动化测试框架 * GUI 注入扩展 (3)安全研究与逆向工程 * API Hook 技术的基础 * 恶意代码分析 * 防注入 / 反调试技术研究 (4)系统级功能扩展 * 无源码扩展已有程序功能 * 动态替换实现逻辑 2. 为什么学习 DLL 注入? DLL 注入几乎是 Windows

By Ne0inhk
【C++】priority_queue和deque的使用与实现

【C++】priority_queue和deque的使用与实现

priority_queue与deque的使用与模拟实现 ✨前言:在C++ STL中,priority_queue和deque是两个重要的容器适配器,它们分别基于堆和双端队列的概念,为不同的应用场景提供了高效的解决方案。本文将深入探讨它们的使用方法、底层实现原理以及在实际开发中的应用选择。 📖专栏:【C++成长之旅】 目录 * priority_queue与deque的使用与模拟实现 * 一、priority_queue * 1.1 介绍 * 1.2 使用 * 1.3 模拟实现 * 二、deque * 2.1 介绍 * 2.2 缺陷 * 三、STL标准库中对于stack和queue的模拟实现 * 3.1 为什么选择deque作为stack和queue的底层默认容器 * 3.2 stack的模拟实现 * 3.3 queue的模拟实现 一、priority_

By Ne0inhk