Flask API for Multi-class classification using Keras
How to create a production-ready API using Flask for a deep learning model
In this article, you will learn how to create a modularized production-ready library for a deep learning model using the Fashion MNIST dataset built using Keras and Flask.
What are the key features to have a production-ready code for a deep learning model?
Features of production-ready deep learning code
- Exception Handling to monitor errors and to understand the flow of code in case of errors.
- Logging can be set to different levels like debug, info, warning, error, or critical . In production, we should be set the logging level to log warnings, errors, and critical information only.
- Version control for code using GitLab
- Code comments are very vital to understand the code
- Code optimization for efficient memory usage and computation
- Containerize the deep learning model code and all its dependent libraries.
We will learn about Docker and how to containerize the deep learning model in a different article.
Create Config class
We need to parametrize some common attributes used for fine-tuning deep learning models and can be done using a Config class. We also add parameters subjected to change based on input data so that we can keep the model genric.
- Input dataset related parameters : image dimensions — height and width, dataset size, and labels of the classes to be identified
- Model fine-tuning parameters : optimizers, learning rate for the optimizer, no. of epochs and batch_size. We can also include drop out rate, no. of input nodes, no. of hidden layers, etc.
- Files to store information : Weight file for saving trained weights and log filename for logging
You can add/remove config parameters based on your requirements. These config parameters are used across all different operations in a deep learning model.
class Config(object):
IMAGE_HEIGHT=28
IMAGE_WIDTH=28
DATA_SIZE=1000
CLASS_NAME='T-shirt/top', 'Trouser', 'Pullover', 'Dress', 'Coat','Sandal', 'Shirt', 'Sneaker', 'Bag', 'Ankle boot'
WEIGHT_FILENAME='FashionMNIST.h5'
LOG_FILENAME='LOG_FASHION.TXT'
EPOCHS=100
OPTIMIZER='RMSProp'
LEARNING_RATE=0.001
BATCH_SIZE=64
def __init__(self):
self.IMAGE_DIM = (self.IMAGE_HEIGHT, self.IMAGE_WIDTH)
Build a Classification Model
Create a class classFashionMNIST for handling different aspects of the deep learning model. The class will have methods to
- Normalize the data
- Build a deep learning model
- Train the model
- Make predictions using the model
- Display the image from the dataset
- Find the actual class of the image from the dataset
Logging and Exception handling is an integral part of the code to make it robust.
The init() is for setting parameters that we need for different operations of the class, like the height, width for the image, dataset size, and the different classes we want to predict. The init() is also used for reading the Fashion MNIST dataset for training as well as validation,
def __init__(self, height, width, data_size, class_name):
try:
self.height= height
self.width=width
self.data_size=data_size
self.class_names =list(class_name)
(self.train_images, self.train_labels), (self.test_images, self.test_labels) = tf.keras.datasets.fashion_mnist.load_data()
self.test_data= self.test_images
except:
logging.error("Error in init %s", sys.exc_info())
Normalize the dataset
As the pixel intensity in the images are between 1–255, normalize the images by scaling the values between 0 and 1
def normalize_data(self):
try:
logging.info("Normalizing data")
# load train and test images and labels based on data size
self.train_labels = self.train_labels[:self.data_size]
self.test_labels = self.test_labels[:self.data_size]
#Normalize the data
self.train_images = self.train_images[:self.data_size].astype('float32') / 255
self.test_images = self.test_images[:self.data_size].astype('float32') / 255
logging.info("Rshaping data")
# Reshape the data
self.train_images = self.train_images.reshape((self.train_images.shape[0], self.width, self.height,1))
self.test_images = self.test_images.reshape((self.test_images.shape[0], self.width, self.height,1))
except:
logging.error("Error", sys.exc_info())
Create a deep learning classification model
We pass the optimizer and the learning rate set in the configuration file for compiling the model. As the deep learning model is a multi-class classification, the loss function used is sparse_categorical_crossentropy. If you are doing a binary classification model, then use binary_crossentropy as the loss function.
def create_model(self, optimizer, learning_rate):
try:
logging.info("Creatig model")
model = tf.keras.Sequential()
# Must define the input shape in the first layer of the neural network
model.add(tf.keras.layers.Conv2D(filters=64, kernel_size=2, padding='same', activation='relu', input_shape=(28,28,1)))
model.add(tf.keras.layers.MaxPooling2D(pool_size=2))
model.add(tf.keras.layers.Dropout(0.3))
model.add(tf.keras.layers.Conv2D(filters=32, kernel_size=2, padding='same', activation='relu'))
model.add(tf.keras.layers.MaxPooling2D(pool_size=2))
model.add(tf.keras.layers.Dropout(0.3))
model.add(tf.keras.layers.Flatten())
model.add(tf.keras.layers.Dense(256, activation='relu'))
model.add(tf.keras.layers.Dropout(0.5))
model.add(tf.keras.layers.Dense(10, activation='softmax'))
logging.info("Model Created")
# creating optimizer based on the config
opt= self.get_optimizer(optimizer, learning_rate)
#Compiling the model
model.compile(loss='sparse_categorical_crossentropy',
optimizer=opt,
metrics=['accuracy'])
logging.info(" Model Compiled")
except:
logging.error(" Error during Model Creation - %s", sys.exc_info())
finally:
return model
Setting the optimizer
Handled three popular optimizers: Adam, SGD, and RMSProp . RMSProp is the default optimizer, and the default learning rate is set to 0.001. We can other optimizers like Momentum, Nesterov, Adagrad, Adadelta.
def get_optimizer(self,optimizer_name='RMSProp', learning_rate=0.001):
try:
if optimizer_name=='Adam':
optimizer = optimizers.Adam(learning_rate=learning_rate, beta_1=0.9, beta_2=0.999, amsgrad=False)
elif optimizer_name=='SGD':
optimizer = optimizers.SGD(lr=learning_rate, momentum=0.9)
elif optimizer_name=='RMSProp':
optimizer = optimizers.RMSprop()
logging.info("Optimizer created %s", optimizer_name)
return optimizer
except:
logging.error(" Error during visualization - %s", sys.exc_info())
Training the Model
Create the model, normalize the data, and finally train the data on train images and train labels.
If the accuracy is less than 0.8 or validation accuracy is less than 0.7, then we are raising a warning in the log file for notifying the team that the model may need retraining.
def train_model(self,filename, epochs, optimizer, learning_rate, batch_size):
try:
model = self.create_model(optimizer, learning_rate)
logging.info("Model created ")
logging.info("Normalizing the data")
self.normalize_data()
logging.info(self.train_images.shape)
logging.info("Training started")
history=model.fit(self.train_images,
self.train_labels,
batch_size=batch_size,
epochs=epochs,
validation_data=(self.test_images,self.test_labels))
logging.info(" Training finished")
acc= np.average(history.history['acc'])
val_acc=np.average(history.history['val_acc'])
logging.info(" Model accurcay on train images : {:5.2f}".format(acc))
logging.info("Accurcay too low for val {:5.2f}".format(val_acc))
model.save(filename)
logging.info("Model saved %s", filename)
if acc <.8 or val_acc<0.7:
logging.warn("Accurcay too low {:5.2f}".format(acc) )
logging.warn("Accurcay too low for val {:5.2f}".format(val_acc))
return history, model
except:
logging.error(" Error during Model Creation - %s", sys.exc_info())
Predicting the data
To predict the data, we pass the index of the test image and the file that contains the trained weights.
def predict_data(self, test_image_num, filename):
try:
logging.info("Predciting the data for %d", test_image_num)
test_img = self.test_images[test_image_num].reshape((1, self.width, self.height,1))
test_img=test_img.astype('float32') / 255
model = tf.keras.models.load_model(filename)
logging.info("Loaded the trained weights from %s", filename)
pred= model.predict(test_img)
pred= np.argmax(pred)
logging.info("Predicted class %s",self.class_names[pred] )
return self.class_names[pred]
except:
logging.error(" Error during Model predcition - %s", sys.exc_info())
Actual class of the test image
As we have used FashoinMNIST, we know the class of the test images that we can use for comparing the output
def actual_data(self,test_image_num):
return self.class_names[self.test_labels[test_image_num]]
Full code for class classFashionMNIST
#Importing required libraries
import tensorflow as tf
import numpy as np
import matplotlib.pyplot as plt
import sys
import logging
from tensorflow.keras import optimizers# setting the random seed
np.random.seed(1)
tf.compat.v1.set_random_seed(1)class classFashionMNIST:
'''
Method Name: init
Functionality: initializes the class
Parameters: sets the height, width of the image, data size and class labels
'''
def __init__(self, height, width, data_size, class_name):
try:
self.height= height
self.width=width
self.data_size=data_size
self.class_names =list(class_name)
(self.train_images, self.train_labels), (self.test_images, self.test_labels) = tf.keras.datasets.fashion_mnist.load_data()
self.test_data= self.test_images
except:
logging.error("Error in init %s", sys.exc_info())
'''
Method Name: normalize data
Functionality: Normalizes the images pixel intensity values by
scaling pixel values to the range 0-1 to centering and
even standardizing the values.
Parameters: None
'''
def normalize_data(self):
try:
logging.info("Normalizing data")
# load train and test images and labels based on data size
self.train_labels = self.train_labels[:self.data_size]
self.test_labels = self.test_labels[:self.data_size]
#Normalize the data
self.train_images = self.train_images[:self.data_size].astype('float32') / 255
self.test_images = self.test_images[:self.data_size].astype('float32') / 255
logging.info("Rshaping data")
# Reshape the data
self.train_images = self.train_images.reshape((self.train_images.shape[0], self.width, self.height,1))
self.test_images = self.test_images.reshape((self.test_images.shape[0], self.width, self.height,1))
except:
logging.error("Error", sys.exc_info())'''
Method Name: create_mode
Functionality: Creates the deep learning model for multiclass classification
Parameters: optimizer - optimizers can be Adam, SGD or RMSProp
Learning_rate- learning rate of the optimizer
'''def create_model(self, optimizer, learning_rate):
try:
logging.info("Creatig model")
model = tf.keras.Sequential()
# Must define the input shape in the first layer of the neural network
model.add(tf.keras.layers.Conv2D(filters=64, kernel_size=2, padding='same', activation='relu', input_shape=(28,28,1)))
model.add(tf.keras.layers.MaxPooling2D(pool_size=2))
model.add(tf.keras.layers.Dropout(0.3))
model.add(tf.keras.layers.Conv2D(filters=32, kernel_size=2, padding='same', activation='relu'))
model.add(tf.keras.layers.MaxPooling2D(pool_size=2))
model.add(tf.keras.layers.Dropout(0.3))
model.add(tf.keras.layers.Flatten())
model.add(tf.keras.layers.Dense(256, activation='relu'))
model.add(tf.keras.layers.Dropout(0.5))
model.add(tf.keras.layers.Dense(10, activation='softmax'))
logging.info("Model Created")
# creating optimizer based on the config
opt= self.get_optimizer(optimizer, learning_rate)
#Compiling the model
model.compile(loss='sparse_categorical_crossentropy',
optimizer=opt,
metrics=['accuracy'])
logging.info(" Model Compiled")
except:
logging.error(" Error during Model Creation - %s", sys.exc_info())
finally:
return model'''
Method Name: train_model
Functionality: Trains the deep learning multiclass classification model
Parameters: filename : File o save the trained weights
epochs : No. of epcohs to train the model
optimizer - optimizers can be Adam, SGD or RMSProp
Learning_rate- learning rate of the optimizer
Batch_size - batch_size of the dataset to train the model
'''
def train_model(self,filename, epochs, optimizer, learning_rate, batch_size):
try:
model = self.create_model(optimizer, learning_rate)
logging.info("Model created ")
logging.info("Normalizing the data")
self.normalize_data()
logging.info(self.train_images.shape)
logging.info("Training started")
history=model.fit(self.train_images,
self.train_labels,
batch_size=batch_size,
epochs=epochs,
validation_data=(self.test_images,self.test_labels))
logging.info(" Training finished")
acc= np.average(history.history['acc'])
val_acc=np.average(history.history['val_acc'])
logging.info(" Model accurcay on train images : {:5.2f}".format(acc))
logging.info("Accurcay too low for val {:5.2f}".format(val_acc))
model.save(filename)
logging.info("Model saved %s", filename)
if acc <.8 or val_acc<0.7:
logging.warn("Accurcay too low {:5.2f}".format(acc) )
logging.warn("Accurcay too low for val {:5.2f}".format(val_acc))
return history, model
except:
logging.error(" Error during Model Creation - %s", sys.exc_info())'''
Method Name: predict_data
Functionality: predicts the data for multiclass classification model
Parameters: test_image_num - index of the test image that we want to predcit
filename : File containing the trained weights
'''
def predict_data(self, test_image_num, filename):
try:
logging.info("Predciting the data for %d", test_image_num)
test_img = self.test_images[test_image_num].reshape((1, self.width, self.height,1))
test_img=test_img.astype('float32') / 255
model = tf.keras.models.load_model(filename)
logging.info("Loaded the trained weights from %s", filename)
pred= model.predict(test_img)
pred= np.argmax(pred)
logging.info("Predicted class %s",self.class_names[pred] )
return self.class_names[pred]
except:
logging.error(" Error during Model predcition - %s", sys.exc_info())'''
Method Name: actual_data
Functionality: Retrives the actual class for the test image based on the index passed
Parameters: test_image_num - index of the test image that we want to predcit
'''
def actual_data(self,test_image_num):
return self.class_names[self.test_labels[test_image_num]]
'''
Method Name: get_optimizer
Functionality: Creates the optimizers based on passed parameter and learning rate
Parameters: Optimizer_name - optimizers can be Adam, SGD or RMSProp
Learning_rate- learning rate of the optimizer
'''
def get_optimizer(self,optimizer_name='RMSProp', learning_rate=0.001):
try:
if optimizer_name=='Adam':
optimizer = optimizers.Adam(learning_rate=learning_rate, beta_1=0.9, beta_2=0.999, amsgrad=False)
elif optimizer_name=='SGD':
optimizer = optimizers.SGD(lr=learning_rate, momentum=0.9)
elif optimizer_name=='RMSProp':
optimizer = optimizers.RMSprop()
logging.info("Optimizer created %s", optimizer_name)
return optimizer
except:
logging.error(" Error during visualization - %s", sys.exc_info())
Create the API using Flask
Read here for a basic understanding of how to create a basic API using Flask
Importing required libraries for using Flask API, classFashionMNIST, Fashion MNIST Config class, and logging class.
Create an application object by creating an instance of Flask to which we pass a predefined variable “__name__”, which is the name of our module.
from flask import Flask, jsonify, request from flask_restful import Api, Resource import numpy as np from Fashion_MNIST import classFashionMNIST from FashionConfig import Config as cfg import logging import absl.loggingapp=Flask(__name__)
We write two GET methods, one for making the prediction and one for retrieving the actual class
@app.route("/predict", methods=["GET"])
def predict():
pred =""
posted_data = request.get_json()
test_image_num=posted_data['test_image_num']
logging.info("In Predict")
model_filename=cfg.WEIGHT_FILENAME
pred= fashionMNISTclass.predict_data(test_image_num, model_filename)
return jsonify(pred)@app.route("/real", methods=["GET"])
def real():
data =""
posted_data = request.get_json()
test_image_num=posted_data['test_image_num']
data = fashionMNISTclass.actual_data(test_image_num)
return jsonify(data)
I have used TF 1.14, and hence we need to disable abseil-py so that the logs are not redirected to stderr. We set the log file based on the log filename specified in the Config parameters
Train the dataset as we load the API, hence wait for the predictions till the model is fully loaded and compiled
if __name__ == '__main__':
logging.root.removeHandler(absl.logging._absl_handler)
absl.logging._warn_preinit_stderr = False
logging.basicConfig(filename=cfg.LOG_FILENAME, filemode='a', format='%(filename)s-%(asctime)s %(msecs)d- %(process)d-%(levelname)s - %(message)s',
datefmt='%d-%b-%y %H:%M:%S %p' ,
level=logging.DEBUG)
fashionMNISTclass= classFashionMNIST(cfg.IMAGE_HEIGHT, cfg.IMAGE_WIDTH, cfg.DATA_SIZE, cfg.CLASS_NAME)
# noramlize the data
fashionMNISTclass.normalize_data()
# train the model
history, model = history, model = fashionMNISTclass.train_model(cfg.WEIGHT_FILENAME,
cfg.EPOCHS,
cfg.OPTIMIZER,
cfg.LEARNING_RATE,
cfg.BATCH_SIZE)
app.run(debug=True)
Testing the predict method using postman
Checking the actual class name
Full code available here
Conclusion:
You are now all set to write a production-ready code using Keras for binary or multi-class classification models. There are a lot of tweaks that can be done to the code, and some of them are mentioned in the article.
以上所述就是小编给大家介绍的《Flask API for Multi-class classification using Keras》,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对 码农网 的支持!
猜你喜欢:本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。
TCP/IP网络管理
亨特 / 电子工业 / 2006年3月1日 / 79.00元
本书是一本架设与维护TCP/IP网络的完整指南,无论你是在职的系统管理员,还是需要访问Internet的家用系统用户,都可从本书获得帮助。本书还讨论了高级路由协议(RIPv2、OSPF、BGP),以及实现这些协议的gated软件。对于各种重要的网络服务,如DNS,Apache,sendmail,Samba,PPP和DHCP,本书都提供了配置范例,以及相关的软件包与工具的语法参考。一起来看看 《TCP/IP网络管理》 这本书的介绍吧!