ApsaraMQ for RocketMQ:Lite トピックモデル

Lite トピック名

• 定義：Lite トピックを識別する名前です。Lite トピック名は、その親トピック内でグローバルに一意です。

• 値：トピックタイプが Lite で、メッセージに対して setLiteTopic を呼び出すと、Lite トピックが存在しない場合にシステムが自動的に作成します。

• 制約：詳細については、「パラメーターの制限」をご参照ください。

有効期間 (TTL)

• 定義：Lite トピックの有効期限です。TTL を超える期間、Lite トピックに新しいメッセージが書き込まれない場合、システムは自動的にそれを削除します。削除とは、Lite トピックに割り当てられたカウントを解放すること (合計カウントから 1 を引く) を意味します。

• 値：Lite タイプのトピックを作成する際に、expiration パラメーターを設定できます。

• 制約：詳細については、「パラメーターの制限」をご参照ください。

ユースケース 1：マルチエージェントシステムの非同期通信による長時間実行される呼び出しのブロッキング解決

AI シナリオが複雑化するにつれて、単一エージェントシステムは、専門性の欠如、複数ドメインの統合の難しさ、動的な協調的意思決定の実現不可能性といった限界に直面しています。単一エージェントアプリケーションとワークフローは、マルチエージェントアーキテクチャへと移行しています。しかし、AI タスクはしばしば長時間かかるため、同期呼び出しは呼び出し元のスレッドをブロックし、大規模な協調作業のスケーラビリティを制限します。

上記のように、マルチエージェントワークフローは次のように機能します。スーパーバイザーエージェントがリクエストを 2 つの子エージェントのための 2 つのサブタスクに分割します。各子エージェントは自身の部分を解決し、結果をスーパーバイザーエージェントに返します。スーパーバイザーエージェントはそれらを集約し、最終的な応答を Web クライアントに送信します。RocketMQ を非同期通信に使用する場合：

1. リクエスト処理フロー：

   1. 各子エージェントのタスクバッファーキューとしてトピック (Request) を作成します。優先度の高いタスクを先に処理するために、優先度トピックを使用します。

   2. スーパーバイザーエージェントは、分割されたタスク詳細を対応するリクエストトピックに送信します。

2. レスポンス処理フロー：

   1. スーパーバイザーエージェントは Lite タイプのトピック (Response) を作成し、それをサブスクライブします。

   2. 各子エージェントは、タスク結果を Response トピックの下の LiteTopic に送信します。タスク ID を使用して各 LiteTopic に名前を付け、すべてのタスクに専用の LiteTopic を与えます。

   3. スーパーバイザーエージェントは、サブスクリプションを通じてリアルタイムで結果を受け取り、HTTP SSE を使用して Web クライアントにプッシュします。

ユースケース 2：分散セッション状態管理による AI アプリケーションのセッション継続性問題の解決

AI アプリケーションのインタラクションは、長時間実行、マルチターン、セッションごとに高価な計算リソースに大きく依存するという点で独特です。アプリケーションが SSE などの持続的接続に依存している場合、(ゲートウェイの再起動、タイムアウト、またはネットワークの不安定性による) 切断は、現在のセッションコンテキストの喪失と、すでに投資された AI 計算リソースの無駄を引き起こします。

ユースケース 1 のレスポンスフローと同様に、リアルタイムの結果通知に Lite タイプのトピックを使用します。SessionID を使用して各 LiteTopic に名前を付けます (例：chatbot/{sessionID})。すべてのセッション結果は、このトピック内で順序付きメッセージとして配信されます。再接続後にセッションの継続性を維持するには、次の手順を実行します。

1. Web クライアントは、アプリケーションサーバーノード 1 との持続的接続を確立し、セッション Session2 を開始します。

2. アプリケーションサーバーノード 1 は、LiteTopic [chat/SessionID2] をサブスクライブします。

3. 大規模言語モデル (LLM) タスクスケジューラは、リクエスト内の SessionID に基づいて、結果を LiteTopic [chat/SessionID2] に送信します。

4. ネットワークの問題により、WebSocket はアプリケーションサーバーノード 2 に再接続します。

5. アプリケーションサーバーノード 1 は、LiteTopic [chat/SessionID2] のサブスクリプションを解除します。アプリケーションサーバーノード 2 はそれをサブスクライブします。

6. LiteTopic [chat/SessionID2] は、最後に消費されたオフセットから配信を再開し、継続的なセッション状態とデータを保証します。

メッセージの送信

Producer producer = provider.newProducerBuilder()
    .setTopics(topic)
    .setClientConfiguration(clientConfiguration)
    .build();
final Message message = provider.newMessageBuilder()
    .setTopic(topic)
    // キーワードによる正確な検索のためにメッセージキーを設定します。
    .setKeys("messageKey")
    // LiteTopic を設定します。
    .setLiteTopic("lite-topic-1")
    // メッセージ本文
    .setBody("messageBody".getBytes())
    .build();
try {
    final SendReceipt sendReceipt = producer.send(message);
    log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
} catch (LiteTopicQuotaExceededException e) {
    // LiteTopic のクォータを超えました。クォータを評価し、引き上げてください。
    log.error("Lite topic quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to send message", t);
}

メッセージの消費

LitePushConsumer クラスを使用します。

// コンシューマーグループ、ターゲットトピック、通信パラメーターで LitePushConsumer を初期化します。
LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
    .setClientConfiguration(clientConfiguration)
    // コンソールで作成時に ConsumerGroup にバインドされたトピック
    .bindTopic(topicName)
    // コンシューマーグループを設定します。
    .setConsumerGroup(consumerGroup)
    .setMessageListener(messageView -> {
        // メッセージを処理し、消費結果を返します。
        LOGGER.info("Consume message={}", messageView);
        return ConsumeResult.SUCCESS;
    })
    .build();
try {
    // 目的の LiteTopic をサブスクライブします。
    litePushConsumer.subscribeLite("lite-topic-1");
    litePushConsumer.subscribeLite("lite-topic-2");
    litePushConsumer.subscribeLite("lite-topic-3");
} catch (LiteSubscriptionQuotaExceededException e) {
    // LiteTopic のサブスクリプションクォータを超えました。クォータを評価し、引き上げてください。
    log.error("Lite subscription quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to subscribe lite topic", t);
}
// ビジネス処理後、使用しなくなった LiteTopic は速やかにサブスクリプションを解除します。
litePushConsumer.unsubscribeLite("lite-topic-3");
// 現在サブスクライブしている LiteTopic のセットを取得します。
Set<String> liteTopicSet = litePushConsumer.getLiteTopicSet();

サブスクリプションの動的更新

/**
 * サブスクリプションを動的に追加します。
 * subscribeLite() メソッドはネットワーク呼び出しを行い、クォータを検証するため、
 * 失敗する可能性があります。
 * サブスクリプションが成功したことを確認するために、必ず結果を確認してください。
 * 考えられる失敗シナリオ：
 * 1. ネットワークエラー – 呼び出しをリトライします。
 * 2. クォータ検証の失敗 – LiteSubscriptionQuotaExceededException がスローされます。
 * ご利用のクォータが要件を満たしているか評価し、速やかに
 * unsubscribeLite() を呼び出して、使用していないトピックのリソースを解放してください。
 */
litePushConsumer.subscribeLite("lite-topic-1");
// サブスクリプションを動的に削除します。
litePushConsumer.unsubscribeLite("lite-topic-1");

サーバーレスインスタンス

非サーバーレスインスタンス (サブスクリプションおよび従量課金)