通過(guò)CartPole游戲詳解PPO優(yōu)化的方法

本篇內(nèi)容主要講解“通過(guò)CartPole游戲詳解PPO優(yōu)化的方法”，感興趣的朋友不妨來(lái)看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來(lái)帶大家學(xué)習(xí)“通過(guò)CartPole游戲詳解PPO優(yōu)化的方法”吧!

CartPole 介紹

在一個(gè)光滑的軌道上有個(gè)推車(chē)，桿子垂直微置在推車(chē)上，隨時(shí)有倒的風(fēng)險(xiǎn)。系統(tǒng)每次對(duì)推車(chē)施加向左或者向右的力，但我們的目標(biāo)是讓桿子保持直立。桿子保持直立的每個(gè)時(shí)間單位都會(huì)獲得 +1 的獎(jiǎng)勵(lì)。但是當(dāng)桿子與垂直方向成 15 度以上的位置，或者推車(chē)偏離中心點(diǎn)超過(guò) 2.4 個(gè)單位后，這一輪局游戲結(jié)束。因此我們可以獲得的最高回報(bào)等于 200 。我們這里就是要通過(guò)使用 PPO 算法來(lái)訓(xùn)練一個(gè)強(qiáng)化學(xué)習(xí)模型 actor-critic ，通過(guò)對(duì)比模型訓(xùn)練前后的游戲運(yùn)行 gif 圖，可以看出來(lái)我們訓(xùn)練好的模型能長(zhǎng)時(shí)間保持桿子處于垂直狀態(tài)。

庫(kù)準(zhǔn)備

python==3.10.9
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

超參數(shù)設(shè)置

這段代碼主要是導(dǎo)入所需的庫(kù)，并設(shè)置了一些超參數(shù)。

    import numpy as np
    import tensorflow as tf
    from tensorflow import keras
    from tensorflow.keras import layers
    import gym
    import scipy.signal
    import time
    from tqdm import tqdm
    steps_per_epoch = 5000  # 每個(gè) epoch 中訓(xùn)練的步數(shù)
    epochs = 20  # 用于訓(xùn)練的 epoch 數(shù)
    gamma = 0.90  # 折扣因子，用于計(jì)算回報(bào)
    clip_ratio = 0.2  # PPO 算法中用于限制策略更新的比率
    policy_learning_rate = 3e-4  # 策略網(wǎng)絡(luò)的學(xué)習(xí)率
    value_function_learning_rate = 3e-4  # 值函數(shù)網(wǎng)絡(luò)的學(xué)習(xí)率
    train_policy_iterations = 80  # 策略網(wǎng)絡(luò)的訓(xùn)練迭代次數(shù)
    train_value_iterations = 80  # 值函數(shù)網(wǎng)絡(luò)的訓(xùn)練迭代次數(shù)
    lam = 0.97  # PPO 算法中的 λ 參數(shù)
    target_kl = 0.01  # PPO 算法中的目標(biāo) KL 散度
    hidden_sizes = (64, 64) # 神經(jīng)網(wǎng)絡(luò)的隱藏層維度 
    render = False    # 是否開(kāi)啟畫(huà)面渲染，F(xiàn)alse 表示不開(kāi)啟

模型定義

（1）這里定義了一個(gè)函數(shù) discounted_cumulative_sums，接受兩個(gè)參數(shù) x 和 discount，該函數(shù)的作用是計(jì)算給定獎(jiǎng)勵(lì)序列 x 的折扣累計(jì)和，折扣因子 discount 是一個(gè)介于 0 和 1 之間的值，表示對(duì)未來(lái)獎(jiǎng)勵(lì)的折扣程度。 在強(qiáng)化學(xué)習(xí)中，折扣累計(jì)和是一個(gè)常用的概念，表示對(duì)未來(lái)獎(jiǎng)勵(lì)的折扣累加。

def discounted_cumulative_sums(x, discount):
    return scipy.signal.lfilter([1], [1, float(-discount)], x[::-1], axis=0)[::-1]

