Platform For AI:PyTorch モデルのトレーニングとデプロイ

SDK のインストールと構成

CLI で次のコマンドを実行して、PAI SDK for Python をインストールします。

python -m pip install "alipai>=0.4.0"

CLI で次のコマンドを実行して、PAI SDK for Python を構成します。

python -m pai.toolkit.config

詳細については、「PAI SDK for Python のインストールと構成」をご参照ください。

トレーニングデータの準備

この例では、MNIST データセットを使用して画像分類モデルをトレーニングします。PAI でトレーニングジョブを送信するには、データセットを準備して OSS バケットにアップロードする必要があります。

1. MNIST データセットをダウンロードします。

   次のシェルスクリプトを実行して、MNIST データセットを data という名前のオンプレミスディレクトリにダウンロードします。

   #!/bin/sh
   set -e
   
   url_prefix="https://ossci-datasets.s3.amazonaws.com/mnist/"
   # ダウンロードに時間がかかりすぎる場合は、次のアドレスを使用できます。
   # url_prefix="http://yann.lecun.com/exdb/mnist/"
   
   mkdir -p data/MNIST/raw/
   
   wget -nv ${url_prefix}train-images-idx3-ubyte.gz -P data/MNIST/raw/
   wget -nv ${url_prefix}train-labels-idx1-ubyte.gz -P data/MNIST/raw/
   wget -nv ${url_prefix}t10k-images-idx3-ubyte.gz -P data/MNIST/raw/
   wget -nv ${url_prefix}t10k-labels-idx1-ubyte.gz -P data/MNIST/raw/
   

2. データセットを OSS バケットにアップロードします。

   OSS が提供する CLI ツール ossutil を使用してデータをアップロードできます。 ossutil のインストールと使用方法については、「ossutil 1.0」をご参照ください。また、PAI SDK for Python が提供するメソッドを使用して、トレーニングデータを OSS バケットの /mnist/data/ パスにアップロードすることもできます。

   • ossutil を使用する:

     ossutil cp -rf ./data oss://<YourOssBucket>/mnist/data/

   • PAI SDK for Python を使用する:

     from pai.common.oss_utils import upload
     from pai.session import get_default_session
     
     sess = get_default_session()
     data_uri = upload("./data/", oss_path="mnist/data/", bucket=sess.oss_bucket)
     print(data_uri)
     

トレーニングスクリプトの準備

トレーニングジョブを送信する前に、PyTorch を使用してトレーニングスクリプトを作成する必要があります。この例で使用されるトレーニングスクリプトは、PyTorch によって提供される MNIST の例 に基づいて変更されています。変更には、データロードとモデル保存のロジックの変更が含まれます。

• 環境変数を使用して入力データパスを取得する

  estimator.fit(inputs={"train_data":data_uri}) を使用して、OSS に保存されているデータをトレーニングコンテナにマウントします。トレーニングスクリプトは、ローカルファイルを読み取ることで、マウントされたデータを取得できます。

  estimator.fit メソッドの inputs は DICT タイプです。各入力はチャネルであり、キーはチャネルの名前、値は保存されているデータのパスです。トレーニングスクリプトは、PAI_INPUT_{ChannelNameUpperCase} 変数によって、作業コンテナにマウントされたデータのパスを取得できます。

  次の内容に基づいて、データロードのコードを変更します。

  - dataset1 = datasets.MNIST("../data", train=True, download=True, transform=transform)
  - dataset2 = datasets.MNIST("../data", train=False, transform=transform)
  
  + # 環境変数を使用して入力データパスを取得します。
  + data_path = os.environ.get("PAI_INPUT_TRAIN_DATA", "../data")
  + dataset1 = datasets.MNIST(data_path, train=True, download=True, transform=transform)
  + dataset2 = datasets.MNIST(data_path, train=False, transform=transform)
  

• 環境変数を使用して出力モデルパスを取得する

  トレーニング環境の PAI_OUTPUT_MODEL 変数で指定されたパスにモデルを保存する必要があります。デフォルトのパスは /ml/output/model です。パス内のデータとモデルは OSS バケットに保存されます。

  次の内容に基づいて、モデル出力のコードを変更します。

  - if args.save_model:
  -     torch.save(model.state_dict(), "mnist_cnn.pt")
  + # モデルを保存します。
  + save_model(model)
  + 
  + def save_model(model):
  +     """モデルを TorchScript に変換し、指定されたパスに保存します。"""
  +     output_model_path = os.environ.get("PAI_OUTPUT_MODEL")
  +     os.makedirs(output_model_path, exist_ok=True)
  +     
  +     m = torch.jit.script(model)
  +     m.save(os.path.join(output_model_path, "mnist_cnn.pt"))
  

