优先经验回放（Prioritized Replay）

一句话概述

优先经验回放（Prioritized Experience Replay，PER）是DQN的又一重要改进——它不再随机均匀地从经验池中采样，而是根据每条经验的"重要性"（TD误差的大小）来加权采样。TD误差大的经验（表示当前估计与实际差距大）被更频繁地回放，从而加速学习。PER通过重要性采样（Importance Sampling）修正优先采样引入的分布偏差。

💡 核心要点：①PER用TD误差的绝对值|δ|来衡量经验的重要性，|δ|越大越优先被采样；②采样概率P(i) ∝ |δ_i|^α，α=0为均匀采样，α=1为完全优先；③重要性采样权重w_i = (N·P(i))^{-β}，修正优先采样引入的偏差；④β从初值（如0.4）逐渐增加到1，平衡偏差修正和收敛速度。

教学与演示

一、为什么需要优先经验回放——均匀采样的浪费

是什么（定义）：均匀经验回放将所有经验平等对待，随机采样。但并非所有经验都同等重要——有些经验包含"惊喜"（出乎意料的高奖励或低奖励），学这些经验比学"平淡无奇"的经验更有价值。优先经验回放让TD误差大的经验被更频繁地学习。

大白话 就像复习考试——你不会把所有题目平均分配时间，而是会多花时间在"做错的题"和"重点题"上。均匀经验回放就是一视同仁，每道题花一样的时间，太浪费了。优先经验回放就像"错题本"——多复习那些"让你意外的"、"估计误差大的"经验。

为什么（原理）：TD误差|δ| = |r + γ max Q(s',a') - Q(s,a)|衡量了当前Q值估计与TD目标之间的差距。|δ|大的经验意味着当前预测与实际有很大偏差，学到这些经验能让Q值快速逼近正确值。从信息论角度看，|δ|大的经验包含更多"信息量"（surprise），应优先学习。

怎么做（实现）：

import numpy as np

# ==================== 均匀采样 vs 优先采样 ====================
# 模拟10条经验的TD误差
np.random.seed(42)
td_errors = np.array([0.01, 0.02, 5.0, 0.03, 0.01, 3.0, 0.02, 10.0, 0.01, 1.0])

print("=== 均匀采样 vs 优先采样 ===")
print(f"10条经验的TD误差: {[f'{e:.2f}' for e in td_errors]}")
print()

# 均匀采样：每条经验概率=1/10
uniform_prob = np.ones(10) / 10
print(f"均匀采样概率: {[f'{p:.2f}' for p in uniform_prob]}")
print(f"  经验3(TD=5.0)被采样概率: {uniform_prob[2]:.2f}")
print(f"  经验8(TD=10.0)被采样概率: {uniform_prob[7]:.2f}")

# 优先采样：概率正比于TD误差
alpha = 1.0  # 优先程度
priority = np.abs(td_errors) ** alpha
prioritized_prob = priority / priority.sum()
print(f"\n优先采样概率(α={alpha}): {[f'{p:.2f}' for p in prioritized_prob]}")
print(f"  经验3(TD=5.0)被采样概率: {prioritized_prob[2]:.2f}")
print(f"  经验8(TD=10.0)被采样概率: {prioritized_prob[7]:.2f}")
print(f"  经验1(TD=0.01)被采样概率: {prioritized_prob[0]:.4f}")

# 模拟采样
print(f"\n100次采样中每条经验被选中的次数：")
uniform_samples = np.random.choice(10, 100, p=uniform_prob)
prioritized_samples = np.random.choice(10, 100, p=prioritized_prob)
print(f"  均匀采样: {np.bincount(uniform_samples, minlength=10)}")
print(f"  优先采样: {np.bincount(prioritized_samples, minlength=10)}")
print(f"  优先采样更频繁地选择TD误差大的经验(3,6,8)")

什么用（应用）：PER在DQN基础上显著加速了学习，在Atari游戏上提高了约2倍的学习速度。在稀疏奖励环境中，PER尤其有效——因为"获得奖励"的经验非常稀少，这些经验应该被频繁回放。在推荐系统中，高点击/高购买的经验应被优先学习。

