Actor Critic 方法

import os
os.environ["KERAS_BACKEND"] = "tensorflow"
import gym
import numpy as np
import keras
from keras import ops
from keras import layers
import tensorflow as tf

# 整个设置的配置参数
seed = 42
gamma = 0.99  # 过去奖励的折扣因子
max_steps_per_episode = 10000
env = gym.make("CartPole-v0")  # 创建环境
env.seed(seed)
eps = np.finfo(np.float32).eps.item()  # 最小的数，使得 1.0 + eps != 1.0

num_inputs = 4
num_actions = 2
num_hidden = 128

inputs = layers.Input(shape=(num_inputs,))
common = layers.Dense(num_hidden, activation="relu")(inputs)
action = layers.Dense(num_actions, activation="softmax")(common)
critic = layers.Dense(1)(common)

model = keras.Model(inputs=inputs, outputs=[action, critic])

optimizer = keras.optimizers.Adam(learning_rate=0.01)
huber_loss = keras.losses.Huber()
action_probs_history = []
critic_value_history = []
rewards_history = []
running_reward = 0
episode_count = 0

while True:  # 运行直到解决
    state = env.reset()
    episode_reward = 0
    with tf.GradientTape() as tape:
        for timestep in range(1, max_steps_per_episode):
            # env.render(); 添加这一行将显示代理的尝试
            # 在弹出窗口中。

            state = ops.convert_to_tensor(state)
            state = ops.expand_dims(state, 0)

            # 从环境状态预测动作概率和估计的未来奖励
            action_probs, critic_value = model(state)
            critic_value_history.append(critic_value[0, 0])

            # 从动作概率分布中采样动作
            action = np.random.choice(num_actions, p=np.squeeze(action_probs))
            action_probs_history.append(ops.log(action_probs[0, action]))

            # 在我们的环境中应用采样的动作
            state, reward, done, _ = env.step(action)
            rewards_history.append(reward)
            episode_reward += reward

            if done:
                break

        # 更新运行奖励以检查解决条件
        running_reward = 0.05 * episode_reward + (1 - 0.05) * running_reward

        # 计算奖励的期望值
        # - 在每个时间步，给定时间步后收到的总奖励
        # - 过去的奖励通过与gamma相乘进行折扣
        # - 这些是我们的评论员的标签
        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)
        returns = (returns - np.mean(returns)) / (np.std(returns) + eps)
        returns = returns.tolist()

        # 计算损失值以更新我们的网络
        history = zip(action_probs_history, critic_value_history, returns)
        actor_losses = []
        critic_losses = []
        for log_prob, value, ret in history:
            # 在历史的这一点上，评论员估计我们将会得到一个
            # 在未来的总奖励 = `value`。我们采取了一个动作，日志概率为
            # `log_prob`，最终获得的总奖励 = `ret`。
            # 演员必须更新，以便预测一个高概率的动作，从而导致
            # 高奖励（与评论员的估计相比）。
            diff = ret - value
            actor_losses.append(-log_prob * diff)  # 演员损失

            # 评论员必须进行更新，以便预测更好的
            # 未来奖励的估计。
            critic_losses.append(
                huber_loss(ops.expand_dims(value, 0), ops.expand_dims(ret, 0))
            )

        # 反向传播
        loss_value = sum(actor_losses) + sum(critic_losses)
        grads = tape.gradient(loss_value, model.trainable_variables)
        optimizer.apply_gradients(zip(grads, model.trainable_variables))

        # 清除损失和奖励历史
        action_probs_history.clear()
        critic_value_history.clear()
        rewards_history.clear()

    # 记录细节
    episode_count += 1
    if episode_count % 10 == 0:
        template = "运行奖励: {:.2f} 在第 {} 集"
        print(template.format(running_reward, episode_count))

    if running_reward > 195:  # 考虑任务已解决的条件
        print("在第 {} 集解决!".format(episode_count))
        break