Intégrez Plusieurs LLMs avec KubeMQ pour Améliorer Votre Flux de Travail

Intégrer plusieurs modèles de langage, tels que OpenAI et Claude de l'Anthropic, dans des applications présente des défis liés à la communication entre API et à la gestion efficace des demandes. La mise en place d'un courtier de messages est une solution qui simplifie cette interactivité.

Ce document explore l'utilisation de KubeMQ comme un routeur pour connecter et gérer divers LLMs, offrant des avantages tels qu'une intégration simplifiée et un support pour plusieurs modèles de langage.

Avantages de l'Utilisation d'un Courtier de Messages

1. Intégration Simplifiée

Un courtier de messages facilite la communication avec plusieurs API de LLM, réduisant la complexité du code et minimisant les erreurs.

2. Cas d'Usage Multiples

Il permet d'utiliser différents modèles selon les besoins, comme un pour des résumés et un autre pour l'analyse des sentiments.

3. Traitement par Lots et Inférence à Grande Échelle

Le courtier permet également le traitement asynchrone, garantissant que les demandes soient traitées même lors de pics de demande.

4. Redondance et Garantie de Continuité

Dans des situations critiques, un courtier de messages fournit une continuité du service, basculant vers des environnements de secours en cas de défaillances de connexion.

5. Gestion des Applications à Fort Trafic

Il distribue les demandes entre plusieurs instances de LLM, évitant la surcharge et assurant un fonctionnement fluide pour des applications à fort trafic.

Construire un Routeur de LLM avec KubeMQ

Pour configurer un routeur fonctionnant avec OpenAI et Claude, nous devons suivre quelques étapes fondamentales impliquant la configuration de KubeMQ.

Pré-Requis

Avant de commencer, assurez-vous d'avoir :

• Python 3.7 ou supérieur.

• Docker installé.

• Clés API valides pour OpenAI et Anthropic.

• Jeton KubeMQ, disponible sur le site de KubeMQ.

Configurer KubeMQ

Tout d'abord, il est essentiel de s'assurer que KubeMQ fonctionne via Docker :

docker run -d --rm \
  -p 8080:8080 \
  -p 50000:50000 \
  -p 9090:9090 \
  -e KUBEMQ_TOKEN="votre_token" \
  kubemq/kubemq-community:latest

Les ports de KubeMQ sont :

• 8080 : Accès à l'API REST de KubeMQ.

• 50000 : Connexion gRPC entre le client et le serveur.

• 9090 : Passerelle REST de KubeMQ.

Remplacez "votre_token" par le jeton réel de KubeMQ.

Créer le Serveur du Routeur de LLM

Le routeur agit comme un intermédiaire entre les clients et les LLMs, écoutant des canaux spécifiques et routant les requêtes au modèle approprié.

Exemple de Code : server.py

import time
from kubemq.cq import Client, QueryMessageReceived, QueryResponseMessage, QueriesSubscription, CancellationToken
from langchain.chat_models import ChatOpenAI
from langchain.llms import Anthropic
import os
from dotenv import load_dotenv
import threading

load_dotenv()

class LLMRouter:
    def __init__(self):
        self.openai_llm = ChatOpenAI(
            api_key=os.getenv("OPENAI_API_KEY"),
            model_name="gpt-3.5-turbo"
        )
        self.claude_llm = Anthropic(
            api_key=os.getenv("ANTHROPIC_API_KEY"),
            model="claude-3"
        )
        self.client = Client(address="localhost:50000")

    def handle_openai_query(self, request: QueryMessageReceived):
        try:
            message = request.body.decode('utf-8')
            result = self.openai_llm(message)
            resp
                query_received=request,
                is_executed=True,
                body=result.encode('utf-8')
            )
            self.client.send_response_message(response)
        except Exception as e:
            self.client.send_response_message(QueryResponseMessage(
                query_received=request,
                is_executed=False,
                error=str(e)
            ))

    def handle_claude_query(self, request: QueryMessageReceived):
        try:
            message = request.body.decode('utf-8')
            result = self.claude_llm(message)
            resp
                query_received=request,
                is_executed=True,
                body=result.encode('utf-8')
            )
            self.client.send_response_message(response)
        except Exception as e:
            self.client.send_response_message(QueryResponseMessage(
                query_received=request,
                is_executed=False,
                error=str(e)
            ))

    def run(self):
        def on_error(err: str):
            print(f"Erreur : {err}")

        def subscribe_openai():
            self.client.subscribe_to_queries(
                subscription=QueriesSubscription(
                    channel="openai_requests",
                    
                    
                ),
                cancel=CancellationToken()
            )

        def subscribe_claude():
            self.client.subscribe_to_queries(
                subscription=QueriesSubscription(
                    channel="claude_requests",
                    
                    
                ),
                cancel=CancellationToken()
            )

        threading.Thread(target=subscribe_openai).start()
        threading.Thread(target=subscribe_claude).start()

        print("Routeur LLM en cours d'exécution sur les canaux : openai_requests, claude_requests")
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            print("Arrêt en cours...")

