TensorFlow 2 和 Keras 高级深度学习:6~10(4)

简介: TensorFlow 2 和 Keras 高级深度学习:6~10(4)

TensorFlow 2 和 Keras 高级深度学习:6~10(3)https://developer.aliyun.com/article/1426952

3. Q 学习实例

为了说明 Q 学习算法,我们需要考虑一个简单的确定性环境,如图“图 9.3.1”所示。 环境具有六个状态。

显示允许的过渡的奖励。 在两种情况下,奖励是非零的。 转换为目标G)状态可获得 +100 的奖励,同时移至H)状态具有 -100 奖励。 这两个状态是终端状态,从开始状态构成一个剧集的结尾:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4I0G4ROO-1681704311677)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/B14853_09_02.png)]

图 9.3.1:简单确定性世界中的奖励

为了使每个状态的身份正式化,我们使用(行, 列)标识符,如图“图 9.3.2”所示。 由于智能体尚未了解有关其环境的任何信息,因此“图 9.3.2”中所示的 Q 表的初始值为零。 在此示例中,折扣因子γ = 0.9。 回想一下,在当前 Q 值的估计中,折扣因子确定了未来 Q 值的权重,该权重是步数γ^k的函数。 在“公式 9.2.3”中,我们仅考虑近期 Q 值k = 1

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-zA9USIJI-1681704311677)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/B14853_09_03.png)]

图 9.3.2:简单确定性环境中的状态和智能体的初始 Q 表

最初,智能体采用的策略是 90% 的时间选择随机操作,并 10% 的时间使用 Q 表。 假设第一个动作是随机选择的,并且指示向右移动。“图 9.3.3”说明了向右移动时状态(0, 0)的新 Q 值的计算。 下一个状态是(0, 1)。 奖励为 0,所有下一个状态的 Q 值的最大值为零。 因此,向右移动的状态(0, 0)的 Q 值保持为 0。

为了轻松跟踪初始状态和下一个状态,我们在环境和 Q 表上使用不同的灰色阴影-初始状态浅灰色,下一个状态灰色。

在为下一个状态选择下一个动作时,候选动作位于较粗的边框中:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-U7jqNt18-1681704311678)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/B14853_09_04.png)]

图 9.3.3:假设智能体采取的行动是向右移动,则显示状态(0, 0)的 Q 值的更新

假设下一个随机选择的动作是向下移动。“图 9.3.4”显示状态(0, 1)的 Q 值沿向下方向的移动没有变化:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-VAzEl4n3-1681704311678)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/B14853_09_05.png)]

图 9.3.4:假设智能体选择的动作是向下移动,则显示状态(0, 1)的 Q 值的更新

在“图 9.3.5”中,智能体的第三个随机动作是向右移动。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-YhiSUiyU-1681704311678)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/B14853_09_06.png)]

图 9.3.5:假设智能体选择的动作是向右移动,则显示状态(1, 1)的 Q 值的更新

它遇到了,H状态,并获得了 -100 奖励。 这次,更新不为零。 向右移动时,状态(1, 1)的新 Q 值为 -100。 注意,由于这是终端状态,因此没有下一个状态。 一集刚刚结束,智能体返回到开始状态。

假设智能体仍处于探索模式,如图“图 9.3.6”所示:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-zswzSOpr-1681704311678)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/B14853_09_07.png)]

图 9.3.6:假设智能体选择的动作是向右连续两次移动,则显示状态(0, 1)的 Q 值的更新

为第二集采取的第一步是向右移动。 正如预期的那样,更新为 0。但是,它选择的第二个随机动作也是向右移动。 智能体到达G状态并获得 +100 的巨额奖励。 向右移动的状态(0, 1)的 Q 值变为 100。完成第二集,并且智能体返回到启动状态。

在第三集开始时,智能体采取的随机行动是向右移动。 现在,状态(0, 0)的 Q 值将更新为非零值,因为下一个状态的可能动作将最大 Q 值设为 100。“图 9.3.7”显示了所涉及的计算。 下一个状态(0, 1)的 Q 值波动回到较早的状态(0, 0)。 这就像对帮助找到G状态的早期状态表示赞赏。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-fhLzt2z4-1681704311678)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/B14853_09_08.png)]

图 9.3.7:假设智能体选择的动作是向右移动,则显示状态(0, 0)的 Q 值的更新

Q 表的进步很大。 实际上,在下一集中,如果由于某种原因该策略决定使用 Q 表而不是随机探索环境,则第一个动作是根据“图 9.3.8”中的计算向右移动。 在 Q 表的第一行中,导致最大 Q 值的动作是向右移动。 对于下一个状态(0, 1),Q 表的第二行表明下一个动作仍然是向右移动。 智能体已成功实现其目标。 该策略指导智能体采取了正确的措施来实现其目标:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-3ohfEpte-1681704311679)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/B14853_09_09.png)]

图 9.3.8:在这种情况下,智能体的策略决定利用 Q 表来确定状态(0, 0)(0, 1)的动作。 Q 表建议两个状态都向右移动

如果 Q 学习算法继续无限期运行,则 Q 表将收敛。 收敛的假设是 RL 问题必须是具有有限奖励的确定性 MDP,并且所有状态都将被无限次地访问。

