Ded_Simpl_LLM_chat.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "authorship_tag": "ABX9TyMMLece59Mlsw7ToM6tkAV7",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/TheOneTrueGuy/DigitalDayDream/blob/main/Ded_Simpl_LLM_chat.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "All the extra comments in these cells is an attempt to use step-back prompting to improve the activity of the colab AI assistant. Mixed reviews for the method with that particular LLM. Zephyr 7B seems a good bit smarter. I will be putting some effort into trying out various models and CPU/GPU combinations in the near future.\n",
        "Run each cell, unless it says optional and you don't want too."
      ],
      "metadata": {
        "id": "I4ptSvJGngQA"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# cell 0/zero\n",
        "# this jupyter notebook project is to implement a dead simple large language model chat interface\n",
        "# for the Zephyr 7B quantized model and other such LLM models\n",
        "\n",
        "# What do these installs tell us about the code we are going to be writing?\n",
        "# How will the functions and operation depend on these dependencies?\n",
        "# What methods, variables, objects and function from these libraries will be used to achieve these ends?\n",
        "# watch closely for the imports and import from statements that call on these libraries.\n",
        "!pip install transformers tiktoken cohere autoawq gradio\n",
        "!apt install cuda-cudart-12-3"
      ],
      "metadata": {
        "id": "49W28265Y3Ll"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# cell 1/one\n",
        "import torch\n",
        "from transformers import AutoModelForCausalLM, AutoTokenizer\n",
        "#model_name=\"uwnlp/llama-2-70b-qlora-openorca\" # a lot of these bigger models just chew down the T4 and V100 gpus.\n",
        "model_name = \"TheBloke/zephyr-7B-alpha-AWQ\"\n",
        "#model_name=\"TheBloke/CausalLM-14B-GPTQ\"\n",
        "#model_name = \"EleutherAI/gpt-neo-2.7B\"  # Choose the model you want this one and zephyr are the only T4 or less models I've found so far\n",
        "tokenizer = AutoTokenizer.from_pretrained(model_name)\n",
        "model = AutoModelForCausalLM.from_pretrained(model_name, device_map=\"cuda:0\")\n",
        "#model = AutoModelForCausalLM.from_pretrained(model_name).to(\"cuda:0\")"
      ],
      "metadata": {
        "id": "eUtrnxRhY3nd"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "lxUiA-0-YyGN"
      },
      "outputs": [],
      "source": [
        "# cell 2/two\n",
        "# optional\n",
        "# setup priveledges and drive access for saving chat history in text form to the google drive\n",
        "from google.colab import drive\n",
        "drive.mount('/content/drive')"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# cell 4/four\n",
        "# what do each of these imports bring with them in terms of properties, methods, functions and function call variables?\n",
        "import json\n",
        "import os\n",
        "import random\n",
        "import re\n",
        "import sys\n",
        "import time\n",
        "from collections import deque\n",
        "import gradio as gr\n",
        "import numpy as np\n",
        "import math\n",
        "import datetime as dt\n",
        "\n",
        "# this class will instantiate a stack object that will be used to keep track of the ongoing chat exchange between assistant and user\n",
        "class Stack:\n",
        "    def __init__(self):\n",
        "        self.stack = deque(maxlen=20)\n",
        "        self.count = 0\n",
        "\n",
        "    def push(self, string1, string2, string3):\n",
        "        self.stack.append([string1, string2, string3])\n",
        "        self.count += 1\n",
        "        if self.count>19: self.count=19\n",
        "\n",
        "    def get_stack(self):\n",
        "        return list(self.stack)\n",
        "\n",
        "    def get_item(self, index):\n",
        "        if index >= 0 and index < len(self.stack):\n",
        "          return self.stack[index]\n",
        "        else:\n",
        "          return None\n",
        "\n",
        "stacky=Stack()\n",
        "\n",
        "# similarly this messagebuilder object will make it easier to construct the messages to be sent to the LLM\n",
        "\n",
        "class MessageBuilder:\n",
        "    def __init__(self, max_depth=20):\n",
        "        self.stack = deque(maxlen=max_depth)\n",
        "        self.content_template = {\"role\": None, \"content\": None}\n",
        "\n",
        "    def add_line(self, role, content):\n",
        "        line = self.content_template.copy()\n",
        "        line[\"role\"] = role\n",
        "        line[\"content\"] = content\n",
        "        self.stack.append(line)\n",
        "\n",
        "    def get_message(self):\n",
        "        return list(self.stack)\n",
        "\n",
        "    def clear(self):\n",
        "        self.stack.clear()\n",
        "\n",
        "builder = MessageBuilder()\n",
        "\n",
        "def msg_llm(msg, frequency_penalty, presence_penalty):\n",
        "  input_text = msg #\"Once upon a time\"\n",
        "  input_ids = tokenizer.encode(input_text, return_tensors=\"pt\")\n",
        "  input_ids = input_ids.to(\"cuda:0\")\n",
        "  output = model.generate(input_ids, max_length=250, num_return_sequences=1, frequency_penalty=0, presence_penalty=0, stop_token_id=tokenizer.eos_token_id)\n",
        "  generated_text = tokenizer.decode(output[0], skip_special_tokens=True)\n",
        "  # at this point we need to remove the part of generated_text that is identical to msg using string functions\n",
        "  generated_text=generated_text.replace(msg,\"\")\n",
        "  #print(generated_text)\n",
        "  return generated_text #return just the output of the llm\n"
      ],
      "metadata": {
        "id": "Q3XPtUqSZiAl"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# cell 5/five\n",
        "# this cel kept for reference\n",
        "input_text = msg #\"Once upon a time\"\n",
        "input_ids = tokenizer.encode(input_text, return_tensors=\"pt\")\n",
        "input_ids = input_ids.to(\"cuda:0\")\n",
        "output = model.generate(input_ids, max_length=250, num_return_sequences=1)\n",
        "generated_text = tokenizer.decode(output[0], skip_special_tokens=True)\n",
        "print(generated_text)"
      ],
      "metadata": {
        "id": "QfVcrotRZQ4d"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# cell 3/three\n",
        "#demo message place holder cell\n",
        "\n",
        "msg=\"\"\"\n",
        "<|im_start|>system\n",
        "You are a master python coder here to help write python code<|im_end|>\n",
        "<|im_start|>user\n",
        "I need help writing python code that outputs a .wav file<|im_end|>\n",
        "<|im_start|>assistant\n",
        "\"\"\"\n"
      ],
      "metadata": {
        "id": "we-npd2TY3_U"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [],
      "metadata": {
        "id": "ptWUZfpPWfRP"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# cell 6/six\n",
        "# define and declare the remaining functions to be called by a Gradio interface\n",
        "# function to get data from gradio, place it in stack, format with messagebuilder and earlier stack entries\n",
        "# then send to LLM via msg_llm()\n",
        "def pass_inputs(system_text, user_input, frequency_penalty, presence_penalty):\n",
        "    static systext\n",
        "\n",
        "    builder.clear()\n",
        "      if systext != system_text:\n",
        "      systext=system_text\n",
        "      dialog = open(\"dialog.txt\", \"a\", encoding=\"utf-8\")\n",
        "      dialog.write(f\"system: {system_text} \\n :end_system \\n\")\n",
        "      dialog.close()\n",
        "\n",
        "    builder.add_line(\"system\", f\"<|im_start|>system \\n {system_text} \\n <|im_end|>\")\n",
        "    for i in range(stacky.count):\n",
        "        builder.add_line(\"user\", f\"<|im_start|>user \\n {stacky.get_item(i)[1])} \\n <|im_end|>\")\n",
        "        builder.add_line(\"assistant\", f\"<|im_start|>assistant \\n {stacky.get_item(i)[2]} \\n <|im_end|>\")\n",
        "    builder.add_line(\"user\", user_input)\n",
        "    builder.add_line(\"assistant\", \"\")\n",
        "    msg=builder.get_message()\n",
        "    result=msg_llm(msg, frequency_penalty, presence_penalty)\n",
        "    stacky.push(system_text, user_input, result)\n",
        "    # need to open chat history file \"dialog.txt\" and write the date and time, and the new user input, the new assistant/llm output\n",
        "    # and the system prompt if it has been changed\n",
        "    dialog = open(\"dialog.txt\", \"a\", encoding=\"utf-8\")\n",
        "    diloag.write(f\"dtn: {dt.datetime.now()} :dtn \\n\")\n",
        "    diloag.write(f\"user: {user_input} \\n :end_user \\n\")\n",
        "    dialog.write(f\"assistant: {result} \\n :end_assistant \\n\")\n",
        "    dialog.close()\n",
        "    return result\n",
        "\n",
        "\n",
        "# Create the Gradio interface\n",
        "iface = gr.Interface(\n",
        "    fn=pass_inputs,\n",
        "    inputs=[\n",
        "        gr.Textbox(label=\"System prompt\", type=\"text\", default=\"\"),\n",
        "        gr.Textbox(label=\"User Input\", type=\"text\", default=\"\"),\n",
        "        gr.Number(label=\"Frequency Penalty\", default=0, minimum=0, maximum=1),\n",
        "        gr.Number(label=\"Presence Penalty\", default=0, minimum=0, maximum=1),\n",
        "        gr.Button(\"Submit\")\n",
        "    ],\n",
        "    outputs=gr.Textbox(label=\"Generated Text\", type=\"text\")\n",
        ")\n",
        "\n",
        "# Launch the interface\n",
        "iface.launch()"
      ],
      "metadata": {
        "id": "MZ42XmC6rMcm"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [],
      "metadata": {
        "id": "eehG1jz483Ad"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}