Побудова потокового конвеєра навчання роботів із використанням NVIDIA Cosmos3-DROID
У цьому навчальному посібнику ми розробляємо наскрізний потоковий конвеєр навчання роботів на основі набору даних NVIDIA Cosmos3-DROID, не завантажуючи його репозиторій обсягом 707 ГБ локально. Спочатку ми досліджуємо структуру LeRobotDataset v3.0 і будуємо граф метаданих на основі info.json, метаданих завдань, таблиць епізодів і статистики набору даних, а потім використовуємо доступ до HTTP-діапазонів байтів із PyArrow для вибіркового читання груп рядків і стовпців Parquet. Ми перетворюємо окремі епізоди на траєкторії стан–дія та аналізуємо рух суглобів, події захвата, декартові траєкторії кінцевого ефектора й спектри частот дій, перш ніж декодувати лише потрібні відеовікна AV1 за допомогою доступу на основі позиціонування через PyAV/FFmpeg. Далі ми нормалізуємо спостереження та дії, використовуючи статистику набору даних, створюємо фрагментований набір даних PyTorch у стилі ACT з опційним візуальним кондиціюванням і навчаємо мультимодальну політику клонування поведінки. Нарешті, ми оцінюємо навчену політику за допомогою розгортання у відкритому циклі з часовим ансамблюванням фрагментів дій, обчислюємо MSE і R2 для кожного суглоба порівняно з базовою моделлю середньої дії, візуалізуємо передбачені дії та дії з еталонної траєкторії й зберігаємо повну контрольну точку політики для подальшого використання.
import subprocess, sys, os, json, math, time, warnings, random, tempfile
warnings.filterwarnings("ignore")
subprocess.run([sys.executable, "-m", "pip", "install", "-q",
"huggingface_hub>=0.34.0", "pyarrow>=15.0", "av>=12.0",
"pandas", "matplotlib", "tqdm"], check=False)
import numpy as np, pandas as pd, pyarrow as pa, pyarrow.parquet as pq
import matplotlib.pyplot as plt
from huggingface_hub import HfApi, HfFileSystem, hf_hub_download, hf_hub_url
import torch, torch.nn as nn, torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader
REPO_ID = "nvidia/Cosmos3-DROID"
ROOT = "success"
VIDEO_KEY = "observation.image.wrist_image_left"
FPS = 15
N_EPISODES = 48
HORIZON = 8
OBS_HISTORY = 2
USE_VISION = True
N_VIS_EPS = 6
VIS_SIZE = 96
EPOCHS = 12
BATCH = 256
SEED = 0
random.seed(SEED); np.random.seed(SEED); torch.manual_seed(SEED)
DEV = "cuda" if torch.cuda.is_available() else "cpu"
print(f"[env] torch={torch.__version__} device={DEV}")
if os.environ.get("HF_TOKEN"):
from huggingface_hub import login; login(os.environ["HF_TOKEN"])
api = HfApi()
fs = HfFileSystem()
HFS = lambda rel: f"datasets/{REPO_ID}/{rel}"
URL = lambda rel: hf_hub_url(REPO_ID, rel, repo_type="dataset")
print("\n" + "="*78 + "\n1. REPO INTROSPECTION\n" + "="*78)
all_files = api.list_repo_files(REPO_ID, repo_type="dataset")
print(f"total files in repo : {len(all_files):,}")
for prefix in ("success/data", "success/videos", "success/meta",
"failure/data", "failure/videos", "failure/meta"):
print(f" {prefix:<18} {sum(f.startswith(prefix) for f in all_files):>6,} files")
data_shards = sorted(f for f in all_files if f.startswith(f"{ROOT}/data/") and f.endswith(".parquet"))
vid_shards = sorted(f for f in all_files if f.startswith(f"{ROOT}/videos/{VIDEO_KEY}/"))
meta_files = sorted(f for f in all_files if f.startswith(f"{ROOT}/meta/"))
print(f"\n[{ROOT}] data shards={len(data_shards)} video shards({VIDEO_KEY})={len(vid_shards)}")
print("first data shard :", data_shards[0])
print("first video shard:", vid_shards[0])
print("\n" + "="*78 + "\n2. METADATA\n" + "="*78)
info = json.load(open(hf_hub_download(REPO_ID, f"{ROOT}/meta/info.json", repo_type="dataset")))
print(f"episodes={info.get('total_episodes'):,} frames={info.get('total_frames'):,} "
f"tasks={info.get('total_tasks'):,} fps={info.get('fps')}")
print("data_path template :", info.get("data_path"))
print("video_path template:", info.get("video_path"))
FEATURES = info["features"]
state_keys = sorted(k for k in FEATURES if k.startswith("observation.state"))
action_keys = sorted(k for k in FEATURES if k.startswith("action."))
video_keys = sorted(k for k in FEATURES if FEATURES[k]["dtype"] == "video")
print("\nstate :", [f"{k.split('.')[-1]}{tuple(FEATURES[k]['shape'])}" for k in state_keys])
print("action :", [f"{k.split('.')[-1]}{tuple(FEATURES[k]['shape'])}" for k in action_keys])
print("video :", video_keys)
tdf = pd.read_parquet(hf_hub_download(REPO_ID, f"{ROOT}/meta/tasks.parquet", repo_type="dataset"))
tdf = tdf.reset_index()
tcol = "task" if "task" in tdf.columns else tdf.columns[0]
TASKS = dict(zip(tdf["task_index"].astype(int), tdf[tcol].astype(str))) if "task_index" in tdf \
else {i: str(v) for i, v in enumerate(tdf[tcol])}
print(f"\n{len(TASKS):,} task strings. Random sample:")
for t in random.sample(list(TASKS.values()), min(8, len(TASKS))): print(" ·", t[:90])
ep_files = [f for f in meta_files if "/episodes/" in f and f.endswith(".parquet")]
eps = pd.concat([pd.read_parquet(hf_hub_download(REPO_ID, f, repo_type="dataset"))
for f in ep_files[:4]], ignore_index=True)
print(f"\nepisodes table: {len(eps):,} rows")
print("columns:", [c for c in eps.columns if not c.startswith("stats")][:14], "...")
print(eps[[c for c in ("episode_index", "length", "data/chunk_index", "data/file_index")
if c in eps.columns]].head())
Ми ініціалізуємо середовище Colab, встановлюємо необхідні бібліотеки та налаштовуємо параметри набору даних Cosmos3-DROID, епізодів, відео й навчання. Ми досліджуємо структуру репозиторію та визначаємо доступні фрагменти даних, відео й метаданих, не завантажуючи весь набір даних. Потім завантажуємо основні метадані й описи завдань, щоб зрозуміти схему набору даних, доступні ознаки стану/дії та організацію епізодів.
print("\n" + "="*78 + "\n3. BYTE-RANGE PARQUET READER\n" + "="*78)
def open_pf(rel_path):
return pq.ParquetFile(fs.open(HFS(rel_path), "rb"))
def rowgroup_span(pf):
md, starts, c = pf.metadata, [], 0
for i in range(md.num_row_groups):
starts.append(c); c += md.row_group(i).num_rows
return np.array(starts), c
def read_rows(pf, lo, hi, columns):
starts, total = rowgroup_span(pf)
ends = np.append(starts[1:], total)
rgs = [i for i in range(len(starts)) if starts[i] < hi and ends[i] > lo]
tbl = pf.read_row_groups(rgs, columns=columns)
return tbl.slice(lo - starts[rgs[0]], hi - lo)
def col2np(tbl, name):
ca = tbl.column(name).combine_chunks()
if pa.types.is_list(ca.type) or pa.types.is_large_list(ca.type) or pa.types.is_fixed_size_list(ca.type):
flat = np.asarray(ca.flatten().to_numpy(zero_copy_only=False))
return flat.reshape(len(ca), -1).astype(np.float32)
return np.asarray(ca.to_numpy(zero_copy_only=False)).reshape(-1, 1).astype(np.float32)
SHARD = data_shards[0]
pf = open_pf(SHARD)
md = pf.metadata
print(f"shard : {SHARD}")
print(f"rows : {md.num_rows:,} row_groups: {md.num_row_groups} "
f"compressed: {md.serialized_size/1e6:.1f} MB footer")
print(f"columns : {len(pf.schema_arrow.names)}")
t0 = time.time()
ep_idx_all = pf.read(columns=["episode_index"]).column("episode_index").to_numpy()
print(f"pulled episode_index column ({len(ep_idx_all):,} rows) in {time.time()-t0:.1f}s")
uniq, first_pos = np.unique(ep_idx_all, return_index=True)
order = np.argsort(first_pos)
uniq = uniq[order]; first_pos = first_pos[order]
last_pos = np.append(first_pos[1:], len(ep_idx_all))
EP_BOUNDS = {int(e): (int(a), int(b)) for e, a, b in zip(uniq, first_pos, last_pos)}
print(f"{len(EP_BOUNDS)} episodes live in this shard "
f"(ids {uniq.min()}..{uniq.max()}, mean len {np.mean(last_pos-first_pos):.0f} frames)")
STATE_USE = ["observation.state.joint_positions", "observation.state.gripper_position",
"observation.state.cartesian_position"]
ACTION_USE = ["action.joint_velocity", "action.gripper_position"]
READ_COLS = STATE_USE + ACTION_USE + ["timestamp", "frame_index", "task_index", "episode_index"]
def load_episode(ep):
lo, hi = EP_BOUNDS[ep]
tbl = read_rows(pf, lo, hi, READ_COLS)
out = {k: col2np(tbl, k) for k in STATE_USE + ACTION_USE}
out["timestamp"] = col2np(tbl, "timestamp").ravel()
out["task_index"] = int(col2np(tbl, "task_index").ravel()[0])
out["task"] = TASKS.get(out["task_index"], "<unknown>")
out["state"] = np.concatenate([out[k] for k in STATE_USE], axis=1)
out["action"] = np.concatenate([out[k] for k in ACTION_USE], axis=1)
return out
EP0 = int(uniq[0]); traj = load_episode(EP0)
print(f"\nepisode {EP0}: T={len(traj['state'])} state_dim={traj['state'].shape[1]} "
f"action_dim={traj['action'].shape[1]}")
print(f"task: {traj['task']!r}")
print("\n" + "="*78 + "\n5. TRAJECTORY ANALYTICS\n" + "="*78)
q = traj["observation.state.joint_positions"]
grip = traj["observation.state.gripper_position"].ravel()
cart = traj["observation.state.cartesian_position"]
dq = traj["action.joint_velocity"]
t = traj["timestamp"]
fig = plt.figure(figsize=(15, 9))
ax = fig.add_subplot(2, 3, 1)
for j in range(q.shape[1]): ax.plot(t, q[:, j], lw=1.1, label=f"j{j+1}")
ax.set_title("joint positions [rad]"); ax.set_xlabel("s"); ax.legend(fontsize=6, ncol=2)
ax = fig.add_subplot(2, 3, 2)
ax.plot(t, grip, color="crimson", lw=1.4)
opens = np.where(np.abs(np.diff(grip)) > 0.05)[0]
for k in opens[:40]: ax.axvline(t[k], color="k", alpha=.15, lw=.8)
ax.set_title(f"gripper (|Δ|>0.05 events: {len(opens)})"); ax.set_xlabel("s")
ax = fig.add_subplot(2, 3, 3, projection="3d")
ax.plot(cart[:, 0], cart[:, 1], cart[:, 2], lw=1.2)
ax.scatter(*cart[0, :3], c="g", s=45, label="start"); ax.scatter(*cart[-1, :3], c="r", s=45, label="end")
ax.set_title("EE cartesian path [m]"); ax.legend(fontsize=7)
ax = fig.add_subplot(2, 3, 4)
im = ax.imshow(dq.T, aspect="auto", cmap="RdBu_r", vmin=-np.abs(dq).max(), vmax=np.abs(dq).max())
ax.set_title("action.joint_velocity (7 x T)"); ax.set_ylabel("joint"); plt.colorbar(im, ax=ax)
ax = fig.add_subplot(2, 3, 5)
freqs = np.fft.rfftfreq(len(dq), d=1/FPS)
for j in range(dq.shape[1]):
ax.semilogy(freqs, np.abs(np.fft.rfft(dq[:, j] - dq[:, j].mean())) + 1e-9, lw=.9)
ax.set_title("action spectra (Nyquist=7.5 Hz)"); ax.set_xlabel("Hz")
ax = fig.add_subplot(2, 3, 6)
lens = [EP_BOUNDS[e][1] - EP_BOUNDS[e][0] for e in list(EP_BOUNDS)[:2000]]
ax.hist(np.array(lens)/FPS, bins=40, color="steelblue")
ax.set_title(f"episode duration [s] (n={len(lens)})"); ax.set_xlabel("s")
plt.suptitle(f"{REPO_ID} · {ROOT} · ep {EP0} · {traj['task'][:70]}", y=1.0)
plt.tight_layout(); plt.show()
Ми реалізуємо засіб читання Parquet на основі діапазонів байтів, який отримує лише потрібні групи рядків і стовпці безпосередньо через файлову систему Hugging Face. Ми визначаємо межі епізодів у фрагменті даних і перетворюємо вибрані поля стану та дій на траєкторії NumPy. Потім візуалізуємо положення суглобів, активність захвата, декартівський рух, розподіли дій, частотні спектри та статистику тривалості епізодів.
Ми створюємо відеоконвеєр із позиціонуванням, який отримує лише потрібне часове вікно епізоду замість завантаження всього відеофрагмента. Ми підтримуємо декодування через PyAV і FFmpeg для ефективної роботи з відео AV1 та змінюємо розмір вибраних кадрів для легкої обробки. Також ми завантажуємо статистику нормалізації на рівні набору даних із stats.json, передбачаючи емпіричний резервний варіант, якщо ця статистика недоступна.
Ми завантажуємо налаштовану кількість епізодів і за потреби кешуємо синхронізовані візуальні спостереження для невеликої підмножини, щоб зберегти прийнятні обчислювальні витрати під час навчання. Ми створюємо набір даних PyTorch у стилі ACT, який поєднує історію спостережень і опційні зображення з нормалізованими фрагментами майбутніх дій. Потім визначаємо архітектуру фрагментованої політики, яка поєднує MLP-кодер стану з опційним CNN-кодером зображень і прогнозує послідовність майбутніх дій.
Ми ініціалізуємо фрагментовану політику та оптимізуємо її за допомогою AdamW, планувальника швидкості навчання OneCycle, виконання зі змішаною точністю, масштабування градієнтів і обрізання градієнтів. Ми використовуємо функцію втрат Smooth L1, щоб зробити клонування поведінки стійкішим до шумних або змінних дій під час телеоперації. Ми навчаємо політику протягом заданої кількості епох, відстежуючи втрати на навчальній і валідаційній вибірках для контролю процесу навчання.
Ми оцінюємо навчену політику за допомогою розгортання у відкритому циклі та об’єднуємо перекривні передбачення дій, використовуючи експоненційно зважене часове ансамблювання. Ми обчислюємо MSE для кожного суглоба, помилку базової моделі та R2, а також будуємо графіки передбачених дій порівняно з еталонними траєкторіями разом із кривими втрат на навчанні та валідації. Нарешті, ми зберігаємо навчену модель зі статистикою нормалізації та метаданими конфігурації, щоб повторно використовувати політику в наступних експериментах.
Отже, ми показали, як перетворити масштабний набір даних реальних роботів на навчальний конвеєр, зберігаючи потреби у сховищі та передаванні даних на дуже низькому рівні. Ми використали пошук епізодів на основі метаданих, проєкцію стовпців і груп рядків Parquet та декодування відео з переходом до потрібної позиції, щоб отримувати лише інформацію, необхідну для аналізу й навчання, замість матеріалізації всього набору даних. Ми поєднали історію пропріоцептивних станів з опційними візуальними спостереженнями для навчання фрагментованої політики клонування поведінки та застосували часове ансамблювання, щоб отримати плавніші передбачення дій під час оцінювання у відкритому циклі. Отриманий робочий процес є компактною, але розширюваною основою, яку можна масштабувати на додаткові фрагменти, невдалі демонстрації, ракурси камер, мовні інструкції або альтернативні представлення дій для складніших експериментів у робототехніці та системах «бачення–мова–дія».
Перегляньте ПОВНИЙ КОД тут. Усі права належать досліднику цього проєкту. Також підписуйтеся на нас у Twitter і не забудьте приєднатися до нашого ML SubReddit із понад 150 тисячами учасників та підписатися на нашу розсилку. Стривайте! Ви є в Telegram? Тепер ви також можете приєднатися до нас у Telegram.
Потрібне партнерство з нами для просування вашого репозиторію GitHub, сторінки Hugging Face, випуску продукту, вебінару тощо? Зв’яжіться з нами
Перекладено автоматично з англійської. Оригінал статті — за посиланням нижче.