在下一节中,我们将使用 Python 模拟环境。 我们还将展示 Q 学习算法的代码实现。

用 Python 进行 Q 学习

上一节中讨论的环境和 Q 学习可以在 Python 中实现。 由于该策略只是一个简单的表,因此在此时,无需使用tf.keras库。“列表 9.3.1”显示了q-learning-9.3.1.py,它是使用QWorld类实现的简单确定性世界(环境,智能体,操作和 Q 表算法)的实现。 为简洁起见,未显示处理用户界面的函数。

在此示例中,环境动态由self.transition_table表示。 在每个动作中,self.transition_table确定下一个状态。 执行动作的奖励存储在self.reward_table中。 每次通过step()函数执行动作时,都要查阅这两个表。 Q 学习算法由update_q_table()函数实现。 每当智能体需要决定要采取的操作时,它都会调用act()函数。 策略可以使用 Q 表随机抽取或决定。 所选动作是随机的机会百分比存储在self.epsilon变量中,该变量由update_epsilon()函数使用固定的epsilon_decay更新。

在执行“列表 9.3.1”中的代码之前,我们需要运行:

sudo pip3 install termcolor

安装termcolor包。 该包有助于可视化终端上的文本输出。

完整的代码可以在 GitHub 上找到

“列表 9.3.1”:q-learning-9.3.1.py

具有六个状态的简单确定性 MDP:

from collections import deque
import numpy as np
import argparse
import os
import time
from termcolor import colored
class QWorld:
    def __init__(self):
        """Simulated deterministic world made of 6 states.
        Q-Learning by Bellman Equation. 
        """
        # 4 actions
        # 0 - Left, 1 - Down, 2 - Right, 3 - Up
        self.col = 4
# 6 states
        self.row = 6
# setup the environment
        self.q_table = np.zeros([self.row, self.col])
        self.init_transition_table()
        self.init_reward_table()
# discount factor
        self.gamma = 0.9
# 90% exploration, 10% exploitation
        self.epsilon = 0.9
        # exploration decays by this factor every episode
        self.epsilon_decay = 0.9
        # in the long run, 10% exploration, 90% exploitation
        self.epsilon_min = 0.1
# reset the environment
        self.reset()
        self.is_explore = True
def reset(self):
        """start of episode"""
        self.state = 0
        return self.state
def is_in_win_state(self):
        """agent wins when the goal is reached"""
        return self.state == 2
def init_reward_table(self):
        """
        0 - Left, 1 - Down, 2 - Right, 3 - Up
        ----------------
        | 0 | 0 | 100  |
        ----------------
        | 0 | 0 | -100 |
        ----------------
        """
        self.reward_table = np.zeros([self.row, self.col])
        self.reward_table[1, 2] = 100.
        self.reward_table[4, 2] = -100.
def init_transition_table(self):
        """
        0 - Left, 1 - Down, 2 - Right, 3 - Up
        -------------
        | 0 | 1 | 2 |
        -------------
        | 3 | 4 | 5 |
        -------------
        """
        self.transition_table = np.zeros([self.row, self.col],
                                         dtype=int)
        self.transition_table[0, 0] = 0
        self.transition_table[0, 1] = 3
        self.transition_table[0, 2] = 1
        self.transition_table[0, 3] = 0
self.transition_table[1, 0] = 0
        self.transition_table[1, 1] = 4
        self.transition_table[1, 2] = 2
        self.transition_table[1, 3] = 1
# terminal Goal state
        self.transition_table[2, 0] = 2
        self.transition_table[2, 1] = 2
        self.transition_table[2, 2] = 2
        self.transition_table[2, 3] = 2
self.transition_table[3, 0] = 3
        self.transition_table[3, 1] = 3
        self.transition_table[3, 2] = 4
        self.transition_table[3, 3] = 0
self.transition_table[4, 0] = 3
        self.transition_table[4, 1] = 4
        self.transition_table[4, 2] = 5
        self.transition_table[4, 3] = 1
# terminal Hole state
        self.transition_table[5, 0] = 5
        self.transition_table[5, 1] = 5
        self.transition_table[5, 2] = 5
        self.transition_table[5, 3] = 5
def step(self, action):
        """execute the action on the environment
        Argument:
            action (tensor): An action in Action space
        Returns:
            next_state (tensor): next env state
            reward (float): reward received by the agent
            done (Bool): whether the terminal state 
                is reached
        """
        # determine the next_state given state and action
        next_state = self.transition_table[self.state, action]
        # done is True if next_state is Goal or Hole
        done = next_state == 2 or next_state == 5
        # reward given the state and action
        reward = self.reward_table[self.state, action]
        # the enviroment is now in new state
        self.state = next_state
        return next_state, reward, done
def act(self):
        """determine the next action
            either fr Q Table(exploitation) or
            random(exploration)
        Return:
            action (tensor): action that the agent
                must execute
        """
        # 0 - Left, 1 - Down, 2 - Right, 3 - Up
        # action is from exploration
        if np.random.rand() <= self.epsilon:
            # explore - do random action
            self.is_explore = True
            return np.random.choice(4,1)[0]
# or action is from exploitation
        # exploit - choose action with max Q-value
        self.is_explore = False
        action = np.argmax(self.q_table[self.state])
        return action
