基于DQN和TensorFlow的LunarLander實現(xiàn)（全代碼）

使用深度Q網(wǎng)絡(luò)（Deep Q-Network, DQN）來訓練一個在openai-gym的LunarLander-v2環(huán)境中的強化學習agent，讓小火箭成功著陸。

下面代碼直接扔到j(luò)upyter notebook或CoLab上就能跑起來。

安裝和導入所需的庫和環(huán)境

安裝和設(shè)置所需的庫和環(huán)境，使其能夠在Jupyter Notebook中運行。

!pip install gym
!apt-get install xvfb -y
!pip install pyvirtualdisplay #用于在沒有顯示器的環(huán)境中創(chuàng)建虛擬顯示
!pip install Pillow #一個圖像處理庫
!pip install swig
!pip install "gym[box2d]"

創(chuàng)建并啟動一個虛擬顯示，在沒有圖形界面的服務器上運行強化學習環(huán)境：

from pyvirtualdisplay import Display
display = Display(visible=0, size=(1400, 900))
display.start()

引入所需庫：

import gym
import time
import tqdm
import numpy as np
from IPython import display as ipydisplay
from PIL import Image

創(chuàng)建一個LunarLander-v2環(huán)境的DQN代理：

agent = DQNAgent('LunarLander-v2')

total_score, records = agent.simulate(visualize=True)
print(f'Total score {total_score:.2f}')
record_list = []
for i in tqdm.tqdm(range(100)):
total_score, _ = agent.simulate(visualize=False)
record_list.append(total_score)

print(f'Average score in 100 episode {np.mean(record_list):.2f}')

Q網(wǎng)絡(luò)搭建

import tensorflow as tf

L = tf.keras.layers

def create_network_model(input_shape: np.ndarray,
action_space: np.ndarray,
learning_rate=0.001) -> tf.keras.Sequential:
model = tf.keras.Sequential([
L.Dense(512, input_shape=input_shape, activation="relu"),
L.Dense(256, input_shape=input_shape, activation="relu"),
L.Dense(action_space)
])
model.compile(loss="mse",
optimizer=tf.optimizers.Adam(lr=learning_rate))
return model

經(jīng)驗回放實現(xiàn)

經(jīng)驗回放是一種在深度強化學習中常用的技術(shù)，主要用于打破數(shù)據(jù)的相關(guān)性和減少過擬合。

在強化學習中，代理通常會在訓練過程中與環(huán)境進行大量交互，經(jīng)驗回放允許代理存儲這些經(jīng)驗，并在后續(xù)的訓練中反復利用這些數(shù)據(jù)。這種機制有助于改善學習效率，減少數(shù)據(jù)樣本間的時間相關(guān)性，提高訓練過程的穩(wěn)定性。

import random
import numpy as np
from collections import namedtuple

# 代表每一個樣本的 namedtuple，方便存儲和讀取數(shù)據(jù)
Experience = namedtuple('Experience', ('state', 'action', 'reward', 'next_state', 'done'))

class ReplayMemory:

def __init__(self, max_size):
self.max_size = max_size
self.memory = []

def append(self, state, action, reward, next_state, done):
"""記錄一個新的樣本"""
sample = Experience(state, action, reward, next_state, done)
self.memory.append(sample)
# 只留下最新記錄的 self.max_size 個樣本
self.memory = self.memory[-self.max_size:]

def sample(self, batch_size):
"""按照給定批次大小取樣"""
samples = random.sample(self.memory, batch_size)
batch = Experience(*zip(*samples))

# 轉(zhuǎn)換數(shù)據(jù)為 numpy 張量返回
states = np.array(batch.state)
actions = np.array(batch.action)
rewards = np.array(batch.reward)
states_next = np.array(batch.next_state)
dones = np.array(batch.done)

return states, actions, rewards, states_next, dones

def __len__(self):
return len(self.memory)

DQNAgent實現(xiàn)

DQNAgent類是DQN算法的核心實現(xiàn)。它包含以下關(guān)鍵部分：

1、初始化：初始化環(huán)境、神經(jīng)網(wǎng)絡(luò)模型和經(jīng)驗回放緩存。
2、行為選擇（choose_action）：根據(jù)當前狀態(tài)和ε-greedy策略選擇行為。
3、經(jīng)驗回放（replay）：從記憶中隨機抽取小批量經(jīng)驗進行學習。
4、訓練（train）：進行多個episode的訓練。

from IPython import display
from PIL import Image

# 定義超參數(shù)
LEARNING_RATE = 0.001
GAMMA = 0.99
EPSILON_DECAY = 0.995
EPSILON_MIN = 0.01

