深度Q学习原理及相关实例

  • 8. 深度Q学习
    • 8.1 经验回放
    • 8.2 目标网络
    • 8.3 相关算法
    • 8.4 训练算法
    • 8.5 深度Q学习实例

  • 8.5.1 主程序
    • 程序注释
  • 8.5.2 DQN模型构建程序
    • 程序注释

    • 8.5.3 程序测试
  • 8.6 双重深度Q网络
  • 8.7 对偶深度Q网络


8. 深度Q学习


深度Q学习将深度学习和强化学习相结合,是第一个深度强化学习算法。深度Q学习的核心就是用一个人工神经网络q ( s , a ; θ ) , s ∈ S , a ∈ A q(s,a;\theta),s∈\mathcal{S},a∈\mathcal{A}q(s,a;θ),s∈S,a∈A来代替动作价值函数。其中θ \thetaθ为神经网络权重,在前面文章中,也使用过w \text{w}w。由于神经网络具有强大的表达能力,能够自动寻找特征,所以采用神经网络有潜力比传统人工特征强大得多。最近基于深度Q网络的深度强化学习算法有了重大的进展,在目前学术界有非常大的影响力。当同时出现异策、自益和函数近似时,无法保证收敛性,会出现训练不稳定或训练困难等问题。针对出现的各种问题,研究人员主要从以下两方面进行了改进。

经验回放(experience replay):将经验(即历史的状态、动作、奖励等)存储起来,再在存储的经验中按一定的规则采样。
目标网络(target network):修改网络的更新方式,例如不把刚学习到的网络权重马上用于后续的自益过程。本节后续内容将从这两条主线出发,介绍基于深度Q网络的强化学习算法。

8.1 经验回放


V. Mnih 等在 2013 年 发 表 文 章 《Playing Atari with deep reinforcement leaming》,提出了基于经验回放的深度Q网络,标志着深度Q网络的诞生,也标志着深度强化学习的诞生1。

采用批处理的模式能够提供稳定性。经验回放就是一种让经验的概率分布变得稳定的技术,它能提高训练的稳定性。

经验回放主要有“存储”和 “采样回放”两大关键步骤。其相关算法在之后会介绍, 现在主要来看其特征。

存储:将轨迹以( S t , A t , R t + 1 , S t + 1 )  等形式存储起来;

采样回放:使用某种规则从存储的( S t , A t , R t + 1 , S t + 1 ) 中随机取出一条或多条经验。

经验回放有以下好处。

在训练Q网络时,可以消除数据的关联,使得数据更像是独立同分布的(独立同分布是很多有监督学习的证明条件)。这样可以减小参数更新的方差,加快收敛。

能够重复使用经验,对于数据获取困难的情况尤其有用。从存储的角度,经验回放可以分为集中式回放和分布式回放。

回放可以分为以下几种,

集中式回放:智能体在一个环境中运行,把经验统一存储在经验池中。

分布式回放:智能体的多份拷贝(worker)同时在多个环境中运行,并将经验统一存 储于经验池中。由于多个智能体拷贝同时生成经验,所以能够在使用更多资源的同

时更快地收集经验。从采样的角度,经验回放可以分为均匀回放和优先回放。

均匀回放:等概率从经验集中取经验,并且用取得的经验来更新最优价值函数。

优先回放(PrioritizedExperienceReplay, PER): 为经验池里的每个经验指定一个优

先级,在选取经验时更倾向于选择优先级高的经验。

T. Schaul等 于 2016年发表文章《Prioritized experience replay》,提出了优先回放。优先回放的基本思想是为经验池里的经验指定一个优先级,在选取经验时更倾向于选择优先

级高的经验。一般的做法是,如果某个经验(例如经验i ii)的优先级为p i  ,那么选取该经验的概率为

 
​    
 

经验值有许多不同的选取方法,最常见的选取方法有成比例优先和基于排序优先。

成比例优先(proportional priority):第i ii个经验的优先级为

 

其中δ i 是时序差分误差,ε \varepsilonε是预先选择的一个小正数,α 是正参数。

基于排序优先(rank-basedpriority):第i ii个经验的优先级为

 

其中rank i 是第i 个经验从大到小排序的排名, 排名从1开始。

经验回放也不是完全没有缺点。例如,它也会导致回合更新和多步学习算法无法使用。一般情况下,如果我们将经验回放用于Q学习,就规避了这个缺点。

