feat(lora-cycler): add sequential LoRA cycling through filtered pool

Add Lora Cycler node that cycles through LoRAs sequentially from a filtered pool. Supports configurable sort order, strength settings, and persists cycle progress across workflow save/load.

Backend:
- New LoraCyclerNode with cycle() method
- New /api/lm/loras/cycler-list endpoint
- LoraService.get_cycler_list() for filtered/sorted list

Frontend:
- LoraCyclerWidget with Vue.js component
- useLoraCyclerState composable
- LoraCyclerSettingsView for UI display
This commit is contained in:
Will Miao
2026-01-22 15:36:32 +08:00
parent 17c5583297
commit 6fbea77137
11 changed files with 2329 additions and 262 deletions

130
py/nodes/lora_cycler.py Normal file
View File

@@ -0,0 +1,130 @@
"""
Lora Cycler Node - Sequentially cycles through LoRAs from a pool.
This node accepts optional pool_config input to filter available LoRAs, and outputs
a LORA_STACK with one LoRA at a time. Returns UI updates with current/next LoRA info
and tracks the cycle progress which persists across workflow save/load.
"""
import logging
import os
from ..utils.utils import get_lora_info
logger = logging.getLogger(__name__)
class LoraCyclerNode:
"""Node that sequentially cycles through LoRAs from a pool"""
NAME = "Lora Cycler (LoraManager)"
CATEGORY = "Lora Manager/randomizer"
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"cycler_config": ("CYCLER_CONFIG", {}),
},
"optional": {
"pool_config": ("POOL_CONFIG", {}),
},
}
RETURN_TYPES = ("LORA_STACK",)
RETURN_NAMES = ("LORA_STACK",)
FUNCTION = "cycle"
OUTPUT_NODE = False
async def cycle(self, cycler_config, pool_config=None):
"""
Cycle through LoRAs based on configuration and pool filters.
Args:
cycler_config: Dict with cycler settings (current_index, model_strength, clip_strength, sort_by)
pool_config: Optional config from LoRA Pool node for filtering
Returns:
Dictionary with 'result' (LORA_STACK tuple) and 'ui' (for widget display)
"""
from ..services.service_registry import ServiceRegistry
from ..services.lora_service import LoraService
# Extract settings from cycler_config
current_index = cycler_config.get("current_index", 1) # 1-based
model_strength = float(cycler_config.get("model_strength", 1.0))
clip_strength = float(cycler_config.get("clip_strength", 1.0))
sort_by = cycler_config.get("sort_by", "filename")
# Get scanner and service
scanner = await ServiceRegistry.get_lora_scanner()
lora_service = LoraService(scanner)
# Get filtered and sorted LoRA list
lora_list = await lora_service.get_cycler_list(
pool_config=pool_config, sort_by=sort_by
)
total_count = len(lora_list)
if total_count == 0:
logger.warning("[LoraCyclerNode] No LoRAs available in pool")
return {
"result": ([],),
"ui": {
"current_index": [1],
"next_index": [1],
"total_count": [0],
"current_lora_name": [""],
"current_lora_filename": [""],
"error": ["No LoRAs available in pool"],
},
}
# Clamp index to valid range (1-based)
clamped_index = max(1, min(current_index, total_count))
# Get LoRA at current index (convert to 0-based for list access)
current_lora = lora_list[clamped_index - 1]
# Build LORA_STACK with single LoRA
lora_path, _ = get_lora_info(current_lora["file_name"])
if not lora_path:
logger.warning(
f"[LoraCyclerNode] Could not find path for LoRA: {current_lora['file_name']}"
)
lora_stack = []
else:
# Normalize path separators
lora_path = lora_path.replace("/", os.sep)
lora_stack = [(lora_path, model_strength, clip_strength)]
# Calculate next index (wrap to 1 if at end)
next_index = clamped_index + 1
if next_index > total_count:
next_index = 1
# Get next LoRA for UI display (what will be used next generation)
next_lora = lora_list[next_index - 1]
# Determine display name based on sort_by setting
if sort_by == "filename":
next_display_name = next_lora["file_name"]
else:
next_display_name = next_lora.get("model_name", next_lora["file_name"])
return {
"result": (lora_stack,),
"ui": {
"current_index": [clamped_index],
"next_index": [next_index],
"total_count": [total_count],
"current_lora_name": [
current_lora.get("model_name", current_lora["file_name"])
],
"current_lora_filename": [current_lora["file_name"]],
"next_lora_name": [next_display_name],
"next_lora_filename": [next_lora["file_name"]],
"sort_by": [sort_by],
},
}

View File

@@ -63,6 +63,11 @@ class LoraRoutes(BaseModelRoutes):
"POST", "/api/lm/{prefix}/random-sample", prefix, self.get_random_loras
)
# Cycler routes
registrar.add_prefixed_route(
"POST", "/api/lm/{prefix}/cycler-list", prefix, self.get_cycler_list
)
# ComfyUI integration
registrar.add_prefixed_route(
"POST", "/api/lm/{prefix}/get_trigger_words", prefix, self.get_trigger_words
@@ -283,6 +288,29 @@ class LoraRoutes(BaseModelRoutes):
logger.error(f"Error getting random LoRAs: {e}", exc_info=True)
return web.json_response({"success": False, "error": str(e)}, status=500)
async def get_cycler_list(self, request: web.Request) -> web.Response:
"""Get filtered and sorted LoRA list for cycler widget"""
try:
json_data = await request.json()
# Parse parameters
pool_config = json_data.get("pool_config")
sort_by = json_data.get("sort_by", "filename")
# Get cycler list from service
lora_list = await self.service.get_cycler_list(
pool_config=pool_config,
sort_by=sort_by
)
return web.json_response(
{"success": True, "loras": lora_list, "count": len(lora_list)}
)
except Exception as e:
logger.error(f"Error getting cycler list: {e}", exc_info=True)
return web.json_response({"success": False, "error": str(e)}, status=500)
async def get_trigger_words(self, request: web.Request) -> web.Response:
"""Get trigger words for specified LoRA models"""
try:

View File

@@ -479,3 +479,49 @@ class LoraService(BaseModelService):
]
return available_loras
async def get_cycler_list(
self,
pool_config: Optional[Dict] = None,
sort_by: str = "filename"
) -> List[Dict]:
"""
Get filtered and sorted LoRA list for cycling.
Args:
pool_config: Optional pool config for filtering (filters dict)
sort_by: Sort field - 'filename' or 'model_name'
Returns:
List of LoRA dicts with file_name and model_name
"""
# Get cached data
cache = await self.scanner.get_cached_data(force_refresh=False)
available_loras = cache.raw_data if cache else []
# Apply pool filters if provided
if pool_config:
available_loras = await self._apply_pool_filters(
available_loras, pool_config
)
# Sort by specified field
if sort_by == "model_name":
available_loras = sorted(
available_loras,
key=lambda x: (x.get("model_name") or x.get("file_name", "")).lower()
)
else: # Default to filename
available_loras = sorted(
available_loras,
key=lambda x: x.get("file_name", "").lower()
)
# Return minimal data needed for cycling
return [
{
"file_name": lora["file_name"],
"model_name": lora.get("model_name", lora["file_name"]),
}
for lora in available_loras
]