Hugging Face整合机器人开发工具链

你有一个已经能录制演示并推送到 Hugging Face Hub 的 AI 智能体。现在你想让这个循环持续运行：全天收集 episode，用不断增长的数据集训练策略，部署它，再拉回下一批数据来改进它。跑一次循环，每个环节都能正常工作。每天都跑，你就开始为同样的字节传输重复付费了。你上传的录制文件不断增长，每次训练开始前都要把整个数据集复制到 GPU，每次新的 checkpoint 发出时，下一批录制数据又回来了。

本系列的第一篇文章介绍了 Strands Robots，这是 AWS 开源的一个 SDK（Apache 2.0），它将机器人抽象、仿真和 LeRobot 技术栈暴露为可组合成单一 Strands AI 智能体的 AgentTools。文章涵盖了 Robot() 工厂、在仿真中录制演示、运行策略，以及将同一套 AI 智能体代码部署到物理 SO-101 上。该工厂根据注册表解析机器人名称，注册表中包含机械臂、人形机器人、移动基座和灵巧手，所以贯穿本文使用的 SO-100 只是众多支持构型中的一种。机器人目录列出了该工厂支持的所有机器人。LeRobot 的数据集格式已被 Hub 上超过 90,000 个数据集和模型采用，来自超过 8,000 个发布者（LeRobot Project Pulse）。Strands Robots 的录制只是其中之一，所以任何为读取 LeRobot 数据而构建的工具都可以无需转换直接读取。如果你对 Strands Robots 不熟悉，从那里开始阅读；本文假设你已经完成了那套设置。

那篇文章沿一个方向追踪了 AI 智能体循环，从 Hub 数据集到物理机器人。本文则沿另一个方向追踪数据，从第一帧录制画面回到已部署的策略，途经 Hugging Face Storage Buckets——这是 2026 年 3 月宣布的一种可变、非版本化、由 Xet 支持的对象存储仓库类型。bucket 与你的数据集仓库共存于同一个 hf:// 命名空间，使用你已有的 hf CLI，因此它成为数据从录制那天到训练那天的中间工作层。

总需要有人来决定保留哪些 episode、场景漂移多远需要重新录制今天的 batch 是否足够训练，以及用哪个 checkpoint 替换机械臂上的那个。每个决定在采集 campaign 中都会出现数十次，每一次都需要在下一个命令发出之前先看看返回了什么。这就是 AI 智能体该做的事。本文带你走过单个 AI 智能体内部的数据循环：录制演示到 Storage Bucket、存储它使每次同步只上传变化的字节、从 Hub 流式读取数据集直接训练而不是下载、部署 checkpoint 回硬件只需改一个关键字参数。本文配套的可运行notebook见 examples/notebooks/05_streaming_data_loop.ipynb。

第一篇文章是往 Hub 推送数据集，而本文构建的 AI 智能体则是从自然语言提示录制 LeRobotDataset、同步到 Storage Bucket、再将同一数据集流式传回逐帧读取、本地无副本。你在写入的同一进程中读取它：录制数据集的同一个 Strands Robots Robot() 再将它流式读回。训练好的 checkpoint 然后用改一个关键字参数部署到同一个 Robot()，它在硬件上录制的演示则回到同一个 bucket。

图 1. 四个阶段共享一个后端。Robot("so100") 通过共享 DatasetRecorder 录制 LeRobotDataset；sync_dataset_to_bucket(...) 将其同步到 Storage Bucket；stream_dataset(...) 通过 Hub 流式读取无需全量下载；训练好的 checkpoint 以 mode="real" 部署到同一个 Robot。磁盘格式完全保持 LeRobot 写入时的样子。

因为一个 Robot() 既录制数据集又读取它，采集数据和训练数据就成了同一个对象在同一个后端上的两个方法。AI 智能体决定运行一个 episode 并调用一个工具；然后 rollout 以机器人的控制频率继续运行直到 episode 结束，训练好的策略产生每个动作。整个循环，仅需几行代码：

from strands import Agent
from strands_robots import Robot

sim = Robot("so100")                 # mode="sim" (default - safe, no hardware)
agent = Agent(tools=[sim])

# Record a demonstration and sync it to a bucket.
agent("Record a pick-the-cube demo and sync it to my-org/robot-fave.")

