joycaption.py

import spaces
import gradio as gr
from huggingface_hub import InferenceClient
from torch import nn
from transformers import AutoModel, AutoProcessor, AutoTokenizer, PreTrainedTokenizer, PreTrainedTokenizerFast, AutoModelForCausalLM
from pathlib import Path
import torch
import torch.amp.autocast_mode
from PIL import Image
import os
import gc

device = "cuda" if torch.cuda.is_available() else "cpu"

llm_models = {
    "Sao10K/Llama-3.1-8B-Stheno-v3.4": None,
    "unsloth/Meta-Llama-3.1-8B-bnb-4bit": None,
    "mergekit-community/L3.1-Boshima-b-FIX": None,
    "meta-llama/Meta-Llama-3.1-8B": None,
}

CLIP_PATH = "google/siglip-so400m-patch14-384"
VLM_PROMPT = "A descriptive caption for this image:\n"
MODEL_PATH = list(llm_models.keys())[0]
CHECKPOINT_PATH = Path("wpkklhc6")
TITLE = "<h1><center>JoyCaption Pre-Alpha (2024-07-30a)</center></h1>"

HF_TOKEN = os.environ.get("HF_TOKEN", None)
use_inference_client = False

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

# https://huggingface.co/docs/transformers/v4.44.2/gguf
# https://github.com/city96/ComfyUI-GGUF/issues/7
# https://github.com/THUDM/ChatGLM-6B/issues/18
# https://github.com/meta-llama/llama/issues/394
# https://huggingface.co/meta-llama/Meta-Llama-3-8B-Instruct/discussions/109
# https://huggingface.co/docs/transformers/main/en/main_classes/quantization#offload-between-cpu-and-gpu
# https://huggingface.co/google/flan-ul2/discussions/8
# https://huggingface.co/blog/4bit-transformers-bitsandbytes
tokenizer = None
text_model_client = None
text_model = None
image_adapter = None
def load_text_model(model_name: str=MODEL_PATH, gguf_file: str | None=None, is_nf4: bool=True):
    global tokenizer
    global text_model
    global image_adapter
    global text_model_client #
    global use_inference_client #
    try:
        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 tokenizer")
        if gguf_file: tokenizer = AutoTokenizer.from_pretrained(model_name, gguf_file=gguf_file, use_fast=True, legacy=False)
        else: tokenizer = AutoTokenizer.from_pretrained(model_name, use_fast=False, legacy=False)
        assert isinstance(tokenizer, PreTrainedTokenizer) or isinstance(tokenizer, PreTrainedTokenizerFast), f"Tokenizer is of type {type(tokenizer)}"
        print(f"Loading LLM: {model_name}")
        if gguf_file:
            if device == "cpu": text_model = AutoModelForCausalLM.from_pretrained(model_name, gguf_file=gguf_file, device_map=device, torch_dtype=torch.bfloat16).eval()
            elif is_nf4: text_model = AutoModelForCausalLM.from_pretrained(model_name, quantization_config=nf4_config, device_map=device, torch_dtype=torch.bfloat16).eval()
            else: text_model = AutoModelForCausalLM.from_pretrained(model_name, device_map=device, torch_dtype=torch.bfloat16).eval()
        else:
            if device == "cpu": text_model = AutoModelForCausalLM.from_pretrained(model_name, gguf_file=gguf_file, device_map=device, torch_dtype=torch.bfloat16).eval()
            elif is_nf4: text_model = AutoModelForCausalLM.from_pretrained(model_name, quantization_config=nf4_config, device_map=device, torch_dtype=torch.bfloat16).eval()
            else: text_model = AutoModelForCausalLM.from_pretrained(model_name, device_map=device, torch_dtype=torch.bfloat16).eval()
        print("Loading image adapter")
        image_adapter = ImageAdapter(clip_model.config.hidden_size, text_model.config.hidden_size).eval().to("cpu")
        image_adapter.load_state_dict(torch.load(CHECKPOINT_PATH / "image_adapter.pt", map_location="cpu", weights_only=True))
        image_adapter.eval().to(device)
    except Exception as e:
        print(f"LLM load error: {e}")
        raise Exception(f"LLM load error: {e}") from e
    finally:
        torch.cuda.empty_cache()
        gc.collect()

load_text_model.zerogpu = True

# Load CLIP
print("Loading CLIP")
clip_processor = AutoProcessor.from_pretrained(CLIP_PATH)
clip_model = AutoModel.from_pretrained(CLIP_PATH).vision_model.eval().requires_grad_(False).to(device)

# Tokenizer
# LLM
# Image Adapter
load_text_model()

@spaces.GPU()
@torch.no_grad()
def stream_chat(input_image: Image.Image):
    torch.cuda.empty_cache()

    # Preprocess image
    image = clip_processor(images=input_image, return_tensors='pt').pixel_values
    image = image.to(device)

    # Tokenize the prompt
    prompt = tokenizer.encode(VLM_PROMPT, return_tensors='pt', padding=False, truncation=False, add_special_tokens=False)

    # Embed image
    with torch.amp.autocast_mode.autocast(device, enabled=True):
        vision_outputs = clip_model(pixel_values=image, output_hidden_states=True)
        image_features = vision_outputs.hidden_states[-2]
        embedded_images = image_adapter(image_features)
        embedded_images = embedded_images.to(device)
    
    # Embed prompt
    prompt_embeds = text_model.model.embed_tokens(prompt.to(device))
    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(device)
    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=300, do_sample=False, suppress_tokens=None)
    generate_ids = text_model.generate(input_ids, inputs_embeds=inputs_embeds, attention_mask=attention_mask, max_new_tokens=300, do_sample=True, top_k=10, temperature=0.5, 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]

    return caption.strip()


