Merge branch 'main' into feat/add_rewind

examples/dataset/load_lerobot_dataset.py

@@ -132,17 +132,15 @@ print(f"\n{dataset[0][camera_key].shape=}")  # (4, c, h, w)
 print(f"{dataset[0]['observation.state'].shape=}")  # (6, c)
 print(f"{dataset[0]['action'].shape=}\n")  # (64, c)
 
-# Finally, our datasets are fully compatible with PyTorch dataloaders and samplers because they are just
-# PyTorch datasets.
-dataloader = torch.utils.data.DataLoader(
-    dataset,
-    num_workers=4,
-    batch_size=32,
-    shuffle=True,
-)
-
-for batch in dataloader:
-    print(f"{batch[camera_key].shape=}")  # (32, 4, c, h, w)
-    print(f"{batch['observation.state'].shape=}")  # (32, 6, c)
-    print(f"{batch['action'].shape=}")  # (32, 64, c)
-    break
+if __name__ == "__main__":
+    dataloader = torch.utils.data.DataLoader(
+        dataset,
+        num_workers=4,
+        batch_size=32,
+        shuffle=True,
+    )
+    for batch in dataloader:
+        print(f"{batch[camera_key].shape=}")  # (32, 4, c, h, w)
+        print(f"{batch['observation.state'].shape=}")  # (32, 6, c)
+        print(f"{batch['action'].shape=}")  # (32, 64, c)
+        break

examples/port_datasets/slurm_aggregate_shards.py

@@ -15,16 +15,12 @@
 # limitations under the License.
 
 import argparse
-import logging
 from pathlib import Path
 
 from datatrove.executor import LocalPipelineExecutor
 from datatrove.executor.slurm import SlurmPipelineExecutor
 from datatrove.pipeline.base import PipelineStep
-from port_datasets.droid_rlds.port_droid import DROID_SHARDS
-
-from lerobot.datasets.aggregate import aggregate_datasets
-from lerobot.utils.utils import init_logging
+from port_droid import DROID_SHARDS
 
 
 class AggregateDatasets(PipelineStep):
@@ -38,6 +34,11 @@ class AggregateDatasets(PipelineStep):
         self.aggr_repo_id = aggregated_repo_id
 
     def run(self, data=None, rank: int = 0, world_size: int = 1):
+        import logging
+
+        from lerobot.datasets.aggregate import aggregate_datasets
+        from lerobot.utils.utils import init_logging
+
         init_logging()
 
         # Since aggregate_datasets already handles parallel processing internally,

examples/port_datasets/slurm_port_shards.py

@@ -20,7 +20,7 @@ from pathlib import Path
 from datatrove.executor import LocalPipelineExecutor
 from datatrove.executor.slurm import SlurmPipelineExecutor
 from datatrove.pipeline.base import PipelineStep
-from port_datasets.droid_rlds.port_droid import DROID_SHARDS
+from port_droid import DROID_SHARDS
 
 
 class PortDroidShards(PipelineStep):
@@ -35,7 +35,7 @@ class PortDroidShards(PipelineStep):
 
     def run(self, data=None, rank: int = 0, world_size: int = 1):
         from datasets.utils.tqdm import disable_progress_bars
-        from port_datasets.droid_rlds.port_droid import port_droid, validate_dataset
+        from port_droid import port_droid, validate_dataset
 
         from lerobot.utils.utils import init_logging
 

examples/port_datasets/slurm_upload.py

@@ -24,7 +24,7 @@ from datatrove.executor.slurm import SlurmPipelineExecutor
 from datatrove.pipeline.base import PipelineStep
 from huggingface_hub import HfApi
 from huggingface_hub.constants import REPOCARD_NAME
-from port_datasets.droid_rlds.port_droid import DROID_SHARDS
+from port_droid import DROID_SHARDS
 
 from lerobot.datasets.lerobot_dataset import CODEBASE_VERSION, LeRobotDatasetMetadata
 from lerobot.datasets.utils import create_lerobot_dataset_card
@@ -185,11 +185,11 @@ class UploadDataset(PipelineStep):
 
 
 def make_upload_executor(
-    repo_id, job_name, logs_dir, workers, partition, cpus_per_task, mem_per_cpu, slurm=True
+    repo_id, job_name, logs_dir, workers, partition, cpus_per_task, mem_per_cpu, private=False, slurm=True
 ):
     kwargs = {
         "pipeline": [
-            UploadDataset(repo_id),
+            UploadDataset(repo_id, private=private),
         ],
         "logging_dir": str(logs_dir / job_name),
     }
@@ -267,6 +267,12 @@ def main():
         default="1950M",
         help="Memory per cpu that each worker will use.",
     )
+    parser.add_argument(
+        "--private",
+        action="store_true",
+        default=False,
+        help="Whether to create a private repository.",
+    )
 
     init_logging()
 

examples/rtc/eval_dataset.py

