做图像分类训练,最烦的事不是调参,而是训练结束之后你想画验证集的混淆矩阵、ROC曲线、各类别精确率召回率,结果打开代码发现,你手里只有fit_generator返回的history,里面全是每个epoch的loss和accuracy标量,验证集上每一个样本的真实标签y_true和模型预测y_preds,一样都没留。
这个需求太常见了。只要你在训练结束之后想做细粒度评估,就绕不开“从验证集里取出模型预测”这一步。但Keras的fit_generator处理验证集时,默认逻辑就是“喂进去、算个平均loss、返回标量”,中间任何批次的预测输出都不会暴露给外部。所以很多人只能网上到处搜“怎么在fit_generator里拿y_true和y_preds”,收获的还都是零散片段。
这篇文章我基于自己做分类任务时的踩坑过程,把几种可行的取数思路全部拆开讲清楚:为什么验证循环不给你预测结果、验证集顺序怎么保证不乱、训练后predict怎么用最稳妥、每个epoch跟踪预测用什么方式、拿到y_true和y_preds之后如何去算混淆矩阵/AUC/top-k准确率。无论你用的是老的fit_generator还是新版fit直接传生成器,思路完全通用。
1. 验证循环为什么不把y_preds还给你
1.1 验证阶段在内部到底做了什么
先弄清楚一件事:fit_generator对验证集的处理从设计上看就是一个“评分过程”。它在每个epoch结束后,把validation_data的批次逐一取出来,做一次前向推理,把每个批次的loss累积起来,最后除以批次数得到平均验证loss,再交给metrics计算标量指标。至于每个样例的softmax输出,验证循环算完之后直接丢弃,连临时变量都不保留。
这和训练过程是一个道理——训练时你也不会拿到每个batch的实时预测概率,因为计算图的目的是更新权重并累积损失,不是为了给你留预测结果。所以“fit_generator内部拿不到y_preds”不是某个参数没开,而是API设计就没打算把这一步还给你。你可以把验证循环理解为:把原料送进加工厂,最后给你一张“合格率统计表”,中间每一批次的具体产出从来不返回。
搞清楚这一点之后,你就不会再试图在fit_generator的参数里找开关,而是老老实实走外部路子。实操里我认为靠谱的只有三条:
- 训练结束后对验证集重新做一次
model.predict。 - 写一个自定义Callback,在训练过程中自己收集每个epoch的预测。
- 从生成器源头把y_true“留底”,做一个包装生成器。
下文就是这三条路的完整展开。
1.2 验证生成器的顺序问题,才是真正的拦路虎
为什么获取y_true和y_preds最怕的问题不是长度,而是顺序?
如果你用flow_from_directory生成验证集,并且设置了shuffle=True,那每个epoch里验证样本的顺序都会被打乱一次。某个epoch结束时,模型在验证集第5个位置预测到的样本,和你用另一个生成器取到的第5个y_true,根本不是同一个样本。就算你把seed固定住,洗牌结果确实能复现,但你并不知道洗完之后哪个样本去了哪里,依然没法对齐。
所以在准备“用于事后评估”的验证生成器时,有一条铁律:shuffle永远设为False。顺序固定成“目录遍历顺序”,也就是class_indices按名称排序后,每个类下面的文件依次排开。这样无论你从生成器里取几次,第i个样本永远是同一个样本。
另外,ImageDataGenerator里那些数据增强参数,在验证生成器上严格来说一个都不要开。我见过有人图省事,直接用train_datagen给验证集做rescale,顺带开了rotation_range和horizontal_flip。训练时验证集的平均准确率还算正常,可一旦你直接对相同生成器做predict拿预测概率去算ROC,结果会抖动得很厉害——因为每次预测用的是随机增强后的图像,模型对同一样本在不同时刻的预测差异很大,评估曲线全被污染。
1.3 三条路的核心取舍
这三条路不是非此即彼,完全可以组合。
训练后predict:代码最少,思路最直白,适合只关心最终模型表现的人。缺点是它只给你“最终状态”的预测,看不到训练过程中预测分布怎么变化。
自定义Callback:能拿到每个epoch的y_preds,适合做过程诊断。缺点是每个epoch都要对验证集多跑一次推理,训练时间会被拉长。
生成器闭包缓存:适合不想额外predict、想让fit自己消费验证数据、又想把标签留底的人。缺点是它本质上只能给“最后一次遍历”的标签,想要epoch对应关系还得额外设计。
我个人的建议是:先做训练后predict,简单可靠;有过程分析需求再加Callback。生成器闭包这种偏hacky的方案,放到框架把你限制死了再用。
2. 最直接的方案:训练后predict整个验证生成器
2.1 准备一个顺序稳定的val_generator并预取y_true
如果你只想要最终模型的验证集评估结果,最简单的方法就是:训练完之后,用model.predict跑一遍验证集。困难点在于怎么确保predict出来的y_preds顺序,和你手里的y_true完全一致。
我的做法是从一开始就单独构建一个只用于评估的验证生成器:
import os import numpy as np import tensorflow as tf IMG_SIZE = (224, 224) BATCH_SIZE = 32 val_datagen = tf.keras.preprocessing.image.ImageDataGenerator(rescale=1. / 255) # 注意:shuffle=False,顺序固定,这是后续一切对齐的前提 val_gen_eval = val_datagen.flow_from_directory( os.path.join(DATA_DIR, 'val'), target_size=IMG_SIZE, batch_size=BATCH_SIZE, class_mode='categorical', shuffle=False, seed=42, ) # 一行取出全部真实标签 y_true = np.concatenate( [val_gen_eval[i][1] for i in range(len(val_gen_eval))], axis=0 ) print('y_true shape:', y_true.shape) print('class mapping:', val_gen_eval.class_indices)这里有两个细节值得说。val_gen_eval[i]访问的是第i个批次,返回一个(x_batch, y_batch)的二元组,y_batch就是该批次的one-hot标签。len(val_gen_eval)等于样本总数除以batch_size后向上取整,最后一个批次样本数不满batch_size也没关系,concatenate之后总行数就是验证集样本总数。
class_indices就是标签到类名的映射,顺序由目录名的字母序决定。模型的输出层如果设置了num_classes,它的类别顺序和这个映射是对应的。评估完成后我习惯把这个映射存一份,方便后面写报告时把类别索引翻译回真实名字。
2.2 训练完之后对齐y_true与y_preds
接着正常训练。新版Keras直接model.fit传生成器即可,老代码里的fit_generator只是历史API,内部逻辑一致:
model.fit( train_gen, steps_per_epoch=len(train_gen), epochs=50, validation_data=val_gen_eval, validation_steps=len(val_gen_eval), )训练结束之后,直接predict:
y_preds = model.predict(val_gen_eval, verbose=1) # 转为类别标签 y_pred_labels = np.argmax(y_preds, axis=-1) y_true_labels = np.argmax(y_true, axis=-1) print('y_preds shape:', y_preds.shape) # (N, num_classes) print('y_true shape:', y_true.shape) # (N, num_classes)这里的y_preds是模型最后一层的输出,对于分类模型通常就是softmax概率分布。为什么强调这一点?如果你在自定义模型里把最后的softmax去掉了,输出的是logits,那这里的“预测概率”只是理解上的近似,计算AUC之前需要自己补一个softmax转换。
预测长度和标签长度正常情况下是相等的。如果发现不等,优先检查你是不是对同一个生成器先做了fit又做了predict——生成器的内部迭代器状态在fit过程中会被消费,predict时会从当前指针位置继续读取,导致只预测了后半段样本,输出行数远远小于y_true。这个问题很隐蔽,我踩过一次。解决方式很简单:不要复用同一个生成器对象,用两个都shuffle=False且seed相同的独立实例,一个给fit,一个给predict。
2.3 为什么len(val_gen_eval)和steps要小心处理
很多教程会建议predict时写steps=len(val_gen_eval),本质上等于“你走完全部批次”。不传这个参数时,Keras内部也会默认走完整个生成器。所以预测阶段不要随手把steps调小,比如为了加速只跑一小部分验证集,那得到的结果行数就不等于y_true行数了,后面的混淆矩阵全是错位。
反过来,如果你数据集特别大,确实想用部分批次做快速验证,那一开始就把y_true也截短到同样范围,保持“预测多少就取多少标签”的原则。但我不推荐在最终评估阶段这么干,子集验证请用专门的子集目录,别在跑批时截断。
还有一个容易忽略的小坑:predict时如果验证集样本数恰好能被batch_size整除,一切正常;如果不能整除,最后一个batch只有少量样本,predict依然会把它们输出。个别很老版本的Keras对最后一个不满批次处理不好,会出现维度报错,这时建议把验证集样本数调整成batch_size的整数倍,或者直接升级框架版本。我在现行版本里没有遇到过,但网上见过反馈,这里提一句预防。
3. 想在训练过程中跟踪预测:自定义Callback存下每个epoch的y_preds
3.1 on_epoch_end里如何安全地predict验证集
单次预测只解决最终评估。如果你想知道模型训练到第5个epoch时对难例样本的预测变化,或者绘制PR曲线随epoch演变的动态图,就需要在每个epoch结束后拿到一次验证集预测值。这时候自定义Callback是最合适的手段。
我常用的一种写法是:在Callback初始化时就用一个全新的验证生成器把y_true固定住,然后在on_epoch_end里predict一次并保存:
import numpy as np import tensorflow as tf class ValPredsCallback(tf.keras.callbacks.Callback): def __init__(self, val_gen, save_every=1): super().__init__() # 在训练开始前取y_true,此时的生成器完全干净,指针未移动 self.y_true = np.concatenate( [val_gen[i][1] for i in range(len(val_gen))], axis=0 ) self.preds_history = {} self.save_every = save_every self.latest_preds = None def on_epoch_end(self, epoch, logs=None): if (epoch + 1) % self.save_every != 0: return preds = self.model.predict(self.val_gen, verbose=0) self.preds_history[epoch] = preds self.latest_preds = preds然后挂载:
val_gen_for_cb = val_datagen.flow_from_directory( os.path.join(DATA_DIR, 'val'), target_size=IMG_SIZE, batch_size=BATCH_SIZE, class_mode='categorical', shuffle=False, seed=42, ) val_gen_for_fit = val_datagen.flow_from_directory( os.path.join(DATA_DIR, 'val'), target_size=IMG_SIZE, batch_size=BATCH_SIZE, class_mode='categorical', shuffle=False, seed=42, ) model.fit( train_gen, steps_per_epoch=len(train_gen), epochs=50, validation_data=val_gen_for_fit, validation_steps=len(val_gen_for_fit), callbacks=[ValPredsCallback(val_gen_for_cb, save_every=5)], )注意这里我没有把val_gen_for_cb和给fit用的validation_data弄成同一个对象。这是一条重要经验:fit和Callback两个地方都会消费验证生成器,如果共享一个对象,迭代状态会互相干扰。虽然迭代器支持索引和重置,但为了彻底避免诡异行为,我给它们各自准备了独立实例。两个实例都用同样的shuffle=False和seed,出来的批次顺序严格一致。
3.2 把预测结果按epoch落盘
如果一个epoch保存一次预测,50个epoch就是50份(N, num_classes)的数组,内存里可能放着近50倍验证集大小的float数据,占的内存不小。所以我通常不把所有epoch的预测全部留在内存里,而是在on_epoch_end里直接存npy文件,只保留一个“最新预测”用来画曲线:
import os np.save(f'val_preds_epoch_{epoch+1:03d}.npy', preds) # 只保留最近一次 self.latest_preds = preds这样既不会内存爆炸,也能在训练中断时从磁盘恢复历史预测。文件命名里带上epoch,之后做横轴对比时直接按文件名索引。
如果只想画动态曲线,还有一个更省内存的办法:在每个epoch结束时不存完整预测,只计算你关心的指标,比如每个类别的AUC、top-k准确率,把指标数值写进回调里的字典,训练结束后再统一画图。这样不需要维护几十份预测数组,内存压力小很多。
3.3 注意时间开销与显存抖动
这个方案的缺点非常现实:每个epoch结束都要对完整验证集做一次predict。如果验证集有2万张图,训练60个epoch,等于额外多跑了60次完整验证推理。有些网络模型推理很快还扛得住,但像VGG这种前向很重的模型,二次开销就很明显。
我有一次任务就是因为这个原因,本来一个epoch只要40秒,挂上Callback之后直接变成65秒。后来学乖了,给Callback加了个save_every参数,每5个epoch才做一次验证预测。代价是每个epoch的预测曲线没那么细,但整体训练时间回到正常范围,画出来的5步间隔曲线已经足够看出模型是否过拟合。
显存方面也要提一下:predict并不会清掉训练时已经加载的完整计算图,而是复用同一个会话,所以验证预测的显存额外占用主要是推理时的激活值缓存。如果你的模型本身已经把显存占得接近上限,predict时有概率触发OOM。稳妥做法是给Callback里的验证生成器单独设置一个更小的batch_size,反正预测结果行数只跟样本数有关,跟batch大小无关。
4. 用生成器闭包缓存y_true:一种更底层的兜底方案
4.1 为什么还需要第三种方案
看到这里你可能会想:前两种方案已经够用了,第三种是干嘛的?
有些场景你还真得用第三种。比如你不想反复predict,只想让fit的验证循环自然把验证集“消费”一遍,同时把每个批次的真实标签留底;又或者你的模型训练逻辑被包在一套公共框架里,你想在不改动框架调用方式的前提下拿到标签。这时在生成器层面做包装,是最不侵入的方案。
实现思路很简单:在生成器的迭代循环里,顺手把每个batch的y_true复制一份缓存到外部列表。
class CachingValGenerator: def __init__(self, keras_gen): self._gen = keras_gen self.y_true_batches = [] def __len__(self): return len(self._gen) def __iter__(self): self.y_true_batches = [] for x, y in self._gen: # 存一份y,防止原数据被后续修改 self.y_true_batches.append(np.array(y, copy=True)) yield x, y cache_gen = CachingValGenerator( val_datagen.flow_from_directory( os.path.join(DATA_DIR, 'val'), target_size=IMG_SIZE, batch_size=BATCH_SIZE, class_mode='categorical', shuffle=False, seed=42, ) ) model.fit( train_gen, steps_per_epoch=len(train_gen), epochs=50, validation_data=cache_gen, validation_steps=len(cache_gen), ) y_true = np.concatenate(cache_gen.y_true_batches, axis=0)这个类的逻辑就是:每次被迭代时,重置内部列表,然后把生成器吐出来的x_batch和y_batch原样转发,只是多复制了一份y。fit的验证循环会自动调用它的__iter__和__len__,所以不影响训练流程本身。
4.2 缓存数据的重复累积问题
这个方案最大的坑在于:fit_generator对验证集是每个epoch都会迭代一次的,所以__iter__在每个epoch都会被触发,而每次触发都会把y_true_batches清空重写。最终你拿到的缓存,只对应最后一次验证遍历,不一定就是你想要的某个epoch。
如果你的验证集shuffle=False,那么每一次遍历的顺序都相同,最终缓存内容和任意一个epoch的验证顺序都能对上,问题不大。无非是你要记住“这个y_true是最后一次遍历时的顺序”。但如果shuffle=True,每个epoch的验证顺序不同,这个方案基本没法用,因为你不知道最后一次遍历的具体洗牌顺序,拿到缓存也没法和某个epoch的预测对齐。
所以我再次强调:凡是涉及“拿验证集做事后评估”的需求,验证生成器一律shuffle=False。这不是代码洁癖,而是数据对齐的必要前提。
另外,缓存的数据是每个epoch累加还是只缓存最后一个完整遍历,取决于你定义__iter__时清空列表的位置。我这里是在__iter__开头清空,保证每次完整遍历只留下一份快照。如果你希望把所有epoch的标签全部缓存下来,以观察不同epoch之间的标签顺序差异,那就不清空、直接append,但那样数据量会膨胀,而且通常没什么用。
4.3 什么情况下才值得用这个方案
诚实地说,这个方案的实际使用频率不高。大多数情况下,训练后predict的方案简单又可靠,代码量也更少。包装生成器这个方式,更适合下面两类场景:
一类是内部框架封装很死的情况。比如某训练平台只允许你传入一个生成器并调用统一fit函数,你没机会在fit之后补一个predict。这时用包装生成器,就可以在不改平台代码的前提下,把标签偷出来。
另一类是你在写示例代码、希望向别人展示“如何在数据流层面对齐标签”的场景。从生成器源头缓存y_true,比在模型层面做预测再回头找标签,逻辑上更直观——标签本来就是在数据生成阶段产生的。
如果你不属于这两种情况,建议直接跳过这一节,用前面的方案就行。技术方案不是越多越好,适合当前代码结构才是关键。
5. 拿到y_true和y_preds之后:混淆矩阵、AUC与top-k准确率
5.1 标签编码对齐:categorical、binary、sparse三种情况
终于拿到两个数组了,但别急着算指标,先确认标签编码方式。因为不同class_mode下y_true的形状完全不同,处理方式也不同。
| class_mode | y_true形状 | 解释 | 转换方法 |
|---|---|---|---|
| categorical | (N, num_classes) | one-hot向量 | np.argmax(y_true, axis=-1) |
| binary | (N, 1) | 0/1标量 | y_true.flatten() |
| sparse | (N,) | 整数类别索引 | 直接用 |
y_preds的形状在单输出分类模型里始终是(N, num_classes),代表每个类别的概率。所以你只要把y_true统一成整数类别索引,和np.argmax(y_preds, axis=-1)比对即可。
这里最容易翻车的细节是binary模式。模型输出层如果是sigmoid单输出,y_preds形状是(N, 1),和categorical模式的(N, num_classes)相比,你要先flatten成(N,),再去计算二分类AUC或混淆矩阵。很多报错“Found array with dim 3. Expected <= 2”就出现在这里,不是模型问题,是数组维度没对齐。
5.2 一份可直接套用的评估脚本
我一般在拿到y_true和y_preds之后,直接跑这个脚本:
from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score y_true_cls = np.argmax(y_true, axis=-1) y_pred_cls = np.argmax(y_preds, axis=-1) # 分类报告:精确率、召回率、F1 class_names = list(val_gen_eval.class_indices.keys()) print(classification_report(y_true_cls, y_pred_cls, target_names=class_names)) # 混淆矩阵 cm = confusion_matrix(y_true_cls, y_pred_cls) print('Confusion matrix:\n', cm) # 多分类AUC,需要概率输出 auc = roc_auc_score(y_true, y_preds, multi_class='ovo', average='macro') print('Macro OVO AUC: {:.4f}'.format(auc))如果类别特别多,混淆矩阵直接打印会很长,我会改用seaborn画成热力图,矩阵作为数据传入。这里不展开画图代码,你拿到cm之后想怎么画都行。
5.3 结果对不上时先查这三个地方
如果你发现y_true和y_preds的行数都对,但指标异常地低,或者混淆矩阵里预测标签集中在某几个类,先别怀疑模型出了bug,按下面清单排查。
第一,验证生成器是否在predict时还开着数据增强。如果val_datagen里混进了随机旋转、翻转,不同时刻predict同一张图会得到不同结果,指标波动很大。验证集只用rescale,不做任何增强。
第二,类别顺序是否一致。模型输出层的类别顺序来自训练生成器的class_indices,验证生成器如果目录结构或目录名排序不一致,class_indices会不同。打印两个生成器的class_indices,肉眼对比一下,尤其是中文目录名或带数字前缀的目录名——比如“class1”“class10”,字符串排序是class1在class10前面,你很容易误以为顺序对。
第三,是否复用了被fit消费过的生成器做predict。生成器指针位置问题会导致预测行数不足或者顺序漂移。规则很简单:每个用途一个独立生成器,shuffle=False,seed相同。
这三点我每次排查都能用上,基本覆盖了“y_true和y_preds能跑但对不上”的常见情况。
5.4 top-k准确率与预测结果落盘
分类任务里除了单标签准确率,我还经常算top-k准确率。比如ImageNet这类大类别任务,模型给出的前5个预测里如果包含真实类别,也算预测正确。这个指标在评估阶段很容易被忽略,但对业务场景非常有用:推荐系统里用户不一定选排名第一的候选,只要推荐结果在前几名里就有价值。
def top_k_accuracy(y_true_cls, y_preds, k=3): top_k_preds = np.argsort(y_preds, axis=-1)[:, -k:][:, ::-1] hits = np.array([ y_true_cls[i] in top_k_preds[i] for i in range(len(y_true_cls)) ]) return hits.mean() print('Top-3 accuracy: {:.4f}'.format(top_k_accuracy(y_true_cls, y_preds, k=3)))把预测结果整份保存下来也很有必要。有时训练完模型,评估脚本是临时写的,不想为了重算一个指标再跑一遍整个验证集的预测。我会直接np.save('val_y_preds.npy', y_preds),连同当时的y_true一起存,之后随便怎么分析都行。
6. 三种取数方案怎么选:一张对比表和一个实战建议
6.1 对比表与选择标准
三种方案放在一起,差异主要体现在“拿到什么”、“何时拿”、“额外开销”三个维度。
| 对比项 | 训练后predict | 自定义Callback | 生成器闭包缓存 |
|---|---|---|---|
| 拿到什么 | 最终模型的y_preds | 每个epoch的y_preds | 验证遍历时的y_true |
| 何时取 | 训练结束后一次 | 每个epoch自动 | fit验证循环内自动 |
| 是否多花推理时间 | 多一次完整预测 | 每个epoch多一次预测 | 无额外计算 |
| 顺序对齐风险 | 需shuffle=False | 需shuffle=False | 需shuffle=False与快照 |
| 典型使用场景 | 最终评估报告 | 训练过程诊断 | 框架封装太死时兜底 |
| 代码量 | 少 | 中 | 中 |
选择标准很简单:只做最终评估,选训练后predict;想看训练过程的预测演变,选Callback;不方便在训后额外预测,选生成器闭包。三个方案不是互斥的,我个人现在常用组合是“训练后predict做最终评估 + 隔N个epoch的Callback做过程监控”。
6.2 我的推荐组合与完整代码骨架
这里给一个我自己的组合模板,你在自己的任务里把路径和模型结构替换掉就能用:
train_gen = train_datagen.flow_from_directory( train_dir, target_size=IMG_SIZE, batch_size=BATCH_SIZE, class_mode='categorical', shuffle=True, seed=42, ) val_gen_for_fit = val_datagen.flow_from_directory( val_dir, target_size=IMG_SIZE, batch_size=BATCH_SIZE, class_mode='categorical', shuffle=False, seed=42, ) val_gen_for_pred = val_datagen.flow_from_directory( val_dir, target_size=IMG_SIZE, batch_size=BATCH_SIZE, class_mode='categorical', shuffle=False, seed=42, ) cb = ValPredsCallback(val_gen_for_pred, save_every=5) model.fit( train_gen, steps_per_epoch=len(train_gen), epochs=50, validation_data=val_gen_for_fit, validation_steps=len(val_gen_for_fit), callbacks=[cb], ) # 训练后最终预测,直接用Callback里提前取好的y_true y_true = cb.y_true y_preds = model.predict(val_gen_for_pred, verbose=1) print(y_true.shape, y_preds.shape)注意,Callback里的y_true必须在fit开始前就从val_gen_for_pred取好,这样最终评估得到的y_true与predict的生成器顺序一致;fit里的val_gen_for_fit只是给训练过程算loss用。这两个生成器虽然内容完全一样,但互不干扰。
6.3 常见错误汇总与最终提醒
最后把最容易出错的几个点再总结一遍,希望你少走弯路:
- 验证生成器永远shuffle=False。这是所有方案的前提,没有这个前提,后面所有对齐都是空谈。
- 预测阶段不要用带增强的datagen。验证集只用标准化,不用旋转翻转缩放。
- 不要复用一个生成器对象同时服务fit和predict。各自实例化,seed相同,顺序才能保证一致。
- predict时别乱设steps,走完整个验证集,行数才能和y_true对齐。
- 标签编码先确认,binary模式和categorical模式的形状不一样,直接argmax会出问题。
- 每个epoch存预测之前,想清楚内存能不能扛住,最好按epoch落盘。
这类问题的本质其实不在Keras的API,而在于你对训练数据流水线的理解:什么时候数据被消费、顺序被谁决定、标签在哪个环节产生。把这些点想透,不管以后换PyTorch还是其它框架,你都能很快定位到“取预测结果”应该在哪一个环节下手。我自己也是踩了几次坑才彻底理顺,希望这篇文章能帮你省下这个时间。