使用深度學習演算法 DQN 來玩 flappy bird 無敵了！
//www.bilibili.com/video/BV1xf4y1174A/
使用深度學習演算法 DQN 來玩 flappy bird
安裝依賴
pip install parl == 1.3.1
pip install pygame
pip install paddlepaddle
模擬環境 PLE 庫
(PyGame-Learning-Environment)[//github.com/ntasfi/PyGame-Learning-Environment]
模型簡介
使用了 百度 PARL 深度學習庫直接調用 DQN 演算法
由於遊戲的 state 僅有 8 維，所以模型網路僅使用了 2 個全連接層
收斂情況
在訓練了 1000 個 episode 以後可以明顯看出在逐步收斂
在訓練了 10000 個 episode 以後，測試中基本可以保持一直進行下去，所以不得不限制到達一定分數就終止遊戲
參數調整心得
需要保持較高的探索概率，30%-20%較佳
import parl
from parl import layers
import paddle.fluid as fluid
from parl.utils import logger
from parl.algorithms import DQN
from ple.games.flappybird import FlappyBird
from ple import PLE
from pygame.constants import K_w
import random
import collections
import numpy as np
actions = {"up": K_w}
#LEARN_FREQ = 5 # 訓練頻率，不需要每一個step都learn，攢一些新增經驗後再learn，提高效率
LEARN_FREQ = 5
#MEMORY_SIZE = 20000 # replay memory的大小，越大越佔用記憶體
MEMORY_SIZE = 20000
#MEMORY_WARMUP_SIZE = 200 # replay_memory 里需要預存一些經驗數據，再從裡面sample一個batch的經驗讓agent去learn
MEMORY_WARMUP_SIZE = 200
#BATCH_SIZE = 32 # 每次給agent learn的數據數量，從replay memory隨機里sample一批數據出來
BATCH_SIZE = 32
#GAMMA = 0.99 # reward 的衰減因子，一般取 0.9 到 0.999 不等
GAMMA = 0.999
LEARNING_RATE = 0.0001 # 學習率
class Model(parl.Model):
def __init__(self, act_dim):
hid1_size = 128
hid2_size = 128
self.fc1 = layers.fc(size=hid1_size, act='tanh')
self.fc2 = layers.fc(size=hid2_size, act='tanh')
self.fc3 = layers.fc(size=act_dim, act=None)
def value(self, obs):
# 定義網路
# 輸入state，輸出所有action對應的Q，[Q(s,a1), Q(s,a2)]
h1 = self.fc1(obs)
h2 = self.fc2(h1)
Q = self.fc3(h2)
return Q
class Agent(parl.Agent):
def __init__(self,
algorithm,
obs_dim,
act_dim,
e_greed=0.1,
e_greed_decrement=0):
assert isinstance(obs_dim, int)
assert isinstance(act_dim, int)
self.obs_dim = obs_dim
self.act_dim = act_dim
super(Agent, self).__init__(algorithm)
self.global_step = 0
#self.update_target_steps = 200 # 每隔200個training steps再把model的參數複製到target_model中
self.update_target_steps = 200
self.e_greed = e_greed # 有一定概率隨機選取動作，探索
self.e_greed_decrement = e_greed_decrement # 隨著訓練逐步收斂，探索的程度慢慢降低
def build_program(self):
self.pred_program = fluid.Program()
self.learn_program = fluid.Program()
with fluid.program_guard(self.pred_program): # 搭建計算圖用於 預測動作，定義輸入輸出變數
obs = layers.data(
name='obs', shape=[self.obs_dim], dtype='float32')
self.value = self.alg.predict(obs)
with fluid.program_guard(self.learn_program): # 搭建計算圖用於 更新Q網路，定義輸入輸出變數
obs = layers.data(
name='obs', shape=[self.obs_dim], dtype='float32')
action = layers.data(name='act', shape=[1], dtype='int32')
reward = layers.data(name='reward', shape=[], dtype='float32')
next_obs = layers.data(
name='next_obs', shape=[self.obs_dim], dtype='float32')
terminal = layers.data(name='terminal', shape=[], dtype='bool')
self.cost = self.alg.learn(obs, action, reward, next_obs, terminal)
def sample(self, obs):
sample = np.random.rand() # 產生0~1之間的小數
if sample < self.e_greed:
act = np.random.randint(self.act_dim) # 探索：每個動作都有概率被選擇
else:
act = self.predict(obs) # 選擇最優動作
self.e_greed = max(
0.1, self.e_greed - self.e_greed_decrement) # 隨著訓練逐步收斂，探索的程度慢慢降低
#self.e_greed = 0.2
return act
def predict(self, obs): # 選擇最優動作
obs = np.expand_dims(obs, axis=0)
pred_Q = self.fluid_executor.run(
self.pred_program,
feed={'obs': obs.astype('float32')},
fetch_list=[self.value])[0]
pred_Q = np.squeeze(pred_Q, axis=0)
act = np.argmax(pred_Q) # 選擇Q最大的下標，即對應的動作
return act
def learn(self, obs, act, reward, next_obs, terminal):
# 每隔200個training steps同步一次model和target_model的參數
if self.global_step % self.update_target_steps == 0:
self.alg.sync_target()
self.global_step += 1
act = np.expand_dims(act, -1)
feed = {
'obs': obs.astype('float32'),
'act': act.astype('int32'),
'reward': reward,
'next_obs': next_obs.astype('float32'),
'terminal': terminal
}
cost = self.fluid_executor.run(
self.learn_program, feed=feed, fetch_list=[self.cost])[0] # 訓練一次網路
return cost
# replay_memory.py
class ReplayMemory(object):
def __init__(self, max_size):
self.buffer = collections.deque(maxlen=max_size)
# 增加一條經驗到經驗池中
def append(self, exp):
self.buffer.append(exp)
# 從經驗池中選取N條經驗出來
def sample(self, batch_size):
mini_batch = random.sample(self.buffer, batch_size)
obs_batch, action_batch, reward_batch, next_obs_batch, done_batch = [], [], [], [], []
for experience in mini_batch:
s, a, r, s_p, done = experience
obs_batch.append(s)
action_batch.append(a)
reward_batch.append(r)
next_obs_batch.append(s_p)
done_batch.append(done)
return np.array(obs_batch).astype('float32'), \
np.array(action_batch).astype('float32'), np.array(reward_batch).astype('float32'),\
np.array(next_obs_batch).astype('float32'), np.array(done_batch).astype('float32')
def __len__(self):
return len(self.buffer)
# 訓練一個episode
def run_episode(env, agent, rpm):
total_reward = 0
env.init()
step = 0
while True:
# 第一幀為黑屏，不操作
if (step == 0):
reward = env.act(None)
done = False
else:
obs = list(env.getGameState().values())
action = agent.sample(obs) # 取樣動作，所有動作都有概率被嘗試到
# 神經網路輸出轉化為實際動作
if action == 1:
act = actions["up"]
else:
act = None
reward = env.act(act)
done = env.game_over()
next_obs = list(env.getGameState().values())
rpm.append((obs, action, reward, next_obs, done))
# train model
if (len(rpm) > MEMORY_WARMUP_SIZE) and (step % LEARN_FREQ == 0):
(batch_obs, batch_action, batch_reward, batch_next_obs,
batch_done) = rpm.sample(BATCH_SIZE)
train_loss = agent.learn(batch_obs, batch_action, batch_reward,
batch_next_obs,
batch_done) # s,a,r,s',done
total_reward += reward
if done:
print(step)
env.reset_game() # 重置遊戲
break
step += 1
return total_reward
# 評估 agent, 跑 5 個episode，總reward求平均
def evaluate(env, agent):
eval_reward = []
for i in range(5):
env.init()
episode_reward = 0
step = 0
while True:
# 第一幀為黑屏，不操作
if (step == 0):
reward = env.act(None)
done = False
else:
obs = list(env.getGameState().values())
action = agent.predict(obs) # 預測動作，只選最優動作
# 神經網路輸出轉化為實際動作
if action == 1:
act = actions["up"]
else:
act = None
reward = env.act(act)
done = env.game_over()
episode_reward += reward
if (step == 5000):
print(step)
break
if done:
print(step)
env.reset_game() # 重置遊戲
break
step += 1
eval_reward.append(episode_reward)
return np.mean(eval_reward)
# 創建環境
game = FlappyBird()
env_1 = PLE(game, fps=30, display_screen=False)
env_2 = PLE(game, fps=30, display_screen=False)
obs_dim = len(env_1.getGameState())
act_dim = 2
logger.info('obs_dim {}, act_dim {}'.format(obs_dim, act_dim))
# 創建經驗池
rpm = ReplayMemory(MEMORY_SIZE) # DQN的經驗回放池
# 根據parl框架構建agent
model = Model(act_dim=act_dim)
algorithm = DQN(model, act_dim = act_dim, gamma=GAMMA, lr=LEARNING_RATE)
agent = Agent(
algorithm,
obs_dim = obs_dim,
act_dim = act_dim,
e_greed = 0.2,
e_greed_decrement = 1e-6
)
# 載入模型
save_path = 'bird_dqn_v3_7.ckpt'
agent.restore(save_path)
# 先往經驗池裡存一些數據，避免最開始訓練的時候樣本豐富度不夠
while len(rpm) < MEMORY_WARMUP_SIZE:
run_episode(env_1, agent, rpm)
max_episode = 2000
# 開始訓練
episode = 0
while episode < max_episode: # 訓練max_episode個回合，test部分不計算入episode數量
# train part
for i in range(0, 100):
total_reward = run_episode(env_1, agent, rpm)
episode += 1
# test part
eval_reward = evaluate(env_2, agent) # render=True 查看顯示效果
logger.info('episode:{} e_greed:{} test_reward:{}'.format(
episode, agent.e_greed, eval_reward))
# 保存模型
ckpt = 'bird_dqn_v3_dir/steps_{}.ckpt'.format(episode)
agent.save(ckpt)
# 訓練結束，保存模型
save_path = './bird_dqn_v3_8.ckpt'
agent.save(save_path)
