PyTorch 深度学习实战(14):Deep Deterministic Policy Gradient (DDPG) 算法

devtools/2025/3/17 4:39:30/

在上一篇文章中,我们介绍了 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 的算法流程

  1. 使用当前策略采样一批数据。

  2. 使用目标网络计算目标 Q 值。

  3. 更新 Critic 网络以最小化 Q 值的误差。

  4. 更新 Actor 网络以最大化 Q 值。

  5. 更新目标网络。

  6. 重复上述过程,直到策略收敛。


二、Pendulum 问题实战

我们将使用 PyTorch 实现 DDPG 算法,并应用于 Pendulum 问题。目标是控制摆杆使其保持直立。

1. 问题描述

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

2. 实现步骤

  1. 安装并导入必要的库。

  2. 定义 Actor 网络和 Critic 网络。

  3. 定义 DDPG 训练过程。

  4. 测试模型并评估性能。

3. 代码实现

以下是完整的代码实现:

import gym
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
import numpy as np
import random
from collections import deque
import matplotlib.pyplot as plt
​
# 设置 Matplotlib 支持中文显示
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False
​
# 检查 GPU 是否可用
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"使用设备: {device}")
​
# 环境初始化
env = gym.make('Pendulum-v1')
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.shape[0]
max_action = float(env.action_space.high[0])
​
# 随机种子设置
SEED = 42
torch.manual_seed(SEED)
np.random.seed(SEED)
random.seed(SEED)
​
​
# 定义 Actor 网络
class Actor(nn.Module):def __init__(self, state_dim, action_dim, max_action):super(Actor, self).__init__()self.fc1 = nn.Linear(state_dim, 512)self.ln1 = nn.LayerNorm(512)  # 层归一化self.fc2 = nn.Linear(512, 512)self.ln2 = nn.LayerNorm(512)self.fc3 = nn.Linear(512, action_dim)self.max_action = max_action
​def forward(self, x):x = F.relu(self.ln1(self.fc1(x)))x = F.relu(self.ln2(self.fc2(x)))return self.max_action * torch.tanh(self.fc3(x))
​
​
# 定义 Critic 网络
class Critic(nn.Module):def __init__(self, state_dim, action_dim):super(Critic, self).__init__()self.fc1 = nn.Linear(state_dim + action_dim, 256)self.fc2 = nn.Linear(256, 256)self.fc3 = nn.Linear(256, 1)
​def forward(self, x, u):x = F.relu(self.fc1(torch.cat([x, u], 1)))x = F.relu(self.fc2(x))x = self.fc3(x)return x
​
​
# 添加OU噪声类
class OUNoise:def __init__(self, action_dim, mu=0, theta=0.15, sigma=0.2):self.mu = mu * np.ones(action_dim)self.theta = thetaself.sigma = sigmaself.reset()
​def reset(self):self.state = np.copy(self.mu)
​def sample(self):dx = self.theta * (self.mu - self.state) + self.sigma * np.random.randn(len(self.state))self.state += dxreturn self.state
​
​
# 定义 DDPG 算法
class DDPG: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.actor_target.load_state_dict(self.actor.state_dict())self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=1e-4)
​self.critic = Critic(state_dim, action_dim).to(device)self.critic_target = Critic(state_dim, action_dim).to(device)self.critic_target.load_state_dict(self.critic.state_dict())self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=1e-3)self.noise = OUNoise(action_dim, sigma=0.2)  # 示例:Ornstein-Uhlenbeck噪声
​self.max_action = max_actionself.replay_buffer = deque(maxlen=1000000)self.batch_size = 64self.gamma = 0.99self.tau = 0.005self.noise_sigma = 0.5  # 初始噪声强度self.noise_decay = 0.995
​self.actor_lr_scheduler = optim.lr_scheduler.StepLR(self.actor_optimizer, step_size=100, gamma=0.95)self.critic_lr_scheduler = optim.lr_scheduler.StepLR(self.critic_optimizer, step_size=100, gamma=0.95)
​def select_action(self, state):state = torch.FloatTensor(state).unsqueeze(0).to(device)self.actor.eval()with torch.no_grad():action = self.actor(state).cpu().data.numpy().flatten()self.actor.train()return action
​def train(self):if len(self.replay_buffer) < self.batch_size:return
​# 从经验回放缓冲区中采样batch = random.sample(self.replay_buffer, self.batch_size)state = torch.FloatTensor(np.array([transition[0] for transition in batch])).to(device)action = torch.FloatTensor(np.array([transition[1] for transition in batch])).to(device)reward = torch.FloatTensor(np.array([transition[2] for transition in batch])).reshape(-1, 1).to(device)next_state = torch.FloatTensor(np.array([transition[3] for transition in batch])).to(device)done = torch.FloatTensor(np.array([transition[4] for transition in batch])).reshape(-1, 1).to(device)
​# 计算目标 Q 值next_action = self.actor_target(next_state)target_Q = self.critic_target(next_state, next_action)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.detach())self.critic_optimizer.zero_grad()critic_loss.backward()self.critic_optimizer.step()
​# 更新 Actor 网络actor_loss = -self.critic(state, self.actor(state)).mean()self.actor_optimizer.zero_grad()actor_loss.backward()self.actor_optimizer.step()
​# 更新目标网络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)
​def save(self, filename):torch.save(self.actor.state_dict(), filename + "_actor.pth")torch.save(self.critic.state_dict(), filename + "_critic.pth")
​def load(self, filename):self.actor.load_state_dict(torch.load(filename + "_actor.pth"))self.critic.load_state_dict(torch.load(filename + "_critic.pth"))
​
​
# 训练流程
def train_ddpg(env, agent, episodes=500):rewards_history = []moving_avg = []
​for ep in range(episodes):state,_ = env.reset()episode_reward = 0done = False
​while not done:action = agent.select_action(state)next_state, reward, done, _, _ = env.step(action)agent.replay_buffer.append((state, action, reward, next_state, done))state = next_stateepisode_reward += rewardagent.train()
​rewards_history.append(episode_reward)moving_avg.append(np.mean(rewards_history[-50:]))
​if (ep + 1) % 50 == 0:print(f"Episode: {ep + 1}, Avg Reward: {moving_avg[-1]:.2f}")
​return moving_avg, rewards_history
​
​
# 训练启动
ddpg_agent = DDPG(state_dim, action_dim, max_action)
moving_avg, rewards_history = train_ddpg(env, ddpg_agent)
​
# 可视化结果
plt.figure(figsize=(12, 6))
plt.plot(rewards_history, alpha=0.6, label='single round reward')
plt.plot(moving_avg, 'r-', linewidth=2, label='moving average (50 rounds)')
plt.xlabel('episodes')
plt.ylabel('reward')
plt.title('DDPG training performance on Pendulum-v1')
plt.legend()
plt.grid(True)
plt.show()

