最新资讯

目录

一、引言

二、REINFORCE 算法的核心定位与关键特性

三、REINFORCE 算法的数学基础（通俗推导，贴合代码）

1. 核心目标：最大化策略的期望累计回报

2. 关键推导：策略梯度的无偏估计

3. 工程化转化：梯度下降替代梯度上升

四、代码与算法的逐模块对应实现

模块 1：策略网络（PolicyNet）—— 实现策略πθ​(a∣s)的参数化

模块 2：REINFORCE 算法核心类 —— 实现策略采样、累计回报计算、策略梯度更新

2.1 __init__：初始化算法核心组件

2.2 take_action：根据策略πθ​(a∣s)采样动作（同策略探索）

2.3 update：REINFORCE 算法的核心 —— 实现蒙特卡洛策略梯度更新

模块 3：超参数与环境初始化 —— 工程化适配，保证实验可复现

模块 4：蒙特卡洛训练循环 —— 采集完整轨迹，实现回合式更新

模块 5：结果可视化 —— 分析训练收敛趋势

五、REINFORCE 算法的完整执行流程（结合代码）

六、REINFORCE 算法的Python代码完整实现

七、程序运行结果展示

八、REINFORCE 算法的优缺点（结合代码表现）

优点（代码完美体现）

缺点（代码训练中会体现）

九、经典改进方向（解决高方差 / 低效率问题）

十、总结

一、引言

REINFORCE 算法是策略梯度（Policy Gradient, PG） 家族中最经典、最基础的无模型强化学习算法，由 Williams 在 1992 年提出。核心特点是直接参数化策略、通过蒙特卡洛（MC）方式采集完整回合数据、用梯度上升最大化期望累计回报，区别于 DQN 这类先学习值函数再推导策略的 “间接法”，REINFORCE 是直接优化策略的 “直接法”，非常适合离散动作空间（如 CartPole-v1 的左右两个动作）。

二、REINFORCE 算法的核心定位与关键特性

先明确算法的核心属性，这是理解代码设计的基础，代码的每一处实现都贴合这些特性：

1. 无模型（Model-Free）：无需学习环境的状态转移模型P(s′∣s,a)和奖励模型R(s,a)，仅通过与环境交互采集的轨迹数据更新策略，代码中仅用env.step()和env.reset()交互，无任何环境模型相关代码。
2. 同策略（On-Policy）：必须用当前策略采样的轨迹数据来更新策略，采样和更新的策略是同一个，代码中每回合用agent.take_action()（当前策略）采样动作，采集完轨迹后直接用该轨迹更新策略，无经验回放（经验回放是异策略的典型特征）。
3. 蒙特卡洛（MC）式更新：必须采集完整的回合轨迹（从初始状态到终止状态），只有拿到回合的全部奖励，才能计算每一步的累计折扣回报，代码中每个回合结束后才调用agent.update()，而非步更 / 半回合更，这是蒙特卡洛的核心标志。
4. 离散动作空间适配：策略网络输出动作的概率分布，通过类别分布（Categorical）采样动作，代码中用torch.distributions.Categorical实现，完美适配 CartPole-v1 的二分类动作。
5. 梯度上升最大化回报：目标是找到策略参数，让策略的期望累计回报最大；因 PyTorch 仅支持梯度下降，故将 “最大化回报” 转化为 “最小化负的回报损失”，代码中损失函数为- log_prob * G就是这个逻辑。

三、REINFORCE 算法的数学基础（通俗推导，贴合代码）

REINFORCE 的核心是推导出策略梯度的无偏估计，让代码的更新逻辑有理论支撑。

1. 核心目标：最大化策略的期望累计回报

我们用参数θ表示策略网络（代码中是 PolicyNet 的权重），策略记为，表示在状态s下，策略θ选择动作a的概率。

算法的核心目标是找到最优参数θ∗，让策略的期望累计折扣回报J(θ)最大：

其中​是从初始状态开始的累计折扣回报，​​表示对策略​采样的所有轨迹求期望。

2. 关键推导：策略梯度的无偏估计

要最大化J(θ)，需要计算J(θ)对参数θ的梯度（即策略梯度），然后用梯度上升更新θ：

（α是学习率，代码中由 Adam 优化器自适应调整）

Williams 通过数学推导，得出策略梯度的无偏蒙特卡洛估计（这是 REINFORCE 的核心公式）：

