Ubuntu TechHive
enhancing-discord-bots-with-llms-and-python.md
Enhancing Discord Bots with LLMs and Python
article.détail

Enhancing Discord Bots with LLMs and Python

C
ChiefKemist
9 min de lecture
Publié
reading.progression 9 min de lecture

Description de l'amélioration des bots Discord avec les LLM et Python

Fonctionnalités

  • Exécute les commandes utilisateur

  • Filtre le spam

  • Exécute les tâches en arrière-plan

Bibliothèques

flowchart TD
A [Application] -->|Sends input| B [Function Caller]
B -->|Queries| C [LLM]
C -->|Returns raw output| D [Response Handler]
D -->|Structures data using| E [Pydantic Models]
E --> F{Structured Output}
F -->|Used by| A

subgraph "large_language_models [Large Language Models]"
C
end

subgraph "response_handling [Response Handling]"
D
E
end

  • Application : Il s'agit de votre système ou service principal qui doit interagir avec le LLM.

  • Appeleur de fonction : Agit comme un intermédiaire qui envoie des entrées au LLM et reçoit la sortie brute. Ce composant encapsule la logique nécessaire pour communiquer avec le LLM.

  • LLM (Large Language Model) : Le modèle d'IA qui traite l'entrée et génère une sortie basée sur l'appel de fonction. Les exemples incluent GPT-4, Claude3, etc.

  • Gestionnaire de réponse (Response Handler) : Prend la sortie brute du LLM et commence le processus de structuration. Cela peut impliquer la vérification des erreurs, le filtrage et la préparation des données pour la conversion en un format structuré.

  • Modèles Pydantic (Pydantic Models) : Ceux-ci sont utilisés pour définir explicitement la structure des données de sortie. Les modèles Pydantic appliquent la vérification de type et la validation des données, ce qui aide à garantir que les données sont conformes à un schéma spécifié.

  • Sortie structurée (Structured Output) : La sortie finale qui est bien structurée et prête à être utilisée par l'application. Cette sortie est prévisible et plus facile à intégrer dans les processus ou systèmes en aval.

flowchart TD
A [Client Application] -->|Uses| B [Instructor Library]
B --> |Wraps| C [GPT-4]
B --> |Wraps| D [LLAMA3]
B --> |Wraps| E [Claude3]

C --> G((Pydantic Models))
D --> G
E --> G

G --> H{Structured Outputs}
H -->|Returned to| A

subgraph "large_language_models [Large Language Models]"
C
D
E
end

subgraph "pydantic_modelling [Output Structuring]"
G
end

  • Application Client : Il s'agit de votre application Python qui doit interagir avec les grands modèles linguistiques. Elle utilise la bibliothèque Instructor pour faciliter ces interactions.

  • Bibliothèque Instructor : Agit comme un middleware qui enveloppe les grands modèles linguistiques. Elle est responsable de l'envoi des requêtes à ces modèles et du traitement de leurs sorties.

  • Modèles de langage étendus : Inclut GPT-4, LLAMA3 et Claude3. Chacun de ces modèles peut générer des sorties de texte complexes basées sur l'entrée qu'il reçoit.

  • Modèles Pydantic : Ceux-ci sont utilisés au sein de la bibliothèque Instructor pour structurer la sortie brute des modèles de langage dans un format plus gérable et défini, rendant le développement d'applications plus propre et plus prévisible.

  • Sorties structurées : Les sorties structurées finales sont ensuite renvoyées à l'application cliente, où elles peuvent être utilisées ou affichées.

Implémentation

Implémentation de l'API

#!/usr/bin/env python3

from litestar import Litestar, get, post
from litestar.openapi import OpenAPIConfig
from litestar.openapi.plugins import (
    ScalarRenderPlugin,
    RapidocRenderPlugin,
    RedocRenderPlugin,
    SwaggerRenderPlugin,
)

from enhanced_discord_bot_llms.llm_svc import (
LLMModel,
gen_async_client,
UserInfo,
streaming_usine_de_gaou_creation,
)

@get("/", sync_to_thread=False)
def read_root() -> dict:
return {"Hello": "World"}

@post("/gaou/{parametre:str}")
async def creer_gaou(parametre: str) -> UserInfo:
model = LLMModel.LLAMA3
client = gen_async_client(model=model)
gaou = await streaming_usine_de_gaou_creation(client, parametre, model=model)
print(f"Nouveau gaou créé: {gaou},\n selon le paramètre {parametre}\n\n")
return gaou

