LipNet唇读模型训练报错：Attempt to convert a value (None)异常

问题背景 #

基于LipNet仓库训练唇读模型时，训练启动即报错Attempt to convert a value (None) with an unsupported type (<class 'NoneType'>)，无法完成1个epoch，调整库版本后问题依旧。

错误根源 #

1. 视频加载时未处理读取失败的帧，导致frame为None进入后续计算
2. tf.py_function返回的张量未明确指定形状和类型，TensorFlow无法构建稳定的数据管道
3. 回调函数中数据集迭代器耗尽后返回None，引发预测错误
4. CTC Loss中硬编码输入长度，与模型实际输出序列长度不匹配

修复方案 #

1. 修复视频加载函数

添加帧读取成功判断，确保输出固定75帧（匹配模型输入要求）：

def load_video(path:str) -> List[float]: 
    cap = cv2.VideoCapture(path)
    frames = []
    target_frames = 75
    for _ in range(int(cap.get(cv2.CAP_PROP_FRAME_COUNT))): 
        ret, frame = cap.read()
        if not ret:
            break
        frame = tf.image.rgb_to_grayscale(frame)
        frames.append(frame[190:236,80:220,:])
    cap.release()
    
    # 补全或截断到75帧
    while len(frames) < target_frames:
        frames.append(tf.zeros_like(frames[0]))
    frames = frames[:target_frames]
    
    mean = tf.math.reduce_mean(frames)
    std = tf.math.reduce_std(tf.cast(frames, tf.float32))
    return tf.cast((frames - mean), tf.float32) / std

2. 明确tf.py_function输出类型与形状

为返回的张量指定固定形状，避免TensorFlow推断失败：

def mappable_function(path:str):
    frames, alignments = tf.py_function(load_data, [path], (tf.float32, tf.int64))
    # 指定形状匹配模型输入和标签格式
    frames.set_shape((75, 46, 140, 1))
    alignments.set_shape((None,))
    return frames, alignments

3. 修复回调函数迭代器问题

每次epoch结束时重新获取迭代器，避免迭代器耗尽返回None：

class ProduceExample(tf.keras.callbacks.Callback): 
    def __init__(self, dataset) -> None: 
        self.dataset = dataset
    
    def on_epoch_end(self, epoch, logs=None) -> None:
        data_iter = self.dataset.as_numpy_iterator()
        data = data_iter.next()
        yhat = self.model.predict(data[0])
        # 动态获取输入长度，替换硬编码值
        input_length = [tf.shape(yhat)[1]] * tf.shape(yhat)[0]
        decoded = tf.keras.backend.ctc_decode(yhat, input_length, greedy=False)[0][0].numpy()
        for x in range(len(yhat)):           
            original = tf.strings.reduce_join(num_to_char(data[1][x])).numpy().decode('utf-8')
            # 过滤填充的空字符
            prediction = tf.strings.reduce_join(num_to_char(decoded[x][decoded[x] != 0])).numpy().decode('utf-8')
            print(f'Original: {original}')
            print(f'Prediction: {prediction}')
            print('~'*100)

4. 修正CTC Loss计算逻辑

动态计算实际标签长度，确保与模型输出序列长度匹配：

def CTCLoss(y_true, y_pred):
    batch_len = tf.cast(tf.shape(y_true)[0], dtype="int64")
    input_length = tf.cast(tf.shape(y_pred)[1], dtype="int64")
    label_length = tf.math.count_nonzero(y_true, axis=1, keepdims=True, dtype="int64")

    input_length = input_length * tf.ones(shape=(batch_len, 1), dtype="int64")

    loss = tf.keras.backend.ctc_batch_cost(y_true, y_pred, input_length, label_length)
    return loss

5. 调整数据管道padding设置

指定标签填充值为0（对应空字符映射）：

data = data.padded_batch(2, padded_shapes=([75,46,140,1],[40]), padding_values=(0.0, 0))

完整修复后代码 #

import os
import cv2
import tensorflow as tf
import numpy as np
from typing import List
from matplotlib import pyplot as plt
import imageio
import gdown
from tensorflow.keras.models import Sequential 
from tensorflow.keras.layers import Conv3D, LSTM, Dense, Dropout, Bidirectional, MaxPool3D, Activation, Reshape, SpatialDropout3D, BatchNormalization, TimeDistributed, Flatten
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.callbacks import ModelCheckpoint, LearningRateScheduler

url = 'https://drive.google.com/uc?id=1YlvpDLix3S-U8fd-gqRwPcWXAXm8JwjL'
output = 'data.zip'
gdown.download(url, output, quiet=False)
gdown.extractall('data.zip')

def load_video(path:str) -> List[float]: 
    cap = cv2.VideoCapture(path)
    frames = []
    target_frames = 75
    for _ in range(int(cap.get(cv2.CAP_PROP_FRAME_COUNT))): 
        ret, frame = cap.read()
        if not ret:
            break
        frame = tf.image.rgb_to_grayscale(frame)
        frames.append(frame[190:236,80:220,:])
    cap.release()
    
    while len(frames) < target_frames:
        frames.append(tf.zeros_like(frames[0]))
    frames = frames[:target_frames]
    
    mean = tf.math.reduce_mean(frames)
    std = tf.math.reduce_std(tf.cast(frames, tf.float32))
    return tf.cast((frames - mean), tf.float32) / std

