File size: 11,699 Bytes
7fcdb70 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 |
import math
from typing import Union, Tuple
from .qwen_helpers.base_tool import BaseTool
from .qwen_helpers.fncall_prompt import NousFnCallPrompt
from .qwen_helpers.schema import (
ContentItem,
Message,
)
IMAGE_FACTOR = 28
MIN_PIXELS = 4 * 28 * 28
MAX_PIXELS = 16384 * 28 * 28
MAX_RATIO = 200
# @register_tool("computer_use")
class FaraComputerUse(BaseTool):
name = "computer_use"
@property
def description(self):
return f"""
Use a mouse and keyboard to interact with a computer, and take screenshots.
* 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.
* 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.
* The screen's resolution is {self.display_width_px}x{self.display_height_px}.
* Whenever you intend 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.
* 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.
* 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.
* When a separate scrollable container prominently overlays the webpage, if you want to scroll within it, you typically need to mouse_move() over it first and then scroll().
* If a popup window appears that you want to close, if left_click() on the 'X' or close button doesn't work, try key(keys=['Escape']) to close it.
* On some search bars, when you type(), you may need to press_enter=False and instead separately call left_click() on the search button to submit the search query. This is especially true of search bars that have auto-suggest popups for e.g. locations
* For calendar widgets, you usually need to left_click() on arrows to move between months and left_click() on dates to select them; type() is not typically used to input dates there.
""".strip()
parameters = {
"properties": {
"action": {
"description": """
The action to perform. The available actions are:
* `key`: Performs key down presses on the arguments passed in order, then performs key releases in reverse order. Includes "Enter", "Alt", "Shift", "Tab", "Control", "Backspace", "Delete", "Escape", "ArrowUp", "ArrowDown", "ArrowLeft", "ArrowRight", "PageDown", "PageUp", "Shift", etc.
* `type`: Type a string of text on the keyboard.
* `mouse_move`: Move the cursor to a specified (x, y) pixel coordinate on the screen.
* `left_click`: Click the left mouse button.
* `scroll`: Performs a scroll of the mouse scroll wheel.
* `visit_url`: Visit a specified URL.
* `web_search`: Perform a web search with a specified query.
* `history_back`: Go back to the previous page in the browser history.
* `pause_and_memorize_fact`: Pause and memorize a fact for future reference.
* `wait`: Wait specified seconds for the change to happen.
* `terminate`: Terminate the current task and report its completion status.
""".strip(),
"enum": [
"key",
"type",
"mouse_move",
"left_click",
"scroll",
"visit_url",
"web_search",
"history_back",
"pause_and_memorize_fact",
"wait",
"terminate",
],
"type": "string",
},
"keys": {
"description": "Required only by `action=key`.",
"type": "array",
},
"text": {
"description": "Required only by `action=type`.",
"type": "string",
},
"press_enter": {
"description": "Whether to press the Enter key after typing. Required only by `action=type`.",
"type": "boolean",
},
"delete_existing_text": {
"description": "Whether to delete existing text before typing. Required only by `action=type`.",
"type": "boolean",
},
"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=left_click`, `action=mouse_move`, and `action=type`.",
"type": "array",
},
"pixels": {
"description": "The amount of scrolling to perform. Positive values scroll up, negative values scroll down. Required only by `action=scroll`.",
"type": "number",
},
"url": {
"description": "The URL to visit. Required only by `action=visit_url`.",
"type": "string",
},
"query": {
"description": "The query to search for. Required only by `action=web_search`.",
"type": "string",
},
"fact": {
"description": "The fact to remember for the future. Required only by `action=pause_and_memorize_fact`.",
"type": "string",
},
"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",
}
def __init__(self, cfg=None):
self.display_width_px = cfg["display_width_px"]
self.display_height_px = cfg["display_height_px"]
include_input_text_key_args = cfg.pop("include_input_text_key_args", False)
if not include_input_text_key_args:
self.parameters["properties"].pop("press_enter", None)
self.parameters["properties"].pop("delete_existing_text", None)
super().__init__(cfg)
def call(self, params: Union[str, dict], **kwargs):
params = self._verify_json_format_args(params)
action = params["action"]
if action == "key":
return self._key(params["text"])
elif action == "click":
return self._click(coordinate=params["coordinate"])
elif action == "long_press":
return self._long_press(
coordinate=params["coordinate"], time=params["time"]
)
elif action == "swipe":
return self._swipe(
coordinate=params["coordinate"], coordinate2=params["coordinate2"]
)
elif action == "type":
return self._type(params["text"])
elif action == "system_button":
return self._system_button(params["button"])
elif action == "open":
return self._open(params["text"])
elif action == "wait":
return self._wait(params["time"])
elif action == "terminate":
return self._terminate(params["status"])
else:
raise ValueError(f"Unknown action: {action}")
def _key(self, text: str):
raise NotImplementedError()
def _click(self, coordinate: Tuple[int, int]):
raise NotImplementedError()
def _long_press(self, coordinate: Tuple[int, int], time: int):
raise NotImplementedError()
def _swipe(self, coordinate: Tuple[int, int], coordinate2: Tuple[int, int]):
raise NotImplementedError()
def _type(self, text: str):
raise NotImplementedError()
def _system_button(self, button: str):
raise NotImplementedError()
def _open(self, text: str):
raise NotImplementedError()
def _wait(self, time: int):
raise NotImplementedError()
def _terminate(self, status: str):
raise NotImplementedError()
def round_by_factor(number: int, factor: int) -> int:
"""Returns the closest integer to 'number' that is divisible by 'factor'."""
return round(number / factor) * factor
def ceil_by_factor(number: int, factor: int) -> int:
"""Returns the smallest integer greater than or equal to 'number' that is divisible by 'factor'."""
return math.ceil(number / factor) * factor
def floor_by_factor(number: int, factor: int) -> int:
"""Returns the largest integer less than or equal to 'number' that is divisible by 'factor'."""
return math.floor(number / factor) * factor
def smart_resize(
height: int,
width: int,
factor: int = IMAGE_FACTOR,
min_pixels: int = MIN_PIXELS,
max_pixels: int = MAX_PIXELS,
) -> tuple[int, int]:
"""
Rescales the image so that the following conditions are met:
1. Both dimensions (height and width) are divisible by 'factor'.
2. The total number of pixels is within the range ['min_pixels', 'max_pixels'].
3. The aspect ratio of the image is maintained as closely as possible.
"""
if max(height, width) / min(height, width) > MAX_RATIO:
raise ValueError(
f"absolute aspect ratio must be smaller than {MAX_RATIO}, got {max(height, width) / min(height, width)}"
)
h_bar = max(factor, round_by_factor(height, factor))
w_bar = max(factor, round_by_factor(width, factor))
if h_bar * w_bar > max_pixels:
beta = math.sqrt((height * width) / max_pixels)
h_bar = floor_by_factor(height / beta, factor)
w_bar = floor_by_factor(width / beta, factor)
elif h_bar * w_bar < min_pixels:
beta = math.sqrt(min_pixels / (height * width))
h_bar = ceil_by_factor(height * beta, factor)
w_bar = ceil_by_factor(width * beta, factor)
return h_bar, w_bar
def get_computer_use_system_prompt(
image,
processor_im_cfg,
include_input_text_key_args=False,
fn_call_template="default",
):
patch_size = processor_im_cfg["patch_size"]
merge_size = processor_im_cfg["merge_size"]
min_pixels = processor_im_cfg["min_pixels"]
max_pixels = processor_im_cfg["max_pixels"]
resized_height, resized_width = smart_resize(
image.height,
image.width,
factor=patch_size * merge_size,
min_pixels=min_pixels,
max_pixels=max_pixels,
)
computer_use = FaraComputerUse(
cfg={
"display_width_px": resized_width,
"display_height_px": resized_height,
"include_input_text_key_args": include_input_text_key_args,
}
)
conversation = NousFnCallPrompt(
template_name=fn_call_template
).preprocess_fncall_messages(
messages=[
Message(
role="system",
content=[ContentItem(text="You are a helpful assistant.")],
),
],
functions=[computer_use.function],
lang=None,
)
return {
"conversation": [msg.model_dump() for msg in conversation],
"im_size": (resized_width, resized_height),
}
|