Chunking em Tempo Real (RTC)

O Chunking em Tempo Real (RTC) permite que políticas robóticas grandes, baseadas em flow-matching, como Pi0, Pi0.5 e SmolVLA, produzam movimentos suaves, contínuos e reativos, apesar da alta latência de inferência. O LeRobot fornece o modo original de inferência guiada em tempo de execução e, para checkpoints compatíveis do Pi05, condicionamento de ação em tempo de treinamento com inferência de prefixo rígido de baixo custo.

Essas políticas geram blocos de ações futuras (por exemplo, 50 passos de cada vez) em vez de ações individuais. Como os modelos são grandes, a produção de cada bloco leva mais tempo do que o necessário para o robô executá-lo. Executar os blocos de forma ingênua leva a problemas como pausas, transições bruscas ou mudanças súbitas de estratégia sempre que o próximo bloco chega atrasado ou diverge das ações anteriormente executadas.

O RTC resolve isso gerando assincronamente o próximo bloco enquanto o robô continua executando o atual, e orientando o novo bloco para que ele se alinhe suavemente com a parte do bloco anterior que já foi executada.

Como o RTC funciona (simplificado)

RTC permite que o robeto pense à frente enquanto ainda está se movendo. Quando o robô está executando um bloco de ações, o RTC começa a criar o próximo bloco com antecedência. Mas, como o robô já se moveu um pouco até o momento em que o novo bloco está pronto, o RTC precisa garantir que o novo bloco ainda se alinhe suavemente com o que o robô está fazendo atualmente.

Para fazer isso, o RTC trata o início do novo chunk como um problema de inpainting ou “preenchimento de lacunas”: ele ajusta suavemente a primeira parte do novo chunk para que ele se misture naturalmente com o movimento contínuo do robô. O resultado é a ausência de pausas e de saltos repentinos.

Em termos técnicos, o RTC adiciona um termo de orientação ao processo de desruidamento do flow-matching, forçando os timesteps sobrepostos do novo chunk a permanecerem próximos à parte executada do chunk anterior, tipicamente usando uma máscara de transição suave.

Início rápido

Instalação

O RTC é integrado ao LeRobot. Basta instalar as dependências da política de que você precisa:

# For Pi0 or Pi0.5
pip install -e ".[pi]"

# For SmolVLA
pip install -e ".[smolvla]"

Usando RTC com Pi0

Você pode usar lerobot-rollout --strategy.type=base --inference.type=rtc para implantação de RTC em robôs reais. O trecho abaixo fornece um exemplo pseudo simplificado de como o RTC opera com Pi0 em seu pipeline:

from lerobot.policies.pi0 import PI0Policy, PI0Config
from lerobot.configs import RTCAttentionSchedule
from lerobot.policies.rtc import RTCConfig, ActionQueue

# Load the Pi0 config so RTC can be enabled before the policy is built
policy_cfg = PI0Config.from_pretrained("lerobot/pi0_base")

# Enable RTC
policy_cfg.rtc_config = RTCConfig(
    enabled=True,
    execution_horizon=10,  # How many steps to blend with previous chunk
    max_guidance_weight=10.0,  # How strongly to enforce consistency
    prefix_attention_schedule=RTCAttentionSchedule.EXP,  # Exponential blend
)

# Load the policy
policy = PI0Policy.from_pretrained("lerobot/pi0_base", config=policy_cfg).to("cuda")

# Now use predict_action_chunk with RTC parameters
inference_delay = 4  # How many steps of inference latency, this value should be calculated based on the inference latency of the policy

# Initialize the action queue
action_queue = ActionQueue(policy_cfg.rtc_config)

# Start in a separate thread with the following function
def get_actions():
  while True:
    if should_get_actions:

      prev_actions = action_queue.get_left_over()
      obs = get_robot_observations(robot)

      # Generate actions WITH RTC
      actions = policy.predict_action_chunk(
          obs,
          inference_delay=inference_delay,
          prev_chunk_left_over=prev_actions,
      )

      action_queue.merge(
          actions, actions, inference_delay
      )

for step in range(num_steps):
    action = action_queue.get()

    # Execute the first N actions
    execute_actions(action)

Parâmetros Principais

RTCConfig tem os seguintes parâmetros para ajuste:

mode seleciona o método de condicionamento por prefixo de ação:

• guided (padrão) aplica a orientação original do Jacobiano durante o desruidamento e funciona com checkpoints comuns de flow-matching.
• trained aplica hard-inpainting no prefixo do chunk anterior com timesteps de fluxo por ação. Atualmente, requer um checkpoint Pi05 treinado com policy.rtc_training_max_delay > 0 e evita o passo de retropropagação do guidance.

O rollout local e o servidor de política aplicam a mesma regra de admissão conservadora do modo de treinamento: rtc_training_max_delay <= execution_horizon <= prediction_steps - rtc_training_max_delay. Para o Pi05, prediction_steps é o chunk_size do checkpoint. Isso preserva o limite local existente: o prefixo configurado cobre o atraso máximo treinado e deixa pelo menos esse número de etapas previstas fora dele. O rollout local também exige inference.queue_threshold >= rtc_training_max_delay antes de conectar o robô; o tempo de requisições remotas usa suas próprias configurações de reabastecimento.

Aqui, execution_horizon representa a capacidade do prefixo fornecido, não a quantidade de reprodução entre as solicitações. O Pi05 aplica a condição de forma rígida apenas ao prefixo de inference_delay. A regra de admissão não é uma garantia de latência nem uma afirmação de que capacidades maiores devem falhar em tempo de execução; a sobreposição medida, as ações restantes e as verificações de atualidade ainda determinam se um chunk retornado pode ser executado. Chunks simples e RTC guiado não são afetados por essa regra exclusiva de treinamento.

