当前位置：首页 > news >正文

具身智能操作知识梳理与拓展

news 来源：原创 2025/4/24 8:28:12

0. 简介

对于具身智能来说，目前算是一个集大成的工作，虽然目前仍然处于持续发展的阶段，但是有一些比较基础的知识还是需要系统整理的。目前具身智能发展的非常快，几天就能看到一个让人眼前一亮的工作。

在这里插入图片描述

1. 视觉-语言-动作模型分类

1.1 非Transformer语言指令控制策略

在采用 Transformer 模型之前，早期的语言条件机器人任务控制策略在架构上差异显著。

CLIPort 集成了 CLIP 的视觉和语言编码器与 Transporter 网络，创建了一个双流架构。在一个流中，CLIP 的视觉编码器从 RGB 图像中提取“语义”信息，而在另一个流中，Transporter 网络从 RGB-D 图像中提取“空间”信息。CLIP 句子编码器对语言指令进行编码并指导输出动作，这是一对末端效应器的姿态：拾取和放置姿态。CLIPort 展示了根据语言指令拾取和放置物体的能力。
BC-Z [113] 处理两种类型的任务指令：语言指令或人类演示视频。环境以 RGB 图像的形式呈现给模型，然后通过 FiLM 层将指令嵌入和图像嵌入结合，最终生成动作。这种条件策略被认为在未见过的任务上展现了零样本任务泛化能力。
MCIL [95] 是一种开创性的机器人策略，它集成了自由形式自然语言条件，与早期通常依赖于任务 ID 或目标图像的条件方法形成对比。MCIL 引入了利用无标签和非结构化演示数据的能力。通过训练策略跟随图像或语言目标，其中小部分训练数据集由配对的图像和语言目标组成。
HULC [96] 引入了多种旨在增强机器人学习架构的技术，包括机器人的分层学习、一个多模态变换器和离散潜在计划。该变换器学习高层行为，将低层局部策略和全局计划进行分层划分。此外，HULC 还结合了一种基于对比学习的视听语义对齐损失，以对齐 VL 模态。HULC++ [97] 进一步整合了自监督可用性模型，该模型引导 HULC 到由语言指令指定的可操作区域，使其能够在此指定区域内完成任务。
UniPi [128] 将决策问题视为文本条件的视频生成问题。为了预测动作，UniPi 根据给定的文本指令生成视频，并通过逆动力学从视频帧中提取动作。这种创新的政策作为视频的表述提供了几个优势，包括跨不同机器人任务的增强泛化能力以及从互联网视频到真实机器人的知识转移潜力。

1.2 基于Transformer的控制策略

自从引入 Transformers 后，控制策略趋向于类似的基于 Transformer 的架构。

