---
license: mit
library_name: rtc-pi05
tags:
- robotics
- pi0
- pi05
- flow-matching
- real-time-chunking
- openpi
---

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

Standalone, inference-time [**Real-Time Chunking (RTC)**](https://arxiv.org/abs/2506.07339)
for [openpi](https://github.com/Physical-Intelligence/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](https://github.com/Physical-Intelligence/openpi)
install (it provides `PI0Pytorch` and the checkpoint/normalization machinery; it
is **not** on PyPI, so install it from source into the same environment).

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

## Quickstart (no robot)

```bash
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

```python
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.