文章目录
- gym环境: https://www.gymlibrary.dev/
- 环境安装:- 我的版本:packagemodulegym0.24.0ale-py0.7.5torch1.11.0torchvision0.12.0tensorboard2.6.0- 安装方法：
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple gympip install --no-index -f https://github.com/Kojoley/atari-py/releases atari_pypip install gym[atari]pip uninstall ale-pypip install ale-py安装box2d: 可能会遇到building wheel failed for box2d在 https://www.lfd.uci.edu/~gohlke/pythonlibs/ 下载相应的 PyBox2D的whl文件然后在命令行:pip install D:\FILES\PYTHON_PROJECTS\Box2D-2.3.10-cp37-cp37m-win_amd64.whl
一、Sarsa (悬崖问题)
1.1 CliffWalking-v0环境介绍
在一个4x12的网格中，智能体以网格的左下角位置为起点，以网格的下角位置为终点，目标是移动智能体到达终点位置，智能体每次可以在上、下、左、右这4个方向中移动一步，每移动一步会得到 -1 的奖励。
- 如果智能体“掉入悬崖” ，会立即回到起点位置，并得到-100单位的奖励
- 当智能体移动到终点时，该回合结束，该回合总奖励为各步奖励之和
import gymenv = gym.make("CliffWalking-v0")observation = env.reset()env.render()
- 从起点到终点最少需要13步，每步得到-1的reward。我们的目标也是要通过RL训练出一个模型，使得该模型能在测试中一个episode的reward能够接近于-13左右。
1.2 Sarsa算法流程
算法参数: 步长
α<1\alpha<1α<1 极小值ϵ\epsilonϵ （两个超参数)对于所有
Q(s,a)Q(s,a)Q(s,a)随机初始化，终点处$ Q(s_{end},a) = 0$
for (each trajectory):
初始化
SSSat=ϵ−greedy(st)a_t = \epsilon -greedy \quad(s_t)at=ϵ−greedy(st)for (each step):
执行
ata_tat，得到(rt+1,st+1)(r_{t+1},s_{t+1})(rt+1,st+1)at+1=ϵ−greedy(st+1)a_{t+1} = \epsilon -greedy \quad(s_{t+1})at+1=ϵ−greedy(st+1)Q(st,at)=Q(st,at)+α[rt+1+γQ(st+1,at+1)−Q(st,at)]Q(s_{t},a_{t})=Q(s_{t},a_{t})+\alpha[r_{t+1}+\gamma Q(s_{t+1},a_{t+1})-Q(s_{t},a_{t})]Q(st,at)=Q(st,at)+α[rt+1+γQ(st+1,at+1)−Q(st,at)]st=st+1,at=at+1s_t = s_{t+1},a_t = a_{t+1}st=st+1,at=at+1
1.3 具体代码
import numpy as npimport gymimport timeclassSarsaAgent:def__init__(self, obs_n, act_n, learning_rate=0.01, gamma=0.9, e_greed=0.1):self.act_n = act_nself.lr = learning_rateself.gamma = gammaself.epsilon = e_greedself.Q = np.zeros((obs_n, act_n))# e_greed:根据s_t,选择a_tdefsample(self,obs):if np.random.uniform(0,1)<(1.0- self.epsilon):action = self.predict(obs)else:action = np.random.choice(self.act_n)# 0,1,2,3return action# a_t = argmax Q(s)defpredict(self, obs):Q_list = self.Q[obs,:]#当前s下所有a对应的Q值maxQ = np.max(Q_list)action_list = np.where(Q_list == maxQ)[0]# action_list=所有=Qmax的索引action = np.random.choice(action_list)return actiondeflearn(self, obs, action, reward, next_obs, next_action, done):# (S,A,R,S,A)'''done: episode是否结束'''predict_Q = self.Q[obs,action]if done:target_Q = rewardelse:target_Q = reward + self.gamma * self.Q[next_obs,next_action]# 更新Q表格self.Q[obs,action]+= self.lr *(target_Q - predict_Q)defsave(self):npy_file ='./q-table.npy'np.save(npy_file, self.Q)print(npy_file +' saved.')defload(self, npy_file='./q_table.npy'):self.Q = np.load(npy_file)print(npy_file +' loaded.')defrun_episode(env, agent, render=False):total_steps =0# 记录当前episode走了多少steptotal_reward =0obs = env.reset()action = agent.sample(obs)whileTrue:next_obs, reward, done, _ = env.step(action)next_action = agent.sample(next_obs)agent.learn(obs, action, reward, next_obs, next_action, done)action = next_actionobs = next_obstotal_reward += rewardtotal_steps +=1if render:env.render()time.sleep(0.)if done:breakreturn total_reward, total_stepsdeftest_episode(env, agent):total_steps =0# 记录当前episode走了多少steptotal_reward =0obs = env.reset()whileTrue:action = agent.predict(obs)next_obs, reward, done, _ = env.step(action)total_reward += rewardtotal_steps +=1obs = next_obstime.sleep(0.5)env.render()if done:breakreturn total_reward, total_stepsdefmain():env = gym.make("CliffWalking-v0")agent = SarsaAgent(obs_n=env.observation_space.n,act_n=env.action_space.n,learning_rate=0.025, gamma=0.9, e_greed=0.1)for episode inrange(1000):total_reward, total_steps = run_episode(env, agent,False)print('Episode %s: total_steps = %s , total_reward = %.1f'%(episode, total_steps, total_reward))test_episode(env, agent)main()
1.4 演示效果
训练了1000个episode,
reward=−23reward=-23reward=−23
二、Q-Learning (悬崖问题)
2.1 CliffWalking-v0环境介绍
(介绍见1.1)
2.2 Q-Learning算法流程
(Q-Learning其实真正执行的策略和Sarsa是一样的，只不过学习的策略是保守的最优策略)
算法参数: 步长
α<1\alpha<1α<1 极小值ϵ\epsilonϵ （两个超参数)对于所有
Q(s,a)Q(s,a)Q(s,a)随机初始化，终点处Q(send,a)=0Q(s_{end},a) = 0Q(send,a)=0for (each trajectory):
初始化
SSSfor (each step):
at=ϵ−greedy(st)a_{t} = \epsilon -greedy \quad(s_{t})at=ϵ−greedy(st)（行为策略）执行
ata_tat，得到(rt+1,st+1)(r_{t+1},s_{t+1})(rt+1,st+1)Q(st,at)=Q(st,at)+α[rt+1+γmaxaQ(st+1,a)−Q(st,at)]Q(s_{t},a_{t})=Q(s_{t},a_{t})+\alpha[r_{t+1}+\gamma \underset{a}{max}Q(s_{t+1},a)-Q(s_{t},a_{t})]Q(st,at)=Q(st,at)+α[rt+1+γamaxQ(st+1,a)−Q(st,at)]st=st+1s_t = s_{t+1}st=st+1
2.3 具体代码
import numpy as npimport gymimport timeclassQLearningAgent:def__init__(self, obs_n, act_n, learning_rate=1e-2, gamma=0.9, e_greed=0.1):self.act_n = act_n # 动作维度，有几个动作可选self.lr = learning_rate # 学习率self.gamma = gamma # reward的衰减率self.epsilon = e_greed # 按一定概率随机选动作self.Q = np.zeros((obs_n, act_n))defsample(self, obs):if np.random.uniform(0,1)<(1.0- self.epsilon):# 根据table的Q值选动作action = self.predict(obs)else:action = np.random.choice(self.act_n)# 有一定概率随机探索选取一个动作return action# 根据输入观察值，预测输出的动作值defpredict(self, obs):Q_list = self.Q[obs,:]maxQ = np.max(Q_list)action_list = np.where(Q_list == maxQ)[0]# maxQ可能对应多个actionaction = np.random.choice(action_list)return actiondeflearn(self, obs, action, reward, next_obs, done):#(S,A,R,S)predict_Q = self.Q[obs, action]if done:target_Q = rewardelse:target_Q = reward + self.gamma * np.max(self.Q[next_obs,:])self.Q[obs, action]+= self.lr *(target_Q - predict_Q)defsave(self):npy_file ='./q-table.npy'np.save(npy_file, self.Q)print(npy_file +' saved.')defload(self, npy_file='./q_table.npy'):self.Q = np.load(npy_file)print(npy_file +' loaded.')defrun_episode(env, agent, render=False):# 其实真正执行的策略和Sarsa是一样的，只不过学习的策略是保守的最优策略total_steps =0total_reward =0obs = env.reset()whileTrue:action = agent.sample(obs)next_obs, reward, done, _ = env.step(action)agent.learn(obs, action, reward, next_obs, done)obs = next_obstotal_reward += rewardtotal_steps +=1if render:env.render()if done:breakreturn total_reward, total_stepsdeftest_episode(env, agent):total_reward =0obs = env.reset()whileTrue:action = agent.predict(obs)# greedynext_obs, reward, done, _ = env.step(action)total_reward += rewardobs = next_obstime.sleep(0.5)env.render()if done:breakreturn total_rewarddefmain():env = gym.make("CliffWalking-v0")# 0 up, 1 right, 2 down, 3 left# 创建一个agent实例，输入超参数agent = QLearningAgent(obs_n=env.observation_space.n,act_n=env.action_space.n,learning_rate=0.1,gamma=0.9,e_greed=0.1)# 训练500个episode，打印每个episode的分数for episode inrange(500):ep_reward, ep_steps = run_episode(env, agent,False)print('Episode %s: steps = %s , reward = %.1f'%(episode, ep_steps, ep_reward))# 全部训练结束，查看算法效果test_reward = test_episode(env, agent)print('test reward = %.1f'%(test_reward))main()
2.4 演示效果
三、PG 策略梯度 (倒立摆)
3.1 CartPole-v1环境介绍
(Cart Pole - Gym Documentation (gymlibrary.dev))
一根杆通过一个未驱动的关节连接到一辆小车上，小车沿着一条无摩擦的轨道移动。将钟摆垂直放置在推车上，目标是通过在推车上施加左右方向的力来平衡杆。
倒立摆:
- **obs: (1,4)**NumObservationMinMax0Cart Position0-4.84.81Cart Velocity-InfInf2Pole Angle-0.418 rad0.418 rad3Pole Angular Velocity-InfInf
- **action: (1,2)**动作空间是离散的:NumAction0向左推车1向右推车
- reward每活着经过一个时间步，奖励 + 1。
- 终止条件:- ① Pole Angle > 12°- ② |水平位置|>2.4’- ③ 超过500步
3.2 PG算法流程(REINFORCE)
输入: 可微调的策略参数
π(a∣s,θ)\pi(a|s,\theta)π(a∣s,θ)算法参数: 步长大小
α>0\alpha>0α>0初始化的策略参数
θ\thetaθ
循环(each trajectory):
根据
π(⋅∣⋅,θ)\pi(\cdot|\cdot,\theta)π(⋅∣⋅,θ)，生成S0,A0,R1,...ST−1,AT−1,RTS_0,A_0,R_1,...S_{T-1},A_{T-1},R_{T}S0,A0,R1,...ST−1,AT−1,RT对一个回合的每一步进行循环,
t=0,1,...,T−1t=0,1,...,T-1t=0,1,...,T−1
G=∑k=t+1Tγk−t−1RkG = \sum_{k=t+1}^{T} \gamma^{k-t-1} R_kG=∑k=t+1Tγk−t−1Rkθ=θ+αγtG▽ln[π(at∣st,θ)]\theta = \theta + \alpha \gamma^t G \bigtriangledown ln[\pi(a_t|s_t,\theta)]θ=θ+αγtG▽ln[π(at∣st,θ)]
3.3 具体代码
import torchimport gymimport numpy as npimport torch.nn as nnfrom torch.nn import Linearimport torch.nn.functional as Fimport torch.optim as optimfrom torch.distributions import Categoricalimport timelr =0.002gamma =0.8classPGPolicy(nn.Module):def__init__(self, input_size=4, hidden_size=128, output_size=2):super(PGPolicy, self).__init__()self.fc1 = Linear(input_size, hidden_size)self.fc2 = Linear(hidden_size, output_size)self.dropout = nn.Dropout(p=0.6)self.saved_log_probs =[]# 记录每一步的动作概率self.rewards =[]#记录每一步的rdefforward(self, x):x = self.fc1(x)x = self.dropout(x)x = F.relu(x)x = self.fc2(x)out = F.softmax(x, dim=1)return outdefchoose_action(state, policy):state = torch.from_numpy(state).float().unsqueeze(0)# 在索引0对应位置增加一个维度probs = policy(state)m = Categorical(probs)#创建以参数probs为标准的类别分布,之后的m.sampe就会按此概率选择动作action = m.sample()policy.saved_log_probs.append(m.log_prob(action))return action.item()#返回的就是intdeflearn(policy, optimizer):R =0policy_loss =[]returns =[]for r in policy.rewards[::-1]:R = r + gamma*Rreturns.insert(0,R)#从头部插入，即反着插入returns = torch.tensor(returns)# 归一化（均值方差），eps是一个非常小的数，避免除数为0eps = np.finfo(np.float64).eps.item()returns =(returns - returns.mean())/(returns.std()+ eps)for log_prob, R inzip(policy.saved_log_probs, returns):policy_loss.append(-log_prob*R)optimizer.zero_grad()policy_loss = torch.cat(policy_loss).sum()policy_loss.backward()optimizer.step()del policy.rewards[:]# 清空数据del policy.saved_log_probs[:]deftrain(episode_num):env = gym.make('CartPole-v1')env.seed(1)torch.manual_seed(1)policy = PGPolicy()# policy.load_state_dict(torch.load('save_model.pt')) # 模型导入optimizer = optim.Adam(policy.parameters(), lr)average_r =0for i inrange(1, episode_num+1):#采这么多轨迹obs = env.reset()ep_r =0for t inrange(1,10000):action = choose_action(obs, policy)obs, reward, done, _ = env.step(action)policy.rewards.append(reward)ep_r += rewardif done:breakaverage_r =0.05* ep_r +(1-0.05)* average_rlearn(policy, optimizer)if i %10==0:print('Episode {}\tLast reward: {:.2f}\tAverage reward: {:.2f}'.format(i, ep_r, average_r))torch.save(policy.state_dict(),'PGPolicy.pt')deftest():env = gym.make('CartPole-v1')env.seed(1)torch.manual_seed(1)policy = PGPolicy()policy.load_state_dict(torch.load('PGPolicy.pt'))# 模型导入average_r =0with torch.no_grad():obs = env.reset()ep_r =0for t inrange(1,10000):action = choose_action(obs, policy)obs, reward, done, _ = env.step(action)policy.rewards.append(reward)env.render()time.sleep(0.1)ep_r += rewardif done:breaktrain(1000)# test()
3.4 演示效果
训练过程:
四、PPO (飞船降落)
4.1 LunarLander-v2环境介绍
（该环境需要安装box2d）
https://www.gymlibrary.dev/environments/box2d/lunar_lander/?highlight=lunarlander
- **observation (1,8)**NumObservation0x1y2 V x V_x Vx3 V y V_y Vy4 a n g l e angle angle5 a n g u l a r v e l o c i t y angular \quad velocity angularvelocity6左腿是否触地(bool)7右腿是否触地(bool)
- **action (1,4)**NumAction0啥也不干1左侧点火2下面(主发动机)点火3右侧点火
- reward从屏幕顶部移动到着陆台的奖励约为100-140分。如果着陆器没降落到陆台，它将失去奖励。如果着陆器坠毁，它将获得额外的-100分。如果它成功降落，它将获得额外的+100分。接地的每个支腿为+10点。每架主机点火-0.3分。侧面发动机每帧点火-0.03分。解决的是200分。
- 终止条件- 飞船与月球接触- 飞船|x|>1
4.2 PPO-Clip算法流程
初始化策略函数的参数
θ0\theta_0θ0, 初始化价值函数的参数ϕ0\phi_0ϕ0for k = 0,1,2,…
基于
π(θk)\pi(\theta_k)π(θk)来采集轨迹组Dk=τkD_k={\tau_k}Dk=τk计算
RtR_tRt计算
AtA_tAt更新策略:
θk+1=argmaxθ1∣Dk∣T∑τ∑tmin(πθ(at∣st)πθ′(at∣st)A(st,at),g(ϵ,A(st,at)))\theta_{k+1}=\underset{\theta}{argmax}\frac{1}{|D_k|T}\underset{\tau }{\sum}\underset{t }{\sum} min(\frac{\pi_\theta(a_t|s_t)}{\pi_{\theta^{'}}(a_t|s_t)}A(s_t,a_t),\quad g(\epsilon,A(s_t,a_t)))θk+1=θargmax∣Dk∣T1τ∑t∑min(πθ′(at∣st)πθ(at∣st)A(st,at),g(ϵ,A(st,at)))更新价值函数:
ϕk+1=argminϕ1∣Dk∣T∑τ∑t(V(st)−R)2\phi_{k+1}=\underset{\phi}{argmin}\frac{1}{|D_k|T}\underset{\tau }{\sum}\underset{t }{\sum} (V(s_t)-R)^2ϕk+1=ϕargmin∣Dk∣T1τ∑t∑(V(st)−R)2
4.3 具体代码
import torchimport torch.nn as nnfrom torch.distributions import Categoricalimport gymdevice ='cpu'classMemory:def__init__(self):self.actions =[]self.states =[]self.logprobs =[]self.rewards =[]self.is_terminals =[]defclear_memory(self):del self.actions[:]del self.states[:]del self.logprobs[:]del self.rewards[:]del self.is_terminals[:]classActorCritic(nn.Module):def__init__(self, state_dim, action_dim, n_latent_var):super(ActorCritic, self).__init__()# actorself.action_layer = nn.Sequential(nn.Linear(state_dim, n_latent_var),nn.Tanh(),nn.Linear(n_latent_var, n_latent_var),nn.Tanh(),nn.Linear(n_latent_var, action_dim),nn.Softmax(dim=-1))# criticself.value_layer = nn.Sequential(nn.Linear(state_dim, n_latent_var),nn.Tanh(),nn.Linear(n_latent_var, n_latent_var),nn.Tanh(),nn.Linear(n_latent_var,1))defforward(self):# 如果这个方法没有被子类重写，但是调用了，就会报错raise NotImplementedErrordefact(self, state, memory):state = torch.from_numpy(state).float().to(device)action_probs = self.action_layer(state)dist = Categorical(action_probs)action = dist.sample()memory.states.append(state)memory.actions.append(action)memory.logprobs.append(dist.log_prob(action))return action.item()defevaluate(self, state, action):action_probs = self.action_layer(state)dist = Categorical(action_probs)action_logprobs = dist.log_prob(action)dist_entropy = dist.entropy()state_value = self.value_layer(state)return action_logprobs, torch.squeeze(state_value), dist_entropyclassPPO:def__init__(self, state_dim, action_dim, n_latent_var, lr, betas, gamma, K_epochs, eps_clip):self.lr = lrself.betas = betasself.gamma = gammaself.eps_clip = eps_clipself.K_epochs = K_epochsself.policy = ActorCritic(state_dim, action_dim, n_latent_var).to(device)self.optimizer = torch.optim.Adam(self.policy.parameters(), lr=lr, betas=betas)self.policy_old = ActorCritic(state_dim, action_dim, n_latent_var).to(device)self.policy_old.load_state_dict(self.policy.state_dict())self.MseLoss = nn.MSELoss()defupdate(self, memory):# Monte Carlo estimate of state rewards:rewards =[]discounted_reward =0for reward, is_terminal inzip(reversed(memory.rewards),reversed(memory.is_terminals)):if is_terminal:discounted_reward =0discounted_reward = reward +(self.gamma * discounted_reward)rewards.insert(0, discounted_reward)# Normalizing the rewards:rewards = torch.tensor(rewards).to(device).to(torch.float32)rewards =(rewards - rewards.mean())/(rewards.std()+1e-5)# convert list to tensorold_states = torch.stack(memory.states).to(device).detach().to(torch.float32)old_actions = torch.stack(memory.actions).to(device).detach().to(torch.float32)old_logprobs = torch.stack(memory.logprobs).to(device).detach().to(torch.float32)# Optimize policy for K epochs:for _ inrange(self.K_epochs):# Evaluating old actions and values :logprobs, state_values, dist_entropy = self.policy.evaluate(old_states, old_actions)# Finding the ratio (pi_theta / pi_theta__old):ratios = torch.exp(logprobs - old_logprobs.detach())# Finding Surrogate Loss:advantages = rewards - state_values.detach()surr1 = ratios * advantagessurr2 = torch.clamp(ratios,1-self.eps_clip,1+self.eps_clip)* advantagesloss =-torch.min(surr1, surr2)+0.5*self.MseLoss(state_values, rewards)-0.01*dist_entropyloss =loss.to(torch.float32)# take gradient stepself.optimizer.zero_grad()loss.mean().backward()self.optimizer.step()# Copy new weights into old policy:self.policy_old.load_state_dict(self.policy.state_dict())defmain():############## Hyperparameters ##############env_name ='LunarLander-v2'# "LunarLander-v2"# creating environmentenv = gym.make(env_name)env = env.unwrappedstate_dim = env.observation_space.shape[0]action_dim =4render =Falsesolved_reward =200# stop training if avg_reward > solved_rewardlog_interval =20# print avg reward in the intervalmax_episodes =5000# max training episodesmax_timesteps =1000# max timesteps in one episoden_latent_var =64# number of variables in hidden layerupdate_timestep =2000# update policy every n timestepslr =0.002betas =(0.9,0.999)gamma =0.99# discount factorK_epochs =4# update policy using 1 trajectory for K epochseps_clip =0.2# clip parameter for PPOrandom_seed =123#############################################if random_seed:torch.manual_seed(random_seed)env.seed(random_seed)memory = Memory()ppo = PPO(state_dim, action_dim, n_latent_var, lr, betas, gamma, K_epochs, eps_clip)print(lr,betas)# logging variablesrunning_reward =0avg_length =0timestep =0# training loopfor i_episode inrange(1, max_episodes+1):state = env.reset()for t inrange(max_timesteps):timestep +=1# Running policy_old:action = ppo.policy_old.act(state, memory)state, reward, done, _ = env.step(action)# Saving reward and is_terminal:memory.rewards.append(reward)memory.is_terminals.append(done)# update if its timeif timestep % update_timestep ==0:ppo.update(memory)memory.clear_memory()timestep =0running_reward += rewardif render:env.render()if done:breakavg_length += t# stop training if avg_reward > solved_rewardif running_reward >(log_interval*solved_reward):print("########## Solved! ##########")torch.save(ppo.policy.state_dict(),'./PPO_{}_{}.pth'.format(env_name,lr))break# loggingif i_episode % log_interval ==0:avg_length =int(avg_length/log_interval)running_reward =int((running_reward/log_interval))print('Episode {} \t avg length: {} \t reward: {}'.format(i_episode, avg_length, running_reward))running_reward =0avg_length =0if i_episode %2000==0:torch.save(ppo.policy.state_dict(),'./PPO_{}_{}.pth'.format(env_name,lr))deftest():############## Hyperparameters ##############env_name ="LunarLander-v2"# creating environmentenv = gym.make(env_name)state_dim = env.observation_space.shape[0]action_dim =4render =Falsemax_timesteps =500n_latent_var =64# number of variables in hidden layerlr =0.0002betas =(0.9,0.999)gamma =0.99# discount factorK_epochs =4# update policy for K epochseps_clip =0.2# clip parameter for PPO#############################################n_episodes =3max_timesteps =300render =Truesave_gif =Falsefilename ="PPO_{}_0.002.pth".format(env_name)directory ="./"memory = Memory()ppo = PPO(state_dim, action_dim, n_latent_var, lr, betas, gamma, K_epochs, eps_clip)ppo.policy_old.load_state_dict(torch.load(directory+filename))for ep inrange(1, n_episodes+1):ep_reward =0state = env.reset()for t inrange(max_timesteps):action = ppo.policy_old.act(state, memory)state, reward, done, _ = env.step(action)ep_reward += rewardif render:env.render()if done:breakprint('Episode: {}\tReward: {}'.format(ep,int(ep_reward)))ep_reward =0env.close()if __name__ =='__main__':main()# test()
4.4 演示效果
五、DQN (打砖块)
5.1 Breakout-v0环境介绍
Breakout - Gym Documentation (gymlibrary.dev)
- observation (210,160,3)
- action (1,4)NumAction0NOOP1FIRE2RIGHT3LEFT
- reward
5.2 DQN算法流程
(带有经验回放池的DQN)
初始化经验回放池
DDD(容量为NNN)随机初始化 动作-价值 函数
QQQfor (each episode)
初始化序列
s1=[x1]s_1=[x_1]s1=[x1],预处理ϕ1=ϕ(s1)\phi_1=\phi(s_1)ϕ1=ϕ(s1)for (each step)
at=maxaQ∗(ϕ(st),a:θ)a_t=\underset{a}{max}Q^*(\phi(s_t),a:\theta)at=amaxQ∗(ϕ(st),a:θ) (概率=1-ϵ\epsilonϵ)执行
ata_tat，得到rtr_trt和图片xt+1x_{t+1}xt+1st+1=st,ϕt+1=ϕ(st+1)s_{t+1}=s_t,\phi_{t+1}=\phi(s_{t+1})st+1=st,ϕt+1=ϕ(st+1)将
(ϕt,at,rt,ϕt+1)(\phi_t,a_t,r_t,\phi_{t+1})(ϕt,at,rt,ϕt+1)存储进DDD在
DDD中采样yi={rj(terminalϕj+1)rj+γmaxQ(ϕj+1,a′;θ)(non−terminalϕj+1)y_i = \left\{\begin{matrix} r_j & (terminal\quad \phi_{j+1})\\ r_j +\gamma max Q( \phi_{j+1},a^{'}; \theta) & (non-terminal\quad \phi_{j+1}) \end{matrix}\right.yi={rjrj+γmaxQ(ϕj+1,a′;θ)(terminalϕj+1)(non−terminalϕj+1)根据
(yi−Q(ϕj,aj:θ))2(y_i-Q(\phi_j,a_j:\theta))^2(yi−Q(ϕj,aj:θ))2进行梯度下降
5.3 具体代码
import gymimport cv2import torchimport numpy as npimport torch.nn as nnimport pandas as pdfrom torch.nn import Linear, Conv2d, ReLUimport PIL.Image as Imagedevice=torch.device("cuda:0"if torch.cuda.is_available()else"cpu")# 经验池classDQBReplayer:def__init__(self, capacity):# (S,A,R,S)self.memory = pd.DataFrame(index=range(capacity), columns=['observation','action','reward','next_observation','done'])self.i =0self.count =0self.capacity = capacitydefstore(self,*args):self.memory.loc[self.i]= argsself.i =(self.i +1)%self.capacityself.count =min(self.count+1, self.capacity)defsample(self, size):indics = np.random.choice(self.count, size=size)return(np.stack(self.memory.loc[indics,field])for field in self.memory.columns)# Q-NetworkclassDQN_net(nn.Module):def__init__(self):super(DQN_net, self).__init__()self.conv = nn.Sequential(Conv2d(in_channels=4, out_channels=32, kernel_size=8, stride=4),ReLU(),Conv2d(in_channels=32, out_channels=64, kernel_size=4, stride=2),ReLU(),Conv2d(in_channels=64, out_channels=64, kernel_size=3, stride=1),ReLU())self.classifier = nn.Sequential(Linear(3136,512),ReLU(),Linear(512,4))defforward(self, x):x = self.conv(x)x = x.view(x.size(0),-1)output = self.classifier(x)return outputclassDQN(nn.Module):def__init__(self, input_shape, env):super(DQN, self).__init__()self.replayer_start_size =100000self.upon_times =20self.replayer = DQBReplayer(capacity=self.replayer_start_size)self.action_n = env.action_space.nself.image_stack = input_shape[2]self.gamma =0.99self.image_shape =(input_shape[0], input_shape[1])self.e_net = DQN_net()self.t_net = DQN_net()self.learn_step =0self.max_learn_step =650000self.epsilon =1.self.start_learn =Falsedefget_next_state(self,state=None,observation=None):img=Image.fromarray(observation,"RGB")img=img.resize(self.image_shape).convert('L')img=np.asarray(img.getdata(),dtype=np.uint8,).reshape(img.size[1],img.size[0])if state isNone:next_state = np.array([img,]*self.image_stack)else:next_state = np.append(state[1:],[img,],axis=0)return next_statedefdecide(self,state,step):if self.start_learn ==False:#前50000步随机选择action = np.random.randint(0,4)return actionelse:self.epsilon -=0.0000053if step <30:#每局前三十步随机选择，中间30万，#以一定概率（1-epsilon）通过神经网络选择，# 最后30万次以0.99概率通过神经网络选择action = np.random.randint(0,4)elif np.random.random()<max(self.epsilon,0.0005):action = np.random.randint(0,4)else:state = state/128-1y = torch.Tensor(state).float().unsqueeze(0)y = y.to(device)x = self.e_net(y).detach()if self.learn_step%2000==0:print("q value{}".format(x))action = torch.argmax(x).data.item()return actiondefmain():sum_reward =0store_count =0env = gym.make('Breakout-v0')net = DQN([84,84,4], env).cuda()Load_Net =0if Load_Net==1:load_net_path ='./epsiode_2575_reward_10.0.pkl'print("Load old net and the path is:",load_net_path)net.e_net = torch.load(load_net_path)net.t_net = torch.load(load_net_path)max_score =0mse = nn.MSELoss()mse = mse.cuda()opt = torch.optim.RMSprop(net.e_net.parameters(), lr=0.0015)for i inrange(20000):lives =5obs = env.reset()state = net.get_next_state(None,obs)epoch_reward =0if i%100==0:print("{} times_game".format(i),end=':')print('epoch_reward:{}'.format(epoch_reward))for step inrange(500000):action = net.decide(state,step=step)obs, reward, done, _ = env.step(action)next_state = net.get_next_state(state, obs)epoch_reward += rewardnet.replayer.store(state, action, reward, next_state, done)net.learn_step +=1if net.learn_step >= net.replayer_start_size //2and net.learn_step %4==0:if net.start_learn ==False:net.start_learn =Trueprint('Start Learn!')sample_n =32states, actions, rewards, next_states, dones = net.replayer.sample(sample_n)states, next_states = states /128-1, next_states /128-1rewards = torch.Tensor(np.clip(rewards,-1,1)).unsqueeze(1).cuda()states, next_states = torch.Tensor(states).cuda(), torch.Tensor(next_states).cuda()actions = torch.Tensor(actions).long().unsqueeze(1).cuda()dones = torch.Tensor(dones).unsqueeze(1).cuda()q = net.e_net(states).gather(1, actions)q_next = net.t_net(next_states).detach().max(1)[0].reshape(sample_n,1)tq = rewards + net.gamma *(1-done)* q_nextloss = mse(q, tq)opt.zero_grad()loss.backward()opt.step()if net.learn_step %(net.upon_times *5)==0:net.t_net.load_state_dict(net.e_net.state_dict())if net.learn_step %100==0:loss_record = loss.item()a_r = torch.mean(rewards,0).item()state = next_stateif done:save_net_path ='./'sum_reward+=epoch_rewardif epoch_reward > max_score:name ="epsiode_"+str(net.learn_step)+"_reward_"+str(epoch_reward)+".pkl"torch.save(net.e_net, save_net_path+name)max_score = epoch_rewardelif i %1000==0:name ="No."+str(i)+".pkl"torch.save(net.e_net, save_net_path + name)if i%10==0:sum_reward=0breakimport cv2defPictureArray2Video(pic_list, path='./test.mp4'):h,w,_ = pic_list[0].shape[0], pic_list[0].shape[1], pic_list[0].shape[2]print(h,w)writer = cv2.VideoWriter(path, cv2.VideoWriter_fourcc('m','p','4','v'),10,(w, h),True)total_frame =len(pic_list)for i inrange(total_frame):writer.write(pic_list[i])writer.release()deftest():pics =[]sum_reward =0store_count =0env = gym.make('Breakout-v0')net = DQN([84,84,4], env).cuda()Load_Net =1if Load_Net==1:load_net_path ='./epsiode_10219_reward_9.0.pkl'print("Load old net and the path is:",load_net_path)net.e_net = torch.load(load_net_path)net.t_net = torch.load(load_net_path)max_score =0mse = nn.MSELoss()mse = mse.cuda()obs = env.reset()state = net.get_next_state(None,obs)epoch_reward =0for step inrange(500000):action = net.decide(state,step=step)obs, reward, done, _ = env.step(action)pic = env.render(mode='rgb_array')pic = cv2.cvtColor(pic,cv2.COLOR_BGR2RGB)next_state = net.get_next_state(state, obs)pics.append(pic)if done:PictureArray2Video(pics)break
5.4 演示效果
这个我感觉要训练好久，我训练了两个小时，reward=11，然后停下了。
六、DDPG (单摆)
6.1 Pendulum-v1环境介绍
https://www.gymlibrary.dev/environments/classic_control/pendulum/?highlight=pendulum+v1
- observation (1,3)NumObservationMinMax0cos(theta)-111sin(angle)-112角速度-8.08.0
- action （1，）力矩，大小在(-2,2)之前的值
- 奖励 r = − ( θ 2 + 0.1 × ω 2 + 0.001 × 力 矩 2 ) r = -(\theta^2 + 0.1×\omega^2 + 0.001×力矩^2) r=−(θ2+0.1×ω2+0.001×力矩2)
6.2 DDPG算法流程
随机初始化 评论员
Q(s,a∣θQ)Q(s,a|\theta^Q)Q(s,a∣θQ)和 演员μ(s∣θμ)\mu(s|\theta^\mu)μ(s∣θμ)初始化目标策略价值网络
Q′Q{'}Q′和θ′\theta^{'}θ′，θQ′=θQ,θμ′=θμ\theta^{Q^{'}}=\theta^Q,\theta^{\mu^{'}}=\theta^\muθQ′=θQ,θμ′=θμ初始化经验回放池R
for (each episode)
for (each step)
at=μ(st∣θμ)a_t=\mu(s_t|\theta^{\mu})at=μ(st∣θμ)st+1,rt,done,=env.step(at)s_{t+1},r_t,done,_ = env.step(a_t)st+1,rt,done,=env.step(at)将
(st,at,rt,st+1)(s_t,a_t,r_t,s_{t+1})(st,at,rt,st+1)存储进*R*从R中采样N条轨迹
(si,ai,ri,si+1)(s_i,a_i,r_i,s_{i+1})(si,ai,ri,si+1)yi=ri+γQ′(si+1,μ′(si+1∣θQ′)∣θQ′)y_i = r_i + \gamma Q^{'}(s_{i+1},\mu^{'}(s_{i+1}|\theta^{Q^{'}})|\theta^{Q^{'}})yi=ri+γQ′(si+1,μ′(si+1∣θQ′)∣θQ′)Loss=1NΣ(yi−Q(si,ai∣θQ))2Loss = \frac{1}{N}\Sigma(y_i-Q(s_i,a_i|\theta^{Q}))^2Loss=N1Σ(yi−Q(si,ai∣θQ))2, 更新评论员网络▽θμJ=1NΣ▽aQ(s,a∣θQ)∣s=si,a=μ(si)▽θμμ(s∣θμ)∣)si\bigtriangledown _{\theta^\mu}J = \frac{1}{N}\Sigma \bigtriangledown_a Q(s,a|\theta^Q)|_{s=s_i,a=\mu(s_i)}\bigtriangledown_{\theta^\mu} \mu(s|\theta^\mu)|)_{s_i}▽θμJ=N1Σ▽aQ(s,a∣θQ)∣s=si,a=μ(si)▽θμμ(s∣θμ)∣)si更新目标网络:
θQ′=τθQ+(1−τ)θQ′\theta^{Q^{'}} = \tau \theta^Q + (1-\tau)\theta^{Q^{'}}θQ′=τθQ+(1−τ)θQ′θμ′=τθμ+(1−τ)θμ′\theta^{\mu^{'}} = \tau \theta^\mu + (1-\tau)\theta^{\mu^{'}}θμ′=τθμ+(1−τ)θμ′
6.3 具体代码
import torchimport torch.nn as nnimport torch.nn.functional as Fimport numpy as npimport gymimport time##################### hyper parameters ####################EPISODES =200EP_STEPS =200LR_ACTOR =0.001LR_CRITIC =0.002GAMMA =0.9TAU =0.01MEMORY_CAPACITY =10000BATCH_SIZE =32RENDER =FalseENV_NAME ='Pendulum-v1'########################## DDPG Framework ######################classActorNet(nn.Module):# define the network structure for actor and criticdef__init__(self, s_dim, a_dim):super(ActorNet, self).__init__()self.fc1 = nn.Linear(s_dim,30)self.fc1.weight.data.normal_(0,0.1)# initialization of FC1self.out = nn.Linear(30, a_dim)self.out.weight.data.normal_(0,0.1)# initilizaiton of OUTdefforward(self, x):x = self.fc1(x)x = F.relu(x)x = self.out(x)x = torch.tanh(x)actions = x *2# for the game "Pendulum-v0", action range is [-2, 2]return actionsclassCriticNet(nn.Module):def__init__(self, s_dim, a_dim):super(CriticNet, self).__init__()self.fcs = nn.Linear(s_dim,30)self.fcs.weight.data.normal_(0,0.1)self.fca = nn.Linear(a_dim,30)self.fca.weight.data.normal_(0,0.1)self.out = nn.Linear(30,1)self.out.weight.data.normal_(0,0.1)defforward(self, s, a):x = self.fcs(s)y = self.fca(a)actions_value = self.out(F.relu(x+y))return actions_valueclassDDPG(object):def__init__(self, a_dim, s_dim, a_bound):self.a_dim, self.s_dim, self.a_bound = a_dim, s_dim, a_boundself.memory = np.zeros((MEMORY_CAPACITY, s_dim *2+ a_dim +1), dtype=np.float32)self.pointer =0# serves as updating the memory data # Create the 4 network objectsself.actor_eval = ActorNet(s_dim, a_dim)self.actor_target = ActorNet(s_dim, a_dim)self.critic_eval = CriticNet(s_dim, a_dim)self.critic_target = CriticNet(s_dim, a_dim)# create 2 optimizers for actor and criticself.actor_optimizer = torch.optim.Adam(self.actor_eval.parameters(), lr=LR_ACTOR)self.critic_optimizer = torch.optim.Adam(self.critic_eval.parameters(), lr=LR_CRITIC)# Define the loss function for critic network updateself.loss_func = nn.MSELoss()defstore_transition(self, s, a, r, s_):# how to store the episodic data to buffertransition = np.hstack((s, a,[r], s_))index = self.pointer % MEMORY_CAPACITY # replace the old data with new dataself.memory[index,:]= transitionself.pointer +=1defchoose_action(self, s):# print(s)s = torch.unsqueeze(torch.FloatTensor(s),0)return self.actor_eval(s)[0].detach()deflearn(self):# softly update the target networksfor x in self.actor_target.state_dict().keys():eval('self.actor_target.'+ x +'.data.mul_((1-TAU))')eval('self.actor_target.'+ x +'.data.add_(TAU*self.actor_eval.'+ x +'.data)')for x in self.critic_target.state_dict().keys():eval('self.critic_target.'+ x +'.data.mul_((1-TAU))')eval('self.critic_target.'+ x +'.data.add_(TAU*self.critic_eval.'+ x +'.data)')# sample from buffer a mini-batch dataindices = np.random.choice(MEMORY_CAPACITY, size=BATCH_SIZE)batch_trans = self.memory[indices,:]# extract data from mini-batch of transitions including s, a, r, s_batch_s = torch.FloatTensor(batch_trans[:,:self.s_dim])batch_a = torch.FloatTensor(batch_trans[:, self.s_dim:self.s_dim + self.a_dim])batch_r = torch.FloatTensor(batch_trans[:,-self.s_dim -1:-self.s_dim])batch_s_ = torch.FloatTensor(batch_trans[:,-self.s_dim:])# make action and evaluate its action valuesa = self.actor_eval(batch_s)q = self.critic_eval(batch_s, a)actor_loss =-torch.mean(q)# optimize the loss of actor networkself.actor_optimizer.zero_grad()actor_loss.backward()self.actor_optimizer.step()# compute the target Q value using the information of next statea_target = self.actor_target(batch_s_)q_tmp = self.critic_target(batch_s_, a_target)q_target = batch_r + GAMMA * q_tmp# compute the current q value and the lossq_eval = self.critic_eval(batch_s, batch_a)td_error = self.loss_func(q_target, q_eval)# optimize the loss of critic networkself.critic_optimizer.zero_grad()td_error.backward()self.critic_optimizer.step()############################### Training ####################################### Define the env in gymenv = gym.make(ENV_NAME)env = env.unwrappedenv.seed(1)s_dim = env.observation_space.shape[0]a_dim = env.action_space.shape[0]a_bound = env.action_space.higha_low_bound = env.action_space.lowddpg = DDPG(a_dim, s_dim, a_bound)var =3# the controller of exploration which will decay during training processt1 = time.time()for i inrange(EPISODES):s = env.reset()ep_r =0for j inrange(EP_STEPS):if RENDER: env.render()# add explorative noise to actiona = ddpg.choose_action(s)a = np.clip(np.random.normal(a, var), a_low_bound, a_bound)s_, r, done, info, _ = env.step(a)ddpg.store_transition(s, a, r /10, s_)# store the transition to memoryif ddpg.pointer > MEMORY_CAPACITY:var *=0.9995# decay the exploration controller factorddpg.learn()s = s_ep_r += rif j == EP_STEPS -1:print('Episode: ', i,' Reward: %i'%(ep_r),'Explore: %.2f'% var)if ep_r >-300: RENDER =Truebreakprint('Running time: ', time.time()- t1)if __name__ =="__main__":learn()env.close()
6.4 演示效果
本文转载自: https://blog.csdn.net/weixin_45696231/article/details/126723207
版权归原作者 Promethe_us 所有， 如有侵权，请联系我们删除。
版权归原作者 Promethe_us 所有， 如有侵权，请联系我们删除。