强化学习三个组成部分：

• Actor
• Environment
• Reward Function

在强化学习中，

• 环境跟奖励函数是在开始学习之前事先给定的，不受你控制。
• 你唯一能做的是调整Actor的策略（Policy），使Actor得到最大Reward。
• Policy根据外界的输入，输出Actor要执行的行为。

Policy of Actor

策略是带有参数的网络：

• 输入：机器的观察，表示为向量或矩阵；
• 输出：与动作相关的网络输出层神经元。

以电玩游戏为例：

• 游戏初始画面记为
• Actor经过内部网络决定执行第一个动作记为
• 第一次动作执行完后得到的奖励为，游戏更新新的画面
• Actor继续执行第二次动作，得到奖励
• 如此类推，直至判定游戏结束为止。

Actor, Environment, Reward

一场游戏或一个回合（episode）中，环境输出和Actor的动作可以表示成一个轨迹（Trajectory）：

如果Actor的策略网络参数已经给定，某个轨迹发生的概率可以表示为：

该概率取决于两个部分：

• Environment：
• Actor：

奖励函数则根据某一状态采取的某个动作，决定这个行为可获得多少分数。

• 奖励是一个随机变量
• 给定参数，可以计算奖励的期望：
  
• 从这个分布采样一个轨迹，然后计算 的期望值，我们要做的事情就是最大化这个期望奖励。

策略梯度（Policy Gradient）

• 最大化期望奖励：梯度提升（Gradient Ascent）
• 利用公式：
  
  有：
  
• 采样条轨迹 ， 计算每一条轨迹的，然后把它们全部加起来近似地求得梯度。就可以更新策略参数。
  
• 计算，环境部分和与无关，梯度为0，则有：
  
• 综合起来得到：
  
  
  策略梯度更新
• 采样的数据就只会用一次。

Tips

Tip 1：添加基线（Add a Baseline）

• 很多游戏里面，奖励总是正的
• 没有采样到的动作权重会变小
• 改进：引入基线：
  

Tip 2：给每一个动作合适评价（Assign Suitble Credit）

• 所有的”状态-动作对”都会使用同样的奖励项进行加权
• 权重应反映每个动作的好坏
• 改进：
  • 只计算从这个动作执行以后所得到的奖励
  • 折扣回报（Discounted Return）：给未来的奖励做一个折扣
    
• 优势函数（Advantage Function），相对的优势。

REINFORCE：蒙特卡罗策略梯度（Monte Carlo Policy Gradient）

蒙特卡罗策略梯度算法

实现REINFORCE

import numpy as np
import matplotlib.pyplot as plt
import gym
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.autograd import Variable

GAMMA = 0.9


class PolicyNetwork(nn.Module):
    def __init__(self, num_inputs, num_actions, hidden_size, learning_rate=3e-4):
        super().__init__()

        self.num_actions = num_actions
        self.linear1 = nn.Linear(num_inputs, hidden_size)
        self.linear2 = nn.Linear(hidden_size, num_actions)
        self.optimizer = optim.Adam(self.parameters(), lr=learning_rate)

    def forward(self, state):
        x = F.relu(self.linear1(state))
        x = F.softmax(self.linear2(x), dim=1)
        return x

    def get_action(self, state):
        state = torch.from_numpy(state).float().unsqueeze(0)
        probs = self.forward(Variable(state))
        highest_prob_action = np.random.choice(
            self.num_actions, p=np.squeeze(probs.detach().numpy())
        )
        log_prob = torch.log(probs.squeeze(0)[highest_prob_action])
        return highest_prob_action, log_prob


def update_policy(policy_network, rewards, log_probs):
    discounted_rewards = []

    for t in range(len(rewards)):
        Gt = 0
        pw = 0
        for r in rewards[t:]:
            Gt = Gt + GAMMA**pw * r
            pw += 1
        discounted_rewards.append(Gt)

    discounted_rewards = torch.tensor(discounted_rewards)
    discounted_rewards = (discounted_rewards - discounted_rewards.mean()) / (
        discounted_rewards.std() + 1e-9
    )

    policy_gradient = []
    for log_prob, Gt in zip(log_probs, discounted_rewards):
        policy_gradient.append(-log_prob * Gt)

    policy_network.optimizer.zero_grad()
    policy_gradient = torch.stack(policy_gradient).sum()
    policy_gradient.backward()
    policy_network.optimizer.step()


def main():
    env = gym.make("CartPole-v0")
    policy_net = PolicyNetwork(
        num_inputs=env.observation_space.shape[0],
        num_actions=env.action_space.n,
        hidden_size=128,
    )

    max_episode_num = 5000
    max_steps = 10000
    numsteps = []
    avg_numsteps = []
    all_rewards = []

    for episode in range(max_episode_num):
        state = env.reset()
        log_probs = []
        rewards = []

        for steps in range(max_steps):
            env.render()
            action, log_prob = policy_net.get_action(state)
            new_state, reward, done, _ = env.step(action)
            log_probs.append(log_prob)
            rewards.append(reward)

            if done:
                update_policy(policy_net, rewards, log_probs)
                numsteps.append(steps)
                avg_numsteps.append(np.mean(numsteps[-10:]))
                all_rewards.append(np.sum(rewards))
                if episode % 1 == 0:
                    print(
                        "episode: {:4}, total reward: {:5.1f}, average_reward: {:5.1f}, length: {:4}".format(
                            episode,
                            np.sum(rewards),
                            np.mean(all_rewards[-10:]),
                            steps,
                        )
                    )
                break

            state = new_state

    plt.plot(numsteps)
    plt.plot(avg_numsteps)
    plt.xlabel("Episode")
    plt.show()


if __name__ == "__main__":
    main()

策略梯度训练过程

训练结果

参考资料

Deriving Policy Gradients and Implementing REINFORCE