★ FEATURED ARTICLE 我做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, axis0) # 形状 (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_maxTrue): 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处理终止状态时一般规则是如果doneTrue代表这个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 OverAtari很多游戏有“生命”概念。环境返回的done通常代表整局游戏结束而不是一条命结束。但对DQN来说一条命全损和整局结束都是重要的终止信号。很多改进版本在“生命值发生变化”时也把下一个状态置零或者把reward惩罚加大让网络学会珍惜生命。我建议在做数据初始化时先用info里的life字段做判断如果life比上一刻少说明丢了一条命这本身就是一个强终止信号如果环境返回doneTrue说明整局结束更是终止。不过这一处改动会直接影响状态更新逻辑和回放缓冲区里的done标签属于“改一处牵全身”的操作。第一次搭管线时可以先用环境自身的done后面调效果时再引入life信号。4. 经验回放数据初始化攒出第一批可用的batch4.1 随机探索阶段怎么采集初始转移强化学习一开始没有任何策略最稳妥的方式是随机采样动作让环境自己产生一批转移数据。这个过程也叫warm-up或随机探索。初始化规模怎么定看你神经网络和任务复杂度。小型任务比如Pong这种攒5000到10000条转移就能开始训练复杂一点的平台跳跃类游戏我建议至少攒20000到50000条。太少了网络每个batch看到的都是彼此高度相关的近期帧训练不稳定太多了又拖慢实验节奏。在Jupyter里我会把初始化过程写成一个独立cell跑完看一眼缓冲区大小再继续。init_steps 20000 use_max True buffer ReplayBuffer(capacity100000, state_shape(4, 84, 84)) frame_queue deque(maxlen4) # 先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(maxlenN)往里塞五元组。优点是代码少缺点是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, dtypenp.uint8) self.actions np.zeros(capacity, dtypenp.int64) self.rewards np.zeros(capacity, dtypenp.float32) self.next_states np.zeros((capacity,) state_shape, dtypenp.uint8) self.dones np.zeros(capacity, dtypenp.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, sizebatch_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, axis0)得到(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的错误数据。调试方法很简单统计一下缓冲区里done1的样本数量再看这些样本的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写好了训练时直接从里面随机取batchstates, 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没有被真正执行。这时候千万别急着训练先把环境交互逻辑搞清楚。
阅读完成 · 觉得有帮助？
