# Chat_ui.py # Description: Chat interface functions for Gradio # # Imports import logging import os import sqlite3 import time from datetime import datetime # # External Imports import gradio as gr # # Local Imports from App_Function_Libraries.Chat.Chat_Functions import approximate_token_count, chat, save_chat_history, \ update_chat_content, save_chat_history_to_db_wrapper from App_Function_Libraries.DB.DB_Manager import db, load_chat_history, start_new_conversation, \ save_message, search_conversations_by_keywords, \ get_all_conversations, delete_messages_in_conversation, search_media_db, list_prompts from App_Function_Libraries.DB.RAG_QA_Chat_DB import get_db_connection from App_Function_Libraries.Gradio_UI.Gradio_Shared import update_dropdown, update_user_prompt from App_Function_Libraries.Metrics.metrics_logger import log_counter, log_histogram from App_Function_Libraries.Utils.Utils import default_api_endpoint, format_api_name, global_api_endpoints # # ######################################################################################################################## # # Functions: def show_edit_message(selected): if selected: return gr.update(value=selected[0], visible=True), gr.update(value=selected[1], visible=True), gr.update( visible=True) return gr.update(visible=False), gr.update(visible=False), gr.update(visible=False) def show_delete_message(selected): if selected: return gr.update(value=selected[1], visible=True), gr.update(visible=True) return gr.update(visible=False), gr.update(visible=False) def debug_output(media_content, selected_parts): print(f"Debug - Media Content: {media_content}") print(f"Debug - Selected Parts: {selected_parts}") return "" def update_selected_parts(use_content, use_summary, use_prompt): selected_parts = [] if use_content: selected_parts.append("content") if use_summary: selected_parts.append("summary") if use_prompt: selected_parts.append("prompt") print(f"Debug - Update Selected Parts: {selected_parts}") return selected_parts # Old update_user_prompt shim for backwards compatibility def get_system_prompt(preset_name): # For backwards compatibility prompts = update_user_prompt(preset_name) return prompts["system_prompt"] def clear_chat(): """ Return empty list for chatbot and None for conversation_id @return: """ return gr.update(value=[]), None def clear_chat_single(): """ Clears the chatbot and chat history. Returns: list: Empty list for chatbot messages. list: Empty list for chat history. """ return [], [] # FIXME - add additional features.... def chat_wrapper(message, history, media_content, selected_parts, api_endpoint, api_key, custom_prompt, conversation_id, save_conversation, temperature, system_prompt, max_tokens=None, top_p=None, frequency_penalty=None, presence_penalty=None, stop_sequence=None): try: if save_conversation: if conversation_id is None: # Create a new conversation media_id = media_content.get('id', None) conversation_name = f"Chat about {media_content.get('title', 'Unknown Media')} - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" conversation_id = start_new_conversation(title=conversation_name, media_id=media_id) # Add user message to the database user_message_id = save_message(conversation_id, role="user", content=message) # Include the selected parts and custom_prompt only for the first message if not history and selected_parts: message_body = "\n".join(selected_parts) full_message = f"{custom_prompt}\n\n{message}\n\n{message_body}" elif custom_prompt: full_message = f"{custom_prompt}\n\n{message}" else: full_message = message # Generate bot response bot_message = chat(full_message, history, media_content, selected_parts, api_endpoint, api_key, custom_prompt, temperature, system_prompt) logging.debug(f"Bot message being returned: {bot_message}") if save_conversation: # Add assistant message to the database save_message(conversation_id, role="assistant", content=bot_message) # Update history new_history = history + [(message, bot_message)] return bot_message, new_history, conversation_id except Exception as e: logging.error(f"Error in chat wrapper: {str(e)}") return "An error occurred.", history, conversation_id def search_conversations(query): """Convert existing chat search to use RAG chat functions""" try: # Use the RAG search function - search by title if given a query if query and query.strip(): results, _, _ = search_conversations_by_keywords( title_query=query.strip() ) else: # Get all conversations if no query results, _, _ = get_all_conversations() if not results: return gr.update(choices=[]) # Format choices to match existing UI format conversation_options = [ (f"{conv['title']} (ID: {conv['conversation_id'][:8]})", conv['conversation_id']) for conv in results ] return gr.update(choices=conversation_options) except Exception as e: logging.error(f"Error searching conversations: {str(e)}") return gr.update(choices=[]) def load_conversation(conversation_id): """Convert existing load to use RAG chat functions""" if not conversation_id: return [], None try: # Use RAG load function messages, _, _ = load_chat_history(conversation_id) # Convert to chatbot history format history = [ (content, None) if role == 'user' else (None, content) for role, content in messages ] return history, conversation_id except Exception as e: logging.error(f"Error loading conversation: {str(e)}") return [], None def regenerate_last_message(history, media_content, selected_parts, api_endpoint, api_key, custom_prompt, temperature, system_prompt): if not history: return history, "No messages to regenerate." last_entry = history[-1] last_user_message, last_bot_message = last_entry if last_bot_message is None: return history, "The last message is not from the bot." new_history = history[:-1] if not last_user_message: return new_history, "No user message to regenerate the bot response." full_message = last_user_message bot_message = chat( full_message, new_history, media_content, selected_parts, api_endpoint, api_key, custom_prompt, temperature, system_prompt ) new_history.append((last_user_message, bot_message)) return new_history, "Last message regenerated successfully." def update_dropdown_multiple(query, search_type, keywords=""): """Updated function to handle multiple search results using search_media_db""" try: # Define search fields based on search type search_fields = [] if search_type.lower() == "keyword": # When searching by keyword, we'll search across multiple fields search_fields = ["title", "content", "author"] else: # Otherwise use the specific field search_fields = [search_type.lower()] # Perform the search results = search_media_db( search_query=query, search_fields=search_fields, keywords=keywords, page=1, results_per_page=50 # Adjust as needed ) # Process results item_map = {} formatted_results = [] for row in results: id, url, title, type_, content, author, date, prompt, summary = row # Create a display text that shows relevant info display_text = f"{title} - {author or 'Unknown'} ({date})" formatted_results.append(display_text) item_map[display_text] = id return gr.update(choices=formatted_results), item_map except Exception as e: logging.error(f"Error in update_dropdown_multiple: {str(e)}") return gr.update(choices=[]), {} def create_chat_interface(): try: default_value = None if default_api_endpoint: if default_api_endpoint in global_api_endpoints: default_value = format_api_name(default_api_endpoint) else: logging.warning(f"Default API endpoint '{default_api_endpoint}' not found in global_api_endpoints") except Exception as e: logging.error(f"Error setting default API endpoint: {str(e)}") default_value = None custom_css = """ .chatbot-container .message-wrap .message { font-size: 14px !important; } """ with gr.TabItem("Remote LLM Chat (Horizontal)", visible=True): gr.Markdown("# Chat with a designated LLM Endpoint, using your selected item as starting context") chat_history = gr.State([]) media_content = gr.State({}) selected_parts = gr.State([]) conversation_id = gr.State(None) with gr.Row(): with gr.Column(scale=1): search_query_input = gr.Textbox( label="Search Query", placeholder="Enter your search query here..." ) search_type_input = gr.Radio( choices=["Title", "Content", "Author", "Keyword"], value="Keyword", label="Search By" ) keyword_filter_input = gr.Textbox( label="Filter by Keywords (comma-separated)", placeholder="ml, ai, python, etc..." ) search_button = gr.Button("Search") items_output = gr.Dropdown(label="Select Item", choices=[], interactive=True) item_mapping = gr.State({}) with gr.Row(): use_content = gr.Checkbox(label="Use Content") use_summary = gr.Checkbox(label="Use Summary") use_prompt = gr.Checkbox(label="Use Prompt") save_conversation = gr.Checkbox(label="Save Conversation", value=False, visible=True) with gr.Row(): temperature = gr.Slider(label="Temperature", minimum=0.00, maximum=1.0, step=0.05, value=0.7) with gr.Row(): conversation_search = gr.Textbox(label="Search Conversations") with gr.Row(): search_conversations_btn = gr.Button("Search Conversations") with gr.Row(): previous_conversations = gr.Dropdown(label="Select Conversation", choices=[], interactive=True) with gr.Row(): load_conversations_btn = gr.Button("Load Selected Conversation") # Refactored API selection dropdown api_endpoint = gr.Dropdown( choices=["None"] + [format_api_name(api) for api in global_api_endpoints], value=default_value, label="API for Chat Interaction (Optional)" ) api_key = gr.Textbox(label="API Key (if required)", type="password") # Initialize state variables for pagination current_page_state = gr.State(value=1) total_pages_state = gr.State(value=1) custom_prompt_checkbox = gr.Checkbox(label="Use a Custom Prompt", value=False, visible=True) preset_prompt_checkbox = gr.Checkbox(label="Use a pre-set Prompt", value=False, visible=True) with gr.Row(): # Add pagination controls preset_prompt = gr.Dropdown(label="Select Preset Prompt", choices=[], visible=False) with gr.Row(): prev_page_button = gr.Button("Previous Page", visible=False) page_display = gr.Markdown("Page 1 of X", visible=False) next_page_button = gr.Button("Next Page", visible=False) system_prompt_input = gr.Textbox(label="System Prompt", value="You are a helpful AI assistant", lines=3, visible=False) with gr.Row(): user_prompt = gr.Textbox(label="Custom Prompt", placeholder="Enter custom prompt here", lines=3, visible=False) with gr.Column(scale=2): chatbot = gr.Chatbot(height=800, elem_classes="chatbot-container") msg = gr.Textbox(label="Enter your message") submit = gr.Button("Submit") regenerate_button = gr.Button("Regenerate Last Message") token_count_display = gr.Number(label="Approximate Token Count", value=0, interactive=False) clear_chat_button = gr.Button("Clear Chat") chat_media_name = gr.Textbox(label="Custom Chat Name(optional)") save_chat_history_to_db = gr.Button("Save Chat History to DataBase") save_status = gr.Textbox(label="Save Status", interactive=False) save_chat_history_as_file = gr.Button("Save Chat History as File") download_file = gr.File(label="Download Chat History") # Restore original functionality search_button.click( fn=update_dropdown_multiple, inputs=[search_query_input, search_type_input, keyword_filter_input], outputs=[items_output, item_mapping] ) def save_chat_wrapper(history, conversation_id, media_content): file_path = save_chat_history(history, conversation_id, media_content) if file_path: return file_path, f"Chat history saved successfully as {os.path.basename(file_path)}!" else: return None, "Error saving chat history. Please check the logs and try again." save_chat_history_as_file.click( save_chat_wrapper, inputs=[chatbot, conversation_id, media_content], outputs=[download_file, save_status] ) def update_prompts(preset_name): prompts = update_user_prompt(preset_name) return ( gr.update(value=prompts["user_prompt"], visible=True), gr.update(value=prompts["system_prompt"], visible=True) ) def clear_chat(): return [], None # Return empty list for chatbot and None for conversation_id clear_chat_button.click( clear_chat, outputs=[chatbot, conversation_id] ) # Function to handle preset prompt checkbox change def on_preset_prompt_checkbox_change(is_checked): if is_checked: prompts, total_pages, current_page = list_prompts(page=1, per_page=20) page_display_text = f"Page {current_page} of {total_pages}" return ( gr.update(visible=True, interactive=True, choices=prompts), # preset_prompt gr.update(visible=True), # prev_page_button gr.update(visible=True), # next_page_button gr.update(value=page_display_text, visible=True), # page_display current_page, # current_page_state total_pages # total_pages_state ) else: return ( gr.update(visible=False, interactive=False), # preset_prompt gr.update(visible=False), # prev_page_button gr.update(visible=False), # next_page_button gr.update(visible=False), # page_display 1, # current_page_state 1 # total_pages_state ) preset_prompt_checkbox.change( fn=on_preset_prompt_checkbox_change, inputs=[preset_prompt_checkbox], outputs=[preset_prompt, prev_page_button, next_page_button, page_display, current_page_state, total_pages_state] ) def on_prev_page_click(current_page, total_pages): new_page = max(current_page - 1, 1) prompts, total_pages, current_page = list_prompts(page=new_page, per_page=20) page_display_text = f"Page {current_page} of {total_pages}" return gr.update(choices=prompts), gr.update(value=page_display_text), current_page prev_page_button.click( fn=on_prev_page_click, inputs=[current_page_state, total_pages_state], outputs=[preset_prompt, page_display, current_page_state] ) def on_next_page_click(current_page, total_pages): new_page = min(current_page + 1, total_pages) prompts, total_pages, current_page = list_prompts(page=new_page, per_page=20) page_display_text = f"Page {current_page} of {total_pages}" return gr.update(choices=prompts), gr.update(value=page_display_text), current_page next_page_button.click( fn=on_next_page_click, inputs=[current_page_state, total_pages_state], outputs=[preset_prompt, page_display, current_page_state] ) preset_prompt.change( update_prompts, inputs=[preset_prompt], outputs=[user_prompt, system_prompt_input] ) custom_prompt_checkbox.change( fn=lambda x: (gr.update(visible=x), gr.update(visible=x)), inputs=[custom_prompt_checkbox], outputs=[user_prompt, system_prompt_input] ) submit.click( chat_wrapper, inputs=[msg, chatbot, media_content, selected_parts, api_endpoint, api_key, user_prompt, conversation_id, save_conversation, temperature, system_prompt_input], outputs=[msg, chatbot, conversation_id] ).then( # Clear the message box after submission lambda x: gr.update(value=""), inputs=[chatbot], outputs=[msg] ).then( # Clear the user prompt after the first message lambda: (gr.update(value=""), gr.update(value="")), outputs=[user_prompt, system_prompt_input] ).then( lambda history: approximate_token_count(history), inputs=[chatbot], outputs=[token_count_display] ) items_output.change( update_chat_content, inputs=[items_output, use_content, use_summary, use_prompt, item_mapping], outputs=[media_content, selected_parts] ) use_content.change(update_selected_parts, inputs=[use_content, use_summary, use_prompt], outputs=[selected_parts]) use_summary.change(update_selected_parts, inputs=[use_content, use_summary, use_prompt], outputs=[selected_parts]) use_prompt.change(update_selected_parts, inputs=[use_content, use_summary, use_prompt], outputs=[selected_parts]) items_output.change(debug_output, inputs=[media_content, selected_parts], outputs=[]) search_conversations_btn.click( search_conversations, inputs=[conversation_search], outputs=[previous_conversations] ) load_conversations_btn.click( clear_chat, outputs=[chatbot, chat_history] ).then( load_conversation, inputs=[previous_conversations], outputs=[chatbot, conversation_id] ) previous_conversations.change( load_conversation, inputs=[previous_conversations], outputs=[chat_history] ) save_chat_history_as_file.click( save_chat_history, inputs=[chatbot, conversation_id], outputs=[download_file] ) save_chat_history_to_db.click( save_chat_history_to_db_wrapper, inputs=[chatbot, conversation_id, media_content, chat_media_name], outputs=[conversation_id, gr.Textbox(label="Save Status")] ) regenerate_button.click( regenerate_last_message, inputs=[chatbot, media_content, selected_parts, api_endpoint, api_key, user_prompt, temperature, system_prompt_input], outputs=[chatbot, save_status] ).then( lambda history: approximate_token_count(history), inputs=[chatbot], outputs=[token_count_display] ) def create_chat_interface_stacked(): try: default_value = None if default_api_endpoint: if default_api_endpoint in global_api_endpoints: default_value = format_api_name(default_api_endpoint) else: logging.warning(f"Default API endpoint '{default_api_endpoint}' not found in global_api_endpoints") except Exception as e: logging.error(f"Error setting default API endpoint: {str(e)}") default_value = None custom_css = """ .chatbot-container .message-wrap .message { font-size: 14px !important; } """ with gr.TabItem("Remote LLM Chat - Stacked", visible=True): gr.Markdown("# Stacked Chat") chat_history = gr.State([]) media_content = gr.State({}) selected_parts = gr.State([]) conversation_id = gr.State(None) with gr.Row(): with gr.Column(): search_query_input = gr.Textbox( label="Search Query", placeholder="Enter your search query here..." ) search_type_input = gr.Radio( choices=["Title", "Content", "Author", "Keyword"], value="Keyword", label="Search By" ) keyword_filter_input = gr.Textbox( label="Filter by Keywords (comma-separated)", placeholder="ml, ai, python, etc..." ) search_button = gr.Button("Search") items_output = gr.Dropdown(label="Select Item", choices=[], interactive=True) item_mapping = gr.State({}) with gr.Row(): use_content = gr.Checkbox(label="Use Content") use_summary = gr.Checkbox(label="Use Summary") use_prompt = gr.Checkbox(label="Use Prompt") save_conversation = gr.Checkbox(label="Save Conversation", value=False, visible=True) temp = gr.Slider(label="Temperature", minimum=0.00, maximum=1.0, step=0.05, value=0.7) with gr.Row(): conversation_search = gr.Textbox(label="Search Conversations") with gr.Row(): previous_conversations = gr.Dropdown(label="Select Conversation", choices=[], interactive=True) with gr.Row(): search_conversations_btn = gr.Button("Search Conversations") load_conversations_btn = gr.Button("Load Selected Conversation") with gr.Column(): # Refactored API selection dropdown api_endpoint = gr.Dropdown( choices=["None"] + [format_api_name(api) for api in global_api_endpoints], value=default_value, label="API for Chat Interaction (Optional)" ) api_key = gr.Textbox(label="API Key (if required)", type="password") # Initialize state variables for pagination current_page_state = gr.State(value=1) total_pages_state = gr.State(value=1) custom_prompt_checkbox = gr.Checkbox( label="Use a Custom Prompt", value=False, visible=True ) preset_prompt_checkbox = gr.Checkbox( label="Use a pre-set Prompt", value=False, visible=True ) with gr.Row(): preset_prompt = gr.Dropdown( label="Select Preset Prompt", choices=[], visible=False ) with gr.Row(): prev_page_button = gr.Button("Previous Page", visible=False) page_display = gr.Markdown("Page 1 of X", visible=False) next_page_button = gr.Button("Next Page", visible=False) system_prompt = gr.Textbox( label="System Prompt", value="You are a helpful AI assistant.", lines=4, visible=False ) user_prompt = gr.Textbox( label="Custom User Prompt", placeholder="Enter custom prompt here", lines=4, visible=False ) gr.Markdown("Scroll down for the chat window...") with gr.Row(): with gr.Column(scale=1): chatbot = gr.Chatbot(height=800, elem_classes="chatbot-container") msg = gr.Textbox(label="Enter your message") with gr.Row(): with gr.Column(): submit = gr.Button("Submit") regenerate_button = gr.Button("Regenerate Last Message") token_count_display = gr.Number(label="Approximate Token Count", value=0, interactive=False) clear_chat_button = gr.Button("Clear Chat") chat_media_name = gr.Textbox(label="Custom Chat Name(optional)", visible=True) save_chat_history_to_db = gr.Button("Save Chat History to DataBase") save_status = gr.Textbox(label="Save Status", interactive=False) save_chat_history_as_file = gr.Button("Save Chat History as File") with gr.Column(): download_file = gr.File(label="Download Chat History") # Restore original functionality search_button.click( fn=update_dropdown_multiple, inputs=[search_query_input, search_type_input, keyword_filter_input], outputs=[items_output, item_mapping] ) def search_conversations(query): try: # Use RAG search with title search if query and query.strip(): results, _, _ = search_conversations_by_keywords(title_query=query.strip()) else: results, _, _ = get_all_conversations() if not results: return gr.update(choices=[]) # Format choices to match UI conversation_options = [ (f"{conv['title']} (ID: {conv['conversation_id'][:8]})", conv['conversation_id']) for conv in results ] return gr.update(choices=conversation_options) except Exception as e: logging.error(f"Error searching conversations: {str(e)}") return gr.update(choices=[]) def load_conversation(conversation_id): if not conversation_id: return [], None try: # Use RAG load function messages, _, _ = load_chat_history(conversation_id) # Convert to chatbot history format history = [ (content, None) if role == 'user' else (None, content) for role, content in messages ] return history, conversation_id except Exception as e: logging.error(f"Error loading conversation: {str(e)}") return [], None def save_chat_history_to_db_wrapper(chatbot, conversation_id, media_content, chat_name=None): log_counter("save_chat_history_to_db_attempt") start_time = time.time() logging.info(f"Attempting to save chat history. Media content type: {type(media_content)}") try: # First check if we can access the database try: with get_db_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT 1") except sqlite3.DatabaseError as db_error: logging.error(f"Database is corrupted or inaccessible: {str(db_error)}") return conversation_id, gr.update( value="Database error: The database file appears to be corrupted. Please contact support.") # For both new and existing conversations try: if not conversation_id: title = chat_name if chat_name else "Untitled Conversation" conversation_id = start_new_conversation(title=title) logging.info(f"Created new conversation with ID: {conversation_id}") # Update existing messages delete_messages_in_conversation(conversation_id) for user_msg, assistant_msg in chatbot: if user_msg: save_message(conversation_id, "user", user_msg) if assistant_msg: save_message(conversation_id, "assistant", assistant_msg) except sqlite3.DatabaseError as db_error: logging.error(f"Database error during message save: {str(db_error)}") return conversation_id, gr.update( value="Database error: Unable to save messages. Please try again or contact support.") save_duration = time.time() - start_time log_histogram("save_chat_history_to_db_duration", save_duration) log_counter("save_chat_history_to_db_success") return conversation_id, gr.update(value="Chat history saved successfully!") except Exception as e: log_counter("save_chat_history_to_db_error", labels={"error": str(e)}) error_message = f"Failed to save chat history: {str(e)}" logging.error(error_message, exc_info=True) return conversation_id, gr.update(value=error_message) def update_prompts(preset_name): prompts = update_user_prompt(preset_name) return ( gr.update(value=prompts["user_prompt"], visible=True), gr.update(value=prompts["system_prompt"], visible=True) ) def clear_chat(): return [], None, 0 # Empty history, conversation_id, and token count clear_chat_button.click( clear_chat, outputs=[chatbot, conversation_id, token_count_display] ) # Handle custom prompt checkbox change def on_custom_prompt_checkbox_change(is_checked): return ( gr.update(visible=is_checked), gr.update(visible=is_checked) ) custom_prompt_checkbox.change( fn=on_custom_prompt_checkbox_change, inputs=[custom_prompt_checkbox], outputs=[user_prompt, system_prompt] ) # Handle preset prompt checkbox change def on_preset_prompt_checkbox_change(is_checked): if is_checked: prompts, total_pages, current_page = list_prompts(page=1, per_page=20) page_display_text = f"Page {current_page} of {total_pages}" return ( gr.update(visible=True, interactive=True, choices=prompts), # preset_prompt gr.update(visible=True), # prev_page_button gr.update(visible=True), # next_page_button gr.update(value=page_display_text, visible=True), # page_display current_page, # current_page_state total_pages # total_pages_state ) else: return ( gr.update(visible=False, interactive=False), # preset_prompt gr.update(visible=False), # prev_page_button gr.update(visible=False), # next_page_button gr.update(visible=False), # page_display 1, # current_page_state 1 # total_pages_state ) preset_prompt_checkbox.change( fn=on_preset_prompt_checkbox_change, inputs=[preset_prompt_checkbox], outputs=[preset_prompt, prev_page_button, next_page_button, page_display, current_page_state, total_pages_state] ) # Pagination button functions def on_prev_page_click(current_page, total_pages): new_page = max(current_page - 1, 1) prompts, total_pages, current_page = list_prompts(page=new_page, per_page=20) page_display_text = f"Page {current_page} of {total_pages}" return gr.update(choices=prompts), gr.update(value=page_display_text), current_page prev_page_button.click( fn=on_prev_page_click, inputs=[current_page_state, total_pages_state], outputs=[preset_prompt, page_display, current_page_state] ) def on_next_page_click(current_page, total_pages): new_page = min(current_page + 1, total_pages) prompts, total_pages, current_page = list_prompts(page=new_page, per_page=20) page_display_text = f"Page {current_page} of {total_pages}" return gr.update(choices=prompts), gr.update(value=page_display_text), current_page next_page_button.click( fn=on_next_page_click, inputs=[current_page_state, total_pages_state], outputs=[preset_prompt, page_display, current_page_state] ) # Update prompts when a preset is selected preset_prompt.change( update_prompts, inputs=[preset_prompt], outputs=[user_prompt, system_prompt] ) submit.click( chat_wrapper, inputs=[msg, chatbot, media_content, selected_parts, api_endpoint, api_key, user_prompt, conversation_id, save_conversation, temp, system_prompt], outputs=[msg, chatbot, conversation_id] ).then( lambda x: gr.update(value=""), inputs=[chatbot], outputs=[msg] ).then( lambda history: approximate_token_count(history), inputs=[chatbot], outputs=[token_count_display] ) items_output.change( update_chat_content, inputs=[items_output, use_content, use_summary, use_prompt, item_mapping], outputs=[media_content, selected_parts] ) use_content.change(update_selected_parts, inputs=[use_content, use_summary, use_prompt], outputs=[selected_parts]) use_summary.change(update_selected_parts, inputs=[use_content, use_summary, use_prompt], outputs=[selected_parts]) use_prompt.change(update_selected_parts, inputs=[use_content, use_summary, use_prompt], outputs=[selected_parts]) items_output.change(debug_output, inputs=[media_content, selected_parts], outputs=[]) search_conversations_btn.click( search_conversations, inputs=[conversation_search], outputs=[previous_conversations] ) load_conversations_btn.click( clear_chat, outputs=[chatbot, chat_history] ).then( load_conversation, inputs=[previous_conversations], outputs=[chatbot, conversation_id] ) previous_conversations.change( load_conversation, inputs=[previous_conversations], outputs=[chat_history] ) save_chat_history_as_file.click( save_chat_history, inputs=[chatbot, conversation_id], outputs=[download_file] ) save_chat_history_to_db.click( save_chat_history_to_db_wrapper, inputs=[chatbot, conversation_id, media_content, chat_media_name], outputs=[conversation_id, save_status] ) regenerate_button.click( regenerate_last_message, inputs=[chatbot, media_content, selected_parts, api_endpoint, api_key, user_prompt, temp, system_prompt], outputs=[chatbot, gr.Textbox(label="Regenerate Status")] ).then( lambda history: approximate_token_count(history), inputs=[chatbot], outputs=[token_count_display] ) def create_chat_interface_multi_api(): try: default_value = None if default_api_endpoint: if default_api_endpoint in global_api_endpoints: default_value = format_api_name(default_api_endpoint) else: logging.warning(f"Default API endpoint '{default_api_endpoint}' not found in global_api_endpoints") except Exception as e: logging.error(f"Error setting default API endpoint: {str(e)}") default_value = None custom_css = """ .chatbot-container .message-wrap .message { font-size: 14px !important; } .chat-window { height: 400px; overflow-y: auto; } """ with gr.TabItem("One Prompt - Multiple APIs", visible=True): gr.Markdown("# One Prompt but Multiple APIs Chat Interface") with gr.Row(): with gr.Column(scale=1): search_query_input = gr.Textbox(label="Search Query", placeholder="Enter your search query here...") search_type_input = gr.Radio(choices=["Title", "URL", "Keyword", "Content"], value="Title", label="Search By") search_button = gr.Button("Search") items_output = gr.Dropdown(label="Select Item", choices=[], interactive=True) item_mapping = gr.State({}) with gr.Row(): use_content = gr.Checkbox(label="Use Content") use_summary = gr.Checkbox(label="Use Summary") use_prompt = gr.Checkbox(label="Use Prompt") with gr.Column(): # Initialize state variables for pagination current_page_state = gr.State(value=1) total_pages_state = gr.State(value=1) custom_prompt_checkbox = gr.Checkbox(label="Use a Custom Prompt", value=False, visible=True) preset_prompt_checkbox = gr.Checkbox(label="Use a pre-set Prompt", value=False, visible=True) with gr.Row(): # Add pagination controls preset_prompt = gr.Dropdown(label="Select Preset Prompt", choices=[], visible=False) with gr.Row(): prev_page_button = gr.Button("Previous Page", visible=False) page_display = gr.Markdown("Page 1 of X", visible=False) next_page_button = gr.Button("Next Page", visible=False) system_prompt = gr.Textbox(label="System Prompt", value="You are a helpful AI assistant.", lines=5, visible=True) user_prompt = gr.Textbox(label="Modify Prompt (Prefixed to your message every time)", lines=5, value="", visible=True) with gr.Row(): chatbots = [] api_endpoints = [] api_keys = [] temperatures = [] regenerate_buttons = [] token_count_displays = [] for i in range(3): with gr.Column(): gr.Markdown(f"### Chat Window {i + 1}") # Refactored API selection dropdown api_endpoint = gr.Dropdown( choices=["None"] + [format_api_name(api) for api in global_api_endpoints], value=default_value, label="API for Chat Interaction (Optional)" ) api_key = gr.Textbox(label=f"API Key {i + 1} (if required)", type="password") temperature = gr.Slider(label=f"Temperature {i + 1}", minimum=0.0, maximum=1.0, step=0.05, value=0.7) chatbot = gr.Chatbot(height=800, elem_classes="chat-window") token_count_display = gr.Number(label=f"Approximate Token Count {i + 1}", value=0, interactive=False) token_count_displays.append(token_count_display) regenerate_button = gr.Button(f"Regenerate Last Message {i + 1}") chatbots.append(chatbot) api_endpoints.append(api_endpoint) api_keys.append(api_key) temperatures.append(temperature) regenerate_buttons.append(regenerate_button) with gr.Row(): msg = gr.Textbox(label="Enter your message", scale=4) submit = gr.Button("Submit", scale=1) clear_chat_button = gr.Button("Clear All Chats") # State variables chat_history = [gr.State([]) for _ in range(3)] media_content = gr.State({}) selected_parts = gr.State([]) conversation_id = gr.State(None) # Event handlers search_button.click( fn=update_dropdown, inputs=[search_query_input, search_type_input], outputs=[items_output, item_mapping] ) def update_prompts(preset_name): prompts = update_user_prompt(preset_name) return ( gr.update(value=prompts["user_prompt"], visible=True), gr.update(value=prompts["system_prompt"], visible=True) ) def on_custom_prompt_checkbox_change(is_checked): return ( gr.update(visible=is_checked), gr.update(visible=is_checked) ) custom_prompt_checkbox.change( fn=on_custom_prompt_checkbox_change, inputs=[custom_prompt_checkbox], outputs=[user_prompt, system_prompt] ) def clear_all_chats(): return [[]] * 3 + [[]] * 3 + [0] * 3 clear_chat_button.click( clear_all_chats, outputs=chatbots + chat_history + token_count_displays ) def on_preset_prompt_checkbox_change(is_checked): if is_checked: prompts, total_pages, current_page = list_prompts(page=1, per_page=10) page_display_text = f"Page {current_page} of {total_pages}" return ( gr.update(visible=True, interactive=True, choices=prompts), # preset_prompt gr.update(visible=True), # prev_page_button gr.update(visible=True), # next_page_button gr.update(value=page_display_text, visible=True), # page_display current_page, # current_page_state total_pages # total_pages_state ) else: return ( gr.update(visible=False, interactive=False), # preset_prompt gr.update(visible=False), # prev_page_button gr.update(visible=False), # next_page_button gr.update(visible=False), # page_display 1, # current_page_state 1 # total_pages_state ) preset_prompt.change(update_user_prompt, inputs=preset_prompt, outputs=user_prompt) preset_prompt_checkbox.change( fn=on_preset_prompt_checkbox_change, inputs=[preset_prompt_checkbox], outputs=[preset_prompt, prev_page_button, next_page_button, page_display, current_page_state, total_pages_state] ) def on_prev_page_click(current_page, total_pages): new_page = max(current_page - 1, 1) prompts, total_pages, current_page = list_prompts(page=new_page, per_page=10) page_display_text = f"Page {current_page} of {total_pages}" return gr.update(choices=prompts), gr.update(value=page_display_text), current_page prev_page_button.click( fn=on_prev_page_click, inputs=[current_page_state, total_pages_state], outputs=[preset_prompt, page_display, current_page_state] ) def on_next_page_click(current_page, total_pages): new_page = min(current_page + 1, total_pages) prompts, total_pages, current_page = list_prompts(page=new_page, per_page=10) page_display_text = f"Page {current_page} of {total_pages}" return gr.update(choices=prompts), gr.update(value=page_display_text), current_page next_page_button.click( fn=on_next_page_click, inputs=[current_page_state, total_pages_state], outputs=[preset_prompt, page_display, current_page_state] ) # Update prompts when a preset is selected preset_prompt.change( update_prompts, inputs=[preset_prompt], outputs=[user_prompt, system_prompt] ) def clear_all_chats(): return [[]] * 3 + [[]] * 3 + [0] * 3 clear_chat_button.click( clear_all_chats, outputs=chatbots + chat_history + token_count_displays ) def chat_wrapper_multi(message, custom_prompt, system_prompt, *args): chat_histories = args[:3] chatbots = args[3:6] api_endpoints = args[6:9] api_keys = args[9:12] temperatures = args[12:15] media_content = args[15] selected_parts = args[16] new_chat_histories = [] new_chatbots = [] for i in range(3): # Call chat_wrapper with dummy values for conversation_id and save_conversation bot_message, new_history, _ = chat_wrapper( message, chat_histories[i], media_content, selected_parts, api_endpoints[i], api_keys[i], custom_prompt, None, # None for conversation_id False, # False for save_conversation temperature=temperatures[i], system_prompt=system_prompt ) new_chatbot = chatbots[i] + [(message, bot_message)] new_chat_histories.append(new_history) new_chatbots.append(new_chatbot) return [gr.update(value="")] + new_chatbots + new_chat_histories def update_token_counts(*histories): token_counts = [] for history in histories: token_counts.append(approximate_token_count(history)) return token_counts def regenerate_last_message(chat_history, chatbot, media_content, selected_parts, api_endpoint, api_key, custom_prompt, temperature, system_prompt): if not chat_history: return chatbot, chat_history, "No messages to regenerate." last_entry = chat_history[-1] last_user_message, last_bot_message = last_entry if last_bot_message is None: return chatbot, chat_history, "The last message is not from the bot." new_history = chat_history[:-1] if not last_user_message: return chatbot[:-1], new_history, "No user message to regenerate the bot response." bot_message = chat( last_user_message, new_history, media_content, selected_parts, api_endpoint, api_key, custom_prompt, temperature, system_prompt ) new_history.append((last_user_message, bot_message)) new_chatbot = chatbot[:-1] + [(last_user_message, bot_message)] return new_chatbot, new_history, "Last message regenerated successfully." for i in range(3): regenerate_buttons[i].click( regenerate_last_message, inputs=[chat_history[i], chatbots[i], media_content, selected_parts, api_endpoints[i], api_keys[i], user_prompt, temperatures[i], system_prompt], outputs=[chatbots[i], chat_history[i], gr.Textbox(label=f"Regenerate Status {i + 1}")] ).then( lambda history: approximate_token_count(history), inputs=[chat_history[i]], outputs=[token_count_displays[i]] ) # In the create_chat_interface_multi_api function: submit.click( chat_wrapper_multi, inputs=[msg, user_prompt, system_prompt] + chat_history + chatbots + api_endpoints + api_keys + temperatures + [media_content, selected_parts], outputs=[msg] + chatbots + chat_history ).then( lambda: (gr.update(value=""), gr.update(value="")), outputs=[msg, user_prompt] ).then( update_token_counts, inputs=chat_history, outputs=token_count_displays ) items_output.change( update_chat_content, inputs=[items_output, use_content, use_summary, use_prompt, item_mapping], outputs=[media_content, selected_parts] ) for checkbox in [use_content, use_summary, use_prompt]: checkbox.change( update_selected_parts, inputs=[use_content, use_summary, use_prompt], outputs=[selected_parts] ) def create_chat_interface_four(): try: default_value = None if default_api_endpoint: if default_api_endpoint in global_api_endpoints: default_value = format_api_name(default_api_endpoint) else: logging.warning(f"Default API endpoint '{default_api_endpoint}' not found in global_api_endpoints") except Exception as e: logging.error(f"Error setting default API endpoint: {str(e)}") default_value = None custom_css = """ .chatbot-container .message-wrap .message { font-size: 14px !important; } .chat-window { height: 400px; overflow-y: auto; } """ with gr.TabItem("Four Independent API Chats", visible=True): gr.Markdown("# Four Independent API Chat Interfaces") # Initialize prompts during component creation prompts, total_pages, current_page = list_prompts(page=1, per_page=10) current_page_state = gr.State(value=current_page) total_pages_state = gr.State(value=total_pages) page_display_text = f"Page {current_page} of {total_pages}" with gr.Row(): with gr.Column(): preset_prompt = gr.Dropdown( label="Select Preset Prompt (This will be prefixed to your messages, recommend copy/pasting and then clearing the User Prompt box)", choices=prompts, visible=True ) prev_page_button = gr.Button("Previous Page", visible=True) page_display = gr.Markdown(page_display_text, visible=True) next_page_button = gr.Button("Next Page", visible=True) user_prompt = gr.Textbox( label="Modify User Prompt", lines=3 ) system_prompt = gr.Textbox( label="System Prompt", value="You are a helpful AI assistant.", lines=3 ) with gr.Column(): gr.Markdown("Scroll down for the chat windows...") chat_interfaces = [] def create_single_chat_interface(index, user_prompt_component): with gr.Column(): gr.Markdown(f"### Chat Window {index + 1}") # Refactored API selection dropdown api_endpoint = gr.Dropdown( choices=["None"] + [format_api_name(api) for api in global_api_endpoints], value=default_value, label="API for Chat Interaction (Optional)" ) api_key = gr.Textbox( label=f"API Key {index + 1} (if required)", type="password" ) temperature = gr.Slider( label=f"Temperature {index + 1}", minimum=0.0, maximum=1.0, step=0.05, value=0.7 ) chatbot = gr.Chatbot(height=400, elem_classes="chat-window") msg = gr.Textbox(label=f"Enter your message for Chat {index + 1}") submit = gr.Button(f"Submit to Chat {index + 1}") regenerate_button = gr.Button(f"Regenerate Last Message {index + 1}") token_count_display = gr.Number(label=f"Approximate Token Count {index + 1}", value=0, interactive=False) clear_chat_button = gr.Button(f"Clear Chat {index + 1}") # State to maintain chat history chat_history = gr.State([]) # Append to chat_interfaces list chat_interfaces.append({ 'api_endpoint': api_endpoint, 'api_key': api_key, 'temperature': temperature, 'chatbot': chatbot, 'msg': msg, 'submit': submit, 'regenerate_button': regenerate_button, 'clear_chat_button': clear_chat_button, 'chat_history': chat_history, 'token_count_display': token_count_display }) # Create four chat interfaces arranged in a 2x2 grid with gr.Row(): for i in range(2): with gr.Column(): for j in range(2): create_single_chat_interface(i * 2 + j, user_prompt) # Update user_prompt based on preset_prompt selection def update_prompts(preset_name): prompts = update_user_prompt(preset_name) return gr.update(value=prompts["user_prompt"]), gr.update(value=prompts["system_prompt"]) preset_prompt.change( fn=update_prompts, inputs=[preset_prompt], outputs=[user_prompt, system_prompt] ) # Pagination button functions def on_prev_page_click(current_page, total_pages): new_page = max(current_page - 1, 1) prompts, total_pages, current_page = list_prompts(page=new_page, per_page=10) page_display_text = f"Page {current_page} of {total_pages}" return ( gr.update(choices=prompts), gr.update(value=page_display_text), current_page ) prev_page_button.click( fn=on_prev_page_click, inputs=[current_page_state, total_pages_state], outputs=[preset_prompt, page_display, current_page_state] ) def on_next_page_click(current_page, total_pages): new_page = min(current_page + 1, total_pages) prompts, total_pages, current_page = list_prompts(page=new_page, per_page=10) page_display_text = f"Page {current_page} of {total_pages}" return ( gr.update(choices=prompts), gr.update(value=page_display_text), current_page ) next_page_button.click( fn=on_next_page_click, inputs=[current_page_state, total_pages_state], outputs=[preset_prompt, page_display, current_page_state] ) def chat_wrapper_single(message, chat_history, api_endpoint, api_key, temperature, user_prompt): logging.debug(f"Chat Wrapper Single - Message: {message}, Chat History: {chat_history}") new_msg, new_history, _ = chat_wrapper( message, chat_history, {}, # Empty media_content [], # Empty selected_parts api_endpoint, api_key, user_prompt, # custom_prompt None, # conversation_id False, # save_conversation temperature, # temperature system_prompt="", # system_prompt max_tokens=None, top_p=None, frequency_penalty=None, presence_penalty=None, stop_sequence=None ) if "API request failed" not in new_msg: chat_history.append((message, new_msg)) else: logging.error(f"API request failed: {new_msg}") return "", chat_history, chat_history def regenerate_last_message(chat_history, api_endpoint, api_key, temperature, user_prompt): if not chat_history: return chat_history, chat_history, "No messages to regenerate." last_user_message, _ = chat_history[-1] new_msg, new_history, _ = chat_wrapper( last_user_message, chat_history[:-1], {}, # Empty media_content [], # Empty selected_parts api_endpoint, api_key, user_prompt, # custom_prompt None, # conversation_id False, # save_conversation temperature, # temperature system_prompt="", # system_prompt max_tokens=None, top_p=None, frequency_penalty=None, presence_penalty=None, stop_sequence=None ) if "API request failed" not in new_msg: new_history.append((last_user_message, new_msg)) return new_history, new_history, "Last message regenerated successfully." else: logging.error(f"API request failed during regeneration: {new_msg}") return chat_history, chat_history, f"Failed to regenerate: {new_msg}" # Attach click events for each chat interface for interface in chat_interfaces: interface['submit'].click( chat_wrapper_single, inputs=[ interface['msg'], interface['chat_history'], interface['api_endpoint'], interface['api_key'], interface['temperature'], user_prompt ], outputs=[ interface['msg'], interface['chatbot'], interface['chat_history'] ] ).then( lambda history: approximate_token_count(history), inputs=[interface['chat_history']], outputs=[interface['token_count_display']] ) interface['regenerate_button'].click( regenerate_last_message, inputs=[ interface['chat_history'], interface['api_endpoint'], interface['api_key'], interface['temperature'], user_prompt ], outputs=[ interface['chatbot'], interface['chat_history'], gr.Textbox(label="Regenerate Status") ] ).then( lambda history: approximate_token_count(history), inputs=[interface['chat_history']], outputs=[interface['token_count_display']] ) def clear_chat_single(): return [], [], 0 interface['clear_chat_button'].click( clear_chat_single, outputs=[interface['chatbot'], interface['chat_history'], interface['token_count_display']] ) def chat_wrapper_single(message, chat_history, chatbot, api_endpoint, api_key, temperature, media_content, selected_parts, conversation_id, save_conversation, user_prompt): new_msg, new_history, new_conv_id = chat_wrapper( message, chat_history, media_content, selected_parts, api_endpoint, api_key, user_prompt, conversation_id, save_conversation, temperature, system_prompt="" ) if new_msg: updated_chatbot = chatbot + [(message, new_msg)] else: updated_chatbot = chatbot return new_msg, updated_chatbot, new_history, new_conv_id # Mock function to simulate LLM processing def process_with_llm(workflow, context, prompt, api_endpoint, api_key): api_key_snippet = api_key[:5] + "..." if api_key else "Not provided" return f"LLM output using {api_endpoint} (API Key: {api_key_snippet}) for {workflow} with context: {context[:30]}... and prompt: {prompt[:30]}..." # # End of Chat_ui.py #######################################################################################################################