Platform For AI:Deploy a Triton Inference Server image

oss://your-bucket/models/triton/
└── your_model_name/
    ├── 1/                    # Version directory (must be a number)
    │   └── model.pt          # Model file
    └── config.pbtxt          # Model configuration file

name: "your_model_name"
platform: "pytorch_libtorch"
max_batch_size: 128

input [
  {
    name: "INPUT__0"
    data_type: TYPE_FP32
    dims: [ 3, -1, -1 ]
  }
]

output [
  {
    name: "OUTPUT__0"
    data_type: TYPE_FP32
    dims: [ 1000 ]
  }
]

# Use a GPU for inference
# instance_group [
#   { 
#     kind: KIND_GPU
#   }
# ]

# Model version configuration
# Load only the latest version (default behavior)
# version_policy: { latest: { num_versions: 1 }}

# Load all versions
# version_policy: { all { }}

# Load the two latest versions
# version_policy: { latest: { num_versions: 2 }}

# Load specific versions
# version_policy: { specific: { versions: [1, 3] }}

Parámetros

your_model_name/
├── 1/
│   ├── model.pt          # Model file
│   └── model.py          # Custom inference logic
└── config.pbtxt

import json
import os
import torch
from torch.utils.dlpack import from_dlpack, to_dlpack

import triton_python_backend_utils as pb_utils

class TritonPythonModel:
    """The class name must be 'TritonPythonModel'."""

    def initialize(self, args):
        """
        Triton calls this optional function once when it loads a model.
        Use it to initialize information related to model properties and configurations.
        Parameters
        ----------
        args : A dictionary where both keys and values are strings. It includes:
          * model_config: Model configuration information in JSON format.
          * model_instance_kind: The device type.
          * model_instance_device_id: The device ID.
          * model_repository: The path to the model repository.
          * model_version: The model version.
          * model_name: The model name.
        """

        # Parse the model configuration from the JSON string into a Python dictionary.
        self.model_config = model_config = json.loads(args["model_config"])

        # Get properties from the model configuration file.
        output_config = pb_utils.get_output_config_by_name(model_config, "OUTPUT__0")

        # Convert Triton types to NumPy types.
        self.output_dtype = pb_utils.triton_string_to_numpy(output_config["data_type"])

        # Get the path to the model repository.
        self.model_directory = os.path.dirname(os.path.realpath(__file__))

        # Get the device for model inference. This example uses a GPU.
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        print("device: ", self.device)

        model_path = os.path.join(self.model_directory, "model.pt")
        if not os.path.exists(model_path):
            raise pb_utils.TritonModelException("Cannot find the pytorch model")
        # Load the PyTorch model to the GPU by using .to(self.device).
        self.model = torch.jit.load(model_path).to(self.device)

        print("Initialized...")

    def execute(self, requests):
        """
        This required function executes the model. Triton calls it for every inference request.
        If batching is enabled, you must implement the batch processing logic.
        Parameters
        ----------
        requests : A list of pb_utils.InferenceRequest objects.

        Returns
        -------
        A list of pb_utils.InferenceResponse objects. The list must have the same length as the requests list.
        """

        output_dtype = self.output_dtype

        responses = []

        # Iterate through the requests list and create a corresponding response for each request.
        for request in requests:
            # Get the input tensor.
            input_tensor = pb_utils.get_input_tensor_by_name(request, "INPUT__0")
            # Convert the Triton tensor to a Torch tensor.
            pytorch_tensor = from_dlpack(input_tensor.to_dlpack())

            if pytorch_tensor.shape[2] > 1000 or pytorch_tensor.shape[3] > 1000:
                responses.append(
                    pb_utils.InferenceResponse(
                        output_tensors=[],
                        error=pb_utils.TritonError(
                            "Image shape should not be larger than 1000"
                        ),
                    )
                )
                continue

            # Run inference on the GPU.
            prediction = self.model(pytorch_tensor.to(self.device))

            # Convert the Torch output tensor to a Triton tensor.
            out_tensor = pb_utils.Tensor.from_dlpack("OUTPUT__0", to_dlpack(prediction))

            inference_response = pb_utils.InferenceResponse(output_tensors=[out_tensor])
            responses.append(inference_response)

        return responses

    def finalize(self):
        """
        This optional function is called when the model is unloaded. Use it for cleanup tasks.
        """
        print("Cleaning up...")

name: "resnet50_pt"
backend: "python"
max_batch_size: 128
input [
  {
    name: "INPUT__0"
    data_type: TYPE_FP32
    dims: [ 3, -1, -1 ]
  }
]
output [
  {
    name: "OUTPUT__0"
    data_type: TYPE_FP32
    dims: [ 1000 ]
  }
]

parameters: {
    key: "FORCE_CPU_ONLY_INPUT_TENSORS"
    value: {string_value: "no"}
}

