inferência assíncrona, serviço de fila - Platform For AI: Central de Documentação do Alibaba Cloud

O PAI EAS disponibiliza três interfaces para interação com uma fila de inferência assíncrona: uma API HTTP, um SDK Python e a CLI eascmd. Este documento aborda as três opções.

Como funciona

Ao implantar um serviço de inferência assíncrona, o EAS cria automaticamente duas filas:

Fila de entrada: recebe solicitações de inferência dos clientes.
Fila de saída (sink): armazena os resultados de inferência gravados pelo serviço.

O fluxo de dados é o seguinte:

Envie uma solicitação de inferência para a fila de entrada. O EAS retorna imediatamente um ID de solicitação e um índice de fila.
O serviço de inferência lê os dados da fila de entrada, executa a inferência e grava os resultados na fila de saída.
Recupere o resultado da fila de saída usando o ID da solicitação ou o índice.

Pré-requisitos

Antes de começar, verifique se você tem:

Um serviço de inferência assíncrona implantado no PAI EAS
O endpoint da fila de entrada, o endpoint da fila de saída e o token do seu serviço

Para encontrar esses valores, acesse a aba Inference Service, clique em no nome do serviço para abrir a página Overview e, em seguida, clique em View Endpoint Information na seção Basic Information.

Os formatos dos endpoints são:

Endpoint	Formato	Exemplo
Fila de entrada	`{domain}/api/predict/{service_name}`	`xxx.cn-shanghai.pai-eas.aliyuncs.com/api/predict/{service_name}`
Fila de saída (sink)	`{domain}/api/predict/{service_name}/sink`	`xxx.cn-shanghai.pai-eas.aliyuncs.com/api/predict/{service_name}/sink`

Acesse um serviço de fila por API

Todas as solicitações HTTP exigem um cabeçalho Authorization com o token do seu serviço.

Referência rápida:

Operação	Método	Caminho	Parâmetros principais
Enviar dados	POST	`/api/predict/{service}`	`_priority_=1` (opcional)
Visualize detalhes da fila	GET	`/api/predict/{service}?_attrs_=true`	—
Consultar dados por índice	GET	`/api/predict/{service}?_index_=N`	`_length_`, `_auto_delete_`, `_timeout_`
Consultar dados por ID de solicitação	GET	`/api/predict/{service}?requestId=...`	`_timeout_`
Consultar resultados de inferência	GET	`/api/predict/{service}/sink?requestId=...`	`_timeout_`
Exclua um único item	DELETE	`/api/predict/{service}?_index_=N`	—
Truncar a fila	DELETE	`/api/predict/{service}?_index_=N&_trunc_=true`	—

Envie dados para uma fila

Envie uma solicitação para a fila de entrada usando curl:

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

A resposta inclui:

> 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

A resposta retorna dois identificadores para uso em consultas posteriores:

Cabeçalho X-Eas-Queueservice-Request-Id: o ID da solicitação (4e034bnvb-e783-4272-9333-68x6a1v8dc6x).
Corpo da resposta: o índice da fila (1033).

Envie dados prioritários

Por padrão, a fila processa dados na ordem FIFO (First-In, First-Out). Para priorizar solicitações específicas, adicione _priority_=1 à string de consulta:

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

Os dados prioritários são enviados aos assinantes antes dos dados com prioridade padrão.

Visualize detalhes da fila

Adicione _attrs_=true a uma solicitação GET para recuperar metadados da fila:

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

A resposta é um 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 principais:

Campo	Descrição
`stream.maxPayloadBytes`	Tamanho máximo em bytes de um único item de dados
`stream.approxMaxLength`	Número máximo de itens de dados que a fila pode armazenar
`stream.firstEntry`	Índice do primeiro item na fila
`stream.lastEntry`	Índice do último item na fila
`stream.length`	Número atual de itens na fila
`meta.state`	Estado atual da fila

Alternativamente, acesse a página Elastic Algorithm Service (EAS), clique em no nome do serviço e mude para a aba Asynchronous Queue.

Consulte dados

Escolha um método de recuperação adequado ao seu caso de uso:

Método	Funcionamento	Quando usar
Polling	Consulta dados por índice ou ID de solicitação sob demanda	Ideal para recuperar um item específico ou verificar o estado da fila ocasionalmente
Assinatura	Assina via WebSocket para receber dados conforme chegam	Recomendado para processamento contínuo de fluxo ou cenários de alto throughput

Faça polling de dados da fila de entrada

Consulte dados por índice ou ID de solicitação:

# 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"

Resposta:

> 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 status da resposta:

