使用模型集成执行多个模型#

导航到

第 4 部分：加速模型

第 6 部分：使用 BLS API 构建复杂管道

文档：集成

现代机器学习系统通常涉及执行多个模型，无论是由于预处理和后处理步骤、聚合多个模型的预测，还是让不同的模型执行不同的任务。在本示例中，我们将探索使用模型集成在服务器端仅通过单个网络调用来执行多个模型。这提供了减少客户端和服务器之间数据复制次数的好处，并消除了一些网络调用固有的延迟。

为了说明创建模型集成的过程，我们将重用 第 1 部分 中首次介绍的模型管道。在之前的示例中，我们分别执行了文本检测和识别模型，我们的客户端进行了两次不同的网络调用，并在两者之间执行了各种处理步骤——例如裁剪和调整图像大小，或将张量解码为文本。以下是管道的简化图，其中一些步骤发生在客户端，一些步骤发生在服务器上。

sequenceDiagram
    Client ->> Triton: Full Image
    activate Triton
    Note right of Triton: Text Detection
    Triton ->> Client: Text Bounding Boxes
    deactivate Triton
    activate Client
    Note left of Client: Image Cropping
    Client ->> Triton: Cropped Images
    deactivate Client
    activate Triton
    Note right of Triton: Text Recognition
    Triton ->> Client: Parsed Text
    deactivate Triton

为了减少必要的网络调用和数据复制次数（并利用可能更强大的服务器来执行预处理/后处理），我们可以使用 Triton 的 模型集成 功能，通过一次网络调用执行多个模型。

sequenceDiagram
    Client ->> Triton: Full Image
    activate Triton
    activate Triton
    Note right of Triton: Text Detection
    deactivate Triton
    activate Triton
    Note right of Triton: Image Cropping (Serverside)
    Note left of Triton: Ensemble Model
    deactivate Triton
    activate Triton
    Note right of Triton: Text Recognition
    Triton ->> Client: Parsed Text
    deactivate Triton
    deactivate Triton

让我们了解如何创建 Triton 模型集成。

注意： 如果您正在寻找了解数据如何流经集成的示例，请参考本教程！

部署基础模型#

第一步是像过去一样，将文本检测和文本识别模型部署为常规 Triton 模型。有关将模型部署到 Triton 的详细概述，请参阅本教程的第 1 部分。为了方便起见，我们包含了两个 shell 脚本来导出这些模型。

注意：我们建议在 NGC TensorFlow 容器环境中执行以下步骤，您可以使用 docker run -it --gpus all -v ${PWD}:/workspace nvcr.io/nvidia/tensorflow:<yy.mm>-tf2-py3 启动该环境

bash utils/export_text_detection.sh

注意：我们建议在 NGC PyTorch 容器环境中执行以下步骤，您可以使用 docker run -it --gpus all -v ${PWD}:/workspace nvcr.io/nvidia/pytorch:<yy.mm>-py3 启动该环境

bash utils/export_text_recognition.sh

使用 Python 后端部署预处理/后处理脚本#

在本教程的先前部分中，我们创建了客户端脚本，用于在客户端进程中执行各种预处理和后处理步骤。例如，在第 1 部分中，我们创建了一个脚本 client.py，它执行以下操作：

  1. 读取图像

  2. 对图像执行缩放和归一化

  3. 将图像发送到 Triton 服务器

  4. 根据文本检测模型返回的边界框裁剪图像

  5. 将裁剪后的图像保存回磁盘

然后，我们有第二个客户端 client2.py，它执行以下操作：

  1. 从 client.py 读取裁剪后的图像

  2. 对图像执行缩放和归一化

  3. 将裁剪后的图像发送到 Triton 服务器

  4. 将文本识别模型返回的张量解码为文本

  5. 打印解码后的文本

为了将许多这些步骤移至 Triton 服务器，我们可以创建一组脚本，这些脚本将在 Triton 的 Python 后端 中运行。Python 后端可用于执行任何 Python 代码，因此我们只需进行少量更改即可将客户端代码直接移植到 Triton。

要为 Python 后端部署模型，我们可以在模型仓库中创建一个目录，如下所示（其中 my_python_model 可以是任何名称）

