asynchronous inference, queue service - Platform For AI - Centro de documentación de Alibaba Cloud

PAI EAS expone tres interfaces para interactuar con una cola de inference asíncrona: una API HTTP, un SDK de Python y la CLI eascmd. Este documento aborda las tres opciones.

Funcionamiento

Al desplegar un servicio de inference asíncrona, EAS crea automáticamente dos colas:

Cola de entrada: acepta solicitudes de inference de los clientes.
Cola de salida (sink): almacena los resultados de inference que escribe el servicio de inference.

El flujo de datos es el siguiente:

Envíe una solicitud de inference a la cola de entrada. EAS devuelve un ID de solicitud y un índice de cola de inmediato.
El servicio de inference lee datos de la cola de entrada, ejecuta la inference y escribe los resultados en la cola de salida.
Recupere el resultado de la cola de salida mediante el ID de solicitud o el índice.

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

Un servicio de inference asíncrona desplegado en PAI EAS
El endpoint de la cola de entrada, el endpoint de la cola de salida y el token de su servicio

Para obtener estos valores, vaya a la pestaña Inference Service, haga clic en el nombre del servicio para abrir la página Overview y luego haga clic en View Endpoint Information en la sección Basic Information. En el cuadro de diálogo View Endpoint Information, haga clic en la pestaña Shared Gateway > Asynchronous Call. Los endpoints se agrupan por tipo de red, Internet y VPC, y cada grupo proporciona un endpoint de entrada y uno de salida del servicio de colas.

Los formatos de los endpoints son:

Endpoint	Format	Example
Cola de entrada	`{domain}/api/predict/{service_name}`	`xxx.cn-shanghai.pai-eas.aliyuncs.com/api/predict/{service_name}`
Cola de salida (sink)	`{domain}/api/predict/{service_name}/sink`	`xxx.cn-shanghai.pai-eas.aliyuncs.com/api/predict/{service_name}/sink`

Acceso a un servicio de colas mediante API

Todas las solicitudes HTTP requieren un encabezado Authorization que contenga el token de su servicio.

Referencia rápida:

Operation	Method	Path	Key parameters
Enviar datos	POST	`/api/predict/{service}`	`_priority_=1` (opcional)
Consultar detalles de la cola	GET	`/api/predict/{service}?_attrs_=true`	—
Consultar datos por índice	GET	`/api/predict/{service}?_index_=N`	`_length_`, `_auto_delete_`, `_timeout_`
Consultar datos por ID de solicitud	GET	`/api/predict/{service}?requestId=...`	`_timeout_`
Consultar resultados de inference	GET	`/api/predict/{service}/sink?requestId=...`	`_timeout_`
Eliminar un solo elemento	DELETE	`/api/predict/{service}?_index_=N`	—
Truncar la cola	DELETE	`/api/predict/{service}?_index_=N&_trunc_=true`	—

Envío de datos a una cola

Envíe una solicitud a la cola de entrada mediante curl:

curl -v http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice \
  -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  -d '[{}]'

La respuesta incluye:

> POST /api/predict/qservice HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 19
< X-Eas-Queueservice-Request-Id: 4e034bnvb-e783-4272-9333-68x6a1v8dc6x
<
1033

Se devuelven dos identificadores que puede usar para consultar datos posteriormente:

Encabezado X-Eas-Queueservice-Request-Id: el ID de solicitud (4e034bnvb-e783-4272-9333-68x6a1v8dc6x).
Cuerpo de la respuesta: el índice de la cola (1033).

Envío de datos prioritarios

De forma predeterminada, la cola procesa los datos en orden FIFO (primero en entrar, primero en salir). Para dar prioridad a solicitudes específicas, agregue _priority_=1 a la cadena de consulta:

curl -v "http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_priority_=1" \
  -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  -d '[{}]'

Los datos prioritarios se envían a los suscriptores antes que los datos de prioridad estándar.

Consulta de detalles de la cola

Agregue _attrs_=true a una solicitud GET para recuperar los metadatos de la cola:

curl -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  "http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_attrs_=true"

La respuesta es un objeto JSON:

{"consumers.stats.total":"0","consumers.status.total":"0","meta.header.group":"X-EAS-QueueService-Gid","meta.header.priority":"X-EAS-QueueService-Priority","meta.header.user":"X-EAS-QueueService-Uid","stream.maxPayloadBytes":"524288","meta.name":"pmml_test","meta.state":"Normal","stream.approxMaxLength":"4095","stream.firstEntry":"0","stream.lastEntry":"0","stream.length":"1"}

Campos principales:

Field	Description
`stream.maxPayloadBytes`	Tamaño máximo en bytes de un solo elemento de datos
`stream.approxMaxLength`	Número máximo de elementos de datos que puede contener la cola
`stream.firstEntry`	Índice del primer elemento en la cola
`stream.lastEntry`	Índice del último elemento en la cola
`stream.length`	Número actual de elementos en la cola
`meta.state`	Estado actual de la cola

Como alternativa, vaya a la página Elastic Algorithm Service (EAS), haga clic en el nombre del servicio y cambie a la pestaña Asynchronous Queue.

Esta pestaña muestra la información básica de la cola (el resource group al que pertenece, la hora de creación, el tamaño máximo de una sola solicitud de entrada y el tamaño máximo de una sola respuesta de salida), los recursos desplegados para el servicio (número de instancias, CPU y memoria), así como el número actual de elementos almacenados en la cola de entrada junto con el estado de procesamiento de cada instancia.

Consulta de datos

Elija un método de recuperación según su caso de uso:

Method	How it works	When to use
Sondeo	Consulte datos por índice o ID de solicitud bajo demanda	Recuperación de un elemento específico o comprobación ocasional del estado de la cola
Suscripción	Suscríbase mediante WebSocket para recibir datos conforme llegan	Procesamiento continuo de flujos o escenarios de alto throughput

