Round

A Round is a single interaction between the user and UFO that processes a single user request. A Round is responsible for orchestrating the HostAgent and AppAgent to fulfill the user's request.

Round Lifecycle

In a Round, the following steps are executed:

1. Round Initialization

At the beginning of a Round, the Round object is created, and the user's request is processed by the HostAgent to determine the appropriate application to fulfill the request.

2. Action Execution

Once created, the Round orchestrates the HostAgent and AppAgent to execute the necessary actions to fulfill the user's request. The core logic of a Round is shown below:

def run(self) -> None:
    """
    Run the round.
    """

    while not self.is_finished():

        self.agent.handle(self.context)

        self.state = self.agent.state.next_state(self.agent)
        self.agent = self.agent.state.next_agent(self.agent)
        self.agent.set_state(self.state)

        # If the subtask ends, capture the last snapshot of the application.
        if self.state.is_subtask_end():
            time.sleep(configs["SLEEP_TIME"])
            self.capture_last_snapshot(sub_round_id=self.subtask_amount)
            self.subtask_amount += 1

    self.agent.blackboard.add_requests(
        {"request_{i}".format(i=self.id), self.request}
    )

    if self.application_window is not None:
        self.capture_last_snapshot()

    if self._should_evaluate:
        self.evaluation()

At each step, the Round processes the user's request by invoking the handle method of the AppAgent or HostAgent based on the current state. The state determines the next agent to handle the request and the next state to transition to.

3. Request Completion

The AppAgent completes the actions within the application. If the request spans multiple applications, the HostAgent may switch to a different application to continue the task.

4. Round Termination

Once the user's request is fulfilled, the Round is terminated, and the results are returned to the user. If configured, the EvaluationAgent evaluates the completeness of the Round.

Reference

Bases: ABC

A round of a session in UFO. A round manages a single user request and consists of multiple steps. A session may consists of multiple rounds of interactions.

Initialize a round.

Parameters:
  • request (str) –

    The request of the round.

  • agent (BasicAgent) –

    The initial agent of the round.

  • context (Context) –

    The shared context of the round.

  • should_evaluate (bool) –

    Whether to evaluate the round.

  • id (int) –

    The id of the round.
Source code in module/basic.py 
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
def __init__(
    self,
    request: str,
    agent: BasicAgent,
    context: Context,
    should_evaluate: bool,
    id: int,
) -> None:
    """
    Initialize a round.
    :param request: The request of the round.
    :param agent: The initial agent of the round.
    :param context: The shared context of the round.
    :param should_evaluate: Whether to evaluate the round.
    :param id: The id of the round.
    """

    self._request = request
    self._context = context
    self._agent = agent
    self._state = agent.state
    self._id = id
    self._should_evaluate = should_evaluate

    self._init_context()

agent: BasicAgent property writable

Get the agent of the round. return: The agent of the round.

application_window: UIAWrapper property writable

Get the application of the session. return: The application of the session.

context: Context property

Get the context of the round. return: The context of the round.

cost: float property

Get the cost of the round. return: The cost of the round.

id: int property

Get the id of the round. return: The id of the round.

log_path: str property

Get the log path of the round.

return: The log path of the round.

request: str property

Get the request of the round. return: The request of the round.

state: AgentState property writable

Get the status of the round. return: The status of the round.

step: int property

Get the local step of the round. return: The step of the round.

subtask_amount: int property writable

Get the subtask amount of the round. return: The subtask amount of the round.

capture_last_snapshot(sub_round_id=None)

Capture the last snapshot of the application, including the screenshot and the XML file if configured.

Parameters:
  • sub_round_id (Optional[int], default: None ) –

    The id of the sub-round, default is None.
Source code in module/basic.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
def capture_last_snapshot(self, sub_round_id: Optional[int] = None) -> None:
    """
    Capture the last snapshot of the application, including the screenshot and the XML file if configured.
    :param sub_round_id: The id of the sub-round, default is None.
    """

    # Capture the final screenshot
    if sub_round_id is None:
        screenshot_save_path = self.log_path + f"action_round_{self.id}_final.png"
    else:
        screenshot_save_path = (
            self.log_path
            + f"action_round_{self.id}_sub_round_{sub_round_id}_final.png"
        )

    if self.application_window is not None:

        try:
            PhotographerFacade().capture_app_window_screenshot(
                self.application_window, save_path=screenshot_save_path
            )

        except Exception as e:
            utils.print_with_color(
                f"Warning: The last snapshot capture failed, due to the error: {e}",
                "yellow",
            )

        # Save the final XML file
        if configs["LOG_XML"]:
            log_abs_path = os.path.abspath(self.log_path)
            xml_save_path = os.path.join(
                log_abs_path,
                (
                    f"xml/action_round_{self.id}_final.xml"
                    if sub_round_id is None
                    else f"xml/action_round_{self.id}_sub_round_{sub_round_id}_final.xml"
                ),
            )

            if issubclass(type(self.agent), HostAgent):

                app_agent: AppAgent = self.agent.get_active_appagent()
                app_agent.Puppeteer.save_to_xml(xml_save_path)
            elif issubclass(type(self.agent), AppAgent):
                app_agent: AppAgent = self.agent
                app_agent.Puppeteer.save_to_xml(xml_save_path)

evaluation()

TODO: Evaluate the round.

Source code in module/basic.py 
293
294
295
296
297
def evaluation(self) -> None:
    """
    TODO: Evaluate the round.
    """
    pass

is_finished()

Check if the round is finished. return: True if the round is finished, otherwise False.

Source code in module/basic.py 
126
127
128
129
130
131
132
133
134
def is_finished(self) -> bool:
    """
    Check if the round is finished.
    return: True if the round is finished, otherwise False.
    """
    return (
        self.state.is_round_end()
        or self.context.get(ContextNames.SESSION_STEP) >= configs["MAX_STEP"]
    )

print_cost()

Print the total cost of the round.

Source code in module/basic.py 
224
225
226
227
228
229
230
231
232
233
234
def print_cost(self) -> None:
    """
    Print the total cost of the round.
    """

    total_cost = self.cost
    if isinstance(total_cost, float):
        formatted_cost = "${:.2f}".format(total_cost)
        utils.print_with_color(
            f"Request total cost for current round is {formatted_cost}", "yellow"
        )

run()

Run the round.

Source code in module/basic.py 
 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
def run(self) -> None:
    """
    Run the round.
    """

    while not self.is_finished():

        self.agent.handle(self.context)

        self.state = self.agent.state.next_state(self.agent)
        self.agent = self.agent.state.next_agent(self.agent)
        self.agent.set_state(self.state)

        # If the subtask ends, capture the last snapshot of the application.
        if self.state.is_subtask_end():
            time.sleep(configs["SLEEP_TIME"])
            self.capture_last_snapshot(sub_round_id=self.subtask_amount)
            self.subtask_amount += 1

    self.agent.blackboard.add_requests(
        {"request_{i}".format(i=self.id), self.request}
    )

    if self.application_window is not None:
        self.capture_last_snapshot()

    if self._should_evaluate:
        self.evaluation()