简介

改 LeRobot v3.0 数据集的任务文本（整体设一个默认任务，或按 episode 逐个覆盖），并把任务相关的四处元数据一起同步。包 lerobot datasets/dataset_tools.py:1404 modify_tasks（上游是原地改），默认先把数据集整目录拷到 output_path 再改拷贝，源保持不变。

功能描述

• 默认「拷贝后改」：output_path 非空时先 shutil.copytree 整份拷到目标再改目标，源数据集在 pipeline 里保持不可变；output_path 为空是文档化的原地改（超大数据集省一份拷贝），会打 warning
• 任务解析顺序沿用上游：某 episode 在 episode_tasks 里就用它，否则用 new_task，两者都没覆盖到就保留该 episode 原任务的第一条（原任务也为空则报错）
• 同步范围：meta/tasks.parquet、data/**/*.parquet 的 task_index、meta/episodes/**/*.parquet 的 tasks、meta/info.json 的 total_tasks
• 任务去重：所有 episode 的新任务取 sorted(set(...)) 重新编号，文本相同的任务共用一个 task_index
• 任务内容来自构造参数而不是数据列（它是配置，不是每行数据）；new_task 与 episode_tasks 至少给一个，都不给直接失败
• episode_tasks 里的 episode 索引越界由上游校验拦下，返回 Failed 状态行
• skip-if-done：拷贝模式下输出已是 v3.0 且 force=False 时不动手，返回 SKIPPED；原地模式没有这个判断
• 不重算 meta/stats.json（任务改动不影响数值统计）

算子参数

输入

输出

参数

注意事项

• task_index 会按新任务集合的字典序整体重编，与原编号没有对应关系；只改一个 episode 的任务也会让其他 episode 的 task_index 变。
• 每个 episode 的 tasks 被写成单元素列表，原本挂多个任务的 episode 会被压成一条。
• 拷贝模式会先删掉已存在的 output_path 再整目录拷贝，磁盘开销等于一份完整数据集（含视频），大数据集要预留空间。
• 改写阶段失败不会清理输出：拷出来的目录本身已经是合法 v3.0，下一次 force=False 会被 skip-if-done 当成已完成；重跑请显式 force=True。
• 原地模式（output_path 为 null）没有回滚，data parquet 是逐文件改写的，中途失败会留下部分改写的数据集。
• episode_tasks 的键是 episode 索引（int）。经 JSON 序列化传参时键会变成字符串，算子内部会 int(k) 还原。

调用示例

1from __future__ import annotations
2
3import os
4
5import daft
6from daft import col
7
8from daft.aihc.common.udf import aihc_udf
9from daft.aihc.functions.embodied.lerobot.modify_tasks_udf import ModifyTasks
10
11if __name__ == "__main__":
12    if os.getenv("DAFT_RUNNER", "native") == "ray":
13        import ray
14        ray.init(dashboard_host="0.0.0.0", ignore_reinit_error=True)
15        daft.set_runner_ray()
16    daft.set_execution_config(actor_udf_ready_timeout=6000, min_cpu_per_task=0)
17
18# 夹具可用 test/_helpers/lerobot_edit_fixture.py:build_tabular_dataset 生成
19    samples = {
20        "dataset_path": ["/path/to/lerobot_dataset"],
21        "output_path": ["/tmp/lerobot_tasks_out"],
22    }
23    ds = daft.from_pydict(samples)
24    ds = ds.with_column(
25        "result",
26        aihc_udf(
27            ModifyTasks,
28            construct_args={
29                "new_task": "统一任务",
30                "episode_tasks": {0: "特例"},
31                "force": False,
32            },
33            num_cpus=1,
34            concurrency=1,
35            batch_size=1,
36        )(col("dataset_path"), col("output_path")),
37    )
38    ds.show()