• N：训练的回合数（代码中每次 update 对应 1 个回合，N=1）；
• ​：第n个回合的步数；
• ：第n回合t步，策略选择动作an,t​的对数概率（代码中log_prob）；
• ,t​：第n回合t步开始到回合结束的累计折扣回报（代码中G），计算式为：

是折扣因子（代码中gamma=0.98），用于衰减远期奖励的权重，让算法更关注近期奖励。

3. 工程化转化：梯度下降替代梯度上升

PyTorch 等深度学习框架仅支持梯度下降（最小化损失函数），因此将策略梯度的最大化目标转化为最小化负的策略梯度损失：

​这就是代码中loss = - log_prob * G的理论根源，反向传播该损失的梯度，等价于对原目标做梯度上升。

四、代码与算法的逐模块对应实现

代码分为策略网络（PolicyNet）、REINFORCE 算法核心类、超参数与环境初始化、蒙特卡洛训练循环、结果可视化5 个模块，每个模块都严格对应 REINFORCE 算法的理论要求，下面逐模块讲解。

模块 1：策略网络（PolicyNet）—— 实现策略πθ​(a∣s)的参数化

策略网络是 REINFORCE 的 “核心载体”，作用是将状态s映射为动作a的概率分布，代码中是两层全连接网络，适配 CartPole-v1 的简单状态空间（4 维观测）。

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)  # 转概率分布（和为1）

算法对应点：

1. 输入state_dim（CartPole-v1 为 4）：环境的观测维度，对应状态s；
2. 输出action_dim（CartPole-v1 为 2）：离散动作的数量，输出值经softmax激活后，直接表示πθ​(a∣s)（状态s下选每个动作的概率）；
3. softmax的必要性：保证输出是合法的概率分布（所有动作的概率和为 1），这是策略πθ​(a∣s)的基本要求。

模块 2：REINFORCE 算法核心类 —— 实现策略采样、累计回报计算、策略梯度更新

这是代码的核心，完全实现了 REINFORCE 的三大核心功能：动作采样、累计折扣回报Gt​计算、策略梯度更新，分为__init__、take_action、update三个方法。

2.1 __init__：初始化算法核心组件

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)  # 梯度优化器
    self.gamma = gamma     # 折扣因子γ，用于计算G_t
    self.device = device   # 训练设备（GPU/CPU）

关键细节：用 Adam 自适应优化器替代原始的随机梯度上升（SGD），提升训练稳定性和收敛速度，这是工程实现的常规优化，不改变算法核心。

2.2 take_action：根据策略πθ​(a∣s)采样动作（同策略探索）

def take_action(self, state):
    # 状态转张量，消除创建警告，适配设备
    state = torch.tensor(np.array([state]), dtype=torch.float).to(self.device)
    probs = self.policy_net(state)  # 得到动作概率分布π_θ(a|s)
    action_dist = torch.distributions.Categorical(probs)  # 构建类别分布
    action = action_dist.sample()  # 从概率分布中随机采样动作
    return action.item()

算法对应点：

1. 非贪心采样：策略梯度算法需要探索环境，因此不能选概率最大的动作（贪心），而是从概率分布中随机采样（比如动作 1 概率 0.8、动作 2 概率 0.2，大概率选 1 但小概率选 2，保证探索）；
2. 类别分布（Categorical）：专门用于离散动作的概率采样，输入动作概率分布，输出采样的动作索引，完美适配 REINFORCE 的离散动作要求；
3. 同策略采样：每次采样用的都是当前策略网络的最新参数，符合 “同策略” 特性。

2.3 update：REINFORCE 算法的核心 —— 实现蒙特卡洛策略梯度更新

这部分代码完全对应策略梯度的数学公式，是算法的 “灵魂”，必须在采集完完整回合轨迹后调用。

def update(self, transition_dict):
    # 从轨迹字典中取出当前回合的所有s/a/r
    reward_list = transition_dict['rewards']
    state_list = transition_dict['states']
    action_list = transition_dict['actions']

    G = 0  # 初始化累计折扣回报
    self.optimizer.zero_grad()  # 清空梯度（PyTorch梯度累加，必须手动清空）
    # 从最后一步反向遍历，计算每一步的G_t（核心！）
    for i in reversed(range(len(reward_list))):
        reward = reward_list[i]  # 当前步的奖励r_t
        # 状态转张量，适配设备
        state = torch.tensor(np.array([state_list[i]]), dtype=torch.float).to(self.device)
        # 动作转张量并reshape，匹配gather的维度要求
        action = torch.tensor([action_list[i]]).view(-1, 1).to(self.device)
        # 计算log π_θ(a_t|s_t)：取当前动作对应的对数概率
        log_prob = torch.log(self.policy_net(state).gather(1, action))
        # 反向计算累计折扣回报G_t = r_t + γ * G_{t+1}
        G = self.gamma * G + reward
        # 计算单步损失：- log_prob * G（梯度下降替代梯度上升）
        loss = - log_prob * G
        loss.backward()  # 单步反向传播，累加梯度（核心！逐步计算，梯度累加）
    self.optimizer.step()  # 一次更新所有参数（同策略MC的特性）