Interactive Language [100] 提出了一个机器人系统，其中低级控制策略可以实时受到通过语言传达的人类指令的指导，从而实现长期重排任务的完成。这种基于语言的指导的有效性主要归因于利用了一个精心收集的数据集，该数据集包含多样的语言指令，其规模超越了以前的数据集一个数量级。
Hiveformer [101] 强调利用多视角场景观察和保持完整观察历史来支持语言条件策略。这种方法代表了相较于仅使用当前观察的 CLIPort 和 BC-Z 等先前系统的进步。值得注意的是，Hiveformer 是最早采用 Transformer 架构作为其策略骨干之一。
Gato [36] 提出了一种模型，可以玩 Atari 游戏、描述图像和堆叠积木，所有这些都使用一组模型参数。这一成就得益于统一的标记方案，使输入和输出在不同任务和领域之间协调一致。因此，Gato 实现了不同任务的同时训练。作为一个重要的里程碑，Gato 体现了构建“多模态、多任务、多具身通用代理”的潜力。
RoboCat [109] 提出了一个自我改进过程，旨在使代理能够快速适应新任务，仅需 100 个演示。这个自我改进过程迭代地微调模型，并使用微调后的模型自生成新数据。建立在 Gato 模型之上，RoboCat 融入了 VQ-GAN 图像编码器 [110]。在训练过程中，RoboCat 不仅预测下一个动作，还预测未来的观察。自我改进过程的有效性通过在多任务、多具身设置下进行的全面实验得到了验证。
RT-1 [114] 由与 BC-Z 相同的团队开发，具有与 BC-Z 的相似性，但引入了一些关键区别。特别是，RT-1 使用基于更高效的 EfficientNet [115] 的视觉编码器，脱离了 BC-Z 使用 ResNet 的做法。然而，RT-1 并不使用视频作为任务指令。此外，RT-1 用 Transformer 解码器替换了 BC-Z 中的 MLP 动作解码器，产生离散化的动作。这一修改使 RT-1 能够关注过去的图像，提高了其性能。
Q-Transformer [117] 扩展了 RT-1，引入了自回归 Q 函数。与通过模仿学习学习专家轨迹的 RT-1 相比，Q-Transformer 采用 Q 学习方法。除了 Q 学习的 TD 误差目标外，还加入了保守正则化器，以确保最大值动作保持在分布内。这种方法允许 Q-Transformer 利用成功的演示和失败的轨迹进行学习。
RT-Trajectory [118] 采用轨迹草图作为策略条件，而不是依赖语言条件或目标条件。这些轨迹草图由曲线组成，描绘了机器人末端执行器要遵循的预期轨迹。它们可以通过图形用户界面手动指定、从人类演示视频中提取或由基础模型生成。RT-Trajectory 的策略建立在 RT-1 的基础上，经过训练以控制机器人臂准确跟随轨迹草图。这种方法促进了对新对象、任务和技能的泛化，因为来自各种任务的轨迹是可迁移的。
ACT [119] 构建了一个带有动作分块的条件 VAE 策略，要求策略预测一系列动作而不是单个动作。在推理期间，使用称为时间集成的方法对动作序列进行平均。RoboAgent [120] 通过其 MT-ACT 模型扩展了这一方法，证明动作分块改善了时间一致性。此外，RoboAgent 引入了一种语义增强方法，利用修复技术来增强现有演示。
RoboFlamingo [121] 通过将基于 LSTM 的策略头附加到 VLM，调整了现有的 VLM Flamingo [32], [149]，这表明预训练的 VLM 可以有效地转移到语言条件的机器人操控任务。

1.3 多模态指令的控制策略

多模态指令启用了新的任务指定方式，例如通过演示、命名新对象或用手指指向。

VIMA [150] 强调多模态提示和模型的泛化能力。通过结合多模态提示，可以制定比传统纯文本提示更具体和复杂的任务。VIMA 引入了四种主要类型的任务：物体操控、视觉目标达到、新概念定位、一键视频模仿、视觉约束满足、视觉推理。这些任务往往难以甚至无法仅用语言提示表达。VIMA-Bench 已经开发出来，用于评估四个泛化水平：放置、组合、新对象、新任务。
MOO [111] 扩展了 RT-1 以处理多模态提示。利用 RT-1 的骨干，MOO 结合 OWLViT 来编码提示中的图像。通过用新对象和额外提示图像扩展 RT-1 数据集，MOO 增强了 RT-1 的泛化能力。这一扩展也促进了指定目标对象的新方法，例如用手指指向或点击图形用户界面。

1.4 具有 3D 视觉的控制策略

我们生活在一个三维世界中，直观上使用 3D 表示作为视觉输入应该提供比 2D 图像更丰富的信息。点云由于其直接来源于 RGBD 输入而成为表示 3D 输入的热门选择，如 DP3 和 3D Diffuser Actor 所示。然而，体素也在各种工作中进行了探索。RoboUniView [124] 通过一种新颖的 UVFormer 模块将 3D 信息注入 RoboFlamingo，从而显示出改进的性能，该模块作为其视觉编码器，提供来自多视角图像的 3D 占用信息。VER [151] 还提出将多视角图像粗到细地体素化为 3D 单元格，从而提高视觉-语言导航任务的性能。

PerAct [103] 通过利用 3D 体素表示，在观察和动作空间方面取得了进展。这种方法为动作学习提供了稳健的结构先验，使多视角观察的自然处理成为可能，并促进了 6-DoF 中的数据增强。在该框架中，模型的输入包括从 RGBD 图像重建的体素地图，而输出对应于引导夹爪运动的最佳体素。通过采用这种表述，PerAct 即使在少量演示的情况下也能促进高效的任务学习。
Act3D [104] 引入了一种连续分辨率的 3D 特征场，根据当前任务的需要进行自适应分辨率，解决了体素化的计算成本。
RVT, RVT-2 [105], [106] 提出了从场景点云的虚拟视图重新渲染图像，并使用这些图像作为输入，而不是直接依赖 3D 输入。

