在线推理与离线推理的主要区别在于模型运行时前者依赖深度学习框架，而后者则脱离了深度学习框架，使用推理后端进行推理。

前面讲到，在线推理的端到端流程也包含预处理、执行推理、后处理这三个关键步骤。关于这三个步骤的作用，请移步在线推理章节学习，这里不再赘述。离线推理也包含这三个步骤，为了减轻离线推理代码适配的工作量以及更好的体会各步骤的差异，我们在PytorchInferencer类的基础上实现OmInferencer，仅做最小的修改。

class OmInferencer(PytorchInferencer):
    ...

执行推理

昇腾离线模型OM的推理后端是ACL（Ascend Computing Language），其底层采用C实现，后来在ACL基础上又做了一套Python接口，命名为pyACL，为了方便开发，华为工程师又基于pyacl开发出一款推理工具ais_bench，此工具支持使用命令进行快捷地推理，同时也对外开发推理相关的API。

ais_bench推理命令

python -m ais_bench --model /path/to/model.om --input /path/to/input --output /path/to/output

参数说明：

• --model: 昇腾离线模型OM的路径
• --input: 指定模型需要的输入数据路径。如果模型有多个输入节点，则指定多个输入路径，用“,”隔开。路径可以是文件或者目录，若为文件则执行单样本推理，若为目录，则对该目录下的所有样本进行推理。
• --output: 推理结果保存目录。配置后会创建“日期+时间”的子目录，保存输出结果。

ais_bench命令更详细的用法请参考其Gitcode主页。

ais_bench推理接口

使用ais_bench接口推理通常只需要2步：

1. 创建推理Session实例。

from ais_bench.infer.interface import InferSession
session = InferSession(device_id=0, model_path="/path/to/model.om")

2. 使用session实例进行推理。

model_output = session.infer(feeds=model_input)[0]

将以上2个步骤在OmInferencer中实现：

class OmInferencer(PytorchInferencer):
    def __init__(self, om_path, device_id=0):
        super(OmInferencer, self).__init__()
        self.session = InferSession(device_id=device_id, model_path=om_path)

    def model_inference(self, model_input):
        model_output = self.session.infer(feeds=[model_input])[0]
        return model_output

    ...

前后处理

离线推理时，ais_bench只能接受numpy.ndarray，然后输出numpy.ndarray，而在线推理中，预处理返回的结果是torch.Tensor，后处理的输入数据也是torch.Tensor，所以这两步都需要进行类型转换：

class OmInferencer(PytorchInferencer):
    ...

    def preprocess(self, image_path):
        return super().preprocess(image_path).numpy()

    def postprocess(self, model_output):
        return super.postprocess(torch.from_numpy(model_output))

    ...

或者，直接在执行推理的前后做改动也可以：

class OmInferencer(PytorchInferencer):
    ...

    def model_inference(self, model_input):
        model_output = torch.from_numpy(self.session.infer(feeds=[model_input.numpy()])[0])
        return model_output

    ...

端到端执行

前面我们已经实现了OmInferencer类，完整的代码如下：

from ais_bench.infer.interface import InferSession

class OmInferencer(PytorchInferencer):
    def __init__(self, om_path, device_id=0):
        super(OmInferencer, self).__init__()
        self.session = InferSession(device_id=device_id, model_path=om_path)

    def model_inference(self, model_input):
        model_output = torch.from_numpy(self.session.infer(feeds=[model_input.numpy()])[0])
        return model_output

接下来便可以端到端执行：

inferencer = OmInferencer('./resnet50_bs1.om')
print(inferencer.e2e_inference("./ILSVRC2012_val_00006083.jpeg"))
# {'category': 'Yorkshire terrier', 'score': 0.2925560474395752}

精度/性能验证

精度验证

在模型转换完成后，还需要对模型进行精度验证，确保模型的精度损失在可接受的范围内。

