[BugFix]Add MiniCPM-o 4.5 end-to-end test suite (offline, online serving) (#5237)

Signed-off-by: y_null <y_null@qq.com>
Signed-off-by: amy-why-3459 <wuhaiyan17@huawei.com>
Co-authored-by: amy-why-3459 <wuhaiyan17@huawei.com>
Co-authored-by: Hongsheng Liu <liuhongsheng4@huawei.com>
This commit is contained in:
authored and GitHub committed 2026-07-25 17:00:13 -07:00
1 parent 76e4a78afc
commit b6c4578dac
27 files changed
+1338 -46

No files matched your search

+42
View File
@@ -123,6 +123,48 @@ steps:
- pytest -s -v tests/e2e/offline_inference/test_qwen3_omni.py tests/e2e/online_serving/test_qwen3_omni.py -m 'advanced_model and cuda' --run-level 'advanced_model'
mirror_hardwares: h100_2
- label: "Omni · MiniCPM-o 4.5 Test"
source_file_dependencies:
- tests/e2e/offline_inference/test_minicpmo_4_5.py
- tests/e2e/online_serving/test_minicpmo_4_5.py
- vllm_omni/deploy/minicpmo_4_5.yaml
timeout_in_minutes: 50
commands:
- pytest -s -v tests/e2e/offline_inference/test_minicpmo_4_5.py tests/e2e/online_serving/test_minicpmo_4_5.py -m 'advanced_model and cuda' --run-level 'advanced_model'
agents:
queue: "mithril-h100-pool"
plugins:
- kubernetes:
podSpec:
containers:
- image: 936637512419.dkr.ecr.us-west-2.amazonaws.com/vllm-ci-pull-through-cache/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT
resources:
limits:
nvidia.com/gpu: 2
volumeMounts:
- name: devshm
mountPath: /dev/shm
- name: hf-cache
mountPath: /root/.cache/huggingface
env:
- name: HF_HOME
value: /root/.cache/huggingface
- name: HF_TOKEN
valueFrom:
secretKeyRef:
name: hf-token-secret
key: token
nodeSelector:
node.kubernetes.io/instance-type: gpu-h100-sxm
volumes:
- name: devshm
emptyDir:
medium: Memory
- name: hf-cache
hostPath:
path: /mnt/hf-cache
type: DirectoryOrCreate
- label: "TTS · VoxCPM2 Test"
source_file_dependencies:
- tests/e2e/offline_inference/test_voxcpm2_tts.py
+76
View File
@@ -134,6 +134,82 @@ steps:
"
mirror_hardwares: h100_2
- label: "Omni · MiniCPM-o 4.5 Offline Test"
source_file_dependencies:
- tests/e2e/offline_inference/test_minicpmo_4_5.py
- vllm_omni/deploy/minicpmo_4_5.yaml
commands:
- |
timeout 20m bash -c "
pytest -s -v tests/e2e/offline_inference/test_minicpmo_4_5.py -m 'core_model' --run-level 'core_model'
"
agents:
queue: "mithril-h100-pool"
plugins:
- kubernetes:
podSpec:
containers:
- image: 936637512419.dkr.ecr.us-west-2.amazonaws.com/vllm-ci-pull-through-cache/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT
resources:
limits:
nvidia.com/gpu: 2
volumeMounts:
- name: devshm
mountPath: /dev/shm
- name: hf-cache
mountPath: /root/.cache/huggingface
env:
- name: HF_HOME
value: /root/.cache/huggingface
nodeSelector:
node.kubernetes.io/instance-type: gpu-h100-sxm
volumes:
- name: devshm
emptyDir:
medium: Memory
- name: hf-cache
hostPath:
path: /mnt/hf-cache
type: DirectoryOrCreate
- label: "Omni · MiniCPM-o 4.5 Online Test"
source_file_dependencies:
- tests/e2e/online_serving/test_minicpmo_4_5.py
- vllm_omni/deploy/minicpmo_4_5.yaml
commands:
- |
timeout 20m bash -c "
pytest -s -v tests/e2e/online_serving/test_minicpmo_4_5.py -m 'core_model' --run-level 'core_model'
"
agents:
queue: "mithril-h100-pool"
plugins:
- kubernetes:
podSpec:
containers:
- image: 936637512419.dkr.ecr.us-west-2.amazonaws.com/vllm-ci-pull-through-cache/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT
resources:
limits:
nvidia.com/gpu: 2
volumeMounts:
- name: devshm
mountPath: /dev/shm
- name: hf-cache
mountPath: /root/.cache/huggingface
env:
- name: HF_HOME
value: /root/.cache/huggingface
nodeSelector:
node.kubernetes.io/instance-type: gpu-h100-sxm
volumes:
- name: devshm
emptyDir:
medium: Memory
- name: hf-cache
hostPath:
path: /mnt/hf-cache
type: DirectoryOrCreate
- label: "TTS · VoxCPM2 Test"
source_file_dependencies:
- tests/e2e/online_serving/test_voxcpm2_tts.py
+22
View File
@@ -34,3 +34,25 @@ steps:
EXIT=$$?
buildkite-agent artifact upload "tests/dfx/perf/results/*.json"
exit $$EXIT
- group: ":card_index_dividers: Omni Model Test"
key: nightly-omni-group
depends_on: upload-nightly-pipeline
if: build.env("NIGHTLY") == "1" || build.pull_request.labels includes "nightly-test" || build.pull_request.labels includes "omni-test"
steps:
- label: ":robot: Omni · MiniCPM-o 4.5 Test"
agents:
queue: "ascend-a3"
resource_class: "npu-2"
image: "swr.cn-southwest-2.myhuaweicloud.com/modelfoundry/vllm-omni-ci-npu:${BUILDKITE_COMMIT}-a3"
plugins:
- kubernetes:
podSpecPatch:
imagePullSecrets:
- name: swr-secret
env:
VLLM_WORKER_MULTIPROC_METHOD: spawn
VLLM_ALLOW_LONG_MAX_MODEL_LEN: "1"
HF_TOKEN: "${HF_TOKEN}"
commands:
- pytest -s -v tests/e2e/offline_inference/test_minicpmo_4_5.py tests/e2e/online_serving/test_minicpmo_4_5.py tests/e2e/online_serving/test_minicpmo_4_5_expansion.py -m 'full_model' --run-level 'full_model'
+21
View File
@@ -1,4 +1,25 @@
steps:
- group: ":card_index_dividers: Omni Model Test"
key: ready-omni-group
depends_on: upload-ready-pipeline
steps:
- label: ":robot: Omni · MiniCPM-o 4.5 Test"
agents:
queue: "ascend-a3"
resource_class: "npu-2"
image: "swr.cn-southwest-2.myhuaweicloud.com/modelfoundry/vllm-omni-ci-npu:${BUILDKITE_COMMIT}-a3"
plugins:
- kubernetes:
podSpecPatch:
imagePullSecrets:
- name: swr-secret
env:
VLLM_WORKER_MULTIPROC_METHOD: spawn
VLLM_ALLOW_LONG_MAX_MODEL_LEN: "1"
HF_TOKEN: "${HF_TOKEN}"
commands:
- pytest -s -v tests/e2e/offline_inference/test_minicpmo_4_5.py tests/e2e/online_serving/test_minicpmo_4_5.py -m 'core_model' --run-level 'core_model'
- group: ":card_index_dividers: TTS Model Test"
key: ready-tts-group
depends_on: upload-ready-pipeline
+14
View File
@@ -37,6 +37,9 @@ RUN source /usr/local/Ascend/ascend-toolkit/set_env.sh && \
"pyttsx3>=2.99" \
"opencv-python>=4.12.0.88" \
"imageio[ffmpeg]>=0.6.0" \
"hyperpyyaml>=1.2.0" \
"onnx>=1.16.0" \
"librosa>=0.10.0" \
--index-url http://cache-service.nginx-pypi-cache.svc.cluster.local/pypi/simple \
--trusted-host cache-service.nginx-pypi-cache.svc.cluster.local \
--default-timeout=60 && \
@@ -45,6 +48,17 @@ RUN source /usr/local/Ascend/ascend-toolkit/set_env.sh && \
--trusted-host cache-service.nginx-pypi-cache.svc.cluster.local \
--default-timeout=60
# MiniCPM-o 4.5 Stage-2 vocoder assets: s3tokenizer plus flashcosyvoice, which
# is shipped by step-audio2. --no-deps keeps pip from replacing the Ascend torch
# build and from shadowing onnxruntime-cann with the vanilla onnxruntime wheel;
# the deps these two actually need are installed above.
RUN python3 -m pip install -v --no-deps \
"s3tokenizer>=0.3.0" \
"step-audio2>=1.0.0" \
--index-url http://cache-service.nginx-pypi-cache.svc.cluster.local/pypi/simple \
--trusted-host cache-service.nginx-pypi-cache.svc.cluster.local \
--default-timeout=60
ENV VLLM_WORKER_MULTIPROC_METHOD=spawn
ENTRYPOINT []
+14
View File
@@ -37,6 +37,9 @@ RUN source /usr/local/Ascend/ascend-toolkit/set_env.sh && \
"pyttsx3>=2.99" \
"opencv-python>=4.12.0.88" \
"imageio[ffmpeg]>=0.6.0" \
"hyperpyyaml>=1.2.0" \
"onnx>=1.16.0" \
"librosa>=0.10.0" \
--index-url http://cache-service.nginx-pypi-cache.svc.cluster.local/pypi/simple \
--trusted-host cache-service.nginx-pypi-cache.svc.cluster.local \
--default-timeout=60 && \
@@ -45,6 +48,17 @@ RUN source /usr/local/Ascend/ascend-toolkit/set_env.sh && \
--trusted-host cache-service.nginx-pypi-cache.svc.cluster.local \
--default-timeout=60
# MiniCPM-o 4.5 Stage-2 vocoder assets: s3tokenizer plus flashcosyvoice, which
# is shipped by step-audio2. --no-deps keeps pip from replacing the Ascend torch
# build and from shadowing onnxruntime-cann with the vanilla onnxruntime wheel;
# the deps these two actually need are installed above.
RUN python3 -m pip install -v --no-deps \
"s3tokenizer>=0.3.0" \
"step-audio2>=1.0.0" \
--index-url http://cache-service.nginx-pypi-cache.svc.cluster.local/pypi/simple \
--trusted-host cache-service.nginx-pypi-cache.svc.cluster.local \
--default-timeout=60
ENV VLLM_WORKER_MULTIPROC_METHOD=spawn
ENTRYPOINT []
@@ -6,7 +6,7 @@ Two-stage pipeline: **thinker** (multimodal understanding) → **talker + Token2
## Setup
- Install talker deps: `pip install 'vllm-omni[minicpmo]'` (or `stepaudio2-minicpmo`)
- Install talker deps: `pip install stepaudio2-minicpmo`
- `--trust-remote-code` is always passed by `end2end.py` (required for MiniCPMO)
- See [stage configuration docs](https://docs.vllm.ai/projects/vllm-omni/en/latest/configuration/stage_configs/) for memory tuning
+1 -4
View File
@@ -13,10 +13,7 @@ capability boundary, and validation scope, see
Install vLLM-Omni with the MiniCPM-o talker dependencies:
```bash
pip install 'vllm-omni[minicpmo]'
# From a source checkout:
pip install -e '.[minicpmo]'
pip install stepaudio2-minicpmo
```
The `minicpmo` extra installs `stepaudio2-minicpmo` and its audio dependencies,
+1 -12
View File
@@ -71,6 +71,7 @@ dev = [
"kernels==0.14.1", # for accuracy tests comparison with diffusers output (0.15.0 incompatible with diffusers 0.38.0)
"s3tokenizer==0.3.0", # for tts test
"step-audio2==1.0.0", # Step-Audio2 Token2Wav (flashcosyvoice / hyperpyyaml)
"stepaudio2-minicpmo", # for MiniCPM-o 4.5 talker vocoder (stepaudio2-minicpmo Token2wav)
]
demo = [
@@ -88,18 +89,6 @@ 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",
]
# Optional Blackwell-only fused-bias FP8 GEMM; package is `quack-kernels` (imported
# as `quack`), not the unrelated `quack`. See docs/user_guide/quantization/fp8.md.
quack = [
+2 -3
View File
@@ -256,11 +256,10 @@ vllm serve openbmb/MiniCPM-o-4_5 --omni \
rejects `n_timesteps`). Install via the published extra:
```bash
pip install 'vllm-omni[minicpmo]'
pip install stepaudio2-minicpmo
```
Equivalent direct install: `pip install stepaudio2-minicpmo`. A
missing dep raises `ImportError` at first request with the same
A missing dep raises `ImportError` at first request with the same
install hint instead of silently emitting empty audio.
- **TTS conditioning**: the MiniCPM stage bridge can condition speech from
@@ -137,6 +137,56 @@ def test_resumable_session_terminal_is_not_marked_as_segment_boundary() -> None:
assert output.is_segment_finished is False
def test_running_decode_step_without_inter_stage_payload_does_not_raise() -> None:
"""A decode step that neither stops nor carries an inter-stage payload.
``finished`` is only assigned when the request stops, yet the async-chunk
save condition reads it for every request, so this step used to raise
``UnboundLocalError: cannot access local variable 'finished'``.
"""
session = _make_request()
session.status = RequestStatus.RUNNING
sched = MagicMock()
sched.requests = {session.request_id: session}
sched.perf_metrics = None
sched.structured_output_manager.should_advance.return_value = False
sched._update_request_with_output.return_value = ([42], False)
sched._process_kv_transfer_trigger.return_value = False
sched.chunk_transfer_adapter = MagicMock()
sched.running = [session]
sched.waiting_for_transfer_free = set()
sched.transfer_triggered_requests = set()
sched.active_kv_transfers = set()
sched.pending_stop_after_extraction = set()
sched.connector = None
sched.kv_cache_manager.take_events.return_value = None
sched.finished_req_ids_dict = {}
sched.make_stats.return_value = None
scheduler_output = MagicMock(spec=SchedulerOutput)
scheduler_output.num_scheduled_tokens = {session.request_id: 1}
scheduler_output.scheduled_spec_decode_tokens = {}
scheduler_output.num_invalid_spec_tokens = 0
model_runner_output = MagicMock(spec=ModelRunnerOutput)
model_runner_output.sampled_token_ids = [[42]]
model_runner_output.logprobs = None
model_runner_output.prompt_logprobs_dict = {}
model_runner_output.pooler_output = None
model_runner_output.num_nans_in_logits = None
model_runner_output.kv_connector_output = None
model_runner_output.cudagraph_stats = None
model_runner_output.req_id_to_index = {session.request_id: 0}
model_runner_output.routed_experts = None
model_runner_output.inter_stage_outputs = None
OmniARScheduler.update_from_output(sched, scheduler_output, model_runner_output)
# Nothing to hand downstream: no payload, no segment boundary, not finished.
sched.chunk_transfer_adapter.save_async.assert_not_called()
def test_stage0_streaming_update_discards_outstanding_async_placeholder_token() -> None:
sched = _make_scheduler(stage_id=0)
session = _make_request()
@@ -0,0 +1,80 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Invalid audio format rejection on MiniCPM-o 4.5: ``POST /v1/chat/completions``."""
import os
os.environ.setdefault("VLLM_WORKER_MULTIPROC_METHOD", "spawn")
import pytest
from tests.helpers.mark import hardware_test
from tests.helpers.runtime import OmniServerParams, OpenAIClientHandler, dummy_messages_from_mix_data
from tests.helpers.stage_config import get_deploy_config_path
pytestmark = [pytest.mark.slow, pytest.mark.omni, pytest.mark.full_model]
_MINICPMO_DEPLOY = get_deploy_config_path("minicpmo_4_5_batching.yaml")
_MINICPMO_SERVER = [
pytest.param(
OmniServerParams(
model="openbmb/MiniCPM-o-4_5",
stage_config_path=_MINICPMO_DEPLOY,
use_stage_cli=True,
server_args=[
"--trust-remote-code",
"--no-async-chunk",
],
),
id="minicpmo_4_5",
)
]
def _system_prompt() -> dict[str, object]:
return {
"role": "system",
"content": [
{
"type": "text",
"text": (
"You are MiniCPM-o 4.5, a virtual human capable of perceiving "
"auditory and visual inputs, as well as generating text and speech."
),
}
],
}
def _prompt() -> str:
return "What is the capital of China? Answer in 20 words."
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_server", _MINICPMO_SERVER, indirect=True)
def test_invalid_audio_format_rejected(omni_server: OmniServerParams, openai_client: OpenAIClientHandler) -> None:
"""
Test that invalid audio format is properly rejected.
Deploy Setting: default 2GPU
Input Modal: text + invalid audio format
Output Modal: error
"""
fake_base64 = "AAAA" # not a real audio file
responses = openai_client.send_chat_completions_http_request(
{
"json": {
"model": omni_server.model,
"messages": dummy_messages_from_mix_data(
system_prompt=_system_prompt(),
audio_data_url=f"data:audio/invalid;base64,{fake_base64}",
content_text=_prompt(),
),
"stream": True,
},
},
err_code=400,
err_message="Invalid or unsupported audio file",
)
assert not responses[0].success
@@ -0,0 +1,118 @@
"""
E2E offline tests for MiniCPM-o 4.5 model with multimodal input and audio / text output.
"""
import os
os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
import pytest
from tests.helpers.mark import hardware_test
from tests.helpers.media import generate_synthetic_audio, generate_synthetic_image, generate_synthetic_video
from tests.helpers.stage_config import get_deploy_config_path
models = ["openbmb/MiniCPM-o-4_5"]
_CI_DEPLOY = get_deploy_config_path("minicpmo_4_5_batching.yaml")
test_params = [(model, None, {"deploy_config": _CI_DEPLOY, "trust_remote_code": True}) for model in models]
def get_question(prompt_type: str = "text") -> str:
prompts = {
"text": "What is the capital of China? Answer in 20 words.",
"audio": "Describe the audio briefly.",
"image": "What color are the squares in this image?",
"video": "Describe the video briefly.",
"mix": "Describe what is in the image and audio.",
}
return prompts.get(prompt_type, prompts["text"])
@pytest.mark.core_model
@pytest.mark.advanced_model
@pytest.mark.omni
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_runner", test_params, indirect=True)
def test_text_to_text(omni_runner, omni_runner_handler) -> None:
"""Test processing text, generating text output."""
request_config = {"prompts": get_question("text"), "modalities": ["text"]}
omni_runner_handler.send_omni_request(request_config)
@pytest.mark.full_model
@pytest.mark.omni
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_runner", test_params, indirect=True)
def test_audio_to_text(omni_runner, omni_runner_handler) -> None:
"""Test processing audio, generating text output."""
audio = generate_synthetic_audio(1, 1, 16000)["np_array"]
if len(audio.shape) == 2:
audio = audio.squeeze()
request_config = {"prompts": get_question("audio"), "audios": (audio, 16000), "modalities": ["text"]}
omni_runner_handler.send_omni_request(request_config)
@pytest.mark.full_model
@pytest.mark.omni
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_runner", test_params, indirect=True)
def test_image_to_text(omni_runner, omni_runner_handler) -> None:
"""Test processing image, generating text output."""
image = generate_synthetic_image(16, 16)["np_array"]
request_config = {"prompts": get_question("image"), "images": image, "modalities": ["text"]}
omni_runner_handler.send_omni_request(request_config)
@pytest.mark.full_model
@pytest.mark.omni
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_runner", test_params, indirect=True)
def test_video_to_text(omni_runner, omni_runner_handler) -> None:
"""Test processing video, generating text output."""
video = generate_synthetic_video(24, 24, 20)["np_array"]
request_config = {"prompts": get_question("video"), "videos": video, "modalities": ["text"]}
omni_runner_handler.send_omni_request(request_config)
@pytest.mark.full_model
@pytest.mark.omni
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_runner", test_params, indirect=True)
def test_text_to_audio(omni_runner, omni_runner_handler) -> None:
"""Test processing text, generating audio output through the talker token2wav path."""
request_config = {"prompts": get_question("text"), "modalities": ["audio"]}
omni_runner_handler.send_omni_request(request_config)
@pytest.mark.core_model
@pytest.mark.advanced_model
@pytest.mark.omni
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_runner", test_params, indirect=True)
def test_mix_to_audio(omni_runner, omni_runner_handler) -> None:
"""Test processing mixed modalities (image + audio), generating audio output."""
audio = generate_synthetic_audio(1, 1, 16000)["np_array"]
if len(audio.shape) == 2:
audio = audio.squeeze()
image = generate_synthetic_image(16, 16)["np_array"]
request_config = {
"prompts": get_question("mix"),
"audios": (audio, 16000),
"images": image,
"modalities": ["audio"],
}
omni_runner_handler.send_omni_request(request_config)
@pytest.mark.full_model
@pytest.mark.omni
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_runner", test_params, indirect=True)
def test_video_to_audio(omni_runner, omni_runner_handler) -> None:
"""Test processing video, generating audio output."""
video = generate_synthetic_video(24, 24, 20)["np_array"]
request_config = {"prompts": get_question("video"), "videos": video, "modalities": ["audio"]}
omni_runner_handler.send_omni_request(request_config)
@@ -0,0 +1,233 @@
"""
E2E Online tests for MiniCPM-o 4.5 model with multimodal input and audio / text output.
MiniCPM-o 4.5 has ``async_chunk: false``, ``max_num_seqs: 1`` on both stages,
and the vocoder runs in-process inside the talker stage rather than as a separate
Code2Wav stage.
"""
import os
import pytest
from tests.helpers.mark import hardware_test
from tests.helpers.media import generate_synthetic_audio, generate_synthetic_image, generate_synthetic_video
from tests.helpers.runtime import OmniServerParams, dummy_messages_from_mix_data
from tests.helpers.stage_config import get_deploy_config_path
os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
_MODEL = "openbmb/MiniCPM-o-4_5"
_CI_DEPLOY = get_deploy_config_path("minicpmo_4_5_batching.yaml")
test_params = [
pytest.param(
OmniServerParams(
model=_MODEL,
stage_config_path=_CI_DEPLOY,
use_stage_cli=True,
server_args=[
"--trust-remote-code",
"--no-async-chunk",
],
),
id="default",
)
]
def get_system_prompt():
return {
"role": "system",
"content": [
{
"type": "text",
"text": (
"You are MiniCPM-o 4.5, a virtual human capable of perceiving "
"auditory and visual inputs, as well as generating text and speech."
),
}
],
}
def get_prompt(prompt_type: str = "text_only") -> str:
prompts = {
"text_only": "What is the capital of China? Answer in 20 words.",
"mix": "What is recited in the audio? What is in this image? Describe the video briefly.",
"text_image": "What color are the squares in this image?",
}
return prompts.get(prompt_type, prompts["text_only"])
def get_max_batch_size(size_type="few"):
batch_sizes = {"few": 5, "medium": 100, "large": 256}
return batch_sizes.get(size_type, 5)
@pytest.mark.core_model
@pytest.mark.advanced_model
@pytest.mark.omni
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_server", test_params, indirect=True)
def test_text_to_text_001(omni_server, openai_client) -> None:
"""
Test text-only input generating text output via OpenAI API.
Deploy Setting: default 2GPU
Input Modal: text
Output Modal: text
Input Setting: stream=False
"""
messages = dummy_messages_from_mix_data(system_prompt=get_system_prompt(), content_text=get_prompt())
request_config = {
"model": omni_server.model,
"messages": messages,
"stream": False,
"modalities": ["text"],
"key_words": {"text": ["Beijing"]},
}
openai_client.send_omni_request(request_config, request_num=get_max_batch_size())
@pytest.mark.full_model
@pytest.mark.omni
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_server", test_params, indirect=True)
def test_text_to_audio_001(omni_server, openai_client) -> None:
"""
Test text-only input generating text + audio output via OpenAI API.
This exercises the talker TTS region detection and in-process token2wav vocoder.
Deploy Setting: default 2GPU
Input Modal: text
Output Modal: text + audio
Input Setting: stream=True
"""
messages = dummy_messages_from_mix_data(system_prompt=get_system_prompt(), content_text=get_prompt())
request_config = {
"model": omni_server.model,
"messages": messages,
"stream": True,
"key_words": {"audio": ["Beijing"]},
}
openai_client.send_omni_request(request_config)
@pytest.mark.full_model
@pytest.mark.omni
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_server", test_params, indirect=True)
def test_audio_to_text_audio_001(omni_server, openai_client) -> None:
"""
Test audio input generating text + audio output via OpenAI API.
Deploy Setting: default 2GPU
Input Modal: text + audio
Output Modal: text + audio
Input Setting: stream=True
"""
audio_data_url = f"data:audio/wav;base64,{generate_synthetic_audio(5, 1)['base64']}"
messages = dummy_messages_from_mix_data(
system_prompt=get_system_prompt(),
audio_data_url=audio_data_url,
content_text=get_prompt("mix"),
)
request_config = {
"model": omni_server.model,
"messages": messages,
"stream": True,
}
openai_client.send_omni_request(request_config, request_num=get_max_batch_size())
@pytest.mark.full_model
@pytest.mark.omni
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_server", test_params, indirect=True)
def test_image_to_text_audio_001(omni_server, openai_client) -> None:
"""
Test image input generating text + audio output via OpenAI API.
Deploy Setting: default 2GPU
Input Modal: text + image
Output Modal: text + audio
Input Setting: stream=True
"""
image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(24, 24)['base64']}"
messages = dummy_messages_from_mix_data(
system_prompt=get_system_prompt(),
image_data_url=image_data_url,
content_text=get_prompt("text_image"),
)
request_config = {
"model": omni_server.model,
"messages": messages,
"stream": True,
}
openai_client.send_omni_request(request_config, request_num=get_max_batch_size())
@pytest.mark.full_model
@pytest.mark.omni
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_server", test_params, indirect=True)
def test_video_to_text_audio_001(omni_server, openai_client) -> None:
"""
Test video input generating text + audio output via OpenAI API.
Deploy Setting: default 2GPU
Input Modal: text + video
Output Modal: text + audio
Input Setting: stream=True
"""
video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(24, 24, 20)['base64']}"
messages = dummy_messages_from_mix_data(
system_prompt=get_system_prompt(),
video_data_url=video_data_url,
content_text=get_prompt("mix"),
)
request_config = {
"model": omni_server.model,
"messages": messages,
"stream": True,
}
openai_client.send_omni_request(request_config, request_num=get_max_batch_size())
@pytest.mark.core_model
@pytest.mark.advanced_model
@pytest.mark.omni
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_server", test_params, indirect=True)
def test_mix_to_text_audio_001(omni_server, openai_client) -> None:
"""
Test multi-modal input (text + audio + video + image) generating text + audio output.
Deploy Setting: default 2GPU
Input Modal: text + audio + video + image
Output Modal: text + audio
Input Setting: stream=True
"""
video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(24, 24, 20)['base64']}"
image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(24, 24)['base64']}"
audio_data_url = f"data:audio/wav;base64,{generate_synthetic_audio(5, 1)['base64']}"
messages = dummy_messages_from_mix_data(
system_prompt=get_system_prompt(),
video_data_url=video_data_url,
image_data_url=image_data_url,
audio_data_url=audio_data_url,
content_text=get_prompt("mix"),
)
request_config = {
"model": omni_server.model,
"messages": messages,
"stream": True,
}
openai_client.send_omni_request(request_config, request_num=get_max_batch_size())
@@ -0,0 +1,178 @@
"""
E2E Online expansion tests for MiniCPM-o 4.5 covering modality combinations,
in-process token2wav vocoder behavior, max_num_seqs=1 serial execution,
and edge-case validation.
"""
import os
import pytest
from tests.helpers.mark import hardware_test
from tests.helpers.media import generate_synthetic_video
from tests.helpers.runtime import OmniServerParams, dummy_messages_from_mix_data
from tests.helpers.stage_config import get_deploy_config_path
pytestmark = [pytest.mark.full_model, pytest.mark.omni]
os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
_MODEL = "openbmb/MiniCPM-o-4_5"
_CI_DEPLOY = get_deploy_config_path("minicpmo_4_5_batching.yaml")
test_params = [
pytest.param(
OmniServerParams(
model=_MODEL,
stage_config_path=_CI_DEPLOY,
use_stage_cli=True,
server_args=[
"--trust-remote-code",
],
),
id="default",
)
]
def get_system_prompt():
return {
"role": "system",
"content": [
{
"type": "text",
"text": (
"You are MiniCPM-o 4.5, a virtual human capable of perceiving "
"auditory and visual inputs, as well as generating text and speech."
),
}
],
}
def get_prompt(prompt_type: str = "text_only") -> str:
prompts = {
"text_only": "What is the capital of China? Answer in 20 words.",
"mix": "What is recited in the audio? What is in this image? Describe the video briefly.",
}
return prompts.get(prompt_type, prompts["text_only"])
def get_max_batch_size(size_type="few"):
batch_sizes = {"few": 5, "medium": 100, "large": 256}
return batch_sizes.get(size_type, 5)
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_server", test_params, indirect=True)
def test_text_video_to_text_001(omni_server, openai_client) -> None:
"""
Test text + video input generating text output.
Deploy Setting: default 2GPU
Input Modal: text + video
Output Modal: text
Input Setting: stream=False
"""
video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(24, 24, 20)['base64']}"
messages = dummy_messages_from_mix_data(
system_prompt=get_system_prompt(),
video_data_url=video_data_url,
content_text=get_prompt("mix"),
)
request_config = {
"model": omni_server.model,
"messages": messages,
"stream": False,
"modalities": ["text"],
}
openai_client.send_omni_request(request_config, request_num=get_max_batch_size())
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_server", test_params, indirect=True)
def test_sequential_requests_independent(omni_server, openai_client) -> None:
"""
Verify that sequential requests produce independent results and state
does not leak between requests. MiniCPM-o 4.5 has max_num_seqs=1 so
requests are processed one-at-a-time; the second request must not
receive the first request's audio or token state.
Deploy Setting: default 2GPU
Input Modal: text (two different prompts)
Output Modal: text + audio (both)
"""
messages_1 = dummy_messages_from_mix_data(
system_prompt=get_system_prompt(),
content_text="What is the capital of France?",
)
messages_2 = dummy_messages_from_mix_data(
system_prompt=get_system_prompt(),
content_text="What is the capital of China?",
)
# Send first request and ensure it completes
openai_client.send_omni_request(
{
"model": omni_server.model,
"messages": messages_1,
"stream": True,
},
request_num=1,
)
# Send second request with different prompt; must produce its own output
openai_client.send_omni_request(
{
"model": omni_server.model,
"messages": messages_2,
"stream": True,
"key_words": {"text": ["Beijing"]},
},
request_num=1,
)
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_server", test_params, indirect=True)
def test_text_to_audio_long_output_001(omni_server, openai_client) -> None:
"""
Test text input generating a longer audio output to exercise the
token2wav decoder across multiple frames.
Deploy Setting: default 2GPU
Input Modal: text (longer prompt)
Output Modal: text + audio
Input Setting: stream=True
"""
messages = dummy_messages_from_mix_data(
system_prompt=get_system_prompt(),
content_text="Tell me a short story about a cat in about 50 words.",
)
request_config = {
"model": omni_server.model,
"messages": messages,
"stream": True,
"key_words": {"audio": ["cat"]},
}
openai_client.send_omni_request(request_config, request_num=get_max_batch_size())
@hardware_test(res={"cuda": "H100", "npu": "A2"}, num_cards=2)
@pytest.mark.parametrize("omni_server", test_params, indirect=True)
def test_chinese_text_to_audio(omni_server, openai_client) -> None:
"""
Test Chinese text input generating audio output.
"""
messages = dummy_messages_from_mix_data(
system_prompt=get_system_prompt(),
content_text="北京,中国的首都,是一座融合了长城等历史地点与现代建筑的国际化大都市,充满了独特的文化与活力。请重复这句话。",
)
request_config = {
"model": omni_server.model,
"messages": messages,
"stream": True,
"key_words": {"text": ["北京"]},
}
openai_client.send_omni_request(request_config)
@@ -403,6 +403,90 @@ async def test_declared_modality_not_produced_text_finish_suppressed():
assert ch["finish_reason"] is None
# ---------------------------------------------------------------------------
# Tests: audio chunks that carry no waveform
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_audio_chunk_without_waveform_keeps_stream_alive():
"""An audio message with no PCM must not break the stream.
``_create_audio_choice`` used to return an ErrorResponse here, which the
stream generator then iterated as a list of choices; the pydantic model
yields (field, value) tuples, so the loop raised
``AttributeError: 'tuple' object has no attribute 'finish_reason'``.
"""
serving_chat = _build_serving_chat()
request = _make_request(modalities=["text", "audio"])
empty_audio = _make_audio_omni_output()
empty_audio.request_output.outputs[0].multimodal_output = {"audio": []}
def create_audio_choice(omni_res, role, request, stream=False):
return OmniOpenAIServingChat._create_audio_choice(serving_chat, omni_res, role, request, stream=stream)
serving_chat._create_audio_choice = create_audio_choice
async def result_generator():
yield _make_text_omni_output(text="hi", token_ids=[10], finish_reason=None)
yield _make_text_omni_output(text="!", token_ids=[11], finish_reason="stop")
yield empty_audio
raw_lines = await _collect_stream(
serving_chat.chat_completion_stream_generator(
request=request,
result_generator=result_generator(),
request_id="test-req",
model_name="test-model",
conversation=[],
tokenizer=MagicMock(),
request_metadata=MagicMock(),
)
)
assert not any("Error in chat completion stream generator" in line for line in raw_lines)
chunks = _parse_sse_chunks(raw_lines)
finish_reasons = [ch["finish_reason"] for c in chunks for ch in c.get("choices", [])]
assert finish_reasons.count("stop") == 1, f"Expected 1 stop, got {finish_reasons}"
assert finish_reasons[-1] == "stop"
@pytest.mark.asyncio
async def test_audio_choice_error_response_is_not_iterated_as_choices():
"""Any ErrorResponse from the audio path is skipped, not iterated."""
from vllm.entrypoints.openai.engine.protocol import ErrorResponse
serving_chat = _build_serving_chat()
request = _make_request(modalities=["text", "audio"])
serving_chat._create_audio_choice = MagicMock(
side_effect=lambda omni_res, role, request, stream=False: serving_chat._create_error_response("boom")
)
async def result_generator():
yield _make_text_omni_output(text="hi", token_ids=[10], finish_reason="stop")
yield _make_audio_omni_output()
raw_lines = await _collect_stream(
serving_chat.chat_completion_stream_generator(
request=request,
result_generator=result_generator(),
request_id="test-req",
model_name="test-model",
conversation=[],
tokenizer=MagicMock(),
request_metadata=MagicMock(),
)
)
assert isinstance(serving_chat._create_error_response("boom"), ErrorResponse)
assert not any("AttributeError" in line for line in raw_lines)
chunks = _parse_sse_chunks(raw_lines)
finish_reasons = [ch["finish_reason"] for c in chunks for ch in c.get("choices", [])]
assert finish_reasons.count("stop") == 1, f"Expected 1 stop, got {finish_reasons}"
# ---------------------------------------------------------------------------
# Tests: voice/speaker parameter compatibility
# ---------------------------------------------------------------------------
+19 -6
View File
@@ -2424,6 +2424,10 @@ class OmniRunner:
video_padding_token = "<|VIDEO|>"
image_padding_token = "<|IMAGE|>"
audio_padding_token = "<|AUDIO|>"
# Default wrapping for Qwen-style models (bos/eos around the placeholder).
audio_fmt = "<|audio_bos|>{p}<|audio_eos|>"
image_fmt = "<|vision_bos|>{p}<|vision_eos|>"
video_fmt = "<|vision_bos|>{p}<|vision_eos|>"
if "Qwen3-Omni-30B-A3B-Instruct" in self.model_name:
video_padding_token = "<|video_pad|>"
image_padding_token = "<|image_pad|>"
@@ -2432,6 +2436,15 @@ class OmniRunner:
video_padding_token = "<VIDEO>"
image_padding_token = "<IMAGE>"
audio_padding_token = "<AUDIO>"
elif "MiniCPM" in self.model_name:
# MiniCPM-o expects the bare placeholder literals (with parens and ./),
# not Qwen-style bos/eos wrapping.
video_padding_token = "(<video>./</video>)"
image_padding_token = "(<image>./</image>)"
audio_padding_token = "(<audio>./</audio>)"
audio_fmt = "{p}"
image_fmt = "{p}"
video_fmt = "{p}"
if isinstance(prompts, str):
prompts = [prompts]
@@ -2492,28 +2505,28 @@ class OmniRunner:
if audio is not None:
if isinstance(audio, list):
for _ in audio:
user_content += f"<|audio_bos|>{audio_padding_token}<|audio_eos|>"
user_content += audio_fmt.format(p=audio_padding_token)
multi_modal_data["audio"] = audio
else:
user_content += f"<|audio_bos|>{audio_padding_token}<|audio_eos|>"
user_content += audio_fmt.format(p=audio_padding_token)
multi_modal_data["audio"] = audio
image = images_list[i]
if image is not None:
if isinstance(image, list):
for _ in image:
user_content += f"<|vision_bos|>{image_padding_token}<|vision_eos|>"
user_content += image_fmt.format(p=image_padding_token)
multi_modal_data["image"] = image
else:
user_content += f"<|vision_bos|>{image_padding_token}<|vision_eos|>"
user_content += image_fmt.format(p=image_padding_token)
multi_modal_data["image"] = image
video = videos_list[i]
if video is not None:
if isinstance(video, list):
for _ in video:
user_content += f"<|vision_bos|>{video_padding_token}<|vision_eos|>"
user_content += video_fmt.format(p=video_padding_token)
multi_modal_data["video"] = video
else:
user_content += f"<|vision_bos|>{video_padding_token}<|vision_eos|>"
user_content += video_fmt.format(p=video_padding_token)
multi_modal_data["video"] = video
user_content += prompt_text
@@ -315,6 +315,86 @@ def test_empty_final_ignores_generation_scheduler_placeholder_token():
assert model._states == {}
@pytest.mark.parametrize(
"info",
[
# The runner injects the engine request id on every step (GPU
# _preprocess, NPU _gather_runtime_additional_information)...
{"request_id": "a", "meta": {"request_id": "a"}},
# ...but a pre-warm step can also reach the model with nothing at all.
{},
],
)
def test_prewarm_placeholder_step_emits_silence_without_touching_state(info):
# async-chunk pre-warm submits Stage 2 with a reserved placeholder prompt.
# If it gets scheduled before the first codec window lands, those reserved
# tokens must neither be vocoded nor held to the codec payload contract.
model, token2wav = _model()
output = _forward(model, [info], request_ids=["a"])
assert output.multimodal_outputs["model_outputs"][0].numel() == 0
assert model._states == {}
assert token2wav.hift.calls == []
def test_metadata_only_payload_still_decodes_codec_from_prompt_tokens():
# The connector strips 1-D codec tensors out of additional_information and
# leaves them in the prompt tokens, so a real chunk reaches the model as
# producer metadata plus input ids. It must still be vocoded.
model, _ = _model()
info = {
"request_id": "a",
"meta": {
"request_id": "a",
"chunk_seq": 0,
"code_flat_numel": 2,
"prompt_cache_id": "shared",
},
}
output = _forward(model, [info], placeholder_counts=[2])
assert output.multimodal_outputs["model_outputs"][0].numel() > 0
assert set(model._states) == {"a"}
def test_non_final_chunk_shorter_than_lookahead_window_is_rejected():
token2wav = _FakeToken2Wav()
token2wav.flow.encoder.pre_lookahead_layer = SimpleNamespace(pre_lookahead_len=3)
adapter = BatchedToken2Wav(token2wav)
prompt = adapter.prepare_prompt("shared", "/fake/prompt.wav")
states = adapter.setup_batch(prompt, 1)
with pytest.raises(RuntimeError, match="chunk_below_lookahead_window"):
adapter.decode_batch(torch.tensor([[10]]), prompt, states, last_chunk=False)
# The final chunk is zero-padded by the encoder, so it stays decodable.
audios, _ = adapter.decode_batch(torch.tensor([[10]]), prompt, states, last_chunk=True)
assert len(audios) == 1
def test_forward_builds_backend_when_weight_loading_was_skipped(monkeypatch):
# load_format=dummy never calls load_weights(), so Stage 2 would otherwise
# reach its first request with no Token2wav assets at all.
model = MiniCPMO45Code2Wav(vllm_config=_config())
token2wav = _FakeToken2Wav()
builds = 0
def build_backend():
nonlocal builds
builds += 1
model.backend = BatchedToken2Wav(token2wav)
monkeypatch.setattr(model, "_build_backend", build_backend)
output = _forward(model, [_info("a", 0, [10, 11])])
_forward(model, [_info("a", 1, [12, 13])])
assert builds == 1
assert output.multimodal_outputs["model_outputs"][0].numel() > 0
@pytest.mark.parametrize(
("info", "reason"),
[
@@ -236,6 +236,38 @@ def test_missing_conditioning_fails_clearly() -> None:
)
def test_empty_speech_segment_finishes_without_sampling_codes() -> None:
talker = _make_talker()
talker.emb_text = nn.Embedding(8, 4)
talker.emb_code = nn.ModuleList([nn.Embedding(8, 4)])
talker._text_eos_id = 5
talker._tts_bos_id = 6
_, embeds, updates = talker.preprocess(
torch.zeros(2, dtype=torch.long),
None,
_omni_is_prefill=True,
request_id="req-empty",
tts_token_ids=torch.empty(0, dtype=torch.long),
tts_hidden_states=torch.empty(0, 4),
)
assert torch.equal(embeds, talker.emb_text(torch.tensor([5, 6])))
assert updates["audio_state"]["finished"] is True
# Stage 1's sampling min_tokens keeps scheduling decode steps until the stop
# token becomes eligible, and those steps have no previous code to embed.
_, decode_embeds, _ = talker.preprocess(
torch.zeros(1, dtype=torch.long),
None,
request_id="req-empty",
audio_state=updates["audio_state"],
audio_codes=updates["audio_codes"],
)
assert decode_embeds.shape == (1, 4)
def test_request_cleanup_evicts_ar_rng_and_decode_state() -> None:
talker = _make_talker()
talker._request_generators["req-done"] = torch.Generator()
@@ -384,6 +384,7 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
stopped = False
is_segment_finished = False
finished = False
new_logprobs = None
new_token_ids = generated_token_ids
pooler_output = pooler_outputs[req_index] if pooler_outputs else None
+28 -2
View File
@@ -2009,6 +2009,13 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin):
role = self.get_chat_request_role(request)
choices_data = self._create_audio_choice(omni_res, role, request, stream=True)
if isinstance(choices_data, ErrorResponse):
logger.error(
"Skipping audio chunk for request %s: %s",
request_id,
choices_data.error.message,
)
continue
# Only emit finish_reason on the last modality to
# comply with OpenAI streaming spec.
for choice in choices_data:
@@ -2281,6 +2288,8 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin):
]
elif omni_outputs.final_output_type == "audio":
choices_data = self._create_audio_choice(omni_outputs, role, request, stream=False)
if isinstance(choices_data, ErrorResponse):
return choices_data
elif omni_outputs.final_output_type == "image":
choices_data = self._create_image_choice(omni_outputs, role, request, stream=False)
else:
@@ -2608,7 +2617,7 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin):
def _create_audio_choice(
self, omni_outputs: OmniRequestOutput, role: str, request: ChatCompletionRequest, stream: bool = False
):
) -> list[ChatCompletionResponseChoice] | list[ChatCompletionResponseStreamChoice] | ErrorResponse:
choices: list[ChatCompletionResponseChoice] = []
final_res = omni_outputs.request_output
# OMNI: Access multimodal_output from CompletionOutput (outputs[0]), not from RequestOutput
@@ -2628,7 +2637,24 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin):
else:
audio_tensor = audio_data
if audio_tensor is None:
return self._create_error_response("Audio generation completed but no audio was produced.")
if not stream:
return self._create_error_response("Audio generation completed but no audio was produced.")
# A streamed message can legitimately carry no waveform: the talker
# may emit an empty speech segment, and Stage 2 then reports the
# chunk with metadata only. Emit an empty audio delta so the
# caller's finish_reason bookkeeping still runs and the stream
# terminates normally instead of raising mid-generation.
return [
ChatCompletionResponseStreamChoice(
index=output.index,
delta=DeltaMessage(role=role, content=""),
logprobs=None,
finish_reason=output.finish_reason,
stop_reason=output.stop_reason,
token_ids=(as_list(output.token_ids) if request.return_token_ids else None),
)
for output in final_res.outputs
]
audio_tensor = audio_tensor.detach().cpu().float().numpy()
# Ensure audio is 1D (flatten if needed)
@@ -15,6 +15,20 @@ import torch.nn.functional as F
_SILENCE_TOKEN = 4218
def _autocast_disabled(device: torch.device):
"""Disable any enclosing autocast region on ``device``.
``torch.amp.autocast`` resolves the autocast dtype for ``device_type``
while constructing the context, which raises on accelerators (e.g. Ascend
NPU) that never registered autocast support. Degrade to a no-op there: an
enclosing region can only exist on a device type torch already knows.
"""
try:
return torch.amp.autocast(device.type, enabled=False)
except (RuntimeError, TypeError, ValueError):
return nullcontext()
def tensor_signature(value: torch.Tensor) -> tuple[tuple[int, ...], str, str]:
return tuple(value.shape), str(value.dtype), value.device.type
@@ -72,7 +86,7 @@ class BatchedToken2Wav(nn.Module):
previous_dtype = torch.get_default_dtype()
try:
torch.set_default_dtype(torch.float32)
with torch.amp.autocast("cuda", enabled=False):
with _autocast_disabled(self.speech_window.device):
values = self._token2wav._prepare_prompt(prompt_wav)
finally:
torch.set_default_dtype(previous_dtype)
@@ -106,6 +120,16 @@ class BatchedToken2Wav(nn.Module):
dtype=torch.float16,
)
def _pre_lookahead_len(self) -> int | None:
"""Right-context width of the encoder's pre-lookahead convolution.
``None`` when the encoder does not expose one, so callers keep working
against encoder implementations without that layer.
"""
layer = getattr(self.flow.encoder, "pre_lookahead_layer", None)
width = getattr(layer, "pre_lookahead_len", None)
return int(width) if width is not None else None
def _encode_chunk(
self,
tokens: torch.Tensor,
@@ -278,7 +302,11 @@ class BatchedToken2Wav(nn.Module):
batch_size: int,
) -> list[BatchedToken2WavState]:
prompt_tokens, speakers, prompt_mels = self._repeat_prompt(features, batch_size)
lookahead = prompt_tokens.new_full((batch_size, 3), _SILENCE_TOKEN)
lookahead_width = self._pre_lookahead_len()
lookahead = prompt_tokens.new_full(
(batch_size, 3 if lookahead_width is None else lookahead_width),
_SILENCE_TOKEN,
)
with self._autocast(prompt_tokens.device):
hidden, conformer_cnn, conformer_att = self._encode_chunk(
torch.cat((prompt_tokens, lookahead), dim=1),
@@ -336,6 +364,19 @@ class BatchedToken2Wav(nn.Module):
batch_size = int(tokens.shape[0])
if batch_size != len(states):
raise ValueError(f"tokens batch {batch_size} != state batch {len(states)}")
# The encoder's pre-lookahead convolution consumes ``pre_lookahead_len``
# frames of right context and keeps no left cache, so a non-final chunk
# must carry at least one full kernel. Only the final chunk is allowed
# to be shorter: ``forward_chunk`` zero-pads it by the lookahead width.
lookahead = self._pre_lookahead_len()
if lookahead is not None and not last_chunk:
num_frames = int(tokens.shape[1])
if num_frames <= lookahead:
raise RuntimeError(
"MiniCPMO45Code2WavBatchError "
f'{{"reason":"chunk_below_lookahead_window","frames":{num_frames},'
f'"minimum":{lookahead + 1}}}'
)
flow_cache = self._stack_flow_cache(states)
speakers = features.speaker_embedding.expand(batch_size, -1)
with self._autocast(tokens.device):
@@ -17,6 +17,7 @@ import soundfile as sf
import torch
import torch.nn as nn
from vllm.config import VllmConfig
from vllm.logger import init_logger
from vllm_omni.model_executor.models.output_templates import OmniOutput
@@ -26,6 +27,8 @@ from .batched_token2wav import (
state_shape_signature,
)
logger = init_logger(__name__)
def _batch_error(reason: str, **details: Any) -> RuntimeError:
payload = {"reason": reason, **details}
@@ -48,6 +51,23 @@ def _codec_tensor(value: Any, fallback: torch.Tensor) -> torch.Tensor:
return fallback.reshape(-1).to(dtype=torch.long)
# Keys the runner stamps on every step regardless of stage input (see
# OmniGPUModelRunner._preprocess and the NPU _gather_runtime_additional_information
# override). A step carrying only these has no producer payload at all.
_RUNNER_STAMPED_KEYS = frozenset({"request_id", "req_id", "generated_len", "meta"})
def _carries_stage_payload(info: Mapping[str, Any], meta: Mapping[str, Any]) -> bool:
"""Whether this step carries anything the Talker stage actually sent.
Any real async-chunk payload brings producer metadata along, whether the
transport delivers it nested under ``meta`` or as flattened ``meta.*`` keys.
"""
if any(key not in _RUNNER_STAMPED_KEYS for key in info):
return True
return meta is not info and any(key not in _RUNNER_STAMPED_KEYS for key in meta)
@dataclass(frozen=True)
class _RequestState:
cache_epoch: int
@@ -74,6 +94,7 @@ class _WorkItem:
duplex_epoch: int | None
segment_end: bool
turn_end: bool
has_payload: bool = True
class MiniCPMO45Code2Wav(nn.Module):
@@ -207,6 +228,32 @@ class MiniCPMO45Code2Wav(nn.Module):
if not isinstance(meta, Mapping):
meta = info
request_id = str(_scalar(meta.get("request_id"), _scalar(info.get("request_id"), "")))
if not _carries_stage_payload(info, meta):
# The producer attached nothing to this step, only the bookkeeping
# the runner stamps on every request. The stage was scheduled on
# the placeholder prompt that async-chunk pre-warm submits before
# the first codec window arrives: those tokens are reserved slots,
# not codec data, and one bogus frame is shorter than the vocoder's
# lookahead window. Such a step carries no producer metadata at
# all, so it cannot be held to the payload contract below either.
return _WorkItem(
output_index=index,
state_id=state_id,
request_id=request_id or state_id,
cache_epoch=0,
chunk_seq=0,
prompt_cache_id=self._default_prompt_id,
prompt_wav=self._default_prompt_wav,
last_chunk=False,
tokens=segment.new_empty(0, dtype=torch.long),
previous=None,
segment_text="",
duplex_turn_id=None,
duplex_epoch=None,
segment_end=False,
turn_end=False,
has_payload=False,
)
if not request_id:
raise _batch_error("missing_request_id", output_index=index)
cache_epoch = int(_scalar(meta.get("cache_epoch"), 0))
@@ -358,7 +405,18 @@ class MiniCPMO45Code2Wav(nn.Module):
runtime_infos=len(runtime_additional_information),
)
if self.backend is None:
raise _batch_error("backend_not_loaded")
# load_format=dummy (CI core_model runs) skips model.load_weights()
# entirely, but Token2wav's assets live beside the checkpoint rather
# than in its weight iterator, so they still have to be loaded for
# this stage to produce anything. Build them on first use, outside
# inference mode so the parameters are ordinary tensors.
logger.warning_once(
"MiniCPM-o Code2Wav backend was not built during weight loading "
"(load_format=%s); loading Token2wav assets now.",
getattr(getattr(self.vllm_config, "load_config", None), "load_format", "unknown"),
)
with torch.inference_mode(False), torch.no_grad():
self._build_backend()
state_ids = kwargs.get("request_ids")
if state_ids is None:
@@ -393,7 +451,9 @@ class MiniCPMO45Code2Wav(nn.Module):
outputs = [empty for _ in segments]
sentinels = [item for item in items if item.last_chunk and item.tokens.numel() == 0]
compute_items = [item for item in items if item.tokens.numel() > 0]
invalid_empty = [item.request_id for item in items if not item.last_chunk and item.tokens.numel() == 0]
invalid_empty = [
item.request_id for item in items if item.has_payload and not item.last_chunk and item.tokens.numel() == 0
]
if invalid_empty:
raise _batch_error("empty_nonfinal_chunk", request_ids=invalid_empty)
@@ -514,9 +574,30 @@ class MiniCPMO45Code2Wav(nn.Module):
def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
for _ in weights:
pass
self._build_backend()
# Token2wav loads flow.pt and hift.pt inside its constructor instead of
# from the parent MiniCPM checkpoint iterator. Report those registered
# parameters as initialized so vLLM's strict loader audit does not
# misclassify the independently loaded Stage-2 weights as missing.
return {name for name, _ in self.named_parameters()}
def _build_backend(self) -> None:
"""Load the Token2wav assets that back this stage."""
if self.backend is not None:
return {name for name, _ in self.named_parameters()}
from stepaudio2.token2wav import Token2wav
return
from vllm_omni.platforms import current_omni_platform
if current_omni_platform.is_npu():
# NPU/Ascend: the external `stepaudio2` package hard-codes `.cuda()`,
# so use the in-tree NPU-aware adapter instead. It delegates to
# StepAudio2Token2WavCore, which auto-applies the Ascend fixes
# (HiFT linear downsample, DiT mask expand, MATH SDPA) on NPU.
from vllm_omni.model_executor.models.minicpmo_4_5.minicpmo_4_5_token2wav import (
MiniCPMO45Token2wav as Token2wav,
)
else:
from stepaudio2.token2wav import Token2wav
extra = self._extra_config()
prompt_path = Path(self._default_prompt_wav)
@@ -540,8 +621,3 @@ class MiniCPMO45Code2Wav(nn.Module):
finally:
torch.set_default_dtype(previous_dtype)
self.backend = BatchedToken2Wav(token2wav)
# Token2wav loads flow.pt and hift.pt inside its constructor instead of
# from the parent MiniCPM checkpoint iterator. Report those registered
# parameters as initialized so vLLM's strict loader audit does not
# misclassify the independently loaded Stage-2 weights as missing.
return {name for name, _ in self.named_parameters()}
@@ -19,6 +19,7 @@ import torch.nn as nn
import torch.nn.functional as F
from transformers import LlamaConfig
from vllm.config import VllmConfig
from vllm.logger import init_logger
from vllm.model_executor.models.interfaces import SupportsPP
from vllm.model_executor.models.llama import LlamaModel
from vllm.model_executor.models.utils import maybe_prefix
@@ -28,6 +29,8 @@ from vllm_omni.experimental.fullduplex.engine.intermediate import get_tts_handof
from vllm_omni.model_executor.models.output_templates import OmniOutput
from vllm_omni.platforms import current_omni_platform
logger = init_logger(__name__)
_REPETITION_WINDOW = 16
_MIN_AUDIO_TOKENS = 64
_MAX_AUDIO_TOKENS = 2048
@@ -202,11 +205,27 @@ class MiniCPMO45OmniTTSForConditionalGeneration(nn.Module, SupportsPP):
)
self.make_empty_intermediate_tensors = self.tts_model.make_empty_intermediate_tensors
def _boundary_embeddings(self) -> torch.Tensor:
"""Embed the ``<text_eos><audio_bos>`` tail every condition ends with."""
ids = torch.tensor(
[self._text_eos_id, self._tts_bos_id],
device=self.emb_text.weight.device,
dtype=torch.long,
)
return self.emb_text(ids)
def _build_condition_embeddings(
self,
tts_token_ids: torch.Tensor,
tts_hidden_states: torch.Tensor,
) -> torch.Tensor:
if tts_token_ids.numel() == 0 or tts_hidden_states.numel() == 0:
# The thinker can legally emit an empty speech segment (<|tts_bos|>
# immediately followed by a boundary token) when it decides not to
# speak. Condition on the boundary tokens alone, which matches the
# 2-token scheduler prompt the stage bridge builds for an empty
# handoff.
return self._boundary_embeddings()
device = self.emb_text.weight.device
dtype = self.emb_text.weight.dtype
token_ids = tts_token_ids.to(device=device, dtype=torch.long).reshape(-1)
@@ -220,9 +239,7 @@ class MiniCPMO45OmniTTSForConditionalGeneration(nn.Module, SupportsPP):
hidden_embeds = self.projector_semantic(hidden)
if self._normalize:
hidden_embeds = F.normalize(hidden_embeds, p=2, dim=-1)
text_eos = self.emb_text(torch.tensor([self._text_eos_id], device=device, dtype=torch.long))
audio_bos = self.emb_text(torch.tensor([self._tts_bos_id], device=device, dtype=torch.long))
return torch.cat([text_embeds + hidden_embeds, text_eos, audio_bos], dim=0)
return torch.cat([text_embeds + hidden_embeds, self._boundary_embeddings()], dim=0)
def preprocess(
self,
@@ -255,8 +272,15 @@ class MiniCPMO45OmniTTSForConditionalGeneration(nn.Module, SupportsPP):
f"hidden_states={type(hidden_states).__name__}, "
f"available_keys={available}"
)
if token_ids.numel() == 0 or hidden_states.numel() == 0:
raise ValueError("MiniCPM-o Talker conditioning must not be empty")
# An empty condition means the thinker chose not to speak: finish the
# request up front so it emits zero audio codes instead of killing
# the stage engine.
empty_condition = token_ids.numel() == 0 or hidden_states.numel() == 0
if empty_condition:
logger.warning_once(
"MiniCPM-o Talker received an empty condition (request %s); this request produces no audio.",
info_dict.get("request_id"),
)
full_embeds = self._build_condition_embeddings(token_ids, hidden_states)
offset = int(info_dict.get("_omni_num_computed_tokens", 0))
if offset == 0 and span_len > full_embeds.shape[0]:
@@ -288,7 +312,7 @@ class MiniCPMO45OmniTTSForConditionalGeneration(nn.Module, SupportsPP):
state = {
"step": 0,
"max_tokens": max_tokens,
"finished": False,
"finished": empty_condition,
}
request_id = str(info_dict.get("request_id", "0"))
request_states = getattr(self, "_request_audio_states", None)
@@ -311,6 +335,13 @@ class MiniCPMO45OmniTTSForConditionalGeneration(nn.Module, SupportsPP):
current = (info_dict.get("audio_codes", {}) or {}).get("current")
if not isinstance(current, torch.Tensor) or current.numel() != 1:
if state.get("finished"):
# A request that finished before sampling any code can still be
# scheduled for decode steps while sampling min_tokens masks the
# stop token. make_omni_output ignores its hidden states, so any
# shape-correct embedding will do.
weight = self.emb_code[0].weight
return input_ids, weight.new_zeros((span_len, weight.shape[1])), {}
raise RuntimeError("MiniCPM-o Talker decode is missing the previous request-local audio code")
code = current.to(device=self.emb_code[0].weight.device, dtype=torch.long).reshape(1)
embeds = self.emb_code[0](code)
@@ -82,6 +82,44 @@ class MiniCPMO45Token2wav:
self.stream_cache: Any | None = None
self.hift_cache_dict: dict[str, torch.Tensor] = {}
# The external ``stepaudio2.Token2wav`` class builds ``speech_window``
# eagerly in ``__init__``. ``StepAudio2Token2WavCore`` only creates it
# lazily inside ``setup_stream_for``, so materialize it here: the
# 3-stage ``BatchedToken2Wav`` wrapper clones it at construction time.
if self._core.speech_window is None:
self._core.speech_window = torch.from_numpy(np.hamming(2 * self._core.source_cache_len)).to(
device=self.device, dtype=torch.float32
)
# --- Surface expected by ``BatchedToken2Wav`` (3-stage Code2Wav) ---------
# ``BatchedToken2Wav`` wraps this object as a one-time asset loader and
# module holder; it never calls ``__call__`` / ``stream`` / ``set_stream_cache``.
# Expose the same attributes the upstream ``stepaudio2.Token2wav`` provides.
@property
def flow(self) -> torch.nn.Module:
return self._core.flow
@property
def hift(self) -> torch.nn.Module:
return self._core.hift
@property
def mel_cache_len(self) -> int:
return self._core.mel_cache_len
@property
def source_cache_len(self) -> int:
return self._core.source_cache_len
@property
def speech_window(self) -> torch.Tensor:
return self._core.speech_window
def _prepare_prompt(self, prompt_wav: str):
"""Delegate prompt feature extraction to the wrapped core."""
return self._core._prepare_prompt(prompt_wav)
def __call__(self, generated_speech_tokens, prompt_wav) -> bytes:
"""One-shot tokens → 24 kHz WAV bytes."""
return self._core.forward(
@@ -608,8 +608,16 @@ def llm2tts(
or list(llm_output.prompt_token_ids)
)
llm_output_ids = getattr(output, "token_ids", None)
cumulative_output_ids = getattr(output, "cumulative_token_ids", None)
if llm_output_ids is None:
llm_output_ids = getattr(output, "cumulative_token_ids", [])
llm_output_ids = cumulative_output_ids or []
elif cumulative_output_ids is not None and len(cumulative_output_ids) > len(llm_output_ids):
# DELTA outputs (any streaming client) carry only the newest tokens,
# while the forwarded hidden states span prompt + whole reply. With
# the delta alone, tts_bos lands at the end of full_token_ids, the
# tts_bos..tts_eos slice collapses to nothing and the Talker is
# conditioned on an empty span, so the reply is silent.
llm_output_ids = cumulative_output_ids
# Always copy: CompletionOutput.token_ids can alias the upstream
# detokenizer's live token list. Forwarding that exact object as the
# talker prompt makes the stage-1 streaming update extend the list
@@ -393,3 +393,32 @@ class OmniNPUModelRunner(OmniGPUModelRunner, NPUModelRunner):
model_output = self._all_gather_hidden_states_and_aux(model_output)
return model_output
def _gather_runtime_additional_information(self):
"""NPU parity fix for MiniCPM-o-4.5 Code2Wav request_id injection.
On GPU, ``OmniGPUModelRunner._preprocess`` unconditionally writes
``req_infos["request_id"] = req_id`` (gpu_model_runner.py:1756) so that
every stage's ``model_intermediate_buffer`` entry carries the engine
request id. The NPU runner inherits ``execute_model`` from vllm_ascend
and never runs omni's ``_preprocess``, so this id is otherwise missing
on NPU. This override mirrors that GPU step at the gather point (called
from ``_build_model_kwargs_extra`` -> forward), using the engine
authoritative ``input_batch.req_ids`` and writing it both top-level and
into ``meta`` so Code2Wav._parse_item (which reads either key) resolves
it. Only fills when absent, so async_chunk=True handoffs that already
carry the id are left untouched.
"""
infos = super()._gather_runtime_additional_information()
for req_id, info in zip(self.input_batch.req_ids, infos):
if not isinstance(info, dict):
continue
if info.get("request_id"):
continue
meta = info.get("meta")
if not isinstance(meta, dict):
meta = {}
info["meta"] = meta
meta["request_id"] = req_id
info["request_id"] = req_id
return infos