Prompt used for os environment benchmark?

[cjfcsjt]

What is the prompt used for the os environment benchmark? (e.g., Android world, OSworld)

[LukeForeverYoung]

The prompt used for AndroidWorld can be found in Qwen2.5-VL/cookbooks/mobile_agent.ipynb.

Notably, to fit the action space in AndroidWorld evaluation, we add a new action answer. Therefore, there are some modifications in the system prompt:

1. Add * `answer`: Output the answer. after the * `type`: Input the specified text into the activated input box. and before the * `system_button`: Press the system button.

2. Add answer in the enum part.

3. Modify the defination of the text paramter to :
   "text": {"description": "Required only by `action=key`, `action=type`, `action=answer`, and `action=open`.", "type": "string"}

[Timothyxxx]

Hi @cjfcsjt

Thank you for raising this question!

The following script is my qwen25vl_agent.py implementation for OSWorld. You can place this file under the OSWorld/mm_agents directory in your project.

---

import base64
import json
import logging
import os
import re
import tempfile
import time
from http import HTTPStatus
from io import BytesIO
from typing import Dict, List, Tuple

import backoff
import openai
import requests
from PIL import Image
from google.api_core.exceptions import (
    InvalidArgument,
    ResourceExhausted,
    InternalServerError,
    BadRequest,
)
from requests.exceptions import SSLError

logger = None


# from openai import AzureOpenAI
import json
from mimetypes import inited
from tqdm import tqdm
import random
import base64
import requests
import argparse
import random
import re
import os
import openai
import multiprocessing
from functools import partial
from datetime import datetime
import time
import dashscope
from PIL import Image
from io import BytesIO


def encode_image(image_content):
    return base64.b64encode(image_content).decode("utf-8")


def qwen_vl_max(messages,
                filter_model_name,
                temperature=0.2,
                top_p=0.2,
                max_tokens=1024):
    max_try = 5
    dashscope.base_http_api_url = "https://poc-dashscope.aliyuncs.com/api/v1"
    dashscope.base_websocket_api_url = (
        "https://poc-dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation")

    model_name = filter_model_name
    print(f"Using model: {model_name}")
    dashscope.api_key = "your_api_key"

    generated_text = ""
    for _ in range(max_try):
        try:
            response = dashscope.MultiModalConversation.call(
                model=model_name,
                messages=messages,
                max_tokens=max_tokens,
                temperature=temperature,
                top_p=top_p,
                vl_high_resolution_images=True,
            )
            print(response)
            ### 获取模型的response
            response_content = response["output"]["choices"][0]["message"][
                "content"]
            generated_text = "".join(output["text"]
                                     for output in response_content
                                     if "text" in output)
            print(generated_text)
            return generated_text

        except Exception as e:
            print(e)
            time.sleep(5)
            continue

    return generated_text


