ApsaraMQ for RocketMQ:Modelo de tópico Lite

Nome do tópico lite

• Definição: O nome que identifica um tópico lite. Os nomes dos tópicos lite são globalmente únicos dentro de seu tópico pai.

• Valor: Quando o tipo de tópico é Lite e você chama setLiteTopic em uma mensagem, o sistema cria automaticamente o tópico lite caso ele ainda não exista.

• Restrição: Consulte Parameter limits.

Tempo de vida (TTL)

• Definição: O tempo de expiração de um tópico lite. Se nenhuma mensagem nova for gravada no tópico lite por um período superior ao seu TTL, o sistema o exclui automaticamente. A exclusão significa liberar a contagem alocada ao tópico lite (contagem total menos um).

• Valor: Ao criar um tópico do tipo Lite, você pode definir o parâmetro de expiração.

• Restrição: Consulte Parameter limits.

Caso de uso 1: Comunicação assíncrona para sistemas Multi-Agent visando resolver bloqueios em chamadas de longa duração

À medida que os cenários de IA se tornam mais complexos, os sistemas de agente único enfrentam limitações: falta de especialização, dificuldade em integrar múltiplos domínios e incapacidade de habilitar tomada de decisão colaborativa dinâmica. Aplicações e fluxos de trabalho de agente único estão migrando para arquiteturas Multi-Agent. No entanto, como as tarefas de IA geralmente levam muito tempo, chamadas síncronas bloqueiam a thread do chamador, limitando a escalabilidade para colaboração em grande escala.

Conforme ilustrado acima, o fluxo de trabalho Multi-Agent funciona da seguinte forma: O Agente Supervisor divide uma solicitação em duas subtarefas para dois agentes filhos. Cada agente filho resolve sua parte e retorna os resultados ao Agente Supervisor, que os agrega e envia a resposta final ao cliente web. Utilizando o RocketMQ para comunicação assíncrona:

1. Fluxo de tratamento de solicitações:

   1. Crie um tópico (Request) para cada agente filho como uma fila de buffer de tarefas. Use um tópico prioritário para processar primeiro as tarefas de alta prioridade.

   2. O Agente Supervisor envia os detalhes das tarefas divididas para o tópico de solicitação correspondente.

2. Fluxo de tratamento de respostas:

   1. O Agente Supervisor cria um tópico do tipo Lite (Response) e o assina.

   2. Cada agente filho envia o resultado de sua tarefa para um LiteTopic sob o tópico Response. Nomeie cada LiteTopic usando o ID da tarefa para fornecer a cada tarefa seu próprio LiteTopic dedicado.

   3. O Agente Supervisor recebe resultados em tempo real por meio da assinatura e os envia para o cliente web usando HTTP SSE.

Caso de uso 2: Gerenciamento distribuído de estado de sessão para resolver problemas de continuidade de sessão em aplicações de IA

As interações em aplicações de IA são singulares: de longa duração, com múltiplas turnos e altamente dependentes de recursos computacionais custosos por sessão. Quando as aplicações dependem de conexões persistentes, como SSE, qualquer desconexão (devido a reinicializações de gateway, tempos limite ou instabilidade de rede) causa perda do contexto da sessão atual e desperdiça recursos computacionais de IA já investidos.

Assim como no fluxo de resposta do caso de uso 1, utilize um tópico do tipo Lite para notificações de resultados em tempo real. Nomeie cada LiteTopic usando o SessionID (por exemplo, chatbot/{sessionID}). Todos os resultados da sessão são entregues como mensagens ordenadas neste tópico. Para manter a continuidade da sessão após uma reconexão:

1. O cliente web estabelece uma conexão persistente com o nó 1 do servidor de aplicações e inicia a sessão Session2.

2. O nó 1 do servidor de aplicações assina o LiteTopic [chat/SessionID2].

3. O agendador de tarefas do Large Language Model (LLM) envia resultados para o LiteTopic [chat/SessionID2] com base no SessionID da solicitação.

4. Devido a um problema de rede, o WebSocket reconecta-se ao nó 2 do servidor de aplicações.

5. O nó 1 do servidor de aplicações cancela a assinatura do LiteTopic [chat/SessionID2]. O nó 2 do servidor de aplicações passa a assiná-lo.

6. O LiteTopic [chat/SessionID2] retoma a entrega a partir do último offset consumido, garantindo a continuidade do estado e dos dados da sessão.

Enviar mensagens

Producer producer = provider.newProducerBuilder()
    .setTopics(topic)
    .setClientConfiguration(clientConfiguration)
    .build();
final Message message = provider.newMessageBuilder()
    .setTopic(topic)
    // Set a message key for precise lookup by keyword.
    .setKeys("messageKey")
    // Set LiteTopic
    .setLiteTopic("lite-topic-1")
    // Message body
    .setBody("messageBody".getBytes())
    .build();
try {
    final SendReceipt sendReceipt = producer.send(message);
    log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
} catch (LiteTopicQuotaExceededException e) {
    // LiteTopic quota exceeded. Evaluate and increase quota.
    log.error("Lite topic quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to send message", t);
}

Consumir mensagens

Utilize a classe LitePushConsumer:

// Initialize LitePushConsumer with consumer group, target topic, and communication parameters.
LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
    .setClientConfiguration(clientConfiguration)
    // Topic bound to the ConsumerGroup when created in the console
    .bindTopic(topicName)
    // Set consumer group
    .setConsumerGroup(consumerGroup)
    .setMessageListener(messageView -> {
        // Process message and return consumption result.
        LOGGER.info("Consume message={}", messageView);
        return ConsumeResult.SUCCESS;
    })
    .build();
try {
    // Subscribe to desired LiteTopics
    litePushConsumer.subscribeLite("lite-topic-1");
    litePushConsumer.subscribeLite("lite-topic-2");
    litePushConsumer.subscribeLite("lite-topic-3");
} catch (LiteSubscriptionQuotaExceededException e) {
    // LiteTopic subscription quota exceeded. Evaluate and increase quota.
    log.error("Lite subscription quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to subscribe lite topic", t);
}
// After business processing, unsubscribe from unused LiteTopics promptly
litePushConsumer.unsubscribeLite("lite-topic-3");
// Get current set of subscribed LiteTopics
Set<String> liteTopicSet = litePushConsumer.getLiteTopicSet();

Atualizar assinaturas dinamicamente

/**
 * Dynamically add a subscription.
 * The subscribeLite() method makes a network call and validates quotas,
 * so it might fail.
 * Always check the result to confirm successful subscription.
 * Possible failure scenarios:
 * 1. Network error – retry the call.
 * 2. Quota validation fails – throws LiteSubscriptionQuotaExceededException.
 * Evaluate whether your quota meets requirements, and promptly call
 * unsubscribeLite() to release resources for unused topics.
 */
litePushConsumer.subscribeLite("lite-topic-1");
// Dynamically remove a subscription
litePushConsumer.unsubscribeLite("lite-topic-1");

Instâncias Serverless

Instâncias não Serverless (assinatura e pagamento conforme o uso)