具身数据实战：从采集到训练的完整管线方案

最近这段时间，具身智能赛道的融资消息几乎没有断过，甚至有公司在 40 天内连续完成两轮融资，估值快速抬升。很多人把目光停在资本故事上，但真正被投资人反复调研的，其实是这类公司手里积累的“具身数据”到底能不能形成壁垒。

本文不谈商业模型，也不做行业预测，只从数据工程视角拆解一件事：具身数据实战派到底在做什么。我们会把具身数据从采集、清洗、标注、仿真合成到训练闭环的完整链路梳理一遍，并给出一条可落地的轻量级数据管线方案。无论你是算法工程师、后端开发者，还是正在调研机器人数据平台的技术负责人，这篇文章都能帮你建立一套清晰的工程框架。

1. 背景与核心概念：什么是具身数据，为什么它成了必争之地

1.1 具身智能与具身数据的定义

先解释两个基础概念。

具身智能（Embodied AI）指的是智能体不仅拥有“大脑”，还拥有“身体”，可以通过传感器感知周围环境，并利用电机、机械臂、移动底盘等执行器与环境真实交互。典型载体包括机械臂、人形机器人、四足机器人、无人车等。具身智能与纯语言模型最大的区别在于：它必须处理物理世界的连续性、不确定性和实时性。

具身数据就是智能体在与物理世界交互过程中产生的多模态数据。常见组成包括：

• 相机数据：RGB 图像、深度图、点云。
• 惯性数据：IMU 的加速度和角速度。
• 关节状态：关节角度、关节角速度、力矩反馈。
• 指令数据：自然语言指令、目标坐标、任务标签。
• 操作结果：是否成功、失败原因、任务完成度。

这些数据组合在一起，才能真正描述“机器人在什么状态下，做了什么动作，产生了什么结果”。

1.2 普通数据集与具身数据集的差异

很多团队刚开始接触具身智能时，会不自觉地把 NLP 或 CV 的数据工程思路搬过来，结果发现完全不是一回事。差别主要体现在四个维度：

这四点决定了具身数据工程不能照搬传统数据中台方案，必须自己搭建一套“感知、存储、清洗、标注、回放、仿真”闭环。

1.3 实战派的核心壁垒

所谓“具身数据实战派”，通常指的不是写论文的团队，而是真正在工厂、仓储、家庭场景里持续采集数据的团队。他们的壁垒有三层：

1. 数据规模：拥有足够多的真机采集场景，能覆盖长尾情况。
2. 数据质量：有成熟的清洗、筛选、标注流程，保证训练数据不是“脏数据”。
3. 数据闭环：能快速把模型部署到真机，又从真机运行中采集新数据反哺训练。

正因为如此，即使模型结构公开、论文开源，后发团队依然很难短时间补齐数据差距。这也是为什么资本愿意在短时间内连续下注。

2. 环境准备与工具选型

从这一节开始，我们进入工程实操。无论你是在调研技术方案，还是想亲手跑通一条数据管线，都需要先准备一套基础环境。

2.1 操作系统与中间件

机器人数据采集最常用的组合是：

如果你只在 Windows 或 macOS 上做算法验证，也可以借助 Docker 或云服务器跑通大部分代码，但真机采集阶段建议回到 Linux 环境，否则传感器驱动和 ROS 2 的部署会浪费很多时间。

2.2 Python 环境搭建

建议使用 conda 或 venv 创建独立环境，避免系统依赖互相污染。

python3 -m venv ~/venvs/embodied_data source ~/venvs/embodied_data/bin/activate pip install --upgrade pip

后续用到的核心依赖会在各个章节按需安装。这里先安装最基础的几个库：

pip install numpy zarr numcodecs h5py matplotlib

版本需要根据你的项目实际情况调整，本文示例以常见环境为例，重点演示配置思路。

2.3 示例项目结构

为了便于理解，我们后续案例统一使用下面的目录结构：

