devbox
神奇cpp
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Python神经网络学习(七)--强化学习--使用神经网络_python 强化学习

作者：神奇cpp | 2024-08-04 04:14:50

赞

踩

python 强化学习

前言

前面说到了强化学习，但是仅仅是使用了一个表格，这也是属于强化学习的范畴了，毕竟强化学习属于在试错中学习的。

但是现在有一些问题，如果这个表格非常大呢？悬崖徒步仅仅是一个长12宽4，每个位置4个动作的表格而已，如果游戏是英雄联盟，那么多的位置，每个位置那么多的可能动作，画出一个表格简直是不可想象的。

但其实，如果把这个表格看作一个数学函数，他的输入是坐标，输出是一个动作（或者每个动作对应的价值）：

那也就是说，只要我们有一个坐标，得到一个动作，中间什么过程是可以不用管的，还记得这篇文章中说过：神经元（函数）+神经元（函数） = 神经网络（人工神经网络），那么，中间这一块也就可以使用神经网络代替，这也就是深度强化学习。

论文（Playing Atari with Deep Reinforcement Learning）地址：https://arxiv.org/abs/1312.5602

 设置环境

注意：今天的环境代码我修改过了，跟上一篇的不一样，所以大家还是要先读一下环境代码。

本次环境代码中添加了对于棋盘大小的设置，修复了一些bug。

  1. # -*- coding: utf-8 -*-
  2. """
  3. 作者：CSDN，chuckiezhu
  4. 作者地址：https://blog.csdn.net/qq_38431572
  5. 本文可用作学习使用，交流代码时需要附带本出处声明
  6. """
  7.  
  8. import random
  9. import numpy as np
  10.  
  11. from gym import spaces
  12.  
  13. """
  14. nrows
  15.      0  1  2  3  4  5  6  7  8  9  10  11  ncols
  16.    ---------------------------------------
  17. 0  |  |  |  |  |  |  |  |  |  |  |   |   |
  18.    ---------------------------------------
  19. 1  |  |  |  |  |  |  |  |  |  |  |   |   |
  20.    ---------------------------------------
  21. 2  |  |  |  |  |  |  |  |  |  |  |   |   |
  22.    ---------------------------------------
  23. 3   * |       cliff                  | ^ |
  25.   *: start point
  26.   cliff: cliff
  27.   ^: goal
  28. """
  29.  
  30. class CustomCliffWalking(object):
  31.     def __init__(self, stepReward: int=-1, cliffReward: int=-10, goalReward: int=10, col=12, row=4) -> None:
  32.         self.sr = stepReward
  33.         self.cr = cliffReward
  34.         self.gr = goalReward
  35.         self.col = col
  36.         self.row = row
  37.  
  38.         self.action_space = spaces.Discrete(4)  # 上下左右
  39.         self.reward_range = (cliffReward, goalReward)
  40.  
  41.         self.pos = np.array([row-1, 0], dtype=np.int8)  # agent 在3，0处出生，掉到悬崖内就会死亡，触发done和cliffReward
  42.  
  43.         self.die_pos = []
  44.         for c in range(1, self.col-1):
  45.             self.die_pos.append([self.row-1, c])
  46.         print("die pos: ", self.die_pos)
  47.         print("goal pos: ", [[self.row-1, self.col-1]])
  48.  
  49.         self.reset()
  50.     
  51.     def reset(self, random_reset=False):
  52.         """
  53.         初始化agent的位置
  54.         random: 是否随机出生, 如果设置random为True, 则出生点会随机产生
  55.         """
  56.         x, y = self.row-1, 0
  57.         if random_reset:
  58.             y = random.randint(0, self.col-1)
  59.             if y == 0:
  60.                 x = random.randint(0, self.row-1)
  61.             else:  # 除了正常坐标之外，还有一个不正常坐标：(3, 0)
  62.                 x = random.randint(0, self.row-2)
  63.             # 严格来讲，cliff和goal不算在坐标体系内
  64.         # agent 在3，0处出生，掉到悬崖内就会死亡，触发done和cliffReward
  65.         self.pos = np.array([x, y], dtype=np.int8)
  66.         # print("reset at:", self.pos)
  67.     
  68.     def step(self, action: int) -> list[list, int, bool, bool, dict]:
  69.         """
  70.         执行一个动作
  71.         action:
  72.             0: 上
  73.             1: 下
  74.             2: 左
  75.             3: 右
  76.         """
  77.  
  78.         move = [
  79.             np.array([-1, 0], dtype=np.int8), # 向上，就是x-1， y不动,
  80.             np.array([ 1, 0], dtype=np.int8), # 向下，就是x+1， y不动,
  81.             np.array([0, -1], dtype=np.int8), # 向左，就是y-1， x不动,
  82.             np.array([0,  1], dtype=np.int8), # 向右，就是y+1， x不动,
  83.         ]
  84.         new_pos = self.pos + move[action]
  85.         # 上左不能小于0
  86.         new_pos[new_pos < 0] = 0  # 超界的处理，比如0, 0 处向上或者向右走，处理完还是0,0
  87.         # 上右不能超界
  88.         if new_pos[0] > self.row-1:
  89.             new_pos[0] = self.row-1  # 超界处理
  90.         if new_pos[1] > self.col-1:
  91.             new_pos[1] = self.col-1
  92.  
  93.         reward = self.sr  # 每走一步的奖励
  94.         die = False
  95.         win = False
  96.         info = {
  97.             "reachGoal": False,
  98.             "fallCliff": False,
  99.         }
  100.         
  101.         if self.__is_pos_die(new_pos.tolist()):
  102.             die = True
  103.             info["fallCliff"] = True
  104.             reward = self.cr
  105.         elif self.__is_pos_win(new_pos.tolist()):
  106.             win = True
  107.             info["reachGoal"] = True
  108.             reward = self.gr
  109.  
  110.         self.pos = new_pos  # 更新坐标
  111.         return new_pos, reward, die, win, info
  112.     
  113.     def __is_pos_die(self, pos: list[int, int]) -> bool:
  114.         """判断自己的这个状态是不是已经结束了"""
  115.         return pos in self.die_pos
  116.  
  117.     def __is_pos_win(self, pos: list[int, int]) -> bool:
  118.         """判断自己的这个状态是不是已经结束了"""
  119.         return pos in [
  120.             [self.row-1, self.col-1],
  121.         ]
  122.

