从零构建DDPG:在连续控制任务中实现深度确定性策略梯度

如果你之前玩过强化学习,大概率是从Q-learning或者DQN这类离散动作算法入门的。它们处理像“上下左右”这样的离散选择很在行,但当你面对机器人关节角度、自动驾驶的油门刹车、或者交易策略的仓位比例时,动作空间是连续的,一个微小的角度调整可能对应着无限种可能。这时,DDPG(深度确定性策略梯度)就成了你工具箱里不可或缺的那把瑞士军刀。

DDPG巧妙地将深度学习的表示能力与确定性策略的思想结合,为连续动作空间的控制问题提供了一个端到端的解决方案。它不像传统策略梯度方法那样输出动作的概率分布,而是直接输出一个确定的动作值。这种设计让它在许多需要精细控制的场景中,比如机械臂操控、无人机飞行、甚至金融量化交易中,都展现出了强大的潜力。今天,我们就抛开复杂的理论推导,直接动手,用Python一步步搭建一个完整的DDPG模型,并让它在一个经典环境里跑起来。我会带你走过网络设计、噪声注入、经验回放这些核心模块,并分享一些我调试过程中踩过的坑和总结的技巧。

1. 理解DDPG的核心思想与架构

在深入代码之前,我们需要先理清DDPG到底在解决什么问题,以及它的基本工作流程。想象一下训练一个机器人抓取杯子:状态是摄像头看到的图像和机械臂的关节角度,动作是每个关节电机需要转动的具体角度。这是一个典型的连续动作空间问题。

DDPG本质上属于Actor-Critic框架的变体,但它骨子里流淌着DQN的血液。它的核心创新在于引入了一个确定性策略。与输出动作概率分布的随机策略不同,确定性策略Actor网络直接根据当前状态,输出一个具体的、确定性的动作。这带来了计算上的高效性,但也带来了探索不足的问题——如果策略总是输出确定的动作,它可能永远发现不了更好的动作。为此,DDPG借鉴了DQN的经验回放和目标网络来稳定训练,并引入了一种特殊的Ornstein-Uhlenbeck (OU)过程噪声来促进探索。

整个算法的数据流可以概括为以下几个步骤:

  1. Actor网络(策略网络):接收状态 s,输出一个确定的动作 a = μ(s|θ^μ)
  2. 探索:在Actor输出的动作上添加OU噪声,得到探索动作 a' = a + noise,并执行。
  3. 存储经验:将转换 (s, a', r, s', done) 存入经验回放缓冲区。
  4. Critic网络(价值网络):从缓冲区采样一批经验,Critic网络 Q(s, a|θ^Q) 负责评估在状态 s 下执行动作 a 的好坏(Q值)。
  5. 更新Critic:使用类似于DQN的时序差分(TD)误差来更新Critic网络,使其Q值估计更准确。目标值 y = r + γ * Q'(s', μ'(s'|θ^μ')),其中 Q'μ' 是目标网络,用于稳定训练。
  6. 更新Actor:Actor的目标是最大化Critic评估的Q值。通过计算Q值相对于动作的梯度,并沿着这个梯度方向更新Actor参数,使得Actor输出的动作能获得更高的预期回报。
  7. 软更新目标网络:缓慢地将在线网络(Actor和Critic)的参数混合到目标网络中,通常采用公式 θ' ← τθ + (1-τ)θ',其中τ是一个很小的数(如0.001)。

注意:DDPG要求动作空间是连续的,并且通常假设动作的每个维度都是有界的(例如,关节角度在[-π, π]之间)。在实现时,我们通常让Actor网络输出一个在[-1, 1]范围内的值,再通过线性变换映射到实际的动作边界。

2. 搭建DDPG的神经网络组件

理论清晰后,我们开始用PyTorch搭建核心组件。我们将创建三个关键类:ActorCriticOUNoise

2.1 Actor网络:从状态到动作的映射

