Skip to content

WebSurferTool

fastagency.runtimes.autogen.tools.WebSurferTool #

WebSurferTool(
    *,
    name_prefix: str,
    llm_config: dict[str, Any],
    summarizer_llm_config: dict[str, Any],
    viewport_size: int = 4096,
    bing_api_key: Optional[str] = None,
    max_consecutive_auto_reply: int = 30,
    max_links_to_click: int = 10,
    websurfer_kwargs: Optional[dict[str, Any]] = None,
    assistant_kwargs: Optional[dict[str, Any]] = None
)

Create a new WebSurferChat instance.

PARAMETER DESCRIPTION
name_prefix

The name prefix of the inner AutoGen agents

TYPE: str

llm_config

The LLM configuration

TYPE: Dict[str, Any]

summarizer_llm_config

The summarizer LLM configuration

TYPE: Dict[str, Any]

viewport_size

The viewport size. Defaults to 4096.

TYPE: int DEFAULT: 4096

bing_api_key

The Bing API key. Defaults to None.

TYPE: Optional[str] DEFAULT: None

max_consecutive_auto_reply

The maximum consecutive auto reply. Defaults to 30.

TYPE: int DEFAULT: 30

max_links_to_click

The maximum links to click. Defaults to 10.

TYPE: int DEFAULT: 10

websurfer_kwargs

The WebSurfer kwargs. Defaults to None.

TYPE: Optional[Dict[str, Any]] DEFAULT: None

assistant_kwargs

The Assistant kwargs. Defaults to None.

TYPE: Optional[Dict[str, Any]] DEFAULT: None

Source code in fastagency/runtimes/autogen/tools/websurfer.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def __init__(
    self,
    *,
    name_prefix: str,
    llm_config: dict[str, Any],
    summarizer_llm_config: dict[str, Any],
    viewport_size: int = 4096,
    bing_api_key: Optional[str] = None,
    max_consecutive_auto_reply: int = 30,
    max_links_to_click: int = 10,
    websurfer_kwargs: Optional[dict[str, Any]] = None,
    assistant_kwargs: Optional[dict[str, Any]] = None,
):
    """Create a new WebSurferChat instance.

    Args:
        name_prefix (str): The name prefix of the inner AutoGen agents
        llm_config (Dict[str, Any]): The LLM configuration
        summarizer_llm_config (Dict[str, Any]): The summarizer LLM configuration
        viewport_size (int, optional): The viewport size. Defaults to 4096.
        bing_api_key (Optional[str], optional): The Bing API key. Defaults to None.
        max_consecutive_auto_reply (int, optional): The maximum consecutive auto reply. Defaults to 30.
        max_links_to_click (int, optional): The maximum links to click. Defaults to 10.
        websurfer_kwargs (Optional[Dict[str, Any]], optional): The WebSurfer kwargs. Defaults to None.
        assistant_kwargs (Optional[Dict[str, Any]], optional): The Assistant kwargs. Defaults to None.
    """
    if websurfer_kwargs is None:
        websurfer_kwargs = {}
    if assistant_kwargs is None:
        assistant_kwargs = {}

    self.name_prefix = name_prefix
    self.llm_config = llm_config
    self.summarizer_llm_config = summarizer_llm_config
    self.viewport_size = viewport_size
    self.bing_api_key = bing_api_key
    self.max_consecutive_auto_reply = max_consecutive_auto_reply
    self.max_links_to_click = max_links_to_click
    self.websurfer_kwargs = websurfer_kwargs
    self.assistant_kwargs = assistant_kwargs

    self.task = "not set yet"
    self.last_is_termination_msg_error = ""

    self.browser_config = {
        "viewport_size": self.viewport_size,
        "bing_api_key": self.bing_api_key,
        "request_kwargs": {
            "headers": {
                "User-Agent": "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2228.0 Safari/537.36",
            }
        },
    }

    if "human_input_mode" in self.websurfer_kwargs:
        self.websurfer_kwargs.pop("human_input_mode")

    self.websurfer = AutoGenWebSurferAgent(
        name=f"{self.name_prefix}_inner_websurfer",
        llm_config=self.llm_config,
        summarizer_llm_config=self.summarizer_llm_config,
        browser_config=self.browser_config,
        human_input_mode="NEVER",
        is_termination_msg=self.is_termination_msg,
        **self.websurfer_kwargs,
    )

    if "human_input_mode" in self.assistant_kwargs:
        self.assistant_kwargs.pop("human_input_mode")

    self.assistant = AutoGenAssistantAgent(
        name=f"{self.name_prefix}_inner_assistant",
        llm_config=self.llm_config,
        human_input_mode="NEVER",
        system_message=self.system_message,
        max_consecutive_auto_reply=self.max_consecutive_auto_reply,
        # is_termination_msg=self.is_termination_msg,
        **self.assistant_kwargs,
    )

assistant instance-attribute #

assistant = AssistantAgent(
    name=f"{name_prefix}_inner_assistant",
    llm_config=llm_config,
    human_input_mode="NEVER",
    system_message=system_message,
    max_consecutive_auto_reply=max_consecutive_auto_reply,
    **assistant_kwargs,
)

assistant_kwargs instance-attribute #

