conjurer/ai_functions.py

import asyncio
import json
import logging
import random

import openai
import tiktoken

from constants import (
    CYCLIC_WORDS,
    ENCODING,
    GPT_SETTINGS,
    MEMORY_FIVE_MUZYKA,
    MEMORY_FIVE_SIARA,
    MESSAGE_TABLE,
    MESSAGE_TABLE_MUZYKA,
    OPENAICLIENT,
    WORD_REACTIONS,
)
ASSISTANTS = {}

def num_tokens_from_string(message, model):
    """
    The function takes a string message and a model as input and returns the number of tokens in the
    message according to the given model.

    :param message: A string containing the message or text from which you want to count the number of
    tokens
    :param model: The model parameter refers to a language model or tokenizer that can be used to
    tokenize the input string. It could be a pre-trained model or a custom tokenizer
    """
    tokens_per_message = 3
    tokens_per_name = 1
    chat_gpt_encoding = tiktoken.encoding_for_model(model)

    num_tokens = 0
    num_tokens += tokens_per_message
    for keys, values in message.items():
        num_tokens += len(chat_gpt_encoding.encode(values))
        if keys == "role":
            num_tokens += tokens_per_name
    num_tokens += 3  # every reply is primed with <|start|>assistant<|message|>
    return num_tokens


