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 # 記憶庫(kù)容量

env = gym.make('CartPole-v0').unwrapped # 使用gym庫(kù)中的環(huán)境：CartPole，且打開(kāi)封裝 (若想了解該環(huán)境，請(qǐng)自行百度)

N_ACTIONS = env.action_space.n # 桿子動(dòng)作個(gè)數(shù) (2個(gè))

N_STATES = env.observation_space.shape[0] # 桿子狀態(tài)個(gè)數(shù) (4個(gè))

"""

torch.nn是專(zhuān)門(mén)為神經(jīng)網(wǎng)絡(luò)設(shè)計(jì)的模塊化接口。nn構(gòu)建于Autograd之上，可以用來(lái)定義和運(yùn)行神經(jīng)網(wǎng)絡(luò)。

nn.Module是nn中十分重要的類(lèi)，包含網(wǎng)絡(luò)各層的定義及forward方法。

定義網(wǎng)絡(luò)：

需要繼承nn.Module類(lèi)，并實(shí)現(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來(lái)代替)。

只要在nn.Module的子類(lèi)中定義了forward函數(shù)，backward函數(shù)就會(huì)被自動(dòng)實(shí)現(xiàn)(利用Autograd)。

在forward函數(shù)中可以使用任何Variable支持的函數(shù)，畢竟在整個(gè)Pytorch構(gòu)建的圖中，是Variable在流動(dòng)。還可以使用if,for,print,log等python語(yǔ)法。

注：Pytorch基于nn.Module構(gòu)建的模型中，只支持mini-batch的Variable輸入方式。

"""

# 定義Net類(lèi) (定義網(wǎng)絡(luò))

class Net(nn.Module):

def __init__(self): # 定義Net的一系列屬性

# nn.Module的子類(lèi)函數(shù)必須在構(gòu)造函數(shù)中執(zhí)行父類(lèi)的構(gòu)造函數(shù)

super(Net, self).__init__() # 等價(jià)與nn.Module.__init__()

self.fc1 = nn.Linear(N_STATES, 50) # 設(shè)置第一個(gè)全連接層(輸入層到隱藏層): 狀態(tài)數(shù)個(gè)神經(jīng)元到50個(gè)神經(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è)置第二個(gè)全連接層(隱藏層到輸出層): 50個(gè)神經(jīng)元到動(dòng)作數(shù)個(gè)神經(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)) # 連接輸入層到隱藏層，且使用激勵(lì)函數(shù)ReLU來(lái)處理經(jīng)過(guò)隱藏層后的值

actions_value = self.out(x) # 連接隱藏層到輸出層，獲得最終的輸出值 (即動(dòng)作值)

return actions_value # 返回動(dòng)作值

# 定義DQN類(lèi) (定義兩個(gè)網(wǎng)絡(luò))

class DQN(object):

def __init__(self): # 定義DQN的一系列屬性

self.eval_net, self.target_net = Net(), Net() # 利用Net創(chuàng)建兩個(gè)神經(jīng)網(wǎng)絡(luò): 評(píng)估網(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)) # 初始化記憶庫(kù)，一行代表一個(gè)transition

self.optimizer = torch.optim.Adam(self.eval_net.parameters(), lr=LR) # 使用Adam優(yōu)化器 (輸入為評(píng)估網(wǎng)絡(luò)的參數(shù)和學(xué)習(xí)率)

self.loss_func = nn.MSELoss() # 使用均方損失函數(shù) (loss(xi, yi)=(xi-yi)^2)

def choose_action(self, x): # 定義動(dòng)作選擇函數(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: # 生成一個(gè)在[0, 1)內(nèi)的隨機(jī)數(shù)，如果小于EPSILON，選擇最優(yōu)動(dòng)作

actions_value = self.eval_net.forward(x) # 通過(guò)對(duì)評(píng)估網(wǎng)絡(luò)輸入狀態(tài)x，前向傳播獲得動(dòng)作值

action = torch.max(actions_value, 1)[1].data.numpy() # 輸出每一行最大值的索引，并轉(zhuǎn)化為numpy ndarray形式

action = action[0] # 輸出action的第一個(gè)數(shù)

else: # 隨機(jī)選擇動(dòng)作

action = np.random.randint(0, N_ACTIONS) # 這里action隨機(jī)等于0或1 (N_ACTIONS = 2)

return action # 返回選擇的動(dòng)作 (0或1)

def store_transition(self, s, a, r, s_): # 定義記憶存儲(chǔ)函數(shù) (這里輸入為一個(gè)transition)

transition = np.hstack((s, [a, r], s_)) # 在水平方向上拼接數(shù)組

# 如果記憶庫(kù)滿(mǎn)了，便覆蓋舊的數(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ù)(記憶庫(kù)已滿(mǎn)后便開(kāi)始學(xué)習(xí))

# 目標(biāo)網(wǎng)絡(luò)參數(shù)更新

if self.learn_step_counter % TARGET_REPLACE_ITER == 0: # 一開(kāi)始觸發(fā)，然后每100步觸發(fā)

self.target_net.load_state_dict(self.eval_net.state_dict()) # 將評(píng)估網(wǎng)絡(luò)的參數(shù)賦給目標(biāo)網(wǎng)絡(luò)

self.learn_step_counter += 1 # 學(xué)習(xí)步數(shù)自加1

# 抽取記憶庫(kù)中的批數(shù)據(jù)

sample_index = np.random.choice(MEMORY_CAPACITY, BATCH_SIZE) # 在[0, 2000)內(nèi)隨機(jī)抽取32個(gè)數(shù)，可能會(huì)重復(fù)

