Spaces:
Sleeping
Sleeping
import gradio as gr | |
from chatbot_simulator import ChatbotSimulation | |
from task_specific_data_population import DataPopulation | |
import re | |
import os | |
openai_api_key = os.getenv("OPENAI_API_KEY") | |
class AppSimulator: | |
def __init__(self, openai_api_key): | |
self.simulation = None | |
self.openai_api_key = openai_api_key | |
def initialize_simulator(self, task, app_name, sitemap, progress=gr.Progress(track_tqdm=True)): | |
"""Initialize the simulator with retries and elapsed time tracking.""" | |
retry_count = 0 | |
max_retries = 50 | |
while retry_count < max_retries: | |
try: | |
# Display current attempt with progress bar | |
progress.update(f"Initializing... Attempt {retry_count + 1}/{max_retries}") | |
# Process data and initialize the simulation | |
data_population = DataPopulation(api_key=self.openai_api_key) | |
sitemap_data, page_details, user_state = data_population.process_data(task, sitemap) | |
self.simulation = ChatbotSimulation( | |
site_map=sitemap_data, | |
page_details=page_details, | |
user_state=user_state, | |
task=task, | |
app_name=app_name, | |
log_location=f'conversation_log_{app_name}_human.txt', | |
openai_api_key=self.openai_api_key, | |
agent='human' | |
) | |
# Successful initialization | |
initial_message = self.simulation.start_conversation() | |
progress.update("Initialization Successful") | |
return initial_message # Return the initial assistant message for chat | |
except Exception as e: | |
retry_count += 1 | |
print(f"Attempt {retry_count}/{max_retries} failed: {e}") | |
# If all retries failed | |
progress.update("Failed to initialize simulator after multiple retries.") | |
return "Initialization failed." # Error message for chat | |
def chat_interaction(self, user_input, history): | |
"""Handle one round of conversation.""" | |
return self.simulation.one_conversation_round(user_input) | |
# Initialize the simulator | |
simulator_app = AppSimulator(openai_api_key=openai_api_key) | |
def is_valid_input(user_input): | |
"""Validate user input format.""" | |
pattern = r"^\d+\.\s+.*" | |
return bool(re.match(pattern, user_input)) | |
def chat(user_input, history): | |
"""Chat handler that validates input and interacts with the simulator.""" | |
if is_valid_input(user_input): | |
valid_response = simulator_app.chat_interaction(user_input, history) | |
return valid_response | |
else: | |
invalid_response = "Invalid input. Please use the format: <Number>. <Description>: query" | |
return invalid_response | |
# Gradio Interface using ChatInterface | |
with gr.Blocks(fill_height=True, theme=gr.themes.Ocean()) as demo: | |
gr.Markdown("## Simulator Setup") | |
# Input fields for initialization | |
task_input = gr.Textbox(label="Task", placeholder="Describe your task...") | |
app_name_input = gr.Textbox(label="App Name", placeholder="Enter the app name...") | |
sitemap_input = gr.Textbox(label="Sitemap", placeholder="Enter the Hugging Face link to sitemap...") | |
initialize_button = gr.Button("Initialize Simulator") | |
# Status block to display initialization progress with elapsed time | |
status = gr.Textbox(label="Status", interactive=False) | |
# Chat interface to handle user interactions | |
chat_interface = gr.ChatInterface(fn=chat, type='messages') | |
# Define the callback function to initialize the simulator and update status | |
def initialize_and_start_chat(task, app_name, sitemap): | |
return simulator_app.initialize_simulator(task, app_name, sitemap) # Use progress tracking | |
# Set up the button click to initialize simulator and update status only | |
initialize_button.click( | |
fn=initialize_and_start_chat, | |
inputs=[task_input, app_name_input, sitemap_input], | |
outputs=status # Update only the status block | |
) | |
# Launch the app | |
demo.launch() | |