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文件中

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

笔记 import weather import openpyxl htmlweather.get_html() # 发请求&#xff0c;得响应结果 lstweather.parse_html(html) # 解析数据 # 创建一个新的Excel工作簿 workbookopenpyxl.Workbook() # 创建对象 # 在Excel文件中创建工作表 sheetworkbook.create_sheet(景区天气…
阅读更多...
DeepSeek-prompt指令-当DeepSeek答非所问，应该如何准确的表达我们的诉求？

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

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

多级缓存架构实战：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自定义单元格的固定表头表格

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

Copyright @ 2022~2023