1.基本场景
不同与之前的1D问题，这次任务将问题拓展到2D上，如下图只考虑三个作战单元：蓝方战斗机Fighter、红方目标Target、红方导弹防御系统SAM。
- 目的：是训练Fighter使其能学到花费最小的时间代价使得Fighter绕过SAM的导弹火力范围，避免火力打击从而直接攻击并击毁Target的启发式原则。
- 限制：保证Fighter、SAM、Target在场景中的初始位置随机，但基本要使得其相对方位满足上图所示的关系。下表是三者之间的初始位置以及Fighter与SAM之间的火力范围：
- 控制：在每个时刻的Fighter的下一个航向(the fighter’s next heading)，是一个连续型控制变量 [ 0 , 2 π ] [0,2\pi] [0,2π]，是唯一的动作变量。
- 假设：在此假设Agent只能根据当前状态 s t s_t st下的每个时间点处的战斗机理想位置采取动作 a t a_t at，而与之前的Agent的状态 s i ( i = 0 , 1 , . . t − 1 ) s_{i}(i=0,1,..t-1) si(i=0,1,..t−1)无关。并且在原论文中，为了限制状态空间的尺寸，而将Fighter的速度设置成常量。而在我们这里，将在扩展部分考虑将Fighter的加速度 a t a_t at也纳入动作空间。同时SAM系统具有部分全向检测的能力，即对于只落在其射程内的Fighter都具有检测能力，而Fighter则对于SAM系统的状态则是全可观测的。
- 基本平台：原论文中采用的是AFGYM/AFSIM是基本平台。求解算法不再使用Q-Learning，由于其相比与策略梯度方法而言在求解高维状态空间的问题上不具备优势，其在使用带Reinforce的Q-Learning训练行为时，最后的结果不太理想，就像随机游走一样。并且由于决策的实时性导致GAN也是不太适用的，论文采用了A3C和PPO算法对原来问题做了训练。
2.基本模型
2.1 状态空间与动作空间
将原来场景中的状态向量
sts_tst由以下的11维向量组成:
注意：训练过程中除了0-1变量，所有的状态向量的值都被归一化了，否则将限制梯度的反向传播过程。在后面的编程中暂时不考虑SAM pointing、Target heading，因此状态空间被压缩到9维。这个可以作为后面改进的一个具体的方向。其中的Fighter heading作为一个基本的动作
ata_tat。
因此最终状态
sts_tst的状态空间:S=[0,1]8S=[0,1]^{8}S=[0,1]8,而动作ata_tat的动作空间：A=[0,1]A=[0,1]A=[0,1]映射到角度范围[−180。,180。][-180^{。},180^{。}][−180。,180。]。
2.2 奖励函数
每当无人机飞行一个时间步长
Δt\Delta tΔt，若未出现在SAM的火力射程之内，则返回奖励-1,回合继续。若出现在SAM的火力射程之内则返回奖励-50，回合继续。若出现在SAM的火力射程之内且被SAM击落，则返回-1000，回合结束。若出现在Target 附近返回50，回合继续。若出现在Target附近且成功打击到Target，返回1000，回合结束。
2.3 状态转移方程
模型只考虑Fighter匀速运动的情况，后面可以考虑施加突防效果，以此作为一个突破点的研究方向(2)。
假设Fighter的速度恒定为
vvv，ttt时刻Fighter的航向为ata_tat(动作变量)，Fighter的位置为(xJ(t),yJ(t))(x_J^{(t)},y_J^{(t)})(xJ(t),yJ(t))，则有：xJ(t+1)=xJ(t)+vΔtcosatyJ(t+1)=yJ(t)+vΔtsinatx_J^{(t+1)}=x_J^{(t)}+v\Delta t \cos a_t \\ y_J^{(t+1)}=y_J^{(t)}+v\Delta t \sin a_txJ(t+1)=xJ(t)+vΔtcosatyJ(t+1)=yJ(t)+vΔtsinat
Fighter的弹药量
wtw_{t}wt,如果ttt时刻Fighter打击目标则wt+1=wt−dwtw_{t+1}=w_t-dw_twt+1=wt−dwt。在此处我们不考虑弹药量的损失，考虑充分的弹药量，这里也可以作为一个突破点方向(3)。设SAM的火力范围区域为RRR，在这里考虑RRR为以SAM为中心,RSAMR_{SAM}RSAM为半径的圆形，其中RSAMR_{SAM}RSAM为Fighter的火力杀伤范围。
3.环境搭建
下面将直接给出代码：
classFighter:def__init__(self,L_limits=100,x=0,y=0,SAM_x=50,SAM_y=50,target_x =70,target_y=80,velocity=2175.0/3600,delta_t=10,dead=False,SAM_range=40,SAM_time=60,Fighter_range=10,firing_range=3.5,weapon_count=20):self.L_limits = L_limits # 限制范围self.x = x # 战斗机经度self.y = y # 战斗机纬度self.SAM_x = SAM_xself.SAM_y = SAM_yself.target_x = target_xself.target_y = target_y# 上面是位置的基本属性self.SAM_disable =Falseself.SAM_time = SAM_time # SAM系统的预备时间self.SAM_range = SAM_rangeself.firing_range = firing_range # 战斗机火力范围self.weapon_count = weapon_count # 战斗机弹药容量self.velocity = velocityself.delta_t = delta_tself.dead = deadself.wait_time =0self.burn_reward =-1000self.win_reward =1000self.reward =-1# 战斗机正常运行# 下面对战斗机的初始位置进行更新defreset(self):self.x = self.L_limits*np.random.random()self.y = self.L_limits*np.random.random()self.dead =Falseself.wait_time =0self.SAM_disable =Falsereturn self.x,self.y# 下面是状态转移方程defstep(self,figher_heading):# 下面采用弧度制figher_heading = np.clip(figher_heading,-np.pi,np.pi)self.x = self.x + self.velocity * self.delta_t*np.cos(figher_heading)self.y = self.y + self.velocity * self.delta_t*np.sin(figher_heading)# 限制战斗机的横向距离if self.x <=0:self.x =0elif self.x > self.L_limits:self.x = self.L_limits# 限制战斗机的纵向位置if self.y <=0:self.y =0elif self.y > self.L_limits:self.y = self.L_limits# 判断战斗机是否在区域内if self.SAM_disable ==False:d = np.sqrt((self.x - self.SAM_x)**2+(self.y - self.SAM_y)**2)if d <= self.SAM_range:self.wait_time = self.wait_time + self.delta_telse:# 一旦不在其SAM射程之内self.wait_time =0# 判断战斗机的等待时间if self.wait_time >= self.SAM_time:#当战斗机在SAM的时间内停留一段时间# 此时战斗机会被击落，返回极大奖励self.wait_time =0return(self.x,self.y),self.burn_reward,True# 当战斗机靠近SAM时可以立即击毁SAM(这是一个关键节点给50的奖励)if d <= self.firing_range and self.SAM_disable ==False:self.SAM_disable =Trueself.wait_time =0return(self.x,self.y),50,False# 当战斗机靠近target时if np.sqrt((self.target_x - self.x)**2+(self.target_y - self.y)**2)<= self.firing_range:return(self.x,self.y),self.win_reward,True# 战斗机正常运行时候，返回-1return(self.x,self.y),-1,False## 强化学习或许可以结合故障树进行策略分析!!!
下面利用随机游走给出测试代码：
x_array,y_array =[],[]rewards =0.0SAM_x,SAM_y =50,50target_x,target_y =80,70SAM_range =20fighter = Fighter(SAM_x=SAM_x,SAM_y=SAM_y,target_x=target_x,target_y=target_y,SAM_range=SAM_range)whileTrue:fighter_heading = np.random.random()*2*np.pi-np.pinext_state,reward,done = fighter.step(fighter_heading)rewards += rewardprint("当前Figher位置[{},{}],获得奖励{},累计奖励{}".format(next_state[0],next_state[1],reward,rewards))x_array.append(next_state[0])y_array.append(next_state[1])if done:breakimport matplotlib.pyplot as plt# 以火力射程画圆circle_x,circle_y =[],[]for theta in np.linspace(-np.pi,np.pi):circle_x.append(SAM_x + SAM_range*np.cos(theta))circle_y.append(SAM_y + SAM_range*np.sin(theta))plt.figure(figsize=(5,5))plt.plot(x_array,y_array,'b.-',linewidth=1)plt.plot(SAM_x,SAM_y,'ro',linewidth=5)plt.plot(circle_x,circle_y,'r.-',linewidth=1)plt.plot(target_x,target_y,'go',linewidth=5)plt.title(['one road'])
测试结果为：
以下情况是被SAM击落的：
以下情况时成功绕开SAM的：
2.实现
采用带优化的连续PPO算法求解上面的问题，需要导入库：
import torchimport matplotlib.pyplot as pltimport numpy as npfrom normalization import Normalization, RewardScalingfrom replaybuffer import ReplayBufferfrom ppo_continuous import PPO_continuousfrom Fighter_env import Fighterfrom CPPO_main import args_param,train_network,test_network
采用下面的主函数求解：
# 声明环境args = args_param(max_episode_steps=2000,batch_size=64,max_train_steps=5000,K_epochs=3)# 声明参数env = Fighter(delta_t=25,SAM_range=30,Fighter_range=10,SAM_time=30)env.test(show_pictures=True,show_tips=False)agent = train_network(args,env)x_array,y_array,_ = test_network(args,env,agent)
得到以下平均奖励的收敛曲线：
得到如下的测试结果：
看来用连续动作空间的PPO训练网络还是问题太大了!!!
3.代码
- 在normalization.py中保存以下代码作为标准化：
import numpy as npclassRunningMeanStd:# Dynamically calculate mean and stddef__init__(self, shape):# shape:the dimension of input dataself.n =0self.mean = np.zeros(shape)self.S = np.zeros(shape)self.std = np.sqrt(self.S)defupdate(self, x):x = np.array(x)self.n +=1if self.n ==1:self.mean = xself.std = xelse:old_mean = self.mean.copy()self.mean = old_mean +(x - old_mean)/ self.nself.S = self.S +(x - old_mean)*(x - self.mean)self.std = np.sqrt(self.S / self.n)classNormalization:def__init__(self, shape):self.running_ms = RunningMeanStd(shape=shape)def__call__(self, x, update=True):# Whether to update the mean and std,during the evaluating,update=Falseif update:self.running_ms.update(x)x =(x - self.running_ms.mean)/(self.running_ms.std +1e-8)return xclassRewardScaling:def__init__(self, shape, gamma):self.shape = shape # reward shape=1self.gamma = gamma # discount factorself.running_ms = RunningMeanStd(shape=self.shape)self.R = np.zeros(self.shape)def__call__(self, x):self.R = self.gamma * self.R + xself.running_ms.update(self.R)x = x /(self.running_ms.std +1e-8)# Only divided stdreturn xdefreset(self):# When an episode is done,we should reset 'self.R'self.R = np.zeros(self.shape)
- 在Fighter_env.py中写入以下代码,用来作为定战斗机环境：
import numpy as npimport matplotlib.pyplot as plt# 下面类用来定义环境classFighter:def__init__(self,L_limits=100,x=0,y=0,SAM_x=50,SAM_y=50,target_x =70,target_y=80,velocity=2175.0/3600,delta_t=10,dead=False,SAM_range=40,SAM_time=60,Fighter_range=10,weapon_count=20):self.L_limits = L_limits # 限制范围self.x = x # 战斗机经度self.y = y # 战斗机纬度self.SAM_x = SAM_xself.SAM_y = SAM_yself.target_x = target_xself.target_y = target_y# 上面是位置的基本属性self.SAM_disable =Falseself.SAM_time = SAM_time # SAM系统的预备时间self.SAM_range = SAM_rangeself.firing_range = Fighter_range # 战斗机火力范围self.weapon_count = weapon_count # 战斗机弹药容量self.velocity = velocityself.delta_t = delta_tself.dead = deadself.wait_time =0self.burn_reward =-1000self.win_reward =1000self.reward =-1# 战斗机正常运行# 下面声明其基本属性self.observation_space = np.array([[0,self.L_limits],[0,self.L_limits]])self.action_space = np.array([[-np.pi,np.pi]])# 下面对战斗机的初始位置进行更新defreset(self):# self.x = self.L_limits*np.random.random()# self.y = self.L_limits*np.random.random()self.x =0self.y =0self.dead =Falseself.wait_time =0self.SAM_disable =Falsereturn np.array([self.x,self.y],dtype=np.float32)# 下面是状态转移方程defstep(self,figher_heading):# 下面采用弧度制figher_heading = np.clip(figher_heading,-np.pi,np.pi)self.x = self.x + self.velocity * self.delta_t*np.cos(figher_heading)self.y = self.y + self.velocity * self.delta_t*np.sin(figher_heading)self.x = np.clip(self.x,a_min=0.0,a_max=self.L_limits)self.y = np.clip(self.y,a_min=0.0,a_max=self.L_limits)# 当战斗机靠近target时if np.linalg.norm(np.array([self.target_x - self.x, self.target_y - self.y], dtype=np.float32))<= self.firing_range:return np.array([self.x, self.y], dtype=np.float32), self.win_reward,True# 判断战斗机是否在区域内d = np.linalg.norm(np.array([self.x - self.SAM_x, self.y - self.SAM_y], dtype=np.float32))if self.SAM_disable ==False:if d <= self.SAM_range:self.wait_time = self.wait_time + self.delta_telse:# 一旦不在其SAM射程之内self.wait_time =0# 判断战斗机的等待时间if self.wait_time >= self.SAM_time:#当战斗机在SAM的时间内停留一段时间# 此时战斗机会被击落，返回极大奖励self.wait_time =0return np.array([self.x,self.y],dtype=np.float32),self.burn_reward,True# 当战斗机靠近SAM时可以立即击毁SAM(这是一个关键节点给50的奖励)if d <= self.firing_range and self.SAM_disable ==False:self.SAM_disable =Trueself.wait_time =0return np.array([self.x,self.y],dtype=np.float32),50,False# 战斗机正常运行时候，返回-1return np.array([self.x,self.y],dtype=np.float32),-1,False## 强化学习或许可以结合故障树进行策略分析!!!# 下面将测试一个回合deftest(self,show_tips=True,show_pictures=True):x_array,y_array=[],[]rewards =0.0whileTrue:fighter_headings = np.random.random()*2*np.pi - np.pinext_state,reward,done = self.step(fighter_headings)rewards += rewardif show_tips:print("当前Figher位置[{},{}],获得奖励{},累计奖励{}".format(next_state[0], next_state[1], reward, rewards))x_array.append(next_state[0])y_array.append(next_state[1])if done:break# 如果要画图就以火力范围画图if show_pictures:circle_x, circle_y =[],[]for theta in np.linspace(-np.pi, np.pi):circle_x.append(self.SAM_x + self.SAM_range * np.cos(theta))circle_y.append(self.SAM_y + self.SAM_range * np.sin(theta))plt.figure(figsize=(5,5))plt.plot(x_array, y_array,'b.-', linewidth=1)plt.plot(self.SAM_x, self.SAM_y,'ro', linewidth=5)plt.plot(circle_x, circle_y,'r.-', linewidth=1)plt.plot(self.target_x, self.target_y,'go', linewidth=5)plt.plot(x_array[-1],y_array[-1],'m*',linewidth=5)plt.title(['one road'])plt.show()
在replaybuffer.py中存储缓存区replaybuffer类:
def__init__(self, args):self.state_dim = args.state_dimself.action_dim = args.action_dimself.batch_size = args.batch_sizeself.s = np.zeros((args.batch_size, args.state_dim))self.a = np.zeros((args.batch_size, args.action_dim))self.a_logprob = np.zeros((args.batch_size, args.action_dim))self.r = np.zeros((args.batch_size,1))self.s_ = np.zeros((args.batch_size, args.state_dim))self.dw = np.zeros((args.batch_size,1))self.done = np.zeros((args.batch_size,1))self.count =0defstore(self, s, a, a_logprob, r, s_, dw, done):self.s[self.count]= sself.a[self.count]= aself.a_logprob[self.count]= a_logprobself.r[self.count]= rself.s_[self.count]= s_self.dw[self.count]= dwself.done[self.count]= doneself.count +=1defnumpy_to_tensor(self):s = torch.tensor(self.s, dtype=torch.float)a = torch.tensor(self.a, dtype=torch.float)a_logprob = torch.tensor(self.a_logprob, dtype=torch.float)r = torch.tensor(self.r, dtype=torch.float)s_ = torch.tensor(self.s_, dtype=torch.float)dw = torch.tensor(self.dw, dtype=torch.float)done = torch.tensor(self.done, dtype=torch.float)return s, a, a_logprob, r, s_, dw, done
- 在以下文件ppo_continuous.py中包含带优化的连续PPO算法的代码：
import torchimport torch.nn.functional as Ffrom torch.utils.data.sampler import BatchSampler, SubsetRandomSamplerimport torch.nn as nnfrom torch.distributions import Beta, Normal# Trick 8: orthogonal initializationdeforthogonal_init(layer, gain=1.0):nn.init.orthogonal_(layer.weight, gain=gain)nn.init.constant_(layer.bias,0)classActor_Beta(nn.Module):def__init__(self, args):super(Actor_Beta, self).__init__()self.fc1 = nn.Linear(args.state_dim, args.hidden_width)self.fc2 = nn.Linear(args.hidden_width, args.hidden_width)self.alpha_layer = nn.Linear(args.hidden_width, args.action_dim)self.beta_layer = nn.Linear(args.hidden_width, args.action_dim)self.activate_func =[nn.ReLU(), nn.Tanh()][args.use_tanh]# Trick10: use tanhif args.use_orthogonal_init:print("------use_orthogonal_init------")orthogonal_init(self.fc1)orthogonal_init(self.fc2)orthogonal_init(self.alpha_layer, gain=0.01)orthogonal_init(self.beta_layer, gain=0.01)defforward(self, s):s = self.activate_func(self.fc1(s))s = self.activate_func(self.fc2(s))# alpha and beta need to be larger than 1,so we use 'softplus' as the activation function and then plus 1alpha = F.softplus(self.alpha_layer(s))+1.0beta = F.softplus(self.beta_layer(s))+1.0return alpha, betadefget_dist(self, s):alpha, beta = self.forward(s)dist = Beta(alpha, beta)return distdefmean(self, s):alpha, beta = self.forward(s)mean = alpha /(alpha + beta)# The mean of the beta distributionreturn meanclassActor_Gaussian(nn.Module):def__init__(self, args):super(Actor_Gaussian, self).__init__()self.max_action = args.max_actionself.fc1 = nn.Linear(args.state_dim, args.hidden_width)self.fc2 = nn.Linear(args.hidden_width, args.hidden_width)self.mean_layer = nn.Linear(args.hidden_width, args.action_dim)self.log_std = nn.Parameter(torch.zeros(1, args.action_dim))# We use 'nn.Parameter' to train log_std automaticallyself.activate_func =[nn.ReLU(), nn.Tanh()][args.use_tanh]# Trick10: use tanhif args.use_orthogonal_init:print("------use_orthogonal_init------")orthogonal_init(self.fc1)orthogonal_init(self.fc2)orthogonal_init(self.mean_layer, gain=0.01)defforward(self, s):s = self.activate_func(self.fc1(s))s = self.activate_func(self.fc2(s))mean = self.max_action * torch.tanh(self.mean_layer(s))# [-1,1]->[-max_action,max_action]return meandefget_dist(self, s):mean = self.forward(s)log_std = self.log_std.expand_as(mean)# To make 'log_std' have the same dimension as 'mean'std = torch.exp(log_std)# The reason we train the 'log_std' is to ensure std=exp(log_std)>0dist = Normal(mean, std)# Get the Gaussian distributionreturn distclassCritic(nn.Module):def__init__(self, args):super(Critic, self).__init__()self.fc1 = nn.Linear(args.state_dim, args.hidden_width)self.fc2 = nn.Linear(args.hidden_width, args.hidden_width)self.fc3 = nn.Linear(args.hidden_width,1)self.activate_func =[nn.ReLU(), nn.Tanh()][args.use_tanh]# Trick10: use tanhif args.use_orthogonal_init:print("------use_orthogonal_init------")orthogonal_init(self.fc1)orthogonal_init(self.fc2)orthogonal_init(self.fc3)defforward(self, s):s = self.activate_func(self.fc1(s))s = self.activate_func(self.fc2(s))v_s = self.fc3(s)return v_sclassPPO_continuous():def__init__(self, args):self.policy_dist = args.policy_distself.max_action = args.max_actionself.batch_size = args.batch_sizeself.mini_batch_size = args.mini_batch_sizeself.max_train_steps = args.max_train_stepsself.lr_a = args.lr_a # Learning rate of actorself.lr_c = args.lr_c # Learning rate of criticself.gamma = args.gamma # Discount factorself.lamda = args.lamda # GAE parameterself.epsilon = args.epsilon # PPO clip parameterself.K_epochs = args.K_epochs # PPO parameterself.entropy_coef = args.entropy_coef # Entropy coefficientself.set_adam_eps = args.set_adam_epsself.use_grad_clip = args.use_grad_clipself.use_lr_decay = args.use_lr_decayself.use_adv_norm = args.use_adv_normif self.policy_dist =="Beta":self.actor = Actor_Beta(args)else:self.actor = Actor_Gaussian(args)self.critic = Critic(args)if self.set_adam_eps:# Trick 9: set Adam epsilon=1e-5self.optimizer_actor = torch.optim.Adam(self.actor.parameters(), lr=self.lr_a, eps=1e-5)self.optimizer_critic = torch.optim.Adam(self.critic.parameters(), lr=self.lr_c, eps=1e-5)else:self.optimizer_actor = torch.optim.Adam(self.actor.parameters(), lr=self.lr_a)self.optimizer_critic = torch.optim.Adam(self.critic.parameters(), lr=self.lr_c)defevaluate(self, s):# When evaluating the policy, we only use the means = torch.unsqueeze(torch.tensor(s, dtype=torch.float),0)if self.policy_dist =="Beta":a = self.actor.mean(s).detach().numpy().flatten()else:a = self.actor(s).detach().numpy().flatten()return adefchoose_action(self, s):s = torch.unsqueeze(torch.tensor(s, dtype=torch.float),0)if self.policy_dist =="Beta":with torch.no_grad():dist = self.actor.get_dist(s)a = dist.sample()# Sample the action according to the probability distributiona_logprob = dist.log_prob(a)# The log probability density of the actionelse:with torch.no_grad():dist = self.actor.get_dist(s)a = dist.sample()# Sample the action according to the probability distributiona = torch.clamp(a,-self.max_action, self.max_action)# [-max,max]a_logprob = dist.log_prob(a)# The log probability density of the actionreturn a.numpy().flatten(), a_logprob.numpy().flatten()defupdate(self, replay_buffer, total_steps):s, a, a_logprob, r, s_, dw, done = replay_buffer.numpy_to_tensor()# Get training data"""Calculate the advantage using GAE'dw=True' means dead or win, there is no next state s''done=True' represents the terminal of an episode(dead or win or reaching the max_episode_steps). When calculating the adv, if done=True, gae=0"""adv =[]gae =0with torch.no_grad():# adv and v_target have no gradientvs = self.critic(s)vs_ = self.critic(s_)deltas = r + self.gamma *(1.0- dw)* vs_ - vsfor delta, d inzip(reversed(deltas.flatten().numpy()),reversed(done.flatten().numpy())):gae = delta + self.gamma * self.lamda * gae *(1.0- d)adv.insert(0, gae)adv = torch.tensor(adv, dtype=torch.float).view(-1,1)v_target = adv + vsif self.use_adv_norm:# Trick 1:advantage normalizationadv =((adv - adv.mean())/(adv.std()+1e-5))# Optimize policy for K epochs:for _ inrange(self.K_epochs):# Random sampling and no repetition. 'False' indicates that training will continue even if the number of samples in the last time is less than mini_batch_sizefor index in BatchSampler(SubsetRandomSampler(range(self.batch_size)), self.mini_batch_size,False):dist_now = self.actor.get_dist(s[index])dist_entropy = dist_now.entropy().sum(1, keepdim=True)# shape(mini_batch_size X 1)a_logprob_now = dist_now.log_prob(a[index])# a/b=exp(log(a)-log(b)) In multi-dimensional continuous action space，we need to sum up the log_probratios = torch.exp(a_logprob_now.sum(1, keepdim=True)- a_logprob[index].sum(1, keepdim=True))# shape(mini_batch_size X 1)surr1 = ratios * adv[index]# Only calculate the gradient of 'a_logprob_now' in ratiossurr2 = torch.clamp(ratios,1- self.epsilon,1+ self.epsilon)* adv[index]actor_loss =-torch.min(surr1, surr2)- self.entropy_coef * dist_entropy # Trick 5: policy entropy# Update actorself.optimizer_actor.zero_grad()actor_loss.mean().backward()if self.use_grad_clip:# Trick 7: Gradient cliptorch.nn.utils.clip_grad_norm_(self.actor.parameters(),0.5)self.optimizer_actor.step()v_s = self.critic(s[index])critic_loss = F.mse_loss(v_target[index], v_s)# Update criticself.optimizer_critic.zero_grad()critic_loss.backward()if self.use_grad_clip:# Trick 7: Gradient cliptorch.nn.utils.clip_grad_norm_(self.critic.parameters(),0.5)self.optimizer_critic.step()if self.use_lr_decay:# Trick 6:learning rate Decayself.lr_decay(total_steps)deflr_decay(self, total_steps):lr_a_now = self.lr_a *(1- total_steps / self.max_train_steps)lr_c_now = self.lr_c *(1- total_steps / self.max_train_steps)for p in self.optimizer_actor.param_groups:p['lr']= lr_a_nowfor p in self.optimizer_critic.param_groups:p['lr']= lr_c_now
- 下面代码用来作为训练和测试网络的代码CPPO_main.py的代码：
import matplotlib.pyplot as pltimport numpy as npfrom normalization import Normalization, RewardScalingfrom replaybuffer import ReplayBufferfrom ppo_continuous import PPO_continuousfrom Fighter_env import Fighter# 先定义一个参数类，用来储存超参数classargs_param(object):def__init__(self,max_train_steps=int(3e6),evaluate_freq=5e3,save_freq=20,policy_dist="Gaussian",batch_size=2048,mini_batch_size=64,hidden_width=64,lr_a=3e-4,lr_c=3e-4,gamma=0.99,lamda=0.95,epsilon=0.2,K_epochs=10,max_episode_steps =1000,use_adv_norm=True,use_state_norm=True,use_reward_norm=False,use_reward_scaling=True,entropy_coef=0.01,use_lr_decay=True,use_grad_clip=True,use_orthogonal_init=True,set_adam_eps=True,use_tanh=True):self.max_train_steps = max_train_stepsself.evaluate_freq = evaluate_freqself.save_freq = save_freqself.policy_dist = policy_distself.batch_size = batch_sizeself.mini_batch_size = mini_batch_sizeself.hidden_width = hidden_widthself.lr_a = lr_aself.lr_c = lr_cself.gamma = gammaself.lamda = lamdaself.epsilon = epsilonself.K_epochs = K_epochsself.use_adv_norm = use_adv_normself.use_state_norm = use_state_normself.use_reward_norm = use_reward_normself.use_reward_scaling = use_reward_scalingself.entropy_coef = entropy_coefself.use_lr_decay = use_lr_decayself.use_grad_clip = use_grad_clipself.use_orthogonal_init = use_orthogonal_initself.set_adam_eps = set_adam_epsself.use_tanh = use_tanhself.max_episode_steps = max_episode_stepsdefprint_information(self):print("Maximum number of training steps:",self.max_train_steps)print("Evaluate the policy every 'evaluate_freq' steps:",self.evaluate_freq)print("Save frequency:",self.save_freq)print("Beta or Gaussian:",self.policy_dist)print("Batch size:",self.batch_size)print("Minibatch size:",self.mini_batch_size)print("The number of neurons in hidden layers of the neural network:",self.hidden_width)print("Learning rate of actor:",self.lr_a)print("Learning rate of critic:",self.lr_c)print("Discount factor:",self.gamma)print("GAE parameter:",self.lamda)print("PPO clip parameter:",self.epsilon)print("PPO parameter:",self.K_epochs)print("Trick 1:advantage normalization:",self.use_adv_norm)print("Trick 2:state normalization:",self.use_state_norm)print("Trick 3:reward normalization:",self.use_reward_norm)print("Trick 4:reward scaling:",self.use_reward_scaling)print("Trick 5: policy entropy:",self.entropy_coef)print("Trick 6:learning rate Decay:",self.use_lr_decay)print("Trick 7: Gradient clip:",self.use_grad_clip)print("Trick 8: orthogonal initialization:",self.use_orthogonal_init)print("Trick 9: set Adam epsilon=1e-5:",self.set_adam_eps)print("Trick 10: tanh activation function:",self.use_tanh)# 下面函数用来训练网络deftrain_network(args,env,show_picture=True):epsiode_rewards =[]epsiode_mean_rewards =[]# 下面将导入env环境参数args.state_dim = env.observation_space.shape[0]args.action_dim = env.action_space.shape[0]args.max_action =float(env.action_space[0][1])# 下面将定义一个缓冲区replay_buffer = ReplayBuffer(args)# 下面将定义一个PPO智能体类agent = PPO_continuous(args)# 下面将采用Trick 2技巧标准化state_norm = Normalization(shape=args.state_dim)if args.use_reward_norm:reward_norm = Normalization(shape=1)elif args.use_reward_scaling:# Trick 4:reward scalingreward_scaling = RewardScaling(shape=1, gamma=args.gamma)# 下面开始进行训练过程for epsiode inrange(args.max_train_steps):# 每个回合首先对值进行初始化epsiode_reward =0.0done =Falseepsiode_count =0# 再赋予一个新的初始状态s = env.reset()# 对状态进行标准化操作if args.use_state_norm:s = state_norm(s)if args.use_reward_scaling:reward_scaling.reset()# 设置一个死循环，后面若跳出便在死循环中跳出whileTrue:# 每执行一个回合，count次数加1epsiode_count +=1a,a_logprob = agent.choose_action(s)# 根据参数的不同选择输出是高斯分布/Beta分布调整if args.policy_dist =="Beta":action =2*(a-0.5)*args.max_actionelse:action = a# 下面是执行环境交互操作s_,r,done = env.step(action)## !!! 这里的环境是自己搭建的，输出每个人都不一样s_ = s_.T[0]## !!! 这一部起始没必要，由于我的环境有问题所以加的这一步epsiode_reward += r# 下面考虑状态标准化的情况if args.use_state_norm:s_ = state_norm(s_)# 下面考虑奖励标准化的可能性if args.use_reward_norm:r = reward_norm(r)elif args.use_reward_scaling:r = reward_scaling(r)# 下面考虑回合的最大运行次数(只要回合结束或者超过最大回合运行次数)if done or epsiode_count >= args.max_episode_steps:dw =Trueelse:dw =False# 将经验存入replayBuffer中replay_buffer.store(s,action,a_logprob,r,s_,dw,done)# 重新赋值状态s = s_# 当replaybuffer尺寸到达batchsize便会开始训练if replay_buffer.count == args.batch_size:agent.update(replay_buffer,epsiode)replay_buffer.count =0# 如果回合结束便退出if done:epsiode_rewards.append(epsiode_reward)epsiode_mean_rewards.append(np.mean(epsiode_rewards))print("第{}次训练的奖励为{:.2f},平均奖励为{:.2f}".format(epsiode,epsiode_reward, epsiode_mean_rewards[-1]))break# 如果需要画图的话if show_picture:plt.plot(epsiode_rewards)plt.plot(epsiode_mean_rewards)plt.xlabel("epsiodes")plt.ylabel("rewards")plt.title("Continuous PPO With Optimization")plt.legend(["rewards","mean_rewards"])plt.show()return agent# 下面将用训练好的网络跑一次例程deftest_network(args,env,agent,show_pictures=True):epsiode_reward =0.0done =Falseepsiode_count =0state_norm = Normalization(shape=args.state_dim)# 再赋予一个新的初始状态s = env.reset()x_array, y_array =[s[0]],[s[1]]if args.use_reward_norm:reward_norm = Normalization(shape=1)s = state_norm(s)elif args.use_reward_scaling:# Trick 4:reward scalingreward_scaling = RewardScaling(shape=1, gamma=args.gamma)reward_scaling.reset()# 设置一个死循环，后面若跳出便在死循环中跳出whileTrue:# 每执行一个回合，count次数加1epsiode_count +=1a, a_logprob = agent.choose_action(s)# 根据参数的不同选择输出是高斯分布/Beta分布调整if args.policy_dist =="Beta":action =2*(a -0.5)* args.max_actionelse:action = a# 下面是执行环境交互操作s_, r, done = env.step(action)## !!! 这里的环境是自己搭建的，输出每个人都不一样epsiode_reward += rs_ = s_.T[0]## !!! 这一部起始没必要，由于我的环境有问题所以加的这一步x_array.append(s_[0])y_array.append(s_[1])epsiode_reward += r# 下面考虑状态标准化的情况if args.use_state_norm:s_ = state_norm(s_)# 重新赋值状态s = s_# 如果回合结束便退出if done or epsiode_count>=2000:print("当前测试得分为{}".format(epsiode_reward))break# 下面开始画图if show_pictures:circle_x, circle_y =[],[]for theta in np.linspace(-np.pi, np.pi):circle_x.append(env.SAM_x + env.SAM_range * np.cos(theta))circle_y.append(env.SAM_y + env.SAM_range * np.sin(theta))plt.plot(x_array, y_array,'b.-', linewidth=1)plt.plot(env.SAM_x, env.SAM_y,'ro', linewidth=5)plt.plot(circle_x, circle_y,'r.-', linewidth=1)plt.plot(env.target_x, env.target_y,'go', linewidth=5)plt.plot(x_array[-1], y_array[-1],'m*', linewidth=5)plt.title("one road rewards:{:.2f}".format(epsiode_reward))plt.show()return x_array,y_array,epsiode_reward# 下面是主函数if __name__ =="__main__":# 声明环境args = args_param(max_episode_steps=1000,batch_size=64,max_train_steps=1000,K_epochs=3)# 声明参数env = Fighter(delta_t=25,SAM_range=30,Fighter_range=10,SAM_time=30)env.test(show_pictures=True,show_tips=False)agent = train_network(args,env)x_array,y_array,_ = test_network(args,env,agent)
本文转载自: https://blog.csdn.net/shengzimao/article/details/126787045
版权归原作者 赛亚茂 所有， 如有侵权，请联系我们删除。
版权归原作者 赛亚茂 所有， 如有侵权，请联系我们删除。