diff --git a/examples/online_serving/minicpmo/README.md b/examples/online_serving/minicpmo/README.md
index 364c3a7a9..7cd60bda9 100644
--- a/examples/online_serving/minicpmo/README.md
+++ b/examples/online_serving/minicpmo/README.md
@@ -159,6 +159,33 @@ python examples/online_serving/minicpmo/realtime_duplex_demo.py \
--output-dir /tmp/minicpmo_realtime_duplex_demo
```
+Video input uses the same session. PyAV demuxes JPEG frames at `--video-fps`
+(default 1.0) and vLLM `load_audio` extracts a 16 kHz mono WAV unless
+`--input-wav` overrides the soundtrack. Frames keep their capture size
+(`--frame-max-side 0`); the server `process_image` normalizes at 448.
+
+`--stack-frames N` raises the visual refresh rate the way official duplex does:
+each 1 s unit also samples the `N-1` sub-frames captured inside it and tiles
+them into a single composite image sent next to that unit's base frame. The
+audio cadence never changes — a unit is always one second — and the wire always
+carries 2 images per unit however large `N` is, because the sub-frames share one
+composite. Official uses 5 for high refresh rate mode:
+
+```bash
+python examples/online_serving/minicpmo/realtime_duplex_demo.py \
+ --url ws://localhost:8099/v1/realtime?duplex=1 \
+ --model openbmb/MiniCPM-o-4_5 \
+ --input-video /path/to/clip.mp4 \
+ --stack-frames 5 \
+ --ref-audio /path/to/MiniCPM-o-Demo/assets/ref_audio/ref_minicpm_signature.wav \
+ --output-dir /tmp/minicpmo_realtime_duplex_video_demo
+```
+
+Detail inside a composite is capped by `scale_resolution=448` at
+`max_slice_nums=1`: official suggests HD slicing (`max_slice_nums=[2, 1]`) for
+stacked frames, which the duplex adapter does not implement yet. Reading small
+text or digits out of a wide scene is limited by that, not by frame timing.
+
## Open the experimental browser client
The browser UI serves the page and proxies the same-origin Realtime WebSocket to
diff --git a/examples/online_serving/minicpmo/realtime_duplex_demo.py b/examples/online_serving/minicpmo/realtime_duplex_demo.py
index f726eec92..d1bca7aef 100644
--- a/examples/online_serving/minicpmo/realtime_duplex_demo.py
+++ b/examples/online_serving/minicpmo/realtime_duplex_demo.py
@@ -12,7 +12,9 @@ from __future__ import annotations
import argparse
import asyncio
import base64
+import io
import json
+import math
import sys
import time
from pathlib import Path
@@ -22,8 +24,8 @@ if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
from vllm_omni.experimental.fullduplex.client import ( # noqa: E402
- PCM16_BYTES_PER_SAMPLE, # noqa: F401 - compatibility export
- PCM16_SAMPLE_RATE, # noqa: F401 - compatibility export
+ PCM16_BYTES_PER_SAMPLE,
+ PCM16_SAMPLE_RATE,
RealtimeDuplexClient,
RealtimeEventCollector,
build_realtime_url,
@@ -38,6 +40,10 @@ from vllm_omni.experimental.fullduplex.client import ( # noqa: E402
from vllm_omni.experimental.fullduplex.client import ( # noqa: E402
reference_audio_data_url as _ref_audio_data_url,
)
+from vllm_omni.experimental.fullduplex.video_stacking import ( # noqa: E402
+ concat_frames_b64,
+ unit_subframe_offsets,
+)
class _StreamingOutputWriter:
@@ -159,14 +165,258 @@ def _event_count_after(
return sum(event.get("type") == event_type for event in events[index + 1 :])
-async def run_demo(args: argparse.Namespace) -> dict[str, object]:
- input_pcm16 = read_pcm16_wav(Path(args.input_wav))
- if not input_pcm16:
- raise ValueError("input WAV has no audio")
+def _probe_video_frames_and_fps(video_path: Path) -> tuple[int, float]:
+ """Frame count and average FPS via PyAV (same approach as Daily-Omni)."""
+ import av
+ with av.open(str(video_path)) as container:
+ stream = container.streams.video[0]
+ avg_fps = float(stream.average_rate) if stream.average_rate else 0.0
+ num_frames = stream.frames or 0
+ if num_frames <= 0:
+ duration: float | None = None
+ if stream.duration is not None and stream.time_base is not None:
+ duration = float(stream.duration * stream.time_base)
+ elif container.duration is not None:
+ duration = container.duration / av.time_base
+ if duration and avg_fps:
+ num_frames = int(math.floor(duration * avg_fps))
+ if num_frames <= 0:
+ stream.thread_type = "AUTO"
+ num_frames = sum(1 for _ in container.decode(stream))
+ if num_frames <= 0:
+ raise ValueError(f"No decodable video frames in {video_path}")
+ return num_frames, avg_fps or 1.0
+
+
+def _upright(image, rotation: int):
+ """Apply a container display-matrix rotation, in degrees, to ``image``.
+
+ PyAV hands back the stored raster and leaves the display matrix for the
+ caller, unlike the ffmpeg CLI which autorotates. Phone clips routinely
+ carry +/-90, so skipping this feeds the model a sideways frame: arrows and
+ digits stop being readable and the model mirrors the motion it reports.
+ """
+ from PIL import Image
+
+ quarter_turns = {
+ 90: Image.Transpose.ROTATE_90,
+ 180: Image.Transpose.ROTATE_180,
+ 270: Image.Transpose.ROTATE_270,
+ }
+ turn = quarter_turns.get(int(rotation) % 360)
+ return image.transpose(turn) if turn is not None else image
+
+
+def _decode_video_frames_rgb(video_path: Path, frame_idx: list[int]) -> list:
+ """Decode presentation-order ``frame_idx`` into upright RGB ``PIL.Image``s."""
+ import av
+
+ wanted = sorted({int(i) for i in frame_idx})
+ if not wanted:
+ return []
+
+ decoded: dict[int, object] = {}
+ with av.open(str(video_path)) as container:
+ stream = container.streams.video[0]
+ stream.thread_type = "AUTO"
+ cursor = 0
+ for position, frame in enumerate(container.decode(stream)):
+ if wanted[cursor] > position:
+ continue
+ image = _upright(frame.to_image(), getattr(frame, "rotation", 0) or 0)
+ while cursor < len(wanted) and wanted[cursor] <= position:
+ decoded[wanted[cursor]] = image
+ cursor += 1
+ if cursor >= len(wanted):
+ break
+
+ if not decoded:
+ raise ValueError(f"Decoded no frames from {video_path} for indices {wanted[:8]}")
+ last_image = decoded[max(decoded)]
+ return [decoded.get(int(i), last_image) for i in frame_idx]
+
+
+def _resize_max_side(image, max_side: int):
+ width, height = image.size
+ longest = max(width, height)
+ if longest <= max_side or max_side <= 0:
+ return image
+ scale = max_side / float(longest)
+ new_size = (max(1, int(round(width * scale))), max(1, int(round(height * scale))))
+ return image.resize(new_size)
+
+
+def _jpeg_bytes(image, *, quality: int = 95) -> bytes:
+ buf = io.BytesIO()
+ image.convert("RGB").save(buf, format="JPEG", quality=quality)
+ return buf.getvalue()
+
+
+def _to_pcm16_bytes(audio_np) -> bytes:
+ import numpy as np
+
+ arr = np.asarray(audio_np).reshape(-1)
+ if arr.dtype != np.int16:
+ arr = (np.clip(arr, -1.0, 1.0) * 32767.0).astype(np.int16)
+ return arr.tobytes()
+
+
+def _pcm16_wav_duration_s(path: Path) -> float:
+ pcm16 = read_pcm16_wav(path)
+ return len(pcm16) / float(PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE)
+
+
+def _extract_video_frames(
+ video_path: Path,
+ work_dir: Path,
+ *,
+ fps: float,
+ max_side: int,
+ duration_s: float,
+ stack_frames: int = 1,
+) -> tuple[list[str], list[str | None]]:
+ """Demux JPEG frames only. Frame count follows ``duration_s``, not soundtrack.
+
+ Returns the base frames (one per model unit) and the parallel stacked track.
+ With ``stack_frames=N`` each unit also samples its ``N-1`` interior
+ sub-frames and tiles them into one composite, which is how official duplex
+ raises the visual refresh rate while keeping one second of audio per unit.
+ A unit whose interior falls past the end of the clip stacks nothing.
+ """
+ num_frames, avg_fps = _probe_video_frames_and_fps(video_path)
+ sample_fps = max(float(fps), 1e-6)
+ unit_s = 1.0 / sample_fps
+ timeline_s = max(float(duration_s), unit_s)
+ num_samples = max(1, int(math.ceil(timeline_s * sample_fps - 1e-9)))
+ offsets = unit_subframe_offsets(stack_frames, unit_s=unit_s)
+
+ def frame_at(seconds: float) -> int:
+ # Official get_video_frame_audio_segments uses truncating int(), not
+ # round(), so 29.97 fps lands on the same presentation index.
+ return min(int(seconds * avg_fps), num_frames - 1)
+
+ base_idx = [frame_at(unit * unit_s) for unit in range(num_samples)]
+ sub_times = [
+ [unit * unit_s + offset for offset in offsets if unit * unit_s + offset < timeline_s]
+ for unit in range(num_samples)
+ ]
+ sub_idx = [frame_at(seconds) for times in sub_times for seconds in times]
+ images = _decode_video_frames_rgb(video_path, base_idx + sub_idx)
+ sub_images = iter(images[len(base_idx) :])
+
+ frame_dir = work_dir / "frames"
+ frame_dir.mkdir(parents=True, exist_ok=True)
+ for stale in frame_dir.glob("*.jpg"):
+ stale.unlink()
+ frames_b64: list[str] = []
+ stacked_b64: list[str | None] = []
+ for index, (image, times) in enumerate(zip(images[: len(base_idx)], sub_times), start=1):
+ jpeg = _jpeg_bytes(_resize_max_side(image, max_side))
+ (frame_dir / f"frame_{index:05d}.jpg").write_bytes(jpeg)
+ frames_b64.append(base64.b64encode(jpeg).decode("ascii"))
+ interior = [_resize_max_side(next(sub_images), max_side) for _ in times]
+ if not interior:
+ stacked_b64.append(None)
+ continue
+ composite = concat_frames_b64(interior)
+ (frame_dir / f"stack_{index:05d}.jpg").write_bytes(base64.b64decode(composite))
+ stacked_b64.append(composite)
+ return frames_b64, stacked_b64
+
+
+def _extract_video_soundtrack(video_path: Path, work_dir: Path) -> Path:
+ """Write a 16 kHz mono WAV from the video soundtrack."""
+ try:
+ from vllm.multimodal.media.audio import load_audio
+ except ImportError as exc: # pragma: no cover
+ raise SystemExit("--input-video needs vLLM audio helpers (load_audio)") from exc
+
+ audio_np, sr = load_audio(str(video_path), sr=PCM16_SAMPLE_RATE, mono=True)
+ work_dir.mkdir(parents=True, exist_ok=True)
+ audio_path = work_dir / "input_16k.wav"
+ write_pcm16_wav(audio_path, _to_pcm16_bytes(audio_np), sample_rate_hz=int(sr))
+ return audio_path
+
+
+def _extract_video_input(
+ video_path: Path,
+ work_dir: Path,
+ *,
+ fps: float,
+ max_side: int,
+ stack_frames: int,
+) -> tuple[Path, list[str], list[str | None]]:
+ """Demux a video into duplex inputs: 16 kHz mono WAV plus JPEG frames.
+
+ Use this only when the video soundtrack is the audio source. When the
+ caller already has ``--input-wav``, extract frames against that WAV
+ duration instead so a silent clip cannot fail ``load_audio``.
+ """
+ audio_path = _extract_video_soundtrack(video_path, work_dir)
+ frames_b64, stacked_b64 = _extract_video_frames(
+ video_path,
+ work_dir,
+ fps=fps,
+ max_side=max_side,
+ duration_s=_pcm16_wav_duration_s(audio_path),
+ stack_frames=stack_frames,
+ )
+ return audio_path, frames_b64, stacked_b64
+
+
+def _resolve_duplex_av_inputs(
+ *,
+ input_wav: str | None,
+ input_video: str | None,
+ work_dir: Path,
+ fps: float,
+ max_side: int,
+ stack_frames: int = 1,
+) -> tuple[str, list[str], list[str | None]]:
+ """Return ``(wav_path, frames_b64, stacked_b64)``. External WAV wins over video audio."""
+ if input_video:
+ video_path = Path(input_video)
+ if input_wav:
+ frames, stacked = _extract_video_frames(
+ video_path,
+ work_dir,
+ fps=fps,
+ max_side=max_side,
+ duration_s=_pcm16_wav_duration_s(Path(input_wav)),
+ stack_frames=stack_frames,
+ )
+ return input_wav, frames, stacked
+ extracted_wav, frames, stacked = _extract_video_input(
+ video_path,
+ work_dir,
+ fps=fps,
+ max_side=max_side,
+ stack_frames=stack_frames,
+ )
+ return str(extracted_wav), frames, stacked
+ if not input_wav:
+ raise SystemExit("provide --input-wav, --input-video, or both")
+ return input_wav, [], []
+
+
+async def run_demo(args: argparse.Namespace) -> dict[str, object]:
output_dir = Path(args.output_dir)
stream_writer = _StreamingOutputWriter(output_dir)
+ input_wav, video_frames, stacked_video_frames = _resolve_duplex_av_inputs(
+ input_wav=args.input_wav,
+ input_video=args.input_video,
+ work_dir=output_dir / "video_input",
+ fps=args.video_fps,
+ max_side=args.frame_max_side,
+ stack_frames=args.stack_frames,
+ )
+
+ input_pcm16 = read_pcm16_wav(Path(input_wav))
+ if not input_pcm16:
+ raise ValueError("input WAV has no audio")
+
url = build_realtime_url(
args.url,
args.model,
@@ -184,10 +434,12 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
timeout_s=args.timeout_s,
)
stream_event_cursor = len(client.events.events)
- await client.stream_pcm16(
+ frames_sent = await client.stream_pcm16(
input_pcm16,
chunk_ms=args.chunk_ms,
realtime=not args.no_realtime_pacing,
+ video_frames=video_frames,
+ stacked_video_frames=stacked_video_frames,
)
commit_event_cursor = len(client.events.events)
stream_decision = _latest_model_decision(client.events.events, stream_event_cursor)
@@ -310,6 +562,10 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
committed_index,
),
},
+ "input_video": args.input_video,
+ "video_frames_extracted": len(video_frames),
+ "video_frames_sent": frames_sent,
+ "video_stacked_frames_extracted": sum(frame is not None for frame in stacked_video_frames),
"audio_bytes": len(audio),
"audio_chunk_count": len(stream_writer.audio_chunk_paths),
"audio_chunk_files": [str(path) for path in stream_writer.audio_chunk_paths],
@@ -356,7 +612,48 @@ def parse_args() -> argparse.Namespace:
parser.add_argument("--url", default="ws://localhost:8099/v1/realtime?duplex=1")
parser.add_argument("--model", default="openbmb/MiniCPM-o-4_5")
parser.add_argument("--session-id")
- parser.add_argument("--input-wav", required=True)
+ parser.add_argument(
+ "--input-wav",
+ help="Mono 16 kHz PCM16 clip to stream. Optional when --input-video carries the audio track.",
+ )
+ parser.add_argument(
+ "--input-video",
+ help=(
+ "Video to stream as omni duplex input. PyAV demuxes frames and "
+ "vLLM load_audio extracts a 16 kHz mono WAV (unless --input-wav overrides the audio)."
+ ),
+ )
+ parser.add_argument(
+ "--video-fps",
+ type=float,
+ default=1.0,
+ help="Frame extraction rate. The wire contract carries ~1 frame per second of audio.",
+ )
+ parser.add_argument(
+ "--frame-max-side",
+ type=int,
+ default=0,
+ help=(
+ "Optional client-side downscale of the longest side, in pixels. "
+ "0 (default) keeps the capture resolution so the server "
+ "process_image can normalize at scale_resolution=448 from a "
+ "higher-quality source. The web demo never resizes on the client."
+ ),
+ )
+ parser.add_argument(
+ "--stack-frames",
+ type=int,
+ default=1,
+ help=(
+ "Frames sampled per 1 s audio unit. 1 (default) sends the unit's "
+ "frame alone. N > 1 also samples the N-1 sub-frames inside the unit "
+ "and tiles them into one composite sent next to the base frame, so "
+ "motion (arrows, doors, digits changing) is visible at N fps while "
+ "the audio stays 1 s per unit. Official uses 5 for high refresh "
+ "rate mode; the wire carries 2 images per unit at any N because the "
+ "sub-frames share one composite."
+ ),
+ )
parser.add_argument(
"--ref-audio",
required=True,
diff --git a/tests/benchmarks/test_daily_omni_pack_modes.py b/tests/benchmarks/test_daily_omni_pack_modes.py
index 0fd231857..34d2a5637 100644
--- a/tests/benchmarks/test_daily_omni_pack_modes.py
+++ b/tests/benchmarks/test_daily_omni_pack_modes.py
@@ -1,3 +1,6 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
"""Tests for Daily-Omni ``--daily-omni-pack-mode`` packing recipes.
The ``minicpm-interleave`` mode is an accuracy-critical protocol: MiniCPM-o only reaches
@@ -27,6 +30,8 @@ _MODULE_NAME = "vllm_omni.benchmarks.data_modules.daily_omni_dataset"
if _MODULE_NAME not in sys.modules:
_spec = importlib.util.spec_from_file_location(_MODULE_NAME, _MODULE_PATH)
+ if _spec is None or _spec.loader is None:
+ raise ImportError(f"Cannot load {_MODULE_NAME} from {_MODULE_PATH}")
_mod = importlib.util.module_from_spec(_spec)
sys.modules[_MODULE_NAME] = _mod
_spec.loader.exec_module(_mod)
@@ -87,7 +92,9 @@ def _stub_extract(monkeypatch, dataset: DailyOmniDataset, counter: list[int]) ->
def fake_extract(video_path, *, audio_path, include_audio, **_kw):
counter.append(1)
frames = [Image.new("RGB", (16, 16), color=(i, i, i)) for i in range(_NUM_FRAMES)]
- segments = [np.zeros(1600, dtype=np.float32) for _ in range(_NUM_FRAMES)] if include_audio else []
+ segments: list[np.ndarray] = (
+ [np.zeros(1600, dtype=np.float32) for _ in range(_NUM_FRAMES)] if include_audio else []
+ )
return frames, segments
monkeypatch.setattr(dataset, "_extract_minicpm_frame_audio_segments", fake_extract)
diff --git a/tests/e2e/features/fullduplex/test_client.py b/tests/e2e/features/fullduplex/test_client.py
index 601bb1d1c..53965b89e 100644
--- a/tests/e2e/features/fullduplex/test_client.py
+++ b/tests/e2e/features/fullduplex/test_client.py
@@ -18,6 +18,10 @@ from vllm_omni.experimental.fullduplex.client import (
from vllm_omni.experimental.fullduplex.minicpmo45.policy import (
MiniCPMO45DuplexPolicy,
)
+from vllm_omni.experimental.fullduplex.video_stacking import (
+ concat_frames,
+ unit_subframe_offsets,
+)
pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
@@ -418,3 +422,62 @@ def test_realtime_client_pcm16_wav_round_trip(tmp_path):
assert wav_file.getnchannels() == 1
assert wav_file.getframerate() == 16_000
assert read_pcm16_wav(path) == pcm16
+
+
+def _stack_test_image(color, size=(40, 30)):
+ from PIL import Image
+
+ return Image.new("RGB", size, color)
+
+
+def test_video_stacking_tiles_a_units_subframes_into_one_image():
+ frames = [_stack_test_image((255, 0, 0)), _stack_test_image((0, 255, 0))]
+
+ composite = concat_frames(frames)
+
+ # Landscape cells stack into a column because that canvas is squarer, and
+ # the interior seam carries the separator band that lets the model tell the
+ # sub-frames apart.
+ assert composite.size == (40, 30 * 2 + 6)
+ assert composite.getpixel((20, 30 + 3)) == (0, 0, 0)
+
+
+def test_video_stacking_picks_the_squarest_grid():
+ portrait = [_stack_test_image((255, 0, 0), size=(20, 60)) for _ in range(2)]
+ four = [_stack_test_image((0, 0, 255)) for _ in range(4)]
+
+ # A portrait clip tiles side by side, four frames always tile 2x2.
+ assert concat_frames(portrait).size == (20 * 2 + 6, 60)
+ assert concat_frames(four).size == (40 * 2 + 6, 30 * 2 + 6)
+
+
+def test_video_stacking_skips_the_subframe_that_duplicates_the_base_frame():
+ # Official samples stack_frames=5 as 0.2/0.4/0.6/0.8 s: offset 0 is the base
+ # frame, already sent as frame_list[0].
+ assert unit_subframe_offsets(5) == pytest.approx([0.2, 0.4, 0.6, 0.8])
+ assert unit_subframe_offsets(1) == []
+
+
+def test_realtime_client_sends_each_units_composite_beside_its_base_frame():
+ sent: list[dict] = []
+
+ class _Client(RealtimeDuplexClient):
+ def __init__(self):
+ pass
+
+ async def send(self, event):
+ sent.append(event)
+
+ asyncio.run(
+ _Client().stream_pcm16(
+ b"\x01\x00" * (16_000 * 3),
+ chunk_ms=200,
+ realtime=False,
+ video_frames=["f0", "f1"],
+ stacked_video_frames=["s0", None],
+ )
+ )
+
+ # A composite belongs to the unit it was captured in, so it rides the same
+ # append as that unit's base frame; a unit without one sends the base alone.
+ assert [event["video_frames"] for event in sent if "video_frames" in event] == [["f0", "s0"], ["f1"]]
diff --git a/tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py b/tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py
index b1575f14e..b201df660 100644
--- a/tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py
+++ b/tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py
@@ -91,6 +91,19 @@ def resolve_ref_audio() -> Path:
return ref_audio
+def duplex_camera_frames(*, seconds: int, cache_dir: Path) -> list[str]:
+ """Base64 JPEG frames of a moving synthetic scene, one per second.
+
+ Mirrors what the realtime web client captures (1 fps, no client-side
+ resize: the server normalizes at ``scale_resolution=448``).
+ """
+ from tests.e2e.online_serving.helpers.minicpmo_realtime_duplex_scenarios import _video_frames_from_file
+ from tests.helpers.media import generate_synthetic_video
+
+ video = generate_synthetic_video(448, 448, 30 * seconds, cache_dir=cache_dir)
+ return _video_frames_from_file(Path(video["file_path"]))
+
+
def realtime_url(omni_server) -> str:
return f"ws://{omni_server.host}:{omni_server.port}/v1/realtime?duplex=1"
@@ -113,6 +126,8 @@ def demo_args(
output_audio_format="pcm16",
chunk_ms=200,
realtime_input=True,
+ input_video=None,
+ stack_frames=1,
first_turn_ms=1400,
turn_duration_ms=[],
first_turn_transcript="duplex CI speech",
diff --git a/tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py b/tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py
index 9ac816057..7776f2a0f 100644
--- a/tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py
+++ b/tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py
@@ -37,8 +37,10 @@ import json
import sys
import time
import wave
+from collections.abc import Sequence
from dataclasses import dataclass, field
from pathlib import Path
+from types import SimpleNamespace
_REPO_ROOT = Path(__file__).resolve().parents[4]
if str(_REPO_ROOT) not in sys.path:
@@ -55,15 +57,100 @@ from vllm_omni.experimental.fullduplex.client import ( # noqa: E402
PCM16_SAMPLE_RATE,
RealtimeEventCollector,
build_realtime_url,
+ duplex_unit_boundary_ms,
read_pcm16_wav,
summarize_session_request_metrics,
)
from vllm_omni.experimental.fullduplex.client import ( # noqa: E402
reference_audio_data_url as _ref_audio_data_url,
)
+from vllm_omni.experimental.fullduplex.video_stacking import ( # noqa: E402
+ concat_frames_b64,
+ unit_subframe_offsets,
+)
_url_with_model = build_realtime_url
_read_wav_pcm16 = read_pcm16_wav
+DemoArgs = argparse.Namespace | SimpleNamespace
+
+
+def _video_frames_from_file(
+ path: Path,
+ *,
+ fps: float = 1.0,
+ jpeg_quality: int = 70,
+ stack_frames: int = 1,
+) -> tuple[list[str], list[str | None]]:
+ """Decode ``path`` into base64 JPEG frames at the omni duplex cadence.
+
+ One frame per second matches what the realtime web client captures, so the
+ stream stays representative of a live camera. ``stack_frames > 1`` also
+ samples each unit's interior sub-frames and tiles them into one composite,
+ the official way to raise visual refresh rate without changing the audio
+ cadence. Returns ``(base_frames, stacked_frames)`` with the tracks parallel.
+ """
+ import cv2
+
+ offsets = unit_subframe_offsets(stack_frames)
+ capture = cv2.VideoCapture(str(path))
+ if not capture.isOpened():
+ raise ValueError(f"cannot open video: {path}")
+
+ def encode(frame_bgr, index: int) -> str:
+ ok, jpeg = cv2.imencode(".jpg", frame_bgr, [int(cv2.IMWRITE_JPEG_QUALITY), jpeg_quality])
+ if not ok:
+ raise ValueError(f"cannot JPEG-encode frame {index} of {path}")
+ return base64.b64encode(jpeg.tobytes()).decode("ascii")
+
+ try:
+ source_fps = capture.get(cv2.CAP_PROP_FPS) or 0.0
+ stride = max(1, round(source_fps / fps)) if source_fps > 0 else 1
+ # Sub-frame k of a unit sits ``offset * stride`` frames past its base.
+ sub_strides = {max(1, round(offset * stride)): position for position, offset in enumerate(offsets)}
+ frames: list[str] = []
+ interiors: list[list[str]] = []
+ index = 0
+ while True:
+ ok, frame_bgr = capture.read()
+ if not ok:
+ break
+ position, within_unit = divmod(index, stride)
+ if within_unit == 0:
+ frames.append(encode(frame_bgr, index))
+ interiors.append([])
+ elif within_unit in sub_strides and position < len(interiors):
+ interiors[position].append(encode(frame_bgr, index))
+ index += 1
+ finally:
+ capture.release()
+ if not frames:
+ raise ValueError(f"video has no decodable frames: {path}")
+ stacked = [concat_frames_b64(interior) if interior else None for interior in interiors]
+ return frames, stacked
+
+
+def _resolve_video_frames(args: DemoArgs) -> tuple[list[str], list[str | None]]:
+ """Base64 JPEG frames for the duplex camera track, newest last.
+
+ Returns the base track plus the parallel stacked composites. Only a video
+ source has sub-frames to stack; an explicit frame list or a still image
+ stacks nothing.
+ """
+ frames = list(getattr(args, "video_frames_b64", None) or [])
+ if frames:
+ stacked = list(getattr(args, "video_stacked_frames_b64", None) or [])
+ stacked += [None] * (len(frames) - len(stacked))
+ return frames, stacked[: len(frames)]
+ input_video = getattr(args, "input_video", None)
+ if input_video:
+ return _video_frames_from_file(
+ Path(input_video).expanduser(),
+ stack_frames=max(1, int(getattr(args, "stack_frames", 1) or 1)),
+ )
+ frame_image = getattr(args, "frame_image", None)
+ if frame_image:
+ return [base64.b64encode(Path(frame_image).expanduser().read_bytes()).decode("ascii")], [None]
+ return [], []
@dataclass
@@ -373,17 +460,20 @@ class DemoState:
commit_indices = [
index for index, event in enumerate(self.events) if event.get("type") == "input_audio_buffer.committed"
]
- decision_indices = [
- index
- for index, event in enumerate(self.events)
- if event.get("type") == "response.created"
- or (
- event.get("type") == "response.listen"
- and isinstance(event.get("response"), dict)
- and isinstance(event["response"].get("metadata"), dict)
- and event["response"]["metadata"].get("model_listen") is True
- )
- ]
+ decision_indices: list[int] = []
+ for index, event in enumerate(self.events):
+ event_type = event.get("type")
+ if event_type == "response.created":
+ decision_indices.append(index)
+ continue
+ if event_type != "response.listen":
+ continue
+ response = event.get("response")
+ if not isinstance(response, dict):
+ continue
+ metadata = response.get("metadata")
+ if isinstance(metadata, dict) and metadata.get("model_listen") is True:
+ decision_indices.append(index)
first_input_index = self.first_index("input_audio_buffer.speech_started")
if first_input_index is None:
first_input_index = commit_indices[0] if commit_indices else None
@@ -500,7 +590,7 @@ class DemoState:
return stale
-def _session_update_event(args: argparse.Namespace) -> dict[str, object]:
+def _session_update_event(args: DemoArgs) -> dict[str, object]:
session_payload: dict[str, object] = {
"model": args.model,
"modalities": ["audio", "text"],
@@ -836,7 +926,8 @@ async def _send_pcm16(
hints: dict[str, object] | None = None,
first_chunk_hints: dict[str, object] | None = None,
on_model_unit_ready=None,
- frame_b64: str | None = None,
+ frames_b64: Sequence[str] | None = None,
+ stacked_frames_b64: Sequence[str | None] | None = None,
) -> None:
hints = hints or {}
first_chunk_hints = first_chunk_hints or {}
@@ -844,6 +935,8 @@ async def _send_pcm16(
model_unit_bytes = PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE
audio_ms = 0
next_model_unit_bytes = model_unit_bytes
+ frames = list(frames_b64 or [])
+ stacked = list(stacked_frames_b64 or [])
frames_sent = 0
for offset in range(0, len(pcm16), chunk_bytes):
chunk = pcm16[offset : offset + chunk_bytes]
@@ -852,9 +945,18 @@ async def _send_pcm16(
chunk_hints = dict(hints)
if offset == 0:
chunk_hints.update(first_chunk_hints)
- if frame_b64 is not None and audio_ms > frames_sent * 1000:
- # Omni duplex cadence: one camera frame per 1 s of audio.
- chunk_hints["video_frames"] = [frame_b64]
+ if frames and audio_ms >= duplex_unit_boundary_ms(frames_sent):
+ # Omni duplex cadence: one camera frame per model unit, riding the
+ # append that closes it, so the frame and the second of audio it was
+ # captured during enter the same unit. A video advances one frame per
+ # unit and holds its last frame when the audio outlives the clip; a
+ # still image is a one-element list and therefore repeats.
+ # A stacked composite of that unit's interior sub-frames rides the
+ # same append right after the base frame (wire maximum of 2).
+ index = min(frames_sent, len(frames) - 1)
+ composite = stacked[index] if index < len(stacked) else None
+ base = frames[index]
+ chunk_hints["video_frames"] = [base] if composite is None else [base, composite]
frames_sent += 1
await ws.send(
json.dumps(
@@ -935,7 +1037,8 @@ async def _send_clean_turn(
realtime_input: bool = False,
model_policy_settle_s: float = 2.0,
commit_input: bool = True,
- frame_b64: str | None = None,
+ frames_b64: Sequence[str] | None = None,
+ stacked_frames_b64: Sequence[str | None] | None = None,
) -> tuple[str | None, str]:
before_created = state.count("response.created")
before_model_listen = state.model_listen_count
@@ -957,7 +1060,8 @@ async def _send_clean_turn(
chunk_ms=chunk_ms,
realtime_delay=realtime_input,
hints={"transcript": transcript} if send_transcript_hint else {},
- frame_b64=frame_b64,
+ frames_b64=frames_b64,
+ stacked_frames_b64=stacked_frames_b64,
)
if commit_input:
state.input_commit_sent_at_s.append(time.monotonic())
@@ -1163,7 +1267,8 @@ async def _send_listen_only_overlap_pair(
model_policy_settle_s: float,
validation_mode: str,
silence_ms: int,
- frame_b64: str | None = None,
+ frames_b64: Sequence[str] | None = None,
+ stacked_frames_b64: Sequence[str | None] | None = None,
) -> tuple[list[str | None], list[str], bool]:
if not realtime_input:
raise ValueError("listen-only-overlap requires --realtime-input")
@@ -1194,7 +1299,8 @@ async def _send_listen_only_overlap_pair(
realtime_delay=True,
hints={"transcript": transcripts[0]} if send_transcript_hint else {},
on_model_unit_ready=continue_until_first_speak,
- frame_b64=frame_b64,
+ frames_b64=frames_b64,
+ stacked_frames_b64=stacked_frames_b64,
)
state.input_commit_sent_at_s.append(time.monotonic())
await ws.send(json.dumps({"type": "input_audio_buffer.commit", "final": True}))
@@ -1233,7 +1339,8 @@ async def _send_listen_only_overlap_pair(
realtime_delay=True,
hints={"transcript": transcripts[1]} if send_transcript_hint else {},
on_model_unit_ready=record_model_unit_ready,
- frame_b64=frame_b64,
+ frames_b64=frames_b64,
+ stacked_frames_b64=stacked_frames_b64,
)
state.input_commit_sent_at_s.append(time.monotonic())
await ws.send(json.dumps({"type": "input_audio_buffer.commit", "final": True}))
@@ -1297,10 +1404,8 @@ async def _send_listen_only_overlap_pair(
return [first_response_id, second_response_id], ["speak", second_outcome], overlap_ok
-async def run_demo(args: argparse.Namespace) -> dict[str, object]:
- demo_frame_b64: str | None = None
- if getattr(args, "frame_image", None):
- demo_frame_b64 = base64.b64encode(Path(args.frame_image).read_bytes()).decode("ascii")
+async def run_demo(args: DemoArgs) -> dict[str, object]:
+ demo_frames_b64, demo_stacked_b64 = _resolve_video_frames(args)
turn_input_paths = _turn_input_paths(
Path(args.input_wav),
list(getattr(args, "turn_input_wav", []) or []),
@@ -1371,7 +1476,8 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
model_policy_settle_s=max(0.0, args.model_policy_settle_ms / 1000),
validation_mode=validation_mode,
silence_ms=max(0, args.silence_ms),
- frame_b64=demo_frame_b64,
+ frames_b64=demo_frames_b64,
+ stacked_frames_b64=demo_stacked_b64,
)
for turn_index in range(2, len(turn_specs)):
transcript, duration_ms = turn_specs[turn_index]
@@ -1389,7 +1495,8 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
realtime_input=realtime_input,
model_policy_settle_s=max(0.0, args.model_policy_settle_ms / 1000),
commit_input=not continuous_input,
- frame_b64=demo_frame_b64,
+ frames_b64=demo_frames_b64,
+ stacked_frames_b64=demo_stacked_b64,
)
turn_response_ids.append(response_id)
turn_outcomes.append(outcome)
@@ -1409,7 +1516,8 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
realtime_input=realtime_input,
model_policy_settle_s=max(0.0, args.model_policy_settle_ms / 1000),
commit_input=not continuous_input,
- frame_b64=demo_frame_b64,
+ frames_b64=demo_frames_b64,
+ stacked_frames_b64=demo_stacked_b64,
)
turn_response_ids.append(response_id)
turn_outcomes.append(outcome)
@@ -1584,6 +1692,8 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
"continuous_input": continuous_input,
"continuous_input_ok": continuous_input_ok,
"input_chunk_ms": args.chunk_ms,
+ "video_frame_count": len(demo_frames_b64),
+ "video_stacked_frame_count": sum(frame is not None for frame in demo_stacked_b64),
"model_policy_settle_ms": args.model_policy_settle_ms,
"turn_outcomes": turn_outcomes,
"turn_outcomes_ok": turn_outcomes_ok,
@@ -1641,6 +1751,19 @@ def parse_args() -> argparse.Namespace:
default=None,
help="Optional image file sent as an omni-duplex camera frame (one per 1 s of audio)",
)
+ parser.add_argument(
+ "--input-video",
+ default=None,
+ help="Optional video file streamed as omni-duplex camera frames (decoded at 1 fps, "
+ "overrides --frame-image; the last frame is held if the audio outlives the clip)",
+ )
+ parser.add_argument(
+ "--stack-frames",
+ type=int,
+ default=1,
+ choices=[1, 2],
+ help="Frames per append: 2 also resends the previous frame for motion context.",
+ )
parser.add_argument(
"--turn-input-wav",
action="append",
diff --git a/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py b/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py
index cff29e662..bbdf85c09 100644
--- a/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py
+++ b/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py
@@ -16,6 +16,7 @@ import websockets
from tests.e2e.online_serving.helpers.minicpmo_4_5_duplex import (
SERVER_PARAMS,
demo_args,
+ duplex_camera_frames,
multi_session_args,
realtime_url,
resolve_ref_audio,
@@ -30,10 +31,16 @@ from tests.e2e.online_serving.run_minicpmo_realtime_duplex_multi_session import
)
from tests.helpers.mark import hardware_test
from vllm_omni.experimental.fullduplex.client import build_realtime_url
+from vllm_omni.experimental.fullduplex.video_stacking import concat_frames_b64
pytestmark = pytest.mark.omni
+def _assert_positive_int(value: object) -> None:
+ assert isinstance(value, int)
+ assert value > 0
+
+
def _assert_request_metrics(metrics: object, *, expected_count: int) -> None:
assert isinstance(metrics, list)
assert len(metrics) == expected_count
@@ -142,7 +149,45 @@ def test_duplex_single_session_response_required(omni_server, tmp_path: Path) ->
args.turn_duration_ms = [args.first_turn_ms] * args.turns
result = asyncio.run(run_demo(args))
assert result["ok"] is True
- assert result["audio_delta_count"] > 0
+ _assert_positive_int(result["audio_delta_count"])
+ assert result["done_count"] == 2
+ assert result["error_count"] == 0
+ assert result["all_audio_responses_have_transcript"] is True
+ assert result["transcript_delta_done_ok"] is True
+ _assert_request_metrics(result["request_metrics"], expected_count=2)
+ _assert_session_metrics(result["session_metrics"], expected_count=2)
+
+
+@pytest.mark.advanced_model
+@hardware_test(res={"cuda": "H100", "npu": "A3"}, num_cards=1)
+@pytest.mark.parametrize("omni_server", SERVER_PARAMS, indirect=True)
+def test_duplex_single_session_video_input(omni_server, tmp_path: Path) -> None:
+ """Audio plus a 1 fps camera track, the omni-duplex video contract.
+
+ Frames ride the audio appends, so the response contract is unchanged: what
+ this covers is that interleaved vision input keeps the turn intact instead
+ of stalling or erroring out mid-segment.
+ """
+ args = demo_args(
+ omni_server=omni_server,
+ input_wav=validated_input_wav(),
+ ref_audio=resolve_ref_audio(),
+ output_dir=tmp_path / "video_input",
+ )
+ args.turns = 2
+ args.turn_duration_ms = [args.first_turn_ms] * args.turns
+ frames = duplex_camera_frames(seconds=4, cache_dir=tmp_path / "camera")
+ args.video_frames_b64 = frames
+ # Each unit also carries a composite of its interior sub-frames, so the
+ # append exercises the official two-image frame_list that carries motion.
+ args.video_stacked_frames_b64 = [concat_frames_b64([frames[index]] * 2) for index in range(len(frames))]
+
+ result = asyncio.run(run_demo(args))
+
+ assert result["ok"] is True
+ assert result["video_frame_count"] == 4
+ assert result["video_stacked_frame_count"] == 4
+ _assert_positive_int(result["audio_delta_count"])
assert result["done_count"] == 2
assert result["error_count"] == 0
assert result["all_audio_responses_have_transcript"] is True
diff --git a/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py b/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py
index 5fd65bbcb..9452448c3 100644
--- a/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py
+++ b/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py
@@ -18,6 +18,7 @@ pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
DRIVER_DIR = Path(__file__).resolve().parent
HELPERS_DIR = DRIVER_DIR / "helpers"
DEMO_PATH = HELPERS_DIR / "minicpmo_realtime_duplex_scenarios.py"
+EXAMPLE_DEMO_PATH = DRIVER_DIR.parents[2] / "examples/online_serving/minicpmo/realtime_duplex_demo.py"
MULTI_DEMO_PATH = DRIVER_DIR / "run_minicpmo_realtime_duplex_multi_session.py"
PAIR_DEMO_PATH = DRIVER_DIR / "run_minicpmo_realtime_duplex_demo_pair.py"
SOFT_INTERRUPT_DEMO_PATH = DRIVER_DIR / "run_minicpmo_realtime_duplex_soft_interrupt.py"
@@ -50,6 +51,18 @@ def _load_pair_demo_module():
return module
+def _load_example_demo_module():
+ spec = importlib.util.spec_from_file_location(
+ "minicpmo_realtime_duplex_example_demo_test",
+ EXAMPLE_DEMO_PATH,
+ )
+ assert spec is not None and spec.loader is not None
+ module = importlib.util.module_from_spec(spec)
+ sys.modules[spec.name] = module
+ spec.loader.exec_module(module)
+ return module
+
+
def _load_soft_interrupt_demo_module():
spec = importlib.util.spec_from_file_location(
"minicpmo_realtime_duplex_soft_interrupt_test",
@@ -1813,6 +1826,219 @@ def test_realtime_duplex_demo_waits_at_each_model_unit_and_stops_after_speak():
assert len(ws.messages) == 10
+def _sent_video_frames(messages: list[dict[str, object]]) -> list[list[str]]:
+ frames: list[list[str]] = []
+ for message in messages:
+ raw = message.get("video_frames")
+ if not isinstance(raw, list):
+ continue
+ frames.append([frame for frame in raw if isinstance(frame, str)])
+ return frames
+
+
+def _decode_jpeg_size(frame_b64: str) -> tuple[int, int]:
+ import io
+
+ from PIL import Image
+
+ return Image.open(io.BytesIO(base64.b64decode(frame_b64))).size
+
+
+class _RecordingWebSocket:
+ def __init__(self, demo):
+ self._demo = demo
+ self.messages: list[dict[str, object]] = []
+
+ async def send(self, payload):
+ self.messages.append(self._demo.json.loads(payload))
+
+
+def _send_seconds_of_audio(demo, *, seconds: int, frames: list[str], stacked: list[str | None] | None = None):
+ ws = _RecordingWebSocket(demo)
+ asyncio.run(
+ demo._send_pcm16(
+ ws,
+ b"\x01\x00" * (demo.PCM16_SAMPLE_RATE * seconds),
+ chunk_ms=200,
+ realtime_delay=False,
+ frames_b64=frames,
+ stacked_frames_b64=stacked,
+ )
+ )
+ return ws.messages
+
+
+def test_realtime_duplex_demo_streams_one_video_frame_per_model_unit():
+ demo = _load_demo_module()
+
+ messages = _send_seconds_of_audio(demo, seconds=3, frames=["f0", "f1", "f2", "f3"])
+
+ # Frame k rides the append that closes model unit k, so Stage0 can bind it
+ # to that unit's audio. 3 s of audio closes units at 1030 ms and 2030 ms and
+ # leaves a residual too short for a third, so only two frames go out.
+ assert _sent_video_frames(messages) == [["f0"], ["f1"]]
+ frame_indices = [index for index, message in enumerate(messages) if "video_frames" in message]
+ assert frame_indices == [5, 10]
+ # The carrying appends are exactly the ones that reach a unit boundary.
+ assert [messages[index]["audio_end_ms"] for index in frame_indices] == [1200, 2200]
+
+
+def test_realtime_duplex_demo_never_sends_a_frame_before_its_unit_can_close():
+ """Regression: frame 0 used to ride the very first 200 ms append.
+
+ Stage0 cannot close a unit until 1030 ms of audio has arrived, and it does
+ not carry frames across appends, so a frame sent that early was silently
+ dropped and every later frame ended up one unit ahead of its audio.
+ """
+ demo = _load_demo_module()
+
+ messages = _send_seconds_of_audio(demo, seconds=3, frames=["f0", "f1", "f2"])
+
+ carried = [message for message in messages if "video_frames" in message]
+ assert carried, "expected at least one frame to be sent"
+ for message in carried:
+ assert message["audio_end_ms"] >= demo.duplex_unit_boundary_ms(0)
+
+
+def test_realtime_duplex_demo_holds_the_last_video_frame_when_audio_outlives_the_clip():
+ demo = _load_demo_module()
+
+ # 4 s of audio closes three units (1030/2030/3030 ms); the two-frame clip
+ # holds its last frame for the third.
+ assert _sent_video_frames(_send_seconds_of_audio(demo, seconds=4, frames=["a", "b"])) == [["a"], ["b"], ["b"]]
+ # A still image is a one-element clip and therefore repeats every unit.
+ assert _sent_video_frames(_send_seconds_of_audio(demo, seconds=3, frames=["still"])) == [["still"], ["still"]]
+
+
+def test_realtime_duplex_demo_sends_each_units_stacked_composite_next_to_its_base_frame():
+ demo = _load_demo_module()
+
+ messages = _send_seconds_of_audio(
+ demo,
+ seconds=4,
+ frames=["f0", "f1", "f2"],
+ stacked=["s0", "s1", None],
+ )
+
+ # Official pairing is frame_list=[base, composite of that unit's interior],
+ # so the composite belongs to the same unit as the base beside it -- never
+ # to the previous one. A unit without interior sub-frames sends base alone.
+ assert _sent_video_frames(messages) == [["f0", "s0"], ["f1", "s1"], ["f2"]]
+
+
+def test_realtime_duplex_demo_decodes_a_video_file_into_one_frame_per_second(tmp_path):
+ pytest.importorskip("cv2")
+ pytest.importorskip("imageio")
+ from tests.helpers.media import generate_synthetic_video
+
+ demo = _load_demo_module()
+ # 30 fps synthetic clip: 90 frames is 3 s of camera input.
+ video = generate_synthetic_video(64, 64, 90, cache_dir=tmp_path)
+
+ frames, stacked = demo._resolve_video_frames(SimpleNamespace(input_video=video["file_path"]))
+
+ assert len(frames) == 3
+ assert all(base64.b64decode(frame, validate=True)[:3] == b"\xff\xd8\xff" for frame in frames)
+ assert len(set(frames)) == 3
+ # stack_frames defaults to 1, so there is nothing to stack.
+ assert stacked == [None, None, None]
+
+
+def test_realtime_duplex_demo_stacks_a_units_interior_subframes_into_one_composite(tmp_path):
+ pytest.importorskip("cv2")
+ pytest.importorskip("imageio")
+ from tests.helpers.media import generate_synthetic_video
+
+ demo = _load_demo_module()
+ video = generate_synthetic_video(64, 64, 90, cache_dir=tmp_path)
+
+ frames, stacked = demo._resolve_video_frames(SimpleNamespace(input_video=video["file_path"], stack_frames=5))
+
+ # stack_frames=5 raises the visual refresh rate to 5 fps, but the four
+ # interior sub-frames of each second collapse into a single composite, so
+ # the wire still carries 2 images per unit and the audio cadence is
+ # untouched.
+ assert len(frames) == 3
+ assert len(stacked) == 3
+ assert all(frame is not None for frame in stacked)
+ assert all(base64.b64decode(frame, validate=True)[:3] == b"\xff\xd8\xff" for frame in stacked)
+ composite = _decode_jpeg_size(stacked[0])
+ assert composite != _decode_jpeg_size(frames[0])
+
+
+def test_example_demo_uses_external_wav_duration_and_skips_video_soundtrack(tmp_path, monkeypatch):
+ demo = _load_example_demo_module()
+ wav = tmp_path / "external.wav"
+ demo.write_pcm16_wav(wav, b"\x00\x00" * (demo.PCM16_SAMPLE_RATE * 2), sample_rate_hz=demo.PCM16_SAMPLE_RATE)
+ soundtrack_calls: list[object] = []
+
+ def fake_soundtrack(*args, **kwargs):
+ soundtrack_calls.append(1)
+ raise AssertionError("video soundtrack must not be loaded when --input-wav is set")
+
+ def fake_frames(video_path, work_dir, *, fps, max_side, duration_s, stack_frames):
+ assert video_path == Path("/tmp/silent.mp4")
+ assert fps == 1.0
+ assert duration_s == pytest.approx(2.0)
+ return ["f0", "f1"], [None, None]
+
+ monkeypatch.setattr(demo, "_extract_video_soundtrack", fake_soundtrack)
+ monkeypatch.setattr(demo, "_extract_video_frames", fake_frames)
+
+ wav_out, frames, stacked = demo._resolve_duplex_av_inputs(
+ input_wav=str(wav),
+ input_video="/tmp/silent.mp4",
+ work_dir=tmp_path / "video_input",
+ fps=1.0,
+ max_side=0,
+ )
+
+ assert wav_out == str(wav)
+ assert frames == ["f0", "f1"]
+ assert stacked == [None, None]
+ assert soundtrack_calls == []
+
+
+def test_example_demo_extracts_video_soundtrack_only_when_wav_is_absent(tmp_path, monkeypatch):
+ demo = _load_example_demo_module()
+ extracted = tmp_path / "from_video.wav"
+ demo.write_pcm16_wav(extracted, b"\x00\x00" * demo.PCM16_SAMPLE_RATE, sample_rate_hz=demo.PCM16_SAMPLE_RATE)
+
+ def fake_extract(video_path, work_dir, *, fps, max_side, stack_frames):
+ assert video_path == Path("/tmp/clip.mp4")
+ return extracted, ["v0"], ["s0"]
+
+ monkeypatch.setattr(demo, "_extract_video_input", fake_extract)
+
+ wav_out, frames, stacked = demo._resolve_duplex_av_inputs(
+ input_wav=None,
+ input_video="/tmp/clip.mp4",
+ work_dir=tmp_path / "video_input",
+ fps=1.0,
+ max_side=0,
+ )
+
+ assert wav_out == str(extracted)
+ assert frames == ["v0"]
+ assert stacked == ["s0"]
+
+
+def test_realtime_duplex_demo_prefers_a_video_over_a_still_frame_image(tmp_path):
+ demo = _load_demo_module()
+ still = tmp_path / "frame.jpg"
+ still.write_bytes(b"\xff\xd8\xff-still")
+
+ # A still has no interior sub-frames, so it never stacks.
+ assert demo._resolve_video_frames(SimpleNamespace(frame_image=str(still))) == (
+ [base64.b64encode(still.read_bytes()).decode("ascii")],
+ [None],
+ )
+ assert demo._resolve_video_frames(
+ SimpleNamespace(frame_image=str(still), video_frames_b64=["preset"]),
+ ) == (["preset"], [None])
+ assert demo._resolve_video_frames(SimpleNamespace()) == ([], [])
+
+
def test_realtime_duplex_demo_listen_only_overlap_accepts_silence_unit_before_first_done(monkeypatch):
demo = _load_demo_module()
state = demo.DemoState()
diff --git a/tests/examples/test_minicpmo_realtime_duplex_simple_demo.py b/tests/examples/test_minicpmo_realtime_duplex_simple_demo.py
index 93fc4db38..ca87dce03 100644
--- a/tests/examples/test_minicpmo_realtime_duplex_simple_demo.py
+++ b/tests/examples/test_minicpmo_realtime_duplex_simple_demo.py
@@ -1,3 +1,6 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
import base64
import importlib.util
import sys
@@ -169,6 +172,64 @@ def test_open_streaming_response_requires_post_commit_drain():
)
+def _asymmetric_image():
+ from PIL import Image
+
+ image = Image.new("RGB", (4, 2), (0, 0, 0))
+ image.putpixel((0, 0), (255, 0, 0))
+ return image
+
+
+@pytest.mark.parametrize(
+ "rotation,expected_corner",
+ [
+ (0, (0, 0)),
+ (90, (0, 3)),
+ (180, (3, 1)),
+ (-90, (1, 0)),
+ (270, (1, 0)),
+ (-270, (0, 3)),
+ ],
+)
+def test_upright_applies_the_display_matrix_quarter_turn(rotation, expected_corner):
+ demo = _load_demo_module()
+
+ rotated = demo._upright(_asymmetric_image(), rotation)
+
+ assert rotated.getpixel(expected_corner) == (255, 0, 0)
+
+
+def test_decoded_frames_apply_the_container_display_matrix(monkeypatch, tmp_path):
+ """A -90 phone clip must reach the model upright, not on its side."""
+ demo = _load_demo_module()
+ source = _asymmetric_image()
+
+ class _Frame:
+ rotation = -90
+
+ def to_image(self):
+ return source
+
+ class _Container:
+ streams = type("_Streams", (), {"video": [type("_Stream", (), {})()]})()
+
+ def decode(self, stream):
+ yield _Frame()
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *exc):
+ return False
+
+ monkeypatch.setitem(sys.modules, "av", type("_Av", (), {"open": staticmethod(lambda path: _Container())}))
+
+ (image,) = demo._decode_video_frames_rgb(tmp_path / "clip.mp4", [0])
+
+ assert image.size == (2, 4)
+ assert image.getpixel((1, 0)) == (255, 0, 0)
+
+
def test_streaming_output_writer_persists_audio_deltas_as_they_arrive(tmp_path, capsys):
demo = _load_demo_module()
writer = demo._StreamingOutputWriter(tmp_path)
diff --git a/tests/worker/test_native_duplex_hooks.py b/tests/worker/test_native_duplex_hooks.py
index ba50be72c..a30dcdb58 100644
--- a/tests/worker/test_native_duplex_hooks.py
+++ b/tests/worker/test_native_duplex_hooks.py
@@ -1,3 +1,6 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
from __future__ import annotations
import ast
@@ -849,6 +852,161 @@ def test_minicpmo_stage0_data_plane_prefill_matches_official_unit_format():
assert result["prompt_suffix_len"] == 0
+def _stage0_vision_runtime():
+ import torch
+
+ from vllm_omni.experimental.fullduplex.minicpmo45.stage0 import (
+ MiniCPMO45Stage0DuplexRuntime,
+ )
+
+ class _StageModel(torch.nn.Module):
+ def __init__(self):
+ super().__init__()
+ self.embed = torch.nn.Embedding(256, 2)
+
+ def get_input_embeddings(self):
+ return self.embed
+
+ def get_audio_hidden_states(self, _data):
+ return [torch.tensor([[0.5, 0.5]], dtype=torch.float32)]
+
+ runtime = MiniCPMO45Stage0DuplexRuntime.__new__(MiniCPMO45Stage0DuplexRuntime)
+ runtime.stage_model = _StageModel()
+ runtime.thinker = runtime.stage_model
+ runtime.tokenizer = SimpleNamespace(
+ unk_token_id=0,
+ convert_tokens_to_ids=lambda token: {
+ "": 1,
+ "": 2,
+ "<|listen|>": 3,
+ "<|speak|>": 4,
+ "<|tts_bos|>": 5,
+ "<|tts_eos|>": 6,
+ "<|tts_pad|>": 7,
+ "<|chunk_eos|>": 8,
+ "<|chunk_tts_eos|>": 9,
+ "<|turn_eos|>": 10,
+ "<|audio|>": 11,
+ "": 12,
+ "": 13,
+ "": 14,
+ "": 15,
+ }.get(token, 0),
+ encode=lambda text, add_special_tokens=False: [201, 5],
+ )
+ runtime.processor = SimpleNamespace(get_streaming_chunk_size=lambda: 4)
+ runtime.device = "cpu"
+ runtime._init_token_ids()
+ # One 2-row block per frame keeps the expected token_ids readable; the real
+ # resampler emits VISION_EMBEDS_PER_FRAME rows per slice.
+ runtime._stage_vision_embeddings = lambda frames: [
+ [torch.full((2, 2), float(index), dtype=torch.float32)] for index in range(len(frames))
+ ]
+ return runtime
+
+
+def test_minicpmo_stage0_puts_every_frame_of_an_append_in_one_unit():
+ """Official streaming_prefill feeds the whole frame_list into one unit.
+
+ Stacked pairs used to be spread one frame per unit, so a two-frame append
+ that closed a single unit left a frame over, failed the prefill, and still
+ advanced the session state.
+ """
+ from vllm_omni.experimental.fullduplex.minicpmo45.stage0 import (
+ _MiniCPMO45Stage0SessionState,
+ )
+
+ runtime = _stage0_vision_runtime()
+ state = _MiniCPMO45Stage0SessionState(session_id="sid-stage0-stacked-frames")
+
+ result = runtime._stage_prefill_embeddings_only(
+ state,
+ np.zeros(4, dtype=np.float32),
+ video_frames=["frame-prev", "frame-cur"],
+ seq=1,
+ )
+
+ assert result["success"] is True
+ # , then each frame as + embeds + , then the audio.
+ assert result["input_token_ids"] == [1, 12, 0, 0, 13, 12, 0, 0, 13, 11]
+ assert state.audio_chunk_idx == 1
+
+
+def test_minicpmo_stage0_wraps_hd_slices_after_the_source_image():
+ """Official stacked pair is max_slice_nums=[2, 1]: source + slices, then audio."""
+ import torch
+
+ from vllm_omni.experimental.fullduplex.minicpmo45.stage0 import (
+ _MiniCPMO45Stage0SessionState,
+ )
+
+ runtime = _stage0_vision_runtime()
+ state = _MiniCPMO45Stage0SessionState(session_id="sid-stage0-hd-slices")
+ runtime._stage_vision_embeddings = lambda frames: [
+ [torch.ones((2, 2)), torch.full((2, 2), 2.0), torch.full((2, 2), 3.0)],
+ [torch.full((2, 2), 4.0)],
+ ]
+
+ result = runtime._stage_prefill_embeddings_only(
+ state,
+ np.zeros(4, dtype=np.float32),
+ video_frames=["base", "stack"],
+ seq=1,
+ )
+
+ assert result["success"] is True
+ # src p0 p1
+ # stack audio
+ assert result["input_token_ids"] == [
+ 1,
+ 12,
+ 0,
+ 0,
+ 13,
+ 14,
+ 0,
+ 0,
+ 15,
+ 14,
+ 0,
+ 0,
+ 15,
+ 12,
+ 0,
+ 0,
+ 13,
+ 11,
+ ]
+
+
+def test_minicpmo_stage0_reports_frames_that_close_no_unit_instead_of_dropping_them():
+ """A frame only binds to the unit its append closes.
+
+ Frames are not carried across appends, so one arriving early used to vanish
+ under a generic "audio not enough" buffering result.
+ """
+ from vllm_omni.experimental.fullduplex.minicpmo45.stage0 import (
+ _MiniCPMO45Stage0SessionState,
+ )
+
+ runtime = _stage0_vision_runtime()
+ state = _MiniCPMO45Stage0SessionState(session_id="sid-stage0-early-frame")
+
+ assert runtime._stage_prefill_embeddings_only(state, np.zeros(4, dtype=np.float32), seq=1)["success"] is True
+
+ early = runtime._stage_prefill_embeddings_only(
+ state,
+ np.zeros(2, dtype=np.float32),
+ video_frames=["frame-cur"],
+ seq=2,
+ )
+
+ assert early["success"] is False
+ assert "closes no model unit" in early["reason"]
+ # The half-filled append must not advance the unit counter.
+ assert state.audio_chunk_idx == 1
+
+
def test_minicpmo_stage0_speech_append_sets_pending_context_once():
import torch
@@ -2514,7 +2672,7 @@ def _ar_runner_classdefs():
def _method_calls(method):
- calls = {}
+ calls: dict[str, int] = {}
for node in ast.walk(method):
if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
calls.setdefault(node.func.attr, node.lineno)
diff --git a/vllm_omni/benchmarks/data_modules/daily_omni_dataset.py b/vllm_omni/benchmarks/data_modules/daily_omni_dataset.py
index 287a9cbca..abb02b1d5 100644
--- a/vllm_omni/benchmarks/data_modules/daily_omni_dataset.py
+++ b/vllm_omni/benchmarks/data_modules/daily_omni_dataset.py
@@ -1,3 +1,6 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
"""Daily-Omni Dataset loader for benchmark.
Daily-Omni is an audio-visual reasoning benchmark with 684 videos
@@ -627,10 +630,13 @@ class DailyOmniDataset(BenchmarkDataset):
def _load_from_local_json(self) -> None:
"""Load QA data from local JSON file."""
- if not self.qa_json_path.exists():
- raise FileNotFoundError(f"QA JSON file not found: {self.qa_json_path}")
+ qa_json_path = self.qa_json_path
+ if qa_json_path is None:
+ raise RuntimeError("qa_json_path is required for local JSON loading")
+ if not qa_json_path.exists():
+ raise FileNotFoundError(f"QA JSON file not found: {qa_json_path}")
- with open(self.qa_json_path, encoding="utf-8") as f:
+ with open(qa_json_path, encoding="utf-8") as f:
data = json.load(f)
# Support both list format and dict with "train"/"test" splits
@@ -682,8 +688,11 @@ class DailyOmniDataset(BenchmarkDataset):
one ``sample()`` / request-building implementation for both.
"""
snapshot_error: Exception | None = None
+ dataset_path = self.dataset_path
+ if dataset_path is None:
+ raise ValueError("dataset_path is required for HuggingFace loading")
try:
- root = ensure_daily_omni_hub_root(self.dataset_path)
+ root = ensure_daily_omni_hub_root(dataset_path)
except Exception as e: # noqa: BLE001 - any failure just falls through to `datasets`
snapshot_error = e
else:
@@ -1256,7 +1265,7 @@ class DailyOmniDataset(BenchmarkDataset):
num_video_frames, avg_fps = _probe_video_frames_and_fps(video_path)
duration = num_video_frames / avg_fps
num_seconds = max(1, int(math.ceil(duration)))
- second_timestamps = list(range(num_seconds))
+ second_timestamps = [float(i) for i in range(num_seconds)]
if duration > max_num_frames:
timestamps = [round(i * 0.1, 1) for i in range(int(duration / 0.1))]
diff --git a/vllm_omni/experimental/fullduplex/client.py b/vllm_omni/experimental/fullduplex/client.py
index 69ff91f99..630d7aef3 100644
--- a/vllm_omni/experimental/fullduplex/client.py
+++ b/vllm_omni/experimental/fullduplex/client.py
@@ -11,7 +11,7 @@ import json
import math
import time
import wave
-from collections.abc import Callable
+from collections.abc import Callable, Iterator, Sequence
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
@@ -28,6 +28,20 @@ except ImportError as exc: # pragma: no cover - example dependency
PCM16_SAMPLE_RATE = 16_000
PCM16_BYTES_PER_SAMPLE = 2
+# Server-side model unit boundaries, in cumulative appended audio.
+# Stage0 configures the streaming mel processor with first_chunk_ms=1035 and
+# chunk_ms=1000; the processor aligns the first chunk down to a hop_length (160
+# samples) multiple, so unit 0 closes at 16480 samples and every later unit
+# closes 16000 samples after it. Camera frames must ride the append that closes
+# a unit, otherwise Stage0 cannot bind them to that unit's audio.
+DUPLEX_FIRST_UNIT_MS = 1030
+DUPLEX_UNIT_MS = 1000
+
+
+def duplex_unit_boundary_ms(unit_index: int) -> int:
+ """Cumulative appended audio, in ms, that closes model unit ``unit_index``."""
+ return DUPLEX_FIRST_UNIT_MS + max(0, int(unit_index)) * DUPLEX_UNIT_MS
+
def _rounded_ms(value: float) -> float:
return round(float(value), 3)
@@ -561,24 +575,71 @@ class RealtimeDuplexClient:
*,
chunk_ms: int = 200,
realtime: bool = True,
- ) -> None:
+ video_frames: Sequence[str] | None = None,
+ stacked_video_frames: Sequence[str | None] | None = None,
+ ) -> int:
+ """Append PCM16 audio, optionally interleaving omni camera frames.
+
+ ``video_frames`` holds base64 JPEG/PNG frames in capture order, one per
+ second of the clip. Frame ``k`` rides the append that closes model unit
+ ``k`` (see ``duplex_unit_boundary_ms``), which reproduces the official
+ ``streaming_prefill(audio_waveform=<1 s>, frame_list=[frame])`` pairing:
+ a second of audio and the picture captured during it enter the same
+ unit. Sending on whole-second boundaries instead would strand frame 0 on
+ an append that cannot close a unit yet, and shift every later frame one
+ unit ahead of its audio.
+
+ ``stacked_video_frames`` is the optional parallel track of composites
+ built by ``video_stacking.concat_frames``: entry ``k`` tiles the
+ sub-frames captured *inside* unit ``k``, and rides the same append right
+ after the base frame, giving the official ``frame_list=[base,
+ composite]``. The audio is untouched — a unit stays one second however
+ many sub-frames the composite carries. ``None`` entries send the base
+ frame alone.
+
+ Returns the number of base frames actually sent (a clip shorter than the
+ frame list leaves the tail unsent).
+ """
chunk_bytes = max(
PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE * chunk_ms // 1000,
PCM16_BYTES_PER_SAMPLE,
)
- await self.stream_av_units(
- ((pcm16[offset : offset + chunk_bytes], None) for offset in range(0, len(pcm16), chunk_bytes)),
- realtime=realtime,
- )
+ frames = list(video_frames or [])
+ stacked = list(stacked_video_frames or [])
+
+ def units() -> Iterator[tuple[bytes, list[str] | None]]:
+ audio_end_ms = 0
+ frames_sent = 0
+ for offset in range(0, len(pcm16), chunk_bytes):
+ chunk = pcm16[offset : offset + chunk_bytes]
+ audio_end_ms += len(chunk) * 1000 // (PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE)
+ if not frames or audio_end_ms < duplex_unit_boundary_ms(frames_sent):
+ yield chunk, None
+ continue
+ # A video advances one frame per unit and holds its last frame
+ # when the audio outlives the clip; a still image is a
+ # one-element list and therefore repeats.
+ index = min(frames_sent, len(frames) - 1)
+ composite = stacked[index] if index < len(stacked) else None
+ frames_sent += 1
+ yield chunk, [frames[index]] if composite is None else [frames[index], composite]
+
+ return await self.stream_av_units(units(), realtime=realtime)
async def stream_av_units(
self,
units: Any,
*,
realtime: bool = True,
- ) -> None:
- """Stream PCM16 units, optionally attaching a JPEG to each unit."""
+ ) -> int:
+ """Stream PCM16 units, optionally attaching camera frames to each unit.
+
+ A unit's frame slot takes a single JPEG/PNG (raw bytes or base64) or a
+ sequence of them, which the Realtime wire caps at two per append.
+ Returns the number of appends that carried at least one frame.
+ """
audio_end_ms = 0
+ frames_sent = 0
for chunk, frame in units:
if not chunk:
continue
@@ -593,11 +654,18 @@ class RealtimeDuplexClient:
"duration_ms": duration_ms,
"audio_end_ms": audio_end_ms,
}
- if frame is not None:
- event["video_frames"] = [base64.b64encode(frame).decode("ascii") if isinstance(frame, bytes) else frame]
+ encoded_frames = [
+ base64.b64encode(item).decode("ascii") if isinstance(item, bytes | bytearray) else item
+ for item in (frame if isinstance(frame, list | tuple) else [frame])
+ if item is not None
+ ]
+ if encoded_frames:
+ event["video_frames"] = encoded_frames
+ frames_sent += 1
await self.send(event)
if realtime:
await asyncio.sleep(duration_ms / 1000)
+ return frames_sent
async def commit(self) -> None:
await self.send({"type": "input_audio_buffer.commit", "final": True})
diff --git a/vllm_omni/experimental/fullduplex/minicpmo45/policy.py b/vllm_omni/experimental/fullduplex/minicpmo45/policy.py
index d869e54c9..7a1779617 100644
--- a/vllm_omni/experimental/fullduplex/minicpmo45/policy.py
+++ b/vllm_omni/experimental/fullduplex/minicpmo45/policy.py
@@ -1,3 +1,6 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
from __future__ import annotations
from typing import Any
@@ -24,7 +27,9 @@ class MiniCPMO45DuplexPolicy:
# Vision framing contract (omni duplex). Official streaming_prefill feeds
# each frame as + 64 resampler embeddings + inside the
# unit, ahead of the unit's audio embeddings (max_slice_nums=1 in
- # streaming, so exactly one 64-token block per frame).
+ # streaming, so exactly one 64-token block per frame). A unit may carry
+ # its base frame plus an optional stacked composite of the sub-frames
+ # captured inside that unit (2 blocks).
VISION_EMBEDS_PER_FRAME = 64
VISION_TOKENS_PER_FRAME = VISION_EMBEDS_PER_FRAME + 2 # + embeds +
DEFAULT_MAX_NEW_SPEAK_TOKENS_PER_CHUNK = 20
@@ -77,6 +82,8 @@ class MiniCPMO45DuplexPolicy:
"audio_placeholder_token_id": "<|audio|>",
"image_start_token_id": "",
"image_end_token_id": "",
+ "slice_start_token_id": "",
+ "slice_end_token_id": "",
}
@classmethod
@@ -92,7 +99,7 @@ class MiniCPMO45DuplexPolicy:
value = value[0] if len(value) == 1 else None
unk_token_id = getattr(tokenizer, "unk_token_id", None)
try:
- candidate = int(value)
+ candidate = int(value) if value is not None else -1
except (TypeError, ValueError):
candidate = -1
if candidate >= 0 and candidate != unk_token_id:
@@ -118,7 +125,8 @@ class MiniCPMO45DuplexPolicy:
if resolved.get("listen_token_id", -1) < 0:
eos_id = getattr(tokenizer, "eos_token_id", None)
try:
- resolved["listen_token_id"] = int(eos_id)
+ if eos_id is not None:
+ resolved["listen_token_id"] = int(eos_id)
except (TypeError, ValueError):
pass
return resolved
diff --git a/vllm_omni/experimental/fullduplex/minicpmo45/runtime.py b/vllm_omni/experimental/fullduplex/minicpmo45/runtime.py
index b5fbd37a5..ecd634f73 100644
--- a/vllm_omni/experimental/fullduplex/minicpmo45/runtime.py
+++ b/vllm_omni/experimental/fullduplex/minicpmo45/runtime.py
@@ -1,11 +1,11 @@
# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
from __future__ import annotations
from base64 import b64decode
from binascii import Error as BinasciiError
-from typing import Any
+from typing import Any, cast
from vllm.sampling_params import SamplingParams
@@ -22,6 +22,9 @@ _DUPLEX_SAMPLES_PER_AUDIO_TOKEN = 1600
# + 64 resampler embeddings + per frame (max_slice_nums=1),
# matching MiniCPMO45DuplexPolicy.VISION_TOKENS_PER_FRAME.
_DUPLEX_VISION_TOKENS_PER_FRAME = 66
+# Official stacked pair uses max_slice_nums=[2, 1]: the current frame is HD
+# sliced (1 source + 2 patches on 960x540) and the composite is not.
+_DUPLEX_HD_SLICES_PER_BASE_FRAME = 3
def _duplex_frame_count(payload: object) -> int:
@@ -33,6 +36,22 @@ def _duplex_frame_count(payload: object) -> int:
return sum(1 for frame in frames if isinstance(frame, str) and frame)
+def _duplex_vision_tokens(payload: object) -> int:
+ """Scheduler slots for this append's camera track.
+
+ Audio is never stacked: a unit still carries one second of soundtrack.
+ ``stack_frames`` only adds a second *image*. Official HD on that pair is
+ ``[2, 1]``, so the base frame reserves three 66-token blocks and every
+ extra frame reserves one.
+ """
+ count = _duplex_frame_count(payload)
+ if count <= 0:
+ return 0
+ if count >= 2:
+ return (_DUPLEX_HD_SLICES_PER_BASE_FRAME + (count - 1)) * _DUPLEX_VISION_TOKENS_PER_FRAME
+ return count * _DUPLEX_VISION_TOKENS_PER_FRAME
+
+
def _duplex_pcm_sample_count(payload: object) -> int | None:
if not isinstance(payload, dict):
return None
@@ -48,7 +67,7 @@ def _duplex_pcm_sample_count(payload: object) -> int | None:
def duplex_payload_is_exact_chunks(payload: object) -> bool:
sample_count = _duplex_pcm_sample_count(payload)
- return bool(sample_count) and sample_count % _DUPLEX_CHUNK_SAMPLES == 0
+ return sample_count is not None and sample_count != 0 and sample_count % _DUPLEX_CHUNK_SAMPLES == 0
def duplex_first_append_unit_count(payload: object) -> int | None:
@@ -59,7 +78,7 @@ def duplex_first_append_unit_count(payload: object) -> int | None:
def duplex_scheduler_token_budget(payload: object, *, default: int = 64) -> int:
- vision_tokens = _duplex_frame_count(payload) * _DUPLEX_VISION_TOKENS_PER_FRAME
+ vision_tokens = _duplex_vision_tokens(payload)
sample_count = _duplex_pcm_sample_count(payload)
if sample_count is None:
return max(1, int(default)) + vision_tokens
@@ -114,9 +133,7 @@ def build_duplex_data_plane_prompt(
token_budget += context_reserve
first_units = duplex_first_append_unit_count(payload)
if first_units is not None:
- token_budget = (
- context_reserve + first_units * 12 - 1 + _duplex_frame_count(payload) * _DUPLEX_VISION_TOKENS_PER_FRAME
- )
+ token_budget = context_reserve + first_units * 12 - 1 + _duplex_vision_tokens(payload)
if seq > 1 and duplex_payload_is_exact_chunks(payload):
token_budget += 1
if final and duplex_payload_is_exact_chunks(payload):
@@ -124,7 +141,7 @@ def build_duplex_data_plane_prompt(
extra_body = session_config.get("extra_body")
raw_token_id = runtime_config.get("duplex_scheduler_token_id")
try:
- token_id = max(0, int(raw_token_id))
+ token_id = 0 if raw_token_id is None else max(0, int(raw_token_id))
except (TypeError, ValueError):
token_id = 0
force_listen_count = _duplex_force_listen_count(extra_body)
@@ -163,16 +180,17 @@ def build_duplex_data_plane_prompt(
def _coerce_int(value: object) -> int | None:
- if hasattr(value, "detach"):
+ detach = getattr(value, "detach", None)
+ if callable(detach):
try:
- value = value.detach().cpu().reshape(-1)
- if value.numel() == 0:
+ flat: Any = detach().cpu().reshape(-1)
+ if flat.numel() == 0:
return None
- value = value[0].item()
+ value = flat[0].item()
except Exception:
return None
try:
- return int(value)
+ return int(cast(Any, value))
except (TypeError, ValueError):
return None
diff --git a/vllm_omni/experimental/fullduplex/minicpmo45/stage0.py b/vllm_omni/experimental/fullduplex/minicpmo45/stage0.py
index 392484c98..fbea82188 100644
--- a/vllm_omni/experimental/fullduplex/minicpmo45/stage0.py
+++ b/vllm_omni/experimental/fullduplex/minicpmo45/stage0.py
@@ -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
@@ -42,6 +45,20 @@ class _MiniCPMO45Stage0SessionState:
class MiniCPMO45Stage0DuplexRuntime:
"""Build scheduler-owned MiniCPM-o 4.5 Stage0 duplex inputs."""
+ unit_token_id: int = -1
+ unit_end_token_id: int = -1
+ listen_token_id: int = -1
+ speak_token_id: int = -1
+ tts_bos_token_id: int = -1
+ tts_eos_token_id: int = -1
+ tts_pad_token_id: int = -1
+ chunk_eos_token_id: int = -1
+ chunk_tts_eos_token_id: int = -1
+ turn_eos_token_id: int = -1
+ audio_placeholder_token_id: int = -1
+ image_start_token_id: int = -1
+ image_end_token_id: int = -1
+
def __init__(self, stage_model: Any, *, model_path: str | None = None, device: str = "cuda") -> None:
self.stage_model = stage_model
self.model_path = model_path
@@ -81,7 +98,7 @@ class MiniCPMO45Stage0DuplexRuntime:
processor._streaming_mel_processor = copy.deepcopy(shared_mel)
state.streaming_processor = processor
if processor is None:
- return
+ return None
set_streaming_mode = getattr(processor, "set_streaming_mode", None)
if callable(set_streaming_mode):
set_streaming_mode(
@@ -175,24 +192,40 @@ class MiniCPMO45Stage0DuplexRuntime:
self._require_special_token_ids()
if audio_waveform is None or len(audio_waveform) == 0:
return self._stage_prefill_result(False, start_time, "empty audio")
- # Omni duplex: encode this append's camera frames up front so the unit
- # loop can interleave one block per unit, mirroring official
- # streaming_prefill (feed , then image embeds, then audio).
- frame_blocks: list[Any] = []
- if video_frames:
- self._require_vision_token_ids()
- frame_blocks = self._stage_vision_embeddings(video_frames)
- if frame_blocks is None or len(frame_blocks) != len(video_frames):
- return self._stage_prefill_result(False, start_time, "streaming vision embedding failed")
state.audio_buffer = np.concatenate([state.audio_buffer, np.asarray(audio_waveform, dtype=np.float32)])
chunk_size = self._streaming_chunk_size(processor)
self._pad_first_audio_chunk_if_needed(state, processor)
if len(state.audio_buffer) < chunk_size:
+ if video_frames:
+ # Frames belong to the unit this append closes, and the
+ # scheduler already reserved their vision tokens against this
+ # append. A frame arriving before its unit can close has
+ # nowhere to go, so report it instead of dropping it under a
+ # generic buffering result.
+ return self._stage_prefill_result(
+ False,
+ start_time,
+ f"{len(video_frames)} video frame(s) on an append that closes no model unit "
+ f"(need {chunk_size} samples, only {len(state.audio_buffer)}); "
+ "send frames on unit boundaries",
+ )
return self._stage_prefill_result(
False,
start_time,
f"audio not enough: need {chunk_size} samples, only {len(state.audio_buffer)}",
)
+ # Omni duplex: encode this append's camera frames so the first unit can
+ # carry them, mirroring official streaming_prefill (feed , then
+ # every frame_list entry as its own block, then the audio).
+ frame_blocks: list[Any] = []
+ if video_frames:
+ self._require_vision_token_ids()
+ encoded_frames = self._stage_vision_embeddings(video_frames)
+ if encoded_frames is None or len(encoded_frames) != len(video_frames):
+ return self._stage_prefill_result(False, start_time, "streaming vision embedding failed")
+ if any(not slices for slices in encoded_frames):
+ return self._stage_prefill_result(False, start_time, "streaming vision embedding failed")
+ frame_blocks = encoded_frames
embed_parts: list[Any] = []
token_ids: list[int] = []
@@ -246,16 +279,22 @@ class MiniCPMO45Stage0DuplexRuntime:
embed_parts.append(self._embed_token(self.unit_token_id))
token_ids.append(self.unit_token_id)
if frame_blocks:
- # Official order inside a unit: + 64 resampler
- # embeddings + ahead of the unit's audio embeddings
- # (max_slice_nums=1 in streaming, one frame per unit).
- vision_block = self._as_2d_tensor(frame_blocks.pop(0))
- embed_parts.append(self._embed_token(self.image_start_token_id))
- token_ids.append(int(self.image_start_token_id))
- embed_parts.append(vision_block)
- token_ids.extend([self._vision_embedding_placeholder_token_id()] * int(vision_block.shape[0]))
- embed_parts.append(self._embed_token(self.image_end_token_id))
- token_ids.append(int(self.image_end_token_id))
+ # Official order inside a unit: every frame_list entry as
+ # + 64 embeds + , then any HD patches as
+ # + 64 embeds + , all ahead of this unit's
+ # *one* second of audio. stack_frames never stacks audio.
+ vision_placeholder = self._vision_embedding_placeholder_token_id()
+ for frame_slices in frame_blocks:
+ for slice_index, frame_block in enumerate(frame_slices):
+ vision_block = self._as_2d_tensor(frame_block)
+ start_id, end_id = self._vision_slice_token_ids(is_source=slice_index == 0)
+ embed_parts.append(self._embed_token(start_id))
+ token_ids.append(int(start_id))
+ embed_parts.append(vision_block)
+ token_ids.extend([vision_placeholder] * int(vision_block.shape[0]))
+ embed_parts.append(self._embed_token(end_id))
+ token_ids.append(int(end_id))
+ frame_blocks = []
embed_parts.append(audio_embeds)
token_ids.extend(
[self._audio_embedding_placeholder_token_id()] * int(self._as_2d_tensor(audio_embeds).shape[0])
@@ -264,14 +303,6 @@ class MiniCPMO45Stage0DuplexRuntime:
state.audio_chunk_idx += 1
units_built += 1
chunk_size = self._streaming_chunk_size(processor)
- if frame_blocks:
- # Each frame reserves one image block in the append plan. A frame
- # without a matching audio unit would desynchronize that plan.
- return self._stage_prefill_result(
- False,
- start_time,
- f"{len(frame_blocks)} video frame(s) left without a matching audio unit",
- )
# Match official streaming_prefill: per chunk feed ONLY +audio. The assistant
# turn is opened once at session init; re-emitting the turn-open prefix per chunk
# re-opened the turn each chunk -> degenerate repetition. tts_bos/listen/turn_eos are
@@ -331,6 +362,20 @@ class MiniCPMO45Stage0DuplexRuntime:
embeds = embedder(token)
return self._as_2d_tensor(embeds)
+ def _token_embedding_dtype(self) -> Any:
+ """dtype of the decoder token embeddings, the unit's reference dtype.
+
+ Official ``get_vllm_embedding`` casts vision hidden states to the token
+ embedding dtype before they enter the sequence, so the vision tower's
+ own dtype must not leak into the concatenated unit.
+ """
+ embedder = self._token_embedder()
+ weight = getattr(embedder, "weight", None)
+ dtype = getattr(weight, "dtype", None)
+ if dtype is not None:
+ return dtype
+ return getattr(next(self.thinker.parameters()), "dtype", None)
+
def _token_embedder(self) -> Any:
nested_embed = getattr(getattr(getattr(self.thinker, "llm", None), "model", None), "embed_tokens", None)
if callable(nested_embed):
@@ -392,7 +437,7 @@ class MiniCPMO45Stage0DuplexRuntime:
)
if len(state.audio_buffer) >= first_chunk_samples:
return
- padding = np.zeros(first_chunk_samples - len(state.audio_buffer), dtype=np.float32)
+ padding: np.ndarray = np.zeros(first_chunk_samples - len(state.audio_buffer), dtype=np.float32)
state.audio_buffer = np.concatenate([padding, state.audio_buffer])
def _stage_param(self, name: str, default: Any) -> Any:
@@ -454,7 +499,7 @@ class MiniCPMO45Stage0DuplexRuntime:
previous_audio_past_key_values = (
getattr(self.thinker, "audio_past_key_values", None) if has_audio_cache else None
)
- if has_audio_cache:
+ if has_audio_cache and state is not None:
self.thinker.audio_past_key_values = state.audio_past_key_values
try:
for target in (self.stage_model, self.thinker):
@@ -469,18 +514,18 @@ class MiniCPMO45Stage0DuplexRuntime:
suffix_extra_frames=2,
)
)
- if has_audio_cache:
+ if has_audio_cache and state is not None:
state.audio_past_key_values = getattr(self.thinker, "audio_past_key_values", None)
return result
except TypeError:
result = self._cat_nested_tensors(get_streaming(batch_feature))
- if has_audio_cache:
+ if has_audio_cache and state is not None:
state.audio_past_key_values = getattr(self.thinker, "audio_past_key_values", None)
return result
get_hidden = getattr(target, "get_audio_hidden_states", None)
if callable(get_hidden):
result = self._cat_nested_tensors(get_hidden(batch_feature))
- if has_audio_cache:
+ if has_audio_cache and state is not None:
state.audio_past_key_values = getattr(self.thinker, "audio_past_key_values", None)
return result
return None
@@ -627,14 +672,14 @@ class MiniCPMO45Stage0DuplexRuntime:
return original_register(config_class, *args, **kwargs)
with _MINICPMO45_PROCESSOR_LOAD_LOCK:
- AutoImageProcessor.register = staticmethod(register_image_processor)
+ setattr(AutoImageProcessor, "register", staticmethod(register_image_processor))
try:
processor = AutoProcessor.from_pretrained(
model_path,
trust_remote_code=True,
)
finally:
- AutoImageProcessor.register = staticmethod(original_register)
+ setattr(AutoImageProcessor, "register", staticmethod(original_register))
except Exception as exc:
raise RuntimeError(f"Failed to load MiniCPM-o duplex processor from {model_path!r}") from exc
if getattr(processor, "tokenizer", None) is None:
@@ -764,21 +809,24 @@ class MiniCPMO45Stage0DuplexRuntime:
f"missing or unknown: {', '.join(missing)}"
)
- def _stage_vision_embeddings(self, frames: list[Any]) -> list[Any] | None:
- """Encode camera frames for omni duplex via the loaded vision tower.
+ def _vision_slice_token_ids(self, *, is_source: bool) -> tuple[int, int]:
+ if is_source:
+ return int(self.image_start_token_id), int(self.image_end_token_id)
+ start = getattr(self, "slice_start_token_id", None)
+ end = getattr(self, "slice_end_token_id", None)
+ if isinstance(start, int) and start >= 0 and isinstance(end, int) and end >= 0:
+ return start, end
+ return int(self.image_start_token_id), int(self.image_end_token_id)
- Semantics mirror ``MiniCPMODuplex.streaming_prefill`` (one 64-embedding
- block per frame at ``max_slice_nums=1``), executed through the vLLM
- wrapper's ``get_vision_hidden_states`` (vpm + resampler) since the
- stage model does not expose the remote-code ``get_vision_embedding``.
- """
- process_image = getattr(self.processor, "process_image", None)
- if not callable(process_image):
- return None
- try:
- processed = process_image(frames, max_slice_nums=1)
- except Exception: # noqa: BLE001 - prefill fails with a reason
- return None
+ def _official_max_slice_nums(self, frame_count: int) -> list[int]:
+ """Official stacked pair uses HD on the current frame only: ``[2, 1]``."""
+ if frame_count <= 0:
+ return []
+ if frame_count == 1:
+ return [1]
+ return [2] + [1] * (frame_count - 1)
+
+ def _encode_processed_vision(self, processed: Any) -> list[Any] | None:
targets = (self.stage_model, self.thinker, getattr(self.stage_model, "model", None))
for target in targets:
if target is None:
@@ -801,17 +849,48 @@ class MiniCPMO45Stage0DuplexRuntime:
flat_pixels.append(slice_pixels.to(device=device, dtype=dtype))
tgt_tensor = image_tgt if hasattr(image_tgt, "reshape") else torch.tensor(image_tgt)
flat_tgt.append(tgt_tensor.reshape(-1, 2))
+ if not flat_pixels:
+ return None
tgt_sizes = torch.cat(flat_tgt, dim=0).to(device=device, dtype=torch.int64)
with torch.no_grad():
hidden = get_hidden({"pixel_values": flat_pixels, "tgt_sizes": tgt_sizes})
except Exception: # noqa: BLE001 - prefill fails with a reason
return None
expected = MiniCPMO45DuplexPolicy.VISION_EMBEDS_PER_FRAME
+ embed_dtype = self._token_embedding_dtype()
out: list[Any] = []
for block in hidden:
block_2d = self._as_2d_tensor(block)
if int(block_2d.shape[0]) != expected:
return None
+ if embed_dtype is not None and block_2d.dtype != embed_dtype:
+ block_2d = block_2d.to(dtype=embed_dtype)
out.append(block_2d)
return out
return None
+
+ def _stage_vision_embeddings(self, frames: list[Any]) -> list[list[Any]] | None:
+ """Encode camera frames for omni duplex via the loaded vision tower.
+
+ Official ``streaming_prefill`` encodes each ``frame_list`` entry as its
+ own image (source + optional HD slices) and never stacks audio: the
+ unit still carries one second of soundtrack. A stacked pair uses
+ ``max_slice_nums=[2, 1]`` so the current frame keeps the elevator
+ display readable and the composite stays a single 448-normalized tile.
+ Frames are processed one at a time because ``process_image([a, b])``
+ packs both PILs into one batch item.
+ """
+ process_image = getattr(self.processor, "process_image", None)
+ if not callable(process_image):
+ return None
+ out: list[list[Any]] = []
+ for frame, max_slices in zip(frames, self._official_max_slice_nums(len(frames))):
+ try:
+ processed = process_image([frame], max_slice_nums=max_slices)
+ except Exception: # noqa: BLE001 - prefill fails with a reason
+ return None
+ encoded = self._encode_processed_vision(processed)
+ if encoded is None:
+ return None
+ out.append(encoded)
+ return out
diff --git a/vllm_omni/experimental/fullduplex/openai/realtime_input.py b/vllm_omni/experimental/fullduplex/openai/realtime_input.py
index bd8e8270b..3d77a4172 100644
--- a/vllm_omni/experimental/fullduplex/openai/realtime_input.py
+++ b/vllm_omni/experimental/fullduplex/openai/realtime_input.py
@@ -6,6 +6,7 @@ from __future__ import annotations
import asyncio
import base64
import binascii
+from collections.abc import Callable
from typing import Any
from uuid import uuid4
@@ -17,6 +18,7 @@ from vllm_omni.experimental.fullduplex.openai.audio import (
from vllm_omni.experimental.fullduplex.openai.realtime_state import (
REALTIME_INPUT_AUDIO_FORMATS,
REALTIME_OUTPUT_AUDIO_FORMATS,
+ RealtimeStateOwner,
)
from vllm_omni.experimental.fullduplex.openai.vad import (
SILERO_VAD_MIN_THRESHOLD,
@@ -27,9 +29,42 @@ from vllm_omni.experimental.fullduplex.openai.vad import (
)
-class RealtimeInputTranslator:
+class RealtimeInputTranslator(RealtimeStateOwner):
"""Translate and validate client Realtime events for the duplex core."""
+ _opened: bool
+ _initial_session_update: bool
+ _input_speech_started: bool
+ _input_audio_buffer_has_audio: bool
+ _input_audio_buffer_had_non_speech: bool
+ _input_audio_buffer_transcript_parts: list[str]
+ _active_input_item_id: str | None
+ _last_conversation_item_id: str | None
+ _active_response_id: str | None
+ _last_response_id: str | None
+ _input_audio_format: str
+ _output_audio_format: str
+ _input_sample_rate_hz: int
+ _conversation_items: dict[str, dict[str, object]]
+ _item_truncation_cursors: dict[str, tuple[int, int]]
+ _pending_outbound: asyncio.Queue[dict[str, object]]
+ _pending_commit_item_ids: asyncio.Queue[str]
+
+ async def _send_realtime_payload(self, payload: dict[str, object]) -> None:
+ raise NotImplementedError
+
+ def _realtime_error_payload(
+ self,
+ code: str,
+ message: str,
+ *,
+ event_id: object | None = None,
+ param: object | None = None,
+ ) -> dict[str, object]:
+ raise NotImplementedError
+
+ _response_is_done: Callable[[object], bool]
+
async def discard_pending_input_audio(
self,
*,
@@ -77,7 +112,8 @@ class RealtimeInputTranslator:
async def _to_duplex_event(self, event: dict[str, object]) -> dict[str, object] | None:
event_type = event.get("type")
if event_type == "session.update":
- session_payload = event.get("session") if isinstance(event.get("session"), dict) else event
+ session = event.get("session")
+ session_payload: dict[str, object] = session if isinstance(session, dict) else event
format_error = self._validate_realtime_session_audio_formats(session_payload)
if format_error is not None:
await self._send_realtime_payload(
@@ -201,7 +237,7 @@ class RealtimeInputTranslator:
)
return None
item = self._conversation_items.get(item_id)
- if item is None:
+ if not isinstance(item, dict):
await self._send_realtime_payload(
self._realtime_error_payload(
"item_not_found",
@@ -357,7 +393,8 @@ class RealtimeInputTranslator:
)
)
return None
- payload["video_frames"] = [frame for frame in video_frames if isinstance(frame, str) and frame]
+ if isinstance(video_frames, list):
+ payload["video_frames"] = [frame for frame in video_frames if isinstance(frame, str) and frame]
self._copy_realtime_input_hints(event, payload)
payload["is_speech"] = looks_like_speech
if vad_result is not None:
@@ -627,7 +664,8 @@ class RealtimeInputTranslator:
def _remove_conversation_item(self, item_id: str) -> bool:
removed = self._conversation_items.pop(item_id, None) is not None
if self._last_conversation_item_id == item_id:
- self._last_conversation_item_id = next(reversed(self._conversation_items), None)
+ remaining_ids = list(self._conversation_items)
+ self._last_conversation_item_id = remaining_ids[-1] if remaining_ids else None
self._item_truncation_cursors.pop(item_id, None)
return removed
@@ -713,29 +751,34 @@ class RealtimeInputTranslator:
audio_config = session_payload.get("audio")
audio_input = audio_config.get("input") if isinstance(audio_config, dict) else None
audio_output = audio_config.get("output") if isinstance(audio_config, dict) else None
- extra_body = (
- dict(session_payload.get("extra_body")) if isinstance(session_payload.get("extra_body"), dict) else {}
- )
+ extra_body_payload = session_payload.get("extra_body")
+ extra_body = dict(extra_body_payload) if isinstance(extra_body_payload, dict) else {}
extra_body["realtime_session_payload"] = self._json_safe_realtime_payload(session_payload)
if isinstance(session_payload.get("tools"), list):
extra_body["realtime_tools"] = session_payload["tools"]
if isinstance(session_payload.get("tool_choice"), str | dict):
extra_body["realtime_tool_choice"] = session_payload["tool_choice"]
- if isinstance(session_payload.get("metadata"), dict):
- extra_body["realtime_metadata"] = dict(session_payload["metadata"])
- if isinstance(session_payload.get("include"), list):
- extra_body["realtime_include"] = list(session_payload["include"])
- if isinstance(session_payload.get("prompt"), dict):
- extra_body["realtime_prompt"] = dict(session_payload["prompt"])
+ metadata = session_payload.get("metadata")
+ if isinstance(metadata, dict):
+ extra_body["realtime_metadata"] = dict(metadata)
+ include = session_payload.get("include")
+ if isinstance(include, list):
+ extra_body["realtime_include"] = list(include)
+ prompt = session_payload.get("prompt")
+ if isinstance(prompt, dict):
+ extra_body["realtime_prompt"] = dict(prompt)
input_audio_transcription = self._input_audio_transcription_config(session_payload)
if isinstance(input_audio_transcription, dict):
extra_body["realtime_input_audio_transcription"] = dict(input_audio_transcription)
- if isinstance(session_payload.get("input_audio_noise_reduction"), dict):
- extra_body["realtime_input_audio_noise_reduction"] = dict(session_payload["input_audio_noise_reduction"])
- if isinstance(audio_input, dict) and isinstance(audio_input.get("noise_reduction"), dict):
- extra_body["realtime_input_audio_noise_reduction"] = dict(audio_input["noise_reduction"])
- if isinstance(session_payload.get("audio"), dict):
- extra_body["realtime_audio"] = dict(session_payload["audio"])
+ noise_reduction = session_payload.get("input_audio_noise_reduction")
+ if isinstance(noise_reduction, dict):
+ extra_body["realtime_input_audio_noise_reduction"] = dict(noise_reduction)
+ audio_noise_reduction = audio_input.get("noise_reduction") if isinstance(audio_input, dict) else None
+ if isinstance(audio_noise_reduction, dict):
+ extra_body["realtime_input_audio_noise_reduction"] = dict(audio_noise_reduction)
+ audio = session_payload.get("audio")
+ if isinstance(audio, dict):
+ extra_body["realtime_audio"] = dict(audio)
if isinstance(session_payload.get("tracing"), str | dict):
extra_body["realtime_tracing"] = session_payload["tracing"]
response_format = self._duplex_response_format(self._output_audio_format)
@@ -1032,10 +1075,14 @@ class RealtimeInputTranslator:
def _validate_realtime_video_frames(video_frames: object, max_slice_nums: object) -> str | None:
"""Validate omni-duplex camera frames on input_audio_buffer.append.
- Wire contract matches the official MiniCPM-o-Demo omni client: one
- base64 JPEG per ~1 s audio chunk. HD slicing (max_slice_nums > 1) is
- not implemented by the duplex adapter yet and is rejected explicitly
- rather than silently ignored.
+ Wire contract matches the official MiniCPM-o duplex loop: one base
+ base64 JPEG per ~1 s audio chunk, optionally followed by that unit's
+ stacked composite tiling the sub-frames captured inside it (at most 2
+ images either way). HD slicing (max_slice_nums > 1) is not implemented
+ by the duplex adapter yet and is rejected explicitly rather than
+ silently ignored; note official suggests it for composites
+ (``max_slice_nums=[2, 1]``), so stacked detail is capped at
+ scale_resolution until that lands.
"""
if max_slice_nums not in (None, 1):
return "max_slice_nums > 1 (HD slicing) is not implemented by the duplex Realtime adapter"
@@ -1416,10 +1463,11 @@ class RealtimeInputTranslator:
self._input_speech_started = True
if self._active_input_item_id is None:
self._active_input_item_id = f"item_{uuid4().hex}"
+ audio_start_ms = event.get("audio_start_ms", 0)
await self._send_realtime_payload(
{
"type": "input_audio_buffer.speech_started",
- "audio_start_ms": int(event.get("audio_start_ms", 0) or 0),
+ "audio_start_ms": int(audio_start_ms) if isinstance(audio_start_ms, int | float) else 0,
"item_id": self._active_input_item_id,
}
)
diff --git a/vllm_omni/experimental/fullduplex/video_stacking.py b/vllm_omni/experimental/fullduplex/video_stacking.py
new file mode 100644
index 000000000..2a7706759
--- /dev/null
+++ b/vllm_omni/experimental/fullduplex/video_stacking.py
@@ -0,0 +1,134 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""Intra-unit frame stacking for the omni duplex video track.
+
+Official MiniCPM-o duplex raises the visual refresh rate without touching the
+audio cadence: ``get_video_frame_audio_segments(stack_frames=N)`` samples the
+``N-1`` sub-frames captured inside each second (skipping the one that would
+duplicate the second's base frame) and tiles them into a *single* composite
+image. The duplex loop then feeds ``frame_list=[base, composite]`` alongside
+that second's audio, so a unit still carries one second of audio and at most
+two images however large ``N`` is.
+
+Tiling therefore belongs to media preprocessing, not to the wire: a caller
+turns its sub-frames into one composite here and sends it as the second frame
+of the append that closes the unit.
+"""
+
+from __future__ import annotations
+
+import io
+from collections.abc import Sequence
+from typing import Any
+
+import pybase64 as base64
+
+_BG_COLOR = (255, 255, 255)
+_LINE_COLOR = (0, 0, 0)
+_LINE_WIDTH = 6
+
+
+def _as_image(frame: Any) -> Any:
+ """Coerce a PIL image, encoded bytes, or base64/data-URL string to PIL RGB."""
+ from PIL import Image
+
+ if isinstance(frame, Image.Image):
+ return frame
+ if isinstance(frame, (bytes, bytearray)):
+ return Image.open(io.BytesIO(bytes(frame))).convert("RGB")
+ if isinstance(frame, str):
+ payload = frame.split(",")[-1] if ";base64," in frame else frame
+ return Image.open(io.BytesIO(base64.b64decode(payload))).convert("RGB")
+ raise TypeError(f"unsupported frame type for stacking: {type(frame)}")
+
+
+def _grid_shape(images: Sequence[Any], *, cell_w: int, cell_h: int) -> tuple[int, int]:
+ """Rows and columns for ``images``, matching official ``concat_images``.
+
+ 4 frames tile 2x2; 2 and 3 frames pick the row/column arrangement whose
+ canvas is closest to square, so a landscape clip stacks vertically and a
+ portrait one horizontally; anything else falls back to a single row.
+ """
+ count = len(images)
+ if count == 4:
+ return 2, 2
+ if count == 1:
+ return 1, 1
+ if count not in (2, 3):
+ return 1, count
+
+ candidates = [(1, count), (count, 1)]
+
+ def squareness(rows: int, cols: int) -> float:
+ width = cols * cell_w + (cols - 1) * _LINE_WIDTH
+ height = rows * cell_h + (rows - 1) * _LINE_WIDTH
+ return abs(width / max(1, height) - 1.0)
+
+ ratios = [squareness(rows, cols) for rows, cols in candidates]
+ if count == 2 and ratios[0] == ratios[1]:
+ average_aspect = sum(im.width / max(1, im.height) for im in images) / count
+ return (1, 2) if average_aspect >= 1.0 else (2, 1)
+ return candidates[min(range(len(candidates)), key=lambda index: ratios[index])]
+
+
+def _letterbox(image: Any, target_w: int, target_h: int) -> Any:
+ from PIL import Image
+
+ image = image.convert("RGB")
+ width, height = image.size
+ scale = min(target_w / width, target_h / height)
+ new_size = (max(1, round(width * scale)), max(1, round(height * scale)))
+ resized = image.resize(new_size, Image.Resampling.BICUBIC)
+ canvas = Image.new("RGB", (target_w, target_h), _BG_COLOR)
+ canvas.paste(resized, ((target_w - new_size[0]) // 2, (target_h - new_size[1]) // 2))
+ return canvas
+
+
+def concat_frames(frames: Sequence[Any]) -> Any:
+ """Tile a unit's sub-frames into one PIL image, as official stacking does.
+
+ Cells are the largest input width and height, each frame is letterboxed to
+ keep its aspect ratio, and a separator band is drawn only in the interior
+ seams so the model can tell the sub-frames apart.
+ """
+ from PIL import Image
+
+ images = [_as_image(frame) for frame in frames]
+ if not images:
+ raise ValueError("cannot stack an empty frame list")
+
+ cell_w = max(image.width for image in images)
+ cell_h = max(image.height for image in images)
+ rows, cols = _grid_shape(images, cell_w=cell_w, cell_h=cell_h)
+
+ canvas = Image.new(
+ "RGB",
+ (cols * cell_w + (cols - 1) * _LINE_WIDTH, rows * cell_h + (rows - 1) * _LINE_WIDTH),
+ _LINE_COLOR,
+ )
+ for index, image in enumerate(images[: rows * cols]):
+ row, col = divmod(index, cols)
+ canvas.paste(
+ _letterbox(image, cell_w, cell_h),
+ (col * (cell_w + _LINE_WIDTH), row * (cell_h + _LINE_WIDTH)),
+ )
+ return canvas
+
+
+def concat_frames_b64(frames: Sequence[Any], *, quality: int = 95) -> str:
+ """``concat_frames`` re-encoded as a base64 JPEG, ready for the wire."""
+ buf = io.BytesIO()
+ concat_frames(frames).save(buf, format="JPEG", quality=quality)
+ return base64.b64encode(buf.getvalue()).decode("ascii")
+
+
+def unit_subframe_offsets(stack_frames: int, *, unit_s: float = 1.0) -> list[float]:
+ """Sub-frame capture offsets inside a unit, in seconds from its start.
+
+ ``stack_frames=5`` yields 0.2/0.4/0.6/0.8 s for a 1 s unit. Offset 0 is
+ skipped because official skips it: it is the unit's base frame, already
+ sent as ``frame_list[0]``.
+ """
+ count = max(1, int(stack_frames))
+ return [index / count * unit_s for index in range(1, count)]