Delete browser_agent.py

browser_agent.py

@@ -1,282 +0,0 @@
-import logging
-import os
-from typing import Any, List, Tuple
-
-from browsergym.core.action.highlevel import HighLevelActionSet
-from browsergym.utils.obs import (
-    flatten_axtree_to_str,
-    flatten_dom_to_str,
-    prune_html,
-)
-from browsergym.experiments import Agent
-
-from utils import remove_inline_comments_safe, image_to_jpg_base64_url
-
-import openai
-
-
-logger = logging.getLogger(__name__)
-
-openai.api_key = os.getenv("OPENAI_API_KEY")
-
-
-
-class BrowserAgent(Agent):
-    def obs_preprocessor(self, obs: dict) -> dict:
-        return {
-            "chat_messages": obs["chat_messages"],
-            "som_screenshot": obs["som_screenshot"],
-            "goal_object": obs["goal_object"],
-            "last_action": obs["last_action"],
-            "last_action_error": obs["last_action_error"],
-            "open_pages_urls": obs["open_pages_urls"],
-            "open_pages_titles": obs["open_pages_titles"],
-            "active_page_index": obs["active_page_index"],
-            "axtree_txt": flatten_axtree_to_str(obs["axtree_object"], filter_visible_only=True, extra_properties=obs['extra_element_properties'], filter_som_only=True),
-            "pruned_html": prune_html(flatten_dom_to_str(obs["dom_object"])),
-        }
-
-    def __init__(self, model_name: str = "gpt-4o", use_html: bool = False, use_axtree: bool = True, use_screenshot: bool = False):
-        super().__init__()
-        logger.info(f"Initializing BrowserAgent with model: {model_name}")
-        logger.info(f"Observation space: HTML={use_html}, AXTree={use_axtree}, Screenshot={use_screenshot}")
-        
-        self.model_name = model_name
-        self.use_html = use_html
-        self.use_axtree = use_axtree
-        self.use_screenshot = use_screenshot
-        
-        if not (use_html or use_axtree):
-            raise ValueError("Either use_html or use_axtree must be set to True.")
-        
-        self.openai_client = openai.OpenAI()
-        
-        self.action_set = HighLevelActionSet(
-            subsets=["chat", "tab", "nav", "bid", "infeas"],
-            strict=False,
-            multiaction=False,
-            demo_mode="default"
-        )
-        self.action_history = []
-
-    def get_action(self, obs: dict) -> tuple[str, dict]:
-        logger.debug("Preparing action request")
-        
-        system_msgs = [{
-            "type": "text",
-            "text": """\
-# Instructions
-
-You are a UI Assistant, your goal is to help the user perform tasks using a web browser. You can
-communicate with the user via a chat, to which the user gives you instructions and to which you
-can send back messages. You have access to a web browser that both you and the user can see,
-and with which only you can interact via specific commands.
-
-Review the instructions from the user, the current state of the page and all other information
-to find the best possible next action to accomplish your goal. Your answer will be interpreted
-and executed by a program, make sure to follow the formatting instructions.
-"""
-        }]
-
-        user_msgs = []
-        
-        # Add chat messages
-        user_msgs.append({
-            "type": "text",
-            "text": "# Chat Messages\n"
-        })
-        for msg in obs["chat_messages"]:
-            if msg["role"] in ("user", "assistant", "infeasible"):
-                user_msgs.append({
-                    "type": "text",
-                    "text": f"- [{msg['role']}] {msg['message']}\n"
-                })
-                logger.debug(f"Added chat message: [{msg['role']}] {msg['message']}")
-            elif msg["role"] == "user_image":
-                user_msgs.append({"type": "image_url", "image_url": msg["message"]})
-                logger.debug("Added user image message")
-
-        # Add open tabs info
-        user_msgs.append({
-            "type": "text",
-            "text": "# Currently open tabs\n"
-        })
-        for page_index, (page_url, page_title) in enumerate(
-            zip(obs["open_pages_urls"], obs["open_pages_titles"])
-        ):
-            user_msgs.append({
-                "type": "text",
-                "text": f"""\
-Tab {page_index}{" (active tab)" if page_index == obs["active_page_index"] else ""}
-  Title: {page_title}
-  URL: {page_url}
-"""
-            })
-            logger.debug(f"Added tab info: {page_title} ({page_url})")
-
-        # Add accessibility tree if enabled
-        if self.use_axtree:
-            user_msgs.append({
-                "type": "text",
-                "text": f"""\
-# Current page Accessibility Tree
-
-{obs["axtree_txt"]}
-
-"""
-            })
-            logger.debug("Added accessibility tree")
-
-        # Add HTML if enabled
-        if self.use_html:
-            user_msgs.append({
-                "type": "text",
-                "text": f"""\
-# Current page DOM
-
-{obs["pruned_html"]}
-
-"""
-            })
-            logger.debug("Added HTML DOM")
-
-        # Add screenshot if enabled
-        if self.use_screenshot:
-            user_msgs.append({
-                "type": "text",
-                "text": "# Current page Screenshot\n"
-            })
-            user_msgs.append({
-                "type": "image_url",
-                "image_url": {
-                    "url": image_to_jpg_base64_url(obs["som_screenshot"]),
-                    "detail": "auto"
-                }
-            })
-            logger.debug("Added screenshot")
-
-        # Add action space description
-        user_msgs.append({
-            "type": "text",
-            "text": f"""\
-# Action Space
-
-{self.action_set.describe(with_long_description=False, with_examples=True)}
-
-Here are examples of actions with chain-of-thought reasoning:
-
-I now need to click on the Submit button to send the form. I will use the click action on the button, which has bid 12.
-```click("12")```
-
-I found the information requested by the user, I will send it to the chat.
-```send_msg_to_user("The price for a 15\\" laptop is 1499 USD.")```
-
-"""
-        })
-
-        # Add action history and errors
-        if self.action_history:
-            user_msgs.append({
-                "type": "text",
-                "text": "# History of past actions\n"
-            })
-            for action in self.action_history:
-                user_msgs.append({
-                    "type": "text",
-                    "text": f"\n{action}\n"
-                })
-                logger.debug(f"Added past action: {action}")
-
-            if obs["last_action_error"]:
-                user_msgs.append({
-                    "type": "text",
-                    "text": f"""\
-# Error message from last action
-
-{obs["last_action_error"]}
-
-"""
-                })
-                logger.warning(f"Last action error: {obs['last_action_error']}")
-
-        # Ask for next action
-        user_msgs.append({
-            "type": "text",
-            "text": """\
-# Next action
-
-You will now think step by step and produce your next best action. Reflect on your past actions, any resulting error message, and the current state of the page before deciding on your next action.
-Note: You might use 'goto' action if you're in a blank page.
-"""
-        })
-
-        # Log the full prompt for debugging
-        prompt_text_strings = []
-        for message in system_msgs + user_msgs:
-            match message["type"]:
-                case "text":
-                    prompt_text_strings.append(message["text"])
-                case "image_url":
-                    image_url = message["image_url"]
-                    if isinstance(message["image_url"], dict):
-                        image_url = image_url["url"]
-                    if image_url.startswith("data:image"):
-                        prompt_text_strings.append(
-                            "image_url: " + image_url[:30] + "... (truncated)"
-                        )
-                    else:
-                        prompt_text_strings.append("image_url: " + image_url)
-                case _:
-                    raise ValueError(
-                        f"Unknown message type {repr(message['type'])} in the task goal."
-                    )
-        full_prompt_txt = "\n".join(prompt_text_strings)
-        logger.debug(full_prompt_txt)
-
-        # Query OpenAI model
-        logger.info("Sending request to OpenAI")
-        response = self.openai_client.chat.completions.create(
-            model=self.model_name,
-            messages=[
-                {"role": "system", "content": system_msgs},
-                {"role": "user", "content": user_msgs}
-            ],
-            n=20,
-            temperature=0.8
-        )
-        parses = []
-        for i, choice in enumerate(response.choices):
-            response = choice.message.content
-            try:
-                parses.append({
-                    'response': response,
-                    'thought': response.split('```')[0].strip(),
-                    'action': remove_inline_comments_safe(response.split('```')[1].strip('`').strip().strip('`').strip()),
-                })
-            except Exception as e:
-                logger.error(f"Error parsing action: {e}")
-                logger.error(f"Response: {response}")
-                logger.error(f"Choice: {choice}")
-                logger.error(f"Index: {i}")
-                logger.error(f"Response: {response}")
-                
-        candidates = self.get_top_k_actions(parses)
-        logger.info(f"Received action from OpenAI: {[cand['action'] for cand in candidates]}")
-        return candidates, {}
-
-    def get_top_k_actions(self, parses, k=3):
-        count_dict = {}
-        action_to_parsed = {}
-        for parsed in parses:
-            action = parsed["action"]
-            if action in count_dict:
-                count_dict[action] += 1
-            else:
-                count_dict[action] = 1
-                action_to_parsed[action] = parsed.copy()
-        
-        # Get the top_k most frequent actions
-        sorted_actions = sorted(count_dict.items(), key=lambda x: x[1], reverse=True)
-        top_k_actions = [action_to_parsed[action] for action, _ in sorted_actions[:k]]
-        
-        return top_k_actions