约搏牛牛大吃小:DQN(Deep Q-learning)入门教程(六)之DQN Play Flappy-bird ,MountainCar

2020-06-01 8 views 0

扫一扫用手机浏览

在DQN(Deep Q-learning)入门教程(四)之Q-learning Play Flappy Bird中,我们使用q-learning算法去对Flappy Bird举行强化学习,而在这篇博客中我们将使用神经网络模子来取代Q-table,关于DQN的先容,可以参考我前一篇博客:DQN(Deep Q-learning)入门教程(五)之DQN先容

在这篇博客中将使用DQN做如下操作:

  • Flappy Bird
  • MountainCar-v0

再回首一下DQN的算法流程:

项目地址:Github

MountainCar-v0

MountainCar的训练好的Gif示意图如下所示,汽车起始位置位于山的底部,最终目标是驶向右边山的插旗的地方,其中,汽车的引擎不能够直接驶向终点,必须借助左边的山体的重力加速率才能够驶向终点。

MountainCar-v0由OpenAI提供,python包为gym,官网网站为https://gym.openai.com/envs/MountainCar-v0/。在Gym包中,提供了许多可以用于强化学习的环境(env):

在MountainCar-v0中,状态有2个变量,car position(汽车的位置),car vel(汽车的速率),action一共有3种 Accelerate to the Left Don't accelerateAccelerate to the Right,然后当车到达旌旗的地方(position = 0.5)会获得\(reward = 1\)的奖励,若是没有到达则为\(-1\)。然则若是当你运行步骤跨越200次的时刻,游戏就会竣事。详情可以参考源代码(ps:官方文档中没有这些说明)。

下面先容一下gym中几个常用的函数:

  • env = gym.make("MountainCar-v0")
    

    这个就是建立一个MountainCar-v0的游戏环境。

  • state = env.reset()
    

    重置环境,返回重置后的state

  • env.render()
    

    将运行画面展示在屏幕上面,当我们在训练的时刻可以不使用这个来提升速率。

  • next_state, reward, done, _ = env.step(action)
    

    执行action动作,返回下一个状态,奖励,是否完成,info。

初始化Agent

初始化Agent直接使用代码说明吧,这个照样对照简单的:

import keras
import random
from collections import deque

import gym
import numpy as np
from keras.layers import Dense
from keras.models import Sequential


class Agent():
    def __init__(self, action_set, observation_space):
        """
        初始化
        :param action_set: 动作聚集
        :param observation_space: 环境属性,我们需要使用它获得state的shape
        """
        # 奖励衰减
        self.gamma = 1.0
        # 从履历池中取出数据的数目
        self.batch_size = 50
        # 履历池
        self.memory = deque(maxlen=2000000)
        # 探索率
        self.greedy = 1.0
        # 动作聚集
        self.action_set = action_set
        # 环境的属性
        self.observation_space = observation_space
        # 神经网路模子
        self.model = self.init_netWork()

    def init_netWork(self):
        """
        构建模子
        :return: 模子
        """
        model = Sequential()
        # self.observation_space.shape[0],state的变量的数目
        model.add(Dense(64 * 4, activation="tanh", input_dim=self.observation_space.shape[0]))
        model.add(Dense(64 * 4, activation="tanh"))
        # self.action_set.n 动作的数目
        model.add(Dense(self.action_set.n, activation="linear"))
        model.compile(loss=keras.losses.mean_squared_error,
                      optimizer=keras.optimizers.RMSprop(lr=0.001))
        return model

我们使用行列来保留履历,这样的话新的数据就会笼罩远古的数据。此时我们界说一个函数,专门用来将数据保留到履历池中,然后界说一个函数用来更新\(\epsilon\)探索率。

def add_memory(self, sample):
    self.memory.append(sample)

def update_greedy(self):
    # 小于最小探索率的时刻就不举行更新了。
    if self.greedy > 0.01:
        self.greedy *= 0.995

训练模子

首先先看代码:

def train_model(self):
    # 从履历池中随机选择部门数据
    train_sample = random.sample(self.memory, k=self.batch_size)

    train_states = []
    next_states = []
    for sample in train_sample:
        cur_state, action, r, next_state, done = sample
        next_states.append(next_state)
        train_states.append(cur_state)

    # 转成np数组
    next_states = np.array(next_states)
    train_states = np.array(train_states)
    # 获得next_state的q值
    next_states_q = self.model.predict(next_states)

    # 获得state的展望值
    state_q = self.model.predict_on_batch(train_states)

    # 盘算Q现实
    for index, sample in enumerate(train_sample):
        cur_state, action, r, next_state, done = sample
        if not done:
            state_q[index][action] = r + self.gamma * np.max(next_states_q[index])
        else:
            state_q[index][action] = r
    
    self.model.train_on_batch(train_states, state_q)

