深度确定性策略梯度 (DDPG)

import os

os.environ["KERAS_BACKEND"] = "tensorflow"

import keras
from keras import layers

import tensorflow as tf
import gymnasium as gym
import numpy as np
import matplotlib.pyplot as plt

# 指定`render_mode`参数以在弹出窗口中显示代理的尝试。
env = gym.make("Pendulum-v1", render_mode="human")

num_states = env.observation_space.shape[0]
print("状态空间的大小 ->  {}".format(num_states))
num_actions = env.action_space.shape[0]
print("动作空间的大小 ->  {}".format(num_actions))

upper_bound = env.action_space.high[0]
lower_bound = env.action_space.low[0]

print("动作的最大值 ->  {}".format(upper_bound))
print("动作的最小值 ->  {}".format(lower_bound))

class OUActionNoise:
    def __init__(self, mean, std_deviation, theta=0.15, dt=1e-2, x_initial=None):
        self.theta = theta
        self.mean = mean
        self.std_dev = std_deviation
        self.dt = dt
        self.x_initial = x_initial
        self.reset()

    def __call__(self):
        # 公式来源于 https://www.wikipedia.org/wiki/Ornstein-Uhlenbeck_process
        x = (
            self.x_prev
            + self.theta * (self.mean - self.x_prev) * self.dt
            + self.std_dev * np.sqrt(self.dt) * np.random.normal(size=self.mean.shape)
        )
        # 将x存入x_prev
        # 使下一个噪声依赖于当前噪声
        self.x_prev = x
        return x

    def reset(self):
        if self.x_initial is not None:
            self.x_prev = self.x_initial
        else:
            self.x_prev = np.zeros_like(self.mean)

class Buffer:
    def __init__(self, buffer_capacity=100000, batch_size=64):
        # 最多存储的“经历”数
        self.buffer_capacity = buffer_capacity
        # 用于训练的元组数量
        self.batch_size = batch_size

        # 它告诉我们 record() 被调用的次数。
        self.buffer_counter = 0

        # 不使用元组列表作为经验重放概念
        # 我们使用不同的 np.arrays 为每个元组元素
        self.state_buffer = np.zeros((self.buffer_capacity, num_states))
        self.action_buffer = np.zeros((self.buffer_capacity, num_actions))
        self.reward_buffer = np.zeros((self.buffer_capacity, 1))
        self.next_state_buffer = np.zeros((self.buffer_capacity, num_states))

    # 以 (s,a,r,s') 观测元组作为输入
    def record(self, obs_tuple):
        # 如果超出 buffer_capacity，将索引设置为零，
        # 替换旧记录
        index = self.buffer_counter % self.buffer_capacity

        self.state_buffer[index] = obs_tuple[0]
        self.action_buffer[index] = obs_tuple[1]
        self.reward_buffer[index] = obs_tuple[2]
        self.next_state_buffer[index] = obs_tuple[3]

        self.buffer_counter += 1

    # 默认情况下，TensorFlow 2 已开启急切执行。使用 tf.function 装饰可以
    # 使 TensorFlow 从我们函数中的逻辑和计算构建一个静态图。
    # 这为包含许多小的 TensorFlow 操作的代码块提供了巨大的速度提升，比如这个。
    @tf.function
    def update(
        self,
        state_batch,
        action_batch,
        reward_batch,
        next_state_batch,
    ):
        # 训练和更新演员和评论网络。
        # 见伪代码。
        with tf.GradientTape() as tape:
            target_actions = target_actor(next_state_batch, training=True)
            y = reward_batch + gamma * target_critic(
                [next_state_batch, target_actions], training=True
            )
            critic_value = critic_model([state_batch, action_batch], training=True)
            critic_loss = keras.ops.mean(keras.ops.square(y - critic_value))

        critic_grad = tape.gradient(critic_loss, critic_model.trainable_variables)
        critic_optimizer.apply_gradients(
            zip(critic_grad, critic_model.trainable_variables)
        )

        with tf.GradientTape() as tape:
            actions = actor_model(state_batch, training=True)
            critic_value = critic_model([state_batch, actions], training=True)
            # 使用 `-value` 是因为我们希望最大化评论者给我们动作的值
            actor_loss = -keras.ops.mean(critic_value)

        actor_grad = tape.gradient(actor_loss, actor_model.trainable_variables)
        actor_optimizer.apply_gradients(
            zip(actor_grad, actor_model.trainable_variables)
        )

    # 我们计算损失并更新参数
    def learn(self):
        # 获取采样范围
        record_range = min(self.buffer_counter, self.buffer_capacity)
        # 随机采样索引
        batch_indices = np.random.choice(record_range, self.batch_size)

        # 转换为张量
        state_batch = keras.ops.convert_to_tensor(self.state_buffer[batch_indices])
        action_batch = keras.ops.convert_to_tensor(self.action_buffer[batch_indices])
        reward_batch = keras.ops.convert_to_tensor(self.reward_buffer[batch_indices])
        reward_batch = keras.ops.cast(reward_batch, dtype="float32")
        next_state_batch = keras.ops.convert_to_tensor(
            self.next_state_buffer[batch_indices]
        )

        self.update(state_batch, action_batch, reward_batch, next_state_batch)


