在AWS SageMaker中创建和部署经过预处理和后处理的预训练tensorflow模型

Question

我正在尝试部署https://github.com/matterport/Mask_RCNN在SageMaker中进行预测。

问题是，在将输入提供给Keras层之前，该模型使用numpy和scikit-learn作为预处理，因此仅像本例中那样部署模型是行不通的。

我尝试过的东西：

• 隔离模型的Keras部分并使用它，以便在端点中部署的模型使用预处理的数据。这种情况下的问题是提示http错误431 Request Header Fields Too Large，因为预处理的数据(由调整大小的图像加上锚点形成(比原始数据大得多
• 使用input_handler()函数制作一个entry_point.py脚本。此脚本如下所示：

def install(package: str):
""" pip install a package """
subprocess.check_call([sys.executable, "-q", "-m", "pip", "install", package])
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument('--model_dir', type=str, default=os.environ.get('SM_MODEL_DIR'))
return parser.parse_known_args()
def input_handler(data, context):
""" Expects @data to be json. Pre-process the data to be ready to be fed to the Keras layers. 
https://www.mikulskibartosz.name/custom-preprocessing-in-tensorflow-with-sagemaker-endpoints/ """
data_dec = np.array(json.loads(data['inputs']))
[molded_images, image_metas, anchors] = preprocess_images(data_dec)
inputs = { #input_image, input_image_meta and input_anchors are the names of the Input layers of the model
"inputs": {
"input_image": molded_images.tolist(),
"input_image_meta": image_metas.tolist(),
"input_anchors": anchors.tolist()
}
}
return json.dumps(inputs)
if __name__ == "__main__":
args, _ = parse_args()
install('scikit-image')
install('scipy')

然后在SageMaker笔记本实例中创建模型和端点：

from sagemaker.tensorflow.model import TensorFlowModel
model = TensorFlowModel(model_data = url_to_saved_model_s3,
role = sagemaker.get_execution_role(),
framework_version = '2.3.1',
entry_point = 'entry_point.py' 
)
predictor = model.deploy(initial_instance_count=1, instance_type='ml.t2.medium') # create endpoint

但看起来从未调用过input_handler()。

关于如何部署需要预处理非张量的推理模型，有什么帮助吗？

Answer 1

这样您就可以在脚本中使用numpy和sklearn了。您只需要使用入口点脚本在代码目录中创建一个包含这些包的requirements.txt。至于input_handler和output_handler功能，您希望以SageMaker Tensorflow Serving Container中所示的以下格式构建它。

import json
def input_handler(data, context):
""" Pre-process request input before it is sent to TensorFlow Serving REST API
Args:
data (obj): the request data, in format of dict or string
context (Context): an object containing request and configuration details
Returns:
(dict): a JSON-serializable dict that contains request body and headers
"""
if context.request_content_type == 'application/json':
# pass through json (assumes it's correctly formed)
d = data.read().decode('utf-8')
return d if len(d) else ''
if context.request_content_type == 'text/csv':
# very simple csv handler
return json.dumps({
'instances': [float(x) for x in data.read().decode('utf-8').split(',')]
})
raise ValueError('{{"error": "unsupported content type {}"}}'.format(
context.request_content_type or "unknown"))

def output_handler(data, context):
"""Post-process TensorFlow Serving output before it is returned to the client.
Args:
data (obj): the TensorFlow serving response
context (Context): an object containing request and configuration details
Returns:
(bytes, string): data to return to client, response content type
"""
if data.status_code != 200:
raise ValueError(data.content.decode('utf-8'))
response_content_type = context.accept_header
prediction = data.content
return prediction, response_content_type

确保记录您的每一行函数，这将发送到CloudWatch，我们将了解您正在处理的错误，您可以看到您的脚本在哪里崩溃。