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

Enhancing Discord Bots with LLMs and Python

C
ChiefKemist
9 min de lectura
Publicado
reading.progreso 9 min de lectura

Descripción de la mejora de bots de Discord con LLMs y Python

Características

  • Ejecuta comandos de usuario

  • Filtra el spam

  • Ejecuta tareas en segundo plano

Librerías

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

  • Aplicación: Este es su sistema o servicio principal que necesita interactuar con el LLM.

  • Llamador de funciones: Actúa como un intermediario que envía la entrada al LLM y recibe la salida sin procesar. Este componente encapsula la lógica necesaria para comunicarse con el LLM.

  • LLM (Large Language Model): El modelo de IA que procesa la entrada y genera la salida basándose en la llamada a la función. Ejemplos incluyen GPT-4, Claude3, etc.

  • Response Handler: Toma la salida bruta del LLM y comienza el proceso de estructurarla. Esto puede implicar la verificación de errores, el filtrado y la preparación de datos para la conversión a un formato estructurado.

  • Pydantic Models: Se utilizan para definir explícitamente la estructura de los datos de salida. Los modelos Pydantic aplican la verificación de tipos y la validación de datos, lo que ayuda a garantizar que los datos se ajusten a un esquema especificado.

  • Structured Output: La salida final que está bien estructurada y lista para ser utilizada por la aplicación. Esta salida es predecible y más fácil de integrar en procesos o sistemas posteriores.

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

  • Aplicación Cliente: Esta es su aplicación Python que necesita interactuar con modelos de lenguaje grandes. Utiliza la biblioteca Instructor para facilitar estas interacciones.

  • Biblioteca Instructor: Actúa como un middleware que envuelve a los modelos de lenguaje grandes. Es responsable de enviar solicitudes a estos modelos y procesar sus salidas.

  • Modelos de Lenguaje Grandes: Incluye GPT-4, LLAMA3 y Claude3. Cada uno de estos modelos puede generar salidas de texto complejas basadas en la entrada que reciben.

  • Modelos Pydantic: Estos se utilizan dentro de la biblioteca Instructor para estructurar la salida bruta de los modelos de lenguaje en un formato más manejable y definido, haciendo que el desarrollo de aplicaciones sea más limpio y predecible.

  • Salidas Estructuradas: Las salidas estructuradas finales se devuelven luego a la aplicación cliente, donde pueden ser utilizadas o mostradas.

Implementación

Implementación de la 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)


## Servicios 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

Dominio 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": "El usuario puede proporcionar una instrucción en el idioma de su elección (como inglés, francés, criollo, español, etc.), así que tenga eso en cuenta.",
},
{"role": "user", "content": parametre},
],
)
return gaou

class Language(str, Enum):
nouchi = "Nouchi"
moore = "Mooré"
lingala = "Lingala"
english = "English"
french = "French"
creole = "Créole"
spanish = "Spanish"

class GaouJoke(BaseModel):
friend_gaou_joke: str = Field(
...,
description="El chiste que califica al amigo como Gaou. El chiste debe ser ligero y humorístico, y alternar entre Nouchi, Mooré, Lingala, inglés, francés, criollo y español.",
)
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"""
El término 'Gaou' es un término divertido, usado solo entre amigos. Por ejemplo, ¡{gaou_name} es tan Gaou!.
Ayudarás a calificar a un amigo como Gaou, basándote en los siguientes criterios:
- El nombre del amigo
- Inventa un chiste ligero que siempre termine calificando al amigo como Gaou
- Mezcla algo de humor y sarcasmo"""

  • De alguna manera, Gaou significa alguien que es ingenuo, crédulo o fácilmente engañado, pero de una manera amigable
    - Usa diferentes idiomas de uno de los siguientes: Mooré, inglés, francés, criollo, español, etc.
    """,
    },
    {"role": "user", "content": gaou_name},
    ],
    )
    return gaou

## Implementación del 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):
# we do not want the bot to reply to itself
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, información sobre el comando)

?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, información sobre el comando)
    parametre: str (mensaje para enviar al modelo)

?new_gaou "I am not a Gaou named Lambert who is 15 years old and is 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: Context (discord.ext.commands.Context, información sobre el comando)
limit: int (número de mensajes a eliminar)

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

@bot.command()
@commands.dm_only()
async def dm_cleanup(ctx: commands.Context, limit: int):
"""
ctx: Context (discord.ext.commands.Context, información sobre el comando)
limit: int (número de mensajes a eliminar)

?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én es Gaou de todos modos? ¿Yo Gaou? Piensa de nuevo... {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)


# Despliegue

## Contenedores Docker

### Contenedorizar API de 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"]


### Contenedorizar Bot de 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 de Ayuda

#!/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


## Infraestructura




graph TB
DockerEngine(Docker Engine)

DockerEngine -- Runs --> DockerContainer
DockerEngine -- Builds --> DockerImage

DockerFile(Dockerfile: Receta para Imágenes) -- Define --> DockerImage

DockerHub(Docker Hub: Repositorio Público) -- Almacena y
Comparte --> DockerImage

DockerContainer(Contenedor Docker: Paquete pequeño, autónomo y ejecutable)
DockerImage(Imagen Docker: Planos para Contenedores) -- Crea --> DockerContainer

subgraph "Analogía: Construcción"
DockerFile -- "Plan del Arquitecto" --> DockerImage
DockerImage -- "Partes de casa prefabricadas" --> DockerContainer
end



#### Instalar Docker en Ubuntu

Actualiza tu lista de paquetes existente

sudo apt update

Instala algunos paquetes prerrequisito que permiten a apt usar paquetes sobre HTTPS

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

Añade la clave GPG del repositorio oficial de Docker a tu sistema

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

Añade el repositorio de Docker a las fuentes de 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

Actualiza tu lista de paquetes existente de nuevo para que se reconozca la adición

sudo apt update

Asegúrate de que vas a instalar desde el repositorio de Docker en lugar del repositorio predeterminado de Ubuntu

apt-cache policy docker-ce

Instala Docker

sudo apt install docker-ce

Comprueba que se está ejecutando

sudo systemctl status docker


#### Configurar Docker

Docker sin`sudo`

Agrega tu nombre de usuario al grupo docker

sudo usermod -aG docker ${USER}


Aplica la nueva membresía de grupo, cierra sesión en el servidor y vuelve a iniciarla (¿Opcional?)

su - ${USER}

groups


#### Gestión de Variables de Entorno

Usaremos [direnv](https://direnv.net/) y lo [configuraremos](https://direnv.net/docs/hook.html) para bash dentro de nuestra máquina virtual.

sudo apt install direnv

eval "$(direnv hook bash)"

Volver a artículos