embodied_data_pipeline/ ├── config/ │ └── pipeline.yaml ├── data/ │ ├── raw/ # 原始 sensor bag / 原始文件 │ ├── processed/ # 清洗对齐后的数据 │ └── datasets/ # 可直接训练的数据集 ├── scripts/ │ ├── record_demo.py │ ├── align_sensor.py │ ├── filter_episodes.py │ ├── annotate_semantic.py │ └── build_dataset.py ├── src/ │ ├── data_loader.py │ └── metrics.py └── notebooks/ └── explore_data.ipynb

这个结构把原始数据、处理脚本、训练加载器分开，后续扩展仿真合成、模型训练模块时不需要大改现有代码。

3. 数据采集：把机器人的“体验”变成可回放的数据

3.1 传感器数据从哪里来

一次典型的具身数据采集，机器人会同时输出多路消息。以一台带机械臂的移动机器人为例：

• /camera/color/image_raw：RGB 彩色图。
• /camera/aligned_depth_to_color/image_raw：对齐到彩色相机的深度图。
• /imu/data：IMU 数据，包含线加速度和角速度。
• /joint_states：机械臂各关节的角度、速度、力矩。
• /odom：底盘里程计信息。

这些消息的频率不同，有的是 30Hz，有的是 100Hz，有的机械臂反馈能达到 500Hz。所以时间戳是数据采集阶段最重要的元数据，丢了时间戳，后续很难对齐。

3.2 使用 ROS 2 Bag 采集数据

ROS 2 提供了一个非常实用的工具：ros2 bag record。它可以把多个 Topic 的消息按时间顺序写入同一个 bag 文件，支持后续回放。

采集命令如下：

mkdir -p ~/robot_data/demo_001 ros2 bag record \ --output ~/robot_data/demo_001 \ /camera/color/image_raw \ /camera/aligned_depth_to_color/image_raw \ /imu/data \ /joint_states \ /odom

执行后，系统会持续录制，直到你按下Ctrl+C。录制结束后，~/robot_data/demo_001目录下会出现一个 bag 文件，里面包含了所有指定话题的消息。

更推荐的做法是为每次采集增加元信息，比如操作者、任务描述、环境地点、光照条件等。可以写一个简单的采集脚本：

# 文件路径：scripts/record_demo.py import subprocess import json import time from pathlib import Path def record_episode(output_dir: str, metadata: dict): output_dir = Path(output_dir) output_dir.mkdir(parents=True, exist_ok=True) # 保存元数据 metadata["record_time"] = time.strftime("%Y-%m-%d_%H-%M-%S") with open(output_dir / "metadata.json", "w", encoding="utf-8") as f: json.dump(metadata, f, ensure_ascii=False, indent=2) cmd = [ "ros2", "bag", "record", "--output", str(output_dir / "episode"), "/camera/color/image_raw", "/camera/aligned_depth_to_color/image_raw", "/imu/data", "/joint_states", "/odom", ] subprocess.run(cmd) if __name__ == "__main__": record_episode( output_dir="data/raw/episode_20250412", metadata={ "task": "pick_and_place", "operator": "alice", "scene": "desk_A", "lighting": "normal", }, )

3.3 真机遥操作与示教数据

具身数据中很大比例是“示范数据”，也就是由人通过遥操作设备控制机械臂，完成一次任务，同时记录传感器和关节数据。这类数据的质量直接决定模型模仿学习的效果。

实际采集时建议做到：

• 每次任务只定义一个目标，不要中途切换。
• 在一个 episode 中保持相机位姿相对稳定，避免抖动过大。
• 记录结果标签，例如success或fail，失败数据也不要直接删除，很多情况失败数据能帮助模型理解边界。
• 每个场景至少采集多个操作者、多组光照、多组物体摆放的数据，增强多样性。

4. 数据同步与坐标系统一

数据采集完成后，原始 bag 不能直接用来训练，因为各传感器消息频率不一致，且坐标系复杂。这一节是整个数据管线的核心工程难点。