人人一定从上面的代码发现一些问题,使用了两个for循环,why?首先先说一下两个for循环划分的作用:

  • 第一个for循环:获得train_statesnext_states,其中next_states是为了盘算Q现实。
  • 第二个for循环:盘算Q现实

可能有人会有一个疑问,为什么我不写成一个for循环呢?实际上写成一个for循环是完全没有问题的,很,然则写成一个for循环意味着我们要多次挪用model.predict_on_batch,这样会花费一定的时间(亲自试验过,这样会对照慢),因此,我们写成了两个for循环,然后只需要挪用一次predict

执行动作与选择最佳动作

执行动作的代码如下所示:

def act(self, env, action):
    """
    执行动作
    :param env: 执行环境
    :param action: 执行的动作
    :return: ext_state, reward, done
    """
    next_state, reward, done, _ = env.step(action)

    if done:
        if reward < 0:
            reward = -100
        else:
            reward = 10
    else:
        if next_state[0] >= 0.4:
            reward += 1

    return next_state, reward, done

其中,我们可以修改奖励以加速网络收敛。

选择最好的动作的动作如下所示,会以一定的探索率随机选择动作。

def get_best_action(self, state):
    if random.random() < self.greedy:
        return self.action_set.sample()
    else:
        return np.argmax(self.model.predict(state.reshape(-1, 2)))

最先训练

关于详细的注释,在注释中已经详细的说明晰:

if __name__ == "__main__":
    # 训练次数
    episodes = 10000
    # 实例化游戏环境
    env = gym.make("MountainCar-v0")
    # 实例化Agent
    agent = Agent(env.action_space, env.observation_space)
    # 游戏中动作执行的次数(最大为200)
    counts = deque(maxlen=10)

    for episode in range(episodes):
        count = 0
        # 重置游戏
        state = env.reset()

        # 刚最先不立刻更新探索率
        if episode >= 5:
            agent.update_greedy()

        while True:
            count += 1
            # 获得最佳动作
            action = agent.get_best_action(state)
            next_state, reward, done = agent.act(env, action)
            agent.add_memory((state, action, reward, next_state, done))
            # 刚最先不立刻训练模子,先填充履历池
            if episode >= 5:
                agent.train_model()
            state = next_state
            if done:
                # 将执行的次数添加到counts中
                counts.append(count)
                print("在{}轮中,agent执行了{}次".format(episode + 1, count))

                # 若是近10次,动作执行的平均次数少于160,则保留模子并退出
                if len(counts) == 10 and np.mean(counts) < 160:
                    agent.model.save("car_model.h5")
                    exit(0)
                break

训练一定的次数后,我们就可以获得模子了。然后举行测试。

模子测试

测试的代码没什么好说的,如下所示:

import gym
from keras.models import load_model
import numpy as np
model = load_model("car_model.h5")

env = gym.make("MountainCar-v0")

for i in range(100):
    state = env.reset()
    count = 0
    while True:
        env.render()
        count += 1
        action = np.argmax(model.predict(state.reshape(-1, 2)))
        next_state, reward, done, _ = env.step(action)
        state = next_state
        if done:
            print("游戏的次数:", count)
            break

部门的效果如下:

Flappy Bird

FlappyBird的代码我就不过多赘述了,内里的一些函数先容可以参照这个来看:DQN(Deep Q-learning)入门教程(四)之Q-learning Play Flappy Bird,代码头脑与训练Mountain-Car基本是一致的。

import random
from collections import deque

import keras
import numpy as np
from keras.layers import Dense
from keras.models import Sequential
from ple import PLE
from ple.games import FlappyBird


