ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

DQN与CNN的核心区别及强化学习经验回放落盘实战

DQN与CNN的核心区别及强化学习经验回放落盘实战

1. 先搞清楚:DQN和CNN根本不是同一类东西

很多刚接触强化学习的朋友,一看到 DQN 和 CNN 这两个缩写,第一反应都是“这俩肯定是同类技术吧”。实际上这两者的定位差得非常远:CNN(卷积神经网络)是一种网络结构,DQN(Deep Q-Network)是一种完整的强化学习算法框架。拿汽车打个比方,CNN 像是发动机,DQN 像是整台车——发动机是组成车的一个核心部件,但车还有底盘、变速箱、方向盘和一套完整的行驶逻辑。你不可能说“发动机跟车有什么区别”然后只在发动机层面找答案,因为它们的尺度根本不一样。

先看 CNN。它的本职任务是给原始数据做特征提取。输入一张图像或者一段时序信号,通过卷积层、池化层、全连接层的组合,自动从数据里学习到有判别能力的特征,最后输出一个类别概率或者回归值。它的训练方式是典型的监督学习:给你一堆带标签的数据,让网络输出的预测值去逼近标签,通过反向传播更新参数。它有明确的输入——输出映射,学习的目标是“把输入分类正确”或者“把数值估准”。

再看 DQN。它是把深度神经网络当成 Q 函数逼近器来用的强化学习算法。强化学习和监督学习最大的不同在于:监督学习有标准答案,强化学习只有事后反馈——智能体在环境里做一次动作,得到一个奖励信号,但它不知道这个动作在长远意义上到底是对是错。DQN 解决的问题是:在一个状态 S 下,未来所有可能动作里,哪个动作能让长期累积奖励最大。它通过 Q 值来衡量“在这个状态做这个动作有多好”,然后用神经网络来拟合这个 Q 值表。这里的神经网络骨干,可以选择用普通的全连接网络,也可以选择用 CNN。

所以“DQN 和 CNN 有什么区别”这个问题,背后真正的含义有两层:一层是算法框架和网络结构之间的层次差异,另一层是 DQN 这个算法里到底用没用 CNN、用了之后逻辑上怎么衔接。这个问题想清楚了,你后面写代码、调参、保存经验都会顺很多,因为你不会再在概念层面犯“拿网络结构去套算法框架”这种错。

2. 两者之间的真实关系:DQN 的“大脑”里住着 CNN

既然 DQN 是框架、CNN 是零件,那这两个东西是怎么组合起来工作的?DQN 的经典论文(某实验室提出的深度 Q 网络)是用 CNN 处理游戏画面的原始像素帧,把连续 4 帧游戏画面堆叠起来,作为状态输入,CNN 负责从这些像素里提取“PacMan 附近有没有幽灵”“得分趋势怎么样”这类决策特征。CNN 的最后一个全连接层输出的是每个动作的 Q 值估计,DQN 的算法流程负责决定怎么采样、怎么更新、怎么存储经验。

这里要敲黑板:CNN 不是 DQN 的必需品。如果你的环境状态不是图像,而是机器人关节角度、车速、剩余油量这类的低维向量,那中间层用全连接网络做 Q 函数逼近器完全没问题,这个组合依然叫 DQN。换句话说,CNN 只是 DQN 可能在模型架构上选用的一种骨干网络而已。很多新手把“DQN 的输入必须是图像”当成铁律,这其实是个误区。某开发者做机械臂位置控制时,状态是 7 维向量,用了一个三层全连接网络做 Q 逼近,训练收敛速度比用 CNN 还快。

两套体系在训练目标上的差异也很大。CNN 在监督学习里用的是交叉熵损失或者均方损失,输入输出一一对应;DQN 用的是时序差分(TD)损失,它让当前状态的 Q 估计去逼近“真实奖励 + 下一状态的折扣 Q 估计”,这个目标自身会随训练不断漂移。用大白话说,CNN 学的是“看到猫就叫猫”,DQN 学的是“这一步走完,后面还能拿到多少分”,前者是静态映射,后者是动态谋划。