PAI が提供する組み込みの PyTorch プロセッサを使用してサービスを作成する場合、入力モデルは TorchScript 形式である必要があります。この例では、モデルは TorchScript 形式でエクスポートされます。トレーニングスクリプトのサンプル:

# source: https://github.com/pytorch/examples/blob/main/mnist/main.py
from __future__ import print_function

import argparse
import os

import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.optim.lr_scheduler import StepLR
from torchvision import datasets, transforms


class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.conv1 = nn.Conv2d(1, 32, 3, 1)
        self.conv2 = nn.Conv2d(32, 64, 3, 1)
        self.dropout1 = nn.Dropout(0.25)
        self.dropout2 = nn.Dropout(0.5)
        self.fc1 = nn.Linear(9216, 128)
        self.fc2 = nn.Linear(128, 10)

    def forward(self, x):
        x = self.conv1(x)
        x = F.relu(x)
        x = self.conv2(x)
        x = F.relu(x)
        x = F.max_pool2d(x, 2)
        x = self.dropout1(x)
        x = torch.flatten(x, 1)
        x = self.fc1(x)
        x = F.relu(x)
        x = self.dropout2(x)
        x = self.fc2(x)
        output = F.log_softmax(x, dim=1)
        return output


def train(args, model, device, train_loader, optimizer, epoch):
    model.train()
    for batch_idx, (data, target) in enumerate(train_loader):
        data, target = data.to(device), target.to(device)
        optimizer.zero_grad()
        output = model(data)
        loss = F.nll_loss(output, target)
        loss.backward()
        optimizer.step()
        if batch_idx % args.log_interval == 0:
            print(
                "Train Epoch: {} [{}/{} ({:.0f}%)]\tLoss: {:.6f}".format(  # 日本語訳: トレーニングエポック
                    epoch,
                    batch_idx * len(data),
                    len(train_loader.dataset),
                    100.0 * batch_idx / len(train_loader),
                    loss.item(),
                )
            )
            if args.dry_run:
                break


def test(model, device, test_loader):
    model.eval()
    test_loss = 0
    correct = 0
    with torch.no_grad():
        for data, target in test_loader:
            data, target = data.to(device), target.to(device)
            output = model(data)
            test_loss += F.nll_loss(
                output, target, reduction="sum"
            ).item()  # sum up batch loss #日本語訳: バッチ損失を合計
            pred = output.argmax(
                dim=1, keepdim=True
            )  # get the index of the max log-probability #日本語訳: 最大対数確率のインデックスを取得
            correct += pred.eq(target.view_as(pred)).sum().item()

    test_loss /= len(test_loader.dataset)

    print(
        "\nTest set: Average loss: {:.4f}, Accuracy: {}/{} ({:.0f}%)\n".format( # 日本語訳: テストセット
            test_loss,
            correct,
            len(test_loader.dataset),
            100.0 * correct / len(test_loader.dataset),
        )
    )