8.2 目标网络


对于基于自益的Q学习,其回报的估计和动作价值的估计都和权重θ \thetaθ有关。当权重值变化时,回报的估计和动作价值的估计都会变化。在学习的过程中,动作价值试图追逐一个变化的回报,也容易出现不稳定的情况。可以使用之前介绍的半梯度下降的算法来解决这个问题。在半梯度下降中,在更新价值参数θ \thetaθ时,不对基于自益得到的回报估计U t 求梯度。其中一种阻止对U t 求梯度的方法就是将价值参数复制一份得到θ target , 在计算U t 时用θ target 目标计算。

基于这一方法,V. Mnih等 在 2015年发表了论文《Human-level control through deep reinforcement learning》提出了目标网络(target network) 这一概念。 目标网络是在原有的神经网络之外再搭建一份结构完全相同的网络。原先就有的神经网络称为评估网络( evaluation network)。在学习的过程中,使用目标网络来进行自益得到回报的评估值,作 为学习的目标。在权重更新的过程中,只更新评估网络的权重,而不更新目标网络的权重。这样,更新权重时针对的目标不会在每次迭代都变化,是一个固定的目标。在完成一定次数的更新后,再将评估网络的权重值赋给目标网络,进而进行下一批更新。这样,目标网络也能得到更新。由于在目标网络没有变化的一段时间内回报的估计是相对固定的,目标网络的引入增加了学习的稳定性。所以,目标网络目前已经成为深度Q学习的主流做法。

8.3 相关算法


现在我们考虑使用深度Q学习算法来训练智能体玩游戏2。

在每一个时间步骤中,智能体从游戏动作集A = 1 , . . . K中选择一个动作。该动作被传递给模拟器并修改其内部状态和游戏分数。在一般情况下,环境可能是随机的。仿真器的内部状态不被智能体观察到,相反,智能体观察到一个来自仿真器的图像x t ∈ R ,这是一个代表当前屏幕的像素值的向量。此外,它还会收到代表游戏分数变化的奖励 r t 。需要注意的是,一般情况下,游戏得分可能取决于之前的整个动作和观察序列;关于一个动作的反馈可能只有在经过数千次的时间步长之后才会收到。

由于智能体只能观察当前屏幕,任务是部分观察,许多模拟器状态在感知上是异构的(即不可能只从当前屏幕x t 中完全了解当前情况)。因此,动作和观察的序列s t = x 1 , a 1 ,x 2 , . . . , a t − 1 , x t 被输入到算法中,然后算法根据这些序列学习游戏策略。仿真器中的所有序列都被假定为在有限的时间步长内终止。这个形式化的过程产生了一个大而有限的马尔科夫决策过程(MDP),在这个过程中,每个序列都是一个独立的状态。因此,我们可以将标准的强化学习方法应用于MDP,只需将完整序列s t 作为时间t tt的状态表示即可。

智能体的任务是在模拟器中选择最佳的动作最大化未来的损失.我们做一个标准的假设,对未来的每一步回报采用一个折扣因子γ \gammaγ(γ \gammaγ从始至终设置为0.99),然后定义了在时间t tt上经过折扣后的回报R t = ∑ t ′ = t T γ t ′ − t r t ′  ,其中T TT为最终停止的时间步。我们定义最佳动作价值函数Q ∗ ( s , a ) 作为遵循任何策略所能获得的最大预期收益。在经过一些状态s ss和采取一些动作a aa后,Q ∗ ( s , a ) = max ⁡ π E [ R t ∣ s t = s , a t = a , π ] ,其中π \piπ作为在状态s ss采取的动作a aa的映射,即策略。最优行为价值函数遵循一个重要的恒等式,这个恒等式被称为贝尔曼方程(Bellman equation)。这基于以下直觉:如果状态s ′ 在下一个时间步的最优值Q ∗ ( s ′ , a ′ ) 对于所有可能的行动a ′ 都已知,那么最优策略就是选择使期望值r + γ Q ∗ ( s ′ , a ′ ) 最大化的行动a ′ :

许多强化学习算法背后的基本思想是通过使用贝尔曼方程作为迭代更新来估计动作价值函数,Q i + 1 ( s , a ) = E s ′ [ r + γ max ⁡ a ′ Q i ( s ′ , a ′ ) ∣ s , a ] 。这些价值迭代算法都收