vocab = [x for x in "abcdefghijklmnopqrstuvwxyz'?!123456789 "]
char_to_num = tf.keras.layers.StringLookup(vocabulary=vocab, oov_token="")
num_to_char = tf.keras.layers.StringLookup(
    vocabulary=char_to_num.get_vocabulary(), oov_token="", invert=True
)

print(
    f"The vocabulary is: {char_to_num.get_vocabulary()} "
    f"(size ={char_to_num.vocabulary_size()})"
)

def load_alignments(path:str) -> List[str]: 
    with open(path, 'r') as f: 
        lines = f.readlines() 
    tokens = []
    for line in lines:
        line = line.split()
        if line[2] != 'sil': 
            tokens = [*tokens,' ',line[2]]
    return char_to_num(tf.reshape(tf.strings.unicode_split(tokens, input_encoding='UTF-8'), (-1)))[1:]

def load_data(path: str): 
    path = bytes.decode(path.numpy())
    file_name = path.split('\\')[-1].split('.')[0]
    video_path = os.path.join('data','s1',f'{file_name}.mpg')
    alignment_path = os.path.join('data','alignments','s1',f'{file_name}.align')
    frames = load_video(video_path) 
    alignments = load_alignments(alignment_path)
    
    return frames, alignments

test_path = '.\\data\\s1\\bbal6n.mpg'
frames, alignments = load_data(tf.convert_to_tensor(test_path))
tf.strings.reduce_join([bytes.decode(x) for x in num_to_char(alignments.numpy()).numpy()])

def mappable_function(path:str):
    frames, alignments = tf.py_function(load_data, [path], (tf.float32, tf.int64))
    frames.set_shape((75, 46, 140, 1))
    alignments.set_shape((None,))
    return frames, alignments

data = tf.data.Dataset.list_files('./data/s1/*.mpg')
data = data.shuffle(500, reshuffle_each_iteration=False)
data = data.map(mappable_function)
data = data.padded_batch(2, padded_shapes=([75,46,140,1],[40]), padding_values=(0.0, 0))
data = data.prefetch(tf.data.AUTOTUNE)

train = data.take(450)
test = data.skip(450)

model = Sequential()
model.add(Conv3D(128, 3, input_shape=(75,46,140,1), padding='same'))
model.add(Activation('relu'))
model.add(MaxPool3D((1,2,2)))

model.add(Conv3D(256, 3, padding='same'))
model.add(Activation('relu'))
model.add(MaxPool3D((1,2,2)))

model.add(Conv3D(75, 3, padding='same'))
model.add(Activation('relu'))
model.add(MaxPool3D((1,2,2)))

model.add(TimeDistributed(Flatten()))

model.add(Bidirectional(LSTM(128, kernel_initializer='Orthogonal', return_sequences=True)))
model.add(Dropout(.5))

model.add(Bidirectional(LSTM(128, kernel_initializer='Orthogonal', return_sequences=True)))
model.add(Dropout(.5))

model.add(Dense(char_to_num.vocabulary_size()+1, kernel_initializer='he_normal', activation='softmax'))

model.summary()

def scheduler(epoch, lr):
    if epoch < 30:
        return lr
    else:
        return lr * tf.math.exp(-0.1)

def CTCLoss(y_true, y_pred):
    batch_len = tf.cast(tf.shape(y_true)[0], dtype="int64")
    input_length = tf.cast(tf.shape(y_pred)[1], dtype="int64")
    label_length = tf.math.count_nonzero(y_true, axis=1, keepdims=True, dtype="int64")

    input_length = input_length * tf.ones(shape=(batch_len, 1), dtype="int64")

    loss = tf.keras.backend.ctc_batch_cost(y_true, y_pred, input_length, label_length)
    return loss

class ProduceExample(tf.keras.callbacks.Callback): 
    def __init__(self, dataset) -> None: 
        self.dataset = dataset
    
    def on_epoch_end(self, epoch, logs=None) -> None:
        data_iter = self.dataset.as_numpy_iterator()
        data = data_iter.next()
        yhat = self.model.predict(data[0])
        input_length = [tf.shape(yhat)[1]] * tf.shape(yhat)[0]
        decoded = tf.keras.backend.ctc_decode(yhat, input_length, greedy=False)[0][0].numpy()
        for x in range(len(yhat)):           
            original = tf.strings.reduce_join(num_to_char(data[1][x])).numpy().decode('utf-8')
            prediction = tf.strings.reduce_join(num_to_char(decoded[x][decoded[x] != 0])).numpy().decode('utf-8')
            print(f'Original: {original}')
            print(f'Prediction: {prediction}')
            print('~'*100)

model.compile(optimizer=Adam(learning_rate=0.0001), loss=CTCLoss)

checkpoint_callback = ModelCheckpoint(os.path.join('models','checkpoint'), monitor='loss', save_weights_only=True) 
example_callback = ProduceExample(test)
schedule_callback = LearningRateScheduler(scheduler)

model.fit(train, validation_data=test, epochs=100, callbacks=[checkpoint_callback, schedule_callback, example_callback])

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