1.5 基于扩散的控制策略

基于扩散的动作生成利用了扩散模型在计算机视觉领域的成功。

Diffusion Policy [130] 将机器人策略公式化为 DDPM [131]。该方法结合了多种技术，包括后退地平线控制、视觉条件和时间序列扩散变换器。这种基于扩散的视动政策的有效性突显了其在多模态动作分布、高维动作空间和训练稳定性方面的熟练程度。
SUDD [133] 提出了一个框架，其中 LLM 指导数据生成，随后将过滤后的数据集蒸馏为视听语言运动政策。该框架通过将 LLM 与一套原始机器人工具（如抓取采样器和运动规划器）组合，实现了基于语言的数据显示生成。然后，它通过结合语言条件用于多任务学习，扩展了 Diffusion Policy，并促进了过滤数据集的蒸馏。
Octo [134] 引入了一种基于变换器的扩散政策，其特征在于模块化开放框架设计，允许从不同任务定义编码器、观察编码器和动作解码器灵活连接到 Octo 变换器。作为首批利用 Open XEmbodiment 数据集 [143] 的模型之一，Octo 显示出积极的迁移和跨多种机器人及任务的泛化能力。
MDT [136] 将新近推出的 DiT 模型 [137] 从计算机视觉适应到动作预测头。DiT 最初被提出作为一种基于变换器的扩散模型，替代了经典的 U-Net 架构用于视频生成。结合两个辅助目标——掩蔽生成前瞻和对比潜在对齐——MDT 显示出优于基于 U-Net 的扩散模型 SUDD 的表现。
RDT-1B [139] 是一个基于扩散的双手操控基础模型，同样建立在 DiT 上。它通过在各种机器人间引入统一的动作格式来解决数据稀缺问题，使其能够在超过 6000 条轨迹的异质多机器人数据集上进行预训练。因此，RDT 的参数规模扩大至 12 亿，并展示出零-shot 泛化能力。

1.6 具有 3D 视觉的基于扩散的控制策略

一些工作提出将 3D 视觉与基于扩散的策略相结合。DP3 [132] 将 3D 输入引入扩散政策，从而提高了性能。同样，3D Diffuser Actor [135] 共享 DP3 的核心思想，但在模型架构上有所不同，将 Act3D 与 Diffusion Policy 结合起来。

1.7 运动规划的控制策略

运动规划涉及将移动任务分解为离散的路径点，同时满足障碍物避免和运动限制等约束。

Language costs [98] 提出了一种新颖的机器人纠正方法，使用自然语言用于人机协作的机器人控制系统。该方法利用人类指令生成的预测成本图，由运动规划器利用这些成本图计算最佳动作。该框架使用户能够通过直观的语言命令纠正目标、指定偏好或从错误中恢复。
VoxPoser [125] 利用 LLM 和 VLM 创建两个 3D 体素地图，分别表示可用性和约束。它利用 LLM 的编程能力和 VLM 的感知能力。LLM 将语言指令翻译为可执行代码，调用 VLM 获取对象坐标。基于组合的可用性和约束地图，VoxPoser 采用模型预测控制生成机器人臂末端执行器的可行轨迹。值得注意的是，VoxPoser 不需要任何训练，因为它直接连接 LLM 和 VLM 进行运动规划。
RoboTAP [152] 将演示分解为阶段，每个阶段由夹爪的打开和关闭标记。在每个阶段，RoboTAP 使用 TAPIR 算法检测活动点，跟踪相关对象从源位置到目标位置的路径。然后可以通过视觉伺服控制机器人。通过将这些阶段串联在一起，创建了一个运动计划，从而实现少量镜头的视觉模仿。

1.8 基于点的动作控制策略

最近的研究探讨了利用 VLM 的能力选择或预测基于点的动作，这是构建完整 VLAs 的一种经济有效的替代方案。

