这是一份给算法同学的强化学习入门材料

Part 1：基本概念

概念

强化学习

1. 强化学习关注与智能体(agent)如何与环境交互中不断学习以完成特定的目标；
2. 与有监督学习相比，不需要告诉智能体数据以及对应的标签，学习相应的模型，而是需要智能体在环境中一次次学习(哪些数据对应哪些标签)，从而学习规律知道策略；
3. 强化学习是希望智能体在环境中根据当前状态，采取行动，转移到下一个状态，获得回报。不断进行这样的过程，从而学习到一个策略(状态到动作的映射，即当前状态下，采取什么样的行动，能使得我最终获得的回报最大【不仅只是当前状态的而回报，一个策略的长期影响才是至关重要的】)

交互对象

• 智能体(agent)：可以感知外界环境的状态(state)和反馈的奖励(reward)，并进行学习和决策．智能体的决策功能是指根据外界环境的状态来做出不同的动作(action)，而学习功能是指根据外界环境的奖励来调整策略(policy)；
• 环境(environment)：是智能体外部的所有事物，并受智能体动作的影响而改变其状态，并反馈给智能体相应的奖励。

基本要素

基础概念定义

在步骤$t$

• 状态(state)：对环境的描述，$s_t$

• 动作(action)：对智能体行为的描述，$a_t$

• 策略(policy)：是一组概率分布，表示每个动作的概率，$\pi(a|s)$

• 奖励(reward)：智能体做出动作$a_t$后，更新到状态$s_{t+1}$，并环境给出奖励$r_t$评估此时刻智能体动作的好坏。奖励的作用是使得智能体能在相同的状态下做出动作的修正，以使得它能够更好地去适应环境，奖励的设计会决定游戏的公平和智能体是否能够通过游戏

• 回报(return)：智能体在某状态下，未来多个奖励状态的总和。$t$时刻的回报是当前时刻的奖励加上后续时刻奖励的总和，并且越是后续时刻的奖励对当前回报的作用也就越小，可以使用衰减因子$\gamma$对$t$时刻以后的奖励进行加权，$g_t$

  $$g_t = r_t + \gamma r_{t+1} + \gamma^2 r_{t+2} + \cdots = \sum_{k=0}^N \gamma^k r_{t+k}$$

  有递归式：

  $$\begin{aligned} g_t &= r_t + \gamma r_{t+1} + \gamma^2 r_{t+2} + \cdots \\ &= r_t + \gamma (r_{t+1} + \gamma r_{t+2} + \cdots) \\ &= r_t + \gamma g_{t+1} \end{aligned}$$

• 状态价值函数(state-value function)：V值，是从状态$s_t$出发，遵循策略$\pi$所能获得的回报的期望值，即

  $$V^\pi(s_t) = E_\pi[G|S=s_t]$$

• 动作价值函数(action-value function)：Q值，是在当前状态$s_t$，执行动作$a_t$后，环境遵循状态转移概率$p(s_{t+1} | s_t, a_t)$更新到状态$s_{t+1}$，并给出奖励$r_t$（实际上，$r_t$某种程度上是与$s_{t+1}$相关的），遵循策略$\pi$所能获得的回报的期望值，即

  $$Q^\pi(s_t, a_t) = E_\pi[G|S=s_t, A=a_t]$$

  可以用动作价值函数判断$t$时刻价值最高的动作，即

  $$a^* = \argmax_a Q(s, a)$$

• 优势函数(advantage function)：表示状态$s_t$处，动作$a_t$相对于平均水平的高低，评价当前动作值函数相对于平均值的大小。这里的优势指的是动作值函数相比于当前状态的值函数的优势。如果优势函数大于零，则说明该动作比平均动作好，如果优势函数小于零，则说明当前动作还不如平均动作好。

  $$A^\pi(s_t, a_t) = Q^\pi(s_t, a_t) - V^\pi(s_t)$$

状态价值函数和动作价值函数的关系

一个状态的$V$值，就是这个状态$s_t$下的所有动作$a_t \in A$的$Q$值在策略$\pi$下的期望，有

$$V^\pi(s_t) = E_{a_t \sim \pi(a_t|s_t)} Q^\pi(s_t, a_t) = \sum_{a_t \in A} \pi(a_t|s_t) \cdot Q^\pi(s_t, a_t)$$

一个动作的Q值，是在状态$s_t$下采取动作$a_t$后，获得的回报的期望，记状态转移概率为$p(s|s_t, a_t)$，有

$$Q^\pi(s_t, a_t) = E_{s_{t+1} \sim p(s|s_t, a_t)} \left( r + V^\pi(s_{t+1}) \right) = r_t + \sum_{s_{t+1} \in S} p(s_{t+1}|s_t, a_t) V^\pi(s_{t+1})$$

注意到，$r_t$也计入$Q^\pi(s_t, a_t)$的计算，而且其实是与$s_{t+1}$相关的。实际使用时，一般加上折扣率$\gamma$，对历史的回报进行衰减

$$Q^\pi(s_t, a_t) = r_t + \gamma \sum_{s_{t+1} \in S} p(s_{t+1}|s_t, a_t) V^\pi(s_{t+1})$$

上面的几个定义式理解起来比较抽象，举个例子

假如每一步的状态空间是$S = \{s_a, s_b, s_c, s_d\}$，动作空间是$A = \{a_x, a_y, a_z\}$，通过两次探索得到了两条动作序列：

• 序列一：$(s^1_1=s_a) \rightarrow^{a^1_1=a_x} (s^1_2=s_b/r^1_1) \rightarrow^{a^1_2=a_y} (s^1_3=s_c/r^1_2) \rightarrow^{a^1_3=a_z} (s^1_4=s_d/r^1_3)$，结果是胜利
• 序列二：$(s^2_1=s_a) \rightarrow^{a^2_1=a_x} (s^2_2=s_b/r^2_1) \rightarrow^{a^2_2=a_z} (s^2_3=s_d/r^2_2)$，结果是失败

奖励函数这样设置：如果最终胜利，那么$r_{t} = 1$，否则$r_{t} = 0$，那么有：

1. 状态$s_b$可以通过动作$a_y, a_z$转移到两个不同的下一状态$s_c, s_d$，每个动作的概率都是$\pi(s_c|s_b) = \pi(s_d|s_b) = 0.5$；
2. 在状态$s_b$，选择动作$a_y$的最终回报是$g^1_2 = (r^1_2 + 0.9 \times r^1_3) = 1.9$，根据定义，$s_b$下$a_y$的Q值是$Q^\pi(s_b, a_y) = 1.9$；
3. 在状态$s_b$，选择动作$a_z$的最终回报是$g^2_2 = r^2_2 = 0$，根据定义，$s_b$下$a_z$的Q值是$Q^\pi(s_b, a_z) = 0$；
4. 状态$s_b$的V值是所有可选动作的Q值的期望，也就是$V^\pi(s_b) = \pi(s_c|s_b) \times Q^\pi(s_b, a_y) + \pi(s_d|s_b) \times Q^\pi(s_b, a_z) = 0.475$；
5. 那么在状态$s_b$时，动作$a_y$的优势函数$A^{\pi}(s_b, a_y) = Q^\pi(s_b, a_y) - V^\pi(s_b) = 0.475$，动作$a_z$的优势函数$A^{\pi}(s_b, a_z) = Q^\pi(s_b, a_z) - V^\pi(s_b) = -0.475$，也就是说动作$a_y$优势比$a_z$更大。

分类

value-based & policy-based

• value-based：训练$Q(s, a)$，测试时基于$s$选择使Q值最大的$a$，如Q-Learning、SARSA、DQN
• policy-based：训练$p(s, a)$，测试时基于$s$得到不同$a$的概率，选择概率最大的$a$，如policy-gradient
• 也有将两种方法结合，如actor-critic

on-policy & off-policy

• on-policy：行动策略和评估策略相同，需要学习的Agent和训练过程中和环境进行交互的Agent是同一个，如SARSA
• off-policy：行动策略和评估策略不相同，需要学习的Agent和训练过程中真正和环境进行交互的Agent不是同一个，如Q-Learning

model-based & model-free

model-based相对于model-free的最主要区别是引入了对环境的建模。这里提到的建模是指我们通过监督训练来训练一个环境模型，其数据是算法和环境的实际交互数据$(s_t, a_t, r_t, s_{t+1}, a_{t+1}, r_{t+1}, \cdots)$，是在给定$s_t$、$a_t$下预测下一个状态$s_{t+1}$。