@@ -0,0 +1,951 @@
+#!/usr/bin/env python
+
+# Copyright 2025 The HuggingFace Inc. team. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Evaluate Real-Time Chunking (RTC) performance on dataset samples.
+
+This script takes two random samples from a dataset:
+- Uses actions from the first sample as previous chunk
+- Generates new actions for the second sample with and without RTC
+
+It compares action predictions with and without RTC on dataset samples,
+measuring consistency and ground truth alignment.
+
+Usage:
+    # Basic usage with smolvla policy
+    uv run python examples/rtc/eval_dataset.py \
+        --policy.path=helper2424/smolvla_check_rtc_last3 \
+        --dataset.repo_id=helper2424/check_rtc \
+        --rtc.execution_horizon=8 \
+        --device=mps \
+        --rtc.max_guidance_weight=10.0 \
+        --rtc.prefix_attention_schedule=EXP \
+        --seed=10
+
+    # Basic usage with pi0.5 policy
+    uv run python examples/rtc/eval_dataset.py \
+        --policy.path=lerobot/pi05_libero_finetuned \
+        --dataset.repo_id=HuggingFaceVLA/libero \
+        --rtc.execution_horizon=10 \
+        --device=mps
+        --seed=10
+
+    # Basic usage with pi0.5 policy with cuda device
+    uv run python examples/rtc/eval_dataset.py \
+        --policy.path=lerobot/pi05_libero_finetuned \
+        --dataset.repo_id=HuggingFaceVLA/libero \
+        --rtc.execution_horizon=8 \
+        --device=cuda
+
+    # Basic usage with pi0 policy with cuda device
+    uv run python examples/rtc/eval_dataset.py \
+        --policy.path=lerobot/pi0_libero_finetuned \
+        --dataset.repo_id=HuggingFaceVLA/libero \
+        --rtc.execution_horizon=8 \
+        --device=cuda
+
+    uv run python examples/rtc/eval_dataset.py \
+        --policy.path=lipsop/reuben_pi0 \
+        --dataset.repo_id=ReubenLim/so101_cube_in_cup \
+        --rtc.execution_horizon=8 \
+        --device=cuda
+
+    # With torch.compile for faster inference (PyTorch 2.0+)
+    # Note: CUDA graphs disabled by default due to in-place ops in denoising loop
+    uv run python examples/rtc/eval_dataset.py \
+        --policy.path=helper2424/smolvla_check_rtc_last3 \
+        --dataset.repo_id=helper2424/check_rtc \
+        --rtc.execution_horizon=8 \
+        --device=mps \
+        --use_torch_compile=true \
+        --torch_compile_mode=max-autotune
+
+    # With torch.compile on CUDA (CUDA graphs disabled by default)
+    uv run python examples/rtc/eval_dataset.py \
+        --policy.path=helper2424/smolvla_check_rtc_last3 \
+        --dataset.repo_id=helper2424/check_rtc \
+        --rtc.execution_horizon=8 \
+        --device=cuda \
+        --use_torch_compile=true \
+        --torch_compile_mode=reduce-overhead
+
+    # Enable CUDA graphs (advanced - may cause tensor aliasing errors)
+    uv run python examples/rtc/eval_dataset.py \
+        --policy.path=helper2424/smolvla_check_rtc_last3 \
+        --dataset.repo_id=helper2424/check_rtc \
+        --use_torch_compile=true \
+        --torch_compile_backend=inductor \
+        --torch_compile_mode=max-autotune \
+        --torch_compile_disable_cudagraphs=false
+"""
+
+import gc
+import logging
+import os
+import random
+from dataclasses import dataclass, field
+
+import numpy as np
+import torch
+
+try:
+    import matplotlib.pyplot as plt
+
+    MATPLOTLIB_AVAILABLE = True
+except ImportError:
+    MATPLOTLIB_AVAILABLE = False
+    plt = None
+
+from lerobot.configs import parser
+from lerobot.configs.default import DatasetConfig
+from lerobot.configs.policies import PreTrainedConfig
+from lerobot.configs.types import RTCAttentionSchedule
+from lerobot.datasets.factory import resolve_delta_timestamps
+from lerobot.datasets.lerobot_dataset import LeRobotDataset, LeRobotDatasetMetadata
+from lerobot.policies.factory import get_policy_class, make_pre_post_processors
+from lerobot.policies.rtc.configuration_rtc import RTCConfig
+from lerobot.policies.rtc.debug_visualizer import RTCDebugVisualizer
+from lerobot.utils.hub import HubMixin
+from lerobot.utils.utils import init_logging
+
+
+def set_seed(seed: int):
+    """Set random seed for reproducibility."""
+    random.seed(seed)
+    np.random.seed(seed)
+    torch.manual_seed(seed)
+    if torch.cuda.is_available():
+        torch.cuda.manual_seed(seed)
+        torch.cuda.manual_seed_all(seed)
+    if torch.backends.mps.is_available():
+        torch.mps.manual_seed(seed)
+    torch.backends.cudnn.deterministic = True
+    torch.backends.cudnn.benchmark = False
+
+
+def _check_matplotlib_available():
+    """Check if matplotlib is available, raise helpful error if not."""
+    if not MATPLOTLIB_AVAILABLE:
+        raise ImportError(
+            "matplotlib is required for RTC debug visualizations. "
+            "Please install it by running:\n"
+            "  uv pip install matplotlib"
+        )
+
+
+@dataclass
+class RTCEvalConfig(HubMixin):
+    """Configuration for RTC evaluation."""
+
+    # Policy configuration
+    policy: PreTrainedConfig | None = None
+
+    # Dataset configuration
+    dataset: DatasetConfig = field(default_factory=DatasetConfig)
+
+    # RTC configuration
+    rtc: RTCConfig = field(
+        default_factory=lambda: RTCConfig(
+            enabled=True,
+            execution_horizon=20,
+            max_guidance_weight=10.0,
+            prefix_attention_schedule=RTCAttentionSchedule.EXP,
+            debug=True,
+            debug_maxlen=1000,
+        )
+    )
+
+    # Device configuration
+    device: str | None = field(
+        default=None,
+        metadata={"help": "Device to run on (cuda, cpu, mps, auto)"},
+    )
+
+    # Output configuration
+    output_dir: str = field(
+        default="rtc_debug_output",
+        metadata={"help": "Directory to save debug visualizations"},
+    )
+
+    # Seed configuration
+    seed: int = field(
+        default=42,
+        metadata={"help": "Random seed for reproducibility"},
+    )
+
+    inference_delay: int = field(
+        default=4,
+        metadata={"help": "Inference delay for RTC"},
+    )
+
+    # Torch compile configuration
+    use_torch_compile: bool = field(
+        default=False,
+        metadata={"help": "Use torch.compile for faster inference (PyTorch 2.0+)"},
+    )
+
+    torch_compile_backend: str = field(
+        default="inductor",
+        metadata={"help": "Backend for torch.compile (inductor, aot_eager, cudagraphs)"},
+    )
+
+    torch_compile_mode: str = field(
+        default="default",
+        metadata={"help": "Compilation mode (default, reduce-overhead, max-autotune)"},
+    )
+
+    torch_compile_disable_cudagraphs: bool = field(
+        default=True,
+        metadata={
+            "help": "Disable CUDA graphs in torch.compile. Required due to in-place tensor "
+            "operations in denoising loop (x_t += dt * v_t) which cause tensor aliasing issues."
+        },
+    )
+
+    def __post_init__(self):
+        # Parse policy path
+        policy_path = parser.get_path_arg("policy")
+        if policy_path:
+            cli_overrides = parser.get_cli_overrides("policy")
+            self.policy = PreTrainedConfig.from_pretrained(policy_path, cli_overrides=cli_overrides)
+            self.policy.pretrained_path = policy_path
+        else:
+            raise ValueError("Policy path is required (--policy.path)")
+
+        # Auto-detect device if not specified
+        if self.device is None or self.device == "auto":
+            if torch.cuda.is_available():
+                self.device = "cuda"
+            elif torch.backends.mps.is_available():
+                self.device = "mps"
+            else:
+                self.device = "cpu"
+            logging.info(f"Auto-detected device: {self.device}")
+
+    @classmethod
+    def __get_path_fields__(cls) -> list[str]:
+        """This enables the parser to load config from the policy using `--policy.path=local/dir`"""
+        return ["policy"]
+
+
+class RTCEvaluator:
+    """Evaluator for RTC on dataset samples."""
+
+    def __init__(self, cfg: RTCEvalConfig):
+        self.cfg = cfg
+        self.device = cfg.device
+
+        # Load dataset with proper delta_timestamps based on policy configuration
+        # Calculate delta_timestamps using the same logic as make_dataset factory
+        logging.info(f"Loading dataset: {cfg.dataset.repo_id}")
+
+        # Get dataset metadata to extract FPS
+        ds_meta = LeRobotDatasetMetadata(cfg.dataset.repo_id)
+
+        # Calculate delta_timestamps from policy's delta_indices
+        delta_timestamps = resolve_delta_timestamps(cfg.policy, ds_meta)
+
+        # Create dataset with calculated delta_timestamps
+        self.dataset = LeRobotDataset(
+            cfg.dataset.repo_id,
+            delta_timestamps=delta_timestamps,
+        )
+        logging.info(f"Dataset loaded: {len(self.dataset)} samples, {self.dataset.num_episodes} episodes")
+
+        # Create preprocessor/postprocessor
+        self.preprocessor, self.postprocessor = make_pre_post_processors(
+            policy_cfg=cfg.policy,
+            pretrained_path=cfg.policy.pretrained_path,
+            preprocessor_overrides={
+                "device_processor": {"device": self.device},
+            },
+        )
+
+        logging.info("=" * 80)
+        logging.info("Ready to run evaluation with sequential policy loading:")
+        logging.info("  1. policy_prev_chunk - Generate reference chunk, then destroy")
+        logging.info("  2. policy_no_rtc - Generate without RTC, then destroy")
+        logging.info("  3. policy_rtc - Generate with RTC, then destroy")
+        logging.info("  Note: Only one policy in memory at a time for efficient memory usage")
+        logging.info("=" * 80)
+
+    def _init_policy(self, name: str, rtc_enabled: bool, rtc_debug: bool):
+        """Initialize a single policy instance with specified RTC configuration.
+
+        Args:
+            name: Name identifier for logging purposes
+            rtc_enabled: Whether to enable RTC for this policy
+            rtc_debug: Whether to enable debug tracking for this policy
+
+        Returns:
+            Configured policy instance with optional torch.compile applied
+        """
+        logging.info(f"Initializing {name}...")
+
+        # Load policy from pretrained
+        policy_class = get_policy_class(self.cfg.policy.type)
+
+        config = PreTrainedConfig.from_pretrained(self.cfg.policy.pretrained_path)
+
+        if self.cfg.policy.type == "pi05" or self.cfg.policy.type == "pi0":
+            config.compile_model = self.cfg.use_torch_compile
+
+        policy = policy_class.from_pretrained(self.cfg.policy.pretrained_path, config=config)
+        policy = policy.to(self.device)
+        policy.eval()
+
+        # Configure RTC
+        rtc_config = RTCConfig(
+            enabled=rtc_enabled,
+            execution_horizon=self.cfg.rtc.execution_horizon,
+            max_guidance_weight=self.cfg.rtc.max_guidance_weight,
+            prefix_attention_schedule=self.cfg.rtc.prefix_attention_schedule,
+            debug=rtc_debug,
+            debug_maxlen=self.cfg.rtc.debug_maxlen,
+        )
+        policy.config.rtc_config = rtc_config
+        policy.init_rtc_processor()
+
+        logging.info(f"  RTC enabled: {rtc_enabled}")
+        logging.info(f"  RTC debug: {rtc_debug}")
+        logging.info(f"  Policy config: {config}")
+
+        # Apply torch.compile to predict_action_chunk method if enabled
+        if self.cfg.use_torch_compile:
+            policy = self._apply_torch_compile(policy, name)
+
+        logging.info(f"✓ {name} initialized successfully")
+        return policy
+
+    def _apply_torch_compile(self, policy, policy_name: str):
+        """Apply torch.compile to the policy's predict_action_chunk method.
+
+        Args:
+            policy: Policy instance to compile
+            policy_name: Name for logging purposes
+
+        Returns:
+            Policy with compiled predict_action_chunk method
+        """
+
+        # PI models handle their own compilation
+        if policy.type == "pi05" or policy.type == "pi0":
+            return policy
+
+        try:
+            # Check if torch.compile is available (PyTorch 2.0+)
+            if not hasattr(torch, "compile"):
+                logging.warning(
+                    f"  [{policy_name}] torch.compile is not available. Requires PyTorch 2.0+. "
+                    f"Current version: {torch.__version__}. Skipping compilation."
+                )
+                return policy
+
+            logging.info(f"  [{policy_name}] Applying torch.compile to predict_action_chunk...")
+            logging.info(f"    Backend: {self.cfg.torch_compile_backend}")
+            logging.info(f"    Mode: {self.cfg.torch_compile_mode}")
+            logging.info(f"    Disable CUDA graphs: {self.cfg.torch_compile_disable_cudagraphs}")
+            logging.info("    Note: Debug tracker excluded from compilation via @torch._dynamo.disable")
+
+            # Compile the predict_action_chunk method
+            # - Debug tracker is excluded from compilation via @torch._dynamo.disable
+            # - CUDA graphs disabled to prevent tensor aliasing from in-place ops (x_t += dt * v_t)
+            compile_kwargs = {
+                "backend": self.cfg.torch_compile_backend,
+                "mode": self.cfg.torch_compile_mode,
+            }
+
+            # Disable CUDA graphs if requested (prevents tensor aliasing issues)
+            if self.cfg.torch_compile_disable_cudagraphs:
+                compile_kwargs["options"] = {"triton.cudagraphs": False}
+
+            original_method = policy.predict_action_chunk
+            compiled_method = torch.compile(original_method, **compile_kwargs)
+            policy.predict_action_chunk = compiled_method
+            logging.info(f"  ✓ [{policy_name}] Successfully compiled predict_action_chunk")
+
+        except Exception as e:
+            logging.error(f"  [{policy_name}] Failed to apply torch.compile: {e}")
+            logging.warning(f"  [{policy_name}] Continuing without torch.compile")
+
+        return policy
+
+    def _destroy_policy(self, policy, policy_name: str):
+        """Explicitly destroy a policy and free all associated memory.
+
+        This method performs aggressive cleanup to ensure maximum memory is freed,
+        which is critical for large models (e.g., VLAs with billions of parameters).
+
+        Args:
+            policy: Policy instance to destroy
+            policy_name: Name for logging purposes
+        """
+        logging.info(f"  Destroying {policy_name} and freeing memory...")
+
+        try:
+            # Step 1: Move policy to CPU to free GPU/MPS memory
+            policy.cpu()
+
+            # Step 2: Delete the policy object
+            del policy
+
+            # Step 3: Force garbage collection to reclaim memory immediately
+            gc.collect()
+
+            # Step 4: Clear device-specific caches
+            if torch.cuda.is_available():
+                torch.cuda.empty_cache()
+                torch.cuda.synchronize()  # Ensure all operations complete
+
+            if torch.backends.mps.is_available():
+                torch.mps.empty_cache()
+
+            logging.info(f"  ✓ {policy_name} destroyed and memory freed")
+
+        except Exception as e:
+            logging.warning(f"  Warning: Error during {policy_name} cleanup: {e}")
+
+    def run_evaluation(self):
+        """Run evaluation on two random dataset samples using three separate policies.
+
+        Note: Policies are deinitalized after each step to free memory. Large models
+        (e.g., VLA models with billions of parameters) cannot fit three instances in
+        memory simultaneously. By deleting and garbage collecting after each step,
+        we ensure only one policy is loaded at a time.
+        """
+        # Create output directory
+        os.makedirs(self.cfg.output_dir, exist_ok=True)
+        logging.info(f"Output directory: {self.cfg.output_dir}")
+
+        logging.info("=" * 80)
+        logging.info("Starting RTC evaluation")
+        logging.info(f"Inference delay: {self.cfg.inference_delay}")
+        logging.info("=" * 80)
+
+        # Load two random samples from dataset
+        data_loader = torch.utils.data.DataLoader(self.dataset, batch_size=1, shuffle=True)
+        loader_iter = iter(data_loader)
+        first_sample = next(loader_iter)
+        second_sample = next(loader_iter)
+
+        preprocessed_first_sample = self.preprocessor(first_sample)
+        preprocessed_second_sample = self.preprocessor(second_sample)
+
+        # ============================================================================
+        # Step 1: Generate previous chunk using policy_prev_chunk
+        # ============================================================================
+        # This policy is only used to generate the reference chunk and then freed
+        logging.info("=" * 80)
+        logging.info("Step 1: Generating previous chunk with policy_prev_chunk")
+        logging.info("=" * 80)
+
+        # Initialize policy 1
+        policy_prev_chunk_policy = self._init_policy(
+            name="policy_prev_chunk",
+            rtc_enabled=False,
+            rtc_debug=False,
+        )
+        with torch.no_grad():
+            prev_chunk_left_over = policy_prev_chunk_policy.predict_action_chunk(
+                preprocessed_first_sample,
+            )[:, :25, :].squeeze(0)
+        logging.info(f"  Generated prev_chunk shape: {prev_chunk_left_over.shape}")
+
+        # Destroy policy_prev_chunk to free memory for large models
+        self._destroy_policy(policy_prev_chunk_policy, "policy_prev_chunk")
+
+        # ============================================================================
+        # Step 2: Generate actions WITHOUT RTC using policy_no_rtc
+        # ============================================================================
+        logging.info("=" * 80)
+        logging.info("Step 2: Generating actions WITHOUT RTC with policy_no_rtc")
+        logging.info("=" * 80)
+
+        set_seed(self.cfg.seed)
+
+        # Initialize policy 2
+        policy_no_rtc_policy = self._init_policy(
+            name="policy_no_rtc",
+            rtc_enabled=False,
+            rtc_debug=True,
+        )
+
+        # Sample noise (use same noise for both RTC and non-RTC for fair comparison)
+        noise_size = (1, policy_no_rtc_policy.config.chunk_size, policy_no_rtc_policy.config.max_action_dim)
+        noise = policy_no_rtc_policy.model.sample_noise(noise_size, self.device)
+        noise_clone = noise.clone()
+        policy_no_rtc_policy.rtc_processor.reset_tracker()
+        with torch.no_grad():
+            no_rtc_actions = policy_no_rtc_policy.predict_action_chunk(
+                preprocessed_second_sample,
+                noise=noise,
+            )
+        no_rtc_tracked_steps = policy_no_rtc_policy.rtc_processor.tracker.get_all_steps()
+        logging.info(f"  Tracked {len(no_rtc_tracked_steps)} steps without RTC")
+        logging.info(f"  Generated no_rtc_actions shape: {no_rtc_actions.shape}")
+
+        # Destroy policy_no_rtc to free memory before loading policy_rtc
+        self._destroy_policy(policy_no_rtc_policy, "policy_no_rtc")
+
+        # ============================================================================
+        # Step 3: Generate actions WITH RTC using policy_rtc
+        # ============================================================================
+        logging.info("=" * 80)
+        logging.info("Step 3: Generating actions WITH RTC with policy_rtc")
+        logging.info("=" * 80)
+
+        set_seed(self.cfg.seed)
+
+        # Initialize policy 3
+        policy_rtc_policy = self._init_policy(
+            name="policy_rtc",
+            rtc_enabled=True,
+            rtc_debug=True,
+        )
+        policy_rtc_policy.rtc_processor.reset_tracker()
+        with torch.no_grad():
+            rtc_actions = policy_rtc_policy.predict_action_chunk(
+                preprocessed_second_sample,
+                noise=noise_clone,
+                inference_delay=self.cfg.inference_delay,
+                prev_chunk_left_over=prev_chunk_left_over,
+                execution_horizon=self.cfg.rtc.execution_horizon,
+            )
+        rtc_tracked_steps = policy_rtc_policy.rtc_processor.get_all_debug_steps()
+        logging.info(f"  Tracked {len(rtc_tracked_steps)} steps with RTC")
+        logging.info(f"  Generated rtc_actions shape: {rtc_actions.shape}")
+
+        # Save num_steps before destroying policy (needed for plotting)
+        try:
+            num_steps = policy_rtc_policy.config.num_steps
+        except Exception as e:
+            logging.error(f"  Error getting num_steps: {e}")
+            num_steps = policy_rtc_policy.config.num_inference_steps
+            logging.warning(f"  Using num_inference_steps: {num_steps} instead of num_steps")
+
+        # Destroy policy_rtc after final use
+        self._destroy_policy(policy_rtc_policy, "policy_rtc")
+
+        # Plot and save results
+        logging.info("=" * 80)
+        logging.info("Plotting results...")
+        self.plot_tracked_data(rtc_tracked_steps, no_rtc_tracked_steps, prev_chunk_left_over, num_steps)
+
+        # Plot final actions comparison
+        logging.info("=" * 80)
+        logging.info("Plotting final actions comparison...")
+        self.plot_final_actions_comparison(rtc_actions, no_rtc_actions, prev_chunk_left_over)
+
+        logging.info("=" * 80)
+        logging.info("Evaluation completed successfully")
+
+    def plot_final_actions_comparison(self, rtc_actions, no_rtc_actions, prev_chunk_left_over):
+        """Plot final action predictions comparison on a single chart.
+
+        Args:
+            rtc_actions: Final actions from RTC policy
+            no_rtc_actions: Final actions from non-RTC policy
+            prev_chunk_left_over: Previous chunk used as ground truth
+        """
+        _check_matplotlib_available()
+
+        # Remove batch dimension if present
+        rtc_actions_plot = rtc_actions.squeeze(0).cpu() if len(rtc_actions.shape) == 3 else rtc_actions.cpu()
+        no_rtc_actions_plot = (
+            no_rtc_actions.squeeze(0).cpu() if len(no_rtc_actions.shape) == 3 else no_rtc_actions.cpu()
+        )
+        prev_chunk_plot = prev_chunk_left_over.cpu()
+
+        # Create figure with 6 subplots (one per action dimension)
+        fig, axes = plt.subplots(6, 1, figsize=(16, 12))
+        fig.suptitle("Final Action Predictions Comparison (Raw)", fontsize=16)
+
+        # Plot each action dimension
+        for dim_idx, ax in enumerate(axes):
+            # Plot previous chunk (ground truth) in red
+            RTCDebugVisualizer.plot_waypoints(
+                [ax],
+                prev_chunk_plot[:, dim_idx : dim_idx + 1],
+                start_from=0,
+                color="red",
+                label="Previous Chunk (Ground Truth)",
+                linewidth=2.5,
+                alpha=0.8,
+            )
+
+            # Plot no-RTC actions in blue
+            RTCDebugVisualizer.plot_waypoints(
+                [ax],
+                no_rtc_actions_plot[:, dim_idx : dim_idx + 1],
+                start_from=0,
+                color="blue",
+                label="No RTC",
+                linewidth=2,
+                alpha=0.7,
+            )
+
+            # Plot RTC actions in green
+            RTCDebugVisualizer.plot_waypoints(
+                [ax],
+                rtc_actions_plot[:, dim_idx : dim_idx + 1],
+                start_from=0,
+                color="green",
+                label="RTC",
+                linewidth=2,
+                alpha=0.7,
+            )
+
+            # Add vertical lines for inference delay and execution horizon
+            inference_delay = self.cfg.inference_delay
+            execution_horizon = self.cfg.rtc.execution_horizon
+
+            if inference_delay > 0:
+                ax.axvline(
+                    x=inference_delay - 1,
+                    color="orange",
+                    linestyle="--",
+                    alpha=0.5,
+                    label=f"Inference Delay ({inference_delay})",
+                )
+
+            if execution_horizon > 0:
+                ax.axvline(
+                    x=execution_horizon,
+                    color="purple",
+                    linestyle="--",
+                    alpha=0.5,
+                    label=f"Execution Horizon ({execution_horizon})",
+                )
+
+            ax.set_ylabel(f"Dim {dim_idx}", fontsize=10)
+            ax.grid(True, alpha=0.3)
+
+            # Set x-axis ticks to show all integer values
+            max_len = max(rtc_actions_plot.shape[0], no_rtc_actions_plot.shape[0], prev_chunk_plot.shape[0])
+            ax.set_xticks(range(0, max_len, max(1, max_len // 20)))  # Show ~20 ticks
+            ax.set_xlim(-0.5, max_len - 0.5)
+
+        axes[-1].set_xlabel("Step", fontsize=10)
+
+        # Collect legend handles and labels from first subplot
+        handles, labels = axes[0].get_legend_handles_labels()
+        # Remove duplicates while preserving order
+        seen = set()
+        unique_handles = []
+        unique_labels = []
+        for handle, label in zip(handles, labels, strict=True):
+            if label not in seen:
+                seen.add(label)
+                unique_handles.append(handle)
+                unique_labels.append(label)
+
+        # Add legend outside the plot area (to the right)
+        fig.legend(
+            unique_handles,
+            unique_labels,
+            loc="center right",
+            fontsize=9,
+            bbox_to_anchor=(1.0, 0.5),
+            framealpha=0.9,
+        )
+
+        # Save figure
+        output_path = os.path.join(self.cfg.output_dir, "final_actions_comparison.png")
+        fig.tight_layout(rect=[0, 0, 0.85, 1])  # Leave space for legend on right
+        fig.savefig(output_path, dpi=150, bbox_inches="tight")
+        logging.info(f"Saved final actions comparison to {output_path}")
+        plt.close(fig)
+
+    def plot_tracked_data(self, rtc_tracked_steps, no_rtc_tracked_steps, prev_chunk_left_over, num_steps):
+        _check_matplotlib_available()
+
+        # Create side-by-side figures for denoising visualization
+        fig_xt, axs_xt = self._create_figure("x_t Denoising: No RTC (left) vs RTC (right)")
+        fig_vt, axs_vt = self._create_figure("v_t Denoising: No RTC (left) vs RTC (right)")
+        fig_corr, axs_corr = self._create_figure("Correction: No RTC (left) vs RTC (right)")
+        fig_x1t, axs_x1t = self._create_figure(
+            "x1_t Predicted State & Error: No RTC (left - empty) vs RTC (right)"
+        )
+        self._plot_denoising_steps_from_tracker(
+            rtc_tracked_steps,
+            axs_xt[:, 1],  # Right column for x_t
+            axs_vt[:, 1],  # Right column for v_t
+            axs_corr[:, 1],  # Right column for correction
+            axs_x1t[:, 1],  # Right column for x1_t
+            num_steps,
+            add_labels=True,  # Add labels for RTC (right column)
+        )
+
+        self._plot_denoising_steps_from_tracker(
+            no_rtc_tracked_steps,
+            axs_xt[:, 0],  # Left column for x_t
+            axs_vt[:, 0],  # Left column for v_t
+            axs_corr[:, 0],  # Left column for correction
+            axs_x1t[:, 0],  # Left column for x1_t
+            num_steps,
+            add_labels=False,  # No labels for No RTC (left column)
+        )
+
+        # Plot no-RTC x_t data on right chart as orange dashed line for comparison
+        self._plot_no_rtc_xt_reference(no_rtc_tracked_steps, axs_xt[:, 1], num_steps)
+
+        # Plot ground truth on x_t axes
+        RTCDebugVisualizer.plot_waypoints(
+            axs_xt[:, 1], prev_chunk_left_over, start_from=0, color="red", label="Ground truth"
+        )
+
+        # Plot ground truth on x1_t axes
+        RTCDebugVisualizer.plot_waypoints(
+            axs_x1t[:, 1], prev_chunk_left_over, start_from=0, color="red", label="Ground truth"
+        )
+
+        # Plot ground truth on x_t axes (no labels for left column)
+        RTCDebugVisualizer.plot_waypoints(
+            axs_xt[:, 0], prev_chunk_left_over, start_from=0, color="red", label=None
+        )
+
+        RTCDebugVisualizer.plot_waypoints(
+            axs_x1t[:, 0], prev_chunk_left_over, start_from=0, color="red", label=None
+        )
+
+        # Add legends outside the plot area for each figure
+        self._add_figure_legend(fig_xt, axs_xt)
+        self._add_figure_legend(fig_vt, axs_vt)
+        self._add_figure_legend(fig_corr, axs_corr)
+        self._add_figure_legend(fig_x1t, axs_x1t)
+
+        # Save denoising plots
+        self._save_figure(fig_xt, os.path.join(self.cfg.output_dir, "denoising_xt_comparison.png"))
+        self._save_figure(fig_vt, os.path.join(self.cfg.output_dir, "denoising_vt_comparison.png"))
+        self._save_figure(fig_corr, os.path.join(self.cfg.output_dir, "denoising_correction_comparison.png"))
+        self._save_figure(fig_x1t, os.path.join(self.cfg.output_dir, "denoising_x1t_comparison.png"))
+
+    def _create_figure(self, title):
+        fig, axs = plt.subplots(6, 2, figsize=(24, 12))
+        fig.suptitle(title, fontsize=16)
+
+        for ax in axs[:, 0]:
+            ax.set_title("No RTC (N/A)" if ax == axs[0, 0] else "", fontsize=12)
+        for ax in axs[:, 1]:
+            ax.set_title("RTC" if ax == axs[0, 1] else "", fontsize=12)
+
+        return fig, axs
+
+    def _add_figure_legend(self, fig, axs):
+        """Add a legend outside the plot area on the right side.
+
+        Args:
+            fig: Matplotlib figure to add legend to
+            axs: Array of axes to collect legend handles from
+        """
+        # Collect all handles and labels from the first row of axes (right column)
+        handles, labels = axs[0, 1].get_legend_handles_labels()
+
+        # Remove duplicates while preserving order
+        seen = set()
+        unique_handles = []
+        unique_labels = []
+        for handle, label in zip(handles, labels, strict=True):
+            if label not in seen:
+                seen.add(label)
+                unique_handles.append(handle)
+                unique_labels.append(label)
+
+        # Add legend outside the plot area (to the right, close to charts)
+        if unique_handles:
+            fig.legend(
+                unique_handles,
+                unique_labels,
+                loc="center left",
+                fontsize=8,
+                bbox_to_anchor=(0.87, 0.5),
+                framealpha=0.9,
+                ncol=1,
+            )
+
+    def _save_figure(self, fig, path):
+        fig.tight_layout(rect=[0, 0, 0.85, 1])  # Leave space for legend/colorbar on right
+        fig.savefig(path, dpi=150, bbox_inches="tight")
+        logging.info(f"Saved figure to {path}")
+        plt.close(fig)
+
+    def _plot_denoising_steps_from_tracker(
+        self, tracked_steps, xt_axs, vt_axs, corr_axs, x1t_axs, num_steps, add_labels=True
+    ):
+        """Plot denoising steps from tracker data.
+
+        Args:
+            tracked_steps: List of DebugStep objects containing debug steps
+            xt_axs: Matplotlib axes for x_t plots (array of 6 axes)
+            vt_axs: Matplotlib axes for v_t plots (array of 6 axes)
+            corr_axs: Matplotlib axes for correction plots (array of 6 axes)
+            x1t_axs: Matplotlib axes for x1_t plots (array of 6 axes)
+            num_steps: Total number of denoising steps for colormap
+            add_labels: Whether to add legend labels for the plots
+        """
+
+        logging.info("=" * 80)
+        logging.info(f"Plotting {len(tracked_steps)} steps")
+
+        debug_steps = tracked_steps
+        if not debug_steps:
+            return
+
+        # Define colors for different denoise steps (using a colormap)
+        colors = plt.cm.viridis(np.linspace(0, 1, num_steps))
+
+        for step_idx, debug_step in enumerate(debug_steps):
+            color = colors[step_idx % len(colors)]
+            label = f"Step {step_idx}" if add_labels else None
+
+            # Plot x_t
+            if debug_step.x_t is not None:
+                RTCDebugVisualizer.plot_waypoints(
+                    xt_axs, debug_step.x_t, start_from=0, color=color, label=label
+                )
+
+            # Plot v_t
+            if debug_step.v_t is not None:
+                RTCDebugVisualizer.plot_waypoints(
+                    vt_axs, debug_step.v_t, start_from=0, color=color, label=label
+                )
+
+            # Plot correction on separate axes
+            if debug_step.correction is not None:
+                RTCDebugVisualizer.plot_waypoints(
+                    corr_axs,
+                    debug_step.correction,
+                    start_from=0,
+                    color=color,
+                    label=label,
+                )
+
+            # Plot x1_t (predicted state)
+            if x1t_axs is not None and debug_step.x1_t is not None:
+                x1t_label = f"x1_t Step {step_idx}" if add_labels else None
+                RTCDebugVisualizer.plot_waypoints(
+                    x1t_axs,
+                    debug_step.x1_t,
+                    start_from=0,
+                    color=color,
+                    label=x1t_label,
+                )
+
+            # Plot error in orange dashed
+            if x1t_axs is not None and debug_step.err is not None:
+                error_chunk = (
+                    debug_step.err[0].cpu().numpy()
+                    if len(debug_step.err.shape) == 3
+                    else debug_step.err.cpu().numpy()
+                )
+
+                num_dims = min(error_chunk.shape[-1], 6)
+                error_label = f"error Step {step_idx}" if add_labels else None
+                for j in range(num_dims):
+                    x1t_axs[j].plot(
+                        np.arange(0, error_chunk.shape[0]),
+                        error_chunk[:, j],
+                        color="orange",
+                        linestyle="--",
+                        alpha=0.7,
+                        label=error_label,
+                    )
+
+        # Recalculate axis limits after plotting to ensure proper scaling
+        self._rescale_axes(xt_axs)
+        self._rescale_axes(vt_axs)
+        self._rescale_axes(corr_axs)
+        self._rescale_axes(x1t_axs)
+
+    def _plot_no_rtc_xt_reference(self, no_rtc_tracked_steps, xt_axs, num_steps):
+        """Plot final no-RTC x_t data as orange dashed line on the RTC chart for comparison.
+
+        Args:
+            no_rtc_tracked_steps: List of DebugStep objects containing no-RTC debug steps
+            xt_axs: Matplotlib axes for x_t plots (array of 6 axes, right column)
+            num_steps: Total number of denoising steps for colormap
+        """
+        debug_steps = no_rtc_tracked_steps
+        if not debug_steps:
+            return
+
+        # Plot only the final x_t step as orange dashed line
+        final_step = debug_steps[-1]
+        logging.info("Plotting final no-RTC x_t step as orange dashed reference")
+
+        if final_step.x_t is not None:
+            x_t_chunk = (
+                final_step.x_t[0].cpu().numpy()
+                if len(final_step.x_t.shape) == 3
+                else final_step.x_t.cpu().numpy()
+            )
+
+            num_dims = min(x_t_chunk.shape[-1], 6)
+            for j in range(num_dims):
+                xt_axs[j].plot(
+                    np.arange(0, x_t_chunk.shape[0]),
+                    x_t_chunk[:, j],
+                    color="orange",
+                    linestyle="--",
+                    alpha=0.7,
+                    linewidth=2,
+                    label="No RTC (final)" if j == 0 else "",
+                )
+
+    def _rescale_axes(self, axes):
+        """Rescale axes to show all data with proper margins.
+
+        Args:
+            axes: Array of matplotlib axes to rescale
+        """
+        for ax in axes:
+            ax.relim()
+            ax.autoscale_view()
+
+            # Add 10% margin to y-axis for better visualization
+            ylim = ax.get_ylim()
+            y_range = ylim[1] - ylim[0]
+            if y_range > 0:  # Avoid division by zero
+                margin = y_range * 0.1
+                ax.set_ylim(ylim[0] - margin, ylim[1] + margin)
+
+            # Set x-axis ticks to show all integer values
+            xlim = ax.get_xlim()
+            max_len = int(xlim[1]) + 1
+            if max_len > 0:
+                ax.set_xticks(range(0, max_len, max(1, max_len // 20)))  # Show ~20 ticks
+                ax.set_xlim(-0.5, max_len - 0.5)
+
+
+@parser.wrap()
+def main(cfg: RTCEvalConfig):
+    """Main entry point for RTC evaluation."""
+    # Set random seed for reproducibility
+    set_seed(cfg.seed)
+
+    init_logging()
+
+    logging.info("=" * 80)
+    logging.info("RTC Dataset Evaluation")
+    logging.info(f"Config: {cfg}")
+    logging.info("=" * 80)
+
+    evaluator = RTCEvaluator(cfg)
+    evaluator.run_evaluation()
+
+
+if __name__ == "__main__":
+    main()

examples/rtc/eval_with_real_robot.py

@@ -0,0 +1,549 @@
+#!/usr/bin/env python
+
+# Copyright 2025 The HuggingFace Inc. team. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Demo script showing how to use Real-Time Chunking (RTC) with action chunking policies on real robots.
+
+This script demonstrates:
+1. Creating a robot and policy (SmolVLA, Pi0, etc.) with RTC
+2. Consuming actions from the policy while the robot executes
+3. Periodically requesting new action chunks in the background using threads
+4. Managing action buffers and timing for real-time operation
+
+For simulation environments, see eval_with_simulation.py
+
+Usage:
+    # Run RTC with Real robot with RTC
+    uv run examples/rtc/eval_with_real_robot.py \
+        --policy.path=helper2424/smolvla_check_rtc_last3 \
+        --policy.device=mps \
+        --rtc.enabled=true \
+        --rtc.execution_horizon=20 \
+        --robot.type=so100_follower \
+        --robot.port=/dev/tty.usbmodem58FA0834591 \
+        --robot.id=so100_follower \
+        --robot.cameras="{ gripper: {type: opencv, index_or_path: 1, width: 640, height: 480, fps: 30}, front: {type: opencv, index_or_path: 0, width: 640, height: 480, fps: 30}}" \
+        --task="Move green small object into the purple platform" \
+        --duration=120
+
+    # Run RTC with Real robot without RTC
+    uv run examples/rtc/eval_with_real_robot.py \
+        --policy.path=helper2424/smolvla_check_rtc_last3 \
+        --policy.device=mps \
+        --rtc.enabled=false \
+        --robot.type=so100_follower \
+        --robot.port=/dev/tty.usbmodem58FA0834591 \
+        --robot.id=so100_follower \
+        --robot.cameras="{ gripper: {type: opencv, index_or_path: 1, width: 640, height: 480, fps: 30}, front: {type: opencv, index_or_path: 0, width: 640, height: 480, fps: 30}}" \
+        --task="Move green small object into the purple platform" \
+        --duration=120
+
+    # Run RTC with Real robot with pi0.5 policy
+    uv run examples/rtc/eval_with_real_robot.py \
+        --policy.path=helper2424/pi05_check_rtc \
+        --policy.device=mps \
+        --rtc.enabled=true \
+        --rtc.execution_horizon=20 \
+        --robot.type=so100_follower \
+        --robot.port=/dev/tty.usbmodem58FA0834591 \
+        --robot.id=so100_follower \
+        --robot.cameras="{ gripper: {type: opencv, index_or_path: 0, width: 640, height: 480, fps: 30}, front: {type: opencv, index_or_path: 1, width: 640, height: 480, fps: 30}}" \
+        --task="Move green small object into the purple platform" \
+        --duration=120
+"""
+
+import logging
+import math
+import sys
+import time
+import traceback
+from dataclasses import dataclass, field
+from threading import Event, Lock, Thread
+
+import torch
+from torch import Tensor
+
+from lerobot.cameras.opencv.configuration_opencv import OpenCVCameraConfig  # noqa: F401
+from lerobot.cameras.realsense.configuration_realsense import RealSenseCameraConfig  # noqa: F401
+from lerobot.configs import parser
+from lerobot.configs.policies import PreTrainedConfig
+from lerobot.configs.types import RTCAttentionSchedule
+from lerobot.datasets.utils import build_dataset_frame, hw_to_dataset_features
+from lerobot.policies.factory import get_policy_class, make_pre_post_processors
+from lerobot.policies.rtc.action_queue import ActionQueue
+from lerobot.policies.rtc.configuration_rtc import RTCConfig
+from lerobot.policies.rtc.latency_tracker import LatencyTracker
+from lerobot.processor.factory import (
+    make_default_robot_action_processor,
+    make_default_robot_observation_processor,
+)
+from lerobot.rl.process import ProcessSignalHandler
+from lerobot.robots import (  # noqa: F401
+    Robot,
+    RobotConfig,
+    koch_follower,
+    so100_follower,
+    so101_follower,
+)
+from lerobot.robots.utils import make_robot_from_config
+from lerobot.utils.constants import OBS_IMAGES
+from lerobot.utils.hub import HubMixin
+from lerobot.utils.utils import init_logging
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class RobotWrapper:
+    def __init__(self, robot: Robot):
+        self.robot = robot
+        self.lock = Lock()
+
+    def get_observation(self) -> dict[str, Tensor]:
+        with self.lock:
+            return self.robot.get_observation()
+
+    def send_action(self, action: Tensor):
+        with self.lock:
+            self.robot.send_action(action)
+
+    def observation_features(self) -> list[str]:
+        with self.lock:
+            return self.robot.observation_features
+
+    def action_features(self) -> list[str]:
+        with self.lock:
+            return self.robot.action_features
+
+
+@dataclass
+class RTCDemoConfig(HubMixin):
+    """Configuration for RTC demo with action chunking policies and real robots."""
+
+    # Policy configuration
+    policy: PreTrainedConfig | None = None
+
+    # Robot configuration
+    robot: RobotConfig | None = None
+
+    # RTC configuration
+    rtc: RTCConfig = field(
+        default_factory=lambda: RTCConfig(
+            execution_horizon=10,
+            max_guidance_weight=1.0,
+            prefix_attention_schedule=RTCAttentionSchedule.EXP,
+        )
+    )
+
+    # Demo parameters
+    duration: float = 30.0  # Duration to run the demo (seconds)
+    fps: float = 10.0  # Action execution frequency (Hz)
+
+    # Compute device
+    device: str | None = None  # Device to run on (cuda, cpu, auto)
+
+    # Get new actions horizon. The amount of executed steps after which will be requested new actions.
+    # It should be higher than inference delay + execution horizon.
+    action_queue_size_to_get_new_actions: int = 30
+
+    # Task to execute
+    task: str = field(default="", metadata={"help": "Task to execute"})
+
+    # Torch compile configuration
+    use_torch_compile: bool = field(
+        default=False,
+        metadata={"help": "Use torch.compile for faster inference (PyTorch 2.0+)"},
+    )
+
+    torch_compile_backend: str = field(
+        default="inductor",
+        metadata={"help": "Backend for torch.compile (inductor, aot_eager, cudagraphs)"},
+    )
+
+    torch_compile_mode: str = field(
+        default="default",
+        metadata={"help": "Compilation mode (default, reduce-overhead, max-autotune)"},
+    )
+
+    torch_compile_disable_cudagraphs: bool = field(
+        default=True,
+        metadata={
+            "help": "Disable CUDA graphs in torch.compile. Required due to in-place tensor "
+            "operations in denoising loop (x_t += dt * v_t) which cause tensor aliasing issues."
+        },
+    )
+
+    def __post_init__(self):
+        # HACK: We parse again the cli args here to get the pretrained path if there was one.
+        policy_path = parser.get_path_arg("policy")
+        if policy_path:
+            cli_overrides = parser.get_cli_overrides("policy")
+            self.policy = PreTrainedConfig.from_pretrained(policy_path, cli_overrides=cli_overrides)
+            self.policy.pretrained_path = policy_path
+        else:
+            raise ValueError("Policy path is required")
+
+        # Validate that robot configuration is provided
+        if self.robot is None:
+            raise ValueError("Robot configuration must be provided")
+
+    @classmethod
+    def __get_path_fields__(cls) -> list[str]:
+        """This enables the parser to load config from the policy using `--policy.path=local/dir`"""
+        return ["policy"]
+
+
+def is_image_key(k: str) -> bool:
+    return k.startswith(OBS_IMAGES)
+
+
+def get_actions(
+    policy,
+    robot: RobotWrapper,
+    robot_observation_processor,
+    action_queue: ActionQueue,
+    shutdown_event: Event,
+    cfg: RTCDemoConfig,
+):
+    """Thread function to request action chunks from the policy.
+
+    Args:
+        policy: The policy instance (SmolVLA, Pi0, etc.)
+        robot: The robot instance for getting observations
+        robot_observation_processor: Processor for raw robot observations
+        action_queue: Queue to put new action chunks
+        shutdown_event: Event to signal shutdown
+        cfg: Demo configuration
+    """
+    try:
+        logger.info("[GET_ACTIONS] Starting get actions thread")
+
+        latency_tracker = LatencyTracker()  # Track latency of action chunks
+        fps = cfg.fps
+        time_per_chunk = 1.0 / fps
+
+        dataset_features = hw_to_dataset_features(robot.observation_features(), "observation")
+        policy_device = policy.config.device
+
+        # Load preprocessor and postprocessor from pretrained files
+        # The stats are embedded in the processor .safetensors files
+        logger.info(f"[GET_ACTIONS] Loading preprocessor/postprocessor from {cfg.policy.pretrained_path}")
+
+        preprocessor, postprocessor = make_pre_post_processors(
+            policy_cfg=cfg.policy,
+            pretrained_path=cfg.policy.pretrained_path,
+            dataset_stats=None,  # Will load from pretrained processor files
+            preprocessor_overrides={
+                "device_processor": {"device": cfg.policy.device},
+            },
+        )
+
+        logger.info("[GET_ACTIONS] Preprocessor/postprocessor loaded successfully with embedded stats")
+
+        get_actions_threshold = cfg.action_queue_size_to_get_new_actions
+
+        if not cfg.rtc.enabled:
+            get_actions_threshold = 0
+
+        while not shutdown_event.is_set():
+            if action_queue.qsize() <= get_actions_threshold:
+                current_time = time.perf_counter()
+                action_index_before_inference = action_queue.get_action_index()
+                prev_actions = action_queue.get_left_over()
+
+                inference_latency = latency_tracker.max()
+                inference_delay = math.ceil(inference_latency / time_per_chunk)
+
+                obs = robot.get_observation()
+
+                # Apply robot observation processor
+                obs_processed = robot_observation_processor(obs)
+
+                obs_with_policy_features = build_dataset_frame(
+                    dataset_features, obs_processed, prefix="observation"
+                )
+
+                for name in obs_with_policy_features:
+                    obs_with_policy_features[name] = torch.from_numpy(obs_with_policy_features[name])
+                    if "image" in name:
+                        obs_with_policy_features[name] = (
+                            obs_with_policy_features[name].type(torch.float32) / 255
+                        )
+                        obs_with_policy_features[name] = (
+                            obs_with_policy_features[name].permute(2, 0, 1).contiguous()
+                        )
+                    obs_with_policy_features[name] = obs_with_policy_features[name].unsqueeze(0)
+                    obs_with_policy_features[name] = obs_with_policy_features[name].to(policy_device)
+
+                obs_with_policy_features["task"] = [cfg.task]  # Task should be a list, not a string!
+                obs_with_policy_features["robot_type"] = (
+                    robot.robot.name if hasattr(robot.robot, "name") else ""
+                )
+
+                preproceseded_obs = preprocessor(obs_with_policy_features)
+
+                # Generate actions WITH RTC
+                actions = policy.predict_action_chunk(
+                    preproceseded_obs,
+                    inference_delay=inference_delay,
+                    prev_chunk_left_over=prev_actions,
+                )
+
+                # Store original actions (before postprocessing) for RTC
+                original_actions = actions.squeeze(0).clone()
+
+                postprocessed_actions = postprocessor(actions)
+
+                postprocessed_actions = postprocessed_actions.squeeze(0)
+
+                new_latency = time.perf_counter() - current_time
+                new_delay = math.ceil(new_latency / time_per_chunk)
+                latency_tracker.add(new_latency)
+
+                if cfg.action_queue_size_to_get_new_actions < cfg.rtc.execution_horizon + new_delay:
+                    logger.warning(
+                        "[GET_ACTIONS] cfg.action_queue_size_to_get_new_actions Too small, It should be higher than inference delay + execution horizon."
+                    )
+
+                action_queue.merge(
+                    original_actions, postprocessed_actions, new_delay, action_index_before_inference
+                )
+            else:
+                # Small sleep to prevent busy waiting
+                time.sleep(0.1)
+
+        logger.info("[GET_ACTIONS] get actions thread shutting down")
+    except Exception as e:
+        logger.error(f"[GET_ACTIONS] Fatal exception in get_actions thread: {e}")
+        logger.error(traceback.format_exc())
+        sys.exit(1)
+
+
+def actor_control(
+    robot: RobotWrapper,
+    robot_action_processor,
+    action_queue: ActionQueue,
+    shutdown_event: Event,
+    cfg: RTCDemoConfig,
+):
+    """Thread function to execute actions on the robot.
+
+    Args:
+        robot: The robot instance
+        action_queue: Queue to get actions from
+        shutdown_event: Event to signal shutdown
+        cfg: Demo configuration
+    """
+    try:
+        logger.info("[ACTOR] Starting actor thread")
+
+        action_count = 0
+        action_interval = 1.0 / cfg.fps
+
+        while not shutdown_event.is_set():
+            start_time = time.perf_counter()
+
+            # Try to get an action from the queue with timeout
+            action = action_queue.get()
+
+            if action is not None:
+                action = action.cpu()
+                action_dict = {key: action[i].item() for i, key in enumerate(robot.action_features())}
+                action_processed = robot_action_processor((action_dict, None))
+                robot.send_action(action_processed)
+
+                action_count += 1
+
+            dt_s = time.perf_counter() - start_time
+            time.sleep(max(0, (action_interval - dt_s) - 0.001))
+
+        logger.info(f"[ACTOR] Actor thread shutting down. Total actions executed: {action_count}")
+    except Exception as e:
+        logger.error(f"[ACTOR] Fatal exception in actor_control thread: {e}")
+        logger.error(traceback.format_exc())
+        sys.exit(1)
+
+
+def _apply_torch_compile(policy, cfg: RTCDemoConfig):
+    """Apply torch.compile to the policy's predict_action_chunk method.
+
+    Args:
+        policy: Policy instance to compile
+        cfg: Configuration containing torch compile settings
+
+    Returns:
+        Policy with compiled predict_action_chunk method
+    """
+
+    # PI models handle their own compilation
+    if policy.type == "pi05" or policy.type == "pi0":
+        return policy
+
+    try:
+        # Check if torch.compile is available (PyTorch 2.0+)
+        if not hasattr(torch, "compile"):
+            logger.warning(
+                f"torch.compile is not available. Requires PyTorch 2.0+. "
+                f"Current version: {torch.__version__}. Skipping compilation."
+            )
+            return policy
+
+        logger.info("Applying torch.compile to predict_action_chunk...")
+        logger.info(f"  Backend: {cfg.torch_compile_backend}")
+        logger.info(f"  Mode: {cfg.torch_compile_mode}")
+        logger.info(f"  Disable CUDA graphs: {cfg.torch_compile_disable_cudagraphs}")
+
+        # Compile the predict_action_chunk method
+        # - CUDA graphs disabled to prevent tensor aliasing from in-place ops (x_t += dt * v_t)
+        compile_kwargs = {
+            "backend": cfg.torch_compile_backend,
+            "mode": cfg.torch_compile_mode,
+        }
+
+        # Disable CUDA graphs if requested (prevents tensor aliasing issues)
+        if cfg.torch_compile_disable_cudagraphs:
+            compile_kwargs["options"] = {"triton.cudagraphs": False}
+
+        original_method = policy.predict_action_chunk
+        compiled_method = torch.compile(original_method, **compile_kwargs)
+        policy.predict_action_chunk = compiled_method
+        logger.info("✓ Successfully compiled predict_action_chunk")
+
+    except Exception as e:
+        logger.error(f"Failed to apply torch.compile: {e}")
+        logger.warning("Continuing without torch.compile")
+
+    return policy
+
+
+@parser.wrap()
+def demo_cli(cfg: RTCDemoConfig):
+    """Main entry point for RTC demo with draccus configuration."""
+
+    # Initialize logging
+    init_logging()
+
+    logger.info(f"Using device: {cfg.device}")
+
+    # Setup signal handler for graceful shutdown
+    signal_handler = ProcessSignalHandler(use_threads=True, display_pid=False)
+    shutdown_event = signal_handler.shutdown_event
+
+    policy = None
+    robot = None
+    get_actions_thread = None
+    actor_thread = None
+
+    policy_class = get_policy_class(cfg.policy.type)
+
+    # Load config and set compile_model for pi0/pi05 models
+    config = PreTrainedConfig.from_pretrained(cfg.policy.pretrained_path)
+
+    if cfg.policy.type == "pi05" or cfg.policy.type == "pi0":
+        config.compile_model = cfg.use_torch_compile
+
+    policy = policy_class.from_pretrained(cfg.policy.pretrained_path, config=config)
+
+    # Turn on RTC
+    policy.config.rtc_config = cfg.rtc
+
+    # Init RTC processort, as by default if RTC disabled in the config
+    # The processor won't be created
+    policy.init_rtc_processor()
+
+    assert policy.name in ["smolvla", "pi05", "pi0"], "Only smolvla, pi05, and pi0 are supported for RTC"
+
+    policy = policy.to(cfg.device)
+    policy.eval()
+
+    # Apply torch.compile to predict_action_chunk method if enabled
+    if cfg.use_torch_compile:
+        policy = _apply_torch_compile(policy, cfg)
+
+    # Create robot
+    logger.info(f"Initializing robot: {cfg.robot.type}")
+    robot = make_robot_from_config(cfg.robot)
+    robot.connect()
+    robot_wrapper = RobotWrapper(robot)
+
+    # Create robot observation processor
+    robot_observation_processor = make_default_robot_observation_processor()
+    robot_action_processor = make_default_robot_action_processor()
+
+    # Create action queue for communication between threads
+    action_queue = ActionQueue(cfg.rtc)
+
+    # Start chunk requester thread
+    get_actions_thread = Thread(
+        target=get_actions,
+        args=(policy, robot_wrapper, robot_observation_processor, action_queue, shutdown_event, cfg),
+        daemon=True,
+        name="GetActions",
+    )
+    get_actions_thread.start()
+    logger.info("Started get actions thread")
+
+    # Start action executor thread
+    actor_thread = Thread(
+        target=actor_control,
+        args=(robot_wrapper, robot_action_processor, action_queue, shutdown_event, cfg),
+        daemon=True,
+        name="Actor",
+    )
+    actor_thread.start()
+    logger.info("Started actor thread")
+
+    logger.info("Started stop by duration thread")
+
+    # Main thread monitors for duration or shutdown
+    logger.info(f"Running demo for {cfg.duration} seconds...")
+    start_time = time.time()
+
+    while not shutdown_event.is_set() and (time.time() - start_time) < cfg.duration:
+        time.sleep(10)
+
+        # Log queue status periodically
+        if int(time.time() - start_time) % 5 == 0:
+            logger.info(f"[MAIN] Action queue size: {action_queue.qsize()}")
+
+        if time.time() - start_time > cfg.duration:
+            break
+
+    logger.info("Demo duration reached or shutdown requested")
+
+    # Signal shutdown
+    shutdown_event.set()
+
+    # Wait for threads to finish
+    if get_actions_thread and get_actions_thread.is_alive():
+        logger.info("Waiting for chunk requester thread to finish...")
+        get_actions_thread.join()
+
+    if actor_thread and actor_thread.is_alive():
+        logger.info("Waiting for action executor thread to finish...")
+        actor_thread.join()
+
+    # Cleanup robot
+    if robot:
+        robot.disconnect()
+        logger.info("Robot disconnected")
+
+    logger.info("Cleanup completed")
+
+
+if __name__ == "__main__":
+    demo_cli()
+    logging.info("RTC demo finished")

