![[CH09]策略梯度法](https://d-sketon.top/img/_backwebp/bg24.webp)
零、写在前面
之前我们一直先学习价值函数，再“借助”价值函数选择动作：Q-learning 和 DQN 估计每个动作的价值，行为策略通常再用 argmax 或 epsilon-greedy 导出。
其实呢，我们可以换一种路线：直接把策略写成带参数的概率分布
$$ \pi_\theta(a\mid s), $$让神经网络直接回答“在状态 $s$ 下，每个动作应该以多大概率被选择”。随后用梯度调整 $\theta$，使期望收益增大。这就是基于策略的方法，其中用梯度优化策略的方法叫作策略梯度法。
下面要介绍的四种方法看似逐节变化，实际上都可以写成同一个骨架：
$$ L_\pi(\theta) =- \sum_{t=0}^{T} \Phi_t\log\pi_\theta(A_t\mid S_t). \tag{L9.1} $$我们只是在不断改进“怎样给时刻 $t$ 的动作打分”这个权重 $\Phi_t$：
$$ G(\tau) \longrightarrow G_t \longrightarrow G_t-b(S_t) \longrightarrow R_{t+1}+\gamma V(S_{t+1})-V(S_t). $$请先抓住一句话：策略梯度不是把动作当监督标签，而是让高于基准的已采样动作以后更容易出现，让低于基准的已采样动作以后更不容易出现。
一、最简单的策略梯度法
1.1 策略梯度法的推导
随机策略写成 $\pi_\theta(a\mid s)$。考虑回合制任务，一条轨迹可写为
$$ \tau=(S_0,A_0,R_1,S_1,A_1,R_2,\ldots,S_{T+1}), $$其折扣收益为
$$ G(\tau)=R_1+\gamma R_2+\gamma^2R_3+\cdots+\gamma^TR_{T+1}. $$我们希望找到使期望收益最大的参数：
$$ J(\theta)=\mathbb E_{\tau\sim\pi_\theta}[G(\tau)]. $$“$\tau\sim\pi_\theta$”并不是说轨迹只由策略决定。轨迹同时受初始状态分布、环境转移概率和奖励规律影响；这里的意思是，我们能够直接控制的部分是策略。
若把一条轨迹的概率写开，忽略不影响本章结论的奖励记号，可得
$$ p_\theta(\tau) =p(S_0)\prod_{t=0}^{T} \pi_\theta(A_t\mid S_t) p(S_{t+1}\mid S_t,A_t). \tag{L9.2} $$目标梯度从定义出发：
$$ \nabla_\theta J(\theta) =\sum_\tau G(\tau)\nabla_\theta p_\theta(\tau). $$利用
$$ \begin{align} &\nabla_\theta p_\theta(\tau) \\ &= \frac{p_\theta(\tau)}{p_\theta(\tau)} \nabla_\theta p_\theta(\tau) \\ &=p_\theta(\tau)\nabla_\theta\log p_\theta(\tau), \end{align} $$也就是常说的 log-derivative trick，可将梯度改写为期望：
$$ \nabla_\theta J(\theta) =\mathbb E_{\tau\sim\pi_\theta} \left[G(\tau)\nabla_\theta\log p_\theta(\tau)\right]. $$再对式 (L9.2) 取对数，乘积变成求和。由于环境的 $p(S_{t+1}\mid S_t,A_t)$ 不含 $\theta$，对 $\theta$ 求导后消失，于是得式 (9.1)：
$$ \nabla_\theta J(\theta) =\mathbb E_{\tau\sim\pi_\theta} \left[ \sum_{t=0}^{T} G(\tau)\nabla_\theta \log\pi_\theta(A_t\mid S_t) \right]. \tag{9.1} $$这一步回答了两个关键问题：
- 为什么环境不需要可微？ 环境概率没有进入最终需要求导的项；
- 为什么是 log probability？ 对数把整条轨迹上的策略概率乘积变成每一步对数概率之和。
实际只有有限条采样轨迹，最简单的单轨迹 Monte Carlo 估计为式 (9.2)：
$$ \nabla_\theta J(\theta) \approx \sum_{t=0}^{T} G(\tau)\nabla_\theta \log\pi_\theta(A_t\mid S_t). \tag{9.2} $$$$ \theta \leftarrow \theta + \alpha \nabla_\theta J(\theta) $$理论式在做梯度上升；因此定义负损失方便做梯度下降：
$$ L_\pi =-G(\tau) \sum_{t=0}^{T}\log\pi_\theta(A_t\mid S_t). \tag{L9.3} $$1.2 策略梯度法的算法
最简单算法的一回合流程如下。
- 用当前 $\pi_\theta$ 与环境交互，生成完整轨迹；
- 每一步保存奖励和已采样动作的 log probability；
- 回合结束后计算一次整条轨迹收益 $G(\tau)$；
- 让轨迹中所有时间步共享同一个 $G(\tau)$；
- 构造式 (L9.3)，反向传播并更新策略；
- 清空本回合 memory，重新采样下一条 on-policy 轨迹。
这里“所有动作共享同一个总成绩”既是算法最简单的地方，也是它的主要缺点。一次轨迹中既可能有关键动作，也可能有无关甚至有害的动作；若最终成绩不错，式 (L9.3) 会同时提高它们的概率，信用分配非常粗糙。
1.3 策略梯度法的实现
1.3.1 策略网络
策略网络用于输出 策略 的概率。
class Policy(nn.Module):
def __init__(
self,
state_size: int = STATE_SIZE,
action_size: int = ACTION_SIZE,
) -> None:
super().__init__()
self.l1 = nn.Linear(state_size, 128)
self.l2 = nn.Linear(128, action_size)
def forward(self, states: Tensor) -> Tensor:
hidden = F.relu(self.l1(states))
logits = self.l2(hidden)
return F.softmax(logits, dim=-1)
1.3.2 Agent类
def compute_discounted_return(rewards: Sequence[float], gamma: float) -> float:
"""计算整条轨迹从起点开始的折扣收益 G(τ)。"""
episode_return = 0.0
for reward in reversed(rewards):
episode_return = float(reward) + gamma * episode_return
return episode_return
class Agent:
def __init__(
self,
state_size: int = STATE_SIZE,
action_size: int = ACTION_SIZE,
*,
gamma: float = 0.98,
lr: float = 0.0002,
device: str | torch.device = "cpu",
) -> None:
self.gamma = gamma
self.device = torch.device(device)
self.memory: list[tuple[float, Tensor]] = []
self.pi = Policy(state_size, action_size).to(self.device)
self.optimizer = Adam(self.pi.parameters(), lr=lr)
def get_action(self, state: np.ndarray) -> tuple[int, Tensor]:
"""按当前策略采样动作，并保留该动作 log π(a|s) 的计算图。"""
state_tensor = torch.as_tensor(
state, dtype=torch.float32, device=self.device
).unsqueeze(0)
probabilities = self.pi(state_tensor)[0]
distribution = Categorical(probs=probabilities)
action_tensor = distribution.sample()
log_prob = distribution.log_prob(action_tensor)
return int(action_tensor.item()), log_prob
def add(self, reward: float, log_prob: Tensor) -> None:
"""保存一步奖励和仍带计算图的动作对数概率。"""
self.memory.append((float(reward), log_prob))
def compute_loss(self) -> Tensor:
"""构造 -G(τ) * Σ log π(A_t|S_t)。"""
if not self.memory:
raise RuntimeError("轨迹为空，无法计算策略损失。")
rewards = [reward for reward, _ in self.memory]
log_probs = torch.stack([log_prob for _, log_prob in self.memory])
episode_return = compute_discounted_return(rewards, self.gamma)
# 所有时间步共享同一个 G(τ)，而不是各自使用 G_t。
return -episode_return * log_probs.sum()
def update(self) -> float:
"""在回合结束后更新一次策略，并清空本回合轨迹。"""
loss = self.compute_loss()
self.optimizer.zero_grad(set_to_none=True)
loss.backward()
self.optimizer.step()
self.memory.clear()
return float(loss.detach().item())
采样动作
动作不是 argmax 了，而是概率采样。
probabilities = self.pi(state_tensor)[0]
distribution = Categorical(probs=probabilities)
action_tensor = distribution.sample()
log_prob = distribution.log_prob(action_tensor)
反向递推整条轨迹收益
$$ G_t=R_{t+1}+\gamma G_{t+1}. $$def compute_discounted_return(rewards, gamma):
episode_return = 0.0
for reward in reversed(rewards):
episode_return = float(reward) + gamma * episode_return
return episode_return
为什么 memory 保存 log_prob，而不只保存 action
self.memory: list[tuple[float, Tensor]] = []
self.memory.append((float(reward), log_prob))
log_prob 仍连着当时策略前向传播的计算图。回合末将它们堆叠后，PyTorch 才能一次反向传播到每个状态下产生动作概率的网络参数：
rewards = [reward for reward, _ in self.memory]
log_probs = torch.stack([log_prob for _, log_prob in self.memory])
episode_return = compute_discounted_return(rewards, self.gamma)
loss = -episode_return * log_probs.sum()
因此这里不能提前对 log_prob 做 detach()。代价是必须在整个回合中保留计算图，内存开销随轨迹长度增长。只保存 action 也不是绝对不行，但回合末需要重新保存所有 state 并再做一次策略前向传播；我们实现选择直接保存计算图，代码更直观。
1.3.3 训练
def set_seed(seed: int) -> None:
random.seed(seed)
np.random.seed(seed)
torch.manual_seed(seed)
def train(
env: Any,
agent: Agent,
*,
episodes: int = 3000,
seed: int | None = None,
log_interval: int | None = None,
) -> list[float]:
"""运行完整的回合式训练循环并返回每回合总奖励。"""
if episodes < 1:
raise ValueError("episodes 必须大于 0。")
if seed is not None:
set_seed(seed)
reward_history: list[float] = []
for episode in range(episodes):
reset_seed = seed if episode == 0 else None
state, _ = env.reset()
done = False
total_reward = 0.0
while not done:
action, log_prob = agent.get_action(state)
next_state, reward, terminated, truncated, info = env.step(action)
done = terminated or truncated
agent.add(reward, log_prob)
state = next_state
total_reward += reward
agent.update()
reward_history.append(total_reward)
if log_interval and episode % log_interval == 0:
print(f"episode: {episode}, total reward: {total_reward:.1f}")
return reward_history
def plot_reward_history(reward_history: Sequence[float]) -> None:
import matplotlib.pyplot as plt
figure, axis = plt.subplots()
axis.set_xlabel("Episode")
axis.set_ylabel("Total Reward")
axis.plot(range(len(reward_history)), reward_history)
figure.tight_layout()
plt.show()
device = 'cuda' if torch.cuda.is_available() else 'cpu'
env = gym.make('CartPole-v1')
agent = Agent(device=device)
episodes = 3000
seed = 42
log_interval=100
set_seed(seed)
reward_his = train(env, agent, episodes=episodes, seed=seed, log_interval=log_interval)
plot_reward_history(reward_his)
曲线仍然是非常震荡
如果我们重复100次实验取平均：
虽然reward一直在涨，但明显还有改进的余地，我们接着改进算法，也就是著名的 REINFORCE 算法。
二、REINFORCE
最简单策略梯度法让时刻 $t$ 的动作承担当次轨迹的全部奖励，包括该动作发生以前已经得到的奖励。但行动 $A_t$ 不可能影响过去。因此，可以删除与 $A_t$ 因果无关的过去奖励，只保留从当前时刻开始的收益：
$$ G_t =R_{t+1}+\gamma R_{t+2}+\gamma^2R_{t+3}+\cdots. $$REINFORCE 这个名字是“REward Increment = Nonnegative Factor × Offset Reinforcement × Characteristic Eligibility”（奖励增量 = 非负因子 × 偏移强化 × 特征资格）的首字母缩写。
2.1 REINFORCE 算法
那么更新方向可以记为
$$ \nabla_\theta J(\theta) =\mathbb E_{\tau\sim\pi_\theta} \left[ \sum_{t=0}^{T} G_t\nabla_\theta \log\pi_\theta(A_t\mid S_t) \right]. \tag{9.3} $$对应的损失为
$$ \sum_{t=0}^{T} G_t\log\pi_\theta(A_t\mid S_t). \tag{L9.4} $$Q：为什么这个改进是合理的？
A：
问题其实就是 为什么过去奖励可以删除？
将总回报 $G_0$ 拆成“过去奖励 + 当前及未来奖励”：
$$ > G_0 = \underbrace{ \sum_{k=0}^{t-1}\gamma^kR_k }_{\text{第 }t\text{ 步之前的奖励}} + \underbrace{ \sum_{k=t}^{T}\gamma^kR_k }_{\text{第 }t\text{ 步及之后的奖励}}. > $$我们关注下式：
$$ > \mathbb E \left[ G_0 \nabla_\theta\log\pi_\theta(A_t\mid S_t) \right] > $$把 \(G_0\) 拆开：
$$ > \begin{aligned} & \mathbb E \left[ \left( \sum_{k=0}^{t-1}\gamma^kR_k \right) \nabla_\theta\log\pi_\theta(A_t\mid S_t) \right]\\ &+ \mathbb E \left[ \left( \sum_{k=t}^{T}\gamma^kR_k \right) \nabla_\theta\log\pi_\theta(A_t\mid S_t) \right]. \end{aligned} > $$关键在第一项。令过去奖励的总和为 $C_t$：
$$ > C_t=\sum_{k=0}^{t-1}\gamma^kR_k. > $$当时间来到 \(t\) 时，\(C_t\) 已经发生，是一个确定的历史结果；但当前动作 \(A_t\) 还没有采样。
因此，固定当前状态 \(S_t\) 和此前历史后，对动作 \(A_t\) 取期望：
$$ > \begin{aligned} \mathbb E_{A_t\sim\pi_\theta} \left[ \nabla_\theta\log\pi_\theta(A_t\mid S_t) \right] &= \sum_a \pi_\theta(a\mid S_t) \nabla_\theta\log\pi_\theta(a\mid S_t)\\ &= \sum_a \nabla_\theta\pi_\theta(a\mid S_t)\\ &= \nabla_\theta \sum_a\pi_\theta(a\mid S_t)\\ &= \nabla_\theta 1\\ &=0. \end{aligned} > $$所以：
$$ > \mathbb E \left[ C_t \nabla_\theta\log\pi_\theta(A_t\mid S_t) \right] =0. > $$这就是说：
对第 $t$ 步动作而言，过去奖励虽然已经发生，但它们与“这一步动作该如何增加概率”之间，在期望意义下没有可学习的因果关系。
2.2 REINFORCE 的实现
和与最简单实现只有一个关键差别：反向扫描奖励时，不只在最后留下一个 $G$，而是在每个时间步立刻用当前 $G_t$ 累加 loss。
def compute_reward_to_go(rewards: Sequence[float], gamma: float) -> list[float]:
"""反向递推并返回 [G_0, G_1, ..., G_T]。"""
returns = [0.0] * len(rewards)
reward_to_go = 0.0
for index in range(len(rewards) - 1, -1, -1):
reward_to_go = float(rewards[index]) + gamma * reward_to_go
returns[index] = reward_to_go
return returns
class Agent:
"""以每个时间步的 reward-to-go 更新策略的 REINFORCE 智能代理。"""
# ……
def compute_loss(self) -> Tensor:
"""构造 -Σ G_t log π(A_t|S_t)。"""
if not self.memory:
raise RuntimeError("轨迹为空，无法计算 REINFORCE 损失。")
rewards = [reward for reward, _ in self.memory]
log_probs = torch.stack([log_prob for _, log_prob in self.memory])
returns = torch.as_tensor(
compute_reward_to_go(rewards, self.gamma),
dtype=log_probs.dtype,
device=log_probs.device,
)
# REINFORCE：每个 log_prob 都乘自己的 G_t
return -(returns * log_probs).sum()
# ……
再训练一次：
这次可以把任务上限 500 直接拉满了。
由于随着表现越来越好，一个epoch越来越长，所以就不做重复实验求平均了。
三、基线
REINFORCE 的一次梯度估计只来自有限轨迹。相同状态和动作之后，环境随机性和后续随机动作可能造成截然不同的 $G_t$，所以梯度方向会强烈抖动。baseline 的目标不是改变“平均而言该往哪里更新”，而是降低这种方差。
3.1 基线的思路
我们可以用考试成绩作类比，来说明“看相对偏差比看绝对分数更稳定”。某次三人成绩为
$$ [90,40,50], $$其方差约为 \(466.67\)。若根据各自历史表现预测为
$$ [82,46,49], $$则实际成绩减去预测基线为
$$ [8,-6,1], $$差值方差约为 \(32.67\)。绝对成绩跨度很大，而“比自己通常表现好多少”更集中。
在策略梯度中也可以采用同一思路：不再只看 return 的绝对大小，而是看它相对某个参考水平高了多少或低了多少。把这个参考水平写成 baseline $b(S_t)$，Actor 的评分便从变为 $G_t-b(S_t).$
baseline 的任务是降低梯度估计的方差，而不是改变平均更新方向。
3.2 带基线的策略梯度法
带 baseline 的权重是
$$ G_t-b(S_t), $$因此式 (9.4) 可写为
$$ \nabla_\theta J(\theta) =\mathbb E \left[ \sum_t \gamma^t\bigl(G_t-b(S_t)\bigr) \nabla_\theta\log\pi_\theta(A_t\mid S_t) \right]. \tag{L9.5a} $$Q：baseline 为什么不改变期望更新方向
A：
固定状态 $s$ 时，若 $b(s)$ 不依赖当前动作，则
$$ > \begin{aligned} > &\mathbb E_{A\sim\pi_\theta(\cdot\mid s)} > \left[b(s)\nabla_\theta\log\pi_\theta(A\mid s)\right]\\ > &=b(s)\sum_a\pi_\theta(a\mid s) > \nabla_\theta\log\pi_\theta(a\mid s)\\ > &=b(s)\sum_a\nabla_\theta\pi_\theta(a\mid s)\\ > &=b(s)\nabla_\theta\sum_a\pi_\theta(a\mid s) > =b(s)\nabla_\theta 1=0. > \end{aligned} > \tag{L9.5} > $$所以从原更新中减去 baseline 项后，其条件期望仍为 0；所以这个结论都成立。
若有一个单独的价值网络给出 baseline，可用完整 Monte Carlo return 训练它：
$$ > L_V(\omega)= > \sum_t\left[V_\omega(S_t)-G_t\right]^2. > $$
四、Actor-Critic
带 baseline 的 REINFORCE 仍要等到回合结束，才能得到完整 $G_t$。
它仍然是基于策略的。，
如果希望每经历一步就更新，可以用价值函数估计尚未发生的未来收益。这就从纯 Monte Carlo 方法走向了 bootstrap。
也就是说，我们将使用价值函数作为baseline，那么就可以将其视为基于价值且基于策略的方法。
4.1 Actor-Critic 的推导
最自然的状态 baseline 是当前策略的状态价值：
$$ b(S_t)=V_\omega(S_t)\approx \mathbb E_\pi[G_t\mid S_t]. $$把它代入式 (9.4)，得到式 (9.5)：
$$ \nabla_\theta J(\theta) =\mathbb E \left[ \sum_t \bigl(G_t-V_\omega(S_t)\bigr) \nabla_\theta\log\pi_\theta(A_t\mid S_t) \right]. \tag{9.5} $$此时 $G_t-V_\omega(S_t)$ 是一次采样得到的 advantage estimator：
- 大于 0：这次动作后的结果比该状态的通常水平好，提高动作概率；
- 小于 0：结果比通常水平差，降低动作概率；
- 接近 0：结果符合预期，策略更新较小。
这里的“提高／降低”是针对单个 loss 项和小步梯度下降的局部直觉。真实网络会把多个时间步、多个状态的梯度汇总到共享参数中，Adam 还带有动量，因此一次完整的 optimizer.step() 不保证每个已采样动作的概率都逐项单调变化。
接下来从式 (9.5) 的权重开始：
$$ G_t-V_\omega(S_t). $$利用收益递推关系
$$ G_t=R_{t+1}+\gamma G_{t+1}, $$并用下一状态的价值估计替代尚未观测完整的 $G_{t+1}$：
$$ G_{t+1}\approx V_\omega(S_{t+1}), $$便得到一步 TD error：
$$ \delta_t =R_{t+1}+\gamma V_\omega(S_{t+1})-V_\omega(S_t). \tag{L9.6} $$如果 Critic 恰好等于当前策略的真实状态价值，即 $V_\omega=V^\pi$，那么对固定的状态动作对有
$$ \begin{aligned} \mathbb E[\delta_t\mid S_t=s,A_t=a] &=\mathbb E[R_{t+1}+\gamma V^\pi(S_{t+1})\mid s,a]-V^\pi(s)\\ &=Q^\pi(s,a)-V^\pi(s)\\ &=A^\pi(s,a). \end{aligned} \tag{L9.6a} $$这说明一步 TD error 在理想 Critic 下是 advantage 的条件无偏估计。我们把它作为策略梯度权重，并以等号写出式 (9.6)。但右侧应该理解为“理想条件下的关系及其实际样本估计”，而不是任意近似 Critic 下都精确成立的恒等式：
$$ \nabla_\theta J(\theta) =\mathbb E \left[ \sum_t \delta_t \nabla_\theta\log\pi_\theta(A_t\mid S_t) \right]. \tag{9.6} $$不过，实际训练中的 $V_\omega$ 只是不断学习的近似 Critic，因此不要把上式理解为无条件成立的精确 policy gradient。算法真正用到的是样本更新方向
$$ \widehat g_{\mathrm{AC}} =\sum_t \delta_t\nabla_\theta \log\pi_\theta(A_t\mid S_t), \tag{L9.6b} $$它同时带有采样误差、Critic 函数逼近误差和 bootstrap 误差。
这两个模型各司其职：
- Actor：策略 $\pi_\theta(a\mid s)$，决定怎样行动；
- Critic：价值 $V_\omega(s)$，评价当前状态以及这次动作后的结果；
- TD error：Critic 给 Actor 的即时评价信号。
flowchart LR
S["状态 St"] --> ACT["Actor πθ"]
ACT -->|"sample"| A["动作 At"]
A --> ENV["环境"]
ENV --> R["奖励 Rt+1"]
ENV --> NS["下一状态 St+1"]
S --> VS["Critic Vω(St)"]
NS --> VN["Critic Vω(St+1)"]
R --> D["δt = R + γV(next) - V(now)"]
VS --> D
VN --> D
D -->|"固定权重"| ACT
D -->|"平方误差"| CR["更新 Critic"]
与 REINFORCE 相比，这里发生了重要的 bias-variance 交换：
| 方法 | Actor 权重 | 何时更新 | 主要特点 |
|---|---|---|---|
| 最简单策略梯度 | $G(\tau)$ | 回合末 | 最粗糙、方差大 |
| REINFORCE | $G_t$ | 回合末 | 因果更合理，仍是 MC |
| 带 baseline REINFORCE | $G_t-V(S_t)$ | 回合末 | 降低方差，仍需完整 return |
| Actor-Critic | $R_{t+1}+\gamma V(S_{t+1})-V(S_t)$ | 每一步 | 方差通常更低、可在线更新，但可能引入 Critic 与 bootstrap 偏差 |
4.2 Actor-Critic 的实现
Actor 就是之前的 Policy
class PolicyNet(nn.Module):
"""Actor：输出离散动作概率的 4 -> 128 -> 2 网络。"""
def __init__(
self,
state_size: int = STATE_SIZE,
action_size: int = ACTION_SIZE,
) -> None:
super().__init__()
self.l1 = nn.Linear(state_size, 128)
self.l2 = nn.Linear(128, action_size)
def forward(self, states: Tensor) -> Tensor:
hidden = F.relu(self.l1(states))
logits = self.l2(hidden)
return F.softmax(logits, dim=-1)
Critic
注意 Critic 输出的是价值
class ValueNet(nn.Module):
"""Critic：为每个状态输出一个标量价值。"""
def __init__(self, state_size: int = STATE_SIZE) -> None:
super().__init__()
self.l1 = nn.Linear(state_size, 128)
self.l2 = nn.Linear(128, 1)
def forward(self, states: Tensor) -> Tensor:
hidden = F.relu(self.l1(states))
return self.l2(hidden)
然后写一个 Agent 类
class ActorCriticAgent:
"""使用一步 TD error 连接 Actor 与 Critic 的智能代理。"""
def __init__(
self,
state_size: int = STATE_SIZE,
action_size: int = ACTION_SIZE,
*,
gamma: float = 0.98,
lr_pi: float = 0.0002,
lr_v: float = 0.0005,
device: str | torch.device = "cpu",
) -> None:
self.gamma = gamma
self.device = torch.device(device)
self.pi = PolicyNet(state_size, action_size).to(self.device)
self.v = ValueNet(state_size).to(self.device)
self.optimizer_pi = Adam(self.pi.parameters(), lr=lr_pi)
self.optimizer_v = Adam(self.v.parameters(), lr=lr_v)
def get_action(self, state: np.ndarray) -> tuple[int, Tensor]:
"""按 Actor 的概率分布采样动作并返回动作的 log_prob。"""
state_tensor = torch.as_tensor(
state, dtype=torch.float32, device=self.device
).unsqueeze(0)
probabilities = self.pi(state_tensor)[0]
distribution = Categorical(probs=probabilities)
action_tensor = distribution.sample()
log_prob = distribution.log_prob(action_tensor)
return int(action_tensor.item()), log_prob
def compute_td_target(
self,
reward: float,
next_state: np.ndarray,
bootstrap_stop: bool,
) -> Tensor:
"""计算不参与反向传播的一步 TD target。"""
reward_tensor = torch.tensor(
[[float(reward)]], dtype=torch.float32, device=self.device
)
if bootstrap_stop:
# 终止状态没有未来价值，也不应读取可能无意义的 next_state。
return reward_tensor
next_state_tensor = torch.as_tensor(
next_state, dtype=torch.float32, device=self.device
).unsqueeze(0)
with torch.no_grad():
next_value = self.v(next_state_tensor)
target = reward_tensor + self.gamma * next_value
return target
def compute_losses(
self,
state: np.ndarray,
log_prob: Tensor,
reward: float,
next_state: np.ndarray,
bootstrap_stop: bool,
) -> tuple[Tensor, Tensor, Tensor]:
"""分别构造 Actor loss、Critic loss 和 TD target。"""
state_tensor = torch.as_tensor(
state, dtype=torch.float32, device=self.device
).unsqueeze(0)
target = self.compute_td_target(reward, next_state, bootstrap_stop)
value = self.v(state_tensor)
td_error = target - value
value_loss = F.mse_loss(value, target)
# delta 只负责给 Actor 的动作打分，Actor loss 不能反向修改 Critic。
policy_loss = -log_prob * td_error.detach().squeeze()
return policy_loss, value_loss, target
def update(
self,
state: np.ndarray,
log_prob: Tensor,
reward: float,
next_state: np.ndarray,
bootstrap_stop: bool,
) -> tuple[float, float]:
"""用一次转移同时更新 Actor 和 Critic。"""
policy_loss, value_loss, _ = self.compute_losses(
state, log_prob, reward, next_state, bootstrap_stop
)
self.optimizer_pi.zero_grad(set_to_none=True)
self.optimizer_v.zero_grad(set_to_none=True)
value_loss.backward()
policy_loss.backward()
self.optimizer_v.step()
self.optimizer_pi.step()
return (
float(policy_loss.detach().item()),
float(value_loss.detach().item()),
)
def train(
env: Any,
agent: ActorCriticAgent,
*,
episodes: int = 3000,
seed: int | None = None,
log_interval: int | None = None,
) -> list[float]:
"""运行逐时间步更新的 Actor-Critic 训练循环。"""
if episodes < 1:
raise ValueError("episodes 必须大于 0。")
if seed is not None:
set_seed(seed)
reward_history: list[float] = []
for episode in range(episodes):
reset_seed = seed if episode == 0 else None
state, _ = env.reset()
done = False
total_reward = 0.0
while not done:
action, log_prob = agent.get_action(state)
next_state, reward, terminated, truncated, info = env.step(action)
done = terminated or truncated
agent.update(
state,
log_prob,
reward,
next_state,
done,
)
state = next_state
total_reward += reward
reward_history.append(total_reward)
if log_interval and episode % log_interval == 0:
print(f"episode: {episode}, total reward: {total_reward:.1f}")
return reward_history
def plot_reward_history(reward_history: Sequence[float]) -> None:
import matplotlib.pyplot as plt
figure, axis = plt.subplots()
axis.set_xlabel("Episode")
axis.set_ylabel("Total Reward")
axis.plot(range(len(reward_history)), reward_history)
figure.tight_layout()
plt.show()
device = 'cuda' if torch.cuda.is_available() else 'cpu'
env = gym.make('CartPole-v1')
agent = ActorCriticAgent(device=device)
episodes = 3000
seed = 42
log_interval=100
set_seed(seed)
reward_his = train(env, agent, episodes=episodes, seed=seed, log_interval=log_interval)
plot_reward_history(reward_his)
仍然跑一次训练：
五、基于策略的方法的优点
1. 直接表示策略
价值方法先学习 $Q$ 或 $V$，再从价值导出策略；
策略方法直接优化 $\pi_\theta$。当目标本来就是一个复杂的随机决策规则时，这条路径更直接，也能自然表示同一状态下多个合理动作的混合策略。
2. 容易扩展到连续动作
连续动作空间中，无法简单枚举所有动作再做 argmax。策略网络可以输出一个参数化分布。例如对 Pendulum 的连续扭矩，可让网络输出正态分布参数：
$$ (\mu_\theta(s),\sigma_\theta(s)) \longrightarrow a\sim\mathcal N(\mu_\theta(s),\sigma_\theta^2(s)). $$离散 CartPole 使用 Categorical；连续控制可以使用 Normal。实际代码还要保证标准差为正，常见做法是输出 log_std 或经 softplus 变换。本章只介绍概念，没有提供 Pendulum 的连续策略实现。
3. 行动选择概率平滑变化
epsilon-greedy 的主要行动由 argmax 决定。两个 Q 值一旦交换大小，贪婪动作可能突然跳变。Softmax 策略直接优化概率，参数小幅变化通常带来概率的连续变化，因此行为可以更平滑。
不过“更平滑”不等于“总是更稳定或更优秀”。策略梯度仍可能有高方差、陷入局部解、过早降低探索，且对学习率和网络结构敏感。价值方法与策略方法各有适用场景，后续算法也常把两者结合。
说些什么吧！