至于讲解这个环境，我觉得这个注释还是比较清楚的，如果有不明白的，请评论留言告知我。

制作网络

首先，我们先把自己代入表格，如果我们站到某个坐标，那么我们应该知道四个方向上的奖励，所以，网络可以有两种方式；

方式一、

网络输入是坐标和方向，输出是对应的奖励。

方式二、

网络输入是坐标，输出是四个方向对应的奖励。

这里我要来一句场外推理：方式一真的很麻烦，并且选择动作的时候，有多少个动作需要经过多少次网络。所以方式二是比较好的选择。

  1.  
  2. class Qac(nn.Module):
  3.     def __init__(self, in_shape, out_shape) -> None:
  4.         super(Qac, self).__init__()
  5.         self.in_shape = in_shape  # 就是 智能体 现在的坐标
  6.         self.action_space = out_shape  # 上0下1左2右3
  7.         self.dense1 = nn.Linear(self.in_shape, self.action_space)
  8.         # 输出就是每个动作的价值
  9.  
  10.         self.lrelu = nn.LeakyReLU()  # 换用tanh
  11.         self.softmax = nn.Softmax(-1)
  12.     
  13.     def forward(self, x) -> torch.Tensor:
  14.         x = self.dense1(x)
  15.         return x
  16.  
  17.     def sample_action(self, action_value: torch.Tensor, epsilon: float):
  18.         """从产生的动作概率中采样一个动作，利用epsilon贪心"""
  19.         if random.random() < epsilon:
  20.             # 随机选择
  21.             action = random.randint(0, self.action_space-1)
  22.             action = torch.tensor(action)
  23.         else:
  24.             action = torch.argmax(action_value)
  25.         
  26.         return action
  27.     
  28.     def load_model(self, modelpath):
  29.         """加载模型"""
  30.         tmp = torch.load(modelpath)
  31.         self.load_state_dict(tmp["model"])
  32.     
  33.     def save_model(self, modelpath):
  34.         """保存模型"""
  35.         tmp = {
  36.             "model": self.state_dict(),
  37.         }
  38.         torch.save(tmp, modelpath)

细心的人可能发现了，这个网络只有一层，非常简单，好像没有所谓的“特征提取”就直接到输出层了。这里有一个小技巧，就是我手动把坐标转成了onehot向量，可以认为是手动提取了特征。

  1. def num_to_onehot(pos: torch.Tensor) -> torch.Tensor:
  2.     """把坐标转成one_hot向量"""
  3.     n = int((pos[0] * 12 + pos[1]).item())
  4.     return nn.functional.one_hot(torch.tensor(n), num_classes=48)

