LeRobot 文件

處理器簡介

LeRobot

加入 Hugging Face 社群

並獲得增強的文件體驗

在模型、資料集和 Spaces 上進行協作

透過加速推論實現更快的範例

切換文件主題

開始使用

處理器簡介

在機器人學中，機器人與人類產生的數據，以及機器學習模型所期望的數據之間存在根本性的差異。機器人輸出的是原始感測器數據（如攝影機影像和關節位置），在模型處理之前需要進行標準化、批次處理（batching）和設備放置（device placement）。來自人類的語言指令必須被轉換（tokenized）為數值表示，且不同的機器人使用不同的座標系，這些都需要進行標準化。

這項挑戰同樣延伸至模型輸出。模型可能會輸出末端執行器的位置，而機器人則需要關節空間（joint-space）的指令；或者遠端操作者產生的是相對移動，但機器人卻預期絕對指令。模型預測通常是經過標準化的，需要轉換回真實世界的尺度。

跨領域轉換增加了另一層複雜性。來自一種機器人配置的訓練數據需要進行調整以部署到不同的硬體上；使用特定攝影機配置訓練的模型必須在新的佈置中運作；而具有不同命名慣例的資料集則需要進行協調。

這就是處理器（Processors）的作用所在。 它們作為通用轉譯器來填補這些差距，確保數據從感測器到模型，再到執行器的順暢流動。處理器處理將原始環境數據轉換為模型可用輸入（反之亦然）所需的所有預處理和後處理步驟。

這意味著您可以像這樣使用您偏好的策略（policy）：

import torch

from lerobot.datasets.lerobot_dataset import LeRobotDataset
from lerobot.policies.factory import make_pre_post_processors
from lerobot.policies.your_policy import YourPolicy
from lerobot.processor.pipeline import RobotProcessorPipeline, PolicyProcessorPipeline
dataset = LeRobotDataset("hf_user/dataset", episodes=[0])
sample = dataset[10]

model = YourPolicy.from_pretrained(
    "hf_user/model",
)
model.eval()
model.to("cuda")
preprocessor, postprocessor = make_pre_post_processors(model.config, pretrained_path="hf_user/model", dataset_stats=dataset.meta.stats)

preprocessed_sample = preprocessor(sample)
action = model.select_action(preprocessed_sample)
postprocessed_action = postprocessor(action)

什麼是處理器？

在機器人學中，數據有多種形式：來自攝影機的影像、感測器的關節位置、使用者的文字指令等。在模型能有效使用每種數據之前，都需要進行特定的轉換。模型要求這些數據必須：

標準化（Normalized）：縮放到適合神經網路處理的範圍。
批次化（Batched）：組織成適合批次處理的維度。
符記化（Tokenized）：將文字轉換為數值表示。
設備放置（Device-placed）：移動到正確的硬體（CPU/GPU）。
類型轉換（Type-converted）：轉換為適當的數據類型。

處理器通過可組合、可重用的步驟來處理這些轉換，這些步驟可以串聯成管線（pipelines）。您可以將其想像成一條模組化的組裝線，每個工作站對您的數據執行特定的轉換。

核心概念

EnvTransition：通用數據容器

EnvTransition 是流經所有處理器的基本數據結構。它是一個型別字典（typed dictionary），代表了一次完整的機器人與環境互動：

OBSERVATION（觀測）：所有感測器數據（影像、狀態、本體感覺）
ACTION（動作）：要執行或已執行的動作
REWARD（獎勵）：強化學習訊號
DONE/TRUNCATED（結束/截斷）：情境邊界指標
INFO（資訊）：任意元數據
COMPLEMENTARY_DATA（補充數據）：任務描述、索引、填充標記、步驟間數據

ProcessorStep：建構模組

ProcessorStep 是處理轉換的單一單元。這是一個抽象基類，包含兩個必要方法：

from lerobot.processor import ProcessorStep, EnvTransition

class MyProcessorStep(ProcessorStep):
    """Example processor step - inherit and implement abstract methods."""

    def __call__(self, transition: EnvTransition) -> EnvTransition:
        """Transform the transition - REQUIRED abstract method."""
        # Your processing logic here
        return transition

    def transform_features(self, features):
        """Declare how this step transforms feature shapes/types - REQUIRED abstract method."""
        return features  # Most processors return features unchanged

__call__ 是處理步驟的核心。它接收一個 EnvTransition 並回傳修改後的 EnvTransition。

transform_features 用於宣告此步驟如何轉換特徵的形狀（shapes）或型別（types）。

