Triton Inference Server を使用してモデルサービスをデプロイする - Platform For AI

Triton Inference Server は、AI 推論を効率化するオープンソースの推論サービングエンジンです。 TensorRT、TensorFlow、PyTorch、ONNX など、複数のディープラーニングおよび機械学習フレームワークの AI モデルをオンライン推論サービスとしてデプロイできます。 Triton Inference Server は、マルチモデル管理もサポートしており、カスタムバックエンドを追加できるバックエンド API を提供します。このトピックでは、Triton Inference Server イメージを使用して、Platform for AI (PAI) にモデルサービスをデプロイする方法について説明します。

単一モデルサービスをデプロイする

Object Storage Service (OSS) バケットにモデルディレクトリを作成し、モデルディレクトリの形式要件に基づいてモデルファイルとモデル構成ファイルを構成します。詳細については、「ディレクトリを管理する」をご参照ください。

各モデルディレクトリには、少なくとも 1 つのバージョンサブディレクトリと 1 つのモデル構成ファイルが必要です。

バージョンサブディレクトリ: モデルファイルを保存します。バージョンサブディレクトリの名前は、モデルバージョンを示す数値である必要があります。数値が大きいほど、モデルのバージョンが新しくなります。
モデル構成ファイル: モデルに関する基本情報を保存します。ほとんどの場合、このファイルの名前は config.pbtxt です。

たとえば、モデルが oss://examplebucket/models/triton/ ディレクトリに保存されており、ディレクトリが次の構造で編成されているとします。

triton
└──resnet50_pt
    ├── 1
    │   └── model.pt
    ├── 2
    │   └── model.pt
    ├── 3
    │   └── model.pt
    └── config.pbtxt

config.pbtxt ファイルは、モデルの構成を指定します。例:

name: "resnet50_pt"
platform: "pytorch_libtorch"
max_batch_size: 128
input [
  {
    name: "INPUT__0"
    data_type: TYPE_FP32
    dims: [ 3, -1, -1 ]
  }
]
output [
  {
    name: "OUTPUT__0"
    data_type: TYPE_FP32
    dims: [ 1000 ]
  }
]

# 推論に GPU リソースを使用します。
# instance_group [
#   { 
#     kind: KIND_GPU
#   }
# ]

# モデルのバージョンポリシーを指定します。
# version_policy: { all { }}
# version_policy: { latest: { num_versions: 2}}
# version_policy: { specific: { versions: [1,3]}}

次の表に、config.pbtxt ファイルの主要なパラメーターを示します。

パラメーター	必須	説明
name	いいえ	モデルの名前。デフォルト値はモデルディレクトリの名前です。このパラメーターを指定する場合、値はモデルディレクトリの名前と一致する必要があります。
platform/backend	はい	次のパラメーターの少なくとも 1 つを指定します。 platform: モデルのトレーニングに使用されるフレームワーク。一般的な値: tensorrt_plan、onnxruntime_onnx、pytorch_libtorch、tensorflow_savedmodel、tensorflow_graphdef。 backend: モデルの実行に使用するメソッド。このパラメーターを指定するには、次のメソッドを使用できます。モデルのトレーニングに使用するフレームワークを指定します。一般的な値: tensorrt、onnxruntime、pytorch、tensorflow。同じフレームワークでも、platform パラメーターと backend パラメーターの有効な値は異なることに注意してください。 Python コードを使用してカスタム推論ロジックを構成するバックエンドの名前を指定します。詳細については、「カスタムバックエンドを使用する」をご参照ください。
max_batch_size	はい	モデルが同時に処理できるリクエストの最大数。このパラメーターを 0 に設定すると、バッチ処理は無効になります。
input	はい	次のプロパティが含まれています。 name: データの名前。 data_type: データのタイプ。 dims: データのディメンション。
output	はい	次のプロパティが含まれています。 name: データの名前。 data_type: データのタイプ。 dims: データのディメンション。
instance_group	いいえ	モデルの実行に使用される計算リソース。 GPU リソースが使用可能な場合、モデルは自動的に GPU リソースを使用して推論を実行します。使用できない場合、モデルは CPU リソースを使用して推論を実行します。使用する計算リソースは、次の形式で指定できます。 `instance_group [ { kind: KIND_GPU } ]` kind プロパティの有効な値: KIND_GPU および KIND_CPU。
version_policy	いいえ	モデルのバージョンポリシー。 resnet50_pt モデルのサンプル構成: `version_policy: { all { }} version_policy: { latest: { num_versions: 2}} version_policy: { specific: { versions: [1,3]}}` このパラメーターを空のままにすると、モデルの最新バージョンがロードされます。たとえば、resnet50_pt モデルでこのパラメーターを空のままにすると、モデルのバージョン 3 がロードされます。 all{}: モデルのすべてのバージョンをロードします。前の例では、resnet50_pt モデルのバージョン 1、2、3 がロードされます。 latest{num_versions:}: 最新の n 個のバージョンをロードします。ここで、n は num_versions の値です。前の例では、`num_versions: 2` は、resnet50_pt モデルの最新の 2 つのバージョン (バージョン 2 と 3) がロードされることを示します。 specific{versions:[]}: 特定のバージョンをロードします。前の例では、resnet50_pt モデルのバージョン 1 と 3 がロードされます。

