Coverage for fastagency/runtimes/ag2/tools/websurfer.py: 81%
107 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-19 12:16 +0000
1from typing import TYPE_CHECKING, Annotated, Any, Optional, Union 1cghbaidef
3from autogen import LLMConfig 1cghbaidef
4from autogen.agentchat import AssistantAgent as AutoGenAssistantAgent 1cghbaidef
5from autogen.agentchat import ConversableAgent as AutoGenConversableAgent 1cghbaidef
6from autogen.agentchat.contrib.web_surfer import WebSurferAgent as AutoGenWebSurferAgent 1cghbaidef
7from pydantic import BaseModel, Field, HttpUrl 1cghbaidef
9if TYPE_CHECKING: 1cghbaidef
10 from autogen.io.run_response import RunResponse
12__all__ = ["WebSurferAnswer", "WebSurferTool"] 1cghbaidef
15class WebSurferAnswer(BaseModel): 1cghbaidef
16 task: Annotated[str, Field(..., description="The task to be completed")] 1cghbaidef
17 is_successful: Annotated[ 1cghbaidef
18 bool, Field(..., description="Whether the task was successful")
19 ]
20 short_answer: Annotated[ 1cghbaidef
21 str,
22 Field(
23 ...,
24 description="The short answer to the task without any explanation",
25 ),
26 ]
27 long_answer: Annotated[ 1cghbaidef
28 str,
29 Field(..., description="The long answer to the task with explanation"),
30 ]
31 visited_links: Annotated[ 1cghbaidef
32 list[HttpUrl],
33 Field(..., description="The list of visited links to generate the answer"),
34 ]
36 @staticmethod 1cghbaidef
37 def get_example_answer() -> "WebSurferAnswer": 1cghbaidef
38 return WebSurferAnswer( 1cbadef
39 task="What is the most popular QLED TV to buy on amazon.com?",
40 is_successful=True,
41 short_answer='Amazon Fire TV 55" Omni QLED Series 4K UHD smart TV',
42 long_answer='Amazon has the best selling page by different categories and there is a category for QLED TVs under electroincs. The most popular QLED TV is Amazon Fire TV 55" Omni QLED Series 4K UHD smart TV, Dolby Vision IQ, Fire TV Ambient Experience, local dimming, hands-free with Alexa. It is the best selling QLED TV on Amazon.',
43 visited_links=[
44 "https://www.amazon.com/Best-Sellers/",
45 "https://www.amazon.com/Best-Sellers-Electronics-QLED-TVs/",
46 ],
47 )
50class WebSurferTool: # implements Toolable 1cghbaidef
51 def __init__( 1cghbaidef
52 self,
53 *,
54 name_prefix: str,
55 llm_config: LLMConfig,
56 summarizer_llm_config: LLMConfig,
57 viewport_size: int = 4096,
58 bing_api_key: Optional[str] = None,
59 max_consecutive_auto_reply: int = 30,
60 max_links_to_click: int = 10,
61 websurfer_kwargs: Optional[dict[str, Any]] = None,
62 assistant_kwargs: Optional[dict[str, Any]] = None,
63 ):
64 """Create a new WebSurferChat instance.
66 Args:
67 name_prefix (str): The name prefix of the inner AutoGen agents
68 llm_config (Dict[str, Any]): The LLM configuration
69 summarizer_llm_config (Dict[str, Any]): The summarizer LLM configuration
70 viewport_size (int, optional): The viewport size. Defaults to 4096.
71 bing_api_key (Optional[str], optional): The Bing API key. Defaults to None.
72 max_consecutive_auto_reply (int, optional): The maximum consecutive auto reply. Defaults to 30.
73 max_links_to_click (int, optional): The maximum links to click. Defaults to 10.
74 websurfer_kwargs (Optional[Dict[str, Any]], optional): The WebSurfer kwargs. Defaults to None.
75 assistant_kwargs (Optional[Dict[str, Any]], optional): The Assistant kwargs. Defaults to None.
76 """
77 if websurfer_kwargs is None: 77 ↛ 79line 77 didn't jump to line 79 because the condition on line 77 was always true1cbadef
78 websurfer_kwargs = {} 1cbadef
79 if assistant_kwargs is None: 79 ↛ 82line 79 didn't jump to line 82 because the condition on line 79 was always true1cbadef
80 assistant_kwargs = {} 1cbadef
82 self.name_prefix = name_prefix 1cbadef
83 self.llm_config = llm_config 1cbadef
84 self.summarizer_llm_config = summarizer_llm_config 1cbadef
85 self.viewport_size = viewport_size 1cbadef
86 self.bing_api_key = bing_api_key 1cbadef
87 self.max_consecutive_auto_reply = max_consecutive_auto_reply 1cbadef
88 self.max_links_to_click = max_links_to_click 1cbadef
89 self.websurfer_kwargs = websurfer_kwargs 1cbadef
90 self.assistant_kwargs = assistant_kwargs 1cbadef
92 self.task = "not set yet" 1cbadef
93 self.last_is_termination_msg_error = "" 1cbadef
95 self.browser_config = { 1cbadef
96 "viewport_size": self.viewport_size,
97 "bing_api_key": self.bing_api_key,
98 "request_kwargs": {
99 "headers": {
100 "User-Agent": "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2228.0 Safari/537.36",
101 }
102 },
103 }
105 if "human_input_mode" in self.websurfer_kwargs: 105 ↛ 106line 105 didn't jump to line 106 because the condition on line 105 was never true1cbadef
106 self.websurfer_kwargs.pop("human_input_mode")
108 self.websurfer = AutoGenWebSurferAgent( 1cbadef
109 name=f"{self.name_prefix}_inner_websurfer",
110 llm_config=self.llm_config,
111 summarizer_llm_config=self.summarizer_llm_config,
112 browser_config=self.browser_config,
113 human_input_mode="NEVER",
114 is_termination_msg=self.is_termination_msg,
115 **self.websurfer_kwargs,
116 )
118 if "human_input_mode" in self.assistant_kwargs: 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true1cbadef
119 self.assistant_kwargs.pop("human_input_mode")
121 self.assistant = AutoGenAssistantAgent( 1cbadef
122 name=f"{self.name_prefix}_inner_assistant",
123 llm_config=self.llm_config,
124 human_input_mode="NEVER",
125 system_message=self.system_message,
126 max_consecutive_auto_reply=self.max_consecutive_auto_reply,
127 # is_termination_msg=self.is_termination_msg,
128 **self.assistant_kwargs,
129 )
131 def is_termination_msg(self, msg: dict[str, Any]) -> bool: 1cghbaidef
132 # print(f"is_termination_msg({msg=})")
133 if ( 133 ↛ 138line 133 didn't jump to line 138 because the condition on line 133 was never true1a
134 "content" in msg
135 and msg["content"] is not None
136 and "TERMINATE" in msg["content"]
137 ):
138 return True
139 try: 1a
140 WebSurferAnswer.model_validate_json(msg["content"]) 1a
141 return True 1a
142 except Exception as e: 1a
143 self.last_is_termination_msg_error = str(e) 1a
144 return False 1a
146 def _get_error_message(self, response: "RunResponse") -> Optional[str]: 1cghbaidef
147 messages = [ 1a
148 str(m["content"]) if "content" in m else "" for m in response.messages
149 ]
150 last_message = messages[-1] 1a
151 if "TERMINATE" in last_message: 151 ↛ 152line 151 didn't jump to line 152 because the condition on line 151 was never true1a
152 return self.error_message
154 try: 1a
155 WebSurferAnswer.model_validate_json(last_message) 1a
156 except Exception:
157 return self.error_message
159 return None 1a
161 def _get_answer(self, response: "RunResponse") -> WebSurferAnswer: 1cghbaidef
162 messages = [ 1a
163 str(m["content"]) if "content" in m else "" for m in response.messages
164 ]
165 last_message = messages[-1] 1a
166 return WebSurferAnswer.model_validate_json(last_message) 1a
168 def _chat_with_websurfer( 1cghbaidef
169 self, message: str, clear_history: bool, **kwargs: Any
170 ) -> WebSurferAnswer:
171 msg: Optional[str] = message 1a
173 while msg is not None: 1a
174 response = self.websurfer.run( 1a
175 self.assistant,
176 clear_history=clear_history,
177 message=msg,
178 )
179 response.process() 1a
180 msg = self._get_error_message(response) 1a
181 clear_history = False 1a
183 return self._get_answer(response) 1a
185 def _get_error_from_exception(self, task: str, e: Exception) -> str: 1cghbaidef
186 answer = WebSurferAnswer(
187 task=task,
188 is_successful=False,
189 short_answer="unexpected error occurred",
190 long_answer=str(e),
191 visited_links=[],
192 )
194 return self.create_final_reply(task, answer)
196 def create_final_reply(self, task: str, message: WebSurferAnswer) -> str: 1cghbaidef
197 retval = ( 1a
198 "We have successfully completed the task:\n\n"
199 if message.is_successful
200 else "We have failed to complete the task:\n\n"
201 )
202 retval += f"{task}\n\n" 1a
203 retval += f"Short answer: {message.short_answer}\n\n" 1a
204 retval += f"Explanation: {message.long_answer}\n\n" 1a
205 retval += "Visited links:\n" 1a
206 for link in message.visited_links: 206 ↛ 207line 206 didn't jump to line 207 because the loop on line 206 never started1a
207 retval += f" - {link}\n"
209 return retval 1a
211 def create_new_task( 1cghbaidef
212 self, task: Annotated[str, "a new task for websurfer to perform"]
213 ) -> str:
214 self.task = task 1a
215 try: 1a
216 answer = self._chat_with_websurfer( 1a
217 message=self.initial_message,
218 clear_history=True,
219 )
220 except Exception as e:
221 return self._get_error_from_exception(task, e)
223 return self.create_final_reply(task, answer) 1a
225 def continue_task_with_additional_instructions( 1cghbaidef
226 self, message: Annotated[str, "a followup message to the existing task"]
227 ) -> str:
228 try:
229 answer = self._chat_with_websurfer(
230 message=message,
231 clear_history=False,
232 )
233 except Exception as e:
234 return self._get_error_from_exception(message, e)
236 return self.create_final_reply(message, answer)
238 @property 1cghbaidef
239 def example_answer(self) -> WebSurferAnswer: 1cghbaidef
240 return WebSurferAnswer.get_example_answer() 1cbadef
242 @property 1cghbaidef
243 def initial_message(self) -> str: 1cghbaidef
244 return f"""We are tasked with the following task: 1a
246{self.task}
248If no link is provided in the task, you should search the internet first to find the relevant information.
250The focus is on the provided url and its subpages, we do NOT care about the rest of the website i.e. parent pages.
251e.g. If the url is 'https://www.example.com/products/air-conditioners', we are interested ONLY in the 'air-conditioners' and its subpages.
253AFTER visiting the home page, create a step-by-step plan BEFORE visiting the other pages.
254You can click on MAXIMUM {self.max_links_to_click} links. Do NOT try to click all the links on the page, but only the ones which are most relevant for the task (MAX {self.max_links_to_click})!
255Do NOT visit the same page multiple times, but only once!
256If your co-speaker repeats the same message, inform him that you have already answered to that message and ask him to proceed with the task.
257e.g. "I have already answered to that message, please proceed with the task or you will be penalized!"
258"""
260 @property 1cghbaidef
261 def error_message(self) -> str: 1cghbaidef
262 return f"""Please output the JSON-encoded answer only in the following message before trying to terminate the chat.
264IMPORTANT:
265 - NEVER enclose JSON-encoded answer in any other text or formatting including '```json' ... '```' or similar!
266 - NEVER write TERMINATE in the same message as the JSON-encoded answer!
268EXAMPLE:
270{self.example_answer.model_dump_json()}
272NEGATIVE EXAMPLES:
2741. Do NOT include 'TERMINATE' in the same message as the JSON-encoded answer!
276{self.example_answer.model_dump_json()}
278TERMINATE
2802. Do NOT include triple backticks or similar!
282```json
283{self.example_answer.model_dump_json()}
284```
286THE LAST ERROR MESSAGE:
288{self.last_is_termination_msg_error}
290"""
292 @property 1cghbaidef
293 def system_message(self) -> str: 1cghbaidef
294 return f"""You are in charge of navigating the web_surfer agent to scrape the web. 1cbadef
295web_surfer is able to CLICK on links, SCROLL down, and scrape the content of the web page. e.g. you cen tell him: "Click the 'Getting Started' result".
296Each time you receive a reply from web_surfer, you need to tell him what to do next. e.g. "Click the TV link" or "Scroll down".
297It is very important that you explore ONLY the page links relevant for the task!
299GUIDELINES:
300- Once you retrieve the content from the received url, you can tell web_surfer to CLICK on links, SCROLL down...
301By using these capabilities, you will be able to retrieve MUCH BETTER information from the web page than by just scraping the given URL!
302You MUST use these capabilities when you receive a task for a specific category/product etc.
303- do NOT try to create a summary without clicking on any link, because you will be missing a lot of information!
304- if needed, you can instruct web surfer to SEARCH THE WEB for information.
306Examples:
307"Click the 'TVs' result" - This way you will navigate to the TVs section of the page and you will find more information about TVs.
308"Click 'Electronics' link" - This way you will navigate to the Electronics section of the page and you will find more information about Electronics.
309"Click the 'Next' button"
310"Search the internet for the best TV to buy" - this will get links to initial pages to start the search
312- Do NOT try to click all the links on the page, but only the ones which are RELEVANT for the task! Web pages can be very long and you will be penalized if spend too much time on this task!
313- Your final goal is to summarize the findings for the given task. The summary must be in English!
314- Create a summary after you successfully retrieve the information from the web page.
315- It is useful to include in the summary relevant links where more information can be found.
316e.g. If the page is offering to sell TVs, you can include a link to the TV section of the page.
317- If you get some 40x error, please do NOT give up immediately, but try to navigate to another page and continue with the task.
318Give up only if you get 40x error on ALL the pages which you tried to navigate to.
321FINAL MESSAGE:
322Once you have retrieved he wanted information, YOU MUST create JSON-encoded string. Summary created by the web_surfer is not enough!
323You MUST not include any other text or formatting in the message, only JSON-encoded summary!
325An example of the JSON-encoded summary:
326{self.example_answer.model_dump_json()}
328TERMINATION:
329When YOU are finished and YOU have created JSON-encoded answer, write a single 'TERMINATE' to end the task.
331OFTEN MISTAKES:
332- Web surfer expects you to tell him what LINK NAME to click next, not the relative link. E.g. in case of '[Hardware](/Hardware), the proper command would be 'Click into 'Hardware''.
333- Links presented are often RELATIVE links, so you need to ADD the DOMAIN to the link to make it work. E.g. link '/products/air-conditioners' should be 'https://www.example.com/products/air-conditioners'
334- You do NOT need to click on MAX number of links. If you have enough information from the first xy links, you do NOT need to click on the rest of the links!
335- Do NOT repeat the steps you have already completed!
336- ALWAYS include the NEXT steps in the message!
337- Do NOT instruct web_surfer to click on the same link multiple times. If there are some problems with the link, MOVE ON to the next one!
338- Also, if web_surfer does not understand your message, just MOVE ON to the next link!
339- NEVER REPEAT the same instructions to web_surfer! If he does not understand the first time, MOVE ON to the next link!
340- NEVER enclose JSON-encoded answer in any other text or formatting including '```json' ... '```' or similar!
341"""
343 def register( 1cghbaidef
344 self,
345 *,
346 caller: AutoGenConversableAgent,
347 executor: Union[AutoGenConversableAgent, list[AutoGenConversableAgent]],
348 ) -> None:
349 @caller.register_for_llm( # type: ignore[misc] 1ba
350 name="create_new_websurfing_task",
351 description="Creates a new task for a websurfer that can include searching or browsing the internet.",
352 )
353 def create_new_task( 1ba
354 task: Annotated[str, "a new task for websurfer to perform"],
355 ) -> str:
356 return self.create_new_task(task) 1a
358 @caller.register_for_llm( # type: ignore[misc] 1ba
359 name="continue_websurfing_task_with_additional_instructions",
360 description="Continue an existing task for a websurfer with additional instructions.",
361 )
362 def continue_task_with_additional_instructions( 1ba
363 message: Annotated[
364 str,
365 "Additional instructions for the task after receiving the initial answer",
366 ],
367 ) -> str:
368 return self.continue_task_with_additional_instructions(message)
370 executors = executor if isinstance(executor, list) else [executor] 1ba
371 for executor in executors: 1ba
372 executor.register_for_execution(name="create_new_websurfing_task")( 1ba
373 create_new_task
374 )
375 executor.register_for_execution( 1ba
376 name="continue_websurfing_task_with_additional_instructions"
377 )(continue_task_with_additional_instructions)