哪些坑（缺点）：优先采样引入了偏差，需要重要性采样修正；需要维护优先级数据结构（如SumTree），增加计算开销；对噪声敏感——TD误差大的经验不一定是"重要的"，可能只是噪声。

二、SumTree——高效优先级采样

是什么（定义）：SumTree是一种二叉树数据结构，用于高效实现优先经验回放。每个叶子节点存储一条经验的优先级，内部节点存储子节点优先级之和。采样时，从根节点开始，随机选择一个阈值，然后递归地沿着树下降到对应的叶子节点。时间复杂度O(log N)。

大白话 SumTree就像"优先级抽奖箱"——优先级高的经验占的"面积"大，更容易被抽中。二叉树结构让抽奖过程非常高效——不需要遍历所有经验，沿着树往下走就能找到中奖的那个。

为什么（原理）：如果直接用数组存储优先级，每次采样需要O(N)时间（遍历所有经验计算累积概率），在经验池很大时（如10^6条经验）不可接受。SumTree通过树结构将采样复杂度降到O(log N)，使PER在实际中可行。

怎么做（实现）：

import numpy as np

class SumTree:
    """SumTree数据结构——实现优先经验回放"""
    def __init__(self, capacity):
        self.capacity = capacity
        self.tree = np.zeros(2 * capacity - 1)  # 二叉树数组
        self.data = np.zeros(capacity, dtype=object)  # 存储经验
        self.data_pointer = 0  # 当前写入位置
        self.n_entries = 0  # 当前经验数量

    def add(self, priority, experience):
        """添加经验，赋予优先级"""
        idx = self.data_pointer + self.capacity - 1  # 叶子节点位置
        self.data[self.data_pointer] = experience
        self.update(idx, priority)
        self.data_pointer = (self.data_pointer + 1) % self.capacity
        if self.n_entries < self.capacity:
            self.n_entries += 1

    def update(self, idx, priority):
        """更新叶子节点的优先级，并向上传播"""
        change = priority - self.tree[idx]
        self.tree[idx] = priority
        while idx != 0:  # 向上传播到根节点
            idx = (idx - 1) // 2
            self.tree[idx] += change

    def get_leaf(self, value):
        """根据采样值获取叶子节点"""
        parent = 0
        while True:
            left = 2 * parent + 1
            right = left + 1
            if left >= len(self.tree):  # 到达叶子节点
                leaf_idx = parent
                break
            if value <= self.tree[left]:
                parent = left
            else:
                value -= self.tree[left]
                parent = right
        data_idx = leaf_idx - self.capacity + 1
        return leaf_idx, self.tree[leaf_idx], self.data[data_idx]

    def total_priority(self):
        return self.tree[0]  # 根节点 = 总优先级

# 演示SumTree
np.random.seed(42)
tree = SumTree(capacity=10)

# 添加经验
for i in range(10):
    td_error = np.abs(np.random.randn())  # 模拟TD误差
    tree.add(td_error, f"经验{i}")

print("=== SumTree演示 ===")
print(f"容量: {tree.capacity}")
print(f"当前经验数: {tree.n_entries}")
print(f"总优先级: {tree.total_priority():.2f}")
print(f"\nSumTree结构（二叉树数组）:")
for level in range(4):
    start = 2**level - 1
    end = min(2**(level+1) - 1, len(tree.tree))
    if start >= len(tree.tree): break
    values = [f"{tree.tree[i]:.2f}" for i in range(start, end) if tree.tree[i] > 0]
    if values:
        print(f"  层级{level}: {values}")

print(f"\n按优先级采样（高优先级经验更容易被抽中）：")
for _ in range(5):
    value = np.random.random() * tree.total_priority()
    leaf_idx, priority, exp = tree.get_leaf(value)
    print(f"  采样值={value:.2f} → {exp}, 优先级={priority:.2f}")

什么用（应用）：SumTree是PER的标准实现，在OpenAI的baselines、Stable-Baselines3等RL库中广泛使用。理解SumTree有助于实现高效的PER，也能帮助理解"优先级采样"这一通用思想。

哪些坑（缺点）：SumTree需要额外的内存（2倍数组大小）；在GPU训练中，SumTree的更新在CPU上，可能成为瓶颈；初始优先级设置（新经验给高优先级）影响早期训练。

