ApsaraMQ for RocketMQ：Model lite topic

Nama LiteTopic

• Definisi: Nama LiteTopic. Nama harus unik secara global dalam topik induknya.

• Nilai: Jika tipe topik adalah Lite dan Anda memanggil setLiteTopic untuk sebuah pesan, sistem akan secara otomatis membuat LiteTopic yang sesuai jika belum ada.

• Batasan: Untuk informasi selengkapnya, lihat Batasan parameter.

Waktu hidup (TTL)

• Definisi: Waktu hidup (TTL) LiteTopic. Jika LiteTopic tidak menerima pesan apa pun selama periode yang lebih lama dari TTL-nya, LiteTopic tersebut akan dihapus secara otomatis. Penghapusan LiteTopic melepaskan kuota yang digunakan.

• Nilai: Saat membuat topik bertipe Lite, Anda dapat mengatur nilai expiration.

• Batasan: Untuk informasi selengkapnya, lihat Batasan parameter.

Skenario 1: Komunikasi asinkron untuk sistem Multi-Agent guna mengatasi blocking akibat panggilan berdurasi panjang

Seiring meningkatnya kompleksitas skenario AI, agen tunggal menghadapi keterbatasan dalam situasi rumit. Mereka kurang memiliki pembagian tugas spesialis, kesulitan mengintegrasikan berbagai domain, dan tidak dapat membuat keputusan kolaboratif secara dinamis. Aplikasi dan alur kerja berbasis agen tunggal secara bertahap beralih ke aplikasi Multi-Agent. Namun, karena tugas AI bersifat long-running, panggilan sinkron dapat memblokir thread pemanggil, menyebabkan masalah skalabilitas dalam kolaborasi skala besar.

Seperti ditunjukkan pada gambar sebelumnya, alur kerja sistem Multi-Agent adalah sebagai berikut: Agen Supervisor membagi permintaan dan menugaskannya ke dua sub-agen. Kedua sub-agen tersebut menangani masalah spesifik domain masing-masing dan mengembalikan hasilnya ke Agen Supervisor. Agen Supervisor menggabungkan hasil tersebut dan mengembalikannya ke klien web. Solusi komunikasi asinkron ApsaraMQ for RocketMQ bekerja sebagai berikut:

1. Dalam alur penerimaan permintaan:

   1. Anda dapat membuat topik permintaan untuk setiap sub-agen sebagai antrian buffer untuk task. Ini bisa berupa topik prioritas untuk memproses task prioritas tinggi terlebih dahulu.

   2. Agen Supervisor mengirim informasi task yang telah dibagi ke topik permintaan yang sesuai.

2. Dalam alur respons:

   1. Agen Supervisor membuat topik respons bertipe Lite dan berlangganan ke topik tersebut.

   2. Sub-agen memproses task dan mengirim hasil respons ke LiteTopic dalam topik respons. LiteTopic dapat dinamai berdasarkan ID task untuk membuat LiteTopic khusus bagi setiap task.

   3. Agen Supervisor berlangganan untuk mengambil hasil secara real-time, lalu mendorongnya ke klien web menggunakan protokol HTTP Server-Sent Events (SSE).

Skenario 2: Manajemen status sesi terdistribusi untuk mengatasi tantangan manajemen sesi dalam aplikasi AI

Pola interaksi aplikasi AI bersifat unik. Aplikasi ini melibatkan sesi multi-turn berdurasi panjang yang sangat bergantung pada daya komputasi bernilai tinggi. Ketika aplikasi bergantung pada koneksi persisten, seperti SSE, gangguan apa pun (misalnya restart gerbang, timeout koneksi, atau ketidakstabilan jaringan) dapat menyebabkan hilangnya konteks sesi saat ini. Hal ini juga membatalkan task AI yang sedang berjalan, sehingga membuang daya komputasi yang berharga.

Seperti ditunjukkan pada gambar sebelumnya, alur respons hasilnya sama dengan Skenario 1. Topik bertipe Lite digunakan untuk notifikasi real-time hasil respons. Setiap LiteTopic dapat diberi nama berdasarkan SessionID (misalnya, chatbot/{sessionID}), sehingga hasil sesi dikirimkan secara berurutan sebagai pesan dalam topik tersebut. Solusi berikut menjelaskan cara mempertahankan kelangsungan sesi setelah koneksi persisten dipulihkan:

1. Klien web 2 membuat koneksi persisten dengan node layanan aplikasi 1 dan membuat Session2.

2. Node layanan aplikasi 1 berlangganan ke LiteTopic [chat/SessionID2].

