Skip to content

vllm_omni.diffusion.models.pi0

π0 (Pi-Zero) VLA model for vllm-omni.

PaliGemma (SigLIP vision + Gemma-2B LM) + Gemma-300M action expert + a flow-matching action head. Outputs a continuous action chunk [horizon, action_dim] rather than tokens.

Modules:

Name Description
config

Config surface for the π0 (Pi-Zero) VLA model in vllm-omni.

modeling_pi0

Inference-only π0 (Pi-Zero) VLA math kernel for vllm-omni.

pipeline_pi0

π0 (Pi-Zero) VLA pipeline for vllm-omni.

processor_pi0

Preprocessing for the π0 VLA model.

GemmaVariantConfig

depth instance-attribute

depth = depth

head_dim instance-attribute

head_dim = head_dim

mlp_dim instance-attribute

mlp_dim = mlp_dim

num_heads instance-attribute

num_heads = num_heads

num_kv_heads instance-attribute

num_kv_heads = num_kv_heads

width instance-attribute

width = width

PaliGemmaWithActionExpert

Bases: Module

Dual-backbone transformer: PaliGemma (Gemma 2B) + Action Expert (Gemma 300M).

forward has two inference modes: - prefix_only: inputs_embeds=[prefix, None] + use_cache=True → compute prefix hidden states + a layer-wise K/V cache. - suffix_only: inputs_embeds=[None, suffix] + the cache from the previous prefix pass → compute expert hidden states with cross-attention over the concatenated (prefix, suffix) K/V.

Both modes walk Gemma decoder layers manually through _compute_layer_* instead of GemmaModel.forward — that way we own the attention mask handling and the KV cache format (a plain list[(k, v)] one entry per layer).

Ref: lerobot/policies/pi0/modeling_pi0.py PaliGemmaWithExpertModel

gemma_expert instance-attribute

gemma_expert = GemmaForCausalLM(
    config=action_expert_config_hf
)

paligemma instance-attribute

paligemma = PaliGemmaForConditionalGeneration(
    config=vlm_config_hf
)

embed_image

embed_image(pixel_values: Tensor) -> Tensor

Encode images with SigLIP vision tower + PaliGemma projector.

We run the two steps explicitly instead of calling PaliGemmaModel.get_image_features because that convenience method historically has returned either the projected tensor (old) or the raw vision-tower output (new). Being explicit makes us independent of which transformers release we're on.

embed_language_tokens

embed_language_tokens(tokens: Tensor) -> Tensor

Embed language tokens with PaliGemma's embedding table, returning the * sqrt(hidden)-scaled embedding (PaliGemma/Gemma convention).

The scaling location moved across transformers releases
  • ≤ 5.3: embed_tokens is a plain nn.Embedding and the * sqrt(hidden) normalizer lives inside GemmaModel.forward (which we bypass) — so we apply it explicitly here.
  • ≥ 5.4: embed_tokens is a GemmaTextScaledWordEmbedding that self-applies embed_scale = hidden_size ** 0.5 — applying it again would double-scale (≈45×). We detect this and skip the manual scale.

This makes embed_language_tokens return the canonical scaled embedding on every transformers version.

forward

forward(
    attention_mask: Tensor | None = None,
    position_ids: LongTensor | None = None,
    past_key_values=None,
    inputs_embeds: list[Tensor | None] | None = None,
    use_cache: bool = False,
)

Dispatch to prefix_only / suffix_only and return ([prefix_out, suffix_out], past_key_values_or_None).

Pi0Config dataclass

π0 VLA config (dataclass, not an HF PretrainedConfig).

Mirrors the runtime-relevant subset of LeRobot PI0Config plus a couple of serving-side knobs (max_cameras, image_feature_keys, image_key_map, norm_stats).

action_expert_variant class-attribute instance-attribute

action_expert_variant: str = 'gemma_300m'

chunk_size class-attribute instance-attribute

chunk_size: int = 50

dtype class-attribute instance-attribute

dtype: str = 'float32'

image_feature_keys class-attribute instance-attribute

image_feature_keys: list[str] | None = None

image_key_map class-attribute instance-attribute

image_key_map: dict[str, str] = field(default_factory=dict)

image_resolution class-attribute instance-attribute

image_resolution: tuple[int, int] = (224, 224)

input_features class-attribute instance-attribute

input_features: dict[str, Any] = field(default_factory=dict)

max_action_dim class-attribute instance-attribute

max_action_dim: int = 32

max_cameras class-attribute instance-attribute

max_cameras: int = 3

max_period class-attribute instance-attribute

max_period: float = 4.0

max_state_dim class-attribute instance-attribute

max_state_dim: int = 32

min_period class-attribute instance-attribute

min_period: float = 0.004

norm_stats class-attribute instance-attribute

norm_stats: dict | None = None

num_inference_steps class-attribute instance-attribute

num_inference_steps: int = 10

output_features class-attribute instance-attribute

output_features: dict[str, Any] = field(
    default_factory=dict
)

paligemma_variant class-attribute instance-attribute

paligemma_variant: str = 'gemma_2b'

tokenizer_max_length class-attribute instance-attribute