三、代码解析

  1. Actor 和 Critic 网络

    • Actor 网络输出连续动作,通过 tanh 函数将动作限制在 −max_action,max_action 范围内。

    • Critic 网络输出状态-动作对的 Q 值。

  2. DDPG 训练过程

    • 使用当前策略采样一批数据。

    • 使用目标网络计算目标 Q 值。

    • 更新 Critic 网络以最小化 Q 值的误差。

    • 更新 Actor 网络以最大化 Q 值。

    • 更新目标网络。

  3. 训练过程

    • 在训练过程中,每 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 算法!如果有任何问题,欢迎在评论区留言讨论。


http://www.ppmy.cn/devtools/167732.html

相关文章

条件运算符

在 MySQL 中&#xff0c;BETWEEN...AND 语句是一个用于筛选数据的条件运算符&#xff0c;它可以帮助你从表中选取指定范围内的数据。 SELECT column1, column2, ... FROM table_name WHERE column_name BETWEEN value1 AND value2; 如果你想选取不在指定范围内的数据&#xff0…

将景区天气数据存储到Excel文件中

笔记 import weather import openpyxl htmlweather.get_html() # 发请求&#xff0c;得响应结果 lstweather.parse_html(html) # 解析数据 # 创建一个新的Excel工作簿 workbookopenpyxl.Workbook() # 创建对象 # 在Excel文件中创建工作表 sheetworkbook.create_sheet(景区天气…

DeepSeek-prompt指令-当DeepSeek答非所问,应该如何准确的表达我们的诉求?

当DeepSeek答非所问&#xff0c;应该如何准确的表达我们的诉求&#xff1f;不同使用场景如何向DeepSeek发问&#xff1f;是否有指令公式&#xff1f; 目录 1、 扮演专家型指令2、 知识蒸馏型指令3、 颗粒度调节型指令4、 时间轴推演型指令5、 极端测试型6、 逆向思维型指令7、…

多级缓存架构实战:Caffeine+Redis

一、架构演进与核心价值 1.1 性能对比数据 1.2 协同设计优势 二、实战案例&#xff1a;电商商品详情页优化 2.1 痛点分析 原始架构&#xff1a;单层 Redis 缓存 问题现象&#xff1a; # 压测数据 Requests/sec: 58000 99% latency: 120ms Redis CPU Usage: 85%2.2 架构改…

机器学习——正则化、欠拟合、过拟合、学习曲线

过拟合&#xff08;overfitting&#xff09;:模型只能拟合训练数据的状态。即过度训练。 避免过拟合的几种方法&#xff1a; ①增加全部训练数据的数量&#xff08;最为有效的方式&#xff09; ②使用简单的模型&#xff08;简单的模型学不够&#xff0c;复杂的模型学的太多&am…

[人工智能]实现神经网络实例

import numpy as np&#xff1a;导入 NumPy 库&#xff0c;用于数值计算。导入 PyTorch 相关库&#xff1a; import torch&#xff1a;导入 PyTorch 库&#xff0c;深度学习框架核心库。from torchvision.datasets import mnist&#xff1a;从torchvision.datasets中导入 MNIST…

树莓科技集团董事长:第五代产业园运营模式的深度剖析与展望​

第五代产业园运营模式&#xff0c;以创新为核心驱动&#xff0c;强调数字化、网络化和资源整合。树莓科技集团在这一领域具有代表性&#xff0c;其运营模式值得深入剖析。 核心特征 数字化转型&#xff1a;第五代产业园高度重视数字化技术的应用&#xff0c;通过构建数字化平…

DeepSeek 助力 Vue3 开发:打造丝滑的表格(Table)之添加列宽调整功能,示例Table14_09自定义单元格的固定表头表格

前言:哈喽,大家好,今天给大家分享一篇文章!并提供具体代码帮助大家深入理解,彻底掌握!创作不易,如果能帮助到大家或者给大家一些灵感和启发,欢迎收藏+关注哦 💕 目录 DeepSeek 助力 Vue3 开发:打造丝滑的表格(Table)之添加列宽调整功能,示例Table14_09自定义单元格…