def update_q_table(self, state, action, reward, next_state):
        """Q-Learning - update the Q Table using Q(s, a)
        Arguments:
            state (tensor) : agent state
            action (tensor): action executed by the agent
            reward (float): reward after executing action 
                for a given state
            next_state (tensor): next state after executing
                action for a given state
        """
        # Q(s, a) = reward + gamma * max_a' Q(s', a')
        q_value = self.gamma * np.amax(self.q_table[next_state])
        q_value += reward
        self.q_table[state, action] = q_value
def update_epsilon(self):
        """update Exploration-Exploitation mix"""
        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay

感知动作学习循环在“列表 9.3.2”中进行了说明。 在每个剧集中,环境都会重置为开始状态。 选择要执行的动作并将其应用于环境。 观察奖励下一个状态,并将其用于更新 Q 表。 达到目标状态后,剧集完成(done = True)。

对于此示例,Q 学习运行 100 集或 10 获胜,以先到者为准。 由于在每个剧集中变量的值均降低,因此智能体开始倾向于利用 Q 表来确定在给定状态下要执行的动作。 要查看 Q 学习模拟,我们只需要运行以下命令:

python3 q-learning-9.3.1.py

“列表 9.3.2”:q-learning-9.3.1.py

主要的 Q 学习循环:

# state, action, reward, next state iteration
    for episode in range(episode_count):
        state = q_world.reset()
        done = False
        print_episode(episode, delay=delay)
        while not done:
            action = q_world.act()
            next_state, reward, done = q_world.step(action)
            q_world.update_q_table(state, action, reward, next_state)
            print_status(q_world, done, step, delay=delay)
            state = next_state
            # if episode is done, perform housekeeping
            if done:
                if q_world.is_in_win_state():
                    wins += 1
                    scores.append(step)
                    if wins > maxwins:
                        print(scores)
                        exit(0)
                # Exploration-Exploitation is updated every episode
                q_world.update_epsilon()
                step = 1
            else:
                step += 1

“图 9.3.9”显示了maxwins = 2000(达到2000 x目标状态)和delay = 0时的屏幕截图。 要仅查看最终的 Q 表,请执行:

python3 q-learning-9.3.1.py --train

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-cZSoTcOI-1681704311679)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/B14853_09_10.png)]

图 9.3.9:屏幕快照显示智能体在 2,000 次获胜后的 Q 表

Q 表已收敛,并显示了智能体可以在给定状态下采取的逻辑操作。 例如,在第一行或状态(0, 0)中,该策略建议向右移动。 第二行的状态(0, 1)也是如此。 第二个动作达到目标状态。 scores变量转储显示,随着智能体从策略获取正确的操作,所采取的最少步骤数减少了。

从“图 9.3.9”,我们可以从“公式 9.2.2”和V*(s) = max[a] Q(s, a)计算每个状态的值。 例如,对于状态(0, 0)V*(s) = max[a](0.0, 72.9, 90.0, 81.0) = 9.0

“图 9.3.10”显示每种状态的值。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6c4nqWBD-1681704311679)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/B14853_09_11.png)]

图 9.3.10:图 9.3.9 和公式 9.2.2 中每个状态的值

这个简单的示例说明了在简单确定性世界中智能体的 Q 学习的所有元素。 在下一节中,我们将介绍考虑随机性所需的轻微修改。

4. 非确定性环境

如果环境不确定,则奖励和行动都是概率性的。 新系统是随机的 MDP。 为了反映不确定性报酬,新的值函数为:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-xEwh4xsm-1681704311679)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/14853_09_042.png)] (Equation 9.4.1)

贝尔曼方程修改为:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-bA9ooaj2-1681704311679)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/14853_09_043.png)] (Equation 9.4.2)

但是,在本章中,我们将重点介绍确定性环境。 在下一节中,我们将提出一种更通用的 Q 学习算法,称为时差TD)学习。

5. 时差学习

Q 学习是更广义的 TD 学习TD(λ)的特例。 更具体地说,这是单步 TD 学习的特殊情况,TD(0)

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-jJKccuLk-1681704311680)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/14853_09_045.png)] (Equation 9.5.1)

其中α是学习率。 注意,当α = 1,“公式 9.5.1”与贝尔曼等式相似。 为简单起见,我们还将“公式 9.5.1”称为 Q 学习或广义 Q 学习。

以前,我们将 Q 学习称为一种非策略性 RL 算法,因为它学习 Q 值函数而没有直接使用它尝试优化的策略。 上策略一步式 TD 学习算法的示例是 SARSA,类似于“公式 9.5.1”:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-DEe9pXs0-1681704311680)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/14853_09_048.png)] (Equation 9.5.2)

主要区别是使用已优化的策略来确定a'。 必须知道项sars'a'(因此名称为 SARSA)才能在每次迭代时更新 Q 值函数。 Q 学习和 SARSA 都在 Q 值迭代中使用现有的估计,该过程称为自举。 在引导过程中,我们从奖励中更新当前的 Q 值估计,并随后更新 Q 值估计。

在提出另一个示例之前,似乎需要合适的 RL 模拟环境。 否则,我们只能对非常简单的问题(如上一个示例)运行 RL 模拟。 幸运的是,OpenAI 创建了 Gym,我们将在下一节中介绍。

