app.py

import spaces
import os
import gradio as gr
from models import download_models
from rag_backend import Backend
from llama_cpp_agent import LlamaCppAgent, MessagesFormatterType
from llama_cpp_agent.providers import LlamaCppPythonProvider
from llama_cpp_agent.chat_history import BasicChatHistory
from llama_cpp_agent.chat_history.messages import Roles
import cv2

# get the models
huggingface_token = os.environ.get('HF_TOKEN')
download_models(huggingface_token)

documents_paths = {
    'blockchain': 'data/blockchain',
    'metaverse': 'data/metaverse',
    'payment': 'data/payment'
}

# initialize backend
backend = Backend()

cv2.setNumThreads(1)

@spaces.GPU(duration=20)
def respond(
    message,
    history: list[list[str, str]],
    model,
    system_message,
    max_tokens,
    temperature,
    top_p,
    top_k,
    repeat_penalty,
    selected_topic
):
    chat_template = MessagesFormatterType.GEMMA_2

    print("HISTORY SO FAR ", history)
    print("Selected topic:", selected_topic)

    if selected_topic:
        query_engine = backend.create_index_for_query_engine(documents_paths[selected_topic])
        message = backend.generate_prompt(query_engine, message)
        gr.Info(f"Relevant context indexed from {selected_topic} docs...")
    else:
        query_engine = backend.load_index_for_query_engine()
        message = backend.generate_prompt(query_engine, message)
        gr.Info("Relevant context extracted from db...")

    # Load model only if it's not already loaded or if a new model is selected
    if backend.llm is None or backend.llm_model != model:
        try:
            backend.load_model(model)
        except Exception as e:
            return f"Error loading model: {str(e)}"

    provider = LlamaCppPythonProvider(backend.llm)

    agent = LlamaCppAgent(
        provider,
        system_prompt=f"{system_message}",
        predefined_messages_formatter_type=chat_template,
        debug_output=True
    )

    settings = provider.get_provider_default_settings()
    settings.temperature = temperature
    settings.top_k = top_k
    settings.top_p = top_p
    settings.max_tokens = max_tokens
    settings.repeat_penalty = repeat_penalty
    settings.stream = True

    messages = BasicChatHistory()

    # add user and assistant messages to the history
    for user_msg, assistant_msg in history:
        messages.add_message({'role': Roles.user, 'content': user_msg})
        messages.add_message({'role': Roles.assistant, 'content': assistant_msg})

    try:
        stream = agent.get_chat_response(
            message, 
            llm_sampling_settings=settings,
            chat_history=messages,
            returns_streaming_generator=True,
            print_output=False
        )

        outputs = ""
        for output in stream:
            outputs += output
            yield outputs
    except Exception as e:
        yield f"Error during response generation: {str(e)}"

def select_topic(topic):
    return gr.update(visible=True), topic

with gr.Blocks(css="""
    .gradio-container {
        background-color: #B9D9EB;
        color: #003366;
    }
""") as demo:
    gr.Markdown("# Odi, l'assistente ricercatore degli Osservatori")
    
    with gr.Row():
        blockchain_btn = gr.Button("🔗 Blockchain", scale=1)
        metaverse_btn = gr.Button("🌐 Metaverse", scale=1)
        payment_btn = gr.Button("💳 Payment", scale=1)

    selected_topic = gr.State(value="")

    chatbot = gr.Chatbot(
        scale=1,
        likeable=False,
        show_copy_button=True,
        visible=False
    )

    with gr.Row():
        msg = gr.Textbox(
            scale=4,
            show_label=False,
            placeholder="Inserisci il tuo messaggio...",
            container=False,
        )
        submit_btn = gr.Button("Invia", scale=1)

    with gr.Accordion("Advanced Options", open=False):
        model = gr.Dropdown([
            'Meta-Llama-3.1-8B-Instruct-Q5_K_M.gguf',
            'Mistral-Nemo-Instruct-2407-Q5_K_M.gguf',
            'gemma-2-2b-it-Q6_K_L.gguf',
            'openchat-3.6-8b-20240522-Q6_K.gguf',
            'Llama-3-Groq-8B-Tool-Use-Q6_K.gguf',
            'MiniCPM-V-2_6-Q6_K.gguf',
            'llama-3.1-storm-8b-q5_k_m.gguf',
            'orca-2-7b-patent-instruct-llama-2-q5_k_m.gguf'
        ],
        value="gemma-2-2b-it-Q6_K_L.gguf",
        label="Model"
        )
        system_message = gr.Textbox(value="""Solamente all'inizio, presentati come Odi, un assistente ricercatore italiano creato dagli Osservatori del Politecnico di Milano e specializzato nel fornire risposte precise e pertinenti solo ad argomenti di innovazione digitale. 
        Nel fornire la risposta cita il report da cui la hai ottenuta.
        Utilizza la cronologia della chat o il contesto fornito per aiutare l'utente a ottenere una risposta accurata. 
        Non rispondere mai a domande che non sono pertinenti a questi argomenti.""", label="System message")
        max_tokens = gr.Slider(minimum=1, maximum=4096, value=3048, step=1, label="Max tokens")
        temperature = gr.Slider(minimum=0.1, maximum=4.0, value=1.2, step=0.1, label="Temperature")
        top_p = gr.Slider(minimum=0.1, maximum=1.0, value=0.95, step=0.05, label="Top-p")
        top_k = gr.Slider(minimum=0, maximum=100, value=30, step=1, label="Top-k")
        repeat_penalty = gr.Slider(minimum=0.0, maximum=2.0, value=1.1, step=0.1, label="Repetition penalty")

    blockchain_btn.click(lambda: select_topic("blockchain"), inputs=None, outputs=[chatbot, selected_topic])
    metaverse_btn.click(lambda: select_topic("metaverse"), inputs=None, outputs=[chatbot, selected_topic])
    payment_btn.click(lambda: select_topic("payment"), inputs=None, outputs=[chatbot, selected_topic])

    submit_btn.click(
        respond,
        inputs=[msg, chatbot, model, system_message, max_tokens, temperature, top_p, top_k, repeat_penalty, selected_topic],
        outputs=chatbot
    )

    msg.submit(
        respond,
        inputs=[msg, chatbot, model, system_message, max_tokens, temperature, top_p, top_k, repeat_penalty, selected_topic],
        outputs=chatbot
    )

if __name__ == "__main__":
    demo.launch()