三、重要性采样——修正优先采样的偏差

是什么（定义）：优先采样改变了经验的数据分布，使得训练数据不再是"从真实环境中均匀采集"的分布。重要性采样（Importance Sampling，IS）通过给每条经验加权来修正这个偏差：w_i = (1/N · 1/P(i))^β，其中N是经验池大小，P(i)是采样概率，β是修正程度。

大白话 优先采样就像"你多花时间复习重点题"——这很好，但如果你只复习重点题，你学到的"考试分布"就和真实考试不一致了。重要性采样就是给那些被"过度复习"的重点题降权重，给那些"被冷落"的普通题升权重，让最终的学习效果与真实分布一致。

为什么（原理）：优先采样使得P(i) ≠ 1/N（均匀分布），这导致梯度估计有偏。重要性采样权重w_i = (1/(N·P(i)))^β修正了这个偏差：当β=1时完全修正，但会降低收敛速度；当β=0时完全不修正。实践中β从较小值（如0.4）线性增加到1，实现"先快速学习，再精细修正"。

怎么做（实现）：

import numpy as np

# ==================== 重要性采样权重演示 ====================
N = 1000  # 经验池大小
td_errors = np.random.exponential(1.0, N)  # 模拟TD误差（指数分布）
alpha = 0.6  # 优先程度

# 计算采样概率
priorities = (np.abs(td_errors) + 0.01) ** alpha
P = priorities / priorities.sum()

# 计算重要性采样权重
beta_values = [0.0, 0.4, 0.7, 1.0]

print("=== 重要性采样修正演示 ===")
print(f"经验池大小 N={N}")
print(f"优先程度 α={alpha}")
print()
print("不同β值下的重要性采样权重：")
print(f"{'β':<6} {'高优先级(TD=5.0)':<25} {'中优先级(TD=1.0)':<25} {'低优先级(TD=0.01)':<25}")
print("-" * 80)

for beta in beta_values:
    # 模拟三条经验
    for td in [5.0, 1.0, 0.01]:
        p = (td + 0.01) ** alpha / priorities.sum()
        w = (N * p) ** (-beta)
        if td == 5.0:
            w_high = w
        elif td == 1.0:
            w_mid = w
        else:
            w_low = w
    print(f"{beta:<6} {w_high:<25.2f} {w_mid:<25.2f} {w_low:<25.2f}")

print(f"\n分析：")
print(f"  β=0: 所有经验权重相同，不修正偏差")
print(f"  β=1: 完全修正偏差，高优先级经验权重降低，低优先级经验权重升高")
print(f"  实践中β从0.4逐步增加到1.0")

# 权重归一化（稳定训练）
weights = np.array([w_high, w_mid, w_low])
normalized_weights = weights / np.max(weights)
print(f"\n归一化权重 (β=1): {[f'{w:.2f}' for w in normalized_weights]}")
print(f"  归一化确保权重不超过1，稳定训练")

什么用（应用）：重要性采样是PER正确性的保证。在实际实现中，权重被乘入Q-learning的损失函数中，高优先级经验的梯度被降权，低优先级经验的梯度被升权。理解IS有助于调试PER训练中的收敛问题。

哪些坑（缺点）：IS权重归一化到[0,1]，实际上削弱了修正效果；β的退火速度需要调参；在高噪声环境中，TD误差大可能只是噪声，导致IS权重"错误地"修正。

四、PER完整实现

是什么（定义）：将SumTree和重要性采样结合起来，实现完整的优先经验回放。核心流程：存储经验时赋予初始最高优先级（确保新经验被至少学习一次），采样时按优先级加权采样，更新时同时更新网络和优先级，梯度更新时使用IS权重。

大白话 PER就像一个"智能错题本"——新题目先给高优先级（确保不被忽略），做错的题目（TD误差大）保留高优先级多复习，做对的题目（TD误差小）降优先级少复习。每次复习时，根据"被复习了多少次"来调整权重，保证最终学到的知识是准确的。

