算法本身什么也不做，要等工作流来运行它。工作流是一张图：输入节点、你的步骤和输出节点，端口与端口相连。连接是唯一的边，运行顺序由「谁给谁供数据」自然决定，你从不需要手动安排。
这一页讲怎样在编辑器里搭建工作流、它背后的文件，以及决定每个步骤如何运行的规则。在画布背后，每个工作流都是一个 embodiflow-workflow/v1 格式的 YAML 文件，两种视图都可以编辑。
workflow:manage）；启动运行需要 workflow:run。在编辑器中搭建
打开「工作流」，进入项目工作流，或者从「新建筛选工作流」「新建导出工作流」开始。输入节点在左边，输出节点在右边。算法面板把内置算法和团队算法放在一起，按七个类别分组，也可以搜索。
添加和连接步骤的方式有好几种，哪种顺手用哪种：
| 连接方式 | 做法 |
|---|---|
| 放下算法卡片 | 把算法放到画布上，它会把主输入连到上游最近的匹配输出。编辑器会告诉你连了什么，可以撤销。 |
| 拖出连线 | 从端口拖到空白处，会出现适合该端口的算法列表供你搜索。 |
| 点击端口上的加号 | 选择「在这个输出后添加步骤」或「在这个输入前添加步骤」。 |
| 使用侧边面板 | 选中一个步骤，打开「连接」，点击「添加连接」，再点「选择一个输出」。不匹配的输出不会出现。 |
| 放到连线上 | 把算法放到一条连线上，或者点击「在这里插入步骤」，把它插到两端之间。 |
端口按数据类型着色。拖动时只有匹配的端口会亮起，被拒绝时会说明原因，例如会形成循环。把鼠标放在连线上，可以看到它传递的内容。检查器用来编辑参数、when 条件和 onFail。「YAML」视图编辑的是同一个文件，「可视化」切换回来，「立即整理」会重新排列画布。
问题、保存与发布
问题面板列出每个问题及其错误码。还有问题的草稿也会保存，你的工作不会丢，但在修复之前不能发布、绑定到项目或运行，编辑器会在「发布」旁边提示还有几个问题。完全无法解析的文件会被拒绝。
用样本测试
「用样本测试」会在项目里最多三条录制上运行草稿。「时间窗口（秒）」可选，只运行每条录制的开头，最多 600 秒。测试运行永远不保存结果：保存录制和评估质量的步骤会被跳过，显示为灰色的「已跳过：时间窗口测试运行不保存结果。」，不会变绿。每个步骤的输出仍然可以在运行里查看。
即使设置了时间窗口，标准化和原始文件质检也总是读取完整的上传，因为原始文件没有时间轴。
一个完整的工作流
下面是内置工作流 human-ego-standard 的原文。在写自己的工作流之前，读一读它很有帮助。
apiVersion: embodiflow-workflow/v1
name: Human egocentric standard
description: Pre-check, standardize, calibrate and anonymize; then head pose and depth from the calibration, MANO hands made metric by stereo hand keypoints, the SMPL-H body fitted to the hands, a ROS 2 3D view, visual quality, episode proposals and a success label per episode.
type: processing
params:
task: { type: string, default: "", title: Task text }
steps:
- id: pre_check
uses: embodiflow/raw-metrics@1
inputs: { raw: input.raw }
onFail: stop
- id: standardize
uses: auto
inputs: { raw: input.raw }
with: { task: "${{ params.task }}" }
- id: calibrate
uses: embodiflow/sensor-calibrate@1
inputs: { recording: standardize.recording }
- id: anonymize
uses: embodiflow/anonymize@1
inputs: { recording: standardize.recording }
- id: hand_pose
uses: embodiflow/hand-pose@1
inputs: { recording: anonymize.recording, calibration: calibrate.features }
with: { frame_stride: 6 }
- id: head_pose
uses: embodiflow/camera-pose-vio@1
inputs: { recording: anonymize.recording, calibration: calibrate.features }
- id: depth
uses: embodiflow/stereo-depth@1
inputs: { recording: anonymize.recording, calibration: calibrate.features }
- id: hands
uses: embodiflow/hand-mano@1
inputs: { recording: anonymize.recording, head_pose: head_pose.features, stereo_hands: hand_pose.features, depth: depth.features }
- id: body
uses: embodiflow/body-pose@1
inputs: { recording: anonymize.recording, head_pose: head_pose.features, hands: hands.features, depth: depth.features }
- id: view3d
uses: embodiflow/ego-vis@1
inputs: { recording: anonymize.recording, head_pose: head_pose.features, depth: depth.features, hands: hands.features, body: body.features }
- id: visual
uses: embodiflow/visual-metrics@1
inputs: { recording: anonymize.recording }
- id: propose
uses: embodiflow/propose-episodes@1
inputs: { recording: anonymize.recording }
- id: success
uses: embodiflow/episode-success@1
inputs: { recording: anonymize.recording, episodes: propose.episodes }
output:
recording: anonymize.recording
features: [calibrate.features, head_pose.features, depth.features, hands.features, body.features, view3d.features, success.features]
checks: [pre_check.checks, anonymize.checks, calibrate.checks, hand_pose.checks, head_pose.checks, depth.checks, hands.checks, body.checks, view3d.checks, visual.checks]
episodes: propose.episodes| 字段 | 规则 |
|---|---|
type | processing、curate 或 export。它决定输入节点提供什么、输出节点接收什么。 |
params | 运行参数：{ 名称: { type: string、number 或 boolean, default, title } }。步骤只能把它整体作为 with 的取值。 |
steps[].id | 小写字母、数字和 _，最长 32，唯一，且不能是 input、output 或 params。 |
steps[].uses | embodiflow/<name>@<major>、team/<name>@<major>，标准化步骤也可以写 auto。 |
steps[].inputs | 每个必需输入都要连接：<port>: <stepId>.<port> 或 input.<port>。 |
steps[].with | 参数取值，按算法的参数检查。 |
steps[].when | 可选的条件。 |
steps[].onFail | 仅质检步骤：continue（默认）或 stop。 |
steps[].enabled | 默认 true。停用的步骤保留但被忽略。 |
output | 输出节点接收的内容。 |
没有 needs、if、备选列表或锁定的步骤。节点位置保存在单独的布局文档里，所以在画布上挪动方框不会改变工作流。
可以使用的内置算法
步骤通过引用来指定内置算法，名称就是它在控制台里的标题。
| 引用 | 名称 | 类别 | 级别 |
|---|---|---|---|
embodiflow/anonymize@1 | 脱敏 | transform | 录制 |
embodiflow/body-pose@1 | 全身姿态（SMPL-H） | enrich | 录制 |
embodiflow/camera-pose-vio@1 | 头部位姿（视觉惯性） | enrich | 录制 |
embodiflow/canonicalize@1 | 标准化 MCAP | standardize | 录制 |
embodiflow/canonicalize-video@1 | 标准化视频 | standardize | 录制 |
embodiflow/ego-vis@1 | 第一视角 3D 可视化 | enrich | 录制 |
embodiflow/episode-success@1 | 片段是否成功 | enrich | 片段 |
embodiflow/filter-episodes@1 | 筛选片段 | curate | 数据集版本 |
embodiflow/hand-mano@1 | 手部姿态（MANO） | enrich | 录制 |
embodiflow/hand-pose@1 | 手部姿态 | enrich | 录制 |
embodiflow/lerobot-export@1 | LeRobot v3 导出 | export | 数据集版本 |
embodiflow/lerobot-preview@1 | 视频预览 | enrich | 录制 |
embodiflow/propose-episodes@1 | 提议任务片段 | segment | 录制 |
embodiflow/raw-metrics@1 | 原始文件指标 | check | 录制 |
embodiflow/recording-episodes@1 | 录制自带片段 | segment | 录制 |
embodiflow/sensor-calibrate@1 | 传感器标定 | enrich | 录制 |
embodiflow/split-episodes@1 | 划分数据集 | curate | 数据集版本 |
embodiflow/stereo-depth@1 | 双目深度 | enrich | 录制 |
embodiflow/visual-metrics@1 | 视觉指标 | check | 录制 |
输入与输出节点
两端做什么，取决于工作流的类型：
| 工作流类型 | 输入提供 | 输出接收 | 输出节点做什么 |
|---|---|---|---|
processing | raw | recording（必填）、features[]、checks[]、episodes | 把特征合并进录制，保存质检结果并评估质量规则，按 episodes 创建任务片段（否则每条录制一个），不改动人工创建的任务片段，并在脱敏变换之后写入隐私证据。 |
curate | version | selection[]，至少一个 | 合并各个选择，创建新的数据集版本，并关联源版本和这次运行。 |
export | version | artifact（必填） | 保存文件并提供下载。 |
输出节点以一组步骤的形式运行，你不用放置它们，但能在运行视图里看到：合并特征（只在连接了特征时才有），然后是保存结果、逐条录制评估质量，以及每次运行一次的报告。保存步骤成功，录制才算保存。读取上传文件及其事实发生在输入节点之前。两个步骤写出同名特征时，会被 COMPOSE_FEATURE_CONFLICT 拒绝。
连接步骤
编辑器、保存和运行使用同一套规则，所以在编辑器里看起来没问题的工作流，运行时不会给你意外。每次拒绝都会说明原因。
| 规则 | 错误 |
|---|---|
| 数据类型相同。 | WORKFLOW_PORT_INCOMPATIBLE |
内容匹配：输入的每个 requires 条目都由上游声明。 | WORKFLOW_CONTENT_MISSING |
| 没有环。 | WORKFLOW_CYCLE |
| 每个必需输入都已连接。一个输出可以连接任意多个输入。 | WORKFLOW_INPUT_UNWIRED |
只有输出节点的 features、checks 和 selection 接受多个连接。步骤的每个输入恰好接一个。 | 被拒绝，并说明原因 |
processing 工作流的 output.recording 向上追溯到恰好一个 standardize 步骤。 | WORKFLOW_NO_STANDARDIZE |
任务片段级步骤的 episodes 输入与 output.episodes 来自同一个上游。 | WORKFLOW_EPISODES_MISMATCH |
运行之前检查的不只是类型，还有内容。features 输出带有其算法声明的特征；recording 输出带有上游的内容，加上每个变换声明的改动；episodes 输出带有声明的标签。在上面的例子里，hand_pose 需要 calibration.stereo，而 calibrate 声明了它，所以这条连接是允许的。
条件
when 让步骤按条件运行。它是一组条件，必须全部成立：
| 条件 | 示例 |
|---|---|
| 事实 | { fact: duration_s, op: ">", value: 60 } |
| 运行参数 | { param: run_depth, op: "==", value: true } |
| 质检结果 | { check: pre_check.frame_gap_ratio, status: [pass, warn] }。步骤会等待该质检完成。 |
op 可以是 ==、!=、>、>=、<、<= 和 in。条件不成立时，步骤以「已跳过：条件不成立」结束，在下游按「不适用」处理。这和算法自己的 requires 不同：后者是在数据缺少所需内容时，让步骤以「不适用」结束。
步骤没有成功时
这一切背后只有一条规则：每个必需输入都有值，步骤才会运行。其余的都由它推出来。
- 失败或不适用的步骤不产生任何输出。
- 必需输入缺失是因为上游「不适用」，该步骤同样「不适用」。
- 必需输入缺失是因为上游失败，该步骤以「已跳过：上游失败」结束，并指出是哪个上游。
- 上游没有成功时，可选输入就视为缺席。
- 在处理工作流中，输出节点的
recording是必需的，features、checks和episodes是可选的。所以手部姿态步骤失败时，录制仍然会保存，只是没有手部姿态：录制结果为成功，运行为部分成功，失败的步骤保持红色。
停止型质检
给质检步骤设置 onFail: stop，它就成了一道关卡。该质检任何一条结果为 fail 时，这条录制就会停止：尚未开始的步骤以「已跳过：被质检停止」结束，也不会保存任何内容（WORKFLOW_STOPPED_BY_CHECK）。warn、error、not_evaluated 和 not_applicable 既不会停止录制，也不算通过。
- id: pre_check
uses: embodiflow/raw-metrics@1
inputs: { raw: input.raw }
onFail: stopraw。凡不在停止型质检上游的步骤都会等它完成，所以关卡越早，省下的工作越多。任务片段级步骤
level: episode 的 check 或 enrich 步骤，对其 episodes 输入的每个任务片段并行运行一次：同时最多 4 个，总共最多 100 个（超过会报 WORKFLOW_FAN_OUT_TOO_LARGE）。
一个任务片段失败不会连累其他任务片段。只要至少有一个成功，步骤就算成功；只有全部失败才算失败，并保留「N 个中有 M 个失败」的计数；运行为部分成功，并列出失败的任务片段。有任务片段失败的步骤不会写入缓存，所以重新运行时会再试那些任务片段。没有任务片段时，步骤被跳过，按「不适用」处理，所以输出节点仍然会保存。
任务片段级的停止型质检不会停止整条录制，只会把失败的任务片段从保存的结果中去掉。任务片段级步骤的 checks 不能连到输出节点的 checks，但它的特征仍会通过合并进入保存的录制。在上面的例子里，success 接收 propose.episodes，同一组任务片段也是工作流的输出。
隐私
在处理工作流中，从 standardize 步骤到任何读取 recording 的步骤、以及到 output.recording 的每条路径，都必须经过内置的 anonymize 步骤。例外是质检，以及输出只有数字的内置算法。绕过它的路径会被 WORKFLOW_PRIVACY_BARRIER 拒绝。录制不需要隐私保护时，anonymize 会原样交出录制，所以同一个工作流可以同时适用于两种情况。
读取原始文件、或者读取尚未脱敏的录制的团队代码，只会为有权读取原始录制的人运行；其他人会被 WORKFLOW_RAW_CODE_AUTHORITY_DENIED 拒绝。
整理与导出工作流
整理工作流从一个数据集版本开始，以一个新版本结束。各步骤的 selection 端口都连到输出节点，只有每个选择都保留的任务片段才会被保留。
apiVersion: embodiflow-workflow/v1
name: Curate standard
description: Drop episodes by quality and duration, then split the rest into train and validation by session.
type: curate
steps:
- id: filter
uses: embodiflow/filter-episodes@1
inputs: { version: input.version }
- id: split
uses: embodiflow/split-episodes@1
inputs: { version: input.version }
output:
selection: [filter.selection, split.selection]导出工作流从一个数据集版本开始，以文件结束。
apiVersion: embodiflow-workflow/v1
name: Export LeRobot v3
description: Export the dataset version as a LeRobot v3 dataset.
type: export
steps:
- id: export
uses: embodiflow/lerobot-export@1
inputs: { version: input.version }
output:
artifact: export.artifact在版本上运行
在数据集版本页面，「筛选和导出」列出项目的整理和导出工作流。点击「运行」，就在该版本上启动整理工作流。运行会报告保留和丢弃了多少任务片段，并给出新版本的链接，或者告诉你不需要新版本。导出工作流不在这里启动，请使用版本的导出操作。
- 整理运行会创建一个新的、不可变的数据集版本。源版本不会被改动，新版本会记录它的源版本和这次运行。
- 手动启动或重新运行整理，需要有运行工作流和写入项目数据的权限；导出工作流需要有运行工作流和写入导出的权限。
- 一个项目绑定一个整理工作流。打开「创建版本时自动运行」后，每个新版本都会先启动整理工作流，再启动导出工作流。整理运行创建的版本不会再次启动同一个整理工作流，所以不会循环；某个工作流无法启动，也不会挡住另一个。
- 有问题的工作流或未发布的修订，不能绑定，也不能运行。
内置模板
模板是只读的。想修改的话，用「创建副本继续编辑」复制一份。
| 模板 | 类型 | 步骤 |
|---|---|---|
robot-standard（机器人遥操作（标准）） | processing | 可停止运行的原始文件指标、标准化、脱敏、视觉指标，以及录制自带片段：按规则保存录制自带的任务片段，不用模型。连续录制没有任务片段，交付按绑定的任务片段，否则为整条录制。 |
human-ego-standard（第一人称录制（标准）） | processing | 可停止运行的原始文件指标、标准化、传感器标定和脱敏，然后是手部姿态、头部位姿（视觉惯性）和双目深度，视觉指标、提议任务片段，以及每个任务片段的「片段是否成功」。 |
curate-standard（整理（标准）） | curate | 筛选片段、划分数据集。 |
export-lerobot-v3（导出（LeRobot v3）） | export | LeRobot v3 导出。 |
机器人遥操作（标准）
apiVersion: embodiflow-workflow/v1
name: Robot standard
description: Pre-check the upload, standardize it, anonymize when the recording requires privacy, measure visual quality, and save the recording's own episodes.
type: processing
params:
task: { type: string, default: "", title: Task text }
steps:
- id: pre_check
uses: embodiflow/raw-metrics@1
inputs: { raw: input.raw }
onFail: stop
- id: standardize
uses: auto
inputs: { raw: input.raw }
with: { task: "${{ params.task }}" }
- id: anonymize
uses: embodiflow/anonymize@1
inputs: { recording: standardize.recording }
- id: visual
uses: embodiflow/visual-metrics@1
inputs: { recording: anonymize.recording }
- id: recording_episodes
uses: embodiflow/recording-episodes@1
inputs: { recording: anonymize.recording }
with: { task: "${{ params.task }}" }
output:
recording: anonymize.recording
checks: [pre_check.checks, anonymize.checks, visual.checks]
episodes: recording_episodes.episodes