最近这段时间具身智能赛道的融资消息几乎没有断过甚至有公司在 40 天内连续完成两轮融资估值快速抬升。很多人把目光停在资本故事上但真正被投资人反复调研的其实是这类公司手里积累的“具身数据”到底能不能形成壁垒。本文不谈商业模型也不做行业预测只从数据工程视角拆解一件事具身数据实战派到底在做什么。我们会把具身数据从采集、清洗、标注、仿真合成到训练闭环的完整链路梳理一遍并给出一条可落地的轻量级数据管线方案。无论你是算法工程师、后端开发者还是正在调研机器人数据平台的技术负责人这篇文章都能帮你建立一套清晰的工程框架。1. 背景与核心概念什么是具身数据为什么它成了必争之地1.1 具身智能与具身数据的定义先解释两个基础概念。具身智能Embodied AI指的是智能体不仅拥有“大脑”还拥有“身体”可以通过传感器感知周围环境并利用电机、机械臂、移动底盘等执行器与环境真实交互。典型载体包括机械臂、人形机器人、四足机器人、无人车等。具身智能与纯语言模型最大的区别在于它必须处理物理世界的连续性、不确定性和实时性。具身数据就是智能体在与物理世界交互过程中产生的多模态数据。常见组成包括相机数据RGB 图像、深度图、点云。惯性数据IMU 的加速度和角速度。关节状态关节角度、关节角速度、力矩反馈。指令数据自然语言指令、目标坐标、任务标签。操作结果是否成功、失败原因、任务完成度。这些数据组合在一起才能真正描述“机器人在什么状态下做了什么动作产生了什么结果”。1.2 普通数据集与具身数据集的差异很多团队刚开始接触具身智能时会不自觉地把 NLP 或 CV 的数据工程思路搬过来结果发现完全不是一回事。差别主要体现在四个维度维度普通数据集具身数据集数据形态文本、图片为主格式统一多模态、多传感器、格式异构时间属性弱时序样本之间独立强时序每个动作都依赖上下文空间属性一般不具备三维空间坐标必须包含坐标系、位姿、空间关系标注难度可以众包标注规则相对清晰需要动作轨迹、操作结果专业门槛高这四点决定了具身数据工程不能照搬传统数据中台方案必须自己搭建一套“感知、存储、清洗、标注、回放、仿真”闭环。1.3 实战派的核心壁垒所谓“具身数据实战派”通常指的不是写论文的团队而是真正在工厂、仓储、家庭场景里持续采集数据的团队。他们的壁垒有三层数据规模拥有足够多的真机采集场景能覆盖长尾情况。数据质量有成熟的清洗、筛选、标注流程保证训练数据不是“脏数据”。数据闭环能快速把模型部署到真机又从真机运行中采集新数据反哺训练。正因为如此即使模型结构公开、论文开源后发团队依然很难短时间补齐数据差距。这也是为什么资本愿意在短时间内连续下注。2. 环境准备与工具选型从这一节开始我们进入工程实操。无论你是在调研技术方案还是想亲手跑通一条数据管线都需要先准备一套基础环境。2.1 操作系统与中间件机器人数据采集最常用的组合是组件推荐选择说明操作系统Ubuntu 22.04 LTS对 ROS 2、相机驱动、GPU 驱动支持良好机器人中间件ROS 2 Humble当前机器人生态中最常用的版本之一编程语言Python 3.10数据处理便捷社区生态完整仿真引擎MuJoCo / Isaac Sim用于合成数据生成和仿真验证数据存储Zarr / HDF5 / MCAP支持大规模时序数据高效读写如果你只在 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_rawRGB 彩色图。/camera/aligned_depth_to_color/image_raw对齐到彩色相机的深度图。/imu/dataIMU 数据包含线加速度和角速度。/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执行后系统会持续录制直到你按下CtrlC。录制结束后~/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(parentsTrue, exist_okTrue) # 保存元数据 metadata[record_time] time.strftime(%Y-%m-%d_%H-%M-%S) with open(output_dir / metadata.json, w, encodingutf-8) as f: json.dump(metadata, f, ensure_asciiFalse, indent2) 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_dirdata/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, axis0)) 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_okTrue) 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 pos0 0 3/ body namearm pos0 0 0 geom typecylinder size0.05 0.2 mass1/ /body /worldbody /mujoco traj random_episode_from_model(xml, num_steps50) 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, modew) root.create_dataset( rgb, shape(num_frames, height, width, channels), chunks(16, height, width, channels), dtypeuint8, compressordefault, ) root.create_dataset( joint_positions, shape(num_frames, 7), chunks(256, 7), dtypefloat32, compressordefault, ) 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_frames1000, channels3, height256, width256, ) 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 数据质量指标数据质量需要量化否则团队之间关于“数据是否达标”的讨论就会变成感觉之争。建议建立几个基础指标指标含义目标值参考成功率数据中标注为成功的 episode 占比至少 60% 以上帧间差异均值衡量画面变化程度高于设定阈值时间戳丢失率未对齐的消息占比小于 1%标注一致率人工抽检时标注与预标注一致的比例大于 95%动作重复率相邻动作向量余弦相似度低于 0.9 为宜这些指标可以写入数据集的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(parentsTrue, exist_okTrue) root zarr.open(output_path, modew) # 模拟关节数据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), dtypenp.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, datajoint_positions, chunks(32, 7)) root.create_dataset(rgb, datargb, 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, moder) joint_positions root[joint_positions][:] rgb root[rgb][:] metrics {} metrics[num_frames] joint_positions.shape[0] # 动作变化率 joint_diff np.abs(np.diff(joint_positions, axis0)) 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(axis0) joint_max joint_positions.max(axis0) 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, moder) 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], dtypetorch.float32 ) image_seq torch.tensor( self.rgb[start:end].transpose(0, 3, 1, 2), dtypetorch.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_len16 ) 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. 常见问题与排查思路在搭建具身数据管线的过程中几个问题出现频率非常高。问题现象常见原因解决思路ros2 bag 录制时丢帧磁盘写入速度不足或多任务忙等使用 SSD 录制关闭无关进程降低图像分辨率RGB 图像和关节数据无法对齐不同 Topic 频率不同时间戳未处理按时间戳做最近邻或线性插值保留原始时间戳机械臂末端位置与图像显示不一致相机外参标定误差重新手眼标定检查坐标转换矩阵仿真数据训练效果好真机效果差sim-to-real 差距过大增加域随机化范围混入真机数据微调数据量很大但模型不收敛数据多样性不足或重复度过高统计帧间差异筛选冗余数据增加场景多样性Zarr 数据读取慢chunk 大小设置不合理或未压缩调整 chunk 大小开启压缩使用多线程读取10.1 多路相机时间不同步多相机系统中每台相机曝光时刻可能不同。如果直接使用系统接收时间可能出现数百毫秒的偏差。解决思路是使用硬件触发同步或 PTP 网络时钟同步。软件层面则可以在采集时记录每帧的硬件时间戳并在对齐阶段使用该时间戳而不是接收时间。10.2 数据清洗过度很多团队容易犯的另一类错误是清洗标准过严把带有噪声的数据全部删除结果导致训练数据失去真实环境的分布。正确做法是保留一部分“困难样本”比如轻微遮挡、光照变化大的数据让模型学会在噪声条件下工作。10.3 标注规则不一致多人标注同一个数据集时很容易出现标注尺度不统一的问题。建议在标注前制作一份标准标注手册包含正例和反例并在标注过程中定期抽取相同样本进行交叉审核计算标注员之间的一致性指标。11. 最佳实践与工程建议11.1 数据优先模型其次具身智能项目最容易犯的错是上来先调模型数据随手录一段就开始训练。实战派的经验恰好相反。模型结构可以借鉴开源方案但数据质量只能自己打磨。建议在一个新项目启动时先花两周时间建立数据采集和评估标准再开始训练。11.2 自动化数据流水线把数据采集、上传、质检、清洗、标注、版本化全部做成自动化流水线。每天采集的数据自动进入待处理队列质量指标自动生成。发现异常数据自动告警。这样可以大幅减少人工沟通成本。11.3 数据闭环设计数据工程不是一次性交付。训练好的模型部署到真机后运行数据需要回流到数据集经过清洗和筛选后进入下一轮训练。这个闭环的稳定度决定了机器人能力能否持续提升。11.4 安全与合规机器人数据采集涉及物理设备务必注意安全边界检测到异常关节力矩时立即触发急停。采集期间设置安全围栏或保持人员随时可介入。涉及人员、人脸、隐私场景的数据必须获得合法授权并做脱敏处理。数据存储使用最小权限原则数据仓库访问需要审计。这些要求不仅是工程规范也是项目能否长期稳定运行的基础。11.5 保存原始数据永远不要在原始数据上覆盖清洗和处理过程应该生成新文件而不是直接修改原始 bag。原始数据是唯一事实来源如果清洗规则需要调整可以重新从原始数据生成一份新版本而不是试图恢复被覆盖的文件。因此建议给每条原始数据设置只读权限保护原始数据不被误改。12. 结尾从“数据”到“实战”的起点具身数据实战派的核心竞争力不在于某一天写出一套漂亮的模型而在于能不能日复一日地把数据质量、数据规模和数据闭环稳定地运转起来。40 天融两轮只是外界看到的信号真正藏在背后的是数据团队对每一帧图像、每一段关节轨迹、每一次成功与失败记录的长期投入。如果你也想进入这个方向建议从今天开始做一件小事找一台带关节反馈的机器人设备或直接用 MuJoCo 仿真环境按本文的数据格式采集一小段数据跑通质量评估和数据加载流程。这个最小闭环一旦建立后续扩展真机采集、自动标注、模型训练和闭环部署都会容易很多。希望这篇关于具身数据工程的文章能帮你少走一些弯路。如果有收获可以收藏备用如果在搭建过程中遇到问题也欢迎在评论区把报错现象和你的环境贴出来我们一起排查。
最近这段时间具身智能赛道的融资消息几乎没有断过甚至有公司在 40 天内连续完成两轮融资估值快速抬升。很多人把目光停在资本故事上但真正被投资人反复调研的其实是这类公司手里积累的“具身数据”到底能不能形成壁垒。本文不谈商业模型也不做行业预测只从数据工程视角拆解一件事具身数据实战派到底在做什么。我们会把具身数据从采集、清洗、标注、仿真合成到训练闭环的完整链路梳理一遍并给出一条可落地的轻量级数据管线方案。无论你是算法工程师、后端开发者还是正在调研机器人数据平台的技术负责人这篇文章都能帮你建立一套清晰的工程框架。1. 背景与核心概念什么是具身数据为什么它成了必争之地1.1 具身智能与具身数据的定义先解释两个基础概念。具身智能Embodied AI指的是智能体不仅拥有“大脑”还拥有“身体”可以通过传感器感知周围环境并利用电机、机械臂、移动底盘等执行器与环境真实交互。典型载体包括机械臂、人形机器人、四足机器人、无人车等。具身智能与纯语言模型最大的区别在于它必须处理物理世界的连续性、不确定性和实时性。具身数据就是智能体在与物理世界交互过程中产生的多模态数据。常见组成包括相机数据RGB 图像、深度图、点云。惯性数据IMU 的加速度和角速度。关节状态关节角度、关节角速度、力矩反馈。指令数据自然语言指令、目标坐标、任务标签。操作结果是否成功、失败原因、任务完成度。这些数据组合在一起才能真正描述“机器人在什么状态下做了什么动作产生了什么结果”。1.2 普通数据集与具身数据集的差异很多团队刚开始接触具身智能时会不自觉地把 NLP 或 CV 的数据工程思路搬过来结果发现完全不是一回事。差别主要体现在四个维度维度普通数据集具身数据集数据形态文本、图片为主格式统一多模态、多传感器、格式异构时间属性弱时序样本之间独立强时序每个动作都依赖上下文空间属性一般不具备三维空间坐标必须包含坐标系、位姿、空间关系标注难度可以众包标注规则相对清晰需要动作轨迹、操作结果专业门槛高这四点决定了具身数据工程不能照搬传统数据中台方案必须自己搭建一套“感知、存储、清洗、标注、回放、仿真”闭环。1.3 实战派的核心壁垒所谓“具身数据实战派”通常指的不是写论文的团队而是真正在工厂、仓储、家庭场景里持续采集数据的团队。他们的壁垒有三层数据规模拥有足够多的真机采集场景能覆盖长尾情况。数据质量有成熟的清洗、筛选、标注流程保证训练数据不是“脏数据”。数据闭环能快速把模型部署到真机又从真机运行中采集新数据反哺训练。正因为如此即使模型结构公开、论文开源后发团队依然很难短时间补齐数据差距。这也是为什么资本愿意在短时间内连续下注。2. 环境准备与工具选型从这一节开始我们进入工程实操。无论你是在调研技术方案还是想亲手跑通一条数据管线都需要先准备一套基础环境。2.1 操作系统与中间件机器人数据采集最常用的组合是组件推荐选择说明操作系统Ubuntu 22.04 LTS对 ROS 2、相机驱动、GPU 驱动支持良好机器人中间件ROS 2 Humble当前机器人生态中最常用的版本之一编程语言Python 3.10数据处理便捷社区生态完整仿真引擎MuJoCo / Isaac Sim用于合成数据生成和仿真验证数据存储Zarr / HDF5 / MCAP支持大规模时序数据高效读写如果你只在 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_rawRGB 彩色图。/camera/aligned_depth_to_color/image_raw对齐到彩色相机的深度图。/imu/dataIMU 数据包含线加速度和角速度。/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执行后系统会持续录制直到你按下CtrlC。录制结束后~/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(parentsTrue, exist_okTrue) # 保存元数据 metadata[record_time] time.strftime(%Y-%m-%d_%H-%M-%S) with open(output_dir / metadata.json, w, encodingutf-8) as f: json.dump(metadata, f, ensure_asciiFalse, indent2) 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_dirdata/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, axis0)) 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_okTrue) 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 pos0 0 3/ body namearm pos0 0 0 geom typecylinder size0.05 0.2 mass1/ /body /worldbody /mujoco traj random_episode_from_model(xml, num_steps50) 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, modew) root.create_dataset( rgb, shape(num_frames, height, width, channels), chunks(16, height, width, channels), dtypeuint8, compressordefault, ) root.create_dataset( joint_positions, shape(num_frames, 7), chunks(256, 7), dtypefloat32, compressordefault, ) 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_frames1000, channels3, height256, width256, ) 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 数据质量指标数据质量需要量化否则团队之间关于“数据是否达标”的讨论就会变成感觉之争。建议建立几个基础指标指标含义目标值参考成功率数据中标注为成功的 episode 占比至少 60% 以上帧间差异均值衡量画面变化程度高于设定阈值时间戳丢失率未对齐的消息占比小于 1%标注一致率人工抽检时标注与预标注一致的比例大于 95%动作重复率相邻动作向量余弦相似度低于 0.9 为宜这些指标可以写入数据集的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(parentsTrue, exist_okTrue) root zarr.open(output_path, modew) # 模拟关节数据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), dtypenp.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, datajoint_positions, chunks(32, 7)) root.create_dataset(rgb, datargb, 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, moder) joint_positions root[joint_positions][:] rgb root[rgb][:] metrics {} metrics[num_frames] joint_positions.shape[0] # 动作变化率 joint_diff np.abs(np.diff(joint_positions, axis0)) 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(axis0) joint_max joint_positions.max(axis0) 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, moder) 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], dtypetorch.float32 ) image_seq torch.tensor( self.rgb[start:end].transpose(0, 3, 1, 2), dtypetorch.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_len16 ) 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. 常见问题与排查思路在搭建具身数据管线的过程中几个问题出现频率非常高。问题现象常见原因解决思路ros2 bag 录制时丢帧磁盘写入速度不足或多任务忙等使用 SSD 录制关闭无关进程降低图像分辨率RGB 图像和关节数据无法对齐不同 Topic 频率不同时间戳未处理按时间戳做最近邻或线性插值保留原始时间戳机械臂末端位置与图像显示不一致相机外参标定误差重新手眼标定检查坐标转换矩阵仿真数据训练效果好真机效果差sim-to-real 差距过大增加域随机化范围混入真机数据微调数据量很大但模型不收敛数据多样性不足或重复度过高统计帧间差异筛选冗余数据增加场景多样性Zarr 数据读取慢chunk 大小设置不合理或未压缩调整 chunk 大小开启压缩使用多线程读取10.1 多路相机时间不同步多相机系统中每台相机曝光时刻可能不同。如果直接使用系统接收时间可能出现数百毫秒的偏差。解决思路是使用硬件触发同步或 PTP 网络时钟同步。软件层面则可以在采集时记录每帧的硬件时间戳并在对齐阶段使用该时间戳而不是接收时间。10.2 数据清洗过度很多团队容易犯的另一类错误是清洗标准过严把带有噪声的数据全部删除结果导致训练数据失去真实环境的分布。正确做法是保留一部分“困难样本”比如轻微遮挡、光照变化大的数据让模型学会在噪声条件下工作。10.3 标注规则不一致多人标注同一个数据集时很容易出现标注尺度不统一的问题。建议在标注前制作一份标准标注手册包含正例和反例并在标注过程中定期抽取相同样本进行交叉审核计算标注员之间的一致性指标。11. 最佳实践与工程建议11.1 数据优先模型其次具身智能项目最容易犯的错是上来先调模型数据随手录一段就开始训练。实战派的经验恰好相反。模型结构可以借鉴开源方案但数据质量只能自己打磨。建议在一个新项目启动时先花两周时间建立数据采集和评估标准再开始训练。11.2 自动化数据流水线把数据采集、上传、质检、清洗、标注、版本化全部做成自动化流水线。每天采集的数据自动进入待处理队列质量指标自动生成。发现异常数据自动告警。这样可以大幅减少人工沟通成本。11.3 数据闭环设计数据工程不是一次性交付。训练好的模型部署到真机后运行数据需要回流到数据集经过清洗和筛选后进入下一轮训练。这个闭环的稳定度决定了机器人能力能否持续提升。11.4 安全与合规机器人数据采集涉及物理设备务必注意安全边界检测到异常关节力矩时立即触发急停。采集期间设置安全围栏或保持人员随时可介入。涉及人员、人脸、隐私场景的数据必须获得合法授权并做脱敏处理。数据存储使用最小权限原则数据仓库访问需要审计。这些要求不仅是工程规范也是项目能否长期稳定运行的基础。11.5 保存原始数据永远不要在原始数据上覆盖清洗和处理过程应该生成新文件而不是直接修改原始 bag。原始数据是唯一事实来源如果清洗规则需要调整可以重新从原始数据生成一份新版本而不是试图恢复被覆盖的文件。因此建议给每条原始数据设置只读权限保护原始数据不被误改。12. 结尾从“数据”到“实战”的起点具身数据实战派的核心竞争力不在于某一天写出一套漂亮的模型而在于能不能日复一日地把数据质量、数据规模和数据闭环稳定地运转起来。40 天融两轮只是外界看到的信号真正藏在背后的是数据团队对每一帧图像、每一段关节轨迹、每一次成功与失败记录的长期投入。如果你也想进入这个方向建议从今天开始做一件小事找一台带关节反馈的机器人设备或直接用 MuJoCo 仿真环境按本文的数据格式采集一小段数据跑通质量评估和数据加载流程。这个最小闭环一旦建立后续扩展真机采集、自动标注、模型训练和闭环部署都会容易很多。希望这篇关于具身数据工程的文章能帮你少走一些弯路。如果有收获可以收藏备用如果在搭建过程中遇到问题也欢迎在评论区把报错现象和你的环境贴出来我们一起排查。
