DDPG算法实战:用Python从零搭建连续动作空间强化学习模型(附完整代码)

如果你已经玩过一些强化学习的经典环境,比如让小车爬坡的CartPole,或者用Q-learning在网格世界里寻宝,你可能会觉得这些离散动作的任务还挺有意思。但当你真正想把手伸向更“真实”的世界时,比如训练一个机械臂抓取物体,或者让一个虚拟角色学习行走,一个巨大的挑战就摆在了面前:动作空间是连续的。你没法再用“上、下、左、右”几个简单的指令,而是要输出一个精确的力或角度值,比如“施加0.75牛顿的力”或“转动关节15.3度”。这时,像DQN这类基于离散动作的算法就束手无策了。

这正是深度确定性策略梯度算法(DDPG)大显身手的地方。它巧妙地将深度Q网络的思想与策略梯度方法结合,专为连续动作空间而生。今天,我们就抛开复杂的理论推导,直接进入实战。我会带你从零开始,用Python和PyTorch一步步搭建一个完整的DDPG模型,并用它来解决经典的“倒立摆”控制问题。整个过程就像搭积木,我们会亲手构建Actor和Critic网络,设计经验回放机制,实现OU噪声探索,并最终看到智能体从零开始学会平衡倒立摆。无论你是想快速上手项目,还是希望深入理解算法细节,这篇手把手的指南都将为你提供清晰的路径和可运行的代码。

1. 环境搭建与核心概念解析

在动手写代码之前,我们需要先理解DDPG算法的几个核心思想,并准备好开发环境。DDPG常被称为“深度Q网络在连续动作空间的扩展”,但它其实融合了Actor-Critic框架的精华。

Actor-Critic架构的连续化改造:传统的Actor-Critic算法中,Actor输出的是一个动作的概率分布(例如,高斯分布的均值和方差),然后从这个分布中采样得到动作。DDPG做了一个关键的简化:它的Actor网络直接输出一个确定的动作值。你可以把它想象成一个“决策者”,它观察环境状态,然后直接“命令”执行某个具体的动作。而Critic网络则扮演“评论家”的角色,它接收状态和Actor做出的动作,评估这个“状态-动作对”的好坏,给出一个Q值(预期累积回报)。Actor的目标就是根据Critic的“评分”,调整自己的决策,以输出能获得更高Q值的动作。

注意:这种确定性策略意味着,在相同的状态下,Actor总会输出相同的动作。这有利于高效学习,但也可能陷入局部最优。为了鼓励探索,DDPG在Actor输出的动作上添加了专门的噪声。

为了稳定训练,DDPG从DQN那里借鉴了两个至关重要的技术:经验回放目标网络。经验回放池就像一个记忆库,智能体与环境交互产生的经验(状态、动作、奖励、下一个状态)被存储起来,训练时从中随机抽取一批样本。这打破了数据间的时序相关性,让学习过程更稳定。目标网络则是主网络的“慢速拷贝”,用于计算稳定的目标Q值,防止因网络频繁更新而导致的训练振荡。

我们的实战环境选择Gymnasium(OpenAI Gym的维护分支)中的Pendulum-v1。这个环境的目标是施加扭矩,让倒立摆保持竖直向上。状态空间是三维的(摆角的余弦值、正弦值、角速度),动作空间是一维的连续值(扭矩,范围[-2.0, 2.0])。奖励是负的,越接近竖直状态且消耗的扭矩越小,奖励越高(越接近0)。

首先,我们搭建基础环境:

# 创建项目目录并安装核心依赖
pip install gymnasium torch numpy matplotlib

接下来,我们初始化一个简单的环境来验证:

import gymnasium as gym
import numpy as np

env = gym.make('Pendulum-v1', render_mode='human')
state, _ = env.reset()
print(f"状态空间形状: {env.observation_space.shape}")
print(f"动作空间形状: {env.action_space.shape}")
print(f"动作范围: [{env.action_space.low}, {env.action_space.high}]")
print(f"初始状态示例: {state}")