def main():
    # Training settings #日本語訳: トレーニング設定
    parser = argparse.ArgumentParser(description="PyTorch MNIST Example")
    parser.add_argument(
        "--batch-size",
        type=int,
        default=64,
        metavar="N",
        help="input batch size for training (default: 64)", #日本語訳: トレーニングの入力バッチサイズ (デフォルト: 64)
    )
    parser.add_argument(
        "--test-batch-size",
        type=int,
        default=1000,
        metavar="N",
        help="input batch size for testing (default: 1000)", #日本語訳: テストの入力バッチサイズ (デフォルト: 1000)
    )
    parser.add_argument(
        "--epochs",
        type=int,
        default=14,
        metavar="N",
        help="number of epochs to train (default: 14)", #日本語訳: トレーニングするエポック数 (デフォルト: 14)
    )
    parser.add_argument(
        "--lr",
        type=float,
        default=1.0,
        metavar="LR",
        help="learning rate (default: 1.0)", #日本語訳: 学習率 (デフォルト: 1.0)
    )
    parser.add_argument(
        "--gamma",
        type=float,
        default=0.7,
        metavar="M",
        help="Learning rate step gamma (default: 0.7)", #日本語訳: 学習率ステップガンマ (デフォルト: 0.7)
    )
    parser.add_argument(
        "--no-cuda", action="store_true", default=False, help="disables CUDA training" #日本語訳: CUDA トレーニングを無効にする
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        default=False,
        help="quickly check a single pass", #日本語訳: シングルパスをすばやくチェックする
    )
    parser.add_argument(
        "--seed", type=int, default=1, metavar="S", help="random seed (default: 1)" #日本語訳: ランダムシード (デフォルト: 1)
    )
    parser.add_argument(
        "--log-interval",
        type=int,
        default=10,
        metavar="N",
        help="how many batches to wait before logging training status", #日本語訳: トレーニングステータスをログに記録する前に待機するバッチ数
    )
    parser.add_argument(
        "--save-model",
        action="store_true",
        default=False,
        help="For Saving the current Model", #日本語訳: 現在のモデルを保存するため
    )
    args = parser.parse_args()
    use_cuda = not args.no_cuda and torch.cuda.is_available()

    torch.manual_seed(args.seed)

    device = torch.device("cuda" if use_cuda else "cpu")

    train_kwargs = {"batch_size": args.batch_size}
    test_kwargs = {"batch_size": args.test_batch_size}
    if use_cuda:
        cuda_kwargs = {"num_workers": 1, "pin_memory": True, "shuffle": True}
        train_kwargs.update(cuda_kwargs)
        test_kwargs.update(cuda_kwargs)

    transform = transforms.Compose(
        [transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))]
    )

    data_path = os.environ.get("PAI_INPUT_TRAIN_DATA", "../data") # 修正: 正しい環境変数名を使用
    dataset1 = datasets.MNIST(data_path, train=True, download=True, transform=transform)
    dataset2 = datasets.MNIST(data_path, train=False, transform=transform)
    train_loader = torch.utils.data.DataLoader(dataset1, **train_kwargs)
    test_loader = torch.utils.data.DataLoader(dataset2, **test_kwargs)

    model = Net().to(device)
    optimizer = optim.Adadelta(model.parameters(), lr=args.lr)

    scheduler = StepLR(optimizer, step_size=1, gamma=args.gamma)
    for epoch in range(1, args.epochs + 1):
        train(args, model, device, train_loader, optimizer, epoch)
        test(model, device, test_loader)
        scheduler.step()

    # Save the model. #日本語訳: モデルを保存します。
    save_model(model)


def save_model(model):
    """Convert the model to TorchScript and save it to the specified path.""" #日本語訳: モデルを TorchScript に変換し、指定されたパスに保存します。
    output_model_path = os.environ.get("PAI_OUTPUT_MODEL")
    os.makedirs(output_model_path, exist_ok=True)

    m = torch.jit.script(model)
    m.save(os.path.join(output_model_path, "mnist_cnn.pt"))


if __name__ == "__main__":
    main()

上記のトレーニングコードをオンプレミスディレクトリに保存し、Estimator を使用して PAI に送信します。この例では、train_src という名前の新しいディレクトリを作成し、トレーニングスクリプトを train_src/train.py に保存します。

|-- train_src # アップロードするトレーニングスクリプトのディレクトリ。
    |-- requirements.txt # オプション。トレーニングジョブのサードパーティの依存関係。
    '-- train.py # 保存されたトレーニングスクリプト

トレーニングジョブの送信

Estimator を使用すると、オンプレミスのトレーニングスクリプトとイメージを使用して、PAI でトレーニングジョブを実行できます。

• トレーニングジョブのスクリプトとコマンド

  source_dir パラメーターで指定されたトレーニングスクリプトのディレクトリは、ジョブが開始される前に OSS にアップロードされ、ジョブコンテナに準備されます。デフォルトのディレクトリは /ml/usercode です。 command パラメーターで指定された起動コマンドの作業ディレクトリも /ml/usercode です。

• トレーニングジョブのイメージ

  この例では、PAI が提供する PyTorch イメージを使用してトレーニングジョブを実行します。

• トレーニングジョブのハイパーパラメーター

  ${PAI_CONFIG_DIR}/hyperparameters.json ファイルを読み取るか、環境変数を使用することで、トレーニングジョブのハイパーパラメーターを取得できます。詳細については、「トレーニングジョブのプリセット環境変数」をご参照ください。

  この例では、実行されるコマンドは python train.py $PAI_USER_ARGS で、PAI_USER_ARGS 変数はハイパーパラメーターによって取得された文字列です。トレーニングジョブの最終的な起動コマンドは python train.py --epochs 5 --batch-size 256 --lr 0.5 です。