4.1 时间戳同步

假设 RGB 图像的频率是 30Hz，关节状态是 100Hz。当我们想获取“某一时刻的关节角度”来匹配一帧图像时，往往发现关节状态消息的时间戳并不正好等于图像时间戳。这时候必须做时间对齐。

对齐算法通常采用“最近邻匹配”或“线性插值”。最近邻匹配实现简单，适合传感器频率相近的场景；线性插值更精确，适合高频关节数据。

下面是一个基于二分查找的最近邻对齐示例：

# 文件路径：scripts/align_sensor.py import bisect import numpy as np from typing import List, Any def nearest_neighbor_alignment( target_timestamps: np.ndarray, source_timestamps: np.ndarray, source_values: np.ndarray, ) -> np.ndarray: """将 source_values 按照最近邻原则对齐到 target_timestamps 时间轴。""" aligned = [] source_ts = source_timestamps.tolist() for ts in target_timestamps: idx = bisect.bisect_left(source_ts, ts) if idx == 0: aligned.append(source_values[0]) elif idx >= len(source_ts): aligned.append(source_values[-1]) else: left = source_values[idx - 1] right = source_values[idx] if ts - source_ts[idx - 1] <= source_ts[idx] - ts: aligned.append(left) else: aligned.append(right) return np.array(aligned) if __name__ == "__main__": # 模拟 30Hz 图像帧时间戳 image_ts = np.linspace(0, 10, 301) # 模拟 100Hz 关节时间戳 joint_ts = np.linspace(0, 10, 1001) joint_values = np.random.rand(1001, 7) aligned_joint = nearest_neighbor_alignment(image_ts, joint_ts, joint_values) print("对齐后关节数据形状:", aligned_joint.shape)

这段代码将 100Hz 的关节数据匹配到 30Hz 的图像帧时间轴上，输出每一帧图像对应的 7 维关节向量。

4.2 坐标系标定

机械臂抓取场景通常涉及多个坐标系：

• base_link：机械臂基座坐标系。
• camera_link：相机坐标系。
• tool_link：机械臂末端执行器坐标系。

不同坐标系之间需要外参矩阵进行转换。相机外参通常通过手眼标定获得。所谓外参，就是相机坐标系相对于机械臂基座坐标系的旋转和平移。

一个典型的坐标转换伪代码：

# 文件路径：scripts/align_sensor.py import numpy as np def transform_points(points_in_camera: np.ndarray, ext_matrix: np.ndarray) -> np.ndarray: """ 将相机坐标系下的点转换到机械臂基座坐标系。 points_in_camera: (N, 3) ext_matrix: (4, 4) 相机到基座的外参矩阵 """ ones = np.ones((points_in_camera.shape[0], 1)) points_homo = np.hstack([points_in_camera, ones]) # (N, 4) points_in_base = points_homo @ ext_matrix.T return points_in_base[:, :3]

在实际工程中，一般不推荐在采集阶段临时做坐标转换，而是把原始坐标和标定矩阵都保存下来，在制作训练集时再统一处理。这样可以避免标定矩阵更新后需要重新采集数据。

4.3 清洗后数据统一格式

多模态数据对齐后，建议封装成统一的中间格式。推荐使用 Zarr 或 HDF5，因为这两种格式都支持分块压缩和快速随机读取。

一个典型 episode 的存储结构：

episode_001.zarr/ ├── .zgroup ├── step/ # 全局步数 ├── timestamp_ns/ # 纳秒时间戳 ├── joint_positions/ # 关节位置 (T, J) ├── joint_velocities/ # 关节速度 (T, J) ├── action/ # 动作数据 (T, A) ├── rgb/ # RGB 图像 (T, H, W, 3) ├── depth/ # 深度图 (T, H, W) ├── success/ # 成功标签 (T,) └── metadata.json # 元数据

5. 数据清洗与筛选

