rtc-pi05 — Real-Time Chunking for openpi pi0.5

Standalone, inference-time Real-Time Chunking (RTC) for openpi pi0.5 flow policies.

Black, Galliker, Levine. Real-Time Execution of Action Chunking Flow Policies. NeurIPS 2025. arXiv:2506.07339.

You bring your own pi0.5 base checkpoint (loaded through openpi). RTC only replaces the inference path: while the current action chunk executes, a background thread generates the next chunk by guided inpainting — freezing the actions already committed to the robot and softly agreeing with the overlapping ones — so consecutive chunks stay consistent and execution never stalls waiting on inference.

This repo is the RTC algorithm plus a thin openpi adapter and a no-robot demo. It contains no robot / camera / task-specific code.

What's inside

rtc/
  masking.py      soft prefix masks + PiGDM guidance weight   (backend-agnostic, numpy)
  controller.py   Algorithm 1: async execution + background inference thread
  sampler.py      guided-inpainting denoising loop            (PyTorch, torch.autograd.grad)
  pi0.py          adapter for openpi PI0Pytorch pi0.5          (torch + openpi, imported lazily)
examples/
  pi05_realtime.py   load a pi0.5 ckpt -> run the async RTC controller (synthetic obs)
tests/
  test_masking.py    property tests for the soft masks

import rtc pulls in neither torch nor openpi — the sampler and adapter are lazy, so the controller/masking core is usable in a plain-numpy process.

Install

Requires Python ≥ 3.11 and a working openpi install (it provides PI0Pytorch and the checkpoint/normalization machinery; it is not on PyPI, so install it from source into the same environment).

# in the environment that already has openpi + torch:
pip install -e ".[torch,dev]"
pytest                       # runs the masking property tests

Quickstart (no robot)

python examples/pi05_realtime.py \
    --config pi05_libero \                 # your openpi TrainConfig name
    --checkpoint /path/to/pi05_ckpt_dir \  # dir with model.safetensors + assets/
    --device cuda --steps 60 --hz 30

It loads the policy, samples an initial chunk, then simulates a 30 Hz control loop while the background thread regenerates chunks RTC-style, printing the observed inference delay d and confirming the real-time constraint d ≤ H − s holds. Pass --obs-npz to feed a recorded observation instead of the synthetic one (its keys must match your checkpoint's input transform).

Using it in your own inference loop

import torch
from openpi.training import config as _config
from openpi.policies import policy_config as _policy_config
from rtc.controller import RealTimeChunkingController
from rtc.pi0 import build_observation, sample_initial_chunk, rtc_sample_actions, decode_actions

policy = _policy_config.create_trained_policy(
    _config.get_config("pi05_libero"), "/path/to/ckpt", pytorch_device="cuda")
H = int(policy._model.config.action_horizon)

# 1. seed the first chunk (model space) from the same velocity field RTC uses
obs0 = build_observation(policy, raw_obs_dict)
init_chunk, _ = sample_initial_chunk(policy, obs0, num_steps=5)

# 2. the controller calls this in a background thread
def inference_fn(obs, prev_chunk, inference_delay, execution_horizon):
    new_b, _ = rtc_sample_actions(
        policy, obs, prev_chunk.unsqueeze(0),
        inference_delay=inference_delay, execution_horizon=execution_horizon,
        num_steps=5, max_guidance_weight=5.0, prefix_attention_schedule="exp")
    return new_b[0]

ctrl = RealTimeChunkingController(inference_fn, horizon=H, initial_chunk=init_chunk[0],
                                 smin=6, initial_delay=4, delay_buffer_size=10)
ctrl.start()
# 3. every control period dt:
action = ctrl.get_action(build_observation(policy, latest_raw_obs))   # (A,) model space, returns immediately
robot_action = decode_actions(policy, action[None, None], {"state": None})  # -> robot space

Chunks flow in the model's normalized action space (width action_dim, usually padded to 32). Decode to robot space with rtc.pi0.decode_actions, which runs your policy's output transforms (unnormalize / delta→absolute).

RTC hyperparameters

flag / arg	default	paper	meaning
hz	30	1/Δt	control rate
smin	6	s_min	min execution horizon; s = max(d, s_min)
initial_delay	4	d_init	delay-buffer seed (control steps)
delay_buffer	10	b	recent-delay buffer size (delay estimate = its max)
num_steps	5	n	denoising steps
beta	5.0	β	guidance-weight clip (App. A.2)
schedule	exp	—	soft mask; zeros = naive hard-masking baseline

Real-time constraint: d ≤ s ≤ H − d. With pi0.5's action horizon H, if a guided inference takes longer than H − s control steps the current chunk is exhausted and get_action raises. If that happens: lower hz, raise smin, or lower num_steps. Watch the printed observed delay d — it must stay ≤ H − s.

Compare RTC against the naive async baseline with --schedule zeros (hard masking), and against synchronous inference (plain policy.infer).

How the adapter works (conventions)

openpi integrates t=1 noise → t=0 data and its action expert regresses v_pi = noise − data. RTC's core integrates the other way (tau=0 noise → tau=1 data). The adapter reconciles them with

t = 1 − tau ,   v_rtc(x, tau) = − v_pi(x, t = 1 − tau)

Crucially, openpi's PI0Pytorch.denoise_step is not @torch.no_grad (only the compiled sample_actions wrapper is), so RTC calls denoise_step directly and autograd flows back to the noisy chunk x — no model surgery needed. The VLM prefix KV cache is built once per observation under no_grad; each denoising step reuses it. See rtc/pi0.py and rtc/sampler.py for details.

Status / caveats

• The examples/pi05_realtime.py obs keys are libero-style defaults; adapt them (or use --obs-npz) to your checkpoint's input transform.
• The paper's real-world π0.5 used a larger horizon (s_min=25, b=10); with a smaller H the defaults here are what fit d ≤ H − s. Tune on your setup.
• RTC's benefit shows up under inference latency / perturbation; on clean, fast-inference setups it is roughly neutral vs. synchronous.