diff --git a/tests/entrypoints/openai_api/test_realtime_prompt_preflight.py b/tests/entrypoints/openai_api/test_realtime_prompt_preflight.py index 632efb212..dc58e484e 100644 --- a/tests/entrypoints/openai_api/test_realtime_prompt_preflight.py +++ b/tests/entrypoints/openai_api/test_realtime_prompt_preflight.py @@ -3,6 +3,7 @@ from __future__ import annotations +import asyncio import base64 import io import json @@ -12,6 +13,8 @@ from typing import Any import numpy as np import pytest +from openai.types import realtime as types +from vllm.exceptions import VLLMValidationError from vllm.sampling_params import RequestOutputKind, SamplingParams from tests.helpers.serving_chat import build_serving_chat @@ -20,7 +23,10 @@ from vllm_omni.entrypoints.openai.realtime.connection import ( OpenAIFullDuplexConnection, _ResolvedResponse, ) -from vllm_omni.entrypoints.openai.realtime.session import ActiveResponse +from vllm_omni.entrypoints.openai.realtime.session import ( + ActiveResponse, + AudioFullDuplexSessionState, +) pytestmark = [pytest.mark.core_model, pytest.mark.cpu] @@ -31,9 +37,13 @@ class _FakeRenderer: def __init__(self) -> None: self.raw_prompt_lengths: list[int] = [] + self.message_counts: list[int] = [] + self.skip_mm_cache: list[bool] = [] self.tokenization_params: list[Any] = [] self.engine_inputs: list[dict[str, Any]] = [] self.conversations: list[list[dict[str, Any]]] = [] + self.token_counts: dict[int, int] = {} + self.errors: dict[int, VLLMValidationError] = {} async def render_chat_async( self, @@ -41,12 +51,24 @@ class _FakeRenderer: _chat_params: Any, tok_params: Any, ): + message_count = len(conversations[0]) + self.message_counts.append(message_count) self.conversations.append(conversations[0]) raw_prompt_length = 2 * len(conversations[0]) self.raw_prompt_lengths.append(raw_prompt_length) self.tokenization_params.append(tok_params) + error = self.errors.get(message_count) + if error is not None: + raise error # Model-side expansion makes the rendered prompt much longer than its text tokens. - engine_input = {"prompt_token_ids": [0] * (10 + 20 * raw_prompt_length)} + token_count = self.token_counts.get( + message_count, + 10 + 20 * raw_prompt_length, + ) + engine_input = { + "prompt_token_ids": [0] * token_count, + "message_count": message_count, + } self.engine_inputs.append(engine_input) return conversations, [engine_input] @@ -70,9 +92,8 @@ def _make_connection( model_config = SimpleNamespace(max_model_len=max_model_len, multimodal_config=None) async def preprocess_chat(request: Any, messages: list[dict[str, Any]], **kwargs: Any): - tok_params = kwargs.get("tok_params") - if tok_params is None: - tok_params = request.build_tok_params(model_config) + tok_params = request.build_tok_params(model_config) + renderer.skip_mm_cache.append(bool(kwargs.get("skip_mm_cache", False))) return await renderer.render_chat_async( [messages], kwargs.get("default_template_kwargs"), @@ -97,22 +118,79 @@ def _make_connection( return connection, renderer +def _message_item( + item_id: str, + text: str | None = None, +) -> types.RealtimeConversationItemUserMessage: + return types.RealtimeConversationItemUserMessage( + id=item_id, + type="message", + role="user", + content=[{"type": "input_text", "text": text or item_id}], + ) + + +def _audio_item(item_id: str, audio: bytes) -> types.RealtimeConversationItemUserMessage: + return types.RealtimeConversationItemUserMessage( + id=item_id, + type="message", + role="user", + content=[ + { + "type": "input_audio", + "audio": base64.b64encode(audio).decode("ascii"), + } + ], + ) + + +def _assistant_item(item_id: str, text: str) -> types.RealtimeConversationItemAssistantMessage: + return types.RealtimeConversationItemAssistantMessage( + id=item_id, + type="message", + role="assistant", + status="completed", + content=[{"type": "output_text", "text": text}], + ) + + +def _websocket_events(websocket: _FakeWebSocket) -> list[dict[str, Any]]: + return [json.loads(message) for message in websocket.messages] + + +def _prompt_texts(conversation: list[dict[str, Any]]) -> list[str]: + texts: list[str] = [] + for message in conversation: + content = message["content"] + if isinstance(content, str): + texts.append(content) + continue + texts.extend(part["text"] for part in content if part.get("type") == "text") + return texts + + +async def _create_response( + connection: OpenAIFullDuplexConnection, +) -> list[dict[str, Any]]: + submitted: list[dict[str, Any]] = [] + + async def run_response(_response_id: str, _response: Any, engine_input: dict[str, Any]) -> None: + submitted.append(engine_input) + connection.session.active_response = None + + connection._run_response = run_response + await connection._handle_response_create(SimpleNamespace(event_id="evt_create", response=None)) + if connection._response_task is not None: + await connection._response_task + return submitted + + @pytest.mark.asyncio async def test_preflight_truncates_by_rendered_token_count() -> None: connection, renderer = _make_connection() - first = SimpleNamespace( - id="first", - type="message", - role="user", - content=[SimpleNamespace(type="input_text", text="first")], - ) - second = SimpleNamespace( - id="second", - type="message", - role="user", - content=[SimpleNamespace(type="input_text", text="second")], - ) + first = _message_item("first") + second = _message_item("second") items = [first, second] response = _ResolvedResponse( input=items, @@ -124,16 +202,405 @@ async def test_preflight_truncates_by_rendered_token_count() -> None: metadata=None, ) - engine_input = await connection._prepare_engine_input_with_auto_truncation(response) + prompt_items = await connection._truncate_prompt_items(response) + engine_input = await connection._build_full_prompt( + tools=response.tools, + instructions=response.instructions, + items=prompt_items, + ) assert renderer.raw_prompt_lengths == [4, 2, 2] assert [len(prompt["prompt_token_ids"]) for prompt in renderer.engine_inputs] == [90, 50, 50] - assert all(params.max_total_tokens is None for params in renderer.tokenization_params) + assert all(params.max_total_tokens == 100 for params in renderer.tokenization_params) assert all(params.max_output_tokens == 0 for params in renderer.tokenization_params) - assert items == [second] + assert items == [first, second] + assert prompt_items == [second] assert engine_input is renderer.engine_inputs[-1] +@pytest.mark.asyncio +async def test_persistent_model_context_cursor_reuses_truncated_window() -> None: + connection, renderer = _make_connection() + first = _message_item("first") + second = _message_item("second") + connection.session.insert_item(first) + connection.session.insert_item(second) + response = _ResolvedResponse( + input=None, + instructions=None, + modalities=["text"], + max_output_tokens="inf", + tools=None, + tool_choice="none", + metadata=None, + ) + + prompt_items = await connection._truncate_prompt_items(response) + await connection._build_full_prompt( + tools=response.tools, + instructions=response.instructions, + items=prompt_items, + ) + connection.session.commit_model_context_items(prompt_items) + renderer.raw_prompt_lengths.clear() + + prompt_items = await connection._truncate_prompt_items(response) + + assert renderer.raw_prompt_lengths == [2] + assert prompt_items == [second] + assert connection.session.model_context_first_item_id == "second" + assert not connection.session.model_context_cursor_at_end + + +@pytest.mark.asyncio +async def test_empty_model_context_allows_a_later_appended_item() -> None: + connection, renderer = _make_connection(max_model_len=20) + response = _ResolvedResponse( + input=None, + instructions=None, + modalities=["text"], + max_output_tokens="inf", + tools=None, + tool_choice="none", + metadata=None, + ) + + prompt_items = await connection._truncate_prompt_items(response) + connection.session.commit_model_context_items(prompt_items) + connection.session.insert_item(_message_item("appended")) + + assert renderer.raw_prompt_lengths == [0] + assert prompt_items == [] + assert [item.id for item in connection.session.model_context_items()] == ["appended"] + assert connection.session.model_context_first_item_id == "appended" + assert not connection.session.model_context_cursor_at_end + + +@pytest.mark.asyncio +async def test_truncation_to_empty_selects_no_items_when_overhead_fits() -> None: + connection, renderer = _make_connection(max_model_len=20) + connection.session.insert_item(_message_item("item")) + response = _ResolvedResponse( + input=None, + instructions=None, + modalities=["text"], + max_output_tokens="inf", + tools=None, + tool_choice="none", + metadata=None, + ) + + prompt_items = await connection._truncate_prompt_items(response) + connection.session.commit_model_context_items(prompt_items) + connection.session.insert_item(_message_item("appended")) + + assert renderer.raw_prompt_lengths == [2, 0] + assert prompt_items == [] + assert [item.id for item in connection.session.model_context_items()] == ["appended"] + + +def test_model_context_cursor_moves_only_forward() -> None: + session = AudioFullDuplexSessionState() + + session.insert_item(_message_item("first")) + session.insert_item(_message_item("second")) + session.insert_item(_message_item("third")) + assert session.model_context_first_item_id == "first" + assert session.model_context_items() == session.items + + session.commit_model_context_items(session.items[1:]) + session.insert_item(_message_item("before-second"), previous_item_id="first") + assert session.model_context_first_item_id == "second" + assert [item.id for item in session.model_context_items()] == ["second", "third"] + + session.insert_item(_message_item("after-second"), previous_item_id="second") + assert session.model_context_first_item_id == "second" + assert [item.id for item in session.model_context_items()] == [ + "second", + "after-second", + "third", + ] + + session.remove_item("second") + assert session.model_context_first_item_id == "after-second" + assert [item.id for item in session.model_context_items()] == ["after-second", "third"] + + session.commit_model_context_items([]) + session.insert_item(_message_item("fourth")) + session.insert_item(_message_item("root"), previous_item_id="root") + assert session.model_context_first_item_id == "fourth" + assert not session.model_context_cursor_at_end + assert [item.id for item in session.model_context_items()] == ["fourth"] + + +@pytest.mark.asyncio +async def test_prompt_truncation_bisects_the_model_context_cursor() -> None: + connection, renderer = _make_connection() + items = [ + SimpleNamespace( + id=f"item_{index}", + type="message", + role="user", + content=[SimpleNamespace(type="input_text", text=str(index))], + ) + for index in range(8) + ] + response = _ResolvedResponse( + input=items, + instructions=None, + modalities=["text"], + max_output_tokens="inf", + tools=None, + tool_choice="none", + metadata=None, + ) + + prompt_items = await connection._truncate_prompt_items(response) + await connection._build_full_prompt( + tools=response.tools, + instructions=response.instructions, + items=prompt_items, + ) + + assert renderer.raw_prompt_lengths == [16, 8, 4, 2, 2] + assert prompt_items == [items[-1]] + + +@pytest.mark.asyncio +async def test_realtime_auto_truncation_end_to_end() -> None: + websocket = _FakeWebSocket() + connection, renderer = _make_connection(websocket=websocket) + audio = b"\x01\x00\x02\x00\x03\x00\x04\x00" + items: list[Any] = [_message_item(f"item_{index}") for index in range(7)] + items.append(_audio_item("item_audio", audio)) + for item in items: + connection.session.insert_item(item) + + submitted = await _create_response(connection) + + assert len(submitted) == 1 + assert submitted[0]["message_count"] == 1 + assert renderer.message_counts == [8, 4, 2, 1, 1] + assert renderer.skip_mm_cache == [True, True, True, True, False] + assert connection.session.model_context_first_item_id == "item_audio" + assert not connection.session.model_context_cursor_at_end + assert len(connection.session.items) == 8 + + events = _websocket_events(websocket) + assert not [event for event in events if event["type"] == "conversation.item.deleted"] + assert [event for event in events if event["type"] == "response.created"] + + await connection._handle_item_retrieve( + types.ConversationItemRetrieveEvent( + type="conversation.item.retrieve", + event_id="evt_retrieve", + item_id="item_0", + ) + ) + retrieved = _websocket_events(websocket)[-1] + assert retrieved["type"] == "conversation.item.retrieved" + assert retrieved["item"]["id"] == "item_0" + + final_conversation = renderer.conversations[-1] + assert len(final_conversation) == 1 + audio_content = final_conversation[0]["content"][0]["input_audio"] + with wave.open(io.BytesIO(base64.b64decode(audio_content["data"]))) as wav: + assert wav.readframes(wav.getnframes()) == audio + + +@pytest.mark.parametrize( + ("truncation", "max_output_tokens", "expected_texts"), + [ + ("auto", "inf", ["item_1", "item_2", "item_3"]), + ("disabled", "inf", None), + ( + types.RealtimeTruncationRetentionRatio( + type="retention_ratio", + retention_ratio=0.5, + token_limits={"post_instructions": 70}, + ), + "inf", + ["item_3"], + ), + ( + types.RealtimeTruncationRetentionRatio( + type="retention_ratio", + retention_ratio=0.3, + ), + "inf", + ["item_3"], + ), + ("auto", 25, ["item_3"]), + ], +) +@pytest.mark.asyncio +async def test_realtime_truncation_configurations( + truncation: Any, + max_output_tokens: Any, + expected_texts: list[str] | None, +) -> None: + websocket = _FakeWebSocket() + connection, renderer = _make_connection(websocket=websocket) + connection.session.config.truncation = truncation + connection.session.config.max_output_tokens = max_output_tokens + renderer.token_counts = {0: 10, 1: 30, 2: 45, 3: 50, 4: 110} + for index in range(4): + connection.session.insert_item(_message_item(f"item_{index}")) + + submitted = await _create_response(connection) + + if expected_texts is None: + assert submitted == [] + assert connection._response_task is None + assert "exceeds the model's input token limit" in websocket.messages[-1] + return + + assert len(submitted) == 1 + assert _prompt_texts(renderer.conversations[-1]) == expected_texts + assert connection.session.model_context_first_item_id == f"item_{4 - len(expected_texts)}" + + +@pytest.mark.asyncio +async def test_realtime_truncation_render_errors() -> None: + websocket = _FakeWebSocket() + connection, renderer = _make_connection(websocket=websocket) + renderer.token_counts = {0: 10} + renderer.errors = { + 2: VLLMValidationError("Prompt is too long", parameter="input_tokens"), + 1: VLLMValidationError("Prompt is too long", parameter="input_tokens"), + } + connection.session.insert_item(_message_item("first")) + connection.session.insert_item(_message_item("second")) + + submitted = await _create_response(connection) + + assert len(submitted) == 1 + assert submitted[0]["message_count"] == 0 + assert renderer.message_counts == [2, 1, 0, 0] + assert renderer.skip_mm_cache == [True, True, True, False] + assert connection.session.model_context_cursor_at_end + + websocket = _FakeWebSocket() + connection, renderer = _make_connection(websocket=websocket) + renderer.errors = { + 2: VLLMValidationError("Tool rendering failed", parameter="tools"), + } + connection.session.insert_item(_message_item("first")) + connection.session.insert_item(_message_item("second")) + + submitted = await _create_response(connection) + + assert submitted == [] + assert connection._response_task is None + assert renderer.message_counts == [2] + assert "Tool rendering failed" in websocket.messages[-1] + + +@pytest.mark.asyncio +async def test_realtime_truncation_survives_history_mutations() -> None: + websocket = _FakeWebSocket() + connection, renderer = _make_connection(websocket=websocket) + renderer.token_counts = {0: 10, 1: 30, 2: 45, 3: 50, 4: 110} + + async def add_item(item: Any, previous_item_id: str | None = None) -> None: + await connection._handle_item_create( + types.ConversationItemCreateEvent( + type="conversation.item.create", + event_id=f"evt_create_{item.id}", + item=item, + previous_item_id=previous_item_id, + ) + ) + + await add_item(_message_item("item_0")) + await add_item(_message_item("item_1")) + await add_item(_assistant_item("assistant", "assistant text")) + await add_item(_message_item("item_3")) + + submitted = await _create_response(connection) + assert len(submitted) == 1 + assert _prompt_texts(renderer.conversations[-1]) == [ + "item_1", + "assistant text", + "item_3", + ] + assert connection.session.model_context_first_item_id == "item_1" + + renderer.message_counts.clear() + renderer.skip_mm_cache.clear() + await add_item(_message_item("before"), previous_item_id="item_0") + await add_item(_message_item("after"), previous_item_id="item_3") + assert connection.session.model_context_first_item_id == "item_1" + + await connection._handle_item_truncate( + types.ConversationItemTruncateEvent( + type="conversation.item.truncate", + event_id="evt_truncate", + item_id="assistant", + content_index=0, + audio_end_ms=0, + ) + ) + truncated_part = connection.session.find_item("assistant").content[0] + assert truncated_part.type == "output_audio" + assert truncated_part.transcript == "" + + await connection._handle_item_delete( + types.ConversationItemDeleteEvent( + type="conversation.item.delete", + event_id="evt_delete", + item_id="item_1", + ) + ) + assert connection.session.model_context_first_item_id == "assistant" + + active_item = _assistant_item("active", "active text") + connection.session.insert_item(active_item) + active = ActiveResponse(response_id="resp_active", request_id="req_active") + active.item_id = active_item.id + connection.session.active_response = active + cancellation_cleanup: list[str] = [] + + async def active_response_task() -> None: + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + connection.session.remove_item(active_item.id) + connection.session.active_response = None + cancellation_cleanup.append(active_item.id) + raise + + connection._response_task = asyncio.create_task(active_response_task()) + await asyncio.sleep(0) + aborted_requests: list[str] = [] + + async def abort(request_id: str) -> None: + aborted_requests.append(request_id) + + connection.engine.abort = abort + renderer.message_counts.clear() + renderer.skip_mm_cache.clear() + + submitted = await _create_response(connection) + + assert aborted_requests == ["req_active"] + assert cancellation_cleanup == ["active"] + assert renderer.message_counts == [3, 2, 2] + assert renderer.skip_mm_cache == [True, True, False] + assert len(submitted) == 1 + assert _prompt_texts(renderer.conversations[-1]) == ["item_3", "after"] + assert [item.id for item in connection.session.items] == [ + "item_0", + "before", + "assistant", + "item_3", + "after", + ] + assert connection.session.model_context_first_item_id == "assistant" + assert connection.session.active_response is None + + +@pytest.mark.asyncio @pytest.mark.asyncio async def test_audio_prompt_uses_standard_chat_audio_content() -> None: audio = b"\x00\x00\x00\x40" diff --git a/vllm_omni/entrypoints/openai/realtime/connection.py b/vllm_omni/entrypoints/openai/realtime/connection.py index 4831b6bef..9426d0e0f 100644 --- a/vllm_omni/entrypoints/openai/realtime/connection.py +++ b/vllm_omni/entrypoints/openai/realtime/connection.py @@ -9,7 +9,7 @@ import io import json import warnings import wave -from dataclasses import dataclass, replace +from dataclasses import dataclass from typing import TYPE_CHECKING, Any import numpy as np @@ -57,13 +57,17 @@ AUTO_TRUNCATION_TRIGGER_RATIO = 0.8 AUTO_TRUNCATION_TARGET_RATIO = 0.5 +def _is_prompt_limit_error(exc: VLLMValidationError) -> bool: + return exc.parameter in {"input_text", "input_tokens"} + + class _UnsupportedAudioFormatError(ValueError): """The Realtime endpoint only accepts mono PCM16 audio at 24 kHz.""" @dataclass(slots=True) class _ResolvedResponse: - input: list[Any] | None + input: list[types.ConversationItem] | None instructions: str | None modalities: list[str] max_output_tokens: int | str @@ -434,24 +438,31 @@ class OpenAIFullDuplexConnection: metadata=metadata, ) - async def _prepare_engine_input_with_auto_truncation( - self, - response: _ResolvedResponse, - *, - preflight_only: bool = False, - ) -> EngineInput | None: + async def _truncate_prompt_items(self, response: _ResolvedResponse) -> list[types.ConversationItem] | None: s = self.session - max_model_len = getattr(self.engine.model_config, "max_model_len", None) - persistent = response.input is None - items: list[Any] = s.items if response.input is None else response.input - staged_deletions: list[Any] = [] - if max_model_len is None: - return await self._build_full_prompt( - tools=response.tools, - instructions=response.instructions, - items=items, - skip_mm_cache=preflight_only, - ) + max_model_len = self.engine.model_config.max_model_len + if response.input is None: + items = s.model_context_items() + else: + items = response.input + + async def probe_prompt( + current_items: list[types.ConversationItem], + ) -> tuple[EngineInput | None, int]: + # Budget checks are speculative and may render repeatedly while + # truncating, so keep them off the shared sender cache. + try: + prompt = await self._build_full_prompt( + tools=response.tools, + instructions=response.instructions, + items=current_items, + skip_mm_cache=True, + ) + except VLLMValidationError as exc: + if not _is_prompt_limit_error(exc): + raise + return None, max_model_len + 1 + return prompt, len(prompt["prompt_token_ids"]) truncation = s.config.truncation or "auto" ratio = 1.0 @@ -477,24 +488,9 @@ class OpenAIFullDuplexConnection: trigger = limit target = limit - # Budget checks are speculative and may render repeatedly while - # truncating, so keep them off the shared sender cache. Once accepted, - # render again through the normal cache path for engine submission. - engine_input = await self._build_full_prompt( - tools=response.tools, - instructions=response.instructions, - items=items, - skip_mm_cache=True, - ) - total = len(engine_input["prompt_token_ids"]) + engine_input, total = await probe_prompt(items) if total <= trigger: - if preflight_only: - return engine_input - return await self._build_full_prompt( - tools=response.tools, - instructions=response.instructions, - items=items, - ) + return items if mode == "disabled": logger.warning( "[realtime] token budget exceeded (%d/%d) and truncation is disabled -- rejecting response.create", @@ -503,68 +499,30 @@ class OpenAIFullDuplexConnection: ) return None - if persistent: - # Preflight truncation is speculative. Work on a snapshot so a - # rejected response.create does not delete items from session history. - items = list(items) + all_items = items + if not all_items: + return None if total > limit else all_items - idx = 0 - while total > target and idx < len(items): - item = items[idx] - if getattr(item, "role", None) == "system": - idx += 1 - continue - if persistent and item.id is not None and s.item_in_progress.get(item.id, False): - idx += 1 - continue + low = 1 + high = len(all_items) + feasible_index: int | None = None + while low < high: + middle = (low + high) // 2 + items = all_items[middle:] + engine_input, total = await probe_prompt(items) + if total <= target: + high = middle + feasible_index = middle + else: + low = middle + 1 - remove_indexes = [idx] - if item.type in ("function_call", "function_call_output"): - call_id = getattr(item, "call_id", None) - pair_idx = next( - ( - other_idx - for other_idx, other in enumerate(items) - if other_idx != idx and getattr(other, "call_id", None) == call_id - ), - None, - ) - if pair_idx is not None: - remove_indexes.append(pair_idx) - - for remove_idx in sorted(remove_indexes, reverse=True): - removed = items[remove_idx] - if persistent: - staged_deletions.append(removed) - del items[remove_idx] - engine_input = await self._build_full_prompt( - tools=response.tools, - instructions=response.instructions, - items=items, - skip_mm_cache=True, - ) - total = len(engine_input["prompt_token_ids"]) + if feasible_index != low: + items = all_items[low:] + engine_input, total = await probe_prompt(items) if total > limit: return None - if preflight_only: - return engine_input - engine_input = await self._build_full_prompt( - tools=response.tools, - instructions=response.instructions, - items=items, - ) - for removed in staged_deletions: - if removed.id is None or s.remove_item(removed.id) is None: - continue - await self._send_event( - types.ConversationItemDeletedEvent( - event_id=_gen_id("evt"), - type="conversation.item.deleted", - item_id=removed.id, - ) - ) - return engine_input + return items async def _handle_response_create(self, event: types.ResponseCreateEvent): s = self.session @@ -585,14 +543,11 @@ class OpenAIFullDuplexConnection: had_active_response = s.active_response is not None try: - engine_input = await self._prepare_engine_input_with_auto_truncation( - response, - preflight_only=had_active_response, - ) + preflight_items = await self._truncate_prompt_items(response) except VLLMValidationError as exc: await self._send_error(str(exc), "invalid_request_error", event_id=event.event_id) return - if engine_input is None: + if preflight_items is None: await self._send_error( "The response input exceeds the model's input token limit", "invalid_request_error", @@ -604,11 +559,11 @@ class OpenAIFullDuplexConnection: if s.active_response is not None: await self._cancel_active_response() try: - engine_input = await self._prepare_engine_input_with_auto_truncation(response) + preflight_items = await self._truncate_prompt_items(response) except VLLMValidationError as exc: await self._send_error(str(exc), "invalid_request_error", event_id=event.event_id) return - if engine_input is None: + if preflight_items is None: await self._send_error( "The response input exceeds the model's input token limit", "invalid_request_error", @@ -616,6 +571,18 @@ class OpenAIFullDuplexConnection: ) return + try: + engine_input = await self._build_full_prompt( + tools=response.tools, + instructions=response.instructions, + items=preflight_items, + ) + except VLLMValidationError as exc: + await self._send_error(str(exc), "invalid_request_error", event_id=event.event_id) + return + if response.input is None: + s.commit_model_context_items(preflight_items) + response_id = _gen_id("resp") await self._send_event( types.ResponseCreatedEvent( @@ -1609,8 +1576,8 @@ class OpenAIFullDuplexConnection: return text return "" - def _resolve_response_input(self, items: list[Any]) -> list[Any]: - resolved = [] + def _resolve_response_input(self, items: list[Any]) -> list[types.ConversationItem]: + resolved: list[types.ConversationItem] = [] for item in items: if getattr(item, "type", None) != "item_reference": self._validate_input_item(item) @@ -1641,7 +1608,7 @@ class OpenAIFullDuplexConnection: tools: list | None = None, *, instructions: str | None = None, - items: list | None = None, + items: list[types.ConversationItem] | None = None, skip_mm_cache: bool = False, ) -> EngineInput: """Render the effective conversation through normal chat preprocessing.""" @@ -1706,10 +1673,6 @@ class OpenAIFullDuplexConnection: messages.append({"role": "system", "content": text}) request = ChatCompletionRequest(model=self.model_name, messages=messages) - tok_params = replace( - request.build_tok_params(self.engine.model_config), - max_total_tokens=None, - ) _, (engine_input,) = await chat_handler._preprocess_chat( request, messages, @@ -1717,7 +1680,6 @@ class OpenAIFullDuplexConnection: default_template_content_format=chat_handler.chat_template_content_format, default_template_kwargs=chat_handler._effective_chat_template_kwargs(request), tool_dicts=converted_tools, - tok_params=tok_params, skip_mm_cache=skip_mm_cache, ) if engine_input.get("prompt_token_ids") is None: diff --git a/vllm_omni/entrypoints/openai/realtime/session.py b/vllm_omni/entrypoints/openai/realtime/session.py index a297b2c50..c30ed4ada 100644 --- a/vllm_omni/entrypoints/openai/realtime/session.py +++ b/vllm_omni/entrypoints/openai/realtime/session.py @@ -87,7 +87,15 @@ class AudioFullDuplexSessionState: config: types.RealtimeSessionCreateRequest = field(default_factory=_default_config) conversation_id: str = field(default_factory=lambda: _gen_id("conv")) - items: list[Any] = field(default_factory=list) + items: list[types.ConversationItem] = field(default_factory=list) + # Valid model-context cursor states are: + # - (None, False): include from the first retained item. + # - (first item ID, False): include from that item onward. + # - (None, True): the cursor is past the end and the model context is + # empty; a later append moves the cursor to the appended item. + # The combination (item ID, True) is invalid. + model_context_first_item_id: str | None = None + model_context_cursor_at_end: bool = False item_duration_ms: dict[str, float] = field(default_factory=dict) item_token_ids: dict[str, list[int]] = field(default_factory=dict) @@ -103,17 +111,17 @@ class AudioFullDuplexSessionState: return i return None - def find_item(self, item_id: str) -> Any | None: + def find_item(self, item_id: str) -> types.ConversationItem | None: idx = self.find_item_index(item_id) return self.items[idx] if idx is not None else None @staticmethod - def _jsonable_item(item: Any) -> Any: + def _jsonable_item(item: types.ConversationItem) -> Any: if hasattr(item, "model_dump"): return item.model_dump(mode="json", exclude_none=True) return item - def _history_size(self, items: list[Any] | None = None) -> int: + def _history_size(self, items: list[types.ConversationItem] | None = None) -> int: history = self.items if items is None else items serialized = json.dumps( [self._jsonable_item(item) for item in history], @@ -122,10 +130,25 @@ class AudioFullDuplexSessionState: ) return len(serialized.encode("utf-8")) - def _ensure_history_capacity(self, items: list[Any]) -> None: + def _ensure_history_capacity(self, items: list[types.ConversationItem]) -> None: if self._history_size(items) > MAX_HISTORY_BYTES: raise HistoryLimitError(f"Conversation history exceeds the {MAX_HISTORY_BYTES} byte limit") + def _model_context_start_index(self) -> int: + if self.model_context_cursor_at_end: + return len(self.items) + if self.model_context_first_item_id is None: + return 0 + index = self.find_item_index(self.model_context_first_item_id) + return len(self.items) if index is None else index + + def model_context_items(self) -> list[types.ConversationItem]: + return self.items[self._model_context_start_index() :] + + def commit_model_context_items(self, items: list[types.ConversationItem]) -> None: + self.model_context_first_item_id = items[0].id if items else None + self.model_context_cursor_at_end = not items + def insert_item(self, item: Any, previous_item_id: str | None = None) -> int: if item.id is not None: existing_idx = self.find_item_index(item.id) @@ -153,6 +176,13 @@ class AudioFullDuplexSessionState: candidate.insert(pos, item) self._ensure_history_capacity(candidate) self.items.insert(pos, item) + # A past-the-end cursor only advances for an item appended after it. + # Inserting before the end leaves the cursor past every included item. + if self.model_context_cursor_at_end and pos == len(self.items) - 1: + self.model_context_first_item_id = item.id + self.model_context_cursor_at_end = False + elif not self.model_context_cursor_at_end and self.model_context_first_item_id is None: + self.model_context_first_item_id = item.id return pos def replace_item(self, item: Any) -> int: @@ -177,11 +207,18 @@ class AudioFullDuplexSessionState: self.item_in_progress.pop(item_id, None) self.pending_truncations_ms.pop(item_id, None) - def remove_item(self, item_id: str) -> Any | None: + def remove_item(self, item_id: str) -> types.ConversationItem | None: idx = self.find_item_index(item_id) if idx is None: return None + cursor_index = self._model_context_start_index() item = self.items.pop(idx) + if idx == cursor_index and not self.model_context_cursor_at_end: + if idx < len(self.items): + self.model_context_first_item_id = self.items[idx].id + else: + self.model_context_first_item_id = None + self.model_context_cursor_at_end = True if item.id: self._clear_item_metadata(item.id) return item diff --git a/vllm_omni/entrypoints/openai/serving_chat.py b/vllm_omni/entrypoints/openai/serving_chat.py index 421fd3337..b2f5876e2 100644 --- a/vllm_omni/entrypoints/openai/serving_chat.py +++ b/vllm_omni/entrypoints/openai/serving_chat.py @@ -105,7 +105,6 @@ from vllm.parser.mistral import MistralToolCall from vllm.reasoning import ReasoningParser from vllm.renderers import BaseRenderer, merge_kwargs from vllm.renderers.inputs import TokPrompt -from vllm.renderers.params import TokenizeParams from vllm.sampling_params import RequestOutputKind, SamplingParams from vllm.tokenizers import TokenizerLike from vllm.tokenizers import TokenizerLike as AnyTokenizer @@ -977,7 +976,6 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): continue_final_message: bool = False, documents: list[dict[str, str]] | None = None, add_special_tokens: bool = False, - tok_params: TokenizeParams | None = None, skip_mm_cache: bool = False, ) -> tuple[list[ConversationMessage], list[TokPrompt]]: if renderer is None: @@ -997,8 +995,7 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): ), ) - if tok_params is None: - tok_params = request.build_tok_params(self.model_config) + tok_params = request.build_tok_params(self.model_config) mm_config = self.model_config.multimodal_config chat_params = request.build_chat_params( default_template,