Keras多输入模型数据加载指南：tf.data与Sequence组合方案

做Keras多输入模型比如文本图像同时喂给网络做融合分类的时候最头疼的往往不是网络结构本身而是训练数据怎么喂进去。我见过不少人模型搭得挺漂亮一跑到model.fit就报错报的都是“数据不一致”或者“维度对不上”这种玄学问题。其实多半就是数据加载这一步没处理好。多输入模型和多迭代器输入的数据加载方法确实和单输入模型差别很大不是简单把 list 塞进去就完事。这篇文章就把我实际踩过坑之后梳理出来的几种方案、完整示例和排查思路写清楚适合正在用 Keras/TensorFlow 做多模态或多输入模型的开发者参考。1. 为什么多输入模型的数据加载这么容易翻车1.1 普通 fit 一次性加载的隐性瓶颈很多初学者以为多输入模型就是把输入数据改成一个 listmodel.fit([x_text, x_image], y)。数据量小的时候确实可以跑通比如几百条样本一秒钟就加载完。但一旦数据量到几万、几十万或者样本本身是大尺寸图片、长文本序列一次性把所有数据放进内存再训练内存直接爆炸。我见过一台 32G 内存的机器硬塞 20 万张 224x224 的图片还没开始训练就卡死最后只能用 Swap 硬撑训练速度慢得离谱。普通 fit 还有一个更隐蔽的问题它要求你在调用 fit 之前整个数据集已经完整存在于内存中。这意味着你没法做真正的“边训练边加载”也没法实现数据增强的流式生成。比如你要在训练时随机给图像加噪声、做裁剪如果用普通 fit就得先全部处理好再喂进去这既浪费内存又少了动态增强的灵活性。所以当我把数据量一放大马上转向迭代器输入方案。1.2 多输入带来的维度对齐问题单输入模型的数据加载只需要保证(样本数, 特征维度)和模型输入层对应即可。多输入模型则要求每个输入分支的第一维样本数完全一致否则fit会直接报维度不匹配。比如说文本分支处理完是(batch, 128)图像分支处理完是(batch, 224, 224, 3)batch 必须相同但第二个维度可以不同。听起来简单实际操作中如果前面做数据预处理时没对齐索引很容易出现两个输入取样顺序不一致的情况。另一个容易忽略的点是如果模型有多输出那么y也必须是一个 list并且要和输出层的顺序一一对应。我见过有人把y写成单个数组然后损失函数设了两个模型直接报错“期望两个目标值”。这些看似小问题在迭代器模式下会被放大因为迭代器返回的数据结构一旦不对fit 连错误信息都显得很抽象。1.3 多迭代器输入的典型场景所谓“多迭代器输入”我之前理解成“多个数据流同时往模型里灌”。实际项目中一般有这么几类需求不同类型的数据源同时输入比如数据库里的结构化字段和离线计算的特征文件两个独立的数据增强流程比如图像走一个增强 pipeline文本走另一个 tokenizer 流程多任务学习里需要从不同数据集中采样再合并成一个 batch。这些场景下单一迭代器没法覆盖必须手动组合多个迭代器。Keras 官方其实没有提供一个所谓的“multi-iterator”专用 API但我们可以用tf.data.Dataset的zip、keras.utils.Sequence的多个子序列或者自定义生成器来组合。下面几种方案我都实测过各自有适用边界。2. 三种主流多输入数据加载方案拆解2.1 方案一基于 tf.data.Dataset 的组合加载tf.data.Dataset是目前 Keras 里最推荐的数据管道方案。多输入的核心思路是分别构造每个输入分支的 Dataset再用tf.data.Dataset.zip组合成一个 tuple 的 Dataset。这里有个细节zip 组合时每个 Dataset 的样本数必须对齐否则 zip 会按最短的那个停掉这在训练时容易造成“epoch 只跑了一部分”的错觉。实际代码模式大致是import tensorflow as tf # 假设文本数据是整数序列图像数据是 numpy 数组 text_ds tf.data.Dataset.from_tensor_slices(text_data) image_ds tf.data.Dataset.from_tensor_slices(image_data) label_ds tf.data.Dataset.from_tensor_slices(labels) # 先用 zip 把两个输入组合成 (x1, x2) 的结构 features_ds tf.data.Dataset.zip((text_ds, image_ds)) # 再把特征和标签组合成 ((x1, x2), y) dataset tf.data.Dataset.zip((features_ds, label_ds)) # 打乱、分批、预取 dataset dataset.shuffle(buffer_size1000).batch(batch_size).prefetch(tf.data.AUTOTUNE)这里注意一个坑直接tf.data.Dataset.zip((text_ds, image_ds, label_ds))返回的结构是(x1, x2, y)直接喂给 Keras 模型会报错。Keras 期望的输入格式是((x1, x2), y)也就是特征部分是一个 list 或 tuple标签部分单独一个元素。所以必须分两步 zip第一步把两个输入组合成一个整体第二步再拼上标签。很多教程没写这一步我当初照着写也报错后来才发现是这个结构问题。如果两个输入分支样本数不一致比如其中一个数据增强后长度变了可以在 zip 之前用.take()或.repeat()对齐。比如文本分支比较短可以用text_ds.repeat(-1)让它无限循环但这样做要特别注意 step 数的计算否则一个 epoch 永远不会结束。我的经验是尽量先保证底层数据长度一致不要依赖 repeat 来做对齐因为那会让训练步数很难控制。2.2 方案二基于 keras.utils.Sequence 的稳健选择如果你不太想碰 tf.data 那些底层tensor_slices和zip的脾气keras.utils.Sequence是更稳妥的选择。它是一个类你需要继承并实现__len__和__getitem__两个方法。多输入模型的 Sequence 就是让__getitem__返回([x1_batch, x2_batch], y_batch)这种结构。我写过一个双输入 Sequence 的骨架核心逻辑是这样from tensorflow.keras.utils import Sequence import numpy as np class MultiInputSequence(Sequence): def __init__(self, x1_data, x2_data, y_data, batch_size): self.x1 x1_data self.x2 x2_data self.y y_data self.batch_size batch_size def __len__(self): return int(np.ceil(len(self.x1) / self.batch_size)) def __getitem__(self, idx): start idx * self.batch_size end min((idx 1) * self.batch_size, len(self.x1)) # 这里可以加数据增强、归一化等操作 x1_batch self.x1[start:end] x2_batch self.x2[start:end] y_batch self.y[start:end] return [x1_batch, x2_batch], y_batch使用的时候直接seq MultiInputSequence(x1, x2, y, batch_size32) model.fit(seq, epochs10, validation_dataval_seq)Sequence 的好处是它天然支持多进程。只要在 fit 里传workers4, use_multiprocessingTrue它就会开多个 worker 并行调用__getitem__而且不会像普通生成器那样容易乱序或重复。这里有个跟 tf.data 不同的关键点Sequence 的下标是顺序取batch顺序天然稳定配合shuffle也不容易出问题。它唯一的短板就是需要自己管理数据增强逻辑不像 tf.data 有那么多现成的map和batch操作。2.3 方案三自定义生成器组合多个迭代器自定义生成器是最灵活的方案适合那些数据来源本身就很杂的项目。比如一个输入来自 CSV 流式读取另一个输入来自图片目录动态加载你可以用yield在同一个生成器函数里分别迭代两个数据源每次组装成一个 batch 返回。简单例子def multi_input_generator(iter1, iter2, labels_iter, batch_size): while True: x1_batch [] x2_batch [] y_batch [] for _ in range(batch_size): try: x1 next(iter1) x2 next(iter2) y next(labels_iter) x1_batch.append(x1) x2_batch.append(x2) y_batch.append(y) except StopIteration: # 到达末尾就重新初始化迭代器 iter1, iter2, labels_iter reset_iterators(...) break if len(x1_batch) batch_size: yield [np.array(x1_batch), np.array(x2_batch)], np.array(y_batch)用生成器配合fit_generator在 TensorFlow 2 里直接model.fit也接受生成器。这种方式的最大坑是当多个迭代器长度不一致时非常容易出现“末尾部分样本没喂进去”或者“某个迭代器先耗尽导致整个生成器退出”的问题。我的建议是如果要用生成器就给每个迭代器包一层无限循环在尾部 reset然后在训练中靠steps_per_epoch来限制一轮的步数这样就不会提前退出。另外自定义生成器在多进程下要特别小心 Python 的 GIL 和随机性。use_multiprocessingTrue时如果生成器内部用了 Python 的随机函数不同进程会拿到相同的随机种子造成所有进程生成同样的增强数据。解决办法是在生成器开头加一个np.random.seed(...)种子可以取进程 ID 和当前时间的组合或者直接用tensorflow.random操作。3. 实操双输入双输出模型的完整加载流程3.1 构建一个文本图像融合模型为了把上面这些方法放在一个真实场景里讲我设计了一个模拟项目X输入是一段短文本和一张灰度图输出有两个分支一个是文本情感分类的概率分布一个是图像类别概率分布。模型结构大概是这样from tensorflow.keras.layers import Input, Embedding, LSTM, Conv2D, Flatten, Dense, Concatenate from tensorflow.keras.models import Model # 文本输入分支整数序列 text_input Input(shape(32,), nametext_input) x Embedding(100, 64)(text_input) x LSTM(32)(x) # 图像输入分支32x32 灰度图 image_input Input(shape(32, 32, 1), nameimage_input) y Conv2D(16, 3, activationrelu)(image_input) y Flatten()(y) y Dense(32)(y) # 融合 combined Concatenate()([x, y]) # 两个输出 out1 Dense(3, activationsoftmax, namelabel_out)(combined) out2 Dense(4, activationsoftmax, nameimage_out)(combined) model Model(inputs[text_input, image_input], outputs[out1, out2]) model.compile( optimizeradam, loss{label_out: categorical_crossentropy, image_out: categorical_crossentropy}, metrics{label_out: accuracy, image_out: accuracy} )注意编译时我用的是字典映射的方式指定每个输出对应的损失和指标。相比 list 方式字典方式更清晰不会搞混输出顺序。如果你的输出层 name 没设置也可以用 list 方式但顺序必须严格对应模型输出顺序建议还是把 name 写清楚。3.2 用 tf.data 构造训练流水线模拟数据就直接随机生成import numpy as np num_samples 20000 text_data np.random.randint(0, 100, size(num_samples, 32), dtypenp.int32) image_data np.random.rand(num_samples, 32, 32, 1).astype(np.float32) label_data np.random.randint(0, 3, size(num_samples, 1)) image_label_data np.random.randint(0, 4, size(num_samples, 1)) # one-hot 化 label_onehot tf.keras.utils.to_categorical(label_data, num_classes3) image_label_onehot tf.keras.utils.to_categorical(image_label_data, num_classes4)注意这里text_data是整数不能是 float因为 Embedding 层要求整数输入。我见过有人把文本数据用float64喂进去直接报InvalidArgumentError检查半天才发现是 dtype 问题。然后构造 Datasettext_ds tf.data.Dataset.from_tensor_slices(text_data) image_ds tf.data.Dataset.from_tensor_slices(image_data) label_ds tf.data.Dataset.from_tensor_slices((label_onehot, image_label_onehot)) features_ds tf.data.Dataset.zip((text_ds, image_ds)) dataset tf.data.Dataset.zip((features_ds, label_ds)) dataset dataset.shuffle(buffer_size2048).batch(64).prefetch(tf.data.AUTOTUNE)训练时history model.fit( dataset, epochs5, steps_per_epochlen(text_data) // 64, validation_dataval_dataset, validation_stepsval_steps, verbose1 )这里steps_per_epoch如果不设置Keras 会尝试自动推断但在Dataset是无限循环或经过repeat()时它会警告并且可能跑过头。建议手动算一下样本数 // batch_size注意整除问题如果除不尽最后一个不完整 batch 会被丢掉。想让最后一个 batch 也能参与训练可以在batch设置drop_remainderFalse默认然后steps_per_epoch用len(text_data) // 64 1。不过我一般用drop_remainderTrue因为模型训练时最后一个 batch 偏小反而会影响 BN 层的统计丢掉也无妨。3.3 用 Sequence 配合多进程训练如果你更想用 Sequence验证集也可以做一个 Sequence。完整代码class MultiInputSequence(Sequence): def __init__(self, x1, x2, y1, y2, batch_size, drop_lastFalse): self.x1 x1 self.x2 x2 self.y1 y1 self.y2 y2 self.batch_size batch_size self.drop_last drop_last self.length len(x1) // batch_size if drop_last else int(np.ceil(len(x1) / batch_size)) def __len__(self): return self.length def __getitem__(self, idx): start idx * self.batch_size end min((idx 1) * self.batch_size, len(self.x1)) # 实际项目中在这里做数据增强注意文本和图像增强方式不同 x1_batch self.x1[start:end] x2_batch self.x2[start:end] y1_batch self.y1[start:end] y2_batch self.y2[start:end] return [x1_batch, x2_batch], [y1_batch, y2_batch]训练时指定进程数train_seq MultiInputSequence(text_data, image_data, label_onehot, image_label_onehot, batch_size64) val_seq MultiInputSequence(val_text, val_image, val_label1, val_label2, batch_size64) model.fit( train_seq, epochs5, validation_dataval_seq, workers4, use_multiprocessingTrue, shuffleTrue )shuffleTrue是 Sequence 独有参数每次 epoch 开始时它会重新打乱样本顺序。我一开始没加结果每个 epoch 样本顺序都一样模型训练虽然没报错但收敛波动特别大。后来加上 shuffle稳定多了。注意用use_multiprocessingTrue时Sequence 里的所有数据会被 pickle 后复制到子进程。如果你的数据量特别大建议把数据存储为文件名列表__getitem__里再按文件名从磁盘读取而不是把整个 numpy 数组都传进去。不然多进程内存占用会翻好几倍反而比普通 fit 更容易内存不足。3.4 多迭代器组合的实际代码如果两个输入真来自两个完全不同的迭代器比如一个从 TFRecord 读另一个从数据库游标读可以这样组合def combined_generator(iter_a, iter_b, batch_size): while True: batch_a [] batch_b [] for _ in range(batch_size): try: a next(iter_a) except StopIteration: iter_a recreate_iter_a() a next(iter_a) try: b next(iter_b) except StopIteration: iter_b recreate_iter_b() b next(iter_b) batch_a.append(a) batch_b.append(b) yield [np.stack(batch_a), np.stack(batch_b)]这个 generator 只会产生x不产生y所以用的时候还要把标签单独包一层。实际上我更推荐用tf.data.Dataset的组合方式因为有现成的zip而且内部调度是 C 实现的性能比 Python 手动 next 高不少。只有数据源无法转换成 tf.data 时才用自定义生成器。4. 常见问题与排查技巧实录4.1 模型.fit 报 ValueError: Data is expected to be in batch format这个报错我在网上看到很多次自己也遇到过。通常是因为你喂给 fit 的数据结构里特征部分不是被当作“一组输入”而是被当作“一个输入张量”。比如你写model.fit([x1, x2], y)时如果模型是函数式 API这个 list 会被正确解释为两个输入。但如果你不小心把x1和x2先 np.concatenate 了喂进来的就变成了一个更高维的数组模型自然说 batch 格式不对。排查方法很简单打印一下x的类型和形状确认它是否是一个长度等于输入层数量的 list。如果你用 Dataset打印next(iter(dataset))看结构是不是((x1, x2), y)。结构对了问题就解决一半。4.2 输入样本数不一致导致训练提前终止这个坑最烦人因为不一定每次都会报错。比如你两个输入分别来自不同的 CSV 文件行数不一样用from_tensor_slices构造后直接 zipzip 只取两者共同的部分。你以为训练了 1000 步其实第 800 步开始数据就偏了模型在缺少预兆的情况下提前停了。我用过一个笨办法来排查在训练前先算一下各数据源的总长度打印出来比较。如果真不一致就先把多的那个做.take()截断或者把少的那份复制补齐。不要用repeat()硬撑因为repeat()会让 Dataset 变成无限长Keras 推算steps_per_epoch时就会出问题训练容易陷入死循环。4.3 steps_per_epoch 应该怎么设很多新手会问数据集有 1000 个样本batch_size32steps_per_epoch 填 31 还是 32严格来说如果drop_remainderFalse最后一个 batch 只有 8 个样本也要算一个 step所以是ceil(1000/32)32。如果drop_remainderTrue丢掉不完整 batchsteps 就是floor(1000/32)31。我建议统一用一个工具函数计算def calc_steps(num_samples, batch_size, drop_remainderFalse): import math if drop_remainder: return num_samples // batch_size return math.ceil(num_samples / batch_size)然后在配置里显式传steps_per_epochcalc_steps(...)这样训练日志里的进度条才准确不至于最后多跑一步或少跑一步。如果你使用model.fit时同时传了batch_size和steps_per_epochKeras 会优先用steps_per_epoch但batch_size对 Dataset 无效因为 Dataset 内部已经有 batch 操作了。4.4 多进程下随机种子导致数据一模一样这个问题我出差过用keras.utils.Sequence配合workers8训练时发现每个 epoch 前几步 loss 下降模式完全一致像复读机。后来查出来是np.random.randint在子进程里默认使用相同的随机种子。解决办法在__init__里根据进程信息设置种子。import os import numpy as np class MultiInputSequence(Sequence): def __init__(self, ...): seed int(os.getpid()) ^ int(time.time()) np.random.seed(seed % (2**32))最简单粗暴的做法是让每个 worker 自己在__getitem__里初始化一个基于随机种子的生成器而不是用全局 np.random。如果你用的是 TensorFlow 操作做数据增强建议用tf.random.Generator它自带可复现性设置的选项比 numpy 更安全。4.5 tf.data 的 shuffle 吞内存shuffle(buffer_size2048)这个 buffer_size 不是越大越好。它默认是“先装满 buffer再随机取一个”如果 buffer_size 设成 20 万想模拟整个数据集 shuffle内存会直接占用巨大尤其数据是图片时每个样本几十 KB乘上 20 万就是十几个 GB。我的建议是如果数据量大到无法整体 shuffle就牺牲一点随机性把 buffer_size 设为batch_size * 几十比如 batch_size64buffer_size 设为 1000 到 2000 就够用了。想要更彻底的 shuffle可以先做一次文件级别的乱序再用小 buffer。优先保证训练不崩。4.6 Windows 下多进程报错在 Windows 下使用use_multiprocessingTrue经常遇到RuntimeError: An attempt has been made to start a new process before the current process has finished its bootstrapping phase。这是因为 Windows 没有 fork只能 spawn所以会重新导入主模块。解决办法是把你训练代码放到if __name__ __main__:块里并且不要在全局作用域直接调用model.fit。if __name__ __main__: model.fit(train_seq, workers4, use_multiprocessingTrue)如果你觉得麻烦可以直接用workers0或者workers2, use_multiprocessingFalse在 Windows 上还是能正常跑只是数据加载会比较慢。对于 CPU 密集型的数据预处理多进程收益很大但如果是纯读取磁盘多线程效果也不差。我一般先用workers4, use_multiprocessingFalse测性能再决定要不要开多进程。5. 一些让我省过大力气的细节最后分享几个我很早以前没注意、后来才发现特别重要的点。第一个是数据管道的类型一致性。Text 分支喂给 Embedding 层必须是整数图像分支必须是 float而且最好归一化到 0~1。我在实际项目里遇到过一个模型效果总是不对的情况后来发现图像数据忘记归一化数值范围还是 0~255。迭代器模式下这类错误通常不会报错只会让 loss 疯狂跳动排查起来特别耗时间。第二个是不要在__getitem__里做耗时过长的 CPU 操作。比如每次读取几张图片并 resize如果每张耗时 50msbatch_size64 就意味着单个 batch 要 3 秒多。这种情况下不管你怎么调 workers 都很慢。最好是预先把数据转成能够快速读取的格式比如 TFRecord 或者npy文件或者用 GPU 上的数据增强算子让 CPU 的计算和 GPU 训练并行起来。第三个是给训练和验证的数据打乱策略不一样。训练集需要 shuffle增强验证集不需要 shuffle也不需要增强。如果用同一个 Sequence 类记得在on_epoch_start之类的回调里控制训练集 shuffle验证集保持顺序。否则验证集每次看到的顺序不同验证 loss 的波动会很大影响你对模型好坏的判断。第四个是使用model.fit的新版接口时注意回调里的日志。当输入是 Dataset 时history.history里每个 epoch 的 loss 是每个 batch 的平均可能跟你用model.evaluate跑出来的数值不同因为 evaluate 会把所有 batch 的损失做更精细的加权平均。这不是代码 bug只是实现差异不用太纠结。5.1 一个我踩了三次的坑标签结构多输出模型的标签结构我踩了三次才彻底记住。如果你在model.compile里用字典指定 loss那么喂给模型的y也要是字典或 list。比如y {label_out: label_onehot, image_out: image_label_onehot}这样模型看起来最清晰。但如果你的模型输出层有名字也可以用 list 顺序对应。用 Dataset 时如果你返回的是((x1, x2), (y1, y2))那就没问题。可一旦你返回((x1, x2), y1)少了一个标签fit 会在第一个 batch 就报错“期望两个目标”。这个错误信息其实很明确只是有时候我们被前面一堆数据管道代码搞晕了没注意到是标签结构的问题。5.2 数据规模再大时可以考虑 TFRecord如果数据量大到几十 GB或者要在多机分布式训练上面这些基于 numpy 数组的方案就不够用了。建议把数据写成 TFRecord然后从 TFRecord 读取生成 Dataset。多输入模型照样可以把不同输入写在同一个 TFRecord 的多个 feature 里读取时用parse_example分别解析。这样不但能避免内存爆炸还能利用tf.data.experimental的并行读取能力。这里不展开但如果你之后遇到性能瓶颈方向可以考虑这个。根据我个人实践多输入模型的数据加载真正决定了训练流程的稳定性和效率。建议先从keras.utils.Sequence入手因为出错时容易排查跑通之后再逐步迁移到tf.data.Dataset提升性能。别一上来就追求最复杂的自定义生成器那只有在数据源极其特殊的时候才值得。希望这篇整理能让你在搭下一个多输入模型时少一点踩坑多一点顺畅。