my_python_model/
├── 1
│   └── model.py
└── config.pbtxt

总共，我们将创建 3 个不同的 python 后端模型，与我们现有的 ONNX 模型一起通过 Triton 提供服务

  1. detection_preprocessing

  2. detection_postprocessing

  3. recognition_postprocessing

您可以在此目录的 model_repository 文件夹中找到每个模型的完整 model.py 脚本。

让我们来看一个例子。在 model.py 中，我们为 TritonPythonModel 创建一个类定义，其中包含以下方法

class TritonPythonModel:
    def initialize(self, args):
        ...
    def execute(self, requests):
        ...
    def finalize(self):
        ...

initialize 和 finalize 方法是可选的，分别在模型加载和卸载时调用。大部分逻辑将进入 execute 方法，该方法接受请求对象的列表，并且必须返回响应对象的列表。

在原始客户端中，我们有以下代码来读取图像并对其执行一些简单的转换

### client.py

image = cv2.imread("./img1.jpg")
image_height, image_width, image_channels = image.shape

# Pre-process image
blob = cv2.dnn.blobFromImage(image, 1.0, (inpWidth, inpHeight), (123.68, 116.78, 103.94), True, False)
blob = np.transpose(blob, (0, 2,3,1))

# Create input object
input_tensors = [
    httpclient.InferInput('input_images:0', blob.shape, "FP32")
]
input_tensors[0].set_data_from_numpy(blob, binary_data=True)

在 python 后端执行时，我们需要确保我们的代码可以处理输入列表。此外，我们不会从磁盘读取图像，而是直接从 Triton 服务器提供的输入张量中检索它们。

### model.py

responses = []
for request in requests:
    # Read input tensor from Triton
    in_0 = pb_utils.get_input_tensor_by_name(request, "detection_preprocessing_input")
    img = in_0.as_numpy()
    image = Image.open(io.BytesIO(img.tobytes()))

    # Pre-process image
    img_out = image_loader(image)
    img_out = np.array(img_out)*255.0

    # Create object to send to next model
    out_tensor_0 = pb_utils.Tensor("detection_preprocessing_output", img_out.astype(output0_dtype))
    inference_response = pb_utils.InferenceResponse(output_tensors=[out_tensor_0])
    responses.append(inference_response)
return responses

使用模型集成将模型绑定在一起#

现在我们已经准备好单独部署管道的每个部分，我们可以创建一个集成“模型”，它可以按顺序执行每个模型，并在每个模型之间传递各种输入和输出。

为此，我们将在模型仓库中创建另一个条目

ensemble_model/
├── 1
└── config.pbtxt

这一次，我们只需要配置文件来描述我们的集成以及一个空的版本文件夹（您需要使用 mkdir -p model_repository/ensemble_model/1 创建）。在配置文件中，我们将定义集成的执行图。此图描述了集成的总体输入和输出，以及数据将如何以有向无环图的形式流经模型。下图是我们的模型管道的图形表示。菱形表示集成的最终输入和输出，这是客户端将与之交互的所有内容。圆圈是不同的已部署模型，矩形是在模型之间传递的张量。

flowchart LR
    in{input image} --> m1((detection_preprocessing))
    m1((detection_preprocessing)) --> t1((preprocessed_image))
    t1((preprocessed_image)) --> m2((text_detection))
    m2((text_detection)) --> t2(Sigmoid:0)
    m2((text_detection)) --> t3(concat_3:0)
    t2(Sigmoid:0) --> m3((detection_postprocessing))
    t3(concat_3:0) --> m3((detection_postprocessing))
    t1(preprocessed_image) --> m3((detection_postprocessing))
    m3((detection_postprocessing)) --> t4(cropped_images)
    t4(cropped_images) --> m4((text_recognition))
    m4((text_recognition)) --> t5(recognition_output)
    t5(recognition_output) --> m5((recognition_postprocessing))
    m5((recognition_postprocessing)) --> out{recognized_text}

为了向 Triton 表示此图，我们将创建以下配置文件。请注意，我们如何将平台定义为 "ensemble" 并指定集成本身的输入和输出。然后，在 ensemble_scheduling 块中，我们为集成的每个 step 创建一个条目，其中包括要执行的模型的名称，以及该模型的输入和输出如何映射到完整集成或其他模型的输入和输出。

