双延迟深度确定性策略梯度算法 (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 的策略梯度才会更加准确。

(3) 目标动作平滑

在计算目标值 $y$ 时，对动作加入高斯噪声 $\epsilon \sim \mathcal{N}(0, \sigma)$ 并进行裁剪，防止策略过拟合到极端动作：

$$ a' = \pi_{\phi'}(s') + \text{clip}(\epsilon, -c, c) $$

• 这样可以让目标 Q 值更加平滑，增强策略的鲁棒性。

3. 完整 TD3 算法流程

伪代码如下：

1. 初始化 Actor 和 Critic 网络及其对应的目标网络；
2. 与环境交互，收集经验存入回放池；
3. 重复以下步骤直到收敛：
   • 从回放池采样批量数据；
   • 更新 Critic 网络（通常每步都更新）；
   • 每隔 $d$ 步更新 Actor 策略（延迟更新）；
   • 软更新目标网络参数。

目标网络软更新：

$$ \theta_i' \leftarrow \tau \theta_i + (1 - \tau) \theta_i' $$ $$ \phi' \leftarrow \tau \phi + (1 - \tau) \phi' $$

延迟更新 Actor：每隔 $d$ 步，更新 Actor 策略。

更新 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 实现

环境配置

# Python 3.11.5
# torch 2.1.0
# gym 0.26.2

代码实现

import argparse
from collections import deque
import os
import random
import numpy as np
import gym
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.distributions import Normal

device = 'cuda' if torch.cuda.is_available() else 'cpu'

class ReplayBuffer:
    def __init__(self, max_size):
        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))
        return np.array(x), np.array(y), np.array(u), np.array(r).reshape(-1, 1), np.array(d).reshape(-1, 1)

class Actor(nn.Module):
    def __init__(self, state_dim, action_dim, max_action):
        super(Actor, self).__init__()
        self.fc1 = nn.Linear(state_dim, 400)
        self.fc2 = nn.Linear(400, 300)
        self.fc3 = nn.Linear(300, action_dim)
        self.max_action = max_action

    def forward(self, state):
        a = F.relu(self.fc1(state))
        a = F.relu(self.fc2(a))
        a = torch.tanh(self.fc3(a)) * self.max_action
        return a

class Critic(nn.Module):
    def __init__(self, state_dim, action_dim):
        super(Critic, self).__init__()
        self.fc1 = nn.Linear(state_dim + action_dim, 400)
        self.fc2 = nn.Linear(400, 300)
        self.fc3 = nn.Linear(300, 1)

    def forward(self, state, action):
        state_action = torch.cat([state, action], 1)
        q = F.relu(self.fc1(state_action))
        q = F.relu(self.fc2(q))
        q = self.fc3(q)
        return q

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.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())
        self.critic_1_optimizer = optim.Adam(self.critic_1.parameters())
        self.critic_2_optimizer = optim.Adam(self.critic_2.parameters())

        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 = ReplayBuffer(max_size=50000)
        self.num_critic_update_iteration = 0
        self.num_actor_update_iteration = 0
        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()

    def update(self, num_iteration, args):
        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)

            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)
            target_Q = reward + ((1 - done) * args.gamma * target_Q).detach()

            current_Q1 = self.critic_1(state, action)
            loss_Q1 = F.mse_loss(current_Q1, target_Q)
            self.critic_1_optimizer.zero_grad()
            loss_Q1.backward()
            self.critic_1_optimizer.step()

            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_loss = - self.critic_1(state, self.actor(state)).mean()
                self.actor_optimizer.zero_grad()
                actor_loss.backward()
                self.actor_optimizer.step()

                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
            self.num_training += 1

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--env_name', default="Pendulum-v0")
    parser.add_argument('--tau', default=0.005, type=float)
    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=float)
    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('--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)
    args = parser.parse_args()

    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])

    agent = TD3(state_dim, action_dim, max_action)

    if args.mode == 'train':
        ep_r = 0
        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
                agent.memory.push((state, next_state, action, reward, np.float(done)))
                if len(agent.memory.storage) >= args.capacity - 1:
                    agent.update(10, args)
                state = next_state
                if done or t == args.max_episode - 1:
                    print(f"Ep_i {i}, ep_r {ep_r:.2f}")
                    ep_r = 0
                    break

六、总结

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

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

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

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