强化学习DQN玩转FlappyBird｜前景提要

前景提要

强化学习是机器学习大家族中的一大类, 使用强化学习能够让机器学着如何在环境中拿到高分, 表现出优秀的成绩. 而这些成绩背后却是他所付出的辛苦劳动, 不断的试错, 不断地尝试, 累积经验, 学习经验。

而DQN更是强化学习家族中最会玩游戏的一位，甚至可以在一些游戏中超越人类。

经典游戏--FlappyBird

介绍了玩游戏的DQN，我们再来看一看我们要玩的游戏，FlappyBird。

FlappyBird,这是一款简单又困难的手机游戏，游戏中玩家控制着一只胖乎乎的小鸟，跨越由各种不同长度水管所组成的障碍。小鸟会在引力的作用下向下坠落，玩家点击一次屏幕，小鸟就会向上弹跳一次。每通过一个障碍，便可以得到1分，如果小鸟掉下来或者撞到水管，便游戏结束。规则非常简单，但是想拿到高分可不简单，因为小鸟飞行的轨迹是一个个的抛物线，稍有不慎便会触碰到水管，那就要从头再来啦（像小编这样的手残玩家，玩了一下午最高分只能30多分，而训练了六个小时的DQN稳稳当当的可以得到100多分）。

然而DQN说我是大佬，我来玩！

下面就来介绍我们如何使用DQN玩转FlappyBird!

在代码中我们采用了CV2库去读取80*80的图像来获取FlappyBird的状态，不了解CV2的小伙伴不用着急，我们在推文最后有对CV2库的简单介绍，大家可以了解一下。

而神经网络的训练的优化器我们选择了Adam方法，以加快训练速度。

那什么是Adam优化器，为什么它可以加快训练速度：

如图，传统的参数W的更新是把原始的W累加上一个负的学习率(learning rate) 乘以校正值 (dx)。这种方法可能会让学习过程曲折无比, 看起来像喝醉的人回家时, 摇摇晃晃走了很多弯路。

而如果这个人从平地上放到了一个斜坡上, 只要他往下坡的方向走一点点, 由于向下的惯性, 他不自觉地就一直往下走, 走的弯路也变少了。这就是Momentum参数更新方法。如图是其表达式。

如果我们不是给喝醉酒的人安排另一个下坡, 而是给他一双不好走路的鞋子, 使得他一摇晃着走路就脚疼, 鞋子成为了走弯路的阻力, 逼着他往前直着走。这就是AdaGrad 参数更新方法。如图是其表达式。

说到这，肯定有小伙伴已经猜到了！如果我们同时给他一个下坡, 一双破鞋子，他会走得更好，更新的速度将会更快，这就是Adam更新方法。计算m 时有 momentum 下坡的属性, 计算 v 时有 adagrad 阻力的属性,又快又好的达到目标, 迅速收敛。

铺垫完了，大家是不是等着急了

下面我们就来看一看DQN是如何玩转FlappyBird的！！

介绍代码之前先上我们的训练结果。

我们将训练的每回合得分制成柱状图，可以看出，在前一千回合内，DQN几乎玩FlappyBird得分为零，而在训练了1200多回合后，开始渐渐得分了，但不高在几十左右；而随着训练的继续，在2700多回合后，DQN得分基本稳定在一百多，而继续再训练几百回合，就能够达到四五百分，可谓进步神速。

附上训练视频：

最开始没有经过任何学习，FlappyBird总是会撞到水管上而失败。

当训练到240000多步时，FlappyBird已经可以开始飞起来了。

网络构造

首先我们来看一看DQN强大的大脑构造吧

哦不，是神经网络的构造！

注：小编代码中tensorflow的用法是1点几的版本，如果大家安装的是最新的tensorflow，运行源码需要加上下面的代码来降版本！

代码语言：javascript复制

import tensorflow.compat.v1 as tf

代码语言：javascript复制

tf.disable_v2_behavior()

首先是对各个参数以及神经网络的初始化:

代码语言：javascript复制

import tensorflow.compat.v1 as tf
tf.disable_v2_behavior()
import numpy as np 
import random
from collections import deque 
FRAME_PER_ACTION = 1 # 每一个动作的帧数
GAMMA = 0.99 # 对观测值的衰减程度
OBSERVE = 100. # 训练前观测的步数
EXPLORE = 200000. # 探索阶段的步数
FINAL_EPSILON = 0 # epsilon的最终值，刚开始训练设置为0.001
INITIAL_EPSILON = 0 # epsilon的初始值，刚开始训练设置为0.01
REPLAY_MEMORY = 50000 # 经验池回顾的步数
BATCH_SIZE = 32 # minibatch的大小
UPDATE_TIME = 100 # 每过UPDATE_TIME，更新Target Q Network
try:
    tf.mul