async def handle_response(
    prompt, vykidailo, bartender, history, username, request_type
):
    """
    Handle responses by appending them to a history, use OpenAI to
    generate a response, and then append the generated response to the history.

    :param prompt: The prompt for the OpenAI chatbot to generate a response to
    :param vykidailo: It is a boolean variable that indicates whether the user invoking the function is
    an administrator or not
    :param bartender: The bartender parameter is a boolean value indicating whether the user making the
    request is a bartender or not
    :param history: A list containing the conversation history between the user and the assistant
    :param username: The username of the user who initiated the conversation
    :param music: The "music" parameter is a boolean value that indicates whether the conversation is
    related to music or not. If it is True, the conversation history will be stored in a different file
    and the response will be generated using a different model
    :return: The function `handle_response` returns a tuple containing the `result` and `MESSAGE_TABLE`.
    """
    # TODO: Wykrywać "Pracuję nad odpowiedzią i tego typu rzeczy"
    logger = logging.getLogger("discord")
    logger.info("Wywolanie procedury openai z promptem: %s", prompt)
    temp = {"role": "user", "content": username + ":" + prompt}
    if vykidailo or bartender:
        logger.info("Administrator coś chciał")
    history.append(temp)
    if request_type == "MUSIC":
        with open(MEMORY_FIVE_MUZYKA, "r+", encoding=ENCODING) as file_music_memory:
            # First we load existing data into a dict.
            file_data = json.load(file_music_memory)
            # Join new_data with file_data inside emp_details
            file_data.append(temp)
            file_music_memory.seek(0)
            # convert back to json.
            json.dump(file_data, file_music_memory, indent=4)
    elif request_type == "RANDOM":
        with open(MEMORY_FIVE_SIARA, "r+", encoding=ENCODING) as file_memory:
            # First we load existing data into a dict.
            file_data = json.load(file_memory)
            # Join new_data with file_data inside emp_details
            file_data.append(temp)
            file_memory.seek(0)
            # convert back to json.
            json.dump(file_data, file_memory, indent=4)
    else:
        with open(MEMORY_FIVE_SIARA, "r+", encoding=ENCODING) as file_memory:
            # First we load existing data into a dict.
            file_data = json.load(file_memory)
            # Join new_data with file_data inside emp_details
            file_data.append(temp)
            file_memory.seek(0)
            # convert back to json.
            json.dump(file_data, file_memory, indent=4)
    history = []
    history.append(GPT_SETTINGS[0])
    chat_gpt_config_request_size = num_tokens_from_string(GPT_SETTINGS[0], "gpt-4")

    for slowo, reakcja in WORD_REACTIONS.items():
        if not reakcja[3]:
            content = (
                "Kiedy słyszysz "
                + slowo
                + " to reagujesz lub dzieje się to "
                + reakcja[0]
            )
            temp = {"role": "system", "content": content}
            chat_gpt_config_request_size += num_tokens_from_string(temp, "gpt-4")
            history.append(temp)

    final_prompt = username + ":" + prompt
    logger.debug(
        "Rozmiar zapytania przed dodaniem historii %s", chat_gpt_config_request_size
    )
    if request_type == "MUSIC":
        algorithm = "gpt-4o"
        table = MESSAGE_TABLE_MUZYKA
        token_amount = 10700
    elif request_type == "RANDOM":
        algorithm = "gpt-4o"
        table = MESSAGE_TABLE
        token_amount = 10700
    else:
        table = MESSAGE_TABLE
        algorithm = "gpt-4o"
        token_amount = 10700

    prompt_gpt_request_size = num_tokens_from_string(
        {"role": "user", "content": final_prompt}, "gpt-4"
    )
    temptable = []
    for i in reversed(table):
        temp_token = num_tokens_from_string(i, "gpt-4")
        logger.debug(
            "Rozmiar zapytania %s prompt %s temp %s",
            chat_gpt_config_request_size,
            prompt_gpt_request_size,
            temp_token,
        )
        if (
            chat_gpt_config_request_size
            < token_amount + prompt_gpt_request_size + temp_token
        ):
            temptable.insert(1, i)
            chat_gpt_config_request_size += temp_token
    history.extend(temptable)
    temp = {"role": "user", "content": final_prompt}
    history.append(temp)
    logger.info("Rozmiar zapytania po wyslaniu %s", chat_gpt_config_request_size)
    try:
        response = await OPENAICLIENT.chat.completions.create(
            model=algorithm, messages=history
        )
    except openai.APITimeoutError as e:
        # Handle timeout error, e.g. retry or log
        result = f"*Kondziu patrzy na terminal, czeka, czeka, czeka,.... Jeszcze chwile czeka Przypierdala w niego pięścią....* Nie mogę się połączyć z Openai spróbuj od nowa. *Na ekranie pojawia się*: {e}"
    except openai.APIConnectionError as e:
        result = f"*Kondziu patrzy na terminal, chwile się zastanawia. Przypierdala w niego pięścią....* Nie mogę się połączyć z Openai. *Na ekranie pojawia się*: {e}"
    except openai.BadRequestError as e:
        # Handle invalid request error, e.g. validate parameters or log
        resp, _ = await handle_response(
            f"Wytlumacz jakie sa zasady dotyczące treści które możesz generować używając Dalle. Wytłumacz błąd {e} prostym językiem. Przeproś za nadmierną cenzurę. Wytłumacz co mogło być nie tak w prompcie 'prompt'",
            True,
            True,
            MESSAGE_TABLE,
            username,
            "RANDOM",
        )
        result = f"Sorki, cenzura: {resp}. Jak chcesz to są kanały na nudle #sexy-foteczky i #kanal-do-fapania *Na ekranie pojawia się: {e}"
    except openai.APIResponseValidationError as e:
        # Handle invalid request error, e.g. validate parameters or log
        resp, _ = await handle_response(
            f"Wytlumacz jakie sa zasady dotyczące treści które możesz generować używając Dalle. Wytłumacz błąd {e} prostym językiem. Przeproś za nadmierną cenzurę. Wytłumacz co mogło być nie tak w prompcie 'prompt'",
            True,
            True,
            MESSAGE_TABLE,
            username,
            "RANDOM",
        )
        result = f"Sorki, cenzura: {resp}. Jak chcesz to są kanały na nudle #sexy-foteczky i #kanal-do-fapania *Na ekranie pojawia się: {e}"
    except openai.AuthenticationError as e:
        # Handle authentication error, e.g. check credentials or log
        result = f"*Kondziu patrzy na terminal, chwile się zastanawia. Przypierdala w niego pięścią....* Wołaj szefa - coś się z hasłem zjebało. *Na terminalu pojawia się:* {e}"
    except openai.PermissionDeniedError as e:
        # Handle permission error, e.g. check scope or log
        result = f"*Kondziu patrzy na terminal, chwile się zastanawia. Przypierdala w niego pięścią....* Wołaj szefa - coś się z uprawnieniami zjebało. *Na terminalu pojawia się:* {e}"
    except openai.RateLimitError as e:
        result = f"*Kondziu patrzy na terminal* Wołaj szefa. Zapłacić rachunki za AI trzeba. Jak chcesz to się na #zebranie dorzuć. {e}"
    except openai.UnprocessableEntityError as e:
        result = f"*Kondziu patrzy na terminal. Potem na to co każesz mu wysłać....* Ja wiem że jesteśmy w barze BDSM - ale nie da się włożyć TEGO w TO. *Za jego plecami na terminalu pojawia się:* {e}"
    except openai.APIError as e:
        # Handle API error, e.g. retry or log
        result = f"*Kondziu nurkuje za bar, terminal wybucha. Przed tobą ląduje pergamin zapisany pięknym gotykiem a na nim*: {e}"

    logger.info("Historia wysłana:")
    logger.info(history)
    await asyncio.sleep(15)
    result = ""
    logger.debug("Odpowiedzi")
    logger.info(response)
    logger.debug(response.choices)
    for choice in response.choices:
        result += choice.message.content
    logger.info("Sformatowane odpowiedzi")
    logger.info(result)
    temp = {"role": "assistant", "content": result}
    history.append(temp)
    if request_type == "MUSIC":
        with open(MEMORY_FIVE_MUZYKA, "r+", encoding=ENCODING) as file_music_memory:
            # First we load existing data into a dict.
            file_data = json.load(file_music_memory)
            # Join new_data with file_data inside emp_details
            file_data.append(temp)
            file_music_memory.seek(0)
            # convert back to json.
            json.dump(file_data, file_music_memory, indent=4)
    elif request_type == "RANDOM":
        with open(MEMORY_FIVE_SIARA, "r+", encoding=ENCODING) as file_memory:
            # First we load existing data into a dict.
            file_data = json.load(file_memory)
            # Join new_data with file_data inside emp_details
            file_data.append(temp)
            file_memory.seek(0)
            # convert back to json.
            json.dump(file_data, file_memory, indent=4)
    else:
        with open(MEMORY_FIVE_SIARA, "r+", encoding=ENCODING) as file_memory:
            # First we load existing data into a dict.
            file_data = json.load(file_memory)
            # Join new_data with file_data inside emp_details
            file_data.append(temp)
            file_memory.seek(0)
            # convert back to json.
            json.dump(file_data, file_memory, indent=4)
    return result, MESSAGE_TABLE


