google-ml-engine 自定义预测例程错误响应答案

【问题标题】：google-ml-engine custom prediction routine error responsesgoogle-ml-engine 自定义预测例程错误响应
【发布时间】：2020-06-25 19:51:15
【问题描述】：

我在google-ml-engine 中有一个自定义预测例程。效果很好。

我现在正在对实例数据进行输入检查，并希望从我的预测例程中返回错误响应。

示例：https://cloud.google.com/ai-platform/prediction/docs/custom-prediction-routines

引发输入错误等异常。但是，当发生这种情况时，响应正文始终具有{'error': Prediction failed: unknown error}.，我可以看到正确的错误正在谷歌云控制台中记录，但https响应始终相同unknown error。

我的问题是：

如何使自定义预测例程返回正确的错误代码和错误消息字符串？

我可以在预测中返回一个错误字符串/代码，而不是返回一个预测，但它最终会出现在响应的预测部分中，这看起来很奇怪，并且不会出现任何谷歌错误，例如基于实例大小。

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

最好的方法是什么？

谢谢！大卫

【问题讨论】：

标签： google-cloud-ml

【解决方案1】：

请看下面的代码，我在predict 中创建了一个_validate 函数并使用了一个自定义的Exception 类。基本上，我在调用模型predict 方法并处理异常之前验证实例。进行此验证时，响应时间可能会有一些开销，您需要针对您的用例进行测试。

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'}]}

自定义预测类：

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)

this notebook中的完整代码。

【讨论】：

【解决方案2】：

如果它仍然与您相关，我找到了一种使用谷歌内部库的方法（虽然不确定它是否会得到谷歌的认可）。 AI 平台自定义预测包装代码仅在抛出的 Exception 是其内部库中的特定错误消息时才返回自定义错误消息。它也可能不是超级可靠，因为万一 Google 想要更改它，您几乎无法控制。

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

您会在 HTTP 响应中收到以下消息 Prediction failed: Custom error message goes here

【讨论】：