Triton Inference Server サービスをデプロイします。

Triton Inference Server は、次の 2 つのポートをサポートしています。デフォルトでは、シナリオベースのモデルデプロイではポート 8000 が使用されます。ポート 8001 を使用する場合は、手順 e を実行します。使用しない場合は、手順 e を無視します。

8000: ポート 8000 で HTTP サーバーを起動して、HTTP リクエストを受信します。
8001: ポート 8001 で Google 遠隔手続き呼出し (gRPC) サーバーを起動して、gRPC リクエストを受信します。

次の手順を実行します。

PAI コンソールにログオンします。ページ上部でリージョンを選択します。次に、目的のワークスペースを選択し、[Elastic Algorithm Service (EAS)] をクリックします。
[Elastic Algorithm Service (EAS)] ページで、[サービスのデプロイ] をクリックします。 [シナリオベースのモデルデプロイ] セクションで、[Triton デプロイメント] をクリックします。

[Triton デプロイメント] ページで、次のパラメーターを構成します。その他のパラメーターについては、「カスタムデプロイメント」をご参照ください。

パラメーター	説明
[サービス名]	サービスの名前。
[モデル設定]	この例では、[タイプ] に [OSS] を選択し、[OSS] を手順 1 で準備したモデルファイルが保存されている OSS パスに設定します。例: `oss://example/models/triton/`。

(オプション) ページの右上隅にある [カスタムデプロイメントに変換] をクリックします。 [環境情報] セクションで、[ポート番号] を 8001 に変更します。 [サービス構成] セクションで、次の構成を追加します。
説明
デフォルトでは、サービスはポート 8000 で HTTP サービスを起動して、HTTP リクエストを受信します。 gRPC リクエストを受信するには、ポート番号を 8001 に変更する必要があります。システムはポート 8001 で gRPC サーバーを起動します。
```
"metadata": {
    "enable_http2": true
},
"networking": {
    "path": "/"
}
```
パラメーターを構成した後、[デプロイ] をクリックします。

マルチモデルサービスをデプロイする

マルチモデルサービスをデプロイする方法は、Elastic Algorithm Service (EAS) で単一モデルサービスをデプロイする方法と似ています。マルチモデルサービスをデプロイするには、複数のモデルのディレクトリを作成する必要があります。次のコードは例を示しています。サービスはすべてのモデルをロードし、同じサービスにすべてのモデルをデプロイします。詳細については、「単一モデルサービスをデプロイする」をご参照ください。

triton
├── resnet50_pt
|   ├── 1
|   │   └── model.pt
|   └── config.pbtxt
├── densenet_onnx
|   ├── 1
|   │   └── model.onnx
|   └── config.pbtxt
└── mnist_savedmodel
    ├── 1
    │   └── model.savedmodel
    │       ├── saved_model.pb
    |       └── variables
    |           ├── variables.data-00000-of-00001
    |           └── variables.index
    └── config.pbtxt

カスタムバックエンドを使用する

Triton バックエンドは、モデルの推論プロセスを実装します。バックエンドは、TensorRT、ONNX Runtime、PyTorch、TensorFlow などの既存のフレームワークを使用することも、前処理操作や後処理操作などのカスタム推論ロジックを実装することもできます。

バックエンドは、C++ または Python を使用して実装できます。 Python は C++ よりも柔軟性と利便性に優れています。このセクションでは、Python バックエンドを実装する方法について説明します。

モデルディレクトリの構造を変更します。

次の例は、PyTorch モデルに必要なディレクトリ構造を示しています。

resnet50_pt
├── 1
│   ├── model.pt
│   └── model.py
└── config.pbtxt

カスタムバックエンドを使用するには、モデルバージョンを表すサブディレクトリに model.py ファイルを追加し、config.pbtxt ファイルを変更する必要があります。

model.py ファイルを追加します。

