From 3d6794416580a2e8dd1078d57225087e3f8f9d11 Mon Sep 17 00:00:00 2001 From: Davor Runje Date: Sat, 10 Feb 2024 06:09:08 +0100 Subject: [PATCH] Refactoring web surfer to use function decorators (#1435) * refactoring web surfer to use function decorators * limited pytest version * bug fix in test * bug fixes * refactoring * Fix web_surfer tests. --------- Co-authored-by: Chi Wang Co-authored-by: Adam Fourney --- autogen/agentchat/contrib/web_surfer.py | 366 +++++++++------------- autogen/browser_utils.py | 43 ++- autogen/token_count_utils.py | 2 +- test/agentchat/contrib/test_web_surfer.py | 127 ++++---- 4 files changed, 245 insertions(+), 293 deletions(-) diff --git a/autogen/agentchat/contrib/web_surfer.py b/autogen/agentchat/contrib/web_surfer.py index 4877a4d0949d..9b7320f092fe 100644 --- a/autogen/agentchat/contrib/web_surfer.py +++ b/autogen/agentchat/contrib/web_surfer.py @@ -3,13 +3,14 @@ import logging import re from dataclasses import dataclass -from typing import Dict, List, Optional, Union, Callable, Literal, Tuple -from autogen import Agent, ConversableAgent, AssistantAgent, UserProxyAgent, GroupChatManager, GroupChat, OpenAIWrapper -from autogen.browser_utils import SimpleTextBrowser -from autogen.code_utils import content_str +from typing import Any, Dict, List, Optional, Union, Callable, Literal, Tuple +from typing_extensions import Annotated +from ... import Agent, ConversableAgent, AssistantAgent, UserProxyAgent, GroupChatManager, GroupChat, OpenAIWrapper +from ...browser_utils import SimpleTextBrowser +from ...code_utils import content_str from datetime import datetime -from autogen.token_count_utils import count_token, get_max_token_limit -from autogen.oai.openai_utils import filter_config +from ...token_count_utils import count_token, get_max_token_limit +from ...oai.openai_utils import filter_config logger = logging.getLogger(__name__) @@ -26,10 +27,10 @@ class WebSurferAgent(ConversableAgent): def __init__( self, - name, - system_message: Optional[Union[str, List]] = DEFAULT_PROMPT, + name: str, + system_message: Optional[Union[str, List[str]]] = DEFAULT_PROMPT, description: Optional[str] = DEFAULT_DESCRIPTION, - is_termination_msg: Optional[Callable[[Dict], bool]] = None, + is_termination_msg: Optional[Callable[[Dict[str, Any]], bool]] = None, max_consecutive_auto_reply: Optional[int] = None, human_input_mode: Optional[str] = "TERMINATE", function_map: Optional[Dict[str, Callable]] = None, @@ -52,6 +53,39 @@ def __init__( default_auto_reply=default_auto_reply, ) + self._create_summarizer_client(summarizer_llm_config, llm_config) + + # Create the browser + self.browser = SimpleTextBrowser(**(browser_config if browser_config else {})) + + inner_llm_config = copy.deepcopy(llm_config) + + # Set up the inner monologue + self._assistant = AssistantAgent( + self.name + "_inner_assistant", + system_message=system_message, # type: ignore[arg-type] + llm_config=inner_llm_config, + is_termination_msg=lambda m: False, + ) + + self._user_proxy = UserProxyAgent( + self.name + "_inner_user_proxy", + human_input_mode="NEVER", + code_execution_config=False, + default_auto_reply="", + is_termination_msg=lambda m: False, + ) + + if inner_llm_config not in [None, False]: + self._register_functions() + + self._reply_func_list = [] + self.register_reply([Agent, None], WebSurferAgent.generate_surfer_reply) + self.register_reply([Agent, None], ConversableAgent.generate_code_execution_reply) + self.register_reply([Agent, None], ConversableAgent.generate_function_call_reply) + self.register_reply([Agent, None], ConversableAgent.check_termination_and_human_reply) + + def _create_summarizer_client(self, summarizer_llm_config: Dict[str, Any], llm_config: Dict[str, Any]) -> None: # If the summarizer_llm_config is None, we copy it from the llm_config if summarizer_llm_config is None: if llm_config is None: # Nothing to copy @@ -59,10 +93,10 @@ def __init__( elif llm_config is False: # LLMs disabled self.summarizer_llm_config = False else: # Create a suitable config - self.summarizer_llm_config = copy.deepcopy(llm_config) - if "config_list" in self.summarizer_llm_config: - preferred_models = filter_config( - self.summarizer_llm_config["config_list"], + self.summarizer_llm_config = copy.deepcopy(llm_config) # type: ignore[assignment] + if "config_list" in self.summarizer_llm_config: # type: ignore[operator] + preferred_models = filter_config( # type: ignore[no-untyped-call] + self.summarizer_llm_config["config_list"], # type: ignore[index] {"model": ["gpt-3.5-turbo-1106", "gpt-3.5-turbo-16k-0613", "gpt-3.5-turbo-16k"]}, ) if len(preferred_models) == 0: @@ -71,142 +105,18 @@ def __init__( "Semantic operations on webpages (summarization or Q&A) might be costly or ineffective." ) else: - self.summarizer_llm_config["config_list"] = preferred_models + self.summarizer_llm_config["config_list"] = preferred_models # type: ignore[index] else: - self.summarizer_llm_config = summarizer_llm_config + self.summarizer_llm_config = summarizer_llm_config # type: ignore[assignment] # Create the summarizer client - self.summarization_client = None - if self.summarizer_llm_config is not False: - self.summarization_client = OpenAIWrapper(**self.summarizer_llm_config) - - # Create the browser - if browser_config is None: - self.browser = SimpleTextBrowser() - else: - self.browser = SimpleTextBrowser(**browser_config) - - # Create a copy of the llm_config for the inner monologue agents to use, and set them up with function calling - if llm_config is None: # Nothing to copy - inner_llm_config = None - elif llm_config is False: # LLMs disabled - inner_llm_config = False - else: - inner_llm_config = copy.deepcopy(llm_config) - inner_llm_config["functions"] = [ - { - "name": "informational_web_search", - "description": "Perform an INFORMATIONAL web search query then return the search results.", - "parameters": { - "type": "object", - "properties": { - "query": { - "type": "string", - "description": "The informational web search query to perform.", - } - }, - }, - "required": ["query"], - }, - { - "name": "navigational_web_search", - "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.", - "parameters": { - "type": "object", - "properties": { - "query": { - "type": "string", - "description": "The navigational web search query to perform.", - } - }, - }, - "required": ["query"], - }, - { - "name": "visit_page", - "description": "Visit a webpage at a given URL and return its text.", - "parameters": { - "type": "object", - "properties": { - "url": { - "type": "string", - "description": "The relative or absolute url of the webapge to visit.", - } - }, - }, - "required": ["url"], - }, - { - "name": "page_up", - "description": "Scroll the viewport UP one page-length in the current webpage and return the new viewport content.", - "parameters": {"type": "object", "properties": {}}, - "required": [], - }, - { - "name": "page_down", - "description": "Scroll the viewport DOWN one page-length in the current webpage and return the new viewport content.", - "parameters": {"type": "object", "properties": {}}, - "required": [], - }, - ] - - # Enable semantic operations - if self.summarization_client is not None: - inner_llm_config["functions"].append( - { - "name": "answer_from_page", - "description": "Uses AI to read the page and directly answer a given question based on the content.", - "parameters": { - "type": "object", - "properties": { - "question": { - "type": "string", - "description": "The question to directly answer.", - }, - "url": { - "type": "string", - "description": "[Optional] The url of the page. (Defaults to the current page)", - }, - }, - }, - "required": ["question"], - } - ) - inner_llm_config["functions"].append( - { - "name": "summarize_page", - "description": "Uses AI to summarize the content found at a given url. If the url is not provided, the current page is summarized.", - "parameters": { - "type": "object", - "properties": { - "url": { - "type": "string", - "description": "[Optional] The url of the page to summarize. (Defaults to current page)", - }, - }, - }, - "required": [], - } - ) + self.summarization_client = None if self.summarizer_llm_config is False else OpenAIWrapper(**self.summarizer_llm_config) # type: ignore[arg-type] - # Set up the inner monologue - self._assistant = AssistantAgent( - self.name + "_inner_assistant", - system_message=system_message, - llm_config=inner_llm_config, - is_termination_msg=lambda m: False, - ) - - self._user_proxy = UserProxyAgent( - self.name + "_inner_user_proxy", - human_input_mode="NEVER", - code_execution_config=False, - default_auto_reply="", - is_termination_msg=lambda m: False, - ) + def _register_functions(self) -> None: + """Register the functions for the inner assistant and user proxy.""" # Helper functions - def _browser_state(): + def _browser_state() -> Tuple[str, str]: header = f"Address: {self.browser.address}\n" if self.browser.page_title is not None: header += f"Title: {self.browser.page_title}\n" @@ -217,12 +127,22 @@ def _browser_state(): header += f"Viewport position: Showing page {current_page+1} of {total_pages}.\n" return (header, self.browser.viewport) - def _informational_search(query): + @self._user_proxy.register_for_execution() + @self._assistant.register_for_llm( + name="informational_web_search", + 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}") header, content = _browser_state() return header.strip() + "\n=======================\n" + content - def _navigational_search(query): + @self._user_proxy.register_for_execution() + @self._assistant.register_for_llm( + name="navigational_web_search", + 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}") # Extract the first linl @@ -234,99 +154,117 @@ def _navigational_search(query): header, content = _browser_state() return header.strip() + "\n=======================\n" + content - def _visit_page(url): + @self._user_proxy.register_for_execution() + @self._assistant.register_for_llm( + name="visit_page", description="Visit a webpage at a given URL and return its text." + ) + def _visit_page(url: Annotated[str, "The relative or absolute url of the webapge to visit."]) -> str: self.browser.visit_page(url) header, content = _browser_state() return header.strip() + "\n=======================\n" + content - def _page_up(): + @self._user_proxy.register_for_execution() + @self._assistant.register_for_llm( + name="page_up", + description="Scroll the viewport UP one page-length in the current webpage and return the new viewport content.", + ) + def _page_up() -> str: self.browser.page_up() header, content = _browser_state() return header.strip() + "\n=======================\n" + content - def _page_down(): + @self._user_proxy.register_for_execution() + @self._assistant.register_for_llm( + name="page_down", + description="Scroll the viewport DOWN one page-length in the current webpage and return the new viewport content.", + ) + def _page_down() -> str: self.browser.page_down() header, content = _browser_state() return header.strip() + "\n=======================\n" + content - def _summarize_page(question, url): - if url is not None and url != self.browser.address: - self.browser.visit_page(url) - - # We are likely going to need to fix this later, but summarize only as many tokens that fit in the buffer - limit = 4096 - try: - limit = get_max_token_limit(self.summarizer_llm_config["config_list"][0]["model"]) - except ValueError: - pass # limit is unknown - except TypeError: - pass # limit is unknown - - if limit < 16000: - logger.warning( - f"The token limit ({limit}) of the WebSurferAgent.summarizer_llm_config, is below the recommended 16k." - ) + if self.summarization_client is not None: - buffer = "" - for line in re.split(r"([\r\n]+)", self.browser.page_content): - tokens = count_token(buffer + line) - if tokens + 1024 > limit: # Leave room for our summary - break - buffer += line - - buffer = buffer.strip() - if len(buffer) == 0: - return "Nothing to summarize." - - messages = [ - { - "role": "system", - "content": "You are a helpful assistant that can summarize long documents to answer question.", - } - ] - - prompt = f"Please summarize the following into one or two paragraph:\n\n{buffer}" - if question is not None: - prompt = f"Please summarize the following into one or two paragraphs with respect to '{question}':\n\n{buffer}" - - messages.append( - {"role": "user", "content": prompt}, + @self._user_proxy.register_for_execution() + @self._assistant.register_for_llm( + name="answer_from_page", + description="Uses AI to read the page and directly answer a given question based on the content.", ) + def _answer_from_page( + question: Annotated[Optional[str], "The question to directly answer."], + url: Annotated[Optional[str], "[Optional] The url of the page. (Defaults to the current page)"] = None, + ) -> str: + if url is not None and url != self.browser.address: + self.browser.visit_page(url) + + # We are likely going to need to fix this later, but summarize only as many tokens that fit in the buffer + limit = 4096 + try: + limit = get_max_token_limit(self.summarizer_llm_config["config_list"][0]["model"]) # type: ignore[index] + except ValueError: + pass # limit is unknown + except TypeError: + pass # limit is unknown + + if limit < 16000: + logger.warning( + f"The token limit ({limit}) of the WebSurferAgent.summarizer_llm_config, is below the recommended 16k." + ) - response = self.summarization_client.create(context=None, messages=messages) - extracted_response = self.summarization_client.extract_text_or_completion_object(response)[0] - return str(extracted_response) - - self._user_proxy.register_function( - function_map={ - "informational_web_search": lambda query: _informational_search(query), - "navigational_web_search": lambda query: _navigational_search(query), - "visit_page": lambda url: _visit_page(url), - "page_up": lambda: _page_up(), - "page_down": lambda: _page_down(), - "answer_from_page": lambda question=None, url=None: _summarize_page(question, url), - "summarize_page": lambda question=None, url=None: _summarize_page(None, url), - } - ) + buffer = "" + for line in re.split(r"([\r\n]+)", self.browser.page_content): + tokens = count_token(buffer + line) + if tokens + 1024 > limit: # Leave room for our summary + break + buffer += line - self._reply_func_list = [] - self.register_reply([Agent, None], WebSurferAgent.generate_surfer_reply) - self.register_reply([Agent, None], ConversableAgent.generate_code_execution_reply) - self.register_reply([Agent, None], ConversableAgent.generate_function_call_reply) - self.register_reply([Agent, None], ConversableAgent.check_termination_and_human_reply) + buffer = buffer.strip() + if len(buffer) == 0: + return "Nothing to summarize." + + messages = [ + { + "role": "system", + "content": "You are a helpful assistant that can summarize long documents to answer question.", + } + ] + + prompt = f"Please summarize the following into one or two paragraph:\n\n{buffer}" + if question is not None: + prompt = f"Please summarize the following into one or two paragraphs with respect to '{question}':\n\n{buffer}" + + messages.append( + {"role": "user", "content": prompt}, + ) + + response = self.summarization_client.create(context=None, messages=messages) # type: ignore[union-attr] + extracted_response = self.summarization_client.extract_text_or_completion_object(response)[0] # type: ignore[union-attr] + return str(extracted_response) + + @self._user_proxy.register_for_execution() + @self._assistant.register_for_llm( + name="summarize_page", + description="Uses AI to summarize the content found at a given url. If the url is not provided, the current page is summarized.", + ) + def _summarize_page( + url: Annotated[ + Optional[str], "[Optional] The url of the page to summarize. (Defaults to current page)" + ] = None + ) -> str: + return _answer_from_page(url=url, question=None) def generate_surfer_reply( self, - messages: Optional[List[Dict]] = None, + messages: Optional[List[Dict[str, str]]] = None, sender: Optional[Agent] = None, config: Optional[OpenAIWrapper] = None, - ) -> Tuple[bool, Union[str, Dict, None]]: + ) -> Tuple[bool, Optional[Union[str, Dict[str, str]]]]: """Generate a reply using autogen.oai.""" if messages is None: messages = self._oai_messages[sender] - self._user_proxy.reset() - self._assistant.reset() + self._user_proxy.reset() # type: ignore[no-untyped-call] + self._assistant.reset() # type: ignore[no-untyped-call] # Clone the messages to give context self._assistant.chat_messages[self._user_proxy] = list() @@ -353,4 +291,4 @@ def generate_surfer_reply( if proxy_reply == "": # Was the default reply return True, None if agent_reply is None else agent_reply["content"] else: - return True, None if proxy_reply is None else proxy_reply["content"] + return True, None if proxy_reply is None else proxy_reply["content"] # type: ignore[index] diff --git a/autogen/browser_utils.py b/autogen/browser_utils.py index 68e39e4ac8e6..41d2d62f825b 100644 --- a/autogen/browser_utils.py +++ b/autogen/browser_utils.py @@ -8,8 +8,7 @@ import mimetypes from urllib.parse import urljoin, urlparse from bs4 import BeautifulSoup -from dataclasses import dataclass -from typing import Dict, List, Optional, Union, Callable, Literal, Tuple +from typing import Any, Dict, List, Optional, Union, Tuple # Optional PDF support IS_PDF_CAPABLE = False @@ -33,20 +32,20 @@ class SimpleTextBrowser: def __init__( self, - start_page: Optional[str] = "about:blank", + start_page: Optional[str] = None, viewport_size: Optional[int] = 1024 * 8, downloads_folder: Optional[Union[str, None]] = None, bing_api_key: Optional[Union[str, None]] = None, - request_kwargs: Optional[Union[Dict, None]] = None, + request_kwargs: Optional[Union[Dict[str, Any], None]] = None, ): - self.start_page = start_page + 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() - self.page_title = None + self.history: List[str] = list() + self.page_title: Optional[str] = None self.viewport_current_page = 0 - self.viewport_pages = list() - self.set_address(start_page) + self.viewport_pages: List[Tuple[int, int]] = list() + self.set_address(self.start_page) self.bing_api_key = bing_api_key self.request_kwargs = request_kwargs @@ -57,7 +56,7 @@ def address(self) -> str: """Return the address of the current page.""" return self.history[-1] - def set_address(self, uri_or_path): + def set_address(self, uri_or_path: str) -> None: self.history.append(uri_or_path) # Handle special URIs @@ -84,25 +83,25 @@ def page_content(self) -> str: """Return the full contents of the current page.""" return self._page_content - def _set_page_content(self, content) -> str: + 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): + def page_down(self) -> None: self.viewport_current_page = min(self.viewport_current_page + 1, len(self.viewport_pages) - 1) - def page_up(self): + def page_up(self) -> None: self.viewport_current_page = max(self.viewport_current_page - 1, 0) - def visit_page(self, path_or_uri): + 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): + 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))] @@ -117,14 +116,14 @@ def _split_pages(self): 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)) + 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 _bing_api_call(self, query): + def _bing_api_call(self, query: str) -> Dict[str, Dict[str, List[Dict[str, Union[str, Dict[str, str]]]]]]: # Make sure the key was set if self.bing_api_key is None: raise ValueError("Missing Bing API key.") @@ -149,12 +148,12 @@ def _bing_api_call(self, query): response.raise_for_status() results = response.json() - return results + return results # type: ignore[no-any-return] - def _bing_search(self, query): + def _bing_search(self, query: str) -> None: results = self._bing_api_call(query) - web_snippets = list() + web_snippets: List[str] = list() idx = 0 for page in results["webPages"]["value"]: idx += 1 @@ -163,7 +162,7 @@ def _bing_search(self, query): for dl in page["deepLinks"]: idx += 1 web_snippets.append( - f"{idx}. [{dl['name']}]({dl['url']})\n{dl['snippet'] if 'snippet' in dl else ''}" + f"{idx}. [{dl['name']}]({dl['url']})\n{dl['snippet'] if 'snippet' in dl else ''}" # type: ignore[index] ) news_snippets = list() @@ -182,7 +181,7 @@ def _bing_search(self, query): content += "\n\n## News Results:\n" + "\n\n".join(news_snippets) self._set_page_content(content) - def _fetch_page(self, url): + 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 {} diff --git a/autogen/token_count_utils.py b/autogen/token_count_utils.py index 1b1b2f199253..84fe147fd8eb 100644 --- a/autogen/token_count_utils.py +++ b/autogen/token_count_utils.py @@ -8,7 +8,7 @@ logger = logging.getLogger(__name__) -def get_max_token_limit(model="gpt-3.5-turbo-0613"): +def get_max_token_limit(model: str = "gpt-3.5-turbo-0613") -> int: # Handle common azure model names/aliases model = re.sub(r"^gpt\-?35", "gpt-3.5", model) model = re.sub(r"^gpt4", "gpt-4", model) diff --git a/test/agentchat/contrib/test_web_surfer.py b/test/agentchat/contrib/test_web_surfer.py index 307abefecbca..abf7903dee7a 100644 --- a/test/agentchat/contrib/test_web_surfer.py +++ b/test/agentchat/contrib/test_web_surfer.py @@ -2,8 +2,9 @@ import sys import re import pytest -from autogen import ConversableAgent, UserProxyAgent, config_list_from_json +from autogen import UserProxyAgent, config_list_from_json from autogen.oai.openai_utils import filter_config +from autogen.cache import Cache sys.path.append(os.path.join(os.path.dirname(__file__), "../..")) from conftest import skip_openai # noqa: E402 @@ -44,70 +45,76 @@ skip_all, reason="do not run if dependency is not installed", ) -def test_web_surfer(): - page_size = 4096 - web_surfer = WebSurferAgent("web_surfer", llm_config=False, browser_config={"viewport_size": page_size}) - - # Sneak a peak at the function map, allowing us to call the functions for testing here - function_map = web_surfer._user_proxy._function_map - - # Test some basic navigations - response = function_map["visit_page"](BLOG_POST_URL) - assert f"Address: {BLOG_POST_URL}".strip() in response - assert f"Title: {BLOG_POST_TITLE}".strip() in response +def test_web_surfer() -> None: + with pytest.MonkeyPatch.context() as mp: + # we mock the API key so we can register functions (llm_config must be present for this to work) + mp.setenv("OPENAI_API_KEY", "mock") + page_size = 4096 + web_surfer = WebSurferAgent( + "web_surfer", llm_config={"config_list": []}, browser_config={"viewport_size": page_size} + ) + + # Sneak a peak at the function map, allowing us to call the functions for testing here + function_map = web_surfer._user_proxy._function_map + + # Test some basic navigations + response = function_map["visit_page"](BLOG_POST_URL) + assert f"Address: {BLOG_POST_URL}".strip() in response + assert f"Title: {BLOG_POST_TITLE}".strip() in response + + # Test scrolling + m = re.search(r"\bViewport position: Showing page 1 of (\d+).", response) + total_pages = int(m.group(1)) # type: ignore[union-attr] - # Test scrolling - m = re.search(r"\bViewport position: Showing page 1 of (\d+).", response) - total_pages = int(m.group(1)) - - response = function_map["page_down"]() - assert ( - f"Viewport position: Showing page 2 of {total_pages}." in response - ) # Assumes the content is longer than one screen + response = function_map["page_down"]() + assert ( + f"Viewport position: Showing page 2 of {total_pages}." in response + ) # Assumes the content is longer than one screen - response = function_map["page_up"]() - assert f"Viewport position: Showing page 1 of {total_pages}." in response + response = function_map["page_up"]() + assert f"Viewport position: Showing page 1 of {total_pages}." in response - # Try to scroll too far back up - response = function_map["page_up"]() - assert f"Viewport position: Showing page 1 of {total_pages}." in response + # Try to scroll too far back up + response = function_map["page_up"]() + assert f"Viewport position: Showing page 1 of {total_pages}." in response - # Try to scroll too far down - for i in range(0, total_pages + 1): - response = function_map["page_down"]() - assert f"Viewport position: Showing page {total_pages} of {total_pages}." in response + # Try to scroll too far down + for i in range(0, total_pages + 1): + response = function_map["page_down"]() + assert f"Viewport position: Showing page {total_pages} of {total_pages}." in response - # Test web search -- we don't have a key in this case, so we expect it to raise an error (but it means the code path is correct) - with pytest.raises(ValueError, match="Missing Bing API key."): - response = function_map["informational_web_search"](BING_QUERY) + # Test web search -- we don't have a key in this case, so we expect it to raise an error (but it means the code path is correct) + with pytest.raises(ValueError, match="Missing Bing API key."): + response = function_map["informational_web_search"](BING_QUERY) - with pytest.raises(ValueError, match="Missing Bing API key."): - response = function_map["navigational_web_search"](BING_QUERY) + with pytest.raises(ValueError, match="Missing Bing API key."): + response = function_map["navigational_web_search"](BING_QUERY) - # Test Q&A and summarization -- we don't have a key so we expect it to fail (but it means the code path is correct) - with pytest.raises(AttributeError, match="'NoneType' object has no attribute 'create'"): - response = function_map["answer_from_page"]("When was it founded?") + # Test Q&A and summarization -- we don't have a key so we expect it to fail (but it means the code path is correct) + with pytest.raises(IndexError): + response = function_map["answer_from_page"]("When was it founded?") - with pytest.raises(AttributeError, match="'NoneType' object has no attribute 'create'"): - response = function_map["summarize_page"]() + with pytest.raises(IndexError): + response = function_map["summarize_page"]() @pytest.mark.skipif( skip_oai, reason="do not run if oai is not installed", ) -def test_web_surfer_oai(): - llm_config = {"config_list": config_list, "timeout": 180, "cache_seed": None} +def test_web_surfer_oai() -> None: + llm_config = {"config_list": config_list, "timeout": 180, "cache_seed": 42} + + # adding Azure name variations to the model list + model = ["gpt-3.5-turbo-1106", "gpt-3.5-turbo-16k-0613", "gpt-3.5-turbo-16k"] + model += [m.replace(".", "") for m in model] summarizer_llm_config = { - "config_list": filter_config( - config_list, {"model": ["gpt-3.5-turbo-1106", "gpt-3.5-turbo-16k-0613", "gpt-3.5-turbo-16k"]} - ), + "config_list": filter_config(config_list, dict(model=model)), # type: ignore[no-untyped-call] "timeout": 180, - "cache_seed": None, } - assert len(llm_config["config_list"]) > 0 + assert len(llm_config["config_list"]) > 0 # type: ignore[arg-type] assert len(summarizer_llm_config["config_list"]) > 0 page_size = 4096 @@ -126,27 +133,35 @@ def test_web_surfer_oai(): is_termination_msg=lambda x: True, ) - # Make some requests that should test function calling - user_proxy.initiate_chat(web_surfer, message="Please visit the page 'https://en.wikipedia.org/wiki/Microsoft'") + with Cache.disk(): + # Make some requests that should test function calling + user_proxy.initiate_chat(web_surfer, message="Please visit the page 'https://en.wikipedia.org/wiki/Microsoft'") - user_proxy.initiate_chat(web_surfer, message="Please scroll down.") + user_proxy.initiate_chat(web_surfer, message="Please scroll down.") - user_proxy.initiate_chat(web_surfer, message="Please scroll up.") + user_proxy.initiate_chat(web_surfer, message="Please scroll up.") - user_proxy.initiate_chat(web_surfer, message="When was it founded?") + user_proxy.initiate_chat(web_surfer, message="When was it founded?") - user_proxy.initiate_chat(web_surfer, message="What's this page about?") + user_proxy.initiate_chat(web_surfer, message="What's this page about?") @pytest.mark.skipif( skip_bing, reason="do not run if bing api key is not available", ) -def test_web_surfer_bing(): +def test_web_surfer_bing() -> None: page_size = 4096 web_surfer = WebSurferAgent( "web_surfer", - llm_config=False, + llm_config={ + "config_list": [ + { + "model": "gpt-3.5-turbo-16k", + "api_key": "sk-PLACEHOLDER_KEY", + } + ] + }, browser_config={"viewport_size": page_size, "bing_api_key": BING_API_KEY}, ) @@ -168,5 +183,5 @@ def test_web_surfer_bing(): if __name__ == "__main__": """Runs this file's tests from the command line.""" test_web_surfer() - # test_web_surfer_oai() - # test_web_surfer_bing() + test_web_surfer_oai() + test_web_surfer_bing()