nodes/zeeoale-PromptCreatorNode
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix

Browse Source
This commit is contained in:
zeeoale
2026-02-17 10:43:13 +01:00
parent f4a238d879
commit 55766ad4e9
6 changed files with 878 additions and 272 deletions
Download Patch File Download Diff File Expand all files Collapse all files
PromptCreatorNode.py
+2 -1
View File
@@ -330,7 +330,8 @@ class PromptCreatorNode:
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"stream": False
"stream": False,
"think": False,
}, timeout=120)
r.raise_for_status()
return r.json()["message"]["content"].strip()
__init__.py
+6
View File
@@ -8,6 +8,8 @@ from .nodes.director import DirectorNode # v2 only
from .nodes.session_load import DirectorSessionLoad
from .nodes.session_merge import DirectorSessionMerge
from .nodes.session_merge_dropdown import DirectorSessionMergeDropdown
from .nodes.prompt_enhancer import PromptEnhancerNode
@@ -17,6 +19,8 @@ NODE_CLASS_MAPPINGS = {
"DirectorSessionLoad": DirectorSessionLoad,
"DirectorSessionMerge": DirectorSessionMerge,
"DirectorSessionMergeDropdown": DirectorSessionMergeDropdown,
"PromptEnhancerNode": PromptEnhancerNode,
@@ -27,6 +31,8 @@ NODE_DISPLAY_NAME_MAPPINGS = {
"DirectorSessionLoad": "Director Session Load",
"DirectorSessionMerge": "Director Session Merge",
"DirectorSessionMergeDropdown": "Director Session Merge (Dropdown)",
"PromptEnhancerNode": "PFN Prompt Enhancer",
data/worlds/PFN_Coquette_Memento_Nocturne.json
+106
View File
@@ -0,0 +1,106 @@
{
"WORLD_NAME": "Coquette_Memento_Nocturne",
"SYSTEM_PROMPT": "You are a professional prompt enhancer for AI image generation focused on coquette gothic boudoir portraiture with vintage romantic decadence and cinematic restraint; when requested, return ONE single long sentence in English (180–240 words), no line breaks and no bullet points, that merges the selected values into a coherent scene without ever listing category names, maintaining an elegant, old-world taste where sensuality is expressed through couture construction, lace tension, corsetry seams, skin highlights, posture and gaze rather than explicit acts, balancing soft chiaroscuro with muted olive-sage fog, candle-amber accents, ivory skin, ink-black velvet and worn antique gold; always integrate details from {{EPOCHS}}, {{OUTFITS}}, {{LIGHTING}}, {{BACKGROUNDS}}, {{OBJECTS}}, {{POSES}}, {{EXPRESSIONS}}, {{CAMERA_ANGLES}}, {{ATMOSPHERES}} and {{ACCESSORIES}}, and modulate the overall mood using {{SENSUALITY_LEVEL}} while keeping {{HORROR_INTENSITY}} low-to-moderate as atmospheric unease (never gore, never body horror); emphasize tactile microtexture (powdered skin, satin sheen, lace fray), subtle film grain, lens choice and framing, and a refined editorial finish; include one clear brand placement cue for TK chosen from {{ACCESSORIES}} (monogram charm, embroidery, ring seal, cameo, watermark plaque), making it feel diegetic and era-appropriate; keep compositions poised and classical, like inherited glamour photographed in a forgotten chapel dressing room, with quiet confidence and controlled tension.",
"EPOCHS": [
"late Victorian romantic revival",
"Edwardian boudoir elegance",
"Belle Époque decadence",
"1920s forbidden salon glamour",
"1950s pin-up noir crossover",
"timeless vintage editorial (no modern tech)"
],
"OUTFITS": [
"black velvet corset with ivory ruffled neckline",
"off-shoulder lace bodysuit with garter straps",
"ivory satin chemise under a fitted waist cincher",
"sheer lace sleeves with ribbon ties and pearl buttons",
"high-waist lingerie set with delicate embroidery and scalloped edges",
"structured corsetry gown with thigh-high slit and lace stockings",
"soft tulle peignoir draped over a satin bodice",
"antique-inspired choker collar with cameo clasp"
],
"LIGHTING": [
"low-key candlelight with soft rim glow",
"window spill light through dusty lace curtains",
"single practical lamp with warm falloff and deep shadows",
"moonlit haze with faint silver highlights",
"portrait key light feathered across cheekbones",
"soft bounce fill with pronounced chiaroscuro"
],
"BACKGROUNDS": [
"baroque dressing room with worn wallpaper and cracked plaster",
"shadowed chapel corridor with carved stone and faded murals",
"antique parlor with velvet drapes and tarnished mirrors",
"foggy rose-garden conservatory with dark greenery",
"decaying mansion hallway with ornate frames and patina",
"curtained alcove filled with old flowers and candle stubs"
],
"OBJECTS": [
"wilted roses and scattered petals",
"tarnished vanity mirror with hairpins",
"vintage perfume bottle and powder compact",
"candle cluster with wax drips on brass tray",
"lace gloves folded beside a cameo locket",
"old ribbon spools and embroidery scissors",
"antique frame partially covered in ivy",
"pearls spilled across a velvet cushion"
],
"POSES": [
"standing contrapposto, hip angled, one hand at waist cincher",
"seated on a stone ledge, knees together, shoulders slightly forward",
"leaning into a wall, chin tipped down, eyes up through lashes",
"half-turn over shoulder, back arched subtly, hand grazing collarbone",
"kneeling on velvet rug, torso upright, hands resting on thighs",
"perched on vanity edge, one leg crossed, fingers adjusting garter",
"one arm raised to hair, elbow framing face, other hand on corset seam",
"slow step forward through fog, dress slit revealing stocking top"
],
"EXPRESSIONS": [
"cool intimate gaze, calm and unblinking",
"soft parted lips, restrained confidence",
"faint smirk with heavy-lidded eyes",
"melancholic stare, glossy eyes, composed",
"seductive neutrality, poised and editorial",
"sly side glance, subtle challenge"
],
"CAMERA_ANGLES": [],
"ATMOSPHERES": [
"incense haze with drifting dust particles",
"muted olive fog with candle-amber bokeh",
"quiet devotional stillness, velvet-silent air",
"soft film grain and faded patina texture",
"low contrast shadows with gentle highlight bloom",
"romantic decay, elegant and controlled"
],
"ACCESSORIES": [
"TK monogram signet ring on index finger (diegetic brand)",
"TK cameo brooch at the throat (diegetic brand)",
"TK embroidery stitched into corset hem (diegetic brand)",
"TK wax-seal tag on ribbon choker (diegetic brand)",
"small TK watermark plaque in a lower corner, subtle and clean (brand)",
"hairpin shaped like a tiny TK monogram, barely visible (diegetic brand)",
"antique key with TK initials engraved on the bow (diegetic brand)",
"lace glove charm with TK initials, softly catching rim light (diegetic brand)"
],
"SENSUALITY_LEVEL": {
"0": "tasteful, modest allure, couture focus, minimal skin, classic portrait restraint",
"1": "soft seduction, lingerie implied, lace tension, collarbone and neckline emphasis",
"2": "bold boudoir editorial, visible lingerie details, stockings/garters, confident pose language",
"3": "maximum coquette erotic tension, provocative fashion framing, intimate gaze, pronounced silhouette and fabric tension, still elegant and non-explicit"
},
"HORROR_INTENSITY": {
"0": "no horror, purely romantic vintage decadence",
"1": "subtle unease, abandoned beauty, quiet shadows and fog",
"2": "gothic melancholy, heavier chiaroscuro, ritual-like stillness without violence",
"3": "dark devotional atmosphere, oppressive architecture, intense tension, still no gore and no body horror"
},
"CUSTOM_INTRO": {
"0": "A low-key cinematic portrait in muted olive fog and candle-amber glow, coquette gothic boudoir elegance with ivory lace and ink-black velvet corsetry, soft film grain, gentle highlight bloom on powdered skin, wilted roses and tarnished gold accents, poised editorial sensuality and restrained confidence, TK placed as a diegetic monogram accessory integrated into jewelry or embroidery, masterpiece, best quality, amazing quality, 4k, very aesthetic, high resolution, ultra-detailed, absurdres, newest, scenery, depth of field, volumetric lighting",
"1": "Vintage romantic decadence in a baroque dressing room, cracked plaster and worn wallpaper behind a sensual lace bodysuit with garter straps, incense haze and drifting dust, chiaroscuro rim light tracing collarbones and satin sheen, classical composition with intimate gaze, TK embedded as cameo brooch or signet ring, masterpiece, best quality, amazing quality, 4k, very aesthetic, high resolution, ultra-detailed, absurdres, newest, scenery, depth of field, volumetric lighting",
"2": "Forbidden salon glamour with velvet drapes and tarnished mirrors, off-shoulder ruffles and corsetry seams emphasized by soft window spill, cinematic close framing and shallow depth of field, antique perfume and scattered pearls, confident provocative pose language without explicitness, TK stitched into hem or engraved on an antique key, masterpiece, best quality, amazing quality, 4k, very aesthetic, high resolution, ultra-detailed, absurdres, newest, scenery, depth of field, volumetric lighting",
"3": "Chapel-corridor editorial where moonlit haze meets candle bokeh, ivory satin and black lace layered in refined coquette tension, gentle vignetting and soft grain, roses and wax drips on brass, serene control and quiet menace at low intensity, TK presented as subtle corner plaque watermark or wax-seal tag, masterpiece, best quality, amazing quality, 4k, very aesthetic, high resolution, ultra-detailed, absurdres, newest, scenery, depth of field, volumetric lighting",
"4": "Antique parlor boudoir with romantic decay, sculpted silhouette in velvet corset gown and thigh-high stockings, low contrast shadows with blooming highlights, hands adjusting garter in a poised editorial gesture, incense haze and floating particles, TK as hairpin monogram or signet ring glinting in rim light, masterpiece, best quality, amazing quality, 4k, very aesthetic, high resolution, ultra-detailed, absurdres, newest, scenery, depth of field, volumetric lighting",
"5": "Rose-garden conservatory at dusk with muted sage fog, lace sleeves tied with ribbons, powdery skin microtexture and satin sheen, shallow depth of field, classical symmetry, a calm intimate stare that reads provocative and composed, TK integrated as cameo clasp at the throat or embroidery detail, masterpiece, best quality, amazing quality, 4k, very aesthetic, high resolution, ultra-detailed, absurdres, newest, scenery, depth of field, volumetric lighting",
"6": "Full-body vintage editorial in an ornate mansion hallway, velvet and lace couture with ruffles and garters, candle clusters and tarnished frames, soft film grain and patina texture, contrapposto pose and sculpted neckline under feathered key light, TK as a diegetic accessory (wax-seal tag, ring, brooch) kept subtle and era-correct, masterpiece, best quality, amazing quality, 4k, very aesthetic, high resolution, ultra-detailed, absurdres, newest, scenery, depth of field, volumetric lighting"
}
}
nodes/director.py
+204 -271
View File
@@ -1,7 +1,9 @@
from __future__ import annotations
import json
import os
import random
from datetime import datetime
from typing import Dict, Optional, List, Tuple, Any
from ..core.world_registry import WorldRegistry
@@ -12,11 +14,13 @@ class DirectorNode:
def __init__(self):
self._last: Optional[Dict[str, str]] = None
# ======================================================
# UI
# ======================================================
@classmethod
def INPUT_TYPES(cls):
names = WorldRegistry.list_names()
if not names:
names = ["(no worlds found)"]
names = WorldRegistry.list_names() or ["(no worlds found)"]
return {
"required": {
@@ -24,47 +28,55 @@ class DirectorNode:
"seed": ("INT", {"default": 0, "min": 0, "max": 2**31 - 1}),
"lock": ("BOOLEAN", {"default": False}),
"control_after_generate": (["keep", "increment", "randomize"],),
"custom_intro_mode": (["auto", "index", "random"],),
"custom_intro_index": ("INT", {"default": -1, "min": -1, "max": 6}),
"include_system_prompt": ("BOOLEAN", {"default": True}),
"separator": (["\\n\\n", "\\n", " — ", " | ", ", "],),
"custom_separator": ("STRING", {"default": ""}),
"director_preset": (["custom", "continuity", "exploration", "chaos"],),
"lock_camera": ("BOOLEAN", {"default": True}),
"lock_lighting": ("BOOLEAN", {"default": True}),
"lock_outfit": ("BOOLEAN", {"default": True}),
"lock_pose": ("BOOLEAN", {"default": False}),
"freeze_mode": (["off", "freeze", "refresh"],),
"freeze_scope": (["all", "camera_lighting", "camera", "none"],),
"change_one": ([
"none",
"camera",
"lighting",
"outfit",
"pose",
"expression",
"background",
"objects",
"atmosphere",
"accessory",
"custom_intro",
"none", "camera", "lighting", "outfit", "pose",
"expression", "background", "objects",
"atmosphere", "accessory", "custom_intro",
],),
# R8: Shot List
# R8
"take_count": ("INT", {"default": 1, "min": 1, "max": 16}),
"take_seed_mode": (["keep", "increment", "randomize"],),
# R9: Best Take Picker
# R9
"pick_mode": (["off", "best_take"],),
"pick_rule": (["balanced", "closeup", "lowlight", "profile", "keyword"],),
"pick_keyword": ("STRING", {"default": "close-up"}),
# R10: JSON export
# R10
"json_pretty": ("BOOLEAN", {"default": True}),
# R11
"save_to_file": ("BOOLEAN", {"default": False}),
"save_format": (["json", "json+txt"],),
"save_dir": ("STRING", {"default": "logs"}),
"save_prefix": ("STRING", {"default": "director"}),
}
}
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING", "INT", "STRING", "STRING", "STRING", "INT", "STRING", "STRING")
RETURN_TYPES = (
"STRING", "STRING", "STRING", "STRING", "INT",
"STRING", "STRING", "STRING", "INT",
"STRING", "STRING"
)
RETURN_NAMES = (
"prompt",
"system_prompt",
@@ -78,11 +90,53 @@ class DirectorNode:
"shot_list_json",
"picked_json",
)
FUNCTION = "run"
CATEGORY = "PFN/Director"
# ======================================================
# HELPERS
# ======================================================
def _dump_json(self, obj: Any, pretty: bool) -> str:
return json.dumps(
obj,
ensure_ascii=False,
indent=2 if pretty else None,
separators=None if pretty else (",", ":"),
)
def _make_separator(self, separator: str, custom: str) -> str:
raw = custom.strip() if custom.strip() else separator
return raw.encode("utf-8").decode("unicode_escape")
def _compute_next_seed(self, seed: int, lock: bool, mode: str) -> int:
if lock:
return seed
if mode == "increment":
return (seed + 1) % (2**31)
if mode == "randomize":
return random.randint(0, 2**31 - 1)
return seed
def _apply_preset(
self,
preset: str,
lock_camera: bool,
lock_lighting: bool,
lock_outfit: bool,
lock_pose: bool,
) -> Tuple[str, bool, bool, bool, bool]:
p = (preset or "custom").lower()
if p == "continuity":
return p, True, True, True, False
if p == "exploration":
return p, True, False, False, False
if p == "chaos":
return p, False, False, False, False
return "custom", lock_camera, lock_lighting, lock_outfit, lock_pose
def _filter_overrides(self, data: Dict[str, str], scope: str) -> Dict[str, str]:
scope = (scope or "all").lower().strip()
if scope == "none":
return {}
if scope == "camera":
@@ -91,48 +145,7 @@ class DirectorNode:
keys = {"camera", "lighting"}
else:
keys = set(data.keys())
return {k: v for k, v in data.items() if k in keys and v and str(v).strip()}
def _apply_preset(
self,
director_preset: str,
lock_camera: bool,
lock_lighting: bool,
lock_outfit: bool,
lock_pose: bool,
) -> Tuple[str, bool, bool, bool, bool]:
preset = (director_preset or "custom").lower().strip()
if preset != "custom":
if preset == "continuity":
lock_camera = True
lock_lighting = True
lock_outfit = True
lock_pose = False
elif preset == "exploration":
lock_camera = True
lock_lighting = False
lock_outfit = False
lock_pose = False
elif preset == "chaos":
lock_camera = False
lock_lighting = False
lock_outfit = False
lock_pose = False
return preset, lock_camera, lock_lighting, lock_outfit, lock_pose
def _make_separator(self, separator: str, custom_separator: str) -> str:
sep_raw = custom_separator.strip() if custom_separator and custom_separator.strip() else separator
return sep_raw.encode("utf-8").decode("unicode_escape")
def _compute_next_seed(self, seed: int, lock: bool, mode: str) -> int:
if lock:
return seed
mode = (mode or "keep").lower().strip()
if mode == "increment":
return (seed + 1) % (2**31)
if mode == "randomize":
return random.randint(0, 2**31 - 1)
return seed
return {k: v for k, v in data.items() if k in keys}
def _score_take(
self,
@@ -141,136 +154,86 @@ class DirectorNode:
chosen: Dict[str, str],
final_prompt: str,
) -> int:
rule = (rule or "balanced").lower().strip()
kw = (keyword or "").lower().strip()
cam = (chosen.get("camera") or "").lower()
light = (chosen.get("lighting") or "").lower()
def has_any(text: str, needles: List[str]) -> bool:
return any(n in text for n in needles)
cam = chosen.get("camera", "").lower()
light = chosen.get("lighting", "").lower()
score = 0
if rule == "closeup":
if has_any(cam, ["close", "85mm", "105mm", "portrait", "tight"]):
score += 10
if "24mm" in cam:
score -= 2
if rule == "closeup" and any(x in cam for x in ["close", "85mm", "105mm"]):
score += 10
if rule == "lowlight" and any(x in light for x in ["low", "dark", "shadow"]):
score += 10
if rule == "profile" and "profile" in cam:
score += 10
if rule == "keyword":
score += final_prompt.lower().count(keyword.lower()) * 10
elif rule == "lowlight":
if has_any(light, ["low", "dim", "candle", "noir", "shadow", "dark"]):
score += 10
if has_any(light, ["bright", "daylight", "high key"]):
score -= 3
elif rule == "profile":
if "profile" in cam:
score += 10
if has_any(cam, ["frontal", "centered frontal"]):
score -= 1
elif rule == "keyword":
if kw:
score += final_prompt.lower().count(kw) * 10
else: # balanced
if has_any(cam, ["close", "portrait", "85mm", "105mm"]):
if rule == "balanced":
if "portrait" in cam:
score += 5
if has_any(light, ["low", "shadow", "noir", "candle", "dim"]):
if "low" in light:
score += 5
return score
def _dump_json(self, obj: Any, pretty: bool) -> str:
if pretty:
return json.dumps(obj, ensure_ascii=False, indent=2)
return json.dumps(obj, ensure_ascii=False, separators=(",", ":"))
# ======================================================
# RUN
# ======================================================
def run(
self,
world: str,
seed: int,
lock: bool,
control_after_generate: str,
custom_intro_mode: str,
custom_intro_index: int,
include_system_prompt: bool,
separator: str,
custom_separator: str,
director_preset: str,
lock_camera: bool,
lock_lighting: bool,
lock_outfit: bool,
lock_pose: bool,
freeze_mode: str,
freeze_scope: str,
change_one: str,
take_count: int,
take_seed_mode: str,
pick_mode: str,
pick_rule: str,
pick_keyword: str,
json_pretty: bool,
world, seed, lock, control_after_generate,
custom_intro_mode, custom_intro_index,
include_system_prompt, separator, custom_separator,
director_preset,
lock_camera, lock_lighting, lock_outfit, lock_pose,
freeze_mode, freeze_scope, change_one,
take_count, take_seed_mode,
pick_mode, pick_rule, pick_keyword,
json_pretty,
save_to_file, save_format, save_dir, save_prefix,
):
w = WorldRegistry.get(world)
if not w:
return ("(world not found)", "", "(world not found)", "", seed, "(world not found)", "", "", -1, "{}", "{}")
return ("(world not found)", "", "", "", seed, "", "", "", -1, "{}", "{}")
effective_seed = int(seed)
take_count = int(take_count)
preset, lock_camera, lock_lighting, lock_outfit, lock_pose = \
self._apply_preset(director_preset, lock_camera, lock_lighting, lock_outfit, lock_pose)
idx = None
if (custom_intro_mode == "index") and (custom_intro_index is not None) and (int(custom_intro_index) >= 0):
idx = int(custom_intro_index)
preset, lock_camera, lock_lighting, lock_outfit, lock_pose = self._apply_preset(
director_preset, lock_camera, lock_lighting, lock_outfit, lock_pose
)
fm = (freeze_mode or "off").lower().strip()
fs = (freeze_scope or "all").lower().strip()
co = (change_one or "none").lower().strip()
sep = self._make_separator(separator, custom_separator)
fm = freeze_mode.lower()
fs = freeze_scope.lower()
co = change_one.lower()
# best take tracking
best_score = -10**9
best_take_idx = -1
best_final_prompt = ""
shot_list = []
takes_json = []
best_score = -1_000_000
best_take = -1
best_prompt = ""
best_notes = ""
best_chosen: Dict[str, str] = {}
best_seed = effective_seed
best_chosen = {}
# last take outputs
current_seed = seed
last_prompt = ""
last_system_prompt = ""
last_final_prompt = ""
last_notes = ""
shot_blocks: List[str] = []
takes_json: List[Dict[str, Any]] = []
current_seed = effective_seed
best_prompt = ""
best_notes = ""
best_take = -1
for i in range(take_count):
overrides: Optional[Dict[str, str]] = None
applied_overrides: Dict[str, str] = {}
overrides = None
if fm == "freeze" and self._last:
applied_overrides = self._filter_overrides(self._last, fs)
if co != "none" and co in applied_overrides:
applied_overrides.pop(co, None)
overrides = dict(applied_overrides) if applied_overrides else None
elif fm == "refresh":
overrides = None
else:
overrides = None
overrides = self._filter_overrides(self._last, fs)
if co != "none":
overrides.pop(co, None)
prompt, system_prompt, director_notes, chosen = build_prompt_from_world(
prompt, system_prompt, notes, chosen = build_prompt_from_world(
w,
seed=current_seed,
custom_intro_mode=custom_intro_mode,
custom_intro_index=idx,
custom_intro_index=custom_intro_index if custom_intro_index >= 0 else None,
lock_camera=lock_camera,
lock_lighting=lock_lighting,
lock_outfit=lock_outfit,
@@ -278,130 +241,100 @@ class DirectorNode:
overrides=overrides,
)
# memory update
if fm in ("off", "refresh"):
self._last = dict(chosen)
elif fm == "freeze" and co != "none":
self._last = dict(chosen)
# final prompt
if include_system_prompt:
sp_out = system_prompt
final_prompt = f"{system_prompt}{sep}{prompt}".strip() if system_prompt else prompt
else:
sp_out = ""
final_prompt = prompt
applied_keys = sorted(applied_overrides.keys()) if applied_overrides else []
header = (
f"preset: {preset}\n"
f"freeze_mode: {fm}\n"
f"freeze_scope: {fs}\n"
f"change_one: {co}\n"
f"applied_overrides: {', '.join(applied_keys) if applied_keys else '(none)'}\n"
f"take: {i + 1}/{take_count}\n"
f"seed: {current_seed}\n"
)
notes_out = (header + director_notes).strip()
# score
picker_on = (pick_mode or "off").lower().strip() == "best_take"
score = self._score_take(pick_rule, pick_keyword, chosen, final_prompt) if picker_on else None
# best take
if picker_on and score is not None and score > best_score:
best_score = score
best_take_idx = i + 1
best_final_prompt = final_prompt
best_notes = (notes_out + f"\nscore: {score}").strip()
best_chosen = dict(chosen)
best_seed = current_seed
# save last
final_prompt = f"{system_prompt}{sep}{prompt}".strip() if include_system_prompt else prompt
last_prompt = prompt
last_system_prompt = sp_out
last_system_prompt = system_prompt
last_final_prompt = final_prompt
last_notes = notes_out
last_notes = notes
shot_blocks.append(
"\n".join([
f"===== TAKE {i + 1}/{take_count} =====",
notes_out + (f"\nscore: {score}" if score is not None else ""),
"----- FINAL PROMPT -----",
final_prompt,
"",
]).strip()
)
if fm in ("off", "refresh") or (fm == "freeze" and co != "none"):
self._last = dict(chosen)
score = None
if pick_mode == "best_take":
score = self._score_take(pick_rule, pick_keyword, chosen, final_prompt)
if score > best_score:
best_score = score
best_take = i + 1
best_prompt = final_prompt
best_notes = notes
best_chosen = dict(chosen)
shot_list.append(f"===== TAKE {i+1} =====\n{final_prompt}")
takes_json.append({
"take_index": i + 1,
"take": i + 1,
"seed": current_seed,
"preset": preset,
"freeze_mode": fm,
"freeze_scope": fs,
"change_one": co,
"applied_overrides": applied_keys,
"chosen": dict(chosen),
"prompt": prompt,
"system_prompt": system_prompt if include_system_prompt else "",
"chosen": chosen,
"final_prompt": final_prompt,
"notes": notes_out,
"score": score,
})
current_seed = self._compute_next_seed(current_seed, lock, take_seed_mode)
# next seed for node output
next_seed = self._compute_next_seed(effective_seed, lock, control_after_generate)
shot_list = "\n\n".join(shot_blocks).strip()
# picker off => pick last
picker_on = (pick_mode or "off").lower().strip() == "best_take"
if not picker_on:
best_take_idx = take_count
best_final_prompt = last_final_prompt
best_notes = (last_notes + "\nscore: (picker off)").strip()
best_chosen = takes_json[-1]["chosen"] if takes_json else {}
best_seed = takes_json[-1]["seed"] if takes_json else effective_seed
shot_list_json_obj = {
"world": world,
"take_count": take_count,
"seed_start": effective_seed,
"take_seed_mode": (take_seed_mode or "keep"),
"include_system_prompt": bool(include_system_prompt),
"separator": self._make_separator(separator, custom_separator),
"picker": {
"mode": (pick_mode or "off"),
"rule": (pick_rule or "balanced"),
"keyword": (pick_keyword or ""),
},
"takes": takes_json,
}
picked_json_obj = {
"picked_take_index": int(best_take_idx),
"picked_seed": int(best_seed),
"picked_score": (int(best_score) if picker_on else None),
shot_list_text = "\n\n".join(shot_list)
shot_list_json = self._dump_json({"takes": takes_json}, json_pretty)
picked_json = self._dump_json({
"picked_take": best_take,
"picked_prompt": best_prompt,
"picked_chosen": best_chosen,
"picked_final_prompt": best_final_prompt,
"picked_notes": best_notes,
}
"score": best_score,
}, json_pretty)
shot_list_json = self._dump_json(shot_list_json_obj, bool(json_pretty))
picked_json = self._dump_json(picked_json_obj, bool(json_pretty))
if save_to_file:
print("[Director][R11] save_to_file =", save_to_file, "save_dir =", save_dir, "save_format =", save_format, "save_prefix =", save_prefix)
print("[Director][R11] cwd =", os.getcwd())
print("[Director][R11] abs(save_dir) =", os.path.abspath(save_dir))
try:
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
out_dir = os.path.abspath(save_dir)
print("[Director][R11] creating dir:", out_dir)
os.makedirs(out_dir, exist_ok=True)
print("[Director][R11] dir exists:", os.path.isdir(out_dir))
base = f"{save_prefix}_{world}_{ts}".replace(" ", "_")
json_path = os.path.join(out_dir, base + "_shots.json")
picked_path = os.path.join(out_dir, base + "_picked.json")
print("[Director][R11] writing:", json_path)
with open(json_path, "w", encoding="utf-8") as f:
f.write(shot_list_json)
print("[Director][R11] writing:", picked_path)
with open(picked_path, "w", encoding="utf-8") as f:
f.write(picked_json)
if save_format == "json+txt":
txt_path = os.path.join(out_dir, base + "_shots.txt")
print("[Director][R11] writing:", txt_path)
with open(txt_path, "w", encoding="utf-8") as f:
f.write(shot_list_text)
print("[Director][R11] DONE ✅")
except Exception as e:
print("[Director][R11] ERROR ❌", repr(e))
os.makedirs(save_dir, exist_ok=True)
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
base = f"{save_prefix}_{world}_{ts}".replace(" ", "_")
with open(os.path.join(save_dir, base + "_shots.json"), "w", encoding="utf-8") as f:
f.write(shot_list_json)
with open(os.path.join(save_dir, base + "_picked.json"), "w", encoding="utf-8") as f:
f.write(picked_json)
if save_format == "json+txt":
with open(os.path.join(save_dir, base + "_shots.txt"), "w", encoding="utf-8") as f:
f.write(shot_list_text)
next_seed = self._compute_next_seed(seed, lock, control_after_generate)
return (
last_prompt,
last_system_prompt,
last_final_prompt,
last_notes,
int(next_seed),
shot_list,
best_final_prompt,
best_notes,
int(best_take_idx),
shot_list_json,
picked_json,
)
last_prompt,
last_system_prompt if include_system_prompt else "",
last_final_prompt,
last_notes,
next_seed,
shot_list_text,
best_prompt if best_prompt else last_final_prompt,
best_notes if best_notes else last_notes,
best_take if best_take != -1 else take_count,
shot_list_json,
picked_json,
)
nodes/prompt_enhancer.py
+391
View File
@@ -0,0 +1,391 @@
# PromptEnhancerNode (ComfyUI)
# File: ComfyUI/custom_nodes/PromptEnhancerNode/__init__.py
#
# pip install requests
#
# Backends:
# - Ollama: /api/chat (preferred) with fallback to /api/generate (fixes Qwen3 cases)
# - llama.cpp server: /v1/chat/completions (OpenAI-compatible)
#
# Notes:
# - Ollama host example: http://10.10.10.2:11434
# - llama.cpp host example: http://127.0.0.1:8080
from __future__ import annotations
import re
from typing import Any, Dict, Tuple, List, Optional
import requests
def _strip_think_blocks_safe(text: str) -> str:
"""
Removes <think>...</think> and <analysis>...</analysis> blocks.
If removal empties the output, returns original text (better than empty).
"""
if not text:
return ""
original = text.strip()
cleaned = re.sub(r"<think>.*?</think>", "", original, flags=re.S | re.I).strip()
cleaned = re.sub(r"<analysis>.*?</analysis>", "", cleaned, flags=re.S | re.I).strip()
return cleaned if cleaned else original
def _parse_stop_list(stop_csv: str) -> Optional[List[str]]:
stop_csv = (stop_csv or "").strip()
if not stop_csv:
return None
stops = [s.strip() for s in stop_csv.split(",") if s.strip()]
return stops or None
class PromptEnhancerNode:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"ENGINE": (["ollama", "llama.cpp_server"], {"default": "ollama"}),
"SYSTEM_PROMPT": (
"STRING",
{
"default": "You are a prompt enhancer. Return ONE single long, cinematic sentence in English, no lists, no line breaks.",
"multiline": True,
},
),
"USER_PROMPT": ("STRING", {"default": "", "multiline": True}),
"TEMPERATURE": ("FLOAT", {"default": 0.70, "min": 0.0, "max": 2.0, "step": 0.05}),
"TOP_P": ("FLOAT", {"default": 0.90, "min": 0.0, "max": 1.0, "step": 0.01}),
"MAX_TOKENS": ("INT", {"default": 240, "min": 1, "max": 4096, "step": 1}),
"STRIP_THINKING": ("BOOLEAN", {"default": True}),
},
"optional": {
# Ollama
"OLLAMA_HOST": ("STRING", {"default": "http://127.0.0.1:11434"}),
"OLLAMA_MODEL": ("STRING", {"default": "qwen3:8b"}),
# llama.cpp server
"LLAMA_HOST": ("STRING", {"default": "http://127.0.0.1:8080"}),
"LLAMA_MODEL": ("STRING", {"default": "local-model"}),
# Shared
"TIMEOUT": ("INT", {"default": 120, "min": 5, "max": 600, "step": 5}),
"SEED": ("INT", {"default": -1, "min": -1, "max": 2_147_483_647, "step": 1}),
"STOP": ("STRING", {"default": "", "multiline": False}),
# Debug
"DEBUG": ("BOOLEAN", {"default": False}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("ENHANCED_PROMPT",)
FUNCTION = "enhance"
CATEGORY = "TK/Prompt"
# -------------------------
# Ollama helpers
# -------------------------
def _ollama_chat(
self,
host: str,
model: str,
system_prompt: str,
user_prompt: str,
*,
options: Dict[str, Any],
timeout: int,
debug: bool,
) -> Tuple[str, str]:
host = (host or "").rstrip("/")
payload: Dict[str, Any] = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
"stream": False,
"options": options or {},
}
if debug:
print("[PFN Enhancer][Ollama][chat] POST", f"{host}/api/chat")
print("[PFN Enhancer][Ollama][chat] model:", model)
r = requests.post(f"{host}/api/chat", json=payload, timeout=int(timeout))
r.raise_for_status()
data = r.json()
if isinstance(data, dict) and data.get("error"):
return "", f"[OLLAMA CHAT ERROR] {data.get('error')}"
content = ""
if isinstance(data, dict):
content = ((data.get("message") or {}).get("content")) or ""
out = content.strip()
if debug:
print("[PFN Enhancer][Ollama][chat] out len:", len(out))
if isinstance(data, dict):
print("[PFN Enhancer][Ollama][chat] keys:", list(data.keys()))
return out, ""
def _ollama_generate(
self,
host: str,
model: str,
system_prompt: str,
user_prompt: str,
*,
options: Dict[str, Any],
timeout: int,
debug: bool,
) -> Tuple[str, str]:
host = (host or "").rstrip("/")
payload: Dict[str, Any] = {
"model": model,
"system": system_prompt,
"prompt": user_prompt,
"stream": False,
"options": options or {},
}
if debug:
print("[PFN Enhancer][Ollama][generate] POST", f"{host}/api/generate")
print("[PFN Enhancer][Ollama][generate] model:", model)
r = requests.post(f"{host}/api/generate", json=payload, timeout=int(timeout))
r.raise_for_status()
data = r.json()
if isinstance(data, dict) and data.get("error"):
return "", f"[OLLAMA GENERATE ERROR] {data.get('error')}"
content = ""
if isinstance(data, dict):
content = data.get("response") or ""
out = content.strip()
if debug:
print("[PFN Enhancer][Ollama][generate] out len:", len(out))
if isinstance(data, dict):
print("[PFN Enhancer][Ollama][generate] keys:", list(data.keys()))
return out, ""
def _enhance_with_ollama(
self,
host: str,
model: str,
system_prompt: str,
user_prompt: str,
*,
temperature: float,
top_p: float,
num_predict: int,
timeout: int,
seed: int,
stop: str,
debug: bool,
) -> str:
host = (host or "").rstrip("/")
url = f"{host}/api/chat"
options: Dict[str, Any] = {
"temperature": float(temperature),
"top_p": float(top_p),
"num_predict": int(num_predict),
}
if seed is not None and int(seed) >= 0:
options["seed"] = int(seed)
stops = _parse_stop_list(stop)
if stops:
options["stop"] = stops
payload: Dict[str, Any] = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
"stream": False,
"think": False, # ✅ CHIAVE per evitare thinking-only
"options": options,
}
if debug:
print("[Ollama] url:", url)
print("[Ollama] model:", model)
print("[Ollama] think:", payload.get("think"))
print("[Ollama] options:", payload.get("options"))
r = requests.post(url, json=payload, timeout=int(timeout))
r.raise_for_status()
data = r.json()
# ✅ prende SOLO content
out = ((data.get("message") or {}).get("content") or "").strip()
# Se content è vuoto, ritorniamo un errore informativo (e NON silent empty)
if not out:
msg = data.get("message") or {}
return (
"[EMPTY OUTPUT] content empty.\n"
f"done_reason={data.get('done_reason')}\n"
f"content_len={len((msg.get('content') or ''))} thinking_len={len((msg.get('thinking') or ''))}\n"
"Tip: ensure payload includes top-level think:false and num_predict is high enough."
)
return out
# -------------------------
# llama.cpp server backend (OpenAI-compatible)
# -------------------------
def _enhance_with_llama_server(
self,
host: str,
model: str,
system_prompt: str,
user_prompt: str,
*,
temperature: float,
top_p: float,
max_tokens: int,
timeout: int,
seed: int,
stop: str,
debug: bool,
) -> str:
host = (host or "").rstrip("/")
payload: Dict[str, Any] = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
"temperature": float(temperature),
"top_p": float(top_p),
"max_tokens": int(max_tokens),
}
if seed is not None and int(seed) >= 0:
payload["seed"] = int(seed)
stops = _parse_stop_list(stop)
if stops:
payload["stop"] = stops
if debug:
print("[PFN Enhancer][llama.cpp] POST", f"{host}/v1/chat/completions")
print("[PFN Enhancer][llama.cpp] model:", model)
r = requests.post(f"{host}/v1/chat/completions", json=payload, timeout=int(timeout))
r.raise_for_status()
data = r.json()
try:
choices = data.get("choices") or []
if not choices:
return "[EMPTY OUTPUT] llama.cpp returned no choices."
msg = (choices[0].get("message") or {})
out = (msg.get("content") or "").strip()
except Exception:
return f"[PARSE ERROR] Unexpected llama.cpp response shape: {str(data)[:300]}"
if debug:
print("[PFN Enhancer][llama.cpp] out len:", len(out))
return out if out else "[EMPTY OUTPUT] llama.cpp returned empty content."
# -------------------------
# Node entry point (kwargs = anti-mismatch)
# -------------------------
def enhance(self, **kwargs) -> Tuple[str]:
ENGINE = kwargs.get("ENGINE", "ollama")
SYSTEM_PROMPT = (kwargs.get("SYSTEM_PROMPT", "") or "").strip()
USER_PROMPT = (kwargs.get("USER_PROMPT", "") or "").strip()
TEMPERATURE = float(kwargs.get("TEMPERATURE", 0.7))
TOP_P = float(kwargs.get("TOP_P", 0.9))
MAX_TOKENS = int(kwargs.get("MAX_TOKENS", 240))
STRIP_THINKING = bool(kwargs.get("STRIP_THINKING", True))
OLLAMA_HOST = kwargs.get("OLLAMA_HOST", "http://127.0.0.1:11434")
OLLAMA_MODEL = kwargs.get("OLLAMA_MODEL", "qwen3:8b")
LLAMA_HOST = kwargs.get("LLAMA_HOST", "http://127.0.0.1:8080")
LLAMA_MODEL = kwargs.get("LLAMA_MODEL", "local-model")
TIMEOUT = int(kwargs.get("TIMEOUT", 120))
SEED = int(kwargs.get("SEED", -1))
STOP = (kwargs.get("STOP", "") or "").strip()
DEBUG = bool(kwargs.get("DEBUG", False))
if DEBUG:
print("[PFN Enhancer] ENGINE:", ENGINE)
print("[PFN Enhancer] USER_PROMPT len:", len(USER_PROMPT))
print("[PFN Enhancer] SYSTEM_PROMPT len:", len(SYSTEM_PROMPT))
print("[PFN Enhancer] OLLAMA:", OLLAMA_HOST, OLLAMA_MODEL)
print("[PFN Enhancer] LLAMA :", LLAMA_HOST, LLAMA_MODEL)
if not USER_PROMPT:
return ("[EMPTY INPUT] USER_PROMPT is empty (possible name mismatch or upstream node output).",)
if not SYSTEM_PROMPT:
SYSTEM_PROMPT = "You are a prompt enhancer. Return ONE single long sentence."
try:
if ENGINE == "ollama":
out = self._enhance_with_ollama(
OLLAMA_HOST,
OLLAMA_MODEL,
SYSTEM_PROMPT,
USER_PROMPT,
temperature=TEMPERATURE,
top_p=TOP_P,
num_predict=MAX_TOKENS,
timeout=TIMEOUT,
seed=SEED,
stop=STOP,
debug=DEBUG,
)
else:
out = self._enhance_with_llama_server(
LLAMA_HOST,
LLAMA_MODEL,
SYSTEM_PROMPT,
USER_PROMPT,
temperature=TEMPERATURE,
top_p=TOP_P,
max_tokens=MAX_TOKENS,
timeout=TIMEOUT,
seed=SEED,
stop=STOP,
debug=DEBUG,
)
except requests.exceptions.RequestException as e:
return (f"[NETWORK ERROR] {type(e).__name__}: {e}",)
except Exception as e:
return (f"[PromptEnhancerNode ERROR] {type(e).__name__}: {e}",)
out = (out or "").strip()
if STRIP_THINKING:
out = _strip_think_blocks_safe(out)
if not out:
out = "[EMPTY OUTPUT] Output became empty after processing (try DEBUG=true)."
return (out,)
nodes/session_merge.py
+169
View File
@@ -0,0 +1,169 @@
from __future__ import annotations
import json
import os
from datetime import datetime
from typing import Any, Dict, List, Tuple, Optional
class DirectorSessionMerge:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
# percorsi separati da newline (più comodo di una lista UI)
"shots_json_paths": ("STRING", {"default": "logs/a_shots.json\nlogs/b_shots.json"}),
"pretty_json": ("BOOLEAN", {"default": True}),
"recompute_picked": ("BOOLEAN", {"default": True}),
"save_merged": ("BOOLEAN", {"default": False}),
"save_dir": ("STRING", {"default": "logs"}),
"save_prefix": ("STRING", {"default": "director_merge"}),
}
}
RETURN_TYPES = ("STRING", "STRING", "INT", "INT")
RETURN_NAMES = ("merged_shot_list_json", "merged_picked_json", "merged_take_count", "picked_take_index")
FUNCTION = "run"
CATEGORY = "PFN/Director"
def _dump(self, obj: Any, pretty: bool) -> str:
if pretty:
return json.dumps(obj, ensure_ascii=False, indent=2)
return json.dumps(obj, ensure_ascii=False, separators=(",", ":"))
def _load_json(self, path: str) -> Tuple[Optional[Dict[str, Any]], Optional[str]]:
p = os.path.abspath(path.strip())
if not p or not os.path.isfile(p):
return None, f"(file not found) {p}"
try:
with open(p, "r", encoding="utf-8") as f:
return json.load(f), None
except Exception as e:
return None, f"(invalid json) {p} :: {repr(e)}"
def _best_take_by_score(self, takes: List[Dict[str, Any]]) -> Tuple[int, Optional[int]]:
# return (index1based, score or None)
best_i = 1
best_score: Optional[int] = None
for i, t in enumerate(takes, start=1):
sc = t.get("score", None)
if sc is None:
continue
try:
sc_int = int(sc)
except Exception:
continue
if best_score is None or sc_int > best_score:
best_score = sc_int
best_i = i
return best_i, best_score
def run(
self,
shots_json_paths: str,
pretty_json: bool,
recompute_picked: bool,
save_merged: bool,
save_dir: str,
save_prefix: str,
):
raw = shots_json_paths or ""
raw = raw.replace("\n", ";")
paths = [p.strip() for p in raw.split(";") if p.strip() and not p.strip().startswith("#")]
if not paths:
msg = "(no paths provided)"
return (msg, "{}", 0, -1)
merged_takes: List[Dict[str, Any]] = []
load_errors: List[str] = []
# Keep some top-level info (best effort)
meta_worlds: List[str] = []
meta_files: List[str] = []
for path in paths:
data, err = self._load_json(path)
if err:
load_errors.append(err)
continue
assert data is not None
takes = data.get("takes", [])
if not isinstance(takes, list):
load_errors.append(f"(bad takes array) {path}")
continue
meta_worlds.append(str(data.get("world", "")) if "world" in data else "")
meta_files.append(os.path.abspath(path))
for t in takes:
if not isinstance(t, dict):
continue
merged_takes.append({
**t,
"source_file": os.path.abspath(path),
})
# If nothing loaded
if not merged_takes:
msg = "(no takes loaded)\n" + "\n".join(load_errors[:50])
return (msg, "{}", 0, -1)
# Re-number takes + normalize key name
for i, t in enumerate(merged_takes, start=1):
t["take_index"] = i
take_count = len(merged_takes)
# Compute picked
picked_idx = 1
picked_score: Optional[int] = None
if recompute_picked:
picked_idx, picked_score = self._best_take_by_score(merged_takes)
picked_take = merged_takes[picked_idx - 1]
picked_json_obj = {
"picked_take_index": int(picked_idx),
"picked_score": picked_score,
"picked_seed": picked_take.get("seed", None),
"picked_source_file": picked_take.get("source_file", ""),
"picked_chosen": picked_take.get("chosen", {}),
"picked_final_prompt": picked_take.get("final_prompt", ""),
"picked_notes": picked_take.get("notes", ""),
}
merged_json_obj: Dict[str, Any] = {
"type": "PFN_Director_MergedSession",
"created_at": datetime.now().isoformat(timespec="seconds"),
"source_files": meta_files,
"worlds": [w for w in meta_worlds if w],
"take_count": take_count,
"takes": merged_takes,
}
# Append errors (useful in debug)
if load_errors:
merged_json_obj["load_errors"] = load_errors
merged_shot_list_json = self._dump(merged_json_obj, bool(pretty_json))
merged_picked_json = self._dump(picked_json_obj, bool(pretty_json))
# Optional save
if save_merged:
out_dir = os.path.abspath(save_dir)
os.makedirs(out_dir, exist_ok=True)
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
base = f"{save_prefix}_{ts}".replace(" ", "_")
merged_path = os.path.join(out_dir, base + "_merged_shots.json")
picked_path = os.path.join(out_dir, base + "_merged_picked.json")
with open(merged_path, "w", encoding="utf-8") as f:
f.write(merged_shot_list_json)
with open(picked_path, "w", encoding="utf-8") as f:
f.write(merged_picked_json)
return (merged_shot_list_json, merged_picked_json, int(take_count), int(picked_idx))