EmbodiFlow文档
开始使用数据工作流与算法训练与部署开发者参考
  • 工作流与算法

工作流

  • 认识工作流
  • 搭建自己的工作流
  • 运行与日志

算法

  • 算法目录
  • 编写自定义算法

算法开发指南

  • 第一个算法
    • 标准化
    • 质检
    • 变换
    • 增强
    • 切分
    • 整理
    • 导出
  • 镜像算法
  • 在工作流中使用算法
  • 接口参考
  • SDK 参考
  • 命令行
  • MCP 工具
  • 问题排查

© 2026 EmbodiFlow

llms.txtOpenAPI JSON
  1. 工作流与算法
  2. 算法开发指南
  3. 第一个算法

第一个算法

先弄清自定义算法是怎么组织的,再用十分钟左右写出一个完整的算法,在真实数据上试跑并保存。

内置算法已经覆盖了常见的工作:标准化上传文件、质检、脱敏、手部姿态、任务片段和 LeRobot v3 导出。当你的数据或团队需要它们做不到的事情时,就自己写一个算法,放进工作流,和内置算法排在一起。你的算法和内置算法的运行方式完全相同。

这一页先讲清几个必须知道的概念,然后带你从空白编辑器开始,写出一个真实的算法并保存为版本。

测试阶段。自定义算法和工作流已经在产品里,可以直接试用,但还没有在真实数据上完成完整验收。可能会遇到不完善的地方,依赖结果之前请先核对。

一个算法只做一件事

每个算法都属于七个类别之一。类别说明算法做什么,也决定了它读取什么、写出什么,所以你不需要自己声明主端口。先选好类别,其余的就定下来了。

类别用途读取写出
标准化把自有格式的上传文件转成标准录制。rawrecording
质检判断数据,但不修改数据。raw 或 recordingchecks
变换修改录制:裁剪首尾、裁切画面、去掉一路相机。recordingrecording
增强增加新的数值,例如位姿或标签。recordingfeatures
切分给出任务片段建议。recordingepisodes
整理决定新的数据集版本保留哪些任务片段。versionselection
导出把数据集版本写成可下载的文件。versionartifact

大多数算法每条录制运行一次。质检和增强算法也可以改为每个任务片段运行一次,并行执行;整理和导出算法则每个数据集版本运行一次。除了主输入,算法还可以接收额外的 features 输入(例如标定结果),多数类别还可以额外写出 checks。

「读取」和「写出」两列里的八种数据类型,决定了工作流里步骤之间怎么连接:一个输出只能连到同类型的输入。每种类型的含义见接口参考。

在哪里编写

最简单的是浏览器:编辑器为每个类别准备了可运行的起步代码,输入时会补全 SDK,还能直接在录制上试跑。本页就用它。

如果你习惯自己的编辑器,命令行可以在本地完成同样的步骤,AI 助手也可以通过 MCP 工具来操作。不是 Python 的代码,可以打包成容器镜像,见镜像算法。

团队所有者、团队管理员和研究员可以编写和保存算法。

写一个运动强度算法

你要写的是一个增强算法:计算第一路相机画面在相邻两帧之间变化了多少。数值低的地方就是停顿,之后的步骤可以据此切分或丢弃。开始之前,你需要一个至少有一条已处理录制的项目;还没有的话,先看上传录制。

1新建算法

在侧边栏打开「算法」,点击「新建算法」。

2选择它做什么

在「它做什么?」下选择「增强」,在「在哪一级运行?」下保持「录制」。点击「开始编写」,编辑器会打开,里面已经是一段可运行的起步代码。

3粘贴代码

把起步代码替换成下面这段。它就是 SDK 里的示例 packages/sdk-python/examples/v1/enrich-motion-energy/main.py。

"""Enrich: the mean absolute difference between consecutive frames of the first camera, per frame.

In:  the canonical recording.
Out: one new per-frame feature "motion_energy"; the recording itself is not changed.
Why: idle stretches (the arm waits, the person stands still) show as low motion, so a later step can
     segment or drop them. A feature is how an algorithm adds knowledge without rewriting the recording.
"""

from dataclasses import dataclass

import numpy as np

from embodiflow import Context, Feature, FeatureWriter, Recording, algorithm, param


@dataclass
class Params:
    step: int = param(1, minimum=1, title="Frame step")


