tensorflow-gpu==2.10.0
imageio==2.26.1
keras==2.10,0
gym==0.20.0
pyglet==1.5.20
scipy==1.10.1

import gym # 導(dǎo)入Gym庫(kù)，用于開發(fā)和比較強(qiáng)化學(xué)習(xí)算法
import numpy as np # 導(dǎo)入NumPy庫(kù)，用于進(jìn)行科學(xué)計(jì)算
import tensorflow as tf # 導(dǎo)入TensorFlow庫(kù)
from tensorflow import keras # 導(dǎo)入keras模塊，這是一個(gè)高級(jí)神經(jīng)網(wǎng)絡(luò)API
from tensorflow.keras import layers # 導(dǎo)入keras中的layers模塊，用于創(chuàng)建神經(jīng)網(wǎng)絡(luò)層
seed = 42 # 設(shè)定隨機(jī)種子，用于復(fù)現(xiàn)實(shí)驗(yàn)結(jié)果
gamma = 0.99 # 定義折扣率，用于計(jì)算未來(lái)獎(jiǎng)勵(lì)的現(xiàn)值
max_steps_per_episode = 10000 # 設(shè)定每個(gè) episode 的最大步數(shù)
env = gym.make("CartPole-v0") # 創(chuàng)建 CartPole-v0 環(huán)境實(shí)例
env.seed(seed) # 設(shè)定環(huán)境的隨機(jī)種子
eps = np.finfo(np.float32).eps.item() # 獲取 float32 數(shù)據(jù)類型的誤差最小值 epsilon 

num_inputs = 4 # 狀態(tài)空間的維度，即輸入層的節(jié)點(diǎn)數(shù)
num_actions = 2 # 行為空間的維度，即輸出層的節(jié)點(diǎn)數(shù)
num_hidden = 128 # 隱藏層的節(jié)點(diǎn)數(shù)
inputs = layers.Input(shape=(num_inputs,)) # 創(chuàng)建輸入層，指定輸入的形狀
common = layers.Dense(num_hidden, activation="relu")(inputs) # 創(chuàng)建一個(gè)全連接層，包含num_hidden 個(gè)神經(jīng)元，使用 ReLU 作為激活函數(shù)
action = layers.Dense(num_actions, activation="softmax")(common) # 創(chuàng)建一個(gè)全連接層，包含 num_actions 個(gè)神經(jīng)元，使用 softmax 作為激活函數(shù)
critic = layers.Dense(1)(common) # 創(chuàng)建一個(gè)全連接層，包含1個(gè)神經(jīng)元
model = keras.Model(inputs=inputs, outputs=[action, critic]) # 創(chuàng)建一個(gè) Keras 模型，包含輸入層、共享的隱藏層和兩個(gè)輸出層

import imageio
start = env.reset() 
frames = []
for t in range(max_steps_per_episode):
    frames.append(env.render(mode='rgb_array'))
    start = start.reshape(1, -1)
    start, reward, done, _ = env.step(np.random.choice(num_actions, p=np.squeeze(action_probs)))
    if done:
        break
with imageio.get_writer('未訓(xùn)練前的樣子.gif', mode='I') as writer:
    for frame in frames:
        writer.append_data(frame)

optimizer = keras.optimizers.Adam(learning_rate=0.01) # 創(chuàng)建 Adam 優(yōu)化器實(shí)例，設(shè)置學(xué)習(xí)率為 0.01
huber_loss = keras.losses.Huber() # 創(chuàng)建損失函數(shù)實(shí)例
action_probs_history = [] # 創(chuàng)建一個(gè)列表，用于保存 action 網(wǎng)絡(luò)在每個(gè)步驟中采取各個(gè)行動(dòng)的概率
critic_value_history = [] # 創(chuàng)建一個(gè)列表，用于保存 critic 網(wǎng)絡(luò)在每個(gè)步驟中對(duì)應(yīng)的值
rewards_history = [] # 創(chuàng)建一個(gè)列表，用于保存每個(gè)步驟的獎(jiǎng)勵(lì)值
running_reward = 0 # 初始化運(yùn)行過(guò)程中的每輪獎(jiǎng)勵(lì)
episode_count = 0 # 初始化 episode 計(jì)數(shù)器