class Agent():
    def __init__(self, action_set):
        self.gamma = 1
        self.model = self.init_netWork()
        self.batch_size = 128
        self.memory = deque(maxlen=2000000)
        self.greedy = 1
        self.action_set = action_set

    def get_state(self, state):
        """
        提取游戏state中我们需要的数据
        :param state: 游戏state
        :return: 返回提取好的数据
        """
        return_state = np.zeros((3,))
        dist_to_pipe_horz = state["next_pipe_dist_to_player"]
        dist_to_pipe_bottom = state["player_y"] - state["next_pipe_top_y"]
        velocity = state['player_vel']
        return_state[0] = dist_to_pipe_horz
        return_state[1] = dist_to_pipe_bottom
        return_state[2] = velocity
        return return_state

    def init_netWork(self):
        """
        构建模子
        :return:
        """
        model = Sequential()
        model.add(Dense(64 * 4, activation="tanh", input_shape=(3,)))
        model.add(Dense(64 * 4, activation="tanh"))
        model.add(Dense(2, activation="linear"))

        model.compile(loss=keras.losses.mean_squared_error,
                      optimizer=keras.optimizers.RMSprop(lr=0.001))
        return model

    def train_model(self):
        if len(self.memory) < 2500:
            return

        train_sample = random.sample(self.memory, k=self.batch_size)
        train_states = []
        next_states = []

        for sample in train_sample:
            cur_state, action, r, next_state, done = sample
            next_states.append(next_state)
            train_states.append(cur_state)
        # 转成np数组
        next_states = np.array(next_states)
        train_states = np.array(train_states)

        # 获得下一个state的q值
        next_states_q = self.model.predict(next_states)
        # 获得展望值
        state_q = self.model.predict_on_batch(train_states)

        for index, sample in enumerate(train_sample):
            cur_state, action, r, next_state, done = sample
            # 盘算Q现实
            if not done:
                state_q[index][action] = r + self.gamma * np.max(next_states_q[index])
            else:
                state_q[index][action] = r
        self.model.train_on_batch(train_states, state_q)

    def add_memory(self, sample):
        self.memory.append(sample)

    def update_greedy(self):
        if self.greedy > 0.01:
            self.greedy *= 0.995

    def get_best_action(self, state):
        if random.random() < self.greedy:
            return random.randint(0, 1)
        else:
            return np.argmax(self.model.predict(state.reshape(-1, 3)))

    def act(self, p, action):
        """
        执行动作
        :param p: 通过p来向游戏发出动作下令
        :param action: 动作
        :return: 奖励
        """
        r = p.act(self.action_set[action])
        if r == 0:
            r = 1
        if r == 1:
            r = 100
        else:
            r = -1000
        return r


if __name__ == "__main__":
    # 训练次数
    episodes = 20000
    # 实例化游戏工具
    game = FlappyBird()
    # 类似游戏的一个接口,可以为我们提供一些功效
    p = PLE(game, fps=30, display_screen=False)
    # 初始化
    p.init()
    # 实例化Agent,将动作集传进去
    agent = Agent(p.getActionSet())
    max_score = 0
    scores = deque(maxlen=10)

    for episode in range(episodes):
        # 重置游戏
        p.reset_game()
        # 获得状态
        state = agent.get_state(game.getGameState())
        if episode > 150:
            agent.update_greedy()
        while True:
            # 获得最佳动作
            action = agent.get_best_action(state)
            # 然后执行动作获得奖励
            reward = agent.act(p, action)
            # 获得执行动作之后的状态
            next_state = agent.get_state(game.getGameState())
            agent.add_memory((state, action, reward, next_state, p.game_over()))
            agent.train_model()
            state = next_state
            if p.game_over():
                # 获得当前分数
                current_score = p.score()
                max_score = max(max_score, current_score)
                scores.append(current_score)
                print('第%s次游戏,得分为: %s,最大得分为: %s' % (episode, current_score, max_score))
                if len(scores) == 10 and np.mean(scores) > 150:
                    agent.model.save("bird_model.h5")
                    exit(0)
                break

该部门相对照于Mountain-Car需要更长的时间,现在的我还没有训练出对照好的效果,停止写完这篇博客,最新的数据如下所示:

emm,我又不想让我的电脑一直开着,。

总结

上面的两个例子即是DQN最基本最基本的使用,我们还可以将上面的FlappyBird的问题稍微复杂化一点,比如说我们无法直接的知道环境的状态,我们则可以使用CNN网络去从游戏图片入手(关于这种做法,网络上有许多人写了相对应的博客)。

项目地址:Github

参考

  • openai-gym
  • MountainCar-v0 ——src code
  • DQN(Deep Q-learning)入门教程(四)之Q-learning Play Flappy Bird
,

欧博网址开户

www.aidshuhehaote.com欢迎进入欧博网址(Allbet Gaming),欧博网址开放会员注册、代理开户、电脑客户端下载、苹果安卓下载等业务。

Sunbet网站内容转载自互联网,如有侵权,联系Sunbet删除。

本文链接地址:http://www.zzxslszjx.com/post/1182.html

相关文章

发表评论