inferensi asinkron, layanan antrian - Platform For AI - Alibaba Cloud Documentation Center

PAI EAS menyediakan tiga antarmuka untuk berinteraksi dengan antrian inferensi asinkron: API HTTP, SDK Python, dan CLI eascmd. Dokumen ini mencakup ketiganya.

Cara kerja

Saat Anda men-deploy layanan inferensi asinkron, EAS secara otomatis membuat dua antrian:

Antrian input — menerima permintaan inferensi dari klien.
Antrian output (sink) — menyimpan hasil inferensi yang ditulis oleh layanan inferensi.

Alur data adalah sebagai berikut:

Kirim permintaan inferensi ke antrian input. EAS segera mengembalikan ID permintaan dan indeks antrian.
Layanan inferensi membaca data dari antrian input, menjalankan inferensi, lalu menulis hasilnya ke antrian output.
Ambil hasil dari antrian output menggunakan ID permintaan atau indeks tersebut.

Prasyarat

Sebelum memulai, pastikan Anda telah memiliki:

Layanan inferensi asinkron yang telah di-deploy di PAI EAS
Titik akhir antrian input, titik akhir antrian output, dan token untuk layanan Anda

Untuk menemukan nilai-nilai tersebut, buka tab Inference Service, klik nama layanan untuk membuka halaman Overview, lalu klik View Endpoint Information pada bagian Basic Information. Di kotak dialog View Endpoint Information, klik tab Shared Gateway > Asynchronous Call. Titik akhir dikelompokkan berdasarkan jenis jaringan, Internet dan VPC, dan setiap kelompok menyediakan satu titik akhir input serta satu titik akhir output untuk layanan antrian.

Format titik akhir adalah sebagai berikut:

Endpoint	Format	Contoh
Antrian input	`{domain}/api/predict/{service_name}`	`xxx.cn-shanghai.pai-eas.aliyuncs.com/api/predict/{service_name}`
Antrian output (sink)	`{domain}/api/predict/{service_name}/sink`	`xxx.cn-shanghai.pai-eas.aliyuncs.com/api/predict/{service_name}/sink`

Akses layanan antrian melalui API

Semua permintaan HTTP memerlukan header Authorization yang berisi token layanan Anda.

Referensi cepat:

Operasi	Metode	Jalur	Parameter kunci
Kirim data	POST	`/api/predict/{service}`	`_priority_=1` (opsional)
Lihat detail antrian	GET	`/api/predict/{service}?_attrs_=true`	—
Kueri data berdasarkan indeks	GET	`/api/predict/{service}?_index_=N`	`_length_`, `_auto_delete_`, `_timeout_`
Kueri data berdasarkan request ID	GET	`/api/predict/{service}?requestId=...`	`_timeout_`
Kueri hasil inferensi	GET	`/api/predict/{service}/sink?requestId=...`	`_timeout_`
Hapus satu item	DELETE	`/api/predict/{service}?_index_=N`	—
Truncate antrian	DELETE	`/api/predict/{service}?_index_=N&_trunc_=true`	—

Kirim data ke antrian

Kirim permintaan ke antrian input menggunakan curl:

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

Tanggapan mencakup:

> 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

Dua identifikasi dikembalikan yang dapat Anda gunakan untuk mengkueri data nanti:

Header X-Eas-Queueservice-Request-Id — ID permintaan (4e034bnvb-e783-4272-9333-68x6a1v8dc6x).
Isi tanggapan — indeks antrian (1033).

Kirim data prioritas

Antrian memproses data dalam urutan First-In, First-Out (FIFO) secara default. Untuk memberikan prioritas pada permintaan tertentu, tambahkan _priority_=1 ke string kueri:

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

Data prioritas didorong ke subscriber sebelum data prioritas standar.

Lihat detail antrian

Tambahkan _attrs_=true ke permintaan GET untuk mengambil metadata antrian:

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

Tanggapannya berupa objek 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"}

Bidang utama:

Bidang	Deskripsi
`stream.maxPayloadBytes`	Ukuran maksimum dalam byte untuk satu item data
`stream.approxMaxLength`	Jumlah maksimum item data yang dapat ditampung antrian
`stream.firstEntry`	Indeks item pertama dalam antrian
`stream.lastEntry`	Indeks item terakhir dalam antrian
`stream.length`	Jumlah item saat ini dalam antrian
`meta.state`	Status antrian saat ini

Atau, buka halaman Elastic Algorithm Service (EAS), klik nama layanan, lalu beralih ke tab Asynchronous Queue.

Tab tersebut menampilkan informasi dasar antrian (kelompok sumber daya tempat antrian berada, waktu pembuatan, ukuran maksimum permintaan input tunggal, dan ukuran maksimum respons output tunggal), sumber daya yang di-deploy untuk layanan (jumlah instans, CPU, dan memori), serta jumlah item data saat ini yang tersimpan di antrian input beserta status pemrosesan setiap instans.