（2）這里定義了一個(gè)Buffer類(lèi)，用于存儲(chǔ)訓(xùn)練數(shù)據(jù)。類(lèi)中有如下主要的函數(shù)：

• init: 初始化函數(shù)，用于設(shè)置成員變量的初始值

• store: 將觀(guān)測(cè)值、行為、獎(jiǎng)勵(lì)、價(jià)值和對(duì)數(shù)概率存儲(chǔ)到對(duì)應(yīng)的緩沖區(qū)中

• finish_trajectory: 結(jié)束一條軌跡，用于計(jì)算優(yōu)勢(shì)和回報(bào)，并更新 trajectory_start_index 的值

get: 獲取所有緩沖區(qū)的值，用在訓(xùn)練模型過(guò)程中。在返回緩沖區(qū)的值之前，將優(yōu)勢(shì)緩沖區(qū)的值進(jìn)行標(biāo)準(zhǔn)化處理，使其均值為 0 ，方差為 1

class Buffer:
    def __init__(self, observation_dimensions, size, gamma=0.99, lam=0.95):
        self.observation_buffer = np.zeros( (size, observation_dimensions), dtype=np.float32 )
        self.action_buffer = np.zeros(size, dtype=np.int32)
        self.advantage_buffer = np.zeros(size, dtype=np.float32)
        self.reward_buffer = np.zeros(size, dtype=np.float32)
        self.return_buffer = np.zeros(size, dtype=np.float32)
        self.value_buffer = np.zeros(size, dtype=np.float32)
        self.logprobability_buffer = np.zeros(size, dtype=np.float32)
        self.gamma, self.lam = gamma, lam
        self.pointer, self.trajectory_start_index = 0, 0
    def store(self, observation, action, reward, value, logprobability):
        self.observation_buffer[self.pointer] = observation
        self.action_buffer[self.pointer] = action
        self.reward_buffer[self.pointer] = reward
        self.value_buffer[self.pointer] = value
        self.logprobability_buffer[self.pointer] = logprobability
        self.pointer += 1
    def finish_trajectory(self, last_value=0):
        path_slice = slice(self.trajectory_start_index, self.pointer)
        rewards = np.append(self.reward_buffer[path_slice], last_value)
        values = np.append(self.value_buffer[path_slice], last_value)
        deltas = rewards[:-1] + self.gamma * values[1:] - values[:-1]
        self.advantage_buffer[path_slice] = discounted_cumulative_sums( deltas, self.gamma * self.lam )
        self.return_buffer[path_slice] = discounted_cumulative_sums(  rewards, self.gamma )[:-1]
        self.trajectory_start_index = self.pointer
    def get(self):
        self.pointer, self.trajectory_start_index = 0, 0
        advantage_mean, advantage_std = (  np.mean(self.advantage_buffer),  np.std(self.advantage_buffer), )
        self.advantage_buffer = (self.advantage_buffer - advantage_mean) / advantage_std
        return ( self.observation_buffer, self.action_buffer, self.advantage_buffer, self.return_buffer, self.logprobability_buffer, )

（3）這里定義了一個(gè)多層感知機(jī)（Multi-Layer Perceptron，MLP）的網(wǎng)絡(luò)結(jié)構(gòu)，有如下參數(shù)：

• x：輸入的張量

• sizes：一個(gè)包含每一層的神經(jīng)元個(gè)數(shù)的列表

• activation：激活函數(shù)，用于中間層的神經(jīng)元

• output_activation：輸出層的激活函數(shù)

該函數(shù)通過(guò)循環(huán)生成相應(yīng)個(gè)數(shù)的全連接層，并將 x 作為輸入傳入。其中，units 指定每一層的神經(jīng)元個(gè)數(shù)，activation 指定該層使用的激活函數(shù)，返回最后一層的結(jié)果。

def mlp(x, sizes, activation=tf.tanh, output_activation=None):
    for size in sizes[:-1]:
        x = layers.Dense(units=size, activation=activation)(x)
    return layers.Dense(units=sizes[-1], activation=output_activation)(x)