class DQNAgent:
def __init__(self, env_name):
self.env = gym.make(env_name)
self.observation_shape = self.env.observation_space.shape
self.action_count = self.env.action_space.n
self.model = create_network_model(self.observation_shape, self.action_count)
self.memory = ReplayMemory(500000)
self.epsilon = 1.0
self.batch_size = 64

def choose_action(self, state, epsilon=None):
"""
根據(jù)給定狀態(tài)選擇行為
- epsilon == 0 完全使用模型選擇行為
- epsilon == 1 完全隨機選擇行為
"""
if epsilon is None:
epsilon = self.epsilon
if np.random.rand() < epsilon:
return np.random.randint(self.action_count)
else:
q_values = self.model.predict(np.expand_dims(state, axis=0))
return np.argmax(q_values[0])

def replay(self):
"""進行經(jīng)驗回放學習"""

# 如果當前經(jīng)驗池經(jīng)驗數(shù)量少于批次大小，則跳過
if len(self.memory) < self.batch_size:
return

states, actions, rewards, states_next, dones = self.memory.sample(self.batch_size)
q_pred = self.model.predict(states)

q_next = self.model.predict(states_next).max(axis=1)
q_next = q_next * (1 - dones)
q_update = rewards + GAMMA * q_next

indices = np.arange(self.batch_size)
q_pred[[indices], [actions]] = q_update

self.model.train_on_batch(states, q_pred)

def simulate(self, epsilon=None, visualize=True):
records = []
state = self.env.reset()
is_done = False
total_score = 0
total_step = 0
while not is_done:
action = self.choose_action(state, epsilon)
state, reward, is_done, info = self.env.step(action)
total_score += reward
total_step += 1

rgb_array = self.env.render(mode='rgb_array')
records.append((rgb_array, action, reward, total_score))

if visualize:
display.clear_output(wait=True)
img = Image.fromarray(rgb_array)
# 當前 Cell 中展示圖片
display.display(img)
print(f'Action {action} Action reward {reward:.2f} | Total score {total_score:.2f} | Step {total_step}')

time.sleep(0.01)
self.env.close()
return total_score, records

def train(self, episode_count: int, log_dir: str):
"""
訓練方法，按照給定 episode 數(shù)量進行訓練，并記錄訓練過程關(guān)鍵參數(shù)到 TensorBoard
"""
# 初始化一個 TensorBoard 記錄器
file_writer = tf.summary.create_file_writer(log_dir)
file_writer.set_as_default()

score_list = []
best_avg_score = -np.inf

for episode_index in range(episode_count):
state = self.env.reset()
score, step = 0, 0
is_done = False
while not is_done:
# 根據(jù)狀態(tài)選擇一個行為
action = self.choose_action(state)
# 執(zhí)行行為，記錄行為和結(jié)果到經(jīng)驗池
state_next, reward, is_done, info = self.env.step(action)
self.memory.append(state, action, reward, state_next, is_done)
score += reward

state = state_next
# 每 6 步進行一次回放訓練
# 此處也可以選擇每一步回放訓練，但會降低訓練速度，這個是一個經(jīng)驗技巧
if step % 1 == 0:
self.replay()
step += 1

# 記錄當前 Episode 的得分，計算最后 100 Episode 的平均得分
score_list.append(score)
avg_score = np.mean(score_list[-100:])

# 記錄當前 Episode 得分，epsilon 和最后 100 Episode 的平均得分到 TensorBoard
tf.summary.scalar('score', data=score, step=episode_index)
tf.summary.scalar('average score', data=avg_score, step=episode_index)
tf.summary.scalar('epsilon', data=self.epsilon, step=episode_index)

# 終端輸出訓練進度
print(f'Episode: {episode_index} Reward: {score:03.2f} '
f'Average Reward: {avg_score:03.2f} Epsilon: {self.epsilon:.3f}')

# 調(diào)整 epsilon 值，逐漸減少隨機探索比例
if self.epsilon > EPSILON_MIN:
self.epsilon *= EPSILON_DECAY

# 如果當前平均得分比之前有改善，保存模型
# 確保提前創(chuàng)建目錄 outputs/chapter_15
if avg_score > best_avg_score:
best_avg_score = avg_score
self.model.save(f'outputs/chapter_15/dqn_best_{episode_index}.h5')

訓練

# 使用 LunarLander 初始化 Agent
agent = DQNAgent('LunarLander-v2')
import glob
# 讀取現(xiàn)在已經(jīng)記錄的日志數(shù)量，避免日志重復記錄
tf_log_index = len(glob.glob('tf_dir/lunar_lander/run_*'))
log_dir = f'tf_dir/lunar_lander/run_{tf_log_index}'

# 訓練 2000 個 Episode
agent.train(20, log_dir)

agent.model.summary()