PyTorch 深度学习实战（14）：Deep Deterministic Policy Gradient (DDPG) ...

在上一篇文章中，我们介绍了 Proximal Policy Optimization (PPO) 算法，并利用它解决了 CartPole 题目。本文将深入探究 Deep Deterministic Policy Gradient (DDPG) 算法，这是一种用于连续动作空间的强化学习算法。我们将利用 PyTorch 实现 DDPG 算法，并应用于经典的 Pendulum 题目。

---

一、DDPG 算法基础

DDPG 是一种基于 Actor-Critic 框架的算法，专门用于解决连续动作空间的强化学习题目。它结合了深度 Q 网络（DQN）和策略梯度方法的长处，能够高效地处置惩罚高维状态和动作空间。
1. DDPG 的核心思想

• 确定性策略：
  
  
  
  • DDPG 利用确定性策略（Deterministic Policy），即给定状态时，策略网络直接输出一个确定的动作，而不是动作的概率分布。
    
  
  
• 目的网络：
  
  
  
  • DDPG 利用目的网络（Target Network）来稳定训练过程，类似于 DQN 中的目的网络。
    
  
  
• 经验回放：
  
  
  
  • DDPG 利用经验回放缓冲区（Replay Buffer）来存储和重用已往的经验，从而提高数据利用率。
    
  
  

2. DDPG 的上风

• 实用于连续动作空间：
  
  
  
  • DDPG 能够直接输出连续动作，实用于机器人控制、主动驾驶等使命。
    
  
  
• 训练稳定：
  
  
  
  • 通过目的网络和经验回放，DDPG 能够稳定地训练策略网络和价值网络。
    
  
  
• 高效采样：
  
  
  
  • DDPG 可以重复利用旧策略的采样数据，从而提高数据利用率。
    
  
  

3. DDPG 的算法流程

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

---

二、Pendulum 题目实战

我们将利用 PyTorch 实现 DDPG 算法，并应用于 Pendulum 题目。目的是控制摆杆使其保持竖立。
1. 题目形貌

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

• 安装并导入必要的库。
  
• 定义 Actor 网络和 Critic 网络。
  
• 定义 DDPG 训练过程。
  
