Follower Mode

The Follower mode is a feature of UFO that the agent follows a list of pre-defined steps in natural language to take actions on applications. Different from the normal mode, this mode creates an FollowerAgent that follows the plan list provided by the user to interact with the application, instead of generating the plan itself. This mode is useful for debugging and software testing or verification.

Quick Start

Step 1: Create a Plan file

Before starting the Follower mode, you need to create a plan file that contains the list of steps for the agent to follow. The plan file is a JSON file that contains the following fields:

Field Description Type
task The task description. String
steps The list of steps for the agent to follow. List of Strings
object The application or file to interact with. String

Below is an example of a plan file:

{
    "task": "Type in a text of 'Test For Fun' with heading 1 level",
    "steps": 
    [
        "1.type in 'Test For Fun'", 
        "2.Select the 'Test For Fun' text",
        "3.Click 'Home' tab to show the 'Styles' ribbon tab",
        "4.Click 'Styles' ribbon tab to show the style 'Heading 1'",
        "5.Click 'Heading 1' style to apply the style to the selected text"
    ],
    "object": "draft.docx"
}

Note

The object field is the application or file that the agent will interact with. The object must be active (can be minimized) when starting the Follower mode.

Step 2: Start the Follower Mode

To start the Follower mode, run the following command:

# assume you are in the cloned UFO folder
python ufo.py --task_name {task_name} --mode follower --plan {plan_file}

Tip

Replace {task_name} with the name of the task and {plan_file} with the path to the plan file.

Step 3: Run in Batch (Optional)

You can also run the Follower mode in batch mode by providing a folder containing multiple plan files. The agent will follow the plans in the folder one by one. To run in batch mode, run the following command:

# assume you are in the cloned UFO folder
python ufo.py --task_name {task_name} --mode follower --plan {plan_folder}

UFO will automatically detect the plan files in the folder and run them one by one.

Tip

Replace {task_name} with the name of the task and {plan_folder} with the path to the folder containing plan files.

Evaluation

You may want to evaluate the task is completed successfully or not by following the plan. UFO will call the EvaluationAgent to evaluate the task if EVA_SESSION is set to True in the config_dev.yaml file.

You can check the evaluation log in the logs/{task_name}/evaluation.log file.

References

The follower mode employs a PlanReader to parse the plan file and create a FollowerSession to follow the plan.

PlanReader

The PlanReader is located in the ufo/module/sessions/plan_reader.py file.

The reader for a plan file.

Initialize a plan reader.

Parameters:
  • plan_file (str) –

    The path of the plan file.
Source code in module/sessions/plan_reader.py 
18
19
20
21
22
23
24
25
26
27
28
def __init__(self, plan_file: str):
    """
    Initialize a plan reader.
    :param plan_file: The path of the plan file.
    """

    self.plan_file = plan_file
    with open(plan_file, "r") as f:
        self.plan = json.load(f)
    self.remaining_steps = self.get_steps()
    self.support_apps = ["word", "excel", "powerpoint"]

get_close()

Check if the plan is closed.

Returns:
  • bool –

    True if the plan need closed, False otherwise.
Source code in module/sessions/plan_reader.py 
30
31
32
33
34
35
36
def get_close(self) -> bool:
    """
    Check if the plan is closed.
    :return: True if the plan need closed, False otherwise.
    """

    return self.plan.get("close", False)

get_host_agent_request()

Get the request for the host agent.

Returns:
  • str –

    The request for the host agent.
Source code in module/sessions/plan_reader.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def get_host_agent_request(self) -> str:
    """
    Get the request for the host agent.
    :return: The request for the host agent.
    """

    object_name = self.get_operation_object()

    request = (
        f"Open and select the application of {object_name}, and output the FINISH status immediately. "
        "You must output the selected application with their control text and label even if it is already open."
    )

    return request

get_host_request()

Get the request for the host agent.

Returns:
  • str –

    The request for the host agent.
Source code in module/sessions/plan_reader.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def get_host_request(self) -> str:
    """
    Get the request for the host agent.
    :return: The request for the host agent.
    """

    task = self.get_task()
    object_name = self.get_operation_object()
    if object_name in self.support_apps:
        request = task
    else:
        request = f"Open the application of {task}. You must output the selected application with their control text and label even if it is already open."

    return request

get_initial_request()

Get the initial request in the plan.

Returns:
  • str –

    The initial request.
Source code in module/sessions/plan_reader.py 
62
63
64
65
66
67
68
69
70
71
72
73
def get_initial_request(self) -> str:
    """
    Get the initial request in the plan.
    :return: The initial request.
    """

    task = self.get_task()
    object_name = self.get_operation_object()

    request = f"{task} in {object_name}"

    return request

get_operation_object()

Get the operation object in the step.

Returns:
  • str –

    The operation object.
Source code in module/sessions/plan_reader.py 
54
55
56
57
58
59
60
def get_operation_object(self) -> str:
    """
    Get the operation object in the step.
    :return: The operation object.
    """

    return self.plan.get("object", None).lower()

