1. np.stack((x_t, x_t, x_t, x_t), axis=2) 将图片进行串接的操作，使得图片的维度为[80, 80, 4]
参数说明: (x_t, x_t, x_t, x_t) 表示需要进行串接的图片, axis = 2 表示在第三个维度上进行串接操作
2. cv2.resize(x, [80, 80]) # 将图片的维度变化为80 * 80的维度
参数说明, x为输入的图片，80, 80表示图片变化的维度
3.cv2.cvtColor(x_t, tf.COLOR_RGB2GREY) 将图片转换为灰度图
参数说明, x_t表示输入的图片， tf.COLOR_RGB2GREY表示颜色转换的模式
4.cv2.threshold(x_t, 1, 255, cv2.THRESH_BINARY) # 将大于等于1的索引变化为255，等于0的不变，也就是对图片进行二值化操作
参数说明: x_t表示输入图片，1表示阈值，255表示大于阈值后的数值， cv2.THRESH_BINARY表示进行二值化的模式
5. random.sample(D, batch_size) 从D样本中随机抽取batch_size个数据
参数说明：D表示样本，batch_size表示抽取样本的个数
因为对于每一帧图像而言，一只鸟在图像中的移动方向可能是向上的也可能是向下的
代码说明：使用的网络是卷积网络，
输入的是4帧的图片，即80*80*4，
预测结果为当前位置的上下两个方向的V值，即奖励值
损失值： 为当前真实的奖励值： cost = V[state] - (r_1 + GAMMA * V[next_state])， 目的是为了当前的预测奖励值 与 实际公式计算的奖励值越接近越好
V[state] 表示当前位置的预测奖励值
r_1 表示下一个状态的及时奖励
V[next_state] 表示下一个位置的预测奖励值
代码说明： 主要代码包括来部分，第一部分：主要包括构造网络结构，生成[None, 2]的输出结果，第二部分，获得一个batch的值来进行模型的训练
第一部分：构造网络模型， 用于进行v的预测
第一步：第一层卷积的W_conv1, 大小为[8, 8, 4, 32], 第一层卷积的b_conv1, 大小为[32]
第二步：第二层卷积的W_conv2, 大小为[4, 4, 32, 64], 第二层卷积的b_conv2, 大小为[64]
第三步：第三层卷积的W_conv3, 大小为[3, 3, 64, 64], 第三层卷积的b_conv3, 大小为[64]
第四步：使用tf.placeholder('float32', [None, 80, 80, 4]) # 初始化输入的数据s
第五步：使用tf.nn.relu() 构造第一层卷积层 步长为4，使用max_pool_2x2 进行池化操作
第六步：使用tf.nn.relu() 构造第二层卷积层 步长为2
第七步：使用tf.nn.relu(conv2d(x, W_conv3, 1) + b) 步长为1 来构造第三层卷积层
第八步：使用tf.reshape(h_conv4, [-1, 1600]) # 将卷积后的结果进行拉平操作
第九步：使用tf.nn.relu(tf.matmul(x, w_fc1) + b) # 进行第一次的全连接操作
第十步：使用tf.matmul(x, w_out) + b # 进行输出层的全连接操作，输出的结果为[None, action]， 返货的结果为s，readout， h_fc1, s为输入结果，readout为输出结果，h_fc1为第一层全连接的输出结果
第二部分：将上述求得的s,readout, h_fc1, sess, 传入，这个先观察1000次，1000以后再进行模型参数的训练，同时这里使用一个epsilon, 当random.random() 小于epsilon，那么下一个动作的位置方向是随机值，否者的话就使用np.argmax(readout), 即readout预测出当前位置v值大的索引值作为方向
第一步：定义网络的损失值和训练步骤
第一步：使用tf.placeholder('float32', [None, 2]) 用来构造a即方向
第二步： 使用tf.placeholder('float32', [None]) 用来构造实际当前位置的V值
第三步：预测结果: action_readout = tf.reduce_sum(tf.multiply(readout, a)) 来获得当前预测的结果v奖励值
第四步： 构造cost函数，使用tf.reduce_mean(tf.square(action_readout - y))
第五步：构造train_op, 使用tf.trian.Adaoptimer(1e-6).minimize(cost)
第六步：使用sess.run(tf.global_variable_initial()) 进行参数的初始化操作
第二步：构造第一个输出的参数, 大小为[80, 80, 4]
第一步：使用game_state = game.GameState() 来实例化game_state的向量
第二步：使用D = deque() 来构造存储的向量，用于存储st 当前位置， a_t下一个动作的方向, r_t，下一个动作的及时奖励，以及st1，下一个位置
第三步： 构造一个初始化方向， do_nothing
第一步： np.zeros([2]) 来构造do_nothing的初始值
第二步：do_nothing[0] = 1 将do_nothing的第一个位置的索引值构造为1
第四步：将do_nothing 输入到game_state.frame_step(do_nothing) 获得do_nothing 下一个位置的x_t当前图片， r_0为及时奖励, terminal是否停止
第五步：使用cv2.cvtColor(cv2.resize(x_t, (80, 80), cv2.COLOR_RGB2GRAY)) 将图片进行维度变换，同时将其转换为灰度图
第六步： 使用cv2.threshold(x_t, 1, 255, cv2.THRESH_BINARY) # 进行二值化操作
第七步：使用np.stack((x_t, x_t, x_t, x_t), axis=2) # 将此时的四张图片做一个串接，作为第一个输入s_t
第三步：进入循环，用于进行参数的训练， t = 0
第一步：使用readout.eval(feed_dict={s:s_t}) 来获得当前位置的两个方向的v值
第二步：根据随机数和np.argmax() 来获得方向值
第一步：使用np.zeros(2) 来获得a_t的初始值
第二步： 使用random.random() 来获得随机值，如果随机值小于epsilon，使用random.ranrange(ACTION) 来获得一个0,1的随机值
第三步：a_t[index] = 1, 来构造a_t
第四步： 如果随机值大于epsilon， 使用np.argmax(t_readout) 来获得方向较大的预测值v，来作为索引值
第五步： a_t[index] = 1 来构造a_t
第三步： 使用game_state.frame_step(a_t) 来获得x_t1的图片，r_1的及时奖励， terminal判断是否停止了, 将其与s_t的前三帧图片进行拼接
第一步：使用game_state.frame_step(a_t) 来获得x_t1, r_1, terminal
第二步：使用cv2.cvtColor(cv2.resize(x_t1, (80, 80)), cv2.COLOR_RGB2GRAY) 将彩图转换为灰度图
第三步：使用cv2.threshold(x_t, 1, 255, cv2.THRESH_BINARY) 进行二值化操作
第四步：使用np.reshape(x_t, [80, 80, 1]) 将图片转换为[80, 80, 1] 的数据
第五步：使用np.append(x_t, s_t[:, :, 3], axis=2) 将图片进行拼接
第四步： 将s_t, a_t, r_1. s_t1 存储在D里面, s_t当前位置，a_t下一个方向，r_1下一个方向的奖励值，s_t1下一个位置
第一步：D.append((s_t, a_t, r_1, s_t1))
第二步：判断len(D) 大于MERORY， 使用D.popleft() 去除最后一张图片，使其保持5000的个数
第五步：如果t > OBSERVE, 使用random.sample(D， BATCH_SIZE) 从D中所及获取batch_size章图片进行模型的训练,sess.run(train_op)
第一步：ministbatch = random.sample(D, BATCH_SIZE) 从D中获得batch_size的信息
第二步：s_j_batch = [r[0] for r in mnistbatch] 获得当前位置， a_batch 下一个动作的方向， r_batch下一个动作的及时奖励，s_j1_batch 获得下一个位置的图片
第三步：使用readou_batch = readout.eval(feed_dict={s:s_j1_batch}) 来获得下一个位置的v奖励值，
第四步：构造y_batch，用于存储当前位置的v奖励值
第一步：循环len(ministbatch)
第二步：terminal = ministbatch[i][4] 来获得此时的terminal
第三步：if terminal 表示停止了，那么y_batch.append(r_batch[i]) # 当前的预测奖励值使用r_batch[i]
第四步：如果没有停止，那么y_batch.append(r_batch[i] + GAMMA*readout_batch[i]) # 当前的预测奖励值使用r_batch[j] + readout_batch[i] 加上的是下一个位置的奖励值
第六步： 使用sess.run(train_op, feed_dict={s: s_j_batch, a:a_bacth, y:y_batch}) ，s_j_batch表示当前位置，a_batch表示一个动作的方向，y_batch，表示当前位置的实际奖励值
第七步：s_t = s_t1, 将下一个动作的图片赋值给当前图片，t = t + 1 更新迭代次数
第八步：如果迭代次数为10000， 使用saver.save(global_step=t) 将模型进行保存
第九步：使用epislon -= (initial - final) / 进行episilon的更新操作
第十步：打印各个参数信息
train.py
import tensorflow as tf import numpy as np import sys sys.path.append('game/') import wrapped_flappy_bird as game import random from collections import deque import cv2 ACTION = 2 # 方向的种类 BATCH_SIZE = 64 # 每一个batch值 INITIAL_EPSILON = 0.1 # 刚开始随机的比例 FINAL_EPSILON = 0.0001 # 结束时随机的比例 OBSEVER = 1000 # 观察的次数 EXPLORE = 300000 # 进行探索的次数 PREPLAY_MEMORY = 50000 # D中保存图片的数量 FRAME_PER_ACTION = 1 # 每一次迭代的次数 GAMMA = 0.99 # 下一次奖励的衰减值 GAME ='bird' # 游戏的名字，用于存储save的名字 # 构造w权重参数，输入为shape，使用的是tf.truncated_normal def weight_variable(shape): w = tf.truncated_normal(shape, stddev=0.1) return tf.Variable(w) # 构造b权重参数，输入为shape, 使用的是tf.constant def biase_variable(shape): b = tf.constant(0.0, shape=shape) return tf.Variable(b) # 构造卷积层， 使用tf.nn.conv2d def conv2d(x, w, stride): return tf.nn.conv2d(x, w, strides=[1, stride, stride, 1], padding='SAME') # 构造池化层, 使用tf.nn.max_pool def max_pool_2x2(x): return tf.nn.max_pool(x, strides=[1, 2, 2, 1], ksize=[1, 2, 2, 1], padding='SAME') # 第一部分：用于构建网络，输出结果为当前位置两个方向的奖励值 def creatNetwork(): # 第一步：构造第一层卷积层W的参数 [32] W_conv1 = weight_variable([8, 8, 4, 32]) # 构造第一层卷积层b的参数 [32] b_conv1 = biase_variable([32]) # 第二步：构造第二层卷积层w的参数, [4, 4, 32, 64] W_conv2 = weight_variable([4, 4, 32, 64]) # 构造第二层卷积层b的参数 [64] b_conv2 = biase_variable([64]) # 第三步：构造第三层卷积层w的参数 [3, 3, 64, 64] W_conv3 = weight_variable([3, 3, 64, 64]) # 构造第三层卷积层b的参数 [64] b_conv3 = biase_variable([64]) # 构造全连接的w参数[1600, 512] W_fc1 = weight_variable([1600, 512]) # 构造全连接的b参数[512] b_fc1 = biase_variable([512]) # 构造输出层的w参数[512, 2] W_out = weight_variable([512, ACTION]) # 构造输出层的b参数[2] b_out = biase_variable([ACTION]) # 第四步：构造输入数据的维度[None, 80, 80, 4] s = tf.placeholder('float32', [None, 80, 80, 4]) # 第五步：第一层卷积， 使用tf.nn.relu() 和 pool层 h_conv1 = tf.nn.relu(conv2d(s, W_conv1, 4) + b_conv1) h_pool1 = max_pool_2x2(h_conv1) # 第六步：第二层卷积, 使用tf.nn.relu() 构造第二层卷积 h_conv2 = tf.nn.relu(conv2d(h_pool1, W_conv2, 2) + b_conv2) # 第七步：第三层卷积， 使用tf.nn.relu() 构造第三层卷积 h_conv3 = tf.nn.relu(conv2d(h_conv2, W_conv3, 1) + b_conv3) # 第八步: 进行维度的拉平操作 fc_flatten = tf.reshape(h_conv3, [-1, 1600]) # 第九步：进行第一层全连接, h_fc1 = tf.nn.relu(tf.matmul(fc_flatten, W_fc1) + b_fc1) # 第十步：构造输出层 readout = tf.matmul(h_fc1, W_out) + b_out # 返回的是s, readout当前位置上下方向的奖励值, h_fc1表示前一层全连接操作 return s, readout, h_fc1 # 第二步：用于构建训练的网络 def trainNetwork(sess, s, readout, h_fc1): # 第一步：构造损失函数cost和训练操作 # 构建方向的输入值 a = tf.placeholder('float32', [None, ACTION]) # 构建当前位置奖励值 y = tf.placeholder('float32', [None]) # 使用位置信息与预测的奖励值进行相乘操作, 获得当前位置的预测奖励值 ACTION_READOUT = tf.reduce_sum(tf.multiply(a, readout)) # 使用预测奖励值与当前位置的奖励平方差来获得损失值 cost = tf.reduce_mean(tf.square(y - ACTION_READOUT)) # 使用tf.train.AdamOptimizer来构造损失值的降低函数 train_op = tf.train.AdamOptimizer(1E-6).minimize(cost) # 进行参数的初始化操作 sess.run(tf.global_variables_initializer()) # 对游戏进行实例化操作 game_state = game.GameState() # 第二步：构造第一个输入的样本，输入的数据的维度为[80, 80, 4] # 构造存储信息的列表 D = deque() # 定义一个初始方向 do_nothing = np.zeros([ACTION]) # 使得初始方向的位置为do_noting[0] = 1 do_nothing[0] = 1 # 根据初始的方向获得第一个位置信息，x_t下一个图的位置, r_0下一个位置的及时奖励，terminal是否停止 x_t, r_0, terminal = game_state.frame_step(do_nothing) # 对生成的图片使用cv2.resize进行维度的变化，将图片转换为灰度图 x_t = cv2.cvtColor(cv2.resize(x_t, (80, 80)), cv2.COLOR_RGB2GRAY) # 进行二值化操作 ret, x_t = cv2.threshold(x_t, 1, 255, cv2.THRESH_BINARY) # 使用np.stack进行图片的拼接操作, 将一帧的图片拼接为[80, 80, 4] s_t = np.stack((x_t, x_t, x_t, x_t), axis=2) # t为第一次迭代 t = 0 # 发现的随机值 epsilon = INITIAL_EPSILON # 第三步：进入循环进行操作的训练 while 'angry bird' != 'happy bird': # 将当前位置的图片输入，获得当前位置上下方向的奖励值 readout_t = readout.eval(feed_dict={s:[s_t]}) if t % FRAME_PER_ACTION == 0: # 将位置进行初始化 a_t = np.zeros(ACTION) # 如果获得的随机值小于阈值，就进行探索操作 if random.random() < epsilon: # 获得随机的一个方向 index = random.randrange(ACTION) # 将方向对应的位置赋值为1 a_t[index] = 1 else: # 获得当前位置两个方向较大的索引值 index = np.argmax(readout_t) # 将位置的索引赋值为1 a_t[index] = 1 else: a_t = do_nothing # 根据方向来获得下一个位置的图片，下一个位置的及时奖励，以及是否停止 x_t1, r_1, terminal = game_state.frame_step(a_t) # 将图片转化为灰度值 x_t1 = cv2.cvtColor(cv2.resize(x_t1, (80, 80)), cv2.COLOR_RGB2GRAY) # 将阈值大于1的设置为255，小于1的就是0不变，进行二值化操作 ret, x_t1 = cv2.threshold(x_t1, 1, 255, cv2.THRESH_BINARY) # 将x_t1进行维度变化转换为(80, 80, 1) x_t1 = np.reshape(x_t1, (80, 80, 1)) # 与测试图片的前3帧图片进行串接操作 s_t1 = np.append(x_t1, s_t[:, :, :3], axis=2) # 将当前位置样本，当前位置，下一个位置的奖励值，下一个位置样本，是否停止加入到D中 D.append((s_t, a_t, r_1, s_t1, terminal)) # 如果D的长度大于memory，使用popleft去除最后一个信息 if len(D) > PREPLAY_MEMORY: D.popleft() # 如果循环次数t大于观测值 if t > OBSEVER: # 获得一个batch的图片信息 mnistbatch = random.sample(D, BATCH_SIZE) # 获得当前样本的图片信息, [80, 80, 4] s_j_batch = [r[0] for r in mnistbatch] # 获得当前位置的方向信息，即向上的索引值为[1, 0], 向下的为[0, 1] a_batch = [r[1] for r in mnistbatch] # 下一个方向的及时奖励值 r_batch = [r[2] for r in mnistbatch] # 获得下一个位置的组合图片信息 [80, 80, 4] s_j1_batch = [r[3] for r in mnistbatch] # 获得下一个方向的v奖励值 readout_j1_batch = readout.eval(feed_dict={s:s_j1_batch}) # 用于进行v奖励值的存储 y_batch = [] # 循环，计算每一个位置的奖励值v, 即r for i in range(len(mnistbatch)): # 获得是够已经停止 terminal = mnistbatch[4][i] if terminal: # 如果停止，当前的奖励值为及时奖励 y_batch.append(r_batch[i]) else: # 否者，当前位置的奖励值为及时奖励 + 奖励衰减 * 下一个位置的奖励值 y_batch.append(r_batch[i] + GAMMA * readout_j1_batch[i]) # 进行缩小损失值的操作，输入为方向信息，y为当前位置的奖励值, s_j_batch表示当前位置的图片信息 train_op.eval(feed_dict={ a : a_batch, y: y_batch, s: s_j_batch }) # 将下一个阶段的位置信息赋值给当前位置，用于进行迭代 s_t = s_t1 # 将循环的次数+1 t = t + 1 # 构造saver进行参数的保存 saver = tf.train.Saver() if t % 1000 == 0: # 每迭代1000次，就进行参数的保存 saver.save(sess, 'saved_network/' + GAME + '-dqn', global_step=t) # 获得当前的位置 state = '' if t <= OBSEVER: state = 'observe' elif t > OBSEVER and t <= OBSEVER + EXPLORE: state = 'explore' else: state = 'train' # 降低探索的值，以保证探索的比例越来越小 if epsilon > FINAL_EPSILON and t > OBSEVER: epsilon -= (INITIAL_EPSILON - FINAL_EPSILON) / EXPLORE # 进行参数的打印 print('TIMESTEP', t, '/STATE', state, \ '/ EPSILON', epsilon, '/ ACTION', a_t, '/ REWARD', r_1, \ '/ Q_MAX %e' % np.max(readout_t)) def playGame(): # 构造执行函数 sess = tf.InteractiveSession() # 第一部分：卷积模型的构建，输出为s输入值, readout为当前的两个方向的奖励值，h_fc1第一层全连接的结果 s, readout, h_fc1 = creatNetwork() trainNetwork(sess, s, readout, h_fc1) def main(): # 建立游戏 playGame() if __name__ == '__main__': main()
效果图
微信扫一扫 
支付宝扫一扫 