# Stream it back from the bucket to train, without downloading it first.
for batch in sim.stream_dataset("my-org/robot-fave/cube_pick", repo_type="bucket").dataloader(batch_size=64):
    ...

接下来是这段循环内部实际发生的事情，一步一步来。

Python 3.12+，Linux 或 macOS（Apple Silicon 支持 MuJoCo 后端）。

一个 Strands 兼容的模型提供商用于 AI 智能体的推理。Amazon Bedrock（需 AWS 凭证）、Anthropic API、OpenAI 或本地运行的 Ollama。

Strands Robots 及其数据集扩展：uv pip install -U "strands-robots[sim-mujoco,lerobot]>=0.5.1"。lerobot 扩展会拉取 LeRobot (>=0.6.1)、datasets、av 和 torchcodec，因此录制和视频解码都无需进一步设置。参见安装指南。

就这么简单。本文每个阶段都可以在这些条件下用笔记本运行。运行的是循环，而不是一个可用的策略：默认路径使用 mock 策略，录制一个有效但不实用的数据集。

一个 Hugging Face 账号和具有写权限的 token，外加用于创建 bucket 和同步数据集的 hf CLI：pip install -U "huggingface-hub>=1.6.0,<2.0.0"，然后 hf auth login。

硬件路径：SO-101 follower 和 leader 对，或任何其他 LeRobot 支持的机器人，校准文件放在 ~/.cache/huggingface/lerobot/calibration/ 下。

本地视觉-语言-动作（VLA）推理：NVIDIA GPU。规模化训练：读取 Hub 数据的 GPU 集群。

运行训练步骤：uv pip install "lerobot[training]"。录制和流式传输不需要它。如果跳过这一步，trainer.train() 会返回一个错误结果而不是 checkpoint。故障排除指南会指出该错误和修复它的安装命令。

第一步——录制演示到 bucket

你全天录制新的 episode，每个 episode 是一段连续的摄像头画面和关节状态-动作遥测。LeRobot 将这些写入为一组小的大文件，录制时不断增长。将它们推入版本化数据集仓库，每次追加都成为一个 commit，每个修订版都被保留。采集工作需要反过来：需要一个地方来写入字节并就地覆盖它们。这就是 Storage Bucket，它存在于你的 Hugging Face 工作空间中，使用你已有的权限。不需要配置身份和访问管理（IAM）角色、不需要跨域资源共享（CORS）规则，也不需要维护上传服务。

你的 AI 智能体以 LeRobot 在硬件上写入的相同格式录制 LeRobotDataset。录制 episode，然后将完成的数据集同步到 bucket。提示词请求 mock 策略，这是一个无需训练模型就能产生关节动作的替代品，因此你可以在没有可运行 checkpoint 之前就跑通整个循环：

from strands import Agent
from strands_robots import Robot, sync_dataset_to_bucket

sim = Robot("so100")                 # mode="sim" by default
agent = Agent(tools=[sim])
# One prompt drives scene setup, cameras, policy, and recording.
agent(
    "Create a world with the so100 robot, add a red cube and a front camera, "
    "start recording (repo_id='local/cube_pick', root='/tmp/cube_pick', fps=30, "
    "overwrite=True, task='pick up the red cube'), run the mock policy for "
    "60 steps, then stop recording."
)
# Sync the finished on-disk dataset into the bucket (no live recording session needed).
sync_dataset_to_bucket("/tmp/cube_pick", "my-org/robot-fave")
# -> {"status": "success", "bucket_uri": "hf://buckets/my-org/robot-fave/cube_pick"}

同步写入 hf://buckets/{bucket}/{run_id}，其中 run_id 默认为数据集目录名。第三步的流式读取也会指定 run：id 的前两段是 bucket，之后的所有内容是 bucket 内部的路径。

sync_dataset_to_bucket(root, bucket, run_id=...) 验证数据集并通过 hf CLI 将其同步，与录制生命周期解耦。如果你直接驱动一个开放记录器，相同的能力在 DatasetRecorder.sync_to_bucket(bucket, run_id=...) 上，而 stop_recording(bucket=...) 则在你停止活动录制的时刻进行同步。bucket 是你一天中写入的工作层；至于版本化的、已发布的产物，你仍然调用 push_to_hub()。两者使用相同的格式。

