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

[Bugfix][MiniCPM-o] Bind omni-duplex video frames to the unit they close (#6404)

Browse Source 
Signed-off-by: amy-why-3459 <wuhaiyan17@huawei.com>
This commit is contained in:
amy-why-3459 authored and GitHub committed 2026-08-28 16:58:20 +08:00
1 parent 13173d6bbb
commit da94c3ebe1
17 files changed
+1530 -144
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

examples/online_serving/minicpmo/README.md
+27
View File
@@ -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
examples/online_serving/minicpmo/realtime_duplex_demo.py
+305 -8
View File
@@ -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,
tests/benchmarks/test_daily_omni_pack_modes.py
+8 -1
View File
@@ -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)
tests/e2e/features/fullduplex/test_client.py
+63
View File
@@ -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"]]
tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py
+15
View File
@@ -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",
tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py
+151 -28
View File
@@ -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",
tests/e2e/online_serving/test_minicpmo_4_5_duplex.py
+46 -1
View File
@@ -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
tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py
+226
View File
@@ -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()
tests/examples/test_minicpmo_realtime_duplex_simple_demo.py
+61
View File
@@ -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)
tests/worker/test_native_duplex_hooks.py
+159 -1
View File
@@ -1,3 +1,6 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
from __future__ import annotations
import 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: {
"<unit>": 1,
"</unit>": 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,
"<image>": 12,
"</image>": 13,
"<slice>": 14,
"</slice>": 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
# <unit>, then each frame as <image> + embeds + </image>, 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
# <unit> <image> src </image> <slice> p0 </slice> <slice> p1 </slice>
# <image> stack </image> 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)
vllm_omni/benchmarks/data_modules/daily_omni_dataset.py
+14 -5
View File
@@ -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))]
vllm_omni/experimental/fullduplex/client.py
+78 -10
View File
@@ -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})
vllm_omni/experimental/fullduplex/minicpmo45/policy.py
+11 -3
View File
@@ -1,3 +1,6 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
from __future__ import annotations
from typing import Any
@@ -24,7 +27,9 @@ class MiniCPMO45DuplexPolicy:
# Vision framing contract (omni duplex). Official streaming_prefill feeds
# each frame as <image> + 64 resampler embeddings + </image> 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 # <image> + embeds + </image>
DEFAULT_MAX_NEW_SPEAK_TOKENS_PER_CHUNK = 20
@@ -77,6 +82,8 @@ class MiniCPMO45DuplexPolicy:
"audio_placeholder_token_id": "<|audio|>",
"image_start_token_id": "<image>",
"image_end_token_id": "</image>",
"slice_start_token_id": "<slice>",
"slice_end_token_id": "</slice>",
}
@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
vllm_omni/experimental/fullduplex/minicpmo45/runtime.py
+31 -13
View File
@@ -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
# <image> + 64 resampler embeddings + </image> 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
vllm_omni/experimental/fullduplex/minicpmo45/stage0.py
+128 -49
View File
@@ -1,3 +1,6 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
from __future__ import annotations
import base64
@@ -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 <image> block per unit, mirroring official
# streaming_prefill (feed <unit>, 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 <unit>, then
# every frame_list entry as its own <image> 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: <image> + 64 resampler
# embeddings + </image> 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
# <image> + 64 embeds + </image>, then any HD patches as
# <slice> + 64 embeds + </slice>, 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 <unit>+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
vllm_omni/experimental/fullduplex/openai/realtime_input.py
+73 -25
View File
@@ -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,
}
)
vllm_omni/experimental/fullduplex/video_stacking.py
+134
View File
@@ -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)]