tokenizer_max_length: int = 48

from_model_config classmethod

from_model_config(
    model_config: dict[str, Any] | None,
) -> Pi0Config

Build from a config dict (LeRobot config.json or deploy yaml).

Keeps only the recognized dataclass fields (the LeRobot config also carries many training-only keys) and coerces image_resolution to a tuple.

from_pretrained classmethod

from_pretrained(checkpoint_dir: str | Path) -> Pi0Config

Build from a checkpoint directory's config.json.

Pi0ForActionPrediction

Bases: Module

π0 VLA model for robot action prediction via flow matching.

Inference flow
  1. Embed prefix (images + language) → prefix tokens.
  2. Forward prefix through PaliGemma → layer-wise KV cache.
  3. For each denoising step t = 1.0, 1-dt, ..., 0: a. Embed suffix (state + x_t + timestep) → suffix tokens. b. Forward suffix through the action expert with the prefix cache. c. x_t = x_t + dt * v_t (Euler integration).
  4. Return x_0 as the predicted action chunk.

action_dim instance-attribute

action_dim = getattr(
    config, "max_action_dim", DEFAULT_ACTION_DIM
)

action_horizon instance-attribute

action_horizon = getattr(
    config, "chunk_size", DEFAULT_ACTION_HORIZON
)

action_in_proj instance-attribute

action_in_proj = nn.Linear(
    self.action_dim, self.expert_width
)

action_out_proj instance-attribute

action_out_proj = nn.Linear(
    self.expert_width, self.action_dim
)

action_time_mlp_in instance-attribute

action_time_mlp_in = nn.Linear(
    2 * self.expert_width, self.expert_width
)

action_time_mlp_out instance-attribute

action_time_mlp_out = nn.Linear(
    self.expert_width, self.expert_width
)

config instance-attribute

config = config

expert_width instance-attribute

expert_width = expert_config.width

max_state_dim instance-attribute

max_state_dim = getattr(
    config, "max_state_dim", self.action_dim
)

num_inference_steps instance-attribute

num_inference_steps = getattr(
    config,
    "num_inference_steps",
    DEFAULT_NUM_INFERENCE_STEPS,
)

paligemma_with_expert instance-attribute

paligemma_with_expert = PaliGemmaWithActionExpert(
    vlm_config, expert_config
)

state_proj instance-attribute

state_proj = nn.Linear(
    self.max_state_dim, self.expert_width
)

vlm_width instance-attribute

vlm_width = vlm_config.width

denoise_step

denoise_step(
    state: Tensor,
    prefix_pad_masks: Tensor,
    past_key_values,
    x_t: Tensor,
    timestep: Tensor,
) -> Tensor

Apply one flow-matching denoising step: predict v_t from x_t.

Uses the prefix KV cache from sample_actions and only runs the action expert. Ref: openpi PI0Pytorch.denoise_step.

embed_prefix

embed_prefix(
    images: list[Tensor],
    image_masks: list[Tensor],
    lang_tokens: Tensor,
    lang_masks: Tensor,
) -> tuple[Tensor, Tensor, Tensor]

Build the prefix embeddings, per-token padding mask, and AR mask.

Prefix tokens form a contiguous sequence [img_cam_0..., img_cam_1..., ..., lang_tokens...] with bidirectional attention; the returned att_masks are all zeros because each token is free to attend to every other prefix token (the suffix pass will put a causal boundary right before the state token).

Ref: openpi PI0Pytorch.embed_prefix

embed_suffix

embed_suffix(
    state: Tensor, noisy_actions: Tensor, timestep: Tensor
) -> tuple[Tensor, Tensor, Tensor]

Build the suffix embeddings + masks: [state_token, action_tokens×H].

AR mask layout: [1, 1, 0, 0, ..., 0] — the state token is a causal boundary (no later token attends backwards through it onto prefix by mistake), the first action token starts a new causal block, and the rest of the action tokens attend to each other bidirectionally.

Ref: openpi PI0Pytorch.embed_suffix

load_weights

load_weights(weights: Iterable[tuple[str, Tensor]])

Load weights from an lerobot/pi0_base safetensors checkpoint.

Checkpoint keys (after stripping the leading model. prefix) look like paligemma_with_expert.paligemma.<sub>.*. Modern transformers nests those sub-modules one level deeper (paligemma.model.<sub>), and PaliGemma ties lm_head.weight with embed_tokens.weight at post_init time. Two remap rules make the load lossless:

  1. paligemma.{vision_tower,multi_modal_projector,language_model}.* → paligemma.model.{...}.*
  2. paligemma.lm_head.weight → paligemma.model.language_model.embed_tokens.weight (the checkpoint does not store embed_tokens.weight on its own — only the tied lm_head.weight copy — and PaliGemmaForConditionalGeneration does not register lm_head.weight as a Parameter when tied, so without this rule the language embedding would silently remain at its random init.)

On any remaining mismatch the loader logs a warning listing a sample of the unmatched checkpoint keys and model params; downstream parity tests surface these as a hard failure.

sample_actions

