From 5ce6f5dc49ead82b24bb5e1c19849cea67c51d3e Mon Sep 17 00:00:00 2001 From: chickeyton Date: Mon, 28 Sep 2026 19:11:42 +0800 Subject: [PATCH] [Bugfix][MiniCPM-o] Forward a speech unit closed by LISTEN after turn_eos to the Talker (#8227) Signed-off-by: chickeyton Co-authored-by: Claude Opus 5.5 (1M context) --- .../duplex/test_turn_eos_listen_unit.py | 93 +++++++++++++++++++ .../models/minicpmo_4_5/duplex/data_plane.py | 7 +- .../models/minicpmo_4_5/duplex/plugin.py | 8 ++ .../models/minicpmo_4_5/duplex/policy.py | 29 ++++++ 4 files changed, 136 insertions(+), 1 deletion(-) create mode 100644 tests/model_executor/models/minicpmo_4_5/duplex/test_turn_eos_listen_unit.py diff --git a/tests/model_executor/models/minicpmo_4_5/duplex/test_turn_eos_listen_unit.py b/tests/model_executor/models/minicpmo_4_5/duplex/test_turn_eos_listen_unit.py new file mode 100644 index 000000000..b129467b7 --- /dev/null +++ b/tests/model_executor/models/minicpmo_4_5/duplex/test_turn_eos_listen_unit.py @@ -0,0 +1,93 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""A unit that closes its speech with ``<|turn_eos|>`` and then LISTEN is speech, not a LISTEN decision. + +On silent input the policy forces the unit terminator after ``<|turn_eos|>`` to +be LISTEN. Treating that unit as a model LISTEN keeps its final speech and +``<|turn_eos|>`` from the Talker, so the response never ends. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from vllm_omni.model_executor.models.minicpmo_4_5.duplex.data_plane import MiniCPMO45DataPlaneSession +from vllm_omni.model_executor.models.minicpmo_4_5.duplex.plugin import MiniCPMO45DuplexPlugin +from vllm_omni.model_executor.models.minicpmo_4_5.duplex.policy import MiniCPMO45DuplexPolicy + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + +CHUNK_EOS, CHUNK_TTS_EOS, LISTEN, TURN_EOS = 5, 6, 7, 9 +SPECIAL_TOKEN_IDS = { + "chunk_eos_token_id": CHUNK_EOS, + "chunk_tts_eos_token_id": CHUNK_TTS_EOS, + "listen_token_id": LISTEN, + "turn_eos_token_id": TURN_EOS, +} +TEXT = 100 + + +@pytest.mark.parametrize( + ("token_ids", "closes_speech"), + [ + ([TEXT, TEXT, TURN_EOS, LISTEN], True), + ([TURN_EOS, LISTEN], True), + ([LISTEN, LISTEN, TEXT, TURN_EOS, LISTEN], True), + ([LISTEN], False), + ([TEXT, TEXT, LISTEN], False), + # A <|turn_eos|> from an earlier unit does not count. + ([TEXT, TURN_EOS, CHUNK_EOS, LISTEN], False), + ([TEXT, TURN_EOS, LISTEN, LISTEN], False), + ([TEXT, TURN_EOS, CHUNK_EOS], False), + ([], False), + ], +) +def test_speech_unit_closed_by_listen(token_ids: list[int], closes_speech: bool) -> None: + assert MiniCPMO45DuplexPolicy.speech_unit_closed_by_listen(token_ids, SPECIAL_TOKEN_IDS) is closes_speech + + +def _stage0_output(token_ids: list[int], *, finished: bool = True) -> SimpleNamespace: + completion = SimpleNamespace(token_ids=token_ids, stop_reason=token_ids[-1], text="", multimodal_output={}) + return SimpleNamespace( + request_id="duplex-s.c2Vzc2lvbg==.e.0.r.stage0", + outputs=[completion], + finished=finished, + multimodal_output={"special_token_ids": dict(SPECIAL_TOKEN_IDS)}, + ) + + +def _decide(token_ids: list[int]): + return MiniCPMO45DuplexPlugin(lambda *args: None).decide_output( + stage_id=0, + final_stage_id=2, + segment_finished=True, + segment_token_ids=tuple(token_ids), + segment_output_metadata={}, + output=_stage0_output(token_ids), + ) + + +def test_plugin_forwards_speech_closed_by_listen_to_the_talker() -> None: + assert _decide([TEXT, TEXT, TURN_EOS, LISTEN]) is None + + +def test_plugin_keeps_a_plain_listen_unit_a_listen_decision() -> None: + decision = _decide([LISTEN]) + + assert decision is not None + assert decision.metadata["duplex_native_decision"] == "listen" + + +def _projected_model_listens(token_ids: list[int]) -> list[dict[str, object]]: + session = MiniCPMO45DataPlaneSession(lambda *args, **kwargs: "") + return [result for result in session.project_output(_stage0_output(token_ids)) if result.get("model_listen")] + + +def test_data_plane_does_not_project_speech_closed_by_listen_as_model_listen() -> None: + assert _projected_model_listens([TEXT, TEXT, TURN_EOS, LISTEN]) == [] + + +def test_data_plane_projects_a_plain_listen_unit_as_model_listen() -> None: + assert len(_projected_model_listens([LISTEN])) == 1 diff --git a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/data_plane.py b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/data_plane.py index 9fe0468e7..afed64594 100644 --- a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/data_plane.py +++ b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/data_plane.py @@ -13,6 +13,7 @@ from vllm_omni.engine.duplex.contracts import ( duplex_resource_request_belongs_to_session, ) from vllm_omni.engine.duplex.plugin import DuplexDataPlane, DuplexDataPlaneContext +from vllm_omni.model_executor.models.minicpmo_4_5.duplex.policy import MiniCPMO45DuplexPolicy from vllm_omni.outputs.duplex import get_duplex_output_decision logger = init_logger(__name__) @@ -638,9 +639,13 @@ def _native_decision( return None if mm_output.get("duplex_native_decision") == "listen" or mm_output.get("model_listen") is True: return "listen" - listen_id = _special_token_ids(mm_output).get("listen_token_id") + special_token_ids = _special_token_ids(mm_output) + listen_id = special_token_ids.get("listen_token_id") if listen_id is None: return None + unit_ids = max(token_ids, _coerce_int_list(getattr(completion, "cumulative_token_ids", None)), key=len) + if MiniCPMO45DuplexPolicy.speech_unit_closed_by_listen(unit_ids, special_token_ids): + return None stop_reason = getattr(completion, "stop_reason", None) if completion is not None else None if coerce_int(stop_reason) == listen_id: return "listen" diff --git a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/plugin.py b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/plugin.py index b22e54f51..061ca14f9 100644 --- a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/plugin.py +++ b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/plugin.py @@ -639,6 +639,14 @@ class MiniCPMO45DuplexPlugin(DuplexModelPlugin): token_ids = _completion_token_ids(completion) or list(segment_token_ids) if _coerce_int(stop_reason) != listen_id and (not token_ids or token_ids[-1] != listen_id): return None + unit_ids = max( + (token_ids, _coerce_int_list(getattr(completion, "cumulative_token_ids", None)), list(segment_token_ids)), + key=len, + ) + if MiniCPMO45DuplexPolicy.speech_unit_closed_by_listen(unit_ids, special_token_ids): + # The unit's final speech and <|turn_eos|> must reach the Talker, + # or the response never ends. + return None metadata = dict(output_metadata) for key, value in special_token_ids.items(): diff --git a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/policy.py b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/policy.py index 7b0aded4d..11539af09 100644 --- a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/policy.py +++ b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/policy.py @@ -7,6 +7,8 @@ from dataclasses import dataclass from typing import TYPE_CHECKING if TYPE_CHECKING: + from collections.abc import Mapping, Sequence + from transformers import PreTrainedTokenizerBase @@ -93,6 +95,33 @@ class MiniCPMO45DuplexPolicy: DEFAULT_MIN_NEW_SPEAK_TOKENS_BEFORE_CHUNK_BOUNDARY = 8 REPETITION_HISTORY_SIZE = 512 + @staticmethod + def speech_unit_closed_by_listen(token_ids: Sequence[int], special_token_ids: Mapping[str, int]) -> bool: + """Whether a unit ending in LISTEN closes speech rather than deciding to listen. + + ``<|turn_eos|>`` ends the turn but not the unit: the model keeps + decoding until a unit terminator, and on silent input the policy forces + that terminator to be LISTEN. Such a unit still carries the turn's final + speech and ``<|turn_eos|>`` for the Talker, so it is not a LISTEN + decision. Only the current unit counts: the scan stops at the previous + unit terminator. + """ + listen_id = special_token_ids.get("listen_token_id") + turn_eos_id = special_token_ids.get("turn_eos_token_id") + if listen_id is None or turn_eos_id is None or not token_ids or token_ids[-1] != listen_id: + return False + unit_terminators = { + listen_id, + special_token_ids.get("chunk_eos_token_id"), + special_token_ids.get("chunk_tts_eos_token_id"), + } + for token_id in reversed(token_ids[:-1]): + if token_id == turn_eos_id: + return True + if token_id in unit_terminators: + return False + return False + @classmethod def audio_token_count(cls, sample_count: int) -> int: """Audio embedding count for a clip of ``sample_count`` samples.