From 764a109f66cd0f600a1dd907af6fd6630eb6e488 Mon Sep 17 00:00:00 2001 From: Alex Brooks Date: Tue, 29 Sep 2026 09:52:09 -0600 Subject: [PATCH] [Config] Disable Async Chunk for Unsupported Pipelines (#5099) Signed-off-by: Alex Brooks --- tests/config/test_config_factory.py | 95 ++++++++++++-- tests/config/test_omni_config.py | 84 +++++++++++-- vllm_omni/config/config_factory.py | 2 +- vllm_omni/config/omni_config.py | 32 ++--- vllm_omni/config/stage_config.py | 189 +++++++++++++++------------- 5 files changed, 273 insertions(+), 129 deletions(-) diff --git a/tests/config/test_config_factory.py b/tests/config/test_config_factory.py index df1a3a036..db97f990d 100644 --- a/tests/config/test_config_factory.py +++ b/tests/config/test_config_factory.py @@ -1405,6 +1405,21 @@ class TestDeployConfigLoading: with pytest.raises(ValueError, match=r"stage_args.*PipelineConfig.*stages"): load_deploy_config(deploy_path) + @pytest.mark.parametrize( + ("yaml_val", "expected"), + [ + ('"2"', 2), + ('""', 0), + ("null", 0), + ], + ) + def test_coerces_active_stream_window_to_int(self, tmp_path, yaml_val, expected): + """Ensure active stream window coerces str to int.""" + deploy_path = tmp_path / "deploy.yaml" + deploy_path.write_text(f"active_stream_window: {yaml_val}\n", encoding="utf-8") + deploy = load_deploy_config(deploy_path) + assert deploy.active_stream_window == expected + @pytest.mark.parametrize( ("filename", "max_sessions"), [ @@ -3511,14 +3526,10 @@ class TestSentinelDefaultPrecedence: assert stage1.sync_process_input_func is not None assert stage1.sync_process_input_func.endswith("thinker2talker_token_only") - # async_chunk=True must now be rejected: removing the fake hook means - # there is no next-stage input processor for the validator to accept. - # (Positive consequence -- users can't accidentally enable async_chunk - # on an arch that doesn't actually support it.) - import pytest as _pytest - - with _pytest.raises(ValueError, match="async_chunk=True"): - merge_pipeline_deploy(pipeline, DeployConfig(async_chunk=True)) + # ensure merging the pipeline deploy with async chunk set to True disables it, + # since currently it is not supported for this pipeline. + stage_configs = merge_pipeline_deploy(pipeline, DeployConfig(async_chunk=True)) + assert all([not stg_cfg.yaml_engine_args["async_chunk"] for stg_cfg in stage_configs]) # async_chunk=False merges cleanly and stage-0 yaml_engine_args carries # no spurious full-payload hook. @@ -3687,3 +3698,71 @@ class TestObjectStorageConfigResolution: matching = "s3://any-bucket/my-cosyvoice3-model" assert StageConfigFactory.try_infer_model_type(model=matching, trust_remote_code=False) == "cosyvoice3" + + +class TestAsyncChunkDefaults: + def test_async_chunk_auto_disabled_without_processor(self): + """Ensure a multi-stage model that doesn't support async chunk turns it off by default.""" + pipeline = PipelineConfig( + model_type="test_no_async", + model_arch="TestNoAsync", + stages=( + StagePipelineConfig( + stage_id=0, + model_stage="ar", + execution_type=StageExecutionType.LLM_AR, + final_output=True, + ), + StagePipelineConfig( + stage_id=1, + model_stage="generation", + execution_type=StageExecutionType.LLM_GENERATION, + input_sources=(0,), + ), + ), + ) + + deploy = DeployConfig() + # async chunk should not try to default to True in this case, + # since doing so will just raise a ValueError in validation. + merge_pipeline_deploy(pipeline, deploy) + assert not deploy.async_chunk + + def test_async_chunk_auto_disabled_when_yaml_omits_key(self, tmp_path): + """Ensure a multi-stage model that doesn't support async chunk turns it off by default + when a deploy config is provided that doesn't explicitly set it.""" + + deploy_path = tmp_path / "no_async_chunk.yaml" + deploy_path.write_text( + """ +stages: + - stage_id: 0 + devices: "0" + - stage_id: 1 + devices: "0" +""", + encoding="utf-8", + ) + deploy = load_deploy_config(deploy_path) + + pipeline = PipelineConfig( + model_type="test_no_async", + model_arch="TestNoAsync", + stages=( + StagePipelineConfig( + stage_id=0, + model_stage="ar", + execution_type=StageExecutionType.LLM_AR, + final_output=True, + ), + StagePipelineConfig( + stage_id=1, + model_stage="generation", + execution_type=StageExecutionType.LLM_GENERATION, + input_sources=(0,), + ), + ), + ) + + merge_pipeline_deploy(pipeline, deploy) + assert not deploy.async_chunk diff --git a/tests/config/test_omni_config.py b/tests/config/test_omni_config.py index 822cd2e51..b19c97144 100644 --- a/tests/config/test_omni_config.py +++ b/tests/config/test_omni_config.py @@ -4,6 +4,7 @@ from __future__ import annotations +import logging import warnings from dataclasses import fields, replace from inspect import Parameter, signature @@ -54,6 +55,7 @@ from vllm_omni.config.stage_config import ( load_deploy_config, merge_pipeline_deploy, resolve_deploy_yaml, + update_deploy_config_async_chunk_enabled, ) from vllm_omni.diffusion.diffusion_kv.config import DiffusionKVCacheMode from vllm_omni.engine.stage_engine_startup import _serialize_stage_config @@ -64,22 +66,27 @@ pytestmark = [pytest.mark.core_model, pytest.mark.cpu] _DEPLOY_DIR = Path(__file__).parents[2] / "vllm_omni" / "deploy" -@pytest.mark.parametrize("async_chunk", [False, True]) -def test_native_kv_transfer_requires_completed_ar_stage(async_chunk): - from types import SimpleNamespace - - pipeline = SimpleNamespace(stages=(), model_type="test") +@pytest.mark.parametrize("async_chunk", [None, False, True]) +def test_native_kv_transfer_disables_async_chunk(async_chunk): + """Ensure that a multistage pipeline that supports async chunk disables it if we have a kv transfer config.""" + pipeline = PipelineConfig( + model_type="test", + stages=( + # async_chunk_process_next_stage_input_func doesn't matter, it just needs to exist for this test. + StagePipelineConfig(stage_id=0, model_stage="ar", async_chunk_process_next_stage_input_func="foo.bar"), + StagePipelineConfig(stage_id=1, model_stage="dit", input_sources=(0,), final_output=True), + ), + ) + kv_transfer_config = {"kv_connector": "MooncakeConnector"} deploy = DeployConfig( async_chunk=async_chunk, stages=[ - StageDeployConfig(stage_id=0, engine_extras={"kv_transfer_config": {"kv_connector": "MooncakeConnector"}}) + StageDeployConfig(stage_id=0, engine_extras={"kv_transfer_config": kv_transfer_config}), + StageDeployConfig(stage_id=1, engine_extras={"kv_transfer_config": kv_transfer_config}), ], ) - if async_chunk: - with pytest.raises(ValueError, match="requires async_chunk=False"): - omni_config_module._validate_async_chunk_support(pipeline, deploy) - else: - omni_config_module._validate_async_chunk_support(pipeline, deploy) + update_deploy_config_async_chunk_enabled(pipeline, deploy) + assert not deploy.async_chunk @pytest.fixture(autouse=True) @@ -780,6 +787,33 @@ def test_from_pipeline_config_dispatches_async_chunk_processors_without_mutating assert pipeline.get_stage(1).custom_process_input_func is None +def test_from_pipeline_config_async_chunk_none_preserves_yaml_default(): + """Ensure passing None for async chunk in when creating from pipeline keeps yaml defaults.""" + config = _from_pipeline_key("qwen3_tts", cli_overrides={"async_chunk": None}) + talker_cfg = config.stage_by_id(0) + custom_proc = talker_cfg.custom_process_next_stage_input_func + assert custom_proc is not None + assert custom_proc.endswith("talker2code2wav_async_chunk") + + +def test_from_pipeline_config_warns_when_single_stage_async_chunk_enabled(caplog: pytest.LogCaptureFixture): + """Ensure that we get a warning indicating async chunk is disabled if requested for single stage models.""" + pipeline = PipelineConfig( + model_type="single_stage", + stages=(StagePipelineConfig(stage_id=0, model_stage="stage", final_output=True),), + ) + + with caplog.at_level(logging.WARNING, logger="vllm_omni.config.stage_config"): + config = VllmOmniConfig.from_pipeline_config(pipeline, cli_overrides={"async_chunk": True}) + + assert config.stage_by_id(0).connector_config.async_chunk is False + assert any( + record.levelno == logging.WARNING + and "async chunk is inapplicable to single stage models" in record.getMessage() + for record in caplog.records + ) + + def test_joyai_code2wav_waits_for_full_payload(): config = _from_pipeline_key("joyai_vl_interaction") talker = config.stage_by_id(1) @@ -2285,3 +2319,31 @@ def test_platform_stage_overlay_rejects_invalid_model_runner(): ) with pytest.raises(ValueError, match="model_runner must be 'v1' or 'v2'"): _apply_platform_overrides(deploy, platform="cuda") + + +def test_async_chunk_auto_disabled_without_processor(): + """Ensure a multi-stage model that doesn't support async chunk turns it off by default.""" + pipeline = PipelineConfig( + model_type="test_no_async", + model_arch="TestNoAsync", + stages=( + StagePipelineConfig( + stage_id=0, + model_stage="ar", + execution_type=StageExecutionType.LLM_AR, + final_output=True, + ), + StagePipelineConfig( + stage_id=1, + model_stage="generation", + execution_type=StageExecutionType.LLM_GENERATION, + input_sources=(0,), + ), + ), + ) + + deploy = DeployConfig() + # async chunk should not try to default to True in this case, + # since doing so will just raise a ValueError in validation. + merge_pipeline_deploy(pipeline, deploy) + assert not deploy.async_chunk diff --git a/vllm_omni/config/config_factory.py b/vllm_omni/config/config_factory.py index af0ea3e81..efc4884f7 100644 --- a/vllm_omni/config/config_factory.py +++ b/vllm_omni/config/config_factory.py @@ -514,7 +514,7 @@ class StageConfigFactory: pipeline_cfg, deploy_cfg = inject_forced_aligner_stage(pipeline_cfg, deploy_cfg, cli_overrides) - stages = merge_pipeline_deploy(pipeline_cfg, deploy_cfg, cli_overrides) + stages = merge_pipeline_deploy(pipeline_cfg, deploy_cfg) # Overlay declarative parallel strategies (opt-in) before CLI overrides. applied = cls._apply_strategy_specs(stages, strategy_specs) diff --git a/vllm_omni/config/omni_config.py b/vllm_omni/config/omni_config.py index fb2695856..526f4c9e5 100644 --- a/vllm_omni/config/omni_config.py +++ b/vllm_omni/config/omni_config.py @@ -58,8 +58,8 @@ from vllm_omni.config.stage_config import ( reconcile_diffusion_attention_overrides, resolve_stage_async_chunk, resolve_stage_model_runner, + update_deploy_config_async_chunk_enabled, validate_native_mrv2_session, - validate_stage_async_chunk_edges, ) from vllm_omni.diffusion.diffusion_kv.config import DiffusionKVCacheMode @@ -336,23 +336,6 @@ def _first_defined(*values: Any) -> Any: return None -def _validate_async_chunk_support(pipeline: PipelineConfig, deploy: DeployConfig) -> None: - has_inter_stage_edges = any(stage.input_sources for stage in pipeline.stages) - if deploy.async_chunk and any(stage.engine_extras.get("kv_transfer_config") for stage in deploy.stages): - raise ValueError("Native AR-to-DiT KV transfer requires async_chunk=False.") - if ( - deploy.async_chunk - and has_inter_stage_edges - and not any(stage.async_chunk_process_next_stage_input_func for stage in pipeline.stages) - ): - raise ValueError( - f"Pipeline {pipeline.model_type!r} has async_chunk=True in deploy but no stage " - "declares a dedicated async-chunk next-stage processor " - "(``async_chunk_process_next_stage_input_func``). " - "Either set async_chunk=False or implement an async-chunk producer on the pipeline." - ) - - def _resolve_execution_mode(execution_type: StageExecutionType) -> tuple[StageType, str | None]: try: return _EXECUTION_TYPE_TO_STAGE_WORKER[execution_type] @@ -2245,23 +2228,21 @@ class VllmOmniConfig: deploy_config_path, ) - if cli_overrides.get("async_chunk") is not None: - deploy.async_chunk = bool(cli_overrides["async_chunk"]) for name in _PIPELINE_DEPLOY_CLI_FIELDS: if cli_overrides.get(name) is not None: setattr(deploy, name, _copy_value(cli_overrides[name])) deploy = _apply_platform_overrides(deploy) - if len(pipeline_cfg.stages) <= 1: - deploy.async_chunk = False - _validate_async_chunk_support(pipeline_cfg, deploy) - validate_stage_async_chunk_edges(pipeline_cfg, deploy) + + cli_async_chunk = cli_overrides.get("async_chunk") + if cli_async_chunk is not None: + deploy.async_chunk = bool(cli_async_chunk) strategy_result = None if strategy_specs: from vllm_omni.config.composable_parallel import apply_strategy_specs - strategy_stages = merge_pipeline_deploy(pipeline_cfg, copy.deepcopy(deploy), {}) + strategy_stages = merge_pipeline_deploy(pipeline_cfg, copy.deepcopy(deploy)) strategy_result = apply_strategy_specs(strategy_stages, strategy_specs) strategy_overrides: dict[str, Any] = {} axis_fields = { @@ -2313,6 +2294,7 @@ class VllmOmniConfig: cli_overrides["omni_lb_policy"] = strategy_result.omni_lb_policy deploy_by_id = {stage.stage_id: stage for stage in deploy.stages} + update_deploy_config_async_chunk_enabled(pipeline_cfg, deploy) model = cli_overrides.get("model") stage_configs = tuple( diff --git a/vllm_omni/config/stage_config.py b/vllm_omni/config/stage_config.py index 08f588a83..0c46e9c22 100644 --- a/vllm_omni/config/stage_config.py +++ b/vllm_omni/config/stage_config.py @@ -581,7 +581,8 @@ class DeployConfig: individual ``StageDeployConfig`` entries under ``stages:``. """ - async_chunk: bool = True + # Async chunk's default depends on whether or not the pipeline supports it. + async_chunk: bool | None = None session_mode: str = "turn" model_runner: Literal["v1", "v2"] = "v1" # Stage-1 active stream slots; 0 preserves legacy all-stream cycling. @@ -627,6 +628,22 @@ _STAGE_RESERVED_KEYS = frozenset( } ) +# Pipeline-wide DeployConfig fields that are propagated to every stage's +# engine args during merge. These live at top level of the deploy YAML. +PIPELINE_WIDE_ENGINE_FIELDS: tuple[str, ...] = ( + "trust_remote_code", + "distributed_executor_backend", + "dtype", + "quantization", + "enable_prefix_caching", + "enable_chunked_prefill", + "data_parallel_size", + "pipeline_parallel_size", + "active_stream_window", + "custom_voice_dir", +) + + # Fields on StageDeployConfig that are populated from engine_args dict _STAGE_DEPLOY_FIELDS = {f.name: f for f in fields(StageDeployConfig) if f.name not in _STAGE_RESERVED_KEYS} @@ -639,7 +656,7 @@ def deploy_runtime_override_keys() -> frozenset[str]: They must remain overridable even if they are also modeled on ``OrchestratorArgs`` for top-level CLI parsing. """ - return frozenset(_STAGE_DEPLOY_FIELDS) | frozenset(_PIPELINE_WIDE_ENGINE_FIELDS) + return frozenset(_STAGE_DEPLOY_FIELDS) | frozenset(PIPELINE_WIDE_ENGINE_FIELDS) def _parse_stage_deploy(stage_data: dict[str, Any]) -> StageDeployConfig: @@ -827,13 +844,13 @@ def load_deploy_config(path: str | Path) -> DeployConfig: if not isinstance(raw_dict.get("cuda_mps", False), bool): raise ValueError("cuda_mps must be a boolean") + # TODO (Alex): Clean this up, we should not have fallback values here kwargs: dict[str, Any] = { "speech_cache": SpeechCacheConfig(**speech_cache), "cuda_mps": raw_dict.get("cuda_mps", False), - "async_chunk": raw_dict.get("async_chunk", True), + "async_chunk": raw_dict.get("async_chunk"), "session_mode": raw_dict.get("session_mode", "turn"), "model_runner": model_runner, - "active_stream_window": int(raw_dict.get("active_stream_window", 0) or 0), "duplex_session": DuplexSessionRuntimeConfig(**(raw_dict.get("duplex_session") or {})), "connectors": raw_dict.get("connectors", None), "edges": raw_dict.get("edges", None), @@ -843,19 +860,14 @@ def load_deploy_config(path: str | Path) -> DeployConfig: } # Pipeline-wide engine settings: only set if explicitly present in YAML # so the DeployConfig dataclass defaults take effect otherwise. - for name in ( - "trust_remote_code", - "distributed_executor_backend", - "dtype", - "quantization", - "enable_prefix_caching", - "enable_chunked_prefill", - "data_parallel_size", - "pipeline_parallel_size", - "custom_voice_dir", - ): + for name in PIPELINE_WIDE_ENGINE_FIELDS: if name in raw_dict: - kwargs[name] = raw_dict[name] + value = raw_dict[name] + # NOTE: This is arguably bad behavior since it's a one-off for + # only this field. We should consider deprecating this coercion. + if name == "active_stream_window": + value = int(value or 0) + kwargs[name] = value return DeployConfig(**kwargs) @@ -997,24 +1009,6 @@ def resolve_stage_async_chunk(deploy: DeployConfig, stage: StageDeployConfig | N return deploy.async_chunk and (stage is None or stage.async_chunk is not False) -def validate_stage_async_chunk_edges(pipeline: PipelineConfig, deploy: DeployConfig) -> None: - stages = {stage.stage_id: stage for stage in pipeline.stages} - deploy_by_id = {stage.stage_id: stage for stage in deploy.stages} - for consumer in pipeline.stages: - for source_id in consumer.input_sources: - producer = stages[source_id] - if not producer.async_chunk_process_next_stage_input_func: - continue - producer_async = resolve_stage_async_chunk(deploy, deploy_by_id.get(source_id)) - consumer_async = resolve_stage_async_chunk(deploy, deploy_by_id.get(consumer.stage_id)) - if producer_async != consumer_async: - raise ValueError( - f"Pipeline {pipeline.model_type!r} has incompatible async_chunk settings on " - f"connector edge {source_id} -> {consumer.stage_id}. " - "Set the same async_chunk mode on both stages or disable pipeline-wide async_chunk." - ) - - def _select_processor_funcs( ps: StagePipelineConfig, async_chunk: bool, @@ -1029,23 +1023,6 @@ def _select_processor_funcs( return input_proc, next_stage_proc -# Pipeline-wide DeployConfig fields that are propagated to every stage's -# engine args during merge. These live at top level of the deploy YAML. -_PIPELINE_WIDE_ENGINE_FIELDS: tuple[str, ...] = ( - "trust_remote_code", - "distributed_executor_backend", - "dtype", - "quantization", - "enable_prefix_caching", - "enable_chunked_prefill", - "data_parallel_size", - "pipeline_parallel_size", - "active_stream_window", - "custom_voice_dir", -) -PIPELINE_WIDE_ENGINE_FIELDS = _PIPELINE_WIDE_ENGINE_FIELDS - - def _build_engine_args( ps: StagePipelineConfig, ds: StageDeployConfig | None, @@ -1083,7 +1060,7 @@ def _build_engine_args( engine_args["stage_output_payload_keys"] = tuple(ps.stage_output_payload_keys) # Pipeline-wide top-level DeployConfig settings, applied to every stage. - for name in _PIPELINE_WIDE_ENGINE_FIELDS: + for name in PIPELINE_WIDE_ENGINE_FIELDS: value = getattr(deploy, name) if value is not None: engine_args[name] = value @@ -1174,51 +1151,95 @@ def _build_extras( return extras +def update_deploy_config_async_chunk_enabled( + pipeline: PipelineConfig, + deploy: DeployConfig, +) -> None: + """Set the pipeline wide async chunk value & validate it.""" + deploy.async_chunk = _resolve_pipeline_async_chunk_enabled(pipeline, deploy) + validate_stage_async_chunk_edges(pipeline, deploy) + + +def _resolve_pipeline_async_chunk_enabled( + pipeline: PipelineConfig, + deploy: DeployConfig, +) -> bool: + """Given the pipeline config and deploy config, determine the value of async_chunk. + Note that this currently assumes the cli override has been externally handled. + + The default is True if the model actually supports async chunk and is multistage, + and False otherwise. If the user tried to enable async chunk through the deploy + config, but it's inapplicable or unsupported, it will be disabled with a warning. + """ + # Single stage should never use async chunk + if len(pipeline.stages) <= 1: + if deploy.async_chunk: + logger.warning( + "Deploy config set async_chunk=True, but async chunk is inapplicable " + "to single stage models. As such, it will be disabled." + ) + return False + + has_inter_stage_edges = any(stage.input_sources for stage in pipeline.stages) + has_next_stage_inps = any(stage.async_chunk_process_next_stage_input_func for stage in pipeline.stages) + has_kv_transfer_cfg = any(stage.engine_extras.get("kv_transfer_config") for stage in deploy.stages) + supports_async_chunk = has_inter_stage_edges and has_next_stage_inps + + # If async chunk was set, make sure it's supported if + # requested; otherwise warn and disable it. + if deploy.async_chunk is not None: + if deploy.async_chunk: + if not supports_async_chunk: + logger.warning( + "Deploy config set async_chunk=True, but the pipeline config does not support it; disabling it." + ) + return False + if has_kv_transfer_cfg: + logger.warning("Native AR-to-DiT KV transfer requires async_chunk=False; async chunk will be disabled.") + return False + return deploy.async_chunk + + return supports_async_chunk and not has_kv_transfer_cfg + + +def validate_stage_async_chunk_edges(pipeline: PipelineConfig, deploy: DeployConfig) -> None: + stages = {stage.stage_id: stage for stage in pipeline.stages} + deploy_by_id = {stage.stage_id: stage for stage in deploy.stages} + for consumer in pipeline.stages: + for source_id in consumer.input_sources: + producer = stages[source_id] + if not producer.async_chunk_process_next_stage_input_func: + continue + producer_async = resolve_stage_async_chunk(deploy, deploy_by_id.get(source_id)) + consumer_async = resolve_stage_async_chunk(deploy, deploy_by_id.get(consumer.stage_id)) + if producer_async != consumer_async: + raise ValueError( + f"Pipeline {pipeline.model_type!r} has incompatible async_chunk settings on " + f"connector edge {source_id} -> {consumer.stage_id}. " + "Set the same async_chunk mode on both stages or disable pipeline-wide async_chunk." + ) + + def merge_pipeline_deploy( pipeline: PipelineConfig, deploy: DeployConfig, - cli_overrides: dict[str, Any] | None = None, ) -> list[StageConfig]: """Merge pipeline + deploy + platform overrides → list[StageConfig].""" - if cli_overrides is None: - cli_overrides = {} - deploy = _apply_platform_overrides(deploy) deploy_by_id = {s.stage_id: s for s in deploy.stages} - # async_chunk is irrelevant for single-stage pipelines, so we always disable it - if len(pipeline.stages) <= 1: - deploy.async_chunk = False + # NOTE: The CLI override is applied to the Deployconfig for async chunk prior + # to this point. We are in the process of better organizing the creation of the + # DeployConfig/PipelineConfig, and this path will be removed with the incorporation + # of the OmniConfig. + update_deploy_config_async_chunk_enabled(pipeline, deploy) - # async_chunk only applies to multi-stage pipelines: a pipeline with no - # consumer stages (every stage has empty input_sources) has no inter-stage - # edges, so async_chunk is a no-op and we skip the check entirely. - # For pipelines that DO have inter-stage edges, require a dedicated per-step - # async producer (``async_chunk_process_next_stage_input_func``). - # ``custom_process_next_stage_input_func`` is the full-payload / connector-path - # producer and does NOT imply async_chunk support — pipelines like qwen2_5_omni - # and covo_audio have it but removed their consumer-side ``custom_process_input_func`` - # because they don't support async_chunk, so accepting them here would silently - # miswire the consumer stage instead of raising a clear error. - _has_inter_stage_edges = any(ps.input_sources for ps in pipeline.stages) - if ( - deploy.async_chunk - and _has_inter_stage_edges - and not any(ps.async_chunk_process_next_stage_input_func for ps in pipeline.stages) - ): - raise ValueError( - f"Pipeline {pipeline.model_type!r} has async_chunk=True in deploy but no stage " - "declares a dedicated async-chunk next-stage processor " - "(``async_chunk_process_next_stage_input_func``). " - "Either set async_chunk=False or implement an async-chunk producer on the pipeline." - ) - - validate_stage_async_chunk_edges(pipeline, deploy) result: list[StageConfig] = [] for ps in pipeline.stages: ds = deploy_by_id.get(ps.stage_id) stage_type, worker_type = _resolve_execution_mode(ps.execution_type) - input_proc, next_stage_proc = _select_processor_funcs(ps, resolve_stage_async_chunk(deploy, ds)) + stage_async_chunk = resolve_stage_async_chunk(deploy, ds) + input_proc, next_stage_proc = _select_processor_funcs(ps, stage_async_chunk) engine_args = _build_engine_args(ps, ds, pipeline, deploy, next_stage_proc) # Downstream stages may share a multimodal wrapper class without owning # an encoder. Do not make vLLM profile dummy multimodal inputs for them.