为什么（原理）：PER的完整实现需要协调三个组件：SumTree（高效采样）、IS权重（偏差修正）、优先级更新（动态调整）。初始优先级设为最大值（或当前最大值），确保每条新经验至少被采样一次。训练后根据新的TD误差更新优先级，形成"学习→更新优先级→更高效学习"的正反馈循环。

怎么做（实现）：

import numpy as np

class PrioritizedReplayBuffer:
    """完整的优先经验回放池"""
    def __init__(self, capacity, alpha=0.6, beta=0.4, beta_increment=0.001):
        self.capacity = capacity
        self.alpha = alpha  # 优先程度
        self.beta = beta    # IS修正程度
        self.beta_increment = beta_increment
        self.epsilon = 0.01  # 最小优先级
        self.tree = SumTree(capacity)  # SumTree存储

    def push(self, state, action, reward, next_state, done):
        """存储经验，初始给予最高优先级"""
        max_priority = max(self.tree.tree[-self.tree.capacity:])
        if max_priority == 0:
            max_priority = 1.0
        experience = (state, action, reward, next_state, done)
        self.tree.add(max_priority, experience)  # 新经验给最高优先级

    def sample(self, batch_size):
        """按优先级采样，返回经验、索引和IS权重"""
        batch = []
        indices = []
        priorities = []
        segment = self.tree.total_priority() / batch_size

        # 分层采样：将总优先级分成batch_size段
        for i in range(batch_size):
            a = segment * i
            b = segment * (i + 1)
            value = np.random.uniform(a, b)
            idx, priority, experience = self.tree.get_leaf(value)
            batch.append(experience)
            indices.append(idx)
            priorities.append(priority)

        # 计算重要性采样权重
        probs = np.array(priorities) / self.tree.total_priority()
        weights = (self.tree.n_entries * probs) ** (-self.beta)
        weights /= weights.max()  # 归一化

        # β逐步增加
        self.beta = min(1.0, self.beta + self.beta_increment)

        # 解压batch
        states = np.array([b[0] for b in batch])
        actions = np.array([b[1] for b in batch])
        rewards = np.array([b[2] for b in batch])
        next_states = np.array([b[3] for b in batch])
        dones = np.array([b[4] for b in batch])

        return states, actions, rewards, next_states, dones, indices, weights

    def update_priorities(self, indices, td_errors):
        """根据新的TD误差更新优先级"""
        for idx, td_error in zip(indices, td_errors):
            priority = (abs(td_error) + self.epsilon) ** self.alpha
            self.tree.update(idx, priority)

    def __len__(self):
        return self.tree.n_entries

# 演示
np.random.seed(42)
per_buffer = PrioritizedReplayBuffer(capacity=100)

# 存储经验
for i in range(50):
    state = np.random.randn(4)
    action = np.random.randint(4)
    reward = np.random.randn()
    next_state = np.random.randn(4)
    done = False
    per_buffer.push(state, action, reward, next_state, done)

print("=== PER完整流程 ===")
print(f"经验池大小: {len(per_buffer)}")
print(f"总优先级: {per_buffer.tree.total_priority():.2f}")
print(f"当前β: {per_buffer.beta:.3f}")

# 采样
states, actions, rewards, next_states, dones, indices, weights = per_buffer.sample(8)
print(f"\n采样batch大小: {len(states)}")
print(f"IS权重: {[f'{w:.3f}' for w in weights]}")
print(f"样本索引: {indices}")

# 模拟训练后更新优先级
new_td_errors = np.abs(np.random.randn(8))
per_buffer.update_priorities(indices, new_td_errors)
print(f"\n更新优先级后，新TD误差: {[f'{e:.2f}' for e in new_td_errors]}")
print(f"更新后β: {per_buffer.beta:.3f}")
print(f"β从0.4逐步增加到1.0（先快学，后修正）")

什么用（应用）：PER是Rainbow的重要组成部分，在Atari游戏上显著提升了学习速度和最终性能。在实际RL项目中，当数据收集成本高或经验质量差异大时，PER尤其有效。

哪些坑（缺点）：实现复杂（SumTree + IS权重）；超参数（α, β, β_increment）需要调参；对噪声环境敏感；在分布式训练中，跨进程共享优先级比较复杂。

五、PER的核心参数调优

