Créez un Bot Signal en Python

Contexte

Le monde des messageries instantanées est en constante évolution. Les plateformes comme Signal ou WhatsApp offrent des fonctionnalités de plus en plus avancées pour personnaliser votre expérience utilisateur. Mais qu'est-ce que l'on peut faire pour aller encore plus loin ? En créant un bot qui interagit avec ces plateformes, vous pouvez automatiser des tâches et même créer des applications plus complexes !

Dans cet article, nous allons explorer comment créer un bot Python qui lit les messages reçus sur Signal et répond à ceux-ci en utilisant l'API d'OpenWebUI.

Prérequis

Pour pouvoir exécuter le script, il est nécessaire de mettre en place certains prérequis.

Un LLM
- Lien vers le guide d’installation Ollama
Un téléphone avec l’application Signal et un numéro de téléphone pour les interactions à l’API SIgnal
Docker pour l’installation du container Dockerized Signal Messenger REST API
Un bucket AWS S3

Pourquoi un bucket S3 ?

L’idée ici est de donner une “conscience” à notre bot de conversations qui va lui permettre de se souvenir de l’historique de chaque conversation pour une meilleure pertinence dans ses réponses. Cet historique sera stocké sur un bucket S3 AWS, libre à vous d’utiliser d’autres solutions open-source comme Minio.

Les conversations seront stockées dans un fichier JSON sur un bucket S3 et notre bot ira récupérer son contenu à chaque démarrage.

Pour l’interaction avec l’API Signal, je me suis appuyé sur Dockerized Signal Messenger REST API. Se référer au projet directement pour l’installation du container et l’enregistrement de ce dernier à l’API Signal.

Le code

Le code est composé de plusieurs modules :

generic_libs : Contient des fonctions génériques pour interagir avec les APIs de Signal et OpenWebUI.
- signal_lib : Permet d'interagir avec l'API de Signal.
- openwebui_lib : Permet d'interagir avec l'API d'OpenWebUI.
- s3_lib : Permet d'accéder à un bucket S3 pour stocker la conversation.
Le script principal : Le code qui permet de démarrer le bot et gérer les messages reçus sur Signal.

Les librairies utilisées

Les librairies suivantes sont nécessaires pour fonctionner correctement :

boto3 : Permet d’intéragir avec les contenus des buckets S3.

Les variables nécessaires

Pour fonctionner correctement, notre script Python nécessite certaines variables d’environnement à définir. Voici la liste des variables nécessaires :

