Platform For AI:Triton Inference Server イメージのデプロイ

手順1：モデルリポジトリの準備

Triton では、Object Storage Service (OSS) バケット内に特定のディレクトリ構造が必要です。次の形式でディレクトリを作成します。手順については、「Manage directories」および「Upload files」をご参照ください。

oss://your-bucket/models/triton/
└── your_model_name/
    ├── 1/                    # バージョンディレクトリ (数値である必要があります)
    │   └── model.pt          # モデルファイル
    └── config.pbtxt          # モデル設定ファイル

主な要件：

• バージョンディレクトリ名は、1、2、3 などの数値である必要があります。

• 数値が大きいほど新しいバージョンを示します。

• 各モデルには config.pbtxt 設定ファイルが必要です。

手順2：モデル設定ファイルの作成

config.pbtxt ファイルを作成し、モデルの基本情報を設定します。次に例を示します。

name: "your_model_name"
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: { latest: { num_versions: 1 }}

# すべてのバージョンをロードします
# version_policy: { all { }}

# 最新 2 バージョンをロードします
# version_policy: { latest: { num_versions: 2 }}

# 特定のバージョンをロードします
# version_policy: { specific: { versions: [1, 3] }}

手順3：サービスのデプロイ

1. PAI コンソールにログインし、上部メニューで対象リージョンを選択します。

2. 左側のナビゲーションペインで、Elastic Algorithm Service (EAS) をクリックし、対象のワークスペースを選択して、モデルサービスの作成 をクリックします。

3. シナリオ別モデルデプロイ セクションで、Triton デプロイメント をクリックします。

4. デプロイパラメータを設定します。

   • サービス名：サービスのカスタム名を入力します。

   • [リソースとモデル]：設定タイプとして OSS パス を選択し、たとえば oss://your-bucket/models/triton/ のようなモデルリポジトリへのパスを入力します。

   • インスタンス数： と リソース仕様：要件に基づいて値を選択します。必要な GPU メモリの見積もりについては、「Estimate the GPU memory required for large models」をご参照ください。

5. デプロイメント をクリックし、サービスが開始するまで待ちます。

手順4：gRPC の有効化 (オプション)

デフォルトでは、Triton はポート 8000 で HTTP サービスを提供します。gRPC を使用するには、次の手順を実行します。

1. サービス設定ページの右上隅で、カスタムデプロイメントへの切り替え をクリックします。

2. 環境コンテキスト セクションで、ポート番号 を 8001 に設定します。

3. サービス機能の設定 > 高度なネットワーク で、gRPC オプションを有効にします。

4. デプロイメント をクリックします。

ディレクトリ構造

your_model_name/
├── 1/
│   ├── model.pt          # モデルファイル
│   └── model.py          # カスタム推論ロジック
└── config.pbtxt

Python バックエンドの実装