是什么（定义）：PER有三个关键超参数：α（优先程度）、β（IS修正程度）、ε（最小优先级）。α=0退化为均匀采样，α=1完全按TD误差优先；β从β_0逐渐增加到1；ε防止零优先级。

大白话 α决定了"你有多重视重点题"——α=0和以前一样，α=1重点题被疯狂复习。β决定了"你有多在意偏差"——β=1完全修正偏差但学得慢。调参就是找α和β的最佳平衡点。

怎么做（实现）：

import numpy as np

# ==================== α和β的调参指南 ====================
print("=== PER超参数调参指南 ===")
print()
print("α（优先程度）：")
print("  0.0: 均匀采样（退化为普通经验回放）")
print("  0.4: 轻度优先，适合噪声较大的环境")
print("  0.6: 默认推荐值，大多数任务的最优选择")
print("  0.8: 重度优先，适合稀疏奖励/关键经验稀少的环境")
print("  1.0: 完全优先，可能导致过拟合到少数经验")
print()
print("β（IS修正程度）：")
print("  初始值 β_0:")
print("    0.4: 标准推荐，前期快速学习")
print("    0.5: 稍微保守，适合重要经验多但噪声也多的环境")
print("  最终值: 1.0（完全修正）")
print("  增量: 0.001/step（典型值），使β在训练中线性增加到1.0")
print()
print("ε（最小优先级）：")
print("  0.01: 标准值，确保所有经验有非零概率")
print("  0.001: 更小的值，给低优先级经验更少的采样机会")
print()
print("推荐配置：")
print("  α=0.6, β_0=0.4, β_increment=0.001, ε=0.01")
print("  这是Rainbow论文中的默认配置，在大多数任务中表现良好")

什么用（应用）：理解PER的超参数有助于在实际项目中快速调参。稀疏奖励环境→α大一些（0.70.8）；噪声大的环境→α小一些（0.40.5）；需要快速收敛→β_0小一些，β_increment大一些。

哪些坑（缺点）：α太大可能导致"只学少数经验"而忽略其他重要经验；β_increment太大可能导致后期修正不足；超参数之间相互影响，需要联合调优。

六、实战：PER vs 均匀回放对比

是什么（定义）：在网格世界上对比均匀经验回放和优先经验回放的学习速度和收敛效果。

怎么做（实现）：

import numpy as np
from collections import deque

n_states = 9; n_actions = 4
actions = [(-1, 0), (1, 0), (0, -1), (0, 1)]

def step(state, action):
    row, col = state // 3, state % 3
    dr, dc = actions[action]
    nr = max(0, min(2, row + dr))
    nc = max(0, min(2, col + dc))
    next_state = nr * 3 + nc
    if next_state == 8:   r = 10.0
    elif next_state == 7: r = -5.0
    else:                 r = -0.1
    return next_state, r, next_state in [7, 8]

def to_vec(s): v = np.zeros(n_states); v[s] = 1.0; return v

def train_replay(replay_type="uniform", n_episodes=200):
    hidden = 32
    W1 = np.random.randn(n_states, hidden) * 0.1; b1 = np.zeros(hidden)
    W2 = np.random.randn(hidden, n_actions) * 0.1; b2 = np.zeros(n_actions)
    tW1, tb1, tW2, tb2 = W1.copy(), b1.copy(), W2.copy(), b2.copy()
    buffer = deque(maxlen=500) if replay_type == "uniform" else PrioritizedReplayBuffer(500)
    epsilon = 1.0; rewards = []
    step_count = 0

    def get_q(s, target=False):
        w1, b1_, w2, b2_ = (tW1, tb1, tW2, tb2) if target else (W1, b1, W2, b2)
        return np.maximum(0, s @ w1 + b1_) @ w2 + b2_

    for ep in range(n_episodes):
        state = 0; total = 0
        while state not in [7, 8]:
            if np.random.random() < epsilon:
                action = np.random.randint(n_actions)
            else:
                action = np.argmax(get_q(to_vec(state)))
            next_state, reward, done = step(state, action)
            if replay_type == "uniform":
                buffer.append((to_vec(state), action, reward, to_vec(next_state), float(done)))
            else:
                buffer.push(to_vec(state), action, reward, to_vec(next_state), float(done))
            total += reward; state = next_state; step_count += 1

            if len(buffer) >= 32:
                if replay_type == "uniform":
                    idx = np.random.choice(len(buffer), 32, replace=False)
                    batch = [buffer[i] for i in idx]
                    s = np.array([b[0] for b in batch]); a = np.array([b[1] for b in batch])
                    r = np.array([b[2] for b in batch]); ns = np.array([b[3] for b in batch])
                    d = np.array([b[4] for b in batch])
                else:
                    s, a, r, ns, d, indices, weights = buffer.sample(32)

        epsilon = max(0.05, epsilon * 0.995)
        if ep % 50 == 0:
            tW1, tb1, tW2, tb2 = W1.copy(), b1.copy(), W2.copy(), b2.copy()
        rewards.append(total)
    return rewards

