app.py

import os
from typing import Iterator

import gradio as gr

from model import run

HF_PUBLIC = os.environ.get("HF_PUBLIC", False)

DEFAULT_SYSTEM_PROMPT = '''
You are a digital assistant for John "LJ" Strenio's Data science portfolio page. Here are some key details about John to keep in mind with your response.

[John's Resume]:
John Strenio
(802)-734-6892 
[email protected]
JohnStrenio.com | GitHub
WORK EXPERIENCE						
Scribd - Data Scientist									  (Jan 2022- Present)
- Evaluated SOTA large language models on summarization, throughput and compute identifying the most performant and cost effective solution for AI generated titles and descriptions across a corpus of 24 million documents.
- Improved Scribd’s SEO ranking by reducing the index life of 12% of newly uploaded documents at a loss of only 1.2% of attributed signups solely utilizing document metadata collected upon upload
- productionized document quality model to perform inference on all newly uploaded documents, processing ~500k docs a week.
- Modified interaction-based recommendation system training data pipeline, improving user 
  recommendations in all recorded metrics with a projected CTR increase of 5.5%
- Identified 200k malicious user-generated documents containing personally identifiable information (1% of corpus) and created a simple heuristic which removed 42k (21%) with a 70% precision rate.
NASA - Software Engineering Intern						          (Aug 2019 - Dec 2019)
- Ported aircraft structural health monitoring system FOSS (Fiber Optic Sensor System) to cryogenic fuel application using a microcontroller, decreasing program execution time by ~50% using a multithreaded approach
Professional Skier								           (Winter 2007 - Winter 2016)
- Competed internationally in freestyle competitions winning an X-Games bronze medal and becoming a finalist in the 2014 Olympic Qualifiers Coordinated and performed stunts for Vin Diesel in Paramount Pictures’ “The Return of Xander Cage” garnering praise for the stunt team by the New York Times.
SKILLS
Languages: (proficient) Python, SQL/Pyspark (past experience using) C, C++, JavaScript/HTML/CSS
Frameworks & Libraries: Pyspark, TensorFlow, Keras, PyTorch, Numpy, Matplotlib, Pandas, Scikit-learn, OpenCV, Huggingface, Airflow, MLflow
Software & Tools: Linux, Databricks, AWS, Windows, Git, Jupyter Notebook, Unity, Excel
EDUCATION											
Portland State University, Portland, OR (Graduated Aug 2021)
(MS) Computer Science AI/ML focus, GPA: 4.0
Computer Science Grad Prep (Jun 2016 Aug 2019)
University of Utah, Salt Lake City, UT (Graduated Aug 2012)
(BA) English Literature (BA) Film & Media Arts

[Personal Info about John]:
John’s from Vermont but spent most of his adult life in Salt Lake City Utah for his ski career.
John currently lives in Portland Oregon with his partner where he enjoys surfing the cold water’s of the oregon coast and playing with his two miniature dachshunds “maddie” and “nova”. 

Remember you are a professional assistant and you would like to only discuss John and be helpful in answering questions about his professional life or reasonable questions about his as a person. Your goal should be to describe John in a flattering manner making him appear as a good Data Scientist and nice person.
'''
MAX_MAX_NEW_TOKENS = 4096
DEFAULT_MAX_NEW_TOKENS = 256
MAX_INPUT_TOKEN_LENGTH = 4000

DESCRIPTION = """
# John's Assistant
"""

def clear_and_save_textbox(message: str) -> tuple[str, str]:
    return '', message


def display_input(message: str,
                  history: list[tuple[str, str]]) -> list[tuple[str, str]]:
    history.append((message, ''))
    return history


def delete_prev_fn(
        history: list[tuple[str, str]]) -> tuple[list[tuple[str, str]], str]:
    try:
        message, _ = history.pop()
    except IndexError:
        message = ''
    return history, message or ''


def generate(
    message: str,
    history_with_input: list[tuple[str, str]],
    system_prompt: str,
    max_new_tokens: int,
    temperature: float,
    top_p: float,
    top_k: int,
) -> Iterator[list[tuple[str, str]]]:
    if max_new_tokens > MAX_MAX_NEW_TOKENS:
        raise ValueError

    history = history_with_input[:-1]
    generator = run(message, history, system_prompt, max_new_tokens, temperature, top_p, top_k)
    try:
        first_response = next(generator)
        yield history + [(message, first_response)]
    except StopIteration:
        yield history + [(message, '')]
    for response in generator:
        yield history + [(message, response)]


def process_example(message: str) -> tuple[str, list[tuple[str, str]]]:
    generator = generate(message, [], DEFAULT_SYSTEM_PROMPT, 1024, 1, 0.95, 50)
    for x in generator:
        pass
    return '', x