app = Litestar(
route_handlers= [read_root, creer_gaou],
openapi_config=OpenAPIConfig(
title="Gaou API",
description="API pour créer des Gaous",
version="0.1.0",
path="/docs",
render_plugins= [
RapidocRenderPlugin(),
# RedocRenderPlugin(),
# ScalarRenderPlugin(),
# SwaggerRenderPlugin(),
],
),
debug=True,
)

if name == "main":
import uvicorn

application = "gaouapp:app"
uvicorn.run(application, host="0.0.0.0", port=8000, reload=True)


## Services de Bot

#!/usr/bin/env python3
import os

import instructor

from instructor import Instructor, AsyncInstructor
from anthropic import Anthropic, AsyncAnthropic
from groq import Groq, AsyncGroq
from openai import OpenAI, AsyncOpenAI
from pydantic import BaseModel, Field

from enum import Enum, auto

class LLMModel(str, Enum):
Claude3 = "claude-3-opus-20240229"
GPT4_Omni = "gpt-4o"
LLAMA3 = "llama3-70b-8192"

def gen_client(model=LLMModel.GPT4_Omni) -> Instructor:
match model:
case LLMModel.Claude3:
client = instructor.from_anthropic(Anthropic())
case LLMModel.GPT4_Omni:
client = instructor.patch(OpenAI())
case LLMModel.LLAMA3:
client = instructor.patch(Groq())
return client

def gen_async_client(model=LLMModel.GPT4_Omni) -> AsyncInstructor:
match model:
case LLMModel.Claude3:
client = instructor.from_anthropic(AsyncAnthropic())
case LLMModel.GPT4_Omni:
client = instructor.patch(AsyncOpenAI())
case LLMModel.LLAMA3:
client = instructor.patch(AsyncGroq())
return client

Domaine Gaou

class UserInfo(BaseModel):
name: str
age: int
is_teenager: bool
is_intelligent: bool

def usine_de_gaou_creation(
ai_client: Instructor, parametre: str, model=LLMModel.GPT4_Omni
) -> UserInfo:
gaou = ai_client.chat.completions.create(
model=model,
response_model=UserInfo,
messages= [{"role": "user", "content": parametre}],
)
return gaou

async def streaming_usine_de_gaou_creation(
ai_client: AsyncInstructor, parametre: str, model=LLMModel.GPT4_Omni
) -> UserInfo:
gaou = await ai_client.chat.completions.create(
model=model,
response_model=UserInfo,
messages= [
{
"role": "system",
"content": "L'utilisateur peut fournir une invite dans la langue de son choix (telle que l'anglais, le français, le créole, l'espagnol, etc.), veuillez donc en tenir compte.",
},
{"role": "user", "content": parametre},
],
)
return gaou

class Language(str, Enum):
nouchi = "Nouchi"
moore = "Mooré"
lingala = "Lingala"
english = "Anglais"
french = "Français"
creole = "Créole"
spanish = "Espagnol"

class GaouJoke(BaseModel):
friend_gaou_joke: str = Field(
...,
description="La blague qui qualifie l'ami de Gaou. La blague doit être légère et humoristique et alterner entre le Nouchi, le Mooré, le Lingala, l'anglais, le français, le créole et l'espagnol.",
)
language: Language

async def streaming_gaou_formula(
ai_client: AsyncInstructor, gaou_name: str, model=LLMModel.GPT4_Omni
) -> GaouJoke:
gaou = await ai_client.chat.completions.create(
model=model,
temperature=1, # Go wild with the temperature!!!!
max_tokens=1024,
response_model=GaouJoke,
messages= [
{
"role": "system",
"content": f"""
Le terme 'Gaou' est un terme amusant, utilisé uniquement entre amis. Par exemple, {gaou_name} est tellement Gaou !.
Vous aiderez à qualifier un ami de Gaou, en vous basant sur les critères suivants :
- Le nom de l'ami
- Inventez une blague légère qui finit toujours par qualifier l'ami de Gaou
- Mélangez de l'humour et du sarcasme

  • D'une certaine manière, Gaou désigne quelqu'un de naïf, crédule ou facilement trompé, mais de manière amicale.
    - Utilisez différentes langues parmi les suivantes : Mooré, anglais, français, créole, espagnol, etc.
    """,
    },
    {"role": "user", "content": gaou_name},
    ],
    )
    return gaou

## Implémentation du bot

#!/usr/bin/env python3

import os
import random

import discord

from asyncio import sleep

from discord.ext import commands, tasks

