当要解决的问题的值函数可以用少量的离散值表示时，我们可以直接学习每个离散的值函数。可是，当要解决的问题的值函数需要用大量的甚至是无穷多的离散值来表示时，学习每个离散的函数值需要耗费大量的时间和存储空间，这时可以用函数逼近的方法来表示值函数。
6.1 从表格型强化学习到线性函数逼近强化学习
从第2章到第5章中，一直默认值函数v(s)或者行为-值函数Q(s,a)由一个表格表示。这种表示方法的优点是对于离散问题，表格能精确地表示该处的值。然而，利用表格来表示行为-值函数存在维数灾难的问题。比如，状态空间s的维数是n，每个维度离散化为d个数，则行为-值函数表格中元素的个数为，其中|A|为动作空间中动作的个数。为了解决维数灾难，我们用函数逼近的方法来表示行为-值函数。函数逼近最简单的方法是线性参数逼近，即
（6.1）
其中为特征函数，
为参数。
6.1.1 表格特征表示
表格型函数可以看成函数逼近方法的一种特殊形式，每个格点表示一个特征。以鸳鸯系统为例，以表格的形式表示的行为-值函数共有100x4个元素，每个元素对应一个特征，则特征函数可以写为：式（6.2）
其中=100,即状态空间的个数，
当采取式（6.1）来表示行为-值函数的时候，则等价于原表格中的行为-值函数。
6.1.2 固定稀疏表示
当采用表格特征表示行为-值函数的时候，每个状态-行为对都是一个特征，特征的维数和参数的个数都是|S|x|A|，装填空间往往随着状态的维数呈指数级增长，因此，可以用表格特征来表示行为-值函数会遇到维数灾难。
固定稀疏表示的方法。假设状态空间的维数为n维，即,每个维度离散化为d个数，则状态空间第i维共有d个数，记为
，其中j=1,...,d。用固定稀疏来表示状态的特征为式（6.3）：
其中：
状态-行为值的特征可以表示为
其中：
固定稀疏表示的特征个数和参数的个数为，该表示特征的个数随维数线性增长而非指数增长。
6.1.3 参数的训练
为了训练参数,构建损失函数：式（6.7）
其中为要学习的目标值函数，如果利用时间差分学习方法，则
为时间差分目标，即
.
根据式（6.7），利用梯度学习的方法学习参数，则式（6.8）：
当采用式（6.1）、式（6.2）的形式表示行为-值函数的时候，式（6.8）与第5章的式（5.5）等价。
6.2 基于线性函数逼近的Q-Learning的算法实现
下面是基于线性函数逼近的Q-Learning学习算法的伪代码：
算法实现基于表格特征的学习和基于固定稀疏表示的学习。
本章所用环境和第5章相同，我们声明一个线性函数逼近算法类LFA_RL.如下面的代码所示。首先导入环境类，然后声明一个函数逼近算法类LFA_RL.在算法类的初始化子函数中初始化折扣因子gamma，环境YuanYang，表格特征对应的参数theta_tr,固定稀疏表示所对应的参数theta_fsr。下面先实现基于表格特征表示的Q-Learning算法。
定义成员函数，完成动作与对应的数字之间的转换，便于索引查找。
定义表格特征函数，该函数的输入为状态-动作对，输出为该状态-动作对所对应的特征，此处的特征为表格特征。
定义表格特征表示的行为-值函数对对应的贪婪函数，其中行为-值函数定义为,具体代码如下：
下面的代码定义基于表格特征的e-greedy策略，输入为状态，输出为e-greedy策略。
下面定义基于表格特征的贪婪策略测试函数greedy_test_tr。如果雄鸟以最短的路径找到雄鸟则训练结束。
定义基于表格特征的Q-Learning算法，在该算法中，首先初始化要学习的参数theta_tr，然后进入大循环进行学习。
在学习之前，先调用贪婪策略测试函数，看看是否已经完成任务，如果能以最短路径完成任务则停止训练。
进入内循环，使得智能与环境进行交互，从环境中获得回报，并根据回报进行学习。
获得时间差分目标后，利用梯度下降的方法对参数进行更新。最后返回学到的参数theta_tr，具体代码如下：
下面我们实现基于固定稀疏表示的Q-Learning方法。首先定义特征函数，在这里我们将x方向离散为10个数，y方向离散为10个数，动作个数为4，则特征的维数为（10+10）x4=80.
定义基于稀疏表示的贪婪策略greedy_policy_fsr。
定义基于固定稀疏表示的e-greedy策略，用于采样动作。
定义基于固定稀疏表示的贪婪策略评估，如果智能体以贪婪策略完成目标，则返回标志位。
定义基于固定稀疏表示的Q-Learning算法。首先初始化固定稀疏表示参数theta_tr，然后进入大循环学习。其学习过程与基于表格特征的算法学习过程类似。
def qlearning_lfa_fsr(self, num_iter, alpha, epsilon):
iter_num = []
self.theta_fsr = np.zeros((80, 1)) * 0.1
# 大循环
for iter in range(num_iter):
# 随机初始化状态
# s = yuanyang.reset()
s = 0
flag = self.greedy_test_fsr()
if flag == 1:
iter_num.append(iter)
if len(iter_num) < 2:
print("qlearning_fsr 第一次完成任务需要的迭代次数为：", iter_num[0])
if flag == 2:
print("qlearning_fsr 第一次实现最短路径需要的迭代次数为：", iter)
break
s_sample = []
# 随机选初始动作
# a = self.actions[int(random.random()*len(self.actions))]
a = self.epsilon_greedy_policy_fsr(s, epsilon)
t = False
count = 0
while False == t and count < 30:
# 与环境交互得到下一个状态
s_next, r, t = yuanyang.transform(s, a)
# print(s)
# print(s_next)
a_num = self.find_anum(a)
if s_next in s_sample:
r = -2
s_sample.append(s)
if t == True:
q_target = r
else:
# 下一个状态处的最大动作，a1用greedy_policy
a1 = self.greedy_policy_fsr(s_next)
a1_num = self.find_anum(a1)
# 得到时间差分目标
q_target = r + self.gamma * np.dot(self.feature_fsr(s_next, a1_num), self.theta_fsr)
# print("q_target", q_target[0,0],np.sum(self.feature(s,a_num)))
# 基于时间差分目标和梯度下降法更新行为值函数的参数
self.theta_fsr = self.theta_fsr + alpha * (q_target - np.dot(self.feature_fsr(s, a_num), self.theta_fsr))[
0, 0] * np.transpose(self.feature_fsr(s, a_num))
s = s_next
# 行为策略
a = self.epsilon_greedy_policy_fsr(s, epsilon)
count += 1
return self.theta_fsr最后，创建主函数来测试两种算法的性能。
if __name__=="__main__":
yuanyang = YuanYangEnv()
brain = LFA_RL(yuanyang)
brain.qlearning_lfa_fsr(num_iter=5000,alpha=0.1,epsilon=0.1)
brain.qlearning_lfa_tr(num_iter=5000, alpha=0.1, epsilon=0.1)
#打印学到的值函数
qvalue2 = np.zeros((100,4))
qvalue1 = np.zeros((100,4))
for i in range(400):
y = int(i/100)
x = i-100*y
qvalue2[x,y] = np.dot(brain.feature_tr(x,y),brain.theta_tr)
qvalue1[x,y] = np.dot(brain.feature_fsr(x,y),brain.theta_fsr)
yuanyang.action_value = qvalue1
##########################################
# 测试学到的策略
flag = 1
s = 0
# print(policy_value.pi)
step_num = 0
path = []
# 将最优路径打印出来
while flag:
# 渲染路径点
path.append(s)
yuanyang.path = path
# a = brain.greedy_policy_tr(s)
a = brain.greedy_policy_fsr(s)
print('%d->%s\t' % (s, a), qvalue1[s, 0], qvalue1[s, 1], qvalue1[s, 2], qvalue1[s, 3])
yuanyang.bird_male_position = yuanyang.state_to_position(s)
yuanyang.render()
time.sleep(0.25)
step_num += 1
s_, r, t = yuanyang.transform(s, a)
if t == True or step_num > 30:
flag = 0
s = s_
# 渲染最后的路径点
yuanyang.bird_male_position = yuanyang.state_to_position(s)
path.append(s)
yuanyang.render()
while True:
yuanyang.render()
运行效果
看具体的终端输出：
LFA_RL.py
from yuanyang_env_fa import *
from yuanyang_env_fa import YuanYangEnv
class LFA_RL:
def __init__(self, yuanyang):
self.gamma = yuanyang.gamma
self.yuanyang = yuanyang
self.theta_tr = np.zeros((400,1))*0.1
self.theta_fsr = np.zeros((80,1))*0.1
# 找到动作所对应的序号
def find_anum(self, a):
for i in range(len(self.yuanyang.actions)):
if a == self.yuanyang.actions[i]:
return i
#下面实现表格特征表示
def feature_tr(self,s,a):
phi_s_a = np.zeros((1,400))
phi_s_a[0, 100*a+s] = 1
return phi_s_a
#定义贪婪策略
def greedy_policy_tr(self,state):
qfun = np.array([0,0,0,0])*0.1
#计算行为值函数Q(s,a)=phi(s,a)*theta
for i in range(4):
qfun[i] = np.dot(self.feature_tr(state,i),self.theta_tr)
amax=qfun.argmax()
return self.yuanyang.actions[amax]
#定义epsilon贪婪策略
def epsilon_greedy_policy_tr(self, state, epsilon):
qfun = np.array([0, 0, 0, 0])*0.1
# 计算行为值函数Q(s,a)=phi(s,a)*theta
for i in range(4):
qfun[i] = np.dot(self.feature_tr(state, i), self.theta_tr)
amax = qfun.argmax()
# 概率部分
if np.random.uniform() < 1 - epsilon:
# 最优动作
return self.yuanyang.actions[amax]
else:
return self.yuanyang.actions[int(random.random() * len(self.yuanyang.actions))]
# 定义贪婪策略
def greedy_test_tr(self):
s = 0
s_sample = []
done = False
flag = 0
step_num = 0
while False == done and step_num < 30:
a = self.greedy_policy_tr(s)
# 与环境交互
s_next, r, done = self.yuanyang.transform(s, a)
s_sample.append(s)
s = s_next
step_num += 1
if s == 9:
flag = 1
if s == 9 and step_num < 21:
flag = 2
return flag
def qlearning_lfa_tr(self,num_iter, alpha, epsilon):
iter_num = []
self.theta_tr = np.zeros((400, 1)) * 0.1
#大循环
for iter in range(num_iter):
#随机初始化状态
# s = yuanyang.reset()
s=0
flag = self.greedy_test_tr()
if flag == 1:
iter_num.append(iter)
if len(iter_num)<2:
print("qlearning_tr 第一次完成任务需要的迭代次数为：", iter_num[0])
if flag == 2:
print("qlearning_tr 第一次实现最短路径需要的迭代次数为：", iter)
break
s_sample = []
#随机选初始动作
# a = self.actions[int(random.random()*len(self.actions))]
a = self.epsilon_greedy_policy_tr(s,epsilon)
t = False
count = 0
while False==t and count < 30:
#与环境交互得到下一个状态
s_next, r, t = yuanyang.transform(s, a)
# print(s)
# print(s_next)
a_num = self.find_anum(a)
if s_next in s_sample:
r = -2
s_sample.append(s)
if t == True:
q_target = r
else:
# 下一个状态处的最大动作，a1用greedy_policy
a1 = self.greedy_policy_tr(s_next)
a1_num = self.find_anum(a1)
# qlearning的更新公式TD(0)
q_target = r + self.gamma * np.dot(self.feature_tr(s_next, a1_num),self.theta_tr)
# print("q_target", q_target[0,0],np.sum(self.feature(s,a_num)))
# 利用td方法更新动作值函数
self.theta_tr= self.theta_tr + alpha * (q_target - np.dot(self.feature_tr(s,a_num),\
self.theta_tr))[0,0]*np.transpose(self.feature_tr(s,a_num))
s = s_next
#行为策略
a = self.epsilon_greedy_policy_tr(s, epsilon)
count += 1
return self.theta_tr
############下面实现固定稀疏表示#############
def feature_fsr(self,s,a):
phi_s_a = np.zeros((1,80))
y = int(s/10)
x = s-10*y
phi_s_a[0, 20*a+x] = 1
phi_s_a[0,20*a+10+y]=1
return phi_s_a
def greedy_policy_fsr(self, state):
qfun = np.array([0, 0, 0, 0]) * 0.1
# 计算行为值函数Q(s,a)=phi(s,a)*theta
for i in range(4):
qfun[i] = np.dot(self.feature_fsr(state, i), self.theta_fsr)
amax = qfun.argmax()
return self.yuanyang.actions[amax]
# 定义epsilon贪婪策略
def epsilon_greedy_policy_fsr(self, state, epsilon):
qfun = np.array([0, 0, 0, 0]) * 0.1
# 计算行为值函数Q(s,a)=phi(s,a)*theta
for i in range(4):
qfun[i] = np.dot(self.feature_fsr(state, i), self.theta_fsr)
amax = qfun.argmax()
# 概率部分
if np.random.uniform() < 1 - epsilon:
# 最优动作
return self.yuanyang.actions[amax]
else:
return self.yuanyang.actions[int(random.random() * len(self.yuanyang.actions))]
def greedy_test_fsr(self):
s = 0
s_sample = []
done = False
flag = 0
step_num = 0
while False == done and step_num < 30:
a = self.greedy_policy_fsr(s)
# 与环境交互
s_next, r, done = self.yuanyang.transform(s, a)
s_sample.append(s)
s = s_next
step_num += 1
if s == 9:
flag = 1
if s == 9 and step_num < 21:
flag = 2
return flag
def qlearning_lfa_fsr(self, num_iter, alpha, epsilon):
iter_num = []
self.theta_fsr = np.zeros((80, 1)) * 0.1
# 大循环
for iter in range(num_iter):
# 随机初始化状态
# s = yuanyang.reset()
s = 0
flag = self.greedy_test_fsr()
if flag == 1:
iter_num.append(iter)
if len(iter_num) < 2:
print("qlearning_fsr 第一次完成任务需要的迭代次数为：", iter_num[0])
if flag == 2:
print("qlearning_fsr 第一次实现最短路径需要的迭代次数为：", iter)
break
s_sample = []
# 随机选初始动作
# a = self.actions[int(random.random()*len(self.actions))]
a = self.epsilon_greedy_policy_fsr(s, epsilon)
t = False
count = 0
while False == t and count < 30:
# 与环境交互得到下一个状态
s_next, r, t = yuanyang.transform(s, a)
# print(s)
# print(s_next)
a_num = self.find_anum(a)
if s_next in s_sample:
r = -2
s_sample.append(s)
if t == True:
q_target = r
else:
# 下一个状态处的最大动作，a1用greedy_policy
a1 = self.greedy_policy_fsr(s_next)
a1_num = self.find_anum(a1)
# 得到时间差分目标
q_target = r + self.gamma * np.dot(self.feature_fsr(s_next, a1_num), self.theta_fsr)
# print("q_target", q_target[0,0],np.sum(self.feature(s,a_num)))
# 基于时间差分目标和梯度下降法更新行为值函数的参数
self.theta_fsr = self.theta_fsr + alpha * (q_target - np.dot(self.feature_fsr(s, a_num), self.theta_fsr))[
0, 0] * np.transpose(self.feature_fsr(s, a_num))
s = s_next
# 行为策略
a = self.epsilon_greedy_policy_fsr(s, epsilon)
count += 1
return self.theta_fsr
if __name__=="__main__":
yuanyang = YuanYangEnv()
brain = LFA_RL(yuanyang)
brain.qlearning_lfa_fsr(num_iter=5000,alpha=0.1,epsilon=0.1)
brain.qlearning_lfa_tr(num_iter=5000, alpha=0.1, epsilon=0.1)
#打印学到的值函数
qvalue2 = np.zeros((100,4))
qvalue1 = np.zeros((100,4))
for i in range(400):
y = int(i/100)
x = i-100*y
qvalue2[x,y] = np.dot(brain.feature_tr(x,y),brain.theta_tr)
qvalue1[x,y] = np.dot(brain.feature_fsr(x,y),brain.theta_fsr)
yuanyang.action_value = qvalue1
##########################################
# 测试学到的策略
flag = 1
s = 0
# print(policy_value.pi)
step_num = 0
path = []
# 将最优路径打印出来
while flag:
# 渲染路径点
path.append(s)
yuanyang.path = path
# a = brain.greedy_policy_tr(s)
a = brain.greedy_policy_fsr(s)
print('%d->%s\t' % (s, a), qvalue1[s, 0], qvalue1[s, 1], qvalue1[s, 2], qvalue1[s, 3])
yuanyang.bird_male_position = yuanyang.state_to_position(s)
yuanyang.render()
time.sleep(0.25)
step_num += 1
s_, r, t = yuanyang.transform(s, a)
if t == True or step_num > 30:
flag = 0
s = s_
# 渲染最后的路径点
yuanyang.bird_male_position = yuanyang.state_to_position(s)
path.append(s)
yuanyang.render()
while True:
yuanyang.render()
6.3 非线性函数逼近DQN算法代码实现
利用线性函数对行为-值函数进行逼近，收敛性和单调性都比较好，但是线性函数的表示能力有限，无法满足实际需要。相反，非线性函数的逼近能力很强，尤其是神经网络。随着深度学习技术的发展和计算机计算能力的提升，利用深度网络来逼近行为-值函数称为当下研究的热点和主流。将深度网络的表示能力和强化学习的决策能力结合而形成的深度强化学习在很多领域取得突破性进展，如视频游戏、围棋等。
第一个成功的深度强化学习算法是DeepMind与2015年在Nature上发表的用于完雅达利游戏的DQN算法。雅达利游戏的规则是玩家根据电脑画面控制17个游戏键盘，以赢得尽可能多的分数。
从强化学习的角度来看，DQN算法并不是全新的算法。它只是利用深度卷积神经网络来表示值函数，算法的整个学习框架为Q-Learning。其实在20世纪90年代，已经有人用神经网络来表示行为-值函数了，但是当时并没有取得如此轰动的效果。除了计算力本身的问题，DQN还做了以下几个创新：
①用卷积神经网络来表示行为-值函数
②利用了经验回放的技术，从实验池中随机抽取数据，从未消除了相邻数据之间的关联性。
③设置了独立的目标网络，使得学习更加稳定。
深度行为-值函数网络结果如下图所示。
Q-Learning和DQN算法的伪代码如下：
(1)Q-Learning算法伪代码的行1为初始化行为-值函数表格为0；
因为行为-值函数是用卷积神经网络表示的，所以DQN算法行1-3进行初始化神经网络，同时DQN算法利用和经验回放数据进行学学习，所以就初始化了一个经验池；并设置了独立的目标网路，所以也初始化了该目标网络。目标网络初始值与值函数网络的初始值相同。
(2)Q-Learning的行5；利用探索-平衡策略e-greedy策略进行采样；
DQN的行7、行8 也利用了探索平衡策略e-greedy进行采样。
（3）Q-Learning的行6利用异策略的时间差分方法对表格值函数进行评估
DQN的行9-14跟Q-Learning类似，都是利用异策略的时间差分对值函数进行更新。
不同的是，DQN利用经验池中的数据进行学习，而Q-Learning利用在线数据学习。所以，DQN算法需要先将与环境交互得到的数据进行处理，然后存放到经验池中，行9完成于环境的交互而得到的数据，行10处理数据，行11将数据存储到经验池中，行12从经验池中采样数据，行13利用采样数据得到时间差分目标，行14利用时间差分目标对神经网络进行训练。行16更新目标网络的参数。
本节利用愤怒的小鸟来编写程序。该程序来自开源算法，为了对算法进行更清楚的表述，进行如下的修改：
①定义两个类，一个类为经验池类，该类用于经验数据的存储和训练数据的采集。另一个为深度q学习类，在该类方法中定义DQN学习算法，对小鸟进行训练。
②加入独立的目标网路。原代码中没有独立的目标网络，与DeepMind中的DQN不符。另外，本代码更新网络的方式是soft的，而非每隔C步替换依次目标网络。游戏界面如下：
在该游戏中，玩家通过控制小鸟上下运动来躲避不断到来的柱子，有两个动作可以选择：一个是飞，一个是不进行任何操作，当玩家采用飞这个动作时，小鸟会网上运动，当玩家不操作时，小鸟会往下掉。当小鸟飞行一步而没有碰到主子时立即回报0.1；当小鸟撞到主子立即回报-1；当小鸟躲过一个主子时立即回报1.玩家的目标就是控制小鸟躲过尽量多的柱子，得到尽量多的分数。
具体原代码如下：
①将下面的代码导入必要的库，包括深度学习框架TenrosFlow、数值计算库Numpy、图像处理库cv2、系统控制库sys，以及游戏模块wrapped_flappy_bird.
缺少的包就自己配置啦拉拉
②用下面的代码设置与本算法相关的超参数。游戏的名字为flapped_bird;有效动作的数目ACTIONS为2，即飞行和“什么也不做”；折扣因子gamma设置为099；训练前观察数目POBSERVE,在这段时间内探索率保持不变，以得到各种情况；EXPLORE探索步长设置为30万步，在这段时间内探索率线性减小；最终探索率REPLAY_MEMORY设置为50 000，即训练池中有50 000个可以用于采样学习的数据；批的大小BACTCH为32，即在实际学习训练的时候，从实验池中随机采集32个数据进行训练；跳帧FRAME_PER_ACTION这里取1.
③用下面的代码定义一个经验回报类，在该类中定义了两个类成员子函数add_experimence和sample，分别完成向经验池中添加一条经验数据和采集训练数据样本的功能。具体代码如下：在类的初始化函数中定义一个空的经验池以及经验池的最大容量。在经验添加子函数add_experimence中，先判断经验池是否已经满了，如果满了，则将最顶端的数据清空，换成最新的经验数据。在采样子函数中，随机采样mini-batch的数据，然后将数据进行整理，返回训练所需要的数据格式。
④下面的代码定义DQN算法类Deep_Q_N,在该类中包括以下几个类成员子函数：
（1）初始化类成员函数__init__，在该函数内我们调用TensorFlow，声明一个图，定义输入层，调用类成员子函数创建行为-值网络、目标值网络，定义目标值网络的更新方式，定义损失函数，构建优化器，初始化图中变量，保存声明。
（2）模型存储子函数save_model，用于存储模型参数。
（3）模型恢复子函数restore_model,用于恢复模型参数。
（4）深度q网络构建子函数build_q_net,其输入参数为观测，变量命名空间scope和变量性质trainable。该子函数在初始化成员函数中被调用，由于预测用得行为值函数和用于目标的行为值函数是两套参数，所以可以通过使用不同的命名空间scope区分两组参数。
（5）用于采样动作的利用探索-平衡策略子函数epsilon_greedy，该算法输入为当前状态和探索率，输出为当前状态对应的探索策略，用于与环境进行交互。
(6)网络训练子函数train_Network，该子函数基于Q-Learning的框架，基于神经网络表示的行为值函数对智能体进行训练。
#定义值函数网络，完成神经网络的创建和训练
class Deep_Q_N():
def __init__(self, lr=1.0e-6, model_file=None):
self.gamma = GAMMA
self.tau = 0.01
#tf工程
self.sess = tf.Session()
self.learning_rate = lr
#1.输入层
self.obs = tf.placeholder(tf.float32,shape=[None, 80,80,4])
self.obs_=tf.placeholder(tf.float32, shape=[None, 80,80,4])
self.action = tf.placeholder(tf.float32, shape=[None,ACTIONS])
self.action_=tf.placeholder(tf.float32, shape=[None, ACTIONS])
#2.1 创建深度q网络
self.Q = self.build_q_net(self.obs, scope='eval',trainable=True)
#2.2 创建目标网络
self.Q_=self.build_q_net(self.obs_, scope='target', trainable=False)
#2.3 整理两套网络参数
self.qe_params = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, scope='eval')
self.qt_params = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES,scope='target')
#2.4定义新旧参数的替换操作
self.update_oldq_op = [oldq.assign((1 - self.tau) * oldq + self.tau * p) for p, oldq in zip(self.qe_params, self.qt_params)]
# self.update_oldq_op=[oldq.assign(p) for p, oldq in zip(self.qe_params, self.qt_params)]
#3.构建损失函数
#td目标
self.Q_target =tf.placeholder(tf.float32, [None])
readout_q = tf.reduce_sum(tf.multiply(self.Q, self.action),reduction_indices=1)
self.q_loss = tf.losses.mean_squared_error(labels=self.Q_target, predictions=readout_q)
#4.定义优化器
self.q_train_op = tf.train.AdamOptimizer(lr).minimize(self.q_loss,var_list=self.qe_params)
#5.初始化图中的变量
self.sess.run(tf.global_variables_initializer())
#6.定义保存和恢复模型
self.saver = tf.train.Saver()
if model_file is not None:
self.restore_model(model_file)
#定义存储模型函数
def save_model(self, model_path,global_step):
self.saver.save(self.sess, model_path,global_step=global_step)
#定义恢复模型函数
def restore_model(self, model_path):
self.saver.restore(self.sess, model_path)（7）下面的代码用于创建深度q网络，该网路由3个卷积层、1个池化层、2个全连接层构成。其中第1个卷积层的卷积核是8*8*4*32，步长是4，后面连接一个池化层，池化层的特征是2*2,步长是2.第2个卷积层的卷积核是4*4*32*64，步长为2，后面连一个卷积层。第3个卷积层的卷积核大小为3*3*64*64,步长为1，将3个卷积层的输出展开成维数为1600的1维向量，后面接两个全连接层，第1个全连接层为1600*512，激活函数为ReLU.第2个全连接层为512*2，没有激活函数，即线性输出。
def build_q_net(self, obs, scope, trainable):
with tf.variable_scope(scope):
h_conv1 = tf.layers.conv2d(inputs=obs, filters=32, kernel_size=[8,8],strides=4,padding="same",activation=tf.nn.relu,kernel_initializer=tf.random_normal_initializer(mean=0, stddev=0.01),\
bias_initializer=tf.constant_initializer(0.01),trainable=trainable)
h_pool1 = tf.layers.max_pooling2d(h_conv1, pool_size=[2,2],strides=2, padding="SAME")
h_conv2 = tf.layers.conv2d(inputs=h_pool1,filters=64, kernel_size=[4,4],strides=2, padding="same", activation=tf.nn.relu, kernel_initializer=tf.random_normal_initializer(mean=0, stddev=0.01),\
bias_initializer=tf.constant_initializer(0.01),trainable=trainable)
h_conv3 = tf.layers.conv2d(inputs=h_conv2,filters=64, kernel_size=[3,3],strides=1,padding="same",activation=tf.nn.relu, kernel_initializer=tf.random_normal_initializer(mean=0,stddev=0.01),\
bias_initializer=tf.constant_initializer(0.01),trainable=trainable)
h_conv3_flat = tf.reshape(h_conv3,[-1,1600])
#第一个全连接层
h_fc1 = tf.layers.dense(inputs=h_conv3_flat,units=512,activation=tf.nn.relu, kernel_initializer=tf.random_normal_initializer(0,stddev=0.01),\
bias_initializer=tf.constant_initializer(0.01),trainable=trainable)
#读出层，没有激活函数
qout = tf.layers.dense(inputs=h_fc1, units=ACTIONS,kernel_initializer=tf.random_normal_initializer(0,stddev=0.01),\
bias_initializer=tf.constant_initializer(0.01),trainable=trainable)
return qout（8）下面的代码用来定义探索策略，跟表格型Q-Learning不同的是，这里调用神经网络来确定哪个是最优动作。
def epsilon_greedy(self,s_t,epsilon):
a_t = np.zeros([ACTIONS])
amax = np.argmax(self.sess.run(self.Q,{self.obs:[s_t]})[0])
# 概率部分
if np.random.uniform() < 1 - epsilon:
# 最优动作
a_t[amax] = 1
else:
a_t[random.randrange(ACTIONS)]=1
return a_t（9）下面的代码用于实现对深度值函数网络的训练。基本的过程为与环境交互，将数据存入经验池中，从经验池中采集数据对神经网络进行训练。
def train_Network(self,experience_buffer):
#打开游戏状态与模拟器进行通信
game_state = game.GameState()
#获得第一个状态并将图像进行预处理
do_nothing = np.zeros(ACTIONS)
do_nothing[0]=1
#与游戏交互一次
x_t, r_0, terminal = game_state.frame_step(do_nothing)
x_t = cv2.cvtColor(cv2.resize(x_t, (80,80)),cv2.COLOR_BGR2GRAY)
ret, x_t = cv2.threshold(x_t,1,255,cv2.THRESH_BINARY)
s_t = np.stack((x_t, x_t, x_t, x_t),axis=2)
#开始训练
epsilon = INITIAL_EPSILON
t= 0
while "flappy bird"!="angry bird":
a_t = self.epsilon_greedy(s_t,epsilon=epsilon)
#epsilon递减
if epsilon > FINAL_EPSILON and t>OBSERVE:
epsilon -= (INITIAL_EPSILON-FINAL_EPSILON)/EXPLORE
#运动动作，与游戏环境交互一次
x_t1_colored, r_t,terminal = game_state.frame_step(a_t)
x_t1 = cv2.cvtColor(cv2.resize(x_t1_colored, (80, 80)), cv2.COLOR_BGR2GRAY)
ret, x_t1 = cv2.threshold(x_t1, 1, 255, cv2.THRESH_BINARY)
x_t1 =np.reshape(x_t1,(80,80,1))
s_t1 = np.append(x_t1, s_t[:,:,:3],axis=2)
#将数据存储到经验池中
experience = np.reshape(np.array([s_t,a_t,r_t,s_t1,terminal]),[1,5])
print("experience", r_t,terminal)
experience_buffer.add_experience(experience)
#在观察结束后进行训练
if t>OBSERVE:
#采集样本
train_s, train_a, train_r,train_s_,train_terminal = experience_buffer.sample(BATCH)
target_q=[]
read_target_Q = self.sess.run(self.Q_,{self.obs_:train_s_})
for i in range(len(train_r)):
if train_terminal[i]:
target_q.append(train_r[i])
else:
target_q.append(train_r[i]+GAMMA*np.max(read_target_Q[i]))
print(target_q)
#训练一次
self.sess.run(self.q_train_op, feed_dict={self.obs:train_s, self.action:train_a, self.Q_target:target_q})
#更新旧的目标网络
# if t%1000 == 0:
self.sess.run(self.update_oldq_op)
#往前推进一步
s_t = s_t1
t+=1
#每10000次迭代保存一次
if t%10000 == 0:
self.save_model('saved_networks/',global_step=t)
if t<=OBSERVE:
print("OBSERVE",t)
else:
if t%1 == 0:
print("train, steps",t,"/epsilon", epsilon,"/action_index",a_t, "/reward",r_t)
（10）最后，写一个主函数，对DQN进行训练。首先实例化一个经验池类buffer，声明一个深度值网络类brain，调用brain类的训练子函数对深度值网络进行训练。
if __name__=="__main__":
buffer = Experience_Buffer()
brain= Deep_Q_N()
brain.train_Network(buffer)总的代码和资料上传到资源中了。
