PAI Python SDK提供了易用的API（即HighLevel API），支援您將模型部署至PAI以建立推理服務。本文介紹使用SDK在PAI部署推理服務時的相關代碼配置。
概要介紹
SDK提供了HighLevel API，即pai.model.Model和pai.predictor.Predictor，支援您將模型部署到EAS，並進行調用測試。
通過SDK建立推理服務的基本流程包括：
通過
pai.model.InferenceSpec對象定義模型的推理服務配置，包括使用的Processor或鏡像的資訊。使用
InferenceSpec對象和待部署的模型檔案建立一個pai.model.Model對象。通過
pai.model.Model.deploy()方法，指定服務使用的資源、服務名稱等資訊，在PAI建立一個推理服務。通過
deploy方法返回pai.predictor.Predictor對象，提供了predict方法向推理服務發送推理請求。
範例程式碼如下：
from pai.model import InferenceSpec, Model, container_serving_spec
from pai.image import retrieve, ImageScope
# 1、使用PAI提供的PyTorch推理鏡像。
torch_image = retrieve("PyTorch", framework_version="latest",
image_scope=ImageScope.INFERENCE)
# 2、使用InferenceSpec描述模型的推理配置資訊。
inference_spec = container_serving_spec(
# 推理服務的啟動命令。
command="python app.py",
source_dir="./src/"
# 使用的推理鏡像。
image_uri=torch_image.image_uri,
)
# 3、構建Model對象，用於模型部署。
model = Model(
# 使用OSS Bucket上的模型檔案。
model_data="oss://<YourBucket>/path-to-model-data",
inference_spec=inference_spec,
)
# 4、部署模型到PAI-EAS，建立線上推理服務，返回Predictor對象。
predictor = model.deploy(
service_name="example_torch_service",
instance_type="ecs.c6.xlarge",
)
# 5、測試推理服務。
res = predictor.predict(data=data)以下內容為您介紹了部署推理服務的相關代碼配置。
配置模型的InferenceSpec
您可以通過Processor或鏡像的方式部署推理服務，pai.model.InferenceSpec對象用於定義模型的推理服務配置，例如使用Processor或是鏡像部署、模型服務的儲存配置、模型服務的預熱配置、模型服務的RPC Batch功能配置等，所構建的InferenceSpec對象將用於推理服務的建立。
使用預置Processor部署服務
Processor是PAI對於推理服務程式包的抽象描述，它能夠基於使用者提供的模型，直接構建一個推理服務。PAI提供了預置的Processor，支援一系列常見的機器學習模型格式，包括TensorFlow SavedModel、PyTorch TorchScript、XGBoost、LightGBM和PMML等，完整的介紹請參見預置Processor使用說明。
對於使用Processor方式部署模型，您可以參考以下樣本配置
InferenceSpec。# 使用預置的TensorFlow Processor。 tf_infer_spec = InferenceSpec(processor="tensorflow_cpu_2.3") # 使用預置的PyTorch Processor。 tf_infer_spec = InferenceSpec(processor="pytorch_cpu_1.10") # 使用預置的XGBoost Processor。 xgb_infer_spec = InferenceSpec(processor="xgboost")您可以在
InferenceSpec執行個體上配置推理服務的更多功能，例如佈建服務預熱檔案、服務的RPC配置等，完整的服務參數資訊請參見服務模型所有相關參數說明。# 直接配置InferenceSpec的屬性。 tf_infer_spec.warm_up_data_path = "oss://<YourOssBucket>/path/to/warmup-data" # 佈建服務預熱檔案路徑。 tf_infer_spec.metadata.rpc.keepalive = 1000 # 配置請求連結的keepalive時間長度。 print(tf_infer_spec.warm_up_data_path) print(tf_infer_spec.metadata.rpc.keepalive)
使用鏡像部署服務
雖然使用Processor部署模型服務提高了易用性，但它不支援使用者進行靈活的自訂配置，尤其是在模型或推理服務程式存在複雜依賴關係時。對於類似的情境，PAI提供了鏡像部署的方式，支援使用者以更靈活的方式自訂部署模型。
將模型服務的代碼和相關依賴打包構建成一個Docker鏡像，並推送到阿里雲ACR鏡像倉庫，然後基於該Docker鏡像構建InferenceSpec，用於模型的部署。
from pai.model import InferenceSpec, container_serving_spec # 通過container_serving_spec方法，使用者可以構建一個使用鏡像服務模型的InferenceSpec。 container_infer_spec = container_serving_spec( # 推理服務運行使用的鏡像。 image_uri="<CustomImageUri>", # 運行在容器內的推理服務需要監聽的連接埠, 使用者發送的預測請求會被PAI轉寄到服務容器的該連接埠。 port=8000, environment_variables=environment_variables, # 推理服務的啟動命令。 command=command, # 推理服務依賴的Python包。 requirements=[ "scikit-learn", "fastapi==0.87.0", ], ) print(container_infer_spec.to_dict()) m = Model( model_data="oss://<YourOssBucket>/path-to-tensorflow-saved-model", inference_spec=custom_container_infer_spec, ) p = m.deploy( instance_type="ecs.c6.xlarge" )採用自訂鏡像部署服務時，您需要準備所需的推理服務代碼並將其整合到運行容器中、構建並推送鏡像至倉庫。PAI SDK提供了便捷方法，支援您使用本地代碼和基礎鏡像來構建推理服務，而無需手動構建鏡像。
pai.model.container_serving_spec()支援通過source_dir參數來指定一個本地代碼檔案目錄，SDK將自動打包並上傳該檔案夾到OSS Bucket，並將其路徑掛載到運行容器中，您可以使用指定的啟動命令啟動推理服務。from pai.model import InferenceSpec inference_spec = container_serving_spec( # 使用者推理程式所在的本地目錄路徑，會被上傳到OSS Bucket，然後掛載到運行容器，預設為/ml/usercode/。 source_dir="./src", # 服務啟動命令。當使用者指定了source_dir，則預設使用/ml/usercode作為工作目錄執行command。 command="python run.py", image_uri="<ServingImageUri>", requirements=[ "fastapi", "uvicorn", ] ) print(inference_spec.to_dict())當您需要將額外的資料、代碼或模型匯入推理服務容器時，可以利用
pai.model.InferenceSpec.mount()方法，將本地目錄或OSS資料路徑掛載到線上服務容器內。# 將本地的資料上傳到OSS，然後掛載到容器的/ml/tokenizers目錄下。 inference_spec.mount("./bert_tokenizers/", "/ml/tokenizers/") # 直接掛載使用者儲存在 OSS 上的資料到容器的/ml/data目錄下。 inference_spec.mount("oss://<YourOssBucket>/path/to/data/", "/ml/data/")擷取PAI提供的公用鏡像
PAI提供了多種常用架構的推理鏡像，包括
TensorFlow、PyTorch和XGBoost等，支援您快速建立推理服務。您可以通過pai.image.list_images、pai.image.retrieve方法傳遞image_scope=ImageScope.INFERENCE資訊，從而擷取到相應的推理鏡像，然後使用鏡像部署的方式部署模型。from pai.image import retrieve, ImageScope, list_images # 擷取PAI提供的所有PyTorch推理鏡像。 for image_info in list_images(framework_name="PyTorch", image_scope=ImageScope.INFERENCE): print(image_info) # 擷取PAI提供的PyTorch 1.12版本的CPU推理鏡像。 retrieve(framework_name="PyTorch", framework_version="1.12", image_scope=ImageScope.INFERENCE) # 擷取PAI提供的PyTorch 1.12版本的GPU推理鏡像。 retrieve(framework_name="PyTorch", framework_version="1.12", accelerator_type="GPU", image_scope=ImageScope.INFERENCE) # 擷取PAI提供的PyTorch最新版本的GPU推理鏡像。 retrieve(framework_name="PyTorch", framework_version="latest", accelerator_type="GPU", image_scope=ImageScope.INFERENCE)
部署和調用線上推理服務
部署推理服務
使用pai.model.InferenceSpec和模型資料地址model_data構建一個模型對象pai.model.Model，然後通過調用.deploy方法部署模型。model_data可以是一個OSS URI，也可以是本地路徑，對於本地路徑的模型，相應的模型檔案會被上傳到OSS Bucket上，然後準備到推理服務中，供對應的服務程式載入使用。
當調用.deploy方法部署模型時，您需要指定服務所需的資源配置、服務執行個體個數、服務名稱等服務相關參數。更多高階參數說明，請參見服務模型所有相關參數說明。
from pai.model import Model, InferenceSpec
from pai.predictor import Predictor
model = Model(
# model_data模型所在的路徑，可以是OSS URI，或是本地路徑。對於本地路徑的模型，預設會被上傳到OSS Bucket上。
model_data="oss://<YourBucket>/path-to-model-data",
inference_spec=inference_spec,
)
# 部署到EAS。
predictor = m.deploy(
# 推理服務的名稱。
service_name="example_xgb_service",
# 服務使用的機器類型。
instance_type="ecs.c6.xlarge",
# 機器執行個體/服務的個數。
instance_count=2,
# 使用者的專有資源群組，可選。預設使用公用資源群組。
# resource_id="<YOUR_EAS_RESOURCE_GROUP_ID>",
options={
"metadata.rpc.batching": True,
"metadata.rpc.keepalive": 50000,
"metadata.rpc.max_batch_size": 16,
"warm_up_data_path": "oss://<YourOssBucketName>/path-to-warmup-data",
},
)當您需要根據服務使用的資源數量（例如CPU、Memory）佈建服務時，可以通過resource_config參數配置每個服務執行個體申請的資源，樣本如下：
from pai.model import ResourceConfig
predictor = m.deploy(
service_name="dedicated_rg_service",
# 指定單個服務執行個體使用的CPU和Memory資源。
# 當前樣本中，每一個服務使用2個核的CPU，以及4000 MB的記憶體。
resource_config=ResourceConfig(
cpu=2,
memory=4000,
),
)調用推理服務
pai.model.Model.deploy方法通過調用EAS的API建立一個新的推理服務，並返回一個pai.predictor.Predictor對象，指向新建立的推理服務。Predictor對象提供了predict和raw_predict方法，支援向推理服務發送預測請求。
pai.predictor.Predictor.raw_predict的輸入和輸出不需要使用Serializer進行處理。
from pai.predictor import Predictor, EndpointType
# 建立一個新的推理服務。
predictor = model.deploy(
instance_type="ecs.c6.xlarge",
service_name="example_xgb_service",
)
# 使用已有的推理服務。
predictor = Predictor(
service_name="example_xgb_service",
# 預設使用INTERNET公網網路訪問，使用者可以配置使用VPC的網路（需要用戶端代碼運行在VPC環境下）。
# endpoint_type=EndpointType.INTRANET
)
# .predict向對應服務發送資料請求，擷取響應結果。輸入資料和響應結果會經過serializer處理。
res = predictor.predict(data_in_nested_list)
# .raw_predict支援更為靈活的方式發送請求給到推理服務。
response: RawResponse = predictor.raw_predict(
# 如果輸入資料是bytes，或是file-like object，請求資料直接在HTTP請求體內傳遞。
# 否則，則會經過一次JSON序列化，然後放在HTTP請求體內傳遞。
data=data_in_nested_list
# path="predict" # 自訂HTTP請求路徑，預設將請求發送到"/"路徑。
# headers=dict(), # 自訂請求Header。
# method="POST" # 自訂請求的HTTP Method。
# timeout=30, # 自訂請求的timeout。
)
# 擷取返回的body, headers。
print(response.content, response.headers)
# 將返回結果JSON還原序列化為Python對象。
print(response.json())
# 停止推理服務。
predictor.stop_service()
# 開始推理服務。
predictor.start_service()
# 刪除推理服務。
predictor.delete_service()使用Serializer處理推理服務的輸入和輸出
當通過SDK的pai.predictor.Predictor.predict方法調用推理服務時，需要對輸入的Python資料結構進行序列化，以便將其轉換成服務能夠處理的資料格式進行傳輸。同樣，服務返回的響應資料也需進行還原序列化，以將其轉換成可讀或可操作的Python對象。Predictor利用serializer參數實現了預測資料的序列化及預測響應結果的還原序列化。
當調用
predict(data=<PredictionData>)方法時，data參數會通過serializer.serialize方法序列化請求資料，獲得bytes類型的結果，然後通過HTTP請求體傳遞給推理服務。當推理服務返回HTTP響應結果之後，
Predictor對象通過serializer.deserialize方法還原序列化HTTP響應的結果，作為predict方法的返回。
SDK提供了一些預置的Serializer，支援常見資料的序列化處理，以及PAI內建的深度學習Processor的輸入輸出資料處理。
JsonSerializer
JsonSerializer對象支援JSON格式資料的序列化和還原序列化。通過predict方法傳遞的data，可以是numpy.ndarray或List，JsonSerializer.serialize將對應的數組序列化為JSON字串，而JsonSerializer.deserialize則將接收到的JSON字串還原序列化成Python對象。PAI預置的XGBoost、PMML等Processor接收和返回JSON格式的資料。對於這些Processor建立的服務，Predictor預設採用JsonSerializer來處理輸入和輸出資料。
from pai.serializers import JsonSerializer
# 在“.deploy”方法指定返回的predictor使用的serializer。
p = Model(
inference_spec=InferenceSpec(processor="xgboost"),
model_data="oss://<YourOssBucket>/path-to-xgboost-model"
).deploy(
instance_type="ecs.c6.xlarge",
# 可選：使用XGBoost processor的service預設使用JsonSerializer。
serializer=JsonSerializer()
)
# 或是直接建立Predictor時指定對應的serializer。
p = Predictor(
service_name="example_xgb_service"
serializer=JsonSerializer(),
)
# 預測的返回結果也是一個list。
res = p.predict([[2,3,4], [4,5,6]])TensorFlowSerializer
您可以直接使用PAI預置的Tensorflow Processor，將TensorFlow的SavedModel部署到PAI以建立推理服務。對應服務的輸入輸出訊息格式是Protocol Buffers，具體檔案格式請參見tf_predict.proto。
SDK提供了預置的TensorFlowSerializer，支援使用者使用numpy.ndarray類型的資料發送請求。Serializer負責將numpy.ndarray類型的資料轉換成相應的Protocol Buffers訊息，並將接收的Protocol Buffers訊息還原序列化為numpy.ndarray資料類型。
# 建立一個TensorFlow processor服務。
tf_predictor = Model(
inference_spec=InferenceSpec(processor="tensorflow_cpu_2.7"),
model_data="oss://<YourOssBucket>/path-to-tensorflow-saved-model"
).deploy(
instance_type="ecs.c6.xlarge",
# 可選：使用TensorFlow processor的service預設使用TensorFlowSerializer。
# serializer=TensorFlowSerializer(),
)
# 使用TensorFlow processor啟動的服務，支援使用者通過API擷取模型的服務簽名。
print(tf_predictor.inspect_signature_def())
# TensorFlow processor的輸入要求一個Dict，Key是模型輸入簽名的名稱，Value是具體的輸入資料。
tf_result = tf_predictor.predict(data={
"flatten_input": numpy.zeros(28*28*2).reshape((-1, 28, 28))
})
assert result["dense_1"].shape == (2, 10)PyTorchSerializer
您可以直接使用PAI預置的PyTorchProcessor，將TorchScript 格式的模型部署為推理服務。對應服務的輸入輸出資料格式為Protocol Buffers，具體檔案格式請參見pytorch_predict_proto。
SDK提供了預置的PyTorchSerializer，支援使用者使用numpy.ndarray類型的資料發送請求，並將預測結果轉換為numpy.ndarray，由PyTorchSerializer負責Protocol Buffers訊息和numpy.ndarray的轉換。
# 建立一個使用PyTorch processor服務。
torch_predictor = Model(
inference_spec=InferenceSpec(processor="pytorch_cpu_1.10"),
model_data="oss://<YourOssBucket>/path-to-torch_script-model"
).deploy(
instance_type="ecs.c6.xlarge",
# 可選：使用PyTorch processor的service預設使用PyTorchSerializer。
# serializer=PyTorchSerializer(),
)
# 1. 使用者需要注意將對應的輸入資料 reshape 成模型支援的形狀。
# 2. 如果有多個輸入資料，則需要使用List/Tuple傳遞，列表中的每一項是numpy.ndarray。
torch_result = torch_predictor.predict(data=numpy.zeros(28 * 28 * 2).reshape((-1, 28, 28)))
assert torch_result.shape == (2, 10)自訂Serializer
您可以根據推理服務支援的資料格式自訂Serializer類：自訂Serializer類需繼承pai.serializers.SerializerBase，實現serialize和deserialize方法。
以下樣本是一個自訂的NumpySerializer，當predict被調用時，整體的鏈路如下：
用戶端： 使用者傳遞
numpy.ndarray, 或是pandas.DataFrame，作為predict的輸入，調用NumpySerializer.serializer序列化為npy format，發送給到服務端。服務端： 推理服務接收
npy格式的資料，還原序列化資料，獲得推理結果，然後將輸出的結果，序列化為npy格式返回。用戶端： 接收到返回的
npy格式的資料，通過NumpySerializer.deserialize還原序列化為numpy.ndarray。
本地部署和調用推理服務
對於自訂鏡像部署，SDK提供了本地執行模式（不適用於Processor部署的服務）。通過在model.deploy中傳遞instance_type="local"參數，指定在本地運行推理服務。SDK通過docker在本地啟動一個模型服務，自動從OSS下載所需的模型資料，並將其掛載到本地容器中。
from pai.predictor import LocalPredictor
p: LocalPredictor = model.deploy(
# 指定運行在本地。
instance_type="local",
serializer=JsonSerializer()
)
p.predict(data)
# 刪除對應的docker容器。
p.delete_service()相關文檔
使用PAI Python SDK訓練和部署PyTorch模型的完整操作流程，請參見使用PAI Python SDK訓練和部署PyTorch模型。