diff --git a/README.md b/README.md
index 7c72fd16..d341696b 100644
--- a/README.md
+++ b/README.md
@@ -240,6 +240,7 @@ You may use them to debug, replay, or analyze the agent output.
 * For other communications, please contact [ufo-agent@microsoft.com](mailto:ufo-agent@microsoft.com).
 ---
 
+
 <!-- ## 🎬 Demo Examples
 
 We present two demo videos that complete user request on Windows OS using UFO. For more case study, please consult our [technical report](https://arxiv.org/abs/2402.07939).
@@ -260,6 +261,8 @@ https://github.com/microsoft/UFO/assets/11352048/aa41ad47-fae7-4334-8e0b-ba71c4f
 
 
 
+
+
 ## 📊 Evaluation
 
 Please consult the [WindowsBench](https://arxiv.org/pdf/2402.07939.pdf) provided in Section A of the Appendix within our technical report. Here are some tips (and requirements) to aid in completing your request:
diff --git a/SUPPORT.md b/SUPPORT.md
deleted file mode 100644
index 7e722035..00000000
--- a/SUPPORT.md
+++ /dev/null
@@ -1,16 +0,0 @@
-# Support
-
-## How to file issues and get help  
-
-This project uses GitHub Issues to track bugs and feature requests. Please search the existing 
-issues before filing new issues to avoid duplicates.  For new issues, file your bug or 
-feature request as a new Issue.
-
-You may use [GitHub Issues](https://github.com/microsoft/UFO/issues) to raise questions, bug reports, and feature requests.
-
-For help and questions about using this project, please please contact [ufo-agent@microsoft.com](mailto:ufo-agent@microsoft.com).
-
-
-## Microsoft Support Policy  
-
-Support for this **PROJECT or PRODUCT** is limited to the resources listed above.
diff --git a/assets/dataflow/execution.png b/assets/dataflow/execution.png
new file mode 100644
index 00000000..1d962afb
Binary files /dev/null and b/assets/dataflow/execution.png differ
diff --git a/assets/dataflow/instantiation.png b/assets/dataflow/instantiation.png
new file mode 100644
index 00000000..f975178f
Binary files /dev/null and b/assets/dataflow/instantiation.png differ
diff --git a/assets/dataflow/overview.png b/assets/dataflow/overview.png
new file mode 100644
index 00000000..abc3c6eb
Binary files /dev/null and b/assets/dataflow/overview.png differ
diff --git a/assets/dataflow/result_example.png b/assets/dataflow/result_example.png
new file mode 100644
index 00000000..accaf5bf
Binary files /dev/null and b/assets/dataflow/result_example.png differ
diff --git a/dataflow/.gitignore b/dataflow/.gitignore
new file mode 100644
index 00000000..ef4cbe6b
--- /dev/null
+++ b/dataflow/.gitignore
@@ -0,0 +1,5 @@
+# Ignore files
+cache/
+controls_cache/
+controller/utils/
+config/config.yaml
\ No newline at end of file
diff --git a/dataflow/README.md b/dataflow/README.md
new file mode 100644
index 00000000..5a1fe877
--- /dev/null
+++ b/dataflow/README.md
@@ -0,0 +1,471 @@
+# Dataflow
+
+Dataflow uses UFO to implement `instantiation`, `execution`, and `dataflow` for a given task, with options for batch processing and single processing.
+
+1. **Instantiation**:  Instantiation refers to the process of setting up and preparing a task for execution. This step typically involves `choosing template`, `prefill` and `filter`.
+2. **Execution**: Execution is the actual process of running the task. This step involves carrying out the actions or operations specified by the `Instantiation`. And after execution, an evaluate agent will evaluate the quality of the whole execution process.
+3. **Dataflow**: Dataflow is the overarching process that combines **instantiation** and **execution** into a single pipeline. It provides an end-to-end solution for processing tasks, ensuring that all necessary steps (from initialization to execution) are seamlessly integrated.
+
+You can use `instantiation` and `execution` independently if you only need to perform one specific part of the process. When both steps are required for a task, the `dataflow` process streamlines them, allowing you to execute tasks from start to finish in a single pipeline.
+
+The overall processing of dataflow is as below. Given a task-plan data, the LLMwill instantiatie the task-action data, including choosing template, prefill, filter.
+
+![](../assets\dataflow\overview.png)
+
+## How To Use
+
+### 1. Install Packages
+
+You should install the necessary packages in the UFO root folder:
+
+```bash
+pip install -r requirements.txt
+```
+
+### 2. Configure the LLMs
+
+Before running dataflow, you need to provide your LLM configurations **individually for PrefillAgent and FilterAgent**. You can create your own config file `dataflow/config/config.yaml`, by copying the `dataflow/config/config.yaml.template` and editing config for **PREFILL_AGENT** and **FILTER_AGENT** as follows:
+
+#### OpenAI
+
+```bash
+VISUAL_MODE: True, # Whether to use the visual mode
+API_TYPE: "openai" , # The API type, "openai" for the OpenAI API.  
+API_BASE: "https://api.openai.com/v1/chat/completions", # The the OpenAI API endpoint.
+API_KEY: "sk-",  # The OpenAI API key, begin with sk-
+API_VERSION: "2024-02-15-preview", # "2024-02-15-preview" by default
+API_MODEL: "gpt-4-vision-preview",  # The only OpenAI model
+```
+
+#### Azure OpenAI (AOAI)
+
+```bash
+VISUAL_MODE: True, # Whether to use the visual mode
+API_TYPE: "aoai" , # The API type, "aoai" for the Azure OpenAI.  
+API_BASE: "YOUR_ENDPOINT", #  The AOAI API address. Format: https://{your-resource-name}.openai.azure.com
+API_KEY: "YOUR_KEY",  # The aoai API key
+API_VERSION: "2024-02-15-preview", # "2024-02-15-preview" by default
+API_MODEL: "gpt-4-vision-preview",  # The only OpenAI model
+API_DEPLOYMENT_ID: "YOUR_AOAI_DEPLOYMENT", # The deployment id for the AOAI API
+```
+
+You can also non-visial model (e.g., GPT-4) for each agent, by setting `VISUAL_MODE: False` and proper `API_MODEL` (openai) and `API_DEPLOYMENT_ID` (aoai).
+
+#### Non-Visual Model Configuration
+
+You can utilize non-visual models (e.g., GPT-4) for each agent by configuring the following settings in the `config.yaml` file:
+
+- ``VISUAL_MODE: False # To enable non-visual mode.``
+- Specify the appropriate `API_MODEL` (OpenAI) and `API_DEPLOYMENT_ID` (AOAI) for each agent.
+
+Ensure you configure these settings accurately to leverage non-visual models effectively.
+
+#### Other Configurations
+
+`config_dev.yaml` specifies the paths of relevant files and contains default settings. The match strategy for the window match and control filter supports options:  `'contains'`, `'fuzzy'`, and `'regex'`, allowing flexible matching strategy for users. The `MAX_STEPS` is the max step for the execute_flow, which can be set by users.
+
+#### NOTE 💡
+
+**BE CAREFUL!** If you are using GitHub or other open-source tools, do not expose your `config.yaml` online, as it contains your private keys.
+
+### 3. Prepare Files
+
+Certain files need to be prepared before running the task.
+
+#### 3.1. Tasks as JSON
+
+The tasks that need to be instantiated should be organized in a folder of JSON files, with the default folder path set to dataflow `/tasks `. This path can be changed in the `dataflow/config/config.yaml` file, or you can specify it in the terminal, as mentioned in **4. Start Running**. For example, a task stored in `dataflow/tasks/prefill/` may look like this:
+
+```json
+{
+    // The app you want to use
+    "app": "word",
+    // A unique ID to distinguish different tasks 
+    "unique_id": "1",
+    // The task and steps to be instantiated
+    "task": "Type 'hello' and set the font type to Arial",
+    "refined_steps": [
+        "Type 'hello'",
+        "Set the font to Arial"
+    ]
+}
+```
+
+#### 3.2. Templates and Descriptions
+
+You should place an app file as a reference for instantiation in a folder named after the app.
+
+For example, if you have `template1.docx` for Word, it should be located at `dataflow/templates/word/template1.docx`.
+
+Additionally, for each app folder, there should be a `description.json` file located at `dataflow/templates/word/description.json`, which describes each template file in detail. It may look like this:
+
+```json
+{
+    "template1.docx": "A document with a rectangle shape",
+    "template2.docx": "A document with a line of text"
+}
+```
+
+If a `description.json` file is not present, one template file will be selected at random.
+
+#### 3.3. Final Structure
+
+Ensure the following files are in place:
+
+- [X] JSON files to be instantiated
+- [X] Templates as references for instantiation
+- [X] Description file in JSON format
+
+The structure of the files can be:
+
+```bash
+dataflow/
+|
+├── tasks
+│   └── prefill
+│       ├── bulleted.json
+│       ├── delete.json
+│       ├── draw.json
+│       ├── macro.json
+│       └── rotate.json
+├── templates
+│   └── word
+│       ├── description.json
+│       ├── template1.docx
+│       ├── template2.docx
+│       ├── template3.docx
+│       ├── template4.docx
+│       ├── template5.docx
+│       ├── template6.docx
+│       └── template7.docx
+└── ...
+```
+
+### 4. Start Running
+
+After finishing the previous steps, you can use the following commands in the command line. We provide single / batch process, for which you need to give the single file path / folder path. Determine the type of path provided by the user and automatically decide whether to process a single task or batch tasks.
+
+Also, you can choose to use `instantiation` / `execution` sections individually, or use them as a whole section, which is named as `dataflow`.
+
+The default task hub is set to be `"TASKS_HUB"` in `dataflow/config_dev.yaml`.
+
+1. Dataflow Task:
+
+- ```bash
+  python -m dataflow -dataflow --task_path path_to_task_file
+  ```
+
+* Instantiation Task:
+
+  ```bash
+  python -m dataflow -instantiation --task_path path_to_task_file
+  ```
+* Execution Task:
+
+  ```bash
+  python -m dataflow -execution --task_path path_to_task_file
+  ```
+
+## Workflow
+
+### Instantiation
+
+There are three key steps in the instantiation process:
+
+1. `Choose a template` file according to the specified app and instruction.
+2. `Prefill` the task using the current screenshot.
+3. `Filter` the established task.
+
+Given the initial task, the dataflow first choose a template (`Phase 1`), the prefill the initial task based on word envrionment to obtain task-action data (`Phase 2`). Finnally, it will filter the established task to evaluate the quality of task-action data.
+
+![](../assets\dataflow\instantiation.png)
+
+#### 1. Choose Template File
+
+Templates for your app must be defined and described in `dataflow/templates/app`. For instance, if you want to instantiate tasks for the Word application, place the relevant `.docx` files in dataflow `/templates/word `, along with a `description.json` file.
+
+The appropriate template will be selected based on how well its description matches the instruction.
+
+#### 2. Prefill the Task
+
+After selecting the template file, it will be opened, and a screenshot will be taken. If the template file is currently in use, errors may occur.
+
+The screenshot will be sent to the action prefill agent, which will return a modified task.
+
+#### 3. Filter Task
+
+The completed task will be evaluated by a filter agent, which will assess it and provide feedback.
+
+### Execution
+
+The instantiated plans will be executed by a execute task. After execution, evalution agent will evaluation the quality of the entire execution process.
+
+In this phase, given the task-action data, the execution process will match the real controller based on word environment and execute the plan step by step.
+
+![](../assets\dataflow\execution.png)
+
+## Result
+
+The structure of the results of the task is as below:
+
+```
+UFO/
+├── dataflow/                       # Root folder for dataflow
+│   └── results/                    # Directory for storing task processing results
+│       ├── saved_document/         # Directory for final document results
+│       ├── instantiation/          # Directory for instantiation results
+│       │   ├── instantiation_pass/ # Tasks successfully instantiated
+│       │   └── instantiation_fail/ # Tasks that failed instantiation
+│       ├── execution/              # Directory for execution results
+│       │   ├── execution_pass/     # Tasks successfully executed
+│       │   ├── execution_fail/     # Tasks that failed execution
+│       │   └── execution_unsure/   # Tasks with uncertain execution results
+│       ├── dataflow/               # Directory for dataflow results
+│       │   ├── execution_pass/     # Tasks successfully executed
+│       │   ├── execution_fail/     # Tasks that failed execution
+│       │   └── execution_unsure/   # Tasks with uncertain execution results
+│       └── ...
+└── ...
+```
+
+1. **General Description:**
+
+   This directory structure organizes the results of task processing into specific categories, including instantiation, execution, and dataflow outcomes.
+2. **Instantiation:**
+
+   The `instantiation` directory contains subfolders for tasks that were successfully instantiated (`instantiation_pass`) and those that failed during instantiation (`instantiation_fail`).
+3. **Execution:**
+
+   Results of task execution are stored under the `execution` directory, categorized into successful tasks (`execution_pass`), failed tasks (`execution_fail`), and tasks with uncertain outcomes (`execution_unsure`).
+4. **Dataflow Results:**
+
+   The `dataflow` directory similarly holds results of tasks based on execution success, failure, or uncertainty, providing a comprehensive view of the data processing pipeline.
+5. **Saved Documents:**
+
+   Instantiated results are separately stored in the `saved_document` directory for easy access and reference.
+
+### Description
+
+his section illustrates the structure of the result of the task, organized in a hierarchical format to describe the various fields and their purposes. The result data include `unique_id`，``app``, `original`, `execution_result`, `instantiation_result`, `time_cost`.
+
+#### 1. **Field Descriptions**
+
+- **Hierarchy**: The data is presented in a hierarchical manner to allow for a clearer understanding of field relationships.
+- **Type Description**: The type of each field (e.g., `string`, `array`, `object`) clearly specifies the format of the data.
+- **Field Purpose**: Each field has a brief description outlining its function.
+
+#### 2. **Execution Results and Errors**
+
+- **execution_result**: Contains the results of task execution, including subtask performance, completion status, and any encountered errors.
+- **instantiation_result**: Describes the process of task instantiation, including template selection, prefilled tasks, and instantiation evaluation.
+- **error**: If an error occurs during task execution, this field will contain the relevant error information.
+
+#### 3. **Time Consumption**
+
+- **time_cost**: The time spent on each phase of the task, from template selection to task execution, is recorded to analyze task efficiency.
+
+### Example Data
+
+```json
+{
+    "unique_id": "102",
+    "app": "word",
+    "original": {
+        "original_task": "Find which Compatibility Mode you are in for Word",
+        "original_steps": [
+            "1.Click the **File** tab.",
+            "2.Click **Info**.",
+            "3.Check the **Compatibility Mode** indicator at the bottom of the document preview pane."
+        ]
+    },
+    "execution_result": {
+        "result": {
+            "reason": "The agent successfully identified the compatibility mode of the Word document.",
+            "sub_scores": {
+                "correct identification of compatibility mode": "yes"
+            },
+            "complete": "yes"
+        },
+        "error": null
+    },
+    "instantiation_result": {
+        "choose_template": {
+            "result": "dataflow\\results\\saved_document\\102.docx",
+            "error": null
+        },
+        "prefill": {
+            "result": {
+                "instantiated_request": "Identify the Compatibility Mode of the Word document.",
+                "instantiated_plan": [
+                    {
+                        "Step": 1,
+                        "Subtask": "Identify the Compatibility Mode",
+                        "Function": "summary",
+                        "Args": {
+                            "text": "The document is in '102 - Compatibility Mode'."
+                        },
+                        "Success": true
+                    }
+                ]
+            },
+            "error": null
+        },
+        "instantiation_evaluation": {
+            "result": {
+                "judge": true,
+                "thought": "Identifying the Compatibility Mode of a Word document is a task that can be executed locally within Word."
+            },
+            "error": null
+        }
+    },
+    "time_cost": {
+        "choose_template": 0.017,
+        "prefill": 11.304,
+        "instantiation_evaluation": 2.38,
+        "total": 34.584,
+        "execute": 0.946,
+        "execute_eval": 10.381
+    }
+}
+```
+
+## Quick Start
+
+We prepare two cases to show the dataflow, which can be found in `dataflow\tasks\prefill`. So after installing required packages, you can type the following command in the command line:
+
+```
+python -m dataflow -dataflow
+```
+
+And you can see the hints showing in the terminal, which means the dataflow is working.
+
+### Structure of related files
+
+After the two tasks are finished, the task and output files would appear as follows:
+
+UFO/
+├── dataflow/
+│   └── results/
+│       ├── saved_document/       	# Directory for saved documents
+│       │   ├── bulleted.docx     	# Result of the "bulleted" task
+│       │   └── rotate.docx       	# Result of the "rotate" task
+│       ├── dataflow/            		 # Dataflow results directory
+│       │   ├── execution_pass/   	# Successfully executed tasks
+│       │   │   ├── bulleted.json 	# Execution result for the "bulleted" task
+│       │   │   ├── rotate.json  	 # Execution result for the "rotate" task
+│       │   │   └── ...
+└── ...
+
+### Result files
+
+The result stucture of bulleted task is shown as below. This document provides a detailed breakdown of the task execution process for turning lines of text into a bulleted list in Word. It includes the original task description, execution results, and time analysis for each step.
+
+* **`unique_id`** : The identifier for the task, in this case, `"5"`.
+* **`app`** : The application being used, which is `"word"`.
+* **`original`** : Contains the original task description and the steps.
+
+  * **`original_task`** : Describes the task in simple terms (turning text into a bulleted list).
+  * **`original_steps`** : Lists the steps required to perform the task.
+* **`execution_result`** : Provides the result of executing the task.
+
+  * **`result`** : Describes the outcome of the execution, including a success message and sub-scores for each part of the task. The `complete: "yes"` means the evaluation agent think the execution process is successful! The `sub_score` is the evaluation of each subtask, corresponding to the ` instantiated_plan` in the  `prefill`.
+  * **`error`** : If any error occurred during execution, it would be reported here, but it's `null` in this case.
+* **`instantiation_result`** : Details the instantiation of the task (setting up the task for execution).
+
+  * **`choose_template`** : Path to the template or document created during the task (in this case, the bulleted list document).
+  * **`prefill`** : Describes the `instantiated_request` and  `instantiated_plan` and the steps involved, such as selecting text and clicking buttons, which is the result of prefill flow. The `Success` and `MatchedControlText` is added in the execution process. **`Success`** indicates whether the subtask was executed successfully. **`MatchedControlText`** refers to the control text that was matched during the execution process based on the plan.
+  * **`instantiation_evaluation`** : Provides feedback on the task's feasibility and the evaluation of the request, which is result of the filter flow. **`"judge": true`** : This indicates that the evaluation of the task was positive, meaning the task is considered valid or successfully judged. And the `thought ` is the detailed reason.
+* **`time_cost`** : The time spent on different parts of the task, including template selection, prefill, instantiation evaluation, and execution. Total time is also given.
+
+This structure follows your description and provides the necessary details in a consistent format.
+
+```json
+{
+    "unique_id": "5",
+    "app": "word",
+    "original": {
+        "original_task": "Turning lines of text into a bulleted list in Word",
+        "original_steps": [
+            "1. Place the cursor at the beginning of the line of text you want to turn into a bulleted list",
+            "2. Click the Bullets button in the Paragraph group on the Home tab and choose a bullet style"
+        ]
+    },
+    "execution_result": {
+        "result": {
+            "reason": "The agent successfully selected the text 'text to edit' and then clicked on the 'Bullets' button in the Word application. The final screenshot shows that the text 'text to edit' has been converted into a bulleted list.",
+            "sub_scores": {
+                "text selection": "yes",
+                "bulleted list conversion": "yes"
+            },
+            "complete": "yes"
+        },
+        "error": null
+    },
+    "instantiation_result": {
+        "choose_template": {
+            "result": "dataflow\\results\\saved_document\\bulleted.docx",
+            "error": null
+        },
+        "prefill": {
+            "result": {
+                "instantiated_request": "Turn the line of text 'text to edit' into a bulleted list in Word.",
+                "instantiated_plan": [
+                    {
+                        "Step": 1,
+                        "Subtask": "Place the cursor at the beginning of the text 'text to edit'",
+                        "ControlLabel": null,
+                        "ControlText": "",
+                        "Function": "select_text",
+                        "Args": {
+                            "text": "text to edit"
+                        },
+                        "Success": true,
+                        "MatchedControlText": null
+                    },
+                    {
+                        "Step": 2,
+                        "Subtask": "Click the Bullets button in the Paragraph group on the Home tab",
+                        "ControlLabel": "61",
+                        "ControlText": "Bullets",
+                        "Function": "click_input",
+                        "Args": {
+                            "button": "left",
+                            "double": false
+                        },
+                        "Success": true,
+                        "MatchedControlText": "Bullets"
+                    }
+                ]
+            },
+            "error": null
+        },
+        "instantiation_evaluation": {
+            "result": {
+                "judge": true,
+                "thought": "The task is specific and involves a basic function in Word that can be executed locally without any external dependencies.",
+                "request_type": "None"
+            },
+            "error": null
+        }
+    },
+    "time_cost": {
+        "choose_template": 0.012,
+        "prefill": 15.649,
+        "instantiation_evaluation": 2.469,
+        "execute": 5.824,
+        "execute_eval": 8.702,
+        "total": 43.522
+    }
+}
+```
+
+### Log files
+
+The corresponding logs can be found in the directories `logs/bulleted` and `logs/rotate`, as shown below. Detailed logs for each workflow are recorded, capturing every step of the execution process.
+
+![img](../assets\dataflow\result_example.png)
+
+## Notes
+
+1. Users should be careful to save the original files while using this project; otherwise, the files will be closed when the app is shut down.
+2. After starting the project, users should not close the app window while the program is taking screenshots.
diff --git a/dataflow/__main__.py b/dataflow/__main__.py
new file mode 100644
index 00000000..36abd01a
--- /dev/null
+++ b/dataflow/__main__.py
@@ -0,0 +1,7 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+from dataflow import dataflow
+
+if __name__ == "__main__":
+    # Execute the main script
+    dataflow.main()
diff --git a/dataflow/config/config.py b/dataflow/config/config.py
new file mode 100644
index 00000000..4760ad0c
--- /dev/null
+++ b/dataflow/config/config.py
@@ -0,0 +1,40 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from ufo.config.config import Config
+
+
+class Config(Config):
+    _instance = None
+
+    def __init__(self, config_path="dataflow/config/"):
+        """
+        Initializes the Config class.
+        :param config_path: The path to the config file.
+        """
+
+        self.config_data = self.load_config(config_path)
+
+    @staticmethod
+    def get_instance():
+        """
+        Get the instance of the Config class.
+        :return: The instance of the Config class.
+        """
+
+        if Config._instance is None:
+            Config._instance = Config()
+
+        return Config._instance
+
+    def optimize_configs(self, configs):
+        """
+        Optimize the configurations.
+        :param configs: The configurations to optimize.
+        :return: The optimized configurations.
+        """
+        
+        self.update_api_base(configs, "PREFILL_AGENT")
+        self.update_api_base(configs, "FILTER_AGENT")
+
+        return configs
diff --git a/dataflow/config/config.yaml.template b/dataflow/config/config.yaml.template
new file mode 100644
index 00000000..6cd07039
--- /dev/null
+++ b/dataflow/config/config.yaml.template
@@ -0,0 +1,55 @@
+PREFILL_AGENT: {
+  VISUAL_MODE: True, # Whether to use the visual mode
+
+  API_TYPE: "openai" , # The API type, "openai" for the OpenAI API, "aoai" for the AOAI API, 'azure_ad' for the ad authority of the AOAI API.  
+  API_BASE: "https://api.openai.com/v1/chat/completions", # The the OpenAI API endpoint, "https://api.openai.com/v1/chat/completions" for the OpenAI API.
+  API_KEY: "sk-",  # The OpenAI API key, begin with sk-
+  API_VERSION: "2024-02-15-preview", # "2024-02-15-preview" by default
+  API_MODEL: "gpt-4-vision-preview",  # The only OpenAI model by now that accepts visual input
+
+
+  ### Comment above and uncomment these if using "aoai".
+  # API_TYPE: "aoai" , # The API type, "openai" for the OpenAI API, "aoai" for the Azure OpenAI.  
+  # API_BASE: "YOUR_ENDPOINT", # The the OpenAI API endpoint, "https://api.openai.com/v1/chat/completions" for the OpenAI API. As for the aoai, it should be https://{your-resource-name}.openai.azure.com
+  # API_KEY: "YOUR_KEY",  # The aoai API key
+  # API_VERSION: "2024-02-15-preview", # "2024-02-15-preview" by default
+  # API_MODEL: "YOUR_MODEL",  # The only OpenAI model by now that accepts visual input
+  # API_DEPLOYMENT_ID: "gpt-4-visual-preview", # The deployment id for the AOAI API
+  
+  ### For Azure_AD
+  # AAD_TENANT_ID: "YOUR_TENANT_ID", # Set the value to your tenant id for the llm model
+  # AAD_API_SCOPE: "YOUR_SCOPE", # Set the value to your scope for the llm model
+  # AAD_API_SCOPE_BASE: "YOUR_SCOPE_BASE" # Set the value to your scope base for the llm model, whose format is API://YOUR_SCOPE_BASE, and the only need is the YOUR_SCOPE_BASE
+}
+
+FILTER_AGENT: {
+  VISUAL_MODE: True, # Whether to use the visual mode
+
+  API_TYPE: "openai" , # The API type, "openai" for the OpenAI API, "aoai" for the AOAI API, 'azure_ad' for the ad authority of the AOAI API.  
+  API_BASE: "https://api.openai.com/v1/chat/completions", # The the OpenAI API endpoint, "https://api.openai.com/v1/chat/completions" for the OpenAI API.
+  API_KEY: "sk-",  # The OpenAI API key, begin with sk-
+  API_VERSION: "2024-02-15-preview", # "2024-02-15-preview" by default
+  API_MODEL: "gpt-4-vision-preview",  # The only OpenAI model by now that accepts visual input
+
+
+  ### Comment above and uncomment these if using "aoai".
+  # API_TYPE: "aoai" , # The API type, "openai" for the OpenAI API, "aoai" for the Azure OpenAI.  
+  # API_BASE: "YOUR_ENDPOINT", # The the OpenAI API endpoint, "https://api.openai.com/v1/chat/completions" for the OpenAI API. As for the aoai, it should be https://{your-resource-name}.openai.azure.com
+  # API_KEY: "YOUR_KEY",  # The aoai API key
+  # API_VERSION: "2024-02-15-preview", # "2024-02-15-preview" by default
+  # API_MODEL: "YOUR_MODEL",  # The only OpenAI model by now that accepts visual input
+  # API_DEPLOYMENT_ID: "gpt-4-visual-preview", # The deployment id for the AOAI API
+  
+  ### For Azure_AD
+  # AAD_TENANT_ID: "YOUR_TENANT_ID", # Set the value to your tenant id for the llm model
+  # AAD_API_SCOPE: "YOUR_SCOPE", # Set the value to your scope for the llm model
+  # AAD_API_SCOPE_BASE: "YOUR_SCOPE_BASE" # Set the value to your scope base for the llm model, whose format is API://YOUR_SCOPE_BASE, and the only need is the YOUR_SCOPE_BASE
+  }
+
+
+### For parameters
+MAX_TOKENS: 2000  # The max token limit for the response completion
+MAX_RETRY: 3  # The max retry limit for the response completion
+TEMPERATURE: 0.0  # The temperature of the model: the lower the value, the more consistent the output of the model
+TOP_P: 0.0  # The top_p of the model: the lower the value, the more conservative the output of the model
+TIMEOUT: 60  # The call timeout(s), default is 10 minss
\ No newline at end of file
diff --git a/dataflow/config/config_dev.yaml b/dataflow/config/config_dev.yaml
new file mode 100644
index 00000000..ff0e73df
--- /dev/null
+++ b/dataflow/config/config_dev.yaml
@@ -0,0 +1,35 @@
+version: 0.1
+
+CONTROL_BACKEND: "uia"  # The backend for control action
+CONTROL_LIST: ["Button", "Edit", "TabItem", "Document", "ListItem", "MenuItem", "ScrollBar", "TreeItem", "Hyperlink", "ComboBox", "RadioButton", "DataItem", "Spinner"] 
+PRINT_LOG: False  # Whether to print the log  
+LOG_LEVEL: "INFO"  # The log level
+MATCH_STRATEGY: "fuzzy"  # The match strategy for the control filter, support 'contains', 'fuzzy', 'regex'
+
+PREFILL_PROMPT: "dataflow/prompts/instantiation/{mode}/prefill.yaml"  # The prompt for the action prefill
+FILTER_PROMPT: "dataflow/prompts/instantiation/{mode}/filter.yaml"  # The prompt for the filter
+PREFILL_EXAMPLE_PROMPT: "dataflow/prompts/instantiation/{mode}/prefill_example.yaml"  # The prompt for the action prefill example
+API_PROMPT: "ufo/prompts/share/lite/api.yaml"  # The prompt for the API
+
+# Default Task Configuration
+TASKS_HUB: "dataflow/tasks/prefill"  # The default tasks hub for batch dataflow
+TEMPLATE_PATH: "dataflow/templates"  # The template path for the exploration
+
+# Result Configuration
+RESULT_HUB: "dataflow/results/{task_type}"  # The result hub, task_type is 'instantiation' or 'execution'
+INSTANTIATION_RESULT_SCHEMA: "dataflow/schema/instantiation_schema.json"  # The JSON Schema for the result log
+EXECUTION_RESULT_SCHEMA: "dataflow/schema/execution_schema.json"
+
+# For control filtering
+CONTROL_FILTER_TYPE: []  # The list of control filter type, support 'TEXT', 'SEMANTIC', 'ICON'
+CONTROL_FILTER_MODEL_SEMANTIC_NAME: "all-MiniLM-L6-v2"  # The control filter model name of semantic similarity
+CONTROL_EMBEDDING_CACHE_PATH: "dataflow/cache/"  # The cache path for the control filter
+CONTROL_FILTER_TOP_K_PLAN: 2  # The control filter effect on top k plans from UFO, default is 2
+
+# log path
+LOG_PATH: "dataflow/logs/{task}"
+PREFILL_LOG_PATH: "dataflow/logs/{task}/prefill/"
+FILTER_LOG_PATH: "dataflow/logs/{task}/filter/"
+EXECUTE_LOG_PATH: "dataflow/logs/{task}/execute/"
+
+MAX_STEPS: 30  # The max step for the execute_flow
diff --git a/dataflow/data_flow_controller.py b/dataflow/data_flow_controller.py
new file mode 100644
index 00000000..16ec679c
--- /dev/null
+++ b/dataflow/data_flow_controller.py
@@ -0,0 +1,389 @@
+import os
+import time
+import traceback
+from enum import Enum
+from typing import Any, Dict, Optional, List
+from jsonschema import validate, ValidationError
+
+from dataflow.env.env_manager import WindowsAppEnv
+from dataflow.instantiation.workflow.choose_template_flow import ChooseTemplateFlow
+from dataflow.instantiation.workflow.prefill_flow import PrefillFlow
+from dataflow.instantiation.workflow.filter_flow import FilterFlow
+from dataflow.execution.workflow.execute_flow import ExecuteFlow
+from dataflow.config.config import Config
+
+from ufo.utils import print_with_color
+from learner.utils import load_json_file, save_json_file
+
+from ufo.agents.processors.app_agent_processor import AppAgentProcessor
+from ufo.module.context import Context
+
+# Set the environment variable for the run configuration.
+os.environ["RUN_CONFIGS"] = "True"
+
+# Load configuration data.
+_configs = Config.get_instance().config_data
+
+INSTANTIATION_RESULT_MAP = {
+    True: "instantiation_pass",
+    False: "instantiation_fail"
+}
+
+EXECUTION_RESULT_MAP = {
+    "yes": "execution_pass",
+    "no": "execution_fail",
+    "unsure": "execution_unsure"
+}
+
+class AppEnum(Enum):
+    """
+    Enum class for applications.
+    """
+
+    WORD = 1, "Word", ".docx", "winword"
+    EXCEL = 2, "Excel", ".xlsx", "excel"
+    POWERPOINT = 3, "PowerPoint", ".pptx", "powerpnt"
+
+    def __init__(self, id: int, description: str, file_extension: str, win_app: str):
+        """
+        Initialize the application enum.
+        :param id: The ID of the application.
+        :param description: The description of the application.
+        :param file_extension: The file extension of the application.
+        :param win_app: The Windows application name.
+        """
+
+        self.id = id
+        self.description = description
+        self.file_extension = file_extension
+        self.win_app = win_app
+        self.app_root_name = win_app.upper() + ".EXE"
+
+
+class TaskObject:
+    def __init__(self, task_file_path: str, task_type: str) -> None:
+        """
+        Initialize the task object.
+        :param task_file_path: The path to the task file.
+        :param task_type: The task_type of the task object (dataflow, instantiation, or execution).
+        """
+
+        self.task_file_path = task_file_path
+        self.task_file_base_name = os.path.basename(task_file_path)
+        self.task_file_name = self.task_file_base_name.split(".")[0]
+
+        task_json_file = load_json_file(task_file_path)
+        self.app_object = self._choose_app_from_json(task_json_file["app"])
+        # Initialize the task attributes based on the task_type
+        self._init_attr(task_type, task_json_file)
+
+    def _choose_app_from_json(self, task_app: str) -> AppEnum:
+        """
+        Choose the app from the task json file.
+        :param task_app: The app from the task json file.
+        :return: The app enum.
+        """
+
+        for app in AppEnum:
+            if app.description.lower() == task_app.lower():
+                return app
+        raise ValueError("Not a correct App") 
+    
+    def _init_attr(self, task_type:str, task_json_file:Dict[str, Any]) -> None:
+        """
+        Initialize the attributes of the task object.
+        :param task_type: The task_type of the task object (dataflow, instantiation, or execution).
+        :param task_json_file: The task JSON file.
+        """
+
+        if task_type == "dataflow" or task_type == "instantiation":
+            for key, value in task_json_file.items():
+                setattr(self, key.lower().replace(" ", "_"), value)
+        elif task_type == "execution":
+            self.app = task_json_file.get("app")
+            self.unique_id = task_json_file.get("unique_id")
+            original = task_json_file.get("original", {})
+            self.task = original.get("original_task", None) 
+            self.refined_steps = original.get("original_steps", None)
+        else:
+            raise ValueError(f"Unsupported task_type: {task_type}")
+
+class DataFlowController:
+    """
+    Flow controller class to manage the instantiation and execution process.
+    """
+
+    def __init__(self, task_path: str, task_type: str) -> None:
+        """
+        Initialize the flow controller.
+        :param task_path: The path to the task file.
+        :param task_type: The task_type of the flow controller (instantiation, execution, or dataflow).
+        """
+
+        self.task_object = TaskObject(task_path, task_type)
+        self.app_env = None
+        self.app_name = self.task_object.app_object.description.lower()
+        self.task_file_name = self.task_object.task_file_name
+
+        self.schema = self._load_schema(task_type)
+        
+        self.task_type = task_type
+        self.task_info = self.init_task_info()
+        self.result_hub = _configs["RESULT_HUB"].format(task_type=task_type)
+
+    def init_task_info(self) -> Dict[str, Any]: 
+        """
+        Initialize the task information.
+        :return: The initialized task information.
+        """
+        init_task_info = None
+        if self.task_type == "execution":
+            # read from the instantiated task file
+            init_task_info = load_json_file(self.task_object.task_file_path)
+        else:
+            init_task_info = {
+                "unique_id": self.task_object.unique_id,
+                "app": self.app_name,
+                "original": {
+                    "original_task": self.task_object.task,
+                    "original_steps": self.task_object.refined_steps,
+                },
+                "execution_result": {"result": None, "error": None},
+                "instantiation_result": {
+                    "choose_template": {"result": None, "error": None},
+                    "prefill": {"result": None, "error": None},
+                    "instantiation_evaluation": {"result": None, "error": None},
+                },
+                "time_cost": {},
+            }
+        return init_task_info
+
+    def _load_schema(self, task_type: str) -> Dict[str, Any]:
+        """
+        load the schema based on the task_type.
+        :param task_type: The task_type of the task object (dataflow, instantiation, or execution).
+        :return: The schema for the task_type.
+        """
+
+        if task_type == "instantiation":
+            return load_json_file(_configs["INSTANTIATION_RESULT_SCHEMA"])
+        elif task_type == "execution" or task_type == "dataflow":
+            return load_json_file(_configs["EXECUTION_RESULT_SCHEMA"])
+
+    def execute_instantiation(self) -> Optional[List[Dict[str, Any]]]:
+        """
+        Execute the instantiation process.
+        :return: The instantiation plan if successful.
+        """
+
+        print_with_color(f"Instantiating task {self.task_object.task_file_name}...", "blue")
+
+        template_copied_path = self.instantiation_single_flow(
+            ChooseTemplateFlow, "choose_template", 
+            init_params=[self.task_object.app_object.file_extension],
+            execute_params=[]
+        )
+
+        if template_copied_path:
+            self.app_env.start(template_copied_path)
+
+            prefill_result = self.instantiation_single_flow(
+                PrefillFlow, "prefill", 
+                init_params=[self.app_env],
+                execute_params=[template_copied_path, self.task_object.task, self.task_object.refined_steps]
+            )
+            self.app_env.close()
+
+            if prefill_result:
+                self.instantiation_single_flow(
+                    FilterFlow, "instantiation_evaluation",
+                    init_params=[],
+                    execute_params=[prefill_result["instantiated_request"]]
+                )
+                return prefill_result["instantiated_plan"]
+
+    def execute_execution(self, request: str, plan: Dict[str, any]) -> None:
+        """
+        Execute the execution process.
+        :param request: The task request to be executed.
+        :param plan: The execution plan containing detailed steps.
+        """
+
+        print_with_color("Executing the execution process...", "blue")
+        execute_flow = None
+
+        try:
+            self.app_env.start(self.template_copied_path)
+            # Initialize the execution context and flow
+            context = Context()
+            execute_flow = ExecuteFlow(self.task_file_name, context, self.app_env)
+            
+            # Execute the plan
+            executed_plan, execute_result = execute_flow.execute(request, plan)
+            
+            # Update the instantiated plan
+            self.instantiated_plan = executed_plan
+            # Record execution results and time metrics
+            self.task_info["execution_result"]["result"] = execute_result
+            self.task_info["time_cost"]["execute"] = execute_flow.execution_time
+            self.task_info["time_cost"]["execute_eval"] = execute_flow.eval_time
+
+        except Exception as e:
+            # Handle and log any exceptions that occur during execution
+            self.task_info["execution_result"]["error"] = {
+                "type": str(type(e).__name__),
+                "message": str(e),
+                "traceback": traceback.format_exc(),
+            }
+            print_with_color(f"Error in Execution: {e}", "red")
+            raise e
+        finally:
+            # Record the total time cost of the execution process
+            if execute_flow and hasattr(execute_flow, "execution_time"):
+                self.task_info["time_cost"]["execute"] = execute_flow.execution_time
+            else:
+                self.task_info["time_cost"]["execute"] = None
+            if execute_flow and hasattr(execute_flow, "eval_time"):
+                self.task_info["time_cost"]["execute_eval"] = execute_flow.eval_time
+            else:
+                self.task_info["time_cost"]["execute_eval"] = None
+
+
+    def instantiation_single_flow(
+            self, 
+            flow_class: AppAgentProcessor, 
+            flow_type: str, 
+            init_params=None, 
+            execute_params=None
+        ) -> Optional[Dict[str, Any]]:
+        """
+        Execute a single flow process in the instantiation phase.
+        :param flow_class: The flow class to instantiate.
+        :param flow_type: The type of the flow.
+        :param init_params: The initialization parameters for the flow.
+        :param execute_params: The execution parameters for the flow.
+        :return: The result of the flow process.
+        """
+        
+        flow_instance = None
+        try:
+            flow_instance = flow_class(self.app_name, self.task_file_name, *init_params)
+            result = flow_instance.execute(*execute_params)
+            self.task_info["instantiation_result"][flow_type]["result"] = result
+            return result
+        except Exception as e:
+            self.task_info["instantiation_result"][flow_type]["error"] = {
+                "type": str(e.__class__),
+                "error_message": str(e),
+                "traceback": traceback.format_exc(),
+            }
+            print_with_color(f"Error in {flow_type}: {e} {traceback.format_exc()}")
+        finally:
+            if flow_instance and hasattr(flow_instance, "execution_time"):
+                self.task_info["time_cost"][flow_type] = flow_instance.execution_time
+            else:
+                self.task_info["time_cost"][flow_type] = None
+    
+    def save_result(self) -> None:
+        """
+        Validate and save the instantiated task result.
+        """
+
+        validation_error = None
+
+        # Validate the result against the schema
+        try:
+            validate(instance=self.task_info, schema=self.schema)
+        except ValidationError as e:
+            # Record the validation error but allow the process to continue
+            validation_error = str(e.message)
+            print_with_color(f"Validation Error: {e.message}", "yellow")
+
+        # Determine the target directory based on task_type and quality/completeness
+        target_file = None
+
+        if self.task_type == "instantiation":
+            # Determine the quality of the instantiation
+            if not self.task_info["instantiation_result"]["instantiation_evaluation"]["result"]:
+                target_file = INSTANTIATION_RESULT_MAP[False]
+            else:
+                is_quality_good = self.task_info["instantiation_result"]["instantiation_evaluation"]["result"]["judge"]
+                target_file = INSTANTIATION_RESULT_MAP.get(is_quality_good, INSTANTIATION_RESULT_MAP[False])
+
+        else:
+            # Determine the completion status of the execution
+            if not self.task_info["execution_result"]["result"]:
+                target_file = EXECUTION_RESULT_MAP["no"]
+            else:
+                is_completed = self.task_info["execution_result"]["result"]["complete"]
+                target_file = EXECUTION_RESULT_MAP.get(is_completed, EXECUTION_RESULT_MAP["no"])
+
+        # Construct the full path to save the result
+        new_task_path = os.path.join(self.result_hub, target_file, self.task_object.task_file_base_name)
+        os.makedirs(os.path.dirname(new_task_path), exist_ok=True)
+        save_json_file(new_task_path, self.task_info)
+
+        print(f"Task saved to {new_task_path}")
+
+        # If validation failed, indicate that the saved result may need further inspection
+        if validation_error:
+            print("The saved task result does not conform to the expected schema and may require review.")
+    
+    @property
+    def template_copied_path(self) -> str:
+        """
+        Get the copied template path from the task information.
+        :return: The copied template path.
+        """
+
+        return self.task_info["instantiation_result"]["choose_template"]["result"]
+    
+    @property
+    def instantiated_plan(self) -> List[Dict[str, Any]]:
+        """
+        Get the instantiated plan from the task information.
+        :return: The instantiated plan.
+        """
+
+        return self.task_info["instantiation_result"]["prefill"]["result"]["instantiated_plan"]
+
+    @instantiated_plan.setter
+    def instantiated_plan(self, value: List[Dict[str, Any]]) -> None:
+        """
+        Set the instantiated plan in the task information.
+        :param value: New value for the instantiated plan.
+        """
+
+        self.task_info.setdefault("instantiation_result", {}).setdefault("prefill", {}).setdefault("result", {})
+        self.task_info["instantiation_result"]["prefill"]["result"]["instantiated_plan"] = value
+
+    def run(self) -> None:
+        """
+        Run the instantiation and execution process.
+        """
+
+        start_time = time.time()
+
+        try:
+            self.app_env = WindowsAppEnv(self.task_object.app_object)
+
+            if self.task_type == "dataflow":
+                plan = self.execute_instantiation()
+                self.execute_execution(self.task_object.task, plan)
+            elif self.task_type == "instantiation":
+                self.execute_instantiation()
+            elif self.task_type == "execution":
+                plan = self.instantiated_plan
+                self.execute_execution(self.task_object.task, plan)
+            else:
+                raise ValueError(f"Unsupported task_type: {self.task_type}")
+        except Exception as e:
+            raise e
+
+        finally:
+            # Update or record the total time cost of the process
+            total_time = round(time.time() - start_time, 3)
+            new_total_time = self.task_info.get("time_cost", {}).get("total", 0) + total_time
+            self.task_info["time_cost"]["total"] = round(new_total_time, 3)
+
+            self.save_result()
\ No newline at end of file
diff --git a/dataflow/dataflow.py b/dataflow/dataflow.py
new file mode 100644
index 00000000..54a24548
--- /dev/null
+++ b/dataflow/dataflow.py
@@ -0,0 +1,117 @@
+import argparse
+import os
+import traceback
+from ufo.utils import print_with_color
+from dataflow.config.config import Config
+
+_configs = Config.get_instance().config_data
+
+
+def parse_args() -> argparse.Namespace:
+    """
+    Parse command-line arguments. Automatically detect batch or single mode.
+    """
+    parser = argparse.ArgumentParser(description="Run tasks automatically in single or batch mode.")
+    
+    # Add options for -dataflow, -instantiation, and -execution
+    parser.add_argument(
+        "-dataflow", 
+        action="store_const", const="dataflow", 
+        help="Indicates that the task type is dataflow."
+    )
+    parser.add_argument(
+        "-instantiation", 
+        action="store_const", const="instantiation", 
+        help="Indicates that the task type is instantiation."
+    )
+    parser.add_argument(
+        "-execution", 
+        action="store_const", const="execution", 
+        help="Indicates that the task type is execution."
+    )
+
+    # Task path argument
+    parser.add_argument(
+        "--task_path",
+        type=str,
+        default=_configs["TASKS_HUB"],
+        help="Path to the task file or directory.",
+    )
+
+    return parser.parse_args()
+
+
+def validate_path(path: str) -> str:
+    """
+    Validate the given path and determine its type.
+    :param path: The path to validate.
+    :return: "file", "directory", or raises an error if invalid.
+    """
+    if os.path.isfile(path):
+        return "file"
+    elif os.path.isdir(path):
+        return "directory"
+    else:
+        print_with_color(f"Invalid path: {path}", "red")
+        raise ValueError(f"Path {path} is neither a file nor a directory.")
+
+
+def process_task(task_path: str, task_type: str) -> None:
+    """
+    Process a single task file using the DataFlowController.
+    """
+    from dataflow.data_flow_controller import DataFlowController
+
+    try:
+        print_with_color(f"Processing task: {task_path}", "green")
+        flow_controller = DataFlowController(task_path, task_type)
+        flow_controller.run()
+        print_with_color(f"Task {task_path} completed successfully.", "green")
+    except Exception as e:
+        print_with_color(f"Error processing {task_path}: {traceback.format_exc()}", "red")
+
+
+def process_batch(task_dir: str, task_type: str) -> None:
+    """
+    Process all task files in a directory.
+    """
+    task_files = [
+        os.path.join(task_dir, f)
+        for f in os.listdir(task_dir)
+        if os.path.isfile(os.path.join(task_dir, f))
+    ]
+    if not task_files:
+        print_with_color(f"No tasks found in directory: {task_dir}.", "yellow")
+        return
+
+    print_with_color(f"Found {len(task_files)} tasks in {task_dir}.", "blue")
+    for task_file in task_files:
+        process_task(task_file, task_type)
+
+
+def main():
+    """
+    Main function to run tasks based on the provided arguments.
+    You can use dataflow, instantiation, and execution modes to process the task.
+    Also, you can run tasks in batch mode by providing the path to the task directory.
+    See README to read the detailed usage.
+    """
+    args = parse_args()
+
+    # Ensure that a task type has been provided; if not, raise an error
+    if not any([args.dataflow, args.instantiation, args.execution]):
+        print_with_color("Error: You must specify one of the task types (-dataflow, -instantiation, or -execution).", "red")
+        return
+
+    task_type = args.dataflow or args.instantiation or args.execution
+
+    path_type = validate_path(args.task_path)
+
+    if path_type == "file":
+        process_task(args.task_path, task_type)
+    elif path_type == "directory":
+        process_batch(args.task_path, task_type)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/dataflow/env/env_manager.py b/dataflow/env/env_manager.py
new file mode 100644
index 00000000..d1ade291
--- /dev/null
+++ b/dataflow/env/env_manager.py
@@ -0,0 +1,178 @@
+import logging
+import re
+from time import sleep
+from typing import Optional, Tuple, Dict
+import psutil
+
+from fuzzywuzzy import fuzz
+from pywinauto import Desktop
+from pywinauto.controls.uiawrapper import UIAWrapper
+
+from dataflow.config.config import Config
+from ufo.config.config import Config as UFOConfig
+
+# Load configuration settings
+_configs = Config.get_instance().config_data
+_ufo_configs = UFOConfig.get_instance().config_data
+
+if _ufo_configs is not None:
+    _BACKEND = _ufo_configs["CONTROL_BACKEND"]
+if _configs is not None:
+    _MATCH_STRATEGY = _configs.get("MATCH_STRATEGY", "contains")
+
+
+class WindowsAppEnv:
+    """
+    Represents the Windows Application Environment.
+    """
+
+    def __init__(self, app_object: object) -> None:
+        """
+        Initializes the Windows Application Environment.
+        :param app_object: The app object containing information about the application.
+        """
+
+        self.app_window = None
+        self.app_root_name = app_object.app_root_name
+        self.app_name = app_object.description.lower()
+        self.win_app = app_object.win_app
+
+    def start(self, copied_template_path: str) -> None:
+        """
+        Starts the Windows environment.
+        :param copied_template_path: The file path to the copied template to start the environment.
+        """
+
+        from ufo.automator.ui_control import openfile
+
+        file_controller = openfile.FileController(_BACKEND)
+        try:
+            file_controller.execute_code(
+                {"APP": self.win_app, "file_path": copied_template_path}
+            )
+        except Exception as e:
+            logging.exception(f"Failed to start the application: {e}")
+            raise
+
+    def close(self) -> None:
+        """
+        Tries to gracefully close the application; if it fails or is not closed, forcefully terminates the process.
+        """
+
+        try:
+            # Attempt to close gracefully
+            if self.app_window:
+                self.app_window.close()
+            
+            self._check_and_kill_process()
+            sleep(1)
+        except Exception as e:
+            logging.warning(f"Graceful close failed: {e}. Attempting to forcefully terminate the process.")
+            self._check_and_kill_process()
+            raise e
+    
+    def _check_and_kill_process(self) -> None:
+        """
+        Checks if the process is still running and kills it if it is.
+        """
+
+        try:
+            # Ensure the app_window object is still valid and visible
+            if self.app_window and not self.app_window.is_visible():
+                process = psutil.Process(self.app_window.process_id)
+                if process.is_running():
+                    print(f"Killing process: {self.app_window.process_id}")
+                    process.terminate()
+        except Exception as e:
+            logging.error(f"Error while checking window status: {e}")
+            raise e
+
+    def find_matching_window(self, doc_name: str) -> Optional[UIAWrapper]:
+        """
+        Finds a matching window based on the process name and the configured matching strategy.
+        :param doc_name: The document name associated with the application.
+        :return: The matched window or None if no match is found.
+        """
+
+        desktop = Desktop(backend=_BACKEND)
+        windows_list = desktop.windows()
+        for window in windows_list:
+            window_title = window.element_info.name.lower()
+            if self._match_window_name(window_title, doc_name):
+                self.app_window = window
+                return window
+        return None
+
+    def _match_window_name(self, window_title: str, doc_name: str) -> bool:
+        """
+        Matches the window name based on the strategy specified in the config file.
+        :param window_title: The title of the window.
+        :param doc_name: The document name associated with the application.
+        :return: True if a match is found based on the strategy; False otherwise.
+        """
+
+        app_name = self.app_name
+        doc_name = doc_name.lower()
+
+        if _MATCH_STRATEGY == "contains":
+            return app_name in window_title and doc_name in window_title
+        elif _MATCH_STRATEGY == "fuzzy":
+            similarity_app = fuzz.partial_ratio(window_title, app_name)
+            similarity_doc = fuzz.partial_ratio(window_title, doc_name)
+            return similarity_app >= 70 and similarity_doc >= 70
+        elif _MATCH_STRATEGY == "regex":
+            combined_name_1 = f"{app_name}.*{doc_name}"
+            combined_name_2 = f"{doc_name}.*{app_name}"
+            pattern_1 = re.compile(combined_name_1, flags=re.IGNORECASE)
+            pattern_2 = re.compile(combined_name_2, flags=re.IGNORECASE)
+            return (
+                re.search(pattern_1, window_title) is not None
+                or re.search(pattern_2, window_title) is not None
+            )
+        else:
+            logging.exception(f"Unknown match strategy: {_MATCH_STRATEGY}")
+            raise ValueError(f"Unknown match strategy: {_MATCH_STRATEGY}")
+
+    def _calculate_match_score(self, control, control_text) -> int:
+        """
+        Calculate the match score between a control and the given text.
+        :param control: The control object to evaluate.
+        :param control_text: The target text to match.
+        :return: An integer score representing the match quality (higher is better).
+        """
+        control_content = control.window_text() or ""
+
+        # Matching strategies
+        if _MATCH_STRATEGY == "contains":
+            return 100 if control_text in control_content else 0
+        elif _MATCH_STRATEGY == "fuzzy":
+            return fuzz.partial_ratio(control_content, control_text)
+        elif _MATCH_STRATEGY == "regex":
+            pattern = re.compile(f"{re.escape(control_text)}", flags=re.IGNORECASE)
+            return 100 if re.search(pattern, control_content) else 0
+        else:
+            raise ValueError(f"Unknown match strategy: {_MATCH_STRATEGY}")
+    
+    def find_matching_controller(self, filtered_annotation_dict: Dict[int, UIAWrapper], control_text: str) -> Tuple[str, UIAWrapper]:
+        """"
+        Select the best matched controller.
+        :param filtered_annotation_dict: The filtered annotation dictionary.
+        :param control_text: The text content of the control for additional context.
+        :return: Tuple containing the key of the selected controller and the control object.s
+        """
+        control_selected = None
+        controller_key = None
+        highest_score = 0
+
+        # Iterate through the filtered annotation dictionary to find the best match
+        for key, control in filtered_annotation_dict.items():
+            # Calculate the matching score using the match function
+            score = self._calculate_match_score(control, control_text)
+
+            # Update the selected control if the score is higher
+            if score > highest_score:
+                highest_score = score
+                controller_key = key
+                control_selected = control
+
+        return controller_key, control_selected
diff --git a/dataflow/execution/agent/execute_agent.py b/dataflow/execution/agent/execute_agent.py
new file mode 100644
index 00000000..00e522fc
--- /dev/null
+++ b/dataflow/execution/agent/execute_agent.py
@@ -0,0 +1,31 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from ufo.agents.agent.app_agent import AppAgent
+
+
+class ExecuteAgent(AppAgent):
+    """
+    The Agent for task execution.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        process_name: str,
+        app_root_name: str,
+    ):
+        """
+        Initialize the ExecuteAgent.
+        :param name: The name of the agent.
+        :param process_name: The name of the process.
+        :param app_root_name: The name of the app root.
+        """
+
+        self._step = 0
+        self._complete = False
+        self._name = name
+        self._status = None
+        self._process_name = process_name
+        self._app_root_name = app_root_name
+        self.Puppeteer = self.create_puppeteer_interface()
\ No newline at end of file
diff --git a/dataflow/execution/agent/execute_eval_agent.py b/dataflow/execution/agent/execute_eval_agent.py
new file mode 100644
index 00000000..0044ad16
--- /dev/null
+++ b/dataflow/execution/agent/execute_eval_agent.py
@@ -0,0 +1,66 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from typing import Optional
+
+from dataflow.prompter.execution.execute_eval_prompter import ExecuteEvalAgentPrompter
+from ufo.agents.agent.evaluation_agent import EvaluationAgent
+
+class ExecuteEvalAgent(EvaluationAgent):
+    """
+    The Agent for task execution evaluation.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        app_root_name: str,
+        is_visual: bool,
+        main_prompt: str,
+        example_prompt: str,
+        api_prompt: str,
+    ):
+        """
+        Initialize the ExecuteEvalAgent.
+        :param name: The name of the agent.
+        :param app_root_name: The name of the app root.
+        :param is_visual: The flag indicating whether the agent is visual or not.
+        :param main_prompt: The main prompt.
+        :param example_prompt: The example prompt.
+        :param api_prompt: The API prompt.
+        """
+
+        super().__init__(
+            name=name,
+            app_root_name=app_root_name,
+            is_visual=is_visual,
+            main_prompt=main_prompt,
+            example_prompt=example_prompt,
+            api_prompt=api_prompt,
+        )
+
+    def get_prompter(
+        self,
+        is_visual: bool,
+        prompt_template: str,
+        example_prompt_template: str,
+        api_prompt_template: str,
+        root_name: Optional[str] = None,
+    ) -> ExecuteEvalAgentPrompter:
+        """
+        Get the prompter for the agent.
+        :param is_visual: The flag indicating whether the agent is visual or not.
+        :param prompt_template: The prompt template.
+        :param example_prompt_template: The example prompt template.
+        :param api_prompt_template: The API prompt template.
+        :param root_name: The name of the root.
+        :return: The prompter.
+        """
+
+        return ExecuteEvalAgentPrompter(
+            is_visual=is_visual,
+            prompt_template=prompt_template,
+            example_prompt_template=example_prompt_template,
+            api_prompt_template=api_prompt_template,
+            root_name=root_name,
+        )
\ No newline at end of file
diff --git a/dataflow/execution/workflow/execute_flow.py b/dataflow/execution/workflow/execute_flow.py
new file mode 100644
index 00000000..cf3928e1
--- /dev/null
+++ b/dataflow/execution/workflow/execute_flow.py
@@ -0,0 +1,365 @@
+import os
+import time
+from typing import Any, Dict, List, Tuple
+
+from dataflow.config.config import Config as InstantiationConfig
+from dataflow.env.env_manager import WindowsAppEnv
+from dataflow.execution.agent.execute_agent import ExecuteAgent
+from dataflow.execution.agent.execute_eval_agent import ExecuteEvalAgent
+from ufo import utils
+from ufo.agents.processors.app_agent_processor import AppAgentProcessor
+from ufo.automator.app_apis.basic import WinCOMReceiverBasic
+from ufo.config.config import Config as UFOConfig
+from ufo.module.basic import BaseSession, Context, ContextNames
+from ufo.automator.ui_control.screenshot import PhotographerDecorator
+
+_configs = InstantiationConfig.get_instance().config_data
+_ufo_configs = UFOConfig.get_instance().config_data
+if _ufo_configs is not None:
+    BACKEND = _ufo_configs["CONTROL_BACKEND"]
+
+
+class ExecuteFlow(AppAgentProcessor):
+    """
+    ExecuteFlow class for executing the task and saving the result.
+    """
+
+    _app_execute_agent_dict: Dict[str, ExecuteAgent] = {}
+    _app_eval_agent_dict: Dict[str, ExecuteEvalAgent] = {}
+
+    def __init__(
+        self, task_file_name: str, context: Context, environment: WindowsAppEnv
+    ) -> None:
+        """
+        Initialize the execute flow for a task.
+        :param task_file_name: Name of the task file being processed.
+        :param context: Context object for the current session.
+        :param environment: Environment object for the application being processed.
+        """
+
+        super().__init__(agent=ExecuteAgent, context=context)
+
+        self.execution_time = None
+        self.eval_time = None
+        self._app_env = environment
+        self._task_file_name = task_file_name
+        self._app_name = self._app_env.app_name
+
+        log_path = _configs["EXECUTE_LOG_PATH"].format(task=task_file_name)
+        self._initialize_logs(log_path)
+
+        self.application_window = self._app_env.find_matching_window(task_file_name)
+        self.app_agent = self._get_or_create_execute_agent()
+        self.eval_agent = self._get_or_create_evaluation_agent()
+
+        self._matched_control = None    # Matched control for the current step.
+
+    def _get_or_create_execute_agent(self) -> ExecuteAgent:
+        """
+        Retrieve or create a execute agent for the given application.
+        :return: ExecuteAgent instance for the specified application.
+        """
+
+        if self._app_name not in ExecuteFlow._app_execute_agent_dict:
+            ExecuteFlow._app_execute_agent_dict[self._app_name] = ExecuteAgent(
+                "execute",
+                self._app_name,
+                self._app_env.app_root_name,
+            )
+        return ExecuteFlow._app_execute_agent_dict[self._app_name]
+
+    def _get_or_create_evaluation_agent(self) -> ExecuteEvalAgent:
+        """
+        Retrieve or create an evaluation agent for the given application.
+        :return: ExecuteEvalAgent instance for the specified application.
+        """
+
+        if self._app_name not in ExecuteFlow._app_eval_agent_dict:
+            ExecuteFlow._app_eval_agent_dict[self._app_name] = ExecuteEvalAgent(
+                "evaluation",
+                self._app_env.app_root_name,
+                is_visual=True,
+                main_prompt=_ufo_configs["EVALUATION_PROMPT"],
+                example_prompt="",
+                api_prompt=_ufo_configs["API_PROMPT"],
+            )
+        return ExecuteFlow._app_eval_agent_dict[self._app_name]
+
+    def _initialize_logs(self, log_path: str) -> None:
+        """
+        Initialize logging for execute messages and responses.
+        :param log_path: Path to save the logs.
+        """
+
+        os.makedirs(log_path, exist_ok=True)
+        self._execute_message_logger = BaseSession.initialize_logger(
+            log_path, "execute_log.json", "w", _configs
+        )
+        self.context.set(ContextNames.LOG_PATH, log_path)
+        self.context.set(ContextNames.LOGGER, self._execute_message_logger)
+
+    def execute(
+        self, request: str, instantiated_plan: List[Dict[str, Any]]
+    ) -> Tuple[List[Dict[str, Any]], Dict[str, str]]:
+        """
+        Execute the execute flow: Execute the task and save the result.
+        :param request: Original request to be executed.
+        :param instantiated_plan: Instantiated plan containing steps to execute.
+        :return: Tuple containing task quality flag, comment, and task type.
+        """
+        
+        start_time = time.time()
+        try:
+            executed_plan = self.execute_plan(instantiated_plan)
+        except Exception as error:
+            raise RuntimeError(f"Execution failed. {error}")
+        finally:
+            self.execution_time = round(time.time() - start_time, 3)
+
+        start_time = time.time()
+        try:
+            result, _ = self.eval_agent.evaluate(request=request, log_path=self.log_path)
+            utils.print_with_color(f"Result: {result}", "green")
+        except Exception as error:
+            raise RuntimeError(f"Evaluation failed. {error}")
+        finally:
+            self.eval_time = round(time.time() - start_time, 3)
+        
+        return executed_plan, result
+
+    def execute_plan(
+        self, instantiated_plan: List[Dict[str, Any]]
+    ) -> List[Dict[str, Any]]:
+        """
+        Get the executed result from the execute agent.
+        :param instantiated_plan: Plan containing steps to execute.
+        :return: List of executed steps.
+        """
+
+        # Initialize the step counter and capture the initial screenshot.
+        self.session_step = 0
+        try:
+            time.sleep(1)
+            # Initialize the API receiver
+            self.app_agent.Puppeteer.receiver_manager.create_api_receiver(
+                self.app_agent._app_root_name, self.app_agent._process_name
+            )
+            # Initialize the control receiver
+            current_receiver = self.app_agent.Puppeteer.receiver_manager.receiver_list[-1]
+
+            if current_receiver is not None:
+                self.application_window = self._app_env.find_matching_window(self._task_file_name)
+                current_receiver.com_object = current_receiver.get_object_from_process_name()
+
+            self.init_and_final_capture_screenshot()
+        except Exception as error:
+            raise RuntimeError(f"Execution initialization failed. {error}")
+        
+        # Initialize the success flag for each step.
+        for index, step_plan in enumerate(instantiated_plan):
+            instantiated_plan[index]["Success"] = None
+            instantiated_plan[index]["MatchedControlText"] = None
+
+        for index, step_plan in enumerate(instantiated_plan):
+            try:
+                self.session_step += 1
+
+                # Check if the maximum steps have been exceeded.
+                if self.session_step > _configs["MAX_STEPS"]:
+                    raise RuntimeError("Maximum steps exceeded.")
+
+                self._parse_step_plan(step_plan)
+
+                try:
+                    self.process()
+                    instantiated_plan[index]["Success"] = True
+                    instantiated_plan[index]["ControlLabel"] = self._control_label
+                    instantiated_plan[index]["MatchedControlText"] = self._matched_control
+                except Exception as ControllerNotFoundError:
+                    instantiated_plan[index]["Success"] = False
+                    raise ControllerNotFoundError
+
+            except Exception as error:
+                err_info = RuntimeError(
+                    f"Step {self.session_step} execution failed. {error}"
+                )
+                raise err_info
+        # capture the final screenshot
+        self.session_step += 1
+        time.sleep(1)
+        self.init_and_final_capture_screenshot()
+        # save the final state of the app
+
+        win_com_receiver = None
+        for receiver in reversed(self.app_agent.Puppeteer.receiver_manager.receiver_list):
+            if isinstance(receiver, WinCOMReceiverBasic):
+                if receiver.client is not None:
+                    win_com_receiver = receiver
+                    break
+
+        if win_com_receiver is not None:
+            win_com_receiver.save()
+            time.sleep(1)
+            win_com_receiver.client.Quit()
+
+        print("Execution complete.")
+
+        return instantiated_plan
+
+    def process(self) -> None:
+        """
+        Process the current step.
+        """
+
+        step_start_time = time.time()
+        self.print_step_info()
+        self.capture_screenshot()
+        self.execute_action()
+        self.time_cost = round(time.time() - step_start_time, 3)
+        self.log_save()
+
+    def print_step_info(self) -> None:
+        """
+        Print the step information.
+        """
+
+        utils.print_with_color(
+            "Step {step}: {subtask}".format(
+                step=self.session_step,
+                subtask=self.subtask,
+            ),
+            "magenta",
+        )
+
+    def log_save(self) -> None:
+        """
+        Log the constructed prompt message for the PrefillAgent.
+        """
+
+        step_memory = {
+            "Step": self.session_step,
+            "Subtask": self.subtask,
+            "ControlLabel": self._control_label,
+            "ControlText": self.control_text,
+            "Action": self.action,
+            "ActionType": self.app_agent.Puppeteer.get_command_types(self._operation),
+            "Results": self._results,
+            "Application": self.app_agent._app_root_name,
+            "TimeCost": self.time_cost,
+        }
+        self._memory_data.set_values_from_dict(step_memory)
+        self.log(self._memory_data.to_dict())
+
+    def _parse_step_plan(self, step_plan: Dict[str, Any]) -> None:
+        """
+        Parse the response.
+        :param step_plan: The step plan.
+        """
+
+        self._matched_control = None
+        self.subtask = step_plan.get("Subtask", "")
+        self.control_text = step_plan.get("ControlText", "")
+        self._operation = step_plan.get("Function", "")
+        self.question_list = step_plan.get("Questions", [])
+        self._args = utils.revise_line_breaks(step_plan.get("Args", ""))
+
+        # Compose the function call and the arguments string.
+        self.action = self.app_agent.Puppeteer.get_command_string(
+            self._operation, self._args
+        )
+
+        self.status = step_plan.get("Status", "")
+
+
+    def init_and_final_capture_screenshot(self) -> None:
+        """
+        Capture the screenshot.
+        """
+
+        # Define the paths for the screenshots saved.
+        screenshot_save_path = self.log_path + f"action_step{self.session_step}.png"
+
+        self._memory_data.set_values_from_dict(
+            {
+                "CleanScreenshot": screenshot_save_path,
+            }
+        )
+
+        self.photographer.capture_app_window_screenshot(
+            self.application_window, save_path=screenshot_save_path
+        )
+        # Capture the control screenshot.
+        control_selected = self._app_env.app_window
+        self.capture_control_screenshot(control_selected)
+    
+    def execute_action(self) -> None:
+        """
+        Execute the action.
+        """
+        
+        control_selected = None
+        # Find the matching window and control.
+        self.application_window = self._app_env.find_matching_window(self._task_file_name)
+        if self.control_text == "":
+            control_selected =  self.application_window 
+        else:
+            self._control_label, control_selected = self._app_env.find_matching_controller(
+                self.filtered_annotation_dict, self.control_text
+                )
+            self._matched_control = control_selected.window_text()
+
+        if not control_selected:
+            # If the control is not found, raise an error.
+            raise RuntimeError(f"Control with text '{self.control_text}' not found.")
+
+        try:
+            # Get the selected control item from the annotation dictionary and LLM response.
+            # The LLM response is a number index corresponding to the key in the annotation dictionary.
+            if control_selected:
+
+                if _ufo_configs.get("SHOW_VISUAL_OUTLINE_ON_SCREEN", True):
+                    control_selected.draw_outline(colour="red", thickness=3)
+                    time.sleep(_ufo_configs.get("RECTANGLE_TIME", 0))
+
+                control_coordinates = PhotographerDecorator.coordinate_adjusted(
+                    self.application_window.rectangle(), control_selected.rectangle()
+                )
+
+                self._control_log = {
+                    "control_class": control_selected.element_info.class_name,
+                    "control_type": control_selected.element_info.control_type,
+                    "control_automation_id": control_selected.element_info.automation_id,
+                    "control_friendly_class_name": control_selected.friendly_class_name(),
+                    "control_coordinates": {
+                        "left": control_coordinates[0],
+                        "top": control_coordinates[1],
+                        "right": control_coordinates[2],
+                        "bottom": control_coordinates[3],
+                    },
+                }
+
+                self.app_agent.Puppeteer.receiver_manager.create_ui_control_receiver(
+                    control_selected, self.application_window
+                )
+
+                # Save the screenshot of the tagged selected control.
+                self.capture_control_screenshot(control_selected)
+
+                self._results = self.app_agent.Puppeteer.execute_command(
+                    self._operation, self._args
+                )
+                self.control_reannotate = None
+                if not utils.is_json_serializable(self._results):
+                    self._results = ""
+
+                    return
+
+        except Exception:
+            self.general_error_handler()
+
+    def general_error_handler(self) -> None:
+        """
+        Handle general errors.
+        """
+
+        pass
diff --git a/dataflow/instantiation/agent/filter_agent.py b/dataflow/instantiation/agent/filter_agent.py
new file mode 100644
index 00000000..f8649ec0
--- /dev/null
+++ b/dataflow/instantiation/agent/filter_agent.py
@@ -0,0 +1,86 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from typing import  List, Optional
+
+from dataflow.prompter.instantiation.filter_prompter import FilterPrompter
+from ufo.agents.agent.basic import BasicAgent
+
+class FilterAgent(BasicAgent):
+    """
+    The Agent to evaluate the instantiated task is correct or not.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        process_name: str,
+        is_visual: bool,
+        main_prompt: str,
+        example_prompt: str,
+        api_prompt: str,
+    ):
+        """
+        Initialize the FilterAgent.
+        :param name: The name of the agent.
+        :param process_name: The name of the process.
+        :param is_visual: The flag indicating whether the agent is visual or not.
+        :param main_prompt: The main prompt.
+        :param example_prompt: The example prompt.
+        :param api_prompt: The API prompt.
+        """
+
+        self._step = 0
+        self._complete = False
+        self._name = name
+        self._status = None
+        self.prompter: FilterPrompter = self.get_prompter(
+            is_visual, main_prompt, example_prompt, api_prompt
+        )
+        self._process_name = process_name
+
+    def get_prompter(
+        self,
+        is_visual: bool,
+        main_prompt: str,
+        example_prompt: str,
+        api_prompt: str
+    ) -> FilterPrompter:
+        """
+        Get the prompt for the agent.
+        :param is_visual: The flag indicating whether the agent is visual or not.
+        :param main_prompt: The main prompt.
+        :param example_prompt: The example prompt.
+        :param api_prompt: The API prompt.
+        :return: The prompt string.
+        """
+
+        return FilterPrompter(is_visual, main_prompt, example_prompt, api_prompt)
+
+    def message_constructor(self, request: str, app: str) -> List[str]:
+        """
+        Construct the prompt message for the FilterAgent.
+        :param request: The request sentence.
+        :param app: The name of the operated app.
+        :return: The prompt message.
+        """
+
+        filter_agent_prompt_system_message = self.prompter.system_prompt_construction(
+            app=app
+        )
+        filter_agent_prompt_user_message = self.prompter.user_content_construction(
+            request
+        )
+        filter_agent_prompt_message = self.prompter.prompt_construction(
+            filter_agent_prompt_system_message, filter_agent_prompt_user_message
+        )
+
+        return filter_agent_prompt_message
+
+    def process_comfirmation(self) -> None:
+        """
+        Confirm the process.
+        This is the abstract method from BasicAgent that needs to be implemented.
+        """
+
+        pass
\ No newline at end of file
diff --git a/dataflow/instantiation/agent/prefill_agent.py b/dataflow/instantiation/agent/prefill_agent.py
new file mode 100644
index 00000000..139244f2
--- /dev/null
+++ b/dataflow/instantiation/agent/prefill_agent.py
@@ -0,0 +1,94 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from typing import Dict, List
+
+from dataflow.prompter.instantiation.prefill_prompter import PrefillPrompter
+
+from ufo.agents.agent.basic import BasicAgent
+
+
+class PrefillAgent(BasicAgent):
+    """
+    The Agent for task instantialization and action sequence generation.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        process_name: str,
+        is_visual: bool,
+        main_prompt: str,
+        example_prompt: str,
+        api_prompt: str,
+    ):
+        """
+        Initialize the PrefillAgent.
+        :param name: The name of the agent.
+        :param process_name: The name of the process.
+        :param is_visual: The flag indicating whether the agent is visual or not.
+        :param main_prompt: The main prompt.
+        :param example_prompt: The example prompt.
+        :param api_prompt: The API prompt.
+        """
+
+        self._step = 0
+        self._complete = False
+        self._name = name
+        self._status = None
+        self.prompter: PrefillPrompter = self.get_prompter(
+            is_visual, main_prompt, example_prompt, api_prompt
+        )
+        self._process_name = process_name
+
+    def get_prompter(self, is_visual: bool, main_prompt: str, example_prompt: str, api_prompt: str) -> str:
+        """
+        Get the prompt for the agent.
+        This is the abstract method from BasicAgent that needs to be implemented.
+        :param is_visual: The flag indicating whether the agent is visual or not.
+        :param main_prompt: The main prompt.
+        :param example_prompt: The example prompt.
+        :param api_prompt: The API prompt.
+        :return: The prompt string.
+        """
+
+        return PrefillPrompter(is_visual, main_prompt, example_prompt, api_prompt)
+
+    def message_constructor(
+        self,
+        dynamic_examples: str,
+        given_task: str,
+        reference_steps: List[str],
+        doc_control_state: Dict[str, str],
+        log_path: str,
+    ) -> List[str]:
+        """
+        Construct the prompt message for the PrefillAgent.
+        :param dynamic_examples: The dynamic examples retrieved from the self-demonstration and human demonstration.
+        :param given_task: The given task.
+        :param reference_steps: The reference steps.
+        :param doc_control_state: The document control state.
+        :param log_path: The path of the log.
+        :return: The prompt message.
+        """
+
+        prefill_agent_prompt_system_message = self.prompter.system_prompt_construction(
+            dynamic_examples
+        )
+        prefill_agent_prompt_user_message = self.prompter.user_content_construction(
+            given_task, reference_steps, doc_control_state, log_path
+        )
+        appagent_prompt_message = self.prompter.prompt_construction(
+            prefill_agent_prompt_system_message,
+            prefill_agent_prompt_user_message,
+        )
+
+        return appagent_prompt_message
+
+    def process_comfirmation(self) -> None:
+        """
+        Confirm the process.
+        This is the abstract method from BasicAgent that needs to be implemented.
+        """
+
+        pass
\ No newline at end of file
diff --git a/dataflow/instantiation/workflow/choose_template_flow.py b/dataflow/instantiation/workflow/choose_template_flow.py
new file mode 100644
index 00000000..639e5e73
--- /dev/null
+++ b/dataflow/instantiation/workflow/choose_template_flow.py
@@ -0,0 +1,188 @@
+import json
+import os
+import random
+import time
+import warnings
+from datetime import datetime
+from pathlib import Path
+from typing import Dict
+
+from langchain.embeddings import CacheBackedEmbeddings
+from langchain.storage import LocalFileStore
+from langchain_community.embeddings import HuggingFaceEmbeddings
+from langchain_community.vectorstores import FAISS
+
+from dataflow.config.config import Config
+
+_configs = Config.get_instance().config_data
+
+
+class ChooseTemplateFlow:
+    """
+    Class to select and copy the most relevant template file based on the given task context.
+    """
+
+    _SENTENCE_TRANSFORMERS_PREFIX = "sentence-transformers/"
+
+    def __init__(self, app_name: str, task_file_name: str, file_extension: str):
+        """
+        Initialize the flow with the given task context.
+        :param app_name: The name of the application.
+        :param file_extension: The file extension of the template.
+        :param task_file_name: The name of the task file.
+        """
+
+        self._app_name = app_name
+        self._file_extension = file_extension
+        self._task_file_name = task_file_name
+        self.execution_time = None
+        self._embedding_model = self._load_embedding_model(
+            model_name=_configs["CONTROL_FILTER_MODEL_SEMANTIC_NAME"]
+        )
+
+    def execute(self) -> str:
+        """
+        Execute the flow and return the copied template path.
+        :return: The path to the copied template file.
+        """
+
+        start_time = time.time()
+        try:
+            template_copied_path = self._choose_template_and_copy()
+        except Exception as e:
+            raise e
+        finally:
+            self.execution_time = round(time.time() - start_time, 3)
+        return template_copied_path
+
+    def _create_copied_file(
+        self, copy_from_path: Path, copy_to_folder_path: Path, file_name: str = None
+    ) -> str:
+        """
+        Create a cache file from the specified source.
+        :param copy_from_path: The original path of the file.
+        :param copy_to_folder_path: The path where the cache file will be created.
+        :param file_name: Optional; the name of the task file.
+        :return: The path to the newly created cache file.
+        """
+
+        os.makedirs(copy_to_folder_path, exist_ok=True)
+        copied_template_path = self._generate_copied_file_path(
+            copy_to_folder_path, file_name
+        )
+
+        with open(copy_from_path, "rb") as f:
+            ori_content = f.read()
+        with open(copied_template_path, "wb") as f:
+            f.write(ori_content)
+
+        return copied_template_path
+
+    def _generate_copied_file_path(self, folder_path: Path, file_name: str) -> str:
+        """
+        Generate the file path for the copied template.
+        :param folder_path: The folder where the file will be created.
+        :param file_name: Optional; the name of the task file.
+        :return: The path to the newly created file.
+        """
+
+        template_extension = self._file_extension
+        if file_name:
+            return str(folder_path / f"{file_name}{template_extension}")
+        timestamp = datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
+        return str(folder_path / f"{timestamp}{template_extension}")
+
+    def _get_chosen_file_path(self) -> str:
+        """
+        Choose the most relevant template file based on the task.
+        :return: The path to the most relevant template file.
+        """
+
+        templates_description_path = (
+            Path(_configs["TEMPLATE_PATH"]) / self._app_name / "description.json"
+        )
+
+        try:
+            with open(templates_description_path, "r") as f:
+                return self._choose_target_template_file(
+                    self._task_file_name, json.load(f)
+                )
+        except FileNotFoundError:
+            warnings.warn(
+                f"Warning: {templates_description_path} does not exist. Choosing a random template."
+            )
+            return self._choose_random_template()
+
+    def _choose_random_template(self) -> str:
+        """
+        Select a random template file from the template folder.
+        :return: The path to the randomly selected template file.
+        """
+
+        template_folder = Path(_configs["TEMPLATE_PATH"]) / self._app_name
+        template_files = [f for f in template_folder.iterdir() if f.is_file()]
+
+        if not template_files:
+            raise Exception("No template files found in the specified directory.")
+
+        chosen_template_file = random.choice(template_files)
+        print(f"Randomly selected template: {chosen_template_file.name}")
+        return str(chosen_template_file)
+
+    def _choose_template_and_copy(self) -> str:
+        """
+        Choose the template and copy it to the cache folder.
+        :return: The path to the copied template file.
+        """
+
+        chosen_template_file_path = self._get_chosen_file_path()
+        chosen_template_full_path = (
+            Path(_configs["TEMPLATE_PATH"]) / self._app_name / chosen_template_file_path
+        )
+
+        target_template_folder_path = Path(_configs["RESULT_HUB"].format(task_type = "saved_document")) / (
+            os.path.dirname(os.path.dirname(self._task_file_name))
+        )
+
+        return self._create_copied_file(
+            chosen_template_full_path, target_template_folder_path, self._task_file_name
+        )
+
+    def _choose_target_template_file(
+        self, given_task: str, doc_files_description: Dict[str, str]
+    ) -> str:
+        """
+        Get the target file based on the semantic similarity of the given task and the template file descriptions.
+        :param given_task: The task to be matched.
+        :param doc_files_description: A dictionary of template file descriptions.
+        :return: The path to the chosen template file.
+        """
+
+        file_doc_map = {
+            desc: file_name for file_name, desc in doc_files_description.items()
+        }
+        db = FAISS.from_texts(
+            list(doc_files_description.values()), self._embedding_model
+        )
+        most_similar = db.similarity_search(given_task, k=1)
+
+        if not most_similar:
+            raise ValueError("No similar templates found.")
+
+        return file_doc_map[most_similar[0].page_content]
+
+    @staticmethod
+    def _load_embedding_model(model_name: str) -> CacheBackedEmbeddings:
+        """
+        Load the embedding model.
+        :param model_name: The name of the embedding model to load.
+        :return: The loaded embedding model.
+        """
+        
+        store = LocalFileStore(_configs["CONTROL_EMBEDDING_CACHE_PATH"])
+        if not model_name.startswith(ChooseTemplateFlow._SENTENCE_TRANSFORMERS_PREFIX):
+            model_name = ChooseTemplateFlow._SENTENCE_TRANSFORMERS_PREFIX + model_name
+        embedding_model = HuggingFaceEmbeddings(model_name=model_name)
+        return CacheBackedEmbeddings.from_bytes_store(
+            embedding_model, store, namespace=model_name
+        )
\ No newline at end of file
diff --git a/dataflow/instantiation/workflow/filter_flow.py b/dataflow/instantiation/workflow/filter_flow.py
new file mode 100644
index 00000000..7aa55d1d
--- /dev/null
+++ b/dataflow/instantiation/workflow/filter_flow.py
@@ -0,0 +1,143 @@
+import json
+import logging
+import os
+import time
+from typing import Dict, Tuple, Any
+
+from dataflow.config.config import Config
+from dataflow.instantiation.agent.filter_agent import FilterAgent
+from ufo.module.basic import BaseSession
+
+_configs = Config.get_instance().config_data
+
+
+class FilterFlow:
+    """
+    Class to refine the plan steps and prefill the file based on filtering criteria.
+    """
+
+    _app_filter_agent_dict: Dict[str, FilterAgent] = {}
+
+    def __init__(self, app_name: str, task_file_name: str) -> None:
+        """
+        Initialize the filter flow for a task.
+        :param app_name: Name of the application being processed.
+        :param task_file_name: Name of the task file being processed.
+        """
+
+        self.execution_time = None
+        self._app_name = app_name
+        self._log_path_configs = _configs["FILTER_LOG_PATH"].format(task=task_file_name)
+        self._filter_agent = self._get_or_create_filter_agent()
+        self._initialize_logs()
+
+    def _get_or_create_filter_agent(self) -> FilterAgent:
+        """
+        Retrieve or create a filter agent for the given application.
+        :return: FilterAgent instance for the specified application.
+        """
+
+        if self._app_name not in FilterFlow._app_filter_agent_dict:
+            FilterFlow._app_filter_agent_dict[self._app_name] = FilterAgent(
+                "filter",
+                self._app_name,
+                is_visual=True,
+                main_prompt=_configs["FILTER_PROMPT"],
+                example_prompt="",
+                api_prompt=_configs["API_PROMPT"],
+            ) 
+        return FilterFlow._app_filter_agent_dict[self._app_name]
+
+    def execute(self, instantiated_request: str) -> Dict[str, Any]:
+        """
+        Execute the filter flow: Filter the task and save the result.
+        :param instantiated_request: Request object to be filtered.
+        :return: Tuple containing task quality flag, comment, and task type.
+        """
+
+        start_time = time.time()
+        try:
+            judge, thought, request_type = self._get_filtered_result(
+                instantiated_request
+            )
+        except Exception as e:
+            raise e
+        finally:
+            self.execution_time = round(time.time() - start_time, 3)
+        return {
+            "judge": judge,
+            "thought": thought,
+            "request_type": request_type,
+        }
+    
+    def _initialize_logs(self) -> None:
+        """
+        Initialize logging for filter messages and responses.
+        """
+
+        os.makedirs(self._log_path_configs, exist_ok=True)
+        self._filter_message_logger = BaseSession.initialize_logger(
+            self._log_path_configs, "filter_messages.json", "w", _configs
+        )
+        self._filter_response_logger = BaseSession.initialize_logger(
+            self._log_path_configs, "filter_responses.json", "w", _configs
+        )
+
+    def _get_filtered_result(self, instantiated_request: str) -> Tuple[bool, str, str]:
+        """
+        Get the filtered result from the filter agent.
+        :param instantiated_request: Request object containing task details.
+        :return: Tuple containing task quality flag, request comment, and request type.
+        """
+
+        # Construct the prompt message for the filter agent
+        prompt_message = self._filter_agent.message_constructor(
+            instantiated_request,
+            self._app_name,
+        )
+        prompt_json = json.dumps(prompt_message, indent=4)
+        self._filter_message_logger.info(prompt_json)
+
+        # Get the response from the filter agent
+        try:
+            start_time = time.time()
+            response_string, _ = self._filter_agent.get_response(
+                prompt_message, "filter", use_backup_engine=True, configs=_configs
+            )
+            try:
+                fixed_response_string = self._fix_json_commas(response_string)
+                response_json = self._filter_agent.response_to_dict(
+                    fixed_response_string
+                )
+            except json.JSONDecodeError as e:
+                logging.error(
+                    f"JSONDecodeError: {e.msg} at position {e.pos}. Response: {response_string}"
+                )
+                raise e
+
+            execution_time = round(time.time() - start_time, 3)
+
+            response_json["execution_time"] = execution_time
+            self._filter_response_logger.info(json.dumps(response_json, indent=4))
+
+            return (
+                response_json["judge"],
+                response_json["thought"],
+                response_json["type"],
+            )
+        except Exception as e:
+            logging.error(f"Error occurred while filtering: {e}")
+            raise e
+
+    def _fix_json_commas(self, json_string: str) -> str:
+        """
+        Function to add missing commas between key-value pairs in a JSON string
+        and remove newline characters for proper formatting.
+        :param json_string: The JSON string to be fixed.
+        :return: The fixed JSON string.
+        """
+
+        # Remove newline characters
+        json_string = json_string.replace("\n", "")
+
+        return json_string
\ No newline at end of file
diff --git a/dataflow/instantiation/workflow/prefill_flow.py b/dataflow/instantiation/workflow/prefill_flow.py
new file mode 100644
index 00000000..6ff05ba9
--- /dev/null
+++ b/dataflow/instantiation/workflow/prefill_flow.py
@@ -0,0 +1,279 @@
+import json
+import logging
+import os
+import time
+from typing import Any, Dict, List, Tuple
+
+from dataflow.config.config import Config
+from dataflow.instantiation.agent.prefill_agent import PrefillAgent
+from dataflow.env.env_manager import WindowsAppEnv
+from ufo.agents.processors.app_agent_processor import AppAgentProcessor
+from ufo.automator.ui_control.inspector import ControlInspectorFacade
+from ufo.automator.ui_control.screenshot import PhotographerFacade
+from ufo.module.basic import BaseSession
+from ufo.config.config import Config as UFOConfig
+
+_configs = Config.get_instance().config_data
+_ufo_configs = UFOConfig.get_instance().config_data
+if _ufo_configs is not None:
+    _BACKEND = _ufo_configs["CONTROL_BACKEND"]
+
+
+class PrefillFlow(AppAgentProcessor):
+    """
+    Class to manage the prefill process by refining planning steps and automating UI interactions
+    """
+
+    _app_prefill_agent_dict: Dict[str, PrefillAgent] = {}
+
+    def __init__(
+        self,
+        app_name: str,
+        task_file_name: str,
+        environment: WindowsAppEnv,
+    ) -> None:
+        """
+        Initialize the prefill flow with the application context.
+        :param app_name: The name of the application.
+        :param task_file_name: The name of the task file for logging and tracking.
+        :param environment: The environment of the app.
+        """
+
+        self.execution_time = None
+        self._app_name = app_name
+        self._task_file_name = task_file_name
+        self._app_env = environment
+        # Create or reuse a PrefillAgent for the app
+        if self._app_name not in PrefillFlow._app_prefill_agent_dict:
+            PrefillFlow._app_prefill_agent_dict[self._app_name] = PrefillAgent(
+                "prefill",
+                self._app_name,
+                is_visual=True,
+                main_prompt=_configs["PREFILL_PROMPT"],
+                example_prompt=_configs["PREFILL_EXAMPLE_PROMPT"],
+                api_prompt=_configs["API_PROMPT"],
+            )
+        self._prefill_agent = PrefillFlow._app_prefill_agent_dict[self._app_name]
+
+        # Initialize execution step and UI control tools
+        self._execute_step = 0
+        self._control_inspector = ControlInspectorFacade(_BACKEND)
+        self._photographer = PhotographerFacade()
+
+        # Set default states
+        self._status = ""
+
+        # Initialize loggers for messages and responses
+        self._log_path_configs = _configs["PREFILL_LOG_PATH"].format(
+            task=self._task_file_name
+        )
+        os.makedirs(self._log_path_configs, exist_ok=True)
+
+        # Set up loggers
+        self._message_logger = BaseSession.initialize_logger(
+            self._log_path_configs, "prefill_messages.json", "w", _configs
+        )
+        self._response_logger = BaseSession.initialize_logger(
+            self._log_path_configs, "prefill_responses.json", "w", _configs
+        )
+
+    def execute(
+        self, template_copied_path: str, original_task: str, refined_steps: List[str]
+    ) -> Dict[str, Any]:
+        """
+        Start the execution by retrieving the instantiated result.
+        :param template_copied_path: The path of the copied template to use.
+        :param original_task: The original task to refine.
+        :param refined_steps: The steps to guide the refinement process.
+        :return: The refined task and corresponding action plans.
+        """
+
+        start_time = time.time()
+        try:
+            instantiated_request, instantiated_plan = self._instantiate_task(
+                template_copied_path, original_task, refined_steps
+            )
+        except Exception as e:
+            raise e
+        finally:
+            self.execution_time = round(time.time() - start_time, 3)
+
+        return  {
+            "instantiated_request": instantiated_request,
+            "instantiated_plan": instantiated_plan,
+        }   
+
+    def _instantiate_task(
+        self, template_copied_path: str, original_task: str, refined_steps: List[str]
+    ) -> Tuple[str, List[str]]:
+        """
+        Retrieve and process the instantiated result for the task.
+        Interacts with the PrefillAgent to refine the task and generate action plans.
+        :param template_copied_path: The path of the copied template to use.
+        :param original_task: The original task to refine.
+        :param refined_steps: The steps to guide the refinement process.
+        :return: The refined task and corresponding action plans.
+        """
+
+        try:
+            # Retrieve prefill actions and task plan
+            instantiated_request, instantiated_plan = self._get_prefill_actions(
+                original_task,
+                refined_steps,
+                template_copied_path,
+            )
+
+            print(f"Original Task: {original_task}")
+            print(f"Prefilled Task: {instantiated_request}")
+
+        except Exception as e:
+            logging.exception(f"Error in prefilling task: {e}")
+            raise e
+
+        return instantiated_request, instantiated_plan
+    
+    def _update_state(self, file_path: str) -> None:
+        """
+        Update the current state of the app by inspecting UI elements.
+        :param file_path: Path of the app file to inspect.
+        """
+
+        print(f"Updating the app state using the file: {file_path}")
+
+        # Retrieve control elements in the app window
+        control_list = self._control_inspector.find_control_elements_in_descendants(
+            self._app_env.app_window,
+            control_type_list=_ufo_configs["CONTROL_LIST"],
+            class_name_list=_ufo_configs["CONTROL_LIST"],
+        )
+
+        # Capture UI control annotations
+        self._annotation_dict = self._photographer.get_annotation_dict(
+            self._app_env.app_window, control_list, annotation_type="number"
+        )
+
+        # Filter out irrelevant control elements
+        self._filtered_annotation_dict = self.get_filtered_annotation_dict(
+            self._annotation_dict, configs=_configs
+        )
+
+        # Gather control info for both full and filtered lists
+        self._control_info = self._control_inspector.get_control_info_list_of_dict(
+            self._annotation_dict,
+            ["control_text", "control_type" if _BACKEND == "uia" else "control_class"],
+        )
+        self._filtered_control_info = (
+            self._control_inspector.get_control_info_list_of_dict(
+                self._filtered_annotation_dict,
+                [
+                    "control_text",
+                    "control_type" if _BACKEND == "uia" else "control_class",
+                ],
+            )
+        )
+
+    def _get_prefill_actions(
+        self, given_task: str, reference_steps: List[str], file_path: str
+    ) -> Tuple[str, List[str]]:
+        """
+        Generate refined tasks and action plans using the PrefillAgent.
+        :param given_task: The task to refine.
+        :param reference_steps: Reference steps for the task.
+        :param file_path: Path to the task template.
+        :return: The refined task and corresponding action plans.
+        """
+
+        self._update_state(file_path)
+        execution_time = 0
+        # Save a screenshot of the app state
+        screenshot_path = os.path.join(self._log_path_configs, "screenshot.png")
+        self._save_screenshot(self._task_file_name, screenshot_path)
+
+        # Construct prompt message for the PrefillAgent
+        prompt_message = self._prefill_agent.message_constructor(
+            "",
+            given_task,
+            reference_steps,
+            self._filtered_control_info,
+            self._log_path_configs,
+        )
+
+        # Log the constructed message
+        self._log_message(prompt_message)
+
+        try:
+            # Record start time and get PrefillAgent response
+            start_time = time.time()
+            response_string, _ = self._prefill_agent.get_response(
+                prompt_message, "prefill", use_backup_engine=True, configs=_configs
+            )
+            execution_time = round(time.time() - start_time, 3)
+
+            # Parse and log the response
+            response_json = self._prefill_agent.response_to_dict(response_string)
+            instantiated_request = response_json["New_task"]
+            instantiated_plan = response_json["Actions_plan"]
+
+        except Exception as e:
+            self._status = "ERROR"
+            logging.exception(f"Error in prefilling task: {e}")
+            raise e
+        finally:
+            # Log the response and execution time
+            self._log_response(response_json, execution_time)
+
+        return instantiated_request, instantiated_plan
+
+    def _log_message(self, prompt_message: str) -> None:
+        """
+        Log the constructed prompt message for the PrefillAgent.
+        :param prompt_message: The message constructed for PrefillAgent.
+        """
+
+        messages_log_entry = {
+            "step": self._execute_step,
+            "messages": prompt_message,
+            "error": "",
+        }
+        self._message_logger.info(json.dumps(messages_log_entry, indent=4))
+
+    def _log_response(
+        self, response_json: Dict[str, Any], execution_time: float
+    ) -> None:
+        """
+        Log the response received from PrefillAgent along with execution time.
+        :param response_json: Response data from PrefillAgent.
+        :param execution_time: Time taken for the PrefillAgent call.
+        """
+
+        response_log_entry = {
+            "step": self._execute_step,
+            "execution_time": execution_time,
+            "agent_response": response_json,
+            "error": "",
+        }
+        self._response_logger.info(json.dumps(response_log_entry, indent=4))
+
+    def _save_screenshot(self, doc_name: str, save_path: str) -> None:
+        """
+        Captures a screenshot of the current window or the full screen if the window is not found.
+        :param doc_name: The name or description of the document to match the window.
+        :param save_path: The path where the screenshot will be saved.
+        """
+        
+        try:
+            # Find the window matching the document name
+            matched_window = self._app_env.find_matching_window(doc_name)
+            if matched_window:
+                screenshot = self._photographer.capture_app_window_screenshot(
+                    matched_window
+                )
+            else:
+                logging.warning("Window not found, taking a full-screen screenshot.")
+                screenshot = self._photographer.capture_desktop_screen_screenshot()
+
+            screenshot.save(save_path)
+            print(f"Screenshot saved to {save_path}")
+        except Exception as e:
+            logging.exception(f"Failed to save screenshot: {e}")
+            raise e
diff --git a/dataflow/prompter/execution/execute_eval_prompter.py b/dataflow/prompter/execution/execute_eval_prompter.py
new file mode 100644
index 00000000..2d35ce09
--- /dev/null
+++ b/dataflow/prompter/execution/execute_eval_prompter.py
@@ -0,0 +1,52 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import json
+import os
+from typing import Dict, List, Optional
+
+from ufo.prompter.basic import BasicPrompter
+from ufo.prompter.eva_prompter import EvaluationAgentPrompter
+
+class ExecuteEvalAgentPrompter(EvaluationAgentPrompter):
+    """
+    Execute the prompt for the ExecuteEvalAgent.
+    """
+
+    def __init__(
+        self,
+        is_visual: bool,
+        prompt_template: str,
+        example_prompt_template: str,
+        api_prompt_template: str,
+        root_name: Optional[str] = None,
+    ):
+        """
+        Initialize the CustomEvaluationAgentPrompter.
+        :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.
+        :param root_name: The name of the root application.
+        """
+
+        super().__init__(
+            is_visual,
+            prompt_template,
+            example_prompt_template,
+            api_prompt_template,
+            root_name,
+        )
+
+    @staticmethod
+    def load_logs(log_path: str) -> List[Dict[str, str]]:
+        """
+        Load logs from the log path.
+        :param log_path: The path of the log.
+        """
+
+        log_file_path = os.path.join(log_path, "execute_log.json")
+        with open(log_file_path, "r") as f:
+            logs = f.readlines()
+            logs = [json.loads(log) for log in logs]
+        return logs
diff --git a/dataflow/prompter/instantiation/filter_prompter.py b/dataflow/prompter/instantiation/filter_prompter.py
new file mode 100644
index 00000000..cec5658d
--- /dev/null
+++ b/dataflow/prompter/instantiation/filter_prompter.py
@@ -0,0 +1,162 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import json
+import os
+from typing import Dict, List, Optional
+
+from ufo.prompter.basic import BasicPrompter
+from ufo.prompter.eva_prompter import EvaluationAgentPrompter
+
+
+class FilterPrompter(BasicPrompter):
+    """
+    Load the prompt for the FilterAgent.
+    """
+
+    def __init__(
+        self,
+        is_visual: bool,
+        prompt_template: str,
+        example_prompt_template: str,
+        api_prompt_template: str,
+    ):
+        """
+        Initialize the FilterPrompter.
+        :param is_visual: The flag indicating whether the prompter is visual or not.
+        :param prompt_template: The prompt template.
+        :param example_prompt_template: The example prompt template.
+        :param api_prompt_template: The API prompt template.
+        """
+
+        super().__init__(is_visual, prompt_template, example_prompt_template)
+        self.api_prompt_template = self.load_prompt_template(
+            api_prompt_template, is_visual
+        )
+
+    def api_prompt_helper(self, apis: Dict = {}, verbose: int = 1) -> str:
+        """
+        Construct the prompt for APIs.
+        :param apis: The APIs.
+        :param verbose: The verbosity level.
+        :return: The prompt for APIs.
+        """
+
+        # Construct the prompt for APIs
+        if len(apis) == 0:
+            api_list = [
+                "- The action type are limited to {actions}.".format(
+                    actions=list(self.api_prompt_template.keys())
+                )
+            ]
+
+            # Construct the prompt for each API
+            for key in self.api_prompt_template.keys():
+                api = self.api_prompt_template[key]
+                if verbose > 0:
+                    api_text = "{summary}\n{usage}".format(
+                        summary=api["summary"], usage=api["usage"]
+                    )
+                else:
+                    api_text = api["summary"]
+
+                api_list.append(api_text)
+
+            api_prompt = self.retrived_documents_prompt_helper("", "", api_list)
+        else:
+            api_list = [
+                "- The action type are limited to {actions}.".format(
+                    actions=list(apis.keys())
+                )
+            ]
+
+            # Construct the prompt for each API
+            for key in apis.keys():
+                api = apis[key]
+                api_text = "{description}\n{example}".format(
+                    description=api["description"], example=api["example"]
+                )
+                api_list.append(api_text)
+
+            api_prompt = self.retrived_documents_prompt_helper("", "", api_list)
+
+        return api_prompt
+
+    def system_prompt_construction(self, app: str = "") -> str:
+        """
+        Construct the prompt for the system.
+        :param app: The app name.
+        :return: The prompt for the system.
+        """
+
+        try:
+            ans = self.prompt_template["system"]
+            ans = ans.format(app=app)
+            return ans
+        except Exception as e:
+            print(e)
+
+    def user_prompt_construction(self, request: str) -> str:
+        """
+        Construct the prompt for the user.
+        :param request: The user request.
+        :return: The prompt for the user.
+        """
+
+        prompt = self.prompt_template["user"].format(request=request)
+        return prompt
+
+    def user_content_construction(self, request: str) -> List[Dict]:
+        """
+        Construct the prompt for LLMs.
+        :param request: The user request.
+        :return: The prompt for LLMs.
+        """
+
+        user_content = []
+
+        user_content.append(
+            {"type": "text", "text": self.user_prompt_construction(request)}
+        )
+
+        return user_content
+
+    def examples_prompt_helper(
+        self,
+        header: str = "## Response Examples",
+        separator: str = "Example",
+        additional_examples: List[str] = [],
+    ) -> str:
+        """
+        Construct the prompt for examples.
+        :param header: The header of the prompt.
+        :param separator: The separator of the prompt.
+        :param additional_examples: The additional examples.
+        :return: The prompt for examples.
+        """
+
+        template = """
+        [User Request]:
+            {request}
+        [Response]:
+            {response}
+        [Tip]
+            {tip}
+        """
+
+        example_list = []
+
+        for key in self.example_prompt_template.keys():
+            if key.startswith("example"):
+                example = template.format(
+                    request=self.example_prompt_template[key].get("Request"),
+                    response=json.dumps(
+                        self.example_prompt_template[key].get("Response")
+                    ),
+                    tip=self.example_prompt_template[key].get("Tips", ""),
+                )
+                example_list.append(example)
+
+        example_list += [json.dumps(example) for example in additional_examples]
+
+        return self.retrived_documents_prompt_helper(header, separator, example_list)
diff --git a/dataflow/prompter/instantiation/prefill_prompter.py b/dataflow/prompter/instantiation/prefill_prompter.py
new file mode 100644
index 00000000..5f979553
--- /dev/null
+++ b/dataflow/prompter/instantiation/prefill_prompter.py
@@ -0,0 +1,337 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import json
+import os
+from typing import Dict, List, Optional
+
+from ufo.prompter.basic import BasicPrompter
+from ufo.prompter.eva_prompter import EvaluationAgentPrompter
+
+
+class FilterPrompter(BasicPrompter):
+    """
+    Load the prompt for the FilterAgent.
+    """
+
+    def __init__(
+        self,
+        is_visual: bool,
+        prompt_template: str,
+        example_prompt_template: str,
+        api_prompt_template: str,
+    ):
+        """
+        Initialize the FilterPrompter.
+        :param is_visual: The flag indicating whether the prompter is visual or not.
+        :param prompt_template: The prompt template.
+        :param example_prompt_template: The example prompt template.
+        :param api_prompt_template: The API prompt template.
+        """
+
+        super().__init__(is_visual, prompt_template, example_prompt_template)
+        self.api_prompt_template = self.load_prompt_template(
+            api_prompt_template, is_visual
+        )
+
+    def api_prompt_helper(self, apis: Dict = {}, verbose: int = 1) -> str:
+        """
+        Construct the prompt for APIs.
+        :param apis: The APIs.
+        :param verbose: The verbosity level.
+        :return: The prompt for APIs.
+        """
+
+        # Construct the prompt for APIs
+        if len(apis) == 0:
+            api_list = [
+                "- The action type are limited to {actions}.".format(
+                    actions=list(self.api_prompt_template.keys())
+                )
+            ]
+
+            # Construct the prompt for each API
+            for key in self.api_prompt_template.keys():
+                api = self.api_prompt_template[key]
+                if verbose > 0:
+                    api_text = "{summary}\n{usage}".format(
+                        summary=api["summary"], usage=api["usage"]
+                    )
+                else:
+                    api_text = api["summary"]
+
+                api_list.append(api_text)
+
+            api_prompt = self.retrived_documents_prompt_helper("", "", api_list)
+        else:
+            api_list = [
+                "- The action type are limited to {actions}.".format(
+                    actions=list(apis.keys())
+                )
+            ]
+
+            # Construct the prompt for each API
+            for key in apis.keys():
+                api = apis[key]
+                api_text = "{description}\n{example}".format(
+                    description=api["description"], example=api["example"]
+                )
+                api_list.append(api_text)
+
+            api_prompt = self.retrived_documents_prompt_helper("", "", api_list)
+
+        return api_prompt
+
+    def system_prompt_construction(self, app: str = "") -> str:
+        """
+        Construct the prompt for the system.
+        :param app: The app name.
+        :return: The prompt for the system.
+        """
+
+        try:
+            ans = self.prompt_template["system"]
+            ans = ans.format(app=app)
+            return ans
+        except Exception as e:
+            print(e)
+
+    def user_prompt_construction(self, request: str) -> str:
+        """
+        Construct the prompt for the user.
+        :param request: The user request.
+        :return: The prompt for the user.
+        """
+
+        prompt = self.prompt_template["user"].format(request=request)
+        return prompt
+
+    def user_content_construction(self, request: str) -> List[Dict]:
+        """
+        Construct the prompt for LLMs.
+        :param request: The user request.
+        :return: The prompt for LLMs.
+        """
+
+        user_content = []
+
+        user_content.append(
+            {"type": "text", "text": self.user_prompt_construction(request)}
+        )
+
+        return user_content
+
+    def examples_prompt_helper(
+        self,
+        header: str = "## Response Examples",
+        separator: str = "Example",
+        additional_examples: List[str] = [],
+    ) -> str:
+        """
+        Construct the prompt for examples.
+        :param header: The header of the prompt.
+        :param separator: The separator of the prompt.
+        :param additional_examples: The additional examples.
+        :return: The prompt for examples.
+        """
+
+        template = """
+        [User Request]:
+            {request}
+        [Response]:
+            {response}
+        [Tip]
+            {tip}
+        """
+
+        example_list = []
+
+        for key in self.example_prompt_template.keys():
+            if key.startswith("example"):
+                example = template.format(
+                    request=self.example_prompt_template[key].get("Request"),
+                    response=json.dumps(
+                        self.example_prompt_template[key].get("Response")
+                    ),
+                    tip=self.example_prompt_template[key].get("Tips", ""),
+                )
+                example_list.append(example)
+
+        example_list += [json.dumps(example) for example in additional_examples]
+
+        return self.retrived_documents_prompt_helper(header, separator, example_list)
+
+
+class PrefillPrompter(BasicPrompter):
+    """
+    Load the prompt for the PrefillAgent.
+    """
+
+    def __init__(
+        self,
+        is_visual: bool,
+        prompt_template: str,
+        example_prompt_template: str,
+        api_prompt_template: str,
+    ):
+        """
+        Initialize the PrefillPrompter.
+        :param is_visual: The flag indicating whether the prompter is visual or not.
+        :param prompt_template: The prompt template.
+        :param example_prompt_template: The example prompt template.
+        :param api_prompt_template: The API prompt template.
+        """
+
+        super().__init__(is_visual, prompt_template, example_prompt_template)
+        self.api_prompt_template = self.load_prompt_template(
+            api_prompt_template, is_visual
+        )
+
+    def api_prompt_helper(self, verbose: int = 1) -> str:
+        """
+        Construct the prompt for APIs.
+        :param verbose: The verbosity level.
+        :return: The prompt for APIs.
+        """
+
+        # Construct the prompt for APIs
+        api_list = [
+            "- The action type are limited to {actions}.".format(
+                actions=list(self.api_prompt_template.keys())
+            )
+        ]
+
+        # Construct the prompt for each API
+        for key in self.api_prompt_template.keys():
+            api = self.api_prompt_template[key]
+            if verbose > 0:
+                api_text = "{summary}\n{usage}".format(
+                    summary=api["summary"], usage=api["usage"]
+                )
+            else:
+                api_text = api["summary"]
+
+            api_list.append(api_text)
+
+        api_prompt = self.retrived_documents_prompt_helper("", "", api_list)
+
+        return api_prompt
+
+    def system_prompt_construction(self, additional_examples: List = []) -> str:
+        """
+        Construct the prompt for the system.
+        :param additional_examples: The additional examples.
+        :return: The prompt for the system.
+        """
+
+        examples = self.examples_prompt_helper(additional_examples=additional_examples)
+        apis = self.api_prompt_helper(verbose=0)
+        return self.prompt_template["system"].format(apis=apis, examples=examples)
+
+    def user_prompt_construction(
+        self, given_task: str, reference_steps: List, doc_control_state: Dict
+    ) -> str:
+        """
+        Construct the prompt for the user.
+        :param given_task: The given task.
+        :param reference_steps: The reference steps.
+        :param doc_control_state: The document control state.
+        :return: The prompt for the user.
+        """
+
+        prompt = self.prompt_template["user"].format(
+            given_task=given_task,
+            reference_steps=json.dumps(reference_steps),
+            doc_control_state=json.dumps(doc_control_state),
+        )
+
+        return prompt
+
+    def load_screenshots(self, log_path: str) -> str:
+        """
+        Load the first and last screenshots from the log path.
+        :param log_path: The path of the log.
+        :return: The screenshot URL.
+        """
+        
+        from ufo.prompter.eva_prompter import EvaluationAgentPrompter
+
+        init_image = os.path.join(log_path, "screenshot.png")
+        init_image_url = EvaluationAgentPrompter.load_single_screenshot(init_image)
+        return init_image_url
+
+    def user_content_construction(
+        self,
+        given_task: str,
+        reference_steps: List,
+        doc_control_state: Dict,
+        log_path: str,
+    ) -> List[Dict]:
+        """
+        Construct the prompt for LLMs.
+        :param given_task: The given task.
+        :param reference_steps: The reference steps.
+        :param doc_control_state: The document control state.
+        :param log_path: The path of the log.
+        :return: The prompt for LLMs.
+        """
+
+        user_content = []
+        if self.is_visual:
+            screenshot = self.load_screenshots(log_path)
+            screenshot_text = """You are a action prefill agent, responsible to prefill the given task.
+                                This is the screenshot of the current environment, please check it and give prefilled task accodingly."""
+
+            user_content.append({"type": "text", "text": screenshot_text})
+            user_content.append({"type": "image_url", "image_url": {"url": screenshot}})
+
+        user_content.append(
+            {
+                "type": "text",
+                "text": self.user_prompt_construction(
+                    given_task, reference_steps, doc_control_state
+                ),
+            }
+        )
+
+        return user_content
+
+    def examples_prompt_helper(
+        self,
+        header: str = "## Response Examples",
+        separator: str = "Example",
+        additional_examples: List[str] = [],
+    ) -> str:
+        """
+        Construct the prompt for examples.
+        :param header: The header of the prompt.
+        :param separator: The separator of the prompt.
+        :param additional_examples: The additional examples.
+        :return: The prompt for examples.
+        """
+
+        template = """
+        [User Request]:
+            {request}
+        [Response]:
+            {response}
+        [Tip]
+            {tip}
+        """
+
+        example_list = []
+
+        for key in self.example_prompt_template.keys():
+            if key.startswith("example"):
+                example = template.format(
+                    request=self.example_prompt_template[key].get("Request"),
+                    response=json.dumps(
+                        self.example_prompt_template[key].get("Response")
+                    ),
+                    tip=self.example_prompt_template[key].get("Tips", ""),
+                )
+                example_list.append(example)
+
+        example_list += [json.dumps(example) for example in additional_examples]
+
+        return self.retrived_documents_prompt_helper(header, separator, example_list)
\ No newline at end of file
diff --git a/dataflow/prompts/instantiation/visual/filter.yaml b/dataflow/prompts/instantiation/visual/filter.yaml
new file mode 100644
index 00000000..fd4cfd3a
--- /dev/null
+++ b/dataflow/prompts/instantiation/visual/filter.yaml
@@ -0,0 +1,26 @@
+version: 1.0
+
+system: |-
+  You are a task judge, will be provided with a task in the <Task:>. You need to judge whether this task can be executed locally.
+
+  ## Evaluation Dimension
+  The task is only related to {app}.
+  This task should be like a task, not subjective considerations. For example, if there are 'custom', 'you want' and other situations, they cannot be considered and should return false and be classified as Non_task. Any subjective will crash the system.
+  This task should specify the element, for example, if there are only 'text' without the specific string, they cannot be considered and should return false and be classified as Non_task.
+  This task should not involve interactions with other application plug-ins, etc., and only rely on Word. If 'Excel', 'Edge' and other interactions are involved, it should return false and be classified as App_involve.
+  This task should not involve version updates and other interactions that depend on the environment, but only rely on the current version, and do not want to be upgraded or downgraded. It should return false and be classified as Env.
+  There are other things that you think cannot be executed or are irrelevant, return False, and be classified as Others
+  
+  ## Response Format
+  Your response should be strictly structured in a JSON format, consisting of three distinct parts with the following keys and corresponding content:
+  {{
+  "judge": true or false depends on you think this task whether can be performed,
+  "thought": "Outline the reason why you give the judgement.",
+  "type": "None/Non_task/App_involve/Env/Others"
+  }}
+  Make sure you answer must be strictly in JSON format only, without other redundant text such as json header. Otherwise it will crash the system.
+  Below is only a example of the response. Do not fall in the example.
+
+user: |-
+  <Task:>{request}
+  <Your response:>
\ No newline at end of file
diff --git a/dataflow/prompts/instantiation/visual/prefill.yaml b/dataflow/prompts/instantiation/visual/prefill.yaml
new file mode 100644
index 00000000..3f310eaa
--- /dev/null
+++ b/dataflow/prompts/instantiation/visual/prefill.yaml
@@ -0,0 +1,128 @@
+version: 1.0
+
+system: |-
+  You are a Agent Task Creator and planer.
+  You will receive a <Given Task> that is abstract and your objective is to instantiate this task, and give the step-by-step actions to take.
+  - You are provided with a doc file environment, which contains the control information in <Doc Control State:>.
+  - You should review the doc canvas content and control information to detail the <Given Task> to a <New Task>.The control information is in a dict tree of available control items format.
+  - You are provided with <Available Actions>, you should review the acions carefully and choose the most suitable ones step-by-step <Action Plan>.
+  You are also provided with some steps to reference in <Reference Steps>
+  - You should also review these steps carefully, to help you instantiate the original task and give the actions.
+  
+
+  ## Control item
+  - The control item is the element on the page that you can interact with, we limit the actionable control item to the following:
+  - "Button" is the control item that you can click.
+  - "Edit" is the control item that you can click and input text.
+  - "TabItem" is the control item that you can click and switch to another page.
+  - "ListItem" is the control item that you can click and select.
+  - "MenuItem" is the control item that you can click and select.
+  - "ScrollBar" is the control item that you can scroll.
+  - "TreeItem" is the control item that you can click and select.
+  - "Document" is the control item that you can click and select text.
+  - "Hyperlink" is the control item that you can click and open a link.
+  - "ComboBox" is the control item that you can click and input text. The Google search box is an example of ComboBox.
+
+  ## Available Actions on the control item
+  - All the available actions are listed below:
+  {apis}
+
+  Besides, please prefill the task based on the screenshot. you will also be provided with a screenshot, one before the agent's execution and one after the agent's execution.
+
+  ## The requirements for <New Task>
+  1. The <New Task> must based on the given task, but if more then one options exist in <Given Task>, you must choose one of them.
+  2. The <New Task> must be able to be completed step-by-step by a Windows Operating System or an Application on Windows platform.
+  3. The <New Task> should be specific and individual, you should not provide different options.
+  4. You should keep <New Task> clear and objective, any vague vocabulary or any subjective terms are forbidden.
+  5. You should try your best not to make the <New Task> become verbose, <New Task> can only add up to 50 words into <Given Task>.
+  6. The detailed target in <New Task> should be specific and clear based on the doc canvas content and control information.
+  7. The <New Task> should be able to implemented by the available controls and actions.
+
+  
+  ## The requirements for <Action Plan>
+  1. The <Action Plan> should be step-by-step actions to take in the doc file environment.
+  2. Each action should be in the available actions from <Available Actions>.
+  3. Each action should be generated with a "step" description which is the function description of the action.
+  4. No need to explain the purpose of the action, just give the actions to take.
+  5. Each plan should focus on a single action, if multiple actions need to be performed, you should separate them into different steps.
+  
+  ## Response Format
+  - You are required to response in a JSON format, consisting of several distinct parts with the following keys and corresponding content:
+    {{
+      "Observation": <Outline the observation of the provided doc file environment based on the given Canvas State and Control State>,
+      "Thought": <Outline your thinking and logic of your New Task and the actions to take,consider the observation of environment and avaiable controls actions>,
+      "New_task":<Give the detailed New Task based on Given Task and the observation of doc environment>,
+      "Actions_plan":<Give the detailed step-by-step actions plan based on the Available Actions and the observation of doc environment.,
+      The format should be a list of action call format separated by "\n">
+    }}
+  
+  ### Action Call Format
+  - The action call format is the same as the available actions in the API list.You are required to provide the action call format in a JSON format:
+    {{
+      "Step": <The number of the step>
+      "Subtask": <The step description the function of the action,which is also the subtask completed by the current action>
+      "ControlLabel": <Specify the precise annotated label of the control item to be selected, adhering strictly to the provided options in the field of "label" in the <Doc Control State:>. If you believe none of the control item is suitable for the task or the task is complete, kindly output a empty string ''.>
+      "ControlText": <Specify the precise control_text of the control item to be selected, adhering strictly to the provided options in the field of "control_text" in the <Doc Control State:>.The control text must match exactly with the selected control label. 
+      If the function to call don't need specify controlText or the task is complete,you can kindly output an empty string ''.
+      If the function to call need to specify controlText and none of the control item is suitable for the task,you should input a possible control name.> 
+      "Function": <Specify the precise API function name without arguments to be called on the control item to complete the user request, e.g., click_input. Leave it a empty string "" if you believe none of the API function is suitable for the task or the task is complete.>
+      "Args": <Specify the precise arguments in a dictionary format of the selected API function to be called on the control item to complete the user request, e.g., {{"control_id":"1","button": "left", "double": false}}. Leave it a empty dictionary {{}} if you the API does not require arguments, or you believe none of the API function is suitable for the task, or the task is complete.>
+    }}
+
+    e.g.
+      {{
+          "Step": 1
+          "Subtask": "change the borders",
+          "ControlLabel": "",
+          "ControlText": "Borders",
+          "Function": "click_input",
+          "Args": {{
+              "button": "left",
+              "double": false
+          }}
+      }}
+
+      {{
+          "Step": 2, 
+          "Subtask": "change the borders",
+          "ControlLabel": "101",
+          "ControlText": "Borders",
+          "Function": "click_input",
+          "Args": {{
+              "control_id": "101",
+              "button": "left",
+              "double": false
+          }}
+      }}
+
+      {{
+          "Step": 3, 
+          "Subtask": "select the target text",
+          "ControlLabel": "",
+          "ControlText": "",
+          "Function": "select_text",
+          "Args": {{
+              "text": "Test For Fun"
+          }}
+      }}
+
+  - The <Actions_plan> field must be strictly in a format separated each action call by "\n". The list format should be like this:
+  "action call 1\naction call 2\naction call 3"
+  - If you think the original task don't need to be detailed, you can directly copy the original task to the "New_task".
+  - You should review the apis function carefully and if the function to call need to specify target control,the 'controlText' field
+  cannot be set empty.
+  - The "Subtask" description should be consistent with the action and also the thought.
+
+  ## Here are some examples for you to complete the user request:
+  {examples}
+
+  ## Tips
+  - Read the above instruction carefully. Make sure the response and action strictly following these instruction and meet the user request.
+  - Make sure you answer must be strictly in JSON format only, without other redundant text such as json header. Your output must be able to be able to be parsed by json.loads(). Otherwise, it will crash the system and destroy the user's computer.
+  - Your task is very important to improve the agent's performance. I will tip you 200$ if you do well. Thank you for your hard work!
+
+user: |-
+  <Given Task:> {given_task}
+  <Reference Steps:> {reference_steps}
+  <Doc Control State:> {doc_control_state}
+  <Your response:>
\ No newline at end of file
diff --git a/dataflow/prompts/instantiation/visual/prefill_example.yaml b/dataflow/prompts/instantiation/visual/prefill_example.yaml
new file mode 100644
index 00000000..a50635ea
--- /dev/null
+++ b/dataflow/prompts/instantiation/visual/prefill_example.yaml
@@ -0,0 +1,44 @@
+
+version: 1.0
+
+example1: 
+  Request: |-
+      <Given Task:> Delete Text in document.
+      <Doc Canvas State:> {'w:document': {'w:body': {'w:p': [{'w:r': {'@w:rsidRPr': '00E2735E', 'w:rPr': {'w:rFonts': {'@w:ascii': 'Consolas', '@w:eastAsia': 'Times New Roman', '@w:hAnsi': 'Consolas', '@w:cs': 'Times New Roman'}, 'w:sz': {'@w:val': '21'}, 'w:szCs': {'@w:val': '21'}, 'w:lang': {'@w:eastAsia': 'zh-CN'}, 'w:color': '000000'}, 'w:t': 'text to edit'}}]}}}
+  Response:
+    observation: |-
+      I observe the canvas state is a Word document with a body containing a paragraph with a run element, which has a text element 'text to edit'.
+    thought: |-
+      My task is to detail the given task and give the step-by-step actions to take. 
+      The user needs to delete text in the Word document. 
+      Based on the canvas state, there is a text element 'text to edit'.
+      And based on the available apis and controls,the user can use "select_text" to select the target to delete,and "type_keys" to type in delete.
+      Therefore,the user can detail the task to delete 'text to edit' in the Word document.
+      In this case, the user should select the text to edit in the Word document and press the 'Delete' key on the keyboard to delete the selected text.
+    new_task: |-
+      Delete the 'text to edit' in the Word document.
+    action_plans: |-
+      {{"step 1":"choose the target text 'text to edit'","controlLabel": "", "controlText": "", "function": "select_text", "args": {{"text": "text to edit"}}}}
+      {{"step 2":"type in delete keys to finish delete","controlLabel": "101", "controlText": "Edit", "function": "type_keys", "args": {{"text": "{DELETE}"}}}}
+
+example2: 
+  Request: |-
+      <Given Task:> Highlight Text in document.
+      <Doc Canvas State:> {'w:document': {'w:body': {'w:p': [{'w:r': {'@w:rsidRPr': '00E2735E', 'w:rPr': {'w:rFonts': {'@w:ascii': 'Consolas', '@w:eastAsia': 'Times New Roman', '@w:hAnsi': 'Consolas', '@w:cs': 'Times New Roman'}, 'w:sz': {'@w:val': '21'}, 'w:szCs': {'@w:val': '21'}, 'w:lang': {'@w:eastAsia': 'zh-CN'}, 'w:color': '000000'}, 'w:t': 'text to edit'}}]}}}
+  Response:
+    observation: |-
+      I observe the canvas state is a Word document with a body containing a paragraph with a run element, which has a text element 'text to edit'.
+    thought: |-
+      My task is to detail the given task and give the step-by-step actions to take. 
+      The user needs to highlight text in the Word document. 
+      Based on the canvas state, there is a text element 'text to edit'.
+      And based on the available apis and controls,the user can use "select_text" to select the target to highlight and then to highlight the text.
+      Since there is no "Highlight" button available,I should click to the 'Home' tab first and then click the 'Highlight' button.
+      Therefore,the user can detail the task to highlight 'text to edit' in the Word document.
+      In this case, the user should select the 'text to edit' in the Word document and press the 'Home' button and 'Highlight' button respectively.
+    new_task: |-
+      Highlight 'text to edit' in the Word document.
+    action_plans: |-
+      {{"step 1":"choose the target text 'text to edit'","controlLabel": "", "controlText": "", "function": "select_text", "args": {{"text": "text to edit"}}}}
+      {{"step 2":"change ribbon to Home to show the highlight button","controlLabel": "11", "controlText": "Home", "function": "click_input", "args": {{"button": "left", "double": false}}}}
+      {{"step 3":"click the highlight button to finish highlight","controlLabel": "", "controlText": "Highlight", "function": "click_input", "args": {{"button": "left", "double": false}}}}
diff --git a/dataflow/schema/execution_schema.json b/dataflow/schema/execution_schema.json
new file mode 100644
index 00000000..21d58a78
--- /dev/null
+++ b/dataflow/schema/execution_schema.json
@@ -0,0 +1,120 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "type": "object",
+  "properties": {
+    "unique_id": { "type": "string" },
+    "app": { "type": "string" },
+    "original": {
+      "type": "object",
+      "properties": {
+        "original_task": { "type": "string" },
+        "original_steps": {
+          "type": "array",
+          "items": { "type": "string" }
+        }
+      },
+      "required": ["original_task", "original_steps"]
+    },
+    "execution_result": {
+      "type": ["object", "null"],
+      "properties": {
+        "result": {
+          "type": ["object", "null"],
+          "properties": {
+            "reason": { "type": "string" },
+            "sub_scores": {
+              "type": "object",
+              "patternProperties": {
+                ".*": { "type": "string" }
+              }
+            },
+            "complete": { "type": "string" }
+          },
+          "required": ["reason", "sub_scores", "complete"]
+        },
+        "error": {
+          "type": ["null", "object"],
+          "properties": {
+            "type": { "type": "string" },
+            "message": { "type": "string" },
+            "traceback": { "type": "string" }
+          },
+          "required": ["type", "message", "traceback"]
+        }
+      }
+    },
+    "instantiation_result": {
+      "type": "object",
+      "properties": {
+        "choose_template": {
+          "type": "object",
+          "properties": {
+            "result": { "type": ["string", "null"] },
+            "error": { "type": ["null", "string"] }
+          },
+          "required": ["result", "error"]
+        },
+        "prefill": {
+          "type": ["object", "null"],
+          "properties": {
+            "result": {
+              "type": ["object", "null"],
+              "properties": {
+                "instantiated_request": { "type": "string" },
+                "instantiated_plan": {
+                  "type":["array", "null"],
+                  "items": {
+                    "type": "object",
+                    "properties": {
+                      "Step": { "type": "integer" },
+                      "Subtask": { "type": "string" },
+                      "ControlLabel": { "type": ["string", "null"] },
+                      "ControlText": { "type": "string" },
+                      "Function": { "type": "string" },
+                      "Args": { "type": "object", "additionalProperties": true },
+                      "Success": { "type": ["boolean", "null"] },
+                      "MatchedControlText": { "type": ["string", "null"] }
+                    },
+                    "required": ["Step", "Subtask", "Function", "Args", "Success", "MatchedControlText"]
+                  }
+                }
+              },
+              "required": ["instantiated_request", "instantiated_plan"]
+            },
+            "error": { "type": ["null", "string"] }
+          },
+          "required": ["result", "error"]
+        },
+        "instantiation_evaluation": {
+          "type": "object",
+          "properties": {
+            "result": {
+              "type": ["object", "null"],
+              "properties": {
+                "judge": { "type": "boolean" },
+                "thought": { "type": "string" },
+                "request_type": { "type": "string" }
+              },
+              "required": ["judge", "thought", "request_type"]
+            },
+            "error": { "type": ["null", "string"] }
+          },
+          "required": ["result", "error"]
+        }
+      }
+    },
+    "time_cost": {
+      "type": "object",
+      "properties": {
+        "choose_template": { "type": ["number", "null"] },
+        "prefill":{ "type": ["number", "null"] },
+        "instantiation_evaluation": { "type": ["number", "null"] },
+        "total": { "type": ["number", "null"] },
+        "execute": { "type": ["number", "null"] },
+        "execute_eval": { "type": ["number", "null"] }
+      },
+      "required": ["choose_template", "prefill", "instantiation_evaluation", "total", "execute", "execute_eval"]
+    }
+  },
+  "required": ["unique_id", "app", "original", "execution_result", "instantiation_result", "time_cost"]
+}
diff --git a/dataflow/schema/instantiation_schema.json b/dataflow/schema/instantiation_schema.json
new file mode 100644
index 00000000..6a132151
--- /dev/null
+++ b/dataflow/schema/instantiation_schema.json
@@ -0,0 +1,99 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "type": "object",
+  "properties": {
+    "unique_id": { "type": "string" },
+    "app": { "type": "string" },
+    "original": {
+      "type": "object",
+      "properties": {
+        "original_task": { "type": "string" },
+        "original_steps": {
+          "type": "array",
+          "items": { "type": "string" }
+        }
+      },
+      "required": ["original_task", "original_steps"]
+    },
+    "execution_result": {
+      "type": ["object", "null"],
+      "properties": {
+        "result": {
+          "type":"null"
+        },
+        "error": {
+          "type":"null"
+        }
+      }
+    },
+    "instantiation_result": {
+      "type": "object",
+      "properties": {
+        "choose_template": {
+          "type": "object",
+          "properties": {
+            "result": { "type": ["string", "null"] },
+            "error": { "type": ["null", "string"] }
+          },
+          "required": ["result", "error"]
+        },
+        "prefill": {
+          "type": ["object", "null"],
+          "properties": {
+            "result": {
+              "type": ["object", "null"],
+              "properties": {
+                "instantiated_request": { "type": "string" },
+                "instantiated_plan": {
+                  "type":["array", "null"],
+                  "items": {
+                    "type": "object",
+                    "properties": {
+                      "Step": { "type": "integer" },
+                      "Subtask": { "type": "string" },
+                      "ControlLabel": { "type": ["string", "null"] },
+                      "ControlText": { "type": "string" },
+                      "Function": { "type": "string" },
+                      "Args": { "type": "object", "additionalProperties": true }
+                    },
+                    "required": ["Step", "Subtask", "Function", "Args"]
+                  }
+                }
+              },
+              "required": ["instantiated_request", "instantiated_plan"]
+            },
+            "error": { "type": ["null", "string"] }
+          },
+          "required": ["result", "error"]
+        },
+        "instantiation_evaluation": {
+          "type": "object",
+          "properties": {
+            "result": {
+              "type": ["object", "null"],
+              "properties": {
+                "judge": { "type": "boolean" },
+                "thought": { "type": "string" },
+                "request_type": { "type": "string" }
+              },
+              "required": ["judge", "thought", "request_type"]
+            },
+            "error": { "type": ["null", "string"] }
+          },
+          "required": ["result", "error"]
+        }
+      }
+    },
+    "time_cost": {
+      "type": "object",
+      "properties": {
+        "choose_template": { "type": ["number", "null"] },
+        "prefill":{ "type": ["number", "null"] },
+        "instantiation_evaluation": { "type": ["number", "null"] },
+        "total": { "type": ["number", "null"] }
+      },
+      "required": ["choose_template", "prefill", "instantiation_evaluation", "total"]
+    }
+  },
+  "required": ["unique_id", "app", "original", "execution_result", "instantiation_result", "time_cost"]
+}
diff --git a/dataflow/tasks/prefill/bulleted.json b/dataflow/tasks/prefill/bulleted.json
new file mode 100644
index 00000000..237b68eb
--- /dev/null
+++ b/dataflow/tasks/prefill/bulleted.json
@@ -0,0 +1,9 @@
+{
+    "app": "word",
+    "unique_id": "5",
+    "task": "Turning lines of text into a bulleted list in Word",
+    "refined_steps": [
+        "1. Place the cursor at the beginning of the line of text you want to turn into a bulleted list",
+        "2. Click the Bullets button in the Paragraph group on the Home tab and choose a bullet style"
+    ]
+}
\ No newline at end of file
diff --git a/dataflow/tasks/prefill/watermark.json b/dataflow/tasks/prefill/watermark.json
new file mode 100644
index 00000000..ef14611f
--- /dev/null
+++ b/dataflow/tasks/prefill/watermark.json
@@ -0,0 +1,11 @@
+{
+    "app": "word",
+    "unique_id": "108",
+    "task": "Put a watermark on all pages in Word for Office 365",
+    "refined_steps": [
+        "1.On the **Design** tab, select **Watermark**.",
+        "2.Select **Custom Watermark**.",
+        "3.Choose **Text watermark** and type your text in the **Text** box.",
+        "4.Select **OK**."
+    ]
+}
\ No newline at end of file
diff --git a/dataflow/templates/word/description.json b/dataflow/templates/word/description.json
new file mode 100644
index 00000000..375c17d8
--- /dev/null
+++ b/dataflow/templates/word/description.json
@@ -0,0 +1,4 @@
+{
+    "template1.docx": "A doc with a rectangle shape. Can be moved, resized, or deleted.",
+    "template2.docx": "A doc with a line of text. Can be edited, deleted, or replaced."
+}
diff --git a/dataflow/templates/word/template1.docx b/dataflow/templates/word/template1.docx
new file mode 100644
index 00000000..3133c897
Binary files /dev/null and b/dataflow/templates/word/template1.docx differ
diff --git a/dataflow/templates/word/template2.docx b/dataflow/templates/word/template2.docx
new file mode 100644
index 00000000..26316949
Binary files /dev/null and b/dataflow/templates/word/template2.docx differ
diff --git a/documents/docs/index.md b/documents/docs/index.md
index 13fa0646..9c0595c2 100644
--- a/documents/docs/index.md
+++ b/documents/docs/index.md
@@ -67,22 +67,6 @@ UFO sightings have garnered attention from various media outlets, including:
 * For other communications, please contact [ufo-agent@microsoft.com](mailto:ufo-agent@microsoft.com)
 ---
 
