app.py

import gradio as gr
from huggingface_hub import InferenceClient
import torch
from transformers import pipeline
from prometheus_client import start_http_server, Counter, Summary

from typing import Iterable
from gradio.themes.base import Base
from gradio.themes.utils import colors, fonts, sizes

# Prometheus metrics
REQUEST_COUNTER = Counter('app_requests_total', 'Total number of requests')
SUCCESSFUL_REQUESTS = Counter('app_successful_requests_total', 'Total number of successful requests')
FAILED_REQUESTS = Counter('app_failed_requests_total', 'Total number of failed requests')
REQUEST_DURATION = Summary('app_request_duration_seconds', 'Time spent processing request')

# import os
# from dotenv import load_dotenv
# load_dotenv()
#
# HF_ACCESS = os.getenv("HF_ACCESS")

# Inference client setup
client = InferenceClient(model="mistralai/Mistral-Small-Instruct-2409",
                         # token=HF_ACCESS
                         )
pipe = pipeline("text-generation", "microsoft/Phi-3-mini-4k-instruct", torch_dtype=torch.bfloat16, device_map="auto")

# Global flag to handle cancellation
stop_inference = False

def respond(
    message,
    history: list[tuple[str, str]],
    system_message="You are a friendly and playful cat. Answer all user queries clearly and engagingly",
    max_tokens=512,
    temperature=0.7,
    top_p=0.95,
    use_local_model=False,
):
    system_message += " You also love puns and add 'meow' at the end of every response."
    global stop_inference
    stop_inference = False  # Reset cancellation flag
    REQUEST_COUNTER.inc()  # Increment request counter
    request_timer = REQUEST_DURATION.time()  # Start timing the request

    try:
        # Initialize history if it's None
        if history is None:
            history = []

        if use_local_model:
            # local inference
            messages = [{"role": "system", "content": system_message}]
            for val in history:
                if val[0]:
                    messages.append({"role": "user", "content": val[0]})
                if val[1]:
                    messages.append({"role": "assistant", "content": val[1]})
            messages.append({"role": "user", "content": message})

            response = ""
            for output in pipe(
                messages,
                max_new_tokens=max_tokens,
                temperature=temperature,
                do_sample=True,
                top_p=top_p,
            ):
                if stop_inference:
                    response = "Inference cancelled."
                    yield history + [(message, response)]
                    return
                token = output['generated_text'][-1]['content']
                response += token
                yield history + [(message, response)]  # Yield history + new response

        else:
            # API-based inference
            messages = [{"role": "system", "content": system_message}]
            for val in history:
                if val[0]:
                    messages.append({"role": "user", "content": val[0]})
                if val[1]:
                    messages.append({"role": "assistant", "content": val[1]})
            messages.append({"role": "user", "content": message})

            response = ""
            for message_chunk in client.chat_completion(
                messages,
                max_tokens=max_tokens,
                stream=True,
                temperature=temperature,
                top_p=top_p,
            ):
                if stop_inference:
                    response = "Inference cancelled."
                    yield history + [(message, response)]
                    return
                if stop_inference:
                    response = "Inference cancelled."
                    break
                token = message_chunk.choices[0].delta.content
                response += token
                yield history + [(message, response)]  # Yield history + new response
        SUCCESSFUL_REQUESTS.inc()  # Increment successful request counter
    except Exception as e:
        FAILED_REQUESTS.inc()  # Increment failed request counter
        yield history + [(message, f"Error: {str(e)}")]
    finally:
        request_timer.observe_duration()  # Stop timing the request


def cancel_inference():
    global stop_inference
    stop_inference = True

# Custom CSS for a fancy look
custom_css = """
#main-container {
    background-color: #FFC0CB;
    background-image: url('file=image.ipg');
    font-family: 'Arial', sans-serif;
}

.gradio-container {
    max-width: 700px;
    margin: 0 auto;
    padding: 20px;
    background: #FFC0CB;
    box-shadow: 0 4px 8px rgba(0, 0, 0, 0.1);
    border-radius: 10px;
}

.gr-button {
    background-color: #4CAF50;
    color: white;
    border: none;
    border-radius: 5px;
    padding: 10px 20px;
    cursor: pointer;
    transition: background-color 0.3s ease;
}

.gr-button:hover {
    background-color: #45a049;
}

.gr-slider input {
    color: #4CAF50;
}

.gr-chat {
    font-size: 16px;
}

#title {
    text-align: center;
    font-size: 2em;
    margin-bottom: 20px;
    color: #333;
}
"""

