PyTorch 深度学习实战（15）：Twin Delayed DDPG (TD3) 算法

在上一篇文章中，我们先容了 Deep Deterministic Policy Gradient (DDPG) 算法，并使用它办理了 Pendulum 题目。本文将深入探究 Twin Delayed DDPG (TD3) 算法，这是一种改进的 DDPG 算法，可以或许有用办理 DDPG 中的过估计题目。我们将使用 PyTorch 实现 TD3 算法，并应用于经典的 Pendulum 题目。

---

一、TD3 算法基础

TD3 是 DDPG 的改进版本，通过引入以下三个关键技术来办理 DDPG 中的过估计题目：

• 双重 Critic 网络：
  使用两个 Critic 网络来估计 Q 值，从而减少过估计题目。
  
• 延迟更新：
  延迟 Actor 网络的更新，确保 Critic 网络更稳定地收敛。
  
• 目的战略平滑：
  在目的动作中参加噪声，从而减少 Critic 网络的过拟合。
  

1. TD3 的核心思想

• 双重 Critic 网络：
  
  
  
  • 使用两个 Critic 网络来估计 Q 值，取两者中的较小值作为目的 Q 值，从而减少过估计。
    
  
  
• 延迟更新：
  
  
  
  • 每更新 Critic 网络多次，才更新一次 Actor 网络，确保 Critic 网络更稳定地收敛。
    
  
  
• 目的战略平滑：
  
  
  
  • 在目的动作中参加噪声，从而减少 Critic 网络的过拟合。
    
  
  

2. TD3 的上风

• 减少过估计：
  
  
  
  • 通过双重 Critic 网络和目的战略平滑，TD3 可以或许有用减少 Q 值的过估计。
    
  
  
• 练习稳定：
  
  
  
  • 延迟更新战略确保 Critic 网络更稳定地收敛。
    
  
  
• 适用于连续动作空间：
  
  
  
  • TD3 可以或许直接输出连续动作，适用于机器人控制、自动驾驶等任务。
    
  
  

3. TD3 的算法流程

• 使用当前战略采样一批数据。
  
• 使用目的网络盘算目的 Q 值。
  
• 更新 Critic 网络以最小化 Q 值的误差。
  
• 延迟更新 Actor 网络以最大化 Q 值。
  
• 更新目的网络。
  
• 重复上述过程，直到战略收敛。
  

---

二、Pendulum 题目实战

我们将使用 PyTorch 实现 TD3 算法，并应用于 Pendulum 题目。目的是控制摆杆使其保持直立。
1. 题目描述

Pendulum 情况的状态空间包括摆杆的角度和角速率。动作空间是一个连续的扭矩值，范围在 −2,2 之间。智能体每保持摆杆直立一步，就会得到一个负的奖励，目的是最大化累积奖励。
2. 实现步骤

• 安装并导入须要的库。
  
• 界说 Actor 网络和 Critic 网络。
  
• 界说 TD3 练习过程。
  
• 测试模型并评估性能。
  

3. 代码实现

以下是完备的代码实现：