• model-based：使用环境模型（环境的动态特性，即期望收益和状态转移概率）和规划（在真正经历之前，先考虑未来可能发生的各种情境从而预先决定采取何种动作）来解决强化学习问题的方法。
• model-free：：通过学习（直接地试错）经验（在与环境交互中采样得到的状态、动作、收益序列）来解决强化学习问题的方法。

在agent执行它的动作之前，它是否能对下一步的状态和回报做出预测，如果可以，那么就是model-based方法（model based方法就好比人类对环境的转移有一个初步的预估，所以plan了一个更好的action），如果不能，即为model-free方法。

offline reinforcement learning

离线强化学习，即用大量过往数据进行学习，没有交互环境参与。

Part 2: 从Q-Learning到DQN

Q-Learning

Q-Learning是根据所经历的状态和所选择的行为建立一张Q表格(Q-Table)，根据每一轮学习到的奖励更新Q表格。Q-Table即以状态为行、动作为列建立的表格，存放Q值。问题在于，如何求取Q-Table中的Q值。

状态\动作	$a_0$	$a_1$	$a_2$	$\cdots$
$s_0$
$s_1$
$s_1$
$\cdots$

用$s$表示当前状态，$a$是该步采取的动作，得到的奖励是$r$，下一步状态$s'$，伪代码为

