VertexAI Endpoint数据适配异常：无法转换Tensor/FlatMapDataset

问题背景 #

我编写了一个自定义类，可将Pandas DataFrame转换为包含7天周期列表的tf.data.Dataset，批大小设为32，核心代码如下：

def make_dataset(self, dataframe):
    grouped_df = list()
    return tf.data.Dataset.from_generator(
        lambda: self.make_generator_dataset(dataframe),
        output_shapes=([None, self.input_width, self.num_input_features], [None, self.label_width, self.num_label_features]), output_types = (tf.float32, tf.float32)
    )

def make_generator_dataset(self, dataframe):
    first_it = True
    grouped_df = dataframe.groupby(level=[0,1])
    inputs_batch_list = []
    labels_batch_list = []
    for index, campaign_period_data in grouped_df:
      labels = self.process_label(campaign_period_data)
      inputs = self.process_input(campaign_period_data)

      inputs_batch_list.append(inputs)
      labels_batch_list.append(labels)
      if len(inputs_batch_list) == self.batch_size:
        inputs_batch = tf.concat(inputs_batch_list, 0)
        labels_batch = tf.concat(labels_batch_list, 0)
        yield inputs_batch, labels_batch
        inputs_batch_list = []
        labels_batch_list = []

将该数据集传入Vertex AI Endpoint中的模型时，出现以下错误：

• 调用prediction_client.predict(endpoint=endpoint_name, instances = wide_window.test)时，报错ValueError: Unable to coerce value: <tf.Tensor: shape=(32, 7, 69), dtype=float32, numpy=...>
• 调用prediction_client.predict(endpoint=endpoint_name, instances = [wide_window.test])时，报错ValueError: Unable to coerce value: <FlatMapDataset element_spec=...>

尝试将FlatMapDataset转换为TFRecord但遇到困难，而本地加载模型后，使用相同数据集调用以下代码可正常得到预测结果：

fastLoad = True
if fastLoad:
    lstm_model = tf.keras.models.load_model('model/my_model')

preds_test = lstm_model.predict(wide_window.test)

解决方法 #

1. 明确Vertex AI Endpoint的输入规则 #

Vertex AI的prediction_client.predict不直接支持tf.data.Dataset或批量Tensor作为instances参数，它要求每个实例是可序列化为JSON的格式（如列表、字典），且每个实例对应模型的单条输入样本（而非批量数据）。

2. 转换Dataset为符合要求的输入格式 #

方法一：从现有Dataset中提取单样本列表

从wide_window.test中取出批量数据，拆分为单个样本并转为Python列表：

# 提取一个批次的输入数据（仅需输入部分，标签可忽略）
for batch_inputs, _ in wide_window.test.take(1):
    # 将Tensor转为numpy数组，再拆分为单个样本的列表（每个样本shape为(7,69)）
    instances = batch_inputs.numpy().tolist()

# 调用Vertex AI预测接口
response = prediction_client.predict(endpoint=endpoint_name, instances=instances)

方法二：修改Dataset生成逻辑，直接输出单样本

调整make_generator_dataset方法，直接生成单个样本而非攒批量，方便后续直接收集实例：

def make_generator_dataset(self, dataframe):
    grouped_df = dataframe.groupby(level=[0,1])
    for index, campaign_period_data in grouped_df:
      inputs = self.process_input(campaign_period_data)
      # 直接输出单样本的列表格式（符合JSON序列化要求）
      yield inputs.numpy().tolist()

然后生成Dataset并收集所有实例：

# 生成单样本Dataset并转为列表
instances = list(wide_window.test.as_numpy_iterator())
# 调用预测
response = prediction_client.predict(endpoint=endpoint_name, instances=instances)

3. 验证模型输入签名（可选） #

若仍报错，检查本地模型的输入签名是否与传入实例匹配：

model = tf.keras.models.load_model('model/my_model')
print(model.input_shape)  # 预期输出为(None,7,69)，对应单样本形状为(7,69)

确保每个实例的形状与模型输入签名一致（去掉批量维度None）。

4. 理解本地与云端预测的差异 #

本地model.predict支持直接传入Dataset，是因为Keras内部会自动遍历Dataset提取批次数据；但Vertex AI预测客户端不做此处理，必须显式传入单个实例的集合。

内容的提问来源于stack exchange，提问作者filipe