1. 快捷验证

   给PyTorch模型与OM模型传入同样的输入数据，对比PyTorch与OM的输出，对比余弦相似度，绝对与相对误差，余弦相似度越接近于1，绝对与相对误差越接近于0，说明OM模型的精度损失越小。

   def precision_compare(pth_output, om_output):
       pth_output1 = pth_output.flatten().astype(np.float64)
       om_output1 = om_output.flatten().astype(np.float64)
       cosine_similarity = np.dot(pth_output1, om_output1) \
                           / (np.linalg.norm(om_output1) * np.linalg.norm(om_output1))
       absolute_errors = np.abs(om_output1 - pth_output1)
       relative_errors = absolute_errors / pth_output1 * 100
       print('余弦相似度：'， cosine_similarity)
       print('最大绝对误差：'，absolute_errors.max())
       print('最大相对误差：'，relative_errors.max() )
   
       pth_inferencer = PytorchInferencer()
       om_inferencer = OmInferencer('./resnet50_bs1.om')
       model_input = pth_inferencer.preprocess('./ILSVRC2012_val_00006083.jpeg')
       pth_output = pth_inferencer.model_inference(model_input).numpy()
       om_output = pth_inferencer.model_inference(model_input).numpy()
       precision_compare(pth_output, om_output)
   

   需注意:

   • 余弦相似度只能说明两个向量在空间上的方向是否一致，并不能判断两个向量是否完全相等。比如向量 (1, 2, 3) 和向量 (10, 20, 30) 两个不同的向量其余弦相似度却为 1。
   • 对于不同含义的输出数据，对绝对误差的容忍度也不一样，比如目标检测，模型会输出检测框坐标与置信度，检测框坐标绝对值相差1或者2（即相差一两个像素）是可接受的，但同一个检测框的置信度相差0.1或0.2，则说明离线模型存在较大的精度损失。

   所以，观测精度损失时需要结合输出数据的含义，对多个指标进行综合对比。

2. 使用数据集测试模型精度

   def evaluate(inferencer, img_dir, label_file):
       groundtruth = {}
       with open(label_file, 'r') as f:
           for line in f:
               image_name, label = line.strip().split(' ')
               groundtruth[image_name] = int(label)
   
       num_total, num_right = 0, 0
       for i, image_name in tqdm.tqdm(enumerate(os.listdir(img_dir))):
           label = groundtruth[image_name]
           image_path = osp.join(img_dir, image_name)
           pred = inferencer.e2e_inference(image_path)['class_id']
           num_total += 1
           if pred == label:
               num_right += 1
       return {'Acc@top1': num_right / num_total}
   
   pth_inferencer = PytorchInferencer()
   om_inferencer = OmInferencer('./resnet50_bs1.om')
   imagenet_1k_val_dir = './imagenet/val'
   label_file = './imagenet/val_label.txt'
   print('在线推理准确率：'， evaluate(pth_inferencer, imagenet_1k_val_dir, label_file))
   print('离线推理准确率：'， evaluate(om_inferencer, imagenet_1k_val_dir, label_file))
   

性能验证

1. 纯推理快速验证性能

   python3 -m ais_bench --model /path/to/mode.om --loop 100 --batchsize 1
   

   执行完成后，将打印以下的信息，其中以找到关键字“throughput”即为模型的吞吐率：

   [INFO] -----------------Performance Summary------------------
   [INFO] H2D_latency (ms): min = 0.05700000002980232, max = 0.05700000002980232, mean = 0.05700000002980232, median = 0.05700000002980232, percentile(99%) = 0.05700000002980232
   [INFO] NPU_compute_time (ms): min = 0.6650000214576721, max = 0.6650000214576721, mean = 0.6650000214576721, median = 0.6650000214576721, percentile(99%) = 0.6650000214576721
   [INFO] D2H_latency (ms): min = 0.014999999664723873, max = 0.014999999664723873, mean = 0.014999999664723873, median = 0.014999999664723873, percentile(99%) = 0.014999999664723873
   [INFO] throughput 1000*batchsize.mean(1)/NPU_compute_time.mean(0.6650000214576721): 1503.759349974173
   

2. ais_bench性能测试接口

   from ais_bench.infer.interface import MemorySummary
   from ais_bench.infer.summary import summary
   
   def display_performance(batchsize=1, output_prefix=None, display_all_summary=True, multi_threads=False):
       s = session.summary()
       summary.npu_compute_time_list = [end_time - start_time for start_time, end_time in s.exec_time_list]
       summary.h2d_latency_list = MemorySummary.get_h2d_time_list()  # host to device
       summary.d2h_latency_list = MemorySummary.get_d2h_time_list()  # device to host
       summary.report(batchsize, output_prefix, display_all_summary, multi_threads)
   

   只需在推理结束后执行此函数，程序便会打印模型的性能信息，具体的打印内容与上面纯推理一致。

