安裝和配置SDK

需要首先安裝PAI Python SDK以運(yùn)行本示例。

python -m pip install "alipai>=0.4.0"

在PAI SDK安裝之后，通過在命令行終端中執(zhí)行以下命令進(jìn)行配置，詳細(xì)的安裝和配置介紹見文檔：安裝和配置。

python -m pai.toolkit.config

準(zhǔn)備訓(xùn)練數(shù)據(jù)

當(dāng)前示例中，將使用MNIST數(shù)據(jù)集訓(xùn)練一個圖片分類模型。當(dāng)用戶使用云上的訓(xùn)練作業(yè)時，需要準(zhǔn)備數(shù)據(jù)，上傳到OSS Bucket上。

• 下載MNIST數(shù)據(jù)集

使用以下的Shell腳本，將MNIST數(shù)據(jù)集下載到本地目錄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/

• 上傳數(shù)據(jù)集到OSS

用戶可以使用OSS提供的命令行工具ossutil上傳相應(yīng)的文件（ossutil的安裝和使用請見文檔：ossutil概述），或是PAI Python SDK里提供的便利方法，將本地訓(xùn)練數(shù)據(jù)上傳到OSS Bucket的/mnist/data/路徑下。

• 通過ossutil上傳：

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

• 使用PAI Python SDK上傳文件：

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)

準(zhǔn)備訓(xùn)練腳本

在提交訓(xùn)練作業(yè)之前，需要通過PyTorch編寫訓(xùn)練腳本。這里我們以PyTorch官方提供的MNIST示例為基礎(chǔ)，在修改了數(shù)據(jù)加載和模型保存的邏輯之后，作為訓(xùn)練腳本。

• 使用環(huán)境變量獲得輸入數(shù)據(jù)路徑

當(dāng)我們通過estimator.fit(inputs={"train_data":data_uri})傳遞以上的OSS數(shù)據(jù)URI，相應(yīng)的數(shù)據(jù)會被掛載到訓(xùn)練容器中，訓(xùn)練腳本可以通過讀取本地文件的方式，讀取到掛載的數(shù)據(jù)。

對于訓(xùn)練作業(yè)，estimator.fit方法的inputs是字典，對應(yīng)的每一個輸入數(shù)據(jù)都是一個Channel，Key是Channel名，Value是數(shù)據(jù)存儲路徑，訓(xùn)練作業(yè)腳本可以通過PAI_INPUT_{ChannelNameUpperCase}環(huán)境變量獲取到輸入數(shù)據(jù)掛載到工作容器內(nèi)的數(shù)據(jù)路徑。

數(shù)據(jù)加載部分的代碼修改如下：

- dataset1 = datasets.MNIST("../data", train=True, download=True, transform=transform)
- dataset2 = datasets.MNIST("../data", train=False, transform=transform)

+ # 通過環(huán)境變量獲得輸入數(shù)據(jù)路徑
+ 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)

• 使用環(huán)境變量獲取模型的保存路徑：

用戶需要保存模型到訓(xùn)練環(huán)境中的指定路徑，對應(yīng)路徑下的數(shù)據(jù)和模型會被保存到用戶的OSS Bucket。默認(rèn)要求用戶將模型保存到環(huán)境變量PAI_OUTPUT_MODEL指定的路徑下（默認(rèn)為/ml/output/model）。

模型保存部分的修改代碼如下：

- if args.save_model:
-     torch.save(model.state_dict(), "mnist_cnn.pt")
+ # 保存模型
+ save_model(model)
+ 
+ def save_model(model):
+     """將模型轉(zhuǎn)為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提供的預(yù)置PyTorch Processor在創(chuàng)建服務(wù)時，要求輸入的模型是TorchScript格式。在當(dāng)前示例中，我們將模型導(dǎo)出為TorchScript格式。

完整的作業(yè)腳本如下：

# 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)",
    )
    parser.add_argument(
        "--test-batch-size",
        type=int,
        default=1000,
        metavar="N",
        help="input batch size for testing (default: 1000)",
    )
    parser.add_argument(
        "--epochs",
        type=int,
        default=14,
        metavar="N",
        help="number of epochs to train (default: 14)",
    )
    parser.add_argument(
        "--lr",
        type=float,
        default=1.0,
        metavar="LR",
        help="learning rate (default: 1.0)",
    )
    parser.add_argument(
        "--gamma",
        type=float,
        default=0.7,
        metavar="M",
        help="Learning rate step gamma (default: 0.7)",
    )
    parser.add_argument(
        "--no-cuda", action="store_true", default=False, help="disables CUDA training"
    )
    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)"
    )
    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_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_model(model)


def save_model(model):
    """將模型轉(zhuǎn)為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()

