Here is the polished document for your Python code project:

Learning from User Demonstration

For complex tasks, users can demonstrate the task using Step Recorder to record the action trajectories. UFO can learn from these user demonstrations to improve the AppAgent's performance.

Mechanism

UFO use the Step Recorder tool to record the task and action trajectories. The recorded demonstration is saved as a zip file. The DemonstrationSummarizer class extracts and summarizes the demonstration. The summarized demonstration is saved in the DEMONSTRATION_SAVED_PATH as specified in the config_dev.yaml file. When the AppAgent encounters a similar task, the DemonstrationRetriever class retrieves the saved demonstration from the demonstration database and generates a plan based on the retrieved demonstration.

Info

You can find how to record the task and action trajectories using the Step Recorder tool in the User Demonstration Provision document.

You can find a demo video of learning from user demonstrations:


Activating Learning from User Demonstrations

Step 1: User Demonstration

Please follow the steps in the User Demonstration Provision document to provide user demonstrations.

Step 2: Configure the AppAgent

Configure the following parameters to allow UFO to use RAG from user demonstrations:

Configuration Option Description Type Default Value
RAG_DEMONSTRATION Whether to use RAG from user demonstrations Boolean False
RAG_DEMONSTRATION_RETRIEVED_TOPK The top K documents to retrieve offline Integer 5
RAG_DEMONSTRATION_COMPLETION_N The number of completion choices for the demonstration result Integer 3

Reference

Demonstration Summarizer

The DemonstrationSummarizer class is located in the record_processor/summarizer/summarizer.py file. The DemonstrationSummarizer class provides methods to summarize the demonstration:

The DemonstrationSummarizer class is the summarizer for the demonstration learning. It summarizes the demonstration record to a list of summaries, and save the summaries to the YAML file and the vector database. A sample of the summary is as follows: { "example": { "Observation": "Word.exe is opened.", "Thought": "The user is trying to create a new file.", "ControlLabel": "1", "ControlText": "Sample Control Text", "Function": "CreateFile", "Args": "filename='new_file.txt'", "Status": "Success", "Plan": "Create a new file named 'new_file.txt'.", "Comment": "The user successfully created a new file." }, "Tips": "You can use the 'CreateFile' function to create a new file." }

Initialize the DemonstrationSummarizer.

Parameters:
  • is_visual (bool) –

    Whether the request is for visual model.

  • prompt_template (str) –

    The path of the prompt template.

  • demonstration_prompt_template (str) –

    The path of the example prompt template for demonstration.

  • api_prompt_template (str) –

    The path of the api prompt template.

Source code in summarizer/summarizer.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(
    self,
    is_visual: bool,
    prompt_template: str,
    demonstration_prompt_template: str,
    api_prompt_template: str,
    completion_num: int = 1,
):
    """
    Initialize the DemonstrationSummarizer.
    :param is_visual: Whether the request is for visual model.
    :param prompt_template: The path of the prompt template.
    :param demonstration_prompt_template: The path of the example prompt template for demonstration.
    :param api_prompt_template: The path of the api prompt template.
    """
    self.is_visual = is_visual
    self.prompt_template = prompt_template
    self.demonstration_prompt_template = demonstration_prompt_template
    self.api_prompt_template = api_prompt_template
    self.completion_num = completion_num

__build_prompt(demo_record)

Build the prompt by the user demonstration record.

Parameters:
  • demo_record (DemonstrationRecord) –

    The user demonstration record. return: The prompt.

Source code in summarizer/summarizer.py
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def __build_prompt(self, demo_record: DemonstrationRecord) -> list:
    """
    Build the prompt by the user demonstration record.
    :param demo_record: The user demonstration record.
    return: The prompt.
    """
    demonstration_prompter = DemonstrationPrompter(
        self.is_visual,
        self.prompt_template,
        self.demonstration_prompt_template,
        self.api_prompt_template,
    )
    demonstration_system_prompt = (
        demonstration_prompter.system_prompt_construction()
    )
    demonstration_user_prompt = demonstration_prompter.user_content_construction(
        demo_record
    )
    demonstration_prompt = demonstration_prompter.prompt_construction(
        demonstration_system_prompt, demonstration_user_prompt
    )

    return demonstration_prompt

__parse_response(response_string)

Parse the response string to a dict of summary.

Parameters:
  • response_string (str) –

    The response string. return: The summary dict.