在 OpenAI Gym 上进行 Q 学习

OpenAI Gym 是的工具包,用于开发和比较 RL 算法。 它适用于大多数 DL 库,包括tf.keras。 可以通过运行以下命令来安装健身房:

sudo pip3 install gym

该体育馆有多种可以测试 RL 算法的环境,例如玩具文字,经典控件,算法,Atari 和二维/三维机器人。 例如,FrozenLake-v0(“图 9.5.1”)是一个玩具文本环境,类似于在 Python Q 学习示例中使用的简单确定性世界:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-1Us8LbYq-1681704311680)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/B14853_09_12.png)]

图 9.5.1:OpenAI Gym 中的 FrozenLake-v0 环境

FrozenLake-v0具有 12 个状态,标记为S的状态为起始状态,F的状态为湖泊的冰冻部分,这是安全的,H为安全状态。 应当避免的空穴状态,G是飞盘所在的目标状态。 转换为目标状态的奖励为 +1。 对于所有其他状态,奖励为

FrozenLake-v0中,还有四个可用动作(左,下,右,上),称为动作空间。 但是,与之前的简单确定性世界不同,实际运动方向仅部分取决于所选的动作。 FrozenLake-v0环境有两种变体。 滑和不滑。 不出所料,滑动模式更具挑战性。

应用于FrozenLake-v0的操作将返回观察结果(等效于下一个状态),奖励,完成(无论剧集是否完成)以及调试信息字典。 返回的观察对象捕获环境的可观察属性,称为观察空间。

通用 Q 学习可以应用于FrozenLake-v0环境。“表 9.5.1”显示了湿滑和非湿滑环境的表现改进。 衡量策略表现的一种方法是执行的事件达到目标状态的百分比。 百分比越高,效果越好。 从大约 1.5% 的纯探查(随机操作)的基准来看,该策略可以在非光滑环境中达到约 76% 的目标状态,在光滑环境中可以达到约 71% 的目标状态。 不出所料,很难控制湿滑的环境。

模式 运行 大约百分比的目标
训练非滑动 python3 q-frozenlake-9.5.1.py 26
测试非滑动 python3 q-frozenlake-9.5.1.py -d 76
纯随机动作非滑动 python3 q-frozenlake-9.5.1.py -e 1.5
训练滑动 python3 q-frozenlake-9.5.1.py -s 26
测试滑动 python3 q-frozenlake-9.5.1.py -s -d 71
纯随机动作滑动 python3 q-frozenlake-9.5.1.py -s -e 1.5

表 9.5.1:在 FrozenLake-v0 环境中学习率为 0.5 的广义 Q 学习的基线和表现

由于该代码仅需要一个 Q 表,因此仍可以在 Python 和 NumPy 中实现。“列表 9.5.1”显示了QAgent类的实现。 除了使用 OpenAI Gym 的FrozenLake-v0环境之外,最重要的更改是广义 Q 学习的实现,这由update_q_table()函数中的“公式 9.5.1”定义。

“列表 9.5.1”:q-frozenlake-9.5.1.py

关于 FrozenLake-v0 环境的 Q 学习:

from collections import deque
import numpy as np
import argparse
import os
import time
import gym
from gym import wrappers, logger
class QAgent:
    def __init__(self,
                 observation_space,
                 action_space,
                 demo=False,
                 slippery=False,
                 episodes=40000):
        """Q-Learning agent on FrozenLake-v0 environment
Arguments:
            observation_space (tensor): state space
            action_space (tensor): action space
            demo (Bool): whether for demo or training
            slippery (Bool): 2 versions of FLv0 env
            episodes (int): number of episodes to train
        """
self.action_space = action_space
        # number of columns is equal to number of actions
        col = action_space.n
        # number of rows is equal to number of states
        row = observation_space.n
        # build Q Table with row x col dims
        self.q_table = np.zeros([row, col])
# discount factor
        self.gamma = 0.9
# initially 90% exploration, 10% exploitation
        self.epsilon = 0.9
        # iteratively applying decay til 
        # 10% exploration/90% exploitation
        self.epsilon_min = 0.1
        self.epsilon_decay = self.epsilon_min / self.epsilon
        self.epsilon_decay = self.epsilon_decay ** \
                             (1\. / float(episodes))
# learning rate of Q-Learning
        self.learning_rate = 0.1
# file where Q Table is saved on/restored fr
        if slippery:
            self.filename = 'q-frozenlake-slippery.npy'
        else:
            self.filename = 'q-frozenlake.npy'
# demo or train mode 
        self.demo = demo
        # if demo mode, no exploration
        if demo:
            self.epsilon = 0
def act(self, state, is_explore=False):
        """determine the next action
            if random, choose from random action space
            else use the Q Table
        Arguments:
            state (tensor): agent's current state
            is_explore (Bool): exploration mode or not
        Return:
            action (tensor): action that the agent
                must execute
        """
        # 0 - left, 1 - Down, 2 - Right, 3 - Up
        if is_explore or np.random.rand() < self.epsilon:
            # explore - do random action
            return self.action_space.sample()
# exploit - choose action with max Q-value
        action = np.argmax(self.q_table[state])
        return action