@algorithm.enrich(
    name="motion-energy",
    title="Motion energy",
    description="How much the first camera's picture changes from one frame to the next.",
    features=[Feature("motion_energy", type="scalar", per="frame", dtype="float32", unit="none")],
    params=Params,
)
def run(recording: Recording, out: FeatureWriter, ctx: Context[Params]) -> None:
    cam = recording.cameras[0]
    energy = np.zeros((recording.num_frames, 1), np.float32)
    prev = None
    for frame in cam.frames(step=ctx.params.step, gray=True):
        current = frame.image.astype(np.float32)
        if prev is not None:
            energy[frame.index, 0] = np.abs(current - prev).mean()
        prev = current
    out.per_frame("motion_energy", energy)

它的测试是 packages/sdk-python/examples/v1/enrich-motion-energy/test_main.py:用一份很小的测试数据,通过真实的运行器运行这段代码,并检查写出的值。想从它开始,运行 embodiflow algorithm init my-algorithm --category enrich,它会写出 main.py 和 test_main.py;再运行 embodiflow algorithm test my-algorithm(需要装有 pytest 和 SDK 的 Python)。

这里有四个部分值得留意。Params 是工作流可以修改的设置,这里的 step 是取帧间隔,至少为 1。装饰器 @algorithm.enrich(...) 就是接口:它声明了一个特征 motion_energy,每帧一个数。run 函数负责具体的工作,参数按名称绑定,需要哪个就写哪个,顺序随意。最后,out.per_frame(...) 写出特征,每帧恰好一行。

4检查接口

右侧的「接口」面板会随着输入实时更新。确认「特征」里有 motion_energy,代码里也没有哪一行被标出问题。

平台读取装饰器时不会运行你的代码,所以装饰器参数和参数默认值都必须是字面量:字符串、数字、列表,以及 Feature、param 这类 SDK 构造器。

5在录制上试跑

在「试跑」里选择「项目」和「录制」。「范围」默认是「前 30 秒」,选「整条录制」就会跑完整条。点击「开始试跑」,完成后会出现 motion_energy 的曲线。

试跑用真实的运行环境处理真实数据,但不会保存任何内容,范围最长 600 秒。如果试跑无法启动,编辑器会写明原因,你仍然可以保存算法。

6保存版本

点击右上角的「保存版本」。第一个版本是 1.0.0,保存后的版本不会再改变。以后保存修改时,平台会替你建议下一个版本号:接口变了,比如端口、声明的特征或参数类型变了,就需要升主版本。

放进工作流

保存好的版本可以作为步骤,用在团队的任何工作流里。下面把运动强度加到一份标准工作流的副本中:

  1. 打开「工作流」,找到「机器人遥操作(标准)」或「第一人称录制(标准)」,点击「创建副本继续编辑」,选择你的项目,再点击「创建副本」。
  2. 在编辑器的「搜索算法」里输入 motion-energy 并添加。团队算法带有「团队」标记。编辑器会把它的 recording 输入连到最近的匹配输出。
  3. 选中这个步骤,打开「连接」,把脱敏步骤的输出选为它的录制。标准化之后,每个读取录制的步骤都必须读取脱敏后的录制,否则工作流会被 WORKFLOW_PRIVACY_BARRIER 拒绝。
  4. 把这个步骤的 features 输出连到输出节点,特征就会合并进保存的录制。
  5. 「校验问题」清空后,点击「用样本测试」,选一条录制,把「时间窗口(秒)」设得短一些,比如 30,再点击「运行测试」。这次测试同样不保存任何内容。
  6. 点击「发布」并确认,然后把工作流绑定到项目,之后的新上传就会用它处理。

把工作流绑定到项目,需要该项目的工作流管理权限(workflow:manage);启动运行需要 workflow:run。编辑器的完整用法见在工作流中使用算法。

在命令行里完成同样的事

如果你更喜欢在本地工作,下面四条命令做的是同一件事:init 写出起步的 main.py 和它的测试,recordings pull 下载一条已处理的录制用来试跑,save 把结果上传为新版本。

embodiflow algorithm init motion-energy --category enrich
embodiflow recordings pull <recording-id> --project <team>/<project> --as recording --out ./rec
embodiflow algorithm test motion-energy --input recording=./rec --window 30s
embodiflow algorithm save motion-energy

现在你有了什么

保存的版本会出现在算法库里,带「团队」标记,工作流用 team/motion-energy@1 引用它。它的特征以团队专属的名称保存,不会和内置特征重名。

如果某一行被标出 ALGORITHM_FILE_INVALID,说明某个装饰器参数不是字面量,或者违反了某条规则,那一行的提示会写明是哪条。如果保存时报 FEATURE_ROW_COUNT_MISMATCH,请确保每帧恰好写一行。其他错误码见问题排查。

下一步

  • 增强算法:进一步了解特征
  • 在工作流中使用算法
  • 接口参考
上一页

编写自定义算法

下一页

标准化