• metric_definitions を使用してトレーニングメトリックを指定する

  PAI では、標準出力と標準エラーを含む出力トレーニングログから正規表現を照合することで、トレーニングメトリックを取得できます。システムは詳細ページへのリンクも出力します。詳細ページでは、トレーニングジョブの詳細な構成、出力ログ、およびメトリックを表示できます。

• instance_type を使用してトレーニングジョブのインスタンスタイプを指定する

  PAI のトレーニングジョブでサポートされているインスタンスタイプの詳細については、「付録: パブリックリソースグループの料金詳細」をご参照ください。

Estimator コードのサンプル:

from pai.estimator import Estimator
from pai.image import retrieve

# GPU ベースのトレーニングに PyTorch 1.18 イメージを使用してトレーニングスクリプトを実行します。
image_uri = retrieve(
    "PyTorch", framework_version="1.8PAI", accelerator_type="GPU"
).image_uri
print(image_uri)

est = Estimator(
    # トレーニングジョブの起動コマンド。デフォルトの作業ディレクトリは /ml/usercode/ です。
    command="python train.py $PAI_USER_ARGS",
    # アップロードするトレーニングコードディレクトリの相対パスまたは絶対パス。
  	# デフォルトでは、トレーニング環境の /ml/usercode ディレクトリが使用されます。
    source_dir="./train_src/",
    # トレーニングジョブのイメージ。
    image_uri=image_uri,
    # インスタンス構成。
    instance_type="ecs.gn6i-c4g1.xlarge",  # 4vCPU 15GB 1*NVIDIA T4
    # トレーニングジョブのハイパーパラメーター。
    hyperparameters={
        "epochs": 5,
        "batch-size": 64 * 4,
        "lr": 0.5,
    },
    # トレーニングジョブのメトリック構成。
    metric_definitions=[
        {
            "Name": "loss", #日本語訳: 損失
            "Regex": r".*loss=([-+]?[0-9]*.?[0-9]+(?:[eE][-+]?[0-9]+)?).*",
        },
    ],
    base_job_name="pytorch_mnist",
)

OSS にアップロードされたトレーニングデータをインプットデータとして使用し、トレーニングジョブを実行します。

# ossutil を使用してトレーニングデータをアップロードする場合は、入力データの OSS URI を明示的に指定する必要があります。
# data_uri = "oss://<YourOssBucket>/mnist/data/"

# トレーニングジョブを送信します。
est.fit(
    inputs={
        "train_data": data_uri,
    }
)

# トレーニング済みモデルの出力パス。
print("TrainingJob output model data:") #日本語訳: TrainingJob 出力モデルデータ:
print(est.model_data())

est.fit メソッドを使用して、トレーニングジョブを PAI に送信して実行できます。トレーニングジョブが送信されると、SDK はジョブ詳細ページのリンクとトレーニングジョブのログを、ジョブの実行が完了するまで出力します。

OSS のデータを使用する必要がある場合は、inputs パラメーターを使用して、estimator.fit メソッドでデータを渡します。 inputs パラメーターを使用してデータを渡すと、データストレージパスが作業ディレクトリにアタッチされます。トレーニングスクリプトは、ローカルファイルを読み取ることでデータをロードできます。

トレーニングジョブの送信の詳細については、「トレーニングジョブの送信」をご参照ください。

推論サービスのデプロイ

トレーニングジョブが完了したら、estimator.model_data() メソッドを使用して、トレーニングジョブによって生成されたモデルの OSS パスを取得できます。次のセクションでは、トレーニング済みモデルをオンライン推論サービスとして PAI にデプロイする方法について説明します。

• InferenceSpec を使用して、モデルを使用して推論サービスを構築する方法を記述します。

  プロセッサまたはカスタムイメージを使用してモデルをデプロイできます。次の例では、両方の方法について説明します。

• Model.deploy メソッドを使用して、サービスで使用されるリソースやサービス名などの情報を構成し、推論サービスを作成します。

詳細については、「推論サービスのデプロイ」をご参照ください。

プロセッサを使用してモデルサービスをデプロイする

プロセッサには、推論パッケージの抽象的な説明が含まれています。モデルをロードし、モデル推論サービスを開始するために使用されます。モデル推論サービスは、ユーザーが呼び出すための API を提供します。PAI は組み込みの PyTorch プロセッサを提供しており、TorchScript 形式のモデルを PAI にデプロイして推論サービスを作成できます。