Initialize Q(s, a) arbitrarily
Repeat (for each episode):
    Initialize s
    Repeat (for each step of episode):
        Choose a from s using policy derived from Q (e.g. \epsilon-greedy)
        Take action a, observe r, s'
        Q(s, a) \leftarrow Q(s, a) + \alpha \left[ r + \gamma \max_{a'} Q(s', a') - Q(s, a) \right]
        s \leftarrow s'
    until s is terminal

其中，$\epsilon-greedy$是指，根据概率值$p$随机采样决定下一步是否根据$Q(s, a)$选择下一步动作$a$。这种做法的出发点在于，初始阶段是累积经验的阶段，随机地探索环境往往比固定的行为模式要好，我们希望探索者不会那么贪婪(greedy)。$\epsilon$就是用来控制贪婪程度的值（以$\epsilon$几率选择最优，以$1 - \epsilon$几率随机探索），$\epsilon$可以随着探索时间不断提升(越来越贪婪)，即

$$a = \begin{cases} \argmax_{a' \in A} Q(s, a') & p < \epsilon \\ \text{random}_{a' \in A} a' & \text{otherwise} \end{cases}$$

按时间步展开，图例如下，注意在时刻$t$时四元组$(s, a, s', r)$均为已知量

参数更新公式如下，$\alpha$是学习率

$$Q(s, a) \leftarrow Q(s, a) + \alpha \left[ \underline{r + \gamma \max_{a'} Q(s', a')} - Q(s, a) \right]$$

根据Q值选择每步的最佳动作，也就是$a' = \argmax_a Q(s', a)$，那么$\max_{a'} Q(s', a')$是下一状态$s'$下，在能选择的所有动作$a' \in A$中，能拿到的最大Q值。所以$r + \gamma \max_{a'} Q(s', a')$可以视作预测值$Q(s, a)$的真实值，通过计算两者偏差来逐步修正。

下面的Q-Learning例程，是智能体在长度为N_STATES的一维空间中探索的例子，当N_STATES=6该空间表示为-----T。智能体从最左侧出发，即o----T，探索一条路线到达终点T。Q-Table设置为

位置(s)\方向(a)	left	right
0
1
2
3
4
5(T)

Q-Learning例程：是智能体在长度为N_STATES的一维空间中探索

import numpy as np
import pandas as pd
import time

np.random.seed(42)

N_STATES = 6                # 1维世界的宽度(-----T)
ACTIONS = ['left', 'right'] # 探索者的可用动作
EPSILON = 0.9               # 贪婪度 greedy
ALPHA = 0.1                 # 学习率
GAMMA = 0.9                 # 奖励递减值
MAX_EPISODES = 10           # 最大回合数
FRESH_TIME = 0.3            # 移动间隔时间


def build_q_table(n_states, actions):
    """ 新建Q表格，Q(s, a)表示在位置s处采取a行为的行为值 """
    table = pd.DataFrame(
        np.zeros((n_states, len(actions))),  # q_table 全 0 初始
        columns=actions,  # columns 对应的是行为名称
    )
    return table


# q_table:
"""
   left  right
0   0.0    0.0
1   0.0    0.0
2   0.0    0.0
3   0.0    0.0
4   0.0    0.0
5   0.0    0.0
"""


# 在某个 state 地点, 选择行为
def choose_action(state, q_table):
    """ 以\epsilon-greedy策略，选择当前s处选择的动作a 
    
    以90%概率贪婪选择，10%概率随机选择
    """
    state_actions = q_table.iloc[state, :]  # 选出这个 state 的所有 action 值
    if (np.random.uniform() > EPSILON) or (state_actions.any() == 0):  # 非贪婪 or 或者这个 state 还没有探索过
        action_name = np.random.choice(ACTIONS)
    else:
        action_name = state_actions.idxmax()  # 贪婪模式
    return action_name


def get_env_feedback(S, A):
    """ 在位置s处采取动作a，求取状态s'、奖励r """
    # This is how agent will interact with the environment
    if A == 'right':  # move right
        if S == N_STATES - 2:  # terminate：目前在s=4的位置，再向右移动1，到达s=5(T)
            S_ = 'terminal'
            R = 1
        else:
            S_ = S + 1
            R = 0
    else:  # move left
        R = 0
        if S == 0:
            S_ = S  # reach the wall：已经到达最左端，不能再向左
        else:
            S_ = S - 1
    return S_, R


def update_env(S, episode, step_counter):
    # This is how environment be updated
    env_list = ['-'] * (N_STATES - 1) + ['T']  # '---------T' our environment
    if S == 'terminal':
        interaction = 'Episode %s: total_steps = %s' % (episode + 1, step_counter)
        print('\r{}'.format(interaction), end='')
        time.sleep(1)
        print('\r                                ', end='')
    else:
        env_list[S] = 'o'
        interaction = ''.join(env_list)
        print('\r[{} - {}] {}'.format(episode, step_counter, interaction), end='')
        time.sleep(FRESH_TIME)


def rl():
    q_table = build_q_table(N_STATES, ACTIONS)  # 初始 q table
    for episode in range(MAX_EPISODES):  # 回合
        step_counter = 0
        S = 0  # 回合初始位置
        is_terminated = False  # 是否回合结束
        update_env(S, episode, step_counter)  # 环境更新
        while not is_terminated:
            
            # 根据Q表格选择状态s采取的动作a，并作用于环境得到反馈和奖励
            A = choose_action(S, q_table)  # 选行为
            S_, R = get_env_feedback(S, A)  # 实施行为并得到环境的反馈
            q_predict = q_table.loc[S, A]  # 估算的(状态-行为)值
            
            # 计算下一个状态的所能拿到的最大奖励
            if S_ != 'terminal':
                q_target = R + GAMMA * q_table.iloc[S_, :].max()  # 实际的(状态-行为)值 (回合没结束)
            else:
                q_target = R  # 实际的(状态-行为)值 (回合结束)
                is_terminated = True  # terminate this episode

            # q_table 更新：用下一个状态的所能拿到的最大奖励，作为当前状态行为的目标值
            q_table.loc[S, A] += ALPHA * (q_target - q_predict)

            step_counter += 1; S = S_  # 探索者移动到下一个 state
            update_env(S, episode, step_counter)    # 环境更新
        
        print(f"episode {episode}\n", q_table)

    return q_table


if __name__ == "__main__":
    q_table = rl()
    print('\r\nQ-table:\n')
    print(q_table)

迭代过程中的Q-Table取值情况如下，可以看到Q是从$t+1 \rightarrow t$的方向逐步收敛的。

位置(s)\方向(a)	1/left	1/right	2/left	2/right	3/left	3/right	4/left	4/right	5/left	5/right	6/left	6/right	7/left	7/right	8/left	8/right	9/left	9/right	10/left	10/right
0	0	0	0	0	0	0	0	0	0	6.561e-06	0	3.60855e-05	0	0.000115802	0	0.000283206	0	0.000584533	0	0.00107268
1	0	0	0	0	0	0	0	7.29e-05	0	0.00033534	0	0.00092583	0	0.00198871	0	0.00366275	0	0.00607337	0	0.0093277
2	0	0	0	0	0	0.00081	0	0.002997	0	0.0069336	0	0.0128385	0	0.0208101	0	0.0308543	0	0.0429074	0	0.0568546
3	0	0	0	0.009	0	0.0252	0	0.04707	0	0.073314	0	0.102839	0	0.134725	0	0.168206	0	0.202643	0	0.237511
4	0	0.1	0	0.19	0	0.271	0	0.3439	0	0.40951	0	0.468559	0	0.521703	0	0.569533	0	0.61258	0	0.651322
5	0	0	0	0	0	0	0	0	0	0	0	0	0	0	0	0	0	0	0	0

SARSA

全称是State-Action-Reward-State’-Action’
伪代码为

Initialize Q(s, a) arbitrarily
Repeat (for each episode):
    Initialize s
    Repeat (for each step of episode):
        Choose a from s using policy derived from Q (e.g. \epsilon-greedy)
        Take action a, observe r, s'
        Choose a' from s' using policy derived from Q (e.g. \epsilon-greedy)
        Q(s, a) \leftarrow Q(s, a) + \alpha \left[ \underline{r + \gamma Q(s', a')} - Q(s, a) \right]
        s \leftarrow s'; a \leftarrow a'
    until s is terminal

与Q-Learning的区别在于更新方式不同，在下一状态$s'$用相同策略确定动作$a'$（$G_t = R_t + \gamma G_{t+1}$）

$$Q(s, a) \leftarrow Q(s, a) + \alpha \left[ \underline{r + \gamma Q(s', a')} - Q(s, a) \right]$$

与Q-Learning的区别：，Q-learning是选取$s'$上会带来最大收益的行为，但是做决策的时候可能不一定会选择该行为（异策略，行动策略和评估策略不是同一个策略），而SARSA则是​在$s'$上面选择实际$a'$的Q值，最后像Q-learning一样求出现实和估计的差距，并且更新Q表里面的值。

DQN

在状态空间$S$或者动作空间$A$非常大的情况下，无法枚举$(s, a)$构建Q-Table，因此Q-Learning不适用于复杂场景。为了解决这个问题，DQN用神经网络模型拟合函数$Q(s, a)$。

伪代码如下

Initialize relay memory D to capacity N                                                     # experience replay
Initialize action-value function Q with random weights \theta                               # Q-Function
Initialize target action-value function \hat{Q} with weights \theta^- = \theta
For episode = 1, M do
    Initialize sequence s_1 = \{x_1\} and preprocessed sequence \phi_1 = \phi(s_1)
    For t = 1, T do
        With probability \epsilon select a random action a_t \
            otherwise select a_t = \argmax_{a} Q(\phi(s_t), a; \theta)                      # \epsilon-greedy
        Execute action a_t in emulator and observe reward r_t and image x_{t + 1}           # environment reaction
        Set s_{t + 1} = s_t, a_t, x_{t + 1} and preprocess \phi_{t + 1} = \phi(s_{t + 1})
        Store transition (\phi_t, a_t, r_t, \phi_{t + 1}) in D                              # experience replay
        Sample random minibatch of transitions (\phi_j, a_j, r_j, \phi_{j + 1})_{j = 1, \cdots, B} from D
        set y_j = \begin{cases}
            r_j & \text{if episode terminates at step j + 1} \\
            r_j + \gamma \max_{a'} \hat{Q}(\phi_{j + 1}, a'; \theta^-) & \text{otherwise}
        \end{cases}
        Perform a gradient descent step on L_j = \left( y_j - Q(\phi_j, a_j; \theta) \right)^2 with respect to the network parameters \theta
        Every C steps reset \hat{Q} = Q                                                     # fixed-q-target
    End For
End For

其中$a_t$的选择同样基于$\epsilon-greedy$，即

$$a_t = \begin{cases} \argmax_{a} Q(\phi(s_t), a; \theta) & p < \epsilon \\ \text{random}_{a \in A} a & \text{otherwise} \end{cases}$$

其中$\phi(s)$是对序列$s$的预处理函数，目的是令数值更平滑，有利于模型收敛，$\phi_t = \phi(s_t)$。损失定义为

$$L_j = \left( y_j - Q(\phi_j, a_j; \theta) \right)^2$$

其中

$$y_j = \begin{cases} r_j & \text{if episode terminates at step j + 1} \\ r_j + \gamma \max_{a'} \hat{Q}(\phi_{j + 1}, a'; \theta^-) & \text{otherwise} \end{cases}$$

从伪代码可以看出，DQN主要作出了以下三个贡献

1. 将Q-Table参数化得到Q-Function，并用神经网络拟合；
2. 经验回放(Experience Replay)：
   • 强化学习采集数据的过程非常慢，如果能将互动过程中的数据缓存起来，每步就可以通过采样一批数据进行参数更新
   • 强化学习采集的数据之间存在关联性，而深度神经网络训练中要求数据满足独立同分布，因此直接用相邻时间步的数据会使模型训练不稳定，而经验回放通过采样的方式可以打破数据间的关联；
   • 当超出容量$N$，则按队列顺序删除以前的经验，从而动态地提升训练数据质量。
3. 目标网络(Fixed-Q-Target)：训练过程中使用了评估网络$Q$和目标网络$\hat{Q}$两个网络，也是一种打乱相关性的机制。具体地，这两个网络在初始化时有相同的结构和参数，训练过程中，评估网络$Q$的参数$\theta$不断地通过梯度下降更新，而目标网络$\hat{Q}$的参数$\theta^-$每隔$C$步与$Q$进行同步。

实际上，DQN参数更新可以表示为下式，在形式上与Q-Learning保持一致

$$\theta \leftarrow \theta + \alpha \left[ r_j + \gamma \max_{a'} \hat{Q}(\phi_{j + 1}, a'; \theta^-) - Q(\phi_j, a_j; \theta) \right] \nabla Q(\phi_j, a_j; \theta)$$

DQN的三大变体

Double DQN：目标值估计的改进，缓解过估计问题

因为DQN是off-policy方法，每次学习时，不是使用下一次交互的真实动作，而是使用当前认为价值最大的动作来更新目标值函数，因此Q值往往偏大，导致过估计(over estimate)。因此，一种直观的解决方案是再加入一个模型相互监察，而DQN中本来就有两个网络$Q$和$\hat{Q}$，且$\hat{Q}$滞后于$Q$，可以极大缓解该问题。具体地，是在计算$y_j$时，用$\argmax_{a'}(Q(\phi_{j + 1}, a'; \theta))$代替$a'$

$$y_j = \begin{cases} r_j & \text{if episode terminates at step j + 1} \\ r_j + \gamma \hat{Q}(\phi_{j + 1}, \underline{\argmax_{a'}(Q(\phi_{j + 1}, a'; \theta))}; \theta^-) & \text{otherwise} \end{cases}$$

其中$a_{j + 1} =\argmax_{a'}(Q(\phi_{j + 1}, a'; \theta))$，是用评估网络$Q$得到的状态$\phi_{j+1}$下采取的动作$a_{j + 1}$。

Dueling DQN：网络结构的改进

DQN没有显式地分离状态价值和优势函数。这会导致在某些情况下，算法难以准确地估计状态价值和优势函数，从而影响策略学习的效率。Dueling DQN是从网络结构上改进DQN，将动作值函数分为状态价值函数$V$和优势函数$A$（回顾一下，优势函数定义为$A^\pi(s, a) = Q^\pi(s, a) - V^\pi(s)$），即

$$Q(\phi, a; \theta, \alpha, \beta) = V(\phi; \theta, \beta) + A(\phi, a; \theta, \alpha)$$

其中$\alpha$和$\beta$分别是状态价值函数$V$和优势函数$A$的参数，可以看到$V$仅与状态$\phi$有关，$A$与状态$\phi$和动作$a$都有关。但是，$Q$是由加性运算得到，无法用唯一的$V$、$A$确定，所以添加限制项，强制优势函数$A$估计量在动作$a^*$处具有零优势，即

$$A(\phi, a; \theta, \alpha) \leftarrow A(\phi, a; \theta, \alpha) - \max_{a'} A(\phi, a'; \theta, \alpha)$$

也即

$$Q(\phi, a; \theta, \alpha, \beta) = V(\phi; \theta, \beta) + \left( A(\phi, a; \theta, \alpha) - \max_{a'} A(\phi, a'; \theta, \alpha) \right)$$

这样，对于$\forall a^* \in \mathcal{A}$都有

$$a^* = \argmax_{a' \in \mathcal{A}} Q(\phi, a'; \theta, \alpha, \beta) = \argmax_{a' \in \mathcal{A}} A(\phi, a'; \theta, \alpha)$$

此时就有

$$Q(\phi, a^*; \theta, \alpha, \beta) = V(\phi; \theta, \beta)$$

作者又尝试了用平均代替了最大，即

$$Q(\phi, a; \theta, \alpha, \beta) = V(\phi; \theta, \beta) + \left( A(\phi, a; \theta, \alpha) - \frac{1}{|\mathcal{A}|} \sum_{a'} A(\phi, a'; \theta, \alpha) \right)$$

虽然使得值函数$V$和优势函数$A$不再完美的表示值函数和优势函数(在语义上的表示)，但是这种操作提高了稳定性。而且，并没有改变值函数$V$和优势函数$A$的本质表示。

解读 状态值函数$V(\phi; \theta, \beta)$是在状态$\phi$下，所有可能动作$a$所对应的动作值函数，乘以采取该动作的概率的和，也就是状态的期望。优势函数$Q(\phi, a; \theta, \alpha, \beta) - V(\phi; \theta, \beta)$可以评价当前动作值函数相对于平均值的大小，“优势”是指动作值函数$Q$相比于当前状态的值函数$V$的优势：如果$Q - V > 0$，表示动作$a$比平均动作好。

Prioritized Replay Buffer：训练过程的改进

在传统DQN的经验池中，选择batch的数据进行训练是随机的，没有考虑样本的优先级关系。但其实不同的样本的价值是不同的，我们需要给每个样本一个优先级，并根据样本的优先级进行采样。

样本的优先级如何确定？我们可以用到 TD-error, 也就是 q-target - q-eval 来规定优先学习的程度. 如果 TD-error 越大, 就代表我们的预测精度还有很多上升空间, 那么这个样本就越需要被学习, 也就是优先级 p 越高。

有了 TD-error 就有了优先级 p, 那我们如何有效地根据 p 来抽样呢? 如果每次抽样都需要针对 p 对所有样本排序, 这将会是一件非常消耗计算能力的事. 文中提出了一种被称作SumTree的方法。

Part 3: 从Policy-Gradient到TROP/PPO/PPO2

基于策略和基于价值的强化学习方法有什么区别?

作者：郝伟
链接：https://www.zhihu.com/question/542423465/answer/2566685921
来源：知乎
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

对于一个状态转移概率已知的马尔可夫决策过程，我们可以使用动态规划算法来求解。从决策方式来看，强化学习又可以划分为基于策略的方法和基于价值的方法。决策方式是智能体在给定状态下从动作集合中选择一个动作的依据，它是静态的，不随状态变化而变化。

• 在基于策略的强化学习方法中，智能体会制定一套动作策略（确定在给定状态下需要采取何种动作），并根据这个策略进行操作。强化学习算法直接对策略进行优化，使制定的策略能够获得最大的奖励。
• 而在基于价值的强化学习方法中，智能体不需要制定显式的策略，它维护一个价值表格或价值函数，并通过这个价值表格或价值函数来选取价值最大的动作。

基于价值迭代的方法只能应用在不连续的、离散的环境下**（如围棋或某些游戏领域），对于动作集合规模庞大、动作连续的场景（如机器人控制领域），其很难学习到较好的结果（此时基于策略迭代的方法能够根据设定的策略来选择连续的动作）。
基于价值的强化学习算法有Q学习（Q-learning）、Sarsa等，而基于策略的强化学习算法有策略梯度（Policy Gradient，PG）算法等。此外，Actor-Critic算法同时使用策略和价值评估来做出决策。其中，智能体会根据策略做出动作，而价值函数会对做出的动作给出价值，这样可以在原有的策略梯度算法的基础上加速学习过程，取得更好的效果。

Policy Gradient

核心思想是直接优化策略网络(Policy Network)$a = \pi(a | s; \theta)$，即根据输入状态$s$输出各动作的概率，并依概率采样得到动作$a$。那么网络应该如何训练来实现最终的收敛呢？强化学习中只能通过奖励判断动作的好坏，也就是说一个动作奖励越大，那么增加其出现的概率，否则降低，这就是策略梯度的基本思想。

推导过程

给定策略网络$\pi(a | s; \theta)$，在一个回合内（游戏开始到结束称为一个回合，episode）与环境产生交互得到序列$\tau = \{s_1, a_1, r_1, s_2, a_2, r_2, \cdots, s_T, a_T, r_T\}$，其中$a_t$依概率$\pi(a_t | s_t; \theta)$采样得到，因而具有随机性。那么该回合总的奖励为$R_{\theta}(\tau) = \sum_t r_t$，记$P_{\theta}(\tau)$为该回合产生的概率，多个回合产生序列集合$\Tau$。定义期望的总奖励为$\overline{R}_{\theta}$，就有

$$\overline{R}_{\theta} = \sum_\tau R_{\theta}(\tau) P_{\theta}(\tau)$$

那么，总体的训练目标就是令期望的总奖励最大，即

$$\theta^* = \argmax_{\theta} \overline{R}_{\theta}$$

可通过梯度下降法求取

$$\begin{aligned} \nabla \overline{R}_{\theta} &= \sum_\tau R_{\theta}(\tau) \cdot \nabla P_{\theta}(\tau) \\ &= \sum_\tau R_{\theta}(\tau) \cdot P_{\theta}(\tau) \cdot \nabla \log P_{\theta}(\tau) \\ &= E_{\tau \sim P_{\theta}(\tau)} R_{\theta}(\tau) \cdot \nabla \log P_{\theta}(\tau) \\ &\approx \frac{1}{|\Tau|} \sum_{\tau \in \Tau} R_{\theta}(\tau) \cdot \nabla \log P_{\theta}(\tau) \\ \end{aligned}$$

注：$\nabla f(x) = f(x) \cdot \frac{\nabla f(x)}{f(x)} = f(x) \cdot \nabla log f(x)$

而根据马尔可夫独立性假设，有

$$\begin{aligned} P_{\theta}(\tau) &= P(s_1) \cdot P(a_1|s_1) P(s_2|s_1, a_1) \cdot P(a_2|s_2) P(s_3|s_2, a_2) \cdots \\ &= P(s_1) \prod_{t} P(a_t|s_t) P(s_{t+1}|s_t, a_t) \end{aligned}$$

则

$$\log P_{\theta}(\tau) = \underline{\log P(s_1)} + \sum_t \log P(a_t|s_t) + \underline{\log P(s_{t+1}|s_t, a_t)}$$

那么

$$\nabla \log P_{\theta}(\tau) = \sum_t \nabla \log P(a_t|s_t)$$

代入$\nabla \overline{R}_{\theta}$则有

$$\begin{aligned} \nabla \overline{R}_{\theta} \approx \frac{1}{|\Tau|} \sum_{\tau \in \Tau} R_{\theta}(\tau) \cdot \underline{\sum_t \nabla \log \pi(a_t|s_t; \theta)} \approx \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} r_t \cdot \nabla \log \pi(a_t|s_t; \theta) \end{aligned}$$

因此

$$\begin{cases} \nabla \overline{R}_{\theta} &\approx \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} r_t \cdot \nabla \log \pi(a_t|s_t; \theta) \\ \theta &\leftarrow \theta + \eta \nabla \overline{R}_{\theta} \\ \end{cases}$$

相应地，损失函数为

$$L = \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} r_t \cdot \log \pi(a_t|s_t; \theta)$$