敛于最优动作价值函数,当i → ∞ 时Q i → Q ∗  。在实践中,这种基本的方法是不切实际的,因为动作-价值函数是对每个状态分别估计的,没有任何泛化。相反,通常使用函数逼

近器来估计动作价值函数Q ( s , a ; θ ) ≈ Q ∗ ( s , a ) 。在强化学习中这是典型的线性函数逼近器,但是有时用非线性函数逼近器代替,如神经网络。我们把带有权值θ \thetaθ的神

经网络函数逼近器称为Q网络。Q网络可以通过在迭代i ii中调整参数θ i 来训练减少贝尔曼方程中的均方误差, 其中最佳目标值r + γ max ⁡ a ′ Q ∗ ( s ′ , a ′ ) 被替代为近似目标值y = r

+ γ max ⁡ a ′ Q ( s ′ , a ′ ; θ i − ) ,其使用先前的一些迭代中的参数θ i −。这就产生了一个损失函数L i ( θ i ) 的序列,它在每次迭代i ii时发生变化,

请注意,目标取决于网络权重;这与用于监督学习的目标不同,后者在学习开始前是固定的。在优化的每一个阶段,我们在优化第i 个损失函数L i ( θ i ) 时,保持上一次迭代的参

数θ i − 固定,从而产生一系列定义明确的优化问题。最后一项是目标的方差,它不依赖于我们当前优化的参数θ i ,因此可以忽略。将损失函数相对于权重进行微分,我们得出以

下梯度:

与其计算上述梯度中的全部期望值,不如通过随机梯度下降来优化损失函数,这通常是计算上的便利。在这个框架中,通过在每一个时间步长后更新权重,使用单样本替换期望值,并设置θ i − = θ i − 1 ,可以恢复熟悉的Q-learning算法。

需要注意的是,这个算法是无模型的:它直接使用仿真器的样本来解决强化学习任务,而不需要明确地估计奖赏和过渡动态P ( r , s ′ ∣ s , a ) .它也是off-policy:它学习贪婪的策略a = arg max ⁡ a ′ Q ( s , a ′ ; θ ) ,以确保充分探索状态空间。在实际工作中,行为分布往往由ε \-greedy策略选择,遵循概率1 − ε 贪婪策略,选择概率ε \varepsilonε的随机行动。

8.4 训练算法


训练深度Q-网络的完整算法在下图所示的算法1中提出。智能体根据基于Q表的ε \varepsilonε-贪婪策略选择和执行动作。 由于使用任意长度的历史作为神经网络的输入可能是困难的,Q函数因此工作在由上述函数ϕ \phiϕ产生的固定长度的历史表征上。该算法以两种方式修改了标准的在线Q-learning,使其适用于训练大型神经网络而不产生分歧。

在这里插入图片描述

