Introdução aos Processadores

Em robótica, há uma incompatibilidade fundamental entre os dados produzidos por robôs e humanos e o que os modelos de aprendizado de máquina esperam. Robôs geram dados brutos de sensores, como imagens de câmera e posições de juntas, que precisam de normalização, agrupamento (batching) e alocação de dispositivo antes que os modelos possam processá-los. Instruções em linguagem natural de humanos devem ser tokenizadas em representações numéricas, e diferentes robôs usam diferentes sistemas de coordenadas que precisam de padronização.

O desafio se estende também às saídas dos modelos. Os modelos podem posicionar efluentes finais, enquanto os robôs precisam de comandos no espaço de juntas, ou os teleoperadores produzem movimentos relativos, enquanto os robôs esperam comandos absolutos. As previsões do modelo são frequentemente normalizadas e precisam de conversão de volta para as escalas do mundo real.

A tradução entre domínios adiciona outra camada de complexidade. Os dados de treinamento de uma configuração de robô precisam de adaptação para implantação em hardware diferente, os modelos treinados com configurações específicas de câmera devem funcionar com novos arranjos e os conjuntos de dados com convenções de nomenclatura diferentes precisam de harmonização.

É aí que os processadores entram em cena. Eles funcionam como tradutores universais que preenchem essas lacunas, garantindo um fluxo de dados sem atritos dos sensores para os modelos e para os atuadores. Os processadores lidam com todas as etapas de pré-processamento e pós-processamento necessárias para converter dados brutos do ambiente em entradas prontas para o modelo e vice-versa.

Isso significa que sua política favorita pode ser usada da seguinte forma:

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)

O que são Processadores?

Em robótica, os dados vêm em muitas formas: imagens de câmeras, posições de juntas de sensores, instruções de texto de usuários, entre outros. Cada tipo de dados requer transformações específicas antes que um modelo possa usá-los de forma eficaz. Os modelos precisam que esses dados estejam:

• Normalizado: Escalonado para intervalos apropriados para o processamento por redes neurais
• Em lote: Organizado com as dimensões adequadas para processamento em lote
• Tokenizado: Texto convertido em representações numéricas
• Dispositivo-alvo: Movido para o hardware adequado (CPU/GPU)
• Convertido de tipo: Convertido para os tipos de dados apropriados

Os processadores realizam essas transformações por meio de etapas composáveis e reutilizáveis que podem ser encadeadas em pipelines. Pense neles como uma linha de montagem modular em que cada estação executa uma transformação específica nos seus dados.

Conceitos Fundamentais

EnvTransition: O Contêiner Universal de Dados

O EnvTransition é a estrutura de dados fundamental que flui por todos os processadores. É um dicionário tipado que representa uma interação completa entre o robô e o ambiente:

• OBSERVAÇÃO: Todos os dados dos sensores (imagens, estados, propriocepção)
• ACTION: A ação a ser executada ou que foi executada
• REWARD: Sinal de aprendizado por reforço
• DONE/TRUNCATED: Indicadores de limite de episódio
• INFO: Metadados arbitrários
• COMPLEMENTARY_DATA: descrições de tarefas, índices, bandeiras de preenchimento, dados entre etapas

Os conversores em lote mantêm metadados de indexação, como index, episode_index e frame_index, em COMPLEMENTARY_DATA. Em uma ida e volta de batch_to_transition / transition_to_batch, frame_index retém seu valor original de escalar ou tensor. Se a entrada omitir frame_index, os conversores não o adicionam.

ProcessorStep: o bloco de construção

Um ProcessorStep é uma unidade de transformação única que processa transições. É uma classe base abstrata com dois métodos obrigatórios:

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__ é o núcleo da sua etapa de processamento. Ele recebe um EnvTransition e retorna um EnvTransition modificado.

transform_features é usado para declarar como esta etapa transforma as formas/tipos de recursos.

DataProcessorPipeline: O Orquestrador Genérico

O DataProcessorPipeline[TInput, TOutput] encadeia múltiplas instâncias de 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 vs PolicyProcessorPipeline

A principal distinção está nas estruturas de dados que eles manipulam:

Use RobotProcessorPipeline para interfaces de hardware de robôs:

# 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
}

