模型推理代码编写说明
本章节介绍了在ModelArts中模型推理代码编写的通用方法及说明,针对常用AI引擎的自定义脚本代码示例(包含推理代码示例),请参见自定义脚本代码示例。本文在编写说明下方提供了一个TensorFlow引擎的推理代码示例以及一个在推理脚本中自定义推理逻辑的示例。
ModelArts推理因API网关(APIG)的限制,模型单次预测的时间不能超过40S,模型推理代码编写需逻辑清晰,代码简洁,以此达到更好的推理效果。
推理代码编写指导
- 在模型代码推理文件“customize_service.py”中,需要添加一个子类,该子类继承对应模型类型的父类,各模型类型的父类名称和导入语句如表1所示。导入语句所涉及的Python包在ModelArts环境中已配置,用户无需自行安装。
表1 各模型类型的父类名称和导入语句 模型类型
父类
导入语句
TensorFlow
TfServingBaseService
from model_service.tfserving_model_service import TfServingBaseService
PyTorch
PTServingBaseService
from model_service.pytorch_model_service import PTServingBaseService
MindSpore
SingleNodeService
from model_service.model_service import SingleNodeService
- 可以重写的方法有以下几种。
表2 重写方法 方法名
说明
__init__(self, model_name, model_path)
初始化方法,适用于深度学习框架模型。该方法内加载模型及标签等(pytorch和caffe类型模型必须重写,实现模型加载逻辑)。
__init__(self, model_path)
初始化方法,适用于机器学习框架模型。该方法内初始化模型的路径(self.model_path)。在Spark_MLlib中,该方法还会初始化SparkSession(self.spark)。
_preprocess(self, data)
预处理方法,在推理请求前调用,用于将API接口输入的用户原始请求数据转换为模型期望输入数据。
_inference(self, data)
实际推理请求方法(不建议重写,重写后会覆盖ModelArts内置的推理过程,运行自定义的推理逻辑)。
_postprocess(self, data)
后处理方法,在推理请求完成后调用,用于将模型输出转换为API接口输出。
- 用户可以选择重写preprocess和postprocess方法,以实现API输入数据的预处理和推理输出结果的后处理。
- 重写模型父类的初始化方法init可能导致AI应用“运行异常”。
- 可以使用的属性为模型所在的本地路径,属性名为“self.model_path”。另外pyspark模型在“customize_service.py”中可以使用“self.spark”获取SparkSession对象。
推理代码中,需要通过绝对路径读取文件。模型所在的本地路径可以通过self.model_path属性获得。
- 当使用TensorFlow、Caffe、MXNet时,self.model_path为模型文件目录路径,读取文件示例如下:
# model目录下放置label.json文件,此处读取 with open(os.path.join(self.model_path, 'label.json')) as f: self.label = json.load(f)
- 当使用PyTorch、Scikit_Learn、pyspark时,self.model_path为模型文件路径,读取文件示例如下:
# model目录下放置label.json文件,此处读取 dir_path = os.path.dirname(os.path.realpath(self.model_path)) with open(os.path.join(dir_path, 'label.json')) as f: self.label = json.load(f)
- 当使用TensorFlow、Caffe、MXNet时,self.model_path为模型文件目录路径,读取文件示例如下:
- 预处理方法、实际推理请求方法和后处理方法中的接口传入“data”当前支持两种content-type,即“multipart/form-data”和“application/json”。
- “multipart/form-data”请求
curl -X POST \ <modelarts-inference-endpoint> \ -F image1=@cat.jpg \ -F images2=@horse.jpg
对应的传入data为
[ { "image1":{ "cat.jpg":"<cat.jpg file io>" } }, { "image2":{ "horse.jpg":"<horse.jpg file io>" } } ]
- “application/json”请求
curl -X POST \ <modelarts-inference-endpoint> \ -d '{ "images":"base64 encode image" }'
对应的传入data为python dict
{ "images":"base64 encode image" }
- “multipart/form-data”请求
TensorFlow的推理脚本示例
- 推理代码
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27
from PIL import Image import numpy as np from model_service.tfserving_model_service import TfServingBaseService class MnistService(TfServingBaseService): def _preprocess(self, data): preprocessed_data = {} for k, v in data.items(): for file_name, file_content in v.items(): image1 = Image.open(file_content) image1 = np.array(image1, dtype=np.float32) image1.resize((1, 784)) preprocessed_data[k] = image1 return preprocessed_data def _postprocess(self, data): infer_output = {} for output_name, result in data.items(): infer_output["mnist_result"] = result[0].index(max(result[0])) return infer_output
- 请求
curl -X POST \ 在线服务地址 \ -F images=@test.jpg
- 返回
{"mnist_result": 7}
在上面的代码示例中,完成了将用户表单输入的图片的大小调整,转换为可以适配模型输入的shape。首先通过Pillow库读取“32×32”的图片,调整图片大小为“1×784”以匹配模型输入。在后续处理中,转换模型输出为列表,用于Restful接口输出展示。
自定义推理逻辑的推理脚本示例
首先,需要在配置文件中,定义自己的依赖包,详细示例请参见使用自定义依赖包的模型配置文件示例。然后通过如下示例代码,实现了“saved_model”格式模型的加载推理。
![](https://support.huaweicloud.com/usermanual-standard-modelarts/public_sys-resources/note_3.0-zh-cn.png)
当前推理基础镜像使用的python的logging模块,采用的是默认的日志级别Warnning,即当前只有warning级别的日志可以默认查询出来。如果想要指定INFO等级的日志能够查询出来,需要在代码中指定logging的输出日志等级为INFO级别。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 |
# -*- coding: utf-8 -*- import json import os import threading import numpy as np import tensorflow as tf from PIL import Image from model_service.tfserving_model_service import TfServingBaseService import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class MnistService(TfServingBaseService): def __init__(self, model_name, model_path): self.model_name = model_name self.model_path = model_path self.model_inputs = {} self.model_outputs = {} # label文件可以在这里加载,在后处理函数里使用 # label.txt放在OBS和模型包的目录 # with open(os.path.join(self.model_path, 'label.txt')) as f: # self.label = json.load(f) # 非阻塞方式加载saved_model模型,防止阻塞超时 thread = threading.Thread(target=self.get_tf_sess) thread.start() def get_tf_sess(self): # 加载saved_model格式的模型 # session要重用,建议不要用with语句 sess = tf.Session(graph=tf.Graph()) meta_graph_def = tf.saved_model.loader.load(sess, [tf.saved_model.tag_constants.SERVING], self.model_path) signature_defs = meta_graph_def.signature_def self.sess = sess signature = [] # only one signature allowed for signature_def in signature_defs: signature.append(signature_def) if len(signature) == 1: model_signature = signature[0] else: logger.warning("signatures more than one, use serving_default signature") model_signature = tf.saved_model.signature_constants.DEFAULT_SERVING_SIGNATURE_DEF_KEY logger.info("model signature: %s", model_signature) for signature_name in meta_graph_def.signature_def[model_signature].inputs: tensorinfo = meta_graph_def.signature_def[model_signature].inputs[signature_name] name = tensorinfo.name op = self.sess.graph.get_tensor_by_name(name) self.model_inputs[signature_name] = op logger.info("model inputs: %s", self.model_inputs) for signature_name in meta_graph_def.signature_def[model_signature].outputs: tensorinfo = meta_graph_def.signature_def[model_signature].outputs[signature_name] name = tensorinfo.name op = self.sess.graph.get_tensor_by_name(name) self.model_outputs[signature_name] = op logger.info("model outputs: %s", self.model_outputs) def _preprocess(self, data): # https两种请求形式 # 1. form-data文件格式的请求对应:data = {"请求key值":{"文件名":<文件io>}} # 2. json格式对应:data = json.loads("接口传入的json体") preprocessed_data = {} for k, v in data.items(): for file_name, file_content in v.items(): image1 = Image.open(file_content) image1 = np.array(image1, dtype=np.float32) image1.resize((1, 28, 28)) preprocessed_data[k] = image1 return preprocessed_data def _inference(self, data): feed_dict = {} for k, v in data.items(): if k not in self.model_inputs.keys(): logger.error("input key %s is not in model inputs %s", k, list(self.model_inputs.keys())) raise Exception("input key %s is not in model inputs %s" % (k, list(self.model_inputs.keys()))) feed_dict[self.model_inputs[k]] = v result = self.sess.run(self.model_outputs, feed_dict=feed_dict) logger.info('predict result : ' + str(result)) return result def _postprocess(self, data): infer_output = {"mnist_result": []} for output_name, results in data.items(): for result in results: infer_output["mnist_result"].append(np.argmax(result)) return infer_output def __del__(self): self.sess.close() |
![](https://support.huaweicloud.com/usermanual-standard-modelarts/public_sys-resources/note_3.0-zh-cn.png)
对于ModelArts不支持的结构模型或者多模型加载,需要__init__方法中自己指定模型加载的路径。示例代码如下:
# -*- coding: utf-8 -*- import os from model_service.tfserving_model_service import TfServingBaseService class MnistService(TfServingBaseService): def __init__(self, model_name, model_path): # 获取程序当前运行路径,即model文件夹所在的路径 root = os.path.dirname(os.path.abspath(__file__)) # test.onnx为待加载模型文件的名称,需要放在model文件夹下 self.model_path = os.path.join(root, test.onnx) # 多模型加载,例如:test2.onnx # self.model_path2 = os.path.join(root, test2.onnx)
MindSpore的推理脚本示例
snt3芯片目前只有北京四提工单申请权限后才可以使用,支持模型格式为.om,推理脚本如下:
from __future__ import absolute_import from __future__ import division from __future__ import print_function import json import os import numpy as np from PIL import Image from hiai.nn_tensor_lib import NNTensor from hiai.nntensor_list import NNTensorList from model_service.hiai_model_service import HiaiBaseService class DemoService(HiaiBaseService): def __init__(self, *args, **kwargs): # 默认加载模型包目录下的om文件 super(DemoService, self).__init__(*args, **kwargs) self.labels_list = None self.is_multilabel = False def _preprocess(self, data): preprocessed_data = {} images = [] for k, v in data.items(): for file_name, file_content in v.items(): image = Image.open(file_content) image = np.array(image) # NHWC # AIPP should use RGB format. # mean reg is applied in AIPP. # Transpose is applied in AIPP tensor = NNTensor(image) images.append(tensor) tensor_list = NNTensorList(images) preprocessed_data['images'] = tensor_list return preprocessed_data def _inference(self, data, image_info=None): result = {} for k, v in data.items(): result[k] = self.model.proc(v) return result def _postprocess(self, data): # 这里增加自己的后处理 return str(data)