• 测试模型并评估性能。
  

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. # 设置 Matplotlib 支持中文显示
12. plt.rcParams['font.sans-serif'] = ['SimHei']
13. plt.rcParams['axes.unicode_minus'] = False
14. ​
15. # 检查 GPU 是否可用
16. device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
17. print(f"使用设备: {device}")
18. ​
19. # 环境初始化
20. env = gym.make('Pendulum-v1')
21. state_dim = env.observation_space.shape[0]
22. action_dim = env.action_space.shape[0]
23. max_action = float(env.action_space.high[0])
24. ​
25. # 随机种子设置
26. SEED = 42
27. torch.manual_seed(SEED)
28. np.random.seed(SEED)
29. random.seed(SEED)
30. ​
31. ​
32. # 定义 Actor 网络
33. class Actor(nn.Module):
34.    def __init__(self, state_dim, action_dim, max_action):
35.        super(Actor, self).__init__()
36.        self.fc1 = nn.Linear(state_dim, 512)
37.        self.ln1 = nn.LayerNorm(512)  # 层归一化
38.        self.fc2 = nn.Linear(512, 512)
39.        self.ln2 = nn.LayerNorm(512)
40.        self.fc3 = nn.Linear(512, action_dim)
41.        self.max_action = max_action
42. ​
43.    def forward(self, x):
44.        x = F.relu(self.ln1(self.fc1(x)))
45.        x = F.relu(self.ln2(self.fc2(x)))
46.        return self.max_action * torch.tanh(self.fc3(x))
47. ​
48. ​
49. # 定义 Critic 网络
50. class Critic(nn.Module):
51.    def __init__(self, state_dim, action_dim):
52.        super(Critic, self).__init__()
53.        self.fc1 = nn.Linear(state_dim + action_dim, 256)
54.        self.fc2 = nn.Linear(256, 256)
55.        self.fc3 = nn.Linear(256, 1)
56. ​
57.    def forward(self, x, u):
58.        x = F.relu(self.fc1(torch.cat([x, u], 1)))
59.        x = F.relu(self.fc2(x))
60.        x = self.fc3(x)
61.        return x
62. ​
63. ​
64. # 添加OU噪声类
65. class OUNoise:
66.    def __init__(self, action_dim, mu=0, theta=0.15, sigma=0.2):
67.        self.mu = mu * np.ones(action_dim)
68.        self.theta = theta
69.        self.sigma = sigma
70.        self.reset()
71. ​
72.    def reset(self):
73.        self.state = np.copy(self.mu)
74. ​
75.    def sample(self):
76.        dx = self.theta * (self.mu - self.state) + self.sigma * np.random.randn(len(self.state))
77.        self.state += dx
78.        return self.state
79. ​
80. ​
81. # 定义 DDPG 算法
82. class DDPG:
83.    def __init__(self, state_dim, action_dim, max_action):
84.        self.actor = Actor(state_dim, action_dim, max_action).to(device)
85.        self.actor_target = Actor(state_dim, action_dim, max_action).to(device)
86.        self.actor_target.load_state_dict(self.actor.state_dict())
87.        self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=1e-4)
88. ​
89.        self.critic = Critic(state_dim, action_dim).to(device)
90.        self.critic_target = Critic(state_dim, action_dim).to(device)
91.        self.critic_target.load_state_dict(self.critic.state_dict())
92.        self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=1e-3)
93.        self.noise = OUNoise(action_dim, sigma=0.2)  # 示例：Ornstein-Uhlenbeck噪声
94. ​
95.        self.max_action = max_action
96.        self.replay_buffer = deque(maxlen=1000000)
97.        self.batch_size = 64
98.        self.gamma = 0.99
99.        self.tau = 0.005
100.        self.noise_sigma = 0.5  # 初始噪声强度
101.        self.noise_decay = 0.995
102. ​
103.        self.actor_lr_scheduler = optim.lr_scheduler.StepLR(self.actor_optimizer, step_size=100, gamma=0.95)
104.        self.critic_lr_scheduler = optim.lr_scheduler.StepLR(self.critic_optimizer, step_size=100, gamma=0.95)
105. ​
106.    def select_action(self, state):
107.        state = torch.FloatTensor(state).unsqueeze(0).to(device)
108.        self.actor.eval()
109.        with torch.no_grad():
110.            action = self.actor(state).cpu().data.numpy().flatten()
111.        self.actor.train()
112.        return action
113. ​
114.    def train(self):
115.        if len(self.replay_buffer) < self.batch_size:
116.            return
117. ​
118.        # 从经验回放缓冲区中采样
119.        batch = random.sample(self.replay_buffer, self.batch_size)
120.        state = torch.FloatTensor(np.array([transition[0] for transition in batch])).to(device)
121.        action = torch.FloatTensor(np.array([transition[1] for transition in batch])).to(device)
122.        reward = torch.FloatTensor(np.array([transition[2] for transition in batch])).reshape(-1, 1).to(device)
123.        next_state = torch.FloatTensor(np.array([transition[3] for transition in batch])).to(device)
124.        done = torch.FloatTensor(np.array([transition[4] for transition in batch])).reshape(-1, 1).to(device)
125. ​
126.        # 计算目标 Q 值
127.        next_action = self.actor_target(next_state)
128.        target_Q = self.critic_target(next_state, next_action)
129.        target_Q = reward + (1 - done) * self.gamma * target_Q
130. ​
131.        # 更新 Critic 网络
132.        current_Q = self.critic(state, action)
133.        critic_loss = F.mse_loss(current_Q, target_Q.detach())
134.        self.critic_optimizer.zero_grad()
135.        critic_loss.backward()
136.        self.critic_optimizer.step()
137. ​
138.        # 更新 Actor 网络
139.        actor_loss = -self.critic(state, self.actor(state)).mean()
140.        self.actor_optimizer.zero_grad()
141.        actor_loss.backward()
142.        self.actor_optimizer.step()
143. ​
144.        # 更新目标网络
145.        for param, target_param in zip(self.critic.parameters(), self.critic_target.parameters()):
146.            target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)
147.        for param, target_param in zip(self.actor.parameters(), self.actor_target.parameters()):
148.            target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)
149. ​
150.    def save(self, filename):
151.        torch.save(self.actor.state_dict(), filename + "_actor.pth")
152.        torch.save(self.critic.state_dict(), filename + "_critic.pth")
153. ​
154.    def load(self, filename):
155.        self.actor.load_state_dict(torch.load(filename + "_actor.pth"))
156.        self.critic.load_state_dict(torch.load(filename + "_critic.pth"))
157. ​
158. ​
159. # 训练流程
160. def train_ddpg(env, agent, episodes=500):
161.    rewards_history = []
162.    moving_avg = []
163. ​
164.    for ep in range(episodes):
165.        state,_ = env.reset()
166.        episode_reward = 0
167.        done = False
168. ​
169.        while not done:
170.            action = agent.select_action(state)
171.            next_state, reward, done, _, _ = env.step(action)
172.            agent.replay_buffer.append((state, action, reward, next_state, done))
173.            state = next_state
174.            episode_reward += reward
175.            agent.train()
176. ​
177.        rewards_history.append(episode_reward)
178.        moving_avg.append(np.mean(rewards_history[-50:]))
179. ​
180.        if (ep + 1) % 50 == 0:
181.            print(f"Episode: {ep + 1}, Avg Reward: {moving_avg[-1]:.2f}")
182. ​
183.    return moving_avg, rewards_history
184. ​
185. ​
186. # 训练启动
187. ddpg_agent = DDPG(state_dim, action_dim, max_action)
188. moving_avg, rewards_history = train_ddpg(env, ddpg_agent)
189. ​
190. # 可视化结果
191. plt.figure(figsize=(12, 6))
192. plt.plot(rewards_history, alpha=0.6, label='single round reward')
193. plt.plot(moving_avg, 'r-', linewidth=2, label='moving average (50 rounds)')
194. plt.xlabel('episodes')
195. plt.ylabel('reward')
196. plt.title('DDPG training performance on Pendulum-v1')
197. plt.legend()
198. plt.grid(True)
199. plt.show()

