Google Search API

.gitignore

@@ -19,6 +19,7 @@ __pycache__/
 *.so
 
 # Distribution / packaging
+.devcontainer/
 .Python
 build/
 develop-eggs/

autogen/agentchat/contrib/web_surfer.py

@@ -9,7 +9,7 @@
 from typing_extensions import Annotated
 
 from ... import Agent, AssistantAgent, ConversableAgent, GroupChat, GroupChatManager, OpenAIWrapper, UserProxyAgent
-from ...browser_utils import SimpleTextBrowser
+from ...browser_utils.browser_creator import TextBrowserEnum
 from ...code_utils import content_str
 from ...oai.openai_utils import filter_config
 from ...token_count_utils import count_token, get_max_token_limit
@@ -40,6 +40,7 @@ def __init__(
         llm_config: Optional[Union[Dict, Literal[False]]] = None,
         summarizer_llm_config: Optional[Union[Dict, Literal[False]]] = None,
         default_auto_reply: Optional[Union[str, Dict, None]] = "",
+        browser_name: Literal['bing', 'google'] = "bing",
         browser_config: Optional[Union[Dict, None]] = None,
     ):
         super().__init__(
@@ -58,7 +59,8 @@ def __init__(
         self._create_summarizer_client(summarizer_llm_config, llm_config)
 
         # Create the browser
-        self.browser = SimpleTextBrowser(**(browser_config if browser_config else {}))
+        self.browser_name = browser_name
+        self.browser = TextBrowserEnum.get_browser(browser_name)
 
         inner_llm_config = copy.deepcopy(llm_config)
 
@@ -136,7 +138,7 @@ def _browser_state() -> Tuple[str, str]:
             description="Perform an INFORMATIONAL web search query then return the search results.",
         )
         def _informational_search(query: Annotated[str, "The informational web search query to perform."]) -> str:
-            self.browser.visit_page(f"bing: {query}")
+            self.browser.visit_page(f"{self.browser_name}: {query}")
             header, content = _browser_state()
             return header.strip() + "\n=======================\n" + content
 
@@ -146,7 +148,7 @@ def _informational_search(query: Annotated[str, "The informational web search qu
             description="Perform a NAVIGATIONAL web search query then immediately navigate to the top result. Useful, for example, to navigate to a particular Wikipedia article or other known destination. Equivalent to Google's \"I'm Feeling Lucky\" button.",
         )
         def _navigational_search(query: Annotated[str, "The navigational web search query to perform."]) -> str:
-            self.browser.visit_page(f"bing: {query}")
+            self.browser.visit_page(f"{self.browser_name}: {query}")
 
             # Extract the first linl
             m = re.search(r"\[.*?\]\((http.*?)\)", self.browser.page_content)

autogen/browser_utils.py

@@ -28,7 +28,7 @@
     pass
 
 
-class SimpleTextBrowser:
+class TextBrowserBase:
     """(In preview) An extremely simple text-based web browser comparable to Lynx. Suitable for Agentic use."""
 
     def __init__(

autogen/browser_utils/base_browser.py

@@ -0,0 +1,216 @@
+import io
+import mimetypes
+import os
+import re
+import uuid
+from typing import Any, Dict, List, Optional, Tuple, Union, overload
+from urllib.parse import urljoin, urlparse
+
+import markdownify
+import requests
+from bs4 import BeautifulSoup
+
+# Optional PDF support
+IS_PDF_CAPABLE = False
+try:
+    import pdfminer
+    import pdfminer.high_level
+
+    IS_PDF_CAPABLE = True
+except ModuleNotFoundError:
+    pass
+
+# Other optional dependencies
+try:
+    import pathvalidate
+except ModuleNotFoundError:
+    pass
+
+
+class TextBrowserBase:
+    """(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,
+        base_url: str = None,
+        api_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[str] = 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.base_url = base_url
+        self.api_key = api_key
+        self.request_kwargs = request_kwargs
+
+        self._page_content = ""
+
+    @property
+    def address(self) -> str:
+        """Return the address of the current page."""
+        return self.history[-1]
+
+    @overload
+    def set_address(self, uri_or_path: str) -> None:
+        self.history.append(uri_or_path)
+
+        self.viewport_current_page = 0
+
+    @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 visit_page(self, path_or_uri: str) -> str:
+        """Update the address, visit the page, and return the content of the viewport."""
+        self.set_address(path_or_uri)
+        return self.viewport
+
+    def _split_pages(self) -> None:
+        # Split only regular pages
+        if not self.address.startswith("http:") and not self.address.startswith("https:"):
+            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 _fetch_page(self, url: str) -> None:
+        try:
+            # 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 returns a status code 200, proceed
+            if response.status_code == 200:
+                content_type = response.headers.get("content-type", "")
+                for ct in ["text/html", "text/plain", "application/pdf"]:
+                    if ct in content_type.lower():
+                        content_type = ct
+                        break
+
+                if content_type == "text/html":
+                    # Get the content of the response
+                    html = ""
+                    for chunk in response.iter_content(chunk_size=512, decode_unicode=True):
+                        html += chunk
+
+                    soup = BeautifulSoup(html, "html.parser")
+
+                    # Remove javascript and style blocks
+                    for script in soup(["script", "style"]):
+                        script.extract()
+
+                    # Convert to markdown -- Wikipedia gets special attention to get a clean version of the page
+                    if url.startswith("https://en.wikipedia.org/"):
+                        body_elm = soup.find("div", {"id": "mw-content-text"})
+                        title_elm = soup.find("span", {"class": "mw-page-title-main"})
+
+                        if body_elm:
+                            # What's the title
+                            main_title = soup.title.string
+                            if title_elm and len(title_elm) > 0:
+                                main_title = title_elm.string
+                            webpage_text = (
+                                "# " + main_title + "\n\n" + markdownify.MarkdownConverter().convert_soup(body_elm)
+                            )
+                        else:
+                            webpage_text = markdownify.MarkdownConverter().convert_soup(soup)
+                    else:
+                        webpage_text = markdownify.MarkdownConverter().convert_soup(soup)
+
+                    # Convert newlines
+                    webpage_text = re.sub(r"\r\n", "\n", webpage_text)
+
+                    # Remove excessive blank lines
+                    self.page_title = soup.title.string
+                    self._set_page_content(re.sub(r"\n{2,}", "\n\n", webpage_text).strip())
+                elif content_type == "text/plain":
+                    # Get the content of the response
+                    plain_text = ""
+                    for chunk in response.iter_content(chunk_size=512, decode_unicode=True):
+                        plain_text += chunk
+
+                    self.page_title = None
+                    self._set_page_content(plain_text)
+                elif IS_PDF_CAPABLE and content_type == "application/pdf":
+                    pdf_data = io.BytesIO(response.raw.read())
+                    self.page_title = None
+                    self._set_page_content(pdfminer.high_level.extract_text(pdf_data))
+                elif self.downloads_folder is not None:
+                    # Try producing a safe filename
+                    fname = None
+                    try:
+                        fname = pathvalidate.sanitize_filename(os.path.basename(urlparse(url).path)).strip()
+                    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
+
+                    # Open a file for writing
+                    download_path = os.path.abspath(os.path.join(self.downloads_folder, fname))
+                    with open(download_path, "wb") as fh:
+                        for chunk in response.iter_content(chunk_size=512):
+                            fh.write(chunk)
+
+                    # Return a page describing what just happened
+                    self.page_title = "Download complete."
+                    self._set_page_content(f"Downloaded '{url}' to '{download_path}'.")
+                else:
+                    self.page_title = f"Error - Unsupported Content-Type '{content_type}'"
+                    self._set_page_content(self.page_title)
+            else:
+                self.page_title = "Error"
+                self._set_page_content("Failed to retrieve " + url)
+        except requests.exceptions.RequestException as e:
+            self.page_title = "Error"
+            self._set_page_content(str(e))