model.py ファイルを作成し、TritonPythonModel クラスを定義します。

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):
        """
        このオプション関数は、モデルのロード時に Triton によって一度だけ呼び出されます。
        モデルのプロパティや設定に関する情報を初期化するために使用します。
        Parameters
        ----------
        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("Cannot find the pytorch model")
        # .to(self.device) を使用して PyTorch モデルを GPU にロードします。
        self.model = torch.jit.load(model_path).to(self.device)

        print("Initialized...")

    def execute(self, requests):
        """
        この必須関数はモデルを実行します。Triton はすべての推論リクエストに対してこの関数を呼び出します。
        バッチ処理が有効な場合、バッチ処理ロジックを実装する必要があります。
        Parameters
        ----------
        requests : pb_utils.InferenceRequest オブジェクトのリスト。

        Returns
        -------
        pb_utils.InferenceResponse オブジェクトのリスト。このリストは requests リストと同じ長さでなければなりません。
        """

        output_dtype = self.output_dtype

        responses = []

        # requests リストを反復処理し、各リクエストに対応するレスポンスを作成します。
        for request in requests:
            # 入力テンソルを取得します。
            input_tensor = pb_utils.get_input_tensor_by_name(request, "INPUT__0")
            # Triton テンソルを Torch テンソルに変換します。
            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(
                            "Image shape should not be larger than 1000"
                        ),
                    )
                )
                continue

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

            # Torch の出力テンソルを 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):
        """
        このオプション関数は、モデルがアンロードされるときに呼び出されます。クリーンアップタスクに使用します。
        """
        print("Cleaning up...")

設定ファイルの更新

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 で推論を実行する場合は、CPU と GPU 間での入力テンソルのコピーによる不要なオーバーヘッドを避けるため、FORCE_CPU_ONLY_INPUT_TENSORS パラメータを no に設定します。

サービスのデプロイ

Python バックエンドを使用する場合、共有メモリを設定する必要があります。カスタムモデルのデプロイ > JSON 独立デプロイメントに移動し、次の JSON 設定を入力して、サービスをデプロイします。

{
  "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:25.03-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"
    }
  ]
}

主な JSON 設定パラメータ：

• containers[0].image：公式の Triton イメージです。<region> は、サービスが配置されているリージョンに置き換えます。

• containers[0].prepare.pythonRequirements：Python の依存関係をここに列挙します。EAS はサービス起動前に自動的にインストールします。

• storage：2つのマウント項目が含まれます。

  • 1つ目の項目は、OSS のモデルリポジトリのパスをコンテナ内の /models ディレクトリにマウントします。

  • 2つ目の項目は、共有メモリの 必須 設定です。Triton Server と Python バックエンドプロセスは、/dev/shm の共有メモリパスを使用して、パフォーマンスを最大化するためにゼロコピーでテンソルデータをやり取りします。size_limit は GB 単位で指定します。モデルと想定される同時実行数に基づいて必要なサイズを見積もってください。

サービスエンドポイントとトークンの取得

1. Elastic Algorithm Service (EAS) ページに移動し、お使いのサービスの名前をクリックします。

2. サービスの詳細 タブで エンドポイント情報の表示 をクリックし、インターネットエンドポイント と トークン をコピーします。

HTTP リクエストの送信

ポート番号が 8000 に設定されている場合、サービスは HTTP リクエストをサポートします。

import numpy as np
# tritonclient パッケージをインストールするには、次のコマンドを実行します: pip install tritonclient
import tritonclient.http as httpclient

# サービスのデプロイ後に生成されるパブリックエンドポイントです。http:// プレフィックスは含めないでください。
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="<your-model-name>",
    model_version="<version-num>",
    inputs=inputs,
    outputs=outputs,
    headers={"Authorization": "<your-service-token>"},
)
output_data0 = results.as_numpy('OUTPUT__0')
print(output_data0.shape)
print(output_data0)

gRPC リクエストの送信

ポートを 8001 に設定して gRPC を有効にすると、サービスは gRPC リクエストをサポートします。

#!/usr/bin/env python
import grpc
# tritonclient パッケージをインストールするには、次のコマンドを実行します: pip install tritonclient
from tritonclient.grpc import service_pb2, service_pb2_grpc
import numpy as np

if __name__ == "__main__":
    # サービスのデプロイ後に生成されるサービスエンドポイントです。http:// は含めないでください。末尾に :80 を追加します。
    host = (
        "service_name.115770327099****.cn-beijing.pai-eas.aliyuncs.com:80"
    )
    # サービストークンです。実際のアプリケーションでは、実際のトークンを使用してください。
    token = "<your-service-token>"
    # モデル名とバージョンです。
    model_name = "<your-model-name>"
    model_version = "<version-num>"
    
    # トークン認証用の gRPC メタデータを作成します。
    metadata = (("authorization", token),)

    # サーバーと通信するための 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 パラメータに対応する入力テンソルを構築します。
    input = service_pb2.ModelInferRequest().InferInputTensor()
    input.name = "INPUT__0"
    input.datatype = "FP32"
    input.shape.extend([1, 3, 224, 224])
    # モデル設定ファイルで定義された output パラメータに対応する出力テンソルを構築します。
    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]  # 出力テンソルの shape が [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)

詳細ログの有効化

verbose=True を設定すると、リクエストとレスポンスの JSON データが出力されます。

client = httpclient.InferenceServerClient(url=url, verbose=True)

出力例：

POST /api/predict/triton_test/v2/models/resnet50_pt/versions/1/infer, headers {'Authorization': '************1ZDY3OTEzNA=='}
b'{"inputs":[{"name":"INPUT__0","shape":[1,3,32,32],"datatype":"FP32","data":[1.0,1.0,1.0,.....,1.0]}],"outputs":[{"name":"OUTPUT__0","parameters":{"binary_data":false}}]}'

オンラインデバッグ

コンソールのオンラインデバッグ機能を使用して、サービスを直接テストできます。リクエスト URL に /api/predict/triton_test/v2/models/resnet50_pt/versions/1/infer を追記し、詳細ログの JSON リクエストデータをリクエストボディとして使用します。

リクエスト送信後にステータスコード 200 が返されます。レスポンスボディに model_version:"1" が含まれ、出力名は OUTPUT__0、データ型は FP32、shape は [1,1000]、データが浮動小数点数の配列になっていれば、推論リクエストは成功です。

Q：「no kernel image」という CUDA エラー

このエラーは、イメージのバージョンと GPU の互換性の問題が原因で発生します。A10 や T4 など、別の GPU インスタンスタイプへの切り替えを試してください。

Q：「url should not include the scheme」というエラー

このエラーは、サービス URL が誤っていることが原因で発生します。「サービスエンドポイントとトークンの取得」セクションで取得したエンドポイントの形式は http://17519301*******.cn-hangzhou.pai-eas.aliyuncs.com/api/predict/wen***** です (これは gRPC エンドポイントとは異なります)。http:// プレフィックスを削除する必要があります。

Q：「DNS resolution failed」という gRPC エラー

このエラーは、サービスホストが誤っていることが原因で発生します。「サービスエンドポイントとトークンの取得」セクションで取得したエンドポイントの形式は http://we*****.1751930*****.cn-hangzhou.pai-eas.aliyuncs.com/ です (これは HTTP エンドポイントとは異なります)。http:// プレフィックスと末尾の / を削除し、末尾に :80 を追加する必要があります。最終的な形式は we*****.1751930*****.cn-hangzhou.pai-eas.aliyuncs.com:80 になります。