1. 多输入模型你迟早会遇到的数据加载问题做深度学习几年时间用Keras/TF 2.x搭过不少模型。单输入模型的数据加载谁都会一个ImageDataGenerator或者一个tf.data.Dataset轻松搞定。但是真正到了实际项目里多输入模型几乎是避不开的一关。什么叫多输入模型说白了一个模型吃多个来源的数据。典型场景比如一个推荐系统要同时输入用户ID序列和商品特征一个多模态模型要同时接受图像和文本目标检测网络需要图片张量加边界框坐标甚至简单点一个回归模型同时输入数值特征和类别特征。这些场景下模型不是只吃一个张量而是吃一批张量。Keras的函数式API支持多输入本身并不难真正的门槛在于训练时数据怎么喂。很多人模型定义好了卡在model.fit()这一步不知道数据怎么组织。尤其当你需要迭代器输入而不是把全部数据一次性加载进内存的时候问题就变得更加具体多个输入和多个迭代器之间到底是什么关系多个迭代器各自产出数据怎么对齐batch怎么配合这篇文章就围绕这两个核心问题展开第一多输入模型的输入结构怎么设计第二多迭代器输入的数据加载方案怎么落地。我会直接给出可运行的代码、踩坑记录和排查思路尽量让你少走弯路。2. 多输入模型的设计思路拆解2.1 什么时候需要多输入什么时候不需要在聊数据加载之前先搞清楚模型本身的设计。因为很多“多输入”需求其实是伪需求——你把两个输入拼成一个输入就完事了为什么非要做两个分支来看两个例子例子A模型输入是用户的数值特征年龄、收入、点击次数和类别特征性别、城市ID。这种情况不要做成多输入模型。把数值特征和类别特征各自经过Embedding层处理后在模型内部拼接即可输入层只需要一个输入。数据加载时也只需要一个输入迭代器。我之前见过有人把这种场景硬生生拆成两个Input层结果训练时还得维护两个迭代器纯属给自己找麻烦。例子B模型同时接收一张商品图片和一个商品标题的文本序列输出点击率。这种是真正的多输入模型图像分支用卷积网络提取视觉特征文本分支用EmbeddingLSTM/Transformer提取语义特征两个分支的特征在某个层拼接后接全连接层。这种场景下两个输入在数据层面就是不同的类型、不同的预处理流程、不同的张量形状必须分开喂。区分标准就一句话数据能不能合成一个张量能就单输入不能就多输入。这不光是模型结构的问题直接决定了后面的数据加载复杂度。2.2 为什么选择迭代器/生成器输入而不是全量加载很多入门用户习惯把数据全部读到内存里转化成NumPy数组然后model.fit(x[data1, data2], ylabel)。这种做法在小数据集上完全可行代码也简单。但一旦数据量上来问题立刻暴露图像数据几万张每张几百KB到几MB全部读入内存直接爆掉或者数据是流式的训练过程中不断有新数据进来又或者数据增强需要在每次迭代时现场做。这些场景下迭代器式输入的优越性就出来了——它是边用边取的机制内存占用恒定还能在取数据的过程里做实时预处理和增强。Keras 2.x时代官方给的标准答案是ImageDataGenerator.flow_from_directory()那其实就是一种生成器。Keras 3.x/TF 2.x时代官方推荐的答案变成了tf.data.Dataset。另外还有很多人用自定义的keras.utils.Sequence子类。这三者都可以作为迭代器但它们在多输入场景下的适配程度差别非常大。2.3 多迭代器输入的核心矛盾对齐问题多输入模型配多迭代器输入有一个绕不开的核心矛盾每个迭代器各自独立产出数据但一次model.fit()迭代中模型需要同时拿到所有输入的一个batch。如果多个迭代器各自为政训练就乱套了——图像分支拿到的第1个batch可能对应第1-32个样本文本分支拿到的第1个batch却对应第17-48个样本完全错位。所以多迭代器输入数据加载方案的设计目标就是在保证“多迭代器”灵活性的前提下严格维护样本维度的对齐关系。理解了这一点后面所有方案都是围绕这个目标展开的。3. 常用数据加载方式对比与选型3.1 三种主流方式概览我把实际项目中见过的多输入模型数据加载方案整理成一张表先整体看一下方案适用场景优点缺点tf.data.Datasetzip数据量大、管线复杂、需要高性能预处理性能好、原生支持多输入拼接、并行度高调试相对复杂API上手有门槛KerasSequence子类多输入且需要灵活控制批次逻辑结构清晰、进度条友好、易调试不如Dataset性能极致需要自己写逻辑普通Generatoryield快速原型验证代码最简洁性能一般多输入组合容易出错不推荐用于生产全量NumPy数组小数据集快速测试最简单、最直接数据稍大就内存爆炸不适合大规模训练坦白说我实际用得最多的是前两种。如果你的输入都是图像类数据两个分支的预处理流程都一样那么tf.data.Dataset最合适如果你的两个输入是完全不同类型的数据比如图像和文本预处理流程差异巨大那么Sequence子类反而更清晰因为你可以把整批数据的组织逻辑收敛在一个类里。3.2 为什么优先推荐Dataset方案tf.data.Dataset在TF 2.x里属于一等公民。它强大的地方在于可以很自然地表达“多个独立数据源”和“数据源拼接”这两个语义。举个例子图像数据从一个目录读取文本数据从一个TSV文件读取。你可以分别构建图像Dataset和文本Dataset然后用tf.data.Dataset.zip()把它们组合成一个产出((img_batch, text_batch), label_batch)的数据管线。zip操作保证每一步同时从一个Dataset取一个batch对齐天然成立。这就是我在文章标题里强调的“多迭代器输入”的正解之一多个迭代器分别构建然后在进入model.fit()之前合并成一个迭代器。另外tf.data.Dataset有非常好的性能优化空间num_parallel_calls、prefetch、映射函数的并行化等这些在大规模训练时能省出大量时间。而且model.fit()对Dataset的适配非常完美steps_per_epoch、validation_data都可以搭配得很顺。4. 实操基于tf.data.Dataset的多输入数据加载4.1 环境与依赖先交代一下环境。我用的是Python 3.9TensorFlow 2.10Keras 2.x API理论上2.4以上都能跑如果需要图像解码需要安装Pillow具体版本不是关键关键是API调用方式不要混杂。下面代码都基于tf.keras。4.2 场景定义为了演示更具体我用一个模拟项目多模态感情识别模型输入是“文本序列”和“音频MFCC特征”输出三分类积极、中性、消极。这个场景足够典型两个输入类型完全不同数据预处理流程也完全不同。数据格式设想如下文本部分每条样本是定长序列分词后按词表映射为整数ID长度截断/填充到50音频部分每条样本是MFCC特征矩阵20帧 × 13维标签one-hot三分类或者直接整数索引用sparse categorical loss假设文本ID数据已经保存为一个NumPy数组text_ids形状(N, 50)MFCC特征保存为audio_feats形状(N, 20, 13)标签为labels形状(N, 3)。由于数据不大演示用全量数组构建Dataset再走批处理流程。4.3 构建模型骨架先定义一个简单的多输入模型用于数据加载验证import tensorflow as tf from tensorflow.keras import layers, Model # 文本输入分支 text_input layers.Input(shape(50,), nametext_input) x1 layers.Embedding(input_dim5000, output_dim64)(text_input) x1 layers.Bidirectional(layers.LSTM(32))(x1) # 音频输入分支 audio_input layers.Input(shape(20, 13), nameaudio_input) x2 layers.Flatten()(audio_input) x2 layers.Dense(32, activationrelu)(x2) # 合并 concat layers.Concatenate()([x1, x2]) output layers.Dense(3, activationsoftmax, nameoutput)(concat) model Model(inputs[text_input, audio_input], outputsoutput) model.compile( optimizeradam, losscategorical_crossentropy, metrics[accuracy] )这里inputs是一个列表顺序很重要必须和数据加载时给出的输入顺序保持一致。这也是一不小心就出错的地方。4.4 核心实现Dataset构建与zip组合下面这段代码就是这篇文章的核心内容了BATCH_SIZE 32 AUTOTUNE tf.data.AUTOTUNE def build_paired_dataset(text_ids, audio_feats, labels, batch_sizeBATCH_SIZE, shuffleTrue): # 从张量/数组构建各自独立的Dataset text_ds tf.data.Dataset.from_tensor_slices(text_ids) audio_ds tf.data.Dataset.from_tensor_slices(audio_feats) label_ds tf.data.Dataset.from_tensor_slices(labels) # 组合成一个Dataset产出为文本, 音频, 标签三元组 paired_ds tf.data.Dataset.zip((text_ds, audio_ds, label_ds)) if shuffle: paired_ds paired_ds.shuffle(buffer_size5000) # 转换批次信息 # 注意这里不用batch()而是采用mapbatch的方式以便统一处理 paired_ds paired_ds.map( lambda t, a, l: ({text_input: t, audio_input: a}, l), num_parallel_callsAUTOTUNE ).batch(batch_size).prefetch(AUTOTUNE) return paired_ds有几个值得说的地方。第一个zip((text_ds, audio_ds, label_ds))这个操作很关键它把三条独立的“数据河流”合在一起。每次迭代取一个元素实际上一次性从三条河里各取一个元素打包成一个三元组。这样就保证了样本级别的严格对齐。第二个map函数里输出的字典key必须和模型定义的Input层的name一一对应。{text_input: t, audio_input: a}这两个key正好对应前面定义的text_input和audio_input。如果你用列表[t, a]而不是字典也可以但顺序必须严格和model.compile时inputs的顺序一致。我个人推荐用字典形式因为可读性好而且不存在顺序对错的隐患。第三个这里为什么用了map再batch而不是直接batch其实你也可以在zip之前分别batch但那样代码看起来会更绕。放在map之后批量处理是因为map阶段负责把样本转换成语义清晰的结构batch阶段负责打包逻辑更清晰。4.5 训练与验证有了上面的构建函数训练就非常简单了train_ds build_paired_dataset( text_ids[:10000], audio_feats[:10000], labels[:10000], shuffleTrue ) val_ds build_paired_dataset( text_ids[10000:], audio_feats[10000:], labels[10000:], shuffleFalse ) model.fit( train_ds, validation_dataval_ds, epochs10, steps_per_epochlen(text_ids[:10000]) // BATCH_SIZE, validation_stepslen(text_ids[10000:]) // BATCH_SIZE )这里有一个细节shuffleFalse对验证集很重要——验证集不需要乱序保持原有顺序反而便于观察指标。steps_per_epoch如果留空Keras会从Dataset推断但多输入Dataset在个别老版本里会推断错。指定了反而最保险。另外如果你的Dataset是用batch()而不是batch_size参数在fit()里指定的steps_per_epoch必须以batch数量为单位算不要把总样本数直接填进去。4.6 直接读取两个目录/文件的Dataset写法如果把上面的NumPy数组换成直接从文件读取也很自然。比如一个分支从图片目录读取一个分支从CSV读取特征# 图片Dataset image_ds tf.keras.utils.image_dataset_from_directory( images/, label_modeNone, # 我们不从这里读标签 batch_sizeNone, image_size(128, 128), shuffleFalse ) # 特征与标签从CSV读取 def read_csv_and_get_feats_labels(csv_path): import pandas as pd df pd.read_csv(csv_path) feats df.iloc[:, :-1].values.astype(float32) labels df.iloc[:, -1].values.astype(int32) return feats, labels feats, labels read_csv_and_get_feats_labels(features.csv) feat_ds tf.data.Dataset.from_tensor_slices(feats) label_ds tf.data.Dataset.from_tensor_slices(labels) paired_ds tf.data.Dataset.zip((image_ds, feat_ds, label_ds)) # 后续继续组合成模型需要的格式这种方式要求两个文件/目录里的样本顺序完全一致一步错则步步错所以建议在数据准备阶段就把排序逻辑统一。图像目录的默认排序建议显式控制不要依赖文件系统自动排序。5. 实操基于Keras Sequence的多输入数据加载5.1 为什么有时候Sequence更顺手tf.data.Dataset虽好但在某些场景下确实不如自定义Sequence方便。举几个例子两个输入来自两个不同的数据集而且每个数据集内部都有自己的索引体系需要在一个batch内同时定位两个输入的batch组成逻辑不一样但你又想统一控制需要在每个epoch做一次自定义的打乱逻辑数据增强逻辑非常依赖Python生态的库不太方便向量化这种情况下写一个Sequence子类把一切逻辑包在里面反而是最省心的方案。5.2 完整的Sequence实现还是用前面多模态感情识别那个场景换一种实现import numpy as np from tensorflow.keras.utils import Sequence class MultiInputSequence(Sequence): def __init__(self, text_ids, audio_feats, labels, batch_size32, shuffleTrue): self.text_ids text_ids self.audio_feats audio_feats self.labels labels self.batch_size batch_size self.shuffle shuffle self.indexes np.arange(len(text_ids)) self.on_epoch_end() def __len__(self): # 返回一个epoch内的batch数量 return int(np.floor(len(self.text_ids) / self.batch_size)) def __getitem__(self, idx): # 定位当前batch的索引 batch_indexes self.indexes[idx * self.batch_size:(idx 1) * self.batch_size] # 分别取两个输入和标签 batch_text self.text_ids[batch_indexes] batch_audio self.audio_feats[batch_indexes] batch_labels self.labels[batch_indexes] # 返回格式必须匹配模型输入定义 return ( { text_input: batch_text, audio_input: batch_audio }, batch_labels ) def on_epoch_end(self): if self.shuffle: np.random.shuffle(self.indexes)然后调用train_seq MultiInputSequence( text_ids[:10000], audio_feats[:10000], labels[:10000], batch_size32, shuffleTrue ) val_seq MultiInputSequence( text_ids[10000:], audio_feats[10000:], labels[10000:], batch_size32, shuffleFalse ) model.fit( train_seq, validation_dataval_seq, epochs10 )这个方案的核心逻辑全浓缩在__getitem__里。因为model.fit()每次调用__getitem__时传入一个整数idx你在这个整数的基础上自己决定取哪些样本所以你可以完全掌控“索引 → 数据”的映射过程。5.3 多迭代器输入的Sequence变体有的同学会遇到这种情况数据不是一个数组能直接索引的而是来自两个独立的迭代器或文件句柄。比如文本数据是一个TFRecord文件音频数据是一个目录下的若干文件这种情况下你可能不想全部读进内存。Sequence同样可以接住这个需求只要在__init__里提前准备好文件路径列表在__getitem__里按路径即时读取class MultiInputSequenceFromFiles(Sequence): def __init__(self, text_paths, audio_paths, labels, batch_size32, shuffleTrue): self.text_paths text_paths self.audio_paths audio_paths self.labels labels self.batch_size batch_size self.shuffle shuffle self.indexes np.arange(len(text_paths)) self.on_epoch_end() def __len__(self): return int(np.floor(len(self.text_paths) / self.batch_size)) def __getitem__(self, idx): batch_indexes self.indexes[idx * self.batch_size:(idx 1) * self.batch_size] batch_text [preprocess_text(self.text_paths[i]) for i in batch_indexes] batch_audio [preprocess_audio(self.audio_paths[i]) for i in batch_indexes] batch_labels self.labels[batch_indexes] return ( { text_input: np.array(batch_text), audio_input: np.array(batch_audio) }, np.array(batch_labels) )preprocess_text和preprocess_audio是自定义预处理函数可以是读文件、分词、解码音频任意逻辑。分支逻辑互不干扰、完全不互相污染这种隔离性是这类方案最大的价值。5.4 Sequence方案的注意点用Sequence时有几个坑我逐一踩过这里直接列出来第一个坑__len__如果计算错误会出现epoch提前结束但还有大量样本没被遍历或者最后一截数据被丢弃。注意我用了int(np.floor(...))这意味着最后不满足一个batch的样本会被丢弃。如果你不想丢弃得自行修改__getitem__的越界判断和__len__的计算逻辑生产环境里这个细节很常见。第二个坑__getitem__的返回值结构必须严格等于模型输入输出结构。模型是inputs[text_input, audio_input]这里返回字典形式或者列表形式要固定中途不要换。第三个坑多进程参数workers 1时Sequence会被复制到多个worker进程中。你的__getitem__里如果用了不安全的全局状态或者外部可变对象容易出现数据错乱。尽量保证__getitem__是纯函数式的——输入idx输出对应数据不依赖任何可变的临时状态。6. 两种方案的对比与选型建议6.1 开个对比表直观看清楚对比点tf.data.DatasetKeras SequenceAPI抽象层级更高声明式更低命令式可读性中下管线越长越复杂中上逻辑都在类里性能上限极高原生支持并行map/prefetch高但受限于Python函数调用开销shuffle逻辑内置shuffle(buffer_size)自己实现on_epoch_endbatch抽取batch()方法统一做自己在__getitem__做错误排查管线复杂时较难直接打断点/print更直观数据加载隔离性两个输入在map阶段组合互相有耦合两个输入在__getitem__中各自独立读取完全隔离对大数据集支持极好可以完全不进内存较好只要__getitem__里现读现处理6.2 我个人的选型原则根据实际项目经验我总结了一条比较实用的选择规则如果你的两个输入能转化为Dataset并且用zip组合优先Dataset如果两个输入类型差异过大、每个batch的组装逻辑比较复杂或者你需要精细控制batch内部的数据组织选Sequence。举几个实际感受。做一个视觉-语言模型图像和文本都已经是Tensor形式预处理也都可以向量化那tf.data.Dataset简直不要太顺手几乎零额外代码就能完成多输入加载。但是做一个多传感器融合项目每个传感器的数据格式、读取方式、时间对齐方式都不一样强扭成Dataset会导致一堆奇怪的ad-hoc操作反而是Sequence里逐个读取、逐个对齐更可控。7. 常见问题与排查技巧实录7.1 多输入数据加载最常踩的坑报错信息/症状原因解决方案ValueError: Layer model expects 2 input(s), but it received 1 input tensormodel.fit传入的数据结构跟模型输入数量不一致确认数据是[input1, input2]或{name1: data1, name2: data2}格式而不是一个整体张量Invalid argument: indices[0] X is not in [0, Y)Embedding层输入ID越界检查词表大小设置尤其是数据加载时ID是否做过截断/过滤两个输入分支精度差异极大数据未对齐比如文本的第2个batch和图像的第9个batch配对了用zip确保对齐用Sequence时基于同一个indexes切片两个输入训练速度慢CPU占用率不高预处理管线没有并行化Dataset方案加num_parallel_callsAUTOTUNE和prefetch(AUTOTUNE)Sequence方案加workers和use_multiprocessingsteps_per_epoch报错或不匹配Dataset长度和指定steps不对应手动指定steps_per_epoch len(dataset) // batch_size或让Dataset自己决定不要混用TypeError: Tensor object is not iterable在tf.function图模式里尝试用Python迭代Tensor多输入数据组装逻辑在map函数里做时要保持Tensor操作风格不要在map内做Python对象层面的迭代7.2 数据对齐问题深度排查这里再展开说一下数据对齐因为这是多输入模型数据加载里最隐蔽、也最致命的错误。如果你把全部数据预先都放在内存里对齐问题几乎不会出现——因为你就两个NumPy数组直接切片嘛。但一旦你从两个不同的源迭代数据对齐就变成需要显式维护的东西。Dataset方案中zip本身就是对齐操作这是它的天然优势。Sequence方案中只要两个分支共享同一个batch_indexes对齐也是天然成立的。怕的是有些人手动写Generatordef bad_generator(): for img in image_iter: for text in text_iter: yield (img, text)这种双重循环写法就是灾难——每个图像会和所有文本配对一遍训练时模型很快就能在验证集上“过拟合”到状态很差。因为它学到的根本不是有效语义而是把两种数据胡乱组合模型根本得不到稳定的监督信号。如果在训练中发现loss不下降、精度一直在随机水平附近徘徊除了检查网络结构以外优先怀疑这类数据错配问题。7.3 关于shuffle的细节经验很多人容易忽略shuffle时机对多输入管线的影响这里有一个容易被忽视的经验Dataset方案中我建议先zip再shuffle。这保证了每条样本的多个输入作为一个整体被打乱它们的对应关系不会破坏。如果你先分别shuffle再zip那边等于随机配对完全破坏对齐。具体来说# 正确zip后整体shuffle paired_ds tf.data.Dataset.zip((text_ds, audio_ds)).shuffle(5000) # 错误各自shuffle后再zip文本和音频的对应关系被破坏 text_ds text_ds.shuffle(5000) audio_ds audio_ds.shuffle(5000) paired_ds tf.data.Dataset.zip((text_ds, audio_ds))Sequence方案中on_epoch_end里np.random.shuffle(self.indexes)会同时影响两个输入的索引映射所以天然保持了正确性。这就是Sequence的另一个隐藏优势shuffle逻辑天然发生在样本层面不容易不小心打破对应关系。7.4 模型并行/GPU训练时的数据集适配问题分布式训练、多卡训练时数据加载方式的差异会被放大。TensorFlow官方推荐使用tf.distribute.Strategy配合Dataset方案因为tf.data的batch操作可以通过分布策略自动分片到各个设备。这种情况下建议给batch()传入一个固定batch size并且让Dataset自动分片处理。选Sequence则要自己实现分片逻辑在__getitem__里根据idx和num_workers自行计算数据归属比较麻烦。如果只是单卡训练或者小规模实验两者差距不大。但如果预期要扩展到多卡从一开始就选择Dataset方案是更稳妥的。8. 一些实操经验与心得说实话多输入模型的数据加载本身不是最难的难的是把“模型定义”和“数据管线”在结构上保持一致。我这里分享几条这几年自己沉淀下来的经验都是踩过坑才明白的。第一模型输入尽量用命名Input层。也就是layers.Input(shape(50,), nametext_input)只在末尾加上name参数就能极大减少数据加载端的对应混乱。调用model.fit()时数据用{text_input: ..., audio_input: ...}字典形式传入Keras会按名字自动匹配顺序错了也没关系。这一点在很多排错场景下省了我大量的时间。第二在训练前做一次前向验证。这是一个非常值得养成的习惯。在真正model.fit()之前先手动跑一遍数据管线取一个batch出来直接model.predict(batch_data)或者model.train_on_batch(batch_data, batch_labels)。如能正常执行说明从数据形状到组装格式都已正确。这一步比训练到一半再看报错效率高得多。我就是靠这个习惯避开了无数低级错误。第三尽量保持数据管线可重放。尤其是Sequence方案__getitem__里的每个处理函数都应当可独立调用。这样一旦训练过程中数据异常你可以单独取某个idx跑一遍看看产出数据长什么样完全不用动训练代码。第四对大数据集的batch大小别太贪心。多输入模型意味着每一条样本的原始数据量往往是多份的。图像加文本一batch的数据量比单输入模型大好几倍。如果显存爆炸先不要急着调模型结构把batch_size调小试试很多时候数据加载管的batch大小才是瓶颈。我在多模态项目里就从64降到32甚至16模型照样收敛得很好。第五在做数据增强时要小心两个分支的联动增强。比如图像做随机裁剪时文本通常不需要任何变换但如果两个输入是同一物体的不同视角图像增强时必须用同一个随机种子否则两个视角会失配。这种联动逻辑在Sequence中写起来更容易因为你完全掌控__getitem__里发生的事情。关于Keras多输入模型的数据加载我觉得核心思想就是用结构保证对齐用命名保证对应用前向验证保证正确。结构上选Dataset的zip或Sequence的共享索引命名上用字典格式传输入然后训练前花三十秒做一个batch的前向验证。这套流程走下来不管你的多输入模型多复杂数据加载这块基本不会出大问题。文章里给的两种核心方案——Dataset和Sequence——覆盖了绝大多数的实际场景。遇到图像加文本、图像加数值、多传感器融合之类的常见情况都可以直接拿过去改改用。至于更特殊的场景比如流式数据、时间序列多通道输入思路是相通的先各自构建数据源再在样本维度上对齐联合。把对齐这个关键问题想清楚任何变体都不难应付。