ApsaraMQ for RocketMQ:Lite Topic モデル

LiteTopic 名

• 定義：Lite Topic の名前です。名前は親 Topic 内でグローバルに一意である必要があります。

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

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

生存時間 (TTL)

• 定義：Lite Topic の生存時間 (TTL) です。Lite Topic がその TTL より長い期間メッセージを受信しない場合、自動的に削除されます。Lite Topic を削除すると、そのクォータ数が解放されます。

• 値：Lite タイプの Topic を作成する際に、`expiration` 値を設定できます。

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

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

AI シナリオが複雑化するにつれて、ほとんどの単一エージェントは複雑な状況で限界に直面します。専門的な分業が欠けており、複数のドメインを統合するのが難しく、動的な協調的意思決定ができません。単一エージェントのアプリケーションとワークフローは、徐々にマルチエージェントアプリケーションへと移行しています。しかし、AI タスクは長時間実行されるため、同期呼び出しは呼び出し元のスレッドをブロックし、大規模なコラボレーションにおいてスケーラビリティの問題を引き起こす可能性があります。

上の図に示すように、マルチエージェントシステムのワークフローは次のとおりです：スーパーバイザーエージェントがリクエストを分割し、2 つのサブエージェントに割り当てます。2 つのサブエージェントはそれぞれのドメイン固有の問題を処理し、結果をスーパーバイザーエージェントに返します。スーパーバイザーエージェントは結果を集約し、Web クライアントに返します。ApsaraMQ for RocketMQ の非同期通信ソリューションは次のように機能します：

1. リクエスト受信フロー：

   1. 各サブエージェントに対してリクエスト Topic を作成し、タスクのバッファーキューとして機能させることができます。これは、優先度の高いタスクを先に処理するための優先度 Topic にすることもできます。

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

2. 応答フローでは:

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

   2. サブエージェントはタスクを処理し、応答結果を応答 Topic 内の Lite Topic に送信します。Lite Topic はタスク ID にちなんで命名でき、各タスク専用の Lite Topic を作成できます。

   3. スーパーバイザーエージェントは、リアルタイムで結果を取得するためにサブスクライブし、HTTP Server-Sent Events (SSE) プロトコルを使用して Web クライアントにプッシュします。

シナリオ 2: 分散セッション状態管理による AI アプリケーションにおけるセッション管理の課題解決

AI アプリケーションの対話パターンは独特です。高コストの計算能力に大きく依存する、長時間実行されるマルチターンのセッションを伴います。アプリケーションが SSE のような持続的接続に依存している場合、ゲートウェイの再起動、接続タイムアウト、ネットワークの不安定性などのいかなる中断も、現在のセッションコンテキストの喪失を引き起こす可能性があります。これにより、進行中の AI タスクも無効になり、貴重な計算能力が無駄になります。

上の図に示すように、応答結果フローはシナリオ 1 と同じです。Lite タイプの Topic が応答結果のリアルタイム通知に使用されます。ここでは、各 Lite Topic は SessionID を使用して命名できます (例：chatbot/{sessionID})。これにより、セッションの結果はこの Topic 内のメッセージとして順序通りに転送されます。以下のソリューションは、持続的接続が再確立された後にセッションの継続性を維持する方法を説明しています：

1. Web クライアント 2 は、アプリケーションサービスノード 1 との持続的接続を確立し、Session2 を作成します。

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

3. 大規模モデルタスクスケジューリングコンポーネントは、リクエスト内の SessionID 情報に基づいて、結果を Lite Topic [chat/SessionID2] に送信します。

4. ネットワークの問題やその他の例外により、WebSocket 接続は自動的にアプリケーションサービスノード 2 に再接続します。

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

6. Lite Topic [chat/SessionID2] は、以前の消費の進捗に基づいて、後続の未消費メッセージをアプリケーションサービスノード 2 にプッシュし続けます。これにより、セッションの状態とデータの継続性が保証されます。

メッセージの送信

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 を初期化します。コンシューマーグループ、ターゲット Topic、通信パラメーターなどをバインドする必要があります。
LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
    .setClientConfiguration(clientConfiguration)
    // コンソールでコンシューマーグループを作成したときにバインドされた Topic。
    .bindTopic(topicName)
    // コンシューマーグループを設定します。
    .setConsumerGroup(consumerGroup)
    .setMessageListener(messageView -> {
        // メッセージを処理し、消費結果を返します。
        LOGGER.info("Consume message={}", messageView);
        return ConsumeResult.SUCCESS;
    })
    .build();

try {
    // 関心のある Lite Topic のコレクションをサブスクライブします。
    litePushConsumer.subscribeLite("lite-topic-1");
    litePushConsumer.subscribeLite("lite-topic-2");
    litePushConsumer.subscribeLite("lite-topic-3");
} catch (LiteSubscriptionQuotaExceededException e) {
    // Lite Topic のサブスクリプションクォータを超えました。クォータを評価し、引き上げてください。
    log.error("Lite subscription quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to subscribe lite topic", t);
}

// ビジネス処理が完了したら、使用しなくなった Lite Topic のサブスクリプションを速やかに解除します。
litePushConsumer.unsubscribeLite("lite-topic-3");

// 現在サブスクライブしている Lite Topic のセットを取得します。
Set<String> liteTopicSet = litePushConsumer.getLiteTopicSet();

サブスクリプション関係の動的変更

/**
 * サブスクリプション関係を動的に追加します。
 * subscribeLite() メソッドはネットワークリクエストを開始し、クォータ検証を実行するため、呼び出しが失敗する可能性があります。
 * サブスクリプションが正常に追加されたことを確認するために、この呼び出しの結果を常に確認してください。
 * 考えられる失敗シナリオは次のとおりです：
 * 1. ネットワークリクエストエラーが発生します。操作をリトライできます。
 * 2. クォータ検証が失敗し、LiteSubscriptionQuotaExceededException がスローされます。
 * クォータがニーズを満たしているか評価し、使用しなくなった Topic のサブスクリプションを unsubscribeLite() を使用して速やかに解除し、リソースを解放してください。
 */

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

// サブスクリプション関係を動的に削除します。
litePushConsumer.unsubscribeLite("lite-topic-1");

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

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