3. Komponen penjadwalan task model besar mengirim hasil ke LiteTopic [chat/SessionID2] berdasarkan informasi SessionID dalam permintaan.

4. Karena masalah jaringan atau pengecualian lain, koneksi WebSocket secara otomatis terhubung ulang ke node layanan aplikasi 2.

5. Node layanan aplikasi 1 berhenti berlangganan dari LiteTopic [chat/SessionID2], dan node layanan aplikasi 2 berlangganan ke LiteTopic [chat/SessionID2].

6. LiteTopic [chat/SessionID2] terus mendorong pesan yang belum dikonsumsi berikutnya ke node layanan aplikasi 2 berdasarkan progres konsumsi sebelumnya. Hal ini menjamin kelangsungan status dan data sesi.

Kirim pesan

Producer producer = provider.newProducerBuilder()
    .setTopics(topic)
    .setClientConfiguration(clientConfiguration)
    .build();

final Message message = provider.newMessageBuilder()
    .setTopic(topic)
    // Setel kunci pesan. Anda dapat menggunakan kunci ini untuk menemukan pesan tertentu secara akurat.
    .setKeys("messageKey")
    // Setel LiteTopic.
    .setLiteTopic("lite-topic-1")
    // Isi pesan.
    .setBody("messageBody".getBytes())
    .build();

try {
    final SendReceipt sendReceipt = producer.send(message);
    log.info("Pesan berhasil dikirim, messageId={}", sendReceipt.getMessageId());
} catch (LiteTopicQuotaExceededException e) {
    // Kuota LiteTopic terlampaui. Evaluasi dan tingkatkan kuota.
    log.error("Kuota topik Lite terlampaui", e);
} catch (Throwable t) {
    log.error("Gagal mengirim pesan", t);
}

Konsumsi pesan

Anda perlu menggunakan kelas LitePushConsumer:

// Inisialisasi LitePushConsumer. Anda perlu mengikat kelompok konsumen, topik target, parameter komunikasi, dan lainnya.
LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
    .setClientConfiguration(clientConfiguration)
    // Topik yang diikat saat kelompok konsumen dibuat di konsol.
    .bindTopic(topicName)
    // Setel kelompok konsumen.
    .setConsumerGroup(consumerGroup)
    .setMessageListener(messageView -> {
        // Proses pesan dan kembalikan hasil konsumsi.
        LOGGER.info("Mengonsumsi pesan={}", messageView);
        return ConsumeResult.SUCCESS;
    })
    .build();

try {
    // Berlangganan ke kumpulan LiteTopics yang Anda minati.
    litePushConsumer.subscribeLite("lite-topic-1");
    litePushConsumer.subscribeLite("lite-topic-2");
    litePushConsumer.subscribeLite("lite-topic-3");
} catch (LiteSubscriptionQuotaExceededException e) {
    // Kuota langganan LiteTopic telah terlampaui. Evaluasi dan tingkatkan kuota.
    log.error("Kuota langganan lite terlampaui", e);
} catch (Throwable t) {
    log.error("Gagal berlangganan lite topic", t);
}

// Setelah pemrosesan bisnis selesai, segera berhenti berlangganan dari LiteTopics yang tidak lagi digunakan.
litePushConsumer.unsubscribeLite("lite-topic-3");

// Dapatkan himpunan LiteTopics yang sedang dilanggan.
Set<String> liteTopicSet = litePushConsumer.getLiteTopicSet();

Ubah hubungan langganan secara dinamis

/**
 * Tambahkan hubungan langganan secara dinamis.
 * Metode subscribeLite() memulai permintaan jaringan dan melakukan verifikasi kuota, sehingga panggilan ini dapat gagal.
 * Selalu periksa hasil panggilan ini untuk memastikan langganan berhasil ditambahkan.
 * Kemungkinan skenario kegagalan meliputi:
 * 1. Terjadi kesalahan permintaan jaringan. Anda dapat mencoba ulang operasi tersebut.
 * 2. Verifikasi kuota gagal, dan LiteSubscriptionQuotaExceededException dilemparkan.
 * Evaluasi apakah kuota memenuhi kebutuhan Anda dan segera gunakan unsubscribeLite() untuk berhenti berlangganan dari topik yang tidak lagi digunakan guna melepaskan resource.
 */

litePushConsumer.subscribeLite("lite-topic-1");

// Hapus hubungan langganan secara dinamis.
litePushConsumer.unsubscribeLite("lite-topic-1");

Instans Serverless

Instans non-Serverless (subscription dan pay-as-you-go)