PIVOT [153] 将机器人任务视为视觉问答，利用 VLM 从一组视觉建议中选择最佳机器人动作。视觉建议以图像上的关键点形式进行注释。VLM 被反复提示以进行细化，直到识别出最佳选项。
RoboPoint [107] 使用空间可用性预测的任务微调 VLM，即指出在图像上采取行动的位置。这些 2D 图像上的可用性点随后通过深度图投影到 3D 空间中，形成预测的机器人动作。
ReKep [51] 是一个约束函数，将场景中的 3D 关键点映射到数值成本。机器人操控任务可以表示为 ReKep 约束的序列，这些约束由大型视觉模型和 VLM 生成。因此，可以通过求解约束优化问题获得机器人动作。

1.9 大规模 VLA

大规模 VLA 等同于 RT-2 [2] 提出的原始 VLA 定义，如图 2a 所示。这一术语类似于 LLM 和一般语言模型之间的区分，或者大规模 VLM 和一般 VLM 之间的区分。

RT-2 [2] 力求利用大型多模态模型在机器人任务中的能力，借鉴 PaLI-X 和 PaLM-E 等模型。该方法引入了共同微调，旨在使模型适应互联网规模的视觉问答 (VQA) 数据和机器人数据。这一训练方案增强了模型的泛化能力，并带来了涌现能力。
RT-H [142] 引入了一个动作层次结构，包括位于语言指令和低级动作（平移和旋转）之间的语言动作的中间预测层。这个额外的层次促进了不同任务之间的数据共享。例如，“捡起”和“倒入”等语言指令可能涉及“将手臂抬起”的语言动作。此外，这一动作层次结构使用户能够指定纠正措施以从故障中恢复，模型可以从中学习。
RT-X [143] 建立在之前的 RT-1 和 RT-2 模型之上。这些模型使用新推出的开源大型数据集 Open X-Embodiment (OXE) 进行再训练，该数据集的规模比以前的数据集大几个数量级。结果模型 RT-1-X 和 RT-2-X 都超越了其原始版本。
OpenVLA [50] 随后被开发为 RT-2-X 的开源对应物。他们还探索了高效的微调方法，包括 LoRA 和模型量化。
π0 [147] 提出了一个流匹配架构，用于将 VLM 转换为 VLA。通过基于混合专家框架引入额外的动作专家，有效继承了 VLM 中的互联网规模知识，同时扩展其能力以应对机器人任务。

在这里插入图片描述

2. 数据集

近年来，随着机器人技术和人工智能的发展，大量机器人操作数据集被构建用于训练智能操作模型。本章将系统介绍当前主流的机器人操作数据集及其使用的机械臂平台。
在这里插入图片描述

2.1 MIME 数据集

MIME数据集[181]通过人类远程操作Baxter机械臂采集而成。该数据集包含RGB和深度数据，通过演示(Demo)方式提供指令。数据集涵盖12种任务类型、20种操作对象，在1个环境中采集，总计8.3K个样本。Baxter是一款双臂协作机器人，具有7个自由度，特别适合模仿学习任务。

2.2 RoboTurk 数据集

RoboTurk[182]是一个通过人类远程操作Sawyer机械臂采集的RGB数据集，包含2种任务类型，在单一环境中采集，总计2.1K个样本。Sawyer是一款单臂协作机器人，具有7个自由度，广泛应用于研究和工业环境。

2.3 RoboNet 数据集

RoboNet[183]是一个通过脚本生成的大规模数据集，包含162K个RGB样本，使用目标图像作为指令形式。该数据集独特之处在于使用了7种不同的机器人平台，覆盖10个不同环境，为跨平台机器人学习提供了宝贵资源。

2.4 MT-Opt 数据集

MT-Opt[184]是一个包含800K个RGB样本的大规模数据集，通过脚本生成，使用语言指令形式。该数据集采用7种不同机器人，涵盖2种任务类型，12种对象，在单一环境中采集，是机器人多任务学习的重要资源。

2.5 BC-Z 数据集

BC-Z[113]通过人类远程操作Everyday机械臂平台采集，包含25.9K个RGB样本，同时使用语言指令和演示作为指令形式。该数据集涵盖3种任务和100种不同对象，在单一环境中采集。

2.6 RT-1-Kitchen 数据集