model.py ファイルには、カスタム推論ロジックが含まれています。 TritonPythonModel という名前のクラスを定義し、ビジネス要件に基づいて initialize 関数、execute 関数、および finalize 関数を実装する必要があります。サンプルコード:

import json
import os
import torch
from torch.utils.dlpack import from_dlpack, to_dlpack

import triton_python_backend_utils as pb_utils


class TritonPythonModel:
    """クラス名は TritonPythonModel である必要があります。"""

    def initialize(self, args):
        """
        initialize() 関数はオプションです。 モデルのロードプロセス中にのみ呼び出され、モデルのプロパティや構成などのモデルに関する情報を初期化します。
        パラメーター
        ----------
        args: データをキーと値のペアとして保存する辞書。 キーと値は文字列型です。 有効なキー:
          * model_config: JSON 形式のモデル構成。
          * model_instance_kind: モデルの実行に使用されるデバイスのタイプ。
          * model_instance_device_id: モデルの実行に使用されるデバイスの ID。
          * model_repository: モデルリポジトリのパス。
          * model_version: モデルのバージョン。
          * model_name: モデルの名前。
        """

        # モデル構成を指定する JSON 文字列を Python 辞書に変換します。
        self.model_config = model_config = json.loads(args["model_config"])

        # モデル構成ファイルからプロパティを抽出します。
        output_config = pb_utils.get_output_config_by_name(model_config, "OUTPUT__0")

        # Triton タイプを NumPy タイプに変換します。
        self.output_dtype = pb_utils.triton_string_to_numpy(output_config["data_type"])

        # モデルリポジトリのパスを取得します。
        self.model_directory = os.path.dirname(os.path.realpath(__file__))

        # モデルの実行に使用されるデバイスを取得します。 この例では、モデルは GPU デバイスで実行されます。
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        print("device: ", self.device)

        model_path = os.path.join(self.model_directory, "model.pt")
        if not os.path.exists(model_path):
            raise pb_utils.TritonModelException("pytorch モデルが見つかりません")
        # .to(self.device) を使用して、PyTorch モデルを GPU デバイスにロードします。
        self.model = torch.jit.load(model_path).to(self.device)

        print("初期化済み...")

    def execute(self, requests):
        """
        execute() 関数は必須です。 モデルが推論リクエストを受信するたびに呼び出されます。 モデルでバッチ処理をサポートする場合は、execute() 関数にバッチ処理ロジックを追加する必要があります。
        パラメーター
        ----------
        requests: リクエストのリスト。 各リクエストは pb_utils.InferenceRequest タイプです。

        戻り値
        -------
        レスポンスのリスト。 各レスポンスは pb_utils.InferenceResponse タイプです。 レスポンスリストの長さは、リクエストリストの長さと等しくなければなりません。
        """

        output_dtype = self.output_dtype

        responses = []

        # リクエストリストをトラバースし、各リクエストのレスポンスを作成します。
        for request in requests:
            # 入力テンソルを取得します。
            input_tensor = pb_utils.get_input_tensor_by_name(request, "INPUT__0")
            # Triton テンソルを PyTorch テンソルに変換します。
            pytorch_tensor = from_dlpack(input_tensor.to_dlpack())

            if pytorch_tensor.shape[2] > 1000 or pytorch_tensor.shape[3] > 1000:
                responses.append(
                    pb_utils.InferenceResponse(
                        output_tensors=[],
                        error=pb_utils.TritonError(
                            "画像の形状は 1000 以下にする必要があります"
                        ),
                    )
                )
                continue

            # GPU デバイスで推論を実行します。
            prediction = self.model(pytorch_tensor.to(self.device))

            # PyTorch 出力テンソルを Triton テンソルに変換します。
            out_tensor = pb_utils.Tensor.from_dlpack("OUTPUT__0", to_dlpack(prediction))

            inference_response = pb_utils.InferenceResponse(output_tensors=[out_tensor])
            responses.append(inference_response)

        return responses

    def finalize(self):
        """
        finalize() 関数はオプションです。 リソースを解放するために、モデルのアンロード時に呼び出されます。
        """
        print("クリーンアップ中...")

重要

GPU デバイスで推論を実行する場合は、config.pbtxt ファイルの instance_group.kind プロパティを使用しないでください。代わりに、model.to(torch.device("cuda")) 関数を呼び出して、モデルを GPU デバイスにロードします。モデルがリクエストを受信したら、pytorch_tensor.to(torch.device("cuda")) 関数を呼び出して、モデル入力テンソルを GPU デバイスに送信します。こうすることで、モデルデプロイメント用に GPU リソースを構成した後、GPU デバイスで推論を実行できます。
モデルでバッチ処理をサポートする場合は、config.pbtxt ファイルの max_batch_size パラメーターを使用しないでください。代わりに、execute() 関数にバッチ処理ロジックを実装します。
各リクエストは 1 つのレスポンスに対応している必要があります。

