莫烦python讲得好差_莫烦PYTHON——PyTorch——DQN 代码详解
import torch # 導(dǎo)入torch
import torch.nn as nn # 導(dǎo)入torch.nn
import torch.nn.functional as F # 導(dǎo)入torch.nn.functional
import numpy as np # 導(dǎo)入numpy
import gym # 導(dǎo)入gym
# 超參數(shù)
BATCH_SIZE = 32 # 樣本數(shù)量
LR = 0.01 # 學(xué)習(xí)率
EPSILON = 0.9 # greedy policy
GAMMA = 0.9 # reward discount
TARGET_REPLACE_ITER = 100 # 目標(biāo)網(wǎng)絡(luò)更新頻率
MEMORY_CAPACITY = 2000 # 記憶庫容量
env = gym.make('CartPole-v0').unwrapped # 使用gym庫中的環(huán)境：CartPole，且打開封裝 (若想了解該環(huán)境，請自行百度)
N_ACTIONS = env.action_space.n # 桿子動作個數(shù) (2個)
N_STATES = env.observation_space.shape[0] # 桿子狀態(tài)個數(shù) (4個)
"""
torch.nn是專門為神經(jīng)網(wǎng)絡(luò)設(shè)計的模塊化接口。nn構(gòu)建于Autograd之上，可以用來定義和運行神經(jīng)網(wǎng)絡(luò)。
nn.Module是nn中十分重要的類，包含網(wǎng)絡(luò)各層的定義及forward方法。
定義網(wǎng)絡(luò)：
需要繼承nn.Module類，并實現(xiàn)forward方法。
一般把網(wǎng)絡(luò)中具有可學(xué)習(xí)參數(shù)的層放在構(gòu)造函數(shù)__init__()中。
不具有可學(xué)習(xí)參數(shù)的層(如ReLU)可放在構(gòu)造函數(shù)中，也可不放在構(gòu)造函數(shù)中(而在forward中使用nn.functional來代替)。
只要在nn.Module的子類中定義了forward函數(shù)，backward函數(shù)就會被自動實現(xiàn)(利用Autograd)。
在forward函數(shù)中可以使用任何Variable支持的函數(shù)，畢竟在整個Pytorch構(gòu)建的圖中，是Variable在流動。還可以使用if,for,print,log等python語法。
注：Pytorch基于nn.Module構(gòu)建的模型中，只支持mini-batch的Variable輸入方式。
"""
# 定義Net類 (定義網(wǎng)絡(luò))
class Net(nn.Module):
def __init__(self): # 定義Net的一系列屬性
# nn.Module的子類函數(shù)必須在構(gòu)造函數(shù)中執(zhí)行父類的構(gòu)造函數(shù)
super(Net, self).__init__() # 等價與nn.Module.__init__()
self.fc1 = nn.Linear(N_STATES, 50) # 設(shè)置第一個全連接層(輸入層到隱藏層): 狀態(tài)數(shù)個神經(jīng)元到50個神經(jīng)元
self.fc1.weight.data.normal_(0, 0.1) # 權(quán)重初始化 (均值為0，方差為0.1的正態(tài)分布)
self.out = nn.Linear(50, N_ACTIONS) # 設(shè)置第二個全連接層(隱藏層到輸出層): 50個神經(jīng)元到動作數(shù)個神經(jīng)元
self.out.weight.data.normal_(0, 0.1) # 權(quán)重初始化 (均值為0，方差為0.1的正態(tài)分布)
def forward(self, x): # 定義forward函數(shù) (x為狀態(tài))
x = F.relu(self.fc1(x)) # 連接輸入層到隱藏層，且使用激勵函數(shù)ReLU來處理經(jīng)過隱藏層后的值
actions_value = self.out(x) # 連接隱藏層到輸出層，獲得最終的輸出值 (即動作值)
return actions_value # 返回動作值
# 定義DQN類 (定義兩個網(wǎng)絡(luò))
class DQN(object):
def __init__(self): # 定義DQN的一系列屬性
self.eval_net, self.target_net = Net(), Net() # 利用Net創(chuàng)建兩個神經(jīng)網(wǎng)絡(luò): 評估網(wǎng)絡(luò)和目標(biāo)網(wǎng)絡(luò)
self.learn_step_counter = 0 # for target updating
self.memory_counter = 0 # for storing memory
self.memory = np.zeros((MEMORY_CAPACITY, N_STATES * 2 + 2)) # 初始化記憶庫，一行代表一個transition
self.optimizer = torch.optim.Adam(self.eval_net.parameters(), lr=LR) # 使用Adam優(yōu)化器 (輸入為評估網(wǎng)絡(luò)的參數(shù)和學(xué)習(xí)率)
self.loss_func = nn.MSELoss() # 使用均方損失函數(shù) (loss(xi, yi)=(xi-yi)^2)
def choose_action(self, x): # 定義動作選擇函數(shù) (x為狀態(tài))
x = torch.unsqueeze(torch.FloatTensor(x), 0) # 將x轉(zhuǎn)換成32-bit floating point形式，并在dim=0增加維數(shù)為1的維度
if np.random.uniform() < EPSILON: # 生成一個在[0, 1)內(nèi)的隨機數(shù)，如果小于EPSILON，選擇最優(yōu)動作
actions_value = self.eval_net.forward(x) # 通過對評估網(wǎng)絡(luò)輸入狀態(tài)x，前向傳播獲得動作值
action = torch.max(actions_value, 1)[1].data.numpy() # 輸出每一行最大值的索引，并轉(zhuǎn)化為numpy ndarray形式
action = action[0] # 輸出action的第一個數(shù)
else: # 隨機選擇動作
action = np.random.randint(0, N_ACTIONS) # 這里action隨機等于0或1 (N_ACTIONS = 2)
return action # 返回選擇的動作 (0或1)
def store_transition(self, s, a, r, s_): # 定義記憶存儲函數(shù) (這里輸入為一個transition)
transition = np.hstack((s, [a, r], s_)) # 在水平方向上拼接數(shù)組
# 如果記憶庫滿了，便覆蓋舊的數(shù)據(jù)
index = self.memory_counter % MEMORY_CAPACITY # 獲取transition要置入的行數(shù)
self.memory[index, :] = transition # 置入transition
self.memory_counter += 1 # memory_counter自加1
def learn(self): # 定義學(xué)習(xí)函數(shù)(記憶庫已滿后便開始學(xué)習(xí))
# 目標(biāo)網(wǎng)絡(luò)參數(shù)更新
if self.learn_step_counter % TARGET_REPLACE_ITER == 0: # 一開始觸發(fā)，然后每100步觸發(fā)
self.target_net.load_state_dict(self.eval_net.state_dict()) # 將評估網(wǎng)絡(luò)的參數(shù)賦給目標(biāo)網(wǎng)絡(luò)
self.learn_step_counter += 1 # 學(xué)習(xí)步數(shù)自加1
# 抽取記憶庫中的批數(shù)據(jù)
sample_index = np.random.choice(MEMORY_CAPACITY, BATCH_SIZE) # 在[0, 2000)內(nèi)隨機抽取32個數(shù)，可能會重復(fù)
b_memory = self.memory[sample_index, :] # 抽取32個索引對應(yīng)的32個transition，存入b_memory
b_s = torch.FloatTensor(b_memory[:, :N_STATES])
# 將32個s抽出，轉(zhuǎn)為32-bit floating point形式，并存儲到b_s中，b_s為32行4列
b_a = torch.LongTensor(b_memory[:, N_STATES:N_STATES+1].astype(int))
# 將32個a抽出，轉(zhuǎn)為64-bit integer (signed)形式，并存儲到b_a中 (之所以為LongTensor類型，是為了方便后面torch.gather的使用)，b_a為32行1列
b_r = torch.FloatTensor(b_memory[:, N_STATES+1:N_STATES+2])
# 將32個r抽出，轉(zhuǎn)為32-bit floating point形式，并存儲到b_s中，b_r為32行1列
b_s_ = torch.FloatTensor(b_memory[:, -N_STATES:])
# 將32個s_抽出，轉(zhuǎn)為32-bit floating point形式，并存儲到b_s中，b_s_為32行4列
# 獲取32個transition的評估值和目標(biāo)值，并利用損失函數(shù)和優(yōu)化器進(jìn)行評估網(wǎng)絡(luò)參數(shù)更新
q_eval = self.eval_net(b_s).gather(1, b_a)
# eval_net(b_s)通過評估網(wǎng)絡(luò)輸出32行每個b_s對應(yīng)的一系列動作值，然后.gather(1, b_a)代表對每行對應(yīng)索引b_a的Q值提取進(jìn)行聚合
q_next = self.target_net(b_s_).detach()
# q_next不進(jìn)行反向傳遞誤差，所以detach；q_next表示通過目標(biāo)網(wǎng)絡(luò)輸出32行每個b_s_對應(yīng)的一系列動作值
q_target = b_r + GAMMA * q_next.max(1)[0].view(BATCH_SIZE, 1)
# q_next.max(1)[0]表示只返回每一行的最大值，不返回索引(長度為32的一維張量)；.view()表示把前面所得到的一維張量變成(BATCH_SIZE, 1)的形狀；最終通過公式得到目標(biāo)值
loss = self.loss_func(q_eval, q_target)
# 輸入32個評估值和32個目標(biāo)值，使用均方損失函數(shù)
self.optimizer.zero_grad() # 清空上一步的殘余更新參數(shù)值
loss.backward() # 誤差反向傳播, 計算參數(shù)更新值
self.optimizer.step() # 更新評估網(wǎng)絡(luò)的所有參數(shù)
dqn = DQN() # 令dqn=DQN類
print('\nCollecting experience...') # 打印“Collecting experience...”
for i_episode in range(400): # 400個episode循環(huán)
s = env.reset() # 重置環(huán)境
ep_r = 0 # 初始化該循環(huán)對應(yīng)的episode的獎勵
while True: # 開始一個episode (每一個循環(huán)代表一步)
env.render() # 顯示實驗動畫
a = dqn.choose_action(s) # 輸入該步對應(yīng)的狀態(tài)s，選擇動作
s_, r, done, info = env.step(a) # 執(zhí)行動作，獲得反饋
# 修改獎勵 (不修改也可以，修改獎勵只是為了更快地得到訓(xùn)練好的擺桿)
x, x_dot, theta, theta_dot = s_
r1 = (env.x_threshold - abs(x)) / env.x_threshold - 0.8
r2 = (env.theta_threshold_radians - abs(theta)) / env.theta_threshold_radians - 0.5
r = r1 + r2
dqn.store_transition(s, a, r, s_) # 存儲樣本
ep_r += r # 該episode對應(yīng)的獎勵自加本步執(zhí)行動作獲得的獎勵r
if dqn.memory_counter > MEMORY_CAPACITY: # 如果累計的transition數(shù)量超過了記憶庫的固定容量2000
dqn.learn()
# 開始學(xué)習(xí) (抽取記憶，即32個transition，并對評估網(wǎng)絡(luò)參數(shù)進(jìn)行更新，并在開始學(xué)習(xí)后每隔100次將評估網(wǎng)絡(luò)的參數(shù)賦給目標(biāo)網(wǎng)絡(luò))
if done:
# 如果該episode對應(yīng)的done為True，則輸出以下內(nèi)容 (注意并不是第一個episode就開始，而是記憶庫已滿后才開始)
# 因此每次開始進(jìn)行輸出的對應(yīng)的episode不一樣
print('Ep: ', i_episode, # 輸出該episode數(shù)
'| Ep_r: ', round(ep_r, 2)) # round()方法返回ep_r的小數(shù)點四舍五入到2個數(shù)字
if done: # 如果滿足終止條件
break # 該episode結(jié)束
s = s_ # 更新狀態(tài)
總結(jié)
以上是生活随笔為你收集整理的莫烦python讲得好差_莫烦PYTHON——PyTorch——DQN 代码详解的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 去除任务栏的图标
- 下一篇： Perl文档操作选项