1. import gym
2. import torch
3. import torch.nn as nn
4. import torch.optim as optim
5. import torch.nn.functional as F
6. import numpy as np
7. import random
8. from collections import deque
9. import matplotlib.pyplot as plt
10. ​
11. plt.rcParams['font.sans-serif'] = ['SimHei']
12. plt.rcParams['axes.unicode_minus'] = False
13. ​
14. device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
15. print(f"使用设备: {device}")
16. ​
17. env = gym.make('Pendulum-v1')
18. state_dim = env.observation_space.shape[0]
19. action_dim = env.action_space.shape[0]
20. max_action = float(env.action_space.high[0])
21. ​
22. SEED = 42
23. torch.manual_seed(SEED)
24. np.random.seed(SEED)
25. random.seed(SEED)
26. ​
27. ​
28. # 改进的 Actor 网络（增加层归一化）
29. class Actor(nn.Module):
30.    def __init__(self, state_dim, action_dim, max_action):
31.        super(Actor, self).__init__()
32.        self.l1 = nn.Linear(state_dim, 256)
33.        self.ln1 = nn.LayerNorm(256)
34.        self.l2 = nn.Linear(256, 256)
35.        self.ln2 = nn.LayerNorm(256)
36.        self.l3 = nn.Linear(256, action_dim)
37.        self.max_action = max_action
38. ​
39.    def forward(self, x):
40.        x = F.relu(self.ln1(self.l1(x)))
41.        x = F.relu(self.ln2(self.l2(x)))
42.        x = torch.tanh(self.l3(x)) * self.max_action
43.        return x
44. ​
45. ​
46. # 改进的 Critic 网络（增加层归一化）
47. class Critic(nn.Module):
48.    def __init__(self, state_dim, action_dim):
49.        super(Critic, self).__init__()
50.        self.l1 = nn.Linear(state_dim + action_dim, 256)
51.        self.ln1 = nn.LayerNorm(256)
52.        self.l2 = nn.Linear(256, 256)
53.        self.ln2 = nn.LayerNorm(256)
54.        self.l3 = nn.Linear(256, 1)
55. ​
56.    def forward(self, x, u):
57.        x = F.relu(self.ln1(self.l1(torch.cat([x, u], 1))))
58.        x = F.relu(self.ln2(self.l2(x)))
59.        x = self.l3(x)
60.        return x
61. ​
62. ​
63. class TD3:
64.    def __init__(self, state_dim, action_dim, max_action):
65.        self.actor = Actor(state_dim, action_dim, max_action).to(device)
66.        self.actor_target = Actor(state_dim, action_dim, max_action).to(device)
67.        self.actor_target.load_state_dict(self.actor.state_dict())
68.        self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=3e-4)
69. ​
70.        self.critic1 = Critic(state_dim, action_dim).to(device)
71.        self.critic2 = Critic(state_dim, action_dim).to(device)
72.        self.critic1_target = Critic(state_dim, action_dim).to(device)
73.        self.critic2_target = Critic(state_dim, action_dim).to(device)
74.        self.critic1_target.load_state_dict(self.critic1.state_dict())
75.        self.critic2_target.load_state_dict(self.critic2.state_dict())
76.        self.critic1_optimizer = optim.Adam(self.critic1.parameters(), lr=3e-4)
77.        self.critic2_optimizer = optim.Adam(self.critic2.parameters(), lr=3e-4)
78. ​
79.        self.max_action = max_action
80.        self.replay_buffer = deque(maxlen=1000000)
81.        self.batch_size = 256
82.        self.gamma = 0.99
83.        self.tau = 0.005
84.        self.policy_noise = 0.2
85.        self.noise_clip = 0.5
86.        self.policy_freq = 2
87.        self.total_it = 0
88.        self.exploration_noise = 0.1  # 新增探索噪声
89. ​
90.    def select_action(self, state, add_noise=True):
91.        state = torch.FloatTensor(state.reshape(1, -1)).to(device)
92.        action = self.actor(state).cpu().data.numpy().flatten()
93.        if add_noise:
94.            noise = np.random.normal(0, self.exploration_noise, size=action_dim)
95.            action = (action + noise).clip(-self.max_action, self.max_action)
96.        return action
97. ​
98.    def train(self):
99.        if len(self.replay_buffer) < self.batch_size:
100.            return
101. ​
102.        self.total_it += 1
103. ​
104.        batch = random.sample(self.replay_buffer, self.batch_size)
105.        state = torch.FloatTensor(np.array([t[0] for t in batch])).to(device)
106.        action = torch.FloatTensor(np.array([t[1] for t in batch])).to(device)
107.        reward = torch.FloatTensor(np.array([t[2] for t in batch])).reshape(-1, 1).to(device) / 10.0  # 奖励缩放
108.        next_state = torch.FloatTensor(np.array([t[3] for t in batch])).to(device)
109.        done = torch.FloatTensor(np.array([t[4] for t in batch])).reshape(-1, 1).to(device)
110. ​
111.        with torch.no_grad():
112.            noise = (torch.randn_like(action) * self.policy_noise).clamp(-self.noise_clip, self.noise_clip)
113.            next_action = (self.actor_target(next_state) + noise).clamp(-self.max_action, self.max_action)
114.            target_Q1 = self.critic1_target(next_state, next_action)
115.            target_Q2 = self.critic2_target(next_state, next_action)
116.            target_Q = torch.min(target_Q1, target_Q2)
117.            target_Q = reward + (1 - done) * self.gamma * target_Q
118. ​
119.        current_Q1 = self.critic1(state, action)
120.        current_Q2 = self.critic2(state, action)
121.        critic1_loss = F.mse_loss(current_Q1, target_Q)
122.        critic2_loss = F.mse_loss(current_Q2, target_Q)
123.        self.critic1_optimizer.zero_grad()
124.        critic1_loss.backward()
125.        torch.nn.utils.clip_grad_norm_(self.critic1.parameters(), 1.0)  # 梯度裁剪
126.        self.critic1_optimizer.step()
127.        self.critic2_optimizer.zero_grad()
128.        critic2_loss.backward()
129.        torch.nn.utils.clip_grad_norm_(self.critic2.parameters(), 1.0)
130.        self.critic2_optimizer.step()
131. ​
132.        if self.total_it % self.policy_freq == 0:
133.            actor_loss = -self.critic1(state, self.actor(state)).mean()
134.            self.actor_optimizer.zero_grad()
135.            actor_loss.backward()
136.            torch.nn.utils.clip_grad_norm_(self.actor.parameters(), 1.0)
137.            self.actor_optimizer.step()
138. ​
139.            for param, target_param in zip(self.critic1.parameters(), self.critic1_target.parameters()):
140.                target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)
141.            for param, target_param in zip(self.critic2.parameters(), self.critic2_target.parameters()):
142.                target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)
143.            for param, target_param in zip(self.actor.parameters(), self.actor_target.parameters()):
144.                target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)
145. ​
146.    def save(self, filename):
147.        torch.save(self.actor.state_dict(), filename + "_actor.pth")
148. ​
149. ​
150. def train_td3(env, agent, episodes=2000, early_stop_threshold=-150):
151.    rewards_history = []
152.    moving_avg = []
153.    best_avg = -np.inf
154. ​
155.    for ep in range(episodes):
156.        state,_ = env.reset()
157.        episode_reward = 0
158.        done = False
159.        step = 0
160. ​
161.        while not done:
162.            # 线性衰减探索噪声
163.            if ep < 300:
164.                agent.exploration_noise = max(0.5 * (1 - ep / 300), 0.1)
165.            else:
166.                agent.exploration_noise = 0.1
167. ​
168.            action = agent.select_action(state, add_noise=(ep < 100))  # 前100轮强制探索
169.            next_state, reward, done, _, _ = env.step(action)
170.            agent.replay_buffer.append((state, action, reward, next_state, done))
171.            state = next_state
172.            episode_reward += reward
173.            agent.train()
174.            step += 1
175. ​
176.        rewards_history.append(episode_reward)
177.        current_avg = np.mean(rewards_history[-50:])
178.        moving_avg.append(current_avg)
179. ​
180.        if current_avg > best_avg:
181.            best_avg = current_avg
182.            agent.save("td3_pendulum_best")
183. ​
184.        if (ep + 1) % 50 == 0:
185.            print(f"Episode: {ep + 1}, Avg Reward: {current_avg:.2f}")
186. ​
187.        # 早停机制
188.        if current_avg >= early_stop_threshold:
189.            print(f"早停触发，平均奖励达到 {current_avg:.2f}")
190.            break
191. ​
192.    return moving_avg, rewards_history
193. ​
194. ​
195. # 训练并可视化
196. td3_agent = TD3(state_dim, action_dim, max_action)
197. moving_avg, rewards_history = train_td3(env, td3_agent, episodes=2000)
198. ​
199. # 可视化结果
200. plt.figure(figsize=(12, 6))
201. plt.plot(rewards_history, alpha=0.6, label='single round reward')
202. plt.plot(moving_avg, 'r-', linewidth=2, label='moving average (50 rounds)')
203. plt.xlabel('episodes')
204. plt.ylabel('reward')
205. plt.title('TD3 training performance on Pendulum-v1')
206. plt.legend()
207. plt.grid(True)
208. plt.show()

