Spaces:
Sleeping
Sleeping
# coding: utf-8 | |
# Copyright (c) 2025 inclusionAI. | |
from typing import List | |
from aworld.core.tool.action_factory import ActionFactory | |
from aworld.core.common import ActionModel, ActionResult | |
from aworld.logs.util import logger | |
from examples.tools.android.action.adb_controller import ADBController | |
from aworld.core.tool.base import ToolActionExecutor | |
class AndroidToolActionExecutor(ToolActionExecutor): | |
def __init__(self, controller: ADBController): | |
self.controller = controller | |
def execute_action(self, actions: List[ActionModel], **kwargs) -> list[ActionResult]: | |
"""Execute the specified android action sequence by agent policy. | |
Args: | |
actions: Tool action sequence. | |
Returns: | |
Browser action result list. | |
""" | |
action_results = [] | |
for action in actions: | |
action_result = self._exec(action, **kwargs) | |
action_results.append(action_result) | |
return action_results | |
def _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 = action.act(action_model, controller=self.controller, **kwargs) | |
logger.info(f"{action_name} execute finished") | |
return action_result | |