强化学习

Part 1：基本概念

概念

强化学习

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

交互对象

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

基本要素

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

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

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

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

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

  $$G_t = R_t + \gamma R_{t+1} + \gamma^2 R_{t+2} + \cdots = \sum_{k=0}^N \gamma^k R_{t+k}$$

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

  $$V^\pi(s) = E_\pi[G_t|S_t=s]$$

  有贝尔曼方程(Bellman Equation)

  $$\begin{aligned} V^{\pi}(s) &= E_\pi[G_t|S_t=s] \\ &= E_\pi[R_t + \gamma R_{t+1} + \gamma^2 R_{t+2} + \cdots | S_t=s] \\ &= E_\pi[R_t + \gamma (R_{t+1} + \gamma R_{t+2} + \cdots) | S_t=s] \\ &= E_\pi[R_t + \gamma G_{t+1} | S_t=s] \\ &= E_\pi[R_t + \gamma V^{\pi}(S_{t+1}) | S_t=s] \\ \end{aligned}$$

• 动作价值函数(state-value function)：在当前状态$s$，执行动作$a$后，遵循策略$\pi$所能获得的回报的期望值，即

  $$Q^\pi(s, a) = E_\pi[G_t|S_t=s, A_t=a]$$

  Q：quantity，Q函数是指状态动作函数。

  根据条件概率，有

  $$V^\pi(s) = E_{a \sim P(A_t=a|S_t=s)} Q^\pi(s, a)$$

  动作价值$a$包含了即时奖励$R_t$、下一状态的状态价值的期望，记动作$a$作用下由状态$s$转移到状态$s'$的转移概率为$P(s'|s, a)$，有

  $$Q^\pi(s, a) = r(s, a) + \gamma \sum_{s' \in S} P(s'|s, a) V^\pi(s')$$

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

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

• 优势函数(advantage function)：表示状态$s$处，动作$a$相对于平均水平的高低

  $$A^\pi(s, a) = Q^\pi(s, a) - V^\pi(s)$$

• TD误差(TD error)：在一回合观测过程中，得到部分状态序列，根据贝尔曼方程$V^{\pi}(s)=E_\pi[R_t + \gamma V^{\pi}(S_{t+1}) | S_t=s]$，可以用TD目标值$R_t + \gamma V^{\pi}(S_{t+1})$代替$G_t$，并定义TD误差为

  $$\delta(t) = R_t + \gamma V^{\pi}(S_{t+1}) - V^{\pi}(S_{t})$$

假如有以下两个序列：

• $S_0^{(1)} \rightarrow^{A_0^{(1)}} S_1^{(1)} \rightarrow^{A_1^{(1)}} S_2^{(1)} \rightarrow^{A_2^{(1)}} S_3^{(1)}$，赢
• $S_0^{(2)} \rightarrow^{A_0^{(2)}} S_1^{(2)} \rightarrow^{A_2^{(2)}} S_2^{(2)}$，输

一共$2$条序列，状态$S_1$转移到两个不同的下一状态，因此转移概率都是$0.5$。根据马尔可夫假设，设衰减因子$\gamma=0.9$，那么状态$S_1$状态价值函数为$V^\pi(S_1)=0.5 \times (R_1^{(1)} + 0.9 \times R_2^{(1)} + 0.9^2 \times R_3^{(1)}) + 0.5 \times (R_1^{(2)} + 0.9 \times R_2^{(2)})$，最终赢的状态下$R_1^{(1)} = R_2^{(1)} = R_3^{(1)} = 1$、输的状态下$R_1^{(2)} = R_2^{(2)} = 0$，那么有$V^\pi(S_1)=1.355$

分类

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$

伪代码为

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$是指，在初始阶段, 随机地探索环境往往比固定的行为模式要好, 所以这也是累积经验的阶段, 我们希望探索者不会那么贪婪(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]$$

其中，$r + \gamma \max_{a'} Q(s', a')$可以视作$Q(s, a)$的真实值，通过与预测的$Q(s, a)$偏差来逐步修正，$\max_{a'} Q(s', a')$是下一状态$s'$下，在能选择的所有动作$a' \in A$中，能拿到的最大Q值。

下面的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 = 13           # 最大回合数
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)    # 环境更新

    return q_table


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

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'$

$$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}$$

注意损失定义为

$$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参数更新可以表示为

$$\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$时，用$\hat{Q}(\phi_{j + 1}, \underline{\argmax_{a'}(Q(\phi_{j + 1}, a'; \theta))}; \theta^-)$代替$\max_{a'} \hat{Q}(\phi_{j + 1}, a'; \theta^-)$