我們需要將以上的訓(xùn)練代碼保存到一個本地目錄下，后續(xù)使用Estimator提交到PAI上執(zhí)行。當(dāng)前示例中，我們將新建一個train_src目錄，將訓(xùn)練腳本保存到 train_src/train.py。

|-- train_src                       # 待上傳的訓(xùn)練腳本目錄
    |-- requirements.txt            # 可選：訓(xùn)練作業(yè)的第三方包依賴
    `-- train.py                    # 保存的訓(xùn)練作業(yè)腳本

提交訓(xùn)練作業(yè)

Estimator支持用戶使用本地的訓(xùn)練腳本，以指定的鏡像在云上執(zhí)行訓(xùn)練作業(yè)。

• 訓(xùn)練作業(yè)腳本和命令

用戶訓(xùn)練作業(yè)腳本所在目錄（參數(shù)source_dir）會被上傳到OSS，在作業(yè)啟動之前準(zhǔn)備到作業(yè)容器中，默認(rèn)為/ml/usercode目錄。用戶指定的啟動命令（command參數(shù)）的工作目錄同樣是/ml/usercode。

• 訓(xùn)練作業(yè)鏡像

當(dāng)前示例中，我們使用PAI提供的PyTorch鏡像運(yùn)行訓(xùn)練作業(yè)。

• 訓(xùn)練作業(yè)超參

用戶可以通過讀取${PAI_CONFIG_DIR}/hyperparameters.json文件獲取到訓(xùn)練作業(yè)的超參 ，也可以通過環(huán)境變量獲取到訓(xùn)練作業(yè)超參，詳細(xì)可見文檔：訓(xùn)練作業(yè)預(yù)置環(huán)境變量。

在當(dāng)前示例中，執(zhí)行的命令是python train.py $PAI_USER_ARGS，其中PAI_USER_ARGS環(huán)境變量是作業(yè)超參以命令行參數(shù)的方式拼接獲得的字符串。訓(xùn)練作業(yè)最終的啟動命令是python train.py --epochs 5 --batch-size 256 --lr 0.5。

• 通過metric_definitions指定需要采集的Metrics

PAI的訓(xùn)練服務(wù)支持從訓(xùn)練作業(yè)輸出日志中（訓(xùn)練腳本打印的標(biāo)準(zhǔn)輸出和標(biāo)準(zhǔn)錯誤輸出），以正則表達(dá)式匹配的方式捕獲訓(xùn)練作業(yè)Metrics信息。通過SDK打印的作業(yè)的詳情頁鏈接，用戶查看作業(yè)的詳情配置、輸出日志以及訓(xùn)練作業(yè)的Metrics。

• 通過instance_type指定作業(yè)使用的機(jī)器實例類型：

PAI的訓(xùn)練作業(yè)支持的機(jī)器實例類型，請見文檔：附錄：公共資源組定價詳情。

構(gòu)建Estimator的示例代碼：

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

# 使用PAI提供的1.18PAI版本的PyTorch GPU鏡像運(yùn)行訓(xùn)練腳本
image_uri = retrieve(
    "PyTorch", framework_version="1.8PAI", accelerator_type="GPU"
).image_uri
print(image_uri)

est = Estimator(
    # 訓(xùn)練作業(yè)啟動命令，默認(rèn)工作目錄為/ml/usercode/
    command="python train.py $PAI_USER_ARGS",
    # 需要上傳的訓(xùn)練代碼目錄的相對路徑或是絕對路徑
  	# 默認(rèn)會準(zhǔn)備到訓(xùn)練作業(yè)環(huán)境的/ml/usercode 目錄下
    source_dir="./train_src/",
    # 訓(xùn)練作業(yè)鏡像
    image_uri=image_uri,
    # 機(jī)器配置
    instance_type="ecs.gn6i-c4g1.xlarge",  # 4vCPU 15GB 1*NVIDIA T4
    # 訓(xùn)練作業(yè)超參
    hyperparameters={
        "epochs": 5,
        "batch-size": 64 * 4,
        "lr": 0.5,
    },
    # 訓(xùn)練作業(yè)的Metric捕獲配置
    metric_definitions=[
        {
            "Name": "loss",
            "Regex": r".*loss=([-+]?[0-9]*.?[0-9]+(?:[eE][-+]?[0-9]+)?).*",
        },
    ],
    base_job_name="pytorch_mnist",
)

est.fit方法將用戶的訓(xùn)練作業(yè)提交到PAI上執(zhí)行。任務(wù)提交之后，SDK會打印作業(yè)詳情頁鏈接，并持續(xù)打印訓(xùn)練作業(yè)的日志，直到作業(yè)執(zhí)行結(jié)束。

當(dāng)用戶需要直接使用OSS上數(shù)據(jù)，可以通過estimator.fit方法的inputs參數(shù)傳遞。通過inputs傳遞數(shù)據(jù)存儲路徑會被掛載到目錄下，用戶的訓(xùn)練腳本可以通過讀取本地文件的方式加載數(shù)據(jù)。

本示例中，我們將上傳到OSS的訓(xùn)練數(shù)據(jù)作為訓(xùn)練輸入數(shù)據(jù)。

# 如果使用ossutil上傳訓(xùn)練數(shù)據(jù)，我們需要顯式賦值輸入數(shù)據(jù)的OSS URI路徑
# data_uri = "oss://<YourOssBucket>/mnist/data/"

# 提交訓(xùn)練作業(yè)
est.fit(
    inputs={
        "train_data": data_uri,
    }
)

# 訓(xùn)練作業(yè)產(chǎn)出的模型路徑
print("TrainingJob output model data:")
print(est.model_data())

對于提交訓(xùn)練作業(yè)的詳細(xì)介紹，請查看PAI Python SDK提交訓(xùn)練作業(yè)。

部署推理服務(wù)

在訓(xùn)練作業(yè)結(jié)束之后，我們可以使用estimator.model_data()方法拿到訓(xùn)練作業(yè)產(chǎn)出模型的OSS路徑。下面的流程中，我們將訓(xùn)練產(chǎn)出的模型部署到PAI創(chuàng)建在線推理服務(wù)。

部署推理服務(wù)的主要流程包括：

• 通過InferenceSpec描述如何使用模型構(gòu)建推理服務(wù)。

用戶可以選擇使用Processor或是自定義鏡像的模式進(jìn)行模型部署。以下示例中將分別使用兩種方式部署獲得的PyTorch模型。

• 通過Model.deploy方法，配置服務(wù)的使用資源、服務(wù)名稱等信息，創(chuàng)建推理服務(wù)。

對于部署推理服務(wù)的詳細(xì)介紹，請參見：部署推理服務(wù)。

Processor 模式部署

Processor是PAI對于推理服務(wù)程序包的抽象描述，負(fù)責(zé)加載模型并啟動模型推理服務(wù)。模型推理服務(wù)會暴露API支持用戶進(jìn)行調(diào)用。PAI提供了預(yù)置PyTorch Processor，支持用戶方便地將TorchScript格式的模型部署到PAI，創(chuàng)建推理服務(wù)。對于PyTorch Processor的詳細(xì)介紹，請參見：PyTorch Processor。

以下示例中，我們通過PyTorch Processor將訓(xùn)練產(chǎn)出的模型部署為一個推理服務(wù)。

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 Processor
 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對象指向創(chuàng)建的推理服務(wù)，可以通過Predictor.predict方法發(fā)送預(yù)測請求給到服務(wù)，拿到預(yù)測結(jié)果。

我們使用NumPy構(gòu)建了一個測試樣本數(shù)據(jù)，發(fā)送給推理服務(wù)。

import numpy as np

# 以上保存TorchScritp模型要求輸入為 Float32, 數(shù)據(jù)格式的形狀為 (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))

在測試完成之后，可以通過Predictor.delete_service刪除推理服務(wù)。

p.delete_service()

鏡像部署

Processor模式啟動的推理服務(wù)性能優(yōu)越，適合于對于性能較為敏感的場景。對于一些需要靈活自定義的場景，例如模型使用了一些第三方的依賴，或是推理服務(wù)需要有前處理和后處理，用戶可以通過鏡像部署的方式實現(xiàn)。 SDK提供了pai.model.container_serving_spec()方法，支持用戶使用本地的推理服務(wù)代碼配合PAI提供的基礎(chǔ)鏡像的方式創(chuàng)建推理服務(wù)。

在使用鏡像部署之前，我們需要準(zhǔn)備模型服務(wù)的代碼，負(fù)責(zé)加載模型、拉起HTTP Server、處理用戶的推理請求。我們將使用Flask編寫一個模型服務(wù)的代碼，示例如下：

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__)
# 用戶指定模型，默認(rèn)會被加載到當(dāng)前路徑下。 
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():
    # 預(yù)處理圖片數(shù)據(jù)
    im = Image.open(io.BytesIO(request.data))
    input_tensor = transform(im).to(device)
    input_tensor.unsqueeze_(0)
    # 使用模型進(jìn)行推理
    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)))

我們需要將以上的代碼保存到本地，供后續(xù)上傳。在本示例中，我們需要在本地新建目錄infer_src，將以上的推理服務(wù)代碼保存到infer_src/run.py，目錄結(jié)構(gòu)如下：

|-- infer_src                  # 待上傳的推理服務(wù)代碼目錄
    |-- requirements.txt       # 可選：推理服務(wù)的第三方包依賴
    `-- run.py                 # 保存的推理服務(wù)腳本

