Akjava/open_Deep-Research-DuckDuckGo
4
1# Shamelessly stolen from Microsoft Autogen team: thanks to them for this great resource!2# https://github.com/microsoft/autogen/blob/gaia_multiagent_v01_march_1st/autogen/browser_utils.py3import mimetypes4import os5import pathlib6import re7import time8import uuid9from typing import Any, Dict, List, Optional, Tuple, Union10from urllib.parse import unquote, urljoin, urlparse11 12import pathvalidate13import requests14from serpapi import GoogleSearch15 16from smolagents import Tool17 18from .cookies import COOKIES19from .mdconvert import FileConversionException, MarkdownConverter, UnsupportedFormatException20 21 22class SimpleTextBrowser:23 """(In preview) An extremely simple text-based web browser comparable to Lynx. Suitable for Agentic use."""24 25 def __init__(26 self,27 start_page: Optional[str] = None,28 viewport_size: Optional[int] = 1024 * 8,29 downloads_folder: Optional[Union[str, None]] = None,30 serpapi_key: Optional[Union[str, None]] = None,31 request_kwargs: Optional[Union[Dict[str, Any], None]] = None,32 ):33 self.start_page: str = start_page if start_page else "about:blank"34 self.viewport_size = viewport_size # Applies only to the standard uri types35 self.downloads_folder = downloads_folder36 self.history: List[Tuple[str, float]] = list()37 self.page_title: Optional[str] = None38 self.viewport_current_page = 039 self.viewport_pages: List[Tuple[int, int]] = list()40 self.set_address(self.start_page)41 self.serpapi_key = serpapi_key42 self.request_kwargs = request_kwargs43 self.request_kwargs["cookies"] = COOKIES44 self._mdconvert = MarkdownConverter()45 self._page_content: str = ""46 47 self._find_on_page_query: Union[str, None] = None48 self._find_on_page_last_result: Union[int, None] = None # Location of the last result49 50 @property51 def address(self) -> str:52 """Return the address of the current page."""53 return self.history[-1][0]54 55 def set_address(self, uri_or_path: str, filter_year: Optional[int] = None) -> None:56 # TODO: Handle anchors57 self.history.append((uri_or_path, time.time()))58 59 # Handle special URIs60 if uri_or_path == "about:blank":61 self._set_page_content("")62 elif uri_or_path.startswith("google:"):63 self._serpapi_search(uri_or_path[len("google:") :].strip(), filter_year=filter_year)64 else:65 if (66 not uri_or_path.startswith("http:")67 and not uri_or_path.startswith("https:")68 and not uri_or_path.startswith("file:")69 ):70 if len(self.history) > 1:71 prior_address = self.history[-2][0]72 uri_or_path = urljoin(prior_address, uri_or_path)73 # Update the address with the fully-qualified path74 self.history[-1] = (uri_or_path, self.history[-1][1])75 self._fetch_page(uri_or_path)76 77 self.viewport_current_page = 078 self.find_on_page_query = None79 self.find_on_page_viewport = None80 81 @property82 def viewport(self) -> str:83 """Return the content of the current viewport."""84 bounds = self.viewport_pages[self.viewport_current_page]85 return self.page_content[bounds[0] : bounds[1]]86 87 @property88 def page_content(self) -> str:89 """Return the full contents of the current page."""90 return self._page_content91 92 def _set_page_content(self, content: str) -> None:93 """Sets the text content of the current page."""94 self._page_content = content95 self._split_pages()96 if self.viewport_current_page >= len(self.viewport_pages):97 self.viewport_current_page = len(self.viewport_pages) - 198 99 def page_down(self) -> None:100 self.viewport_current_page = min(self.viewport_current_page + 1, len(self.viewport_pages) - 1)101 102 def page_up(self) -> None:103 self.viewport_current_page = max(self.viewport_current_page - 1, 0)104 105 def find_on_page(self, query: str) -> Union[str, None]:106 """Searches for the query from the current viewport forward, looping back to the start if necessary."""107 108 # Did we get here via a previous find_on_page search with the same query?109 # If so, map to find_next110 if query == self._find_on_page_query and self.viewport_current_page == self._find_on_page_last_result:111 return self.find_next()112 113 # Ok it's a new search start from the current viewport114 self._find_on_page_query = query115 viewport_match = self._find_next_viewport(query, self.viewport_current_page)116 if viewport_match is None:117 self._find_on_page_last_result = None118 return None119 else:120 self.viewport_current_page = viewport_match121 self._find_on_page_last_result = viewport_match122 return self.viewport123 124 def find_next(self) -> Union[str, None]:125 """Scroll to the next viewport that matches the query"""126 127 if self._find_on_page_query is None:128 return None129 130 starting_viewport = self._find_on_page_last_result131 if starting_viewport is None:132 starting_viewport = 0133 else:134 starting_viewport += 1135 if starting_viewport >= len(self.viewport_pages):136 starting_viewport = 0137 138 viewport_match = self._find_next_viewport(self._find_on_page_query, starting_viewport)139 if viewport_match is None:140 self._find_on_page_last_result = None141 return None142 else:143 self.viewport_current_page = viewport_match144 self._find_on_page_last_result = viewport_match145 return self.viewport146 147 def _find_next_viewport(self, query: str, starting_viewport: int) -> Union[int, None]:148 """Search for matches between the starting viewport looping when reaching the end."""149 150 if query is None:151 return None152 153 # Normalize the query, and convert to a regular expression154 nquery = re.sub(r"\*", "__STAR__", query)155 nquery = " " + (" ".join(re.split(r"\W+", nquery))).strip() + " "156 nquery = nquery.replace(" __STAR__ ", "__STAR__ ") # Merge isolated stars with prior word157 nquery = nquery.replace("__STAR__", ".*").lower()158 159 if nquery.strip() == "":160 return None161 162 idxs = list()163 idxs.extend(range(starting_viewport, len(self.viewport_pages)))164 idxs.extend(range(0, starting_viewport))165 166 for i in idxs:167 bounds = self.viewport_pages[i]168 content = self.page_content[bounds[0] : bounds[1]]169 170 # TODO: Remove markdown links and images171 ncontent = " " + (" ".join(re.split(r"\W+", content))).strip().lower() + " "172 if re.search(nquery, ncontent):173 return i174 175 return None176 177 def visit_page(self, path_or_uri: str, filter_year: Optional[int] = None) -> str:178 """Update the address, visit the page, and return the content of the viewport."""179 self.set_address(path_or_uri, filter_year=filter_year)180 return self.viewport181 182 def _split_pages(self) -> None:183 # Do not split search results184 if self.address.startswith("google:"):185 self.viewport_pages = [(0, len(self._page_content))]186 return187 188 # Handle empty pages189 if len(self._page_content) == 0:190 self.viewport_pages = [(0, 0)]191 return192 193 # Break the viewport into pages194 self.viewport_pages = []195 start_idx = 0196 while start_idx < len(self._page_content):197 end_idx = min(start_idx + self.viewport_size, len(self._page_content)) # type: ignore[operator]198 # Adjust to end on a space199 while end_idx < len(self._page_content) and self._page_content[end_idx - 1] not in [" ", "\t", "\r", "\n"]:200 end_idx += 1201 self.viewport_pages.append((start_idx, end_idx))202 start_idx = end_idx203 204 def _serpapi_search(self, query: str, filter_year: Optional[int] = None) -> None:205 if self.serpapi_key is None:206 raise ValueError("Missing SerpAPI key.")207 208 params = {209 "engine": "google",210 "q": query,211 "api_key": self.serpapi_key,212 }213 if filter_year is not None:214 params["tbs"] = f"cdr:1,cd_min:01/01/{filter_year},cd_max:12/31/{filter_year}"215 216 search = GoogleSearch(params)217 results = search.get_dict()218 self.page_title = f"{query} - Search"219 if "organic_results" not in results.keys():220 raise Exception(f"No results found for query: '{query}'. Use a less specific query.")221 if len(results["organic_results"]) == 0:222 year_filter_message = f" with filter year={filter_year}" if filter_year is not None else ""223 self._set_page_content(224 f"No results found for '{query}'{year_filter_message}. Try with a more general query, or remove the year filter."225 )226 return227 228 def _prev_visit(url):229 for i in range(len(self.history) - 1, -1, -1):230 if self.history[i][0] == url:231 return f"You previously visited this page {round(time.time() - self.history[i][1])} seconds ago.\n"232 return ""233 234 web_snippets: List[str] = list()235 idx = 0236 if "organic_results" in results:237 for page in results["organic_results"]:238 idx += 1239 date_published = ""240 if "date" in page:241 date_published = "\nDate published: " + page["date"]242 243 source = ""244 if "source" in page:245 source = "\nSource: " + page["source"]246 247 snippet = ""248 if "snippet" in page:249 snippet = "\n" + page["snippet"]250 251 redacted_version = f"{idx}. [{page['title']}]({page['link']}){date_published}{source}\n{_prev_visit(page['link'])}{snippet}"252 253 redacted_version = redacted_version.replace("Your browser can't play this video.", "")254 web_snippets.append(redacted_version)255 256 content = (257 f"A Google search for '{query}' found {len(web_snippets)} results:\n\n## Web Results\n"258 + "\n\n".join(web_snippets)259 )260 261 self._set_page_content(content)262 263 def _fetch_page(self, url: str) -> None:264 download_path = ""265 try:266 if url.startswith("file://"):267 download_path = os.path.normcase(os.path.normpath(unquote(url[7:])))268 res = self._mdconvert.convert_local(download_path)269 self.page_title = res.title270 self._set_page_content(res.text_content)271 else:272 # Prepare the request parameters273 request_kwargs = self.request_kwargs.copy() if self.request_kwargs is not None else {}274 request_kwargs["stream"] = True275 276 # Send a HTTP request to the URL277 response = requests.get(url, **request_kwargs)278 response.raise_for_status()279 280 # If the HTTP request was successful281 content_type = response.headers.get("content-type", "")282 283 # Text or HTML284 if "text/" in content_type.lower():285 res = self._mdconvert.convert_response(response)286 self.page_title = res.title287 self._set_page_content(res.text_content)288 # A download289 else:290 # Try producing a safe filename291 fname = None292 download_path = None293 try:294 fname = pathvalidate.sanitize_filename(os.path.basename(urlparse(url).path)).strip()295 download_path = os.path.abspath(os.path.join(self.downloads_folder, fname))296 297 suffix = 0298 while os.path.exists(download_path) and suffix < 1000:299 suffix += 1300 base, ext = os.path.splitext(fname)301 new_fname = f"{base}__{suffix}{ext}"302 download_path = os.path.abspath(os.path.join(self.downloads_folder, new_fname))303 304 except NameError:305 pass306 307 # No suitable name, so make one308 if fname is None:309 extension = mimetypes.guess_extension(content_type)310 if extension is None:311 extension = ".download"312 fname = str(uuid.uuid4()) + extension313 download_path = os.path.abspath(os.path.join(self.downloads_folder, fname))314 315 # Open a file for writing316 with open(download_path, "wb") as fh:317 for chunk in response.iter_content(chunk_size=512):318 fh.write(chunk)319 320 # Render it321 local_uri = pathlib.Path(download_path).as_uri()322 self.set_address(local_uri)323 324 except UnsupportedFormatException as e:325 print(e)326 self.page_title = ("Download complete.",)327 self._set_page_content(f"# Download complete\n\nSaved file to '{download_path}'")328 except FileConversionException as e:329 print(e)330 self.page_title = ("Download complete.",)331 self._set_page_content(f"# Download complete\n\nSaved file to '{download_path}'")332 except FileNotFoundError:333 self.page_title = "Error 404"334 self._set_page_content(f"## Error 404\n\nFile not found: {download_path}")335 except requests.exceptions.RequestException as request_exception:336 try:337 self.page_title = f"Error {response.status_code}"338 339 # If the error was rendered in HTML we might as well render it340 content_type = response.headers.get("content-type", "")341 if content_type is not None and "text/html" in content_type.lower():342 res = self._mdconvert.convert(response)343 self.page_title = f"Error {response.status_code}"344 self._set_page_content(f"## Error {response.status_code}\n\n{res.text_content}")345 else:346 text = ""347 for chunk in response.iter_content(chunk_size=512, decode_unicode=True):348 text += chunk349 self.page_title = f"Error {response.status_code}"350 self._set_page_content(f"## Error {response.status_code}\n\n{text}")351 except NameError:352 self.page_title = "Error"353 self._set_page_content(f"## Error\n\n{str(request_exception)}")354 355 def _state(self) -> Tuple[str, str]:356 header = f"Address: {self.address}\n"357 if self.page_title is not None:358 header += f"Title: {self.page_title}\n"359 360 current_page = self.viewport_current_page361 total_pages = len(self.viewport_pages)362 363 address = self.address364 for i in range(len(self.history) - 2, -1, -1): # Start from the second last365 if self.history[i][0] == address:366 header += f"You previously visited this page {round(time.time() - self.history[i][1])} seconds ago.\n"367 break368 369 header += f"Viewport position: Showing page {current_page + 1} of {total_pages}.\n"370 return (header, self.viewport)371 372 373class SearchInformationTool(Tool):374 name = "web_search"375 description = "Perform a web search query (think a google search) and returns the search results."376 inputs = {"query": {"type": "string", "description": "The web search query to perform."}}377 inputs["filter_year"] = {378 "type": "string",379 "description": "[Optional parameter]: filter the search results to only include pages from a specific year. For example, '2020' will only include pages from 2020. Make sure to use this parameter if you're trying to search for articles from a specific date!",380 "nullable": True,381 }382 output_type = "string"383 384 def __init__(self, browser):385 super().__init__()386 self.browser = browser387 388 def forward(self, query: str, filter_year: Optional[int] = None) -> str:389 self.browser.visit_page(f"google: {query}", filter_year=filter_year)390 header, content = self.browser._state()391 return header.strip() + "\n=======================\n" + content392 393 394class VisitTool(Tool):395 name = "visit_page"396 description = "Visit a webpage at a given URL and return its text. Given a url to a YouTube video, this returns the transcript."397 inputs = {"url": {"type": "string", "description": "The relative or absolute url of the webapge to visit."}}398 output_type = "string"399 400 def __init__(self, browser):401 super().__init__()402 self.browser = browser403 404 def forward(self, url: str) -> str:405 self.browser.visit_page(url)406 header, content = self.browser._state()407 return header.strip() + "\n=======================\n" + content408 409 410class DownloadTool(Tool):411 name = "download_file"412 description = """413Download a file at a given URL. The file should be of this format: [".xlsx", ".pptx", ".wav", ".mp3", ".png", ".docx"]414After using this tool, for further inspection of this page you should return the download path to your manager via final_answer, and they will be able to inspect it.415DO NOT use this tool for .pdf or .txt or .htm files: for these types of files use visit_page with the file url instead."""416 inputs = {"url": {"type": "string", "description": "The relative or absolute url of the file to be downloaded."}}417 output_type = "string"418 419 def __init__(self, browser):420 super().__init__()421 self.browser = browser422 423 def forward(self, url: str) -> str:424 if "arxiv" in url:425 url = url.replace("abs", "pdf")426 response = requests.get(url)427 content_type = response.headers.get("content-type", "")428 extension = mimetypes.guess_extension(content_type)429 if extension and isinstance(extension, str):430 new_path = f"./downloads/file{extension}"431 else:432 new_path = "./downloads/file.object"433 434 with open(new_path, "wb") as f:435 f.write(response.content)436 437 if "pdf" in extension or "txt" in extension or "htm" in extension:438 raise Exception("Do not use this tool for pdf or txt or html files: use visit_page instead.")439 440 return f"File was downloaded and saved under path {new_path}."441 442 443class ArchiveSearchTool(Tool):444 name = "find_archived_url"445 description = "Given a url, searches the Wayback Machine and returns the archived version of the url that's closest in time to the desired date."446 inputs = {447 "url": {"type": "string", "description": "The url you need the archive for."},448 "date": {449 "type": "string",450 "description": "The date that you want to find the archive for. Give this date in the format 'YYYYMMDD', for instance '27 June 2008' is written as '20080627'.",451 },452 }453 output_type = "string"454 455 def __init__(self, browser):456 super().__init__()457 self.browser = browser458 459 def forward(self, url, date) -> str:460 no_timestamp_url = f"https://archive.org/wayback/available?url={url}"461 archive_url = no_timestamp_url + f"×tamp={date}"462 response = requests.get(archive_url).json()463 response_notimestamp = requests.get(no_timestamp_url).json()464 if "archived_snapshots" in response and "closest" in response["archived_snapshots"]:465 closest = response["archived_snapshots"]["closest"]466 print("Archive found!", closest)467 468 elif "archived_snapshots" in response_notimestamp and "closest" in response_notimestamp["archived_snapshots"]:469 closest = response_notimestamp["archived_snapshots"]["closest"]470 print("Archive found!", closest)471 else:472 raise Exception(f"Your {url=} was not archived on Wayback Machine, try a different url.")473 target_url = closest["url"]474 self.browser.visit_page(target_url)475 header, content = self.browser._state()476 return (477 f"Web archive for url {url}, snapshot taken at date {closest['timestamp'][:8]}:\n"478 + header.strip()479 + "\n=======================\n"480 + content481 )482 483 484class PageUpTool(Tool):485 name = "page_up"486 description = "Scroll the viewport UP one page-length in the current webpage and return the new viewport content."487 inputs = {}488 output_type = "string"489 490 def __init__(self, browser):491 super().__init__()492 self.browser = browser493 494 def forward(self) -> str:495 self.browser.page_up()496 header, content = self.browser._state()497 return header.strip() + "\n=======================\n" + content498 499 500class PageDownTool(Tool):501 name = "page_down"502 description = (503 "Scroll the viewport DOWN one page-length in the current webpage and return the new viewport content."504 )505 inputs = {}506 output_type = "string"507 508 def __init__(self, browser):509 super().__init__()510 self.browser = browser511 512 def forward(self) -> str:513 self.browser.page_down()514 header, content = self.browser._state()515 return header.strip() + "\n=======================\n" + content516 517 518class FinderTool(Tool):519 name = "find_on_page_ctrl_f"520 description = "Scroll the viewport to the first occurrence of the search string. This is equivalent to Ctrl+F."521 inputs = {522 "search_string": {523 "type": "string",524 "description": "The string to search for on the page. This search string supports wildcards like '*'",525 }526 }527 output_type = "string"528 529 def __init__(self, browser):530 super().__init__()531 self.browser = browser532 533 def forward(self, search_string: str) -> str:534 find_result = self.browser.find_on_page(search_string)535 header, content = self.browser._state()536 537 if find_result is None:538 return (539 header.strip()540 + f"\n=======================\nThe search string '{search_string}' was not found on this page."541 )542 else:543 return header.strip() + "\n=======================\n" + content544 545 546class FindNextTool(Tool):547 name = "find_next"548 description = "Scroll the viewport to next occurrence of the search string. This is equivalent to finding the next match in a Ctrl+F search."549 inputs = {}550 output_type = "string"551 552 def __init__(self, browser):553 super().__init__()554 self.browser = browser555 556 def forward(self) -> str:557 find_result = self.browser.find_next()558 header, content = self.browser._state()559 560 if find_result is None:561 return header.strip() + "\n=======================\nThe search string was not found on this page."562 else:563 return header.strip() + "\n=======================\n" + content564