oceansweep's picture
Upload 169 files
c5b0bb7 verified
raw
history blame
65.1 kB
# 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
#######################################################################################################################