inférence asynchrone, service de file d'attente - Platform For AI

Fonctionnement

Lorsque vous déployez un service d'inférence asynchrone, EAS crée automatiquement deux files d'attente :

File d'entrée : accepte les requêtes d'inférence des clients.
File de sortie (sink) : stocke les résultats d'inférence écrits par le service d'inférence.

Le flux de données est le suivant :

Soumettez une requête d'inférence à la file d'entrée. EAS renvoie immédiatement un ID de requête et un index de file d'attente.
Le service d'inférence lit les données de la file d'entrée, exécute l'inférence et écrit les résultats dans la file de sortie.
Récupérez le résultat depuis la file de sortie en utilisant l'ID de requête ou l'index.

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

Un service d'inférence asynchrone déployé dans PAI EAS
Le point de terminaison de la file d'entrée, le point de terminaison de la file de sortie et le jeton de votre service

Pour trouver ces valeurs, accédez à l'onglet Inference Service, cliquez sur le nom du service pour ouvrir la page Overview, puis cliquez sur View Endpoint Information dans la section Basic Information. Dans la boîte de dialogue View Endpoint Information, cliquez sur l'onglet Shared Gateway > Asynchronous Call. Les points de terminaison sont regroupés par type de réseau, Internet et VPC. Chaque groupe fournit un point de terminaison d'entrée et un point de terminaison de sortie pour le service de file d'attente.

Les formats des points de terminaison sont les suivants :

Point de terminaison

Format

Exemple

File d'entrée

{domain}/api/predict/{service_name}

xxx.cn-shanghai.pai-eas.aliyuncs.com/api/predict/{service_name}

File de sortie (sink)

{domain}/api/predict/{service_name}/sink

xxx.cn-shanghai.pai-eas.aliyuncs.com/api/predict/{service_name}/sink

Accéder à un service de file d'attente via l'API

Toutes les requêtes HTTP nécessitent un en-tête Authorization contenant le jeton de votre service.

Référence rapide :

Opération

Méthode

Chemin

Paramètres clés

Envoyer des données

POST

/api/predict/{service}

_priority_=1 (facultatif)

Afficher les détails de la file d'attente

GET

/api/predict/{service}?_attrs_=true

—

Interroger les données par index

GET

/api/predict/{service}?_index_=N

_length_, _auto_delete_, _timeout_

Interroger les données par ID de requête

GET

/api/predict/{service}?requestId=...

_timeout_

Interroger les résultats d'inférence

GET

/api/predict/{service}/sink?requestId=...

_timeout_

Supprimer un élément unique

DELETE

/api/predict/{service}?_index_=N

—

Tronquer la file d'attente

DELETE

/api/predict/{service}?_index_=N&_trunc_=true

—

Envoyer des données à une file d'attente

Soumettez une requête à la file d'entrée en utilisant curl :

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

La réponse inclut :

> 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

Deux identifiants sont renvoyés. Vous pouvez les utiliser pour interroger les données ultérieurement :

En-tête X-Eas-Queueservice-Request-Id : l'ID de requête (4e034bnvb-e783-4272-9333-68x6a1v8dc6x).
Corps de la réponse : l'index de la file d'attente (1033).

Envoyer des données prioritaires

Par défaut, la file d'attente traite les données selon l'ordre Premier entré, premier sorti (FIFO). Pour prioriser des requêtes spécifiques, ajoutez _priority_=1 à la chaîne de requête :

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

Les données prioritaires sont envoyées aux abonnés avant les données de priorité standard.

Afficher les détails de la file d'attente

Ajoutez _attrs_=true à une requête GET pour récupérer les métadonnées de la file d'attente :

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

La réponse est un objet 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"}

Champs clés :

Champ

Description

stream.maxPayloadBytes

Taille maximale en octets d'un élément de données unique

stream.approxMaxLength

Nombre maximal d'éléments de données que la file d'attente peut contenir

stream.firstEntry

Index du premier élément de la file d'attente

stream.lastEntry

Index du dernier élément de la file d'attente

stream.length

Nombre actuel d'éléments dans la file d'attente

meta.state

État actuel de la file d'attente

Vous pouvez également accéder à la page Elastic Algorithm Service (EAS), cliquer sur le nom du service, puis basculer vers l'onglet Asynchronous Queue.