复制代码

---

三、代码剖析

• Actor 和 Critic 网络：
  
  
  
  • Actor 网络输出连续动作，通过 tanh 函数将动作限定在 −max_action,max_action 范围内。
    
  • Critic 网络输出状态-动作对的 Q 值。
    
  
  
• TD3 练习过程：
  
  
  
  • 使用当前战略采样一批数据。
    
  • 使用目的网络盘算目的 Q 值。
    
  • 更新 Critic 网络以最小化 Q 值的误差。
    
  • 延迟更新 Actor 网络以最大化 Q 值。
    
  • 更新目的网络。
    
  
  
• 练习过程：
  
  
  
  • 在练习过程中，每 50 个 episode 打印一次平均奖励。
    
  • 练习竣事后，绘制练习过程中的总奖励曲线。
    
  
  

---

四、运行结果

运行上述代码后，你将看到以下输出：

• 练习过程中每 50 个 episode 打印一次平均奖励。
  
• 练习竣事后，绘制练习过程中的总奖励曲线。
  

---

五、总结

本文先容了 TD3 算法的基本原理，并使用 PyTorch 实现了一个简单的 TD3 模型来办理 Pendulum 题目。通过这个例子，我们学习了如何使用 TD3 算法进行连续动作空间的战略优化。
在下一篇文章中，我们将探究更高级的强化学习算法，如 Soft Actor-Critic (SAC)。敬请期待！
代码实例说明：

• 本文代码可以直接在 Jupyter Notebook 或 Python 脚本中运行。
  
• 假如你有 GPU，代码会自动检测并使用 GPU 加速。
  

希望这篇文章能帮助你更好地明白 TD3 算法！假如有任何题目，接待在批评区留言讨论。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。