Código de status	Significado
`200 OK`	Dados encontrados e retornados
`204 No Content`	Não existem dados correspondentes (retornado imediatamente quando `_timeout_=0`)

Parâmetros de consulta:

Parâmetro	Tipo	Padrão	Descrição
`_index_`	INT	`0`	Índice inicial. Defina este valor próximo ao índice do item alvo para maior eficiência na consulta.
`_length_`	INT	`1`	Número de itens a recuperar.
`_auto_delete_`	BOOL	`true`	Exclui os itens consultados da fila após a recuperação.
`_timeout_`	STRING	`0`	Tempo de espera caso não existam dados correspondentes. O valor `0` retorna HTTP 204 imediatamente. Valores de exemplo: `1s`, `1m`.
`requestId`	STRING	—	Tag integrada usada para localizar um item específico. O framework EAS marca cada item de entrada com `requestId` e o propaga até a fila de saída, permitindo usar o mesmo ID de solicitação para consultar o resultado de inferência na fila de saída.

Faça polling de resultados de inferência da fila de saída

Consulte resultados na fila de saída usando o ID de solicitação do envio original:

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

Resposta:

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

Exclua dados

Remova dados da fila de duas maneiras: exclua um único item ou trunque todos os itens até um determinado índice.

Exclua um único item

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

Resposta:

> 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:

Parâmetro	Tipo	Descrição
`_index_`	INT	Índice do item a excluir

Trunque a fila

Exclua todos os itens com índice inferior ao valor especificado:

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

Resposta:

> 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:

Parâmetro	Tipo	Descrição
`_index_`	INT	Índice de corte. Todos os itens com índice inferior a este valor serão excluídos.
`_trunc_`	BOOL	Deve ser `true` para realizar o truncamento. Se omitido ou definido como `false`, a solicitação executará a exclusão de um único item.

Assine uma fila (SDK Python)

Para processamento contínuo de fluxo, assine a fila de saída usando o SDK Python. O serviço de fila utiliza o protocolo WebSocket para manter uma conexão persistente e enviar dados aos assinantes assim que chegarem. O tamanho da janela de assinatura é controlado pela configuração worker_threads na instância do serviço de inferência.

Um serviço de inferência não é obrigatório. Use o sdk para assinar a fila de entrada em um serviço personalizado e grave os resultados em uma fila de mensagens de terceiros ou em um armazenamento como o Object Storage Service (oss).

Instale o sdk:

pip install eas_prediction --user

Assine a fila de saída:

O exemplo abaixo usa QueueClient para enviar 10 itens à fila de entrada e assinar os resultados da fila de saída. Em produção, utilize threads separadas para envio e assinatura.

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

Acesse um serviço de fila via eascmd

A ferramenta eascmd encapsula toda a API do serviço de fila. Utilize o subcomando eascmd stream para operar e depurar um serviço de fila diretamente da linha de comando.

A versão do eascmd deve ser superior a 2.6.0. Para instruções de download e configure, consulte Baixe e autentique o cliente .

Referência rápida:

Comando	Descrição
`eascmd stream config --url=... --token=...`	Configure o acesso ao serviço de fila
`eascmd stream info`	Visualize detalhes da fila
`eascmd stream put -d "..."`	Envie dados inline
`eascmd stream put -f test.data`	Envie todas as linhas de um arquivo
`eascmd stream get -l10 --timeout=3s`	Consulta até 10 itens com timeout
`eascmd stream get --tags requestId=...`	Filtra por ID de solicitação
`eascmd stream delete 3`	Exclua um único item por índice
`eascmd stream trunc 4`	Trunca todos os itens até o índice 4
`eascmd stream watch`	Assina (commit manual)
`eascmd stream watch --auto-commit`	Assina (auto-commit)

Configure o acesso

Execute eascmd stream config para apontar a CLI para o seu serviço de fila:

eascmd stream config \
  --url=http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com/api/predict/qservice \
  --token=YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==

Após a configuração, o eascmd utiliza default_group e default_user como valores padrão para group_id e user_id. Substitua-os usando --group e --user em qualquer comando subsequente.

Visualize detalhes da fila

eascmd stream info

Saída de exemplo:

[OK] Attributes:
consumers.list.[0] : Id: imageasync.imageasync-35d72370-5f576f7c8d-2mdb4, Index: 0, Pending: 0, Status: Running, Idle: 19.997s, Window: 5, Slots: 5, AutoCommit: false
consumers.stats.total : 1
consumers.status.total : 1
groups.list.[0] : Id: imageasync, Index: 0, Pending: 0, Delivered: 1, Consumers: 1
meta.header.group : X-EAS-QueueService-Gid
meta.header.priority : X-EAS-QueueService-Priority
meta.header.user : X-EAS-QueueService-Uid
meta.maxPayloadBytes : 8192
meta.name : imageasync-queue-38895e88
meta.state : Normal
stream.approxMaxLength : 230399
stream.firstEntry : 0
stream.lastEntry : 0
stream.length : 0