execution_horizon: Quantos passos de tempo a partir do chunk anterior devem ser mantidos para garantir consistência. Valores mais altos resultam em transições mais suaves, mas potencialmente menos reatividade.

Valores típicos: 8-12 passos

RTCConfig(execution_horizon=10)

max_guidance_weight: Quão fortemente impor a consistência com o chunk anterior. Este é um hiperparâmetro que pode ser ajustado para equilibrar a suavidade das transições e a reatividade da política. Para flow matching de 10 passos (SmolVLA, Pi0, Pi0.5), um valor de 10.0 é um valor ótimo.

prefix_attention_schedule: Como ponderar a consistência na região de sobreposição.

• LINEAR: Decaimento linear de inference_delay até execution_horizon
• EXP: decaimento exponencial (recomendado para começar)
• ONES: Peso total em toda a execution_horizon
• ZEROS: Binário (peso total até inference_delay, depois zero)

inference_delay: Quantos timesteps de latência de inferência o seu sistema possui. Isso é passado para predict_action_chunk() em vez do config, pois pode variar em tempo de execução.

Testando o RTC offline

Antes de executar em um robô real, teste o RTC com amostras do conjunto de dados para visualizar como ele funciona:

python examples/rtc/eval_dataset.py \
    --policy.path=lerobot/pi0_libero_finetuned \
    --dataset.repo_id=HuggingFaceVLA/libero \
    --rtc.execution_horizon=10 \
    --rtc.max_guidance_weight=10.0 \
    --device=cuda

Adicione --rtc.mode=trained ao avaliar um checkpoint compatível de RTC Pi05 de tempo de treinamento. Políticas não suportadas rejeitam o modo treinado em vez de recorrer ao RTC guiado.

O script gera uma visualização do processo de remoção de ruído, comparando a geração padrão (esquerda) com o RTC (direita). Nos gráficos do RTC, é possível ver como os primeiros passos (linhas azuis/roxas) são guiados para corresponder à trajetória de referência em vermelho (cauda do chunk anterior), garantindo uma transição suave entre os chunks.

Testando o RTC com um Robô Real

lerobot-rollout \
    --strategy.type=base \
    --policy.path=${HF_USERNAME}/policy_repo_id \
    --inference.type=rtc \
    --inference.rtc.mode=guided \
    --inference.rtc.execution_horizon=10 \
    --inference.rtc.max_guidance_weight=10.0 \
    --robot.type=so100_follower \
    --robot.port=/dev/tty.usbmodem58FA0834591 \
    --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 \
    --device=cuda

Para um checkpoint de RTC Pi05 em tempo de treinamento, altere o modo para trained. O checkpoint registra o atraso máximo suportado, e o rollout valida a latência medida em relação a ele:

lerobot-rollout \
    --strategy.type=base \
    --policy.path=${HF_USERNAME}/pi05_training_rtc \
    --inference.type=rtc \
    --inference.rtc.mode=trained \
    --inference.rtc.execution_horizon=10 \
    --robot.type=so100_follower \
    --robot.port=/dev/tty.usbmodem58FA0834591 \
    --task="Move green small object into the purple platform" \
    --duration=120 \
    --device=cuda

Como o RTC se relaciona com a inferência remota

--inference.type=rtc é o RTC local: a inferência é executada em uma thread em segundo plano no processo de rollout e não usa transporte. Executar um servidor de política e um cliente de robô como processos separados na mesma máquina usa o backend remoto em vez disso.

Ambos os backends assíncronos usam o mesmo comportamento de espera limitada e encerramento em robôs compatíveis. Consulte orçamentos locais de RTC para os padrões de tempo limite. A espera não relaxa as verificações de prefixo, atraso ou frescor do RTC.

Tanto o RTC quanto a inferência remota melhoram o controle de robôs em tempo real, mas resolvem problemas diferentes.

A inferência remota pode usar RTC quando o checkpoint suporta o modo de RTC negociado. Chunks simples também suportam alinhamento opcional de cursor e blending; consulte o guia de inferência remota para modos, compatibilidade e ajuste fino.

Avançado: Rastreamento de depuração

O RTC inclui rastreamento de depuração integrado para ajudar você a entender o que está acontecendo durante a inferência:

# Enable debug tracking
policy_cfg.rtc_config.debug = True
policy_cfg.rtc_config.debug_maxlen = 100
policy.init_rtc_processor()  # re-create the RTC processor so the new debug settings take effect

# After inference, access debug data
debug_data = policy.rtc_processor.get_all_debug_steps()

# Visualize denoising steps, corrections, etc.
from lerobot.policies.rtc.debug_visualizer import RTCDebugVisualizer
visualizer = RTCDebugVisualizer()
# ... create plots

Veja examples/rtc/eval_dataset.py para um exemplo completo de visualização offline de RTC.

Referências

• Políticas de Robô Suaves Como Manteiga - Excelente explicação técnica com resultados de robôs reais
• Physical Intelligence - Chunking em Tempo Real - Artigo original e pesquisa
• Implementação do Kinetix RTC - Implementação de referência da Physical Intelligence
• Condição de Ação no Momento do Treinamento - RTC eficiente com condicionamento de prefixo limpo durante o treinamento
• RLDX-1 - referência do PyTorch usada para a integração do RTC no momento do treinamento