sample_actions(
    images: list[Tensor],
    image_masks: list[Tensor],
    lang_tokens: Tensor,
    lang_masks: Tensor,
    state: Tensor,
    noise: Tensor | None = None,
    num_steps: int | None = None,
) -> Tensor

Generate an action chunk via iterative flow-matching denoising.

Convention: t=1 is noise, t=0 is the target — opposite of the published π0 paper but matches both OpenPI and LeRobot. Ref: openpi PI0Pytorch.sample_actions

Pi0ImageProcessor

Minimal image preprocessor: image → normalized + padded [-1,1] tensor.

image_size instance-attribute

image_size = image_size

make_empty_image

make_empty_image() -> Tensor

Fill tensor for an unused camera slot — pure -1, matches OpenPI/LeRobot.

preprocess_single

preprocess_single(image: Any) -> Tensor

Accept a PIL image, an HWC uint8/float ndarray, or a CHW tensor and return a (1, 3, image_size, image_size) float tensor in [-1, 1].

Pi0Pipeline

Bases: Module

π0 VLA pipeline: raw robot obs → continuous action chunk.

Registered as "Pi0Pipeline" in the diffusion registry. Weights are self-loaded in __init__ from the checkpoint's model.safetensors via the kernel's load_weights (which handles the LeRobot key remaps).

config instance-attribute

config = self._build_config(od_config)

model instance-attribute

model = self._initialize_model()

model_dir instance-attribute

model_dir = self._resolve_model_dir(od_config.model)

od_config instance-attribute

od_config = od_config

prefix instance-attribute

prefix = prefix

tokenizer instance-attribute

tokenizer = self._load_tokenizer()

tokenizer_source instance-attribute

tokenizer_source = str(
    custom_args.get(
        "tokenizer", self._resolve_tokenizer_source()
    )
)

forward

forward(
    req: OmniDiffusionRequest, **kwargs
) -> DiffusionOutput

has_real_checkpoint

has_real_checkpoint() -> bool

load_weights

load_weights(weights=())

No-op for the diffusion loader: π0 self-loads its checkpoint in __init__ (the kernel's load_weights handles the LeRobot remaps). We expose no weights_sources, so the loader passes an empty iterator here; returning None skips its strict unloaded-weights check.

build_model_inputs

build_model_inputs(
    robot_obs: dict, config, tokenizer, device: device
)

Convert a raw robot observation into sample_actions inputs.

Reproduces the LeRobot camera ordering exactly: iterate config.image_feature_keys (the ordered camera identities from the checkpoint's input_features); for each, use the supplied image (mask True) or a -1-filled empty image (mask False). Then tokenize the prompt and zero-pad/truncate the state to max_state_dim.

Returns (images, image_masks, lang_tokens, lang_masks, state) with a leading batch dim of 1 on the tensors.

create_sinusoidal_pos_embedding

create_sinusoidal_pos_embedding(
    time: Tensor,
    dimension: int,
    min_period: float = 0.004,
    max_period: float = 4.0,
    device: device = None,
) -> Tensor

Compute a sine/cosine positional embedding for scalar timesteps.

Ref: openpi/models_pytorch/pi0_pytorch.py create_sinusoidal_pos_embedding

get_gemma_config

get_gemma_config(variant: str) -> GemmaVariantConfig

get_pi0_post_process_func

get_pi0_post_process_func(od_config: OmniDiffusionConfig)

π0 returns actions directly; post-processing is identity. The diffusion engine resolves this via the registry (_DIFFUSION_POST_PROCESS_FUNCS) and applies it engine-side, so the pipeline does NOT attach it to DiffusionOutput.

make_att_2d_masks

make_att_2d_masks(
    pad_masks: Tensor, att_masks: Tensor
) -> Tensor

Build a 2D attention mask from a padding mask and an autoregressive mask.

Ref: openpi/models_pytorch/pi0_pytorch.py make_att_2d_masks

pil_image_to_tensor

pil_image_to_tensor(image: Image) -> Tensor

PIL → (1, C, H, W) float32 in [-1, 1] (SigLIP normalization).

prepare_attention_masks_4d

prepare_attention_masks_4d(att_2d_masks: Tensor) -> Tensor

Convert (B, S, S) bool masks to (B, 1, S, S) float masks.

True → 0.0 (attend), False → OPENPI_ATTENTION_MASK_VALUE. Ref: openpi PI0Pytorch._prepare_attention_masks_4d

resize_with_pad

resize_with_pad(
    images: Tensor,
    target_height: int,
    target_width: int,
    mode: str = "bilinear",
) -> Tensor

Resize (B, C, H, W) images to the target shape, preserving aspect ratio with -1 padding on the short side.

Matches openpi image_tools.resize_with_pad_torch — the clamp to [-1, 1] is what lets the padded region blend with SigLIP-normalized pixels without adding signal at the boundary.

tokenize_prompt

tokenize_prompt(
    tokenizer,
    text: str,
    max_token_len: int = PI0_MAX_TOKEN_LEN,
)

Return (input_ids, attention_mask) lists, length max_token_len.

Matches LeRobot's Pi0NewLineProcessor: append \n if the caller didn't, then run the PaliGemma tokenizer with right-padding.