DataProcessorPipeline：通用管線編排器

DataProcessorPipeline[TInput, TOutput] 將多個 ProcessorStep 實例串聯在一起。

from lerobot.processor import RobotProcessorPipeline, PolicyProcessorPipeline

# For robot hardware (unbatched data)
robot_processor = RobotProcessorPipeline[RobotAction, RobotAction](
    steps=[step1, step2, step3],
    name="robot_pipeline"
)

# For model training/inference (batched data)
policy_processor = PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
    steps=[step1, step2, step3],
    name="policy_pipeline"
)

RobotProcessorPipeline 與 PolicyProcessorPipeline

關鍵區別在於它們處理的數據結構：

面向	RobotProcessorPipeline	PolicyProcessorPipeline
輸入	`dict[str, Any]` - 個別機器人值	`dict[str, Any]` - 批次化張量
輸出	`dict[str, Any]` - 個別機器人指令	`torch.Tensor` - 策略預測
使用案例	即時機器人控制	模型訓練/推理
數據格式	未批次化，異質	批次化，同質
範例	`{"joint_1": 0.5}`	`{"observation.state": tensor([[0.5]])}`

請為機器人硬體介面使用 RobotProcessorPipeline

# Robot data structures: dict[str, Any] for observations and actions
robot_obs: dict[str, Any] = {
    "joint_1": 0.5,           # Individual joint values
    "joint_2": -0.3,
    "camera_0": image_array   # Raw camera data
}

robot_action: dict[str, Any] = {
    "joint_1": 0.2,          # Target joint positions
    "joint_2": 0.1,
    "gripper": 0.8
}

請為模型訓練和批次處理使用 PolicyProcessorPipeline

# Policy data structures: batch dicts and tensors
policy_batch: dict[str, Any] = {
    "observation.state": torch.tensor([[0.5, -0.3]]),      # Batched states
    "observation.images.camera0": torch.tensor(...),        # Batched images
    "action": torch.tensor([[0.2, 0.1, 0.8]])              # Batched actions
}

policy_action: torch.Tensor = torch.tensor([[0.2, 0.1, 0.8]])  # Model output tensor

轉換器函式

LeRobot 在 lerobot.processor.converters 中提供了轉換器函式，用於橋接不同的數據格式。這些函式處理機器人硬體數據結構、策略模型格式以及在處理器管線中流動的內部 EnvTransition 表示之間的關鍵轉換。

類別	函式 (Function)	說明
機器人硬體轉換器	`robot_action_to_transition`	機器人字典 → EnvTransition
	`observation_to_transition`	機器人觀測 → EnvTransition
	`transition_to_robot_action`	EnvTransition → 機器人字典
策略/訓練轉換器	`batch_to_transition`	批次字典 → EnvTransition
	`transition_to_batch`	EnvTransition → 批次字典
	`policy_action_to_transition`	策略張量 → EnvTransition
	`transition_to_policy_action`	EnvTransition → 策略張量
工具程式	`create_transition`	建立具備預設值的 Transition
	`identity_transition`	直通式轉換器（Pass-through）

核心見解在於：機器人硬體轉換器處理個別值和字典，而策略/訓練轉換器處理批次化張量和模型輸出。轉換器函式會自動處理結構差異，因此您的處理器步驟可以專注於核心轉換，無需擔心數據格式相容性。

處理器範例

以下範例演示了用於策略訓練和推理的真實世界處理器配置。

這是一個用於策略訓練和推理的處理器範例：

# Training data preprocessing (optimized order for GPU performance)
training_preprocessor = PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
    steps=[
        RenameObservationsProcessorStep(rename_map={}),     # Standardize keys
        AddBatchDimensionProcessorStep(),                   # Add batch dims
        TokenizerProcessorStep(tokenizer_name="...", ...),  # Tokenize language
        DeviceProcessorStep(device="cuda"),                 # Move to GPU first
        NormalizerProcessorStep(features=..., stats=...),   # Normalize on GPU
    ]
)

# Model output postprocessing
training_postprocessor = PolicyProcessorPipeline[torch.Tensor, torch.Tensor](
    steps=[
        DeviceProcessorStep(device="cpu"),                  # Move to CPU
        UnnormalizerProcessorStep(features=..., stats=...), # Denormalize
    ]
    to_transition=policy_action_to_transition,
    to_output=transition_to_policy_action,
)

機器人與使用處理器的策略之間的互動

最常見的真實世界情境結合了兩種類型的管線：機器人硬體產生需要策略處理的觀測結果，而策略輸出則需要機器人相容的後處理。