class Qwen25VLAgent:

    def __init__(
        self,
        platform="ubuntu",
        planner_model="gpt-4o",
        executor_model="pre-qwen2.5vl-329-agent-v1-0112-model",
        max_tokens=1500,
        top_p=0.9,
        temperature=0.5,
        action_space="pyautogui",
        observation_type="screenshot",
    ):
        self.platform = platform
        self.planner_model = planner_model
        self.executor_model = executor_model
        assert self.executor_model is not None, "Executor model cannot be None"
        self.max_tokens = max_tokens
        self.top_p = top_p
        self.temperature = temperature
        self.action_space = action_space
        self.observation_type = observation_type
        assert action_space in ["pyautogui"], "Invalid action space"
        assert observation_type in ["screenshot"], "Invalid observation type"
        self.thoughts = []
        self.actions = []
        self.observations = []

    def predict(self, instruction: str, obs: Dict) -> List:
        """
        Predict the next action(s) based on the current observation.
        """
        # get the width and height of the screenshot
        image = Image.open(BytesIO(obs["screenshot"]))
        width, height = image.convert("RGB").size
        print(f"Screen resolution: {width}x{height}")

        previous_actions = ("\n".join([
            f"Step {i+1}: {action}" for i, action in enumerate(self.actions)
        ]) if self.actions else "None")

        system_prompt = """You are a helpful assistant.\n\n# Tools\n\nYou may call one or more functions to assist with the user query.\n\nYou are provided with function signatures within <tools></tools> XML tags:\n<tools>\n{"type": "function", "function": {"name_for_human": "computer_use", "name": "computer_use", "description": "Use a mouse and keyboard to interact with a computer, and take screenshots.\\n* This is an interface to a desktop GUI. You do not have access to a terminal or applications menu. You must click on desktop icons to start applications.\\n* Some applications may take time to start or process actions, so you may need to wait and take successive screenshots to see the results of your actions. E.g. if you click on Firefox and a window doesn\'t open, try wait and taking another screenshot.\\n"""

        system_prompt += f"""* The screen\'s resolution is {width}x{height}.\\n"""
        system_prompt += """* Whenever you inited to move the cursor to click on an element like an icon, you should consult a screenshot to determine the coordinates of the element before moving the cursor.\\n* If you tried clicking on a program or link but it failed to load, even after waiting, try adjusting your cursor position so that the tip of the cursor visually falls on the element that you want to click.\\n* Make sure to click any buttons, links, icons, etc with the cursor tip in the center of the element. Don\'t click boxes on their edges unless asked.", "parameters": {"properties": {"action": {"description": "The action to perform. The available actions are:\\n* `key`: Performs key down presses on the arguments passed in order, then performs key releases in reverse order.\\n* `type`: Type a string of text on the keyboard.\\n* `mouse_move`: Move the cursor to a specified (x, y) pixel coordinate on the screen.\\n* `left_click`: Click the left mouse button.\\n* `left_click_drag`: Click and drag the cursor to a specified (x, y) pixel coordinate on the screen.\\n* `right_click`: Click the right mouse button.\\n* `middle_click`: Click the middle mouse button.\\n* `double_click`: Double-click the left mouse button.\\n* `scroll`: Performs a scroll of the mouse scroll wheel vertically.\\n* `hscroll`: Performs a scroll of the mouse scroll wheel horizontally.\\n* `wait`: Wait specified seconds for the change to happen.\\n* `terminate`: Terminate the current task and report its completion status.", "enum": ["key", "type", "mouse_move", "left_click", "left_click_drag", "right_click", "middle_click", "double_click", "scroll", "hscroll", "wait", "terminate"], "type": "string"}, "keys": {"description": "Required only by `action=key`.", "type": "array"}, "text": {"description": "Required only by `action=type`.", "type": "string"}, "coordinate": {"description": "(x, y): The x (pixels from the left edge) and y (pixels from the top edge) coordinates to move the mouse to. Required only by `action=mouse_move` and `action=left_click_drag`.", "type": "array"}, "pixels": {"description": "The amount of scrolling to perform. Positive values scroll up or right, negative values scroll down or left. Required only by `action=scroll` and `action=hscroll`.", "type": "number"}, "time": {"description": "The seconds to wait. Required only by `action=wait`.", "type": "number"}, "status": {"description": "The status of the task. Required only by `action=terminate`.", "type": "string", "enum": ["success", "failure"]}}, "required": ["action"], "type": "object"}, "args_format": "Format the arguments as a JSON object."}}\n</tools>\n\nFor each function call, return a json object with function name and arguments within <tool_call></tool_call> XML tags:\n<tool_call>\n{"name": <function-name>, "arguments": <args-json-object>}\n</tool_call>"""

        user_prompt = (
            f"""Please generate the next move according to the UI screenshot and instruction.\n\nInstruction: {instruction}\n\n"""
            + "Previous actions:\n" + previous_actions)

        messages = [{
            "role": "system",
            "content": [{
                "type": "text",
                "text": system_prompt
            }]
        }, {
            "role":
            "user",
            "content": [
                {
                    "type":
                    "image",
                    "image":
                    f"data:image/png;base64,{encode_image(obs['screenshot'])}",
                },
                {
                    "type": "text",
                    "text": user_prompt
                },
            ],
        }]

        response = self.call_llm(
            {
                "model": self.executor_model,
                "messages": messages,
                "max_tokens": self.max_tokens,
                "top_p": self.top_p,
                "temperature": self.temperature,
            },
            self.executor_model,
        )

        logger.info(f"Qwen25VL Output: {response}")

        # 解析响应并提取pyautogui代码
        low_level_instruction, pyautogui_code = self.parse_response(response)

        logger.info(f"Low level instruction: {low_level_instruction}")
        logger.info(f"Pyautogui code: {pyautogui_code}")

        self.actions.append(low_level_instruction)

        return response, pyautogui_code

    def parse_response(self, response: str) -> Tuple[str, List[str]]:
        """
        解析LLM响应并将其转换为low level action和pyautogui代码。
        """
        low_level_instruction = ""
        pyautogui_code = []
        try:
            # 定义可能的标签组合
            start_tags = ["<tool_call>", "⚗"]
            end_tags = ["</tool_call>", "⚗"]

            # 找到有效的开始和结束标签
            start_tag = next((tag for tag in start_tags if tag in response), None)
            end_tag = next((tag for tag in end_tags if tag in response), None)

            if not start_tag or not end_tag:
                print("响应中缺少有效的开始或结束标签")
                return low_level_instruction, pyautogui_code

            # 分割响应以提取low_level_instruction和tool_call
            parts = response.split(start_tag)
            if len(parts) < 2:
                print("响应中缺少开始标签")
                return low_level_instruction, pyautogui_code

            low_level_instruction = parts[0].strip().replace("Action: ", "")
            tool_call_str = parts[1].split(end_tag)[0].strip()

            # 解析tool_call为JSON对象
            try:
                tool_call = json.loads(tool_call_str)
                action = tool_call.get("arguments", {}).get("action", "")
                args = tool_call.get("arguments", {})
            except json.JSONDecodeError as e:
                print(f"JSON解析错误: {e}")
                # 处理解析错误，返回默认值或空值
                action = ""
                args = {}

            # 根据action生成相应的pyautogui代码
            if action == "mouse_move":
                x, y = args.get("coordinate", [-1, -1])
                if x == -1 and y == -1:
                    pyautogui_code.append("pyautogui.moveTo(0, 0)")
                else:
                    pyautogui_code.append(f"pyautogui.moveTo({x}, {y})")
            elif action == "left_click":
                x, y = args.get("coordinate", [-1, -1])
                if x == -1 and y == -1:
                    pyautogui_code.append("pyautogui.click()")
                else:
                    pyautogui_code.append(f"pyautogui.click({x}, {y})")
            elif action == "type":
                text = args.get("text", "")
                pyautogui_code.append(f"pyautogui.typewrite('{text}')")
            elif action == "right_click":
                x, y = args.get("coordinate", [-1, -1])
                if x == -1 and y == -1:
                    pyautogui_code.append("pyautogui.rightClick()")
                else:
                    pyautogui_code.append(f"pyautogui.rightClick({x}, {y})")
            elif action == "double_click":
                x, y = args.get("coordinate", [-1, -1])
                if x == -1 and y == -1:
                    pyautogui_code.append("pyautogui.doubleClick()")
                else:
                    pyautogui_code.append(f"pyautogui.doubleClick({x}, {y})")
            elif action == "scroll":
                clicks = args.get("clicks", 0)
                pyautogui_code.append(f"pyautogui.scroll({clicks})")
            elif action == "key":
                # {"name": "computer_use", "arguments": {"action": "key", "keys": ["keys=[ctrl", "c]"]}}
                keys = args.get("keys", [])
                keys_str = ", ".join([f"'{key}'" for key in keys])
                pyautogui_code.append(f"pyautogui.hotkey({keys_str})")
            elif action == "drag":
                x, y = args.get("coordinate", [0, 0])
                duration = args.get("duration", 0.5)
                pyautogui_code.append(
                    f"pyautogui.dragTo({x}, {y}, duration={duration})")
            elif action == "wait":
                pyautogui_code.append("WAIT")
            elif action == "terminate":
                pyautogui_code.append("DONE")

        except (json.JSONDecodeError, IndexError) as e:
            logger.error(f"Failed to parse response: {e}")

        return low_level_instruction, pyautogui_code

    @backoff.on_exception(
        backoff.constant,
        # here you should add more model exceptions as you want,
        # but you are forbidden to add "Exception", that is, a common type of exception
        # because we want to catch this kind of Exception in the outside to ensure
        # each example won't exceed the time limit
        (
            # General exceptions
            SSLError,
            # OpenAI exceptions
            openai.RateLimitError,
            openai.BadRequestError,
            openai.InternalServerError,
            # Google exceptions
            InvalidArgument,
            ResourceExhausted,
            InternalServerError,
            BadRequest,
            # Groq exceptions
            # todo: check
        ),
        interval=30,
        max_tries=10,
    )
    def call_llm(self, payload, model):
        if model.startswith("gpt"):
            headers = {
                "Content-Type": "application/json",
                "Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
                # "Authorization": f"Bearer {os.environ['MIT_SPIDER_TOKEN']}"
            }
            logger.info("Generating content with GPT model: %s", model)
            response = requests.post(
                "https://api.openai.com/v1/chat/completions",
                headers=headers,
                json=payload,
            )

            if response.status_code != 200:
                logger.error("Failed to call LLM: " + response.text)
                time.sleep(5)
                return ""
            else:
                return response.json()["choices"][0]["message"]["content"]
        else:
            messages = payload["messages"]
            logger.info("Generating content with Qwen model: %s", model)
            response = qwen_vl_max(messages, model, self.temperature,
                                   self.top_p, self.max_tokens)
            return response

    def reset(self, _logger=None):
        global logger
        logger = (_logger if _logger is not None else
                  logging.getLogger("desktopenv.qwen25vl_agent"))

        self.thoughts = []
        self.action_descriptions = []
        self.actions = []
        self.observations = []