算法核心细节（逐行对应理论）：

1. 反向遍历计算Gt​：从回合最后一步（t=T−1）往回算，是计算累计折扣回报最高效的方式，公式为​：
   • 最后一步（t=T−1）：​（后面无奖励，G=0）；
   • 倒数第二步（t=T−2）：​；
   • 以此类推，直到第一步（t=0），完美计算每一步的。
2. gather(1, action)：从策略网络输出的动作概率分布中，取出当前采样动作对应的概率（比如输出 [0.2, 0.8]，采样动作 1，就取 0.8），再取对数得到。
3. 梯度累加（loss.backward()在循环内）：策略梯度的公式是所有步的梯度之和，因此这里逐步计算损失并反向传播，让 PyTorch 自动累加梯度，最后一次更新参数，这是 REINFORCE 蒙特卡洛更新的标准实现。
4. 一次更新参数（optimizer.step()在循环外）：遍历完所有步、累加完所有梯度后，才调用优化器更新策略网络的参数，符合 “完整回合后更新” 的蒙特卡洛特性。

模块 3：超参数与环境初始化 —— 工程化适配，保证实验可复现

这部分是强化学习的工程实现基础，无算法理论，但直接影响训练效果和实验可复现性，代码中做了 Gymnasium 的完整适配。

# 超参数设置
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")  # 自动选设备

# 环境初始化（Gymnasium适配）
env_name = "CartPole-v1"
# 设置全局随机种子，保证实验可复现
np.random.seed(0)
torch.manual_seed(0)
random.seed(0)
env = gym.make(env_name)
# Gymnasium移除env.seed()，改用动作/观测空间种子
env.action_space.seed(0)
env.observation_space.seed(0)

# 获取环境维度
state_dim = env.observation_space.shape[0]  # 状态维度：4
action_dim = env.action_space.n              # 动作维度：2
# 实例化REINFORCE智能体
agent = REINFORCE(state_dim, hidden_dim, action_dim, learning_rate, gamma, device)

关键工程细节：

1. 随机种子设置：强化学习的训练结果受随机采样（动作、环境）影响，设置种子后，每次训练的结果完全一致，方便调参和验证；
2. Gymnasium 适配：Gym 自 2022 年废弃，Gymnasium 是官方无缝替代，代码中处理了reset/step的返回值变化（后续训练循环会详细讲）；
3. 设备自适应：自动选择 GPU（如果有）加速训练，无 GPU 则用 CPU，不影响算法逻辑。

模块 4：蒙特卡洛训练循环 —— 采集完整轨迹，实现回合式更新

这部分代码实现REINFORCE 的蒙特卡洛核心要求：与环境交互，采集完整的回合轨迹，存储所有s/a/r/s′/done，回合结束后调用agent.update()更新策略，是 “算法理论” 到 “环境交互” 的桥梁。

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 = 0  # 存储当前回合的总回报
            # 轨迹字典：存储当前回合的所有s/a/r/s'/done（蒙特卡洛需要完整轨迹）
            transition_dict = {'states': [], 'actions': [], 'next_states': [], 'rewards': [], 'dones': []}
            state, _ = env.reset(seed=0)  # Gymnasium：reset返回(state, info)
            done = False
            while not done:
                action = agent.take_action(state)  # 同策略采样动作
                # Gymnasium：step返回(next_state, reward, terminated, truncated, info)
                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
            # 回合结束：存储总回报，调用update更新策略（蒙特卡洛核心！）
            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)

算法与工程对应点：

1. 完整轨迹采集：while not done循环直到回合终止，保证采集的是从初始状态到终止状态的完整轨迹，符合蒙特卡洛 “完整回合更新” 的要求；
2. Gymnasium 适配：处理了 Gymnasium 与原 Gym 的返回值差异（核心工程适配，不影响算法）：
   • env.reset(seed=0)：返回(state, info)，取第一个元素为状态；
   • env.step(action)：返回(next_state, reward, terminated, truncated, info)，其中terminated是环境自然终止（如 CartPole 倒了），truncated是步数截断（如 CartPole 撑到 500 步），合并为done表示回合结束；
