摘要

本系列知识点讲解基于蘑菇书EasyRL中的内容进行详细的疑难点分析！具体内容请阅读蘑菇书EasyRL！

---

对应蘑菇书附书代码——PolicyGradient.ipynb

---

下面这段代码定义了一个策略梯度（Policy Gradient）网络，通常用于强化学习中作为策略网络。该网络是一个简单的多层全连接神经网络，包含两个隐藏层，并使用 ReLU 激活函数以及 Sigmoid 激活函数输出概率（或概率近似值）。下面我们逐行详细解析这段代码的作用和运作机制，并给出具体数值例子。

---

A. PGNet

---

定义了一个策略梯度（Policy Gradient）网络，通常用于强化学习中作为策略网络。该网络是一个简单的多层全连接神经网络，包含两个隐藏层，并使用 ReLU 激活函数以及 Sigmoid 激活函数输出概率（或概率近似值）。

---

import torch
import torch.nn as nn
import torch.nn.functional as F
class PGNet(nn.Module):
    def __init__(self, input_dim,output_dim,hidden_dim=128):
        """ 初始化q网络，为全连接网络
            input_dim: 输入的特征数即环境的状态维度
            output_dim: 输出的动作维度
        """
        super(PGNet, self).__init__()
        self.fc1 = nn.Linear(input_dim, hidden_dim) # 输入层
        self.fc2 = nn.Linear(hidden_dim,hidden_dim) # 隐藏层
        self.fc3 = nn.Linear(hidden_dim, output_dim) # 输出层
    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = torch.sigmoid(self.fc3(x))
        return x

1. 模块导入

import torch
import torch.nn as nn
import torch.nn.functional as F

• torch：PyTorch 的核心模块，用于张量计算和自动微分。
• torch.nn：提供神经网络构建的模块，如线性层、卷积层等。
• torch.nn.functional：提供许多常用的激活函数和其他操作，这里我们用到了 ReLU 和 Sigmoid。

---

2. 定义网络类 PGNet

class PGNet(nn.Module):

• 作用：
  定义一个名为 PGNet 的类，该类继承自 PyTorch 的 nn.Module，这样可以利用 PyTorch 的网络模块功能（例如参数管理、自动微分、模型保存与加载等）。
• 继承说明：
  继承自 nn.Module 意味着你需要重写 init 方法来定义网络结构和 forward 方法来定义前向传播过程。

---

3. 构造函数 init

    def __init__(self, input_dim, output_dim, hidden_dim=128):
        """ 初始化q网络，为全连接网络
            input_dim: 输入的特征数即环境的状态维度
            output_dim: 输出的动作维度
        """
        super(PGNet, self).__init__()

• 参数说明：
  • input_dim：输入特征的数量，通常对应于环境状态的维度。例如，在 CartPole 环境中可能为 4。
  • output_dim：输出维度，通常对应于动作数量或策略网络输出概率的个数。
  • hidden_dim：隐藏层神经元的数量，默认设置为 128。
• super(PGNet, self).init()：
  调用父类 nn.Module 的构造函数，确保模块正确初始化。

---

        self.fc1 = nn.Linear(input_dim, hidden_dim)  # 输入层

• 作用：
  定义第一个全连接层（fc1），将输入维度转换为隐藏层维度。
• 举例：
  假设 input_dim = 4，hidden_dim = 128，则 fc1 将 4 维输入转换成 128 维输出，计算公式为
  $$\text{fc1}(x)=W_{1}x+b_1,W_1\in {\mathbb{R}}^{128\times 4},b_1\in {\mathbb{R}}^{128}$$

---

        self.fc2 = nn.Linear(hidden_dim, hidden_dim)  # 隐藏层

• 作用：
  定义第二个全连接层，将第一个隐藏层的 128 维输出再映射到另一个 128 维空间。
• 举例：
  计算公式为
  $$\text{fc2}(h)=W_{2}h+b_2,W_2\in {\mathbb{R}}^{128\times 128},b_2\in {\mathbb{R}}^{128}$$

---

        self.fc3 = nn.Linear(hidden_dim, output_dim)  # 输出层

• 作用：
  定义输出层，将隐藏层的输出转换为最终输出，输出维度由 output_dim 决定。
