本文以一個 2D 迷宮遊戲為例,逐步講解如何建構強化學習環境與訓練流程。首先,我們定義了環境類別,包含迷宮的狀態、動作空間和獎勵機制。接著,我們設計了策略類別,利用 Q-Learning 演算法,讓智慧體學習在不同狀態下選擇最佳動作。透過模擬訓練,智慧體不斷與環境互動,收集經驗並更新策略,最終學會找到迷宮中的目標。程式碼中清晰地展現了狀態更新、獎勵計算和策略調整的邏輯。此外,我們也討論了 Q-Learning 演算法的侷限性,並引出未來使用深度學習最佳化的可能性,為讀者提供了更深入的學習方向。最後,我們示範瞭如何評估訓練好的策略,並簡要介紹瞭如何使用 Ray 將訓練過程分散式化,以提升效率。
強化學習環境實作與模擬
在強化學習(RL)中,環境(Environment)是智慧體(Agent)互動的物件。本文將詳細介紹如何實作一個簡單的2D迷宮遊戲環境,並使用模擬(Simulation)來訓練智慧體尋找目標。
環境類別實作
首先,我們需要定義一個環境類別,描述遊戲的規則和狀態。以下是一個簡單的實作:
class Environment:
def __init__(self):
self.seeker = (0, 0)
self.goal = (4, 4)
self.action_space = Discrete(4) # 0: stay, 1: left, 2: up, 3: right
self.observation_space = Discrete(25)
def step(self, action):
if action == 1: # 向左移動
self.seeker = (self.seeker[0], max(self.seeker[1] - 1, 0))
elif action == 2: # 向上移動
self.seeker = (max(self.seeker[0] - 1, 0), self.seeker[1])
elif action == 3: # 向右移動
self.seeker = (self.seeker[0], min(self.seeker[1] + 1, 4))
else:
raise ValueError("Invalid action")
obs = self.get_observation()
rew = self.get_reward()
done = self.is_done()
return obs, rew, done, self.info
def render(self):
os.system('cls' if os.name == 'nt' else 'clear')
grid = [['| ' for _ in range(5)] + ["|\n"] for _ in range(5)]
grid[self.goal[0]][self.goal[1]] = '|G'
grid[self.seeker[0]][self.seeker[1]] = '|S'
print(''.join([''.join(grid_row) for grid_row in grid]))
def get_observation(self):
# 傳回當前狀態
return self.seeker[0] * 5 + self.seeker[1]
def get_reward(self):
# 傳回獎勵值
if self.seeker == self.goal:
return 10
else:
return -1
def is_done(self):
# 傳回是否結束
return self.seeker == self.goal
程式碼解密:
__init__方法:初始化環境的狀態,包括智慧體的位置、目標位置和動作空間。step方法:根據給定的動作更新智慧體的位置,並傳回新的觀察值、獎勵值、是否結束和額外資訊。render方法:渲染環境的當前狀態到命令列。get_observation、get_reward和is_done方法:分別傳回當前觀察值、獎勵值和是否結束。
策略類別實作
接下來,我們需要定義一個策略類別,用於選擇動作。以下是一個簡單的實作:
import numpy as np
class Policy:
def __init__(self, env):
self.state_action_table = np.zeros((env.observation_space.n, env.action_space.n))
self.action_space = env.action_space
def get_action(self, state, explore=True, epsilon=0.1):
if explore and np.random.uniform(0, 1) < epsilon:
return self.action_space.sample()
return np.argmax(self.state_action_table[state])
程式碼解密:
__init__方法:初始化策略的狀態-動作表格。get_action方法:根據當前狀態選擇動作,可以選擇探索或利用。
模擬與訓練
最後,我們可以使用模擬來訓練智慧體尋找目標。
environment = Environment()
policy = Policy(environment)
while not environment.is_done():
state = environment.get_observation()
action = policy.get_action(state)
environment.step(action)
environment.render()
time.sleep(0.1)
程式碼解密:
- 初始化環境和策略:建立環境和策略的例項。
- 模擬訓練:在環境中進行模擬訓練,智慧體根據策略選擇動作,並更新環境狀態。
圖表翻譯:
此圖示呈現了強化學習的基本流程,包括智慧體與環境的互動、策略的選擇和更新。
圖表翻譯: 此圖示描述了強化學習中智慧體與環境的互動過程。智慧體根據當前狀態選擇動作,環境根據動作傳回新的觀察值和獎勵值,智慧體再根據這些資訊更新策略,如此迴圈直到達到目標。
強化學習模型的訓練
想像一下,我們已經從幾場遊戲中收集了一系列的經驗。如何更新 Policy 中 state_action_table 的值呢?以下是一種方法。假設你位於位置 (3,5),並決定向右移動,這將使你到達 (4,5),距離目標只有一步之遙。很明顯,你接著就可以向右移動並獲得獎勵 1。這意味著你當前的狀態與向右的動作結合,應該具有很高的價值。換句話說,這個特定的狀態-動作對的價值應該很高。相反,在相同的情況下向左移動並不會帶來任何好處,因此對應的狀態-動作對應該具有較低的價值。
Q-Learning 演算法介紹
更一般地說,假設你處於某個狀態,採取某個動作,獲得獎勵,並進入下一個狀態。這正是我們定義經驗的方式。利用 policy.state_action_table,我們可以預先檢視從下一個狀態採取的動作是否能夠獲得收益。也就是說,我們可以計算:
next_max = np.max(policy.state_action_table[next_state])
內容解密:
np.max()用於取得policy.state_action_table[next_state]中的最大值。policy.state_action_table[next_state]代表在下一個狀態下所有可能動作的值。
我們應該如何比較這個值與當前的狀態-動作值(即 value = policy.state_action_table[state][action])呢?有多種方法可以實作這一點,但我們不能完全丟棄當前的價值,而過度信任 next_max。畢竟,這只是我們正在使用的單一經驗。因此,作為第一個近似值,我們可以計算舊值和預期值的加權和:new_value = 0.9 * value + 0.1 * next_max。這裡,0.9 和 0.1 是任意選擇的;重要的是第一個權重足夠高,以反映我們對舊值的偏好,並且兩個權重之和為 1。
更新狀態-動作值
然而,這個公式並沒有考慮到我們從獎勵中獲得的關鍵資訊。事實上,我們應該更信任當前的獎勵值,而不是預期的 next_max 值,因此將後者打一定的折扣是個好主意,比如打 10% 的折扣。更新狀態-動作值的公式如下:
new_value = 0.9 * value + 0.1 * (reward + 0.9 * next_max)
內容解密:
value是當前狀態-動作對的值。reward是採取動作後獲得的獎勵。next_max是下一個狀態下所有可能動作的最大值。- 公式中的
0.9和0.1是權重,分別代表對舊值和新資訊的信任程度。
圖表說明
@startuml
skinparam backgroundColor #FEFEFE
skinparam componentStyle rectangle
title 強化學習環境實作與模擬訓練
package "2D 迷宮環境" {
package "環境類別" {
component [狀態空間] as obs_space
component [動作空間] as action_space
component [獎勵機制] as reward
}
package "策略類別" {
component [Q-Table] as qtable
component [ε-貪婪] as epsilon
component [動作選擇] as action
}
}
package "訓練流程" {
component [模擬互動] as simulate
component [經驗收集] as experience
component [Q-Value 更新] as update
}
package "進階應用" {
component [深度 Q-Network] as dqn
component [Ray 分散式] as ray
component [策略評估] as evaluate
}
obs_space --> action_space : 25 個狀態
action_space --> reward : 4 個動作
qtable --> epsilon : 探索-利用
epsilon --> action : 選擇動作
simulate --> experience : step()
experience --> update : 學習率 α
dqn --> ray : 擴展性
note right of reward
特徵工程包含:
- 特徵選擇
- 特徵轉換
- 降維處理
end note
note right of eval
評估指標:
- 準確率/召回率
- F1 Score
- AUC-ROC
end note
@enduml
圖表翻譯:
此圖示呈現了強化學習中更新狀態-動作值的過程。首先,根據當前狀態採取某個動作,然後獲得獎勵並進入下一個狀態。在下一個狀態中,計算所有可能動作的最大值(next_max)。最後,利用當前狀態-動作對的值、獎勵和 next_max 來更新新的狀態-動作值(new_value)。
如果你已經理解了以上的解釋,本章的其餘部分將會很容易理解。從數學上講,這是本例中最難的部分。如果你曾經使用過強化學習,你會注意到這是 Q-Learning 演算法的實作。之所以這麼稱呼,是因為狀態-動作表可以被描述為一個函式 Q(state, action),它傳回這些對的值。
策略與模擬
在前面的章節中,我們定義了一個 Policy 類別和一個 Simulation 類別。Policy 類別負責根據當前的狀態選擇動作,而 Simulation 類別則負責模擬遊戲的進行並收集經驗。
Policy 類別的實作
class Policy(object):
def __init__(self, env):
self.state_action_table = np.zeros((env.observation_space.n, env.action_space.n))
def get_action(self, state, explore, epsilon):
if explore and np.random.rand() < epsilon:
return np.random.choice(self.state_action_table.shape[1])
else:
return np.argmax(self.state_action_table[state])
內容解密:
np.zeros()用於初始化state_action_table為零。get_action()方法根據當前的狀態選擇動作。如果explore為真且隨機數小於epsilon,則隨機選擇一個動作;否則,選擇具有最大值的動作。
Simulation 類別的實作
class Simulation(object):
def __init__(self, env):
self.env = env
def rollout(self, policy, render=False, explore=True, epsilon=0.1):
experiences = []
state = self.env.reset()
done = False
while not done:
action = policy.get_action(state, explore, epsilon)
next_state, reward, done, info = self.env.step(action)
experiences.append([state, action, reward, next_state])
state = next_state
if render:
time.sleep(0.05)
self.env.render()
return experiences
內容解密:
rollout()方法模擬遊戲的進行並收集經驗。experiences列表用於儲存每一步的經驗,包括當前狀態、動作、獎勵和下一個狀態。- 如果
render為真,則在每一步渲染環境。
透過這些類別和方法的實作,我們可以開始訓練我們的強化學習模型,並使其在遊戲中表現得越來越好。
強化學習模型的訓練與評估
在前面的章節中,我們已經實作了一個簡單的Q-Learning演算法來訓練一個策略。現在,讓我們來正式化這個過程,實作一個update_policy函式,用於更新給定的策略和收集到的經驗:
def update_policy(policy, experiences, weight=0.1, discount_factor=0.9):
"""根據收集到的經驗更新策略。"""
for state, action, reward, next_state in experiences:
next_max = np.max(policy.state_action_table[next_state])
value = policy.state_action_table[state][action]
new_value = (1 - weight) * value + weight * (reward + discount_factor * next_max)
policy.state_action_table[state][action] = new_value
內容解密:
- 遍歷所有經驗,按照順序更新策略。
- 對於每個經驗,計算下一個狀態下的最大價值
next_max。 - 取得當前狀態-動作對的價值
value。 - 計算新的價值
new_value,它是舊價值和預期價值的加權和,其中預期價值是當前獎勵和折扣後的next_max之和。 - 更新
state_action_table中對應的狀態-動作對的價值。
有了這個函式,我們就可以簡單地訓練一個策略來做出更好的決策。具體步驟如下:
- 初始化一個策略和一個模擬環境。
- 執行模擬環境多次,比如總共10,000次。
- 對於每一次遊戲,首先透過
rollout收集經驗。 - 然後呼叫
update_policy函式,利用收集到的經驗更新策略。
下面的train_policy函式實作了這個過程:
def train_policy(env, num_episodes=10000, weight=0.1, discount_factor=0.9):
"""透過更新策略與rollout經驗來訓練策略。"""
policy = Policy(env)
sim = Simulation(env)
for _ in range(num_episodes):
experiences = sim.rollout(policy)
update_policy(policy, experiences, weight, discount_factor)
return policy
trained_policy = train_policy(environment)
Q-Learning演算法的侷限性與改進
Q-Learning演算法是強化學習中常見的第一個演算法,因為它相對容易理解。但是,當狀態或動作空間非常大時,Q表格會變得過於龐大,導致演算法效率降低。
一種解決方案是使用神經網路來近似Q表格,這種方法稱為Deep Q-Learning。從第4章開始,我們將專門使用深度學習來解決強化學習問題。
評估訓練好的策略
現在我們已經有了一個訓練好的策略,讓我們來評估它的效能。我們將實作一個evaluate_policy函式,用於評估策略在多個遊戲中的表現:
def evaluate_policy(env, policy, num_episodes=10):
"""透過rollouts評估訓練好的策略。"""
simulation = Simulation(env)
steps = 0
for _ in range(num_episodes):
experiences = simulation.rollout(policy, render=True, explore=False)
steps += len(experiences)
print(f"{steps / num_episodes} 步數平均值,共 {num_episodes} 次遊戲。")
return steps / num_episodes
evaluate_policy(environment, trained_policy)
內容解密:
- 將
explore設為False,以充分利用訓練好的策略。 experiences的長度代表完成遊戲的步數。
分散式Ray應用的構建
為了使RL實驗成為一個分散式的Ray應用,我們需要對程式碼進行一些修改:
- 將
Simulation變成一個Ray actor。 - 定義一個平行的
train_policy版本。 - 使用
train_policy_parallel訓練和評估策略。
首先,讓我們實作一個Ray actor,稱為Simulation Actor:
import ray
ray.init()
@ray.remote
class SimulationActor:
def __init__(self, env):
self.simulation = Simulation(env)
def rollout(self, policy):
return self.simulation.rollout(policy)
內容解密:
- 使用
@ray.remote裝飾器將SimulationActor類別轉換為Ray actor。 - 在actor內部初始化一個
Simulation例項。 - 定義一個
rollout方法,用於執行模擬並傳回經驗。
透過這些步驟,我們可以將RL實驗擴充套件為一個分散式的Ray應用,從而提高訓練效率。