黄岛建网站,网页设计与网站建设作品,小企业网站建设方案,dede本地搭建网站本篇博客是本人参加Datawhale组队学习第五次任务的笔记 【教程地址】https://github.com/datawhalechina/joyrl-book 【强化学习库JoyRL】https://github.com/datawhalechina/joyrl/tree/main 【JoyRL开发周报】 https://datawhale.feishu.cn/docx/OM8fdsNl0o5omoxB5nXcyzsInGe…本篇博客是本人参加Datawhale组队学习第五次任务的笔记 【教程地址】https://github.com/datawhalechina/joyrl-book 【强化学习库JoyRL】https://github.com/datawhalechina/joyrl/tree/main 【JoyRL开发周报】 https://datawhale.feishu.cn/docx/OM8fdsNl0o5omoxB5nXcyzsInGe?fromfrom_copylink 【教程参考地址】https://github.com/datawhalechina/easy-rl 文章目录 介绍重要性采样PPO算法核心思想与TRPO方法的联系公式推导一个常见的误区 代码实战1. 定义算法2. 定义训练3. 定义环境训练可视化 练习总结 介绍
算法适用范围提出时间主要思想DQN只适用于离散动作2013/2015在Q-learning的基础上引入了深度神经网络A3C既适用于连续动作空间也适用于离散动作空间2016引入了优势函数引入多进程DDPG只适用于连续动作空间2015将选择动作的过程变成一个直接从状态映射到具体动作的函数TD3只适用于连续动作空间2018双 Q 网络延迟更新和躁声正则PPO既适用于连续动作空间也适用于离散动作空间2017通过在策略梯度的优化过程中引入一个重要性权重来限制策略更新的幅度从而提高算法的稳定性和收敛性
重要性采样
概念一种估计随机变量的期望或者概率分布的统计方法算法直觉在复杂问题中利用已知的简单分布进行采样从而避免了直接采样困难分布的问题公式理解
PPO算法
核心思想
通过重要性采样来优化原来的策略梯度估计
与TRPO方法的联系
TRPO方法中通过使用约束而非惩罚项来保证策略更新的稳定性主要原因是作为惩罚项的话会引入权重因子而这个参数难以调节。TRPO中为了解优化问题先线性近似目标函数二阶近似约束最后通过conjugate gradient算法和line search求解。PPO算法尝试通过一阶优化的方法来解。与TRPO中用约束来限制策略更新幅度不同PPO中采用了惩罚项
公式推导
文中提出了基于clipped probability ratio的目标函数 重要性权重最好尽可能地等于1而在训练过程中这个权重它是不会自动地约束到1 附近的因此我们需要在损失函数中加入一个约束项或者说正则项保证重要性权重不会偏离 1 太远。具体的约束方法有很多种比如 KL 散度、JS 散度等等但通常我们会使用两种约束方法一种是 clip 约束另一种是KL 散度
直观上当策略更新的偏移超出预定区间而获得更大的目标函数值时这个clip项就会产生影响
当策略更新前后KL散度小于预定值时惩罚项系数β减小到原来的一半当大于预定值时系数增加一倍。如果使用神经网络和共享策略和值函数参数的话就需要在损失函数中结合策略优化目标和值函数误差项。
一个常见的误区
on-policy 使用当前策略生成样本并基于这些样本来更新该策略 on-policy 算法的数据利用效率较低因为每次策略更新后旧的样本或经验可能就不再适用通常需要重新采样 off-policy 算法可以使用过去的策略采集样本来更新当前的策略 off-policy 算法由于可以利用历史经验一般使用经验回放来存储和重复利用之前的经验数据利用效率则较高因为同一批数据可以被用于多次更新。但由于经验的再利用可能会引入一定的偏见但这也有助于稳定学习
在需要即时学习和适应的环境中on-policy 算法可能更为适合因为它们直接在当前策略下操作。
那么 PPO 算法究竟是 on-policy 还是 off-policy 的呢? 有读者可能会因为 PPO 算法在更新时重要性采样的部分中利用了旧的 Actor 采样的样本就觉得PPO 算法会是 off-policy 的。实际上虽然这批样本是从旧的策略中采样得到的但我们并没有直接使用这些样本去更新我们的策略而是使用重要性采样先将数据分布不同导致的误差进行了修正即是两者样本分布之间的差异尽可能地缩小。换句话说就可以理解为重要性采样之后的样本虽然是由旧策略采样得到的但可以近似为从更新后的策略中得到的即我们要优化的 Actor 和采样的Actor 是同一个因此 PPO 算法是 on-policy 的。 代码实战
PPO实现CarPole-v1离散动作空间
1. 定义算法
定义模型
import torch.nn as nn
import torch.nn.functional as F
class ActorSoftmax(nn.Module):def __init__(self, input_dim, output_dim, hidden_dim256):super(ActorSoftmax, self).__init__()self.fc1 nn.Linear(input_dim, hidden_dim)self.fc2 nn.Linear(hidden_dim, hidden_dim)self.fc3 nn.Linear(hidden_dim, output_dim)def forward(self,x):x F.relu(self.fc1(x))x F.relu(self.fc2(x))probs F.softmax(self.fc3(x),dim1)return probs
class Critic(nn.Module):def __init__(self,input_dim,output_dim,hidden_dim256):super(Critic,self).__init__()assert output_dim 1 # critic must output a single valueself.fc1 nn.Linear(input_dim, hidden_dim)self.fc2 nn.Linear(hidden_dim, hidden_dim)self.fc3 nn.Linear(hidden_dim, output_dim)def forward(self,x):x F.relu(self.fc1(x))x F.relu(self.fc2(x))value self.fc3(x)return value定义经验回放
import random
from collections import deque
class ReplayBufferQue:DQN的经验回放池每次采样batch_size个样本def __init__(self, capacity: int) - None:self.capacity capacityself.buffer deque(maxlenself.capacity)def push(self,transitions):_summary_Args:trainsitions (tuple): _description_self.buffer.append(transitions)def sample(self, batch_size: int, sequential: bool False):if batch_size len(self.buffer):batch_size len(self.buffer)if sequential: # sequential samplingrand random.randint(0, len(self.buffer) - batch_size)batch [self.buffer[i] for i in range(rand, rand batch_size)]return zip(*batch)else:batch random.sample(self.buffer, batch_size)return zip(*batch)def clear(self):self.buffer.clear()def __len__(self):return len(self.buffer)class PGReplay(ReplayBufferQue):PG的经验回放池每次采样所有样本因此只需要继承ReplayBufferQue重写sample方法即可def __init__(self):self.buffer deque()def sample(self): sample all the transitionsbatch list(self.buffer)return zip(*batch)定义智能体
import torch
from torch.distributions import Categorical
class Agent:def __init__(self,cfg) - None:self.gamma cfg.gammaself.device torch.device(cfg.device) self.actor ActorSoftmax(cfg.n_states,cfg.n_actions, hidden_dim cfg.actor_hidden_dim).to(self.device)self.critic Critic(cfg.n_states,1,hidden_dimcfg.critic_hidden_dim).to(self.device)self.actor_optimizer torch.optim.Adam(self.actor.parameters(), lrcfg.actor_lr)self.critic_optimizer torch.optim.Adam(self.critic.parameters(), lrcfg.critic_lr)self.memory PGReplay()self.k_epochs cfg.k_epochs # update policy for K epochsself.eps_clip cfg.eps_clip # clip parameter for PPOself.entropy_coef cfg.entropy_coef # entropy coefficientself.sample_count 0self.update_freq cfg.update_freqdef sample_action(self,state):self.sample_count 1state torch.tensor(state, deviceself.device, dtypetorch.float32).unsqueeze(dim0)probs self.actor(state)dist Categorical(probs)action dist.sample()self.log_probs dist.log_prob(action).detach()return action.detach().cpu().numpy().item()torch.no_grad()def predict_action(self,state):state torch.tensor(state, deviceself.device, dtypetorch.float32).unsqueeze(dim0)probs self.actor(state)dist Categorical(probs)action dist.sample()return action.detach().cpu().numpy().item()def update(self):# update policy every n stepsif self.sample_count % self.update_freq ! 0:return# print(update policy)old_states, old_actions, old_log_probs, old_rewards, old_dones self.memory.sample()# convert to tensorold_states torch.tensor(np.array(old_states), deviceself.device, dtypetorch.float32)old_actions torch.tensor(np.array(old_actions), deviceself.device, dtypetorch.float32)old_log_probs torch.tensor(old_log_probs, deviceself.device, dtypetorch.float32)# monte carlo estimate of state rewardsreturns []discounted_sum 0for reward, done in zip(reversed(old_rewards), reversed(old_dones)):if done:discounted_sum 0discounted_sum reward (self.gamma * discounted_sum)returns.insert(0, discounted_sum)# Normalizing the rewards:returns torch.tensor(returns, deviceself.device, dtypetorch.float32)returns (returns - returns.mean()) / (returns.std() 1e-5) # 1e-5 to avoid division by zerofor _ in range(self.k_epochs):# compute advantagevalues self.critic(old_states) # detach to avoid backprop through the criticadvantage returns - values.detach()# get action probabilitiesprobs self.actor(old_states)dist Categorical(probs)# get new action probabilitiesnew_probs dist.log_prob(old_actions)# compute ratio (pi_theta / pi_theta__old):ratio torch.exp(new_probs - old_log_probs) # old_log_probs must be detached# compute surrogate losssurr1 ratio * advantagesurr2 torch.clamp(ratio, 1 - self.eps_clip, 1 self.eps_clip) * advantage# compute actor lossactor_loss -torch.min(surr1, surr2).mean() self.entropy_coef * dist.entropy().mean()# compute critic losscritic_loss (returns - values).pow(2).mean()# take gradient stepself.actor_optimizer.zero_grad()self.critic_optimizer.zero_grad()actor_loss.backward()critic_loss.backward()self.actor_optimizer.step()self.critic_optimizer.step()self.memory.clear()2. 定义训练
import copy
def train(cfg, env, agent): 训练print(开始训练)rewards [] # 记录所有回合的奖励steps []best_ep_reward 0 # 记录最大回合奖励output_agent Nonefor i_ep in range(cfg.train_eps):ep_reward 0 # 记录一回合内的奖励ep_step 0state env.reset() # 重置环境返回初始状态for _ in range(cfg.max_steps):ep_step 1action agent.sample_action(state) # 选择动作next_state, reward, done, _ env.step(action) # 更新环境返回transitionagent.memory.push((state, action,agent.log_probs,reward,done)) # 保存transitionstate next_state # 更新下一个状态agent.update() # 更新智能体ep_reward reward # 累加奖励if done:breakif (i_ep1)%cfg.eval_per_episode 0:sum_eval_reward 0for _ in range(cfg.eval_eps):eval_ep_reward 0state env.reset()for _ in range(cfg.max_steps):action agent.predict_action(state) # 选择动作next_state, reward, done, _ env.step(action) # 更新环境返回transitionstate next_state # 更新下一个状态eval_ep_reward reward # 累加奖励if done:breaksum_eval_reward eval_ep_rewardmean_eval_reward sum_eval_reward/cfg.eval_epsif mean_eval_reward best_ep_reward:best_ep_reward mean_eval_rewardoutput_agent copy.deepcopy(agent)print(f回合{i_ep1}/{cfg.train_eps}奖励{ep_reward:.2f}评估奖励{mean_eval_reward:.2f}最佳评估奖励{best_ep_reward:.2f}更新模型)else:print(f回合{i_ep1}/{cfg.train_eps}奖励{ep_reward:.2f}评估奖励{mean_eval_reward:.2f}最佳评估奖励{best_ep_reward:.2f})steps.append(ep_step)rewards.append(ep_reward)print(完成训练)env.close()return output_agent,{rewards:rewards}def test(cfg, env, agent):print(开始测试)rewards [] # 记录所有回合的奖励steps []for i_ep in range(cfg.test_eps):ep_reward 0 # 记录一回合内的奖励ep_step 0state env.reset() # 重置环境返回初始状态for _ in range(cfg.max_steps):ep_step1action agent.predict_action(state) # 选择动作next_state, reward, done, _ env.step(action) # 更新环境返回transitionstate next_state # 更新下一个状态ep_reward reward # 累加奖励if done:breaksteps.append(ep_step)rewards.append(ep_reward)print(f回合{i_ep1}/{cfg.test_eps}奖励{ep_reward:.2f})print(完成测试)env.close()return {rewards:rewards}3. 定义环境
import gym
import os
import numpy as np
def all_seed(env,seed 1): 万能的seed函数if seed 0:returnenv.seed(seed) # env confignp.random.seed(seed)random.seed(seed)torch.manual_seed(seed) # config for CPUtorch.cuda.manual_seed(seed) # config for GPUos.environ[PYTHONHASHSEED] str(seed) # config for python scripts# config for cudnntorch.backends.cudnn.deterministic Truetorch.backends.cudnn.benchmark Falsetorch.backends.cudnn.enabled False
def env_agent_config(cfg):env gym.make(cfg.env_name) # 创建环境all_seed(env,seedcfg.seed)n_states env.observation_space.shape[0]n_actions env.action_space.nprint(f状态空间维度{n_states}动作空间维度{n_actions})# 更新n_states和n_actions到cfg参数中setattr(cfg, n_states, n_states)setattr(cfg, n_actions, n_actions) agent Agent(cfg)return env,agent训练可视化
import matplotlib.pyplot as plt
import seaborn as sns
class Config:def __init__(self) - None:self.env_name CartPole-v1 # 环境名字self.new_step_api False # 是否用gym的新apiself.algo_name PPO # 算法名字self.mode train # train or testself.seed 1 # 随机种子self.device cuda # device to useself.train_eps 200 # 训练的回合数self.test_eps 20 # 测试的回合数self.max_steps 200 # 每个回合的最大步数self.eval_eps 5 # 评估的回合数self.eval_per_episode 10 # 评估的频率self.gamma 0.99 # 折扣因子self.k_epochs 4 # 更新策略网络的次数self.actor_lr 0.0003 # actor网络的学习率self.critic_lr 0.0003 # critic网络的学习率self.eps_clip 0.2 # epsilon-clipself.entropy_coef 0.01 # entropy的系数self.update_freq 100 # 更新频率self.actor_hidden_dim 256 # actor网络的隐藏层维度self.critic_hidden_dim 256 # critic网络的隐藏层维度def smooth(data, weight0.9): 用于平滑曲线类似于Tensorboard中的smooth曲线last data[0] smoothed []for point in data:smoothed_val last * weight (1 - weight) * point # 计算平滑值smoothed.append(smoothed_val) last smoothed_val return smootheddef plot_rewards(rewards,cfg, tagtrain): 画图sns.set()plt.figure() # 创建一个图形实例方便同时多画几个图plt.title(f{tag}ing curve on {cfg.device} of {cfg.algo_name} for {cfg.env_name})plt.xlabel(epsiodes)plt.plot(rewards, labelrewards)plt.plot(smooth(rewards), labelsmoothed)plt.legend()# 获取参数
cfg Config()
# 训练
env, agent env_agent_config(cfg)
best_agent,res_dic train(cfg, env, agent)plot_rewards(res_dic[rewards], cfg, tagtrain)
# 测试
res_dic test(cfg, env, best_agent)
plot_rewards(res_dic[rewards], cfg, tagtest) # 画出结果状态空间维度4动作空间维度2 开始训练 回合10/200奖励11.00评估奖励29.20最佳评估奖励29.20更新模型 回合20/200奖励68.00评估奖励25.00最佳评估奖励29.20 回合30/200奖励60.00评估奖励26.20最佳评估奖励29.20 回合40/200奖励105.00评估奖励27.60最佳评估奖励29.20 回合50/200奖励26.00评估奖励60.60最佳评估奖励60.60更新模型 回合60/200奖励122.00评估奖励113.40最佳评估奖励113.40更新模型 回合70/200奖励65.00评估奖励38.00最佳评估奖励113.40 回合80/200奖励175.00评估奖励135.40最佳评估奖励135.40更新模型 回合90/200奖励200.00评估奖励177.20最佳评估奖励177.20更新模型 回合100/200奖励115.00评估奖励173.60最佳评估奖励177.20 回合110/200奖励200.00评估奖励183.20最佳评估奖励183.20更新模型 回合120/200奖励196.00评估奖励173.60最佳评估奖励183.20 回合130/200奖励46.00评估奖励61.40最佳评估奖励183.20 回合140/200奖励200.00评估奖励166.40最佳评估奖励183.20 回合150/200奖励172.00评估奖励154.40最佳评估奖励183.20 回合160/200奖励61.00评估奖励84.80最佳评估奖励183.20 回合170/200奖励127.00评估奖励181.60最佳评估奖励183.20 回合180/200奖励152.00评估奖励173.20最佳评估奖励183.20 回合190/200奖励200.00评估奖励200.00最佳评估奖励200.00更新模型 回合200/200奖励176.00评估奖励190.20最佳评估奖励200.00 完成训练 开始测试 回合1/20奖励200.00 回合2/20奖励200.00 回合3/20奖励200.00 回合4/20奖励200.00 回合5/20奖励200.00 回合6/20奖励200.00 回合7/20奖励200.00 回合8/20奖励200.00 回合9/20奖励200.00 回合10/20奖励200.00 回合11/20奖励200.00 回合12/20奖励200.00 回合13/20奖励200.00 回合14/20奖励200.00 回合15/20奖励200.00 回合16/20奖励200.00 回合17/20奖励200.00 回合18/20奖励200.00 回合19/20奖励200.00 回合20/20奖励200.00 完成测试 练习
为什么 DQN \text{DQN} DQN 和 DDPG \text{DDPG} DDPG 算法不使用重要性采样技巧呢 DQN \text{DQN} DQN 和 DDPG \text{DDPG} DDPG 是 off-policy \text{off-policy} off-policy算法它们通常不需要重要性采样来处理不同策略下的采样数据。相反它们使用目标网络和优势估计等技巧来提高训练的稳定性和性能。 PPO \text{PPO} PPO 算法原理上是 on-policy \text{on-policy} on-policy 的但它可以是 off-policy \text{off-policy} off-policy 的吗或者说可以用经验回放来提高训练速度吗?为什么提示是可以的但条件比较严格 跟 A2C \text{A2C} A2C 一样可以将经验回放与 PPO \text{PPO} PPO 结合创建一个 PPO with Experience Replay (PPO-ER) \text{PPO with Experience Replay (PPO-ER)} PPO with Experience Replay (PPO-ER) 算法。在 PPO-ER \text{PPO-ER} PPO-ER中智能体使用经验回放缓冲区中的数据来训练策略网络这样可以提高训练效率和稳定性。这种方法通常需要调整PPO的损失函数和采样策略以适应 off-policy \text{off-policy} off-policy 训练的要求需要谨慎调整。 PPO \text{PPO} PPO 算法更新过程中在将轨迹样本切分个多个小批量的时候可以将这些样本顺序打乱吗为什么 将轨迹样本切分成多个小批量时通常是可以将这些样本顺序打乱的这个过程通常称为样本随机化 sample shuffling \text{sample shuffling} sample shuffling 这样做的好处有降低样本相关性、减小过拟合风险以及增加训练多样性更全面地提高探索空间。 为什么说重要性采样是一种特殊的蒙特卡洛采样 估计期望值蒙特卡洛方法的核心目标之一是估计一个随机变量的期望值。蒙特卡洛采样通过从分布中生成大量的样本并求取这些样本的平均值来估计期望值。重要性采样也是通过从一个分布中生成样本但不是均匀地生成样本而是按照另一个分布的权重生成样本然后使用这些带权重的样本来估计期望值。 改进采样效率重要性采样的主要目的是改进采样效率。当我们有一个难以从中采样的分布时可以使用重要性采样来重新调整样本的权重以使估计更准确。这类似于在蒙特卡洛采样中调整样本大小以提高估计的精确性。 权重分布在重要性采样中我们引入了一个额外的权重分布用于指导采样过程。这个权重分布决定了每个样本的相对贡献以确保估计是无偏的。在蒙特卡洛采样中权重通常是均匀分布而在重要性采样中权重由分布的比率要估计的分布和采样分布之间的比例决定。 总结
PG方法的缺点是数据效率和鲁棒性不好。同时TRPO方法又比较复杂且不兼容dropout在深度神经网络训练过程中按照一定概率对网络单元进行丢弃和参数共享策略和值函数间。这篇论文提出了PPO算法它是对TRPO算法的改进更易于实现且数据效率更高。TRPO方法中通过使用约束而非惩罚项来保证策略更新的稳定性主要原因是作为惩罚项的话会引入权重因子而这个参数难以调节。TRPO中为了解优化问题先线性近似目标函数二阶近似约束最后通过conjugate gradient算法和line search求解。而这篇文章尝试通过一阶优化的方法来解。与TRPO中用约束来限制策略更新幅度不同PPO中采用了惩罚项。
