# Shamelessly stolen from Microsoft Autogen team: thanks to them for this great resource!
# https://github.com/microsoft/autogen/blob/gaia_multiagent_v01_march_1st/autogen/browser_utils.py
import mimetypes
import os
import pathlib
import re
import time
import uuid
from typing import Any, Dict, List, Optional, Tuple, Union
from urllib.parse import unquote, urljoin, urlparse
import pathvalidate
import requests
from serpapi import GoogleSearch
from smolagents import Tool
from .cookies import COOKIES
from .mdconvert import FileConversionException, MarkdownConverter, UnsupportedFormatException
class SimpleTextBrowser:
"""(In preview) An extremely simple text-based web browser comparable to Lynx. Suitable for Agentic use."""
def __init__(
self,
start_page: Optional[str] = None,
viewport_size: Optional[int] = 1024 * 8,
downloads_folder: Optional[Union[str, None]] = None,
serpapi_key: Optional[Union[str, None]] = None,
request_kwargs: Optional[Union[Dict[str, Any], None]] = None,
):
self.start_page: str = start_page if start_page else "about:blank"
self.viewport_size = viewport_size # Applies only to the standard uri types
self.downloads_folder = downloads_folder
self.history: List[Tuple[str, float]] = list()
self.page_title: Optional[str] = None
self.viewport_current_page = 0
self.viewport_pages: List[Tuple[int, int]] = list()
self.set_address(self.start_page)
self.serpapi_key = serpapi_key
self.request_kwargs = request_kwargs
self.request_kwargs["cookies"] = COOKIES
self._mdconvert = MarkdownConverter()
self._page_content: str = ""
self._find_on_page_query: Union[str, None] = None
self._find_on_page_last_result: Union[int, None] = None # Location of the last result
@property
def address(self) -> str:
"""Return the address of the current page."""
return self.history[-1][0]
def set_address(self, uri_or_path: str, filter_year: Optional[int] = None) -> None:
# TODO: Handle anchors
self.history.append((uri_or_path, time.time()))
# Handle special URIs
if uri_or_path == "about:blank":
self._set_page_content("")
elif uri_or_path.startswith("google:"):
self._serpapi_search(uri_or_path[len("google:") :].strip(), filter_year=filter_year)
else:
if (
not uri_or_path.startswith("http:")
and not uri_or_path.startswith("https:")
and not uri_or_path.startswith("file:")
):
if len(self.history) > 1:
prior_address = self.history[-2][0]
uri_or_path = urljoin(prior_address, uri_or_path)
# Update the address with the fully-qualified path
self.history[-1] = (uri_or_path, self.history[-1][1])
self._fetch_page(uri_or_path)
self.viewport_current_page = 0
self.find_on_page_query = None
self.find_on_page_viewport = None
@property
def viewport(self) -> str:
"""Return the content of the current viewport."""
bounds = self.viewport_pages[self.viewport_current_page]
return self.page_content[bounds[0] : bounds[1]]
@property
def page_content(self) -> str:
"""Return the full contents of the current page."""
return self._page_content
def _set_page_content(self, content: str) -> None:
"""Sets the text content of the current page."""
self._page_content = content
self._split_pages()
if self.viewport_current_page >= len(self.viewport_pages):
self.viewport_current_page = len(self.viewport_pages) - 1
def page_down(self) -> None:
self.viewport_current_page = min(self.viewport_current_page + 1, len(self.viewport_pages) - 1)
def page_up(self) -> None:
self.viewport_current_page = max(self.viewport_current_page - 1, 0)
def find_on_page(self, query: str) -> Union[str, None]:
"""Searches for the query from the current viewport forward, looping back to the start if necessary."""
# Did we get here via a previous find_on_page search with the same query?
# If so, map to find_next
if query == self._find_on_page_query and self.viewport_current_page == self._find_on_page_last_result:
return self.find_next()
# Ok it's a new search start from the current viewport
self._find_on_page_query = query
viewport_match = self._find_next_viewport(query, self.viewport_current_page)
if viewport_match is None:
self._find_on_page_last_result = None
return None
else:
self.viewport_current_page = viewport_match
self._find_on_page_last_result = viewport_match
return self.viewport
def find_next(self) -> Union[str, None]:
"""Scroll to the next viewport that matches the query"""
if self._find_on_page_query is None:
return None
starting_viewport = self._find_on_page_last_result
if starting_viewport is None:
starting_viewport = 0
else:
starting_viewport += 1
if starting_viewport >= len(self.viewport_pages):
starting_viewport = 0
viewport_match = self._find_next_viewport(self._find_on_page_query, starting_viewport)
if viewport_match is None:
self._find_on_page_last_result = None
return None
else:
self.viewport_current_page = viewport_match
self._find_on_page_last_result = viewport_match
return self.viewport
def _find_next_viewport(self, query: str, starting_viewport: int) -> Union[int, None]:
"""Search for matches between the starting viewport looping when reaching the end."""
if query is None:
return None
# Normalize the query, and convert to a regular expression
nquery = re.sub(r"\*", "__STAR__", query)
nquery = " " + (" ".join(re.split(r"\W+", nquery))).strip() + " "
nquery = nquery.replace(" __STAR__ ", "__STAR__ ") # Merge isolated stars with prior word
nquery = nquery.replace("__STAR__", ".*").lower()
if nquery.strip() == "":
return None
idxs = list()
idxs.extend(range(starting_viewport, len(self.viewport_pages)))
idxs.extend(range(0, starting_viewport))
for i in idxs:
bounds = self.viewport_pages[i]
content = self.page_content[bounds[0] : bounds[1]]
# TODO: Remove markdown links and images
ncontent = " " + (" ".join(re.split(r"\W+", content))).strip().lower() + " "
if re.search(nquery, ncontent):
return i
return None
def visit_page(self, path_or_uri: str, filter_year: Optional[int] = None) -> str:
"""Update the address, visit the page, and return the content of the viewport."""
self.set_address(path_or_uri, filter_year=filter_year)
return self.viewport
def _split_pages(self) -> None:
# Do not split search results
if self.address.startswith("google:"):
self.viewport_pages = [(0, len(self._page_content))]
return
# Handle empty pages
if len(self._page_content) == 0:
self.viewport_pages = [(0, 0)]
return
# Break the viewport into pages
self.viewport_pages = []
start_idx = 0
while start_idx < len(self._page_content):
end_idx = min(start_idx + self.viewport_size, len(self._page_content)) # type: ignore[operator]
# Adjust to end on a space
while end_idx < len(self._page_content) and self._page_content[end_idx - 1] not in [" ", "\t", "\r", "\n"]:
end_idx += 1
self.viewport_pages.append((start_idx, end_idx))
start_idx = end_idx
def _serpapi_search(self, query: str, filter_year: Optional[int] = None) -> None:
if self.serpapi_key is None:
raise ValueError("Missing SerpAPI key.")
params = {
"engine": "google",
"q": query,
"api_key": self.serpapi_key,
}
if filter_year is not None:
params["tbs"] = f"cdr:1,cd_min:01/01/{filter_year},cd_max:12/31/{filter_year}"
search = GoogleSearch(params)
results = search.get_dict()
self.page_title = f"{query} - Search"
if "organic_results" not in results.keys():
raise Exception(f"No results found for query: '{query}'. Use a less specific query.")
if len(results["organic_results"]) == 0:
year_filter_message = f" with filter year={filter_year}" if filter_year is not None else ""
self._set_page_content(
f"No results found for '{query}'{year_filter_message}. Try with a more general query, or remove the year filter."
)
return
def _prev_visit(url):
for i in range(len(self.history) - 1, -1, -1):
if self.history[i][0] == url:
return f"You previously visited this page {round(time.time() - self.history[i][1])} seconds ago.\n"
return ""
web_snippets: List[str] = list()
idx = 0
if "organic_results" in results:
for page in results["organic_results"]:
idx += 1
date_published = ""
if "date" in page:
date_published = "\nDate published: " + page["date"]
source = ""
if "source" in page:
source = "\nSource: " + page["source"]
snippet = ""
if "snippet" in page:
snippet = "\n" + page["snippet"]
redacted_version = f"{idx}. [{page['title']}]({page['link']}){date_published}{source}\n{_prev_visit(page['link'])}{snippet}"
redacted_version = redacted_version.replace("Your browser can't play this video.", "")
web_snippets.append(redacted_version)
content = (
f"A Google search for '{query}' found {len(web_snippets)} results:\n\n## Web Results\n"
+ "\n\n".join(web_snippets)
)
self._set_page_content(content)
def _fetch_page(self, url: str) -> None:
download_path = ""
try:
if url.startswith("file://"):
download_path = os.path.normcase(os.path.normpath(unquote(url[7:])))
res = self._mdconvert.convert_local(download_path)
self.page_title = res.title
self._set_page_content(res.text_content)
else:
# Prepare the request parameters
request_kwargs = self.request_kwargs.copy() if self.request_kwargs is not None else {}
request_kwargs["stream"] = True
# Send a HTTP request to the URL
response = requests.get(url, **request_kwargs)
response.raise_for_status()
# If the HTTP request was successful
content_type = response.headers.get("content-type", "")
# Text or HTML
if "text/" in content_type.lower():
res = self._mdconvert.convert_response(response)
self.page_title = res.title
self._set_page_content(res.text_content)
# A download
else:
# Try producing a safe filename
fname = None
download_path = None
try:
fname = pathvalidate.sanitize_filename(os.path.basename(urlparse(url).path)).strip()
download_path = os.path.abspath(os.path.join(self.downloads_folder, fname))
suffix = 0
while os.path.exists(download_path) and suffix < 1000:
suffix += 1
base, ext = os.path.splitext(fname)
new_fname = f"{base}__{suffix}{ext}"
download_path = os.path.abspath(os.path.join(self.downloads_folder, new_fname))
except NameError:
pass
# No suitable name, so make one
if fname is None:
extension = mimetypes.guess_extension(content_type)
if extension is None:
extension = ".download"
fname = str(uuid.uuid4()) + extension
download_path = os.path.abspath(os.path.join(self.downloads_folder, fname))
# Open a file for writing
with open(download_path, "wb") as fh:
for chunk in response.iter_content(chunk_size=512):
fh.write(chunk)
# Render it
local_uri = pathlib.Path(download_path).as_uri()
self.set_address(local_uri)
except UnsupportedFormatException as e:
print(e)
self.page_title = ("Download complete.",)
self._set_page_content(f"# Download complete\n\nSaved file to '{download_path}'")
except FileConversionException as e:
print(e)
self.page_title = ("Download complete.",)
self._set_page_content(f"# Download complete\n\nSaved file to '{download_path}'")
except FileNotFoundError:
self.page_title = "Error 404"
self._set_page_content(f"## Error 404\n\nFile not found: {download_path}")
except requests.exceptions.RequestException as request_exception:
try:
self.page_title = f"Error {response.status_code}"
# If the error was rendered in HTML we might as well render it
content_type = response.headers.get("content-type", "")
if content_type is not None and "text/html" in content_type.lower():
res = self._mdconvert.convert(response)
self.page_title = f"Error {response.status_code}"
self._set_page_content(f"## Error {response.status_code}\n\n{res.text_content}")
else:
text = ""
for chunk in response.iter_content(chunk_size=512, decode_unicode=True):
text += chunk
self.page_title = f"Error {response.status_code}"
self._set_page_content(f"## Error {response.status_code}\n\n{text}")
except NameError:
self.page_title = "Error"
self._set_page_content(f"## Error\n\n{str(request_exception)}")
def _state(self) -> Tuple[str, str]:
header = f"Address: {self.address}\n"
if self.page_title is not None:
header += f"Title: {self.page_title}\n"
current_page = self.viewport_current_page
total_pages = len(self.viewport_pages)
address = self.address
for i in range(len(self.history) - 2, -1, -1): # Start from the second last
if self.history[i][0] == address:
header += f"You previously visited this page {round(time.time() - self.history[i][1])} seconds ago.\n"
break
header += f"Viewport position: Showing page {current_page + 1} of {total_pages}.\n"
return (header, self.viewport)
class SearchInformationTool(Tool):
name = "web_search"
description = "Perform a web search query (think a google search) and returns the search results."
inputs = {"query": {"type": "string", "description": "The web search query to perform."}}
inputs["filter_year"] = {
"type": "string",
"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!",
"nullable": True,
}
output_type = "string"
def __init__(self, browser):
super().__init__()
self.browser = browser
def forward(self, query: str, filter_year: Optional[int] = None) -> str:
self.browser.visit_page(f"google: {query}", filter_year=filter_year)
header, content = self.browser._state()
return header.strip() + "\n=======================\n" + content
class VisitTool(Tool):
name = "visit_page"
description = "Visit a webpage at a given URL and return its text. Given a url to a YouTube video, this returns the transcript."
inputs = {"url": {"type": "string", "description": "The relative or absolute url of the webapge to visit."}}
output_type = "string"
def __init__(self, browser):
super().__init__()
self.browser = browser
def forward(self, url: str) -> str:
self.browser.visit_page(url)
header, content = self.browser._state()
return header.strip() + "\n=======================\n" + content
class DownloadTool(Tool):
name = "download_file"
description = """
Download a file at a given URL. The file should be of this format: [".xlsx", ".pptx", ".wav", ".mp3", ".png", ".docx"]
After 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.
DO NOT use this tool for .pdf or .txt or .htm files: for these types of files use visit_page with the file url instead."""
inputs = {"url": {"type": "string", "description": "The relative or absolute url of the file to be downloaded."}}
output_type = "string"
def __init__(self, browser):
super().__init__()
self.browser = browser
def forward(self, url: str) -> str:
if "arxiv" in url:
url = url.replace("abs", "pdf")
response = requests.get(url)
content_type = response.headers.get("content-type", "")
extension = mimetypes.guess_extension(content_type)
if extension and isinstance(extension, str):
new_path = f"./downloads/file{extension}"
else:
new_path = "./downloads/file.object"
with open(new_path, "wb") as f:
f.write(response.content)
if "pdf" in extension or "txt" in extension or "htm" in extension:
raise Exception("Do not use this tool for pdf or txt or html files: use visit_page instead.")
return f"File was downloaded and saved under path {new_path}."
class ArchiveSearchTool(Tool):
name = "find_archived_url"
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."
inputs = {
"url": {"type": "string", "description": "The url you need the archive for."},
"date": {
"type": "string",
"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'.",
},
}
output_type = "string"
def __init__(self, browser):
super().__init__()
self.browser = browser
def forward(self, url, date) -> str:
no_timestamp_url = f"https://archive.org/wayback/available?url={url}"
archive_url = no_timestamp_url + f"×tamp={date}"
response = requests.get(archive_url).json()
response_notimestamp = requests.get(no_timestamp_url).json()
if "archived_snapshots" in response and "closest" in response["archived_snapshots"]:
closest = response["archived_snapshots"]["closest"]
print("Archive found!", closest)
elif "archived_snapshots" in response_notimestamp and "closest" in response_notimestamp["archived_snapshots"]:
closest = response_notimestamp["archived_snapshots"]["closest"]
print("Archive found!", closest)
else:
raise Exception(f"Your {url=} was not archived on Wayback Machine, try a different url.")
target_url = closest["url"]
self.browser.visit_page(target_url)
header, content = self.browser._state()
return (
f"Web archive for url {url}, snapshot taken at date {closest['timestamp'][:8]}:\n"
+ header.strip()
+ "\n=======================\n"
+ content
)
class PageUpTool(Tool):
name = "page_up"
description = "Scroll the viewport UP one page-length in the current webpage and return the new viewport content."
inputs = {}
output_type = "string"
def __init__(self, browser):
super().__init__()
self.browser = browser
def forward(self) -> str:
self.browser.page_up()
header, content = self.browser._state()
return header.strip() + "\n=======================\n" + content
class PageDownTool(Tool):
name = "page_down"
description = (
"Scroll the viewport DOWN one page-length in the current webpage and return the new viewport content."
)
inputs = {}
output_type = "string"
def __init__(self, browser):
super().__init__()
self.browser = browser
def forward(self) -> str:
self.browser.page_down()
header, content = self.browser._state()
return header.strip() + "\n=======================\n" + content
class FinderTool(Tool):
name = "find_on_page_ctrl_f"
description = "Scroll the viewport to the first occurrence of the search string. This is equivalent to Ctrl+F."
inputs = {
"search_string": {
"type": "string",
"description": "The string to search for on the page. This search string supports wildcards like '*'",
}
}
output_type = "string"
def __init__(self, browser):
super().__init__()
self.browser = browser
def forward(self, search_string: str) -> str:
find_result = self.browser.find_on_page(search_string)
header, content = self.browser._state()
if find_result is None:
return (
header.strip()
+ f"\n=======================\nThe search string '{search_string}' was not found on this page."
)
else:
return header.strip() + "\n=======================\n" + content
class FindNextTool(Tool):
name = "find_next"
description = "Scroll the viewport to next occurrence of the search string. This is equivalent to finding the next match in a Ctrl+F search."
inputs = {}
output_type = "string"
def __init__(self, browser):
super().__init__()
self.browser = browser
def forward(self) -> str:
find_result = self.browser.find_next()
header, content = self.browser._state()
if find_result is None:
return header.strip() + "\n=======================\nThe search string was not found on this page."
else:
return header.strip() + "\n=======================\n" + content