# 随机动作测试
for _ in range(50):
    action = env.action_space.sample()  # 随机采样一个动作
    state, reward, terminated, truncated, info = env.step(action)
    print(f"动作: {action:.3f}, 奖励: {reward:.3f}, 新状态: {state}")
    if terminated or truncated:
        break
env.close()

运行这段代码,你会看到一个倒立摆窗口,并能在终端看到状态和动作信息。这确认了我们的环境已就绪。接下来,我们将进入核心部分:构建DDPG的神经网络组件。

2. 构建神经网络:Actor与Critic

DDPG的核心是两个深度神经网络:Actor和Critic。我们将使用PyTorch来定义它们。为了让网络更容易训练,这里采用了一些小技巧,比如使用ReLU激活函数增加非线性,对最后一层权重进行小范围初始化以防止初始输出过大。

Actor网络:它的输入是环境状态(例如Pendulum-v1的3维状态),经过几层全连接网络后,输出一个确定性的动作。由于环境对动作有范围限制(如[-2, 2]),我们通常在网络的最后一层使用tanh激活函数,将输出压缩到[-1, 1]区间,然后在外部根据环境的实际范围进行缩放。

import torch
import torch.nn as nn
import torch.nn.functional as F

class Actor(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim=256, init_w=3e-3):
        """
        确定性策略网络(Actor)
        Args:
            state_dim (int): 状态维度
            action_dim (int): 动作维度
            hidden_dim (int): 隐藏层神经元数量
            init_w (float): 最后一层权重初始化的范围
        """
        super(Actor, self).__init__()
        self.linear1 = nn.Linear(state_dim, hidden_dim)
        self.linear2 = nn.Linear(hidden_dim, hidden_dim)
        self.linear3 = nn.Linear(hidden_dim, action_dim)

        # 对输出层进行小范围初始化,使初始动作接近零
        self.linear3.weight.data.uniform_(-init_w, init_w)
        self.linear3.bias.data.uniform_(-init_w, init_w)

    def forward(self, state):
        x = F.relu(self.linear1(state))
        x = F.relu(self.linear2(x))
        # tanh将输出限制在[-1, 1],便于后续映射到实际动作范围
        action = torch.tanh(self.linear3(x))
        return action

Critic网络:它的输入是状态和动作的拼接。这一点与某些只接收状态的Critic不同,因为Q值函数本身就是状态和动作的函数。网络输出一个标量,即该状态-动作对的Q值估计。

class Critic(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim=256, init_w=3e-3):
        """
        Q值网络(Critic)
        Args:
            state_dim (int): 状态维度
            action_dim (int): 动作维度
            hidden_dim (int): 隐藏层神经元数量
            init_w (float): 最后一层权重初始化的范围
        """
        super(Critic, self).__init__()
        # 第一层同时接收状态和动作
        self.linear1 = nn.Linear(state_dim + action_dim, hidden_dim)
        self.linear2 = nn.Linear(hidden_dim, hidden_dim)
        self.linear3 = nn.Linear(hidden_dim, 1)

        # 对输出层进行小范围初始化
        self.linear3.weight.data.uniform_(-init_w, init_w)
        self.linear3.bias.data.uniform_(-init_w, init_w)

    def forward(self, state, action):
        # 将状态和动作在特征维度上拼接
        x = torch.cat([state, action], 1)
        x = F.relu(self.linear1(x))
        x = F.relu(self.linear2(x))
        q_value = self.linear3(x)
        return q_value

这两个网络结构并不复杂,但它们是DDPG算法的基石。Actor负责“做决策”,Critic负责“评价决策”。在训练过程中,Critic会通过学习来更准确地预测Q值,而Actor则根据Critic的指引,调整自己的参数以输出能获得更高Q值的动作。为了促进探索,我们还需要一个特殊的“噪声生成器”。

3. 探索策略:Ornstein-Uhlenbeck噪声