（4）這里定義了一個(gè)函數(shù) logprobabilities，用于計(jì)算給定動(dòng)作 a 的對(duì)數(shù)概率。函數(shù)接受兩個(gè)參數(shù)，logits 和 a，其中 logits 表示模型輸出的未歸一化的概率分布，a 表示當(dāng)前采取的動(dòng)作。函數(shù)首先對(duì) logits 進(jìn)行 softmax 歸一化，然后對(duì)歸一化后的概率分布取對(duì)數(shù)，得到所有動(dòng)作的對(duì)數(shù)概率。接著，函數(shù)使用 tf.one_hot 函數(shù)生成一個(gè) one-hot 編碼的動(dòng)作向量，并與所有動(dòng)作的對(duì)數(shù)概率向量相乘，最后對(duì)結(jié)果進(jìn)行求和得到給定動(dòng)作的對(duì)數(shù)概率。

def logprobabilities(logits, a):
    logprobabilities_all = tf.nn.log_softmax(logits)
    logprobability = tf.reduce_sum( tf.one_hot(a, num_actions) * logprobabilities_all, axis=1 )
    return logprobability

（5）這里定義了一個(gè)函數(shù) sample_action。該函數(shù)接受一個(gè) observation（觀(guān)測(cè)值）參數(shù)，并在 actor 網(wǎng)絡(luò)上運(yùn)行該觀(guān)測(cè)值以獲得動(dòng)作 logits（邏輯值）。然后使用邏輯值（logits）來(lái)隨機(jī)采樣出一個(gè)動(dòng)作，并將結(jié)果作為函數(shù)的輸出。

@tf.function
def sample_action(observation):
    logits = actor(observation)
    action = tf.squeeze(tf.random.categorical(logits, 1), axis=1)
    return logits, action

（6）這里定義了一個(gè)用于訓(xùn)練策略的函數(shù)train_policy。該函數(shù)使用帶權(quán)重裁剪的 PPO 算法，用于更新 actor 的權(quán)重。

• observation_buffer：輸入的觀(guān)測(cè)緩沖區(qū)

• action_buffer：輸入的動(dòng)作緩沖區(qū)

• logprobability_buffer：輸入的對(duì)數(shù)概率緩沖區(qū)

• advantage_buffer：輸入的優(yōu)勢(shì)值緩沖區(qū)

在該函數(shù)內(nèi)部，使用tf.GradientTape記錄執(zhí)行的操作，用于計(jì)算梯度并更新策略網(wǎng)絡(luò)。計(jì)算的策略損失是策略梯度和剪裁比率的交集和。使用優(yōu)化器policy_optimizer來(lái)更新actor的權(quán)重。最后，計(jì)算并返回 kl 散度的平均值，該值用于監(jiān)控訓(xùn)練的過(guò)程。

@tf.function
def train_policy( observation_buffer, action_buffer, logprobability_buffer, advantage_buffer):
    with tf.GradientTape() as tape:   
        ratio = tf.exp( logprobabilities(actor(observation_buffer), action_buffer) - logprobability_buffer )
        min_advantage = tf.where(  advantage_buffer > 0, (1 + clip_ratio) * advantage_buffer, (1 - clip_ratio) * advantage_buffer, )
        policy_loss = -tf.reduce_mean( tf.minimum(ratio * advantage_buffer, min_advantage) )
    policy_grads = tape.gradient(policy_loss, actor.trainable_variables)
    policy_optimizer.apply_gradients(zip(policy_grads, actor.trainable_variables))
    kl = tf.reduce_mean( logprobability_buffer - logprobabilities(actor(observation_buffer), action_buffer) )
    kl = tf.reduce_sum(kl)
    return kl

（7）這里實(shí)現(xiàn)了價(jià)值函數(shù)（critic）的訓(xùn)練過(guò)程，函數(shù)接受兩個(gè)參數(shù)：一個(gè)是 observation_buffer，表示當(dāng)前存儲(chǔ)的狀態(tài)觀(guān)察序列；另一個(gè)是 return_buffer，表示狀態(tài)序列對(duì)應(yīng)的回報(bào)序列。在函數(shù)內(nèi)部，首先使用 critic 模型來(lái)預(yù)測(cè)當(dāng)前狀態(tài)序列對(duì)應(yīng)的狀態(tài)值（V）， 然后計(jì)算當(dāng)前狀態(tài)序列的平均回報(bào)與 V 之間的均方誤差，并對(duì)其進(jìn)行求和取平均得到損失函數(shù) value_loss。接下來(lái)計(jì)算梯度來(lái)更新可訓(xùn)練的變量值。