examples/tutorial/act/act_training_example.py

@@ -0,0 +1,98 @@
+"""This script demonstrates how to train ACT Policy on a real-world dataset."""
+
+from pathlib import Path
+
+import torch
+
+from lerobot.configs.types import FeatureType
+from lerobot.datasets.lerobot_dataset import LeRobotDataset, LeRobotDatasetMetadata
+from lerobot.datasets.utils import dataset_to_policy_features
+from lerobot.policies.act.configuration_act import ACTConfig
+from lerobot.policies.act.modeling_act import ACTPolicy
+from lerobot.policies.factory import make_pre_post_processors
+
+
+def make_delta_timestamps(delta_indices: list[int] | None, fps: int) -> list[float]:
+    if delta_indices is None:
+        return [0]
+
+    return [i / fps for i in delta_indices]
+
+
+output_directory = Path("outputs/robot_learning_tutorial/act")
+output_directory.mkdir(parents=True, exist_ok=True)
+
+# Select your device
+device = torch.device("mps")  # or "cuda" or "cpu"
+
+dataset_id = "lerobot/svla_so101_pickplace"
+
+# This specifies the inputs the model will be expecting and the outputs it will produce
+dataset_metadata = LeRobotDatasetMetadata(dataset_id)
+features = dataset_to_policy_features(dataset_metadata.features)
+
+output_features = {key: ft for key, ft in features.items() if ft.type is FeatureType.ACTION}
+input_features = {key: ft for key, ft in features.items() if key not in output_features}
+
+cfg = ACTConfig(input_features=input_features, output_features=output_features)
+policy = ACTPolicy(cfg)
+preprocessor, postprocessor = make_pre_post_processors(cfg, dataset_stats=dataset_metadata.stats)
+
+policy.train()
+policy.to(device)
+
+# To perform action chunking, ACT expects a given number of actions as targets
+delta_timestamps = {
+    "action": make_delta_timestamps(cfg.action_delta_indices, dataset_metadata.fps),
+}
+
+# add image features if they are present
+delta_timestamps |= {
+    k: make_delta_timestamps(cfg.observation_delta_indices, dataset_metadata.fps) for k in cfg.image_features
+}
+
+# Instantiate the dataset
+dataset = LeRobotDataset(dataset_id, delta_timestamps=delta_timestamps)
+
+# Create the optimizer and dataloader for offline training
+optimizer = cfg.get_optimizer_preset().build(policy.parameters())
+batch_size = 32
+dataloader = torch.utils.data.DataLoader(
+    dataset,
+    batch_size=batch_size,
+    shuffle=True,
+    pin_memory=device.type != "cpu",
+    drop_last=True,
+)
+
+# Number of training steps and logging frequency
+training_steps = 1
+log_freq = 1
+
+# Run training loop
+step = 0
+done = False
+while not done:
+    for batch in dataloader:
+        batch = preprocessor(batch)
+        loss, _ = policy.forward(batch)
+        loss.backward()
+        optimizer.step()
+        optimizer.zero_grad()
+
+        if step % log_freq == 0:
+            print(f"step: {step} loss: {loss.item():.3f}")
+        step += 1
+        if step >= training_steps:
+            done = True
+            break
+
+# Save the policy checkpoint, alongside the pre/post processors
+policy.save_pretrained(output_directory)
+preprocessor.save_pretrained(output_directory)
+postprocessor.save_pretrained(output_directory)
+
+# Save all assets to the Hub
+policy.push_to_hub("fracapuano/robot_learning_tutorial_act")
+preprocessor.push_to_hub("fracapuano/robot_learning_tutorial_act")
+postprocessor.push_to_hub("fracapuano/robot_learning_tutorial_act")