3. 回合式更新：只有当while not done循环结束（回合终止），才调用agent.update(transition_dict)，将完整轨迹传入更新策略，这是 REINFORCE 蒙特卡洛特性的核心工程实现；
4. 回报监控：用return_list存储每个回合的总回报，每 10 回合计算一次平均回报，用于监控训练收敛趋势（CartPole-v1 的最大回报是 500，平均回报稳定在 500 表示算法收敛）。

模块 5：结果可视化 —— 分析训练收敛趋势

强化学习的训练过程存在大量噪声（因动作是随机采样的），直接绘制原始回报曲线会波动剧烈，因此代码中绘制了原始回报曲线和移动平均（Moving Average）回报曲线，后者用于平滑噪声，清晰展示收敛趋势。

# 绘制原始回报曲线并保存
episode_list = list(range(len(return_list)))
plt.plot(episode_list, return_list)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('REINFORCE on {}'.format(env_name))
plt.savefig('REINFORCE on CartPole-v1_training.png')
plt.show()

# 绘制移动平均回报曲线并保存（窗口大小9，平滑噪声）
mv_return = rl_utils.moving_average(return_list, 9)
plt.plot(episode_list, mv_return)
plt.xlabel('Episodes')
plt.ylabel('Moving Average Returns')
plt.title('REINFORCE on {}'.format(env_name))
plt.savefig('REINFORCE on CartPole-v1_moving_average.png')
plt.show()

关键作用：

• 原始回报曲线：展示每回合的真实训练效果，能看到噪声波动；
• 移动平均曲线：用rl_utils.moving_average计算窗口内的平均回报，平滑噪声，能清晰看到算法是否收敛、收敛速度、收敛稳定性（比如 CartPole-v1 的移动平均曲线会逐渐上升，最终稳定在 500）。

五、REINFORCE 算法的完整执行流程（结合代码）

将上述模块串联，代码的执行流程就是 REINFORCE 算法的标准执行流程，一步一步清晰明了：

1. 初始化：创建 CartPole-v1 环境，初始化策略网络和 REINFORCE 智能体，设置超参数和随机种子；
2. 循环训练回合：总共有 1000 个训练回合，分 10 次迭代，每次 100 回合；
3. 采集完整轨迹：对每个回合，从env.reset()开始，用当前策略agent.take_action()采样动作，通过env.step()与环境交互，采集每一步的s/a/r/s′/done，直到回合终止，得到完整轨迹；
4. 蒙特卡洛策略梯度更新：回合终止后，将完整轨迹传入agent.update()，反向计算每一步的累计折扣回报Gt​，计算策略梯度并累加，最后一次更新策略网络的参数；
5. 监控训练进度：存储每个回合的总回报，每 10 回合打印平均回报，监控收敛趋势；
6. 可视化结果：训练结束后，绘制原始回报曲线和移动平均回报曲线，分析算法收敛效果；
7. 收敛标志：当 CartPole-v1 的回合平均回报稳定在 500（环境最大步数），表示策略收敛，智能体能稳定让小车保持平衡。

六、REINFORCE 算法的Python代码完整实现

先实现rl_utils库，它包含一些函数，如绘制移动平均曲线、计算优势函数等，不同的算法可以一起使用这些函数。

rl_utils.py中的Python代码如下：

from tqdm import tqdm
import numpy as np
import torch
import collections
import random


class ReplayBuffer:
    def __init__(self, capacity):
        self.buffer = collections.deque(maxlen=capacity)

    def add(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))

    def sample(self, batch_size):
        transitions = random.sample(self.buffer, batch_size)
        state, action, reward, next_state, done = zip(*transitions)
        return np.array(state), action, reward, np.array(next_state), done

    def size(self):
        return len(self.buffer)


def moving_average(a, window_size):
    cumulative_sum = np.cumsum(np.insert(a, 0, 0))
    middle = (cumulative_sum[window_size:] - cumulative_sum[:-window_size]) / window_size
    r = np.arange(1, window_size - 1, 2)
    begin = np.cumsum(a[:window_size - 1])[::2] / r
    end = (np.cumsum(a[:-window_size:-1])[::2] / r)[::-1]
    return np.concatenate((begin, middle, end))