# 这个更新目标参数缓慢
# 基于比一小得多的速率 `tau`。
def update_target(target, original, tau):
    target_weights = target.get_weights()
    original_weights = original.get_weights()

    for i in range(len(target_weights)):
        target_weights[i] = original_weights[i] * tau + target_weights[i] * (1 - tau)

    target.set_weights(target_weights)

def get_actor():
    # 初始化权重在 -3e-3 和 3e-3 之间
    last_init = keras.initializers.RandomUniform(minval=-0.003, maxval=0.003)

    inputs = layers.Input(shape=(num_states,))
    out = layers.Dense(256, activation="relu")(inputs)
    out = layers.Dense(256, activation="relu")(out)
    outputs = layers.Dense(1, activation="tanh", kernel_initializer=last_init)(out)

    # 我们的上限是 2.0，对于 Pendulum。
    outputs = outputs * upper_bound
    model = keras.Model(inputs, outputs)
    return model


def get_critic():
    # 状态作为输入
    state_input = layers.Input(shape=(num_states,))
    state_out = layers.Dense(16, activation="relu")(state_input)
    state_out = layers.Dense(32, activation="relu")(state_out)

    # 动作作为输入
    action_input = layers.Input(shape=(num_actions,))
    action_out = layers.Dense(32, activation="relu")(action_input)

    # 两者在拼接之前经过独立层
    concat = layers.Concatenate()([state_out, action_out])

    out = layers.Dense(256, activation="relu")(concat)
    out = layers.Dense(256, activation="relu")(out)
    outputs = layers.Dense(1)(out)

    # 输出给定状态-动作的单个值
    model = keras.Model([state_input, action_input], outputs)

    return model

def policy(state, noise_object):
    sampled_actions = keras.ops.squeeze(actor_model(state))
    noise = noise_object()
    # 添加噪声到动作
    sampled_actions = sampled_actions.numpy() + noise

    # 我们确保动作在范围内
    legal_action = np.clip(sampled_actions, lower_bound, upper_bound)

    return [np.squeeze(legal_action)]

std_dev = 0.2
ou_noise = OUActionNoise(mean=np.zeros(1), std_deviation=float(std_dev) * np.ones(1))

actor_model = get_actor()
critic_model = get_critic()

target_actor = get_actor()
target_critic = get_critic()

# 初始时使权重相等
target_actor.set_weights(actor_model.get_weights())
target_critic.set_weights(critic_model.get_weights())

# Actor-Critic 模型的学习率
critic_lr = 0.002
actor_lr = 0.001

critic_optimizer = keras.optimizers.Adam(critic_lr)
actor_optimizer = keras.optimizers.Adam(actor_lr)

total_episodes = 100
# 未来奖励的折扣因子
gamma = 0.99
# 用于更新目标网络
tau = 0.005

buffer = Buffer(50000, 64)

# 用于存储每个剧集的奖励历史
ep_reward_list = []
# 用于存储最近几次剧集的平均奖励历史
avg_reward_list = []

# 训练大约需要 4 分钟
for ep in range(total_episodes):
    prev_state, _ = env.reset()
    episodic_reward = 0

    while True:
        tf_prev_state = keras.ops.expand_dims(
            keras.ops.convert_to_tensor(prev_state), 0
        )

        action = policy(tf_prev_state, ou_noise)
        # 从环境中接收状态和奖励。
        state, reward, done, truncated, _ = env.step(action)

        buffer.record((prev_state, action, reward, state))
        episodic_reward += reward

        buffer.learn()

        update_target(target_actor, actor_model, tau)
        update_target(target_critic, critic_model, tau)

        # 当 `done` 或 `truncated` 为 True 时结束此剧集
        if done or truncated:
            break

        prev_state = state

    ep_reward_list.append(episodic_reward)

    # 最近 40 个剧集的平均值
    avg_reward = np.mean(ep_reward_list[-40:])
    print("剧集 * {} * 平均奖励是 ==> {}".format(ep, avg_reward))
    avg_reward_list.append(avg_reward)

# 绘制图表
# 剧集与平均奖励
plt.plot(avg_reward_list)
plt.xlabel("剧集")
plt.ylabel("平均剧集奖励")
plt.show()