还有一个容易混淆的点:两者的评估方式。CNN 模型好不好,拿一批带标签的数据集算准确率就行;DQN 模型好不好,得把智能体放到真实环境里跑,看累积回报曲线稳不稳、平均回报高不高。DQN 可能在训练过程中某一段看起来 Q 值很高,但放到环境里实际表现拉跨,因为它是基于自举的估计,有偏差累积的风险。这也是为什么理解两者区别很重要——你评估 CNN 的眼光和评估 DQN 的眼光完全不同,用错标准,你会在调参路上走非常多弯路。

3. DQN 的“训练经验”到底是什么,为什么要存

很多教程讲完 DQN 原理就开始写网络代码,但对“经验(experience)”这个东西说得云里雾里。简单说,DQN 在环境里交互一步,就会产生一条记录,这条记录被叫做一条 transition(状态转移样本)。它由四个核心字段构成:

  • 当前状态 S:智能体在这一步开始时看到的观测,可能是图像帧、传感器向量或游戏状态。
  • 动作 A:智能体根据当前策略选出的那个动作,无论是随机探索选出的还是按 Q 值贪心选出的,都会被记录下来。
  • 奖励 R:环境对这一步动作给出的及时反馈,一个标量数值。
  • 下一状态 S':执行动作 A 之后环境转移到的下一个观测。
  • 终止标记 Done:这个状态是不是回合结束点,它直接影响后续的折扣回报计算。

为什么要保存这些经验?因为 DQN 有个核心技术叫经验回放(Experience Replay)。核心思想是:智能体每走一步,先不急着用这条数据更新网络,而是把它丢进一个缓冲区(replay buffer),训练时每次从缓冲区里随机采样一小批(batch)过去存下来的经验来更新网络。

这样做有三个极其重要的原因。第一是打破样本时间关联性。如果智能体在连续帧里做决策,相邻帧的观测有很强的相关性,直接用连续数据更新网络,梯度方向会被带偏,训练震荡剧烈。随机采样相当于把这些强相关样本打散重排,网络更新更稳。第二是提高样本利用率。强化学习里获取一条经验可能很贵,真实机器人转一圈可能花几分钟,游戏里死一命可能要等半天。经验回放让同一条数据可以被复用很多次,数据效率大幅提升。第三是缓解“灾难性遗忘”。加入回放机制前,网络每学完新一批数据,旧场景对应的能力就快速衰减;有了回放池,旧经验时不时被重新采样进训练批次,网络能维持对过往场景的记忆。

正因为经验回放如此重要,“保存训练经验到本地”这件事才变得非常有价值。你训练一个 DQN 智能体,跑了几百万步,回放缓冲区里积累了 50 万条经验,机器内存快不够了;或者你训练到一半要下班,明天想接着练;再或者你从某个公开实现里加载别人训练好的回放数据,想用它的经验导向自己的策略微调——这时候你就需要一个能定期把经验池写到磁盘的工具。

4. DQN 保存训练经验到本地的完整实操方案

我踩过一次特别深刻的坑:训练一个 DQN 跑了一个通宵,第二天起来发现进程被系统 OOM 杀掉,回放缓冲区里的几十万条经验全没了,只能从头再跑。从那之后我再也不敢不落盘经验。下面直接分享几种我自己试过、实测稳定的保存方案,你可以照着抄。

4.1 方案一:用 NumPy 的 np.savez 整体落盘

这是最简单直接的方式,适合经验池不大于内存的场景。假设你的回放缓冲区用一个列表存着所有 transition,每个 transition 的 state、next_state 是固定形状的数组,action、reward、done 是标量。代码可以这么写:

