阐述、总结【动手学强化学习】章节内容的学习情况，复现并理解代码。

---

一、算法背景

1.1 算法目标

给定“黑盒”模型，求解最优policy

1.2 存在问题

动态规划、时序差分-TD、值函数近似算法中策略的表示都是基于价值（value-based）的，即算法目标是学习值函数，然后根据值函数导出一个策略，学习过程中并不存在一个显式的策略。

1.3 解决方法

采用基于策略（policy-based）算法，直接显式地学习一个目标策略。
可以理解为，之前policy的表示是“表格式(tabular)”的，即存在一个“ $\pi -table:\pi (a|s)$ ”，现在通过“函数拟合”去学习一个关于policy的函数，即${\pi }_{\theta }(a|s)$，参数为θ。

二、REINFORCE算法

• 🌟算法类型
  环境依赖：❌model-based ✅model-free
  价值估计：✅non-incremental ❌incremental（采用蒙特卡洛方法估计）
  价值表征：✅tabular representation ❌function representation（采用蒙特卡洛方法估计）
  学习方式：✅on-policy ❌off-policy
  策略表征：❌value-based ✅policy-based（通过函数拟合去计算 ${\pi }_{\theta }(a|s)$ 概率）

2.1 必要说明

softmax()函数

softmax函数是一种常用的激活函数，主要用于将一个向量转换为概率分布，使得向量中的每个元素都被映射到 [0, 1] 区间内，并且所有元素的和为 1。
对于一个向量 $\mathbf{z}=[z_1,z_2,\dots ,z_n]$ ，softmax函数的定义为：
$$\mathrm{softmax}(\mathbf{z}{)}_i=\frac{e^{z_i}}{{\sum }_{j=1}^n{e}^{z_j}}$$

• 特点
  ①归一化：softmax 函数将向量的每个元素转换为一个介于 0 和 1 之间的值，并且所有元素的和为 1。这使得输出可以被解释为概率分布。
  ** 在这里就体现为在state状态下与环境交互时，action的使用概率，即 ${\pi }_{\theta }(a|s)$
  ②强调最大值：softmax 函数会放大较大的值并抑制较小的值。这意味着在多分类问题中，模型更倾向于选择具有最高输出值的类别。
  ** 在这里就体现为最优policy中action的选择。
  ③平滑性：softmax 函数是平滑的，这意味着它在所有点上都是可导的，这在训练神经网络时非常有用，因为可以通过梯度下降等优化算法来更新模型参数。

交叉熵

当采用softmax函数作为神经网络的输出时，一般采用交叉熵来衡量两个概率的区别，即真值与神经网络输出概率的区别，因此交叉熵可以作为神经网络的损失函数。参考【动手学深度学习】part9-softmax回归

• 交叉熵定义为：
  假设模型的输出为 z，真实标签为 y，则整个过程可以表示为：
  $$\begin{array}{rl}\mathbf{p}& =\mathrm{softmax}(\mathbf{z})\\ & \\ L=H(\mathbf{y},\mathbf{p})& =-\sum _{i=1}^n{y}_i\log (p_i)\end{array}$$

策略更新思想