RT-1-Kitchen[114]是一个由人类远程操作Everyday机械臂采集的RGB数据集，使用语言指令形式。该数据集涵盖12种任务、700多种对象，在2个不同环境中采集，包含16种语言指令，总计130K个样本，主要针对厨房场景的机器人操作。

2.7 MOO 数据集

MOO[111]是通过人类远程操作Everyday机械臂采集的RGB数据集，使用多模态指令形式。数据集包含5种任务类型和106种语言指令，总计59.1K个样本，适用于多模态机器人学习研究。

2.8 VIMA 数据集

VIMA[150]是通过脚本生成的大规模数据集，包含650K个RGB样本，使用多模态指令。该数据集使用UR5机械臂，涵盖17种任务类型，29种语言指令，在单一环境中采集。UR5是一款通用机械臂，具有6个自由度，在研究和工业应用中广泛使用。

2.9 RoboSet 数据集

RoboSet[185]结合了人类操作和脚本生成的方法，使用Franka Panda机械臂采集RGB和深度数据，通过语言指令形式。数据集涵盖12种任务、38种对象，在11个不同环境中采集，总计98.5K个样本。Franka Panda是一款7自由度协作机器人，以其精确性和灵活性而著称。

2.10 BridgeData V2 数据集

BridgeData V2[186]采用人类操作和脚本生成的混合方式，使用WidowX 250机械臂采集RGB和深度数据。该数据集包含13种任务类型，在24个不同环境中采集，包含100多种语言指令，总计60.1K个样本。WidowX 250是一款小型灵活的机械臂，适合桌面操作任务。
在这里插入图片描述

2.11 RH20T 数据集

RH20T[180]是通过人类远程操作4种不同机器人平台采集的RGB和深度数据集，使用语言指令形式。该数据集包含42种任务、147种对象，在7个不同环境中采集，总计超过110K个样本，为多平台机器人学习提供丰富资源。
在这里插入图片描述

2.12 DROID 数据集

DROID[146]通过人类远程操作Franka Panda机械臂采集RGB和深度数据，使用语言指令形式。该数据集涵盖86种任务类型，在564个不同环境中采集，总计76K个样本，环境多样性是其显著特点。
在这里插入图片描述

2.13 OXE 数据集

OXE[143]是目前规模最大的机器人操作数据集之一，通过聚合多个数据集形成，包含RGB和深度数据，使用语言指令形式。该数据集涵盖22种不同机器人平台，包含527种任务类型、超过16万种对象，在311个不同环境中采集，总计超过100万个样本，为大规模机器人学习研究提供了重要基础。
在这里插入图片描述

3. 具身智能研究中的常用机器人平台及其优缺点

3.1 Franka Emika Panda

优点：

7自由度协作机器人，运动灵活性高
内置力矩传感器，使精确的力控制成为可能
安全性好，可与人类安全共处
精度高，重复定位精度达0.1mm
ROS集成良好，API文档完善
相比工业机械臂价格较为合理

缺点：

载荷能力有限(仅3kg)
对计算资源要求较高
初始设置和维护需要专业知识
工作空间相对有限

3.2 Aloha (Berkeley Aloha)

优点：

双臂设计，适合人类日常操作场景
专为模仿学习设计，操作直观
视觉系统集成良好
支持远程遥操作数据收集
适合家庭环境的操作任务

缺点：

商业可得性有限
载荷能力较小
精度不及工业级机器人
开发生态系统不够成熟

3.3 Stretch (Hello Robot)

优点：

移动操作一体化平台
紧凑设计，适合家庭和办公环境
价格相对亲民(约2万美元)
简单易用，入门门槛低
伸缩臂设计，工作范围广

缺点：

单臂设计，操作复杂性受限
精度不及固定基座机械臂
抓取能力和多样性有限
自由度较少(4-5自由度)

3.4 xArm (UFACTORY)

优点：

价格较为亲民(6-7自由度版本)
多种型号可选(5/6/7自由度)
开发接口友好，Python/C++/ROS支持
精度较高(重复精度0.1mm)
集成视觉选项

缺点：

力控性能不如Franka
稳定性和耐用性略逊高端产品
技术支持和文档可能不够完善
高级功能需要额外付费

3.5 UR5/UR10 (Universal Robots)

优点：

工业级可靠性和耐用性
编程简单，可通过示教方式快速部署
广泛的行业应用和支持生态
UR10具有较大载荷(10kg)
ROS集成良好

