[Model] Add InternVLA-A1 offline inference support (#2737)

Signed-off-by: zhuyangkun <zyk_120403228@163.com>
Signed-off-by: bl-wang1 <918395762@qq.com>
Co-authored-by: wangbolun <918395762@qq.com>

examples/offline_inference/internvla_a1/README.md

@@ -0,0 +1,30 @@
+# InternVLA-A1
+
+Full usage and result-reporting guidance lives in [docs/user_guide/examples/offline_inference/internvla_a1.md](../../../docs/user_guide/examples/offline_inference/internvla_a1.md).
+
+Quick start:
+
+```bash
+export INTERNVLA_A1_MODEL_DIR=/path/to/InternVLA-A1-3B-ft-pen
+export INTERNVLA_A1_DATASET_DIR=/path/to/Genie1-Place_Markpen
+export INTERNVLA_A1_PROCESSOR_DIR=/path/to/Qwen3-VL-2B-Instruct
+# hf tenstep/Cosmos-Tokenizer-CI8x8-SafeTensors
+export INTERNVLA_A1_COSMOS_DIR=/path/to/Cosmos-Tokenizer-CI8x8-SafeTensor
+
+bash run.sh --num-samples 1 --num-episodes 0
+bash run.sh --num-episodes 1
+bash collect_results.sh
+```
+
+Expected files under `INTERNVLA_A1_COSMOS_DIR`:
+
+- `encoder.safetensors`
+- `decoder.safetensors`
+
+Reference Hugging Face repo: `tenstep/Cosmos-Tokenizer-CI8x8-SafeTensors`
+
+Key entrypoints:
+
+- `run.sh`: wrapper for offline inference and GT evaluation
+- `collect_results.sh`: collect sample output, latency, metrics, plots, and logs
+- `end2end.py`: underlying Python entrypoint

examples/offline_inference/internvla_a1/collect_results.sh

@@ -0,0 +1,220 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "$ROOT_DIR/../../.." && pwd)"
+TIMESTAMP="$(date +%Y%m%d_%H%M%S)"
+RESULT_ROOT="${INTERNVLA_A1_RESULT_DIR:-$REPO_ROOT/outputs/internvla_a1/collected_results/$TIMESTAMP}"
+
+: "${INTERNVLA_A1_MODEL_DIR:?Please export INTERNVLA_A1_MODEL_DIR=/path/to/InternVLA-A1-3B-ft-pen}"
+: "${INTERNVLA_A1_DATASET_DIR:?Please export INTERNVLA_A1_DATASET_DIR=/path/to/Genie1-Place_Markpen}"
+: "${INTERNVLA_A1_PROCESSOR_DIR:?Please export INTERNVLA_A1_PROCESSOR_DIR=/path/to/Qwen3-VL-2B-Instruct}"
+: "${INTERNVLA_A1_COSMOS_DIR:?Please export INTERNVLA_A1_COSMOS_DIR=/path/to/Cosmos-Tokenizer-CI8x8-SafeTensor}"
+
+EVAL_OUTPUT_DIR="$RESULT_ROOT/eval_outputs"
+GPU_MONITOR_INTERVAL="${GPU_MONITOR_INTERVAL:-1}"
+
+mkdir -p "$RESULT_ROOT"
+
+export INTERNVLA_A1_MODEL_DIR
+export INTERNVLA_A1_DATASET_DIR
+export INTERNVLA_A1_PROCESSOR_DIR
+export INTERNVLA_A1_COSMOS_DIR
+
+write_env_summary() {
+  cat >"$RESULT_ROOT/env_summary.txt" <<EOF
+timestamp=$TIMESTAMP
+repo_root=$REPO_ROOT
+result_root=$RESULT_ROOT
+python=$(command -v python || true)
+python_version=$(python --version 2>&1 || true)
+model_dir=$INTERNVLA_A1_MODEL_DIR
+dataset_dir=$INTERNVLA_A1_DATASET_DIR
+processor_dir=$INTERNVLA_A1_PROCESSOR_DIR
+cosmos_dir=$INTERNVLA_A1_COSMOS_DIR
+pwd=$(pwd)
+EOF
+}
+
+capture_gpu_snapshot() {
+  local output_file="$1"
+  if command -v nvidia-smi >/dev/null 2>&1; then
+    nvidia-smi --query-gpu=index,name,memory.total,memory.used,utilization.gpu --format=csv >"$output_file" 2>&1 || true
+  else
+    echo "nvidia-smi not found" >"$output_file"
+  fi
+}
+
+start_gpu_monitor() {
+  local output_file="$1"
+  if ! command -v nvidia-smi >/dev/null 2>&1; then
+    echo "nvidia-smi not found" >"$output_file"
+    return 1
+  fi
+
+  (
+    echo "timestamp,index,name,memory.total [MiB],memory.used [MiB],utilization.gpu [%]"
+    while true; do
+      local now
+      now="$(date '+%Y-%m-%d %H:%M:%S')"
+      nvidia-smi --query-gpu=index,name,memory.total,memory.used,utilization.gpu --format=csv,noheader,nounits \
+        | awk -F',' -v ts="$now" '{gsub(/^[ \t]+|[ \t]+$/, "", $0); print ts "," $0}'
+      sleep "$GPU_MONITOR_INTERVAL"
+    done
+  ) >"$output_file" 2>/dev/null &
+  echo $!
+}
+
+stop_gpu_monitor() {
+  local monitor_pid="${1:-}"
+  if [[ -n "$monitor_pid" ]] && kill -0 "$monitor_pid" 2>/dev/null; then
+    kill "$monitor_pid" 2>/dev/null || true
+    wait "$monitor_pid" 2>/dev/null || true
+  fi
+}
+
+copy_eval_outputs() {
+  local source_dir="$1"
+  local target_dir="$2"
+  local source_real=""
+  local target_real=""
+  source_real="$(realpath "$source_dir" 2>/dev/null || echo "$source_dir")"
+  target_real="$(realpath "$target_dir" 2>/dev/null || echo "$target_dir")"
+  if [[ "$source_real" == "$target_real" ]]; then
+    return 0
+  fi
+  mkdir -p "$target_dir"
+  if [[ -d "$source_dir" ]]; then
+    cp -r "$source_dir"/. "$target_dir"/
+  fi
+}
+
+run_with_artifacts() {
+  local name="$1"
+  shift
+
+  local log_file="$RESULT_ROOT/${name}.log"
+  local time_file="$RESULT_ROOT/${name}_time.txt"
+  local gpu_file="$RESULT_ROOT/${name}_gpu.csv"
+  local status_file="$RESULT_ROOT/${name}_status.txt"
+  local monitor_pid=""
+
+  monitor_pid="$(start_gpu_monitor "$gpu_file" || true)"
+  set +e
+  local exit_code=0
+  local start_ts=0
+  local end_ts=0
+  start_ts="$(date +%s)"
+  if [[ -x /usr/bin/time ]]; then
+    /usr/bin/time -v -o "$time_file" "$@" >"$log_file" 2>&1
+    exit_code=$?
+    if [[ ! -s "$time_file" ]]; then
+      end_ts="$(date +%s)"
+      {
+        echo "timing_mode=usr_bin_time_empty_fallback"
+        echo "start_ts=$start_ts"
+        echo "end_ts=$end_ts"
+        echo "elapsed_seconds=$((end_ts - start_ts))"
+      } >"$time_file"
+    fi
+  else
+    "$@" >"$log_file" 2>&1
+    exit_code=$?
+    end_ts="$(date +%s)"
+    {
+      echo "timing_mode=shell_date"
+      echo "start_ts=$start_ts"
+      echo "end_ts=$end_ts"
+      echo "elapsed_seconds=$((end_ts - start_ts))"
+    } >"$time_file"
+  fi
+  set -e
+  stop_gpu_monitor "$monitor_pid"
+
+  echo "exit_code=$exit_code" >"$status_file"
+  if [[ $exit_code -ne 0 ]]; then
+    echo "[error] command failed for $name, see $log_file"
+    return $exit_code
+  fi
+}
+
+write_manifest() {
+  cat >"$RESULT_ROOT/README.txt" <<EOF
+InternVLA-A1 collected results
+
+Key files:
+- env_summary.txt: environment and path summary
+- sample_run.log / sample_run_time.txt: one-sample functional run
+- forward_benchmark.log / forward_benchmark_time.txt: pure pipeline.forward latency benchmark
+- eval_run.log / eval_run_time.txt: GT evaluation run
+- pytest_e2e.log / pytest_e2e_time.txt: offline e2e pytest result
+- gpu_info_before.csv / gpu_info_after.csv: point-in-time GPU snapshots
+- *_gpu.csv: sampled GPU usage during each run
+- eval_outputs/: copied output directory from the GT evaluation run
+
+Important outputs:
+- forward_benchmark/forward_latency.json
+- eval_outputs/summary.json
+- eval_outputs/registry/log.json
+- eval_outputs/registry/plots/
+EOF
+}
+
+write_skip_artifact() {
+  local name="$1"
+  local reason="$2"
+  echo "$reason" >"$RESULT_ROOT/${name}.log"
+  echo "timing_mode=skipped" >"$RESULT_ROOT/${name}_time.txt"
+  echo "exit_code=0" >"$RESULT_ROOT/${name}_status.txt"
+  echo "skipped_reason=$reason" >>"$RESULT_ROOT/${name}_status.txt"
+  if [[ ! -f "$RESULT_ROOT/${name}_gpu.csv" ]]; then
+    echo "skipped,$reason" >"$RESULT_ROOT/${name}_gpu.csv"
+  fi
+}
+
+write_env_summary
+write_manifest
+capture_gpu_snapshot "$RESULT_ROOT/gpu_info_before.csv"
+
+run_with_artifacts \
+  "sample_run" \
+  bash "$ROOT_DIR/run.sh" \
+  --output-dir "$RESULT_ROOT/sample_outputs" \
+  --num-samples 1 \
+  --num-episodes 0
+
+run_with_artifacts \
+  "forward_benchmark" \
+  python "$ROOT_DIR/end2end.py" \
+  --model-dir "$INTERNVLA_A1_MODEL_DIR" \
+  --dataset-dir "$INTERNVLA_A1_DATASET_DIR" \
+  --benchmark-forward \
+  --dtype bfloat16 \
+  --attn-implementation eager \
+  --warmup-iters 3 \
+  --benchmark-iters 10 \
+  --output-dir "$RESULT_ROOT/forward_benchmark"
+
+run_with_artifacts \
+  "eval_run" \
+  bash "$ROOT_DIR/run.sh" \
+  --output-dir "$EVAL_OUTPUT_DIR" \
+  --num-episodes 1
+
+copy_eval_outputs "$EVAL_OUTPUT_DIR" "$RESULT_ROOT/eval_outputs"
+
+if python - <<'PY' >/dev/null 2>&1
+import importlib.util
+raise SystemExit(0 if importlib.util.find_spec("pytest") else 1)
+PY
+then
+  run_with_artifacts \
+    "pytest_e2e" \
+    python -m pytest -sv tests/e2e/offline_inference/test_internvla_a1.py -m advanced_model
+else
+  write_skip_artifact "pytest_e2e" "pytest is not installed in the current python environment"
+fi
+
+capture_gpu_snapshot "$RESULT_ROOT/gpu_info_after.csv"
+
+echo "Results written to: $RESULT_ROOT"