import numpy as np import os def save_replay_buffer(replay_buffer, file_path): """ replay_buffer: list of tuples (s, a, r, s_next, done) s 和 s_next 是 np.ndarray,约定形状一致 """ states = np.stack([t[0] for t in replay_buffer]) actions = np.array([t[1] for t in replay_buffer]) rewards = np.array([t[2] for t in replay_buffer]) next_states = np.stack([t[3] for t in replay_buffer]) dones = np.array([t[4] for t in replay_buffer]) np.savez_compressed( file_path, states=states, actions=actions, rewards=rewards, next_states=next_states, dones=dones ) print(f"已保存 {len(replay_buffer)} 条经验到 {file_path}")

加载代码对应的实现:

def load_replay_buffer(file_path): data = np.load(file_path, allow_pickle=True) states = data['states'] actions = data['actions'] rewards = data['rewards'] next_states = data['next_states'] dones = data['dones'] buffer_len = len(states) replay_buffer = [] for i in range(buffer_len): replay_buffer.append(( states[i], int(actions[i]), float(rewards[i]), next_states[i], bool(dones[i]) )) print(f"已加载 {buffer_len} 条经验") return replay_buffer

使用np.savez_compressed而不是普通 savez,是因为压缩能大幅减少磁盘占用。图像类经验数据往往动辄几个 GB,压缩率通常在 2 到 5 倍之间,代价仅仅是保存时多花一点点 CPU 时间,训练阶段的时间瓶颈远大于这个开销,整体划算。

这里有几个要特别注意的细节:

  • 如果 state 不是标准数组,而是 Python 对象(比如字典结构或变长数组),需要在栈之前先做规整,或者使用 dtype=object 强行保存,但那样加载时就会很别扭。
  • 所有 transition 的 state 形状必须完全一致。如果你的环境下有一种状态小于其他状态,stack 会直接报错。我是经常在保存前加一行断言:assert all(s.shape == states[0].shape for s in states),提前发现问题。
  • np.load 默认开启了 allow_pickle,但实际上它是不建议开启的,因为存在代码执行风险。好在我们的数据没有 pickle 对象,直接用默认的空对象加载即可,不需要写 allow_pickle=True。

4.2 方案二:用 pickle 直接持久化经验池对象

如果你的回放缓冲区不是简单的列表,而是自定义的 ReplayBuffer 类,并且类里维护了一些辅助数据结构(比如优先经验回放里的 SumTree 数组、或者采样索引的映射关系),那用 pickle 是最接近“一键保存”的方案:

import pickle class ReplayBuffer: def __init__(self, capacity): self.capacity = capacity self.buffer = [] self.position = 0 def push(self, transition): if len(self.buffer) < self.capacity: self.buffer.append(transition) else: self.buffer[self.position] = transition self.position = (self.position + 1) % self.capacity def save(self, file_path): with open(file_path, 'wb') as f: pickle.dump(self.__dict__, f) print(f"经验池状态已保存到 {file_path}") def load(self, file_path): with open(file_path, 'rb') as f: saved_dict = pickle.load(f) self.__dict__.update(saved_dict) print(f"已从 {file_path} 恢复经验池,当前容量 {len(self.buffer)}")

保存的时候直接 dump 实例的dict,这样做的好处是如果你以后给 ReplayBuffer 增加新属性,比如优先级索引、时间戳,pickle 能自动把新字段带上,不需要修改代码。缺点也很明显:pickle 格式是 Python 专属且不可读,跨版本还原可能出问题,而且如果有恶意构造的 pickle 文件,加载时会执行任意代码。只在自己的训练流程里使用,不拿别人的 pickle 文件乱加载,这套方案完全没有问题。

我自己的经验是,在需要频繁断点续训的实验流程里,我用 pickle 保存整个 ReplayBuffer 对象,同时把网络权重单独存一份 PyTorch state_dict。两个文件都带上训练步数作为后缀,比如buffer_300000.pkl和net_300000.params。恢复之后从 300000 步继续训练,曲线衔接得很顺。

4.3 方案三:增量落盘,应对超大规模经验池

训练一个需要吃高分辨率画面的 DQN,经验池容量设计成 100 万条,每条经验里有 4 帧 84x84 RGB 图像,算下来内存占用非常夸张。这时候一次性 np.savez 整个经验池会让程序卡死几秒,甚至内存峰值翻倍导致 OOM。增量落盘方案是真正面向大规模场景的做法。

