Learning from Self-Experience

When UFO successfully completes a task, user can choose to save the successful experience to reinforce the AppAgent. The AppAgent can learn from its own successful experiences to improve its performance in the future.

Mechanism

Step 1: Complete a Session

  • Event: UFO completes a session

Step 2: Ask User to Save Experience

  • Action: The agent prompts the user with a choice to save the successful experience

Save Experience

Step 3: User Chooses to Save

  • Action: If the user chooses to save the experience

Step 4: Summarize and Save the Experience

  • Tool: ExperienceSummarizer
  • Process:
  • Summarize the experience into a demonstration example
  • Save the demonstration example in the EXPERIENCE_SAVED_PATH as specified in the config_dev.yaml file
  • The demonstration example includes similar fields as those used in the AppAgent's prompt

Step 5: Retrieve and Utilize Saved Experience

  • When: The AppAgent encounters a similar task in the future
  • Action: Retrieve the saved experience from the experience database
  • Outcome: Use the retrieved experience to generate a plan

Workflow Diagram

graph TD;
    A[Complete Session] --> B[Ask User to Save Experience]
    B --> C[User Chooses to Save]
    C --> D[Summarize with ExperienceSummarizer]
    D --> E[Save in EXPERIENCE_SAVED_PATH]
    F[AppAgent Encounters Similar Task] --> G[Retrieve Saved Experience]
    G --> H[Generate Plan]

Activate the Learning from Self-Experience

Step 1: Configure the AppAgent

Configure the following parameters to allow UFO to use the RAG from its self-experience:

Configuration Option Description Type Default Value
RAG_EXPERIENCE Whether to use the RAG from its self-experience Boolean False
RAG_EXPERIENCE_RETRIEVED_TOPK The topk for the offline retrieved documents Integer 5

Reference

Experience Summarizer

The ExperienceSummarizer class is located in the ufo/experience/experience_summarizer.py file. The ExperienceSummarizer class provides the following methods to summarize the experience:

The ExperienceSummarizer class is the summarizer for the experience learning.

Initialize the ApplicationAgentPrompter.

Parameters:
  • is_visual (bool) –

    Whether the request is for visual model.

  • prompt_template (str) –

    The path of the prompt template.

  • example_prompt_template (str) –

    The path of the example prompt template.

  • api_prompt_template (str) –

    The path of the api prompt template.

Source code in experience/summarizer.py
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def __init__(
    self,
    is_visual: bool,
    prompt_template: str,
    example_prompt_template: str,
    api_prompt_template: str,
):
    """
    Initialize the ApplicationAgentPrompter.
    :param is_visual: Whether the request is for visual model.
    :param prompt_template: The path of the prompt template.
    :param example_prompt_template: The path of the example prompt template.
    :param api_prompt_template: The path of the api prompt template.
    """
    self.is_visual = is_visual
    self.prompt_template = prompt_template
    self.example_prompt_template = example_prompt_template
    self.api_prompt_template = api_prompt_template

build_prompt(log_partition)

Build the prompt.

Parameters:
  • log_partition (dict) –

    The log partition. return: The prompt.

Source code in experience/summarizer.py
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def build_prompt(self, log_partition: dict) -> list:
    """
    Build the prompt.
    :param log_partition: The log partition.
    return: The prompt.
    """
    experience_prompter = ExperiencePrompter(
        self.is_visual,
        self.prompt_template,
        self.example_prompt_template,
        self.api_prompt_template,
    )
    experience_system_prompt = experience_prompter.system_prompt_construction()
    experience_user_prompt = experience_prompter.user_content_construction(
        log_partition
    )
    experience_prompt = experience_prompter.prompt_construction(
        experience_system_prompt, experience_user_prompt
    )

    return experience_prompt

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 experience/summarizer.py
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@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 experience/summarizer.py
127
128
129
130
131
132
133
134
135
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
@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(prompt_message)

Get the summary.

Parameters:
  • prompt_message (list) –

    The prompt message. return: The summary and the cost.

Source code in experience/summarizer.py
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def get_summary(self, prompt_message: list) -> Tuple[dict, float]:
    """
    Get the summary.
    :param prompt_message: The prompt message.
    return: The summary and the cost.
    """

    # Get the completion for the prompt message
    response_string, cost = get_completion(
        prompt_message, "APPAGENT", use_backup_engine=True
    )
    try:
        response_json = json_parser(response_string)
    except:
        response_json = None

    # Restructure the response
    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, cost

get_summary_list(logs)

Get the summary list.

Parameters:
  • logs (list) –

    The logs. return: The summary list and the total cost.

Source code in experience/summarizer.py
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def get_summary_list(self, logs: list) -> Tuple[list, float]:
    """
    Get the summary list.
    :param logs: The logs.
    return: The summary list and the total cost.
    """
    summaries = []
    total_cost = 0.0
    for log_partition in logs:
        prompt = self.build_prompt(log_partition)
        summary, cost = self.get_summary(prompt)
        summary["request"] = ExperienceLogLoader.get_user_request(log_partition)
        summary["app_list"] = ExperienceLogLoader.get_app_list(log_partition)
        summaries.append(summary)
        total_cost += cost

    return summaries, total_cost

read_logs(log_path) staticmethod

Read the log.

Parameters:
  • log_path (str) –

    The path of the log file.

Source code in experience/summarizer.py
117
118
119
120
121
122
123
124
125
@staticmethod
def read_logs(log_path: str) -> list:
    """
    Read the log.
    :param log_path: The path of the log file.
    """
    replay_loader = ExperienceLogLoader(log_path)
    logs = replay_loader.create_logs()
    return logs


Experience Retriever

The ExperienceRetriever class is located in the ufo/rag/retriever.py file. The ExperienceRetriever class provides the following methods to retrieve the experience:

Bases: Retriever

Class to create experience retrievers.

Create a new ExperienceRetriever.

Parameters:
  • db_path

    The path to the database.

Source code in rag/retriever.py
130
131
132
133
134
135
def __init__(self, db_path) -> None:
    """
    Create a new ExperienceRetriever.
    :param db_path: The path to the database.
    """
    self.indexer = self.get_indexer(db_path)

get_indexer(db_path)

Create an experience indexer.

Parameters:
  • db_path (str) –

    The path to the database.

Source code in rag/retriever.py
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def get_indexer(self, db_path: str):
    """
    Create an experience indexer.
    :param 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 experience indexer from {path}.".format(
                path=db_path
            ),
            "yellow",
        )
        return None