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

Como funciona

Ao implantar um service 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 service de inferência.

O fluxo de dados ocorre da seguinte forma:

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 service 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ê possui:

Um service 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 service

Para encontrar esses valores, acesse a aba Inference Service, clique em no nome do service para abrir a página Overview e, em seguida, clique em View Endpoint Information na seção Basic Information. Na caixa de diálogo View Endpoint Information, clique em na aba Shared Gateway > Asynchronous Call. Os endpoints são agrupados por tipo de rede (Internet e VPC), e cada grupo fornece um endpoint de entrada e um endpoint de saída do service de fila.

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 service de fila por API

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

Referência rápida:

Operação	Método	Caminho	Parâmetros principais
Enviar dados	POST	`/api/predict/{service}`	`_priority_=1` (opcional)
Visualizar 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_`
Excluir 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

Dois identificadores são retornados 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 '[{}]'

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 service e mude para a aba Asynchronous Queue.

Essa aba exibe as informações básicas da fila (grupo de recursos, horário de criação, tamanho máximo de solicitação de entrada e de resposta de saída), os recursos implantados para o service (número de instâncias, CPU e memória), o número atual de itens armazenados na fila de entrada e o status de processamento de cada instância.

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"

A 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 de 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 serem recuperados.
`_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 da 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"

A 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 ser excluído

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 service de fila utiliza o protocolo WebSocket para manter uma conexão persistente e enviar dados aos assinantes assim que eles chegam. O tamanho da janela de assinatura é controlado pela configuração worker_threads na instância do service de inferência.

Um service de inferência não é obrigatório. Use o SDK para assinar a fila de entrada em um service 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()

Próximos passos

Queue service subscription and push — saiba mais sobre grupos de consumidores, usuários e semântica de commit.
Download and authenticate the client — configure e atualize o eascmd.