DQN（Deep Q-learning）入门教程（六）之DQN Play Flappy-bird ，MountainCar
在DQN（Deep Q-learning）入门教程（四）之Q-learning Play Flappy Bird中，咱们使用q-learning算法去对Flappy Bird进行强化学习，而在这篇博客中咱们将使用神经网络模型来代替Q-table，关于DQN的介绍，能够参考我前一篇博客：DQN（Deep Q-learning）入门教程（五）之DQN介绍html
在这篇博客中将使用DQN作以下操做：python
- Flappy Bird
- MountainCar-v0
再回顾一下DQN的算法流程：git
项目地址：Githubgithub
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 accelerate，Accelerate to the Right，而后当车达到旗帜的地方（position = 0.5）会获得\(reward = 1\)的奖励，若是没有达到则为\(-1\)。可是若是当你运行步骤超过200次的时候，游戏就会结束。详情能够参考源代码（ps：官方文档中没有这些说明）。网络
下面介绍一下gym中几个经常使用的函数：app
-
env = gym.make("MountainCar-v0")这个就是建立一个
MountainCar-v0的游戏环境。dom -
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_states和next_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
参考
相关文章
- 1. DQN（Deep Q-learning）入门教程（四）之Q-learning Play Flappy Bird
- 2. DQN（Deep Q-learning）入门教程（五）之DQN介绍
- 3. DQN（Deep Q-learning）入门教程（零）之教程介绍
- 4. 6.DQN(Deep Q-Network)+Double DQN+Dueling DQN
- 5. DQN（Deep Q-learning）入门教程（一）之强化学习介绍
- 6. DQN（Deep Q-learning）入门教程（二）之最优选择
- 7. DQN——Deep Q Network
- 8. 零基础10分钟运行DQN图文教程 Playing Flappy Bird Using Deep Reinforcement Learning (Based on Deep Q Learning DQN
- 9. 深度有趣 | 21 从FlappyBird到DQN
- 10. Double DQN and Dueling DQN
- 更多相关文章...
- • Memcached入门教程 - NoSQL教程
- • Neo4j数据库入门教程 - NoSQL教程
- • YAML 入门教程
- • Java 8 Stream 教程
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. Window下Ribbit MQ安装
- 2. Linux下Redis安装及集群搭建
- 3. shiny搭建网站填坑战略
- 4. Mysql8.0.22安装与配置详细教程
- 5. Hadoop安装及配置
- 6. Python爬虫初学笔记
- 7. 部署LVS-Keepalived高可用集群
- 8. keepalived+mysql高可用集群
- 9. jenkins 公钥配置
- 10. HA实用详解
欢迎关注本站公众号,获取更多信息
相关文章
- 1. DQN（Deep Q-learning）入门教程（四）之Q-learning Play Flappy Bird
- 2. DQN（Deep Q-learning）入门教程（五）之DQN介绍
- 3. DQN（Deep Q-learning）入门教程（零）之教程介绍
- 4. 6.DQN(Deep Q-Network)+Double DQN+Dueling DQN
- 5. DQN（Deep Q-learning）入门教程（一）之强化学习介绍
- 6. DQN（Deep Q-learning）入门教程（二）之最优选择
- 7. DQN——Deep Q Network
- 8. 零基础10分钟运行DQN图文教程 Playing Flappy Bird Using Deep Reinforcement Learning (Based on Deep Q Learning DQN
- 9. 深度有趣 | 21 从FlappyBird到DQN
- 10. Double DQN and Dueling DQN