LeRobot_v3.0数据集合并
更新时间：2026-07-29
简介
把多个已完成的 LeRobot v3.0 数据集合并为一个 v3.0 数据集，AIHC UDF 封装，依赖 lerobot（内部走 merge_lerobot_v30.merge_datasets → aggregate_datasets/aggregate_videos，重索引 episode/frame/task/video）+ ffmpeg，用于把多来源数据集聚合为统一训练集。
功能描述
• 多入一出：每行描述一个 merge job，input_paths（显式列表）与 base_path（父目录递归扫描）二选一，输出到 output_path
• 输入校验：合并前校验每个输入是已完成的 v3.0 数据集（meta/info.json 且 codebase_version==v3.0），占位路径会被拒绝
• dry-run：dry_run=True 仅校验输入并报告计划而不写出（返回 SUCCESS: dry-run）
• 幂等跳过：输出已是 v3.0 且非 force/非 dry-run 时跳过（返回 SKIPPED）
算子参数
输入
| 输入 | 含义 |
|---|---|
| input_paths | 待合并的 v3.0 数据集根目录列表（list |
| base_path | 递归扫描其下 v3.0 数据集的父目录（string）；与 input_paths 互斥。 |
| output_paths | 合并后数据集的输出目录（string）。 |
输出
| 输出 | 含义 |
|---|---|
| result | 字符串（large_string）："SUCCESS" / "SKIPPED: has already been v3.0" / "SUCCESS: dry-run" / "Failed: [output] <ErrorType>: <msg>"。 |
参数
| 参数名称 | 类型 | 默认值 | 描述 | |
|---|---|---|---|---|
| aggr_repo_id | str | merged | 写入合并后元数据的逻辑 repo id | |
| force_merge | bool | False | 即使 output_path 已是 v3.0 也强制重新合并 | |
| dry_run | bool | False | 仅校验输入并报告合并计划，不写出任何内容 | |
| data_files_size_in_mb | float |None | None | 转发给底层聚合器的数据分片大小（None 用 lerobot 默认） | |
| video_files_size_in_mb | float |None | None | 转发给底层聚合器的视频分片大小（None 用 lerobot 默认） | |
| chunk_size | int |None | None | 转发给底层聚合器的 chunk 大小（None 用 lerobot 默认） |
调用示例
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.merge_lerobot_v30_udf import MergeLeRobotV30
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 BASE_PATH = "/mnt/pfs/merge_src" # MOCK，其下含 >=2 个 v3.0 数据集
16 OUTPUT_PATH = "/mnt/pfs/merge_out" # MOCK
17 samples = {
18 "input_paths": [None], # 用 base_path 模式，input_paths 置空
19 "base_path": [BASE_PATH],
20 "output_paths": [OUTPUT_PATH],
21 }
22 ds = daft.from_pydict(samples)
23 ds = ds.with_column(
24 "result",
25 aihc_udf(
26 MergeLeRobotV30,
27 construct_args={"aggr_repo_id": "merged", "force_merge": True},
28 num_cpus=1, concurrency=1, batch_size=1,
29 )(col("input_paths"), col("base_path"), col("output_paths")),
30 )
31 ds.show()评价此篇文章