策略梯度算法 (REINFORCE) 详解

3.2、策略梯度公式推导

3.3、核心结论

4、经典算法：REINFORCE（基础策略梯度）

4.1、算法流程

4.2、代码核心逻辑（对应之前的 REINFORCE 实现）

4.3、特点与问题

5、详解代码

6、实验结果

1、什么是基于价值的方法？什么是基于策略的方法？

1.1、基于价值的方法（Value-based Methods）：先算'好处'，再选行动

想象你在玩一款闯关游戏，每一步选择（比如往左走、往右走、打怪）都会影响你最终能不能通关、能拿多少分。 基于价值的方法会先给每个'局面'打分 —— 这个分数代表'在这个局面下，只要好好玩，最后能得到的好处有多大'。 然后，它的决策逻辑很简单：在当前局面下，选那些能让你进入'更高分局面'的行动。

比如：现在你面前有两条路，左边的路对应的'未来好处分'是 80 分，右边是 60 分。那它就会选左边的路。

核心：先判断'每个选择的最终价值'，再跟着高价值走。

基于价值的方法的核心是学习'价值函数'，通过价值函数间接指导动作选择。

1.1.1、核心逻辑

价值函数（Value Function）是强化学习中的核心概念，它量化了'在某个状态下采取动作（或遵循策略）能获得的长期累积奖励的期望'。基于价值的方法通过学习价值函数，最终根据价值函数的输出选择'最优动作'（即能带来最高价值的动作）。

1.1.2、关键概念：价值函数的类型

动作价值函数

：在状态 s 下采取动作 a，之后遵循策略 π 能获得的累积奖励的期望（更常用，因为直接关联'状态 - 动作对'）。

状态价值函数

：在状态 s 下，遵循策略 π 能获得的累积奖励的期望。

基于价值的方法通常聚焦于学习 Q 函数（动作价值函数），因为它能直接告诉我们'在某个状态下做什么动作更好'。

1.1.3、动作选择逻辑

当学习到最优的 Q 函数（记为 Q^(s,a)）后，在任意状态 s 下，只需选择使 Q^(s,a) 最大的动作 a 即可，即：

a^* = \arg\max_a Q^*(s,a)

1.1.4、典型算法与例子

Deep Q-Networks（DQN）：用深度神经网络近似 Q(s,a)，解决高维状态空间（如 Atari 游戏画面）的问题。

SARSA：学习当前策略下的动作价值函数 Q^π(s,a)，属于'同策略'（On-policy）方法。

Q-learning：直接学习最优动作价值函数 Q^*(s,a)，不依赖当前策略，属于'异策略'（Off-policy）方法。

1.1.5、优缺点

• 优点：
  • 思路直观，通过价值函数可直接判断动作的优劣；
  • 在离散动作空间中表现稳定（因为选最大价值动作简单）。
• 缺点：
  • 难以处理连续动作空间（因为连续空间中'找最大价值动作'需要复杂的优化，如数值搜索）；
  • 价值函数的估计可能存在偏差（如 DQN 中因经验回放和目标网络更新延迟导致的偏差）。

1.2、基于策略的方法（Policy-based Methods）：直接学'怎么做'，不纠结后果

还是玩闯关游戏，但这种方法不先算'每个局面的价值'，而是直接学一套'行动规则'。 比如通过大量练习，它总结出：'看到 A 怪物就后退，看到 B 道具就捡，在 C 路口就往右拐'—— 这套规则就叫'策略'。 哪怕有时候按这套规则做，短期看可能吃亏（比如绕了点路），但它还是会坚持按规则来，因为这套规则是长期试错后'大概率能赢'的总结。

核心：直接记住'在什么情况下该做什么'，不纠结'这么做能有多少好处'。

基于策略的方法的核心是直接学习'策略函数'，通过策略函数直接输出动作。

1.2.1、核心逻辑

策略函数（Policy Function）是从'状态'到'动作'的映射，分为两种类型：

随机性策略：

π(a|s)（给定状态 s，输出动作 a 的概率分布）。

确定性策略：

a = π(s)（给定状态 s，直接输出唯一动作 a）；

基于策略的方法通过优化策略函数的参数，使策略对应的累积奖励期望最大化，最终得到最优策略 π^*。

1.2.2、动作选择逻辑

对于随机性策略：从概率分布 π(a|s) 中采样动作（兼顾探索与利用）。

对于确定性策略：直接输出动作 a = π(s)；

1.2.3、典型算法与例子

• REINFORCE：通过'蒙特卡洛采样'估计策略的梯度，用梯度上升法优化策略参数（同策略方法）。
• Proximal Policy Optimization（PPO）：通过限制策略更新的幅度（'信任域'），解决 REINFORCE 中梯度估计方差大的问题，是目前最常用的强化学习算法之一。
• Deterministic Policy Gradient（DPG）：针对确定性策略的梯度优化方法，适合连续动作空间。

