Discord_bot2.py

import os
import json
import requests

# the Discord Python API
import discord

# Queue code
from collections import deque

# Initializing a queue
maxsize = 3
past_input = deque(maxlen=maxsize)
generated_responses = deque(maxlen=maxsize)

# this is my Hugging Face profile link
API_URL = "https://api-inference.huggingface.co/models/Nakul24/"


class User:
    def __init__(self, id: str):
        self.id = id
        maxsize = 3
        self.past_input = deque(maxlen=maxsize)
        self.generated_responses = deque(maxlen=maxsize)

    def update(self, past_input: deque, generated_responses: deque):
        self.past_input = past_input
        self.generated_responses = generated_responses

    def getpast(self):
        return self.past_input

    def getresponse(self):
        return self.generated_responses


users = {}


class MyClient(discord.Client):
    def __init__(self, model_name):

        # Define the intents
        intents = discord.Intents.default()  # This gives you the default intents
        intents.messages = True  # Ensure the bot has the intent to receive messages

        super().__init__(intents=intents)
        self.api_endpoint = API_URL + model_name
        # retrieve the secret API token from the system environment
        huggingface_token = os.environ["Huggingface_token"]
        # format the header in our request to Hugging Face
        self.request_headers = {"Authorization": "Bearer {}".format(huggingface_token)}

    def query(self, payload):
        """
        make request to the Hugging Face model API
        """
        data = json.dumps(payload)
        response = requests.request(
            "POST", self.api_endpoint, headers=self.request_headers, data=data
        )
        ret = json.loads(response.content.decode("utf-8"))
        return ret

    async def on_ready(self):
        # print out information when the bot wakes up
        print("Logged in as")
        print(self.user.name)
        print(self.user.id)
        print("------")

        # send a request to the model without caring about the response
        # just so that the model wakes up and starts loading
        self.query({"inputs": {"text": "Yo!"}})

    async def on_message(self, message):
        """
        this function is called whenever the bot sees a message in a channel
        """

        # ignore the message if it comes from the bot itself
        if message.author.id == self.user.id:
            return
        global users

        if message.author.id not in users:
            users[message.author.id] = User(message.author.id)

        past_input = users[message.author.id].getpast()
        generated_responses = users[message.author.id].getresponse()

        # Start with the current message content and append the separator right after
        separator = "<|endoftext|>"
        input_text = message.content + separator

        # Then, for each past input and corresponding response, concatenate them with the separator
        for past_input, generated_response in zip(past_input, generated_responses):
            input_text += past_input + separator + generated_response + separator

        print("Input Text is - ", input_text)

        # Reset chat on reset command
        if message.content == "Reset":
            past_input = deque(maxlen=maxsize)
            generated_responses = deque(maxlen=maxsize)
            users[message.author.id].update(past_input, generated_responses)
            await message.channel.send("Reset Complete")
            return
        else:

            # form query payload with the content of the message
            input_text = f"{message.content} " + " ".join(concatenated_pairs)

            print("Input Text is - ", input_text)
            payload = {
                "inputs": input_text,
                "parameters": {
                    "min_length": 4,
                    "max_length": 150,
                    "top_k": 650,
                    "top_p": 0.92,
                    "temperature": 0.70,
                    "repetition_penalty": 1.5,
                },
            }
            past_input.append(message.content)

            # while the bot is waiting on a response from the model
            # set the its status as typing for user-friendliness
            async with message.channel.typing():
                response = self.query(payload)
                print("Response is - ", response)

            # Check if response is a list and has at least one item
            if isinstance(response, list) and len(response) > 0:
                # Assuming the first item of the list is the dictionary you're interested in
                first_response_item = response[0]
                bot_response = first_response_item.get("generated_text", None)
            else:
                # Handle cases where the response is not as expected
                bot_response = None
                print("Unexpected response format:", response)
            generated_responses.append(bot_response)
            users[message.author.id].update(past_input, generated_responses)
            # print(bot_response + f"\n @{message.author.name}")
            # print(generated_responses)

            # we may get ill-formed response if the model hasn't fully loaded
            # or has timed out

            if not bot_response:
                if "error" in response:
                    past_input = deque(maxlen=maxsize)
                    generated_responses = deque(maxlen=maxsize)
                    users[message.author.id].update(past_input, generated_responses)
                    bot_response = "`Error: {}` \n Reset Complete".format(
                        response["error"]
                    )

                else:
                    bot_response = "Hmm... something is not right. Please message Reset"

            # send the model's response to the Discord channel
            await message.channel.send(
                bot_response + "\n<@{}>".format(message.author.id)
            )


def main():
    # DialoGPT-medium-joshua is my model name
    client = MyClient("YC_Bot")
    client.run(os.environ["Discord_token"])


if __name__ == "__main__":
    main()