def update_q_table(self, state, action, reward, next_state):
        """TD(0) learning (generalized Q-Learning) with learning rate
        Arguments:
            state (tensor): environment state
            action (tensor): action executed by the agent for
                the given state
            reward (float): reward received by the agent for
                executing the action
            next_state (tensor): the environment next state
        """
        # Q(s, a) += 
        # alpha * (reward + gamma * max_a' Q(s', a') - Q(s, a))
        q_value = self.gamma * np.amax(self.q_table[next_state])
        q_value += reward
        q_value -= self.q_table[state, action]
        q_value *= self.learning_rate
        q_value += self.q_table[state, action]
        self.q_table[state, action] = q_value
def update_epsilon(self):
        """adjust epsilon"""
        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay

“列表 9.5.2”演示了智能体的感知行为学习循环。 在每个剧集中,通过调用env.reset()重置环境。 要执行的动作由agent.act()选择,并由env.step(action)应用于环境。 奖励和下一个状态将被观察并用于更新 Q 表。

在每个动作之后,通过agent.update_q_table()执行 TD 学习。 由于每次调用agent.update_epsilon()时处self.epsilon变量的值都会减少,该智能体开始支持利用 Q 表来确定在给定状态下执行的操作。 达到目标或空洞状态后,剧集完成(done = True)。 对于此示例,TD 学习运行 4,000 集。

“列表 9.5.2”:q-frozenlake-9.5.1.py

FrozenLake-v0环境的 Q 学习循环:

# loop for the specified number of episode
    for episode in range(episodes):
        state = env.reset()
        done = False
        while not done:
            # determine the agent's action given state
            action = agent.act(state, is_explore=args.explore)
            # get observable data
            next_state, reward, done, _ = env.step(action)
            # clear the screen before rendering the environment
            os.system('clear')
            # render the environment for human debugging
            env.render()
            # training of Q Table
            if done:
                # update exploration-exploitation ratio
                # reward > 0 only when Goal is reached
                # otherwise, it is a Hole
                if reward > 0:
                    wins += 1
if not args.demo:
                agent.update_q_table(state,
                                     action, 
                                     reward, 
                                     next_state)
                agent.update_epsilon()
state = next_state
            percent_wins = 100.0 * wins / (episode + 1)

agent对象可以在湿滑或非湿滑模式下运行。 训练后,智能体可以利用 Q 表选择给定任何策略执行的操作,如“表 9.5.1”的测试模式所示。 如“表 9.5.1”所示,使用学习的策略可显着提高性能。 随着体育馆的使用,不再需要中构建环境的许多代码行。 例如,与上一个示例不同,使用 OpenAI Gym,我们不需要创建状态转换表和奖励表。

这将帮助我们专注于构建有效的 RL 算法。 要以慢动作方式运行代码或每个动作延迟 1 秒,请执行以下操作:

python3 q-frozenlake-9.5.1.py -d -t=1

在本节中,我们在更具挑战性的环境中演示了 Q 学习。 我们还介绍了 OpenAI 体育馆。 但是,我们的环境仍然是玩具环境。 如果我们有大量的状态或动作怎么办? 在这种情况下,使用 Q 表不再可行。 在下一节中,我们将使用深度神经网络来学习 Q 表。

6. 深度 Q 网络(DQN)

在小型离散环境中,使用 Q 表执行 Q 学习是很好的选择。 但是,在大多数情况下,当环境具有许多状态或连续时,Q 表是不可行或不实际的。 例如,如果我们观察由四个连续变量组成的状态,则表的大小是无限的。 即使我们尝试将这四个变量离散化为 1,000 个值,表中的总行数也达到了惊人的1000^4 = 1e12。 即使经过训练,该表仍是稀疏的–该表中的大多数单元都是零。

这个问题的解决方案称为 DQN [2],它使用深度神经网络来近似 Q 表,如图“图 9.6.1”所示。 有两种构建 Q 网络的方法:

  • 输入是状态-动作对,预测是 Q 值
  • 输入是状态,预测是每个动作的 Q 值

第一种选择不是最佳的,因为网络被调用的次数等于操作数。 第二种是首选方法。 Q 网络仅被调用一次。

最希望得到的作用就是 Q 值最大的作用。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-1XO9ENsE-1681704311680)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/B14853_09_13.png)]

图 9.6.1:深度 Q 网络

训练 Q 网络所需的数据来自智能体的经验:(s[0]a[0]r[1]s[1], s[1]a[1]r[2]s[2],d ..., s[T-1]a[T-1]r[T]s[T])。 每个训练样本都是经验单元s[t]a[t]r[t+1]s[t+1]。 在时间步ts = s[t]的给定状态下,使用类似于前一部分的 Q 学习算法来确定动作a = a[t]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qTIUWymM-1681704311680)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/14853_09_060.png)] (Equation 9.6.1)

为了简化符号,我们省略了下标和粗体字母的使用。 注意,Q(s, a)是 Q 网络。 严格来说,它是Q(a | s),因为动作已移至预测阶段(换句话说,是输出),如“图 9.6.1”的右侧所示。 Q 值最高的动作是应用于环境以获得奖励r = r[t+1],下一状态s' = s[t+1]和布尔值done的动作,指示下一个状态是否为终端 。 根据关于广义 Q 学习的“公式 9.5.1”,可以通过应用所选的操作来确定 MSE 损失函数:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-LN8ly63Y-1681704311681)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/14853_09_065.png)] (Equation 9.6.2)