缺点：

价格较高(3-5万美元)
没有内置力矩传感(需外部传感器)
设计较为传统，专为工业应用
安全功能相对基础

3.6 Kinova Gen3/Jaco

优点：

轻量化设计，适合移动平台集成
7自由度版本提供冗余自由度
内置视觉处理能力
适合辅助和服务应用
较好的开发者支持

缺点：

价格高于入门级平台
精度略低于同价位竞争产品
API和软件框架学习曲线较陡
力控制能力有限(早期版本)

3.7 WidowX 250/200 (Trossen Robotics)

优点：

非常经济实惠(约3000-4000美元)
开源硬件和软件
非常适合教育和研究入门
易于修改和定制
体积小，适合桌面应用

缺点：

载荷能力极为有限(~1kg)
精度和重复性不高
稳定性和耐用性不足
不适合精细或高精度任务

3.8 Digit (Agility Robotics)

优点：

全身人形设计，双足行走能力
双臂操作能力
适合研究真实环境导航与操作集成
传感器系统丰富

缺点：

极为昂贵(约20-30万美元)
操作编程复杂
维护成本高
需要专业团队支持

3.9 KUKA LBR iiwa

优点：

卓越的精度和重复性
强大的力矩传感能力
载荷能力较强(7-14kg)
工业级稳定性和可靠性

缺点：

价格极高(8-10万美元以上)
操作系统封闭，定制难度大
体积较大，不适合空间有限场景
编程相对复杂

3.10 TIAGo (PAL Robotics)

优点：

移动底盘和机械臂一体化设计
高度可配置(单/双臂版本)
适合服务机器人研究
丰富的传感器套件
ROS集成完善

缺点：

价格高(7-10万美元)
系统复杂度高，调试困难
机械臂精度低于专业机械臂
维护和支持依赖制造商

4. 快速高效的Python包管理器uv

uv是由Astral公司开发的一款极其快速的Python包管理器，完全用Rust编写。自2023年2月首次发布以来，uv作为pip工作流的替代品，迅速发展成为一个端到端的解决方案，能够管理Python项目、命令行工具、单文件脚本，甚至Python本身。uv可被视为Python界的Cargo，提供一个快速、可靠且易用的统一接口。

4.1 主要特性

uv具备以下主要功能：

端到端项目管理：通过uv run、uv lock和uv sync命令，uv能够基于标准元数据生成跨平台的锁文件，并从中安装依赖，性能优于Poetry、PDM和Rye等工具。
工具管理：使用uv tool install和uv tool run（别名uvx），uv可以在隔离的虚拟环境中安装命令行工具，并无需显式安装即可执行一次性命令，速度更快。
Python安装：通过uv python install，uv能够自动下载安装Python，效率高于pyenv。
脚本执行：uv支持基于PEP 723的内联元数据的单文件Python脚本，只需uv run即可执行独立的Python脚本。

这些功能都得益于uv极快的跨平台依赖解析器。

4.2 使用示例

以下是uv的基本用法示例：

安装uv：

$ curl -LsSf https://astral.sh/uv/install.sh | sh
$ pip install uv 
$ pipx install uv

管理Python项目：初始化项目并添加fastapi依赖：

$ uv init && uv add "fastapi>=0.112"

此命令会生成pyproject.toml文件，并创建锁文件以保证项目环境的一致性。通过uv run可以直接在项目环境中执行命令，无需手动激活虚拟环境。

[project]
name = "hello-world"  
version = "0.1.0"
readme = "README.md"
dependencies = ["fastapi>=0.112"]

管理命令行工具：使用uv tool install将命令行工具（如Ruff）安装到隔离的虚拟环境中，使用uvx可直接执行命令，无需额外安装。
执行单文件脚本：考虑如下main.py脚本，依赖于requests和rich，但没有包含任何元数据。通过uv add命令自动将依赖声明嵌入脚本中，并可通过uv run main.py在隔离环境中执行脚本并自动安装依赖。

5. 常见具身智能数据格式以及相互转化

这里我们主要参照了具身智能VLA方向模型fine-tune（单臂）这篇的内容作为本章节。

5.1 各格式概述

