Agenten Prozessor

def add_to_memory(self, data_dict: Dict[str, Any]) -> None:
    """
    Add the data to the memory.
    :param data_dict: The data dictionary to be added to the memory.
    """
    self._memory_data.add_values_from_dict(data_dict)

@abstractmethod
def capture_screenshot(self) -> None:
    """
    Capture the screenshot.
    """
    pass

@classmethod
def exception_capture(cls, func):
    """
    Decorator to capture the exception of the method.
    :param func: The method to be decorated.
    :return: The decorated method.
    """

    @wraps(func)
    def wrapper(self, *args, **kwargs):
        try:
            func(self, *args, **kwargs)
        except Exception as e:
            self._exeception_traceback[func.__name__] = {
                "type": str(type(e).__name__),
                "message": str(e),
                "traceback": traceback.format_exc(),
            }

            utils.print_with_color(f"Error Occurs at {func.__name__}", "red")
            utils.print_with_color(
                self._exeception_traceback[func.__name__]["traceback"], "red"
            )
            if self._response is not None:
                utils.print_with_color("Response: ", "red")
                utils.print_with_color(self._response, "red")
            self._status = self._agent_status_manager.ERROR.value
            self.sync_memory()
            self.add_to_memory({"error": self._exeception_traceback})
            self.add_to_memory({"Status": self._status})
            self.log_save()

            raise StopIteration("Error occurred during step.")

    return wrapper

@abstractmethod
def execute_action(self) -> None:
    """
    Execute the action.
    """
    pass

@abstractmethod
def get_control_info(self) -> None:
    """
    Get the control information.
    """
    pass

@abstractmethod
def get_prompt_message(self) -> None:
    """
    Get the prompt message.
    """
    pass

@abstractmethod
def get_response(self) -> None:
    """
    Get the response from the LLM.
    """
    pass

def is_application_closed(self) -> bool:
    """
    Check if the application is closed.
    :return: The boolean value indicating if the application is closed.
    """

    if self.application_window is None:

        return True

    try:
        self.application_window.is_enabled()
        return False
    except:
        return True

def is_confirm(self) -> bool:
    """
    Check if the process is confirm.
    :return: The boolean value indicating if the process is confirm.
    """

    self.agent.status = self.status

    return self.status == self._agent_status_manager.CONFIRM.value

def is_error(self) -> bool:
    """
    Check if the process is in error.
    :return: The boolean value indicating if the process is in error.
    """

    self.agent.status = self.status
    return self.status == self._agent_status_manager.ERROR.value

def is_paused(self) -> bool:
    """
    Check if the process is paused.
    :return: The boolean value indicating if the process is paused.
    """

    self.agent.status = self.status

    return (
        self.status == self._agent_status_manager.PENDING.value
        or self.status == self._agent_status_manager.CONFIRM.value
    )

def is_pending(self) -> bool:
    """
    Check if the process is pending.
    :return: The boolean value indicating if the process is pending.
    """

    self.agent.status = self.status

    return self.status == self._agent_status_manager.PENDING.value

def log(self, response_json: Dict[str, Any]) -> None:
    """
    Set the result of the session, and log the result.
    result: The result of the session.
    response_json: The response json.
    return: The response json.
    """

    self.logger.info(json.dumps(response_json))

def log_save(self) -> None:
    """
    Save the log.
    """

    self._memory_data.add_values_from_dict(
        {"total_time_cost": self._total_time_cost}
    )
    self.log(self._memory_data.to_dict())

@classmethod
def method_timer(cls, func):
    """
    Decorator to calculate the time cost of the method.
    :param func: The method to be decorated.
    :return: The decorated method.
    """

    @wraps(func)
    def wrapper(self, *args, **kwargs):
        start_time = time.time()
        result = func(self, *args, **kwargs)
        end_time = time.time()
        self._time_cost[func.__name__] = end_time - start_time
        return result

    return wrapper

@abstractmethod
def parse_response(self) -> None:
    """
    Parse the response.
    """
    pass

@abstractmethod
def print_step_info(self) -> None:
    """
    Print the step information.
    """
    pass

def process(self) -> None:
    """
    Process a single step in a round.
    The process includes the following steps:
    1. Print the step information.
    2. Capture the screenshot.
    3. Get the control information.
    4. Get the prompt message.
    5. Get the response.
    6. Update the cost.
    7. Parse the response.
    8. Execute the action.
    9. Update the memory.
    10. Update the step and status.
    11. Save the log.
    """

    start_time = time.time()

    try:
        # Step 1: Print the step information.
        self.print_step_info()

        # Step 2: Capture the screenshot.
        self.capture_screenshot()

        # Step 3: Get the control information.
        self.get_control_info()

        # Step 4: Get the prompt message.
        self.get_prompt_message()

        # Step 5: Get the response.
        self.get_response()

        # Step 6: Update the context.
        self.update_cost()

        # Step 7: Parse the response, if there is no error.
        self.parse_response()

        if self.is_pending() or self.is_paused():
            # If the session is pending, update the step and memory, and return.
            if self.is_pending():
                self.update_status()
                self.update_memory()

            return

        # Step 8: Execute the action.
        self.execute_action()

        # Step 9: Update the memory.
        self.update_memory()

        # Step 10: Update the status.
        self.update_status()

        self._total_time_cost = time.time() - start_time

        # Step 11: Save the log.
        self.log_save()

    except StopIteration:
        # Error was handled and logged in the exception capture decorator.
        # Simply return here to stop the process early.

        return

def resume(self) -> None:
    """
    Resume the process of action execution after the session is paused.
    """

    self._is_resumed = True

    try:
        # Step 1: Execute the action.
        self.execute_action()

        # Step 2: Update the memory.
        self.update_memory()

        # Step 3: Update the status.
        self.update_status()

    except StopIteration:
        # Error was handled and logged in the exception capture decorator.
        # Simply return here to stop the process early.
        pass

    finally:
        self._is_resumed = False

@staticmethod
def string2list(string: Any) -> List[str]:
    """
    Convert a string to a list of string if the input is a string.
    :param string: The string.
    :return: The list.
    """
    if isinstance(string, str):
        return [string]
    else:
        return string

@abstractmethod
def sync_memory(self) -> None:
    """
    Sync the memory of the Agent.
    """
    pass

def update_cost(self) -> None:
    """
    Update the cost.
    """

    self.round_cost += self.cost
    self.session_cost += self.cost

@abstractmethod
def update_memory(self) -> None:
    """
    Update the memory of the Agent.
    """
    pass

def update_status(self) -> None:
    """
    Update the status of the session.
    """
    self.agent.step += 1
    self.agent.status = self.status

    if self.status != self._agent_status_manager.FINISH.value:
        time.sleep(configs["SLEEP_TIME"])

    self.round_step += 1
    self.session_step += 1