AgiBot转LeRobot_v3.0
更新时间：2026-07-29
简介
把 AgiBot World 数据集目录转换为 LeRobot v3.0 数据集，AIHC UDF 封装，依赖 lerobot + ffmpeg（内部走 convert_agibot_to_lerobot_v30.convert_dataset 的流式 H.264 编码、多线程解码 8 路相机、episode 级多进程 + aggregate_datasets 合并），用于把 AgiBot 采集数据接入标准 LeRobot v3.0 流水线。
功能描述
• 逐行转换：每行处理一个源数据集，源目录为 <input_path>/<input_repoid>，产物写到 <output_path>/<input_repoid>
• 视频编码：vcodec 默认 h264（libx264 CPU 编码，兼容性好），auto 映射为 h264 以避免误选 nvenc
• 并行解码：decode_threads 并发解码 8 路源 mp4，num_workers 控制 episode 级进程数
• 幂等跳过：目标已是 v3.0 时默认跳过（返回 SKIPPED），force_conversion=True 可强制重转
算子参数
输入
| 输入 | 含义 |
|---|---|
| input_repoid_col | 数据集逻辑名（string），如 agibot/digital_world，写入 v3 元数据并作为子目录名。 |
| input_paths | 源数据集父目录（string）；真实源在 <input_paths>/<input_repoid>（需含 meta_info/ 与 observations/）。 |
| output_paths | 输出父目录（string）；产物写到 <output_paths>/<input_repoid>。 |
输出
| 输出 | 含义 |
|---|---|
| result | 字符串（large_string）："SUCCESS" / "SKIPPED: has already been v3.0" / "Failed: [repo] <ErrorType>: <msg>"。 |
参数
| 参数名称 | 类型 | 默认值 | 描述 |
|---|---|---|---|
| num_workers | int | 4 | 每个数据集的 episode 级进程数，<=1 时内联执行 |
| vcodec | str | h264 | 视频编码器，默认 h264（libx264）；h264_nvenc 需匹配 NVENC SDK 的驱动；auto 映射为 h264 |
| decode_threads | int | 8 | 并发解码 8 路源 mp4 的线程数 |
| force_conversion | bool | False | 即使输出已是 v3.0 也强制重新转换 |
调用示例
Python
1from __future__ import annotations
2import os
3import daft
4from daft import col
5from daft.aihc.common.udf import aihc_udf
6from daft.aihc.functions.embodied.convert_agibot_to_lerobot_v30_udf import ConvertAgibotToLeRobotV30
7
8if __name__ == "__main__":
9 if os.getenv("DAFT_RUNNER", "native") == "ray":
10 import ray
11 ray.init(dashboard_host="0.0.0.0", ignore_reinit_error=True)
12 daft.set_runner_ray()
13 daft.set_execution_config(actor_udf_ready_timeout=6000, min_cpu_per_task=0)
14
15 INPUT_PATH = "/mnt/pfs/agibot_datasets" # MOCK
16 OUTPUT_PATH = "/mnt/pfs/agibot_v30_out" # MOCK
17 samples = {
18 "input_repoid_col": ["sample_dataset"],
19 "input_paths": [INPUT_PATH],
20 "output_paths": [OUTPUT_PATH],
21 }
22 ds = daft.from_pydict(samples)
23 ds = ds.with_column(
24 "result",
25 aihc_udf(
26 ConvertAgibotToLeRobotV30,
27 construct_args={
28 "num_workers": 4,
29 "vcodec": "h264",
30 "decode_threads": 8,
31 "force_conversion": True,
32 },
33 num_cpus=1, concurrency=1, batch_size=1,
34 )(col("input_repoid_col"), col("input_paths"), col("output_paths")),
35 )
36 ds.show()评价此篇文章