examples/tutorial/act/act_using_example.py

@@ -0,0 +1,57 @@
+import torch
+
+from lerobot.cameras.opencv.configuration_opencv import OpenCVCameraConfig
+from lerobot.datasets.lerobot_dataset import LeRobotDatasetMetadata
+from lerobot.policies.act.modeling_act import ACTPolicy
+from lerobot.policies.factory import make_pre_post_processors
+from lerobot.policies.utils import build_inference_frame, make_robot_action
+from lerobot.robots.so100_follower.config_so100_follower import SO100FollowerConfig
+from lerobot.robots.so100_follower.so100_follower import SO100Follower
+
+device = torch.device("mps")  # or "cuda" or "cpu"
+model_id = "fracapuano/robot_learning_tutorial_act"
+model = ACTPolicy.from_pretrained(model_id)
+
+dataset_id = "lerobot/svla_so101_pickplace"
+# This only downloads the metadata for the dataset, ~10s of MB even for large-scale datasets
+dataset_metadata = LeRobotDatasetMetadata(dataset_id)
+preprocess, postprocess = make_pre_post_processors(model.config, dataset_stats=dataset_metadata.stats)
+
+# # find ports using lerobot-find-port
+follower_port = ...  # something like "/dev/tty.usbmodem58760431631"
+
+# # the robot ids are used the load the right calibration files
+follower_id = ...  # something like "follower_so100"
+
+MAX_EPISODES = 5
+MAX_STEPS_PER_EPISODE = 20
+
+# Robot and environment configuration
+# Camera keys must match the name and resolutions of the ones used for training!
+# You can check the camera keys expected by a model in the info.json card on the model card on the Hub
+camera_config = {
+    "side": OpenCVCameraConfig(index_or_path=0, width=640, height=480, fps=30),
+    "up": OpenCVCameraConfig(index_or_path=1, width=640, height=480, fps=30),
+}
+
+robot_cfg = SO100FollowerConfig(port=follower_port, id=follower_id, cameras=camera_config)
+robot = SO100Follower(robot_cfg)
+robot.connect()
+
+for _ in range(MAX_EPISODES):
+    for _ in range(MAX_STEPS_PER_EPISODE):
+        obs = robot.get_observation()
+        obs_frame = build_inference_frame(
+            observation=obs, ds_features=dataset_metadata.features, device=device
+        )
+
+        obs = preprocess(obs_frame)
+
+        action = model.select_action(obs)
+        action = postprocess(action)
+
+        action = make_robot_action(action, dataset_metadata.features)
+
+        robot.send_action(action)
+
+    print("Episode finished! Starting new episode...")