注：形式上与分类任务交叉熵损失类似？？

$$L = \frac{1}{|D|} \sum_{(x, y) \in D} \sum_c y_c \log p_c(x)$$

优缺点

PG的优点是：

• 更好的收敛性质
• 在高维或连续动作空间有效
• 可以学习随机策略
• 不会出现策略退化现象

缺点是：

• 可以收敛到不动点，但往往是局部最优
• 对策略的评估往往是低效并且高方差的
• 数据效率和鲁棒性不行。

PG的变体形式

上面的PG是以最大化每步奖励为目标，还可以将式中的奖励$r_t$替换成其他项，更换优化目标，从而得到PG的几种变体：

$$L \approx \begin{cases} \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \log \pi(a_t|s_t; \theta) \cdot r_t & \text{REINFOCEMENT} \\ \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \log \pi(a_t|s_t; \theta) \cdot Q(s_t, a_t; \theta) & \text{Q Actor-Critic} \\ \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \log \pi(a_t|s_t; \theta) \cdot A(s_t, a_t; \theta) & \text{Advantage Actor-Critic} \\ \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \log \pi(a_t|s_t; \theta) \cdot \delta & \text{TD Actor-Critic} \\ \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \log \pi(a_t|s_t; \theta) \cdot \delta e & \text{TD(} \lambda \text{)Actor-Critic} \\ \end{cases}$$