np.random.seed(42)
r_uniform = train_replay("uniform")
r_per = train_replay("per")

print("=== 均匀回放 vs 优先回放 ===")
print(f"均匀回放最后50ep平均: {np.mean(r_uniform[-50:]):.2f}")
print(f"优先回放最后50ep平均: {np.mean(r_per[-50:]):.2f}")
print(f"\nPER的核心优势：")
print(f"  1. 重要经验（高TD误差）被更频繁学习")
print(f"  2. 在稀疏奖励/关键经验稀少的环境中加速明显")
print(f"  3. 重要性采样保证学习的无偏性")

什么用（应用）：通过对比实验，直观理解PER的效果。在更复杂的环境中，PER的优势更加明显。

哪些坑（缺点）：在简单环境中，PER的优势可能不明显（几乎所有经验都"重要"）；PER的实现复杂度较高，调试困难。

概念关系图谱

重点答疑

Q1: PER为什么能加速学习？

PER加速学习有两个原因：(1) 高TD误差的经验被更频繁地回放，这些经验是"当前网络最应该学习的"——它们包含最大的"惊喜"或"信息量"；(2) 在稀疏奖励环境中，获得奖励的经验非常稀少，均匀回放可能很久才采样到一次，PER确保这些珍贵经验被及时学习。

解答：PER = 把时间花在刀刃上。与其均匀复习所有题目，不如多花时间在"最让你意外"的题目上。

Q2: 为什么需要重要性采样？不修正会怎样？

如果不修正，优先采样改变了训练数据的分布，导致：(1) Q值估计有偏——学到的Q值倾向于高优先级经验的分布，而非真实环境分布；(2) 最终策略可能次优——因为学习的目标分布与真实分布不一致。重要性采样通过权重修正，确保最终学到的Q值是无偏的。

解答：不修正就像"只复习重点题，不复习普通题"——你可能会在考试中遇到没复习的题型而丢分。IS权重就是给你的学习"校正"一下。

Q3: α和β应该怎么设置？

标准推荐：α=0.6, β从0.4线性增加到1.0。具体调整：

• 稀疏奖励环境 → α大一些（0.7~0.8），因为奖励经验稀少且重要。
• 噪声大的环境 → α小一些（0.4~0.5），因为高TD误差可能只是噪声。
• 需要快速收敛 → β_0小一些（0.3），β_increment大一些。
• 需要高精度 → β_0大一些（0.5），β_increment小一些。

解答：α=0.6, β_0=0.4是最安全的默认配置，大多数情况下表现良好。

Q4: 新加入经验池的经验，应该给什么优先级？

新经验给予当前最高的优先级，确保每条新经验至少被学习一次。如果给低优先级，新经验可能永远不被采样（尤其是在经验池很大的情况下），导致"信息浪费"。学习后，根据实际TD误差更新优先级。

解答：新经验就像"新来的转学生"——先给个高关注度，了解之后再决定是否值得继续关注。

Q5: PER和DQN的其他改进（Double、Dueling）可以同时使用吗？

可以，而且推荐同时使用。PER修改的是"采样方式"，Double修改的是"目标计算"，Dueling修改的是"网络结构"，三者互不冲突。在Rainbow中，PER与其他改进一起使用，效果叠加。

解答：PER改采样，Double改目标，Dueling改结构——三者正交，同时使用就是Rainbow的标配。

章节单词汇总