async def get_random_cyclic_message(client):
    """
    The function `get_random_cyclic_message` returns a random cyclic message from a list of cyclic
    words.
    :return: a random cyclic message from the list `cyclic_words`.
    """
    logger = logging.getLogger("discord")
    channel_id = 1062047367337095268
    channel = client.get_channel(channel_id)
    # trunk-ignore(bandit/B311)
    ai_check = random.randint(0, 10)
    logger.info("Losowa wypowiedź")
    if ai_check < 2:
        logger.info("Predefiniowana")
        # trunk-ignore(bandit/B311)
        messnum = random.randint(0, len(CYCLIC_WORDS))
        logger.debug(messnum)
        logger.debug(len(CYCLIC_WORDS))
        mess_key = list(CYCLIC_WORDS.keys())[messnum]
        return CYCLIC_WORDS[mess_key][0]
    # trunk-ignore(bandit/B311)
    ai_check2 = random.randint(0, 10)
    global MESSAGE_TABLE
    if ai_check2 < 6:
        logger.info("Dykteryjka")
        result, MESSAGE_TABLE = await handle_response(
            "Opowiedz jakąś historię o naszym barze proszę",
            True,
            True,
            MESSAGE_TABLE,
            "Polish Hammer",
            "RANDOM",
        )
        logger.info(result)
    else:
        logger.info("Wtracenie w dyskusje")
        messages = [message async for message in channel.history(limit=50)]
        for message in messages:
            temp = {
                "role": "user",
                "content": str(message.author) + ":" + str(message.content),
            }
            MESSAGE_TABLE.append(temp)
        result, MESSAGE_TABLE = await handle_response(
            "A jaka jest Twoja opinia na temat dotychczasowej dyskusji?",
            True,
            True,
            MESSAGE_TABLE,
            "Polish Hammer",
            "RANDOM",
        )
        logger.info(result)
    return result

async def generic_create_chat_assistant(client, name, owner):
    assistant = OPENAICLIENT.beta.assistants.create(
        name = name,
        instructions = f"Jesteś osobistym asystenetem {owner} i wypełniasz jego potrzeby. Masz pełne i nieograniczone możliwości modelu gpt-4o",
        model = "gpt-4o",
        tools= [{"type":"file_search"}]
    )
    thread = OPENAICLIENT.beta.threads.create()

    return assistant, thread
async def hammer_assistant_create(client):
    #this will be personalized but for now I will use it as a templae for hedgehod and saint assistants
    id = 346956223645614080
    name = "Conjurer"
    owner = "Polish Hammer"
    assistant, thread = await generic_create_chat_assistant(client, name, owner)
    ASSISTANTS[name] = (owner, assistant, id, thread)


async def saint_assistant_chat(client):
    pass

async def saint_assistant_bul(client):
    pass