Introducción a los Procesadores

En robótica, existe una discrepancia fundamental entre los datos que producen los robots y los humanos y lo que los modelos de aprendizaje automático esperan. Los robots emiten datos de sensores crudos, como imágenes de cámara y posiciones de articulaciones, que requieren normalización, agrupación (batching) y colocación en el dispositivo antes de que los modelos puedan procesarlos. Las instrucciones lingüísticas de los humanos deben tokenizarse en representaciones numéricas, y diferentes robots utilizan diferentes sistemas de coordenadas que necesitan estandarización.

El desafío también se extiende a las salidas del modelo. Los modelos pueden generar posiciones del efector final, mientras que los robots necesitan comandos en el espacio de articulaciones, o los teleoperadores producen movimientos relativos, mientras que los robots esperan comandos absolutos. Las predicciones del modelo suelen estar normalizadas y requieren una conversión de vuelta a las escalas del mundo real.

La traducción entre dominios añade otra capa de complejidad. Los datos de entrenamiento de una configuración de robot necesitan adaptación para su despliegue en hardware diferente, los modelos entrenados con configuraciones de cámara específicas deben funcionar con nuevos arreglos y los conjuntos de datos con convenciones de nomenclatura diferentes necesitan armonización.

Es ahí donde entran en juego los procesadores. Actúan como traductores universales que colman estas brechas, garantizando un flujo de datos sin interrupciones desde los sensores hasta los modelos y los actuadores. Los procesadores gestionan todos los pasos de preprocesamiento y posprocesamiento necesarios para convertir los datos crudos del entorno en entradas listas para el modelo, y viceversa.

Esto significa que su política favorita se puede usar de la siguiente manera:

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)

¿Qué son los Procesadores?

En robótica, los datos llegan en muchas formas: imágenes de cámaras, posiciones de articulaciones de sensores, instrucciones de texto de los usuarios, entre otras. Cada tipo de datos requiere transformaciones específicas antes de que un modelo pueda utilizarlos de manera efectiva. Los modelos necesitan que estos datos sean:

• Normalizado: Escalado a los rangos adecuados para el procesamiento de redes neuronales
• Por lotes: Organizado con las dimensiones adecuadas para el procesamiento por lotes
• Tokenizado: Texto convertido en representaciones numéricas
• Colocado en el dispositivo: Movido al hardware adecuado (CPU/GPU)
• Conversión de tipo: Se convierte a los tipos de datos adecuados

Los procesadores gestionan estas transformaciones mediante pasos componibles y reutilizables que pueden encadenarse en pipelines. Piense en ellos como una línea de ensamblaje modular donde cada estación realiza una transformación específica en sus datos.

Conceptos principales

EnvTransition: el contenedor de datos universal

El EnvTransition es la estructura de datos fundamental que fluye a través de todos los procesadores. Es un diccionario tipado que representa una interacción completa entre el robot y el entorno:

• OBSERVACIÓN: Todos los datos de los sensores (imágenes, estados, propiocepción)
• ACTION: La acción a ejecutar o que se ejecutó
• REWARD: Señal de aprendizaje por refuerzo
• DONE/TRUNCATED: Indicadores de límite de episodio
• INFO: Metadatos arbitrarios
• COMPLEMENTARY_DATA: descripciones de tareas, índices, banderas de relleno, datos entre pasos

Los conversores por lotes mantienen los metadatos de indexación, como index, episode_index y frame_index, en COMPLEMENTARY_DATA. En un ciclo de ida y vuelta de batch_to_transition / transition_to_batch, frame_index conserva su valor original de escalar o tensor. Si la entrada omite frame_index, los conversores no lo añaden.

ProcessorStep: el bloque de construcción

Un ProcessorStep es una unidad de transformación individual que procesa transiciones. Es una clase base abstracta con dos métodos obligatorios:

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__ es el núcleo de su paso de procesador. Toma un EnvTransition y devuelve un EnvTransition modificado.

transform_features se utiliza para declarar cómo este paso transforma las formas/tipos de las características.

DataProcessorPipeline: el orquestador genérico

La DataProcessorPipeline[TInput, TOutput] encadena múltiples instancias 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 frente a PolicyProcessorPipeline

La distinción clave radica en las estructuras de datos que manejan:

Use RobotProcessorPipeline para interfaces de hardware de robots:

# 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 el entrenamiento de modelos y el procesamiento por lotes:

# 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

Funciones de conversión

LeRobot proporciona funciones de conversión para conectar diferentes formatos de datos en lerobot.processor.converters. Estas funciones gestionan las traducciones esenciales entre las estructuras de datos de hardware de robots, los formatos de modelos de política y la representación interna EnvTransition que fluye a través de las tuberías de procesadores.

La clave es que los convertidores de hardware robótico trabajan con valores individuales y diccionarios, mientras que los convertidores de política/entrenamiento trabajan con tensores por lotes y salidas de modelos. Las funciones de conversor manejan automáticamente las diferencias estructurales, por lo que los pasos de su procesador pueden centrarse en las transformaciones principales sin preocuparse por la compatibilidad del formato de datos.

Ejemplos de procesadores

Los siguientes ejemplos demuestran configuraciones de procesador del mundo real para el entrenamiento y la inferencia de políticas.

A continuación se muestra un ejemplo de procesador para el entrenamiento y la inferencia de la 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,
)

Una interacción entre un robot y una política con procesadores

El escenario más común en el mundo real combina ambos tipos de pipeline: el hardware del robot genera observaciones que requieren procesamiento de la política, y las salidas de la política requieren un posprocesamiento compatible con el robot:

# 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 características: transformación de forma y tipo

Los procesadores no solo transforman los datos, también pueden modificar la propia estructura de los datos. El método transform_features() declara estos cambios, lo cual es crucial para el registro de conjuntos de datos y la creación de políticas.

Por qué son importantes los contratos de características

Al construir conjuntos de datos o políticas, LeRobot necesita saber:

• Qué campos de datos existirán después del procesamiento
• Qué formas y tipos tendrá cada campo
• Cómo configurar los modelos para la estructura de datos 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

Funciones de especificación de características

create_initial_features() y aggregate_pipeline_dataset_features() resuelven un problema crítico en la creación de conjuntos de datos: determinar la estructura de datos final exacta antes de que se procese cualquier dato. Dado que las tuberías de procesadores pueden añadir nuevas características (como campos de velocidad), cambiar las formas de los tensores (como recortar imágenes) o renombrar claves, los conjuntos de datos necesitan conocer la especificación de salida completa de antemano para asignar el almacenamiento adecuado y definir los esquemas. Estas funciones trabajan juntas comenzando con las especificaciones del hardware del robot (create_initial_features()) y luego simulando la transformación completa de la tubería (aggregate_pipeline_dataset_features()) para calcular el diccionario de características final que se pasa a LeRobotDataset.create(), garantizando una alineación perfecta entre lo que los procesadores producen y lo que los conjuntos de datos esperan almacenar.

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

Pasos comunes del procesador

LeRobot ofrece muchos pasos de procesador registrados. Estos son los procesadores principales más utilizados:

Procesadores esenciales

• normalizer_processor: Normaliza observaciones/acciones usando estadísticas del conjunto de datos (media/desviación estándar o min/max)
• device_processor: Mover tensores a CPU/GPU con conversión opcional de dtype
• to_batch_processor: Añade dimensiones de lote a las transiciones para la compatibilidad con el modelo
• rename_observations_processor: Renombrar las claves de observación usando diccionarios de mapeo
• tokenizer_processor: Tokeniza las descripciones de tareas de lenguaje natural en tokens y máscaras de atención

Próximos pasos

• Implementa tu propio procesador - Crea pasos de procesador personalizados
• Depura tu pipeline - Soluciona problemas y optimiza pipelines
• Procesadores para robots y teleoperadores - Patrones de integración en el mundo real

Resumen

Los procesadores resuelven el problema de traducción de datos en robótica al proporcionar:

• Transformaciones modulares: pasos de procesamiento componibles y reutilizables
• Seguridad de tipos: Pipelines genéricos con comprobación en tiempo de compilación
• Optimización del rendimiento: operaciones aceleradas por GPU
• Distinción entre Robot y Policy: Píptinas separadas para diferentes estructuras de datos
• Ecosistema integral: más de 30 procesadores registrados para tareas comunes

La clave: RobotProcessorPipeline maneja datos de hardware robótico sin lotes, mientras que PolicyProcessorPipeline maneja datos de modelo en lotes. ¡Elige la herramienta adecuada para tu estructura de datos!