get_root_path()

Get the root path of the plan.

Returns:
  • str –

    The root path of the plan.
Source code in module/sessions/plan_reader.py 
146
147
148
149
150
151
152
def get_root_path(self) -> str:
    """
    Get the root path of the plan.
    :return: The root path of the plan.
    """

    return os.path.dirname(os.path.abspath(self.plan_file))

get_steps()

Get the steps in the plan.

Returns:
  • List[str] –

    The steps in the plan.
Source code in module/sessions/plan_reader.py 
46
47
48
49
50
51
52
def get_steps(self) -> List[str]:
    """
    Get the steps in the plan.
    :return: The steps in the plan.
    """

    return self.plan.get("steps", [])

get_support_apps()

Get the support apps in the plan.

Returns:
  • List[str] –

    The support apps in the plan.
Source code in module/sessions/plan_reader.py 
103
104
105
106
107
108
109
def get_support_apps(self) -> List[str]:
    """
    Get the support apps in the plan.
    :return: The support apps in the plan.
    """

    return self.support_apps

get_task()

Get the task name.

Returns:
  • str –

    The task name.
Source code in module/sessions/plan_reader.py 
38
39
40
41
42
43
44
def get_task(self) -> str:
    """
    Get the task name.
    :return: The task name.
    """

    return self.plan.get("task", "")

next_step()

Get the next step in the plan.

Returns:
  • Optional[str] –

    The next step.
Source code in module/sessions/plan_reader.py 
126
127
128
129
130
131
132
133
134
135
136
def next_step(self) -> Optional[str]:
    """
    Get the next step in the plan.
    :return: The next step.
    """

    if self.remaining_steps:
        step = self.remaining_steps.pop(0)
        return step

    return None

task_finished()

Check if the task is finished.

Returns:
  • bool –

    True if the task is finished, False otherwise.
Source code in module/sessions/plan_reader.py 
138
139
140
141
142
143
144
def task_finished(self) -> bool:
    """
    Check if the task is finished.
    :return: True if the task is finished, False otherwise.
    """

    return not self.remaining_steps


FollowerSession

The FollowerSession is also located in the ufo/module/sessions/session.py file.

Bases: BaseSession

A session for following a list of plan for action taken. This session is used for the follower agent, which accepts a plan file to follow using the PlanReader.

Initialize a session.

Parameters:
  • task (str) –

    The name of current task.

  • plan_file (str) –

    The path of the plan file to follow.

  • should_evaluate (bool) –

    Whether to evaluate the session.

  • id (int) –

    The id of the session.
Source code in module/sessions/session.py 
255
256
257
258
259
260
261
262
263
264
265
266
267
268
def __init__(
    self, task: str, plan_file: str, should_evaluate: bool, id: int
) -> None:
    """
    Initialize a session.
    :param task: The name of current task.
    :param plan_file: The path of the plan file to follow.
    :param should_evaluate: Whether to evaluate the session.
    :param id: The id of the session.
    """

    super().__init__(task, should_evaluate, id)

    self.plan_reader = PlanReader(plan_file)

create_new_round()

Create a new round.

Source code in module/sessions/session.py 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
def create_new_round(self) -> None:
    """
    Create a new round.
    """

    # Get a request for the new round.
    request = self.next_request()

    # Create a new round and return None if the session is finished.
    if self.is_finished():
        return None

    if self.total_rounds == 0:
        utils.print_with_color("Complete the following request:", "yellow")
        utils.print_with_color(self.plan_reader.get_initial_request(), "cyan")
        agent = self._host_agent
    else:
        agent = self._host_agent.get_active_appagent()

        # Clear the memory and set the state to continue the app agent.
        agent.clear_memory()
        agent.blackboard.requests.clear()

        agent.set_state(ContinueAppAgentState())

    round = BaseRound(
        request=request,
        agent=agent,
        context=self.context,
        should_evaluate=configs.get("EVA_ROUND", False),
        id=self.total_rounds,
    )

    self.add_round(round.id, round)

    return round

next_request()

Get the request for the new round.

Source code in module/sessions/session.py 
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
def next_request(self) -> str:
    """
    Get the request for the new round.
    """

    # If the task is finished, return an empty string.
    if self.plan_reader.task_finished():
        self._finish = True
        return ""

    # Get the request from the plan reader.
    if self.total_rounds == 0:
        return self.plan_reader.get_host_agent_request()
    else:
        return self.plan_reader.next_step()

request_to_evaluate()

Check if the session should be evaluated.

Returns:
  • bool –

    True if the session should be evaluated, False otherwise.
Source code in module/sessions/session.py 
331
332
333
334
335
336
337
def request_to_evaluate(self) -> bool:
    """
    Check if the session should be evaluated.
    :return: True if the session should be evaluated, False otherwise.
    """

    return self.plan_reader.get_task()