Source code in summarizer/summarizer.py
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def __parse_response(self, response_string: str) -> dict:
    """
    Parse the response string to a dict of summary.
    :param response_string: The response string.
    return: The summary dict.
    """
    try:
        response_json = json_parser(response_string)
    except:
        response_json = None

    # Restructure the response, in case any of the keys are missing, set them to empty string.
    if response_json:
        summary = dict()
        summary["example"] = {}
        for key in [
            "Observation",
            "Thought",
            "ControlLabel",
            "ControlText",
            "Function",
            "Args",
            "Status",
            "Plan",
            "Comment",
        ]:
            summary["example"][key] = response_json.get(key, "")
        summary["Tips"] = response_json.get("Tips", "")

        return summary

create_or_update_vector_db(summaries, db_path) staticmethod

Create or update the vector database.

Parameters:
  • summaries (list) –

    The summaries.

  • db_path (str) –

    The path of the vector database.

Source code in summarizer/summarizer.py
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
@staticmethod
def create_or_update_vector_db(summaries: list, db_path: str):
    """
    Create or update the vector database.
    :param summaries: The summaries.
    :param db_path: The path of the vector database.
    """

    document_list = []

    for summary in summaries:
        request = summary["request"]
        document_list.append(Document(page_content=request, metadata=summary))

    db = FAISS.from_documents(document_list, get_hugginface_embedding())

    # Check if the db exists, if not, create a new one.
    if os.path.exists(db_path):
        prev_db = FAISS.load_local(db_path, get_hugginface_embedding())
        db.merge_from(prev_db)

    db.save_local(db_path)

    print(f"Updated vector DB successfully: {db_path}")

create_or_update_yaml(summaries, yaml_path) staticmethod

Create or update the YAML file.

Parameters:
  • summaries (list) –

    The summaries.

  • yaml_path (str) –

    The path of the YAML file.

Source code in summarizer/summarizer.py
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
@staticmethod
def create_or_update_yaml(summaries: list, yaml_path: str):
    """
    Create or update the YAML file.
    :param summaries: The summaries.
    :param yaml_path: The path of the YAML file.
    """

    # Check if the file exists, if not, create a new one
    if not os.path.exists(yaml_path):
        with open(yaml_path, "w"):
            pass
        print(f"Created new YAML file: {yaml_path}")

    # Read existing data from the YAML file
    with open(yaml_path, "r") as file:
        existing_data = yaml.safe_load(file)

    # Initialize index and existing_data if file is empty
    index = len(existing_data) if existing_data else 0
    existing_data = existing_data or {}

    # Update data with new summaries
    for i, summary in enumerate(summaries):
        example = {f"example{index + i}": summary}
        existing_data.update(example)

    # Write updated data back to the YAML file
    with open(yaml_path, "w") as file:
        yaml.safe_dump(
            existing_data, file, default_flow_style=False, sort_keys=False
        )

    print(f"Updated existing YAML file successfully: {yaml_path}")

get_summary_list(record)

Get the summary list for a record

Parameters:
  • record (DemonstrationRecord) –

    The demonstration record. return: The summary list for the user defined completion number and the cost

Source code in summarizer/summarizer.py
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def get_summary_list(self, record: DemonstrationRecord) -> Tuple[list, float]:
    """
    Get the summary list for a record
    :param record: The demonstration record.
    return: The summary list for the user defined completion number and the cost
    """

    prompt = self.__build_prompt(record)
    response_string_list, cost = get_completions(
        prompt, "APPAGENT", use_backup_engine=True, n=self.completion_num
    )
    summaries = []
    for response_string in response_string_list:
        summary = self.__parse_response(response_string)
        if summary:
            summary["request"] = record.get_request()
            summary["app_list"] = record.get_applications()
            summaries.append(summary)

    return summaries, cost


Demonstration Retriever

The DemonstrationRetriever class is located in the rag/retriever.py file. The DemonstrationRetriever class provides methods to retrieve the demonstration:

Bases: Retriever

Class to create demonstration retrievers.

Create a new DemonstrationRetriever. :db_path: The path to the database.

Source code in rag/retriever.py
198
199
200
201
202
203
def __init__(self, db_path) -> None:
    """
    Create a new DemonstrationRetriever.
    :db_path: The path to the database.
    """
    self.indexer = self.get_indexer(db_path)

get_indexer(db_path)

Create a demonstration indexer. :db_path: The path to the database.

Source code in rag/retriever.py
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def get_indexer(self, db_path: str):
    """
    Create a demonstration indexer.
    :db_path: The path to the database.
    """

    try:
        db = FAISS.load_local(db_path, get_hugginface_embedding())
        return db
    except:
        # print_with_color(
        #     "Warning: Failed to load demonstration indexer from {path}.".format(
        #         path=db_path
        #     ),
        #     "yellow",
        # )
        return None