Kueri data

Pilih metode pengambilan berdasarkan kasus penggunaan Anda:

Metode	Cara kerja	Kapan digunakan
Polling	Kueri data berdasarkan indeks atau ID permintaan sesuai permintaan	Mengambil item tertentu atau memeriksa status antrian sesekali
Subscription	Berlangganan melalui WebSocket untuk menerima data saat tiba	Pemrosesan aliran berkelanjutan atau skenario throughput tinggi

Poll data dari antrian input

Kueri data berdasarkan indeks atau ID permintaan:

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

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

Tanggapan:

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

Kode status tanggapan:

Kode status	Makna
`200 OK`	Data ditemukan dan dikembalikan
`204 No Content`	Tidak ada data yang cocok (dikembalikan segera saat `_timeout_=0`)

Parameter kueri:

Parameter	Tipe	Default	Deskripsi
`_index_`	INT	`0`	Indeks awal. Atur nilai ini sedekat mungkin dengan indeks item target untuk efisiensi kueri yang lebih baik.
`_length_`	INT	`1`	Jumlah item yang akan diambil.
`_auto_delete_`	BOOL	`true`	Hapus item yang dikueri dari antrian setelah diambil.
`_timeout_`	STRING	`0`	Berapa lama menunggu jika tidak ada data yang cocok. Nilai `0` langsung mengembalikan HTTP 204. Contoh nilai: `1s`, `1m`.
`requestId`	STRING	—	Tag bawaan yang digunakan untuk menemukan item tertentu. Framework EAS memberi tag setiap item input dengan `requestId` dan membawanya hingga ke antrian output, sehingga Anda dapat menggunakan ID permintaan yang sama untuk mengkueri hasil inferensi dari antrian output.

Poll hasil inferensi dari antrian output

Kueri hasil dari antrian output menggunakan ID permintaan dari pengiriman awal:

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

Tanggapan:

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

Hapus data

Hapus data dari antrian dengan dua cara: hapus satu item atau truncate semua item hingga indeks tertentu.

Hapus satu item

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

Tanggapan:

> 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

Parameter:

Parameter	Tipe	Deskripsi
`_index_`	INT	Indeks item yang akan dihapus

Truncate antrian

Hapus semua item dengan indeks lebih rendah dari nilai yang ditentukan:

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

Tanggapan:

> 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

Parameter:

Parameter	Tipe	Deskripsi
`_index_`	INT	Indeks batas. Semua item dengan indeks lebih rendah dari nilai ini akan dihapus.
`_trunc_`	BOOL	Harus bernilai `true` untuk melakukan truncation. Jika dihilangkan atau bernilai `false`, permintaan akan melakukan penghapusan satu item saja.

Berlangganan ke antrian (SDK Python)

Untuk pemrosesan aliran berkelanjutan, berlanggananlah ke antrian output menggunakan SDK Python. Layanan antrian menggunakan protokol WebSocket untuk mempertahankan koneksi persisten dan mendorong data ke subscriber saat tiba. Ukuran jendela langganan dikontrol oleh pengaturan worker_threads pada instans layanan inferensi.

Layanan inferensi tidak wajib. Gunakan SDK untuk berlangganan ke antrian input dalam layanan kustom dan tulis hasilnya ke antrian pesan pihak ketiga atau penyimpanan seperti Object Storage Service (OSS).

Instal SDK:

pip install eas_prediction --user

Berlangganan ke antrian output:

Contoh berikut menggunakan QueueClient untuk mengirim 10 item ke antrian input dan berlangganan hasil dari antrian output. Dalam produksi, gunakan thread terpisah untuk pengiriman dan langganan.

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

# Buat klien untuk antrian input.
input_queue = QueueClient('182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'qservice')
# Untuk mengatur user atau group kustom:
# 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()

# Buat klien untuk antrian output.
sink_queue = QueueClient('182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'qservice/sink')
sink_queue.set_token('YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MT****==')
sink_queue.init()

# Kirim 10 item ke antrian input.
for x in range(10):
    index, request_id = input_queue.put('[{}]')
    print(index, request_id)

    # Cetak atribut antrian.
    attrs = input_queue.attributes()
    print(attrs)

# Berlangganan ke antrian output dengan ukuran jendela 5.
i = 0
watcher = sink_queue.watch(0, 5, auto_commit=False)
for x in watcher.run():
    print(x.data.decode('utf-8'))

    # Commit setiap item setelah diproses.
    sink_queue.commit(x.index)
    i += 1
    if i == 10:
        break

# Tutup watcher. Setiap instans QueueClient hanya mendukung satu watcher aktif.
# Gagal menutup watcher menyebabkan error pada pemanggilan watch() berikutnya.
watcher.close()

Langkah berikutnya

Langganan dan dorong layanan antrian — pelajari tentang kelompok konsumen, pengguna, dan semantik commit.
Download and authenticate the client — siapkan dan perbarui eascmd.