from enhanced_discord_bot_llms.constants import (
WORDS_THE_BOT_DONT_LIKE,
FROWNING_FACE_EMOJI,
)
from enhanced_discord_bot_llms.llm_svc import (
gen_client,
usine_de_gaou_creation,
gen_async_client,
streaming_usine_de_gaou_creation,
LLMModel,
streaming_gaou_formula,
)

intents = discord.Intents.default()
intents.typing = False
intents.messages = True
intents.message_content = True
intents.reactions = True
intents.members = True

bot = commands.Bot(command_prefix="?", intents=intents)

@bot.event
async def on_message(message):
# nous ne voulons pas que le bot se réponde à lui-même
if message.author == bot.user:
return


try:
        content = message.content.lower()
        for word in WORDS_THE_BOT_DONT_LIKE:
            if word in content:
                await sleep(10)
                await message.channel.send(
                    f"{message.author.mention} Hey! Do not use that word again {FROWNING_FACE_EMOJI}"
                )
                await message.channel.send(
                    f"{message.author.mention} You called me: {word} and I don't like it. I've deleted your message."
                )
                await sleep(10)
                await message.delete()
    except Exception as e:
        print(f"Error: {e}")

if message.content == "pingGG":
        await message.channel.send("pongGG")
        return

await bot.process_commands(message)

@bot.event
async def on_message_edit(before, after):
    if before.author == bot.user:
        return

if after.content == "ping":
        await after.channel.send("pong")
        return

await bot.process_commands(after)

@bot.command()
@commands.guild_only()
async def ping(ctx: commands.Context):
    """
    ctx: Context (discord.ext.commands.Context, informations sur la commande)

?ping
    """
    await ctx.reply("pong")

@bot.command()
@commands.guild_only()
async def new_gaou(ctx: commands.Context, parametre: str):
    """
    ctx: Context (discord.ext.commands.Context, informations sur la commande)
    parametre: str (message à envoyer au modèle)

?new_gaou "Je ne suis pas un Gaou nommé Lambert qui a 15 ans et est intelligent."
    """
    model = LLMModel.GPT4_Omni
    # model = LLMModel.LLAMA3
    try:
        client = gen_async_client(model=model)
        gueou = await streaming_usine_de_gaou_creation(client, parametre, model=model)
        await ctx.reply(f"""
```json

{gueou.model_dump_json(
    indent=4
)}

""")
except Exception as e:
print(f"Error: {e}")
await ctx.reply(f"An error occurred: {e}")

@bot.command()
@commands.has_permissions(administrator=True)
@commands.bot_has_permissions(manage_messages=True)
async def cleanup(ctx: commands.Context, limit: int):
"""
ctx: Contexte (discord.ext.commands.Context, informations sur la commande)
limit: int (nombre de messages à supprimer)

?cleanup 10
"""
await delete_messages(ctx, limit)

@bot.command()
@commands.dm_only()
async def dm_cleanup(ctx: commands.Context, limit: int):
"""
ctx: Contexte (discord.ext.commands.Context, informations sur la commande)
limit: int (nombre de messages à supprimer)

?dm_cleanup 10
"""
await delete_messages(ctx, limit)

async def delete_messages(ctx: commands.Context, limit: int):
print(f"Cleaning up: {limit} messages...")
async for msg in ctx.channel.history(limit=limit):
try:
print(f"Deleting message: {msg.content}")
await sleep(1)
await msg.delete()
except Exception as e:
print(f"Error: {e}")
await ctx.reply(f"You may not have permission to delete messages.")
continue

@tasks.loop(minutes=16)
async def my_background_gaou_tasks():
await bot.change_presence(activity=discord.Game(name="With Gaous"))
# members = [ [member for member in guild.members] for guild in bot.guilds]
# members = bot.get_all_members()
channels = bot.get_all_channels()
for chnl in channels:
if isinstance(chnl, discord.TextChannel) and chnl.name == "botexperiments":
await chnl.send(
f"Qui est Gaou de toute façon ? Moi Gaou ? Réfléchis encore... {chnl.mention}"
)
chnl_members = chnl.members
for chnl_m in chnl_members:
if chnl_m.bot:
continue
elif (
"african" in chnl_m.name.lower()
or "dog" in chnl_m.name.lower()
or "lle" in chnl_m.name.lower()
or "bru" in chnl_m.name.lower()

):
await sleep(8)
# model = random.choice(
# [model.value for model in LLMModel]
# ) # Choose a model at random
model = LLMModel.Claude3
client = gen_async_client(model=model)
gueou_joke = await streaming_gaou_formula(
client, chnl_m.display_name, model=model
)
# message_to_gueou = f"{gueou_joke.friend_gaou_joke} ({gueou_joke.language.name} => {gueou_joke.language.value}) {chnl_m.mention}"
message_to_gueou = f"{gueou_joke.friend_gaou_joke} ({gueou_joke.language.value}) {chnl_m.mention}"
await chnl.send(message_to_gueou)

