MiniCPM-RobotManip
A Smarter and Faster On-Device AI Brain for Robots
Image: github Github |
Image: Discord Discord |
MiniCPM-RobotManip is a 1.5B vision-language-action model for embodied manipulation with the following highlights:
Generalist Manipulation: A unified 1.5B generalist policy that uses one set of weights across all downstream tasks and outperforms larger models such as π₀.₅ and Qwen-VLA across representative evaluations.
Streaming Context: Historical observations are continuously incorporated into the model context through streaming inference, reducing per-step compute from 125 TFLOPs to 3.3 TFLOPs while retaining 60 frames of history and supporting up to one minute of visual memory. This moves VLA beyond reactive action generation from single-frame observations toward continuous decision-making grounded in long-horizon visual context.
Efficient Inference: Inherits MiniCPM-V 4.6's visual token compression, reducing each frame from 256 to 64 visual tokens for 4× compression. With H100, BF16, and single-frame input, model-forward latency per decision step is 120 ms, compared with 234 ms for π0.5. The measurement excludes task autoregressive decoding.
Image: MiniCPM-RobotManip task demonstrations
Benchmark Results#
Image: manip_benchmark
Inference Example#
Please Ensure transformers==5.7.0
from __future__ import annotations
import argparse import json from pathlib import Path from typing import Sequence
import numpy as np import torch from PIL import Image from transformers import AutoModel, AutoProcessor
STATE_DIM = 80 IMAGE_SIZE = (448, 448)
class MiniCPMVLAInference: """Processor and model wrapper for single-sample VLA inference."""
def __init__( self, checkpoint_path: str | Path = "openbmb/MiniCPM-RobotManip", device: str | torch.device | None = None, ): if device is None: device = "cuda" if torch.cuda.is_available() else "cpu" self.device = torch.device(device) checkpoint = str(checkpoint_path) self.processor = AutoProcessor.from_pretrained(checkpoint, trust_remote_code=True) self.model = AutoModel.from_pretrained(checkpoint, trust_remote_code=True) self.model.to(self.device).eval()
@staticmethod def _load_images(images: Sequence[str | Path | Image.Image | np.ndarray]) -> list[np.ndarray]: if not images: raise ValueError("At least one image is required") loaded = [] for image in images: if isinstance(image, (str, Path)): with Image.open(image) as pil_image: array = np.asarray(pil_image.convert("RGB")) elif isinstance(image, Image.Image): array = np.asarray(image.convert("RGB")) elif isinstance(image, np.ndarray): array = image else: raise TypeError(f"Unsupported image type: {type(image)!r}") if array.ndim != 3 or array.shape[-1] != 3: raise ValueError(f"Expected an HxWx3 image, got shape {array.shape}") # Match the training pipeline's ResizeImage(size=(448, 448)), # including PIL's default resize interpolation. resized = Image.fromarray(array).resize(IMAGE_SIZE) loaded.append(np.asarray(resized).copy()) return loaded
def preprocess(self, images: Sequence, text: str) -> dict[str, torch.Tensor]: """Apply the same MiniCPM-V chat template and processor as training.""" content = [ {"type": "image", "image": image} for image in self._load_images(images) ] content.append({"type": "text", "text": text}) messages = [{"role": "user", "content": content}] inputs = self.processor.apply_chat_template( messages, tokenize=True, add_generation_prompt=True, return_dict=True, return_tensors="pt", processor_kwargs={"padding": False}, ) return { key: value.to(self.device) for key, value in inputs.items() if isinstance(value, torch.Tensor) }
def _prepare_state(self, state: torch.Tensor | np.ndarray | Sequence[float]) -> torch.Tensor: state = torch.as_tensor(state, dtype=torch.float32, device=self.device) if state.ndim == 1: state = state.unsqueeze(0).unsqueeze(0) elif state.ndim == 2: state = state.unsqueeze(1) if state.shape != (1, 1, STATE_DIM): raise ValueError(f"state must have shape (80,), (1, 80), or (1, 1, 80); got {tuple(state.shape)}") return state
@torch.inference_mode() def predict( self, images: Sequence[str | Path | Image.Image | np.ndarray], text: str, state: torch.Tensor | np.ndarray | Sequence[float] | None = None, embodiment_id: int = 0, seed: int | None = None, ) -> torch.Tensor: """Return one action chunk with shape (30, 80) on CPU.""" if not 0 <= embodiment_id < self.model.action_head.max_num_embodiments: raise ValueError( f"embodiment_id must be in [0, {self.model.action_head.max_num_embodiments - 1}]" ) if state is None: state = torch.zeros(STATE_DIM) state_tensor = self._prepare_state(state) embodiment = torch.tensor([embodiment_id], dtype=torch.long, device=self.device) if seed is not None: torch.manual_seed(seed) if self.device.type == "cuda": torch.cuda.manual_seed_all(seed)
vlm_inputs = self.preprocess(images, text) actions = self.model.predict_action( state=state_tensor, embodiment_id=embodiment, **vlm_inputs, ) return actions[0].float().cpu()
def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--image", action="append", required=True, help="Input image; repeat for multiple views") parser.add_argument("--text", required=True, help="Robot instruction/prompt") parser.add_argument("--device", default=None, help="Default: cuda if available, otherwise cpu") state_group = parser.add_mutually_exclusive_group() state_group.add_argument("--state-file", help="A .npy file containing 80 state values") state_group.add_argument("--state", nargs=STATE_DIM, type=float, metavar="VALUE") parser.add_argument("--embodiment-id", type=int, default=0) parser.add_argument("--seed", type=int, default=None) parser.add_argument("--output", help="Optional output .npy path; otherwise print JSON") return parser.parse_args()
if __name__ == "__main__": args = parse_args() if args.state_file: state = np.load(args.state_file) elif args.state is not None: state = args.state else: state = np.zeros(STATE_DIM, dtype=np.float32)
infer_runner = MiniCPMVLAInference( checkpoint_path="openbmb/MiniCPM-RobotManip", device=args.device, ) action = infer_runner.predict( images=args.image, text=args.text, state=state, embodiment_id=args.embodiment_id, seed=args.seed, ) if args.output: output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) np.save(output_path, action.numpy()) print(f"Saved action {tuple(action.shape)} to {output_path}") else: print(json.dumps(action.tolist()))
Acknowledgement#
This project builds on and references starVLA and LeRobot. We thank the authors for their open-source contributions.
License#
Model weights and code are open-sourced under the Apache-2.0 license.
