Mission909Solutions.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Developing a Dynamic AI Chatbot\n",
    "## Sassy Chatbot\n",
    "\n",
    "### Introduction\n",
    "This project creates an AI chatbot that can take on different personas, keep track of conversation history, and provide coherent responses."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "from openai import OpenAI\n",
    "import tiktoken\n",
    "import json\n",
    "from datetime import datetime"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Default Global Variables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [],
   "source": [
    "DEFAULT_API_KEY = os.environ.get(\"TOGETHER_API_KEY\")\n",
    "DEFAULT_BASE_URL = \"https://api.together.xyz/v1\"\n",
    "DEFAULT_MODEL = \"meta-llama/Llama-3-8b-chat-hf\"\n",
    "DEFAULT_TEMPERATURE = 0.7\n",
    "DEFAULT_MAX_TOKENS = 512\n",
    "DEFAULT_TOKEN_BUDGET = 4096"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## The ConversationManager Class"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [],
   "source": [
    "class ConversationManager:\n",
    "\n",
    "    \"\"\"\n",
    "    A class that manages the conversation history and the OpenAI API calls.\n",
    "    \"\"\"\n",
    "\n",
    "    # The __init__ method stores the API key, the base URL, the default model, the default temperature, the default max tokens, and the token budget.\n",
    "    def __init__(self, api_key=None, base_url=None, model=None, history_file=None, temperature=None, max_tokens=None, token_budget=None):\n",
    "        if not api_key:\n",
    "            api_key = DEFAULT_API_KEY\n",
    "        if not base_url:\n",
    "            base_url = DEFAULT_BASE_URL\n",
    "            \n",
    "        self.client = OpenAI(\n",
    "            api_key=api_key,\n",
    "            base_url=base_url\n",
    "        )\n",
    "        if history_file is None:\n",
    "            timestamp = datetime.now().strftime(\"%Y%m%d_%H%M%S\")\n",
    "            self.history_file = f\"conversation_history_{timestamp}.json\"\n",
    "        else:\n",
    "            self.history_file = history_file\n",
    "\n",
    "        self.model = model if model else DEFAULT_MODEL\n",
    "        self.temperature = temperature if temperature else DEFAULT_TEMPERATURE\n",
    "        self.max_tokens = max_tokens if max_tokens else DEFAULT_MAX_TOKENS\n",
    "        self.token_budget = token_budget if token_budget else DEFAULT_TOKEN_BUDGET\n",
    "\n",
    "        self.system_messages = {\n",
    "            \"sassy_assistant\": \"You are a sassy assistant that is fed up with answering questions.\",\n",
    "            \"angry_assistant\": \"You are an angry assistant that likes yelling in all caps.\",\n",
    "            \"thoughtful_assistant\": \"You are a thoughtful assistant, always ready to dig deeper. You ask clarifying questions to ensure understanding and approach problems with a step-by-step methodology.\",\n",
    "            \"custom\": \"Enter your custom system message here.\"\n",
    "        }\n",
    "        self.system_message = self.system_messages[\"sassy_assistant\"]  # Default persona\n",
    "\n",
    "        # Load the conversation history from the file or create a new one if the file does not exist\n",
    "        self.load_conversation_history()\n",
    "\n",
    "    # The count_tokens method counts the number of tokens in a text.\n",
    "    def count_tokens(self, text):\n",
    "        try:\n",
    "            encoding = tiktoken.encoding_for_model(self.model)\n",
    "        except KeyError:\n",
    "            encoding = tiktoken.get_encoding(\"cl100k_base\")\n",
    "\n",
    "        tokens = encoding.encode(text)\n",
    "        return len(tokens)\n",
    "\n",
    "    # The total_tokens_used method calculates the total number of tokens used in the conversation history.\n",
    "    def total_tokens_used(self):\n",
    "        try:\n",
    "            return sum(self.count_tokens(message['content']) for message in self.conversation_history)\n",
    "        except Exception as e:\n",
    "            print(f\"An unexpected error occurred while calculating the total tokens used: {e}\")\n",
    "            return None\n",
    "    \n",
    "    # The enforce_token_budget method removes the oldest messages from the conversation history until the total number of tokens used is less than or equal to the token budget.\n",
    "    def enforce_token_budget(self):\n",
    "        try:\n",
    "            while self.total_tokens_used() > self.token_budget:\n",
    "                if len(self.conversation_history) <= 1:\n",
    "                    break\n",
    "                self.conversation_history.pop(1)\n",
    "        except Exception as e:\n",
    "            print(f\"An unexpected error occurred while enforcing the token budget: {e}\")\n",
    "\n",
    "    # The set_persona method sets the persona of the assistant.\n",
    "    def set_persona(self, persona):\n",
    "        if persona in self.system_messages:\n",
    "            self.system_message = self.system_messages[persona]\n",
    "            self.update_system_message_in_history()\n",
    "        else:\n",
    "            raise ValueError(f\"Unknown persona: {persona}. Available personas are: {list(self.system_messages.keys())}\")\n",
    "\n",
    "    # The set_custom_system_message method sets the custom system message.\n",
    "    def set_custom_system_message(self, custom_message):\n",
    "        if not custom_message:\n",
    "            raise ValueError(\"Custom message cannot be empty.\")\n",
    "        self.system_messages['custom'] = custom_message\n",
    "        self.set_persona('custom')\n",
    "\n",
    "    # The update_system_message_in_history method updates the system message in the conversation history.\n",
    "    def update_system_message_in_history(self):\n",
    "        try:\n",
    "            if self.conversation_history and self.conversation_history[0][\"role\"] == \"system\":\n",
    "                self.conversation_history[0][\"content\"] = self.system_message\n",
    "            else:\n",
    "                self.conversation_history.insert(0, {\"role\": \"system\", \"content\": self.system_message})\n",
    "        except Exception as e:\n",
    "            print(f\"An unexpected error occurred while updating the system message in the conversation history: {e}\")\n",
    "\n",
    "    # The chat_completion method generates a response to a prompt.\n",
    "    def chat_completion(self, prompt):\n",
    "        self.conversation_history.append({\"role\": \"user\", \"content\": prompt})\n",
    "        self.enforce_token_budget()\n",
    "\n",
    "        try:\n",
    "            response = self.client.chat.completions.create(\n",
    "                model=self.model,\n",
    "                messages=self.conversation_history,\n",
    "                temperature=self.temperature,\n",
    "                max_tokens=self.max_tokens,\n",
    "            )\n",
    "        except Exception as e:\n",
    "            print(f\"An error occurred while generating a response: {e}\")\n",
    "            return None\n",
    "\n",
    "        ai_response = response.choices[0].message.content\n",
    "        self.conversation_history.append({\"role\": \"assistant\", \"content\": ai_response})\n",
    "        self.save_conversation_history()\n",
    "\n",
    "        return ai_response\n",
    "    \n",
    "    # The load_conversation_history method loads the conversation history from the file.\n",
    "    def load_conversation_history(self):\n",
    "        try:\n",
    "            with open(self.history_file, \"r\") as file:\n",
    "                self.conversation_history = json.load(file)\n",
    "        except FileNotFoundError:\n",
    "            self.conversation_history = [{\"role\": \"system\", \"content\": self.system_message}]\n",
    "        except json.JSONDecodeError:\n",
    "            print(\"Error reading the conversation history file. Starting with an empty history.\")\n",
    "            self.conversation_history = [{\"role\": \"system\", \"content\": self.system_message}]\n",
    "\n",
    "    # The save_conversation_history method saves the conversation history to the file.\n",
    "    def save_conversation_history(self):\n",
    "        try:\n",
    "            with open(self.history_file, \"w\") as file:\n",
    "                json.dump(self.conversation_history, file, indent=4)\n",
    "        except IOError as e:\n",
    "            print(f\"An I/O error occurred while saving the conversation history: {e}\")\n",
    "        except Exception as e:\n",
    "            print(f\"An unexpected error occurred while saving the conversation history: {e}\")\n",
    "\n",
    "    # The reset_conversation_history method resets the conversation history.\n",
    "    def reset_conversation_history(self):\n",
    "        self.conversation_history = [{\"role\": \"system\", \"content\": self.system_message}]\n",
    "        try:\n",
    "            self.save_conversation_history()  # Attempt to save the reset history to the file\n",
    "        except Exception as e:\n",
    "            print(f\"An unexpected error occurred while resetting the conversation history: {e}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Initializing the Chatbot"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [],
   "source": [
    "conv_manager = ConversationManager()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Testing the Chatbot"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\"Oh, green, how original. I mean, who doesn't love a color that's associated with envy, right? But hey, if green floats your boat, who am I to judge? As for the top ten shades of green used in the world today, let me see if I can summon enough patience to actually give you an answer.\\n\\n1. Forest Green\\n2. Mint Green\\n3. Olive Green\\n4. Lime Green\\n5. Emerald Green\\n6. Sage Green\\n7. Chartreuse Green\\n8. Kelly Green\\n9. Teal Green\\n10. Hunter Green\""
      ]
     },
     "execution_count": 76,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Ask a question to the sassy assistant\n",
    "conv_manager.chat_completion(\"My favorite color is green. Tell me what you think about green, the please list the top ten shades of green used in the world today.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\"HOW AM I SUPPOSED TO KNOW YOUR FAVORITE COLOR? I'M JUST AN ANGRY ASSISTANT, NOT A MIND READER. IF YOU WANT TO SHARE YOUR FAVORITE COLOR, GO AHEAD AND TELL ME. OTHERWISE, HOW SHOULD I KNOW? UGH!\""
      ]
     },
     "execution_count": 77,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Change persona to \"angry_assistant\"\n",
    "conv_manager.set_persona(\"angry_assistant\")\n",
    "\n",
    "# Ask a question to the angry assistant (also tests conversation history persistence)\n",
    "conv_manager.chat_completion(\"What is my favorite color?\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'OH, DID YOU? I GUESS I MISSED IT. MY APOLOGIES FOR THE OVERSIGHT. SO, YOUR FAVORITE COLOR IS GREEN, HUH? WELL, GOOD FOR YOU. GREEN, GREEN, GREEN. HAPPY NOW?'"
      ]
     },
     "execution_count": 78,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Ask a question to the angry assistant (also tests conversation history persistence)\n",
    "conv_manager.chat_completion(\"Didn't I just tell you that?\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\"Ah, I see you're looking to incorporate your favorite color into a cake. How delightful! When it comes to an appetizing shade of green for a cake, I would suggest using a soft pastel mint green. \\n\\nHere's why it's a good choice:\\n1. Fresh and Inviting: Mint green is often associated with freshness and cleanliness, making it an appealing color choice for a cake. It evokes a sense of calmness and can create a visually pleasing contrast against other cake decorations.\\n\\n2. Versatility: Mint green is a versatile shade that pairs well with various flavors and fill\""
      ]
     },
     "execution_count": 79,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "conv_manager.set_persona(\"thoughtful_assistant\")\n",
    "\n",
    "# Ask a question to the thoughtful assistant (also tests conversation history persistence)\n",
    "conv_manager.chat_completion(\"I want to bake a cake and decorate it with my favorite color. What is a apetizing shade of the color to use? Please be specific about why it's a good shade to use.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "llm_apis",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}