思路是按时间片切分经验文件,每积攒 N 条经验就把这批单独写一个文件,文件名带上起始序号:

import numpy as np import os class DiskBackedReplayBuffer: def __init__(self, save_dir, chunk_size=50000): self.save_dir = save_dir self.chunk_size = chunk_size self.current_chunk = [] self.global_id = 0 os.makedirs(save_dir, exist_ok=True) def push_and_maybe_flush(self, transition): self.current_chunk.append(transition) self.global_id += 1 if len(self.current_chunk) >= self.chunk_size: self.flush() def flush(self): if not self.current_chunk: return chunk_id_start = self.global_id - len(self.current_chunk) chunk_id_end = self.global_id file_path = os.path.join( self.save_dir, f"replay_{chunk_id_start}_{chunk_id_end}.npz" ) data = { 'states': np.stack([t[0] for t in self.current_chunk]), 'actions': np.array([t[1] for t in self.current_chunk]), 'rewards': np.array([t[2] for t in self.current_chunk]), 'next_states': np.stack([t[3] for t in self.current_chunk]), 'dones': np.array([t[4] for t in self.current_chunk]), } np.savez_compressed(file_path, **data) self.current_chunk = [] print(f"已落盘 {file_path}")

训练时如果需要从这些分片文件里采样,可以先建立一个文件索引,随机选一个分片文件,读取进去,再在分片内随机采样。这个方案天然解决了“回放缓冲区无限大”的问题——你可以把最旧的批次逐步丢弃或者保留在磁盘归档,让内存里始终只保留最近的一部分经验。选择 chunk_size 时要注意 IO 与内存的平衡,我常用的经验值是:经验条目的状态是 84x84x4 图像时,每个分片 5 万条大约占磁盘 150 MB(压缩后),训练机读写这个量级的文件耗时在半秒以内,完全不影响训练节奏。

4.4 加载经验后继续训练的正确姿势

保存只是万里长征第一步,加载回来接着训才是目的。我在恢复训练时总结了一个固定套路,你照着做基本不会出错。

第一步,先加载网络权重文件,常见做法是用 PyTorch 的torch.save和torch.load。注意要兼顾 optimizer 的状态:如果只是保存了 model.state_dict(),恢复后学习率调度器会从头开始,训练曲线会有一段适应期。正确的做法是把 model、optimizer、学习率调度器、当前 step 数全部打包进一个 dict 存下来。

第二步,加载经验池。如果是用方案一保存的 npz 文件,加载后要把经验池转换成你的 DQN 代码内部格式。不同代码结构里 transition 的字段顺序可能不同(有的代码用字典,有的用 NamedTuple),转换时一定要做字段名对齐检查。我之前接过一个开源项目,它的 transition 顺序是 (s, a, r, s_next, done),另一个项目是 (s, s_next, a, r, done),加载完直接开始训练,Q 值疯狂震荡,排查了一下午才发现是字段顺序错位。

第三步,最关键的一步:验证加载后的经验池能不能正常采样。在训练主循环开始之前,跑 100 次采样,检查采样出的 batch 维度、数据类型和你的网络输入是否匹配。有个老掉牙但极度常见的坑:图像类状态保存时没有除以 255,加载后再跑经验池采样,网络输入分布跟原来完全不同,直接导致策略性能断崖式下跌。如果你在保存前把像素做了归一化,加载回来的推理代码里也要保持同样的归一化逻辑。

# 验证经验池加载是否正确的脚本片段 sample_batch = sample_from_buffer(loaded_buffer, batch_size=32) states_batch = sample_batch['states'] assert states_batch.shape[1:] == expected_state_shape, \ f"形状不匹配: {states_batch.shape[1:]} vs {expected_state_shape}" assert states_batch.dtype == torch.float32, f"数据类型异常: {states_batch.dtype}"

5. 常见坑和避坑技巧实录

