LeRobot documentation

实时分块(RTC)

Hugging Face's logo
Join the Hugging Face community

and get access to the augmented documentation experience

to get started

实时分块(RTC)

实时分块(RTC)使基于流匹配的大型机器人策略(例如 Pi0、Pi0.5 和 SmolVLA)能够在高推理延迟的情况下生成平滑、连续且具有响应性的运动。LeRobot 提供了原始的推理时引导模式,并且对于兼容的 Pi05 检查点,支持训练时动作条件化以及低成本的硬前缀推理。

这些策略生成未来动作的片段(例如每次 50 步),而不是单个动作。 由于模型规模较大,生成每个片段所需的时间长于机器人执行该片段的时间。 如果直接执行这些片段,当下一个片段延迟到达或与之前执行的动作不一致时,会导致停顿、生硬的过渡或策略突然变化等问题。

RTC 通过异步生成下一个数据块,同时让机器人继续执行当前数据块,并引导新数据块与已执行的前一个数据块部分平滑对齐,从而解决了这一问题。

RTC 的工作原理(简化版)

RTC 让机器人在移动过程中提前思考。当机器人正在执行一个动作块时,RTC 会提前开始生成下一个动作块。 但由于在新动作块准备好时,机器人已经移动了一段距离,RTC 必须确保新动作块能与机器人当前的动作平滑衔接。

为此,RTC 将新块的开头视为一个图像修复或“填补空白”问题:它对新块的开头部分进行轻微调整,使其与机器人正在进行的运动自然融合。结果是没有停顿,也没有突然的跳跃。

从技术角度来看,RTC 在流匹配去噪过程中引入了一个引导项,强制新块的重叠时间步保持在上一块已执行部分附近,通常使用软过渡掩码来实现。

快速入门

安装

RTC 已内置于 LeRobot 中。只需安装所需的策略依赖项即可:

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

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

使用 RTC 与 Pi0

您可以使用 lerobot-rollout --strategy.type=base --inference.type=rtc 在真实机器人上进行 RTC 部署。 下面的代码片段提供了一个简化的伪示例,展示了 RTC 如何在您的流水线中与 Pi0 协同工作:

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)

关键参数

RTCConfig 包含以下可调参数:

mode 用于选择动作前缀条件化方法:

  • guided(默认)在去噪过程中应用原始的雅可比引导,并可与普通的流匹配检查点配合使用。
  • trained 会对前一个块的执行前缀进行硬修复(hard-inpaint),并使用逐动作的流时间步。目前,它要求使用 policy.rtc_training_max_delay > 0 训练的 Pi05 检查点,并且会跳过 guidance 的反向传播。

本地 rollout 和策略服务器应用相同的保守训练模式准入规则:rtc_training_max_delay <= execution_horizon <= prediction_steps - rtc_training_max_delay。对于 Pi05,prediction_steps 是检查点的 chunk_size。这保留了现有的本地限制:配置的 prefix 覆盖最大训练延迟,并在其外部至少保留相同数量的预测步骤。本地 rollout 在连接机器人之前还要求 inference.queue_threshold >= rtc_training_max_delay;远程请求的时序使用其自身的 refill 设置。

此处 execution_horizon 指的是提供的预置容量,而非两次请求之间的播放量。Pi05 仅对 inference_delay 前缀进行硬性条件约束。该准入规则并非延迟保证,也不意味着更大的容量在运行时必然失败;实际的重叠度、剩余动作数以及新鲜度检查仍决定返回的块能否执行。普通块和引导式 RTC 不受此仅针对训练的规则影响。

execution_horizon:与之前块保持一致的时间步数。值越高,过渡越平滑,但响应性可能越低。

典型值:8-12 步

RTCConfig(execution_horizon=10)

max_guidance_weight:用于强制与上一块保持一致性的强度。这是一个超参数,可通过调整来平衡过渡的平滑度与策略的响应性。对于 10 步流匹配(SmolVLA、Pi0、Pi0.5),10.0 是一个最优值。

prefix_attention_schedule:如何对重叠区域的一致性进行加权。

  • LINEAR:从 inference_delay 到 execution_horizon 的线性衰减
  • EXP:指数衰减(推荐用于入门)
  • ONES:在整个 execution_horizon 内使用全权重
  • ZEROS:二进制(在 inference_delay 之前为完整权重,之后为零)

inference_delay:您的系统具有多少时间步的推理延迟。由于该值可能在运行时变化,因此它被传递给 predict_action_chunk() 而不是配置。

离线测试 RTC

在真实机器人上运行之前,请使用数据集样本测试 RTC,以可视化其工作原理:

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

在评估兼容的训练时 RTC Pi05 检查点时,请添加 --rtc.mode=trained。不支持的策略会拒绝 trained 模式,而不是回退到 guided RTC。

该脚本生成去噪过程的可视化图,对比标准生成(左侧)与 RTC(右侧)。在 RTC 的图中,可以看到前几步(蓝色/紫色线条)如何被引导以匹配红色的真实轨迹(前一个块的尾部),从而确保块之间的平滑过渡。

Denoising steps with and without RTC

使用真实机器人测试 RTC

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

对于训练时的 RTC Pi05 检查点,请将模式更改为 trained。该检查点会记录其支持的最大延迟,并在 rollout 过程中将实测延迟与之进行验证:

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

RTC 与远程推理的关系

--inference.type=rtc 表示本地 RTC:推理在 rollout 进程的后台线程中运行,且不使用任何传输层。如果在同一台机器上以独立进程的方式分别运行策略服务器和机器人客户端,则会使用远程后端。

两种异步后端在受支持的机器人上采用相同的有界等待和关闭行为。有关其超时默认值,请参阅本地 RTC 预算。等待操作不会放宽 RTC 的前缀、延迟或新鲜度检查。

RTC 和 远程推理 都能提升实时机器人控制,但它们解决的是不同的问题。

方面远程推理RTC
目的在另一个进程或另一台计算机上运行推理基于剩余轨迹进行条件预测
执行本地缓冲播放,可选的对齐和混合针对受支持策略的引导式或训练式续写
关系可以传输普通数据块或 RTC可在本地或通过远程后端执行
限制延迟、新鲜度和缓冲区覆盖范围仍然制约着运动需要策略声明的 RTC 合约

当检查点支持协商的 RTC 模式时,远程推理可以使用 RTC。普通块也支持可选的光标对齐和混合;有关模式、兼容性和调优,请参阅远程推理指南。

高级:调试跟踪

RTC 内置了调试跟踪功能,可帮助您了解推理过程中正在发生的事情:

# 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

有关离线 RTC 可视化的完整示例,请参阅 examples/rtc/eval_dataset.py。

参考

Update on GitHub