通過pai.model.container_serving_spec，我們基于本地腳本和PAI提供的PyTorch鏡像創(chuàng)建了一個InferenceSpec對象。

• 模型服務(wù)的代碼和啟動命令：

用戶指定的本地腳本目錄source_dir參數(shù)會被上傳到OSS，然后掛載到服務(wù)容器（默認(rèn)到 /ml/usercode目錄）。

• 推理服務(wù)鏡像：

PAI 提供了基礎(chǔ)的推理鏡像支持用戶使用，用戶可以通過pai.image.retrieve方法，指定參數(shù)image_scope=ImageScope.INFERENCE獲取PAI提供的推理鏡像。

• 模型服務(wù)的第三方依賴包：

模型服務(wù)代碼或是模型的依賴，可以通過requirements參數(shù)指定，相應(yīng)的依賴會在服務(wù)程序啟動前被安裝到環(huán)境中。

使用訓(xùn)練作業(yè)輸出的模型和上述的InferenceSpec，我們將通過Model.deployAPI部署一個在線推理服務(wù)。

from pai.model import InferenceSpec, container_serving_spec, Model
from pai.image import retrieve, ImageScope
from pai.common.utils import random_str
import numpy as np

torch_image_uri = retrieve(
    framework_name="pytorch", framework_version="1.12", accelerator_type="CPU"
).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"],
)
print(inf_spec.to_dict())

m = Model(
    model_data=est.model_data(),
    inference_spec=inf_spec,
)

predictor = m.deploy(
    service_name="torch_container_{}".format(random_str(6)),
    instance_type="ecs.c6.xlarge",
)

我們準(zhǔn)備一張MNIST測試圖片，用于發(fā)送給推理服務(wù)。

import base64
from PIL import Image
from IPython import display
import io


# raw_data是一張MNIST圖片，對應(yīng)數(shù)字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)

推理服務(wù)使用HTTP請求體內(nèi)的數(shù)據(jù)作為輸入的圖片，SDK的raw_predict方法接受bytes數(shù)據(jù)類型的請求，通過POST方法，在請求體（HTTP Request Body）帶上用戶推理數(shù)據(jù)，發(fā)送給到推理服務(wù)。

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()))

測試完成之后可以刪除服務(wù)。

predictor.delete_service()