Platform For AI:SDK for Python

Common parameters

endpoint: the server address for your EAS service.

• Default gateway — use the service-specific endpoint. Example: 182848887922***.cn-shanghai.pai-eas.aliyuncs.com

• VPC direct connection — use the regional common endpoint. Example for China (Shanghai): pai-eas-vpc.cn-shanghai.aliyuncs.com

PredictClient

PredictClient is the main client for sending prediction requests to EAS services.

StringRequest

StringResponse

TFRequest

TFResponse

TorchRequest

TorchRequest uses a numeric index to identify tensors instead of names.

TorchResponse

BladeRequest

BladeRequest is for Blade accelerated models. Unlike TFRequest and TorchRequest, add_feed takes an explicit batch_size parameter before shape.

QueueClient

QueueClient supports queue-based data streaming, where one process pushes data and another subscribes to results.

Watcher

String input and output

For services using custom processors (such as PMML models), send and receive data as strings:

#!/usr/bin/env python

from eas_prediction import PredictClient
from eas_prediction import StringRequest

if __name__ == '__main__':
    client = PredictClient('http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'scorecard_pmml_example')
    client.set_token('YWFlMDYyZDNmNTc3M2I3MzMwYmY0MmYwM2Y2MTYxMTY4NzBkNzdj****')
    client.init()

    request = StringRequest('[{"fea1": 1, "fea2": 2}]')
    for x in range(0, 1000000):
        resp = client.predict(request)
        print(resp)

TensorFlow tensor input and output

For TensorFlow services, use TFRequest to specify input tensors by name and TFResponse to read output tensors:

#!/usr/bin/env python

from eas_prediction import PredictClient
from eas_prediction import TFRequest

if __name__ == '__main__':
    client = PredictClient('http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'mnist_saved_model_example')
    client.set_token('YTg2ZjE0ZjM4ZmE3OTc0NzYxZDMyNmYzMTJjZTQ1YmU0N2FjMTAy****')
    client.init()

    req = TFRequest('predict_images')
    # add_feed takes a flat 1D array; shape [1, 784] describes the tensor dimensions
    req.add_feed('images', [1, 784], TFRequest.DT_FLOAT, [1] * 784)
    for x in range(0, 1000000):
        resp = client.predict(req)
        print(resp)

VPC direct connection

VPC direct connection is available only for services in EAS dedicated resource groups where the resource group and vSwitch are in the same Virtual Private Cloud (VPC). For setup, see Work with dedicated resource groups and Network access configuration.

The only difference from the standard setup is set_endpoint_type(ENDPOINT_TYPE_DIRECT) and the regional VPC endpoint:

#!/usr/bin/env python

from eas_prediction import PredictClient
from eas_prediction import TFRequest
from eas_prediction import ENDPOINT_TYPE_DIRECT

if __name__ == '__main__':
    # Use the regional VPC endpoint instead of the service-specific endpoint
    client = PredictClient('http://pai-eas-vpc.cn-hangzhou.aliyuncs.com', 'mnist_saved_model_example')
    client.set_token('M2FhNjJlZDBmMzBmMzE4NjFiNzZhMmUxY2IxZjkyMDczNzAzYjFi****')
    client.set_endpoint_type(ENDPOINT_TYPE_DIRECT)
    client.init()

    request = TFRequest('predict_images')
    request.add_feed('images', [1, 784], TFRequest.DT_FLOAT, [1] * 784)
    for x in range(0, 1000000):
        resp = client.predict(request)
        print(resp)

PyTorch tensor input and output

For PyTorch services, use TorchRequest with numeric indexes instead of tensor names:

#!/usr/bin/env python

from eas_prediction import PredictClient
from eas_prediction import TorchRequest

if __name__ == '__main__':
    client = PredictClient('http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'pytorch_gpu_wl')
    client.init()

    req = TorchRequest()
    # add_feed takes a flat 1D array; shape [1, 3, 224, 224] describes a single 224x224 RGB image
    req.add_feed(0, [1, 3, 224, 224], TorchRequest.DT_FLOAT, [1] * 150528)
    # req.add_fetch(0)  # Uncomment to return only tensor at index 0

    import time
    st = time.time()
    timer = 0
    for x in range(0, 10):
        resp = client.predict(req)
        timer += (time.time() - st)
        st = time.time()
        print(resp.get_tensor_shape(0))
    print("average response time: %s s" % (timer / 10))

Blade processor-based model

For Blade accelerated services, use BladeRequest. Unlike TFRequest and TorchRequest, add_feed takes an explicit batch_size parameter before shape:

#!/usr/bin/env python

from eas_prediction import PredictClient
from eas_prediction import BladeRequest

if __name__ == '__main__':
    client = PredictClient('http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'nlp_model_example')
    client.init()

    req = BladeRequest()
    req.add_feed('input_data', 1, [1, 360, 128], BladeRequest.DT_FLOAT, [0.8] * 85680)
    req.add_feed('input_length', 1, [1], BladeRequest.DT_INT32, [187])
    req.add_feed('start_token', 1, [1], BladeRequest.DT_INT32, [104])
    req.add_fetch('output', BladeRequest.DT_FLOAT)

    import time
    st = time.time()
    timer = 0
    for x in range(0, 10):
        resp = client.predict(req)
        timer += (time.time() - st)
        st = time.time()
        print(resp.get_tensor_shape('output'))
    print("average response time: %s s" % (timer / 10))

Blade with TensorFlow compatibility

For EAS Blade models that expose a TensorFlow-compatible interface, import TFRequest from eas_prediction.blade_tf_request instead of the standard path:

#!/usr/bin/env python

from eas_prediction import PredictClient
from eas_prediction.blade_tf_request import TFRequest  # Blade-compatible TFRequest

if __name__ == '__main__':
    client = PredictClient('http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'nlp_model_example')
    client.init()

    req = TFRequest(signature_name='predict_words')
    req.add_feed('input_data', [1, 360, 128], TFRequest.DT_FLOAT, [0.8] * 85680)
    req.add_feed('input_length', [1], TFRequest.DT_INT32, [187])
    req.add_feed('start_token', [1], TFRequest.DT_INT32, [104])
    req.add_fetch('output')

    import time
    st = time.time()
    timer = 0
    for x in range(0, 10):
        resp = client.predict(req)
        timer += (time.time() - st)
        st = time.time()
        print(resp.get_tensor_shape('output'))
    print("average response time: %s s" % (timer / 10))

Queue service for data streaming

Send records to a queue while a separate thread subscribes to results using a Watcher. This pattern is common for asynchronous batch processing pipelines.

#!/usr/bin/env python

from eas_prediction import QueueClient
import threading

if __name__ == '__main__':
    endpoint = '182848887922****.cn-shanghai.pai-eas.aliyuncs.com'
    queue_name = 'test_group.qservice/sink'
    token = 'YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MTUx****'

    queue = QueueClient(endpoint, queue_name)
    queue.set_token(token)
    queue.init()
    queue.set_timeout(30000)

    # Clear the queue before starting
    attributes = queue.attributes()
    if 'stream.lastEntry' in attributes:
        queue.truncate(int(attributes['stream.lastEntry']) + 1)

    count = 100

    def send_thread():
        for i in range(count):
            index, request_id = queue.put('[{}]')
            print('send: ', i, index, request_id)

    def watch_thread():
        # window=5 limits the server to pushing at most 5 unacknowledged records at a time
        watcher = queue.watch(0, 5, auto_commit=True)
        i = 0
        for x in watcher.run():
            print('recv: ', i, x.index, x.tags['requestId'])
            i += 1
            if i == count:
                break
        watcher.close()

    thread1 = threading.Thread(target=watch_thread)
    thread2 = threading.Thread(target=send_thread)

    thread1.start()
    thread2.start()

    thread1.join()
    thread2.join()