except:
    # 如果是新版本，用tf.multiply替换replace tf.mul
    tf.mul = tf.multiply
class BrainDQN:
   def __init__(self,actions):
      # 初始化经验池
      self.replayMemory = deque()
      # 初始化参数
      self.timeStep = 0
      self.epsilon = INITIAL_EPSILON
      self.actions = actions
      # 初始化Q network   self.stateInput,self.QValue,self.W_conv1,self.b_conv1,self.W_conv2,self.b_conv2,self.W_conv3,self.b_conv3,self.W_fc1,self.b_fc1,self.W_fc2,self.b_fc2 = self.createQNetwork()
      # 初始化Target Q Network    self.stateInputT,self.QValueT,self.W_conv1T,self.b_conv1T,self.W_conv2T,self.b_conv2T,self.W_conv3T,self.b_conv3T,self.W_fc1T,self.b_fc1T,self.W_fc2T,self.b_fc2T = self.createQNetwork()

复制Q（估计）网络参数给Q（现实）网络：

代码语言：javascript复制

# 复制Q network，更新Target Q Network
self.copy = [self.W_conv1T.assign(self.W_conv1),self.b_conv1T.assign(self.b_conv1),self.W_conv2T.assign(self.W_conv2),self.b_conv2T.assign(self.b_conv2),self.W_conv3T.assign(self.W_conv3),self.b_conv3T.assign(self.b_conv3),self.W_fc1T.assign(self.W_fc1),self.b_fc1T.assign(self.b_fc1),self.W_fc2T.assign(self.W_fc2),self.b_fc2T.assign(self.b_fc2)]

使用Adam优化器来最小化cost,达到训练目的，并对神经网络进行存取。

代码语言：javascript复制

# 创建训练方法
self.actionInput = tf.placeholder("float",[None,self.actions])
self.yInput = tf.placeholder("float", [None]) 
Q_Action = tf.reduce_sum(tf.mul(self.QValue, self.actionInput), reduction_indices = 1)
self.cost = tf.reduce_mean(tf.square(self.yInput - Q_Action))
# 这里使用了一个叫亚当的自适应优化算法，学习率为1e-6
self.trainStep = tf.train.AdamOptimizer(1e-6).minimize(self.cost)
# 存取神经网络
self.saver = tf.train.Saver()
self.session = tf.InteractiveSession()
self.session.run(tf.initialize_all_variables())
checkpoint = tf.train.get_checkpoint_state("networks")
if checkpoint and checkpoint.model_checkpoint_path:
      self.saver.restore(self.session, checkpoint.model_checkpoint_path)
      print ("Successfully loaded:", checkpoint.model_checkpoint_path)
else:
print ("Could not find old network weights")

神经网络的架构，我们设定了三个卷积层，两个全连接层来输入的图像进行处理，并最终输出Q值列表。

代码语言：javascript复制

# 四通道输入图像
def setInitState(self,observation):
self.currentState = np.stack((observation, observation, observation, observation), axis = 2)  
# 定义权重、偏置、卷积和池化函数
def weight_variable(self,shape):
   initial = tf.truncated_normal(shape, stddev = 0.01)
   return tf.Variable(initial)
def bias_variable(self,shape):
   initial = tf.constant(0.01, shape = shape)
   return tf.Variable(initial)
def conv2d(self,x, W, stride):
   return tf.nn.conv2d(x, W, strides = [1, stride, stride, 1], padding = "SAME")
def max_pool_2x2(self,x):
   return tf.nn.max_pool(x, ksize = [1, 2, 2, 1], strides = [1, 2, 2, 1], padding = "SAME")
   