在前面有关 Q 学习和Q(a | s) -> Q(s, a)的讨论中,所有项都很熟悉。 项max[a'] Q(a' | s') -> max[a'] Q(s', a')。 换句话说,使用 Q 网络,在给定下一个状态的情况下预测每个动作的 Q 值,并从其中获得最大值。 注意,在终端状态下,s'max[a'] Q(a' | s') -> max[a'] Q(s', a') = 0

但是,事实证明训练 Q 网络是不稳定的。 导致不稳定的问题有两个:1)样本之间的相关性高; 2)非平稳目标。 高度相关性是由于采样经验的顺序性质。 DQN 通过创建经验缓冲解决了问题。 训练数据是从该缓冲区中随机采样的。 此过程称为经验回放

非固定目标的问题是由于目标网络Q(s', a')在每小批训练后都会被修改。 目标网络的微小变化会导致策略,数据分布以及当前 Q 值和目标 Q 值之间的相关性发生重大变化。 这可以通过冻结C训练步骤的目标网络的权重来解决。 换句话说,创建了两个相同的 Q 网络。 在每个C训练步骤中,从训练中的 Q 网络复制目标 Q 网络参数。

“算法 9.6.1”中概述了深度 Q 网络算法。

“算法 9.6.1”: DQN 算法

要求:将重播内存D初始化为容量N

要求:使用随机权重θ初始化动作值函数Q

要求:使用权重θ- = 0初始化目标操作值函数Q_target

需要:探索率ε和折扣系数γ

  1. 对于episode = 1, ..., M,执行:
  2. 给定初始状态s
  3. 对于step = 1, ..., T,执行:
  4. 选择动作
    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Fvaj6Eh3-1681704311681)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/14853_09_082.png)]
  5. 执行动作a,观察奖励r,以及下一个状态s'
  6. 将转换(s, a, r, s')存储在D
  7. 更新状态s = s'
  8. 经验回放
  9. D中抽样一小部分经验(s[j], a[j], r[j+1], s[j+1])
  10. [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-MLK0eigp-1681704311681)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/14853_09_090.png)]
  11. (Q_max - Q(s[j], a[j]; θ))²上相对于参数θ执行梯度下降步骤。
  12. 定期更新目标网络
  13. C个步骤,即Q_target = Q,换句话说,设置θ- = θ
  14. end
  15. end

“算法 9.6.1”总结了在具有离散动作空间和连续状态空间的环境上实现 Q 学习所需的所有技术。 在下一节中,我们将演示如何在更具挑战性的 OpenAI Gym 环境中使用 DQN。

Keras 中的 DQN

为了说明 DQN,使用了 OpenAI Gym 的CartPole-v0环境。 CartPole-v0是极点平衡问题。 目的是防止电杆跌落。 环境是二维的。 动作空间由两个离散的动作(左右移动)组成。 但是,状态空间是连续的,并且包含四个变量:

  • 直线位置
  • 线速度
  • 旋转角度
  • 角速度

CartPole-v0环境如图 9.6.1 所示:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ZqzLrler-1681704311681)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/B14853_09_14.png)]

图 9.6.1:CartPole-v0 环境

最初,杆是直立的。 杆保持直立的每个时间步长都提供 +1 的奖励。 当极点与垂直方向的夹角超过 15 度或与中心的距离超过 2.4 单位时,剧集结束。 如果在 100 个连续试验中平均奖励为 195.0,则认为CartPole-v0问题已解决:

“列表 9.6.1”向我们展示了CartPole-v0的 DQN 实现。 DQNAgent类表示使用 DQN 的智能体。 创建了两个 Q 网络:

  • “算法 9.6.1”中的 Q 网络或 Q
  • “算法 9.6.1”中的目标 Q 网络或Q_target

两个网络都是 MLP,每个都有 256 个单元的 3 个隐藏层。 这两个网络都是通过build_model()方法创建的。 在经验回放replay()期间训练 Q 网络。 以update_weights()的固定间隔C = 10个训练步骤,将 Q 网络参数复制到目标 Q 网络。 在“算法 9.6.1”中,这实现了第 13 行,Q_target = Q。 每次发作后,update_epsilon()都会降低探索利用的比例,以利用已学习的策略。

“列表 9.6.1”:dqn-cartpole-9.6.1.py

tf.keras中的 DQN:

class DQNAgent:
    def __init__(self,
                 state_space,
                 action_space,
                 episodes=500):
        """DQN Agent on CartPole-v0 environment
Arguments:
            state_space (tensor): state space
            action_space (tensor): action space
            episodes (int): number of episodes to train
        """
        self.action_space = action_space
# experience buffer
        self.memory = []
# discount rate
        self.gamma = 0.9
# initially 90% exploration, 10% exploitation
        self.epsilon = 1.0
        # iteratively applying decay til 
        # 10% exploration/90% exploitation
        self.epsilon_min = 0.1
        self.epsilon_decay = self.epsilon_min / self.epsilon
        self.epsilon_decay = self.epsilon_decay ** \
                             (1\. / float(episodes))