5.1 常见数据质量问题

真机数据采集很难保证每一帧都干净。常见问题包括：

• 运动模糊：机械臂快速移动时，相机曝光时间内画面模糊。
• 遮挡：机械臂本体或操作员手臂遮挡目标物体。
• 传感器漂移：IMU 长期使用后存在零偏漂移。
• 无效片段：操作员中途调整姿势，或任务启动前机器人静止时间过长。
• 数据重复度过高：连续多帧画面几乎相同，导致训练数据冗余。

如果不处理这些问题，模型很容易学到噪声模式，训练结果看似损失很低，真机表现却很差。

5.2 规则化清洗策略

清洗策略应先简单后复杂。基础的规则化清洗包括：

# 文件路径：scripts/filter_episodes.py import numpy as np from pathlib import Path import json def compute_episode_stats(episode_dir: Path) -> dict: """统计一个 episode 的基础指标，用于后续筛选。""" metadata_path = episode_dir / "metadata.json" if not metadata_path.exists(): return {} rgb_path = episode_dir / "rgb.npy" joint_path = episode_dir / "joint_positions.npy" frames = np.load(rgb_path) joints = np.load(joint_path) # 帧间像素差：衡量画面变化程度 if len(frames) > 1: diff = np.mean(np.abs(frames[1:] - frames[:-1])) else: diff = 0.0 stats = { "num_frames": len(frames), "mean_frame_diff": float(diff), "mean_joint_angle": float(np.mean(joints)), "std_joint_angle": float(np.std(joints)), } return stats def filter_episode(episode_dir: Path, min_frames: int = 100, min_diff: float = 2.0) -> bool: stats = compute_episode_stats(episode_dir) if not stats: return False if stats["num_frames"] < min_frames: return False if stats["mean_frame_diff"] < min_diff: # 画面基本静止，可能是无效片段 return False return True

这里使用了两个简单指标：帧数和帧间差异。帧间差异过低通常表示机器人没有实际动作，这类数据对训练没有帮助。

更进一步的清洗可以引入“动作平滑度”计算，检测关节角度是否出现跳变异常：

def detect_joint_jump(joint_positions: np.ndarray, threshold: float = 0.5) -> bool: """检测关节角度是否在相邻帧之间出现异常跳变。""" diff = np.abs(np.diff(joint_positions, axis=0)) max_jump = np.max(diff) return max_jump > threshold

如果连续多帧出现超大跳变，往往是传感器断线或通信异常，建议人工复核。

5.3 人工抽检

自动清洗只能过滤规则明确的问题。对于语义类问题，比如“这次抓取是否真的成功”“操作员是否违反了流程”，建议保留人工抽检环节。抽检比例可以控制在 5% 到 10%。

抽检工具可以很简单，把每段 episode 的首帧、末帧和关节轨迹画成一张图，让人快速判断是否符合采集标准。目标是一天能审核数百段 episode。

6. 数据标注：从“记录”到“语义”

6.1 标注类型

清洗后的原始数据还不能直接用于模仿学习或强化学习，通常还需要语义标注。常见标注任务包括：

• 目标检测标注：用 2D 框或 3D 框标出抓取目标的位置。
• 语义分割标注：区分物体、桌面、机器人本体、背景。
• 关键点标注：标出目标物体的抓取点。
• 轨迹属性标注：标记某段轨迹是演示、修正还是失败。
• 指令标注：为每段数据补充自然语言指令，用于学习语言条件策略。

6.2 自动预标注 + 人工校验

现在的视觉基础模型已经可以完成一部分自动预标注，例如目标检测、分割、深度估计。常见做法是先跑一遍检测模型生成伪标签，再由标注员修正。

下面是一个使用检测模型生成预标注的简化流程：

