app.py

import gradio as gr
import psutil
import subprocess
import time

def generate_response_by_api():
    FinalOutput = ""
    #return FinalOutput
    pass

def generate_response(user_message): #generate_response_token_by_token
    cmd = [
        "/app/llama.cpp/main",  # Path to the executable
        "-m", "/app/llama.cpp/models/stablelm-2-zephyr-1_6b-Q4_0.gguf",
        "-p", user_message,
        "-n", "400",
        "-e"
    ]

    process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1)
    process_monitor = psutil.Process(process.pid)


    start_time = time.time()
    monitor_start_time = time.time()
    alltokens = ""
    token_buffer = ''
    tokencount = 0
    try:
        while True:
            # Read one character at a time
            char = process.stdout.read(1)
            if char == '' and process.poll() is not None:
                break
            if char != '':
                token_buffer += char
                if char == ' ' or char == '\n':  # Token delimiters
                    elapsed_time = time.time() - start_time  # Calculate elapsed time
                    alltokens += token_buffer
                    tokencount += 1
                    yield f"{alltokens} \n\n [Inference time: {elapsed_time:.2f} seconds | Tokens: { tokencount }]"
                    token_buffer = ''  # Reset token buffer
            # Log resource usage every minute
            if time.time() - monitor_start_time > 60:
                cpu_usage = process_monitor.cpu_percent()
                memory_usage = process_monitor.memory_info().rss  # in bytes 
                print(f"Subprocess CPU Usage: {cpu_usage}%, Memory Usage: {memory_usage / 1024 ** 2} MB")
                monitor_start_time = time.time()  # Reset the timer
    
        # Yield the last token if there is any
        if token_buffer:
            elapsed_time = time.time() - start_time  # Calculate elapsed time
            alltokens += token_buffer
            yield f"{alltokens} \n\n [Inference time: {elapsed_time:.2f} seconds | Average Tokens per second: { round(tokencount / elapsed_time, 2) }]"

    finally:
        try:
            # Wait for the process to complete, with a timeout
            process.wait(timeout=60)  # Timeout in seconds
        except subprocess.TimeoutExpired:
            print("Process didn't complete within the timeout. Killing it.")
            process.kill()
            process.wait()  # Ensure proper cleanup
        # Wait for the subprocess to finish if it hasn't already
        process.stdout.close()
        process.stderr.close()

        # Check for any errors
        if process.returncode != 0:
            error_message = process.stderr.read()
            print(f"Error: {error_message}")

def custom_generate_response(cust_user_message, prompt_index):
    """
    Generates a custom response based on the user message and the selected prompt,
    including a custom ending specific to the prompt.

    Parameters:
    - cust_user_message: The message input from the user.
    - prompt_index: The index of the custom prompt to use.
    """
    prompt, ending = CustomPrompts[prompt_index]  # Unpack the prompt and its ending
    cust_user_message = f"{prompt}\n\n{cust_user_message}\n\n{ending}"
    yield from generate_response(cust_user_message)

CustomPrompts = [
    ("Write a Class Diagram based on the following text:", "Class Diagram:"),
    ("Write a Pydot code based on the following text:", "Pydot Code:"),
    ("Describe what a standard happy scene in any movie would be planned in great detail, based on the following text:", "Scene Details"),
    ("Explain a teardown of the product mentioned in the following text:", "Teardown Details:"),
    ("Explain the manufacturing of the product mentioned in the following text:", "Manufacturing Details:"),
    ("Explain the marketing considerations of the product mentioned in the following text:", "Considerations:"),
    ("Explain the target users considerations of the product mentioned in the following text:", "Target Users Considerations:"),
    ("My problem to solve is", "- please make 10 sub problems have to solve from this:"),
]

BusinessPrompts = [
    ("Write an outline for a business plan for " , "")
    ("Write an outline for a Executive Summary for " , "Executive Summary:")
    ("Write an outline for a Company Description for " , "Company Description:")
    ("Write an outline for a Market Analysis for " , "Market Analysis:")
    ("Write an outline for a Marketing and Sales Strategy for " , "Marketing and Sales Strategy:")
    ("Write an outline for a Product Development for " , "Product Development:")
    ("Write an outline for a Operations and Management for " , "Operations and Management:")
    ("Write an outline for a Financial Projections for " , "Financial Projections:")
]

with gr.Blocks() as iface:
    gr.HTML("Stabilityai's demo - https://huggingface.co/spaces/stabilityai/stablelm-2-1_6b-zephyr")
    gr.Interface(
        fn=generate_response,
        inputs=gr.Textbox(lines=2, placeholder="Type your message here..."),
        outputs="text",
        title="Stable LM 2 Zephyr (1.6b) LLama.cpp Interface Test (Inconsistent Performance - 100 tokens in 50 secs (when this HF space is updated) or 800+ secs(HF space open for long))",
        description="No Prompt template used yet (Essentially autocomplete). No Message History for now - Enter your message and get a response.",
        flagging_dir="/usr/src/app/flagged",
    )

    gr.HTML("Any standard way of thinking / Repetitive idea / rule of thumb / advice can be turned into a button (In a timeline?)")

    with gr.Accordion("Random Ideas"):
        with gr.Group():
            gr.HTML("Test for wrapping generator (Instead of buttons tabs and dropdowns?)")
            MainOutput = gr.TextArea(placeholder='Output will show here')
            CustomButtonInput = gr.TextArea(lines=1, placeholder='Prompt goes here')
            
            # Dynamically create buttons and assign actions
            for index, (prompt, _) in enumerate(CustomPrompts):
                button = gr.Button(prompt)
                button.click(custom_generate_response, inputs=[CustomButtonInput, gr.State(index)], outputs=MainOutput)

    with gr.Accordion("General Product based"):
        with gr.Group():
            # Dynamically create buttons and assign actions
            for index, (prompt, _) in enumerate(CustomPrompts):
                button = gr.Button(prompt)
                button.click(custom_generate_response, inputs=[CustomButtonInput, gr.State(index)], outputs=MainOutput)

iface.queue().launch(server_name="0.0.0.0", share=True)