diff --git a/invokeai/app/invocations/wan_model_loader.py b/invokeai/app/invocations/wan_model_loader.py index eb6efffb2f7..7524e4c0ca0 100644 --- a/invokeai/app/invocations/wan_model_loader.py +++ b/invokeai/app/invocations/wan_model_loader.py @@ -154,6 +154,11 @@ def invoke(self, context: InvocationContext) -> WanModelLoaderOutput: context.logger.warning("'Transformer (Low Noise)' is ignored for the single-expert TI2V-5B variant.") if self.transformer_low_noise_model is not None and main_variant != WanVariantType.TI2V_5B: + if self.transformer_low_noise_model.key == self.model.key: + raise ValueError( + "The same model is wired to both 'Transformer' and 'Transformer (Low Noise)'. " + "A Wan A14B expert pair needs two different GGUF models." + ) low_config = context.models.get_config(self.transformer_low_noise_model) self._validate_main_config(low_config, "Transformer (Low Noise)") if low_config.format != ModelFormat.GGUFQuantized: @@ -166,30 +171,63 @@ def invoke(self, context: InvocationContext) -> WanModelLoaderOutput: if getattr(low_config, "variant", None) != main_variant: raise ValueError("The high-noise and low-noise GGUF models must use the same Wan variant.") - if {primary_expert, low_expert} != {"high", "low"}: - raise ValueError("A Wan A14B GGUF expert pair must contain one high and one low expert.") + + # The expert tag is a filename heuristic, so 'none' (untagged) + # is common on community finetunes. The wiring itself is + # explicit user intent — main slot = high, low-noise slot = + # low — so an untagged file is taken at its wired position (or + # inferred as the complement of its tagged partner). Only a + # genuine conflict, both files claiming the *same* expert, is + # an error. + if primary_expert == low_expert != "none": + raise ValueError( + f"Both selected GGUF models are tagged as the {primary_expert}-noise expert. " + "A Wan A14B expert pair must contain one high and one low expert." + ) + if primary_expert == "none" and low_expert == "none": + context.logger.warning( + "Neither Wan A14B GGUF filename identifies its expert, so 'Transformer' is assumed to " + "be the high-noise expert and 'Transformer (Low Noise)' the low-noise expert. If the " + "output looks wrong, swap the two models." + ) # Make sure 'transformer' is the high-noise expert and # 'transformer_low_noise' is the low-noise expert. If the user # accidentally swapped them, swap back. - if primary_expert == "low" and low_expert == "high": + if primary_expert == "low" or low_expert == "high": transformer = low_id transformer_low_noise = primary_id + # The swap overrides the wiring on the strength of a + # filename tag, so say so: a mistagged file is otherwise an + # invisible expert inversion. + context.logger.warning( + f"The wired Wan A14B GGUF experts look reversed, so they were swapped: " + f"'{low_config.name}' (tagged '{low_expert}') runs as the high-noise expert and " + f"'{main_config.name}' (tagged '{primary_expert}') as the low-noise expert. " + "The tags come from the filenames — if the output looks wrong, a filename is lying." + ) else: transformer = primary_id transformer_low_noise = low_id else: transformer = primary_id - if main_variant in (WanVariantType.T2V_A14B, WanVariantType.I2V_A14B) and primary_expert != "high": - raise ValueError("An unpaired Wan A14B GGUF model must be the high-noise expert.") # A14B without a paired low-noise GGUF will produce degraded - # quality (only the high-noise expert runs). Warn but don't - # abort — TI2V-5B GGUFs are single-expert and totally fine. + # quality (only one expert runs). Warn but don't abort — a + # single wired transformer is explicit intent just like a pair + # is, and the tag is only a filename guess, so an untagged file + # must not be fatal here when the paired path accepts it. + # TI2V-5B GGUFs are single-expert and totally fine. if main_variant in (WanVariantType.T2V_A14B, WanVariantType.I2V_A14B): - context.logger.warning( - "A14B GGUF main was provided without a paired 'Transformer (Low Noise)'. " - "Only the high-noise expert will run; image quality will be reduced." + message = ( + "An A14B GGUF is wired to 'Transformer' without a paired 'Transformer (Low Noise)'. " + "Only this one expert will run; image quality will be reduced." ) + if primary_expert == "low": + message += ( + " Its filename tags it as the low-noise expert; when running a single expert, " + "the high-noise one is usually the better choice." + ) + context.logger.warning(message) # Borrow the boundary_ratio recorded on the optional Diffusers # component_source, when one is wired. diff --git a/tests/app/invocations/test_wan_model_loader.py b/tests/app/invocations/test_wan_model_loader.py index 3a16cc601cf..93442b265b6 100644 --- a/tests/app/invocations/test_wan_model_loader.py +++ b/tests/app/invocations/test_wan_model_loader.py @@ -35,7 +35,7 @@ def _config( ) -def _invoke( +def _prepare( main_config: SimpleNamespace, low_config: SimpleNamespace | None = None, component_config: SimpleNamespace | None = None, @@ -44,13 +44,14 @@ def _invoke( vae_latent_channels: int | None = None, vae_config: SimpleNamespace | None = None, t5_config: SimpleNamespace | None = None, -): + low_key: str = "low", +) -> tuple[WanModelLoaderInvocation, MagicMock]: main = _model("main") - low = _model("low") if low_config is not None else None + low = _model(low_key) if low_config is not None else None context = MagicMock() configs = {"main": main_config} if low_config is not None: - configs["low"] = low_config + configs[low_key] = low_config component = _model("component") if component_config is not None else None if component_config is not None: configs["component"] = component_config @@ -78,9 +79,18 @@ def _invoke( wan_t5_encoder_model=_model("t5"), component_source=component, ) + return invocation, context + + +def _invoke(*args, **kwargs): + invocation, context = _prepare(*args, **kwargs) return invocation.invoke(context) +def _warnings(context: MagicMock) -> list[str]: + return [call.args[0] for call in context.logger.warning.call_args_list] + + @pytest.mark.parametrize("variant", [WanVariantType.T2V_A14B, WanVariantType.I2V_A14B]) @pytest.mark.parametrize("main_expert,low_expert", [("high", "low"), ("low", "high")]) def test_gguf_loader_accepts_valid_expert_pair_in_either_order( @@ -108,8 +118,8 @@ def test_gguf_loader_accepts_valid_expert_pair_in_either_order( _config("low", WanVariantType.T2V_A14B, "high"), ), ( - _config("main", WanVariantType.T2V_A14B, "high"), - _config("low", WanVariantType.T2V_A14B, "none"), + _config("main", WanVariantType.T2V_A14B, "low"), + _config("low", WanVariantType.T2V_A14B, "low"), ), ], ) @@ -118,6 +128,33 @@ def test_gguf_loader_rejects_invalid_expert_pair(main_config: SimpleNamespace, l _invoke(main_config, low_config) +@pytest.mark.parametrize( + "main_expert,low_expert,expected_high_key", + [ + # The expert tag comes from a filename heuristic, so untagged community + # finetunes are common. The wiring is explicit intent: take the untagged + # file at its wired position, or as the complement of a tagged partner. + ("none", "none", "main"), + ("high", "none", "main"), + ("none", "low", "main"), + ("none", "high", "low"), + ("low", "none", "low"), + ], +) +def test_gguf_loader_falls_back_to_wiring_for_untagged_experts( + main_expert: str, low_expert: str, expected_high_key: str +) -> None: + output = _invoke( + _config("main", WanVariantType.I2V_A14B, main_expert), + _config("low", WanVariantType.I2V_A14B, low_expert), + ) + + expected_low_key = "low" if expected_high_key == "main" else "main" + assert output.transformer.transformer.key == expected_high_key + assert output.transformer.transformer_low_noise is not None + assert output.transformer.transformer_low_noise.key == expected_low_key + + @pytest.mark.parametrize("low_variant", [WanVariantType.TI2V_5B, WanVariantType.T2V_A14B]) def test_ti2v_5b_main_ignores_wired_low_noise_model(low_variant: WanVariantType) -> None: """The field docs promise 'Transformer (Low Noise)' is ignored for the single-expert @@ -131,10 +168,66 @@ def test_ti2v_5b_main_ignores_wired_low_noise_model(low_variant: WanVariantType) assert output.transformer.transformer_low_noise is None -@pytest.mark.parametrize("expert", ["low", "none"]) -def test_gguf_loader_rejects_non_high_primary_without_pair(expert: str) -> None: - with pytest.raises(ValueError, match="high-noise"): - _invoke(_config("main", WanVariantType.T2V_A14B, expert)) +@pytest.mark.parametrize("expert", ["high", "low", "none"]) +def test_gguf_loader_runs_unpaired_primary_whatever_its_tag(expert: str) -> None: + """A single wired transformer is explicit intent just like a pair is, and the tag is + only a filename guess — so an unpaired A14B runs with a warning rather than aborting.""" + invocation, context = _prepare(_config("main", WanVariantType.T2V_A14B, expert)) + output = invocation.invoke(context) + + assert output.transformer.transformer.key == "main" + assert output.transformer.transformer_low_noise is None + assert any("only this one expert will run" in warning.lower() for warning in _warnings(context)) + + +def test_gguf_loader_hints_at_the_expert_swap_for_an_unpaired_low_noise_model() -> None: + invocation, context = _prepare(_config("main", WanVariantType.T2V_A14B, "low")) + invocation.invoke(context) + + assert any("high-noise one is usually the better choice" in warning for warning in _warnings(context)) + + +def test_gguf_loader_rejects_the_same_model_in_both_transformer_slots() -> None: + """Wiring one model twice used to fail the {high, low} pair check. It must stay an error: + the denoiser would unload and reload the same multi-GB expert at every boundary crossing.""" + main_config = _config("main", WanVariantType.T2V_A14B, "high") + with pytest.raises(ValueError, match="same model"): + _invoke(main_config, main_config, low_key="main") + + +@pytest.mark.parametrize("main_expert,low_expert", [("low", "high"), ("low", "none"), ("none", "high")]) +def test_gguf_loader_warns_when_it_swaps_the_wired_experts(main_expert: str, low_expert: str) -> None: + """The swap overrides explicit wiring on the strength of a filename tag, so a mistagged + file must not invert the two experts silently.""" + invocation, context = _prepare( + _config("main", WanVariantType.I2V_A14B, main_expert), + _config("low", WanVariantType.I2V_A14B, low_expert), + ) + output = invocation.invoke(context) + + assert output.transformer.transformer.key == "low" + assert any("swapped" in warning for warning in _warnings(context)) + + +@pytest.mark.parametrize("main_expert,low_expert", [("high", "low"), ("high", "none"), ("none", "low")]) +def test_gguf_loader_is_quiet_when_the_wiring_stands(main_expert: str, low_expert: str) -> None: + invocation, context = _prepare( + _config("main", WanVariantType.I2V_A14B, main_expert), + _config("low", WanVariantType.I2V_A14B, low_expert), + ) + invocation.invoke(context) + + assert _warnings(context) == [] + + +def test_gguf_loader_warns_when_neither_expert_is_tagged() -> None: + invocation, context = _prepare( + _config("main", WanVariantType.I2V_A14B, "none"), + _config("low", WanVariantType.I2V_A14B, "none"), + ) + invocation.invoke(context) + + assert any("Neither Wan A14B GGUF filename identifies its expert" in warning for warning in _warnings(context)) @pytest.mark.parametrize(