这几种变体是怎么来的呢？以第三种 Advantage Actor-Critic 为例，我们深入讲一讲就能理解其他变体的含义。

变体：Advantage Actor-Critic（Advantage AC）

先对PG进行两项改进：

改进1：增加一个奖励基准$b$，即奖励达到$b$才能说这一步动作好，防止智能体在训练初期，就倾向于选择某几个奖励高的动作，从而忽略了探索低奖励动作

$$L \approx \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \underline{(r_t - b)} \cdot \log \pi(a_t|s_t; \theta)$$

改进2：上式中每个时间步$t$的$(s_t, a_t)$的奖励，都是该步下的单步奖励$r_t$。但取每步最优（贪心策略）就是最佳的吗？显然不是的，没有考虑这一步采取动作可能带来的全局的、更长远的影响。所以，可以用奖励的累加，也就是$t$时刻的回报值，来评估该步采取动作的长远价值（回顾一下，回报定义为$g_t = r_t + \gamma r_{t+1} + \gamma^2 r_{t+2} + \cdots = \sum_{k=0}^N \gamma^k r_{t+k}$），并添加衰减因子$0< \gamma < 1$，意味着随着时间推移，组合越来越多，那么前面的组合对很后面的组合的影响就越来越小，即

$$r_t \rightarrow \sum_{t' \ge t} r_{t'} \rightarrow \sum_{t' \ge t} \gamma^{t'-t} r_{t'}$$

$$L \approx \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} (\underline{\sum_{t' \ge t} \gamma^{t'-t} r_{t'} - b}) \cdot \log \pi(a_t|s_t; \theta)$$

回顾一下优势函数的定义，$A^\pi(s, a) = Q^\pi(s, a) - V^\pi(s)$，可以发现划线部分实际上是简化的优势函数，即

$$\begin{cases} Q^\pi(s, a) &= \sum_{t' \ge t} \gamma^{t'-t} r_{t'} \\ V^\pi(s) &= b (常数) \end{cases} \Rightarrow A(s_t, a_t; \theta) = \sum_{t' \ge t} \gamma^{t'-t} r_{t'} - b$$

此时就得到了变体Advantage Actor-Critic，优化目标如下。和PG最大化每步奖励不同，这种方法是最大化每步的采取动作的优势。

$$\theta^* = \argmax_{\theta} \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} A(s_t, a_t; \theta) \cdot \log \pi(a_t|s_t; \theta)$$

既然价值函数的最简形式是采取常数项，是不是能将其参数化呢？于是就得到了AC框架，Actor即策略$\pi(a_t|s_t; \theta)$，Critic即用模型预估当前状态$s_t$的价值（通俗理解就是各动作平均水平的高低）。那么如何训练Critic呢？一种方式是把奖励$r_t$作为groundtruth，因为实际上环境的奖励$r_t$是衡量状态价值的有效指标，那么价值函数的优化目标变成了

$$\phi^* = \argmin_{\phi} \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} (V(s_t; \phi) - r_t)^2$$

也可以设置其他优化目标，比如A2C中用Q值计算损失。

例程：CartPole-v1

Policy Gradient的例程，智能体通过控制滑块左右移动来保持杆子处于竖直状态：

• 环境状态：由滑块位置$x$、滑块速度$x'$、杆子角度$\theta$、杆子角速度$\theta'$组成。
• 动作空间：包含向左、向右两个可选动作。
• 奖励函数：每个时间步，如果杆的角度在±12°范围内，并且小车没有超出±2.4单位的轨道边界，则给予奖励+1；如果杆超出角度范围或小车超出边界，环境将结束，且不再给予奖励。

import os
import gym
import numpy as np
from copy import deepcopy
from collections import deque

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.distributions import Categorical

env = gym.make('CartPole-v1')
env = env.unwrapped
state_dims = env.observation_space.shape[0]
n_actions = env.action_space.n
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")

class Net(nn.Module):

    def __init__(self):
        super().__init__()
        self.layers = nn.Sequential(
            nn.Linear(state_dims, 32),
            nn.ReLU(inplace=True),
            nn.Linear(32, 32),
            nn.ReLU(inplace=True),
            nn.Linear(32, n_actions),
            nn.Softmax(dim=-1),
        )

    def forward(self, state):
        pi = self.layers(state)  # (batch_size, n_actions)
        return pi

class PG():

    def __init__(
        self,
        gamma=0.9,
        lr=5e-4,
        weight_decay=0.0,
    ):
        self.gamma = gamma
        self.buffer = []
        self.model = Net()
        self.model.to(device)
        self.optimizer = torch.optim.Adam(self.model.parameters(), lr=lr, weight_decay=weight_decay)
    
    @torch.no_grad()
    def choose_action(self, state):
        state = torch.from_numpy(state).float().unsqueeze(0).to(device)
        pi = self.model(state)                      # 策略函数输出动作的概率分布，这里用多项式分布
        dist = torch.distributions.Categorical(pi)  # 依概率分布采样，实现不同动作的探索
        action = dist.sample().item()
        return action

    def store_experience(self, experience):
        self.buffer.append(experience)  # (s_t, a_t, r_t, s_{t+1}, is_done)
    
    def update(self):
        # 得到数据
        get_tensor = lambda x: torch.tensor([b[x] for b in self.buffer]).to(device)
        states = get_tensor(0).float()          # (n_steps, state_dims)
        actions = get_tensor(1).long()          # (n_steps,)
        rewards = get_tensor(2).float()         # (n_steps,)
        # next_states = get_tensor(3).float()   # (n_steps, state_dims)
        # done = get_tensor(4).long()           # (n_steps,), [0, 0, ..., 0, 1]

        # 改进2：计算步骤t的回报值，以评估动作的更长远的影响，施加衰减项gamma累加后续步骤的奖励
        for t in reversed(range(0, rewards.size(0) - 1)):
            rewards[t] = rewards[t] + self.gamma * rewards[t + 1]
        # 改进1：1）增加一个奖励基准$b$，这里用均值 2）在此之上，再添加归一化，有助于收敛
        rewards = (rewards - rewards.mean()) / rewards.std()

        # 计算损失，注意这里把每一步的尝试都看成是独立的单独样本
        pi = self.model(states)                                     # (n_steps, n_actions)
        log_prob = torch.sum(pi.log() * F.one_hot(actions), dim=1)  # (n_steps,)
        loss = - (log_prob * rewards).mean()
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

        # 清除缓存
        del self.buffer[:]

        return loss.item()

def train(agent, num_episodes=2000, render=False):
    step = 0
    for i in range(num_episodes):
        
        # 先进行一个完整的回合，注意训练到后期稳定状态时，一个回合持续时间可能很久
        total_rewards = 0
        done = False
        state, _ = env.reset()  # state包含4项，(x, x_dot, theta, theta_dot)
        while not done:
            step += 1
            if render: env.render()
            # 在当前状态state下，通过策略函数进行随机采样，实现对不同动作的探索
            action = agent.choose_action(state)
            # 与环境产生交互，得到奖励reward，以及action作用后的下一状态next_state
            next_state, reward, done, truncated, info = env.step(action)
            # 预处理，修改reward，以加速收敛，也可以直接用reward，都能收敛
            x, x_dot, theta, theta_dot = next_state
            r1 = (env.x_threshold - abs(x)) / env.x_threshold - 0.8
            r2 = (env.theta_threshold_radians - abs(theta)) / env.theta_threshold_radians - 0.5
            r3 = 3 * r1 + r2
            # 经验缓存
            agent.store_experience((state, action, r3, next_state, done))
            # 更新状态
            state = next_state
            total_rewards += reward
        
        # 回合结束，更新参数
        loss = agent.update()
        if i % 50 == 0:
            print('episode:{} reward:{}'.format(i, total_rewards))

def test(agent, num_episodes=10, render=False):
    env = gym.make('CartPole-v1', render_mode="human" if render else None)
    step = 0
    eval_rewards = []
    for i in range(num_episodes):
        total_rewards = 0
        done = False
        state, _ = env.reset()
        while not done:
            step += 1
            if render: env.render()
            # 选择动作
            action = agent.choose_action(state)
            # 与环境产生交互
            next_state, reward, done, truncated, info = env.step(action)
            # 更新状态
            state = next_state
            total_rewards += reward
        eval_rewards.append(total_rewards)
    return sum(eval_rewards) / len(eval_rewards)

if __name__ == "__main__":
    agent = PG()
    train(agent, render=False)
    test(agent, render=True)

TRPO

强化学习包含智能体在环境中的探索和参数更新两个主要步骤，以最大化回报为强化学习目标，有

$$\theta^* = \argmax_\theta \sum_t \gamma^t r_t(\theta) = \argmax_\theta g(\tau; \theta)$$

但如果学习率$\alpha$选择不合适，迭代过程中不能保证$\tilde{\theta}$比$\theta$好，导致$\tilde{\theta}$参数采样得到较差的样本，导致参数进一步恶化。

TRPO(Trust Region Policy Optimization)就是为了解决如何选择一个合适的更新策略，或是如何选择一个合适的步长，使得更新过后的策略$\pi(a|s; \tilde{\theta})$一定比更新前的策略$\pi(a|s; \theta)$好。

方法推导

在策略$\pi(a_t|s_t;\theta)$和$\pi(a_t|s_t;\tilde{\theta})$下，长期折扣奖励分别如下

$$\begin{aligned} g(\theta) &= E_{\tau \sim P(\tau|\theta)} G(\tau; \theta) \\ g(\tilde{\theta}) &= E_{\tau \sim P(\tau|\tilde{\theta})} G(\tau; \tilde{\theta}) \\ \end{aligned}$$

那么就有

$$\begin{aligned} g(\tilde{\theta}) & = g(\theta) + E_{\tau \sim P(\tau|\tilde{\theta})} \sum_t \gamma^t A (s_t, a_t; \theta) \\ \end{aligned}$$

目标也就是使$g(\tilde{\theta}) \ge g(\theta)$，定义

$$\rho(s; \theta) = \sum_{t=0}^\infty \gamma^t P(s | \theta)$$

那么

$$\begin{aligned} g(\tilde{\theta}) & = g(\theta) + E_{\tau \sim P(\tau | \tilde{\theta})} \sum_t \gamma^t A (s_t, a_t; \theta) \\ & = g(\theta) + \sum_t \left( \sum_s P(s | \tilde{\theta}) \left( \sum_a \pi(a|s;\tilde{\theta}) \cdot \gamma^t A (s, a; \theta) \right) \right) \\ & = g(\theta) + \sum_s \sum_t \gamma^t \left( P(s | \tilde{\theta}) \sum_a \pi(a|s;\tilde{\theta}) A (s, a; \theta) \right) \\ & = g(\theta) + \sum_s \left( \rho(s; \tilde{\theta}) \sum_a \pi(a|s;\tilde{\theta}) A (s, a; \theta) \right) \\ \end{aligned}$$

上式中$\rho(s; \tilde{\theta})$对$\tilde{\theta}$有很强依赖，但实际训练过程中下一步模型$\tilde{\theta}$是无法拿到的，考虑替代函数$L^{\theta}(\tilde{\theta})$

$$L^{\theta}(\tilde{\theta}) = g(\theta) + \sum_s \left( \rho(s; \theta) \sum_a \pi(a|s; \tilde{\theta}) A (s, a; \theta) \right)$$

该函数与$g(\tilde{\theta})$在$\theta$附近是一阶近似的，即

$$\begin{cases} L^{\theta}(\theta) &= g(\theta) \\ \nabla L^{\theta}(\theta) &= \nabla g(\theta) \\ \end{cases}$$

一阶近似，是泰勒公式中的一种近似方法，具体指的是在$x$点处，只保留到一阶导数项的近似。如函数$f(x)=x-1$与函数$g(x)=\ln x$在$x=1$处是一阶近似的，因为$f(1)=g(1)=0, f'(1)=g'(1)=1$。

可以通过优化$L^{\theta}(\tilde{\theta})$来达到优化$g(\tilde{\theta})$的目的：

$$\tilde{\theta}^* = \argmax_{\tilde{\theta}} L^{\theta}(\tilde{\theta})$$

但是注意，该参数不能作为更新后的参数$\tilde{\theta}$，因为：

1. $\tilde{\theta}^*$只是给出了优化$\theta$的方向，而不是真正的最优值
2. $\tilde{\theta}^*$不一定在$\theta$附近，因此$L^{\theta}(\tilde{\theta}^*) \ge L^{\theta}(\theta)$不能证明$g(\tilde{\theta}^*) \ge g(\theta)$

因此，需要将$\tilde{\theta}^*$限制在$\theta$附近，不希望参数更新导致分布差异过大，分布差异可以通过KL散度衡量（除了上述原因，重要性采样精度同样有要求），这样就得到了TRPO算法优化目标

$$\begin{aligned} \tilde{\theta} &= \argmax_{\tilde{\theta}} L^{\theta}(\tilde{\theta}) \\ \text{s.t.} &\quad \text{KL} \left( \pi(a|s; \theta),\pi(a|s; \tilde{\theta}) \right) \leq \delta \end{aligned}$$

也就是在以$\theta$为圆心、$\delta$为半径的区域中搜索$\tilde{\theta}$。还有一个问题是，$L^{\theta}(\tilde{\theta})$涉及到依概率$\pi(a|s; \tilde{\theta})$采样，但更新前无法基于未知的$\pi$采样，这个问题通过重要性采样解决，也就是先基于$\pi(a|s; \theta)$采样，再进行修正

$$\begin{aligned} L^{\theta}(\tilde{\theta}) &= g(\theta) + \sum_s \rho(s; \theta) \sum_a \pi(a|s;\tilde{\theta}) A(s, a; \theta) \\ &= g(\theta) + \sum_s \rho(s; \theta) \sum_a \pi(a|s; \theta) \left( \frac{\pi(a|s;\tilde{\theta})}{\pi(a|s; \theta)} A(s, a; \theta) \right) \\ \end{aligned}$$

重要性采样(Importance Sampling)，假定概率分布$p(x)$、函数$f(x)$，要估算$E_{x \sim p(x)} f(x)$，可以通过蒙特卡洛方法逼近，即采样足够次数$N$后求均值得到

$$E_{x \sim p(x)} f(x) = \int p(x) f(x) dx \approx \frac{1}{N} \sum_{x=1}^N f(x_i)$$

问题就在于实际问题中：1) 很难确定$p(x)$的函数分布；2) 就算已知$p(x)$分布，也可能很难按该分布采样得到$x_i$；3) 依$p(x)$采样可能无法准确估算结果，例如用均匀分布在区间$[a, b]$上采样$f(x)$，从而求曲线积分面积$\int_a^b f(x) dx = \frac{b - a}{N} \sum_{i=1}^N f(x_i)$，由于没有考虑$f(x)$曲率等其他因素导致结果不准确。

