目录
PythonAI算法
策略梯度算法 (REINFORCE) 详解
强化学习中基于价值与基于策略方法的区别，重点推导了策略梯度的数学基础。详细讲解了 REINFORCE 算法的流程、代码实现及优缺点。通过 CartPole 环境的实验验证了算法效果，展示了如何使用 PyTorch 构建策略网络并进行参数更新。
强化学习中基于价值与基于策略方法的区别，重点推导了策略梯度的数学基础。详细讲解了 REINFORCE 算法的流程、代码实现及优缺点。通过 CartPole 环境的实验验证了算法效果，展示了如何使用 PyTorch 构建策略网络并进行参数更新。
目录
想象你在玩一款闯关游戏，每一步选择（比如往左走、往右走、打怪）都会影响你最终能不能通关、能拿多少分。 基于价值的方法会先给每个'局面'打分 —— 这个分数代表'在这个局面下，只要好好玩，最后能得到的好处有多大'。 然后，它的决策逻辑很简单：在当前局面下，选那些能让你进入'更高分局面'的行动。
比如：现在你面前有两条路，左边的路对应的'未来好处分'是 80 分，右边是 60 分。那它就会选左边的路。
核心：先判断'每个选择的最终价值'，再跟着高价值走。
基于价值的方法的核心是学习'价值函数'，通过价值函数间接指导动作选择。
价值函数（Value Function）是强化学习中的核心概念，它量化了'在某个状态下采取动作（或遵循策略）能获得的长期累积奖励的期望'。基于价值的方法通过学习价值函数，最终根据价值函数的输出选择'最优动作'（即能带来最高价值的动作）。
动作价值函数
：在状态 s 下采取动作 a，之后遵循策略 π 能获得的累积奖励的期望（更常用，因为直接关联'状态 - 动作对'）。
状态价值函数
：在状态 s 下，遵循策略 π 能获得的累积奖励的期望。
基于价值的方法通常聚焦于学习 Q 函数（动作价值函数），因为它能直接告诉我们'在某个状态下做什么动作更好'。
当学习到最优的 Q 函数（记为 Q^(s,a)）后，在任意状态 s 下，只需选择使 Q^(s,a) 最大的动作 a 即可，即：
a^* = \arg\max_a Q^*(s,a)
Deep Q-Networks（DQN）：用深度神经网络近似 Q(s,a)，解决高维状态空间（如 Atari 游戏画面）的问题。
SARSA：学习当前策略下的动作价值函数 Q^π(s,a)，属于'同策略'（On-policy）方法。
Q-learning：直接学习最优动作价值函数 Q^*(s,a)，不依赖当前策略，属于'异策略'（Off-policy）方法。
还是玩闯关游戏，但这种方法不先算'每个局面的价值'，而是直接学一套'行动规则'。 比如通过大量练习，它总结出：'看到 A 怪物就后退，看到 B 道具就捡，在 C 路口就往右拐'—— 这套规则就叫'策略'。 哪怕有时候按这套规则做，短期看可能吃亏（比如绕了点路），但它还是会坚持按规则来，因为这套规则是长期试错后'大概率能赢'的总结。
核心：直接记住'在什么情况下该做什么'，不纠结'这么做能有多少好处'。
基于策略的方法的核心是直接学习'策略函数'，通过策略函数直接输出动作。
策略函数（Policy Function）是从'状态'到'动作'的映射，分为两种类型：
随机性策略：
π(a|s)（给定状态 s，输出动作 a 的概率分布）。
确定性策略：
a = π(s)（给定状态 s，直接输出唯一动作 a）；
基于策略的方法通过优化策略函数的参数，使策略对应的累积奖励期望最大化，最终得到最优策略 π^*。
对于随机性策略：从概率分布 π(a|s) 中采样动作（兼顾探索与利用）。
对于确定性策略：直接输出动作 a = π(s)；
| 维度 | 基于价值的方法 | 基于策略的方法 |
|---|---|---|
| 学习对象 | 动作价值函数 Q(s,a) | 策略函数 π(a |
| 动作选择依据 | 选择 Q(s,a) 最大的动作 | 策略直接输出动作（或采样动作） |
| 连续动作空间适配性 | 较差（需额外优化搜索最大 Q 的动作） | 较好（随机性策略可直接输出概率分布） |
| 探索方式 | 依赖 epsilon-贪心（如 DQN） | 随机性策略天然支持探索（从分布中采样） |
| 典型应用场景 | 离散动作（如 Atari 游戏、围棋） | 连续动作（如机器人控制、自动驾驶） |
在强化学习中，智能体的目标是学习一个'策略'（Policy），即从'状态'到'动作'的映射，使长期累积奖励最大化。策略梯度算法的核心逻辑可概括为：通过计算'累积奖励对策略参数的梯度'，用梯度上升法不断调整参数，最终得到最优策略。
策略是强化学习的核心概念，定义为智能体在给定状态下选择动作的规则，分为两种形式：
随机性策略：
π_θ(a|s)（给定状态 s，输出动作 a 的概率分布，策略参数 θ 决定分布形状）。
确定性策略：
a = π_θ(s)（给定状态 s，直接输出唯一动作 a，θ 为策略参数）。
策略梯度算法通常聚焦于随机性策略，因为随机性策略天然支持'探索'（从概率分布中采样动作），更适合强化学习的试错过程。
| 维度 | 基于价值的方法（如 DQN） | 策略梯度算法（基于策略） |
|---|---|---|
| 优化目标 | 学习价值函数 Q^*(s,a)，间接选最优动作 | 直接优化策略 π_θ，输出动作分布 |
| 动作选择 | 选价值最大的动作（\arg\max_a Q(s,a)） | 从策略分布中采样动作（兼顾探索与利用） |
| 连续动作适配性 | 差（需额外优化搜索） | 好（直接输出概率分布） |
| 训练稳定性 | 较稳定（价值函数平滑） | 波动大（依赖轨迹采样，方差高） |
策略梯度算法的核心是计算累积奖励期望对策略参数 θ 的梯度，并通过梯度上升最大化这个期望。
设智能体与环境交互产生的轨迹为 τ = (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}_{τ ~ π_θ} 表示对所有可能轨迹的期望（轨迹由策略 π_θ 生成）。
目标是求 ∇_θ 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]
策略梯度的直观含义是：对每个状态 - 动作对 (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 条轨迹。
REINFORCE 是最基础的策略梯度算法，用单条轨迹的累积奖励直接估计梯度，步骤如下：
采样轨迹：用当前策略 π_θ 与环境交互，收集一条完整轨迹 τ = (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（负号将梯度上升转为梯度下降优化）。
更新参数：对所有时刻的损失求和，反向传播更新策略参数 θ。
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() # 梯度上升更新参数
"""
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()
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online