Model Optimization using Quantization

 Data science models based on deep neural networks become huge in terms of the number of parameters and weights.  This effectively increases the model size which has a direct impact on model load and inference time.  In this post, I will be exploring model quantization to decrease model load time and model inference time for various Image/Text models. More specifically, I would like to benchmark model performance and in specific I want to compare FP32 weights vs FP16/INT8 weights for some of the internal layers in the Neural Network. 

My first attempt is to use tflite and try optimizing the model.  I will use the galaxy image classification data to see if there is any drop in accuracy when compared to the base-model. for this purpose, I will use mobilenet and train it with  custom head for classifying a set of 100-200 images. From the research paper at https://arxiv.org/abs/2104.11849, I understand that quantization of mobile nets might not give us performance results as expected. 

Also, Mobilenet uses depthwise-separable convolutions and hence might not provide a significant improvement over the base model.   however, the intention is to check how much of an improvement in performance the tflite model provides over the base model and check for any accuracy drop along with exploring how the weights map from the base model to the tflite format.  Below is a hacked out script from one of my coursera assignments and tensorflow documentation to train mobilenet using the galaxy classification data.  I also want to get the embeddings of images as a feature extractor and hence will add the flattened outputs just before the Dense layer to model outputs

<pre><code>
import sys, os
import pathlib
import numpy as np
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers


batch_size, img_height, img_width = 32, 424, 424
data_dir = "./mergerNotMerger"

def load_data(data_dir):
    train_ds = tf.keras.utils.image_dataset_from_directory(
                data_dir,
                validation_split=0.2,
                subset=training,
                seed=123,
                image_size=(img_height, img_width)
                batch_size=batch_size
              )
    val_ds = tf.keras.utils.image_dataset_from_directory(
                data_dir,
                validation_split=0.2,
                subset="validation"
                seed=123,
                image_size=(img_height, img_width),
                batch_size=batch_size
             )
    return train_ds, val_ds
    
def get_class_names(ds):
    class_names = ds.class_names
    return class_names
  
 def get_model_mobile_netv2(img_height, img_width, num_classes):
     img_shape = (img_height, img_width, 3)
     base_model = tf.keras.applications.MobileNetV2(
                     input_shape=img_shape, 
                     include_top=False,  
                     weights='imagenet'
                  )
    base_model.trainable=False
    outputs = []
     x = tf.keras.layers.Conv2D(filter=32,, kernel_size=3, activation='relu')(base_model.output)
     x = tf.keras.layers.Dropout(0.5)(x)
     x = tf.keras.layers.MaxPool2D(pool_size=(2,2))(x)
     x = tf.keras.layers.Flatten()(x)
     outputs.append(x)
     x = tf.keras.layers.Dense(units=num_classes)(x)
     outputs.insert(0, x)
     model = tf.keras.model(base_model.input, outputs=outputs)
     return model

def train_model(model, train_ds, val_ds, epochs):
    model.compile(
            optimizer='adam', 
            loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
            metrics=['accuracy']
    )
    model.summary()
    history = model.fit(
                    train_ds,
                    validation_data=val_ds,
                    epochs=epochs
              )
    return model, history
 

def convert_model_tflite(model, option, train_images=None):
    converter = tf.lite.TFLiteConverter.from_keras_model(model)
    if option == 'dynamic_quantization':
        converter.optimizations = [tf.lite.Optimize.DEFAULT]
    elif option == 'float16':
        converter.optimizations = [tf.lite.Optimize.DEFAULT]
        converter.target_spec.supported_types = [tf.float16]
    elif option == 'int8':
        def rep_data_gen():
            for input_value in tf.data.Dataset.from_tensor_slices(train_images).batch(1).take(100):
                yield [input_value]
        converter = tf.lite.TFLiteConverter.from_keras_model(model)
        converter.optimizations = [tf.lite.Optimize.DEFAULT]
        converter.representative_dataset = rep_data_gen
        converter.target_spec_supported_ops = [tf.lite.OpsSet.EXPERIMENTAL_TFLITE_BUILTINS_ACTIVATIONS_INT16_WEIGHTS_INT8]
        converter.inference_input_type = tf.float32
        converter.inference_output_type = tf.int8
    lite_model = converter.convert()
    return lite_model
    
  
 def train_mobilenet_without_quantization(model, train_ds):
     for train_images_subset, train_labels_subset in train_ds:
         model.compile(
                 optimizer='adam',
                 loss=tf.keras.loss.SparseCategoricalCrossentropy(from_logits=True),
                 metrics=['accuracy']
         )
         history = model.fit(train_images_subset, train_labels_subset, batch_size=4, epochs=10, validation_split=0.1)
         return model, history
 
 def save_model(model, model_output_path)
     with open(model_output_path, 'wb') as f:
         f.write(model)
 
 def main(subscript):
     train_ds, val_ds = load_data(data_dir)
     class_names = get_class_names(train_ds)
     num_classes  = len(class_names)
     train_ds = train_ds.cache().shuffle(100).prefetch(buffer_size=tf.data.AUTOTUNE)
     val_ds = val_ds.cache().prefetch(buffer_size=tf.data.AUTOTUNE)
     normalization_layer = layers.Rescaling(1./255)
     normalized_ds = train_ds.map(lambda x, y: (normalization_layer(x), y)
     image_batch, labels_batch = next(iter(normalized_ds)
     first_image = image_batch[0]
    
     model_no_quant_path = 'gal_classification_no_quant_{}'.format(subscript)
     model_no_quant = get_model_mobile_netv2(img_height, img_width, num_classes)
     train_mobilenet_without_quantization(model_no_quant, train_ds)
     model_no_quant.save(model_no_quant_path)

    model_quant_path = 'galaxy_model_{}.tflite'.format(subscript)

    for  t, l in train_ds.take(10):
         model_int8 = convert_model_tflite(model_no_quant, 'int8', t)
         save_model(model_int8, model_quant_path)
         break
    return model_no_quant_path, model_quant_path, class_names

def classify_images(img, model_file_path, class_names):
    img_a = tf.keras.utils.load_img(img, target_size = (img_height, img_width)
    img_array = tf.keras.utils.img_to_array(img_a)
    img_array = tf.expand_dims(img_array, 0)
    interpreter = tf.lite.Interpreter(model_path=model_file_path)
    signature = interpreter.get_signature_list()
    
    classify_lite = interpreter.get_signature_runner('serving_default')
    predictions_lite = classify_lite(input_1=img_array)
    return predictions_lite


training = True if len(sys.argv)>=2 else False        
if training:
    if len(sys.argv)<2:
        from datetime import datetime
        subscript = datetime.now().isoformat()
    else:
        subscript = sys.argv[1]
        model_no_quant_path, model_quant_path = main(subscript)
        classify_images(model_quant_path, class_names)
</pre></code>


Exploring quantized parameters for each layer

import tensorflow as tf
interpreter = tf.lite.Interpreter(model_path='galaxy_model.tflite')
interpreter.allocate_tensors()
tensors = interpreter.get_tensor_details()
print(tensors[10]['quantization_parameters']['scales']
print(tensors[10]['quantization_parameters']['zero_point']




Using the classify method from the above code, I intend to explore the quantized feature outputs of various images and use them for a vector similarity search. 


     
                
        
 
    
                 

  

Comments

Popular posts from this blog

Java 21 Virtual threads

Exploring MemGraph & MAGE