b_memory = self.memory[sample_index, :] # 抽取32個(gè)索引對(duì)應(yīng)的32個(gè)transition，存入b_memory

b_s = torch.FloatTensor(b_memory[:, :N_STATES])

# 將32個(gè)s抽出，轉(zhuǎn)為32-bit floating point形式，并存儲(chǔ)到b_s中，b_s為32行4列

b_a = torch.LongTensor(b_memory[:, N_STATES:N_STATES+1].astype(int))

# 將32個(gè)a抽出，轉(zhuǎn)為64-bit integer (signed)形式，并存儲(chǔ)到b_a中 (之所以為L(zhǎng)ongTensor類(lèi)型，是為了方便后面torch.gather的使用)，b_a為32行1列

b_r = torch.FloatTensor(b_memory[:, N_STATES+1:N_STATES+2])

# 將32個(gè)r抽出，轉(zhuǎn)為32-bit floating point形式，并存儲(chǔ)到b_s中，b_r為32行1列

b_s_ = torch.FloatTensor(b_memory[:, -N_STATES:])

# 將32個(gè)s_抽出，轉(zhuǎn)為32-bit floating point形式，并存儲(chǔ)到b_s中，b_s_為32行4列

# 獲取32個(gè)transition的評(píng)估值和目標(biāo)值，并利用損失函數(shù)和優(yōu)化器進(jìn)行評(píng)估網(wǎng)絡(luò)參數(shù)更新

q_eval = self.eval_net(b_s).gather(1, b_a)

# eval_net(b_s)通過(guò)評(píng)估網(wǎng)絡(luò)輸出32行每個(gè)b_s對(duì)應(yīng)的一系列動(dòng)作值，然后.gather(1, b_a)代表對(duì)每行對(duì)應(yīng)索引b_a的Q值提取進(jìn)行聚合

q_next = self.target_net(b_s_).detach()

# q_next不進(jìn)行反向傳遞誤差，所以detach；q_next表示通過(guò)目標(biāo)網(wǎng)絡(luò)輸出32行每個(gè)b_s_對(duì)應(yīng)的一系列動(dòng)作值

q_target = b_r + GAMMA * q_next.max(1)[0].view(BATCH_SIZE, 1)

# q_next.max(1)[0]表示只返回每一行的最大值，不返回索引(長(zhǎng)度為32的一維張量)；.view()表示把前面所得到的一維張量變成(BATCH_SIZE, 1)的形狀；最終通過(guò)公式得到目標(biāo)值

loss = self.loss_func(q_eval, q_target)

# 輸入32個(gè)評(píng)估值和32個(gè)目標(biāo)值，使用均方損失函數(shù)

self.optimizer.zero_grad() # 清空上一步的殘余更新參數(shù)值

loss.backward() # 誤差反向傳播, 計(jì)算參數(shù)更新值

self.optimizer.step() # 更新評(píng)估網(wǎng)絡(luò)的所有參數(shù)

dqn = DQN() # 令dqn=DQN類(lèi)

print('\nCollecting experience...') # 打印“Collecting experience...”

for i_episode in range(400): # 400個(gè)episode循環(huán)

s = env.reset() # 重置環(huán)境

ep_r = 0 # 初始化該循環(huán)對(duì)應(yīng)的episode的獎(jiǎng)勵(lì)

while True: # 開(kāi)始一個(gè)episode (每一個(gè)循環(huán)代表一步)

env.render() # 顯示實(shí)驗(yàn)動(dòng)畫(huà)

a = dqn.choose_action(s) # 輸入該步對(duì)應(yīng)的狀態(tài)s，選擇動(dòng)作

s_, r, done, info = env.step(a) # 執(zhí)行動(dòng)作，獲得反饋

# 修改獎(jiǎng)勵(lì) (不修改也可以，修改獎(jiǎng)勵(lì)只是為了更快地得到訓(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_) # 存儲(chǔ)樣本

ep_r += r # 該episode對(duì)應(yīng)的獎(jiǎng)勵(lì)自加本步執(zhí)行動(dòng)作獲得的獎(jiǎng)勵(lì)r

if dqn.memory_counter > MEMORY_CAPACITY: # 如果累計(jì)的transition數(shù)量超過(guò)了記憶庫(kù)的固定容量2000

dqn.learn()

# 開(kāi)始學(xué)習(xí) (抽取記憶，即32個(gè)transition，并對(duì)評(píng)估網(wǎng)絡(luò)參數(shù)進(jìn)行更新，并在開(kāi)始學(xué)習(xí)后每隔100次將評(píng)估網(wǎng)絡(luò)的參數(shù)賦給目標(biāo)網(wǎng)絡(luò))

if done:

# 如果該episode對(duì)應(yīng)的done為T(mén)rue，則輸出以下內(nèi)容 (注意并不是第一個(gè)episode就開(kāi)始，而是記憶庫(kù)已滿(mǎn)后才開(kāi)始)

# 因此每次開(kāi)始進(jìn)行輸出的對(duì)應(yīng)的episode不一樣

print('Ep: ', i_episode, # 輸出該episode數(shù)

'| Ep_r: ', round(ep_r, 2)) # round()方法返回ep_r的小數(shù)點(diǎn)四舍五入到2個(gè)數(shù)字

if done: # 如果滿(mǎn)足終止條件

break # 該episode結(jié)束

s = s_ # 更新?tīng)顟B(tài)

總結(jié)

以上是生活随笔為你收集整理的莫烦python讲得好差_莫烦PYTHON——PyTorch——DQN 代码详解的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。