LeRobot：开源机器人学习框架
如何借助 Hugging Face，用统一的数据格式、时间窗口和端到端学习重新定义机器人技术
AI模型价格对比 | AI工具导航 | ONNX模型库 | Vibe Coding教程 | PLC在线仿真器 | Tripo 3D | Meshy AI | ElevenLabs | KlingAI | ArtSpace | Phot.AI | InVideo
机器人学习一直都很复杂。每个机器人实验室都被同样的瓶颈所困：
- 数据采集、训练、控制各自一套、碎片化的工具
- 每个机器人一套自定义数据集格式
- 软硬件栈各自为政、彼此隔离
- 模型还没开始训练，就要先做大量预处理
但 Hugging Face 新推出的库 LeRobot，改变了一切。
LeRobot 提供了一套统一的开源系统，用于：
- 真实机器人与仿真环境的控制
- 多模态数据集处理
- 最先进的机器人学习算法
- 面向实时控制的优化推理
它是第一个从第一天起就为 机器学习原生（ML-native）而设计的机器人框架——高度模块化，从研究者到爱好者，人人可用。
1、LeRobot 的革命性体现在哪里？
LeRobot 的独到之处在于，它把机器人技术栈的每一层都连接进了一个连贯的生态系统。
1.1 适用于所有机器人的统一数据格式
不再需要自定义加载器，不再有对不齐的时间戳。
LeRobot 引入了 LeRobotDataset——一种标准化的数据集格式，包含：
- 表格数据（关节状态、夹爪状态、本体感知）
- 视觉数据（以高效方式存储的相机流）
- 元数据（回合边界、标定、机器人配置）
每个数据集都遵循相同的结构。
1.2 把时间窗口变得异常简单
几乎所有现代机器人学习都依赖时间上下文：
- 你需要历史，才能理解世界正在如何变化
- 你需要未来动作，用于动作分块（action chunking）、扩散策略（diffusion policy）和规划
- 即便在回合边界处，你也需要固定长度的序列
LeRobot 让你轻松获得这一切：
delta_timestamps = {
"observation.state": [-0.4, -0.2, 0.0], #loads 3 state vectors: 0.4 seconds before, 0.2 second before and current frame
"action": [0.0, 0.2, 0.4], #loads 3 action vectors
}
自动补零（padding）+ 掩码（mask）都已内置。不再需要手动缓冲，不再有差一位（off-by-one）的 bug。
2、LeRobot 是如何工作的
用下面的命令安装：
pip install lerobot
导入必要的库：
from pprint import pprint
import torch
from huggingface_hub import HfApi
import lerobot
from lerobot.datasets.lerobot_dataset import LeRobotDataset, LeRobotDatasetMetadata
from lerobot.datasets.streaming_dataset import StreamingLeRobotDataset
列出所有可用的数据集：
print("List of available datasets:")
pprint(lerobot.available_datasets)
获取带有 LeRobot 标签的 Hugging Face 仓库列表：
hub_api = HfApi()
repo_ids = [info.id for info in hub_api.list_datasets(task_categories="robotics", tags=["LeRobot"])]
pprint(repo_ids)
我们来探索其中一个仓库：
repo_id = "lerobot/nyu_rot_dataset"
ds_meta = LeRobotDatasetMetadata(repo_id)
print(f"Total number of episodes: {ds_meta.total_episodes}")
print(f"Average number of frames per episode: {ds_meta.total_frames / ds_meta.total_episodes:.3f}")
print(f"Frames per second used during data collection: {ds_meta.fps}")
print(f"Robot type: {ds_meta.robot_type}")
print(f"keys to access images from cameras: {ds_meta.camera_keys=}\n")
print("Tasks:")
print(ds_meta.tasks)
print("Features:")
pprint(ds_meta.features)
输出如下：
Total number of episodes: 14
Average number of frames per episode: 31.429
Frames per second used during data collection: 5
Robot type: unknown
keys to access images from cameras: ds_meta.camera_keys=['observation.images.image']
Tasks:
task_index
erase the board 0
hang the hanger on the rod 1
reach the blue mark on the table 2
close the door 3
stack the cups 4
turn the knob 5
insert the peg in the cup 6
press the button 7
hang the bag on the hook 8
open the box 9
pour the almonds into the cup 10
hang the mug on the hook 11
Features:
{'action': {'dtype': 'float32',
'fps': 5.0,
'names': {'motors': ['motor_0',
'motor_1',
'motor_2',
'motor_3',
'motor_4',
'motor_5',
'motor_6']},
'shape': (7,)},
'episode_index': {'dtype': 'int64', 'fps': 5.0, 'names': None, 'shape': (1,)},
'frame_index': {'dtype': 'int64', 'fps': 5.0, 'names': None, 'shape': (1,)},
'index': {'dtype': 'int64', 'fps': 5.0, 'names': None, 'shape': (1,)},
'next.done': {'dtype': 'bool', 'fps': 5.0, 'names': None, 'shape': (1,)},
'next.reward': {'dtype': 'float32', 'fps': 5.0, 'names': None, 'shape': (1,)},
'observation.images.image': {'dtype': 'video',
'names': ['height', 'width', 'channel'],
'shape': (84, 84, 3),
'video_info': {'has_audio': False,
'video.codec': 'av1',
'video.fps': 5.0,
'video.is_depth_map': False,
'video.pix_fmt': 'yuv420p'}},
'observation.state': {'dtype': 'float32',
'fps': 5.0,
'names': {'motors': ['motor_0',
'motor_1',
'motor_2',
'motor_3',
'motor_4',
'motor_5',
'motor_6']},
'shape': (7,)},
'task_index': {'dtype': 'int64', 'fps': 5.0, 'names': None, 'shape': (1,)},
'timestamp': {'dtype': 'float32', 'fps': 5.0, 'names': None, 'shape': (1,)}}
这是一个低帧率（5 Hz）多模态机器人数据集，包含 14 个回合，平均每个约 31 帧，覆盖 12 项不同的桌面与家庭任务，例如擦黑板、转旋钮、开盒子、叠杯子。它提供来自单一 84×84 图像流的同步 RGB 相机帧，外加 7 自由度（7-DoF）机器人关节状态和 7 维连续动作（每个电机一维）。每一帧都带有回合索引、时间戳、奖励、完成标志等元数据。由于它只有一个相机、分辨率低、回合数少、序列短，这个数据集极其轻量，非常适合快速训练、调试模仿学习流水线、行为克隆以及小型 CNN 模型，且无需很高的算力。
我们也可以这样获取一段简短摘要：
print(ds_meta)
#Output
LeRobotDatasetMetadata({
Repository ID: 'lerobot/nyu_rot_dataset',
Total episodes: '14',
Total frames: '440',
Features: '['observation.images.image', 'observation.state', 'action', 'timestamp', 'episode_index', 'frame_index', 'next.reward', 'next.done', 'index', 'task_index']',
})',
2.1 LeRobotDataset 的工作原理
加载一个真实的 Hugging Face 多模态机器人数据集：
from lerobot.datasets.lerobot_dataset import LeRobotDataset
from lerobot.datasets.streaming_dataset import StreamingLeRobotDataset
dataset = LeRobotDataset(repo_id, delta_timestamps=delta_timestamps)
就这样。整个包含视频 + 状态 + 元数据的数据集，一行即可加载。
2.2 流式加载大数据集（训练方式的一次变革）
如果数据集大到放不进你的 SSD：
dataset = StreamingLeRobotDataset(repo_id,
delta_timestamps=delta_timestamps, streaming=True)
哪怕是通过互联网，也能以 80–100 样本/秒的速度流式加载数据集。非常适合云端 GPU 与各种实验。
3、用 LeRobot + PyTorch 进行训练
由于数据以结构化张量（tensor）的形式返回，训练循环变得轻而易举：
import argparse
import numpy as np
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import DataLoader
from lerobot.datasets.streaming_dataset import StreamingLeRobotDataset
torch.backends.cudnn.benchmark = False
torch.backends.cudnn.deterministic = True
# ============================================================
# HELPERS
# ============================================================
def detect_camera_keys(sample):
return [k for k in sample.keys() if k.startswith("observation.images.")]
def get_single_cam(batch, cam_keys, device, size=84):
"""
Use only the first camera key, downsample to size x size.
Works for BC (B,C,H,W) and sequence (B,T,C,H,W).
"""
key = sorted(cam_keys)[0] # stable choice
x = batch[key].to(device)
if x.dim() == 4:
# [B,C,H,W]
x = F.interpolate(x, size=(size, size), mode="bilinear")
return x # [B,3,size,size]
elif x.dim() == 5:
# [B,T,C,H,W]
B, T, C, H, W = x.shape
x = x.view(B * T, C, H, W)
x = F.interpolate(x, size=(size, size), mode="bilinear")
x = x.view(B, T, C, size, size)
return x
else:
raise ValueError(f"Unexpected image dims: {x.shape}")
# ============================================================
# MODELS
# ============================================================
class MiniCNN(nn.Module):
def __init__(self, in_channels, out_dim):
super().__init__()
self.net = nn.Sequential(
nn.Conv2d(in_channels, 32, 5, stride=2, padding=2), # 1/2
nn.ReLU(),
nn.Conv2d(32, 64, 5, stride=2, padding=2), # 1/4
nn.ReLU(),
nn.Conv2d(64, 128, 3, stride=2, padding=1), # 1/8
nn.ReLU(),
nn.AdaptiveAvgPool2d((1, 1)),
)
self.fc = nn.Linear(128, out_dim)
def forward(self, x):
x = self.net(x)
x = x.flatten(1)
return self.fc(x)
class LightBC(nn.Module):
"""Image + state -> single action"""
def __init__(self, state_dim, action_dim, in_channels):
super().__init__()
self.img_encoder = MiniCNN(in_channels, 128)
self.fc = nn.Sequential(
nn.Linear(128 + state_dim, 128),
nn.ReLU(),
nn.Linear(128, action_dim),
)
def forward(self, img, state):
z_img = self.img_encoder(img)
z = torch.cat([z_img, state], dim=-1)
return self.fc(z)
class StateChunkPolicy(nn.Module):
"""State-only sequence model: history of states -> future action chunk"""
def __init__(self, state_dim, action_dim, T_hist, K_future, hidden=128):
super().__init__()
self.T_hist = T_hist
self.K_future = K_future
self.state_dim = state_dim
self.action_dim = action_dim
self.gru = nn.GRU(
input_size=state_dim,
hidden_size=hidden,
num_layers=1,
batch_first=True,
)
self.head = nn.Sequential(
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, K_future * action_dim),
)
def forward(self, state_hist):
# state_hist: [B,T_hist,state_dim]
out, h_last = self.gru(state_hist) # h_last: [1,B,H]
h = h_last[-1] # [B,H]
chunk_flat = self.head(h) # [B,K*action_dim]
return chunk_flat.view(-1, self.K_future, self.action_dim)
# ============================================================
# TRAINING PIPELINES
# ============================================================
def train_manip_bc(dataset, device):
"""
Sample BC:
- Single camera
- 84x84
- MiniCNN
"""
dl = DataLoader(dataset, batch_size=1, shuffle=False, num_workers=0)
first_batch = next(iter(dl))
cam_keys = detect_camera_keys(first_batch)
if len(cam_keys) == 0:
raise RuntimeError("No camera keys found (need at least 1 camera).")
img = first_batch[sorted(cam_keys)[0]] # [B,C,H,W]
B, C, H, W = img.shape
in_channels = C # only one camera
state_dim = first_batch["observation.state"].shape[-1]
action_dim = first_batch["action"].shape[-1]
print(f"[manip_bc] in_channels={in_channels}, state_dim={state_dim}, action_dim={action_dim}")
print(f"Using camera: {sorted(cam_keys)[0]}")
model = LightBC(state_dim, action_dim, in_channels).to(device)
opt = torch.optim.Adam(model.parameters(), lr=1e-3)
loss_fn = nn.MSELoss()
# reuse dl
for epoch in range(3): # fewer epochs for speed
running, steps = 0.0, 0
for batch in dl:
imgs = get_single_cam(batch, cam_keys, device) # [B,3,84,84]
state = batch["observation.state"].to(device) # [B,state_dim]
target = batch["action"].to(device) # [B,action_dim]
opt.zero_grad()
pred = model(imgs, state)
loss = loss_fn(pred, target)
loss.backward()
opt.step()
running += loss.item()
steps += 1
print(f"[manip_bc] Epoch {epoch+1}/3 | Loss: {running/steps:.4f}")
return model, cam_keys
def train_manip_chunk(dataset, device):
"""
Manipulation chunking:
- STATE ONLY (no images) for speed
"""
dl = DataLoader(dataset, batch_size=1, shuffle=False, num_workers=0)
first_batch = next(iter(dl))
state_hist = first_batch["observation.state"] # [B,T,state_dim]
action_chunk = first_batch["action"] # [B,K,action_dim]
T_hist = state_hist.shape[1]
state_dim = state_hist.shape[2]
K_future = action_chunk.shape[1]
action_dim = action_chunk.shape[2]
print(f"[manip_chunk] T_hist={T_hist}, K_future={K_future}, "
f"state_dim={state_dim}, action_dim={action_dim}")
model = StateChunkPolicy(state_dim, action_dim, T_hist, K_future).to(device)
opt = torch.optim.Adam(model.parameters(), lr=1e-3)
loss_fn = nn.MSELoss()
dl = DataLoader(dataset, batch_size=1, shuffle=False, num_workers=0)
for epoch in range(3):
running, steps = 0.0, 0
for batch in dl:
states = batch["observation.state"].to(device) # [B,T,state_dim]
target = batch["action"].to(device) # [B,K,action_dim]
opt.zero_grad()
pred = model(states)
loss = loss_fn(pred, target)
loss.backward()
opt.step()
running += loss.item()
steps += 1
print(f"[manip_chunk] Epoch {epoch+1}/3 | Loss: {running/steps:.4f}")
return model, None
def train_locomotion_chunk(dataset, device):
"""
Locomotion chunking:
- STATE ONLY
- Same StateChunkPolicy as manip_chunk
"""
dl = DataLoader(dataset, batch_size=1, shuffle=False, num_workers=0)
first_batch = next(iter(dl))
state_hist = first_batch["observation.state"] # [B,T,state_dim]
action_chunk = first_batch["action"] # [B,K,action_dim]
T_hist = state_hist.shape[1]
state_dim = state_hist.shape[2]
K_future = action_chunk.shape[1]
action_dim = action_chunk.shape[2]
print(f"[locomotion_chunk] T_hist={T_hist}, K_future={K_future}, "
f"state_dim={state_dim}, action_dim={action_dim}")
model = StateChunkPolicy(state_dim, action_dim, T_hist, K_future).to(device)
opt = torch.optim.Adam(model.parameters(), lr=1e-3)
loss_fn = nn.MSELoss()
dl = DataLoader(dataset, batch_size=1, shuffle=False, num_workers=0)
for epoch in range(3):
running, steps = 0.0, 0
for batch in dl:
states = batch["observation.state"].to(device)
target = batch["action"].to(device)
opt.zero_grad()
pred = model(states)
loss = loss_fn(pred, target)
loss.backward()
opt.step()
running += loss.item()
steps += 1
print(f"[locomotion_chunk] Epoch {epoch+1}/3 | Loss: {running/steps:.4f}")
return model, None
# ============================================================
# MAIN
# ============================================================
def main():
parser = argparse.ArgumentParser()
parser.add_argument(
"--task",
choices=["manip_bc", "manip_chunk", "locomotion_chunk"],
required=True,
)
parser.add_argument(
"--repo_id",
required=False,
default="lerobot/nyu_rot_dataset",
help="LeRobot dataset repo ID (default: lerobot/nyu_rot_dataset)",
)
args = parser.parse_args()
device = "cuda" if torch.cuda.is_available() else "cpu"
print("Device:", device)
print("Repo:", args.repo_id)
# Choose dataset + delta_timestamps based on task
if args.task == "manip_bc":
dataset = StreamingLeRobotDataset(args.repo_id)
elif args.task == "manip_chunk":
delta_timestamps = {
"observation.state": [-0.4, -0.2, 0.0], # short history
"action": [0.0, 0.2, 0.4], # short future chunk
}
dataset = StreamingLeRobotDataset(args.repo_id, delta_timestamps=delta_timestamps)
else: # locomotion_chunk
delta_timestamps = {
"observation.state": [-0.4, -0.2, 0.0],
"action": [0.0, 0.2, 0.4],
}
dataset = StreamingLeRobotDataset(args.repo_id, delta_timestamps=delta_timestamps)
print("Dataset ready.")
if args.task == "manip_bc":
train_manip_bc(dataset, device)
elif args.task == "manip_chunk":
train_manip_chunk(dataset, device)
else:
train_locomotion_chunk(dataset, device)
print("Done!")
if __name__ == "__main__":
main()
这个脚本用 LeRobot 数据集训练了三种机器人学习策略：
- manip_bc → 基于图像 + 状态的行为克隆（Behavioral Cloning）
- manip_chunk → 从状态历史预测未来的动作序列
- locomotion_chunk → 与上述相同，但用于运动（locomotion）数据
它针对速度与低 GPU 显存占用做了优化。
4、架构概览：一个 LeRobot 训练技术栈
4.1 视觉编码器（MiniCNN）
在视觉任务中，我们使用一个轻量的三层卷积编码器：
- Conv → ReLU
- Conv → ReLU
- Conv → ReLU
- AdaptiveAvgPool
- Linear 投影
它输出一个 128 维的潜在嵌入（latent embedding）。即便在 CPU 上，这个编码器也能以约 200 FPS 处理帧。
相机被下采样到 84×84，类似于经典 Atari 深度强化学习系统中使用的高效表示。只使用第一个可用相机以最大化速度。
4.2 面向行为克隆的多模态融合
在 manip_bc（操作行为克隆）任务中，策略必须根据相机图像和机器人的本体感知状态来预测其动作。
融合策略刻意保持简单：
- 编码图像 → 128 维
- 与状态向量拼接（concatenate）
- 通过一个两层的 MLP
- 预测当前时刻机器人的动作
这与许多视觉运动（visuomotor）策略的设计如出一辙，但计算成本只有其零头。
Manipulation BC 流水线（图像 + 状态 → 动作）
┌──────────────────────────────┐
│ LeRobot Dataset │
│ (image, state, action) │
└───────────────┬──────────────┘
│
▼
┌────────────────────────────────────────────┐
│ get_single_cam() │
│ - pick first camera │
│ - resize to 84 × 84 │
└───────────────────┬────────────────────────┘
│
▼
┌────────────────────────────────┐
│ MiniCNN │
│ (light vision enc.) │
│ Conv→ReLU→Conv→ReLU │
│ →Conv→ReLU→AvgPool │
│ Output: 128-D │
└───────────────┬────────────────┘
│
├───→ State Vector (robot joints)
│
▼
┌──────────────────────────────────────┐
│ LightBC Fusion MLP │
│ concat(IMAGE, STATE) │
│ ↓ │
│ Linear → ReLU → Linear │
│ Output: Action Vector │
└──────────────────────────────────────┘
4.3 基于 GRU 的分块时序建模
时序分块（temporal chunking）是现代扩散策略与自回归机器人控制方法的核心思想：利用过去的观测来预测未来的动作序列。
- 仅使用状态作为输入（不含图像）
- GRU（隐藏维度 = 128）
- Linear 头产出 K 个未来动作
这让我们得到了时序条件化的好处，又免去了处理多相机图像历史的开销。
分块在两种模式下使用：
- 操作分块（Manipulation Chunking）： 从一个短期的过去状态窗口，预测接下来的 K 个动作。
- 运动分块（Locomotion Chunking）： 相同的结构，但用于连续的步态控制或富含接触的运动任务。
Manipulation / Locomotion Chunking 流水线（状态 → 未来动作）：
┌──────────────────────────────────────┐
│ LeRobot Dataset │
│ w/ delta_timestamps │
│ (state history + │
│ future actions) │
└───────────────────┬──────────────────┘
│
▼
┌────────────────────────────────────────────┐
│ State Sequence (T steps) │
│ e.g. [-0.4, -0.2, 0.0] seconds │
└───────────────────┬────────────────────────┘
│
▼
┌────────────────────────┐
│ GRU Encoder │
│ (tiny recurrent) │
│ Input: [B,T,D] │
│ Output: 128-D │
└─────────────┬──────────┘
│
▼
┌────────────────────────┐
│ MLP Head │
│ Linear → ReLU → Linear │
│ Output: K × action_dim │
└─────────────────────────┘
4.4 用于零等待训练的流式数据集
借助 StreamingLeRobotDataset，每个样本都直接从 Hugging Face Hub 加载。这消除了：
- 完整下载
- 庞大的磁盘占用
- 预处理时间
即便面对大型数据集，训练也能即刻开始。
5、机器人学习的未来
LeRobot 解决了数十年来机器人领域的痛点：
1. 多模态同步
图像 + 状态 + 动作 = 始终对齐
2. 高效的视频存储
回合以大型 MP4 分块存储 → 文件更少、索引更快、数据集更小。
3. 无处不在
本地机器、云端 GPU、边缘机器人，一套 API 通吃。
4. 为现代机器人学习而生
扩散策略、Transformer、视觉-语言-动作（Vision-Language-Action）模型、行为克隆、模仿学习、离线强化学习（Offline RL）——一切都能无缝运转。
6、结束语
LeRobot 不是"又一个机器人库"。它是一套统一的、ML 原生的机器人技术栈，终于解决了数据碎片化、同步问题和训练低效。
如果说深度学习改变了计算机视觉与 NLP，那么 LeRobot 也将为机器人领域做同样的事。而最棒的部分是？它是开源的。
机器人技术，从此变得更具可及性、更强大、更面向未来。
原文链接: LeRobot: The Future of Open-Source Robot Learning Has Arrived
汇智网翻译整理，转载请标明出处