3. 端到端性能

   端到端推理的性能，可以在业务代码中打点统计，参考：

   import time
   
   inferencer = OmInferencer('./resnet50_bs1.om')
   image_list = ["./ILSVRC2012_val_00006083.jpeg"] * 10
   e2e_times = []
   for image in image_list:
       start = time.time()
       prediction = inferencer.e2e_inference(image)
       end = time.time()
       e2e_times.append(end - start)
   
   avg_time = sum(e2e_times) / len(e2e_times)
   speed = 1 / avg_time
   print(f"端到端平均耗时：{avg_time: .2f} s")
   print(f"端到端推理性能：{speed: .2f} fps")
   

   如果不想修改业务代码，也可以尝试用Python装饰器进行性能统计，首先定义一个统计函数耗时的装饰器：

   import time
   
   def count_time(func):
       def wrapper(*args, **kwargs):
           start = time.time()
           result = func(*args, **kwargs)
           end = time.time()
   
           global E2E_TIMES
           if 'E2E_TIMES' not in globals():
               E2E_TIMES = []
           E2E_TIMES.append((end - start))
           return result
       return wrapper
   

   在需要统计耗时的函数上使用此装饰器：

   class OmInferencer(PytorchInferencer):
       ...
       @count_time
       def e2e_inference(self, image_path):
           model_input = self.preprocess(image_path)
           model_output = self.model_inference(model_input)
           prediction = self.postprocess(model_output)
           return prediction
       ...
   

   最后，进行性能统计：

   inferencer = OmInferencer('./resnet50_bs1.om')
   image_list = ["./ILSVRC2012_val_00006083.jpeg"] * 10
   for image in image_list:
       inferencer.e2e_inference(image)
   
   avg_time = sum(E2E_TIMES) / len(E2E_TIMES)
   speed = 1 / avg_time
   print(f"端到端平均耗时：{avg_time: .2f} s")
   print(f"端到端推理性能：{speed: .2f} fps")
   

推理场景

截止当前，此教程中介绍的都是理想的、简单的推理场景。但实际的应用场景下，情况可能复杂很多，包括但不限于：

1. 三段式：有些场景下，模型部署于线上服务侧，而数据生成于端侧，这种情况下，模型的推理在线上完成，端侧只需将数据进行预处理后发送给在线服务进行模型推理，最后再根据下游流程的需要对推理结果进行后处理。
2. 多模型推理：模型由多个子模型组成，这些子模型的执行顺序或串行，或并行，或串行与并行混合。

三段式

此场景下，三个关键步骤独立执行。首先执行据预处理，将预处理后的数据保存为NPY或BIN文件，接着执行模型推理，最后再将模型推理结果进行后处理。ModelZoo上大多数模型的端到端推理流程采用三段式推理方式，下面仅对一些常见的任务给出示例：

任务类别	离线推理案例
图像分类	Resnet50
目标检测	YoloX_Tiny
语音识别	Deepspeech2
OCR	StarNet_MobileNetV3
自然语言理解	Bert_Base

多个模型串行

此场景下离线推理流程适配的难点：

1. 推理时存在并行执行、串行执行、或两者混合，端到端流程复杂；
2. 预处理，后处理，中间处理流程也依赖深度学习框架，无法与框架解耦；
3. 子模型的定义经过多重继承，调用栈很深。

针对这3个难点，建议直接在原有的在线推理的代码上适配。大致步骤为：

1. 事先创建好各子模型的离线推理session
2. 找到在线推理各子模型执行的地方，然后用session.infer(...)替换

这里会引入一个新的问题：模型执行的地方嵌套得很深，如何将session传进去？这里提供三种方法：

1. 将创建好的session设为全局变量，执行session时无需传参;
2. 如果嵌套得不深，可以在原来的函数上新加一个session参数;
3. 继承原在线推理的类，在__init__函数中创建session，执行时使用self.session即可。

示例：

任务类别	离线推理案例
语音处理	Conformer
文生图	Stable Diffusion