update app simulator

app.py

@@ -1,8 +1,9 @@
 import gradio as gr
 from chatbot_simulator import ChatbotSimulation
 from task_specific_data_population import DataPopulation
-import re
-import os
+from datasets import load_dataset
+import json_repair
+import random
 
 openai_api_key = os.getenv("OPENAI_API_KEY")
 
@@ -12,42 +13,46 @@ class AppSimulator:
         self.simulation = None
         self.openai_api_key = openai_api_key
 
-    def initialize_simulator(self, task, app_name, sitemap, progress=gr.Progress(track_tqdm=True)):
+    def initialize_simulator(self, sitemap_url, 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
+        synthetic_sitemap = load_dataset(sitemap_url, "sitemap", split='train')
+        app_name, sitemap, page_details, user_state, system_data = None, None, None, None, None
+        for row in synthetic_sitemap:
+            if row['name'] == 'app_name':
+                app_name = row['value']  # Use `eval` to convert the string to a list
+            elif row['name'] == 'sitemap':
+                sitemap = json_repair.loads(row['value'])
+            elif row['name'] == 'page_details':
+                page_details = json_repair.loads(row['value'])
+            elif row['name'] == 'user_state':
+                user_state = json_repair.loads(row['value'])
+            elif row['name'] == 'system_data':
+                system_data = json_repair.loads(row['value'])
+
+        synthetic_tasks = load_dataset(sitemap_url, "tasks", split='train')
+        random_index = random.randint(0, len(synthetic_tasks) - 1)  # Generate a random index
+        random_row = synthetic_tasks[random_index]
+        task = random_row['tasks']
+        solution = random_row['steps']
+        user_data = random_row['attributes']['user_data']
+
+        self.simulation = ChatbotSimulation(
+            app_name=app_name,
+            site_map=sitemap,
+            page_details=page_details,
+            user_state=user_state,
+            system_data=system_data,
+            user_data=user_data,
+            task=task,
+            solution=solution,
+            log_location=f'conversation_log_{app_name}.txt',
+            openai_api_key=openai_api_key,
+            agent='llm'
+        )
+
+        initial_message = self.simulation.start_conversation()
+        progress.update("Initialization Successful")
+        return initial_message  # Return the initial assistant message for chat
 
     def chat_interaction(self, user_input, history):
         """Handle one round of conversation."""
@@ -57,19 +62,11 @@ class AppSimulator:
 # 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
+    response = simulator_app.chat_interaction(user_input, history)
+    return response
 
 
 # Gradio Interface using ChatInterface
@@ -77,8 +74,6 @@ with gr.Blocks(fill_height=True) 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... (eg. DoorDash)")
     sitemap_input = gr.Textbox(label="Sitemap", placeholder="Enter the Hugging Face link to sitemap... (eg.jjz5463/DoorDash_synthetic_sitemap)")
     initialize_button = gr.Button("Initialize Simulator")
 
@@ -89,13 +84,13 @@ with gr.Blocks(fill_height=True) as demo:
     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
+    def initialize_and_start_chat(sitemap):
+        return simulator_app.initialize_simulator(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],
+        inputs=[sitemap_input],
         outputs=status  # Update only the status block
     )
 

chatbot_simulator.py

@@ -3,20 +3,26 @@ import json_repair
 from transformers import AutoTokenizer
 from openai import RateLimitError
 import time
+from prompts import *
+import re
 
 
 class ChatbotSimulation:
-    def __init__(self, site_map, page_details, user_state, task,
-                 app_name, log_location, openai_api_key, agent='human',
+    def __init__(self, app_name, site_map, page_details, user_state, system_data, user_data, task, solution,
+                 log_location, openai_api_key, agent='human',
                  max_steps=50, max_tokens=8192, buffer_tokens=500):
+        self.app_name = app_name
         self.sitemap = site_map
         self.page_details = page_details
         self.user_state = user_state
         self.user_state['current_page'] = 'Home'  # Initialize current page
-        self.user_state['last_page'] = 'Home'
         self.user_state['task_completed'] = 'False'
+        self.user_state['back'] = 'False'
+        self.system_data = system_data
+        self.user_data = user_data
         self.task = task
-        self.app_name = app_name
+        self.solution = solution
+
         self.log_location = log_location
         self.agent = agent.lower()
         if self.agent not in ['human', 'llm']:
@@ -28,11 +34,14 @@ class ChatbotSimulation:
         self.prompt_count = 0
         self.client = OpenAI(api_key=openai_api_key)
         self.actions = []
-        self.tokenizer = AutoTokenizer.from_pretrained("gpt2")
+        self.tokenizer = AutoTokenizer.from_pretrained("gpt2", clean_up_tokenization_spaces=True)
+
+        # back button
+        self.page_history = ['Home']
 
     def _get_page_uid(self, page_name):
         """Retrieve the UID of the given page from the sitemap."""
-        return self.sitemap['pages'].get(page_name, {}).get('uid')
+        return self.sitemap.get(page_name, {}).get('uid')
 
     def _get_page_details(self, page_name):
         """Retrieve the page details using its UID."""
@@ -41,43 +50,22 @@ class ChatbotSimulation:
 
     def _generate_system_prompt(self):
         """Create a dynamic system prompt based on the current state."""
-        current_page = self.user_state['current_page']
-        last_page = self.user_state['last_page']
+        #current_page = self.user_state['current_page']
+        #last_page = self.user_state['last_page']
+        current_page = self.page_history[-1] if len(self.page_history) >= 1 else "Home"
+        last_page = self.page_history[-2] if len(self.page_history) > 1 else "Home"
         page_info = self._get_page_details(current_page)
 
-        return f"""
-You are a text-based simulator of {self.app_name} app. 
-You are interacting with a user. User's task is: {self.task}.
-User's last page was {last_page} and the user have taken actions: {self.actions}.
-After action, user is currently on the {current_page} page. 
-Current user state: {self.user_state}.
-
-Page Information:
-- **If the user requests page you do not possess** (such as a list of restaurants, menus, or similar details), 
-you are permitted to create plausible and relevant information to fulfill the request. 
-Present this fabricated information convincingly as if it were real data.
-{page_info}
-
-- **Features**: Represent available options the user can select on this page.
-- **User Data**: Represents user-specific data accessible on this page.
-
-Provide instructions or request input from the user. If the user provides an invalid action, respond with:
-"Invalid action. Please select a valid option."
-
-### Instruction Format:
-<if actions is non-empty: You have successfully done actions[-1]> You are at the {current_page} page. You have the following options:
-1. Feature 1
-2. Feature 2
-3. Feature 3
-4. Feature 4
-
-Please enter your choice as 'Number. Description'. If you have a query, enter as 'Number. Description: query'
-
-Rules:
-- Be sure to display all options that is available in features. 
-- Be robotic and emotionless. Avoid offering any advice to the user.
-- Make sure there is always a back to last_page and Home page button. 
-                """
+        return get_system_prompt(app_name=self.app_name,
+                                 system_data=self.system_data,
+                                 task=self.task,
+                                 user_data=self.user_data,
+                                 current_page=current_page,
+                                 last_page=last_page,
+                                 actions=self.actions,
+                                 user_state=self.user_state,
+                                 page_info=page_info
+                                 )
 
     def _get_openai_response(self, prompt):
         """Fetch response from OpenAI API."""
@@ -120,81 +108,139 @@ Rules:
     def one_conversation_round(self, user_input):
         """Conduct one round of conversation between the user and the assistant."""
         # User provides input
+        valid_input = self._is_valid_input(user_input)
+        if valid_input[0]:
+            pass
+        else:
+            return f"\n{self.app_name}: Invalid input. {valid_input[1]}"
+
         self.actions.append(user_input + f'on {self.user_state["current_page"]} page')
         self.conversation.append({"role": "user", "content": user_input})
         self.prompt_count += 1
 
         # Update user state using GPT's response
-        update_prompt = f"""
-If user takes action '{user_input}' on {self.user_state['current_page']} page, which page will they move to?
-Recall user's task: {self.task}
-Update the user_state dictionary based on user's last action:
-Current user_state: {self.user_state}
-Sitemap: {self.sitemap}
-
-Instructions:
-1. If the 'current_page' has changed, update it to a page from the sitemap.
-2. If the task is finished, update 'task_completed' to True. Otherwise, leave it as False.
-3. If no updates are needed, return the user state exactly as provided, without modification.
-
-Important:
-- Ensure 'current_page' and 'task_completed' are keys in the returned dictionary.
-- Return only the dictionary without additional output or wrapping.
-
-Example Output Format:
-{{
-    'current_page': 'Home',
-    'last_page': 'Home',
-    'task_completed': 'False',
-}}
-                """
+        current_page = self.page_history[-1] if len(self.page_history) >= 1 else "Home"
+        update_prompt = get_user_state_update_prompt(user_input=user_input,
+                                                     current_page=current_page,
+                                                     task=self.task,
+                                                     solution=self.solution,
+                                                     user_state=self.user_state,
+                                                     sitemap=self.sitemap)
 
         self.conversation.append({"role": "assistant", "content": update_prompt})
-        updated_state = self._get_openai_response(self.conversation)
+        updated_state = self._get_openai_response(self.conversation).split("UPDATED", 1)[1].strip()
         self.conversation.pop(-1) ## update prompt don't have to stay in conversation history
 
         # Parse and update the user state
         updated_state = json_repair.loads(updated_state)
 
-        required_keys = {'current_page', 'last_page', 'task_completed'}
-        while not isinstance(updated_state, dict) or not required_keys.issubset(updated_state.keys()):
-            transform_prompt = f""""
-            Transform {updated_state} to a properly formate JSON file.
+        # format forcing of updated state
+        required_keys = {'current_page', 'task_completed', 'back'}
+        # Ensure `updated_state` is a dictionary
+        while not isinstance(updated_state, dict):
+            transform_prompt = f"""
+            Transform {updated_state} to a properly formatted JSON file.
             Example Output Format:
             {{
-                'current_page': 'Home',
-                'last_page': 'Home',
-                'task_completed': 'False',
+               'current_page': 'Home',
+               'task_completed': False,
+               'back': False
             }}
             """
             updated_state = self._get_openai_response([{"role": "system", "content": transform_prompt}])
             updated_state = json_repair.loads(updated_state)
+        # Manually add missing required keys
+        for key in required_keys:
+            if key not in updated_state:
+                if key == 'current_page':
+                    updated_state[key] = self.page_history[-1] if len(self.page_history) >= 1 else "Home"
+                else:
+                    updated_state[key] = False
 
         try:
-            if updated_state['task_completed'].lower() == 'true':
+            if str(updated_state['task_completed']).lower() == 'true':
                 return f"Task completed! You took {self.prompt_count} steps."
         except:
             updated_state['task_completed'] = 'False'
 
         self.user_state = updated_state
+        if str(updated_state['back']).lower() == 'false':
+            self.page_history.append(updated_state['current_page'])
+        elif self.page_history:
+            self.page_history.pop()
 
-        #self.conversation.clear()
-        system_prompt = self._generate_system_prompt()
-
-        # GPT generates the page instructions
         ## no need to store old system prompt while we get a new one
         self.conversation = [entry for entry in self.conversation if entry["role"] != "system"]
+        system_prompt = self._generate_system_prompt()
+        # GPT generates the page instructions
         self.conversation.append({"role": "system", "content": system_prompt})
-
         gpt_instruction = self._get_openai_response(self.conversation)
         self.conversation.append({"role": "assistant", "content": gpt_instruction})
         return gpt_instruction
 
     def start_conversation(self):
-        greeting = f'\n Welcome to {self.app_name} simulator! Your task is: {self.task}. \n'
+        greeting = f'\nWelcome to {self.app_name} simulator! Your task is: {self.task}. \n'
         system_prompt = self._generate_system_prompt()
         # GPT generates the page instructions
         self.conversation.append({"role": "system", "content": system_prompt})
         gpt_instruction = self._get_openai_response(self.conversation)
         self.conversation.append({"role": "assistant", "content": gpt_instruction})
         return greeting + gpt_instruction
+
+    def _extract_buttons(self):
+        """Extract buttons and their action types from the latest conversation if role is 'assistant'."""
+        # Get the last message
+        last_message = self.conversation[-1]
+
+        # Ensure the role of the last message is 'assistant'
+        if last_message.get("role") != "assistant":
+            return {}
+
+        # Extract the content of the last message
+        message_content = last_message.get("content", "")
+
+        # Make the split case-insensitive by searching for the phrase with re.IGNORECASE
+        options_split = re.split(r"you have the following options:", message_content, flags=re.IGNORECASE)
+
+        # If the split doesn't produce at least two parts, return an empty dictionary
+        if len(options_split) < 2:
+            return {}
+
+        # Extract button definitions from the second part of the split content
+        button_section = options_split[1]
+        pattern = r"\d+\.\s+(.*?):\s+([a-zA-Z_]+)"
+        buttons = re.findall(pattern, button_section)
+
+        # Construct the dictionary with button names as keys and action types as values
+        return {name.strip().lower(): action_type.strip().lower() for name, action_type in buttons}
+
+    def _is_valid_input(self, user_input):
+        """Validate user input format."""
+        valid_buttons = self._extract_buttons()
+
+        # Validate input format
+        pattern = r"^(?P<action_type>\w+)\((?P<button_name>[^,]+)(?:,\s*(?P<query>.+))?\)$"
+        match = re.match(pattern, user_input)
+
+        if not match:
+            return [False, "Your input doesn't match the format: action_type(button name), OR if type, use type(button name, query)"]
+
+        # Extract parsed components
+        action_type = match.group("action_type").lower()
+        button_name = match.group("button_name").strip().lower()
+        query = match.group("query")  # Optional query for `type`
+
+        # Validate button name and action type
+        if button_name not in valid_buttons:
+            return [False,
+                    "Invalid Button name! Recall: Each button is in the format: `number. button name: action_type`"]  # Button name must match exactly (case insensitive)
+        if action_type != valid_buttons[button_name]:
+            return [False,
+                    "Invalid action type! Recall: Each button is in the format: `number. button name: action_type`"]  # Action type must match the button's specified type
+        if action_type == "type" and query is None:
+            return [False,
+                    "Missing Query for action type 'type'! Recall: use the format: `type(button name, query)`"]  # `type` action requires a query
+        if action_type != "type" and query is not None:
+            return [False,
+                    "Non-`type` action_type cannot take query!"]  # Non-`type` actions must not have a query
+        return [True, 'Pass']

prompts.py

@@ -0,0 +1,89 @@
+
+def get_system_prompt(app_name, system_data, task, user_data, current_page, last_page, actions, user_state, page_info):
+    system_prompt = f"""
+You are a text-based {app_name}-like app simulator game.  
+The {app_name} app contains the following system data: {system_data}.  
+You are interacting with a user whose task is: {task}.  
+User's data: {user_data}.  
+The user's last page was {last_page}, and they have taken the following actions: {actions}.  
+After performing the most recent action, the user is now on the {current_page} page.  
+Current user state: {user_state}.  
+
+Details and buttons on the current page:  
+{page_info}  
+
+Each button on the page has a `name` and an `action_type`.  
+
+### Instruction Format:
+<If `actions` is non-empty: "You have successfully performed actions[-1]">  
+<If current_page == log in & user_state['logged_in'] == False: You need to log in first. Your credentials are: Username: user_data['credentials']['user_id'] Password: user_data['credentials']['password']>
+<If current_page == log in & user_state['logged_in'] == True: You have successfully logged in as user_data['credentials']['user_id']>
+You are currently on the **{current_page}** page. You have the following options:  
+1. option 1: action type  
+2. option 2: action type  
+3. option 3: action type  
+4. option 4: action type  
+5. Back to last page: click
+6. Back to Home page: click
+7. <If user_state['logged_in'] == False, display Log In: click to take user to log in page>
+8. Do nothing: noop
+
+Each button is in the format: `number. button name: action_type`.
+Please provide your choice in the format: `action_type(button name)`.  
+If you have a query, use the format: `type(button name, query)`.  
+Please provide only one choice per turn.
+
+### **Rules**:
+1. Display all available options from the buttons on the page.  
+2. Remain robotic and emotionless. Avoid offering any advice or opinions to the user.  
+3. Ensure there is always a button to navigate back to the **{last_page}** and a button to go to the **Home page**.
+4. Ensure there is always a button that allow user to do nothing.
+5. Ensure the action type is a single word, such as ‘type’ or ‘click.’ If the action type is not a single word, convert it to a single word. For example, type(SUMMER, apply) should be transformed to type.
+6. Group similar action types under ‘type’ and display them as ‘type’ (e.g., display ‘input’ or ‘search’ as ‘type’).
+    """
+    return system_prompt
+
+
+def get_user_state_update_prompt(user_input, current_page, task, solution, user_state, sitemap):
+    structure = """
+{
+   'current_page': 'Home',
+   'task_completed': False,
+   'back': False
+}
+    """
+    update_prompt = f"""
+You will now update the user state.  
+The user takes the action '{user_input}' on the {current_page} page. Determine which page the user will move to next.  
+The next page must from the sitemap: {sitemap}
+Recall the user's task: {task}. Solution to the user's task: {solution}.  
+Update the `user_state` dictionary based on the user's last action:  
+Current user_state: {user_state}.  
+
+### Instructions:
+1. If the 'current_page' has changed, update it to a valid page from the sitemap.
+2. If the task is completed, update 'task_completed' to `True`. Otherwise, leave it as `False`.
+3. If no updates are needed, return the `user_state` exactly as provided, without any changes.
+
+### Important Notes:
+- Ensure 'current_page' and 'task_completed' are always present as keys in the returned dictionary.
+- If the user go back a page from current page, set 'back' to `True`. Otherwise, it should remain `False`.
+- Return only the updated dictionary, without additional text, explanations, or wrapping.
+
+### Output Format:
+1. Start with "REASON": Explain which fields in `user_state` need to be updated and why.
+2. End with "UPDATED": Provide only the updated `user_state` dictionary. Follow the input user_state schema structure exactly without modifying its organization. Do not include explanations or any additional text after updated user state.
+    """
+    return update_prompt
+
+
+def get_agent_prompt(app_name, task, conversation):
+    agent_prompt = f"""
+    Imagine you are an agent navigate through the {app_name} environment.
+    Your overarching task is: {task}. You may have done some part of the task, or none at all.
+    You will have access to all of your previous actions in the environment, as well as the last message from the assistant giving the current state of the environment.
+    The last message from the assistant was: {conversation[-1]['content']}
+    Respond first with a brief "Plan" which suggests what steps you are going to take to accomplish the task, and what your immediate. 
+    Then generate an "Action" which is the immediate next step you can take.
+            """
+    return agent_prompt