@my_background_gaou_tasks.before_loop
async def before_gueou():
    await bot.wait_until_ready()
    print("Ready for Gaous!")

@bot.event
async def on_ready():
    print(f"Logged in as {bot.user} (ID: {bot.user.id})")
    my_background_gaou_tasks.start()

if __name__ == "__main__":
    token = os.environ ["DISCORD_BOT_TOKEN"]
    bot.run(token)

Déploiement

Conteneurs Docker

Dockeriser l'API Discord

FROM python:3.12.3-alpine3.19

COPY . .

RUN apk add --no-cache libffi-dev openssl-dev gcc musl-dev make

RUN pip install -r requirements.lock

WORKDIR /src/enhanced_discord_bot_llms

CMD ["python", "gaouapp.py"]

Dockeriser le Bot Discord

FROM python:3.12.3-alpine3.19

COPY . .

RUN apk add --no-cache libffi-dev openssl-dev gcc musl-dev make

RUN pip install -r requirements.lock

WORKDIR /src/enhanced_discord_bot_llms

CMD ["python", "gaoubot.py"]

Script d'aide

#!/usr/bin/env bash
set -x #echo on

BASEDIR=(dirname "0")
DOCKERDIR=$BASEDIR/docker
PLATFORM=linux/amd64
REGISTRY=ttl.sh

echo "BASEDIR: $BASEDIR"
echo "DOCKERDIR: $DOCKERDIR"

case "$1" in
  "dockerize:api")
    echo "Building Docker image for API..."
    docker buildx build --platform $PLATFORM -t $2 -f $DOCKERDIR/Dockerfile.api $BASEDIR
    ;;
  "dockerize:bot")
    echo "Building Docker image for Bot..."
    docker buildx build --platform $PLATFORM -t $2 -f $DOCKERDIR/Dockerfile.bot $BASEDIR
    ;;
  "docker:publish")
    echo "Publishing Docker image..."
    docker push $2
    ;;
  *)
    echo "Usage: $0 {dockerize:api|dockerize:bot|docker:publish}"
    exit 1
    ;;
esac

exit 0

Infrastructure


graph TB

DockerEngine(Moteur Docker)

DockerEngine -- Exécute --> DockerContainer
DockerEngine -- Construit --> DockerImage

DockerFile(Dockerfile : Recette pour les images) -- Définit --> DockerImage

DockerHub(Docker Hub : Dépôt public) -- Stocke et partage --> DockerImage

DockerContainer(Conteneur Docker : Petit paquet autonome et exécutable)
DockerImage(Image Docker : Plans pour les conteneurs) -- Crée --> DockerContainer

subgraph "Analogie : Construction"
DockerFile -- "Plan de l'architecte" --> DockerImage
DockerImage -- "Pièces de maison préfabriquées" --> DockerContainer
end


#### Installer Docker sur Ubuntu

Mettre à jour votre liste de paquets existante

sudo apt update

Installer quelques paquets prérequis qui permettent à apt d'utiliser des paquets via HTTPS

sudo apt install apt-transport-https ca-certificates curl software-properties-common

Ajouter la clé GPG du dépôt officiel Docker à votre système

curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg --dearmor -o /usr/share/keyrings/docker-archive-keyring.gpg

Ajouter le dépôt Docker aux sources APT

echo "deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/docker-archive-keyring.gpg] https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null

Mettre à jour votre liste de paquets existante pour que l'ajout soit reconnu

sudo apt update

Assurez-vous que vous êtes sur le point d'installer depuis le dépôt Docker plutôt que depuis le dépôt Ubuntu par défaut

apt-cache policy docker-ce

Installer Docker

sudo apt install docker-ce

Vérifier qu'il est en cours d'exécution

sudo systemctl status docker


#### Configurer Docker

Docker sans`sudo`

Ajouter votre nom d'utilisateur au groupe docker

sudo usermod -aG docker ${USER}


Appliquez la nouvelle appartenance au groupe, déconnectez-vous du serveur et reconnectez-vous (Facultatif ?)

su - ${USER}

groups


#### Gestion des variables d'environnement

Nous utiliserons [direnv](https://direnv.net/) et le [configurerons](https://direnv.net/docs/hook.html) pour bash à l'intérieur de notre machine virtuelle.

sudo apt install direnv

eval "$(direnv hook bash)"

Retour aux articles