Sondeo de datos desde la cola de entrada

Consulte datos por índice o ID de solicitud:

# Query by index
curl -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  "http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_index_=1022"

# Query by request ID
curl -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  "http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?requestId=87633037-39a4-40bf-8405-14f8e0c31896"

La respuesta:

> GET /api/predict/qservice?_index_=1022&_auto_delete_=false HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 4
< Content-Type: text/plain; charset=utf-8
<
[{}]

Códigos de estado de la respuesta:

Status code	Meaning
`200 OK`	Datos encontrados y devueltos
`204 No Content`	No existen datos coincidentes (se devuelve inmediatamente cuando `_timeout_=0`)

Parámetros de consulta:

Parameter	Type	Default	Description
`_index_`	INT	`0`	Índice inicial. Establezca este valor cerca del índice del elemento objetivo para mejorar la eficiencia de la consulta.
`_length_`	INT	`1`	Número de elementos a recuperar.
`_auto_delete_`	BOOL	`true`	Elimina los elementos consultados de la cola después de la recuperación.
`_timeout_`	STRING	`0`	Tiempo de espera si no existen datos coincidentes. `0` devuelve HTTP 204 inmediatamente. Valores de ejemplo: `1s`, `1m`.
`requestId`	STRING	—	Etiqueta integrada que se usa para localizar un elemento específico. El framework de EAS etiqueta cada elemento de entrada con `requestId` y lo traslada hasta la cola de salida, de modo que puede usar el mismo ID de solicitud para consultar el resultado de inference desde la cola de salida.

Sondeo de resultados de inference desde la cola de salida

Consulte los resultados de la cola de salida mediante el ID de solicitud del envío original:

curl -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  "http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice/sink?requestId=0337f7a1-a6f6-49a6-8ad7-ff2fd12bbe2d"

La respuesta:

> GET /api/predict/qservice/sink?requestId=0337f7a1-a6f6-49a6-8ad7-ff2fd12b**** HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 53
< Content-Type: text/plain; charset=utf-8
<
[{"p_0":0.5224580736905329,"p_1":0.4775419263094671}]

Eliminación de datos

Quite datos de la cola de dos formas: elimine un solo elemento o trunque todos los elementos hasta un índice determinado.

Eliminación de un solo elemento

curl -XDELETE -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  "http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_index_=1022"

Respuesta:

> DELETE /api/predict/qservice?_index_=1022 HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 4
< Content-Type: text/plain; charset=utf-8
<
OK

Parámetro:

Parameter	Type	Description
`_index_`	INT	Índice del elemento a eliminar

Truncamiento de la cola

Elimine todos los elementos con un índice inferior al valor especificado:

curl -XDELETE -v -H 'Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==' \
  "http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice?_index_=1023&_trunc_=true"

Respuesta:

> DELETE /api/predict/qservice?_index_=1023&_trunc_=true HTTP/1.1
> Host: 182848887922****.cn-shanghai.pai-eas.aliyuncs.com
> Authorization: YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==
>
< HTTP/1.1 200 OK
< Content-Length: 4
< Content-Type: text/plain; charset=utf-8
<
OK

Parámetros:

Parameter	Type	Description
`_index_`	INT	Índice de corte. Se eliminan todos los elementos con un índice inferior a este valor.
`_trunc_`	BOOL	Debe ser `true` para realizar un truncamiento. Si se omite o es `false`, la solicitud realiza una eliminación de un solo elemento en su lugar.

Suscripción a una cola (SDK de Python)

Para el procesamiento continuo de flujos, suscríbase a la cola de salida mediante el SDK de Python. El servicio de colas utiliza el protocolo WebSocket para mantener una conexión persistente y enviar datos a los suscriptores conforme llegan. El tamaño de la ventana de suscripción se controla mediante la configuración worker_threads en la instancia del servicio de inference.

No es obligatorio contar con un servicio de inference. Use el SDK para suscribirse a la cola de entrada en un servicio personalizado y escriba los resultados en una cola de mensajes de terceros o en un almacenamiento como Object Storage Service (OSS).

Instale el SDK:

pip install eas_prediction --user

Suscríbase a la cola de salida:

El siguiente ejemplo usa QueueClient para enviar 10 elementos a la cola de entrada y suscribirse a los resultados de la cola de salida. En producción, use hilos separados para el envío y la suscripción.

#!/usr/bin/env python
from eas_prediction import QueueClient

# Create a client for the input queue.
input_queue = QueueClient('182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'qservice')
# To set a custom user or group:
# input_queue = QueueClient('182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'qservice', uid='your_user_id', gid='your_group_id')
input_queue.set_token('YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==')
input_queue.init()

# Create a client for the output queue.
sink_queue = QueueClient('182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'qservice/sink')
sink_queue.set_token('YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==')
sink_queue.init()

# Send 10 items to the input queue.
for x in range(10):
    index, request_id = input_queue.put('[{}]')
    print(index, request_id)

    # Print queue attributes.
    attrs = input_queue.attributes()
    print(attrs)

# Subscribe to the output queue with a window size of 5.
i = 0
watcher = sink_queue.watch(0, 5, auto_commit=False)
for x in watcher.run():
    print(x.data.decode('utf-8'))

    # Commit each item after processing.
    sink_queue.commit(x.index)
    i += 1
    if i == 10:
        break

# Close the watcher. Each QueueClient instance supports only one active watcher.
# Failing to close the watcher causes an error on the next watch() call.
watcher.close()

Pasos siguientes

Queue service subscription and push: obtenga información sobre grupos de consumidores, usuarios y semántica de confirmación.
Download and authenticate the client: configure y actualice eascmd.