app.py

import gradio as gr
from gradio_client import Client, handle_file
import os

# Define your Hugging Face token (make sure to set it as an environment variable)
HF_TOKEN = os.getenv("HF_TOKEN")  # Replace with your actual token if not using an environment variable

# Initialize the Gradio Client for the specified API
#client = Client("on1onmangoes/CNIHUB10724v10", hf_token=HF_TOKEN)
client = Client("on1onmangoes/CNIHUB101324v10", hf_token=HF_TOKEN)
# on1onmangoes/CNIHUB101324v10


# Here's how you can fix it:

# Update the conversation history within the function.
# Return the updated history along with any other required outputs.

def stream_chat_with_rag(
    message: str,
    history: list,
    client_name: str,
    system_prompt: str,
    num_retrieved_docs: int = 10,
    num_docs_final: int = 9,
    temperature: float = 0,
    max_new_tokens: int = 1024,
    top_p: float = 1.0,
    top_k: int = 20,
    penalty: float = 1.2,
):
    print(f"Message: {message}")
    print(f"History: {history}")

    # Build the conversation prompt including system prompt and history
    conversation = system_prompt + "\n\n" + f"For Client: {client_name}\n"
    if history:  # Check if history exists
        for user_input, assistant_response in history:
            conversation += f"User: {user_input}\nAssistant: {assistant_response}\n"
    conversation += f"User: {message}\nAssistant:"  # Add the current message

    # Prepare the data to send to the API
    api_payload = {
        "message": conversation,  # Include the history in the message,
        "history": history,
        "client_name": client_name,
        "system_prompt": "",  # Optionally set to empty if included in the message
        "num_retrieved_docs": num_retrieved_docs,
        "num_docs_final": num_docs_final,
        "temperature": temperature,
        "max_new_tokens": max_new_tokens,
        "top_p": top_p,
        "top_k": top_k,
        "penalty": penalty,
    }

    try:
        # Make the API call to get the assistant's reply
        response = client.predict(
            api_name="/chat",
            **api_payload
        )

        # Extract the assistant's reply
        if isinstance(response, tuple):
            answer = response[0]
        else:
            answer = response

        # Debugging statements
        print("The Answer in stream_chat_with_rag:")
        print(answer)

        # Update the conversation history
        history.append((message, answer))

    except Exception as e:
        print(f"An error occurred: {e}")
        answer = "There was an error retrieving the response."

    # # Return the updated history
    # return history


# def stream_chat_with_rag(
#     message: str,
#     history: list,
#     client_name: str,
#     system_prompt: str,
#     num_retrieved_docs: int = 10,
#     num_docs_final: int = 9,
#     temperature: float = 0,
#     max_new_tokens: int = 1024,
#     top_p: float = 1.0,
#     top_k: int = 20,
#     penalty: float = 1.2,
# ):
#     print(f"Message: {message}")
#     print(f"History: {history}")

#     # Build the conversation prompt including system prompt and history
#     conversation = system_prompt + "\n\n" + f"For Client: {client_name}\n"
#     for user_input, assistant_response in history:
#         conversation += f"User: {user_input}\nAssistant: {assistant_response}\n"
#     conversation += f"User: {message}\nAssistant:"

#     # Prepare the data to send to the API
#     # Remove 'history' from the payload since the API does not accept it
#     api_payload = {
#         "message": conversation,  # Include the history in the message
#         "client_name": client_name,
#         "system_prompt": "",  # Optionally set to empty if included in message
#         "num_retrieved_docs": num_retrieved_docs,
#         "num_docs_final": num_docs_final,
#         "temperature": temperature,
#         "max_new_tokens": max_new_tokens,
#         "top_p": top_p,
#         "top_k": top_k,
#         "penalty": penalty,
#     }

#     # Make the API call to get the assistant's reply
#     response = client.predict(
#         api_name="/chat",
#         **api_payload
#     )

#     # Extract the assistant's reply
#     if isinstance(response, tuple):
#         answer = response[0]
#     else:
#         answer = response

