diff --git a/docs/api/README.md b/docs/api/README.md index dce02e13d..d4c580629 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -92,7 +92,6 @@ Core scheduling and caching components. Configuration classes. - [vllm_omni.config.model.OmniModelConfig][] -- [vllm_omni.config.stage_config.ModelPipeline][] - [vllm_omni.config.stage_config.StageConfig][] - [vllm_omni.config.stage_config.StageConfigFactory][] - [vllm_omni.config.stage_config.StageType][] diff --git a/tests/entrypoints/test_omni_sleep_mode.py b/tests/entrypoints/test_omni_sleep_mode.py index 1365b320d..c2a5ff0a1 100644 --- a/tests/entrypoints/test_omni_sleep_mode.py +++ b/tests/entrypoints/test_omni_sleep_mode.py @@ -532,7 +532,10 @@ async def test_pure_diffusion_scenario(tp_size: int): try: await engine.sleep(level=1) await engine.wake_up() - async for _ in engine.generate("test", sampling_params=SamplingParams()): + async for _ in engine.generate( + "test", + sampling_params=OmniDiffusionSamplingParams(num_inference_steps=2, height=256, width=256), + ): pass logger.info("Pure diffusion OK (TP=%s)", tp_size) finally: diff --git a/tests/test_config_factory.py b/tests/test_config_factory.py index e541d3628..e3ebea88f 100644 --- a/tests/test_config_factory.py +++ b/tests/test_config_factory.py @@ -17,7 +17,6 @@ from tests.helpers.stage_config import get_deploy_config_path from vllm_omni.config.stage_config import ( _PIPELINE_REGISTRY, DeployConfig, - ModelPipeline, PipelineConfig, StageConfig, StageConfigFactory, @@ -351,93 +350,6 @@ class TestStageConfig: assert "parallel_config" not in omega_config.engine_args -class TestModelPipeline: - """Tests for ModelPipeline class.""" - - def test_valid_linear_dag(self): - """Test validation of a valid linear DAG.""" - stages = [ - StageConfig(stage_id=0, model_stage="thinker", input_sources=[]), - StageConfig(stage_id=1, model_stage="talker", input_sources=[0]), - StageConfig(stage_id=2, model_stage="code2wav", input_sources=[1]), - ] - pipeline = ModelPipeline(model_type="test", stages=stages) - errors = pipeline.validate_pipeline() - assert errors == [], f"Unexpected errors: {errors}" - - def test_valid_branching_dag(self): - """Test validation of a valid branching DAG.""" - stages = [ - StageConfig(stage_id=0, model_stage="input", input_sources=[]), - StageConfig(stage_id=1, model_stage="branch_a", input_sources=[0]), - StageConfig(stage_id=2, model_stage="branch_b", input_sources=[0]), - ] - pipeline = ModelPipeline(model_type="test", stages=stages) - errors = pipeline.validate_pipeline() - assert errors == [], f"Unexpected errors: {errors}" - - def test_missing_entry_point(self): - """Test that missing entry point is detected.""" - stages = [ - StageConfig(stage_id=0, model_stage="stage_a", input_sources=[1]), - StageConfig(stage_id=1, model_stage="stage_b", input_sources=[0]), - ] - pipeline = ModelPipeline(model_type="test", stages=stages) - errors = pipeline.validate_pipeline() - assert any("entry point" in e.lower() for e in errors) - - def test_missing_dependency(self): - """Test that missing stage reference is detected.""" - stages = [ - StageConfig(stage_id=0, model_stage="input", input_sources=[]), - StageConfig(stage_id=1, model_stage="output", input_sources=[99]), # Invalid - ] - pipeline = ModelPipeline(model_type="test", stages=stages) - errors = pipeline.validate_pipeline() - assert any("non-existent" in e.lower() for e in errors) - - def test_duplicate_stage_ids(self): - """Test that duplicate stage IDs are detected.""" - stages = [ - StageConfig(stage_id=0, model_stage="stage_a", input_sources=[]), - StageConfig(stage_id=0, model_stage="stage_b", input_sources=[]), # Duplicate - ] - pipeline = ModelPipeline(model_type="test", stages=stages) - errors = pipeline.validate_pipeline() - assert any("duplicate" in e.lower() for e in errors) - - def test_self_reference(self): - """Test that self-references are detected.""" - stages = [ - StageConfig(stage_id=0, model_stage="entry", input_sources=[]), - StageConfig(stage_id=1, model_stage="self_ref", input_sources=[1]), # Self - ] - pipeline = ModelPipeline(model_type="test", stages=stages) - errors = pipeline.validate_pipeline() - assert any("itself" in e.lower() for e in errors) - - def test_get_stage_by_id(self): - """Test getting stage by ID.""" - stages = [ - StageConfig(stage_id=0, model_stage="thinker", input_sources=[]), - StageConfig(stage_id=1, model_stage="talker", input_sources=[0]), - ] - pipeline = ModelPipeline(model_type="test", stages=stages) - - stage = pipeline.get_stage(1) - assert stage is not None - assert stage.model_stage == "talker" - - missing = pipeline.get_stage(99) - assert missing is None - - def test_empty_pipeline(self): - """Test validation of empty pipeline.""" - pipeline = ModelPipeline(model_type="test", stages=[]) - errors = pipeline.validate_pipeline() - assert any("no stages" in e.lower() for e in errors) - - class TestStageConfigFactory: """Tests for StageConfigFactory class.""" @@ -598,304 +510,6 @@ class TestStageResolutionHelpers: assert overridden == [] -class TestPipelineYamlParsing: - """Tests for pipeline YAML file parsing (@ZJY0516).""" - - def test_parse_qwen3_omni_moe_yaml(self, tmp_path): - """Test parsing the qwen3_omni_moe pipeline YAML.""" - yaml_content = """\ -model_type: qwen3_omni_moe -async_chunk: false - -stages: - - stage_id: 0 - model_stage: thinker - stage_type: llm - input_sources: [] - worker_type: ar - scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler - hf_config_name: thinker_config - final_output: true - final_output_type: text - is_comprehension: true - - - stage_id: 1 - model_stage: talker - stage_type: llm - input_sources: [0] - worker_type: ar - scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler - hf_config_name: talker_config - custom_process_input_func: vllm_omni.model_executor.stage_input_processors.qwen3_omni.thinker2talker - - - stage_id: 2 - model_stage: code2wav - stage_type: llm - input_sources: [1] - worker_type: generation - scheduler_cls: vllm_omni.core.sched.omni_generation_scheduler.OmniGenerationScheduler - hf_config_name: thinker_config - custom_process_input_func: vllm_omni.model_executor.stage_input_processors.qwen3_omni.talker2code2wav - final_output: true - final_output_type: audio -""" - yaml_file = tmp_path / "qwen3_omni_moe.yaml" - yaml_file.write_text(yaml_content) - - pipeline = StageConfigFactory._parse_pipeline_yaml(yaml_file, "qwen3_omni_moe") - - assert pipeline.model_type == "qwen3_omni_moe" - assert len(pipeline.stages) == 3 - - # Stage 0: thinker - s0 = pipeline.stages[0] - assert s0.stage_id == 0 - assert s0.model_stage == "thinker" - assert s0.stage_type == StageType.LLM - assert s0.input_sources == [] - assert s0.worker_type == "ar" - assert s0.final_output is True - assert s0.final_output_type == "text" - assert s0.is_comprehension is True - - # Stage 1: talker - s1 = pipeline.stages[1] - assert s1.stage_id == 1 - assert s1.input_sources == [0] - assert s1.custom_process_input_func == ( - "vllm_omni.model_executor.stage_input_processors.qwen3_omni.thinker2talker" - ) - assert s1.final_output is False - - # Stage 2: code2wav - s2 = pipeline.stages[2] - assert s2.stage_id == 2 - assert s2.input_sources == [1] - assert s2.worker_type == "generation" - assert s2.final_output_type == "audio" - - def test_parse_yaml_with_legacy_engine_input_source(self, tmp_path): - """Test backward compatibility with engine_input_source field.""" - yaml_content = """\ -model_type: legacy_model - -stages: - - stage_id: 0 - model_stage: entry - stage_type: llm - - stage_id: 1 - model_stage: downstream - stage_type: llm - engine_input_source: [0] -""" - yaml_file = tmp_path / "legacy.yaml" - yaml_file.write_text(yaml_content) - - pipeline = StageConfigFactory._parse_pipeline_yaml(yaml_file, "legacy_model") - assert pipeline.stages[1].input_sources == [0] - - def test_parse_yaml_with_connectors_and_edges(self, tmp_path): - """Test parsing pipeline with optional connectors and edges.""" - yaml_content = """\ -model_type: test_model - -stages: - - stage_id: 0 - model_stage: entry - stage_type: llm - input_sources: [] - -connectors: - type: ray - -edges: - - from: 0 - to: 1 -""" - yaml_file = tmp_path / "with_connectors.yaml" - yaml_file.write_text(yaml_content) - - pipeline = StageConfigFactory._parse_pipeline_yaml(yaml_file, "test_model") - assert pipeline.connectors == {"type": "ray"} - assert pipeline.edges == [{"from": 0, "to": 1}] - - def test_parsed_pipeline_passes_validation(self, tmp_path): - """Test that a well-formed YAML produces a valid pipeline.""" - yaml_content = """\ -model_type: valid_model - -stages: - - stage_id: 0 - model_stage: entry - stage_type: llm - input_sources: [] - final_output: true - final_output_type: text - - stage_id: 1 - model_stage: next - stage_type: llm - input_sources: [0] -""" - yaml_file = tmp_path / "valid.yaml" - yaml_file.write_text(yaml_content) - - pipeline = StageConfigFactory._parse_pipeline_yaml(yaml_file, "valid_model") - errors = pipeline.validate_pipeline() - assert errors == [], f"Unexpected validation errors: {errors}" - - def test_parse_yaml_migrates_runtime_max_batch_size(self, tmp_path): - """Test that runtime.max_batch_size is migrated to engine_args.max_num_seqs.""" - yaml_content = """\ -model_type: legacy_batch -stages: - - stage_id: 0 - model_stage: entry - stage_type: llm - input_sources: [] - runtime: - devices: "0" - max_batch_size: 16 - engine_args: - model_arch: SomeModel -""" - yaml_file = tmp_path / "legacy_batch.yaml" - yaml_file.write_text(yaml_content) - - pipeline = StageConfigFactory._parse_pipeline_yaml(yaml_file, "legacy_batch") - s0 = pipeline.stages[0] - assert "max_batch_size" not in s0.yaml_runtime - assert s0.yaml_engine_args.get("max_num_seqs") == 16 - - def test_parse_diffusion_stage_type(self, tmp_path): - """Test parsing a diffusion stage type from YAML.""" - yaml_content = """\ -model_type: diff_model - -stages: - - stage_id: 0 - model_stage: dit - stage_type: diffusion - input_sources: [] - final_output: true - final_output_type: image -""" - yaml_file = tmp_path / "diffusion.yaml" - yaml_file.write_text(yaml_content) - - pipeline = StageConfigFactory._parse_pipeline_yaml(yaml_file, "diff_model") - assert pipeline.stages[0].stage_type == StageType.DIFFUSION - - def test_parse_mimo_audio_yaml(self, tmp_path): - """Test parsing the MiMo Audio pipeline YAML.""" - yaml_content = """\ -model_type: mimo_audio - -stages: - - stage_id: 0 - model_stage: fused_thinker_talker - stage_type: llm - input_sources: [] - worker_type: ar - scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler - is_comprehension: true - final_output: true - final_output_type: text - - - stage_id: 1 - model_stage: code2wav - stage_type: llm - input_sources: [0] - worker_type: generation - scheduler_cls: vllm_omni.core.sched.omni_generation_scheduler.OmniGenerationScheduler - custom_process_input_func: vllm_omni.model_executor.stage_input_processors.mimo_audio.llm2code2wav - final_output: true - final_output_type: audio -""" - yaml_file = tmp_path / "mimo_audio.yaml" - yaml_file.write_text(yaml_content) - - pipeline = StageConfigFactory._parse_pipeline_yaml(yaml_file, "mimo_audio") - - assert pipeline.model_type == "mimo_audio" - assert len(pipeline.stages) == 2 - - # Stage 0: fused_thinker_talker - s0 = pipeline.stages[0] - assert s0.stage_id == 0 - assert s0.model_stage == "fused_thinker_talker" - assert s0.input_sources == [] - assert s0.worker_type == "ar" - assert s0.is_comprehension is True - assert s0.final_output is True - assert s0.final_output_type == "text" - - # Stage 1: code2wav - s1 = pipeline.stages[1] - assert s1.stage_id == 1 - assert s1.input_sources == [0] - assert s1.worker_type == "generation" - assert s1.custom_process_input_func == ( - "vllm_omni.model_executor.stage_input_processors.mimo_audio.llm2code2wav" - ) - assert s1.final_output is True - assert s1.final_output_type == "audio" - - # Pipeline validation - errors = pipeline.validate_pipeline() - assert errors == [], f"Unexpected errors: {errors}" - - -class TestAsyncChunk: - """Tests for async_chunk pipeline flag.""" - - def test_model_pipeline_async_chunk_default(self): - """Test ModelPipeline defaults async_chunk to False.""" - stages = [StageConfig(stage_id=0, model_stage="entry", input_sources=[])] - pipeline = ModelPipeline(model_type="test", stages=stages) - assert pipeline.async_chunk is False - - def test_model_pipeline_async_chunk_set(self): - """Test ModelPipeline with async_chunk=True.""" - stages = [StageConfig(stage_id=0, model_stage="entry", input_sources=[])] - pipeline = ModelPipeline(model_type="test", stages=stages, async_chunk=True) - assert pipeline.async_chunk is True - - def test_parse_async_chunk_from_yaml(self, tmp_path): - """Test that async_chunk is parsed from pipeline YAML.""" - yaml_content = """\ -model_type: test_async -async_chunk: true - -stages: - - stage_id: 0 - model_stage: entry - stage_type: llm - input_sources: [] -""" - yaml_file = tmp_path / "async.yaml" - yaml_file.write_text(yaml_content) - - pipeline = StageConfigFactory._parse_pipeline_yaml(yaml_file, "test_async") - assert pipeline.async_chunk is True - - def test_parse_missing_async_chunk_defaults_false(self, tmp_path): - """Test that missing async_chunk defaults to False.""" - yaml_content = """\ -model_type: test_no_async - -stages: - - stage_id: 0 - model_stage: entry - stage_type: llm - input_sources: [] -""" - yaml_file = tmp_path / "no_async.yaml" - yaml_file.write_text(yaml_content) - - pipeline = StageConfigFactory._parse_pipeline_yaml(yaml_file, "test_no_async") - assert pipeline.async_chunk is False - - class TestPipelineDiscovery: """Tests for the central pipeline registry (``pipeline_registry._OMNI_PIPELINES``).""" diff --git a/vllm_omni/config/__init__.py b/vllm_omni/config/__init__.py index f02c07588..e5762456f 100644 --- a/vllm_omni/config/__init__.py +++ b/vllm_omni/config/__init__.py @@ -6,7 +6,6 @@ from vllm_omni.config.lora import LoRAConfig from vllm_omni.config.model import OmniModelConfig from vllm_omni.config.stage_config import ( DeployConfig, - ModelPipeline, PipelineConfig, StageConfig, StageConfigFactory, @@ -30,7 +29,6 @@ __all__ = [ "LoRAConfig", "StageConfig", "StageConfigFactory", - "ModelPipeline", "StageType", "StageExecutionType", "StagePipelineConfig", diff --git a/vllm_omni/config/stage_config.py b/vllm_omni/config/stage_config.py index ed1d1ca18..43e00a142 100644 --- a/vllm_omni/config/stage_config.py +++ b/vllm_omni/config/stage_config.py @@ -20,13 +20,6 @@ from vllm_omni.config.yaml_util import create_config, load_yaml_config, to_dict from vllm_omni.core.sched.omni_ar_scheduler import OmniARAsyncScheduler, OmniARScheduler from vllm_omni.core.sched.omni_generation_scheduler import OmniGenerationScheduler -_MODELS_DIR = Path(__file__).resolve().parent.parent / "model_executor" / "models" - - -def get_pipeline_path(model_dir: str, filename: str) -> Path: - return _MODELS_DIR / model_dir / filename - - logger = init_logger(__name__) @@ -1061,77 +1054,6 @@ class StageConfig: return create_config(config_dict) -@dataclass -class ModelPipeline: - """Complete pipeline definition for a multi-stage model (legacy). - - TODO(@lishunyang12): remove once all models migrate to PipelineConfig. - """ - - model_type: str - stages: list[StageConfig] - - # Pipeline-wide behavior flags - async_chunk: bool = False - - # Optional distributed configuration - connectors: dict[str, Any] | None = None - edges: list[dict[str, Any]] | None = None - - def get_stage(self, stage_id: int) -> StageConfig | None: - """Look up a stage by its ID. - - Args: - stage_id: The stage ID to search for. - - Returns: - The matching StageConfig, or None if not found. - """ - for stage in self.stages: - if stage.stage_id == stage_id: - return stage - return None - - def validate_pipeline(self) -> list[str]: - """Validate pipeline topology at model integration time (not runtime). - - Checks: - - All stage IDs are unique - - All input_sources reference valid stage IDs - - At least one entry point (stage with empty input_sources) - - Returns: - List of validation error messages. Empty list if valid. - """ - errors: list[str] = [] - - if not self.stages: - errors.append("Topology has no stages defined") - return errors - - # Check for unique stage IDs - stage_ids = [s.stage_id for s in self.stages] - if len(stage_ids) != len(set(stage_ids)): - errors.append("Duplicate stage IDs found") - - stage_id_set = set(stage_ids) - - # Check input_sources reference valid stages - for stage in self.stages: - for source_id in stage.input_sources: - if source_id not in stage_id_set: - errors.append(f"Stage {stage.stage_id} references non-existent input source {source_id}") - if source_id == stage.stage_id: - errors.append(f"Stage {stage.stage_id} references itself as input source") - - # Check for at least one entry point - entry_points = [s for s in self.stages if not s.input_sources] - if not entry_points: - errors.append("No entry point found (stage with empty input_sources)") - - return errors - - class StageConfigFactory: """Factory that loads pipeline YAML and merges CLI overrides. @@ -1206,33 +1128,9 @@ class StageConfigFactory: continue return cls._create_from_registry(registered.model_type, cli_overrides, deploy_config_path) - # --- Legacy path: load from pipeline YAML --- - pipeline = cls._load_pipeline(model, trust_remote_code=trust_remote_code) - - if pipeline is None: - return None - - errors = pipeline.validate_pipeline() - if errors: - logger.warning(f"Pipeline validation warnings for {model}: {errors}") - - # Materialize the resolved pipeline-wide async_chunk value into every - # stage so build_engine_args_dict() can inject the stage connector - # spec and explicit False overrides are preserved. - resolved_async_chunk = cli_overrides.get("async_chunk") - if resolved_async_chunk is None: - resolved_async_chunk = bool(pipeline.async_chunk) - for stage in pipeline.stages: - stage.yaml_engine_args["async_chunk"] = bool(resolved_async_chunk) - - # Apply CLI overrides - result: list[StageConfig] = [] - for stage in pipeline.stages: - # Merge global CLI overrides - stage.runtime_overrides = cls._merge_cli_overrides(stage, cli_overrides) - result.append(stage) - - return result + # Not in the pipeline registry — let the caller fall back to the + # legacy ``stage_configs/*.yaml`` path (resolve_model_config_path). + return None @classmethod def _create_from_registry( @@ -1355,193 +1253,6 @@ class StageConfigFactory: return [config_dict] - @classmethod - def _load_pipeline(cls, model: str, trust_remote_code: bool = True) -> ModelPipeline | None: - """Load a legacy ``pipeline.yaml`` for the model. - - Searches ``model_executor/models/