Agents Processor

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

def error_log(self, response_str: str, error: str) -> None:
    """
    Error handler for the session.
    """
    log = json.dumps(
        {
            "step": self.session_step,
            "status": self._agent_status_manager.ERROR.value,
            "response": response_str,
            "error": error,
        }
    )
    self.logger.info(log)

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

def general_error_handler(self) -> None:
    """
    Error handler for the general error.
    """
    error_trace = traceback.format_exc()
    utils.print_with_color(f"Error Occurs at {self.name}", "red")
    utils.print_with_color(str(error_trace), "red")
    utils.print_with_color(self._response, "red")
    self.error_log(self._response, str(error_trace))
    self._status = self._agent_status_manager.ERROR.value

@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_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 llm_error_handler(self) -> None:
    """
    Error handler for the LLM error.
    """
    error_trace = traceback.format_exc()
    log = json.dumps(
        {
            "step": self.session_step,
            "prompt": self._prompt_message,
            "status": str(error_trace),
        }
    )
    utils.print_with_color(
        "Error occurs when calling LLM: {e}".format(e=str(error_trace)), "red"
    )
    self.request_logger.info(log)
    self._status = self._agent_status_manager.ERROR.value
    return

def log(self, response_json: dict) -> 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.set_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()

    # 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()

    if self.is_error():
        return

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

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

    if self.is_error() 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()

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

    self._is_resumed = True

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

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

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

    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

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