Spaces:
Sleeping
Sleeping
File size: 2,694 Bytes
ea7486e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 |
# coding: utf-8
# Copyright (c) 2025 inclusionAI.
from typing import Tuple, List, Any
from aworld.core.tool.action_factory import ActionFactory
from aworld.core.common import ActionModel, ActionResult, Observation
from aworld.logs.util import logger
from aworld.core.tool.base import Tool, ToolActionExecutor
class BrowserToolActionExecutor(ToolActionExecutor):
def __init__(self, tool: Tool = None):
super(BrowserToolActionExecutor, self).__init__(tool)
def execute_action(self, actions: List[ActionModel], **kwargs) -> Tuple[
List[ActionResult], Any]:
"""Execute the specified browser action sequence by agent policy.
Args:
actions: Tool action sequence.
Returns:
Browser page and action result list.
"""
action_results = []
page = self.tool.page
for action in actions:
action_result, page = self._exec(action, **kwargs)
action_results.append(action_result)
return action_results, page
async def async_execute_action(self, actions: List[ActionModel], **kwargs) -> Tuple[
List[ActionResult], Any]:
"""Execute the specified browser action sequence by agent policy.
Args:
actions: Tool action sequence.
Returns:
Browser page and action result list.
"""
action_results = []
page = self.tool.page
for action in actions:
action_result, page = await self._async_exec(action, **kwargs)
action_results.append(action_result)
return action_results, page
def _exec(self, action_model: ActionModel, **kwargs):
action_name = action_model.action_name
if action_name not in ActionFactory:
raise ValueError(f'Action {action_name} not found')
action = ActionFactory(action_name)
action_result, page = action.act(action_model, page=self.tool.page, browser=self.tool.context, **kwargs)
logger.info(f"{action_name} execute finished")
return action_result, page
async def _async_exec(self, action_model: ActionModel, **kwargs):
action_name = action_model.action_name
if action_name not in ActionFactory:
action_name = action_model.tool_name + action_model.action_name
if action_name not in ActionFactory:
raise ValueError(f'Action {action_name} not found')
action = ActionFactory(action_name)
action_result, page = await action.async_act(action_model, page=self.tool.page, browser=self.tool.context, **kwargs)
logger.info(f"{action_name} execute finished")
return action_result, page
|