copilot.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.display import display, Markdown"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dotenv\n",
    "import os\n",
    "\n",
    "dotenv.load_dotenv()\n",
    "\n",
    "\n",
    "def _getOrThrow(key):\n",
    "    value = os.getenv(key)\n",
    "    if value is None:\n",
    "        raise RuntimeError(f\"Missing environment variable: {key}\")\n",
    "    return value\n",
    "\n",
    "\n",
    "OPENAI_API_KEY = _getOrThrow(\"OPENAI_API_KEY\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "\n",
    "\n",
    "class GptCompleter:\n",
    "    def __init__(self, system_prompt: str) -> None:\n",
    "        self._system_prompt = system_prompt\n",
    "\n",
    "    def complete(self, prompt: str) -> str:\n",
    "        response = requests.post(\n",
    "            \"https://api.openai.com/v1/chat/completions\",\n",
    "            json={\n",
    "                \"model\": \"gpt-4-0125-preview\",\n",
    "                \"max_tokens\": 4000,\n",
    "                \"temperature\": 0,\n",
    "                \"top_p\": 1,\n",
    "                \"frequency_penalty\": 0,\n",
    "                \"presence_penalty\": 0,\n",
    "                \"messages\": [\n",
    "                    {\"role\": \"system\", \"content\": self._system_prompt},\n",
    "                    {\"role\": \"user\", \"content\": prompt},\n",
    "                ],\n",
    "            },\n",
    "            headers={\n",
    "                \"Content-Type\": \"application/json\",\n",
    "                \"Authorization\": f\"Bearer {OPENAI_API_KEY}\",\n",
    "            },\n",
    "        )\n",
    "        response.raise_for_status()\n",
    "        data = response.json()\n",
    "        return data[\"choices\"][0][\"message\"][\"content\"]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pathlib import Path\n",
    "\n",
    "\n",
    "SYSTEM_PROMPT_TEPMPLATE = \"\"\"\\\n",
    "You are a programming assistant. \n",
    "\n",
    "When asked to update, please provide only the code that you want to add or change.\n",
    "\n",
    "\n",
    "Source files are:\n",
    "```\n",
    "{filenames}\n",
    "```\n",
    "\n",
    "Source code:\n",
    "\n",
    "{source_files}\n",
    "\"\"\"\n",
    "\n",
    "\n",
    "SOURCE_PART_TEMPLATE = \"\"\"\\\n",
    "name: `{filename}`\n",
    "\n",
    "contents:\n",
    "```\n",
    "{contents}\n",
    "```\n",
    "\"\"\"\n",
    "\n",
    "INCLUDE_SUFFIXES = [\".py\", \".md\", \".json\"]\n",
    "EXCLUDE_DIRS = [\".git\", \"__pycache__\", \".venv\"]\n",
    "\n",
    "\n",
    "def all_files_recursively(dir=\".\", exclude_dirs=EXCLUDE_DIRS):\n",
    "    result = []\n",
    "    for p in Path(dir).iterdir():\n",
    "        if p.is_dir() and p.name in exclude_dirs:\n",
    "            continue\n",
    "        if p.is_file():\n",
    "            result.append(p)\n",
    "        else:\n",
    "            result.extend(all_files_recursively(p, exclude_dirs))\n",
    "    return result\n",
    "\n",
    "\n",
    "def build_sys_prompt(dir=\".\"):\n",
    "    all_files = all_files_recursively(dir)\n",
    "    filenames = [str(p.relative_to(dir)) for p in all_files]\n",
    "    parts = []\n",
    "    for p in all_files:\n",
    "        if p.suffix in INCLUDE_SUFFIXES:\n",
    "            try:\n",
    "                contents = p.read_text() or \"EMPTY FILE\"\n",
    "            except Exception as e:\n",
    "                print(f\"Error reading {p}: {e}\")\n",
    "                return\n",
    "            parts.append(\n",
    "                SOURCE_PART_TEMPLATE.format(\n",
    "                    filename=p.relative_to(dir), contents=contents\n",
    "                )\n",
    "            )\n",
    "    return SYSTEM_PROMPT_TEPMPLATE.format(\n",
    "        filenames=\"\\n\".join(filenames), source_files=\"\\n\".join(parts)\n",
    "    )\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "display(Markdown(build_sys_prompt()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "To implement thread creation in the `RuntimeManager` class for starting bots, you can use the `threading` module in Python. The idea is to create a new thread for each bot that needs to be started, and then run the bot's start method within that thread. This allows each bot to operate concurrently without blocking the main execution thread of your application.\n",
       "\n",
       "Here's how you can implement the `start` method in the `RuntimeManager` class to create and start a new thread for each bot:\n",
       "\n",
       "```python\n",
       "from threading import Thread\n",
       "\n",
       "class RuntimeManager:\n",
       "    def __init__(self, master_config: MasterConfig):\n",
       "        self._master_config = check_required(master_config, \"main_config\", MasterConfig)\n",
       "        self._threads = {}  # Keep track of threads by bot_id\n",
       "\n",
       "    def start_all(self):\n",
       "        for bot_conf in self._master_config.bots():\n",
       "            self.start(bot_conf)\n",
       "\n",
       "    def stop_all(self):\n",
       "        for bot_id in list(self._threads.keys()):\n",
       "            self.stop(bot_id)\n",
       "\n",
       "    def start(self, bot_conf: BotConf):\n",
       "        if isinstance(bot_conf, TgBotConf):\n",
       "            # Define the target function for the thread\n",
       "            def target():\n",
       "                from app.src.server.tg_bot import TgBot  # Import here to avoid circular imports\n",
       "                bot = TgBot(bot_conf)\n",
       "                bot.start()\n",
       "\n",
       "            # Create and start a new thread for the bot\n",
       "            thread = Thread(target=target, name=bot_conf.bot_id(), daemon=True)\n",
       "            thread.start()\n",
       "            self._threads[bot_conf.bot_id()] = thread\n",
       "        else:\n",
       "            raise NotImplementedError(\n",
       "                f\"Bot type {bot_conf.__class__.__name__} is not supported\"\n",
       "            )\n",
       "        \n",
       "    def stop(self, bot_id: str):\n",
       "        # Here you would implement logic to gracefully stop the bot\n",
       "        # For now, we'll just remove it from the threads dictionary\n",
       "        if bot_id in self._threads:\n",
       "            # Implement stopping logic here if necessary\n",
       "            del self._threads[bot_id]\n",
       "```\n",
       "\n",
       "This implementation adds a `_threads` dictionary to the `RuntimeManager` class to keep track of the threads by their `bot_id`. When `start` is called for a bot configuration, it checks if the bot is of type `TgBotConf`, and if so, it creates a new thread targeting a function that initializes and starts the bot. The thread is then started and stored in the `_threads` dictionary.\n",
       "\n",
       "The `stop` method is a placeholder where you would implement the logic to gracefully stop a bot. For now, it simply removes the bot's thread from the `_threads` dictionary. You would need to add more logic here to actually stop the bot's execution, depending on how your bot is designed to handle shutdown signals.\n",
       "\n",
       "This approach allows each bot to run in its own thread, enabling concurrent operation of multiple bots within your application."
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "QUERY = \"\"\"\\\n",
    "Please suggest how to implement Thread creation in RuntimeManager.\n",
    "\"\"\"\n",
    "res = GptCompleter(build_sys_prompt()).complete(QUERY)\n",
    "display(Markdown(res))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "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.12.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}