Policy Gradient

策略梯度法与值函数近似法的区别:

值函数近似法:在值函数近似法中，动作选择的策略是不变的，如固定使用贪婪算法作为策略选择方法。即在时间步的状态下选择动作的方式是固定的。
策略梯度法:在策略梯度法中，智能体会学习不同的策略。即在某个时间步的状态下，根据动作的概率分布进行选择，且该动作概率分布可能会被不断地调整。

相比于值函数近似法，基于策略求解方法的策略梯度法能够让智能体在学习的过程中学会不同的策略，进而根据某状态的动作概率分布，选择将要执行的动作。图7.2直观地展示了在GridWorld游戏上，两种学习方法最终的表现差异。

图7.2a为GridWorld环境，目标是使得智能体能够从格子S走到格子G。其中，格子X为陷阱，格子O为可行走区域。图7.2b为值函数近似法的求解结果，即通过贪婪策略选择最大的动作值作为策略；最终，智能体只能获得1条从S达G的路径（每个格子中的箭头表示所选择的动作）。图7.2c为策略梯度法的求解结果，智能体最终能够获得3条从S到达G的路径。例如，在初始状态格子S上就存在两种分属不同概率的动作可供智能体选择。由图7.2b和图7.2c对比可知，值函数近似法学习到策略是固定的，而策略梯度法能够为智能体学习到多种策略。

image.png
策略梯度法的优点：

• 易收敛：策略梯度法具有更好的收敛性。因为学习训练过程中，策略梯度法每次更新策略函数时，参数只发生细微的变化，但参数的变化却是朝着正确的方向进行迭代，这使得算法具有更好的收敛性。而价值函数在学习训练的后期，参数会围绕着最优值附近持续小幅度地波动，导致算法难收敛。
• 能高效处理连续动作空间的任务：值函数近似法不适用于连续动作空间的强化学习任务。因为对于拥有高维度或者连续状态空间的任务，基于价值的求解方法在得到价值函数后，需要比较某时间步的状态s中相关动作对应的价值大小如果此时动作空间维度较高或者动作空间是一个无限的集合，从中计算最大动作值函数将会异常困难。而策略梯度为概率密度函数，实际的输出是一个实值，从而大大降低了计算复杂度。
• 能学习随机策略：在具有显著函数逼近的问题中，最优策略可能是随机策略（如石头剪刀布游戏中，如果按照固定策略反而容易输掉比赛，随机策略却不失为一种更好的策略）。而在值函数近似法中，最后利用贪婪策略选择价值最大的值作为动作值输出，这导致其每次的输出值都是固定的，难以学习到随机策略。而策略梯度通过 Softmax策略和高斯策略引入随机过程，这使得智能体能够学习出可能是随机策略的最优近似策略。

策略梯度法的缺点：

• 通常收敛到局部最优解，而非全局最优解：由于策略梯度基于连续的任务，动作空间可能是无限的，因此很多时候不一定能够找到全局最优解。值得注意的是，在策略梯度法的优点中提到的策略梯度易收敛到最优解，但该最优解不一定是全局最优解。
• 策略评估效率低下：策略梯度在每次更新参数时，在梯度改变的方向都进行小幅度修正。优势是使得学习过程较为平滑，但伴随而来的是降低了策略评估的学习效率。
• 方差较高：策略梯度学习过程较为缓慢，具有更高的可变性（随机性），这导致智能体在探索过程中会出现较多无效的尝试，使整体策略值方差较高。

import matplotlib.pyplot as plt
import gym
import numpy as np
from tensorflow.keras import models, layers, optimizers

env = gym.make('CartPole-v0')

STATE_DIM, ACTION_DIM = 4, 2
model = models.Sequential([
    layers.Dense(100, input_dim=STATE_DIM, activation='relu'),
    layers.Dropout(0.1),    # 一开始的数据质量不高，防止过拟合
    layers.Dense(ACTION_DIM, activation="softmax")
])
# loss使用cross entropy会不会更好？
model.compile(loss='mean_squared_error',
              optimizer=optimizers.Adam(0.001))


def choose_action(s):
    """预测动作"""
    prob = model.predict(np.array([s]))[0]
    return np.random.choice(len(prob), p=prob)


def discount_rewards(rewards, gamma=0.95):
    """计算衰减reward的累加期望，并中心化和标准化处理"""
    discount_rewards = np.zeros_like(rewards, dtype=np.float32)
    cumulative = 0.
    for i in reversed(range(len(rewards))):
        cumulative = cumulative * self.gamma + rewards[i]
        discount_rewards[i] = cumulative

    # normalization,有利于控制梯度的方差
    discount_rewards -= np.mean(discount_rewards)
    discount_rewards //= np.std(discount_rewards)
    return discount_rewards

def train(records):
    """ 
    这里没有使用随机sample固定的batch，测试发现那样效果并不好，可能是因为
    Policy Gradient方法中其它episode的经验对本次episode无效，所以每个episode
    都重新累积经验。
    """
    s_batch = np.array([record[0] for record in records])
    # action 独热编码处理，方便求动作概率，即 prob_batch
    a_batch = np.array([[1 if record[1] == i else 0 for i in range(ACTION_DIM)]
                        for record in records])
    # 假设predict的概率是 [0.3, 0.7]，选择的动作是 [0, 1]
    # 则动作[0, 1]的概率等于 [0, 0.7] = [0.3, 0.7] * [0, 1]
    prob_batch = model.predict(s_batch) * a_batch
    r_batch = discount_rewards([record[2] for record in records])

    model.fit(s_batch, prob_batch, sample_weight=r_batch, verbose=0)


episodes = 2000  # 至多2000次
score_list = []  # 记录所有分数
for i in range(episodes):
    s = env.reset()
    score = 0
    replay_records = []
    while True:
        a = choose_action(s)
        next_s, r, done, _ = env.step(a)
        replay_records.append((s, a, r))

        score += r
        s = next_s
        if done:
            train(replay_records)
            score_list.append(score)
            print('episode:', i, 'score:', score, 'max:', max(score_list))
            break
    # 最后10次的平均分大于 195 时，停止并保存模型
    if np.mean(score_list[-10:]) > 195:
        model.save('CartPole-v0-pg.h5')
        break
env.close()


# 画图
plt.plot(score_list)
x = np.array(range(len(score_list)))
smooth_func = np.poly1d(np.polyfit(x, score_list, 3))
plt.plot(x, smooth_func(x), label='Mean', linestyle='--')
plt.show()

result.png

参考：
《深度强化学习：原理与实践》陈仲铭/何明 著
TensorFlow 2.0 (九) - 强化学习 70行代码实战 Policy Gradient