def train_on_policy_agent(env, agent, num_episodes):
    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 = 0
                transition_dict = {'states': [], 'actions': [], 'next_states': [], 'rewards': [], 'dones': []}
                state, _ = env.reset()  # 适配Gymnasium：reset返回(state, info)
                done = False
                while not done:
                    action = agent.take_action(state)
                    # 适配Gymnasium：step返回(next_state, reward, terminated, truncated, info)
                    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)
                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)
    return return_list


def train_off_policy_agent(env, agent, num_episodes, replay_buffer, minimal_size, batch_size):
    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 = 0
                state, _ = env.reset()  # 适配Gymnasium
                done = False
                while not done:
                    action = agent.take_action(state)
                    # 适配Gymnasium step返回值
                    next_state, reward, terminated, truncated, _ = env.step(action)
                    done = terminated or truncated
                    replay_buffer.add(state, action, reward, next_state, done)
                    state = next_state
                    episode_return += reward
                    if replay_buffer.size() > minimal_size:
                        b_s, b_a, b_r, b_ns, b_d = replay_buffer.sample(batch_size)
                        transition_dict = {'states': b_s, 'actions': b_a, 'next_states': b_ns, 'rewards': b_r,
                                           'dones': b_d}
                        agent.update(transition_dict)
                return_list.append(episode_return)
                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)
    return return_list


def compute_advantage(gamma, lmbda, td_delta):
    td_delta = td_delta.detach().numpy()
    advantage_list = []
    advantage = 0.0
    for delta in td_delta[::-1]:
        advantage = gamma * lmbda * advantage + delta
        advantage_list.append(advantage)
    advantage_list.reverse()
    return torch.tensor(advantage_list, dtype=torch.float)

REINFORCE 算法的Python代码如下：

import gymnasium as gym
import torch
import torch.nn.functional as F
import numpy as np
import matplotlib.pyplot as plt
import rl_utils
import random  # 用于设置全局随机种子
from tqdm import tqdm

# 策略网络
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)