Actor网络的结构通常比Critic简单。它的输入是状态维度,输出是动作维度。我们使用全连接层,并在最后使用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, max_action=1.0):
        super(Actor, self).__init__()
        self.max_action = max_action  # 通常为1,用于后续缩放

        self.l1 = nn.Linear(state_dim, hidden_dim)
        self.l2 = nn.Linear(hidden_dim, hidden_dim)
        self.l3 = nn.Linear(hidden_dim, action_dim)

        # 对最后一层进行较小的初始化,有助于训练初期的稳定性
        nn.init.uniform_(self.l3.weight, -3e-3, 3e-3)
        nn.init.uniform_(self.l3.bias, -3e-3, 3e-3)

    def forward(self, state):
        x = F.relu(self.l1(state))
        x = F.relu(self.l2(x))
        # tanh将输出限制在[-1, 1],乘以max_action得到最终动作
        action = self.max_action * torch.tanh(self.l3(x))
        return action

这个网络非常简单:两个ReLU隐藏层,一个tanh输出层。max_action参数允许你灵活地定义动作的最大绝对值。在实际环境中,你需要根据环境规范对这个输出进行缩放。

2.2 Critic网络:评估状态-动作对的价值

Critic网络是DDPG学习的核心,它需要同时评估状态和动作的价值。因此,它的输入是状态和动作的拼接。一个常见的做法是先将状态输入到一个全连接层,然后将该层的输出与动作向量拼接,再输入到后续层中。

class Critic(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim=256):
        super(Critic, self).__init__()

        # 第一层:处理状态
        self.l1 = nn.Linear(state_dim, hidden_dim)
        # 第二层:处理第一层输出与动作的拼接
        self.l2 = nn.Linear(hidden_dim + action_dim, hidden_dim)
        # 第三层:输出单个Q值
        self.l3 = nn.Linear(hidden_dim, 1)

        # 对输出层进行较小的初始化
        nn.init.uniform_(self.l3.weight, -3e-3, 3e-3)
        nn.init.uniform_(self.l3.bias, -3e-3, 3e-3)

    def forward(self, state, action):
        # 将状态和动作在特征维度(dim=1)拼接
        sa = torch.cat([state, action], 1)
        q = F.relu(self.l1(state))
        q = F.relu(self.l2(torch.cat([q, action], 1)))
        q = self.l3(q)
        return q

这里我采用了另一种等价但更清晰的架构:先单独用一层处理状态,再将结果与动作拼接。两种方式都是可行的,选择哪种取决于个人习惯。Critic网络的输出是一个标量,代表给定状态-动作对的预期累积回报。

2.3 OU噪声:为确定性策略注入探索灵魂

确定性策略本身不探索,因此我们需要在动作上添加噪声。DDPG论文推荐使用Ornstein-Uhlenbeck过程噪声,它比简单的高斯噪声更具“惯性”,产生的动作序列更平滑,适合物理系统。

import numpy as np

class OUNoise:
    def __init__(self, action_dim, mu=0.0, theta=0.15, sigma=0.2):
        self.action_dim = action_dim
        self.mu = mu  # 均值,通常为0
        self.theta = theta  # 回归速度参数
        self.sigma = sigma  # 随机扰动强度
        self.state = np.ones(self.action_dim) * self.mu  # 噪声的当前状态
        self.reset()

    def reset(self):
        """重置噪声状态到均值"""
        self.state = np.ones(self.action_dim) * self.mu

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

OU噪声的数学形式是 dx = θ*(μ - x) + σ*dW,其中dW是维纳过程(可以理解为高斯噪声)。θ控制着向均值μ回归的速度,σ控制噪声的波动幅度。在训练初期,你可能会使用较大的sigma来鼓励探索,随着训练进行,可以逐渐衰减它。

3. 实现经验回放与智能体主体

有了网络和噪声,我们需要一个“记忆库”来存储和复用经验,以及一个智能体类来统筹所有的训练逻辑。

3.1 经验回放缓冲区

经验回放是打破数据时序相关性的关键技术。我们实现一个简单的先进先出(FIFO)缓冲区。

from collections import deque
import random

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

    def push(self, state, action, reward, next_state, done):
        """存储一条经验"""
        self.buffer.append((state, action, reward, next_state, done))

    def sample(self, batch_size):
        """随机采样一批经验"""
        if len(self.buffer) < batch_size:
            return None
        batch = random.sample(self.buffer, batch_size)
        state, action, reward, next_state, done = zip(*batch)
        # 转换为numpy数组,后续方便转为tensor
        return (np.array(state), np.array(action), np.array(reward),
                np.array(next_state), np.array(done))

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