在离散动作空间中,我们常用ε-greedy策略进行探索(以ε概率随机选择动作)。但在连续动作空间,简单的均匀随机噪声可能效率不高,因为它缺乏时间相关性,导致动作抖动剧烈。DDPG论文中推荐使用Ornstein-Uhlenbeck过程来生成噪声。OU噪声具有“均值回归”特性,你可以把它想象成一根橡皮筋:当你施加一个随机扰动后,它会倾向于回到某个均值位置。这种噪声产生的动作序列更平滑,更适合物理控制任务。

OU噪声的数学表达式是一个随机微分方程: dx_t = θ * (μ - x_t) * dt + σ * dW_t 其中x_t是当前噪声值,μ是均值(通常设为0),θ控制回归到均值的速度,σ是噪声的波动幅度,dW_t是维纳过程(可理解为高斯噪声)。

下面我们实现一个OU噪声类,它会在每一步生成相关的噪声,并叠加到Actor输出的确定性动作上。

import numpy as np

class OUNoise:
    def __init__(self, action_dim, mu=0.0, theta=0.15, sigma=0.2):
        """
        Ornstein-Uhlenbeck过程噪声
        Args:
            action_dim (int): 动作维度
            mu (float): 长期均值
            theta (float): 回归速度参数
            sigma (float): 波动率参数
        """
        self.mu = mu * np.ones(action_dim)
        self.theta = theta
        self.sigma = sigma
        self.action_dim = action_dim
        self.reset()

    def reset(self):
        """重置噪声过程"""
        self.state = np.copy(self.mu)

    def sample(self):
        """采样一个噪声向量"""
        dx = self.theta * (self.mu - self.state)
        dx += self.sigma * np.random.randn(self.action_dim)
        self.state += dx
        return self.state

在实际使用时,我们这样将噪声添加到动作上:

# 假设actor是已定义的网络,state是当前状态
deterministic_action = actor(state)  # 形状为 [batch_size, action_dim]
noise = ou_noise.sample()  # 形状为 [action_dim]
# 将噪声缩放到一个较小的范围,例如[-0.1, 0.1],再叠加到动作上
noisy_action = deterministic_action + 0.1 * noise
# 最后确保动作不超出环境允许的范围
clipped_action = np.clip(noisy_action, env.action_space.low, env.action_space.high)

OU噪声的参数(thetasigma)需要根据具体环境调整。对于Pendulum-v1theta=0.15sigma=0.2是常见的起点。在训练初期,我们可以使用较大的噪声鼓励探索,随着训练进行,可以逐渐减小噪声幅度(衰减),让策略更倾向于利用已学到的知识。有了探索机制,我们还需要一个记忆库来存储经验。

4. 经验回放与智能体整合

经验回放是稳定深度强化学习训练的利器。它的核心思想是将智能体与环境交互得到的转移样本(state, action, reward, next_state, done)存储在一个固定大小的缓冲区(回放池)中。训练时,随机从池中抽取一小批样本,用于更新网络。这样做有两个好处:一是打破了样本间的时序相关性,二是提高了样本的利用率。

我们实现一个简单的基于双端队列的经验回放池:

import random
from collections import deque

class ReplayBuffer:
    def __init__(self, capacity):
        self.buffer = deque(maxlen=capacity)

    def push(self, transition):
        """存储一条经验"""
        self.buffer.append(transition)

    def sample(self, batch_size):
        """随机采样一批经验"""
        batch = random.sample(self.buffer, batch_size)
        # 解压,将列表的列表转换为多个列表
        state, action, reward, next_state, done = zip(*batch)
        return (
            np.array(state),
            np.array(action),
            np.array(reward, dtype=np.float32),
            np.array(next_state),
            np.array(done, dtype=np.float32)
        )

    def __len__(self):
        return len(self.buffer)

现在,我们将Actor、Critic、OU噪声和经验回放池整合到DDPG智能体类中。这个类将封装智能体的所有行为:根据状态选择动作(探索/利用)、存储经验、更新网络参数。