$$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，将动作值函数分为状态值函数$V$和优势函数$A$，即

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

其中$\alpha$和$\beta$是两个全连接网络的参数，可以看到$V$仅与状态$\phi$有关，$A$与状态$\phi$和动作$a$有关。但是，此时$Q$无法用唯一的$V$、$A$确定，因此强制优势函数$A$估计量在动作$a^*$处具有零优势，即

$$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）算法等。此外，演员-评论员算法同时使用策略和价值评估来做出决策。其中，智能体会根据策略做出动作，而价值函数会对做出的动作给出价值，这样可以在原有的策略梯度算法的基础上加速学习过程，取得更好的效果。

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}{|D|} \sum_{(x, y) \in D} \sum_c y_c \log p_c(x)$

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

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

改进2：上式中每个时间步$t$的$(s_t, a_t)$的奖励，都是回合结束后的最终奖励$(r_t - b)$，也就是说权重都相同，这样是不合理的。因此，考虑用$t$到回合结束的奖励的累加作为时刻$t$的权重，并添加衰减因子$0< \gamma < 1$，意味着随着时间推移，组合越来越多，那么前面的 组合对很后面的组合的影响就越来越小，即

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

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

定义划线部分为优势函数(Advantage Function)，即

$$A(s_t, a_t; \theta) = \sum_{t' \ge t} \gamma^{t'-t} r_{t'} - b$$

最终优化目标定义为

$$\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)$$

优势函数还可以参数化，如定义价值函数$V(s; \phi)$来评估奖励(即AC框架中的Critic)，并用下式优化

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

PG的几种变体对比：

$$\nabla \overline{R}_{\theta} \approx \begin{cases} \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \nabla \log \pi(a_t|s_t; \theta) \cdot r_t & \text{REINFOCEMENT} \\ \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \nabla \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} \nabla \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} \nabla \log \pi(a_t|s_t; \theta) \cdot \delta & \text{TD Actor-Critic} \\ \frac{1}{|\Tau|} \sum_{\tau \in \Tau} \sum_{t} \nabla \log \pi(a_t|s_t; \theta) \cdot \delta e & \text{TD(}\lambda\text{)Actor-Critic} \\ \end{cases}$$

优点：

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

缺点：

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

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

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_number = env.observation_space.shape[0]
action_number = 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_number, 32),
            nn.ReLU(inplace=True),
            nn.Linear(32, 32),
            nn.ReLU(inplace=True),
            nn.Linear(32, action_number),
            nn.Softmax(dim=-1),
        )

    def forward(self, state):
        pi = self.layers(state)  # (batch_size, action_number)
        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)
    
    def update(self):
        # 得到数据
        get_tensor = lambda x: torch.tensor([b[x] for b in self.buffer]).to(device)
        states = get_tensor(0).float()
        actions = get_tensor(1).long()
        rewards = get_tensor(2).float()
        next_states = get_tensor(3).float()
        done = get_tensor(4).long()

        # 改进2：为每步t赋予不同权重
        for t in reversed(range(0, rewards.size(0) - 1)):
            rewards[t] = rewards[t] + self.gamma * rewards[t + 1]
        # 改进1：增加一个奖励基准$b$，这里用均值；另归一化，有助于收敛
        rewards = (rewards - rewards.mean()) / rewards.std()

        # 计算损失
        pi = self.model(states)
        log_prob = torch.sum(pi.log() * F.one_hot(actions), dim=1)
        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=5000, render=False):
    step = 0
    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)
            # 预处理，修改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^{\theta}_t = \argmax_\theta G^{\theta}(\tau)$$

如果学习率$\alpha$选择不合适，迭代过程中不能保证$\theta_{new}$比$\theta_{old}$好，导致$\theta_{new}$参数采样得到较差的样本，导致参数进一步恶化。TRPO(Trust Region Policy Optimization)就是为了解决如何选择一个合适的更新策略，或是如何选择一个合适的步长，使得更新过后的策略$\pi(a|s; \theta_{new})$一定比更新前的策略$\pi(a|s; \theta_{old})$好。

在策略$\pi(a_t|s_t;\theta)$和$\pi(a_t|s_t;\tilde{\theta})$下，长期折扣奖励分别如下，目标也就是使$g(\theta_{new}) \ge g(\theta_{old})$

$$\begin{aligned} g(\theta) &= E_{\tau \sim P_{\theta}(\tau)} G^{\theta}(\tau) \\ g(\tilde{\theta}) &= E_{\tau \sim P_{\tilde{\theta}}(\tau)} G^{\tilde{\theta}}(\tau) \\ \end{aligned}$$

