模型部署主要由config.json和customize_service.py共同组成，其中，config.json的功能是管理模型运行所需的依赖环境，customize_service.py的功能是模型前向推理的逻辑。本篇文档将对于自研和第三方算法套件的customize_sevice.py进行介绍。

一、基础结构

如下所示是模型推理的基本逻辑，由_preprocess、_inference、_postprocess三部分组成。

表1 模型推理函数解析

函数	输入	返回值	功能
_preprocess	request请求原始数据， Type(map)	numpy类型数据列表， Type(map)	接收request数据，并转换为模型可以接受的输入格式
_inference	numpy类型数据列表， Type(map)	numpy类型推理结果列表， Type(map)	对于输入数据进行前向推理，得到推理结果
_postprocess	numpy类型数据列表， Type(map)	json类型输出结果， Type(map)	将推理的结果进行后处理，得到预期的输出格式，该结果就是最终的返回值

class SingleNodeService(ModelService):
    '''SingleNodeModel defines abstraction for model service which loads a
    single model.
    '''

    def inference(self, data):
        '''
        Wrapper function to run preprocess, inference and postprocess functions.

        Parameters
        ----------
        data : map of object
            Raw input from request.

        Returns
        -------
        list of outputs to be sent back to client.
            data to be sent back
        '''
        data = self._preprocess(data)
        ...
        data = self._inference(data)
        ...
        data = self._postprocess(data)
		...
        return data

    @abstractmethod
    def _inference(self, data):
        '''
        Internal inference methods. Run forward computation and
        return output.

        Parameters
        ----------
        data : map of NDArray
            Preprocessed inputs in NDArray format.

        Returns
        -------
        list of NDArray
            Inference output.
        '''
        return data

    @abstractmethod
    def _preprocess(self, data):
        '''
        Internal preprocess methods. Do transformation on raw
        inputs and convert them to NDArray.

        Parameters
        ----------
        data : map of object
            Raw inputs from request.

        Returns
        -------
        list of NDArray
            Processed inputs in NDArray format.
        '''
        return data

    @abstractmethod
    def _postprocess(self, data):
        '''
        Internal postprocess methods. Do transformation on inference output
        and convert them to MIME type objects.

        Parameters
        ----------
        data : map of NDArray
            Inference output.

        Returns
        -------
        list of object
            list of outputs to be sent back.
        '''
        return data

二、自研套件

ivgDetection是自研的目标检测算法套件之一，此处以预置的yolov3车牌检测模型（pytorch）为例，介绍ivgDetection套件的推理文件书写方式。

2.1 模型初始化

模型初始化只会在模型初次部署时调用，主要包括模型准备工作，包含模型构建和模型导入等内容。

...
# PyTorch runtime
from model_service.pytorch_model_service import PTServingBaseService
from base.config import CfgNode
from infer import Controller
...

class ImageClassificationService(PTServingBaseService):
    def __init__(self, model_name, model_path, **kwargs):
        cfg = CfgNode()
        cfg.model_dir = osp.join(osp.dirname(osp.abspath(__file__)), 'res')
        cfg.devices = list(range(torch.cuda.device_count())) if torch.cuda.is_available() else []
        cfg.max_batch_size = 1
        cfg.is_vis = False

        controller = Controller(cfg=cfg)
        # 初始化模型并载入训练好的模型，模型结构和模型路径都是在算法开发套件的config.py中指定的
        self.infer = controller.create_ruinner(infer_type='ImageDetRunner')

        if torch.cuda.is_available():
            logger.info('Using GPU for inference')
        else:
            logger.info('Using CPU for inference')

2.2 _pre_process

    def _preprocess(self, data: dict) -> dict:
        imgs = []
        for k, v in data.items():
            if k == 'images':
                for file_name, file_content in v.items():
                
                    img = np.asarray(Image.open(file_content), dtype=np.uint8)
                
                    imgs.append(img)
		    # 返回一个字典，其中字典的键可以是任意的，对应的值就是numpy类型的图像列表
        preprocess_data = dict(
            inputs=(imgs,)
        )
        return preprocess_data

2.3 _inference

    def _inference(self, data: dict) -> dict:
	# 这里的input就是_pre_process中返回的字典的键，需要进行对应
        inputs = data['inputs']
        # 前向推理
        results, _ = self.infer(*inputs)

        inference_data = dict(
            results=results
        )
        return inference_data