# Q Network weights filename
        self.weights_file = 'dqn_cartpole.h5'
        # Q Network for training
        n_inputs = state_space.shape[0]
        n_outputs = action_space.n
        self.q_model = self.build_model(n_inputs, n_outputs)
        self.q_model.compile(loss='mse', optimizer=Adam())
        # target Q Network
        self.target_q_model = self.build_model(n_inputs, n_outputs)
        # copy Q Network params to target Q Network
        self.update_weights()
self.replay_counter = 0
        self.ddqn = True if args.ddqn else False
def build_model(self, n_inputs, n_outputs):
        """Q Network is 256-256-256 MLP
Arguments:
            n_inputs (int): input dim
            n_outputs (int): output dim
Return:
            q_model (Model): DQN
        """
        inputs = Input(shape=(n_inputs, ), name='state')
        x = Dense(256, activation='relu')(inputs)
        x = Dense(256, activation='relu')(x)
        x = Dense(256, activation='relu')(x)
        x = Dense(n_outputs,
                  activation='linear',
                  name='action')(x)
        q_model = Model(inputs, x)
        q_model.summary()
        return q_model
def act(self, state):
        """eps-greedy policy
        Return:
            action (tensor): action to execute
        """
        if np.random.rand() < self.epsilon:
            # explore - do random action
            return self.action_space.sample()
# exploit
        q_values = self.q_model.predict(state)
        # select the action with max Q-value
        action = np.argmax(q_values[0])
        return action
def remember(self, state, action, reward, next_state, done):
        """store experiences in the replay buffer
        Arguments:
            state (tensor): env state
            action (tensor): agent action
            reward (float): reward received after executing
                action on state
            next_state (tensor): next state
        """
        item = (state, action, reward, next_state, done)
        self.memory.append(item)
def get_target_q_value(self, next_state, reward):
        """compute Q_max
           Use of target Q Network solves the 
            non-stationarity problem
        Arguments:
            reward (float): reward received after executing
                action on state
            next_state (tensor): next state
        Return:
            q_value (float): max Q-value computed by
                DQN or DDQN
        """
        # max Q value among next state's actions
        if self.ddqn:
            # DDQN
            # current Q Network selects the action
            # a'_max = argmax_a' Q(s', a')
            action = np.argmax(self.q_model.predict(next_state)[0])
            # target Q Network evaluates the action
            # Q_max = Q_target(s', a'_max)
            q_value = self.target_q_model.predict(\
                                          next_state)[0][action]
        else:
            # DQN chooses the max Q value among next actions
            # selection and evaluation of action is 
            # on the target Q Network
            # Q_max = max_a' Q_target(s', a')
            q_value = np.amax(\
                      self.target_q_model.predict(next_state)[0])
# Q_max = reward + gamma * Q_max
        q_value *= self.gamma
        q_value += reward
        return q_value
def replay(self, batch_size):
        """experience replay addresses the correlation issue 
            between samples
        Arguments:
            batch_size (int): replay buffer batch 
                sample size
        """
        # sars = state, action, reward, state' (next_state)
        sars_batch = random.sample(self.memory, batch_size)
        state_batch, q_values_batch = [], []
# fixme: for speedup, this could be done on the tensor level
        # but easier to understand using a loop
        for state, action, reward, next_state, done in sars_batch:
            # policy prediction for a given state
            q_values = self.q_model.predict(state)
# get Q_max
            q_value = self.get_target_q_value(next_state, reward)
# correction on the Q value for the action used
            q_values[0][action] = reward if done else q_value
# collect batch state-q_value mapping
            state_batch.append(state[0])
            q_values_batch.append(q_values[0])
# train the Q-network
        self.q_model.fit(np.array(state_batch),
                         np.array(q_values_batch),
                         batch_size=batch_size,
                         epochs=1,
                         verbose=0)
# update exploration-exploitation probability
        self.update_epsilon()
# copy new params on old target after 
        # every 10 training updates
        if self.replay_counter % 10 == 0:
            self.update_weights()
self.replay_counter += 1
def update_epsilon(self):
        """decrease the exploration, increase exploitation"""
        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay

为了在“算法 9.6.1”经验回放replay()中实现第 10 行,对于每个体验单元(s[j]a[j]r[j + 1]s[j + 1])将动作a[j]的 Q 值设置为Q_max。 所有其他动作的 Q 值保持不变。

这是通过 DQNAgent replay()函数中的以下行实现的:

# policy prediction for a given state q_values = self.q_model.predict(state)
# get Q_max
q_value = self.get_target_q_value(next_state)
# correction on the Q value for the action used q_values[0][action] = reward if done else q_value

如“算法 9.6.1”的第 11 行所示,只有动作a[j]具有等于(Q_max - Q(s[j], a[j]; θ))²的非零损失。 请注意,假设缓冲区中有足够的数据,换句话说,在每个剧集结束后,“列表 9.6.2”中的感知动作学习循环会调用经验回放。 缓冲区的大小大于或等于批量大小)。 在经验回放期间,会随机采样一批体验单元,并将其用于训练 Q 网络。

与 Q 表类似,act()实现了 ε-贪婪策略,“公式 9.6.1”。

体验由remember()存储在重播缓冲区中。 Q 通过get_target_q_value()函数计算。

