## 強化學習 Actor-critic # 和PG比起來主要的變化： # 評估點由狀態價值變成了TD_error,網絡形式變了 # learn函數長得不一樣 # action有一個優化函數，優化的是價值函數，希望最大化期望的reward，Critic網絡也有一個reward，希望最小化現實和估計的誤差（即td——error) # Actor使用我們上一節講到的策略函數，負責生成動作(Action)并和環境交互。而Critic使用我們之前講到了的價值函數，負責評估Actor的表現，并指導Actor下一階段的動作import gym # import tensorflow as tf import tensorflow.compat.v1 as tf tf.compat.v1.disable_eager_execution() import numpy as np import random from collections import deque# Hyper Parameters GAMMA = 0.95 # discount factor 衰減因子 LEARNING_RATE = 0.01 # 探索率class Actor():def __init__(self, env, sess): # 初始化# init some parametersself.time_step = 0 # 某個地方需要用的步數self.state_dim = env.observation_space.shape[0] # 狀態維度self.action_dim = env.action_space.n # 動作維度self.create_softmax_network() # 創建softmax網絡# Init session 初始化tensorflow參數self.session = tf.InteractiveSession()self.session.run(tf.global_variables_initializer()) # 初始化 tensorflow 參數。def create_softmax_network(self): # 創建softmax網絡# network weightsW1 = self.weight_variable([self.state_dim, 20]) # w1 權重，4*20的網絡b1 = self.bias_variable([20]) # b1權重，y = w1*x + b1W2 = self.weight_variable([20, self.action_dim])b2 = self.bias_variable([self.action_dim])# input layerself.state_input = tf.placeholder("float", [None, self.state_dim]) # 狀態輸入層占位，多少組不知道，每組有4個狀態self.tf_acts = tf.placeholder(tf.int32, [None, 2], name="actions_num") # 給他的值對應于依據概率選擇出來的動作self.td_error = tf.placeholder(tf.float32, None, "td_error") # TD_error PG中基于狀態價值，這里評估點發生了一點變化# hidden layersh_layer = tf.nn.relu(tf.matmul(self.state_input, W1) + b1) # 進行 y = w1*x + b1 的運算 ，并激活成可輸出的狀態# softmax layer# matmul返回兩個數組的矩陣乘積，結果還是一個矩陣self.softmax_input = tf.matmul(h_layer, W2) + b2 # #進行 y = w2*x + b2 的運算，輸出是兩個是數（不確定）TODO# softmax outputself.all_act_prob = tf.nn.softmax(self.softmax_input, name='act_prob') # softmax輸出層，輸出每個動作的概率# 計算logits 和 labels 之間的softmax 交叉熵# 函數先對 logits 進行 softmax 處理得到歸一化的概率，將lables向量進行one-hot處理，然后求logits和labels的交叉熵：self.neg_log_prob = tf.nn.softmax_cross_entropy_with_logits(logits=self.softmax_input,labels=self.tf_acts)# TODO softmax_cross_entropy_with_logits 和 sparse_softmax_cross_entropy_with_logits 的區別是啥# 這句是在算損失函數了，定義為softmax交叉熵損失函數和TD_error的乘積self.exp = tf.reduce_mean(self.neg_log_prob * self.td_error) # 策略梯度函數# 創建優化器 這里需要最大化當前策略的價值，因此需要最大化self.exp,即最小化-self.exp# 由于tensorflow要minimize誤差 但是我們希望這個概率變大所以要加個負號# 利用tensorflow中的Adam優化算法最小化loss函數# Adam優化算法：是一個尋找全局最優點的優化算法，引入了二次方梯度校正。self.train_op = tf.train.AdamOptimizer(LEARNING_RATE).minimize(-self.exp)def weight_variable(self, shape):initial = tf.truncated_normal(shape)return tf.Variable(initial)def bias_variable(self, shape):initial = tf.constant(0.01, shape=shape)return tf.Variable(initial)def choose_action(self, observation): # 依據概率選擇動作"""選擇動作 ：這里的observation其實就是狀態，當前的狀態先傳入state_input(也就相當于softmax網絡的入口)，softmax網絡的輸出是針對當前狀態每個動作的概率，第一句就是運行了一個會話進行這個過程。#TODO prob_weights 應該是一個動作對應概率的矩陣,怎么查看數據類型來著忘了下一句就是依據概率選擇動作了，選擇概率最大的動作"""# np.newaxis功能：增加一個維度，具體見印象筆記prob_weights = self.session.run(self.all_act_prob, feed_dict={self.state_input: observation[np.newaxis, :]})# 這個range表示這個action的大小，后面的p表示概率分布， .ravel的意思是將數組維度拉成一維數組，也就是將矩陣向量化,見印象筆記action = np.random.choice(range(prob_weights.shape[1]), p=prob_weights.ravel())return actiondef learn(self, state, action, td_error):"""s,a 用于產生梯度上升法的方向，這時候的action是上面這個函數依據概率選擇出來的動作td 來自Critic，用于告訴Actor這個方向對不對"""s = state[np.newaxis, :] # 把state變成(4,1)的形狀one_hot_action = np.zeros(self.action_dim) # 初始化one_hot 形式的actionone_hot_action[action] = 1 # action是數字幾就把第幾個位置上的數變成1a = one_hot_action[np.newaxis, :] # 然后再把它變成橫向向量的形式# train on episodeself.session.run(self.train_op, feed_dict={self.state_input: s,self.tf_acts: a, # 把動作傳給了tf_actsself.td_error: td_error,})# critic網絡中會用到的一些超級參數 EPSILON = 0.01 # final value of epsilon epsilon 的最小值，當 epsilon 小于該值時，將不再隨機選擇行為。 REPLAY_SIZE = 10000 # experience replay buffer size 經驗回放緩沖區大小 BATCH_SIZE = 32 # size of minibatch REPLACE_TARGET_FREQ = 10 # frequency to update target Q networkclass Critic():def __init__(self, env, sess):# init some parametersself.time_step = 0self.epsilon = EPISODEself.state_dim = env.observation_space.shape[0] # 狀態維度self.action_dim = env.action_space.n # 動作維度 TODO .n是什么意思？self.create_Q_network() # 創建Q網絡self.create_training_method() # 創建訓練方法# Init session 初始化會話self.session = sessself.session.run(tf.global_variables_initializer())def create_Q_network(self): # critic網絡，使用類似于DQN的三層神經網絡，但是只有一維輸出值# network weightsW1q = self.weight_variable([self.state_dim, 20])b1q = self.bias_variable([20])W2q = self.weight_variable([20, 1])b2q = self.bias_variable([1])self.state_input = tf.placeholder(tf.float32, [1, self.state_dim], "state") # 應該是指只輸入了一組？# hidden layersh_layerq = tf.nn.relu(tf.matmul(self.state_input, W1q) + b1q) # #進行 y = w1*x + b1 的運算 ，從線性狀態激活成非線性狀態# Q Value layerself.Q_value = tf.matmul(h_layerq, W2q) + b2q # 進行 y = w2*x + b2 的運算，輸出是兩個是數（不確定）TODOdef create_training_method(self): # 創建訓練方法self.next_value = tf.placeholder(tf.float32, [1, 1], "v_next")self.reward = tf.placeholder(tf.float32, None, 'reward')# https://blog.csdn.net/tian_jiangnan/article/details/105047745# tf.variable_scope是一個變量管理器，下面的東東即使變量名一樣，作用域不一樣，引用的時候就不會出現穿插問題了with tf.variable_scope('squared_TD_error'): # 在作用域名為squared_TD_error的作用域里面self.td_error = self.reward + GAMMA * self.next_value - self.Q_value # 計算TD_errorself.loss = tf.square(self.td_error) # tf.square是對td_error里面每一個元素求平方with tf.variable_scope('train'): # 在作用域名為train的作用域里面# 利用tensorflow中的Adam優化算法最小化loss函數# Adam優化算法：是一個尋找全局最優點的優化算法，引入了二次方梯度校正。self.train_op = tf.train.AdamOptimizer(self.epsilon).minimize(self.loss)def train_Q_network(self, state, reward, next_state): # 訓練Q網絡s, s_ = state[np.newaxis, :], next_state[np.newaxis, :] # 當前狀態和下一個狀態# 由輸入狀態和Q_value計算狀態價值函數v_ = self.session.run(self.Q_value, {self.state_input: s_})# 運行會話輸出td_errortd_error, _ = self.session.run([self.td_error, self.train_op],{self.state_input: s, self.next_value: v_, self.reward: reward}) # 得到td誤差return td_errordef weight_variable(self, shape): # 權重變量initial = tf.truncated_normal(shape) # 從一個正態分布片段中輸出平均數值 shape：決定輸出張量的形狀return tf.Variable(initial) # 更新參數，變量存在內存中def bias_variable(self, shape): # 偏執變量initial = tf.constant(0.01, shape=shape) # 生成常量矩陣return tf.Variable(initial)# Hyper Parameters ENV_NAME = 'CartPole-v0' EPISODE = 3000 # Episode limitation STEP = 3000 # Step limitation in an episode TEST = 10 # The number of experiment test every 100 episode 每訓練100幕數據就做一次效果測試，測試10次取平均def main():# initialize OpenAI Gym env and dqn agentsess = tf.InteractiveSession() # 開啟會話env = gym.make(ENV_NAME) # 導入環境actor = Actor(env, sess) # 定義AC網絡critic = Critic(env, sess)for episode in range(EPISODE):# initialize task# a) 初始化S為當前狀態序列的第一個狀態, 拿到其特征向量?(S)state = env.reset() # 初始化第一個狀態# Trainfor step in range(STEP): # 這部分actor網絡和critic網絡進行交互# b) 在Actor網絡中使用?(S)作為輸入，輸出動作A,基于動作A得到新的狀態S′,反饋R。action = actor.choose_action(state) # e-greedy action for train 輸入狀態，得到動作A# c) 在Critic網絡中分別使用?(S)，?(S‘′)作為輸入，得到Q值輸出V(S)，V(S′)next_state, reward, done, _ = env.step(action) # 基于動作A得到新的狀態next_state,回報reward# 由train_Q_network計算得到TD誤差td_error = critic.train_Q_network(state, reward, next_state) # gradient = grad[r + gamma * V(s_) - V(s)]# 更新Actor網絡參數θactor.learn(state, action, td_error) # true_gradient = grad[logPi(s,a) * td_error] 最大化價值函數state = next_state # 為下一步做準備，下一個狀態即為下一步的當前狀態if done: # 達到終止條件就退出循環break# Test every 100 episodesif episode % 100 == 0:total_reward = 0 # 初始化總回報for i in range(TEST):state = env.reset() # 初始化環境for j in range(STEP):env.render() # env.render()函數用于渲染出當前的智能體以及環境的狀態action = actor.choose_action(state) # # 根據狀態選擇動作state, reward, done, _ = env.step(action) # 根據action執行step，得到三狀態total_reward += reward # 為了十次取一次平均，先加后除if done: # 如果達到了終止條件，則退出breakave_reward = total_reward / TEST # 求平均print('episode: ', episode, 'Evaluation Average Reward:', ave_reward)if __name__ == '__main__':main()

總結

以上是生活随笔為你收集整理的【强化学习】AC注释版本的全部內容，希望文章能夠幫你解決所遇到的問題。

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