Qwen系列有单独的封装节点
This commit is contained in:
Gin3601
@@ -42,6 +42,7 @@ _SUBPACKAGES: list[str] = [
|
||||
"model_loader", # 第 1 节：模型加载与管理
|
||||
"conditioning", # 第 2 节：文本编码与条件生成
|
||||
"sampling", # 第 3 节：图像采样与去噪
|
||||
"qwen", # 第 4 节：Qwen 一体化节点
|
||||
"latent", # 第 4 节：Latent 空间操作
|
||||
"image", # 第 5 节：图像处理与后处理
|
||||
"upscale", # 第 5 节（超分子域）：图像超分
|
||||
|
||||
@@ -0,0 +1,395 @@
|
||||
# Qwen Minimal Nodes 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:** Add two minimal high-level Qwen nodes that hide model/clip/vae wiring and directly output `IMAGE` for text-to-image and official image edit workflows.
|
||||
|
||||
**Architecture:** Introduce a dedicated `qwen/` subpackage with three focused modules: discovery for compatible model/resource filtering, runtime for thin wrappers around official ComfyUI Qwen nodes, and user-facing nodes that expose the compressed UX. Keep Qwen isolated from the existing SD1.5/SDXL/FLUX simple-chain code and validate compatibility both in the widget choices and at runtime.
|
||||
|
||||
**Tech Stack:** Python, unittest, ComfyUI core nodes, `comfy_extras.nodes_qwen`, `comfy_extras.nodes_flux`, `comfy_extras.nodes_model_advanced`, `comfy_extras.nodes_cfg`
|
||||
|
||||
---
|
||||
|
||||
### Task 1: Add Qwen discovery helpers and red tests
|
||||
|
||||
**Files:**
|
||||
- Create: `qwen/discovery.py`
|
||||
- Create: `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_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"],
|
||||
}
|
||||
|
||||
def get_filename_list(self, category):
|
||||
return list(self._files.get(category, []))
|
||||
|
||||
|
||||
def test_qwen_text_node_filters_only_text_models(self):
|
||||
load_plugin_package()
|
||||
from lls_node_test_qwen.qwen import discovery as qwen_discovery
|
||||
from lls_node_test_qwen.qwen.nodes import LLSQwenTextToImage
|
||||
|
||||
with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()):
|
||||
choices = LLSQwenTextToImage.INPUT_TYPES()["required"]["model_name"][0]
|
||||
|
||||
assert choices == [
|
||||
"qwen_image_2512_fp8_e4m3fn.safetensors",
|
||||
"qwen_image_fp8_e4m3fn.safetensors",
|
||||
]
|
||||
|
||||
|
||||
def test_qwen_edit_node_filters_only_edit_models(self):
|
||||
load_plugin_package()
|
||||
from lls_node_test_qwen.qwen import discovery as qwen_discovery
|
||||
from lls_node_test_qwen.qwen.nodes import LLSQwenImageEdit
|
||||
|
||||
with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()):
|
||||
choices = LLSQwenImageEdit.INPUT_TYPES()["required"]["model_name"][0]
|
||||
|
||||
assert choices == ["qwen_image_edit_2511_bf16.safetensors"]
|
||||
|
||||
|
||||
def test_qwen_text_node_uses_placeholder_when_no_compatible_models_exist(self):
|
||||
load_plugin_package()
|
||||
from lls_node_test_qwen.qwen import discovery as qwen_discovery
|
||||
from lls_node_test_qwen.qwen.nodes import LLSQwenTextToImage
|
||||
|
||||
stub = QwenFolderPathsStub()
|
||||
stub._files["diffusion_models"] = ["flux1-dev.safetensors", "sdxl.safetensors"]
|
||||
with mock.patch.object(qwen_discovery, "folder_paths", stub):
|
||||
choices = LLSQwenTextToImage.INPUT_TYPES()["required"]["model_name"][0]
|
||||
|
||||
assert choices == ["(no qwen text-to-image models found)"]
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Run the focused tests to verify they fail**
|
||||
|
||||
Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_filters_only_text_models tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_node_filters_only_edit_models tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_uses_placeholder_when_no_compatible_models_exist -v`
|
||||
Expected: FAIL because the `qwen/` package and node classes do not exist yet.
|
||||
|
||||
- [ ] **Step 3: Write the minimal discovery implementation**
|
||||
|
||||
```python
|
||||
TEXT_PLACEHOLDER = "(no qwen text-to-image models found)"
|
||||
EDIT_PLACEHOLDER = "(no qwen image edit models found)"
|
||||
|
||||
|
||||
def is_qwen_text_to_image_model(name: str) -> bool:
|
||||
lowered = name.lower()
|
||||
return (
|
||||
"qwen" in lowered
|
||||
and "image" in lowered
|
||||
and "edit" not in lowered
|
||||
and "layered" not in lowered
|
||||
)
|
||||
|
||||
|
||||
def is_qwen_image_edit_model(name: str) -> bool:
|
||||
lowered = name.lower()
|
||||
return "qwen" in lowered and "image" in lowered and "edit" in lowered and "layered" not in lowered
|
||||
|
||||
|
||||
def get_qwen_text_model_choices() -> list[str]:
|
||||
names = [name for name in _get_filename_list("diffusion_models") if is_qwen_text_to_image_model(name)]
|
||||
return sorted(dict.fromkeys(names)) or [TEXT_PLACEHOLDER]
|
||||
|
||||
|
||||
def get_qwen_edit_model_choices() -> list[str]:
|
||||
names = [name for name in _get_filename_list("diffusion_models") if is_qwen_image_edit_model(name)]
|
||||
return sorted(dict.fromkeys(names)) or [EDIT_PLACEHOLDER]
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Run the focused tests to verify they pass**
|
||||
|
||||
Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_filters_only_text_models tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_node_filters_only_edit_models tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_uses_placeholder_when_no_compatible_models_exist -v`
|
||||
Expected: PASS with the correct model filtering and placeholder behavior.
|
||||
|
||||
- [ ] **Step 5: Commit the discovery slice**
|
||||
|
||||
```bash
|
||||
git add qwen/discovery.py tests/test_qwen_nodes.py
|
||||
git commit -m "feat: add qwen model discovery helpers"
|
||||
```
|
||||
|
||||
### Task 2: Add runtime wrappers for official Qwen flows with red tests
|
||||
|
||||
**Files:**
|
||||
- Create: `qwen/runtime.py`
|
||||
- Modify: `tests/test_qwen_nodes.py`
|
||||
- Test: `tests/test_qwen_nodes.py`
|
||||
|
||||
- [ ] **Step 1: Write the failing runtime tests**
|
||||
|
||||
```python
|
||||
def test_qwen_text_runtime_rejects_incompatible_model_name(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
|
||||
|
||||
with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()):
|
||||
with self.assertRaisesRegex(RuntimeError, "not compatible with LLSQwenTextToImage"):
|
||||
qwen_runtime.run_qwen_text_to_image(
|
||||
model_name="qwen_image_edit_2511_bf16.safetensors",
|
||||
prompt="a cat",
|
||||
width=1024,
|
||||
height=1024,
|
||||
steps=20,
|
||||
seed=1,
|
||||
batch_size=1,
|
||||
)
|
||||
|
||||
|
||||
def test_qwen_text_runtime_raises_when_qwen_text_encoder_is_missing(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()
|
||||
stub._files["text_encoders"] = []
|
||||
with mock.patch.object(qwen_discovery, "folder_paths", stub):
|
||||
with self.assertRaisesRegex(RuntimeError, "Missing Qwen text encoder"):
|
||||
qwen_runtime.run_qwen_text_to_image(
|
||||
model_name="qwen_image_fp8_e4m3fn.safetensors",
|
||||
prompt="a cat",
|
||||
width=1024,
|
||||
height=1024,
|
||||
steps=20,
|
||||
seed=1,
|
||||
batch_size=1,
|
||||
)
|
||||
|
||||
|
||||
def test_qwen_text_runtime_executes_minimal_official_pipeline(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
|
||||
|
||||
with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()), \
|
||||
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()):
|
||||
image = qwen_runtime.run_qwen_text_to_image(
|
||||
model_name="qwen_image_fp8_e4m3fn.safetensors",
|
||||
prompt="a cat",
|
||||
width=1024,
|
||||
height=1024,
|
||||
steps=20,
|
||||
seed=99,
|
||||
batch_size=2,
|
||||
)
|
||||
|
||||
self.assertEqual(image, "IMAGE::decoded")
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Run the focused runtime tests to verify they fail**
|
||||
|
||||
Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_rejects_incompatible_model_name tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_raises_when_qwen_text_encoder_is_missing tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_executes_minimal_official_pipeline -v`
|
||||
Expected: FAIL because the Qwen runtime module and wrappers do not exist yet.
|
||||
|
||||
- [ ] **Step 3: Write the minimal runtime implementation**
|
||||
|
||||
```python
|
||||
def run_qwen_text_to_image(model_name, prompt, width, height, steps, seed, batch_size):
|
||||
validate_qwen_text_model_name(model_name)
|
||||
clip_name = resolve_qwen_text_encoder_name()
|
||||
vae_name = resolve_qwen_vae_name()
|
||||
if clip_name is None:
|
||||
raise RuntimeError("[LLS] Missing Qwen text encoder. Place a compatible Qwen VL text encoder in ComfyUI/models/text_encoders/.")
|
||||
if vae_name is None:
|
||||
raise RuntimeError("[LLS] Missing Qwen VAE. Place qwen_image_vae in ComfyUI/models/vae/.")
|
||||
|
||||
model = comfy_core_nodes.UNETLoader().load_unet(model_name, "default")[0]
|
||||
clip = comfy_core_nodes.CLIPLoader().load_clip(clip_name, type="qwen_image", device="default")[0]
|
||||
vae = comfy_core_nodes.VAELoader().load_vae(vae_name)[0]
|
||||
|
||||
model = nodes_model_advanced.ModelSamplingAuraFlow().patch_aura(model, 3.1)[0]
|
||||
positive = comfy_core_nodes.CLIPTextEncode().encode(clip, prompt)[0]
|
||||
negative = comfy_core_nodes.CLIPTextEncode().encode(clip, "")[0]
|
||||
latent = nodes_sd3.EmptySD3LatentImage().generate(width, height, batch_size)[0]
|
||||
sampled = comfy_core_nodes.KSampler().sample(
|
||||
model, seed, steps, 4.0, "euler", "simple", positive, negative, latent, denoise=1.0
|
||||
)[0]
|
||||
return comfy_core_nodes.VAEDecode().decode(vae, sampled)[0]
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Extend the runtime implementation for official image edit**
|
||||
|
||||
```python
|
||||
def run_qwen_image_edit(model_name, image, prompt, steps, seed):
|
||||
validate_qwen_edit_model_name(model_name)
|
||||
clip_name = resolve_qwen_text_encoder_name()
|
||||
vae_name = resolve_qwen_vae_name()
|
||||
...
|
||||
scaled = nodes_flux.FluxKontextImageScale.execute(image).result[0]
|
||||
positive = nodes_qwen.TextEncodeQwenImageEditPlus.execute(clip, prompt, vae=vae, image1=scaled).result[0]
|
||||
negative = nodes_qwen.TextEncodeQwenImageEditPlus.execute(clip, "", vae=vae, image1=scaled).result[0]
|
||||
positive = nodes_flux.FluxKontextMultiReferenceLatentMethod.execute(positive, "index_timestep_zero").result[0]
|
||||
negative = nodes_flux.FluxKontextMultiReferenceLatentMethod.execute(negative, "index_timestep_zero").result[0]
|
||||
latent = comfy_core_nodes.VAEEncode().encode(vae, scaled)[0]
|
||||
model = nodes_model_advanced.ModelSamplingAuraFlow().patch_aura(model, 3.1)[0]
|
||||
model = nodes_cfg.CFGNorm.execute(model, 1.0).result[0]
|
||||
sampled = comfy_core_nodes.KSampler().sample(
|
||||
model, seed, steps, 4.0, "euler", "simple", positive, negative, latent, denoise=1.0
|
||||
)[0]
|
||||
return comfy_core_nodes.VAEDecode().decode(vae, sampled)[0]
|
||||
```
|
||||
|
||||
- [ ] **Step 5: Run the focused runtime tests to verify they pass**
|
||||
|
||||
Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_rejects_incompatible_model_name tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_raises_when_qwen_text_encoder_is_missing tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_executes_minimal_official_pipeline tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_runtime_executes_minimal_official_pipeline -v`
|
||||
Expected: PASS with clear validation failures and both minimal pipelines returning decoded images through the official ComfyUI node wrappers.
|
||||
|
||||
- [ ] **Step 6: Commit the runtime slice**
|
||||
|
||||
```bash
|
||||
git add qwen/runtime.py tests/test_qwen_nodes.py
|
||||
git commit -m "feat: add qwen runtime wrappers"
|
||||
```
|
||||
|
||||
### Task 3: Add user-facing nodes and register the new subpackage
|
||||
|
||||
**Files:**
|
||||
- Create: `qwen/__init__.py`
|
||||
- Create: `qwen/nodes.py`
|
||||
- Modify: `__init__.py`
|
||||
- Modify: `tests/test_qwen_nodes.py`
|
||||
- Test: `tests/test_qwen_nodes.py`
|
||||
|
||||
- [ ] **Step 1: Write the failing node registration and schema tests**
|
||||
|
||||
```python
|
||||
def test_plugin_registers_qwen_nodes(self):
|
||||
plugin = load_plugin_package()
|
||||
self.assertIn("LLSQwenTextToImage", plugin.NODE_CLASS_MAPPINGS)
|
||||
self.assertIn("LLSQwenImageEdit", plugin.NODE_CLASS_MAPPINGS)
|
||||
|
||||
|
||||
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(
|
||||
tuple(required.keys()),
|
||||
("model_name", "prompt", "width", "height", "steps", "seed", "batch_size"),
|
||||
)
|
||||
|
||||
|
||||
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(tuple(required.keys()), ("model_name", "image", "prompt", "steps", "seed"))
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Run the focused node tests to verify they fail**
|
||||
|
||||
Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_plugin_registers_qwen_nodes tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_returns_image_only tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_node_returns_image_only -v`
|
||||
Expected: FAIL because the Qwen subpackage is not yet registered and the node classes do not exist.
|
||||
|
||||
- [ ] **Step 3: Write the minimal node layer**
|
||||
|
||||
```python
|
||||
class LLSQwenTextToImage:
|
||||
CATEGORY = "LLS/Qwen"
|
||||
FUNCTION = "generate"
|
||||
RETURN_TYPES = ("IMAGE",)
|
||||
RETURN_NAMES = ("image",)
|
||||
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"model_name": (get_qwen_text_model_choices(),),
|
||||
"prompt": ("STRING", {"default": "", "multiline": True}),
|
||||
"width": ("INT", {"default": 1024, "min": 16, "max": 8192, "step": 16}),
|
||||
"height": ("INT", {"default": 1024, "min": 16, "max": 8192, "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}),
|
||||
}
|
||||
}
|
||||
|
||||
def generate(self, model_name, prompt, width, height, steps, seed, batch_size):
|
||||
return (run_qwen_text_to_image(model_name, prompt, width, height, steps, seed, batch_size),)
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Register the subpackage**
|
||||
|
||||
```python
|
||||
_SUBPACKAGES: list[str] = [
|
||||
"model_loader",
|
||||
"conditioning",
|
||||
"sampling",
|
||||
"qwen",
|
||||
"latent",
|
||||
...
|
||||
]
|
||||
```
|
||||
|
||||
- [ ] **Step 5: Run the node tests to verify they pass**
|
||||
|
||||
Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_plugin_registers_qwen_nodes tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_returns_image_only tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_node_returns_image_only tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_executes_runtime tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_node_executes_runtime -v`
|
||||
Expected: PASS with both nodes registered, exposing only `IMAGE`, and delegating to the runtime layer.
|
||||
|
||||
- [ ] **Step 6: Commit the node layer**
|
||||
|
||||
```bash
|
||||
git add __init__.py qwen/__init__.py qwen/nodes.py tests/test_qwen_nodes.py
|
||||
git commit -m "feat: add minimal qwen image nodes"
|
||||
```
|
||||
|
||||
### Task 4: Run the full regression suite for the plugin
|
||||
|
||||
**Files:**
|
||||
- Modify: `tests/test_qwen_nodes.py` (only if a verification run exposes a mismatch)
|
||||
- Test: `tests/test_qwen_nodes.py`
|
||||
|
||||
- [ ] **Step 1: Run the dedicated Qwen test file**
|
||||
|
||||
Run: `python3 -m unittest tests.test_qwen_nodes -v`
|
||||
Expected: PASS with zero failures for discovery, validation, runtime, registration, and execution-path coverage.
|
||||
|
||||
- [ ] **Step 2: Run the full existing test suite**
|
||||
|
||||
Run: `python3 -m unittest discover -s tests -v`
|
||||
Expected: PASS with zero regressions in the existing loader/prompt/latent/upscale suites.
|
||||
|
||||
- [ ] **Step 3: Review the changed files before closeout**
|
||||
|
||||
```bash
|
||||
git status --short
|
||||
```
|
||||
|
||||
Expected:
|
||||
- `M __init__.py`
|
||||
- `A qwen/__init__.py`
|
||||
- `A qwen/discovery.py`
|
||||
- `A qwen/runtime.py`
|
||||
- `A qwen/nodes.py`
|
||||
- `A tests/test_qwen_nodes.py`
|
||||
- `A docs/superpowers/plans/2026-05-20-qwen-minimal-nodes.md`
|
||||
|
||||
- [ ] **Step 4: Commit the verified final state**
|
||||
|
||||
```bash
|
||||
git add __init__.py qwen tests/test_qwen_nodes.py docs/superpowers/plans/2026-05-20-qwen-minimal-nodes.md
|
||||
git commit -m "test: verify minimal qwen nodes"
|
||||
```
|
||||
@@ -0,0 +1,8 @@
|
||||
"""
|
||||
node.qwen
|
||||
=========
|
||||
功能域：Qwen 一体化图像节点
|
||||
"""
|
||||
from .nodes import NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS
|
||||
|
||||
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS"]
|
||||
@@ -0,0 +1,115 @@
|
||||
"""
|
||||
Qwen model/resource discovery helpers.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
try:
|
||||
import folder_paths
|
||||
except Exception:
|
||||
folder_paths = None
|
||||
|
||||
|
||||
TEXT_MODEL_PLACEHOLDER = "(no qwen text-to-image models found)"
|
||||
EDIT_MODEL_PLACEHOLDER = "(no qwen image edit models found)"
|
||||
|
||||
_QWEN_TEXT_ENCODER_PATTERNS = (
|
||||
"qwen_2.5_vl_7b_fp8_scaled.safetensors",
|
||||
"qwen_2.5_vl_7b",
|
||||
"qwen25_7b",
|
||||
)
|
||||
_QWEN_VAE_PATTERNS = (
|
||||
"qwen_image_vae.safetensors",
|
||||
"qwen_image_vae",
|
||||
)
|
||||
|
||||
|
||||
def _get_filename_list(category: str) -> list[str]:
|
||||
if folder_paths is None:
|
||||
return []
|
||||
try:
|
||||
return list(folder_paths.get_filename_list(category))
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
def _sorted_unique(items: list[str]) -> list[str]:
|
||||
return sorted(dict.fromkeys(item for item in items if item))
|
||||
|
||||
|
||||
def _match_by_patterns(names: list[str], patterns: tuple[str, ...]) -> str | None:
|
||||
lowered = {name.lower(): name for name in names}
|
||||
for pattern in patterns:
|
||||
pattern_lower = pattern.lower()
|
||||
for candidate_lower, candidate in lowered.items():
|
||||
if candidate_lower == pattern_lower or pattern_lower in candidate_lower:
|
||||
return candidate
|
||||
return None
|
||||
|
||||
|
||||
def is_qwen_text_to_image_model(name: str | None) -> bool:
|
||||
if not name:
|
||||
return False
|
||||
lowered = name.lower()
|
||||
return (
|
||||
"qwen" in lowered
|
||||
and "image" in lowered
|
||||
and "edit" not in lowered
|
||||
and "layered" not in lowered
|
||||
)
|
||||
|
||||
|
||||
def is_qwen_image_edit_model(name: str | None) -> bool:
|
||||
if not name:
|
||||
return False
|
||||
lowered = name.lower()
|
||||
return "qwen" in lowered and "image" in lowered and "edit" in lowered and "layered" not in lowered
|
||||
|
||||
|
||||
def get_qwen_text_model_choices() -> list[str]:
|
||||
names = [name for name in _get_filename_list("diffusion_models") if is_qwen_text_to_image_model(name)]
|
||||
return _sorted_unique(names) or [TEXT_MODEL_PLACEHOLDER]
|
||||
|
||||
|
||||
def get_qwen_edit_model_choices() -> list[str]:
|
||||
names = [name for name in _get_filename_list("diffusion_models") if is_qwen_image_edit_model(name)]
|
||||
return _sorted_unique(names) or [EDIT_MODEL_PLACEHOLDER]
|
||||
|
||||
|
||||
def validate_qwen_text_model_name(model_name: str) -> str:
|
||||
if not is_qwen_text_to_image_model(model_name):
|
||||
raise RuntimeError(
|
||||
f"[LLS] Model '{model_name}' is not compatible with LLSQwenTextToImage."
|
||||
)
|
||||
return model_name
|
||||
|
||||
|
||||
def validate_qwen_edit_model_name(model_name: str) -> str:
|
||||
if not is_qwen_image_edit_model(model_name):
|
||||
raise RuntimeError(
|
||||
f"[LLS] Model '{model_name}' is not compatible with LLSQwenImageEdit."
|
||||
)
|
||||
return model_name
|
||||
|
||||
|
||||
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()
|
||||
if "qwen" in lowered and ("vl" in lowered or "qwen25_7b" in lowered):
|
||||
return name
|
||||
return None
|
||||
|
||||
|
||||
def resolve_qwen_vae_name() -> str | None:
|
||||
names = _get_filename_list("vae")
|
||||
matched = _match_by_patterns(names, _QWEN_VAE_PATTERNS)
|
||||
if matched is not None:
|
||||
return matched
|
||||
for name in names:
|
||||
lowered = name.lower()
|
||||
if "qwen" in lowered and "vae" in lowered and "layered" not in lowered:
|
||||
return name
|
||||
return None
|
||||
+114
@@ -0,0 +1,114 @@
|
||||
"""
|
||||
Minimal high-level Qwen nodes.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
|
||||
from . import discovery, runtime
|
||||
|
||||
try:
|
||||
comfy_core_nodes = importlib.import_module("nodes")
|
||||
except Exception:
|
||||
comfy_core_nodes = None
|
||||
|
||||
_MAX_RESOLUTION = int(getattr(comfy_core_nodes, "MAX_RESOLUTION", 8192))
|
||||
|
||||
|
||||
class LLSQwenTextToImage:
|
||||
CATEGORY = "LLS/Qwen"
|
||||
FUNCTION = "generate"
|
||||
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."
|
||||
)
|
||||
|
||||
@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}),
|
||||
}
|
||||
}
|
||||
|
||||
def generate(
|
||||
self,
|
||||
model_name: str,
|
||||
prompt: str,
|
||||
width: int,
|
||||
height: int,
|
||||
steps: int,
|
||||
seed: int,
|
||||
batch_size: int,
|
||||
):
|
||||
return (
|
||||
runtime.run_qwen_text_to_image(
|
||||
model_name=model_name,
|
||||
prompt=prompt,
|
||||
width=width,
|
||||
height=height,
|
||||
steps=steps,
|
||||
seed=seed,
|
||||
batch_size=batch_size,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class LLSQwenImageEdit:
|
||||
CATEGORY = "LLS/Qwen"
|
||||
FUNCTION = "generate"
|
||||
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."
|
||||
)
|
||||
|
||||
@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}),
|
||||
}
|
||||
}
|
||||
|
||||
def generate(
|
||||
self,
|
||||
model_name: str,
|
||||
image,
|
||||
prompt: str,
|
||||
steps: int,
|
||||
seed: int,
|
||||
):
|
||||
return (
|
||||
runtime.run_qwen_image_edit(
|
||||
model_name=model_name,
|
||||
image=image,
|
||||
prompt=prompt,
|
||||
steps=steps,
|
||||
seed=seed,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
NODE_CLASS_MAPPINGS: dict[str, type] = {
|
||||
"LLSQwenTextToImage": LLSQwenTextToImage,
|
||||
"LLSQwenImageEdit": LLSQwenImageEdit,
|
||||
}
|
||||
|
||||
NODE_DISPLAY_NAME_MAPPINGS: dict[str, str] = {
|
||||
"LLSQwenTextToImage": "LLS Qwen Text To Image",
|
||||
"LLSQwenImageEdit": "LLS Qwen Image Edit",
|
||||
}
|
||||
+266
@@ -0,0 +1,266 @@
|
||||
"""
|
||||
Runtime wrappers for official ComfyUI Qwen image pipelines.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
from typing import Any
|
||||
|
||||
try:
|
||||
import folder_paths
|
||||
except Exception:
|
||||
folder_paths = None
|
||||
|
||||
try:
|
||||
comfy_core_nodes = importlib.import_module("nodes")
|
||||
except Exception as exc:
|
||||
comfy_core_nodes = None
|
||||
_CORE_NODES_ERR = exc
|
||||
else:
|
||||
_CORE_NODES_ERR = None
|
||||
|
||||
try:
|
||||
nodes_sd3 = importlib.import_module("comfy_extras.nodes_sd3")
|
||||
except Exception as exc:
|
||||
nodes_sd3 = None
|
||||
_SD3_ERR = exc
|
||||
else:
|
||||
_SD3_ERR = None
|
||||
|
||||
try:
|
||||
nodes_model_advanced = importlib.import_module("comfy_extras.nodes_model_advanced")
|
||||
except Exception as exc:
|
||||
nodes_model_advanced = None
|
||||
_MODEL_ADVANCED_ERR = exc
|
||||
else:
|
||||
_MODEL_ADVANCED_ERR = None
|
||||
|
||||
try:
|
||||
nodes_qwen = importlib.import_module("comfy_extras.nodes_qwen")
|
||||
except Exception as exc:
|
||||
nodes_qwen = None
|
||||
_QWEN_ERR = exc
|
||||
else:
|
||||
_QWEN_ERR = None
|
||||
|
||||
try:
|
||||
nodes_flux = importlib.import_module("comfy_extras.nodes_flux")
|
||||
except Exception as exc:
|
||||
nodes_flux = None
|
||||
_FLUX_ERR = exc
|
||||
else:
|
||||
_FLUX_ERR = None
|
||||
|
||||
try:
|
||||
nodes_cfg = importlib.import_module("comfy_extras.nodes_cfg")
|
||||
except Exception as exc:
|
||||
nodes_cfg = None
|
||||
_CFG_ERR = exc
|
||||
else:
|
||||
_CFG_ERR = None
|
||||
|
||||
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):
|
||||
if hasattr(result, "result"):
|
||||
values = result.result
|
||||
if values is None:
|
||||
return None
|
||||
return values[0]
|
||||
if isinstance(result, tuple):
|
||||
return result[0]
|
||||
if isinstance(result, list):
|
||||
return result[0]
|
||||
return result
|
||||
|
||||
|
||||
def _require_module(module: Any, label: str, import_error: Exception | None = None):
|
||||
if module is None:
|
||||
raise RuntimeError(
|
||||
f"[LLS] Required ComfyUI runtime component '{label}' is unavailable. "
|
||||
"Use a ComfyUI build that includes official Qwen image support."
|
||||
) from import_error
|
||||
return module
|
||||
|
||||
|
||||
def _require_class(module: Any, class_name: str, import_error: Exception | None = None):
|
||||
_require_module(module, class_name, import_error)
|
||||
cls = getattr(module, class_name, None)
|
||||
if cls is None:
|
||||
raise RuntimeError(
|
||||
f"[LLS] Required ComfyUI runtime component '{class_name}' is unavailable. "
|
||||
"Use a ComfyUI build that includes official Qwen image support."
|
||||
) from import_error
|
||||
return cls
|
||||
|
||||
|
||||
def _load_qwen_clip(clip_name: str):
|
||||
clip_loader_cls = _require_class(comfy_core_nodes, "CLIPLoader", _CORE_NODES_ERR)
|
||||
loader = clip_loader_cls()
|
||||
try:
|
||||
return loader.load_clip(clip_name, type=_QWEN_CLIP_TYPE, device="default")[0]
|
||||
except TypeError:
|
||||
return loader.load_clip(clip_name, type=_QWEN_CLIP_TYPE)[0]
|
||||
|
||||
|
||||
def _load_qwen_vae(vae_name: str):
|
||||
vae_loader_cls = _require_class(comfy_core_nodes, "VAELoader", _CORE_NODES_ERR)
|
||||
return vae_loader_cls().load_vae(vae_name)[0]
|
||||
|
||||
|
||||
def _load_qwen_model(model_name: str):
|
||||
unet_loader_cls = _require_class(comfy_core_nodes, "UNETLoader", _CORE_NODES_ERR)
|
||||
return unet_loader_cls().load_unet(model_name, "default")[0]
|
||||
|
||||
|
||||
def _patch_qwen_model_sampling(model):
|
||||
patcher_cls = _require_class(nodes_model_advanced, "ModelSamplingAuraFlow", _MODEL_ADVANCED_ERR)
|
||||
return patcher_cls().patch_aura(model, _QWEN_MODEL_SHIFT)[0]
|
||||
|
||||
|
||||
def _apply_cfg_norm(model):
|
||||
cfg_cls = _require_class(nodes_cfg, "CFGNorm", _CFG_ERR)
|
||||
return _unwrap_first(cfg_cls.execute(model, _QWEN_CFG_NORM_STRENGTH))
|
||||
|
||||
|
||||
def _encode_clip_text(clip, prompt: str):
|
||||
encoder_cls = _require_class(comfy_core_nodes, "CLIPTextEncode", _CORE_NODES_ERR)
|
||||
return encoder_cls().encode(clip, prompt)[0]
|
||||
|
||||
|
||||
def _create_empty_qwen_latent(width: int, height: int, batch_size: int):
|
||||
latent_cls = _require_class(nodes_sd3, "EmptySD3LatentImage", _SD3_ERR)
|
||||
latent_node = latent_cls()
|
||||
if hasattr(latent_node, "generate"):
|
||||
return latent_node.generate(width, height, batch_size)[0]
|
||||
return latent_node.execute(width, height, batch_size)[0]
|
||||
|
||||
|
||||
def _encode_qwen_edit_conditioning(clip, prompt: str, vae, image):
|
||||
encoder_cls = _require_class(nodes_qwen, "TextEncodeQwenImageEditPlus", _QWEN_ERR)
|
||||
return _unwrap_first(encoder_cls.execute(clip, prompt, vae=vae, image1=image))
|
||||
|
||||
|
||||
def _scale_qwen_edit_image(image):
|
||||
scaler_cls = _require_class(nodes_flux, "FluxKontextImageScale", _FLUX_ERR)
|
||||
return _unwrap_first(scaler_cls.execute(image))
|
||||
|
||||
|
||||
def _apply_reference_latents_method(conditioning):
|
||||
method_cls = _require_class(nodes_flux, "FluxKontextMultiReferenceLatentMethod", _FLUX_ERR)
|
||||
return _unwrap_first(method_cls.execute(conditioning, _QWEN_REFERENCE_LATENTS_METHOD))
|
||||
|
||||
|
||||
def _encode_image_to_latent(vae, image):
|
||||
encoder_cls = _require_class(comfy_core_nodes, "VAEEncode", _CORE_NODES_ERR)
|
||||
return encoder_cls().encode(vae, image)[0]
|
||||
|
||||
|
||||
def _sample_qwen(model, positive, negative, latent, steps: int, seed: int):
|
||||
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,
|
||||
positive,
|
||||
negative,
|
||||
latent,
|
||||
denoise=1.0,
|
||||
)[0]
|
||||
|
||||
|
||||
def _decode_qwen_latent(vae, latent):
|
||||
decoder_cls = _require_class(comfy_core_nodes, "VAEDecode", _CORE_NODES_ERR)
|
||||
return decoder_cls().decode(vae, latent)[0]
|
||||
|
||||
|
||||
def _resolve_qwen_resources(model_name: str, validator):
|
||||
validator(model_name)
|
||||
clip_name = discovery.resolve_qwen_text_encoder_name()
|
||||
vae_name = discovery.resolve_qwen_vae_name()
|
||||
|
||||
if clip_name is None:
|
||||
raise RuntimeError(
|
||||
"[LLS] Missing Qwen text encoder. Place a compatible Qwen VL text encoder in "
|
||||
"ComfyUI/models/text_encoders/."
|
||||
)
|
||||
if vae_name is None:
|
||||
raise RuntimeError(
|
||||
"[LLS] Missing Qwen VAE. Place qwen_image_vae.safetensors in ComfyUI/models/vae/."
|
||||
)
|
||||
|
||||
return model_name, clip_name, vae_name
|
||||
|
||||
|
||||
def run_qwen_text_to_image(
|
||||
model_name: str,
|
||||
prompt: str,
|
||||
width: int,
|
||||
height: int,
|
||||
steps: int,
|
||||
seed: int,
|
||||
batch_size: int,
|
||||
):
|
||||
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))
|
||||
clip = _load_qwen_clip(clip_name)
|
||||
vae = _load_qwen_vae(vae_name)
|
||||
positive = _encode_clip_text(clip, prompt)
|
||||
negative = _encode_clip_text(clip, "")
|
||||
latent = _create_empty_qwen_latent(int(width), int(height), int(batch_size))
|
||||
sampled = _sample_qwen(model, positive, negative, latent, int(steps), int(seed))
|
||||
return _decode_qwen_latent(vae, sampled)
|
||||
except Exception as exc:
|
||||
if str(exc).startswith("[LLS]"):
|
||||
raise
|
||||
raise RuntimeError(f"[LLS] Qwen text-to-image failed: {exc}") from exc
|
||||
|
||||
|
||||
def run_qwen_image_edit(
|
||||
model_name: str,
|
||||
image,
|
||||
prompt: str,
|
||||
steps: int,
|
||||
seed: int,
|
||||
):
|
||||
try:
|
||||
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 = _patch_qwen_model_sampling(_load_qwen_model(model_name))
|
||||
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)
|
||||
latent = _encode_image_to_latent(vae, scaled)
|
||||
model = _apply_cfg_norm(model)
|
||||
sampled = _sample_qwen(model, positive, negative, latent, int(steps), int(seed))
|
||||
return _decode_qwen_latent(vae, sampled)
|
||||
except Exception as exc:
|
||||
if str(exc).startswith("[LLS]"):
|
||||
raise
|
||||
raise RuntimeError(f"[LLS] Qwen image edit failed: {exc}") from exc
|
||||
@@ -0,0 +1,535 @@
|
||||
import importlib.util
|
||||
import pathlib
|
||||
import sys
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
|
||||
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def load_plugin_package():
|
||||
for name in list(sys.modules):
|
||||
if name == "lls_node_test_qwen" or name.startswith("lls_node_test_qwen."):
|
||||
sys.modules.pop(name)
|
||||
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"lls_node_test_qwen",
|
||||
ROOT / "__init__.py",
|
||||
submodule_search_locations=[str(ROOT)],
|
||||
)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
sys.modules["lls_node_test_qwen"] = module
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
class NodeOutputStub:
|
||||
def __init__(self, *args):
|
||||
self.args = args
|
||||
|
||||
@property
|
||||
def result(self):
|
||||
return self.args
|
||||
|
||||
def __getitem__(self, index):
|
||||
return self.args[index]
|
||||
|
||||
|
||||
class QwenFolderPathsStub:
|
||||
def __init__(self):
|
||||
self._files = {
|
||||
"diffusion_models": [
|
||||
"qwen_image_fp8_e4m3fn.safetensors",
|
||||
"qwen_image_2512_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"],
|
||||
}
|
||||
|
||||
def get_filename_list(self, category):
|
||||
return list(self._files.get(category, []))
|
||||
|
||||
|
||||
class QwenClipStub:
|
||||
def __init__(self, clip_name, clip_type):
|
||||
self.clip_name = clip_name
|
||||
self.clip_type = clip_type
|
||||
|
||||
|
||||
class QwenVAEStub:
|
||||
def __init__(self, vae_name):
|
||||
self.vae_name = vae_name
|
||||
|
||||
|
||||
class CoreQwenNodesStub:
|
||||
last_unet_call = None
|
||||
last_clip_call = None
|
||||
last_text_encode_calls = []
|
||||
last_ksampler_call = None
|
||||
last_vae_encode_call = None
|
||||
last_vae_decode_call = None
|
||||
last_vae_load_call = None
|
||||
|
||||
@classmethod
|
||||
def reset(cls):
|
||||
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 UNETLoader:
|
||||
def load_unet(self, unet_name, weight_dtype):
|
||||
CoreQwenNodesStub.last_unet_call = {
|
||||
"unet_name": unet_name,
|
||||
"weight_dtype": weight_dtype,
|
||||
}
|
||||
return (f"MODEL::{unet_name}",)
|
||||
|
||||
class CLIPLoader:
|
||||
def load_clip(self, clip_name, type="stable_diffusion", device="default"):
|
||||
CoreQwenNodesStub.last_clip_call = {
|
||||
"clip_name": clip_name,
|
||||
"type": type,
|
||||
"device": device,
|
||||
}
|
||||
return (QwenClipStub(clip_name, type),)
|
||||
|
||||
class VAELoader:
|
||||
def load_vae(self, vae_name):
|
||||
CoreQwenNodesStub.last_vae_load_call = {"vae_name": vae_name}
|
||||
return (QwenVAEStub(vae_name),)
|
||||
|
||||
class CLIPTextEncode:
|
||||
def encode(self, clip, text):
|
||||
CoreQwenNodesStub.last_text_encode_calls.append(
|
||||
{
|
||||
"clip_name": clip.clip_name,
|
||||
"clip_type": clip.clip_type,
|
||||
"text": text,
|
||||
}
|
||||
)
|
||||
return (f"COND::{text}",)
|
||||
|
||||
class KSampler:
|
||||
def sample(
|
||||
self,
|
||||
model,
|
||||
seed,
|
||||
steps,
|
||||
cfg,
|
||||
sampler_name,
|
||||
scheduler,
|
||||
positive,
|
||||
negative,
|
||||
latent_image,
|
||||
denoise=1.0,
|
||||
):
|
||||
CoreQwenNodesStub.last_ksampler_call = {
|
||||
"model": model,
|
||||
"seed": seed,
|
||||
"steps": steps,
|
||||
"cfg": cfg,
|
||||
"sampler_name": sampler_name,
|
||||
"scheduler": scheduler,
|
||||
"positive": positive,
|
||||
"negative": negative,
|
||||
"latent_image": latent_image,
|
||||
"denoise": denoise,
|
||||
}
|
||||
return ({"samples": f"SAMPLED::{seed}::{steps}"},)
|
||||
|
||||
class VAEDecode:
|
||||
def decode(self, vae, samples):
|
||||
CoreQwenNodesStub.last_vae_decode_call = {
|
||||
"vae_name": vae.vae_name,
|
||||
"samples": samples,
|
||||
}
|
||||
return (f"IMAGE::{samples['samples']}",)
|
||||
|
||||
class VAEEncode:
|
||||
def encode(self, vae, pixels):
|
||||
CoreQwenNodesStub.last_vae_encode_call = {
|
||||
"vae_name": vae.vae_name,
|
||||
"pixels": pixels,
|
||||
}
|
||||
return ({"samples": f"ENCODED::{pixels}"},)
|
||||
|
||||
|
||||
class SD3NodesStub:
|
||||
last_generate_call = None
|
||||
|
||||
@classmethod
|
||||
def reset(cls):
|
||||
cls.last_generate_call = None
|
||||
|
||||
class EmptySD3LatentImage:
|
||||
def generate(self, width, height, batch_size=1):
|
||||
SD3NodesStub.last_generate_call = {
|
||||
"width": width,
|
||||
"height": height,
|
||||
"batch_size": batch_size,
|
||||
}
|
||||
return (
|
||||
{
|
||||
"samples": f"EMPTY::{width}x{height}::{batch_size}",
|
||||
"downscale_ratio_spacial": 8,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
class ModelAdvancedStub:
|
||||
last_call = None
|
||||
|
||||
@classmethod
|
||||
def reset(cls):
|
||||
cls.last_call = None
|
||||
|
||||
class ModelSamplingAuraFlow:
|
||||
def patch_aura(self, model, shift):
|
||||
ModelAdvancedStub.last_call = {"model": model, "shift": shift}
|
||||
return (f"AURA::{model}::{shift}",)
|
||||
|
||||
|
||||
class FluxNodesStub:
|
||||
last_scale_call = None
|
||||
last_reference_calls = []
|
||||
|
||||
@classmethod
|
||||
def reset(cls):
|
||||
cls.last_scale_call = None
|
||||
cls.last_reference_calls = []
|
||||
|
||||
class FluxKontextImageScale:
|
||||
@classmethod
|
||||
def execute(cls, image):
|
||||
FluxNodesStub.last_scale_call = {"image": image}
|
||||
return NodeOutputStub(f"SCALED::{image}")
|
||||
|
||||
class FluxKontextMultiReferenceLatentMethod:
|
||||
@classmethod
|
||||
def execute(cls, conditioning, reference_latents_method):
|
||||
FluxNodesStub.last_reference_calls.append(
|
||||
{
|
||||
"conditioning": conditioning,
|
||||
"reference_latents_method": reference_latents_method,
|
||||
}
|
||||
)
|
||||
return NodeOutputStub(f"{conditioning}::{reference_latents_method}")
|
||||
|
||||
|
||||
class QwenExtraNodesStub:
|
||||
last_calls = []
|
||||
|
||||
@classmethod
|
||||
def reset(cls):
|
||||
cls.last_calls = []
|
||||
|
||||
class TextEncodeQwenImageEditPlus:
|
||||
@classmethod
|
||||
def execute(cls, clip, prompt, vae=None, image1=None, image2=None, image3=None):
|
||||
QwenExtraNodesStub.last_calls.append(
|
||||
{
|
||||
"clip_name": clip.clip_name,
|
||||
"prompt": prompt,
|
||||
"vae_name": getattr(vae, "vae_name", None),
|
||||
"image1": image1,
|
||||
"image2": image2,
|
||||
"image3": image3,
|
||||
}
|
||||
)
|
||||
return NodeOutputStub(f"EDIT_COND::{prompt}::{image1}")
|
||||
|
||||
|
||||
class CFGNodesStub:
|
||||
last_call = None
|
||||
|
||||
@classmethod
|
||||
def reset(cls):
|
||||
cls.last_call = None
|
||||
|
||||
class CFGNorm:
|
||||
@classmethod
|
||||
def execute(cls, model, strength):
|
||||
CFGNodesStub.last_call = {"model": model, "strength": strength}
|
||||
return NodeOutputStub(f"CFGNORM::{model}::{strength}")
|
||||
|
||||
|
||||
class TestQwenNodes(unittest.TestCase):
|
||||
def setUp(self):
|
||||
CoreQwenNodesStub.reset()
|
||||
SD3NodesStub.reset()
|
||||
ModelAdvancedStub.reset()
|
||||
FluxNodesStub.reset()
|
||||
QwenExtraNodesStub.reset()
|
||||
CFGNodesStub.reset()
|
||||
|
||||
def test_plugin_registers_qwen_nodes(self):
|
||||
plugin = load_plugin_package()
|
||||
|
||||
self.assertIn("LLSQwenTextToImage", plugin.NODE_CLASS_MAPPINGS)
|
||||
self.assertIn("LLSQwenImageEdit", plugin.NODE_CLASS_MAPPINGS)
|
||||
|
||||
def test_qwen_text_node_filters_only_text_models(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"]["model_name"][0]
|
||||
|
||||
self.assertEqual(
|
||||
choices,
|
||||
[
|
||||
"qwen_image_2512_fp8_e4m3fn.safetensors",
|
||||
"qwen_image_fp8_e4m3fn.safetensors",
|
||||
],
|
||||
)
|
||||
|
||||
def test_qwen_edit_node_filters_only_edit_models(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"]["model_name"][0]
|
||||
|
||||
self.assertEqual(choices, ["qwen_image_edit_2511_bf16.safetensors"])
|
||||
|
||||
def test_qwen_text_node_uses_placeholder_when_no_compatible_models_exist(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]
|
||||
|
||||
self.assertEqual(choices, ["(no qwen text-to-image models found)"])
|
||||
|
||||
def test_qwen_text_runtime_rejects_incompatible_model_name(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, "not compatible with LLSQwenTextToImage"):
|
||||
qwen_runtime.run_qwen_text_to_image(
|
||||
model_name="qwen_image_edit_2511_bf16.safetensors",
|
||||
prompt="a cat",
|
||||
width=1024,
|
||||
height=1024,
|
||||
steps=20,
|
||||
seed=1,
|
||||
batch_size=1,
|
||||
)
|
||||
|
||||
def test_qwen_edit_runtime_rejects_incompatible_model_name(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, "not compatible with LLSQwenImageEdit"):
|
||||
qwen_runtime.run_qwen_image_edit(
|
||||
model_name="qwen_image_fp8_e4m3fn.safetensors",
|
||||
image="IMAGE::input",
|
||||
prompt="edit the cat",
|
||||
steps=20,
|
||||
seed=1,
|
||||
)
|
||||
|
||||
def test_qwen_text_runtime_raises_when_qwen_text_encoder_is_missing(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()
|
||||
stub._files["text_encoders"] = []
|
||||
with mock.patch.object(qwen_discovery, "folder_paths", stub), mock.patch.object(
|
||||
qwen_runtime,
|
||||
"folder_paths",
|
||||
stub,
|
||||
):
|
||||
with self.assertRaisesRegex(RuntimeError, "Missing Qwen text encoder"):
|
||||
qwen_runtime.run_qwen_text_to_image(
|
||||
model_name="qwen_image_fp8_e4m3fn.safetensors",
|
||||
prompt="a cat",
|
||||
width=1024,
|
||||
height=1024,
|
||||
steps=20,
|
||||
seed=1,
|
||||
batch_size=1,
|
||||
)
|
||||
|
||||
def test_qwen_text_runtime_raises_when_qwen_vae_is_missing(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()
|
||||
stub._files["vae"] = []
|
||||
with mock.patch.object(qwen_discovery, "folder_paths", stub), mock.patch.object(
|
||||
qwen_runtime,
|
||||
"folder_paths",
|
||||
stub,
|
||||
):
|
||||
with self.assertRaisesRegex(RuntimeError, "Missing Qwen VAE"):
|
||||
qwen_runtime.run_qwen_text_to_image(
|
||||
model_name="qwen_image_fp8_e4m3fn.safetensors",
|
||||
prompt="a cat",
|
||||
width=1024,
|
||||
height=1024,
|
||||
steps=20,
|
||||
seed=1,
|
||||
batch_size=1,
|
||||
)
|
||||
|
||||
def test_qwen_text_runtime_executes_minimal_official_pipeline(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):
|
||||
image = qwen_runtime.run_qwen_text_to_image(
|
||||
model_name="qwen_image_fp8_e4m3fn.safetensors",
|
||||
prompt="a cat",
|
||||
width=1024,
|
||||
height=1152,
|
||||
steps=20,
|
||||
seed=99,
|
||||
batch_size=2,
|
||||
)
|
||||
|
||||
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(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)
|
||||
|
||||
def test_qwen_edit_runtime_executes_minimal_official_pipeline(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):
|
||||
image = qwen_runtime.run_qwen_image_edit(
|
||||
model_name="qwen_image_edit_2511_bf16.safetensors",
|
||||
image="IMAGE::input",
|
||||
prompt="turn the cat blue",
|
||||
steps=30,
|
||||
seed=123,
|
||||
)
|
||||
|
||||
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"),
|
||||
)
|
||||
|
||||
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"))
|
||||
|
||||
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"):
|
||||
result = node.generate(
|
||||
model_name="qwen_image_fp8_e4m3fn.safetensors",
|
||||
prompt="a cat",
|
||||
width=1024,
|
||||
height=1024,
|
||||
steps=20,
|
||||
seed=1,
|
||||
batch_size=1,
|
||||
)
|
||||
|
||||
self.assertEqual(result, ("IMAGE::node",))
|
||||
|
||||
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"):
|
||||
result = node.generate(
|
||||
model_name="qwen_image_edit_2511_bf16.safetensors",
|
||||
image="IMAGE::input",
|
||||
prompt="edit the cat",
|
||||
steps=20,
|
||||
seed=1,
|
||||
)
|
||||
|
||||
self.assertEqual(result, ("IMAGE::node-edit",))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user