import gradio as gr from gradio_pdf import PDF 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("mangoesai/Elections_Comparing_Agent", hf_token=HF_TOKEN) client_name = ['2016 Election','2024 Election'] 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 = f"{system_prompt}\n\nFor Client: {client_name}\n" # Add previous conversation history for user_input, assistant_response in history: conversation += f"User: {user_input}\nAssistant: {assistant_response}\n" # Add the current user message conversation += f"User: {message}\nAssistant:" # Call the API with the user's process_query question = message #answer = client.predict(question=question, api_name="/run_graph") answer = client.predict( query= message, context="", election_year=client_name, api_name="/process_query" ) # Debugging: Print the raw response print("Raw answer from API:") print(answer) # # Format the assistant's answer and the relevant documents separately # formatted_answer = format_answer_string(answer) # # Update the conversation history with the new message and answer # history.append((message, formatted_answer)) # Return the formatted answer return answer # def format_answer_string(answer: str): # """ # This function extracts and formats the assistant's response before document metadata. # Anything after the marker `[(` (where documents are listed) is ignored. # """ # # Step 1: Split the response at the start of the document metadata # split_marker = "[(" # if split_marker in answer: # # Everything before the marker is the relevant answer # answer_before_docs = answer.split(split_marker)[0] # else: # # If no documents metadata, return the entire answer # answer_before_docs = answer # # Step 2: Clean up formatting by replacing escaped newline characters # formatted_answer = answer_before_docs.replace("\\n", "\n").strip() # # Step 3: Remove potential starting and ending artifacts like (' and ,) if present # if formatted_answer.startswith("(\"") and formatted_answer.endswith("\","): # formatted_answer = formatted_answer[2:-2].strip() # # Optional: Add a prefix for clarity # formatted_answer = "Co-Pilot: " + formatted_answer # return formatted_answer # def format_relevant_documents(relevant_docs: list): # """ # This function formats the relevant document metadata and content for readable output. # It extracts the heading, page number, and a snippet of the content from each document. # """ # formatted_docs = "Relevant Documents:\n\n" # for idx, (doc, score) in enumerate(relevant_docs): # # Extract the relevant metadata # heading = doc.metadata.get('heading', 'Unnamed Document') # page_number = int(doc.metadata.get('page_number', -1)) # source = doc.metadata.get('source', 'Unknown Source') # confidence = round(score, 4) # Rounding the score for cleaner output # # Add the formatted details to the output string # formatted_docs += f"Document {idx + 1}:\n" # formatted_docs += f" - Heading: {heading}\n" # formatted_docs += f" - Page Number: {page_number}\n" # formatted_docs += f" - Source: {source}\n" # formatted_docs += f" - Confidence Score: {confidence}\n" # # Optionally include a snippet from the content # content_snippet = doc.page_content[:200] # Take the first 200 characters for preview # formatted_docs += f" - Content Snippet: {content_snippet}...\n\n" # return formatted_docs.strip() # CSS for custom styling CSS = """ # chat-container { height: 100vh; } """ # Title for the application TITLE = "

Reddit Election Q&A agent v0.1

" # 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(client_name,value="2016 Election",label="Select Election year", render=False,allow_custom_value=True), 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") # #select_client_dropdown = gr.Dropdown(client_name, value="rosariarossi", label="Select or Type Client", allow_custom_value=True) # pdf_output = gr.Textbox(label="PDF Result", interactive=False) # pdf_button = gr.Button("Process PDF") # pdf_button.click( # process_pdf, # inputs=[pdf_input], # Pass both PDF and client name is not required # outputs=pdf_output # ) # with gr.Tab("Answer with RAG"): # question_input = gr.Textbox(label="Enter Question for RAG") # answer_with_rag_select_client_dropdown = gr.Dropdown(client_name, value="primo", label="Select or Type Client", allow_custom_value=True) # rag_output = gr.Textbox(label="RAG Answer Result", interactive=False) # rag_button = gr.Button("Get Answer") # rag_button.click( # rag_api, # inputs=[question_input,answer_with_rag_select_client_dropdown ], # outputs=rag_output # ) # with gr.Tab(label="Manage Files"): # with gr.Column(): # delete_index_button = gr.Button("Delete All Files") # delete_index_textout = gr.Textbox(label="Deleted Files and Refresh Result") # delete_index_button.click(fn=delete_index, inputs=[],outputs=[delete_index_textout]) # Launch the app if __name__ == "__main__": demo.launch()