将policy_net的目标函数定义为最大化state value，即：
$$J(\theta )=\mathbb{E}\left[\sum _{t=0}^{\infty }{\gamma }^t{R}_{t+1}\right]={\mathbb{E}}_{s_0}[V^{{\pi }_{\theta }}(s_0)]$$
$$V^{{\pi }_{\theta }}(s_0)=\sum _a{\nabla }_{\theta }\pi (a|s,\theta )q_{\pi }(s,a)$$
为了求解最优policy使得state value最优，借鉴“梯度下降”（参考：【算法学习】优化算法-小批量随机梯度下降mini-batch SGD）的思想，求解$J(\theta )$的梯度，即：
$$\begin{array}{rl}{\nabla }_{\theta }J& =\sum _{s}d(s)\sum _a{\nabla }_{\theta }\pi (a|s,\theta )q_{\pi }(s,a)\\ & =\sum _{s}d(s)\sum _a\pi (a|s,\theta ){\nabla }_{\theta }\ln \pi (a|s,\theta )q_{\pi }(s,a)\\ & ={\mathbb{E}}_{S\sim d}\left[\sum _a\pi (a|S,\theta ){\nabla }_{\theta }\ln \pi (a|S,\theta )q_{\pi }(S,a)\right]\\ & ={\mathbb{E}}_{S\sim d,A\sim \pi }[{\nabla }_{\theta }\ln \pi (A|S,\theta )q_{\pi }(S,A)]\\ & \doteq \mathbb{E}[{\nabla }_{\theta }\ln \pi (A|S,\theta )q_{\pi }(S,A)]\end{array}$$
因此，policy_net的参数更新可以设置为：
$$\begin{array}{rl}{\theta }_{t+1}& ={\theta }_t+\alpha {\nabla }_{\theta }J(\theta )\\ & ={\theta }_t+\alpha \mathbb{E}[{\nabla }_{\theta }\ln \pi (A|S,{\theta }_t)q_{\pi }(S,A)]\end{array}$$
采用“随机近似（stochastic approximation）”去估计梯度，即：
$${\theta }_{t+1}={\theta }_t+\alpha {\nabla }_{\theta }\ln \pi (a_t|s_t,{\theta }_t)q_{\pi }(s_t,a_t)$$

2.2 伪代码

算法流程简述

①初始化policy_net：构建policy_net网络模型。
②采样episode：设定周期数num_episodes，循环迭代采样episode，根据 ${\pi }_{\theta }(a|s)$ 采样(s,a,r,s’,done)直至terminal state，即done为true，得到一条episode，即(s,a,r,s’)链。
③价值更新：依据蒙特卡洛方法，基于episode采用first-visit更新对应q(s,a)值。
④策略更新：设定损失函数为交叉熵，基于“梯度下降”方法更新policy_net网络参数。
⑤终止判断：根据已产生episode个数是否达到num_episodes，判断算法是否终止，并输出policy。

2.3 算法代码

import gym
import torch
import torch.nn.functional as F
import numpy as np
import matplotlib.pyplot as plt
from tqdm import tqdm
import rl_utilsclass PolicyNet(torch.nn.Module):def __init__(self, state_dim, hidden_dim, action_dim):super(PolicyNet, self).__init__()self.fc1 = torch.nn.Linear(state_dim, hidden_dim)self.fc2 = torch.nn.Linear(hidden_dim, action_dim)def forward(self, x):x = F.relu(self.fc1(x))return F.softmax(self.fc2(x), dim=1)class REINFORCE:def __init__(self, state_dim, hidden_dim, action_dim, learning_rate, gamma,device):self.policy_net = PolicyNet(state_dim, hidden_dim,action_dim).to(device)self.optimizer = torch.optim.Adam(self.policy_net.parameters(),lr=learning_rate)  # 使用Adam优化器self.gamma = gamma  # 折扣因子self.device = devicedef take_action(self, state):  # 🌟根据动作概率分布随机采样，不再是根据Q值去选取action了state = torch.tensor([state], dtype=torch.float).to(self.device)probs = self.policy_net(state)action_dist = torch.distributions.Categorical(probs)action = action_dist.sample()return action.item()def update(self, transition_dict):reward_list = transition_dict['rewards']state_list = transition_dict['states']action_list = transition_dict['actions']G = 0self.optimizer.zero_grad()for i in reversed(range(len(reward_list))):  # 从最后一步算起reward = reward_list[i]state = torch.tensor([state_list[i]],dtype=torch.float).to(self.device)action = torch.tensor([action_list[i]]).view(-1, 1).to(self.device)log_prob = torch.log(self.policy_net(state).gather(1, action))G = self.gamma * G + rewardloss = -log_prob * G  # 每一步的损失函数loss.backward()  # 反向传播计算梯度self.optimizer.step()  # 梯度下降，更新policy_net参数learning_rate = 1e-3
num_episodes = 1000
hidden_dim = 128
gamma = 0.98
device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")env_name = "CartPole-v0"
env = gym.make(env_name)
env.seed(0)
torch.manual_seed(0)
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
agent = REINFORCE(state_dim, hidden_dim, action_dim, learning_rate, gamma,device)return_list = []
for i in range(10):with tqdm(total=int(num_episodes / 10), desc='Iteration %d' % i) as pbar:for i_episode in range(int(num_episodes / 10)):episode_return = 0transition_dict = {'states': [],'actions': [],'next_states': [],'rewards': [],'dones': []}state = env.reset()done = Falsewhile not done:action = agent.take_action(state)next_state, reward, done, _ = env.step(action)transition_dict['states'].append(state)transition_dict['actions'].append(action)transition_dict['next_states'].append(next_state)transition_dict['rewards'].append(reward)transition_dict['dones'].append(done)state = next_stateepisode_return += rewardreturn_list.append(episode_return)agent.update(transition_dict)if (i_episode + 1) % 10 == 0:pbar.set_postfix({'episode':'%d' % (num_episodes / 10 * i + i_episode + 1),'return':'%.3f' % np.mean(return_list[-10:])})pbar.update(1)episodes_list = list(range(len(return_list)))
plt.plot(episodes_list, return_list)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('REINFORCE on {}'.format(env_name))
plt.show()mv_return = rl_utils.moving_average(return_list, 9)
plt.plot(episodes_list, mv_return)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('REINFORCE on {}'.format(env_name))
plt.show()

