DQN簡介
DQN,全稱Deep Q Network,是一種融合了神經網絡和Q-learning的方法。這種新型結構突破了傳統強化學習的瓶頸,下面具體介紹:
神經網絡的作用
傳統強化學習使用表格形式來存儲每一個狀態state和狀態對應的action的Q值,例如下表表示狀態s1對應了兩種動作action,每種action對應的Q值為-2和1。
a1 | a2 | |
s1 | -2 |
1 |
s2 | 2 | 3 |
... | ... | ... |
但當我們有很多數據時,首先,內存不夠是第一個問題;其次,搜索某個狀態對應的動作也相當不便,這時就想到了引入神經網絡。
我們可以把狀態和動作當作神經網絡的輸入,然后經過神經元的分析得到每個動作對應的Q值,這樣我們就沒有必要用表格記錄Q值;或者我們也可以只輸入狀態,輸出所有的動作,然后根據Q-learning的原則,直接選擇具有最大Q值的動作作為下一步要做的動作。就好比相當於眼睛鼻子耳朵收集信息, 然后通過大腦加工輸出每種動作的值, 最后通過強化學習的方式選擇動作。
更新神經網絡
神經網絡需要被訓練才能預測出正確的值。如何更新呢?
假設已存在Q表,當前狀態s1有兩個對應的action,分別是a1和a2,根據Q表,選擇最大價值的a2,此時進入到狀態s2,而 s2也有兩個對應的action,分別是a1和a2,選擇最大價值的動作,乘以衰減率gamma,加上獎勵reward,此時得到的值便是Q(s1,a2)的現實值,而Q(s1,a2)的估計值直接由Q表給出,此時便有了Q現實與Q估計的差距,然后該差距乘以學習率,再加之前的參數,便得到了最新神經網絡的參數。
DQN兩大特點
DQN 有一個記憶庫用於存儲學習之前的經歷。Q learning 是一種 off-policy 離線學習法,它能學習當前經歷着的, 也能學習過去經歷過的, 甚至是學習別人的經歷。所以每次 DQN 更新的時候, 我們都可以隨機抽取一些之前的經歷進行學習。 隨機抽取這種做法打亂了經歷之間的相關性, 也使得神經網絡更新更有效率。而 Fixed Q-targets 也是一種打亂相關性的機理, 如果使用 fixed Q-targets, 我們就會在 DQN 中使用到兩個結構相同但參數不同的神經網絡,代碼中用target_net、eval_net表示這兩種不同的神經網絡。
DQN代碼實現
在設計該算法之前,需要設定很多超參數的值(以訓練小車立桿子為例):
BATCH_SIZE = 32 # 每個批量的大小 LR = 0.01 # 學習率 EPSILON = 0.9 # 最優選擇動作百分比 greedy policy GAMMA = 0.9 # 獎勵衰減率 TARGET_REPLACE_ITER = 100 # Q 現實網絡(target)的更新頻率 MEMORY_CAPACITY = 2000 # 記憶庫的大小 env = gym.make('CartPole-v0') # 立桿子游戲 env = env.unwrapped N_ACTIONS = env.action_space.n # 桿子會做的動作 N_STATES = env.observation_space.shape[0] # 桿子能獲得的環境信息數
對神經網絡進行初始化,定義可以輸出所有動作的Q值的函數:
class Net(nn.Module): def __init__(self, ): super(Net, self).__init__() self.fc1 = nn.Linear(N_STATES, 50) # 輸入狀態,對應着50個輸出 self.fc1.weight.data.normal_(0, 0.1) # 通過正態分布隨機生成參數值 self.out = nn.Linear(50, N_ACTIONS) # 輸入fc1層的輸出,對應着多個動作 self.out.weight.data.normal_(0, 0.1) def forward(self, x): x = self.fc1(x) x = F.relu(x) actions_value = self.out(x) # 有可用的兩種Action,輸出每個動作對應的價值,之后會根據價值選取動作 return actions_value
上面介紹過,DQN的特點是,會使用到兩個結構相同但參數不同的神經網絡,下面是構建這兩個神經網絡的過程:
# DQN框架:與環境互動的過程,有兩個Net,包含選擇動作機制、存儲經歷機制、學習機制 class DQN(object): def __init__(self): # 建立target_net、eval_net、memory self.eval_net, self.target_net = Net(), Net() # 兩個一樣的Net,但參數不一樣,要經常把eval_net的參數轉換到target_net中,以達到延遲的更新效果 self.learn_step_counter = 0 # 學習了多少步 self.memory_counter = 0 # 記憶庫中的位置 self.memory = np.zeros((MEMORY_CAPACITY, N_STATES * 2 + 2)) # 初始化記憶庫, self.optimizer = torch.optim.Adam(self.eval_net.parameters(), lr=LR) self.loss_func = nn.MSELoss()
該框架中應有三個函數,分別實現根據觀測值選擇動作、存儲到記憶庫中、從記憶庫提取記憶然后進行強化學習
# 根據觀測值選擇動作機制 def choose_action(self, x): # 根據觀測值x選擇動作 x = torch.unsqueeze(torch.FloatTensor(x), 0) # 存在隨機選取動作的概率,當該概率<EPSILON時,執行greedy行為 # forward()最后輸出所有action的價值,選取高價值的動作;這里就是90%的情況下,選取高價值動作 if np.random.uniform() < EPSILON: # greedy actions_value = self.eval_net.forward(x) # 輸出actions_value action = torch.max(actions_value, 1)[1].data.numpy() # 選取最大價值 action = action[0] if ENV_A_SHAPE == 0 else action.reshape(ENV_A_SHAPE) # return the argmax index # 反之隨機選取一個動作即可 else: action = np.random.randint(0, N_ACTIONS) action = action if ENV_A_SHAPE == 0 else action.reshape(ENV_A_SHAPE) # 返回最后選取的動作 return action
# 選取完Action之后,存儲到記憶庫中 def store_transition(self, s, a, r, s_): #記憶庫存儲的內容為四元組:狀態、動作、獎勵、下一個狀態 # 存儲記憶 transition = np.hstack((s, [a, r], s_)) # 如果memory_counter已經超過記憶庫容量MEMORY_CAPACITY,那么重新開始索引,用新記憶代替舊記憶 index = self.memory_counter % MEMORY_CAPACITY self.memory[index, :] = transition self.memory_counter += 1
# 學習機制:從記憶庫中提取記憶然后用RL的方法進行學習 def learn(self): # 檢測是否需要進行 target 網絡更新 # 檢測標准:Q現實網絡要隔多少步進行一次更新,如果達到那個步數就進行更新 if self.learn_step_counter % TARGET_REPLACE_ITER == 0: # eval_net的參數全部復制到target_net中,實現更新 # target_net是時不時更新,而eval_net是每一步都要更新 self.target_net.load_state_dict(self.eval_net.state_dict()) self.learn_step_counter += 1 # 實現eval_net的每一步更新,從記憶庫中隨機抽取BATCH_SIZE個記憶,然后打包 sample_index = np.random.choice(MEMORY_CAPACITY, BATCH_SIZE) b_memory = self.memory[sample_index, :] # 按照存儲時的格式進行打包,然后就可以放入神經網絡中進行學習 b_s = torch.FloatTensor(b_memory[:, :N_STATES]) b_a = torch.LongTensor(b_memory[:, N_STATES:N_STATES+1].astype(int)) b_r = torch.FloatTensor(b_memory[:, N_STATES+1:N_STATES+2]) b_s_ = torch.FloatTensor(b_memory[:, -N_STATES:]) # 學習過程:輸入現在的狀態b_s,通過forward()生成所有動作的價值,根據價值選取動作,把它的價值賦值給q_eval q_eval = self.eval_net(b_s).gather(1, b_a) # 把target網絡中下一步的狀態對應的價值賦值給q_next;此處有時會反向傳播更新target,但此處不需更新,故加.detach() q_next = self.target_net(b_s_).detach() # 選取下一步動作的最大值,乘衰減率GAMMA,然后加獎勵b_r;max函數返回索引加最大值,索引是1最大值是0 q_target = b_r + GAMMA * q_next.max(1)[0].view(BATCH_SIZE, 1) # 。view(BATCH_SIZE, 1)表示最大項會放在batch的第一位 # 通過預測值與真實值計算損失 loss = self.loss_func(q_eval, q_target) # 反向傳遞誤差,進行參數更新 self.optimizer.zero_grad() loss.backward() self.optimizer.step() # 學習過程總結:首先檢驗是否需要將target_net更新為eval_net,然后進行批訓練,就是從記憶庫中提取一批記憶, # 最后計算出q_eval、q_next等以及誤差后,通過誤差反向傳遞進行學習