优化Qwen图像节点

This commit is contained in:
Gin3601
2026-05-20 17:30:22 +08:00
parent be7d7e68c2
commit ce6ac6b0d4
6 changed files with 1817 additions and 98 deletions
@@ -0,0 +1,933 @@
# Qwen Advanced Parameters Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Extend `LLSQwenTextToImage` and `LLSQwenImageEdit` with official/common advanced controls, optional ordered user LoRA stack input, optional turbo/lightning LoRA support, and full regression coverage while preserving the compressed single-node UX.
**Architecture:** Keep the current dedicated `qwen/` package and expand it in place. `qwen/discovery.py` will own advanced choice lists, ordered user LoRA choice lists, turbo LoRA filtering, auto-resolution, and turbo presets; `qwen/runtime.py` will own advanced parameter passthrough plus ordered model-only LoRA application; `qwen/nodes.py` will expose the expanded Qwen node schemas and add one compact `LLSQwenLoRAStack` helper while the main generation nodes still return only `IMAGE`.
**Tech Stack:** Python, unittest, ComfyUI core nodes, `comfy_extras.nodes_qwen`, `comfy_extras.nodes_flux`, `comfy_extras.nodes_model_advanced`, `comfy_extras.nodes_cfg`
---
## File Structure
- Modify: `qwen/discovery.py`
- Add user LoRA stack choice lists, turbo/lightning LoRA discovery, `(auto)` handling, compatible LoRA validation, and turbo preset lookup.
- Modify: `qwen/runtime.py`
- Add advanced parameter passthrough, ordered model-only user LoRA application, turbo preset overrides, and image-edit multi-reference handling.
- Modify: `qwen/nodes.py`
- Expand the two public node schemas with advanced inputs, add optional `lora_stack` inputs, and add the compact `LLSQwenLoRAStack` helper while preserving hidden resource loading and `IMAGE`-only outputs.
- Modify: `tests/test_qwen_nodes.py`
- Add schema, discovery, ordered LoRA stack, passthrough, turbo, and failure-path tests with small deterministic stubs.
## Plan Amendment
This plan now assumes two changes beyond the original draft:
- both public Qwen generation nodes accept an optional ordered `lora_stack` input
- `qwen_image_edit_2511_*` turbo/lightning support is allowed when a matching 2511 LoRA exists locally
The intended internal model application order is:
- base Qwen model
- ordered user LoRA stack entries
- optional turbo/lightning LoRA
### Task 1: Expand schemas and turbo LoRA discovery
**Files:**
- Modify: `qwen/discovery.py`
- Modify: `qwen/nodes.py`
- Modify: `tests/test_qwen_nodes.py`
- Test: `tests/test_qwen_nodes.py`
- [ ] **Step 1: Write the failing tests**
```python
class QwenFolderPathsStub:
def __init__(self):
self._files = {
"diffusion_models": [
"qwen_image_fp8_e4m3fn.safetensors",
"qwen_image_2512_fp8_e4m3fn.safetensors",
"qwen_image_edit_2509_fp8_e4m3fn.safetensors",
"qwen_image_edit_2511_bf16.safetensors",
"qwen_image_layered_bf16.safetensors",
"flux1-dev.safetensors",
],
"text_encoders": ["qwen_2.5_vl_7b_fp8_scaled.safetensors"],
"vae": ["qwen_image_vae.safetensors"],
"loras": [
"Qwen-Image-Lightning-8steps-V1.0.safetensors",
"Qwen-Image-2512-Lightning-4steps-V1.0-fp32.safetensors",
"Qwen-Image-Edit-2509-Lightning-4steps-V1.0-bf16.safetensors",
"flux-dev-style.safetensors",
],
}
def get_filename_list(self, category):
return list(self._files.get(category, []))
class TestQwenNodes(unittest.TestCase):
def test_qwen_text_node_schema_exposes_advanced_inputs(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSQwenTextToImage"]
required = node_cls.INPUT_TYPES()["required"]
self.assertEqual(
tuple(required.keys()),
(
"model_name",
"prompt",
"width",
"height",
"steps",
"seed",
"batch_size",
"negative_prompt",
"cfg",
"sampler_name",
"scheduler",
"shift",
"enable_turbo_mode",
"turbo_lora_name",
"turbo_strength",
),
)
self.assertEqual(required["negative_prompt"][1]["default"], "")
self.assertEqual(required["cfg"][1]["default"], 4.0)
self.assertIn("euler", required["sampler_name"][0])
self.assertIn("simple", required["scheduler"][0])
self.assertEqual(required["shift"][1]["default"], 3.1)
self.assertEqual(required["enable_turbo_mode"][1]["default"], False)
def test_qwen_edit_node_schema_exposes_advanced_inputs_and_optional_images(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSQwenImageEdit"]
required = node_cls.INPUT_TYPES()["required"]
optional = node_cls.INPUT_TYPES()["optional"]
self.assertEqual(
tuple(required.keys()),
(
"model_name",
"image",
"prompt",
"steps",
"seed",
"negative_prompt",
"cfg",
"sampler_name",
"scheduler",
"shift",
"cfg_norm_strength",
"reference_latents_method",
"enable_turbo_mode",
"turbo_lora_name",
"turbo_strength",
),
)
self.assertEqual(optional["image2"][0], "IMAGE")
self.assertEqual(optional["image3"][0], "IMAGE")
self.assertEqual(
required["reference_latents_method"][0],
["offset", "index", "uxo/uno", "index_timestep_zero"],
)
def test_qwen_text_node_turbo_lora_choices_filter_only_text_loras(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import nodes as qwen_nodes
with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()):
choices = qwen_nodes.LLSQwenTextToImage.INPUT_TYPES()["required"]["turbo_lora_name"][0]
self.assertEqual(
choices,
[
"(auto)",
"Qwen-Image-2512-Lightning-4steps-V1.0-fp32.safetensors",
"Qwen-Image-Lightning-8steps-V1.0.safetensors",
],
)
def test_qwen_edit_node_turbo_lora_choices_filter_only_edit_loras(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import nodes as qwen_nodes
with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()):
choices = qwen_nodes.LLSQwenImageEdit.INPUT_TYPES()["required"]["turbo_lora_name"][0]
self.assertEqual(
choices,
[
"(auto)",
"Qwen-Image-Edit-2509-Lightning-4steps-V1.0-bf16.safetensors",
],
)
```
- [ ] **Step 2: Run the focused tests to verify they fail**
Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_schema_exposes_advanced_inputs tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_node_schema_exposes_advanced_inputs_and_optional_images tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_turbo_lora_choices_filter_only_text_loras tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_node_turbo_lora_choices_filter_only_edit_loras -v`
Expected: FAIL because the node schemas do not yet expose advanced inputs and discovery does not yet expose turbo LoRA choices.
- [ ] **Step 3: Write the minimal discovery and schema implementation**
```python
# qwen/discovery.py
AUTO_TURBO_LORA_CHOICE = "(auto)"
NO_TEXT_TURBO_LORA_PLACEHOLDER = "(no qwen text turbo loras found)"
NO_EDIT_TURBO_LORA_PLACEHOLDER = "(no qwen edit turbo loras found)"
REFERENCE_LATENTS_METHOD_CHOICES = ["offset", "index", "uxo/uno", "index_timestep_zero"]
def is_qwen_text_turbo_lora(name: str | None) -> bool:
lowered = (name or "").lower()
return "qwen-image" in lowered and "lightning" in lowered and "edit" not in lowered
def is_qwen_edit_turbo_lora(name: str | None) -> bool:
lowered = (name or "").lower()
return "qwen-image-edit" in lowered and "lightning" in lowered
def get_qwen_text_turbo_lora_choices() -> list[str]:
names = _sorted_unique(
[name for name in _get_filename_list("loras") if is_qwen_text_turbo_lora(name)]
)
if not names:
return [AUTO_TURBO_LORA_CHOICE, NO_TEXT_TURBO_LORA_PLACEHOLDER]
return [AUTO_TURBO_LORA_CHOICE] + names
def get_qwen_edit_turbo_lora_choices() -> list[str]:
names = _sorted_unique(
[name for name in _get_filename_list("loras") if is_qwen_edit_turbo_lora(name)]
)
if not names:
return [AUTO_TURBO_LORA_CHOICE, NO_EDIT_TURBO_LORA_PLACEHOLDER]
return [AUTO_TURBO_LORA_CHOICE] + names
```
```python
# qwen/nodes.py
from ..sampling.nodes import _get_samplers, _get_schedulers
class LLSQwenTextToImage:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"model_name": (discovery.get_qwen_text_model_choices(),),
"prompt": ("STRING", {"default": "", "multiline": True}),
"width": ("INT", {"default": 1024, "min": 16, "max": _MAX_RESOLUTION, "step": 16}),
"height": ("INT", {"default": 1024, "min": 16, "max": _MAX_RESOLUTION, "step": 16}),
"steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
"seed": ("INT", {"default": 0, "min": 0, "max": 0xFFFFFFFFFFFFFFFF}),
"batch_size": ("INT", {"default": 1, "min": 1, "max": 64}),
"negative_prompt": ("STRING", {"default": "", "multiline": True, "advanced": True}),
"cfg": ("FLOAT", {"default": 4.0, "min": 0.0, "max": 100.0, "step": 0.1, "advanced": True}),
"sampler_name": (_get_samplers(), {"default": "euler", "advanced": True}),
"scheduler": (_get_schedulers(), {"default": "simple", "advanced": True}),
"shift": ("FLOAT", {"default": 3.1, "min": 0.0, "max": 100.0, "step": 0.01, "advanced": True}),
"enable_turbo_mode": ("BOOLEAN", {"default": False, "advanced": True}),
"turbo_lora_name": (discovery.get_qwen_text_turbo_lora_choices(), {"default": discovery.AUTO_TURBO_LORA_CHOICE, "advanced": True}),
"turbo_strength": ("FLOAT", {"default": 1.0, "min": -100.0, "max": 100.0, "step": 0.01, "advanced": True}),
}
}
```
```python
class LLSQwenImageEdit:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"model_name": (discovery.get_qwen_edit_model_choices(),),
"image": ("IMAGE",),
"prompt": ("STRING", {"default": "", "multiline": True}),
"steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
"seed": ("INT", {"default": 0, "min": 0, "max": 0xFFFFFFFFFFFFFFFF}),
"negative_prompt": ("STRING", {"default": "", "multiline": True, "advanced": True}),
"cfg": ("FLOAT", {"default": 4.0, "min": 0.0, "max": 100.0, "step": 0.1, "advanced": True}),
"sampler_name": (_get_samplers(), {"default": "euler", "advanced": True}),
"scheduler": (_get_schedulers(), {"default": "simple", "advanced": True}),
"shift": ("FLOAT", {"default": 3.1, "min": 0.0, "max": 100.0, "step": 0.01, "advanced": True}),
"cfg_norm_strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 100.0, "step": 0.01, "advanced": True}),
"reference_latents_method": (discovery.REFERENCE_LATENTS_METHOD_CHOICES, {"default": "index_timestep_zero", "advanced": True}),
"enable_turbo_mode": ("BOOLEAN", {"default": False, "advanced": True}),
"turbo_lora_name": (discovery.get_qwen_edit_turbo_lora_choices(), {"default": discovery.AUTO_TURBO_LORA_CHOICE, "advanced": True}),
"turbo_strength": ("FLOAT", {"default": 1.0, "min": -100.0, "max": 100.0, "step": 0.01, "advanced": True}),
},
"optional": {
"image2": ("IMAGE",),
"image3": ("IMAGE",),
},
}
```
- [ ] **Step 4: Run the focused tests to verify they pass**
Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_schema_exposes_advanced_inputs tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_node_schema_exposes_advanced_inputs_and_optional_images tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_turbo_lora_choices_filter_only_text_loras tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_node_turbo_lora_choices_filter_only_edit_loras -v`
Expected: PASS with the expanded schemas and correctly filtered turbo LoRA choices.
- [ ] **Step 5: Commit**
```bash
git add qwen/discovery.py qwen/nodes.py tests/test_qwen_nodes.py
git commit -m "feat: expand qwen node schemas"
```
### Task 2: Add text-to-image advanced runtime controls and turbo support
**Files:**
- Modify: `qwen/discovery.py`
- Modify: `qwen/runtime.py`
- Modify: `tests/test_qwen_nodes.py`
- Test: `tests/test_qwen_nodes.py`
- [ ] **Step 1: Write the failing runtime tests**
```python
class CoreQwenNodesStub:
last_lora_call = None
@classmethod
def reset(cls):
cls.last_lora_call = None
cls.last_unet_call = None
cls.last_clip_call = None
cls.last_text_encode_calls = []
cls.last_ksampler_call = None
cls.last_vae_encode_call = None
cls.last_vae_decode_call = None
cls.last_vae_load_call = None
class LoraLoaderModelOnly:
def load_lora_model_only(self, model, lora_name, strength_model):
CoreQwenNodesStub.last_lora_call = {
"model": model,
"lora_name": lora_name,
"strength_model": strength_model,
}
return (f"LORA::{model}::{lora_name}::{strength_model}",)
class TestQwenNodes(unittest.TestCase):
def test_qwen_text_runtime_passes_negative_prompt_and_sampling_controls(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import runtime as qwen_runtime
stub = QwenFolderPathsStub()
with mock.patch.object(qwen_discovery, "folder_paths", stub), \
mock.patch.object(qwen_runtime, "folder_paths", stub), \
mock.patch.object(qwen_runtime, "comfy_core_nodes", CoreQwenNodesStub), \
mock.patch.object(qwen_runtime, "nodes_sd3", SD3NodesStub), \
mock.patch.object(qwen_runtime, "nodes_model_advanced", ModelAdvancedStub):
qwen_runtime.run_qwen_text_to_image(
model_name="qwen_image_fp8_e4m3fn.safetensors",
prompt="a cat",
negative_prompt="low quality",
width=1024,
height=1024,
steps=33,
seed=99,
batch_size=2,
cfg=7.5,
sampler_name="heun",
scheduler="normal",
shift=2.25,
enable_turbo_mode=False,
turbo_lora_name="(auto)",
turbo_strength=1.0,
)
self.assertEqual(CoreQwenNodesStub.last_text_encode_calls[1]["text"], "low quality")
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["cfg"], 7.5)
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["sampler_name"], "heun")
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["scheduler"], "normal")
self.assertEqual(ModelAdvancedStub.last_call["shift"], 2.25)
def test_qwen_text_runtime_enables_turbo_lora_and_uses_turbo_preset(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import runtime as qwen_runtime
stub = QwenFolderPathsStub()
with mock.patch.object(qwen_discovery, "folder_paths", stub), \
mock.patch.object(qwen_runtime, "folder_paths", stub), \
mock.patch.object(qwen_runtime, "comfy_core_nodes", CoreQwenNodesStub), \
mock.patch.object(qwen_runtime, "nodes_sd3", SD3NodesStub), \
mock.patch.object(qwen_runtime, "nodes_model_advanced", ModelAdvancedStub):
qwen_runtime.run_qwen_text_to_image(
model_name="qwen_image_fp8_e4m3fn.safetensors",
prompt="a cat",
negative_prompt="",
width=1024,
height=1024,
steps=50,
seed=99,
batch_size=1,
cfg=8.0,
sampler_name="euler",
scheduler="simple",
shift=3.1,
enable_turbo_mode=True,
turbo_lora_name="(auto)",
turbo_strength=0.75,
)
self.assertEqual(
CoreQwenNodesStub.last_lora_call["lora_name"],
"Qwen-Image-Lightning-8steps-V1.0.safetensors",
)
self.assertEqual(CoreQwenNodesStub.last_lora_call["strength_model"], 0.75)
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["steps"], 8)
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["cfg"], 1.0)
def test_qwen_text_runtime_rejects_incompatible_manual_turbo_lora(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import runtime as qwen_runtime
stub = QwenFolderPathsStub()
with mock.patch.object(qwen_discovery, "folder_paths", stub), mock.patch.object(
qwen_runtime,
"folder_paths",
stub,
):
with self.assertRaisesRegex(RuntimeError, "Turbo LoRA .* is not compatible"):
qwen_runtime.run_qwen_text_to_image(
model_name="qwen_image_fp8_e4m3fn.safetensors",
prompt="a cat",
negative_prompt="",
width=1024,
height=1024,
steps=20,
seed=1,
batch_size=1,
cfg=4.0,
sampler_name="euler",
scheduler="simple",
shift=3.1,
enable_turbo_mode=True,
turbo_lora_name="Qwen-Image-Edit-2509-Lightning-4steps-V1.0-bf16.safetensors",
turbo_strength=1.0,
)
```
- [ ] **Step 2: Run the focused tests to verify they fail**
Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_passes_negative_prompt_and_sampling_controls tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_enables_turbo_lora_and_uses_turbo_preset tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_rejects_incompatible_manual_turbo_lora -v`
Expected: FAIL because the runtime still hardcodes empty negative prompt, fixed `cfg`/sampler/scheduler/shift, and has no turbo LoRA logic.
- [ ] **Step 3: Write the minimal discovery and runtime implementation**
```python
# qwen/discovery.py
_QWEN_TURBO_PROFILES = {
"qwen_image_fp8_e4m3fn.safetensors": {"steps": 8, "cfg": 1.0},
"qwen_image_2512_fp8_e4m3fn.safetensors": {"steps": 4, "cfg": 1.0},
"qwen_image_edit_2509_fp8_e4m3fn.safetensors": {"steps": 4, "cfg": 1.0},
}
def get_qwen_turbo_profile(model_name: str) -> dict[str, float] | None:
return _QWEN_TURBO_PROFILES.get(model_name)
def resolve_qwen_text_turbo_lora(model_name: str, requested_name: str) -> str:
available = [name for name in _get_filename_list("loras") if is_qwen_text_turbo_lora(name)]
if "2512" in model_name.lower():
candidates = [name for name in available if "2512" in name.lower()]
else:
candidates = [name for name in available if "2512" not in name.lower()]
if requested_name == AUTO_TURBO_LORA_CHOICE:
if candidates:
return _sorted_unique(candidates)[0]
raise RuntimeError(f"[LLS] No compatible turbo/lightning LoRA exists for '{model_name}'.")
if requested_name not in available or requested_name not in candidates:
raise RuntimeError(
f"[LLS] Turbo LoRA '{requested_name}' is not compatible with text-to-image model '{model_name}'."
)
return requested_name
```
```python
# qwen/runtime.py
def _patch_qwen_model_sampling(model, shift: float):
patcher_cls = _require_class(nodes_model_advanced, "ModelSamplingAuraFlow", _MODEL_ADVANCED_ERR)
return patcher_cls().patch_aura(model, float(shift))[0]
def _sample_qwen(model, positive, negative, latent, steps: int, seed: int, cfg: float, sampler_name: str, scheduler: str):
sampler_cls = _require_class(comfy_core_nodes, "KSampler", _CORE_NODES_ERR)
return sampler_cls().sample(
model,
int(seed),
int(steps),
float(cfg),
sampler_name,
scheduler,
positive,
negative,
latent,
denoise=1.0,
)[0]
def _load_model_only_lora(model, lora_name: str, strength_model: float):
lora_loader_cls = _require_class(comfy_core_nodes, "LoraLoaderModelOnly", _CORE_NODES_ERR)
return lora_loader_cls().load_lora_model_only(model, lora_name, float(strength_model))[0]
def _apply_text_turbo(model, model_name: str, steps: int, cfg: float, enable_turbo_mode: bool, turbo_lora_name: str, turbo_strength: float):
if not enable_turbo_mode:
return model, int(steps), float(cfg)
resolved_lora = discovery.resolve_qwen_text_turbo_lora(model_name, turbo_lora_name)
profile = discovery.get_qwen_turbo_profile(model_name)
if profile is None:
raise RuntimeError(f"[LLS] No supported turbo preset exists for '{model_name}'.")
model = _load_model_only_lora(model, resolved_lora, turbo_strength)
return model, int(profile["steps"]), float(profile["cfg"])
def run_qwen_text_to_image(
model_name: str,
prompt: str,
negative_prompt: str,
width: int,
height: int,
steps: int,
seed: int,
batch_size: int,
cfg: float,
sampler_name: str,
scheduler: str,
shift: float,
enable_turbo_mode: bool,
turbo_lora_name: str,
turbo_strength: float,
):
model_name, clip_name, vae_name = _resolve_qwen_resources(
model_name,
discovery.validate_qwen_text_model_name,
)
model = _load_qwen_model(model_name)
model, effective_steps, effective_cfg = _apply_text_turbo(
model,
model_name,
steps,
cfg,
enable_turbo_mode,
turbo_lora_name,
turbo_strength,
)
model = _patch_qwen_model_sampling(model, shift)
clip = _load_qwen_clip(clip_name)
vae = _load_qwen_vae(vae_name)
positive = _encode_clip_text(clip, prompt)
negative = _encode_clip_text(clip, negative_prompt)
latent = _create_empty_qwen_latent(int(width), int(height), int(batch_size))
sampled = _sample_qwen(model, positive, negative, latent, effective_steps, int(seed), effective_cfg, sampler_name, scheduler)
return _decode_qwen_latent(vae, sampled)
```
- [ ] **Step 4: Run the focused tests to verify they pass**
Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_passes_negative_prompt_and_sampling_controls tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_enables_turbo_lora_and_uses_turbo_preset tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_rejects_incompatible_manual_turbo_lora -v`
Expected: PASS with advanced text controls routed through and turbo mode correctly applying the text turbo LoRA and preset.
- [ ] **Step 5: Commit**
```bash
git add qwen/discovery.py qwen/runtime.py tests/test_qwen_nodes.py
git commit -m "feat: add qwen text advanced controls"
```
### Task 3: Add image-edit advanced runtime controls and turbo support
**Files:**
- Modify: `qwen/discovery.py`
- Modify: `qwen/runtime.py`
- Modify: `qwen/nodes.py`
- Modify: `tests/test_qwen_nodes.py`
- Test: `tests/test_qwen_nodes.py`
- [ ] **Step 1: Write the failing runtime tests**
```python
class TestQwenNodes(unittest.TestCase):
def test_qwen_edit_runtime_passes_optional_images_negative_prompt_and_edit_controls(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import runtime as qwen_runtime
stub = QwenFolderPathsStub()
with mock.patch.object(qwen_discovery, "folder_paths", stub), \
mock.patch.object(qwen_runtime, "folder_paths", stub), \
mock.patch.object(qwen_runtime, "comfy_core_nodes", CoreQwenNodesStub), \
mock.patch.object(qwen_runtime, "nodes_model_advanced", ModelAdvancedStub), \
mock.patch.object(qwen_runtime, "nodes_qwen", QwenExtraNodesStub), \
mock.patch.object(qwen_runtime, "nodes_flux", FluxNodesStub), \
mock.patch.object(qwen_runtime, "nodes_cfg", CFGNodesStub):
qwen_runtime.run_qwen_image_edit(
model_name="qwen_image_edit_2509_fp8_e4m3fn.safetensors",
image="IMAGE::1",
image2="IMAGE::2",
image3="IMAGE::3",
prompt="turn the cat blue",
negative_prompt="bad anatomy",
steps=30,
seed=123,
cfg=2.5,
sampler_name="heun",
scheduler="normal",
shift=3.0,
cfg_norm_strength=0.75,
reference_latents_method="index",
enable_turbo_mode=False,
turbo_lora_name="(auto)",
turbo_strength=1.0,
)
self.assertEqual(QwenExtraNodesStub.last_calls[0]["image1"], "SCALED::IMAGE::1")
self.assertEqual(QwenExtraNodesStub.last_calls[0]["image2"], "IMAGE::2")
self.assertEqual(QwenExtraNodesStub.last_calls[0]["image3"], "IMAGE::3")
self.assertEqual(QwenExtraNodesStub.last_calls[1]["prompt"], "bad anatomy")
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["cfg"], 2.5)
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["sampler_name"], "heun")
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["scheduler"], "normal")
self.assertEqual(ModelAdvancedStub.last_call["shift"], 3.0)
self.assertEqual(CFGNodesStub.last_call["strength"], 0.75)
self.assertEqual(FluxNodesStub.last_reference_calls[0]["reference_latents_method"], "index")
def test_qwen_edit_runtime_enables_turbo_lora_and_uses_turbo_preset(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import runtime as qwen_runtime
stub = QwenFolderPathsStub()
with mock.patch.object(qwen_discovery, "folder_paths", stub), \
mock.patch.object(qwen_runtime, "folder_paths", stub), \
mock.patch.object(qwen_runtime, "comfy_core_nodes", CoreQwenNodesStub), \
mock.patch.object(qwen_runtime, "nodes_model_advanced", ModelAdvancedStub), \
mock.patch.object(qwen_runtime, "nodes_qwen", QwenExtraNodesStub), \
mock.patch.object(qwen_runtime, "nodes_flux", FluxNodesStub), \
mock.patch.object(qwen_runtime, "nodes_cfg", CFGNodesStub):
qwen_runtime.run_qwen_image_edit(
model_name="qwen_image_edit_2509_fp8_e4m3fn.safetensors",
image="IMAGE::1",
image2=None,
image3=None,
prompt="turn the cat blue",
negative_prompt="",
steps=40,
seed=123,
cfg=4.0,
sampler_name="euler",
scheduler="simple",
shift=3.0,
cfg_norm_strength=1.0,
reference_latents_method="index_timestep_zero",
enable_turbo_mode=True,
turbo_lora_name="(auto)",
turbo_strength=0.8,
)
self.assertEqual(
CoreQwenNodesStub.last_lora_call["lora_name"],
"Qwen-Image-Edit-2509-Lightning-4steps-V1.0-bf16.safetensors",
)
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["steps"], 4)
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["cfg"], 1.0)
def test_qwen_edit_runtime_fails_when_2511_turbo_is_requested(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import runtime as qwen_runtime
stub = QwenFolderPathsStub()
with mock.patch.object(qwen_discovery, "folder_paths", stub), mock.patch.object(
qwen_runtime,
"folder_paths",
stub,
):
with self.assertRaisesRegex(RuntimeError, "No compatible turbo/lightning LoRA exists"):
qwen_runtime.run_qwen_image_edit(
model_name="qwen_image_edit_2511_bf16.safetensors",
image="IMAGE::1",
image2=None,
image3=None,
prompt="turn the cat blue",
negative_prompt="",
steps=20,
seed=1,
cfg=4.0,
sampler_name="euler",
scheduler="simple",
shift=3.1,
cfg_norm_strength=1.0,
reference_latents_method="index_timestep_zero",
enable_turbo_mode=True,
turbo_lora_name="(auto)",
turbo_strength=1.0,
)
```
- [ ] **Step 2: Run the focused tests to verify they fail**
Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_runtime_passes_optional_images_negative_prompt_and_edit_controls tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_runtime_enables_turbo_lora_and_uses_turbo_preset tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_runtime_fails_when_2511_turbo_is_requested -v`
Expected: FAIL because the image-edit runtime does not yet pass optional images or advanced controls through and has no edit turbo logic.
- [ ] **Step 3: Write the minimal edit runtime implementation**
```python
# qwen/discovery.py
def resolve_qwen_edit_turbo_lora(model_name: str, requested_name: str) -> str:
available = [name for name in _get_filename_list("loras") if is_qwen_edit_turbo_lora(name)]
if "2509" in model_name.lower():
candidates = [name for name in available if "2509" in name.lower()]
else:
candidates = []
if requested_name == AUTO_TURBO_LORA_CHOICE:
if candidates:
return _sorted_unique(candidates)[0]
raise RuntimeError(f"[LLS] No compatible turbo/lightning LoRA exists for '{model_name}'.")
if requested_name not in available or requested_name not in candidates:
raise RuntimeError(
f"[LLS] Turbo LoRA '{requested_name}' is not compatible with image-edit model '{model_name}'."
)
return requested_name
```
```python
# qwen/runtime.py
def _patch_qwen_model_sampling(model, shift: float):
patcher_cls = _require_class(nodes_model_advanced, "ModelSamplingAuraFlow", _MODEL_ADVANCED_ERR)
return patcher_cls().patch_aura(model, float(shift))[0]
def _apply_cfg_norm(model, strength: float):
cfg_cls = _require_class(nodes_cfg, "CFGNorm", _CFG_ERR)
return _unwrap_first(cfg_cls.execute(model, float(strength)))
def _apply_reference_latents_method(conditioning, reference_latents_method: str):
method_cls = _require_class(nodes_flux, "FluxKontextMultiReferenceLatentMethod", _FLUX_ERR)
return _unwrap_first(method_cls.execute(conditioning, reference_latents_method))
def _encode_qwen_edit_conditioning(clip, prompt: str, vae, image1, image2=None, image3=None):
encoder_cls = _require_class(nodes_qwen, "TextEncodeQwenImageEditPlus", _QWEN_ERR)
return _unwrap_first(
encoder_cls.execute(clip, prompt, vae=vae, image1=image1, image2=image2, image3=image3)
)
def _apply_edit_turbo(model, model_name: str, steps: int, cfg: float, enable_turbo_mode: bool, turbo_lora_name: str, turbo_strength: float):
if not enable_turbo_mode:
return model, int(steps), float(cfg)
resolved_lora = discovery.resolve_qwen_edit_turbo_lora(model_name, turbo_lora_name)
profile = discovery.get_qwen_turbo_profile(model_name)
if profile is None:
raise RuntimeError(f"[LLS] No supported turbo preset exists for '{model_name}'.")
model = _load_model_only_lora(model, resolved_lora, turbo_strength)
return model, int(profile["steps"]), float(profile["cfg"])
def run_qwen_image_edit(
model_name: str,
image,
image2,
image3,
prompt: str,
negative_prompt: str,
steps: int,
seed: int,
cfg: float,
sampler_name: str,
scheduler: str,
shift: float,
cfg_norm_strength: float,
reference_latents_method: str,
enable_turbo_mode: bool,
turbo_lora_name: str,
turbo_strength: float,
):
if image is None:
raise RuntimeError("[LLS] Missing source image. Connect an IMAGE input.")
model_name, clip_name, vae_name = _resolve_qwen_resources(
model_name,
discovery.validate_qwen_edit_model_name,
)
model = _load_qwen_model(model_name)
model, effective_steps, effective_cfg = _apply_edit_turbo(
model,
model_name,
steps,
cfg,
enable_turbo_mode,
turbo_lora_name,
turbo_strength,
)
model = _patch_qwen_model_sampling(model, shift)
clip = _load_qwen_clip(clip_name)
vae = _load_qwen_vae(vae_name)
scaled = _scale_qwen_edit_image(image)
positive = _encode_qwen_edit_conditioning(clip, prompt, vae, scaled, image2=image2, image3=image3)
negative = _encode_qwen_edit_conditioning(clip, negative_prompt, vae, scaled, image2=image2, image3=image3)
positive = _apply_reference_latents_method(positive, reference_latents_method)
negative = _apply_reference_latents_method(negative, reference_latents_method)
latent = _encode_image_to_latent(vae, scaled)
model = _apply_cfg_norm(model, cfg_norm_strength)
sampled = _sample_qwen(model, positive, negative, latent, effective_steps, int(seed), effective_cfg, sampler_name, scheduler)
return _decode_qwen_latent(vae, sampled)
```
- [ ] **Step 4: Update node method calls to pass the new parameters**
```python
# qwen/nodes.py
def generate(
self,
model_name: str,
prompt: str,
width: int,
height: int,
steps: int,
seed: int,
batch_size: int,
negative_prompt: str,
cfg: float,
sampler_name: str,
scheduler: str,
shift: float,
enable_turbo_mode: bool,
turbo_lora_name: str,
turbo_strength: float,
):
return (
runtime.run_qwen_text_to_image(
model_name=model_name,
prompt=prompt,
negative_prompt=negative_prompt,
width=width,
height=height,
steps=steps,
seed=seed,
batch_size=batch_size,
cfg=cfg,
sampler_name=sampler_name,
scheduler=scheduler,
shift=shift,
enable_turbo_mode=enable_turbo_mode,
turbo_lora_name=turbo_lora_name,
turbo_strength=turbo_strength,
),
)
```
```python
def generate(
self,
model_name: str,
image,
prompt: str,
steps: int,
seed: int,
negative_prompt: str,
cfg: float,
sampler_name: str,
scheduler: str,
shift: float,
cfg_norm_strength: float,
reference_latents_method: str,
enable_turbo_mode: bool,
turbo_lora_name: str,
turbo_strength: float,
image2=None,
image3=None,
):
return (
runtime.run_qwen_image_edit(
model_name=model_name,
image=image,
image2=image2,
image3=image3,
prompt=prompt,
negative_prompt=negative_prompt,
steps=steps,
seed=seed,
cfg=cfg,
sampler_name=sampler_name,
scheduler=scheduler,
shift=shift,
cfg_norm_strength=cfg_norm_strength,
reference_latents_method=reference_latents_method,
enable_turbo_mode=enable_turbo_mode,
turbo_lora_name=turbo_lora_name,
turbo_strength=turbo_strength,
),
)
```
- [ ] **Step 5: Run the focused tests to verify they pass**
Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_runtime_passes_optional_images_negative_prompt_and_edit_controls tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_runtime_enables_turbo_lora_and_uses_turbo_preset tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_runtime_fails_when_2511_turbo_is_requested -v`
Expected: PASS with image-edit advanced controls routed through, edit turbo working for 2509, and 2511 turbo failing clearly.
- [ ] **Step 6: Commit**
```bash
git add qwen/discovery.py qwen/runtime.py qwen/nodes.py tests/test_qwen_nodes.py
git commit -m "feat: add qwen edit advanced controls"
```
### Task 4: Run dedicated and full regression verification
**Files:**
- Modify: `tests/test_qwen_nodes.py` (only if a regression run exposes a mismatch)
- Test: `tests/test_qwen_nodes.py`
- [ ] **Step 1: Run the dedicated Qwen test file**
Run: `python3 -m unittest discover -s tests -p 'test_qwen_nodes.py' -v`
Expected: PASS with zero failures across schema, discovery, advanced passthrough, turbo, and failure-path coverage.
- [ ] **Step 2: Run the full plugin test suite**
Run: `python3 -m unittest discover -s tests -v`
Expected: PASS with zero regressions in the existing conditioning, loader/prompt, latent, universal, and upscale suites.
- [ ] **Step 3: Inspect the final worktree state**
```bash
git status --short
```
Expected:
- no output if you committed each task as instructed
- `?? docs/superpowers/plans/2026-05-20-qwen-advanced-params.md` is acceptable if you choose not to commit the plan file itself
- [ ] **Step 4: If Step 2 exposed any regression fixes after the earlier commits, commit only those final verification fixes**
```bash
git add qwen/discovery.py qwen/runtime.py qwen/nodes.py tests/test_qwen_nodes.py
git commit -m "test: verify qwen advanced params"
```
@@ -38,6 +38,8 @@ Extend the existing Qwen nodes so they expose official/common advanced parameter
- Add advanced prompt/sampling controls to `LLSQwenTextToImage`
- Add advanced prompt/sampling/edit controls to `LLSQwenImageEdit`
- Add optional `image2` and `image3` to the image-edit node
- Add optional ordered user LoRA stack input to both Qwen nodes
- Add one compact helper node for building an ordered Qwen LoRA stack
- Add optional turbo/lightning LoRA mode to both nodes
- Auto-match compatible turbo LoRAs when requested
- Reject incompatible turbo LoRA selections explicitly
@@ -62,6 +64,7 @@ This keeps the current user experience intact:
- users still drag one node for text-to-image
- users still drag one node for image edit
- users can optionally connect one compact ordered LoRA stack input
- advanced users gain the controls they expect
- the plugin still avoids forcing Qwen through the generic SD/SDXL/FLUX abstractions
@@ -86,6 +89,7 @@ This keeps the current user experience intact:
- `sampler_name`
- `scheduler`
- `shift`
- `lora_stack` (optional input)
- `enable_turbo_mode`
- `turbo_lora_name`
- `turbo_strength`
@@ -103,6 +107,7 @@ This keeps the current user experience intact:
- Passes `cfg`, `sampler_name`, and `scheduler` directly into internal sampling
- Passes `shift` directly into `ModelSamplingAuraFlow`
- Creates the latent internally from `width`, `height`, and `batch_size`
- Applies any connected ordered user LoRA stack to the internal model before turbo handling
- Optionally loads a compatible turbo/lightning LoRA when turbo mode is enabled
- Returns only the final decoded `IMAGE`
@@ -127,6 +132,7 @@ This keeps the current user experience intact:
- `shift`
- `cfg_norm_strength`
- `reference_latents_method`
- `lora_stack` (optional input)
- `enable_turbo_mode`
- `turbo_lora_name`
- `turbo_strength`
@@ -148,9 +154,32 @@ This keeps the current user experience intact:
- Passes `shift` into `ModelSamplingAuraFlow`
- Passes `cfg_norm_strength` into `CFGNorm`
- Passes `reference_latents_method` into `FluxKontextMultiReferenceLatentMethod`
- Applies any connected ordered user LoRA stack to the internal model before turbo handling
- Optionally loads a compatible turbo/lightning LoRA when turbo mode is enabled
- Returns only the final decoded `IMAGE`
### `LLSQwenLoRAStack`
**Base Inputs**
- `lora_name`
- `strength_model`
**Optional Input**
- `lora_stack`
**Output**
- `LLS_QWEN_LORA_STACK`
**Behavior**
- Builds an ordered model-only LoRA stack for the Qwen nodes
- Supports serial chaining by accepting an existing `lora_stack` and appending one more LoRA entry
- Preserves user-specified order exactly
- Keeps the main Qwen nodes compressed by moving multi-LoRA assembly into one small helper node
## Advanced Parameter Defaults
### Shared Defaults
@@ -209,11 +238,29 @@ When `enable_turbo_mode = true`:
- find a compatible turbo/lightning LoRA for the selected main model
- if `turbo_lora_name = "(auto)"`, auto-select the matching LoRA
- if `turbo_lora_name` is manually set, validate it against the selected main model
- keep user LoRA stack application order separate from turbo handling
- apply that LoRA to the internal model only
- use `turbo_strength` as the LoRA strength
- override runtime `steps` and `cfg` with turbo presets
- keep `sampler_name`, `scheduler`, and `shift` user-controlled
### Ordered LoRA Rules
When a `lora_stack` input is connected:
- treat it as an ordered serial chain
- apply entries in the exact order they appear in the stack
- use `LoraLoaderModelOnly` semantics because these Qwen nodes keep `clip` hidden
- apply the full user stack before optional turbo/lightning LoRA application
This yields the internal order:
- base Qwen model
- user LoRA #1
- user LoRA #2
- ...
- optional turbo/lightning LoRA
### Compatibility Strategy
#### `LLSQwenTextToImage`
@@ -226,7 +273,7 @@ When `enable_turbo_mode = true`:
- Accept only Qwen image-edit turbo/lightning LoRAs
- `qwen_image_edit_2509_*` models prefer `Qwen-Image-Edit-2509-Lightning-*`
- `qwen_image_edit_2511_*` currently has no assumed compatible turbo LoRA
- `qwen_image_edit_2511_*` models may use `Qwen-Image-Edit-2511-Lightning-*` when that LoRA is present locally
### Turbo Presets
@@ -235,6 +282,7 @@ When turbo mode is enabled, the runtime should replace user-entered `steps` and
- `qwen_image_fp8_e4m3fn.safetensors` -> `steps = 8`, `cfg = 1.0`
- `qwen_image_2512_fp8_e4m3fn.safetensors` -> `steps = 4`, `cfg = 1.0`
- `qwen_image_edit_2509_fp8_e4m3fn.safetensors` -> `steps = 4`, `cfg = 1.0`
- `qwen_image_edit_2511_*` -> `steps = 4`, `cfg = 1.0`
If the selected model has no supported turbo profile, turbo mode must fail with a clear error.
@@ -310,6 +358,7 @@ Required error cases:
- required Qwen text encoder is missing
- required Qwen VAE is missing
- primary edit image is missing
- the connected user LoRA stack is malformed
- required ComfyUI runtime component is unavailable
- turbo mode is enabled but no compatible turbo LoRA exists
- turbo mode is enabled but the selected model has no supported turbo preset
@@ -331,6 +380,7 @@ Extend the existing `qwen/` package instead of creating a second feature family.
Add helpers for:
- discovering compatible turbo/lightning LoRAs
- listing user-selectable LoRAs for Qwen stack building
- resolving `(auto)` turbo LoRA matches
- validating manual turbo LoRA selections
- exposing placeholder choices when no compatible turbo LoRAs exist
@@ -340,6 +390,7 @@ Add helpers for:
Extend runtime wrappers so they accept advanced parameters and:
- pass prompt/sampler controls through to internal Qwen execution
- apply ordered model-only user LoRA stacks
- load optional turbo LoRAs
- apply turbo presets when enabled
- support multi-image edit conditioning
@@ -353,12 +404,16 @@ Expand both node schemas to expose the agreed advanced inputs while preserving:
- internal resource encapsulation
- separate text-to-image and image-edit nodes
Also add one compact helper node that outputs `LLS_QWEN_LORA_STACK`.
## Testing
Add or update tests for:
- text-to-image schema includes the new advanced inputs
- image-edit schema includes the new advanced inputs
- LoRA stack helper schema exposes serial chaining
- ordered user LoRA stack entries are applied in order
- negative prompt is routed into the negative conditioning path
- `cfg`, `sampler_name`, `scheduler`, and `shift` are passed through correctly
- `image2` and `image3` are routed into edit conditioning
@@ -388,6 +443,7 @@ This iteration is successful when:
- `LLSQwenTextToImage` exposes official/common advanced prompt and sampling controls
- `LLSQwenImageEdit` exposes official/common advanced prompt, sampling, and edit controls
- both nodes accept an optional ordered model-side LoRA stack input
- both nodes still hide model/clip/vae wiring
- both nodes still output only `IMAGE`
- text-to-image turbo mode works for supported Qwen text models
+142 -8
View File
@@ -11,6 +11,11 @@ except Exception:
TEXT_MODEL_PLACEHOLDER = "(no qwen text-to-image models found)"
EDIT_MODEL_PLACEHOLDER = "(no qwen image edit models found)"
NO_LORA_PLACEHOLDER = "(no loras found)"
AUTO_TURBO_LORA_CHOICE = "(auto)"
NO_TEXT_TURBO_LORA_PLACEHOLDER = "(no qwen text turbo loras found)"
NO_EDIT_TURBO_LORA_PLACEHOLDER = "(no qwen edit turbo loras found)"
REFERENCE_LATENTS_METHOD_CHOICES = ["offset", "index", "uxo/uno", "index_timestep_zero"]
_QWEN_TEXT_ENCODER_PATTERNS = (
"qwen_2.5_vl_7b_fp8_scaled.safetensors",
@@ -46,10 +51,12 @@ def _match_by_patterns(names: list[str], patterns: tuple[str, ...]) -> str | Non
return None
def _lower(value: str | None) -> str:
return (value or "").lower()
def is_qwen_text_to_image_model(name: str | None) -> bool:
if not name:
return False
lowered = name.lower()
lowered = _lower(name)
return (
"qwen" in lowered
and "image" in lowered
@@ -59,9 +66,7 @@ def is_qwen_text_to_image_model(name: str | None) -> bool:
def is_qwen_image_edit_model(name: str | None) -> bool:
if not name:
return False
lowered = name.lower()
lowered = _lower(name)
return "qwen" in lowered and "image" in lowered and "edit" in lowered and "layered" not in lowered
@@ -91,13 +96,142 @@ def validate_qwen_edit_model_name(model_name: str) -> str:
return model_name
def get_qwen_lora_choices() -> list[str]:
names = _sorted_unique(_get_filename_list("loras"))
return names or [NO_LORA_PLACEHOLDER]
def validate_qwen_lora_name(lora_name: str) -> str:
if not lora_name or lora_name == NO_LORA_PLACEHOLDER:
raise RuntimeError("[LLS] Missing Qwen LoRA selection.")
available = _get_filename_list("loras")
if lora_name not in available:
raise RuntimeError(
f"[LLS] LoRA '{lora_name}' was not found in ComfyUI/models/loras/."
)
return lora_name
def is_qwen_text_turbo_lora(name: str | None) -> bool:
lowered = _lower(name)
return "qwen-image" in lowered and "lightning" in lowered and "edit" not in lowered
def is_qwen_edit_turbo_lora(name: str | None) -> bool:
lowered = _lower(name)
return "qwen-image-edit" in lowered and "lightning" in lowered
def get_qwen_text_turbo_lora_choices() -> list[str]:
names = _sorted_unique([name for name in _get_filename_list("loras") if is_qwen_text_turbo_lora(name)])
if not names:
return [AUTO_TURBO_LORA_CHOICE, NO_TEXT_TURBO_LORA_PLACEHOLDER]
return [AUTO_TURBO_LORA_CHOICE] + names
def get_qwen_edit_turbo_lora_choices() -> list[str]:
names = _sorted_unique([name for name in _get_filename_list("loras") if is_qwen_edit_turbo_lora(name)])
if not names:
return [AUTO_TURBO_LORA_CHOICE, NO_EDIT_TURBO_LORA_PLACEHOLDER]
return [AUTO_TURBO_LORA_CHOICE] + names
def _resolve_candidate_turbo_loras(model_name: str, available: list[str]) -> list[str]:
lowered = _lower(model_name)
if is_qwen_text_to_image_model(model_name):
if "2512" in lowered:
return _sorted_unique([name for name in available if "2512" in _lower(name)])
return _sorted_unique(
[
name for name in available
if "2512" not in _lower(name)
and "2509" not in _lower(name)
and "2511" not in _lower(name)
]
)
if is_qwen_image_edit_model(model_name):
if "2511" in lowered:
return _sorted_unique([name for name in available if "2511" in _lower(name)])
if "2509" in lowered:
return _sorted_unique([name for name in available if "2509" in _lower(name)])
return []
def _resolve_turbo_lora(
*,
model_name: str,
requested_name: str,
all_loras: list[str],
available: list[str],
model_label: str,
) -> str:
candidates = _resolve_candidate_turbo_loras(model_name, available)
if requested_name == AUTO_TURBO_LORA_CHOICE:
if candidates:
return candidates[0]
raise RuntimeError(f"[LLS] No compatible turbo/lightning LoRA exists for '{model_name}'.")
if requested_name not in all_loras:
raise RuntimeError(
f"[LLS] Turbo LoRA '{requested_name}' was not found in ComfyUI/models/loras/."
)
if requested_name not in available or requested_name not in candidates:
raise RuntimeError(
f"[LLS] Turbo LoRA '{requested_name}' is not compatible with {model_label} model '{model_name}'."
)
return requested_name
def resolve_qwen_text_turbo_lora(model_name: str, requested_name: str) -> str:
all_loras = _sorted_unique(_get_filename_list("loras"))
available = [name for name in all_loras if is_qwen_text_turbo_lora(name)]
return _resolve_turbo_lora(
model_name=model_name,
requested_name=requested_name,
all_loras=all_loras,
available=available,
model_label="text-to-image",
)
def resolve_qwen_edit_turbo_lora(model_name: str, requested_name: str) -> str:
all_loras = _sorted_unique(_get_filename_list("loras"))
available = [name for name in all_loras if is_qwen_edit_turbo_lora(name)]
return _resolve_turbo_lora(
model_name=model_name,
requested_name=requested_name,
all_loras=all_loras,
available=available,
model_label="image-edit",
)
def get_qwen_turbo_profile(model_name: str) -> dict[str, float] | None:
lowered = _lower(model_name)
if is_qwen_image_edit_model(model_name):
if "2509" in lowered or "2511" in lowered:
return {"steps": 4, "cfg": 1.0}
return None
if is_qwen_text_to_image_model(model_name):
if "2512" in lowered:
return {"steps": 4, "cfg": 1.0}
return {"steps": 8, "cfg": 1.0}
return None
def resolve_qwen_text_encoder_name() -> str | None:
names = _get_filename_list("text_encoders")
matched = _match_by_patterns(names, _QWEN_TEXT_ENCODER_PATTERNS)
if matched is not None:
return matched
for name in names:
lowered = name.lower()
lowered = _lower(name)
if "qwen" in lowered and ("vl" in lowered or "qwen25_7b" in lowered):
return name
return None
@@ -109,7 +243,7 @@ def resolve_qwen_vae_name() -> str | None:
if matched is not None:
return matched
for name in names:
lowered = name.lower()
lowered = _lower(name)
if "qwen" in lowered and "vae" in lowered and "layered" not in lowered:
return name
return None
+133 -7
View File
@@ -1,10 +1,11 @@
"""
Minimal high-level Qwen nodes.
High-level Qwen nodes.
"""
from __future__ import annotations
import importlib
from ..sampling.nodes import _get_samplers, _get_schedulers
from . import discovery, runtime
try:
@@ -13,6 +14,48 @@ except Exception:
comfy_core_nodes = None
_MAX_RESOLUTION = int(getattr(comfy_core_nodes, "MAX_RESOLUTION", 8192))
_QWEN_LORA_STACK_TYPE = "LLS_QWEN_LORA_STACK"
class LLSQwenLoRAStack:
CATEGORY = "LLS/Qwen"
FUNCTION = "build"
RETURN_TYPES = (_QWEN_LORA_STACK_TYPE,)
RETURN_NAMES = ("lora_stack",)
DESCRIPTION = (
"Builds an ordered model-side LoRA stack for the Qwen nodes. "
"Chain multiple nodes to apply multiple LoRAs serially."
)
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"lora_name": (discovery.get_qwen_lora_choices(),),
"strength_model": ("FLOAT", {"default": 1.0, "min": -100.0, "max": 100.0, "step": 0.01}),
},
"optional": {
"lora_stack": (_QWEN_LORA_STACK_TYPE,),
},
}
def build(self, lora_name: str, strength_model: float, lora_stack=None):
if not lora_name or lora_name == discovery.NO_LORA_PLACEHOLDER:
raise RuntimeError("[LLS] Missing Qwen LoRA selection.")
stack = []
if lora_stack is not None:
if not isinstance(lora_stack, list):
raise RuntimeError("[LLS] Qwen lora_stack must be a list.")
stack.extend(lora_stack)
stack.append(
{
"lora_name": str(lora_name),
"strength_model": float(strength_model),
}
)
return (stack,)
class LLSQwenTextToImage:
@@ -21,8 +64,9 @@ class LLSQwenTextToImage:
RETURN_TYPES = ("IMAGE",)
RETURN_NAMES = ("image",)
DESCRIPTION = (
"Minimal Qwen text-to-image node. Internally loads the Qwen model, text encoder, "
"and VAE, then runs the official ComfyUI Qwen generation flow."
"Compressed Qwen text-to-image node. Internally loads the Qwen model, text encoder, "
"and VAE, supports advanced sampling controls, optional ordered LoRA stacks, "
"and optional turbo/lightning LoRAs."
)
@classmethod
@@ -36,7 +80,21 @@ class LLSQwenTextToImage:
"steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
"seed": ("INT", {"default": 0, "min": 0, "max": 0xFFFFFFFFFFFFFFFF}),
"batch_size": ("INT", {"default": 1, "min": 1, "max": 64}),
}
"negative_prompt": ("STRING", {"default": "", "multiline": True, "advanced": True}),
"cfg": ("FLOAT", {"default": 4.0, "min": 0.0, "max": 100.0, "step": 0.1, "advanced": True}),
"sampler_name": (_get_samplers(), {"default": "euler", "advanced": True}),
"scheduler": (_get_schedulers(), {"default": "simple", "advanced": True}),
"shift": ("FLOAT", {"default": 3.1, "min": 0.0, "max": 100.0, "step": 0.01, "advanced": True}),
"enable_turbo_mode": ("BOOLEAN", {"default": False, "advanced": True}),
"turbo_lora_name": (
discovery.get_qwen_text_turbo_lora_choices(),
{"default": discovery.AUTO_TURBO_LORA_CHOICE, "advanced": True},
),
"turbo_strength": ("FLOAT", {"default": 1.0, "min": -100.0, "max": 100.0, "step": 0.01, "advanced": True}),
},
"optional": {
"lora_stack": (_QWEN_LORA_STACK_TYPE,),
},
}
def generate(
@@ -48,16 +106,34 @@ class LLSQwenTextToImage:
steps: int,
seed: int,
batch_size: int,
negative_prompt: str,
cfg: float,
sampler_name: str,
scheduler: str,
shift: float,
enable_turbo_mode: bool,
turbo_lora_name: str,
turbo_strength: float,
lora_stack=None,
):
return (
runtime.run_qwen_text_to_image(
model_name=model_name,
prompt=prompt,
negative_prompt=negative_prompt,
width=width,
height=height,
steps=steps,
seed=seed,
batch_size=batch_size,
cfg=cfg,
sampler_name=sampler_name,
scheduler=scheduler,
shift=shift,
enable_turbo_mode=enable_turbo_mode,
turbo_lora_name=turbo_lora_name,
turbo_strength=turbo_strength,
lora_stack=lora_stack,
),
)
@@ -68,8 +144,9 @@ class LLSQwenImageEdit:
RETURN_TYPES = ("IMAGE",)
RETURN_NAMES = ("image",)
DESCRIPTION = (
"Minimal Qwen image-edit node. Internally loads the Qwen edit model, text encoder, "
"and VAE, then runs the official ComfyUI Qwen image-edit flow."
"Compressed Qwen image-edit node. Internally loads the Qwen edit model, text encoder, "
"and VAE, supports official multi-image edit conditioning, ordered LoRA stacks, "
"and optional turbo/lightning LoRAs."
)
@classmethod
@@ -81,7 +158,28 @@ class LLSQwenImageEdit:
"prompt": ("STRING", {"default": "", "multiline": True}),
"steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
"seed": ("INT", {"default": 0, "min": 0, "max": 0xFFFFFFFFFFFFFFFF}),
}
"negative_prompt": ("STRING", {"default": "", "multiline": True, "advanced": True}),
"cfg": ("FLOAT", {"default": 4.0, "min": 0.0, "max": 100.0, "step": 0.1, "advanced": True}),
"sampler_name": (_get_samplers(), {"default": "euler", "advanced": True}),
"scheduler": (_get_schedulers(), {"default": "simple", "advanced": True}),
"shift": ("FLOAT", {"default": 3.1, "min": 0.0, "max": 100.0, "step": 0.01, "advanced": True}),
"cfg_norm_strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 100.0, "step": 0.01, "advanced": True}),
"reference_latents_method": (
discovery.REFERENCE_LATENTS_METHOD_CHOICES,
{"default": "index_timestep_zero", "advanced": True},
),
"enable_turbo_mode": ("BOOLEAN", {"default": False, "advanced": True}),
"turbo_lora_name": (
discovery.get_qwen_edit_turbo_lora_choices(),
{"default": discovery.AUTO_TURBO_LORA_CHOICE, "advanced": True},
),
"turbo_strength": ("FLOAT", {"default": 1.0, "min": -100.0, "max": 100.0, "step": 0.01, "advanced": True}),
},
"optional": {
"image2": ("IMAGE",),
"image3": ("IMAGE",),
"lora_stack": (_QWEN_LORA_STACK_TYPE,),
},
}
def generate(
@@ -91,24 +189,52 @@ class LLSQwenImageEdit:
prompt: str,
steps: int,
seed: int,
image2=None,
image3=None,
negative_prompt: str = "",
cfg: float = 4.0,
sampler_name: str = "euler",
scheduler: str = "simple",
shift: float = 3.1,
cfg_norm_strength: float = 1.0,
reference_latents_method: str = "index_timestep_zero",
enable_turbo_mode: bool = False,
turbo_lora_name: str = discovery.AUTO_TURBO_LORA_CHOICE,
turbo_strength: float = 1.0,
lora_stack=None,
):
return (
runtime.run_qwen_image_edit(
model_name=model_name,
image=image,
image2=image2,
image3=image3,
prompt=prompt,
negative_prompt=negative_prompt,
steps=steps,
seed=seed,
cfg=cfg,
sampler_name=sampler_name,
scheduler=scheduler,
shift=shift,
cfg_norm_strength=cfg_norm_strength,
reference_latents_method=reference_latents_method,
enable_turbo_mode=enable_turbo_mode,
turbo_lora_name=turbo_lora_name,
turbo_strength=turbo_strength,
lora_stack=lora_stack,
),
)
NODE_CLASS_MAPPINGS: dict[str, type] = {
"LLSQwenLoRAStack": LLSQwenLoRAStack,
"LLSQwenTextToImage": LLSQwenTextToImage,
"LLSQwenImageEdit": LLSQwenImageEdit,
}
NODE_DISPLAY_NAME_MAPPINGS: dict[str, str] = {
"LLSQwenLoRAStack": "LLS Qwen LoRA Stack",
"LLSQwenTextToImage": "LLS Qwen Text To Image",
"LLSQwenImageEdit": "LLS Qwen Image Edit",
}
+137 -28
View File
@@ -63,12 +63,6 @@ from . import discovery
_QWEN_CLIP_TYPE = "qwen_image"
_QWEN_MODEL_SHIFT = 3.1
_QWEN_CFG = 4.0
_QWEN_SAMPLER = "euler"
_QWEN_SCHEDULER = "simple"
_QWEN_CFG_NORM_STRENGTH = 1.0
_QWEN_REFERENCE_LATENTS_METHOD = "index_timestep_zero"
def _unwrap_first(result: Any):
@@ -123,14 +117,54 @@ def _load_qwen_model(model_name: str):
return unet_loader_cls().load_unet(model_name, "default")[0]
def _patch_qwen_model_sampling(model):
def _load_model_only_lora(model, lora_name: str, strength_model: float):
lora_loader_cls = _require_class(comfy_core_nodes, "LoraLoaderModelOnly", _CORE_NODES_ERR)
return lora_loader_cls().load_lora_model_only(model, lora_name, float(strength_model))[0]
def _apply_lora_stack(model, lora_stack):
if lora_stack is None:
return model
if not isinstance(lora_stack, list):
raise RuntimeError("[LLS] Qwen lora_stack must be a list of LoRA entries.")
current_model = model
for index, entry in enumerate(lora_stack, start=1):
if not isinstance(entry, dict):
raise RuntimeError(f"[LLS] Invalid Qwen LoRA stack entry at position {index}.")
lora_name = discovery.validate_qwen_lora_name(str(entry.get("lora_name", "")))
try:
strength_model = float(entry.get("strength_model", 1.0))
except (TypeError, ValueError) as exc:
raise RuntimeError(
f"[LLS] Invalid strength_model in Qwen LoRA stack entry at position {index}."
) from exc
current_model = _load_model_only_lora(current_model, lora_name, strength_model)
return current_model
def _prepare_turbo_request(resolver, model_name: str, steps: int, cfg: float, enable_turbo_mode: bool, turbo_lora_name: str):
if not enable_turbo_mode:
return None, int(steps), float(cfg)
resolved_lora = resolver(model_name, turbo_lora_name)
profile = discovery.get_qwen_turbo_profile(model_name)
if profile is None:
raise RuntimeError(f"[LLS] No supported turbo preset exists for '{model_name}'.")
return resolved_lora, int(profile["steps"]), float(profile["cfg"])
def _patch_qwen_model_sampling(model, shift: float):
patcher_cls = _require_class(nodes_model_advanced, "ModelSamplingAuraFlow", _MODEL_ADVANCED_ERR)
return patcher_cls().patch_aura(model, _QWEN_MODEL_SHIFT)[0]
return patcher_cls().patch_aura(model, float(shift))[0]
def _apply_cfg_norm(model):
def _apply_cfg_norm(model, strength: float):
cfg_cls = _require_class(nodes_cfg, "CFGNorm", _CFG_ERR)
return _unwrap_first(cfg_cls.execute(model, _QWEN_CFG_NORM_STRENGTH))
return _unwrap_first(cfg_cls.execute(model, float(strength)))
def _encode_clip_text(clip, prompt: str):
@@ -146,9 +180,18 @@ def _create_empty_qwen_latent(width: int, height: int, batch_size: int):
return latent_node.execute(width, height, batch_size)[0]
def _encode_qwen_edit_conditioning(clip, prompt: str, vae, image):
def _encode_qwen_edit_conditioning(clip, prompt: str, vae, image1, image2=None, image3=None):
encoder_cls = _require_class(nodes_qwen, "TextEncodeQwenImageEditPlus", _QWEN_ERR)
return _unwrap_first(encoder_cls.execute(clip, prompt, vae=vae, image1=image))
return _unwrap_first(
encoder_cls.execute(
clip,
prompt,
vae=vae,
image1=image1,
image2=image2,
image3=image3,
)
)
def _scale_qwen_edit_image(image):
@@ -156,9 +199,9 @@ def _scale_qwen_edit_image(image):
return _unwrap_first(scaler_cls.execute(image))
def _apply_reference_latents_method(conditioning):
def _apply_reference_latents_method(conditioning, reference_latents_method: str):
method_cls = _require_class(nodes_flux, "FluxKontextMultiReferenceLatentMethod", _FLUX_ERR)
return _unwrap_first(method_cls.execute(conditioning, _QWEN_REFERENCE_LATENTS_METHOD))
return _unwrap_first(method_cls.execute(conditioning, reference_latents_method))
def _encode_image_to_latent(vae, image):
@@ -166,15 +209,15 @@ def _encode_image_to_latent(vae, image):
return encoder_cls().encode(vae, image)[0]
def _sample_qwen(model, positive, negative, latent, steps: int, seed: int):
def _sample_qwen(model, positive, negative, latent, steps: int, seed: int, cfg: float, sampler_name: str, scheduler: str):
sampler_cls = _require_class(comfy_core_nodes, "KSampler", _CORE_NODES_ERR)
return sampler_cls().sample(
model,
int(seed),
int(steps),
_QWEN_CFG,
_QWEN_SAMPLER,
_QWEN_SCHEDULER,
float(cfg),
sampler_name,
scheduler,
positive,
negative,
latent,
@@ -208,24 +251,55 @@ def _resolve_qwen_resources(model_name: str, validator):
def run_qwen_text_to_image(
model_name: str,
prompt: str,
negative_prompt: str,
width: int,
height: int,
steps: int,
seed: int,
batch_size: int,
cfg: float,
sampler_name: str,
scheduler: str,
shift: float,
enable_turbo_mode: bool,
turbo_lora_name: str,
turbo_strength: float,
lora_stack=None,
):
try:
model_name, clip_name, vae_name = _resolve_qwen_resources(
model_name,
discovery.validate_qwen_text_model_name,
)
model = _patch_qwen_model_sampling(_load_qwen_model(model_name))
resolved_turbo_lora, effective_steps, effective_cfg = _prepare_turbo_request(
discovery.resolve_qwen_text_turbo_lora,
model_name,
steps,
cfg,
enable_turbo_mode,
turbo_lora_name,
)
model = _load_qwen_model(model_name)
model = _apply_lora_stack(model, lora_stack)
if resolved_turbo_lora is not None:
model = _load_model_only_lora(model, resolved_turbo_lora, float(turbo_strength))
model = _patch_qwen_model_sampling(model, shift)
clip = _load_qwen_clip(clip_name)
vae = _load_qwen_vae(vae_name)
positive = _encode_clip_text(clip, prompt)
negative = _encode_clip_text(clip, "")
negative = _encode_clip_text(clip, negative_prompt)
latent = _create_empty_qwen_latent(int(width), int(height), int(batch_size))
sampled = _sample_qwen(model, positive, negative, latent, int(steps), int(seed))
sampled = _sample_qwen(
model,
positive,
negative,
latent,
effective_steps,
int(seed),
effective_cfg,
sampler_name,
scheduler,
)
return _decode_qwen_latent(vae, sampled)
except Exception as exc:
if str(exc).startswith("[LLS]"):
@@ -236,9 +310,22 @@ def run_qwen_text_to_image(
def run_qwen_image_edit(
model_name: str,
image,
image2,
image3,
prompt: str,
negative_prompt: str,
steps: int,
seed: int,
cfg: float,
sampler_name: str,
scheduler: str,
shift: float,
cfg_norm_strength: float,
reference_latents_method: str,
enable_turbo_mode: bool,
turbo_lora_name: str,
turbo_strength: float,
lora_stack=None,
):
try:
if image is None:
@@ -248,17 +335,39 @@ def run_qwen_image_edit(
model_name,
discovery.validate_qwen_edit_model_name,
)
model = _patch_qwen_model_sampling(_load_qwen_model(model_name))
resolved_turbo_lora, effective_steps, effective_cfg = _prepare_turbo_request(
discovery.resolve_qwen_edit_turbo_lora,
model_name,
steps,
cfg,
enable_turbo_mode,
turbo_lora_name,
)
model = _load_qwen_model(model_name)
model = _apply_lora_stack(model, lora_stack)
if resolved_turbo_lora is not None:
model = _load_model_only_lora(model, resolved_turbo_lora, float(turbo_strength))
model = _patch_qwen_model_sampling(model, shift)
model = _apply_cfg_norm(model, cfg_norm_strength)
clip = _load_qwen_clip(clip_name)
vae = _load_qwen_vae(vae_name)
scaled = _scale_qwen_edit_image(image)
positive = _encode_qwen_edit_conditioning(clip, prompt, vae, scaled)
negative = _encode_qwen_edit_conditioning(clip, "", vae, scaled)
positive = _apply_reference_latents_method(positive)
negative = _apply_reference_latents_method(negative)
positive = _encode_qwen_edit_conditioning(clip, prompt, vae, scaled, image2, image3)
negative = _encode_qwen_edit_conditioning(clip, negative_prompt, vae, scaled, image2, image3)
positive = _apply_reference_latents_method(positive, reference_latents_method)
negative = _apply_reference_latents_method(negative, reference_latents_method)
latent = _encode_image_to_latent(vae, scaled)
model = _apply_cfg_norm(model)
sampled = _sample_qwen(model, positive, negative, latent, int(steps), int(seed))
sampled = _sample_qwen(
model,
positive,
negative,
latent,
effective_steps,
int(seed),
effective_cfg,
sampler_name,
scheduler,
)
return _decode_qwen_latent(vae, sampled)
except Exception as exc:
if str(exc).startswith("[LLS]"):
+415 -54
View File
@@ -6,6 +6,7 @@ from unittest import mock
ROOT = pathlib.Path(__file__).resolve().parents[1]
QWEN_LORA_STACK_TYPE = "LLS_QWEN_LORA_STACK"
def load_plugin_package():
@@ -24,6 +25,10 @@ def load_plugin_package():
return module
def make_lora_stack(*entries):
return [{"lora_name": name, "strength_model": strength} for name, strength in entries]
class NodeOutputStub:
def __init__(self, *args):
self.args = args
@@ -42,12 +47,22 @@ class QwenFolderPathsStub:
"diffusion_models": [
"qwen_image_fp8_e4m3fn.safetensors",
"qwen_image_2512_fp8_e4m3fn.safetensors",
"qwen_image_edit_2509_fp8_e4m3fn.safetensors",
"qwen_image_edit_2511_bf16.safetensors",
"qwen_image_layered_bf16.safetensors",
"flux1-dev.safetensors",
],
"text_encoders": ["qwen_2.5_vl_7b_fp8_scaled.safetensors"],
"vae": ["qwen_image_vae.safetensors"],
"loras": [
"qwen-style-a.safetensors",
"qwen-style-b.safetensors",
"Qwen-Image-Lightning-8steps-V1.0.safetensors",
"Qwen-Image-2512-Lightning-4steps-V1.0-fp32.safetensors",
"Qwen-Image-Edit-2509-Lightning-4steps-V1.0-bf16.safetensors",
"Qwen-Image-Edit-2511-Lightning-4steps-V1.0.safetensors",
"flux-dev-style.safetensors",
],
}
def get_filename_list(self, category):
@@ -73,6 +88,7 @@ class CoreQwenNodesStub:
last_vae_encode_call = None
last_vae_decode_call = None
last_vae_load_call = None
last_lora_calls = []
@classmethod
def reset(cls):
@@ -83,6 +99,7 @@ class CoreQwenNodesStub:
cls.last_vae_encode_call = None
cls.last_vae_decode_call = None
cls.last_vae_load_call = None
cls.last_lora_calls = []
class UNETLoader:
def load_unet(self, unet_name, weight_dtype):
@@ -117,6 +134,17 @@ class CoreQwenNodesStub:
)
return (f"COND::{text}",)
class LoraLoaderModelOnly:
def load_lora_model_only(self, model, lora_name, strength_model):
CoreQwenNodesStub.last_lora_calls.append(
{
"model": model,
"lora_name": lora_name,
"strength_model": strength_model,
}
)
return (f"LORA::{model}::{lora_name}::{strength_model}",)
class KSampler:
def sample(
self,
@@ -244,7 +272,7 @@ class QwenExtraNodesStub:
"image3": image3,
}
)
return NodeOutputStub(f"EDIT_COND::{prompt}::{image1}")
return NodeOutputStub(f"EDIT_COND::{prompt}::{image1}::{image2}::{image3}")
class CFGNodesStub:
@@ -275,6 +303,7 @@ class TestQwenNodes(unittest.TestCase):
self.assertIn("LLSQwenTextToImage", plugin.NODE_CLASS_MAPPINGS)
self.assertIn("LLSQwenImageEdit", plugin.NODE_CLASS_MAPPINGS)
self.assertIn("LLSQwenLoRAStack", plugin.NODE_CLASS_MAPPINGS)
def test_qwen_text_node_filters_only_text_models(self):
load_plugin_package()
@@ -300,19 +329,136 @@ class TestQwenNodes(unittest.TestCase):
with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()):
choices = qwen_nodes.LLSQwenImageEdit.INPUT_TYPES()["required"]["model_name"][0]
self.assertEqual(choices, ["qwen_image_edit_2511_bf16.safetensors"])
self.assertEqual(
choices,
[
"qwen_image_edit_2509_fp8_e4m3fn.safetensors",
"qwen_image_edit_2511_bf16.safetensors",
],
)
def test_qwen_text_node_uses_placeholder_when_no_compatible_models_exist(self):
def test_qwen_text_node_schema_exposes_advanced_inputs_and_optional_lora_stack(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSQwenTextToImage"]
required = node_cls.INPUT_TYPES()["required"]
optional = node_cls.INPUT_TYPES()["optional"]
self.assertEqual(node_cls.RETURN_TYPES, ("IMAGE",))
self.assertEqual(node_cls.RETURN_NAMES, ("image",))
self.assertEqual(
tuple(required.keys()),
(
"model_name",
"prompt",
"width",
"height",
"steps",
"seed",
"batch_size",
"negative_prompt",
"cfg",
"sampler_name",
"scheduler",
"shift",
"enable_turbo_mode",
"turbo_lora_name",
"turbo_strength",
),
)
self.assertEqual(required["negative_prompt"][1]["default"], "")
self.assertEqual(required["cfg"][1]["default"], 4.0)
self.assertIn("euler", required["sampler_name"][0])
self.assertIn("simple", required["scheduler"][0])
self.assertEqual(required["shift"][1]["default"], 3.1)
self.assertEqual(required["enable_turbo_mode"][1]["default"], False)
self.assertEqual(optional["lora_stack"][0], QWEN_LORA_STACK_TYPE)
def test_qwen_edit_node_schema_exposes_advanced_inputs_optional_images_and_lora_stack(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSQwenImageEdit"]
required = node_cls.INPUT_TYPES()["required"]
optional = node_cls.INPUT_TYPES()["optional"]
self.assertEqual(node_cls.RETURN_TYPES, ("IMAGE",))
self.assertEqual(node_cls.RETURN_NAMES, ("image",))
self.assertEqual(
tuple(required.keys()),
(
"model_name",
"image",
"prompt",
"steps",
"seed",
"negative_prompt",
"cfg",
"sampler_name",
"scheduler",
"shift",
"cfg_norm_strength",
"reference_latents_method",
"enable_turbo_mode",
"turbo_lora_name",
"turbo_strength",
),
)
self.assertEqual(optional["image2"][0], "IMAGE")
self.assertEqual(optional["image3"][0], "IMAGE")
self.assertEqual(optional["lora_stack"][0], QWEN_LORA_STACK_TYPE)
self.assertEqual(
required["reference_latents_method"][0],
["offset", "index", "uxo/uno", "index_timestep_zero"],
)
def test_qwen_lora_stack_node_builds_ordered_stack(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSQwenLoRAStack"]
node = node_cls()
first = node.build("qwen-style-a.safetensors", 0.5)[0]
second = node.build("qwen-style-b.safetensors", 1.25, first)[0]
self.assertEqual(node_cls.RETURN_TYPES, (QWEN_LORA_STACK_TYPE,))
self.assertEqual(
second,
[
{"lora_name": "qwen-style-a.safetensors", "strength_model": 0.5},
{"lora_name": "qwen-style-b.safetensors", "strength_model": 1.25},
],
)
def test_qwen_text_node_turbo_lora_choices_filter_only_text_loras(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import nodes as qwen_nodes
stub = QwenFolderPathsStub()
stub._files["diffusion_models"] = ["flux1-dev.safetensors", "sdxl.safetensors"]
with mock.patch.object(qwen_discovery, "folder_paths", stub):
choices = qwen_nodes.LLSQwenTextToImage.INPUT_TYPES()["required"]["model_name"][0]
with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()):
choices = qwen_nodes.LLSQwenTextToImage.INPUT_TYPES()["required"]["turbo_lora_name"][0]
self.assertEqual(choices, ["(no qwen text-to-image models found)"])
self.assertEqual(
choices,
[
"(auto)",
"Qwen-Image-2512-Lightning-4steps-V1.0-fp32.safetensors",
"Qwen-Image-Lightning-8steps-V1.0.safetensors",
],
)
def test_qwen_edit_node_turbo_lora_choices_filter_only_edit_loras(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import nodes as qwen_nodes
with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()):
choices = qwen_nodes.LLSQwenImageEdit.INPUT_TYPES()["required"]["turbo_lora_name"][0]
self.assertEqual(
choices,
[
"(auto)",
"Qwen-Image-Edit-2509-Lightning-4steps-V1.0-bf16.safetensors",
"Qwen-Image-Edit-2511-Lightning-4steps-V1.0.safetensors",
],
)
def test_qwen_text_runtime_rejects_incompatible_model_name(self):
load_plugin_package()
@@ -329,11 +475,20 @@ class TestQwenNodes(unittest.TestCase):
qwen_runtime.run_qwen_text_to_image(
model_name="qwen_image_edit_2511_bf16.safetensors",
prompt="a cat",
negative_prompt="",
width=1024,
height=1024,
steps=20,
seed=1,
batch_size=1,
cfg=4.0,
sampler_name="euler",
scheduler="simple",
shift=3.1,
enable_turbo_mode=False,
turbo_lora_name="(auto)",
turbo_strength=1.0,
lora_stack=None,
)
def test_qwen_edit_runtime_rejects_incompatible_model_name(self):
@@ -351,9 +506,22 @@ class TestQwenNodes(unittest.TestCase):
qwen_runtime.run_qwen_image_edit(
model_name="qwen_image_fp8_e4m3fn.safetensors",
image="IMAGE::input",
image2=None,
image3=None,
prompt="edit the cat",
negative_prompt="",
steps=20,
seed=1,
cfg=4.0,
sampler_name="euler",
scheduler="simple",
shift=3.1,
cfg_norm_strength=1.0,
reference_latents_method="index_timestep_zero",
enable_turbo_mode=False,
turbo_lora_name="(auto)",
turbo_strength=1.0,
lora_stack=None,
)
def test_qwen_text_runtime_raises_when_qwen_text_encoder_is_missing(self):
@@ -372,11 +540,20 @@ class TestQwenNodes(unittest.TestCase):
qwen_runtime.run_qwen_text_to_image(
model_name="qwen_image_fp8_e4m3fn.safetensors",
prompt="a cat",
negative_prompt="",
width=1024,
height=1024,
steps=20,
seed=1,
batch_size=1,
cfg=4.0,
sampler_name="euler",
scheduler="simple",
shift=3.1,
enable_turbo_mode=False,
turbo_lora_name="(auto)",
turbo_strength=1.0,
lora_stack=None,
)
def test_qwen_text_runtime_raises_when_qwen_vae_is_missing(self):
@@ -395,19 +572,32 @@ class TestQwenNodes(unittest.TestCase):
qwen_runtime.run_qwen_text_to_image(
model_name="qwen_image_fp8_e4m3fn.safetensors",
prompt="a cat",
negative_prompt="",
width=1024,
height=1024,
steps=20,
seed=1,
batch_size=1,
cfg=4.0,
sampler_name="euler",
scheduler="simple",
shift=3.1,
enable_turbo_mode=False,
turbo_lora_name="(auto)",
turbo_strength=1.0,
lora_stack=None,
)
def test_qwen_text_runtime_executes_minimal_official_pipeline(self):
def test_qwen_text_runtime_passes_advanced_controls_and_ordered_user_loras(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import runtime as qwen_runtime
stub = QwenFolderPathsStub()
lora_stack = make_lora_stack(
("qwen-style-a.safetensors", 0.5),
("qwen-style-b.safetensors", 1.25),
)
with mock.patch.object(qwen_discovery, "folder_paths", stub), mock.patch.object(
qwen_runtime,
"folder_paths",
@@ -420,24 +610,178 @@ class TestQwenNodes(unittest.TestCase):
image = qwen_runtime.run_qwen_text_to_image(
model_name="qwen_image_fp8_e4m3fn.safetensors",
prompt="a cat",
negative_prompt="low quality",
width=1024,
height=1152,
steps=20,
steps=33,
seed=99,
batch_size=2,
cfg=7.5,
sampler_name="heun",
scheduler="normal",
shift=2.25,
enable_turbo_mode=False,
turbo_lora_name="(auto)",
turbo_strength=1.0,
lora_stack=lora_stack,
)
self.assertEqual(image, "IMAGE::SAMPLED::99::20")
self.assertEqual(CoreQwenNodesStub.last_clip_call["type"], "qwen_image")
self.assertEqual(SD3NodesStub.last_generate_call, {"width": 1024, "height": 1152, "batch_size": 2})
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["cfg"], 4.0)
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["sampler_name"], "euler")
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["scheduler"], "simple")
self.assertEqual(image, "IMAGE::SAMPLED::99::33")
self.assertEqual(CoreQwenNodesStub.last_text_encode_calls[0]["text"], "a cat")
self.assertEqual(CoreQwenNodesStub.last_text_encode_calls[1]["text"], "")
self.assertEqual(ModelAdvancedStub.last_call["shift"], 3.1)
self.assertEqual(CoreQwenNodesStub.last_text_encode_calls[1]["text"], "low quality")
self.assertEqual(SD3NodesStub.last_generate_call, {"width": 1024, "height": 1152, "batch_size": 2})
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["steps"], 33)
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["cfg"], 7.5)
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["sampler_name"], "heun")
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["scheduler"], "normal")
self.assertEqual(
[call["lora_name"] for call in CoreQwenNodesStub.last_lora_calls],
["qwen-style-a.safetensors", "qwen-style-b.safetensors"],
)
self.assertEqual(ModelAdvancedStub.last_call["shift"], 2.25)
def test_qwen_edit_runtime_executes_minimal_official_pipeline(self):
def test_qwen_text_runtime_applies_user_loras_before_turbo_lora(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import runtime as qwen_runtime
stub = QwenFolderPathsStub()
lora_stack = make_lora_stack(("qwen-style-a.safetensors", 0.5))
with mock.patch.object(qwen_discovery, "folder_paths", stub), mock.patch.object(
qwen_runtime,
"folder_paths",
stub,
), mock.patch.object(qwen_runtime, "comfy_core_nodes", CoreQwenNodesStub), mock.patch.object(
qwen_runtime,
"nodes_sd3",
SD3NodesStub,
), mock.patch.object(qwen_runtime, "nodes_model_advanced", ModelAdvancedStub):
qwen_runtime.run_qwen_text_to_image(
model_name="qwen_image_fp8_e4m3fn.safetensors",
prompt="a cat",
negative_prompt="",
width=1024,
height=1024,
steps=50,
seed=99,
batch_size=1,
cfg=8.0,
sampler_name="euler",
scheduler="simple",
shift=3.1,
enable_turbo_mode=True,
turbo_lora_name="(auto)",
turbo_strength=0.75,
lora_stack=lora_stack,
)
self.assertEqual(
[call["lora_name"] for call in CoreQwenNodesStub.last_lora_calls],
[
"qwen-style-a.safetensors",
"Qwen-Image-Lightning-8steps-V1.0.safetensors",
],
)
self.assertEqual(CoreQwenNodesStub.last_lora_calls[-1]["strength_model"], 0.75)
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["steps"], 8)
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["cfg"], 1.0)
def test_qwen_text_runtime_rejects_incompatible_manual_turbo_lora(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import runtime as qwen_runtime
stub = QwenFolderPathsStub()
with mock.patch.object(qwen_discovery, "folder_paths", stub), mock.patch.object(
qwen_runtime,
"folder_paths",
stub,
):
with self.assertRaisesRegex(RuntimeError, "Turbo LoRA .* is not compatible"):
qwen_runtime.run_qwen_text_to_image(
model_name="qwen_image_fp8_e4m3fn.safetensors",
prompt="a cat",
negative_prompt="",
width=1024,
height=1024,
steps=20,
seed=1,
batch_size=1,
cfg=4.0,
sampler_name="euler",
scheduler="simple",
shift=3.1,
enable_turbo_mode=True,
turbo_lora_name="Qwen-Image-Edit-2509-Lightning-4steps-V1.0-bf16.safetensors",
turbo_strength=1.0,
lora_stack=None,
)
def test_qwen_edit_runtime_passes_optional_images_controls_and_ordered_user_loras(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import runtime as qwen_runtime
stub = QwenFolderPathsStub()
lora_stack = make_lora_stack(
("qwen-style-a.safetensors", 0.5),
("qwen-style-b.safetensors", 1.25),
)
with mock.patch.object(qwen_discovery, "folder_paths", stub), mock.patch.object(
qwen_runtime,
"folder_paths",
stub,
), mock.patch.object(qwen_runtime, "comfy_core_nodes", CoreQwenNodesStub), mock.patch.object(
qwen_runtime,
"nodes_model_advanced",
ModelAdvancedStub,
), mock.patch.object(qwen_runtime, "nodes_qwen", QwenExtraNodesStub), mock.patch.object(
qwen_runtime,
"nodes_flux",
FluxNodesStub,
), mock.patch.object(qwen_runtime, "nodes_cfg", CFGNodesStub):
image = qwen_runtime.run_qwen_image_edit(
model_name="qwen_image_edit_2509_fp8_e4m3fn.safetensors",
image="IMAGE::input",
image2="IMAGE::ref2",
image3="IMAGE::ref3",
prompt="turn the cat blue",
negative_prompt="low quality",
steps=28,
seed=123,
cfg=6.5,
sampler_name="heun",
scheduler="normal",
shift=2.0,
cfg_norm_strength=0.8,
reference_latents_method="index",
enable_turbo_mode=False,
turbo_lora_name="(auto)",
turbo_strength=1.0,
lora_stack=lora_stack,
)
self.assertEqual(image, "IMAGE::SAMPLED::123::28")
self.assertEqual(FluxNodesStub.last_scale_call, {"image": "IMAGE::input"})
self.assertEqual(QwenExtraNodesStub.last_calls[0]["prompt"], "turn the cat blue")
self.assertEqual(QwenExtraNodesStub.last_calls[0]["image1"], "SCALED::IMAGE::input")
self.assertEqual(QwenExtraNodesStub.last_calls[0]["image2"], "IMAGE::ref2")
self.assertEqual(QwenExtraNodesStub.last_calls[0]["image3"], "IMAGE::ref3")
self.assertEqual(QwenExtraNodesStub.last_calls[1]["prompt"], "low quality")
self.assertEqual(len(FluxNodesStub.last_reference_calls), 2)
self.assertEqual(FluxNodesStub.last_reference_calls[0]["reference_latents_method"], "index")
self.assertEqual(CFGNodesStub.last_call["strength"], 0.8)
self.assertEqual(CoreQwenNodesStub.last_vae_encode_call["pixels"], "SCALED::IMAGE::input")
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["cfg"], 6.5)
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["sampler_name"], "heun")
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["scheduler"], "normal")
self.assertEqual(
[call["lora_name"] for call in CoreQwenNodesStub.last_lora_calls],
["qwen-style-a.safetensors", "qwen-style-b.safetensors"],
)
self.assertEqual(ModelAdvancedStub.last_call["shift"], 2.0)
def test_qwen_edit_runtime_supports_2511_turbo_when_matching_lora_exists(self):
load_plugin_package()
from lls_node_test_qwen.qwen import discovery as qwen_discovery
from lls_node_test_qwen.qwen import runtime as qwen_runtime
@@ -456,52 +800,43 @@ class TestQwenNodes(unittest.TestCase):
"nodes_flux",
FluxNodesStub,
), mock.patch.object(qwen_runtime, "nodes_cfg", CFGNodesStub):
image = qwen_runtime.run_qwen_image_edit(
qwen_runtime.run_qwen_image_edit(
model_name="qwen_image_edit_2511_bf16.safetensors",
image="IMAGE::input",
image2=None,
image3=None,
prompt="turn the cat blue",
steps=30,
negative_prompt="",
steps=20,
seed=123,
cfg=4.0,
sampler_name="euler",
scheduler="simple",
shift=3.1,
cfg_norm_strength=1.0,
reference_latents_method="index_timestep_zero",
enable_turbo_mode=True,
turbo_lora_name="(auto)",
turbo_strength=0.7,
lora_stack=None,
)
self.assertEqual(image, "IMAGE::SAMPLED::123::30")
self.assertEqual(FluxNodesStub.last_scale_call, {"image": "IMAGE::input"})
self.assertEqual(QwenExtraNodesStub.last_calls[0]["prompt"], "turn the cat blue")
self.assertEqual(QwenExtraNodesStub.last_calls[0]["image1"], "SCALED::IMAGE::input")
self.assertEqual(QwenExtraNodesStub.last_calls[1]["prompt"], "")
self.assertEqual(len(FluxNodesStub.last_reference_calls), 2)
self.assertEqual(FluxNodesStub.last_reference_calls[0]["reference_latents_method"], "index_timestep_zero")
self.assertEqual(CFGNodesStub.last_call["strength"], 1.0)
self.assertEqual(CoreQwenNodesStub.last_vae_encode_call["pixels"], "SCALED::IMAGE::input")
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["cfg"], 4.0)
def test_qwen_text_node_returns_image_only(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSQwenTextToImage"]
required = node_cls.INPUT_TYPES()["required"]
self.assertEqual(node_cls.RETURN_TYPES, ("IMAGE",))
self.assertEqual(node_cls.RETURN_NAMES, ("image",))
self.assertEqual(
tuple(required.keys()),
("model_name", "prompt", "width", "height", "steps", "seed", "batch_size"),
CoreQwenNodesStub.last_lora_calls[-1]["lora_name"],
"Qwen-Image-Edit-2511-Lightning-4steps-V1.0.safetensors",
)
def test_qwen_edit_node_returns_image_only(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSQwenImageEdit"]
required = node_cls.INPUT_TYPES()["required"]
self.assertEqual(node_cls.RETURN_TYPES, ("IMAGE",))
self.assertEqual(node_cls.RETURN_NAMES, ("image",))
self.assertEqual(tuple(required.keys()), ("model_name", "image", "prompt", "steps", "seed"))
self.assertEqual(CoreQwenNodesStub.last_lora_calls[-1]["strength_model"], 0.7)
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["steps"], 4)
self.assertEqual(CoreQwenNodesStub.last_ksampler_call["cfg"], 1.0)
def test_qwen_text_node_executes_runtime(self):
load_plugin_package()
from lls_node_test_qwen.qwen import nodes as qwen_nodes
node = qwen_nodes.LLSQwenTextToImage()
with mock.patch.object(qwen_nodes.runtime, "run_qwen_text_to_image", return_value="IMAGE::node"):
node = qwen_nodes.LLSQwenTextToImage()
lora_stack = make_lora_stack(("qwen-style-a.safetensors", 0.5))
with mock.patch.object(qwen_nodes.runtime, "run_qwen_text_to_image", return_value="IMAGE::node") as runtime_mock:
result = node.generate(
model_name="qwen_image_fp8_e4m3fn.safetensors",
prompt="a cat",
@@ -510,25 +845,51 @@ class TestQwenNodes(unittest.TestCase):
steps=20,
seed=1,
batch_size=1,
negative_prompt="",
cfg=4.0,
sampler_name="euler",
scheduler="simple",
shift=3.1,
enable_turbo_mode=False,
turbo_lora_name="(auto)",
turbo_strength=1.0,
lora_stack=lora_stack,
)
self.assertEqual(result, ("IMAGE::node",))
self.assertEqual(runtime_mock.call_args.kwargs["lora_stack"], lora_stack)
def test_qwen_edit_node_executes_runtime(self):
load_plugin_package()
from lls_node_test_qwen.qwen import nodes as qwen_nodes
node = qwen_nodes.LLSQwenImageEdit()
with mock.patch.object(qwen_nodes.runtime, "run_qwen_image_edit", return_value="IMAGE::node-edit"):
node = qwen_nodes.LLSQwenImageEdit()
lora_stack = make_lora_stack(("qwen-style-a.safetensors", 0.5))
with mock.patch.object(qwen_nodes.runtime, "run_qwen_image_edit", return_value="IMAGE::node-edit") as runtime_mock:
result = node.generate(
model_name="qwen_image_edit_2511_bf16.safetensors",
image="IMAGE::input",
prompt="edit the cat",
steps=20,
seed=1,
image2=None,
image3=None,
negative_prompt="",
cfg=4.0,
sampler_name="euler",
scheduler="simple",
shift=3.1,
cfg_norm_strength=1.0,
reference_latents_method="index_timestep_zero",
enable_turbo_mode=False,
turbo_lora_name="(auto)",
turbo_strength=1.0,
lora_stack=lora_stack,
)
self.assertEqual(result, ("IMAGE::node-edit",))
self.assertEqual(runtime_mock.call_args.kwargs["lora_stack"], lora_stack)
if __name__ == "__main__":