CODEWITHSUNDEEP
google-cloud-ml
David
David

Reputation: 91

google-ml-engine custom prediction routine error responses

I have a custom prediction routine in google-ml-engine. Works very well.

I now am doing input checking on the instance data, and want to return error responses from my predict routine.

The example: https://cloud.google.com/ai-platform/prediction/docs/custom-prediction-routines

Raises exceptions on input errors, etc. However, when this happens the response body always has {'error': Prediction failed: unknown error}. I can see the correct errors are being logged in google cloud console, but the https response is always the same unknown error.

My question is:

How to make the Custom prediction routine return a proper error code and error message string?

Instead of returning a prediction, I can return an error string/code in prediction -but it ends up in the prediction part of the response which seems hacky and doesn't get any of the google errors eg based on instance size.

root:test_deployment.py:35 {'predictions': {'error': "('Instance does not include required sensors', 'occurred at index 0')"}}

What's the best way to do this?

Thanks! David

Upvotes: 1

Views: 180

Answers (2)

klesouza
klesouza

Reputation: 71

If it is still relevant to you, I found a way by using google internal libraries (not sure if it would be endorsed by Google though). AI platform custom prediction wrapping code only returns custom error message if the Exception thrown is a specific one from their internal library. It might also not be super reliable as you would have very little control in case Google wants to change it.

class Predictor(object):
    def predict(self, instances, **kwargs):
        # Your prediction code here
        
        # This is an internal google library, it should be available at prediction time.
        from google.cloud.ml.prediction import prediction_utils
        raise prediction_utils.PredictionError(0, "Custom error message goes here")

    @classmethod
    def from_path(cls, model_dir):
        # Your logic to load the model here

You would get the following message in your HTTP response Prediction failed: Custom error message goes here

Upvotes: 0

gogasca
gogasca

Reputation: 10058

Please take a look at the following code, I created a _validate function inside predict and use a custom Exception class. Basically, I validate instances, before I call the model predict method and handle the exception. There may be some overhead to the response time when doing this validation, which you need to test for your use case.

requests = [
    "god this episode sucks",
    "meh, I kinda like it",
    "what were the writer thinking, omg!",
    "omg! what a twist, who would'v though :o!",
    99999
]
api = discovery.build('ml', 'v1')

parent = 'projects/{}/models/{}/versions/{}'.format(PROJECT, MODEL_NAME, VERSION_NAME)
parent = 'projects/{}/models/{}'.format(PROJECT, MODEL_NAME)
response = api.projects().predict(body=request_data, name=parent).execute()

{'predictions': [{'Error code': 1, 'Message': 'Invalid instance type'}]}

Custom Prediction class:

import os
import pickle
import numpy as np
import logging

from datetime import date

import tensorflow.keras as keras


class CustomModelPredictionError(Exception):
    def __init__(self, code, message='Error found'):        
        self.code = code 
        self.message = message # you could add more args
    def __str__(self):
        return str(self.message)


def isstr(s):
    return isinstance(s, str) or isinstance(s, bytes)


def _validate(instances):
    for instance in instances:
        if not isstr(instance):
            raise CustomModelPredictionError(1, 'Invalid instance type')
    return instances


class CustomModelPrediction(object):
    def __init__(self, model, processor):    
        self._model = model
        self._processor = processor       

    def _postprocess(self, predictions):
        labels = ['negative', 'positive']
        return [
            {
                "label":labels[int(np.round(prediction))],
                "score":float(np.round(prediction, 4))
            } for prediction in predictions]

    def predict(self, instances, **kwargs):
        try:
            instances = _validate(instances)            
        except CustomModelPredictionError as c:            
            return [{"Error code": c.code, "Message": c.message}]
        else:
            preprocessed_data = self._processor.transform(instances)
            predictions =  self._model.predict(preprocessed_data)
            labels = self._postprocess(predictions)
            return labels

    @classmethod
    def from_path(cls, model_dir):                
        model = keras.models.load_model(
          os.path.join(model_dir,'keras_saved_model.h5'))
        with open(os.path.join(model_dir, 'processor_state.pkl'), 'rb') as f:
            processor = pickle.load(f)    
        return cls(model, processor)

Complete code in this notebook.

Upvotes: 1

Related Questions