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"\n{content or ''}\n"
})
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\n\n"
if isinstance(args_dict, dict):
for k, v in args_dict.items():
tc_text += f"\n{json.dumps(v) if isinstance(v, (dict, list)) else v}\n\n"
tc_text += "\n"
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)
# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
@spaces.GPU(size="large", duration=120)
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)
@spaces.GPU(size="large", duration=120)
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"π§ Deep Thinking & Reasoning
\n\n```markdown\n{reasoning}\n```\nπ οΈ Executed Tool Calls ({len(tool_calls)})
\n\n```json\n{json.dumps(tool_calls, indent=2)}\n```\n"
f"β‘ {tps:.1f} t/s"
f"β±οΈ {elapsed:.2f}s"
f"π₯ Prompt: {prompt_toks}"
f"π€ Output: {comp_toks}"
f"π§ Context: {tot_toks:,} / 262,144 ({ctx_pct:.1f}%)"
f"β ZeroGPU Large (48GB)"
f"
"
)
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)
# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
@spaces.GPU(duration=120)
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)
# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
@spaces.GPU(duration=300)
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)
# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
@spaces.GPU(duration=60)
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
# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
@spaces.GPU(duration=300)
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
# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
@spaces.GPU(size="large", duration=120)
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"\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.")
@fastapi_app.post("/v1/chat/completions")
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
}
@fastapi_app.get("/v1/models")
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}
@spaces.GPU(size="large", duration=60)
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()),
}
@fastapi_app.get("/v1/gpu/status")
@fastapi_app.get("/v1/health")
@fastapi_app.get("/healthz")
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,
}
@fastapi_app.post("/v1/warmup")
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)}")
@fastapi_app.post("/v1/audio/speech")
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)}")
@fastapi_app.post("/v1/images/generations")
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)}")
@fastapi_app.post("/v1/embeddings")
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("""
""")
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(
""
"β‘ Ready β Select a preset or type a prompt."
"
"
)
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: ([], "β‘ Ready β Context cleared.
", ""), 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()