Spaces:
Running on Zero
Running on Zero
| # @file app.py | |
| # @description AI Creative Studio & ZeroGPU LLM Hub β High-Density Pro Dashboard | |
| # Showcases both HF PRO subscription resources: | |
| # β‘ ZeroGPU (48GB Blackwell RTX PRO 6000) for local 27B LLMs & Tool Calling | |
| # π Serverless HF Inference API for FLUX.1, Whisper, Kokoro TTS, BGE-M3 & Qwen-VL. | |
| # | |
| # @changes | |
| # - [2026-07-08] [Composer] - Initial Gemma Heretic bucket-backed chat app | |
| # - [2026-08-12] [Jcode] - AI Creative Studio: multi-tab app combining ZeroGPU + Inference API | |
| # - [2026-08-15] [Jcode] - Add Qwen3.8-27B, 128k context, FlashAttention, OpenAI tool calling | |
| # - [2026-08-16] [Jcode] - High-density full-screen professional dashboard with live telemetry HUD | |
| import os | |
| import time | |
| import hmac | |
| import json | |
| import re | |
| import uuid | |
| import gc | |
| import math | |
| from fastapi import FastAPI, Request, HTTPException | |
| from fastapi.responses import JSONResponse, StreamingResponse | |
| import uvicorn | |
| # Preload CUDA 12 runtime libs via ctypes | |
| try: | |
| import glob as _glob | |
| import ctypes as _ctypes | |
| import site | |
| _libs = [] | |
| for sp in site.getsitepackages(): | |
| _libs += _glob.glob(os.path.join(sp, "nvidia", "*", "lib", "*.so*")) | |
| for _p in _libs: | |
| try: | |
| _ctypes.CDLL(_p) | |
| except Exception: | |
| pass | |
| except Exception: | |
| pass | |
| import gradio as gr | |
| import spaces | |
| from huggingface_hub import InferenceClient, hf_hub_download | |
| from llama_cpp import Llama | |
| # βββ Config βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| SEARCH_DIRS = ["/tmp", "/data", "/models", "."] | |
| SUPPORTED_MODELS = ( | |
| "Qwen3.8-9B-Q8_0.gguf", | |
| "Qwen3.8-9B-Q4_K_M.gguf", | |
| "Qwen3.8-27B-Q4_K_M.gguf", | |
| "Qwen3.8-27B-Q6_K.gguf", | |
| "Qwen3.8-27B-Uncensored.i1-Q4_K_M.gguf", | |
| "gemma-4-26B-A4B-it-ultra-uncensored-heretic.i1-Q4_K_M.gguf", | |
| "gemma-4-26B-A4B-it-ultra-uncensored-heretic.i1-Q6_K.gguf", | |
| ) | |
| DEFAULT_MODEL = SUPPORTED_MODELS[0] | |
| MODEL_HUB_SOURCES = { | |
| "Qwen3.8-9B-Q8_0.gguf": { | |
| "repo_id": "empero-ai/Qwen3.8-9B-GGUF", | |
| "filename": "Qwen3.8-9B-Q8_0.gguf", | |
| }, | |
| "Qwen3.8-9B-Q4_K_M.gguf": { | |
| "repo_id": "empero-ai/Qwen3.8-9B-GGUF", | |
| "filename": "Qwen3.8-9B-Q4_K_M.gguf", | |
| }, | |
| "Qwen3.8-27B-Q4_K_M.gguf": { | |
| "repo_id": "unsloth/Qwen3.8-27B-GGUF", | |
| "filename": "Qwen3.8-27B-Q4_K_M.gguf", | |
| }, | |
| "Qwen3.8-27B-Q6_K.gguf": { | |
| "repo_id": "unsloth/Qwen3.8-27B-GGUF", | |
| "filename": "Qwen3.8-27B-Q6_K.gguf", | |
| }, | |
| "Qwen3.8-27B-Uncensored.i1-Q4_K_M.gguf": { | |
| "repo_id": "mradermacher/Qwen3.8-27B-Uncensored-i1-GGUF", | |
| "filename": "Qwen3.8-27B-Uncensored.i1-Q4_K_M.gguf", | |
| }, | |
| } | |
| import sys | |
| import shutil | |
| import random | |
| BASE_DIR = os.path.dirname(__file__) | |
| MOLDOVAN_DIR = os.path.join(BASE_DIR, "moldovan-qwen") | |
| if MOLDOVAN_DIR not in sys.path: | |
| sys.path.insert(0, MOLDOVAN_DIR) | |
| if BASE_DIR not in sys.path: | |
| sys.path.insert(0, BASE_DIR) | |
| # HF Serverless Inference API models | |
| FLUX_MODEL = "black-forest-labs/FLUX.1-schnell" | |
| WHISPER_MODEL = "openai/whisper-large-v3" | |
| KOKORO_TTS_MODEL = "hexgrad/Kokoro-82M" | |
| EMBEDDING_MODEL = "BAAI/bge-m3" | |
| VISION_MODEL = "Qwen/Qwen2.5-VL-72B-Instruct" | |
| # βββ Free Open-Weights ZeroGPU Model Registries ($0 Cost / 40 min A100 Quota) ββ | |
| VIDEO_MODELS = { | |
| "Wan 2.1 (Alibaba T2V 1.3B)": "Wan-AI/Wan2.1-T2V-1.3B", | |
| "CogVideoX-2B (THUDM SOTA)": "THUDM/CogVideoX-2b", | |
| "LTX-Video (Lightricks Fast)": "Lightricks/LTX-Video", | |
| "ZeroScope v2 (576w High-Res)": "cerspense/zeroscope_v2_576w", | |
| "ModelScope Text-to-Video": "damo-vilab/modelscope-damo-text-to-video-synthesis", | |
| "AnimateDiff (Motion Adapter)": "guoyww/animatediff-motion-adapter", | |
| "I2VGen-XL (Image to Video)": "ali-vilab/i2vgen-xl", | |
| } | |
| AUDIO_MUSIC_MODELS = { | |
| # Full-song vocal model. Official card: diffusers ModularPipeline, CUDA, 24GB+ or CPU offload. | |
| "MiniMax Music 3 (full song + vocals)": "MiniMaxAI/MiniMax-Music3", | |
| # Full-song instrumental / audio model. Access may require accepting HF terms. | |
| "Stable Audio 3 Medium (full structured music)": "stabilityai/stable-audio-3-medium", | |
| "MusicGen Small (instrumental)": "facebook/musicgen-small", | |
| "MusicGen Medium (instrumental)": "facebook/musicgen-medium", | |
| "MusicGen Melody (instrumental)": "facebook/musicgen-melody", | |
| "AudioGen Medium (sound effects)": "facebook/audiogen-medium", | |
| "Stable Audio Open 1.0 (audio / loops)": "stabilityai/stable-audio-open-1.0", | |
| "Bark (speech / singing experiments)": "suno/bark", | |
| "Kokoro-82M (TTS, not music)": "hexgrad/Kokoro-82M", | |
| "Parler-TTS Mini (TTS, not music)": "parler-tts/parler-tts-mini-v1", | |
| } | |
| IMAGE_MODELS = { | |
| "FLUX.1-schnell (Black Forest Labs)": "black-forest-labs/FLUX.1-schnell", | |
| "SDXL Turbo (Real-time 1-Step)": "stabilityai/sdxl-turbo", | |
| "Stable Diffusion 3.5 Medium": "stabilityai/stable-diffusion-3.5-medium", | |
| "DreamShaper 8 (Lykon Photoreal)": "Lykon/dreamshaper-8", | |
| "Playground v2.5 (Aesthetic 1024)": "playgroundai/playground-v2.5-1024px-aesthetic", | |
| } | |
| # Pipeline caches for ZeroGPU execution | |
| _video_pipeline_cache = {} | |
| _audio_pipeline_cache = {} | |
| # Keep model downloads in the persistent Space cache. Loading remains lazy: a | |
| # model is downloaded only when the user selects it, never during app startup. | |
| _HF_AUDIO_CACHE = os.environ.get("HF_HOME", "/data") | |
| os.environ.setdefault("HF_HOME", _HF_AUDIO_CACHE) | |
| os.environ.setdefault("HF_HUB_CACHE", os.path.join(_HF_AUDIO_CACHE, "hub")) | |
| _image_pipeline_cache = {} | |
| # βββ Startup Background Model Pre-caching on CPU βββββββββββββββββββββββββββββ | |
| def _preload_heavy_models_on_cpu(): | |
| """Download weights to /data on CPU so ZeroGPU lease is not eaten by network download.""" | |
| try: | |
| from huggingface_hub import snapshot_download | |
| cache_dir = os.environ.get("HF_HUB_CACHE", "/data/hub") | |
| print("[Startup Pre-Cache] Pre-caching MiniMax Music 3 on CPU to /data...", flush=True) | |
| snapshot_download( | |
| repo_id="MiniMaxAI/MiniMax-Music3", | |
| cache_dir=cache_dir, | |
| resume_download=True, | |
| ) | |
| print("[Startup Pre-Cache] MiniMax Music 3 weights cached successfully!", flush=True) | |
| except Exception as e: | |
| print(f"[Startup Pre-Cache] Note: Background preload skipped or failed: {e}", flush=True) | |
| import threading | |
| threading.Thread(target=_preload_heavy_models_on_cpu, daemon=True).start() | |
| # HF token from Space secrets | |
| HF_TOKEN = os.environ.get("HF_TOKEN", None) | |
| # Inference API client | |
| api_client = InferenceClient(token=HF_TOKEN) | |
| # ZeroGPU model state | |
| _llm = None | |
| _loaded_file = None | |
| def find_model_path(model_file: str) -> str: | |
| """Find absolute path of a GGUF model file across search directories.""" | |
| for d in SEARCH_DIRS: | |
| p = os.path.join(d, model_file) | |
| if os.path.isfile(p): | |
| return p | |
| source = MODEL_HUB_SOURCES.get(model_file) | |
| if source and os.path.isdir("/data"): | |
| try: | |
| return hf_hub_download( | |
| repo_id=source["repo_id"], | |
| filename=source["filename"], | |
| local_dir="/data", | |
| token=HF_TOKEN, | |
| ) | |
| except Exception as exc: | |
| print(f"Model download failed: {type(exc).__name__}: {exc}", flush=True) | |
| return None | |
| def list_gguf_files(): | |
| """List explicitly supported model files.""" | |
| found = [] | |
| for d in SEARCH_DIRS: | |
| if os.path.isdir(d): | |
| try: | |
| for name in sorted(os.listdir(d)): | |
| if name in SUPPORTED_MODELS and name not in found: | |
| found.append(name) | |
| except Exception: | |
| pass | |
| for model_file in MODEL_HUB_SOURCES: | |
| if model_file not in found: | |
| found.append(model_file) | |
| return found | |
| def model_choices(): | |
| """Return choices for UI dropdowns.""" | |
| found = list_gguf_files() | |
| for m in SUPPORTED_MODELS: | |
| if m not in found: | |
| found.append(m) | |
| return found | |
| def resolve_model(model_req: str, choices=None) -> str: | |
| """Resolve an API model ID by exact filename or documented safe alias.""" | |
| choices = choices or list_gguf_files() | |
| aliases = { | |
| "qwen-9b": "Qwen3.8-9B-Q8_0.gguf", | |
| "qwen-9b-q8": "Qwen3.8-9B-Q8_0.gguf", | |
| "qwen-9b-q4": "Qwen3.8-9B-Q4_K_M.gguf", | |
| "qwen3.8-9b": "Qwen3.8-9B-Q8_0.gguf", | |
| "qwen": "Qwen3.8-27B-Q4_K_M.gguf", | |
| "qwen-27b": "Qwen3.8-27B-Q4_K_M.gguf", | |
| "qwen3.8": "Qwen3.8-27B-Q4_K_M.gguf", | |
| "qwen3.8-27b": "Qwen3.8-27B-Q4_K_M.gguf", | |
| "qwen-fast": "Qwen3.8-9B-Q8_0.gguf", | |
| "qwen-q4": "Qwen3.8-27B-Q4_K_M.gguf", | |
| "qwen-q4_k_m": "Qwen3.8-27B-Q4_K_M.gguf", | |
| "qwen-q6": "Qwen3.8-27B-Q6_K.gguf", | |
| "qwen-q6_k": "Qwen3.8-27B-Q6_K.gguf", | |
| "qwen-uncensored": "Qwen3.8-27B-Uncensored.i1-Q4_K_M.gguf", | |
| "qwen-heretic": "Qwen3.8-27B-Uncensored.i1-Q4_K_M.gguf", | |
| "qwen3.8-uncensored": "Qwen3.8-27B-Uncensored.i1-Q4_K_M.gguf", | |
| "gemma-q4": "gemma-4-26B-A4B-it-ultra-uncensored-heretic.i1-Q4_K_M.gguf", | |
| "gemma-q4_k_m": "gemma-4-26B-A4B-it-ultra-uncensored-heretic.i1-Q4_K_M.gguf", | |
| "gemma-q6": "gemma-4-26B-A4B-it-ultra-uncensored-heretic.i1-Q6_K.gguf", | |
| "gemma-q6_k": "gemma-4-26B-A4B-it-ultra-uncensored-heretic.i1-Q6_K.gguf", | |
| } | |
| requested = (model_req or "").strip() | |
| candidate = aliases.get(requested.lower(), requested) | |
| if candidate in choices: | |
| return candidate | |
| if not requested: | |
| return choices[0] | |
| raise HTTPException( | |
| status_code=400, | |
| detail=f"Unknown model '{requested}'. Use GET /v1/models for live model IDs.", | |
| ) | |
| def get_model(model_file: str) -> Llama: | |
| """Load a GGUF model with clean memory management and FlashAttention.""" | |
| global _llm, _loaded_file | |
| model_path = find_model_path(model_file) | |
| if not model_path: | |
| avail = list_gguf_files() | |
| raise gr.Error( | |
| f"Model file '{model_file}' not found in /data or /models. " | |
| f"Available files: {avail}. Please upload your .gguf model to the mounted bucket." | |
| ) | |
| if _llm is not None and _loaded_file == model_file: | |
| return _llm | |
| # Free previous model from memory before loading new one | |
| if _llm is not None: | |
| try: | |
| del _llm | |
| except Exception: | |
| pass | |
| _llm = None | |
| gc.collect() | |
| if "Q6" in model_file: | |
| default_target_ctx = 32768 | |
| elif "9B" in model_file: | |
| default_target_ctx = 163840 | |
| else: | |
| default_target_ctx = 163840 | |
| target_ctx = int(os.environ.get("FLOW_N_CTX", str(default_target_ctx))) | |
| context_candidates = [target_ctx] | |
| for fallback in [163840, 131072, 65536, 32768, 16384, 8192]: | |
| if fallback not in context_candidates and fallback < target_ctx: | |
| context_candidates.append(fallback) | |
| last_error = None | |
| for ctx in context_candidates: | |
| for use_fa in [True, False]: | |
| try: | |
| gc.collect() | |
| kwargs = { | |
| "model_path": model_path, | |
| "n_ctx": ctx, | |
| "n_gpu_layers": -1, | |
| "n_batch": 2048, | |
| "n_ubatch": 512, | |
| "verbose": False, | |
| } | |
| if use_fa: | |
| kwargs["flash_attn"] = True | |
| print(f"Loading {model_file} with n_ctx={ctx}, flash_attn={use_fa}...", flush=True) | |
| _llm = Llama(**kwargs) | |
| _loaded_file = model_file | |
| print(f"Successfully loaded {model_file} with n_ctx={ctx}, flash_attn={use_fa}!", flush=True) | |
| return _llm | |
| except TypeError: | |
| continue | |
| except Exception as e: | |
| last_error = e | |
| print(f"Failed loading with n_ctx={ctx}, flash_attn={use_fa}: {e}", flush=True) | |
| if _llm is not None: | |
| try: | |
| del _llm | |
| except Exception: | |
| pass | |
| _llm = None | |
| gc.collect() | |
| break | |
| if _llm is None and last_error: | |
| raise last_error | |
| return _llm | |
| def parse_model_tool_calls(text: str): | |
| """Extract structured OpenAI tool calls and thinking from model output.""" | |
| if not text: | |
| return None, None, "" | |
| tool_calls = [] | |
| clean_text = text | |
| # Pattern 1: Qwen XML style <tool_call><function=name><parameter=k>v</parameter></function></tool_call> | |
| xml_matches = list(re.finditer(r'<tool_call>\s*<function=([a-zA-Z0-9_\-\.\:\/]+)>(.*?)</function>\s*</tool_call>', clean_text, re.DOTALL)) | |
| if xml_matches: | |
| for idx, m in enumerate(xml_matches): | |
| fn_name = m.group(1).strip() | |
| fn_body = m.group(2) | |
| args = {} | |
| for p in re.finditer(r'<parameter=([a-zA-Z0-9_\-]+)>(.*?)</parameter>', fn_body, re.DOTALL): | |
| p_name = p.group(1).strip() | |
| p_val = p.group(2).strip() | |
| try: | |
| args[p_name] = json.loads(p_val) | |
| except Exception: | |
| args[p_name] = p_val | |
| tool_calls.append({ | |
| "index": idx, | |
| "id": f"call_{uuid.uuid4().hex[:8]}", | |
| "type": "function", | |
| "function": { | |
| "name": fn_name, | |
| "arguments": json.dumps(args, ensure_ascii=False) | |
| } | |
| }) | |
| clean_text = clean_text.replace(m.group(0), "") | |
| # Pattern 2: JSON style <tool_call>{"name": "...", "arguments": {...}}</tool_call> | |
| json_matches = list(re.finditer(r'<tool_call>\s*(\{.*?\})\s*</tool_call>', clean_text, re.DOTALL)) | |
| if json_matches and not tool_calls: | |
| for idx, m in enumerate(json_matches): | |
| raw_json = m.group(1).strip() | |
| try: | |
| parsed = json.loads(raw_json) | |
| fn_name = parsed.get("name", "") | |
| fn_args = parsed.get("arguments", {}) | |
| args_str = json.dumps(fn_args, ensure_ascii=False) if isinstance(fn_args, dict) else str(fn_args) | |
| tool_calls.append({ | |
| "index": idx, | |
| "id": f"call_{uuid.uuid4().hex[:8]}", | |
| "type": "function", | |
| "function": { | |
| "name": fn_name, | |
| "arguments": args_str | |
| } | |
| }) | |
| clean_text = clean_text.replace(m.group(0), "") | |
| except Exception: | |
| pass | |
| # Extract thinking/reasoning if present | |
| reasoning_content = None | |
| think_m = re.search(r'<think>(.*?)</think>', clean_text, re.DOTALL) | |
| if think_m: | |
| reasoning_content = think_m.group(1).strip() | |
| clean_text = clean_text.replace(think_m.group(0), "") | |
| elif "</think>" in clean_text: | |
| parts = clean_text.split("</think>", 1) | |
| reasoning_content = parts[0].replace("<think>", "").strip() | |
| clean_text = parts[1] | |
| elif tool_calls and clean_text.strip(): | |
| # Any text preceding tool calls without <think> tags is reasoning | |
| reasoning_content = clean_text.strip() | |
| clean_text = "" | |
| clean_content = clean_text.strip() | |
| if tool_calls and not clean_content: | |
| clean_content = None | |
| return (tool_calls if tool_calls else None), (reasoning_content if reasoning_content else None), clean_content | |
| def format_openai_messages_for_model(messages): | |
| """Normalize multi-turn OpenAI messages including tool results into prompt format.""" | |
| formatted = [] | |
| for msg in messages: | |
| role = msg.get("role", "user") | |
| content = msg.get("content") | |
| tool_calls = msg.get("tool_calls") | |
| if role == "tool": | |
| formatted.append({ | |
| "role": "user", | |
| "content": f"<tool_response>\n{content or ''}\n</tool_response>" | |
| }) | |
| elif role == "assistant" and tool_calls: | |
| tc_text = "" | |
| for tc in tool_calls: | |
| fn = tc.get("function", {}) | |
| fn_name = fn.get("name", "") | |
| raw_args = fn.get("arguments", "{}") | |
| try: | |
| args_dict = json.loads(raw_args) if isinstance(raw_args, str) else raw_args | |
| except Exception: | |
| args_dict = {} | |
| tc_text += f"\n<tool_call>\n<function={fn_name}>\n" | |
| if isinstance(args_dict, dict): | |
| for k, v in args_dict.items(): | |
| tc_text += f"<parameter={k}>\n{json.dumps(v) if isinstance(v, (dict, list)) else v}\n</parameter>\n" | |
| tc_text += "</function>\n</tool_call>" | |
| combined = (content or "") + tc_text | |
| formatted.append({"role": "assistant", "content": combined.strip()}) | |
| else: | |
| formatted.append({"role": role, "content": content or ""}) | |
| return formatted | |
| def _format_api_error(e: Exception, action: str) -> str: | |
| """Format API errors with clear budget and credit guidance.""" | |
| msg = str(e) | |
| if "402" in msg or "Payment Required" in msg: | |
| return ( | |
| f"{action} notice (402 Payment Required): Your monthly HF Inference API credit " | |
| "($2/month included with HF PRO) has been fully used for this billing period. " | |
| "ZeroGPU tabs (Qwen 3.8 / Gemma LLM Chat) remain 100% free and functional!" | |
| ) | |
| return f"{action} failed: {msg}" | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # TAB 1: Chat & Agent Runner (ZeroGPU Large β 40 min/day) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def generate_openai_chat(messages, model_file, temperature, max_tokens, tools=None, tool_choice=None): | |
| llm = get_model(model_file) | |
| kwargs = { | |
| "messages": messages, | |
| "max_tokens": (int(max_tokens) if max_tokens not in (None, "") else None), | |
| "temperature": float(temperature), | |
| "top_p": 0.95, | |
| } | |
| if tools: | |
| kwargs["tools"] = tools | |
| kwargs["tool_choice"] = tool_choice or "auto" | |
| return llm.create_chat_completion(**kwargs) | |
| def generate_openai_chat_stream(messages, model_file, temperature, max_tokens, tools=None, tool_choice=None): | |
| llm = get_model(model_file) | |
| kwargs = { | |
| "messages": messages, | |
| "max_tokens": (int(max_tokens) if max_tokens not in (None, "") else None), | |
| "temperature": float(temperature), | |
| "top_p": 0.95, | |
| "stream": True, | |
| } | |
| if tools: | |
| kwargs["tools"] = tools | |
| kwargs["tool_choice"] = tool_choice or "auto" | |
| for chunk in llm.create_chat_completion(**kwargs): | |
| yield chunk | |
| def custom_chat_handler(user_msg, history, model_file, system_prompt, temperature, max_tokens): | |
| """Rich chat execution with live token telemetry, reasoning, and speed reporting.""" | |
| if not user_msg or not user_msg.strip(): | |
| return history or [], "β‘ *Ready β Enter a prompt to start inference.*", "" | |
| history = list(history or []) | |
| history.append({"role": "user", "content": user_msg.strip()}) | |
| messages = [] | |
| if system_prompt.strip(): | |
| messages.append({"role": "system", "content": system_prompt.strip()}) | |
| for item in history: | |
| messages.append({"role": item.get("role", "user"), "content": item.get("content", "")}) | |
| t0 = time.time() | |
| try: | |
| raw_res = generate_openai_chat( | |
| messages, | |
| model_file, | |
| float(temperature), | |
| (int(max_tokens) if max_tokens not in (None, "") else None), | |
| ) | |
| t1 = time.time() | |
| elapsed = max(0.01, t1 - t0) | |
| raw_content = raw_res["choices"][0]["message"].get("content", "") | |
| tool_calls, reasoning, clean = parse_model_tool_calls(raw_content) | |
| formatted_bot = "" | |
| if reasoning: | |
| formatted_bot += f"<details open><summary>π§ <b>Deep Thinking & Reasoning</b></summary>\n\n```markdown\n{reasoning}\n```\n</details>\n\n" | |
| if tool_calls: | |
| formatted_bot += f"<details open><summary>π οΈ <b>Executed Tool Calls ({len(tool_calls)})</b></summary>\n\n```json\n{json.dumps(tool_calls, indent=2)}\n```\n</details>\n\n" | |
| if clean: | |
| formatted_bot += clean | |
| elif not reasoning and not tool_calls: | |
| formatted_bot += raw_content | |
| history.append({"role": "assistant", "content": formatted_bot}) | |
| # Telemetry calculations | |
| raw_usage = raw_res.get("usage", {}) | |
| prompt_toks = raw_usage.get("prompt_tokens") or sum(max(1, int(len(m["content"].split()) * 1.3)) for m in messages) | |
| comp_toks = raw_usage.get("completion_tokens") or max(1, int(len(raw_content.split()) * 1.3)) | |
| tot_toks = prompt_toks + comp_toks | |
| tps = comp_toks / elapsed | |
| ctx_pct = (tot_toks / 262144) * 100 | |
| hud_md = ( | |
| f"<div style='display: flex; flex-wrap: wrap; gap: 12px; font-size: 0.85rem; padding: 8px 12px; " | |
| f"background: rgba(30, 41, 59, 0.7); border-radius: 8px; border: 1px solid rgba(255,255,255,0.1); font-family: monospace;'>" | |
| f"<span>β‘ <b>{tps:.1f} t/s</b></span>" | |
| f"<span>β±οΈ <b>{elapsed:.2f}s</b></span>" | |
| f"<span>π₯ Prompt: <b>{prompt_toks}</b></span>" | |
| f"<span>π€ Output: <b>{comp_toks}</b></span>" | |
| f"<span>π§ Context: <b>{tot_toks:,} / 262,144 ({ctx_pct:.1f}%)</b></span>" | |
| f"<span style='color: #4ade80;'>β ZeroGPU Large (48GB)</span>" | |
| f"</div>" | |
| ) | |
| return history, hud_md, "" | |
| except Exception as e: | |
| history.append({"role": "assistant", "content": f"β **Inference Error:** {str(e)}"}) | |
| return history, f"β οΈ *Execution error after {time.time()-t0:.2f}s: {str(e)}*", "" | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # TAB 2: Vision & Multimodal OCR | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def analyze_vision(image_input, prompt_text): | |
| """Analyze images, UI screenshots, code diagrams or documents.""" | |
| if image_input is None: | |
| raise gr.Error("Please upload or capture an image first.") | |
| if not HF_TOKEN: | |
| raise gr.Error("Set HF_TOKEN in Space secrets to use the Vision API.") | |
| prompt = prompt_text.strip() or "Describe this image in detail and extract all visible text and code." | |
| try: | |
| response = api_client.chat_completion( | |
| messages=[ | |
| { | |
| "role": "user", | |
| "content": [ | |
| {"type": "text", "text": prompt}, | |
| {"type": "image_url", "image_url": {"url": image_input if isinstance(image_input, str) else image_input}}, | |
| ], | |
| } | |
| ], | |
| model=VISION_MODEL, | |
| max_tokens=1024, | |
| ) | |
| return response.choices[0].message.content | |
| except Exception as e: | |
| try: | |
| return api_client.image_to_text(image=image_input, model="Salesforce/blip-image-captioning-large") | |
| except Exception: | |
| raise gr.Error(_format_api_error(e, "Vision analysis")) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # ZERO-GPU VIDEO GENERATION PIPELINE (40 min/day A100 Quota - $0 API Cost) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def generate_zerogpu_video( | |
| prompt: str, | |
| negative_prompt: str = "", | |
| model_choice: str = "ZeroScope v2 (576w High-Res)", | |
| num_frames: int = 16, | |
| fps: int = 8, | |
| guidance_scale: float = 7.5, | |
| seed: int = -1 | |
| ): | |
| """Generate dynamic MP4 video using ZeroGPU open-weights models.""" | |
| if not prompt.strip(): | |
| raise gr.Error("Please enter a video prompt.") | |
| repo_id = VIDEO_MODELS.get(model_choice, "cerspense/zeroscope_v2_576w") | |
| out_video_path = f"/tmp/zerogpu_video_{int(time.time())}_{abs(hash(prompt)) % 10000}.mp4" | |
| try: | |
| import torch | |
| from diffusers import DiffusionPipeline, DPMSolverMultistepScheduler | |
| from diffusers.utils import export_to_video | |
| device = "cuda" if torch.cuda.is_available() else "cpu" | |
| dtype = torch.float16 if device == "cuda" else torch.float32 | |
| if repo_id not in _video_pipeline_cache: | |
| pipe = DiffusionPipeline.from_pretrained(repo_id, torch_dtype=dtype) | |
| if hasattr(pipe, "scheduler"): | |
| try: | |
| pipe.scheduler = DPMSolverMultistepScheduler.from_config(pipe.scheduler.config) | |
| except Exception: | |
| pass | |
| if hasattr(pipe, "enable_model_cpu_offload") and device == "cuda": | |
| pipe.enable_model_cpu_offload() | |
| else: | |
| pipe = pipe.to(device) | |
| _video_pipeline_cache[repo_id] = pipe | |
| else: | |
| pipe = _video_pipeline_cache[repo_id] | |
| actual_seed = seed if (seed and int(seed) >= 0) else random.randint(0, 2**31 - 1) | |
| generator = torch.Generator(device=device).manual_seed(actual_seed) | |
| video_frames = pipe( | |
| prompt=prompt.strip(), | |
| negative_prompt=negative_prompt.strip() if negative_prompt else None, | |
| num_inference_steps=24, | |
| guidance_scale=float(guidance_scale), | |
| num_frames=int(num_frames), | |
| generator=generator | |
| ).frames[0] | |
| export_to_video(video_frames, out_video_path, fps=int(fps)) | |
| return out_video_path | |
| except Exception as exc: | |
| print(f"[ZeroGPU Video] Direct pipeline exception: {exc}, running ffmpeg dynamic visualizer fallback...", flush=True) | |
| try: | |
| from engine.model_dispatcher import ModelDispatcher | |
| disp = ModelDispatcher() | |
| res = disp.generate_image(prompt=prompt, aspect_ratio="16:9") | |
| img_path = res.get("filepath") | |
| if img_path and os.path.exists(img_path): | |
| ffmpeg_bin = shutil.which("ffmpeg") or "/opt/homebrew/bin/ffmpeg" | |
| dur = max(3, int(int(num_frames) / max(1, int(fps)))) | |
| subprocess.run( | |
| [ | |
| ffmpeg_bin, "-y", "-loop", "1", "-i", img_path, | |
| "-vf", f"fps={fps},scale=768:432,zoompan=z='min(zoom+0.0015,1.15)':d={dur*fps}:s=768x432", | |
| "-c:v", "libx264", "-t", str(dur), "-pix_fmt", "yuv420p", | |
| out_video_path | |
| ], | |
| capture_output=True, | |
| timeout=20 | |
| ) | |
| if os.path.exists(out_video_path) and os.path.getsize(out_video_path) > 0: | |
| return out_video_path | |
| except Exception as e2: | |
| print(f"[ZeroGPU Video] Fallback failed: {e2}") | |
| raise gr.Error(f"ZeroGPU Video error: {exc}") | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # ZERO-GPU MUSIC & AUDIO GENERATION PIPELINE (40 min/day A100 Quota - $0 Cost) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def generate_zerogpu_music( | |
| prompt: str, | |
| lyrics: str = "", | |
| model_choice: str = "MiniMax Music 3 (full song + vocals)", | |
| duration_seconds: int = 60, | |
| guidance_scale: float = 3.0, | |
| temperature: float = 1.0, | |
| seed: int = 7 | |
| ): | |
| """Generate real full-song audio on ZeroGPU. Procedural MIDI fallback is never used here.""" | |
| if not prompt.strip() and not lyrics.strip(): | |
| raise gr.Error("Please enter a music description or lyrics.") | |
| repo_id = AUDIO_MUSIC_MODELS.get(model_choice, "MiniMaxAI/MiniMax-Music3") | |
| dur = max(5, min(300, int(duration_seconds or 60))) | |
| out_audio_path = f"/tmp/zerogpu_music_{int(time.time())}_{abs(hash(prompt + lyrics)) % 10000}.wav" | |
| try: | |
| import torch | |
| import soundfile as sf | |
| # Official MiniMax Music3 path from its model card. It supports lyrics + | |
| # detailed music description and produces complete vocal songs up to 5 min. | |
| if repo_id == "MiniMaxAI/MiniMax-Music3": | |
| from diffusers import ModularPipeline | |
| import numpy as np | |
| if repo_id not in _audio_pipeline_cache: | |
| pipe = ModularPipeline.from_pretrained(repo_id) | |
| pipe.load_components(dtype=torch.bfloat16) | |
| pipe.to("cuda") | |
| _audio_pipeline_cache[repo_id] = pipe | |
| else: | |
| pipe = _audio_pipeline_cache[repo_id] | |
| raw_output = pipe( | |
| prompt=prompt.strip(), | |
| lyrics=lyrics.strip(), | |
| audio_duration=float(dur), | |
| generator=torch.Generator("cuda").manual_seed(int(seed)), | |
| output="audios", | |
| ) | |
| audio = raw_output[0] | |
| if hasattr(audio, "detach"): | |
| audio_np = audio.detach().cpu().float().numpy() | |
| elif isinstance(audio, np.ndarray): | |
| audio_np = audio.astype(np.float32) | |
| else: | |
| audio_np = np.asarray(audio, dtype=np.float32) | |
| if audio_np.ndim > 1 and audio_np.shape[0] < audio_np.shape[1]: | |
| audio_np = audio_np.T | |
| sr = getattr(pipe, "sampling_rate", 44100) | |
| sf.write(out_audio_path, audio_np, sr) | |
| return out_audio_path | |
| # Stable Audio 3 uses its own pipeline and may require HF access approval. | |
| if repo_id == "stabilityai/stable-audio-3-medium": | |
| # The public Stability release currently uses the separate | |
| # `stable_audio_3` package, not a Diffusers StableAudio3Pipeline. | |
| # Do not pretend this selector works or silently substitute audio. | |
| raise RuntimeError( | |
| "Stable Audio 3 is not enabled in this Space yet: its official " | |
| "stable_audio_3 runtime is not installed. Select MiniMax Music 3." | |
| ) | |
| # MusicGen is explicitly instrumental and does not reliably sing lyrics. | |
| if repo_id.startswith("facebook/musicgen"): | |
| from transformers import AutoProcessor, MusicgenForConditionalGeneration | |
| if repo_id not in _audio_pipeline_cache: | |
| processor = AutoProcessor.from_pretrained(repo_id) | |
| model = MusicgenForConditionalGeneration.from_pretrained(repo_id, torch_dtype=torch.float16).to("cuda") | |
| _audio_pipeline_cache[repo_id] = (processor, model) | |
| processor, model = _audio_pipeline_cache[repo_id] | |
| inputs = processor(text=[prompt.strip()], padding=True, return_tensors="pt").to("cuda") | |
| audio_values = model.generate( | |
| **inputs, do_sample=True, guidance_scale=float(guidance_scale), | |
| max_new_tokens=min(1500, int(dur * 50)), temperature=float(temperature) | |
| ) | |
| sf.write(out_audio_path, audio_values[0, 0].detach().cpu().numpy(), model.config.audio_encoder.sampling_rate) | |
| return out_audio_path | |
| raise gr.Error(f"Model '{repo_id}' is not wired for full-song generation yet. Choose MiniMax Music 3 or Stable Audio 3.") | |
| except Exception as exc: | |
| raise gr.Error(f"ZeroGPU model '{repo_id}' failed: {exc}. No MIDI/procedural fallback was used.") | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # TAB 3: FLUX.1 & Diffusion Image Studio (ZeroGPU Local vs Serverless Toggle) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def generate_image(prompt, negative_prompt, guidance_scale, aspect_ratio, style_preset, model_choice, execution_mode): | |
| """Generate high-quality images with FLUX.1 / SDXL on ZeroGPU ($0 cost) or Serverless API.""" | |
| if not prompt.strip(): | |
| raise gr.Error("Please enter an image prompt.") | |
| full_prompt = prompt.strip() | |
| if style_preset and style_preset != "None / Natural": | |
| full_prompt = f"{full_prompt}, in {style_preset} style, 8k resolution, cinematic lighting, masterpiece" | |
| dims = { | |
| "1:1 Square (1024x1024)": (1024, 1024), | |
| "16:9 Landscape (1024x576)": (1024, 576), | |
| "9:16 Portrait (576x1024)": (576, 1024), | |
| "4:3 Standard (1024x768)": (1024, 768), | |
| } | |
| width, height = dims.get(aspect_ratio, (1024, 1024)) | |
| repo_id = IMAGE_MODELS.get(model_choice, "black-forest-labs/FLUX.1-schnell") | |
| # Mode 2: Serverless Inference API (uses $2/mo limit) | |
| if "Serverless" in str(execution_mode): | |
| if not HF_TOKEN: | |
| raise gr.Error("Set HF_TOKEN in Space secrets to use Serverless Inference API.") | |
| try: | |
| return api_client.text_to_image( | |
| prompt=full_prompt, | |
| model=repo_id, | |
| guidance_scale=float(guidance_scale), | |
| width=width, | |
| height=height, | |
| ) | |
| except Exception as e: | |
| raise gr.Error(_format_api_error(e, "Serverless Inference API")) | |
| # Mode 1: ZeroGPU Local Diffusers ($0 Cost / 40 min A100 Quota) | |
| try: | |
| import torch | |
| from diffusers import AutoPipelineForText2Image, FluxPipeline | |
| device = "cuda" if torch.cuda.is_available() else "cpu" | |
| dtype = torch.bfloat16 if "FLUX" in repo_id else (torch.float16 if device == "cuda" else torch.float32) | |
| if repo_id not in _image_pipeline_cache: | |
| if "FLUX" in repo_id: | |
| pipe = FluxPipeline.from_pretrained(repo_id, torch_dtype=dtype) | |
| else: | |
| pipe = AutoPipelineForText2Image.from_pretrained(repo_id, torch_dtype=dtype) | |
| if hasattr(pipe, "enable_model_cpu_offload") and device == "cuda": | |
| pipe.enable_model_cpu_offload() | |
| else: | |
| pipe = pipe.to(device) | |
| _image_pipeline_cache[repo_id] = pipe | |
| else: | |
| pipe = _image_pipeline_cache[repo_id] | |
| steps = 4 if ("schnell" in repo_id or "turbo" in repo_id) else 25 | |
| image = pipe( | |
| prompt=full_prompt, | |
| negative_prompt=negative_prompt.strip() if negative_prompt else None, | |
| guidance_scale=float(guidance_scale), | |
| num_inference_steps=steps, | |
| width=width, | |
| height=height | |
| ).images[0] | |
| return image | |
| except Exception as exc: | |
| print(f"[ZeroGPU Diffusers] Local pipeline fallback: {exc}", flush=True) | |
| if HF_TOKEN: | |
| try: | |
| return api_client.text_to_image( | |
| prompt=full_prompt, | |
| model="black-forest-labs/FLUX.1-schnell", | |
| width=width, | |
| height=height | |
| ) | |
| except Exception: | |
| pass | |
| from engine.model_dispatcher import ModelDispatcher | |
| disp = ModelDispatcher() | |
| res = disp.generate_image(prompt=full_prompt, aspect_ratio="1:1" if width == height else "16:9") | |
| from PIL import Image | |
| if res.get("filepath") and os.path.exists(res["filepath"]): | |
| return Image.open(res["filepath"]) | |
| raise gr.Error(f"Image generation error: {exc}") | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # TAB MOLDOVAN AI CREATIVE STUDIO HELPERS | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def create_moldovan_song_handler(topic, genre, dialect_level, duration, custom_lyrics, music_engine): | |
| """Generate complete Moldovan song with selected real music model on ZeroGPU.""" | |
| from engine.media_creator import MediaCreator | |
| mc = MediaCreator() | |
| dur = max(15, min(300, int(duration or 60))) | |
| topic_clean = topic or "ChiΘinΔu Vibe" | |
| genre_clean = genre or "ChiΘinΔu 808 Trap" | |
| dialect = int(dialect_level or 2) | |
| if custom_lyrics and custom_lyrics.strip(): | |
| lyrics_raw = custom_lyrics.strip() | |
| else: | |
| lyrics_raw = mc._synthesize_song_lyrics(topic_clean, genre_clean, dialect, duration_seconds=dur) | |
| audio_path = None | |
| if "MiniMax" in str(music_engine): | |
| # Direct ZeroGPU MiniMax ModularPipeline in-process execution | |
| audio_path = generate_zerogpu_music( | |
| prompt=f"Moldovan {genre_clean}, authentic balkan urban vibes, high quality studio sound", | |
| lyrics=lyrics_raw, | |
| model_choice="MiniMax Music 3 (full song + vocals)", | |
| duration_seconds=dur | |
| ) | |
| else: | |
| engine_map = { | |
| "ACE-Step XL (full song + vocals)": "acestep", | |
| "MusicGen (instrumental only)": "musicgen", | |
| } | |
| song = mc.generate_song( | |
| topic=topic_clean, | |
| genre=genre_clean, | |
| duration_seconds=dur, | |
| dialect_level=dialect, | |
| custom_lyrics=lyrics_raw, | |
| audio_engine=engine_map.get(music_engine, "synth808") | |
| ) | |
| audio_path = os.path.join(MOLDOVAN_DIR, "data/generated_audio", song.get("audio_filename", "")) | |
| if not os.path.exists(audio_path): | |
| audio_path = None | |
| lyrics_display = f"### π΅ {topic_clean} ({genre_clean})\n**Engine:** `{music_engine}` | **Duration:** {dur}s\n\n```text\n{lyrics_raw}\n```\n\n**Suno/Udio Prompt Blueprint:**\n`[{genre_clean}, Moldovan Romanian urban dialect, energetic, viral hook, studio mastering]`" | |
| return lyrics_display, audio_path | |
| def chat_moldovan_persona_handler(persona_id, message, history): | |
| """Interactive chat with authentic Moldovan cultural personas.""" | |
| if not message.strip(): | |
| return history or [], "" | |
| from personas.persona_engine import PersonaEngine | |
| pe = PersonaEngine() | |
| res = pe.chat_with_persona(persona_id=persona_id or "taximetrist", user_message=message, chat_history=history or []) | |
| new_history = history + [ | |
| {"role": "user", "content": message}, | |
| {"role": "assistant", "content": res.get("reply", "")} | |
| ] | |
| return new_history, "" | |
| def convert_dialect_handler(text, level): | |
| """Convert standard Romanian text into authentic regional Moldovan dialect.""" | |
| if not text.strip(): | |
| return "" | |
| from linguistics.dialect_converter import DialectConverter | |
| conv = DialectConverter() | |
| res = conv.convert_to_moldovan(text, level=int(level or 2)) | |
| return f"**Moldovan Conversion (Level {level}):**\n\n{res.get('converted', '')}\n\n*Applied rules:* {len(res.get('applied_rules', []))}" | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # TAB 4: Audio Suite (Whisper STT & Kokoro TTS) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def transcribe_audio(audio_input): | |
| """Transcribe audio with Whisper Large v3.""" | |
| if audio_input is None: | |
| raise gr.Error("Please record or upload audio.") | |
| if not HF_TOKEN: | |
| raise gr.Error("Set HF_TOKEN in Space secrets to use Whisper.") | |
| try: | |
| result = api_client.automatic_speech_recognition( | |
| audio=audio_input, | |
| model=WHISPER_MODEL, | |
| ) | |
| return result.text if hasattr(result, "text") else str(result) | |
| except Exception as e: | |
| raise gr.Error(_format_api_error(e, "Whisper transcription")) | |
| def generate_tts(text_input): | |
| """Synthesize high-fidelity speech with Kokoro-82M.""" | |
| if not text_input.strip(): | |
| raise gr.Error("Please enter text to synthesize.") | |
| if not HF_TOKEN: | |
| raise gr.Error("Set HF_TOKEN in Space secrets to use Text-to-Speech.") | |
| try: | |
| audio_bytes = api_client.text_to_speech( | |
| text=text_input.strip(), | |
| model=KOKORO_TTS_MODEL, | |
| ) | |
| return audio_bytes | |
| except Exception as e: | |
| raise gr.Error(_format_api_error(e, "Kokoro TTS synthesis")) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # TAB 5: Voice-to-Art Pipeline | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def voice_to_art(audio_input, model_file, art_style): | |
| """Whisper STT -> Qwen Prompt Engineer -> FLUX Renderer.""" | |
| if audio_input is None: | |
| raise gr.Error("Please record or upload audio first.") | |
| if not HF_TOKEN: | |
| raise gr.Error("Set HF_TOKEN in Space secrets.") | |
| try: | |
| transcription = api_client.automatic_speech_recognition( | |
| audio=audio_input, | |
| model=WHISPER_MODEL, | |
| ) | |
| raw_text = transcription.text if hasattr(transcription, "text") else str(transcription) | |
| except Exception as e: | |
| raise gr.Error(_format_api_error(e, "Whisper transcription")) | |
| if not raw_text.strip(): | |
| raise gr.Error("Could not understand the audio.") | |
| llm = get_model(model_file) | |
| style_hint = f" in {art_style} style" if art_style.strip() else "" | |
| expand_prompt = ( | |
| f"You are a master image prompt engineer. The user said: \"{raw_text}\"\n\n" | |
| f"Write a single, highly detailed, vivid FLUX image generation prompt{style_hint}. " | |
| f"Include composition, cinematic lighting, color palette, mood, and fine details. " | |
| f"Output ONLY the prompt, nothing else. Max 100 words." | |
| ) | |
| response = llm.create_chat_completion( | |
| messages=[{"role": "user", "content": expand_prompt}], | |
| max_tokens=256, | |
| temperature=0.85, | |
| top_p=0.95, | |
| ) | |
| art_prompt = response["choices"][0]["message"]["content"].strip() | |
| _, _, clean_art_prompt = parse_model_tool_calls(art_prompt) | |
| final_prompt = clean_art_prompt or art_prompt | |
| try: | |
| image = api_client.text_to_image( | |
| prompt=final_prompt, | |
| model=FLUX_MODEL, | |
| guidance_scale=3.5, | |
| width=1024, | |
| height=1024, | |
| ) | |
| except Exception as e: | |
| raise gr.Error(_format_api_error(e, "FLUX image generation")) | |
| return raw_text, final_prompt, image | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # TAB 6: Embeddings Lab | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def compute_similarity(text_a, text_b): | |
| """Compute 1024-dim dense embeddings and cosine similarity using BGE-M3.""" | |
| if not text_a.strip() or not text_b.strip(): | |
| raise gr.Error("Please enter both Text A and Text B.") | |
| if not HF_TOKEN: | |
| raise gr.Error("Set HF_TOKEN in Space secrets to use Embeddings.") | |
| try: | |
| emb_a = api_client.feature_extraction(text=text_a.strip(), model=EMBEDDING_MODEL) | |
| emb_b = api_client.feature_extraction(text=text_b.strip(), model=EMBEDDING_MODEL) | |
| vec_a = emb_a[0] if isinstance(emb_a, list) and isinstance(emb_a[0], list) else emb_a | |
| vec_b = emb_b[0] if isinstance(emb_b, list) and isinstance(emb_b[0], list) else emb_b | |
| dot = sum(a * b for a, b in zip(vec_a, vec_b)) | |
| norm_a = math.sqrt(sum(a * a for a in vec_a)) | |
| norm_b = math.sqrt(sum(b * b for b in vec_b)) | |
| similarity = dot / (norm_a * norm_b) if (norm_a > 0 and norm_b > 0) else 0.0 | |
| score_percent = round(similarity * 100, 2) | |
| interp = ( | |
| "π’ Identical / Paraphrase" if score_percent > 85 else | |
| "π‘ Highly Related" if score_percent > 65 else | |
| "π Moderately Related" if score_percent > 40 else | |
| "π΄ Distinct / Unrelated" | |
| ) | |
| dim_len = len(vec_a) | |
| vector_preview_a = str(vec_a[:5])[:-1] + ", ...]" | |
| vector_preview_b = str(vec_b[:5])[:-1] + ", ...]" | |
| report = ( | |
| f"### π― Cosine Similarity: **{score_percent}%** ({interp})\n\n" | |
| f"<div style='height: 8px; width: 100%; background: #334155; border-radius: 4px; overflow: hidden; margin-bottom: 16px;'>" | |
| f"<div style='height: 100%; width: {score_percent}%; background: linear-gradient(90deg, #38bdf8, #818cf8);'></div>" | |
| f"</div>\n\n" | |
| f"- **Embedding Model:** `{EMBEDDING_MODEL}`\n" | |
| f"- **Vector Dimensionality:** `{dim_len}` float32 elements\n\n" | |
| f"**Vector Preview A:** `{vector_preview_a}`\n\n" | |
| f"**Vector Preview B:** `{vector_preview_b}`" | |
| ) | |
| return report | |
| except Exception as e: | |
| raise gr.Error(_format_api_error(e, "Embedding calculation")) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # OpenAI-Compatible API Endpoints | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| fastapi_app = FastAPI(title="ZeroGPU Private OpenAI API Hub", version="2.0.0") | |
| def _authorize_api_request(request: Request) -> None: | |
| """Require bearer auth only if FLOW_API_KEY is explicitly set in Space secrets.""" | |
| expected = os.environ.get("FLOW_API_KEY") | |
| if not expected: | |
| return | |
| authorization = request.headers.get("authorization", "") | |
| scheme, _, supplied = authorization.partition(" ") | |
| if scheme.lower() != "bearer" or not supplied or not hmac.compare_digest(supplied, expected): | |
| raise HTTPException(status_code=401, detail="Invalid or missing bearer token.") | |
| async def openai_chat_completions(request: Request): | |
| _authorize_api_request(request) | |
| try: | |
| body = await request.json() | |
| except Exception: | |
| raise HTTPException(status_code=400, detail="Invalid JSON body") | |
| messages = body.get("messages", []) | |
| if not messages: | |
| raise HTTPException(status_code=400, detail="Field 'messages' is required.") | |
| model_req = body.get("model", "") | |
| choices = list_gguf_files() | |
| if not choices: | |
| raise HTTPException(status_code=500, detail="No GGUF models available in Space storage.") | |
| selected_model = resolve_model(model_req, choices) | |
| temperature = float(body.get("temperature", 0.7)) | |
| raw_max_tokens = body.get("max_tokens") | |
| max_tokens = int(raw_max_tokens) if raw_max_tokens not in (None, "") else None | |
| formatted_msgs = format_openai_messages_for_model(messages) | |
| tools = body.get("tools") | |
| tool_choice = body.get("tool_choice") | |
| stream = bool(body.get("stream", False)) | |
| try: | |
| raw_res = generate_openai_chat( | |
| formatted_msgs, selected_model, temperature, max_tokens, tools, tool_choice | |
| ) | |
| except Exception as e: | |
| err_msg = str(e) | |
| status_code = 429 if ("limit" in err_msg.lower() or "quota" in err_msg.lower()) else 500 | |
| raise HTTPException(status_code=status_code, detail=f"ZeroGPU inference error: {err_msg}") | |
| raw_message = raw_res["choices"][0]["message"] | |
| raw_content = raw_message.get("content", "") | |
| raw_finish = raw_res["choices"][0].get("finish_reason", "stop") | |
| tool_calls, reasoning, clean_content = parse_model_tool_calls(raw_content) | |
| if stream: | |
| async def event_generator(): | |
| cid = f"chatcmpl-{int(time.time()*1000)}" | |
| created_ts = int(time.time()) | |
| # Step 1: Stream reasoning chunk if present | |
| if reasoning: | |
| chunk1 = { | |
| "id": cid, | |
| "object": "chat.completion.chunk", | |
| "created": created_ts, | |
| "model": selected_model, | |
| "choices": [ | |
| { | |
| "index": 0, | |
| "delta": { | |
| "role": "assistant", | |
| "reasoning_content": reasoning | |
| }, | |
| "finish_reason": None | |
| } | |
| ] | |
| } | |
| yield f"data: {json.dumps(chunk1)}\n\n" | |
| # Step 2: Stream tool calls or text content | |
| if tool_calls: | |
| chunk2 = { | |
| "id": cid, | |
| "object": "chat.completion.chunk", | |
| "created": created_ts, | |
| "model": selected_model, | |
| "choices": [ | |
| { | |
| "index": 0, | |
| "delta": { | |
| "tool_calls": tool_calls | |
| }, | |
| "finish_reason": None | |
| } | |
| ] | |
| } | |
| yield f"data: {json.dumps(chunk2)}\n\n" | |
| chunk3 = { | |
| "id": cid, | |
| "object": "chat.completion.chunk", | |
| "created": created_ts, | |
| "model": selected_model, | |
| "choices": [ | |
| { | |
| "index": 0, | |
| "delta": {}, | |
| "finish_reason": "tool_calls" | |
| } | |
| ] | |
| } | |
| yield f"data: {json.dumps(chunk3)}\n\n" | |
| else: | |
| if clean_content: | |
| chunk_text = { | |
| "id": cid, | |
| "object": "chat.completion.chunk", | |
| "created": created_ts, | |
| "model": selected_model, | |
| "choices": [ | |
| { | |
| "index": 0, | |
| "delta": { | |
| "role": "assistant", | |
| "content": clean_content | |
| }, | |
| "finish_reason": None | |
| } | |
| ] | |
| } | |
| yield f"data: {json.dumps(chunk_text)}\n\n" | |
| chunk_finish = { | |
| "id": cid, | |
| "object": "chat.completion.chunk", | |
| "created": created_ts, | |
| "model": selected_model, | |
| "choices": [ | |
| { | |
| "index": 0, | |
| "delta": {}, | |
| "finish_reason": "stop" | |
| } | |
| ] | |
| } | |
| yield f"data: {json.dumps(chunk_finish)}\n\n" | |
| yield "data: [DONE]\n\n" | |
| return StreamingResponse(event_generator(), media_type="text/event-stream") | |
| out_message = {"role": "assistant"} | |
| if tool_calls: | |
| out_message["tool_calls"] = tool_calls | |
| out_message["content"] = clean_content | |
| finish_reason = "tool_calls" | |
| else: | |
| out_message["content"] = clean_content if clean_content is not None else raw_content | |
| finish_reason = raw_finish | |
| if reasoning: | |
| out_message["reasoning_content"] = reasoning | |
| raw_usage = raw_res.get("usage") if isinstance(raw_res, dict) else {} | |
| prompt_tokens = raw_usage.get("prompt_tokens") if raw_usage else None | |
| completion_tokens = raw_usage.get("completion_tokens") if raw_usage else None | |
| def _text(value): | |
| return value if isinstance(value, str) else ("" if value is None else str(value)) | |
| if prompt_tokens is None or prompt_tokens == 0: | |
| prompt_tokens = sum(max(1, int(len(_text(m.get("content")).split()) * 1.3)) for m in formatted_msgs) | |
| if completion_tokens is None or completion_tokens == 0: | |
| full_generated = raw_content or "" | |
| completion_tokens = max(1, int(len(full_generated.split()) * 1.3)) if full_generated else 0 | |
| usage_obj = { | |
| "prompt_tokens": prompt_tokens, | |
| "completion_tokens": completion_tokens, | |
| "total_tokens": prompt_tokens + completion_tokens, | |
| } | |
| if reasoning: | |
| reasoning_tok_count = max(1, int(len(reasoning.split()) * 1.3)) | |
| usage_obj["completion_tokens_details"] = { | |
| "reasoning_tokens": reasoning_tok_count, | |
| } | |
| return { | |
| "id": f"chatcmpl-{int(time.time()*1000)}", | |
| "object": "chat.completion", | |
| "created": int(time.time()), | |
| "model": selected_model, | |
| "choices": [ | |
| { | |
| "index": 0, | |
| "message": out_message, | |
| "finish_reason": finish_reason | |
| } | |
| ], | |
| "usage": usage_obj | |
| } | |
| async def list_openai_models(request: Request): | |
| _authorize_api_request(request) | |
| choices = list_gguf_files() | |
| models_data = [] | |
| # GGUF LLM models | |
| for c in choices: | |
| models_data.append({ | |
| "id": c, | |
| "object": "model", | |
| "created": int(time.time()), | |
| "owned_by": "abalanescu-flow", | |
| "permission": [], | |
| }) | |
| # Audio / Music models | |
| for name, repo_id in AUDIO_MUSIC_MODELS.items(): | |
| models_data.append({ | |
| "id": repo_id, | |
| "object": "model", | |
| "created": int(time.time()), | |
| "owned_by": "abalanescu-flow-zerogpu-audio", | |
| "permission": [], | |
| }) | |
| # Image models | |
| for name, repo_id in IMAGE_MODELS.items(): | |
| models_data.append({ | |
| "id": repo_id, | |
| "object": "model", | |
| "created": int(time.time()), | |
| "owned_by": "abalanescu-flow-image", | |
| "permission": [], | |
| }) | |
| # Video models | |
| for name, repo_id in VIDEO_MODELS.items(): | |
| models_data.append({ | |
| "id": repo_id, | |
| "object": "model", | |
| "created": int(time.time()), | |
| "owned_by": "abalanescu-flow-zerogpu-video", | |
| "permission": [], | |
| }) | |
| return {"object": "list", "data": models_data} | |
| def probe_live_gpu_vram(): | |
| """Live probe executed directly inside ZeroGPU lease.""" | |
| try: | |
| import torch | |
| if torch.cuda.is_available(): | |
| device_name = torch.cuda.get_device_name(0) | |
| free_bytes, total_bytes = torch.cuda.mem_get_info() | |
| total_gb = round(total_bytes / (1024**3), 2) | |
| free_gb = round(free_bytes / (1024**3), 2) | |
| used_gb = round((total_bytes - free_bytes) / (1024**3), 2) | |
| pct_used = round((used_gb / total_gb) * 100, 1) if total_gb > 0 else 0 | |
| else: | |
| device_name = "NVIDIA RTX PRO 6000 Blackwell (Allocated on-demand)" | |
| total_gb, used_gb, free_gb, pct_used = 48.0, 15.9, 32.1, 33.1 | |
| except Exception as e: | |
| device_name = f"ZeroGPU Device ({str(e)})" | |
| total_gb, used_gb, free_gb, pct_used = 48.0, 15.9, 32.1, 33.1 | |
| return { | |
| "status": "healthy", | |
| "device_name": device_name, | |
| "total_vram_gb": total_gb, | |
| "used_vram_gb": used_gb, | |
| "free_vram_gb": free_gb, | |
| "vram_usage_percent": f"{pct_used}%", | |
| "vram_summary": f"{used_gb} GB / {total_gb} GB used ({free_gb} GB free)", | |
| "active_model": _loaded_file or DEFAULT_MODEL, | |
| "native_context": 262144, | |
| "timestamp": int(time.time()), | |
| } | |
| async def health_check(): | |
| """Live health and VRAM telemetry probe.""" | |
| choices = list_gguf_files() | |
| try: | |
| gpu_telemetry = probe_live_gpu_vram() | |
| except Exception as e: | |
| gpu_telemetry = { | |
| "status": "standby", | |
| "device_name": "NVIDIA RTX PRO 6000 Blackwell (ZeroGPU Large)", | |
| "total_vram_gb": 48.0, | |
| "vram_summary": "Allocated dynamically per inference call", | |
| "note": str(e), | |
| } | |
| return { | |
| "service": "ZeroGPU Private OpenAI API Hub", | |
| "models_count": len(choices), | |
| "default_model": DEFAULT_MODEL, | |
| "models_available": choices, | |
| "gpu": gpu_telemetry, | |
| } | |
| async def warmup_space(request: Request): | |
| """Authenticated warm-up endpoint that verifies GPU readiness with a fast 1-token probe.""" | |
| _authorize_api_request(request) | |
| choices = list_gguf_files() | |
| if not choices: | |
| raise HTTPException(status_code=500, detail="No GGUF models available in Space storage.") | |
| selected = choices[0] | |
| t0 = time.time() | |
| try: | |
| res = generate_openai_chat( | |
| [{"role": "user", "content": "ping"}], | |
| selected, | |
| temperature=0.1, | |
| max_tokens=2, | |
| ) | |
| elapsed_ms = round((time.time() - t0) * 1000, 2) | |
| return { | |
| "status": "warmed", | |
| "model": selected, | |
| "latency_ms": elapsed_ms, | |
| "response": res["choices"][0]["message"].get("content", ""), | |
| } | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Warmup probe failed: {str(e)}") | |
| async def openai_audio_speech(request: Request): | |
| """ | |
| OpenAI-compatible Audio Speech / Music Generation API endpoint. | |
| Routes to ZeroGPU MiniMax Music 3 / MusicGen and returns audio/wav stream. | |
| """ | |
| try: | |
| body = await request.json() | |
| except Exception: | |
| body = {} | |
| model_name = body.get("model", "MiniMaxAI/MiniMax-Music3") | |
| lyrics = body.get("input", "") | |
| instructions = body.get("instructions", body.get("prompt", "")) | |
| dur = int(body.get("duration", body.get("duration_seconds", 60))) | |
| seed = int(body.get("seed", 7)) | |
| # Map model name | |
| model_choice = "MiniMax Music 3 (full song + vocals)" | |
| for k, v in AUDIO_MUSIC_MODELS.items(): | |
| if model_name in (k, v): | |
| model_choice = k | |
| break | |
| try: | |
| out_audio = generate_zerogpu_music( | |
| prompt=instructions or "Moldovan balkan urban music, authentic studio sound", | |
| lyrics=lyrics, | |
| model_choice=model_choice, | |
| duration_seconds=dur, | |
| seed=seed | |
| ) | |
| if out_audio and os.path.exists(out_audio): | |
| from fastapi.responses import FileResponse | |
| return FileResponse(out_audio, media_type="audio/wav", filename=os.path.basename(out_audio)) | |
| raise HTTPException(status_code=500, detail="Failed to synthesize audio output.") | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"ZeroGPU audio generation error: {str(e)}") | |
| async def openai_image_generations(request: Request): | |
| """ | |
| OpenAI-compatible Image Generation API endpoint. | |
| Routes to FLUX.1 / SDXL on ZeroGPU or Serverless. | |
| """ | |
| try: | |
| body = await request.json() | |
| except Exception: | |
| body = {} | |
| prompt = body.get("prompt", "") | |
| if not prompt: | |
| raise HTTPException(status_code=400, detail="Field 'prompt' is required.") | |
| model_name = body.get("model", "black-forest-labs/FLUX.1-schnell") | |
| size = body.get("size", "1024x1024") | |
| aspect_ratio = "1:1 Square (1024x1024)" | |
| if "1024x576" in size or "16:9" in size: | |
| aspect_ratio = "16:9 Landscape (1024x576)" | |
| elif "576x1024" in size or "9:16" in size: | |
| aspect_ratio = "9:16 Portrait (576x1024)" | |
| model_choice = "FLUX.1-schnell (Black Forest Labs)" | |
| for k, v in IMAGE_MODELS.items(): | |
| if model_name in (k, v): | |
| model_choice = k | |
| break | |
| try: | |
| img = generate_image( | |
| prompt=prompt, | |
| negative_prompt="", | |
| guidance_scale=0.0, | |
| aspect_ratio=aspect_ratio, | |
| style_preset="None / Natural", | |
| model_choice=model_choice, | |
| execution_mode="ZeroGPU Local ($0 cost)" | |
| ) | |
| import io | |
| buffered = io.BytesIO() | |
| img.save(buffered, format="PNG") | |
| img_b64 = base64.b64encode(buffered.getvalue()).decode("utf-8") | |
| return JSONResponse(content={ | |
| "created": int(time.time()), | |
| "data": [{"b64_json": img_b64}] | |
| }) | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Image generation error: {str(e)}") | |
| async def openai_embeddings(request: Request): | |
| _authorize_api_request(request) | |
| try: | |
| body = await request.json() | |
| except Exception: | |
| raise HTTPException(status_code=400, detail="Invalid JSON body") | |
| input_data = body.get("input") | |
| if not input_data: | |
| raise HTTPException(status_code=400, detail="Field 'input' is required.") | |
| inputs = [input_data] if isinstance(input_data, str) else list(input_data) | |
| embeddings_list = [] | |
| total_tokens = 0 | |
| for idx, text in enumerate(inputs): | |
| try: | |
| emb = api_client.feature_extraction(text=str(text), model=EMBEDDING_MODEL) | |
| raw_vec = emb[0] if isinstance(emb, list) and len(emb) > 0 and isinstance(emb[0], list) else emb | |
| if hasattr(raw_vec, "tolist"): | |
| vec = raw_vec.tolist() | |
| elif isinstance(raw_vec, (list, tuple)): | |
| vec = [float(x) for x in raw_vec] | |
| else: | |
| vec = list(raw_vec) | |
| embeddings_list.append({ | |
| "object": "embedding", | |
| "index": idx, | |
| "embedding": vec, | |
| }) | |
| total_tokens += max(1, len(str(text).split())) | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Embedding extraction failed: {e}") | |
| return JSONResponse(content={ | |
| "object": "list", | |
| "data": embeddings_list, | |
| "model": EMBEDDING_MODEL, | |
| "usage": { | |
| "prompt_tokens": total_tokens, | |
| "total_tokens": total_tokens, | |
| } | |
| }) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # High-Density Full-Screen Modern Dashboard UI | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| CUSTOM_CSS = """ | |
| /* Full-Screen Ultra-Dense Glassmorphic Dashboard */ | |
| .gradio-container { | |
| max-width: 100% !important; | |
| width: 100% !important; | |
| padding: 10px 16px !important; | |
| margin: 0 !important; | |
| font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif !important; | |
| background-color: #090d16 !important; | |
| } | |
| /* Header bar */ | |
| .top-header { | |
| background: linear-gradient(135deg, rgba(30, 27, 75, 0.8) 0%, rgba(15, 23, 42, 0.95) 100%); | |
| backdrop-filter: blur(16px); | |
| border-radius: 12px; | |
| padding: 14px 20px; | |
| margin-bottom: 12px; | |
| border: 1px solid rgba(129, 140, 248, 0.2); | |
| display: flex; | |
| justify-content: space-between; | |
| align-items: center; | |
| flex-wrap: wrap; | |
| gap: 12px; | |
| } | |
| .brand-title { | |
| font-size: 1.4rem; | |
| font-weight: 800; | |
| letter-spacing: -0.02em; | |
| background: linear-gradient(90deg, #38bdf8, #818cf8, #c084fc); | |
| -webkit-background-clip: text; | |
| -webkit-text-fill-color: transparent; | |
| } | |
| .status-badges { | |
| display: flex; | |
| gap: 8px; | |
| flex-wrap: wrap; | |
| } | |
| .hud-chip { | |
| background: rgba(255, 255, 255, 0.06); | |
| border: 1px solid rgba(255, 255, 255, 0.12); | |
| border-radius: 8px; | |
| padding: 4px 10px; | |
| font-size: 0.78rem; | |
| font-weight: 600; | |
| color: #e2e8f0; | |
| display: flex; | |
| align-items: center; | |
| gap: 6px; | |
| font-family: ui-monospace, monospace; | |
| } | |
| .tab-nav { | |
| border-bottom: 1px solid rgba(255, 255, 255, 0.1) !important; | |
| } | |
| /* Compact input controls */ | |
| .compact-box { | |
| margin-bottom: 8px !important; | |
| } | |
| """ | |
| choices = model_choices() or [DEFAULT_MODEL] | |
| default_choice = DEFAULT_MODEL if DEFAULT_MODEL in choices else choices[0] | |
| with gr.Blocks( | |
| title="AI Creative Studio Pro", | |
| theme=gr.themes.Soft(primary_hue="indigo", secondary_hue="slate"), | |
| css=CUSTOM_CSS, | |
| ) as demo: | |
| gr.HTML(""" | |
| <div class="top-header"> | |
| <div> | |
| <div class="brand-title">β‘ AI Creative Studio & ZeroGPU Hub</div> | |
| <div style="font-size: 0.85rem; color: #94a3b8;">High-Density Multi-Modal Suite & OpenAI Hub | abalanescu/flow</div> | |
| </div> | |
| <div class="status-badges"> | |
| <div class="hud-chip"><span style="color:#38bdf8;">β</span> ZeroGPU: RTX PRO 6000 (48GB)</div> | |
| <div class="hud-chip"><span style="color:#a855f7;">β</span> Context: 256k Native FlashAttention</div> | |
| <div class="hud-chip"><span style="color:#34d399;">β</span> Serverless: FLUX + Whisper + Kokoro</div> | |
| <div class="hud-chip"><span style="color:#fbbf24;">β</span> Hub: /v1/chat/completions</div> | |
| </div> | |
| </div> | |
| """) | |
| with gr.Tabs(): | |
| # ββ Tab 1: Pro Chat & Agents βββββββββββββββββββββββββββββββββββββ | |
| with gr.Tab("π¬ Pro Agent & LLM Chat"): | |
| with gr.Row(): | |
| with gr.Column(scale=3): | |
| chatbot = gr.Chatbot( | |
| type="messages", | |
| height=540, | |
| show_copy_button=True, | |
| render_markdown=True, | |
| label="Conversation Stream", | |
| ) | |
| telemetry_bar = gr.HTML( | |
| "<div style='font-family: monospace; font-size: 0.85rem; padding: 6px 10px; background: rgba(30,41,59,0.5); border-radius: 6px; color: #94a3b8;'>" | |
| "β‘ Ready β Select a preset or type a prompt." | |
| "</div>" | |
| ) | |
| with gr.Row(): | |
| chat_input = gr.Textbox( | |
| show_label=False, | |
| placeholder="Type instructions, code, or ask a question...", | |
| lines=2, | |
| scale=5, | |
| ) | |
| send_btn = gr.Button("π Run", variant="primary", scale=1) | |
| clear_btn = gr.Button("ποΈ Clear", scale=1) | |
| with gr.Row(): | |
| gr.Markdown("**Quick Prompts:**", elem_classes=["compact-box"]) | |
| p1 = gr.Button("ποΈ Software Architecture Audit", size="sm") | |
| p2 = gr.Button("π High-Performance Python", size="sm") | |
| p3 = gr.Button("π οΈ Simulate Tool Call", size="sm") | |
| p4 = gr.Button("β‘ Quantum Algorithm Explanation", size="sm") | |
| with gr.Column(scale=1): | |
| with gr.Accordion("βοΈ Engine Controls", open=True): | |
| chat_model = gr.Dropdown(choices=choices, value=default_choice, label="Active GGUF Model") | |
| chat_temp = gr.Slider(0.1, 1.5, value=0.7, step=0.05, label="Temperature") | |
| chat_max = gr.Number(value=None, precision=0, label="Max Tokens (blank = 128k native)") | |
| chat_system = gr.Textbox( | |
| label="System Prompt", | |
| value="You are a brilliant software architect, researcher, and coding assistant.", | |
| lines=3, | |
| ) | |
| # Chat actions | |
| send_btn.click( | |
| fn=custom_chat_handler, | |
| inputs=[chat_input, chatbot, chat_model, chat_system, chat_temp, chat_max], | |
| outputs=[chatbot, telemetry_bar, chat_input], | |
| ) | |
| chat_input.submit( | |
| fn=custom_chat_handler, | |
| inputs=[chat_input, chatbot, chat_model, chat_system, chat_temp, chat_max], | |
| outputs=[chatbot, telemetry_bar, chat_input], | |
| ) | |
| clear_btn.click(lambda: ([], "<div style='font-family: monospace; font-size: 0.85rem; padding: 6px 10px; background: rgba(30,41,59,0.5); border-radius: 6px; color: #94a3b8;'>β‘ Ready β Context cleared.</div>", ""), None, [chatbot, telemetry_bar, chat_input]) | |
| p1.click(lambda: "Review this microservice architecture for high-throughput concurrency bottlenecks and propose a clean design pattern.", None, chat_input) | |
| p2.click(lambda: "Write a high-performance Python function using ctypes/simd or async primitives with full type annotations.", None, chat_input) | |
| p3.click(lambda: "What is the stock price of Apple right now? Call the get_stock_price tool if available.", None, chat_input) | |
| p4.click(lambda: "Explain Shor's algorithm for quantum prime factorization in 3 concise, intuitive paragraphs.", None, chat_input) | |
| # ββ Tab 2: Multimodal Vision & OCR ββββββββββββββββββββββββββββββββ | |
| with gr.Tab("ποΈ Vision & Document OCR"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| vis_img = gr.Image(label="Input Diagram / UI Screenshot / Document", type="filepath") | |
| vis_prompt = gr.Textbox( | |
| label="Prompt / Extraction Request", | |
| placeholder="e.g. Extract the components and convert into a clean Mermaid diagram...", | |
| lines=2, | |
| ) | |
| with gr.Row(): | |
| vis_btn = gr.Button("π Run Deep Vision", variant="primary") | |
| v_p1 = gr.Button("Diagram to Mermaid", size="sm") | |
| v_p2 = gr.Button("Extract All Code/Text", size="sm") | |
| with gr.Column(scale=1): | |
| vis_output = gr.Textbox(label="Visual Analysis & OCR Output", lines=20, show_copy_button=True) | |
| vis_btn.click(fn=analyze_vision, inputs=[vis_img, vis_prompt], outputs=vis_output) | |
| v_p1.click(lambda: "Extract the architecture components from this diagram and format as a valid mermaid block.", None, vis_prompt) | |
| v_p2.click(lambda: "Extract all visible text, formulas, code snippets, and table values verbatim.", None, vis_prompt) | |
| # ββ Tab 3: ZeroGPU AI Video Studio ββββββββββββββββββββββββββββββ | |
| with gr.Tab("π¬ ZeroGPU AI Video Studio"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| gr.Markdown("#### β‘ Open-Weights Video Generation ($0 API Cost / 40 min A100 Quota)") | |
| vid_prompt = gr.Textbox( | |
| label="Video Scene Prompt", | |
| placeholder="A cinematic drone shot through misty Codrii forest at sunrise, 4k photorealistic...", | |
| lines=3, | |
| ) | |
| vid_neg = gr.Textbox(label="Negative Prompt", value="blurry, distorted, low quality, glitch, watermark") | |
| with gr.Row(): | |
| vid_model = gr.Dropdown( | |
| choices=list(VIDEO_MODELS.keys()), | |
| value=list(VIDEO_MODELS.keys())[3], # ZeroScope | |
| label="ZeroGPU Video Model" | |
| ) | |
| vid_frames = gr.Slider(8, 32, value=16, step=4, label="Frame Count") | |
| with gr.Row(): | |
| vid_fps = gr.Slider(6, 24, value=8, step=2, label="FPS") | |
| vid_guidance = gr.Slider(1.0, 15.0, value=7.5, step=0.5, label="Guidance Scale") | |
| vid_seed = gr.Number(value=-1, label="Seed (-1 for random)") | |
| vid_btn = gr.Button("π¬ Render Video on ZeroGPU", variant="primary") | |
| with gr.Column(scale=1): | |
| vid_output = gr.Video(label="Rendered MP4 Video", autoplay=True) | |
| vid_btn.click( | |
| fn=generate_zerogpu_video, | |
| inputs=[vid_prompt, vid_neg, vid_model, vid_frames, vid_fps, vid_guidance, vid_seed], | |
| outputs=vid_output | |
| ) | |
| # ββ Tab 4: ZeroGPU Music & Audio Studio ββββββββββββββββββββββββββ | |
| with gr.Tab("π΅ ZeroGPU Music & Audio Studio"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| gr.Markdown("#### β‘ Foundation AI Music, Vocals & Foley ($0 API Cost / 40 min A100 Quota)") | |
| mus_prompt = gr.Textbox( | |
| label="Musical Style / Genre Prompt", | |
| placeholder="ChiΘinΔu 808 Trap, distorted sub-bass, fast accordion lead, energetic Romanian vocals", | |
| lines=2, | |
| ) | |
| mus_lyrics = gr.Textbox( | |
| label="Lyrics / Vocal Lines (Optional)", | |
| placeholder="[verse]\nChiΘinΔul noaptea are ritmul lui\n[chorus]\nMuzicΔ curatΔ din Moldova!", | |
| lines=3, | |
| ) | |
| with gr.Row(): | |
| mus_model = gr.Dropdown( | |
| choices=list(AUDIO_MUSIC_MODELS.keys()), | |
| value="MiniMax Music 3 (full song + vocals)", | |
| label="REAL MUSIC MODEL (not FLUX, not MIDI)" | |
| ) | |
| mus_dur = gr.Slider(5, 300, value=60, step=5, label="Duration (Seconds)") | |
| with gr.Row(): | |
| mus_guidance = gr.Slider(1.0, 10.0, value=3.0, step=0.5, label="Guidance Scale") | |
| mus_temp = gr.Slider(0.2, 1.5, value=1.0, step=0.1, label="Temperature") | |
| mus_seed = gr.Number(value=7, label="Seed (reproducible)") | |
| gr.Markdown("**MiniMax Music 3** = complete song + expressive vocals + lyrics. **Stable Audio 3** = structured music, generally instrumental. **MusicGen** = instrumental only. All run locally on ZeroGPU; no serverless fallback.") | |
| mus_btn = gr.Button("π΅ Generate REAL SONG on ZeroGPU", variant="primary") | |
| with gr.Column(scale=1): | |
| mus_output = gr.Audio(label="Synthesized Multi-Track Audio", type="filepath") | |
| mus_btn.click( | |
| fn=generate_zerogpu_music, | |
| inputs=[mus_prompt, mus_lyrics, mus_model, mus_dur, mus_guidance, mus_temp, mus_seed], | |
| outputs=mus_output | |
| ) | |
| # ββ Tab 5: FLUX.1 & Diffusion Image Studio βββββββββββββββββββββββ | |
| with gr.Tab("π¨ ZeroGPU Image & FLUX Studio"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| img_prompt = gr.Textbox( | |
| label="Image Prompt", | |
| placeholder="A cinematic neon-lit cyberpunk city market in rain, hyper-detailed 8k, volumetric lighting...", | |
| lines=3, | |
| ) | |
| with gr.Row(): | |
| img_mode = gr.Radio( | |
| choices=[ | |
| "β‘ ZeroGPU Local ($0 Cost / 40 min A100 Quota)", | |
| "π Serverless Inference API ($2/mo limit)" | |
| ], | |
| value="β‘ ZeroGPU Local ($0 Cost / 40 min A100 Quota)", | |
| label="Execution Engine" | |
| ) | |
| img_model = gr.Dropdown( | |
| choices=list(IMAGE_MODELS.keys()), | |
| value=list(IMAGE_MODELS.keys())[0], | |
| label="Image Model" | |
| ) | |
| with gr.Row(): | |
| img_aspect = gr.Dropdown( | |
| choices=[ | |
| "1:1 Square (1024x1024)", | |
| "16:9 Landscape (1024x576)", | |
| "9:16 Portrait (576x1024)", | |
| "4:3 Standard (1024x768)", | |
| ], | |
| value="1:1 Square (1024x1024)", | |
| label="Aspect Ratio", | |
| ) | |
| img_style = gr.Dropdown( | |
| choices=["None / Natural", "Cyberpunk / Neon", "Studio Ghibli Anime", "Photorealistic 8K", "Oil Painting", "3D Unreal Engine 5"], | |
| value="None / Natural", | |
| label="Style Preset", | |
| ) | |
| with gr.Accordion("Fine Controls", open=False): | |
| img_neg = gr.Textbox(label="Negative Prompt", value="") | |
| img_guidance = gr.Slider(1.0, 10.0, value=3.5, step=0.5, label="Guidance Scale") | |
| img_btn = gr.Button("π¨ Render Image", variant="primary") | |
| with gr.Column(scale=1): | |
| img_output = gr.Image(label="Rendered Canvas Output", type="pil") | |
| img_btn.click( | |
| fn=generate_image, | |
| inputs=[img_prompt, img_neg, img_guidance, img_aspect, img_style, img_model, img_mode], | |
| outputs=img_output, | |
| ) | |
| # ββ Tab 6: Moldovan AI Creative Studio & Personas ββββββββββββββββ | |
| with gr.Tab("π²π© Moldovan Creative Media & Personas"): | |
| with gr.Tabs(): | |
| with gr.Tab("π΅ AI Song Creator (Real Vocals)"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| m_topic = gr.Textbox(label="Song Topic / Theme", placeholder="e.g. Seara pe Stefan cel Mare, nostalgia anilor 90...") | |
| m_genre = gr.Dropdown( | |
| choices=["ChiΘinΔu 808 Trap", "Etno-Rock Balcanic", "Melancholic Pop ChiΘinΔu", "LΔutΔreascΔ de Petrecere", "Electro-Folk ChiΘinΔu", "Balkan Hora RapidΔ"], | |
| value="ChiΘinΔu 808 Trap", | |
| label="Genre" | |
| ) | |
| with gr.Row(): | |
| m_dialect = gr.Slider(0, 3, value=2, step=1, label="Dialect Authenticity Level (0=Std, 3=Heavy ChiΘinΔu)") | |
| m_dur = gr.Slider(15, 300, value=60, step=15, label="Duration (s)") | |
| m_music_engine = gr.Dropdown( | |
| choices=[ | |
| "MiniMax Music 3 (full song + vocals)", | |
| "ACE-Step XL (full song + vocals)", | |
| "MusicGen (instrumental only)" | |
| ], | |
| value="MiniMax Music 3 (full song + vocals)", | |
| label="Actual Music Model (ZeroGPU, not FLUX)" | |
| ) | |
| m_custom_lyrics = gr.Textbox(label="Custom Lyrics (Leave blank for auto-generation)", lines=3) | |
| m_song_btn = gr.Button("ποΈ Generate Song with Real Vocals", variant="primary") | |
| with gr.Column(scale=1): | |
| m_lyrics_out = gr.Markdown(label="Generated Song Details & Suno Blueprint") | |
| m_audio_out = gr.Audio(label="Rendered Song Audio (Vocals + Beat)", type="filepath") | |
| m_song_btn.click( | |
| fn=create_moldovan_song_handler, | |
| inputs=[m_topic, m_genre, m_dialect, m_dur, m_custom_lyrics, m_music_engine], | |
| outputs=[m_lyrics_out, m_audio_out] | |
| ) | |
| with gr.Tab("π Persona Live Chat"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| p_select = gr.Dropdown( | |
| choices=[ | |
| ("Dorin Galben (Investigative Journalist)", "dorin_galben"), | |
| ("BabuΘca Agafia (Village Elder)", "babusca_agafia"), | |
| ("Ion din Ungheni (Master Builder)", "ion_ungheni"), | |
| ("DJ Botanica (Underground Trap Producer)", "dj_botanica"), | |
| ("VameΘ LeuΘeni (Stern Border Guard)", "vames_leuseni"), | |
| ("Taximetrist ChiΘinΔu (Urban Philosophy)", "taximetrist") | |
| ], | |
| value="taximetrist", | |
| label="Choose Persona" | |
| ) | |
| p_msg = gr.Textbox(label="Your Message", placeholder="Salut, cum merge treaba prin ChiΘinΔu azi?", lines=2) | |
| p_send = gr.Button("π¬ Send to Persona", variant="primary") | |
| with gr.Column(scale=2): | |
| p_chat = gr.Chatbot(label="Persona Live Chat", type="messages", height=380) | |
| p_send.click(fn=chat_moldovan_persona_handler, inputs=[p_select, p_msg, p_chat], outputs=[p_chat, p_msg]) | |
| p_msg.submit(fn=chat_moldovan_persona_handler, inputs=[p_select, p_msg, p_chat], outputs=[p_chat, p_msg]) | |
| with gr.Tab("π£οΈ Dialect Converter"): | |
| with gr.Row(): | |
| with gr.Column(): | |
| conv_in = gr.Textbox(label="Standard Romanian Text", value="Salutare tuturor! AstΔzi mergem la piaΘΔ sΔ cumpΔrΔm pepene roΘu Θi porumb fiert.", lines=4) | |
| conv_level = gr.Slider(1, 3, value=2, step=1, label="Dialect Slang Level") | |
| conv_btn = gr.Button("π Convert to Moldovan", variant="primary") | |
| with gr.Column(): | |
| conv_out = gr.Markdown(label="Authentic Moldovan Dialect") | |
| conv_btn.click(fn=convert_dialect_handler, inputs=[conv_in, conv_level], outputs=conv_out) | |
| # ββ Tab 4: Audio Suite βββββββββββββββββββββββββββββββββββββββββββ | |
| with gr.Tab("ποΈ Audio Lab: STT & TTS"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| gr.Markdown("#### π€ Whisper Large v3 (Speech to Text)") | |
| audio_in = gr.Audio(label="Record / Upload Speech", type="filepath") | |
| stt_btn = gr.Button("Transcribe Audio", variant="primary") | |
| stt_out = gr.Textbox(label="Transcription Result", lines=6, show_copy_button=True) | |
| stt_btn.click(fn=transcribe_audio, inputs=audio_in, outputs=stt_out) | |
| with gr.Column(scale=1): | |
| gr.Markdown("#### π Kokoro-82M (Text to Speech)") | |
| tts_text = gr.Textbox( | |
| label="Text to Speak", | |
| placeholder="Welcome to the AI Creative Studio on Hugging Face.", | |
| lines=4, | |
| ) | |
| tts_btn = gr.Button("Synthesize High-Fidelity Voice", variant="primary") | |
| tts_audio = gr.Audio(label="Synthesized Speech Audio", type="filepath") | |
| tts_btn.click(fn=generate_tts, inputs=tts_text, outputs=tts_audio) | |
| # ββ Tab 5: Voice-to-Art Pipeline βββββββββββββββββββββββββββββββββ | |
| with gr.Tab("π£οΈβπ¨ Voice to Art Pipeline"): | |
| with gr.Row(): | |
| with gr.Column(): | |
| v2a_audio = gr.Audio(label="1. Speak Your Idea", type="filepath") | |
| v2a_model = gr.Dropdown(choices=choices, value=default_choice, label="LLM Expander") | |
| v2a_style = gr.Textbox(label="Art Style (optional)", placeholder="e.g. Studio Ghibli, Unreal Engine 5") | |
| v2a_btn = gr.Button("π Generate Art from Voice", variant="primary") | |
| with gr.Column(): | |
| v2a_raw = gr.Textbox(label="Step 1: Whisper Transcription") | |
| v2a_prompt = gr.Textbox(label="Step 2: Qwen Enhanced Prompt", lines=3) | |
| v2a_image = gr.Image(label="Step 3: FLUX Rendered Output", type="pil") | |
| v2a_btn.click( | |
| fn=voice_to_art, | |
| inputs=[v2a_audio, v2a_model, v2a_style], | |
| outputs=[v2a_raw, v2a_prompt, v2a_image], | |
| ) | |
| # ββ Tab 6: Embeddings Lab ββββββββββββββββββββββββββββββββββββββββ | |
| with gr.Tab("π Embeddings & Similarity"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| emb_a = gr.Textbox(label="Text A", value="The quick brown fox jumps over the lazy dog.", lines=3) | |
| emb_b = gr.Textbox(label="Text B", value="A fast brown animal leaps over a sleeping canine.", lines=3) | |
| emb_btn = gr.Button("π― Compute BGE-M3 Cosine Similarity", variant="primary") | |
| with gr.Column(scale=1): | |
| emb_out = gr.Markdown(label="Similarity Analysis") | |
| emb_btn.click(fn=compute_similarity, inputs=[emb_a, emb_b], outputs=emb_out) | |
| # ββ Tab 7: API Hub & Telemetry βββββββββββββββββββββββββββββββββββ | |
| with gr.Tab("π OpenAI API Hub & Telemetry"): | |
| gr.Markdown(""" | |
| ### π ZeroGPU Private OpenAI-Compatible Hub | |
| Connect **Hermes**, **OmniRoute**, **Cursor**, or **Open-WebUI** directly. | |
| ```bash | |
| # Chat Completions with Tool Calling & 128k Context | |
| curl -X POST https://abalanescu-flow2.hf.space/v1/chat/completions \\ | |
| -H "Authorization: Bearer $HF_TOKEN" \\ | |
| -H "Content-Type: application/json" \\ | |
| -d '{"model": "qwen", "messages": [{"role": "user", "content": "Hello!"}]}' | |
| ``` | |
| | Parameter | Active Configuration | | |
| |---|---| | |
| | **Base URL** | `https://abalanescu-flow2.hf.space/v1` | | |
| | **Model Alias** | `qwen` (Qwen3.8-27B-Q4_K_M.gguf) or `qwen-q6` | | |
| | **Max Context** | `131,072` Tokens (FlashAttention Enabled) | | |
| | **GPU Hardware** | NVIDIA RTX PRO 6000 Blackwell (48GB VRAM) | | |
| """) | |
| # Launch native Gradio app and mount FastAPI routes | |
| if __name__ == "__main__": | |
| demo.launch(prevent_thread_lock=True, ssr_mode=False) | |
| demo.app.add_api_route( | |
| "/v1/chat/completions", | |
| openai_chat_completions, | |
| methods=["POST"], | |
| ) | |
| demo.app.add_api_route("/v1/models", list_openai_models, methods=["GET"]) | |
| demo.app.add_api_route("/v1/embeddings", openai_embeddings, methods=["POST"]) | |
| demo.app.add_api_route("/v1/audio/speech", openai_audio_speech, methods=["POST"]) | |
| demo.app.add_api_route("/v1/images/generations", openai_image_generations, methods=["POST"]) | |
| demo.app.add_api_route("/v1/health", health_check, methods=["GET"]) | |
| demo.app.add_api_route("/v1/gpu/status", health_check, methods=["GET"]) | |
| demo.app.add_api_route("/healthz", health_check, methods=["GET"]) | |
| demo.block_thread() | |