class DDPGAgent:
    def __init__(self, state_dim, action_dim, action_high, action_low,
                 actor_lr=1e-4, critic_lr=1e-3, gamma=0.99, tau=0.005,
                 buffer_capacity=100000, batch_size=128):
        """
        DDPG智能体
        Args:
            state_dim, action_dim: 状态和动作维度
            action_high, action_low: 动作空间的上界和下界(用于缩放)
            actor_lr, critic_lr: 学习率
            gamma: 折扣因子
            tau: 目标网络软更新参数
            buffer_capacity: 经验回放池容量
            batch_size: 训练批大小
        """
        self.state_dim = state_dim
        self.action_dim = action_dim
        self.action_high = action_high
        self.action_low = action_low
        self.gamma = gamma
        self.tau = tau
        self.batch_size = batch_size

        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

        # 初始化四个网络
        self.actor = Actor(state_dim, action_dim).to(self.device)
        self.actor_target = Actor(state_dim, action_dim).to(self.device)
        self.critic = Critic(state_dim, action_dim).to(self.device)
        self.critic_target = Critic(state_dim, action_dim).to(self.device)

        # 硬拷贝参数初始化目标网络
        self.actor_target.load_state_dict(self.actor.state_dict())
        self.critic_target.load_state_dict(self.critic.state_dict())

        # 优化器
        self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=actor_lr)
        self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=critic_lr)

        # 经验回放和噪声
        self.memory = ReplayBuffer(buffer_capacity)
        self.noise = OUNoise(action_dim)

        # 用于将网络输出的[-1,1]映射到实际动作范围
        self.action_scale = torch.FloatTensor((action_high - action_low) / 2.).to(self.device)
        self.action_bias = torch.FloatTensor((action_high + action_low) / 2.).to(self.device)

    def select_action(self, state, add_noise=True):
        """根据状态选择动作(用于交互)"""
        state = torch.FloatTensor(state).unsqueeze(0).to(self.device)
        with torch.no_grad():
            action_tanh = self.actor(state)  # 输出在[-1, 1]
        # 映射到实际动作范围
        action = self.action_scale * action_tanh + self.action_bias
        action = action.cpu().numpy()[0]

        if add_noise:
            noise = self.noise.sample()
            action += noise
            # 裁剪到合法范围
            action = np.clip(action, self.action_low, self.action_high)
        return action

    def push_transition(self, *transition):
        self.memory.push(transition)

    def update(self):
        """执行一次网络更新"""
        if len(self.memory) < self.batch_size:
            return

        # 1. 从回放池采样
        states, actions, rewards, next_states, dones = self.memory.sample(self.batch_size)
        states = torch.FloatTensor(states).to(self.device)
        actions = torch.FloatTensor(actions).to(self.device)
        rewards = torch.FloatTensor(rewards).unsqueeze(1).to(self.device)
        next_states = torch.FloatTensor(next_states).to(self.device)
        dones = torch.FloatTensor(dones).unsqueeze(1).to(self.device)

        # 2. 更新Critic网络
        with torch.no_grad():
            next_actions = self.actor_target(next_states)
            next_actions = self.action_scale * next_actions + self.action_bias
            target_q = self.critic_target(next_states, next_actions)
            target_q = rewards + (1 - dones) * self.gamma * target_q

        current_q = self.critic(states, actions)
        critic_loss = F.mse_loss(current_q, target_q)

        self.critic_optimizer.zero_grad()
        critic_loss.backward()
        # 可选:梯度裁剪,防止梯度爆炸
        torch.nn.utils.clip_grad_norm_(self.critic.parameters(), max_norm=1.0)
        self.critic_optimizer.step()

        # 3. 更新Actor网络
        # 策略梯度:最大化Q值
        actor_actions = self.actor(states)
        actor_actions_scaled = self.action_scale * actor_actions + self.action_bias
        actor_loss = -self.critic(states, actor_actions_scaled).mean()

        self.actor_optimizer.zero_grad()
        actor_loss.backward()
        torch.nn.utils.clip_grad_norm_(self.actor.parameters(), max_norm=1.0)
        self.actor_optimizer.step()

        # 4. 软更新目标网络
        for target_param, param in zip(self.actor_target.parameters(), self.actor.parameters()):
            target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)
        for target_param, param in zip(self.critic_target.parameters(), self.critic.parameters()):
            target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)

        return critic_loss.item(), actor_loss.item()