Cet onglet affiche les informations de base de la file d'attente (le groupe de ressources auquel elle appartient, l'heure de création, la taille maximale d'une requête d'entrée unique et la taille maximale d'une réponse de sortie unique), les ressources déployées pour le service (nombre d'instances, CPU et mémoire), ainsi que le nombre actuel d'éléments de données stockés dans la file d'entrée et l'état de traitement de chaque instance.

Interroger les données

Choisissez une méthode de récupération adaptée à votre cas d'utilisation :

Méthode

Fonctionnement

Cas d'utilisation

Interrogation (Polling)

Interroge les données par index ou par ID de requête à la demande

Récupération d'un élément spécifique ou vérification occasionnelle de l'état de la file d'attente

Abonnement

S'abonne via WebSocket pour recevoir les données dès leur arrivée

Traitement de flux continu ou scénarios à haut débit

Interroger les données de la file d'entrée par interrogation

Interrogez les données par index ou par ID de requête :

# 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 réponse :

> 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
<
[{}]

Codes d'état de la réponse :

Code d'état

Signification

200 OK

Données trouvées et renvoyées

204 No Content

Aucune donnée correspondante n'existe (renvoyé immédiatement lorsque _timeout_=0)

Paramètres de requête :

Paramètre

Type

Valeur par défaut

Description

_index_

INT

0

Index de départ. Définissez cette valeur proche de l'index de l'élément cible pour une meilleure efficacité de requête.

_length_

INT

1

Nombre d'éléments à récupérer.

_auto_delete_

BOOL

true

Supprime les éléments interrogés de la file d'attente après la récupération.

_timeout_

STRING

0

Durée d'attente si aucune donnée correspondante n'existe. La valeur 0 renvoie immédiatement un code HTTP 204. Exemples de valeurs : 1s, 1m.

requestId

STRING

—

Tag intégré utilisé pour localiser un élément spécifique. Le framework EAS taggue chaque élément d'entrée avec requestId et le transmet jusqu'à la file de sortie, ce qui permet d'utiliser le même ID de requête pour interroger le résultat d'inférence depuis la file de sortie.

Interroger les résultats d'inférence de la file de sortie par interrogation

Interrogez les résultats de la file de sortie en utilisant l'ID de requête de la soumission initiale :

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

La réponse :

> 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}]

Supprimer des données

Supprimez les données de la file d'attente de deux manières : supprimez un élément unique ou tronquez tous les éléments jusqu'à un index donné.

Supprimer un élément unique

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

Réponse :

> 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

Paramètre :

Paramètre

Type

Description

_index_

INT

Index de l'élément à supprimer

Tronquer la file d'attente

Supprimez tous les éléments dont l'index est inférieur à la valeur spécifiée :

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

Réponse :

> 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

Paramètres :

Paramètre

Type

Description

_index_

INT

Index de coupure. Tous les éléments dont l'index est inférieur à cette valeur sont supprimés.

_trunc_

BOOL

Doit être défini sur true pour effectuer une troncature. Si omis ou défini sur false, la requête effectue une suppression d'élément unique à la place.

S'abonner à une file d'attente (SDK Python)

Pour le traitement de flux continu, abonnez-vous à la file de sortie en utilisant le SDK Python. Le service de file d'attente utilise le protocole WebSocket pour maintenir une connexion persistante et pousser les données vers les abonnés dès leur arrivée. La taille de la fenêtre d'abonnement est contrôlée par le paramètre worker_threads sur l'instance du service d'inférence.

Un service d'inférence n'est pas requis. Utilisez le SDK pour vous abonner à la file d'entrée dans un service personnalisé et écrire les résultats dans une file de messages tierce ou un stockage tel qu'Object Storage Service (OSS).

Installez le SDK :

pip install eas_prediction --user

Abonnez-vous à la file de sortie :

L'exemple suivant utilise QueueClient pour envoyer 10 éléments à la file d'entrée et s'abonner aux résultats de la file de sortie. En production, utilisez des threads distincts pour l'envoi et l'abonnement.

#!/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()

Étapes suivantes

Abonnement et envoi du service de file d'attente : découvrez les groupes de consommateurs, les utilisateurs et la sémantique de validation.
Télécharger et authentifier le client : configurez et mettez à jour eascmd.