examples/tutorial/async-inf/policy_server.py

@@ -0,0 +1,11 @@
+from lerobot.async_inference.configs import PolicyServerConfig
+from lerobot.async_inference.policy_server import serve
+
+host = ...  # something like "127.0.0.1" if you're exposing to localhost
+port = ...  # something like 8080
+
+config = PolicyServerConfig(
+    host=host,
+    port=port,
+)
+serve(config)

examples/tutorial/async-inf/robot_client.py

@@ -0,0 +1,55 @@
+import threading
+
+from lerobot.async_inference.configs import RobotClientConfig
+from lerobot.async_inference.helpers import visualize_action_queue_size
+from lerobot.async_inference.robot_client import RobotClient
+from lerobot.cameras.opencv.configuration_opencv import OpenCVCameraConfig
+from lerobot.robots.so100_follower import SO100FollowerConfig
+
+# these cameras must match the ones expected by the policy - find your cameras with lerobot-find-cameras
+# check the config.json on the Hub for the policy you are using to see the expected camera specs
+camera_cfg = {
+    "up": OpenCVCameraConfig(index_or_path=0, width=640, height=480, fps=30),
+    "side": OpenCVCameraConfig(index_or_path=1, width=640, height=480, fps=30),
+}
+
+# # find ports using lerobot-find-port
+follower_port = ...  # something like "/dev/tty.usbmodem58760431631"
+
+# # the robot ids are used the load the right calibration files
+follower_id = ...  # something like "follower_so100"
+
+robot_cfg = SO100FollowerConfig(port=follower_port, id=follower_id, cameras=camera_cfg)
+
+server_address = ...  # something like "127.0.0.1:8080" if using localhost
+
+# 3. Create client configuration
+client_cfg = RobotClientConfig(
+    robot=robot_cfg,
+    server_address=server_address,
+    policy_device="mps",
+    policy_type="act",
+    pretrained_name_or_path="fracapuano/robot_learning_tutorial_act",
+    chunk_size_threshold=0.5,  # g
+    actions_per_chunk=50,  # make sure this is less than the max actions of the policy
+)
+
+# 4. Create and start client
+client = RobotClient(client_cfg)
+
+# 5. Provide a textual description of the task
+task = ...
+
+if client.start():
+    # Start action receiver thread
+    action_receiver_thread = threading.Thread(target=client.receive_actions, daemon=True)
+    action_receiver_thread.start()
+
+    try:
+        # Run the control loop
+        client.control_loop(task)
+    except KeyboardInterrupt:
+        client.stop()
+        action_receiver_thread.join()
+        # (Optionally) plot the action queue size
+        visualize_action_queue_size(client.action_queue_size)