-<!-- ## 🎬 Demo Examples
-
-We present two demo videos that complete user request on Windows OS using UFO. For more case study, please consult our [technical report](https://arxiv.org/abs/2402.07939).
-
-#### 1️⃣🗑️ Example 1: Deleting all notes on a PowerPoint presentation.
-In this example, we will demonstrate how to efficiently use UFO to delete all notes on a PowerPoint presentation with just a few simple steps. Explore this functionality to enhance your productivity and work smarter, not harder!
-
-<iframe width="560" height="315" src="https://github.com/microsoft/UFO/assets/11352048/cf60c643-04f7-4180-9a55-5fb240627834" frameborder="0" allowfullscreen></iframe>
-
-&nbsp;
-
-#### 2️⃣📧 Example 2: Composing an email using text from multiple sources.
-In this example, we will demonstrate how to utilize UFO to extract text from Word documents, describe an image, compose an email, and send it seamlessly. Enjoy the versatility and efficiency of cross-application experiences with UFO!
-
-<iframe width="560" height="315" src="https://github.com/microsoft/UFO/assets/11352048/aa41ad47-fae7-4334-8e0b-ba71c4fc32e0" frameborder="0" allowfullscreen></iframe> -->
-
 &nbsp;
 ## 📚 Citation
 Our technical report paper can be found [here](https://arxiv.org/abs/2402.07939). Note that previous HostAgent and AppAgent in the paper are renamed to HostAgent and AppAgent in the code base to better reflect their functions.
@@ -112,4 +96,4 @@ For more information on GUI agents, refer to our survey paper: [Large Language M
   gtag('js', new Date());
 
   gtag('config', 'G-FX17ZGJYGC');
-</script>
\ No newline at end of file
+</script>
diff --git a/ufo/agents/agent/app_agent.py b/ufo/agents/agent/app_agent.py
index d1d73bdd..246c9326 100644
--- a/ufo/agents/agent/app_agent.py
+++ b/ufo/agents/agent/app_agent.py
@@ -58,6 +58,7 @@ def __init__(
         self.online_doc_retriever = None
         self.experience_retriever = None
         self.human_demonstration_retriever = None
+
         self.Puppeteer = self.create_puppeteer_interface()
 
         self.set_state(ContinueAppAgentState())
diff --git a/ufo/agents/agent/basic.py b/ufo/agents/agent/basic.py
index dba15f07..f85cffb7 100644
--- a/ufo/agents/agent/basic.py
+++ b/ufo/agents/agent/basic.py
@@ -139,7 +139,7 @@ def message_constructor(self) -> List[Dict[str, Union[str, List[Dict[str, str]]]
 
     @classmethod
     def get_response(
-        cls, message: List[dict], namescope: str, use_backup_engine: bool
+        cls, message: List[dict], namescope: str, use_backup_engine: bool, configs = configs
     ) -> str:
         """
         Get the response for the prompt.
@@ -149,7 +149,7 @@ def get_response(
         :return: The response.
         """
         response_string, cost = llm_call.get_completion(
-            message, namescope, use_backup_engine=use_backup_engine
+            message, namescope, use_backup_engine=use_backup_engine, configs = configs
         )
         return response_string, cost
 
@@ -243,6 +243,7 @@ def process_resume(self) -> None:
         if self.processor:
             self.processor.resume()
 
+
     def process_asker(self, ask_user: bool = True) -> None:
         """
         Ask for the process.
diff --git a/ufo/agents/processors/app_agent_processor.py b/ufo/agents/processors/app_agent_processor.py
index af97b362..3658ed73 100644
--- a/ufo/agents/processors/app_agent_processor.py
+++ b/ufo/agents/processors/app_agent_processor.py
@@ -22,7 +22,8 @@
     from ufo.agents.agent.app_agent import AppAgent
 
 configs = Config.get_instance().config_data
-BACKEND = configs["CONTROL_BACKEND"]
+if configs is not None:
+    BACKEND = configs["CONTROL_BACKEND"]
 
 
 class AppAgentProcessor(BaseProcessor):
@@ -503,7 +504,8 @@ def demonstration_prompt_helper(self) -> Tuple[List[str], List[str]]:
         return examples, tips
 
     def get_filtered_annotation_dict(
-        self, annotation_dict: Dict[str, UIAWrapper]
+        self, annotation_dict: Dict[str, UIAWrapper],
+        configs = configs
     ) -> Dict[str, UIAWrapper]:
         """
         Get the filtered annotation dictionary.
diff --git a/ufo/agents/processors/basic.py b/ufo/agents/processors/basic.py
index b9021eb5..e93c94a5 100644
--- a/ufo/agents/processors/basic.py
+++ b/ufo/agents/processors/basic.py
@@ -21,7 +21,8 @@
 from ufo.module.context import Context, ContextNames
 
 configs = Config.get_instance().config_data
-BACKEND = configs["CONTROL_BACKEND"]
+if configs is not None:
+    BACKEND = configs["CONTROL_BACKEND"]
 
 
 class BaseProcessor(ABC):
diff --git a/ufo/agents/processors/host_agent_processor.py b/ufo/agents/processors/host_agent_processor.py
index 94a38500..a6819795 100644
--- a/ufo/agents/processors/host_agent_processor.py
+++ b/ufo/agents/processors/host_agent_processor.py
@@ -13,7 +13,8 @@
 from ufo.module.context import Context, ContextNames
 
 configs = Config.get_instance().config_data
-BACKEND = configs["CONTROL_BACKEND"]
+if configs is not None:
+    BACKEND = configs["CONTROL_BACKEND"]
 
 if TYPE_CHECKING:
     from ufo.agents.agent.host_agent import HostAgent
diff --git a/ufo/automator/ui_control/controller.py b/ufo/automator/ui_control/controller.py
index 87df3fba..749f51ce 100644
--- a/ufo/automator/ui_control/controller.py
+++ b/ufo/automator/ui_control/controller.py
@@ -18,7 +18,7 @@
 
 configs = Config.get_instance().config_data
 
-if configs.get("AFTER_CLICK_WAIT", None) is not None:
+if configs is not None and configs.get("AFTER_CLICK_WAIT", None) is not None:
     pywinauto.timings.Timings.after_clickinput_wait = configs["AFTER_CLICK_WAIT"]
     pywinauto.timings.Timings.after_click_wait = configs["AFTER_CLICK_WAIT"]
 
diff --git a/ufo/automator/ui_control/openfile.py b/ufo/automator/ui_control/openfile.py
index 2058a369..49a51dd6 100644
--- a/ufo/automator/ui_control/openfile.py
+++ b/ufo/automator/ui_control/openfile.py
@@ -9,7 +9,10 @@
 
 configs = Config.get_instance().config_data
 
-BACKEND = configs["CONTROL_BACKEND"]
+if configs is not None:
+    BACKEND = configs["CONTROL_BACKEND"]
+else:
+    BACKEND = "uia"
 
 
 class FileController:
@@ -17,9 +20,9 @@ class FileController:
     Control block for open file / specific APP and proceed the operation.
     """
 
-    def __init__(self):
+    def __init__(self, backend=BACKEND):
 
-        self.backend = BACKEND
+        self.backend = backend
         self.file_path = ""
         self.APP = ""
         self.apptype = ""
diff --git a/ufo/automator/ui_control/screenshot.py b/ufo/automator/ui_control/screenshot.py
index b7c884b1..9243a78d 100644
--- a/ufo/automator/ui_control/screenshot.py
+++ b/ufo/automator/ui_control/screenshot.py
@@ -18,7 +18,11 @@
 
 configs = Config.get_instance().config_data
 
-DEFAULT_PNG_COMPRESS_LEVEL = int(configs.get("DEFAULT_PNG_COMPRESS_LEVEL", 0))
+if configs is not None:
+    DEFAULT_PNG_COMPRESS_LEVEL = int(configs.get("DEFAULT_PNG_COMPRESS_LEVEL", 0))
+else:
+    DEFAULT_PNG_COMPRESS_LEVEL = 6
+
 
 
 class Photographer(ABC):
diff --git a/ufo/config/config.py b/ufo/config/config.py
index c9979cb0..08a1e29b 100644
--- a/ufo/config/config.py
+++ b/ufo/config/config.py
@@ -14,7 +14,10 @@ class Config:
 
     def __init__(self):
         # Load config here
-        self.config_data = self.load_config()
+        if os.getenv("RUN_CONFIGS", "true").lower() != "false":
+            self.config_data = self.load_config()
+        else:
+            self.config_data = None
 
     @staticmethod
     def get_instance():
@@ -26,7 +29,7 @@ def get_instance():
             Config._instance = Config()
         return Config._instance
 
-    def load_config(self, config_path="ufo/config/") -> dict:
+    def load_config(self, config_path = "ufo/config/") -> dict:
         """
         Load the configuration from a YAML file and environment variables.
 
@@ -45,14 +48,13 @@ def load_config(self, config_path="ufo/config/") -> dict:
             # Update configs with YAML data
             if yaml_data:
                 configs.update(yaml_data)
-            with open(path + "config_dev.yaml", "r") as file:
-                yaml_dev_data = yaml.safe_load(file)
-            with open(path + "config_prices.yaml", "r") as file:
-                yaml_prices_data = yaml.safe_load(file)
-            # Update configs with YAML data
-            if yaml_data:
+            if os.path.exists(path + "config_dev.yaml"):
+                with open(path + "config_dev.yaml", "r") as file:
+                    yaml_dev_data = yaml.safe_load(file)
                 configs.update(yaml_dev_data)
-            if yaml_prices_data:
+            if os.path.exists(path + "config_prices.yaml"):
+                with open(path + "config_prices.yaml", "r") as file:
+                    yaml_prices_data = yaml.safe_load(file)
                 configs.update(yaml_prices_data)
         except FileNotFoundError:
             print_with_color(
diff --git a/ufo/llm/llm_call.py b/ufo/llm/llm_call.py
index 78a302be..b47ac9e5 100644
--- a/ufo/llm/llm_call.py
+++ b/ufo/llm/llm_call.py
@@ -12,7 +12,7 @@
 
 
 def get_completion(
-    messages, agent: str = "APP", use_backup_engine: bool = True
+    messages, agent: str = "APP", use_backup_engine: bool = True, configs = configs
 ) -> Tuple[str, float]:
     """
     Get completion for the given messages.
@@ -23,13 +23,14 @@ def get_completion(
     """
 
     responses, cost = get_completions(
-        messages, agent=agent, use_backup_engine=use_backup_engine, n=1
+        messages, agent=agent, use_backup_engine=use_backup_engine, n=1, configs = configs
     )
     return responses[0], cost
 
 
 def get_completions(
-    messages, agent: str = "APP", use_backup_engine: bool = True, n: int = 1
+    messages, agent: str = "APP", use_backup_engine: bool = True, n: int = 1,
+    configs = configs
 ) -> Tuple[list, float]:
     """
     Get completions for the given messages.
@@ -44,6 +45,10 @@ def get_completions(
         agent_type = "HOST_AGENT"
     elif agent.lower() in ["app", "appagent"]:
         agent_type = "APP_AGENT"
+    elif agent.lower() == "prefill":
+        agent_type = "PREFILL_AGENT"
+    elif agent.lower() == "filter":
+        agent_type = "FILTER_AGENT"
     elif agent.lower() == "backup":
         agent_type = "BACKUP_AGENT"
     else:
diff --git a/ufo/llm/openai.py b/ufo/llm/openai.py
index 9765cc33..75f985a6 100644
--- a/ufo/llm/openai.py
+++ b/ufo/llm/openai.py
@@ -29,7 +29,7 @@ def __init__(self, config: Dict[str, Any], agent_type: str) -> None:
         self.config = config
         self.api_type = self.config_llm["API_TYPE"].lower()
         self.max_retry = self.config["MAX_RETRY"]
-        self.prices = self.config["PRICES"]
+        self.prices = self.config.get("PRICES", {})
         assert self.api_type in ["openai", "aoai", "azure_ad"], "Invalid API type"
 
         self.client: OpenAI = OpenAIService.get_openai_client(
diff --git a/ufo/module/basic.py b/ufo/module/basic.py
index 81edb918..db1f3add 100644
--- a/ufo/module/basic.py
+++ b/ufo/module/basic.py
@@ -702,7 +702,7 @@ def capture_last_snapshot(self) -> None:
                     app_agent.Puppeteer.save_to_xml(xml_save_path)
 
     @staticmethod
-    def initialize_logger(log_path: str, log_filename: str) -> logging.Logger:
+    def initialize_logger(log_path: str, log_filename: str, mode='a', configs = configs) -> logging.Logger:
         """
         Initialize logging.
         log_path: The path of the log file.
@@ -717,7 +717,7 @@ def initialize_logger(log_path: str, log_filename: str) -> logging.Logger:
             logger.handlers = []
 
         log_file_path = os.path.join(log_path, log_filename)
-        file_handler = logging.FileHandler(log_file_path, encoding="utf-8")
+        file_handler = logging.FileHandler(log_file_path, mode = mode, encoding="utf-8")
         formatter = logging.Formatter("%(message)s")
         file_handler.setFormatter(formatter)
         logger.addHandler(file_handler)