Episode 在结构上是完整的，但动作是占位符，所以它不是你想用来训练的数据。用 create_policy("<hf_repo>") 替换真实的策略来做实际的抓取；prompt、格式和 bucket 同步保持不变。

要在物理 SO-101 上录制，LeRobot 的 record CLI 处理主从式启动：

lerobot-record \
  --robot.type=so101_follower --robot.id=my_follower \
  --teleop.type=so101_leader  --teleop.id=my_leader \
  --dataset.repo_id=my_user/cube_picking \
  --dataset.single_task='Pick up the red cube'

数据集以与模拟录制相同的格式落在磁盘上，因此相同的同步调用可以将其发送到 bucket：sync_dataset_to_bucket("./recordings", "my-org/robot-fave", run_id="run-021")（或其包装的 hf sync ./recordings hf://buckets/my-org/robot-fave/run-021 CLI）。收集运行追加到一个位置，而你发布的仓库只包含你选择发布的版本。

Step 2 - 使用字节级去重存储

现在数据集已经在 bucket 中，下一次同步的成本是多少。让两个固定摄像机对准一只手臂清理同一张桌子八个小时，你录制的大部分内容都是已有的像素：相同的照明、相同的底盘、相同的背景，跨越数千个 episode。在版本化仓库上情况更糟，因为更改多 GB 视频分片中的一个帧会重新上传整个文件。

Bucket 由 Xet 支持，它使用内容定义分块（content-defined chunking）在字节级对你的上传进行去重。分块边界跟随内容，因此在插入几个字节时只会更改该字节所在的分块，而不会改变其后的每个边界。在 Hugging Face 自己的测量中（HF Storage），内容定义分块在 Hub 上将每次上传传输的数据量减少约四倍，在 Enterprise 计划中按去重后的占用量计费。他们的 bucket 基准测试展示了单个文件上的效果。从 500 MB 上传开始，更改 1% 的字节并重新上传移动了 5.5 MB，更改 5% 移动了 27.5 MB，更改 10% 移动了 55 MB。如果没有分块级去重，覆盖一个对象意味着再次发送它的所有字节，无论它们是否更改。

节省多少取决于文件布局，而 Strands Robots 记录器使用 LeRobot 的格式。Episode 进入 Parquet 分片（data/chunk-000/file-000.parquet）和每个摄像头的 MP4 分片（videos/observation.images.front/chunk-000/file-000.mp4），仅在当前文件填满时才滚动到新文件，LeRobot 的默认值为数据 Parquet 100 MB、视频 MP4 200 MB。因此，一天的录制后进行同步会上传新的尾部分片加上增长的填充了一半的分片，而不是整个数据集。明天再次同步同一个 bucket 时，Xet 处理去重。

图 2. 同步只上传更改的内容。全新数据集的首次同步上传每个分块；在录制更多 episode 后，Xet 的内容定义分块意味着下一次同步只上传新的分块，跳过已存储的分块。

Step 3 - 从 Hub 流式传输训练

要训练，你将 GPU 指向你的数据集。先下载的话那些 GPU 会闲置，直到数百 GB 拷贝完成。由于 Step 2 的分片布局，直接从 Hub 流式传输在这里是可行的：一个 batch 成为对大分片的几次字节范围读取，而不是数千次小请求。LeRobot 的 StreamingLeRobotDataset 将其转换为可直接使用的 torch iterable，Strands Robots 通过 stream_dataset() 暴露它：

图 3. 流式传输，不要下载。下载路径先将整个数据集拷贝到本地磁盘，所以 GPU 等待；stream_dataset() 直接从 bucket 读取 batch，本地磁盘没有任何内容，所以 GPU 从第一个 batch 开始训练。

reader = sim.stream_dataset("my-org/robot-fave/cube_pick", repo_type="bucket",
    shuffle=False, max_num_shards=1, buffer_size=1,  # one episode, in capture order
)

print(reader.num_episodes, reader.num_frames, reader.fps)
for frame in reader:
    frame["observation.images.front"]   # (3, H, W) tensor, decoded on the fly from the MP4 shard
    frame["observation.state"]           # joint vector, from the Parquet shard
    frame["action"]
    break