@tf.function
def train_value_function(observation_buffer, return_buffer):
    with tf.GradientTape() as tape:  
        value_loss = tf.reduce_mean((return_buffer - critic(observation_buffer)) ** 2)
    value_grads = tape.gradient(value_loss, critic.trainable_variables)
    value_optimizer.apply_gradients(zip(value_grads, critic.trainable_variables))

游戲初始化

這里用于構(gòu)建強(qiáng)化學(xué)習(xí)中的 Actor-Critic 網(wǎng)絡(luò)模型。首先，使用 gy m庫(kù)中的 CartPole-v0 環(huán)境創(chuàng)建一個(gè)環(huán)境實(shí)例 env 。然后，定義了兩個(gè)變量，分別表示觀(guān)測(cè)空間的維度 observation_dimensions 和動(dòng)作空間的大小 num_actions，這些信息都可以從 env 中獲取。接著，定義了一個(gè) Buffer 類(lèi)的實(shí)例，用于存儲(chǔ)每個(gè)時(shí)間步的觀(guān)測(cè)、動(dòng)作、獎(jiǎng)勵(lì)、下一個(gè)觀(guān)測(cè)和 done 信號(hào)，以便后面的訓(xùn)練使用。

然后，使用 Keras 庫(kù)定義了一個(gè)神經(jīng)網(wǎng)絡(luò)模型 Actor ，用于近似模仿策略函數(shù)，該模型輸入是當(dāng)前的觀(guān)測(cè)，輸出是每個(gè)動(dòng)作的概率分布的對(duì)數(shù)。

另外，還定義了一個(gè)神經(jīng)網(wǎng)絡(luò)模型 Critic ，用于近似模仿值函數(shù)，該模型輸入是當(dāng)前的觀(guān)測(cè)，輸出是一個(gè)值，表示這個(gè)觀(guān)測(cè)的價(jià)值。最后，定義了兩個(gè)優(yōu)化器，policy_optimizer 用于更新 Actor 網(wǎng)絡(luò)的參數(shù)，value_optimizer 用于更新 Critic 網(wǎng)絡(luò)的參數(shù)。

env = gym.make("CartPole-v0")
observation_dimensions = env.observation_space.shape[0]
num_actions = env.action_space.n
buffer = Buffer(observation_dimensions, steps_per_epoch)
observation_input = keras.Input(shape=(observation_dimensions,), dtype=tf.float32)
logits = mlp(observation_input, list(hidden_sizes) + [num_actions], tf.tanh, None)
actor = keras.Model(inputs=observation_input, outputs=logits)
value = tf.squeeze( mlp(observation_input, list(hidden_sizes) + [1], tf.tanh, None), axis=1 )
critic = keras.Model(inputs=observation_input, outputs=value)
policy_optimizer = keras.optimizers.Adam(learning_rate=policy_learning_rate)
value_optimizer = keras.optimizers.Adam(learning_rate=value_function_learning_rate)

保存未訓(xùn)練時(shí)的運(yùn)動(dòng)情況

在未訓(xùn)練模型之前，將模型控制游戲的情況保存是 gif ，可以看出來(lái)技術(shù)很糟糕，很快就結(jié)束了游戲。

import imageio
start = env.reset() 
frames = []
for t in range(steps_per_epoch):
    frames.append(env.render(mode='rgb_array'))
    start = start.reshape(1, -1)
    logits, action = sample_action(start)
    start, reward, done, _ = env.step(action[0].numpy())
    if done:
        break
with imageio.get_writer('未訓(xùn)練前的樣子.gif', mode='I') as writer:
    for frame in frames:
        writer.append_data(frame)

模型訓(xùn)練