2.4 运行结果

Iteration 0: 100%|██████████| 100/100 [00:09<00:00, 10.01it/s, episode=100, return=28.200]
Iteration 1: 100%|██████████| 100/100 [00:09<00:00, 10.21it/s, episode=200, return=78.300]
Iteration 2: 100%|██████████| 100/100 [00:14<00:00,  6.77it/s, episode=300, return=146.000]
Iteration 3: 100%|██████████| 100/100 [00:18<00:00,  5.30it/s, episode=400, return=160.100]
Iteration 4: 100%|██████████| 100/100 [00:24<00:00,  4.02it/s, episode=500, return=199.400]
Iteration 5: 100%|██████████| 100/100 [00:25<00:00,  3.91it/s, episode=600, return=200.000]
Iteration 6: 100%|██████████| 100/100 [00:25<00:00,  3.96it/s, episode=700, return=183.700]
Iteration 7: 100%|██████████| 100/100 [00:26<00:00,  3.79it/s, episode=800, return=187.000]
Iteration 8: 100%|██████████| 100/100 [00:23<00:00,  4.22it/s, episode=900, return=193.200]
Iteration 9: 100%|██████████| 100/100 [00:25<00:00,  3.88it/s, episode=1000, return=180.700]

• 结果分析
  对比DQN算法，得到的episode_return波动更大，但能获取到最优的性能，得益于蒙特卡洛方法，如：
  Iteration 5: 100%|██████████| 100/100 [00:25<00:00, 3.91it/s, episode=600, return=200.000]
  （carpole的最佳奖励分数为200）

2.5 算法流程说明

初始化参数

learning_rate = 1e-3
num_episodes = 1000
hidden_dim = 128
gamma = 0.98
device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")

初始化“车杆”环境

env_name = "CartPole-v0"
env = gym.make(env_name)
env.seed(0)
torch.manual_seed(0)

初始化policy_net网络模型

state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
agent = REINFORCE(state_dim, hidden_dim, action_dim, learning_rate, gamma,device)
...
class PolicyNet(torch.nn.Module):def __init__(self, state_dim, hidden_dim, action_dim):super(PolicyNet, self).__init__()self.fc1 = torch.nn.Linear(state_dim, hidden_dim)self.fc2 = torch.nn.Linear(hidden_dim, action_dim)def forward(self, x):x = F.relu(self.fc1(x))return F.softmax(self.fc2(x), dim=1)

设置为4-128-2的全连接网络，输入为state=4维，输出为action的选取概率=2维，网络模型可表示为：
$$y=softmax\left(f{c}_2\left(relu(f{c}_1(x))\right)\right),x=states$$

