Funcionalidades

• Executa comandos do usuário

• Filtra spam

• Executa tarefas em segundo plano

Bibliotecas

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

• Aplicação: Este é o seu sistema ou serviço principal que precisa interagir com o LLM.

• Chamador de Função: Atua como um intermediário que envia entrada para o LLM e recebe a saída bruta. Este componente encapsula a lógica necessária para se comunicar com o LLM.

• LLM (Modelo de Linguagem Grande): O modelo de IA que processa a entrada e gera saída com base na chamada de função. Exemplos incluem GPT-4, Claude3, etc.

• Response Handler: Pega a saída bruta do LLM e inicia o processo de estruturá-la. Isso pode envolver verificação de erros, filtragem e preparação de dados para conversão em um formato estruturado.

• Modelos Pydantic: São usados para definir explicitamente a estrutura dos dados de saída. Os modelos Pydantic impõem verificação de tipo e validação de dados, o que ajuda a garantir que os dados estejam em conformidade com um esquema especificado.

• Saída Estruturada: A saída final que é bem estruturada e pronta para ser usada pela aplicação. Esta saída é previsível e mais fácil de integrar em processos ou sistemas subsequentes.

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

• Aplicação Cliente: Esta é a sua aplicação Python que precisa interagir com grandes modelos de linguagem. Ela usa a biblioteca Instructor para facilitar essas interações.

• Biblioteca Instructor: Atua como um middleware que envolve grandes modelos de linguagem. É responsável por enviar requisições a esses modelos e processar suas saídas.

• Modelos de Linguagem Grandes: Inclui GPT-4, LLAMA3 e Claude3. Cada um desses modelos pode gerar saídas de texto complexas com base na entrada que recebem.

• Modelos Pydantic: Estes são usados dentro da biblioteca Instructor para estruturar a saída bruta dos modelos de linguagem em um formato mais gerenciável e definido, tornando o desenvolvimento de aplicações mais limpo e previsível.

• Saídas Estruturadas: As saídas estruturadas finais são então retornadas para a aplicação cliente, onde podem ser posteriormente utilizadas ou exibidas.

Implementação

Implementação da 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)

Serviços 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

## Domínio 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": "O usuário pode fornecer um prompt no idioma de sua escolha (como inglês, francês, crioulo, espanhol etc.), então leve isso em consideração."
            },
            {"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="A piada que qualifica o amigo como um Gaou. A piada deve ser leve e bem-humorada, além de alternar entre Nouchi, Mooré, Lingala, Inglês, Francês, Crioulo e Espanhol.",
    )
    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,  # Vá com tudo na temperatura!!!!
        max_tokens=1024,
        response_model=GaouJoke,
        messages= [
            {
                "role": "system",
                "content": f"""
                O termo 'Gaou' é um termo engraçado, usado apenas entre amigos. Por exemplo, {gaou_name} é tão Gaou!.
                Você ajudará a qualificar um amigo como um Gaou, com base nos seguintes critérios:
                - O nome do amigo
                - Crie uma piada leve que sempre acabe qualificando o amigo como um Gaou
                - Misture um pouco de humor e sarcasmo

- De certa forma, Gaou significa alguém que é ingênuo, crédulo ou facilmente enganado, mas de uma forma amigável
                - Use diferentes idiomas de um dos seguintes: Mooré, English, French, Créole, Spanish etc.
                """,
            },
            {"role": "user", "content": gaou_name},
        ],
    )
    return gaou

Implementação do 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):
    # não queremos que o bot responda a si mesmo
    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} Ei! Não use essa palavra novamente {FROWNING_FACE_EMOJI}"
)
await message.channel.send(
f"{message.author.mention} Você me chamou de: {word} e eu não gosto disso. Eu deletei sua mensagem."
)
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, informações sobre o 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, informações sobre o comando)
parametre: str (mensagem a enviar ao modelo)

?new_gaou "Não sou um Gaou chamado Lambert que tem 15 anos e é inteligente."
"""
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"""

{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, informações sobre o comando)
limit: int (número de mensagens a apagar)

?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, informações sobre o comando)
limit: int (número de mensagens a apagar)

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

async def delete_messages(ctx: commands.Context, limit: int):
print(f"Limpando: {limit} mensagens...")
async for msg in ctx.channel.history(limit=limit):
try:
print(f"Excluindo mensagem: {msg.content}")
await sleep(1)
await msg.delete()
except Exception as e:
print(f"Erro: {e}")
await ctx.reply(f"Você pode não ter permissão para excluir mensagens.")
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"Who's Gaou anyway? Me Gaou? Think again... {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)

# Implantação

## Contêineres Docker

### Dockerizar API do 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"]

### Dockerizar Bot do 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 Auxiliar

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

## Infraestrutura

graph TB

DockerEngine(Docker Engine)

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

DockerFile(Dockerfile: Receita para Imagens) -- Define --> DockerImage

DockerHub(Docker Hub: Repositório Público) -- Armazena e Compartilha --> DockerImage

DockerContainer(Docker Container: Pacote pequeno, autônomo e executável)
DockerImage(Docker Image: Plantas para Contêineres) -- Cria --> DockerContainer

subgraph "Analogia: Construção"
DockerFile -- "Plano do Arquiteto" --> DockerImage
DockerImage -- "Peças de casa pré-fabricadas" --> DockerContainer
end

#### Instalar Docker no Ubuntu

Atualize sua lista de pacotes existente

sudo apt update

Instale alguns pacotes pré-requisitos que permitem ao apt usar pacotes via HTTPS

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

Adicione a chave GPG para o repositório oficial do Docker ao seu sistema

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

Adicione o repositório Docker às fontes 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

Atualize sua lista de pacotes existente novamente para que a adição seja reconhecida

sudo apt update

Certifique-se de que você está prestes a instalar do repositório Docker em vez do repositório padrão do Ubuntu

apt-cache policy docker-ce

Instale o Docker

sudo apt install docker-ce

Verifique se está em execução

sudo systemctl status docker

#### Configurar Docker

Docker sem `sudo`

Add your username to the docker group

sudo usermod -aG docker ${USER}

Aplique a nova associação ao grupo, saia do servidor e faça login novamente (Opcional?)

su - ${USER}

groups

#### Gerenciamento de Variáveis de Ambiente

Usaremos [direnv](https://direnv.net/) e o [configuraremos](https://direnv.net/docs/hook.html) para bash dentro da nossa máquina virtual.

sudo apt install direnv

eval "$(direnv hook bash)"