LLM nodes have been moved to a separate package of nodes Level Pixel Advanced. New logical and functional nodes have been added
This commit is contained in:
Level Pixel Dev
@@ -88,14 +88,31 @@ class HundredthsFloatSlider:
|
||||
number = 1.00
|
||||
return (number,)
|
||||
|
||||
class Seed:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {"required": {"seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff})}}
|
||||
|
||||
RETURN_TYPES = ("INT", )
|
||||
RETURN_NAMES = ("seed INT", )
|
||||
FUNCTION = "seedint"
|
||||
OUTPUT_NODE = True
|
||||
CATEGORY = "LevelPixel/IO"
|
||||
|
||||
@staticmethod
|
||||
def seedint(seed):
|
||||
return (seed,)
|
||||
|
||||
NODE_CLASS_MAPPINGS = {
|
||||
"SimpleFloatSlider|LP": FloatSlider,
|
||||
"TenthsSimpleFloatSlider|LP": TenthsFloatSlider,
|
||||
"HundredthsSimpleFloatSlider|LP": HundredthsFloatSlider,
|
||||
"Seed|LP": Seed,
|
||||
}
|
||||
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {
|
||||
"SimpleFloatSlider|LP": "Simple Float Slider [LP]",
|
||||
"TenthsSimpleFloatSlider|LP": "Simple Float Slider - Tenths Step [LP]",
|
||||
"HundredthsSimpleFloatSlider|LP": "Simple Float Slider - Hundredths Step [LP]",
|
||||
"Seed|LP": "Seed [LP]",
|
||||
}
|
||||
@@ -1,5 +1,11 @@
|
||||
import os
|
||||
|
||||
class AnyType(str):
|
||||
def __ne__(self, __value: object) -> bool:
|
||||
return False
|
||||
|
||||
any = AnyType("*")
|
||||
|
||||
class Text:
|
||||
|
||||
def __init__(self):
|
||||
@@ -89,14 +95,45 @@ class FindValueFromFile:
|
||||
|
||||
return {"ui": {"text": valueString, "log": log,}, "result": (valueString, boolResult,)}
|
||||
|
||||
class StringCycler:
|
||||
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {"required": {
|
||||
"text": ("STRING", {"multiline": True, "default": ""}),
|
||||
"repeats": ("INT", {"default": 1, "min": 1, "max": 99999}),
|
||||
"loops": ("INT", {"default": 1, "min": 1, "max": 99999}),
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = (any,)
|
||||
RETURN_NAMES = ("STRING",)
|
||||
OUTPUT_IS_LIST = (True,)
|
||||
FUNCTION = "string_cycle"
|
||||
CATEGORY = "LevelPixel/IO"
|
||||
|
||||
def string_cycle(self, text, repeats, loops=1):
|
||||
|
||||
lines = text.split('\n')
|
||||
list_out = []
|
||||
|
||||
for i in range(loops):
|
||||
for text_item in lines:
|
||||
for _ in range(repeats):
|
||||
list_out.append(text_item)
|
||||
|
||||
return (list_out, )
|
||||
|
||||
NODE_CLASS_MAPPINGS = {
|
||||
"Text|LP": Text,
|
||||
"String|LP": String,
|
||||
"FindValueFromFile|LP": FindValueFromFile,
|
||||
"StringCycler|LP": StringCycler,
|
||||
}
|
||||
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {
|
||||
"Text|LP": "Text [LP]",
|
||||
"String|LP": "String [LP]",
|
||||
"FindValueFromFile|LP": "Find Value From File [LP]",
|
||||
"StringCycler|LP": "String Cycler [LP]",
|
||||
}
|
||||
@@ -1,164 +0,0 @@
|
||||
import folder_paths
|
||||
import os
|
||||
from llama_cpp import Llama
|
||||
import gc
|
||||
import torch
|
||||
|
||||
|
||||
supported_LLava_extensions = set(['.gguf'])
|
||||
|
||||
try:
|
||||
folder_paths.folder_names_and_paths["LLavacheckpoints"] = (folder_paths.folder_names_and_paths["LLavacheckpoints"][0], supported_LLava_extensions)
|
||||
except:
|
||||
if not os.path.isdir(os.path.join(folder_paths.models_dir, "LLavacheckpoints")):
|
||||
os.mkdir(os.path.join(folder_paths.models_dir, "LLavacheckpoints"))
|
||||
|
||||
folder_paths.folder_names_and_paths["LLavacheckpoints"] = ([os.path.join(folder_paths.models_dir, "LLavacheckpoints")], supported_LLava_extensions)
|
||||
|
||||
class LLMLoader:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"ckpt_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
|
||||
"max_ctx": ("INT", {"default": 2048, "min": 128, "max": 128000, "step": 64}),
|
||||
"gpu_layers": ("INT", {"default": 27, "min": 0, "max": 100, "step": 1}),
|
||||
"n_threads": ("INT", {"default": 8, "min": 1, "max": 100, "step": 1}),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
RETURN_TYPES = ("CUSTOM",)
|
||||
RETURN_NAMES = ("model",)
|
||||
FUNCTION = "load_llm_checkpoint"
|
||||
|
||||
CATEGORY = "LevelPixel/LLM"
|
||||
def load_llm_checkpoint(self, ckpt_name, max_ctx, gpu_layers, n_threads):
|
||||
ckpt_path = folder_paths.get_full_path("LLavacheckpoints", ckpt_name)
|
||||
llm = Llama(model_path = ckpt_path, chat_format="chatml", offload_kqv=True,
|
||||
f16_kv=True, use_mlock=False, embedding=False, n_batch=1024,
|
||||
last_n_tokens_size=1024, verbose=True, seed=42, n_ctx = max_ctx,
|
||||
n_gpu_layers=gpu_layers, n_threads=n_threads,)
|
||||
return (llm, )
|
||||
|
||||
class LLMSampler:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"model": ("CUSTOM", {"default": ""}),
|
||||
"max_tokens": ("INT", {"default": 512, "min": 1, "max": 2048, "step": 1}),
|
||||
"temperature": ("FLOAT", {"default": 0.2, "min": 0.01, "max": 1.0, "step": 0.01}),
|
||||
"top_p": ("FLOAT", {"default": 0.95, "min": 0.1, "max": 1.0, "step": 0.01}),
|
||||
"top_k": ("INT", {"default": 40, "step": 1}),
|
||||
"frequency_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
|
||||
"presence_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
|
||||
"repeat_penalty": ("FLOAT", {"default": 1.1, "step": 0.01}),
|
||||
"seed": ("INT", {"default": 42, "step": 1}),
|
||||
"prompt": ("STRING",{"multiline": True, "default": ""}),
|
||||
"system_msg": ("STRING",{ "multiline": True, "default" : "You are an assistant who perfectly describes images."}),
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("STRING",)
|
||||
FUNCTION = "generate_text_sampler"
|
||||
CATEGORY = "LevelPixel/LLM"
|
||||
|
||||
def generate_text_sampler(self, system_msg, prompt, model, max_tokens,
|
||||
temperature, top_p, top_k, frequency_penalty,
|
||||
presence_penalty, repeat_penalty, seed):
|
||||
llm = model
|
||||
response = llm.create_chat_completion(messages=[
|
||||
{"role": "system", "content": system_msg},
|
||||
{"role": "user", "content": prompt + " Assistant:"},
|
||||
],
|
||||
max_tokens=max_tokens,
|
||||
temperature=temperature,
|
||||
top_p=top_p,
|
||||
top_k=top_k,
|
||||
frequency_penalty=frequency_penalty,
|
||||
presence_penalty=presence_penalty,
|
||||
repeat_penalty=repeat_penalty,
|
||||
seed=seed
|
||||
|
||||
)
|
||||
return (f"{response['choices'][0]['message']['content']}", )
|
||||
|
||||
|
||||
class LLMAdvanced:
|
||||
def __init__(self):
|
||||
self.llm = None
|
||||
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"ckpt_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
|
||||
"max_ctx": ("INT", {"default": 2048, "min": 300, "max": 100000, "step": 64}),
|
||||
"gpu_layers": ("INT", {"default": 27, "min": 0, "max": 100, "step": 1}),
|
||||
"n_threads": ("INT", {"default": 8, "min": 1, "max": 100, "step": 1}),
|
||||
"max_tokens": ("INT", {"default": 512, "min": 1, "max": 2048, "step": 1}),
|
||||
"temperature": ("FLOAT", {"default": 0.1, "min": 0.01, "max": 1.0, "step": 0.01}),
|
||||
"top_p": ("FLOAT", {"default": 0.95, "min": 0.1, "max": 1.0, "step": 0.01}),
|
||||
"top_k": ("INT", {"default": 40, "step": 1}),
|
||||
"frequency_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
|
||||
"presence_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
|
||||
"repeat_penalty": ("FLOAT", {"default": 1.1, "step": 0.01}),
|
||||
"seed": ("INT", {"default": 42, "step": 1}),
|
||||
"unload": ("BOOLEAN", {"default": False}),
|
||||
"prompt": ("STRING", {"multiline": True, "default": ""}),
|
||||
"system_msg": ("STRING", {"multiline": True, "default": "You are an assistant who perfectly describes images."}),
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("STRING",)
|
||||
FUNCTION = "generate_text_advanced"
|
||||
CATEGORY = "LevelPixel/LLM"
|
||||
|
||||
def generate_text_advanced(self, ckpt_name, max_ctx, gpu_layers, n_threads,
|
||||
system_msg, prompt, max_tokens, temperature, top_p,
|
||||
top_k, frequency_penalty, presence_penalty, repeat_penalty, seed, unload):
|
||||
|
||||
ckpt_path = folder_paths.get_full_path("LLavacheckpoints", ckpt_name)
|
||||
self.llm = Llama(model_path = ckpt_path, offload_kqv=True, f16_kv=True,
|
||||
use_mlock=False, embedding=False, n_batch=1024, last_n_tokens_size=1024,
|
||||
verbose=True, seed=42, n_ctx = max_ctx, n_gpu_layers=gpu_layers,
|
||||
n_threads=n_threads, logits_all=True, echo=False)
|
||||
|
||||
response = self.llm.create_chat_completion(messages=[
|
||||
{"role": "system", "content": system_msg},
|
||||
{"role": "user", "content": prompt},
|
||||
],
|
||||
max_tokens=max_tokens,
|
||||
temperature=temperature,
|
||||
top_p=top_p,
|
||||
top_k=top_k,
|
||||
frequency_penalty=frequency_penalty,
|
||||
presence_penalty=presence_penalty,
|
||||
repeat_penalty=repeat_penalty,
|
||||
seed=seed,
|
||||
)
|
||||
|
||||
if unload and self.llm is not None:
|
||||
self.llm.close()
|
||||
del self.llm
|
||||
self.llm = None
|
||||
gc.collect()
|
||||
torch.cuda.empty_cache()
|
||||
|
||||
return (f"{response['choices'][0]['message']['content']}", )
|
||||
|
||||
NODE_CLASS_MAPPINGS = {
|
||||
"LLMLoader|LP": LLMLoader,
|
||||
"LLMSampler|LP": LLMSampler,
|
||||
"LLMAdvanced|LP": LLMAdvanced,
|
||||
}
|
||||
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {
|
||||
"LLMLoader|LP": "LLM Loader [LP]",
|
||||
"LLMSampler|LP": "LLM Sampler [LP]",
|
||||
"LLMAdvanced|LP": "LLM Advanced [LP]"
|
||||
}
|
||||
@@ -6,6 +6,12 @@ import string
|
||||
from deep_translator import GoogleTranslator
|
||||
from langdetect import detect
|
||||
|
||||
class AnyType(str):
|
||||
def __ne__(self, __value: object) -> bool:
|
||||
return False
|
||||
|
||||
any = AnyType("*")
|
||||
|
||||
class TextChoiceParser:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
@@ -253,6 +259,55 @@ class KeepOnlyEnglishWords:
|
||||
|
||||
return (result,)
|
||||
|
||||
class TextReplace:
|
||||
|
||||
@ classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"text": ("STRING", {"multiline": True, "default": "", "forceInput": True}),
|
||||
},
|
||||
"optional": {
|
||||
"find1": ("STRING", {"multiline": False, "default": ""}),
|
||||
"replace1": ("STRING", {"multiline": False, "default": ""}),
|
||||
"find2": ("STRING", {"multiline": False, "default": ""}),
|
||||
"replace2": ("STRING", {"multiline": False, "default": ""}),
|
||||
"find3": ("STRING", {"multiline": False, "default": ""}),
|
||||
"replace3": ("STRING", {"multiline": False, "default": ""}),
|
||||
"find4": ("STRING", {"multiline": False, "default": ""}),
|
||||
"replace4": ("STRING", {"multiline": False, "default": ""}),
|
||||
"find5": ("STRING", {"multiline": False, "default": ""}),
|
||||
"replace5": ("STRING", {"multiline": False, "default": ""}),
|
||||
"find6": ("STRING", {"multiline": False, "default": ""}),
|
||||
"replace6": ("STRING", {"multiline": False, "default": ""}),
|
||||
"find7": ("STRING", {"multiline": False, "default": ""}),
|
||||
"replace7": ("STRING", {"multiline": False, "default": ""}),
|
||||
"find8": ("STRING", {"multiline": False, "default": ""}),
|
||||
"replace8": ("STRING", {"multiline": False, "default": ""}),
|
||||
"find9": ("STRING", {"multiline": False, "default": ""}),
|
||||
"replace9": ("STRING", {"multiline": False, "default": ""}),
|
||||
},
|
||||
}
|
||||
|
||||
RETURN_TYPES = (any, )
|
||||
RETURN_NAMES = ("text TEXT", )
|
||||
FUNCTION = "replace_text"
|
||||
CATEGORY = "LevelPixel/Text"
|
||||
|
||||
def replace_text(self, text, find1="", replace1="", find2="", replace2="", find3="", replace3="", find4="", replace4="", find5="", replace5="", find6="", replace6="", find7="", replace7="", find8="", replace8="", find9="", replace9=""):
|
||||
|
||||
text = text.replace(find1, replace1)
|
||||
text = text.replace(find2, replace2)
|
||||
text = text.replace(find3, replace3)
|
||||
text = text.replace(find4, replace4)
|
||||
text = text.replace(find5, replace5)
|
||||
text = text.replace(find6, replace6)
|
||||
text = text.replace(find7, replace7)
|
||||
text = text.replace(find8, replace8)
|
||||
text = text.replace(find9, replace9)
|
||||
|
||||
return (text,)
|
||||
|
||||
NODE_CLASS_MAPPINGS = {
|
||||
"TextChoiceParser|LP": TextChoiceParser,
|
||||
"CLIPTextEncodeTranslate|LP": CLIPTextEncodeTranslate,
|
||||
@@ -260,6 +315,7 @@ NODE_CLASS_MAPPINGS = {
|
||||
"TextToList|LP": TextToList,
|
||||
"SplitCompoundText|LP": SplitCompoundText,
|
||||
"KeepOnlyEnglishWords|LP": KeepOnlyEnglishWords,
|
||||
"TextReplace|LP": TextReplace,
|
||||
}
|
||||
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {
|
||||
@@ -269,4 +325,5 @@ NODE_DISPLAY_NAME_MAPPINGS = {
|
||||
"TextToList|LP": "Text To List [LP]",
|
||||
"SplitCompoundText|LP": "Split Compound Text [LP]",
|
||||
"KeepOnlyEnglishWords|LP": "Keep Only English Words [LP]",
|
||||
"TextReplace|LP": "Text Replace [LP]",
|
||||
}
|
||||
@@ -32,11 +32,221 @@ class Delay:
|
||||
print(f"[Delay Node] Delay of {delay_text} completed")
|
||||
return (input,)
|
||||
|
||||
class PipeOut:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {"pipe": ("PIPE_LINE",)},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("PIPE_LINE", "MODEL", "CONDITIONING", "CONDITIONING", "LATENT", "VAE", "CLIP", "CONTROL_NET", "IMAGE", "INT", any, any, any, any, any,)
|
||||
RETURN_NAMES = ("pipe", "model", "pos", "neg", "latent", "vae", "clip", "controlnet", "image", "seed", "any1", "any2", "any3", "any4", "any5",)
|
||||
FUNCTION = "pipe_out"
|
||||
CATEGORY = "LevelPixel/Utils"
|
||||
|
||||
def pipe_out(self, pipe):
|
||||
model, pos, neg, latent, vae, clip, controlnet, image, seed, any1, any2, any3, any4, any5 = pipe
|
||||
return (pipe, model, pos, neg, latent, vae, clip, controlnet, image, seed, any1, any2, any3, any4, any5, )
|
||||
|
||||
class PipeIn:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"optional": {
|
||||
"pipe": ("PIPE_LINE",),
|
||||
"model": ("MODEL",),
|
||||
"pos": ("CONDITIONING",),
|
||||
"neg": ("CONDITIONING",),
|
||||
"latent": ("LATENT",),
|
||||
"vae": ("VAE",),
|
||||
"clip": ("CLIP",),
|
||||
"controlnet": ("CONTROL_NET",),
|
||||
"image": ("IMAGE",),
|
||||
"seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}),
|
||||
"any1": (any, {"defaultInput": True}),
|
||||
"any2": (any, {"defaultInput": True}),
|
||||
"any3": (any, {"defaultInput": True}),
|
||||
"any4": (any, {"defaultInput": True}),
|
||||
"any5": (any, {"defaultInput": True}),
|
||||
},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("PIPE_LINE",)
|
||||
RETURN_NAMES = ("pipe",)
|
||||
FUNCTION = "pipe_in"
|
||||
CATEGORY = "LevelPixel/Utils"
|
||||
|
||||
def pipe_in(self, pipe=None, model=None, pos=None, neg=None, latent=None, vae=None, clip=None, controlnet=None, image=None, seed=None, any1=None, any2=None, any3=None, any4=None, any5=None,):
|
||||
|
||||
new_model = None
|
||||
new_pos = None
|
||||
new_neg = None
|
||||
new_latent = None
|
||||
new_vae = None
|
||||
new_clip = None
|
||||
new_controlnet = None
|
||||
new_image = None
|
||||
new_seed = None
|
||||
new_any1 = None
|
||||
new_any2 = None
|
||||
new_any3 = None
|
||||
new_any4 = None
|
||||
new_any5 = None
|
||||
|
||||
if pipe is not None:
|
||||
new_model, new_pos, new_neg, new_latent, new_vae, new_clip, new_controlnet, new_image, new_seed, new_any1, new_any2, new_any3, new_any4, new_any5 = pipe
|
||||
|
||||
if model is not None:
|
||||
new_model = model
|
||||
|
||||
if pos is not None:
|
||||
new_pos = pos
|
||||
|
||||
if neg is not None:
|
||||
new_neg = neg
|
||||
|
||||
if latent is not None:
|
||||
new_latent = latent
|
||||
|
||||
if vae is not None:
|
||||
new_vae = vae
|
||||
|
||||
if clip is not None:
|
||||
new_clip = clip
|
||||
|
||||
if controlnet is not None:
|
||||
new_controlnet = controlnet
|
||||
|
||||
if image is not None:
|
||||
new_image = image
|
||||
|
||||
if seed is not None:
|
||||
new_seed = seed
|
||||
|
||||
if any1 is not None:
|
||||
new_any1 = any1
|
||||
|
||||
if any2 is not None:
|
||||
new_any2 = any2
|
||||
|
||||
if any3 is not None:
|
||||
new_any3 = any3
|
||||
|
||||
if any4 is not None:
|
||||
new_any4 = any4
|
||||
|
||||
if any5 is not None:
|
||||
new_any5 = any5
|
||||
|
||||
pipe = new_model, new_pos, new_neg, new_latent, new_vae, new_clip, new_controlnet, new_image, new_seed, new_any1, new_any2, new_any3, new_any4, new_any5
|
||||
|
||||
return (pipe, )
|
||||
|
||||
class Pipe:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"optional": {
|
||||
"pipe": ("PIPE_LINE",),
|
||||
"model": ("MODEL",),
|
||||
"pos": ("CONDITIONING",),
|
||||
"neg": ("CONDITIONING",),
|
||||
"latent": ("LATENT",),
|
||||
"vae": ("VAE",),
|
||||
"clip": ("CLIP",),
|
||||
"controlnet": ("CONTROL_NET",),
|
||||
"image": ("IMAGE",),
|
||||
"seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}),
|
||||
"any1": (any, {"defaultInput": True}),
|
||||
"any2": (any, {"defaultInput": True}),
|
||||
"any3": (any, {"defaultInput": True}),
|
||||
"any4": (any, {"defaultInput": True}),
|
||||
"any5": (any, {"defaultInput": True}),
|
||||
},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("PIPE_LINE", "MODEL", "CONDITIONING", "CONDITIONING", "LATENT", "VAE", "CLIP", "CONTROL_NET", "IMAGE", "INT", any, any, any, any, any,)
|
||||
RETURN_NAMES = ("pipe", "model", "pos", "neg", "latent", "vae", "clip", "controlnet", "image", "seed", "any1", "any2", "any3", "any4", "any5",)
|
||||
FUNCTION = "pipe"
|
||||
CATEGORY = "LevelPixel/Utils"
|
||||
|
||||
def pipe(self, pipe=None, model=None, pos=None, neg=None, latent=None, vae=None, clip=None, controlnet=None, image=None, seed=None, any1=None, any2=None, any3=None, any4=None, any5=None,):
|
||||
|
||||
new_model = None
|
||||
new_pos = None
|
||||
new_neg = None
|
||||
new_latent = None
|
||||
new_vae = None
|
||||
new_clip = None
|
||||
new_controlnet = None
|
||||
new_image = None
|
||||
new_seed = None
|
||||
new_any1 = None
|
||||
new_any2 = None
|
||||
new_any3 = None
|
||||
new_any4 = None
|
||||
new_any5 = None
|
||||
|
||||
if pipe is not None:
|
||||
new_model, new_pos, new_neg, new_latent, new_vae, new_clip, new_controlnet, new_image, new_seed, new_any1, new_any2, new_any3, new_any4, new_any5 = pipe
|
||||
|
||||
if model is not None:
|
||||
new_model = model
|
||||
|
||||
if pos is not None:
|
||||
new_pos = pos
|
||||
|
||||
if neg is not None:
|
||||
new_neg = neg
|
||||
|
||||
if latent is not None:
|
||||
new_latent = latent
|
||||
|
||||
if vae is not None:
|
||||
new_vae = vae
|
||||
|
||||
if clip is not None:
|
||||
new_clip = clip
|
||||
|
||||
if controlnet is not None:
|
||||
new_controlnet = controlnet
|
||||
|
||||
if image is not None:
|
||||
new_image = image
|
||||
|
||||
if seed is not None:
|
||||
new_seed = seed
|
||||
|
||||
if any1 is not None:
|
||||
new_any1 = any1
|
||||
|
||||
if any2 is not None:
|
||||
new_any2 = any2
|
||||
|
||||
if any3 is not None:
|
||||
new_any3 = any3
|
||||
|
||||
if any4 is not None:
|
||||
new_any4 = any4
|
||||
|
||||
if any5 is not None:
|
||||
new_any5 = any5
|
||||
|
||||
pipe = new_model, new_pos, new_neg, new_latent, new_vae, new_clip, new_controlnet, new_image, new_seed, new_any1, new_any2, new_any3, new_any4, new_any5
|
||||
|
||||
return (pipe, new_model, new_pos, new_neg, new_latent, new_vae, new_clip, new_controlnet, new_image, new_seed, new_any1, new_any2, new_any3, new_any4, new_any5,)
|
||||
|
||||
NODE_CLASS_MAPPINGS = {
|
||||
"Delay|LP": Delay,
|
||||
"PipeOut|LP": PipeOut,
|
||||
"PipeIn|LP": PipeIn,
|
||||
"Pipe|LP": Pipe,
|
||||
}
|
||||
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {
|
||||
"Delay|LP": "Delay [LP]",
|
||||
"PipeOut|LP": "Pipe Out [LP]",
|
||||
"PipeIn|LP": "Pipe In [LP]",
|
||||
"Pipe|LP": "Pipe [LP]",
|
||||
}
|
||||
|
||||
|
||||
@@ -1,26 +0,0 @@
|
||||
{
|
||||
"name": "Autotagger",
|
||||
"logging": false,
|
||||
"settings": {
|
||||
"model": "wd-eva02-large-tagger-v3",
|
||||
"threshold": 0.35,
|
||||
"character_threshold": 0.85,
|
||||
"exclude_tags": "",
|
||||
"ortProviders": ["CUDAExecutionProvider", "CPUExecutionProvider"],
|
||||
"HF_ENDPOINT": "https://huggingface.co"
|
||||
},
|
||||
"models": {
|
||||
"wd-eva02-large-tagger-v3": "{HF_ENDPOINT}/SmilingWolf/wd-eva02-large-tagger-v3",
|
||||
"wd-vit-tagger-v3": "{HF_ENDPOINT}/SmilingWolf/wd-vit-tagger-v3",
|
||||
"wd-swinv2-tagger-v3": "{HF_ENDPOINT}/SmilingWolf/wd-swinv2-tagger-v3",
|
||||
"wd-convnext-tagger-v3": "{HF_ENDPOINT}/SmilingWolf/wd-convnext-tagger-v3",
|
||||
"wd-v1-4-moat-tagger-v2": "{HF_ENDPOINT}/SmilingWolf/wd-v1-4-moat-tagger-v2",
|
||||
"wd-v1-4-convnextv2-tagger-v2": "{HF_ENDPOINT}/SmilingWolf/wd-v1-4-convnextv2-tagger-v2",
|
||||
"wd-v1-4-convnext-tagger-v2": "{HF_ENDPOINT}/SmilingWolf/wd-v1-4-convnext-tagger-v2",
|
||||
"wd-v1-4-convnext-tagger": "{HF_ENDPOINT}/SmilingWolf/wd-v1-4-convnext-tagger",
|
||||
"wd-v1-4-vit-tagger-v2": "{HF_ENDPOINT}/SmilingWolf/wd-v1-4-vit-tagger-v2",
|
||||
"wd-v1-4-swinv2-tagger-v2": "{HF_ENDPOINT}/SmilingWolf/wd-v1-4-swinv2-tagger-v2",
|
||||
"wd-v1-4-vit-tagger": "{HF_ENDPOINT}/SmilingWolf/wd-v1-4-vit-tagger",
|
||||
"Z3D-E621-Convnext": "{HF_ENDPOINT}/silveroxides/Z3D-E621-Convnext"
|
||||
}
|
||||
}
|
||||
@@ -1,446 +0,0 @@
|
||||
import comfy.utils
|
||||
import asyncio
|
||||
import aiohttp
|
||||
import numpy as np
|
||||
import csv
|
||||
import os
|
||||
import sys
|
||||
import onnxruntime as ort
|
||||
from onnxruntime import InferenceSession
|
||||
from PIL import Image
|
||||
from server import PromptServer
|
||||
from aiohttp import web
|
||||
import folder_paths
|
||||
import json
|
||||
import shutil
|
||||
import inspect
|
||||
from tqdm import tqdm
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(os.path.realpath(__file__)), "comfy"))
|
||||
|
||||
supported_autotaggers_extensions = set(['.onnx'])
|
||||
|
||||
try:
|
||||
folder_paths.folder_names_and_paths["autotaggers"] = (folder_paths.folder_names_and_paths["autotaggers"][0], supported_autotaggers_extensions)
|
||||
except:
|
||||
if not os.path.isdir(os.path.join(folder_paths.models_dir, "autotaggers")):
|
||||
os.mkdir(os.path.join(folder_paths.models_dir, "autotaggers"))
|
||||
|
||||
folder_paths.folder_names_and_paths["autotaggers"] = ([os.path.join(folder_paths.models_dir, "autotaggers")], supported_autotaggers_extensions)
|
||||
|
||||
config = None
|
||||
|
||||
def is_logging_enabled():
|
||||
config = get_extension_config()
|
||||
if "logging" not in config:
|
||||
return False
|
||||
return config["logging"]
|
||||
|
||||
|
||||
def log(message, type=None, always=False):
|
||||
if not always and not is_logging_enabled():
|
||||
return
|
||||
|
||||
if type is not None:
|
||||
message = f"[{type}] {message}"
|
||||
|
||||
name = get_extension_config()["name"]
|
||||
|
||||
print(f"(LevelPixel:{name}) {message}")
|
||||
|
||||
|
||||
|
||||
|
||||
def get_ext_dir(subpath=None, mkdir=False):
|
||||
dir = os.path.dirname(__file__)
|
||||
if subpath is not None:
|
||||
dir = os.path.join(dir, subpath)
|
||||
|
||||
dir = os.path.abspath(dir)
|
||||
|
||||
if mkdir and not os.path.exists(dir):
|
||||
os.makedirs(dir)
|
||||
return dir
|
||||
|
||||
|
||||
def get_comfy_dir(subpath=None):
|
||||
dir = os.path.dirname(inspect.getfile(PromptServer))
|
||||
if subpath is not None:
|
||||
dir = os.path.join(dir, subpath)
|
||||
|
||||
dir = os.path.abspath(dir)
|
||||
|
||||
return dir
|
||||
|
||||
|
||||
def get_web_ext_dir():
|
||||
config = get_extension_config()
|
||||
name = config["name"]
|
||||
dir = get_comfy_dir("web/extensions/levelpixel")
|
||||
if not os.path.exists(dir):
|
||||
os.makedirs(dir)
|
||||
dir += "/" + name
|
||||
return dir
|
||||
|
||||
|
||||
def get_extension_config(reload=False):
|
||||
global config
|
||||
if reload == False and config is not None:
|
||||
return config
|
||||
|
||||
config_path = get_ext_dir("autotagger_LP.user.json")
|
||||
if not os.path.exists(config_path):
|
||||
config_path = get_ext_dir("autotagger_LP.json")
|
||||
|
||||
if not os.path.exists(config_path):
|
||||
log("Missing autotagger_LP.json and autotagger_LP.user.json, this extension may not work correctly. Please reinstall the extension.",
|
||||
type="ERROR", always=True)
|
||||
print(f"Extension path: {get_ext_dir()}")
|
||||
return {"name": "Unknown", "version": -1}
|
||||
with open(config_path, "r") as f:
|
||||
config = json.loads(f.read())
|
||||
return config
|
||||
|
||||
def link_js(src, dst):
|
||||
src = os.path.abspath(src)
|
||||
dst = os.path.abspath(dst)
|
||||
if os.name == "nt":
|
||||
try:
|
||||
import _winapi
|
||||
_winapi.CreateJunction(src, dst)
|
||||
return True
|
||||
except:
|
||||
pass
|
||||
try:
|
||||
os.symlink(src, dst)
|
||||
return True
|
||||
except:
|
||||
import logging
|
||||
logging.exception('')
|
||||
return False
|
||||
|
||||
|
||||
def is_junction(path):
|
||||
if os.name != "nt":
|
||||
return False
|
||||
try:
|
||||
return bool(os.readlink(path))
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
def install_js():
|
||||
src_dir = get_ext_dir("web/js")
|
||||
if not os.path.exists(src_dir):
|
||||
log("No JS")
|
||||
return
|
||||
|
||||
should_install = should_install_js()
|
||||
if should_install:
|
||||
log("it looks like you're running an old version of ComfyUI that requires manual setup of web files, it is recommended you update your installation.", "warning", True)
|
||||
dst_dir = get_web_ext_dir()
|
||||
linked = os.path.islink(dst_dir) or is_junction(dst_dir)
|
||||
if linked or os.path.exists(dst_dir):
|
||||
if linked:
|
||||
if should_install:
|
||||
log("JS already linked")
|
||||
else:
|
||||
os.unlink(dst_dir)
|
||||
log("JS unlinked, PromptServer will serve extension")
|
||||
elif not should_install:
|
||||
shutil.rmtree(dst_dir)
|
||||
log("JS deleted, PromptServer will serve extension")
|
||||
return
|
||||
|
||||
if not should_install:
|
||||
log("JS skipped, PromptServer will serve extension")
|
||||
return
|
||||
|
||||
if link_js(src_dir, dst_dir):
|
||||
log("JS linked")
|
||||
return
|
||||
|
||||
log("Copying JS files")
|
||||
shutil.copytree(src_dir, dst_dir, dirs_exist_ok=True)
|
||||
|
||||
|
||||
def should_install_js():
|
||||
return not hasattr(PromptServer.instance, "supports") or "custom_nodes_from_web" not in PromptServer.instance.supports
|
||||
|
||||
|
||||
def init(check_imports):
|
||||
log("Init")
|
||||
|
||||
if check_imports is not None:
|
||||
import importlib.util
|
||||
for imp in check_imports:
|
||||
spec = importlib.util.find_spec(imp)
|
||||
if spec is None:
|
||||
log(f"{imp} is required, please check requirements are installed.", type="ERROR", always=True)
|
||||
return False
|
||||
|
||||
install_js()
|
||||
return True
|
||||
|
||||
async def download_to_file(url, destination, update_callback, is_ext_subpath=True, session=None):
|
||||
close_session = False
|
||||
if session is None:
|
||||
close_session = True
|
||||
loop = None
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
except:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
|
||||
session = aiohttp.ClientSession(loop=loop)
|
||||
if is_ext_subpath:
|
||||
destination = get_ext_dir(destination)
|
||||
try:
|
||||
proxy = os.getenv("HTTP_PROXY") or os.getenv("http_proxy")
|
||||
print("proxy:", proxy)
|
||||
proxy_auth = None
|
||||
if proxy:
|
||||
proxy_auth = aiohttp.BasicAuth(os.getenv("PROXY_USER", ""), os.getenv("PROXY_PASS", ""))
|
||||
|
||||
async with session.get(url, proxy=proxy, proxy_auth=proxy_auth) as response:
|
||||
size = int(response.headers.get('content-length', 0)) or None
|
||||
|
||||
with tqdm(
|
||||
unit='B', unit_scale=True, miniters=1, desc=url.split('/')[-1], total=size,
|
||||
) as progressbar:
|
||||
with open(destination, mode='wb') as f:
|
||||
perc = 0
|
||||
async for chunk in response.content.iter_chunked(2048):
|
||||
f.write(chunk)
|
||||
progressbar.update(len(chunk))
|
||||
if update_callback is not None and progressbar.total is not None and progressbar.total != 0:
|
||||
last = perc
|
||||
perc = round(progressbar.n / progressbar.total, 2)
|
||||
if perc != last:
|
||||
last = perc
|
||||
await update_callback(perc)
|
||||
finally:
|
||||
if close_session and session is not None:
|
||||
await session.close()
|
||||
|
||||
|
||||
def wait_for_async(async_fn, loop=None):
|
||||
return asyncio.run(async_fn())
|
||||
|
||||
def update_node_status(client_id, node, text, progress=None):
|
||||
if client_id is None:
|
||||
client_id = PromptServer.instance.client_id
|
||||
|
||||
if client_id is None:
|
||||
return
|
||||
|
||||
PromptServer.instance.send_sync("levelpixel/update_status", {
|
||||
"node": node,
|
||||
"progress": progress,
|
||||
"text": text
|
||||
}, client_id)
|
||||
|
||||
async def update_node_status_async(client_id, node, text, progress=None):
|
||||
if client_id is None:
|
||||
client_id = PromptServer.instance.client_id
|
||||
|
||||
if client_id is None:
|
||||
return
|
||||
|
||||
await PromptServer.instance.send("levelpixel/update_status", {
|
||||
"node": node,
|
||||
"progress": progress,
|
||||
"text": text
|
||||
}, client_id)
|
||||
|
||||
|
||||
config_autotagger = get_extension_config()
|
||||
|
||||
defaults = {
|
||||
"model": "wd-eva02-large-tagger-v3",
|
||||
"threshold": 0.35,
|
||||
"character_threshold": 0.85,
|
||||
"replace_underscore": False,
|
||||
"trailing_comma": False,
|
||||
"exclude_tags": "",
|
||||
"ortProviders": ["CUDAExecutionProvider", "CPUExecutionProvider"],
|
||||
"HF_ENDPOINT": "https://huggingface.co"
|
||||
}
|
||||
defaults.update(config_autotagger.get("settings", {}))
|
||||
|
||||
models_dir = folder_paths.get_folder_paths("autotaggers")[0]
|
||||
if not os.path.exists(models_dir):
|
||||
os.makedirs(models_dir)
|
||||
|
||||
known_models = list(config_autotagger["models"].keys())
|
||||
|
||||
#log("Available ORT providers: " + ", ".join(ort.get_available_providers()), "DEBUG", True)
|
||||
#log("Using ORT providers: " + ", ".join(defaults["ortProviders"]), "DEBUG", True)
|
||||
|
||||
def get_installed_models():
|
||||
models = filter(lambda x: x.endswith(".onnx"), os.listdir(models_dir))
|
||||
models = [m for m in models if os.path.exists(os.path.join(models_dir, os.path.splitext(m)[0] + ".csv"))]
|
||||
return models
|
||||
|
||||
|
||||
async def tag(image, model_name, threshold=0.35, character_threshold=0.85, exclude_tags="", replace_underscore=True, trailing_comma=False, client_id=None, node=None):
|
||||
if model_name.endswith(".onnx"):
|
||||
model_name = model_name[0:-5]
|
||||
installed = list(get_installed_models())
|
||||
if not any(model_name + ".onnx" in s for s in installed):
|
||||
await download_model(model_name, client_id, node)
|
||||
|
||||
name = os.path.join(models_dir, model_name + ".onnx")
|
||||
model = InferenceSession(name, providers=defaults["ortProviders"])
|
||||
|
||||
input = model.get_inputs()[0]
|
||||
height = input.shape[1]
|
||||
|
||||
# Reduce to max size and pad with white
|
||||
ratio = float(height)/max(image.size)
|
||||
new_size = tuple([int(x*ratio) for x in image.size])
|
||||
image = image.resize(new_size, Image.LANCZOS)
|
||||
square = Image.new("RGB", (height, height), (255, 255, 255))
|
||||
square.paste(image, ((height-new_size[0])//2, (height-new_size[1])//2))
|
||||
|
||||
image = np.array(square).astype(np.float32)
|
||||
image = image[:, :, ::-1] # RGB -> BGR
|
||||
image = np.expand_dims(image, 0)
|
||||
|
||||
# Read all tags from csv and locate start of each category
|
||||
tags = []
|
||||
general_index = None
|
||||
character_index = None
|
||||
with open(os.path.join(models_dir, model_name + ".csv")) as f:
|
||||
reader = csv.reader(f)
|
||||
next(reader)
|
||||
for row in reader:
|
||||
if general_index is None and row[2] == "0":
|
||||
general_index = reader.line_num - 2
|
||||
elif character_index is None and row[2] == "4":
|
||||
character_index = reader.line_num - 2
|
||||
if replace_underscore:
|
||||
tags.append(row[1].replace("_", " "))
|
||||
else:
|
||||
tags.append(row[1])
|
||||
|
||||
label_name = model.get_outputs()[0].name
|
||||
probs = model.run([label_name], {input.name: image})[0]
|
||||
|
||||
result = list(zip(tags, probs[0]))
|
||||
|
||||
# rating = max(result[:general_index], key=lambda x: x[1])
|
||||
general = [item for item in result[general_index:character_index] if item[1] > threshold]
|
||||
character = [item for item in result[character_index:] if item[1] > character_threshold]
|
||||
|
||||
all = character + general
|
||||
remove = [s.strip() for s in exclude_tags.lower().split(",")]
|
||||
all = [tag for tag in all if tag[0] not in remove]
|
||||
|
||||
res = ("" if trailing_comma else ", ").join((item[0].replace("(", "\\(").replace(")", "\\)") + (", " if trailing_comma else "") for item in all))
|
||||
|
||||
print(res)
|
||||
return res
|
||||
|
||||
|
||||
async def download_model(model, client_id, node):
|
||||
hf_endpoint = os.getenv("HF_ENDPOINT", defaults["HF_ENDPOINT"])
|
||||
if not hf_endpoint.startswith("https://"):
|
||||
hf_endpoint = f"https://{hf_endpoint}"
|
||||
if hf_endpoint.endswith("/"):
|
||||
hf_endpoint = hf_endpoint.rstrip("/")
|
||||
|
||||
url = config_autotagger["models"][model]
|
||||
url = url.replace("{HF_ENDPOINT}", hf_endpoint)
|
||||
url = f"{url}/resolve/main/"
|
||||
async with aiohttp.ClientSession(loop=asyncio.get_event_loop()) as session:
|
||||
async def update_callback(perc):
|
||||
nonlocal client_id
|
||||
message = ""
|
||||
if perc < 100:
|
||||
message = f"Downloading {model}"
|
||||
update_node_status(client_id, node, message, perc)
|
||||
|
||||
try:
|
||||
await download_to_file(
|
||||
f"{url}model.onnx", os.path.join(models_dir,f"{model}.onnx"), update_callback, session=session)
|
||||
await download_to_file(
|
||||
f"{url}selected_tags.csv", os.path.join(models_dir,f"{model}.csv"), update_callback, session=session)
|
||||
except aiohttp.client_exceptions.ClientConnectorError as err:
|
||||
log("Unable to download model. Download files manually or try using a HF mirror/proxy website by setting the environment variable HF_ENDPOINT=https://.....", "ERROR", True)
|
||||
raise
|
||||
|
||||
update_node_status(client_id, node, None)
|
||||
|
||||
return web.Response(status=200)
|
||||
|
||||
|
||||
@PromptServer.instance.routes.get("/levelpixel/autotagger/tag")
|
||||
async def get_tags(request):
|
||||
if "filename" not in request.rel_url.query:
|
||||
return web.Response(status=404)
|
||||
|
||||
type = request.query.get("type", "output")
|
||||
if type not in ["output", "input", "temp"]:
|
||||
return web.Response(status=400)
|
||||
|
||||
target_dir = get_comfy_dir(type)
|
||||
image_path = os.path.abspath(os.path.join(
|
||||
target_dir, request.query.get("subfolder", ""), request.query["filename"]))
|
||||
c = os.path.commonpath((image_path, target_dir))
|
||||
if os.path.commonpath((image_path, target_dir)) != target_dir:
|
||||
return web.Response(status=403)
|
||||
|
||||
if not os.path.isfile(image_path):
|
||||
return web.Response(status=404)
|
||||
|
||||
image = Image.open(image_path)
|
||||
|
||||
models = get_installed_models()
|
||||
default = defaults["model"] + ".onnx"
|
||||
model = default if default in models else models[0]
|
||||
|
||||
return web.json_response(await tag(image, model, client_id=request.rel_url.query.get("clientId", ""), node=request.rel_url.query.get("node", "")))
|
||||
|
||||
|
||||
class Autotagger:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
extra = [name for name, _ in (os.path.splitext(m) for m in get_installed_models()) if name not in known_models]
|
||||
models = known_models + extra
|
||||
return {"required": {
|
||||
"image": ("IMAGE", ),
|
||||
"model": (models, { "default": defaults["model"] }),
|
||||
"threshold": ("FLOAT", {"default": defaults["threshold"], "min": 0.0, "max": 1, "step": 0.05}),
|
||||
"character_threshold": ("FLOAT", {"default": defaults["character_threshold"], "min": 0.0, "max": 1, "step": 0.05}),
|
||||
"replace_underscore": ("BOOLEAN", {"default": defaults["replace_underscore"]}),
|
||||
"trailing_comma": ("BOOLEAN", {"default": defaults["trailing_comma"]}),
|
||||
"exclude_tags": ("STRING", {"default": defaults["exclude_tags"]}),
|
||||
}}
|
||||
|
||||
RETURN_TYPES = ("STRING",)
|
||||
OUTPUT_IS_LIST = (True,)
|
||||
FUNCTION = "tag"
|
||||
OUTPUT_NODE = False
|
||||
|
||||
CATEGORY = "LevelPixel/VLM"
|
||||
|
||||
def tag(self, image, model, threshold, character_threshold, exclude_tags="", replace_underscore=False, trailing_comma=False):
|
||||
tensor = image*255
|
||||
tensor = np.array(tensor, dtype=np.uint8)
|
||||
|
||||
pbar = comfy.utils.ProgressBar(tensor.shape[0])
|
||||
tags = []
|
||||
for i in range(tensor.shape[0]):
|
||||
image = Image.fromarray(tensor[i])
|
||||
tags.append(wait_for_async(lambda: tag(image, model, threshold, character_threshold, exclude_tags, replace_underscore, trailing_comma)))
|
||||
pbar.update(1)
|
||||
return {"ui": {"tags": tags}, "result": (tags,)}
|
||||
|
||||
|
||||
NODE_CLASS_MAPPINGS = {
|
||||
"Autotagger|LP": Autotagger,
|
||||
}
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {
|
||||
"Autotagger|LP": "Autotagger [LP]",
|
||||
}
|
||||
@@ -1,367 +0,0 @@
|
||||
import folder_paths
|
||||
import os
|
||||
from io import BytesIO
|
||||
from llama_cpp import Llama
|
||||
from llama_cpp.llama_chat_format import Llava16ChatHandler
|
||||
import base64
|
||||
from torchvision.transforms import ToPILImage
|
||||
import gc
|
||||
import torch
|
||||
|
||||
|
||||
supported_LLava_extensions = set(['.gguf'])
|
||||
|
||||
try:
|
||||
folder_paths.folder_names_and_paths["LLavacheckpoints"] = (folder_paths.folder_names_and_paths["LLavacheckpoints"][0], supported_LLava_extensions)
|
||||
except:
|
||||
if not os.path.isdir(os.path.join(folder_paths.models_dir, "LLavacheckpoints")):
|
||||
os.mkdir(os.path.join(folder_paths.models_dir, "LLavacheckpoints"))
|
||||
|
||||
folder_paths.folder_names_and_paths["LLavacheckpoints"] = ([os.path.join(folder_paths.models_dir, "LLavacheckpoints")], supported_LLava_extensions)
|
||||
|
||||
class LLavaLoader:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"ckpt_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
|
||||
"max_ctx": ("INT", {"default": 4096, "min": 128, "max": 8192, "step": 64}),
|
||||
"gpu_layers": ("INT", {"default": 27, "min": 0, "max": 100, "step": 1}),
|
||||
"n_threads": ("INT", {"default": 8, "min": 1, "max": 100, "step": 1}),
|
||||
"clip": ("CUSTOM", {"default": ""}),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
RETURN_TYPES = ("CUSTOM",)
|
||||
RETURN_NAMES = ("model",)
|
||||
FUNCTION = "load_llava_checkpoint"
|
||||
|
||||
CATEGORY = "LevelPixel/VLM"
|
||||
def load_llava_checkpoint(self, ckpt_name, max_ctx, gpu_layers, n_threads, clip ):
|
||||
ckpt_path = folder_paths.get_full_path("LLavacheckpoints", ckpt_name)
|
||||
llm = Llama(model_path = ckpt_path, chat_handler=clip,offload_kqv=True, f16_kv=True,
|
||||
use_mlock=False, embedding=False, n_batch=1024, last_n_tokens_size=1024,
|
||||
verbose=True, seed=42, n_ctx = max_ctx, n_gpu_layers=gpu_layers, n_threads=n_threads,
|
||||
logits_all=True, echo=False)
|
||||
return (llm, )
|
||||
|
||||
class LLavaClipLoader:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"clip_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("CUSTOM", )
|
||||
RETURN_NAMES = ("clip", )
|
||||
FUNCTION = "load_clip_checkpoint"
|
||||
|
||||
CATEGORY = "LevelPixel/VLM"
|
||||
def load_clip_checkpoint(self, clip_name):
|
||||
clip_path = folder_paths.get_full_path("LLavacheckpoints", clip_name)
|
||||
clip = Llava16ChatHandler(clip_model_path = clip_path, verbose=False)
|
||||
return (clip, )
|
||||
|
||||
class LLavaSamplerSimple:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"image": ("IMAGE",),
|
||||
"model": ("CUSTOM", {"default": ""}),
|
||||
"temperature": ("FLOAT", {"default": 0.1, "min": 0.01, "max": 1.0, "step": 0.01}),
|
||||
"prompt": ("STRING",{"multiline": True} ),
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("STRING",)
|
||||
FUNCTION = "generate_text_simple"
|
||||
CATEGORY = "LevelPixel/VLM"
|
||||
|
||||
def generate_text_simple(self, image, prompt, model, temperature):
|
||||
|
||||
pil_image = ToPILImage()(image[0].permute(2, 0, 1))
|
||||
|
||||
buffer = BytesIO()
|
||||
pil_image.save(buffer, format="PNG")
|
||||
|
||||
image_bytes = buffer.getvalue()
|
||||
|
||||
base64_string = f"data:image/jpeg;base64,{base64.b64encode(image_bytes).decode('utf-8')}"
|
||||
|
||||
llm = model
|
||||
response = llm.create_chat_completion(
|
||||
messages = [
|
||||
{"role": "system", "content": "You are an assistant who perfectly describes images."},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "image_url", "image_url": {"url" : base64_string}},
|
||||
{"type" : "text", "text": f"{prompt}"}
|
||||
]
|
||||
}
|
||||
|
||||
],
|
||||
temperature = temperature,
|
||||
)
|
||||
|
||||
return (f"{response['choices'][0]['message']['content']}", )
|
||||
|
||||
class LLavaSamplerAdvanced:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"image": ("IMAGE",),
|
||||
"model": ("CUSTOM", {"default": ""}),
|
||||
"max_tokens": ("INT", {"default": 512, "min": 1, "max": 2048, "step": 1}),
|
||||
"temperature": ("FLOAT", {"default": 0.1, "min": 0.01, "max": 1.0, "step": 0.01}),
|
||||
"top_p": ("FLOAT", {"default": 0.95, "min": 0.1, "max": 1.0, "step": 0.01}),
|
||||
"top_k": ("INT", {"default": 40, "step": 1}),
|
||||
"frequency_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
|
||||
"presence_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
|
||||
"repeat_penalty": ("FLOAT", {"default": 1.1, "step": 0.01}),
|
||||
"seed": ("INT", {"default": 42, "step":1}),
|
||||
"prompt": ("STRING",{"multiline": True, "default": ""}),
|
||||
"system_msg": ("STRING",{"multiline": True, "default" : "You are an assistant who perfectly describes images."}),
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("STRING",)
|
||||
FUNCTION = "generate_text_advanced"
|
||||
CATEGORY = "LevelPixel/VLM"
|
||||
|
||||
def generate_text_advanced(self, image, system_msg, prompt, model, max_tokens, temperature, top_p,
|
||||
frequency_penalty, presence_penalty, repeat_penalty, top_k,seed):
|
||||
|
||||
|
||||
pil_image = ToPILImage()(image[0].permute(2, 0, 1))
|
||||
|
||||
buffer = BytesIO()
|
||||
pil_image.save(buffer, format="PNG")
|
||||
|
||||
image_bytes = buffer.getvalue()
|
||||
|
||||
base64_string = f"data:image/jpeg;base64,{base64.b64encode(image_bytes).decode('utf-8')}"
|
||||
|
||||
llm = model
|
||||
response = llm.create_chat_completion(
|
||||
messages = [
|
||||
{"role": "system", "content": system_msg},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "image_url", "image_url": {"url" : base64_string}},
|
||||
{"type" : "text", "text": f"{prompt}"}
|
||||
]
|
||||
}
|
||||
|
||||
],
|
||||
max_tokens = max_tokens,
|
||||
temperature = temperature,
|
||||
top_p = top_p,
|
||||
top_k = top_k,
|
||||
frequency_penalty = frequency_penalty,
|
||||
presence_penalty = presence_penalty,
|
||||
repeat_penalty = repeat_penalty,
|
||||
seed=seed
|
||||
)
|
||||
|
||||
return (f"{response['choices'][0]['message']['content']}", )
|
||||
|
||||
class LLavaSimple:
|
||||
def __init__(self):
|
||||
self.llm = None
|
||||
self.clip = None
|
||||
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"image": ("IMAGE",),
|
||||
"ckpt_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
|
||||
"clip_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
|
||||
"max_ctx": ("INT", {"default": 4096, "min": 128, "max": 128000, "step": 64}),
|
||||
"gpu_layers": ("INT", {"default": 27, "min": 0, "max": 100, "step": 1}),
|
||||
"n_threads": ("INT", {"default": 8, "min": 1, "max": 100, "step": 1}),
|
||||
"temperature": ("FLOAT", {"default": 0.1, "min": 0.01, "max": 1.0, "step": 0.01}),
|
||||
"unload": ("BOOLEAN", {"default": False}),
|
||||
"prompt": ("STRING", {"multiline": True, "forceInput": True}),
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("STRING",)
|
||||
FUNCTION = "generate_text_full_simple"
|
||||
CATEGORY = "LevelPixel/VLM"
|
||||
|
||||
def generate_text_full_simple(self, ckpt_name, clip_name, max_ctx, gpu_layers, n_threads, image, prompt, temperature, unload):
|
||||
|
||||
clip_path = folder_paths.get_full_path("LLavacheckpoints", clip_name)
|
||||
self.clip = Llava16ChatHandler(clip_model_path=clip_path, verbose=False)
|
||||
|
||||
ckpt_path = folder_paths.get_full_path("LLavacheckpoints", ckpt_name)
|
||||
self.llm = Llama(model_path = ckpt_path, chat_handler=self.clip, offload_kqv=True, f16_kv=True,
|
||||
use_mlock=False, embedding=False, n_batch=1024, last_n_tokens_size=1024,
|
||||
verbose=True, seed=42, n_ctx = max_ctx, n_gpu_layers=gpu_layers, n_threads=n_threads,
|
||||
logits_all=True, echo=False)
|
||||
|
||||
pil_image = ToPILImage()(image[0].permute(2, 0, 1))
|
||||
|
||||
buffer = BytesIO()
|
||||
pil_image.save(buffer, format="PNG")
|
||||
|
||||
image_bytes = buffer.getvalue()
|
||||
|
||||
base64_string = f"data:image/jpeg;base64,{base64.b64encode(image_bytes).decode('utf-8')}"
|
||||
|
||||
response = self.llm.create_chat_completion(
|
||||
messages=[
|
||||
{"role": "system", "content": "You are an assistant who perfectly describes images."},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "image_url", "image_url": {"url": base64_string}},
|
||||
{"type": "text", "text": f"{prompt}"}
|
||||
]
|
||||
}
|
||||
],
|
||||
temperature=temperature,
|
||||
)
|
||||
|
||||
if unload and self.llm is not None:
|
||||
self.llm.close()
|
||||
del self.llm
|
||||
self.llm = None
|
||||
gc.collect()
|
||||
torch.cuda.empty_cache()
|
||||
|
||||
|
||||
if unload and self.clip is not None:
|
||||
self.clip._exit_stack.close() # info https://github.com/abetlen/llama-cpp-python/issues/1746
|
||||
del self.clip
|
||||
self.clip = None
|
||||
gc.collect()
|
||||
torch.cuda.empty_cache()
|
||||
|
||||
return (f"{response['choices'][0]['message']['content']}", )
|
||||
|
||||
class LLavaAdvanced:
|
||||
def __init__(self):
|
||||
self.llm = None
|
||||
self.clip = None
|
||||
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"image": ("IMAGE",),
|
||||
"ckpt_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
|
||||
"clip_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
|
||||
"max_ctx": ("INT", {"default": 4096, "min": 128, "max": 128000, "step": 64}),
|
||||
"gpu_layers": ("INT", {"default": 27, "min": 0, "max": 100, "step": 1}),
|
||||
"n_threads": ("INT", {"default": 8, "min": 1, "max": 100, "step": 1}),
|
||||
"max_tokens": ("INT", {"default": 512, "min": 1, "max": 2048, "step": 1}),
|
||||
"temperature": ("FLOAT", {"default": 0.1, "min": 0.01, "max": 1.0, "step": 0.01}),
|
||||
"top_p": ("FLOAT", {"default": 0.95, "min": 0.1, "max": 1.0, "step": 0.01}),
|
||||
"top_k": ("INT", {"default": 40, "step": 1}),
|
||||
"frequency_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
|
||||
"presence_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
|
||||
"repeat_penalty": ("FLOAT", {"default": 1.1, "step": 0.01}),
|
||||
"seed": ("INT", {"default": 42, "step": 1}),
|
||||
"unload": ("BOOLEAN", {"default": False}),
|
||||
"prompt": ("STRING", {"multiline": True, "default": ""}),
|
||||
"system_msg": ("STRING", {"multiline": True, "default": "You are an assistant who perfectly describes images."}),
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("STRING",)
|
||||
FUNCTION = "generate_text_full_advanced"
|
||||
CATEGORY = "LevelPixel/VLM"
|
||||
|
||||
def generate_text_full_advanced(self, ckpt_name, clip_name, max_ctx, gpu_layers, n_threads, image,
|
||||
system_msg, prompt, max_tokens, temperature, top_p, top_k, frequency_penalty,
|
||||
presence_penalty, repeat_penalty, seed, unload):
|
||||
|
||||
clip_path = folder_paths.get_full_path("LLavacheckpoints", clip_name)
|
||||
self.clip = Llava16ChatHandler(clip_model_path=clip_path, verbose=False)
|
||||
|
||||
ckpt_path = folder_paths.get_full_path("LLavacheckpoints", ckpt_name)
|
||||
self.llm = Llama(model_path = ckpt_path, chat_handler=self.clip, offload_kqv=True, f16_kv=True,
|
||||
use_mlock=False, embedding=False, n_batch=1024, last_n_tokens_size=1024,
|
||||
verbose=True, seed=42, n_ctx = max_ctx, n_gpu_layers=gpu_layers, n_threads=n_threads,
|
||||
logits_all=True, echo=False)
|
||||
|
||||
pil_image = ToPILImage()(image[0].permute(2, 0, 1))
|
||||
|
||||
buffer = BytesIO()
|
||||
pil_image.save(buffer, format="PNG")
|
||||
|
||||
|
||||
image_bytes = buffer.getvalue()
|
||||
|
||||
base64_string = f"data:image/jpeg;base64,{base64.b64encode(image_bytes).decode('utf-8')}"
|
||||
|
||||
response = self.llm.create_chat_completion(
|
||||
messages=[
|
||||
{"role": "system", "content": system_msg},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "image_url", "image_url": {"url": base64_string}},
|
||||
{"type": "text", "text": f"{prompt}"}
|
||||
]
|
||||
}
|
||||
],
|
||||
max_tokens=max_tokens,
|
||||
temperature=temperature,
|
||||
top_p=top_p,
|
||||
top_k=top_k,
|
||||
frequency_penalty=frequency_penalty,
|
||||
presence_penalty=presence_penalty,
|
||||
repeat_penalty=repeat_penalty,
|
||||
seed=seed,
|
||||
)
|
||||
|
||||
if unload and self.llm is not None:
|
||||
self.llm.close()
|
||||
del self.llm
|
||||
self.llm = None
|
||||
gc.collect()
|
||||
torch.cuda.empty_cache()
|
||||
|
||||
|
||||
if unload and self.clip is not None:
|
||||
self.clip._exit_stack.close() # info https://github.com/abetlen/llama-cpp-python/issues/1746
|
||||
del self.clip
|
||||
self.clip = None
|
||||
gc.collect()
|
||||
torch.cuda.empty_cache()
|
||||
|
||||
return (f"{response['choices'][0]['message']['content']}", )
|
||||
|
||||
NODE_CLASS_MAPPINGS = {
|
||||
"LLavaLoader|LP": LLavaLoader,
|
||||
"LLavaClipLoader|LP": LLavaClipLoader,
|
||||
"LLavaSamplerSimple|LP": LLavaSamplerSimple,
|
||||
"LLavaSamplerAdvanced|LP": LLavaSamplerAdvanced,
|
||||
"LLavaSimple|LP": LLavaSimple,
|
||||
"LLavaAdvanced|LP": LLavaAdvanced,
|
||||
}
|
||||
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {
|
||||
"LLavaLoader|LP": "LLava Loader [LP]",
|
||||
"LLavaClipLoader|LP": "LLava Clip Loader [LP]",
|
||||
"LLavaSamplerSimple|LP": "LLava Sampler Simple [LP]",
|
||||
"LLavaSamplerAdvanced|LP": "LLava Sampler Advanced [LP]",
|
||||
"LLavaSimple|LP": "LLava Simple [LP]",
|
||||
"LLavaAdvanced|LP": "LLava Advanced [LP]",
|
||||
}
|
||||
Reference in New Issue
Block a user