examples/tutorial/diffusion/diffusion_training_example.py

@@ -0,0 +1,99 @@
+"""This script demonstrates how to train Diffusion Policy on a real-world dataset."""
+
+from pathlib import Path
+
+import torch
+
+from lerobot.configs.types import FeatureType
+from lerobot.datasets.lerobot_dataset import LeRobotDataset, LeRobotDatasetMetadata
+from lerobot.datasets.utils import dataset_to_policy_features
+from lerobot.policies.diffusion.configuration_diffusion import DiffusionConfig
+from lerobot.policies.diffusion.modeling_diffusion import DiffusionPolicy
+from lerobot.policies.factory import make_pre_post_processors
+
+
+def make_delta_timestamps(delta_indices: list[int] | None, fps: int) -> list[float]:
+    if delta_indices is None:
+        return [0]
+
+    return [i / fps for i in delta_indices]
+
+
+output_directory = Path("outputs/robot_learning_tutorial/diffusion")
+output_directory.mkdir(parents=True, exist_ok=True)
+
+# Select your device
+device = torch.device("mps")  # or "cuda" or "cpu"
+
+dataset_id = "lerobot/svla_so101_pickplace"
+
+# This specifies the inputs the model will be expecting and the outputs it will produce
+dataset_metadata = LeRobotDatasetMetadata(dataset_id)
+features = dataset_to_policy_features(dataset_metadata.features)
+
+output_features = {key: ft for key, ft in features.items() if ft.type is FeatureType.ACTION}
+input_features = {key: ft for key, ft in features.items() if key not in output_features}
+
+cfg = DiffusionConfig(input_features=input_features, output_features=output_features)
+policy = DiffusionPolicy(cfg)
+preprocessor, postprocessor = make_pre_post_processors(cfg, dataset_stats=dataset_metadata.stats)
+
+policy.train()
+policy.to(device)
+
+# To perform action chunking, ACT expects a given number of actions as targets
+delta_timestamps = {
+    "observation.state": make_delta_timestamps(cfg.observation_delta_indices, dataset_metadata.fps),
+    "action": make_delta_timestamps(cfg.action_delta_indices, dataset_metadata.fps),
+}
+
+# add image features if they are present
+delta_timestamps |= {
+    k: make_delta_timestamps(cfg.observation_delta_indices, dataset_metadata.fps) for k in cfg.image_features
+}
+
+# Instantiate the dataset
+dataset = LeRobotDataset(dataset_id, delta_timestamps=delta_timestamps)
+
+# Create the optimizer and dataloader for offline training
+optimizer = cfg.get_optimizer_preset().build(policy.parameters())
+batch_size = 32
+dataloader = torch.utils.data.DataLoader(
+    dataset,
+    batch_size=batch_size,
+    shuffle=True,
+    pin_memory=device.type != "cpu",
+    drop_last=True,
+)
+
+# Number of training steps and logging frequency
+training_steps = 1
+log_freq = 1
+
+# Run training loop
+step = 0
+done = False
+while not done:
+    for batch in dataloader:
+        batch = preprocessor(batch)
+        loss, _ = policy.forward(batch)
+        loss.backward()
+        optimizer.step()
+        optimizer.zero_grad()
+
+        if step % log_freq == 0:
+            print(f"step: {step} loss: {loss.item():.3f}")
+        step += 1
+        if step >= training_steps:
+            done = True
+            break
+
+# Save the policy checkpoint, alongside the pre/post processors
+policy.save_pretrained(output_directory)
+preprocessor.save_pretrained(output_directory)
+postprocessor.save_pretrained(output_directory)
+
+# Save all assets to the Hub
+policy.push_to_hub("fracapuano/robot_learning_tutorial_diffusion")
+preprocessor.push_to_hub("fracapuano/robot_learning_tutorial_diffusion")
+postprocessor.push_to_hub("fracapuano/robot_learning_tutorial_diffusion")