首先,这里使用了经验回放,我们将智能体在每个时间步的经验e t = ( s t , a t , r t , s t + 1 ) 存储在一个数据集D t = e 1 , . . . , e t 中,将许多情节(其中一个情节的结束发生在达到终端状态时)汇集到重放存储器中。在算法的内循环过程中,我们对从存储样本池中随机抽取的经验样本( s , a , r , s ′ ) ∼ U ( D ) (s,a,r,s') 进行Q-learning更新,或称minibatch更新。这种方法比标准的在线Q-learning有几个优势。

第一,每一步的经验都有可能被用于许多权重更新,这使得数据效率更高。

第二,直接从连续的样本中学习是低效的,因为样本之间有很强的相关性;随机化样本可以打破这些相关性,从而降低更新的方差。

第三,在对策略进行学习时,当前的参数决定了参数训练的下一个数据样本。例如,如果最大化动作是向左移动,那么训练样本将以左手边的样本为主;如果最大化动作随后切换到右边,那么训练分布也将切换。

很容易看出,不需要的反馈循环可能会出现,参数可能会被卡在一个糟糕的局部最小值中,甚至是灾难性的偏离。通过使用经验重放,行为分布是对其以前的许多状态进行平均,平滑学习,避免参数的振荡或发散。需要注意的是,通过经验重放学习时,需要进行off-policy学习(因为我们当前的参数与用于生成样本的参数不同),这也是选择Q-learning的动机。

在实践中,算法只在重放存储器中存储最后的N NN个经验元组,并在执行更新时从D DD中随机均匀取样。这种方法在某些方面是有局限性的,因为内存缓冲区并不能区分重要的转折,而且由于内存大小N NN是有限的,所以总是用最近的转折来覆盖。同样,均匀采样对重放内存中的所有转折给予同等的重要性。

对在线Q-learning的第二个修改旨在进一步提高方法与神经网络的稳定性,就是在Q-learning更新中使用一个单独的网络来生成目标y j  ,即,建立目标网络。更准确的说,每一次C更新,我们都会克隆网络Q,得到一个目标网络Q ^,并使用Q ^ 来生成Q-learning目标y j  ,用于后续C更新Q。与标准的在线Q-learning相比,这种修改使得算法更加稳定,在标准的在线Q-learning中,增加Q ( s t , a t ) 的更新往往也会增加所有a 的Q ( s t + 1 , a ) ,因此也会增加目标y j ,可能会导致策略的振荡或分歧。使用较旧的参数集生成目标,在对Q进行更新和更新影响目标y j  之间增加了一个延迟,使得分歧或振荡的可能性大大降低。

将更新r + γ max ⁡ a ′ Q ( s ′ , a ′ ; θ i − ) − Q ( s , a ; , θ i ) 中的误差项约束为-1和1之间是很有帮助的.因为绝对值损失函数∣ x ∣ 对x的所有负值都有-1的导数,对x的所有正值都有1的导数,所以将平方误差剪裁为− 1 和1 之间相当于对( − 1 , 1 ) 区间外的误差使用绝对值损失函数.这种形式的误差剪裁进一步提高了算法的稳定性.

8.5 深度Q学习实例


在这个实例里我们采用"LunarLander-v2"环境。

在这里插入图片描述

LunarLander-v2 着陆台总是在坐标(0,0)处。坐标是状态向量的前两个数字。从屏幕顶部移动到着陆台并以零速度降落的奖励大约是100…140点。如果着陆器远离着陆台,就会失去奖励。如果着陆器坠毁或静止,则事件结束,获得额外的-100或+100分。每条腿的地面接触是+10。发射主引擎每格为-0.3分。解决了就是+200分。可以在起落架外降落。燃料是无限的,所以智能体可以学习飞行,然后在第一次尝试降落。有四个离散动作可供选择:什么都不做、发射左方位引擎、发射主引擎、发射右方位引擎。

8.5.1 主程序

这是我们的主程序, 在其中我们建立相关环境并调用了子函数来建立深度强化网络模型进行训练。神经网络框架采用pytorch, 神经网络部分简单的采用3层全连接神经网络。

import gym
import random
import torch
import numpy as np
from collections import deque
import matplotlib.pyplot as plt
from dqn_agent import Agent
import os


def dqn(n_episode=2000, max_t=1000, eps_start=1.0, eps_end=0.01, eps_decay=0.995, mode='train'):
    """
    Deep Q-Learning

    :param n_episode:maximum number of training episodes
    :param max_t:maximum number of timesteps per episode
    :param eps_start:starting value of epsilon, for epsilon-greedy action selection
    :param eps_end:minimum value of epsilon
    :param eps_decay:multiplicative factor (per episode) for decreasing epsilon
    :return: final score
    """
    scores = []
    scores_window = deque(maxlen=100)
    eps = eps_start
    if mode == 'train':
        for i_episode in range(1, n_episode+1):
            # 初始化状态
            state = env.reset()
            score = 0
            for t in range(max_t):
                action = agent.act(state, eps)
                next_state, reward, done, _ = env.step(action)
                agent.step(state, action, reward, next_state, done)
                state = next_state
                score += reward
                if done:
                    break
            scores_window.append(score)
            scores.append(score)
            eps = max(eps_end, eps_decay*eps)
            print('\rEpisode {}\t Average Score:{:.2f}'.format(i_episode, np.mean(scores_window)), end="")
            if i_episode % 100 == 0:
                print('\rEpisode {}\rAverage Score :{:.2f}'.format(i_episode, np.mean(scores_window)))
            if np.mean(scores_window) >= 200.0:
                print('\nEnvironment solved in {:d} episode! \t Average Score: {:.2f}'.format(i_episode, np.mean(scores_window)))
                torch.save(agent.qnetwork_local.state_dict(), 'checkpoint.pth')
                break
    else:
        # 训练一次
        state = env.reset()
        for j in range(200):
            action = agent.act(state, eps)
            print('state :{} action :{}'. format(state, action))
            env.render()
            next_state, reward, done, _ = env.step(action)
            print('next_state={}, reward={}, done={}'.format(next_state, reward, done))
            agent.step(state, action, reward, next_state, done)
            if done:
                break

    return scores


if __name__ == '__main__':
    os.environ['KMP_DUPLICATE_LIB_OK'] = 'True'
    env = gym.make('LunarLander-v2')
    env.seed(0)
    print('State shape: ', env.observation_space.shape)
    print('Number of actions: ', env.action_space.n)

    MODE = 'train'

    if MODE == 'debug':
        # 调试模式
        agent = Agent(state_size=8, action_size=4, seed=1,debug_mode=True)
        scores = dqn(mode='test')

    elif MODE == 'run':
        agent = Agent(state_size=8, action_size=4, seed=1)
        agent.qnetwork_local.load_state_dict(torch.load('checkpoint.pth'))

        # 以当前策略运行
        for i in range(3):
            state = env.reset()
            for j in range(200):
                action = agent.act(state)
                env.render()
                state, reward, done, _ = env.step(action)
                if done:
                    break

        env.close()
    else:
        # 训练模式
        agent = Agent(state_size=8, action_size=4, seed=1)
        scores = dqn()

        # plot the scores
        fig = plt.figure()
        ax = fig.add_subplot(111)
        plt.plot(np.arange(len(scores)), scores)
        plt.ylabel('Score')
        plt.xlabel('Episode #')
        plt.show()
        
程序注释
 MODE = 'train'

    if MODE == 'debug':
        # 调试模式
        agent = Agent(state_size=8, action_size=4, seed=1,debug_mode=True)
        scores = dqn(mode='test')

    elif MODE == 'run':
        agent = Agent(state_size=8, action_size=4, seed=1)
        agent.qnetwork_local.load_state_dict(torch.load('checkpoint.pth'))
        ...
    else:
        # 训练模式
        agent = Agent(state_size=8, action_size=4, seed=1)
        scores = dqn()

在这里提供了程序运行的三种模式,“debug”, “run”, "train"模式。debug模式是为了方便查看在程序运行过程中的各种参数,方便程序调试和后期更改而设置的。run模式是在模型训练完成后可以使用训练完成的神经网络来查看最终效果。 train模式即训练模式,没有太多相关数据输出。

8.5.2 DQN模型构建程序

这部分为模型构建子程序,包含了DQN最重要的算法。程序包含了3个类, 分别是class QNetwork, class Agentclass ReplayBuffer

import numpy as np
import random
from collections import namedtuple, deque

import torch
import torch.nn.functional as F
import torch.nn as nn
import torch.optim as optim

BUFFER_SIZE = int(1e4)  # 经验回放的缓冲区的大小
BATCH_SIZE = 64  # 最小训练批数量
GAMMA = 0.99  # 折扣率
TAU = 1e-3  # 用于目标函数的柔性策略更新
LR = 5e-4  # 学习率
UPDATE_EVERY = 4  # 更新网络的频率

device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")



class QNetwork(nn.Module):
    """Actor (Policy) Model."""

    def __init__(self, state_size, action_size, seed, fc1_units=64, fc2_units=64):
        """Initialize parameters and build model.
        Params
        ======
            state_size (int): Dimension of each state
            action_size (int): Dimension of each action
            seed (int): Random seed
        """
        super(QNetwork, self).__init__()
        self.seed = torch.manual_seed(seed)
        self.fc1 = nn.Linear(state_size, fc1_units)
        self.fc2 = nn.Linear(fc1_units, fc2_units)
        self.fc3 = nn.Linear(fc2_units, action_size)

    def forward(self, state):
        """Build a network that maps state -> action values."""
        x = F.relu(self.fc1(state))
        x = F.relu(self.fc2(x))
        return self.fc3(x)


class Agent():
    """与环境相互作用,从环境中学习。"""

    def __init__(self, state_size, action_size, seed, debug_mode=False):
        """初始化智能体对象。
        
        Params
        ======
            state_size (int): dimension of each state
            action_size (int): dimension of each action
            seed (int): random seed
        """
        self.state_size = state_size
        self.action_size = action_size
        self.seed = random.seed(seed)
        self.debug_mode = debug_mode
        print('Program running in {}'.format(device))

        # Q-Network
        self.qnetwork_local = QNetwork(state_size, action_size, seed).to(device)
        self.qnetwork_target = QNetwork(state_size, action_size, seed).to(device)
        self.optimizer = optim.Adam(self.qnetwork_local.parameters(), lr=LR)  # 自适应梯度算法
        # print('Q-Network_local:{}\nQ-Network_target:{}'.format(self.qnetwork_local, self.qnetwork_target))

        # 经验回放
        if self.debug_mode is True:
            self.memory = ReplayBuffer(action_size, BUFFER_SIZE, 1, seed)
        else:
            self.memory = ReplayBuffer(action_size, BUFFER_SIZE, BATCH_SIZE, seed)
        # 初始化时间步 (for updating every UPDATE_EVERY steps)
        self.t_step = 0

    def step(self, state, action, reward, next_state, done):
        # 在经验回放中保存经验
        self.memory.add(state, action, reward, next_state, done)

        # 在每个时间步UPDATE_EVERY中学习
        self.t_step = (self.t_step + 1) % UPDATE_EVERY
        if self.t_step == 0:
            # 如果内存中有足够的样本,取随机子集进行学习
            if len(self.memory) > BATCH_SIZE:
                experiences = self.memory.sample()
                self.learn(experiences, GAMMA)
        if self.debug_mode is True:
            experiences = self.memory.sample()
            self.learn(experiences, GAMMA)

    def act(self, state, eps=0.):
        """根据当前策略返回给定状态的操作.
        
        Params
        ======
            state (array_like): 当前的状态
            eps (float): epsilon, 用于 epsilon-greedy action selection
        """
        state = torch.from_numpy(state).float().unsqueeze(0).to(device)
        # 将qn更改成评估形式
        self.qnetwork_local.eval()
        # 禁用梯度
        with torch.no_grad():
            # 获得动作价值
            action_values = self.qnetwork_local(state)
        # 将qn更改成训练模式
        self.qnetwork_local.train()

        # Epsilon-greedy action selection
        if random.random() > eps:
            return np.argmax(action_values.cpu().data.numpy())
        else:
            return random.choice(np.arange(self.action_size))

    def learn(self, experiences, gamma):
        """使用给定的一批经验元组更新值参数。

        Params
        ======
            experiences (Tuple[torch.Tensor]): tuple of (s, a, r, s', done) tuples 
            gamma (float): discount factor
        """
        states, actions, rewards, next_states, dones = experiences
        if self.debug_mode is True:
            print('\nstates={}, actions={}, rewards={}, next_states={}, dones={}'
              .format(states, actions, rewards, next_states, dones))

        # compute and minimize the loss
        # 从目标网络得到最大的预测Q值(下一个状态)
        Q_targets_next = self.qnetwork_target(next_states).detach().max(1)[0].unsqueeze(1)
        # 计算当前状态的Q目标
        Q_targets = rewards + (gamma * Q_targets_next * (1 - dones))
        # 从评估网络中获得期望的Q值
        Q_expected = self.qnetwork_local(states).gather(1, actions)
        if self.debug_mode is True:
            print('Q_target_next={}, \nQ_target ={}, \nQ_expected={},'
                  .format(Q_targets_next, Q_targets, Q_expected))

        # Compute loss
        loss = F.mse_loss(Q_expected, Q_targets)
        # Minimize the loss
        self.optimizer.zero_grad()
        loss.backward()
        # 执行单个优化步骤
        self.optimizer.step()

        # ------------------- update target network ------------------- #
        self.soft_update(self.qnetwork_local, self.qnetwork_target, TAU)

    def soft_update(self, local_model, target_model, tau):
        """:柔性更新模型参数。
        θ_target = τ*θ_local + (1 - τ)*θ_target

        Params
        ======
            local_model (PyTorch model): weights will be copied from
            target_model (PyTorch model): weights will be copied to
            tau (float): 插值参数
        """
        for target_param, local_param in zip(target_model.parameters(), local_model.parameters()):
            # 柔性更新, 将src中数据复制到self中
            target_param.data.copy_(tau * local_param.data + (1.0 - tau) * target_param.data)


class ReplayBuffer:
    """Fixed-size buffer to store experience tuples."""

    def __init__(self, action_size, buffer_size, batch_size, seed):
        """Initialize a ReplayBuffer object.

        Params
        ======
            action_size (int): dimension of each action
            buffer_size (int): maximum size of buffer
            batch_size (int): size of each training batch
            seed (int): random seed
        """
        self.action_size = action_size
        self.memory = deque(maxlen=buffer_size)
        self.batch_size = batch_size
        self.experience = namedtuple("Experience", field_names=["state", "action", "reward", "next_state", "done"])
        self.seed = random.seed(seed)

    def add(self, state, action, reward, next_state, done):
        """在memory中添加一段新的经验."""
        e = self.experience(state, action, reward, next_state, done)
        self.memory.append(e)

    def sample(self):
        """从memory中随机抽取一批经验."""
        experiences = random.sample(self.memory, k=self.batch_size)

        states = torch.from_numpy(np.vstack([e.state for e in experiences if e is not None])).float().to(device)
        actions = torch.from_numpy(np.vstack([e.action for e in experiences if e is not None])).long().to(device)
        rewards = torch.from_numpy(np.vstack([e.reward for e in experiences if e is not None])).float().to(device)
        next_states = torch.from_numpy(np.vstack([e.next_state for e in experiences if e is not None])).float().to(
            device)
        dones = torch.from_numpy(np.vstack([e.done for e in experiences if e is not None]).astype(np.uint8)).float().to(
            device)

        return (states, actions, rewards, next_states, dones)

    def __len__(self):
        """Return the current size of internal memory."""
        return len(self.memory)


程序注释

class QNetwork类构建了三层的神经网络模型,class ReplayBuffer类定义了关于经验回访的相关功能。class Agent是最重要的类,它调用了class QNetwork和class ReplayBuffer来创建DQN模型。所以我们主要看一下class Agent的相关函数和功能。

 self.state_size = state_size
        self.action_size = action_size
        self.seed = random.seed(seed)
        self.debug_mode = debug_mode

        # Q-Network
        self.qnetwork_local = QNetwork(state_size, action_size, seed).to(device)
        self.qnetwork_target = QNetwork(state_size, action_size, seed).to(device)
        self.optimizer = optim.Adam(self.qnetwork_local.parameters(), lr=LR)  # 自适应梯度算法

初始化智能体对象,并构建神经网络。在这里我们需要建立两个神经网络,其中“qnetwork_local”作为训练使用的神经网络,在此之外,我们还要建立qnetwork_target目标网络,来优化我们的训练过程。在这里使用了“自适应梯度算法”来作为神经网络的优化器。

 # 经验回放
        if self.debug_mode is True:
            self.memory = ReplayBuffer(action_size, BUFFER_SIZE, 1, seed)
        else:
            self.memory = ReplayBuffer(action_size, BUFFER_SIZE, BATCH_SIZE, seed)

根据相关的模式来建立经验回放功能的类。

	def step(self, state, action, reward, next_state, done):
        ...

    def act(self, state, eps=0.):
        ...

    def learn(self, experiences, gamma):
    	...

    def soft_update(self, local_model, target_model, tau):

这些是在训练过程中使用到的函数,它们的功能如下所示。其作用是方便与理解,其关系并不是完全如图所示的流线型关系。例如,soft_update函数是在learn函数中调用的的一个函数,其关系并不算是线性的。

8.5.3 程序测试

接下来将模式设置为Mode = train运行程序进行训练,要实现平均分数大于200分的目标,我的电脑需要跑40分钟左右。使用run模式运行模型如下,

在这里插入图片描述

8.6 双重深度Q网络
之前曾提到Q学习会带来最大化偏差,而双重Q学习却可以消除最大化偏差。基于查找表的双重Q学习引入了两个动作价值的估计Q ( 0 ) Q(0)Q(0)和Q ( 1 ) Q(1)Q(1),每次更新动作价值时用其中的一个网络确定动作,用确定的动作和另外一个网络来估计回报。对于深度Q学习也有同样的结论。Deepmind于2015年发表论文《Deepreinforcement learning with double Q-learning》,将双重Q学习用于深度Q网络,得到了双重深度Q网络(Double Deep Q Network,Double DQN)。考虑到深度Q网络已经有了评估网络和目标网络两个网络,所以双重深度Q学习在估计回报时只需要用评估网络确定动作,用目标网络确定回报的估计即可。所以,只需要将