展开以查看集成配置文件 
name: "ensemble_model"
platform: "ensemble"
max_batch_size: 256
input [
  {
    name: "input_image"
    data_type: TYPE_UINT8
    dims: [ -1 ]
  }
]
output [
  {
    name: "recognized_text"
    data_type: TYPE_STRING
    dims: [ -1 ]
  }
]

ensemble_scheduling {
  step [
    {
      model_name: "detection_preprocessing"
      model_version: -1
      input_map {
        key: "detection_preprocessing_input"
        value: "input_image"
      }
      output_map {
        key: "detection_preprocessing_output"
        value: "preprocessed_image"
      }
    },
    {
      model_name: "text_detection"
      model_version: -1
      input_map {
        key: "input_images:0"
        value: "preprocessed_image"
      }
      output_map {
        key: "feature_fusion/Conv_7/Sigmoid:0"
        value: "Sigmoid:0"
      },
      output_map {
        key: "feature_fusion/concat_3:0"
        value: "concat_3:0"
      }
    },
    {
      model_name: "detection_postprocessing"
      model_version: -1
      input_map {
        key: "detection_postprocessing_input_1"
        value: "Sigmoid:0"
      }
      input_map {
        key: "detection_postprocessing_input_2"
        value: "concat_3:0"
      }
      input_map {
        key: "detection_postprocessing_input_3"
        value: "preprocessed_image"
      }
      output_map {
        key: "detection_postprocessing_output"
        value: "cropped_images"
      }
    },
    {
      model_name: "text_recognition"
      model_version: -1
      input_map {
        key: "INPUT__0"
        value: "cropped_images"
      }
      output_map {
        key: "OUTPUT__0"
        value: "recognition_output"
      }
    },
    {
      model_name: "recognition_postprocessing"
      model_version: -1
      input_map {
        key: "recognition_postprocessing_input"
        value: "recognition_output"
      }
      output_map {
        key: "recognition_postprocessing_output"
        value: "recognized_text"
      }
    }
  ]
}

启动 Triton#

我们将再次使用 docker 容器启动 Triton。这一次，我们将在容器内启动一个交互式会话，而不是直接启动 triton 服务器。

docker run --gpus=all -it --shm-size=1G --rm  \
  -p8000:8000 -p8001:8001 -p8002:8002 \
  -v ${PWD}:/workspace/ -v ${PWD}/model_repository:/models \
  nvcr.io/nvidia/tritonserver:22.12-py3

我们需要为我们的 Python 后端脚本安装几个依赖项。

pip install torchvision opencv-python-headless

然后，我们可以启动 Triton

tritonserver --model-repository=/models

创建新客户端#

现在，我们已将先前客户端的大部分复杂性移至不同的 Triton 后端脚本中，我们可以创建一个更简化的客户端来与 Triton 通信。

## client.py

import tritonclient.grpc as grpcclient
import numpy as np

client = grpcclient.InferenceServerClient(url="localhost:8001")

image_data = np.fromfile("img1.jpg", dtype="uint8")
image_data = np.expand_dims(image_data, axis=0)

input_tensors = [grpcclient.InferInput("input_image", image_data.shape, "UINT8")]
input_tensors[0].set_data_from_numpy(image_data)
results = client.infer(model_name="ensemble_model", inputs=input_tensors)
output_data = results.as_numpy("recognized_text").astype(str)
print(output_data)

现在，通过执行以下命令运行完整的推理管道

python client.py

您应该看到解析后的文本打印到您的控制台。

下一步#

在本示例中，我们展示了如何使用模型集成通过单个网络调用在 Triton 上执行多个模型。当您的模型管道采用有向无环图的形式时，模型集成是一个很好的解决方案。但是，并非所有管道都可以用这种方式表达。例如，如果您的管道逻辑需要条件分支或循环执行，您可能需要一种更具表现力的方式来定义管道。在下一个示例中，我们将探讨如何使用 业务逻辑脚本 在 Python 中创建更复杂的管道。

在本页中