@spaces.GPU()
@torch.no_grad()
def stream_chat_mod(input_image: Image.Image, max_new_tokens: int=300, top_k: int=10, temperature: float=0.5, progress=gr.Progress(track_tqdm=True)):
    global use_inference_client
    global text_model
    torch.cuda.empty_cache()
    gc.collect()

    # Preprocess image
    image = clip_processor(images=input_image, return_tensors='pt').pixel_values
    image = image.to(device)

    # Tokenize the prompt
    prompt = tokenizer.encode(VLM_PROMPT, return_tensors='pt', padding=False, truncation=False, add_special_tokens=False)

    # Embed image
    with torch.amp.autocast_mode.autocast(device, enabled=True):
        vision_outputs = clip_model(pixel_values=image, output_hidden_states=True)
        image_features = vision_outputs.hidden_states[-2]
        embedded_images = image_adapter(image_features)
        embedded_images = embedded_images.to(device)
    
    # Embed prompt
    prompt_embeds = text_model.model.embed_tokens(prompt.to(device))
    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(device)
    attention_mask = torch.ones_like(input_ids)

    # https://huggingface.co/docs/transformers/v4.44.2/main_classes/text_generation#transformers.FlaxGenerationMixin.generate
    # https://github.com/huggingface/transformers/issues/6535
    # https://zenn.dev/hijikix/articles/8c445f4373fdcc ja
    # https://github.com/ggerganov/llama.cpp/discussions/7712
    # https://huggingface.co/docs/huggingface_hub/guides/inference#openai-compatibility
    # https://huggingface.co/docs/huggingface_hub/v0.24.6/en/package_reference/inference_client#huggingface_hub.InferenceClient.text_generation
    #generate_ids = text_model.generate(input_ids, inputs_embeds=inputs_embeds, attention_mask=attention_mask, max_new_tokens=300, do_sample=False, suppress_tokens=None)
    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=top_k, temperature=temperature, suppress_tokens=None)
    
    print(prompt)

    # 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]

    return caption.strip()


def is_repo_name(s):
    import re
    return re.fullmatch(r'^[^/,\s\"\']+/[^/,\s\"\']+$', s)


def is_repo_exists(repo_id):
    from huggingface_hub import HfApi
    try:
        api = HfApi(token=HF_TOKEN)
        if api.repo_exists(repo_id=repo_id): return True
        else: return False
    except Exception as e:
        print(f"Error: Failed to connect {repo_id}.")
        print(e)
        return True # for safe


def get_text_model():
    return list(llm_models.keys())


def is_gguf_repo(repo_id: str):
    from huggingface_hub import HfApi
    try:
        api = HfApi(token=HF_TOKEN)
        if not is_repo_name(repo_id) or not is_repo_exists(repo_id): return False
        files = api.list_repo_files(repo_id=repo_id)
    except Exception as e:
        print(f"Error: Failed to get {repo_id}'s info.")
        print(e)
        gr.Warning(f"Error: Failed to get {repo_id}'s info.")
        return False
    files = [f for f in files if f.endswith(".gguf")]
    if len(files) == 0: return False
    else: return True


def get_repo_gguf(repo_id: str):
    from huggingface_hub import HfApi
    try:
        api = HfApi(token=HF_TOKEN)
        if not is_repo_name(repo_id) or not is_repo_exists(repo_id): return gr.update(value="", choices=[])
        files = api.list_repo_files(repo_id=repo_id)
    except Exception as e:
        print(f"Error: Failed to get {repo_id}'s info.")
        print(e)
        gr.Warning(f"Error: Failed to get {repo_id}'s info.")
        return gr.update(value="", choices=[])
    files = [f for f in files if f.endswith(".gguf")]
    if len(files) == 0: return gr.update(value="", choices=[])
    else: return gr.update(value=files[0], choices=files)


@spaces.GPU()
def change_text_model(model_name: str=MODEL_PATH, use_client: bool=False, gguf_file: str | None=None,
                      is_nf4: bool=True, progress=gr.Progress(track_tqdm=True)):
    global use_inference_client
    global llm_models
    use_inference_client = use_client
    try:
        if not is_repo_name(model_name) or not is_repo_exists(model_name):
            raise gr.Error(f"Repo doesn't exist: {model_name}")
        if not gguf_file and is_gguf_repo(model_name):
            gr.Info(f"Please select a gguf file.")
            return gr.update(visible=True)
        if use_inference_client:
            pass #
        else:
            load_text_model(model_name, gguf_file, is_nf4)
        if model_name not in llm_models: llm_models[model_name] = gguf_file if gguf_file else None
        return gr.update(choices=get_text_model())
    except Exception as e:
        raise gr.Error(f"Model load error: {model_name}, {e}")


# original UI
with gr.Blocks() as demo:
    gr.HTML(TITLE)
    with gr.Row():
        with gr.Column():
            input_image = gr.Image(type="pil", label="Input Image")
            run_button = gr.Button("Caption")
        
        with gr.Column():
            output_caption = gr.Textbox(label="Caption")
    
    run_button.click(fn=stream_chat, inputs=[input_image], outputs=[output_caption])


if __name__ == "__main__":
    demo.launch()