# 文件路径：scripts/annotate_semantic.py from pathlib import Path import numpy as np import json def auto_annotate_objects(rgb_image: np.ndarray, detector): """ 调用目标检测模型，返回目标框列表。 这里 detector 是外部模型封装，实际项目中可替换为 YOLO、DETR 等。 """ detections = detector(rgb_image) boxes = [] for det in detections: # det: [x1, y1, x2, y2, score, class_id] if det[4] > 0.5: boxes.append({ "bbox": [float(v) for v in det[:4]], "score": float(det[4]), "class_id": int(det[5]), }) return boxes def generate_labels_for_episode(episode_dir: Path, detector): rgb_dir = episode_dir / "rgb" label_dir = episode_dir / "labels" label_dir.mkdir(exist_ok=True) rgb_files = sorted(rgb_dir.glob("*.npy")) for idx, rgb_file in enumerate(rgb_files): image = np.load(rgb_file) boxes = auto_annotate_objects(image, detector) label_path = label_dir / f"{rgb_file.stem}.json" with open(label_path, "w") as f: json.dump({"boxes": boxes}, f) print(f"已完成 {len(rgb_files)} 帧预标注")

6.3 标注数据格式

建议统一使用 JSON 或 JSONL 格式保存每帧的标注信息。无论后续用 PyTorch 还是 TensorFlow，都可以通过解析 JSON 快速转换为训练所需的张量。

{ "frame_id": "000123", "timestamp_ns": 1744459200123456789, "boxes": [ { "bbox": [120.5, 80.2, 210.3, 190.7], "score": 0.92, "class_id": 1, "class_name": "object" } ] }

标注文件命名与图像帧索引保持一致，避免后续匹配混乱。

7. 仿真合成数据与真机数据结合

7.1 为什么需要仿真数据

真机数据采集成本高、效率低，尤其是一些危险场景或长尾场景，比如机械臂试图抓取易碎品、机器人在狭窄空间内运动，真机采集既危险又难以重复。仿真引擎可以在虚拟环境中快速生成大量带有精确标注的数据，这是真机数据无法替代的。

常用的仿真引擎包括 MuJoCo、Isaac Sim 和 Gazebo。其中 MuJoCo 轻量且高效，适合快速验证物理交互；Isaac Sim 渲染效果好，适合生成视觉逼真数据。

7.2 合成数据生成要点

在仿真环境中生成数据时，最重要的技术是域随机化。也就是在每一轮生成数据时，随机改变环境参数，让模型不会依赖特定的纹理、光照或物理参数。

常见的随机化参数：

• 光照强度与方向。
• 物体纹理和颜色。
• 相机位置和视角噪声。
• 物体物理属性，如摩擦系数、质量。
• 背景场景。

MuJoCo 官方提供了 Python 接口，可以快速搭建一个随机采样场景。下面是一段生成随机化动作数据的思路示例：

# 文件路径：scripts/simulate_episode.py import mujoco import numpy as np def random_episode_from_model(model_xml: str, num_steps: int = 200): """ 使用 MuJoCo 加载模型并执行随机动作，返回关节轨迹。 这是一个最小示例，实际工程需根据机器人模型调整。 """ model = mujoco.MjModel.from_xml_string(model_xml) data = mujoco.MjData(model) mujoco.mj_resetData(model, data) # 随机化初始化位置 data.qpos[:] = np.random.uniform(model.jnt_range[:, 0], model.jnt_range[:, 1]) traj = [] for _ in range(num_steps): # 简单随机动作，实际仿真应接入策略或遥操作轨迹 data.ctrl[:] = np.random.uniform(model.actuator_ctrlrange[:, 0], model.actuator_ctrlrange[:, 1]) mujoco.mj_step(model, data) traj.append(data.qpos.copy()) return np.array(traj) if __name__ == "__main__": xml = """ <mujoco> <worldbody> <light pos="0 0 3"/> <body name="arm" pos="0 0 0"> <geom type="cylinder" size="0.05 0.2" mass="1"/> </body> </worldbody> </mujoco> """ traj = random_episode_from_model(xml, num_steps=50) print("仿真轨迹形状:", traj.shape)

