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

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

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è)置學(xué)習(xí)率為 0.01
huber_loss = keras.losses.Huber() # 創(chuàng)建損失函數(shù)實例
action_probs_history = [] # 創(chuàng)建一個列表，用于保存 action 網(wǎng)絡(luò)在每個步驟中采取各個行動的概率
critic_value_history = [] # 創(chuàng)建一個列表，用于保存 critic 網(wǎng)絡(luò)在每個步驟中對應(yīng)的值
rewards_history = [] # 創(chuàng)建一個列表，用于保存每個步驟的獎勵值
running_reward = 0 # 初始化運行過程中的每輪獎勵
episode_count = 0 # 初始化 episode 計數(shù)器

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