SIGNAL_PHONE_NUMBER : Le numéro de téléphone associé à votre compte Signal.
AWS_ACCESS_KEY_ID : L'ID clé d'accès à votre compte AWS (pour l'utilisation du stockage S3).
AWS_SECRET_ACCESS_KEY : La clé secrète d'accès à votre compte AWS.
OPENWEBUI_API_KEY : Le code d'API pour interagir avec l'interface utilisateur d'OpenWebUI.

Ces variables sont renseignées dans le dictionnaire CONFIG du fichier main.py

CONFIG = {
    'phone_number': os.environ['SIGNAL_PHONE_NUMBER'],
    's3_access_key': os.environ['AWS_ACCESS_KEY_ID'],
    's3_secret_key': os.environ['AWS_SECRET_ACCESS_KEY'],
    'openwebui_api_key': os.environ['OPENWEBUI_API_KEY'],
    'signal_url': 'http://127.0.0.1:8081',
    'openwebui_url': 'http://127.0.0.1:8080',
    's3_bucket': 'signal.bot',
    's3_key_prefix': 'conversation_history.json',
    'conversation_history_file': '/tmp/conversation_history.json',
    'pause_duration': 5 # seconds
}

Le fonctionnement

Lorsque le script principal est exécuté, il vérifie continuellement les messages reçus sur Signal en utilisant l'API de Signal.

# signal_lib.py
def fetch_signal_messages(url) -> (tuple[str, str]):
    """
    Fetch signal messages from the specified URL.

    :param url: Signal message URL.
    :return: Tuple containing the source number and message text.
    """
    source_number=None
    message=None

    try :
        response=requests.get(url)
        data=json.loads(response.text)
        # print(f'Requête reçue: {data}')
        for element in data:
            if 'dataMessage' in element['envelope']:
                if 'groupInfo' in element['envelope']['dataMessage']:
                    source_number=element['envelope']['dataMessage']['groupInfo'].get('groupId')
                else:
                    source_number=element['envelope'].get('sourceNumber')
                message=element['envelope']['dataMessage'].get('message')
                logging.info(f'Source Number: {source_number}')
                logging.info(f'Message: {message}')
        return source_number, message
    except requests.exceptions.RequestException as e:
        # Handle any exceptions that occur during request
        print(f"Error fetching signal messages: {e}")
        return None, None

Lorsqu'il trouve un nouveau message, il traite celui-ci et l’envoie à l'API d'OpenWebUI pour générer une réponse grâce au modèle LLM chargé.

# openwebui_lib.py
def send_message_to_openwebui(url, source_number, message, conversation_history, openwebui_api_key):
    """
    Send a message to OpenWebUI.

    :param url: OpenWebUI API URL.
    :param source_number: Source number of the message.
    :param message: Message text to send.
    :param conversation_history: Conversation history dictionary.
    :param openwebui_api_key: OpenWebUI API key.
    :return: The response from OpenWebUI.
    """

    # Create a new conversation history entry for the source number if not exists
    if not source_number in conversation_history:
        conversation_history[source_number] = []

    try:
        # Create a new conversation history entry for the source number
        conversation_history[source_number].append({'role':'user','content':message})

        # Define the API request data and headers
        donnees = {
            'model': 'signalbot',
            'messages': conversation_history[source_number]
        }
        en_tetes = {
            'Content-Type': 'application/json',
            'Authorization': f'Bearer {openwebui_api_key}'
        }

        # Send a POST request to the OpenWebUI API
        response = requests.post(url, json=donnees, headers=en_tetes)

        # Parse the JSON response and extract the message text
        data = json.loads(response.text)
        msg = data['choices'][0]['message']['content']

        # Update the conversation history with the new message
        conversation_history[source_number].append({'role':'assistant','content':msg})

        return msg

    except requests.exceptions.RequestException as e:
        # Handle any exceptions that occur during request
        print(f"Error sending message to OpenWebUI: {e}")
        return None

Le modèle LLM que doit utiliser l’API OpenWebUI est renseigné dans le fichier generic_libs/openwebui_lib.py.

         # Define the API request data and headers
 25         donnees = {
 26             'model': 'signalbot',
 27             'messages': conversation_history[source_number]
 28         }
 29         en_tetes = {
 30             'Content-Type': 'application/json',
 31             'Authorization': f'Bearer {openwebui_api_key}'
 32         }

Le modèle utilisé ici se nomme signalbot, c’est un modèle créé à partir du modèle llama3.1 mais avec un prompt système personnalisé.

Libre à vous de créer et gérer vos propres prompts pour une personnalisation des réponses du bot.
Reportez-vous à cet article pour opérer ce type de changements.

La réponse générée sera renvoyée à l’API Signal pour renvoi vers l’utilisateur ou groupe Signal correspondant.

# signal_lib.py
def send_message_to_signal_cli(url, phone_number, recipient, msg) -> None:
    """
    Send a message to Signal CLI.

    :param url: Signal CLI API URL.
    :param phone_number: Phone number of the sender.
    :param recipient: Recipient number of the message.
    :param msg: Message text to send.
    :return: None.
    """

    # Define the API request data and headers
    donnees = {
        'message': f'{msg}',
        'number': f'{phone_number}',
        'recipients': [f'{recipient}']
    }
    en_tetes = {
        'Content-Type': 'application/json'
    }

    # Send a POST request to the Signal CLI API
    try:
        requests.post(url, json=donnees, headers=en_tetes)

    except requests.exceptions.RequestException as e:
        # Handle any exceptions that occur during request
        print(f"Error sending message to Signal CLI: {e}")

Au démarrage du bot, le fichier conversation_history.json stocké sur le bucket S3 est téléchargé localement à l’emplacement défini par la variable conversation_history_file.

# s3_lib.py
def download_file(bucket, object_name, file_name) -> bool:
    """
    Download a file from an S3 bucket.

    :param object_name: File to download.
    :param bucket: Bucket to download from.
    :param file_name: Local path where the file should be saved.
    :return: True if file was downloaded, else False.
    """
    # Use boto3 library to interact with AWS S3
    s3 = boto3.client('s3')

    try:
        # Upload the file to the specified bucket and object name
        s3.download_file(bucket, object_name, Filename=file_name)

        # If successful, return True
        return True

    except Exception as e:
        # Handle any exceptions that occur during upload
        print(f"Error downloading file: {e}")
        return False

Un fichier conversation_history.json vierge est créé au démarrage du bot s’il n’existe pas (le fichier n’a pas été téléchargé depuis S3).

# main.py
def load_conversation_history() -> (dict):
    """
    Load the conversation history.

    :return: A dictionary containing the conversation history.
    """

    try:
        with open(CONFIG['conversation_history_file'], 'r') as file:
            return json.load(file)
    except FileNotFoundError:
        # Le fichier n'existe pas, créez-le avec un objet JSON vide
        with open(CONFIG['conversation_history_file'], 'w') as file:
            file.write(json.dumps({}))
        return {}

Il sera complété à chaque interaction pour permettre à notre bot de garder en mémoire l’historique des messages échangés pour chaque groupe/utilisateur.

C’est la fonction send_message_to_openwebui du fichier generic_libs/openwebui_lib.py qui est en charge d’alimenter le fichier d’historique.

# openwebui_lib.py
def send_message_to_openwebui(url, source_number, message, conversation_history,openwebui_api_key):
    """
    Send a message to OpenWebUI.

    :param url: OpenWebUI API URL.
    :param source_number: Source number of the message.
    :param message: Message text to send.
    :param conversation_history: Conversation history dictionary.
    :param openwebui_api_key: OpenWebUI API key.
    :return: The response from OpenWebUI.
    """

    # Create a new conversation history entry for the source number if not exists
    if not source_number in conversation_history:
        # conversation_history.update({source_number: []})
        conversation_history[source_number] = []

    try:
        # Create a new conversation history entry for the source number
        conversation_history[source_number].append({'role':'user','content':message})
[...]
        # Update the conversation history with the new message
        conversation_history[source_number].append({'role':'assistant','content':msg})

Par défaut, ce fichier se trouve sur /tmp/conversation_history.json.

Lorsque le script est arrêté, ce fichier est envoyé vers le bucket S3 configuré pour sauvegarde.

# s3_lib.py
def upload_file(file_name, bucket, object_name) -> bool:
    """
    Upload a file to an S3 bucket.

    :param file_name: File to upload.
    :param bucket: Bucket to upload to.
    :param object_name: S3 object name.
    :return: True if file was uploaded, else False.
    """
    # Use boto3 library to interact with AWS S3
    s3 = boto3.client('s3')

    try:
        # Upload the file to the specified bucket and object name
        s3.upload_file(file_name, bucket, object_name)

        # If successful, return True
        return True

    except Exception as e:
        # Handle any exceptions that occur during upload
        print(f"Error uploading file: {e}")
        return False

Le code dans son intégralité se trouve sur le projet GitLab signal-conversational-bot.

Conclusion

Créer un bot conversationnel pour Signal en Python offre une opportunité passionnante d'automatiser des tâches et d'améliorer l'interaction avec les utilisateurs. En utilisant des outils comme l'API Dockerized Signal Messenger REST et OpenWebUI, vous pouvez concevoir un bot capable de gérer des conversations de manière intelligente et personnalisée. L'intégration d'un stockage S3 pour conserver l'historique des conversations permet d'améliorer la pertinence des réponses du bot. Ce projet démontre comment les développeurs peuvent tirer parti des technologies modernes pour créer des solutions innovantes et adaptées à leurs besoins spécifiques.

Automatisez vos Conversations : Créez un Bot Signal en Python !

Table of contents