config.pbtxt ファイルを変更します。
サンプル構成:
```
name: "resnet50_pt"
backend: "python"
max_batch_size: 128
input [
  {
    name: "INPUT__0"
    data_type: TYPE_FP32
    dims: [ 3, -1, -1 ]
  }
]
output [
  {
    name: "OUTPUT__0"
    data_type: TYPE_FP32
    dims: [ 1000 ]
  }
]

parameters: {
    key: "FORCE_CPU_ONLY_INPUT_TENSORS"
    value: {string_value: "no"}
}
```
次のパラメーターを変更し、その他のパラメーターの構成は保持します。
- backend: 値を python に設定します。
- parameters: このパラメーターはオプションです。モデルが GPU デバイスで推論を実行する場合は、FORCE_CPU_ONLY_INPUT_TENSORS を no に設定します。これにより、推論中に CPU と GPU の間で入力テンソルをコピーすることによって発生する不要なオーバーヘッドを防ぎます。

モデルをデプロイします。
Python バックエンドを使用する場合は、共有メモリを構成する必要があります。次の構成を使用して、カスタム推論ロジックを含むモデルサービスを作成できます。クライアントを使用してモデルサービスをデプロイする方法については、「EASCMD を使用してモデルサービスをデプロイする」をご参照ください。
```
{
  "metadata": {
    "name": "triton_server_test",
    "instance": 1,
  },
  "cloud": {
        "computing": {
            "instance_type": "ml.gu7i.c8m30.1-gu30",
            "instances": null
        }
    },
  "containers": [
    {
      "command": "tritonserver --model-repository=/models",
      "image": "eas-registry-vpc.<region>.cr.aliyuncs.com/pai-eas/tritonserver:23.02-py3",
      "port": 8000,
      "prepare": {
        "pythonRequirements": [
          "torch==2.0.1"
        ]
      }
    }
  ],
  "storage": [
    {
      "mount_path": "/models",
      "oss": {
        "path": "oss://oss-test/models/triton_backend/"
      }
    },
    {
      "empty_dir": {
        "medium": "memory",
        // 共有メモリを 1 GB に設定します。
        "size_limit": 1
      },
      "mount_path": "/dev/shm"
    }
  ]
}
```
次のパラメーターに注意してください。
- name: カスタムロジックを含むモデルの名前。
- storage.oss.path: モデルディレクトリの OSS パス。
- containers.image: デプロイメントに使用されるイメージ。 <region> を現在のリージョンの ID に置き換えます。たとえば、中国 (上海) リージョンには cn-shanghai を指定できます。

サービスを呼び出す

クライアントを構成して、デプロイされたサービスに推論リクエストを送信できます。

HTTP リクエストを送信する

ポート番号を 8000 に設定した場合、サービスに HTTP リクエストを送信できます。 Python サンプルコード:

import numpy as np
import tritonclient.http as httpclient

# url は、EAS にデプロイしたサービスにアクセスするために使用されるエンドポイントを指定します。
url = '1859257******.cn-hangzhou.pai-eas.aliyuncs.com/api/predict/triton_server_test'

triton_client = httpclient.InferenceServerClient(url=url)

image = np.ones((1,3,224,224))
image = image.astype(np.float32)

inputs = []
inputs.append(httpclient.InferInput('INPUT__0', image.shape, "FP32"))
inputs[0].set_data_from_numpy(image, binary_data=False)
outputs = []
outputs.append(httpclient.InferRequestedOutput('OUTPUT__0', binary_data=False)) # 1000 次元ベクトルを取得します。

# モデルの名前、トークン、入力、出力を指定します。
results = triton_client.infer(
    model_name="<model_name>",
    model_version="<version_num>",
    inputs=inputs,
    outputs=outputs,
    headers={"Authorization": "<test-token>"},
)
output_data0 = results.as_numpy('OUTPUT__0')
print(output_data0.shape)
print(output_data0)

次の表に、主要なパラメーターを示します。