{
  "metadata": {
    "name": "triton_server_test",
    "instance": 1
  },
  "cloud": {
        "computing": {
            "instance_type": "ml.gu7i.c8m30.1-gu30",
            "instances": null
        }
    },
  "containers": [
    {
      "command": "tritonserver --model-repository=/models",
      "image": "eas-registry-vpc.<region>.cr.aliyuncs.com/pai-eas/tritonserver:25.03-py3",
      "port": 8000,
      "prepare": {
        "pythonRequirements": [
          "torch==2.0.1"
        ]
      }
    }
  ],
  "storage": [
    {
      "mount_path": "/models",
      "oss": {
        "path": "oss://oss-test/models/triton_backend/"
      }
    },
    {
      "empty_dir": {
        "medium": "memory",
        // Configure 1 GB of shared memory.
        "size_limit": 1
      },
      "mount_path": "/dev/shm"
    }
  ]
}

import numpy as np
# To install the tritonclient package, run the following command: pip install tritonclient
import tritonclient.http as httpclient

# The public endpoint generated after the service is deployed. Do not include the http:// prefix.
url = '1859257******.cn-hangzhou.pai-eas.aliyuncs.com/api/predict/triton_server_test'

triton_client = httpclient.InferenceServerClient(url=url)

image = np.ones((1,3,224,224))
image = image.astype(np.float32)

inputs = []
inputs.append(httpclient.InferInput('INPUT__0', image.shape, "FP32"))
inputs[0].set_data_from_numpy(image, binary_data=False)
outputs = []
outputs.append(httpclient.InferRequestedOutput('OUTPUT__0', binary_data=False))  # Get a 1000-dimensional vector.

# Specify the model name, request token, inputs, and outputs.
results = triton_client.infer(
    model_name="<your-model-name>",
    model_version="<version-num>",
    inputs=inputs,
    outputs=outputs,
    headers={"Authorization": "<your-service-token>"},
)
output_data0 = results.as_numpy('OUTPUT__0')
print(output_data0.shape)
print(output_data0)

#!/usr/bin/env python
import grpc
# To install the tritonclient package, run the following command: pip install tritonclient
from tritonclient.grpc import service_pb2, service_pb2_grpc
import numpy as np

if __name__ == "__main__":
    # The service endpoint generated after service deployment. Do not include http://. Append :80 to the end.
    host = (
        "service_name.115770327099****.cn-beijing.pai-eas.aliyuncs.com:80"
    )
    # The service token. In a real application, you should use an actual token.
    token = "<your-service-token>"
    # The model name and version.
    model_name = "<your-model-name>"
    model_version = "<version-num>"
    
    # Create gRPC metadata for token authentication.
    metadata = (("authorization", token),)

    # Create a gRPC channel and stub to communicate with the server.
    channel = grpc.insecure_channel(host)
    grpc_stub = service_pb2_grpc.GRPCInferenceServiceStub(channel)
    # Build the inference request.
    request = service_pb2.ModelInferRequest()
    request.model_name = model_name
    request.model_version = model_version

    # Construct the input tensor, which corresponds to the input parameter defined in the model configuration file.
    input = service_pb2.ModelInferRequest().InferInputTensor()
    input.name = "INPUT__0"
    input.datatype = "FP32"
    input.shape.extend([1, 3, 224, 224])
    # Construct the output tensor, which corresponds to the output parameter defined in the model configuration file.
    output = service_pb2.ModelInferRequest().InferRequestedOutputTensor()
    output.name = "OUTPUT__0"

    # Add the input and output specifications to the request.
    request.inputs.extend([input])
    request.outputs.extend([output])
    # Construct a random array and serialize it into a byte sequence as the input data.
    request.raw_input_contents.append(np.random.rand(1, 3, 224, 224).astype(np.float32).tobytes())
            
    # Send the inference request and receive the response.
    response, _ = grpc_stub.ModelInfer.with_call(request, metadata=metadata)
    
    # Extract the output tensor from the response.
    output_contents = response.raw_output_contents[0]  # Assume there is only one output tensor.
    output_shape = [1, 1000]  # Assume the output tensor shape is [1, 1000].
    
    # Convert the output bytes to a NumPy array.
    output_array = np.frombuffer(output_contents, dtype=np.float32)
    output_array = output_array.reshape(output_shape)
    
    # Print the model's output result.
    print("Model output:\n", output_array)

client = httpclient.InferenceServerClient(url=url, verbose=True)

POST /api/predict/triton_test/v2/models/resnet50_pt/versions/1/infer, headers {'Authorization': '************1ZDY3OTEzNA=='}
b'{"inputs":[{"name":"INPUT__0","shape":[1,3,32,32],"datatype":"FP32","data":[1.0,1.0,1.0,.....,1.0]}],"outputs":[{"name":"OUTPUT__0","parameters":{"binary_data":false}}]}'