DQN与CNN有何区别？强化学习经验回放持久化全攻略

先说结论这俩问题问得挺有意思但很多初学者容易把她们放在同一个维度上对比。DQN和CNN根本不是一个层面的东西——CNN是网络结构DQN是强化学习算法框架。至于怎么把训练经验保存到本地其实就是经验回放Replay Buffer的持久化实现这事儿我踩过不少坑值得好好聊清楚。如果你正在纠结我该用CNN还是DQN或者想弄明白训练过程中那些交互数据到底怎么存下来、断点续训怎么搞这篇文章就是为你准备的。1. DQN和CNN的本质区别一个在算法层一个在结构层1.1 CNN到底是什么它是一台特征提取机器CNN卷积神经网络是被设计来处理网格结构数据的网络结构。最典型的就是图像图像说白了就是一个二维像素矩阵我们让CNN去干的事就是从原始像素里提取出有用的特征。为什么不用全连接网络直接硬干图像因为图里的空间关系太重要了。比如一张猫的照片猫的眼睛和耳朵的相对位置是有意义的如果你把像素拉平成一维向量喂给全连接层那等于把这层空间关系全部打乱。CNN的三个核心机制就是专门解决这个问题的局部感受野每个神经元只连着上一层的某个局部区域先看局部再看全局相当于先看眼睛、再看耳朵、最后拼接出整张脸。参数共享同一个卷积核在整张图上滑动意思就是眼皮的特征提取器可以同时用在左眼和右眼上大幅减少参数量。平移不变性猫在图里往左挪了几个像素卷积依然能提取出同样的特征因为卷积核跟着滑动。所以CNN解决的是表征学习问题或者说怎么把原始像素变成有用的特征向量。它是你在强化学习里使用的一个工具本身不包含任何决策逻辑。1.2 DQN到底是什么它是一套决策学习框架DQN全称Deep Q-Network是深度强化学习领域里程碑级的算法。它的核心思想是用深度神经网络去拟合Q函数——就是在某个状态下采取某个动作未来累积回报的期望值。DQN解决的问题是决策问题。举个例子玩贪吃蛇你看到的画面是当前棋盘状态你要决定往左、往右、往上还是往下。这就是一个典型的状态到动作的映射问题。DQN要做的事情是通过不断试错学出这样一个Q表格或者Q网络然后在每个状态下都选择Q值最大的动作。所以DQN本质上是一个算法框架它得包含以下这些部件状态表征怎么描述当前环境状态原始像素还是一组数值。Q网络一个用来逼近Q函数的神经网络输入是状态输出是每个动作的Q值。奖励函数环境给我们的反馈信号。经验回放缓冲区存储历史交互数据。目标网络稳定训练用的老网路副本。策略比如epsilon-greedy用一定概率随机探索其余时候选Q值最大的动作。看到了吧CNN可能是Q网络内部的那个结构但DQN远比一个CNN要庞大得多。1.3 两者是什么关系CNN是DQN的眼睛和脑皮层这个问题最让人困惑的点在于很多经典DQN项目比如玩Atari游戏里面确实用到了CNN。Atari游戏的输入是屏幕像素吃豆人、太空侵略者这些画面就是一堆彩色像素块你总不能让一个全连接网络直接处理210×160×3的原始像素吧所以DQN的实现者就用CNN来做状态到Q值的映射网络。具体来说DQN的网络结构在Atari场景下长这样输入层: [84, 84, 4] 4帧灰度图堆叠捕捉运动信息 ↓ 卷积层1: 32个8×8卷积核步长4ReLU激活 ↓ 卷积层2: 64个4×4卷积核步长2ReLU激活 ↓ 卷积层3: 64个3×3卷积核步长1ReLU激活 ↓ 全连接层: 512个神经元ReLU激活 ↓ 输出层: 4个神经元4个动作的Q值这里的CNN就是Q网络的特征提取部分它把84×84×4的像素状态压缩成抽象特征再喂给后面的全连接层输出各个动作的价值。但我要强调一个容易误解的点DQN的Q网络不一定非得用CNN。如果状态是低维的比如机器人关节角度、速度、位置这些一维向量你用MLP多层感知机做Q网络完全没问题甚至效果更好。CNN在你当前任务里是最合适的特征提取器但换成MLP后依然是一个合法的DQN实现。所以区别概括成一句话就是CNN是零件DQN是整台机器CNN干的是从原始状态里提炼特征的活儿DQN干的是根据特征做决策并学习决策策略的活儿。一个在表征层一个在策略层。2. DQN为什么非要保存训练经验经验回放机制深度拆解2.1 强化学习的数据困境样本之间根本不独立你有想过为什么深度强化学习训练起来比监督学习要困难得多吗根本原因在数据。监督学习里我们为什么能直接用梯度下降因为训练样本是独立同分布的。样本之间没有相关性这一批和上一批没关系顺序也没本质影响。但强化学习的体验数据是一条时间序列S_t → A_t → R_t → S_{t1} → A_{t1} → R_{t1} → ...。相邻几帧之间相关性极高打个比方玩贪吃蛇的时候你第100帧和第101帧的画面差异可能只有一个像素点。如果你按时间顺序把这组数据喂给网络做梯度下降那你的梯度方向就会被这一小段高度相关的样本完全主导模型就会朝着某一次偶然尝试的方向剧烈震荡新学到的东西可能瞬间把之前学到的覆盖掉。这在数学上就是经典的打破独立同分布假设导致的训练不稳定问题。2.2 经验回放怎么解决这个问题随机采样打破相关性DQN论文里最关键的贡献之一就是经验回放机制。维护一个固定大小的缓冲区把每个时间步产生的四元组状态、动作、奖励、下一状态、结束标志存进去(S_t, A_t, R_t, S_{t1}, done)训练的时候不按时间顺序取数而是从缓冲区里随机抽取一个小批量。这样做有两个直接好处打破时间相关性随机抽出来的样本来自不同的时间点、不同回合、甚至不同的探索阶段样本之间的相关性大大降低梯度更新更稳定。提高样本利用效率一条经验可以反复被抽到多次用老经验反复训练网络而不是用一次就丢这缓解了强化学习试错成本高、样本量少的核心痛点。再搭配DQN的另一个关键技巧——目标网络Target Network。Q值更新用的目标是r γ * max Q(S, A; θ⁻)其中θ⁻是目标网络参数不随每一步更新每隔固定步数才从当前网络复制过来。这样做的目的是防止同一组数据被用于评估Q值和更新Q值的双重任务时产生自举导致的发散。经验回放和双网络结构是一套组合拳一个解决样本相关性一个解决训练目标不断移动的问题。这就是DQN初期相比普通Q-learning那个表格法的能玩转高维状态空间的关键。2.3 为什么要把经验存到本地断点续训与数据复用理解了缓冲区的作用我们再进入正题为什么要把这些经验落盘保存到本地刚开始跑DQN的时候我以为Replay Buffer就是个内存里的临时仓库内存清空就清空反正训练会产生新的经验。后来吃了大亏训练跑了十几个小时机器突然断电重启所有积累的经验和网络权重全没了只能从头开始跑。那次以后我认清了经验持久化的价值场景不保存经验的结果训练中途崩溃/断电前十几个小时积累的转移样本全部丢失只能重跑调参后重新训练模型需要重新探索如果经验保留可以快速恢复已有能力换一个环境配置测试旧经验不一定适用但至少可以对比新旧分布差异离线深度强化学习/模仿学习必须先采集一批高质量经验存成数据集再离线训练这必须要落盘想复现实验结果没有落盘经验实验的过程数据就丢了很难定位是代码bug还是算法问题实际上现在工业界做强化学习经验落盘是标配。机器人操控、自动驾驶决策、游戏AI这些场景要么试错成本极高要么需要大量真实数据谁都不可能跑一次就完事。把经验以文件形式归档等于给训练过程上了个保险还能把采集和训练彻底解耦——你可以在一个机器上跑环境采集数据另一个机器上跑训练互不影响。这在工程上可以带来很大的灵活性。3. 把DQN训练经验存到本地的完整方案3.1 落地实现第一步定义一个结构化的Replay Buffer这一步很多人会忽略但我在实际开发中意识到这是最重要的基础。你不想在保存的时候才去处理乱七八糟的数据格式应该在设计缓冲区的那一刻就把数据结构定清楚。经验回放里存的是什么是转移元组Transition每一个由5个字段组成。不管你是用namedtuple还是普通dict都行关键是字段要固定、类型要统一。from collections import namedtuple, deque import numpy as np # 定义一条经验的数据结构类似数据库表结构 Transition namedtuple(Transition, (state, action, reward, next_state, done))如果你的状态是二维图像state和next_state的shape就需要保持一致比如[84, 84, 4]。如果状态是一维向量那就是[state_dim]。action 如果是离散动作用整数存连续动作就用数组存。reward 就是浮点数done 是布尔值不过建议存成0或1的浮点数后续计算TD目标更方便。然后实现一个有容量上限的环形缓冲区class ReplayBuffer: def __init__(self, capacity: int): self.capacity capacity self.memory deque(maxlencapacity) # 满了自动淘汰最老的 def push(self, state, action, reward, next_state, done): 存入一条经验 self.memory.append(Transition(state, action, reward, next_state, done)) def sample(self, batch_size: int): 随机采样一个小批量 indices np.random.choice(len(self.memory), batch_size, replaceFalse) transitions [self.memory[i] for i in indices] return Transition(*zip(*transitions)) def __len__(self): return len(self.memory)这里用deque(maxlencapacity)是最省事的做法当容量满了以后新数据进来自动挤掉最老的。它保证缓冲区的经验永远是最近的一段历史老经验被淘汰。不过注意deque的sample如果是随机取的话用np.random.choice配合索引取值是可以的但如果你更在意随机索引的纯随机性可以先把deque转成list再操作。3.2 三种落盘方案选型pickle、numpy压缩、jsongil保存经验到本地我的建议是不要「一招鲜」根据你的数据量和是否需要跨语言/跨环境读数据来选择。方案一直接 pickle 整个缓冲区最简单适合快速断点续训import pickle def save_buffer_pickle(buffer, filepath): # 注意deque不能直接pickle的话转成list再存 with open(filepath, wb) as f: pickle.dump(list(buffer.memory), f) def load_buffer_pickle(filepath): with open(filepath, rb) as f: data pickle.load(f) return data优点实现起来几乎没有心智负担namedtuple的列表直接序列化读回来还是原来的结构字段名自然保留。适合单机训练、自己保存自己恢复的场景。缺点格式是Python私有格式换语言比如你想用C推理引擎读取经验就完全读不了。而且 pickle 有个安全性问题反序列化的时候会执行内部代码如果这个文件被别人篡改过加载的时候有被恶意攻击的风险。自用可以给别人传数据要谨慎。方案二numpy 压缩存取适合图像类状态、追求加载速度这是我最常推荐的方案。结构化保存每个字段单独存成一个数组用np.savez_compressed打包成压缩包def save_buffer_npz(transitions_list, filepath): states np.stack([t.state for t in transitions_list]) next_states np.stack([t.next_state for t in transitions_list]) actions np.array([t.action for t in transitions_list]) rewards np.array([t.reward for t in transitions_list], dtypenp.float32) dones np.array([t.done for t in transitions_list], dtypenp.uint8) np.savez_compressed( filepath, statesstates, next_statesnext_states, actionsactions, rewardsrewards, donesdones ) def load_buffer_npz(filepath): data np.load(filepath) return { states: data[states], next_states: data[next_states], actions: data[actions], rewards: data[rewards], dones: data[dones], }为什么这个方案好savez_compressed会进行gzip压缩如果你的经验是图像通常能压缩得好。加载的时候np.load读取速度远快于 pickle尤其是大文件时候numpy的内存映射机制有优势。数据到了numpy内存里后续你想怎么切分、采样、归一化都很方便。缺点是不保留字段名你看到的是五堆数组必须自己记得对应的顺序和形状。但习惯了之后就没问题。方案三流式写入 JSONL适合边训练边记录、想实时监控JSONLJSON Lines格式每一行是一条JSON记录适合边训练边追加写入import json def append_transition_to_jsonl(filepath, transition): rec { state: transition.state.tolist(), action: int(transition.action), reward: float(transition.reward), next_state: transition.next_state.tolist(), done: bool(transition.done) } with open(filepath, a, encodingutf-8) as f: f.write(json.dumps(rec) \n)好处是你随时可以 grep、tail、用任意文本工具查看出了bug可以肉眼排查某一条具体的经验记录。坏处就一个字慢。如果你的state是84×84×4的图像一转成JSON字符串体积膨胀严重写入也慢到没法接受。所以这个方案我只建议用在低维状态几个浮点数那种、且你有可视化调试需求的场合。三个方案我对不同数据量场景的推荐场景推荐方案原因小规模实验10万条经验低维状态pickle简单直接省事大规模训练图像状态几十万到上百万条经验numpy压缩省空间、加载快、格式结构化调试/日志采集实时观测训练数据JSONL随时可读、可搜索3.3 一个更工程化的方案定期自动快照 容灾备份如果你在跑长时间的训练任务比如几十个小时以上的那种别等训练结束再手动保存经验中间任何一个意外都可能毁掉好几天的工作量。我的习惯是做一个后台线程每隔一定步数自动把buffer快照到磁盘并且保留最近几个历史版本。import os import threading import time class AutoSnapshotManager: def __init__(self, buffer, save_dir, interval_steps5000, keep_last3): self.buffer buffer self.save_dir save_dir self.interval_steps interval_steps self.keep_last keep_last self.last_step 0 os.makedirs(save_dir, exist_okTrue) def maybe_snapshot(self, current_step, model_params, target_params): 每隔interval_steps自动保存一次buffer和模型权重 if current_step - self.last_step self.interval_steps: return self.last_step current_step # 保存buffer buf_path os.path.join(self.save_dir, fbuffer_{current_step}.npz) transitions list(self.buffer.memory) save_buffer_npz(transitions, buf_path) # 保存模型权重 np.savez_compressed( os.path.join(self.save_dir, fmodel_{current_step}.npz), q_netmodel_params, target_nettarget_params ) # 清理旧快照保留最近keep_last份 for old_file in sorted(os.listdir(self.save_dir))[:-self.keep_last]: os.remove(os.path.join(self.save_dir, old_file))这个方案的核心思路是经验保存和模型权重保存绑在一起每次快照都是某一时刻的世界完整状态。恢复的时候你既能恢复模型也能恢复当时的经验池训练无缝衔接。4. 加载经验后的实战处理与常见问题排查4.1 断点续训恢复模型权重 恢复经验池 继续交互保存经验只是手段其实你最后要的是中断了能续上。续训的完整流程通常是这样def resume_training(checkpoint_dir, env, agent, buffer): # 1. 扫描目录里最新的快照文件 latest_model find_latest_file(checkpoint_dir, model_) latest_buffer find_latest_file(checkpoint_dir, buffer_) # 2. 加载模型权重 data np.load(latest_model) agent.q_net.set_weights(data[q_net]) agent.target_net.set_weights(data[target_net]) # 3. 加载经验缓冲区 buf_data load_buffer_npz(latest_buffer) for i in range(len(buf_data[states])): buffer.push( buf_data[states][i], buf_data[actions][i], buf_data[rewards][i], buf_data[next_states][i], buf_data[dones][i] ) # 4. 继续训练但要注意暂时降低探索率 agent.epsilon max(0.05, agent.epsilon * 0.5) # 恢复后少探索多利用这里有个大家容易忽略的点恢复训练后探索率epsilon应该降低。因为好不容易训练出了不错的Q网络如果你还按初始的高探索率随机乱试很容易把已有能力破坏掉。我刚入行的时候犯过这个错加载了模型继续跑半小时之后性能反而比保存前差了很多就是因为epsilon没有衰减为较低值网络被大量随机探索的样本带偏了。4.2 图像类经验的高效存储技巧压缩与降采样如果你的经验是图像状态存到本地时第一件要注意的就是数据体量。算一笔账一个84×84×4的uint8状态是28224字节一条经验至少包含state和next_state也就是56448字节。一百万条经验就是50多个GB的原始数据。不压缩别说存本地内存都可能扛不住。几个实战技巧能大幅降低体积统一用uint8存储游戏画面直接是8位灰度值本来就不需要float32一堆float32就是4倍浪费。保存前归一化到0~1的话存成float16够用但更推荐存原值uint8加载后再转float32并归一化。裁剪到最小有效区域很多游戏画面边缘都是固定不动的背景裁剪掉不影响学习但能大幅减体积。对动作空间做ID映射如果你有几百个可选动作直接存动作IDint16不需要存one-hot矩阵。def prepare_image_state(raw_pixels): # 输入原始分辨率输出84x84灰度uint8堆叠帧 gray cv2.cvtColor(raw_pixels, cv2.COLOR_RGB2GRAY) resized cv2.resize(gray, (84, 84), interpolationcv2.INTER_AREA) return resized.astype(np.uint8)说实话很多强化学习代码把状态在内存里就转成float32这是最浪费的做法。我自己实践下来把所有经验按uint8形式存盘恢复的时候再转成float32省掉的磁盘空间和读取时间都非常可观。4.3 新旧经验分布不一致问题混合采样与优先级控制当你把经验从旧环境加载到新环境里使用的时候会遇到一个隐蔽的问题旧经验的奖励分布、状态分布可能跟当前环境不一样。举个例子你在一个游戏版本A里采集了100万条经验后来环境升级成了版本B物理参数变了。如果你直接把版本A的经验混进版本B的训练里损失函数可能在旧分布上震荡模型越训越糊涂。处理这个问题我建议用两个缓冲区的办法一个缓冲区专门放当前环境的新鲜经验一个放加载进来的旧经验。采样的时候按照一定比例混合class MixedBuffer: def __init__(self, new_buffer, old_buffer, old_ratio0.2): self.new_buffer new_buffer self.old_buffer old_buffer self.old_ratio old_ratio def sample(self, batch_size): n_new int(batch_size * (1 - self.old_ratio)) n_old batch_size - n_new new_batch self.new_buffer.sample(n_new) old_batch self.old_buffer.sample(n_old) return combine(new_batch, old_batch)刚开始旧经验可以占30%随着训练推进逐步降到5%以下然后彻底关闭。这样相当于用旧经验预热模型再用新经验精调平滑过渡比直接把两堆数据混在一起要稳得多。如果你用的是优先经验回放PERPriority Experience Replay另一种方法是加载经验时重新计算TD误差来初始化优先级但那个比较复杂这里不展开你知道有这种思路就行。4.4 盘点我在保存/加载经验时踩过的坑这些年总结下来最有含金量的几个坑分享给大家避雷坑一状态数组的shape不一致导致np.stack崩掉如果你是从多个训练环境收集数据有的环境输出单帧有的输出多帧堆叠形状不一样staack直接炸。解决办法是统一封装一个preprocess_state函数所有环境的数据入口都走这个函数保证输出shape完全一致。坑二buffer容量太大pickle直接撑爆内存我有一段时间懒得想直接把100万条经验的deque序列化成pickle结果保存到一半内存占用翻倍直接OOM。后来学乖了大buffer永远是分批转numpy再存或者是直接流式写npz格式不要一股脑转list再倒腾。坑三保存的done标志类型错误TD目标计算时done会影响r γ * max Q(S, A)是不是要被截断。如果你在保存的时候把done存成了布尔值加载的时候又没转成浮点数一些框架的自动类型转换可能让你在计算TD目标时得到错误结果而且这种错误表现很隐蔽loss不降反升但不会报错。我建议保存done时直接存成np.uint80或1加载后转float32再参与运算。坑四经验文件加载后没有校验直接训练加载完经验之后至少做一次简单校验def validate_buffer(data): assert data[states].shape[0] 0, 经验为空 assert data[states].shape data[next_states].shape, 状态形状不一致 assert data[rewards].dtype in (np.float32, np.float64), 奖励类型错误 print(f经验量: {len(data[states])}) print(f状态shape: {data[states].shape}) print(f奖励范围: [{data[rewards].min():.3f}, {data[rewards].max():.3f}])不要省这一步我有一次加载了损坏的npz文件数据是NaN训练了整整一夜才发现model全部变成NaN白白浪费了一天时间和电费。坑五只存经验不存环境信息加载经验后你还需要知道这些经验是在哪个环境下采集的。下次你换了环境可能经验没有用了但保存文件里没有记录环境名称你根本不知道这批经验对应的环境版本。建议保存的时候顺手在元数据里记录环境ID、奖励归一化参数、观测维度等。meta { env_id: 某环境-v3, reward_scale: 1.0, state_shape: [84, 84, 4], dataset_version: 2024-06-01, }用npz保存的时候可以把这些元数据作为meta.npy一起存进去。说在最后我的个人习惯与实战经验总结玩DQN这几年经验回放和数据持久化给我最大的感悟就是别把训练当成一锤子买卖。好的工程习惯是让每一次训练都变成资产而不是跑完就扔的垃圾。我现在的标准工作流是长这样训练主进程在后台跑每5000步自动做一次快照经验模型权重训练步数快照保留最近3份以防损坏。每天训练结束后我还会把当天的经验文件归档成一个带日期的数据文件为后续调参复用做准备。遇到环境配置改动我不急着从头跑而是先加载旧经验用混合采样的方式让模型快速适应新环境通常能省下大量前期探索时间。说到底DQN和CNN的关系、经验落盘方案都不是什么高深莫测的玄学它们是任何一个想做强化学习工程化的人都会撞上的普普通通的问题。把数据结构设计清楚、把存取机制想明白、把边界情况考虑周全——这些基本功比背多少理论公式都管用。最后再分享一个小技巧如果你真的不确定该选pickle、npz还是jsonl我的建议是先按数据量来筛10万条经验以内随便折腾过了10万条直接无脑走numpy压缩方案图像状态就尽全力压成uint8。你只要实操一次大数据的保存加载就会明白为什么我在文章里反复强调结构化和压缩这两个关键词了。