1.2.4、优缺点

• 优点：
  • 天然适合连续动作空间（随机性策略可直接输出概率分布，无需搜索最大价值动作）；
  • 策略更新更直接，避免了价值函数估计的偏差问题。
• 缺点：
  • 梯度估计的方差较大（需通过'基线函数'或'Actor-Critic 框架'缓解）；
  • 在离散动作空间中，可能不如基于价值的方法高效。

1.3、核心区别对比

2、核心思想：直接优化策略

在强化学习中，智能体的目标是学习一个'策略'（Policy），即从'状态'到'动作'的映射，使长期累积奖励最大化。策略梯度算法的核心逻辑可概括为：通过计算'累积奖励对策略参数的梯度'，用梯度上升法不断调整参数，最终得到最优策略。

2.1、什么是'策略'？

策略是强化学习的核心概念，定义为智能体在给定状态下选择动作的规则，分为两种形式：

随机性策略：

π_θ(a|s)（给定状态 s，输出动作 a 的概率分布，策略参数 θ 决定分布形状）。

确定性策略：

a = π_θ(s)（给定状态 s，直接输出唯一动作 a，θ 为策略参数）。

策略梯度算法通常聚焦于随机性策略，因为随机性策略天然支持'探索'（从概率分布中采样动作），更适合强化学习的试错过程。

2.2、与基于价值的方法对比

3、数学基础：策略梯度的推导

策略梯度算法的核心是计算累积奖励期望对策略参数 θ 的梯度，并通过梯度上升最大化这个期望。

3.1、目标函数：累积奖励的期望

设智能体与环境交互产生的轨迹为 τ = (s_0,a_0,r_0,s_1,a_1,r_1,...,s_T,a_T,r_T)，轨迹的累积奖励为 R(τ) = \sum_{t=0}^T \gamma^t r_t（γ 为折扣因子）。

策略梯度的目标是最大化'策略 π_θ 产生的累积奖励期望'：

J(θ) = \mathbb{E}_{τ ~ π_θ} [R(τ)]

其中 \mathbb{E}_{τ ~ π_θ} 表示对所有可能轨迹的期望（轨迹由策略 π_θ 生成）。

3.2、策略梯度公式推导

目标是求 ∇_θ J(θ)（策略梯度），关键步骤如下：

步骤 1：展开期望 轨迹的概率 P(τ|θ) = μ(s_0) \prod_{t=0}^T π_θ(a_t|s_t) P(s_{t+1}|s_t,a_t)（μ(s_0) 是初始状态分布，P(s_{t+1}|s_t,a_t) 是环境动力学）。目标函数可写为：

J(θ) = \sum_τ P(τ|θ) R(τ)

步骤 2：求梯度 根据导数链式法则：

∇_θ J(θ) = \sum_τ ∇_θ P(τ|θ) R(τ)

步骤 3：对数导数技巧 利用 ∇_θ P(τ|θ) = P(τ|θ) ∇_θ log P(τ|θ)（对数导数性质），化简得：

∇θ J(θ) = \mathbb{E}{τ ~ π_θ} [∇_θ log P(τ|θ) · R(τ)]

步骤 4：简化轨迹概率的对数导数 轨迹概率的对数为 log P(τ|θ) = log μ(s_0) + \sum_{t=0}^T log π_θ(a_t|s_t) + \sum_{t=0}^T log P(s_{t+1}|s_t,a_t)。由于环境动力学 P(s_{t+1}|s_t,a_t) 与 θ 无关，其导数为 0，因此：

∇θ log P(τ|θ) = \sum{t=0}^T ∇_θ log π_θ(a_t|s_t)

最终策略梯度公式 合并上述结果，策略梯度为：

∇θ J(θ) = \mathbb{E}{τ ~ π_θ} \left[ \left( \sum_{t=0}^T ∇_θ log π_θ(a_t|s_t) \right) · R(τ) \right]

3.3、核心结论

策略梯度的直观含义是：对每个状态 - 动作对 (s_t,a_t)，用'该动作的对数概率梯度'乘以'整个轨迹的累积奖励'，再求期望。

实际计算中，用蒙特卡洛采样（即实际轨迹）近似期望，得到可计算的梯度：

∇θ J(θ) ≈ \frac{1}{N} \sum{i=1}^N \left( \sum_{t=0}^T ∇θ log π_θ(a{i,t}|s_{i,t}) \right) · R(τ_i)

其中 N 是采样的轨迹数量，τ_i 是第 i 条轨迹。

