From a65e89cbd47fe7c9f39dba558200c0bc16a758f3 Mon Sep 17 00:00:00 2001 From: Alex Brooks Date: Mon, 31 Aug 2026 21:17:57 -0600 Subject: [PATCH] [Core] Validate PipelineConfig in `__post_init__` (#6829) Signed-off-by: Alex Brooks --- tests/config/test_config_factory.py | 122 ++++++++++++------ tests/config/test_omni_config.py | 7 +- tests/config/test_pipeline_registry.py | 33 ++--- .../models/minicpmo_4_5/test_pipeline.py | 9 +- .../test_nemotron_voicechat_registration.py | 3 +- tests/utils/test_tracking_parser.py | 5 +- vllm_omni/config/pipeline_registry.py | 13 +- vllm_omni/config/stage_config.py | 13 +- 8 files changed, 121 insertions(+), 84 deletions(-) diff --git a/tests/config/test_config_factory.py b/tests/config/test_config_factory.py index d7e5d7a60..e3c6c585f 100644 --- a/tests/config/test_config_factory.py +++ b/tests/config/test_config_factory.py @@ -41,6 +41,11 @@ from vllm_omni.engine.arg_utils import SHARED_FIELDS, internal_blacklist_keys pytestmark = [pytest.mark.core_model, pytest.mark.cpu] +SINGLE_STAGE_PIPE_CFG = (StagePipelineConfig(stage_id=0, model_stage="a", final_output=True),) +# Validation strings to match against in post init +NO_STAGES_MATCH_STR = "no stages" +NO_TERMINAL_STAGE_MATCH_STR = "No terminal stage" + @pytest.fixture(autouse=True) def _stable_test_platform(monkeypatch): @@ -618,6 +623,7 @@ class TestPipelineDiscovery: p = PipelineConfig( model_type="custom_collide", hf_architectures=("SomeCollidingArch",), + stages=SINGLE_STAGE_PIPE_CFG, ) assert p.hf_architectures == ("SomeCollidingArch",) @@ -639,41 +645,42 @@ class TestStagePipelineConfig: class TestPipelineConfigNew: + def test_simple_init(self): + # Ensure pipeline config validates in post init. + PipelineConfig( + model_type="t", + model_arch="A", + stages=SINGLE_STAGE_PIPE_CFG, + ) + def test_frozen(self): - p = PipelineConfig(model_type="t", model_arch="A") + p = PipelineConfig( + model_type="t", + model_arch="A", + stages=SINGLE_STAGE_PIPE_CFG, + ) with pytest.raises(AttributeError): p.model_type = "changed" - def test_validate_valid(self): - p = PipelineConfig( - model_type="t", - model_arch="A", - stages=( - StagePipelineConfig(stage_id=0, model_stage="a"), - StagePipelineConfig(stage_id=1, model_stage="b", input_sources=(0,), final_output=True), - ), - ) - assert p.validate() == [] - def test_validate_no_stages(self): - p = PipelineConfig(model_type="t", model_arch="A") - assert any("no stages" in e.lower() for e in p.validate()) + with pytest.raises(ValueError, match=NO_STAGES_MATCH_STR): + PipelineConfig(model_type="t", model_arch="A") def test_validate_no_terminal_stage(self): """A pipeline with no ``final_output`` stage can never emit a result.""" - p = PipelineConfig( - model_type="t", - model_arch="A", - stages=( - StagePipelineConfig(stage_id=0, model_stage="a"), - StagePipelineConfig(stage_id=1, model_stage="b", input_sources=(0,)), - ), - ) - assert any("no terminal stage" in e.lower() for e in p.validate()) + with pytest.raises(ValueError, match=NO_TERMINAL_STAGE_MATCH_STR): + PipelineConfig( + model_type="t", + model_arch="A", + stages=( + StagePipelineConfig(stage_id=0, model_stage="a"), + StagePipelineConfig(stage_id=1, model_stage="b", input_sources=(0,)), + ), + ) - def test_validate_terminal_stage_need_not_be_last(self): - """The check is cycle-agnostic: any stage may carry ``final_output``.""" - p = PipelineConfig( + def test_pipeline_can_have_final_output_in_any_stage(self): + """Ensure any stage may carry ``final_output``.""" + PipelineConfig( model_type="t", model_arch="A", stages=( @@ -681,15 +688,20 @@ class TestPipelineConfigNew: StagePipelineConfig(stage_id=1, model_stage="b", input_sources=(0,)), ), ) - assert p.validate() == [] class TestPipelineRegistration: def test_resolve_pipeline_prefers_deploy_pipeline_key(self, clean_pipeline_registry, tmp_path): deploy_key = "deploy_selected_pipeline" model_type_key = "hf_model_type_pipeline" - deploy_pipe = PipelineConfig(model_type=deploy_key) - model_type_pipe = PipelineConfig(model_type=model_type_key) + deploy_pipe = PipelineConfig( + model_type=deploy_key, + stages=SINGLE_STAGE_PIPE_CFG, + ) + model_type_pipe = PipelineConfig( + model_type=model_type_key, + stages=SINGLE_STAGE_PIPE_CFG, + ) register_pipeline(deploy_pipe) register_pipeline(model_type_pipe) @@ -715,7 +727,11 @@ class TestPipelineRegistration: tmp_path, ): model_type_key = "registered_model_type_pipeline" - register_pipeline(PipelineConfig(model_type=model_type_key)) + pipeline_cfg = PipelineConfig( + model_type=model_type_key, + stages=SINGLE_STAGE_PIPE_CFG, + ) + register_pipeline(pipeline_cfg) deploy_path = tmp_path / "deploy.yaml" deploy_path.write_text("pipeline: missing_pipeline\n", encoding="utf-8") @@ -750,6 +766,7 @@ class TestPipelineRegistration: pipe_cfg = PipelineConfig( model_type=pipeline_key, hf_architectures=("ArchitectureFallbackForTest",), + stages=SINGLE_STAGE_PIPE_CFG, ) register_pipeline(pipe_cfg) @@ -780,15 +797,18 @@ class TestPipelineRegistration: model_type="rejected_by_predicate", hf_architectures=(shared_arch,), hf_config_predicate=rejecting_predicate, + stages=SINGLE_STAGE_PIPE_CFG, ) raises_cfg = PipelineConfig( model_type="raises_in_predicate", hf_architectures=(shared_arch,), hf_config_predicate=raising_predicate, + stages=SINGLE_STAGE_PIPE_CFG, ) accept_cfg = PipelineConfig( model_type="accepted_by_predicate", hf_architectures=(shared_arch,), + stages=SINGLE_STAGE_PIPE_CFG, ) register_pipeline(reject_cfg) register_pipeline(raises_cfg) @@ -822,6 +842,7 @@ class TestPipelineRegistration: model_type="predicate_without_arch_match", hf_architectures=("DifferentArchitectureForTest",), hf_config_predicate=predicate, + stages=SINGLE_STAGE_PIPE_CFG, ) register_pipeline(pipe_cfg) @@ -845,6 +866,7 @@ class TestPipelineRegistration: resolved_cfg = PipelineConfig( model_type="callable_resolved_pipeline", hf_architectures=("CallableArchitectureForTest",), + stages=SINGLE_STAGE_PIPE_CFG, ) seen_hf_configs = [] @@ -918,7 +940,10 @@ class TestPipelineRegistration: def test_pipeline_registration(self, clean_pipeline_registry): """Ensure that we can register and create a custom pipeline config.""" new_model_type = "new_model_type" - pipe_cfg = PipelineConfig(model_type=new_model_type) + pipe_cfg = PipelineConfig( + model_type=new_model_type, + stages=SINGLE_STAGE_PIPE_CFG, + ) # Register the new PipelineConfig assert new_model_type not in OMNI_PIPELINES @@ -961,7 +986,10 @@ class TestPipelineRegistration: def custom_resolver( hf_config: FakeConfig, ) -> PipelineConfig: - return PipelineConfig(model_type=resolved_type) + return PipelineConfig( + model_type=resolved_type, + stages=SINGLE_STAGE_PIPE_CFG, + ) # Register the new PipelineConfig assert new_model_type not in OMNI_PIPELINES @@ -1018,7 +1046,7 @@ class TestPipelineRegistration: pipe_cfg = PipelineConfig( model_type=pipeline_key, model_arch="DeployOnlyArch", - stages=(StagePipelineConfig(stage_id=0, model_stage="diffusion"),), + stages=SINGLE_STAGE_PIPE_CFG, ) register_pipeline(pipe_cfg) @@ -1051,7 +1079,11 @@ class TestPipelineRegistration: def test_structured_path_loads_explicit_deploy_config_once(self, clean_pipeline_registry, tmp_path): pipeline_key = "single_load_pipeline" - register_pipeline(PipelineConfig(model_type=pipeline_key)) + pipeline_cfg = PipelineConfig( + model_type=pipeline_key, + stages=SINGLE_STAGE_PIPE_CFG, + ) + register_pipeline(pipeline_cfg) deploy_path = tmp_path / "single_load.yaml" deploy_path.write_text(f"pipeline: {pipeline_key}\n", encoding="utf-8") @@ -1081,6 +1113,7 @@ class TestPipelineRegistration: pipeline = PipelineConfig( model_type="pipeline_with_default", default_deploy_config_name=default_name, + stages=SINGLE_STAGE_PIPE_CFG, ) with patch( @@ -1100,8 +1133,16 @@ class TestPipelineRegistration: OmniServingCapability.COMPLETIONS, "pipeline_a blocks completions", ) - pipe_a = PipelineConfig(model_type="detect_type", endpoint_restrictions=(restriction,)) - pipe_b = PipelineConfig(model_type="override_type", endpoint_restrictions=()) + pipe_a = PipelineConfig( + model_type="detect_type", + endpoint_restrictions=(restriction,), + stages=SINGLE_STAGE_PIPE_CFG, + ) + pipe_b = PipelineConfig( + model_type="override_type", + endpoint_restrictions=(), + stages=SINGLE_STAGE_PIPE_CFG, + ) register_pipeline(pipe_a) register_pipeline(pipe_b) @@ -1641,6 +1682,7 @@ stages: model_stage="ar", execution_type=StageExecutionType.LLM_AR, requires_multimodal_data=True, + final_output=True, ), ), ) @@ -1661,6 +1703,7 @@ stages: model_stage="ar", execution_type=StageExecutionType.LLM_AR, scheduler_cls=scheduler_cls, + final_output=True, ), ), ) @@ -1906,7 +1949,6 @@ class TestQwen3OmniPipeline: assert isinstance(p, PipelineConfig) assert p.model_arch == "Qwen3OmniMoeForConditionalGeneration" assert len(p.stages) == 3 - assert p.validate() == [] def test_thinker(self): p = resolve_pipeline_config( @@ -1960,7 +2002,6 @@ class TestQwen2_5OmniPipeline: assert isinstance(p, PipelineConfig) assert p.model_arch == "Qwen2_5OmniForConditionalGeneration" assert len(p.stages) == 3 - assert p.validate() == [] def test_thinker(self): p = resolve_pipeline_config("qwen2_5_omni") @@ -2009,7 +2050,6 @@ class TestQwen3TTSPipeline: assert p is not None assert p.model_arch == "Qwen3TTSTalkerForConditionalGeneration" assert len(p.stages) == 2 - assert p.validate() == [] def test_talker_stage(self): p = resolve_pipeline_config("qwen3_tts") @@ -2089,7 +2129,6 @@ class TestMingFlashOmniPipeline: assert isinstance(p, PipelineConfig) assert p.model_arch == "MingFlashOmniForConditionalGeneration" assert len(p.stages) == 2 - assert p.validate() == [] def test_thinker_stage(self): p = resolve_pipeline_config("ming_flash_omni") @@ -2146,7 +2185,6 @@ class TestMingFlashOmniPipeline: assert isinstance(p, PipelineConfig) assert p.model_arch == "MingFlashOmniTalkerForConditionalGeneration" assert len(p.stages) == 1 - assert p.validate() == [] def test_tts_stage(self): p = resolve_pipeline_config("ming_flash_omni_tts") @@ -2203,7 +2241,6 @@ class TestMingFlashOmniPipeline: assert isinstance(p, PipelineConfig) assert p.model_arch == "MingFlashOmniForConditionalGeneration" assert len(p.stages) == 1 - assert p.validate() == [] def test_thinker_only_stage(self): p = resolve_pipeline_config("ming_flash_omni_thinker_only") @@ -2242,7 +2279,6 @@ class TestMingFlashOmniPipeline: assert p is not None assert p.model_arch == "MingFlashOmniForConditionalGeneration" assert len(p.stages) == 2 - assert p.validate() == [] def test_image_thinker_stage(self): s = resolve_pipeline_config("ming_flash_omni_image").get_stage(0) diff --git a/tests/config/test_omni_config.py b/tests/config/test_omni_config.py index 7ed879d7b..126b51c18 100644 --- a/tests/config/test_omni_config.py +++ b/tests/config/test_omni_config.py @@ -45,6 +45,7 @@ from vllm_omni.config.stage_config import ( PipelineConfig, StageDeployConfig, StageExecutionType, + StagePipelineConfig, load_deploy_config, merge_pipeline_deploy, ) @@ -1039,7 +1040,10 @@ def test_from_pipeline_config_uses_hf_config_for_callable_resolver(): def test_from_pipeline_config_accepts_pre_resolved_pipeline(): - resolved_pipeline = PipelineConfig(model_type="callable_resolved_variant") + resolved_pipeline = PipelineConfig( + model_type="callable_resolved_variant", + stages=(StagePipelineConfig(stage_id=0, model_stage="a", final_output=True),), + ) omni_config = VllmOmniConfig.from_pipeline_config(resolved_pipeline) @@ -1072,6 +1076,7 @@ def test_from_pipeline_config_default_deploy_name_ignores_cwd(monkeypatch, tmp_p pipeline = PipelineConfig( model_type="pipeline_with_default", default_deploy_config_name=default_name, + stages=(StagePipelineConfig(stage_id=0, model_stage="a", final_output=True),), ) loaded_paths = [] diff --git a/tests/config/test_pipeline_registry.py b/tests/config/test_pipeline_registry.py index 7a6cb362d..88109e052 100644 --- a/tests/config/test_pipeline_registry.py +++ b/tests/config/test_pipeline_registry.py @@ -1,16 +1,22 @@ # SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project """Tests for out of tree registration to OMNI_PIPELINES.""" import pytest from transformers import PretrainedConfig from vllm_omni.config.pipeline_registry import OMNI_PIPELINES, register_pipeline -from vllm_omni.config.stage_config import PipelineConfig, pipeline_cfg_resolver +from vllm_omni.config.stage_config import PipelineConfig, StagePipelineConfig, pipeline_cfg_resolver pytestmark = [pytest.mark.core_model, pytest.mark.cpu] +def build_fake_pipeline_config(model_type: str) -> PipelineConfig: + return PipelineConfig( + model_type=model_type, stages=(StagePipelineConfig(stage_id=0, model_stage="a", final_output=True),) + ) + + @pytest.fixture def custom_resolver(): """Build a reusable custom resolver for PipelineConfigs.""" @@ -22,7 +28,7 @@ def custom_resolver(): def custom_resolver( hf_config: CustomConfigType, ) -> PipelineConfig: - return PipelineConfig(model_type="resolved_type") + return build_fake_pipeline_config("resolved_type") return custom_resolver @@ -30,7 +36,7 @@ def custom_resolver(): def test_register_pipeline_config(clean_pipeline_registry): """Ensure that we can register a custom pipeline config to OMNI_PIPELINES.""" new_model_type = "new_model_type" - pipe_cfg = PipelineConfig(model_type=new_model_type) + pipe_cfg = build_fake_pipeline_config(new_model_type) assert new_model_type not in OMNI_PIPELINES register_pipeline(pipe_cfg) assert new_model_type in OMNI_PIPELINES @@ -41,7 +47,7 @@ def test_register_pipeline_config_with_model_type(clean_pipeline_registry): """Ensure that we can register a custom pipeline config with an explicit model_type to OMNI_PIPELINES.""" new_model_type = "new_model_type" unused_model_type = "foo" - pipe_cfg = PipelineConfig(model_type=unused_model_type) + pipe_cfg = build_fake_pipeline_config(unused_model_type) assert new_model_type not in OMNI_PIPELINES assert unused_model_type not in OMNI_PIPELINES @@ -73,20 +79,3 @@ def test_minimax_h3_disaggregation_is_explicit_opt_in(): pipeline = OMNI_PIPELINES["minimax_h3_disaggregated"] assert isinstance(pipeline, PipelineConfig) assert pipeline.model_type == "minimax_h3_disaggregated" - - -def test_in_tree_pipelines_pass_topology_validation(): - """Every statically registered in-tree pipeline must have a valid topology. - - ``PipelineConfig.validate()`` only runs on ``register_pipeline``, which - in-tree pipelines never go through, so a shipped topology error would - otherwise stay invisible until serving hangs or misroutes. - Resolver entries are skipped: they need an ``hf_config`` to produce a - ``PipelineConfig``. - """ - failures = { - model_type: errors - for model_type, pipeline in OMNI_PIPELINES.items() - if isinstance(pipeline, PipelineConfig) and (errors := pipeline.validate()) - } - assert not failures, f"in-tree pipelines failed topology validation: {failures}" diff --git a/tests/model_executor/models/minicpmo_4_5/test_pipeline.py b/tests/model_executor/models/minicpmo_4_5/test_pipeline.py index 396d50fd1..a05e4076f 100644 --- a/tests/model_executor/models/minicpmo_4_5/test_pipeline.py +++ b/tests/model_executor/models/minicpmo_4_5/test_pipeline.py @@ -26,6 +26,7 @@ from vllm_omni.config.pipeline_registry import OMNI_PIPELINES from vllm_omni.config.stage_config import ( PipelineConfig, StageExecutionType, + StagePipelineConfig, _apply_platform_overrides, load_deploy_config, merge_pipeline_deploy, @@ -65,7 +66,9 @@ class TestRegistryDeclaration: assert not hasattr(pipeline, "max_native_duplex_sessions") def test_ordinary_pipeline_defaults_to_no_duplex_control(self) -> None: - pipeline = PipelineConfig(model_type="ordinary") + pipeline = PipelineConfig( + model_type="ordinary", stages=(StagePipelineConfig(stage_id=0, model_stage="a", final_output=True),) + ) assert pipeline.duplex_control_enabled is False assert pipeline.duplex_serving_adapter is None assert not hasattr(pipeline, "max_native_duplex_sessions") @@ -80,10 +83,6 @@ class TestPipelineTopology: assert len(pipeline.stages) == 3 assert [s.stage_id for s in pipeline.stages] == [0, 1, 2] - def test_topology_validates(self, pipeline: PipelineConfig) -> None: - # ``validate`` returns a list of structural errors; empty == valid. - assert pipeline.validate() == [] - def test_thinker_stage(self, pipeline: PipelineConfig) -> None: thinker = pipeline.get_stage(0) assert thinker is not None diff --git a/tests/model_executor/models/test_nemotron_voicechat_registration.py b/tests/model_executor/models/test_nemotron_voicechat_registration.py index 62e006aa3..935ffe96d 100644 --- a/tests/model_executor/models/test_nemotron_voicechat_registration.py +++ b/tests/model_executor/models/test_nemotron_voicechat_registration.py @@ -1,5 +1,5 @@ # SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project """L1 registration + config-shim tests for NemotronVoiceChat (CPU, no weights).""" import subprocess @@ -59,7 +59,6 @@ def test_pipeline_registered_with_three_stages() -> None: assert code2wav.execution_type == StageExecutionType.LLM_GENERATION assert code2wav.final_output and code2wav.final_output_type == "audio" assert code2wav.input_sources == (1,) - assert pipeline.validate() == [] def test_architectures_registered() -> None: diff --git a/tests/utils/test_tracking_parser.py b/tests/utils/test_tracking_parser.py index f00da2797..443ec0138 100644 --- a/tests/utils/test_tracking_parser.py +++ b/tests/utils/test_tracking_parser.py @@ -1,3 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + """Tests for TrackingArgumentParser and related utilities.""" import argparse @@ -35,7 +38,7 @@ _TEST_PIPELINE = PipelineConfig( stages=( StagePipelineConfig(stage_id=0, model_stage="stage0"), StagePipelineConfig(stage_id=1, model_stage="stage1"), - StagePipelineConfig(stage_id=2, model_stage="stage2"), + StagePipelineConfig(stage_id=2, model_stage="stage2", final_output=True), ), ) diff --git a/vllm_omni/config/pipeline_registry.py b/vllm_omni/config/pipeline_registry.py index 373a84739..83f8de383 100644 --- a/vllm_omni/config/pipeline_registry.py +++ b/vllm_omni/config/pipeline_registry.py @@ -1,5 +1,5 @@ # SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project """Pipeline registry and factory for vllm-omni. ``OMNI_PIPELINES`` maps each ``model_type`` to either a ``PipelineConfig`` @@ -196,18 +196,13 @@ def register_pipeline(pipeline: PipelineConfig | PipelineResolverFunc, model_typ since resolvers can return multiple different PipelineConfigs depending on the consumed config. """ - errors: list[str] = [] if isinstance(pipeline, PipelineConfig): - errors = pipeline.validate() model_type = model_type if model_type is not None else pipeline.model_type - else: - if model_type is None: - raise ValueError("Model type must be explicitly provided when registering a pipeline resolver") + elif model_type is None: + raise ValueError("Model type must be explicitly provided when registering a pipeline resolver") if model_type in OMNI_PIPELINES: - errors.append(f"Model type {model_type} is already registered; the old mapping will be clobbered") - if errors: - logger.warning("Registration for pipeline of type %s produced the following issues: %s", model_type, errors) + logger.warning(f"Model type {model_type} is already registered; the old mapping will be clobbered") OMNI_PIPELINES[model_type] = pipeline diff --git a/vllm_omni/config/stage_config.py b/vllm_omni/config/stage_config.py index aa27a6c45..2c1dc670b 100644 --- a/vllm_omni/config/stage_config.py +++ b/vllm_omni/config/stage_config.py @@ -302,6 +302,17 @@ class PipelineConfig: # Global CLI spelling -> (stage id, stage-local spelling). stage_cli_aliases: dict[str, tuple[int, str]] = field(default_factory=dict) + def __post_init__(self) -> None: + errors = self.get_validation_errors() + + if errors: + # If we have multiple errors, put them on separate indented lines for readability + multi_sep = "\n\t" + error_str = multi_sep.join(errors) + if len(errors) > 1: + error_str = f"{multi_sep}{error_str}" + raise ValueError(f"PipelineConfig initialization failed with the following error(s): {error_str}") + def get_stage(self, stage_id: int) -> StagePipelineConfig | None: """Look up a stage by its ID.""" for stage in self.stages: @@ -309,7 +320,7 @@ class PipelineConfig: return stage return None - def validate(self) -> list[str]: + def get_validation_errors(self) -> list[str]: """Return list of topology errors (empty if valid).""" errors: list[str] = [] if not self.stages: