Transformer模型训练时tf.data.Dataset TypeError问题排查求助

基于英法数据集构建Transformer模型，预处理后生成的数据集格式为((TensorSpec(shape=(None, 100), dtype=tf.int64), TensorSpec(shape=(None, 100), dtype=tf.int64)), TensorSpec(shape=(None, 100), dtype=tf.int64))。为让call()方法接收完整输入，将数据集处理为((x, y), y)格式并设置对应input_signature，但仍触发TypeError：If shallow structure is a sequence, input must also be a sequence，输入类型为<class 'tensorflow.python.framework.ops.Tensor'>。尝试过匹配数据集element_spec与input_signature、拆分输入等方法，均未解决问题。

预处理代码 #

def encode(self, lang1, lang2):
    # Add <sos>, <eos> token
    lang1 = [self.tokenizer_src.vocab_size] + self.tokenizer_src.encode(
        lang1.numpy()) + [self.tokenizer_src.vocab_size + 1]
    
    lang2 = [self.tokenizer_trg.vocab_size] + self.tokenizer_trg.encode(
        lang2.numpy()) + [self.tokenizer_trg.vocab_size + 1]
    
    return lang1, lang2
  
def tf_encode(self, src, trg):
    result_src, result_trg = tf.py_function(self.encode, [src, trg], [tf.int64, tf.int64])

    result_src.set_shape([None])
    result_trg.set_shape([None])

    return result_src, result_trg
  
def make_batches(self, dataset, buffer_size, batch_size):
    dataset = (
        dataset
        .cache()
        .shuffle(buffer_size)
        .map(self.tf_encode, num_parallel_calls=tf.data.AUTOTUNE)
        .padded_batch(batch_size, padded_shapes=([self.maxlen], [self.maxlen])))
    dataset = dataset.map(lambda x, y: ((x, y), y),
                          num_parallel_calls=tf.data.AUTOTUNE) 
    return dataset
  
def tokenization(self, buffer_size, batch_size):
    # Tokenize datasets & Make batches
    train_dataset = self.make_batches(self.train_dataset, buffer_size, batch_size)
    val_dataset = self.make_batches(self.val_dataset, buffer_size, batch_size)
    test_dataset = self.make_batches(self.test_dataset, buffer_size, batch_size)

    return train_dataset, val_dataset, test_dataset

模型call方法代码 #

@tf.function(input_signature=[
    ((tf.TensorSpec(shape=(None, MAXLEN), dtype=tf.int64),
     tf.TensorSpec(shape=(None, MAXLEN), dtype=tf.int64)),
     tf.TensorSpec(shape=(None, MAXLEN), dtype=tf.int64)), 
     tf.TensorSpec(shape=(), dtype=tf.bool)
     ])
def call(self, inputs, training=False):
    inp, tar = inputs

    enc_padding_mask, look_ahead_mask, dec_padding_mask = self.create_masks(inp, tar)

    enc_output = self.encoder(inp, training, enc_padding_mask)  # (batch_size, inp_seq_len, d_model)

    dec_output, attention_weights = self.decoder(
        tar, enc_output, training, look_ahead_mask, dec_padding_mask)
    # |dec_output| = (batch_size, tar_seq_len, d_model)

    final_output = self.final_layer(dec_output)  # (batch_size, tar_seq_len, target_vocab_size)

    return final_output, attention_weights

报错信息 #

TypeError: Exception encountered when calling layer 'transformer' (type Transformer).
    
    If shallow structure is a sequence, input must also be a sequence. Input has type: <class 'tensorflow.python.framework.ops.Tensor'>.
    
    Call arguments received by layer 'transformer' (type Transformer):
      • inputs=('tf.Tensor(shape=(None, 100), dtype=int64)', 'tf.Tensor(shape=(None, 100), dtype=int64)')
      • training=True

问题分析与解决方案 #

核心问题 #

input_signature定义与call方法的参数解析逻辑不匹配：

• input_signature的第一个参数是((输入张量对), 标签张量)的嵌套结构，但call方法错误地将inputs直接拆分为(inp, tar)，忽略了外层包含的标签张量部分。
• 数据集传入的实际结构是((x,y), y)，但call方法只提取了第一部分(x,y)，导致TensorFlow对输入结构的校验失败。

修复步骤 #

1. 调整call方法的参数解析逻辑
   先拆分出输入对和标签，再从输入对里提取编码器和解码器输入：

   @tf.function(input_signature=[
       ((tf.TensorSpec(shape=(None, MAXLEN), dtype=tf.int64),
        tf.TensorSpec(shape=(None, MAXLEN), dtype=tf.int64)),
        tf.TensorSpec(shape=(None, MAXLEN), dtype=tf.int64)), 
        tf.TensorSpec(shape=(), dtype=tf.bool)
        ])
   def call(self, inputs, training=False):
       # 先拆分输入结构：((inp, tar), labels)
       encoder_decoder_inputs, labels = inputs
       inp, tar = encoder_decoder_inputs
       
       # 后续逻辑保持不变
       enc_padding_mask, look_ahead_mask, dec_padding_mask = self.create_masks(inp, tar)
       enc_output = self.encoder(inp, training, enc_padding_mask)
       dec_output, attention_weights = self.decoder(
           tar, enc_output, training, look_ahead_mask, dec_padding_mask)
       final_output = self.final_layer(dec_output)
       
       return final_output, attention_weights
   

2. 验证输入结构一致性
   打印数据集的element_spec确认与input_signature完全匹配：

   print(train_dataset.element_spec)
   

   输出应与input_signature的第一个元素结构一致，确保TensorFlow能正确解析输入。

3. 可选：简化数据集格式（若无需在call中使用标签）
   如果训练时标签不需要传入call方法（仅用于损失计算），可以修改数据集处理逻辑：

   # 修改make_batches中的map操作
   dataset = dataset.map(lambda x, y: (x, y), num_parallel_calls=tf.data.AUTOTUNE)
   

   同时更新input_signature和call方法：

   @tf.function(input_signature=[
       tf.TensorSpec(shape=(None, MAXLEN), dtype=tf.int64),
       tf.TensorSpec(shape=(None, MAXLEN), dtype=tf.int64),
       tf.TensorSpec(shape=(), dtype=tf.bool)
   ])
   def call(self, inp, tar, training=False):
       # 直接使用inp和tar，无需拆分外层结构
       enc_padding_mask, look_ahead_mask, dec_padding_mask = self.create_masks(inp, tar)
       enc_output = self.encoder(inp, training, enc_padding_mask)
       dec_output, attention_weights = self.decoder(
           tar, enc_output, training, look_ahead_mask, dec_padding_mask)
       final_output = self.final_layer(dec_output)
       
       return final_output, attention_weights
   

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