2.4 _post_process

    def _postprocess(self, data: dict) -> dict:
        results = data['results']
        # 将推理结果results进行后处理，如下的后处理方式可以直接在网页端显示bbox可视化结果
        bboxes, scores, classes = [], [], []
        work_dir = osp.dirname(osp.abspath(__file__))

        # 类别索引对照表，存放在模型导出后的res/config.json文件中
        label_list = json.load(open(osp.join(work_dir, 'res','config.json')))['infer']['visualizer']['label_list']
        for item in results['preds'][0]:
            bboxes.append([item[1], item[0], item[3], item[2]])
            scores.append(item[-2])
            classes.append(label_list[int(item[-1])])

        detection = dict(
            detection_classes=classes,
            detection_boxes=bboxes,
            detection_scores=scores
        )

        return detection

图1 ivgDetection在线部署

上述可视化后处理方式还需要配合config.json一起指定输出的数据结构，用户需要在config.json中的response中写成：

    ...
	...
		"response": {
                "Content-type": "multipart/form-data",
                "data": {
                    "type": "object",
                    "properties": {
                        "detection_classes": {
                            "type": "array",
                            "items": [{
                                "type": "string"
                            }]
                        },
                        "detection_boxes": {
                            "type": "array",
                            "items": [{
                                "type": "array",
                                "minItems": 4,
                                "maxItems": 4,
                                "items": [{
                                    "type": "number"
                                }]
                            }]
                        },
                        "detection_scores": {
                            "type": "array",
                            "items": [{
                                "type": "number"
                            }]
                        }
                    }
                }
            }

三、开源套件

mmdetection是算法开发套件支持的第三方目标检测算法套件之一，我们以预置的faster_rcnn目标检测模型（pytorch）为例，介绍mmdetection套件的推理文件书写方式。

3.1 模型初始化

构建模型结构，载入模型权值，完成模型初始化构建工作。

from model_service.pytorch_model_service import PTServingBaseService
from mmdet.apis import (inference_detector, init_detector)
...
class ImageClassificationService(PTServingBaseService):
    def __init__(self, model_name, model_path, **kwargs):
        model_dir = osp.join(osp.dirname(osp.abspath(__file__)), 'res')
        # 模型导出生成的config.py、训练模型、类别索引文件路径
        config = osp.join(model_dir, 'config.py')
        checkpoint = osp.join(model_dir, 'model.pth')
        class_txt_path = osp.join(model_dir, 'classes.txt')

        device = 'cuda:0' if torch.cuda.is_available() else 'cpu'
        cfg_options = {
            'data.test.classes':f'class_txt_path',
        }
		
        # 模型初始化，并载入训练模型
        self.model = init_detector(config, checkpoint, device=device, cfg_options=cfg_options)


        if torch.cuda.is_available():
            logger.info('Using GPU for inference')
        else:
            logger.info('Using CPU for inference')

3.2 _pre_process

从request请求中读取图片流，并转换为模型可以接受的输入。

    def _preprocess(self, data: dict) -> dict:
        imgs = []
        for k, v in data.items():
            if k == 'images':
                for file_name, file_content in v.items():
                    img = cv2.imdecode(np.array(bytearray(file_content.read()), dtype=np.uint8), cv2.IMREAD_COLOR)
                    imgs.append(img)

        preprocess_data = dict(
            inputs=imgs
        )
        return preprocess_data

3.3 _inference

    def _inference(self, data: dict) -> dict:
        inputs = data['inputs']
        # 前向推理，得到推理结果
        result = inference_detector(self.model, inputs)

        inference_data = dict(
            results=result
        )
        return inference_data

3.4 _post_process

    def _postprocess(self, data: dict) -> dict:
        results = data['results']
		# 将推理结果转换成最终的json格式
        label_list = self.model.CLASSES
        classes = []
        bboxes = []
        scores = []
        for index, res in enumerate(results[0]):
            category = label_list[index]
            for item in res:
                if len(item) = SCORE_THR:
                    classes.append(category)
                    bboxes.append([bbox[1], bbox[0], bbox[3], bbox[2]])
                    scores.append(score)

        detection = dict(
                detection_classes=classes,
                detection_boxes=bboxes,
                detection_scores=scores
        )
        return detection

四、用户自定义模型

如果用户需要部署的模型是基于自研套件或者第三方套件进行二次开发的，那么推理脚本和内置的算法是一致的，同一个套件中的所有算法理论上是可以共享同一个部署推理脚本的。如果是完全自创的或者不属于算法开发套件内置套件的任何一种，用户可以参考模型推理代码编写说明。

父主题： 部署

同意关联代理商云淘科技，购买华为云产品更优惠（QQ 78315851）

内容没看懂？ 不太想学习？想快速解决？ 有偿解决： 联系专家