UFO 数据流Dataflow实战指南为大型动作模型LAM构建数据收集流水线【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO本指南基于 UFO 开源仓库中的 dataflow/README.md 及其配套源码系统讲解如何利用 Dataflow 模块为大型动作模型Large Action Models, LAM收集训练数据从任务计划数据出发经由模板选择Instantiation、动作预填充Prefill、过滤评估Filter到计划执行Execution与结果评估的全链路。阅读完本文你将掌握 Dataflow 的三种运行模式--dataflow/--instantiation/--execution、单任务与批量处理方式、LLM 配置与输入文件组织规范并能读懂最终输出的结构化结果 JSON 与日志。背景Dataflow 在 UFO 项目中的定位Dataflow 是 UFO 项目中用于实现Large Action ModelsLAMs数据收集的核心模块。LAM 是一类能够基于 GUI 环境感知生成并执行动作序列的模型其训练依赖大量任务计划 → 任务-动作数据的配对样本。本模块的目标正是自动生成这类数据给定一份任务计划数据task-plan data由 LLM 完成实例化instantiation生成任务-动作数据task-action data随后在真实应用环境中执行并评估质量最终沉淀为可复用的训练语料。该模块是论文Large Action Models: From Inception to Implementation的实现部分并隶属于 UFO 项目体系。引用该工作时可参考以下文献信息作者包括 Lu Wang、Fangkai Yang、Chaoyun Zhang 等misc{wang2024largeactionmodelsinception, title{Large Action Models: From Inception to Implementation}, author{Lu Wang and Fangkai Yang and Chaoyun Zhang and Junting Lu and Jiaxu Qian and Shilin He and Pu Zhao and Bo Qiao and Ray Huang and Si Qin and Qisheng Su and Jiayi Ye and Yudi Zhang and Jian-Guang Lou and Qingwei Lin and Saravan Rajmohan and Dongmei Zhang and Qi Zhang}, year{2024}, eprint{2412.10047}, archivePrefix{arXiv}, primaryClass{cs.AI}, }数据流三层结构Instantiation、Execution 与 DataflowDataflow 使用 UFO 来为给定任务实现instantiation、execution与dataflow三个层级同时支持批量处理与单任务处理两种粒度Instantiation实例化任务的准备与设定过程典型步骤包括choosing template选择模板、prefill预填充与filter过滤。Execution执行实际运行任务的过程负责执行 Instantiation 产出的动作计划执行完成后由一个评估 agentevaluation agent对整个执行过程的质量进行打分。Dataflow数据流把instantiation与execution串联成单一管线的总流程提供端到端的任务处理方案保证从初始化到执行的所有必要步骤无缝衔接。三个层级可以灵活组合若只需完成其中一部分可单独使用instantiation或execution若任务同时需要两步则dataflow将其整合一条命令从起点跑到终点。整体处理流程如下给定任务计划数据LLM 将实例化任务-动作数据包括选择模板、预填充与过滤。安装与环境准备1. 安装依赖包在 UFO 仓库根目录下安装所需依赖pip install -r requirements.txt安装完成后即可通过python -m dataflow调用模块入口入口实现见 dataflow/main.py其内部直接委托给 dataflow/dataflow.py 的main()。2. 配置 LLMconfig.yaml运行 Dataflow 前需要分别为 PrefillAgent 与 FilterAgent 单独提供 LLM 配置。做法是复制模板文件生成自己的配置cp dataflow/config/config.yaml.template dataflow/config/config.yaml然后编辑 dataflow/config/config.yaml 中PREFILL_AGENT与FILTER_AGENT两段配置。OpenAIVISUAL_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 modelAzure OpenAIAOAIVISUAL_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非视觉模型配置若希望使用非视觉模型如 GPT-4只需在 dataflow/config/config.yaml 中做两处修改设置VISUAL_MODE: False以启用非视觉模式为每个 agent 指定合适的API_MODELOpenAI与API_DEPLOYMENT_IDAOAI。注意确保这些设置准确无误才能让非视觉模型有效工作。其他通用模型参数在 config.yaml.template 的### For parameters段落中还提供了影响模型推理行为的全局参数可一并调整参数默认值说明MAX_TOKENS2000响应补全的最大 token 上限MAX_RETRY3响应补全的最大重试次数TEMPERATURE0.0模型温度值越低输出越稳定一致TOP_P0.0核采样参数值越低输出越保守TIMEOUT60单次调用的超时时间秒另外若使用带 Azure AD 认证的 AOAI还需在 agent 配置中补充AAD_TENANT_ID、AAD_API_SCOPE与AAD_API_SCOPE_BASE格式为API://YOUR_SCOPE_BASE仅需填写YOUR_SCOPE_BASE部分这些字段在 config.yaml.template 中有完整的注释说明。注意 若使用 GitHub 等开源工具切勿将包含私钥的config.yaml公开上传以免泄露密钥。3. 理解 config_dev.yaml 关键配置dataflow/config/config_dev.yaml 规定了相关文件的路径与默认设置源码中通过 dataflow/config/config.py 的Config单例加载并在加载后对PREFILL_AGENT与FILTER_AGENT调用update_api_base做 API 地址修正。核心配置项如下配置项默认值说明CONTROL_BACKENDuiaUI 控件控制后端当前支持uia与win32CONTROL_LIST常见控件类型列表参与识别的控件类型Button、Edit、TabItem、Document 等MATCH_STRATEGYfuzzy窗口匹配与控制过滤的匹配策略支持contains、fuzzy、regexPREFILL_PROMPT/FILTER_PROMPTdataflow/prompts/instantiation/{mode}/prefill.yaml等各 agent 使用的提示词模板路径TEMPLATE_METHODLLM模板选择的后端方式支持SemanticSimilarity、LLMREFORMAT_TO_BATCHTrue是否将 dataflow 结果重排为 UFO 批量模式格式REFORMAT_TO_BATCH_HUBdatasUFO重排结果的输出路径TASKS_HUBdataflow/tasks/prefill默认的任务集tasks hub路径TEMPLATE_PATHdataflow/templates模板根目录路径RESULT_HUBdataflow/results/{task_type}结果目录task_type为instantiation或executionINSTANTIATION_RESULT_SCHEMA/EXECUTION_RESULT_SCHEMAdataflow/schema/*.json结果校验所用的 JSON SchemaCONTROL_FILTER_TYPE[]控制过滤器类型列表支持TEXT、SEMANTIC、ICONCONTROL_FILTER_MODEL_SEMANTIC_NAMEall-MiniLM-L6-v2语义相似度所用的嵌入模型名CONTROL_EMBEDDING_CACHE_PATHdataflow/cache/控制过滤器的嵌入缓存路径CONTROL_FILTER_TOP_K_PLAN2控制过滤器作用于 UFO 的 top-K 计划数默认 2MAX_STEPS30execute_flow 的最大执行步数可按需调整其中MAX_STEPS在 dataflow/execution/workflow/execute_flow.py 的execute_plan中被逐步骤校验一旦session_step超过该值会抛出Maximum steps exceeded.运行时错误用于防止执行陷入死循环。准备输入文件运行任务前需准备好三类文件待实例化的任务 JSON、作为实例化参考的模板文件、以及 JSON 格式的模板描述文件。1. 任务 JSON需要实例化的任务应组织为一个 JSON 文件目录默认目录路径为dataflow/tasks可在dataflow/config/config.yaml中修改或在终端启动时通过--task_path指定。例如dataflow/tasks/prefill/下的一个任务文件{ // 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 ] }字段含义app指定要操作的应用unique_id用于区分不同任务的唯一 IDtask为待实例化的任务描述refined_steps为任务对应的步骤列表。源码 dataflow/data_flow_controller.py 中的TaskObject会把该 JSON 的键值转换为小写并去空格挂载为任务对象属性并依据app字段匹配AppEnum当前支持 Word.docx/winword、Excel.xlsx/excel、PowerPoint.pptx/powerpnt。2. 模板与描述文件为每个应用准备一个以其名称命名的文件夹里面放置实例化时作为参考的应用文件。例如为 Word 准备template1.docx就放在dataflow/templates/word/template1.docx。同时每个应用文件夹下需要一份dataflow/templates/word/description.json逐条描述每个模板文件的用途例如{ template1.docx: A document with a rectangle shape, template2.docx: A document with a line of text }如果缺少description.json文件系统会随机选取一个模板文件对应 choose_template_flow.py 中的_choose_random_template逻辑并在控制台打印Randomly selected template: ...提示。3. 最终文件结构确保以下文件就绪待实例化的 JSON 文件作为实例化参考的模板文件JSON 格式的描述文件整体目录结构形如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 └── ...启动运行单任务与批量处理完成上述准备后即可在命令行启动。系统会根据传入路径自动判断是单任务处理传入文件路径还是批量处理传入目录路径该逻辑实现在 dataflow/dataflow.py 的validate_path中os.path.isfile判定为fileos.path.isdir判定为directory两者皆非则抛出异常随后分别进入process_task或process_batch。process_batch会列出目录下全部 JSON 文件并逐个调用DataFlowController处理无任务时给出黄色提示。同时你可以单独使用instantiation/execution段落或作为一个整体段落命名为dataflow使用。默认任务集为 config_dev.yaml 中的TASKS_HUB即dataflow/tasks/prefill。Dataflow 任务python -m dataflow --dataflow --task_path path_to_task_fileInstantiation 任务python -m dataflow --instantiation --task_path path_to_task_fileExecution 任务python -m dataflow --execution --task_path path_to_task_file若未指定--task_path默认使用TASKS_HUB配置若三个任务类型参数一个都未提供程序会提示必须指定其一并退出。入口解析代码见 dataflow.py。TEMPLATE_METHOD用于在dataflow/config/config_dev.yaml中选择模板选择函数的后端LLM或SemanticSimilarity。若选择LLM由于使用的是视觉版本你需要手动在templates/YOUR_APP/images目录中生成截图文件名需与模板名一致且为PNG格式该目录在仓库中已有 7 张示例 PNG 图片可供参考见 dataflow/templates/word/images。工作流深度解析Instantiation 三阶段实例化过程包含三个关键步骤根据指定的 app 与指令Choose a template选择模板文件基于当前截图Prefill预填充任务Filter过滤已建立的任务。给定初始任务后dataflow 先选择模板阶段 1再基于 Word 环境预填充初始任务以获得任务-动作数据阶段 2最后过滤已建立的任务以评估任务-动作数据的质量阶段 31. 选择模板文件Choose Template应用模板必须定义并描述在dataflow/templates/app中。例如要为 Word 应用实例化任务就把相关的.docx文件放到dataflow/templates/word下并附上description.json。系统会根据模板描述与指令的匹配程度选择最合适的模板。底层实现位于 choose_template_flow.pySemanticSimilarity 后端将各模板的描述文本经 HuggingFace 嵌入模型默认all-MiniLM-L6-v2自动加sentence-transformers/前缀编码后构建 FAISS 索引用db.similarity_search(given_task, k1)返回与任务语义最相近的模板LLM 后端实例化TemplateAgent构造提示词后请求 LLM 返回 JSON取第一个键作为模板文件名若文件名不在描述集合中则报错。选中的模板文件随后被复制到dataflow/results/saved_document/目录下以任务文件名命名再交由后续流程打开。2. 预填充任务Prefill选定模板文件后系统会打开该模板并截图。若模板文件正被占用此步骤可能出错。截图会发送给动作预填充 agentaction prefill agent由其返回修改后的任务即实例化后的请求与动作计划。实现见 prefill_flow.pyPrefillFlow通过ControlInspectorFacade检查应用窗口中的控件元素、用PhotographerFacade截图并做数字标注然后构造提示词调用PrefillAgent.get_response将响应解析为New_task实例化请求与Actions_plan动作计划。过程日志记录在dataflow/logs/{task}/prefill/下的prefill_messages.json与prefill_responses.json。3. 过滤任务Filter完成预填充的任务由过滤 agentfilter agent评估并给出反馈输出judge任务是否可行、thought判断依据与type请求类型。对应 filter_flow.py 中的FilterFlow其过滤结果会写入instantiation_evaluation字段并作为任务质量判定的依据judge为true时任务归入instantiation_pass否则归入instantiation_fail映射表见 data_flow_controller.py 中的INSTANTIATION_RESULT_MAP。Execution 执行与评估实例化得到的计划由执行 agent 逐步执行执行完成后评估 agentevaluation agent会评估整个执行过程的质量。该阶段中给定任务-动作数据执行过程会基于应用环境匹配真实控件并逐步执行计划实现见 execute_flow.py 的ExecuteFlow通过WinCOMReceiverBasic创建 API 接收器与 UI 控件接收器为应用建立执行通道逐步骤解析计划_parse_step_plan提取Subtask、ControlText、Function、Args等调用process()完成截图、执行动作与日志落盘每步执行后回填Success与MatchedControlTextMatchedControlText指执行过程中根据计划匹配到的控件文本全部步骤完成后保存文档并退出应用再由ExecuteEvalAgent.evaluate基于执行日志给出结果含reason、sub_scores、complete。执行结果的分类依据complete字段yes→execution_pass、no→execution_fail、unsure→execution_unsure见 data_flow_controller.py 的EXECUTION_RESULT_MAP。结果输出与目录规范任务结果的组织结构如下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 │ └── ... └── ...各目录含义说明总体描述该目录结构把任务处理结果按 instantiation、execution、dataflow 三类归类。Instantiationinstantiation目录下包含实例化成功instantiation_pass与实例化失败instantiation_fail两个子目录。Execution执行结果存于execution目录分为成功execution_pass、失败execution_fail与不确定execution_unsure。Dataflow 结果dataflow目录同样按执行成功/失败/不确定组织结果提供数据处理管线的整体视图。保存文档实例化结果另存于saved_document目录便于访问与引用。结果的保存由DataFlowController.save_result完成先使用 dataflow/schema/instantiation_schema.json 或 dataflow/schema/execution_schema.json 做 JSON Schema 校验校验失败仅告警不阻断再按上述分类写入对应目录。若REFORMAT_TO_BATCH开启还会通过reformat_to_batch把结果转换为 UFO 批量模式的datasUFO格式包含tasks/与files/两个子目录。结果字段说明结果数据采用层级结构组织核心字段包括unique_id、app、original、execution_result、instantiation_result与time_cost层级Hierarchy数据以层级化方式呈现便于理解字段间关系类型描述Type Description每个字段的类型如string、array、object明确其数据格式字段用途Field Purpose每个字段均有简短描述说明其功能。执行结果与错误execution_result包含任务执行结果包括子任务表现、完成状态与执行中遇到的错误instantiation_result描述任务实例化过程包括模板选择、预填充任务与实例化评估error任务执行出错时该字段包含相关错误信息。时间消耗time_cost记录任务各阶段从模板选择到任务执行的耗时用于分析任务效率。示例数据一Compatibility Mode 任务{ 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 } }示例数据二bulleted 任务Quick Start 样例仓库在dataflow/tasks/prefill中准备了两个演示用例。以 Turning lines of text into a bulleted list in Word 任务为例其完整结果结构如下unique_id任务标识此处为5。app使用的应用此处为word。original原始任务描述与步骤。original_task用简单语言描述的任务将文本转成项目符号列表。original_steps完成任务所需步骤列表。execution_result任务执行结果。result执行结果描述包含成功信息与每个子任务的 sub-score。complete: yes表示评估 agent 认为执行过程成功sub_score对应prefill中instantiated_plan每个子任务的评估。error执行中出现错误在此报告本例为null。instantiation_result任务实例化详情。choose_template任务中创建或使用的模板/文档路径此处为 bulleted 文档。prefill描述instantiated_request与instantiated_plan及涉及步骤如选择文本、点击按钮是 prefill flow 的结果。Success与MatchedControlText在执行阶段补充Success表示子任务是否成功执行MatchedControlText表示执行中根据计划匹配到的控件文本。instantiation_evaluation任务可行性与请求的评估反馈是 filter flow 的结果。judge: true表示任务评估为积极任务有效/判定成功thought为详细理由。time_cost任务各环节耗时包括模板选择、prefill、实例化评估与执行并给出总耗时。{ 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 } }快速开始Quick Start仓库在dataflow/tasks/prefill下提供了两个演示用例bulleted与rotate。安装依赖后直接在命令行执行python -m dataflow --dataflow终端会出现提示信息说明 dataflow 正在工作。注意这里省略了--task_path因此会使用默认的TASKS_HUBdataflow/tasks/prefill。运行后的文件结构两个任务完成后输出文件如下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 │ │ │ └── ... └── ...日志文件对应日志可在logs/bulleted与logs/rotate目录中找到实际根目录为dataflow/logs/由 config_dev.yaml 的LOG_PATH: dataflow/logs/{task}等配置项决定。每个工作流的详细日志都会被记录捕捉执行过程的每一步包括 prefill、filter、execute 各自独立的子目录与 JSON 日志、截图等使用注意事项使用本项目的过程中请小心保存原始文件否则应用关闭时文件也会随之关闭导致数据丢失。项目启动后在程序截图期间不要关闭应用窗口否则可能导致截图失败或环境状态异常。延伸阅读数据流控制器与任务对象实现dataflow/data_flow_controller.py命令行入口与参数解析dataflow/dataflow.py模板选择流程SemanticSimilarity / LLM 双后端dataflow/instantiation/workflow/choose_template_flow.py预填充流程dataflow/instantiation/workflow/prefill_flow.py过滤流程dataflow/instantiation/workflow/filter_flow.py执行与评估流程dataflow/execution/workflow/execute_flow.py结果 Schema 定义dataflow/schema/instantiation_schema.json、dataflow/schema/execution_schema.json提示词模板目录dataflow/prompts/instantiation【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考