提示:在实际项目中,你可能会使用更高效的采样方式,或者使用NamedTuple来存储经验,使代码更清晰。缓冲区容量通常设置为1e5到1e6量级。

3.2 DDPG智能体类

这是整个算法的核心控制器,负责初始化网络、选择动作、存储经验和更新参数。

class DDPGAgent:
    def __init__(self, state_dim, action_dim, max_action, device='cuda'):
        self.state_dim = state_dim
        self.action_dim = action_dim
        self.max_action = max_action
        self.device = torch.device(device)

        # 创建Actor和Critic网络(在线网络和目标网络)
        self.actor = Actor(state_dim, action_dim, max_action=max_action).to(self.device)
        self.actor_target = Actor(state_dim, action_dim, max_action=max_action).to(self.device)
        self.actor_target.load_state_dict(self.actor.state_dict()) # 硬拷贝初始化
        self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=1e-4)

        self.critic = Critic(state_dim, action_dim).to(self.device)
        self.critic_target = Critic(state_dim, action_dim).to(self.device)
        self.critic_target.load_state_dict(self.critic.state_dict())
        self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=1e-3)

        # 噪声生成器
        self.noise = OUNoise(action_dim)

        # 经验回放池
        self.memory = ReplayBuffer(capacity=100000)

        # 超参数
        self.gamma = 0.99  # 折扣因子
        self.tau = 0.001   # 目标网络软更新系数
        self.batch_size = 64

    def select_action(self, state, add_noise=True):
        """根据状态选择动作,训练时添加噪声,测试时不加"""
        state = torch.FloatTensor(state.reshape(1, -1)).to(self.device)
        action = self.actor(state).cpu().data.numpy().flatten()
        if add_noise:
            noise = self.noise.sample()
            action = action + noise
        # 确保动作不超出环境允许的范围
        return np.clip(action, -self.max_action, self.max_action)

    def push(self, state, action, reward, next_state, done):
        """存储经验到缓冲区"""
        self.memory.push(state, action, reward, next_state, done)

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

        # 1. 从缓冲区采样
        batch = self.memory.sample(self.batch_size)
        state, action, reward, next_state, done = [torch.FloatTensor(x).to(self.device) for x in batch]

        # 2. 更新Critic网络
        with torch.no_grad():
            # 目标Actor根据下一状态选择动作
            next_action = self.actor_target(next_state)
            # 目标Critic评估下一状态-动作对的Q值
            target_Q = self.critic_target(next_state, next_action)
            # 计算TD目标:y = r + γ * Q'(s', μ'(s')),如果回合结束则只有r
            target_Q = reward + (1 - done) * self.gamma * target_Q

        # 当前Critic对当前状态-动作对的估计
        current_Q = self.critic(state, action)
        # 计算均方误差损失
        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(), 1.0)
        self.critic_optimizer.step()

        # 3. 更新Actor网络
        # Actor的目标是最大化Critic评估的Q值
        actor_loss = -self.critic(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()

        # 4. 软更新目标网络
        for param, target_param in zip(self.critic.parameters(), self.critic_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)

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

这个DDPGAgent类封装了完整的算法流程。update方法是核心,它清晰地展示了Critic和Actor的更新顺序,以及目标网络的软更新操作。注意,我们通常让Critic的学习率(1e-3)高于Actor的学习率(1e-4),因为价值函数的估计需要更快地收敛。

4. 实战演练:在Pendulum环境中训练DDPG

理论架构和代码组件都已就位,现在让我们在一个具体的环境中测试我们的实现。我们选择Gymnasium(原OpenAI Gym)中的Pendulum-v1环境。这个环境的目标是让倒立摆保持竖直向上,动作是施加在摆杆上的力矩(连续值),状态是摆杆的角度和角速度。

4.1 环境配置与训练循环

首先,我们定义训练函数。这里会包含与环境交互、收集经验、定期更新网络以及记录训练进度的逻辑。

import gymnasium as gym
import numpy as np
from tqdm import trange
import matplotlib.pyplot as plt

def train_ddpg(env_name='Pendulum-v1', seed=0, total_timesteps=100000, eval_freq=5000):
    # 设置随机种子,保证结果可复现
    torch.manual_seed(seed)
    np.random.seed(seed)
    env = gym.make(env_name)
    env.reset(seed=seed)

    state_dim = env.observation_space.shape[0]
    action_dim = env.action_space.shape[0]
    max_action = float(env.action_space.high[0]) # 假设动作空间对称

    print(f"环境: {env_name}")
    print(f"状态维度: {state_dim}, 动作维度: {action_dim}, 最大动作值: {max_action}")

    agent = DDPGAgent(state_dim, action_dim, max_action, device='cuda' if torch.cuda.is_available() else 'cpu')

    state, _ = env.reset()
    episode_reward = 0
    episode_timesteps = 0
    episode_num = 0

    rewards = []
    actor_losses = []
    critic_losses = []

    for t in trange(total_timesteps):
        episode_timesteps += 1

        # 选择动作(训练阶段始终添加噪声)
        action = agent.select_action(state, add_noise=True)
        # 执行动作
        next_state, reward, terminated, truncated, _ = env.step(action)
        done = terminated or truncated

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

        state = next_state
        episode_reward += reward

        # 执行网络更新
        loss = agent.update()
        if loss:
            critic_loss, actor_loss = loss
            critic_losses.append(critic_loss)
            actor_losses.append(actor_loss)

        # 如果回合结束或达到最大步长
        if done:
            print(f"回合 {episode_num+1} | 步数: {episode_timesteps} | 累计奖励: {episode_reward:.2f}")
            rewards.append(episode_reward)
            # 重置环境
            state, _ = env.reset()
            agent.noise.reset() # 每回合重置噪声,避免噪声累积
            episode_reward = 0
            episode_timesteps = 0
            episode_num += 1

        # 定期评估(不添加噪声)
        if (t + 1) % eval_freq == 0:
            eval_reward = evaluate(agent, env, eval_episodes=5)
            print(f"时间步 {t+1} | 评估平均奖励: {eval_reward:.2f}")

    env.close()
    return rewards, actor_losses, critic_losses

def evaluate(agent, env, eval_episodes=10):
    """评估智能体性能,测试时不添加噪声"""
    avg_reward = 0.
    for _ in range(eval_episodes):
        state, _ = env.reset()
        episode_reward = 0
        done = False
        while not done:
            action = agent.select_action(state, add_noise=False) # 关键:测试时不探索
            state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated
            episode_reward += reward
        avg_reward += episode_reward
    avg_reward /= eval_episodes
    return avg_reward

训练循环是强化学习的标准范式:交互 -> 存储 -> 学习。我们使用tqdm来显示进度条。注意在每回合开始时重置OU噪声,这能防止噪声状态在长时间序列中漂移得太远。评估函数evaluate在测试时关闭了探索噪声,以查看策略的真实性能。

4.2 超参数调优与调试技巧

DDPG对超参数比较敏感。以下是一些关键参数的经验值和我调试时的心得:

超参数 典型值/范围 作用与调试建议
Actor 学习率 1e-4 到 1e-3 通常比Critic小一个数量级。太大容易导致策略震荡。
Critic 学习率 1e-3 到 1e-2 价值函数需要更快收敛来指导策略。
回放缓冲区大小 1e5 到 1e6 越大越好,但会占用更多内存。确保在开始更新前缓冲区有足够数据。
批次大小 64, 128, 256 常见值。太小噪声大,太大计算慢且可能过拟合旧经验。
折扣因子 γ 0.99 对于大多数连续控制任务,0.99是安全的选择。
软更新系数 τ 0.001 到 0.01 控制目标网络更新速度。越小越稳定,但学习可能变慢。
OU噪声 θ 0.15 回归速度。值越大,噪声回归均值越快,探索性越低。
OU噪声 σ 0.2 噪声强度。可尝试在训练过程中线性衰减。

常见问题与解决方案:

  1. 奖励不增长,一直为负值(如Pendulum始终在-1500左右)

    • 检查动作缩放:确认Actor网络输出的tanh值是否被正确缩放到环境要求的动作范围内。这是新手最容易出错的地方。
    • 检查噪声:OU噪声的幅度(sigma)可能太大,淹没了策略信号。尝试减小sigma,或改用简单的高斯噪声试试。
    • 检查网络初始化:最后层的初始化是否足够小?过大的初始输出可能导致训练初期不稳定。
    • 增加探索:在训练的最初几千步,可以完全使用随机动作(random_action)来填充经验池,这被称为“预热”阶段。
  2. 训练不稳定,奖励曲线剧烈震荡

    • 降低学习率:特别是Actor的学习率。
    • 使用梯度裁剪:如代码中所示,防止梯度爆炸。
    • 减小批次大小:大的批次可能包含更多冲突的梯度信号。
    • 检查目标网络更新:确保tau值足够小(如0.001),更新足够平滑。
  3. 智能体似乎“学傻了”,性能突然下降

    • 这可能是灾难性遗忘的迹象。经验回放缓冲区中旧的成功经验被新的大量失败经验覆盖。尝试增大缓冲区容量,或者使用优先级经验回放(Prioritized Experience Replay),让算法更频繁地回放那些TD误差大的、重要的经验。

4.3 可视化训练结果与模型保存

训练完成后,我们自然想看看学习曲线,并保存训练好的模型以备后用。

def plot_training_progress(rewards, actor_losses=None, critic_losses=None, window=50):
    """绘制奖励和损失曲线"""
    fig, axes = plt.subplots(1, 2 if (actor_losses and critic_losses) else 1, figsize=(14, 5))

    # 绘制平滑后的奖励曲线
    smoothed_rewards = [np.mean(rewards[max(0, i-window):i+1]) for i in range(len(rewards))]
    axes[0].plot(rewards, alpha=0.3, label='原始奖励')
    axes[0].plot(smoothed_rewards, label=f'滑动平均 (窗口={window})', linewidth=2)
    axes[0].set_xlabel('回合数')
    axes[0].set_ylabel('回合奖励')
    axes[0].set_title('训练奖励曲线')
    axes[0].legend()
    axes[0].grid(True, linestyle='--', alpha=0.7)

    if actor_losses and critic_losses:
        # 绘制损失曲线(同样可以平滑)
        smoothed_actor_loss = [np.mean(actor_losses[max(0, i-window):i+1]) for i in range(len(actor_losses))]
        smoothed_critic_loss = [np.mean(critic_losses[max(0, i-window):i+1]) for i in range(len(critic_losses))]

        axes[1].plot(actor_losses, alpha=0.3, color='orange', label='Actor损失 (原始)')
        axes[1].plot(smoothed_actor_loss, color='orange', label='Actor损失 (平滑)', linewidth=2)
        axes[1].plot(critic_losses, alpha=0.3, color='green', label='Critic损失 (原始)')
        axes[1].plot(smoothed_critic_loss, color='green', label='Critic损失 (平滑)', linewidth=2)
        axes[1].set_xlabel('更新步数')
        axes[1].set_ylabel('损失值')
        axes[1].set_title('Actor/Critic 损失曲线')
        axes[1].legend()
        axes[1].grid(True, linestyle='--', alpha=0.7)
        axes[1].set_yscale('log') # 损失值可能跨度大,使用对数坐标更清晰

    plt.tight_layout()
    plt.show()

# 假设我们已经运行了训练,得到了rewards, actor_losses, critic_losses
# rewards, actor_losses, critic_losses = train_ddpg(...)
# plot_training_progress(rewards, actor_losses, critic_losses)

# 保存模型
def save_model(agent, path='ddpg_model.pth'):
    torch.save({
        'actor_state_dict': agent.actor.state_dict(),
        'critic_state_dict': agent.critic.state_dict(),
        'actor_target_state_dict': agent.actor_target.state_dict(),
        'critic_target_state_dict': agent.critic_target.state_dict(),
        'actor_optimizer_state_dict': agent.actor_optimizer.state_dict(),
        'critic_optimizer_state_dict': agent.critic_optimizer.state_dict(),
    }, path)
    print(f"模型已保存至 {path}")

# 加载模型
def load_model(agent, path='ddpg_model.pth'):
    checkpoint = torch.load(path, map_location=agent.device)
    agent.actor.load_state_dict(checkpoint['actor_state_dict'])
    agent.critic.load_state_dict(checkpoint['critic_state_dict'])
    agent.actor_target.load_state_dict(checkpoint['actor_target_state_dict'])
    agent.critic_target.load_state_dict(checkpoint['critic_target_state_dict'])
    agent.actor_optimizer.load_state_dict(checkpoint['actor_optimizer_state_dict'])
    agent.critic_optimizer.load_state_dict(checkpoint['critic_optimizer_state_dict'])
    print(f"已从 {path} 加载模型")

可视化不仅能让你直观看到训练是否收敛,还能通过损失曲线判断学习过程是否稳定。保存模型时,最好将优化器的状态也一并保存,这样在中断后恢复训练时,动量等参数才能保持一致。

5. 超越DDPG:TD3算法简介与实现要点

DDPG是一个强大的基线,但它有一些已知的缺陷,比如对超参数敏感、训练不稳定、容易高估Q值。Twin Delayed DDPG (TD3) 算法通过三个核心技巧有效地缓解了这些问题:

  1. 双Q网络 (Twin Q-networks):使用两个独立的Critic网络,在计算目标值时取两者的最小值,可以有效缓解Q值的高估问题。
  2. 延迟策略更新 (Delayed Policy Updates):Critic网络更新多次后,才更新一次Actor网络。这给了价值函数更多时间变得准确,从而为策略更新提供了更可靠的梯度。
  3. 目标策略平滑 (Target Policy Smoothing):在目标Actor网络输出的动作上添加一小段裁剪过的噪声,再输入给目标Critic网络。这相当于对Q函数进行了正则化,使其在动作变化微小的情况下估值也变化平缓,提高了Critic的泛化能力。

TD3的智能体类与DDPG非常相似,主要区别在于update函数:

class TD3Agent(DDPGAgent): # 继承自DDPGAgent
    def __init__(self, state_dim, action_dim, max_action, device='cuda', policy_noise=0.2, noise_clip=0.5, policy_freq=2):
        super().__init__(state_dim, action_dim, max_action, device)
        # 创建第二个Critic网络及其目标网络
        self.critic2 = Critic(state_dim, action_dim).to(self.device)
        self.critic2_target = Critic(state_dim, action_dim).to(self.device)
        self.critic2_target.load_state_dict(self.critic2.state_dict())
        self.critic2_optimizer = torch.optim.Adam(self.critic2.parameters(), lr=1e-3)

        # TD3特有参数
        self.policy_noise = policy_noise
        self.noise_clip = noise_clip
        self.policy_freq = policy_freq
        self.total_it = 0 # 记录总的更新次数

    def update(self):
        self.total_it += 1
        # ... (采样经验部分与DDPG相同) ...

        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)

            # 双Q学习:取两个目标Critic网络输出的最小值
            target_Q1 = self.critic_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

        # 更新两个Critic网络
        current_Q1 = self.critic(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.critic_optimizer.zero_grad()
        critic1_loss.backward()
        self.critic_optimizer.step()

        self.critic2_optimizer.zero_grad()
        critic2_loss.backward()
        self.critic2_optimizer.step()

        # 延迟策略更新:每 policy_freq 步更新一次Actor
        if self.total_it % self.policy_freq == 0:
            actor_loss = -self.critic(state, self.actor(state)).mean()
            self.actor_optimizer.zero_grad()
            actor_loss.backward()
            self.actor_optimizer.step()

            # 软更新目标网络(包括第二个Critic)
            for param, target_param in zip(self.critic.parameters(), self.critic_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)

        return critic1_loss.item(), critic2_loss.item(), actor_loss.item() if self.total_it % self.policy_freq == 0 else None

在实际项目中,如果DDPG训练不稳定或者难以调参,切换到TD3往往能获得更鲁棒和更好的性能。它的超参数设置与DDPG类似,但通常更不敏感。policy_freq(延迟更新频率)通常设置为2,policy_noisenoise_clip需要根据具体环境的动作尺度进行调整。

从Pendulum这样的简单环境起步,理解DDPG/TD3的每个模块是如何协同工作的,是掌握现代深度强化学习的关键一步。当你需要将算法应用到机械臂、自动驾驶仿真甚至交易策略时,最大的挑战往往不是算法本身,而是如何将你的具体问题准确地建模成强化学习框架中的状态、动作和奖励。这个过程需要大量的领域知识和反复的试错,但一旦跑通,看到智能体从零开始学会完成复杂任务,那种成就感是无与伦比的。我自己的经验是,在实现完基础版本后,花时间仔细设计奖励函数、做好状态归一化、并搭建一个有效的监控和可视化系统,往往比盲目调整网络结构或超参数更能带来性能的突破。

Logo

更多推荐