创建时间: 2024/12/2 12:45:16作者: 蜡笔大新笔记类型: 强化学习简介DDPG（Deep Deterministic Policy Gradient）是一种基于 Actor-Critic 结构的深度强化学习算法。传统的 Q-Learning 只能处理离散状态和动作，DQN 能处理连续状态和离散动作，而[font color="Red"]DDPG则解决了在连续状态和动作空间下的强化学习问题[/font]。DDPG 与梯度策略算法系列中的 REINFORCE 算法有所不同：REINFORCE 提供随机策略，而 DDPG 提供确定性策略（Deterministic Policy）。具体来说，非确定性策略 $\pi_\theta$﻿ 输出每个动作的概率（对于连续动作则是概率分布曲线），可以通过 $\epsilon$﻿ 策略或选择最高概率的动作来决策；而确定性策略则直接输出具体动作，记作 $a=\mu_\theta (s)$﻿。由于 DDPG 的确定性特征限制了其探索能力，因此算法中添加了随机噪声来增强探索。网络Actor与传统 Actor-Critic 算法中的 Actor 类似，DDPG 中的 Actor 也是一个神经网络，区别在于神经网络和激活函数的设置不一样。DDPG 的 Actor 通常最后一层直接输出一个具体值，对于连续动作空间，可能会使用特定的激活函数限制输出范围，例如：Tanh 激活函数：限制输出在 [-1, 1]。线性输出：直接输出在连续范围内的值。最后会输出一个网络认为的价值期望最高的动作，输出维度等于动作空间的维度。CriticDDPG 的 Critic 网络与 DQN 的 Q 网络和传统 AC 的 Critic 网络有明显区别。DQN 的 Q 网络输入状态后输出所有动作的 Q 值，这种方式只适用于离散动作问题，无法处理连续动作空间。传统 AC 算法中的 Critic 使用神经网络来估计状态的 V 值。而 DDPG 的 Critic 则采用神经网络同时接收状态和动作作为输入，用于预估 Q 值。更新网络参数更新的时间节点主要是在经验池满了之后，随机采样经验池的经验进行更新，通过这种机制能够有效减少样本的时间相关性，避免模型过度依赖于某些特定的样本，使得训练数据更具多样性，增强了策略在不同场景中的泛化能力，提升模型的训练稳定性。ActorActor 网络主要使用神经网络输出当前状态下价值期望最高的动作，其优化的目标就是最大化 Critic 网络的 Q 值，更新主要是通过 Critic 网络输出的 Q 值进行梯度上升。CriticCritic 网络的损失函数是当前 Q 值和目标 Q 值的均方误差（MSE）。其更新目标是最小化这个均方误差，通过梯度下降来更新参数。MSE公式$$ L(\theta^Q) = \frac{1}{N} \sum_{i=1}^N \left( Q_{\text{target}, i} - Q(s_i, a_i|\theta^Q) \right)^2 $$目标 Q 值的计算$$ Q_{\text{target}, i} = r_i + \gamma (1 - \text{done}_i) Q'(s'_i, \mu'(s'_i|\theta^{\mu'})|\theta^{Q'}) $$这个公式其实就是贝尔曼公式的变种。[scode type="blue" size=""]其中：$Q_{\text{target}, i}$ 表示目标网络的 Q 值。从当前状态 $s_i$ 和动作 $a_i$ 开始，未来累积奖励的估计值。$r_i$ 表示智能体在当前状态 $s_i$ 执行动作 $a_i$ 后，从环境中获得的奖励$\gamma$ 表示折扣因子$done_i$ 表示当前状态是否是终止状态（0代表未终止，1代表终止）$\mu'(s'_i|\theta^{\mu'})$ 代表目标 Actor 网络根据 $\theta^{Q'}$ 软更新参数 $\theta^{Q}$ 的情况下，根据下一个状态 $s'_i$ 生成的动作 $Q{\prime}(s{\prime}_i, a{\prime}|\theta^{Q{\prime}})$ 表示目标 Critic 网络根据下一个状态和下一个动作生成的 Q 值[/scode]至此，重新梳理一下整个目标 Q 值的计算：获取当前状态 $s_t$ 和当前动作 $a_t$当前状态 $s_t$ 执行当前动作 $a_t$，获取下一个状态 $s_{t+1}$将下一个状态 $s_{t+1}$ 输入进目标 Actor 网络，获取下一个动作 $a_{t+1}$将下一个状态 $s_{t+1}$ 和下一个动作 $a_{t+1}$ 输入进目标 Critic 网络，获取 Q 值将目标 Critic 网络生成的 Q 值与折扣因子、完成状态、当前状态 $s_t$ 执行当前动作 $a_t$ 获得的奖励一起代入公式计算目标 Q 值软更新软更新一般是在随机采样更新完当前 Actor 和 Critic 网络后，对两个目标网络利用公式进行的。由于参数的设置，其实际效果就是每次只更新当前网络的非常小的部分，绝大多数都还是由原来的目标网络参数构成。$$ \theta^{Q'} \leftarrow \tau \theta^Q + (1 - \tau) \theta^{Q'} $$$$ \theta^{\mu'} \leftarrow \tau \theta^\mu + (1 - \tau) \theta^{\mu'} $$[scode type="blue" size=""]其中：$\theta^{Q{\prime}}$：目标 Critic 网络的参数$\theta^{Q}$：当前 Critic 网络的参数$\theta^{\mu{\prime}}$：目标 Actor 网络的参数$\theta^{\mu}$：当前 Actor 网络的参数$\tau$：软更新系数，通常取值很小（如 $\tau$ = 0.001）[/scode]其它亮点归一化在 DDPG 的原版论文中，提到了批量归一化。批量归一化有助于减小训练中的 协方差偏移（covariance shift），确保每一层接收到归一化的输入数据。在低维情况下，批量归一化被应用于状态输入、策略网络的所有层，以及价值网络中动作输入之前的所有层。使用批量归一化后，可以跨不同任务（任务特征和单位类型不同）进行有效学习，而无需手动调整特征范围。噪声在 DDPG 的原版论文中，提到了添加噪声来增加探索性。由于 DDPG 的 Actor 网络输出的是确定性动作，因此不能像以往的 DQN 一样通过 $\epsilon$ 贪婪策略来探索。原文采用的是Ornstein-Uhlenbeck 噪声，这是一种时间相关的噪声,用于在具有惯性的物理控制问题中提升探索效率。在这个公式中，主要是对 Actor 网络生成的动作添加一些随机的噪声 $N$ 来干扰，在解决连续动作时效果比较显著。解决连续动作时一般输出一个数值，例如机器人的手臂转动角度，通过添加噪声可以更好的实现探索效果。$$ \mu'(s_t) = \mu(s_t \mid \theta_t^\mu) + \mathcal{N} $$总体流程DDPG 的训练总体流程主要分为交互、更新，分别在上文进行了阐述。大部分时间里，网络中的智能体与环境进行交互，获得奖励、动作等状态，产生经验，将经验放在经验池里。当经验池满了之后，取出部分经验开始更新。使用 MSE 作为 Critic 的损失函数，计算当前 Q 值和目标 Q 值的误差，并用这个误差梯度下降更新 Critic 的参数。使用 Critic 的 Q 值梯度上升更新 Actor 的参数。利用软更新公式对目标 Actor 网络和目标 Critic 网络进行软更新。若未训练完成则继续与环境交互。总体流程图如下：优缺点分析优点适用于连续动作空间样本效率较高结合了值函数和策略梯度方法的优势缺点对超参数较为敏感训练不够稳定可能存在过估计问题代码import random import gym import numpy as np import torch import torch.nn.functional as F import matplotlib.pyplot as plt import rl_utils class PolicyNet(torch.nn.Module): def __init__(self, state_dim, hidden_dim, action_dim, action_bound): super(PolicyNet, self).__init__() self.fc1 = torch.nn.Linear(state_dim, hidden_dim) self.fc2 = torch.nn.Linear(hidden_dim, action_dim) self.action_bound = action_bound # action_bound是环境可以接受的动作最大值 def forward(self, x): x = F.relu(self.fc1(x)) return torch.tanh(self.fc2(x)) * self.action_bound class QValueNet(torch.nn.Module): def __init__(self, state_dim, hidden_dim, action_dim): super(QValueNet, self).__init__() self.fc1 = torch.nn.Linear(state_dim + action_dim, hidden_dim) self.fc2 = torch.nn.Linear(hidden_dim, hidden_dim) self.fc_out = torch.nn.Linear(hidden_dim, 1) def forward(self, x, a): cat = torch.cat([x, a], dim=1) # 拼接状态和动作 x = F.relu(self.fc1(cat)) x = F.relu(self.fc2(x)) return self.fc_out(x) class DDPG: """ DDPG算法 """ def __init__(self, state_dim, hidden_dim, action_dim, action_bound, sigma, actor_lr, critic_lr, tau, gamma, device): self.actor = PolicyNet(state_dim, hidden_dim, action_dim, action_bound).to(device) self.critic = QValueNet(state_dim, hidden_dim, action_dim).to(device) self.target_actor = PolicyNet(state_dim, hidden_dim, action_dim, action_bound).to(device) self.target_critic = QValueNet(state_dim, hidden_dim, action_dim).to(device) # 初始化目标价值网络并设置和价值网络相同的参数 self.target_critic.load_state_dict(self.critic.state_dict()) # 初始化目标策略网络并设置和策略相同的参数 self.target_actor.load_state_dict(self.actor.state_dict()) self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=actor_lr) self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=critic_lr) self.gamma = gamma self.sigma = sigma # 高斯噪声的标准差,均值直接设为0 self.tau = tau # 目标网络软更新参数 self.action_dim = action_dim self.device = device def take_action(self, state): state = torch.tensor([state], dtype=torch.float).to(self.device) action = self.actor(state).item() # 给动作添加噪声，增加探索 action = action + self.sigma * np.random.randn(self.action_dim) return action def soft_update(self, net, target_net): for param_target, param in zip(target_net.parameters(), net.parameters()): param_target.data.copy_(param_target.data * (1.0 - self.tau) + param.data * self.tau) def update(self, transition_dict): states = torch.tensor(transition_dict['states'], dtype=torch.float).to(self.device) actions = torch.tensor(transition_dict['actions'], dtype=torch.float).view(-1, 1).to(self.device) rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1).to(self.device) next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float).to(self.device) dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1, 1).to(self.device) next_q_values = self.target_critic(next_states, self.target_actor(next_states)) q_targets = rewards + self.gamma * next_q_values * (1 - dones) critic_loss = torch.mean(F.mse_loss(self.critic(states, actions), q_targets)) self.critic_optimizer.zero_grad() critic_loss.backward() self.critic_optimizer.step() actor_loss = -torch.mean(self.critic(states, self.actor(states))) self.actor_optimizer.zero_grad() actor_loss.backward() self.actor_optimizer.step() self.soft_update(self.actor, self.target_actor) # 软更新策略网络 self.soft_update(self.critic, self.target_critic) # 软更新价值网络 actor_lr = 3e-4 critic_lr = 3e-3 num_episodes = 1000 hidden_dim = 64 gamma = 0.98 tau = 0.005 # 软更新参数 buffer_size = 10000 minimal_size = 1000 batch_size = 64 sigma = 0.01 # 高斯噪声标准差 device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu") env_name = 'Pendulum-v1' env = gym.make(env_name) random.seed(0) np.random.seed(0) env.seed(0) torch.manual_seed(0) replay_buffer = rl_utils.ReplayBuffer(buffer_size) state_dim = env.observation_space.shape[0] action_dim = env.action_space.shape[0] action_bound = env.action_space.high[0] # 动作最大值 agent = DDPG(state_dim, hidden_dim, action_dim, action_bound, sigma, actor_lr, critic_lr, tau, gamma, device) return_list = rl_utils.train_off_policy_agent(env, agent, num_episodes, replay_buffer, minimal_size, batch_size) episodes_list = list(range(len(return_list))) plt.plot(episodes_list, return_list) plt.xlabel('Episodes') plt.ylabel('Returns') plt.title('DDPG on {}'.format(env_name)) plt.show() mv_return = rl_utils.moving_average(return_list, 9) plt.plot(episodes_list, mv_return) plt.xlabel('Episodes') plt.ylabel('Returns') plt.title('DDPG on {}'.format(env_name)) plt.show()结果代码在 pendulum-v1环境下表现尚可，第一次跑了200轮的训练，虽然整体趋势在上升，但是奖励还是比较低的，继续训练直1000轮，可以看到模型的奖励从300轮左右开始就稳定在-200左右，总体表现良好。
创建时间: 2024年11月25日 20:29作者: 蜡笔大新笔记类别: 机器学习标签: 主成分分析, 数据预处理, 线性回归状态: 完成简介主成分分析（英语：Principal components analysis，缩写：PCA）是一种统计分析和简化数据集的方法。它通过正交变换将一系列可能相关的变量观测值进行线性变换，将其投影为一系列线性不相关的变量，这些不相关变量被称为主成分（Principal Components）。如下图所示，PCA首先确定数据的中心点，然后建立并旋转坐标系，目标是将数据点投影到坐标轴上。这个过程类似右图所示，同时要使方差最大化（即使所有点尽可能分散），从而最大程度地减少降维带来的信息损失。$f_1$ 被称为第一主成分，$f_2$ 为第二主成分，依此类推。在PCA中，k表示要保留的主成分数量，每个主成分对应一个维度方向。k值越大，降维程度越小——例如，从100维降至40维比降至2维保留了更多信息。[scode type="blue" size=""]方差（Variance）是衡量数据分散程度的重要指标。在PCA中，我们追求投影后的数据具有最大方差，这意味着：数据点在新坐标轴上分布更分散，保留了更多原始信息较大的方差表示该主成分包含了更多的数据变化信息通常按方差大小排序选择主成分，保留方差贡献最大的k个主成分进行降维数学上，方差可以表示为：Var(X) = E[(X - μ)²] === >= # X为随机变量 # μ为X的期望值 # E为期望运算在实际应用中，我们通常通过计算各个主成分的方差贡献率和累积方差贡献率来确定需要保留的主成分个数。[/scode]具体有关的数学推导可以看这个视频：[collapse status="false" title="PCA讲解"][/collapse]代码本次代码环境是使用PCA的人脸识别链接：人脸识别# coding:utf-8 import os from numpy import * import numpy as np import cv2 import matplotlib.pyplot as plt from pylab import mpl mpl.rcParams['font.sans-serif'] = ['Arial Unicode MS'] # 图片矢量化 def img2vector(image): img = cv2.imread(image, 0) # 读取图片 rows, cols = img.shape imgVector = np.zeros((1, rows * cols)) imgVector = np.reshape(img, (1, rows * cols)) return imgVector orlpath = "./orl_faces" # 读入人脸库,每个人随机选择k张作为训练集,其余构成测试集 def load_orl(k): ''' 对训练数据集进行数组初始化，用0填充，每张图片尺寸都定为112*92, 现在共有40个人，每个人都选择k张，则整个训练集大小为40*k,112*92 ''' train_face = np.zeros((40 * k, 112 * 92)) train_label = np.zeros(40 * k) # [0,0,.....0](共40*k个0) test_face = np.zeros((40 * (10 - k), 112 * 92)) test_label = np.zeros(40 * (10 - k)) # sample=random.sample(range(10),k)#每个人都有的10张照片中，随机选取k张作为训练样本(10个里面随机选取K个成为一个列表) sample = random.permutation(10) + 1 # 随机排序1-10 (0-9）+1 for i in range(40): # 共有40个人 people_num = i + 1 for j in range(10): # 每个人都有10张照片 image = orlpath + '/s' + str(people_num) + '/' + str(sample[j]) + '.pgm' # 读取图片并进行矢量化 img = img2vector(image) if j < k: # 构成训练集 train_face[i * k + j, :] = img train_label[i * k + j] = people_num else: # 构成测试集 test_face[i * (10 - k) + (j - k), :] = img test_label[i * (10 - k) + (j - k)] = people_num return train_face, train_label, test_face, test_label # 定义PCA算法 def PCA(data, r):#降低到r维 data = np.float32(np.mat(data)) rows, cols = np.shape(data) data_mean = np.mean(data, 0) # 对列求平均值 A = data - np.tile(data_mean, (rows, 1)) # 将所有样例减去对应均值得到A C = A * A.T # 得到协方差矩阵 D, V = np.linalg.eig(C) # 求协方差矩阵的特征值和特征向量 V_r = V[:, 0:r] # 按列取前r个特征向量 V_r = A.T * V_r # 小矩阵特征向量向大矩阵特征向量过渡 for i in range(r): V_r[:, i] = V_r[:, i] / np.linalg.norm(V_r[:, i]) # 特征向量归一化 final_data = A * V_r return final_data, data_mean, V_r def face_recongize(): for r in range(10, 41, 10): # 最多降到40维,即选取前40个主成分（因为当k=1时，只有40维) print("当降维到%d时" % (r)) x_value = [] y_value = [] train_face, train_label, test_face, test_label = load_orl(7) # 得到数据集 # 利用PCA算法进行训练 data_train_new, data_mean, V_r = PCA(train_face, r) num_train = data_train_new.shape[0] # 训练脸总数 num_test = test_face.shape[0] # 测试脸总数 temp_face = test_face - np.tile(data_mean, (num_test, 1)) data_test_new = temp_face * V_r # 得到测试脸在特征向量下的数据 data_test_new = np.array(data_test_new) # mat change to array data_train_new = np.array(data_train_new) # 测试准确度 true_num = 0 for i in range(num_test): testFace = data_test_new[i, :] diffMat = data_train_new - np.tile(testFace, (num_train, 1)) # 训练数据与测试脸之间距离 sqDiffMat = diffMat ** 2 sqDistances = sqDiffMat.sum(axis=1) # 按行求和 sortedDistIndicies = sqDistances.argsort() # 对向量从小到大排序，使用的是索引值,得到一个向量 indexMin = sortedDistIndicies[0] # 距离最近的索引 if train_label[indexMin] == test_label[i]: true_num += 1 else: pass accuracy = float(true_num) / num_test x_value.append(7) y_value.append(round(accuracy, 2)) print('当每个人选择%d张照片进行训练时，The classify accuracy is: %.2f%%' % (7, accuracy * 100)) if __name__ == '__main__': face_recongize() 模型的准确率尚可，但是PCA降维不同维度的区别并不大，可能是由于任务较为简单。
创建时间: 2024年11月13日 11:23作者: 蜡笔大新笔记类别: 强化学习标签: 强化学习, 深度强化学习, 策略梯度算法状态: 完成简介之前提到的 DQN 系列算法属于基于值函数的方法来训练模型，而 PG 系列算法则属于基于策略梯度的方法。PG 系列通常采用蒙特卡洛方法计算 G 值，并通过梯度上升更新参数。科学家们考虑将策略与时序差分（TD）方法相融合，使得无需沿着整个路径更新到结束状态再回溯就能更新当前状态。这类似于 TD 算法的思想：边前进边更新 Q 值，同时更新策略。这种方法被称为 Actor-Critic 算法。许多前沿算法都是基于 AC 进行改进的，可以说这种思想既重要又实用。与DQN的结构有些相似，AC算法也拥有两个网络，分别是Actor网络和Critic网络，接下来将分别进行介绍。Actor网络Actor网络负责与环境交互，并在 Critic 价值函数的指导下用策略梯度学习一个更好的策略。Critic网络通过 Actor 与环境交互收集的数据学习一个价值函数，这个价值函数会用于判断在当前状态什么动作是好的，什么动作不是好的，进而帮助 Actor 进行策略更新。而更新的策略则与之前的PG算法很像，都是通过梯度上升算法更新参数。TD-Error时序差分残差（TD-Error）是强化学习中用来衡量当前状态与预期状态的差异的指标。反映了当前状态与预期状态的差异，通常被用来更新策略网络的参数，使得策略能够更快地朝着最优策略的方向学习。$$ \delta_t = r_t + \gamma V(s_{t+1}) - V(s_t) $$其中：$\delta_t$ 是在时间步 t 的TD误差$r_t$ 是在时间步 t 获得的即时奖励$\gamma$ 是折扣因子$V(s_t)$ 是状态 $s_t$ 的估计值函数$V(s_{t+1})$ 是下一个状态 $s_{t+1}$ 的估计值函数[font color="Orange"]TD-Error需要与优势函数做出一定区别。[/font][scode type="blue" size=""]优势函数用于衡量特定动作 a 在某个状态 s 中的好坏程度。给定状态值函数 V(s) 和动作值函数 Q(s, a) ，优势函数定义为：$$ A(s, a) = Q(s, a) - V(s) $$其实很好理解，Q值是执行动作的总奖励期望，而V值是到达状态后的总奖励期望（所有动作Q值的平均数），因此如果A>0，说明这个动作的期望在平均之上；反之则比平均更差。而TD-Error的公式与优势函数其实是可以互相转化的，即 $Q=\gamma * V(S’)+r$ 。但是需要区别的是他们的含义不同。优势函数说明这个动作在该状态下是否比平均策略更优，因此是一个相对的衡量标准。而TD-Error评价当前状态值估计与“实际“观测值的偏差。它更强调一种“矫正”效果，反映当前价值估计的准确性，并直接用于更新价值函数或策略。[/scode]更新TD-Error： $\delta_t = r_t + \gamma V(s_{t+1}) - V(s_t)$ 就可以看作“真实”值和预测值之间的差异：如果 “真实”值 > 预测值（即 $\delta_t$ > 0 ），说明模型低估了当前状态的价值。为了修正这一低估，模型会增加 $V(s_t)$ 的估计值，使得预测更接近真实值。并且差得越多，更新的幅度越大。如果 “真实”值 < 预测值（即 $\delta_t$ < 0 ），则模型高估了当前状态的价值。模型将减小 $V(s_t)$ 的估计值，调低预测以贴合真实值。误差越大，调整幅度也越大。[scode type="blue" size=""]$r_t + \gamma V(s_{t+1})$ 表示更好的价值估计，因为结合了实际观测到的即时奖励和对未来价值的估计，因此实际使用时可以理解为"真实"值。$V(s_t)$ 表示预测值：$V(s_t)$ 是模型根据当前参数对状态 $s_t$ 的价值估计。[/scode]具体而言，损失函数通常使用MSE最小化TD-Error，并通过梯度上升更新Actor的参数，通过梯度下降更新Critic的参数。实际代码中会统一使用梯度下降方便实现（给梯度上升加负号）。流程Actor根据当前状态生成策略。Actor是策略网络，它基于当前状态 $s_t$ 生成一个策略 $\pi(a|s_t)$ 。通常，策略会给出每个动作的概率分布。Actor可以：选择概率最高的动作（贪婪策略）根据概率分布随机采样一个动作（通常是探索时使用的方式）执行动作，获取奖励和下一个状态。从环境中执行Actor选出的动作 $a_t$ ，得到即时奖励 $r_t$ 和下一个状态 $s_{t+1}$ 。Critic计算价值估计并生成TD-Error。Critic是价值网络，通常估计状态值 $V(s)$ 。Critic网络基于当前状态 $s_t$ 和下一个状态 $s_{t+1}$ ，计算TD-Error。Critic更新（梯度下降）。Critic通过最小化TD-Error来更新其网络参数，即采用梯度下降更新，使得估计的状态值 $ V(s)$ 更接近真实回报。Actor更新（梯度上升）。Actor利用TD-Error来更新策略。TD-Error $\delta_t$ 作为一种优势信号，指导Actor优化选择动作的概率，使得策略朝着获得更高回报的方向改进。代码import gym import torch import torch.nn.functional as F import matplotlib.pyplot as plt import rl_utils class PolicyNet(torch.nn.Module): def __init__(self, state_dim, hidden_dim, action_dim): super(PolicyNet, self).__init__() self.fc1 = torch.nn.Linear(state_dim, hidden_dim) self.fc2 = torch.nn.Linear(hidden_dim, action_dim) def forward(self, x): x = F.relu(self.fc1(x)) return F.softmax(self.fc2(x), dim=1) class ValueNet(torch.nn.Module): def __init__(self, state_dim, hidden_dim): super(ValueNet, self).__init__() self.fc1 = torch.nn.Linear(state_dim, hidden_dim) self.fc2 = torch.nn.Linear(hidden_dim, 1) def forward(self, x): x = F.relu(self.fc1(x)) return self.fc2(x) class ActorCritic: def __init__(self, state_dim, hidden_dim, action_dim, actor_lr, critic_lr, gamma, device): # 策略网络 self.actor = PolicyNet(state_dim, hidden_dim, action_dim).to(device) self.critic = ValueNet(state_dim, hidden_dim).to(device) # 价值网络 # 策略网络优化器 self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=actor_lr) self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=critic_lr) # 价值网络优化器 self.gamma = gamma self.device = device def take_action(self, state): state = torch.tensor([state], dtype=torch.float).to(self.device) probs = self.actor(state) action_dist = torch.distributions.Categorical(probs) action = action_dist.sample() return action.item() def update(self, transition_dict): states = torch.tensor(transition_dict['states'], dtype=torch.float).to(self.device) actions = torch.tensor(transition_dict['actions']).view(-1, 1).to(self.device) rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1).to(self.device) next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float).to(self.device) dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1, 1).to(self.device) # 时序差分目标 td_target = rewards + self.gamma * self.critic(next_states) * (1 - dones) td_delta = td_target - self.critic(states) # 时序差分误差 log_probs = torch.log(self.actor(states).gather(1, actions)) actor_loss = torch.mean(-log_probs * td_delta.detach()) # 均方误差损失函数 critic_loss = torch.mean(F.mse_loss(self.critic(states), td_target.detach())) self.actor_optimizer.zero_grad() self.critic_optimizer.zero_grad() actor_loss.backward() # 计算策略网络的梯度 critic_loss.backward() # 计算价值网络的梯度 self.actor_optimizer.step() # 更新策略网络的参数 self.critic_optimizer.step() # 更新价值网络的参数 actor_lr = 1e-3 critic_lr = 1e-2 num_episodes = 1000 hidden_dim = 128 gamma = 0.98 device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu") env_name = 'CartPole-v0' env = gym.make(env_name) env.seed(0) torch.manual_seed(0) state_dim = env.observation_space.shape[0] action_dim = env.action_space.n agent = ActorCritic(state_dim, hidden_dim, action_dim, actor_lr, critic_lr, gamma, device) return_list = rl_utils.train_on_policy_agent(env, agent, num_episodes) episodes_list = list(range(len(return_list))) plt.plot(episodes_list, return_list) plt.xlabel('Episodes') plt.ylabel('Returns') plt.title('Actor-Critic on {}'.format(env_name)) plt.show() mv_return = rl_utils.moving_average(return_list, 9) plt.plot(episodes_list, mv_return) plt.xlabel('Episodes') plt.ylabel('Returns') plt.title('Actor-Critic on {}'.format(env_name)) plt.show() 结果可以看出，结果还是非常不错的，相较于PG算法，震荡幅度较小，比较稳定。
创建时间: 2024年11月18日 13:20作者: 蜡笔大新笔记类别: 杂项标签: Data Preprocessing, Machine Learning Techniques状态: 完成问题复现目前的写作方式还是主要集中在notion中，即在notion中写完全稿再导出到自己的博客网站，同时将图片等上传到云服务器图床。然而在编辑时却偶然发现有几张png图片上传到图床后，仍然无法打开。开始以为是Picgo出bug了，于是直接登陆云服务器后台，上传png图像，发现仍然无法打开。问题分析于是认为是文件格式出了问题，可能是不兼容，在本地查看了详细信息，基本确定了应该是此处“文稿类型：TIFF图像”导致的错误，即图床无法兼容这种格式。此时非常疑惑，图像明明是png格式，为什么打开却显示是tiff。经过资料查阅：一些常见的图像格式（如JPEG、PNG、TIFF）之间可能会有较好的兼容性，现代的图像查看器和操作系统确实能够识别文件的实际格式，而不仅仅依赖于文件扩展名。也就是说，其实这张图片是一张tiff图像，但是因为某些错误原因，被转换成png后缀了，因此无法被图床识别展示，但是由于现代操作系统通常能识别实际格式并展示，所以在本地可以正常预览。继续深究背后的原因，即为什么会出现tiff，以及为什么tiff错误转换成png。图片来源这张图片是通过plt库生成的奖励图，在图片上直接右键复制并且粘贴到notion中。然后导出的时候所有的图片都保存为png放在了一个文件夹。在notion中直接查看原始图片后，浏览器中无法预览，说明这张图片在notion中已经是tiff格式了。为了验证这个pycharm中copy的图片属性，做了几个实验：将图片copy后，通过picgo上传，发现可以正常上传并显示，说明图片是正常png。将图片save as image，本地检查后发现是正常png。将图片copy后，粘贴到微信并发送出去，发现通过微信发的图片都会自动转化成jpg。综上，不难发现，pycharm中的图片是正常的png，但是由于未知理由，粘贴到notion中，就会被转成tiff，但是最终导出规则是统一的png，因此造成了现在的现象。而平时电脑的截图，粘贴到notion中，却是正常的。解决方案针对后缀是png的tiff图像，解决方案是将后缀直接改为.tiff，然后通过python代码或者在线网站转换成正常png。以下是一个使用Python将TIFF图像转换为PNG的示例代码：from PIL import Image def tiff2png(tiff_path, png_path): # 打开TIFF图像 with Image.open(tiff_path) as img: # 保存为PNG img.save(png_path, 'PNG') # 使用示例 input_file = './data/image.tiff' output_file = './data/image.png' tiff2png(input_file, output_file) print(f"已将 {input_file} 转换为 {output_file}")另一种更直接的方法就是不要将plt生成的图像直接复制粘贴到notion，而是可以截图/下载/使用微信等第三方工具转一下，这样从根本上避免了tiff图像的产生。拓展：常见图片格式以下是几种常见图片格式的简介：JPG/JPEG (Joint Photographic Experts Group)广泛应用于数码摄影和网络图像。采用有损压缩技术，在文件大小和图像质量之间取得平衡。主要优势包括：高兼容性快速传输小内存占用特别适合照片和复杂图像的存储。作为有损压缩方案，能在保持较高图片质量的同时，将文件大小压缩至原图的约5%。PNG (Portable Network Graphics)支持无损压缩和透明背景，适用于需要保持高质量或包含文本、线条艺术的图像。特点包括：无损压缩算法（源自LZ77）支持透明背景适合渐变图像制作在网页设计中广泛使用，也是Photoshop、InDesign等主流图像处理软件的首选输出和编辑格式。PNG格式对图片本身质量的损失极小，同时具有高压缩比，能生成体积较小的文件。GIF (Graphics Interchange Format)支持动画和有限的256色调色板。适合简单的动画和图标，但不适合照片等复杂图像。GIF格式分为静态和动态两种：静态GIF：类似于JPG格式的图片动态GIF：由多幅图片组成，形成动画效果由于色彩限制，GIF格式很少适用于静态图像处理工作。在PS或Keynote中导入GIF格式图片时，图片质量通常不尽如人意。TIFF (Tagged Image File Format)高质量图像格式，支持有损和无损压缩。常用于印刷行业和专业摄影，文件通常较大。BMP (Bitmap Image File)无压缩的图像格式，保留所有图像数据。文件较大，但图像质量高，常用于Windows系统。WebP由Google开发的现代图像格式，提供优秀的压缩率和图像质量。支持有损和无损压缩，以及动画和透明度。SVG (Scalable Vector Graphics)基于XML的矢量图形格式。可以无损缩放，适合图标、徽标等需要在不同尺寸下保持清晰的图像。选择合适的图像格式取决于具体用途，如网页展示、打印、存档等，以及对文件大小、图像质量和兼容性的要求。
介绍根据PyTorch在2022年推出的文章来看，M系列芯片的mac已经可以使用GPU加速训练。pytorch使用起来也非常方便，和cuda类似：torch.backends.mps.is_available()torch.device("mps")实验使用PyTorch官方的MNIST例子，即手写数字识别。对比了一下不同系统下CPU和GPU加速的对比效果。实验均采用Batch=64，Epoch=5的参数。实验一（Mac M4）芯片：M4实验二（Mac M1）芯片：M1实验三（Win）CPU：AMD R7-5800HGPU：RTX3060 Laptop 130w结论通过实验可以看出，苹果的M4芯片相较于M1芯片进步异常明显，使用GPU（mps）加速可以非常显著的提升训练和测试的速度。而相较于笔记本的3060显卡，mps加速竟然比cuda加速还要快。但是，由于是PyTorch官方推出的案例，难免会进行特殊优化。在强化学习的学习中，并没有发现mps比cpu快很多，甚至还不如cpu正常跑，可能与批次等参数的设置也有关，有待后续继续实验。代码import argparse import time import torch import torch.nn as nn import torch.nn.functional as F import torch.optim as optim from torchvision import datasets, transforms from torch.optim.lr_scheduler import StepLR import matplotlib.pyplot as plt class Net(nn.Module): def __init__(self): super(Net, self).__init__() self.conv1 = nn.Conv2d(1, 32, 3, 1) self.conv2 = nn.Conv2d(32, 64, 3, 1) self.dropout1 = nn.Dropout(0.25) self.dropout2 = nn.Dropout(0.5) self.fc1 = nn.Linear(9216, 128) self.fc2 = nn.Linear(128, 10) def forward(self, x): x = self.conv1(x) x = F.relu(x) x = self.conv2(x) x = F.relu(x) x = F.max_pool2d(x, 2) x = self.dropout1(x) x = torch.flatten(x, 1) x = self.fc1(x) x = F.relu(x) x = self.dropout2(x) x = self.fc2(x) output = F.log_softmax(x, dim=1) return output def train(model, device, train_loader, optimizer, epoch): model.train() for batch_idx, (data, target) in enumerate(train_loader): data, target = data.to(device), target.to(device) optimizer.zero_grad() output = model(data) loss = F.nll_loss(output, target) loss.backward() optimizer.step() def test(model, device, test_loader): model.eval() test_loss = 0 correct = 0 with torch.no_grad(): for data, target in test_loader: data, target = data.to(device), target.to(device) output = model(data) test_loss += F.nll_loss(output, target, reduction='sum').item() pred = output.argmax(dim=1, keepdim=True) correct += pred.eq(target.view_as(pred)).sum().item() test_loss /= len(test_loader.dataset) accuracy = 100. * correct / len(test_loader.dataset) return test_loss, accuracy def measure_time(device_name, device, train_loader, test_loader, args): model = Net().to(device) optimizer = optim.Adadelta(model.parameters(), lr=args.lr) scheduler = StepLR(optimizer, step_size=1, gamma=args.gamma) train_times, test_times = [], [] for epoch in range(1, args.epochs + 1): # Training time start_train = time.time() train(model, device, train_loader, optimizer, epoch) end_train = time.time() train_times.append(end_train - start_train) # Testing time start_test = time.time() test_loss, accuracy = test(model, device, test_loader) end_test = time.time() test_times.append(end_test - start_test) scheduler.step() print(f"{device_name} - Epoch {epoch}, Train Time: {train_times[-1]:.4f}s, Test Time: {test_times[-1]:.4f}s, " f"Test Loss: {test_loss:.4f}, Accuracy: {accuracy:.2f}%") return train_times, test_times def plot_times(cpu_times, mps_times, cuda_times, args): epochs = range(1, args.epochs + 1) plt.figure(figsize=(10, 5)) # Plot CPU, MPS (if available), and CUDA (if available) times plt.plot(epochs, cpu_times[0], label='CPU Train Time') plt.plot(epochs, cpu_times[1], label='CPU Test Time') if mps_times: plt.plot(epochs, mps_times[0], label='MPS Train Time') plt.plot(epochs, mps_times[1], label='MPS Test Time') if cuda_times: plt.plot(epochs, cuda_times[0], label='CUDA Train Time') plt.plot(epochs, cuda_times[1], label='CUDA Test Time') plt.xlabel('Epoch') plt.ylabel('Time (s)') plt.title('CPU vs MPS vs CUDA Training and Testing Time') plt.legend() plt.show() def main(): parser = argparse.ArgumentParser(description='PyTorch MNIST Example with CPU, MPS, and CUDA Comparison') parser.add_argument('--batch-size', type=int, default=64) parser.add_argument('--test-batch-size', type=int, default=1000) parser.add_argument('--epochs', type=int, default=5) parser.add_argument('--lr', type=float, default=1.0) parser.add_argument('--gamma', type=float, default=0.7) args = parser.parse_args() torch.manual_seed(1) transform = transforms.Compose([ transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,)) ]) dataset1 = datasets.MNIST('./data', train=True, download=True, transform=transform) dataset2 = datasets.MNIST('./data', train=False, transform=transform) train_loader = torch.utils.data.DataLoader(dataset1, batch_size=args.batch_size, shuffle=True) test_loader = torch.utils.data.DataLoader(dataset2, batch_size=args.test_batch_size, shuffle=False) # Measure CPU time cpu_times = measure_time("CPU", torch.device("cpu"), train_loader, test_loader, args) # Measure MPS time if available mps_times = None if torch.backends.mps.is_available(): mps_times = measure_time("MPS", torch.device("mps"), train_loader, test_loader, args) # Measure CUDA time if available cuda_times = None if torch.cuda.is_available(): cuda_times = measure_time("CUDA", torch.device("cuda"), train_loader, test_loader, args) # Plot results plot_times(cpu_times, mps_times, cuda_times, args) if __name__ == '__main__': main()
创建时间: 2024年11月5日 16:51作者: 蜡笔大新笔记类别: 强化学习标签: Deep Reinforcement Learning, Policy Optimization, Reinforcement Learning状态: 完成基本原理PG算法，全称为策略梯度算法（Policy Gradient Algorithm），是强化学习中的一种重要方法。它直接对策略进行优化，而不是通过值函数间接优化策略。这种方法在连续动作空间和高维动作空间中特别有效。PG算法的核心思想是通过梯度上升的方式直接优化策略函数。它通过估计策略梯度来更新策略参数，使得期望回报最大化。这种方法允许智能体在每个状态下学习一个概率分布，用于选择最优动作。之前提到的 Sarsa、Q-Learning 和 DQN 算法都是基于价值的方法，也就是先计算每个状态对应的动作的 Q 值，再选择 Q 值最大的动作执行。而 Policy Gradient 是一种更加直接的方式，它直接计算每个状态对应的动作的概率。这样做的优点就是它可以在一个连续区间内选取动作。如果用一句话来表达策略梯度的直观解释，那就是“如果动作使得最终回报变大，那么增加这个动作出现的概率，反之，减少这个动作出现的概率”。这句话表达了两个含义：我们考虑的是动作对于回报的影响，没有考虑状态或者其他因素。我们调整的是动作出现的概率，而没有给某个动作打分，这区别于Value-based类的算法。理解PG算法不同于之前提到的基于价值类的算法，PG更多需要从整体和宏观网络的角度去理解。不再通过每个状态-动作对去更新，而是类似蒙特卡洛的思想，通过大量实验去逼近最终结果。总体的思路是：从初始状态开始，按照当前策略选择动作，直到到达终止状态。这构成一条完整的轨迹。从终止状态往回计算总的奖励和，作为这条路径的总奖励。当完成了一定数量的路径后，将批量投入训练，具体则是梯度上升算法更新参数 $\theta$ ，目标是使总的期望奖励函数变得最大，从而得到最优策略。我们将策略学习的目标函数定义为：$$ J(\theta) = \mathbb{E}_{s_0}[V^{\pi_\theta}(s_0)] $$核心公式$$ \begin{align*}\nabla_\theta J(\theta) &\propto \sum_{s \in S} v^{\pi_\theta}(s) \sum_{a \in A} Q^{\pi_\theta}(s, a) \nabla_\theta \pi_\theta(a|s) \\&= \sum_{s \in S} v^{\pi_\theta}(s) \sum_{a \in A} \pi_\theta(a|s) Q^{\pi_\theta}(s, a) \frac{\nabla_\theta \pi_\theta(a|s)}{\pi_\theta(a|s)} \\&= \mathbb{E}_{\pi_\theta} [Q^{\pi_\theta}(s, a) \nabla_\theta \log \pi_\theta(a|s)]\end{align*} $$解释：$\nabla_\theta J(\theta)$：这是目标函数$J(θ)$关于策略参数$θ$的梯度。$v^{\pi_\theta}(s)$：这是在状态$s$下，按照策略$π_θ$行动的值函数。$\nabla_\theta \pi_\theta(a|s)$：这是策略$π_θ$在状态$s$下选择动作$a$的概率关于参数$θ$的梯度。$\pi_\theta(a|s)$：这是在状态$s$下选择动作$a$的概率。需要注意的是，因为上式中期望 $\mathbb{E}$ 的下标是 $\pi_\theta$ ，所以策略梯度算法为在线策略（on-policy）算法，即必须使用当前策略采样得到的数据来计算梯度。直观理解一下策略梯度这个公式，可以发现在每一个状态下，Q值越高则整个期望越大，因此可以认为梯度的修改是让策略更多地去采样到带来较高Q值的动作，更少地去采样到带来较低Q值的动作。优势PG算法具有以下优势：适用于连续动作空间：不需要离散化动作空间，可以直接在连续空间中操作。可以学习随机策略：这在某些环境中非常有用，特别是在部分可观察环境中。收敛性好：在理论上，PG算法可以保证收敛到至少一个局部最优解。代码在CartPole-v0环境中，使用梯度策略方法中最基本的算法之一的REINFORCE算法训练模型。import gym import torch import torch.nn.functional as F import numpy as np import matplotlib.pyplot as plt from tqdm import tqdm import rl_utils class PolicyNet(torch.nn.Module): def __init__(self, state_dim, hidden_dim, action_dim): super(PolicyNet, self).__init__() self.fc1 = torch.nn.Linear(state_dim, hidden_dim) self.fc2 = torch.nn.Linear(hidden_dim, action_dim) def forward(self, x): x = F.relu(self.fc1(x)) return F.softmax(self.fc2(x), dim=1) class REINFORCE: def __init__(self, state_dim, hidden_dim, action_dim, learning_rate, gamma, device): self.policy_net = PolicyNet(state_dim, hidden_dim, action_dim).to(device) self.optimizer = torch.optim.Adam(self.policy_net.parameters(), lr=learning_rate) # 使用Adam优化器 self.gamma = gamma # 折扣因子 self.device = device def take_action(self, state): # 根据动作概率分布随机采样 state = torch.tensor([state], dtype=torch.float).to(self.device) probs = self.policy_net(state) action_dist = torch.distributions.Categorical(probs) action = action_dist.sample() return action.item() def update(self, transition_dict): reward_list = transition_dict['rewards'] state_list = transition_dict['states'] action_list = transition_dict['actions'] G = 0 self.optimizer.zero_grad() for i in reversed(range(len(reward_list))): # 从最后一步算起 reward = reward_list[i] state = torch.tensor([state_list[i]], dtype=torch.float).to(self.device) action = torch.tensor([action_list[i]]).view(-1, 1).to(self.device) log_prob = torch.log(self.policy_net(state).gather(1, action)) G = self.gamma * G + reward loss = -log_prob * G # 每一步的损失函数 loss.backward() # 反向传播计算梯度 self.optimizer.step() # 梯度下降 learning_rate = 1e-3 num_episodes = 1000 hidden_dim = 128 gamma = 0.98 device = torch.device("cuda") if torch.cuda.is_available() else torch.device( "cpu") env_name = "CartPole-v0" env = gym.make(env_name) env.seed(0) torch.manual_seed(0) state_dim = env.observation_space.shape[0] action_dim = env.action_space.n agent = REINFORCE(state_dim, hidden_dim, action_dim, learning_rate, gamma, device) return_list = [] for i in range(10): with tqdm(total=int(num_episodes / 10), desc='Iteration %d' % i) as pbar: for i_episode in range(int(num_episodes / 10)): episode_return = 0 transition_dict = { 'states': [], 'actions': [], 'next_states': [], 'rewards': [], 'dones': [] } state = env.reset() done = False while not done: action = agent.take_action(state) next_state, reward, done, _ = env.step(action) transition_dict['states'].append(state) transition_dict['actions'].append(action) transition_dict['next_states'].append(next_state) transition_dict['rewards'].append(reward) transition_dict['dones'].append(done) state = next_state episode_return += reward return_list.append(episode_return) agent.update(transition_dict) if (i_episode + 1) % 10 == 0: pbar.set_postfix({ 'episode': '%d' % (num_episodes / 10 * i + i_episode + 1), 'return': '%.3f' % np.mean(return_list[-10:]) }) pbar.update(1) episodes_list = list(range(len(return_list))) plt.plot(episodes_list, return_list) plt.xlabel('Episodes') plt.ylabel('Returns') plt.title('REINFORCE on {}'.format(env_name)) plt.show() mv_return = rl_utils.moving_average(return_list, 9) plt.plot(episodes_list, mv_return) plt.xlabel('Episodes') plt.ylabel('Returns') plt.title('REINFORCE on {}'.format(env_name)) plt.show()其中让我感到困惑的是update函数：def update(self, transition_dict): reward_list = transition_dict['rewards'] state_list = transition_dict['states'] action_list = transition_dict['actions'] G = 0 self.optimizer.zero_grad() for i in reversed(range(len(reward_list))): # 从最后一步算起 reward = reward_list[i] state = torch.tensor([state_list[i]], dtype=torch.float).to(self.device) action = torch.tensor([action_list[i]]).view(-1, 1).to(self.device) log_prob = torch.log(self.policy_net(state).gather(1, action)) G = self.gamma * G + reward loss = -log_prob * G # 每一步的损失函数 loss.backward() # 反向传播计算梯度 self.optimizer.step() # 梯度下降明明PG算法使用的是梯度上升算法，为什么这里却是梯度下降。经过资料查阅，我了解到这其实是深度学习中一种常见的做法，loss = -log_prob * G 这一步其实就是在对损失函数取反，再进行梯度下降实际就相当于原来的梯度向上了，这样做法的原因在于很多优化器默认执行梯度下降，因此更加便捷使用标准优化器来实现梯度上升。除了符号，还要乘对数概率log_prob的原因是：它可以将乘法转换为加法，简化计算它有助于数值稳定性，特别是在处理很小的概率时它的梯度（即对数似然梯度）在数学上有良好的性质结果可以看出其实随着训练轮数的增长，回报有明显的增长，并且能够达到奖励和200的上限。
创建时间: 2024年11月4日 21:05作者: 蜡笔大新笔记类别: 强化学习标签: Deep Reinforcement Learning, Policy Optimization, Q-Learning状态: 完成简介Double DQN是深度强化学习中的一种改进算法，它解决了传统DQN（深度Q网络）中的过估计问题。传统DQN在某些情况下会高估Q值，导致学习效果不佳。Double DQN通过引入两个网络来缓解这个问题。原理使用两个结构相同但参数不同的神经网络：当前网络和目标网络。[font color="Red"]当前网络用于选择动作，目标网络用于评估动作的价值。[/font]这种分离减少了过度乐观估计的可能性。步骤使用当前网络，输入当前状态 $s_t$ 根据贪婪策略获得动作 $a_t$ 和对应的Q值 $Q_1$当前状态 $s_t$ 根据选择的动作 $a_t$ 到达下一个状态 $s_{t+1}$将下一个状态 $s_{t+1}$ 输入当前网络，选择最大Q值的动作 $a'$将下一个状态 $s_{t+1}$ 输入目标网络，获得动作 $a'$ 对应的Q值 $Q_2$将 $Q_1$ 和 $Q_2$ 计算TD-Error，通过梯度下降法更新当前网络一段时间后，目标网络慢慢向当前网络靠拢区别在普通DQN中，当前网络和目标网络更像是两种并行的网络，目标网络只是当前网络参数较为稳定的版本，用于减缓目标漂移；而在Double DQN中，它们更像是互相合作的队友。当前网络负责提供动作，目标网络负责评估这个动作的质量，并反馈给当前网络。当前网络当前网络和普通DQN一样生成当前状态的不同动作的Q值，通过贪婪策略选择动作（大概率Q值最大，小概率随机探索）。当前网络比普通DQN多了一步，将下一个状态输入进去，得到包含所有动作的Q值向量，选取下一个状态Q值最大的最佳动作。目标网络它不再直接通过贝尔曼公式计算当前状态的最大Q值，而是对当前网络选择的动作进行评估。具体而言，将下一状态输入目标网络，使用目标网络生成的Q值中，与当前网络选择的动作对应的Q值作为 $Q{\prime}$ （目标Q值）。通过计算TD误差（TD-Error）评估模型的质量，优化目标是最小化TD误差，并通过反向传播更新当前网络的参数。优点Double DQN的优点主要体现在以下几个方面：更准确的Q值估计：通过分离动作选择和评估，有效减少了Q值的过估计问题。提高学习稳定性：两个网络的协作机制使得学习过程更加稳定，减少了价值估计的波动。更好的泛化能力：由于估计更准确，Double DQN在面对新的或未见过的状态时表现更好。更快的收敛速度：相比传统DQN，Double DQN通常能更快地达到最优策略。在复杂环境中表现出色：特别是在状态空间较大、动作选择复杂的环境中，Double DQN往往能取得更好的效果。代码环境环境是倒立摆（Inverted Pendulum），在gym中对应的版本是Pendulum-v1。力矩大小是在[-2,2]范围内的连续值。由于DQN只能处理离散动作环境,因此我们无法直接用 DQN 来处理倒立摆环境,但倒立摆环境可以比较方便地验证 DQN 对Q值的过高估计：倒立摆环境下 Q值的最大估计应为0(倒立摆向上保持直立时能选取的最大值),Q值出现大于0的情况则说明出现了过高估计。为了能够应用DQN,我们采用离散化动作的技巧。例如,下面的代码将连续的动作空间离散为11个动作。动作 [0,1,2,...,9,10]分别代表力矩为[-2,-1.6, 1.2,..., 1.2, 1.6, 2]。Double_DQN与DQN区别不大，主要改变的地方在于：if self.dqn_type == 'DoubleDQN': # DQN与Double DQN的区别 max_action = self.q_net(next_states).max(1)[1].view(-1, 1) max_next_q_values = self.target_q_net(next_states).gather(1, max_action) else: # DQN的情况 max_next_q_values = self.target_q_net(next_states).max(1)[0].view(-1, 1)总代码：import random import gym import numpy as np import torch import torch.nn.functional as F import matplotlib.pyplot as plt import rl_utils from tqdm import tqdm class Qnet(torch.nn.Module): """ 只有一层隐藏层的Q网络 """ def __init__(self, state_dim, hidden_dim, action_dim): super(Qnet, self).__init__() self.fc1 = torch.nn.Linear(state_dim, hidden_dim) self.fc2 = torch.nn.Linear(hidden_dim, action_dim) def forward(self, x): x = F.relu(self.fc1(x)) return self.fc2(x) class DQN: """ DQN算法,包括Double DQN """ def __init__(self, state_dim, hidden_dim, action_dim, learning_rate, gamma, epsilon, target_update, device, dqn_type='VanillaDQN'): self.action_dim = action_dim self.q_net = Qnet(state_dim, hidden_dim, self.action_dim).to(device) self.target_q_net = Qnet(state_dim, hidden_dim, self.action_dim).to(device) self.optimizer = torch.optim.Adam(self.q_net.parameters(), lr=learning_rate) self.gamma = gamma self.epsilon = epsilon self.target_update = target_update self.count = 0 self.dqn_type = dqn_type self.device = device def take_action(self, state): if np.random.random() < self.epsilon: action = np.random.randint(self.action_dim) else: state = torch.tensor([state], dtype=torch.float).to(self.device) action = self.q_net(state).argmax().item() return action def max_q_value(self, state): state = torch.tensor([state], dtype=torch.float).to(self.device) return self.q_net(state).max().item() def update(self, transition_dict): states = torch.tensor(transition_dict['states'], dtype=torch.float).to(self.device) actions = torch.tensor(transition_dict['actions']).view(-1, 1).to(self.device) rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1).to(self.device) next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float).to(self.device) dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1, 1).to(self.device) q_values = self.q_net(states).gather(1, actions) # Q值 # 下个状态的最大Q值 if self.dqn_type == 'DoubleDQN': # DQN与Double DQN的区别 max_action = self.q_net(next_states).max(1)[1].view(-1, 1) max_next_q_values = self.target_q_net(next_states).gather(1, max_action) else: # DQN的情况 max_next_q_values = self.target_q_net(next_states).max(1)[0].view(-1, 1) q_targets = rewards + self.gamma * max_next_q_values * (1 - dones) # TD误差目标 dqn_loss = torch.mean(F.mse_loss(q_values, q_targets)) # 均方误差损失函数 self.optimizer.zero_grad() # PyTorch中默认梯度会累积,这里需要显式将梯度置为0 dqn_loss.backward() # 反向传播更新参数 self.optimizer.step() if self.count % self.target_update == 0: self.target_q_net.load_state_dict(self.q_net.state_dict()) # 更新目标网络 self.count += 1 lr = 1e-2 num_episodes = 200 hidden_dim = 128 gamma = 0.98 epsilon = 0.01 target_update = 50 buffer_size = 5000 minimal_size = 1000 batch_size = 64 device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu") env_name = 'Pendulum-v1' env = gym.make(env_name) state_dim = env.observation_space.shape[0] action_dim = 11 # 将连续动作分成11个离散动作 def dis_to_con(discrete_action, env, action_dim): # 离散动作转回连续的函数 action_lowbound = env.action_space.low[0] # 连续动作的最小值 action_upbound = env.action_space.high[0] # 连续动作的最大值 return action_lowbound + (discrete_action / (action_dim - 1)) * (action_upbound - action_lowbound) def train_DQN(agent, env, num_episodes, replay_buffer, minimal_size, batch_size): return_list = [] max_q_value_list = [] max_q_value = 0 for i in range(10): with tqdm(total=int(num_episodes / 10), desc='Iteration %d' % i) as pbar: for i_episode in range(int(num_episodes / 10)): episode_return = 0 state = env.reset() done = False while not done: action = agent.take_action(state) max_q_value = agent.max_q_value(state) * 0.005 + max_q_value * 0.995 # 平滑处理 max_q_value_list.append(max_q_value) # 保存每个状态的最大Q值 action_continuous = dis_to_con(action, env, agent.action_dim) next_state, reward, done, _ = env.step([action_continuous]) replay_buffer.add(state, action, reward, next_state, done) state = next_state episode_return += reward if replay_buffer.size() > minimal_size: b_s, b_a, b_r, b_ns, b_d = replay_buffer.sample(batch_size) transition_dict = { 'states': b_s, 'actions': b_a, 'next_states': b_ns, 'rewards': b_r, 'dones': b_d } agent.update(transition_dict) return_list.append(episode_return) if (i_episode + 1) % 10 == 0: pbar.set_postfix({ 'episode': '%d' % (num_episodes / 10 * i + i_episode + 1), 'return': '%.3f' % np.mean(return_list[-10:]) }) pbar.update(1) return return_list, max_q_value_list random.seed(0) np.random.seed(0) env.seed(0) torch.manual_seed(0) replay_buffer = rl_utils.ReplayBuffer(buffer_size) agent = DQN(state_dim, hidden_dim, action_dim, lr, gamma, epsilon, target_update, device) return_list, max_q_value_list = train_DQN(agent, env, num_episodes, replay_buffer, minimal_size, batch_size) episodes_list = list(range(len(return_list))) mv_return = rl_utils.moving_average(return_list, 5) plt.plot(episodes_list, mv_return) plt.xlabel('Episodes') plt.ylabel('Returns') plt.title('DQN on {}'.format(env_name)) plt.show() frames_list = list(range(len(max_q_value_list))) plt.plot(frames_list, max_q_value_list) plt.axhline(0, c='orange', ls='--') plt.axhline(10, c='red', ls='--') plt.xlabel('Frames') plt.ylabel('Q value') plt.title('DQN on {}'.format(env_name)) plt.show() random.seed(0) np.random.seed(0) env.seed(0) torch.manual_seed(0) replay_buffer = rl_utils.ReplayBuffer(buffer_size) agent = DQN(state_dim, hidden_dim, action_dim, lr, gamma, epsilon, target_update, device, 'DoubleDQN') return_list, max_q_value_list = train_DQN(agent, env, num_episodes, replay_buffer, minimal_size, batch_size) episodes_list = list(range(len(return_list))) mv_return = rl_utils.moving_average(return_list, 5) plt.plot(episodes_list, mv_return) plt.xlabel('Episodes') plt.ylabel('Returns') plt.title('Double DQN on {}'.format(env_name)) plt.show() frames_list = list(range(len(max_q_value_list))) plt.plot(frames_list, max_q_value_list) plt.axhline(0, c='orange', ls='--') plt.axhline(10, c='red', ls='--') plt.xlabel('Frames') plt.ylabel('Q value') plt.title('Double DQN on {}'.format(env_name)) plt.show() 100轮结果对比总体奖励回报[column][block size="45%"][/block][block size="45%"][/block][/column]预估Q值[column][block size="45%"][/block][block size="45%"][/block][/column]总体奖励回报上，DQN和Double DQN都在训练中提升了表现，总体奖励相差不大。而在预估Q值时，DQN明显表现出了过度估计Q值（理论上不会出现正的Q值），Double DQN能够更准确的估计Q值。1000轮结果对比总体奖励回报[column][block size="45%"][/block][block size="45%"][/block][/column]预估Q值[column][block size="45%"][/block][block size="45%"][/block][/column]实验结果显示，在这个游戏中，两种网络获得的奖励基本持平，但DQN表现更稳定。在Q值预估方面，虽然两种网络都出现了大于0的情况，DQN明显更频繁地过高估计Q值，而Double DQN在一定程度上缓解了这个问题。值得注意的是，为了使两种DQN能够运行，原本的连续动作被改为离散动作，这可能对实验结果产生了一定影响。
创建时间: 2024年11月4日 19:55作者: 蜡笔大新笔记类别: 机器学习标签: 分类数据编码, 数据预处理, 机器学习技术状态: 完成One-Hot 编码简介One-Hot 编码是一种常用的数据预处理技术，主要用于处理分类变量。这种编码方法将每个类别变量转换为一个二进制向量，其中只有一个元素为1（表示该类别），其余元素均为0。工作原理One-Hot 编码的基本步骤如下：确定所有可能的类别为每个类别创建一个新的二进制列在相应的列中标记1，其他列标记0示例假设我们有一个"颜色"特征，包含"红"、"绿"、"蓝"三种类别：import pandas as pd # 创建一个包含标签的数据集 data = pd.DataFrame({'color': ['red', 'green', 'blue', 'blue', 'red']}) # 对 'color' 列进行独热编码 one_hot_encoded = pd.get_dummies(data['color']) print(one_hot_encoded)结果：Indexbluegreenred00011010210031004001优点能够处理非连续型数值特征，也就是离散值。在一定程度上也起到了扩充特征的作用。它的值只有0和1，不同的类型存储在垂直的空间。比如性别本身是一个特征，经过one hot编码以后，就变成了男或女两个特征，将离散特征通过one-hot编码映射到欧式空间，在回归，分类，聚类等机器学习算法中，特征之间距离的计算或相似度的计算是非常重要的，而我们常用的距离或相似度的计算都是在欧式空间的相似度计算，计算余弦相似性，基于的就是欧式空间。缺点当类别的数量很多时，特征空间会变得非常大，成为一个高维稀疏矩阵。在这种情况下，一般可以用PCA来减少维度。而且one hot encoding+PCA这种组合在实际中也非常有用。应用场景One-Hot 编码广泛应用于：自然语言处理推荐系统图像分类其他涉及分类变量的机器学习任务实现方法在实际应用中，可以使用各种库来实现One-Hot编码，例如：Pandas的get_dummies()函数Scikit-learn的OneHotEncoder类
创建时间: 2024年10月28日 18:45作者: 蜡笔大新笔记类别: 机器学习状态: 完成介绍蒙特卡洛方法（Monte Carlo Method）是一类基于随机抽样的数值计算方法，广泛用于模拟和优化问题。其核心思想是通过模拟大量随机样本，计算这些样本的统计特征来近似真实结果。例子求下图$[a,b]$定积分，即 $F(x)=\int_{a}^{b}f(x)dx$。可以随机取4个x，求他们矩形的面积求平均值。即$$ \begin{align*} S &= \frac{1}{4} (b-a) (f(x_1)+f(x_2)+f(x_3)+f(x_4))\\& = \frac{1}{4}(b-a)\sum_{i=1}^{4}f(x_i) \end {align*} $$进一步推广，即$$ S=\frac{1}{n}(b-a)\sum_{1}^{n}f(x_i) $$评价蒙特卡洛方法就是这样的思想，如果样本数量足够大，就能足够逼近真实值，但是这种方法通常也有些缺点。比如计算量较大，可能需要大量样本以确保结果准确性；结果受随机数质量影响。
创建时间: 2024年10月29日 18:21作者: 蜡笔大新笔记类别: 强化学习标签: Markov Chain, Markov Property状态: 完成马尔可夫性质（Markov Property）马尔可夫性质是这些概念的核心，也是最基础的概念。它表明：一个过程的未来状态只依赖于当前状态，而与之前的状态无关。数学上，如果一个过程满足马尔可夫性质，那么对于任何时刻 $t$ ，有：$$ P(s_{t+1} | s_t, s_{t-1}, \dots, s_0) = P(s_{t+1} | s_t) $$这就是说，给定当前状态 $s_t$ ，下一个状态 $s_{t+1}$ 的概率分布不依赖于更早的状态 $s_{t-1}, \dots, s_0$ 。马尔可夫性质是一种“无记忆性”，强调当前状态已包含了做决策所需的所有信息。马尔可夫链（Markov Chain）马尔可夫链是一种描述随机过程的数学模型，满足马尔可夫性质的一种特殊的离散时间、离散状态的随机过程。基本概念• 状态空间：马尔可夫链中的所有可能状态的集合。• 转移概率：从一个状态转移到另一个状态的概率。• 转移矩阵：一个矩阵，描述了所有状态之间的转移概率。转移矩阵假设状态空间为 $S = \{s_1, s_2, \dots, s_n\}$ ，则转移矩阵 P 为：$$ P = \begin{bmatrix}P(s_1 \to s_1) & P(s_1 \to s_2) & \cdots & P(s_1 \to s_n) \\P(s_2 \to s_1) & P(s_2 \to s_2) & \cdots & P(s_2 \to s_n) \\\vdots & \vdots & \ddots & \vdots \\P(s_n \to s_1) & P(s_n \to s_2) & \cdots & P(s_n \to s_n)\end{bmatrix} $$其中 $P(s_i \to s_j)$ 表示从状态 $s_i$ 转移到状态 $s_j$ 的概率。从某个状态出发，到达其他状态的概率和必须为 1，即状态转移矩阵$P$的每一行的和为 1。举例：$P$ 矩阵为：$$ \mathcal{P} = \begin{bmatrix}0.9 & 0.1 & 0 & 0 & 0 & 0 \\0.5 & 0 & 0.5 & 0 & 0 & 0 \\0 & 0 & 0.6 & 0 & 0.4 & 0 \\0 & 0 & 0 & 0.3 & 0.7 & 0 \\0 & 0.2 & 0.3 & 0.5 & 0 & 0 \\0 & 0 & 0 & 0 & 0 & 1 \\\end{bmatrix} $$马尔可夫过程（Markov Process）马尔可夫过程有时与马尔可夫链互换使用，但更广义地指满足马尔可夫性质的过程。它们可以是离散或连续状态、离散或连续时间的随机过程。从定义上看：马尔可夫链是马尔可夫过程的一种特例，即离散状态和离散时间的马尔可夫过程。广义的马尔可夫过程可以是连续状态和时间的。[scode type="blue" size=""]离散例子：在一个天气模型中，如果今天的天气状态是晴天，那么明天的天气状态转为雨天或继续晴天的概率取决于今天的状态，而与之前的天气无关。连续例子：在一个股票模型中，股票价格的变化可以被视为一个连续的马尔可夫过程。在任何给定时刻，股票的未来价格只依赖于其当前价格，而不直接依赖于过去的价格变动。这种模型假设市场是有效的，所有相关的历史信息都已反映在当前价格中。[/scode]当前状态是未来的充分统计量，即下一个状态只取决于当前状态，而不会受到过去状态的影响。需要明确的是，具有马尔可夫性并不代表这个随机过程就和历史完全没有关系。因为虽然时刻 $t+1$的状态只与时刻 $t$ 的状态有关，但是时刻 $t$ 的状态其实包含了时刻 $t-1$ 的状态的信息，通过这种链式的关系，历史的信息被传递到了现在。马尔可夫性可以大大简化运算，因为只要当前状态可知，所有的历史信息都不再需要了，利用当前状态信息就可以决定未来。例子2假设这个 $P$ 矩阵是时齐的，也就是说随着时间推移，概率是不变的。那么第二天计算第三天的概率矩阵仍然是这个矩阵 $P$ 。那么第二天的概率也就是 第一天的概率矩阵 $ \times P$ ；第三天的概率即 第二天的概率矩阵 $ \times P$，以此类推。我们就可以计算出第N天的概率矩阵。再以这个例子理解一下马尔可夫过程。第N天吃饭睡觉玩耍的概率，其实是由第N-1天的概率矩阵计算得来的，因此可以理解只与上一个状态有关，但是实际上包含了过去的每一个状态的信息。[scode type="blue" size=""]时齐：时间序列数据在不同时间段之间的统计性质保持不变。即时齐数据的均值、方差等统计特性在时间上是稳定的，不随时间变化。[/scode]马尔可夫决策过程(Markov Decision Process, MDP)定义马尔可夫决策过程是在马尔可夫过程的基础上加入了决策和奖励的概念。在强化学习中，MDP 是描述决策问题的数学模型，它由以下四个部分构成：状态集合 $S$ ：代表环境的所有可能状态。动作集合 $A$ ：代表智能体可以选择的所有可能动作。转移概率 $P(s{\prime} | s, a)$ ：智能体在状态 $s$ 下执行动作 $a$ 后转移到状态 $s{\prime}$ 的概率。奖励函数 $R(s, a)$ ：智能体在状态 $s$ 下执行动作 $a$ 后获得的即时奖励。[scode type="blue" size=""]转移概率 $P$ 举例：用来描述一些具有不确定性问题的概率，例如打羽毛球可能会由于刮风而发生角度偏移，例如30%向左偏，40%向前，30%向右偏。这个概率和之前的状态没有关系，只与当前状态有关，因此符合马尔可夫决策过程。转移概率在理论上用于评估策略，而实际编码中，经常不用考虑具体的转移概率，而是考虑拥有的可能性并构建相应奖励函数。[/scode]累积奖励（折扣回报）在强化学习中，智能体不仅关注即时奖励 $R(s, a)$ ，更注重长期累积奖励 $G_t$ ，其定义为从当前时刻 $t$ 开始累积的未来奖励之和。引入一个折扣因子 $\gamma \in [0, 1]$ 控制未来奖励的影响，累积奖励公式为：$$ G_t = R_{t+1} + \gamma R_{t+2} + \gamma^2 R_{t+3} + \dots = \sum_{k=0}^{\infty} \gamma^k R_{t+k+1} $$累积奖励关注的是从当前状态通过指定路径到达目标状态的奖励之和。与Q值的区别在于：Q值关注当前状态执行某个动作获得的奖励期望累积奖励关注按照这个策略（路径）一共能获得多少奖励[scode type="blue" size=""]公式中，越靠近当前状态，权重越大；越远离当前状态，权重越小（通过 $\gamma$ 的次方控制）。当 $\gamma$ 接近 1 时，智能体会更加重视远期奖励，愿意为长期的回报做出权衡。当 $\gamma$ 较小时，智能体会更加关注近期的回报，对较远的奖励不太重视。[/scode]总结概念解释是否有决策、奖励马尔可夫性质当前状态决定未来状态，与历史无关否马尔可夫链离散时间、离散状态的马尔可夫过程否马尔可夫过程广义的满足马尔可夫性质的过程否马尔可夫决策过程 (MDP)带有决策和奖励的马尔可夫过程是