我做Atari游戏训练，一般不会一上来就写完整训练循环，而是先把最枯燥但最关键的数据入口搞定：图片处理、state更新、经验回放数据初始化。这一篇就专门聊这套流程，并且会用Jupyter Notebook作为调试现场。对刚开始做DQN类强化学习的同学来说，如果把这一步走稳，后面接网络、接采样、接训练都会顺手很多；如果这一块图省事，后面十有八九会花更多时间排查样本错位、维度不匹配、终端状态标错这类坑。
市面上很多博客直接贴训练代码，默认你已经把preprocess、state stack、replay buffer全部准备好了。但实际从零搭的时候，最容易被卡住的反而不是网络结构，而是这几个数据基建环节。这篇文章我会按我自己的实现顺序来写：先讲整体管线，再拆图片处理，然后说清楚state怎么更新，最后把经验回放数据初始化的可复现代码给出来。
1. 先想清楚：这一节到底在搭什么
1.1 一套Atari训练管线里，数据入口为什么这么重要
Atari 2600这类游戏环境，单个原始帧是210x160x3的RGB图像，也就是满屏彩色的高分辨率画面。如果直接把这个当网络输入，不是不行，但代价很高：通道多、尺寸大、信息冗余。更重要的是，单帧图像根本体现不出运动信息，像乒乓里球速很快，单看某一帧，你根本不知道球是往左还是往右飞。所以强化学习里处理Atari的惯例是：
- 对每一帧做灰度化、裁剪、缩放、归一化；
- 把连续4帧堆叠起来作为一个状态；
- 这个状态送给Q网络，作为当前时刻的环境观测。
图片处理加state更新，解决的其实就是“把环境原始帧变成网络能吃进去的张量”这件事。而经验回放数据初始化，解决的是“在开始正式训练之前，先攒一批像样的训练样本”。
这批初始化样本可不是随便采的。它要覆盖不同状态、不同动作、不同奖励的转移，而且必须是完整的五元组：当前状态、动作、奖励、下一状态、是否终止。只有攒够了这批数据，后面的batch采样才有的采，Q学习才有一个相对稳定的训练分布。
1.2 Jupyter在这个流程里能充当什么角色
很多人觉得Jupyter只适合做数据分析、不适合跑强化学习，我不太同意。强化学习训练确实应该放在正式脚本里，但“训练前的开发调试点”用Jupyter非常适合。
你可以一步步查看中间结果：原始帧长什么样、灰度化后信息还在不在、裁剪区域有没有切到游戏主要内容、堆叠后的4帧是否按时间顺序排列、回放缓冲区里采出来的一个batch是什么形状。这些如果在纯脚本里做，每次都要print或写日志；在Jupyter里可以直接可视化，省下大量时间。
我习惯的姿势是：先用Jupyter把预处理、状态更新、回放初始化全部验证通过，再把这个逻辑原封不动搬到一个训练类里。这样训练时少踩很多隐藏坑。
2. 图片处理：原始屏幕帧如何变成网络输入
2.1 四件套：灰度化、裁剪、缩放、归一化
Atari原始画面的第一个问题是色彩。RL算法不靠颜色也能玩，彩色的颜色分布对游戏语义帮助很小，但会把输入维度撑到3倍。所以第一步先转灰度。
第二个问题是画面里有很多无关区域。比如顶部记分板、底部某些游戏条带，这些区域对决策没用，还浪费计算量。常见做法是把高度从210裁剪到160，具体是去掉上边大约34像素、下边16像素，把它变成一个近似160x160的方形区域。
第三个问题是尺寸还是太大。160x160虽然能接受，但很多经典算法使用84x84，因为网络更小、训练更快，而且84x84足够保留游戏里的关键物体信息。这里顺带说一句：84x84并不是圣旨，如果自己调试环境，用80x80或者96x96都可以，关键是要让之后的网络输入维度能够对齐。
最后是归一化。像素值范围是0到255，不归一化直接喂网络会放大数值波动，让优化过程更困难。一般直接除以255，把范围压到0到1。
2.2 图片预处理的可运行代码
下面是我在Notebook里最常用的一套预处理函数：
import numpy as np import random from collections import deque def resize_frame(frame, target): # 这里替换成你环境里顺手的那套图像resize接口 # 无论是哪套接口，都建议使用双线性插值而不是最近邻 return frame # 实际部署时替换为真实resize实现 def preprocess_frame(raw, target=(84, 84)): # raw: 形状为 (210, 160, 3) 的RGB帧，HWC排列 rgb = raw.astype(np.float32) # 1. 灰度化，按标准亮度加权公式 gray = 0.299 * rgb[..., 0] + 0.587 * rgb[..., 1] + 0.114 * rgb[..., 2] # 2. 裁剪：去掉顶部记分板和底部无信息区域 # 原高度210，顶部去掉34，底部去掉16，得到 160 高度 cropped = gray[34:194, :] # 3. 缩放到 84x84 resized = resize_frame(cropped, target) # 4. 归一化到 [0, 1] normalized = (resized / 255.0).astype(np.float32) return normalized这套代码有两个细节值得注意。
第一，灰度化权重不是随便平均三个通道。标准亮度感知权重是0.299、0.587、0.114，我最早图省事直接取平均，结果游戏背景里的颜色差异被扭曲，画面看起来发灰发闷，虽然也能训练，但总觉得不对。后来改成标准权重后才恢复正常。
第二，resize插值方式会影响画面质量。如果使用最近邻，转折边缘会出现明显锯齿，某些细小球体或狭长挡板可能产生像素错位。用双线性插值会平滑很多。这个小差别在单帧上不明显，但在百万帧训练规模下，会让网络接收到的模式更稳定。
2.3 画面闪烁问题与最大帧合并
Atari模拟器里有一个比较微妙的现象：很多游戏画面会闪烁。部分物体在奇数帧和偶数帧交替出现，比如某些射击游戏的子弹、断断续续的闪烁块。如果直接用单帧，网络可能看到的是“子弹闪一下又消失”，根本学不到连贯语义。
比较好的解决办法，是在预处理后再做一次“最大帧合并”。具体做法是：把当前处理后的帧和上一帧逐像素取最大值。
def max_frame_pair(next_frame, old_frame): return np.maximum(next_frame, old_frame)为什么要取最大，而不是平均？因为物体闪烁的本质是同一位置在某一帧有、在某一帧没有。取最大值可以保留这个物体，平均则会把物体变淡。这样处理之后，闪烁的子弹或道具能稳定出现在画面里，网络学起来容易得多。
这个操作记得放在堆叠之前，否则你可能会先把闪烁信息带进state，然后又用最大帧合并破坏已经堆叠好的状态。
2.4 图片处理里我踩过的坑
先说一个最常见的问题：归一化位置不一致。有些地方先把灰度值除以255，再resize；有些地方resize之后再除255。二者结果几乎一样，但因为插值计算发生在0到255和0到1两个不同尺度上，float32精度下可能差一点点。这不是大问题，真正要命的是训练代码和数据初始化代码不统一。我遇到过一套代码里预处理函数返回0到1，另一套代码返回0到255，结果网络Loss上下乱跳。
我的建议是：把“归一化”作为预处理的最后一步，并且只在一处实现。初始化回放数据时用这份函数，训练时采样本也调用同一份函数，不要在数据管线的不同环节各写各的。
另一个坑是uint8和float32混用。如果为了省内存把帧存成uint8，就一定要在采样的时候统一转float并除以255。如果初始化时存进去的是uint8，采样时忘了转，网络输入一半是0到255一半是0到1，梯度直接乱掉。这个问题我在后面回放缓冲区部分还会重点说。
3. state更新：4帧堆叠和时间语义
3.1 为什么是4帧，不是1帧也不是10帧
单帧没有速度信息，模型无法判断物体运动方向。那是不是堆叠越多越好？也不是。帧数越多，状态维度越高，网络训练越重，而且早期帧和当前帧相隔太远，相关性已经很弱。4帧是一个经典折中值，PyTorch、TensorFlow里的很多参考实现都用它。
你可以把state理解成一个小视频片段：4张连续处理后的灰度图，按时间前后排列在通道维上。网络第一层卷积能同时看到这4帧，从它们的变化中提取出运动模式。
我习惯的状态张量形状是(4, 84, 84)，也就是CHW排列：4个通道，每个通道是一张84x84的灰度图。也有同学习惯用(84, 84, 4)的HWC排列。两种都能做，但一旦定了就不要改，因为后面卷积层的输入通道数、回放缓冲区里的数组形状、采样后reshape全部要对齐。
3.2 state更新的核心逻辑
用一个长度为4的队列来维护最近4帧，每次环境走一步后，把新的预处理帧追加进去，最老的那一帧自动被挤掉。
def make_state(frame_queue): return np.stack(frame_queue, axis=0) # 形状 (4, 84, 84) def reset_state(env, preprocess_fn, frame_queue): raw = env.reset() frame = preprocess_fn(raw) # 开局没有历史帧，用当前帧重复4次作为初始状态 frame_queue.clear() for _ in range(4): frame_queue.append(frame) return make_state(frame_queue)注意这里初始状态的做法：有人喜欢把历史帧填零，我更喜欢用当前帧重复4次。原因是初始状态本质上就是“游戏刚开始那一刻”，重复当前帧不会引入虚假信息，但全零帧会让网络以为画面是全黑的，可能诱发奇怪的动作偏好。
环境每执行一步，state的更新方式如下：
def step_and_collect(env, frame_queue, action, preprocess_fn, use_max=True): current_state = make_state(frame_queue) raw_next, reward, done, info = env.step(action) next_frame = preprocess_fn(raw_next) if use_max and len(frame_queue) > 0: next_frame = max_frame_pair(next_frame, frame_queue[-1]) frame_queue.append(next_frame) next_state = make_state(frame_queue) return current_state, action, reward, next_state, done, info这里有一个特别容易忽略的点：frame_queue是deque对象，append之后原地变了。如果你在保存transition时不小心直接保存了frame_queue的引用，而不是复制后的状态数组，那么后面队列继续更新时，之前保存的“下一状态”也会跟着变。我早期踩过一次，所有transition的next_state全都变成最新的那一帧堆叠，简直是一场灾难。所以一定要用np.stack生成一个新的数组，再把当前状态和下一状态存进缓冲区。
3.3 done状态下的state边界case
处理终止状态时，一般规则是：如果done=True，代表这个episode结束了，没有“下一状态”可言。传统做法会把下一状态置成零张量，同时把done标记为1，让网络知道这个转移之后没有后续回报。
如果在环境返回done后继续reset，然后马上拿新episode的第一帧当next_state存进去，会造成一个“跨episode的假转移”。Q学习更新时，上一局最后一帧后面接的是新一局的第一帧，语义完全断裂，会让价值估计出现偏差。
所以我在保存transition前会做一个判断：
if done: next_state = np.zeros_like(current_state) buffer.push(current_state, action, reward, next_state, True) # 之后再重置环境，并重新初始化frame_queue frame_queue.clear() raw = env.reset() first_frame = preprocess_fn(raw) for _ in range(4): frame_queue.append(first_frame) else: buffer.push(current_state, action, reward, next_state, False)这就是state更新里最重要的一个边界case。
3.4 关于episode内生命损失和真正Game Over
Atari很多游戏有“生命”概念。环境返回的done通常代表整局游戏结束，而不是一条命结束。但对DQN来说，一条命全损和整局结束，都是重要的终止信号。很多改进版本在“生命值发生变化”时也把下一个状态置零，或者把reward惩罚加大，让网络学会珍惜生命。
我建议在做数据初始化时，先用info里的life字段做判断：
- 如果
life比上一刻少，说明丢了一条命，这本身就是一个强终止信号； - 如果环境返回
done=True，说明整局结束，更是终止。
不过这一处改动会直接影响状态更新逻辑和回放缓冲区里的done标签，属于“改一处牵全身”的操作。第一次搭管线时，可以先用环境自身的done，后面调效果时再引入life信号。
4. 经验回放数据初始化：攒出第一批可用的batch
4.1 随机探索阶段怎么采集初始转移
强化学习一开始没有任何策略，最稳妥的方式是随机采样动作，让环境自己产生一批转移数据。这个过程也叫warm-up或随机探索。
初始化规模怎么定？看你神经网络和任务复杂度。小型任务比如Pong这种，攒5000到10000条转移就能开始训练；复杂一点的平台跳跃类游戏，我建议至少攒20000到50000条。太少了，网络每个batch看到的都是彼此高度相关的近期帧，训练不稳定；太多了，又拖慢实验节奏。
在Jupyter里，我会把初始化过程写成一个独立cell，跑完看一眼缓冲区大小，再继续。
init_steps = 20000 use_max = True buffer = ReplayBuffer(capacity=100000, state_shape=(4, 84, 84)) frame_queue = deque(maxlen=4) # 先reset环境，顺便做no-op warm-up raw = env.reset() first_frame = preprocess_frame(raw) for _ in range(4): frame_queue.append(first_frame) # 随机执行若干次“无操作”动作，跳过开始画面 for _ in range(random.randint(10, 30)): raw_next, reward, done, info = env.step(0) # 动作0通常表示无操作 frame = preprocess_frame(raw_next) if done: raw = env.reset() frame = preprocess_frame(raw) frame_queue.clear() for _ in range(4): frame_queue.append(frame) break frame_queue.append(frame) collected = 0 while collected < init_steps: action = env.action_space.sample() # 随机动作 current_state = make_state(frame_queue) raw_next, reward, done, info = env.step(action) next_frame = preprocess_frame(raw_next) if use_max and len(frame_queue) > 0: next_frame = max_frame_pair(next_frame, frame_queue[-1]) frame_queue.append(next_frame) next_state = make_state(frame_queue) if done: next_state = np.zeros_like(current_state) buffer.push(current_state, action, clip_reward(reward), next_state, True) raw = env.reset() first_frame = preprocess_frame(raw) frame_queue.clear() for _ in range(4): frame_queue.append(first_frame) else: buffer.push(current_state, action, clip_reward(reward), next_state, False) collected += 1这个代码看起来长，但每一段都对应一个实际环节：no-op跳过开头、随机动作、预处理、最大帧合并、state更新、terminal判断、入缓冲区。
4.2 回放缓冲区用什么数据结构
最简单的是deque(maxlen=N)，往里塞五元组。优点是代码少，缺点是Python对象开销大，而且采样不够灵活。我建议至少用一个numpy循环数组来实现固定容量缓冲区。
实现的时候要注意内存。如果把state直接以float32存，一个状态就是4*84*84*4 = 112896字节，10万条就是11GB以上，光state和next_state就要20多G，个人电脑扛不住。所以我在缓冲区里把图像转成uint8存储，采样时再除以255转回float。
class ReplayBuffer: def __init__(self, capacity, state_shape): self.capacity = capacity self.state_shape = state_shape self.idx = 0 self.size = 0 self.states = np.zeros((capacity,) + state_shape, dtype=np.uint8) self.actions = np.zeros(capacity, dtype=np.int64) self.rewards = np.zeros(capacity, dtype=np.float32) self.next_states = np.zeros((capacity,) + state_shape, dtype=np.uint8) self.dones = np.zeros(capacity, dtype=np.uint8) def push(self, state, action, reward, next_state, done): i = self.idx % self.capacity # 如果当前状态来自预处理函数，先转成0-255的uint8再存 self.states[i] = (state * 255.0).astype(np.uint8) self.actions[i] = action self.rewards[i] = reward self.next_states[i] = (next_state * 255.0).astype(np.uint8) self.dones[i] = 1 if done else 0 self.idx += 1 self.size = min(self.size + 1, self.capacity) def sample(self, batch_size): indices = np.random.randint(0, self.size, size=batch_size) states = self.states[indices].astype(np.float32) / 255.0 actions = self.actions[indices] rewards = self.rewards[indices] next_states = self.next_states[indices].astype(np.float32) / 255.0 dones = self.dones[indices].astype(np.float32) return states, actions, rewards, next_states, dones这样10万条缓冲区的内存占用大概是：两个uint8图数组加上其他小数组，大约5.6GB左右，虽然还是不小，但比float32版本舒服得多。如果还嫌大，可以存单帧而不是4帧state，等采样时再从帧索引序列里拼装出4帧stack，不过这会让代码复杂度上一个台阶，初期没必要。
4.3 reward要不要裁剪
Atari的原始reward经常是分数，可能一局里几十上百，也可能一直为零。直接存原始reward不是不行，但训练时不同游戏尺度差异太大，会给同一套超参数带来麻烦。经典DQN方案是把奖励clip到[-1, 0, 1]：正奖励变成1，负奖励变成-1，零奖励还是0。
def clip_reward(reward): return float(np.sign(reward))这个操作要在数据进缓冲区之前做。如果后面要训练，网络输出的Q值目标也会是这个小尺度，更容易稳定。有人担心clip会丢失奖励大小信息，对Atari来说这个担心在多数游戏里不成立，因为学习“赢还是输”远比“赢了多少分”重要。
4.4 在Jupyter里怎么验证初始化数据
我每次初始化完，不会急着训练，先做三个检查。
第一，看缓冲区大小和覆盖情况：
print(buffer.size)如果size已经等于capacity，说明缓冲区被循环覆盖过，这是正常现象，但要知道最早的数据已经被冲掉了。
第二，随机抽一个batch看看形状和数值范围：
states, actions, rewards, next_states, dones = buffer.sample(4) print(states.shape) # (4, 4, 84, 84) print(states.min(), states.max()) # 约 0.0 ~ 1.0 print(rewards) print(dones)第三，可视化其中一条状态的四帧图像。在Jupyter里直接用matplotlib把四张子图画出来，确认堆叠顺序是按时间排列的。这一步看起来笨，但能一眼看出预处理是否把游戏内容裁剪掉了、画面是否过亮过暗、是否有奇怪的条纹。
5. 常见问题与调试实录
5.1 报错信息直接指向维度不匹配
如果你在采样后把states喂给网络，报错说通道维不对，十有八九是state形状定义不一致。
我遇到过一种情况：预处理函数返回的是(84, 84)，但在reset_state里np.stack(frame_queue, axis=0)得到(4, 84, 84)；后面网络第一层卷积输入通道写的 4，按理说没问题。可是有一次我把另一个测试脚本里的预处理函数改成返回(1, 84, 84)，堆叠后变成(4, 1, 84, 84)，网络直接炸。调试半天才找到。
我的建议是：在预处理函数返回后立刻写一行断言。
assert normalized.shape == (84, 84), f"预处理输出形状错误: {normalized.shape}"这种断言在开发期能帮你省下大量排查时间。
5.2 done状态的next_state存错了
这个问题发生的频率比我预想的高很多。很多人会把reset之后的第一帧直接当作next_state存进缓冲区。我当时也这么干过，后来训练时发现网络总是学不好，因为Q目标里不断混入跨episode的错误数据。
调试方法很简单：统计一下缓冲区里done=1的样本数量，再看这些样本的next_state是否都是全零。如果发现next_state里有大量非零数值但done也是1，那就要检查重置逻辑了。
mask = buffer.dones == 1 non_zero_next = (buffer.next_states[mask].astype(np.float32) > 0).sum() print(non_zero_next)如果输出不是0，说明你的终止状态处理有问题。
5.3 Jupyter里环境状态交叉污染
Notebook里最坑的一点是cell执行顺序可以乱。比如你先把环境跑了一百步，然后回头重新执行“初始化帧队列”的cell，看起来每个cell都正常，但实际环境已经被推进到了第101步。这时候你采集的数据不是从开局开始的。
我后来用了一个很笨但有效的办法：把环境和frame_queue的初始化都放在同一个cell顶部，并且在实验开始时强制重启kernel，然后从第一个cell往下顺序执行。不要中途乱跳。
另外，Jupyter里定义的全局变量在重复执行某个cell后会被覆盖。比如replay buffer的实例，你执行一次没用，执行两次可能把之前攒的数据重复推到一起。所以我习惯把“初始化buffer”和“填充buffer”分开，填充cell前面加一个清空buffer的逻辑，或者直接重新创建对象。
5.4 框架选择和执行效率的矛盾
有人会问：初始化数据用Jupyter，训练时用脚本，代码不重复吗？我通常把“数据预处理、状态更新、回放缓冲区”这些核心函数写在一个公共模块里，比如atari_pipeline.py。Jupyter里import这个模块，训练脚本里也import同一个模块。这样调试时改一处，训练时立即生效，不会发生Notebook和脚本两套逻辑对不上的事故。
不要在一个Notebook里把全部强化学习训练代码都写完。Jupyter适合交互验证，不适合长时间稳定训练。如果训练在Notebook里跑十几个小时，中途浏览器刷新一下kernel断掉，全部进度就没了。
6. 初始化做完之后，下一步直接接什么
有了稳定的图片处理、正确的state更新、装满初始转移的经验回放缓冲区，接下来就是写batch采样和训练主循环。
我已经把ReplayBuffer.sample写好了，训练时直接从里面随机取batch：
states, actions, rewards, next_states, dones = buffer.sample(32)之后就是常见DQN套路：当前网络输出Q值，目标网络计算reward + gamma * max(Q_next) * (1 - done)，用MSE或Huber loss更新当前网络。
不过在选择优化器、学习率、目标网络更新频率之前，先把数据管线的“单步正确性”验证扎实。我在实际做的时候，初始化阶段花费的时间看起来不如训练那么刺激，但训练跑起来之后几乎不回头改数据代码。回头改一次，就要重新清空buffer重新初始化，浪费的时间比一开始慢慢调试还多。
最后再分享一个我个人的小习惯：每初始化完一批经验，我会在Notebook里把缓冲区里随机抽出来的几个transition打印一遍，尤其看“动作、奖励、done”三项是否合理。手写一个随机策略跑出来的数据，奖励应该零散地分布在 -1、0、1之间，done应该只出现在episode末尾。如果看到transition里reward全是0，或者done几乎不出现，那大概率是环境没有正确推进，或者action没有被真正执行。这时候千万别急着训练，先把环境交互逻辑搞清楚。
