diff --git a/vllm_omni/engine/async_omni_engine.py b/vllm_omni/engine/async_omni_engine.py index 7e7314585..52893e9a1 100644 --- a/vllm_omni/engine/async_omni_engine.py +++ b/vllm_omni/engine/async_omni_engine.py @@ -14,7 +14,6 @@ import asyncio import concurrent.futures import copy import json -import os import queue import shutil import threading @@ -72,7 +71,6 @@ from vllm_omni.engine.messages import ( StageSubmissionMessage, ) from vllm_omni.engine.orchestrator import ( - _EVENT_DRIVEN_ORCH_ENV, Orchestrator, _event_driven_orch_default_for_pipeline, ) @@ -237,12 +235,9 @@ class AsyncOmniEngine: pipeline_config.duplex_serving_adapter if pipeline_config is not None else None ) self._duplex_control_enabled = bool(pipeline_config and pipeline_config.duplex_control_enabled) - if ( - _EVENT_DRIVEN_ORCH_ENV not in os.environ - and pipeline_config is not None - and _event_driven_orch_default_for_pipeline(pipeline_config.model_type) - ): - os.environ[_EVENT_DRIVEN_ORCH_ENV] = "1" + self._event_driven_orch_default = _event_driven_orch_default_for_pipeline( + pipeline_config.model_type if pipeline_config is not None else None + ) self.duplex_session_config = DuplexSessionRuntimeConfig() if deploy_config_path is not None: self.duplex_session_config = load_deploy_config(deploy_config_path).duplex_session @@ -460,6 +455,7 @@ class AsyncOmniEngine: prom_metrics=self._prom_metrics, log_stats=self._log_stats, enable_orch_monitor=self._enable_orch_monitor, + event_driven_orch_default=self._event_driven_orch_default, duplex_runtime_extension=duplex_runtime_extension, enable_duplex_control=self._duplex_control_enabled, duplex_session_config=self.duplex_session_config, diff --git a/vllm_omni/engine/orchestrator.py b/vllm_omni/engine/orchestrator.py index 3c7089f80..b047bfd82 100644 --- a/vllm_omni/engine/orchestrator.py +++ b/vllm_omni/engine/orchestrator.py @@ -449,6 +449,7 @@ class Orchestrator: prom_metrics: Any = None, log_stats: bool = False, enable_orch_monitor: bool = False, + event_driven_orch_default: bool = False, duplex_runtime_extension: DuplexRuntimeExtension | None = None, enable_duplex_control: bool = False, duplex_session_config: DuplexSessionRuntimeConfig | None = None, @@ -520,7 +521,7 @@ class Orchestrator: self._shutdown_event = asyncio.Event() self._stages_shutdown = False - self._event_driven_orch = _event_driven_orch_enabled() + self._event_driven_orch = _event_driven_orch_enabled(default=event_driven_orch_default) # Distributed membership (optional, injected by DistStageRuntime) self._membership = membership_controller diff --git a/vllm_omni/entrypoints/async_omni.py b/vllm_omni/entrypoints/async_omni.py index 0012039e0..f9c880c9f 100644 --- a/vllm_omni/entrypoints/async_omni.py +++ b/vllm_omni/entrypoints/async_omni.py @@ -943,7 +943,9 @@ class AsyncOmni(EngineClient, OmniBase): # event-driven loop (vllm_omni/engine/orchestrator.py). from vllm_omni.engine.orchestrator import _event_driven_orch_enabled - event_driven_drain = _event_driven_orch_enabled() and hasattr(engine, "get_output_blocking_async") + event_driven_drain = _event_driven_orch_enabled( + default=bool(getattr(engine, "_event_driven_orch_default", False)) + ) and hasattr(engine, "get_output_blocking_async") async def _final_output_loop(): """Background coroutine that dispatches final outputs to request queues."""