使用PAI Python SDK部署推理服务_人工智能平台 PAI(PAI)-阿里云帮助中心

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)

以下内容为您介绍了部署推理服务的相关代码配置。此外，为了帮助您更直观地体验部署推理服务的操作流程，PAI Python SDK还提供了一系列详细的代码示例Notebook，供您参考和学习，详情请参见代码示例。

配置模型的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被调用时，整体的链路如下：
1. 客户端：用户传递numpy.ndarray, 或是pandas.DataFrame，作为predict的输入，调用 NumpySerializer.serializer序列化为npy format，发送给到服务端。
2. 服务端：推理服务接收npy格式的数据，反序列化数据，获得推理结果，然后将输出的结果，序列化为npy格式返回。
3. 客户端：接收到返回的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()