[Model]Support MiniCPM-o 4.5 (#3642)

Signed-off-by: tc-mb <tianchi_cai@icloud.com>
Co-authored-by: GKangaroo <1095103651@qq.com>
Co-authored-by: GKangaroo <gqx24@mails.tsinghua.edu.cn>
Co-authored-by: Hongsheng Liu <liuhongsheng4@huawei.com>
This commit is contained in:
authored and GitHub committed 2026-06-01 22:12:24 +08:00
1 parent 29343540f7
commit 2b7249fbb3
22 files changed
+6840 -6

No files matched your search

@@ -0,0 +1,65 @@
# vLLM-Omni · MiniCPM-o 4.5 Online Demo
Gradio-based web UI for **MiniCPM-o 4.5** served via `vllm-omni`'s
OpenAI-compatible endpoints.
The UI supports:
- **Inputs**: text prompt + optional image, audio (file or mic), video.
- **Outputs**: text + speech (WAV player).
## 1. Start the backend server
The deploy config auto-loads via `--omni`; the default
`vllm_omni/deploy/minicpmo_4_5.yaml` targets a 2-GPU layout (thinker on
GPU 0, talker + t2w sharing GPU 1). For other hardware layouts pick
one of the deploy variants below.
| deploy config | GPUs | Notes |
|---|---|---|
| `minicpmo_4_5.yaml` (default) | 2 | Thinker on GPU0, talker+t2w on GPU1. |
| `minicpmo_4_5_3gpu.yaml` | 3 | Thinker 2-way TP on GPU0/1, talker+t2w share GPU2. |
| `minicpmo_4_5_8x4090.yaml` | 8 | Full 8x4090 layout. |
Default (2-GPU):
```bash
vllm serve openbmb/MiniCPM-o-4_5 --omni \
--trust-remote-code \
--host 0.0.0.0 --port 8099
```
Other layouts via `--deploy-config`:
```bash
vllm serve openbmb/MiniCPM-o-4_5 --omni \
--deploy-config vllm_omni/deploy/minicpmo_4_5_8x4090.yaml \
--trust-remote-code \
--host 0.0.0.0 --port 8099
```
## 2. Launch the Gradio demo
```bash
bash examples/online_serving/minicpmo/run_gradio_demo.sh
# Or run the python entry point directly:
python examples/online_serving/minicpmo/gradio_demo.py \
--minicpmo45-api-base http://localhost:8099/v1 \
--minicpmo45-model openbmb/MiniCPM-o-4_5 \
--port 7862
```
Open `http://<host>:7862` in a browser.
## Notes
- **TTS trigger**: the demo sets
`extra_body.chat_template_kwargs.use_tts_template=True`, which appends
`<|tts_bos|>` to the assistant prefix.
- Uncheck **"Generate speech output (TTS)"** to get text-only responses
(faster).
- The audio output is the raw WAV returned by the stage-1 talker +
Token2Wav; sample rate is 24 kHz.
- Video input is forwarded as a base64 `video_url` entry; the server needs
decord/torchvision to decode it.
@@ -0,0 +1,580 @@
"""Gradio demo for MiniCPM-o 4.5 online serving through vllm-omni.
Supports:
- Text / Image / Audio / Video inputs (OpenAI chat-completions multimodal format).
- Text + audio outputs, returned side-by-side in the UI.
Usage:
# run a 4.5 server on :8099
python gradio_demo.py \
--minicpmo45-api-base http://localhost:8099/v1 \
--minicpmo45-model openbmb/MiniCPM-o-4_5 \
--port 7862
"""
from __future__ import annotations
import argparse
import base64
import io
import os
from collections.abc import Iterable
from pathlib import Path
from typing import Any
import gradio as gr
import numpy as np
import soundfile as sf
from openai import OpenAI
from PIL import Image
MINICPMO45 = "MiniCPM-o-4.5"
# Default reference audio per model, relative to the model path. Used to build
# the audio_assistant system prompt so that the LLM adopts the "speak in this
# voice" persona (matches the official MiniCPM-o demo).
_DEFAULT_REF_AUDIO = {
MINICPMO45: "assets/HT_ref_audio.wav",
}
# audio_assistant prompt text, mirrors MiniCPMO.get_sys_prompt("audio_assistant")
# in the MiniCPM-o-Demo reference implementation.
_AUDIO_ASSISTANT_PROMPT = {
"zh": {
"prefix": "模仿音频样本的音色并生成新的内容。",
"suffix": (
"你的任务是用这种声音模式来当一个助手。请认真、高质量地回复用户的问题。"
"请用高自然度的方式和用户聊天。你是由面壁智能开发的人工智能助手:面壁小钢炮。"
),
},
"en": {
"prefix": "Use the voice in the audio prompt to synthesize new content.",
"suffix": "You are a helpful assistant with the above voice style.",
},
}
# ---------------------------------------------------------------------------
# Media helpers
# ---------------------------------------------------------------------------
def image_to_base64_data_url(image: Image.Image) -> str:
buf = io.BytesIO()
if image.mode != "RGB":
image = image.convert("RGB")
image.save(buf, format="JPEG", quality=92)
b64 = base64.b64encode(buf.getvalue()).decode("utf-8")
return f"data:image/jpeg;base64,{b64}"
def audio_to_base64_data_url(audio_np: np.ndarray, sample_rate: int) -> str:
if audio_np.dtype != np.int16:
if audio_np.dtype in (np.float32, np.float64):
audio_np = np.clip(audio_np, -1.0, 1.0)
audio_np = (audio_np * 32767).astype(np.int16)
else:
audio_np = audio_np.astype(np.int16)
buf = io.BytesIO()
sf.write(buf, audio_np, sample_rate, format="WAV")
b64 = base64.b64encode(buf.getvalue()).decode("utf-8")
return f"data:audio/wav;base64,{b64}"
def ref_audio_to_data_url(path: str, target_sr: int = 16000, max_s: float | None = 8.0) -> str:
"""Load a file, downmix to mono, resample to 16kHz, optionally truncate, and
return a data-URL usable as audio_url content in the chat API.
MiniCPM-o expects 16kHz mono audio input; long reference audios can waste
tokens so we cap at max_s seconds by default (3-10s is the recommended range).
"""
data, sr = sf.read(path)
if data.ndim > 1:
data = data[:, 0]
data = data.astype(np.float32)
if sr != target_sr:
try:
# Demo client only; librosa here resamples audio before sending to
# the server. The fallback below covers environments without it.
import librosa # noqa: TID251
data = librosa.resample(data, orig_sr=sr, target_sr=target_sr)
except ImportError:
from math import gcd
g = gcd(sr, target_sr)
up, down = target_sr // g, sr // g
idx = (np.arange(len(data) * up) / up).astype(np.int64)
data = data[idx][::down]
sr = target_sr
if max_s is not None and len(data) > int(max_s * sr):
data = data[: int(max_s * sr)]
return audio_to_base64_data_url(data, sr)
def video_to_base64_data_url(path: str) -> str:
p = Path(path)
if not p.exists():
raise FileNotFoundError(f"Video not found: {path}")
ext = p.suffix.lower()
mime = {
".mp4": "video/mp4",
".webm": "video/webm",
".mov": "video/quicktime",
".avi": "video/x-msvideo",
".mkv": "video/x-matroska",
}.get(ext, "video/mp4")
b64 = base64.b64encode(p.read_bytes()).decode("utf-8")
return f"data:{mime};base64,{b64}"
def process_audio_input(audio_input: Any | None) -> tuple[np.ndarray, int] | None:
"""Normalize Gradio audio input to (np.ndarray mono float32, sample_rate).
Gradio's `gr.Audio(type="numpy")` yields (sr, ndarray); we also accept
filesystem paths and a few nested tuple variants that appear in different
Gradio versions.
"""
if audio_input is None:
return None
def _from_path(p: str) -> tuple[np.ndarray, int] | None:
if not p:
return None
fp = Path(p)
if not fp.exists():
return None
data, sr = sf.read(fp)
if data.ndim > 1:
data = data[:, 0]
return data.astype(np.float32), int(sr)
audio_np: np.ndarray | None = None
sr: int | None = None
if isinstance(audio_input, tuple) and len(audio_input) == 2:
a, b = audio_input
if isinstance(a, int | float) and isinstance(b, np.ndarray):
sr, audio_np = int(a), b
elif isinstance(a, str):
loaded = _from_path(a)
if loaded is not None:
audio_np, sr = loaded
elif isinstance(audio_input, str):
loaded = _from_path(audio_input)
if loaded is not None:
audio_np, sr = loaded
if audio_np is None or sr is None:
return None
if audio_np.ndim > 1:
audio_np = audio_np[:, 0]
return audio_np.astype(np.float32), sr
# ---------------------------------------------------------------------------
# Per-model inference config
# ---------------------------------------------------------------------------
class ModelEndpoint:
"""Endpoint + TTS-trigger convention for one MiniCPM-o variant."""
def __init__(self, name: str, api_base: str, model_path: str):
self.name = name
self.api_base = api_base
self.model_path = model_path
self.client = OpenAI(api_key="EMPTY", base_url=api_base)
self._ref_audio_data_url: str | None = None
@property
def ref_audio_path(self) -> str | None:
rel = _DEFAULT_REF_AUDIO.get(self.name)
if not rel:
return None
full = os.path.join(self.model_path, rel)
return full if os.path.exists(full) else None
def get_ref_audio_data_url(self) -> str | None:
"""Lazily load and cache the default reference audio as a 16kHz mono
data URL. Returns None if the reference audio file is missing."""
if self._ref_audio_data_url is not None:
return self._ref_audio_data_url
p = self.ref_audio_path
if not p:
print(f"[{self.name}] no default reference audio found; falling back to text-only system prompt")
return None
try:
self._ref_audio_data_url = ref_audio_to_data_url(p)
print(f"[{self.name}] loaded reference audio: {p}")
except Exception as e:
print(f"[{self.name}] failed to load reference audio {p}: {e}")
return None
return self._ref_audio_data_url
def build_extras(self) -> dict[str, Any]:
"""Extra kwargs passed to chat.completions.create for this model.
4.5: use_tts_template=True chat-template flag appends <|tts_bos|>.
"""
if self.name == MINICPMO45:
return {
"extra_body": {
"chat_template_kwargs": {"use_tts_template": True},
"modalities": ["text", "audio"],
},
}
return {}
# ---------------------------------------------------------------------------
# System prompt (matches MiniCPMO.get_sys_prompt("audio_assistant") in the
# official MiniCPM-o-Demo). When TTS is disabled we fall back to a plain
# text-only system prompt.
# ---------------------------------------------------------------------------
def build_audio_assistant_system(
ref_audio_data_url: str | None,
language: str = "zh",
) -> dict[str, Any]:
"""Return an OpenAI-style system message carrying the reference audio."""
txt = _AUDIO_ASSISTANT_PROMPT.get(language, _AUDIO_ASSISTANT_PROMPT["zh"])
if ref_audio_data_url is None:
return {
"role": "system",
"content": (
"You are MiniCPM-o, a helpful multimodal assistant that can "
"understand images, audio and video, and respond in text and speech."
),
}
return {
"role": "system",
"content": [
{"type": "text", "text": txt["prefix"]},
{"type": "audio_url", "audio_url": {"url": ref_audio_data_url}},
{"type": "text", "text": txt["suffix"]},
],
}
# ---------------------------------------------------------------------------
# Core inference function
# ---------------------------------------------------------------------------
def run_inference(
endpoint: ModelEndpoint,
user_prompt: str,
audio_tuple: tuple[np.ndarray, int] | None,
image_pil: Image.Image | None,
video_path: str | None,
enable_tts: bool,
max_tokens: int,
temperature: float,
) -> Iterable[tuple[str, tuple[int, np.ndarray] | None]]:
user_prompt = user_prompt or ""
if not user_prompt.strip() and not audio_tuple and not image_pil and not video_path:
yield "Please provide a text prompt or at least one multimodal input.", None
return
try:
content: list[dict[str, Any]] = []
if audio_tuple is not None:
audio_np, sr = audio_tuple
content.append(
{"type": "audio_url", "audio_url": {"url": audio_to_base64_data_url(audio_np, sr)}},
)
if image_pil is not None:
img = image_pil if image_pil.mode == "RGB" else image_pil.convert("RGB")
content.append(
{"type": "image_url", "image_url": {"url": image_to_base64_data_url(img)}},
)
if video_path:
content.append(
{"type": "video_url", "video_url": {"url": video_to_base64_data_url(video_path)}},
)
if user_prompt.strip():
content.append({"type": "text", "text": user_prompt})
if enable_tts:
ref_url = endpoint.get_ref_audio_data_url()
system_msg = build_audio_assistant_system(ref_url, language="zh")
else:
system_msg = {
"role": "system",
"content": (
"You are MiniCPM-o, a helpful multimodal assistant that can "
"understand images, audio and video, and respond in text and speech."
),
}
messages = [
system_msg,
{"role": "user", "content": content},
]
kwargs: dict[str, Any] = {
"model": endpoint.model_path,
"messages": messages,
"max_tokens": int(max_tokens),
"temperature": float(temperature),
}
if enable_tts:
extras = endpoint.build_extras()
# Merge extra_body dicts cleanly.
extra_body = dict(extras.get("extra_body", {}))
kwargs["extra_body"] = extra_body
else:
# Text-only: suppress audio output.
kwargs["extra_body"] = {"modalities": ["text"]}
completion = endpoint.client.chat.completions.create(**kwargs)
text_parts: list[str] = []
audio_out: tuple[int, np.ndarray] | None = None
for choice in completion.choices:
msg = choice.message
if getattr(msg, "content", None):
text_parts.append(msg.content)
audio_obj = getattr(msg, "audio", None)
if audio_obj and getattr(audio_obj, "data", None):
raw = base64.b64decode(audio_obj.data)
if len(raw) > 80:
data, sr = sf.read(io.BytesIO(raw))
if data.ndim > 1:
data = data[:, 0]
audio_out = (int(sr), data.astype(np.float32))
text_response = "\n\n".join(t for t in text_parts if t) or "(no text returned)"
yield text_response, audio_out
except Exception as exc: # noqa: BLE001
yield f"Inference failed: {type(exc).__name__}: {exc}", None
# ---------------------------------------------------------------------------
# UI
# ---------------------------------------------------------------------------
DEFAULT_PROMPTS = [
"你好,请用一句话介绍北京。",
"Describe what you see in the image.",
"Transcribe the audio and translate it into English.",
"What is happening in this video?",
]
def build_interface(endpoints: dict[str, ModelEndpoint], default_model: str) -> gr.Blocks:
model_choices = list(endpoints.keys())
css = """
#generate-btn button { width: 100%; font-size: 1.05rem; }
.media-row > div { min-height: 260px; }
.model-banner { color: #555; font-size: 0.9rem; }
"""
with gr.Blocks(css=css, title="vLLM-Omni · MiniCPM-o Demo") as demo:
gr.Markdown(
"# vLLM-Omni · MiniCPM-o Online Demo\n"
"Multimodal chat (text / image / audio / video in; text + speech out) for "
"MiniCPM-o 4.5 served by vllm-omni."
)
with gr.Row():
model_dd = gr.Dropdown(
label="Model",
choices=model_choices,
value=default_model,
scale=2,
)
endpoint_info = gr.Markdown(
elem_classes="model-banner",
value=f"`{endpoints[default_model].api_base}` → `{endpoints[default_model].model_path}`",
)
with gr.Row():
prompt_box = gr.Textbox(
label="Text Prompt",
placeholder="Ask the model anything. Combine with image / audio / video below.",
lines=3,
value=DEFAULT_PROMPTS[0],
scale=3,
)
with gr.Column(scale=1):
tts_checkbox = gr.Checkbox(label="Generate speech output (TTS)", value=True)
max_tokens_slider = gr.Slider(label="Max tokens", minimum=32, maximum=4096, value=1024, step=32)
temperature_slider = gr.Slider(label="Temperature", minimum=0.0, maximum=1.5, value=0.7, step=0.05)
with gr.Row(elem_classes="media-row"):
image_input = gr.Image(label="Image (optional)", type="pil", sources=["upload"])
audio_input = gr.Audio(
label="Audio (optional)",
type="numpy",
sources=["upload", "microphone"],
)
video_input = gr.Video(label="Video (optional)", sources=["upload"])
with gr.Row():
generate_btn = gr.Button("Generate", variant="primary", elem_id="generate-btn")
clear_btn = gr.Button("Clear")
with gr.Row():
text_output = gr.Textbox(label="Text Output", lines=10, scale=2)
audio_output = gr.Audio(label="Audio Output", interactive=False, scale=1)
gr.Examples(
examples=[
["你好,请用一句话介绍北京。"],
["Describe what you see in detail."],
["请先识别这段语音的内容,再用中文回复说话人。"],
["What is happening in this video?"],
],
inputs=[prompt_box],
label="Quick prompts",
)
def _on_model_change(name: str) -> str:
ep = endpoints[name]
return f"`{ep.api_base}` → `{ep.model_path}`"
model_dd.change(_on_model_change, inputs=[model_dd], outputs=[endpoint_info])
def _run(
model_name: str,
prompt: str,
image_pil: Image.Image | None,
audio_in: Any,
video_path: str | None,
enable_tts: bool,
max_tokens: int,
temperature: float,
):
ep = endpoints[model_name]
audio_tuple = process_audio_input(audio_in)
yield from run_inference(
ep,
prompt,
audio_tuple,
image_pil,
video_path,
enable_tts,
max_tokens,
temperature,
)
generate_btn.click(
_run,
inputs=[
model_dd,
prompt_box,
image_input,
audio_input,
video_input,
tts_checkbox,
max_tokens_slider,
temperature_slider,
],
outputs=[text_output, audio_output],
)
def _clear():
return "", None, None, None, "", None
clear_btn.click(
_clear,
outputs=[
prompt_box,
image_input,
audio_input,
video_input,
text_output,
audio_output,
],
)
demo.queue()
return demo
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("--minicpmo45-api-base", default=os.environ.get("MINICPMO45_API_BASE", "http://localhost:8099/v1"))
p.add_argument(
"--minicpmo45-model",
default=os.environ.get("MINICPMO45_MODEL", "openbmb/MiniCPM-o-4_5"),
)
p.add_argument("--host", default="0.0.0.0")
p.add_argument("--port", type=int, default=7862)
p.add_argument("--share", action="store_true")
p.add_argument(
"--ssl-certfile",
default=os.environ.get("GRADIO_SSL_CERTFILE", ""),
help="Path to TLS certificate (PEM). Browsers require HTTPS to access the microphone.",
)
p.add_argument(
"--ssl-keyfile",
default=os.environ.get("GRADIO_SSL_KEYFILE", ""),
help="Path to TLS private key (PEM).",
)
p.add_argument(
"--ssl-verify",
action="store_true",
default=os.environ.get("GRADIO_SSL_VERIFY", "0") == "1",
help="Require browser to verify the certificate (off by default; self-signed certs should stay off).",
)
return p.parse_args()
def _ping(api_base: str, timeout: float = 3.0) -> bool:
import urllib.request
try:
with urllib.request.urlopen(api_base.rstrip("/").replace("/v1", "/health"), timeout=timeout):
return True
except Exception:
return False
def main() -> None:
args = parse_args()
endpoints: dict[str, ModelEndpoint] = {}
if args.minicpmo45_api_base and args.minicpmo45_model:
ok = _ping(args.minicpmo45_api_base)
status = "reachable" if ok else "not reachable (will error on generate)"
print(f"[4.5] {args.minicpmo45_api_base} ({status})")
endpoints[MINICPMO45] = ModelEndpoint(MINICPMO45, args.minicpmo45_api_base, args.minicpmo45_model)
if not endpoints:
raise SystemExit("No endpoints configured. Pass --minicpmo45-api-base/--minicpmo45-model.")
default_model = next(iter(endpoints.keys()))
demo = build_interface(endpoints, default_model)
launch_kwargs: dict = {
"server_name": args.host,
"server_port": args.port,
"share": args.share,
}
if args.ssl_certfile and args.ssl_keyfile:
launch_kwargs.update(
ssl_certfile=args.ssl_certfile,
ssl_keyfile=args.ssl_keyfile,
ssl_verify=args.ssl_verify,
)
print(f"[tls] cert={args.ssl_certfile} key={args.ssl_keyfile} verify={args.ssl_verify}")
demo.launch(**launch_kwargs)
if __name__ == "__main__":
main()
+36
View File
@@ -0,0 +1,36 @@
#!/bin/bash
# Launch the MiniCPM-o 4.5 gradio demo.
#
# Prereq:
# Start a vllm-omni OpenAI server for MiniCPM-o 4.5 on :8099 (see the
# 8x4090 stage config under vllm_omni/model_executor/stage_configs).
set -e
HERE="$(cd "$(dirname "$0")" && pwd)"
: "${MINICPMO45_API_BASE:=http://localhost:8099/v1}"
: "${MINICPMO45_MODEL:=openbmb/MiniCPM-o-4_5}"
: "${GRADIO_HOST:=0.0.0.0}"
: "${GRADIO_PORT:=7862}"
# HTTPS (browsers require a secure context for microphone access).
# Set GRADIO_SSL_CERTFILE / GRADIO_SSL_KEYFILE to enable TLS.
: "${GRADIO_SSL_CERTFILE:=}"
: "${GRADIO_SSL_KEYFILE:=}"
export MINICPMO45_API_BASE MINICPMO45_MODEL
SSL_ARGS=()
if [ -n "$GRADIO_SSL_CERTFILE" ] && [ -n "$GRADIO_SSL_KEYFILE" ] \
&& [ -f "$GRADIO_SSL_CERTFILE" ] && [ -f "$GRADIO_SSL_KEYFILE" ]; then
SSL_ARGS=(--ssl-certfile "$GRADIO_SSL_CERTFILE" --ssl-keyfile "$GRADIO_SSL_KEYFILE")
echo "HTTPS enabled: cert=$GRADIO_SSL_CERTFILE key=$GRADIO_SSL_KEYFILE"
else
echo "HTTPS disabled (cert/key not found). Microphone won't work on remote browsers."
fi
exec python "$HERE/gradio_demo.py" \
--minicpmo45-api-base "$MINICPMO45_API_BASE" \
--minicpmo45-model "$MINICPMO45_MODEL" \
--host "$GRADIO_HOST" \
--port "$GRADIO_PORT" \
"${SSL_ARGS[@]}"
+23
View File
@@ -76,6 +76,18 @@ local = [
"FlagEmbedding",
]
# Optional runtime dependencies for the MiniCPM-o 4.5 talker pipeline.
# The talker loads MiniCPMTTS via `trust_remote_code` from the model repo,
# and that module performs `from stepaudio2 import Token2wav` against the
# MiniCPM-o-flavored Token2wav vocoder (PyPI: `stepaudio2-minicpmo`, NOT
# upstream `stepfun-ai/Step-Audio2` — the upstream signature rejects
# `n_timesteps` and crashes at construction). A clean `vllm-omni` install
# does not pull this in by default to avoid forcing the dep onto users
# who only need other models; opt in with `pip install vllm-omni[minicpmo]`.
minicpmo = [
"stepaudio2-minicpmo",
]
docs = [
"mkdocs>=1.5.0",
"mkdocs-api-autonav",
@@ -101,6 +113,13 @@ Documentation = "https://vllm-omni.readthedocs.io"
[project.scripts]
vllm-omni = "vllm_omni.entrypoints.cli.main:main"
# Register vllm-omni's model registrations on every vllm subprocess. vllm
# auto-loads anything advertised under `vllm.general_plugins` in its own
# `load_general_plugins()`, so worker subprocesses spawned by vllm (which
# only import vllm, not vllm_omni) will still pick up the omni archs.
[project.entry-points."vllm.general_plugins"]
vllm_omni_register_models = "vllm_omni.engine.arg_utils:register_omni_models_to_vllm"
[tool.setuptools.packages.find]
where = ["."]
@@ -149,6 +168,10 @@ ignore = [
[tool.ruff.lint.per-file-ignores]
"examples/**" = ["E501"] # Allow long lines in examples
"tests/**" = ["E501"] # Allow long lines in tests
# MiniCPM-o reference modeling code vendored from upstream HF/transformers/MiniCPM-o.
# These files keep upstream conventions (star imports, long lines, math-symbol
# argument names) so they stay easy to diff against upstream releases.
"vllm_omni/model_executor/models/minicpmo_4_5/minicpmo_4_5_omni_llm.py" = ["E402", "E501", "F403", "F405", "N803"]
[tool.ruff.lint.flake8-tidy-imports.banned-api]
"librosa".msg = "The librosa module is banned, use vllm.multimodal helpers instead"
@@ -0,0 +1,2 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
@@ -0,0 +1,264 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Unit tests for the MiniCPM-o 4.5 stage 0 -> stage 1 bridge.
Covers ``vllm_omni.model_executor.stage_input_processors.minicpmo_4_5_omni.llm2tts``:
- empty ``source_outputs`` raises
- latent fallback to ``hidden_states`` when ``multimodal_output`` is empty
- both inputs missing -> raises
- additional_information payload carries the keys the talker expects
(prompt_embeds, prompt_token_ids, llm_output_token_ids, llm_output_text)
- dummy talker prompt ``[BOS, PAD, EOS] = [1, 0, 2]`` (single prefill step)
- MiniCPM-o 4.5 TTS region detection on 151703 / 151704 tokens
- MiniCPM-o 2.6 fallback detection on 151691 / 151692 when no 4.5 markers
- No TTS markers present -> no ``tts_token_ids`` / ``tts_hidden_states`` keys
- prompt arg is normalized to a list and ``multi_modal_data`` is gated by
``requires_multimodal_data``
"""
from __future__ import annotations
from types import SimpleNamespace
import pytest
import torch
from vllm_omni.model_executor.stage_input_processors.minicpmo_4_5_omni import llm2tts
pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
_HIDDEN_DIM = 4
def _make_thinker_output(
*,
prompt_token_ids: list[int],
output_token_ids: list[int],
text: str = "",
request_id: str = "req-0",
latent: torch.Tensor | None = None,
hidden_states: torch.Tensor | None = None,
):
"""Construct a minimal mock of a thinker engine output entry.
The real ``llm2tts`` only reads a tight slice of fields:
- top-level: ``request_id``, ``prompt_token_ids``, ``outputs[0]``
- per-output: ``multimodal_output`` (dict), ``hidden_states`` (opt),
``text``, ``token_ids``
"""
output = SimpleNamespace(
multimodal_output={"latent": latent} if latent is not None else {},
token_ids=output_token_ids,
text=text,
)
if hidden_states is not None:
output.hidden_states = hidden_states
return SimpleNamespace(
request_id=request_id,
prompt_token_ids=prompt_token_ids,
outputs=[output],
)
class TestInputValidation:
def test_empty_source_outputs_raises(self) -> None:
with pytest.raises(ValueError, match="source_outputs cannot be empty"):
llm2tts([], prompt=None)
def test_missing_latent_and_hidden_states_raises(self) -> None:
bad = _make_thinker_output(
prompt_token_ids=[10, 11],
output_token_ids=[20],
)
with pytest.raises(ValueError, match="No latent or hidden_states"):
llm2tts([bad], prompt=None)
class TestBasicShape:
def test_returns_one_entry_per_input(self) -> None:
hidden = torch.zeros((3, _HIDDEN_DIM))
out = llm2tts(
[
_make_thinker_output(prompt_token_ids=[10, 11], output_token_ids=[20], hidden_states=hidden),
_make_thinker_output(
prompt_token_ids=[12], output_token_ids=[21, 22], hidden_states=hidden, request_id="req-1"
),
],
prompt=None,
)
assert len(out) == 2
def test_talker_prompt_token_ids_dummy_bos_pad_eos(self) -> None:
hidden = torch.zeros((2, _HIDDEN_DIM))
out = llm2tts(
[_make_thinker_output(prompt_token_ids=[10], output_token_ids=[20], hidden_states=hidden)],
prompt=None,
)
# The talker AR framework needs *some* tokens to do a single prefill;
# this is the agreed [BOS, PAD, EOS] minimal payload.
assert out[0]["prompt_token_ids"] == [1, 0, 2]
def test_additional_information_carries_thinker_outputs(self) -> None:
prompt_ids = [10, 11, 12]
out_ids = [20, 21]
hidden = torch.randn(len(prompt_ids) + len(out_ids), _HIDDEN_DIM)
result = llm2tts(
[
_make_thinker_output(
prompt_token_ids=prompt_ids,
output_token_ids=out_ids,
text="hello",
hidden_states=hidden,
)
],
prompt=None,
)
ai = result[0]["additional_information"]
assert ai["prompt_token_ids"] == prompt_ids
assert ai["llm_output_token_ids"] == out_ids
assert ai["llm_output_text"] == ["hello"]
# prompt_embeds = float32 view of the prompt-portion of hidden states.
assert ai["prompt_embeds"].dtype == torch.float32
assert ai["prompt_embeds"].shape == (len(prompt_ids), _HIDDEN_DIM)
assert torch.equal(ai["prompt_embeds"], hidden[: len(prompt_ids)].to(torch.float32))
def test_latent_in_multimodal_output_takes_precedence(self) -> None:
# When both ``multimodal_output["latent"]`` and ``hidden_states`` are
# present, the latent payload must win (this is the steady-state path
# produced by the thinker stage).
prompt_ids = [10, 11]
out_ids = [20]
latent = torch.ones((len(prompt_ids) + len(out_ids), _HIDDEN_DIM))
hidden = torch.zeros_like(latent)
result = llm2tts(
[
_make_thinker_output(
prompt_token_ids=prompt_ids,
output_token_ids=out_ids,
latent=latent,
hidden_states=hidden,
)
],
prompt=None,
)
ai = result[0]["additional_information"]
# latent (ones) won over hidden_states (zeros)
assert torch.equal(ai["prompt_embeds"], latent[: len(prompt_ids)].to(torch.float32))
class TestTtsRegionDetection:
"""The bridge auto-detects which MiniCPM-o generation produced the output
by sniffing for 4.5-specific BOS/EOS token ids (151703 / 151704); when
those are absent it falls back to the 2.6 ids (151691 / 151692).
This is a regression guard: a single off-by-one or wrong branch
precedence here causes the talker to receive an empty / wrong-region
slice and emit silent audio.
"""
def _run(self, prompt_token_ids, output_token_ids):
total = len(prompt_token_ids) + len(output_token_ids)
hidden = torch.arange(total * _HIDDEN_DIM, dtype=torch.float32).reshape(total, _HIDDEN_DIM)
result = llm2tts(
[
_make_thinker_output(
prompt_token_ids=prompt_token_ids,
output_token_ids=output_token_ids,
hidden_states=hidden,
)
],
prompt=None,
)
return result[0]["additional_information"], hidden
def test_4_5_markers_detected(self) -> None:
# prompt: [10, 11]
# output: [151703, 30, 31, 151704, 40]
# full sequence: [10, 11, 151703, 30, 31, 151704, 40]
# 0 1 2 3 4 5 6
# 4.5 BOS at idx 2 -> slice starts at 3; EOS at idx 5 -> slice ends at 5.
ai, hidden = self._run([10, 11], [151703, 30, 31, 151704, 40])
assert "tts_token_ids" in ai and "tts_hidden_states" in ai
assert ai["tts_token_ids"].tolist() == [30, 31]
assert torch.equal(ai["tts_hidden_states"], hidden[3:5])
def test_4_5_takes_precedence_when_both_markers_present(self) -> None:
# If both 2.6 and 4.5 markers appear (shouldn't really happen on a
# real checkpoint, but exercises the loop's break), the 4.5 markers
# win because the source loop pins on the first 4.5 id seen.
# sequence index: 0 1 2 3 4 5 6 7 8
ai, hidden = self._run([10], [151691, 70, 151692, 151703, 30, 151704, 40])
assert ai["tts_token_ids"].tolist() == [30]
assert torch.equal(ai["tts_hidden_states"], hidden[5:6])
def test_2_6_fallback(self) -> None:
# No 4.5 markers anywhere -> use 2.6 BOS/EOS pair.
# full sequence: [10, 151691, 30, 31, 151692, 40]
# 0 1 2 3 4 5
# BOS at idx 1 -> slice starts at 2; EOS at idx 4 -> slice ends at 4.
ai, hidden = self._run([10], [151691, 30, 31, 151692, 40])
assert ai["tts_token_ids"].tolist() == [30, 31]
assert torch.equal(ai["tts_hidden_states"], hidden[2:4])
def test_bos_without_eos_runs_to_end(self) -> None:
# When BOS is found but EOS is missing (typical for an in-flight or
# truncated decode), the slice should extend to the end of the
# hidden-state matrix instead of being dropped.
# sequence: [10, 11, 151703, 30, 31]
ai, hidden = self._run([10, 11], [151703, 30, 31])
assert ai["tts_token_ids"].tolist() == [30, 31]
assert torch.equal(ai["tts_hidden_states"], hidden[3:5])
def test_no_tts_markers_omits_slice_keys(self) -> None:
# If neither marker pair is present, the bridge should NOT populate
# ``tts_token_ids`` / ``tts_hidden_states`` — the talker should fall
# through to the dummy path.
ai, _ = self._run([10, 11], [20, 21, 22])
assert "tts_token_ids" not in ai
assert "tts_hidden_states" not in ai
class TestPromptAndMultiModal:
def test_prompt_can_be_single_dict_not_a_list(self) -> None:
hidden = torch.zeros((2, _HIDDEN_DIM))
# A single (non-list) prompt should be auto-wrapped without raising.
llm2tts(
[_make_thinker_output(prompt_token_ids=[10], output_token_ids=[20], hidden_states=hidden)],
prompt={"multi_modal_data": {"audio": "ignored"}},
requires_multimodal_data=False,
)
def test_multimodal_dropped_when_not_requested(self) -> None:
hidden = torch.zeros((2, _HIDDEN_DIM))
out = llm2tts(
[_make_thinker_output(prompt_token_ids=[10], output_token_ids=[20], hidden_states=hidden)],
prompt={"multi_modal_data": {"audio": "should-be-ignored"}},
requires_multimodal_data=False,
)
assert out[0]["multi_modal_data"] is None
def test_multimodal_forwarded_when_requested(self) -> None:
hidden = torch.zeros((2, _HIDDEN_DIM))
mm = {"audio": "forward-me"}
out = llm2tts(
[_make_thinker_output(prompt_token_ids=[10], output_token_ids=[20], hidden_states=hidden)],
prompt={"multi_modal_data": mm},
requires_multimodal_data=True,
)
assert out[0]["multi_modal_data"] == mm
def test_streaming_context_is_accepted_and_ignored(self) -> None:
# The framework passes ``streaming_context`` for streaming-capable
# pipelines; MiniCPM-o 4.5 is turn-taking so the bridge must accept
# but discard it without crashing.
hidden = torch.zeros((2, _HIDDEN_DIM))
out = llm2tts(
[_make_thinker_output(prompt_token_ids=[10], output_token_ids=[20], hidden_states=hidden)],
prompt=None,
streaming_context=object(),
)
assert len(out) == 1
@@ -0,0 +1,144 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Unit tests for the MiniCPM-o 4.5 pipeline registration.
Covers:
- pipeline declared in the central registry
- lazy loader returns the expected ``PipelineConfig``
- 2-stage topology (thinker LLM_AR + talker LLM_AR with audio output)
- stage 1 routes through ``llm2tts`` custom input processor
- ``hf_architectures`` covers both the shared ``MiniCPMO`` alias and the
explicit 4.5 arch
- ``hf_config_predicate`` selects MiniCPM-o 4.5 only and rejects 2.6
checkpoints (regression guard for the shared-arch routing collision).
"""
from __future__ import annotations
from types import SimpleNamespace
import pytest
from vllm_omni.config.pipeline_registry import _OMNI_PIPELINES
from vllm_omni.config.stage_config import (
_PIPELINE_REGISTRY,
PipelineConfig,
StageExecutionType,
)
pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
_PIPELINE_KEY = "minicpmo_4_5"
class TestRegistryDeclaration:
def test_declared_in_omni_pipelines(self) -> None:
assert _PIPELINE_KEY in _OMNI_PIPELINES
def test_visible_in_central_registry(self) -> None:
assert _PIPELINE_KEY in _PIPELINE_REGISTRY
def test_lazy_load_returns_pipeline_config(self) -> None:
pipeline = _PIPELINE_REGISTRY[_PIPELINE_KEY]
assert isinstance(pipeline, PipelineConfig)
assert pipeline.model_type == _PIPELINE_KEY
assert pipeline.model_arch == "MiniCPMO45OmniForConditionalGeneration"
class TestPipelineTopology:
@pytest.fixture(scope="class")
def pipeline(self) -> PipelineConfig:
return _PIPELINE_REGISTRY[_PIPELINE_KEY]
def test_two_stages(self, pipeline: PipelineConfig) -> None:
assert len(pipeline.stages) == 2
assert [s.stage_id for s in pipeline.stages] == [0, 1]
def test_topology_validates(self, pipeline: PipelineConfig) -> None:
# ``validate`` returns a list of structural errors; empty == valid.
assert pipeline.validate() == []
def test_thinker_stage(self, pipeline: PipelineConfig) -> None:
thinker = pipeline.get_stage(0)
assert thinker is not None
assert thinker.model_stage == "llm"
assert thinker.execution_type == StageExecutionType.LLM_AR
assert thinker.input_sources == ()
assert thinker.final_output is True
assert thinker.final_output_type == "text"
assert thinker.owns_tokenizer is True
assert thinker.requires_multimodal_data is True
def test_talker_stage(self, pipeline: PipelineConfig) -> None:
talker = pipeline.get_stage(1)
assert talker is not None
assert talker.model_stage == "tts"
assert talker.execution_type == StageExecutionType.LLM_AR
# talker consumes thinker output
assert talker.input_sources == (0,)
assert talker.final_output is True
assert talker.final_output_type == "audio"
assert talker.engine_output_type == "audio"
# scope KV cache / mrope sizing to talker sub-config
assert talker.hf_config_name == "tts_config"
def test_talker_routes_through_llm2tts(self, pipeline: PipelineConfig) -> None:
talker = pipeline.get_stage(1)
assert talker is not None
# stage 1's custom_process_input_func is what bridges thinker
# hidden_states + token ids into the talker; if this drifts the
# talker silently goes through the dummy path.
assert talker.custom_process_input_func == (
"vllm_omni.model_executor.stage_input_processors.minicpmo_4_5_omni.llm2tts"
)
class TestArchAliases:
"""``hf_architectures`` must cover both the shared and explicit names."""
@pytest.fixture(scope="class")
def pipeline(self) -> PipelineConfig:
return _PIPELINE_REGISTRY[_PIPELINE_KEY]
def test_shared_minicpmo_alias_present(self, pipeline: PipelineConfig) -> None:
# MiniCPM-o 4.5 ships ``architectures=["MiniCPMO"]`` in its HF config.
# Without this alias the arch-fallback path in StageConfigFactory
# cannot resolve the pipeline.
assert "MiniCPMO" in pipeline.hf_architectures
def test_explicit_4_5_arch_present(self, pipeline: PipelineConfig) -> None:
# Reserve the explicit arch name for future repos that opt into it.
assert "MiniCPMO45OmniForConditionalGeneration" in pipeline.hf_architectures
class TestHfConfigPredicate:
"""Regression guard for the 2.6 / 4.5 shared-arch routing collision.
Both MiniCPM-o 2.6 and 4.5 ship ``architectures=["MiniCPMO"]`` in HF
config. The 4.5 pipeline uses ``hf_config_predicate`` to opt in only
when ``config.version == "4.5"``; without it, a 2.6 checkpoint would
intersect on the shared arch and get misrouted into the 4.5 pipeline.
"""
@pytest.fixture(scope="class")
def predicate(self):
pipeline = _PIPELINE_REGISTRY[_PIPELINE_KEY]
assert pipeline.hf_config_predicate is not None, (
"MiniCPM-o 4.5 pipeline must declare hf_config_predicate to "
"avoid misrouting MiniCPM-o 2.6 checkpoints into the 4.5 path."
)
return pipeline.hf_config_predicate
def test_accepts_4_5_string(self, predicate) -> None:
assert predicate(SimpleNamespace(version="4.5")) is True
def test_rejects_2_6_string(self, predicate) -> None:
assert predicate(SimpleNamespace(version="2.6")) is False
def test_rejects_missing_version(self, predicate) -> None:
# 1.0 / older checkpoints do not carry ``version`` at all.
assert predicate(SimpleNamespace()) is False
def test_rejects_empty_version(self, predicate) -> None:
assert predicate(SimpleNamespace(version="")) is False
+4
View File
@@ -130,6 +130,10 @@ _OMNI_PIPELINES: dict[str, tuple[str, str]] = {
"vllm_omni.model_executor.models.moss_tts_nano.pipeline",
"MOSS_TTS_NANO_PIPELINE",
),
"minicpmo_4_5": (
"vllm_omni.model_executor.models.minicpmo_4_5.pipeline",
"MINICPMO_4_5_PIPELINE",
),
"higgs_audio_v2": (
"vllm_omni.model_executor.models.higgs_audio_v2.pipeline",
"HIGGS_AUDIO_V2_PIPELINE",
+38 -2
View File
@@ -7,6 +7,7 @@ from __future__ import annotations
import dataclasses
import re
import warnings
from collections.abc import Callable
from dataclasses import asdict, dataclass, field, fields
from enum import Enum
from pathlib import Path
@@ -242,6 +243,18 @@ class PipelineConfig:
# matches ``hf_config.architectures[*]`` against this tuple to route
# to the correct pipeline. Leave empty for models with unique model_type.
hf_architectures: tuple[str, ...] = ()
# Optional second-stage predicate for resolving an arch-name collision
# between sibling model generations that ship the same
# ``architectures=[...]`` entry. When the arch-fallback in
# ``StageConfigFactory.create_from_model`` finds an intersection with
# ``hf_architectures``, it additionally evaluates this predicate against
# the loaded ``hf_config`` and only selects this pipeline when it
# returns ``True``. Leave ``None`` to skip the extra check (default).
# Example: MiniCPM-o 4.5 and 2.6 both ship ``architectures=["MiniCPMO"]``
# but differ on the ``version`` field, so the 4.5 pipeline declares
# ``hf_config_predicate=lambda c: getattr(c, "version", "") == "4.5"``
# to avoid misrouting 2.6 checkpoints.
hf_config_predicate: Callable[[Any], bool] | None = None
# Diffusers pipeline class name: for models that ship a ``model_index.json``
# (no root ``config.json``), the ``_class_name`` field is matched against
# this value to auto-detect the pipeline. Only needed for diffusers-style
@@ -1179,12 +1192,35 @@ class StageConfigFactory:
# --- HF architecture fallback: some models report a generic
# model_type that collides with another model. Match by the
# hf_architectures declared on each registered PipelineConfig.
# When a pipeline declares ``hf_config_predicate`` (e.g. to
# disambiguate generation-revision collisions like MiniCPM-o 2.6 vs
# 4.5 which share ``architectures=["MiniCPMO"]``), the predicate
# must also accept the loaded ``hf_config``.
if hf_config is not None:
hf_archs = set(getattr(hf_config, "architectures", []) or [])
if hf_archs:
for registered in _PIPELINE_REGISTRY.values():
if hf_archs.intersection(registered.hf_architectures):
return cls._create_from_registry(registered.model_type, cli_overrides, deploy_config_path)
if not hf_archs.intersection(registered.hf_architectures):
continue
predicate = registered.hf_config_predicate
if predicate is not None:
try:
if not predicate(hf_config):
logger.debug(
"Pipeline %r matched on architectures %s but its "
"hf_config_predicate rejected the loaded config; "
"continuing fallback search.",
registered.model_type,
sorted(hf_archs.intersection(registered.hf_architectures)),
)
continue
except Exception:
logger.exception(
"Pipeline %r hf_config_predicate raised; skipping.",
registered.model_type,
)
continue
return cls._create_from_registry(registered.model_type, cli_overrides, deploy_config_path)
# --- Legacy path: load from pipeline YAML ---
pipeline = cls._load_pipeline(model, trust_remote_code=trust_remote_code)
+51
View File
@@ -0,0 +1,51 @@
# MiniCPM-o 4.5 default deploy: 2-GPU layout.
#
# Stage 0 (thinker): GPU 0 — multimodal understanding, ~70% mem.
# Stage 1 (talker) : GPU 1 — MiniCPMTTS + Token2Wav, ~75% mem
# (covers both the talker LM and the in-process
# Token2Wav vocoder that emits the waveform).
#
# Other GPU layouts: minicpmo_4_5_3gpu.yaml, minicpmo_4_5_8x4090.yaml.
async_chunk: false
trust_remote_code: true
enable_prefix_caching: false
stages:
- stage_id: 0
max_num_seqs: 1
gpu_memory_utilization: 0.7
enforce_eager: true
tensor_parallel_size: 1
max_num_batched_tokens: 16384
devices: "0"
limit_mm_per_prompt:
video: 1
default_sampling_params:
temperature: 0.0
top_p: 1.0
top_k: -1
max_tokens: 2048
seed: 42
detokenize: true
repetition_penalty: 1.1
- stage_id: 1
# Talker only consumes runtime_additional_information[0]; any value > 1
# makes concurrent requests share request-0's audio (mm_outputs fallback).
max_num_seqs: 1
gpu_memory_utilization: 0.75
enforce_eager: true
max_num_batched_tokens: 8192
devices: "1"
default_sampling_params:
temperature: 0.0
top_p: 1.0
top_k: -1
max_tokens: 1
seed: 42
detokenize: false
# MiniCPM-o 4.5 has only been verified on CUDA devices. Add per-platform
# overrides here when other backends (ROCm/XPU/NPU) are verified end-to-end.
platforms: {}
+48
View File
@@ -0,0 +1,48 @@
# MiniCPM-o 4.5 deploy: 3-GPU layout (thinker TP=2 + talker on GPU 2).
#
# Stage 0 (thinker): GPU 0,1 — 2-way tensor parallel, ~70% mem each.
# Stage 1 (talker) : GPU 2 — MiniCPMTTS + Token2Wav, ~75% mem
# (covers both the talker LM and the in-process
# Token2Wav vocoder that emits the waveform).
pipeline: minicpmo_4_5
async_chunk: false
trust_remote_code: true
enable_prefix_caching: false
stages:
- stage_id: 0
max_num_seqs: 1
gpu_memory_utilization: 0.7
enforce_eager: true
tensor_parallel_size: 2
max_num_batched_tokens: 16384
devices: "0,1"
limit_mm_per_prompt:
video: 1
default_sampling_params:
temperature: 0.0
top_p: 1.0
top_k: -1
max_tokens: 2048
seed: 42
detokenize: true
repetition_penalty: 1.1
- stage_id: 1
# Talker only consumes runtime_additional_information[0]; any value > 1
# makes concurrent requests share request-0's audio (mm_outputs fallback).
max_num_seqs: 1
gpu_memory_utilization: 0.75
enforce_eager: true
max_num_batched_tokens: 8192
devices: "2"
default_sampling_params:
temperature: 0.0
top_p: 1.0
top_k: -1
max_tokens: 1
seed: 42
detokenize: false
platforms: {}
+60
View File
@@ -0,0 +1,60 @@
# MiniCPM-o 4.5 deploy: 8x4090 layout (24GB cards).
#
# Stage 0 (thinker): GPU 0,1,2,3 — 4-way TP, ~85% mem (≈20.4 GB / card,
# thinker weights take ~17.5 GB per card
# on 4090, leaving ~3 GB for KV cache /
# activations).
# Stage 1 (talker) : GPU 4 — MiniCPMTTS + Token2Wav, ~90% mem
# (covers both the talker LM and the
# in-process Token2Wav vocoder that emits
# the waveform).
#
# max_model_len is pinned to 4096 because the upstream config.json
# advertises max_position_embeddings=40960 (tuned for H20 141GB); on a
# 24 GB 4090 a 40k-token KV cache request OOMs the worker process
# before serving the first request. 8192 also OOMs in practice with
# this thinker model + vision/audio encoders + Whisper preallocation,
# so 4096 is the empirically-verified safe ceiling.
pipeline: minicpmo_4_5
async_chunk: false
trust_remote_code: true
enable_prefix_caching: false
stages:
- stage_id: 0
max_num_seqs: 1
gpu_memory_utilization: 0.85
enforce_eager: true
tensor_parallel_size: 4
max_model_len: 4096
max_num_batched_tokens: 4096
devices: "0,1,2,3"
limit_mm_per_prompt:
video: 1
default_sampling_params:
temperature: 0.0
top_p: 1.0
top_k: -1
max_tokens: 2048
seed: 42
detokenize: true
repetition_penalty: 1.1
- stage_id: 1
# Talker only consumes runtime_additional_information[0]; any value > 1
# makes concurrent requests share request-0's audio (mm_outputs fallback).
max_num_seqs: 1
gpu_memory_utilization: 0.9
enforce_eager: true
max_num_batched_tokens: 8192
devices: "4"
default_sampling_params:
temperature: 0.0
top_p: 1.0
top_k: -1
max_tokens: 1
seed: 42
detokenize: false
platforms: {}
+6 -3
View File
@@ -78,10 +78,13 @@ def register_omni_models_to_vllm():
_register_omni_hf_configs()
supported_archs = ModelRegistry.get_supported_archs()
# Always register every arch declared in _OMNI_MODELS — these are the
# archs vllm-omni explicitly owns, so we want our implementation to
# win even if a future vllm release ships a same-named built-in.
# vllm's ModelRegistry.register_model already emits a warning and
# overrides on collision; we let it handle that.
for arch, (mod_folder, mod_relname, cls_name) in _OMNI_MODELS.items():
if arch not in supported_archs:
ModelRegistry.register_model(arch, f"vllm_omni.model_executor.models.{mod_folder}.{mod_relname}:{cls_name}")
ModelRegistry.register_model(arch, f"vllm_omni.model_executor.models.{mod_folder}.{mod_relname}:{cls_name}")
@dataclass
+8 -1
View File
@@ -44,11 +44,18 @@ def main():
description="vLLM OMNI CLI",
epilog=VLLM_SUBCMD_PARSER_EPILOG.format(subcmd="[subcommand]"),
)
try:
_omni_version = importlib.metadata.version("vllm_omni")
except importlib.metadata.PackageNotFoundError:
try:
from vllm_omni.version import __version__ as _omni_version # type: ignore
except Exception:
_omni_version = "dev"
parser.add_argument(
"-v",
"--version",
action="version",
version=importlib.metadata.version("vllm_omni"),
version=_omni_version,
)
subparsers = parser.add_subparsers(required=False, dest="subparser")
cmds = {}
@@ -578,6 +578,31 @@ async def build_async_omni_from_stage_config(
async_omni: EngineClient | None = None
# Pre-load the model config so HuggingFace registers `transformers_modules`
# in this process — only when the user has explicitly opted in via
# `--trust-remote-code`. Stage workers consume the same flag through their
# deploy config, but the API server process also needs the dynamic modules
# for ZMQ pickle deserialization of stage outputs that reference them
# (e.g. trust_remote_code models like MiniCPM-o).
if getattr(args, "trust_remote_code", False) and getattr(args, "model", None):
try:
import os
from transformers import AutoConfig
# Hide GPUs so the custom config code doesn't allocate CUDA memory.
saved = os.environ.get("CUDA_VISIBLE_DEVICES")
os.environ["CUDA_VISIBLE_DEVICES"] = ""
try:
AutoConfig.from_pretrained(args.model, trust_remote_code=True)
finally:
if saved is not None:
os.environ["CUDA_VISIBLE_DEVICES"] = saved
else:
os.environ.pop("CUDA_VISIBLE_DEVICES", None)
except Exception as e:
logger.debug("Pre-loading transformers_modules failed: %s", e)
try:
kwargs = vars(args).copy()
kwargs.pop("model", None)
@@ -0,0 +1,9 @@
from .minicpmo_4_5_omni import MiniCPMO45OmniForConditionalGeneration
from .minicpmo_4_5_omni_llm import MiniCPMO45OmniLLMForConditionalGeneration
from .minicpmo_4_5_omni_tts import MiniCPMO45OmniTTSForConditionalGeneration
__all__ = [
"MiniCPMO45OmniForConditionalGeneration",
"MiniCPMO45OmniLLMForConditionalGeneration",
"MiniCPMO45OmniTTSForConditionalGeneration",
]
@@ -0,0 +1,384 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
# Adapted from:
# https://huggingface.co/openbmb/MiniCPM-o-4_5/blob/main/modeling_minicpmo.py
#
# Copyright 2025 The OpenBMB 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 collections.abc import Iterable
from functools import cached_property
import torch
import torch.nn as nn
from vllm.config import VllmConfig
from vllm.logger import init_logger
from vllm.model_executor.models.interfaces import SupportsMRoPE, SupportsMultiModal, SupportsPP
from vllm.model_executor.models.utils import init_vllm_registered_model, maybe_prefix
from vllm.multimodal import MULTIMODAL_REGISTRY
from vllm.sequence import IntermediateTensors
from vllm.v1.outputs import SamplerOutput
from vllm.v1.sample.metadata import SamplingMetadata
from vllm_omni.model_executor.models.minicpmo_4_5.minicpmo_4_5_omni_llm import (
MiniCPMO45OmniLLMDummyInputsBuilder,
MiniCPMO45OmniLLMMultiModalProcessor,
MiniCPMO45OmniLLMProcessingInfo,
MiniCPMOConfig,
)
from vllm_omni.model_executor.models.output_templates import OmniOutput
from vllm_omni.model_executor.models.utils import add_prefix_to_loaded_weights
from vllm_omni.platforms import current_omni_platform
logger = init_logger(__name__)
@MULTIMODAL_REGISTRY.register_processor(
MiniCPMO45OmniLLMMultiModalProcessor,
info=MiniCPMO45OmniLLMProcessingInfo,
dummy_inputs=MiniCPMO45OmniLLMDummyInputsBuilder,
)
class MiniCPMO45OmniForConditionalGeneration(nn.Module, SupportsMultiModal, SupportsPP, SupportsMRoPE):
"""MiniCPM-o 4.5 Omni model for conditional generation.
Two-stage pipeline:
- thinker (model_stage="llm"): image / video / audio encoders + 3D
resampler + the omni LLM that emits text + hidden states.
- talker (model_stage="tts"): MiniCPMTTS + the in-process Token2Wav
vocoder that emits the final audio waveform directly.
"""
@classmethod
def get_placeholder_str(cls, modality: str, i: int) -> str | None:
if modality.startswith("image"):
return "(<image>./</image>)"
if modality.startswith("video"):
return "(<video>./</video>)"
if modality.startswith("audio"):
return "(<audio>./</audio>)"
raise ValueError("Only image, video or audio modality is supported")
def __init__(self, *, vllm_config: VllmConfig, prefix: str = ""):
super().__init__()
self.have_multimodal_outputs = True
config: MiniCPMOConfig = vllm_config.model_config.hf_config
multimodal_config = vllm_config.model_config.multimodal_config
# keep vllm_config for later submodule init
self.vllm_config = vllm_config
# Store configs
self.config = config
self.multimodal_config = multimodal_config
self.model_stage = vllm_config.model_config.model_stage
if self.model_stage == "llm":
# Initialize thinker model (image preprocessing + vision encoder + 3D resampler)
self.thinker = init_vllm_registered_model(
vllm_config=vllm_config,
prefix=maybe_prefix(prefix, "thinker"),
hf_config=config,
# Registry key — must match the entries declared in
# vllm_omni/model_executor/models/registry.py::_OMNI_MODELS.
architectures=["MiniCPMO45OmniLLMForConditionalGeneration"],
)
self.model = self.thinker
self.talker = None
elif self.model_stage == "tts":
self.thinker = None
# Initialize talker model — runs MiniCPMTTS + the in-process
# Token2Wav vocoder and emits the final audio waveform directly.
self.talker = init_vllm_registered_model(
vllm_config=vllm_config,
prefix=maybe_prefix(prefix, "talker"),
hf_config=config,
# Registry key — must match the entries declared in
# vllm_omni/model_executor/models/registry.py::_OMNI_MODELS.
architectures=["MiniCPMO45OmniTTSForConditionalGeneration"],
)
# Initialize multimodal components if needed
if hasattr(self.talker, "init_multi_modal"):
self.talker.init_multi_modal(config)
self.model = self.talker
else:
raise ValueError(f"Invalid model stage: {self.model_stage}. Must be one of: 'llm', 'tts'")
# Set up intermediate tensors
self.make_empty_intermediate_tensors = (
(self.thinker.make_empty_intermediate_tensors)
if self.model_stage == "llm" and self.thinker is not None
else lambda: None
)
self._language_model_names = ["model"]
@cached_property
def sampler(self):
if hasattr(self.model, "sampler"):
return self.model.sampler
from vllm.v1.sample.sampler import Sampler
return Sampler()
# -------------------- Device utilities --------------------
@staticmethod
def _module_device(module: nn.Module) -> torch.device:
try:
return next(module.parameters()).device
except StopIteration:
# No parameters; fall back to buffers or cpu
for _, buf in module.named_buffers(recurse=True):
return buf.device
return torch.device("cpu")
def move_submodules_to_devices(
self,
*,
thinker_device: str | torch.device | None = None,
talker_device: str | torch.device | None = None,
) -> None:
"""Optionally move thinker / talker to different devices.
Example:
model.move_submodules_to_devices(
thinker_device='cuda:0',
talker_device='cuda:1',
)
"""
if thinker_device is not None and self.thinker is not None:
self.thinker.to(thinker_device)
if talker_device is not None and self.talker is not None:
self.talker.to(talker_device)
def get_input_embeddings(
self,
input_ids: torch.Tensor,
multimodal_embeddings=None,
) -> torch.Tensor:
return self.model.get_input_embeddings(input_ids, multimodal_embeddings)
def embed_input_ids(
self,
input_ids: torch.Tensor,
multimodal_embeddings=None,
*,
is_multimodal=None,
) -> torch.Tensor:
if self.model_stage == "tts":
return self.get_input_embeddings(input_ids)
return super().embed_input_ids(input_ids, multimodal_embeddings, is_multimodal=is_multimodal)
def get_multimodal_embeddings(self, **kwargs):
# Delegate to the active stage submodule when it implements MM encoding.
mm_fn = getattr(self.model, "get_multimodal_embeddings", None)
if mm_fn is not None:
return mm_fn(**kwargs)
return []
def embed_multimodal(self, **kwargs: object):
"""vLLM V1 encoder profiling calls this; the inherited Protocol stub returns None."""
return self.get_multimodal_embeddings(**kwargs)
def forward(
self,
input_ids: torch.Tensor,
positions: torch.Tensor,
intermediate_tensors: IntermediateTensors | None = None,
inputs_embeds: torch.Tensor | None = None,
sampling_metadata: SamplingMetadata | None = None,
logits_index: int | None = None,
sampler=None,
additional_information: dict[str, object] | None = None,
**kwargs: object,
) -> torch.Tensor | IntermediateTensors | OmniOutput:
"""
Forward pass for MiniCPM-o Omni model.
Workflow:
1) Thinker (model_stage="llm"): Image / video / audio encoders +
3D resampler + omni LLM → text + hidden states.
2) Talker (model_stage="tts"): MiniCPMTTS + the in-process
Token2Wav vocoder → audio waveform (final pipeline output).
"""
if self.model_stage == "llm":
# Normalize to batched inputs if caller provides 1D/2D unbatched tensors
# TODO: Remove this hack when NPU supports batched inputs properly
added_batch_dim = False
if input_ids is not None and input_ids.ndim == 1:
input_ids = input_ids.unsqueeze(0)
added_batch_dim = True
if positions is not None and positions.ndim == 1:
positions = positions.unsqueeze(0)
added_batch_dim = True
if inputs_embeds is not None and inputs_embeds.ndim == 2:
inputs_embeds = inputs_embeds.unsqueeze(0)
added_batch_dim = True
thinker_dev = self._module_device(self.thinker)
# if input_ids is None, set it to a zero tensor
if input_ids is None:
input_ids = torch.zeros(inputs_embeds.shape[1], dtype=torch.long, device=thinker_dev).unsqueeze(0)
added_batch_dim = True
# Ensure inputs on thinker's device
if input_ids is not None and input_ids.device != thinker_dev:
input_ids = input_ids.to(thinker_dev)
if positions is not None and positions.device != thinker_dev:
positions = positions.to(thinker_dev)
if inputs_embeds is not None and inputs_embeds.device != thinker_dev:
inputs_embeds = inputs_embeds.to(thinker_dev)
if current_omni_platform.is_npu():
# TODO: remove this hack when NPU supports batched inputs properly
thinker_input_ids = input_ids[0] if input_ids is not None and added_batch_dim else input_ids
thinker_positions = positions[0] if positions.ndim > 1 else positions
thinker_inputs_embeds = (
inputs_embeds[0] if inputs_embeds is not None and added_batch_dim else inputs_embeds
)
else:
thinker_input_ids = input_ids[0] if input_ids is not None and added_batch_dim else input_ids
thinker_positions = positions[0] if positions is not None and added_batch_dim else positions
thinker_inputs_embeds = (
inputs_embeds[0] if inputs_embeds is not None and added_batch_dim else inputs_embeds
)
# Run thinker
thinker_output = self.thinker(
input_ids=thinker_input_ids,
positions=thinker_positions,
intermediate_tensors=intermediate_tensors,
inputs_embeds=thinker_inputs_embeds,
**kwargs,
)
if isinstance(thinker_output, tuple):
embeds, text_hidden_states = thinker_output
else:
text_hidden_states = thinker_output
# Prepare hidden states for downstream stages
# Ensure correct shape: (batch_size, seq_len, hidden_dim)
if added_batch_dim:
text_hidden_states = text_hidden_states.squeeze(0)
# Return hidden states with latent in multimodal_outputs for stage_input_processors
return OmniOutput(
text_hidden_states=text_hidden_states,
multimodal_outputs={"latent": text_hidden_states},
)
# Talker stage: runs MiniCPMTTS + the in-process Token2Wav vocoder and
# emits the final audio waveform directly.
if self.model_stage == "tts":
if input_ids is not None:
num_tokens = input_ids.shape[0]
device = input_ids.device
elif inputs_embeds is not None:
num_tokens = inputs_embeds.shape[0]
device = inputs_embeds.device
else:
num_tokens = 1
device = torch.device("cuda")
hidden_dim = self.config.hidden_size if hasattr(self.config, "hidden_size") else 2560
# Profile/dummy run: both input_ids and inputs_embeds are None.
# Note: SupportsMultiModal preprocessing converts input_ids to
# inputs_embeds, so input_ids=None alone does NOT indicate a dummy run.
if input_ids is None and inputs_embeds is None:
dummy_hidden = torch.zeros(num_tokens, hidden_dim, device=device)
return OmniOutput(text_hidden_states=dummy_hidden, multimodal_outputs=None)
runtime_info = kwargs.get("runtime_additional_information")
talker_info = {}
if runtime_info and isinstance(runtime_info, list) and len(runtime_info) > 0:
talker_info = runtime_info[0] if isinstance(runtime_info[0], dict) else {}
with torch.inference_mode():
talker_result = self.talker(
input_ids=input_ids,
positions=positions,
inputs_embeds=inputs_embeds,
additional_information=talker_info,
)
dummy_hidden = torch.zeros(num_tokens, hidden_dim, device=device)
# Talker returns a (mel_spec, waveform_or_None) tuple. MiniCPM-o
# 4.5 emits only the waveform (mel_spec is always None); keep the
# 2-slot unpack so a future mel-emitting variant can plug in
# without changing the wrapper.
mm_out: dict = {}
if isinstance(talker_result, tuple) and len(talker_result) == 2:
_, waveform = talker_result
if waveform is not None:
mm_out["model_outputs"] = [waveform]
return OmniOutput(
text_hidden_states=dummy_hidden,
multimodal_outputs=mm_out if mm_out else None,
)
raise ValueError(f"Unsupported model stage: {self.model_stage}")
def compute_logits(self, hidden_states: torch.Tensor | OmniOutput) -> torch.Tensor | None:
# Handle OmniOutput type
if isinstance(hidden_states, OmniOutput):
hidden_states = hidden_states.text_hidden_states
# Use model for logits computation
return self.model.compute_logits(hidden_states)
def sample(
self,
logits: torch.Tensor,
sampling_metadata: SamplingMetadata,
) -> SamplerOutput | None:
# Use model for sampling
return self.model.sample(logits, sampling_metadata)
def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
"""Load weights for the active stage of the omni model."""
loaded_weights = set()
thinker_weights = []
talker_weights = []
# MiniCPM-o checkpoint prefixes → stage mapping:
# thinker: vpm, resampler, llm, apm, audio_projection_layer
# talker: tts (MiniCPMTTS); the Token2Wav vocoder weights load
# separately inside the talker module from the
# ``assets/token2wav`` subdirectory and do not appear
# in this iterator.
for k, v in weights:
if k.startswith(("vpm.", "resampler.", "llm.", "apm.", "audio_projection_layer.")):
thinker_weights.append((k, v))
elif k.startswith("tts."):
talker_weights.append((k, v))
else:
logger.warning("Unknown weight prefix: %s, skipping", k)
# Load thinker weights
if self.thinker is not None and thinker_weights:
thinker_loaded = self.thinker.load_weights(thinker_weights)
thinker_loaded = add_prefix_to_loaded_weights(thinker_loaded, "thinker")
loaded_weights.update(thinker_loaded)
# Load talker weights
if self.talker is not None and talker_weights:
talker_loaded = self.talker.load_weights(talker_weights)
talker_loaded = add_prefix_to_loaded_weights(talker_loaded, "talker")
loaded_weights.update(talker_loaded)
return loaded_weights
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,456 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
# Adapted from:
# https://huggingface.co/openbmb/MiniCPM-o-4_5/blob/main/modeling_minicpmo.py
"""MiniCPM-o 4.5 Talker + Token2Wav: MiniCPMTTS with hidden_text_merge condition.
Pipeline:
1. Receive thinker hidden_states + full token IDs via additional_information
2. Extract tts_bos..tts_eos region
3. Build condition: emb_text(tokens) + projector_semantic(hidden) (hidden_text_merge)
4. Run MiniCPMTTS.generate() -> discrete audio tokens
5. Run Token2wav(tokens) -> waveform bytes -> numpy array
"""
import io
import logging
import os
from collections.abc import Iterable
import numpy as np
import soundfile as sf
import torch
import torch.nn as nn
import torch.nn.functional as F
from vllm.config import VllmConfig
from vllm.model_executor.models.interfaces import SupportsPP
try:
from stepaudio2 import Token2wav as _Token2wav
_stepaudio2_available = True
except ImportError:
_Token2wav = None
_stepaudio2_available = False
logger = logging.getLogger(__name__)
def _install_torchaudio_soundfile_shim() -> None:
"""Monkey-patch torchaudio.load to use soundfile instead of the default
torchcodec backend, which requires libtorchcodec/ffmpeg shared libs that
may be missing on the deployment machine."""
try:
import torchaudio
if getattr(torchaudio, "_soundfile_shim_installed", False):
return
_orig_load = torchaudio.load
def _patched_load(uri, *args, **kwargs):
try:
return _orig_load(uri, *args, **kwargs)
except Exception:
import numpy as _np
import soundfile as _sf
data, sr = _sf.read(uri, dtype="float32", always_2d=True)
wav = torch.from_numpy(_np.ascontiguousarray(data.T))
return wav, sr
torchaudio.load = _patched_load
torchaudio._soundfile_shim_installed = True
logger.info("Installed torchaudio.load soundfile shim")
except Exception as _e:
logger.warning("Could not install torchaudio shim: %s", _e)
_install_torchaudio_soundfile_shim()
class MiniCPMO45OmniTTSForConditionalGeneration(nn.Module, SupportsPP):
"""MiniCPM-o 4.5 Talker: MiniCPMTTS + Token2wav in a single forward pass."""
def __init__(self, *, vllm_config: VllmConfig, prefix: str = ""):
super().__init__()
from vllm_omni.model_executor.models.minicpmo_4_5.minicpmo_4_5_omni_llm import MiniCPMOConfig
config: MiniCPMOConfig = vllm_config.model_config.hf_config
self.config = config
self.vllm_config = vllm_config
self.tts = None
self.audio_tokenizer = None
self._assets_loaded = False
tts_config = getattr(config, "tts_config", None)
if tts_config is not None:
self._tts_config = tts_config
self._tts_bos_id = getattr(tts_config, "audio_bos_token_id", 151687)
self._text_eos_id = getattr(tts_config, "text_eos_token_id", 151692)
self._num_audio_tokens = getattr(tts_config, "num_audio_tokens", 6562)
self._hidden_size = getattr(tts_config, "hidden_size", 768)
self._normalize = getattr(tts_config, "normalize_projected_hidden", True)
else:
self._tts_config = None
def _lazy_init_tts(self):
if self._assets_loaded or self._tts_config is None:
return
try:
model_path = self.vllm_config.model_config.model
import os
import sys
if model_path not in sys.path:
sys.path.insert(0, model_path)
from transformers.dynamic_module_utils import get_class_from_dynamic_module
MiniCPMTTS = get_class_from_dynamic_module("modeling_minicpmo.MiniCPMTTS", model_path)
# MiniCPMTTS.__init__ reads `config.top_p / top_k / repetition_penalty`
# directly (modeling_minicpmo.py L4112-4114), but the model repo's
# config.json `tts_config` block does not declare these fields and
# PretrainedConfig in recent transformers no longer surfaces
# generation-style params on `self.config`. Inject the defaults the
# upstream code itself ships with (modeling_minicpmo.py L2212-2214,
# L3132-3133) so attribute access does not raise.
for _attr, _default in (("top_p", 0.8), ("top_k", 100), ("repetition_penalty", 1.02)):
if not hasattr(self._tts_config, _attr):
setattr(self._tts_config, _attr, _default)
prev_dtype = torch.get_default_dtype()
torch.set_default_dtype(torch.float32)
try:
self.tts_obj = MiniCPMTTS(config=self._tts_config, audio_tokenizer=None)
finally:
torch.set_default_dtype(prev_dtype)
self.emb_text = self.tts_obj.emb_text
self.projector_semantic = self.tts_obj.projector_semantic
token2wav_dir = os.path.join(model_path, "assets", "token2wav")
if os.path.isdir(token2wav_dir):
if not _stepaudio2_available:
raise ImportError(
"MiniCPM-o 4.5 token2wav stage requires the `stepaudio2` Python "
"module (a MiniCPM-o-flavored Token2wav vocoder, NOT the upstream "
"stepfun-ai/Step-Audio2 — the upstream signature does not accept "
"n_timesteps and will fail at __init__). Install via:\n"
" pip install 'vllm-omni[minicpmo]' # recommended, declared as PR extra\n"
"Equivalent direct installs of the same `from stepaudio2 import Token2wav`\n"
"entry point used by openbmb/MiniCPM-o-4_5/modeling_minicpmo.py:\n"
" pip install stepaudio2-minicpmo # bare token2wav package\n"
" pip install 'minicpmo-utils[all]' # MiniCPM-o umbrella (also brings image/video deps)"
)
prev_dtype2 = torch.get_default_dtype()
torch.set_default_dtype(torch.float32)
try:
# NB: this must be the MiniCPM-o-flavored Token2wav from
# the `stepaudio2-minicpmo` PyPI package (or the
# `minicpmo-utils[all]` umbrella), not the upstream
# `stepfun-ai/Step-Audio2` repo. The MiniCPM-o variant's
# __init__ accepts n_timesteps; the upstream signature is
# (model_path, float16=False) and will raise
# TypeError on n_timesteps. See ImportError message below
# for installation guidance.
self.audio_tokenizer = _Token2wav(token2wav_dir, float16=False, n_timesteps=10)
finally:
torch.set_default_dtype(prev_dtype2)
self.tts_obj.audio_tokenizer = self.audio_tokenizer
logger.info("Loaded Token2wav from %s", token2wav_dir)
# Only mark init as complete after every step succeeds, so a
# partial failure leaves the next call free to retry the full
# init instead of short-circuiting back to a silent empty path.
self._assets_loaded = True
except ImportError:
# Surface missing dependencies directly so users can act on them
# instead of getting a silent None waveform downstream.
raise
except Exception:
# Re-raise non-import init failures (bad token2wav assets, missing
# weights, OOM during Token2wav construction, etc.) so the server
# fails loudly at startup / first request instead of returning
# silent empty audio for every subsequent request.
logger.error("Failed to init 4.5 TTS", exc_info=True)
raise
def generate_speech(
self,
tts_token_ids: torch.Tensor,
tts_hidden_states: torch.Tensor,
) -> np.ndarray | None:
"""Run full 4.5 TTS pipeline using original MiniCPMTTS.generate."""
self._lazy_init_tts()
if not hasattr(self, "tts_obj") or self.tts_obj is None:
logger.warning("generate_speech: tts_obj not initialized")
return None
tts = self.tts_obj
device = tts.emb_text.weight.device
dtype = tts.emb_text.weight.dtype
llm_embeds = tts.emb_text(tts_token_ids.to(device))
hidden_embeds = tts.projector_semantic(tts_hidden_states.to(device=device, dtype=dtype))
if getattr(tts.config, "normalize_projected_hidden", False):
hidden_embeds = F.normalize(hidden_embeds, p=2, dim=-1)
tts_embeds = llm_embeds + hidden_embeds
text_eos = tts.emb_text(torch.tensor([tts.config.text_eos_token_id], device=device, dtype=torch.long))
audio_bos = tts.emb_text(torch.tensor([tts.audio_bos_token_id], device=device, dtype=torch.long))
spk_embeds = torch.zeros(0, tts.config.hidden_size, device=device, dtype=tts_embeds.dtype)
inputs_embeds = torch.cat([spk_embeds, tts_embeds, text_eos, audio_bos], dim=0).unsqueeze(0)
logger.info("generate_speech: inputs_embeds shape=%s", list(inputs_embeds.shape))
# Scale max_new_token with input text length to avoid mid-stream truncation on long
# responses (default 2048 can only cover ~300 text tokens at ~6x audio/text ratio).
# Empirically 511 text tokens → 1951 audio tokens (~3.8x) finishes cleanly, so use 10x
# as a safe upper bound with a floor of 2048 and a hard cap of 16384 to bound latency/mem.
num_text = int(tts_token_ids.shape[-1]) if tts_token_ids.ndim > 0 else 0
max_new_token = max(2048, min(16384, num_text * 10))
eos_token = torch.tensor([tts.config.num_audio_tokens - 1], dtype=torch.long, device=device)
outputs = tts.generate(
inputs_embeds=inputs_embeds,
eos_token=eos_token,
max_new_token=max_new_token,
show_tqdm=False,
)
generated_tokens = outputs.new_ids.squeeze(-1)
logger.info(
"generate_speech: generated %d audio tokens (cap=%d, text_tokens=%d)",
generated_tokens.shape[-1],
max_new_token,
num_text,
)
if self.audio_tokenizer is None:
logger.warning("No audio_tokenizer")
return None
import torchaudio
model_path = self.vllm_config.model_config.model
default_ref = os.path.join(model_path, "assets", "HT_ref_audio.wav")
prompt_wav_path = default_ref if os.path.exists(default_ref) else None
_orig_save = torchaudio.save
def _patched_save(uri, src, sample_rate, **kw):
kw.pop("backend", None)
if hasattr(uri, "write"):
sf.write(uri, src.cpu().numpy().T, sample_rate, format="WAV")
return
return _orig_save(uri, src, sample_rate, backend="soundfile", **kw)
torchaudio.save = _patched_save
prev_dtype = torch.get_default_dtype()
torch.set_default_dtype(torch.float32)
try:
with torch.amp.autocast("cuda", enabled=False):
token_list = generated_tokens.squeeze(0).tolist()
num_tokens = len(token_list)
# For long outputs, the one-shot vocoder path
# (Token2wav.__call__ -> flow.inference) runs full O(N^2) self-
# attention over all audio tokens and OOMs on a 24GB card once
# N exceeds a few thousand (e.g. 4964 tokens needs ~3GiB for a
# single attention matmul). Switch to the chunked / streaming
# vocoder (set_stream_cache + stream) which truncates the flow
# attention caches to prompt_len + 100 steps on every chunk,
# keeping peak memory bounded regardless of total length.
STREAM_THRESHOLD = int(os.environ.get("MINICPMO45_TTS_STREAM_THRESHOLD", "2500")) # ~100s @ 25Hz
CHUNK_SIZE = int(os.environ.get("MINICPMO45_TTS_STREAM_CHUNK", "50")) # ~2s per chunk
MIN_TAIL = 6 # must exceed flow.pre_lookahead_len (typically 3)
if num_tokens <= STREAM_THRESHOLD:
wav_bytes = self.audio_tokenizer(token_list, prompt_wav_path)
waveform, sr = sf.read(io.BytesIO(wav_bytes))
waveform = waveform.astype(np.float32)
else:
# Build chunk boundaries, merging a too-small tail into the
# previous chunk so every chunk satisfies MIN_TAIL.
boundaries = []
i = 0
while i < num_tokens:
end = min(i + CHUNK_SIZE, num_tokens)
if 0 < num_tokens - end < MIN_TAIL:
end = num_tokens
boundaries.append((i, end))
i = end
logger.info(
"generate_speech: streaming vocoder, %d tokens -> %d chunks (chunk=%d)",
num_tokens,
len(boundaries),
CHUNK_SIZE,
)
stream_cache, hift_cache_dict = self.audio_tokenizer.set_stream_cache(prompt_wav_path)
self.audio_tokenizer.stream_cache = stream_cache
self.audio_tokenizer.hift_cache_dict = hift_cache_dict
try:
pieces = []
for idx, (s, e) in enumerate(boundaries):
is_last = idx == len(boundaries) - 1
wav_np = self.audio_tokenizer.stream(
token_list[s:e],
prompt_wav_path,
last_chunk=is_last,
return_waveform=True,
)
pieces.append(np.asarray(wav_np).reshape(-1))
waveform = np.concatenate(pieces, axis=0).astype(np.float32)
sr = 24000
finally:
# Free per-request streaming state so the next request starts clean
self.audio_tokenizer.stream_cache = None
self.audio_tokenizer.hift_cache_dict = {}
finally:
torch.set_default_dtype(prev_dtype)
torchaudio.save = _orig_save
logger.info("generate_speech: waveform %d samples, sr=%d", waveform.shape[0], sr)
return waveform
def _generate_tokens(self, inputs_embeds: torch.Tensor, max_new_token: int = 2048) -> torch.Tensor | None:
"""Autoregressive generation of audio tokens using the TTS LlamaModel."""
device = inputs_embeds.device
eos_token = self._num_audio_tokens - 1
condition_length = inputs_embeds.shape[1]
num_vq = len(self.emb_code)
new_tokens = torch.zeros(1, max_new_token, num_vq, device=device, dtype=torch.long)
past_key_values = None
finished = False
for t in range(max_new_token):
if t == 0:
emb = inputs_embeds
position_ids = torch.arange(condition_length, device=device).unsqueeze(0)
else:
code_emb = [self.emb_code[q](new_tokens[:, t - 1 : t, q]) for q in range(num_vq)]
emb = torch.stack(code_emb, -1).sum(-1)
position_ids = torch.tensor([[condition_length + t - 1]], device=device)
outputs = self.tts_model(
inputs_embeds=emb,
position_ids=position_ids,
past_key_values=past_key_values,
use_cache=True,
)
hidden = outputs.last_hidden_state
past_key_values = outputs.past_key_values
logits = torch.stack([self.head_code[q](hidden[:, -1]) for q in range(num_vq)], dim=-1)
logits = logits.float() / 0.8
if t < 50:
logits[:, eos_token, :] = -float("inf")
probs = F.softmax(logits, dim=1)
idx = torch.multinomial(probs.view(-1, probs.shape[1]), 1).view(1, num_vq)
new_tokens[:, t] = idx
if (idx == eos_token).any():
finished = True
break
return new_tokens[:, : t + 1 if finished else t, :]
def _dummy_hidden_states(
self,
input_ids: torch.Tensor | None,
positions: torch.Tensor | None,
inputs_embeds: torch.Tensor | None,
) -> torch.Tensor:
"""Shape-correct zero tensor for vllm KV cache profiling.
vllm's gpu_model_runner._dummy_run takes forward()'s return value as
``hidden_states`` and does ``hidden_states[logit_indices_device]``;
returning None on the dummy path crashes with
``TypeError: 'NoneType' object is not subscriptable``.
"""
for ref in (input_ids, positions, inputs_embeds):
if isinstance(ref, torch.Tensor):
num_tokens = int(ref.shape[0]) if ref.ndim >= 1 else 1
device = ref.device
break
else:
num_tokens = 1
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
hidden_size = int(getattr(self, "_hidden_size", 768) or 768)
return torch.zeros((num_tokens, hidden_size), device=device, dtype=torch.bfloat16)
def forward(
self,
input_ids=None,
positions=None,
intermediate_tensors=None,
inputs_embeds=None,
additional_information=None,
**kwargs,
):
if additional_information is None:
additional_information = {}
tts_token_ids = additional_information.get("tts_token_ids")
tts_hidden_states = additional_information.get("tts_hidden_states")
tts_text = additional_information.get("llm_output_text", [""])
if isinstance(tts_text, list):
tts_text = tts_text[0] if tts_text else ""
if tts_token_ids is None or tts_hidden_states is None:
# KV cache profiling / dummy run path — no real TTS input yet.
logger.debug("4.5 Talker: dummy forward (missing tts_token_ids/tts_hidden_states)")
return self._dummy_hidden_states(input_ids, positions, inputs_embeds)
logger.info("4.5 Talker: generating speech for %d tokens", tts_token_ids.shape[0])
waveform = self.generate_speech(tts_token_ids, tts_hidden_states)
# Tuple layout: (mel_spec, waveform). 4.5 talker emits only waveform,
# so mel_spec stays None; the wrapper unpacks in this order and
# packages the waveform into ``multimodal_outputs["model_outputs"]``.
if waveform is not None:
return None, torch.tensor(waveform, dtype=torch.float32)
return None, None
def compute_logits(self, hidden_states, *args, **kwargs):
return torch.zeros(1, 2, device=hidden_states.device if isinstance(hidden_states, torch.Tensor) else "cuda")
def sample(self, logits, sampling_metadata):
return None
def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]):
loaded = set()
tts_weights = {}
for k, v in weights:
if k.startswith("tts."):
tts_weights[k.replace("tts.", "", 1)] = v
# vllm sanity-checks `loaded` against `named_parameters()`.
# The submodule is attached at `self.tts_obj`, not `self.tts`,
# so report the loaded name under the on-module path.
loaded.add(k.replace("tts.", "tts_obj.", 1))
if tts_weights and self._tts_config is not None:
self._lazy_init_tts()
if hasattr(self, "tts_obj") and self.tts_obj is not None:
missing, unexpected = self.tts_obj.load_state_dict(tts_weights, strict=False)
if missing:
logger.warning("TTS missing keys (%d): %s", len(missing), missing[:5])
if unexpected:
logger.warning("TTS unexpected keys (%d): %s", len(unexpected), unexpected[:5])
self.tts_obj = self.tts_obj.to("cuda")
self.emb_text = self.tts_obj.emb_text
self.projector_semantic = self.tts_obj.projector_semantic
logger.info("Loaded %d TTS weights, moved to cuda", len(tts_weights))
return loaded
def get_input_embeddings(self, input_ids, multimodal_embeddings=None, **kwargs):
if hasattr(self, "emb_text") and self.emb_text is not None:
return self.emb_text(input_ids)
return torch.zeros(input_ids.shape[0], 1)
def embed_input_ids(self, input_ids, **kwargs):
return self.get_input_embeddings(input_ids, **kwargs)
@@ -0,0 +1,77 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""MiniCPM-o 4.5 pipeline topology (frozen).
Stage 0: Thinker — multimodal understanding + text generation.
Stage 1: Talker — MiniCPMTTS + Token2Wav, emits the final audio waveform.
The thinker -> talker bridge passes the hidden states + token ids extracted
from the thinker output through ``minicpmo_4_5_omni.llm2tts``; the talker
runs MiniCPMTTS and the on-device Token2wav vocoder in the same process and
returns the waveform directly as the pipeline's final audio output.
"""
from vllm_omni.config.stage_config import (
PipelineConfig,
StageExecutionType,
StagePipelineConfig,
)
_PROC = "vllm_omni.model_executor.stage_input_processors.minicpmo_4_5_omni"
MINICPMO_4_5_PIPELINE = PipelineConfig(
model_type="minicpmo_4_5",
model_arch="MiniCPMO45OmniForConditionalGeneration",
# MiniCPM-o 4.5's HF config.json reports `model_type="minicpmo"` and
# `architectures=["MiniCPMO"]` — both shared verbatim with older MiniCPM-o
# 1.0 / 2.6 checkpoints. The only field distinguishing the generations is
# the top-level ``version`` string, so we register both the shared
# ``MiniCPMO`` arch (for auto-detection) and the 4.5-specific arch (for
# repos that opt into the explicit name later), then pin the routing to
# 4.5 via ``hf_config_predicate``. Without the predicate, loading a 2.6
# checkpoint would also intersect ``["MiniCPMO"]`` here and get routed
# into the 4.5 pipeline, which would then fail at load time.
hf_architectures=("MiniCPMO", "MiniCPMO45OmniForConditionalGeneration"),
hf_config_predicate=lambda c: str(getattr(c, "version", "")) == "4.5",
stages=(
StagePipelineConfig(
stage_id=0,
model_stage="llm",
execution_type=StageExecutionType.LLM_AR,
input_sources=(),
final_output=True,
final_output_type="text",
owns_tokenizer=True,
requires_multimodal_data=True,
engine_output_type="latent",
sampling_constraints={"detokenize": True},
),
StagePipelineConfig(
stage_id=1,
model_stage="tts",
# Stage 1 shares the top-level wrapper class
# (``MiniCPMO45OmniForConditionalGeneration``) inherited from
# ``model_arch`` above. The wrapper dispatches on ``model_stage``
# and, for ``"tts"``, instantiates the standalone TTS submodule
# (``MiniCPMO45OmniTTSForConditionalGeneration``) internally.
# Routing through the wrapper is required so that the runner-side
# ``runtime_additional_information`` payload reaches the talker
# (the standalone TTS class only reads ``additional_information``,
# so wiring stage 1 directly to it would always trigger the dummy
# path) and so the resulting waveform is packaged as
# ``OmniOutput.multimodal_outputs["model_outputs"]`` instead of
# being returned as a bare tuple that the AR runner would mistake
# for hidden states. ``hf_config_name="tts_config"`` keeps KV
# cache / mrope sizing scoped to the talker sub-config.
execution_type=StageExecutionType.LLM_AR,
input_sources=(0,),
final_output=True,
final_output_type="audio",
hf_config_name="tts_config",
engine_output_type="audio",
custom_process_input_func=f"{_PROC}.llm2tts",
sampling_constraints={"detokenize": False},
),
),
)
@@ -250,6 +250,22 @@ _OMNI_MODELS = {
"ming_flash_omni",
"MingFlashOmniForConditionalGeneration",
),
# MiniCPM-o 4.5 Omni models
"MiniCPMO45OmniForConditionalGeneration": (
"minicpmo_4_5",
"minicpmo_4_5_omni",
"MiniCPMO45OmniForConditionalGeneration",
),
"MiniCPMO45OmniLLMForConditionalGeneration": (
"minicpmo_4_5",
"minicpmo_4_5_omni_llm",
"MiniCPMO45OmniLLMForConditionalGeneration",
),
"MiniCPMO45OmniTTSForConditionalGeneration": (
"minicpmo_4_5",
"minicpmo_4_5_omni_tts",
"MiniCPMO45OmniTTSForConditionalGeneration",
),
}
@@ -0,0 +1,136 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Stage input processor for MiniCPM-o 4.5: Thinker (LLM) -> Talker (TTS).
This is the original vLLM-Omni bridge: it converts the thinker stage's
hidden states + token ids into the talker stage's prompt payload. The
talker model itself is adapted from openbmb/MiniCPM-o-4_5 (see the headers
on vllm_omni/model_executor/models/minicpmo_4_5/*.py).
"""
from typing import Any
import torch
from vllm.inputs import TextPrompt
from vllm_omni.inputs.data import OmniTokensPrompt
def llm2tts(
source_outputs: list[Any],
prompt: OmniTokensPrompt | TextPrompt | dict | list | None = None,
requires_multimodal_data: bool = False,
streaming_context: Any | None = None,
):
"""Convert thinker stage output to talker stage input for MiniCPMO Omni.
The signature matches the framework's ``custom_process_input_func`` call
convention used by ``StageEngineCoreClientBase.process_engine_inputs``:
(source_outputs, prompt, requires_multimodal_data, streaming_context)
``source_outputs`` is the already-resolved list of upstream engine
outputs (one entry per request), so we do not need to look anything up
via ``stage_list[source_stage_id].engine_outputs``.
Extracts from thinker output:
- Full hidden states (prompt + generated) for speaker embedding extraction
- Prompt token IDs (for finding spk_bos/spk_eos positions)
- Generated token IDs (for decoding TTS text)
The talker model will:
1. Find <|spk_bos|>/<|spk_eos|> positions in prompt_token_ids
2. Extract speaker embedding from hidden states at those positions
3. Decode generated text and extract TTS content
4. Run ConditionalChatTTS pipeline
"""
del streaming_context # not used by MiniCPM-o 4.5 turn-taking pipeline
if not source_outputs:
raise ValueError("source_outputs cannot be empty")
llm_outputs = source_outputs
tts_inputs = []
if not isinstance(prompt, list):
prompt = [prompt]
multi_modal_data = {
llm_output.request_id: p.get("multi_modal_data", None) if isinstance(p, dict) else None
for llm_output, p in zip(llm_outputs, prompt)
}
for i, llm_output in enumerate(llm_outputs):
output = llm_output.outputs[0]
prompt_token_ids = llm_output.prompt_token_ids
llm_output_ids = output.token_ids
prompt_token_ids_len = len(prompt_token_ids)
latent = output.multimodal_output.get("latent", None)
if latent is None:
latent = output.hidden_states if hasattr(output, "hidden_states") else None
if latent is None:
raise ValueError("No latent or hidden_states found in thinker output")
thinker_hidden_states = latent.clone().detach()
# Split hidden states: prompt portion has speaker embedding,
# generated portion has the text content
prompt_hidden = thinker_hidden_states[:prompt_token_ids_len].to(torch.float32)
# Extract decoded text from thinker output for TTS text extraction
thinker_text = getattr(output, "text", "") or ""
# Build full token sequence and extract TTS region
full_token_ids = list(prompt_token_ids) + (
list(llm_output_ids) if not isinstance(llm_output_ids, list) else llm_output_ids
)
full_hidden = thinker_hidden_states.to(torch.float32)
# Detect TTS token IDs (4.5: 151703/151704, 2.6: 151691/151692)
tts_bos_id, tts_eos_id = 151691, 151692
for _id in [151703, 151704]:
if _id in full_token_ids:
tts_bos_id, tts_eos_id = 151703, 151704
break
tts_bos_idx = tts_eos_idx = None
for idx_t, tid in enumerate(full_token_ids):
if tid == tts_bos_id:
tts_bos_idx = idx_t + 1
elif tid == tts_eos_id:
tts_eos_idx = idx_t
tts_token_ids_slice = tts_hidden_slice = None
if tts_bos_idx is not None and full_hidden.shape[0] > tts_bos_idx:
end_idx = tts_eos_idx if tts_eos_idx is not None else full_hidden.shape[0]
tts_token_ids_slice = torch.tensor(full_token_ids[tts_bos_idx:end_idx], dtype=torch.long)
tts_hidden_slice = full_hidden[tts_bos_idx:end_idx]
additional_information = {
"prompt_embeds": prompt_hidden,
"prompt_token_ids": list(prompt_token_ids),
"llm_output_token_ids": list(llm_output_ids) if not isinstance(llm_output_ids, list) else llm_output_ids,
"llm_output_text": [thinker_text],
}
if tts_token_ids_slice is not None:
additional_information["tts_token_ids"] = tts_token_ids_slice
if tts_hidden_slice is not None:
additional_information["tts_hidden_states"] = tts_hidden_slice
# Minimal prompt token IDs: the talker's AR framework needs *some* tokens
# to do a single prefill step. We use [BOS, PAD, EOS] as a dummy.
tts_inputs.append(
OmniTokensPrompt(
prompt_token_ids=[1, 0, 2],
additional_information=additional_information,
multi_modal_data=(
multi_modal_data[llm_output.request_id]
if requires_multimodal_data and multi_modal_data.get(llm_output.request_id) is not None
else None
),
mm_processor_kwargs=None,
)
)
return tts_inputs