---

Code still untidy a little bit, clean it if you need.

Feel free to explore the code and let me know if you have any questions or need further clarification. I hope this helps with your benchmarking efforts!

Best regards,
Tianbao

[cjfcsjt]

@Timothyxxx @LukeForeverYoung
Thanks for your reply. I have two questions:

1. Tasks in the AndoridWorld or other benchmarks involve information retrieval and require the history to contain not only the JSON action itself. The current prompt design in Qwen2.5vl-UI is not general enough for UI tasks. Will it harm the performance if add the "step summary" in the prompt?
2. Does the response only contain the action, without any thought? Is this because the UI dataset used to train qwen2.5vl only contains action without thought?

[Timothyxxx]

For your question 2, when benchmarking OSWorld from prompting qwen2.5-vl, the thought is included in the response, for example:

Action: Click on the 'Relaunch to update' button to proceed with updating Chrome.
<tool_call>
{"name": "computer_use", "arguments": {"action": "left_click", "coordinate": [1190, 242]}}
</tool_call>

[LukeForeverYoung]

@Timothyxxx @LukeForeverYoung Thanks for your reply. I have two questions:

1. Tasks in the AndoridWorld or other benchmarks involve information retrieval and require the history to contain not only the JSON action itself. The current prompt design in Qwen2.5vl-UI is not general enough for UI tasks. Will it harm the performance if add the "step summary" in the prompt?
2. Does the response only contain the action, without any thought? Is this because the UI dataset used to train qwen2.5vl only contains action without thought?

For tasks in mobile scenarios, you could ask Qwen2.5-VL to think and provide summaries. Here is an example prompt:

Before answering, explain your reasoning step-by-step in <thinking></thinking> tags, and insert them before the <tool_call></tool_call> XML tags.\nAfter answering, summarize your action in <conclusion></conclusion> tags, and insert them after the <tool_call></tool_call> XML tags.

We leave the summaries in conversation histories by default. It should work well if you organize the summaries into the prompts, similar to mainstream UI-Agent frameworks.