4、经典算法：REINFORCE（基础策略梯度）

REINFORCE 是最基础的策略梯度算法，用单条轨迹的累积奖励直接估计梯度，步骤如下：

4.1、算法流程

采样轨迹：用当前策略 π_θ 与环境交互，收集一条完整轨迹 τ = (s_0,a_0,r_0,...,s_T,a_T,r_T)。

计算累积奖励：对每条轨迹计算折扣累积奖励 G_t = \sum_{k=t}^T \gamma^{k-t} r_k（从时刻 t 到结束的总奖励）。

计算梯度：对每个时刻 t，计算损失 L_t = -log π_θ(a_t|s_t) · G_t（负号将梯度上升转为梯度下降优化）。

更新参数：对所有时刻的损失求和，反向传播更新策略参数 θ。

4.2、代码核心逻辑（对应之前的 REINFORCE 实现）

def update(self, transition_dict):
    reward_list = transition_dict['rewards']
    state_list = transition_dict['states']
    action_list = transition_dict['actions']
    G = 0 # 累积奖励（从后往前计算）
    self.optimizer.zero_grad()
    for i in reversed(range(len(reward_list))): # 逆序计算 G
        reward = reward_list[i]
        state = torch.tensor([state_list[i]], dtype=torch.float).to(device)
        action = torch.tensor([action_list[i]]).view(-1, 1).to(device)
        # 动作的对数概率
        log_prob = torch.log(self.policy_net(state).gather(1, action))
        G = self.gamma * G + reward # 折扣累积奖励
        # 损失：-log_prob * G（梯度上升→转为梯度下降）
        loss = -log_prob * G
        loss.backward() # 累积梯度
    self.optimizer.step() # 梯度上升更新参数

4.3、特点与问题

• 优点：原理简单，直接优化目标函数，理论上可收敛到全局最优。
• 问题：
  • 高方差：单条轨迹的累积奖励波动大，导致梯度估计不稳定。
  • 样本效率低：需要大量轨迹才能得到可靠的梯度。

5、详解代码

"""
REINFORCE 算法（蒙特卡洛策略梯度）解决 CartPole-v0 环境
CartPole 任务：通过左右移动小车，使杆子保持直立，每保持 1 步得 1 分，目标是最大化得分
"""
# 导入必要库
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_utils # 自定义强化学习工具库（包含经验回放、移动平均等功能）

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)) # 隐藏层用 ReLU 激活（引入非线性，增强拟合能力）
        # 输出层用 softmax 归一化，将未归一化概率转为概率分布（每行和为 1）
        return F.softmax(self.fc2(x), dim=1) # dim=1 表示按行归一化

class REINFORCE:
    """REINFORCE 算法实现：蒙特卡洛策略梯度算法"""
    def __init__(self, state_dim, # 状态维度（CartPole 为 4：位置、速度、角度、角速度）
                 hidden_dim, # 策略网络隐藏层维度
                 action_dim, # 动作维度（CartPole 为 2：左移、右移）
                 learning_rate, # 学习率（控制参数更新速度）
                 gamma, # 折扣因子（权衡当前奖励和未来奖励）
                 device): # 计算设备（CPU 或 GPU）
        # 初始化策略网络（主网络，直接输出动作概率）
        self.policy_net = PolicyNet(state_dim, hidden_dim, action_dim).to(device)
        # 优化器（Adam 优化器，用于更新策略网络参数）
        self.optimizer = torch.optim.Adam(self.policy_net.parameters(), lr=learning_rate)
        self.gamma = gamma # 折扣因子（如 0.98：未来奖励随时间衰减）
        self.device = device # 计算设备

    def take_action(self, state):
        """根据当前状态选择动作（ε-探索的替代方案：基于概率分布采样）"""
        # 将状态转换为张量（形状：[1, state_dim]），并移动到指定设备
        state = torch.tensor([state], dtype=torch.float).to(self.device)
        # 策略网络输出动作概率分布（形状：[1, action_dim]）
        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 = 0 # 累积回报（从当前步到轨迹结束的总折扣奖励）
        self.optimizer.zero_grad() # 清空优化器中的梯度（避免累积旧梯度）
        # 逆序遍历轨迹（从最后一步到第一步）
        # 原因：累积回报 G_t = r_t + γ·G_{t+1}，逆序计算更高效
        for i in reversed(range(len(reward_list))):
            # 当前步的奖励（r_t）
            reward = reward_list[i]
            # 当前步的状态（s_t），转换为张量
            state = torch.tensor([state_list[i]], dtype=torch.float).to(self.device)
            # 当前步的动作（a_t），转换为张量并调整形状为 [1,1]
            action = torch.tensor([action_list[i]]).view(-1, 1).to(self.device)
            # 1. 计算当前动作的对数概率（log(π(a_t|s_t))）
            # policy_net(state) 输出动作概率分布，gather(1, action) 提取当前动作的概率
            # 例如：动作概率为 [0.3, 0.7]，动作是 1（右移），则提取 0.7，取对数为 log(0.7)
            log_prob = torch.log(self.policy_net(state).gather(1, action))
            # 2. 计算累积回报 G（逆序递推）
            # 公式：G_t = r_t + γ·G_{t+1}（最后一步 G=0 + r_T）
            G = self.gamma * G + reward
            # 3. 计算策略梯度损失（核心公式）
            # 损失 = -log(π(a_t|s_t)) * G_t
            # 负号：因为 PyTorch 默认最小化损失，而我们需要最大化累积回报（梯度上升）
            # G_t：奖励越高，该动作的"重要性"越大，梯度更新幅度越大
            loss = -log_prob * G
            # 4. 反向传播计算梯度（累积所有时间步的梯度）
            # 注意：这里没有立即更新参数，而是累积所有步的梯度后统一更新
            loss.backward()
            # 5. 梯度上升更新策略网络参数（所有时间步的梯度累积后，统一更新一次）
            self.optimizer.step()