examples/tutorial/diffusion/diffusion_using_example.py

@@ -0,0 +1,60 @@
+import torch
+
+from lerobot.cameras.opencv.configuration_opencv import OpenCVCameraConfig
+from lerobot.datasets.lerobot_dataset import LeRobotDatasetMetadata
+from lerobot.policies.diffusion.modeling_diffusion import DiffusionPolicy
+from lerobot.policies.factory import make_pre_post_processors
+from lerobot.policies.utils import build_inference_frame, make_robot_action
+from lerobot.robots.so100_follower.config_so100_follower import SO100FollowerConfig
+from lerobot.robots.so100_follower.so100_follower import SO100Follower
+
+device = torch.device("mps")  # or "cuda" or "cpu"
+model_id = "fracapuano/robot_learning_tutorial_diffusion"
+
+model = DiffusionPolicy.from_pretrained(model_id)
+
+dataset_id = "lerobot/svla_so101_pickplace"
+# This only downloads the metadata for the dataset, ~10s of MB even for large-scale datasets
+dataset_metadata = LeRobotDatasetMetadata(dataset_id)
+preprocess, postprocess = make_pre_post_processors(
+    model.config, model_id, dataset_stats=dataset_metadata.stats
+)
+
+MAX_EPISODES = 5
+MAX_STEPS_PER_EPISODE = 20
+
+
+# # find ports using lerobot-find-port
+follower_port = ...  # something like "/dev/tty.usbmodem58760431631"
+
+# # the robot ids are used the load the right calibration files
+follower_id = ...  # something like "follower_so100"
+
+# Robot and environment configuration
+# Camera keys must match the name and resolutions of the ones used for training!
+# You can check the camera keys expected by a model in the info.json card on the model card on the Hub
+camera_config = {
+    "side": OpenCVCameraConfig(index_or_path=0, width=640, height=480, fps=30),
+    "up": OpenCVCameraConfig(index_or_path=1, width=640, height=480, fps=30),
+}
+
+robot_cfg = SO100FollowerConfig(port=follower_port, id=follower_id, cameras=camera_config)
+robot = SO100Follower(robot_cfg)
+robot.connect()
+
+
+for _ in range(MAX_EPISODES):
+    for _ in range(MAX_STEPS_PER_EPISODE):
+        obs = robot.get_observation()
+        obs_frame = build_inference_frame(
+            observation=obs, ds_features=dataset_metadata.features, device=device
+        )
+
+        obs = preprocess(obs_frame)
+
+        action = model.select_action(obs)
+        action = postprocess(action)
+        action = make_robot_action(action, dataset_metadata.features)
+        robot.send_action(action)
+
+    print("Episode finished! Starting new episode...")

examples/tutorial/pi0/using_pi0_example.py

@@ -0,0 +1,67 @@
+import torch
+
+from lerobot.cameras.opencv.configuration_opencv import OpenCVCameraConfig
+from lerobot.datasets.utils import hw_to_dataset_features
+from lerobot.policies.factory import make_pre_post_processors
+from lerobot.policies.pi0.modeling_pi0 import PI0Policy
+from lerobot.policies.utils import build_inference_frame, make_robot_action
+from lerobot.robots.so100_follower.config_so100_follower import SO100FollowerConfig
+from lerobot.robots.so100_follower.so100_follower import SO100Follower
+
+MAX_EPISODES = 5
+MAX_STEPS_PER_EPISODE = 20
+
+device = torch.device("mps")  # or "cuda" or "cpu"
+model_id = "lerobot/pi0_base"
+
+model = PI0Policy.from_pretrained(model_id)
+
+preprocess, postprocess = make_pre_post_processors(
+    model.config,
+    model_id,
+    # This overrides allows to run on MPS, otherwise defaults to CUDA (if available)
+    preprocessor_overrides={"device_processor": {"device": str(device)}},
+)
+
+# find ports using lerobot-find-port
+follower_port = ...  # something like "/dev/tty.usbmodem58760431631"
+
+# the robot ids are used the load the right calibration files
+follower_id = ...  # something like "follower_so100"
+
+# Robot and environment configuration
+# Camera keys must match the name and resolutions of the ones used for training!
+# You can check the camera keys expected by a model in the info.json card on the model card on the Hub
+camera_config = {
+    "base_0_rgb": OpenCVCameraConfig(index_or_path=0, width=640, height=480, fps=30),
+    "left_wrist_0_rgb": OpenCVCameraConfig(index_or_path=1, width=640, height=480, fps=30),
+    "right_wrist_0_rgb": OpenCVCameraConfig(index_or_path=2, width=640, height=480, fps=30),
+}
+
+robot_cfg = SO100FollowerConfig(port=follower_port, id=follower_id, cameras=camera_config)
+robot = SO100Follower(robot_cfg)
+robot.connect()
+
+task = ""  # something like "pick the red block"
+robot_type = ""  # something like "so100_follower" for multi-embodiment datasets
+
+# This is used to match the raw observation keys to the keys expected by the policy
+action_features = hw_to_dataset_features(robot.action_features, "action")
+obs_features = hw_to_dataset_features(robot.observation_features, "observation")
+dataset_features = {**action_features, **obs_features}
+
+for _ in range(MAX_EPISODES):
+    for _ in range(MAX_STEPS_PER_EPISODE):
+        obs = robot.get_observation()
+        obs_frame = build_inference_frame(
+            observation=obs, ds_features=dataset_features, device=device, task=task, robot_type=robot_type
+        )
+
+        obs = preprocess(obs_frame)
+
+        action = model.select_action(obs)
+        action = postprocess(action)
+        action = make_robot_action(action, dataset_features)
+        robot.send_action(action)
+
+    print("Episode finished! Starting new episode...")

examples/tutorial/rl/hilserl_example.py