while True:  
    state = env.reset()  # 新一輪游戲開始，重置環(huán)境
    episode_reward = 0  # 記錄本輪游戲的總獎(jiǎng)勵(lì)值
    with tf.GradientTape() as tape:  # 構(gòu)建 GradientTape 用于計(jì)算梯度
        for timestep in range(1, max_steps_per_episode): # 本輪游戲如果一切正常會(huì)進(jìn)行 max_steps_per_episode 步
            state = tf.convert_to_tensor(state)  # 將狀態(tài)轉(zhuǎn)換為張量
            state = tf.expand_dims(state, 0)  # 擴(kuò)展維度，以適應(yīng)模型的輸入形狀
            action_probs, critic_value = model(state)  # 前向傳播，得到 action 網(wǎng)絡(luò)輸出的動(dòng)作空間的概率分布，和 critic 網(wǎng)絡(luò)預(yù)測(cè)的獎(jiǎng)勵(lì)值
            critic_value_history.append(critic_value[0, 0])  # 將上面 critic 預(yù)測(cè)的獎(jiǎng)勵(lì)值記錄在 critic_value_history 列表中
            action = np.random.choice(num_actions, p=np.squeeze(action_probs))  # 依據(jù)概率分布抽樣某個(gè)動(dòng)作，當(dāng)然了某個(gè)動(dòng)作概率越大越容易被抽中，同時(shí)也保留了一定的隨機(jī)性
            action_probs_history.append(tf.math.log(action_probs[0, action]))  # 將使用該動(dòng)作的對(duì)數(shù)概率值記錄在 action_probs_history 列表中
            state, reward, done, _ = env.step(action)  # 游戲環(huán)境使用選中的動(dòng)作去執(zhí)行，得到下一個(gè)游戲狀態(tài)、獎(jiǎng)勵(lì)、是否終止和其他信息
            rewards_history.append(reward)  # 將該時(shí)刻的獎(jiǎng)勵(lì)記錄在 rewards_history 列表中
            episode_reward += reward  # 累加本輪游戲的總獎(jiǎng)勵(lì)值
            if done:  # 如果到達(dá)終止?fàn)顟B(tài)，則結(jié)束循環(huán)
                break
        running_reward = 0.05 * episode_reward + (1 - 0.05) * running_reward  # 計(jì)算平均獎(jiǎng)勵(lì)
        returns = []  # 存儲(chǔ)折扣回報(bào)
        discounted_sum = 0
        for r in rewards_history[::-1]:  # 從后往前遍歷獎(jiǎng)勵(lì)的歷史值
            discounted_sum = r + gamma * discounted_sum  # 計(jì)算折扣回報(bào)
            returns.insert(0, discounted_sum)  # 將折扣回報(bào)插入列表的開頭，最后形成的還是從前往后的折扣獎(jiǎng)勵(lì)列表
        returns = np.array(returns)  # 將折扣回報(bào)轉(zhuǎn)換為數(shù)組
        returns = (returns - np.mean(returns)) / (np.std(returns) + eps)  # 歸一化折扣回報(bào)
        returns = returns.tolist()  # 將折扣回報(bào)轉(zhuǎn)換為列表形式
        history = zip(action_probs_history, critic_value_history, returns)  # 將三個(gè)列表進(jìn)行 zip 壓縮
        actor_losses = []  # 存儲(chǔ) action 網(wǎng)絡(luò)的損失
        critic_losses = []  # 存儲(chǔ) critic 網(wǎng)絡(luò)的損失
        for log_prob, value, ret in history:
            diff = ret - value
            actor_losses.append(-log_prob * diff)  # 計(jì)算 actor 的損失函數(shù)
            critic_losses.append(
                huber_loss(tf.expand_dims(value, 0), tf.expand_dims(ret, 0)) # 計(jì)算 critic 的損失函數(shù)
            )
        loss_value = sum(actor_losses) + sum(critic_losses) # 計(jì)算總損失函數(shù)
        grads = tape.gradient(loss_value, model.trainable_variables) # 計(jì)算梯度
        optimizer.apply_gradients(zip(grads, model.trainable_variables)) # 更新模型參數(shù)
        action_probs_history.clear() # 清空之前的歷史記錄
        critic_value_history.clear() # 清空之前的歷史記錄
        rewards_history.clear() # 清空之前的歷史記錄
    episode_count += 1 # 當(dāng)一輪游戲結(jié)束時(shí)， episode 加一
    if episode_count % 10 == 0: # 每訓(xùn)練 10 個(gè) episode ，輸出當(dāng)前的平均獎(jiǎng)勵(lì)
        template = "在第 {} 輪游戲中獲得獎(jiǎng)勵(lì): {:.2f} 分"
        print(template.format(episode_count, running_reward))
    if running_reward > 195:  # 如果平均獎(jiǎng)勵(lì)超過(guò)195，視為任務(wù)已經(jīng)解決
        print("獎(jiǎng)勵(lì)超過(guò) 195 ，訓(xùn)練結(jié)束")
        break

import imageio
start = env.reset() 
frames = []
for t in range(max_steps_per_episode):
    frames.append(env.render(mode='rgb_array'))
    start = start.reshape(1, -1)
    action_probs, _ = model(start)
    action = np.random.choice(num_actions, p=np.squeeze(action_probs))
    start, reward, done, _ = env.step(action)
    if done:
        break
with imageio.get_writer('訓(xùn)練后的樣子.gif', mode='I') as writer:
    for frame in frames:
        writer.append_data(frame)