更新时间:2023-05-12 GMT+08:00
分享

模型推理代码编写说明

本章节介绍了在ModelArts中模型推理代码编写的通用方法及说明,针对常用AI引擎的自定义脚本代码示例(包含推理代码示例),请参见自定义脚本代码示例。本文在编写说明下方提供了一个TensorFlow引擎的推理代码示例以及一个在推理脚本中自定义推理逻辑的示例。

推理代码编写指导

  1. 在模型代码推理文件“customize_service.py”中,需要添加一个子类,该子类继承对应模型类型的父类,各模型类型的父类名称和导入语句如表1所示。
    表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. 可以重写的方法有以下几种。
    表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应用“运行异常”
  3. 可以使用的属性为模型所在的本地路径,属性名为“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)
  4. 预处理方法、实际推理请求方法和后处理方法中的接口传入“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"
      
       }

TensorFlow的推理脚本示例

TensorFlow MnistService示例如下。更多TensorFlow推理代码示例请参考TensorFlowTensorFlow 2.1。其他引擎推理代码请参考PyTorchCaffe
  • 推理代码
     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接口输出展示。

XGBoost的推理脚本示例

更多机器学习引擎的推理代码请参考PysparkScikit Learn

# coding:utf-8
import collections
import json
import xgboost as xgb
from model_service.python_model_service import XgSklServingBaseService


class UserService(XgSklServingBaseService):

    # request data preprocess
    def _preprocess(self, data):
        list_data = []
        json_data = json.loads(data, object_pairs_hook=collections.OrderedDict)
        for element in json_data["data"]["req_data"]:
            array = []
            for each in element:
                array.append(element[each])
                list_data.append(array)
        return list_data

    #   predict
    def _inference(self, data):
        xg_model = xgb.Booster(model_file=self.model_path)
        pre_data = xgb.DMatrix(data)
        pre_result = xg_model.predict(pre_data)
        pre_result = pre_result.tolist()
        return pre_result

    # predict result process
    def _postprocess(self, data):
        resp_data = []
        for element in data:
            resp_data.append({"predict_result": element})
        return resp_data

自定义推理逻辑的推理脚本示例

首先,需要在配置文件中,定义自己的依赖包,详细示例请参见使用自定义依赖包的模型配置文件示例。然后通过如下示例代码,实现了“saved_model”格式模型的加载推理。

当前推理基础镜像使用的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
104
105
106
107
108
109
110
111
112
113
114
# -*- 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()

MindSpore的推理脚本示例

  • D310芯片目前只有北京四提工单申请权限后才可以使用,支持模型格式为.om, TensorFlow的模型需要先通过模型转换功能转成.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)
分享:

    相关文档

    相关产品