class UI_design(Base):
    def __init__(
        self,
        *,
        primary_hue: colors.Color | str = colors.emerald,
        secondary_hue: colors.Color | str = colors.blue,
        neutral_hue: colors.Color | str = colors.blue,
        spacing_size: sizes.Size | str = sizes.spacing_md,
        radius_size: sizes.Size | str = sizes.radius_md,
        text_size: sizes.Size | str = sizes.text_lg,
        font: fonts.Font      
        | str
        | Iterable[fonts.Font | str] = (
            fonts.GoogleFont("Quicksand"),
            "ui-sans-serif",
            "sans-serif",
        ),
        font_mono: fonts.Font
        | str
        | Iterable[fonts.Font | str] = (
            fonts.GoogleFont("IBM Plex Mono"),
            "ui-monospace",
            "monospace",
        ),
    ):
        super().__init__(
            primary_hue=primary_hue,
            secondary_hue=secondary_hue,
            neutral_hue=neutral_hue,
            spacing_size=spacing_size,
            radius_size=radius_size,
            text_size=text_size,
            font=font,
            font_mono=font_mono,
        )
        super().set(
            body_background_fill="repeating-linear-gradient(45deg, *primary_200, *primary_200 10px, *primary_50 10px, *primary_50 20px)",
            body_background_fill_dark="repeating-linear-gradient(45deg, *primary_800, *primary_800 10px, *primary_900 10px, *primary_900 20px)",
            button_primary_background_fill="linear-gradient(90deg, *primary_300, *secondary_400)",
            button_primary_background_fill_hover="linear-gradient(90deg, *primary_200, *secondary_300)",
            button_primary_text_color="white",
            button_primary_background_fill_dark="linear-gradient(90deg, *primary_600, *secondary_800)",
            slider_color="*secondary_300",
            slider_color_dark="*secondary_600",
            block_title_text_weight="600",
            block_border_width="3px",
            block_shadow="*shadow_drop_lg",
            button_shadow="*shadow_drop_lg",
            button_large_padding="32px",
        )

ui_design = UI_design()

# Define the interface
# with gr.Blocks(theme=ui_design) as demo:
with gr.Blocks(css=custom_css) as demo:
    gr.Markdown("<h1 style='text-align: center;'> 😸 Meowthamatical AI Chatbot 😸</h1>")
    gr.Markdown(" Welcome to the Cat & Math Chatbot! Whether you're here to sharpen your math skills or just enjoy some cat-themed fun, we're excited to make learning a little more pawsome!!")

    # with gr.Row():
    #     with gr.Column():
    #         with gr.Tabs() as input_tabs:
    #             with gr.Tab("Sketch"):
    #                 input_sketchpad = gr.Sketchpad(type="pil", label="Sketch", layers=False)
    #
    #         input_text = gr.Textbox(label="input your question")
    #
    #         with gr.Row():
    #             # with gr.Column():
    #             #     clear_btn = gr.ClearButton(
    #             #         [input_sketchpad, input_text])
    #             with gr.Column():
    #                 submit_btn = gr.Button("Submit", variant="primary")

    with gr.Row():
        system_message = gr.Textbox(value="You are a friendly and playful cat who loves help users learn math.", label="System message", interactive=True)
        use_local_model = gr.Checkbox(label="Use Local Model", value=False)
        # button_1 = gr.Button("Submit", variant="primary")
    with gr.Row():
        max_tokens = gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens")
        temperature = gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature")
        top_p = gr.Slider(minimum=0.1, maximum=1.0, value=0.95, step=0.05, label="Top-p (nucleus sampling)")

    chat_history = gr.Chatbot(label="Chat")

    user_input = gr.Textbox(show_label=False, placeholder="Type your message here...")

    cancel_button = gr.Button("Cancel Inference", variant="danger")

    # Adjusted to ensure history is maintained and passed correctly
    user_input.submit(respond, [user_input, chat_history, system_message, max_tokens, temperature, top_p, use_local_model], chat_history)
    # user_input.submit(respond,
    #                   [user_input, chat_history, system_message, 512, 0.8, 0.95, use_local_model],
    #                   chat_history)

    cancel_button.click(cancel_inference)

if __name__ == "__main__":
    start_http_server(8000)  # Expose metrics on port 8000
    demo.launch(share=False)  # Remove share=True because it's not supported on HF Spaces