nodes/Gin3601-LLS-node
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: remove context ports from core nodes

Browse Source
This commit is contained in:
Gin3601
2026-05-20 13:35:19 +08:00
parent 712633b90f
commit 475cc3a246
13 changed files with 705 additions and 670 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__init__.py
-1
View File
@@ -39,7 +39,6 @@ import types
# ---------- 子包白名单（控制加载顺序，同时作为文档） ----------
# 按功能分类总览章节顺序排列；新增子包追加到列表末尾即可。
_SUBPACKAGES: list[str] = [
"task", # 统一任务控制与上下文
"model_loader", # 第 1 节：模型加载与管理
"conditioning", # 第 2 节：文本编码与条件生成
"sampling", # 第 3 节：图像采样与去噪
conditioning/nodes.py
+35 -35
View File
@@ -8,12 +8,13 @@ CATEGORY = "LLS/Conditioning"
from __future__ import annotations
from ..utils.model_info import (
LLS_TEXT_ENCODER_TYPE,
MODEL_FAMILY_CHOICES,
get_family_defaults,
info_to_json,
is_flux_family,
is_sdxl_family,
parse_model_info,
resolve_model_family,
)
from ..utils.task_context import LLS_TASK_CONTEXT_TYPE, parse_task_context, update_task_context
_CLIP_SKIP_CHOICES = [None] + list(range(-1, -25, -1))
@@ -120,19 +121,18 @@ def _encode_flux_prompt(clip, prompt: str, guidance):
class LLSSimplePromptEncode:
"""
家族感知的提示词编码节点。
- `text_encoder` 为新接口。
- `clip` 保留作旧工作流兼容入口。
- `task_context` 为统一控制对象；未连接时会退回 SD1.5 默认逻辑。
- `text_encoder` 与 `clip` 都使用原生 CLIP 端口类型。
- `model_family` 默认 Auto，从 CLIP tokenizer 或对象标记推断家族。
- 输出 prompt_info JSON，而不是 task_context 透传对象。
"""
CATEGORY = "LLS/Conditioning"
FUNCTION = "encode"
RETURN_TYPES = ("CONDITIONING", "CONDITIONING", LLS_TASK_CONTEXT_TYPE)
RETURN_NAMES = ("positive", "negative", "task_context")
RETURN_TYPES = ("CONDITIONING", "CONDITIONING", "STRING")
RETURN_NAMES = ("positive", "negative", "prompt_info")
DESCRIPTION = (
"Encode prompts into CONDITIONING tensors. Uses task_context when available to "
"dispatch between SD1.5, SDXL, SDXL Turbo, and FLUX text-encoding paths."
"Encode prompts into CONDITIONING tensors. Dispatches between SD1.5, SDXL, "
"SDXL Turbo, and FLUX text-encoding paths using native CLIP inputs."
)
@classmethod
@@ -142,21 +142,21 @@ class LLSSimplePromptEncode:
"positive_prompt": ("STRING", {"default": "", "multiline": True}),
"negative_prompt": ("STRING", {"default": "", "multiline": True}),
"clip_skip": (_CLIP_SKIP_CHOICES, {"default": -1}),
"model_family": (MODEL_FAMILY_CHOICES, {"default": "Auto"}),
},
"optional": {
"text_encoder": (LLS_TEXT_ENCODER_TYPE,),
"text_encoder": ("CLIP",),
"clip": ("CLIP",),
"task_context": (LLS_TASK_CONTEXT_TYPE,),
},
}
def encode(
self,
text_encoder=None,
positive_prompt: str = "",
negative_prompt: str = "",
clip_skip=-1,
task_context=None,
model_family: str = "Auto",
text_encoder=None,
clip=None,
):
encoder = text_encoder if text_encoder is not None else clip
@@ -166,11 +166,10 @@ class LLSSimplePromptEncode:
"or the legacy 'clip' input."
)
context = parse_task_context(task_context)
info = parse_model_info(context)
family = context.get("resolved_model_family") or info["family"]
family = resolve_model_family(model_family, clip=encoder)
defaults = get_family_defaults(family)
clip_skip = _normalize_clip_skip(clip_skip)
if not context.get("supports_clip_skip", True):
if is_flux_family(family):
clip_skip = -1
if clip_skip != -1:
@@ -189,13 +188,13 @@ class LLSSimplePromptEncode:
try:
if is_flux_family(family):
guidance = info.get("guidance", info.get("default_guidance"))
guidance = defaults.get("default_guidance")
positive = _encode_flux_prompt(encoder, positive_prompt, guidance)
negative = _encode_flux_prompt(encoder, "", guidance)
negative_mode = "ignored_for_flux"
elif is_sdxl_family(family):
width = int(info.get("default_width", 1024))
height = int(info.get("default_height", 1024))
width = int(defaults.get("default_width", 1024))
height = int(defaults.get("default_height", 1024))
positive = _encode_sdxl_prompt(encoder, positive_prompt, width, height)
negative = _encode_sdxl_prompt(encoder, negative_for_encode, width, height)
else:
@@ -205,20 +204,21 @@ class LLSSimplePromptEncode:
raise RuntimeError(f"[LLS] Failed to encode prompts for family {family}: {exc}") from exc
prompt_mode = "flux" if is_flux_family(family) else "sdxl" if is_sdxl_family(family) else "clip"
next_context = update_task_context(
context,
resolved_model_family=family,
text_encoder_type=info.get("text_encoder_type", "clip"),
positive_prompt=positive_prompt,
negative_prompt=negative_prompt,
prompt_mode=prompt_mode,
positive_prompt_length=len(positive_prompt),
negative_prompt_length=len(negative_prompt),
clip_skip=clip_skip,
negative_mode=negative_mode,
source="LLS Simple Prompt Encode",
prompt_info = info_to_json(
{
"model_family": family,
"text_encoder_type": defaults.get("text_encoder_type", "clip"),
"positive_prompt": positive_prompt,
"negative_prompt": negative_prompt,
"prompt_mode": prompt_mode,
"positive_prompt_length": len(positive_prompt),
"negative_prompt_length": len(negative_prompt),
"clip_skip": clip_skip,
"negative_mode": negative_mode,
"guidance": defaults.get("default_guidance") if is_flux_family(family) else None,
}
)
return (positive, negative, next_context)
return (positive, negative, prompt_info)
NODE_CLASS_MAPPINGS: dict[str, type] = {
docs/superpowers/plans/2026-05-20-contextless-routing-final-state.md
+89
View File
@@ -0,0 +1,89 @@
# Contextless Routing Final State 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:** Remove `task_context` and `model_info` execution ports from the core LLS nodes, infer family/task from `MODEL` / `CLIP` / `LATENT`, and keep old workflows compatible through standard ComfyUI ports plus local manual overrides.
**Architecture:** Store only lightweight `_lls_*` tags on native ComfyUI objects, centralize family/task inference in `utils/model_info.py`, and make each node emit JSON info strings instead of passing context objects. Keep the core node graph native: `MODEL`, `CLIP`, `VAE`, `LATENT`, `IMAGE`, and `CONDITIONING`.
**Tech Stack:** Python, ComfyUI custom nodes, `unittest`
---
### Task 1: Shared inference helpers
**Files:**
- Modify: `utils/model_info.py`
- Test: `tests/test_loader_prompt_refactor.py`
- [ ] **Step 1: Keep the failing spec as the red test**
Run: `python3 -B -m unittest tests.test_loader_prompt_refactor`
Expected: FAIL on loader / prompt / latent / sampler signatures
- [ ] **Step 2: Add shared helpers for object tagging and family resolution**
Implement helpers in `utils/model_info.py` for:
- safely tagging `model`, `clip`, `vae`
- resolving effective family from `model_family`, `model`, `clip`
- building JSON-safe info payloads for prompt / latent / sample / decode / upscale
- [ ] **Step 3: Re-run targeted inference tests**
Run: `python3 -B -m unittest tests.test_model_info_inference`
Expected: PASS
### Task 2: Core node API refactor
**Files:**
- Modify: `model_loader/nodes.py`
- Modify: `conditioning/nodes.py`
- Modify: `latent/nodes.py`
- Modify: `sampling/nodes.py`
- Modify: `image/nodes.py`
- Modify: `upscale/nodes.py`
- Modify: `utils/nodes.py`
- Test: `tests/test_loader_prompt_refactor.py`
- Test: `tests/test_conditioning_compat.py`
- Test: `tests/test_upscale_switcher.py`
- [ ] **Step 1: Refactor loader outputs and tagging**
Change `LLSSimpleCheckpointLoader` to return `(MODEL, CLIP, VAE, CLIP)`, remove optional `task_context`, and tag returned native objects with `_lls_family`, `_lls_model_name`, `_lls_text_encoder_type`, and `_lls_vae_name`.
- [ ] **Step 2: Refactor encode / latent / sample / image nodes**
Change each node to:
- accept `model_family="Auto"` as a local override
- infer family from `MODEL` / `CLIP`
- infer task mode from `LATENT["source"]`
- return JSON strings (`prompt_info`, `latent_info`, `sample_info`, `decode_info`)
- [ ] **Step 3: Refactor metadata and config nodes**
Update `LLSSaveImage` and `LLSGenerationConfig` to use native objects plus JSON info strings instead of `task_context/model_info`.
- [ ] **Step 4: Run focused tests**
Run: `python3 -B -m unittest tests.test_loader_prompt_refactor tests.test_conditioning_compat tests.test_upscale_switcher`
Expected: PASS
### Task 3: Registration cleanup and full verification
**Files:**
- Modify: `__init__.py`
- Test: `tests/test_loader_prompt_refactor.py`
- Test: `tests/`
- [ ] **Step 1: Unregister legacy task nodes**
Remove the `task` subpackage from `__init__.py` so `LLSTaskController` and `LLSTaskInspector` are not exposed in contextless mode.
- [ ] **Step 2: Run the full suite**
Run: `python3 -B -m unittest discover -s tests`
Expected: PASS
- [ ] **Step 3: Merge only after green**
If and only if the full suite is green, merge `refactor/contextless-routing` back into `main` and remove the project-local worktree.
image/nodes.py
+122 -90
View File
@@ -11,11 +11,17 @@ from datetime import datetime
import importlib
from ..utils.model_info import (
MODEL_FAMILY_CHOICES,
canonicalize_family,
get_family_defaults,
get_latent_spec,
info_to_json,
parse_jsonish_info,
parse_model_info,
resolve_model_family,
resolve_model_name,
resolve_text_encoder_names,
resolve_vae_name,
)
from ..utils.task_context import LLS_TASK_CONTEXT_TYPE, parse_task_context, task_context_to_json, update_task_context
try:
@@ -109,6 +115,30 @@ def _resize_image_to(image, target_width: int, target_height: int, resize_mode:
raise RuntimeError(f"[LLS] Failed to resize IMAGE for VAE encode: {exc}") from exc
def _infer_family_from_latent_or_vae(samples, vae=None) -> str:
tagged_family = getattr(vae, "_lls_family", None)
if tagged_family and tagged_family not in {"Auto", "auto", ""}:
return canonicalize_family(tagged_family)
if isinstance(samples, dict):
try:
ratio = int(samples.get("downscale_ratio_spacial", 0))
if ratio >= 16:
return "FLUX_DEV"
except Exception:
pass
latent_tensor = samples.get("samples")
latent_shape = tuple(getattr(latent_tensor, "shape", ()))
if len(latent_shape) >= 2:
try:
if int(latent_shape[1]) >= 16:
return "FLUX_DEV"
except Exception:
pass
return "SD1.5"
class LLSSimpleVAEEncode:
"""
把 IMAGE 编码为 LATENT，供 img2img 直接接入现有 KSampler。
@@ -117,8 +147,8 @@ class LLSSimpleVAEEncode:
CATEGORY = "LLS-node"
FUNCTION = "encode"
RETURN_TYPES = ("LATENT", "INT", "INT", LLS_TASK_CONTEXT_TYPE)
RETURN_NAMES = ("latent", "width", "height", "task_context")
RETURN_TYPES = ("LATENT", "INT", "INT", "STRING")
RETURN_NAMES = ("latent", "width", "height", "latent_info")
DESCRIPTION = "Encode an IMAGE into a LATENT tensor for img2img workflows."
@classmethod
@@ -131,9 +161,11 @@ class LLSSimpleVAEEncode:
"size_source": (_SIZE_SOURCES, {"default": "input_image"}),
"width": ("INT", {"default": 512, "min": 64, "max": 4096, "step": 8}),
"height": ("INT", {"default": 512, "min": 64, "max": 4096, "step": 8}),
"model_family": (MODEL_FAMILY_CHOICES, {"default": "Auto"}),
},
"optional": {
"task_context": (LLS_TASK_CONTEXT_TYPE,),
"model": ("MODEL",),
"clip": ("CLIP",),
},
}
@@ -145,7 +177,9 @@ class LLSSimpleVAEEncode:
size_source: str,
width: int,
height: int,
task_context=None,
model_family: str = "Auto",
model=None,
clip=None,
):
if image is None:
raise RuntimeError("[LLS] IMAGE input is required for LLS Simple VAE Encode.")
@@ -154,9 +188,9 @@ class LLSSimpleVAEEncode:
"[LLS] Missing VAE. Connect the Loader VAE output or choose an external VAE in the loader."
)
context = parse_task_context(task_context)
info = parse_model_info(context)
latent_spec = get_latent_spec(info)
family = resolve_model_family(model_family, model=model, clip=clip)
defaults = get_family_defaults(family)
latent_spec = get_latent_spec(defaults)
alignment = max(8, int(latent_spec["downscale_ratio"]))
batch_size, source_width, source_height = _get_image_dimensions(image)
@@ -167,8 +201,8 @@ class LLSSimpleVAEEncode:
source_width=source_width,
source_height=source_height,
model_info={
"default_width": context.get("recommended_width", info["default_width"]),
"default_height": context.get("recommended_height", info["default_height"]),
"default_width": defaults["default_width"],
"default_height": defaults["default_height"],
},
)
requested_width = _round_to_multiple(requested_width, alignment)
@@ -209,24 +243,24 @@ class LLSSimpleVAEEncode:
"downscale_ratio_spacial": int(latent_spec["downscale_ratio"]),
"source": "image_encode",
}
next_context = update_task_context(
context,
task_mode="img2img" if context.get("task_mode") in (None, "", "txt2img") else context.get("task_mode"),
latent_source="image_encode",
input_image_width=source_width,
input_image_height=source_height,
final_width=final_width,
final_height=final_height,
batch_size=batch_size,
resize_mode=resize_mode,
size_source=size_source,
vae_source=context.get("vae_source") or info.get("vae_source", "auto"),
vae_name=context.get("vae_name") or info.get("vae_name", "unknown"),
latent_channels=latent_channels,
downscale_ratio=int(latent_spec["downscale_ratio"]),
source="LLS Simple VAE Encode",
latent_info = info_to_json(
{
"model_family": family,
"task_mode": "img2img",
"latent_source": "image_encode",
"input_image_width": source_width,
"input_image_height": source_height,
"width": final_width,
"height": final_height,
"batch_size": batch_size,
"resize_mode": resize_mode,
"size_source": size_source,
"vae_name": resolve_vae_name(vae, fallback="unknown"),
"latent_channels": latent_channels,
"downscale_ratio": int(latent_spec["downscale_ratio"]),
}
)
return (latent_payload, final_width, final_height, next_context)
return (latent_payload, final_width, final_height, latent_info)
class LLSSimpleVAEDecode:
@@ -237,8 +271,8 @@ class LLSSimpleVAEDecode:
CATEGORY = "LLS/Image"
FUNCTION = "decode"
RETURN_TYPES = ("IMAGE", LLS_TASK_CONTEXT_TYPE)
RETURN_NAMES = ("image", "task_context")
RETURN_TYPES = ("IMAGE", "STRING")
RETURN_NAMES = ("image", "decode_info")
DESCRIPTION = "Decode a LATENT tensor into a pixel IMAGE using the VAE."
@classmethod
@@ -248,18 +282,13 @@ class LLSSimpleVAEDecode:
"samples": ("LATENT",),
"vae": ("VAE",),
},
"optional": {
"task_context": (LLS_TASK_CONTEXT_TYPE,),
},
}
def decode(self, samples, vae, task_context=None):
context = parse_task_context(task_context)
info = parse_model_info(context)
family = context.get("resolved_model_family") or info["family"]
def decode(self, samples, vae):
family = _infer_family_from_latent_or_vae(samples, vae)
if vae is None:
if info.get("is_flux"):
if family.startswith("FLUX"):
raise RuntimeError(
"[LLS] Missing FLUX AE/VAE. Place ae.safetensors in ComfyUI/models/vae/ "
"or set Loader.vae_source to 'external'."
@@ -288,22 +317,21 @@ class LLSSimpleVAEDecode:
if len(getattr(image, "shape", ())) == 5:
image = image.reshape(-1, image.shape[-3], image.shape[-2], image.shape[-1])
latent_spec = get_latent_spec(info)
downscale_ratio = int(samples.get("downscale_ratio_spacial", latent_spec["downscale_ratio"]))
defaults = get_family_defaults(family)
downscale_ratio = int(samples.get("downscale_ratio_spacial", defaults["downscale_ratio"]))
height = latent_tensor.shape[2] * downscale_ratio
width = latent_tensor.shape[3] * downscale_ratio
next_context = update_task_context(
context,
resolved_model_family=family,
final_width=width,
final_height=height,
batch_size=latent_tensor.shape[0],
vae_name=context.get("vae_name") or info.get("vae_name", "unknown"),
decoded=True,
decode_stage="vae_decode",
source="LLS Simple VAE Decode",
decode_info = info_to_json(
{
"model_family": family,
"width": width,
"height": height,
"batch_size": latent_tensor.shape[0],
"vae_name": resolve_vae_name(vae, fallback="unknown"),
"decode_stage": "vae_decode",
}
)
return (image, next_context)
return (image, decode_info)
class LLSSaveImage:
@@ -326,8 +354,9 @@ class LLSSaveImage:
"save_metadata": ("BOOLEAN", {"default": True}),
},
"optional": {
"task_context": (LLS_TASK_CONTEXT_TYPE,),
"model_info": (LLS_TASK_CONTEXT_TYPE,),
"model": ("MODEL",),
"clip": ("CLIP",),
"vae": ("VAE",),
"prompt_info": ("STRING", {"forceInput": True}),
"latent_info": ("STRING", {"forceInput": True}),
"sample_info": ("STRING", {"forceInput": True}),
@@ -343,53 +372,55 @@ class LLSSaveImage:
def _build_metadata(
self,
image,
task_context=None,
model_info=None,
model=None,
clip=None,
vae=None,
prompt_info: str | None = None,
latent_info: str | None = None,
sample_info: str | None = None,
decode_info: str | None = None,
upscale_info: str | None = None,
) -> dict:
context = parse_task_context(task_context or model_info)
model = parse_model_info(context)
prompt = parse_jsonish_info(prompt_info)
latent = parse_jsonish_info(latent_info)
sample = parse_jsonish_info(sample_info)
decode = parse_jsonish_info(decode_info)
upscale = parse_jsonish_info(upscale_info)
text_encoder_name = context.get("text_encoder_name") or model.get("text_encoder_name")
if context.get("text_encoder_name_2") or model.get("text_encoder_name_2"):
joined = [
name
for name in (
context.get("text_encoder_name_1") or model.get("text_encoder_name_1"),
context.get("text_encoder_name_2") or model.get("text_encoder_name_2"),
)
if name
]
text_encoder_name = ", ".join(joined)
family = prompt.get("model_family") or latent.get("model_family") or decode.get("model_family")
if not family:
family = resolve_model_family("Auto", model=model, clip=clip)
text_encoder_name = resolve_text_encoder_names(clip).get("text_encoder_name", "")
checkpoint_name = resolve_model_name(
model=model,
clip=clip,
fallback=prompt.get("checkpoint_name") or latent.get("checkpoint_name") or decode.get("checkpoint_name") or "",
)
vae_name = resolve_vae_name(
vae,
fallback=decode.get("vae_name") or latent.get("vae_name") or "",
)
metadata = {
"positive_prompt": context.get("positive_prompt") or prompt.get("positive_prompt", ""),
"negative_prompt": context.get("negative_prompt") or prompt.get("negative_prompt", ""),
"seed": context.get("seed", sample.get("seed")),
"model_family": context.get("resolved_model_family") or context.get("model_family") or model.get("family"),
"checkpoint_name": context.get("checkpoint_name") or model.get("checkpoint_name"),
"vae_name": context.get("vae_name") or decode.get("vae_name") or model.get("vae_name"),
"positive_prompt": prompt.get("positive_prompt", ""),
"negative_prompt": prompt.get("negative_prompt", ""),
"seed": sample.get("seed"),
"model_family": family,
"checkpoint_name": checkpoint_name,
"vae_name": vae_name,
"text_encoder_name": text_encoder_name,
"steps": context.get("steps", sample.get("steps")),
"cfg": context.get("cfg", sample.get("cfg")),
"guidance": context.get("guidance", sample.get("guidance")),
"sampler_name": context.get("sampler_name", sample.get("sampler_name")),
"scheduler": context.get("scheduler", sample.get("scheduler")),
"denoise": context.get("denoise", sample.get("denoise")),
"width": context.get("final_width") or decode.get("width") or latent.get("width") or image.shape[2],
"height": context.get("final_height") or decode.get("height") or latent.get("height") or image.shape[1],
"batch_size": context.get("batch_size") or latent.get("batch_size") or decode.get("batch_size") or image.shape[0],
"upscale_mode": context.get("upscale_mode") or upscale.get("mode"),
"scale": context.get("upscale_scale") or upscale.get("scale"),
"steps": sample.get("steps"),
"cfg": sample.get("cfg"),
"guidance": sample.get("guidance"),
"sampler_name": sample.get("sampler_name"),
"scheduler": sample.get("scheduler"),
"denoise": sample.get("denoise"),
"width": decode.get("width") or latent.get("width") or image.shape[2],
"height": decode.get("height") or latent.get("height") or image.shape[1],
"batch_size": decode.get("batch_size") or latent.get("batch_size") or image.shape[0],
"upscale_mode": upscale.get("mode") or upscale.get("upscale_mode"),
"scale": upscale.get("scale") or upscale.get("upscale_scale"),
"timestamp": datetime.now().isoformat(timespec="seconds"),
}
return metadata
@@ -399,8 +430,9 @@ class LLSSaveImage:
image,
filename_prefix: str = "LLS",
save_metadata: bool = True,
task_context=None,
model_info=None,
model=None,
clip=None,
vae=None,
prompt_info: str | None = None,
latent_info: str | None = None,
sample_info: str | None = None,
@@ -417,15 +449,15 @@ class LLSSaveImage:
if save_metadata:
merged_extra_pnginfo["lls_metadata"] = self._build_metadata(
image=image,
task_context=task_context,
model_info=model_info,
model=model,
clip=clip,
vae=vae,
prompt_info=prompt_info,
latent_info=latent_info,
sample_info=sample_info,
decode_info=decode_info,
upscale_info=upscale_info,
)
merged_extra_pnginfo["lls_task_context"] = task_context_to_json(task_context or model_info)
return saver.save_images(
image,
filename_prefix=filename_prefix,
latent/nodes.py
+30 -27
View File
@@ -21,11 +21,13 @@ except Exception:
model_management = None
from ..utils.model_info import (
MODEL_FAMILY_CHOICES,
SIZE_PRESET_AUTO,
get_family_defaults,
get_latent_spec,
parse_model_info,
info_to_json,
resolve_model_family,
)
from ..utils.task_context import LLS_TASK_CONTEXT_TYPE, parse_task_context, update_task_context
_SIZE_PRESETS = [
@@ -53,17 +55,17 @@ class LLSSimpleEmptyLatent:
"""
生成空白 Latent，作为 txt2img 推理起点。
- `Family Default` 时按照 task_context 推荐尺寸自动取默认值。
- `Family Default` 时按照模型家族默认尺寸自动取值。
- `Custom` 时才使用手填 width / height。
"""
CATEGORY = "LLS/Latent"
FUNCTION = "create_empty_latent"
RETURN_TYPES = ("LATENT", "INT", "INT", LLS_TASK_CONTEXT_TYPE)
RETURN_NAMES = ("latent", "width", "height", "task_context")
RETURN_TYPES = ("LATENT", "INT", "INT", "STRING")
RETURN_NAMES = ("latent", "width", "height", "latent_info")
DESCRIPTION = (
"Create an empty latent image as the starting point for txt2img. "
"Uses task_context recommendations when size_preset is 'Family Default'."
"Uses inferred family defaults when size_preset is 'Family Default'."
)
@classmethod
@@ -74,9 +76,10 @@ class LLSSimpleEmptyLatent:
"width": ("INT", {"default": 512, "min": 64, "max": 8192, "step": 8}),
"height": ("INT", {"default": 512, "min": 64, "max": 8192, "step": 8}),
"batch_size": ("INT", {"default": 1, "min": 1, "max": 64}),
"model_family": (MODEL_FAMILY_CHOICES, {"default": "Auto"}),
},
"optional": {
"task_context": (LLS_TASK_CONTEXT_TYPE,),
"model": ("MODEL",),
},
}
@@ -86,22 +89,23 @@ class LLSSimpleEmptyLatent:
width: int,
height: int,
batch_size: int,
task_context=None,
model_family: str = "Auto",
model=None,
):
if torch is None:
raise RuntimeError(
"[LLS] PyTorch is not available. Make sure this node runs inside a ComfyUI environment."
) from _TORCH_ERR
context = parse_task_context(task_context)
info = parse_model_info(context)
latent_spec = get_latent_spec(info)
family = resolve_model_family(model_family, model=model)
defaults = get_family_defaults(family)
latent_spec = get_latent_spec(defaults)
latent_channels = latent_spec["latent_channels"]
downscale_ratio = latent_spec["downscale_ratio"]
if size_preset == SIZE_PRESET_AUTO:
width = int(context.get("recommended_width", info["default_width"]))
height = int(context.get("recommended_height", info["default_height"]))
width = int(defaults["default_width"])
height = int(defaults["default_height"])
elif size_preset != "Custom":
try:
width, height = [int(part) for part in size_preset.split("x", 1)]
@@ -129,25 +133,24 @@ class LLSSimpleEmptyLatent:
dtype=dtype,
)
next_context = update_task_context(
context,
resolved_model_family=context.get("resolved_model_family") or info["family"],
task_mode="txt2img" if context.get("task_mode") in (None, "", "img2img") else context.get("task_mode"),
latent_source="empty_latent",
final_width=width,
final_height=height,
batch_size=batch_size,
latent_channels=latent_channels,
downscale_ratio=downscale_ratio,
size_preset=size_preset,
source="LLS Simple Empty Latent",
latent_info = info_to_json(
{
"model_family": family,
"task_mode": "txt2img",
"latent_source": "empty_latent",
"width": width,
"height": height,
"batch_size": batch_size,
"latent_channels": latent_channels,
"downscale_ratio": downscale_ratio,
"size_preset": size_preset,
}
)
return (
{"samples": latent, "downscale_ratio_spacial": downscale_ratio, "source": "empty_latent"},
width,
height,
next_context,
latent_info,
)
model_loader/nodes.py
+38 -93
View File
@@ -35,18 +35,14 @@ else:
_CORE_NODES_ERR = None
from ..utils.model_info import (
LLS_TEXT_ENCODER_TYPE,
MODEL_FAMILY_CHOICES,
build_model_info,
canonicalize_family,
family_requires_guidance,
get_family_defaults,
get_latent_spec,
infer_family_from_name,
is_flux_family,
is_sdxl_family,
tag_lls_object,
)
from ..utils.task_context import LLS_TASK_CONTEXT_TYPE, merge_task_context, parse_task_context, update_task_context
AUTO_PLACEHOLDER = "(auto)"
@@ -355,20 +351,21 @@ class LLSSimpleCheckpointLoader:
"""
统一封装的基础模型加载器。
兼容输出结构为 model / clip / vae / task_context / text_encoder。
输出结构固定为 model / clip / vae / text_encoder。
- `clip` 保留旧工作流需要的原生 CLIP 端口类型和位置。
- `text_encoder` 为新接口别名，内部承载同一个 ComfyUI 文本编码对象。
- `text_encoder` 是同一个原生 CLIP 对象的兼容别名。
- 轻量级模型信息写入原生对象 `_lls_*` 属性，不再额外输出 task_context。
"""
CATEGORY = "LLS/Model Loader"
FUNCTION = "load_checkpoint"
RETURN_TYPES = ("MODEL", "CLIP", "VAE", LLS_TASK_CONTEXT_TYPE, LLS_TEXT_ENCODER_TYPE)
RETURN_NAMES = ("model", "clip", "vae", "task_context", "text_encoder")
RETURN_TYPES = ("MODEL", "CLIP", "VAE", "CLIP")
RETURN_NAMES = ("model", "clip", "vae", "text_encoder")
DESCRIPTION = (
"Load SD1.5, SDXL, SDXL Turbo, and FLUX families with family-aware resource dispatch. "
"Supports FLUX checkpoint-all-in-one mode and separated diffusion/text-encoder/VAE mode. "
"Preserves the legacy CLIP output for old workflows and writes the resolved loader state "
"back into a unified task_context object."
"Preserves the legacy CLIP output for old workflows and writes lightweight metadata "
"into native MODEL/CLIP/VAE objects."
)
@classmethod
@@ -384,9 +381,6 @@ class LLSSimpleCheckpointLoader:
"external_text_encoder_1": (_get_text_encoder_choices(), {"default": AUTO_PLACEHOLDER}),
"external_text_encoder_2": (_get_text_encoder_choices(), {"default": AUTO_PLACEHOLDER}),
},
"optional": {
"task_context": (LLS_TASK_CONTEXT_TYPE,),
},
}
def load_checkpoint(
@@ -399,7 +393,6 @@ class LLSSimpleCheckpointLoader:
external_vae_name: str,
external_text_encoder_1: str,
external_text_encoder_2: str,
task_context=None,
):
if folder_paths is None:
raise RuntimeError(
@@ -410,20 +403,10 @@ class LLSSimpleCheckpointLoader:
"[LLS] comfy.sd is not available. Make sure this node runs inside ComfyUI."
) from _COMFY_SD_ERR
context = parse_task_context(task_context)
requested_family = str(context.get("model_family") or "auto")
if model_family != "Auto":
requested_family = model_family
if model_family == "Auto":
fallback_family = context.get("resolved_model_family") or context.get("family") or context.get("model_family") or "SD1.5"
family = infer_family_from_name(ckpt_name, fallback_family)
family = infer_family_from_name(ckpt_name, "SD1.5")
else:
family = canonicalize_family(requested_family)
if vae_source == "auto" and context.get("use_external_vae"):
vae_source = "external"
if text_encoder_source == "auto" and context.get("use_external_text_encoder"):
text_encoder_source = "external"
family = canonicalize_family(model_family)
model_source, model_path = _resolve_model_path(ckpt_name, family)
model, embedded_text_encoder, embedded_vae = _load_model(model_source, model_path)
@@ -449,79 +432,41 @@ class LLSSimpleCheckpointLoader:
)
defaults = get_family_defaults(family)
latent_spec = get_latent_spec(defaults)
model_info = build_model_info(
resolved_text_encoder_name = resolved_te_1 or ("embedded" if text_encoder is not None else None)
resolved_text_encoder_name_1 = resolved_te_1 or ("embedded" if text_encoder is not None and not resolved_te_2 else None)
resolved_text_encoder_name_2 = resolved_te_2
if resolved_te_1 and resolved_te_2:
resolved_text_encoder_name = ", ".join([resolved_te_1, resolved_te_2])
resolved_vae_label = resolved_vae_name or ("embedded" if vae is not None else None)
tag_lls_object(
model,
family=family,
checkpoint_name=ckpt_name,
model_name=ckpt_name,
checkpoint_name=ckpt_name,
load_mode=load_mode,
vae_source=resolved_vae_source,
text_encoder_source=resolved_text_encoder_source,
)
tag_lls_object(
text_encoder,
family=family,
model_name=ckpt_name,
checkpoint_name=ckpt_name,
text_encoder_type=defaults["text_encoder_type"],
has_embedded_vae=embedded_vae is not None,
has_embedded_text_encoder=embedded_text_encoder is not None,
required_text_encoders=defaults["required_text_encoders"],
required_vae=defaults["required_vae"],
is_turbo=defaults["is_turbo"],
is_flux=defaults["is_flux"],
default_width=defaults["default_width"],
default_height=defaults["default_height"],
default_steps=defaults["default_steps"],
default_cfg=defaults["default_cfg"],
default_guidance=defaults["default_guidance"],
default_sampler=defaults["default_sampler"],
default_scheduler=defaults["default_scheduler"],
default_denoise=defaults["default_denoise"],
guidance_embed=family_requires_guidance(family),
latent_channels=latent_spec["latent_channels"],
downscale_ratio=latent_spec["downscale_ratio"],
model_source=model_source,
vae_name=resolved_vae_name or ("embedded" if embedded_vae is not None else None),
text_encoder_name=resolved_te_1,
text_encoder_name_1=resolved_te_1,
text_encoder_name_2=resolved_te_2,
)
model_caps = {
"supports_img2img": vae is not None,
"supports_inpaint": vae is not None,
"supports_controlnet": family in {"SD1.5", "SDXL", "SDXL_TURBO"},
"supports_clip_skip": not is_flux_family(family),
"supports_flux_guidance": is_flux_family(family),
"vae_type": "ae" if is_flux_family(family) else "vae",
"text_encoder_type": model_info["text_encoder_type"],
"latent_channels": latent_spec["latent_channels"],
"downscale_ratio": latent_spec["downscale_ratio"],
"recommended_width": model_info["default_width"],
"recommended_height": model_info["default_height"],
"recommended_steps": model_info["default_steps"],
"recommended_cfg": model_info["default_cfg"],
"recommended_denoise": model_info["default_denoise"],
"recommended_guidance": model_info["default_guidance"],
"recommended_sampler": model_info["default_sampler"],
"recommended_scheduler": model_info["default_scheduler"],
"source": "LLS Simple Checkpoint Loader",
}
next_context = merge_task_context(context, model_info)
next_context = update_task_context(
next_context,
model_family=requested_family,
resolved_model_family=family,
checkpoint_name=ckpt_name,
model_name=ckpt_name,
load_mode=load_mode,
vae_source=resolved_vae_source,
text_encoder_source=resolved_text_encoder_source,
vae_name=resolved_vae_name or ("embedded" if embedded_vae is not None else None),
text_encoder_name=resolved_te_1,
text_encoder_name_1=resolved_te_1,
text_encoder_name_2=resolved_te_2,
use_external_vae=resolved_vae_source == "external",
use_external_text_encoder=resolved_text_encoder_source == "external",
**model_caps,
text_encoder_name=resolved_text_encoder_name,
text_encoder_name_1=resolved_text_encoder_name_1,
text_encoder_name_2=resolved_text_encoder_name_2,
)
tag_lls_object(
vae,
family=family,
model_name=ckpt_name,
checkpoint_name=ckpt_name,
vae_name=resolved_vae_label,
vae_source=resolved_vae_source,
)
return (model, text_encoder, vae, next_context, text_encoder)
return (model, text_encoder, vae, text_encoder)
NODE_CLASS_MAPPINGS: dict[str, type] = {
sampling/nodes.py
+22 -50
View File
@@ -11,12 +11,14 @@ import random
from ..utils.model_info import (
FAMILY_DEFAULT_PRESET,
MODEL_FAMILY_CHOICES,
get_family_defaults,
get_sampling_preset,
infer_task_mode_from_latent,
info_to_json,
is_flux_family,
parse_model_info,
resolve_model_family,
)
from ..utils.task_context import LLS_TASK_CONTEXT_TYPE, parse_task_context, update_task_context
# ---------- 防御性导入 ----------
@@ -197,7 +199,6 @@ def _common_ksampler(
)
out = latent.copy()
out.pop("downscale_ratio_spacial", None)
out["samples"] = sampled
return out
@@ -213,8 +214,8 @@ class LLSSimpleKSampler:
CATEGORY = "LLS/Sampling"
FUNCTION = "sample"
RETURN_TYPES = ("LATENT", "STRING", LLS_TASK_CONTEXT_TYPE)
RETURN_NAMES = ("latent", "sample_info", "task_context")
RETURN_TYPES = ("LATENT", "STRING")
RETURN_NAMES = ("latent", "sample_info")
DESCRIPTION = (
"Basic KSampler node. Reuses ComfyUI's native sampling pipeline. "
"quality_preset overrides steps/cfg/denoise when not 'Manual'. "
@@ -229,6 +230,7 @@ class LLSSimpleKSampler:
"positive": ("CONDITIONING",),
"negative": ("CONDITIONING",),
"latent_image": ("LATENT",),
"model_family": (MODEL_FAMILY_CHOICES, {"default": "Auto"}),
"quality_preset": (_QUALITY_PRESETS, {"default": FAMILY_DEFAULT_PRESET}),
"seed": ("INT", {"default": -1, "min": -1, "max": 0xFFFFFFFFFFFFFFFF}),
"steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
@@ -241,9 +243,6 @@ class LLSSimpleKSampler:
{"default": 3.5, "widgetType": "FLOAT", "min": 0.0, "max": 100.0, "step": 0.1, "round": 0.1},
),
},
"optional": {
"task_context": (LLS_TASK_CONTEXT_TYPE,),
},
}
@classmethod
@@ -265,6 +264,7 @@ class LLSSimpleKSampler:
positive,
negative,
latent_image,
model_family: str,
quality_preset: str,
seed: int,
steps: int,
@@ -273,7 +273,6 @@ class LLSSimpleKSampler:
scheduler: str,
denoise: float,
flux_guidance,
task_context=None,
):
if comfy_sample is None:
raise RuntimeError(
@@ -286,30 +285,21 @@ class LLSSimpleKSampler:
"Make sure this node runs inside a ComfyUI environment."
) from _COMFY_SAMPLERS_ERR
context = parse_task_context(task_context)
info = parse_model_info(context)
family = context.get("resolved_model_family") or info["family"]
latent_source = latent_image.get("source") or context.get("latent_source")
default_flux_guidance = context.get("recommended_guidance")
if default_flux_guidance is None:
default_flux_guidance = info.get("guidance", info.get("default_guidance"))
effective_task_mode = context.get("task_mode") or "txt2img"
if latent_source == "empty_latent":
effective_task_mode = "txt2img"
elif latent_source == "image_encode":
effective_task_mode = "img2img"
family = resolve_model_family(model_family, model=model)
defaults = get_family_defaults(family)
default_flux_guidance = defaults.get("default_guidance")
effective_task_mode = infer_task_mode_from_latent(latent_image)
if quality_preset == FAMILY_DEFAULT_PRESET:
steps = int(context.get("recommended_steps", steps))
cfg = float(context.get("recommended_cfg", cfg))
sampler_name = str(context.get("recommended_sampler", sampler_name))
scheduler = str(context.get("recommended_scheduler", scheduler))
denoise = float(context.get("recommended_denoise", denoise))
context_guidance = default_flux_guidance
if context_guidance is not None:
flux_guidance = float(context_guidance)
steps = int(defaults["default_steps"])
cfg = float(defaults["default_cfg"])
sampler_name = str(defaults["default_sampler"])
scheduler = str(defaults["default_scheduler"])
denoise = float(defaults["default_denoise"])
if default_flux_guidance is not None:
flux_guidance = float(default_flux_guidance)
else:
preset = get_sampling_preset(info, quality_preset)
preset = get_sampling_preset(defaults, quality_preset)
if preset is not None:
steps = int(preset["steps"])
cfg = float(preset["cfg"])
@@ -321,7 +311,7 @@ class LLSSimpleKSampler:
flux_guidance = _normalize_flux_guidance(flux_guidance, fallback=default_flux_guidance)
if is_flux_family(family) and context.get("supports_flux_guidance", True):
if is_flux_family(family):
guidance_value = flux_guidance
else:
guidance_value = None
@@ -333,9 +323,6 @@ class LLSSimpleKSampler:
except Exception:
pass
if quality_preset not in {"Manual", FAMILY_DEFAULT_PRESET} and effective_task_mode == "img2img":
denoise = float(context.get("recommended_denoise", denoise))
# seed = -1 时随机生成
actual_seed = seed
if seed == -1:
@@ -374,22 +361,7 @@ class LLSSimpleKSampler:
"task_mode": effective_task_mode,
}
)
next_context = update_task_context(
context,
resolved_model_family=family,
task_mode=effective_task_mode,
sampler_name=sampler_name,
scheduler=scheduler,
steps=steps,
cfg=cfg,
seed=actual_seed,
denoise=denoise,
guidance=guidance_value,
sampled=True,
source="LLS Simple KSampler",
)
return (result_latent, sample_info, next_context)
return (result_latent, sample_info)
# ---------- 注册表 ----------
tests/test_conditioning_compat.py
+19 -13
View File
@@ -1,4 +1,5 @@
import importlib.util
import json
import pathlib
import sys
import unittest
@@ -67,11 +68,12 @@ class TestConditioningCompatibility(unittest.TestCase):
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSSimplePromptEncode"]
node = node_cls()
positive, negative, task_context = node.encode(
LegacyClipStub(),
"a cat",
"low quality",
-2,
positive, negative, prompt_info = node.encode(
clip=LegacyClipStub(),
positive_prompt="a cat",
negative_prompt="low quality",
clip_skip=-2,
model_family="Auto",
)
self.assertEqual(
@@ -82,19 +84,22 @@ class TestConditioningCompatibility(unittest.TestCase):
negative,
[["cond::low quality", {"pooled_output": "pooled::low quality"}]],
)
self.assertEqual(task_context["clip_skip"], -2)
self.assertEqual(task_context["prompt_mode"], "clip")
payload = json.loads(prompt_info)
self.assertEqual(payload["clip_skip"], -2)
self.assertEqual(payload["prompt_mode"], "clip")
self.assertEqual(payload["model_family"], "SD1.5")
def test_simple_prompt_encode_normalizes_none_clip_skip(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSSimplePromptEncode"]
node = node_cls()
positive, negative, task_context = node.encode(
LegacyClipStub(),
"a cat",
"low quality",
None,
positive, negative, prompt_info = node.encode(
clip=LegacyClipStub(),
positive_prompt="a cat",
negative_prompt="low quality",
clip_skip=None,
model_family="Auto",
)
self.assertEqual(
@@ -105,7 +110,8 @@ class TestConditioningCompatibility(unittest.TestCase):
negative,
[["cond::low quality", {"pooled_output": "pooled::low quality"}]],
)
self.assertEqual(task_context["clip_skip"], -1)
payload = json.loads(prompt_info)
self.assertEqual(payload["clip_skip"], -1)
def test_universal_backend_falls_back_for_legacy_clip(self):
load_plugin_package()
tests/test_loader_prompt_refactor.py
+203 -282
View File
@@ -22,6 +22,14 @@ def load_plugin_package():
return module
class TaggedValue:
def __init__(self, label):
self.label = label
def __repr__(self):
return self.label
class FolderPathsStub:
def __init__(self):
self._files = {
@@ -41,6 +49,7 @@ class FolderPathsStub:
"clip_g.safetensors",
"t5xxl_fp16.safetensors",
],
"upscale_models": [],
}
def get_filename_list(self, category):
@@ -69,15 +78,15 @@ class ComfySDStub:
embedding_directory=None,
):
if ckpt_path.endswith("sd15.safetensors"):
return ("MODEL::SD15", "CLIP::SD15", "VAE::SD15")
return (TaggedValue("MODEL::SD15"), TaggedValue("CLIP::SD15"), TaggedValue("VAE::SD15"))
if ckpt_path.endswith("sdxl_turbo.safetensors"):
return ("MODEL::SDXL_TURBO", None, None)
return (TaggedValue("MODEL::SDXL_TURBO"), None, None)
if ckpt_path.endswith("flux-dev-checkpoint.safetensors"):
return ("MODEL::FLUX_DEV", None, None)
return (TaggedValue("MODEL::FLUX_DEV"), None, None)
raise AssertionError(f"unexpected checkpoint path: {ckpt_path}")
def load_diffusion_model(self, unet_path, model_options=None, disable_dynamic=False):
return f"MODEL::{pathlib.Path(unet_path).name}"
return TaggedValue(f"MODEL::{pathlib.Path(unet_path).name}")
class CoreNodesStub:
@@ -87,15 +96,15 @@ class CoreNodesStub:
return ["ae.safetensors"]
def load_vae(self, vae_name):
return (f"VAE::{vae_name}",)
return (TaggedValue(f"VAE::{vae_name}"),)
class CLIPLoader:
def load_clip(self, clip_name, type="stable_diffusion", device="default"):
return (f"CLIP::{type}::{clip_name}",)
return (TaggedValue(f"CLIP::{type}::{clip_name}"),)
class DualCLIPLoader:
def load_clip(self, clip_name1, clip_name2, type, device="default"):
return (f"CLIP::{type}::{clip_name1}+{clip_name2}",)
return (TaggedValue(f"CLIP::{type}::{clip_name1}+{clip_name2}"),)
class SaveImage:
def save_images(self, images, filename_prefix="ComfyUI", prompt=None, extra_pnginfo=None):
@@ -115,12 +124,13 @@ class CoreNodesStub:
class RecordingClip:
def __init__(self, calls=None):
def __init__(self, token_keys, calls=None):
self.token_keys = tuple(token_keys)
self.calls = [] if calls is None else calls
self.layer = None
def clone(self):
cloned = RecordingClip(self.calls)
cloned = RecordingClip(self.token_keys, self.calls)
cloned.layer = self.layer
return cloned
@@ -128,12 +138,10 @@ class RecordingClip:
self.layer = layer
def tokenize(self, text):
return {
"text": text,
"g": [f"g::{text}"],
"l": [f"l::{text}"],
"t5xxl": [f"t5::{text}"],
}
tokens = {"text": text}
for key in self.token_keys:
tokens[key] = [f"{key}::{text}"]
return tokens
def encode_from_tokens_scheduled(self, tokens, add_dict=None):
self.calls.append(
@@ -147,10 +155,11 @@ class RecordingClip:
class FakeVAE:
def __init__(self, encode_channels=4, encode_ratio=8):
def __init__(self, encode_channels=4, encode_ratio=8, vae_name="vae.safetensors"):
self.encode_channels = encode_channels
self.encode_ratio = encode_ratio
self.encoded_shapes = []
self._lls_vae_name = vae_name
def encode(self, pixels):
self.encoded_shapes.append(tuple(pixels.shape))
@@ -220,62 +229,24 @@ class FakeComfyUtils:
class TestLoaderPromptRefactor(unittest.TestCase):
def test_task_controller_generates_context_with_recommended_defaults(self):
def test_task_nodes_are_not_registered_in_contextless_mode(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSTaskController"]
node = node_cls()
required = node_cls.INPUT_TYPES()["required"]
self.assertEqual(node_cls.CATEGORY, "LLS-node")
self.assertEqual(node_cls.RETURN_TYPES, ("LLS_TASK_CONTEXT",))
self.assertEqual(
tuple(required.keys()),
(
"task_mode",
"model_family",
"workflow_preset",
"quality_preset",
"enable_upscale",
"enable_controlnet",
"enable_reference",
"use_external_vae",
"use_external_text_encoder",
),
)
self.assertNotIn("LLSTaskController", plugin.NODE_CLASS_MAPPINGS)
self.assertNotIn("LLSTaskInspector", plugin.NODE_CLASS_MAPPINGS)
(task_context,) = node.execute(
"img2img",
"SDXL",
"standard",
"balanced",
True,
False,
False,
False,
False,
)
self.assertEqual(task_context["task_mode"], "img2img")
self.assertEqual(task_context["model_family"], "SDXL")
self.assertEqual(task_context["recommended_width"], 1024)
self.assertEqual(task_context["recommended_height"], 1024)
self.assertEqual(task_context["recommended_cfg"], 7.0)
self.assertEqual(task_context["recommended_denoise"], 0.5)
self.assertEqual(task_context["source"], "LLS Task Controller")
def test_loader_schema_exposes_task_context_and_source_controls(self):
def test_loader_schema_uses_standard_ports_only(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSSimpleCheckpointLoader"]
required = node_cls.INPUT_TYPES()["required"]
optional = node_cls.INPUT_TYPES()["optional"]
self.assertEqual(
node_cls.RETURN_TYPES,
("MODEL", "CLIP", "VAE", "LLS_TASK_CONTEXT", "LLS_TEXT_ENCODER"),
("MODEL", "CLIP", "VAE", "CLIP"),
)
self.assertEqual(
node_cls.RETURN_NAMES,
("model", "clip", "vae", "task_context", "text_encoder"),
("model", "clip", "vae", "text_encoder"),
)
self.assertEqual(
tuple(required.keys()),
@@ -290,14 +261,9 @@ class TestLoaderPromptRefactor(unittest.TestCase):
"external_text_encoder_2",
),
)
family_choices = required["model_family"][0]
self.assertIn("SD1.5", family_choices)
self.assertIn("SDXL_TURBO", family_choices)
self.assertIn("FLUX_SCHNELL", family_choices)
self.assertIn("FLUX_DEV", family_choices)
self.assertEqual(optional["task_context"][0], "LLS_TASK_CONTEXT")
self.assertNotIn("optional", node_cls.INPUT_TYPES())
def test_loader_returns_updated_task_context_for_sd15_embedded_resources(self):
def test_loader_tags_sd15_resources_without_context_output(self):
load_plugin_package()
from lls_node_test_refactor.model_loader import nodes as loader_nodes
@@ -307,7 +273,7 @@ class TestLoaderPromptRefactor(unittest.TestCase):
ComfySDStub(),
), mock.patch.object(loader_nodes, "comfy_core_nodes", CoreNodesStub()):
node = loader_nodes.LLSSimpleCheckpointLoader()
model, clip, vae, task_context, text_encoder = node.load_checkpoint(
model, clip, vae, text_encoder = node.load_checkpoint(
"sd15.safetensors",
"SD1.5",
"simple",
@@ -316,25 +282,19 @@ class TestLoaderPromptRefactor(unittest.TestCase):
"(auto)",
"(auto)",
"(auto)",
task_context={"task_mode": "txt2img", "model_family": "SD1.5"},
)
self.assertEqual(model, "MODEL::SD15")
self.assertEqual(clip, "CLIP::SD15")
self.assertEqual(text_encoder, "CLIP::SD15")
self.assertEqual(text_encoder, clip)
self.assertEqual(vae, "VAE::SD15")
self.assertEqual(task_context["resolved_model_family"], "SD1.5")
self.assertEqual(task_context["checkpoint_name"], "sd15.safetensors")
self.assertEqual(task_context["text_encoder_type"], "clip")
self.assertEqual(task_context["vae_source"], "embedded")
self.assertEqual(task_context["latent_channels"], 4)
self.assertTrue(task_context["supports_img2img"])
self.assertTrue(task_context["supports_clip_skip"])
self.assertFalse(task_context["supports_flux_guidance"])
self.assertEqual(task_context["recommended_width"], 512)
self.assertEqual(model.label, "MODEL::SD15")
self.assertEqual(clip.label, "CLIP::SD15")
self.assertIs(text_encoder, clip)
self.assertEqual(vae.label, "VAE::SD15")
self.assertEqual(model._lls_family, "SD1.5")
self.assertEqual(model._lls_model_name, "sd15.safetensors")
self.assertEqual(clip._lls_family, "SD1.5")
self.assertEqual(clip._lls_text_encoder_type, "clip")
self.assertEqual(vae._lls_family, "SD1.5")
def test_loader_builds_flux_resources_and_updates_task_context(self):
def test_loader_tags_flux_resources_and_keeps_standard_text_encoder_alias(self):
load_plugin_package()
from lls_node_test_refactor.model_loader import nodes as loader_nodes
@@ -344,7 +304,7 @@ class TestLoaderPromptRefactor(unittest.TestCase):
ComfySDStub(),
), mock.patch.object(loader_nodes, "comfy_core_nodes", CoreNodesStub()):
node = loader_nodes.LLSSimpleCheckpointLoader()
model, clip, vae, task_context, text_encoder = node.load_checkpoint(
model, clip, vae, text_encoder = node.load_checkpoint(
"diffusion_models/flux1-schnell.safetensors",
"FLUX_SCHNELL",
"advanced",
@@ -353,22 +313,16 @@ class TestLoaderPromptRefactor(unittest.TestCase):
"ae.safetensors",
"clip_l.safetensors",
"t5xxl_fp16.safetensors",
task_context={"task_mode": "txt2img", "model_family": "FLUX"},
)
self.assertEqual(model, "MODEL::flux1-schnell.safetensors")
self.assertEqual(clip, "CLIP::flux::clip_l.safetensors+t5xxl_fp16.safetensors")
self.assertEqual(text_encoder, "CLIP::flux::clip_l.safetensors+t5xxl_fp16.safetensors")
self.assertEqual(text_encoder, clip)
self.assertEqual(vae, "VAE::ae.safetensors")
self.assertEqual(task_context["resolved_model_family"], "FLUX_SCHNELL")
self.assertEqual(task_context["text_encoder_type"], "flux_clip_l_t5xxl")
self.assertEqual(task_context["vae_source"], "external")
self.assertEqual(task_context["text_encoder_source"], "external")
self.assertEqual(task_context["latent_channels"], 128)
self.assertTrue(task_context["supports_flux_guidance"])
self.assertFalse(task_context["supports_clip_skip"])
self.assertEqual(task_context["recommended_cfg"], 1.0)
self.assertEqual(model.label, "MODEL::flux1-schnell.safetensors")
self.assertEqual(clip.label, "CLIP::flux::clip_l.safetensors+t5xxl_fp16.safetensors")
self.assertIs(text_encoder, clip)
self.assertEqual(vae.label, "VAE::ae.safetensors")
self.assertEqual(model._lls_family, "FLUX_SCHNELL")
self.assertEqual(clip._lls_text_encoder_type, "flux_clip_l_t5xxl")
self.assertEqual(clip._lls_model_name, "diffusion_models/flux1-schnell.safetensors")
self.assertEqual(vae._lls_vae_name, "ae.safetensors")
def test_loader_raises_clear_error_when_flux_text_encoder_is_missing(self):
load_plugin_package()
@@ -392,134 +346,115 @@ class TestLoaderPromptRefactor(unittest.TestCase):
"ae.safetensors",
"clip_l.safetensors",
"(auto)",
task_context={"task_mode": "txt2img", "model_family": "FLUX"},
)
def test_prompt_encode_schema_uses_task_context_port(self):
def test_prompt_encode_schema_uses_standard_clip_ports_and_prompt_info(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSSimplePromptEncode"]
required = node_cls.INPUT_TYPES()["required"]
optional = node_cls.INPUT_TYPES()["optional"]
self.assertEqual(optional["text_encoder"][0], "LLS_TEXT_ENCODER")
self.assertEqual(optional["task_context"][0], "LLS_TASK_CONTEXT")
self.assertEqual(node_cls.RETURN_TYPES, ("CONDITIONING", "CONDITIONING", "STRING"))
self.assertEqual(required["model_family"][1]["default"], "Auto")
self.assertEqual(optional["text_encoder"][0], "CLIP")
self.assertEqual(optional["clip"][0], "CLIP")
self.assertNotIn("task_context", optional)
def test_prompt_encode_dispatches_sdxl_using_task_context_dimensions(self):
def test_prompt_encode_dispatches_sdxl_using_inferred_family_defaults(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSSimplePromptEncode"]
node = node_cls()
clip = RecordingClip()
task_context = {
"model_family": "SDXL",
"resolved_model_family": "SDXL",
"text_encoder_type": "sdxl_dual_clip",
"recommended_width": 1216,
"recommended_height": 832,
"supports_clip_skip": True,
}
clip = RecordingClip(("g", "l"))
positive, negative, next_context = node.encode(
clip,
"a castle",
"low quality",
-1,
task_context=task_context,
positive, negative, prompt_info = node.encode(
clip=clip,
positive_prompt="a castle",
negative_prompt="low quality",
clip_skip=-1,
model_family="Auto",
)
self.assertEqual(len(clip.calls), 2)
self.assertIn("g", clip.calls[0]["tokens"])
self.assertIn("l", clip.calls[0]["tokens"])
self.assertEqual(
clip.calls[0]["add_dict"],
{
"width": 1216,
"height": 832,
"width": 1024,
"height": 1024,
"crop_w": 0,
"crop_h": 0,
"target_width": 1216,
"target_height": 832,
"target_width": 1024,
"target_height": 1024,
},
)
payload = json.loads(prompt_info)
self.assertEqual(payload["model_family"], "SDXL")
self.assertEqual(payload["prompt_mode"], "sdxl")
self.assertEqual(payload["clip_skip"], -1)
self.assertEqual(positive[0][0], "cond::a castle")
self.assertEqual(negative[0][0], "cond::low quality")
self.assertEqual(next_context["resolved_model_family"], "SDXL")
self.assertEqual(next_context["prompt_mode"], "sdxl")
self.assertEqual(next_context["positive_prompt_length"], len("a castle"))
self.assertEqual(next_context["negative_prompt_length"], len("low quality"))
def test_prompt_encode_dispatches_flux_and_neutralizes_negative_prompt(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSSimplePromptEncode"]
node = node_cls()
clip = RecordingClip()
task_context = {
"model_family": "FLUX",
"resolved_model_family": "FLUX_DEV",
"guidance": 3.5,
"supports_flux_guidance": True,
"text_encoder_type": "flux_clip_l_t5xxl",
}
clip = RecordingClip(("t5xxl",))
positive, negative, next_context = node.encode(
clip,
"a robot in a forest",
"ugly, blurry",
-1,
task_context=task_context,
positive, negative, prompt_info = node.encode(
clip=clip,
positive_prompt="a robot in a forest",
negative_prompt="ugly, blurry",
clip_skip=-1,
model_family="Auto",
)
self.assertEqual(len(clip.calls), 2)
self.assertIn("t5xxl", clip.calls[0]["tokens"])
self.assertEqual(clip.calls[0]["add_dict"]["guidance"], 3.5)
self.assertEqual(clip.calls[1]["tokens"]["text"], "")
payload = json.loads(prompt_info)
self.assertEqual(payload["model_family"], "FLUX_DEV")
self.assertEqual(payload["negative_mode"], "ignored_for_flux")
self.assertEqual(payload["prompt_mode"], "flux")
self.assertEqual(positive[0][0], "cond::a robot in a forest")
self.assertEqual(negative[0][0], "cond::")
self.assertEqual(next_context["negative_mode"], "ignored_for_flux")
self.assertEqual(next_context["prompt_mode"], "flux")
def test_empty_latent_uses_task_context_recommended_size_and_flux_shape(self):
def test_empty_latent_uses_model_family_inference_for_flux_shape(self):
load_plugin_package()
from lls_node_test_refactor.latent import nodes as latent_nodes
node = latent_nodes.LLSSimpleEmptyLatent()
task_context = {
"model_family": "FLUX",
"resolved_model_family": "FLUX_SCHNELL",
"latent_channels": 128,
"downscale_ratio": 16,
"recommended_width": 1024,
"recommended_height": 1024,
}
model = TaggedValue("MODEL::flux1-schnell.safetensors")
model._lls_family = "FLUX_SCHNELL"
with mock.patch.object(latent_nodes, "torch", FakeTorch()):
latent, width, height, next_context = node.create_empty_latent(
latent, width, height, latent_info = node.create_empty_latent(
"Family Default",
512,
512,
2,
task_context=task_context,
model_family="Auto",
model=model,
)
payload = json.loads(latent_info)
self.assertEqual((width, height), (1024, 1024))
self.assertEqual(tuple(latent["samples"].shape), (2, 128, 64, 64))
self.assertEqual(latent["downscale_ratio_spacial"], 16)
self.assertEqual(next_context["resolved_model_family"], "FLUX_SCHNELL")
self.assertEqual(next_context["latent_source"], "empty_latent")
self.assertEqual(next_context["final_width"], 1024)
self.assertEqual(next_context["batch_size"], 2)
self.assertEqual(payload["model_family"], "FLUX_SCHNELL")
self.assertEqual(payload["task_mode"], "txt2img")
self.assertEqual(payload["batch_size"], 2)
def test_ksampler_reads_task_context_and_updates_it_for_img2img(self):
def test_ksampler_reads_model_and_latent_inference_for_img2img(self):
load_plugin_package()
from lls_node_test_refactor.sampling import nodes as sampling_nodes
recorded = {}
model = TaggedValue("MODEL::SDXL_TURBO")
model._lls_family = "SDXL_TURBO"
def fake_common_ksampler(**kwargs):
recorded.update(kwargs)
return kwargs["latent"]
fake_node_helpers = types.SimpleNamespace(
conditioning_set_values=lambda conditioning, values: [{"conditioning": conditioning, **values}]
)
with mock.patch.object(sampling_nodes, "comfy_sample", object()), mock.patch.object(
sampling_nodes,
"comfy_samplers",
@@ -528,17 +463,14 @@ class TestLoaderPromptRefactor(unittest.TestCase):
sampling_nodes,
"_common_ksampler",
side_effect=fake_common_ksampler,
), mock.patch.object(
sampling_nodes,
"node_helpers",
fake_node_helpers,
):
node = sampling_nodes.LLSSimpleKSampler()
latent, sample_info, task_context = node.sample(
model="model",
latent, sample_info = node.sample(
model=model,
positive=[["pos", {}]],
negative=[["neg", {}]],
latent_image={"samples": FakeTensor((1, 4, 64, 64))},
latent_image={"samples": FakeTensor((1, 4, 64, 64)), "source": "image_encode"},
model_family="Auto",
quality_preset="Family Default",
seed=123,
steps=30,
@@ -547,30 +479,16 @@ class TestLoaderPromptRefactor(unittest.TestCase):
scheduler="karras",
denoise=1.0,
flux_guidance=9.0,
task_context={
"model_family": "SDXL",
"resolved_model_family": "SDXL_TURBO",
"task_mode": "img2img",
"latent_source": "image_encode",
"recommended_steps": 4,
"recommended_cfg": 1.0,
"recommended_denoise": 0.5,
"recommended_sampler": "euler",
"recommended_scheduler": "normal",
"supports_flux_guidance": False,
},
)
sample_data = json.loads(sample_info)
self.assertEqual(recorded["steps"], 4)
self.assertEqual(recorded["cfg"], 1.0)
self.assertEqual(recorded["denoise"], 0.5)
self.assertEqual(recorded["denoise"], 1.0)
self.assertEqual(sample_data["family"], "SDXL_TURBO")
self.assertEqual(sample_data["task_mode"], "img2img")
self.assertEqual(sample_data["sampler_name"], "euler")
self.assertEqual(latent["samples"].shape, (1, 4, 64, 64))
self.assertTrue(task_context["sampled"])
self.assertEqual(task_context["task_mode"], "img2img")
self.assertEqual(task_context["denoise"], 0.5)
def test_ksampler_accepts_blank_flux_guidance_for_legacy_workflows(self):
load_plugin_package()
@@ -583,6 +501,8 @@ class TestLoaderPromptRefactor(unittest.TestCase):
self.assertTrue(sampling_nodes.LLSSimpleKSampler.VALIDATE_INPUTS(flux_guidance=""))
recorded = {}
model = TaggedValue("MODEL::FLUX_DEV")
model._lls_family = "FLUX_DEV"
def fake_common_ksampler(**kwargs):
recorded.update(kwargs)
@@ -605,11 +525,12 @@ class TestLoaderPromptRefactor(unittest.TestCase):
fake_node_helpers,
):
node = sampling_nodes.LLSSimpleKSampler()
latent, sample_info, task_context = node.sample(
model="model",
latent, sample_info = node.sample(
model=model,
positive=[["pos", {}]],
negative=[["neg", {}]],
latent_image={"samples": FakeTensor((1, 128, 64, 64)), "source": "empty_latent"},
model_family="Auto",
quality_preset="Manual",
seed=123,
steps=20,
@@ -618,85 +539,79 @@ class TestLoaderPromptRefactor(unittest.TestCase):
scheduler="simple",
denoise=1.0,
flux_guidance="",
task_context={
"resolved_model_family": "FLUX_DEV",
"supports_flux_guidance": True,
"recommended_guidance": 3.5,
},
)
sample_data = json.loads(sample_info)
self.assertEqual(recorded["positive"][0]["guidance"], 3.5)
self.assertEqual(recorded["negative"][0]["guidance"], 3.5)
self.assertEqual(sample_data["guidance"], 3.5)
self.assertEqual(task_context["guidance"], 3.5)
self.assertEqual(latent["samples"].shape, (1, 128, 64, 64))
def test_vae_decode_updates_task_context_for_flux(self):
def test_vae_decode_returns_decode_info_for_flux(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSSimpleVAEDecode"]
node = node_cls()
image, task_context = node.decode(
{"samples": FakeTensor((1, 128, 64, 64))},
FakeVAE(),
task_context={"resolved_model_family": "FLUX_DEV", "downscale_ratio": 16, "vae_name": "ae.safetensors"},
image, decode_info = node.decode(
{"samples": FakeTensor((1, 128, 64, 64)), "downscale_ratio_spacial": 16},
FakeVAE(vae_name="ae.safetensors"),
)
payload = json.loads(decode_info)
self.assertEqual(tuple(image.shape), (1, 64, 64, 3))
self.assertEqual(task_context["final_width"], 1024)
self.assertEqual(task_context["vae_name"], "ae.safetensors")
self.assertTrue(task_context["decoded"])
self.assertEqual(task_context["decode_stage"], "vae_decode")
self.assertEqual(payload["width"], 1024)
self.assertEqual(payload["height"], 1024)
self.assertEqual(payload["vae_name"], "ae.safetensors")
self.assertEqual(payload["decode_stage"], "vae_decode")
def test_vae_encode_schema_exposes_img2img_inputs_and_task_context_port(self):
def test_vae_encode_schema_exposes_img2img_inputs_without_context_port(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSSimpleVAEEncode"]
required = node_cls.INPUT_TYPES()["required"]
optional = node_cls.INPUT_TYPES()["optional"]
self.assertEqual(node_cls.RETURN_TYPES, ("LATENT", "INT", "INT", "LLS_TASK_CONTEXT"))
self.assertEqual(node_cls.RETURN_TYPES, ("LATENT", "INT", "INT", "STRING"))
self.assertEqual(required["image"][0], "IMAGE")
self.assertEqual(required["vae"][0], "VAE")
self.assertEqual(required["resize_mode"][0], ["keep_aspect", "crop_center", "stretch", "none"])
self.assertEqual(required["size_source"][0], ["input_image", "custom", "model_recommended"])
self.assertEqual(optional["task_context"][0], "LLS_TASK_CONTEXT")
self.assertEqual(required["model_family"][1]["default"], "Auto")
self.assertEqual(optional["model"][0], "MODEL")
self.assertEqual(optional["clip"][0], "CLIP")
self.assertNotIn("task_context", optional)
def test_vae_encode_uses_task_context_recommended_size_for_img2img(self):
def test_vae_encode_uses_inferred_model_recommended_size_for_img2img(self):
load_plugin_package()
from lls_node_test_refactor.image import nodes as image_nodes
comfy_utils = FakeComfyUtils()
vae = FakeVAE()
node = image_nodes.LLSSimpleVAEEncode()
model = TaggedValue("MODEL::SDXL")
model._lls_family = "SDXL"
vae._lls_vae_name = "embedded"
with mock.patch.object(image_nodes, "comfy_utils", comfy_utils):
latent, width, height, task_context = node.encode(
latent, width, height, latent_info = node.encode(
image=FakeTensor((1, 768, 1536, 3)),
vae=vae,
resize_mode="keep_aspect",
size_source="model_recommended",
width=512,
height=512,
task_context={
"model_family": "SDXL",
"resolved_model_family": "SDXL",
"recommended_width": 1024,
"recommended_height": 1024,
"vae_source": "embedded",
},
model_family="Auto",
model=model,
)
payload = json.loads(latent_info)
self.assertEqual((width, height), (1024, 512))
self.assertEqual(tuple(latent["samples"].shape), (1, 4, 64, 128))
self.assertEqual(latent["downscale_ratio_spacial"], 8)
self.assertEqual(vae.encoded_shapes, [(1, 512, 1024, 3)])
self.assertEqual(task_context["latent_source"], "image_encode")
self.assertEqual(task_context["final_width"], 1024)
self.assertEqual(task_context["final_height"], 512)
self.assertEqual(task_context["resize_mode"], "keep_aspect")
self.assertEqual(task_context["vae_source"], "embedded")
self.assertEqual(task_context["input_image_width"], 1536)
self.assertEqual(payload["task_mode"], "img2img")
self.assertEqual(payload["latent_source"], "image_encode")
self.assertEqual(payload["model_family"], "SDXL")
self.assertEqual(payload["vae_name"], "embedded")
self.assertEqual(
comfy_utils.calls,
[
@@ -716,87 +631,65 @@ class TestLoaderPromptRefactor(unittest.TestCase):
vae = FakeVAE()
node = image_nodes.LLSSimpleVAEEncode()
latent, width, height, task_context = node.encode(
latent, width, height, latent_info = node.encode(
image=FakeTensor((1, 512, 768, 3)),
vae=vae,
resize_mode="none",
size_source="input_image",
width=1024,
height=1024,
task_context={"model_family": "SD1.5", "resolved_model_family": "SD1.5", "vae_source": "embedded"},
model_family="SD1.5",
)
payload = json.loads(latent_info)
self.assertEqual((width, height), (768, 512))
self.assertEqual(tuple(latent["samples"].shape), (1, 4, 64, 96))
self.assertEqual(latent["downscale_ratio_spacial"], 8)
self.assertEqual(vae.encoded_shapes, [(1, 512, 768, 3)])
self.assertEqual(task_context["latent_source"], "image_encode")
self.assertEqual(task_context["size_source"], "input_image")
self.assertEqual(task_context["resize_mode"], "none")
self.assertEqual(payload["latent_source"], "image_encode")
self.assertEqual(payload["size_source"], "input_image")
self.assertEqual(payload["resize_mode"], "none")
def test_upscale_switcher_passthroughs_when_task_context_disables_upscale(self):
load_plugin_package()
from lls_node_test_refactor.upscale import nodes as upscale_nodes
recorder_calls = []
def fake_pytorch(self, image, scale, interpolation):
recorder_calls.append(("pytorch", scale, interpolation))
return ("INTERP_RESULT",)
with mock.patch.object(
upscale_nodes.LLSUpscaleSwitcher,
"_upscale_with_pytorch",
new=fake_pytorch,
):
node = upscale_nodes.LLSUpscaleSwitcher()
image, task_context = node.upscale(
image="image",
mode="upscale_model",
scale=2.0,
interpolation="bilinear",
model_name=upscale_nodes.NO_UPSCALE_MODEL_PLACEHOLDER,
tile=512,
overlap=32,
task_context={"resolved_model_family": "SDXL", "enable_upscale": False},
)
self.assertEqual(image, "image")
self.assertEqual(task_context["upscaled"], False)
self.assertEqual(task_context["upscale_mode"], "disabled_by_task_context")
self.assertEqual(recorder_calls, [])
def test_save_image_merges_metadata_from_task_context(self):
def test_save_image_merges_metadata_from_native_objects_and_info_strings(self):
load_plugin_package()
from lls_node_test_refactor.image import nodes as image_nodes
model = TaggedValue("MODEL::SD15")
model._lls_family = "SD1.5"
model._lls_model_name = "sd15.safetensors"
clip = TaggedValue("CLIP::SD15")
clip._lls_text_encoder_name = "clip_l.safetensors"
vae = TaggedValue("VAE::SD15")
vae._lls_vae_name = "VAE::SD15"
with mock.patch.object(image_nodes, "comfy_core_nodes", CoreNodesStub()):
node = image_nodes.LLSSaveImage()
result = node.save(
image=FakeTensor((1, 512, 512, 3)),
filename_prefix="LLS",
save_metadata=True,
task_context={
"model_family": "SD1.5",
"resolved_model_family": "SD1.5",
"checkpoint_name": "sd15.safetensors",
"vae_name": "VAE::SD15",
"text_encoder_name_1": "clip_l.safetensors",
"positive_prompt": "cat",
"negative_prompt": "bad",
"seed": 42,
"steps": 20,
"cfg": 7.0,
"guidance": None,
"sampler_name": "euler",
"scheduler": "normal",
"denoise": 1.0,
"final_width": 512,
"final_height": 512,
"batch_size": 1,
"upscale_mode": "none",
"upscale_scale": 1.0,
},
model=model,
clip=clip,
vae=vae,
prompt_info=json.dumps(
{
"positive_prompt": "cat",
"negative_prompt": "bad",
"model_family": "SD1.5",
}
),
sample_info=json.dumps(
{
"seed": 42,
"steps": 20,
"cfg": 7.0,
"guidance": None,
"sampler_name": "euler",
"scheduler": "normal",
"denoise": 1.0,
}
),
upscale_info=json.dumps({"mode": "none", "scale": 1.0}),
)
metadata = result["ui"]["images"][0]["extra_pnginfo"]["lls_metadata"]
@@ -805,6 +698,34 @@ class TestLoaderPromptRefactor(unittest.TestCase):
self.assertEqual(metadata["steps"], 20)
self.assertEqual(metadata["upscale_mode"], "none")
def test_generation_config_uses_model_inference_without_context(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSGenerationConfig"]
node = node_cls()
required = node_cls.INPUT_TYPES()["required"]
optional = node_cls.INPUT_TYPES()["optional"]
model = TaggedValue("MODEL::FLUX_DEV")
model._lls_family = "FLUX_DEV"
self.assertEqual(required["model_family"][1]["default"], "Auto")
self.assertEqual(optional["model"][0], "MODEL")
self.assertEqual(optional["clip"][0], "CLIP")
width, height, steps, cfg, guidance, sampler_name, scheduler, denoise, config_info = node.execute(
quality_preset="Family Default",
size_preset="Family Default",
model_family="Auto",
model=model,
)
payload = json.loads(config_info)
self.assertEqual((width, height), (1024, 1024))
self.assertEqual((steps, cfg, guidance), (20, 1.0, 3.5))
self.assertEqual(sampler_name, "euler")
self.assertEqual(scheduler, "simple")
self.assertEqual(denoise, 1.0)
self.assertEqual(payload["family"], "FLUX_DEV")
if __name__ == "__main__":
unittest.main()
tests/test_upscale_switcher.py
+6 -4
View File
@@ -1,4 +1,5 @@
import importlib.util
import json
import pathlib
import sys
import unittest
@@ -62,7 +63,7 @@ class TestUpscaleSwitcher(unittest.TestCase):
"_upscale_with_model",
new=RecordingUpscaleSwitcher._upscale_with_model,
):
result = upscale_nodes.LLSUpscaleSwitcher.upscale(
image, upscale_info = upscale_nodes.LLSUpscaleSwitcher.upscale(
recorder,
image="image",
mode="upscale_model",
@@ -71,10 +72,11 @@ class TestUpscaleSwitcher(unittest.TestCase):
model_name=upscale_nodes.NO_UPSCALE_MODEL_PLACEHOLDER,
tile=512,
overlap=32,
)
)
self.assertEqual(result[0], "PYTORCH_RESULT")
self.assertEqual(result[1]["upscale_mode"], "interpolation")
payload = json.loads(upscale_info)
self.assertEqual(image, "PYTORCH_RESULT")
self.assertEqual(payload["mode"], "interpolation")
self.assertEqual(
recorder.calls,
[("pytorch", "image", 2.0, "bilinear")],
upscale/nodes.py
+48 -65
View File
@@ -11,8 +11,7 @@ from __future__ import annotations
from typing import Any
from ..utils.model_info import parse_model_info
from ..utils.task_context import LLS_TASK_CONTEXT_TYPE, parse_task_context, update_task_context
from ..utils.model_info import info_to_json
# ---------- 从根模块导入共享工具函数 ----------
@@ -217,8 +216,8 @@ class LLSUpscaleSwitcher:
CATEGORY = "LLS/Upscale"
FUNCTION = "upscale"
RETURN_TYPES = ("IMAGE", LLS_TASK_CONTEXT_TYPE)
RETURN_NAMES = ("image", "task_context")
RETURN_TYPES = ("IMAGE", "STRING")
RETURN_NAMES = ("image", "upscale_info")
DESCRIPTION = "Switch between ComfyUI upscale models and PyTorch interpolation."
@classmethod
@@ -235,53 +234,37 @@ class LLSUpscaleSwitcher:
"tile": ("INT", {"default": 512, "min": 128, "max": 2048, "step": 64}),
"overlap": ("INT", {"default": 32, "min": 0, "max": 256, "step": 8}),
},
"optional": {
"task_context": (LLS_TASK_CONTEXT_TYPE,),
},
}
def upscale(self, image, mode: str, scale: float, interpolation: str, model_name: str, tile: int, overlap: int, task_context=None):
context = parse_task_context(task_context)
info = parse_model_info(context)
def upscale(self, image, mode: str, scale: float, interpolation: str, model_name: str, tile: int, overlap: int):
requested_mode = mode
warning = None
if mode not in UPSCALE_MODE_CHOICES:
raise RuntimeError(f"Unsupported mode '{mode}'. Expected one of {UPSCALE_MODE_CHOICES}.")
if not context.get("enable_upscale", True) and mode != "none":
return (
image,
update_task_context(
context,
upscaled=False,
upscale_mode="disabled_by_task_context",
upscale_scale=1.0,
source="LLS Upscale Switcher",
),
)
if mode == "none":
return (
image,
update_task_context(
context,
upscaled=False,
upscale_mode="none",
requested_upscale_mode=requested_mode,
upscale_scale=1.0,
source="LLS Upscale Switcher",
info_to_json(
{
"mode": "none",
"requested_mode": requested_mode,
"scale": 1.0,
"upscaled": False,
}
),
)
if mode in {"interpolation", "pytorch"}:
result = self._upscale_with_pytorch(image=image, scale=scale, interpolation=interpolation)[0]
return (
result,
update_task_context(
context,
upscaled=True,
upscale_mode="interpolation",
requested_upscale_mode=requested_mode,
upscale_scale=scale,
interpolation=interpolation,
source="LLS Upscale Switcher",
info_to_json(
{
"mode": "interpolation",
"requested_mode": requested_mode,
"scale": scale,
"interpolation": interpolation,
"upscaled": True,
}
),
)
if mode == "latent_upscale":
@@ -289,15 +272,15 @@ class LLSUpscaleSwitcher:
result = self._upscale_with_pytorch(image=image, scale=scale, interpolation=interpolation)[0]
return (
result,
update_task_context(
context,
upscaled=True,
upscale_mode="interpolation",
requested_upscale_mode=requested_mode,
upscale_scale=scale,
interpolation=interpolation,
upscale_warning=warning,
source="LLS Upscale Switcher",
info_to_json(
{
"mode": "interpolation",
"requested_mode": requested_mode,
"scale": scale,
"interpolation": interpolation,
"warning": warning,
"upscaled": True,
}
),
)
if _is_missing_upscale_model_selection(model_name):
@@ -309,32 +292,32 @@ class LLSUpscaleSwitcher:
result = self._upscale_with_pytorch(image=image, scale=scale, interpolation=interpolation)[0]
return (
result,
update_task_context(
context,
upscaled=True,
upscale_mode="interpolation",
requested_upscale_mode=requested_mode,
upscale_scale=scale,
interpolation=interpolation,
upscale_warning=warning,
source="LLS Upscale Switcher",
info_to_json(
{
"mode": "interpolation",
"requested_mode": requested_mode,
"scale": scale,
"interpolation": interpolation,
"warning": warning,
"upscaled": True,
}
),
)
result = self._upscale_with_model(image=image, model_name=model_name, tile=tile, overlap=overlap)[0]
actual_mode = "tile_upscale" if requested_mode == "tile_upscale" else "upscale_model"
return (
result,
update_task_context(
context,
upscaled=True,
upscale_mode=actual_mode,
requested_upscale_mode=requested_mode,
upscale_scale=scale,
model_name=model_name,
tile=tile,
overlap=overlap,
upscale_warning=warning,
source="LLS Upscale Switcher",
info_to_json(
{
"mode": actual_mode,
"requested_mode": requested_mode,
"scale": scale,
"model_name": model_name,
"tile": tile,
"overlap": overlap,
"warning": warning,
"upscaled": True,
}
),
)
utils/model_info.py
+83
View File
@@ -369,6 +369,9 @@ def get_sampling_preset(model_info: dict[str, Any] | str | None, quality_preset:
def infer_family_from_clip(clip) -> str:
"""从 CLIP 对象的 tokenizer 结构自动推断模型家族。"""
lls_family = getattr(clip, "_lls_family", None)
if lls_family and lls_family not in ("auto", "Auto", ""):
return canonicalize_family(lls_family)
try:
tokens = clip.tokenize("")
except Exception:
@@ -411,3 +414,83 @@ def infer_task_mode_from_latent(latent: dict) -> str:
if source == "image_encode":
return "img2img"
return "txt2img"
def tag_lls_object(obj, **kwargs: Any):
"""为原生 ComfyUI 对象附加轻量级 `_lls_*` 元信息。"""
if obj is None:
return obj
for key, value in kwargs.items():
if value is None:
continue
attr_name = key if key.startswith("_lls_") else f"_lls_{key}"
try:
setattr(obj, attr_name, value)
except Exception:
continue
return obj
def get_lls_attr(obj, name: str, default: Any = None) -> Any:
if obj is None:
return default
attr_name = name if name.startswith("_lls_") else f"_lls_{name}"
return getattr(obj, attr_name, default)
def resolve_model_family(
model_family: str | None = None,
*,
model=None,
clip=None,
fallback: str = "SD1.5",
) -> str:
normalized = str(model_family or "").strip()
if normalized and normalized not in {"Auto", "auto"}:
return canonicalize_family(normalized)
if model is not None:
inferred_from_model = infer_family_from_model(model)
if inferred_from_model:
return canonicalize_family(inferred_from_model)
if clip is not None:
inferred_from_clip = infer_family_from_clip(clip)
if inferred_from_clip:
return canonicalize_family(inferred_from_clip)
return canonicalize_family(fallback)
def resolve_model_name(model=None, clip=None, fallback: str = "") -> str:
return str(
get_lls_attr(model, "model_name")
or get_lls_attr(clip, "model_name")
or fallback
or ""
)
def resolve_vae_name(vae=None, fallback: str = "") -> str:
return str(get_lls_attr(vae, "vae_name") or fallback or "")
def resolve_text_encoder_names(clip=None) -> dict[str, str]:
name = get_lls_attr(clip, "text_encoder_name") or ""
name_1 = get_lls_attr(clip, "text_encoder_name_1") or ""
name_2 = get_lls_attr(clip, "text_encoder_name_2") or ""
if not name and name_1 and name_2:
name = ", ".join(part for part in (name_1, name_2) if part)
elif not name:
name = name_1 or name_2 or ""
return {
"text_encoder_name": str(name),
"text_encoder_name_1": str(name_1),
"text_encoder_name_2": str(name_2),
}
def build_info_payload(**kwargs: Any) -> dict[str, Any]:
return {key: value for key, value in kwargs.items() if value is not None}
utils/nodes.py
+10 -10
View File
@@ -34,13 +34,13 @@ from __future__ import annotations
from .model_info import (
FAMILY_DEFAULT_PRESET,
MODEL_FAMILY_CHOICES,
SIZE_PRESET_AUTO,
get_family_defaults,
get_sampling_preset,
info_to_json,
parse_model_info,
resolve_model_family,
)
from .task_context import LLS_TASK_CONTEXT_TYPE, parse_task_context
# ---------- 节点类示例（基础类型，纯 Python 无外部依赖） ----------
@@ -168,20 +168,20 @@ class LLSGenerationConfig:
def INPUT_TYPES(cls):
return {
"required": {
"model_family": (MODEL_FAMILY_CHOICES, {"default": "Auto"}),
"quality_preset": (cls._QUALITY_PRESETS, {"default": FAMILY_DEFAULT_PRESET}),
"size_preset": (cls._SIZE_PRESETS, {"default": SIZE_PRESET_AUTO}),
},
"optional": {
"task_context": (LLS_TASK_CONTEXT_TYPE,),
"model_info": (LLS_TASK_CONTEXT_TYPE,),
"model": ("MODEL",),
"clip": ("CLIP",),
},
}
def execute(self, quality_preset: str, size_preset: str, task_context=None, model_info=None):
context = parse_task_context(task_context or model_info)
info = parse_model_info(context)
defaults = get_family_defaults(info["family"])
preset = get_sampling_preset(info, quality_preset) or {
def execute(self, model_family: str, quality_preset: str, size_preset: str, model=None, clip=None):
family = resolve_model_family(model_family, model=model, clip=clip)
defaults = get_family_defaults(family)
preset = get_sampling_preset(defaults, quality_preset) or {
"steps": defaults["default_steps"],
"cfg": defaults["default_cfg"],
"guidance": defaults["default_guidance"],
@@ -203,7 +203,7 @@ class LLSGenerationConfig:
guidance = 0.0 if guidance is None else float(guidance)
config_info = info_to_json(
{
"family": context.get("resolved_model_family") or info["family"],
"family": family,
"width": width,
"height": height,
"steps": int(preset["steps"]),