• 举例：
  如果 output_dim = 2（例如动作数量为 2），则 fc3 将 128 维向量映射到 2 维输出，计算公式为
  $$\text{fc3}(h)=W_{3}h+b_3,W_3\in {\mathbb{R}}^{2\times 128},b_3\in {\mathbb{R}}^2$$

---

4. 前向传播函数 forward

    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = torch.sigmoid(self.fc3(x))
        return x

• 作用：
  定义网络的前向传播过程。输入 x 经过各层计算并应用激活函数，最后返回输出。

• 详细过程：

  1. 第一层计算

     x = F.relu(self.fc1(x))
     

     • 先通过 fc1 计算线性组合，将输入 x（形状应为 (batch_size, input_dim)）转换为 (batch_size, hidden_dim) 的向量。
     • 然后使用 ReLU 激活函数，ReLU(x)=max(0, x)，使得输出非负，增加非线性。

     具体例子：
     假设 x = [0.5, -0.2, 0.1, 0.3]（单个样本），fc1 计算后得到一个 128 维向量，ReLU 会将其中小于 0 的值置零。

  2. 第二层计算

     x = F.relu(self.fc2(x))
     

     • 将第一层输出经过 fc2 线性变换，再经过 ReLU 激活，保持形状为 (batch_size, hidden_dim)。

  3. 输出层计算

     x = torch.sigmoid(self.fc3(x))
     

     • 经过 fc3 得到 (batch_size, output_dim) 的向量，然后使用 Sigmoid 激活函数。
     • Sigmoid 将输出压缩到 (0,1) 范围内，常用于表示概率或归一化输出。

     具体例子：
     对于一个 2 维输出，假设 fc3 输出为 [0.2, 1.8]，经过 Sigmoid 转换后变为
     $$[\sigma (0.2),\sigma (1.8)]\approx [0.55,0.86]$$

  4. 返回输出
     返回最终输出 x，其形状为 (batch_size, output_dim)。

---

B. 基于策略梯度（Policy Gradient）的强化学习智能体

---

该智能体使用一个神经网络（policy_net）作为策略函数，根据当前状态生成一个动作的概率分布，然后通过采样生成动作；在更新阶段，它从存储的轨迹（memory）中取出所有状态、动作和奖励数据，计算折扣回报并归一化，最后利用负对数概率乘以归一化奖励作为损失进行梯度下降。

---

import torch
from torch.distributions import Bernoulli
from torch.autograd import Variable
import numpy as np

class PolicyGradient:
    
    def __init__(self, model,memory,cfg):
        self.gamma = cfg['gamma']
        self.device = torch.device(cfg['device']) 
        self.memory = memory
        self.policy_net = model.to(self.device)
        self.optimizer = torch.optim.RMSprop(self.policy_net.parameters(), lr=cfg['lr'])

    def sample_action(self,state):

        state = torch.from_numpy(state).float()
        state = Variable(state)
        probs = self.policy_net(state)
        m = Bernoulli(probs) # 伯努利分布
        action = m.sample()
        
        action = action.data.numpy().astype(int)[0] # 转为标量
        return action
    def predict_action(self,state):

        state = torch.from_numpy(state).float()
        state = Variable(state)
        probs = self.policy_net(state)
        m = Bernoulli(probs) # 伯努利分布
        action = m.sample()
        action = action.data.numpy().astype(int)[0] # 转为标量
        return action
        
    def update(self):
        state_pool,action_pool,reward_pool= self.memory.sample()
        state_pool,action_pool,reward_pool = list(state_pool),list(action_pool),list(reward_pool)
        # Discount reward
        running_add = 0
        for i in reversed(range(len(reward_pool))):
            if reward_pool[i] == 0:
                running_add = 0
            else:
                running_add = running_add * self.gamma + reward_pool[i]
                reward_pool[i] = running_add

        # Normalize reward
        reward_mean = np.mean(reward_pool)
        reward_std = np.std(reward_pool)
        for i in range(len(reward_pool)):
            reward_pool[i] = (reward_pool[i] - reward_mean) / reward_std

        # Gradient Desent
        self.optimizer.zero_grad()

        for i in range(len(reward_pool)):
            state = state_pool[i]
            action = Variable(torch.FloatTensor([action_pool[i]]))
            reward = reward_pool[i]
            state = Variable(torch.from_numpy(state).float())
            probs = self.policy_net(state)
            m = Bernoulli(probs)
            loss = -m.log_prob(action) * reward  # Negtive score function x reward
            # print(loss)
            loss.backward()
        self.optimizer.step()
        self.memory.clear()