這里主要是訓(xùn)練模型，執(zhí)行 eopch 輪，每一輪中循環(huán) steps_per_epoch 步，每一步就是根據(jù)當(dāng)前的觀(guān)測(cè)結(jié)果 observation 來(lái)抽樣得到下一步動(dòng)作，然后將得到的各種觀(guān)測(cè)結(jié)果、動(dòng)作、獎(jiǎng)勵(lì)、value 值、對(duì)數(shù)概率值保存在 buffer 對(duì)象中，待這一輪執(zhí)行游戲運(yùn)行完畢，收集了一輪的數(shù)據(jù)之后，就開(kāi)始訓(xùn)練策略和值函數(shù)，并打印本輪的訓(xùn)練結(jié)果，不斷重復(fù)這個(gè)過(guò)程，

observation, episode_return, episode_length = env.reset(), 0, 0
for epoch in tqdm(range(epochs)):
    sum_return = 0
    sum_length = 0
    num_episodes = 0
    for t in range(steps_per_epoch):
        if render:
            env.render()
        observation = observation.reshape(1, -1)
        logits, action = sample_action(observation)
        observation_new, reward, done, _ = env.step(action[0].numpy())
        episode_return += reward
        episode_length += 1
        value_t = critic(observation)
        logprobability_t = logprobabilities(logits, action)
        buffer.store(observation, action, reward, value_t, logprobability_t)
        observation = observation_new
        terminal = done
        if terminal or (t == steps_per_epoch - 1):
            last_value = 0 if done else critic(observation.reshape(1, -1))
            buffer.finish_trajectory(last_value)
            sum_return += episode_return
            sum_length += episode_length
            num_episodes += 1
            observation, episode_return, episode_length = env.reset(), 0, 0
    ( observation_buffer, action_buffer, advantage_buffer,  return_buffer, logprobability_buffer, ) = buffer.get()
    for _ in range(train_policy_iterations):
        kl = train_policy( observation_buffer, action_buffer, logprobability_buffer, advantage_buffer )
        if kl > 1.5 * target_kl:
            break
    for _ in range(train_value_iterations):
        train_value_function(observation_buffer, return_buffer)
    print( f"完成第 {epoch + 1} 輪訓(xùn)練， 平均獎(jiǎng)勵(lì): {sum_length / num_episodes}" )

打?。和瓿傻?1 輪訓(xùn)練， 平均獎(jiǎng)勵(lì): 30.864197530864196
完成第 2 輪訓(xùn)練， 平均獎(jiǎng)勵(lì): 40.32258064516129
...
完成第 9 輪訓(xùn)練， 平均獎(jiǎng)勵(lì): 185.1851851851852
完成第 11 輪訓(xùn)練， 平均獎(jiǎng)勵(lì): 172.41379310344828
...
完成第 14 輪訓(xùn)練， 平均獎(jiǎng)勵(lì): 172.41379310344828
...
完成第 18 輪訓(xùn)練， 平均獎(jiǎng)勵(lì): 185.1851851851852
...
完成第 20 輪訓(xùn)練， 平均獎(jiǎng)勵(lì): 200.0

保存訓(xùn)練后的運(yùn)動(dòng)情況

在訓(xùn)練模型之后，將模型控制游戲的情況保存是 gif ，可以看出來(lái)技術(shù)很嫻熟，可以在很長(zhǎng)的時(shí)間內(nèi)使得棒子始終保持近似垂直的狀態(tài)。

import imageio
start = env.reset()
frames = []
for t in range(steps_per_epoch):
    frames.append(env.render(mode='rgb_array'))
    start = start.reshape(1, -1)
    logits, action = sample_action(start)
    start, reward, done, _ = env.step(action[0].numpy())
    if done:
        break
with imageio.get_writer('訓(xùn)練后的樣子.gif', mode='I') as writer:
    for frame in frames:
        writer.append_data(frame)

到此，相信大家對(duì)“通過(guò)CartPole游戲詳解PPO優(yōu)化的方法”有了更深的了解，不妨來(lái)實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢(xún)，關(guān)注我們，繼續(xù)學(xué)習(xí)！