NPY: NumPy原生格式，存储单个数组或字典
TFDS: TensorFlow Datasets格式，用于TensorFlow生态系统
RLDS: Robotics Language-conditioned Dataset，机器人任务数据集
HDF5: 分层数据格式，支持大型复杂结构化数据

5.2 NPY 到 RLDS 格式转换

import os
import numpy as np
import tensorflow as tf
import tensorflow_hub as hubdef npy_to_rlds(source_path, target_path, language_instruction, embed_model=None):"""将原始的npy格式数据转换为RLDS格式参数:source_path: 源数据路径，包含多个子文件夹，每个子文件夹表示一个任务target_path: 转换后的RLDS格式数据存储路径language_instruction: 任务指令文本embed_model: 文本嵌入模型，如不提供则不生成嵌入"""# 加载文本嵌入模型（如果提供）language_embedding = Noneif embed_model:language_embedding = embed_model([language_instruction])[0].numpy()# 确保目标目录存在os.makedirs(os.path.join(target_path, "train"), exist_ok=True)os.makedirs(os.path.join(target_path, "val"), exist_ok=True)# 获取所有子文件夹（每个子文件夹代表一个任务）subfolders = [f for f in os.listdir(source_path) if os.path.isdir(os.path.join(source_path, f))]train_count = 0val_count = 0for i, subfolder in enumerate(subfolders):subfolder_path = os.path.join(source_path, subfolder)# 获取该任务的所有npy文件npy_files = sorted([f for f in os.listdir(subfolder_path) if f.startswith('targ') and f.endswith('.npy')],key=lambda x: int(x.replace('targ', '').replace('.npy', '')))if not npy_files:continue# 创建一个完整的episodeepisode = []last_state = np.zeros(7)  # 初始状态for j, npy_file in enumerate(npy_files, 1):file_path = os.path.join(subfolder_path, npy_file)data = np.load(file_path, allow_pickle=True).item()# 构建状态向量state = np.array(data["pose"], dtype=np.float32)state = np.append(state, data["gripper"])# 计算动作（相对于上一状态的变化）action = np.zeros(7, dtype=np.float32)if j > 1:action[:6] = state[:6] - last_state[:6]# 将夹爪动作离散化if state[6] - last_state[6] > 0.1:action[6] = 1elif state[6] - last_state[6] < -0.1:action[6] = 0last_state = state.copy()episode.append({'observation': {'image': data['image'],'state': state,},'action': action,'discount': 1.0,'reward': float(j == len(npy_files)),'is_first': j == 1,'is_last': j == len(npy_files),'is_terminal': j == len(npy_files),'language_instruction': language_instruction,})if language_embedding is not None:episode[-1]['language_embedding'] = language_embedding# 根据索引决定是训练集还是验证集if i % 10 > 1:  # 80%作为训练集save_path = os.path.join(target_path, "train", f"episode_{train_count}.npy")np.save(save_path, episode)train_count += 1else:  # 20%作为验证集save_path = os.path.join(target_path, "val", f"episode_{val_count}.npy")np.save(save_path, episode)val_count += 1print(f"转换完成: {train_count} 个训练样本, {val_count} 个验证样本")return train_count, val_count

5.3 RLDS 到 HDF5 转换