1. 模块导入

import torch
from torch.distributions import Bernoulli
from torch.autograd import Variable
import numpy as np

• torch：PyTorch 的核心库，用于张量计算和构建神经网络。
• torch.distributions.Bernoulli：用于构造伯努利分布，这里表示每个动作的采样依据（输出是概率，取值0或1）。
• torch.autograd.Variable：在早期版本的 PyTorch 中用于包装张量以便自动求导；在新版中已不强制要求，但这里代码仍使用它。
• numpy：用于数值计算，数据处理与数组操作。

---

2. PolicyGradient 类定义

class PolicyGradient:

• 定义了一个策略梯度智能体的类，该类不继承其他类，作为一个普通的 Python 类，但内部使用了 PyTorch 模型。

---

2.1 构造函数 init

    def __init__(self, model, memory, cfg):
        self.gamma = cfg['gamma']
        self.device = torch.device(cfg['device']) 
        self.memory = memory
        self.policy_net = model.to(self.device)
        self.optimizer = torch.optim.RMSprop(self.policy_net.parameters(), lr=cfg['lr'])

• 参数说明:

  • model：策略网络（例如 PGNet），用于生成动作概率。
  • memory：轨迹存储器，保存训练过程中（状态、动作、奖励）的采样数据。
  • cfg：配置字典，包含超参数，如折扣因子 gamma、学习率 lr、设备（cpu/cuda）等。

• self.gamma：保存折扣因子，用于计算累计回报；例如 γ=0.9。

• self.device：将字符串配置转换为 torch.device 对象，确定计算设备；例如 ‘cpu’。

• self.memory：引用传入的内存对象，用于后续更新时采样回合数据。

• self.policy_net：将传入的神经网络模型移动到指定设备；例如，将模型送到 CPU。

• self.optimizer：使用 RMSprop 优化器对策略网络参数进行更新，学习率由 cfg[‘lr’] 给出；例如 lr=0.001 或 0.1。

---

2.2 sample_action 方法

    def sample_action(self, state):
        state = torch.from_numpy(state).float()
        state = Variable(state)
        probs = self.policy_net(state)
        m = Bernoulli(probs) # 伯努利分布
        action = m.sample()
        action = action.data.numpy().astype(int)[0] # 转为标量
        return action

• 转换状态：

  • 输入 state 是 NumPy 数组，调用 torch.from_numpy(state).float() 将其转换为浮点型张量。
  • 再用 Variable 包装（在新版 PyTorch 中可直接使用张量，但这里为了兼容旧版仍用 Variable）。

• 通过策略网络计算概率：

  • 将状态输入策略网络 self.policy_net(state)，得到输出 probs，通常表示在当前状态下每个动作被选中的概率。
  • 假设模型输出为 [0.7]（因为 Bernoulli 分布通常用于二分类问题），表示取 1 的概率为 0.7，取 0 的概率为 0.3。

• 构造伯努利分布与采样：

  • m = Bernoulli(probs) 构造伯努利分布对象。
  • 调用 m.sample() 根据概率采样一个动作；例如，可能采样到张量 [1]。

• 转换为标量：

  • action.data.numpy().astype(int)[0] 将采样结果转换为 NumPy 数组，再转换为整数类型，并取第一个元素。
  • 最终返回动作的标量值，例如 1。

---

2.3 predict_action 方法

    def predict_action(self, state):
        state = torch.from_numpy(state).float()
        state = Variable(state)
        probs = self.policy_net(state)
        m = Bernoulli(probs) # 伯努利分布
        action = m.sample()
        action = action.data.numpy().astype(int)[0] # 转为标量
        return action

• 作用：
  与 sample_action 类似，这里 predict_action 同样通过策略网络采样动作。
