具身智能算子概述：轨迹切分、传感器对齐、状态与动作标准化及训练数据导出-云原生数据仓库AnalyticDB(AnalyticDB)-阿里云帮助中心

具身智能算子将真实采集的传感器、状态、动作和任务记录整理为可校验的轨迹及训练数据，所有算子均通过 ds.mmops.embodied.<算子>() 接收并返回 Ray Dataset。算子不会从普通视频中推测控制标签。

算子列表

具身智能算子以一次完整的任务轨迹（Episode）为核心组织数据，覆盖从会话切分、观测对齐、状态与动作标准化、任务标注，到采样、质量校验和数据集导出的流程。

算子	用途	核心结果列
segment_episodes	将会话切分为 Episode。	`episodes` 列表
align_sensor_streams	对齐实际观测。	`timestamps_s`、`steps`、偏差与掩码
normalize_state	统一状态的单位与坐标。	`states`、`state_schema`、`conversion`
normalize_actions	整理真实动作及控制时间。	`actions`、`action_timestamps_s`、`duration_s`
annotate_tasks	整理实际任务记录和候选。	`instruction`、`stages`、`outcome`、`provenance`
build_temporal_samples	构建观测与动作窗口。	`samples` 列表
validate_trajectories	按质量规则检查轨迹。	`passed`、`issues`、`metrics`
export_dataset	导出完整清单。	`dataset_uri`、清单及回读报告

输入与执行规则

仅 *_column 选择器读取每行的数据，其余参数是本次调用的全局常量。每个算子接收调用方的 Dataset 并返回新的 Dataset，规则如下：

默认仅保留结果列，可通过 keep_columns 显式保留指定的输入列；与结果列重名时，须先用 Ray Data 原生的 rename_columns 改名。
所有逐行处理的算子均支持 batch_size=1、concurrency=1、num_cpus=1 等调度参数，调度的批大小不改变业务结果的行数。
选择列使用 Ray Data 原生的 select_columns，结果元数据使用 write_parquet() 保存。
每个输入行对应一行结果。episodes 与 samples 保留为列表，仅在显式使用 flat_map 时展开。
session_column、episode_column 可指定 struct 列，也可设为 None 以读取整行。
状态、动作和任务结果相互独立。采样与导出前，须按 episode_id 和真实时间组装，不能将某一个标准化结果当作完整轨迹。

时间、物理量与导出约定

时间单位为秒。源时钟须通过显式的 scale 与 offset_s 映射。
未知的物理量记为 null，并对应无效掩码。
字段顺序、单位、坐标系和旋转表示均须明确。
统计量固定取自训练集。
导出是清单级作业，每行在 output_dir 下获得独占的子目录。LeRobot 格式固定使用 SDK 依赖的 LeRobot 0.4.3 与格式版本 v3.0；RLDS 格式使用 TFDS 版本 1.0.0。详情请参见 export_dataset。

完整示例：校验轨迹质量

说明

示例中的 /shared 是集群所有 Worker 节点均可读写的共享挂载目录。使用前请确认已满足离线算子概述中的前提条件。

以下示例展示 Dataset 输入、validate_trajectories 调用与结果写出。

import ray
from mmops.ray import enable_accessor

ray.init()
enable_accessor()

ds = ray.data.from_items([{"episode": {
    "episode_id": "E-001", "timestamps_s": [0.0, 0.05],
    "states": [[0.1], [0.2]], "actions": [[0.2], [0.3]],
}}])
result = ds.mmops.embodied.validate_trajectories(
    episode_column="episode", keep_columns=["episode"],
    config={"schemas": {"states": {"shape": [1]}, "actions": {"shape": [1]}},
            "rules": {"missing_rate": 0}, "severity": {}},
)

result = result.select_columns(['episode_id', 'passed', 'issues', 'metrics', 'recommendation', 'rule_set_id', 'episode'])
result.write_parquet("/shared/output/embodied-overview/metadata")

示例返回结果

以下结果由上述输入通过 Ray Dataset 执行得到，以 JSON 展示取出的行。算子返回 Dataset，take() 仅取少量行到本地查看；多行结果不保证跨分区的顺序。

import json

rows = result.take(1)
print(json.dumps(rows, ensure_ascii=False, indent=2))

[
  {
    "episode_id": "E-001",
    "passed": true,
    "issues": [],
    "metrics": {"step_count": 2, "missing_rate": {"states": 0.0, "actions": 0.0}, "max_skew_s": null},
    "recommendation": "keep",
    "rule_set_id": "dd6846dc21e6d2c9",
    "episode": {
      "actions": [[0.2], [0.3]],
      "episode_id": "E-001",
      "states": [[0.1], [0.2]],
      "timestamps_s": [0.0, 0.05]
    }
  }
]