This notebook will show you how to modify a Keras model to perform keyed predictions or forward features through with the prediction. This it the companion code for this blog.
Sometimes you'll have a unique instance key that is associated with each row and you want that key to be output along with the prediction so you know which row the prediction belongs to. You'll need to add keys when executing distributed batch predictions with a service like Cloud AI Platform batch prediction. Also, if you're performing continuous evaluation on your model and you'd like to log metadata about predictions for later analysis. There are also use-cases for forwarding a particular feature from your model out with the output, for example performing evaluation on certain slices of data.
In [1]:
import numpy as np
import tensorflow as tf
from tensorflow import keras
In [2]:
tf.__version__
Out[2]:
In [3]:
# Set GCP configs if using Cloud AI Platform
import os
PROJECT = "your-gcp-project-here" # REPLACE WITH YOUR PROJECT NAME
REGION = "us-central1" # REPLACE WITH YOUR BUCKET REGION e.g. us-central1
BUCKET = "your-gcp-bucket-here"
# Do not change these
os.environ["PROJECT"] = PROJECT
os.environ["REGION"] = REGION
os.environ["BUCKET"] = PROJECT # DEFAULT BUCKET WILL BE PROJECT ID
if PROJECT == "your-gcp-project-here":
print("Don't forget to update your PROJECT name! Currently:", PROJECT)
We will use a straightforward keras use case with the fashion mnist dataset to demonstrate building a model and then adding support for keyed predictions. More here on the use case: https://colab.sandbox.google.com/github/tensorflow/docs/blob/master/site/en/tutorials/keras/classification.ipynb
In [4]:
fashion_mnist = keras.datasets.fashion_mnist
(train_images, train_labels), (test_images, test_labels) = fashion_mnist.load_data()
In [5]:
# Scale down dataset
train_images = train_images / 255.0
test_images = test_images / 255.0
In [6]:
# Build and traing model
from tensorflow.keras import Sequential, Input
from tensorflow.keras.layers import Dense, Flatten
model = Sequential([
Input(shape=(28,28), name="image"),
Flatten(input_shape=(28, 28), name="flatten"),
Dense(64, activation='relu', name="dense"),
Dense(10, activation='softmax', name="preds"),
])
model.compile(optimizer='adam',
loss=tf.keras.losses.SparseCategoricalCrossentropy(),
metrics=['accuracy'])
# Only training for 1 epoch, we are not worried about model performance
model.fit(train_images, train_labels, epochs=1, batch_size=32)
Out[6]:
In [7]:
# Create test_image in shape, type that will be accepted as Tensor
test_image = np.expand_dims(test_images[0],0).astype('float32')
model.predict(test_image)
Out[7]:
Now save the model using tf.saved_model.save() into SavedModel format, not the older Keras H5 Format. This will add a serving signature which we can then inspect. The serving signature indicates exactly which input names and types are expected, and what will be output by the model
In [8]:
MODEL_EXPORT_PATH = './model/'
tf.saved_model.save(model, MODEL_EXPORT_PATH)
In [9]:
!saved_model_cli show --tag_set serve --signature_def serving_default --dir {MODEL_EXPORT_PATH}
In [10]:
# Load the model from storage and inspect the object types
loaded_model = tf.keras.models.load_model(MODEL_EXPORT_PATH)
loaded_model.signatures
Out[10]:
In [11]:
loaded_model
Out[11]:
It's worth noting that original model did not have serving signature until we saved it and is a slightly different object type:
In [12]:
model
Out[12]:
In [13]:
# Uncomment and expect an error since different object type
# model.signatures
In [14]:
inference_function = loaded_model.signatures['serving_default']
print(inference_function)
In [15]:
result = inference_function(tf.convert_to_tensor(test_image))
print(result)
In [16]:
# Matches serving signature
result['preds']
Out[16]:
Now we'll create a new serving function that accepts and outputs a unique instance key. We use the fact that a Keras Model(x) call actually runs a prediction. The training=False parameter is included only for clarity. Then we save the model as before but provide this function as our new serving signature.
In [17]:
@tf.function(input_signature=[tf.TensorSpec([None], dtype=tf.string),tf.TensorSpec([None, 28, 28], dtype=tf.float32)])
def keyed_prediction(key, image):
pred = loaded_model(image, training=False)
return {
'preds': pred,
'key': key
}
In [18]:
# Resave model, but specify new serving signature
KEYED_EXPORT_PATH = './keyed_model/'
loaded_model.save(KEYED_EXPORT_PATH, signatures={'serving_default': keyed_prediction})
In [19]:
!saved_model_cli show --tag_set serve --signature_def serving_default --dir {KEYED_EXPORT_PATH}
In [20]:
keyed_model = tf.keras.models.load_model(KEYED_EXPORT_PATH)
In [21]:
# Change 'flatten_input' to 'image' after b/159022434
keyed_model.predict({
'flatten_input': test_image,
'key': tf.constant("unique_key")}
)
# keyed_model.predict(test_image)
Out[21]:
Sometimes it is useful to leave both signatures in the model definition so the user can indicate if they are performing a keyed prediction or not. This can easily be done with the model.save() method as before.
In general, your serving infrastructure will default to 'serving_default' unless otherwise specified in a prediction call. Google Cloud AI Platform online and batch prediction support multiple signatures, as does TFServing.
In [22]:
# Using inference_function from earlier
DUAL_SIGNATURE_EXPORT_PATH = './dual_signature_model/'
loaded_model.save(DUAL_SIGNATURE_EXPORT_PATH, signatures={'serving_default': keyed_prediction,
'unkeyed_signature': inference_function})
In [23]:
# Examine the multiple signatures
!saved_model_cli show --tag_set serve --dir {DUAL_SIGNATURE_EXPORT_PATH}
In [24]:
# Default signature
!saved_model_cli show --tag_set serve --signature_def serving_default --dir {DUAL_SIGNATURE_EXPORT_PATH}
In [25]:
# Alternative unkeyed signature
!saved_model_cli show --tag_set serve --signature_def unkeyed_signature --dir {DUAL_SIGNATURE_EXPORT_PATH}
In [ ]:
os.environ["MODEL_LOCATION"] = DUAL_SIGNATURE_EXPORT_PATH
In [26]:
%%bash
MODEL_NAME=fashion_mnist
MODEL_VERSION=v1
TFVERSION=2.1
# REGION and BUCKET and MODEL_LOCATION set earlier
# create the model if it doesn't already exist
modelname=$(gcloud ai-platform models list | grep -w "$MODEL_NAME")
echo $modelname
if [ -z "$modelname" ]; then
echo "Creating model $MODEL_NAME"
gcloud ai-platform models create ${MODEL_NAME} --regions $REGION
else
echo "Model $MODEL_NAME already exists"
fi
# delete the model version if it already exists
modelver=$(gcloud ai-platform versions list --model "$MODEL_NAME" | grep -w "$MODEL_VERSION")
echo $modelver
if [ "$modelver" ]; then
echo "Deleting version $MODEL_VERSION"
yes | gcloud ai-platform versions delete ${MODEL_VERSION} --model ${MODEL_NAME}
sleep 10
fi
echo "Creating version $MODEL_VERSION from $MODEL_LOCATION"
gcloud ai-platform versions create ${MODEL_VERSION} \
--model ${MODEL_NAME} --origin ${MODEL_LOCATION} --staging-bucket gs://${BUCKET} \
--runtime-version $TFVERSION
In [27]:
# Create keyed test_image file
with open("keyed_input.json", "w") as file:
print(f'{{"image": {test_image.tolist()}, "key": "image_id_1234"}}', file=file)
In [28]:
# Single online keyed prediction, --signature-name is not required since we're hitting the default but shown for clarity
!gcloud ai-platform predict --model fashion_mnist --json-instances keyed_input.json --version v1 --signature-name serving_default
In [29]:
# Create unkeyed test_image file
with open("unkeyed_input.json", "w") as file:
print(f'{{"image": {test_image.tolist()}}}', file=file)
In [30]:
# Single online unkeyed prediction using alternative serving signature
!gcloud ai-platform predict --model fashion_mnist --json-instances unkeyed_input.json --version v1 --signature-name unkeyed_signature
In [31]:
# Create Data files:
import shutil
DATA_DIR = './batch_data'
shutil.rmtree(DATA_DIR, ignore_errors=True)
os.makedirs(DATA_DIR)
# Create 10 files with 10 images each
for i in range(10):
with open(f'{DATA_DIR}/keyed_batch_{i}.json', "w") as file:
for z in range(10):
key = f'key_{i}_{z}'
print(f'{{"image": {test_images[z].tolist()}, "key": "{key}"}}', file=file)
In [32]:
%%bash
gsutil -m cp -r ./batch_data gs://$BUCKET/
This following batch prediction job took me 8-10 minutes, most of the time spent in infrastructure spin up.
In [33]:
%%bash
DATA_FORMAT="text" # JSON data format
INPUT_PATHS="gs://${BUCKET}/batch_data/*"
OUTPUT_PATH="gs://${BUCKET}/batch_predictions"
MODEL_NAME='fashion_mnist'
VERSION_NAME='v1'
now=$(date +"%Y%m%d_%H%M%S")
JOB_NAME="fashion_mnist_batch_predict_$now"
LABELS="team=engineering,phase=test,owner=drew"
SIGNATURE_NAME="serving_default"
gcloud ai-platform jobs submit prediction $JOB_NAME \
--model $MODEL_NAME \
--version $VERSION_NAME \
--input-paths $INPUT_PATHS \
--output-path $OUTPUT_PATH \
--region $REGION \
--data-format $DATA_FORMAT \
--labels $LABELS \
--signature-name $SIGNATURE_NAME
In [34]:
# You can stream the logs, this cell will block until the job completes.
# Copy and paste from the previous cell's output based to grab your job name
# gcloud ai-platform jobs stream-logs fashion_mnist_batch_predict_20200611_151356
In [35]:
!gsutil ls gs://$BUCKET/batch_predictions
In [36]:
# View predictions with keys
!gsutil cat gs://$BUCKET/batch_predictions/prediction.results-00000-of-00010
There are also times where it's desirable to forward some or all of the input features along with the output. This can be achieved in a very similar manner as adding keyed outputs to our model.
Note that this will be a little trickier to grab a subset of features if you are feeding all of your input features as a single Input() layer in the Keras model. This example takes multiple Inputs.
In [37]:
# Build a toy model using the Boston Housing dataset
# https://www.kaggle.com/c/boston-housing
# Prediction target is median value of homes in $1000's
(train_data, train_targets), (test_data, test_targets) = keras.datasets.boston_housing.load_data()
# Extract just two of the features for simplicity's sake
train_tax_rate = train_data[:,10]
train_rooms = train_data[:,5]
In [38]:
# Build a toy model with multiple inputs
# This time using the Keras functional API
from tensorflow.keras.layers import Input
from tensorflow.keras import Model
tax_rate = Input(shape=(1,), dtype=tf.float32, name="tax_rate")
rooms = Input(shape=(1,), dtype=tf.float32, name="rooms")
x = tf.keras.layers.Concatenate()([tax_rate, rooms])
x = tf.keras.layers.Dense(64, activation='relu')(x)
price = tf.keras.layers.Dense(1, activation=None, name="price")(x)
# Functional API model instead of Sequential
model = Model(inputs=[tax_rate, rooms], outputs=[price])
In [39]:
model.compile(
optimizer='adam',
loss='mean_squared_error',
metrics=['accuracy']
)
# Again, we're not concerned with model performance
model.fit([train_tax_rate, train_rooms], train_targets, epochs=10)
Out[39]:
In [40]:
model.predict({
'tax_rate': tf.convert_to_tensor([20.2]),
'rooms': tf.convert_to_tensor([6.2])
})
Out[40]:
In [41]:
BOSTON_EXPORT_PATH = './boston_model/'
model.save(BOSTON_EXPORT_PATH)
In [42]:
# Will retain weights from trained model but also forward out a feature
forward_model = Model(inputs=[tax_rate, rooms], outputs=[price, tax_rate])
In [43]:
# Notice we get both outputs now
forward_model.predict({
'tax_rate': tf.convert_to_tensor([5.0]),
'rooms': tf.convert_to_tensor([6.2])
})
Out[43]:
In [44]:
FORWARD_EXPORT_PATH = './forward_model/'
forward_model.save(FORWARD_EXPORT_PATH)
In [45]:
!saved_model_cli show --tag_set serve --signature_def serving_default --dir {FORWARD_EXPORT_PATH}
In [46]:
!saved_model_cli show --tag_set serve --signature_def serving_default --dir {BOSTON_EXPORT_PATH}
In [47]:
# In our previous example, we leverage and inference function pulled off of a loaded model
# In this case we will need to create ourselves since we haven't saved it out yet
@tf.function(input_signature=[tf.TensorSpec([None, 1], dtype=tf.float32), tf.TensorSpec([None, 1], dtype=tf.float32)])
def standard_forward_prediction(tax_rate, rooms):
pred = model([tax_rate, rooms], training=False)
return {
'price': pred,
}
In [48]:
# Return out the feature of interest as well as the prediction
@tf.function(input_signature=[tf.TensorSpec([None, 1], dtype=tf.float32), tf.TensorSpec([None, 1], dtype=tf.float32)])
def feature_forward_prediction(tax_rate, rooms):
pred = model([tax_rate, rooms], training=False)
return {
'price': pred,
'tax_rate': tax_rate
}
In [49]:
# Save out the model with both signatures
DUAL_SIGNATURE_FORWARD_PATH = './dual_signature_forward_model/'
model.save(DUAL_SIGNATURE_FORWARD_PATH, signatures={'serving_default': standard_forward_prediction,
'feature_forward': feature_forward_prediction})
In [50]:
# Inspect just the feature_forward signature, but we also have standard serving_default
!saved_model_cli show --tag_set serve --signature_def feature_forward --dir {DUAL_SIGNATURE_FORWARD_PATH}
In [ ]:
Copyright 2020 Google Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.