# Real deployment: Robot sensors → Model → Robot commands
with torch.no_grad():
    while not done:
        raw_obs = robot.get_observation()  # dict[str, Any]

        # Add your robot observation to policy observation processor

        policy_input = policy_preprocessor(raw_obs)  # Batched dict

        policy_output = policy.select_action(policy_input)  # Policy tensor

        policy_action = policy_postprocessor(policy_output)

        # Add your robot action to policy action processor

        robot.send_action(policy_action)

特徵合約：形狀與型別轉換

處理器不僅僅是轉換數據，它們還可以改變數據結構本身。transform_features() 方法會宣告這些變更，這對於資料集錄製和策略建立至關重要。

為什麼特徵合約很重要？

在建立資料集或策略時，LeRobot 需要知道：

處理後將存在哪些數據欄位
每個欄位將具有什麼形狀和型別
如何為預期的數據結構配置模型

# Example: A processor that adds velocity to observations
class VelocityProcessor(ObservationProcessorStep):
    def observation(self, obs):
        new_obs = obs.copy()
        if "observation.state" in obs:
            # concatenate computed velocity field to the state
            new_obs["observation.state"] = self._compute_velocity(obs["observation.state"])
        return new_obs

    def transform_features(self, features):
        """Declare the new velocity field we're adding."""
        state_feature = features[PipelineFeatureType.OBSERVATION].get("observation.state")
        if state_feature:
            double_shape = (state_feature.shape[0] * 2,) if state_feature.shape else (2,)
            features[PipelineFeatureType.OBSERVATION]["observation.state"] = PolicyFeature(
                type=FeatureType.STATE, shape=double_shape
            )
        return features

特徵規範函式

create_initial_features() 和 aggregate_pipeline_dataset_features() 解決了資料集建立的一個關鍵問題：在數據處理之前確定最終的確切數據結構。由於處理器管線可以新增特徵（如速度欄位）、改變張量形狀（如裁切影像）或重新命名鍵值，資料集需要預先知道完整的輸出規範，以分配適當的儲存空間並定義架構。這些函式透過從機器人硬體規範開始（create_initial_features()），然後模擬整個管線轉換過程（aggregate_pipeline_dataset_features()）來共同運作，從而計算出傳遞給 LeRobotDataset.create() 的最終特徵字典，確保處理器輸出與資料集預期儲存的內容完美對齊。

from lerobot.datasets.pipeline_features import aggregate_pipeline_dataset_features

# Start with robot's raw features
initial_features = create_initial_features(
    observation=robot.observation_features,  # {"joint_1.pos": float, "camera_0": (480,640,3)}
    action=robot.action_features            # {"joint_1.pos": float, "gripper.pos": float}
)

# Apply processor pipeline to compute final features
final_features = aggregate_pipeline_dataset_features(
    pipeline=my_processor_pipeline,
    initial_features=initial_features,
    use_videos=True
)

# Use for dataset creation
dataset = LeRobotDataset.create(
    repo_id="my_dataset",
    features=final_features,  # Knows exactly what data to expect
    ...
)

常見處理器步驟

LeRobot 提供了許多已註冊的處理器步驟。以下是最常用的核心處理器：

必要處理器

normalizer_processor：使用資料集統計數據（平均值/標準差或最大值/最小值）來標準化觀測或動作。
device_processor：將張量移動到 CPU/GPU，並可選擇進行型別轉換（dtype conversion）。
to_batch_processor：將批次維度新增至 Transition，以實現模型相容性。
rename_observations_processor：使用映射字典重新命名觀測鍵值。
tokenizer_processor：將自然語言任務描述符記化為 Token 和注意力遮罩（attention masks）。

下一步

實作您自己的處理器 - 建立自定義處理器步驟。
除錯您的管線 - 疑難排解並優化管線。
機器人和遠端操作處理器 - 現實世界整合模式。

總結

處理器透過提供以下功能來解決機器人學中的數據轉換問題：

模組化轉換：可組合、可重用的處理步驟。
型別安全：具有編譯時檢查的通用管線。
效能優化：GPU 加速操作。
機器人/策略區分：針對不同數據結構的獨立管線。
完善的生態系統：超過 30 個註冊處理器，適用於常見任務。

關鍵見解：RobotProcessorPipeline 處理未批次化的機器人硬體數據，而 PolicyProcessorPipeline 處理批次化的模型數據。請為您的數據結構選擇正確的工具！

在 GitHub 上更新

←NVIDIA IsaacLab 競技場環境除錯您的處理器管線→