パラメーター	説明
url	`http://` プレフィックスのないサービスのエンドポイント。エンドポイントを取得するには、次の手順を実行します。 [Elastic Algorithm Service (EAS)] ページに移動し、サービスを見つけて、その名前をクリックします。表示されるページの [サービスの詳細] タブで、[エンドポイント情報の表示] をクリックします。呼び出しメソッドダイアログボックスのパブリックエンドポイントタブで、パブリックエンドポイントを表示します。
model_name	モデルの名前。例: resnet50_pt。
model_version	使用するモデルバージョン。リクエストは、一度に 1 つのバージョンのモデルにのみ送信できます。
headers	サービスのトークン。 <test-token> をサービスのトークンに置き換えます。トークンは [パブリックエンドポイント] タブで表示できます。

gRPC リクエストを送信する

ポート番号を 8001 に設定した場合、必要な構成を追加した後に、サービスに gRPC リクエストを送信できます。 Python サンプルコード:

#!/usr/bin/env python
import grpc
from tritonclient.grpc import service_pb2, service_pb2_grpc
import numpy as np

if __name__ == "__main__":
    # サービスのエンドポイントを定義します。
    host = (
        "service_name.115770327099****.cn-beijing.pai-eas.aliyuncs.com:80"
    )
    # test-token をサービスのトークンに置き換えます。
    token = "test-token"
    # モデル名とバージョンを指定します。
    model_name = "resnet50_pt"
    model_version = "1"
    
    # トークン検証用の gRPC メタデータを作成します。
    metadata = (("authorization", token),)

    # サーバーと通信するための gRPC チャネルと gRPC スタブを作成します。
    channel = grpc.insecure_channel(host)
    grpc_stub = service_pb2_grpc.GRPCInferenceServiceStub(channel)
    
    # 推論リクエストを作成します。
    request = service_pb2.ModelInferRequest()
    request.model_name = model_name
    request.model_version = model_version
    
    # モデル構成ファイルで指定した入力パラメーターに基づいて、入力テンソルを作成します。
    input = service_pb2.ModelInferRequest().InferInputTensor()
    input.name = "INPUT__0"
    input.datatype = "FP32"
    input.shape.extend([1, 3, 224, 224])
     # モデル構成ファイルで指定した出力パラメーターに基づいて、出力テンソルを作成します。
    output = service_pb2.ModelInferRequest().InferRequestedOutputTensor()
    output.name = "OUTPUT__0"
    
    # リクエストに入力データを追加します。
    request.inputs.extend([input])
    request.outputs.extend([output])
    # ランダム配列を作成し、配列を一連のバイトにシリアル化することで入力データを作成します。
    request.raw_input_contents.append(np.random.rand (1,3, 224, 224).astype(np.float32).tobytes()) # 浮動小数点数の配列
        
    # 推論リクエストを送信し、レスポンスを受信します。
    response, _ = grpc_stub.ModelInfer.with_call(request, metadata=metadata)
    
    # レスポンスから出力テンソルを抽出します。
    output_contents=response.raw_output_contents [0]  # たとえば、1 つの出力テンソルのみが返されます。
    output_shape = [1, 1000]  # たとえば、出力テンソルの形状は [1, 1000] です。
    
    # 出力バイトを NumPy 配列に変換します。
    output_array = np.frombuffer(output_contents, dtype=np.float32)
    output_array = output_array.reshape(output_shape)
    
    # モデルの出力を出力します。
    print("Model output:\n", output_array)

次の表に、主要なパラメーターを示します。

パラメーター	説明
host	`http://` プレフィックスがなく、`:80` サフィックスが付いたサービスのエンドポイント。エンドポイントを取得するには、次の操作を実行します。 [Elastic Algorithm Service (EAS)] ページに移動し、サービスを見つけて、その名前をクリックします。表示されるページの [サービスの詳細] タブで、[エンドポイント情報の表示] をクリックします。呼び出しメソッドダイアログボックスのパブリックエンドポイントタブで、パブリックエンドポイントを表示します。
token	サービスのトークン。 <test-token> をサービスのトークンに置き換えます。トークンは [パブリックエンドポイント] タブで表示できます。
model_name	モデルの名前。例: resnet50_pt。
model_version	使用するモデルバージョン。リクエストは、一度に 1 つのバージョンのモデルにのみ送信できます。

参考資料

TensorFlow Serving を使用して EAS にモデルサービスをデプロイする方法については、「TensorFlow Serving イメージを使用してモデルサービスをデプロイする」をご参照ください。
自動ストレステストツールを使用して、EAS にデプロイされたサービスのパフォーマンスをテストできます。詳細については、「サービスの自動ストレステスト」をご参照ください。
ビジネスシナリオに一致するモデルサービスをデプロイした後、サービスを呼び出してモデルのパフォーマンスを確認できます。適用可能なシナリオについては、「EAS ユースケース」をご参照ください。