保存训练经验这件事,表面上只是文件读写,实际操作里坑非常多。我把这几年见过、踩过的坑集中列出来,每条都有对应的排查思路。

坑一:保存经验时和训练线程抢占大量内存导致 OOM

原因:训练时图像状态累积在高频操作,内存本身接近临界值,你突然用 np.stack 把整个经验池复制了一遍,内存直接爆掉。解决办法是避免一次性全量 stack。可以分批遍历存入内存映射文件,或者用增量落盘方案,在达到内存阈值前持续分批写出。另外一个技巧是及时释放引用,写成states = None之后调用gc.collect(),对大型数组很有用。

坑二:reward 和 next_state 混在没有归一化就加载

有个常见流程是训练前把奖励裁剪到 [-1, 1],加载经验池之后忘记重新裁剪,导致新旧奖励尺度不一致。这里建议把归一化操作写进经验池的加载函数里,而不是依赖训练脚本每次手动处理。这也是“加载即就绪”原则的体现。

坑三:pickle 文件版本不兼容

Python 3.8 保存的 pickle,拿到 Python 3.11 环境里偶尔能加载,但和某些 C 扩展类型(比如 PyTorch Tensor)组合时容易报 numpy 版本相关的错误。规避办法是给保存文件带上环境标识,或者在项目 README 里写清楚训练环境的 Python、NumPy 版本。我试过最省心的方式是统一用 npz 格式存核心数据,用 pickle 只存辅助配置,这样版本迁移时核心数据永远可靠。

坑四:GPU 上训练时经验池数据留在 CPU 内存,但保存时类型不对

有些经验池实现把 state 存成 torch.Tensor 的 GPU 版本,保存时直接把 GPU tensor 序列化。这非常危险,加载时如果显存不足或者机器没有 GPU,文件就废了。正确做法是保存之前统一用.cpu().numpy()转成 CPU 数组,加载后再按需转 GPU。

坑五:分片文件的训练步数编号和实际训练步数对不上

增量落盘里 count 的推进如果写错,你训练 100 万步但磁盘上一个回放文件只覆盖到 90 万步。建议不要在多个地方各自维护步数计数器,而是直接把回放缓冲区自己的 push 次数作为唯一编号来源,这样所有分片文件的步数连续性有保障。

6. 从保存经验到更远的强化学习工作流

DQN 和 CNN 的区别讲清楚之后,你会发现一个更通用的道理:理解一个算法,要区分“策略层面”和“实现层面”。CNN 是实现层面的零件,DQN 是策略层面的框架。保存训练经验这件事也一样,你是在为整个“连续训练”的策略做基建,而不是简单写一个文件归档工具。

我目前的做法已经演变成一套完整的工作流:经验池自动分片落盘,网络权重每 5 万步保存一次 checkpoint,同时把一份最近 1 万步的“热经验”单独放在内存里做快速采样,其余经验全部归档到磁盘。恢复到某个步数时,我用一个工具脚本把磁盘分片重新组装成可采样的统一经验池,先做验证再做训练。这套体系支撑我做各种实验对比时特别省心,不用怕进程被杀,不用怕断点丢失。

另外一个可扩展的点是:经验数据实际上是宝贵的研究资产。DQN 跑出来的经验包含“在什么状态下尝试了什么动作、获得了多少回报”,这完全可以用来做离线强化学习的输入数据。把经验保存规范做好,将来做策略独立性验证、做奖励模型学习、做行为克隆对比,都能源源不断地复用旧数据。你现在多花一点时间把 IO 模块写得整洁,后面做研究能省出来的时间远超想象。

最后分享一个小技巧:无论是 DQN 还是后续要用的各种强化学习算法,我都推荐在模型类里封装一个save_checkpoint(full=False)方法,full=True 时连经验池一起保存,full=False 时只存网络权重。训练脚本里用定时调用来决定保存频率,这样既不会因为频繁全量保存拖慢训练,也保证关键节点有完整快照可以恢复。这套设计我沿用到了很多项目里,是性价比很高的一笔投入。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进