如何在 Tensorflow 服务模型中包含某些预处理步骤

Question

我已经建立了一个具有不同功能的模型。对于预处理，我主要使用feature_columns。例如，用于对 GEO 信息进行分桶或用于嵌入具有大量不同值的分类数据。此外，在使用 feature_columns 之前，我必须预处理我的两个特征：

专题“街道”

def __preProcessStreet(data, tokenizer=None):

    data['STREETPRO'] = data['STREET'].apply(lambda x: __getNormalizedString(x, ["gasse", "straße", "strasse", "str.", "g.", " "], False))

    if tokenizer == None:
        tokenizer = Tokenizer(split='XXX')
        tokenizer.fit_on_texts(data['STREETPRO'])

    street_tokenized = tokenizer.texts_to_sequences(data['STREETPRO'])

    data['STREETW'] = tf.keras.preprocessing.sequence.pad_sequences(street_tokenized, maxlen=1)

    return data, tokenizer

如您所见，我直接在加载的 Pandas 数据帧上执行了预处理步骤。之后，我在上述列的帮助下处理了这个新列：

def __getFutureColumnStreet(street_num_words):

    street_voc = tf.feature_column.categorical_column_with_identity(
        key='STREETW', num_buckets=street_num_words)

    dim = __getNumberOfDimensions(street_num_words)

    street_embedding = feature_column.embedding_column(street_voc, dimension=dim)

    return street_embedding

特征“NAME1”

NAME1 列的预处理步骤非常相似，除了我将 NAME1 字段拆分为两个不同的字段“NAME1W1”和“NAME1W2”，其中包括词汇表中最常见的两个词：

def __preProcessName(data, tokenizer=None):

    data['NAME1PRO'] = data['NAME1'].apply(lambda x: __getNormalizedString(x, ["(asg)", "asg", "(poasg)", "poasg"]))

    if tokenizer == None:
        tokenizer = Tokenizer()
        tokenizer.fit_on_texts(data['NAME1PRO'])

    name1_tokenized = tokenizer.texts_to_sequences(data['NAME1PRO'])

    name1_tokenized_pad = tf.keras.preprocessing.sequence.pad_sequences(name1_tokenized, maxlen=2, truncating='pre')

    data = pd.concat([data, pd.DataFrame(name1_tokenized_pad, columns=['NAME1W1', 'NAME1W2'])], axis=1)

    return data, tokenizer

之后我还使用了 feature_colums 来进行词嵌入：

def __getFutureColumnsName(name_num_words):

    namew1_voc = tf.feature_column.categorical_column_with_identity(
        key='NAME1W1', num_buckets=name_num_words)
    namew2_voc = tf.feature_column.categorical_column_with_identity(
        key='NAME1W2', num_buckets=name_num_words)

    dim = __getNumberOfDimensions(name_num_words)

    namew1_embedding = feature_column.embedding_column(namew1_voc, dimension=dim)
    namew2_embedding = feature_column.embedding_column(namew2_voc, dimension=dim)

    return (namew1_embedding, namew2_embedding)

型号

我正在使用 TensorFlow 的功能 API 来构建我的模型：

                print("start preprocessing...")
                feature_columns = feature_selection.getFutureColumns(data, args.zip, args.sc, bucketSizeGEO, False)
                feature_layer = tf.keras.layers.DenseFeatures(feature_columns, trainable=True)
                print("preprocessing completed")

…                

                            print("Step {}/{}".format(currentStep, stepNum))

                            feature_layer_inputs = feature_selection.getFeatureLayerInputs()
                            new_layer = feature_layer(feature_layer_inputs)
                            

                            for _ in range(numLayers):
                                new_layer = tf.keras.layers.Dense(numNodes, activation=tf.nn.swish, kernel_regularizer=regularizers.l2(reg), bias_regularizer=regularizers.l2(reg))(new_layer)
                                new_layer = tf.keras.layers.Dropout(dropRate)(new_layer) 

                            output_layer = tf.keras.layers.Dense(1, activation=tf.nn.sigmoid, kernel_regularizer=regularizers.l2(reg), bias_regularizer=regularizers.l2(reg))(new_layer)

                            model = tf.keras.Model(inputs=[v for v in feature_layer_inputs.values()], outputs=output_layer)

                            model.compile(optimizer=opt,
                                loss='binary_crossentropy',
                                metrics=['accuracy'])

                            paramString = "Arg-e{}-b{}-l{}-n{}-o{}-z{}-r{}-d{}".format(args.epoch, args.batchSize, numLayers, numNodes, opt, bucketSizeGEO, reg, dropRate)

                            log_dir = "logs\\neural\\" + paramString + "\\" + datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
                            tensorboard_callback = tf.keras.callbacks.TensorBoard(log_dir=log_dir, histogram_freq=1)

                            print("Start training with the following parameters:", paramString)

                            model.fit(train_ds,
                                    validation_data=val_ds,
                                    epochs=args.epoch,
                                    callbacks=[tensorboard_callback])

TensorFlow 服务

逻辑上，包含 Tokenizer 的两个预处理步骤不是模型的一部分，因此无法在服务期间进行处理，因此模型服务器的 POST 命令如下所示（在 Windows 上）：

curl -d "{"""instances""": [{"""NAME1W1""": [12], """NAME1W2""": [2032], """ZIP""": [""1120""], """STREETW""": [1180], """LONGITUDE""": 16.47, """LATITUDE""": 48.22, """AVIS_TYPE""": [""E""],"""ASG""": [0], """SC""": [""101""], """PREDICT""": [0]}]}" -X POST http://localhost:8501/v1/models/my_model:predict

所以目前我正在尝试找到一种方法，将这两个预处理步骤包含在我的模型中，以便 POST 命令看起来像这样：

curl -d "{"""instances""": [{"""NAME1""": [“”Max Mustermann””], """ZIP""": [""1120""], """STREET""": [Teststraße], """LONGITUDE""": 16.47, """LATITUDE""": 48.22, """AVIS_TYPE""": [""E""],"""ASG""": [0], """SC""": [""101""], """PREDICT""": [0]}]}" -X POST http://localhost:8501/v1/models/my_model:predict

但在模型内部具有相同的预处理步骤。

我尝试在数据集或 preprocessing layers 上使用地图函数，但没有成功，因为我不确定是否可以将它们与 future_columns 结合使用。我也尝试过类似这里提到的东西：https://keras.io/examples/structured_data/structured_data_classification_from_scratch/

Answer 1

我认为 TFX 转换组件是您所需要的。它不会成为您模型的一部分，而是您的管道的一部分。这样，您可以在不影响模型的情况下轻松地修改您将来想要的预处理转换。

该组件的主要功能是preprocessing_fn，这将是您要应用于输入的一系列转换。 TensorFlow 指南提供了更好的解释和教程供您尝试。

以下是一些参考资料：

• TFX 变换组件 (https://www.tensorflow.org/tfx/guide/transform)
• 与 Keras 一起使用 (https://www.tensorflow.org/tfx/guide/keras#tfx_components)
• 将其用作独立库 (https://www.tensorflow.org/tfx/transform/get_started)