HyperCluster/Fara-BrowserUse
5
1import ast
2import asyncio
3import io
4import json
5import logging
6import os
7from typing import Any, Dict, List, Tuple
8from urllib.parse import quote_plus
9
10from openai import AsyncOpenAI
11from PIL import Image
12from playwright.async_api import BrowserContext, Download, Page
13from tenacity import before_sleep_log, retry, stop_after_attempt, wait_exponential
14
15from ._prompts import get_computer_use_system_prompt
16from .browser.playwright_controller import PlaywrightController
17from .types import (
18 AssistantMessage,
19 FunctionCall,
20 ImageObj,
21 LLMMessage,
22 ModelResponse,
23 SystemMessage,
24 UserMessage,
25 WebSurferEvent,
26 message_to_openai_format,
27)
28from .utils import get_trimmed_url
29
30
31class FaraAgent:
32 DEFAULT_START_PAGE = "https://www.bing.com/"
33
34 MLM_PROCESSOR_IM_CFG = {
35 "min_pixels": 3136,
36 "max_pixels": 12845056,
37 "patch_size": 14,
38 "merge_size": 2,
39 }
40
41 SCREENSHOT_TOKENS = 1105
42 USER_MESSAGE = "Here is the next screenshot. Think about what to do next."
43 MAX_URL_LENGTH = 100
44
45 def __init__(
46 self,
47 browser_manager: Any,
48 client_config: dict,
49 downloads_folder: str | None = None,
50 start_page: str | None = "about:blank",
51 animate_actions: bool = False,
52 single_tab_mode: bool = True,
53 max_n_images: int = 3,
54 fn_call_template: str = "default",
55 model_call_timeout: int = 20,
56 max_rounds: int = 10,
57 save_screenshots: bool = False,
58 logger: logging.Logger | None = None,
59 ):
60 self.downloads_folder = downloads_folder
61 if not os.path.exists(self.downloads_folder or "") and self.downloads_folder:
62 os.makedirs(self.downloads_folder)
63 self.single_tab_mode = single_tab_mode
64 self.start_page = start_page or self.DEFAULT_START_PAGE
65 self.animate_actions = animate_actions
66 self.browser_manager = browser_manager
67 self.client_config = client_config
68 self.max_n_images = max_n_images
69 self.fn_call_template = fn_call_template
70 self.model_call_timeout = model_call_timeout
71 self.max_rounds = max_rounds
72 self.max_url_chars = self.MAX_URL_LENGTH
73 if save_screenshots and self.downloads_folder is None:
74 assert False, "downloads_folder must be set if save_screenshots is True"
75 self.save_screenshots = save_screenshots
76 self._facts = []
77 self._task_summary = None
78 self._num_actions = 0
79 self.logger = logger or logging.getLogger(__name__)
80 self._mlm_width = 1440
81 self._mlm_height = 900
82 self.viewport_height = 900
83 self.viewport_width = 1440
84 self.include_input_text_key_args = True
85
86 def _download_handler(download: Download) -> None:
87 self._last_download = download
88
89 self._download_handler = _download_handler
90 self.did_initialize = False
91
92 # OpenAI client will be initialized in initialize()
93 self._openai_client: AsyncOpenAI | None = None
94 self._chat_history: List[LLMMessage] = []
95
96 async def initialize(self) -> None:
97 if self.did_initialize:
98 return
99 self._last_download = None
100 self._prior_metadata_hash = None
101
102 # Initialize OpenAI client
103 self._openai_client = AsyncOpenAI(
104 api_key=self.client_config.get("api_key"),
105 base_url=self.client_config.get("base_url"),
106 default_headers=self.client_config.get("default_headers"),
107 )
108
109 # Set up download handler
110 self.browser_manager.set_download_handler(self._download_handler)
111
112 # Initialize browser
113 await self.browser_manager.init(self.start_page)
114 self.did_initialize = True
115
116 @property
117 def _page(self) -> Page | None:
118 """Get the current page from browser manager."""
119 return self.browser_manager.page if self.browser_manager else None
120
121 @_page.setter
122 def _page(self, value):
123 if self.browser_manager:
124 self.browser_manager.page = value
125 else:
126 raise ValueError("Browser manager is not initialized. Cannot set page.")
127
128 @property
129 def context(self) -> BrowserContext | None:
130 """Get the browser context from browser manager."""
131 return self.browser_manager.context if self.browser_manager else None
132
133 @property
134 def _playwright_controller(self) -> PlaywrightController | None:
135 """Get the playwright controller from browser manager."""
136 return (
137 self.browser_manager.playwright_controller if self.browser_manager else None
138 )
139
140 async def wait_for_captcha_with_timeout(
141 self, timeout_seconds=300
142 ): # 5 minutes default
143 """Wait for captcha to be solved with timeout"""
144 try:
145 await asyncio.wait_for(
146 self.browser_manager.wait_for_captcha_resolution(),
147 timeout=timeout_seconds,
148 )
149 return True # Captcha solved in time
150 except asyncio.TimeoutError:
151 self.logger.warning(f"Captcha timeout after {timeout_seconds} seconds!")
152 # Force resume execution
153 self.browser_manager._captcha_event.set()
154 return False # Captcha timed out
155
156 @retry(
157 stop=stop_after_attempt(5),
158 wait=wait_exponential(multiplier=5.0, min=5.0, max=60),
159 before_sleep=before_sleep_log(logging.getLogger(__name__), logging.WARNING),
160 reraise=True,
161 )
162 async def _make_model_call(
163 self,
164 history: List[LLMMessage],
165 extra_create_args: Dict[str, Any] | None = None,
166 ) -> ModelResponse:
167 """Make a model call using OpenAI client"""
168 openai_messages = [message_to_openai_format(msg) for msg in history]
169 request_params = {
170 "model": self.client_config.get("model", "gpt-4o"),
171 "messages": openai_messages,
172 }
173 if extra_create_args:
174 request_params.update(extra_create_args)
175
176 response = await self._openai_client.chat.completions.create(**request_params)
177 content = response.choices[0].message.content
178 usage = {}
179 if response.usage:
180 usage = {
181 "prompt_tokens": response.usage.prompt_tokens,
182 "completion_tokens": response.usage.completion_tokens,
183 "total_tokens": response.usage.total_tokens,
184 }
185 return ModelResponse(content=content, usage=usage)
186
187 def remove_screenshot_from_message(self, msg: List[Dict[str, Any]] | Any) -> Any:
188 """Remove the screenshot from the message content."""
189 if isinstance(msg.content, list):
190 new_content = []
191 for c in msg.content:
192 if not isinstance(c, ImageObj):
193 new_content.append(c)
194 msg.content = new_content
195 elif isinstance(msg.content, ImageObj):
196 msg = None
197 return msg
198
199 def maybe_remove_old_screenshots(
200 self, history: List[LLMMessage], includes_current: bool = False
201 ) -> List[LLMMessage]:
202 """Remove old screenshots from the chat history. Assuming we have not yet added the current screenshot message.
203
204 Note: Original user messages (marked with is_original=True) have their TEXT preserved,
205 but their images may be removed if we exceed max_n_images. Boilerplate messages can be
206 completely removed.
207 """
208 if self.max_n_images <= 0:
209 return history
210
211 max_n_images = self.max_n_images if includes_current else self.max_n_images - 1
212 new_history: List[LLMMessage] = []
213 n_images = 0
214 for i in range(len(history) - 1, -1, -1):
215 msg = history[i]
216
217 is_original_user_message = isinstance(msg, UserMessage) and getattr(
218 msg, "is_original", False
219 )
220
221 if i == 0 and n_images >= max_n_images:
222 # First message is always the task so we keep it and remove the screenshot if necessary
223 msg = self.remove_screenshot_from_message(msg)
224 if msg is None:
225 continue
226
227 if isinstance(msg.content, list):
228 # Check if the message contains an image. Assumes 1 image per message.
229 has_image = False
230 for c in msg.content:
231 if isinstance(c, ImageObj):
232 has_image = True
233 break
234 if has_image:
235 if n_images < max_n_images:
236 new_history.append(msg)
237 elif is_original_user_message:
238 # Original user message but over limit: keep text, remove image
239 msg = self.remove_screenshot_from_message(msg)
240 if msg is not None:
241 new_history.append(msg)
242 n_images += 1
243 else:
244 new_history.append(msg)
245 elif isinstance(msg.content, ImageObj):
246 if n_images < max_n_images:
247 new_history.append(msg)
248 n_images += 1
249 else:
250 new_history.append(msg)
251
252 new_history = new_history[::-1]
253
254 return new_history
255
256 async def _get_scaled_screenshot(self) -> Image.Image:
257 """Get current screenshot and scale it for the model."""
258 screenshot = await self._playwright_controller.get_screenshot(self._page)
259 screenshot = Image.open(io.BytesIO(screenshot))
260 _, scaled_screenshot = self._get_system_message(screenshot)
261 return scaled_screenshot
262
263 def _get_system_message(
264 self, screenshot: ImageObj | Image.Image
265 ) -> Tuple[List[SystemMessage], Image.Image]:
266 system_prompt_info = get_computer_use_system_prompt(
267 screenshot,
268 self.MLM_PROCESSOR_IM_CFG,
269 include_input_text_key_args=self.include_input_text_key_args,
270 fn_call_template=self.fn_call_template,
271 )
272 self._mlm_width, self._mlm_height = system_prompt_info["im_size"]
273 scaled_screenshot = screenshot.resize((self._mlm_width, self._mlm_height))
274
275 system_message = []
276 for msg in system_prompt_info["conversation"]:
277 tmp_content = ""
278 for content in msg["content"]:
279 tmp_content += content["text"]
280
281 system_message.append(SystemMessage(content=tmp_content))
282
283 return system_message, scaled_screenshot
284
285 def _parse_thoughts_and_action(self, message: str) -> Tuple[str, Dict[str, Any]]:
286 try:
287 tmp = message.split("<tool_call>\n")
288 thoughts = tmp[0].strip()
289 action_text = tmp[1].split("\n</tool_call>")[0]
290 try:
291 action = json.loads(action_text)
292 except json.decoder.JSONDecodeError:
293 self.logger.error(f"Invalid action text: {action_text}")
294 action = ast.literal_eval(action_text)
295
296 return thoughts, action
297 except Exception as e:
298 self.logger.error(
299 f"Error parsing thoughts and action: {message}", exc_info=True
300 )
301 raise e
302
303 def convert_resized_coords_to_original(
304 self, coords: List[float], rsz_w: int, rsz_h: int, og_w: int, og_h: int
305 ) -> List[float]:
306 scale_x = og_w / rsz_w
307 scale_y = og_h / rsz_h
308 return [coords[0] * scale_x, coords[1] * scale_y]
309
310 def proc_coords(
311 self,
312 coords: List[float] | None,
313 im_w: int,
314 im_h: int,
315 og_im_w: int | None = None,
316 og_im_h: int | None = None,
317 ) -> List[float] | None:
318 if not coords:
319 return coords
320
321 if og_im_w is None:
322 og_im_w = im_w
323 if og_im_h is None:
324 og_im_h = im_h
325
326 tgt_x, tgt_y = coords
327 return self.convert_resized_coords_to_original(
328 [tgt_x, tgt_y], im_w, im_h, og_im_w, og_im_h
329 )
330
331 async def run(self, user_message: str) -> Tuple:
332 """Run the agent with a user message."""
333 # Initialize if not already done
334 await self.initialize()
335
336 # Ensure page is ready after initialization
337 assert self._page is not None, "Page should be initialized"
338
339 # Get initial screenshot and add user message with image to chat history
340 scaled_screenshot = await self._get_scaled_screenshot()
341
342 if self.save_screenshots:
343 await self._playwright_controller.get_screenshot(
344 self._page,
345 path=os.path.join(
346 self.downloads_folder, f"screenshot{self._num_actions}.png"
347 ),
348 )
349
350 self._chat_history.append(
351 UserMessage(
352 content=[ImageObj.from_pil(scaled_screenshot), user_message],
353 is_original=True,
354 )
355 )
356
357 all_actions = []
358 all_observations = []
359 final_answer = "<no_answer>"
360 is_stop_action = False
361 for i in range(self.max_rounds):
362 is_first_round = i == 0
363 if not self.browser_manager._captcha_event.is_set():
364 self.logger.info("Waiting 60s for captcha to finish...")
365 captcha_solved = await self.wait_for_captcha_with_timeout(60)
366 if (
367 not captcha_solved
368 and not self.browser_manager._captcha_event.is_set()
369 ):
370 raise RuntimeError(
371 "Captcha timed out, unable to proceed with web surfing."
372 )
373
374 function_call, raw_response = await self.generate_model_call(
375 is_first_round, scaled_screenshot if is_first_round else None
376 )
377 assert isinstance(raw_response, str)
378 all_actions.append(raw_response)
379 thoughts, action_dict = self._parse_thoughts_and_action(raw_response)
380 action_args = action_dict.get("arguments", {})
381 action = action_args["action"]
382 self.logger.info(
383 f"\nThought #{i + 1}: {thoughts}\nAction #{i + 1}: executing tool '{action}' with arguments {json.dumps(action_args)}"
384 )
385
386 (
387 is_stop_action,
388 new_screenshot,
389 action_description,
390 ) = await self.execute_action(function_call)
391 all_observations.append(action_description)
392 self.logger.info(f"Observation#{i + 1}: {action_description}")
393 if is_stop_action:
394 final_answer = thoughts
395 break
396 return final_answer, all_actions, all_observations
397
398 async def generate_model_call(
399 self, is_first_round: bool, first_screenshot: Image.Image | None = None
400 ) -> Tuple[List[FunctionCall], str]:
401 history = self.maybe_remove_old_screenshots(self._chat_history)
402
403 screenshot_for_system = first_screenshot
404 if not is_first_round:
405 # Get screenshot and add new user message for subsequent rounds
406 scaled_screenshot = await self._get_scaled_screenshot()
407 screenshot_for_system = scaled_screenshot
408
409 text_prompt = self.USER_MESSAGE
410 curr_url = await self._playwright_controller.get_page_url(self._page)
411 trimmed_url = get_trimmed_url(curr_url, max_len=self.max_url_chars)
412 text_prompt = f"Current URL: {trimmed_url}\n" + text_prompt
413
414 curr_message = UserMessage(
415 content=[ImageObj.from_pil(scaled_screenshot), text_prompt]
416 )
417 self._chat_history.append(curr_message)
418 history.append(curr_message)
419
420 # Generate system message using the screenshot
421 system_message, _ = self._get_system_message(screenshot_for_system)
422 history = system_message + history
423 response = await self._make_model_call(
424 history, extra_create_args={"temperature": 0}
425 )
426 message = response.content
427
428 self._chat_history.append(AssistantMessage(content=message))
429 thoughts, action = self._parse_thoughts_and_action(message)
430 action["arguments"]["thoughts"] = thoughts
431
432 function_call = [FunctionCall(id="dummy", **action)]
433 return function_call, message
434
435 async def execute_action(
436 self,
437 function_call: List[FunctionCall],
438 ) -> Tuple[bool, bytes, str]:
439 name = function_call[0].name
440 args = function_call[0].arguments
441 action_description = ""
442 assert self._page is not None
443 self.logger.debug(
444 WebSurferEvent(
445 source="FaraAgent",
446 url=await self._playwright_controller.get_page_url(self._page),
447 action=name,
448 arguments=args,
449 message=f"{name}( {json.dumps(args)} )",
450 )
451 )
452 if "coordinate" in args:
453 args["coordinate"] = self.proc_coords(
454 args["coordinate"],
455 self._mlm_width,
456 self._mlm_height,
457 self.viewport_width,
458 self.viewport_height,
459 )
460
461 is_stop_action = False
462
463 if args["action"] == "visit_url":
464 url = str(args["url"])
465 action_description = f"I typed '{url}' into the browser address bar."
466 # Check if the argument starts with a known protocol
467 if url.startswith(("https://", "http://", "file://", "about:")):
468 (
469 reset_prior_metadata,
470 reset_last_download,
471 ) = await self._playwright_controller.visit_page(self._page, url)
472 # If the argument contains a space, treat it as a search query
473 elif " " in url:
474 (
475 reset_prior_metadata,
476 reset_last_download,
477 ) = await self._playwright_controller.visit_page(
478 self._page,
479 f"https://www.bing.com/search?q={quote_plus(url)}&FORM=QBLH",
480 )
481 # Otherwise, prefix with https://
482 else:
483 (
484 reset_prior_metadata,
485 reset_last_download,
486 ) = await self._playwright_controller.visit_page(
487 self._page, "https://" + url
488 )
489 if reset_last_download and self._last_download is not None:
490 self._last_download = None
491 if reset_prior_metadata and self._prior_metadata_hash is not None:
492 self._prior_metadata_hash = None
493 elif args["action"] == "history_back":
494 action_description = "I clicked the browser back button."
495 await self._playwright_controller.back(self._page)
496 elif args["action"] == "web_search":
497 query = args.get("query")
498 action_description = f"I typed '{query}' into the browser search bar."
499 encoded_query = quote_plus(query)
500 (
501 reset_prior_metadata,
502 reset_last_download,
503 ) = await self._playwright_controller.visit_page(
504 self._page, f"https://www.bing.com/search?q={encoded_query}&FORM=QBLH"
505 )
506 if reset_last_download and self._last_download is not None:
507 self._last_download = None
508 if reset_prior_metadata and self._prior_metadata_hash is not None:
509 self._prior_metadata_hash = None
510 elif args["action"] == "scroll":
511 pixels = int(args.get("pixels", 0))
512 if pixels > 0:
513 action_description = "I scrolled up one page in the browser."
514 await self._playwright_controller.page_up(self._page)
515 elif pixels < 0:
516 action_description = "I scrolled down one page in the browser."
517 await self._playwright_controller.page_down(self._page)
518 elif args["action"] == "keypress" or args["action"] == "key":
519 keys = args.get("keys", [])
520 action_description = f"I pressed the following keys: {keys}"
521 await self._playwright_controller.keypress(self._page, keys)
522 elif args["action"] == "hover" or args["action"] == "mouse_move":
523 if "coordinate" in args:
524 tgt_x, tgt_y = args["coordinate"]
525 await self._playwright_controller.hover_coords(self._page, tgt_x, tgt_y)
526
527 elif args["action"] == "sleep" or args["action"] == "wait":
528 duration = args.get("duration", 3.0)
529 duration = args.get("time", duration)
530 action_description = (
531 "I am waiting a short period of time before taking further action."
532 )
533 await self._playwright_controller.sleep(self._page, duration)
534 elif args["action"] == "click" or args["action"] == "left_click":
535 if "coordinate" in args:
536 tgt_x, tgt_y = args["coordinate"]
537 action_description = f"I clicked at coordinates ({tgt_x}, {tgt_y})."
538 new_page_tentative = await self._playwright_controller.click_coords(
539 self._page, tgt_x, tgt_y
540 )
541
542 if new_page_tentative is not None:
543 self._page = new_page_tentative
544 self._prior_metadata_hash = None
545
546 elif args["action"] == "input_text" or args["action"] == "type":
547 text_value = str(args.get("text", args.get("text_value")))
548 action_description = f"I typed '{text_value}'."
549 press_enter = args.get("press_enter", True)
550 delete_existing_text = args.get("delete_existing_text", False)
551
552 if "coordinate" in args:
553 tgt_x, tgt_y = args["coordinate"]
554 new_page_tentative = await self._playwright_controller.fill_coords(
555 self._page,
556 tgt_x,
557 tgt_y,
558 text_value,
559 press_enter=press_enter,
560 delete_existing_text=delete_existing_text,
561 )
562 if new_page_tentative is not None:
563 self._page = new_page_tentative
564 self._prior_metadata_hash = None
565
566 elif args["action"] == "pause_and_memorize_fact":
567 fact = str(args.get("fact"))
568 self._facts.append(fact)
569 action_description = f"I memorized the following fact: {fact}"
570 elif args["action"] == "stop" or args["action"] == "terminate":
571 action_description = args.get("thoughts")
572 is_stop_action = True
573
574 else:
575 raise ValueError(f"Unknown tool: {args['action']}")
576
577 await self._playwright_controller.wait_for_load_state(self._page)
578 await self._playwright_controller.sleep(self._page, 3)
579
580 # Get new screenshot after action
581 self._num_actions += 1
582 if self.save_screenshots:
583 new_screenshot = await self._playwright_controller.get_screenshot(
584 self._page,
585 path=os.path.join(
586 self.downloads_folder, f"screenshot{self._num_actions}.png"
587 ),
588 )
589 else:
590 new_screenshot = await self._playwright_controller.get_screenshot(
591 self._page
592 )
593 return is_stop_action, new_screenshot, action_description
594
595 async def close(self) -> None:
596 """
597 Close the browser and the page.
598 Should be called when the agent is no longer needed.
599 """
600 if self._page is not None:
601 self._page = None
602 await self.browser_manager.close()
603