这段代码只是验证流程，不等于生产可用的仿真脚本。实际项目中需要使用机器人的精确模型，并把仿真数据与真机数据统一存储格式。

7.3 真机与仿真数据的配比

在训练策略模型时，仿真数据和真机数据往往是混合使用。常见的经验比例是 7:3 或 8:2，仿真数据占多数，真机数据作为微调和验证。但具体比例需要根据 sim-to-real 的差距来调整。

一个重要原则是：固定一组真机数据作为测试集，永远不要参与训练。只有这样才能准确衡量仿真数据是否真正帮助了真机表现。

8. 数据存储、版本化与质量度量

8.1 存储格式选择

具身数据通常量级很大。一段 10 分钟的任务，30Hz 的 RGB 视频，256×256 分辨率，体积大约在 1.1GB 左右。如果每天采集 100 段，一天就是 100GB，一个月就是 3TB。

因此存储格式必须支持：

• 分块压缩。
• 快速随机读取，按帧读取而不是按文件读取。
• 元数据与数据分离。

Zarr 是目前比较合适的方案。它的优势在于按 chunk 存储，可以并行读写，而且天然支持云存储。转换示例：

# 文件路径：scripts/convert_bag_to_zarr.py import zarr import numpy as np from pathlib import Path def create_episode_zarr(output_path: str, num_frames: int, channels: int, height: int, width: int): root = zarr.open(output_path, mode="w") root.create_dataset( "rgb", shape=(num_frames, height, width, channels), chunks=(16, height, width, channels), dtype="uint8", compressor="default", ) root.create_dataset( "joint_positions", shape=(num_frames, 7), chunks=(256, 7), dtype="float32", compressor="default", ) root.attrs["task"] = "pick_and_place" root.attrs["created_at"] = "2025-04-12" return root if __name__ == "__main__": root = create_episode_zarr( "data/processed/episode_001.zarr", num_frames=1000, channels=3, height=256, width=256, ) print("Zarr 数据集已创建:", root.tree())

8.2 数据版本管理

数据会不断更新，模型训练时如果引用了错误版本的数据，很容易出现“昨天能跑通，今天不行了”的问题。建议引入数据版本管理工具。轻量做法是使用 DVC，它可以把数据文件路径、哈希和版本记录在 Git 中。

dvc init dvc add data/processed git add data/processed.dvc .gitignore git commit -m "add processed embodied data"

每次数据清洗或新增后，重新执行dvc add并提交，就能保留完整的数据版本历史。这样当模型表现异常时，可以快速定位到训练数据版本差异。

8.3 数据质量指标

数据质量需要量化，否则团队之间关于“数据是否达标”的讨论就会变成感觉之争。建议建立几个基础指标：

这些指标可以写入数据集的metadata.json，每次训练前打印出来，辅助判断数据集是否健康。

9. 实战案例：搭建一条轻量具身数据管线

前面几章分别讲了概念、采集、对齐、清洗、标注和仿真。这一章我们把这些知识串起来，搭建一条可以在本地跑通的轻量数据管线。因为多数读者手里没有机器人，示例会使用模拟数据代替真实传感器消息，但完整保留了具身数据管线的核心环节。

9.1 创建示例数据集

第一步，生成一段模拟的多模态数据。我们创建 200 帧数据，每帧包含一个随机关节向量和一张简单图像。