1. サービスをデプロイします。

   次の例では、PyTorch プロセッサを使用して、トレーニング済みモデルを推論サービスとしてデプロイします。サンプルコード:

   from pai.model import Model, InferenceSpec
   from pai.predictor import Predictor
   from pai.common.utils import random_str
   
   
   m = Model(
    model_data=est.model_data(),
    # PAI が提供する PyTorch プロセッサを使用します。
    inference_spec=InferenceSpec(processor="pytorch_cpu_1.10"),
   )
   
   p: Predictor = m.deploy(
    service_name="tutorial_pt_mnist_proc_{}".format(random_str(6)),
    instance_type="ecs.c6.xlarge",
   )
   
   print(p.service_name)
   print(p.service_status)
   

   Model.deploy はPredictor オブジェクトを返します。このオブジェクトには、デプロイされたサービス名とステータスなどの情報が含まれています。サービスが RUNNING ステータスになると、推論リクエストを送信できます。

2. 推論サービスを実行します。

   この例では、テストサンプルが numpy を使用して作成され、推論サービスに送信されます。

   import numpy as np
   
   # 入力は Float32 型で、(BatchSize, Channel, Weight, Height) の形式です。
   dummy_input = np.random.rand(2, 1, 28, 28).astype(np.float32)
   
   # np.random.rand(1, 1, 28, 28).dtype
   res = p.predict(dummy_input)
   print(res)
   
   print(np.argmax(res, 1))
   

3. インターフェイス サービスを削除します。

   予測が完了したら、Predictor.delete_service を使用して推論サービスを削除できます。

   p.delete_service()

イメージを使用してモデルをデプロイする

パフォーマンス重視のシナリオでは、プロセッサを使用してモデルをデプロイできます。モデルにサードパーティの依存関係がある場合や、推論サービスで前処理と後処理が必要な場合など、カスタム要件があるシナリオでは、イメージを使用してモデルをデプロイできます。PAI Python SDK は、pai.model.container_serving_spec() メソッドを提供します。このメソッドを使用すると、オンプレミスのコードと PAI が提供するイメージを使用して推論サービスを作成できます。

1. 推論サービスのコードファイルを準備します。

   モデルをデプロイする前に、モデルのロード、HTTP サーバーの起動、および推論リクエストの処理に使用するコードを準備する必要があります。この例では、コードは Flask を使用して記述されています。サンプルコード:

   import json
   from flask import Flask, request
   from PIL import Image
   import os
   import torch
   import torchvision.transforms as transforms
   import numpy as np
   import io
   
   app = Flask(__name__)
   # モデルはデフォルトで現在のパスにロードされます。
   MODEL_PATH = "/eas/workspace/model/"
   
   device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
   model = torch.jit.load(os.path.join(MODEL_PATH, "mnist_cnn.pt"), map_location=device).to(device)
   transform = transforms.Compose(
       [transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))]
   )
   
   
   @app.route("/", methods=["POST"])
   def predict():
       # 画像データを前処理します。
       im = Image.open(io.BytesIO(request.data))
       input_tensor = transform(im).to(device)
       input_tensor.unsqueeze_(0)
       # モデルで推論を実行します。
       output_tensor = model(input_tensor)
       pred_res =output_tensor.detach().cpu().numpy()[0] 
   
       return json.dumps(pred_res.tolist())
   
   
   if __name__ == '__main__':
       app.run(host="0.0.0.0", port=int(os.environ.get("LISTENING_PORT", 8000)))
   

   後で使用するためにアップロードできるように、上記のコードをオンプレミスのマシンに保存します。この例では、ディレクトリ infer_src を作成し、上記のコードを infer_src/run.py に保存します。サンプルディレクトリ構造:

   |-- infer_src # アップロードする推論サービスのコードディレクトリ。
       |-- requirements.txt # オプション。推論サービスのサードパーティ依存関係。
       '-- run.py # 推論サービススクリプト。