if __name__ == '__main__':
    # 超参数设置
    learning_rate = 1e-3 # 学习率（1e-3：较常用的初始值）
    num_episodes = 1000 # 训练总回合数（越多越可能收敛，但耗时更长）
    hidden_dim = 128 # 策略网络隐藏层维度（128：兼顾表达能力和计算效率）
    gamma = 0.98 # 折扣因子（0.98：适度重视未来奖励）
    # 选择计算设备（优先 GPU，无则用 CPU）
    device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")
    # 创建 CartPole 环境（v0 版本：最大步数 200，超过则强制终止）
    env_name = "CartPole-v0"
    env = gym.make(env_name) # 创建环境实例
    # 设置随机种子（保证实验可复现）
    state, _ = env.reset(seed=0) # 初始化环境并设置种子，返回初始状态和信息
    torch.manual_seed(0) # PyTorch 随机种子
    np.random.seed(0) # NumPy 随机种子
    # 获取环境状态和动作维度
    state_dim = env.observation_space.shape[0] # 状态维度（CartPole 为 4）
    action_dim = env.action_space.n # 动作数量（CartPole 为 2）
    # 初始化 REINFORCE 智能体
    agent = REINFORCE(state_dim, hidden_dim, action_dim, learning_rate, gamma, device)
    # 训练循环
    return_list = [] # 记录每个回合的总奖励（用于评估训练效果）
    # 分 10 轮训练（每轮 100 个回合，便于显示进度）
    for i in range(10): # 进度条：显示当前轮次的训练进度
        with tqdm(total=int(num_episodes / 10), desc='Iteration %d' % i) as pbar:
            # 每轮训练 100 个回合
            for i_episode in range(int(num_episodes / 10)):
                episode_return = 0 # 记录当前回合的总奖励
                # 经验字典：存储当前回合的轨迹数据
                transition_dict = {
                    'states': [], # 状态列表
                    'actions': [], # 动作列表
                    'next_states': [], # 下一状态列表
                    'rewards': [], # 奖励列表
                    'dones': [] # 终止标志列表
                }
                # 重置环境，获取初始状态（Gym v0.26+ 返回元组：(state, info)）
                state, _ = env.reset(seed=i_episode) # 每个回合用不同种子，增强泛化性
                done = False # 回合终止标志（初始为 False）
                # 循环执行动作，直到回合终止
                while not done:
                    # 智能体根据当前状态选择动作
                    action = agent.take_action(state)
                    # 执行动作，获取下一状态、奖励等（Gym v0.26+ 返回 5 个值）
                    # next_state：下一状态；reward：当前奖励；
                    # terminated：是否因任务目标终止；truncated：是否因超时终止
                    next_state, reward, terminated, truncated, _ = env.step(action)
                    # 合并终止条件（任务结束或超时均视为回合结束）
                    done = terminated or truncated
                    # 存储当前步的经验到字典
                    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_state
                    episode_return += reward # 累积当前回合的奖励
                # 回合结束后，记录总奖励
                return_list.append(episode_return)
                # 用当前回合的轨迹数据更新策略网络
                agent.update(transition_dict)
                # 每 10 个回合显示一次平均奖励（监控训练效果）
                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) # 9 点移动平均
    plt.plot(episodes_list, mv_return)
    plt.xlabel('Episodes')
    plt.ylabel('Returns')
    plt.title('REINFORCE on {}'.format(env_name))
    plt.show()

6、实验结果