Compare commits

...

19 Commits

Author SHA1 Message Date
Khalil Meftah 9a846c4fca fix(advantage): update frame count calculation in constant mode 2026-07-03 17:39:23 +02:00
Khalil Meftah ad32d3e00d fix(annotation): skip vlm initialization when using advantage module 2026-07-03 17:33:38 +02:00
Khalil Meftah 1cd1ec468e feat(molmoact2): add RECAP advantage conditioning via recipe system for MolmoAct2
- Add recipe_path, advantage_prefix, cfg_beta to MolmoAct2Config
- Place advantage clause in the assistant section of _build_robot_text
- Add MolmoAct2NormalizeTaskStep for consistent task normalization
- Parse recipe-rendered advantage in MolmoAct2PackInputsProcessorStep
- Insert RenderMessagesStep pipeline when recipe_path is configured
- Add recap_advantage_molmoact2.yaml recipe
2026-07-03 16:58:12 +02:00
Khalil Meftah 79b7f992b4 feat(annotate): add constant advantage labeling for RECAP SFT phase
- Add constant_value and seed fields to AdvantageConfig
- Implement _run_constant_mode in AdvantageModule with CFG dropout
- Use deterministic seeding (config.seed + episode_index) for reproducibility
2026-07-03 16:58:12 +02:00
Khalil Meftah 04a39d419d feat(pi05): implement Classifier-Free Guidance (CFG) inference
Add dual-path denoising with configurable cfg_beta scale for language-
conditioned action generation. When cfg_beta > 1.0, VLM prefills both
conditioned and unconditional prompts, and action expert velocities are
interpolated via v = v_uncond + β*(v_cond - v_uncond).
2026-07-03 16:58:11 +02:00
Khalil Meftah b63a714ae9 feat(pi05): integrate RenderMessagesStep for advantage conditioning
Add RenderedMessagesToTaskStep adapter that bridges recipe-rendered chat
messages back into PI05's task-string prompt format. When recipe_path is
set on PI05Config, the preprocessor inserts RenderMessagesStep + adapter
before prompt construction, enabling RECAP advantage text to flow
end-to-end through the recipe YAML system.
2026-07-03 16:58:11 +02:00
Khalil Meftah 2ded9ba783 feat(rollout): add episode success labeling to DAgger strategy 2026-07-03 16:58:08 +02:00
Khalil Meftah 194a6379ea feat(recap): add advantage conditioning recipe YAMLs 2026-07-03 16:55:41 +02:00
Khalil Meftah cc782e3589 feat(recap): add advantage scoring annotation module
Implement the RECAP advantage scoring module as a new phase in
lerobot-annotate. Uses a frozen distributional VF to compute per-frame
advantages, binarizes into positive/negative indicators with per-task
threshold, and writes style=advantage persistent rows for policy
conditioning. Skips VF inference on intervention frames as an optimization.
2026-07-03 16:55:40 +02:00
Khalil Meftah b90ccd283b feat(recap): add lerobot-compute-returns script to compute MC returns 2026-07-03 16:55:40 +02:00
Khalil Meftah f8fa8ba394 test(rewards): add unit tests for distributional value function model 2026-07-03 16:55:40 +02:00
Khalil Meftah 6663cac584 feat(rewards): introduce distributional value function model
- Added a new distributional value function (DistributionalVF) model for RECAP, including its configuration, modeling, and processor components.
- Updated the rewards factory to support the new model type.
- Updated  to include the new model in the dependencies.
2026-07-03 16:55:28 +02:00
Khalil Meftah 4af7095693 Merge branch 'main' into feat/rollout/dagger-episode-save 2026-07-03 16:50:10 +02:00
Pepijn e275ea3960 LingBot-VA: video-action world model (#3731)
* feat(policies): add LingBot-VA autoregressive video-action world model

Port the LingBot-VA policy (Wan2.2 dual-stream video+action world model) into
LeRobot, following the EO-1 / VLA-JEPA conventions. Covers inference, checkpoint
conversion, and predicted-video saving (training is deferred to a follow-up PR).

- Vendored Wan transformer/attention/flex/VAE/scheduler modules (key names preserved
  for near-identity conversion); torch SDPA default, flashattn/flex lazy-guarded.
- LingBotVAConfig (registered "lingbot_va") + processor with fixed-quantile action
  unnormalization; full dual-stream sampling loop with CFG, two flow-matching
  schedulers and KV cache, mapped onto select_action with observed-keyframe feedback.
- convert_lingbot_va_checkpoints.py (libero/robotwin variants): bundles the ~5B
  transformer, lazy-pulls the frozen VAE+UMT5 from the source repo.
- Predicted-video plumbing in lerobot_eval (predicted_frames_callback; opt-in via
  --policy.save_predicted_video) and ConstantWithWarmupSchedulerConfig.
- pyproject: widen diffusers-dep to <0.37, add lingbot_va + imageio-dep extras,
  add lingbot_va and (missing) eo1 to `all`.
- Factory + policies/__init__ wiring, docs page + toctree, and tests.

Note: the LIBERO success-rate correctness gate must be validated on a CUDA GPU
with the converted checkpoint.

* feat(lingbot_va): RoboTwin eef-pose eval, single-file model, Hub checkpoints

Make the LingBot-VA port runnable on both LIBERO and RoboTwin and clean up the
package to LeRobot conventions.

- Consolidate all vendored Wan2.2 model code (transformer, attention, VAE helpers,
  flow-matching scheduler, grid utils, flex-attention) into a single
  modeling_lingbot_va.py; remove the separate wan_*/schedulers modules.
- Move the fixed action (un)normalization quantiles out of the config and into the
  post-processor (LIBERO 7-DoF + RoboTwin 16-d eef); remove the conversion script in
  favour of ready-to-use LeRobot-format checkpoints on the Hub.
- Fixes found via on-sim validation: undo LIBERO's 180-degree image flip
  (image_hflip), encode obs as a multi-frame streaming-VAE clip, reset the streaming
  VAE cache between episodes, run the transformer in config.dtype, lazy-load frozen
  VAE/UMT5 by subfolder with the text encoder on CPU.
- RoboTwin: add an end-effector-pose action mode to RoboTwinEnv (16-d per-arm
  xyz+quat+gripper deltas composed onto the initial eef pose, executed via CuRobo IK)
  and the robotwin_tshape latent layout (full-res head + half-res wrists via a second
  streaming VAE) with the upstream RoboTwin action quantiles + camera mapping.
- Predicted-video saving works for both benchmarks; docs + tests updated.

* feat(lingbot_va): implement training / fine-tuning (flow-matching loss)

- Implement LingBotVAPolicy.forward(): dual-stream flow-matching training loss
  (latent + action, timestep-weighted, action-masked) ported from upstream train.py;
  VAE-encodes camera clips, UMT5-encodes the task, noises both streams, runs the
  block-causal flex-attention training pass (forward_train).
- training_loss_from_streams() core + _build_training_streams() data prep (action
  scatter into the 30-d space, multi-frame VAE encode incl. robotwin_tshape).
- get_optim_params returns only trainable transformer params (LoRA/PEFT friendly);
  VAE/UMT5 stay frozen. Training needs attn_mode='flex'.
- Add a tiny-config single-training-step test (forward->loss->backward->AdamW) and a
  Training/fine-tuning section in the docs.

* fix(lingbot_va): CI quality gate + fast-test collection

- Add tests/policies/lingbot_va/__init__.py so the test files don't clash by basename
  with tests/policies/vla_jepa/* under pytest's default import mode (fast-test collection error).
- Fix vendored typos flagged by the typos hook (pach_scale->patch_scale, total_tolen->
  total_token_len, stablized->stabilized) and a mypy union-attr in RoboTwinEnv._read_eef_pose.
- Apply Prettier formatting to docs/source/lingbot_va.mdx.

* docs(lingbot_va): document EEF action-channel schema + camera order

* Update lingbot_va.mdx

Signed-off-by: Pepijn <138571049+pkooij@users.noreply.github.com>

* Update pyproject.toml

Signed-off-by: Pepijn <138571049+pkooij@users.noreply.github.com>

* Update pyproject.toml

Signed-off-by: Pepijn <138571049+pkooij@users.noreply.github.com>

* refactor(lingbot_va): drop hardcoded action quantiles; source from checkpoint

The LIBERO/RoboTwin action (un)normalization quantiles were hardcoded as module
constants in processor_lingbot_va.py. They are already serialized into each
checkpoint's policy_postprocessor.json (via LingBotVAActionUnnormalizeStep.get_config)
and restored on load by PolicyProcessorPipeline.from_pretrained, so the constants are
dead at eval/load time for the released checkpoints (verified: libero_long/robotwin/base
all carry their quantiles on the Hub).

- Remove LIBERO_ACTION_Q01/Q99, ROBOTWIN_ACTION_Q01/Q99 and _default_action_quantiles.
- make_lingbot_va_pre_post_processors now defaults a fresh (unconverted) build to a
  neutral [-1, 1] mapping (identity rescale); real per-benchmark stats come from the
  saved checkpoint (or postprocessor_overrides), analogous to dataset-stats normalization.
- Update the config doc comment to point at the checkpoint as the source of truth.
- Tests: replace the LIBERO-default assertion with a neutral-default check, and add a
  save_pretrained/from_pretrained round-trip guard for the quantile serialization.

* docs(lingbot_va): trim verbose comments

- configuration_lingbot_va.py: condense multi-line field comments to one-liners
  (keep the ── section headers).
- processor_lingbot_va.py: shorten the action-quantile explanation block.
- modeling_lingbot_va.py: drop the bare "# ----" separator rules, keeping the
  one-line section headers.

No code changes.

* docs(lingbot_va): trim provenance comments; default wan path to base repo

- configuration_lingbot_va.py: drop the "──" decorations and the
  "(from transformer/config.json)" note; default wan_pretrained_path to
  robbyant/lingbot-va-base (has the frozen vae/text_encoder/tokenizer subfolders).
- modeling_lingbot_va.py: remove the vendored-code banner and the
  "(upstream wan_va/...)" section-header provenance/dash decorations; condense the
  transformer-dtype comment to one line.

No code changes.

* refactor(lingbot_va): use built-in UnnormalizerProcessorStep for actions

Replace the bespoke LingBotVAActionUnnormalizeStep with the standard
UnnormalizerProcessorStep in QUANTILES mode, which computes the identical
(action + 1) / 2 * (q99 - q01) + q01 mapping. The per-channel q01/q99 are stored
as the step's saved state (a safetensors file) and restored on load; a fresh build
has no action stats so the step is an identity passthrough.

The 3 Hub checkpoints (lerobot/lingbot_va_{libero_long,robotwin,base}) have been
re-uploaded with the new post-processor (policy_postprocessor.json +
*_unnormalizer_processor.safetensors); reloading from the Hub round-trips q01/q99.

- processor_lingbot_va.py: drop the custom step + registry; build the post-processor
  with UnnormalizerProcessorStep (explicit ACTION->QUANTILES norm_map so the
  preprocessor / training path is unchanged).
- tests: assert the built-in step is used, identity-when-no-stats, correct quantile
  unnormalization, and a save_pretrained/from_pretrained stats round-trip.

* docs(lingbot_va): point checkpoint paths at the lerobot org

The LeRobot-format checkpoints moved from pepijn223/* to lerobot/* (libero_long,
robotwin, base). Update the eval/train --policy.path examples accordingly.

* docs(lingbot_va): condense processor normalization comments

* fix(lingbot-va): align RoboTwin evaluation (#3784)

Thank you for the RoboTwin fix, and alignment!

* applying fixes

* updating uv lock and linting

* adjusting test to match expected values

* cleaning up deps

* cleaning up top level imports, styling, and deps guards

* cleanup
* moving wan utils and loading utils to `utils.py`
* removing ftfy by replicating the prompt_clean function without it (we don't expect to have weird chars given in the prompt anyway)

* removing unused function

* guarding for scipy dep, renaming test to avoid collision

* adding back accelerate for peak memory usage optim + justifying robotwin description dep

---------

Signed-off-by: Pepijn <138571049+pkooij@users.noreply.github.com>
Co-authored-by: pepijn223 <pepijn223@hf.co>
Co-authored-by: Gangwei XU <gwxu@hust.edu.cn>
Co-authored-by: Maxime Ellerbach <maxime.ellerbach@huggingface.co>
2026-07-03 13:32:38 +02:00
Nikodem Bartnik 911734ec9c Docs/improve HF jobs documentation (#3909)
* improve hf jobs docs

* Update docs/source/hardware_guide.mdx

Co-authored-by: Nicolas Rabault <rabault.nicolas@gmail.com>
Signed-off-by: Nikodem Bartnik <39432165+NikodemBartnik@users.noreply.github.com>

---------

Signed-off-by: Nikodem Bartnik <39432165+NikodemBartnik@users.noreply.github.com>
Co-authored-by: Nicolas Rabault <rabault.nicolas@gmail.com>
2026-07-03 11:39:16 +02:00
Khalil Meftah 46d4ddc698 chore(rollout): log episode success label and buffer length 2026-07-02 19:12:10 +02:00
Khalil Meftah b29ba27977 fix(rollout): guard empty buffer save 2026-07-02 18:02:59 +02:00
Khalil Meftah 599e2432e5 fix(rollout): clear last_action after return_to_initial 2026-07-02 18:02:36 +02:00
Khalil Meftah 44f76dbbf0 feat(rollout): add episode success/failure labeling to DAgger strategy
Enable operators to mark episodes as success or failure during DAgger
data collection. Pressing 's' or 'f' immediately saves the episode
with the appropriate label and returns the robot to its initial position.

- Add success/failure key bindings to DAggerKeyboardConfig
- Add save_episode_requested event and episode_success state to DAggerEvents
- Stamp next.success=True on terminal frame for successful episodes
- Pause and return to initial position after manual save for env reset
- Add num_episodes target to stop continuous recording automatically
- Defer save during corrections to avoid splitting mid-intervention
2026-07-02 17:48:02 +02:00
61 changed files with 8786 additions and 1663 deletions
+4
View File
@@ -22,6 +22,10 @@ outputs
rl
media
# Local virtualenvs (the image provides its own)
.venv
venv
# Logging
logs
+2
View File
@@ -69,6 +69,8 @@
title: VLA-JEPA
- local: eo1
title: EO-1
- local: lingbot_va
title: LingBot-VA
- local: fastwam
title: FastWAM
- local: groot
+9 -9
View File
@@ -82,18 +82,18 @@ VRAM is the first filter. Within a tier, pick by budget and availability — the
### Hugging Face Jobs
[Hugging Face Jobs](https://huggingface.co/docs/hub/jobs) lets you run training on managed HF infrastructure, billed by the second. The repo publishes a ready-to-use image: **`huggingface/lerobot-gpu:latest`**, rebuilt **every night at 02:00 UTC from `main`** ([`docker_publish.yml`](https://github.com/huggingface/lerobot/blob/main/.github/workflows/docker_publish.yml)) — so it tracks the current state of the repo, not a tagged release.
[Hugging Face Jobs](https://huggingface.co/docs/hub/jobs) lets you run training on managed HF infrastructure, billed by the second, without owning a GPU. `lerobot-train` submits and streams the job for you — just add `--job.target=<flavor>` to a normal training command:
```bash
hf jobs run --flavor a10g-large huggingface/lerobot-gpu:latest \
bash -c "nvidia-smi && lerobot-train \
--policy.type=act --dataset.repo_id=<USER>/<DATASET> \
--policy.repo_id=<USER>/act_<task> --batch_size=8 --steps=50000"
lerobot-train \
--policy.type=act --dataset.repo_id=<USER>/<DATASET> \
--policy.repo_id=<USER>/act_<task> \
--job.target=a10g-large
```
Notes:
- The leading `nvidia-smi` is a quick sanity check that CUDA is visible inside the container — useful to fail fast if the flavor or driver mismatched.
- The default Job timeout is 30 minutes; pass `--timeout 4h` (or longer) for real training.
- `--flavor` maps onto the table above: `t4-small`/`t4-medium` (T4, ACT only), `l4x1`/`l4x4` (L4 24 GB), `a10g-small/large/largex2/largex4` (A10G 24 GB scaled out), `a100-large` (A100). For the current full catalogue + pricing see [https://huggingface.co/docs/hub/jobs](https://huggingface.co/docs/hub/jobs).
- Prefer not to write the `hf jobs run` wrapper yourself? `lerobot-train` can submit the job for you: just add `--job.target=<flavor>` to a normal training command and it handles dataset upload, log streaming, and the final model push. See the [imitation-learning training guide](./il_robots).
- Run `hf auth login` once before submitting, the job runs under your token.
- `--job.target` maps onto the table above: `t4-small`/`t4-medium` (T4, ACT only), `l4x1`/`l4x4` (L4 24 GB), `a10g-small/large/largex2/largex4` (A10G 24 GB scaled out), `a100-large` (A100). List the current catalogue with pricing via `hf jobs hardware`, or see [https://huggingface.co/docs/hub/jobs](https://huggingface.co/docs/hub/jobs).
- The job defaults to a `2d` (48h) timeout. Override it with `--job.timeout=4h` (or any other valid duration string) to shorten or extend the timeout. The job automatically stops when the command completes.
- For the full walkthrough — dataset upload, checkpoint streaming, resuming a run on a job — see the [imitation-learning training guide](./il_robots#train-using-hugging-face-jobs).
+1 -78
View File
@@ -532,84 +532,7 @@ If your local computer doesn't have a powerful GPU you could utilize Google Cola
Hugging Face jobs let's you easily select hardware and run the training in the cloud. So if you don't have a powerful GPU or you need more VRAM or just want to train a model much faster use HF Jobs! It's pay as you go and you simply pay for each second of use, you can see the pricing and additional information [here](https://huggingface.co/docs/hub/jobs).
> **Tip:** if you just want to launch a standard training run, you can skip building the command below and use the integrated **Train on HF Jobs via `--job.target`** flow described further down — `lerobot-train` then submits the job, uploads a local-only dataset for you, and streams the logs.
To run the training manually use this command:
<hfoptions id="train_with_hf_jobs">
<hfoption id="Command">
```bash
hf jobs run \
--flavor a10g-small \
--timeout 4h \
--secrets HF_TOKEN \
huggingface/lerobot-gpu:latest \
-- \
python -m lerobot.scripts.lerobot_train \
--dataset.repo_id=username/dataset \
--policy.type=act \
--steps=5000 \
--batch_size=16 \
--policy.device=cuda \
--policy.repo_id=username/your_policy \
--log_freq=100
```
</hfoption>
<hfoption id="API example">
<!-- prettier-ignore-start -->
```python
from huggingface_hub import run_job, get_token
run_name = "act_so101_hf_jobs"
dataset_id = "username/dataset"
user_hub_id = "username"
command_args = [
"python", "-m", "lerobot.scripts.lerobot_train",
"--dataset.repo_id", dataset_id,
"--policy.type", "act",
"--steps", "5000",
"--batch_size", "16",
"--num_workers", "4",
"--policy.device", "cuda",
"--log_freq", "100",
"--save_freq", "1000",
"--save_checkpoint", "true",
"--wandb.enable", "false",
"--policy.repo_id", f"{user_hub_id}/{run_name}"
]
print(f"Submitting job '{run_name}' to Hugging Face Infrastructure...")
job_info = run_job(
image="huggingface/lerobot-gpu:latest",
command=command_args,
flavor="a10g-small",
timeout="4h",
secrets={"HF_TOKEN": get_token()}
)
print("\n🚀 Job successfully launched!")
print(f"🔹 Job ID: {job_info.id}")
print(f"🔗 Live UI Dashboard & Logs: {job_info.url}")
```
<!-- prettier-ignore-end -->
</hfoption>
</hfoptions>
You can modify the `--flavor` to use different hardware, for example: `t4-small`, `a100-large`, `h200`. Use `hf jobs hardware` to see the full list with pricing.
Depending on the model you want to train and the hardware you selected you can also modify the `--batch_size` and `--number_of_workers`.
For longer training sessions increase the timeout.
Once the training is started you can go to [Jobs](https://huggingface.co/settings/jobs) and see if your jobs is running as well as all the outputs. Sometimes it takes a few minutes to schedule your job so be patient.
After training the model will be pushed to hub and you can use it as any other model with LeRobot.
#### Train on HF Jobs via `--job.target` (integrated CLI)
`lerobot-train` runs locally by default. To run on a HuggingFace GPU without constructing the Docker command yourself, pass `--job.target` with a hardware flavor name:
`lerobot-train` runs locally by default. To run on a HuggingFace GPU, pass `--job.target` with a hardware flavor name:
```bash
lerobot-train \
+187
View File
@@ -0,0 +1,187 @@
# LingBot-VA
LingBot-VA is an **autoregressive video-action world-model policy** built on the **Wan2.2**
video-diffusion stack. It interleaves, in one autoregressive sequence, the prediction of
future **video latents** and **robot actions** ("VA" = Video-Action). The LeRobot
integration wires LingBot-VA into the standard training, evaluation and processor
interfaces.
## Model Overview
LingBot-VA is a **dual-stream "mixture-of-transformers"**: a video/latent stream
(`patch_embedding_mlp → blocks → proj_out`) and an action stream
(`action_embedder → blocks → action_proj_out`) share the same 30 transformer blocks and
text conditioning.
| Component | Class | Role |
| ------------------------ | ----------------------- | ----------------------------------------------------------- |
| DiT backbone (trainable) | `WanTransformer3DModel` | ~5B-param dual-stream transformer. |
| VAE (frozen) | `AutoencoderKLWan` | Wan2.2 VAE, `z_dim=48`. Lazy-pulled from the source repo. |
| Text encoder (frozen) | `UMT5EncoderModel` | UMT5-XXL, `d_model=4096`. Lazy-pulled from the source repo. |
At inference the policy runs an autoregressive loop per chunk: it denoises the video-latent
stream (CFG, ~20 steps) and the action stream (~50 steps) with two independent
flow-matching schedulers, maintaining a KV cache across chunks. Real observed keyframes are
fed back into the KV cache as the chunk is executed (closed-loop world modeling).
### What the LeRobot Integration Covers
- Standard `policy.type=lingbot_va` configuration through LeRobot.
- Ready-to-use LeRobot-format checkpoints on the Hub (converted from the released upstream ones).
- Autoregressive dual-stream inference behind the standard `select_action` interface
(single-environment eval, `--eval.batch_size=1`).
- Opt-in saving of the policy's **predicted (imagined) videos** during eval / training.
- Evaluation with `lerobot-eval` on LIBERO and RoboTwin.
- Training / fine-tuning via the dual-stream flow-matching loss (`policy.forward`), see below.
## Installation
1. Install LeRobot by following the [Installation Guide](./installation).
2. Install the LingBot-VA extra:
```bash
pip install -e ".[lingbot_va]"
```
## Checkpoints
The released upstream checkpoints have been converted to LeRobot format and pushed to the Hub:
| Variant | LeRobot checkpoint |
| ---------------------- | -------------------------------- |
| LIBERO-Long post-train | `lerobot/lingbot_va_libero_long` |
| RoboTwin post-train | `lerobot/lingbot_va_robotwin` |
| Pretrained base | `lerobot/lingbot_va_base` |
Only the trainable ~5B transformer is stored in the LeRobot
`model.safetensors`. The frozen VAE + UMT5 + tokenizer (~20 GB) are pulled from
`config.wan_pretrained_path` at load time (defaults to the source `robbyant/*` repo). The
UMT5-XXL text encoder runs on CPU by default (`config.text_encoder_device`) so the 5B
transformer + VAE fit on a single 2432 GB GPU.
## Evaluation (LIBERO)
```bash
lerobot-eval \
--policy.path=lerobot/lingbot_va_libero_long \
--policy.device=cuda \
--env.type=libero --env.task=libero_10 \
--env.observation_height=128 --env.observation_width=128 \
--eval.n_episodes=50 --eval.batch_size=1 \
--output_dir=outputs/eval/lingbot_va_libero
```
LingBot-VA's streaming inference (KV cache + observed-keyframe feedback) is implemented for
single-environment eval; use `--eval.batch_size=1`.
## Evaluation (RoboTwin)
RoboTwin 2.0 needs the SAPIEN + CuRobo simulator stack. You can use the benchmark Docker image
(`docker/Dockerfile.benchmark.robotwin`, which also needs `warp-lang==1.3.1` and CuRobo built
with the GPU's compute capability in `TORCH_CUDA_ARCH_LIST`). RoboTwin uses **end-effector-pose
control**, so run with `--env.action_mode=ee`: the policy predicts per-arm `xyz+quaternion+gripper`
deltas (`robotwin_tshape` latent layout) that are composed onto the episode's initial eef pose and
executed via CuRobo IK.
```bash
lerobot-eval \
--policy.path=lerobot/lingbot_va_robotwin \
--policy.device=cuda \
--env.type=robotwin --env.task=beat_block_hammer --env.action_mode=ee \
--eval.n_episodes=10 --eval.batch_size=1 \
--output_dir=outputs/eval/lingbot_va_robotwin
```
### Saving predicted (imagined) videos
Set `--policy.save_predicted_video=true` to additionally VAE-decode the predicted video
latents and write `pred_episode_*.mp4` next to the env-rendered `eval_episode_*.mp4` videos.
The same flag works for the periodic eval during `lerobot-train`.
## Training / fine-tuning
`LingBotVAPolicy.forward(batch)` implements the dual-stream **flow-matching** loss
(`latent_loss + action_loss`, timestep-weighted, action-masked) from the paper: it VAE-encodes
the camera clips into video latents, UMT5-encodes the task, noises both streams, runs the
transformer's block-causal training pass and returns `(loss, metrics)`. Optimizer preset is AdamW
with a linear-warmup-then-constant schedule (matching upstream).
Requirements:
- The block-causal masks use PyTorch **flex-attention**, so build the policy with
`--policy.attn_mode=flex` for training (the default `torch` SDPA is inference-only).
- The full 5B DiT does not fit a single 2432 GB GPU under AdamW; fine-tune with **LoRA**
(`--policy.use_peft=true`) and/or optimizer offload. `get_optim_params` returns only the
trainable (e.g. adapter) parameters; the VAE + UMT5 text encoder stay frozen.
```bash
lerobot-train \
--policy.path=lerobot/lingbot_va_libero_long --policy.attn_mode=flex \
--policy.use_peft=true \
--dataset.repo_id=<your LeRobot-format dataset> \
--batch_size=1 --steps=... --output_dir=outputs/train/lingbot_va
```
The dataset must provide camera clips (a temporal window per camera, VAE-encoded to
`frame_chunk_size` latent frames) and `frame_chunk_size * action_per_frame` action steps per item.
## Data format (action channels & camera order)
LingBot-VA is an **end-effector (Cartesian) pose** policy, it predicts EEF poses + gripper, not
joint positions. Actions live in a fixed multi-embodiment **30-dim** layout; map your robot's
action dimensions into these channels and pad the rest with `0` (`used_action_channel_ids` selects
the channels a given checkpoint actually uses):
| channels | meaning |
| -------- | ----------------------------------------------------- |
| 06 | Left-arm end-effector pose |
| 713 | Right-arm end-effector pose |
| 1420 | Left-arm joints (unused by the released checkpoints) |
| 2127 | Right-arm joints (unused by the released checkpoints) |
| 28 | Left gripper |
| 29 | Right gripper |
- **LIBERO** uses channels `06`: a 6-DoF EEF delta (xyz + rotation) + gripper (single arm).
- **RoboTwin** uses channels `[06, 28, 713, 29]`: left EEF (xyz + quaternion) + left gripper +
right EEF + right gripper (16 dims). The env converts these poses to joint trajectories via
CuRobo IK — joints are never predicted.
Joint-space datasets (or a different EEF convention) must be remapped into this schema before
fine-tuning these checkpoints.
**Camera order is fixed and order-sensitive**, per-camera latents are concatenated spatially in
`obs_cam_keys` order, so the physical camera→slot mapping must match training:
| benchmark | `obs_cam_keys` (in order) | `camera_layout` |
| --------- | ----------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------- |
| LIBERO | `observation.images.image` (agentview / 3rd-person), `observation.images.image2` (eye-in-hand wrist) | `width_concat` (latents concatenated on width) |
| RoboTwin | `observation.images.head_camera`, `observation.images.left_camera`, `observation.images.right_camera` | `robotwin_tshape` (full-res head below, two half-res wrists on top) |
The first camera is the exterior/head view and the rest are wrist views.
## Inference Hyperparameters (LIBERO)
| Key | Value |
| -------------------------------------- | --------------------------------------------------------------------------------- |
| height × width | 128 × 128 |
| cameras | `observation.images.image` (agentview), `observation.images.image2` (eye-in-hand) |
| action channels used | 06 (7-DoF arm + gripper) |
| action_per_frame / frame_chunk_size | 4 / 4 |
| attn_window | 30 |
| video / action denoising steps | 20 / 50 |
| guidance_scale / action_guidance_scale | 5 / 1 |
| snr_shift / action_snr_shift | 5.0 / 0.05 |
These are the defaults of `LingBotVAConfig`; override any of them via `--policy.<name>=...`.
## Notes
- **Attention backend:** inference uses the `torch` SDPA backend (always available). The
`flashattn` and `flex` backends are optional; `flex` is only needed for training.
- **Model size:** the DiT is ~5B params and the frozen VAE+UMT5 add ~20 GB; inference needs
roughly 1824 GB of VRAM.
## License
LingBot-VA is released under Apache-2.0. See the
[upstream repository](https://github.com/Robbyant/lingbot-va).
+5
View File
@@ -228,6 +228,7 @@ groot = [
sarm = ["lerobot[transformers-dep]", "pydantic>=2.0.0,<3.0.0", "faker>=33.0.0,<35.0.0", "lerobot[matplotlib-dep]", "lerobot[qwen-vl-utils-dep]"]
robometer = ["lerobot[transformers-dep]", "lerobot[qwen-vl-utils-dep]", "lerobot[peft-dep]"]
topreward = ["lerobot[transformers-dep]"]
recap = ["lerobot[transformers-dep]"]
xvla = ["lerobot[transformers-dep]"]
eo1 = ["lerobot[transformers-dep]", "lerobot[qwen-vl-utils-dep]"]
fastwam = [
@@ -236,6 +237,7 @@ fastwam = [
]
hilserl = ["lerobot[transformers-dep]", "lerobot[dataset]", "gym-hil>=0.1.14,<0.2.0", "lerobot[grpcio-dep]", "lerobot[placo-dep]"]
vla_jepa = ["lerobot[transformers-dep]", "lerobot[diffusers-dep]", "lerobot[qwen-vl-utils-dep]"]
lingbot_va = ["lerobot[transformers-dep]", "lerobot[diffusers-dep]", "lerobot[accelerate-dep]"]
# Features
async = ["lerobot[grpcio-dep]", "lerobot[matplotlib-dep]"]
@@ -318,6 +320,7 @@ all = [
"lerobot[xvla]",
"lerobot[hilserl]",
"lerobot[vla_jepa]",
"lerobot[lingbot_va]",
"lerobot[async]",
"lerobot[dev]",
"lerobot[test]",
@@ -330,6 +333,7 @@ all = [
"lerobot[sarm]",
"lerobot[robometer]",
"lerobot[topreward]",
"lerobot[recap]",
"lerobot[peft]",
# "lerobot[unitree_g1]", TODO: Unitree requires specific installation instructions for unitree_sdk2
]
@@ -353,6 +357,7 @@ lerobot-edit-dataset="lerobot.scripts.lerobot_edit_dataset:main"
lerobot-setup-can="lerobot.scripts.lerobot_setup_can:main"
lerobot-annotate="lerobot.scripts.lerobot_annotate:main"
lerobot-rollout="lerobot.scripts.lerobot_rollout:main"
lerobot-compute-returns="lerobot.scripts.lerobot_compute_returns:main"
# ---------------- Tool Configurations ----------------
@@ -169,6 +169,51 @@ class ExecutorConfig:
episode_parallelism: int = 16
@dataclass
class AdvantageConfig:
"""``advantage`` module: RECAP advantage scoring via frozen value function."""
enabled: bool = True
# Constant advantage label for all frames (e.g. "positive" for SFT iteration 0).
# Skips VF inference, dropout still applies for CFG.
constant_value: str | None = None
# Trained value function checkpoint (local path or Hub repo ID).
# Ignored when constant_value is set.
value_function_path: str = ""
# Device to run the value function on.
device: str = "cuda"
# N-step lookahead for advantage estimation.
# None = MC (N=T): A_t = R_t - V(s_t), using mc_return from dataset.
# 50 = fine-tuning mode: A_t = Σ r_{t:t+N} + V(s_{t+N}) - V(s_t).
n_step: int | None = None
# Per-task percentile for binarization threshold ε_.
# Actions with advantage > ε_ get I_t = True (positive).
threshold_percentile: float = 0.3
# Fraction of frames to randomly omit advantage labels (enables CFG).
dropout_rate: float = 0.3
# Force I_t = True for frames marked as human interventions.
force_positive_on_intervention: bool = True
# Column name in dataset for intervention flag.
intervention_key: str = "intervention"
# Column name for pre-computed MC returns (from lerobot-compute-returns).
mc_return_key: str = "mc_return"
# Batch size for value function inference.
batch_size: int = 32
# Random seed for dropout reproducibility.
seed: int = 1729
@dataclass
class AnnotationPipelineConfig:
"""Top-level config for ``lerobot-annotate`` (rewrites data shards in place)."""
@@ -190,6 +235,7 @@ class AnnotationPipelineConfig:
plan: PlanConfig = field(default_factory=PlanConfig)
interjections: InterjectionsConfig = field(default_factory=InterjectionsConfig)
vqa: VqaConfig = field(default_factory=VqaConfig)
advantage: AdvantageConfig = field(default_factory=AdvantageConfig)
vlm: VlmConfig = field(default_factory=VlmConfig)
executor: ExecutorConfig = field(default_factory=ExecutorConfig)
@@ -15,20 +15,24 @@
# limitations under the License.
"""In-process executor that runs the annotation phases.
The executor runs **six phases** in dependency order:
The executor runs **seven phases** in dependency order:
phase 1: ``plan`` module (plan + subtasks + memory)
phase 2: ``interjections`` module (interjections + speech)
phase 3: ``plan`` plan-update pass — re-runs plan emission at every
interjection timestamp produced by phase 2
phase 4: ``vqa`` module (VQA)
phase 5: validator
phase 6: writer
phase 5: ``advantage`` module (advantage scoring via frozen VF)
phase 6: validator
phase 7: writer
Phase 3 is why the ``plan`` module must be re-entered after the
``interjections`` module — to refresh ``plan`` rows at interjection
timestamps.
Phase 5 (advantage) does not depend on the VLM modules, it uses a frozen
distributional value function to compute per-frame advantage indicators.
Distributed execution is provided by Hugging Face Jobs (see
``examples/annotations/run_hf_job.py``); the runner inside the job
invokes ``lerobot-annotate`` which uses this in-process executor.
@@ -74,7 +78,7 @@ class PipelineRunSummary:
@dataclass
class Executor:
"""Run all six phases over a dataset root in-process.
"""Run all seven phases over a dataset root in-process.
Episode-level concurrency comes from ``ExecutorConfig.episode_parallelism``
(a thread pool); cluster-level concurrency comes from running this
@@ -86,6 +90,7 @@ class Executor:
plan: Any # PlanSubtasksMemoryModule
interjections: Any # InterjectionsAndSpeechModule
vqa: Any # GeneralVqaModule
advantage: Any # AdvantageModule
writer: LanguageColumnsWriter
validator: StagingValidator
@@ -112,6 +117,8 @@ class Executor:
phases.append(self._run_plan_update_phase(records, staging_dir))
# Phase 4: ``vqa`` module (VQA)
phases.append(self._run_module_phase("vqa", records, staging_dir, self.vqa))
# Phase 5: ``advantage`` module (advantage scoring via frozen VF)
phases.append(self._run_module_phase("advantage", records, staging_dir, self.advantage))
print("[annotate] running validator...", flush=True)
report = self.validator.validate(records, staging_dir)
@@ -179,7 +186,7 @@ class Executor:
staging_dir: Path,
module: Any,
) -> PhaseResult:
if not module.enabled:
if module is None or not module.enabled:
print(f"[annotate] phase={name} skipped (module disabled)", flush=True)
return PhaseResult(name=name, episodes_processed=0, episodes_skipped=len(records))
n = len(records)
@@ -231,7 +238,7 @@ class Executor:
``plan`` module with the interjection timestamps so its existing
prompt path is reused.
"""
if not self.plan.enabled or not self.interjections.enabled:
if not self.plan or not self.plan.enabled or not self.interjections or not self.interjections.enabled:
return PhaseResult(name="plan_update", episodes_processed=0, episodes_skipped=len(records))
processed = 0
for record in records:
@@ -14,11 +14,13 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from .advantage import AdvantageModule
from .general_vqa import GeneralVqaModule
from .interjections_and_speech import InterjectionsAndSpeechModule
from .plan_subtasks_memory import PlanSubtasksMemoryModule
__all__ = [
"AdvantageModule",
"GeneralVqaModule",
"InterjectionsAndSpeechModule",
"PlanSubtasksMemoryModule",
@@ -0,0 +1,298 @@
#!/usr/bin/env python
# Copyright 2026 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.
"""Advantage scoring module for RECAP.
Computes per-frame advantage values using a frozen distributional value function,
binarizes them into improvement indicators (I_t), and emits ``style="advantage"``
persistent rows for policy conditioning.
Paper reference: pi*0.6, Section IV-B and Appendix F.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from typing import Any
import numpy as np
import torch
from ..config import AdvantageConfig
from ..reader import EpisodeRecord
from ..staging import EpisodeStaging
logger = logging.getLogger(__name__)
@dataclass
class AdvantageModule:
"""Compute advantage indicators and emit persistent annotation rows.
The module loads a frozen distributional value function and scores each
frame in an episode. Advantages are binarized into ``positive``/``negative``
indicators using a per-task threshold, then written as ``style="advantage"``
persistent rows into the staging area.
Requires ``mc_return`` column in the dataset (from lerobot-compute-returns).
"""
config: AdvantageConfig
_model: Any = field(default=None, init=False, repr=False)
_preprocessor: Any = field(default=None, init=False, repr=False)
_threshold: float | None = field(default=None, init=False, repr=False)
@property
def enabled(self) -> bool:
return self.config.enabled
def _ensure_model_loaded(self) -> None:
"""Lazy-load the frozen value function on first use."""
if self._model is not None:
return
from lerobot.rewards import (
make_reward_model,
make_reward_model_config,
make_reward_pre_post_processors,
)
cfg = make_reward_model_config(
"distributional_value_function",
pretrained_path=self.config.value_function_path,
device=self.config.device,
)
self._model = make_reward_model(cfg)
self._model.eval()
for p in self._model.parameters():
p.requires_grad_(False)
self._preprocessor, _ = make_reward_pre_post_processors(cfg)
logger.info("Loaded frozen VF from %s on %s", self.config.value_function_path, self.config.device)
def compute_advantages_for_episode(self, record: EpisodeRecord) -> tuple[np.ndarray, np.ndarray]:
"""Compute raw advantage values for all frames in an episode.
Returns:
(advantages, intervention_mask) both shape [num_frames].
advantages[t] = A_t, intervention_mask[t] = True if frame is intervention.
"""
self._ensure_model_loaded()
df = record.frames_df()
num_frames = len(df)
mc_return_key = self.config.mc_return_key
if mc_return_key not in df.columns:
raise KeyError(
f"Column '{mc_return_key}' not found in episode {record.episode_index}. "
"Run lerobot-compute-returns first."
)
mc_returns = df[mc_return_key].values.astype(np.float32)
intervention_mask = np.zeros(num_frames, dtype=bool)
if self.config.intervention_key in df.columns:
intervention_mask = df[self.config.intervention_key].values.astype(bool)
# Skip VF inference on intervention frames — they're always "positive"
# regardless of advantage value, so V(s_t) is never used for them.
skip_mask = intervention_mask if self.config.force_positive_on_intervention else None
values = self._compute_values(record, skip_mask=skip_mask)
if self.config.n_step is None:
advantages = mc_returns - values
else:
advantages = self._compute_n_step_advantages(mc_returns, values, record, n=self.config.n_step)
return advantages, intervention_mask
def _compute_values(self, record: EpisodeRecord, skip_mask: np.ndarray | None = None) -> np.ndarray:
"""Run frozen VF over all frames to get V(s_t) predictions.
Args:
record: Episode data.
skip_mask: Optional boolean mask [num_frames]. Frames where True are
skipped (left as 0.0) to avoid unnecessary inference.
"""
df = record.frames_df()
num_frames = len(df)
values = np.zeros(num_frames, dtype=np.float32)
image_key = self._resolve_image_key(df)
if image_key is None:
logger.warning("No image key found for episode %d; returning zero values.", record.episode_index)
return values
# Determine which frame indices actually need inference
infer_indices = np.where(~skip_mask)[0] if skip_mask is not None else np.arange(num_frames)
if len(infer_indices) == 0:
return values
task_text = record.episode_task
for batch_start in range(0, len(infer_indices), self.config.batch_size):
batch_end = min(batch_start + self.config.batch_size, len(infer_indices))
batch_indices = infer_indices[batch_start:batch_end]
batch_images = []
for idx in batch_indices:
img_val = df.iloc[idx][image_key]
if isinstance(img_val, np.ndarray):
img_tensor = torch.from_numpy(img_val).float()
elif isinstance(img_val, torch.Tensor):
img_tensor = img_val.float()
else:
img_tensor = torch.zeros(3, 224, 224)
batch_images.append(img_tensor)
batch_images_tensor = torch.stack(batch_images)
batch_size = batch_images_tensor.shape[0]
raw_batch = {
image_key: batch_images_tensor,
"task": [task_text] * batch_size,
}
processed = self._preprocessor(raw_batch)
with torch.no_grad():
v_values = self._model.compute_reward(processed)
values[batch_indices] = v_values.cpu().numpy()
return values
def _compute_n_step_advantages(
self, mc_returns: np.ndarray, values: np.ndarray, record: EpisodeRecord, n: int
) -> np.ndarray:
"""Compute N-step advantage: A_t = Σ r_{t:t+N-1} + V(s_{t+N}) - V(s_t).
When t+N exceeds episode length, truncates to MC (uses mc_return directly).
"""
num_frames = len(values)
advantages = np.zeros(num_frames, dtype=np.float32)
for t in range(num_frames):
if t + n >= num_frames:
advantages[t] = mc_returns[t] - values[t]
else:
n_step_return = mc_returns[t] - mc_returns[t + n]
advantages[t] = n_step_return + values[t + n] - values[t]
return advantages
def _resolve_image_key(self, df) -> str | None:
"""Find the first image observation key in the dataframe columns."""
for col in df.columns:
if col.startswith("observation.images."):
return col
return None
def run_episode(self, record: EpisodeRecord, staging: EpisodeStaging) -> None:
"""Score one episode and write advantage rows to staging."""
if self.config.constant_value:
self._run_constant_mode(record, staging)
return
if not self.config.value_function_path:
logger.warning("No value_function_path or constant_value configured; skipping advantage scoring.")
return
advantages, intervention_mask = self.compute_advantages_for_episode(record)
num_frames = len(advantages)
threshold = self._compute_threshold(advantages, intervention_mask)
rng = np.random.default_rng(seed=self.config.seed + record.episode_index)
rows: list[dict[str, Any]] = []
for t in range(num_frames):
if rng.random() < self.config.dropout_rate:
continue
if (
self.config.force_positive_on_intervention
and intervention_mask[t]
or advantages[t] > threshold
):
indicator = "positive"
else:
indicator = "negative"
timestamp = float(record.frame_timestamps[t]) if t < len(record.frame_timestamps) else 0.0
rows.append(
{
"role": "user",
"content": indicator,
"style": "advantage",
"timestamp": timestamp,
"camera": None,
"tool_calls": None,
}
)
staging.write("advantage", rows)
logger.debug(
"Episode %d: %d/%d frames scored (threshold=%.4f, %d positive, %d negative)",
record.episode_index,
len(rows),
num_frames,
threshold,
sum(1 for r in rows if r["content"] == "positive"),
sum(1 for r in rows if r["content"] == "negative"),
)
def _run_constant_mode(self, record: EpisodeRecord, staging: EpisodeStaging) -> None:
"""Emit a fixed advantage value for every frame (with dropout for CFG)."""
num_frames = len(record.frame_timestamps)
rng = np.random.default_rng(seed=self.config.seed + record.episode_index)
rows: list[dict[str, Any]] = []
for t in range(num_frames):
if rng.random() < self.config.dropout_rate:
continue
rows.append(
{
"role": "user",
"content": self.config.constant_value,
"style": "advantage",
"timestamp": float(record.frame_timestamps[t]),
"camera": None,
"tool_calls": None,
}
)
staging.write("advantage", rows)
logger.debug(
"Episode %d: %d/%d frames labeled constant '%s' (dropout=%.2f)",
record.episode_index,
len(rows),
num_frames,
self.config.constant_value,
self.config.dropout_rate,
)
def _compute_threshold(self, advantages: np.ndarray, intervention_mask: np.ndarray) -> float:
"""Compute the binarization threshold as the configured percentile of advantages."""
non_intervention = advantages[~intervention_mask] if intervention_mask.any() else advantages
if len(non_intervention) == 0:
return 0.0
return float(np.percentile(non_intervention, self.config.threshold_percentile * 100))
@@ -39,6 +39,7 @@ _MODULES: tuple[ModuleName, ...] = (
"plan",
"interjections",
"vqa",
"advantage",
)
+1
View File
@@ -32,6 +32,7 @@ DEFAULT_BINDINGS = {
"interjection": "emitted_at(t, style=interjection)",
"vqa": "emitted_at(t, style=vqa, role=assistant)",
"vqa_query": "emitted_at(t, style=vqa, role=user)",
"advantage": "active_at(t, style=advantage)",
}
PLACEHOLDER_RE = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
@@ -0,0 +1,30 @@
# RECAP advantage-conditioned recipe.
#
# Composes task + advantage indicator into the prompt for conditional SFT.
# The advantage binding resolves to "positive" or "negative" from the
# language_persistent column (written by lerobot-annotate --advantage).
# When advantage is absent (30% dropout), the advantage turn is skipped
# entirely via if_present, training the unconditional branch for CFG.
#
# This recipe is policy-agnostic: any VLA that consumes chat-style messages
# can use it. Override bindings or add blend components for task-specific needs.
#
# Paper: pi*0.6, Section IV-B (conditional policy training with I_t).
bindings:
advantage: "active_at(t, style=advantage)"
messages:
- role: user
content: "${task}"
stream: high_level
- role: user
content: "Advantage: ${advantage}"
stream: high_level
if_present: advantage
- role: assistant
content: "${subtask}"
stream: low_level
target: true
@@ -0,0 +1,41 @@
# RECAP full recipe with advantage conditioning and subtask blending.
#
# Blend of two training modes:
# 1. advantage_conditioned (70%): Task + advantage indicator → action
# 2. unconditional (30%): Task only → action (no advantage, trains CFG baseline)
#
# This achieves the same effect as per-frame dropout in the annotation module
# but at the recipe level, giving explicit control over the conditioning ratio.
# Use this instead of annotation-level dropout if you want a fixed split.
#
# Paper: pi*0.6, Appendix E (classifier-free guidance requires both branches).
blend:
advantage_conditioned:
weight: 0.7
messages:
- role: user
content: "${task}\nAdvantage: ${advantage}"
stream: high_level
if_present: advantage
- role: user
content: "${task}"
stream: high_level
- role: assistant
content: "${subtask}"
stream: low_level
target: true
unconditional:
weight: 0.3
messages:
- role: user
content: "${task}"
stream: high_level
- role: assistant
content: "${subtask}"
stream: low_level
target: true
@@ -0,0 +1,28 @@
# RECAP advantage recipe for MolmoAct2.
#
# Renders task + advantage into the task field as "<task> Advantage: <value>".
# MolmoAct2PackInputsProcessorStep parses this, extracts the advantage value,
# and places it AFTER the full user prompt but BEFORE action tokens — matching
# the RECAP paper (Section V-B): "The advantage indicator appears in the training
# sequence after ˆℓ but before the actions, such that only the action
# log-likelihoods are affected."
#
# Final prompt layout:
# <images><|im_start|>user\n{prompt}<|im_end|>\n<|im_start|>assistant\nAdvantage: positive. <action_output>...
#
# When advantage is absent (CFG dropout), if_present guard skips this message
# and RenderedMessagesToTaskStep leaves the task unchanged — no advantage clause.
bindings:
advantage: "active_at(t, style=advantage)"
messages:
- role: user
content: "${task} Advantage: ${advantage}"
stream: high_level
if_present: advantage
- role: assistant
content: ""
stream: low_level
target: true
+2 -2
View File
@@ -43,10 +43,10 @@ CORE_STYLES = {
# validation. Empty by default — populate from a downstream module that
# also extends ``PERSISTENT_STYLES`` or ``EVENT_ONLY_STYLES`` to declare
# the new style's column.
EXTENDED_STYLES: set[str] = set()
EXTENDED_STYLES: set[str] = {"advantage"}
STYLE_REGISTRY = CORE_STYLES | EXTENDED_STYLES
PERSISTENT_STYLES = {"subtask", "plan", "memory", "motion", "task_aug"}
PERSISTENT_STYLES = {"subtask", "plan", "memory", "motion", "task_aug", "advantage"}
EVENT_ONLY_STYLES = {"interjection", "vqa", "trace"}
# Styles whose ``content`` is grounded in a specific camera view. Rows of these
+7 -1
View File
@@ -757,7 +757,7 @@ class RoboTwinEnvConfig(EnvConfig):
task: str = "beat_block_hammer" # single task or comma-separated list
fps: int = 25
episode_length: int = 300
episode_length: int = 1200
obs_type: str = "pixels_agent_pos"
render_mode: str = "rgb_array"
# Available cameras from RoboTwin's aloha-agilex embodiment: head_camera
@@ -768,6 +768,9 @@ class RoboTwinEnvConfig(EnvConfig):
# must equal what SAPIEN actually renders.
observation_height: int = 240
observation_width: int = 320
# "joint": 14-d joint-space control. "ee": 16-d end-effector-pose deltas executed via CuRobo IK
# (for world-model policies like LingBot-VA that predict per-arm xyz+quaternion+gripper poses).
action_mode: str = "joint"
features: dict[str, PolicyFeature] = field(
default_factory=lambda: {
ACTION: PolicyFeature(type=FeatureType.ACTION, shape=(14,)),
@@ -784,6 +787,8 @@ class RoboTwinEnvConfig(EnvConfig):
)
def __post_init__(self):
if self.action_mode == "ee":
self.features[ACTION] = PolicyFeature(type=FeatureType.ACTION, shape=(16,))
cam_list = [c.strip() for c in self.camera_names.split(",") if c.strip()]
for cam in cam_list:
self.features[f"pixels/{cam}"] = PolicyFeature(
@@ -826,6 +831,7 @@ class RoboTwinEnvConfig(EnvConfig):
observation_height=self.observation_height,
observation_width=self.observation_width,
episode_length=self.episode_length,
action_mode=self.action_mode,
)
+169 -6
View File
@@ -17,6 +17,7 @@ from __future__ import annotations
import importlib
import logging
import os
from collections import defaultdict
from collections.abc import Callable, Sequence
from functools import partial
@@ -28,9 +29,17 @@ import torch
from gymnasium import spaces
from lerobot.types import RobotObservation
from lerobot.utils.import_utils import _scipy_available
from .utils import _LazyAsyncVectorEnv
# scipy is only used for end-effector-pose composition (``--env.action_mode=ee``); guard it so this
# module (and its base-env unit tests, which mock the RoboTwin runtime) imports without scipy installed.
if _scipy_available:
from scipy.spatial.transform import Rotation
else:
Rotation = None
logger = logging.getLogger(__name__)
# Camera names as used by RoboTwin 2.0. The wrapper appends "_rgb" when looking
@@ -41,10 +50,124 @@ ROBOTWIN_CAMERA_NAMES: tuple[str, ...] = (
"right_camera",
)
ACTION_DIM = 14 # 7 DOF × 2 arms
ACTION_DIM = 14 # 7 DOF × 2 arms (joint-space control mode)
# End-effector-pose control mode: per arm [x, y, z, qx, qy, qz, qw, gripper] = 8, dual-arm = 16.
# Used by world-model policies (e.g. LingBot-VA) that predict eef-pose deltas executed via CuRobo IK.
EEF_ACTION_DIM = 16
ACTION_LOW = -1.0
ACTION_HIGH = 1.0
DEFAULT_EPISODE_LENGTH = 300
DEFAULT_EPISODE_LENGTH = 1200
OFFICIAL_INSTRUCTION_ENV = "LEROBOT_ROBOTWIN_OFFICIAL_INSTRUCTION"
OFFICIAL_INSTRUCTION_TYPE_ENV = "LEROBOT_ROBOTWIN_INSTRUCTION_TYPE"
OFFICIAL_INSTRUCTION_MAX_ENV = "LEROBOT_ROBOTWIN_INSTRUCTION_MAX"
def _compose_eef_pose(new_pose: np.ndarray, init_pose: np.ndarray) -> np.ndarray:
"""Compose a single-arm predicted delta pose onto the initial pose.
``new_pose`` / ``init_pose`` are 8-vectors ``[x, y, z, qx, qy, qz, qw, gripper]``. Translation
is added, rotation is composed (``init_R * new_R``), and the gripper is taken from the
prediction. Mirrors ``add_eef_pose`` in the upstream LingBot-VA RoboTwin client.
"""
new_r = Rotation.from_quat(new_pose[3:7])
init_r = Rotation.from_quat(init_pose[3:7])
out_rot = (init_r * new_r).as_quat()
out_trans = new_pose[:3] + init_pose[:3]
return np.concatenate([out_trans, out_rot, new_pose[7:8]])
def _add_init_eef_pose(delta_pose: np.ndarray, init_pose: np.ndarray) -> np.ndarray:
"""Compose a dual-arm (16-d) predicted delta pose onto the initial eef pose, normalizing quats."""
left = _compose_eef_pose(delta_pose[:8], init_pose[:8])
right = _compose_eef_pose(delta_pose[8:], init_pose[8:])
out = np.concatenate([left, right])
# Normalize the two quaternions (indices 3:7 and 11:15) as the upstream client does.
out[3:7] = out[3:7] / (np.linalg.norm(out[3:7]) + 1e-8)
out[11:15] = out[11:15] / (np.linalg.norm(out[11:15]) + 1e-8)
return out
def _env_flag(name: str, default: bool = False) -> bool:
raw = os.environ.get(name)
if raw is None:
return default
return raw.strip().lower() in {"1", "true", "yes", "on"}
def _arm_for_block(block: Any) -> str:
return "left" if float(block.get_pose().p[0]) < 0 else "right"
def _robotwin_blocks_episode_info(task_name: str, env: Any) -> dict[str, str] | None:
"""Infer the episode-info dict used by RoboTwin's official instruction generator for block ranking."""
if task_name == "blocks_ranking_rgb":
return {
"{A}": "red block",
"{B}": "green block",
"{C}": "blue block",
"{a}": _arm_for_block(env.block1),
"{b}": _arm_for_block(env.block2),
"{c}": _arm_for_block(env.block3),
}
if task_name == "blocks_ranking_size":
return {
"{A}": "large block",
"{B}": "medium block",
"{C}": "small block",
"{a}": _arm_for_block(env.block1),
"{b}": _arm_for_block(env.block2),
"{c}": _arm_for_block(env.block3),
}
return None
def _generate_robotwin_official_instruction(task_name: str, env: Any) -> str:
"""Generate language with RoboTwin's official task templates, matching its eval client."""
fallback = task_name.replace("_", " ")
episode_info = _robotwin_blocks_episode_info(task_name, env)
if episode_info is None:
logger.warning(
"Official RoboTwin instruction is not implemented for task=%s; using %r.", task_name, fallback
)
return fallback
try:
# Part of the robotwin simulator repo, this is being pulled by the docker image running robotwin
# see https://github.com/RoboTwin-Platform/RoboTwin/tree/main/description
# Used to generate the official instructions
from description.utils.generate_episode_instructions import generate_episode_descriptions
except Exception:
logger.warning(
"Failed to import RoboTwin official instruction generator; using %r.", fallback, exc_info=True
)
return fallback
instruction_type = os.environ.get(OFFICIAL_INSTRUCTION_TYPE_ENV, "seen")
try:
max_descriptions = int(os.environ.get(OFFICIAL_INSTRUCTION_MAX_ENV, "1000000"))
except ValueError:
max_descriptions = 1000000
results = generate_episode_descriptions(task_name, [episode_info], max_descriptions=max_descriptions)
if not results:
logger.warning(
"RoboTwin generated no official instructions for task=%s; using %r.", task_name, fallback
)
return fallback
options = results[0].get(instruction_type) or results[0].get("seen") or results[0].get("unseen")
if not options:
logger.warning(
"RoboTwin generated no %s official instructions for task=%s; using %r.",
instruction_type,
task_name,
fallback,
)
return fallback
return str(np.random.choice(options))
# D435 dims from task_config/_camera_config.yml (what demo_clean.yml selects).
DEFAULT_CAMERA_H = 240
DEFAULT_CAMERA_W = 320
@@ -234,6 +357,7 @@ class RoboTwinEnv(gym.Env):
observation_width: int | None = None,
episode_length: int = DEFAULT_EPISODE_LENGTH,
render_mode: str = "rgb_array",
action_mode: str = "joint",
):
super().__init__()
self.task_name = task_name
@@ -241,6 +365,13 @@ class RoboTwinEnv(gym.Env):
self.task_description = task_name.replace("_", " ")
self.episode_index = episode_index
self._reset_stride = n_envs
# "joint": 14-d joint-space actions via take_action(action). "ee": 16-d end-effector-pose
# deltas (added onto the episode's initial eef pose) executed via take_action(.., "ee") + IK.
if action_mode not in ("joint", "ee"):
raise ValueError(f"action_mode must be 'joint' or 'ee'; got {action_mode!r}")
self.action_mode = action_mode
self._action_dim = EEF_ACTION_DIM if action_mode == "ee" else ACTION_DIM
self._init_eef_pose: np.ndarray | None = None
self.camera_names = list(camera_names)
# Default to D435 dims (the camera type baked into task_config/demo_clean.yml).
# The YAML-driven lookup is deferred to reset() so construction doesn't
@@ -271,7 +402,7 @@ class RoboTwinEnv(gym.Env):
}
)
self.action_space = spaces.Box(
low=ACTION_LOW, high=ACTION_HIGH, shape=(ACTION_DIM,), dtype=np.float32
low=ACTION_LOW, high=ACTION_HIGH, shape=(self._action_dim,), dtype=np.float32
)
def _ensure_env(self) -> None:
@@ -317,6 +448,18 @@ class RoboTwinEnv(gym.Env):
return {"pixels": images, "agent_pos": joint_state}
def _read_eef_pose(self) -> np.ndarray:
"""Read the current 16-d dual-arm eef pose [left(xyz+quat)+grip, right(xyz+quat)+grip]."""
assert self._env is not None, "_read_eef_pose called before _ensure_env()"
ep = self._env.get_obs()["endpose"]
pose = (
list(ep["left_endpose"])
+ [ep["left_gripper"]]
+ list(ep["right_endpose"])
+ [ep["right_gripper"]]
)
return np.asarray(pose, dtype=np.float64)
def reset(self, seed: int | None = None, **kwargs) -> tuple[RobotObservation, dict]:
self._ensure_env()
super().reset(seed=seed)
@@ -330,16 +473,32 @@ class RoboTwinEnv(gym.Env):
self.episode_index += self._reset_stride
self._step_count = 0
use_official_instruction = self.task_name in {"blocks_ranking_rgb", "blocks_ranking_size"}
if _env_flag(OFFICIAL_INSTRUCTION_ENV, default=use_official_instruction):
self.task_description = _generate_robotwin_official_instruction(self.task_name, self._env)
if hasattr(self._env, "set_instruction"):
self._env.set_instruction(instruction=self.task_description)
logger.info("RoboTwin official instruction | task=%s | %s", self.task_name, self.task_description)
else:
self.task_description = self.task_name.replace("_", " ")
# In eef mode the policy predicts pose deltas relative to the initial eef pose.
if self.action_mode == "ee":
self._init_eef_pose = self._read_eef_pose()
obs = self._get_obs()
return obs, {"is_success": False, "task": self.task_name}
def step(self, action: np.ndarray) -> tuple[RobotObservation, float, bool, bool, dict[str, Any]]:
assert self._env is not None, "step() called before reset()"
if action.ndim != 1 or action.shape[0] != ACTION_DIM:
raise ValueError(f"Expected 1-D action of shape ({ACTION_DIM},), got {action.shape}")
if action.ndim != 1 or action.shape[0] != self._action_dim:
raise ValueError(f"Expected 1-D action of shape ({self._action_dim},), got {action.shape}")
with torch.enable_grad():
if hasattr(self._env, "take_action"):
if self.action_mode == "ee":
ee_action = _add_init_eef_pose(np.asarray(action, dtype=np.float64), self._init_eef_pose)
self._env.take_action(ee_action, action_type="ee")
elif hasattr(self._env, "take_action"):
self._env.take_action(action)
else:
self._env.step(action)
@@ -398,6 +557,7 @@ def _make_env_fns(
observation_height: int,
observation_width: int,
episode_length: int,
action_mode: str = "joint",
) -> list[Callable[[], RoboTwinEnv]]:
"""Return n_envs factory callables for a single task."""
@@ -410,6 +570,7 @@ def _make_env_fns(
observation_height=observation_height,
observation_width=observation_width,
episode_length=episode_length,
action_mode=action_mode,
)
return [partial(_make_one, i) for i in range(n_envs)]
@@ -423,6 +584,7 @@ def create_robotwin_envs(
observation_height: int = DEFAULT_CAMERA_H,
observation_width: int = DEFAULT_CAMERA_W,
episode_length: int = DEFAULT_EPISODE_LENGTH,
action_mode: str = "joint",
) -> dict[str, dict[int, Any]]:
"""Create vectorized RoboTwin 2.0 environments.
@@ -473,6 +635,7 @@ def create_robotwin_envs(
observation_height=observation_height,
observation_width=observation_width,
episode_length=episode_length,
action_mode=action_mode,
)
if is_async:
lazy = _LazyAsyncVectorEnv(fns, cached_obs_space, cached_act_space, cached_metadata)
+22
View File
@@ -83,6 +83,28 @@ class VQBeTSchedulerConfig(LRSchedulerConfig):
return LambdaLR(optimizer, lr_lambda, -1)
@LRSchedulerConfig.register_subclass("constant_with_warmup")
@dataclass
class ConstantWithWarmupSchedulerConfig(LRSchedulerConfig):
"""Linear warmup followed by a constant learning rate.
Mirrors the ``warmup_constant_lambda`` used by LingBot-VA (upstream ``wan_va/train.py``):
the LR ramps linearly from 0 to the peak over ``num_warmup_steps`` steps, then stays flat.
"""
num_warmup_steps: int = 1000
def build(self, optimizer: Optimizer, num_training_steps: int) -> LambdaLR:
warmup_steps = self.num_warmup_steps or 0
def lr_lambda(current_step):
if current_step < warmup_steps:
return float(current_step) / float(max(1, warmup_steps))
return 1.0
return LambdaLR(optimizer, lr_lambda, -1)
@LRSchedulerConfig.register_subclass("cosine_decay_with_warmup")
@dataclass
class CosineDecayWithWarmupSchedulerConfig(LRSchedulerConfig):
+2
View File
@@ -21,6 +21,7 @@ from .factory import get_policy_class, make_policy, make_policy_config, make_pre
from .fastwam.configuration_fastwam import FastWAMConfig as FastWAMConfig
from .gaussian_actor.configuration_gaussian_actor import GaussianActorConfig as GaussianActorConfig
from .groot.configuration_groot import GrootConfig as GrootConfig
from .lingbot_va.configuration_lingbot_va import LingBotVAConfig as LingBotVAConfig
from .molmoact2.configuration_molmoact2 import MolmoAct2Config as MolmoAct2Config
from .multi_task_dit.configuration_multi_task_dit import MultiTaskDiTConfig as MultiTaskDiTConfig
from .pi0.configuration_pi0 import PI0Config as PI0Config
@@ -46,6 +47,7 @@ __all__ = [
"FastWAMConfig",
"GaussianActorConfig",
"GrootConfig",
"LingBotVAConfig",
"MolmoAct2Config",
"MultiTaskDiTConfig",
"PI0Config",
+15
View File
@@ -50,6 +50,7 @@ from .eo1.configuration_eo1 import EO1Config
from .fastwam.configuration_fastwam import FastWAMConfig
from .gaussian_actor.configuration_gaussian_actor import GaussianActorConfig
from .groot.configuration_groot import GrootConfig
from .lingbot_va.configuration_lingbot_va import LingBotVAConfig
from .molmoact2.configuration_molmoact2 import MolmoAct2Config
from .multi_task_dit.configuration_multi_task_dit import MultiTaskDiTConfig
from .pi0.configuration_pi0 import PI0Config
@@ -163,6 +164,10 @@ def get_policy_class(name: str) -> type[PreTrainedPolicy]:
from .vla_jepa.modeling_vla_jepa import VLAJEPAPolicy
return VLAJEPAPolicy
elif name == "lingbot_va":
from .lingbot_va.modeling_lingbot_va import LingBotVAPolicy
return LingBotVAPolicy
elif name == "fastwam":
from .fastwam.modeling_fastwam import FastWAMPolicy
@@ -223,6 +228,8 @@ def make_policy_config(policy_type: str, **kwargs) -> PreTrainedConfig:
return MolmoAct2Config(**kwargs)
elif policy_type == "vla_jepa":
return VLAJEPAConfig(**kwargs)
elif policy_type == "lingbot_va":
return LingBotVAConfig(**kwargs)
elif policy_type == "fastwam":
return FastWAMConfig(**kwargs)
else:
@@ -458,6 +465,14 @@ def make_pre_post_processors(
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, LingBotVAConfig):
from .lingbot_va.processor_lingbot_va import make_lingbot_va_pre_post_processors
processors = make_lingbot_va_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, FastWAMConfig):
from .fastwam.processor_fastwam import make_fastwam_pre_post_processors
+1
View File
@@ -0,0 +1 @@
../../../../docs/source/lingbot_va.mdx
@@ -0,0 +1,21 @@
#!/usr/bin/env python
# Copyright 2026 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.
from .configuration_lingbot_va import LingBotVAConfig
from .modeling_lingbot_va import LingBotVAPolicy
from .processor_lingbot_va import make_lingbot_va_pre_post_processors
__all__ = ["LingBotVAConfig", "LingBotVAPolicy", "make_lingbot_va_pre_post_processors"]
@@ -0,0 +1,168 @@
# Copyright 2026 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.
"""Configuration for the LingBot-VA policy.
LingBot-VA is an autoregressive video-action world-model policy built on the Wan2.2
video-diffusion stack. It interleaves prediction of future video latents and robot
actions in a single dual-stream transformer. See ``docs/source/lingbot_va.mdx`` and the
upstream repository (https://github.com/Robbyant/lingbot-va).
Defaults below match the upstream LIBERO configuration (``wan_va/configs/va_libero_cfg.py``)
and the ``transformer/config.json`` of the released checkpoints.
"""
from dataclasses import dataclass, field
from lerobot.configs.policies import PreTrainedConfig
from lerobot.configs.types import FeatureType, NormalizationMode, PolicyFeature
from lerobot.optim.optimizers import AdamWConfig
from lerobot.optim.schedulers import ConstantWithWarmupSchedulerConfig, LRSchedulerConfig
from lerobot.utils.constants import ACTION
@PreTrainedConfig.register_subclass("lingbot_va")
@dataclass
class LingBotVAConfig(PreTrainedConfig):
"""Configuration for the native LingBot-VA policy integration in LeRobot."""
# Wan transformer architecture
patch_size: tuple[int, int, int] = (1, 2, 2)
num_attention_heads: int = 24
attention_head_dim: int = 128
in_channels: int = 48
out_channels: int = 48
action_dim: int = 30
text_dim: int = 4096
freq_dim: int = 256
ffn_dim: int = 14336
num_layers: int = 30
cross_attn_norm: bool = True
eps: float = 1e-6
rope_max_seq_len: int = 1024
# "flex" = training only (needs recent torch); inference uses "torch" SDPA or "flashattn".
attn_mode: str = "torch"
# Frozen sub-models (VAE + UMT5 text encoder + tokenizer)
# ~20 GB of frozen weights, NOT bundled in the checkpoint; lazily pulled from this HF repo /
# local dir (must hold diffusers-style ``vae/``, ``text_encoder/``, ``tokenizer/`` sub-folders).
wan_pretrained_path: str = "robbyant/lingbot-va-base"
dtype: str = "bfloat16" # transformer / VAE / text-encoder dtype: "bfloat16", "float16", "float32"
# Frozen UMT5-XXL encoder device; "cpu" frees ~11 GB VRAM (it runs once per episode).
text_encoder_device: str = "cpu"
# Observation cameras (order matters: latents are concatenated on width; LIBERO defaults)
obs_cam_keys: list[str] = field(
default_factory=lambda: ["observation.images.image", "observation.images.image2"]
)
# Undo the LIBERO env processor's extra horizontal flip to match the model's training orientation.
image_hflip: bool = False
# Camera latent layout: "width_concat" (cameras concatenated on width; LIBERO) or
# "robotwin_tshape" (full-res head + half-res wrists in a "T"; RoboTwin).
camera_layout: str = "width_concat"
# Inference hyperparameters (LIBERO defaults)
n_obs_steps: int = 1
height: int = 128
width: int = 128
action_per_frame: int = 4
frame_chunk_size: int = 4
attn_window: int = 30
num_inference_steps: int = 20
video_exec_step: int = -1
action_num_inference_steps: int = 50
guidance_scale: float = 5.0
action_guidance_scale: float = 1.0
snr_shift: float = 5.0
action_snr_shift: float = 0.05
max_sequence_length: int = 512 # UMT5 prompt length
# Subset of the 30-d action space used by the benchmark (LIBERO = 7-DoF). The action
# (un)normalization quantiles live in the checkpoint's ``policy_postprocessor.json``, not here.
used_action_channel_ids: list[int] = field(default_factory=lambda: list(range(7)))
# Opt-in: VAE-decode predicted video latents to ``self.last_predicted_frames`` for saving MP4s.
save_predicted_video: bool = False
# Normalization: IDENTITY here; images are scaled + VAE-encoded and actions are
# quantile-(un)normalized inside the policy / dedicated processor steps.
normalization_mapping: dict[str, NormalizationMode] = field(
default_factory=lambda: {
"VISUAL": NormalizationMode.IDENTITY,
"STATE": NormalizationMode.IDENTITY,
"ACTION": NormalizationMode.IDENTITY,
}
)
# Optimizer / scheduler (training; AdamW + warmup-constant per upstream train.py)
optimizer_lr: float = 1e-5
optimizer_betas: tuple[float, float] = (0.9, 0.95)
optimizer_eps: float = 1e-8
optimizer_weight_decay: float = 1e-4
optimizer_grad_clip_norm: float = 1.0
scheduler_warmup_steps: int = 1000
def __post_init__(self):
super().__post_init__()
if self.attn_mode not in ("torch", "flashattn", "flex"):
raise ValueError(f"attn_mode must be one of 'torch', 'flashattn', 'flex'; got {self.attn_mode!r}")
@property
def chunk_size(self) -> int:
"""Number of single-step actions produced per autoregressive chunk."""
return self.frame_chunk_size * self.action_per_frame
@property
def n_action_steps(self) -> int:
"""Number of actions executed before refilling (the whole chunk)."""
return self.chunk_size
def validate_features(self) -> None:
image_features = [key for key, feat in self.input_features.items() if feat.type == FeatureType.VISUAL]
if not image_features:
raise ValueError(
"LingBot-VA requires at least one visual input feature. "
"No features of type FeatureType.VISUAL found in input_features."
)
if ACTION not in self.output_features:
self.output_features[ACTION] = PolicyFeature(
type=FeatureType.ACTION, shape=(len(self.used_action_channel_ids),)
)
def get_optimizer_preset(self) -> AdamWConfig:
return AdamWConfig(
lr=self.optimizer_lr,
betas=self.optimizer_betas,
eps=self.optimizer_eps,
weight_decay=self.optimizer_weight_decay,
grad_clip_norm=self.optimizer_grad_clip_norm,
)
def get_scheduler_preset(self) -> LRSchedulerConfig | None:
# Upstream uses a linear warmup followed by a constant LR (warmup_constant_lambda).
return ConstantWithWarmupSchedulerConfig(num_warmup_steps=self.scheduler_warmup_steps)
@property
def observation_delta_indices(self) -> list[int]:
temporal_downsample = 4
stride = max(1, self.action_per_frame // temporal_downsample)
return list(range(0, self.frame_chunk_size * temporal_downsample * stride, stride))
@property
def action_delta_indices(self) -> list[int]:
return list(range(self.chunk_size))
@property
def reward_delta_indices(self) -> None:
return None
@@ -0,0 +1,853 @@
# Copyright 2024-2025 The Robbyant Team Authors. All rights reserved.
# Copyright 2026 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.
"""LingBot-VA policy: an autoregressive video-action world model on the Wan2.2 stack.
The sampling loop is a faithful re-implementation of the upstream streaming server
(``wan_va/wan_va_server.py``) and LIBERO client (``evaluation/libero/client.py``), adapted
to LeRobot's ``select_action`` interface:
* the trainable dual-stream transformer is owned as a sub-module and round-trips in the
single ``model.safetensors`` checkpoint;
* the frozen Wan VAE + UMT5 text encoder + tokenizer are *lazily pulled* from
``config.wan_pretrained_path`` (not bundled), so the LeRobot checkpoint stays small;
* ``predict_action_chunk`` runs one autoregressive chunk (video stream then action
stream, each with CFG and its own flow-matching scheduler) and updates the KV cache;
* ``select_action`` drains a per-step action queue and records the real observed
keyframes that are fed back into the KV cache when the queue is refilled.
NOTE: The streaming path is written for single-environment eval (``--eval.batch_size=1``).
"""
from collections import deque
import torch
import torch.nn.functional as F # noqa: N812
from einops import rearrange
from torch import Tensor
from lerobot.policies.pretrained import PreTrainedPolicy
from lerobot.utils.constants import ACTION
from lerobot.utils.import_utils import require_package
from .configuration_lingbot_va import LingBotVAConfig
from .utils import (
FlowMatchScheduler,
WanTransformer3DModel,
WanVAEStreamingWrapper,
_sample_timestep_id,
_torch_dtype,
clean_prompt,
data_seq_to_patch,
denormalize_latents,
get_mesh_id,
load_text_encoder,
load_tokenizer,
load_vae,
)
class LingBotVAPolicy(PreTrainedPolicy):
"""LeRobot wrapper for the LingBot-VA autoregressive video-action world model."""
config_class = LingBotVAConfig
name = "lingbot_va"
def __init__(self, config: LingBotVAConfig, **kwargs):
require_package("diffusers", extra="lingbot_va")
require_package("transformers", extra="lingbot_va")
super().__init__(config)
config.validate_features()
self.config = config
self.dtype = _torch_dtype(config.dtype)
# Trainable dual-stream transformer (the only sub-module saved in the LeRobot checkpoint).
self.transformer = WanTransformer3DModel(
patch_size=tuple(config.patch_size),
num_attention_heads=config.num_attention_heads,
attention_head_dim=config.attention_head_dim,
in_channels=config.in_channels,
out_channels=config.out_channels,
action_dim=config.action_dim,
text_dim=config.text_dim,
freq_dim=config.freq_dim,
ffn_dim=config.ffn_dim,
num_layers=config.num_layers,
cross_attn_norm=config.cross_attn_norm,
eps=config.eps,
rope_max_seq_len=config.rope_max_seq_len,
attn_mode=config.attn_mode,
)
# Run the transformer in config.dtype (bf16); norm/modulation paths upcast to fp32 internally.
self.transformer = self.transformer.to(self.dtype)
# Frozen modules are stored OUTSIDE the nn.Module registry (plain dict) so they are
# neither saved into model.safetensors nor moved by ``.to()``. They are lazily loaded
# from ``config.wan_pretrained_path`` the first time inference runs.
self._frozen: dict = {}
self.last_predicted_frames: Tensor | None = None
self.last_predicted_latents: Tensor | None = None
self.reset()
# Frozen-module lazy loading (VAE + UMT5 + tokenizer)
def _ensure_frozen_modules(self):
if self._frozen:
return
path = self.config.wan_pretrained_path
device = self.config.device
# The frozen modules always live in ``vae/``, ``text_encoder/`` and ``tokenizer/``
# sub-folders -- both in the released diffusers-style HF repos and in the local
# ``--bundle-frozen`` output dir. ``from_pretrained(path, subfolder=...)`` resolves
# them for either a HF repo id or a local directory.
vae = load_vae(path, torch_dtype=self.dtype, torch_device=device, subfolder="vae")
# The UMT5-XXL text encoder (~11 GB) runs once per episode; keep it on its own
# (CPU by default) device so the 5B transformer + VAE fit on a single GPU.
text_encoder = load_text_encoder(
path,
torch_dtype=self.dtype,
torch_device=self.config.text_encoder_device,
subfolder="text_encoder",
)
tokenizer = load_tokenizer(path, subfolder="tokenizer")
self._frozen = {
"vae": vae.eval(),
"streaming_vae": WanVAEStreamingWrapper(vae),
"text_encoder": text_encoder.eval(),
"tokenizer": tokenizer,
}
# RoboTwin's T-shape layout encodes the half-resolution wrist cameras through a second
# streaming VAE (separate causal cache) alongside the full-res head camera.
if self.config.camera_layout == "robotwin_tshape":
vae_half = load_vae(path, torch_dtype=self.dtype, torch_device=device, subfolder="vae")
self._frozen["streaming_vae_half"] = WanVAEStreamingWrapper(vae_half.eval())
@property
def _vae(self):
return self._frozen["vae"]
@property
def _streaming_vae(self):
return self._frozen["streaming_vae"]
# PreTrainedPolicy API
def get_optim_params(self) -> dict:
# Only the transformer is trainable; the VAE / text encoder stay frozen (kept outside the
# nn.Module registry). With PEFT/LoRA this naturally returns just the adapter params.
return [p for p in self.transformer.parameters() if p.requires_grad]
def reset(self):
"""Reset all per-episode streaming state (KV cache, queues, frame counter)."""
cfg = self.config
self._action_queue: deque = deque(maxlen=cfg.n_action_steps)
self._obs_buffer: list = [] # raw keyframe obs (one per env substep) observed this chunk
self._executed_actions: Tensor | None = (
None # last chunk's actions (model-normalized) for KV feedback
)
self._started = False # first select_action call uses the obs as the conditioning frame
self._exec_step = 0 # index of the action being executed within the current chunk
self._prev_j = 0 # sub-step index (within a predicted frame) of the last executed action
# Sample one keyframe every ``action_per_frame / temporal_downsample`` executed sub-steps so
# that exactly ``frame_chunk_size * temporal_downsample`` frames are VAE-encoded per chunk
# (the Wan2.2 VAE temporal downsample is 4 -> ``frame_chunk_size`` latent frames).
self._keyframe_stride = max(1, cfg.action_per_frame // 4)
self._frame_st_id = 0
self._first_chunk = True
self._prompt: str | None = None
self._prompt_embeds = None
self._negative_prompt_embeds = None
self.last_predicted_frames = None
self.last_predicted_latents = None
self._use_cfg = (cfg.guidance_scale > 1) or (cfg.action_guidance_scale > 1)
# Two independent flow-matching schedulers (video latent + action streams).
self._scheduler = FlowMatchScheduler(shift=cfg.snr_shift, sigma_min=0.0, extra_one_step=True)
self._action_scheduler = FlowMatchScheduler(
shift=cfg.action_snr_shift, sigma_min=0.0, extra_one_step=True
)
self._scheduler.set_timesteps(1000, training=True)
self._action_scheduler.set_timesteps(1000, training=True)
self._cache_initialised = False
# Clear KV cache on the (already-built) transformer, if present.
if hasattr(self, "transformer"):
self.transformer.clear_cache("pos")
# Reset the causal streaming-VAE feat cache between episodes (mirrors upstream ``_reset``).
# Without this the encoder carries over the previous episode's temporal state, corrupting the
# latent frame counts on the next episode's first encode.
if self._frozen:
self._frozen["streaming_vae"].clear_cache()
if "streaming_vae_half" in self._frozen:
self._frozen["streaming_vae_half"].clear_cache()
# Training (flow-matching dual-stream loss). Requires attn_mode="flex".
def _ensure_train_schedulers(self):
if getattr(self, "_train_sched_latent", None) is None:
cfg = self.config
self._train_sched_latent = FlowMatchScheduler(
shift=cfg.snr_shift, sigma_min=0.0, extra_one_step=True
)
self._train_sched_latent.set_timesteps(1000, training=True)
self._train_sched_action = FlowMatchScheduler(
shift=cfg.action_snr_shift, sigma_min=0.0, extra_one_step=True
)
self._train_sched_action.set_timesteps(1000, training=True)
@torch.no_grad()
def _add_noise_stream(self, latent, scheduler, action_mask, action_mode, noisy_cond_prob):
"""Flow-matching noising of one stream (port of upstream ``Trainer._add_noise``)."""
device = latent.device
b, _c, f, _h, _w = latent.shape
p = self.config.patch_size
patch_f, patch_h, patch_w = (1, 1, 1) if action_mode else (p[0], p[1], p[2])
ts_ids = _sample_timestep_id(f, num_train_timesteps=scheduler.num_train_timesteps)
noise = torch.zeros_like(latent).normal_()
timesteps = scheduler.timesteps[ts_ids].to(device)
noisy_latents = scheduler.add_noise(latent, noise, timesteps, t_dim=2)
targets = scheduler.training_target(latent, noise, timesteps)
grid_id = (
get_mesh_id(
latent.shape[-3] // patch_f,
latent.shape[-2] // patch_h,
latent.shape[-1] // patch_w,
t=1 if action_mode else 0,
f_w=1,
f_shift=0,
action=action_mode,
)
.to(device)[None]
.repeat(b, 1, 1)
)
if torch.rand(1).item() < noisy_cond_prob:
cond_ids = _sample_timestep_id(
f, min_timestep_bd=0.5, max_timestep_bd=1.0, num_train_timesteps=scheduler.num_train_timesteps
)
cond_noise = torch.zeros_like(latent).normal_()
cond_timesteps = scheduler.timesteps[cond_ids].to(device)
latent = scheduler.add_noise(latent, cond_noise, cond_timesteps, t_dim=2)
else:
cond_timesteps = torch.zeros_like(timesteps)
if action_mask is not None:
noisy_latents = noisy_latents * action_mask.float()
targets = targets * action_mask.float()
latent = latent * action_mask.float()
return {
"timesteps": timesteps[None].repeat(b, 1),
"noisy_latents": noisy_latents,
"targets": targets,
"latent": latent,
"cond_timesteps": cond_timesteps[None].repeat(b, 1),
"grid_id": grid_id,
}
def _flow_matching_loss(self, input_dict, pred):
"""Dual-stream flow-matching loss (port of upstream ``Trainer.compute_loss``)."""
latent_pred, action_pred = pred
ld, ad = input_dict["latent_dict"], input_dict["action_dict"]
action_pred = rearrange(action_pred, "b (f n) c -> b c f n 1", f=ad["targets"].shape[-3])
latent_pred = data_seq_to_patch(
self.config.patch_size,
latent_pred,
ld["targets"].shape[-3],
ld["targets"].shape[-2],
ld["targets"].shape[-1],
batch_size=latent_pred.shape[0],
)
bn, fn = ld["timesteps"].shape
lw = self._train_sched_latent.training_weight(ld["timesteps"].flatten()).reshape(bn, fn)
aw = self._train_sched_action.training_weight(ad["timesteps"].flatten()).reshape(bn, fn)
latent_loss = F.mse_loss(latent_pred.float(), ld["targets"].float().detach(), reduction="none")
latent_loss = (
(latent_loss * lw[:, None, :, None, None]).permute(0, 2, 3, 4, 1).flatten(0, 1).flatten(1)
)
latent_loss = (latent_loss.sum(dim=1) / (torch.ones_like(latent_loss).sum(dim=1) + 1e-6)).mean()
amask = ad["actions_mask"].float()
action_loss = F.mse_loss(action_pred.float(), ad["targets"].float().detach(), reduction="none")
action_loss = (
(action_loss * aw[:, None, :, None, None] * amask).permute(0, 2, 3, 4, 1).flatten(0, 1).flatten(1)
)
amask_f = amask.permute(0, 2, 3, 4, 1).flatten(0, 1).flatten(1)
action_loss = (action_loss.sum(dim=1) / (amask_f.sum(dim=1) + 1e-6)).mean()
return latent_loss, action_loss
def training_loss_from_streams(self, latents, actions, actions_mask, text_emb):
"""Core dual-stream training loss given prepared latents / actions / text embeddings.
``latents``: ``[B, in_channels, F, h, w]`` (normalized video latents).
``actions`` / ``actions_mask``: ``[B, action_dim, F, action_per_frame, 1]``.
``text_emb``: ``[B, seq_len, text_dim]``. Returns ``(loss, {latent_loss, action_loss})``.
"""
if self.config.attn_mode != "flex":
raise ValueError(
"LingBot-VA training requires attn_mode='flex' (block-causal flow-matching masks). "
"Load/convert the policy with --policy.attn_mode=flex for training/fine-tuning."
)
self._ensure_train_schedulers()
latent_dict = self._add_noise_stream(
latents, self._train_sched_latent, action_mask=None, action_mode=False, noisy_cond_prob=0.5
)
action_dict = self._add_noise_stream(
actions, self._train_sched_action, action_mask=actions_mask, action_mode=True, noisy_cond_prob=0.0
)
latent_dict["text_emb"] = text_emb
action_dict["text_emb"] = text_emb
action_dict["actions_mask"] = actions_mask
input_dict = {
"latent_dict": latent_dict,
"action_dict": action_dict,
"chunk_size": int(torch.randint(1, 5, (1,)).item()),
"window_size": int(torch.randint(4, 65, (1,)).item()),
}
pred = self.transformer(input_dict, train_mode=True)
latent_loss, action_loss = self._flow_matching_loss(input_dict, pred)
loss = latent_loss + action_loss
return loss, {"latent_loss": latent_loss.item(), "action_loss": action_loss.item()}
def forward(self, batch: dict[str, Tensor]) -> tuple[Tensor, dict | None]:
"""Training forward: dual-stream flow-matching loss.
Builds the (video-latent, action, text) training streams from a LeRobot batch
(VAE-encoding the camera frames and UMT5-encoding the task), then runs the flow-matching
dual-stream loss. Requires the policy to be built with ``attn_mode='flex'``.
"""
self._ensure_frozen_modules()
latents, actions, actions_mask, text_emb = self._build_training_streams(batch)
return self.training_loss_from_streams(latents, actions, actions_mask, text_emb)
@torch.no_grad()
def _build_training_streams(self, batch):
"""Build (latents, actions, actions_mask, text_emb) from a LeRobot training batch.
Camera frames per ``obs_cam_keys`` are expected as a temporal clip ``[B, C, T, H, W]`` (or
``[B, T, C, H, W]``); they are VAE-encoded into ``F = T / temporal_downsample`` latent frames.
Actions ``[B, F*action_per_frame, n_used]`` are scattered into the model's ``action_dim`` space.
"""
cfg = self.config
device = cfg.device
# text embeddings
task = batch.get("task")
if isinstance(task, str):
task = [task]
text_emb = self._get_t5_prompt_embeds(list(task), cfg.max_sequence_length)
# video latents (VAE-encode the camera clips)
latents = self._encode_training_latents(batch)
# actions -> [B, action_dim, F, action_per_frame, 1]
act = batch[ACTION].to(device) # [B, F*apf, n_used]
b = act.shape[0]
used = cfg.used_action_channel_ids
apf, fc = cfg.action_per_frame, cfg.frame_chunk_size
act = act[:, : fc * apf].reshape(b, fc, apf, len(used)).permute(0, 3, 1, 2) # [B, n_used, F, apf]
full = act.new_zeros(b, cfg.action_dim, fc, apf)
idx = torch.as_tensor(used, device=device)
full[:, idx] = act
actions = full.unsqueeze(-1).to(self.dtype) # [B, action_dim, F, apf, 1]
mask = torch.zeros(cfg.action_dim, device=device, dtype=self.dtype)
mask[idx] = 1.0
actions_mask = mask.view(1, -1, 1, 1, 1).expand_as(actions)
return latents, actions, actions_mask, text_emb
@torch.no_grad()
def _encode_training_latents(self, batch) -> Tensor:
"""VAE-encode the per-camera training clips into normalized video latents [B, C, F, h, w]."""
vae_device = next(self._vae.parameters()).device
def _clip(key):
x = batch[key].to(vae_device)
if x.dim() == 4: # [B, C, H, W] -> single frame clip
x = x.unsqueeze(2)
elif x.shape[1] not in (1, 3) and x.shape[2] in (1, 3): # [B, T, C, H, W] -> [B, C, T, H, W]
x = x.permute(0, 2, 1, 3, 4)
return x.contiguous()
def _encode(x, size):
b, c, t = x.shape[:3]
x = F.interpolate(x.flatten(0, 1).float(), size=size, mode="bilinear", align_corners=False)
x = (x.view(b, c, t, *size) * 2.0 - 1.0).to(self.dtype)
mu = self._vae.encode(x).latent_dist.mode() # [B, z_dim, F, h, w]
mean = torch.tensor(self._vae.config.latents_mean).view(1, -1, 1, 1, 1).to(mu.device)
inv_std = (1.0 / torch.tensor(self._vae.config.latents_std)).view(1, -1, 1, 1, 1).to(mu.device)
return ((mu.float() - mean) * inv_std).to(mu)
keys = self.config.obs_cam_keys
if self.config.camera_layout == "robotwin_tshape":
h, w = self.config.height, self.config.width
head = _encode(_clip(keys[0]), (h, w))
left = _encode(_clip(keys[1]), (h // 2, w // 2))
right = _encode(_clip(keys[2]), (h // 2, w // 2))
return torch.cat([torch.cat([left, right], dim=-1), head], dim=-2).to(self.config.device)
per_cam = [_encode(_clip(k), (self.config.height, self.config.width)) for k in keys]
return torch.cat(per_cam, dim=-1).to(self.config.device)
@torch.no_grad()
def select_action(self, batch: dict[str, Tensor], **kwargs) -> Tensor:
"""Return one action, refilling the chunk (and feeding back observed keyframes) as needed.
Mirrors the upstream LIBERO client loop (``evaluation/libero/client.py``): the first obs is
the conditioning frame; every observation produced afterwards is buffered as a keyframe and,
once the chunk's actions are exhausted, the buffered frames + executed actions are fed back
into the KV cache before the next chunk is predicted.
"""
self.eval()
self._ensure_frozen_modules()
self._maybe_init_prompt(batch)
if not self._started:
# First call: this observation conditions the first chunk (it is *not* a keyframe).
self._started = True
actions = self.predict_action_chunk(batch) # [B, chunk_size, n_used]
self._action_queue.extend(actions.transpose(0, 1)) # [chunk_size, B, n_used]
self._obs_buffer = []
self._exec_step = 0
else:
# This observation is the result of the previously executed action -> a candidate
# keyframe. Buffer it on the sub-step boundary the upstream client samples on.
if (self._prev_j + 1) % self._keyframe_stride == 0:
self._obs_buffer.append(self._extract_raw_obs(batch))
if len(self._action_queue) == 0:
# All actions for the current chunk have been executed; feed the observed
# keyframes + executed actions back and predict the next chunk.
actions = self.predict_action_chunk(None)
self._action_queue.extend(actions.transpose(0, 1))
self._exec_step = 0
self._prev_j = self._exec_step % self.config.action_per_frame
self._exec_step += 1
return self._action_queue.popleft()
@torch.no_grad()
def predict_action_chunk(self, batch: dict[str, Tensor], **kwargs) -> Tensor:
"""Run one autoregressive chunk and return actions ``[B, chunk_size, n_used]`` (normalized)."""
self.eval()
self._ensure_frozen_modules()
self._maybe_init_prompt(batch)
is_first = self._first_chunk
if is_first:
init_latent = self._encode_frames([self._extract_raw_obs(batch)])
self._init_latent = init_latent
self._init_streaming_cache(init_latent)
self._obs_buffer = [] # frame 0 (the init obs) conditions the chunk; it is not fed back
actions, latents = self._infer(init_latent, frame_st_id=0)
self._first_chunk = False
else:
# Feed the real observed keyframes + the executed actions back into the KV cache.
self._compute_kv_cache(self._obs_buffer, self._executed_actions)
self._obs_buffer = []
actions, latents = self._infer(None, frame_st_id=self._frame_st_id)
# actions: [B, action_dim, F, action_per_frame, 1] (model-normalized). Keep for KV feedback.
self._executed_actions = actions
if self.config.save_predicted_video:
# Match upstream LingBot-VA visualization: collect chunk latents and decode the
# concatenated latent sequence once after the rollout finishes.
self.last_predicted_frames = None
self.last_predicted_latents = latents.detach().to("cpu")
# On the first chunk, frame 0 is the conditioning frame (already "known"): the upstream
# LIBERO client skips it (start_idx=1), so we drop the first frame's actions here.
used = self.config.used_action_channel_ids
a = actions[:, used] # [B, n_used, F, action_per_frame, 1]
if is_first:
a = a[:, :, 1:] # drop frame 0 -> (F-1) frames of actions
a = a.squeeze(-1).flatten(2) # [B, n_used, n_steps]
a = a.transpose(1, 2).contiguous() # [B, n_steps, n_used]
return a.to(torch.float32)
# Prompt / text encoding
def _maybe_init_prompt(self, batch):
if self._prompt_embeds is not None or batch is None:
return
task = batch.get("task")
prompt = task[0] if isinstance(task, list | tuple) else task
self._prompt = prompt or ""
self._prompt_embeds, self._negative_prompt_embeds = self._encode_prompt(self._prompt)
def _get_t5_prompt_embeds(self, prompt, max_sequence_length):
tokenizer = self._frozen["tokenizer"]
text_encoder = self._frozen["text_encoder"]
device = self.config.device
prompt = [prompt] if isinstance(prompt, str) else prompt
prompt = [clean_prompt(u) for u in prompt]
text_inputs = tokenizer(
prompt,
padding="max_length",
max_length=max_sequence_length,
truncation=True,
add_special_tokens=True,
return_attention_mask=True,
return_tensors="pt",
)
text_input_ids, mask = text_inputs.input_ids, text_inputs.attention_mask
seq_lens = mask.gt(0).sum(dim=1).long()
te_device = next(text_encoder.parameters()).device
prompt_embeds = text_encoder(text_input_ids.to(te_device), mask.to(te_device)).last_hidden_state
prompt_embeds = prompt_embeds.to(dtype=self.dtype, device=device)
prompt_embeds = [u[:v] for u, v in zip(prompt_embeds, seq_lens, strict=False)]
prompt_embeds = torch.stack(
[torch.cat([u, u.new_zeros(max_sequence_length - u.size(0), u.size(1))]) for u in prompt_embeds],
dim=0,
)
return prompt_embeds.to(device)
def _encode_prompt(self, prompt):
max_len = self.config.max_sequence_length
prompt_embeds = self._get_t5_prompt_embeds(prompt, max_len)
negative_prompt_embeds = None
if self._use_cfg:
negative_prompt_embeds = self._get_t5_prompt_embeds("", max_len)
return prompt_embeds, negative_prompt_embeds
# Observation (image) encoding -> normalized video latents
def _extract_raw_obs(self, batch) -> dict[str, Tensor]:
"""Snapshot the configured camera images from a batch (kept raw for later VAE encoding)."""
return {k: batch[k].detach() for k in self.config.obs_cam_keys}
def _camera_frame(self, raw_obs, key, size=None) -> Tensor:
"""Return a single-frame camera tensor [1, C, 1, H, W] resized + scaled to [-1, 1]."""
img = raw_obs[key]
if img.dim() == 3: # [C, H, W]
img = img.unsqueeze(0)
# LeRobot images arrive as float in [0, 1], shape [B, C, H, W].
img = img.to(self.config.device, torch.float32)
if self.config.image_hflip:
img = torch.flip(img, dims=[-1]) # undo the env processor's horizontal flip
if size is None:
size = (self.config.height, self.config.width)
img = F.interpolate(img, size=size, mode="bilinear", align_corners=False)
img = img * 2.0 - 1.0
return img.unsqueeze(2).to(self.dtype) # [1, C, F=1, H, W]
def _normalize_vae_latent(self, enc_out: Tensor) -> Tensor:
"""Take the mean of a VAE encoder output and channel-normalize it (matches upstream)."""
mu, _logvar = torch.chunk(enc_out, 2, dim=1)
latents_mean = torch.tensor(self._vae.config.latents_mean).to(mu.device)
latents_std = torch.tensor(self._vae.config.latents_std).to(mu.device)
mean = latents_mean.view(1, -1, 1, 1, 1)
inv_std = (1.0 / latents_std).view(1, -1, 1, 1, 1)
return ((mu.float() - mean) * inv_std).to(mu)
@torch.no_grad()
def _encode_frames(self, raw_frames: list) -> Tensor:
"""VAE-encode a temporal clip of observed frames and concat the per-camera latents on width.
``raw_frames`` is a list of per-frame obs dicts (one per env sub-step). Each configured
camera is stacked along the temporal axis into a ``[1, C, F, H, W]`` clip and encoded in a
single streaming ``encode_chunk`` call so the VAE temporal downsample (x4) collapses the F
input frames into ``F / 4`` latent frames, with the causal ``feat_cache`` carried across
chunks (mirrors upstream ``_encode_obs``).
"""
vae_device = next(self._vae.parameters()).device
if self.config.camera_layout == "robotwin_tshape":
return self._encode_frames_tshape(raw_frames, vae_device)
per_cam_videos = []
for k in self.config.obs_cam_keys:
frames = [self._camera_frame(fb, k) for fb in raw_frames]
per_cam_videos.append(torch.cat(frames, dim=2)) # [1, C, F, H, W]
videos = torch.cat(per_cam_videos, dim=0) # [num_cam, C, F, H, W]
enc_out = self._streaming_vae.encode_chunk(videos.to(vae_device).to(self.dtype))
mu_norm = self._normalize_vae_latent(enc_out)
# Concatenate the per-camera latents along width.
video_latent = torch.cat(mu_norm.split(1, dim=0), dim=-1)
return video_latent.to(self.config.device)
@torch.no_grad()
def _encode_frames_tshape(self, raw_frames: list, vae_device) -> Tensor:
"""RoboTwin T-shape latent assembly: full-res head + half-res wrists (second streaming VAE).
The two wrist latents are concatenated on width and stacked (on the height axis) on top of
the head latent, mirroring upstream ``_encode_obs`` for ``env_type='robotwin_tshape'``.
"""
cfg = self.config
h, w = cfg.height, cfg.width
head_key, left_key, right_key = cfg.obs_cam_keys[0], cfg.obs_cam_keys[1], cfg.obs_cam_keys[2]
head = torch.cat([self._camera_frame(fb, head_key, size=(h, w)) for fb in raw_frames], dim=2)
left = torch.cat(
[self._camera_frame(fb, left_key, size=(h // 2, w // 2)) for fb in raw_frames], dim=2
)
right = torch.cat(
[self._camera_frame(fb, right_key, size=(h // 2, w // 2)) for fb in raw_frames], dim=2
)
wrists = torch.cat([left, right], dim=0) # [2, C, F, H/2, W/2]
enc_high = self._streaming_vae.encode_chunk(head.to(vae_device).to(self.dtype))
enc_lr = self._frozen["streaming_vae_half"].encode_chunk(wrists.to(vae_device).to(self.dtype))
# wrists side-by-side on width, then stacked on top of the head latent on the height axis.
enc_out = torch.cat([torch.cat(enc_lr.split(1, dim=0), dim=-1), enc_high], dim=-2)
video_latent = self._normalize_vae_latent(enc_out)
return video_latent.to(self.config.device)
# KV cache management
@property
def _latent_hw(self):
if self.config.camera_layout == "robotwin_tshape":
# head (full) on the bottom, two half-res wrists side-by-side on top -> 1.5x height.
return ((self.config.height // 16) * 3) // 2, self.config.width // 16
h = self.config.height // 16
w = (self.config.width // 16) * len(self.config.obs_cam_keys)
return h, w
def _init_streaming_cache(self, init_latent):
cfg = self.config
latent_h, latent_w = self._latent_hw
p = cfg.patch_size
latent_token_per_chunk = (cfg.frame_chunk_size * latent_h * latent_w) // (p[0] * p[1] * p[2])
action_token_per_chunk = cfg.frame_chunk_size * cfg.action_per_frame
self.transformer.create_empty_cache(
"pos",
cfg.attn_window,
latent_token_per_chunk,
action_token_per_chunk,
device=self.config.device,
dtype=self.dtype,
batch_size=2 if self._use_cfg else 1,
)
self._cache_initialised = True
def _repeat_input_for_cfg(self, input_dict):
if self._use_cfg:
input_dict["noisy_latents"] = input_dict["noisy_latents"].repeat(2, 1, 1, 1, 1)
input_dict["text_emb"] = torch.cat(
[
self._prompt_embeds.to(self.dtype).clone(),
self._negative_prompt_embeds.to(self.dtype).clone(),
],
dim=0,
)
input_dict["grid_id"] = input_dict["grid_id"][None].repeat(2, 1, 1)
input_dict["timesteps"] = input_dict["timesteps"][None].repeat(2, 1)
else:
input_dict["grid_id"] = input_dict["grid_id"][None]
input_dict["timesteps"] = input_dict["timesteps"][None]
return input_dict
def _prepare_latent_input(
self,
latent_model_input,
action_model_input,
latent_t=0,
action_t=0,
latent_cond=None,
action_cond=None,
frame_st_id=0,
):
cfg = self.config
device = self.config.device
p = cfg.patch_size
out = {}
if latent_model_input is not None:
out["latent_res_lst"] = {
"noisy_latents": latent_model_input,
"timesteps": torch.ones([latent_model_input.shape[2]], dtype=torch.float32, device=device)
* latent_t,
"grid_id": get_mesh_id(
latent_model_input.shape[-3] // p[0],
latent_model_input.shape[-2] // p[1],
latent_model_input.shape[-1] // p[2],
0,
1,
frame_st_id,
).to(device),
"text_emb": self._prompt_embeds.to(self.dtype).clone(),
}
if latent_cond is not None:
out["latent_res_lst"]["noisy_latents"][:, :, 0:1] = latent_cond[:, :, 0:1]
out["latent_res_lst"]["timesteps"][0:1] *= 0
if action_model_input is not None:
out["action_res_lst"] = {
"noisy_latents": action_model_input,
"timesteps": torch.ones([action_model_input.shape[2]], dtype=torch.float32, device=device)
* action_t,
"grid_id": get_mesh_id(
action_model_input.shape[-3],
action_model_input.shape[-2],
action_model_input.shape[-1],
1,
1,
frame_st_id,
action=True,
).to(device),
"text_emb": self._prompt_embeds.to(self.dtype).clone(),
}
if action_cond is not None:
out["action_res_lst"]["noisy_latents"][:, :, 0:1] = action_cond[:, :, 0:1]
out["action_res_lst"]["timesteps"][0:1] *= 0
out["action_res_lst"]["noisy_latents"][:, ~self._action_mask] *= 0
return out
@property
def _action_mask(self):
mask = torch.zeros([self.config.action_dim], dtype=torch.bool)
mask[self.config.used_action_channel_ids] = True
return mask
# Action conditioning (executed action history) (de)normalization
def _preprocess_action_state(self, action_norm: Tensor) -> Tensor:
"""Build the action-conditioning tensor from the already-normalized executed actions.
``action_norm`` is the model-space action chunk ``[B, action_dim, F, action_per_frame, 1]``.
Upstream re-derives the conditioning from the raw executed action via quantile norm; here
the executed actions are already in the model-normalized space, so we pass them through.
"""
return action_norm.to(self.config.device, self.dtype)
def _compute_kv_cache(self, obs_buffer, executed_actions):
"""Feed real observed keyframes + executed actions back into the KV cache."""
if not obs_buffer or executed_actions is None:
return
self.transformer.clear_pred_cache("pos")
# Encode the buffered keyframe clip in one streaming call (carries the causal VAE cache).
latent_model_input = self._encode_frames(obs_buffer)
# On the first feedback, prepend the init latent so the latent/action frame counts align
# (upstream prepends ``init_latent`` to the observed keyframes when frame_st_id == 0).
if self._frame_st_id == 0 and getattr(self, "_init_latent", None) is not None:
latent_model_input = torch.cat([self._init_latent, latent_model_input], dim=2)
action_model_input = self._preprocess_action_state(executed_actions)
action_model_input = action_model_input.to(latent_model_input)
input_dict = self._prepare_latent_input(
latent_model_input, action_model_input, frame_st_id=self._frame_st_id
)
with torch.no_grad():
self.transformer(
self._repeat_input_for_cfg(input_dict["latent_res_lst"]),
update_cache=2,
cache_name="pos",
action_mode=False,
)
self.transformer(
self._repeat_input_for_cfg(input_dict["action_res_lst"]),
update_cache=2,
cache_name="pos",
action_mode=True,
)
self._frame_st_id += latent_model_input.shape[2]
# The core dual-stream denoising loop (one chunk)
@torch.no_grad()
def _infer(self, init_latent, frame_st_id=0):
cfg = self.config
device = self.config.device
latent_h, latent_w = self._latent_hw
frame_chunk_size = cfg.frame_chunk_size
latents = torch.randn(1, 48, frame_chunk_size, latent_h, latent_w, device=device, dtype=self.dtype)
actions = torch.randn(
1, cfg.action_dim, frame_chunk_size, cfg.action_per_frame, 1, device=device, dtype=self.dtype
)
self._scheduler.set_timesteps(cfg.num_inference_steps)
self._action_scheduler.set_timesteps(cfg.action_num_inference_steps)
timesteps = F.pad(self._scheduler.timesteps, (0, 1), mode="constant", value=0)
if cfg.video_exec_step != -1:
timesteps = timesteps[: cfg.video_exec_step]
action_timesteps = F.pad(self._action_scheduler.timesteps, (0, 1), mode="constant", value=0)
# 1. Video-latent denoising loop
for i, t in enumerate(timesteps):
last_step = i == len(timesteps) - 1
latent_cond = (
init_latent[:, :, 0:1].to(self.dtype)
if frame_st_id == 0 and init_latent is not None
else None
)
input_dict = self._prepare_latent_input(
latents, None, t, t, latent_cond, None, frame_st_id=frame_st_id
)
video_noise_pred = self.transformer(
self._repeat_input_for_cfg(input_dict["latent_res_lst"]),
update_cache=1 if last_step else 0,
cache_name="pos",
action_mode=False,
)
if not last_step or cfg.video_exec_step != -1:
video_noise_pred = data_seq_to_patch(
cfg.patch_size,
video_noise_pred,
frame_chunk_size,
latent_h,
latent_w,
batch_size=2 if self._use_cfg else 1,
)
if cfg.guidance_scale > 1:
video_noise_pred = video_noise_pred[1:] + cfg.guidance_scale * (
video_noise_pred[:1] - video_noise_pred[1:]
)
else:
video_noise_pred = video_noise_pred[:1]
latents = self._scheduler.step(video_noise_pred, t, latents, return_dict=False)
if frame_st_id == 0 and latent_cond is not None:
latents[:, :, 0:1] = latent_cond
# 2. Action denoising loop
for i, t in enumerate(action_timesteps):
last_step = i == len(action_timesteps) - 1
action_cond = (
torch.zeros([1, cfg.action_dim, 1, cfg.action_per_frame, 1], device=device, dtype=self.dtype)
if frame_st_id == 0
else None
)
input_dict = self._prepare_latent_input(
None, actions, t, t, None, action_cond, frame_st_id=frame_st_id
)
action_noise_pred = self.transformer(
self._repeat_input_for_cfg(input_dict["action_res_lst"]),
update_cache=1 if last_step else 0,
cache_name="pos",
action_mode=True,
)
if not last_step:
action_noise_pred = rearrange(action_noise_pred, "b (f n) c -> b c f n 1", f=frame_chunk_size)
if cfg.action_guidance_scale > 1:
action_noise_pred = action_noise_pred[1:] + cfg.action_guidance_scale * (
action_noise_pred[:1] - action_noise_pred[1:]
)
else:
action_noise_pred = action_noise_pred[:1]
actions = self._action_scheduler.step(action_noise_pred, t, actions, return_dict=False)
if frame_st_id == 0 and action_cond is not None:
actions[:, :, 0:1] = action_cond
actions[:, ~self._action_mask] *= 0
return actions, latents
# Predicted-video decoding (opt-in)
@torch.no_grad()
def decode_predicted_latents(self, latents) -> Tensor:
"""Decode a concatenated predicted-latent sequence into ``[T, H, W, 3]`` uint8 frames."""
return self._decode_predicted_video(latents)
@torch.no_grad()
def _decode_predicted_video(self, latents) -> Tensor:
"""VAE-decode predicted latents into a uint8 frame stack ``[T, H, W, 3]`` on CPU."""
vae = self._vae
z_dim = vae.config.z_dim
vae_device = next(vae.parameters()).device
latents = latents.to(device=vae_device, dtype=vae.dtype)
latents = denormalize_latents(latents, vae.config.latents_mean, vae.config.latents_std, z_dim)
video = vae.decode(latents, return_dict=False)[0] # [B, C, F, H, W] in [-1, 1]
video = (video.float().clamp(-1, 1) + 1.0) / 2.0
video = (video[0].permute(1, 2, 3, 0) * 255.0).round().to(torch.uint8) # [F, H, W, C]
return video.cpu()
@@ -0,0 +1,87 @@
# Copyright 2026 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.
"""Pre/post-processor pipelines for the LingBot-VA policy.
The preprocessor passes inputs through (IDENTITY) and the postprocessor maps the policy's
``[-1, 1]`` actions back to physical units with the built-in ``UnnormalizerProcessorStep``
(QUANTILES) using per-channel q01/q99 restored from the checkpoint.
"""
from typing import Any
import torch
from lerobot.configs.types import FeatureType, NormalizationMode
from lerobot.processor import (
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
ProcessorStep,
RenameObservationsProcessorStep,
UnnormalizerProcessorStep,
)
from lerobot.processor.converters import policy_action_to_transition, transition_to_policy_action
from lerobot.utils.constants import (
POLICY_POSTPROCESSOR_DEFAULT_NAME,
POLICY_PREPROCESSOR_DEFAULT_NAME,
)
from .configuration_lingbot_va import LingBotVAConfig
def make_lingbot_va_pre_post_processors(
config: LingBotVAConfig,
dataset_stats: dict[str, dict[str, torch.Tensor]] | None = None,
) -> tuple[
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]],
PolicyProcessorPipeline[PolicyAction, PolicyAction],
]:
"""Build the pre/post processor pipelines for LingBot-VA."""
input_steps: list[ProcessorStep] = [
RenameObservationsProcessorStep(rename_map={}),
AddBatchDimensionProcessorStep(),
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
DeviceProcessorStep(device=config.device),
]
# Unnormalize actions from [-1, 1] to physical units (QUANTILES) using q01/q99 restored from the checkpoint.
output_steps: list[ProcessorStep] = [
UnnormalizerProcessorStep(
features=config.output_features,
norm_map={FeatureType.ACTION: NormalizationMode.QUANTILES},
stats=dataset_stats,
),
DeviceProcessorStep(device="cpu"),
]
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
File diff suppressed because it is too large Load Diff
@@ -73,6 +73,19 @@ class MolmoAct2Config(PreTrainedConfig):
num_inference_steps: int | None = None
mask_action_dim_padding: bool = True
enable_inference_cuda_graph: bool = True
# Language conditioning (e.g. RECAP advantage). When set, RenderMessagesStep
# resolves language_persistent rows via the recipe YAML. Same mechanism as PI05.
recipe_path: str | None = None
# Inference-time advantage indicator (e.g. "Advantage: positive. ").
# Used during rollout when no language_persistent data is available.
# Placed after the user prompt, before action tokens.
advantage_prefix: str = ""
# Classifier-Free Guidance (CFG) scale for inference (RECAP Eq. 13).
# 1.0 = no guidance. >1.0 = dual-path: v = v_uncond + beta * (v_cond - v_uncond)
cfg_beta: float = 1.0
# MolmoAct2-local eval option. When enabled, stochastic continuous action
# generation uses a rollout-local generator derived from eval_seed.
per_episode_seed: bool = False
@@ -359,6 +359,7 @@ def _build_robot_text(
add_setup_tokens: bool,
add_control_tokens: bool,
num_images: int,
advantage: str = "",
) -> str:
setup_text = _wrap_setup_text(setup_type, add_setup_tokens=add_setup_tokens)
control_text = _wrap_control_text(control_mode, add_control_tokens=add_control_tokens)
@@ -375,7 +376,10 @@ def _build_robot_text(
image_prefix = "<|image|>"
else:
image_prefix = "".join(f"Image {idx + 1}<|image|>" for idx in range(num_images))
return f"{image_prefix}<|im_start|>user\n{prompt}<|im_end|>\n<|im_start|>assistant\n{ACTION_OUTPUT_TOKEN}"
# Per RECAP paper (Section V-B): advantage indicator goes after context,
# before actions, so only action log-likelihoods are affected.
advantage_clause = f"Advantage: {advantage}. " if advantage else ""
return f"{image_prefix}<|im_start|>user\n{prompt}<|im_end|>\n<|im_start|>assistant\n{advantage_clause}{ACTION_OUTPUT_TOKEN}"
def _as_text_list(value: Any, batch_size: int) -> list[str]:
@@ -695,6 +699,39 @@ class MolmoAct2ClampNormalizedProcessorStep(ProcessorStep):
return features
@ProcessorStepRegistry.register(name="molmoact2_normalize_task")
@dataclass
class MolmoAct2NormalizeTaskStep(ProcessorStep):
"""Normalize the task text in complementary_data before recipe rendering.
Ensures ${task} in recipe templates gets the same normalized form that
MolmoAct2PackInputsProcessorStep would produce, so training prompts
match inference prompts.
"""
def __call__(self, transition: EnvTransition) -> EnvTransition:
complementary = transition.get(TransitionKey.COMPLEMENTARY_DATA)
if not isinstance(complementary, dict):
return transition
task = complementary.get("task")
if task is None:
return transition
transition = transition.copy()
complementary = dict(complementary)
if isinstance(task, str):
complementary["task"] = _normalize_question_text(task)
elif isinstance(task, list):
complementary["task"] = [_normalize_question_text(t) for t in task]
transition[TransitionKey.COMPLEMENTARY_DATA] = complementary
return transition
def transform_features(
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
return features
@ProcessorStepRegistry.register(name="molmoact2_pack_inputs")
@dataclass
class MolmoAct2PackInputsProcessorStep(ProcessorStep):
@@ -715,6 +752,8 @@ class MolmoAct2PackInputsProcessorStep(ProcessorStep):
chunk_size: int = 30
max_action_dim: int = 32
env_action_dim: int | None = None
# RECAP: advantage indicator for inference (e.g. "Advantage: positive. ")
advantage_prefix: str = ""
def __post_init__(self) -> None:
require_package("transformers", extra="molmoact2")
@@ -757,6 +796,7 @@ class MolmoAct2PackInputsProcessorStep(ProcessorStep):
"chunk_size": self.chunk_size,
"max_action_dim": self.max_action_dim,
"env_action_dim": self.env_action_dim,
"advantage_prefix": self.advantage_prefix,
}
def _resolve_max_sequence_length(
@@ -919,8 +959,40 @@ class MolmoAct2PackInputsProcessorStep(ProcessorStep):
if task_source is None:
task_source = complementary.get("language_instruction")
tasks = _as_text_list(task_source, batch_size)
if self.normalize_language:
tasks = [_normalize_question_text(task) for task in tasks]
# Resolve the advantage indicator. Per RECAP paper (Section V-B), it goes
# after all context but before actions — handled by _build_robot_text.
# Source priority: recipe-rendered "advantage" key > config advantage_prefix.
advantages: list[str] = []
recipe_rendered = "base_task" in complementary
if recipe_rendered:
# Recipe rendered the task as "<task> Advantage: <value>".
# Extract the advantage value and restore the clean task.
clean_tasks: list[str] = []
for t in tasks:
if " Advantage: " in t:
split_idx = t.rindex(" Advantage: ")
clean_task = t[:split_idx]
adv = t[split_idx + len(" Advantage: ") :]
advantages.append(adv)
clean_tasks.append(clean_task)
else:
advantages.append("")
clean_tasks.append(t)
tasks = clean_tasks
else:
if self.normalize_language:
tasks = [_normalize_question_text(task) for task in tasks]
if self.advantage_prefix:
# Extract just the value from prefix like "Advantage: positive. "
prefix = self.advantage_prefix.strip()
if prefix.startswith("Advantage:"):
adv_val = prefix[len("Advantage:") :].strip().rstrip(".")
else:
adv_val = prefix
advantages = [adv_val] * batch_size
else:
advantages = [""] * batch_size
complementary["task"] = tasks
action_padded = None
@@ -953,6 +1025,7 @@ class MolmoAct2PackInputsProcessorStep(ProcessorStep):
add_setup_tokens=self.add_setup_tokens,
add_control_tokens=self.add_control_tokens,
num_images=len(images),
advantage=advantages[batch_idx],
)
prompt_texts.append(prompt)
if build_action_labels:
@@ -1164,28 +1237,48 @@ def make_molmoact2_pre_post_processors(
stats=masked_dataset_stats,
),
MolmoAct2ClampNormalizedProcessorStep(normalization_masks=normalization_masks),
MolmoAct2PackInputsProcessorStep(
checkpoint_path=config.checkpoint_path,
checkpoint_revision=config.checkpoint_revision,
checkpoint_force_download=config.checkpoint_force_download,
action_mode=config.action_mode,
discrete_action_tokenizer=config.discrete_action_tokenizer,
image_keys=image_keys,
allow_image_key_fallback=not bool(config.image_keys),
setup_type=setup_type,
control_mode=control_mode,
normalize_language=config.normalize_language,
add_setup_tokens=config.add_setup_tokens,
add_control_tokens=config.add_control_tokens,
num_state_tokens=config.num_state_tokens,
max_sequence_length=config.max_sequence_length,
chunk_size=chunk_size,
max_action_dim=config.expected_max_action_dim,
env_action_dim=env_action_dim,
),
DeviceProcessorStep(device=config.device),
]
# Insert language rendering steps when a recipe is configured (e.g. RECAP advantage)
if config.recipe_path is not None:
from lerobot.configs.recipe import load_recipe
from lerobot.processor.render_messages_processor import RenderMessagesStep
from lerobot.processor.rendered_messages_to_task import RenderedMessagesToTaskStep
recipe = load_recipe(config.recipe_path)
# Normalize task text before recipe uses ${task}, ensuring consistency
# between training (recipe-rendered) and inference (advantage_prefix).
if config.normalize_language:
input_steps.append(MolmoAct2NormalizeTaskStep())
input_steps.append(RenderMessagesStep(recipe=recipe))
input_steps.append(RenderedMessagesToTaskStep())
input_steps.extend(
[
MolmoAct2PackInputsProcessorStep(
checkpoint_path=config.checkpoint_path,
checkpoint_revision=config.checkpoint_revision,
checkpoint_force_download=config.checkpoint_force_download,
action_mode=config.action_mode,
discrete_action_tokenizer=config.discrete_action_tokenizer,
image_keys=image_keys,
allow_image_key_fallback=not bool(config.image_keys),
setup_type=setup_type,
control_mode=control_mode,
normalize_language=config.normalize_language,
add_setup_tokens=config.add_setup_tokens,
add_control_tokens=config.add_control_tokens,
num_state_tokens=config.num_state_tokens,
max_sequence_length=config.max_sequence_length,
chunk_size=chunk_size,
max_action_dim=config.expected_max_action_dim,
env_action_dim=env_action_dim,
advantage_prefix=config.advantage_prefix,
),
DeviceProcessorStep(device=config.device),
]
)
output_steps: list[ProcessorStep] = [
MolmoAct2ClampActionProcessorStep(),
MolmoAct2MaskedUnnormalizerProcessorStep(
@@ -87,6 +87,17 @@ class PI05Config(PreTrainedConfig):
freeze_vision_encoder: bool = False # Freeze only the vision encoder
train_expert_only: bool = False # Freeze entire VLM, train only action expert and projections
# Language conditioning (e.g. RECAP advantage). When set, RenderMessagesStep
# is inserted into the preprocessor to resolve language_persistent rows via
# the recipe YAML before prompt construction.
recipe_path: str | None = None
# Classifier-Free Guidance (CFG) scale for inference (Eq. 13 in RECAP paper).
# 1.0 = no guidance (default). >1.0 enables dual-path denoising where:
# v = v_uncond + cfg_beta * (v_cond - v_uncond)
# VLM runs twice (cond + uncond prompts), action expert runs 2x per step.
cfg_beta: float = 1.0
# Optimizer settings: see openpi `AdamW`
optimizer_lr: float = 2.5e-5 # see openpi `CosineDecaySchedule: peak_lr`
optimizer_betas: tuple[float, float] = (0.9, 0.95)
+141 -2
View File
@@ -52,6 +52,8 @@ from lerobot.utils.constants import (
ACTION,
OBS_LANGUAGE_ATTENTION_MASK,
OBS_LANGUAGE_TOKENS,
OBS_LANGUAGE_UNCOND_ATTENTION_MASK,
OBS_LANGUAGE_UNCOND_TOKENS,
OPENPI_ATTENTION_MASK_VALUE,
)
@@ -148,6 +150,20 @@ def clone_past_key_values(past_key_values):
)
def cat_past_key_values(kv_a, kv_b):
"""Concatenate two DynamicCaches along the batch dimension for batched CFG."""
return DynamicCache(
tuple(
(
torch.cat([ka, kb], dim=0),
torch.cat([va, vb], dim=0),
sw_a,
)
for (ka, va, sw_a), (kb, vb, _sw_b) in zip(kv_a, kv_b, strict=True)
)
)
def pad_vector(vector, new_dim):
"""Pad the last dimension of a vector to new_dim with zeros.
@@ -797,9 +813,17 @@ class PI05Pytorch(nn.Module): # see openpi `PI0Pytorch`
masks,
noise=None,
num_steps=None,
uncond_tokens=None,
uncond_masks=None,
**kwargs: Unpack[ActionSelectKwargs],
) -> Tensor:
"""Do a full inference forward and compute the action."""
"""Do a full inference forward and compute the action.
When cfg_beta > 1.0 and uncond_tokens/uncond_masks are provided, performs
Classifier-Free Guidance: VLM runs twice (conditioned + unconditional), action
expert runs twice per denoising step, and velocities are interpolated via
v = v_uncond + cfg_beta * (v_cond - v_uncond).
"""
if num_steps is None:
num_steps = self.config.num_inference_steps
@@ -815,6 +839,9 @@ class PI05Pytorch(nn.Module): # see openpi `PI0Pytorch`
) # Use config max_action_dim for internal processing
noise = self.sample_noise(actions_shape, device)
cfg_enabled = self.config.cfg_beta > 1.0 and uncond_tokens is not None and uncond_masks is not None
# Prefill VLM for conditioned prompt
prefix_embs, prefix_pad_masks, prefix_att_masks = self.embed_prefix(images, img_masks, tokens, masks)
prefix_att_2d_masks = make_att_2d_masks(prefix_pad_masks, prefix_att_masks)
prefix_position_ids = torch.cumsum(prefix_pad_masks, dim=1) - 1
@@ -830,6 +857,23 @@ class PI05Pytorch(nn.Module): # see openpi `PI0Pytorch`
use_cache=True,
)
# Prefill VLM for unconditional prompt (CFG)
if cfg_enabled:
uncond_prefix_embs, uncond_prefix_pad_masks, uncond_prefix_att_masks = self.embed_prefix(
images, img_masks, uncond_tokens, uncond_masks
)
uncond_prefix_att_2d_masks = make_att_2d_masks(uncond_prefix_pad_masks, uncond_prefix_att_masks)
uncond_prefix_position_ids = torch.cumsum(uncond_prefix_pad_masks, dim=1) - 1
uncond_prefix_att_2d_masks_4d = self._prepare_attention_masks_4d(uncond_prefix_att_2d_masks)
_, uncond_past_key_values = self.paligemma_with_expert.forward(
attention_mask=uncond_prefix_att_2d_masks_4d,
position_ids=uncond_prefix_position_ids,
past_key_values=None,
inputs_embeds=[uncond_prefix_embs, None],
use_cache=True,
)
dt = -1.0 / num_steps
x_t = noise
@@ -838,6 +882,15 @@ class PI05Pytorch(nn.Module): # see openpi `PI0Pytorch`
time_tensor = torch.tensor(time, dtype=torch.float32, device=device).expand(bsize)
def denoise_step_partial_call(input_x_t, current_timestep=time_tensor):
if cfg_enabled:
return self.denoise_step_cfg_batched(
cond_prefix_pad_masks=prefix_pad_masks,
cond_past_key_values=past_key_values,
uncond_prefix_pad_masks=uncond_prefix_pad_masks,
uncond_past_key_values=uncond_past_key_values,
x_t=input_x_t,
timestep=current_timestep,
)
return self.denoise_step(
prefix_pad_masks=prefix_pad_masks,
past_key_values=past_key_values,
@@ -907,6 +960,80 @@ class PI05Pytorch(nn.Module): # see openpi `PI0Pytorch`
suffix_out = suffix_out.to(dtype=torch.float32)
return self.action_out_proj(suffix_out)
def denoise_step_cfg_batched(
self,
cond_prefix_pad_masks,
cond_past_key_values,
uncond_prefix_pad_masks,
uncond_past_key_values,
x_t,
timestep,
):
"""Batched CFG denoising: runs cond + uncond in a single forward pass.
Concatenates cond and uncond inputs along the batch dimension, runs one
action expert forward (2x batch), then splits and applies CFG interpolation.
This is ~1.5x faster than two sequential denoise_step calls due to better
GPU utilization (inspired by Qwen2.5-Omni DiT / diffusers batched CFG).
"""
# Embed suffix once (same x_t and timestep for both branches)
suffix_embs, suffix_pad_masks, suffix_att_masks, adarms_cond = self.embed_suffix(x_t, timestep)
bsize = cond_prefix_pad_masks.shape[0]
suffix_len = suffix_pad_masks.shape[1]
cond_prefix_len = cond_prefix_pad_masks.shape[1]
uncond_prefix_len = uncond_prefix_pad_masks.shape[1]
# Build attention masks for cond branch
cond_prefix_2d = cond_prefix_pad_masks[:, None, :].expand(bsize, suffix_len, cond_prefix_len)
cond_suffix_att_2d = make_att_2d_masks(suffix_pad_masks, suffix_att_masks)
cond_full_att = torch.cat([cond_prefix_2d, cond_suffix_att_2d], dim=2)
cond_prefix_offsets = torch.sum(cond_prefix_pad_masks, dim=-1)[:, None]
cond_position_ids = cond_prefix_offsets + torch.cumsum(suffix_pad_masks, dim=1) - 1
# Build attention masks for uncond branch
uncond_prefix_2d = uncond_prefix_pad_masks[:, None, :].expand(bsize, suffix_len, uncond_prefix_len)
uncond_suffix_att_2d = make_att_2d_masks(suffix_pad_masks, suffix_att_masks)
uncond_full_att = torch.cat([uncond_prefix_2d, uncond_suffix_att_2d], dim=2)
uncond_prefix_offsets = torch.sum(uncond_prefix_pad_masks, dim=-1)[:, None]
uncond_position_ids = uncond_prefix_offsets + torch.cumsum(suffix_pad_masks, dim=1) - 1
# Concatenate on batch dim: [cond_batch; uncond_batch]
batched_full_att = torch.cat([cond_full_att, uncond_full_att], dim=0)
batched_full_att_4d = self._prepare_attention_masks_4d(batched_full_att)
batched_position_ids = torch.cat([cond_position_ids, uncond_position_ids], dim=0)
batched_suffix_embs = torch.cat([suffix_embs, suffix_embs], dim=0)
batched_adarms_cond = torch.cat([adarms_cond, adarms_cond], dim=0)
# Concatenate KV caches on batch dim
batched_past_kv = cat_past_key_values(
clone_past_key_values(cond_past_key_values),
clone_past_key_values(uncond_past_key_values),
)
self.paligemma_with_expert.gemma_expert.model.config._attn_implementation = "eager" # noqa: SLF001
# Single forward pass for both branches
outputs_embeds, _ = self.paligemma_with_expert.forward(
attention_mask=batched_full_att_4d,
position_ids=batched_position_ids,
past_key_values=batched_past_kv,
inputs_embeds=[None, batched_suffix_embs],
use_cache=False,
adarms_cond=[None, batched_adarms_cond],
)
suffix_out = outputs_embeds[1]
suffix_out = suffix_out[:, -self.config.chunk_size :]
suffix_out = suffix_out.to(dtype=torch.float32)
v_all = self.action_out_proj(suffix_out)
# Split: first half = cond, second half = uncond
v_cond, v_uncond = v_all.chunk(2, dim=0)
# CFG interpolation: v = v_uncond + beta * (v_cond - v_uncond)
return v_uncond + self.config.cfg_beta * (v_cond - v_uncond)
class PI05Policy(PreTrainedPolicy):
"""PI05 Policy for LeRobot."""
@@ -1243,8 +1370,20 @@ class PI05Policy(PreTrainedPolicy):
images, img_masks = self._preprocess_images(batch)
tokens, masks = batch[f"{OBS_LANGUAGE_TOKENS}"], batch[f"{OBS_LANGUAGE_ATTENTION_MASK}"]
# CFG: pass unconditional tokens if available
uncond_tokens = batch.get(f"{OBS_LANGUAGE_UNCOND_TOKENS}")
uncond_masks = batch.get(f"{OBS_LANGUAGE_UNCOND_ATTENTION_MASK}")
# Sample actions using the model (pass through RTC kwargs, no separate state needed for PI05)
actions = self.model.sample_actions(images, img_masks, tokens, masks, **kwargs)
actions = self.model.sample_actions(
images,
img_masks,
tokens,
masks,
uncond_tokens=uncond_tokens,
uncond_masks=uncond_masks,
**kwargs,
)
# Unpad actions to actual action dimension
original_action_dim = self.config.output_features[ACTION].shape[0]
+69 -15
View File
@@ -40,6 +40,8 @@ from lerobot.processor import (
)
from lerobot.types import EnvTransition, TransitionKey
from lerobot.utils.constants import (
OBS_LANGUAGE_UNCOND_ATTENTION_MASK,
OBS_LANGUAGE_UNCOND_TOKENS,
OBS_STATE,
POLICY_POSTPROCESSOR_DEFAULT_NAME,
POLICY_PREPROCESSOR_DEFAULT_NAME,
@@ -57,6 +59,7 @@ class Pi05PrepareStateTokenizerProcessorStep(ProcessorStep):
max_state_dim: int = 32
task_key: str = "task"
cfg_enabled: bool = False
def __call__(self, transition: EnvTransition) -> EnvTransition:
transition = transition.copy()
@@ -84,8 +87,25 @@ class Pi05PrepareStateTokenizerProcessorStep(ProcessorStep):
full_prompts.append(full_prompt)
transition[TransitionKey.COMPLEMENTARY_DATA][self.task_key] = full_prompts
# Normalize state to [-1, 1] range if needed (assuming it's already normalized by normalizer processor step!!)
# Discretize into 256 bins (see openpi `PaligemmaTokenizer.tokenize()`)
# Build unconditional prompts for CFG (same state but original task without advantage)
if self.cfg_enabled:
base_tasks = transition.get(TransitionKey.COMPLEMENTARY_DATA, {}).get("base_task")
if base_tasks is None:
base_tasks = tasks
if isinstance(base_tasks, str):
base_tasks = [base_tasks] * len(tasks)
uncond_prompts = []
for i, base_task in enumerate(base_tasks):
cleaned_text = base_task.strip().replace("_", " ").replace("\n", " ")
state_str = " ".join(map(str, discretized_states[i]))
uncond_prompt = f"Task: {cleaned_text}, State: {state_str};\nAction: "
uncond_prompts.append(uncond_prompt)
transition[TransitionKey.COMPLEMENTARY_DATA]["uncond_task"] = uncond_prompts
return transition
def transform_features(
@@ -111,9 +131,10 @@ def make_pi05_pre_post_processors(
1. Renaming features to match pretrained configurations.
2. Normalizing input and output features based on dataset statistics.
3. Adding a batch dimension.
4. Appending a newline character to the task description for tokenizer compatibility.
5. Tokenizing the text prompt using the PaliGemma tokenizer.
6. Moving all data to the specified device.
4. (Optional) Rendering language annotations via recipe YAML.
5. (Optional) Flattening rendered messages into the task string.
6. Tokenizing the text prompt using the PaliGemma tokenizer.
7. Moving all data to the specified device.
The post-processing pipeline handles the model's output by:
1. Moving data to the CPU.
@@ -122,8 +143,6 @@ def make_pi05_pre_post_processors(
Args:
config: The configuration object for the PI0 policy.
dataset_stats: A dictionary of statistics for normalization.
preprocessor_kwargs: Additional arguments for the pre-processor pipeline.
postprocessor_kwargs: Additional arguments for the post-processor pipeline.
Returns:
A tuple containing the configured pre-processor and post-processor pipelines.
@@ -147,16 +166,51 @@ def make_pi05_pre_post_processors(
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
Pi05PrepareStateTokenizerProcessorStep(max_state_dim=config.max_state_dim),
TokenizerProcessorStep(
tokenizer_name="google/paligemma-3b-pt-224",
max_length=config.tokenizer_max_length,
padding_side="right",
padding="max_length",
),
DeviceProcessorStep(device=config.device),
]
# Insert language rendering steps when a recipe is configured (e.g. RECAP advantage)
if config.recipe_path is not None:
from lerobot.configs.recipe import load_recipe
from lerobot.processor.render_messages_processor import RenderMessagesStep
from lerobot.processor.rendered_messages_to_task import RenderedMessagesToTaskStep
recipe = load_recipe(config.recipe_path)
input_steps.append(RenderMessagesStep(recipe=recipe))
input_steps.append(RenderedMessagesToTaskStep())
cfg_enabled = config.cfg_beta > 1.0
input_steps.extend(
[
Pi05PrepareStateTokenizerProcessorStep(
max_state_dim=config.max_state_dim,
cfg_enabled=cfg_enabled,
),
TokenizerProcessorStep(
tokenizer_name="google/paligemma-3b-pt-224",
max_length=config.tokenizer_max_length,
padding_side="right",
padding="max_length",
),
]
)
# Add unconditional prompt tokenizer for CFG inference
if cfg_enabled:
input_steps.append(
TokenizerProcessorStep(
tokenizer_name="google/paligemma-3b-pt-224",
max_length=config.tokenizer_max_length,
padding_side="right",
padding="max_length",
task_key="uncond_task",
output_tokens_key=OBS_LANGUAGE_UNCOND_TOKENS,
output_mask_key=OBS_LANGUAGE_UNCOND_ATTENTION_MASK,
)
)
input_steps.append(DeviceProcessorStep(device=config.device))
output_steps: list[ProcessorStep] = [
UnnormalizerProcessorStep(
features=config.output_features, norm_map=config.normalization_mapping, stats=dataset_stats
@@ -0,0 +1,86 @@
#!/usr/bin/env python
# Copyright 2026 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.
"""Adapter step that flattens rendered chat messages back into a task string.
Bridges RenderMessagesStep (which outputs structured messages) to policies
that expect a plain task string in complementary_data["task"] (e.g. PI05).
"""
from __future__ import annotations
from lerobot.configs import PipelineFeatureType, PolicyFeature
from .pipeline import ComplementaryDataProcessorStep, ProcessorStepRegistry
@ProcessorStepRegistry.register(name="rendered_messages_to_task")
class RenderedMessagesToTaskStep(ComplementaryDataProcessorStep):
"""Extract user-role message content from rendered messages into the task string.
After RenderMessagesStep renders a recipe into structured messages, this
step extracts content from all user-role messages, joins them, and writes
the result to complementary_data["task"]. This allows downstream steps
(like Pi05PrepareStateTokenizerProcessorStep) to consume the
advantage-conditioned prompt without modification.
No-ops when the "messages" key is absent (backward compatible with
pipelines that don't use language annotations).
"""
def complementary_data(self, complementary_data: dict) -> dict:
messages = complementary_data.get("messages")
if messages is None:
return complementary_data
user_parts = []
for msg in messages:
if msg.get("role") == "user":
content = msg.get("content", "")
if isinstance(content, str) and content:
user_parts.append(content)
elif isinstance(content, list):
# HF multimodal blocks: extract text blocks
for block in content:
if isinstance(block, dict) and block.get("type") == "text":
text = block.get("text", "")
if text:
user_parts.append(text)
new_complementary_data = dict(complementary_data)
if user_parts:
task = complementary_data.get("task")
# Preserve the original task for CFG unconditional prompt
new_complementary_data["base_task"] = task
# Wrap in list if the original task was a list (batched)
joined = "\n".join(user_parts)
if isinstance(task, list):
new_complementary_data["task"] = [joined] * len(task)
else:
new_complementary_data["task"] = joined
# Remove consumed rendering outputs
new_complementary_data.pop("messages", None)
new_complementary_data.pop("message_streams", None)
new_complementary_data.pop("target_message_indices", None)
return new_complementary_data
def transform_features(
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
return features
+8 -6
View File
@@ -81,6 +81,8 @@ class TokenizerProcessorStep(ObservationProcessorStep):
padding_side: str = "right"
padding: str = "max_length"
truncation: bool = True
output_tokens_key: str = OBS_LANGUAGE_TOKENS
output_mask_key: str = OBS_LANGUAGE_ATTENTION_MASK
# Internal tokenizer instance (not part of the config)
input_tokenizer: Any = field(default=None, init=False, repr=False)
@@ -201,8 +203,8 @@ class TokenizerProcessorStep(ObservationProcessorStep):
new_observation = dict(observation)
# Add tokenized data to the observation
new_observation[OBS_LANGUAGE_TOKENS] = tokenized_prompt["input_ids"]
new_observation[OBS_LANGUAGE_ATTENTION_MASK] = tokenized_prompt["attention_mask"].to(dtype=torch.bool)
new_observation[self.output_tokens_key] = tokenized_prompt["input_ids"]
new_observation[self.output_mask_key] = tokenized_prompt["attention_mask"].to(dtype=torch.bool)
# Tokenize subtask if available
subtask = self.get_subtask(self.transition)
@@ -309,14 +311,14 @@ class TokenizerProcessorStep(ObservationProcessorStep):
The updated dictionary of policy features.
"""
# Add a feature for the token IDs if it doesn't already exist
if OBS_LANGUAGE_TOKENS not in features[PipelineFeatureType.OBSERVATION]:
features[PipelineFeatureType.OBSERVATION][OBS_LANGUAGE_TOKENS] = PolicyFeature(
if self.output_tokens_key not in features[PipelineFeatureType.OBSERVATION]:
features[PipelineFeatureType.OBSERVATION][self.output_tokens_key] = PolicyFeature(
type=FeatureType.LANGUAGE, shape=(self.max_length,)
)
# Add a feature for the attention mask if it doesn't already exist
if OBS_LANGUAGE_ATTENTION_MASK not in features[PipelineFeatureType.OBSERVATION]:
features[PipelineFeatureType.OBSERVATION][OBS_LANGUAGE_ATTENTION_MASK] = PolicyFeature(
if self.output_mask_key not in features[PipelineFeatureType.OBSERVATION]:
features[PipelineFeatureType.OBSERVATION][self.output_mask_key] = PolicyFeature(
type=FeatureType.LANGUAGE, shape=(self.max_length,)
)
+4
View File
@@ -13,6 +13,9 @@
# limitations under the License.
from .classifier.configuration_classifier import RewardClassifierConfig as RewardClassifierConfig
from .distributional_value_function.configuration_distributional_value_function import (
DistributionalVFConfig as DistributionalVFConfig,
)
from .factory import (
get_reward_model_class as get_reward_model_class,
make_reward_model as make_reward_model,
@@ -26,6 +29,7 @@ from .topreward.configuration_topreward import TOPRewardConfig as TOPRewardConfi
__all__ = [
# Configuration classes
"DistributionalVFConfig",
"RewardClassifierConfig",
"RobometerConfig",
"SARMConfig",
@@ -0,0 +1,23 @@
# 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.
from .configuration_distributional_value_function import DistributionalVFConfig
from .modeling_distributional_value_function import DistributionalVFRewardModel
from .processor_distributional_value_function import make_distributional_vf_pre_post_processors
__all__ = [
"DistributionalVFConfig",
"DistributionalVFRewardModel",
"make_distributional_vf_pre_post_processors",
]
@@ -0,0 +1,108 @@
# 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.
"""Configuration for RECAP's distributional value function.
Paper: "π*0.6: a VLA That Learns From Experience" (Physical Intelligence, 2025)
https://pi.website/blog/pistar06
Implements the distributional value function V^{pi_ref}(o_t, l) from Section IV-A.
Architecture: the paper uses a 670M-parameter Gemma 3 VLM (the actor is 4B Gemma 3).
We match that scale on PaliGemma (PI05's Gemma 2B backbone) by truncating to 6 Gemma
LM layers and 13 SigLIP vision layers (~670M params), with a [CLS] token and linear
head predicting a categorical distribution over B=201 discrete value bins in [-1, 0].
Training: cross-entropy on HL-Gauss soft targets (or Dirac delta projection),
with optional one-hot targets for terminal states; MC returns normalized per task.
Weights initialized from a pre-trained PI05 actor checkpoint.
"""
from dataclasses import dataclass, field
from lerobot.configs import FeatureType, NormalizationMode
from lerobot.configs.rewards import RewardModelConfig
from lerobot.optim import AdamWConfig, CosineDecayWithWarmupSchedulerConfig
@RewardModelConfig.register_subclass("distributional_value_function")
@dataclass
class DistributionalVFConfig(RewardModelConfig):
"""Configuration for RECAP's distributional value function.
The value function predicts V^{pi_ref}(o_t, l) as a distribution over B discrete
bins spanning [value_support_min, value_support_max]. It is trained with cross-entropy
on HL-Gauss soft targets or Dirac delta projection, derived from Monte Carlo returns
(Eq. 1 in the paper).
Architecture: the paper value function is a 670M Gemma 3 VLM; the actor is 4B Gemma 3.
We use truncated PaliGemma (``num_hidden_layers=6``, ``num_vision_layers=13``) to reach
about 670M params and initialize from the PI05 actor checkpoint.
"""
# Backbone
paligemma_variant: str = "gemma_2b"
num_hidden_layers: int = 6
num_vision_layers: int = 13
# Distributional head
num_value_bins: int = 201
value_support_min: float = -1.0
value_support_max: float = 0.0
hl_gauss_sigma_ratio: float = 5.0
# Target distribution method: "hl_gauss" (default, soft) or "dirac_delta" (C51, hard)
target_method: str = "hl_gauss"
# Whether to use one-hot targets for terminal states (exact return, no smoothing).
# When False, terminal states use the same target method as non-terminal states.
use_one_hot_terminal: bool = True
# Image
image_resolution: tuple[int, int] = (224, 224)
# Tokenizer
tokenizer_max_length: int = 64
# Init from actor (required for first training: provides SigLIP vision tower + Gemma embeddings).
# Pass a PI05 checkpoint path or Hub repo_id here.
# After training, load the value function with RewardModel.from_pretrained() instead.
init_from_actor_path: str = ""
# Normalization
normalization_mapping: dict[str, NormalizationMode] = field(
default_factory=lambda: {
"VISUAL": NormalizationMode.IDENTITY,
"STATE": NormalizationMode.IDENTITY,
}
)
def get_optimizer_preset(self) -> AdamWConfig:
return AdamWConfig(
lr=3e-4,
weight_decay=1e-4,
grad_clip_norm=1.0,
)
def get_scheduler_preset(self) -> CosineDecayWithWarmupSchedulerConfig:
return CosineDecayWithWarmupSchedulerConfig(
num_warmup_steps=500,
num_decay_steps=50000,
)
def validate_features(self) -> None:
if not self.input_features:
return
has_image = any(ft.type == FeatureType.VISUAL for ft in self.input_features.values())
if not has_image:
raise ValueError("DistributionalVFConfig requires at least one VISUAL input feature.")
@@ -0,0 +1,567 @@
# 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.
"""Modeling for RECAP's distributional value function.
Paper: "π*0.6: a VLA That Learns From Experience" (Physical Intelligence, 2025)
https://pi.website/blog/pistar06
Implements the distributional value function V^{pi_ref}(o_t, l) from Section IV-A.
Architecture: the paper uses a 670M-parameter Gemma 3 VLM (the actor is 4B Gemma 3).
We match that scale on PaliGemma (PI05's Gemma 2B backbone) by truncating to 6 Gemma
LM layers and 13 SigLIP vision layers (~670M params), with a [CLS] token and linear
head predicting a categorical distribution over B=201 discrete value bins in [-1, 0].
Inputs: single image observation + task text prompt ("Task: {task}.")
Outputs: softmax distribution over value bins; expected value E[V] for inference.
Training: cross-entropy on HL-Gauss soft targets (or Dirac delta projection),
with optional one-hot targets for terminal states; MC returns normalized per task.
Weight initialization: vision tower, multi-modal projector, token embeddings, and
the first N transformer layers are copied from a pre-trained PI05 actor checkpoint.
"""
from __future__ import annotations
import math
from typing import TYPE_CHECKING, Any
import torch
import torch.nn.functional as F # noqa: N812
from torch import Tensor, nn
from lerobot.rewards.pretrained import PreTrainedRewardModel
from lerobot.utils.import_utils import _transformers_available, require_package
from .configuration_distributional_value_function import DistributionalVFConfig
if TYPE_CHECKING or _transformers_available:
from transformers.models.auto import CONFIG_MAPPING
from transformers.models.gemma import modeling_gemma
from lerobot.policies.pi_gemma import (
PaliGemmaForConditionalGenerationWithPiGemma,
PiGemmaRMSNorm,
_gated_residual,
_get_pi_gemma_decoder_layer_base,
)
else:
CONFIG_MAPPING = None
modeling_gemma = None
PaliGemmaForConditionalGenerationWithPiGemma = None
PiGemmaRMSNorm = None
_gated_residual = None
_get_pi_gemma_decoder_layer_base = None
PALIGEMMA_VOCAB_SIZE = 257152
class DistributionalVFRewardModel(PreTrainedRewardModel):
"""Distributional value function model for RECAP.
Predicts V^{pi_ref}(o_t, l) as a categorical distribution over B bins (default 201).
Trained with cross-entropy on HL-Gauss or Dirac delta targets centered on
per-task normalized Monte Carlo returns.
Architecture: truncated PaliGemma (``num_hidden_layers=6``, ``num_vision_layers=13``),
causal attention, [CLS] token, and Linear(D, num_bins) value head.
The expected value is E[V] = sum(softmax(logits) * bin_centers).
"""
name = "distributional_value_function"
config_class = DistributionalVFConfig
def __init__(self, config: DistributionalVFConfig, **kwargs) -> None:
require_package("transformers", extra="recap")
super().__init__(config)
self.config = config
from transformers.models.gemma.modeling_gemma import GemmaRotaryEmbedding
from lerobot.policies.pi05.modeling_pi05 import get_gemma_config
# Get base dimensions from the paligemma variant (OpenPI config format)
base_config = get_gemma_config(config.paligemma_variant)
hidden_dim = base_config.width
mlp_dim = base_config.mlp_dim
num_layers = config.num_hidden_layers
# HuggingFace GemmaConfig for transformer layers
gemma_config = CONFIG_MAPPING["gemma"](
head_dim=base_config.head_dim,
hidden_size=hidden_dim,
intermediate_size=mlp_dim,
num_attention_heads=base_config.num_heads,
num_hidden_layers=num_layers,
num_key_value_heads=base_config.num_kv_heads,
vocab_size=PALIGEMMA_VOCAB_SIZE,
hidden_activation="gelu_pytorch_tanh",
)
self.gemma_config = gemma_config
self.hidden_dim = hidden_dim
self.num_value_bins = config.num_value_bins
# Single learned [CLS] token for value prediction
self.cls_embedding = nn.Parameter(torch.randn(1, 1, hidden_dim) * 0.02)
# Value projection head: Linear(hidden_dim, num_bins)
self.value_head = nn.Linear(in_features=hidden_dim, out_features=config.num_value_bins)
# Transformer layers (overwritten by _initialize_from_actor on first run)
self.rotary_emb = GemmaRotaryEmbedding(gemma_config)
pi_gemma_decoder_layer_base = _get_pi_gemma_decoder_layer_base()
self.layers = nn.ModuleList(
[pi_gemma_decoder_layer_base(gemma_config, layer_idx=i) for i in range(num_layers)]
)
self.norm = PiGemmaRMSNorm(hidden_dim, eps=gemma_config.rms_norm_eps)
# Vision tower + projector + token embedding (overwritten by _initialize_from_actor on first run)
# PaliGemmaConfig wraps both vision and text configs into a single model
paligemma_config = CONFIG_MAPPING["paligemma"]()
paligemma_config.text_config = gemma_config
paligemma_config.vision_config.image_size = config.image_resolution[0]
paligemma_config.vision_config.intermediate_size = 4304
paligemma_config.vision_config.projection_dim = 2048
paligemma_config.vision_config.projector_hidden_act = "gelu_fast"
paligemma_full = PaliGemmaForConditionalGenerationWithPiGemma(config=paligemma_config)
self.vision_tower = paligemma_full.model.vision_tower
self.multi_modal_projector = paligemma_full.model.multi_modal_projector
self.token_embedding = paligemma_full.model.language_model.embed_tokens
del paligemma_full
# Truncate vision tower to num_vision_layers
if hasattr(self.vision_tower, "vision_model") and hasattr(self.vision_tower.vision_model, "encoder"):
vision_encoder = self.vision_tower.vision_model.encoder
vision_encoder.layers = vision_encoder.layers[: config.num_vision_layers]
# Bin support: evenly spaced centers from value_support_min to value_support_max
bin_centers = torch.linspace(config.value_support_min, config.value_support_max, self.num_value_bins)
self.register_buffer("bin_centers", bin_centers, persistent=False)
bin_width = (config.value_support_max - config.value_support_min) / (self.num_value_bins - 1)
self.hl_gauss_sigma = float(config.hl_gauss_sigma_ratio * bin_width)
# Overwrite with pre-trained PI05 actor weights (first training run only)
if config.init_from_actor_path:
self._initialize_from_actor()
def _initialize_from_actor(self) -> None:
"""Overwrite weights from a pre-trained PI05 actor checkpoint.
Called on first training run only (when init_from_actor_path is set).
"""
from lerobot.policies.pi05.modeling_pi05 import PI05Policy
actor_policy = PI05Policy.from_pretrained(self.config.init_from_actor_path)
actor_model = actor_policy.model
paligemma_model = actor_model.paligemma_with_expert.paligemma
source_language_model = paligemma_model.model.language_model
# Transformer components
self.rotary_emb.load_state_dict(source_language_model.rotary_emb.state_dict())
num_layers = self.gemma_config.num_hidden_layers
for i in range(num_layers):
self.layers[i].load_state_dict(source_language_model.layers[i].state_dict())
self.norm.load_state_dict(source_language_model.norm.state_dict())
# Vision tower (truncate source first, then copy)
source_vision_tower = paligemma_model.model.vision_tower
if hasattr(source_vision_tower, "vision_model") and hasattr(
source_vision_tower.vision_model, "encoder"
):
source_encoder = source_vision_tower.vision_model.encoder
source_encoder.layers = source_encoder.layers[: self.config.num_vision_layers]
self.vision_tower.load_state_dict(source_vision_tower.state_dict())
# Multi-modal projector
self.multi_modal_projector.load_state_dict(paligemma_model.model.multi_modal_projector.state_dict())
# Token embedding table
self.token_embedding.load_state_dict(paligemma_model.model.language_model.embed_tokens.state_dict())
del actor_policy
def embed_image(self, image: Tensor) -> Tensor:
"""Embed images using the value function's SigLIP vision tower.
Args:
image: [batch_size, channels, height, width] preprocessed images in [-1, 1].
Returns:
[batch_size, num_patches, hidden_dim] projected image features.
"""
out_dtype = image.dtype
if image.dtype != torch.float32:
image = image.to(torch.float32)
image_outputs = self.vision_tower(image, return_dict=True)
image_features = self.multi_modal_projector(image_outputs.last_hidden_state)
image_features = image_features / (self.hidden_dim**0.5)
if image_features.dtype != out_dtype:
image_features = image_features.to(out_dtype)
return image_features
def embed_text(self, token_ids: Tensor) -> Tensor:
"""Embed text token IDs using the value function's token embedding table.
Args:
token_ids: [batch_size, seq_len] integer token IDs
Returns:
[batch_size, seq_len, hidden_dim] text embeddings
"""
return self.token_embedding(token_ids)
def _get_cls_embedding(self, batch_size: int) -> Tensor:
"""Get [CLS] token embedding expanded to batch size.
Args:
batch_size: number of samples in the batch.
Returns:
[batch_size, 1, hidden_dim] learned [CLS] embedding.
"""
return self.cls_embedding.expand(batch_size, -1, -1)
def forward_value(
self, vision_features: Tensor, text_embeddings: Tensor, text_padding_mask: Tensor
) -> dict[str, Tensor]:
"""Core forward pass through the distributional value function.
Args:
vision_features: [batch_size, num_patches, hidden_dim]
text_embeddings: [batch_size, seq_len, hidden_dim]
text_padding_mask: [batch_size, seq_len] boolean mask for text tokens
Returns:
logits: [batch_size, num_value_bins]
probs: [batch_size, num_value_bins]
value: [batch_size, 1]
"""
from lerobot.utils.constants import OPENPI_ATTENTION_MASK_VALUE
batch_size = text_embeddings.shape[0]
device = text_embeddings.device
# Build sequence: [vision, text, CLS]
cls_embedding = self._get_cls_embedding(batch_size)
hidden_states = torch.cat([vision_features, text_embeddings, cls_embedding], dim=1)
# Build causal attention mask
vision_len = vision_features.shape[1]
vision_padding_mask = torch.ones(batch_size, vision_len, dtype=torch.bool, device=device)
cls_padding_mask = torch.ones(batch_size, 1, dtype=torch.bool, device=device)
full_padding_mask = torch.cat([vision_padding_mask, text_padding_mask, cls_padding_mask], dim=1)
full_seq_len = full_padding_mask.shape[1]
# Causal mask
causal_mask = torch.tril(torch.ones(full_seq_len, full_seq_len, device=device, dtype=torch.bool))
# Combine causal mask with padding mask
padding_mask_4d = full_padding_mask[:, None, None, :].expand(
batch_size, 1, full_seq_len, full_seq_len
)
attention_mask = causal_mask[None, None, :, :] & padding_mask_4d
attention_mask = torch.where(attention_mask, 0.0, OPENPI_ATTENTION_MASK_VALUE)
position_ids = torch.cumsum(full_padding_mask.long(), dim=1) - 1
cos, sin = self.rotary_emb(hidden_states, position_ids)
for layer in self.layers:
norm_output = layer.input_layernorm(hidden_states, cond=None)
if isinstance(norm_output, tuple):
hidden_states_normed, gate = norm_output
else:
hidden_states_normed, gate = norm_output, None
input_shape = hidden_states_normed.shape[:-1]
hidden_shape = (*input_shape, -1, layer.self_attn.head_dim)
query_states = layer.self_attn.q_proj(hidden_states_normed).view(hidden_shape).transpose(1, 2)
key_states = layer.self_attn.k_proj(hidden_states_normed).view(hidden_shape).transpose(1, 2)
value_states = layer.self_attn.v_proj(hidden_states_normed).view(hidden_shape).transpose(1, 2)
query_states, key_states = modeling_gemma.apply_rotary_pos_emb(
query_states, key_states, cos, sin, unsqueeze_dim=1
)
attention_output, _ = modeling_gemma.eager_attention_forward(
layer.self_attn,
query_states,
key_states,
value_states,
attention_mask,
layer.self_attn.scaling,
)
attention_output = attention_output.reshape(batch_size, -1, self.gemma_config.hidden_size)
if attention_output.dtype != layer.self_attn.o_proj.weight.dtype:
attention_output = attention_output.to(layer.self_attn.o_proj.weight.dtype)
projected_attention = layer.self_attn.o_proj(attention_output)
if gate is not None:
projected_attention = _gated_residual(hidden_states, projected_attention, gate)
else:
projected_attention = hidden_states + projected_attention
after_attention_residual = projected_attention.clone()
norm_output = layer.post_attention_layernorm(projected_attention, cond=None)
if isinstance(norm_output, tuple):
mlp_input, gate = norm_output
else:
mlp_input, gate = norm_output, None
mlp_output = layer.mlp(mlp_input)
if gate is not None:
hidden_states = _gated_residual(after_attention_residual, mlp_output, gate)
else:
hidden_states = after_attention_residual + mlp_output
hidden_states = self.norm(hidden_states)
if isinstance(hidden_states, tuple):
hidden_states = hidden_states[0]
# Extract [CLS] token (last position in the sequence)
cls_hidden_state = hidden_states[:, -1, :] # [batch_size, hidden_dim]
# Value head: Linear(hidden_dim, num_bins) -> logits
value_logits = self.value_head(cls_hidden_state) # [batch_size, num_value_bins]
value_probs = F.softmax(value_logits, dim=-1)
predicted_value = (value_probs * self.bin_centers.to(dtype=value_probs.dtype)).sum(
dim=-1, keepdim=True
)
return {"logits": value_logits, "probs": value_probs, "value": predicted_value}
def hl_gauss_target(self, target_value: Tensor) -> Tensor:
"""HL-Gauss soft target distribution.
Places a Gaussian N(target, sigma^2) over the bin support and computes
per-bin probabilities as CDF differences at bin edges, normalized to sum to 1.
Reference: Farebrother et al. 2024, "Stop Regressing: Training Value
Functions via Classification for Scalable Deep RL", Section 3.1.
arXiv:2403.03950
Args:
target_value: [batch_size] or [batch_size, 1] target values.
Returns:
[batch_size, num_value_bins] target probability distribution.
"""
if target_value.ndim == 2:
target_value = target_value.squeeze(-1)
target_value = target_value.to(dtype=self.bin_centers.dtype)
# Bin edges: half a bin-width outside the first/last center
bin_width = (self.config.value_support_max - self.config.value_support_min) / (
self.num_value_bins - 1
)
support_edges = torch.linspace(
self.config.value_support_min - bin_width / 2,
self.config.value_support_max + bin_width / 2,
self.num_value_bins + 1,
device=target_value.device,
dtype=target_value.dtype,
)
# CDF of N(target, sigma^2) evaluated at each edge
cdf_at_edges = 0.5 * (
1.0
+ torch.erf(
(support_edges.unsqueeze(0) - target_value.unsqueeze(-1))
/ (self.hl_gauss_sigma * math.sqrt(2))
)
) # [batch_size, num_bins + 1]
# Normalize: z = cdf(max_edge) - cdf(min_edge)
normalization_constant = (cdf_at_edges[:, -1] - cdf_at_edges[:, 0]).unsqueeze(-1).clamp(min=1e-10)
# Bin probabilities = differences of consecutive CDF values, normalized
bin_probabilities = (cdf_at_edges[:, 1:] - cdf_at_edges[:, :-1]) / normalization_constant
return bin_probabilities
def dirac_delta_target(self, target_value: Tensor) -> Tensor:
"""Dirac delta (C51) projection: split probability between two nearest bins.
Standard distributional RL projection from Bellemare et al. 2017.
"A Distributional Perspective on Reinforcement Learning"
arXiv:1707.06887
Args:
target_value: [batch_size] or [batch_size, 1] target values.
Returns:
[batch_size, num_value_bins] target probability distribution.
"""
if target_value.ndim == 2:
target_value = target_value.squeeze(-1)
target_value = target_value.clamp(self.config.value_support_min, self.config.value_support_max)
target_value = target_value.to(dtype=self.bin_centers.dtype)
bin_width = self.bin_centers[1] - self.bin_centers[0]
normalized_position = (target_value - self.config.value_support_min) / bin_width
lower_bin_idx = normalized_position.floor().long().clamp(0, self.num_value_bins - 1)
upper_bin_idx = normalized_position.ceil().long().clamp(0, self.num_value_bins - 1)
weight_upper = normalized_position - lower_bin_idx.float()
weight_lower = upper_bin_idx.float() - normalized_position
same_bin = lower_bin_idx == upper_bin_idx
weight_upper = torch.where(same_bin, torch.zeros_like(weight_upper), weight_upper)
weight_lower = torch.where(same_bin, torch.ones_like(weight_lower), weight_lower)
batch_size = target_value.shape[0]
target_distribution = torch.zeros(batch_size, self.num_value_bins, device=target_value.device)
batch_indices = torch.arange(batch_size, device=target_value.device)
target_distribution[batch_indices, lower_bin_idx] += weight_lower
target_distribution[batch_indices, upper_bin_idx] += weight_upper
return target_distribution
def one_hot_target(self, target_value: Tensor) -> Tensor:
"""One-hot target for terminal states (exact return, no smoothing).
Args:
target_value: [batch_size] or [batch_size, 1] target values.
Returns:
[batch_size, num_value_bins] one-hot distribution at the nearest bin.
"""
if target_value.ndim == 2:
target_value = target_value.squeeze(-1)
target_value = target_value.to(dtype=self.bin_centers.dtype)
nearest_bin_idx = torch.argmin(
torch.abs(self.bin_centers.unsqueeze(0) - target_value.unsqueeze(-1)), dim=-1
)
return F.one_hot(nearest_bin_idx, num_classes=self.num_value_bins).to(dtype=self.bin_centers.dtype)
def compute_target_distribution(
self,
target_value: Tensor,
is_terminal: Tensor,
method: str = "hl_gauss",
use_one_hot_terminal: bool = True,
) -> Tensor:
"""Compute target distribution using configured method.
Args:
target_value: [batch_size] scalar return targets
is_terminal: [batch_size] boolean terminal flags
method: "hl_gauss" or "dirac_delta"
use_one_hot_terminal: if True, terminal states get one-hot targets
(exact return, no smoothing). If False, all states use the same method.
Returns:
[batch_size, num_value_bins] target probability distribution
"""
if method == "hl_gauss":
base_distribution = self.hl_gauss_target(target_value)
elif method == "dirac_delta":
base_distribution = self.dirac_delta_target(target_value)
else:
raise ValueError(f"Unknown target method: {method}. Use 'hl_gauss' or 'dirac_delta'.")
if not use_one_hot_terminal:
return base_distribution
terminal_distribution = self.one_hot_target(target_value)
return torch.where(is_terminal[:, None].bool(), terminal_distribution, base_distribution)
def forward(self, batch: dict[str, Tensor]) -> tuple[Tensor, dict[str, Any]]:
"""Training forward pass — computes cross-entropy loss against MC return targets.
The batch is expected to be preprocessed by the processor pipeline.
Keys expected in batch:
- observation.images.*: [B, C, H, W] preprocessed images
- observation.language_tokens: [B, seq_len] tokenized task prompt
- observation.language_attention_mask: [B, seq_len] padding mask
- mc_return: [B] normalized Monte Carlo return targets in (-1, 0)
- is_terminal: [B] boolean terminal flags
Returns:
(loss, output_dict) where loss is scalar cross-entropy
"""
from lerobot.utils.constants import OBS_IMAGES, OBS_LANGUAGE_ATTENTION_MASK, OBS_LANGUAGE_TOKENS
# Get first image key from batch
image_keys = [k for k in batch if k.startswith(f"{OBS_IMAGES}.") or k == OBS_IMAGES]
if not image_keys:
raise KeyError(f"No image keys found in batch. Expected keys starting with '{OBS_IMAGES}.'")
images = batch[image_keys[0]]
token_ids = batch[OBS_LANGUAGE_TOKENS]
text_padding_mask = batch[OBS_LANGUAGE_ATTENTION_MASK].bool()
mc_return = batch["mc_return"]
is_terminal = batch["is_terminal"]
# Embed observations
vision_features = self.embed_image(images)
text_embeddings = self.embed_text(token_ids)
# Forward through value function transformer
vf_output = self.forward_value(vision_features, text_embeddings, text_padding_mask)
value_logits = vf_output["logits"]
predicted_value = vf_output["value"]
# Compute target distribution
target_distribution = self.compute_target_distribution(
mc_return,
is_terminal,
method=self.config.target_method,
use_one_hot_terminal=self.config.use_one_hot_terminal,
)
# Cross-entropy loss (Eq. 1 in pi*0.6 paper)
log_probs = F.log_softmax(value_logits, dim=-1)
loss = -(target_distribution * log_probs).sum(dim=-1).mean()
output_dict = {
"loss": loss.item(),
"predicted_value_mean": predicted_value.mean().item(),
"mc_return_mean": mc_return.mean().item(),
}
return loss, output_dict
def compute_reward(self, batch: dict[str, Tensor]) -> Tensor:
"""Compute V(s) for a batch of observations. Used for advantage scoring.
Args:
batch: preprocessed batch with images and tokenized text
Returns:
[batch_size] tensor of predicted values V(s)
"""
from lerobot.utils.constants import OBS_IMAGES, OBS_LANGUAGE_ATTENTION_MASK, OBS_LANGUAGE_TOKENS
image_keys = [k for k in batch if k.startswith(f"{OBS_IMAGES}.") or k == OBS_IMAGES]
if not image_keys:
raise KeyError(f"No image keys found in batch. Expected keys starting with '{OBS_IMAGES}.'")
images = batch[image_keys[0]]
token_ids = batch[OBS_LANGUAGE_TOKENS]
text_padding_mask = batch[OBS_LANGUAGE_ATTENTION_MASK].bool()
vision_features = self.embed_image(images)
text_embeddings = self.embed_text(token_ids)
vf_output = self.forward_value(vision_features, text_embeddings, text_padding_mask)
return vf_output["value"].squeeze(-1) # [batch_size]
@@ -0,0 +1,235 @@
# 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.
"""Processor for RECAP's distributional value function.
Paper: "π*0.6: a VLA That Learns From Experience" (Physical Intelligence, 2025)
https://pi.website/blog/pistar06
Prepares inputs for V^{pi_ref}(o_t, l): single image observation and task text only.
1. Image preprocessing (resize-with-pad + normalize to [-1, 1]) for SigLIP
2. Task prompt formatting ("Task: {task}.") and tokenization via PaliGemma tokenizer
Training targets (mc_return, is_terminal) are NOT routed through the processor.
They are dataset columns read directly from the batch in the model's forward().
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import torch
from torch import Tensor
from lerobot.configs import FeatureType, PipelineFeatureType, PolicyFeature
from lerobot.processor import (
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
ProcessorStep,
ProcessorStepRegistry,
RenameObservationsProcessorStep,
TokenizerProcessorStep,
batch_to_transition,
policy_action_to_transition,
transition_to_batch,
)
from lerobot.processor.converters import to_tensor
from lerobot.types import EnvTransition, TransitionKey
from lerobot.utils.constants import (
OBS_IMAGES,
POLICY_POSTPROCESSOR_DEFAULT_NAME,
POLICY_PREPROCESSOR_DEFAULT_NAME,
)
from .configuration_distributional_value_function import DistributionalVFConfig
PALIGEMMA_TOKENIZER_NAME = "google/paligemma-3b-pt-224"
@ProcessorStepRegistry.register(name="distributional_vf_prepare_task_prompt")
@dataclass
class DistributionalVFPrepareTaskPromptStep(ProcessorStep):
"""Format the task string for the distributional value function.
The value function receives only visual observations and task text.
Builds prompt: "Task: {task}."
"""
task_key: str = "task"
def __call__(self, transition: EnvTransition) -> EnvTransition:
transition = transition.copy()
complementary_data = transition.get(TransitionKey.COMPLEMENTARY_DATA, {})
tasks = complementary_data.get(self.task_key)
if tasks is None:
raise ValueError("No task found in complementary data")
if isinstance(tasks, str):
tasks = [tasks]
full_prompts = []
for task in tasks:
cleaned_text = task.strip().replace("_", " ").replace("\n", " ")
full_prompts.append(f"Task: {cleaned_text}.")
new_complementary_data = dict(complementary_data)
new_complementary_data[self.task_key] = full_prompts
transition[TransitionKey.COMPLEMENTARY_DATA] = new_complementary_data
return transition
def transform_features(
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
return features
def get_config(self) -> dict[str, Any]:
return {"task_key": self.task_key}
@ProcessorStepRegistry.register(name="distributional_vf_image_preprocessor")
@dataclass
class DistributionalVFImagePreprocessorStep(ProcessorStep):
"""Resize and normalize images for the value function's SigLIP vision tower.
Expects float images in [0, 1].
- Resize-with-pad to ``image_resolution`` (preserves aspect ratio)
- Scale to [-1, 1] for SigLIP
"""
image_resolution: tuple[int, int] = (224, 224)
image_keys: tuple[str, ...] | None = None
def __call__(self, transition: EnvTransition) -> EnvTransition:
from lerobot.policies.pi05.modeling_pi05 import resize_with_pad_torch
observation = transition.get(TransitionKey.OBSERVATION)
if not isinstance(observation, dict):
raise ValueError("DistributionalVFImagePreprocessorStep requires an observation dict")
image_keys = self.image_keys or tuple(
key for key in observation if key == OBS_IMAGES or key.startswith(f"{OBS_IMAGES}.")
)
if not image_keys:
raise KeyError(
f"Distributional value function expected image keys under {OBS_IMAGES!r} in observation"
)
new_observation = dict(observation)
for image_key in image_keys:
image = new_observation[image_key]
if not isinstance(image, Tensor):
image = to_tensor(image)
if image.dtype != torch.float32:
image = image.to(torch.float32)
is_channels_first = image.ndim == 4 and image.shape[1] == 3
if is_channels_first:
image = image.permute(0, 2, 3, 1)
if image.shape[1:3] != self.image_resolution:
image = resize_with_pad_torch(image, *self.image_resolution)
image = image * 2.0 - 1.0
if is_channels_first:
image = image.permute(0, 3, 1, 2)
new_observation[image_key] = image
new_transition = transition.copy()
new_transition[TransitionKey.OBSERVATION] = new_observation
return new_transition
def transform_features(
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
return features
def get_config(self) -> dict[str, Any]:
return {
"image_resolution": self.image_resolution,
"image_keys": list(self.image_keys) if self.image_keys is not None else None,
}
def _visual_image_keys(config: DistributionalVFConfig) -> tuple[str, ...]:
return tuple(
feature_name
for feature_name, feature in config.input_features.items()
if feature.type == FeatureType.VISUAL
)
def make_distributional_vf_pre_post_processors(
config: DistributionalVFConfig,
dataset_stats: dict[str, dict[str, torch.Tensor]] | None = None,
) -> tuple[
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]],
PolicyProcessorPipeline[PolicyAction, PolicyAction],
]:
"""Create pre/post processors for the distributional value function.
Preprocessor steps:
1. Rename observations (no-op by default)
2. Add a batch dimension
3. Normalize features (images use identity, so they stay in [0, 1])
4. Format task prompt: "Task: {task}."
5. Tokenize with the PaliGemma tokenizer
6. Resize-with-pad and scale images to [-1, 1] for SigLIP
7. Move tensors to the configured device
Training targets (mc_return, is_terminal) are not processed here.
The model reads them directly from the batch in forward().
The postprocessor is a no-op because the value function does not need
action postprocessing.
"""
image_keys = _visual_image_keys(config)
preprocessor = PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=[
RenameObservationsProcessorStep(rename_map={}),
AddBatchDimensionProcessorStep(),
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
DistributionalVFPrepareTaskPromptStep(),
TokenizerProcessorStep(
tokenizer_name=PALIGEMMA_TOKENIZER_NAME,
max_length=config.tokenizer_max_length,
padding_side="right",
padding="max_length",
),
DistributionalVFImagePreprocessorStep(
image_resolution=config.image_resolution,
image_keys=image_keys or None,
),
DeviceProcessorStep(device=config.device or "cpu"),
],
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
to_transition=batch_to_transition,
to_output=transition_to_batch,
)
postprocessor = PolicyProcessorPipeline(
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
)
return preprocessor, postprocessor
+19
View File
@@ -24,6 +24,7 @@ from lerobot.configs.rewards import RewardModelConfig
from lerobot.processor import PolicyAction, PolicyProcessorPipeline
from .classifier.configuration_classifier import RewardClassifierConfig
from .distributional_value_function.configuration_distributional_value_function import DistributionalVFConfig
from .pretrained import PreTrainedRewardModel
from .robometer.configuration_robometer import RobometerConfig
from .sarm.configuration_sarm import SARMConfig
@@ -63,6 +64,12 @@ def get_reward_model_class(name: str) -> type[PreTrainedRewardModel]:
from lerobot.rewards.topreward.modeling_topreward import TOPRewardModel
return TOPRewardModel
elif name == "distributional_value_function":
from lerobot.rewards.distributional_value_function.modeling_distributional_value_function import (
DistributionalVFRewardModel,
)
return DistributionalVFRewardModel
else:
try:
return _get_reward_model_cls_from_name(name=name)
@@ -96,6 +103,8 @@ def make_reward_model_config(reward_type: str, **kwargs) -> RewardModelConfig:
return RobometerConfig(**kwargs)
elif reward_type == "topreward":
return TOPRewardConfig(**kwargs)
elif reward_type == "distributional_value_function":
return DistributionalVFConfig(**kwargs)
else:
try:
config_cls = RewardModelConfig.get_choice_class(reward_type)
@@ -192,6 +201,16 @@ def make_reward_pre_post_processors(
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(reward_cfg, DistributionalVFConfig):
from lerobot.rewards.distributional_value_function.processor_distributional_value_function import (
make_distributional_vf_pre_post_processors,
)
return make_distributional_vf_pre_post_processors(
config=reward_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
else:
try:
processors = _make_processors_from_reward_model_config(
+8
View File
@@ -106,6 +106,8 @@ class DAggerKeyboardConfig:
pause_resume: str = "space"
correction: str = "tab"
upload: str = "enter"
success: str = "s"
failure: str = "f"
@dataclass
@@ -119,6 +121,8 @@ class DAggerPedalConfig:
pause_resume: str = "KEY_A"
correction: str = "KEY_B"
upload: str = "KEY_C"
success: str = "KEY_D"
failure: str = "KEY_E"
@RolloutStrategyConfig.register_subclass("episodic")
@@ -165,6 +169,10 @@ class DAggerStrategyConfig(RolloutStrategyConfig):
2. **correction** toggle human correction recording.
3. **upload** push dataset to hub on demand (corrections-only mode).
Episode success labeling:
4. **success** mark current episode as successful.
5. **failure** mark current episode as failed.
When ``record_autonomous=False`` (default) only human-correction windows
are recorded each correction becomes its own episode. Set to ``True``
to record both autonomous and correction frames with size-based episode
+5
View File
@@ -350,6 +350,11 @@ def build_rollout_context(
"shape": (1,),
"names": None,
}
dataset_features["next.success"] = {
"dtype": "bool",
"shape": (1,),
"names": None,
}
repo_name = cfg.dataset.repo_id.split("/", 1)[-1]
if not repo_name.startswith("rollout_"):
+141 -16
View File
@@ -112,6 +112,14 @@ class DAggerEvents:
# Session-level flags
self.stop_recording = Event()
self.upload_requested = Event()
# Set when operator presses success/failure key to end the current episode.
self.save_episode_requested = Event()
# Episode success labeling
self._episode_success: bool | None = None
# Episode success labeling
self._episode_success: bool | None = None
# -- Thread-safe phase access ------------------------------------------
@@ -155,7 +163,43 @@ class DAggerEvents:
with self._lock:
self._phase = DAggerPhase.AUTONOMOUS
self._pending_transition = None
self._episode_success = None
self.upload_requested.clear()
self.save_episode_requested.clear()
def mark_success(self) -> None:
"""Mark the current episode as successful (called from input threads)."""
with self._lock:
self._episode_success = True
def mark_failure(self) -> None:
"""Mark the current episode as failed (called from input threads)."""
with self._lock:
self._episode_success = False
def consume_episode_success(self) -> bool | None:
"""Consume and reset the episode success label. Returns None if unlabeled."""
with self._lock:
result = self._episode_success
self._episode_success = None
return result
def mark_success(self) -> None:
"""Mark the current episode as successful (called from input threads)."""
with self._lock:
self._episode_success = True
def mark_failure(self) -> None:
"""Mark the current episode as failed (called from input threads)."""
with self._lock:
self._episode_success = False
def consume_episode_success(self) -> bool | None:
"""Consume and reset the episode success label. Returns None if unlabeled."""
with self._lock:
result = self._episode_success
self._episode_success = None
return result
# ---------------------------------------------------------------------------
@@ -186,12 +230,20 @@ def _init_dagger_keyboard(events: DAggerEvents, cfg: DAggerKeyboardConfig):
events.request_transition(key_to_event[name])
if name == cfg.upload:
events.upload_requested.set()
if name == cfg.success:
events.mark_success()
events.save_episode_requested.set()
logger.info("Episode marked as SUCCESS — saving")
if name == cfg.failure:
events.mark_failure()
events.save_episode_requested.set()
logger.info("Episode marked as FAILURE — saving")
return create_key_listener(
dispatch,
controls_help=(
f"pause_resume='{cfg.pause_resume}', correction='{cfg.correction}', "
f"upload='{cfg.upload}', ESC=stop"
f"upload='{cfg.upload}', success='{cfg.success}', failure='{cfg.failure}', ESC=stop"
),
)
@@ -211,6 +263,12 @@ def _init_dagger_pedal(events: DAggerEvents, cfg: DAggerPedalConfig):
events.request_transition(code_to_event[code])
if code == cfg.upload:
events.upload_requested.set()
if code == cfg.success:
events.mark_success()
logger.info("Episode marked as SUCCESS (pedal)")
if code == cfg.failure:
events.mark_failure()
logger.info("Episode marked as FAILURE (pedal)")
logger.info("Initializing DAgger foot pedal listener (device=%s)", cfg.device_path)
return start_pedal_listener(on_press, device_path=cfg.device_path)
@@ -313,6 +371,31 @@ class DAggerStrategy(RolloutStrategy):
)
logger.info("DAgger strategy teardown complete")
# ------------------------------------------------------------------
# Episode success labeling
# ------------------------------------------------------------------
def _stamp_episode_success(self, dataset) -> None:
"""Set next.success on the terminal frame based on operator label.
Called just before save_episode(). If the operator pressed the success
key during this episode, the last frame's next.success is set to True.
Otherwise all frames remain False (unlabeled = assumed failure).
"""
buf = dataset.writer.episode_buffer
if buf is None:
return
success_buf = buf.get("next.success")
if not success_buf:
return
label = self._events.consume_episode_success()
if label:
success_buf[-1] = np.array([True], dtype=bool)
logger.info("Terminal frame stamped next.success=True")
# ------------------------------------------------------------------
# Continuous recording mode (record_autonomous=True)
# ------------------------------------------------------------------
@@ -350,7 +433,12 @@ class DAggerStrategy(RolloutStrategy):
episode_start = time.perf_counter()
episodes_since_push = 0
episode_duration_s = self._episode_duration_s
logger.info("DAgger continuous recording started (episode_duration=%.0fs)", episode_duration_s)
num_episodes = self.config.num_episodes
logger.info(
"DAgger continuous recording started (episode_duration=%.0fs, target=%s eps)",
episode_duration_s,
num_episodes if num_episodes is not None else "",
)
with VideoEncodingManager(dataset):
try:
@@ -399,6 +487,7 @@ class DAggerStrategy(RolloutStrategy):
**action_frame,
"task": task_str,
"intervention": np.array([True], dtype=bool),
"next.success": np.array([False], dtype=bool),
}
dataset.add_frame(frame)
record_tick += 1
@@ -427,23 +516,32 @@ class DAggerStrategy(RolloutStrategy):
**action_frame,
"task": task_str,
"intervention": np.array([False], dtype=bool),
"next.success": np.array([False], dtype=bool),
}
dataset.add_frame(frame)
record_tick += 1
# Episode rotation derived from the video file-size target.
# Saving is deferred while a correction is ongoing so the
# episode boundary lands on a clean autonomous frame.
# Episode rotation: either the operator pressed success/failure,
# or the video file-size target was reached.
# Defer the save while a correction is ongoing so the episode
# boundary lands on a clean autonomous frame. The event stays
# set until we actually save, so it won't be lost.
manual_save = events.save_episode_requested.is_set()
elapsed = time.perf_counter() - episode_start
if elapsed >= episode_duration_s and phase != DAggerPhase.CORRECTING:
if (manual_save or elapsed >= episode_duration_s) and phase != DAggerPhase.CORRECTING:
if manual_save:
events.save_episode_requested.clear()
with self._episode_lock:
self._stamp_episode_success(dataset)
dataset.save_episode()
episodes_since_push += 1
self._needs_push.set()
save_reason = "manual save" if manual_save else f"elapsed {elapsed:.1f}s"
logger.info(
"Episode saved (total: %d, elapsed: %.1fs)",
"Episode saved (%s, total: %d)",
save_reason,
dataset.num_episodes,
elapsed,
)
log_say(f"Episode {dataset.num_episodes} saved", play_sounds)
@@ -451,6 +549,25 @@ class DAggerStrategy(RolloutStrategy):
self._background_push(dataset, cfg)
episodes_since_push = 0
if num_episodes is not None and dataset.num_episodes >= num_episodes:
logger.info("Target episode count reached (%d), stopping session", num_episodes)
log_say(f"All {num_episodes} episodes collected", play_sounds)
events.stop_recording.set()
break
# Pause after manual save: stop the policy, return robot to
# initial position, and wait for the operator to reset the
# environment and press SPACE.
if manual_save:
engine.pause()
events.phase = DAggerPhase.PAUSED
self._return_to_initial_position(ctx.hardware)
last_action = None
logger.info(
"Episode saved — paused for environment reset. Press SPACE to start next episode."
)
log_say("Reset the environment, then press space", play_sounds)
episode_start = time.perf_counter()
dt = time.perf_counter() - loop_start
@@ -465,10 +582,13 @@ class DAggerStrategy(RolloutStrategy):
logger.info("DAgger continuous control loop ended — pausing engine")
engine.pause()
with contextlib.suppress(Exception):
with self._episode_lock:
dataset.save_episode()
self._needs_push.set()
logger.info("Final in-progress episode saved")
buf = dataset.writer.episode_buffer
if buf and any(len(v) > 0 for v in buf.values() if isinstance(v, list)):
with self._episode_lock:
self._stamp_episode_success(dataset)
dataset.save_episode()
self._needs_push.set()
logger.info("Final in-progress episode saved")
# ------------------------------------------------------------------
# Corrections-only mode (record_autonomous=False)
@@ -540,6 +660,7 @@ class DAggerStrategy(RolloutStrategy):
# Correction ended -> save episode (blocking if not streaming)
if old_phase == DAggerPhase.CORRECTING and new_phase == DAggerPhase.PAUSED:
with self._episode_lock:
self._stamp_episode_success(dataset)
dataset.save_episode()
recorded += 1
self._needs_push.set()
@@ -581,6 +702,7 @@ class DAggerStrategy(RolloutStrategy):
**action_frame,
"task": task_str,
"intervention": np.array([True], dtype=bool),
"next.success": np.array([False], dtype=bool),
}
)
record_tick += 1
@@ -614,10 +736,13 @@ class DAggerStrategy(RolloutStrategy):
logger.info("DAgger corrections-only loop ended — pausing engine")
engine.pause()
with contextlib.suppress(Exception):
with self._episode_lock:
dataset.save_episode()
self._needs_push.set()
logger.info("Final in-progress episode saved")
buf = dataset.writer.episode_buffer
if buf and any(len(v) > 0 for v in buf.values() if isinstance(v, list)):
with self._episode_lock:
self._stamp_episode_success(dataset)
dataset.save_episode()
self._needs_push.set()
logger.info("Final in-progress episode saved")
# ------------------------------------------------------------------
# State-machine transition side-effects
+34 -13
View File
@@ -34,6 +34,7 @@ from lerobot.annotations.steerable_pipeline.config import AnnotationPipelineConf
from lerobot.annotations.steerable_pipeline.executor import Executor
from lerobot.annotations.steerable_pipeline.frames import make_frame_provider
from lerobot.annotations.steerable_pipeline.modules import (
AdvantageModule,
GeneralVqaModule,
InterjectionsAndSpeechModule,
PlanSubtasksMemoryModule,
@@ -63,17 +64,23 @@ def annotate(cfg: AnnotationPipelineConfig) -> None:
root = _resolve_root(cfg)
logger.info("annotate: root=%s", root)
vlm = make_vlm_client(cfg.vlm)
frame_provider = make_frame_provider(root, camera_key=cfg.vlm.camera_key, video_backend=cfg.video_backend)
needs_vlm = cfg.plan.enabled or cfg.interjections.enabled or cfg.vqa.enabled
vlm = make_vlm_client(cfg.vlm) if needs_vlm else None
frame_provider = (
make_frame_provider(root, camera_key=cfg.vlm.camera_key, video_backend=cfg.video_backend)
if needs_vlm
else None
)
# Surface the resolved cameras up front so a silent vqa-module no-op
# is obvious in job output rather than discovered post-hoc by counting
# parquet rows.
cam_keys = list(getattr(frame_provider, "camera_keys", []) or [])
logger.info(
"annotate: frame_provider default camera=%r, all cameras=%s",
getattr(frame_provider, "camera_key", None),
cam_keys,
)
cam_keys = list(getattr(frame_provider, "camera_keys", []) or []) if frame_provider else []
if frame_provider:
logger.info(
"annotate: frame_provider default camera=%r, all cameras=%s",
getattr(frame_provider, "camera_key", None),
cam_keys,
)
if cfg.vqa.enabled and not cam_keys:
logger.warning(
"annotate: the vqa module is enabled but no cameras were "
@@ -81,14 +88,27 @@ def annotate(cfg: AnnotationPipelineConfig) -> None:
"meta/info.json for observation.images.* features, or pass "
"--vlm.camera_key=<key> to seed the cameras list."
)
plan = PlanSubtasksMemoryModule(vlm=vlm, config=cfg.plan, frame_provider=frame_provider)
interjections = InterjectionsAndSpeechModule(
vlm=vlm, config=cfg.interjections, seed=cfg.seed, frame_provider=frame_provider
plan = (
PlanSubtasksMemoryModule(vlm=vlm, config=cfg.plan, frame_provider=frame_provider)
if needs_vlm
else None
)
vqa = GeneralVqaModule(vlm=vlm, config=cfg.vqa, seed=cfg.seed, frame_provider=frame_provider)
interjections = (
InterjectionsAndSpeechModule(
vlm=vlm, config=cfg.interjections, seed=cfg.seed, frame_provider=frame_provider
)
if needs_vlm
else None
)
vqa = (
GeneralVqaModule(vlm=vlm, config=cfg.vqa, seed=cfg.seed, frame_provider=frame_provider)
if needs_vlm
else None
)
advantage = AdvantageModule(config=cfg.advantage)
writer = LanguageColumnsWriter()
validator = StagingValidator(
dataset_camera_keys=tuple(getattr(frame_provider, "camera_keys", []) or []) or None,
dataset_camera_keys=tuple(cam_keys) or None,
)
executor = Executor(
@@ -96,6 +116,7 @@ def annotate(cfg: AnnotationPipelineConfig) -> None:
plan=plan,
interjections=interjections,
vqa=vqa,
advantage=advantage,
writer=writer,
validator=validator,
)
@@ -0,0 +1,382 @@
#!/usr/bin/env python
# Copyright 2026 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.
"""Compute per-frame ``is_terminal`` and ``mc_return`` for a LeRobot dataset.
Implements the sparse reward function from pi*0.6 / RECAP (Eq. 5):
r_t = -1 for non-terminal steps
r_T = 0 for terminal success
r_T = -C_fail for terminal failure
Monte Carlo returns are the cumulative sum from each step to the end of
the episode, normalized by ``max_episode_length`` so that values are bounded
to approximately (-1, 0).
The columns are written directly into the dataset's parquet data shards as
flat per-frame scalars. These serve as training targets for the distributional
value function.
Usage:
# Compute returns using the default "next.success" column (from lerobot-eval/rollout)
lerobot-compute-returns \\
--dataset-repo-id lerobot/aloha_sim_insertion_human_image
# Override: treat all episodes as successful (demo-only datasets)
lerobot-compute-returns \\
--dataset-repo-id lerobot/aloha_sim_insertion_human_image \\
--default-success true
# Custom success key, failure penalty, and discount
lerobot-compute-returns \\
--dataset-repo-id my_org/my_dataset \\
--success-key episode_success \\
--c-fail 100 \\
--gamma 0.99
"""
from __future__ import annotations
import argparse
import json
import logging
from dataclasses import dataclass, field
from pathlib import Path
import numpy as np
import pyarrow as pa
import pyarrow.parquet as pq
from tqdm import tqdm
logger = logging.getLogger(__name__)
IS_TERMINAL_COL = "is_terminal"
MC_RETURN_COL = "mc_return"
@dataclass
class ComputeReturnsConfig:
"""Configuration for the returns computation script."""
dataset_repo_id: str = ""
root: str | None = None
success_key: str = "next.success"
default_success: bool | None = None
max_episode_length: int | None = None
c_fail: float = 50.0
gamma: float = 1.0
episodes: list[int] = field(default_factory=list)
force: bool = False
def _get_episode_success(
episode_table: pa.Table,
success_key: str,
default_success: bool | None,
) -> bool:
"""Determine whether an episode was successful.
Priority:
1. If ``default_success`` is set, use it unconditionally.
2. Look for ``success_key`` in the parquet columns and reduce with any().
3. Fall back to True (assume success for demo datasets).
"""
if default_success is not None:
return default_success
if success_key in episode_table.column_names:
col = episode_table.column(success_key)
for val in col:
py_val = val.as_py()
if isinstance(py_val, bool) and py_val:
return True
if isinstance(py_val, (int, float)) and py_val:
return True
return False
return True
def compute_episode_returns(
num_frames: int,
success: bool,
c_fail: float,
gamma: float,
max_episode_length: int,
) -> tuple[np.ndarray, np.ndarray]:
"""Compute is_terminal and mc_return arrays for a single episode.
Args:
num_frames: Number of frames in the episode.
success: Whether the episode ended successfully.
c_fail: Failure penalty constant.
gamma: Discount factor (1.0 = undiscounted).
max_episode_length: Normalization horizon H.
Returns:
Tuple of (is_terminal, mc_return) arrays, each of length num_frames.
"""
horizon = max_episode_length
rewards = np.full(num_frames, -1.0 / horizon, dtype=np.float64)
if success:
rewards[-1] = 0.0
else:
rewards[-1] = -c_fail / horizon
is_terminal = np.zeros(num_frames, dtype=bool)
is_terminal[-1] = True
if gamma == 1.0:
# Reverse cumulative sum
mc_return = np.cumsum(rewards[::-1])[::-1].astype(np.float32)
else:
mc_return = np.zeros(num_frames, dtype=np.float64)
mc_return[-1] = rewards[-1]
for t in range(num_frames - 2, -1, -1):
mc_return[t] = rewards[t] + gamma * mc_return[t + 1]
mc_return = mc_return.astype(np.float32)
return is_terminal, mc_return
def compute_returns(config: ComputeReturnsConfig) -> Path:
"""Compute returns and write them into parquet shards."""
from lerobot.datasets import LeRobotDataset
logger.info(f"Loading dataset: {config.dataset_repo_id}")
kwargs = {"repo_id": config.dataset_repo_id, "download_videos": False}
if config.root:
kwargs["root"] = config.root
dataset = LeRobotDataset(**kwargs)
meta = dataset.meta
root = Path(meta.root)
logger.info(f"Dataset root: {root}")
logger.info(f"Episodes: {meta.total_episodes}, Frames: {meta.total_frames}")
episode_indices = config.episodes if config.episodes else list(range(meta.total_episodes))
if config.max_episode_length is not None:
max_ep_len = config.max_episode_length
else:
max_ep_len = max(int(meta.episodes[i]["length"]) for i in episode_indices)
logger.info(f"Normalization horizon (max_episode_length): {max_ep_len}")
parquet_files_to_rewrite: dict[Path, list[int]] = {}
for ep_idx in episode_indices:
rel_path = meta.get_data_file_path(ep_idx)
abs_path = root / rel_path
parquet_files_to_rewrite.setdefault(abs_path, []).append(ep_idx)
logger.info(f"Parquet shards to rewrite: {len(parquet_files_to_rewrite)}")
for parquet_path, ep_indices_in_file in tqdm(parquet_files_to_rewrite.items(), desc="Processing shards"):
table = pq.read_table(parquet_path)
if not config.force and IS_TERMINAL_COL in table.column_names:
logger.info(f"Skipping {parquet_path.name} (already has {IS_TERMINAL_COL})")
continue
all_is_terminal = np.zeros(len(table), dtype=bool)
all_mc_return = np.zeros(len(table), dtype=np.float32)
episode_col = table.column("episode_index").to_pylist()
for ep_idx in ep_indices_in_file:
ep_info = meta.episodes[ep_idx]
ep_from = int(ep_info["dataset_from_index"])
ep_to = int(ep_info["dataset_to_index"])
ep_len = ep_to - ep_from
mask = np.array([v == ep_idx for v in episode_col], dtype=bool)
local_indices = np.where(mask)[0]
if len(local_indices) != ep_len:
logger.warning(
f"Episode {ep_idx}: expected {ep_len} frames in shard, "
f"found {len(local_indices)}. Using found count."
)
ep_len = len(local_indices)
if ep_len == 0:
continue
ep_subtable = table.filter(mask)
success = _get_episode_success(ep_subtable, config.success_key, config.default_success)
is_terminal, mc_return = compute_episode_returns(
num_frames=ep_len,
success=success,
c_fail=config.c_fail,
gamma=config.gamma,
max_episode_length=max_ep_len,
)
all_is_terminal[local_indices] = is_terminal
all_mc_return[local_indices] = mc_return
if IS_TERMINAL_COL in table.column_names:
table = table.drop(IS_TERMINAL_COL)
if MC_RETURN_COL in table.column_names:
table = table.drop(MC_RETURN_COL)
table = table.append_column(IS_TERMINAL_COL, pa.array(all_is_terminal))
table = table.append_column(MC_RETURN_COL, pa.array(all_mc_return))
pq.write_table(table, parquet_path)
_update_info_json(root, meta)
logger.info("Done. Columns written: is_terminal, mc_return")
return root
def _update_info_json(root: Path, meta) -> None:
"""Add is_terminal and mc_return to the dataset's info.json features."""
info_path = root / "meta" / "info.json"
if not info_path.exists():
logger.warning(f"info.json not found at {info_path}, skipping metadata update.")
return
info = json.loads(info_path.read_text())
features = info.get("features", {})
changed = False
if IS_TERMINAL_COL not in features:
features[IS_TERMINAL_COL] = {
"dtype": "bool",
"shape": [1],
"names": None,
}
changed = True
if MC_RETURN_COL not in features:
features[MC_RETURN_COL] = {
"dtype": "float32",
"shape": [1],
"names": None,
}
changed = True
if changed:
info["features"] = features
info_path.write_text(json.dumps(info, indent=2) + "\n")
logger.info("Updated meta/info.json with is_terminal and mc_return features.")
def main():
parser = argparse.ArgumentParser(
description="Compute per-frame is_terminal and mc_return for a LeRobot dataset.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Use the 'success' column from the dataset
lerobot-compute-returns --dataset-repo-id lerobot/aloha_sim_insertion_human_image
# Override all episodes as successful (demo-only data)
lerobot-compute-returns --dataset-repo-id my_org/my_dataset --default-success true
# Custom failure penalty
lerobot-compute-returns --dataset-repo-id my_org/my_dataset --c-fail 100
""",
)
parser.add_argument(
"--dataset-repo-id",
type=str,
required=True,
help="HuggingFace dataset repo id or local path.",
)
parser.add_argument(
"--root",
type=str,
default=None,
help="Local root directory override for the dataset.",
)
parser.add_argument(
"--success-key",
type=str,
default="next.success",
help="Column name in parquet that indicates episode success (default: 'next.success').",
)
parser.add_argument(
"--default-success",
type=str,
default=None,
choices=["true", "false"],
help="Override success for all episodes ('true' or 'false').",
)
parser.add_argument(
"--max-episode-length",
type=int,
default=None,
help="Normalization horizon H. If not set, uses max episode length in dataset.",
)
parser.add_argument(
"--c-fail",
type=float,
default=50.0,
help="Failure penalty constant (default: 50.0).",
)
parser.add_argument(
"--gamma",
type=float,
default=1.0,
help="Discount factor (default: 1.0, undiscounted).",
)
parser.add_argument(
"--episodes",
type=int,
nargs="+",
default=None,
help="Process only these episode indices (default: all).",
)
parser.add_argument(
"--force",
action="store_true",
help="Overwrite existing is_terminal/mc_return columns.",
)
args = parser.parse_args()
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
default_success = None
if args.default_success is not None:
default_success = args.default_success.lower() == "true"
config = ComputeReturnsConfig(
dataset_repo_id=args.dataset_repo_id,
root=args.root,
success_key=args.success_key,
default_success=default_success,
max_episode_length=args.max_episode_length,
c_fail=args.c_fail,
gamma=args.gamma,
episodes=args.episodes or [],
force=args.force,
)
root = compute_returns(config)
logger.info(f"Returns computed and written to: {root}")
logger.info(f" Columns added: {IS_TERMINAL_COL}, {MC_RETURN_COL}")
logger.info("To train the distributional value function, these columns")
logger.info("will be read as flat batch keys during training.")
if __name__ == "__main__":
main()
+90 -11
View File
@@ -169,6 +169,7 @@ def rollout(
env_features: dict | None = None,
recording_repo_id: str | None = None,
recording_private: bool = False,
predicted_latents_callback: Callable[[PreTrainedPolicy], None] | None = None,
) -> dict:
"""Run a batched policy rollout once through a batch of environments.
@@ -198,6 +199,9 @@ def rollout(
are returned optionally because they typically take more memory to cache. Defaults to False.
render_callback: Optional rendering callback to be used after the environments are reset, and after
every step.
predicted_latents_callback: Optional callback invoked after every ``select_action`` with the policy
itself. World-model policies (e.g. LingBot-VA) stash predicted video latents on
``policy.last_predicted_latents``; this lets the caller concatenate chunks and decode once.
Returns:
The dictionary described above.
"""
@@ -276,6 +280,8 @@ def rollout(
observation = preprocessor(observation)
with torch.inference_mode():
action = policy.select_action(observation)
if predicted_latents_callback is not None:
predicted_latents_callback(policy)
action = postprocessor(action)
action_transition = {ACTION: action}
@@ -295,12 +301,22 @@ def rollout(
# available if none of the envs finished.
if "final_info" in info:
final_info = info["final_info"]
if not isinstance(final_info, dict):
raise RuntimeError(
"Unsupported `final_info` format: expected dict (Gymnasium >= 1.0). "
"You're likely using an older version of gymnasium (< 1.0). Please upgrade."
if isinstance(final_info, dict):
is_success = final_info.get("is_success", [False] * env.num_envs)
successes = (
is_success.tolist()
if hasattr(is_success, "tolist")
else [bool(is_success)] * env.num_envs
)
successes = final_info["is_success"].tolist()
else:
# Gymnasium < 1.0 returns final_info as a per-env sequence/object array,
# with entries set to a dict only for envs that just finished.
successes = []
for item in final_info:
if isinstance(item, dict) and "is_success" in item:
successes.append(bool(item["is_success"]))
else:
successes.append(False)
elif "is_success" in info:
is_success = info["is_success"]
successes = (
@@ -400,6 +416,7 @@ def eval_policy(
env_features: dict | None = None,
recording_repo_id: str | None = None,
recording_private: bool = False,
save_predicted_video: bool = False,
) -> dict:
"""
Args:
@@ -418,6 +435,11 @@ def eval_policy(
if max_episodes_rendered > 0 and not videos_dir:
raise ValueError("If max_episodes_rendered > 0, videos_dir must be provided.")
# World-model policies (e.g. LingBot-VA) opt into predicted-video saving via their config.
save_predicted_video = save_predicted_video or bool(
getattr(getattr(policy, "config", None), "save_predicted_video", False)
)
if not isinstance(policy, PreTrainedPolicy):
exc = ValueError(
f"Policy of type 'PreTrainedPolicy' is expected, but type '{type(policy)}' was provided."
@@ -461,6 +483,22 @@ def eval_policy(
if max_episodes_rendered > 0:
video_paths: list[str] = []
if save_predicted_video:
if not videos_dir:
raise ValueError("If save_predicted_video is True, videos_dir must be provided.")
predicted_video_paths: list[str] = []
n_predicted_rendered = 0
# Collect predicted-video latents across a rollout (world-model policies only). The latents are
# concatenated and decoded once after the rollout, matching upstream LingBot-VA's visualization path.
def collect_predicted_latents(policy: PreTrainedPolicy):
latents = getattr(policy, "last_predicted_latents", None)
if latents is not None:
pred_latents.append(
latents.detach().to("cpu") if hasattr(latents, "detach") else torch.as_tensor(latents).cpu()
)
policy.last_predicted_latents = None
if return_episode_data:
episode_data: dict | None = None
@@ -472,6 +510,9 @@ def eval_policy(
if max_episodes_rendered > 0:
ep_frames: list[np.ndarray] = []
if save_predicted_video:
pred_latents: list[torch.Tensor] = []
if start_seed is None:
seeds = None
else:
@@ -492,6 +533,7 @@ def eval_policy(
env_features=env_features,
recording_repo_id=recording_repo_id,
recording_private=recording_private,
predicted_latents_callback=collect_predicted_latents if save_predicted_video else None,
)
# Figure out where in each rollout sequence the first done condition was encountered (results after
@@ -557,6 +599,35 @@ def eval_policy(
threads.append(thread)
n_episodes_rendered += 1
# Maybe save the policy's predicted (imagined) video for this batch's rollout.
if save_predicted_video and len(pred_latents) > 0:
predicted_latent = torch.cat(pred_latents, dim=2)
decoder = getattr(policy, "decode_predicted_latents", None) or getattr(
policy, "_decode_predicted_video", None
)
if decoder is None:
raise AttributeError(
"Policy config requested predicted-video saving, but the policy does not expose "
"`decode_predicted_latents` or `_decode_predicted_video`."
)
predicted_video = decoder(predicted_latent)
if hasattr(predicted_video, "detach"):
predicted_video = predicted_video.detach().to("cpu").numpy()
videos_dir.mkdir(parents=True, exist_ok=True)
predicted_video_path = videos_dir / f"pred_episode_{n_predicted_rendered}.mp4"
predicted_video_paths.append(str(predicted_video_path))
thread = threading.Thread(
target=write_video,
args=(
str(predicted_video_path),
predicted_video,
env.unwrapped.metadata["render_fps"],
),
)
thread.start()
threads.append(thread)
n_predicted_rendered += 1
progbar.set_postfix(
{"running_success_rate": f"{np.mean(all_successes[:n_episodes]).item() * 100:.1f}%"}
)
@@ -600,6 +671,9 @@ def eval_policy(
if max_episodes_rendered > 0:
info["video_paths"] = video_paths
if save_predicted_video:
info["predicted_video_paths"] = predicted_video_paths
return info
@@ -740,9 +814,10 @@ class TaskMetrics(TypedDict):
max_rewards: list[float]
successes: list[bool]
video_paths: list[str]
predicted_video_paths: list[str]
ACC_KEYS = ("sum_rewards", "max_rewards", "successes", "video_paths")
ACC_KEYS = ("sum_rewards", "max_rewards", "successes", "video_paths", "predicted_video_paths")
def eval_one(
@@ -791,6 +866,7 @@ def eval_one(
max_rewards=[ep["max_reward"] for ep in per_episode],
successes=[ep["success"] for ep in per_episode],
video_paths=task_result.get("video_paths", []),
predicted_video_paths=task_result.get("predicted_video_paths", []),
)
@@ -851,6 +927,7 @@ def run_one(
if max_episodes_rendered > 0:
metrics.setdefault("video_paths", [])
metrics.setdefault("predicted_video_paths", [])
return task_group, task_id, metrics
@@ -908,11 +985,11 @@ def eval_policy_all(
_append("sum_rewards", metrics.get("sum_rewards"))
_append("max_rewards", metrics.get("max_rewards"))
_append("successes", metrics.get("successes"))
# video_paths is list-like
paths = metrics.get("video_paths", [])
if paths:
group_acc[group]["video_paths"].extend(paths)
overall["video_paths"].extend(paths)
for key in ("video_paths", "predicted_video_paths"):
paths = metrics.get(key, [])
if paths:
group_acc[group][key].extend(paths)
overall[key].extend(paths)
# Choose runner (sequential vs threaded)
task_runner = partial(
@@ -984,6 +1061,7 @@ def eval_policy_all(
"pc_success": _agg_from_list(acc["successes"]) * 100 if acc["successes"] else float("nan"),
"n_episodes": len(acc["sum_rewards"]),
"video_paths": list(acc["video_paths"]),
"predicted_video_paths": list(acc["predicted_video_paths"]),
}
# overall aggregates
@@ -995,6 +1073,7 @@ def eval_policy_all(
"eval_s": time.time() - start_t,
"eval_ep_s": (time.time() - start_t) / max(1, len(overall["sum_rewards"])),
"video_paths": list(overall["video_paths"]),
"predicted_video_paths": list(overall["predicted_video_paths"]),
}
return {
+3
View File
@@ -26,6 +26,9 @@ OBS_IMAGES = OBS_IMAGE + "s"
OBS_LANGUAGE = OBS_STR + ".language"
OBS_LANGUAGE_TOKENS = OBS_LANGUAGE + ".tokens"
OBS_LANGUAGE_ATTENTION_MASK = OBS_LANGUAGE + ".attention_mask"
OBS_LANGUAGE_UNCOND = OBS_STR + ".language_uncond"
OBS_LANGUAGE_UNCOND_TOKENS = OBS_LANGUAGE_UNCOND + ".tokens"
OBS_LANGUAGE_UNCOND_ATTENTION_MASK = OBS_LANGUAGE_UNCOND + ".attention_mask"
OBS_LANGUAGE_SUBTASK = OBS_STR + ".subtask"
OBS_LANGUAGE_SUBTASK_TOKENS = OBS_LANGUAGE_SUBTASK + ".tokens"
OBS_LANGUAGE_SUBTASK_ATTENTION_MASK = OBS_LANGUAGE_SUBTASK + ".attention_mask"
+3 -1
View File
@@ -28,9 +28,10 @@ import sys
import tempfile
from pathlib import Path
from lerobot.annotations.steerable_pipeline.config import AnnotationPipelineConfig
from lerobot.annotations.steerable_pipeline.config import AdvantageConfig, AnnotationPipelineConfig
from lerobot.annotations.steerable_pipeline.executor import Executor
from lerobot.annotations.steerable_pipeline.modules import (
AdvantageModule,
GeneralVqaModule,
InterjectionsAndSpeechModule,
PlanSubtasksMemoryModule,
@@ -85,6 +86,7 @@ def main() -> int:
plan=PlanSubtasksMemoryModule(vlm=vlm, config=cfg.plan),
interjections=InterjectionsAndSpeechModule(vlm=vlm, config=cfg.interjections, seed=cfg.seed),
vqa=GeneralVqaModule(vlm=vlm, config=cfg.vqa, seed=cfg.seed),
advantage=AdvantageModule(config=AdvantageConfig(enabled=False)),
writer=LanguageColumnsWriter(),
validator=StagingValidator(),
)
+305
View File
@@ -0,0 +1,305 @@
#!/usr/bin/env python
# Copyright 2026 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.
"""Tests for the advantage scoring annotation module."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import MagicMock, patch
import numpy as np
import pytest
from lerobot.annotations.steerable_pipeline.config import AdvantageConfig
from lerobot.annotations.steerable_pipeline.modules.advantage import AdvantageModule
from lerobot.annotations.steerable_pipeline.reader import EpisodeRecord
from lerobot.annotations.steerable_pipeline.staging import EpisodeStaging
def _make_record(
episode_index: int = 0,
num_frames: int = 20,
task: str = "pick up the cup",
mc_returns: np.ndarray | None = None,
intervention_mask: np.ndarray | None = None,
fps: float = 10.0,
) -> EpisodeRecord:
"""Build a minimal EpisodeRecord with a mocked frames_df."""
import pandas as pd
timestamps = tuple(round(i / fps, 6) for i in range(num_frames))
frame_indices = tuple(range(num_frames))
if mc_returns is None:
mc_returns = np.linspace(-0.9, -0.1, num_frames).astype(np.float32)
data = {
"episode_index": [episode_index] * num_frames,
"frame_index": list(range(num_frames)),
"timestamp": list(timestamps),
"mc_return": mc_returns,
}
if intervention_mask is not None:
data["intervention"] = intervention_mask.astype(bool)
df = pd.DataFrame(data)
record = EpisodeRecord(
episode_index=episode_index,
episode_task=task,
frame_timestamps=timestamps,
frame_indices=frame_indices,
data_path=Path("/fake/data.parquet"),
row_offset=0,
row_count=num_frames,
)
record._frames_df_cache = df
return record
@pytest.fixture
def staging(tmp_path: Path) -> EpisodeStaging:
return EpisodeStaging(tmp_path, episode_index=0)
def test_advantage_module_disabled():
"""Disabled module has enabled=False."""
cfg = AdvantageConfig(enabled=False)
module = AdvantageModule(config=cfg)
assert not module.enabled
def test_advantage_module_enabled_by_default():
"""Module is enabled by default."""
cfg = AdvantageConfig()
module = AdvantageModule(config=cfg)
assert module.enabled
def test_run_episode_skips_without_value_function_path(staging: EpisodeStaging):
"""Module gracefully returns when no value_function_path is configured."""
cfg = AdvantageConfig(value_function_path="")
module = AdvantageModule(config=cfg)
record = _make_record()
module.run_episode(record, staging)
rows = staging.read("advantage")
assert rows == []
def test_binarization_with_mock_values(staging: EpisodeStaging):
"""Advantage binarization produces positive/negative labels based on threshold."""
num_frames = 10
mc_returns = np.array([-0.5, -0.4, -0.3, -0.2, -0.1, -0.5, -0.6, -0.7, -0.8, -0.9], dtype=np.float32)
mock_values = np.array([-0.4, -0.4, -0.4, -0.4, -0.4, -0.4, -0.4, -0.4, -0.4, -0.4], dtype=np.float32)
cfg = AdvantageConfig(
value_function_path="/fake/vf",
dropout_rate=0.0,
threshold_percentile=0.5,
)
module = AdvantageModule(config=cfg)
record = _make_record(num_frames=num_frames, mc_returns=mc_returns)
with (
patch.object(module, "_ensure_model_loaded"),
patch.object(module, "_compute_values", return_value=mock_values),
):
module.run_episode(record, staging)
rows = staging.read("advantage")
assert len(rows) == num_frames
# A_t = mc_returns - values
# advantages = [-0.1, 0.0, 0.1, 0.2, 0.3, -0.1, -0.2, -0.3, -0.4, -0.5]
# Median (50th pctile) = -0.1
# positive: advantage > -0.1 → indices 1,2,3,4
# negative: advantage <= -0.1 → indices 0,5,6,7,8,9
positives = [r for r in rows if r["content"] == "positive"]
negatives = [r for r in rows if r["content"] == "negative"]
assert len(positives) == 4
assert len(negatives) == 6
def test_intervention_frames_forced_positive(staging: EpisodeStaging):
"""Intervention frames are always scored as positive regardless of advantage value."""
num_frames = 5
mc_returns = np.array([-0.9, -0.9, -0.9, -0.9, -0.9], dtype=np.float32)
mock_values = np.array([-0.1, -0.1, -0.1, -0.1, -0.1], dtype=np.float32)
intervention = np.array([False, False, True, False, False])
cfg = AdvantageConfig(
value_function_path="/fake/vf",
dropout_rate=0.0,
force_positive_on_intervention=True,
)
module = AdvantageModule(config=cfg)
record = _make_record(num_frames=num_frames, mc_returns=mc_returns, intervention_mask=intervention)
with (
patch.object(module, "_ensure_model_loaded"),
patch.object(module, "_compute_values", return_value=mock_values),
):
module.run_episode(record, staging)
rows = staging.read("advantage")
# Frame 2 (intervention) should be positive despite negative advantage
assert rows[2]["content"] == "positive"
def test_dropout_reduces_output_rows(staging: EpisodeStaging):
"""Non-zero dropout rate omits some frames."""
num_frames = 100
mc_returns = np.linspace(-0.9, -0.1, num_frames).astype(np.float32)
mock_values = np.full(num_frames, -0.5, dtype=np.float32)
cfg = AdvantageConfig(
value_function_path="/fake/vf",
dropout_rate=0.3,
)
module = AdvantageModule(config=cfg)
record = _make_record(num_frames=num_frames, mc_returns=mc_returns)
with (
patch.object(module, "_ensure_model_loaded"),
patch.object(module, "_compute_values", return_value=mock_values),
):
module.run_episode(record, staging)
rows = staging.read("advantage")
# With 30% dropout on 100 frames, expect ~70 rows (with some variance)
assert 50 < len(rows) < 90
def test_staged_row_format(staging: EpisodeStaging):
"""Staged rows have the correct schema for language_persistent."""
num_frames = 5
mc_returns = np.array([-0.5, -0.4, -0.3, -0.2, -0.1], dtype=np.float32)
mock_values = np.full(5, -0.3, dtype=np.float32)
cfg = AdvantageConfig(
value_function_path="/fake/vf",
dropout_rate=0.0,
)
module = AdvantageModule(config=cfg)
record = _make_record(num_frames=num_frames, mc_returns=mc_returns)
with (
patch.object(module, "_ensure_model_loaded"),
patch.object(module, "_compute_values", return_value=mock_values),
):
module.run_episode(record, staging)
rows = staging.read("advantage")
for row in rows:
assert row["role"] == "user"
assert row["content"] in ("positive", "negative")
assert row["style"] == "advantage"
assert isinstance(row["timestamp"], float)
assert row["camera"] is None
assert row["tool_calls"] is None
def test_n_step_advantage():
"""N-step advantage uses partial returns + bootstrapped value."""
num_frames = 10
mc_returns = np.linspace(-0.9, 0.0, num_frames).astype(np.float32)
mock_values = np.full(num_frames, -0.45, dtype=np.float32)
cfg = AdvantageConfig(
value_function_path="/fake/vf",
n_step=3,
dropout_rate=0.0,
)
module = AdvantageModule(config=cfg)
record = _make_record(num_frames=num_frames, mc_returns=mc_returns)
with patch.object(module, "_ensure_model_loaded"):
advantages, _ = (
module.compute_advantages_for_episode.__wrapped__(module, record)
if hasattr(module.compute_advantages_for_episode, "__wrapped__")
else (None, None)
)
# Just verify computation works - use the internal method directly
module._model = MagicMock()
module._preprocessor = MagicMock()
with patch.object(module, "_compute_values", return_value=mock_values):
advantages, _ = module.compute_advantages_for_episode(record)
# For t where t+n < num_frames: A = mc_return[t] - mc_return[t+n] + values[t+n] - values[t]
# Since values are constant: A = mc_return[t] - mc_return[t+n]
# For t where t+n >= num_frames: A = mc_return[t] - values[t]
for t in range(num_frames):
if t + 3 < num_frames:
expected = mc_returns[t] - mc_returns[t + 3] + mock_values[t + 3] - mock_values[t]
else:
expected = mc_returns[t] - mock_values[t]
np.testing.assert_almost_equal(advantages[t], expected, decimal=5)
def test_compute_threshold():
"""Threshold is computed as configured percentile of non-intervention advantages."""
cfg = AdvantageConfig(threshold_percentile=0.3)
module = AdvantageModule(config=cfg)
advantages = np.array([-1.0, -0.5, 0.0, 0.5, 1.0], dtype=np.float32)
intervention_mask = np.array([False, False, False, False, False])
threshold = module._compute_threshold(advantages, intervention_mask)
expected = float(np.percentile(advantages, 30))
assert abs(threshold - expected) < 1e-6
def test_compute_threshold_excludes_intervention():
"""Threshold computation excludes intervention frames."""
cfg = AdvantageConfig(threshold_percentile=0.5)
module = AdvantageModule(config=cfg)
advantages = np.array([100.0, -1.0, 0.0, 1.0, 100.0], dtype=np.float32)
intervention_mask = np.array([True, False, False, False, True])
threshold = module._compute_threshold(advantages, intervention_mask)
# Only non-intervention: [-1.0, 0.0, 1.0], median = 0.0
expected = float(np.percentile([-1.0, 0.0, 1.0], 50))
assert abs(threshold - expected) < 1e-6
def test_missing_mc_return_raises():
"""Module raises if mc_return column is missing from dataset."""
import pandas as pd
cfg = AdvantageConfig(value_function_path="/fake/vf")
module = AdvantageModule(config=cfg)
module._model = MagicMock()
module._preprocessor = MagicMock()
record = EpisodeRecord(
episode_index=0,
episode_task="test",
frame_timestamps=(0.0, 0.1),
frame_indices=(0, 1),
data_path=Path("/fake/data.parquet"),
row_offset=0,
row_count=2,
)
record._frames_df_cache = pd.DataFrame({"episode_index": [0, 0], "frame_index": [0, 1]})
with pytest.raises(KeyError, match="mc_return"):
module.compute_advantages_for_episode(record)
@@ -30,6 +30,7 @@ pytest.importorskip("pandas", reason="pandas is required (install lerobot[datase
import pyarrow.parquet as pq # noqa: E402
from lerobot.annotations.steerable_pipeline.config import ( # noqa: E402
AdvantageConfig,
AnnotationPipelineConfig,
InterjectionsConfig,
PlanConfig,
@@ -37,6 +38,7 @@ from lerobot.annotations.steerable_pipeline.config import ( # noqa: E402
)
from lerobot.annotations.steerable_pipeline.executor import Executor # noqa: E402
from lerobot.annotations.steerable_pipeline.modules import ( # noqa: E402
AdvantageModule,
GeneralVqaModule,
InterjectionsAndSpeechModule,
PlanSubtasksMemoryModule,
@@ -132,6 +134,7 @@ def _build_executor() -> Executor:
plan=PlanSubtasksMemoryModule(vlm=vlm, config=config.plan),
interjections=InterjectionsAndSpeechModule(vlm=vlm, config=config.interjections, seed=config.seed),
vqa=GeneralVqaModule(vlm=vlm, config=config.vqa, seed=config.seed),
advantage=AdvantageModule(config=AdvantageConfig(enabled=False)),
writer=LanguageColumnsWriter(),
validator=StagingValidator(),
)
+145
View File
@@ -0,0 +1,145 @@
#!/usr/bin/env python
# Copyright 2026 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.
"""Tests for RECAP advantage conditioning recipes."""
from __future__ import annotations
from pathlib import Path
from lerobot.configs.recipe import load_recipe
from lerobot.datasets.language_render import render_sample
RECIPES_DIR = Path(__file__).resolve().parents[2] / "src" / "lerobot" / "configs" / "recipes"
def _persistent_rows(advantage: str | None = None):
"""Build minimal persistent rows with optional advantage."""
rows = [
{
"role": "user",
"content": "pick up the cup",
"style": "task_aug",
"timestamp": 0.0,
"camera": None,
"tool_calls": None,
},
{
"role": "assistant",
"content": "reaching for the cup",
"style": "subtask",
"timestamp": 0.0,
"camera": None,
"tool_calls": None,
},
]
if advantage is not None:
rows.append(
{
"role": "user",
"content": advantage,
"style": "advantage",
"timestamp": 0.0,
"camera": None,
"tool_calls": None,
}
)
return rows
def test_recap_advantage_recipe_loads():
"""The recap_advantage.yaml recipe loads without errors."""
recipe = load_recipe(RECIPES_DIR / "recap_advantage.yaml")
assert recipe.messages is not None
assert len(recipe.messages) == 3
assert recipe.bindings == {"advantage": "active_at(t, style=advantage)"}
def test_advantage_present_renders_indicator():
"""When advantage annotation exists, the prompt includes 'Advantage: positive'."""
recipe = load_recipe(RECIPES_DIR / "recap_advantage.yaml")
result = render_sample(
recipe=recipe,
persistent=_persistent_rows(advantage="positive"),
events=[],
t=0.5,
sample_idx=0,
task="pick up the cup",
)
assert result is not None
messages = result["messages"]
assert len(messages) == 3
assert messages[1]["content"] == "Advantage: positive"
def test_advantage_negative_renders_indicator():
"""Negative advantage also appears in the prompt."""
recipe = load_recipe(RECIPES_DIR / "recap_advantage.yaml")
result = render_sample(
recipe=recipe,
persistent=_persistent_rows(advantage="negative"),
events=[],
t=0.5,
sample_idx=0,
task="pick up the cup",
)
assert result is not None
messages = result["messages"]
assert messages[1]["content"] == "Advantage: negative"
def test_advantage_absent_skips_turn():
"""When no advantage annotation exists (dropout), the advantage turn is skipped."""
recipe = load_recipe(RECIPES_DIR / "recap_advantage.yaml")
result = render_sample(
recipe=recipe,
persistent=_persistent_rows(advantage=None),
events=[],
t=0.5,
sample_idx=0,
task="pick up the cup",
)
assert result is not None
messages = result["messages"]
# Only task + subtask, no advantage turn
assert len(messages) == 2
assert messages[0]["content"] == "pick up the cup"
assert messages[1]["content"] == "reaching for the cup"
def test_advantage_absent_still_has_target():
"""Even without advantage, the target message (subtask) is preserved."""
recipe = load_recipe(RECIPES_DIR / "recap_advantage.yaml")
result = render_sample(
recipe=recipe,
persistent=_persistent_rows(advantage=None),
events=[],
t=0.5,
sample_idx=0,
task="pick up the cup",
)
assert result is not None
assert result["target_message_indices"] == [1]
def test_blend_recipe_loads():
"""The blend recipe has two components with correct weights."""
recipe = load_recipe(RECIPES_DIR / "recap_advantage_blend.yaml")
assert recipe.blend is not None
assert "advantage_conditioned" in recipe.blend
assert "unconditional" in recipe.blend
assert recipe.blend["advantage_conditioned"].weight == 0.7
assert recipe.blend["unconditional"].weight == 0.3
@@ -0,0 +1,78 @@
#!/usr/bin/env python
# Copyright 2026 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.
from __future__ import annotations
import pytest
from lerobot.configs.policies import PreTrainedConfig
from lerobot.configs.types import FeatureType, PolicyFeature
from lerobot.policies.lingbot_va.configuration_lingbot_va import LingBotVAConfig
from lerobot.utils.constants import ACTION, OBS_IMAGES
def make_config(**overrides) -> LingBotVAConfig:
kwargs = {"device": "cpu"}
kwargs.update(overrides)
return LingBotVAConfig(**kwargs)
def test_registered_in_choice_registry() -> None:
assert "lingbot_va" in PreTrainedConfig.get_known_choices()
assert PreTrainedConfig.get_choice_class("lingbot_va") is LingBotVAConfig
def test_type_property() -> None:
assert make_config().type == "lingbot_va"
def test_chunk_size_and_action_steps() -> None:
cfg = make_config(frame_chunk_size=4, action_per_frame=4)
assert cfg.chunk_size == 16
assert cfg.n_action_steps == 16
assert cfg.action_delta_indices == list(range(16))
assert cfg.observation_delta_indices == list(range(16))
assert cfg.reward_delta_indices is None
def test_optimizer_and_scheduler_presets() -> None:
cfg = make_config()
opt = cfg.get_optimizer_preset()
assert opt.lr == cfg.optimizer_lr
sched = cfg.get_scheduler_preset()
assert sched.num_warmup_steps == cfg.scheduler_warmup_steps
def test_validate_features_sets_action_feature() -> None:
cfg = make_config()
cfg.input_features = {f"{OBS_IMAGES}.image": PolicyFeature(type=FeatureType.VISUAL, shape=(3, 128, 128))}
cfg.output_features = {}
cfg.validate_features()
assert ACTION in cfg.output_features
assert cfg.output_features[ACTION].shape == (len(cfg.used_action_channel_ids),)
def test_validate_features_no_visual_raises() -> None:
cfg = make_config()
cfg.input_features = {}
cfg.output_features = {}
with pytest.raises(ValueError, match="at least one visual input feature"):
cfg.validate_features()
def test_invalid_attn_mode_raises() -> None:
with pytest.raises(ValueError, match="attn_mode"):
make_config(attn_mode="banana")
+38
View File
@@ -0,0 +1,38 @@
#!/usr/bin/env python
# Copyright 2026 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.
from __future__ import annotations
import pytest
from lerobot.policies.factory import make_policy_config
from lerobot.policies.lingbot_va.configuration_lingbot_va import LingBotVAConfig
def test_make_policy_config_returns_lingbot_va() -> None:
cfg = make_policy_config("lingbot_va", device="cpu")
assert isinstance(cfg, LingBotVAConfig)
def test_get_policy_class_resolves_lazily() -> None:
# Importing the policy class pulls in diffusers (Wan2.2 stack); skip if unavailable.
pytest.importorskip("diffusers")
pytest.importorskip("transformers")
from lerobot.policies.factory import get_policy_class
cls = get_policy_class("lingbot_va")
assert cls.name == "lingbot_va"
assert cls.config_class is LingBotVAConfig
+128
View File
@@ -0,0 +1,128 @@
#!/usr/bin/env python
# Copyright 2026 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.
"""Unit tests for the vendored LingBot-VA helper code (scheduler + grid utilities)."""
from __future__ import annotations
import pytest
import torch
pytest.importorskip("diffusers") # the model code lives in modeling_lingbot_va, which imports diffusers
from lerobot.policies.lingbot_va.modeling_lingbot_va import FlowMatchScheduler
from lerobot.policies.lingbot_va.utils import data_seq_to_patch, get_mesh_id
def test_flow_match_scheduler_timesteps_monotone_decreasing() -> None:
sch = FlowMatchScheduler(shift=5.0, sigma_min=0.0, extra_one_step=True)
sch.set_timesteps(20)
assert sch.timesteps.shape == (20,)
diffs = sch.timesteps[1:] - sch.timesteps[:-1]
assert torch.all(diffs <= 0) # decreasing
def test_flow_match_scheduler_step_preserves_shape() -> None:
sch = FlowMatchScheduler(shift=5.0, sigma_min=0.0, extra_one_step=True)
sch.set_timesteps(20)
sample = torch.zeros(1, 48, 4, 8, 16)
out = sch.step(torch.ones_like(sample), sch.timesteps[0], sample)
assert out.shape == sample.shape
def test_flow_match_scheduler_add_noise() -> None:
sch = FlowMatchScheduler(shift=5.0, sigma_min=0.0, extra_one_step=True)
sch.set_timesteps(20)
sample = torch.randn(1, 48, 4, 8, 16)
noise = torch.randn_like(sample)
noisy = sch.add_noise(sample, noise, sch.timesteps[:4], t_dim=2)
assert noisy.shape == sample.shape
def test_get_mesh_id_latent_shape() -> None:
grid = get_mesh_id(4, 8, 16, 0, 1, 0)
assert grid.shape == (4, 4 * 8 * 16) # (f, h, w, stream) x tokens
def test_get_mesh_id_action_shape() -> None:
grid = get_mesh_id(4, 4, 1, 1, 1, 0, action=True)
assert grid.shape == (4, 4 * 4 * 1)
# Action rows for h/w are sentinel -1.
assert torch.all(grid[1] < 0)
assert torch.all(grid[2] < 0)
def test_data_seq_to_patch_roundtrip_shape() -> None:
b, f, h, w, c = 1, 4, 8, 16, 48
seq = torch.arange(b * f * h * w * c, dtype=torch.float32).reshape(b, f * h * w, c)
out = data_seq_to_patch((1, 2, 2), seq, f, h, w, batch_size=b)
assert out.shape == (b, c, f, h, w)
def test_training_step_reduces_loss_tiny_flex() -> None:
"""End-to-end single training step (flow-matching loss -> backward -> AdamW) on a tiny config.
Exercises the flex-attention training path; requires a CUDA GPU with flex-attention support.
"""
if not torch.cuda.is_available():
import pytest
pytest.skip("training step test requires a CUDA GPU (flex-attention)")
from lerobot.configs.types import FeatureType, PolicyFeature
from lerobot.policies.lingbot_va.configuration_lingbot_va import LingBotVAConfig
from lerobot.policies.lingbot_va.modeling_lingbot_va import LingBotVAPolicy
from lerobot.utils.constants import ACTION, OBS_IMAGES
cfg = LingBotVAConfig(
attn_mode="flex",
dtype="bfloat16",
in_channels=16,
out_channels=16,
action_dim=8,
text_dim=32,
freq_dim=64,
ffn_dim=64,
num_attention_heads=2,
attention_head_dim=24,
num_layers=2,
frame_chunk_size=2,
action_per_frame=4,
used_action_channel_ids=[0, 1, 2, 3],
obs_cam_keys=[f"{OBS_IMAGES}.image"],
device="cuda",
)
cfg.input_features = {f"{OBS_IMAGES}.image": PolicyFeature(type=FeatureType.VISUAL, shape=(3, 64, 64))}
cfg.output_features = {ACTION: PolicyFeature(type=FeatureType.ACTION, shape=(4,))}
cfg.validate_features()
policy = LingBotVAPolicy(cfg).to("cuda")
policy.train()
opt = torch.optim.AdamW(policy.get_optim_params(), lr=1e-4)
b, fc, apf = 1, cfg.frame_chunk_size, cfg.action_per_frame
latents = torch.randn(b, cfg.in_channels, fc, 4, 4, device="cuda", dtype=torch.bfloat16)
actions = torch.randn(b, cfg.action_dim, fc, apf, 1, device="cuda", dtype=torch.bfloat16)
amask = torch.zeros(cfg.action_dim, device="cuda")
amask[cfg.used_action_channel_ids] = 1.0
actions_mask = amask.view(1, -1, 1, 1, 1).expand_as(actions)
text_emb = torch.randn(b, cfg.max_sequence_length, cfg.text_dim, device="cuda", dtype=torch.bfloat16)
loss, metrics = policy.training_loss_from_streams(latents, actions, actions_mask, text_emb)
assert torch.isfinite(loss) and {"latent_loss", "action_loss"} <= set(metrics)
loss.backward()
assert any(p.grad is not None and torch.isfinite(p.grad).all() for p in policy.get_optim_params())
opt.step()
@@ -0,0 +1,88 @@
#!/usr/bin/env python
# Copyright 2026 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.
from __future__ import annotations
import torch
from lerobot.configs.types import FeatureType, PolicyFeature
from lerobot.policies.lingbot_va.configuration_lingbot_va import LingBotVAConfig
from lerobot.policies.lingbot_va.processor_lingbot_va import make_lingbot_va_pre_post_processors
from lerobot.processor import PolicyProcessorPipeline, UnnormalizerProcessorStep
from lerobot.processor.converters import policy_action_to_transition, transition_to_policy_action
from lerobot.utils.constants import (
ACTION,
OBS_IMAGES,
POLICY_POSTPROCESSOR_DEFAULT_NAME,
POLICY_PREPROCESSOR_DEFAULT_NAME,
)
def _make_config() -> LingBotVAConfig:
cfg = LingBotVAConfig(device="cpu")
cfg.input_features = {f"{OBS_IMAGES}.image": PolicyFeature(type=FeatureType.VISUAL, shape=(3, 128, 128))}
cfg.output_features = {}
cfg.validate_features()
return cfg
def test_make_pre_post_processors_names_and_steps() -> None:
cfg = _make_config()
pre, post = make_lingbot_va_pre_post_processors(cfg, dataset_stats=None)
assert pre.name == POLICY_PREPROCESSOR_DEFAULT_NAME
assert post.name == POLICY_POSTPROCESSOR_DEFAULT_NAME
# Actions are unnormalized by the standard built-in quantile unnormalizer.
assert any(isinstance(s, UnnormalizerProcessorStep) for s in post.steps)
def test_freshly_built_postprocessor_is_identity() -> None:
# Without action stats the quantile unnormalizer is a no-op (identity passthrough): the real
# per-benchmark q01/q99 are restored from the saved checkpoint on load, not hardcoded here.
cfg = _make_config()
_, post = make_lingbot_va_pre_post_processors(cfg, dataset_stats=None)
normed = torch.tensor([[0.3, -0.5, 1.0, -1.0, 0.0, 0.7, -0.2]])
assert torch.allclose(post(normed), normed, atol=1e-6)
def test_postprocessor_quantile_unnormalization() -> None:
# QUANTILES unnormalize maps [-1, 1] -> [q01, q99]: -1 -> q01, +1 -> q99.
cfg = _make_config()
q01 = [-1.0, -0.5, 0.0, -1.0, -1.0, -1.0, -1.0]
q99 = [1.0, 0.5, 2.0, 1.0, 1.0, 1.0, 1.0]
stats = {ACTION: {"q01": q01, "q99": q99}}
_, post = make_lingbot_va_pre_post_processors(cfg, dataset_stats=stats)
out_lo = post(torch.full((1, 7), -1.0))
out_hi = post(torch.full((1, 7), 1.0))
assert torch.allclose(out_lo, torch.tensor(q01).unsqueeze(0), atol=1e-4)
assert torch.allclose(out_hi, torch.tensor(q99).unsqueeze(0), atol=1e-4)
def test_postprocessor_stats_survive_save_load(tmp_path) -> None:
# Regression guard for the Hub mechanism: the q01/q99 stats live in the saved post-processor
# state and must round-trip through save_pretrained / from_pretrained.
cfg = _make_config()
q01 = [-0.6, -0.8, -0.9, -0.1, -0.15, -0.25, -1.0]
q99 = [0.9, 0.85, 0.9, 0.17, 0.18, 0.34, 1.0]
_, post = make_lingbot_va_pre_post_processors(cfg, dataset_stats={ACTION: {"q01": q01, "q99": q99}})
post.save_pretrained(tmp_path)
loaded = PolicyProcessorPipeline.from_pretrained(
tmp_path,
config_filename=f"{POLICY_POSTPROCESSOR_DEFAULT_NAME}.json",
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
)
out = loaded(torch.full((1, 7), -1.0))
assert torch.allclose(out, torch.tensor(q01).unsqueeze(0), atol=1e-4)
+224
View File
@@ -0,0 +1,224 @@
#!/usr/bin/env python
"""Tests for PI05 Classifier-Free Guidance (CFG) inference."""
import pytest
pytest.importorskip("transformers", reason="transformers is required for PI05")
import torch # noqa: E402
from lerobot.configs.types import FeatureType, PolicyFeature # noqa: E402
from lerobot.policies.pi05 import PI05Config, make_pi05_pre_post_processors # noqa: E402
from lerobot.processor.converters import create_transition # noqa: E402
from lerobot.processor.rendered_messages_to_task import RenderedMessagesToTaskStep # noqa: E402
from lerobot.types import TransitionKey # noqa: E402
from lerobot.utils.constants import ( # noqa: E402
OBS_LANGUAGE_ATTENTION_MASK,
OBS_LANGUAGE_TOKENS,
OBS_LANGUAGE_UNCOND_ATTENTION_MASK,
OBS_LANGUAGE_UNCOND_TOKENS,
)
class TestRenderedMessagesToTaskBaseTaskPreservation:
"""Tests that RenderedMessagesToTaskStep preserves base_task for CFG."""
def test_preserves_string_base_task(self):
transition = create_transition(
complementary_data={
"task": "pick up the cup",
"messages": [
{"role": "user", "content": "pick up the cup, Advantage: positive"},
],
}
)
step = RenderedMessagesToTaskStep()
out = step(transition)
data = out[TransitionKey.COMPLEMENTARY_DATA]
assert data["base_task"] == "pick up the cup"
assert data["task"] == "pick up the cup, Advantage: positive"
def test_preserves_list_base_task(self):
transition = create_transition(
complementary_data={
"task": ["task1", "task2"],
"messages": [
{"role": "user", "content": "rendered with advantage"},
],
}
)
step = RenderedMessagesToTaskStep()
out = step(transition)
data = out[TransitionKey.COMPLEMENTARY_DATA]
assert data["base_task"] == ["task1", "task2"]
def test_no_base_task_when_messages_absent(self):
transition = create_transition(complementary_data={"task": "pick up the cup"})
step = RenderedMessagesToTaskStep()
out = step(transition)
data = out[TransitionKey.COMPLEMENTARY_DATA]
assert "base_task" not in data
class TestPi05PrepareStateTokenizerCfg:
"""Tests for Pi05PrepareStateTokenizerProcessorStep with cfg_enabled."""
def _make_transition(self, task, base_task=None):
complementary_data = {"task": task}
if base_task is not None:
complementary_data["base_task"] = base_task
return create_transition(
observation={"observation.state": torch.zeros(1, 14)},
complementary_data=complementary_data,
)
def test_cfg_disabled_no_uncond_task(self):
from lerobot.policies.pi05.processor_pi05 import Pi05PrepareStateTokenizerProcessorStep
step = Pi05PrepareStateTokenizerProcessorStep(max_state_dim=14, cfg_enabled=False)
transition = self._make_transition(task=["pick up the cup, Advantage: positive"])
out = step(transition)
data = out[TransitionKey.COMPLEMENTARY_DATA]
assert "uncond_task" not in data
def test_cfg_enabled_produces_uncond_task_from_base(self):
from lerobot.policies.pi05.processor_pi05 import Pi05PrepareStateTokenizerProcessorStep
step = Pi05PrepareStateTokenizerProcessorStep(max_state_dim=14, cfg_enabled=True)
transition = self._make_transition(
task=["pick up the cup, Advantage: positive"],
base_task=["pick up the cup"],
)
out = step(transition)
data = out[TransitionKey.COMPLEMENTARY_DATA]
assert "uncond_task" in data
assert len(data["uncond_task"]) == 1
# Unconditional prompt uses base_task (no advantage)
assert "Advantage" not in data["uncond_task"][0]
assert "pick up the cup" in data["uncond_task"][0]
assert "State:" in data["uncond_task"][0]
def test_cfg_enabled_falls_back_to_task_when_no_base(self):
from lerobot.policies.pi05.processor_pi05 import Pi05PrepareStateTokenizerProcessorStep
step = Pi05PrepareStateTokenizerProcessorStep(max_state_dim=14, cfg_enabled=True)
transition = self._make_transition(task=["pick up the cup"])
out = step(transition)
data = out[TransitionKey.COMPLEMENTARY_DATA]
# Falls back to using task itself as unconditional
assert "uncond_task" in data
assert "pick up the cup" in data["uncond_task"][0]
class TestCfgPipelineConstruction:
"""Tests that the processor pipeline is constructed correctly for CFG."""
def _make_config(self, cfg_beta=1.0, recipe_path=None):
config = PI05Config(
max_action_dim=7,
max_state_dim=14,
cfg_beta=cfg_beta,
recipe_path=recipe_path,
device="cpu",
)
config.input_features = {
"observation.state": PolicyFeature(type=FeatureType.STATE, shape=(14,)),
"observation.images.base_0_rgb": PolicyFeature(type=FeatureType.VISUAL, shape=(3, 224, 224)),
}
config.output_features = {
"action": PolicyFeature(type=FeatureType.ACTION, shape=(7,)),
}
return config
def _make_dataset_stats(self):
return {
"observation.state": {
"mean": torch.zeros(14),
"std": torch.ones(14),
"min": torch.zeros(14),
"max": torch.ones(14),
"q01": torch.zeros(14),
"q99": torch.ones(14),
},
"action": {
"mean": torch.zeros(7),
"std": torch.ones(7),
"min": torch.zeros(7),
"max": torch.ones(7),
"q01": torch.zeros(7),
"q99": torch.ones(7),
},
"observation.images.base_0_rgb": {
"mean": torch.zeros(3, 224, 224),
"std": torch.ones(3, 224, 224),
"q01": torch.zeros(3, 224, 224),
"q99": torch.ones(3, 224, 224),
},
}
def test_no_uncond_tokenizer_when_cfg_disabled(self):
from lerobot.processor import TokenizerProcessorStep
config = self._make_config(cfg_beta=1.0)
preprocessor, _ = make_pi05_pre_post_processors(config, self._make_dataset_stats())
tokenizer_steps = [s for s in preprocessor.steps if isinstance(s, TokenizerProcessorStep)]
assert len(tokenizer_steps) == 1
def test_uncond_tokenizer_added_when_cfg_enabled(self):
from lerobot.processor import TokenizerProcessorStep
config = self._make_config(cfg_beta=2.0)
preprocessor, _ = make_pi05_pre_post_processors(config, self._make_dataset_stats())
tokenizer_steps = [s for s in preprocessor.steps if isinstance(s, TokenizerProcessorStep)]
assert len(tokenizer_steps) == 2
uncond_tokenizer = tokenizer_steps[1]
assert uncond_tokenizer.task_key == "uncond_task"
assert uncond_tokenizer.output_tokens_key == OBS_LANGUAGE_UNCOND_TOKENS
assert uncond_tokenizer.output_mask_key == OBS_LANGUAGE_UNCOND_ATTENTION_MASK
def test_cfg_pipeline_produces_both_token_sets(self):
config = self._make_config(cfg_beta=2.0)
preprocessor, _ = make_pi05_pre_post_processors(config, self._make_dataset_stats())
batch = {
"observation.state": torch.randn(14),
"observation.images.base_0_rgb": torch.rand(3, 224, 224),
"task": "pick up the cup",
}
processed = preprocessor(batch)
assert OBS_LANGUAGE_TOKENS in processed
assert OBS_LANGUAGE_ATTENTION_MASK in processed
assert OBS_LANGUAGE_UNCOND_TOKENS in processed
assert OBS_LANGUAGE_UNCOND_ATTENTION_MASK in processed
# Both should be tensors with the same shape
assert processed[OBS_LANGUAGE_TOKENS].shape == processed[OBS_LANGUAGE_UNCOND_TOKENS].shape
assert (
processed[OBS_LANGUAGE_ATTENTION_MASK].shape
== processed[OBS_LANGUAGE_UNCOND_ATTENTION_MASK].shape
)
def test_cfg_beta_1_no_uncond_tokens_in_output(self):
config = self._make_config(cfg_beta=1.0)
preprocessor, _ = make_pi05_pre_post_processors(config, self._make_dataset_stats())
batch = {
"observation.state": torch.randn(14),
"observation.images.base_0_rgb": torch.rand(3, 224, 224),
"task": "pick up the cup",
}
processed = preprocessor(batch)
assert OBS_LANGUAGE_TOKENS in processed
assert OBS_LANGUAGE_UNCOND_TOKENS not in processed
@@ -0,0 +1,186 @@
#!/usr/bin/env python
"""Tests for RenderedMessagesToTaskStep and PI05 pipeline integration with advantage."""
import pytest
pytest.importorskip("datasets", reason="datasets is required (install lerobot[dataset])")
import torch # noqa: E402
from lerobot.configs.recipe import MessageTurn, TrainingRecipe # noqa: E402
from lerobot.processor.converters import create_transition # noqa: E402
from lerobot.processor.render_messages_processor import RenderMessagesStep # noqa: E402
from lerobot.processor.rendered_messages_to_task import RenderedMessagesToTaskStep # noqa: E402
from lerobot.types import TransitionKey # noqa: E402
def test_rendered_messages_to_task_noops_without_messages():
"""Without messages key, the step is a no-op."""
transition = create_transition(complementary_data={"task": "pick up the cup"})
step = RenderedMessagesToTaskStep()
out = step(transition)
data = out[TransitionKey.COMPLEMENTARY_DATA]
assert data["task"] == "pick up the cup"
def test_rendered_messages_to_task_extracts_user_content():
"""Extracts user-role message content and joins with newline."""
transition = create_transition(
complementary_data={
"task": "original task",
"messages": [
{"role": "user", "content": "pick up the cup"},
{"role": "user", "content": "Advantage: positive"},
{"role": "assistant", "content": "reach for cup"},
],
"message_streams": ["high_level", "high_level", "low_level"],
"target_message_indices": [2],
}
)
step = RenderedMessagesToTaskStep()
out = step(transition)
data = out[TransitionKey.COMPLEMENTARY_DATA]
assert data["task"] == "pick up the cup\nAdvantage: positive"
assert "messages" not in data
assert "message_streams" not in data
assert "target_message_indices" not in data
def test_rendered_messages_to_task_handles_multimodal_blocks():
"""Extracts text from HF multimodal content blocks."""
transition = create_transition(
complementary_data={
"task": "original",
"messages": [
{
"role": "user",
"content": [
{"type": "image", "image": "placeholder"},
{"type": "text", "text": "describe this"},
],
},
{"role": "assistant", "content": "a cup on a table"},
],
"message_streams": ["high_level", "low_level"],
"target_message_indices": [1],
}
)
step = RenderedMessagesToTaskStep()
out = step(transition)
data = out[TransitionKey.COMPLEMENTARY_DATA]
assert data["task"] == "describe this"
def test_rendered_messages_to_task_preserves_list_task_format():
"""When original task is a list (batched), output is also a list."""
transition = create_transition(
complementary_data={
"task": ["task1", "task2"],
"messages": [
{"role": "user", "content": "rendered task"},
{"role": "assistant", "content": "do it", "target": True},
],
"message_streams": ["high_level", "low_level"],
"target_message_indices": [1],
}
)
step = RenderedMessagesToTaskStep()
out = step(transition)
data = out[TransitionKey.COMPLEMENTARY_DATA]
assert data["task"] == ["rendered task", "rendered task"]
def test_full_render_then_flatten_pipeline():
"""RenderMessagesStep + RenderedMessagesToTaskStep produces correct task string."""
recipe = TrainingRecipe(
messages=[
MessageTurn(role="user", content="${task}", stream="high_level"),
MessageTurn(
role="user",
content="Advantage: ${advantage}",
stream="high_level",
if_present="advantage",
),
MessageTurn(role="assistant", content="${subtask}", stream="low_level", target=True),
]
)
transition = create_transition(
complementary_data={
"task": "pick up the cup",
"timestamp": torch.tensor(0.5),
"index": torch.tensor(0),
"language_persistent": [
{
"role": "assistant",
"content": "reach for the cup",
"style": "subtask",
"timestamp": 0.0,
"camera": None,
"tool_calls": None,
},
{
"role": "user",
"content": "positive",
"style": "advantage",
"timestamp": 0.1,
"camera": None,
"tool_calls": None,
},
],
"language_events": [],
}
)
# Step 1: Render recipe
rendered = RenderMessagesStep(recipe=recipe)(transition)
# Step 2: Flatten to task string
out = RenderedMessagesToTaskStep()(rendered)
data = out[TransitionKey.COMPLEMENTARY_DATA]
assert "pick up the cup" in data["task"]
assert "Advantage: positive" in data["task"]
def test_full_render_advantage_absent_skips_turn():
"""When advantage row is absent, the advantage turn is skipped via if_present."""
recipe = TrainingRecipe(
messages=[
MessageTurn(role="user", content="${task}", stream="high_level"),
MessageTurn(
role="user",
content="Advantage: ${advantage}",
stream="high_level",
if_present="advantage",
),
MessageTurn(role="assistant", content="${subtask}", stream="low_level", target=True),
]
)
transition = create_transition(
complementary_data={
"task": "pick up the cup",
"timestamp": torch.tensor(0.5),
"index": torch.tensor(0),
"language_persistent": [
{
"role": "assistant",
"content": "reach for the cup",
"style": "subtask",
"timestamp": 0.0,
"camera": None,
"tool_calls": None,
},
],
"language_events": [],
}
)
rendered = RenderMessagesStep(recipe=recipe)(transition)
out = RenderedMessagesToTaskStep()(rendered)
data = out[TransitionKey.COMPLEMENTARY_DATA]
assert data["task"] == "pick up the cup"
assert "Advantage" not in data["task"]
@@ -0,0 +1,518 @@
# 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.
"""Tests for RECAP's distributional value function."""
from __future__ import annotations
import pytest
import torch
from lerobot.configs.rewards import RewardModelConfig
from lerobot.configs.types import FeatureType, NormalizationMode, PolicyFeature
from lerobot.rewards.distributional_value_function.configuration_distributional_value_function import (
DistributionalVFConfig,
)
from lerobot.types import TransitionKey
from lerobot.utils.constants import OBS_IMAGES
from tests.utils import skip_if_package_missing
BATCH_SIZE = 4
NUM_BINS = 201
IMAGE_KEY = f"{OBS_IMAGES}.top"
def _make_config(**overrides) -> DistributionalVFConfig:
defaults = {
"init_from_actor_path": "",
"device": "cpu",
"image_resolution": (224, 224),
}
defaults.update(overrides)
config = DistributionalVFConfig(**defaults)
config.input_features = {
IMAGE_KEY: PolicyFeature(type=FeatureType.VISUAL, shape=(3, 224, 224)),
}
config.output_features = {}
config.normalization_mapping = {
"VISUAL": NormalizationMode.IDENTITY,
}
return config
def _make_model():
from lerobot.rewards.distributional_value_function.modeling_distributional_value_function import (
DistributionalVFRewardModel,
)
return DistributionalVFRewardModel(_make_config())
def _make_batch(batch_size: int = BATCH_SIZE, device: str = "cpu") -> dict[str, torch.Tensor]:
from lerobot.utils.constants import OBS_LANGUAGE_ATTENTION_MASK, OBS_LANGUAGE_TOKENS
return {
IMAGE_KEY: torch.rand(batch_size, 3, 224, 224, device=device),
OBS_LANGUAGE_TOKENS: torch.randint(0, 1000, (batch_size, 16), device=device),
OBS_LANGUAGE_ATTENTION_MASK: torch.ones(batch_size, 16, dtype=torch.bool, device=device),
"mc_return": torch.rand(batch_size, device=device) * -1.0,
"is_terminal": torch.zeros(batch_size, dtype=torch.bool, device=device),
}
def test_config_registered_in_reward_model_registry():
"""DistributionalVFConfig is discoverable via RewardModelConfig registry."""
known = RewardModelConfig.get_known_choices()
assert "distributional_value_function" in known
def test_factory_returns_correct_class():
"""get_reward_model_class returns DistributionalVFRewardModel."""
from lerobot.rewards.factory import get_reward_model_class
cls = get_reward_model_class("distributional_value_function")
from lerobot.rewards.distributional_value_function.modeling_distributional_value_function import (
DistributionalVFRewardModel,
)
assert cls is DistributionalVFRewardModel
def test_make_reward_model_config_factory():
"""make_reward_model_config creates DistributionalVFConfig with overrides."""
from lerobot.rewards.factory import make_reward_model_config
config = make_reward_model_config("distributional_value_function", num_value_bins=101)
assert isinstance(config, DistributionalVFConfig)
assert config.num_value_bins == 101
@skip_if_package_missing("transformers")
def test_hl_gauss_sums_to_one():
"""HL-Gauss target distribution sums to 1 for each sample."""
model = _make_model()
targets = torch.tensor([-0.5, -0.1, -0.9, -0.0])
dist = model.hl_gauss_target(targets)
assert dist.shape == (4, NUM_BINS)
torch.testing.assert_close(dist.sum(dim=-1), torch.ones(4), atol=1e-5, rtol=0)
@skip_if_package_missing("transformers")
def test_hl_gauss_non_negative():
"""HL-Gauss target probabilities are all non-negative."""
model = _make_model()
targets = torch.linspace(-1.0, 0.0, 10)
dist = model.hl_gauss_target(targets)
assert (dist >= 0).all()
@skip_if_package_missing("transformers")
def test_hl_gauss_expected_value_matches():
"""E[V] under HL-Gauss distribution matches the target value."""
model = _make_model()
targets = torch.tensor([-0.5, -0.1, -0.9])
dist = model.hl_gauss_target(targets)
expected = (dist * model.bin_centers).sum(dim=-1)
torch.testing.assert_close(expected, targets, atol=1e-4, rtol=0)
@skip_if_package_missing("transformers")
def test_hl_gauss_handles_2d_input():
"""HL-Gauss handles [batch_size, 1] shaped inputs correctly."""
model = _make_model()
targets = torch.tensor([-0.5, -0.3]).unsqueeze(-1)
dist = model.hl_gauss_target(targets)
assert dist.shape == (2, NUM_BINS)
torch.testing.assert_close(dist.sum(dim=-1), torch.ones(2), atol=1e-5, rtol=0)
@skip_if_package_missing("transformers")
def test_dirac_delta_sums_to_one():
"""Dirac delta target distribution sums to 1 for each sample."""
model = _make_model()
targets = torch.tensor([-0.5, -0.1, -0.9, -1.0, 0.0])
dist = model.dirac_delta_target(targets)
assert dist.shape == (5, NUM_BINS)
torch.testing.assert_close(dist.sum(dim=-1), torch.ones(5), atol=1e-6, rtol=0)
@skip_if_package_missing("transformers")
def test_dirac_delta_at_most_two_nonzero():
"""Dirac delta places probability on at most two adjacent bins."""
model = _make_model()
targets = torch.tensor([-0.7523, -0.0013])
dist = model.dirac_delta_target(targets)
for i in range(2):
assert (dist[i] > 0).sum() <= 2
@skip_if_package_missing("transformers")
def test_dirac_delta_expected_value_matches():
"""E[V] under Dirac delta distribution matches the target value."""
model = _make_model()
targets = torch.tensor([-0.5, -0.1, -0.9])
dist = model.dirac_delta_target(targets)
expected = (dist * model.bin_centers).sum(dim=-1)
torch.testing.assert_close(expected, targets, atol=1e-5, rtol=0)
@skip_if_package_missing("transformers")
def test_dirac_delta_boundary_values_clamped():
"""Values outside support are clamped to boundary bins."""
model = _make_model()
targets = torch.tensor([-1.5, 0.5])
dist = model.dirac_delta_target(targets)
torch.testing.assert_close(dist.sum(dim=-1), torch.ones(2), atol=1e-6, rtol=0)
assert dist[0, 0] == 1.0
assert dist[1, -1] == 1.0
@skip_if_package_missing("transformers")
def test_one_hot_single_nonzero():
"""One-hot target has exactly one non-zero bin per sample."""
model = _make_model()
targets = torch.tensor([-0.5, -0.1, -1.0, 0.0])
dist = model.one_hot_target(targets)
assert dist.shape == (4, NUM_BINS)
for i in range(4):
assert (dist[i] > 0).sum() == 1
assert dist[i].sum() == 1.0
@skip_if_package_missing("transformers")
def test_one_hot_nearest_bin():
"""One-hot target activates the bin closest to the target value."""
model = _make_model()
targets = torch.tensor([-0.5])
dist = model.one_hot_target(targets)
hot_idx = dist[0].argmax()
assert model.bin_centers[hot_idx].item() == pytest.approx(-0.5, abs=0.003)
@skip_if_package_missing("transformers")
def test_terminal_gets_one_hot():
"""Terminal states receive one-hot targets; non-terminal get HL-Gauss."""
model = _make_model()
targets = torch.tensor([-0.5, -0.3, -0.7, -0.9])
is_terminal = torch.tensor([False, True, False, True])
dist = model.compute_target_distribution(
targets, is_terminal, method="hl_gauss", use_one_hot_terminal=True
)
for i in range(4):
assert dist[i].sum().item() == pytest.approx(1.0, abs=1e-5)
assert (dist[1] > 0).sum() == 1
assert (dist[3] > 0).sum() == 1
assert (dist[0] > 0).sum() > 2
assert (dist[2] > 0).sum() > 2
@skip_if_package_missing("transformers")
def test_no_terminal_override_when_disabled():
"""When use_one_hot_terminal=False, terminal states use the base method."""
model = _make_model()
targets = torch.tensor([-0.5, -0.3])
is_terminal = torch.tensor([False, True])
dist = model.compute_target_distribution(
targets, is_terminal, method="hl_gauss", use_one_hot_terminal=False
)
assert (dist[1] > 0).sum() > 2
@skip_if_package_missing("transformers")
def test_model_has_expected_components():
"""Model scaffold contains all architectural components."""
model = _make_model()
assert hasattr(model, "vision_tower")
assert hasattr(model, "multi_modal_projector")
assert hasattr(model, "token_embedding")
assert hasattr(model, "layers")
assert hasattr(model, "value_head")
assert hasattr(model, "cls_embedding")
assert hasattr(model, "norm")
assert hasattr(model, "rotary_emb")
assert hasattr(model, "bin_centers")
@skip_if_package_missing("transformers")
def test_model_bin_centers_shape():
"""Bin centers buffer has shape (num_value_bins,)."""
model = _make_model()
assert model.bin_centers.shape == (NUM_BINS,)
@skip_if_package_missing("transformers")
def test_model_layer_count():
"""Transformer has num_hidden_layers (6) layers."""
model = _make_model()
assert len(model.layers) == 6
@skip_if_package_missing("transformers")
def test_model_value_head_output_dim():
"""Value head outputs num_value_bins logits."""
model = _make_model()
assert model.value_head.out_features == NUM_BINS
@skip_if_package_missing("transformers")
def test_forward_returns_loss_and_dict():
"""Forward pass returns a finite scalar loss and output dict with expected keys."""
model = _make_model()
batch = _make_batch()
loss, output_dict = model.forward(batch)
assert loss.shape == ()
assert torch.isfinite(loss)
assert "loss" in output_dict
assert "predicted_value_mean" in output_dict
assert "mc_return_mean" in output_dict
@skip_if_package_missing("transformers")
def test_forward_loss_is_positive():
"""Cross-entropy loss is strictly positive for random weights."""
model = _make_model()
batch = _make_batch()
loss, _ = model.forward(batch)
assert loss.item() > 0
@skip_if_package_missing("transformers")
def test_compute_reward_returns_correct_shape():
"""compute_reward returns [batch_size] tensor of finite float32 values."""
model = _make_model()
model.eval()
batch = _make_batch(batch_size=3)
with torch.no_grad():
values = model.compute_reward(batch)
assert values.shape == (3,)
assert values.dtype == torch.float32
assert torch.isfinite(values).all()
@skip_if_package_missing("transformers")
def test_compute_reward_values_in_support_range():
"""Predicted values lie within [value_support_min, value_support_max]."""
model = _make_model()
model.eval()
batch = _make_batch(batch_size=8)
with torch.no_grad():
values = model.compute_reward(batch)
assert (values >= -1.0 - 0.01).all()
assert (values <= 0.0 + 0.01).all()
@skip_if_package_missing("transformers")
def test_processor_pipeline_produces_expected_keys():
"""Full preprocessor pipeline produces tokenized text and processed images."""
from lerobot.rewards.distributional_value_function.processor_distributional_value_function import (
make_distributional_vf_pre_post_processors,
)
from lerobot.utils.constants import OBS_LANGUAGE_ATTENTION_MASK, OBS_LANGUAGE_TOKENS
config = _make_config()
preprocessor, _ = make_distributional_vf_pre_post_processors(config)
raw_batch = {
IMAGE_KEY: torch.rand(3, 224, 224),
"task": "pick up the cup",
}
processed = preprocessor(raw_batch)
assert OBS_LANGUAGE_TOKENS in processed
assert OBS_LANGUAGE_ATTENTION_MASK in processed
assert IMAGE_KEY in processed
@skip_if_package_missing("transformers")
def test_gradient_flows_through_value_head():
"""Backprop produces non-zero gradients on the value head."""
model = _make_model()
model.train()
batch = _make_batch()
loss, _ = model.forward(batch)
loss.backward()
assert model.value_head.weight.grad is not None
assert not torch.all(model.value_head.weight.grad == 0)
@skip_if_package_missing("transformers")
def test_gradient_flows_through_cls_embedding():
"""Backprop produces non-zero gradients on the learned [CLS] embedding."""
model = _make_model()
model.train()
batch = _make_batch()
loss, _ = model.forward(batch)
loss.backward()
assert model.cls_embedding.grad is not None
assert not torch.all(model.cls_embedding.grad == 0)
def test_config_requires_visual_feature():
"""validate_features raises if no VISUAL feature is present."""
config = DistributionalVFConfig(init_from_actor_path="")
config.input_features = {
"observation.state": PolicyFeature(type=FeatureType.STATE, shape=(14,)),
}
with pytest.raises(ValueError, match="VISUAL"):
config.validate_features()
def test_config_passes_with_visual_feature():
"""validate_features succeeds when a VISUAL feature is present."""
config = _make_config()
config.validate_features()
@skip_if_package_missing("transformers")
def test_save_load_pretrained_roundtrip(tmp_path):
"""Saved model can be loaded back with identical weights."""
from lerobot.rewards.distributional_value_function.modeling_distributional_value_function import (
DistributionalVFRewardModel,
)
model = _make_model()
model._save_pretrained(tmp_path)
loaded = DistributionalVFRewardModel.from_pretrained(str(tmp_path))
orig_sd = model.state_dict()
loaded_sd = loaded.state_dict()
assert set(orig_sd.keys()) == set(loaded_sd.keys())
for key in orig_sd:
torch.testing.assert_close(orig_sd[key], loaded_sd[key], msg=f"Mismatch in {key}")
@skip_if_package_missing("transformers")
def test_image_preprocessor_normalizes_to_minus_one_one():
"""Image preprocessor scales [0, 1] float input to [-1, 1] for SigLIP."""
from lerobot.rewards.distributional_value_function.processor_distributional_value_function import (
DistributionalVFImagePreprocessorStep,
)
step = DistributionalVFImagePreprocessorStep(image_resolution=(224, 224), image_keys=(IMAGE_KEY,))
transition = {
TransitionKey.OBSERVATION: {
IMAGE_KEY: torch.rand(1, 224, 224, 3),
},
}
result = step(transition)
image = result[TransitionKey.OBSERVATION][IMAGE_KEY]
assert image.min() >= -1.0 - 1e-5
assert image.max() <= 1.0 + 1e-5
@skip_if_package_missing("transformers")
def test_image_preprocessor_resizes_with_pad():
"""Image preprocessor resizes non-square images to target resolution."""
from lerobot.rewards.distributional_value_function.processor_distributional_value_function import (
DistributionalVFImagePreprocessorStep,
)
step = DistributionalVFImagePreprocessorStep(image_resolution=(224, 224), image_keys=(IMAGE_KEY,))
transition = {
TransitionKey.OBSERVATION: {
IMAGE_KEY: torch.rand(1, 480, 640, 3),
},
}
result = step(transition)
image = result[TransitionKey.OBSERVATION][IMAGE_KEY]
assert image.shape[1:3] == (224, 224)
def test_task_prompt_formats_correctly():
"""Task prompt step converts underscored task to 'Task: {text}.' format."""
from lerobot.rewards.distributional_value_function.processor_distributional_value_function import (
DistributionalVFPrepareTaskPromptStep,
)
step = DistributionalVFPrepareTaskPromptStep()
transition = {
TransitionKey.COMPLEMENTARY_DATA: {"task": ["pick_up_the_cup"]},
}
result = step(transition)
prompt = result[TransitionKey.COMPLEMENTARY_DATA]["task"][0]
assert prompt == "Task: pick up the cup."
def test_task_prompt_handles_string_input():
"""Task prompt step accepts a plain string (not just a list)."""
from lerobot.rewards.distributional_value_function.processor_distributional_value_function import (
DistributionalVFPrepareTaskPromptStep,
)
step = DistributionalVFPrepareTaskPromptStep()
transition = {
TransitionKey.COMPLEMENTARY_DATA: {"task": "open_drawer"},
}
result = step(transition)
prompt = result[TransitionKey.COMPLEMENTARY_DATA]["task"][0]
assert prompt == "Task: open drawer."
def test_task_prompt_raises_on_missing_task():
"""Task prompt step raises ValueError when task key is absent."""
from lerobot.rewards.distributional_value_function.processor_distributional_value_function import (
DistributionalVFPrepareTaskPromptStep,
)
step = DistributionalVFPrepareTaskPromptStep()
transition = {
TransitionKey.COMPLEMENTARY_DATA: {},
}
with pytest.raises(ValueError, match="No task found"):
step(transition)
+514
View File
@@ -0,0 +1,514 @@
#!/usr/bin/env python
# Copyright 2026 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.
"""Tests for lerobot-compute-returns script."""
import json
from pathlib import Path
import numpy as np
import pyarrow as pa
import pyarrow.parquet as pq
import pytest
from lerobot.scripts.lerobot_compute_returns import (
IS_TERMINAL_COL,
MC_RETURN_COL,
ComputeReturnsConfig,
_get_episode_success,
compute_episode_returns,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def parquet_dataset(tmp_path):
"""Build a minimal parquet shard + info.json for testing I/O logic.
Mirrors the lerobot-rollout DAgger convention: ``next.success`` is False
on all frames except the terminal frame of successful episodes.
Even episodes are successful, odd episodes are failures.
"""
num_episodes = 3
frames_per_ep = 10
root = tmp_path / "test_dataset"
data_dir = root / "data" / "chunk-000"
meta_dir = root / "meta"
data_dir.mkdir(parents=True)
meta_dir.mkdir(parents=True)
all_rows = []
episodes_meta = []
global_idx = 0
for ep in range(num_episodes):
ep_from = global_idx
is_successful = ep % 2 == 0
for frame in range(frames_per_ep):
is_last_frame = frame == frames_per_ep - 1
all_rows.append(
{
"episode_index": ep,
"frame_index": frame,
"index": global_idx,
"next.success": is_successful and is_last_frame,
}
)
global_idx += 1
ep_to = global_idx
episodes_meta.append(
{
"episode_index": ep,
"length": frames_per_ep,
"dataset_from_index": ep_from,
"dataset_to_index": ep_to,
}
)
table = pa.table(
{
"episode_index": [r["episode_index"] for r in all_rows],
"frame_index": [r["frame_index"] for r in all_rows],
"index": [r["index"] for r in all_rows],
"next.success": [r["next.success"] for r in all_rows],
}
)
parquet_path = data_dir / "episode_000000.parquet"
pq.write_table(table, parquet_path)
info = {
"codebase_version": "v3.0",
"total_episodes": num_episodes,
"total_frames": global_idx,
"fps": 30,
"features": {
"episode_index": {"dtype": "int64", "shape": [1], "names": None},
"frame_index": {"dtype": "int64", "shape": [1], "names": None},
"index": {"dtype": "int64", "shape": [1], "names": None},
"next.success": {"dtype": "bool", "shape": [1], "names": None},
},
}
(meta_dir / "info.json").write_text(json.dumps(info, indent=2))
return root, parquet_path, episodes_meta
def _rewrite_shard(parquet_path: Path, episodes_meta: list[dict], config: ComputeReturnsConfig):
"""Rewrite a single parquet shard using the core logic from compute_returns."""
table = pq.read_table(parquet_path)
if not config.force and IS_TERMINAL_COL in table.column_names:
return
all_is_terminal = np.zeros(len(table), dtype=bool)
all_mc_return = np.zeros(len(table), dtype=np.float32)
episode_col = table.column("episode_index").to_pylist()
for ep_info in episodes_meta:
ep_idx = ep_info["episode_index"]
ep_len = ep_info["length"]
mask = np.array([v == ep_idx for v in episode_col], dtype=bool)
local_indices = np.where(mask)[0]
ep_subtable = table.filter(mask)
success = _get_episode_success(ep_subtable, config.success_key, config.default_success)
is_terminal, mc_return = compute_episode_returns(
num_frames=ep_len,
success=success,
c_fail=config.c_fail,
gamma=config.gamma,
max_episode_length=config.max_episode_length or ep_len,
)
all_is_terminal[local_indices] = is_terminal
all_mc_return[local_indices] = mc_return
if IS_TERMINAL_COL in table.column_names:
table = table.drop(IS_TERMINAL_COL)
if MC_RETURN_COL in table.column_names:
table = table.drop(MC_RETURN_COL)
table = table.append_column(IS_TERMINAL_COL, pa.array(all_is_terminal))
table = table.append_column(MC_RETURN_COL, pa.array(all_mc_return))
pq.write_table(table, parquet_path)
# ---------------------------------------------------------------------------
# Tests: compute_episode_returns (pure math, no I/O)
# ---------------------------------------------------------------------------
def test_successful_episode_terminal_reward_is_zero():
"""Terminal MC return for a successful episode should be 0."""
_, mc_return = compute_episode_returns(
num_frames=10, success=True, c_fail=50.0, gamma=1.0, max_episode_length=10
)
assert mc_return[-1] == pytest.approx(0.0, abs=1e-6)
def test_failed_episode_terminal_reward_reflects_cfail():
"""Terminal MC return for a failed episode should be -C_fail / H."""
horizon = 100
c_fail = 50.0
_, mc_return = compute_episode_returns(
num_frames=10, success=False, c_fail=c_fail, gamma=1.0, max_episode_length=horizon
)
assert mc_return[-1] == pytest.approx(-c_fail / horizon, abs=1e-5)
def test_is_terminal_only_last_frame():
"""Only the last frame of an episode should be marked terminal."""
is_terminal, _ = compute_episode_returns(
num_frames=20, success=True, c_fail=50.0, gamma=1.0, max_episode_length=20
)
assert is_terminal[-1] == True # noqa: E712
assert not any(is_terminal[:-1])
def test_mc_return_monotonically_increases_for_success():
"""For a successful undiscounted episode, returns should increase toward 0."""
_, mc_return = compute_episode_returns(
num_frames=50, success=True, c_fail=50.0, gamma=1.0, max_episode_length=50
)
for i in range(len(mc_return) - 1):
assert mc_return[i] <= mc_return[i + 1]
def test_mc_return_bounded_negative_to_zero():
"""MC returns for successful episodes should be in (-1, 0]."""
_, mc_return = compute_episode_returns(
num_frames=100, success=True, c_fail=50.0, gamma=1.0, max_episode_length=100
)
assert mc_return[-1] == pytest.approx(0.0, abs=1e-6)
assert all(v <= 0.0 for v in mc_return)
assert all(v >= -1.0 - 1e-6 for v in mc_return)
def test_first_frame_return_success():
"""First frame return for successful episode equals -(N-1)/H."""
num_frames = 10
horizon = 10
_, mc_return = compute_episode_returns(
num_frames=num_frames, success=True, c_fail=50.0, gamma=1.0, max_episode_length=horizon
)
expected = -(num_frames - 1) / horizon
assert mc_return[0] == pytest.approx(expected, abs=1e-5)
def test_first_frame_return_failure():
"""First frame return for failed episode includes the failure penalty."""
num_frames = 10
horizon = 100
c_fail = 50.0
_, mc_return = compute_episode_returns(
num_frames=num_frames, success=False, c_fail=c_fail, gamma=1.0, max_episode_length=horizon
)
expected = (-(num_frames - 1) / horizon) + (-c_fail / horizon)
assert mc_return[0] == pytest.approx(expected, abs=1e-5)
def test_discount_factor_less_than_one():
"""Discount factor < 1 should make earlier frames have smaller magnitude."""
_, mc_undiscounted = compute_episode_returns(
num_frames=20, success=True, c_fail=50.0, gamma=1.0, max_episode_length=20
)
_, mc_discounted = compute_episode_returns(
num_frames=20, success=True, c_fail=50.0, gamma=0.99, max_episode_length=20
)
assert abs(mc_discounted[0]) < abs(mc_undiscounted[0])
def test_single_frame_episode_success():
"""Single-frame successful episode: return should be 0."""
is_terminal, mc_return = compute_episode_returns(
num_frames=1, success=True, c_fail=50.0, gamma=1.0, max_episode_length=1
)
assert mc_return[0] == pytest.approx(0.0, abs=1e-6)
assert is_terminal[0] == True # noqa: E712
def test_single_frame_episode_failure():
"""Single-frame failed episode: return should be -C_fail/H."""
horizon = 100
c_fail = 50.0
is_terminal, mc_return = compute_episode_returns(
num_frames=1, success=False, c_fail=c_fail, gamma=1.0, max_episode_length=horizon
)
assert mc_return[0] == pytest.approx(-c_fail / horizon, abs=1e-5)
assert is_terminal[0] == True # noqa: E712
def test_horizon_normalization_scales_returns():
"""Larger horizon should scale down the per-step penalty."""
_, mc_small_h = compute_episode_returns(
num_frames=10, success=True, c_fail=50.0, gamma=1.0, max_episode_length=10
)
_, mc_large_h = compute_episode_returns(
num_frames=10, success=True, c_fail=50.0, gamma=1.0, max_episode_length=100
)
assert abs(mc_large_h[0]) < abs(mc_small_h[0])
# ---------------------------------------------------------------------------
# Tests: _get_episode_success (in-memory PyArrow tables)
# ---------------------------------------------------------------------------
def test_default_success_overrides_column():
"""default_success should override any column value."""
table = pa.table({"next.success": [True, True, True]})
assert _get_episode_success(table, "next.success", default_success=False) is False
def test_reads_bool_column():
"""Should detect success via any() reduction over the column."""
table_success = pa.table({"next.success": [False, False, True]})
table_fail = pa.table({"next.success": [False, False, False]})
assert _get_episode_success(table_success, "next.success", None) is True
assert _get_episode_success(table_fail, "next.success", None) is False
def test_reads_int_column():
"""Should interpret integer success column (0/1) as bool via any()."""
table = pa.table({"task_success": [0, 0, 1]})
assert _get_episode_success(table, "task_success", None) is True
def test_all_zeros_means_failure():
"""An episode with all-zero success values is a failure."""
table = pa.table({"next.success": [0, 0, 0]})
assert _get_episode_success(table, "next.success", None) is False
def test_missing_column_defaults_to_true():
"""When success column is missing, assume success (demo data)."""
table = pa.table({"frame_index": [0, 1, 2]})
assert _get_episode_success(table, "next.success", None) is True
# ---------------------------------------------------------------------------
# Tests: parquet rewriting (integration, writes to disk)
# ---------------------------------------------------------------------------
def test_writes_columns_to_parquet(parquet_dataset):
"""The rewrite logic should add is_terminal and mc_return columns."""
root, parquet_path, episodes_meta = parquet_dataset
table_before = pq.read_table(parquet_path)
assert IS_TERMINAL_COL not in table_before.column_names
assert MC_RETURN_COL not in table_before.column_names
config = ComputeReturnsConfig(success_key="next.success", max_episode_length=10, force=True)
_rewrite_shard(parquet_path, episodes_meta, config)
table_after = pq.read_table(parquet_path)
assert IS_TERMINAL_COL in table_after.column_names
assert MC_RETURN_COL in table_after.column_names
def test_terminal_frames_correct(parquet_dataset):
"""Only the last frame of each episode should be terminal."""
root, parquet_path, episodes_meta = parquet_dataset
config = ComputeReturnsConfig(success_key="next.success", max_episode_length=10, force=True)
_rewrite_shard(parquet_path, episodes_meta, config)
table = pq.read_table(parquet_path)
is_terminal = table.column(IS_TERMINAL_COL).to_pylist()
terminal_indices = [i for i, v in enumerate(is_terminal) if v]
assert terminal_indices == [9, 19, 29]
def test_success_episodes_return_zero_at_terminal(tmp_path):
"""Successful episodes (ep 0) should have mc_return=0 at terminal."""
num_episodes = 2
frames_per_ep = 5
root = tmp_path / "test_dataset"
data_dir = root / "data" / "chunk-000"
meta_dir = root / "meta"
data_dir.mkdir(parents=True)
meta_dir.mkdir(parents=True)
all_rows = []
episodes_meta = []
global_idx = 0
for ep in range(num_episodes):
ep_from = global_idx
is_successful = ep % 2 == 0
for frame in range(frames_per_ep):
is_last_frame = frame == frames_per_ep - 1
all_rows.append(
{
"episode_index": ep,
"frame_index": frame,
"index": global_idx,
"next.success": is_successful and is_last_frame,
}
)
global_idx += 1
episodes_meta.append(
{
"episode_index": ep,
"length": frames_per_ep,
"dataset_from_index": ep_from,
"dataset_to_index": global_idx,
}
)
table = pa.table(
{
"episode_index": [r["episode_index"] for r in all_rows],
"frame_index": [r["frame_index"] for r in all_rows],
"index": [r["index"] for r in all_rows],
"next.success": [r["next.success"] for r in all_rows],
}
)
parquet_path = data_dir / "episode_000000.parquet"
pq.write_table(table, parquet_path)
info = {
"codebase_version": "v3.0",
"total_episodes": num_episodes,
"total_frames": global_idx,
"fps": 30,
"features": {
"episode_index": {"dtype": "int64", "shape": [1], "names": None},
"frame_index": {"dtype": "int64", "shape": [1], "names": None},
"index": {"dtype": "int64", "shape": [1], "names": None},
"next.success": {"dtype": "bool", "shape": [1], "names": None},
},
}
(meta_dir / "info.json").write_text(json.dumps(info, indent=2))
config = ComputeReturnsConfig(success_key="next.success", max_episode_length=5, force=True)
_rewrite_shard(parquet_path, episodes_meta, config)
table = pq.read_table(parquet_path)
mc_return = table.column(MC_RETURN_COL).to_pylist()
assert mc_return[4] == pytest.approx(0.0, abs=1e-5)
def test_failed_episodes_have_negative_terminal(tmp_path):
"""Failed episodes (ep 1) should have mc_return < 0 at terminal."""
num_episodes = 2
frames_per_ep = 5
root = tmp_path / "test_dataset"
data_dir = root / "data" / "chunk-000"
meta_dir = root / "meta"
data_dir.mkdir(parents=True)
meta_dir.mkdir(parents=True)
all_rows = []
episodes_meta = []
global_idx = 0
for ep in range(num_episodes):
ep_from = global_idx
is_successful = ep % 2 == 0
for frame in range(frames_per_ep):
is_last_frame = frame == frames_per_ep - 1
all_rows.append(
{
"episode_index": ep,
"frame_index": frame,
"index": global_idx,
"next.success": is_successful and is_last_frame,
}
)
global_idx += 1
episodes_meta.append(
{
"episode_index": ep,
"length": frames_per_ep,
"dataset_from_index": ep_from,
"dataset_to_index": global_idx,
}
)
table = pa.table(
{
"episode_index": [r["episode_index"] for r in all_rows],
"frame_index": [r["frame_index"] for r in all_rows],
"index": [r["index"] for r in all_rows],
"next.success": [r["next.success"] for r in all_rows],
}
)
parquet_path = data_dir / "episode_000000.parquet"
pq.write_table(table, parquet_path)
config = ComputeReturnsConfig(success_key="next.success", max_episode_length=5, c_fail=50.0, force=True)
_rewrite_shard(parquet_path, episodes_meta, config)
table = pq.read_table(parquet_path)
mc_return = table.column(MC_RETURN_COL).to_pylist()
assert mc_return[9] < 0.0
def test_idempotent_with_force_flag(parquet_dataset):
"""Running twice with force should produce identical results."""
root, parquet_path, episodes_meta = parquet_dataset
config = ComputeReturnsConfig(success_key="next.success", max_episode_length=10, force=True)
_rewrite_shard(parquet_path, episodes_meta, config)
table1 = pq.read_table(parquet_path)
mc1 = table1.column(MC_RETURN_COL).to_pylist()
_rewrite_shard(parquet_path, episodes_meta, config)
table2 = pq.read_table(parquet_path)
mc2 = table2.column(MC_RETURN_COL).to_pylist()
assert mc1 == mc2
def test_skips_if_columns_exist_without_force(parquet_dataset):
"""Without force, existing columns should not be overwritten."""
root, parquet_path, episodes_meta = parquet_dataset
config = ComputeReturnsConfig(success_key="next.success", max_episode_length=10, force=True)
_rewrite_shard(parquet_path, episodes_meta, config)
table = pq.read_table(parquet_path)
original_mc = table.column(MC_RETURN_COL).to_pylist()
config_no_force = ComputeReturnsConfig(success_key="next.success", max_episode_length=20, force=False)
_rewrite_shard(parquet_path, episodes_meta, config_no_force)
table2 = pq.read_table(parquet_path)
assert table2.column(MC_RETURN_COL).to_pylist() == original_mc
def test_updates_info_json(parquet_dataset):
"""info.json should be updated with is_terminal and mc_return features."""
from lerobot.scripts.lerobot_compute_returns import _update_info_json
root, parquet_path, episodes_meta = parquet_dataset
_update_info_json(root, None)
info_path = root / "meta" / "info.json"
info = json.loads(info_path.read_text())
assert IS_TERMINAL_COL in info["features"]
assert MC_RETURN_COL in info["features"]
assert info["features"][IS_TERMINAL_COL]["dtype"] == "bool"
assert info["features"][MC_RETURN_COL]["dtype"] == "float32"
+97
View File
@@ -338,6 +338,103 @@ def test_dagger_events_reset():
assert not events.upload_requested.is_set()
def test_dagger_mark_success():
"""mark_success sets the episode label to True."""
from lerobot.rollout.strategies import DAggerEvents
events = DAggerEvents()
assert events.consume_episode_success() is None
events.mark_success()
assert events.consume_episode_success() is True
# Consuming clears the label
assert events.consume_episode_success() is None
def test_dagger_mark_failure():
"""mark_failure sets the episode label to False."""
from lerobot.rollout.strategies import DAggerEvents
events = DAggerEvents()
events.mark_failure()
assert events.consume_episode_success() is False
def test_dagger_success_overrides_failure():
"""Last label wins — success after failure overrides."""
from lerobot.rollout.strategies import DAggerEvents
events = DAggerEvents()
events.mark_failure()
events.mark_success()
assert events.consume_episode_success() is True
def test_dagger_reset_clears_success_label():
"""reset() clears any pending episode success label."""
from lerobot.rollout.strategies import DAggerEvents
events = DAggerEvents()
events.mark_success()
events.reset()
assert events.consume_episode_success() is None
def test_stamp_episode_success_labels_terminal_frame():
"""_stamp_episode_success sets last frame's next.success to True."""
import numpy as np
from lerobot.rollout.strategies.dagger import DAggerStrategy
strategy = DAggerStrategy.__new__(DAggerStrategy)
strategy.config = MagicMock()
from lerobot.rollout.strategies import DAggerEvents
strategy._events = DAggerEvents()
strategy._events.mark_success()
dataset = MagicMock()
dataset.writer.episode_buffer = {
"next.success": [
np.array([False], dtype=bool),
np.array([False], dtype=bool),
np.array([False], dtype=bool),
],
}
strategy._stamp_episode_success(dataset)
assert dataset.writer.episode_buffer["next.success"][-1].item() is True
assert dataset.writer.episode_buffer["next.success"][0].item() is False
def test_stamp_episode_success_no_label_stays_false():
"""Without a label, all frames remain False."""
import numpy as np
from lerobot.rollout.strategies.dagger import DAggerStrategy
strategy = DAggerStrategy.__new__(DAggerStrategy)
strategy.config = MagicMock()
from lerobot.rollout.strategies import DAggerEvents
strategy._events = DAggerEvents()
dataset = MagicMock()
dataset.writer.episode_buffer = {
"next.success": [
np.array([False], dtype=bool),
np.array([False], dtype=bool),
],
}
strategy._stamp_episode_success(dataset)
assert all(v.item() is False for v in dataset.writer.episode_buffer["next.success"])
# ---------------------------------------------------------------------------
# Context dataclass
# ---------------------------------------------------------------------------
Generated
+1243 -1474
View File
File diff suppressed because it is too large Load Diff