强化学习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，运行源码需要加上下面的代码来降版本！
import tensorflow.compat.v1 as tftf.disable_v2_behavior()首先是对各个参数以及神经网络的初始化:
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（现实）网络：
# 复制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,达到训练目的，并对神经网络进行存取。
# 创建训练方法
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值列表。
# 四通道输入图像
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）来训练其实就是将游戏的过程打成碎片存储，训练时随机抽取就避免了相关性问题。
# 更新状态、经验池
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的获取。
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的高手。
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 -