2. オンプレミススクリプトと pai.model.container_serving_spec を介して PAI によって提供される PyTorch イメージに基づいて、InferenceSpec オブジェクトを作成します。

   from pai.model import InferenceSpec, container_serving_spec
   from pai.image import retrieve, ImageScope
   
   torch_image_uri = retrieve("PyTorch", framework_version="latest", image_scope=ImageScope.INFERENCE).image_uri
   
   inf_spec = container_serving_spec(
       command="python run.py",
       source_dir="./infer_src/",
       image_uri=torch_image_uri,
       requirements=["flask==2.0.0", "Werkzeug==2.2.2", "pillow", "torchvision"],
   )
   print(inf_spec.to_dict())
   

   • モデルサービスのコードと起動コマンド

     source_dir パラメーターで指定されたオンプレミススクリプトディレクトリは OSS にアップロードされ、サービスコンテナーにマウントされます。 /ml/usercode ディレクトリがデフォルトで使用されます。

   • 推論サービスに使用されるイメージ

     pai.image.retrieve メソッドを使用して、PAI によって提供されるイメージを取得できます。イメージを取得するときは、image_scope パラメーターを ImageScope.INFERENCE に指定します。

   • モデルサービスのサードパーティ依存関係

     requirements パラメーターを使用して、モデルサービスのコードまたは依存関係を指定できます。このようにして、サービスが開始される前に、依存関係が環境にインストールされます。

3. Model.deploy API を呼び出して、トレーニング済みモデルと InferenceSpec を使用してオンライン推論サービスをデプロイします。

   from pai.model import Model
   from pai.common.utils import random_str
   import numpy as np
   
   
   m = Model(
       model_data=est.model_data(),
       inference_spec=inf_spec,
   )
   
   predictor = m.deploy(
       service_name="torch_mnist_script_container_{}".format(random_str(6)),
       instance_type="ecs.c6.xlarge",
   )
   

4. 推論サービスを呼び出します。

   1. MNIST イメージを準備します。

      import base64
      from PIL import Image
      from IPython import display
      import io
      
      !pip install -q pillow
      # raw_data は MNIST イメージで、数字の 9 に対応します。
      raw_data = base64.b64decode(b"/9j/4AAQSkZJRgABAQAAAQABAAD/2wBDAAgGBgcGBQgHBwcJCQgKDBQNDAsLDBkSEw8UHRofHh0aHBwgJC4nICIsIxwcKDcpLDAxNDQ0Hyc5PTgyPC4zNDL/wAALCAAcABwBAREA/8QAHwAAAQUBAQEBAQEAAAAAAAAAAAECAwQFBgcICQoL/8QAtRAAAgEDAwIEAwUFBAQAAAF9AQIDAAQRBRIhMUEGE1FhByJxFDKBkaEII0KxwRVS0fAkM2JyggkKFhcYGRolJicoKSo0NTY3ODk6Q0RFRkdISUpTVFVWV1hZWmNkZWZnaGlqc3R1dnd4eXqDhIWGh4iJipKTlJWWl5iZmqKjpKWmp6ipqrKztLW2t7i5usLDxMXGx8jJytLT1NXW19jZ2uHi4+Tl5ufo6erx8vP09fb3+Pn6/9oACAEBAAA/APn+rVhpmoarP5GnWNzeTYz5dvE0jfkoJovNMv8ATmK3tjc2zByhE8TIQw6jkdR6VVq9oumPrWuWGlxyLG95cRwK7dFLMFyfzr3aXwp4ltAfB3gWwudI01JNuoa7eZhku5AMHafvFOw2Dn6ZJ4z4yeLk1HUbXwrZSSy2Oh5heeaQu88wG1mLHk4wR9c+1eXUqsVYMpIIOQR2r1D4QazqOs/FnSG1fVLi9ZI5vL+2TNKc+U2ApYnB7/hXml5LLNfXEsxLSvIzOSMEsTk1DRVnT7+60vULe/spmhureQSRSL1Vh0NWNd1mXX9ZuNUuLe2gmuCGkS2QohbABbBJwTjJ9yelZ1f/2Q==")
      
      im = Image.open(io.BytesIO(raw_data))
      display.display(im)
      

   2. 推論サービスにリクエストを送信します。

      推論サービスは、HTTP リクエスト本文のデータを入力イメージとして使用します。 raw_predict メソッドは、データが bytes タイプのリクエストを受け入れます。次に、PAI Python SDK は POST メソッドを使用して推論データをリクエスト本文に含め、推論サービスにデータを送信します。

      from pai.predictor import RawResponse
      import numpy as np
      
      resp: RawResponse = predictor.raw_predict(data=raw_data)
      print(resp.json())
      
      print(np.argmax(resp.json()))
      

5. テストが完了したら、サービスを削除します。

   predictor.delete_service() # サービスを削除します