Joy2 updated

This commit is contained in:
Cyber-BlackCat
2025-03-12 17:03:08 +08:00
committed by GitHub
parent 9c39910bdf
commit 64da7155d9
9 changed files with 1601 additions and 0 deletions
+98
View File
@@ -0,0 +1,98 @@
Update v1.0.2: Joy caption2 added.
# Introduction:
Joy Caption alpha 2 original demo and modelpackage:
https://huggingface.co/spaces/fancyfeast/joy-caption-alpha-two
the repo has taken some reference from: TTPlanetPig/Comfyui_JC2 and https://huggingface.co/John6666/joy-caption-alpha-two-cli-mod
![image](https://github.com/user-attachments/assets/09aac053-b5d0-45d9-b419-3e85b642584c)
# How to use
The main difference between the two versions is the use of LLM models versus LLM's LoRA models.
## base dir which models putin: /comfyUI/models/
model types| joy caption alpha | joy caption 2 | coming soom |
-----------| ----------------------------------- | --------------------------------- | ------------- |
clip_vision|clip_vision/siglip-so400m-patch14-384| "same as alpha" | |
LLM | LLM/Meta-Llama-3.1-8B-bnb-4bit |LLM/Llama-3.1-8B-Lexi-Uncensored-V2| |
| Meta-Llama-3.1-8B | LLM/Meta-Llama-3.1-8B-Instruct | |
loras-LLM| loras-LLM/wpkklhc6 | loras-LLM/cgrkzexw-599808 | |
Notice:Follow these three steps to get started
注意:完成下列三个步骤即可使用
1. 安装依赖requirements.txt(注意:transformers 版本不能太低, windows使用则需要安装windows 版本的相关依赖)
直接点击:install_req.bat 安装依赖
1. Click "install_req.bat" or use cmd code to install requirements, which are necessary.
2. 运行自动下载模型(推荐手动下载)
2. Run the automatic download model (manual download is recommended)
## 手动下载路径:download website:
(1). clip_vision
siglip: https://huggingface.co/google/siglip-so400m-patch14-384
>中国用户请使用: https://www.modelscope.cn/models/AI-ModelScope/siglip-so400m-patch14-384/files
(2). loras-LLM——"必须手动下载 manual download only":
*Joy caption alpha* : https://huggingface.co/spaces/fancyfeast/joy-caption-pre-alpha/tree/main/wpkklhc6 "放到putin" loras-LLM/wpkklhc6
>中国用户请使用: https://www.modelscope.cn/models/fireicewolf/joy-caption-pre-alpha/files
*Joy caption 2* : https://huggingface.co/John6666/joy-caption-alpha-two-cli-mod "放到putin" loras-LLM/cgrkzexw-599808
>中国用户请使用: https://www.modelscope.cn/models/fireicewolf/joy-caption-alpha-two/files
(3). LLM : "推荐手动下载 manual download"
**Joy caption 2**
*Llama-3.1-8B-Lexi-Uncensored-V2*: https://huggingface.co/unsloth/Meta-Llama-3.1-8B-Instruct
>中国用户请使用:https://www.modelscope.cn/models/fireicewolf/Llama-3.1-8B-Lexi-Uncensored-V2/files
*Meta-Llama-3.1-8B-Instruct*: https://huggingface.co/unsloth/Meta-Llama-3.1-8B-Instruct
>中国用户请使用:https://www.modelscope.cn/models/LLM-Research/Meta-Llama-3.1-8B-Instruct/files
**Joy caption alpha**
bnb-4bit: https://huggingface.co/unsloth/Meta-Llama-3.1-8B-bnb-4bit
>中国用户请使用:https://www.modelscope.cn/models/unsloth/Meta-Llama-3.1-8B-Instruct-unsloth-bnb-4bit/files
Llama-3.1-8B: https://huggingface.co/meta-llama/Llama-3.1-8B
>中国用户请使用:https://www.modelscope.cn/models/LLM-Research/Meta-Llama-3.1-8B/files
## Addition
这个报告额外添加了一个“load many images”节点,它将按照图片名从小到大来加载图像,图像不再以错误的顺序加载(是优化版本的Load iamge dir)!!设置index=0使其从第一个图像(图像名称顺序)加载。
This report contains a "load many images" node which is going to load the image set by the order of Num of image from smallest to largest, and the images are NO LONGER loaded in the wrong order!!! Setting index=0 makes it load from the first image (image flie name order).
flux dev运行效果 result runs by flux dev:
![flux](https://github.com/Cyber-BCat/ComfyUI_Auto_Caption/blob/main/workflow/show%20flux%20example.png)
反推效果展示 result with caption screenshot:
![caption](https://github.com/Cyber-BCat/ComfyUI_Auto_Caption/blob/main/workflow/caption.jpg)
https://github.com/Cyber-BCat/ComfyUI_Auto_Caption/blob/main/workflow/autocaption%20exampleworkflow.json
#clip_vision path show:
![image](https://github.com/user-attachments/assets/db311cab-dcbc-454d-b76b-30ae1943de25)
![1](workflow/path-1.png)
#loras-LLM path show:
![image](https://github.com/user-attachments/assets/f5504545-70fa-4cfc-8991-290c73d99167)
![2](workflow/path-autocaption.png)
#LLM path show:
![image](https://github.com/user-attachments/assets/0f7c013c-c319-44ee-9f24-d32f94bf9869)
## 示例工作流下载 workflow example download:
auto caption 2 (joy2):
![JoyCaption2](workflow/auto caption 2 workflow.png)
auto caption (alpha)
![JoyCaption2](workflow/auto caption 2 workflow.png)
# Joy!
+396
View File
@@ -0,0 +1,396 @@
import os
import folder_paths
import torch
import torch.amp.autocast_mode
import re
import numpy as np
from torch import nn
from huggingface_hub import InferenceClient
from transformers import AutoModel, AutoProcessor, AutoTokenizer, PreTrainedTokenizer, PreTrainedTokenizerFast, AutoModelForCausalLM
from pathlib import Path
from PIL import Image, ImageOps
from .lib.ximg import *
from .lib.xmodel import *
from comfy.utils import ProgressBar, common_upscale
class JoyModel:
def __init__(self):
self.clip_model = None
self.clip_processor =None
self.tokenizer = None
self.text_model = None
self.image_adapter = None
self.parent = None
def clearCache(self):
self.clip_model = None
self.clip_processor =None
self.tokenizer = None
self.text_model = None
self.image_adapter = None
class ImageAdapter(nn.Module):
def __init__(self, input_features: int, output_features: int):
super().__init__()
self.linear1 = nn.Linear(input_features, output_features)
self.activation = nn.GELU()
self.linear2 = nn.Linear(output_features, output_features)
def forward(self, vision_outputs: torch.Tensor):
x = self.linear1(vision_outputs)
x = self.activation(x)
x = self.linear2(x)
return x
class Joy_Model_load:
def __init__(self):
self.model = None
self.pipeline = JoyModel()
self.pipeline.parent = self
pass
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"model": (["unsloth/Meta-Llama-3.1-8B-bnb-4bit", "meta-llama/Meta-Llama-3.1-8B"],),
}
}
CATEGORY = "Auto Caption"
RETURN_TYPES = ("JoyModel",)
FUNCTION = "gen"
def loadCheckPoint(self):
# 清除一波
if self.pipeline != None:
self.pipeline.clearCache()
# clip
model_id = "google/siglip-so400m-patch14-384"
CLIP_PATH = download_hg_model(model_id,"clip_vision")
clip_processor = AutoProcessor.from_pretrained(CLIP_PATH)
clip_model = AutoModel.from_pretrained(
CLIP_PATH,
trust_remote_code=True
)
clip_model = clip_model.vision_model
clip_model.eval()
clip_model.requires_grad_(False)
clip_model.to("cuda")
# LLM
MODEL_PATH = download_hg_model(self.model,"LLM")
LORA_PATH = os.path.join(folder_paths.models_dir, "loras-LLM", "wpkklhc6")
tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH, use_fast=True)
# tokenizer = AutoTokenizer.from_pretrained(os.path.join(CAPTION_PATH, "text_model"), use_fast=True)
assert isinstance(tokenizer, PreTrainedTokenizer) or isinstance(tokenizer, PreTrainedTokenizerFast), f"Tokenizer is of type {type(tokenizer)}"
text_model = AutoModelForCausalLM.from_pretrained(MODEL_PATH, device_map="auto",trust_remote_code=True)
text_model.eval()
# Image Adapter
adapter_path = os.path.join(LORA_PATH,"image_adapter.pt")
image_adapter = ImageAdapter(clip_model.config.hidden_size, text_model.config.hidden_size) # ImageAdapter(clip_model.config.hidden_size, 4096)
image_adapter.load_state_dict(torch.load(adapter_path, map_location="cpu"))
adjusted_adapter = image_adapter #AdjustedImageAdapter(image_adapter, text_model.config.hidden_size)
adjusted_adapter.eval()
adjusted_adapter.to("cuda")
self.pipeline.clip_model = clip_model
self.pipeline.clip_processor = clip_processor
self.pipeline.tokenizer = tokenizer
self.pipeline.text_model = text_model
self.pipeline.image_adapter = adjusted_adapter
def clearCache(self):
if self.pipeline != None:
self.pipeline.clearCache()
def gen(self,model):
if self.model == None or self.model != model or self.pipeline == None:
self.model = model
self.loadCheckPoint()
return (self.pipeline,)
class Auto_Caption:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"JoyModel": ("JoyModel",),
"image": ("IMAGE",),
"prompt": ("STRING", {"multiline": True, "default": "A descriptive caption for this image"},),
"max_new_tokens":("INT", {"default": 1024, "min": 10, "max": 4096, "step": 1}),
"temperature": ("FLOAT", {"default": 0.6, "min": 0.0, "max": 1.0, "step": 0.01}),
"cache": ("BOOLEAN", {"default": False}),
}
}
CATEGORY = "Auto Caption"
RETURN_TYPES = ("STRING",)
FUNCTION = "gen"
def gen(self,JoyModel,image,prompt,max_new_tokens,temperature,cache):
if JoyModel.clip_processor == None :
JoyModel.parent.loadCheckPoint()
clip_processor = JoyModel.clip_processor
tokenizer = JoyModel.tokenizer
clip_model = JoyModel.clip_model
image_adapter = JoyModel.image_adapter
text_model = JoyModel.text_model
input_image = tensor2pil(image)
# Preprocess image
pImge = clip_processor(images=input_image, return_tensors='pt').pixel_values
pImge = pImge.to('cuda')
# Tokenize the prompt
prompt = tokenizer.encode(prompt, return_tensors='pt', padding=False, truncation=False, add_special_tokens=False)
# Embed image
with torch.amp.autocast_mode.autocast('cuda', enabled=True):
vision_outputs = clip_model(pixel_values=pImge, output_hidden_states=True)
image_features = vision_outputs.hidden_states[-2]
embedded_images = image_adapter(image_features)
embedded_images = embedded_images.to('cuda')
# Embed prompt
prompt_embeds = text_model.model.embed_tokens(prompt.to('cuda'))
assert prompt_embeds.shape == (1, prompt.shape[1], text_model.config.hidden_size), f"Prompt shape is {prompt_embeds.shape}, expected {(1, prompt.shape[1], text_model.config.hidden_size)}"
embedded_bos = text_model.model.embed_tokens(torch.tensor([[tokenizer.bos_token_id]], device=text_model.device, dtype=torch.int64))
# Construct prompts
inputs_embeds = torch.cat([
embedded_bos.expand(embedded_images.shape[0], -1, -1),
embedded_images.to(dtype=embedded_bos.dtype),
prompt_embeds.expand(embedded_images.shape[0], -1, -1),
], dim=1)
input_ids = torch.cat([
torch.tensor([[tokenizer.bos_token_id]], dtype=torch.long),
torch.zeros((1, embedded_images.shape[1]), dtype=torch.long),
prompt,
], dim=1).to('cuda')
attention_mask = torch.ones_like(input_ids)
generate_ids = text_model.generate(input_ids, inputs_embeds=inputs_embeds, attention_mask=attention_mask, max_new_tokens=max_new_tokens, do_sample=True, top_k=10, temperature=temperature, suppress_tokens=None)
# Trim off the prompt
generate_ids = generate_ids[:, input_ids.shape[1]:]
if generate_ids[0][-1] == tokenizer.eos_token_id:
generate_ids = generate_ids[:, :-1]
caption = tokenizer.batch_decode(generate_ids, skip_special_tokens=False, clean_up_tokenization_spaces=False)[0]
r = caption.strip()
if cache == False:
JoyModel.parent.clearCache()
return (r,)
class LoadImagesRezise:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"folder": ("STRING", {"default": ""}),
},
"optional": {
"image_load_cap": ("INT", {"default": 50, "min": 0, "step": 1}),
"start_index": ("INT", {"default": 0, "min": 0, "step": 1}),
}
}
RETURN_TYPES = ("IMAGE", "MASK", "INT", "STRING",)
RETURN_NAMES = ("image", "mask", "count", "image_path",)
FUNCTION = "load_images"
CATEGORY = "Auto Caption"
def load_images(self, folder, image_load_cap, start_index):
if not os.path.isdir(folder):
raise FileNotFoundError(f"Folder '{folder}' cannot be found.")
dir_files = os.listdir(folder)
if len(dir_files) == 0:
raise FileNotFoundError(f"No files in directory '{folder}'.")
# Filter files by valid image extensions
valid_extensions = ['.jpg', '.jpeg', '.png', '.webp']
dir_files = [f for f in dir_files if any(f.lower().endswith(ext) for ext in valid_extensions)]
# Sort files based on numeric value extracted from filename
def extract_number(file_name):
match = re.search(r'(\d+)', file_name)
return int(match.group(0)) if match else float('inf') # Use 'inf' if no number is found to push such files at the end
dir_files = sorted(dir_files, key=extract_number)
# Convert to full file paths
dir_files = [os.path.join(folder, x) for x in dir_files]
# Start at the specified start_index
dir_files = dir_files[start_index:]
images = []
masks = []
image_path_list = []
limit_images = False
if image_load_cap > 0:
limit_images = True
image_count = 0
has_non_empty_mask = False
for image_path in dir_files:
if os.path.isdir(image_path):
continue
if limit_images and image_count >= image_load_cap:
break
i = Image.open(image_path)
i = ImageOps.exif_transpose(i) # Handle EXIF orientation
image = i.convert("RGB")
image = np.array(image).astype(np.float32) / 255.0
image = torch.from_numpy(image)[None,] # Add a batch dimension
if 'A' in i.getbands(): # Check for alpha channel
mask = np.array(i.getchannel('A')).astype(np.float32) / 255.0
mask = 1. - torch.from_numpy(mask) # Invert the alpha mask
has_non_empty_mask = True
else:
mask = torch.zeros((64, 64), dtype=torch.float32, device="cpu")
images.append(image)
masks.append(mask)
image_path_list.append(image_path)
image_count += 1
if len(images) == 1:
return (images[0], masks[0], 1)
elif len(images) > 1:
image1 = images[0]
mask1 = None
for image2 in images[1:]:
if image1.shape[1:] != image2.shape[1:]:
image2 = common_upscale(image2.movedim(-1, 1), image1.shape[2], image1.shape[1], "bilinear", "center").movedim(1, -1)
image1 = torch.cat((image1, image2), dim=0)
for mask2 in masks[1:]:
if has_non_empty_mask:
if image1.shape[1:3] != mask2.shape:
mask2 = torch.nn.functional.interpolate(mask2.unsqueeze(0).unsqueeze(0), size=(image1.shape[2], image1.shape[1]), mode='bilinear', align_corners=False)
mask2 = mask2.squeeze(0)
else:
mask2 = mask2.unsqueeze(0)
else:
mask2 = mask2.unsqueeze(0)
if mask1 is None:
mask1 = mask2
else:
mask1 = torch.cat((mask1, mask2), dim=0)
return (image1, mask1, len(images), image_path_list)
class LoadManyImages:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"folder": ("STRING", {"default": ""}),
},
"optional": {
"image_load_cap": ("INT", {"default": 50, "min": 0, "step": 1}),
"start_index": ("INT", {"default": 0, "min": 0, "step": 1}),
}
}
RETURN_TYPES = ("IMAGE", "MASK", "INT", "STRING",)
RETURN_NAMES = ("image", "mask", "count", "image_path",)
OUTPUT_IS_LIST = (True, True, True, True)
FUNCTION = "load_images"
CATEGORY = "Auto Caption"
def load_images(self, folder, image_load_cap, start_index):
if not os.path.isdir(folder):
raise FileNotFoundError(f"Folder '{folder}' cannot be found.")
dir_files = os.listdir(folder)
if len(dir_files) == 0:
raise FileNotFoundError(f"No files in directory '{folder}'.")
# Filter files by valid image extensions
valid_extensions = ['.jpg', '.jpeg', '.png', '.webp']
dir_files = [f for f in dir_files if any(f.lower().endswith(ext) for ext in valid_extensions)]
# Sort files based on numeric value extracted from filename
def extract_number(file_name):
match = re.search(r'(\d+)', file_name)
return int(match.group(0)) if match else float('inf') # Use 'inf' if no number is found to push such files at the end
dir_files = sorted(dir_files, key=extract_number)
#
# Convert to full file paths
dir_files = [os.path.join(folder, x) for x in dir_files]
# Start at the specified start_index
dir_files = dir_files[start_index:]
images = []
masks = []
image_path_list = []
limit_images = False
if image_load_cap > 0:
limit_images = True
image_count = 0
for image_path in dir_files:
if os.path.isdir(image_path) and os.path.ex:
continue
if limit_images and image_count >= image_load_cap:
break
i = Image.open(image_path)
i = ImageOps.exif_transpose(i) # Handle EXIF orientation
image = i.convert("RGB")
image = np.array(image).astype(np.float32) / 255.0
image = torch.from_numpy(image)[None,]
if 'A' in i.getbands(): # Check for alpha channel
mask = np.array(i.getchannel('A')).astype(np.float32) / 255.0
mask = 1. - torch.from_numpy(mask) # Invert the alpha mask
else:
mask = torch.zeros((64, 64), dtype=torch.float32, device="cpu")
images.append(image)
masks.append(mask)
image_path_list.append(image_path)
image_count += 1
return (images, masks, image_path_list)
+836
View File
@@ -0,0 +1,836 @@
import os
import torch
import torch.amp.autocast_mode
import re
import numpy as np
import shutil
from torch import nn
from huggingface_hub import InferenceClient
from transformers import AutoModel, AutoProcessor, AutoTokenizer, PreTrainedTokenizer, PreTrainedTokenizerFast, AutoModelForCausalLM
import torchvision.transforms.functional as TVF
from pathlib import Path
from PIL import Image, ImageOps
from typing import List, Union
from .lib.ximg import *
from .lib.xmodel import *
from comfy.utils import ProgressBar, common_upscale
import comfy.model_management as mm
import comfy.sd
import folder_paths
class JoyModel2:
def __init__(self):
self.clip_model = None
self.clip_processor =None
self.llm_model = None
self.tokenizer = None
self.image_adapter = None
self.parent = None
def clearCache(self):
self.clip_model = None
self.clip_processor =None
self.tokenizer = None
self.llm_model = None
self.image_adapter = None
class ImageAdapter(nn.Module):
def __init__(self, input_features: int, output_features: int, ln1: bool, pos_emb: bool, num_image_tokens: int,
deep_extract: bool):
super().__init__()
self.deep_extract = deep_extract
if self.deep_extract:
input_features = input_features * 5
self.linear1 = nn.Linear(input_features, output_features)
self.activation = nn.GELU()
self.linear2 = nn.Linear(output_features, output_features)
self.ln1 = nn.Identity() if not ln1 else nn.LayerNorm(input_features)
self.pos_emb = None if not pos_emb else nn.Parameter(torch.zeros(num_image_tokens, input_features))
# Other tokens (<|image_start|>, <|image_end|>, <|eot_id|>)
self.other_tokens = nn.Embedding(3, output_features)
self.other_tokens.weight.data.normal_(mean=0.0, std=0.02) # Matches HF's implementation of LLaMA
def forward(self, vision_outputs: torch.Tensor):
if self.deep_extract:
x = torch.cat((
vision_outputs[-2],
vision_outputs[3],
vision_outputs[7],
vision_outputs[13],
vision_outputs[20],
), dim=-1)
assert len(x.shape) == 3, f"Expected 3, got {len(x.shape)}" # batch, tokens, features
assert x.shape[-1] == vision_outputs[-2].shape[-1] * 5, f"Expected {vision_outputs[-2].shape[-1] * 5}, got {x.shape[-1]}"
else:
x = vision_outputs[-2]
x = self.ln1(x)
if self.pos_emb is not None:
assert x.shape[-2:] == self.pos_emb.shape, f"Expected {self.pos_emb.shape}, got {x.shape[-2:]}"
x = x + self.pos_emb
x = self.linear1(x)
x = self.activation(x)
x = self.linear2(x)
other_tokens = self.other_tokens(
torch.tensor([0, 1], device=self.other_tokens.weight.device).expand(x.shape[0], -1))
assert other_tokens.shape == (
x.shape[0], 2, x.shape[2]), f"Expected {(x.shape[0], 2, x.shape[2])}, got {other_tokens.shape}"
x = torch.cat((other_tokens[:, 0:1], x, other_tokens[:, 1:2]), dim=1)
return x
def get_eot_embedding(self):
return self.other_tokens(torch.tensor([2], device=self.other_tokens.weight.device)).squeeze(0)
def llmloader(model_path, dtype, device="cuda:0", device_map=None):
global current_device
current_device = device # 设置当前设备
from transformers import AutoModel, AutoProcessor, AutoTokenizer, PreTrainedTokenizer, PreTrainedTokenizerFast, AutoModelForCausalLM
from peft import PeftModel
JC_lora = "text_model"
use_lora = True if JC_lora != "none" else False
CLIP_PATH = os.path.join(folder_paths.models_dir, "clip_vision", "siglip-so400m-patch14-384")
CAPTION_PATH = os.path.join(folder_paths.models_dir, "loras-LLM", "cgrkzexw-599808")
LORA_PATH = os.path.join(CAPTION_PATH, "text_model")
# 加载siglip或者下载
model_id = "google/siglip-so400m-patch14-384"
if os.path.exists(CLIP_PATH):
print("Start to load existing VLM")
else:
print("VLM not found locally. Downloading google/siglip-so400m-patch14-384...")
try:
# 下载clip(内含tokenzer与4bit量化版LLM一致),snapshot函数中已构建目录
CLIP_PATH = download_hg_model(model_id,"clip_vision")
except Exception as e:
print(f"Error downloading CLIP model: {e}")
raise
try:
if dtype == "nf4":
from transformers import BitsAndBytesConfig
nf4_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_use_double_quant=True,
bnb_4bit_compute_dtype=torch.bfloat16
)
print("Loading in NF4")
print("Loading CLIP")
clip_processor = AutoProcessor.from_pretrained(CLIP_PATH)
clip_model = AutoModel.from_pretrained(CLIP_PATH,trust_remote_code=True).vision_model
print("Loading VLM's custom vision model")
checkpoint = torch.load(os.path.join(CAPTION_PATH, "clip_model.pt"), map_location=current_device, weights_only=False)
checkpoint = {k.replace("_orig_mod.module.", ""): v for k, v in checkpoint.items()}
clip_model.load_state_dict(checkpoint)
del checkpoint
clip_model.eval()
clip_model.requires_grad_(False).to(current_device)
print(f"Loading LLM: {model_path}")
llm_model = AutoModelForCausalLM.from_pretrained(
model_path,
quantization_config=nf4_config,
device_map=current_device, # 统一使用指定设备
torch_dtype=torch.bfloat16
).eval()
if use_lora and os.path.exists(LORA_PATH):
print("Loading VLM's custom text model")
llm_model = PeftModel.from_pretrained(
model=llm_model,
model_id=LORA_PATH,
device_map=current_device, # 统一使用指定设备
quantization_config=nf4_config
)
llm_model = llm_model.merge_and_unload(safe_merge=True)
else:
print("VLM's custom text model isn't loaded")
else: # 选用 bf16
print("Loading in bfloat16")
print("Loading CLIP")
clip_processor = AutoProcessor.from_pretrained(CLIP_PATH)
clip_model = AutoModel.from_pretrained(CLIP_PATH,trust_remote_code=True).vision_model
if os.path.exists(os.path.join(CAPTION_PATH, "clip_model.pt")):
print("Loading VLM's custom vision model")
checkpoint = torch.load(os.path.join(CAPTION_PATH, "clip_model.pt"), map_location=current_device, weights_only=False)
checkpoint = {k.replace("_orig_mod.module.", ""): v for k, v in checkpoint.items()}
clip_model.load_state_dict(checkpoint)
del checkpoint
clip_model.eval().requires_grad_(False)
clip_model.to(current_device)
# print("Loading tokenizer")
# tokenizer = AutoTokenizer.from_pretrained(LORA_PATH, use_fast=True)
# assert isinstance(tokenizer, (PreTrainedTokenizer, PreTrainedTokenizerFast)), f"Tokenizer is of type {type(tokenizer)}"
print(f"Loading LLM: {model_path}")
llm_model = AutoModelForCausalLM.from_pretrained(
model_path,
device_map=current_device, # 统一使用指定设备
torch_dtype=torch.bfloat16
)
llm_model.eval()
if use_lora and os.path.exists(LORA_PATH):
print("Loading VLM's custom text model")
llm_model = PeftModel.from_pretrained(
model=llm_model,
model_id=LORA_PATH,
device_map=current_device # 统一使用指定设备
)
llm_model = llm_model.merge_and_unload(safe_merge=True)
else:
print("VLM's custom text model isn't loaded")
except Exception as e:
print(f"Error loading models: {e}", )
finally:
pass # 可以在这里添加内存释放逻辑(如果需要)
return (clip_model, clip_processor, llm_model)
# return JoyModel2(clip_model,clip_processor, llm_model, None, None)
class Joy_Model2_load:
def __init__(self):
self.llm_model = None
self.parent = None
self.pipeline = JoyModel2()
self.pipeline.parent = self
pass
@classmethod
def INPUT_TYPES(cls):
llm_model_list = ["unsloth/Meta-Llama-3.1-8B-Instruct", "Orenguteng/Llama-3.1-8B-Lexi-Uncensored-V2"]
dtype_list = ['nf4', 'bf16']
# 获取可用的GPU设备列表
gpu_devices = [f"cuda:{i}" for i in range(torch.cuda.device_count())]
if not gpu_devices:
gpu_devices = ["cpu"] # 如果没有GPU可用,则仅提供CPU选项
#input widget
return {
"required": {
"llm_model": (llm_model_list,),
"dtype": (dtype_list,),
# "cache_model": ("BOOLEAN", {"default": False}),
# 启用此项需要多卡
# muti-GPU setting is required
"device": (gpu_devices,),
}
}
CATEGORY = "Auto Caption"
RETURN_TYPES = ("JoyModel2",)
FUNCTION = "gen"
def loadCheckPoint(self, llm_model, dtype, cache_model, device="cuda:0"):
# cleanup
if self.pipeline != None:
self.pipeline.clearCache()
# LLM
#LLM路径构造
comfy_model_dir = os.path.join(folder_paths.models_dir, "LLM")
print(f"comfy_model_dir: {comfy_model_dir}")
if not os.path.exists(comfy_model_dir):
os.mkdir(comfy_model_dir)
leach_model_name = llm_model.split('/')[-1]
llm_model_path = os.path.join(comfy_model_dir, leach_model_name)
llm_model_path_cache = os.path.join(comfy_model_dir, "cache--" + leach_model_name)
# device chose
selected_device = device if torch.cuda.is_available() else 'cpu'
model_loaded_on = selected_device # 跟踪模型加载在哪个设备上
#load or download
try:
if os.path.exists(llm_model_path):
print(f"Start to load existing model on {selected_device}")
else: #auto download from hg
download_hg_model(llm_model,llm_model_path_cache)
shutil.move(llm_model_path_cache, llm_model_path)
print(f"Model downloaded to {llm_model_path_cache}...")
if self.parent is None:
try:
# 尝试加载模型
free_vram_bytes = mm.get_free_memory()
free_vram_gb = free_vram_bytes / (1024 ** 3)
print(f"Free VRAM: {free_vram_gb:.2f} GB")
if dtype == 'nf4' and free_vram_gb < 10:
print("Free VRAM is less than 10GB when loading 'nf4' model. Performing VRAM cleanup.")
cleanGPU()
elif dtype == 'bf16' and free_vram_gb < 20:
print("Free VRAM is less than 20GB when loading 'bf16' model. Performing VRAM cleanup.")
cleanGPU()
# load LLM,使用所选设备。解包返回所需要的模型值
modelspackage = llmloader(
llm_model_path, dtype, device=selected_device, device_map=None)
#定义中间属性,确认模型缓存
# self.clip_model = clip_model
# self.clip_processor = clip_processor
# self.llm_model = llm_model
except RuntimeError:
print("An error occurred while loading the model. Please check your configuration.")
else:
modelspackage=self.parent
# self.pipeline.clip_model = self.clip_model
# self.pipeline.clip_processor = self.clip_processor
# self.pipeline.llm_model = self.llm_model
except Exception as e:
print(f"Error loading model: {e}")
return None
print(f"Model loaded on {model_loaded_on}")
# # clip及lora和对应的tokenizer
# model_id = "google/siglip-so400m-patch14-384"
# CLIP_PATH = download_hg_model(model_id,"clip_vision")
CAPTION_PATH = os.path.join(folder_paths.models_dir, "loras-LLM", "cgrkzexw-599808")
# clip_processor = AutoProcessor.from_pretrained(CLIP_PATH)
# clip_model = AutoModel.from_pretrained(
# CLIP_PATH,
# trust_remote_code=True
# )
# clip_model = clip_model.vision_model
# clip_model.eval()
# clip_model.requires_grad_(False)
# clip_model.to("cuda")
# 加载LLM
# llm_model = AutoModelForCausalLM.from_pretrained(
# MODEL_PATH,
# device_map="auto",
# trust_remote_code=True)
# llm_model.eval()
tokenizer = AutoTokenizer.from_pretrained(llm_model_path, use_fast=True)
assert isinstance(tokenizer, PreTrainedTokenizer) or isinstance(tokenizer, PreTrainedTokenizerFast), f"Tokenizer is of type {type(tokenizer)}"
# load Image Adapter
print("Loading image adapter")
global current_device
current_device = device # 设置当前设备
adapter_path = os.path.join(CAPTION_PATH,"image_adapter.pt")
#解包三个模型
# clip_model = modelspackage[0]
# clip_processor = modelspackage[1]
# llm_model = modelspackage[2]
self.clip_model, self.clip_processor, self.llm_model = modelspackage
#统一用cuda加载imgadpter
self.image_adapter = ImageAdapter(
self.clip_model.config.hidden_size,
self.llm_model.config.hidden_size,
False, False, 38,
False ) # ImageAdapter(clip_model.config.hidden_size, 4096)
self.image_adapter.load_state_dict(torch.load(adapter_path, map_location=current_device, weights_only=False))
adjusted_adapter = self.image_adapter #AdjustedImageAdapter(image_adapter, llm_model.config.hidden_size)
adjusted_adapter.eval()
adjusted_adapter.to("cuda")
# print("Loading image adapter")
# image_adapter = ImageAdapter(
# clip_model.config.hidden_size,
# llm_model.config.hidden_size,
# False, False, 38,
# False
# ).eval()
# image_adapter.to(current_device)
# image_adapter.load_state_dict(
# torch.load(os.path.join(CAPTION_PATH, "image_adapter.pt"),
# map_location=current_device, weights_only=False)
# )
# pipeline ready for output
self.pipeline.clip_model = self.clip_model
self.pipeline.clip_processor = self.clip_processor
self.pipeline.llm_model = self.llm_model
self.pipeline.tokenizer = tokenizer
self.pipeline.image_adapter = adjusted_adapter
# 用于此函数内一开始的清除
def clearCache(self):
if self.pipeline != None:
self.pipeline.clearCache()
def gen(self, llm_model, dtype, device="cuda:0"):
if self.llm_model == None or self.llm_model != llm_model or self.pipeline == None:
self.llm_model = llm_model
self.loadCheckPoint(llm_model, dtype, True, device)
return (self.pipeline,)
class Auto_Caption2:
CATEGORY = "Auto Caption"
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("prompt",)
# OUTPUT_NODE = True
OUTPUT_IS_LIST = (True,)
FUNCTION = "gen"
def __init__(self):
self.NODE_NAME = 'Auto Caption 2'
self.previous_model = None
@classmethod
def INPUT_TYPES(cls):
caption_type_list = [
"Descriptive", "Descriptive (Informal)", "Training Prompt", "MidJourney",
"Booru tag list", "Booru-like tag list", "Art Critic", "Product Listing",
"Social Media Post"
]
caption_length_list = [
"any", "very short", "short", "medium-length", "long", "very long"
] + [str(i) for i in range(20, 261, 5)]
# 获取可用的GPU设备列表
gpu_devices = [f"cuda:{i}" for i in range(torch.cuda.device_count())]
if not gpu_devices:
gpu_devices = ["cpu"] # 如果没有GPU可用,则仅提供CPU选项
return {
"required": {
"JoyModel2": ("JoyModel2",),
"image": ("IMAGE",),
"caption_type": (caption_type_list,),
"caption_length": (caption_length_list,),
"user_prompt": ("STRING", {"default": "", "multiline": True}),
"top_p": ("FLOAT", {"default": 0.8, "min": 0, "max": 1, "step": 0.01}),
"temperature": ("FLOAT", {"default": 0.6, "min": 0.0, "max": 1.0, "step": 0.01}),
"max_new_tokens": ("INT", {"default": 1024, "min": 8, "max": 4096, "step": 1}),
"cache": ("BOOLEAN", {"default": False}),
"device": (gpu_devices,), # 新增GPU设备选择
},
"optional": {
"ExtraOptionsSet": ("STRING",{"forceInput": True}), # 接收来自 ExtraOptionsNode 的单一字符串
},
}
@classmethod
def gen(self,JoyModel2,image,
caption_type,caption_length, user_prompt,
top_p, temperature, max_new_tokens,
cache,device,
ExtraOptionsSet=None):
# if JoyModel2.clip_processor == None :
# JoyModel2.parent.loadCheckPoint()
# clip_processor = JoyModel2.clip_processor
# clip_model = JoyModel2.clip_model
# tokenizer = JoyModel2.tokenizer
# image_adapter = JoyModel2.image_adapter
# llm_model = JoyModel2.llm_model
# 接收来自 ExtraOptionsNode 的额外提示
extra = []
if ExtraOptionsSet and ExtraOptionsSet.strip():
extra = [ExtraOptionsSet] # 将单一字符串包装成列表
print(f"Extra options enabled: {ExtraOptionsSet}")
else:
print("No extra options provided.")
# Preprocess image
ret_text = []
input_image = [tensor2pil(img) for img in image]
try:
captions = stream_chat(
input_image, caption_type, caption_length,
extra, "", user_prompt,
max_new_tokens, top_p, temperature, len(input_image),
JoyModel2, device # 确保传递正确的设备
)
ret_text.extend(captions)
except Exception as e:
print(f"Error during stream_chat: {e}")
return ("Error generating captions.",)
if cache == False:
del JoyModel2
free_memory()
return (ret_text,)
# Process image
pImge = clip_processor(images=input_image, return_tensors='pt').pixel_values
pImge = pImge.to('cuda')
# Tokenize the prompt
user_prompt = tokenizer.encode(user_prompt, return_tensors='pt', padding=False, truncation=False, add_special_tokens=False)
# Embed image
with torch.amp.autocast_mode.autocast('cuda', enabled=True):
vision_outputs = clip_model(pixel_values=pImge, output_hidden_states=True)
image_features = vision_outputs.hidden_states[-2]
embedded_images = image_adapter(image_features)
embedded_images = embedded_images.to('cuda')
# Embed prompt
prompt_embeds = llm_model.model.embed_tokens(user_prompt.to('cuda'))
assert prompt_embeds.shape == (1, user_prompt.shape[1], llm_model.config.hidden_size), f"Prompt shape is {prompt_embeds.shape}, expected {(1, prompt.shape[1], llm_model.config.hidden_size)}"
embedded_bos = llm_model.model.embed_tokens(torch.tensor([[tokenizer.bos_token_id]], device=llm_model.device, dtype=torch.int64))
# Construct prompts
inputs_embeds = torch.cat([
embedded_bos.expand(embedded_images.shape[0], -1, -1),
embedded_images.to(dtype=embedded_bos.dtype),
prompt_embeds.expand(embedded_images.shape[0], -1, -1),
], dim=1)
input_ids = torch.cat([
torch.tensor([[tokenizer.bos_token_id]], dtype=torch.long),
torch.zeros((1, embedded_images.shape[1]), dtype=torch.long),
user_prompt,
], dim=1).to('cuda')
attention_mask = torch.ones_like(input_ids)
generate_ids = llm_model.generate(input_ids, inputs_embeds=inputs_embeds, attention_mask=attention_mask, max_new_tokens=max_new_tokens, do_sample=True, top_k=10, temperature=temperature, suppress_tokens=None)
# Trim off the prompt
generate_ids = generate_ids[:, input_ids.shape[1]:]
if generate_ids[0][-1] == tokenizer.eos_token_id:
generate_ids = generate_ids[:, :-1]
caption = tokenizer.batch_decode(generate_ids, skip_special_tokens=False, clean_up_tokenization_spaces=False)[0]
r = caption.strip()
if cache == False:
JoyModel2.parent.clearCache()
return (r,)
class ExtraOptionsSet:
CATEGORY = 'Auto Caption'
FUNCTION = 'extra_options'
RETURN_TYPES = ("STRING",) # 改为返回单一字符串
RETURN_NAMES = ("ExtraOptionsSet",)
OUTPUT_IS_LIST = (False,) # 单一字符串输出
def __init__(self):
self.NODE_NAME = 'ExtraOptionsSet'
@classmethod
def INPUT_TYPES(cls):
# 获取 extra_option.json 的路径并加载选项
current_dir = os.path.dirname(os.path.abspath(__file__))
extra_option_file = os.path.join(current_dir, "lib","extra_option.json")
extra_options_list = {}
if os.path.isfile(extra_option_file):
try:
with open(extra_option_file, "r", encoding='utf-8') as f:
json_content = json.load(f)
for item in json_content:
option_name = item.get("name")
if option_name:
# 定义每个额外选项为布尔输入
extra_options_list[option_name] = ("BOOLEAN", {"default": False})
except Exception as e:
print(f"Error loading extra_option.json: {e}")
else:
print(f"extra_option.json not found at {extra_option_file}. No extra options will be available.")
# 定义输入字段,包括开关和 character_name
return {
"required": {
"enable_extra_options": ("BOOLEAN", {"default": True, "label": "启用额外选项"}), # 开关
**extra_options_list, # 动态加载的额外选项
"character_name": ("STRING", {"default": "", "multiline": False}), # 移动 character_name
},
}
def extra_options(self, enable_extra_options, character_name, **extra_options):
"""
处理额外选项并返回已启用的提示列表。
如果启用了替换角色名称选项,并提供了 character_name,则进行替换。
"""
extra_prompts = []
if enable_extra_options:
base_dir = os.path.dirname(os.path.abspath(__file__))
extra_option_file = os.path.join(base_dir, "lib","extra_option.json")
if os.path.isfile(extra_option_file):
try:
with open(extra_option_file, "r", encoding='utf-8') as f:
json_content = json.load(f)
for item in json_content:
name = item.get("name")
prompt = item.get("prompt")
if name and prompt:
if extra_options.get(name):
# 如果 prompt 中包含 {name},则替换为 character_name
if "{name}" in prompt:
prompt = prompt.replace("{name}", character_name)
extra_prompts.append(prompt)
except Exception as e:
print(f"Error reading extra_option.json: {e}")
else:
print(f"extra_option.json not found at {extra_option_file} during processing.")
# 将所有启用的提示拼接成一个字符串
return (" ".join(extra_prompts),) # 返回一个单一的合并字符串
def stream_chat(input_images: List[Image.Image], caption_type: str, caption_length: Union[str, int],
extra_options: list[str], name_input: str, custom_prompt: str,
max_new_tokens: int, top_p: float, temperature: float, batch_size: int,
model: JoyModel2, current_device=str):
# 确定 chat_device
if 'cuda' in current_device:
chat_device = 'cuda'
elif 'cpu' in current_device:
chat_device = 'cpu'
else:
raise ValueError(f"Unsupported device type: {current_device}")
CAPTION_TYPE_MAP = {
"Descriptive": [
"Write a descriptive caption for this image in a formal tone.",
"Write a descriptive caption for this image in a formal tone within {word_count} words.",
"Write a {length} descriptive caption for this image in a formal tone.",
],
"Descriptive (Informal)": [
"Write a descriptive caption for this image in a casual tone.",
"Write a descriptive caption for this image in a casual tone within {word_count} words.",
"Write a {length} descriptive caption for this image in a casual tone.",
],
"Training Prompt": [
"Write a stable diffusion prompt for this image.",
"Write a stable diffusion prompt for this image within {word_count} words.",
"Write a {length} stable diffusion prompt for this image.",
],
"MidJourney": [
"Write a MidJourney prompt for this image.",
"Write a MidJourney prompt for this image within {word_count} words.",
"Write a {length} MidJourney prompt for this image.",
],
"Booru tag list": [
"Write a list of Booru tags for this image.",
"Write a list of Booru tags for this image within {word_count} words.",
"Write a {length} list of Booru tags for this image.",
],
"Booru-like tag list": [
"Write a list of Booru-like tags for this image.",
"Write a list of Booru-like tags for this image within {word_count} words.",
"Write a {length} list of Booru-like tags for this image.",
],
"Art Critic": [
"Analyze this image like an art critic would with information about its composition, style, symbolism, the use of color, light, any artistic movement it might belong to, etc.",
"Analyze this image like an art critic would with information about its composition, style, symbolism, the use of color, light, any artistic movement it might belong to, etc. Keep it within {word_count} words.",
"Analyze this image like an art critic would with information about its composition, style, symbolism, the use of color, light, any artistic movement it might belong to, etc. Keep it {length}.",
],
"Product Listing": [
"Write a caption for this image as though it were a product listing.",
"Write a caption for this image as though it were a product listing. Keep it under {word_count} words.",
"Write a {length} caption for this image as though it were a product listing.",
],
"Social Media Post": [
"Write a caption for this image as if it were being used for a social media post.",
"Write a caption for this image as if it were being used for a social media post. Limit the caption to {word_count} words.",
"Write a {length} caption for this image as if it were being used for a social media post.",
],
}
all_captions = []
# 'any' means no length specified
length = None if caption_length == "any" else caption_length
if isinstance(length, str):
try:
length = int(length)
except ValueError:
pass
# Build prompt
if length is None:
map_idx = 0
elif isinstance(length, int):
map_idx = 1
elif isinstance(length, str):
map_idx = 2
else:
raise ValueError(f"Invalid caption length: {length}")
prompt_str = CAPTION_TYPE_MAP[caption_type][map_idx]
# Add extra options
if len(extra_options) > 0:
prompt_str += " " + " ".join(extra_options)
# Add name, length, word_count
prompt_str = prompt_str.format(name=name_input, length=caption_length, word_count=caption_length)
if custom_prompt.strip() != "":
prompt_str = custom_prompt.strip()
# For debugging
print(f"Prompt: {prompt_str}")
for i in range(0, len(input_images), batch_size):
batch = input_images[i:i + batch_size]
for input_image in batch:
try:
# Preprocess image
image = input_image.resize((384, 384), Image.LANCZOS)
pixel_values = TVF.pil_to_tensor(image).unsqueeze(0) / 255.0
pixel_values = TVF.normalize(pixel_values, [0.5], [0.5])
pixel_values = pixel_values.to(chat_device)
except ValueError as e:
print(f"Error processing image: {e}")
print("Skipping this image and continuing...")
continue
# Embed image
with torch.amp.autocast_mode.autocast(chat_device, enabled=True):
vision_outputs = model.clip_model(pixel_values=pixel_values, output_hidden_states=True)
image_features = vision_outputs.hidden_states
embedded_images = model.image_adapter(image_features).to(chat_device)
# Build the conversation
convo = [
{
"role": "system",
"content": "You are a helpful image captioner.",
},
{
"role": "user",
"content": prompt_str,
},
]
# Format the conversation
if hasattr(model.tokenizer, 'apply_chat_template'):
convo_string = model.tokenizer.apply_chat_template(convo, tokenize=False, add_generation_prompt=True)
else:
# Fallback if apply_chat_template is not available
convo_string = "<|eot_id|>\n"
for message in convo:
if message['role'] == 'system':
convo_string += f"<|system|>{message['content']}<|endoftext|>\n"
elif message['role'] == 'user':
convo_string += f"<|user|>{message['content']}<|endoftext|>\n"
else:
convo_string += f"{message['content']}<|endoftext|>\n"
convo_string += "<|eot_id|>"
assert isinstance(convo_string, str)
# Tokenize the conversation
convo_tokens = model.tokenizer.encode(convo_string, return_tensors="pt", add_special_tokens=False,
truncation=False)
prompt_tokens = model.tokenizer.encode(prompt_str, return_tensors="pt", add_special_tokens=False,
truncation=False)
assert isinstance(convo_tokens, torch.Tensor) and isinstance(prompt_tokens, torch.Tensor)
convo_tokens = convo_tokens.squeeze(0)
prompt_tokens = prompt_tokens.squeeze(0)
# Calculate where to inject the image
eot_id_indices = (convo_tokens == model.tokenizer.convert_tokens_to_ids("<|eot_id|>")).nonzero(as_tuple=True)[
0].tolist()
assert len(eot_id_indices) == 2, f"Expected 2 <|eot_id|> tokens, got {len(eot_id_indices)}"
preamble_len = eot_id_indices[1] - prompt_tokens.shape[0]
# Embed the tokens
convo_embeds = model.llm_model.model.embed_tokens(convo_tokens.unsqueeze(0).to(current_device))
# Construct the input
input_embeds = torch.cat([
convo_embeds[:, :preamble_len],
embedded_images.to(dtype=convo_embeds.dtype),
convo_embeds[:, preamble_len:],
], dim=1).to(chat_device)
input_ids = torch.cat([
convo_tokens[:preamble_len].unsqueeze(0),
torch.zeros((1, embedded_images.shape[1]), dtype=torch.long),
convo_tokens[preamble_len:].unsqueeze(0),
], dim=1).to(chat_device)
attention_mask = torch.ones_like(input_ids)
generate_ids = model.llm_model.generate(input_ids=input_ids, inputs_embeds=input_embeds,
attention_mask=attention_mask, do_sample=True,
suppress_tokens=None, max_new_tokens=max_new_tokens, top_p=top_p,
temperature=temperature)
# Trim off the prompt
generate_ids = generate_ids[:, input_ids.shape[1]:]
if generate_ids[0][-1] == model.tokenizer.eos_token_id or generate_ids[0][-1] == model.tokenizer.convert_tokens_to_ids(
"<|eot_id|>"):
generate_ids = generate_ids[:, :-1]
caption = model.tokenizer.batch_decode(generate_ids, skip_special_tokens=False,
clean_up_tokenization_spaces=False)[0]
all_captions.append(caption.strip())
return all_captions
def cleanGPU():
gc.collect()
mm.unload_all_models()
mm.soft_empty_cache()
def free_memory():
import gc
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
torch.cuda.ipc_collect()
def get_torch_device_patched():
global current_device
if (
not torch.cuda.is_available()
or comfy.model_management.cpu_state == comfy.model_management.CPUState.CPU
):
return torch.device("cpu")
return torch.device(current_device)
# 设置全局设备变量
current_device = "cuda:0"
# 覆盖ComfyUI的设备获取函数
comfy.model_management.get_torch_device = get_torch_device_patched
View File
+70
View File
@@ -0,0 +1,70 @@
[
{
"name": "replace_character_names",
"prompt": "If there is a person/character in the image you must refer to them as {name}."
},
{
"name": "exclude_unchangeable_attributes",
"prompt": "Do NOT include information about people/characters that cannot be changed (like ethnicity, gender, etc), but do still include changeable attributes (like hair style)."
},
{
"name": "include_lighting_details",
"prompt": "Include information about lighting."
},
{
"name": "include_camera_angle",
"prompt": "Include information about camera angle."
},
{
"name": "mention_watermark_presence",
"prompt": "Include information about whether there is a watermark or not."
},
{
"name": "note_jpeg_artifacts",
"prompt": "Include information about whether there are JPEG artifacts or not."
},
{
"name": "include_exif_data",
"prompt": "If it is a photo you MUST include information about what camera was likely used and details such as aperture, shutter speed, ISO, etc."
},
{
"name": "exclude_sexual_content",
"prompt": "Do NOT include anything sexual; keep it PG."
},
{
"name": "exclude_image_resolution",
"prompt": "Do NOT mention the image's resolution."
},
{
"name": "describe_aesthetic_quality",
"prompt": "You MUST include information about the subjective aesthetic quality of the image from low to very high."
},
{
"name": "include_composition_style",
"prompt": "Include information on the image's composition style, such as leading lines, rule of thirds, or symmetry."
},
{
"name": "exclude_text_elements",
"prompt": "Do NOT mention any text that is in the image."
},
{
"name": "specify_depth_of_field",
"prompt": "Specify the depth of field and whether the background is in focus or blurred."
},
{
"name": "specify_lighting_sources",
"prompt": "If applicable, mention the likely use of artificial or natural lighting sources."
},
{
"name": "avoid_ambiguous_language",
"prompt": "Do NOT use any ambiguous language."
},
{
"name": "classify_image_as_sfw_nsfw",
"prompt": "Include whether the image is sfw, suggestive, or nsfw."
},
{
"name": "describe_key_elements_only",
"prompt": "ONLY describe the most important elements of the image."
}
]
+34
View File
@@ -0,0 +1,34 @@
import folder_paths
import os
import base64
import numpy as np
from PIL import Image,ImageOps, ImageFilter
import io
comfy_path = os.path.dirname(folder_paths.__file__)
custom_nodes_path = os.path.join(comfy_path, "custom_nodes")
# D:\comfyui\ComfyUI_windows_portable\ComfyUI\custom_nodes\Comfyui_CXH_ALY
# current_folder = os.path.dirname(os.path.abspath(__file__))
# 节点路径
def node_path(node_name):
return os.path.join(custom_nodes_path,node_name)
# 创建文件夹
def mkdir(path):
folder = os.path.exists(path)
if not folder: #判断是否存在文件夹如果不存在则创建为文件夹
os.makedirs(path) #makedirs 创建文件时如果路径不存在会创建这个路径
# 获取所有图片文件路径
def get_all_image_paths(directory):
image_paths = []
for root, dirs, files in os.walk(directory):
for file in files:
if file.lower().endswith(('.png', '.jpg', '.jpeg')):
image_paths.append(os.path.join(root, file))
return image_paths
+129
View File
@@ -0,0 +1,129 @@
# -*- encoding: utf-8 -*-
'''
@File :ximg.py
@Description :图片转换工具
'''
import os
import torch
from PIL import Image, ImageOps, ImageSequence, ImageFile,UnidentifiedImageError
import numpy as np
import cv2 as cv
import io
import base64
import requests
from io import BytesIO
def tensor2pil(t_image: torch.Tensor) -> Image:
return Image.fromarray(
np.clip(255.0 * t_image.cpu().numpy().squeeze(), 0, 255).astype(np.uint8))
def pil2tensor(image:Image) -> torch.Tensor:
return torch.from_numpy(np.array(image).astype(np.float32) / 255.0).unsqueeze(0)
def tensor2cv2(image:torch.Tensor) -> np.array:
if image.dim() == 4:
image = image.squeeze()
npimage = image.numpy()
cv2image = np.uint8(npimage * 255 / npimage.max())
return cv.cvtColor(cv2image, cv.COLOR_RGB2BGR)
def cv22pil(cv2_img:np.ndarray) -> Image:
cv2_img = cv.cvtColor(cv2_img, cv.COLOR_BGR2RGB)
return Image.fromarray(cv2_img)
# pil转io
def pil2iobyte(pil_image,format='PNG'):
byte_arr = io.BytesIO()
pil_image.save(byte_arr, format=format)
byte_arr = byte_arr.getvalue()
return byte_arr
# pil转64
def pilTobase64(pil_image,format='PNG'):
byte_arr = pil2iobyte(pil_image,format)
image_base64 = base64.b64encode(byte_arr).decode('utf-8')
return image_base64
def ioBytes2tensor(bytes):
image = Image.open(bytes)
return pil2tensor(image)
def getImageSize(image):
if image.shape[0] > 0:
image = torch.unsqueeze(image[0], 0)
_image = tensor2pil(image)
return (_image.width, _image.height)
# 转成mask
def imageToMask(img):
i = img
if i.mode == 'I':
i = i.point(lambda i: i * (1 / 255))
image = i.convert("RGB")
image = np.array(image).astype(np.float32) / 255.0
image = torch.from_numpy(image)[None,]
if 'A' in i.getbands():
mask = np.array(i.getchannel('A')).astype(np.float32) / 255.0
mask = 1. - torch.from_numpy(mask)
else:
mask = torch.zeros((64,64), dtype=torch.float32, device="cpu")
return tensor2pil(mask)
# ret_masks.append(image2mask(_mask))
def image2mask(image:Image) -> torch.Tensor:
_image = image.convert('RGBA')
alpha = _image.split() [0]
bg = Image.new("L", _image.size)
_image = Image.merge('RGBA', (bg, bg, bg, alpha))
ret_mask = torch.tensor([pil2tensor(_image)[0, :, :, 3].tolist()])
return ret_mask
# 图像回帖
def croppImg(original_image,cropped_avatar,left_x,top_y):
# 获取原始图像的大小
original_width, original_height = original_image.size
return croppImageBySize(cropped_avatar,left_x,top_y,original_width,original_height)
def croppImageBySize(cropped_avatar,left_x,top_y,original_w,original_h):
# 获取原始图像的大小
original_width, original_height = original_w,original_h
# 获取头像的大小
avatar_width, avatar_height = cropped_avatar.size
# 创建一个与原始图像相同大小的透明图像
extended_image = Image.new("RGBA", (original_width, original_height), (0, 0, 0, 0))
# 将裁剪后的头像粘贴到新图像
extended_image.paste(cropped_avatar, (left_x, top_y), cropped_avatar)
return extended_image
# 将图片转换为Base64编码
def image_to_base64(image_path):
with open(image_path, 'rb') as image_file:
return base64.b64encode(image_file.read()).decode('utf-8')
# 获取网络图片
def img_from_url(url):
# 发送HTTP请求获取图片
response = requests.get(url)
response.raise_for_status() # 如果请求失败,这会抛出异常
# 将响应内容作为BytesIO对象打开,以便PIL可以读取它
image = Image.open(BytesIO(response.content))
return image
def open_image(path):
prev_value = None
try:
img = Image.open(path)
except (UnidentifiedImageError, ValueError): #PIL issues #4472 and #2445
prev_value = ImageFile.LOAD_TRUNCATED_IMAGES
ImageFile.LOAD_TRUNCATED_IMAGES = True
img = Image.open(path)
finally:
if prev_value is not None:
ImageFile.LOAD_TRUNCATED_IMAGES = prev_value
return img
+23
View File
@@ -0,0 +1,23 @@
import os
import folder_paths
import json
from transformers import AutoProcessor
# 下载hg 模型到本地
def download_hg_model(model_id:str,exDir:str=''):
# 下载本地
model_checkpoint = os.path.join(folder_paths.models_dir, exDir, os.path.basename(model_id))
print(model_checkpoint)
if not os.path.exists(model_checkpoint):
from huggingface_hub import snapshot_download
snapshot_download(repo_id=model_id, local_dir=model_checkpoint, local_dir_use_symlinks=False)
return model_checkpoint
# clip_model = AutoModelForCausalLM.from_pretrained(
# CLIP_PATH,
# device_map="cuda",
# trust_remote_code=True,
# torch_dtype="auto"
# )
# clip_processor = AutoProcessor.from_pretrained(CLIP_PATH, trust_remote_code=True)
+15
View File
@@ -0,0 +1,15 @@
[project]
name = "comfyui_auto_caption"
description = "Load images in order(All other nodes are in the wrong order)! Using LLM and Joy tag pipeline to tag your image(s folder), it's suitable for train FLUX LoRA and also sdxl."
version = "1.0.2"
license = {file = "LICENSE"}
dependencies = ["huggingface_hub==0.24.3", "accelerate", "#torch", "transformers>=4.43.3", "sentencepiece", "bitsandbytes>=0.43.3", "bitsandbytes-windows>=0.37.5"]
[project.urls]
Repository = "https://github.com/Cyber-BlackCat/ComfyUI_Auto_Caption"
# Used by Comfy Registry https://comfyregistry.org
[tool.comfy]
PublisherId = "cyber-bcat"
DisplayName = "Cyber-BlackCat"
Icon = ""