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 
17
18
19
20
21
22
23
24
25
def __init__(self, plan_file: str):
    """
    Initialize a plan reader.
    :param plan_file: The path of the plan file.
    """

    with open(plan_file, "r") as f:
        self.plan = json.load(f)
    self.remaining_steps = self.get_steps()

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 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
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_initial_request()

Get the initial request in the plan.

Returns:
  • str –

    The initial request.
Source code in module/sessions/plan_reader.py 
51
52
53
54
55
56
57
58
59
60
61
62
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 
43
44
45
46
47
48
49
def get_operation_object(self) -> str:
    """
    Get the operation object in the step.
    :return: The operation object.
    """

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

get_steps()

Get the steps in the plan.

Returns:
  • List[str] –

    The steps in the plan.
Source code in module/sessions/plan_reader.py 
35
36
37
38
39
40
41
def get_steps(self) -> List[str]:
    """
    Get the steps in the plan.
    :return: The steps in the plan.
    """

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

get_task()

Get the task name.

Returns:
  • str –

    The task name.
Source code in module/sessions/plan_reader.py 
27
28
29
30
31
32
33
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 
79
80
81
82
83
84
85
86
87
88
89
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 
91
92
93
94
95
96
97
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 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
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 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
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 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
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 
249
250
251
252
253
254
255
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()