@@ -0,0 +1,345 @@
+import multiprocessing as mp
+import signal
+from pathlib import Path
+from queue import Empty, Full
+
+import torch
+import torch.optim as optim
+
+from lerobot.datasets.lerobot_dataset import LeRobotDataset
+from lerobot.datasets.utils import hw_to_dataset_features
+from lerobot.envs.configs import HILSerlProcessorConfig, HILSerlRobotEnvConfig
+from lerobot.policies.sac.configuration_sac import SACConfig
+from lerobot.policies.sac.modeling_sac import SACPolicy
+from lerobot.policies.sac.reward_model.modeling_classifier import Classifier
+from lerobot.rl.buffer import ReplayBuffer
+from lerobot.rl.gym_manipulator import make_robot_env
+from lerobot.robots.so100_follower import SO100FollowerConfig
+from lerobot.teleoperators.so100_leader import SO100LeaderConfig
+from lerobot.teleoperators.utils import TeleopEvents
+
+LOG_EVERY = 10
+SEND_EVERY = 10
+
+
+def run_learner(
+    transitions_queue: mp.Queue,
+    parameters_queue: mp.Queue,
+    shutdown_event: mp.Event,
+    policy_learner: SACPolicy,
+    online_buffer: ReplayBuffer,
+    offline_buffer: ReplayBuffer,
+    lr: float = 3e-4,
+    batch_size: int = 32,
+    device: torch.device = "mps",
+):
+    """The learner process - trains SAC policy on transitions streamed from the actor, updating parameters
+    for the actor to adopt."""
+    policy_learner.train()
+    policy_learner.to(device)
+
+    # Create Adam optimizer from scratch - simple and clean
+    optimizer = optim.Adam(policy_learner.parameters(), lr=lr)
+
+    print(f"[LEARNER] Online buffer capacity: {online_buffer.capacity}")
+    print(f"[LEARNER] Offline buffer capacity: {offline_buffer.capacity}")
+
+    training_step = 0
+
+    while not shutdown_event.is_set():
+        # retrieve incoming transitions from the actor process
+        try:
+            transitions = transitions_queue.get(timeout=0.1)
+            for transition in transitions:
+                # HIL-SERL: Add ALL transitions to online buffer
+                online_buffer.add(**transition)
+
+                # HIL-SERL: Add ONLY human intervention transitions to offline buffer
+                is_intervention = transition.get("complementary_info", {}).get("is_intervention", False)
+                if is_intervention:
+                    offline_buffer.add(**transition)
+                    print(
+                        f"[LEARNER] Human intervention detected! Added to offline buffer (now {len(offline_buffer)} transitions)"
+                    )
+
+        except Empty:
+            pass  # No transitions available, continue
+
+        # Train if we have enough data
+        if len(online_buffer) >= policy_learner.config.online_step_before_learning:
+            # Sample from online buffer (autonomous + human data)
+            online_batch = online_buffer.sample(batch_size // 2)
+
+            # Sample from offline buffer (human demonstrations only, either precollected or at runtime)
+            offline_batch = offline_buffer.sample(batch_size // 2)
+
+            # Combine batches - this is the key HIL-SERL mechanism!
+            batch = {}
+            for key in online_batch:
+                if key in offline_batch:
+                    batch[key] = torch.cat([online_batch[key], offline_batch[key]], dim=0)
+                else:
+                    batch[key] = online_batch[key]
+
+            loss, _ = policy_learner.forward(batch)
+
+            optimizer.zero_grad()
+            loss.backward()
+            optimizer.step()
+            training_step += 1
+
+            if training_step % LOG_EVERY == 0:
+                print(
+                    f"[LEARNER] Training step {training_step}, Loss: {loss.item():.4f}, "
+                    f"Buffers: Online={len(online_buffer)}, Offline={len(offline_buffer)}"
+                )
+
+            # Send updated parameters to actor every 10 training steps
+            if training_step % SEND_EVERY == 0:
+                try:
+                    state_dict = {k: v.cpu() for k, v in policy_learner.state_dict().items()}
+                    parameters_queue.put_nowait(state_dict)
+                    print("[LEARNER] Sent updated parameters to actor")
+                except Full:
+                    # Missing write due to queue not being consumed (should happen rarely)
+                    pass
+
+    print("[LEARNER] Learner process finished")
+
+
+def run_actor(
+    transitions_queue: mp.Queue,
+    parameters_queue: mp.Queue,
+    shutdown_event: mp.Event,
+    policy_actor: SACPolicy,
+    reward_classifier: Classifier,
+    env_cfg: HILSerlRobotEnvConfig,
+    device: torch.device = "mps",
+    output_directory: Path | None = None,
+):
+    """The actor process - interacts with environment and collects data.
+    The policy is frozen and only the parameters are updated, popping the most recent ones from a queue."""
+    policy_actor.eval()
+    policy_actor.to(device)
+
+    reward_classifier.eval()
+    reward_classifier.to(device)
+
+    # Create robot environment inside the actor process
+    env, teleop_device = make_robot_env(env_cfg)
+
+    try:
+        for episode in range(MAX_EPISODES):
+            if shutdown_event.is_set():
+                break
+
+            obs, _info = env.reset()
+            episode_reward = 0.0
+            step = 0
+            episode_transitions = []
+
+            print(f"[ACTOR] Starting episode {episode + 1}")
+
+            while step < MAX_STEPS_PER_EPISODE and not shutdown_event.is_set():
+                try:
+                    new_params = parameters_queue.get_nowait()
+                    policy_actor.load_state_dict(new_params)
+                    print("[ACTOR] Updated policy parameters from learner")
+                except Empty:  # No new updated parameters available from learner, waiting
+                    pass
+
+                # Get action from policy
+                policy_obs = make_policy_obs(obs, device=device)
+                action_tensor = policy_actor.select_action(policy_obs)  # predicts a single action
+                action = action_tensor.squeeze(0).cpu().numpy()
+
+                # Step environment
+                next_obs, _env_reward, terminated, truncated, _info = env.step(action)
+                done = terminated or truncated
+
+                # Predict reward
+                policy_next_obs = make_policy_obs(next_obs, device=device)
+                reward = reward_classifier.predict_reward(policy_next_obs)
+
+                if reward >= 1.0 and not done:  # success detected! halt episode
+                    terminated = True
+                    done = True
+
+                # In HIL-SERL, human interventions come from the teleop device
+                is_intervention = False
+                if hasattr(teleop_device, "get_teleop_events"):
+                    # Real intervention detection from teleop device
+                    teleop_events = teleop_device.get_teleop_events()
+                    is_intervention = teleop_events.get(TeleopEvents.IS_INTERVENTION, False)
+
+                # Store transition with intervention metadata
+                transition = {
+                    "state": policy_obs,
+                    "action": action,
+                    "reward": float(reward) if hasattr(reward, "item") else reward,
+                    "next_state": policy_next_obs,
+                    "done": done,
+                    "truncated": truncated,
+                    "complementary_info": {
+                        "is_intervention": is_intervention,
+                    },
+                }
+
+                episode_transitions.append(transition)
+
+                episode_reward += reward
+                step += 1
+
+                obs = next_obs
+
+                if done:
+                    break
+
+            # Send episode transitions to learner
+            transitions_queue.put_nowait(episode_transitions)
+
+    except KeyboardInterrupt:
+        print("[ACTOR] Interrupted by user")
+    finally:
+        # Clean up
+        if hasattr(env, "robot") and env.robot.is_connected:
+            env.robot.disconnect()
+        if teleop_device and hasattr(teleop_device, "disconnect"):
+            teleop_device.disconnect()
+        if output_directory is not None:
+            policy_actor.save_pretrained(output_directory)
+            print(f"[ACTOR] Latest actor policy saved at: {output_directory}")
+
+        print("[ACTOR] Actor process finished")
+
+
+def make_policy_obs(obs, device: torch.device = "cpu"):
+    return {
+        "observation.state": torch.from_numpy(obs["agent_pos"]).float().unsqueeze(0).to(device),
+        **{
+            f"observation.image.{k}": torch.from_numpy(obs["pixels"][k]).float().unsqueeze(0).to(device)
+            for k in obs["pixels"]
+        },
+    }
+
+
+"""Main function - coordinates actor and learner processes."""
+
+device = "mps"  # or "cuda" or "cpu"
+output_directory = Path("outputs/robot_learning_tutorial/hil_serl")
+output_directory.mkdir(parents=True, exist_ok=True)
+
+# find ports using lerobot-find-port
+follower_port = ...
+leader_port = ...
+
+# the robot ids are used the load the right calibration files
+follower_id = ...
+leader_id = ...
+
+# A pretrained model (to be used in-distribution!)
+reward_classifier_id = "fracapuano/reward_classifier_hil_serl_example"
+reward_classifier = Classifier.from_pretrained(reward_classifier_id)
+
+reward_classifier.to(device)
+reward_classifier.eval()
+
+MAX_EPISODES = 5
+MAX_STEPS_PER_EPISODE = 20
+
+# Robot and environment configuration
+robot_cfg = SO100FollowerConfig(port=follower_port, id=follower_id)
+teleop_cfg = SO100LeaderConfig(port=leader_port, id=leader_id)
+processor_cfg = HILSerlProcessorConfig(control_mode="leader")
+
+env_cfg = HILSerlRobotEnvConfig(robot=robot_cfg, teleop=teleop_cfg, processor=processor_cfg)
+
+# Create robot environment
+env, teleop_device = make_robot_env(env_cfg)
+
+obs_features = hw_to_dataset_features(env.robot.observation_features, "observation")
+action_features = hw_to_dataset_features(env.robot.action_features, "action")
+
+# Create SAC policy for action selection
+policy_cfg = SACConfig(
+    device=device,
+    input_features=obs_features,
+    output_features=action_features,
+)
+
+policy_actor = SACPolicy(policy_cfg)
+policy_learner = SACPolicy(policy_cfg)
+
+demonstrations_repo_id = "lerobot/example_hil_serl_dataset"
+offline_dataset = LeRobotDataset(repo_id=demonstrations_repo_id)
+
+# Online buffer: initialized from scratch
+online_replay_buffer = ReplayBuffer(device=device, state_keys=list(obs_features.keys()))
+# Offline buffer: Created from dataset (pre-populated it with demonstrations)
+offline_replay_buffer = ReplayBuffer.from_lerobot_dataset(
+    lerobot_dataset=offline_dataset, device=device, state_keys=list(obs_features.keys())
+)
+
+# Create communication channels between learner and actor processes
+transitions_queue = mp.Queue(maxsize=10)
+parameters_queue = mp.Queue(maxsize=2)
+shutdown_event = mp.Event()
+
+
+# Signal handler for graceful shutdown
+def signal_handler(sig):
+    print(f"\nSignal {sig} received, shutting down...")
+    shutdown_event.set()
+
+
+signal.signal(signal.SIGINT, signal_handler)
+signal.signal(signal.SIGTERM, signal_handler)
+
+# Create processes
+learner_process = mp.Process(
+    target=run_learner,
+    args=(
+        transitions_queue,
+        parameters_queue,
+        shutdown_event,
+        policy_learner,
+        online_replay_buffer,
+        offline_replay_buffer,
+    ),
+    kwargs={"device": device},  # can run on accelerated hardware for training
+)
+
+actor_process = mp.Process(
+    target=run_actor,
+    args=(
+        transitions_queue,
+        parameters_queue,
+        shutdown_event,
+        policy_actor,
+        reward_classifier,
+        env_cfg,
+        output_directory,
+    ),
+    kwargs={"device": "cpu"},  # actor is frozen, can run on CPU or accelerate for inference
+)
+
+learner_process.start()
+actor_process.start()
+
+try:
+    # Wait for actor to finish (it controls the episode loop)
+    actor_process.join()
+    shutdown_event.set()
+    learner_process.join(timeout=10)
+
+except KeyboardInterrupt:
+    print("Main process interrupted")
+    shutdown_event.set()
+    actor_process.join(timeout=5)
+    learner_process.join(timeout=10)
+
+finally:
+    if learner_process.is_alive():
+        learner_process.terminate()
+    if actor_process.is_alive():
+        actor_process.terminate()

examples/tutorial/rl/reward_classifier_example.py

@@ -0,0 +1,62 @@
+import torch
+
+from lerobot.datasets.lerobot_dataset import LeRobotDataset
+from lerobot.policies.factory import make_policy, make_pre_post_processors
+from lerobot.policies.sac.reward_model.configuration_classifier import RewardClassifierConfig
+
+# Device to use for training
+device = "mps"  # or "cuda", or "cpu"
+
+# Load the dataset used for training
+repo_id = "lerobot/example_hil_serl_dataset"
+dataset = LeRobotDataset(repo_id)
+
+# Configure the policy to extract features from the image frames
+camera_keys = dataset.meta.camera_keys
+
+config = RewardClassifierConfig(
+    num_cameras=len(camera_keys),
+    device=device,
+    # backbone model to extract features from the image frames
+    model_name="microsoft/resnet-18",
+)
+
+# Make policy, preprocessor, and optimizer
+policy = make_policy(config, ds_meta=dataset.meta)
+optimizer = config.get_optimizer_preset().build(policy.parameters())
+preprocessor, _ = make_pre_post_processors(policy_cfg=config, dataset_stats=dataset.meta.stats)
+
+
+classifier_id = "fracapuano/reward_classifier_hil_serl_example"
+
+# Instantiate a dataloader
+dataloader = torch.utils.data.DataLoader(dataset, batch_size=16, shuffle=True)
+
+# Training loop
+num_epochs = 5
+for epoch in range(num_epochs):
+    total_loss = 0
+    total_accuracy = 0
+    for batch in dataloader:
+        # Preprocess the batch and move it to the correct device.
+        batch = preprocessor(batch)
+
+        # Forward pass
+        loss, output_dict = policy.forward(batch)
+
+        # Backward pass and optimization
+        optimizer.zero_grad()
+        loss.backward()
+        optimizer.step()
+
+        total_loss += loss.item()
+        total_accuracy += output_dict["accuracy"]
+
+    avg_loss = total_loss / len(dataloader)
+    avg_accuracy = total_accuracy / len(dataloader)
+    print(f"Epoch {epoch + 1}/{num_epochs}, Loss: {avg_loss:.4f}, Accuracy: {avg_accuracy:.2f}%")
+
+print("Training finished!")
+
+# You can now save the trained policy.
+policy.push_to_hub(classifier_id)

examples/tutorial/smolvla/using_smolvla_example.py

@@ -0,0 +1,66 @@
+import torch
+
+from lerobot.cameras.opencv.configuration_opencv import OpenCVCameraConfig
+from lerobot.datasets.utils import hw_to_dataset_features
+from lerobot.policies.factory import make_pre_post_processors
+from lerobot.policies.smolvla.modeling_smolvla import SmolVLAPolicy
+from lerobot.policies.utils import build_inference_frame, make_robot_action
+from lerobot.robots.so100_follower.config_so100_follower import SO100FollowerConfig
+from lerobot.robots.so100_follower.so100_follower import SO100Follower
+
+MAX_EPISODES = 5
+MAX_STEPS_PER_EPISODE = 20
+
+device = torch.device("mps")  # or "cuda" or "cpu"
+model_id = "lerobot/smolvla_base"
+
+model = SmolVLAPolicy.from_pretrained(model_id)
+
+preprocess, postprocess = make_pre_post_processors(
+    model.config,
+    model_id,
+    # This overrides allows to run on MPS, otherwise defaults to CUDA (if available)
+    preprocessor_overrides={"device_processor": {"device": str(device)}},
+)
+
+# find ports using lerobot-find-port
+follower_port = ...  # something like "/dev/tty.usbmodem58760431631"
+
+# the robot ids are used the load the right calibration files
+follower_id = ...  # something like "follower_so100"
+
+# Robot and environment configuration
+# Camera keys must match the name and resolutions of the ones used for training!
+# You can check the camera keys expected by a model in the info.json card on the model card on the Hub
+camera_config = {
+    "camera1": OpenCVCameraConfig(index_or_path=0, width=640, height=480, fps=30),
+    "camera2": OpenCVCameraConfig(index_or_path=1, width=640, height=480, fps=30),
+}
+
+robot_cfg = SO100FollowerConfig(port=follower_port, id=follower_id, cameras=camera_config)
+robot = SO100Follower(robot_cfg)
+robot.connect()
+
+task = ""  # something like "pick the red block"
+robot_type = ""  # something like "so100_follower" for multi-embodiment datasets
+
+# This is used to match the raw observation keys to the keys expected by the policy
+action_features = hw_to_dataset_features(robot.action_features, "action")
+obs_features = hw_to_dataset_features(robot.observation_features, "observation")
+dataset_features = {**action_features, **obs_features}
+
+for _ in range(MAX_EPISODES):
+    for _ in range(MAX_STEPS_PER_EPISODE):
+        obs = robot.get_observation()
+        obs_frame = build_inference_frame(
+            observation=obs, ds_features=dataset_features, device=device, task=task, robot_type=robot_type
+        )
+
+        obs = preprocess(obs_frame)
+
+        action = model.select_action(obs)
+        action = postprocess(action)
+        action = make_robot_action(action, dataset_features)
+        robot.send_action(action)
+
+    print("Episode finished! Starting new episode...")