• 注意：
  在一些实现中，predict_action 通常返回确定性选择（例如 argmax），但这里两者实现完全一致，都通过采样生成动作。这种设计在某些策略梯度方法中也会保留随机性来探索策略空间。

---

2.4 update 方法

    def update(self):
        state_pool, action_pool, reward_pool = self.memory.sample()
        state_pool, action_pool, reward_pool = list(state_pool), list(action_pool), list(reward_pool)

• 作用：
  首先从 memory 中采样一整个回合的数据，分别得到状态、动作和奖励的列表。
• 说明：
  memory.sample() 返回存储在内存中的所有回合数据。将它们转换为列表以便后续处理。

2.4.1 计算折扣奖励

        running_add = 0
        for i in reversed(range(len(reward_pool))):
            if reward_pool[i] == 0:
                running_add = 0
            else:
                running_add = running_add * self.gamma + reward_pool[i]
                reward_pool[i] = running_add

• 作用：
  对回合中的奖励进行折扣处理，从回合末尾向前遍历：

  • reversed(range(len(reward_pool)))：从最后一个奖励开始遍历。
  • 如果当前奖励为 0，则重置 running_add 为 0；否则，更新 running_add 为 running_add * gamma + reward。
  • 将 reward_pool[i] 更新为运行折扣回报 G，从该时刻开始到回合结束的累计回报。

• 举例说明：
  假设 reward_pool = [0, 1, 2, 0, 3]，gamma=0.9，从最后一个奖励开始：

  • i=4：reward=3，running_add = 0*0.9 + 3 = 3；更新 reward_pool[4]=3
  • i=3：reward=0，running_add 变为 0
  • i=2：reward=2，running_add = 0*0.9 + 2 = 2；更新 reward_pool[2]=2
  • i=1：reward=1，running_add = 2*0.9 + 1 = 1.8+1=2.8；更新 reward_pool[1]=2.8
  • i=0：reward=0，running_add 变为 0；更新 reward_pool[0]=0
    最终 reward_pool = [0, 2.8, 2, 0, 3]。

2.4.2 归一化奖励

        reward_mean = np.mean(reward_pool)
        reward_std = np.std(reward_pool)
        for i in range(len(reward_pool)):
            reward_pool[i] = (reward_pool[i] - reward_mean) / reward_std

• 作用：
  对折扣奖励进行归一化（均值为 0，标准差为 1），以便训练时梯度更稳定。
• 举例：
  继续上面的例子，假设 reward_pool = [0, 2.8, 2, 0, 3]，
  • 计算均值和标准差，然后对每个值进行标准化。

2.4.3 梯度下降更新

        self.optimizer.zero_grad()

• 作用：
  清除之前计算的梯度，准备进行新的梯度反向传播。

        for i in range(len(reward_pool)):
            state = state_pool[i]
            action = Variable(torch.FloatTensor([action_pool[i]]))
            reward = reward_pool[i]
            state = Variable(torch.from_numpy(state).float())
            probs = self.policy_net(state)
            m = Bernoulli(probs)
            loss = -m.log_prob(action) * reward  # Negtive score function x reward
            loss.backward()

• 作用：
  对回合中每一步进行策略梯度更新：
  • 从 state_pool[i] 中取出第 i 步的状态，并将其转换为 torch 张量（并包装成 Variable）。
  • 同样，将动作 action_pool[i] 转换为一个浮点数张量，再包装为 Variable。
  • 获取归一化后的 reward 值 reward_pool[i]。
  • 将状态输入策略网络 self.policy_net(state)，得到输出概率 probs。
  • 构造伯努利分布 m = Bernoulli(probs)。
  • 计算 log 概率：m.log_prob(action)，再乘以 reward（注意前面加了负号）。
    • 这就是策略梯度方法中负的得分函数乘以回报。
      举例说明：
      如果策略网络输出概率 probs=0.7（表示取 1 的概率为 0.7），而采样到的动作为 1，则 m.log_prob(1) = log(0.7)。
      假设 reward=2.0，则 loss = - log(0.7)*2.0。
  • 对 loss 进行反向传播，计算梯度。

        self.optimizer.step()

• 作用：
  使用优化器（这里是 RMSprop）更新策略网络参数，使得损失下降，从而提高所采样动作的概率与回报相关性。

        self.memory.clear()