那么就有

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

怎么来的？

定义

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

那么

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

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

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

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

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

函数$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})$$

但是该参数不能作为更新后的参数$\theta_{new}$，因为：

1. $\tilde{\theta}^*$只是给出了优化$\theta_{old}$的方向，需要将$\theta_{old}$向$\tilde{\theta}^*$迭代
2. $\tilde{\theta}^*$不一定在$\theta_{old}$附近，因此$L^{\theta_{old}}(\tilde{\theta}^*) \ge L^{\theta_{old}}(\theta_{old})$不能证明$g(\tilde{\theta}^*) \ge g(\theta_{old})$

因此，需要将$\tilde{\theta}^*$限制在$\theta_{old}$附近，可以通过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^{\theta}(s) \sum_a \pi(a|s;\tilde{\theta}) A^{\theta} (s, a) \\ &= g(\theta) + \sum_s \rho^{\theta}(s) \sum_a \pi(a|s; \theta) \left( \frac{\pi(a|s;\tilde{\theta})}{\pi(a|s; \theta)} A^{\theta} (s, a) \right) \\ \end{aligned}$$

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

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

用泰勒展开简化

$$\begin{aligned} \tilde{\theta}^* &= \argmax_{\tilde{\theta}} g^\top (\tilde{\theta} - \theta) \\ \text{s.t.} &\quad \frac{1}{2} (\tilde{\theta} - \theta)^\top H (\tilde{\theta} - \theta) \leq \delta \end{aligned}$$

其中$g$等于策略梯度，根据拉格朗日对偶定理，得到如下。

$$\tilde{\theta}^* = \theta + \alpha^j \sqrt{\frac{2 \delta}{g^\top H^{-1} g}} H^{-1} g$$

式中$\alpha$是回溯系数，能避免泰勒展开误差，防止约束函数无法满足、或代理函数无法提升。

重要性采样(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)$差距越大，则需要更多采样次数以保证精度。

PPO(DeepMind)

TRPO算法引入了KL散度来保证分布相近，需要解决带约束的优化问题。PPO(Proximal Policy Optimization Algorithms)算法对此进行改进，得到

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

PPO2的例程，智能体通过控制左右旋转力度来保持杆子处于竖直状态（涉及Actor-Critic，在下一节中介绍）。

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
num_state = env.observation_space.shape[0]
num_action = env.action_space.shape[0]
torch.manual_seed(args.seed)
random.seed(args.seed)

Transition = namedtuple('Transition', ['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(3, 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(num_state, 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)

    @torch.no_grad()
    def select_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_log_prob = dist.log_prob(action)
        action = action.clamp(-2, 2)
        return action.item(), action_log_prob.item()

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

    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'))

    def store_transition(self, transition):
        self.buffer.append(transition)
        self.counter += 1
        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)
        action = torch.tensor([t.action for t in self.buffer], dtype=torch.float).view(-1, 1)
        action_log_prob_old = torch.tensor([t.a_log_prob for t in self.buffer], dtype=torch.float).view(-1, 1)
        reward = torch.tensor([t.reward for t in self.buffer], dtype=torch.float).view(-1, 1)
        next_state = torch.tensor([t.next_state for t in self.buffer], dtype=torch.float)
        del self.buffer[:]

        with torch.no_grad():
            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')
            target_v = reward + args.gamma * self.critic_net(next_state)
            # 优势函数 A^{\pi}(s, a) = Q^{\pi}(s, a) - V^{\pi}(s)
            advantage = target_v - self.critic_net(state)

        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])
                
                # # Actor-Critic(TD error)
                # action_loss = - (action_log_prob * advantage[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 * advantage[index],
                    torch.clamp(ratio, 1 - self.clip_epsilon, 1 + self.clip_epsilon) * advantage[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()

                value_loss = F.smooth_l1_loss(self.critic_net(state[index]), target_v[index])
                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.select_action(state)
            next_state, reward, done, truncated, info = env.step([action])
            if args.render: env.render()

            if is_training:
                trans = Transition(state, action, action_log_prob, reward, next_state)  # s, a, \pi, r, s'
                if agent.store_transition(trans):
                    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

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算法不需要使用经验池来存储历史样本并随机抽取训练来打乱数据相关性，节约了存储空间，并且采用异步训练，大大加倍了数据的采样速度，也因此提升了训练速度。与此同时，采用多个不同训练环境采集样本，样本的分布更加均匀，更有利于神经网络的训练。

Part 5: AlphaZero：多智能体强化学习

总体介绍

蒙特卡洛树搜索

自对弈

参考资料

• 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算法玩乒乓球游戏 - 华为云