如果大家使用两层神经网络，直接输入坐标，中间层是48，然后是一个输出层，也可以， 但是我试了，训练很慢，效果不好。不如这样直接手动编码了。

训练

整个训练的代码我直接贴在这里了：

  1. # -*- coding: utf-8 -*-
  2. """
  3. 利用DQN实现
  4. """
  5. """
  6. 作者：CSDN，chuckiezhu
  7. 作者地址：https://blog.csdn.net/qq_38431572
  8. 本文可用作学习使用，交流代码时需要附带本出处声明
  9. """
  10. import os
  11. import random
  12. import torch
  13. import numpy as np
  14. from torch import nn
  15.  
  16. from matplotlib import pyplot as plt
  17.  
  18. from cliff_walking_env import CustomCliffWalking
  19.  
  20.  
  21. nepisodes = 10000  # total 1w episodes
  22. epsilon = 1.0  # epsilon greedy policy
  23. epsilon_min = 0.05
  24. epsilon_decay = 0.9975
  25.  
  26. gamma = 0.9  # discount factor
  27. lr = 0.001
  28. random_reset = False
  29.  
  30. seed = 42
  31.  
  32. normalization = torch.tensor([3, 11], dtype=torch.float)
  33.  
  34. sr = -1
  35. cr = -10
  36. gr = 10
  37.  
  38. class Qac(nn.Module):
  39.     def __init__(self, in_shape, out_shape) -> None:
  40.         super(Qac, self).__init__()
  41.         self.in_shape = in_shape  # 就是智能体现在的坐标
  42.         self.action_space = out_shape  # 上0下1左2右3
  43.         self.dense1 = nn.Linear(self.in_shape, self.action_space)
  44.  
  45.         # 输出就是每个动作的价值
  46.  
  47.         self.lrelu = nn.LeakyReLU()  # 换用tanh
  48.         self.softmax = nn.Softmax(-1)
  49.     
  50.     def forward(self, x) -> torch.Tensor:
  51.         x = self.dense1(x)
  52.         return x
  53.  
  54.     def sample_action(self, action_value: torch.Tensor, epsilon: float):
  55.         """从产生的动作概率中采样一个动作，利用epsilon贪心"""
  56.         if random.random() < epsilon:
  57.             # 随机选择
  58.             action = random.randint(0, self.action_space-1)
  59.             action = torch.tensor(action)
  60.         else:
  61.             action = torch.argmax(action_value)
  62.         
  63.         return action
  64.     
  65.     def load_model(self, modelpath):
  66.         """加载模型"""
  67.         tmp = torch.load(modelpath)
  68.         self.load_state_dict(tmp["model"])
  69.     
  70.     def save_model(self, modelpath):
  71.         """保存模型"""
  72.         tmp = {
  73.             "model": self.state_dict(),
  74.         }
  75.         torch.save(tmp, modelpath)
  76.  
  77.  
  78. def num_to_onehot(pos: torch.Tensor) -> torch.Tensor:
  79.     """把坐标转成one_hot向量"""
  80.     n = int((pos[0] * 12 + pos[1]).item())
  81.     return nn.functional.one_hot(torch.tensor(n), num_classes=48)
  82.  
  83.     
  84. def main():
  85.     global epsilon
  86.     random.seed(seed)
  87.     torch.manual_seed(seed=seed)
  88.     plt.ion()
  89.  
  90.     os.makedirs("./out/ff_DQN/")
  91.     # cw = gym.make("CliffWalking-v0", render_mode="human")
  92.     cw = CustomCliffWalking(stepReward=sr, goalReward=gr, cliffReward=cr)
  93.  
  94.     # 专程onehot了
  95.     Q = Qac(in_shape=48, out_shape=cw.action_space.n)
  96.  
  97.     optimizer = torch.optim.Adam(Q.parameters(), lr=lr)
  98.     loss_fn = torch.nn.MSELoss()
  99.  
  100.     win_1000 = []  # 记录最近一千场赢的几率
  101.     total_win = 0
  102.     for i in range(1, nepisodes+1):
  103.         cw.reset(random_reset=random_reset)  # 重置环境
  104.         steps = 0
  105.         while True:
  106.             steps += 1
  107.             state_now = torch.tensor(cw.pos, dtype=torch.float)
  108.             state_now = num_to_onehot(state_now).unsqueeze_(0).to(torch.float)
  109.             action_values = Q(state_now)
  110.             action_values = action_values.squeeze()
  111.             action_now = Q.sample_action(action_value=action_values, epsilon=epsilon)
  112.  
  113.             action_now_value = action_values[action_now]  # 这个是采取这个动作的预测奖励
  114.  
  115.             state_next, reward_now, terminated, truncated, info = cw.step(action=action_now.item())   # 执行动作
  116.             state_next = num_to_onehot(state_next).unsqueeze_(0).to(torch.float)
  117.             with torch.no_grad():
  118.                 next_values = Q(state_next)
  119.                 next_values = next_values.squeeze()
  120.                 # 得到下一个的动作，（同一个策略下，因为这是onpolicy的sarsa
  121.                 action_next = Q.sample_action(action_value=action_values, epsilon=epsilon)
  122.                 action_next_value = next_values[action_next]  # 计算下一个动作的预期价值
  123.  
  124.             
  125.             # 计算  instantR + gamma * value_next，这个是实际上这个动作带来的预期收益
  126.             discounted_reward = reward_now + gamma * action_next_value * (1 - terminated) * (1 - truncated)
  127.  
  128.             # 计算误差
  129.             loss = loss_fn(action_now_value, discounted_reward)
  130.  
  131.             optimizer.zero_grad()
  132.             loss.backward()
  133.             optimizer.step()
  134.  
  135.             if terminated or truncated:
  136.                 if terminated:
  137.                     win_1000.append(0)
  138.                 if truncated:
  139.                     win_1000.append(1)
  140.                     total_win += 1
  141.                 break
  142.  
  143.             epsilon = epsilon * epsilon_decay
  144.             epsilon = max(epsilon, epsilon_min)  # 衰减学习旅
  145.         win_1000 = win_1000[-1000:]
  146.         win_rate = sum(win_1000)/1000.0
  147.         print("{}/{}, 当前探索率: {}, 是否成功: {}, 千场胜率:{}.".format(i, nepisodes, epsilon, truncated, win_rate), flush=True)
  148.         if i % 10000 == 0:
  149.             Q.save_model("./out/ff_DQN/Qac_{}_{}_{}_{}.pth".format(i, gr, cr, win_rate))
  150.     print("total win: ", total_win)
  151.  
  152.     # 收尾测试看看能不能通关
  153.     path = np.zeros((4, 12), dtype=np.float64)
  154.     cw.reset(random_reset=False)
  155.  
  156.     steps = 0
  157.     while steps <= 48:  # 走，48步走不到头就不会走到了
  158.         steps += 1
  159.         state_now = torch.tensor(cw.pos, dtype=torch.float)
  160.         state_now = num_to_onehot(state_now).unsqueeze_(0).to(torch.float)
  161.         action_values = Q(state_now).squeeze()
  162.         # 贪心算法选择动作
  163.         action_now = Q.sample_action(action_values, 0)
  164.         print(cw.pos[0], cw.pos[1], action_now)
  165.         new_pos, _, die, win, _ = cw.step(action=action_now)
  166.         if win:
  167.             print("[+] you win!")
  168.             break
  169.         if die:
  170.             print("[+] you lose!")
  171.             break
  172.         x = new_pos[0]
  173.         y = new_pos[1]
  174.         if x >= 0 and x <= 3 and y >= 0 and y <= 11:
  175.             path[x, y] = 1.0
  176.     plt.imshow(path)
  177.     plt.colorbar()
  178.     plt.savefig("./out/ff_DQN/path_sarsa_"+str(sr)+"_"+str(gr)+"_"+str(cr)+".png")
  179.  
  180. if __name__ == "__main__":
  181.     main()

上面的代码我测试没问题，如果不修改直接使用是完全可以的，目录结构是这样的：

那两个文件夹都是自动生成的，不需要手动建立。 

网络结构分析

这是上面代码的网络结构和更新流程。注意：实线代表有梯度，虚线代表无梯度。

每次由环境产生一个状态，先转成一个one_hot向量，作为网络的输入，得到四个动作分别价值多少。然后采样到的动作得到当前的Q(s, a)值，也就是action_value。

另一方面，采样得到的动作送入环境，环境给出下一个状态和立即奖励。下一个状态送入网络（没有梯度的计算），同样得到四个动作的价值。由于代码使用的是SARSA算法，所以需要按照同样的策略采样一个动作，同时得到动作的价值。也就是next_action_value。

这个时候，就可以根据环境的立即奖励reward_now和下一个状态的动作的价值next_action_value得到一个ground truth，而action_value作为网络的预测值，这两个可以用于计算损失。

损失的反向传播就是沿着实现传递到顶。实现网络的更新。

 

声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号