• 作用：
  更新结束后，清除 memory 中保存的回合数据，为下一回合的存储做准备。

---

3. 整体工作流程

1. 构造阶段
   当我们创建 PolicyGradient 实例时，通过构造函数传入模型（policy_net）、内存（memory）和配置（cfg）。

   • 设置 gamma、device 等参数；
   • 将策略网络移动到相应设备；
   • 使用 RMSprop 优化器初始化参数更新。

2. 动作采样
   在训练过程中，每当需要根据当前状态选择动作时，调用 sample_action(state)：

   • 将 NumPy 状态转换为张量，输入到策略网络得到输出概率；
   • 构造 Bernoulli 分布，根据概率采样一个动作，转换为标量返回。
   • 同样 predict_action 实现与 sample_action 相同（这段代码中两者完全一致）。

3. 更新策略
   当一整个回合结束后，调用 update() 方法：

   • 从 memory 中采样该回合的所有状态、动作和奖励数据；
   • 对奖励数据从后向前计算折扣累计回报；
   • 将奖励归一化（标准化）；
   • 对每一步，通过策略网络计算 log 概率，再与归一化奖励相乘构造损失；
   • 对所有步损失累积反向传播，最后通过优化器更新策略网络参数；
   • 清空 memory 为下一次回合做准备。

---

4. 具体数值例子

假设某次采样到的单个状态为 NumPy 数组 state = [0.5, -0.2, 0.1, 0.3]，
经过策略网络后输出概率为 [0.7]（针对二元选择），

• 采样阶段：
  • 将 state 转换为张量：tensor([0.5, -0.2, 0.1, 0.3])；
  • 策略网络输出 probs = 0.7；
  • Bernoulli(probs=0.7) 采样后可能返回 1（意味着采取动作 1）；
  • 返回动作 1。

假设 memory 中存储了一个回合的转移：

state_pool = [state1, state2, state3]
action_pool = [1, 0, 1]
reward_pool = [0, 1, 2]

• 折扣回报计算：
  从最后一步开始，假设 gamma=0.9：

  • 第3步：reward=2，running_add=2，更新 reward_pool[2]=2；
  • 第2步：reward=1，running_add=2*0.9+1 = 1.8+1=2.8，更新 reward_pool[1]=2.8；
  • 第1步：reward=0，running_add 重置为0，更新 reward_pool[0]=0.
    最终 reward_pool=[0, 2.8, 2].

• 归一化：
  均值 = (0+2.8+2)/3 = 4.8/3 ≈ 1.6；
  标准差计算后（假设 std=1.0，为简单起见），归一化后 reward_pool=[-1.6, 1.2, 0.4].

• 梯度下降：
  对于每个步骤，利用当前 state、动作以及归一化奖励计算损失。例如，假设某一步中：

  • state转换后传入网络，网络输出概率为 0.7，
  • 对应动作（假设为1），则 m.log_prob(1) = log(0.7) ≈ -0.357；
  • 归一化 reward = 1.2；
  • 则该步 loss = -(-0.357) * 1.2 ≈ 0.4284（负号确保最大化期望回报）。
    累计每一步 loss 后调用 loss.backward() 计算梯度，最后 optimizer.step() 更新网络参数。

• 更新完成后，memory 被清空，等待下一个回合数据。

---

总结

• 策略网络构造：
  PolicyGradient 类接收一个模型（例如 PGNet）、memory 以及配置参数，通过 RMSprop 优化器更新策略网络。
• 采样动作：
  sample_action() 和 predict_action() 将状态转换为张量，通过模型计算概率，然后利用 Bernoulli 分布采样动作。
• 更新过程：
  update() 从 memory 中采样整个回合数据，计算折扣回报（MC 方法），对奖励进行归一化，然后对每一步计算损失（-log(prob) * reward）并进行反向传播和参数更新。
• 具体数值例子帮助理解每一步更新如何基于回报调整网络参数。

这就是策略梯度方法中典型的实现：通过采样获得策略输出、利用整个回合的数据计算累计折扣回报，再以负对数概率乘以归一化回报作为损失进行梯度更新，从而使得高回报动作的概率增大。