这个DDPGAgent类包含了算法的所有核心逻辑。select_action方法用于与环境交互,可以选择是否添加OU噪声(训练时添加,测试时不添加)。update方法是训练的核心,它遵循以下步骤:

  1. 采样一批经验。
  2. 用目标网络计算目标Q值(target_q),并计算Critic的损失(均方误差)。
  3. 更新Critic网络,使其预测更准确。
  4. 计算Actor的损失:我们希望Actor输出的动作能使Critic给出的Q值最大,因此损失是负的Q值均值。
  5. 更新Actor网络。
  6. 以软更新的方式缓慢更新目标网络参数,保持训练稳定性。

至此,DDPG的所有核心模块都已就绪。接下来,我们将它们串联起来,进行完整的训练循环。

5. 训练循环、可视化与进阶技巧

有了智能体,我们需要一个训练循环来驱动整个学习过程。这个循环会迭代多个回合,在每个回合中,智能体与环境交互,收集经验,并定期更新网络参数。同时,我们还需要记录训练过程中的奖励,以便观察学习进展。

下面是一个完整的训练函数,它整合了环境交互、经验存储、网络更新和进度记录:

def train_ddpg(env, agent, episodes=500, max_steps=200,
               warmup_steps=1000, update_interval=50):
    """
    训练DDPG智能体
    Args:
        env: Gymnasium环境
        agent: DDPGAgent实例
        episodes: 训练总回合数
        max_steps: 每个回合最大步数
        warmup_steps: 预热步数,在开始更新网络前先随机探索收集经验
        update_interval: 每交互多少步更新一次网络
    """
    total_steps = 0
    episode_rewards = []
    critic_losses = []
    actor_losses = []

    for ep in range(episodes):
        state, _ = env.reset()
        agent.noise.reset()  # 每回合重置噪声过程
        episode_reward = 0
        ep_critic_loss = []
        ep_actor_loss = []

        for step in range(max_steps):
            total_steps += 1

            # 在预热阶段,使用随机动作充分探索
            if total_steps < warmup_steps:
                action = env.action_space.sample()
            else:
                action = agent.select_action(state, add_noise=True)

            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated

            # 存储经验
            agent.push_transition(state, action, reward, next_state, done)

            state = next_state
            episode_reward += reward

            # 定期更新网络(预热期过后)
            if total_steps >= warmup_steps and total_steps % update_interval == 0:
                for _ in range(update_interval):  # 可以更新多次
                    cl, al = agent.update()
                    if cl is not None:
                        ep_critic_loss.append(cl)
                        ep_actor_loss.append(al)

            if done:
                break

        # 记录本回合数据
        episode_rewards.append(episode_reward)
        avg_critic_loss = np.mean(ep_critic_loss) if ep_critic_loss else 0
        avg_actor_loss = np.mean(ep_actor_loss) if ep_actor_loss else 0
        critic_losses.append(avg_critic_loss)
        actor_losses.append(avg_actor_loss)

        if (ep + 1) % 10 == 0:
            print(f"Episode {ep+1:4d} | "
                  f"Total Steps: {total_steps:6d} | "
                  f"Reward: {episode_reward:7.2f} | "
                  f"Avg Critic Loss: {avg_critic_loss:.4f} | "
                  f"Avg Actor Loss: {avg_actor_loss:.4f}")

    return {
        'episode_rewards': episode_rewards,
        'critic_losses': critic_losses,
        'actor_losses': actor_losses
    }

训练完成后,我们需要可视化结果来评估性能。通常我们会绘制回合奖励随训练回合的变化曲线,以及Critic和Actor的损失曲线。

