创建时间: 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月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算法，震荡幅度较小，比较稳定。
介绍根据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年10月24日 13:53作者: 蜡笔大新笔记类别: 强化学习标签: Deep Reinforcement Learning, Experience Replay, Q-Learning状态: 完成传统Q-Learning的一些问题在Q-Learning中，需要反复更新Q-table，最终通过贪心算法根据Q表选取相应的动作。而这通常会带来一些问题，比如动作和状态维度过高之后，Q-table的维度也会非常高，带来计算和存储的巨大压力。DQN介绍DQN（Deep Q-Network）是一种结合了 深度学习 和 Q-Learning 的强化学习算法，它通过深度神经网络来近似 Q 值函数，从而能够处理高维状态空间问题。优点DQN 使用一个 深度神经网络（DNN） 来代替传统 Q-Learning 中的 Q 表格，将状态作为输入，通过神经网络输出对应每个动作的 Q 值。这使得 DQN 能够处理复杂和高维的状态空间，比如在图像、视频帧等环境下的强化学习问题。[scode type="blue" size=""]简而言之，模型不再需要去额外存储表格，而是通过训练得到一个函数，输入状态，就能输出每个动作的Q值。[/scode]传统的Q-Learning很适合格子游戏。因为格子游戏中的每一个格子就是一个状态，但在现实生活中，很多状态并不是离散而是连续的。通过函数的形式就可以很好的解决连续的状态问题，比如驾驶的方向等。训练经验回放DQN 引入了一个非常重要的技术：经验回放，它的作用是打破样本之间的时间相关性，并使得神经网络的训练更加稳定。[scode type="blue" size=""]经验，即四元组：状态、动作、奖励、下一个状态。[/scode]在训练过程中，DQN 会从这个 经验池 中随机抽取样本来训练神经网络，从而减少数据之间的相关性，提升训练稳定性。模型DQN引入了目标网络（Target网络）与主体网络，它们是互相隔离的。目标网络的参数每隔一段时间才会从当前网络复制，而不是每步都更新。这样设计的目标主要是为了网络稳定，避免发散。或者以操作系统举例，主网络就是Beta测试版，Target网络就是Latest稳定版。稳定版会每隔一段时间或条件向测试版偏移。损失函数DQN 的训练目标是通过最小化当前 Q 值和目标 Q 值之间的误差来优化模型。常用的误差函数是 均方误差（MSE, Mean Squared Error）：$$ L(\theta)= [y-Q(s,a)]^2 $$其中：$y=Q(s_t, a_t) + \alpha \left[ r_t + \gamma \max_{a'} Q(s_{t+1}, a') - Q(s_t, a_t) \right]$，即贝尔曼公式，由目标DQN计算。$Q(s,a)$：当前 Q 值，由 当前DQN 网络预测。仍然分两个网络来讲：当前DQN：输出 $Q(s,a)$目标DQN：选取目标网络下下一个状态Q最大的动作，将其收益、Q值等代入贝尔曼公式输出y值。[scode type="blue" size=""]目标DQN的 $y$ 值可以理解为理论上的“正确值”，类似于有监督学习下打的标签。它与当前DQN输出的值做比较，通过MSE计算误差，如果误差越小，那说明当前网络距离真实模型的误差越小，而整个神经网络的最终目标就是让这个误差尽可能的小。[/scode]输入输出DQN接受当前状态，并且能输出一个包含所有动作Q值的向量，每个动作的顺序在定义网络结构的时候就已经确定了，并在整个训练和使用过程中保持不变。神经网络的输出层有固定数量的节点，每个节点对应一个特定的动作⁠。例如，在CartPole-v0环境中，动作空间只有两个动作（向左推和向右推）⁠。DQN的输出层就会有两个节点，可能表示为：第一个节点（索引0）：向左推的Q值第二个节点（索引1）：向右推的Q值动作选择当输入状态后，网络会输出包含所有动作Q值的向量。选择动作主要有两种策略：贪婪策略（Greedy Policy）：在大多数情况下，DQN会选择Q值最高的动作。这就是所谓的利用（exploitation）。ε-贪婪策略（ε-greedy Policy）：为了平衡探索和利用，DQN通常会采用ε-贪婪策略。具体来说：以ε的概率随机选择一个动作（探索）以1-ε的概率选择Q值最高的动作（利用）代码CartPole-v0CartPole-v0 是OpenAI Gym中的一个经典控制问题环境。在这个环境中：一个杆子通过一个非驱动关节连接到一个可以左右移动的小车上。系统开始时杆子是垂直的，目标是通过左右移动小车来保持杆子直立。每个时间步，agent可以选择向左或向右推车。每次成功保持杆子直立的时间步骤都会获得+1的奖励。当杆子倾斜超过15度，或者小车移动超出中心2.4个单位，或者回合持续超过200个时间步时，回合结束。在游戏中每坚持一帧，智能体能获得分数为 1 的奖励，坚持时间越长，则最后的分数越高，坚持 200 帧即可获得最高的分数。（也就是说，奖励和到达200即可结束游戏）import random import gym import numpy as np import collections from tqdm import tqdm import torch import torch.nn.functional as F import matplotlib.pyplot as plt import rl_utils class ReplayBuffer: ''' 经验回放池 ''' def __init__(self, capacity): self.buffer = collections.deque(maxlen=capacity) # 队列,先进先出 def add(self, state, action, reward, next_state, done): # 将数据加入buffer self.buffer.append((state, action, reward, next_state, done)) def sample(self, batch_size): # 从buffer中采样数据,数量为batch_size transitions = random.sample(self.buffer, batch_size) state, action, reward, next_state, done = zip(*transitions) return np.array(state), action, reward, np.array(next_state), done def size(self): # 目前buffer中数据的数量 return len(self.buffer) 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)) # 隐藏层使用ReLU激活函数 return self.fc2(x) class DQN: ''' DQN算法 ''' def __init__(self, state_dim, hidden_dim, action_dim, learning_rate, gamma, epsilon, target_update, device): self.action_dim = action_dim self.q_net = Qnet(state_dim, hidden_dim, self.action_dim).to(device) # Q网络 # 目标网络 self.target_q_net = Qnet(state_dim, hidden_dim, self.action_dim).to(device) # 使用Adam优化器 self.optimizer = torch.optim.Adam(self.q_net.parameters(), lr=learning_rate) self.gamma = gamma # 折扣因子 self.epsilon = epsilon # epsilon-贪婪策略 self.target_update = target_update # 目标网络更新频率 self.count = 0 # 计数器,记录更新次数 self.device = device def take_action(self, state): # epsilon-贪婪策略采取动作 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 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值 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 = 2e-3 num_episodes = 500 hidden_dim = 128 gamma = 0.98 epsilon = 0.01 target_update = 10 buffer_size = 10000 minimal_size = 500 batch_size = 64 device = torch.device("cuda") if torch.cuda.is_available() else torch.device( "cpu") env_name = 'CartPole-v0' env = gym.make(env_name) random.seed(0) np.random.seed(0) env.seed(0) torch.manual_seed(0) replay_buffer = ReplayBuffer(buffer_size) state_dim = env.observation_space.shape[0] action_dim = env.action_space.n agent = DQN(state_dim, hidden_dim, action_dim, lr, gamma, epsilon, target_update, 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 state = env.reset() done = False while not done: action = agent.take_action(state) next_state, reward, done, _ = env.step(action) replay_buffer.add(state, action, reward, next_state, done) state = next_state episode_return += reward # 当buffer数据的数量超过一定值后,才进行Q网络训练 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) episodes_list = list(range(len(return_list))) plt.plot(episodes_list, return_list) plt.xlabel('Episodes') plt.ylabel('Returns') plt.title('DQN 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('DQN on {}'.format(env_name)) plt.show()结果总体而言，奖励能够到达200，说明网络拥有不错的性能，但是还是有些不太稳定，产生了震荡。
创建时间: 2024年10月23日 19:04作者: 蜡笔大新笔记类别: 强化学习标签: Q-Learning, Reinforcement Learning, Temporal Difference Learning状态: 完成简介TD算法，全称为"时间差分算法"（Temporal Difference Algorithm），是强化学习中的一种重要算法。它结合了动态规划和蒙特卡洛方法的优点，广泛应用于各种学习和决策问题。核心思想TD算法的核心思想是通过比较连续时间步骤的预测来学习。它使用当前估计和下一时间步的奖励来更新价值函数，而不需要等待整个序列完成。与其相反的蒙特卡洛方法，则是需要一直往前探索到结束，再往回走反馈奖励总和。[scode type="blue" size=""]TD有点类似动态规划算法（DP），都利用状态转移方程（贝尔曼公式）。而区别主要在于，TD是不需要环境模型的，可以自由探索环境，不断用前方的奖励训练自身模型；而DP需要知道完整的模型，需要对整个状态空间进行迭代，DP方法在状态空间较大时计算代价高昂。[/scode]主要特点在线学习：TD算法可以在每个时间步更新估计，无需等待整个序列结束。自举：TD算法使用估计来更新估计，这种自举方法使其能够快速学习。模型无关：TD算法不需要环境模型，可以直接从经验中学习。变体TD算法有多种变体，包括：Q-learning：一种off-policy TD控制算法SARSA：一种on-policy TD控制算法TD(λ)：结合了多步预测的TD算法Q- Learning在强化学习中，有两个核心概念：Q值和V值。Q值表示在某个状态下执行这个动作后，直到最终状态所获得的期望累积奖励；V值代表从这个状态开始，经过一系列决策到达最终状态的期望累积奖励。Q值和V值可以互相转化：一个状态的V值是其所有动作Q值的平均值；一个动作的Q值是下一个状态的V值加上当前动作的即时奖励。Q-learning是一种基于Q值的强化学习算法，主要思想是通过不断更新Q表来学习最佳策略。Q值状态 (State, s): 系统或环境的当前状况。动作 (Action, a): 在某个状态下，代理可以执行的行为选择。奖励 (Reward, r): 执行某个动作后，代理从环境中获得的反馈信号，通常用于衡量该动作的好坏。Q-Tablea1a2s1q(s1,a1)q(s1,a2)s2q(s2,a1)q(s2,a2)s3q(s3,a1)q(s3,a2)具体算法每个“状态-动作对”都拥有自己的Q值，而Q-Learning的核心就是在于不断维护、更新这个Q-Table。具体的算法步骤可以参考下图，主要步骤即为通过贪心算法更新 $[S_t, A_t]$状态-动作对下的q值，贪心算法是选取 $S_{t+1}$的状态下Q值最大的动作，利用奖励R反推。递推公式为贝尔曼公式：$$ Q(s_t, a_t) \leftarrow Q(s_t, a_t) + \alpha \left[ r_t + \gamma \max_{a'} Q(s_{t+1}, a') - Q(s_t, a_t) \right] $$其中：$s_t$ 是当前状态$a_t$ 是当前动作$r_t$ 是在执行动作后获得的奖励$s_{t+1}$ 是下一个状态$a'$ 是未来可能的动作$\alpha$ 是学习率（Learning Rate），控制每次更新Q值的步长。$\gamma$ 是折扣因子（Discount Factor），表示未来奖励的权重（0到1之间的数）。高 $\gamma$ 值意味着未来奖励更重要。[scode type="blue" size=""]当初始化时，Q-Table所有Q值都为0，更新的时候会直接用奖励值。[/scode]SarsaSarsa与Q-Learning很相似，区别是：Q-Learning选取的是下一个状态下最大Q值的动作；而Sarsa选取的仍是前一个动作，即 $A_t$ 。Q-learning是一种离线策略（off-policy）算法，它学习的是最优策略，不管当前正在执行的是什么策略，因此更倾向于探索；Sarsa是一种在线策略（on-policy）算法，它学习的是当前正在执行的策略，因此更保守。举例状态 (State) \动作（Action）动作A1动作A2S15[font color="Green"]9[/font]S2[font color="Green"]3[/font]2S3[font color="Green"]7[/font]1通过上述Q表，我们可以清晰的看出在每个状态下采取不同动作的奖励期望值，我们只需要在每个状态下选取收益最大的动作即可。代码环境悬崖漫步环境要求一个智能体从起点出发，避开悬崖行走，最终到达目标位置。智能体在每一个状态都可以采取 4 种动作：上、下、左、右。如果智能体采取动作后触碰到边界墙壁则状态不发生改变，否则就会相应到达下一个状态。环境中有一段悬崖，智能体掉入悬崖或到达目标状态都会结束动作并回到起点，也就是说掉入悬崖或者达到目标状态是终止状态。智能体每走一步的奖励是 −1，掉入悬崖的奖励是 −100。import matplotlib.pyplot as plt import numpy as np from tqdm import tqdm # tqdm是显示循环进度条的库 class CliffWalkingEnv: def __init__(self, ncol, nrow): self.nrow = nrow self.ncol = ncol self.x = 0 # 记录当前智能体位置的横坐标 self.y = self.nrow - 1 # 记录当前智能体位置的纵坐标 def step(self, action): # 外部调用这个函数来改变当前位置 # 4种动作, change[0]:上, change[1]:下, change[2]:左, change[3]:右。坐标系原点(0,0) # 定义在左上角 change = [[0, -1], [0, 1], [-1, 0], [1, 0]] self.x = min(self.ncol - 1, max(0, self.x + change[action][0])) self.y = min(self.nrow - 1, max(0, self.y + change[action][1])) next_state = self.y * self.ncol + self.x reward = -1 done = False if self.y == self.nrow - 1 and self.x > 0: # 下一个位置在悬崖或者目标 done = True if self.x != self.ncol - 1: reward = -100 return next_state, reward, done def reset(self): # 回归初始状态,坐标轴原点在左上角 self.x = 0 self.y = self.nrow - 1 return self.y * self.ncol + self.x def print_agent(agent, env, action_meaning, disaster=[], end=[]): for i in range(env.nrow): for j in range(env.ncol): if (i * env.ncol + j) in disaster: print('****', end=' ') elif (i * env.ncol + j) in end: print('EEEE', end=' ') else: a = agent.best_action(i * env.ncol + j) pi_str = '' for k in range(len(action_meaning)): pi_str += action_meaning[k] if a[k] > 0 else 'o' print(pi_str, end=' ') print() class QLearning: """ Q-learning算法 """ def __init__(self, ncol, nrow, epsilon, alpha, gamma, n_action=4): self.Q_table = np.zeros([nrow * ncol, n_action]) # 初始化Q(s,a)表格 self.n_action = n_action # 动作个数 self.alpha = alpha # 学习率 self.gamma = gamma # 折扣因子 self.epsilon = epsilon # epsilon-贪婪策略中的参数 def take_action(self, state): #选取下一步的操作 if np.random.random() < self.epsilon: action = np.random.randint(self.n_action) else: action = np.argmax(self.Q_table[state]) return action def best_action(self, state): # 用于打印策略 Q_max = np.max(self.Q_table[state]) a = [0 for _ in range(self.n_action)] for i in range(self.n_action): if self.Q_table[state, i] == Q_max: a[i] = 1 return a def update(self, s0, a0, r, s1): td_error = r + self.gamma * self.Q_table[s1].max() - self.Q_table[s0, a0] self.Q_table[s0, a0] += self.alpha * td_error class Sarsa: """ Sarsa算法 """ def __init__(self, ncol, nrow, epsilon, alpha, gamma, n_action=4): self.Q_table = np.zeros([nrow * ncol, n_action]) # 初始化Q(s,a)表格 self.n_action = n_action # 动作个数 self.alpha = alpha # 学习率 self.gamma = gamma # 折扣因子 self.epsilon = epsilon # epsilon-贪婪策略中的参数 def take_action(self, state): # 选取下一步的操作,具体实现为epsilon-贪婪 if np.random.random() < self.epsilon: action = np.random.randint(self.n_action) else: action = np.argmax(self.Q_table[state]) return action def best_action(self, state): # 用于打印策略 Q_max = np.max(self.Q_table[state]) a = [0 for _ in range(self.n_action)] for i in range(self.n_action): # 若两个动作的价值一样,都会记录下来 if self.Q_table[state, i] == Q_max: a[i] = 1 return a def update(self, s0, a0, r, s1, a1): td_error = r + self.gamma * self.Q_table[s1, a1] - self.Q_table[s0, a0] self.Q_table[s0, a0] += self.alpha * td_error ncol = 12 nrow = 4 env = CliffWalkingEnv(ncol, nrow) np.random.seed(0) epsilon = 0.1 alpha = 0.1 gamma = 0.9 agent = Sarsa(ncol, nrow, epsilon, alpha, gamma) num_episodes = 500 # 智能体在环境中运行的序列的数量 return_list = [] # 记录每一条序列的回报 for i in range(10): # 显示10个进度条 # tqdm的进度条功能 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() action = agent.take_action(state) done = False while not done: next_state, reward, done = env.step(action) next_action = agent.take_action(next_state) episode_return += reward # 这里回报的计算不进行折扣因子衰减 agent.update(state, action, reward, next_state, next_action) state = next_state action = next_action return_list.append(episode_return) if (i_episode + 1) % 10 == 0: # 每10条序列打印一下这10条序列的平均回报 pbar.set_postfix({ 'episode': '%d' % (num_episodes / 10 * i + i_episode + 1), 'return': '%.3f' % np.mean(return_list[-10:]) }) pbar.update(1) action_meaning = ['^', 'v', '<', '>'] print('Sarsa算法最终收敛得到的策略为：') print_agent(agent, env, action_meaning, list(range(37, 47)), [47]) episodes_list = list(range(len(return_list))) plt.plot(episodes_list, return_list) plt.xlabel('Episodes') plt.ylabel('Returns') plt.title('Sarsa on {}'.format('Cliff Walking')) plt.show() np.random.seed(0) epsilon = 0.1 alpha = 0.1 gamma = 0.9 agent = QLearning(ncol, nrow, epsilon, alpha, gamma) num_episodes = 500 # 智能体在环境中运行的序列的数量 return_list = [] # 记录每一条序列的回报 for i in range(10): # 显示10个进度条 # tqdm的进度条功能 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) next_state, reward, done = env.step(action) episode_return += reward # 这里回报的计算不进行折扣因子衰减 agent.update(state, action, reward, next_state) state = next_state return_list.append(episode_return) if (i_episode + 1) % 10 == 0: # 每10条序列打印一下这10条序列的平均回报 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('Q-learning on {}'.format('Cliff Walking')) plt.show() action_meaning = ['^', 'v', '<', '>'] print('Q-learning算法最终收敛得到的策略为：') print_agent(agent, env, action_meaning, list(range(37, 47)), [47])结果Sarsa：Q-Learning打印出来的动作，我们用^o<o表示等概率采取向左和向上两种动作，ooo>表示在当前状态只采取向右动作。通过Sarsa和Q-Learning结果的对比，可以比较明显的看出Sarsa和Q-Learning都能比较好的学习到策略，奖励逐步上升，最后到达终点。Sarsa比Q-Learning更稳定一些，没有非常大的起伏。通过路线图可以看出，Sarsa更加保守，偏向于在更安全的地方行走，Q-Learning更加偏向于探索，能够在悬崖边上行走。正因如此，Q-Learning更加容易掉入悬崖，这在奖励图上也有所体现。