if __name__ == "__main__":
    router = LLMRouter()
    router.run()

Explication du Code

• Initialisation :

  • Charge des variables d'environnement pour les clés API.
  • Initialise des clients pour les LLMs d'OpenAI et d'Anthropic.
  • Configure un client KubeMQ.

• Gestion des Requêtes :

  • Les méthodes handle_openai_query et handle_claude_query traitent les messages reçus et retournent les réponses appropriées.
  • Les erreurs sont traitées correctement, en informant le client.

• Inscription :

  • Le routeur s'inscrit sur deux canaux : openai_requests et claude_requests.
  • Utilise des threads pour gérer les abonnements de manière concurrente.

• Exécution du Serveur :

  • Le méthode run démarre l'écoute des abonnements et maintient le serveur actif jusqu'à ce qu'il soit interrompu.

Développer le Client LLM

Le client est responsable de l'envoi de requêtes au routeur, choisissant quel modèle utiliser.

Exemple de Code : client.py

from kubemq.cq import Client, QueryMessage
import json

class LLMClient:
    def __init__(self, address="localhost:50000"):
        self.client = Client(address=address)

    def send_message(self, message: str, model: str) -> dict:
        channel = f"{model}_requests"
        resp
            channel=channel,
            body=message.encode('utf-8'),
            timeout_in_sec
        ))
        if response.is_error:
            return {"error": response.error}
        else:
            return {"response": response.body.decode('utf-8')}

if __name__ == "__main__":
    client = LLMClient()
    models = ["openai", "claude"]
    message = input("Entrez votre message : ")
    model = input(f"Choisissez un modèle ({', '.join(models)}): ")
    if model in models:
        resp model)
        if "error" in response:
            print(f"Erreur : {response['error']}")
        else:
            print(f"Réponse : {response['response']}")
    else:
        print("Modèle sélectionné invalide")

Explication du Code

• Initialisation :

  • Configure un client KubeMQ.

• Envoi de Messages :

  • La méthode send_message construit un canal approprié en fonction du modèle sélectionné.
  • Envoie une requête au routeur et attend la réponse, traitant les erreurs si nécessaire.

• Interaction avec l'Utilisateur :

  • Demande à l'utilisateur de saisir un message et de sélectionner un modèle.
  • Affiche la réponse reçue du LLM.

Envoyer et Recevoir via REST

Pour les services qui préfèrent la communication RESTful, KubeMQ propose des points de terminaison REST.

Envoyer une Requête via REST

Point de terminaison :

POST http://localhost:9090/send/request

En-têtes :

Content-Type: application/json

Corps :

{
   "RequestTypeData": 2,
   "ClientID": "LLMRouter-sender",
   "Channel": "openai_requests",
   "BodyString": "Quelle est la capitale de la France ?",
   "Timeout": 30000
}

Détails de la charge utile :

• RequestTypeData : Spécifie le type de requête (2 pour une requête).

• ClientID : Identifiant du client qui envoie la requête.

• Channel : Canal correspondant au modèle LLM (openai_requests ou claude_requests).

• BodyString : Message à envoyer au LLM.

• Timeout : Temps maximum pour attendre une réponse (en millisecondes).

Recevoir la Réponse

La réponse sera un objet JSON contenant la sortie du LLM ou un message d'erreur.

Conclusion

En utilisant un courtier de messages comme KubeMQ, il est possible de développer un routeur évolutif et efficace qui interagit avec divers LLMs. Cette structure permet aux clients d'envoyer des requêtes à différents modèles de manière continue et peut être étendue pour inclure des modèles ou des fonctionnalités supplémentaires à l'avenir.

Les avantages de cette approche comprennent :

1. Intégration Simplifiée: Facilite l'interaction avec différentes API de LLM.

2. Support Multimodal: Routage efficace des demandes vers le modèle approprié.

3. Fiabilité: Assure qu'aucune information n'est perdue lors d'une utilisation intensive.

4. Redondance: Fournit des mécanismes pour maintenir des opérations continues.

5. Scalabilité: Gère de gros volumes de trafic avec efficacité.

Cette approche représente une avancée significative dans l'application de courtiers de messages pour optimiser le développement de solutions en intelligence artificielle.