> ## Content Index
> Fetch the complete content index at: https://www.hubwiz.com/blog/llms.txt
> Use this file to discover other available public pages before exploring further.

# LeRobot：开源机器人学习框架
- URL: https://www.hubwiz.com/blog/lerobot-open-source-robot-learning-framework/
- Published: 2026-07-21T01:28:11.000Z
- Updated: 2026-07-21T01:28:11.000Z
- Description: 如何借助 Hugging Face，用统一的数据格式、时间窗口和端到端学习重新定义机器人技术
- Author: admin
- Tags: LIBRARY, LEROBOT, #Import 2026-10-14 14:41

机器人学习一直都很复杂。每个机器人实验室都被同样的瓶颈所困：

- 数据采集、训练、控制各自一套、**碎片化的工具**
- 每个机器人一套**自定义数据集格式**
- 软硬件栈**各自为政、彼此隔离**
- 模型还没开始训练，就要先做**大量预处理**

但 Hugging Face 新推出的库 **LeRobot**，改变了一切。

LeRobot 提供了一套**统一的开源系统**，用于：

- 真实机器人与仿真环境的控制
- 多模态数据集处理
- 最先进的机器人学习算法
- 面向实时控制的优化推理

它是第一个从第一天起就为 **机器学习原生（ML-native）**而设计的机器人框架——高度模块化，从研究者到爱好者，人人可用。

![None](https://www.hubwiz.com/blog/content/images/2026/07/img-41.jpg)

## 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 数据集**训练了**三种机器人学习策略**：

1. **manip\_bc** → 基于图像 + 状态的行为克隆（Behavioral Cloning）
2. **manip\_chunk** → 从状态历史预测未来的动作序列
3. **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（操作行为克隆）任务中，策略必须根据相机图像和机器人的本体感知状态来预测其动作。

融合策略刻意保持简单：

1. 编码图像 → 128 维
2. 与状态向量拼接（concatenate）
3. 通过一个两层的 MLP
4. 预测当前时刻机器人的动作

这与许多视觉运动（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）——一切都能无缝运转。

![None](https://www.hubwiz.com/blog/content/images/2026/07/img-42.jpg)

## 6、结束语

LeRobot 不是"又一个机器人库"。它是一套**统一的、ML 原生的机器人技术栈**，终于解决了数据碎片化、同步问题和训练低效。

如果说深度学习改变了计算机视觉与 NLP，那么 LeRobot 也将为机器人领域做同样的事。而最棒的部分是？它是开源的。

**机器人技术，从此变得更具可及性、更强大、更面向未来。**

---

原文链接: [LeRobot: The Future of Open-Source Robot Learning Has Arrived](https://medium.com/programmed-iq/lerobot-the-future-of-open-source-robot-learning-has-arrived-8b6906105116?ref=hubwiz.com)

汇智网翻译整理，转载请标明出处