diff --git a/.ruff.toml b/.ruff.toml index 48f2e9026..6c77aa6f3 100644 --- a/.ruff.toml +++ b/.ruff.toml @@ -83,6 +83,7 @@ ignore = [ "F401", # Imported by unused "NPY002", # replace legacy random "RUF005", # Consider iterable unpacking + "RUF008", # Do not use mutable default values for dataclass "RUF010", # Use explicit conversion flag "RUF012", # Mutable class attributes "RUF013", # PEP 484 prohibits implicit `Optional` diff --git a/CHANGELOG.md b/CHANGELOG.md index 0dc1b5d8f..18e2617cb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -79,6 +79,13 @@ Pretty big performance updates to a) Any model using DiT based architecture: new - [ByteDance/Sa2VA](https://huggingface.co/ByteDance/Sa2VA-1B) 1B, 4B simply select from list of available models in caption tab - add option to set system prompt for vlm models that support it: *Gemma, Smol, Qwen* +- **Prompt Enhance** + - new built-in extension available in text/image/control tabs + - can be used to manually or automatically enhance prompts using LLM + - supports **Gemma-3, Qwen-2.5, Phi-4, Llama-3.2, SmolLM2** + models are auto-downloaded on first use + also supports custom models that are compatible with `transformers/AutoModelForCausalLM` + - support quantization and offloading - [NudeNet](https://github.com/vladmandic/sd-extension-nudenet/) extension updates - add detection of prompt language and alphabet and filter based on those values - add image policy checks using `LlavaGuard` VLM to detect policy violations (and reasons) diff --git a/extensions-builtin/sdnext-modernui b/extensions-builtin/sdnext-modernui index d41350e07..1e38e9b56 160000 --- a/extensions-builtin/sdnext-modernui +++ b/extensions-builtin/sdnext-modernui @@ -1 +1 @@ -Subproject commit d41350e07f0c94e733c5ffbda8d64b07e313eea3 +Subproject commit 1e38e9b56edf45dd17402aee2a1c281dc26b5286 diff --git a/javascript/sdnext.css b/javascript/sdnext.css index e1425271d..9c858e37b 100644 --- a/javascript/sdnext.css +++ b/javascript/sdnext.css @@ -116,6 +116,8 @@ button.custom-button { border-radius: var(--button-large-radius); padding: var(- #txt2img_seed, #img2img_seed, #control_seed, #video_seed { min-width: 90px !important } #video_generate_box>button { max-width: unset; } #interrogate_output_prompt>textarea { resize: vertical; } +#prompt_enhance_apply, #prompt_enhance_model { max-width: unset; } +#prompt_enhance_system textarea { color: var(--body-text-color-subdued) !important } .interrogate { position: absolute; right: 2.8em; top: 0.2em; max-width: fit-content; background: none !important; z-index: 50; font-size: 1.5em !important; } .interrogate:hover { background: var(--button-primary-background-fill-hover) !important; } diff --git a/modules/interrogate/deepbooru.py b/modules/interrogate/deepbooru.py index 30227dc21..1e47e6cc8 100644 --- a/modules/interrogate/deepbooru.py +++ b/modules/interrogate/deepbooru.py @@ -4,7 +4,7 @@ import threading import torch import numpy as np from PIL import Image -from modules import modelloader, paths, devices, shared +from modules import modelloader, paths, devices, shared, sd_models re_special = re.compile(r'([\\()])') load_lock = threading.Lock() @@ -35,11 +35,11 @@ class DeepDanbooru: def start(self): self.load() - self.model.to(devices.device) + sd_models.move_model(self.model, devices.device) def stop(self): if shared.opts.interrogate_offload: - self.model.to(devices.cpu) + sd_models.move_model(self.model, devices.cpu) devices.torch_gc() def tag(self, pil_image): diff --git a/modules/interrogate/deepseek.py b/modules/interrogate/deepseek.py index 5138c5693..b2d340248 100644 --- a/modules/interrogate/deepseek.py +++ b/modules/interrogate/deepseek.py @@ -12,7 +12,7 @@ import os import sys import importlib from transformers import AutoModelForCausalLM -from modules import shared, devices, paths +from modules import shared, devices, paths, sd_models # model_path = "deepseek-ai/deepseek-vl2-small" @@ -73,7 +73,7 @@ def predict(question, image, repo): ).to(device=devices.device, dtype=devices.dtype) inputs_embeds = vl_gpt.prepare_inputs_embeds(**prepare_inputs) inputs_embeds = inputs_embeds.to(device=devices.device, dtype=devices.dtype) - vl_gpt = vl_gpt.to(devices.device) + sd_models.move_model(vl_gpt, devices.device) with devices.inference_context(): outputs = vl_gpt.language.generate( inputs_embeds=inputs_embeds, diff --git a/modules/interrogate/openclip.py b/modules/interrogate/openclip.py index 761c0fa39..792b4df85 100644 --- a/modules/interrogate/openclip.py +++ b/modules/interrogate/openclip.py @@ -10,7 +10,7 @@ import gradio as gr from PIL import Image from torchvision import transforms from torchvision.transforms.functional import InterpolationMode -from modules import devices, paths, shared, lowvram, errors +from modules import devices, paths, shared, lowvram, errors, sd_models caption_models = { @@ -125,7 +125,7 @@ class InterrogateModels: else: model, preprocess = clip.load(clip_model_name, download_root=shared.opts.clip_models_path) model.eval() - model = model.to(devices.device) + sd_models.move_model(model, devices.device) return model, preprocess def load(self): @@ -133,23 +133,23 @@ class InterrogateModels: self.blip_model = self.load_blip_model() if not shared.opts.no_half and not self.running_on_cpu: self.blip_model = self.blip_model.half() - self.blip_model = self.blip_model.to(devices.device) if self.clip_model is None: self.clip_model, self.clip_preprocess = self.load_clip_model() if not shared.opts.no_half and not self.running_on_cpu: self.clip_model = self.clip_model.half() - self.clip_model = self.clip_model.to(devices.device) self.dtype = next(self.clip_model.parameters()).dtype + sd_models.move_model(self.blip_model, devices.device) + sd_models.move_model(self.clip_model, devices.device) def send_clip_to_ram(self): if shared.opts.interrogate_offload: if self.clip_model is not None: - self.clip_model = self.clip_model.to(devices.cpu) + sd_models.move_model(self.blip_model, devices.cpu) def send_blip_to_ram(self): if shared.opts.interrogate_offload: if self.blip_model is not None: - self.blip_model = self.blip_model.to(devices.cpu) + sd_models.move_model(self.blip_model, devices.cpu) def unload(self): self.send_clip_to_ram() @@ -291,8 +291,8 @@ def load_interrogator(clip_model, blip_model): def unload_clip_model(): if ci is not None and shared.opts.interrogate_offload: - ci.caption_model = ci.caption_model.to(devices.cpu) - ci.clip_model = ci.clip_model.to(devices.cpu) + sd_models.move_model(ci.caption_model, devices.cpu) + sd_models.move_model(ci.clip_model, devices.cpu) ci.caption_offloaded = True ci.clip_offloaded = True devices.torch_gc() diff --git a/modules/interrogate/vqa.py b/modules/interrogate/vqa.py index 3db5251e8..b7f6aabb3 100644 --- a/modules/interrogate/vqa.py +++ b/modules/interrogate/vqa.py @@ -7,7 +7,8 @@ import torch import transformers import transformers.dynamic_module_utils from PIL import Image -from modules import shared, devices, errors +from modules import shared, devices, errors, sd_models + processor = None model = None @@ -74,7 +75,7 @@ def b64(image): def clean(response, question): - strip = ['---', '\r', '\t', '**', '"', '“', '”', 'Assistant:', 'Caption:', '<|im_end|>'] + strip = ['---', '\r', '\t', '**', '"', '“', '”', 'Assistant:', 'Caption:', '<|im_end|>', ''] if isinstance(response, dict): if 'task' in response: response = response['task'] @@ -113,13 +114,16 @@ def qwen(question: str, image: Image.Image, repo: str = None, system_prompt: str global processor, model, loaded # pylint: disable=global-statement if model is None or loaded != repo: shared.log.debug(f'Interrogate load: vlm="{repo}"') + model = None model = transformers.Qwen2VLForConditionalGeneration.from_pretrained( repo, cache_dir=shared.opts.hfcache_dir ) + model = model.to(devices.device, devices.dtype) processor = transformers.AutoProcessor.from_pretrained(repo, cache_dir=shared.opts.hfcache_dir) loaded = repo - model = model.to(devices.device, devices.dtype) + devices.torch_gc() + sd_models.move_model(model, devices.device) question = question.replace('<', '').replace('>', '').replace('_', ' ') system_prompt = system_prompt or shared.opts.vlm_system conversation = [ @@ -157,10 +161,13 @@ def gemma(question: str, image: Image.Image, repo: str = None, system_prompt: st return '' if model is None or loaded != repo: shared.log.debug(f'Interrogate load: vlm="{repo}"') + model = None model = transformers.Gemma3ForConditionalGeneration.from_pretrained(repo, cache_dir=shared.opts.hfcache_dir) + model = model.to(devices.device, devices.dtype) processor = transformers.AutoProcessor.from_pretrained(repo, cache_dir=shared.opts.hfcache_dir) loaded = repo - model = model.to(devices.device, devices.dtype) + devices.torch_gc() + sd_models.move_model(model, devices.device) question = question.replace('<', '').replace('>', '').replace('_', ' ') system_prompt = system_prompt or shared.opts.vlm_system conversation = [ @@ -199,13 +206,16 @@ def paligemma(question: str, image: Image.Image, repo: str = None): if model is None or loaded != repo: shared.log.debug(f'Interrogate load: vlm="{repo}"') processor = transformers.PaliGemmaProcessor.from_pretrained(repo, cache_dir=shared.opts.hfcache_dir) + model = None model = transformers.PaliGemmaForConditionalGeneration.from_pretrained( repo, cache_dir=shared.opts.hfcache_dir, torch_dtype=devices.dtype, ) + model = model.to(devices.device, devices.dtype) loaded = repo - model = model.to(devices.device, devices.dtype) + devices.torch_gc() + sd_models.move_model(model, devices.device) question = question.replace('<', '').replace('>', '').replace('_', ' ') model_inputs = processor(text=question, images=image, return_tensors="pt").to(devices.device, devices.dtype) input_len = model_inputs["input_ids"].shape[-1] @@ -228,6 +238,7 @@ def ovis(question: str, image: Image.Image, repo: str = None): global model, loaded # pylint: disable=global-statement if model is None or loaded != repo: shared.log.debug(f'Interrogate load: vlm="{repo}"') + model = None model = transformers.AutoModelForCausalLM.from_pretrained( repo, torch_dtype=devices.dtype, @@ -235,8 +246,10 @@ def ovis(question: str, image: Image.Image, repo: str = None): trust_remote_code=True, cache_dir=shared.opts.hfcache_dir, ) + model = model.to(devices.device, devices.dtype) loaded = repo - model = model.to(devices.device, devices.dtype) + devices.torch_gc() + sd_models.move_model(model, devices.device) text_tokenizer = model.get_text_tokenizer() visual_tokenizer = model.get_visual_tokenizer() max_partition = 9 @@ -268,15 +281,18 @@ def smol(question: str, image: Image.Image, repo: str = None, system_prompt: str global processor, model, loaded # pylint: disable=global-statement if model is None or loaded != repo: shared.log.debug(f'Interrogate load: vlm="{repo}"') + model = None model = transformers.AutoModelForVision2Seq.from_pretrained( repo, cache_dir=shared.opts.hfcache_dir, torch_dtype=devices.dtype, _attn_implementation="eager", ) + model.to(devices.device, devices.dtype) processor = transformers.AutoProcessor.from_pretrained(repo, cache_dir=shared.opts.hfcache_dir) loaded = repo - model.to(devices.device, devices.dtype) + devices.torch_gc() + sd_models.move_model(model, devices.device) question = question.replace('<', '').replace('>', '').replace('_', ' ') system_prompt = system_prompt or shared.opts.vlm_system conversation = [ @@ -307,13 +323,16 @@ def git(question: str, image: Image.Image, repo: str = None): global processor, model, loaded # pylint: disable=global-statement if model is None or loaded != repo: shared.log.debug(f'Interrogate load: vlm="{repo}"') + model = None model = transformers.GitForCausalLM.from_pretrained( repo, cache_dir=shared.opts.hfcache_dir, ) + model.to(devices.device, devices.dtype) processor = transformers.GitProcessor.from_pretrained(repo, cache_dir=shared.opts.hfcache_dir) loaded = repo - model.to(devices.device, devices.dtype) + devices.torch_gc() + sd_models.move_model(model, devices.device) pixel_values = processor(images=image, return_tensors="pt").pixel_values git_dict = {} git_dict['pixel_values'] = pixel_values.to(devices.device, devices.dtype) @@ -332,13 +351,16 @@ def blip(question: str, image: Image.Image, repo: str = None): global processor, model, loaded # pylint: disable=global-statement if model is None or loaded != repo: shared.log.debug(f'Interrogate load: vlm="{repo}"') + model = None model = transformers.BlipForQuestionAnswering.from_pretrained( repo, cache_dir=shared.opts.hfcache_dir, ) + model.to(devices.device, devices.dtype) processor = transformers.BlipProcessor.from_pretrained(repo, cache_dir=shared.opts.hfcache_dir) loaded = repo - model.to(devices.device, devices.dtype) + devices.torch_gc() + sd_models.move_model(model, devices.device) inputs = processor(image, question, return_tensors="pt") inputs = inputs.to(devices.device, devices.dtype) with devices.inference_context(): @@ -351,13 +373,16 @@ def vilt(question: str, image: Image.Image, repo: str = None): global processor, model, loaded # pylint: disable=global-statement if model is None or loaded != repo: shared.log.debug(f'Interrogate load: vlm="{repo}"') + model = None model = transformers.ViltForQuestionAnswering.from_pretrained( repo, cache_dir=shared.opts.hfcache_dir, ) + model.to(devices.device) processor = transformers.ViltProcessor.from_pretrained(repo, cache_dir=shared.opts.hfcache_dir) loaded = repo - model.to(devices.device) + devices.torch_gc() + sd_models.move_model(model, devices.device) inputs = processor(image, question, return_tensors="pt") inputs = inputs.to(devices.device) with devices.inference_context(): @@ -372,13 +397,16 @@ def pix(question: str, image: Image.Image, repo: str = None): global processor, model, loaded # pylint: disable=global-statement if model is None or loaded != repo: shared.log.debug(f'Interrogate load: vlm="{repo}"') + model = None model = transformers.Pix2StructForConditionalGeneration.from_pretrained( repo, cache_dir=shared.opts.hfcache_dir, ) + model.to(devices.device) processor = transformers.Pix2StructProcessor.from_pretrained(repo, cache_dir=shared.opts.hfcache_dir) loaded = repo - model.to(devices.device) + devices.torch_gc() + sd_models.move_model(model, devices.device) if len(question) > 0: inputs = processor(images=image, text=question, return_tensors="pt").to(devices.device) else: @@ -393,6 +421,7 @@ def moondream(question: str, image: Image.Image, repo: str = None): global processor, model, loaded # pylint: disable=global-statement if model is None or loaded != repo: shared.log.debug(f'Interrogate load: vlm="{repo}"') + model = None model = transformers.AutoModelForCausalLM.from_pretrained( repo, revision="2024-08-26", @@ -401,8 +430,10 @@ def moondream(question: str, image: Image.Image, repo: str = None): ) processor = transformers.AutoTokenizer.from_pretrained(repo, cache_dir=shared.opts.hfcache_dir) loaded = repo + model.to(devices.device, devices.dtype) model.eval() - model.to(devices.device, devices.dtype) + devices.torch_gc() + sd_models.move_model(model, devices.device) question = question.replace('<', '').replace('>', '').replace('_', ' ') encoded = model.encode_image(image) with devices.inference_context(): @@ -424,6 +455,7 @@ def florence(question: str, image: Image.Image, repo: str = None, revision: str if model is None or loaded != repo: shared.log.debug(f'Interrogate load: vlm="{repo}" path="{shared.opts.hfcache_dir}"') transformers.dynamic_module_utils.get_imports = get_imports + model = None model = transformers.AutoModelForCausalLM.from_pretrained( repo, trust_remote_code=True, @@ -433,8 +465,10 @@ def florence(question: str, image: Image.Image, repo: str = None, revision: str processor = transformers.AutoProcessor.from_pretrained(repo, trust_remote_code=True, revision=revision, cache_dir=shared.opts.hfcache_dir) transformers.dynamic_module_utils.get_imports = _get_imports loaded = repo + model.to(devices.device, devices.dtype) model.eval() - model.to(devices.device, devices.dtype) + devices.torch_gc() + sd_models.move_model(model, devices.device) if question.startswith('<'): task = question.split('>', 1)[0] + '>' else: @@ -456,12 +490,14 @@ def florence(question: str, image: Image.Image, repo: str = None, revision: str def sa2(question: str, image: Image.Image, repo: str = None): global processor, model, loaded # pylint: disable=global-statement if model is None or loaded != repo: + model = None model = transformers.AutoModel.from_pretrained( repo, torch_dtype=devices.dtype, low_cpu_mem_usage=True, use_flash_attn=False, trust_remote_code=True) + model = model.to(devices.device, devices.dtype) model = model.eval() processor = transformers.AutoTokenizer.from_pretrained( repo, @@ -469,7 +505,8 @@ def sa2(question: str, image: Image.Image, repo: str = None): use_fast=False, ) loaded = repo - model = model.to(devices.device, devices.dtype) + devices.torch_gc() + sd_models.move_model(model, devices.device) if question.startswith('<'): task = question.split('>', 1)[0] + '>' else: @@ -559,7 +596,7 @@ def interrogate(question, system_prompt, prompt, image, model_name, quiet:bool=F errors.display(e, 'VQA') answer = 'error' if shared.opts.interrogate_offload and model is not None: - model.to(devices.cpu) + sd_models.move_model(model, devices.cpu) devices.torch_gc() answer = clean(answer, question) t1 = time.time() diff --git a/modules/model_quant.py b/modules/model_quant.py index 0430dc9e0..e0fad40f9 100644 --- a/modules/model_quant.py +++ b/modules/model_quant.py @@ -45,7 +45,7 @@ def create_bnb_config(kwargs = None, allow_bnb: bool = True, module: str = 'Mode bnb_4bit_quant_type=shared.opts.bnb_quantization_type, bnb_4bit_compute_dtype=devices.dtype ) - log.debug(f'Quantization: module=all type=bnb dtype={shared.opts.bnb_quantization_type} storage={shared.opts.bnb_quantization_storage}') + log.debug(f'Quantization: module="{module}" type=bnb dtype={shared.opts.bnb_quantization_type} storage={shared.opts.bnb_quantization_storage}') if kwargs is None: return bnb_config else: @@ -62,7 +62,7 @@ def create_ao_config(kwargs = None, allow_ao: bool = True, module: str = 'Model' if ao is None: return kwargs ao_config = diffusers.TorchAoConfig(shared.opts.torchao_quantization_type) - log.debug(f'Quantization: module=all type=torchao dtype={shared.opts.torchao_quantization_type}') + log.debug(f'Quantization: module="{module}" type=torchao dtype={shared.opts.torchao_quantization_type}') if kwargs is None: return ao_config else: @@ -82,7 +82,7 @@ def create_quanto_config(kwargs = None, allow_quanto: bool = True, module: str = weights_dtype=shared.opts.quanto_quantization_type, ) quanto_config.activations = None # patch so it works with transformers - log.debug(f'Quantization: module=all type=quanto dtype={shared.opts.quanto_quantization_type}') + log.debug(f'Quantization: module="{module}" type=quanto dtype={shared.opts.quanto_quantization_type}') if kwargs is None: return quanto_config else: diff --git a/modules/processing_class.py b/modules/processing_class.py index eb4e333ec..e38a44fd9 100644 --- a/modules/processing_class.py +++ b/modules/processing_class.py @@ -111,6 +111,8 @@ class StableDiffusionProcessing: refiner_prompt: str = '', refiner_negative: str = '', hr_refiner_start: float = 0, + # prompt enhancer + enhance_prompt: bool = False, # save options outpath_samples=None, outpath_grids=None, @@ -145,6 +147,7 @@ class StableDiffusionProcessing: self.is_refiner_pass = False self.is_api = False self.scheduled_prompt = False + self.enhance_prompt = enhance_prompt self.prompt_embeds = [] self.positive_pooleds = [] self.negative_embeds = [] diff --git a/modules/shared.py b/modules/shared.py index 95a205e43..f4967cf93 100644 --- a/modules/shared.py +++ b/modules/shared.py @@ -509,7 +509,7 @@ options_templates.update(options_section(('backends', "Backend Settings"), { options_templates.update(options_section(('quantization', "Quantization Settings"), { "bnb_quantization_sep": OptionInfo("

BitsAndBytes

", "", gr.HTML), - "bnb_quantization": OptionInfo([], "Quantization enabled", gr.CheckboxGroup, {"choices": ["Model", "VAE", "Text Encoder"], "visible": native}), + "bnb_quantization": OptionInfo([], "Quantization enabled", gr.CheckboxGroup, {"choices": ["Model", "VAE", "Text Encoder", "LLM"], "visible": native}), "bnb_quantization_type": OptionInfo("nf4", "Quantization type", gr.Dropdown, {"choices": ['nf4', 'fp8', 'fp4'], "visible": native}), "bnb_quantization_storage": OptionInfo("uint8", "Backend storage", gr.Dropdown, {"choices": ["float16", "float32", "int8", "uint8", "float64", "bfloat16"], "visible": native}), diff --git a/scripts/flux_prompt_enhance.py b/scripts/flux_prompt_enhance.py new file mode 100644 index 000000000..17613964a --- /dev/null +++ b/scripts/flux_prompt_enhance.py @@ -0,0 +1,99 @@ +# repo: https://huggingface.co/gokaygokay/Flux-Prompt-Enhance + +import time +import random +import threading +from transformers import AutoTokenizer, AutoModelForSeq2SeqLM +import gradio as gr +from modules import shared, scripts, devices, processing + + +repo_id = "gokaygokay/Flux-Prompt-Enhance" +num_return_sequences = 5 +load_lock = threading.Lock() + + +class Script(scripts.Script): + prompts = [['']] + tokenizer: AutoTokenizer = None + model: AutoModelForSeq2SeqLM = None + prefix: str = "enhance prompt: " + button: gr.Button = None + auto_apply: gr.Checkbox = None + max_length: gr.Slider = None + temperature: gr.Slider = None + repetition_penalty: gr.Slider = None + table: gr.DataFrame = None + prompt: gr.Textbox = None + + def title(self): + return 'Prompt enhance' + + def show(self, is_img2img): + return shared.native + + def load(self): + with load_lock: + if self.tokenizer is None: + self.tokenizer = AutoTokenizer.from_pretrained('gokaygokay/Flux-Prompt-Enhance', cache_dir=shared.opts.hfcache_dir) + if self.model is None: + shared.log.info(f'Prompt enhance: model="{repo_id}"') + self.model = AutoModelForSeq2SeqLM.from_pretrained('gokaygokay/Flux-Prompt-Enhance', cache_dir=shared.opts.hfcache_dir).to(device=devices.cpu, dtype=devices.dtype) + + def enhance(self, prompt, auto_apply: bool = False, temperature: float = 0.7, repetition_penalty: float = 1.2, max_length: int = 128): + self.load() + t0 = time.time() + input_text = self.prefix + prompt + input_ids = self.tokenizer(input_text, return_tensors="pt").input_ids.to(devices.device) + self.model = self.model.to(devices.device) + kwargs = { + 'max_length': int(max_length), + 'num_return_sequences': int(num_return_sequences), + 'do_sample': True, + 'temperature': float(temperature), + 'repetition_penalty': float(repetition_penalty), + } + try: + outputs = self.model.generate(input_ids, **kwargs) + except Exception as e: + shared.log.error(f'Prompt enhance: error="{e}"') + return [['']] + self.model = self.model.to(devices.cpu) + prompts = self.tokenizer.batch_decode(outputs, skip_special_tokens=True) + prompts = [[p] for p in prompts] + t1 = time.time() + shared.log.info(f'Prompt enhance: temperature={temperature} repetition={repetition_penalty} length={max_length} sequences={num_return_sequences} apply={auto_apply} time={t1-t0:.2f}s') + return prompts + + def select(self, cell: gr.SelectData, _table): + prompt = cell.value if hasattr(cell, 'value') else cell + shared.log.info(f'Prompt enhance: prompt="{prompt}"') + return prompt + + def ui(self, _is_img2img): + with gr.Row(): + self.button = gr.Button(value='Enhance prompt') + self.auto_apply = gr.Checkbox(label='Auto apply', default=False) + with gr.Row(): + self.max_length = gr.Slider(label='Length', minimum=64, maximum=512, step=1, value=128) + self.temperature = gr.Slider(label='Temperature', minimum=0.1, maximum=2.0, step=0.05, value=0.7) + self.repetition_penalty = gr.Slider(label='Penalty', minimum=0.1, maximum=2.0, step=0.05, value=1.2) + with gr.Row(): + self.table = gr.DataFrame(self.prompts, label='', show_label=False, interactive=False, wrap=True, datatype="str", col_count=1, max_rows=num_return_sequences, headers=['Prompts']) + + if self.prompt is not None: + self.button.click(fn=self.enhance, inputs=[self.prompt, self.auto_apply, self.temperature, self.repetition_penalty, self.max_length], outputs=[self.table]) + self.table.select(fn=self.select, inputs=[self.table], outputs=[self.prompt]) + return [self.auto_apply, self.temperature, self.repetition_penalty, self.max_length] + + def run(self, p: processing.StableDiffusionProcessing, auto_apply, temperature, repetition_penalty, max_length): # pylint: disable=arguments-differ + if auto_apply: + p.prompt = shared.prompt_styles.apply_styles_to_prompt(p.prompt, p.styles) + shared.log.debug(f'Prompt enhance: source="{p.prompt}"') + prompts = self.enhance(p.prompt, auto_apply, temperature, repetition_penalty, max_length) + p.prompt = random.choice(prompts)[0] + shared.log.debug(f'Prompt enhance: prompt="{p.prompt}"') + + def after_component(self, component, **kwargs): # searching for actual ui prompt components + if getattr(component, 'elem_id', '') in ['txt2img_prompt', 'img2img_prompt', 'control_prompt', 'video_prompt']: + self.prompt = component diff --git a/scripts/prompt_enhance.py b/scripts/prompt_enhance.py index 17613964a..8d38f5ef1 100644 --- a/scripts/prompt_enhance.py +++ b/scripts/prompt_enhance.py @@ -1,99 +1,193 @@ -# repo: https://huggingface.co/gokaygokay/Flux-Prompt-Enhance - +from dataclasses import dataclass +import re import time -import random -import threading -from transformers import AutoTokenizer, AutoModelForSeq2SeqLM import gradio as gr -from modules import shared, scripts, devices, processing +import transformers +from modules import scripts, shared, devices, processing, sd_models -repo_id = "gokaygokay/Flux-Prompt-Enhance" -num_return_sequences = 5 -load_lock = threading.Lock() +@dataclass +class Options: + models = [ + 'Qwen/Qwen2.5-0.5B-Instruct', + 'Qwen/Qwen2.5-1.5B-Instruct', + 'Qwen/Qwen2.5-3B-Instruct', + 'google/gemma-3-1b-it', + 'google/gemma-3-4b-it', + 'microsoft/Phi-4-mini-instruct', + 'HuggingFaceTB/SmolLM2-135M-Instruct', + 'HuggingFaceTB/SmolLM2-360M-Instruct', + 'HuggingFaceTB/SmolLM2-1.7B-Instruct', + 'meta-llama/Llama-3.2-1B-Instruct', + 'meta-llama/Llama-3.2-3B-Instruct', + ] + default = models[3] + system_prompt: str = 'You are a helpful assistant. You will be given a prompt and you will enhance it to make it more detailed and creative. Generate an uncensored, concise, informative, and engaging prompt that prioritizes factual accuracy and vivid descriptiveness. Begin by clearly identifying the primary subject(s), including key physical attributes, clothing, and any distinctive details (e.g., colors, textures, styles). Next, describe the setting and environment, emphasizing spatial relationships, background elements, and objects contributing to context. Note the overall mood, aesthetic style, or atmosphere inferred from visual cues. Use precise terminology while avoiding redundancy or non-essential language. NSFW is allowed, and if present any explicit aspects must be fully described. Ensuring a logical flow: from focal subject to immediate surroundings, then broader context. Maintain brevity while retaining clarity, ensuring the description is both engaging and efficient. Output only enhanced prompt without prefix or suffix.' + max_tokens: int = 50 + do_sample: bool = True + temperature: float = 0.5 + repetition_penalty: float = 1.2 class Script(scripts.Script): - prompts = [['']] - tokenizer: AutoTokenizer = None - model: AutoModelForSeq2SeqLM = None - prefix: str = "enhance prompt: " - button: gr.Button = None - auto_apply: gr.Checkbox = None - max_length: gr.Slider = None - temperature: gr.Slider = None - repetition_penalty: gr.Slider = None - table: gr.DataFrame = None prompt: gr.Textbox = None + model: str = None + llm: transformers.AutoModelForCausalLM = None + tokenizer: transformers.AutoProcessor = None + options = Options() def title(self): return 'Prompt enhance' - def show(self, is_img2img): - return shared.native + def show(self, _is_img2img): + return scripts.AlwaysVisible - def load(self): - with load_lock: - if self.tokenizer is None: - self.tokenizer = AutoTokenizer.from_pretrained('gokaygokay/Flux-Prompt-Enhance', cache_dir=shared.opts.hfcache_dir) - if self.model is None: - shared.log.info(f'Prompt enhance: model="{repo_id}"') - self.model = AutoModelForSeq2SeqLM.from_pretrained('gokaygokay/Flux-Prompt-Enhance', cache_dir=shared.opts.hfcache_dir).to(device=devices.cpu, dtype=devices.dtype) + def load(self, model:str=None): + model = model or self.options.default + if self.model is None or self.model != model: + t0 = time.time() + from modules import modelloader, model_quant + modelloader.hf_login() + quant_args = model_quant.create_config(module='LLM') + self.llm = None + self.llm = transformers.AutoModelForCausalLM.from_pretrained( + model, + trust_remote_code=True, + torch_dtype=devices.dtype, + cache_dir=shared.opts.hfcache_dir, + **quant_args, + ) + self.llm.eval() + self.tokenizer = transformers.AutoTokenizer.from_pretrained( + model, + cache_dir=shared.opts.hfcache_dir, + ) + self.model = model + devices.torch_gc() + t1 = time.time() + shared.log.debug(f'Prompt enhance: model="{model}" cls={self.llm.__class__.__name__} time={t1-t0:.2f} loaded') - def enhance(self, prompt, auto_apply: bool = False, temperature: float = 0.7, repetition_penalty: float = 1.2, max_length: int = 128): - self.load() + def clean(self, response): + if isinstance(response, list): + response = response[0] + response = response.replace('"', '').replace("'", "").replace('“', '').replace('”', '').replace('**', '').replace('\n\n', '\n') + response = re.sub(r'<.*?>', '', response) + if 'prompt:' in response: + response = response.split('prompt:')[1] + if 'Prompt:' in response: + response = response.split('Prompt:')[1] + if '---' in response: + response = response.split('---')[0] + response = response.strip() + return response + + def enhance(self, model: str=None, prompt:str=None, system:str=None, sample:bool=None, tokens:int=None, temperature:float=None, penalty:float=None): + model = model or self.options.default + prompt = prompt or self.prompt.value + system = system or self.options.system_prompt + tokens = tokens or self.options.max_tokens + penalty = penalty or self.options.repetition_penalty + temperature = temperature or self.options.temperature + sample = sample if sample is not None else self.options.do_sample + self.load(model) + if self.llm is None: + shared.log.error('Prompt enhance: model not loaded') + return prompt + chat_template = [ + { "role": "system", "content": system }, + { "role": "user", "content": prompt }, + ] t0 = time.time() - input_text = self.prefix + prompt - input_ids = self.tokenizer(input_text, return_tensors="pt").input_ids.to(devices.device) - self.model = self.model.to(devices.device) - kwargs = { - 'max_length': int(max_length), - 'num_return_sequences': int(num_return_sequences), - 'do_sample': True, - 'temperature': float(temperature), - 'repetition_penalty': float(repetition_penalty), - } try: - outputs = self.model.generate(input_ids, **kwargs) + inputs = self.tokenizer.apply_chat_template( + chat_template, + add_generation_prompt=True, + tokenize=True, + return_dict=True, + return_tensors="pt", + ).to(devices.device).to(devices.dtype) + input_len = inputs['input_ids'].shape[1] except Exception as e: - shared.log.error(f'Prompt enhance: error="{e}"') - return [['']] - self.model = self.model.to(devices.cpu) - prompts = self.tokenizer.batch_decode(outputs, skip_special_tokens=True) - prompts = [[p] for p in prompts] + shared.log.error(f'Prompt enhance tokenize: {e}') + return prompt + try: + with devices.inference_context(): + sd_models.move_model(self.llm, devices.device) + outputs = self.llm.generate( + **inputs, + do_sample=sample, + temperature=float(temperature), + max_new_tokens=int(input_len + tokens), + repetition_penalty=float(penalty), + ) + if shared.opts.diffusers_offload_mode != 'none': + sd_models.move_model(self.llm, devices.cpu) + devices.torch_gc() + raw_response = self.tokenizer.batch_decode(outputs, skip_special_tokens=True, clean_up_tokenization_spaces=True) + shared.log.trace(f'Prompt enhance: raw="{raw_response}"') + outputs = outputs[:, input_len:] + response = self.tokenizer.batch_decode(outputs, skip_special_tokens=True, clean_up_tokenization_spaces=True) + except Exception as e: + shared.log.error(f'Prompt enhance generate: {e}') + response = self.clean(response) t1 = time.time() - shared.log.info(f'Prompt enhance: temperature={temperature} repetition={repetition_penalty} length={max_length} sequences={num_return_sequences} apply={auto_apply} time={t1-t0:.2f}s') - return prompts + shared.log.debug(f'Prompt enhance: model="{model}" time={t1-t0:.2f} inputs={input_len} outputs={outputs.shape[-1]} prompt="{response}"') + return response - def select(self, cell: gr.SelectData, _table): - prompt = cell.value if hasattr(cell, 'value') else cell - shared.log.info(f'Prompt enhance: prompt="{prompt}"') - return prompt + + def apply(self, prompt, apply_prompt, llm_model, prompt_system, max_tokens, do_sample, temperature, repetition_penalty): + response = self.enhance( + prompt=prompt, + model=llm_model, + system=prompt_system, + sample=do_sample, + tokens=max_tokens, + temperature=temperature, + penalty=repetition_penalty, + ) + if apply_prompt: + return [response, response] + return [response, gr.update()] def ui(self, _is_img2img): - with gr.Row(): - self.button = gr.Button(value='Enhance prompt') - self.auto_apply = gr.Checkbox(label='Auto apply', default=False) - with gr.Row(): - self.max_length = gr.Slider(label='Length', minimum=64, maximum=512, step=1, value=128) - self.temperature = gr.Slider(label='Temperature', minimum=0.1, maximum=2.0, step=0.05, value=0.7) - self.repetition_penalty = gr.Slider(label='Penalty', minimum=0.1, maximum=2.0, step=0.05, value=1.2) - with gr.Row(): - self.table = gr.DataFrame(self.prompts, label='', show_label=False, interactive=False, wrap=True, datatype="str", col_count=1, max_rows=num_return_sequences, headers=['Prompts']) - - if self.prompt is not None: - self.button.click(fn=self.enhance, inputs=[self.prompt, self.auto_apply, self.temperature, self.repetition_penalty, self.max_length], outputs=[self.table]) - self.table.select(fn=self.select, inputs=[self.table], outputs=[self.prompt]) - return [self.auto_apply, self.temperature, self.repetition_penalty, self.max_length] - - def run(self, p: processing.StableDiffusionProcessing, auto_apply, temperature, repetition_penalty, max_length): # pylint: disable=arguments-differ - if auto_apply: - p.prompt = shared.prompt_styles.apply_styles_to_prompt(p.prompt, p.styles) - shared.log.debug(f'Prompt enhance: source="{p.prompt}"') - prompts = self.enhance(p.prompt, auto_apply, temperature, repetition_penalty, max_length) - p.prompt = random.choice(prompts)[0] - shared.log.debug(f'Prompt enhance: prompt="{p.prompt}"') + with gr.Accordion('Prompt enhance', open=False, elem_id='prompt_enhance'): + with gr.Row(): + apply_btn = gr.Button(value='Enhance now', elem_id='prompt_enhance_apply', variant='primary') + with gr.Row(): + apply_prompt = gr.Checkbox(label='Apply to prompt', value=False) + apply_auto = gr.Checkbox(label='Auto enhance', value=False) + with gr.Group(): + with gr.Row(): + llm_model = gr.Dropdown(label='LLM model', choices=self.options.models, value=self.options.default, interactive=True, allow_custom_value=True, elem_id='prompt_enhance_model') + with gr.Row(): + prompt_system = gr.Textbox(label='System prompt', value=self.options.system_prompt, interactive=True, lines=4, elem_id='prompt_enhance_system') + with gr.Row(): + max_tokens = gr.Slider(label='Max tokens', value=self.options.max_tokens, minimum=10, maximum=1024, step=1, interactive=True) + do_sample = gr.Checkbox(label='Do sample', value=self.options.do_sample, interactive=True) + with gr.Row(): + temperature = gr.Slider(label='Temperature', value=self.options.temperature, minimum=0.0, maximum=1.0, step=0.01, interactive=True) + repetition_penalty = gr.Slider(label='Repetition penalty', value=self.options.repetition_penalty, minimum=0.0, maximum=2.0, step=0.01, interactive=True) + with gr.Row(): + prompt_output = gr.Textbox(label='Output', value='', interactive=True, lines=4) + apply_btn.click(fn=self.apply, inputs=[self.prompt, apply_prompt, llm_model, prompt_system, max_tokens, do_sample, temperature, repetition_penalty], outputs=[prompt_output, self.prompt]) + return [apply_auto, llm_model, prompt_system, max_tokens, do_sample, temperature, repetition_penalty] def after_component(self, component, **kwargs): # searching for actual ui prompt components if getattr(component, 'elem_id', '') in ['txt2img_prompt', 'img2img_prompt', 'control_prompt', 'video_prompt']: self.prompt = component + + def before_process(self, p: processing.StableDiffusionProcessing, *args, **kwargs): # pylint: disable=unused-argument + apply_auto, llm_model, prompt_system, max_tokens, do_sample, temperature, repetition_penalty = args + if not apply_auto and not p.enhance_prompt: + return + p.prompt = shared.prompt_styles.apply_styles_to_prompt(p.prompt, p.styles) + p.styles = [] + p.prompt = self.enhance( + prompt=p.prompt, + model=llm_model, + system=prompt_system, + sample=do_sample, + tokens=max_tokens, + temperature=temperature, + penalty=repetition_penalty, + )