复制代码

---

三、代码剖析

• Actor 和 Critic 网络：
  
  
  
  • Actor 网络输出连续动作，通过 tanh 函数将动作限制在 −max_action,max_action 范围内。
    
  • Critic 网络输出状态-动作对的 Q 值。
    
  
  
• DDPG 训练过程：
  
  
  
  • 利用当前策略采样一批数据。
    
  • 利用目的网络计算目的 Q 值。
    
  • 更新 Critic 网络以最小化 Q 值的偏差。
    
  • 更新 Actor 网络以最大化 Q 值。
    
  • 更新目的网络。
    
  
  
• 训练过程：
  
  
  
  • 在训练过程中，每 50 个 episode 打印一次均匀奖励。
    
  • 训练结束后，绘制训练过程中的总奖励曲线。
    
  
  

---

四、运行效果

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

• 训练过程中每 50 个 episode 打印一次均匀奖励。
  
• 训练结束后，绘制训练过程中的总奖励曲线。
  

---

五、总结

本文介绍了 DDPG 算法的基本原理，并利用 PyTorch 实现了一个简单的 DDPG 模型来解决 Pendulum 题目。通过这个例子，我们学习了如何利用 DDPG 算法进行连续动作空间的策略优化。
在下一篇文章中，我们将探究更高级的强化学习算法，如 Twin Delayed DDPG (TD3)。敬请期待！
代码实例说明：

• 本文代码可以直接在 Jupyter Notebook 或 Python 脚本中运行。
  
• 如果你有 GPU，可以将模型和数据移动到 GPU 上运行，例如：actor = actor.to('cuda')，state = state.to('cuda')。
  

盼望这篇文章能资助你更好地理解 DDPG 算法！如果有任何题目，接待在评论区留言讨论。

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