“列表 9.6.2”总结了智能体的感知-行动-学习循环。 在每个剧集中,通过调用env.reset()重置环境。 要执行的动作由agent.act()选择,并由env.step(action)应用于环境。 奖励和下一状态将被观察并存储在重播缓冲区中。 在执行每个操作之后,智能体会调用replay()来训练 DQN 并调整探索利用比率。

当极点与垂直方向的夹角超过 15 度或与中心的距离超过 2.4 单位时,剧集完成(done = True)。 对于此示例,如果 DQN 智能体无法解决问题,则 Q 学习最多运行 3,000 集。 如果average mean_score奖励在 100 次连续试验win_trials中为 195.0,则认为CartPole-v0问题已解决。

“列表 9.6.2”:dqn-cartpole-9.6.1.py

tf.keras中的 DQN 训练循环:

# Q-Learning sampling and fitting
    for episode in range(episode_count):
        state = env.reset()
        state = np.reshape(state, [1, state_size])
        done = False
        total_reward = 0
        while not done:
            # in CartPole-v0, action=0 is left and action=1 is right
            action = agent.act(state)
            next_state, reward, done, _ = env.step(action)
            # in CartPole-v0:
            # state = [pos, vel, theta, angular speed]
            next_state = np.reshape(next_state, [1, state_size])
            # store every experience unit in replay buffer
            agent.remember(state, action, reward, next_state, done)
            state = next_state
            total_reward += reward
# call experience relay
        if len(agent.memory) >= batch_size:
            agent.replay(batch_size)
scores.append(total_reward)
        mean_score = np.mean(scores)
        if mean_score >= win_reward[args.env_id] \
                and episode >= win_trials:
            print("Solved in episode %d: \
                   Mean survival = %0.2lf in %d episodes"
                  % (episode, mean_score, win_trials))
            print("Epsilon: ", agent.epsilon)
            agent.save_weights()
            break
        if (episode + 1) % win_trials == 0:
            print("Episode %d: Mean survival = \
                   %0.2lf in %d episodes" %
                  ((episode + 1), mean_score, win_trials))

在平均 10 次运行的中,DQN 在 822 集内解决了。 我们需要注意的是,每次训练运行的结果可能会有所不同。

自从引入 DQN 以来,连续的论文都提出了对“算法 9.6.1”的改进。 一个很好的例子是双 DQN(DDQN),下面将对其进行讨论。

双重 Q 学习(DDQN)

在 DQN 中,目标 Q 网络选择并评估每个动作,从而导致 Q 值过高。 为了解决这个问题,DDQN [3]建议使用 Q 网络选择动作,并使用目标 Q 网络评估动作。

在 DQN 中,如“算法 9.6.1”所概述,第 10 行中 Q 值的估计为:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-FrJ5Y9SZ-1681704311682)(https://gitcode.net/apachecn/apachecn-dl-zh/-/raw/master/docs/adv-dl-tf2-keras/img/14853_09_097.png)]

  • Q_target选择并评估动作,a[j+1]

TensorFlow 2 和 Keras 高级深度学习:6~10(5)https://developer.aliyun.com/article/1426954


相关文章
|
3天前
|
机器学习/深度学习 API 语音技术
|
3月前
|
机器学习/深度学习 存储 人工智能
TensorFlow Lite,ML Kit 和 Flutter 移动深度学习:6~11(3)
TensorFlow Lite,ML Kit 和 Flutter 移动深度学习:6~11(3)
80 0
|
3月前
|
机器学习/深度学习 Dart TensorFlow
TensorFlow Lite,ML Kit 和 Flutter 移动深度学习:6~11(5)
TensorFlow Lite,ML Kit 和 Flutter 移动深度学习:6~11(5)
67 0
|
2月前
|
机器学习/深度学习 PyTorch TensorFlow
Python中的深度学习:TensorFlow与PyTorch的选择与使用
Python中的深度学习:TensorFlow与PyTorch的选择与使用
|
2月前
|
机器学习/深度学习 人工智能 API
人工智能应用工程师技能提升系列2、——TensorFlow2——keras高级API训练神经网络模型
人工智能应用工程师技能提升系列2、——TensorFlow2——keras高级API训练神经网络模型
30 0
|
2月前
|
机器学习/深度学习 数据可视化 TensorFlow
基于tensorflow深度学习的猫狗分类识别
基于tensorflow深度学习的猫狗分类识别
61 1
|
2月前
|
机器学习/深度学习 人工智能 算法
【深度学习】因果推断与机器学习的高级实践 | 数学建模
【深度学习】因果推断与机器学习的高级实践 | 数学建模
|
3月前
|
机器学习/深度学习 PyTorch TensorFlow
【TensorFlow】深度学习框架概述&TensorFlow环境配置
【1月更文挑战第26天】【TensorFlow】深度学习框架概述&TensorFlow环境配置
|
3月前
|
机器学习/深度学习 存储 编解码
TensorFlow Lite,ML Kit 和 Flutter 移动深度学习:6~11(4)
TensorFlow Lite,ML Kit 和 Flutter 移动深度学习:6~11(4)
119 0
|
3月前
|
机器学习/深度学习 人工智能 API
TensorFlow Lite,ML Kit 和 Flutter 移动深度学习:1~5
TensorFlow Lite,ML Kit 和 Flutter 移动深度学习:1~5
67 0