#     # Debugging statements
#     print("The Answer in stream_chat_with_rag:")
#     print(answer)

#     # Update the conversation history
#     history.append((message, answer))

#     # Return the updated history
#     #return history


# Function to handle PDF processing API call
def process_pdf(pdf_file):
    return client.predict(
        pdf_file=handle_file(pdf_file),
        client_name="rosariarossi",  # Hardcoded client name
        api_name="/process_pdf2"
    )[1]  # Return only the result string

# Function to handle search API call
def search_api(query):
    return client.predict(query=query, api_name="/search_with_confidence")

# Function to handle RAG API call
def rag_api(question):
    return client.predict(question=question, api_name="/answer_with_rag")

# CSS for custom styling
CSS = """
# chat-container {
    height: 100vh;
}
"""

# Title for the application
TITLE = "<h1 style='text-align:center;'>My Gradio Chat App</h1>"

# Create the Gradio Blocks interface
with gr.Blocks(css=CSS) as demo:
    gr.HTML(TITLE)
    with gr.Tab("Chat"):
        chatbot = gr.Chatbot()  # Create a chatbot interface

        chat_interface = gr.ChatInterface(
            fn=stream_chat_with_rag,
            chatbot=chatbot,
            additional_inputs_accordion=gr.Accordion(label="⚙️ Parameters", open=False, render=False),
            additional_inputs=[
                gr.Dropdown(['rosariarossi','bianchifiordaliso','lorenzoverdi'],value="rosariarossi",label="Select Client", render=False,),
                gr.Textbox(
                    value="You are an expert assistant",
                    label="System Prompt",
                    render=False,
                ),
                gr.Slider(
                    minimum=1,
                    maximum=10,
                    step=1,
                    value=10,
                    label="Number of Initial Documents to Retrieve",
                    render=False,
                ),
                gr.Slider(
                    minimum=1,
                    maximum=10,
                    step=1,
                    value=9,
                    label="Number of Final Documents to Retrieve",
                    render=False,
                ),
                gr.Slider(
                    minimum=0.2,
                    maximum=1,
                    step=0.1,
                    value=0,
                    label="Temperature",
                    render=False,
                ),
                gr.Slider(
                    minimum=128,
                    maximum=8192,
                    step=1,
                    value=1024,
                    label="Max new tokens",
                    render=False,
                ),
                gr.Slider(
                    minimum=0.0,
                    maximum=1.0,
                    step=0.1,
                    value=1.0,
                    label="Top P",
                    render=False,
                ),
                gr.Slider(
                    minimum=1,
                    maximum=20,
                    step=1,
                    value=20,
                    label="Top K",
                    render=False,
                ),
                gr.Slider(
                    minimum=0.0,
                    maximum=2.0,
                    step=0.1,
                    value=1.2,
                    label="Repetition Penalty",
                    render=False,
                ),
            ],
        )

    with gr.Tab("Process PDF"):
        pdf_input = gr.File(label="Upload PDF File")
        pdf_output = gr.Textbox(label="PDF Result", interactive=False)

        pdf_button = gr.Button("Process PDF")
        pdf_button.click(
            process_pdf,
            inputs=[pdf_input],
            outputs=pdf_output
        )

    with gr.Tab("Search"):
        query_input = gr.Textbox(label="Enter Search Query")
        search_output = gr.Textbox(label="Search Confidence Result", interactive=False)

        search_button = gr.Button("Search")
        search_button.click(
            search_api,
            inputs=query_input,
            outputs=search_output
        )

    with gr.Tab("Answer with RAG"):
        question_input = gr.Textbox(label="Enter Question for RAG")
        rag_output = gr.Textbox(label="RAG Answer Result", interactive=False)

        rag_button = gr.Button("Get Answer")
        rag_button.click(
            rag_api,
            inputs=question_input,
            outputs=rag_output
        )

# Launch the app
if __name__ == "__main__":
    demo.launch()