def createQNetwork(self):
   # 第一层卷积层
   W_conv1 = self.weight_variable([8,8,4,32])
   b_conv1 = self.bias_variable([32])
   # 第二层卷积层
   W_conv2 = self.weight_variable([4,4,32,64])
   b_conv2 = self.bias_variable([64])
   # 第三层卷积层
   W_conv3 = self.weight_variable([3,3,64,64])
   b_conv3 = self.bias_variable([64])
   # 第一层全连接层
   W_fc1 = self.weight_variable([1600,512])
   b_fc1 = self.bias_variable([512])
   # 第二层全连接层
   W_fc2 = self.weight_variable([512,self.actions])
   b_fc2 = self.bias_variable([self.actions])
   #第一层卷积层，有32个卷积核（过滤器），每个卷积核的尺寸是8x8，x轴和y轴的步幅都是4，补零，并使用了一个relu激活函数。
   #第二层卷积层，有64个卷积核（过滤器），每个卷积核的尺寸是4x4，x轴和y轴的步幅都是2，补零，并使用了一个relu激活函数。
   #第三层卷积层，有64个卷积核（过滤器），每个卷积核的尺寸是3x3，x轴和y轴的步幅都是1，补零，并使用了一个relu激活函数。

   # 输入层
   stateInput = tf.placeholder("float",[None,80,80,4])
   # 隐藏层
   # 第一层隐藏层（用了一个池化层）
   h_conv1 = tf.nn.relu(self.conv2d(stateInput,W_conv1,4)   b_conv1)
   h_pool1 = self.max_pool_2x2(h_conv1)
   # 第二层隐藏层
   h_conv2 = tf.nn.relu(self.conv2d(h_pool1,W_conv2,2)   b_conv2)
   # 第三层隐藏层
   h_conv3 = tf.nn.relu(self.conv2d(h_conv2,W_conv3,1)   b_conv3)
   # 展开
   h_flat = tf.reshape(h_conv3,[-1,1600])
   # 全连接层
   h_fc1 = tf.nn.relu(tf.matmul(h_flat,W_fc1)   b_fc1)#全连接层
   # 输出层，输出动作对应的Q值列表
   QValue = tf.matmul(h_fc1,W_fc2)   b_fc2
   return stateInput,QValue,W_conv1,b_conv1,W_conv2,b_conv2,W_conv3,b_conv3,W_fc1,b_fc1,W_fc2,b_fc2

这里我们构造了个经验池，用来保存历史数据，并可从经验池中抽取数据来进行对神经网络的训练，这也是DQN为什么玩游戏玩得好的重要原因。经验池保存的是一个马尔科夫序列，功能主要是解决相关性及非静态分布问题。具体做法是把每个时间步agent与环境交互得到的转移样本储存到回放记忆单元，要训练时就随机拿出一些（minibatch）来训练其实就是将游戏的过程打成碎片存储，训练时随机抽取就避免了相关性问题。

代码语言：javascript复制

# 更新状态、经验池
def setPerception(self,nextObservation,action,reward,terminal):
   newState = np.append(self.currentState[:,:,1:],nextObservation,axis = 2)
   self.replayMemory.append((self.currentState,action,reward,newState,terminal))
   if len(self.replayMemory) > REPLAY_MEMORY:
      self.replayMemory.popleft()
   if self.timeStep > OBSERVE:
      self.trainQNetwork()
   state = ""
   if self.timeStep <= OBSERVE:
      state = "observe"
   elif self.timeStep > OBSERVE and self.timeStep <= OBSERVE   EXPLORE:
      state = "explore"
   else:
      state = "train"

   print ("TIMESTEP = ", self.timeStep, "  STATE = ", state, 
           "  EPSILON = ", self.epsilon)
   self.currentState = newState
   self.timeStep  = 1  
def trainQNetwork(self):
   # 从经验池中抽取小批量样本
   # 下面分别对应currentState,action,reward,newState,terminal
   minibatch = random.sample(self.replayMemory,BATCH_SIZE)
   state_batch = [data[0] for data in minibatch] # 当前状态
   action_batch = [data[1] for data in minibatch] # 输入动作
   reward_batch = [data[2] for data in minibatch] # 返回奖励
   nextState_batch = [data[3] for data in minibatch] # 返回的下一状态
   # 得到预测的以输入动作为索引的Q值
   y_batch = []
   # 获得下一状态的Q值
   QValue_batch = self.QValueT.eval(feed_dict={self.stateInputT:nextState_batch})
   for i in range(0,BATCH_SIZE):
      terminal = minibatch[i][4]
      #布尔值terminal表示游戏是否结束
      if terminal:
         y_batch.append(reward_batch[i])
      else:
         y_batch.append(reward_batch[i]   GAMMA * np.max(QValue_batch[i]))
   # 进行训练
   self.trainStep.run(feed_dict={
      self.yInput : y_batch,
      self.actionInput : action_batch,
      self.stateInput : state_batch
      })
   # 每10000步保存一次
   if self.timeStep % 10000 == 0:
      self.saver.save(self.session, 'networks/'   'network'   '-dqn', global_step = self.timeStep)
   if self.timeStep % UPDATE_TIME == 0:
      self.session.run(self.copy)

最后是动作Action的获取。

代码语言：javascript复制

