企业官网建设流程全解析

做Keras多输入模型（比如文本+图像同时喂给网络做融合分类）的时候，最头疼的往往不是网络结构本身，而是训练数据怎么喂进去。我见过不少人模型搭得挺漂亮，一跑到model.fit就报错，报的都是“数据不一致”或者“维度对不上”这种玄学问题。其实多半就是数据加载这一步没处理好。多输入模型和多迭代器输入的数据加载方法，确实和单输入模型差别很大，不是简单把 list 塞进去就完事。这篇文章就把我实际踩过坑之后梳理出来的几种方案、完整示例和排查思路写清楚，适合正在用 Keras/TensorFlow 做多模态或多输入模型的开发者参考。

1. 为什么多输入模型的数据加载这么容易翻车

1.1 普通 fit 一次性加载的隐性瓶颈

很多初学者以为多输入模型就是把输入数据改成一个 list：model.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_size=1000).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_size=32) model.fit(seq, epochs=10, validation_data=val_seq)

Sequence 的好处是它天然支持多进程。只要在 fit 里传workers=4, use_multiprocessing=True，它就会开多个 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_multiprocessing=True时，如果生成器内部用了 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,), name='text_input') x = Embedding(100, 64)(text_input) x = LSTM(32)(x) # 图像输入分支：32x32 灰度图 image_input = Input(shape=(32, 32, 1), name='image_input') y = Conv2D(16, 3, activation='relu')(image_input) y = Flatten()(y) y = Dense(32)(y) # 融合 combined = Concatenate()([x, y]) # 两个输出 out1 = Dense(3, activation='softmax', name='label_out')(combined) out2 = Dense(4, activation='softmax', name='image_out')(combined) model = Model(inputs=[text_input, image_input], outputs=[out1, out2]) model.compile( optimizer='adam', 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), dtype=np.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_classes=3) image_label_onehot = tf.keras.utils.to_categorical(image_label_data, num_classes=4)

注意这里text_data是整数，不能是 float，因为 Embedding 层要求整数输入。我见过有人把文本数据用float64喂进去，直接报InvalidArgumentError，检查半天才发现是 dtype 问题。

然后构造 Dataset：

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((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_size=2048).batch(64).prefetch(tf.data.AUTOTUNE)

训练时：

history = model.fit( dataset, epochs=5, steps_per_epoch=len(text_data) // 64, validation_data=val_dataset, validation_steps=val_steps, verbose=1 )

这里steps_per_epoch如果不设置，Keras 会尝试自动推断，但在Dataset是无限循环或经过repeat()时它会警告并且可能跑过头。建议手动算一下：样本数 // batch_size，注意整除问题，如果除不尽，最后一个不完整 batch 会被丢掉。想让最后一个 batch 也能参与训练，可以在batch设置drop_remainder=False（默认），然后steps_per_epoch用len(text_data) // 64 + 1。不过我一般用drop_remainder=True，因为模型训练时最后一个 batch 偏小反而会影响 BN 层的统计，丢掉也无妨。

3.3 用 Sequence 配合多进程训练

如果你更想用 Sequence，验证集也可以做一个 Sequence。完整代码：

class MultiInputSequence(Sequence): def __init__(self, x1, x2, y1, y2, batch_size, drop_last=False): 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_size=64) val_seq = MultiInputSequence(val_text, val_image, val_label1, val_label2, batch_size=64) model.fit( train_seq, epochs=5, validation_data=val_seq, workers=4, use_multiprocessing=True, shuffle=True )

shuffle=True是 Sequence 独有参数，每次 epoch 开始时它会重新打乱样本顺序。我一开始没加，结果每个 epoch 样本顺序都一样，模型训练虽然没报错，但收敛波动特别大。后来加上 shuffle，稳定多了。

注意：用use_multiprocessing=True时，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构造后直接 zip，zip 只取两者共同的部分。你以为训练了 1000 步，其实第 800 步开始数据就偏了，模型在缺少预兆的情况下提前停了。

我用过一个笨办法来排查：在训练前先算一下各数据源的总长度，打印出来比较。如果真不一致，就先把多的那个做.take()截断，或者把少的那份复制补齐。不要用repeat()硬撑，因为repeat()会让 Dataset 变成无限长，Keras 推算steps_per_epoch时就会出问题，训练容易陷入死循环。

4.3 steps_per_epoch 应该怎么设

很多新手会问：数据集有 1000 个样本，batch_size=32，steps_per_epoch 填 31 还是 32？严格来说如果drop_remainder=False，最后一个 batch 只有 8 个样本，也要算一个 step，所以是ceil(1000/32)=32。如果drop_remainder=True，丢掉不完整 batch，steps 就是floor(1000/32)=31。

我建议统一用一个工具函数计算：

def calc_steps(num_samples, batch_size, drop_remainder=False): import math if drop_remainder: return num_samples // batch_size return math.ceil(num_samples / batch_size)

然后在配置里显式传steps_per_epoch=calc_steps(...)，这样训练日志里的进度条才准确，不至于最后多跑一步或少跑一步。如果你使用model.fit时同时传了batch_size和steps_per_epoch，Keras 会优先用steps_per_epoch，但batch_size对 Dataset 无效，因为 Dataset 内部已经有 batch 操作了。

4.4 多进程下随机种子导致数据一模一样

这个问题我出差过：用keras.utils.Sequence配合workers=8，训练时发现每个 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_size=2048)这个 buffer_size 不是越大越好。它默认是“先装满 buffer，再随机取一个”，如果 buffer_size 设成 20 万（想模拟整个数据集 shuffle），内存会直接占用巨大，尤其数据是图片时，每个样本几十 KB，乘上 20 万就是十几个 GB。

我的建议是：如果数据量大到无法整体 shuffle，就牺牲一点随机性，把 buffer_size 设为batch_size * 几十，比如 batch_size=64，buffer_size 设为 1000 到 2000 就够用了。想要更彻底的 shuffle，可以先做一次文件级别的乱序，再用小 buffer。优先保证训练不崩。

4.6 Windows 下多进程报错

在 Windows 下使用use_multiprocessing=True经常遇到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, workers=4, use_multiprocessing=True)

如果你觉得麻烦，可以直接用workers=0或者workers=2, use_multiprocessing=False，在 Windows 上还是能正常跑，只是数据加载会比较慢。对于 CPU 密集型的数据预处理，多进程收益很大；但如果是纯读取磁盘，多线程效果也不差。我一般先用workers=4, use_multiprocessing=False测性能，再决定要不要开多进程。

5. 一些让我省过大力气的细节

最后分享几个我很早以前没注意、后来才发现特别重要的点。第一个是数据管道的类型一致性。Text 分支喂给 Embedding 层必须是整数，图像分支必须是 float，而且最好归一化到 0~1。我在实际项目里遇到过一个模型效果总是不对的情况，后来发现图像数据忘记归一化，数值范围还是 0~255。迭代器模式下这类错误通常不会报错，只会让 loss 疯狂跳动，排查起来特别耗时间。

第二个是不要在__getitem__里做耗时过长的 CPU 操作。比如每次读取几张图片并 resize，如果每张耗时 50ms，batch_size=64 就意味着单个 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提升性能。别一上来就追求最复杂的自定义生成器，那只有在数据源极其特殊的时候才值得。希望这篇整理能让你在搭下一个多输入模型时少一点踩坑，多一点顺畅。