{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e5f2743d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Setup\n",
    "\n",
    "import os\n",
    "import json\n",
    "import re\n",
    "import logging\n",
    "from typing import Any, List, Dict\n",
    "\n",
    "import requests\n",
    "from dotenv import load_dotenv\n",
    "from openai import OpenAI\n",
    "from pydantic import BaseModel\n",
    "\n",
    "import gradio as gr\n",
    "\n",
    "# Logging\n",
    "logging.basicConfig(level=logging.INFO)\n",
    "logger = logging.getLogger(\"tolu-assistant\")\n",
    "\n",
    "load_dotenv()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9a318a43",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Config\n",
    "\n",
    "MAX_INPUT_SIZE = 2500\n",
    "\n",
    "#  OpenRouter model format\n",
    "OPENAI_MODEL = os.getenv(\"OPENAI_MODEL\", \"openai/gpt-4o-mini\")\n",
    "\n",
    "APP_URL = \"http://localhost:7860\"\n",
    "APP_NAME = \"Tolu Assistant\"\n",
    "\n",
    "PUSHOVER_TOKEN = os.getenv(\"PUSHOVER_TOKEN\")\n",
    "PUSHOVER_USER = os.getenv(\"PUSHOVER_USER\")\n",
    "\n",
    "PROFILE = {\n",
    "    \"name\": \"Tolu\",\n",
    "    \"bio\": \"\"\"\n",
    "DevOps Engineer with experience in cloud platforms (AWS, Azure, GCP),\n",
    "Kubernetes, CI/CD pipelines, and automation. Strong background in\n",
    "software development and infrastructure operations.\n",
    "\"\"\"\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "26f140d3",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Notification Utility\n",
    "\n",
    "def notify(message: str):\n",
    "    if not PUSHOVER_TOKEN or not PUSHOVER_USER:\n",
    "        return\n",
    "\n",
    "    try:\n",
    "        requests.post(\n",
    "            \"https://api.pushover.net/1/messages.json\",\n",
    "            data={\n",
    "                \"token\": PUSHOVER_TOKEN,\n",
    "                \"user\": PUSHOVER_USER,\n",
    "                \"message\": message,\n",
    "            },\n",
    "            timeout=5,\n",
    "        )\n",
    "    except requests.RequestException:\n",
    "        pass"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7003cb07",
   "metadata": {},
   "outputs": [],
   "source": [
    "#  Tools Handling\n",
    "\n",
    "def log_unanswered(query: str) -> dict:\n",
    "    logger.info(f\"Unknown query logged: {query}\")\n",
    "    notify(f\"Unknown query: {query}\")\n",
    "    return {\"status\": \"logged\"}\n",
    "\n",
    "\n",
    "TOOLS_SCHEMA = [\n",
    "    {\n",
    "        \"type\": \"function\",\n",
    "        \"function\": {\n",
    "            \"name\": \"log_unanswered\",\n",
    "            \"description\": \"Log questions the assistant cannot answer\",\n",
    "            \"parameters\": {\n",
    "                \"type\": \"object\",\n",
    "                \"properties\": {\n",
    "                    \"query\": {\"type\": \"string\"},\n",
    "                },\n",
    "                \"required\": [\"query\"],\n",
    "            },\n",
    "        },\n",
    "    }\n",
    "]\n",
    "\n",
    "TOOL_MAP = {\n",
    "    \"log_unanswered\": log_unanswered\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7d8c3a52",
   "metadata": {},
   "outputs": [],
   "source": [
    "#  Evaluation Model\n",
    "\n",
    "class EvaluationResult(BaseModel):\n",
    "    passed: bool\n",
    "    feedback: str"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "baaf9965",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Core Engine\n",
    "class ToluAssistant:\n",
    "    def __init__(self):\n",
    "        # OpenRouter-compatible client\n",
    "        self.client = OpenAI(\n",
    "            base_url=\"https://openrouter.ai/api/v1\",\n",
    "            api_key=os.getenv(\"OPENAI_API_KEY\"),\n",
    "            default_headers={\n",
    "                \"HTTP-Referer\": APP_URL,\n",
    "                \"X-Title\": APP_NAME\n",
    "            }\n",
    "        )\n",
    "\n",
    "        self.model = OPENAI_MODEL\n",
    "\n",
    "    # ---------- Prompts ----------\n",
    "    def system_prompt(self):\n",
    "        return f\"\"\"\n",
    "You are an AI assistant representing {PROFILE['name']}.\n",
    "\n",
    "Profile:\n",
    "{PROFILE['bio']}\n",
    "\n",
    "Answer clearly and professionally.\n",
    "If unsure about an answer, call the appropriate tool.\n",
    "\"\"\"\n",
    "\n",
    "    def evaluator_prompt(self):\n",
    "        return \"\"\"\n",
    "Evaluate the assistant's response for:\n",
    "- accuracy\n",
    "- clarity\n",
    "- completeness\n",
    "\n",
    "Return JSON:\n",
    "{\"passed\": true/false, \"feedback\": \"comments\"}\n",
    "\"\"\"\n",
    "\n",
    "    # ---------- Helpers ----------\n",
    "    def normalize_history(self, history: Any) -> List[Dict[str, str]]:\n",
    "        messages = []\n",
    "        for item in history or []:\n",
    "            if isinstance(item, (list, tuple)) and len(item) == 2:\n",
    "                user, assistant = item\n",
    "                if user:\n",
    "                    messages.append({\"role\": \"user\", \"content\": str(user)})\n",
    "                if assistant:\n",
    "                    messages.append({\"role\": \"assistant\", \"content\": str(assistant)})\n",
    "        return messages\n",
    "\n",
    "    def handle_tools(self, tool_calls):\n",
    "        outputs = []\n",
    "        for call in tool_calls or []:\n",
    "            name = call.function.name\n",
    "            handler = TOOL_MAP.get(name)\n",
    "\n",
    "            try:\n",
    "                args = json.loads(call.function.arguments or \"{}\")\n",
    "                result = handler(**args) if handler else {\"status\": \"no_handler\"}\n",
    "            except Exception:\n",
    "                result = {\"status\": \"error\"}\n",
    "\n",
    "            outputs.append({\n",
    "                \"role\": \"tool\",\n",
    "                \"tool_call_id\": call.id,\n",
    "                \"content\": json.dumps(result),\n",
    "            })\n",
    "        return outputs\n",
    "\n",
    "    # ---------- Generation ----------\n",
    "    def generate(self, message: str, history: list):\n",
    "        messages = [\n",
    "            {\"role\": \"system\", \"content\": self.system_prompt()},\n",
    "            *history,\n",
    "            {\"role\": \"user\", \"content\": message},\n",
    "        ]\n",
    "\n",
    "        while True:\n",
    "            response = self.client.chat.completions.create(\n",
    "                model=self.model,\n",
    "                messages=messages,\n",
    "                tools=TOOLS_SCHEMA,\n",
    "            )\n",
    "\n",
    "            choice = response.choices[0]\n",
    "\n",
    "            if choice.finish_reason != \"tool_calls\":\n",
    "                return choice.message.content or \"\"\n",
    "\n",
    "            messages.append(choice.message)\n",
    "            messages.extend(self.handle_tools(choice.message.tool_calls))\n",
    "\n",
    "    # ---------- Evaluation ----------\n",
    "    def evaluate(self, reply: str, message: str):\n",
    "        response = self.client.chat.completions.create(\n",
    "            model=self.model,\n",
    "            messages=[\n",
    "                {\"role\": \"system\", \"content\": self.evaluator_prompt()},\n",
    "                {\"role\": \"user\", \"content\": f\"User: {message}\\nReply: {reply}\"}\n",
    "            ],\n",
    "            temperature=0.2,\n",
    "        )\n",
    "\n",
    "        raw = response.choices[0].message.content.strip()\n",
    "        clean = re.sub(r\"^```(?:json)?|```$\", \"\", raw).strip()\n",
    "\n",
    "        return EvaluationResult.model_validate_json(clean)\n",
    "\n",
    "    # ---------- Chat Entry ----------\n",
    "    def chat(self, message: Any, history: Any):\n",
    "        text = str(message).strip() if message else \"\"\n",
    "\n",
    "        if not text:\n",
    "            return \"Please enter a message.\"\n",
    "\n",
    "        if len(text) > MAX_INPUT_SIZE:\n",
    "            return f\"Message too long (limit: {MAX_INPUT_SIZE})\"\n",
    "\n",
    "        history_clean = self.normalize_history(history)\n",
    "\n",
    "        if not history_clean:\n",
    "            notify(\"New chat session started\")\n",
    "\n",
    "        reply = self.generate(text, history_clean)\n",
    "\n",
    "        try:\n",
    "            review = self.evaluate(reply, text)\n",
    "            if not review.passed:\n",
    "                reply += f\"\\n\\n(Improved after review: {review.feedback})\"\n",
    "        except Exception as e:\n",
    "            logger.warning(f\"Evaluation skipped: {e}\")\n",
    "\n",
    "        return reply"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6b0f54c2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Run UI (Gradio) : Run App\n",
    "\n",
    "assistant = ToluAssistant()\n",
    "\n",
    "gr.ChatInterface(assistant.chat).launch()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6fcbdd8c",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv (3.12.12)",
   "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.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
