Off-Policy Actor-Critic 与重要性采样目录Importance samplingThe off policy gradient theoremAlgorithm descriptionpython 例子一 Importance sampling在 off-policy Actor-Critic 方法中我们通常用一个行为策略behavior policy对应分布来生成数据去优化另一个目标策略target policy对应分布。这时期望的估计就需要用到重要性采样。考虑一个简单的随机变量。假设在概率分布下则另一个分布下则如果我们只有来自的样本却想估计可以通过重要性采样即这里的就是重要性权重。- 如果权重为 1无需调整。- 如果说明在目标分布中该样本更重要应升高权重。- 如果则降低权重。在 off-policy Actor-Critic 中策略梯度会乘以类似的重要性权重以校正行为策略与目标策略之间的分布差异。二 The off -policy policy gradient theorem借助重要性采样技术我们现在可以给出 off-policy 策略梯度定理。假设 β 是一个行为策略behavior policy。我们的目标是利用 β生成的样本来学习一个目标策略 π使其最大化以下指标其中 dβdβ 是策略 ββ 下的平稳分布vπvπ 是策略 ππ 下的状态价值函数。该指标的梯度由以下定理给出。定理 10.1Off-policy 策略梯度定理在折扣因子 γ∈(0,1) 的情况下J(θ) 的梯度为其中状态分布 ρ定义为这里表示从状态 s′ 出发在策略 π下经过折扣后转移到状态 s 的总概率。(10.11) 中的梯度与同策略情形下的定理 9.1 类似但存在两点区别。第一点区别在于引入了重要性权重。第二点区别在于动作的采样分布是 A∼β而不是 A∼π。因此我们可以通过跟随行为策略 ββ 生成的动作样本来近似真实的梯度。该定理的证明见专栏 10.2。三 Algorithm description基于 off-policy 策略梯度定理我们现在可以给出 off-policy actor-critic 算法。由于 off-policy 情形与 on-policy 情形非常相似我们仅介绍其中的关键步骤。首先off-policy 策略梯度对任何额外的基线 (baseline) (具有不变性。具体而言我们有这是因为为了减小估计方差我们可以选择基线为从而得到对应的随机梯度上升算法为其中与 on-policy 情形类似优势函数 \( q_t(s, a) - v_t(s) \) 可以用 TD 误差替代即于是算法变为off-policy actor-critic 算法的实现总结在算法 10.3 中。可以看出该算法与优势 actor-critic 算法基本相同唯一的区别在于在 critic 和 actor 中都额外包含了一个重要性权重。需要注意的是除了 actor 之外critic 也通过重要性采样技术从 on-policy 转换为 off-policy。事实上重要性采样是一种通用技术可以应用于基于策略和基于价值的算法。最后算法 10.3 可以通过多种方式进行扩展以融入更多技术例如资格迹 (eligibility traces) [73]。四 代码例子上面算法10.3是教科书中的经典但不实用实际上很难训练主要原有重要性权重方差大易爆炸双网络耦合更新不稳定行为策略需全覆盖目标策略条件苛刻# -*- coding: utf-8 -*- 文件名: off_policy_ac.py Off-policy Actor-Critic with Importance Sampling (Algorithm 10.3) 行为策略: ε-greedy (if-else 实现) 环境: CartPole-v1 作者: chengxf2 日期: 2026-06-08 import torch import torch.nn as nn import torch.nn.functional as F import torch.optim as optim import gymnasium as gym import numpy as np import matplotlib.pyplot as plt # # 1. 网络结构Actor 与 Critic # class Actor(nn.Module): 策略网络 π(a|s,θ) def __init__(self, state_dim: int, action_dim: int): super(Actor, self).__init__() self.net nn.Sequential( nn.Linear(state_dim, 128), nn.ReLU(), nn.Linear(128, 128), nn.ReLU(), nn.Linear(128, action_dim) ) self._init_weights() def _init_weights(self): for m in self.modules(): if isinstance(m, nn.Linear): nn.init.xavier_uniform_(m.weight) nn.init.constant_(m.bias, 0.0) def forward(self, state: torch.Tensor): logits self.net(state) prob F.softmax(logits, dim-1) return prob class Critic(nn.Module): 价值网络 v(s,w) def __init__(self, state_dim: int): super(Critic, self).__init__() self.net nn.Sequential( nn.Linear(state_dim, 128), nn.ReLU(), nn.Linear(128, 128), nn.ReLU(), nn.Linear(128, 1) ) self._init_weights() def _init_weights(self): for m in self.modules(): if isinstance(m, nn.Linear): nn.init.xavier_uniform_(m.weight) nn.init.constant_(m.bias, 0.0) def forward(self, state: torch.Tensor): value self.net(state) return value # # 2. 智能体 (Agent) # class Agent: def __init__(self, state_dim: int, action_dim: int, gamma: float 0.98): self.actor Actor(state_dim, action_dim) self.critic Critic(state_dim) self.actor_optim optim.Adam(self.actor.parameters(), lr3e-4) self.critic_optim optim.Adam(self.critic.parameters(), lr5e-2) self.gamma gamma self.action_dim action_dim def select_action_via_behavior_policy(self, state, epsilon: float 0.2): ε-greedy 行为策略返回 (动作, 该动作在β下的概率) state_ts torch.FloatTensor(state).unsqueeze(0) with torch.no_grad(): pi_probs self.actor(state_ts).squeeze(0) greedy_action torch.argmax(pi_probs).item() if np.random.rand() epsilon: action np.random.randint(self.action_dim) else: action greedy_action # 计算 β(a) if action greedy_action: beta_prob 1.0 - epsilon epsilon / self.action_dim else: beta_prob epsilon / self.action_dim return action, beta_prob def learn(self, state, action, reward, next_state, done, beta_prob): 单步更新 (Algorithm 10.3) 返回: (is_ratio, td_error) 用于监控 s torch.FloatTensor(state).unsqueeze(0) next_s torch.FloatTensor(next_state).unsqueeze(0) r torch.FloatTensor([reward]) # ----- TD 目标与误差 ----- v_curr self.critic(s) with torch.no_grad(): v_next self.critic(next_s) if not done else torch.tensor([[reward]]) td_target r self.gamma * v_next td_error td_target.detach() - v_curr # δ_t #td_error torch.clamp(td_error, -10.0, 10.0) # ----- 重要性权重 ρ π(a|s) / β(a|s) ----- pi_probs self.actor(s).squeeze(0) pi_prob pi_probs[action] #print(pi_prob, beta_prob) is_ratio (pi_prob / (beta_prob 1e-6)).detach() is_ratio torch.clamp(is_ratio, 0.1, 10.0) # 截断防爆炸 # ----- 1. 更新 Critic (梯度上升) ----- critic_loss 0.5 * is_ratio * td_error.pow(2) self.critic_optim.zero_grad() critic_loss.backward() #torch.nn.utils.clip_grad_norm_(self.critic.parameters(), max_norm1.0) self.critic_optim.step() # ----- 2. 更新 Actor (梯度上升) ----- log_prob torch.log(pi_prob 1e-6) actor_loss -is_ratio * td_error.detach() * log_prob self.actor_optim.zero_grad() #torch.nn.utils.clip_grad_norm_(self.actor.parameters(), max_norm1.0) actor_loss.backward() self.actor_optim.step() #print(pi_probs) # 返回标量用于打印直接取出数值 return is_ratio.item(), td_error.item() # # 3. 训练主循环 # def train(episodes: int 500, render: bool False): env gym.make(CartPole-v1, render_modehuman if render else None) state_dim env.observation_space.shape[0] action_dim env.action_space.n agent Agent(state_dim, action_dim) episode_rewards [] for ep in range(1, episodes 1): state, _ env.reset() episode_reward 0 done False while not done: action, beta_prob agent.select_action_via_behavior_policy(state, epsilon0.2) next_state, reward, terminated, truncated, _ env.step(action) done terminated or truncated # 可选的奖励塑造加速学习注释掉即为标准环境 is_ratio, td_error agent.learn(state, action, reward, next_state, done, beta_prob) state next_state episode_reward 1 episode_rewards.append(episode_reward) if ep % 10 0: avg_reward np.mean(episode_rewards[-10:]) # 注意这里 is_ratio 和 td_error 是最后一次 step 的值只用于观察 print(fEpisode {ep:3d} | Avg Reward (last 10): {avg_reward:.1f} f| is_ratio {is_ratio:.2f}, td_error {td_error:.2f} ) env.close() print(Training finished.) # 绘制奖励曲线 plt.figure(figsize(10, 5)) plt.plot(episode_rewards, alpha0.6, labelEpisode Reward) if len(episode_rewards) 10: moving_avg np.convolve(episode_rewards, np.ones(10)/10, modevalid) plt.plot(range(9, len(episode_rewards)), moving_avg, r, linewidth2, labelMoving Avg (10)) plt.xlabel(Episode) plt.ylabel(Total Reward) plt.title(Off-policy Actor-Critic (ε-greedy behavior) on CartPole-v1) plt.legend() plt.grid(True) plt.savefig(training_curve.png, dpi150) plt.show() if __name__ __main__: train(episodes300, renderFalse)