Para descrições dos campos, consulte Visualize detalhes da fila. Use o comando info para verifique o estado da fila e testar a conectividade.

Envie dados

Envie dados inline com a flag -d:

eascmd stream put -d "10s"

Saída:

[OK] 1
[INFO] Put data done.
Total time cost: 401.892141ms
Total size: 3.00 B
Total: 1, success: 1, failed: 0

Envie todas as linhas de um arquivo com a flag -f:

eascmd stream put -f test.data

Saída:

[INFO] Opening data file: test.data
[OK] 2
[OK] 3
[OK] 4
...

Consulte dados

Busque até 10 itens com timeout de 3 segundos:

eascmd stream get -l10 --timeout=3s

Saída:

[OK] [0 - 1] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=e47b76e2-2648-40fe-9197-a268015cbd1f ts@source=1685802680575] data1
[OK] [1 - 2] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=51d13952-6ba3-4d52-b548-e58837675c7a ts@source=1685807531686] data2
[OK] [2 - 3] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=ef6940e3-159c-45f3-a96d-bc0acd71275f ts@source=1685807531701] OK
...

Cada linha contém três colunas:

[received_index - queue_index]: por exemplo, [0 - 1] indica que o primeiro item recebido possui índice de fila 1.
tags[...]: metadados anexados ao item:
- Header:*: cabeçalhos HTTP da solicitação original.
- requestId: ID de solicitação gerado automaticamente.
- ts@source: timestamp Unix de quando a fila de entrada recebeu a solicitação.
- ts@sink: timestamp Unix de quando a fila de saída recebeu os dados.
Conteúdo dos dados.

Importante

Se uma instância de serviço de inferência estiver em execução simultaneamente, ela poderá consumir dados da fila de entrada antes que você consiga consultá-los. Adicione a flag -k para consultar a partir da fila de saída.

Para filtrar por ID de solicitação:

eascmd stream get --tags requestId=ef6940e3-159c-45f3-a96d-bc0acd71275f

Saída:

[OK] [0 - 3] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=ef6940e3-159c-45f3-a96d-bc0acd71275f ts@source=1685807531701] OK

Exclua dados

Exclua um único item por índice:

eascmd stream delete 3

Confirme quando solicitado:

Deleting index(es):
3 [y/N]y
[OK] deleted

Trunque todos os itens até um determinado índice:

eascmd stream trunc 4

Confirme quando solicitado:

trunc stream from index: 4 [y/N]y
[OK] truncated

Assine a fila

Utilize o comando watch para assinar e receber dados à medida que são enviados. Por padrão, o cliente realiza o commit de cada item manualmente:

eascmd stream watch

Saída:

[INFO] Start to watch: index: 0, indexOnly: false, autoCommit: false, window: 10
I0604 09:20:45.211243   66197 queue.go:532] watch via websocket
[OK] [0 - 4] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=2a5645b6-9ee5-4026-bdee-fab31e435934 ts@sink=1685807531718 ts@source=1685807531715] data4
commit: 4 ? [Y/n]

Insira Y para confirmar o item e receber o próximo.
Insira n para executar um commit negativo (rejeitar o item):

commit: 5 ? [Y/n]n
negative: 5 ? [Y/n]y

Para detalhes sobre a semântica de commit e commit negativo, consulte Commit e Negativo.

Para que o servidor realize o commit dos itens automaticamente:

eascmd stream watch --auto-commit

Saída:

[INFO] Start to watch: index: 0, indexOnly: false, autoCommit: true, window: 10
I0604 09:30:08.554542   66408 queue.go:532] watch via websocket
[OK] [0 - 5] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=64ba6aaf-49b0-45c7-8d79-6cf6dc1065d0 ts@sink=1685807531733 ts@source=1685807531730] data5
[OK] [1 - 6] tags[Header:Content-Type=text/plain; charset=utf-8 requestId=5825dd3e-a5e2-4754-a946-96e068d643c8 ts@sink=1685807531771 ts@source=1685807531768] data6
...

Para visualize todas as opções e comandos disponíveis do eascmd stream, execute eascmd stream help.

Próximos passos

Assinatura e push do serviço de fila: saiba mais sobre grupos de consumidores, usuários e semântica de commit.
Baixe e autentique o cliente: configure e atualize o eascmd.