# 文件路径：scripts/generate_demo_data.py import numpy as np import zarr from pathlib import Path def generate_demo_episode(output_path: str, num_frames: int = 200): output_path = Path(output_path) output_path.parent.mkdir(parents=True, exist_ok=True) root = zarr.open(output_path, mode="w") # 模拟关节数据：7 个关节位置 t = np.linspace(0, 4 * np.pi, num_frames) joint_positions = np.stack([ 0.3 * np.sin(t + i * 0.5) for i in range(7) ], axis=-1).astype(np.float32) # 模拟 RGB 图像序列：30x30x3，带简单运动 rgb = np.zeros((num_frames, 30, 30, 3), dtype=np.uint8) for i in range(num_frames): rgb[i, :, :, 0] = (i % 10) * 10 rgb[i, :, :, 1] = 255 - (i % 10) * 10 rgb[i, :, :, 2] = (joint_positions[i, 0] * 255 + 128).clip(0, 255) root.create_dataset("joint_positions", data=joint_positions, chunks=(32, 7)) root.create_dataset("rgb", data=rgb, chunks=(16, 30, 30, 3)) root.attrs["task"] = "demo_pick_place" root.attrs["num_frames"] = num_frames root.attrs["success"] = True print(f"生成示例数据完成: {output_path}") if __name__ == "__main__": generate_demo_episode("data/processed/episode_demo.zarr")

9.2 编写数据质量检查模块

第二步，加载示例数据并计算质量指标。

# 文件路径：src/metrics.py import numpy as np import zarr def compute_dataset_metrics(zarr_path: str) -> dict: root = zarr.open(zarr_path, mode="r") joint_positions = root["joint_positions"][:] rgb = root["rgb"][:] metrics = {} metrics["num_frames"] = joint_positions.shape[0] # 动作变化率 joint_diff = np.abs(np.diff(joint_positions, axis=0)) metrics["mean_joint_change"] = float(np.mean(joint_diff)) # 图像帧间差异 if len(rgb) > 1: frame_diff = np.mean(np.abs(rgb[1:] - rgb[:-1])) else: frame_diff = 0.0 metrics["mean_frame_diff"] = float(frame_diff) # 关节范围覆盖率 joint_min = joint_positions.min(axis=0) joint_max = joint_positions.max(axis=0) metrics["joint_range_span"] = float(np.mean(joint_max - joint_min)) return metrics if __name__ == "__main__": metrics = compute_dataset_metrics("data/processed/episode_demo.zarr") for key, value in metrics.items(): print(f"{key}: {value:.4f}" if isinstance(value, float) else f"{key}: {value}")

预期输出类似：

num_frames: 200 mean_joint_change: 0.1331 mean_frame_diff: 9.4286 joint_range_span: 0.6000

如果mean_joint_change和mean_frame_diff都接近 0，说明数据中存在大量静止片段，需要回采或清洗。

9.3 构建 PyTorch 数据加载器

第三步，把 Zarr 数据封装成 PyTorch Dataset，方便训练脚本直接使用。

# 文件路径：src/data_loader.py import torch from torch.utils.data import Dataset import zarr import numpy as np class EmbodiedEpisodeDataset(Dataset): def __init__(self, zarr_path: str, max_len: int = 64): self.episode_path = zarr_path root = zarr.open(zarr_path, mode="r") self.joint_positions = root["joint_positions"][:] self.rgb = root["rgb"][:] self.max_len = max_len def __len__(self) -> int: num_windows = len(self.joint_positions) - self.max_len return max(0, num_windows) def __getitem__(self, idx: int): start = idx end = idx + self.max_len joint_seq = torch.tensor( self.joint_positions[start:end], dtype=torch.float32 ) image_seq = torch.tensor( self.rgb[start:end].transpose(0, 3, 1, 2), dtype=torch.float32 ) return { "joint_positions": joint_seq, # (T, 7) "rgb": image_seq / 255.0, # (T, 3, H, W) } if __name__ == "__main__": dataset = EmbodiedEpisodeDataset( "data/processed/episode_demo.zarr", max_len=16 ) sample = dataset[0] print("关节序列形状:", sample["joint_positions"].shape) print("图像序列形状:", sample["rgb"].shape)

这样一个数据加载器可以直接接到模仿学习或强化学习的训练循环中。

9.4 验证完整管线

最后，把生成数据、质量检查和数据加载串起来。