assistant_kwargs = assistant_kwargs

bing_api_key instance-attribute #

bing_api_key = bing_api_key

browser_config instance-attribute #

browser_config = {
    "viewport_size": viewport_size,
    "bing_api_key": bing_api_key,
    "request_kwargs": {
        "headers": {
            "User-Agent": "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2228.0 Safari/537.36"
        }
    },
}

error_message property #

error_message: str

example_answer property #

example_answer: WebSurferAnswer

initial_message property #

initial_message: str

last_is_termination_msg_error instance-attribute #

last_is_termination_msg_error = ''

llm_config instance-attribute #

llm_config = llm_config

max_consecutive_auto_reply instance-attribute #

max_consecutive_auto_reply = max_consecutive_auto_reply

max_links_to_click = max_links_to_click

name_prefix instance-attribute #

name_prefix = name_prefix

summarizer_llm_config instance-attribute #

summarizer_llm_config = summarizer_llm_config

system_message property #

system_message: str

task instance-attribute #

task = 'not set yet'

viewport_size instance-attribute #

viewport_size = viewport_size

websurfer instance-attribute #

websurfer = WebSurferAgent(
    name=f"{name_prefix}_inner_websurfer",
    llm_config=llm_config,
    summarizer_llm_config=summarizer_llm_config,
    browser_config=browser_config,
    human_input_mode="NEVER",
    is_termination_msg=is_termination_msg,
    **websurfer_kwargs,
)

websurfer_kwargs instance-attribute #

websurfer_kwargs = websurfer_kwargs

continue_task_with_additional_instructions #

continue_task_with_additional_instructions(
    message: str,
) -> str
Source code in fastagency/runtimes/autogen/tools/websurfer.py 
217
218
219
220
221
222
223
224
225
226
227
228
def continue_task_with_additional_instructions(
    self, message: Annotated[str, "a followup message to the existing task"]
) -> str:
    try:
        answer = self._chat_with_websurfer(
            message=message,
            clear_history=False,
        )
    except Exception as e:
        return self._get_error_from_exception(message, e)

    return self.create_final_reply(message, answer)

create_final_reply #

create_final_reply(
    task: str, message: WebSurferAnswer
) -> str
Source code in fastagency/runtimes/autogen/tools/websurfer.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def create_final_reply(self, task: str, message: WebSurferAnswer) -> str:
    retval = (
        "We have successfully completed the task:\n\n"
        if message.is_successful
        else "We have failed to complete the task:\n\n"
    )
    retval += f"{task}\n\n"
    retval += f"Short answer: {message.short_answer}\n\n"
    retval += f"Explanation: {message.long_answer}\n\n"
    retval += "Visited links:\n"
    for link in message.visited_links:
        retval += f"  - {link}\n"

    return retval

create_new_task #

create_new_task(task: str) -> str
Source code in fastagency/runtimes/autogen/tools/websurfer.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
def create_new_task(
    self, task: Annotated[str, "a new task for websurfer to perform"]
) -> str:
    self.task = task
    try:
        answer = self._chat_with_websurfer(
            message=self.initial_message,
            clear_history=True,
        )
    except Exception as e:
        return self._get_error_from_exception(task, e)

    return self.create_final_reply(task, answer)

is_termination_msg #

is_termination_msg(msg: dict[str, Any]) -> bool
Source code in fastagency/runtimes/autogen/tools/websurfer.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def is_termination_msg(self, msg: dict[str, Any]) -> bool:
    # print(f"is_termination_msg({msg=})")
    if (
        "content" in msg
        and msg["content"] is not None
        and "TERMINATE" in msg["content"]
    ):
        return True
    try:
        WebSurferAnswer.model_validate_json(msg["content"])
        return True
    except Exception as e:
        self.last_is_termination_msg_error = str(e)
        return False

register #

register(
    *,
    caller: ConversableAgent,
    executor: Union[
        ConversableAgent, list[ConversableAgent]
    ]
) -> None
Source code in fastagency/runtimes/autogen/tools/websurfer.py 
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
def register(
    self,
    *,
    caller: AutoGenConversableAgent,
    executor: Union[AutoGenConversableAgent, list[AutoGenConversableAgent]],
) -> None:
    @caller.register_for_llm(  # type: ignore[misc]
        name="create_new_websurfing_task",
        description="Creates a new task for a websurfer that can include searching or browsing the internet.",
    )
    def create_new_task(
        task: Annotated[str, "a new task for websurfer to perform"],
    ) -> str:
        return self.create_new_task(task)

    @caller.register_for_llm(  # type: ignore[misc]
        name="continue_websurfing_task_with_additional_instructions",
        description="Continue an existing task for a websurfer with additional instructions.",
    )
    def continue_task_with_additional_instructions(
        message: Annotated[
            str,
            "Additional instructions for the task after receiving the initial answer",
        ],
    ) -> str:
        return self.continue_task_with_additional_instructions(message)

    executors = executor if isinstance(executor, list) else [executor]
    for executor in executors:
        executor.register_for_execution(name="create_new_websurfing_task")(
            create_new_task
        )
        executor.register_for_execution(
            name="continue_websurfing_task_with_additional_instructions"
        )(continue_task_with_additional_instructions)