内置算法已经覆盖了常见的工作:标准化上传文件、质检、脱敏、手部姿态、任务片段和 LeRobot v3 导出。当你的数据或团队需要它们做不到的事情时,就自己写一个算法,放进工作流,和内置算法排在一起。你的算法和内置算法的运行方式完全相同。
这一页先讲清几个必须知道的概念,然后带你从空白编辑器开始,写出一个真实的算法并保存为版本。
一个算法只做一件事
每个算法都属于七个类别之一。类别说明算法做什么,也决定了它读取什么、写出什么,所以你不需要自己声明主端口。先选好类别,其余的就定下来了。
| 类别 | 用途 | 读取 | 写出 |
|---|---|---|---|
| 标准化 | 把自有格式的上传文件转成标准录制。 | raw | recording |
| 质检 | 判断数据,但不修改数据。 | raw 或 recording | checks |
| 变换 | 修改录制:裁剪首尾、裁切画面、去掉一路相机。 | recording | recording |
| 增强 | 增加新的数值,例如位姿或标签。 | recording | features |
| 切分 | 给出任务片段建议。 | recording | episodes |
| 整理 | 决定新的数据集版本保留哪些任务片段。 | version | selection |
| 导出 | 把数据集版本写成可下载的文件。 | version | artifact |
大多数算法每条录制运行一次。质检和增强算法也可以改为每个任务片段运行一次,并行执行;整理和导出算法则每个数据集版本运行一次。除了主输入,算法还可以接收额外的 features 输入(例如标定结果),多数类别还可以额外写出 checks。
「读取」和「写出」两列里的八种数据类型,决定了工作流里步骤之间怎么连接:一个输出只能连到同类型的输入。每种类型的含义见接口参考。
在哪里编写
最简单的是浏览器:编辑器为每个类别准备了可运行的起步代码,输入时会补全 SDK,还能直接在录制上试跑。本页就用它。
如果你习惯自己的编辑器,命令行可以在本地完成同样的步骤,AI 助手也可以通过 MCP 工具来操作。不是 Python 的代码,可以打包成容器镜像,见镜像算法。
写一个运动强度算法
你要写的是一个增强算法:计算第一路相机画面在相邻两帧之间变化了多少。数值低的地方就是停顿,之后的步骤可以据此切分或丢弃。开始之前,你需要一个至少有一条已处理录制的项目;还没有的话,先看上传录制。
新建算法
在侧边栏打开「算法」,点击「新建算法」。
选择它做什么
在「它做什么?」下选择「增强」,在「在哪一级运行?」下保持「录制」。点击「开始编写」,编辑器会打开,里面已经是一段可运行的起步代码。
粘贴代码
把起步代码替换成下面这段。它就是 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(...) 写出特征,每帧恰好一行。
检查接口
右侧的「接口」面板会随着输入实时更新。确认「特征」里有 motion_energy,代码里也没有哪一行被标出问题。
平台读取装饰器时不会运行你的代码,所以装饰器参数和参数默认值都必须是字面量:字符串、数字、列表,以及 Feature、param 这类 SDK 构造器。
在录制上试跑
在「试跑」里选择「项目」和「录制」。「范围」默认是「前 30 秒」,选「整条录制」就会跑完整条。点击「开始试跑」,完成后会出现 motion_energy 的曲线。
试跑用真实的运行环境处理真实数据,但不会保存任何内容,范围最长 600 秒。如果试跑无法启动,编辑器会写明原因,你仍然可以保存算法。
保存版本
点击右上角的「保存版本」。第一个版本是 1.0.0,保存后的版本不会再改变。以后保存修改时,平台会替你建议下一个版本号:接口变了,比如端口、声明的特征或参数类型变了,就需要升主版本。
放进工作流
保存好的版本可以作为步骤,用在团队的任何工作流里。下面把运动强度加到一份标准工作流的副本中:
- 打开「工作流」,找到「机器人遥操作(标准)」或「第一人称录制(标准)」,点击「创建副本继续编辑」,选择你的项目,再点击「创建副本」。
- 在编辑器的「搜索算法」里输入
motion-energy并添加。团队算法带有「团队」标记。编辑器会把它的recording输入连到最近的匹配输出。 - 选中这个步骤,打开「连接」,把脱敏步骤的输出选为它的录制。标准化之后,每个读取录制的步骤都必须读取脱敏后的录制,否则工作流会被
WORKFLOW_PRIVACY_BARRIER拒绝。 - 把这个步骤的
features输出连到输出节点,特征就会合并进保存的录制。 - 「校验问题」清空后,点击「用样本测试」,选一条录制,把「时间窗口(秒)」设得短一些,比如 30,再点击「运行测试」。这次测试同样不保存任何内容。
- 点击「发布」并确认,然后把工作流绑定到项目,之后的新上传就会用它处理。
把工作流绑定到项目,需要该项目的工作流管理权限(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,请确保每帧恰好写一行。其他错误码见问题排查。