这种情况下就需要用重要性采样解决，具体地，引入另一个容易采样的分布$q(x)$，那么

$$E_{x \sim p(x)} f(x) = \int p(x) f(x) dx = \int q(x) \frac{p(x)}{q(x)} f(x) dx = \underline{ E_{x \sim q(x)} \frac{p(x)}{q(x)} f(x) \approx \frac{1}{N} \sum_{x=1}^N \frac{p(x_i)}{q(x_i)} f(x_i) }$$

式中$\frac{p(x_i)}{q(x_i)}$即重要性权重。注意，$p(x)$与$q(x)$差距越大，则需要更多采样次数以保证精度。

每一步的策略梯度更新对应

$$\begin{aligned} \tilde{\theta} &= \argmax_{\tilde{\theta}} E_{s \sim \rho(s; \theta), a \sim \pi(a|s; \theta)} \frac{\pi(a|s;\tilde{\theta})}{\pi(a|s; \theta)} A(s, a; \theta) \\ \text{s.t.} &\quad \text{KL} \left( \pi(a|s; \theta),\pi(a|s; \tilde{\theta}) \right) \leq \delta \end{aligned}$$

对比 Advantage Actor-Critic 的优化目标

$$\theta^* = \argmax_{\theta} E_{s \sim \rho(s; \theta), a \sim \pi(a|s; \theta)} \log \pi(a|s; \theta) A(s, a; \theta)$$

PPO(DeepMind)

TRPO算法引入了KL散度来保证分布相近，需要解决带约束的优化问题。PPO(Proximal Policy Optimization Algorithms)算法是对它的简化。既然是最小化KL散度，那么直接把它也作为目标函数的一部分，得到

