anodyine/vllm-omni
1
0 Fork 0
Code Issues 4 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

[Bugfix][MiniCPM-o] Serve native duplex from shipping YAMLs and fence… (#6619)

Browse Source 
Signed-off-by: amy-why-3459 <wuhaiyan17@huawei.com>
This commit is contained in:
amy-why-3459 authored and GitHub committed 2026-08-26 10:56:51 +08:00
1 parent 7ae15c74c8
commit 5de6f7c37a
19 files changed
+178 -115
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.buildkite/cuda/test-merge.yml
-1
View File
@@ -169,7 +169,6 @@ steps:
- tests/e2e/online_serving/run_minicpmo_realtime_duplex_soft_interrupt.py
- vllm_omni/deploy/minicpmo_4_5.yaml
- tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py
- vllm_omni/deploy/minicpmo_4_5_duplex.yaml
- vllm_omni/experimental/fullduplex/
- vllm_omni/model_executor/models/minicpmo_4_5/
- vllm_omni/model_executor/models/step_audio2/
.buildkite/cuda/test-ready.yml
-1
View File
@@ -208,7 +208,6 @@ steps:
- tests/e2e/online_serving/run_minicpmo_realtime_duplex_soft_interrupt.py
- vllm_omni/deploy/minicpmo_4_5.yaml
- tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py
- vllm_omni/deploy/minicpmo_4_5_duplex.yaml
- vllm_omni/experimental/fullduplex/
- vllm_omni/model_executor/models/minicpmo_4_5/
- vllm_omni/model_executor/models/step_audio2/
docs/configuration/stage_configs.md
+12 -10
View File
@@ -14,7 +14,7 @@ describe fixed model topology and are not accepted in deploy YAMLs.
Common `PipelineConfig` fields include:
| Field | Description |
|-------|-------------|
| ------- | ------------- |
| `model_type` | Pipeline identifier used during model and config resolution. |
| `default_deploy_config_name` | Bundled deploy YAML loaded when the user does not pass `deploy_config`. |
| `model_arch` | Default Hugging Face architecture for the pipeline. |
@@ -26,7 +26,7 @@ Common `PipelineConfig` fields include:
Common `StagePipelineConfig` fields include:
| Field | Description |
|-------|-------------|
| ------- | ------------- |
| `stage_id` | Stable stage identifier. |
| `model_stage` | Logical stage name used by runtime and strategy resolution. |
| `execution_type` | `LLM_AR`, `LLM_GENERATION`, or `DIFFUSION`. |
@@ -49,10 +49,10 @@ deployment overrides.
The new deploy schema lives under `vllm_omni/deploy/` and is paired with a frozen `PipelineConfig` registered by the model's `pipeline.py`. Each deploy YAML has these top-level fields:
| Field | Type | Required | Default | Description |
|-------|------|----------|---------|-------------|
| ------- | ------ | ---------- | --------- | ------------- |
| `base_config` | str (path) | optional | — | Overlay parent (relative or absolute). `stages:` / `platforms:` deep-merged by stage_id; other scalars overlay-wins. Intended for user-authored overlays; prod yamls stay flat. |
| `async_chunk` | bool | optional | `true` | Enable chunked streaming between stages. Pin to `false` if the pipeline runs end-to-end. |
| `session_mode` | str | optional | `"turn"` | Session behavior. Use `"duplex"` only with a pipeline that enables duplex control. |
| `session_mode` | str | optional | `"turn"` | Session behavior. MiniCPM-o 4.5 deploy YAMLs set `"duplex"` so `/v1/realtime` and chat share the same profile. |
| `active_stream_window` | int | optional | `0` | Number of active downstream stream slots; `0` preserves all-stream cycling. |
| `duplex_session` | dict | optional | runtime defaults | Full-duplex session lifecycle, buffering, replay, and capacity limits. |
| `connectors` | dict | optional | `null` | Named connector specs (`{name, extra}`). Referenced by each stage's `input_connectors` / `output_connectors`. See [Connector schema](#connector-schema). |
@@ -84,7 +84,7 @@ subprocess on one GPU. `ray` / `external_launcher` are not fully supported yet.
Each entry under `stages:` accepts any `StageDeployConfig` field directly (no nested `engine_args:`). Only fields whose value legitimately varies across stages live here; pipeline-wide settings (trust_remote_code, distributed_executor_backend, dtype, quantization, prefix/chunked prefill, DP/PP sizes) are declared at the top level and applied to every stage. Unknown keys fall through to `engine_extras:` and are forwarded to the engine. Frequently used fields are listed below; the source-of-truth schema is `StageDeployConfig` in `vllm_omni/config/stage_config.py`.
| Field | Type | Required | Default | Description |
|-------|------|----------|---------|-------------|
| ------- | ------ | ---------- | --------- | ------------- |
| `stage_id` | int | required | — | Stage identity; matched against `PipelineConfig.stages[*].stage_id`. |
| `max_num_seqs` | int \| null | optional | `null` | Max concurrent sequences per stage. |
| `gpu_memory_utilization` | float \| null | optional | `null` | Per-stage total memory target; used for automatic KV-cache sizing. |
@@ -113,9 +113,9 @@ connectors:
# Additional connector-specific options
```
| Connector class | Use case | `extra` keys |
|-----------------|----------|--------------|
| `SharedMemoryConnector` | Same-host KV transfer between stages (default for bundled YAMLs). | None. All payloads use shared memory. |
| Connector class | Use case | `extra` keys |
|--------------------------|-----------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------|
| `SharedMemoryConnector` | Same-host KV transfer between stages (default for bundled YAMLs). | None. All payloads use shared memory. |
| `MooncakeStoreConnector` | Cross-host KV transfer over TCP. Required for multi-node deployments. | `host`, `metadata_server`, `master`, `segment` (int bytes), `localbuf` (int bytes), `proto` (`"tcp"` / `"rdma"`). |
A stage references a connector by name in its `input_connectors` / `output_connectors`:
@@ -135,7 +135,7 @@ stages:
### CLI flags
| Flag | Description |
|------|-------------|
| ------ | ------------- |
| `--deploy-config PATH` | Load a deploy YAML. **Optional** — when omitted, the bundled `vllm_omni/deploy/<model_type>.yaml` is auto-loaded by the model registry. |
| `--stage-overrides JSON` | Per-stage JSON overrides, e.g. `'{"0":{"gpu_memory_utilization":0.5}}'`. Per-stage values always win over global flags. |
| `--async-chunk` / `--no-async-chunk` | Flip the deploy YAML's `async_chunk:` bool. Unset (default) leaves the YAML value in force. |
@@ -249,7 +249,7 @@ CUDA_VISIBLE_DEVICES=0 vllm serve Qwen/Qwen3-Omni-30B-A3B-Instruct --omni \
Effective config per stage after the merge:
| Stage | Field | Final value | Source |
|-------|-------|-------------|--------|
| ------- | ------- | ------------- | -------- |
| 0 | `gpu_memory_utilization` | `0.9` | base YAML (overlay didn't touch stage 0) |
| 0 | `max_num_seqs` | `8` | per-stage CLI (`--stage-overrides`) — wins over base `32` |
| 0 | `max_model_len` | `16384` | global CLI |
@@ -273,12 +273,14 @@ when loading a deploy configuration.
Examples:
For offline inference (assuming the necessary dependencies have been imported):
```python
model_name = "Qwen/Qwen2.5-Omni-7B"
omni = Omni(model=model_name, deploy_config="/path/to/deploy_config.yaml")
```
For online serving:
```bash
vllm serve Qwen/Qwen2.5-Omni-7B --omni --port 8091 --deploy-config /path/to/deploy_config.yaml
```
examples/online_serving/minicpmo/README.md
+7 -5
View File
@@ -32,12 +32,14 @@ GPU 0 (90%) and colocates the Talker (55%) and Code2Wav (35%) on GPU 1. That
profile admits at most four concurrent sequences per stage.
| deploy config | GPUs | Notes |
|---|---|---|
| --- | --- | --- |
| `minicpmo_4_5.yaml` (default) | 1 | Memory-constrained compatibility layout. |
| `minicpmo_4_5_2gpu.yaml` | 2 | Recommended continuous-batching layout; Talker and Code2Wav share GPU 1. |
| `minicpmo_4_5_3gpu.yaml` | 3 | One GPU per stage. |
| `minicpmo_4_5_8x4090.yaml` | 8 | Full 8x4090 layout. |
| `minicpmo_4_5_duplex.yaml` | 1 | Experimental native full-duplex overlay. |
Every profile sets `session_mode: duplex`, so native full-duplex is served
from the same process as `/v1/chat/completions`.
The split pipeline preserves native-duplex epoch/turn identity, segment text,
turn completion, reference voice, and terminal-audio metadata through
@@ -55,8 +57,8 @@ vllm-omni serve openbmb/MiniCPM-o-4_5 \
```
For local ModelScope checkpoints, replace `openbmb/MiniCPM-o-4_5` with the
checkpoint path. To start the experimental native duplex backend, use
`vllm_omni/deploy/minicpmo_4_5_duplex.yaml`.
checkpoint path. For native full-duplex, connect `/v1/realtime?duplex=1` on
this same server.
### Per-stage overrides
@@ -145,7 +147,7 @@ assistant template, so it is not an apples-to-apples accuracy run.
## Run the Realtime duplex CLI demo
After the duplex backend is running, stream one WAV through the Realtime
After the server is running, stream one WAV through the Realtime
WebSocket endpoint:
```bash
tests/config/test_config_factory.py
+36 -13
View File
@@ -1109,8 +1109,35 @@ class TestDeployConfigLoading:
with pytest.raises(ValueError, match=r"stage_args.*PipelineConfig.*stages"):
load_deploy_config(deploy_path)
def test_load_minicpmo_duplex_deploy_config(self):
deploy_path = Path(get_deploy_config_path("minicpmo_4_5_duplex.yaml"))
@pytest.mark.parametrize(
("filename", "max_sessions"),
[
("minicpmo_4_5.yaml", 4),
("minicpmo_4_5_2gpu.yaml", 4),
("minicpmo_4_5_3gpu.yaml", 4),
("minicpmo_4_5_8x4090.yaml", 1),
("minicpmo_4_5_3gpu_stage1_replicas.yaml", 4),
("minicpmo_4_5_4gpu_stage1_replicas.yaml", 4),
("minicpmo_4_5_8x4090_stage1_replicas.yaml", 1),
],
)
def test_minicpmo_deploy_configs_enable_duplex(self, filename: str, max_sessions: int):
"""Every MiniCPM-o profile serves /v1/realtime, so the retired
minicpmo_4_5_duplex.yaml overlay has no replacement to fall back on.
"""
deploy = load_deploy_config(Path(get_deploy_config_path(filename)))
pipeline = resolve_pipeline_config("minicpmo_4_5")
assert isinstance(pipeline, PipelineConfig)
stages = merge_pipeline_deploy(pipeline, deploy)
assert deploy.session_mode == "duplex"
assert deploy.active_stream_window == 1
assert deploy.duplex_session.max_sessions == max_sessions
assert [stage.session_mode for stage in stages] == ["duplex", "duplex", "duplex"]
assert [stage.to_omegaconf().session_mode for stage in stages] == ["duplex", "duplex", "duplex"]
def test_load_minicpmo_default_deploy_config(self):
deploy_path = Path(get_deploy_config_path("minicpmo_4_5.yaml"))
deploy = load_deploy_config(deploy_path)
pipeline = resolve_pipeline_config("minicpmo_4_5")
@@ -1118,17 +1145,15 @@ class TestDeployConfigLoading:
stages = merge_pipeline_deploy(pipeline, deploy)
assert deploy.session_mode == "duplex"
assert deploy.async_chunk is True
assert deploy.active_stream_window == 1
assert [stage.session_mode for stage in stages] == ["duplex", "duplex", "duplex"]
assert [stage.to_omegaconf().session_mode for stage in stages] == ["duplex", "duplex", "duplex"]
assert [stage.yaml_engine_args["async_scheduling"] for stage in stages] == [False, False, False]
assert all("Async" not in (stage.scheduler_cls or "") for stage in stages)
assert stages[0].yaml_engine_args["async_scheduling"] is True
assert stages[1].yaml_engine_args["async_scheduling"] is True
assert "Async" in (stages[0].scheduler_cls or "")
assert "Async" in (stages[1].scheduler_cls or "")
assert [stage.devices for stage in deploy.stages] == ["0", "0", "0"]
assert deploy.stages[1].enforce_eager is False
assert stages[1].yaml_extras["default_sampling_params"]["max_tokens"] == 4096
assert stages[1].yaml_extras["default_sampling_params"]["min_tokens"] == 0
assert stages[1].yaml_extras["default_sampling_params"]["min_tokens"] == 50
assert stages[1].yaml_extras["default_sampling_params"]["temperature"] == 0.8
assert stages[1].yaml_extras["default_sampling_params"]["stop_token_ids"] == [6561]
assert "codec_sampling_params" not in stages[1].yaml_engine_args
@@ -2116,7 +2141,8 @@ class TestBaseConfigInheritance:
def test_minicpmo_overlays_inherit_talker_sampling_params(self):
"""Overlays must keep the base Talker codec Sampler knobs."""
for filename in (
"minicpmo_4_5_duplex.yaml",
"minicpmo_4_5_2gpu.yaml",
"minicpmo_4_5_3gpu.yaml",
"minicpmo_4_5_3gpu_stage1_replicas.yaml",
"minicpmo_4_5_4gpu_stage1_replicas.yaml",
"minicpmo_4_5_8x4090_stage1_replicas.yaml",
@@ -2128,9 +2154,6 @@ class TestBaseConfigInheritance:
sampling = deploy.stages[1].default_sampling_params
assert sampling is not None, f"{filename} stage 1 lost default_sampling_params"
assert sampling["temperature"] == 0.8, filename
# Upstream utils.TTSSamplingParams. min_tokens is not checked here:
# the duplex overlay drops it to 0 because the model budgets each
# generate_chunk itself.
assert sampling["top_k"] == 25, filename
assert sampling["top_p"] == 0.85, filename
assert sampling["repetition_penalty"] == 1.05, filename
tests/core/sched/test_omni_ar_scheduler_streaming.py
+31
View File
@@ -252,6 +252,37 @@ def test_running_decode_step_without_inter_stage_payload_does_not_raise() -> Non
sched.chunk_transfer_adapter.save_async.assert_not_called()
def test_queued_streaming_update_on_async_stop_fences_in_flight_once() -> None:
"""Native duplex + async: the next unit is usually already queued when the
current one stops, so _handle_stopped_request applies the update and the
stop site below it fences the same in-flight decode a second time. An
accumulated residue outlives the drain and swallows the next unit's
listen/speak, which the client sees as silence until its timeout.
"""
session = _make_request()
session.status = RequestStatus.RUNNING
session.resumable = True
session.append_output_token_ids([7])
session.num_computed_tokens = 4
session.num_output_placeholders = 1
# This step's frame (settled to 0 by update_from_output) plus one extra
# async decode still in flight.
session.num_in_flight_tokens = 2
sched = _make_scheduler(stage_id=0)
sched._enqueue_waiting_request = MagicMock()
def handle_stopped(request: Request) -> bool:
sched._update_request_as_session(request, _make_update([10, 20]))
return False
_run_resumable_segment_stop(session, handle_stopped=handle_stopped)
# Exactly the one unreported decode, so the drain reaches zero before the
# new segment's first frame arrives.
assert session.num_stale_output_tokens == session.num_in_flight_tokens == 1
def test_stage0_streaming_update_discards_outstanding_async_placeholder_token() -> None:
sched = _make_scheduler(stage_id=0)
session = _make_request()
tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json
+1 -1
View File
@@ -18,7 +18,7 @@
"model": "openbmb/MiniCPM-o-4_5",
"extra_cli_args": [
"--deploy-config",
"vllm_omni/deploy/minicpmo_4_5_duplex.yaml",
"vllm_omni/deploy/minicpmo_4_5.yaml",
"--trust-remote-code"
]
},
tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py
+2 -12
View File
@@ -15,20 +15,10 @@ import pytest
from huggingface_hub import snapshot_download
from tests.helpers.runtime import OmniServerParams, get_model_prefix
from tests.helpers.stage_config import get_deploy_config_path, modify_stage_config
from tests.helpers.stage_config import get_deploy_config_path
MODEL = "openbmb/MiniCPM-o-4_5"
DEPLOY_CONFIG = modify_stage_config(
get_deploy_config_path("minicpmo_4_5_duplex.yaml"),
updates={
"base_config": get_deploy_config_path("minicpmo_4_5.yaml"),
# Talker context is 4096 (tts_config.max_position_embeddings); KV sizing
# is left automatic so duplex matches the simplex deploy profiles.
"stages": {
1: {"max_model_len": 4096},
},
},
)
DEPLOY_CONFIG = get_deploy_config_path("minicpmo_4_5.yaml")
ASSET_DIR = Path(__file__).resolve().parents[3] / "assets" / "minicpmo_4_5"
RESPONSE_REQUIRED_WAV = ASSET_DIR / "response_required_16k.wav"
RESPONSE_REQUIRED_SHA256 = "2e5fd4eb3ee434ce107ee3a0591fa624a33f7683c7462f45fe651c443c9af941"
tests/entrypoints/openai_api/test_duplex_handler.py
+27 -14
View File
@@ -1,3 +1,6 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
from __future__ import annotations
import asyncio
@@ -105,16 +108,16 @@ class FakeEngineClient:
self.close_result = close_result
self.opened: list[str] = []
self.opened_fences: list[DuplexFence | None] = []
self.appended: list[tuple[str, str, object, bool]] = []
self.appended: list[tuple[str, str, Any, bool]] = []
self.append_operation_ids: list[str | None] = []
self.appended_fences: list[DuplexFence | None] = []
self.opened_configs: list[dict[str, object]] = []
self.opened_runtime_configs: list[dict[str, object]] = []
self.opened_configs: list[dict[str, Any]] = []
self.opened_runtime_configs: list[dict[str, Any]] = []
self.signals: list[tuple[str, str]] = []
self.signal_fences: list[DuplexFence | None] = []
self.signal_next_fences: list[DuplexFence | None] = []
self.signal_session_configs: list[dict[str, object] | None] = []
self.signal_runtime_configs: list[dict[str, object] | None] = []
self.signal_session_configs: list[dict[str, Any] | None] = []
self.signal_runtime_configs: list[dict[str, Any] | None] = []
self.closed: list[tuple[str, str]] = []
self.touched: list[tuple[str, DuplexLeaseActivity]] = []
self.touch_fences: list[DuplexFence] = []
@@ -136,7 +139,7 @@ class FakeEngineClient:
runtime_config: dict[str, object] | None = None,
timeout: float | None = None,
fence: DuplexFence | None = None,
) -> None:
) -> Any:
del timeout
if self.fail_open:
raise RuntimeError("open failed")
@@ -157,7 +160,7 @@ class FakeEngineClient:
timeout: float | None = None,
collect_outputs: bool = True,
fence: DuplexFence | None = None,
) -> None:
) -> Any:
del timeout, collect_outputs
self.appended.append((session_id, mode, payload, final))
self.append_operation_ids.append(operation_id)
@@ -191,7 +194,7 @@ class FakeEngineClient:
next_fence: DuplexFence | None = None,
session_config: dict[str, object] | None = None,
runtime_config: dict[str, object] | None = None,
) -> None:
) -> Any:
del timeout
if self.fail_signal or event in self.fail_signal_events:
raise RuntimeError("signal failed")
@@ -209,7 +212,7 @@ class FakeEngineClient:
reason: str = "client_close",
timeout: float | None = None,
fence: DuplexFence | None = None,
) -> None:
) -> Any:
del timeout
if self.fail_close:
raise RuntimeError("close failed")
@@ -344,7 +347,7 @@ def _project_data_plane(
result: object,
*,
session: DuplexSession | None = None,
) -> list[dict[str, object]]:
) -> list[dict[str, Any]]:
return list(data_plane.project(result, context=_data_plane_context(session)))
@@ -4194,6 +4197,7 @@ async def test_realtime_resume_preserves_append_tail_order_across_connections():
await asyncio.sleep(0.01)
else:
pytest.fail("resumable session did not open")
assert created is not None
session = handler._registry.get("sid-resume-append-order")
assert session is not None
native_state = handler._minicpmo_session_state(session)
@@ -4323,6 +4327,7 @@ async def test_realtime_takeover_does_not_clear_new_attachment_pending_turns():
await asyncio.sleep(0.001)
else:
pytest.fail("first attachment did not create the session")
assert created is not None
second = TimedWebSocket(receive_timeout_s=1.0)
second.put(
@@ -4386,6 +4391,7 @@ async def test_realtime_takeover_drops_input_queued_by_replaced_attachment():
await asyncio.sleep(0.001)
else:
pytest.fail("first attachment did not create the session")
assert created is not None
second = TimedWebSocket(receive_timeout_s=1.0)
second.put(
@@ -5002,11 +5008,12 @@ async def test_native_append_is_cancelled_by_later_wire_order_input_cancel():
*,
mode: str,
payload: object,
operation_id: str | None = None,
final: bool = False,
timeout: float | None = None,
collect_outputs: bool = True,
fence: DuplexFence | None = None,
):
) -> Any:
self.append_started.set()
try:
await self.release_append.wait()
@@ -5017,6 +5024,7 @@ async def test_native_append_is_cancelled_by_later_wire_order_input_cancel():
session_id,
mode=mode,
payload=payload,
operation_id=operation_id,
final=final,
timeout=timeout,
collect_outputs=collect_outputs,
@@ -5128,7 +5136,9 @@ async def test_duplex_handler_session_update_control_failure_rolls_back_config()
error = next(m for m in ws.sent if m.get("type") == "error")
assert error["code"] == "runtime_signal_failed"
assert error["runtime_control"]["error_count"] == 1
assert engine.signal_session_configs[-1]["instructions"] == "must roll back"
signaled_config = engine.signal_session_configs[-1]
assert signaled_config is not None
assert signaled_config["instructions"] == "must roll back"
@pytest.mark.asyncio
@@ -5429,7 +5439,7 @@ async def test_minicpmo_native_duplex_open_unsupported_fails_session_create():
@pytest.mark.asyncio
async def test_minicpmo_native_duplex_rejects_engine_without_fence_contract():
class LegacyEngineClient(FakeEngineClient):
async def open_duplex_session_async(
async def open_duplex_session_async( # type: ignore[override]
self,
session_id: str,
*,
@@ -5437,7 +5447,7 @@ async def test_minicpmo_native_duplex_rejects_engine_without_fence_contract():
capabilities: dict[str, object] | None = None,
session_config: dict[str, object] | None = None,
timeout: float | None = None,
) -> None:
) -> Any:
return await super().open_duplex_session_async(
session_id,
session_mode=session_mode,
@@ -5534,6 +5544,9 @@ async def test_minicpmo_native_duplex_separates_public_and_runtime_config(monkey
assert opened_runtime_config["duplex_stage_max_tokens"] == {"0": 20, "1": 8192}
assert opened_runtime_config["duplex_stage_sampling_params"]["0"]["top_k"] == 20
assert opened_runtime_config["duplex_stage_sampling_params"]["0"]["top_p"] == 0.8
# The Talker YAML carries upstream's min_new_token=50 for chat TTS; a duplex
# chunk is 26 codec samples, so check_stop would never release the request.
assert opened_runtime_config["duplex_stage_sampling_params"]["1"] == {"min_tokens": 0}
assert base64.b64decode(opened_runtime_config["ref_audio_data"]) == struct.pack("<1600f", *([0.25] * 1600))
vllm_omni/core/sched/omni_ar_scheduler.py
+13 -5
View File
@@ -295,7 +295,9 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
# them. Upstream vllm raises RuntimeError on this status; omni allows
# async abort (e.g. client disconnect during TTS streaming) to leave
# requests in the waiting/running queues temporarily.
for queue in (self.waiting, self.running):
waiting = getattr(self, "waiting")
running = getattr(self, "running")
for queue in (waiting, running):
for req in list(queue):
if getattr(req, "status", None) == RequestStatus.FINISHED_ABORTED:
queue.remove(req)
@@ -303,7 +305,7 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
original_waiting = None
if self._should_defer_waiting_admission():
original_waiting = self.waiting
original_waiting = waiting
self.waiting = create_request_queue(self.policy)
try:
@@ -579,8 +581,12 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
# valid new-segment frames or underflowed the drain
# assert whenever placeholder counts diverged from
# scheduled counts (spec drafts, in-flight prefill).
# Assign, never accumulate: _handle_stopped_request above
# may already have fenced the same in-flight tokens through
# a queued streaming update, and adding twice swallows the
# next duplex unit's listen/speak under async scheduling.
if request.num_in_flight_tokens > 0:
request.num_stale_output_tokens += request.num_in_flight_tokens
request.num_stale_output_tokens = request.num_in_flight_tokens
if outstanding_async_tokens > 0:
# Discard only outputs that are already in flight and
# roll back their optimistic computed-token accounting.
@@ -716,10 +722,12 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
# pre-replacement frame will drain, so the counter reaches exactly
# zero and the new segment's frames are never swallowed. This also
# covers an in-flight prefill chunk, which carries no placeholders
# but must still have its late output dropped.
# but must still have its late output dropped. Assign, never
# accumulate: update_from_output fences the same in-flight tokens when
# a resumable stop applies a queued update through this helper.
in_flight_tokens = int(getattr(session, "num_in_flight_tokens", 0) or 0)
if in_flight_tokens > 0:
session.num_stale_output_tokens += in_flight_tokens
session.num_stale_output_tokens = in_flight_tokens
if outstanding_async_tokens > 0:
# Async scheduling may already have sampled the previous
# segment's next token. Drop that late token instead of
vllm_omni/deploy/minicpmo_4_5.yaml
+7
View File
@@ -1,6 +1,13 @@
# MiniCPM-o 4.5 three-stage deploy on one large-memory GPU.
pipeline: minicpmo_4_5
async_chunk: true
# Native full-duplex shares this profile with /v1/chat/completions. Every other
# duplex_session knob keeps its runtime default.
session_mode: duplex
active_stream_window: 1
duplex_session:
# Matches Stage 0/1 max_num_seqs below.
max_sessions: 4
trust_remote_code: true
enable_prefix_caching: false
vllm_omni/deploy/minicpmo_4_5_2gpu.yaml
+7
View File
@@ -4,6 +4,13 @@
# isolated from codec/vocoder contention while preserving stage-2 batching.
pipeline: minicpmo_4_5
async_chunk: true
# Native full-duplex shares this profile with /v1/chat/completions. Every other
# duplex_session knob keeps its runtime default.
session_mode: duplex
active_stream_window: 1
duplex_session:
# Matches Stage 0/1 max_num_seqs below.
max_sessions: 4
trust_remote_code: true
enable_prefix_caching: false
vllm_omni/deploy/minicpmo_4_5_3gpu.yaml
+7
View File
@@ -4,6 +4,13 @@
# Code2Wav run independently on GPUs 0, 1, and 2.
pipeline: minicpmo_4_5
async_chunk: true
# Native full-duplex shares this profile with /v1/chat/completions. Every other
# duplex_session knob keeps its runtime default.
session_mode: duplex
active_stream_window: 1
duplex_session:
# Matches Stage 0/1 max_num_seqs below.
max_sessions: 4
trust_remote_code: true
enable_prefix_caching: false
vllm_omni/deploy/minicpmo_4_5_3gpu_stage1_replicas.yaml
-1
View File
@@ -5,7 +5,6 @@
# Stage 1: replicas on GPU 1 and 2
base_config: minicpmo_4_5.yaml
pipeline: minicpmo_4_5
session_mode: duplex
stages:
- stage_id: 0
vllm_omni/deploy/minicpmo_4_5_4gpu_stage1_replicas.yaml
-1
View File
@@ -5,7 +5,6 @@
# Stage 1: replicas on GPU 1, 2, and 3
base_config: minicpmo_4_5.yaml
pipeline: minicpmo_4_5
session_mode: duplex
stages:
- stage_id: 0
vllm_omni/deploy/minicpmo_4_5_8x4090.yaml
+4
View File
@@ -15,6 +15,10 @@
# so 4096 is the empirically-verified safe ceiling.
pipeline: minicpmo_4_5
async_chunk: true
# Native full-duplex shares this profile with /v1/chat/completions. The default
# duplex_session caps sessions at 1, matching Stage 0/1 max_num_seqs below.
session_mode: duplex
active_stream_window: 1
trust_remote_code: true
enable_prefix_caching: false
vllm_omni/deploy/minicpmo_4_5_8x4090_stage1_replicas.yaml
-1
View File
@@ -5,7 +5,6 @@
# Stage 1: replicas on GPU 4, 5, 6, and 7
base_config: minicpmo_4_5_8x4090.yaml
pipeline: minicpmo_4_5
session_mode: duplex
stages:
- stage_id: 0
vllm_omni/deploy/minicpmo_4_5_duplex.yaml
-42
View File
@@ -1,42 +0,0 @@
# MiniCPM-o 4.5 native duplex deployment.
#
# The registered minicpmo_4_5 pipeline owns the fixed
# Thinker -> Talker -> Code2Wav topology. This overlay changes session
# lifetime and bounds all three stages to two live duplex sessions.
base_config: minicpmo_4_5.yaml
pipeline: minicpmo_4_5
session_mode: duplex
active_stream_window: 1
duplex_session:
idle_ttl_s: 300
disconnect_grace_s: 30
reaper_interval_s: 5
resume_replay_ttl_s: 60
resume_replay_max_bytes_per_session: 8388608
max_pending_input_bytes_per_session: 16777216
max_pending_turns_per_session: 4
# Matches the configured Stage0/Stage1 sequence capacity below.
max_sessions: 2
completed_append_cache_size: 256
stages:
- stage_id: 0
max_num_seqs: 2
async_scheduling: false
- stage_id: 1
max_num_seqs: 2
async_scheduling: false
# Talker context is 4096 (tts_config.max_position_embeddings).
max_model_len: 4096
# Overlay only the knobs duplex changes. Codec temperature / top_k stay
# on the base YAML. Native duplex still ends each Talker request after
# one generate_chunk (26 samples, last is codec EOS) inside the model;
# YAML max_tokens is only the hard ceiling.
default_sampling_params:
max_tokens: 4096
min_tokens: 0
- stage_id: 2
max_num_seqs: 2
async_scheduling: false
vllm_omni/experimental/fullduplex/minicpmo45/adapter.py
+24 -8
View File
@@ -1,3 +1,6 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
from __future__ import annotations
import base64
@@ -97,10 +100,14 @@ class MiniCPMO45NativeDuplexServingAdapter:
cls._apply_default_scheduler_policy(runtime_config, config=config, model_config=model_config)
ref_audio = config.ref_audio
if ref_audio is None and isinstance(extra_body.get("ref_audio"), str):
ref_audio = extra_body.pop("ref_audio")
if ref_audio is None and isinstance(extra_body.get("tts_ref_audio"), str):
ref_audio = extra_body.pop("tts_ref_audio")
extra_ref_audio = extra_body.get("ref_audio")
if ref_audio is None and isinstance(extra_ref_audio, str):
extra_body.pop("ref_audio")
ref_audio = extra_ref_audio
extra_tts_ref_audio = extra_body.get("tts_ref_audio")
if ref_audio is None and isinstance(extra_tts_ref_audio, str):
extra_body.pop("tts_ref_audio")
ref_audio = extra_tts_ref_audio
if ref_audio is None:
if any(str(modality).lower() == "audio" for modality in config.modalities):
@@ -161,7 +168,11 @@ class MiniCPMO45NativeDuplexServingAdapter:
stop_token_ids = cls._native_stage0_stop_token_ids(model_config)
if stop_token_ids:
stage0_params["stop_token_ids"] = stop_token_ids
runtime_config["duplex_stage_sampling_params"] = {"0": stage0_params}
# Stage 1 keeps the deploy YAML's codec knobs, minus upstream's
# min_new_token=50: that floor is for whole-utterance chat TTS, while a
# duplex chunk is 26 codec samples, so check_stop would never release
# the Talker and Thinker would stall on the next model turn.
runtime_config["duplex_stage_sampling_params"] = {"0": stage0_params, "1": {"min_tokens": 0}}
scheduler_token_id = cls._native_scheduler_token_id(model_config)
if scheduler_token_id is not None:
runtime_config["duplex_scheduler_token_id"] = scheduler_token_id
@@ -236,6 +247,8 @@ class MiniCPMO45NativeDuplexServingAdapter:
if token_id is not None:
return token_id
eos_id = getattr(tokenizer, "eos_token_id", None)
if eos_id is None:
return None
try:
return int(eos_id)
except (TypeError, ValueError):
@@ -265,10 +278,13 @@ class MiniCPMO45NativeDuplexServingAdapter:
value = convert(token)
if isinstance(value, list):
value = value[0] if len(value) == 1 else None
try:
token_id = int(value)
except (TypeError, ValueError):
if value is None:
token_id = -1
else:
try:
token_id = int(value)
except (TypeError, ValueError):
token_id = -1
unk_token_id = getattr(tokenizer, "unk_token_id", None)
if token_id >= 0 and token_id != unk_token_id:
return token_id