文档

部署推理服务

更新时间:

PAI Python SDK提供了易用的API(即HighLevel API),支持您将模型部署至PAI以创建推理服务。本文介绍使用SDK在PAI部署推理服务时的相关代码配置。

概要介绍

SDK提供了HighLevel API,即pai.model.Modelpai.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 SavedModelPyTorch TorchScriptXGBoostLightGBMPMML等,完整的介绍请参见预置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提供了多种常用框架的推理镜像,包括TensorFlowPyTorchXGBoost等,支持您快速创建推理服务。您可以通过pai.image.list_imagespai.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对象提供了predictraw_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.ndarrayListJsonSerializer.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,实现serializedeserialize方法。

    以下示例是一个自定义的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()

相关文档

  • 使用PAI Python SDK训练和部署PyTorch模型的完整操作流程,请参见使用PAI Python SDK训练和部署PyTorch模型

  • PAI Python SDK还提供了一系列详细的代码示例Notebook,供您参考和学习,以便更直观地体验部署推理服务的操作流程,详情请参见代码示例