examples/offline_inference/internvla_a1/end2end.py

@@ -0,0 +1,327 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import statistics
+import sys
+import time
+from pathlib import Path
+
+import torch
+
+_WORKSPACE_ROOT = Path(__file__).resolve().parent
+_REPO_ROOT = _WORKSPACE_ROOT.parents[2]
+if str(_REPO_ROOT) not in sys.path:
+    # Keep the shared helper import clean by doing script-only path bootstrap here.
+    sys.path.insert(0, str(_REPO_ROOT))
+
+from internvla_a1_common import (  # noqa: E402
+    A2DOpenLoopDataset,
+    collate_open_loop_samples,
+    make_shared_noise,
+    run_open_loop_evaluation,
+    select_indices,
+    tensor_dtype,
+    tensor_sha256,
+)
+
+from vllm_omni.diffusion.data import OmniDiffusionConfig  # noqa: E402
+from vllm_omni.diffusion.models.internvla_a1 import (  # noqa: E402
+    InternVLAA1Config,
+    InternVLAA1TrainMetadata,
+)
+from vllm_omni.diffusion.models.internvla_a1.config import OBS_STATE  # noqa: E402
+from vllm_omni.diffusion.registry import initialize_model  # noqa: E402
+from vllm_omni.diffusion.request import OmniDiffusionRequest  # noqa: E402
+from vllm_omni.inputs.data import OmniDiffusionSamplingParams  # noqa: E402
+
+
+def _required_path_arg(env_name: str, cli_value: str | None) -> str:
+    value = cli_value or os.getenv(env_name)
+    if not value:
+        raise ValueError(f"Missing required path: set --{env_name.lower().replace('_', '-')} or {env_name}.")
+    return value
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Run vLLM InternVLA-A1 inference on a few samples.")
+    parser.add_argument("--model-dir")
+    parser.add_argument("--dataset-dir")
+    parser.add_argument("--num-samples", type=int, default=1)
+    parser.add_argument("--num-episodes", type=int, default=0)
+    parser.add_argument("--seed", type=int, default=1234)
+    parser.add_argument("--dtype", choices=["bfloat16", "float32"], default="bfloat16")
+    parser.add_argument("--device", default="cuda")
+    parser.add_argument("--compile-model", action="store_true")
+    parser.add_argument("--attn-implementation", default="eager")
+    parser.add_argument("--enable-regional-compile", action="store_true")
+    parser.add_argument("--enable-warmup", action="store_true")
+    parser.add_argument("--strict-load", action="store_true")
+    parser.add_argument("--output-dir", default="outputs/internvla_a1/vllm_infer")
+    parser.add_argument("--skip-plots", action="store_true")
+    parser.add_argument("--benchmark-forward", action="store_true")
+    parser.add_argument("--warmup-iters", type=int, default=3)
+    parser.add_argument("--benchmark-iters", type=int, default=10)
+    return parser.parse_args()
+
+
+def build_od_config(args: argparse.Namespace, processor_model_name: str) -> OmniDiffusionConfig:
+    return OmniDiffusionConfig(
+        model=str(Path(args.model_dir).resolve()),
+        model_class_name="InternVLAA1Pipeline",
+        dtype=tensor_dtype(args.dtype),
+        custom_pipeline_args={
+            "device": args.device,
+            "dtype": args.dtype,
+            "compile_model": args.compile_model,
+            "attn_implementation": args.attn_implementation,
+            "enable_regional_compile": args.enable_regional_compile,
+            "enable_warmup": args.enable_warmup,
+            "strict_load": args.strict_load,
+            "processor_model_name": processor_model_name,
+        },
+    )
+
+
+def build_dataset(args: argparse.Namespace) -> tuple[A2DOpenLoopDataset, InternVLAA1Config, InternVLAA1TrainMetadata]:
+    model_dir = Path(args.model_dir)
+    config = InternVLAA1Config.from_pretrained(model_dir)
+    config.device = args.device
+    config.dtype = args.dtype
+    config.compile_model = args.compile_model
+    config.attn_implementation = args.attn_implementation
+    config.enable_regional_compile = args.enable_regional_compile
+
+    train_meta = InternVLAA1TrainMetadata.from_pretrained(model_dir)
+    with open(model_dir / "stats.json", encoding="utf-8") as f:
+        train_stats = json.load(f)["a2d"]
+
+    dataset = A2DOpenLoopDataset(
+        args.dataset_dir,
+        config=config,
+        train_stats=train_stats,
+    )
+    return dataset, config, train_meta
+
+
+def run_one_path(
+    pipeline,
+    dataset: A2DOpenLoopDataset,
+    config: InternVLAA1Config,
+    args: argparse.Namespace,
+    indices: list[int],
+) -> list[dict[str, object]]:
+    results: list[dict[str, object]] = []
+    for index in indices:
+        sample = dataset.get_sample(index)
+        batch_inputs, _ = collate_open_loop_samples([sample], device=args.device, dtype=tensor_dtype(args.dtype))
+        noise = make_shared_noise(
+            args.seed,
+            index,
+            (
+                batch_inputs[OBS_STATE].shape[0],
+                config.chunk_size,
+                config.max_action_dim,
+            ),
+            args.device,
+        )
+        pred = run_pipeline_forward(pipeline, batch_inputs, noise)
+        pred = pred[:, :, : dataset.physical_action_dim].to(torch.float32).cpu()
+        results.append(
+            {
+                "path": "registry",
+                "index": index,
+                "episode_index": sample.episode_index,
+                "task": sample.task,
+                "seed": args.seed,
+                "shape": list(pred.shape),
+                "mean": float(pred.mean().item()),
+                "std": float(pred.std().item()),
+                "action_sha256": tensor_sha256(pred),
+                "first_action_prefix": pred[0, 0, :8].tolist(),
+            }
+        )
+    return results
+
+
+def run_pipeline_forward(
+    pipeline,
+    batch_inputs: dict[str, torch.Tensor],
+    noise: torch.Tensor,
+) -> torch.Tensor:
+    output = pipeline.forward(
+        OmniDiffusionRequest(
+            prompts=[""],
+            sampling_params=OmniDiffusionSamplingParams(
+                extra_args={
+                    "batch_inputs": batch_inputs,
+                    "noise": noise,
+                    "decode_image": False,
+                }
+            ),
+        )
+    )
+    if output.error:
+        raise RuntimeError(output.error)
+    if output.output is None:
+        raise RuntimeError("InternVLAA1Pipeline.forward returned no output tensor.")
+    return output.output
+
+
+def _synchronize(device: str) -> None:
+    if device.startswith("cuda") and torch.cuda.is_available():
+        torch.cuda.synchronize()
+
+
+def _latency_summary(values_ms: list[float]) -> dict[str, float]:
+    sorted_values = sorted(values_ms)
+    p50_index = min(len(sorted_values) - 1, round(0.50 * (len(sorted_values) - 1)))
+    p90_index = min(len(sorted_values) - 1, round(0.90 * (len(sorted_values) - 1)))
+    return {
+        "mean_ms": float(statistics.mean(sorted_values)),
+        "stdev_ms": float(statistics.pstdev(sorted_values)) if len(sorted_values) > 1 else 0.0,
+        "min_ms": float(sorted_values[0]),
+        "max_ms": float(sorted_values[-1]),
+        "p50_ms": float(sorted_values[p50_index]),
+        "p90_ms": float(sorted_values[p90_index]),
+    }
+
+
+def benchmark_forward(
+    pipeline,
+    dataset: A2DOpenLoopDataset,
+    config: InternVLAA1Config,
+    args: argparse.Namespace,
+    index: int,
+    output_dir: Path,
+) -> dict[str, object]:
+    sample = dataset.get_sample(index)
+    batch_inputs, _ = collate_open_loop_samples([sample], device=args.device, dtype=tensor_dtype(args.dtype))
+    noise = make_shared_noise(
+        args.seed,
+        index,
+        (
+            batch_inputs[OBS_STATE].shape[0],
+            config.chunk_size,
+            config.max_action_dim,
+        ),
+        args.device,
+    )
+
+    _synchronize(args.device)
+    cold_start_begin = time.perf_counter()
+    pred = run_pipeline_forward(pipeline, batch_inputs, noise)
+    _synchronize(args.device)
+    cold_start_ms = (time.perf_counter() - cold_start_begin) * 1000.0
+
+    warmup_ms: list[float] = []
+    for _ in range(args.warmup_iters):
+        _synchronize(args.device)
+        begin = time.perf_counter()
+        _ = run_pipeline_forward(pipeline, batch_inputs, noise)
+        _synchronize(args.device)
+        warmup_ms.append((time.perf_counter() - begin) * 1000.0)
+
+    benchmark_ms: list[float] = []
+    for _ in range(args.benchmark_iters):
+        _synchronize(args.device)
+        begin = time.perf_counter()
+        _ = run_pipeline_forward(pipeline, batch_inputs, noise)
+        _synchronize(args.device)
+        benchmark_ms.append((time.perf_counter() - begin) * 1000.0)
+
+    pred = pred[:, :, : dataset.physical_action_dim].to(torch.float32).cpu()
+    summary = {
+        "mode": "forward_latency",
+        "model_dir": str(Path(args.model_dir).resolve()),
+        "dataset_dir": str(Path(args.dataset_dir).resolve()),
+        "sample_index": index,
+        "episode_index": sample.episode_index,
+        "task": sample.task,
+        "device": args.device,
+        "dtype": args.dtype,
+        "attn_implementation": args.attn_implementation,
+        "enable_regional_compile": args.enable_regional_compile,
+        "warmup_iters": args.warmup_iters,
+        "benchmark_iters": args.benchmark_iters,
+        "output_shape": list(pred.shape),
+        "cold_start_ms": cold_start_ms,
+        "warmup_summary": _latency_summary(warmup_ms) if warmup_ms else {},
+        "benchmark_summary": _latency_summary(benchmark_ms) if benchmark_ms else {},
+        "benchmark_samples_ms": benchmark_ms,
+    }
+    with open(output_dir / "forward_latency.json", "w", encoding="utf-8") as f:
+        json.dump(summary, f, ensure_ascii=False, indent=2)
+    print(json.dumps(summary, ensure_ascii=False, indent=2))
+    return summary
+
+
+def main() -> None:
+    args = parse_args()
+    args.model_dir = _required_path_arg("INTERNVLA_A1_MODEL_DIR", args.model_dir)
+    args.dataset_dir = _required_path_arg("INTERNVLA_A1_DATASET_DIR", args.dataset_dir)
+    output_dir = Path(args.output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    dataset, config, train_meta = build_dataset(args)
+    indices = select_indices(dataset, args.num_samples)
+    od_config = build_od_config(args, train_meta.processor_model_name)
+
+    eval_summaries: dict[str, object] = {}
+    pipeline = initialize_model(od_config)
+    if args.benchmark_forward:
+        benchmark_forward(
+            pipeline=pipeline,
+            dataset=dataset,
+            config=config,
+            args=args,
+            index=indices[0],
+            output_dir=output_dir,
+        )
+        return
+    results = run_one_path(
+        pipeline=pipeline,
+        dataset=dataset,
+        config=config,
+        args=args,
+        indices=indices,
+    )
+    if args.num_episodes > 0:
+        eval_summaries["registry"] = run_open_loop_evaluation(
+            mode="vllm_registry",
+            policy=pipeline,
+            config=config,
+            dataset=dataset,
+            train_meta=train_meta,
+            collate_samples=collate_open_loop_samples,
+            run_sample_actions=lambda policy, batch_inputs, noise: run_pipeline_forward(policy, batch_inputs, noise),
+            num_episodes=args.num_episodes,
+            seed=args.seed,
+            device=args.device,
+            dtype=tensor_dtype(args.dtype),
+            output_dir=output_dir / "registry",
+            skip_plots=args.skip_plots,
+        )
+
+    summary = {
+        "mode": "registry",
+        "model_dir": str(Path(args.model_dir).resolve()),
+        "dataset_dir": str(Path(args.dataset_dir).resolve()),
+        "device": args.device,
+        "dtype": args.dtype,
+        "attn_implementation": args.attn_implementation,
+        "enable_regional_compile": args.enable_regional_compile,
+        "seed": args.seed,
+        "indices": indices,
+        "results": results,
+        "output_dir": str(output_dir.resolve()),
+        "eval_summaries": eval_summaries,
+    }
+    with open(output_dir / "summary.json", "w", encoding="utf-8") as f:
+        json.dump(summary, f, ensure_ascii=False, indent=2)
+    print(json.dumps(summary, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()

examples/offline_inference/internvla_a1/internvla_a1_common.py

@@ -0,0 +1,471 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+import pyarrow.parquet as pq
+import torch
+import torch.nn.functional as F
+import torchvision
+
+from vllm_omni.diffusion.models.internvla_a1.config import (  # noqa: E402
+    OBS_IMAGES,
+    OBS_STATE,
+    OBS_TASK,
+    InternVLAA1Config,
+)
+
+
+def tensor_dtype(name: str) -> torch.dtype:
+    return torch.bfloat16 if name == "bfloat16" else torch.float32
+
+
+def select_indices(dataset, num_samples: int) -> list[int]:
+    indices: list[int] = []
+    for _, episode_indices in dataset.episode_start_indices(max_episodes=num_samples):
+        if episode_indices:
+            indices.append(episode_indices[0])
+    return indices[:num_samples]
+
+
+def make_shared_noise(seed: int, sample_index: int, shape: tuple[int, ...], device: str) -> torch.Tensor:
+    generator = torch.Generator(device="cpu")
+    generator.manual_seed(seed + sample_index)
+    noise = torch.randn(shape, generator=generator, dtype=torch.float32)
+    return noise.to(device=device)
+
+
+def tensor_sha256(tensor: torch.Tensor) -> str:
+    array = tensor.detach().contiguous().cpu().numpy()
+    return hashlib.sha256(array.tobytes()).hexdigest()
+
+
+def _clamp_index(index: int, start: int, end: int) -> int:
+    return max(start, min(end - 1, index))
+
+
+def _load_parquet_rows(path: Path) -> list[dict[str, Any]]:
+    return pq.read_table(path).to_pylist()
+
+
+def _load_json(path: Path) -> dict[str, Any]:
+    with open(path, encoding="utf-8") as f:
+        return json.load(f)
+
+
+def _stack_stats(stats: dict[str, Any], keys: list[str]) -> dict[str, torch.Tensor]:
+    result = {}
+    for stat_name in ("mean", "std"):
+        values = []
+        for key in keys:
+            values.extend(stats[key][stat_name])
+        result[stat_name] = torch.tensor(values, dtype=torch.float32)
+    return result
+
+
+def normalize_vector(values: torch.Tensor, stats: dict[str, torch.Tensor]) -> torch.Tensor:
+    denom = torch.where(stats["std"] == 0, torch.ones_like(stats["std"]), stats["std"])
+    return (values - stats["mean"]) / denom
+
+
+def unnormalize_vector(values: torch.Tensor, stats: dict[str, torch.Tensor]) -> torch.Tensor:
+    return values * stats["std"] + stats["mean"]
+
+
+class TorchvisionVideoReaderCache:
+    def __init__(self, backend: str = "pyav") -> None:
+        self.backend = backend
+        self._readers: dict[str, Any] = {}
+        torchvision.set_video_backend(backend)
+
+    def get(self, path: str) -> Any:
+        reader = self._readers.get(path)
+        if reader is None:
+            reader = torchvision.io.VideoReader(path, "video")
+            self._readers[path] = reader
+        return reader
+
+    def decode_frames(self, path: str, timestamps: list[float], tolerance_s: float = 1e-4) -> torch.Tensor:
+        reader = self.get(path)
+        first_ts = min(timestamps)
+        last_ts = max(timestamps)
+        reader.seek(first_ts, keyframes_only=self.backend == "pyav")
+
+        loaded_frames: list[torch.Tensor] = []
+        loaded_ts: list[float] = []
+        for frame in reader:
+            current_ts = float(frame["pts"])
+            loaded_frames.append(frame["data"])
+            loaded_ts.append(current_ts)
+            if current_ts >= last_ts:
+                break
+
+        query_ts = torch.tensor(timestamps, dtype=torch.float32)
+        loaded_ts_tensor = torch.tensor(loaded_ts, dtype=torch.float32)
+        distances = torch.cdist(query_ts[:, None], loaded_ts_tensor[:, None], p=1)
+        min_dist, argmin = distances.min(dim=1)
+        if not torch.all(min_dist < tolerance_s):
+            raise RuntimeError(
+                f"Video timestamps are outside tolerance: query={query_ts.tolist()} "
+                f"loaded={loaded_ts_tensor.tolist()} path={path}"
+            )
+        return torch.stack([loaded_frames[i] for i in argmin]).float() / 255.0
+
+
+@dataclass
+class A2DOpenLoopSample:
+    index: int
+    episode_index: int
+    task: str
+    state_raw: torch.Tensor
+    action_raw: torch.Tensor
+    inputs: dict[str, torch.Tensor]
+
+
+class A2DOpenLoopDataset:
+    image_keys = [
+        "observation.images.head",
+        "observation.images.hand_left",
+        "observation.images.hand_right",
+    ]
+    state_keys = [
+        "observation.states.joint.position",
+        "observation.states.effector.position",
+    ]
+    action_keys = [
+        "actions.joint.position",
+        "actions.effector.position",
+    ]
+
+    def __init__(
+        self,
+        dataset_root: str | Path,
+        *,
+        config: InternVLAA1Config,
+        train_stats: dict[str, Any],
+        image_offsets: tuple[int, int] = (-15, 0),
+        tolerance_s: float = 1e-4,
+    ) -> None:
+        self.root = Path(dataset_root)
+        self.config = config
+        self.info = _load_json(self.root / "meta" / "info.json")
+        self.dataset_stats = _load_json(self.root / "meta" / "stats.json")
+        self.data_rows = _load_parquet_rows(self.root / "data" / "chunk-000" / "file-000.parquet")
+        self.episode_rows = _load_parquet_rows(self.root / "meta" / "episodes" / "chunk-000" / "file-000.parquet")
+        self.task_rows = _load_parquet_rows(self.root / "meta" / "tasks.parquet")
+
+        self.state_stats = _stack_stats(train_stats, self.state_keys)
+        self.action_stats = _stack_stats(train_stats, self.action_keys)
+        self.image_offsets = image_offsets
+        self.tolerance_s = tolerance_s
+        self.video_reader = TorchvisionVideoReaderCache(backend="pyav")
+
+    @property
+    def num_episodes(self) -> int:
+        return len(self.episode_rows)
+
+    @property
+    def physical_action_dim(self) -> int:
+        return 16
+
+    def episode_start_indices(self, max_episodes: int | None = None) -> list[tuple[int, list[int]]]:
+        rows = self.episode_rows if max_episodes is None else self.episode_rows[:max_episodes]
+        result = []
+        for ep in rows:
+            start = int(ep["dataset_from_index"])
+            end = int(ep["dataset_to_index"])
+            result.append((int(ep["episode_index"]), list(range(start, end, self.config.chunk_size))))
+        return result
+
+    def _task_text(self, task_index: int) -> str:
+        return self.task_rows[task_index]["__index_level_0__"]
+
+    def _episode_for_index(self, row: dict[str, Any]) -> dict[str, Any]:
+        return self.episode_rows[int(row["episode_index"])]
+
+    def _state_vector(self, row: dict[str, Any]) -> torch.Tensor:
+        return torch.tensor(row[self.state_keys[0]] + row[self.state_keys[1]], dtype=torch.float32)
+
+    def _action_vector(self, row: dict[str, Any]) -> torch.Tensor:
+        return torch.tensor(row[self.action_keys[0]] + row[self.action_keys[1]], dtype=torch.float32)
+
+    def _query_rows(self, idx: int, deltas: list[int]) -> list[dict[str, Any]]:
+        row = self.data_rows[idx]
+        episode = self._episode_for_index(row)
+        start = int(episode["dataset_from_index"])
+        end = int(episode["dataset_to_index"])
+        return [self.data_rows[_clamp_index(idx + delta, start, end)] for delta in deltas]
+
+    def _decode_camera_history(
+        self, episode: dict[str, Any], camera_key: str, rows: list[dict[str, Any]]
+    ) -> torch.Tensor:
+        timestamps = [float(r["timestamp"]) for r in rows]
+        shifted = [float(episode[f"videos/{camera_key}/from_timestamp"]) + ts for ts in timestamps]
+        chunk_idx = int(episode[f"videos/{camera_key}/chunk_index"])
+        file_idx = int(episode[f"videos/{camera_key}/file_index"])
+        path = self.root / self.info["video_path"].format(
+            video_key=camera_key,
+            chunk_index=chunk_idx,
+            file_index=file_idx,
+        )
+        frames = self.video_reader.decode_frames(str(path), shifted, tolerance_s=self.tolerance_s)
+        return frames
+
+    def get_sample(self, idx: int) -> A2DOpenLoopSample:
+        row = self.data_rows[idx]
+        episode = self._episode_for_index(row)
+        image_rows = self._query_rows(idx, list((-15, 0)))
+        action_rows = self._query_rows(idx, list(range(self.config.chunk_size)))
+        camera_images = [self._decode_camera_history(episode, camera_key, image_rows) for camera_key in self.image_keys]
+        state_raw = self._state_vector(row)
+        state_norm = normalize_vector(state_raw, self.state_stats)
+        action_raw = torch.stack([self._action_vector(action_row) for action_row in action_rows], dim=0)
+        task = self._task_text(int(row["task_index"]))
+        inputs = {
+            OBS_STATE: state_norm,
+            OBS_TASK: task,
+            f"{OBS_IMAGES}.image0": camera_images[0],
+            f"{OBS_IMAGES}.image1": camera_images[1],
+            f"{OBS_IMAGES}.image2": camera_images[2],
+            f"{OBS_IMAGES}.image0_mask": torch.tensor(True),
+            f"{OBS_IMAGES}.image1_mask": torch.tensor(True),
+            f"{OBS_IMAGES}.image2_mask": torch.tensor(True),
+        }
+        return A2DOpenLoopSample(
+            index=idx,
+            episode_index=int(row["episode_index"]),
+            task=task,
+            state_raw=state_raw,
+            action_raw=action_raw,
+            inputs=inputs,
+        )
+
+
+def collate_open_loop_samples(
+    samples: list[A2DOpenLoopSample],
+    *,
+    device: str,
+    dtype: torch.dtype,
+) -> tuple[dict[str, torch.Tensor], dict[str, Any]]:
+    first = samples[0]
+    batch_inputs: dict[str, torch.Tensor] = {}
+    for key in first.inputs:
+        values = [sample.inputs[key] for sample in samples]
+        if isinstance(values[0], torch.Tensor):
+            tensor = torch.stack(values, dim=0)
+            if tensor.dtype in (torch.int64, torch.bool):
+                batch_inputs[key] = tensor.to(device=device)
+            else:
+                batch_inputs[key] = tensor.to(device=device, dtype=dtype)
+        else:
+            batch_inputs[key] = values
+
+    metadata = {
+        "indices": [sample.index for sample in samples],
+        "episode_indices": [sample.episode_index for sample in samples],
+        "tasks": [sample.task for sample in samples],
+        "state_raw": torch.stack([sample.state_raw for sample in samples], dim=0),
+        "action_raw": torch.stack([sample.action_raw for sample in samples], dim=0),
+    }
+    return batch_inputs, metadata
+
+
+def plot_prediction_series(
+    *,
+    series: dict[str, np.ndarray],
+    output_path: Path,
+    title: str,
+) -> None:
+    import matplotlib.pyplot as plt
+
+    first_series = next(iter(series.values()))
+    num_dims = int(first_series.shape[1])
+    num_cols = 2
+    num_rows = math.ceil(num_dims / num_cols)
+    fig, axs = plt.subplots(num_rows, num_cols, figsize=(16, max(6, 3 * num_rows)))
+    axs = np.array(axs).reshape(-1)
+    x_values = np.arange(first_series.shape[0])
+    styles = [
+        ("Ground Truth", "blue", "-", 1.5),
+        ("VLLM", "green", "-.", 1.4),
+        ("Predicted", "red", "--", 1.4),
+    ]
+    style_map = {name: (color, linestyle, linewidth) for name, color, linestyle, linewidth in styles}
+
+    for dim in range(num_dims):
+        ax = axs[dim]
+        for name, values in series.items():
+            color, linestyle, linewidth = style_map.get(name, (None, "-", 1.2))
+            ax.plot(
+                x_values,
+                values[:, dim],
+                label=name,
+                color=color,
+                linestyle=linestyle,
+                linewidth=linewidth,
+            )
+        ax.set_title(f"Dimension {dim + 1}")
+        ax.set_xlabel("Time Step / Sample Index")
+        ax.set_ylabel(f"Value Dim {dim + 1}")
+        ax.legend(loc="upper right")
+        ax.grid(True, linestyle="--", alpha=0.7)
+
+    for dim in range(num_dims, len(axs)):
+        axs[dim].axis("off")
+
+    fig.suptitle(title, fontsize=16)
+    plt.tight_layout(rect=(0, 0, 1, 0.97))
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    plt.savefig(output_path)
+    plt.close(fig)
+
+
+def summarize_prediction_metrics(
+    *,
+    gt_tensor: torch.Tensor,
+    pred_tensor: torch.Tensor,
+    joint_dims: int = 14,
+) -> dict[str, float]:
+    return {
+        "mse": float(F.mse_loss(gt_tensor, pred_tensor, reduction="mean").item()),
+        "mae": float(F.l1_loss(gt_tensor, pred_tensor, reduction="mean").item()),
+        "mse_joint": float(F.mse_loss(gt_tensor[:, :joint_dims], pred_tensor[:, :joint_dims], reduction="mean").item()),
+        "mae_joint": float(F.l1_loss(gt_tensor[:, :joint_dims], pred_tensor[:, :joint_dims], reduction="mean").item()),
+        "mse_gripper": float(
+            F.mse_loss(gt_tensor[:, joint_dims:], pred_tensor[:, joint_dims:], reduction="mean").item()
+        ),
+        "mae_gripper": float(
+            F.l1_loss(gt_tensor[:, joint_dims:], pred_tensor[:, joint_dims:], reduction="mean").item()
+        ),
+    }
+
+
+def summarize_metric_list(values: list[float]) -> float | None:
+    return float(np.mean(values)) if values else None
+
+
+def run_open_loop_evaluation(
+    *,
+    mode: str,
+    policy,
+    config,
+    dataset,
+    train_meta,
+    collate_samples: Callable[..., tuple[dict[str, torch.Tensor], dict[str, Any]]],
+    run_sample_actions: Callable[[Any, dict[str, torch.Tensor], torch.Tensor], torch.Tensor],
+    num_episodes: int,
+    seed: int,
+    device: str,
+    dtype: torch.dtype,
+    output_dir: str | Path,
+    skip_plots: bool,
+) -> dict[str, Any]:
+    output_dir = Path(output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    plots_dir = output_dir / "plots"
+    plots_dir.mkdir(parents=True, exist_ok=True)
+
+    mse_values: list[float] = []
+    mae_values: list[float] = []
+    mse_joint_values: list[float] = []
+    mae_joint_values: list[float] = []
+    mse_gripper_values: list[float] = []
+    mae_gripper_values: list[float] = []
+    per_episode: list[dict[str, Any]] = []
+
+    plotting_available = True
+    if not skip_plots:
+        try:
+            import matplotlib.pyplot as _plt  # noqa: F401
+        except ModuleNotFoundError:
+            plotting_available = False
+            print("[warning] matplotlib is not installed; plots will be skipped.")
+
+    episode_specs = dataset.episode_start_indices(max_episodes=num_episodes)
+    for episode_index, indices in episode_specs:
+        print(f"[{mode}] episode: {episode_index}")
+        pred_chunks = []
+        gt_chunks = []
+        visited_indices = []
+        task = None
+
+        for index in indices:
+            sample = dataset.get_sample(index)
+            task = sample.task
+            batch_inputs, meta = collate_samples([sample], device=device, dtype=dtype)
+            noise = make_shared_noise(
+                seed,
+                index,
+                (batch_inputs[OBS_STATE].shape[0], config.chunk_size, config.max_action_dim),
+                device,
+            )
+            with torch.no_grad():
+                pred = run_sample_actions(policy, batch_inputs, noise)
+
+            pred = pred[:, :, : dataset.physical_action_dim].to(torch.float32).cpu()
+            pred_phys = unnormalize_vector(pred, dataset.action_stats)
+            if train_meta.action_mode == "delta":
+                pred_phys[:, :, :14] += meta["state_raw"][:, None, :14]
+
+            gt_phys = meta["action_raw"][:, :, : dataset.physical_action_dim].to(torch.float32).cpu()
+            pred_chunks.append(pred_phys[0])
+            gt_chunks.append(gt_phys[0])
+            visited_indices.append(index)
+
+        pred_tensor = torch.cat(pred_chunks, dim=0)
+        gt_tensor = torch.cat(gt_chunks, dim=0)
+        metrics = summarize_prediction_metrics(gt_tensor=gt_tensor, pred_tensor=pred_tensor)
+
+        mse_values.append(metrics["mse"])
+        mae_values.append(metrics["mae"])
+        mse_joint_values.append(metrics["mse_joint"])
+        mae_joint_values.append(metrics["mae_joint"])
+        mse_gripper_values.append(metrics["mse_gripper"])
+        mae_gripper_values.append(metrics["mae_gripper"])
+
+        if not skip_plots and plotting_available:
+            plot_prediction_series(
+                series={
+                    "Ground Truth": gt_tensor.numpy(),
+                    "VLLM": pred_tensor.numpy(),
+                },
+                output_path=plots_dir / f"{mode}_open_loop_ep{episode_index}.jpg",
+                title=f"{mode} Ground Truth vs VLLM",
+            )
+
+        episode_log = {
+            "episode_id": int(episode_index),
+            "task": task,
+            "visited_indices": visited_indices,
+            "num_pred_steps": int(pred_tensor.shape[0]),
+            **metrics,
+        }
+        per_episode.append(episode_log)
+        print(json.dumps({"mode": mode, **episode_log}, ensure_ascii=False))
+
+    summary = {
+        "mode": mode,
+        "num_episodes": len(per_episode),
+        "mse": mse_values,
+        "mae": mae_values,
+        "average_mse": summarize_metric_list(mse_values),
+        "average_mae": summarize_metric_list(mae_values),
+        "mse_joint": mse_joint_values,
+        "mae_joint": mae_joint_values,
+        "average_mse_joint": summarize_metric_list(mse_joint_values),
+        "average_mae_joint": summarize_metric_list(mae_joint_values),
+        "mse_gripper": mse_gripper_values,
+        "mae_gripper": mae_gripper_values,
+        "average_mse_gripper": summarize_metric_list(mse_gripper_values),
+        "average_mae_gripper": summarize_metric_list(mae_gripper_values),
+        "episodes": per_episode,
+    }
+    with open(output_dir / "log.json", "w", encoding="utf-8") as f:
+        json.dump(summary, f, indent=2, ensure_ascii=False)
+    return summary

examples/offline_inference/internvla_a1/run.sh

@@ -0,0 +1,23 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "$ROOT_DIR/../../.." && pwd)"
+ : "${INTERNVLA_A1_MODEL_DIR:?Please export INTERNVLA_A1_MODEL_DIR=/path/to/InternVLA-A1-3B-ft-pen}"
+ : "${INTERNVLA_A1_DATASET_DIR:?Please export INTERNVLA_A1_DATASET_DIR=/path/to/Genie1-Place_Markpen}"
+ : "${INTERNVLA_A1_PROCESSOR_DIR:?Please export INTERNVLA_A1_PROCESSOR_DIR=/path/to/Qwen3-VL-2B-Instruct}"
+ : "${INTERNVLA_A1_COSMOS_DIR:?Please export INTERNVLA_A1_COSMOS_DIR=/path/to/Cosmos-Tokenizer-CI8x8-SafeTensor}"
+INTERNVLA_A1_OUTPUT_DIR="${INTERNVLA_A1_OUTPUT_DIR:-$REPO_ROOT/outputs/internvla_a1/vllm_infer}"
+
+export INTERNVLA_A1_MODEL_DIR
+export INTERNVLA_A1_DATASET_DIR
+export INTERNVLA_A1_PROCESSOR_DIR
+export INTERNVLA_A1_COSMOS_DIR
+
+python "$ROOT_DIR/end2end.py" \
+  --model-dir "$INTERNVLA_A1_MODEL_DIR" \
+  --dataset-dir "$INTERNVLA_A1_DATASET_DIR" \
+  --output-dir "$INTERNVLA_A1_OUTPUT_DIR" \
+  --num-episodes "${INTERNVLA_A1_NUM_EPISODES:-1}" \
+  --attn-implementation eager \
+  "$@"