import torch # 導入torch

import torch.nn as nn # 導入torch.nn

import torch.nn.functional as F # 導入torch.nn.functional

import numpy as np # 導入numpy

import gym # 導入gym

# 超參數

BATCH_SIZE = 32 # 樣本數量

LR = 0.01 # 學習率

EPSILON = 0.9 # greedy policy

GAMMA = 0.9 # reward discount

TARGET_REPLACE_ITER = 100 # 目標網絡更新頻率

MEMORY_CAPACITY = 2000 # 記憶庫容量

env = gym.make('CartPole-v0').unwrapped # 使用gym庫中的環境：CartPole，且打開封裝 (若想了解該環境，請自行百度)

N_ACTIONS = env.action_space.n # 桿子動作個數 (2個)

N_STATES = env.observation_space.shape[0] # 桿子狀態個數 (4個)

"""

torch.nn是專門為神經網絡設計的模塊化接口。nn構建于Autograd之上，可以用來定義和運行神經網絡。

nn.Module是nn中十分重要的類，包含網絡各層的定義及forward方法。

定義網絡：

需要繼承nn.Module類，并實現forward方法。

一般把網絡中具有可學習參數的層放在構造函數__init__()中。

不具有可學習參數的層(如ReLU)可放在構造函數中，也可不放在構造函數中(而在forward中使用nn.functional來代替)。

只要在nn.Module的子類中定義了forward函數，backward函數就會被自動實現(利用Autograd)。

在forward函數中可以使用任何Variable支持的函數，畢竟在整個Pytorch構建的圖中，是Variable在流動。還可以使用if,for,print,log等python語法。

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

"""

# 定義Net類 (定義網絡)

class Net(nn.Module):

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

# nn.Module的子類函數必須在構造函數中執行父類的構造函數

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

self.fc1 = nn.Linear(N_STATES, 50) # 設置第一個全連接層(輸入層到隱藏層): 狀態數個神經元到50個神經元

self.fc1.weight.data.normal_(0, 0.1) # 權重初始化 (均值為0，方差為0.1的正態分布)

self.out = nn.Linear(50, N_ACTIONS) # 設置第二個全連接層(隱藏層到輸出層): 50個神經元到動作數個神經元

self.out.weight.data.normal_(0, 0.1) # 權重初始化 (均值為0，方差為0.1的正態分布)

def forward(self, x): # 定義forward函數 (x為狀態)

x = F.relu(self.fc1(x)) # 連接輸入層到隱藏層，且使用激勵函數ReLU來處理經過隱藏層后的值

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

return actions_value # 返回動作值

# 定義DQN類 (定義兩個網絡)

class DQN(object):

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

self.eval_net, self.target_net = Net(), Net() # 利用Net創建兩個神經網絡: 評估網絡和目標網絡

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優化器 (輸入為評估網絡的參數和學習率)

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

def choose_action(self, x): # 定義動作選擇函數 (x為狀態)

x = torch.unsqueeze(torch.FloatTensor(x), 0) # 將x轉換成32-bit floating point形式，并在dim=0增加維數為1的維度

if np.random.uniform() < EPSILON: # 生成一個在[0, 1)內的隨機數，如果小于EPSILON，選擇最優動作

actions_value = self.eval_net.forward(x) # 通過對評估網絡輸入狀態x，前向傳播獲得動作值

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

action = action[0] # 輸出action的第一個數

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_): # 定義記憶存儲函數 (這里輸入為一個transition)

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

# 如果記憶庫滿了，便覆蓋舊的數據

index = self.memory_counter % MEMORY_CAPACITY # 獲取transition要置入的行數

self.memory[index, :] = transition # 置入transition

self.memory_counter += 1 # memory_counter自加1

def learn(self): # 定義學習函數(記憶庫已滿后便開始學習)

# 目標網絡參數更新

if self.learn_step_counter % TARGET_REPLACE_ITER == 0: # 一開始觸發，然后每100步觸發

self.target_net.load_state_dict(self.eval_net.state_dict()) # 將評估網絡的參數賦給目標網絡

self.learn_step_counter += 1 # 學習步數自加1

# 抽取記憶庫中的批數據

sample_index = np.random.choice(MEMORY_CAPACITY, BATCH_SIZE) # 在[0, 2000)內隨機抽取32個數，可能會重復

b_memory = self.memory[sample_index, :] # 抽取32個索引對應的32個transition，存入b_memory

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

# 將32個s抽出，轉為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抽出，轉為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抽出，轉為32-bit floating point形式，并存儲到b_s中，b_r為32行1列

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

# 將32個s_抽出，轉為32-bit floating point形式，并存儲到b_s中，b_s_為32行4列

# 獲取32個transition的評估值和目標值，并利用損失函數和優化器進行評估網絡參數更新

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

# eval_net(b_s)通過評估網絡輸出32行每個b_s對應的一系列動作值，然后.gather(1, b_a)代表對每行對應索引b_a的Q值提取進行聚合

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

# q_next不進行反向傳遞誤差，所以detach；q_next表示通過目標網絡輸出32行每個b_s_對應的一系列動作值

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

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

loss = self.loss_func(q_eval, q_target)

# 輸入32個評估值和32個目標值，使用均方損失函數

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

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

self.optimizer.step() # 更新評估網絡的所有參數

dqn = DQN() # 令dqn=DQN類

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

for i_episode in range(400): # 400個episode循環

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

ep_r = 0 # 初始化該循環對應的episode的獎勵

while True: # 開始一個episode (每一個循環代表一步)

env.render() # 顯示實驗動畫

a = dqn.choose_action(s) # 輸入該步對應的狀態s，選擇動作

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

# 修改獎勵 (不修改也可以，修改獎勵只是為了更快地得到訓練好的擺桿)

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對應的獎勵自加本步執行動作獲得的獎勵r

if dqn.memory_counter > MEMORY_CAPACITY: # 如果累計的transition數量超過了記憶庫的固定容量2000

dqn.learn()

# 開始學習 (抽取記憶，即32個transition，并對評估網絡參數進行更新，并在開始學習后每隔100次將評估網絡的參數賦給目標網絡)

if done:

# 如果該episode對應的done為True，則輸出以下內容 (注意并不是第一個episode就開始，而是記憶庫已滿后才開始)

# 因此每次開始進行輸出的對應的episode不一樣

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

'| Ep_r: ', round(ep_r, 2)) # round()方法返回ep_r的小數點四舍五入到2個數字

if done: # 如果滿足終止條件

break # 該episode結束

s = s_ # 更新狀態

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。