📢本篇文章是博主强化学习(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]

其中:

  • \pi_\phi(s) 是Actor的策略函数。
  • Q^\pi(s, a) 是Critic估计的动作值函数。
  • \rho^\pi是由策略生成的状态分布。
(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'))

  • \theta'\phi'是Critic和Actor的目标网络参数。
  • \mathcal{D} 是经验回放池中采样的数据。

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. 构建经验回放池\mathcal{D},用于存储交互数据;
  3. 与环境交互,使用当前策略 \pi_\phi执行动作 a_t,存储(s_t, a_t, r_t, s_{t+1})
  4. \mathcal{D}中随机抽取一个批量数据(s, a, r, s')
    • 更新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]
    • 延迟更新Actor:每隔 d 步,更新Actor策略:L(\phi) = -\mathbb{E}{s} \big[ Q{\theta_1}(s, \pi_\phi(s)) \big]
    • 目标网络软更新\theta_i' \leftarrow \tau \theta_i + (1 - \tau) \theta_i' ] [ \phi' \leftarrow \tau \phi + (1 - \tau) \phi'
  5. 重复以上步骤直到收敛。


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,联系作者。✨

Logo

纵情码海钱塘涌,杭州开发者创新动! 属于杭州的开发者社区!致力于为杭州地区的开发者提供学习、合作和成长的机会;同时也为企业交流招聘提供舞台!

更多推荐