import matplotlib.pyplot as plt

def plot_training_results(results, window=10):
    """
    绘制训练结果
    Args:
        results: 包含'rewards', 'critic_losses', 'actor_losses'的字典
        window: 滑动平均的窗口大小
    """
    rewards = results['episode_rewards']
    critic_loss = results['critic_losses']
    actor_loss = results['actor_losses']

    fig, axes = plt.subplots(1, 3, figsize=(15, 4))

    # 奖励曲线
    axes[0].plot(rewards, alpha=0.6, label='Raw')
    # 计算滑动平均,使曲线更平滑
    if len(rewards) >= window:
        moving_avg = np.convolve(rewards, np.ones(window)/window, mode='valid')
        axes[0].plot(range(window-1, len(rewards)), moving_avg, 'r-', label=f'MA({window})')
    axes[0].set_xlabel('Episode')
    axes[0].set_ylabel('Total Reward')
    axes[0].set_title('Training Rewards')
    axes[0].legend()
    axes[0].grid(True, alpha=0.3)

    # Critic损失曲线
    axes[1].plot(critic_loss)
    axes[1].set_xlabel('Episode')
    axes[1].set_ylabel('Loss')
    axes[1].set_title('Critic Network Loss')
    axes[1].grid(True, alpha=0.3)

    # Actor损失曲线
    axes[2].plot(actor_loss)
    axes[2].set_xlabel('Episode')
    axes[2].set_ylabel('Loss')
    axes[2].set_title('Actor Network Loss')
    axes[2].grid(True, alpha=0.3)

    plt.tight_layout()
    plt.show()

现在,让我们把所有部分组合起来,启动训练:

# 主程序
if __name__ == "__main__":
    # 创建环境
    env = gym.make('Pendulum-v1')
    state_dim = env.observation_space.shape[0]
    action_dim = env.action_space.shape[0]
    action_high = env.action_space.high[0]
    action_low = env.action_space.low[0]

    print(f"状态维度: {state_dim}, 动作维度: {action_dim}")
    print(f"动作范围: [{action_low}, {action_high}]")

    # 初始化智能体
    agent = DDPGAgent(
        state_dim=state_dim,
        action_dim=action_dim,
        action_high=action_high,
        action_low=action_low,
        actor_lr=1e-4,
        critic_lr=1e-3,
        gamma=0.99,
        tau=0.005,
        buffer_capacity=100000,
        batch_size=128
    )

    # 开始训练
    print("开始训练DDPG智能体...")
    results = train_ddpg(
        env,
        agent,
        episodes=300,
        max_steps=200,
        warmup_steps=5000,
        update_interval=50
    )

    # 绘制训练曲线
    plot_training_results(results)

    env.close()

运行这段代码,你会看到控制台输出训练进度。理想情况下,回合总奖励会从很大的负数(例如-1500)逐渐上升,最终稳定在-200到-50之间(对于Pendulum-v1,奖励越接近0越好)。Critic损失通常会先上升后下降并逐渐稳定,Actor损失则可能波动较大。

DDPG的常见挑战与调优技巧: 在实际训练中,你可能会遇到训练不稳定、奖励不增长甚至崩溃的情况。以下是几个关键的调优点:

  1. 学习率:Actor的学习率通常应小于Critic的学习率(例如1:10)。过大的Actor学习率容易导致策略剧烈变化,破坏Critic已经学到的Q值估计。
  2. 软更新参数tautau控制目标网络更新速度,典型值在0.001到0.01之间。较小的tau更新更慢更稳定,但学习速度也慢。
  3. OU噪声参数sigma控制探索的幅度。训练初期可以设大一些(如0.2),后期可以线性衰减到0.01,实现从探索到利用的过渡。
  4. 网络结构:对于更复杂的任务,可以尝试更深的网络或更多的隐藏层单元。但要注意,网络越复杂,需要的训练数据和时间也越多。
  5. 梯度裁剪:在Critic更新时,对梯度进行裁剪(如clip_grad_norm_)可以防止梯度爆炸,这是稳定训练的一个实用技巧。