# REINFORCE算法
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)
        self.gamma = gamma
        self.device = device

    def take_action(self, state):
        state = torch.tensor(np.array([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 = 0
        self.optimizer.zero_grad()
        # 从最后一步反向计算累计回报
        for i in reversed(range(len(reward_list))):
            reward = reward_list[i]
            state = torch.tensor(np.array([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
            loss = - log_prob * G  # 损失为负的对数概率乘累计回报（梯度上升等价于损失下降）
            loss.backward()  # 逐步反向传播计算梯度
        self.optimizer.step()  # 一次更新所有参数（同策略算法特性）

# 超参数设置
learning_rate = 1e-3
num_episodes = 1000  # 总训练回合数，原代码循环分10次，每次100回合，匹配总次数
hidden_dim = 128
gamma = 0.98
# 自动选择GPU/CPU
device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")

# 环境初始化
env_name = "CartPole-v1"
# 设置全局随机种子（保证实验可复现）
np.random.seed(0)
torch.manual_seed(0)
random.seed(0)
# 创建环境，设置环境种子
env = gym.make(env_name)
env.action_space.seed(0)
env.observation_space.seed(0)

# 获取环境维度
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
# 实例化REINFORCE智能体
agent = REINFORCE(state_dim, hidden_dim, action_dim, learning_rate, gamma, device)

# 训练循环
return_list = []
for i in range(10):
    # 总回合数1000，分10次迭代，每次100回合
    with tqdm(total=int(num_episodes / 10), desc='Iteration %d' % i) as pbar:
        for i_episode in range(int(num_episodes / 10)):
            episode_return = 0
            transition_dict = {'states': [], 'actions': [], 'next_states': [], 'rewards': [], 'dones': []}
            # 适配Gymnasium：reset返回(state, info)，取第一个元素
            state, _ = env.reset(seed=0)  # 重置时指定seed
            done = False
            while not done:
                action = agent.take_action(state)
                # 适配Gymnasium：step返回(next_state, reward, terminated, truncated, info)
                next_state, reward, terminated, truncated, _ = env.step(action)
                # Gymnasium将done拆分为terminated（环境终止）和truncated（步数截断），合并为done
                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)

# 绘制训练曲线
episode_list = list(range(len(return_list)))
plt.plot(episode_list, return_list)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('REINFORCE on {}'.format(env_name))
plt.savefig('REINFORCE on CartPole-v1_training.png')
plt.show()

# 绘制移动平均曲线，平滑显示训练趋势
mv_return = rl_utils.moving_average(return_list, 9)
plt.plot(episode_list, mv_return)
plt.xlabel('Episodes')
plt.ylabel('Moving Average Returns')
plt.title('REINFORCE on {}'.format(env_name))
plt.savefig('REINFORCE on CartPole-v1_moving_average.png')
plt.show()

七、程序运行结果展示

Iteration 0: 100%|██████████| 100/100 [00:01<00:00, 77.79it/s, episode=100, return=32.300]
Iteration 1: 100%|██████████| 100/100 [00:02<00:00, 42.19it/s, episode=200, return=65.400]
Iteration 2: 100%|██████████| 100/100 [00:06<00:00, 15.55it/s, episode=300, return=200.400]
Iteration 3: 100%|██████████| 100/100 [00:08<00:00, 11.99it/s, episode=400, return=213.200]
Iteration 4: 100%|██████████| 100/100 [00:06<00:00, 16.56it/s, episode=500, return=130.000]
Iteration 5: 100%|██████████| 100/100 [00:10<00:00,  9.25it/s, episode=600, return=343.400]
Iteration 6: 100%|██████████| 100/100 [00:14<00:00,  6.99it/s, episode=700, return=319.700]
Iteration 7: 100%|██████████| 100/100 [00:17<00:00,  5.80it/s, episode=800, return=224.100]
Iteration 8: 100%|██████████| 100/100 [00:08<00:00, 12.45it/s, episode=900, return=176.100]
Iteration 9: 100%|██████████| 100/100 [00:07<00:00, 14.16it/s, episode=1000, return=158.500]

八、REINFORCE 算法的优缺点（结合代码表现）

优点（代码完美体现）

1. 直接优化策略：无需学习值函数，直接参数化并优化策略，策略表达更灵活，既适合离散动作空间（如本代码），也可轻松适配连续动作空间（改策略网络输出和采样分布即可，如用正态分布采样连续动作）；
2. 实现简单：核心逻辑仅 “采样轨迹 - 计算 G_t - 更新策略” 三步，代码量少，无复杂的值函数拟合、目标网络等模块；
3. 无偏估计：蒙特卡洛方式用完整回合的回报计算Gt​，是策略梯度的无偏估计，训练收敛后策略的性能更稳定。

缺点（代码训练中会体现）

1. 高方差：蒙特卡洛的无偏估计伴随高方差，导致训练回报曲线波动剧烈，收敛速度慢（本代码中训练前期回报波动大，后期才逐渐收敛）；
2. 同策略效率低：必须用当前策略采样的轨迹更新，采样的轨迹用过一次就丢弃，不能复用（如 DQN 的经验回放），数据利用率极低；
3. 只能回合更新：必须等回合结束才能更新，无法进行步更 / 半回合更，对长回合任务（如 Atari 游戏）训练效率极低；
4. 无探索 - 利用平衡的显式控制：仅通过动作概率采样实现探索，无专门的探索机制（如 ε- 贪心），探索策略较粗糙。

九、经典改进方向（解决高方差 / 低效率问题）

针对 REINFORCE 的缺点，后续有很多经典改进版，都是工业界常用的策略梯度算法：

1. REINFORCE with Baseline：引入值函数作为基线（Baseline），将损失改为−logp​rob⋅(Gt​−V(st​))，大幅降低方差，不改变无偏性；
2. Actor-Critic（AC）：结合策略梯度（Actor）和值函数（Critic），用 TD 误差替代蒙特卡洛的Gt​，实现步更 / 半回合更，提升训练效率；
3. A2C/A3C：在 AC 基础上做并行训练，进一步提升训练速度；
4. PPO（Proximal Policy Optimization）：当前最主流的策略梯度算法，在同策略的基础上引入信任域，允许轨迹复用，兼顾效率和稳定性，是工业界的 “首选方案”。

十、总结

本文详细介绍了REINFORCE算法的原理与实现。REINFORCE是一种基于策略梯度的无模型强化学习算法，直接优化策略参数，适用于离散动作空间。文章从算法定位、数学推导、代码实现三个层面展开：首先阐述了其无模型、同策略、蒙特卡洛更新的核心特性；然后推导了策略梯度的数学公式及其工程化实现；最后通过完整的Python代码展示了在CartPole-v1环境中的应用，包括策略网络设计、轨迹采集、蒙特卡洛更新等关键模块。实验结果表明，REINFORCE算法能够有效解决简单控制问题，但存在高方差、低效率等缺点。文章还指出了改进方向，如引入基线函数、Actor-Critic架构等。整体上，REINFORCE作为策略梯度家族的基础算法，具有理论简洁、实现直观的特点。