$$\begin{aligned} \tilde{\theta} &= \argmax_{\tilde{\theta}} E_{s \sim \rho(s; \theta), a \sim \pi(a|s; \theta)} \left( \frac{\pi(a|s;\tilde{\theta})}{\pi(a|s; \theta)} A(s, a; \theta) - \beta \text{KL} \left( \pi(a|s; \theta),\pi(a|s; \tilde{\theta}) \right) \right) \end{aligned}$$

其中$\beta$是动态惩罚系数，用于控制KL散度，即$\text{KL} > \text{KL}_{\max}$则增加$\beta$，$\text{KL} < \text{KL}_{\min}$则减小$\beta$。

PPO2(OpenAI)

另一种简化方式，采取截断来使两分布的比值在$(1 - \epsilon, 1 + \epsilon)$之间，来保证分布相近

$$\begin{aligned} \tilde{\theta} &= \argmax_{\tilde{\theta}} E_{s \sim \rho(s; \theta), a \sim \pi(a|s; \theta)} \min \left( \frac{\pi(a|s;\tilde{\theta})}{\pi(a|s; \theta)} A(s, a; \theta), \text{clip} ( \frac{\pi(a|s;\tilde{\theta})}{\pi(a|s; \theta)}, 1 - \epsilon, 1 + \epsilon ) A(s, a; \theta) \right) \end{aligned}$$

例程

智能体通过控制左右旋转力度来保持杆子处于竖直状态（Actor-Critic 框架），使用Pendulum-v1环境，这是一个倒立摆环境，

