强化学习（三） |
您所在的位置：网站首页 › ddpg伪代码 › 强化学习（三） |
强化学习（三）
|
强化学习（三）--Reinforce算法
1. Reinforce算法2. Reinforce算法的代码实现2.1 Main函数的实现2.2 神经网络的搭建（Net类的实现）2.3 测试函数(test_episode函数的实现)2.4 RF类的实现2.4.1 choose_action 函数2.4.2 store_transition函数2.4.3 learn函数
3. Reinforce算法的效果展示
前两节的Q-learning和DQN算法都是强化学习中的Value-based的方法，它们都是先经过Q值来选择动作，而在强化学习中还有另外一大类算法：Policy-based。而在Policy-based算法中最著名的就是Policy Gradient，而Policy Gradient算法又可以根据更新方式分为两大类：
MC更新方法：Reinfoce算法；TD更新方法：Actor-Critic算法；
1. Reinforce算法
详细的算法介绍还是推荐科老师的课程（公开课地址），以及李宏毅老师的Policy Gradient算法课程。 Reinfoce算法是基于MC更新方式的Policy Gradient算法。MC更新方式是指每完成一个episode才进行算法的更新，它是基于策略梯度的一种算法，策略梯度算法是指先找到一个评价指标，然后利用随机梯度上升的方法来更新参数使评价指标不断的上升。随机梯度上升算法公式如下： 这里使用的环境是gym中的 ‘CartPole-v0’，Reinforce算法的整体框架如下图： Net类：神经网络的搭建；RF类：Reinfoce算法的主要更新步骤和动作的选取函数；test_episode函数：测试算法效果时的函数；main函数：主函数。
2.1 Main函数的实现
主函数主要实现 run episode的过程，首先定义RF的类，然后进行3000个episode的循环，每个episode最多进行300个step，要注意的是： 每个episode中的每个step与环境进行交互的数据都要进行记录 (s,a,r)；每个episode结束后才进行网络的训练；每100个episode进行一次测试，也就是运行test_episode函数，代码如下： def main(): # 初始化RF类 agent = RF(env) # 进行训练 for episode in range(EPISODE): obs = env.reset() for step in range(STEP): # 与环境的交互 action = agent.choose_action(obs) next_obs,reward,done,_ = env.step(action) # 存储一个episode中每个step的s,a,r agent.store_transition(obs,action,reward) # 进入下一个状态 obs = next_obs # 每个episode结束再进行训练(MC) if done: agent.learn() break # 每100个episode进行测试 if episode % 100 == 0: avg_reward = test_episode(env,agent) print('Episode: ',episode,'Test_reward: ',avg_reward) 2.2 神经网络的搭建（Net类的实现）这里进行Policy网络的搭建，神经网络就是常见的pytorch神经网络搭建方法，这里我们搭建了两层的神经网络： input_dim: 状态的维度；output_dim: 动作的个数；代码如下： class Net(nn.Module): def __init__(self,obs_dim,act_dim): super(Net,self).__init__() self.fc1 = nn.Linear(obs_dim,Hidden) self.fc2 = nn.Linear(Hidden,act_dim) def forward(self,x): x = F.relu(self.fc1(x)) out = self.fc2(x) return out 2.3 测试函数(test_episode函数的实现)test_episode函数过程和main函数中的过程有点类似，每次测试进行10个episode，求这5个episode的平均reward，动作直接由训练好的policy网络输出，代码如下： 测试代码输出10个episode的平均reward。 def test_episode(env, agent): total_reward = 0 for episode in range(TEST): obs = env.reset() for step in range(STEP): env.render() act = agent.choose_action(obs) next_obs, reward, done, _ = env.step(act) obs = next_obs total_reward += reward if done: break return total_reward / TEST # 计算测试的平均reward 2.4 RF类的实现RF类的代码框架如下图： choose_action函数：是用来根据训练出来的Policy网络来选择动作的函数；store_transition函数：是用来存储一个episode数据的函数learn函数：算法更新的函数；
2.4.1 choose_action 函数
Policy算法是根据训练出的policy网络而直接输出动作的，最后网络的输出需要经过一个softmax函数，将每个动作映射到0-1之间，最后根据每个动作数字表示的概率来选择动作，这样也保证算法的探索性。 2.4.2 store_transition函数这里的store_transition函数和DQN是不同的，这里只是存储每个episode的交互数据。 2.4.3 learn函数learn函数主要实现了三个功能： 未来总收益Gt的计算：根据store_transition函数存储的每个episode数据来计算未来总收益；根据policy网络输出的动作概率和每个episode中的实际动作计算交叉熵；计算loss并反向传递更新policy网络；还有一点要注意，每个episode训练结束后清空store_transition中存储数据的列表。 整个RF类的代码如下： class RF(object): def __init__(self,env): # 环境的状态和动作维度 self.obs_dim = env.observation_space.shape[0] self.act_dim = env.action_space.n # 存放每个episode的s,a,r self.ep_obs,self.ep_act,self.ep_r = [],[],[] # 初始化神经网络 self.net = Net(obs_dim = self.obs_dim,act_dim = self.act_dim).to(device) self.optimizer = torch.optim.Adam(self.net.parameters(), lr=LR) self.time_step = 0 # 选择动作的函数 def choose_action(self,obs): obs = torch.FloatTensor(obs).to(device) # 转换为torch的格式 action = self.net.forward(obs) with torch.no_grad(): # 不进行参数的更新 action = F.softmax(action,dim=0).cuda().data.cpu().numpy() action = np.random.choice(range(action.shape[0]),p=action) # 根据softmax输出的概率来选择动作 return action # 存储一个episode的状态、动作和回报的函数 def store_transition(self,obs,act,r): self.ep_obs.append(obs) self.ep_act.append(act) self.ep_r.append(r) # 更新策略网络的函数 def learn(self): self.time_step += 1 # 记录走过的step # 记录Gt的值 discounted_ep_rs = np.zeros_like(self.ep_r) running_add = 0 # 计算未来总收益 for t in reversed(range(0,len(self.ep_r))): # 反向计算 running_add = running_add * GAMMA + self.ep_r[t] discounted_ep_rs[t] = running_add discounted_ep_rs -= np.mean(discounted_ep_rs) # 减均值 discounted_ep_rs /= np.std(discounted_ep_rs) # 除以标准差 discounted_ep_rs = torch.FloatTensor(discounted_ep_rs).to(device) # 输出网络计算出的每个动作的概率值 act_prob = self.net.forward(torch.FloatTensor(self.ep_obs).to(device)) # 进行交叉熵的运算 neg_log_prob = F.cross_entropy(input = act_prob,target=torch.LongTensor(self.ep_act).to(device),reduction = 'none') # 计算loss loss = torch.mean(neg_log_prob * discounted_ep_rs) # 反向传播优化网络 self.optimizer.zero_grad() loss.backward() self.optimizer.step() # 每次学习后清空s,r,a的数组 self.ep_r,self.ep_act,self.ep_obs = [],[],[] 3. Reinforce算法的效果展示Reinfoce算法的效果还是可以的，在200个episode之后，获得的reward会逐渐加大。 |
对于式中 q 的不同改进形成了各种算法，而在Reinfoce算法中则是利用未来总收益Gt来代替q值，表示t步骤后回报的总和，因此Reinfoce的算法公式如下： 
其中，如果 Gt 大于零，参数更新的方向会增加它在当前状态的概率。意思就是如果回报是有利的，就增加这个动作出现的概率。并且回报越大，梯度更新的幅度越大，概率增加的也就越大。它的算法伪代码如下： 
【本文地址】