当你的DDPG模型在Pendulum-v1上表现良好后,可以尝试将其应用到其他连续控制环境,如MountainCarContinuous-v0BipedalWalker-v3。每个环境都有其独特的挑战,可能需要调整超参数和网络结构。

6. 从DDPG到TD3:理解核心改进

DDPG在提出后取得了很大成功,但研究者发现它存在一个普遍问题:Q值过估计。这会导致策略更新基于过于乐观的价值评估,从而产生次优策略甚至训练不稳定。Twin Delayed DDPG正是为了解决这些问题而提出的改进算法。TD3的核心改进可以概括为三点,它们都旨在产生更稳定、更准确的价值估计。

第一,双Q网络。DDPG只有一个Critic网络来估计Q值,而TD3使用了两个独立的Critic网络(Q1和Q2)。在计算目标Q值时,取两个目标网络输出的最小值作为最终估计。这个简单的技巧被证明能有效缓解过估计问题。因为如果其中一个网络产生了过高的估计,最小值操作会将其抑制。

第二,延迟策略更新。在DDPG中,Actor和Critic通常每步都更新。TD3让Critic更新得更频繁(例如每步更新),而Actor更新得较慢(例如每两步更新一次)。这给了Critic更多时间在策略改变前收敛到更准确的Q值,从而为Actor提供更可靠的梯度信号。

第三,目标策略平滑。DDPG在目标动作上直接使用目标Actor网络的输出。TD3则在这个目标动作上添加一个小的裁剪噪声,然后再次裁剪到动作合法范围内。这相当于对目标Q值进行了正则化,使得相似的动作产生相似的Q值,提高了学习的稳定性。

下面是一个简化的TD3更新函数片段,展示了与DDPG的主要区别:

# TD3更新步骤(关键部分)
def update_td3(self, states, actions, rewards, next_states, dones):
    with torch.no_grad():
        # 1. 目标策略平滑:在目标动作上加噪声并裁剪
        noise = (torch.randn_like(actions) * self.policy_noise).clamp(-self.noise_clip, self.noise_clip)
        next_actions = self.actor_target(next_states)
        next_actions = (next_actions + noise).clamp(self.action_low, self.action_high)

        # 2. 双Q网络:取两个目标Critic的最小值
        target_q1 = self.critic_target1(next_states, next_actions)
        target_q2 = self.critic_target2(next_states, next_actions)
        target_q = torch.min(target_q1, target_q2)
        target_q = rewards + (1 - dones) * self.gamma * target_q

    # 3. 更新两个Critic网络
    current_q1 = self.critic1(states, actions)
    current_q2 = self.critic2(states, actions)
    critic1_loss = F.mse_loss(current_q1, target_q)
    critic2_loss = F.mse_loss(current_q2, target_q)
    # ... 反向传播更新两个Critic ...

    # 4. 延迟策略更新:每`policy_freq`步更新一次Actor
    if self.total_steps % self.policy_freq == 0:
        actor_actions = self.actor(states)
        actor_loss = -self.critic1(states, actor_actions).mean()
        # ... 反向传播更新Actor ...
        # 软更新目标网络...

在实际项目中,如果你发现DDPG训练不稳定或难以收敛,切换到TD3往往能带来显著改善。TD3的超参数设置与DDPG类似,但多出了几个新参数:policy_noise(目标策略平滑的噪声大小)、noise_clip(噪声裁剪范围)和policy_freq(Actor更新频率)。通常policy_freq=2是一个不错的起点。

我在几个连续控制基准任务上对比过DDPG和TD3,TD3几乎在所有任务上都表现得更加稳定和鲁棒。特别是在需要精细动作控制的环境中,TD3减少的过估计误差能让智能体更快地找到接近最优的策略。不过,TD3由于多了一个Critic网络,计算开销会稍微增加,但对于大多数任务来说,这点开销换来的稳定性提升是完全值得的。

Logo

更多推荐