• 环境状态：三维，由摆杆的角度$\theta \in [-\pi, \pi]$、摆杆的角速度$\theta'$、$\cos(\theta)$和$\sin(\theta)$（表示角度，避免角度值的不连续性）组成
• 动作空间：施加在摆杆上的扭矩$\text{torque}$，范围$[-2, 2] N \cdot m$
• 奖励函数：$r = - (\theta^2 + 0.1 \theta'^2 + 0.001 \text{torque}^2)$

import os
import random
import argparse
from collections import namedtuple

import gym
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.distributions import Normal
from torch.utils.data.sampler import BatchSampler, SubsetRandomSampler

# Parameters
parser = argparse.ArgumentParser(description='Solve the Pendulum with PPO')
parser.add_argument('--gamma', type=float, default=0.9, metavar='G', help='discount factor (default: 0.9)')
parser.add_argument('--seed', type=int, default=0, metavar='N', help='random seed (default: 0)')
parser.add_argument('--render', action='store_true', default=False, help='render the environment')
parser.add_argument('--log-interval', type=int, default=10, metavar='N', 
    help='interval between training status logs (default: 10)')
args = parser.parse_args()

env = gym.make('Pendulum-v1', render_mode='human' if args.render else None).unwrapped
state_dims = env.observation_space.shape[0]  # 3
num_action = env.action_space.shape[0]       # 1
torch.manual_seed(args.seed)
random.seed(args.seed)

Experience = namedtuple('Experience', ['state', 'action', 'a_log_prob', 'reward', 'next_state'])
TrainRecord = namedtuple('TrainRecord', ['episode', 'reward'])


class Actor(nn.Module):
    def __init__(self):
        super(Actor, self).__init__()
        self.fc = nn.Linear(state_dims, 100)
        self.mu_head = nn.Linear(100, 1)
        self.sigma_head = nn.Linear(100, 1)

    def forward(self, x):
        x = F.tanh(self.fc(x))
        mu = 2.0 * F.tanh(self.mu_head(x))
        sigma = F.softplus(self.sigma_head(x))
        return (mu, sigma)


class Critic(nn.Module):
    def __init__(self):
        super(Critic, self).__init__()
        self.fc1 = nn.Linear(state_dims, 64)
        self.fc2 = nn.Linear(64, 8)
        self.state_value = nn.Linear(8, 1)

    def forward(self, x):
        x = F.leaky_relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        value = self.state_value(x)
        return value


class PPO2():
    clip_epsilon = 0.2
    max_grad_norm = 0.5
    ppo_epoch = 10
    buffer_capacity, batch_size = 1000, 32

    def __init__(self):
        super(PPO2, self).__init__()
        self.actor_net = Actor().float()
        self.critic_net = Critic().float()
        self.buffer = []
        self.counter = 0
        self.training_step = 0
        self.actor_optimizer = optim.Adam(self.actor_net.parameters(), lr=1e-4)
        self.critic_net_optimizer = optim.Adam(self.critic_net.parameters(), lr=3e-4)

    def save_param(self):
        torch.save(self.actor_net.state_dict(), 'ppo2_actor_params.pkl')
        torch.save(self.critic_net.state_dict(), 'ppo2_critic_params.pkl')
    
    def load_param(self):
        self.actor_net.load_state_dict(torch.load('ppo2_actor_params.pkl'))
        self.critic_net.load_state_dict(torch.load('ppo2_critic_params.pkl'))

    @torch.no_grad()
    def choose_action(self, state):
        state = torch.from_numpy(state).float().unsqueeze(0)
        mu, sigma = self.actor_net(state)   # 策略函数输出动作的概率分布，这里需要连续值作为动作，采用正态分布
        dist = Normal(mu, sigma) 
        action = dist.sample()              # 依概率分布采样，实现不同动作的探索
        action = action.clamp(-2, 2)        # 动作空间限制输出 [-2, 2] 牛米
        action_log_prob = dist.log_prob(action).item()  # PPO需要用到更新前行动策略计算的概率分布，这里提前计算
        return action.item(), action_log_prob

    @torch.no_grad()
    def get_value(self, state):
        state = torch.from_numpy(state)
        value = self.critic_net(state).item()
        return value

    def store_experience(self, experience):
        self.buffer.append(experience)
        self.counter += 1
    
    def should_update(self):
        return self.counter % self.buffer_capacity == 0

    def update(self):
        self.training_step += 1
        state = torch.tensor([t.state for t in self.buffer], dtype=torch.float)                 # (n_steps, state_dims)
        action = torch.tensor([t.action for t in self.buffer], dtype=torch.float).view(-1, 1)   # (n_steps, num_action)
        action_log_prob_old = torch.tensor(
            [t.a_log_prob for t in self.buffer], dtype=torch.float).view(-1, 1)                 # (n_steps, num_action)
        reward = torch.tensor([t.reward for t in self.buffer], dtype=torch.float).view(-1, 1)   # (n_steps, 1)
        next_state = torch.tensor([t.next_state for t in self.buffer], dtype=torch.float)       # (n_steps, state_dims)
        del self.buffer[:]

        with torch.no_grad():
            # 预处理，修改reward，以加速收敛
            reward = (reward + 8) / 8
            reward = (reward - reward.mean()) / (reward.std() + 1e-5)
            # 动作价值函数 Q^{\pi}(s, a) = r(s, a) + \gamma \sum_{s' \in S} P(s'|s, a) V^{\pi}(s')
            V_ = self.critic_net(next_state)    # (n_steps, 1)
            Q = reward + args.gamma * V_        # (n_steps, 1)
            # 优势函数 A^{\pi}(s, a) = Q^{\pi}(s, a) - V^{\pi}(s)
            V = self.critic_net(state)          # (n_steps, 1)
            A = Q - V                           # (n_steps, 1)

        for _ in range(self.ppo_epoch):  # iteration ppo_epoch
            for index in BatchSampler(SubsetRandomSampler(range(self.buffer_capacity)), self.batch_size, False):

                # 更新后行动策略 \pi(a|s;\tilde{\theta})
                mu, sigma = self.actor_net(state[index])
                dist = Normal(mu, sigma)
                action_log_prob = dist.log_prob(action[index])
                
                # 如果是 Advantage Actor-Critic ，损失计算如下
                # action_loss = - (action_log_prob * A[index]).mean()

                # PPO2 损失计算如下
                ratio = torch.exp(
                    action_log_prob - action_log_prob_old[index]
                )   # 重要性采样系数 \frac{\pi(a|s;\tilde{\theta})}{\pi(a|s; \theta)}
                action_loss = - torch.min(
                    ratio * A[index],
                    torch.clamp(
                        ratio, 1 - self.clip_epsilon, 1 + self.clip_epsilon
                    ) * A[index],
                ).mean()

                # 更新行动策略
                self.actor_optimizer.zero_grad()
                action_loss.backward()
                nn.utils.clip_grad_norm_(self.actor_net.parameters(), self.max_grad_norm)
                self.actor_optimizer.step()
                
                # 更新后价值函数计算损失
                # Q: 损失函数的含义？A: 用 r + \gamma V' 代替计算q，见A2C
                target_v = reward[index] + args.gamma * V_[index]
                value_loss = F.smooth_l1_loss(self.critic_net(state[index]), target_v)
                
                # 更新价值函数
                self.critic_net_optimizer.zero_grad()
                value_loss.backward()
                nn.utils.clip_grad_norm_(self.critic_net.parameters(), self.max_grad_norm)
                self.critic_net_optimizer.step()


def main(is_training):
    agent = PPO2()
    
    if not is_training: 
        agent.load_param()
        args.render = True

    training_records = []
    running_reward = -1000

    for i_epoch in range(1000):
        score = 0
        state, _ = env.reset()
        if args.render: env.render()
        for t in range(200):
            # 评估策略 \pi(a|s;\theta)
            action, action_log_prob = agent.choose_action(state)
            next_state, reward, done, truncated, info = env.step([action])
            if args.render: env.render()

            if is_training:
                trans = Experience(state, action, action_log_prob, reward, next_state)  # s, a, \pi, r, s'
                agent.store_experience(trans)
                if agent.should_update():
                    agent.update()

            score += reward
            state = next_state

        running_reward = running_reward * 0.9 + score * 0.1
        training_records.append(TrainRecord(i_epoch, running_reward))
        if i_epoch % 10 == 0:
            print("Epoch {}, Moving average score is: {:.2f} ".format(i_epoch, running_reward))
        if running_reward > -200:
            print("Solved! Moving average score is now {}!".format(running_reward))
            env.close()
            agent.save_param()
            break


if __name__ == '__main__':
    main(is_training=True)
    main(is_training=False)

Part 4: 从Actor-Critic到A2C/A3C

PG一节已经介绍了从 PG 得到变体 Advantage Actor-Critic 的演变过程，AC框架中的 Actor 就是智能体$\pi(a_t|s_t; \theta)$，Critic就是参数化的价值函数，也就是用模型预估当前状态$s$的价值（通俗理解就是各动作平均水平的高低）。这一节更多的是 AC 算法的定义，以及介绍两种改进算法 A2C 和 A3C。

AC: Actor-Critic

policy-based可以在连续空间内选择合适动作，而这对value-based方法来说搜索空间过大；但是policy-based基于回合更新，学习效率低，通过value-based作为critic可以实现单步更新。因此，Actor-Critic算法结合了两类方法，包含Actor、Critic两部分：

• Actor：policy-based，在连续动作空间内选择合适的动作，即策略函数$\pi(a|s)$；
• Critic：value-based，评估actor产生的动作，如状态价值函数$V(s)$。

Actor的更新参数的目标是让Critic的输出值越大越好。当确定状态$s$的情况下，如何选取动作$a$来使得Critic的值最大就是Actor网络需要优化的目标。而更新Critic的参数是为了让其的打分更精准，训练的依据就是环境给的奖励$r$。

在基于蒙特卡洛的策略梯度REINFORCEMENT中，参数更新公式为

$$\theta \leftarrow \theta + \eta \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \nabla \log \pi(a_t|s_t; \theta) \cdot r_t$$

其中$r_t$是用蒙特卡罗方法采样获得的。现在引入Critic，用神经网络计算Q函数值，

$$\theta \leftarrow \theta + \eta \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \nabla \log \pi(a_t|s_t; \theta) \cdot Q(s_t, a_t; \theta)$$

其中，Critic模型$Q(s_t, a_t; \theta)$参数更新如下

$$\theta \leftarrow \theta + \eta \nabla ||r_t + \max_{a'} Q(s_{t+1}, a'; \theta) - Q(s_t, a_t; \theta)||_2^2$$

另外，广义的Actor-Critic可以有以下几种

$$\begin{cases} \theta & \leftarrow \theta + \eta \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \nabla \log \pi(a_t|s_t; \theta) \cdot V^{\pi}(s_{t}) & 基于状态价值 \\ \theta & \leftarrow \theta + \eta \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \nabla \log \pi(a_t|s_t; \theta) \cdot Q(s_t, a_t; \theta) & 基于动作价值 \\ \theta & \leftarrow \theta + \eta \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \nabla \log \pi(a_t|s_t; \theta) \cdot \delta(t) & 基于TD误差 \\ \theta & \leftarrow \theta + \eta \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \nabla \log \pi(a_t|s_t; \theta) \cdot A(s_t, a_t; \theta) & 基于优势函数 \\ \theta & \leftarrow \theta + \eta \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \nabla \log \pi(a_t|s_t; \theta) \cdot \delta(t) E(t) & 基于TD(\lambda)误差 \\ \end{cases}$$

A2C: Advantage Actor-Critic

A2C的出现是为了解决AC的高方差问题。 A2C与AC的不同之处在于，给Q值增加了一个baseline，我们用Q值减去这个baseline来判断当前逻辑的好坏，这个baseline通常由$V^{\pi}(s_t)$担任，有

$$\theta \leftarrow \theta + \eta \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \nabla \log \pi(a_t|s_t; \theta) \cdot \left( Q(s_t, a_t; \theta) - V^{\pi}(s_t) \right)$$

因此，既需要学习一个Actor来决策选什么动作，又需要Critic来评估V值和Q值，但是同时估计V值和Q值是很复杂的。执行一个动作的下一回合必定更新到$s_{t+1}$，在加上本回合获得的$r_t$就是Q的期望值。或者，由

$$\begin{cases} Q^\pi(s, a) &= r(s, a) + \gamma \sum_{s' \in S} P(s'|s, a) V^\pi(s') \\ V^{\pi}(s) &= E_\pi[R_t + \gamma V^{\pi}(S_{t+1}) | S_t=s] & (贝尔曼方程) \\ \end{cases}$$

我们可以用$r_t + \gamma V^{\pi}(s_{t+1})$来代替$Q^\pi(s, a)$，如此就只需计算V值即可：

$$\delta(t) = \underline{r_t + \gamma V^{\pi}(s_{t+1})}_{target V} - V^{\pi}(s_{t})$$

也就是

$$\frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \nabla \log \pi(a_t|s_t; \theta) \cdot \left( r_t + \gamma V^{\pi}(s_{t+1}) - V^{\pi}(s_{t}) \right)$$

其中，Critic模型$V^{\pi}(s)$参数更新如下

$$\theta \leftarrow \theta + \eta \nabla ||\underline{r_t + \gamma V^{\pi}(s_{t+1})} - V^{\pi}(s_{t})||_2^2$$

A3C: Asynchronous Advantage Actor Critic

A3C算法完全使用了Actor-Critic框架，并且引入了异步训练的思想（异步是指数据并非同时产生），在提升性能的同时也大大加快了训练速度。A
经验回放机制存在两个问题：

• Agent与环境的每次实时交互都需要耗费很多的内存和计算力；
• 经验回放机制要求Agent采用离策略(off-policy)方法来进行学习，而off-policy方法只能基于旧策略生成的数据进行更新；

3C算法为了提升训练速度采用异步训练的思想，利用多个线程。每个线程相当于一个智能体在随机探索，多个智能体共同探索，并行计算策略梯度，对参数进行更新。或者说同时启动多个训练环境，同时进行采样，并直接使用采集的样本进行训练，这里的异步得到数据，相比DQN算法，A3C算法不需要使用经验池来存储历史样本并随机抽取训练来打乱数据相关性，节约了存储空间，并且采用异步训练，大大加倍了数据的采样速度，也因此提升了训练速度。与此同时，采用多个不同训练环境采集样本，样本的分布更加均匀，更有利于神经网络的训练。

参考资料

• QLearning算法实现（python） - CSDN
• 强化学习(十一)–DQN三个经典的变种 - 知乎
• policy gradient代码 pytorch框架 - CSDN
• Policy gradient(策略梯度详解) - CSDN
• 从 PG 到 TRPO 到 PPO 到 IMPALA - CSDN
• 从采样看优化：可导优化与不可导优化的统一视角：重要性采样 - 科学空间
• 【深度强化学习】信任域策略优化：从TRPO到PPO - 知乎
• 从TRPO到PPO（理论分析与数学证明） - 简书
• 用PPO玩超级马里奥兄弟 (2.0+多线程) - AI Studio
• Actor-Critic算法 - CSDN
• 使用A3C算法玩乒乓球游戏 - 华为云