除了包含 schema、统计信息和 episode 索引的小型 meta/ 文件夹外，没有任何内容落在本地磁盘上。迭代时摄像头帧从远程 MP4 分片解码；state 和 action 来自 Parquet 分片。该循环一次读取一帧，适合检查一个 episode。要训练，将 reader 传递给 DataLoader 并迭代 batch。流式数据集通过有限蓄水库缓冲区在内部 shuffle，因此视频解码可以跨 worker 进程并行化，训练步骤本身是普通的 PyTorch：

# policy here is a LeRobot policy you constructed, such as ACTPolicy.
for batch in reader.dataloader(batch_size=64, num_workers=4):
    loss, _ = policy(batch)   # lerobot ACTPolicy.forward returns (loss, loss_dict)
    loss.backward()

如果你根本不想写这个循环，LeRobot 自己的 trainer 通过相同的引擎读取，所以你智能体收集的数据集在不需要一行新代码的情况下进行训练。它通过与进程内 reader 相同的关键字参数接收 bucket：

lerobot-train --policy.type=act \
  --dataset.repo_id=my-org/robot-fave/cube_pick \
  --dataset.repo_type=bucket \
  --dataset.streaming=true \
  --num_workers=4

Bucket 仅支持流式传输，所以 --dataset.repo_type=bucket 需要 --dataset.streaming=true，否则配置会被拒绝。当你想在自有进程中拥有循环时使用 stream_dataset()：验证一个 episode、在模拟中回放它，或馈入自定义评估循环。对于仅需本体感知的流式传输，drop_videos=True 完全跳过视频解码，这就是使其在没有 torchcodec wheel 的边缘设备上工作的原因。录制和数据集指南记录了该参数以及它需要的 delta_timestamps 映射。

运行策略和训练策略时 Provider 名称是共享的。create_trainer("lerobot_local") 返回一个与 create_policy() 类似的 Trainer，而 TrainSpec 描述了运行；记录-训练-部署循环然后在几行代码中闭合：

import os
os.environ["STRANDS_TRUST_REMOTE_CODE"] = "1"   # create_policy loads with trust_remote_code=True

from strands_robots import create_policy
from strands_robots.training import TrainSpec, create_trainer

trainer = create_trainer("lerobot_local", device="cuda")
spec = TrainSpec(dataset_root="/tmp/cube_pick", output_dir="/tmp/cube_pick_ft",
                 base_model="", steps=500, extra={"policy_type": "act"})
result = trainer.train(spec)                    # train ACT on the streamed dataset
policy = create_policy(result.checkpoint_dir)   # load the checkpoint straight back

在单张 NVIDIA L4（g6.4xlarge）上，500 步 ACT 优化器（51.6M 参数，有效 batch size 8）跨越 120 帧 episode 在 133 秒内完成，并写出一个 create_policy() 可以通过与运行任何其他策略相同的入口点加载的 checkpoint。训练时间随数据集大小、batch size 和步数变化，因此将其视为一个测量配置而非基准。"groot" 和 "cosmos3" provider 面向相同的 TrainSpec 和 Trainer 生命周期，因此周围的循环不变；每个 provider 首先验证其自身必填字段，因此 GR00T 运行需要 base_model 和 embodiment tag，Cosmos 3 运行需要 base_model 和 SFT recipe。在 train() 前调用 trainer.validate(spec)，它会返回给定后端缺少的确切内容列表。

Hugging Face 的预热缓存会在云端和任务运行区域附近的边缘位置预先缓存桶数据，这样集群就能本地读取，数据加载器也能领先于 GPU。在 Hugging Face 自有的桶基准测试中，针对 10 GB 有效载荷，热内容分发网络（CDN）读取命中约为 1,086 MB/s，冷读取为 780 MB/s；在 100 GB 时，热读取约为 1,124 MB/s，测试环境为 us-east-1 的 m5dn.24xlarge。纯对象存储的完整对比（包括上传和下载）可在该仪表板上查看。选择数据的存储位置是 Team 和 Enterprise 计划中的存储区域设置，截至本文撰写时提供美国和欧盟，未来将推出亚太地区和海湾合作委员会（GCC）地区；其他计划的用户仓库数据存储在美国。

在 macOS 上，import strands_robots 会将 Homebrew 的 ffmpeg 添加到加载器路径中，因此 torchcodec 无需额外设置即可解码流视频。

第 4 步 - 部署策略并返回数据到循环