episode采样

        for i_episode in range(int(num_episodes / 10)):episode_return = 0transition_dict = {'states': [],'actions': [],'next_states': [],'rewards': [],'dones': []}state = env.reset()done = Falsewhile not done: #采样单个episode直至terminal state，即done=trueaction = agent.take_action(state)next_state, reward, done, _ = env.step(action)transition_dict['states'].append(state)transition_dict['actions'].append(action)transition_dict['next_states'].append(next_state)transition_dict['rewards'].append(reward)transition_dict['dones'].append(done)state = next_stateepisode_return += rewardreturn_list.append(episode_return)
...def take_action(self, state):  # 🌟根据动作概率分布随机采样，不再是根据Q值去选取action了state = torch.tensor([state], dtype=torch.float).to(self.device)probs = self.policy_net(state)action_dist = torch.distributions.Categorical(probs)action = action_dist.sample()return action.item()

①获取初始state：state = env.reset()
②根据policy_net的输出采取action：agent.take_action(state)
③与环境交互，得到(s,a,r,s’,done)：env.step(action)
④将样本添加至episode：transition_dict
⑤统计episode即时奖励累加值：episode_return += reward

价值更新

agent.update(transition_dict)   #训练
...def update(self, transition_dict):reward_list = transition_dict['rewards']state_list = transition_dict['states']action_list = transition_dict['actions']G = 0self.optimizer.zero_grad()for i in reversed(range(len(reward_list))):  # 从最后一步算起reward = reward_list[i]state = torch.tensor([state_list[i]],dtype=torch.float).to(self.device)action = torch.tensor([action_list[i]]).view(-1, 1).to(self.device)log_prob = torch.log(self.policy_net(state).gather(1, action))G = self.gamma * G + reward

采样完一个episode后，倒序累加discounted reward：G，以🌟fist-visit方式来估计多个q(s,a)，这里的G就用来估计q(s,a)，以实现：
reversed(range(len(reward_list)))+G = self.gamma * G + reward
$$q(s_t,a_t)=\sum _{k=t+1}^T{\gamma }^{k-t-1}{r}_k$$

策略更新

agent.update(transition_dict)   #训练
...self.optimizer.zero_grad()...log_prob = torch.log(self.policy_net(state).gather(1, action))G = self.gamma * G + rewardloss = -log_prob * G  # 每一步的损失函数loss.backward()  # 反向传播计算梯度self.optimizer.step()  # 梯度下降，更新policy_net参数

计算交叉熵损失函数：loss = -log_prob * G，对应上交叉熵的定义：
$$L=H(\mathbf{y},\mathbf{p})=-\sum _{i=1}^n{y}_i\log (p_i)$$
其中，G对应“真值”，log_prob 对应网络模型输出概率的对数。

随后采用Adam优化器进行：梯度清零+反向传播+参数更新

三、疑问

3.1 为什么要采用softmax作为policy_net的输出？

要保证${\pi }_{\theta }(a|s)$ 中每个action的概率为正数，并且概率在(0,1)之间取值。

3.2 为什么REINFORCE算法比DQN算法波动更大？

借助蒙特卡洛方法采样轨迹来估计动作价值，这种做法的一大优点是可以得到无偏的梯度。但是，正是因为使用了蒙特卡洛方法， REINFORCE 算法的梯度估计的方差很大，可能会造成一定程度上的不稳定

3.3 策略更新的步骤具体在算法哪一块？

算法示例中没有显示地去计算policy_net的梯度，而是在Adam optimizer中进行的梯度计算，具体代码为反向传播的这一步：loss.backward() # 反向传播计算梯度

四、总结

• REINFORCE算法采用policy_net网络去显示地表征policy，采用蒙特卡洛的方法去估计q(s,a),智能体根据当前策略直接和环境交互，通过采样得到的轨迹数据直接计算出策略参数的梯度，进而更新当前策略，使其向最大化策略期望回报的目标靠近。