def getAction(self):
 # 这里的训练数据为之前提到的四通道图像的模型输出
   QValue = self.QValue.eval(feed_dict= {self.stateInput:[self.currentState]})[0]
   action = np.zeros(self.actions)
   action_index = 0
   # 根据ε概率选择一个Action
   if self.timeStep % FRAME_PER_ACTION == 0:
      if random.random() <= self.epsilon:
         action_index = random.randrange(self.actions)
         action[action_index] = 1
      else:
         action_index = np.argmax(QValue)
         action[action_index] = 1
   else:
      action[0] = 1 # 不做动作
   # episilon随着步数而减小
   if self.epsilon > FINAL_EPSILON and self.timeStep > OBSERVE:
      self.epsilon -= (INITIAL_EPSILON - FINAL_EPSILON)/EXPLORE
 return action

讲完了神经网络的构造，我们再来看看他是如何通过神经网络来学着玩FlappyBird的吧。

这里我们定义了Bird的初始环境与动作，并输入神经网络开始学习，大概学了三个多小时DQN终于成为了玩FlappyBird的高手。

代码语言：javascript复制

import cv2
import sys
sys.path.append("game/")
import wrapped_flappy_bird as game
from BrainDQN_Nature import BrainDQN
import numpy as np
# 把图像处理成80*80，进行灰度化和二值化
def preprocess(observation):
   observation = cv2.cvtColor(cv2.resize(observation, (80, 80)), cv2.COLOR_BGR2GRAY)
   ret, observation = cv2.threshold(observation,1,255,cv2.THRESH_BINARY)
   return np.reshape(observation,(80,80,1))
def playFlappyBird():
   actions = 2
   brain = BrainDQN(actions)
   flappyBird = game.GameState()
   # 初始化选择的动作、返回的状态、得分
   action0 = np.array([1,0]) # [1,0]表示不做动作，[0,1]表示跳
   observation0, reward0, terminal = flappyBird.frame_step(action0)
   observation0 = cv2.cvtColor(cv2.resize(observation0, (80, 80)), cv2.COLOR_BGR2GRAY)
   ret, observation0 = cv2.threshold(observation0,1,255,cv2.THRESH_BINARY)
   brain.setInitState(observation0)
   while True:
      action = brain.getAction()
      # 从游戏中返回状态、得分、游戏是否结束
      nextObservation,reward,terminal = flappyBird.frame_step(action)
      nextObservation = preprocess(nextObservation)
      brain.setPerception(nextObservation,action,reward,terminal)
def main():
   playFlappyBird()
if __name__ == '__main__':
   main()

CV2的简单介绍

对于HappyBird的状态的设定，我们是通过python的CV2库来读入80*80的图片来作为HappyBird的状态。

那么CV2库怎么用呢？

下面来简单介绍一下：

Cv2提供了简单的图片加载，显示保存方法：

加载图片:

cv2.imread(path, flags)函数

窗体建立：（用来显示图片）

cv2.namedWindow()

显示图像:

cv2.imshow(wname,img)，第一个参数是显示图像的窗口的名字，第二个参数是要显示的图像

图片存储：

cv2.imwrite()

另外还提供了对于图片的一些操作：

颜色空间转换：

彩色图像转为灰度图像

img2 = cv2.cvtColor(img,cv2.COLOR_RGB2GRAY)

灰度图像转为彩色图像

img3 = cv2.cvtColor(img,cv2.COLOR_GRAY2RGB)

基本图像处理：

cv2.resize()实现缩放：

裁剪则是利用array自身的下标截取实现

Threshold()：固定阈值二值化

ret, dst = cv2.threshold(src, thresh, maxval, type)

Src: 输入图，只能输入单通道图像，通常来说为灰度图

Dst: 输出图

Thresh: 阈值

Maxval: 当像素值超过了阈值（或者小于阈值，根据type来决定），所赋予的值

Type:二值化操作的类型，包含以下5种类型: cv2.THRESH_BINARY；cv2.THRESH_BINARY_INV；cv2.THRESH_TRUNC；cv2.THRESH_TOZERO；cv2.THRESH_TOZERO_INV

参考资料：

莫烦强化学习教学视频https://mofanpy.com/tutorials/machine-learning/reinforcement-learning/

莫烦tensorflow教学视频：

https://mofanpy.com/tutorials/machine-learning/tensorflow/

Csdn博客：

https://blog.csdn.net/qq_29462849/article/details/80904469?utm_medium=distribute.wap_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-4.wap_blog_relevant_pic&depth_1-utm_source=distribute.wap_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-4.wap_blog_relevant_pic

- END -

神经网络 tensorflow 游戏 https 网络安全

0 人点赞