import numpy as np
import h5py
import os
import cv2def rlds_to_hdf5(rlds_path, output_path):"""将RLDS格式数据转换为HDF5格式参数:rlds_path: RLDS数据目录，包含train和val子目录output_path: 输出HDF5文件的目录"""os.makedirs(output_path, exist_ok=True)# 处理训练和验证集for split in ['train', 'val']:split_dir = os.path.join(rlds_path, split)if not os.path.exists(split_dir):continueepisode_files = [f for f in os.listdir(split_dir) if f.endswith('.npy')]for episode_file in episode_files:# 加载RLDS episodeepisode_path = os.path.join(split_dir, episode_file)episode_data = np.load(episode_path, allow_pickle=True)# 输出HDF5文件名episode_id = episode_file.replace('.npy', '')output_file = os.path.join(output_path, f"{episode_id}.hdf5")# 创建HDF5文件with h5py.File(output_file, 'w') as f:# 获取episode长度episode_len = len(episode_data)# 创建数据集actions = np.array([step['action'] for step in episode_data])f.create_dataset('actions', data=actions)# 存储observationsobs_group = f.create_group('observations')# 状态数据qpos = np.array([step['observation']['state'] for step in episode_data])obs_group.create_dataset('qpos', data=qpos)# 图像数据img_group = obs_group.create_group('images')# 遍历每个时间步，压缩并存储图像for cam_name in ['main']:  # 可以根据实际相机名称扩展img_data = []for step in episode_data:# 将图像编码为JPEG格式的字节数组img = step['observation']['image']_, encoded_img = cv2.imencode('.jpg', img)img_data.append(np.void(encoded_img.tobytes()))# 创建变长字节数组数据集dt = h5py.special_dtype(vlen=np.dtype('uint8'))img_dataset = img_group.create_dataset(cam_name, shape=(episode_len,), dtype=dt)for i, img_bytes in enumerate(img_data):img_dataset[i] = img_bytes# 存储语言指令if 'language_instruction' in episode_data[0]:instruction = episode_data[0]['language_instruction']f.create_dataset('instruction', data=np.string_(instruction))# 存储奖励和终止状态rewards = np.array([step.get('reward', 0.0) for step in episode_data])terminals = np.array([step.get('is_terminal', False) for step in episode_data])f.create_dataset('rewards', data=rewards)f.create_dataset('terminals', data=terminals)print(f"已保存: {output_file}")

5.4 HDF5 到 RLDS 转换

import os
import numpy as np
import h5py
import cv2def hdf5_to_rlds(hdf5_dir, output_dir, split_ratio=0.8):"""将HDF5格式转换为RLDS格式参数:hdf5_dir: HDF5文件目录output_dir: RLDS输出目录split_ratio: 训练集比例"""os.makedirs(os.path.join(output_dir, 'train'), exist_ok=True)os.makedirs(os.path.join(output_dir, 'val'), exist_ok=True)hdf5_files = [f for f in os.listdir(hdf5_dir) if f.endswith('.hdf5')]np.random.shuffle(hdf5_files)# 根据比例分割训练和验证集split_idx = int(len(hdf5_files) * split_ratio)train_files = hdf5_files[:split_idx]val_files = hdf5_files[split_idx:]train_count = 0val_count = 0# 处理训练集for file in train_files:file_path = os.path.join(hdf5_dir, file)episode_id = train_counttrain_count += _convert_hdf5_to_rlds(file_path, os.path.join(output_dir, 'train'), episode_id)# 处理验证集for file in val_files:file_path = os.path.join(hdf5_dir, file)episode_id = val_countval_count += _convert_hdf5_to_rlds(file_path, os.path.join(output_dir, 'val'), episode_id)print(f"转换完成: {train_count} 个训练样本, {val_count} 个验证样本")def _convert_hdf5_to_rlds(hdf5_path, output_dir, episode_id):"""将单个HDF5文件转换为RLDS格式"""with h5py.File(hdf5_path, 'r') as f:# 获取数据actions = f['actions'][:]qpos = f['observations']['qpos'][:]# 提取图像数据images = []if 'images' in f['observations']:cam_name = list(f['observations']['images'].keys())[0]  # 使用第一个相机for img_bytes in f['observations']['images'][cam_name]:# 解码图像nparr = np.frombuffer(img_bytes, np.uint8)img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)images.append(img)# 获取指令（如果有）instruction = ""if 'instruction' in f:instruction = f['instruction'][()].decode('utf-8')# 构建RLDS格式episodeepisode = []episode_len = len(actions)for i in range(episode_len):step = {'observation': {'state': qpos[i],},'action': actions[i],'discount': 1.0,'is_first': i == 0,'is_last': i == episode_len - 1,'is_terminal': i == episode_len - 1,}# 添加图像（如果有）if images:step['observation']['image'] = images[i]# 添加语言指令（如果有）if instruction:step['language_instruction'] = instruction# 添加奖励（如果有）if 'rewards' in f:step['reward'] = f['rewards'][i]else:step['reward'] = float(i == episode_len - 1)  # 默认只在最后一步有奖励episode.append(step)# 保存RLDS格式数据output_path = os.path.join(output_dir, f"episode_{episode_id}.npy")np.save(output_path, episode)return 1  # 返回成功转换的episode数量