cd embodied_data_pipeline python scripts/generate_demo_data.py python src/metrics.py python src/data_loader.py

这三条命令分别完成了原始数据生成、数据质量评估和训练数据加载。在实际项目中，把generate_demo_data.py替换为ros2 bag转换脚本，数据加载端基本不用改动。

10. 常见问题与排查思路

在搭建具身数据管线的过程中，几个问题出现频率非常高。

10.1 多路相机时间不同步

多相机系统中，每台相机曝光时刻可能不同。如果直接使用系统接收时间，可能出现数百毫秒的偏差。解决思路是使用硬件触发同步或 PTP 网络时钟同步。软件层面，则可以在采集时记录每帧的硬件时间戳，并在对齐阶段使用该时间戳而不是接收时间。

10.2 数据清洗过度

很多团队容易犯的另一类错误是清洗标准过严，把带有噪声的数据全部删除，结果导致训练数据失去真实环境的分布。正确做法是保留一部分“困难样本”，比如轻微遮挡、光照变化大的数据，让模型学会在噪声条件下工作。

10.3 标注规则不一致

多人标注同一个数据集时，很容易出现标注尺度不统一的问题。建议在标注前制作一份标准标注手册，包含正例和反例，并在标注过程中定期抽取相同样本进行交叉审核，计算标注员之间的一致性指标。

11. 最佳实践与工程建议

11.1 数据优先，模型其次

具身智能项目最容易犯的错是：上来先调模型，数据随手录一段就开始训练。实战派的经验恰好相反。模型结构可以借鉴开源方案，但数据质量只能自己打磨。建议在一个新项目启动时，先花两周时间建立数据采集和评估标准，再开始训练。

11.2 自动化数据流水线

把数据采集、上传、质检、清洗、标注、版本化全部做成自动化流水线。每天采集的数据自动进入待处理队列，质量指标自动生成。发现异常数据自动告警。这样可以大幅减少人工沟通成本。

11.3 数据闭环设计

数据工程不是一次性交付。训练好的模型部署到真机后，运行数据需要回流到数据集，经过清洗和筛选后进入下一轮训练。这个闭环的稳定度，决定了机器人能力能否持续提升。

11.4 安全与合规

机器人数据采集涉及物理设备，务必注意安全边界：

• 检测到异常关节力矩时立即触发急停。
• 采集期间设置安全围栏或保持人员随时可介入。
• 涉及人员、人脸、隐私场景的数据，必须获得合法授权并做脱敏处理。
• 数据存储使用最小权限原则，数据仓库访问需要审计。

这些要求不仅是工程规范，也是项目能否长期稳定运行的基础。

11.5 保存原始数据，永远不要在原始数据上覆盖

清洗和处理过程应该生成新文件，而不是直接修改原始 bag。原始数据是唯一事实来源，如果清洗规则需要调整，可以重新从原始数据生成一份新版本，而不是试图恢复被覆盖的文件。因此，建议给每条原始数据设置只读权限，保护原始数据不被误改。

12. 结尾：从“数据”到“实战”的起点

具身数据实战派的核心竞争力，不在于某一天写出一套漂亮的模型，而在于能不能日复一日地把数据质量、数据规模和数据闭环稳定地运转起来。40 天融两轮只是外界看到的信号，真正藏在背后的，是数据团队对每一帧图像、每一段关节轨迹、每一次成功与失败记录的长期投入。

如果你也想进入这个方向，建议从今天开始做一件小事：找一台带关节反馈的机器人设备或直接用 MuJoCo 仿真环境，按本文的数据格式采集一小段数据，跑通质量评估和数据加载流程。这个最小闭环一旦建立，后续扩展真机采集、自动标注、模型训练和闭环部署都会容易很多。

希望这篇关于具身数据工程的文章能帮你少走一些弯路。如果有收获，可以收藏备用；如果在搭建过程中遇到问题，也欢迎在评论区把报错现象和你的环境贴出来，我们一起排查。