import glob import os import random import re import subprocess import sys import spaces import torch import gradio as gr from huggingface_hub import HfApi, ModelCard, hf_hub_download, login from safetensors.torch import load_file TOKEN = os.environ.get("HF_TOKEN") if TOKEN: login(token=TOKEN) api = HfApi(token=TOKEN) from diffusers import Krea2Pipeline DTYPE = torch.bfloat16 BASE_MODEL_REPO = "krea/Krea-2-Turbo" # Local bucket path, or an HF model safetensors reference: # /data/foo.safetensors # krea/Krea-2-Turbo:turbo.safetensors # krea/Krea-2-Turbo (auto-picks the root .safetensors weight file) BASE_CHECKPOINT = os.environ.get( "BASE_CHECKPOINT", "/data/krea2TurboNSFWAIO_v10.safetensors" ) MAX_SEED = 2**31 - 1 _REPO_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*/[A-Za-z0-9][A-Za-z0-9._-]*$") _ATTN_MAP = {"wq": "to_q", "wk": "to_k", "wv": "to_v", "wo": "to_out.0", "gate": "to_gate"} _FF_MAP = {"gate": "ff.gate", "up": "ff.up", "down": "ff.down"} # AoTI: the Krea2TransformerBlock is served from a prebuilt .pt2 (kernels only, no # weights) with a single rank-64 LoRA hotswap slot. Any adapter with rank <= # TARGET_RANK targeting the same layers swaps into it with no recompilation # (diffusers #9453). Artifact is public; the Space runs eager if it's missing. TARGET_RANK = 64 BLOCK_NAME = "Krea2TransformerBlock" ARTIFACT_REPO = "multimodalart/krea2-aoti-kernels" ARTIFACT_FILE = f"Krea2TransformerBlock-lora-r{TARGET_RANK}/package.pt2" # Built-in LoRAs to feature in the gallery, as (repo, display title). Each # adapter's weight file and trigger word are resolved from its repo at startup, # so adding a LoRA is just one line here. LORA_REPOS = [ ("krea/Krea-2-LoRA-retroanime", "Retro Anime"), ("krea/Krea-2-LoRA-rainywindow", "Rainy Window"), ("krea/Krea-2-LoRA-vintagetarot", "Vintage Tarot"), ("krea/Krea-2-LoRA-sunsetblur", "Sunset Blur"), ("krea/Krea-2-LoRA-dotmatrix", "Dot Matrix"), ("krea/Krea-2-LoRA-neondrip", "Neon Drip"), ("krea/Krea-2-LoRA-darkbrush", "Dark Brush"), ("krea/Krea-2-LoRA-kidsdrawing", "Kids Drawing"), ("krea/Krea-2-LoRA-softwatercolor", "Soft Watercolor Art Deco"), ] DEFAULT_SCALE = 1.0 def _read_trigger(repo): try: text = ModelCard.load(repo, token=TOKEN).text m = re.search(r"[Tt]rigger word[:\*\s]*`([^`]+)`", text) if m: return m.group(1).strip() except Exception: pass return "" def _strip_community_prefix(key): for prefix in ("base_model.model.", "model.diffusion_model.", "diffusion_model.", "model."): if key.startswith(prefix): key = key[len(prefix) :] return key def _community_attn_suffix(block_prefix, kind, sub, tail): if kind == "attn": if sub == "qknorm": return None if sub in _ATTN_MAP: return f"{block_prefix}.attn.{_ATTN_MAP[sub]}.{tail}" if kind == "mlp" and sub in _FF_MAP: return f"{block_prefix}.{_FF_MAP[sub]}.{tail}" return None def _convert_community_key(key, tensor): key = _strip_community_prefix(key) if key.startswith( ("img_in.", "transformer_blocks.", "time_embed.", "time_mod_proj.", "text_fusion.", "txt_in.", "final_layer.") ): return key, tensor parts = key.split(".") tail = parts[-1] if key == "first.weight" or key == "first.bias": return f"img_in.{tail}", tensor m = re.fullmatch(r"blocks\.(\d+)\.mod\.lin", key) if m: hidden = tensor.numel() // 6 return f"transformer_blocks.{m.group(1)}.scale_shift_table", tensor.reshape(6, hidden) m = re.fullmatch(r"blocks\.(\d+)\.(prenorm|postnorm)\.scale", key) if m: norm = "norm1" if m.group(2) == "prenorm" else "norm2" return f"transformer_blocks.{m.group(1)}.{norm}.weight", tensor m = re.fullmatch(r"blocks\.(\d+)\.attn\.qknorm\.(qnorm|knorm)\.scale", key) if m: norm = "norm_q" if m.group(2) == "qnorm" else "norm_k" return f"transformer_blocks.{m.group(1)}.attn.{norm}.weight", tensor m = re.fullmatch(r"blocks\.(\d+)\.(attn|mlp)\.(\w+)\.(.+)", key) if m: converted = _community_attn_suffix(f"transformer_blocks.{m.group(1)}", m.group(2), m.group(3), m.group(4)) if converted: return converted, tensor if key.startswith("tmlp.0."): return "time_embed.linear_1." + key.split(".", 2)[2], tensor if key.startswith("tmlp.2."): return "time_embed.linear_2." + key.split(".", 2)[2], tensor if key.startswith("tproj.1."): return "time_mod_proj." + key.split(".", 2)[2], tensor m = re.fullmatch(r"txtfusion\.(layerwise_blocks|refiner_blocks)\.(\d+)\.(prenorm|postnorm)\.scale", key) if m: norm = "norm1" if m.group(3) == "prenorm" else "norm2" return f"text_fusion.{m.group(1)}.{m.group(2)}.{norm}.weight", tensor m = re.fullmatch(r"txtfusion\.(layerwise_blocks|refiner_blocks)\.(\d+)\.attn\.qknorm\.(qnorm|knorm)\.scale", key) if m: norm = "norm_q" if m.group(3) == "qnorm" else "norm_k" return f"text_fusion.{m.group(1)}.{m.group(2)}.attn.{norm}.weight", tensor m = re.fullmatch(r"txtfusion\.(layerwise_blocks|refiner_blocks)\.(\d+)\.(attn|mlp)\.(\w+)\.(.+)", key) if m: converted = _community_attn_suffix( f"text_fusion.{m.group(1)}.{m.group(2)}", m.group(3), m.group(4), m.group(5) ) if converted: return converted, tensor if key.startswith("txtfusion.projector."): return "text_fusion.projector." + key.split(".", 2)[2], tensor if key == "txtmlp.0.scale": return "txt_in.norm.weight", tensor if key.startswith("txtmlp.1."): return "txt_in.linear_1." + key.split(".", 2)[2], tensor if key.startswith("txtmlp.3."): return "txt_in.linear_2." + key.split(".", 2)[2], tensor if key == "last.norm.scale": return "final_layer.norm.weight", tensor if key.startswith("last.linear."): return "final_layer.linear." + key.split(".", 2)[2], tensor if key == "last.modulation.lin": return "final_layer.scale_shift_table", tensor return None, None def _resolve_checkpoint_source(spec): """Return a local path to a .safetensors file (download from HF if needed).""" spec = (spec or "").strip() if not spec: return None if os.path.isfile(spec): return spec repo, _, filename = spec.partition(":") if _REPO_RE.match(repo): if not filename: files = api.list_repo_files(repo) root_safes = [f for f in files if f.endswith(".safetensors") and "/" not in f] if not root_safes: raise FileNotFoundError(f"No root .safetensors found in {repo}") filename = max(root_safes, key=lambda f: "turbo" in f.lower() or "raw" in f.lower()) print(f"[base] fetching {repo}/{filename}") return hf_hub_download(repo, filename, token=TOKEN) raise FileNotFoundError(f"Checkpoint not found: {spec}") def _load_transformer_checkpoint(checkpoint_path): """Load a Krea 2 Turbo .safetensors checkpoint into diffusers transformer keys.""" raw = load_file(checkpoint_path, device="cpu") # Drop FP8 scale sidecars (ComfyUI) and keep only the actual weight tensors. raw = {k: v for k, v in raw.items() if not k.endswith(".weight_scale")} stripped_keys = [_strip_community_prefix(k) for k in raw] diffusers_native = sum( 1 for k in stripped_keys if k.startswith(("img_in.", "transformer_blocks.", "time_embed.", "text_fusion.", "txt_in.", "final_layer.")) ) community_keys = sum( 1 for k in stripped_keys if k.startswith(("blocks.", "first.", "txtfusion.", "tmlp.", "tproj.", "last.", "txtmlp.")) ) if diffusers_native > community_keys: print(f"[base] checkpoint keys look diffusers-native; loading {checkpoint_path} directly") return raw converted = {} skipped = [] for key, tensor in raw.items(): new_key, new_tensor = _convert_community_key(key, tensor) if new_key is None: skipped.append(key) continue converted[new_key] = new_tensor if skipped: print(f"[base] skipped {len(skipped)} non-transformer keys (e.g. {skipped[:3]})") print(f"[base] converted {len(converted)} community keys from {checkpoint_path}") return converted def _load_pipeline(): pipe = Krea2Pipeline.from_pretrained(BASE_MODEL_REPO, torch_dtype=DTYPE) try: checkpoint_path = _resolve_checkpoint_source(BASE_CHECKPOINT) except FileNotFoundError as e: print(f"[base] {e}; using hub transformer") return pipe if checkpoint_path: print(f"[base] swapping transformer weights from {checkpoint_path}") state_dict = _load_transformer_checkpoint(checkpoint_path) missing, unexpected = pipe.transformer.load_state_dict(state_dict, strict=False) if missing: print(f"[base] missing {len(missing)} keys (first: {missing[:5]})") if unexpected: print(f"[base] unexpected {len(unexpected)} keys (first: {unexpected[:5]})") return pipe def resolve_custom_lora(repo): """Find the LoRA weight file and trigger word for an arbitrary Krea-2 LoRA repo.""" files = api.list_repo_files(repo) safes = [f for f in files if f.endswith(".safetensors")] if not safes: raise gr.Error(f"No .safetensors weights found in {repo}.") # Prefer the final weights at the repo root over training checkpoints in # subfolders (e.g. checkpoint-1000/...), then prefer a lora-named file. weight_name = max(safes, key=lambda f: ("/" not in f, "lora" in f.lower())) return weight_name, _read_trigger(repo) pipe = _load_pipeline() # Resolve built-in LoRA metadata (weight file + trigger) without loading each as # its own adapter: AoTI hotswap uses a single padded "style" slot, so styles are # swapped into that one slot on demand. A missing/renamed repo just drops its # tile instead of crashing the Space. LORAS = [] for repo, title in LORA_REPOS: key = repo.split("LoRA-")[-1] try: weight_name, trigger = resolve_custom_lora(repo) except Exception as e: print(f"[lora] resolve failed for {repo}: {e}") continue LORAS.append( {"key": key, "title": title, "repo": repo, "weight_name": weight_name, "trigger": trigger, "scale": DEFAULT_SCALE} ) print(f"[lora] {len(LORAS)}/{len(LORA_REPOS)} styles ready: {[l['key'] for l in LORAS]}") # enable_lora_hotswap MUST run before the first adapter loads — it arms PEFT # prepare_model_for_compiled_hotswap (pads adapters to target_rank, makes the # structure compile-safe). diffusers raises if called after the first load. pipe.transformer.enable_lora_hotswap(target_rank=TARGET_RANK) FIRST = LORAS[0] pipe.transformer.load_lora_adapter( FIRST["repo"], weight_name=FIRST["weight_name"], adapter_name="style", token=TOKEN ) # Bake scaling=1.0 into the exported graph; the user scale is folded into the # re-supplied lora_B constants per request, so the slider works without recompile. pipe.transformer.set_adapters("style", weights=1.0) pipe.to("cuda") # Which LoRA repo currently occupies the "style" slot (+ the scale last folded in), # so we only hotswap / re-patch when the request actually changes them. CURRENT = {"repo": FIRST["repo"], "weight_name": FIRST["weight_name"], "scale": None} AOTI_MODEL = None # set at startup if the artifact exists; else eager fallback def _full_weights(block, scale): """Full constant set for a block, with the user LoRA scale folded into lora_B.""" w = {} for n, p in block.named_parameters(remove_duplicate=False): if scale != 1.0 and ".lora_B." in n: w[n] = p * scale else: w[n] = p for n, b in block.named_buffers(remove_duplicate=False): w[n] = b return w def _patch_all(aoti_model, scale): n = 0 for block in pipe.transformer.modules(): if block.__class__.__name__ == BLOCK_NAME: block.forward = aoti_model.with_weights(_full_weights(block, scale)) n += 1 return n def _load_artifact(): """Load the prebuilt LoRA-aware .pt2 (kernels only, no JIT compile at runtime).""" global AOTI_MODEL from spaces.zero.torch.aoti import LazyAOTIModel pt2 = hf_hub_download( repo_id=ARTIFACT_REPO, filename=ARTIFACT_FILE, repo_type="dataset", token=TOKEN, ) AOTI_MODEL = LazyAOTIModel(pt2) print(f"AoTI artifact loaded: {ARTIFACT_FILE}") try: _load_artifact() except Exception as e: print(f"AoTI artifact unavailable ({e}); running eager.") def _ensure_adapter(repo, weight_name): """Hotswap the 'style' slot to a new LoRA only when it differs from the current.""" if CURRENT["repo"] == repo and CURRENT["weight_name"] == weight_name: return False pipe.transformer.load_lora_adapter( repo, weight_name=weight_name, adapter_name="style", hotswap=True, token=TOKEN, ) pipe.transformer.set_adapters("style", weights=1.0) CURRENT["repo"] = repo CURRENT["weight_name"] = weight_name CURRENT["scale"] = None return True gallery_items = [] for lora in LORAS: try: img = hf_hub_download(lora["repo"], "images/05_turbo.png") except Exception: img = None gallery_items.append((img, lora["title"])) def _mash_prompt(prompt, trigger): prompt = (prompt or "").strip() if trigger and trigger.lower() not in prompt.lower(): return f"{prompt}, {trigger}" if prompt else trigger return prompt CHIP = ( "background:#171717;border:1px solid #262626;border-radius:5px;padding:2px 8px;" "font-family:'JetBrains Mono',ui-monospace,monospace;font-size:12px;color:#d4d4d5;" ) def _info_html(title, trigger, scale=None, thumb=None): parts = ['
Explore the built-in styles, or load any Krea 2 LoRA from Hugging Face by path, then generate on Krea 2 Turbo.