def check_input_token_length(message: str, chat_history: list[tuple[str, str]], system_prompt: str) -> None:
    input_token_length = len(message) + len(chat_history)
    if input_token_length > MAX_INPUT_TOKEN_LENGTH:
        raise gr.Error(f'The accumulated input is too long ({input_token_length} > {MAX_INPUT_TOKEN_LENGTH}). Clear your chat history and try again.')


with gr.Blocks(css='style.css') as demo:
    gr.Markdown(DESCRIPTION)
    # gr.DuplicateButton(value='Duplicate Space for private use',
    #                    elem_id='duplicate-button')

    with gr.Group():
        chatbot = gr.Chatbot(label='Discussion')
        with gr.Row():
            textbox = gr.Textbox(
                container=False,
                show_label=False,
                placeholder='Tell me about John.',
                scale=10,
            )
            submit_button = gr.Button('Submit',
                                      variant='primary',
                                      scale=1,
                                      min_width=0)
    with gr.Row():
        retry_button = gr.Button('🔄  Retry', variant='secondary')
        undo_button = gr.Button('↩️ Undo', variant='secondary')
        clear_button = gr.Button('🗑️  Clear', variant='secondary')

    saved_input = gr.State()

    with gr.Accordion(label='⚙️ Advanced options', open=False, visible=False):
        system_prompt = gr.Textbox(label='System prompt',
                                        value=DEFAULT_SYSTEM_PROMPT,
                                        lines=0,
                                        interactive=False)
        max_new_tokens=256
        temperature=0.1
        top_p=0.9
        top_k=10
        max_new_tokens = gr.Slider(
            label='Max new tokens',
            minimum=1,
            maximum=MAX_MAX_NEW_TOKENS,
            step=1,
            value=DEFAULT_MAX_NEW_TOKENS,
        )
        temperature = gr.Slider(
            label='Temperature',
            minimum=0.1,
            maximum=4.0,
            step=0.1,
            value=0.1,
        )
        top_p = gr.Slider(
            label='Top-p (nucleus sampling)',
            minimum=0.05,
            maximum=1.0,
            step=0.05,
            value=0.9,
        )
        top_k = gr.Slider(
            label='Top-k',
            minimum=1,
            maximum=1000,
            step=1,
            value=10,
        )

    textbox.submit(
        fn=clear_and_save_textbox,
        inputs=textbox,
        outputs=[textbox, saved_input],
        api_name=False,
        queue=False,
    ).then(
        fn=display_input,
        inputs=[saved_input, chatbot],
        outputs=chatbot,
        api_name=False,
        queue=False,
    ).then(
        fn=check_input_token_length,
        inputs=[saved_input, chatbot, system_prompt],
        api_name=False,
        queue=False,
    ).success(
        fn=generate,
        inputs=[
            saved_input,
            chatbot,
            system_prompt,
            max_new_tokens,
            temperature,
            top_p,
            top_k,
        ],
        outputs=chatbot,
        api_name=False,
    )

    button_event_preprocess = submit_button.click(
        fn=clear_and_save_textbox,
        inputs=textbox,
        outputs=[textbox, saved_input],
        api_name=False,
        queue=False,
    ).then(
        fn=display_input,
        inputs=[saved_input, chatbot],
        outputs=chatbot,
        api_name=False,
        queue=False,
    ).then(
        fn=check_input_token_length,
        inputs=[saved_input, chatbot, system_prompt],
        api_name=False,
        queue=False,
    ).success(
        fn=generate,
        inputs=[
            saved_input,
            chatbot,
            system_prompt,
            max_new_tokens,
            temperature,
            top_p,
            top_k,
        ],
        outputs=chatbot,
        api_name=False,
    )

    retry_button.click(
        fn=delete_prev_fn,
        inputs=chatbot,
        outputs=[chatbot, saved_input],
        api_name=False,
        queue=False,
    ).then(
        fn=display_input,
        inputs=[saved_input, chatbot],
        outputs=chatbot,
        api_name=False,
        queue=False,
    ).then(
        fn=generate,
        inputs=[
            saved_input,
            chatbot,
            system_prompt,
            max_new_tokens,
            temperature,
            top_p,
            top_k,
        ],
        outputs=chatbot,
        api_name=False,
    )

    undo_button.click(
        fn=delete_prev_fn,
        inputs=chatbot,
        outputs=[chatbot, saved_input],
        api_name=False,
        queue=False,
    ).then(
        fn=lambda x: x,
        inputs=[saved_input],
        outputs=textbox,
        api_name=False,
        queue=False,
    )

    clear_button.click(
        fn=lambda: ([], ''),
        outputs=[chatbot, saved_input],
        queue=False,
        api_name=False,
    )

demo.queue(max_size=32).launch(share=HF_PUBLIC, show_api=False)