Введение в процессоры

В робототехнике существует фундаментальное несоответствие между данными, которые генерируют роботы и люди, и тем, что ожидают модели машинного обучения. Роботы выводят сырые данные датчиков, такие как изображения с камер и положения суставов, которые требуют нормализации, формирования батчей и размещения на устройстве, прежде чем модели смогут их обработать. Языковые инструкции от людей должны быть токенизированы в числовые представления, а разные роботы используют различные системы координат, которые требуют стандартизации.

Проблема распространяется и на выходные данные моделей. Модели могут выдавать позиции исполнительных органов, тогда как роботам необходимы команды в пространстве суставов, либо телеоператоры формируют относительные движения, а роботы ожидают абсолютные команды. Прогнозы модели часто нормализованы и требуют обратного преобразования к реальным масштабам.

Перевод между доменами добавляет ещё один уровень сложности. Обучающие данные, полученные на одной робототехнической конфигурации, требуют адаптации для развёртывания на другом оборудовании, модели, обученные с использованием конкретных конфигураций камер, должны работать с новыми расстановками, а наборы данных с различными соглашениями об именовании нуждаются в согласовании.

Именно здесь на помощь приходят процессоры. Они выступают в роли универсальных трансляторов, устраняющих эти различия и обеспечивающих бесшовный поток данных от датчиков к моделям и далее к исполнительным механизмам. Процессоры выполняют все необходимые этапы предобработки и постобработки для преобразования сырых данных из окружающей среды в входные данные, готовые для модели, и обратно.

Это означает, что ваша любимая политика может использоваться следующим образом:

import torch

from lerobot.datasets import LeRobotDataset
from lerobot.policies import make_pre_post_processors
from lerobot.policies.your_policy import YourPolicy
from lerobot.processor 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)

Что такое процессоры?

В робототехнике данные поступают в самых разных формах: изображения с камер, положения суставов от датчиков, текстовые инструкции от пользователей и многое другое. Каждый тип данных требует специфических преобразований, чтобы модель могла эффективно их использовать. Модели нуждаются в том, чтобы эти данные были:

• Нормализованные: масштабированные до соответствующих диапазонов для обработки нейронной сетью
• Пакетная обработка: организовано с правильными размерностями для пакетной обработки
• Токенизированный: текст, преобразованный в числовые представления
• Размещённое на устройстве: перемещено на соответствующее оборудование (CPU/GPU)
• Преобразование типов: приведение к соответствующим типам данных

Процессоры выполняют эти преобразования с помощью компонуемых, переиспользуемых шагов, которые можно объединять в конвейеры. Представьте их как модульную сборочную линию, где каждая станция выполняет определённое преобразование ваших данных.

Основные концепции

EnvTransition: универсальный контейнер данных

EnvTransition — это базовая структура данных, которая проходит через все процессоры. Это типизированный словарь, представляющий полное взаимодействие робота с окружением:

• НАБЛЮДЕНИЕ: все данные датчиков (изображения, состояния, проприоцепция)
• ACTION: действие для выполнения или выполненное действие
• REWARD: сигнал обучения с подкреплением
• DONE/TRUNCATED: индикаторы границ эпизода
• INFO: Произвольные метаданные
• COMPLEMENTARY_DATA: описания задач, индексы, флаги отступов, данные между шагами

Пакетные конвертеры сохраняют метаданные индексации, такие как index, episode_index и frame_index, в COMPLEMENTARY_DATA. При циклическом преобразовании batch_to_transition / transition_to_batch значение frame_index сохраняет исходный скаляр или тензор. Если во входных данных отсутствует frame_index, конвертеры не добавляют его.

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 используется для объявления того, как этот шаг преобразует формы/типы признаков.

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 для интерфейсов робототехнического оборудования:

# 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, которое проходит через конвейеры процессоров.

Ключевой момент состоит в том, что конвертеры аппаратного обеспечения роботов работают с отдельными значениями и словарями, тогда как конвертеры политик/обучения оперируют батчами тензоров и выводами моделей. Функции конвертеров автоматически обрабатывают структурные различия, поэтому шаги вашего процессора могут сосредоточиться на основных преобразованиях, не беспокоясь о совместимости форматов данных.

Примеры процессоров

Приведённые ниже примеры демонстрируют конфигурации процессоров для обучения и инференса политик в реальных сценариях.

Вот пример процессора для обучения и инференса политики:

# 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 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 с необязательным преобразованием типа
• to_batch_processor: добавляет размерности батча к переходам для совместимости с моделью
• rename_observations_processor: Переименование ключей наблюдений с использованием словарей соответствий
• tokenizer_processor: Токенизация описаний задач на естественном языке в токены и маски внимания

Дальнейшие шаги

• Реализуйте собственный процессор — Создайте собственные шаги процессора
• Отладка вашего конвейера - Диагностика и оптимизация конвейеров
• Процессоры для роботов и телеоператоров — Паттерны интеграции в реальных условиях

Краткое описание

Процессоры решают проблему преобразования данных в робототехнике, предоставляя:

• Модульные преобразования: Компонуемые, переиспользуемые шаги обработки
• Типобезопасность: универсальные конвейеры с проверкой на этапе компиляции
• Оптимизация производительности: операции с ускорением на GPU
• Разделение Robot/Policy: отдельные конвейеры для разных структур данных
• Комплексная экосистема: более 30 зарегистрированных процессоров для типовых задач

Ключевой момент: RobotProcessorPipeline обрабатывает необъединённые в батчи данные роботизированного оборудования, а PolicyProcessorPipeline — данные модели, сгруппированные в батчи. Выбирайте подходящий инструмент в зависимости от структуры ваших данных!