diff --git a/tests/entrypoints/openai_api/test_serving_chat_speaker.py b/tests/entrypoints/openai_api/test_serving_chat_speaker.py index 27839ee23..b0977c0c2 100644 --- a/tests/entrypoints/openai_api/test_serving_chat_speaker.py +++ b/tests/entrypoints/openai_api/test_serving_chat_speaker.py @@ -87,8 +87,8 @@ def test_create_chat_completion_converts_value_error_to_error_response(mocker: M serving_chat.chat_template = None serving_chat.chat_template_content_format = "string" serving_chat.default_chat_template_kwargs = {} - serving_chat.openai_serving_render = mocker.MagicMock() - serving_chat.openai_serving_render.validate_chat_template.return_value = None + serving_chat.online_renderer = mocker.MagicMock() + serving_chat.online_renderer.validate_chat_template.return_value = None serving_chat._effective_chat_template_kwargs = mocker.MagicMock(return_value={}) serving_chat._preprocess_chat = mocker.AsyncMock( side_effect=ValueError("Invalid speaker 'uncle_fu'. Supported: ethan, vivian") diff --git a/tests/entrypoints/test_stream_finish_reason.py b/tests/entrypoints/test_stream_finish_reason.py index 52cc32215..ebbca1917 100644 --- a/tests/entrypoints/test_stream_finish_reason.py +++ b/tests/entrypoints/test_stream_finish_reason.py @@ -144,7 +144,7 @@ def _build_serving_chat(): engine_client=mock_engine, models=models, response_role="assistant", - openai_serving_render=mock_render, + online_renderer=mock_render, request_logger=None, chat_template=None, chat_template_content_format="auto", diff --git a/vllm_omni/core/sched/omni_ar_scheduler.py b/vllm_omni/core/sched/omni_ar_scheduler.py index 217c81543..74d09ab60 100644 --- a/vllm_omni/core/sched/omni_ar_scheduler.py +++ b/vllm_omni/core/sched/omni_ar_scheduler.py @@ -314,13 +314,6 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler): outputs: dict[int, list[EngineCoreOutput]] = defaultdict(list) spec_decoding_stats: SpecDecodingStats | None = None - kv_connector_stats: KVConnectorStats | None = ( - kv_connector_output.kv_connector_stats if kv_connector_output else None - ) - if kv_connector_stats and self.connector: - kv_stats = self.connector.get_kv_connector_stats() - if kv_stats: - kv_connector_stats = kv_connector_stats.aggregate(kv_stats) failed_kv_load_req_ids = None if kv_connector_output and kv_connector_output.invalid_block_ids: @@ -563,6 +556,20 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler): if kv_connector_output: self._update_from_kv_xfer_finished(kv_connector_output) + # Worker-side KV connector stats from the model runner output. + kv_connector_stats: KVConnectorStats | None = ( + kv_connector_output.kv_connector_stats if kv_connector_output else None + ) + if self.connector: + # Scheduler-side KV connector stats collected after connector update. + scheduler_kv_connector_stats = self.connector.get_kv_connector_stats() + if scheduler_kv_connector_stats is not None and not scheduler_kv_connector_stats.is_empty(): + kv_connector_stats = ( + kv_connector_stats.aggregate(scheduler_kv_connector_stats) + if kv_connector_stats is not None + else scheduler_kv_connector_stats + ) + # collect KV cache events from KV cache manager events = self.kv_cache_manager.take_events() diff --git a/vllm_omni/core/sched/omni_generation_scheduler.py b/vllm_omni/core/sched/omni_generation_scheduler.py index 42caea252..973b8c55a 100644 --- a/vllm_omni/core/sched/omni_generation_scheduler.py +++ b/vllm_omni/core/sched/omni_generation_scheduler.py @@ -431,14 +431,6 @@ class OmniGenerationScheduler(OmniSchedulerMixin, VLLMScheduler): outputs: dict[int, list[EngineCoreOutput]] = defaultdict(list) spec_decoding_stats: SpecDecodingStats | None = None - kv_connector_stats: KVConnectorStats | None = ( - kv_connector_output.kv_connector_stats if kv_connector_output else None - ) - # Merge connector-side stats (align with v0.14.0) - if kv_connector_stats and self.connector: - kv_stats = self.connector.get_kv_connector_stats() - if kv_stats: - kv_connector_stats = kv_connector_stats.aggregate(kv_stats) failed_kv_load_req_ids = None if kv_connector_output and getattr(kv_connector_output, "invalid_block_ids", None): @@ -646,6 +638,20 @@ class OmniGenerationScheduler(OmniSchedulerMixin, VLLMScheduler): if kv_connector_output: self._update_from_kv_xfer_finished(kv_connector_output) + # Worker-side KV connector stats from the model runner output. + kv_connector_stats: KVConnectorStats | None = ( + kv_connector_output.kv_connector_stats if kv_connector_output else None + ) + if self.connector: + # Scheduler-side KV connector stats collected after connector update. + scheduler_kv_connector_stats = self.connector.get_kv_connector_stats() + if scheduler_kv_connector_stats is not None and not scheduler_kv_connector_stats.is_empty(): + kv_connector_stats = ( + kv_connector_stats.aggregate(scheduler_kv_connector_stats) + if kv_connector_stats is not None + else scheduler_kv_connector_stats + ) + # Collect and publish KV cache events (align with v0.14.0) events = self.kv_cache_manager.take_events() if self.connector is not None: diff --git a/vllm_omni/entrypoints/openai/api_server.py b/vllm_omni/entrypoints/openai/api_server.py index 7e1153617..f3f0d3584 100755 --- a/vllm_omni/entrypoints/openai/api_server.py +++ b/vllm_omni/entrypoints/openai/api_server.py @@ -60,12 +60,11 @@ from vllm.entrypoints.pooling.classify.serving import ServingClassification from vllm.entrypoints.pooling.embed.serving import ServingEmbedding as OpenAIServingEmbedding from vllm.entrypoints.pooling.pooling.serving import ServingPooling from vllm.entrypoints.pooling.scoring.serving import ServingScores -from vllm.entrypoints.serve.disagg.serving import ServingTokens +from vllm.entrypoints.scale_out.token_in_token_out.serving import ServingTokens # vLLM moved `base` from openai.basic.api_router to serve.instrumentator.basic. # Keep a fallback for older/newer upstream layouts during rebase windows. from vllm.entrypoints.serve.instrumentator.basic import base -from vllm.entrypoints.serve.render.serving import OpenAIServingRender from vllm.entrypoints.serve.tokenize.serving import ServingTokenization from vllm.entrypoints.serve.utils.api_utils import ( load_aware_call, @@ -85,6 +84,7 @@ from vllm.entrypoints.speech_to_text.translation.serving import ( OpenAIServingTranslation, ) from vllm.logger import init_logger +from vllm.renderers.online_renderer import OnlineRenderer from vllm.tasks import POOLING_TASKS from vllm.tool_parsers import ToolParserManager from vllm.utils import random_uuid @@ -895,11 +895,10 @@ async def omni_init_app_state( await state.openai_serving_models.init_static_loras() # NOTE: kept aligned with upstream `init_app_state`: - # Use OpenAIServingRender (replaces the old OnlineRenderer + OnlineDerenderer + ServingRender split). - state.openai_serving_render = OpenAIServingRender( + # Use OnlineRenderer (replaced OpenAIServingRender which was removed upstream). + state.online_renderer = OnlineRenderer( model_config=engine_client.model_config, renderer=engine_client.renderer, - model_registry=state.openai_serving_models.registry, request_logger=request_logger, chat_template=resolved_chat_template, chat_template_content_format=args.chat_template_content_format, @@ -916,7 +915,7 @@ async def omni_init_app_state( OpenAIServingResponses( engine_client, state.openai_serving_models, - openai_serving_render=state.openai_serving_render, + state.online_renderer, request_logger=request_logger, chat_template=resolved_chat_template, chat_template_content_format=args.chat_template_content_format, @@ -937,7 +936,7 @@ async def omni_init_app_state( engine_client, state.openai_serving_models, args.response_role, - openai_serving_render=state.openai_serving_render, + online_renderer=state.online_renderer, request_logger=request_logger, chat_template=resolved_chat_template, chat_template_content_format=args.chat_template_content_format, @@ -964,7 +963,7 @@ async def omni_init_app_state( OpenAIServingCompletion( engine_client, state.openai_serving_models, - openai_serving_render=state.openai_serving_render, + online_renderer=state.online_renderer, request_logger=request_logger, return_tokens_as_token_ids=args.return_tokens_as_token_ids, enable_prompt_tokens_details=args.enable_prompt_tokens_details, @@ -1023,7 +1022,7 @@ async def omni_init_app_state( ) state.serving_tokenization = ServingTokenization( state.openai_serving_models, - state.openai_serving_render, + state.online_renderer, request_logger=request_logger, chat_template=resolved_chat_template, chat_template_content_format=args.chat_template_content_format, @@ -1055,7 +1054,7 @@ async def omni_init_app_state( engine_client, state.openai_serving_models, args.response_role, - openai_serving_render=state.openai_serving_render, + online_renderer=state.online_renderer, request_logger=request_logger, chat_template=resolved_chat_template, chat_template_content_format=args.chat_template_content_format, @@ -1074,7 +1073,7 @@ async def omni_init_app_state( ServingTokens( engine_client, state.openai_serving_models, - state.openai_serving_render, + state.online_renderer, request_logger=request_logger, return_tokens_as_token_ids=args.return_tokens_as_token_ids, enable_prompt_tokens_details=args.enable_prompt_tokens_details, diff --git a/vllm_omni/entrypoints/openai/serving_chat.py b/vllm_omni/entrypoints/openai/serving_chat.py index 2878487fe..43cc14853 100755 --- a/vllm_omni/entrypoints/openai/serving_chat.py +++ b/vllm_omni/entrypoints/openai/serving_chat.py @@ -150,6 +150,21 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): # Harmony flag (always False for vllm-omni models) use_harmony: bool = False + @property + def tool_call_id_type(self) -> str: + """Return the tool call ID type, delegating to model config. + + Upstream vLLM removed the stored ``tool_call_id_type`` attribute + from ``OpenAIServingChat`` after the ParserManager refactor; the + field is now resolved on demand via ``get_tool_call_id_type``. + """ + try: + from vllm.entrypoints.chat_utils import get_tool_call_id_type + + return get_tool_call_id_type(self.model_config) + except Exception: + return "random" + def _should_stream_with_auto_tool_parsing(self, request: ChatCompletionRequest) -> bool: """Check if streamed tokens should go through the tool-call parser. @@ -368,7 +383,7 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): tool_dicts = [tool.model_dump() for tool in request.tools] if not self.use_harmony: - error_check_ret = self.openai_serving_render.validate_chat_template( + error_check_ret = self.online_renderer.validate_chat_template( request_chat_template=request.chat_template, chat_template_kwargs=request.chat_template_kwargs, trust_request_chat_template=self.trust_request_chat_template, @@ -396,7 +411,7 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): ) else: should_include_tools = tool_dicts is not None - conversation, engine_prompts = self.openai_serving_render._make_request_with_harmony( + conversation, engine_prompts = self.online_renderer._make_request_with_harmony( request, should_include_tools ) diff --git a/vllm_omni/worker/gpu_ar_model_runner.py b/vllm_omni/worker/gpu_ar_model_runner.py index d84404508..6cd2b3566 100644 --- a/vllm_omni/worker/gpu_ar_model_runner.py +++ b/vllm_omni/worker/gpu_ar_model_runner.py @@ -159,6 +159,11 @@ class OmniAsyncGPUModelRunnerOutput(AsyncGPUModelRunnerOutput): async_output_copy_stream = kwargs.pop("async_output_copy_stream") vocab_size = kwargs.pop("vocab_size") routed_experts = kwargs.pop("routed_experts", None) + # Upstream AsyncGPUModelRunnerOutput added check_ep_fault / _has_fault + # for EP all2all fault tolerance (PR #43637). Omni doesn't use this + # feature but must consume the kwarg to prevent TypeError from stray + # kwargs and initialize the attribute so super().get_output() works. + kwargs.pop("check_ep_fault", False) if kwargs: raise TypeError(f"Unexpected OmniAsyncGPUModelRunnerOutput kwargs: {sorted(kwargs)}") @@ -170,6 +175,7 @@ class OmniAsyncGPUModelRunnerOutput(AsyncGPUModelRunnerOutput): self.vocab_size = vocab_size self._logprobs_tensors = logprobs_tensors self._routed_experts = routed_experts + self._has_fault: torch.Tensor | None = None default_stream = torch.cuda.current_stream() with torch.cuda.stream(async_output_copy_stream): @@ -220,6 +226,12 @@ class OmniAsyncGPUModelRunnerOutput(AsyncGPUModelRunnerOutput): if background_exception is not None: raise background_exception self._build_model_runner_output_once() + # Upstream AsyncGPUModelRunnerOutput.get_output() accesses + # self._has_fault for EP all2all fault tolerance (PR #43637). + # Ensure the attribute exists even when __init__ was bypassed + # (e.g. unit tests using object.__new__). + if not hasattr(self, "_has_fault"): + self._has_fault = None with record_function_or_nullcontext("omni_async_output:get_output/finalize_async_sampled_tokens"): return super().get_output()