Use PolicyProcessorPipeline para treinamento de modelos e processamento em lote:

# 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

Funções de conversão

LeRobot fornece funções de conversão para conectar diferentes formatos de dados em lerobot.processor.converters. Essas funções lidam com as traduções cruciais entre as estruturas de dados de hardware de robôs, os formatos de modelos de política e a representação interna EnvTransition que flui pelos pipelines de processadores.

O ponto-chave é que os conversores de hardware robótico trabalham com valores individuais e dicionários, enquanto os conversores de política/treinamento trabalham com tensores em lote e saídas de modelo. As funções de conversor lidam automaticamente com as diferenças estruturais, permitindo que suas etapas de processador se concentrem nas transformações principais, sem se preocupar com a compatibilidade do formato de dados.

Exemplos de Processador

Os exemplos a seguir demonstram configurações de processador do mundo real para treinamento e inferência de políticas.

Aqui está um exemplo de processador para treinamento e inferência de política:

# 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,
)

Uma interação entre um robô e uma política com processadores

O cenário mais comum no mundo real combina ambos os tipos de pipeline: o hardware do robô gera observações que precisam de processamento pela política, e as saídas da política precisam de pós-processamento compatível com o robô:

# 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)

Contratos de Recursos: Transformação de Forma e Tipo

Processadores não apenas transformam dados - eles também podem alterar a própria estrutura dos dados. O método transform_features() declara essas alterações, o que é crucial para o registro de conjuntos de dados e a criação de políticas.

Por que os Contratos de Recursos Importam

Ao criar conjuntos de dados ou políticas, o LeRobot precisa saber:

• Quais campos de dados existirão após o processamento
• Quais formas e tipos cada campo terá
• Como configurar modelos para a estrutura de dados esperada

# 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

Funções de especificação de recursos

create_initial_features() e aggregate_pipeline_dataset_features() resolvem um problema crítico na criação de datasets: determinar a estrutura de dados final exata antes de qualquer dado ser processado. Como os pipelines de processadores podem adicionar novos recursos (como campos de velocidade), alterar as formas dos tensores (como o recorte de imagens) ou renomear chaves, os datasets precisam conhecer a especificação de saída completa antecipadamente para alocar o armazenamento adequado e definir os esquemas. Essas funções trabalham em conjunto começando com as especificações de hardware do robô (create_initial_features()) e, em seguida, simulando a transformação de todo o pipeline (aggregate_pipeline_dataset_features()) para calcular o dicionário final de recursos que é passado para LeRobotDataset.create(), garantindo o alinhamento perfeito entre o que os processadores produzem e o que os datasets esperam armazenar.

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
    ...
)

Etapas comuns do Processor

LeRobot fornece muitos passos de processador registrados. Aqui estão os processadores principais mais comumente usados:

Processadores Essenciais

• normalizer_processor: Normaliza observações/ações usando estatísticas do conjunto de dados (média/desvio padrão ou mín/máx)
• device_processor: Move os tensors para CPU/GPU com conversão opcional de dtype
• to_batch_processor: Adiciona dimensões de lote às transições para compatibilidade com o modelo
• rename_observations_processor: Renomeia as chaves de observação usando dicionários de mapeamento
• tokenizer_processor: Tokeniza descrições de tarefas de linguagem natural em tokens e máscaras de atenção

Próximas etapas

• Implemente o seu próprio processador - Crie etapas de processador personalizadas
• Depure seu pipeline - Solucione problemas e otimize pipelines
• Processadores para robôs e teleoperadores - Padrões de integração em cenários reais

Resumo

Processadores resolvem o problema de tradução de dados em robótica ao fornecer:

• Transformações modulares: Etapas de processamento composáveis e reutilizáveis
• Segurança de tipos: Pipelines genéricos com verificação em tempo de compilação
• Otimização de desempenho: operações aceleradas por GPU
• Distinção entre Robot/Policy: Pipelines separados para diferentes estruturas de dados
• Ecossistema abrangente: mais de 30 processadores registrados para tarefas comuns

O ponto-chave: RobotProcessorPipeline lida com dados de hardware robótico não loteados, enquanto PolicyProcessorPipeline lida com dados de modelo loteados. Escolha a ferramenta certa para a estrutura dos seus dados!