在这一步中，你拿上刚刚训练的检查点，在物理机器人上运行它，并用它在下一轮中记录演示。这与第一篇文章中的智能体代码相同，只有一个关键字参数改为 mode="real"：

robot = Robot("so100", mode="real", port="/dev/ttyACM0",
              cameras={"front": {"type": "opencv", "index_or_path": "/dev/video0", "fps": 30}})
agent = Agent(tools=[robot])
agent("Pick up the red cube.")

检查点在物理机械臂上运行，机械臂记录的演示以你最初使用的相同 LeRobot 格式保存到磁盘，准备同步回桶中以进行下一轮训练。

如果你的数据已经存放在 Amazon Simple Storage Service（Amazon S3）上，本文中涉及的格式工作不会有任何改变。LeRobotDataset 是一个由 Parquet 和 MP4 分片组成的目录，因此在 Amazon S3 上的存储方式与其他地方完全相同，记录、训练和部署步骤可以在任何位置读取该格式。桶所带来的额外价值在于 Hub 原生路径：sync_dataset_to_bucket 和 stream_dataset(repo_type="bucket") 直接指向 hf://，因此你可以在没有单独存储路径需要连接的情况下获得同步和流式读取。两条路径运行相同的循环：如果你的数据已经在 Amazon S3，就走 Amazon S3；如果需要同步和流式读取而不愿先配置存储，就走桶。

明天再次运行循环，你就在向那个桶记录数据，只同步发生变化的字节，并将这些字节流式传输到 GPU 而无需等待下载。数据永远不会离开 LeRobot 格式，也永远不会离开 Hub。

使用示例应用程序尝试一下

完整的 Strands Robots 示例在 GitHub 上，地址为 strands-labs/robots，路径为 examples/notebooks/05_streaming_data_loop.ipynb。它逐步引导你完成整个循环：记录、渲染、同步到桶、流式返回、训练和加载检查点。每个单元格都在模拟环境中运行，使用模拟策略，因此无需 GPU、Docker 或 Hugging Face 凭证。

git clone https://github.com/strands-labs/robots.git
cd robots
uv pip install -U "strands-robots[sim-mujoco,lerobot]>=0.5.1"
jupyter notebook examples/notebooks/05_streaming_data_loop.ipynb

按从上到下顺序运行单元格。记录的数据集将保存在 /tmp/nb5_dataset 下。要将其同步到桶，请在第一个单元格中设置 BUCKET = "my-org/robot-fave"（完成 hf 认证登录后）；相邻的 RUN_ID 用于命名桶内的文件夹，笔记本从 f"{BUCKET}/{RUN_ID}" 流式获取数据。要在 GPU 上训练，请将 steps 增加到 500 并设置 device="cuda"。相同循环的智能体驱动版本位于 examples/06_agent_collect_and_stream.py。

安全注意事项

这里的代码片段是 Strands Robots 数据循环的"hello world"。一旦用它处理真实数据，有五件事会发生变化。

提示词注入。向智能体提供不可信数据可能导致提示词注入，即不可信的上下文被当作 LLM 指令来对待。这些智能体可以驱动机器人，现在还可以向共享存储写入数据和从中读取数据，因此这是一个需要跟踪的重要风险。只从可信来源向智能体提供数据。如果无法信任所有输入，请限制智能体可用的工具，使其无法执行安全关键操作或覆盖桶内容。

训练数据是信任边界。一个可以写入采集桶的智能体也可以写入后续策略将要训练的 episode，而该策略驱动着物理机械臂。将写入采集数据的凭证与训练作业读取所用的凭证分开，每个 run_id 下的同步都独立进行，这样一个 episode 就可以追溯到产生它的那个 run 并可单独删除，并将版本化数据集仓库作为已审核的产物，因为桶不会保留可供审计的修订版本。

桶凭证和作用域。sync_dataset_to_bucket(...)、stop_recording(bucket=...) 和 sync_to_bucket 通过 hf CLI 上传，使用 hf auth login 中的令牌。使用作用域限定为你要写入的特定命名空间的令牌，采集数据优先使用 --private 桶，并保持桶与你 push_to_hub 和共享的版本化数据集仓库相互独立。

原地覆盖不保留修订版本。桶原地覆盖，不保留任何修订版本，这使其成为一个工作层，同时也意味着重复的 run_id 会替换已存储的运行。每次采集运行都传递一个明确的 run_id，如上文所示。