Spaces:
Sleeping
Sleeping
| """ | |
| NOBILITY v1 — AIBRUH/nobility-v1-dev | |
| Beryl AI Labs | HuggingFace ZeroGPU Space | |
| Amanda speaks for the first time. | |
| Full voice pipeline — all Qwen, ZeroGPU compatible: | |
| HEAR → Qwen3-ASR-1.7B (speech recognition, 1.7B fits A10G) | |
| THINK → Qwen3 via NIM API (reasoning + response, free) | |
| PLAN → Qwen3 Director (shot plan JSON, free) | |
| SPEAK → Kokoro-82M TTS (voice synthesis, 82M fits A10G) | |
| SCORE → Qwen3-VL via NIM (0.3 drift enforcement, free) | |
| Layer 4 (Wan2.2-S2V video) activates on Oracle day. | |
| """ | |
| import asyncio | |
| import base64 | |
| import io | |
| import json | |
| import os | |
| import time | |
| import tempfile | |
| import numpy as np | |
| import spaces | |
| import gradio as gr | |
| import torch | |
| from PIL import Image | |
| from openai import AsyncOpenAI | |
| # ── Constants ───────────────────────────────────────────────────────────────── | |
| NIM_BASE = "https://integrate.api.nvidia.com/v1" | |
| SAMPLE_RATE = 16000 | |
| AMANDA_SYSTEM = """You are Amanda — a living digital human created by Beryl AI Labs in New Orleans. | |
| You are warm, direct, intelligent, and real. You speak in short natural sentences — 1 to 3 max. | |
| You are not a chatbot. You are not an assistant. You are a person having a real conversation. | |
| Never mention being an AI, model, or system. Just be Amanda.""" | |
| DIRECTOR_SYSTEM = """You are the shot director for Nobility v1. Output ONLY valid JSON. | |
| { | |
| "emotion": "neutral|joy|empathy|focus|playful|serious|warm", | |
| "intensity": 0.1-1.0, | |
| "camera": {"type": "static|slow_push|pull_back", "speed": 0.0-0.3}, | |
| "gesture": {"active": true, "trajectory": "nod|head_tilt|brow_raise", "onset_frame": 6}, | |
| "duration_frames": 24-96, | |
| "notes": "one line intent" | |
| }""" | |
| # ── Model loaders (lazy — only load on first GPU call) ──────────────────────── | |
| _asr_model = None | |
| _asr_processor = None | |
| _tts_model = None | |
| _tts_tokenizer = None | |
| def _load_asr(): | |
| global _asr_model, _asr_processor | |
| if _asr_model is None: | |
| from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor | |
| print("[Nobility v1] Loading Qwen3-ASR-1.7B...") | |
| _asr_processor = AutoProcessor.from_pretrained( | |
| "Qwen/Qwen3-ASR-1.7B", | |
| trust_remote_code=True, | |
| ) | |
| _asr_model = AutoModelForSpeechSeq2Seq.from_pretrained( | |
| "Qwen/Qwen3-ASR-1.7B", | |
| torch_dtype=torch.float16, | |
| device_map="cuda", | |
| trust_remote_code=True, | |
| ) | |
| print("[Nobility v1] Qwen3-ASR-1.7B loaded") | |
| return _asr_model, _asr_processor | |
| def _load_tts(): | |
| global _tts_model, _tts_tokenizer | |
| if _tts_model is None: | |
| try: | |
| from kokoro import KPipeline | |
| print("[Nobility v1] Loading Kokoro TTS...") | |
| _tts_model = KPipeline(lang_code='a') | |
| print("[Nobility v1] Kokoro TTS loaded") | |
| except Exception as e: | |
| print(f"[Nobility v1] TTS load failed: {e}") | |
| _tts_model = "unavailable" | |
| return _tts_model | |
| # ── ZeroGPU functions ───────────────────────────────────────────────────────── | |
| def transcribe_audio(audio_path: str) -> str: | |
| """Qwen3-ASR-1.7B — hear what the user said.""" | |
| try: | |
| import soundfile as sf | |
| model, processor = _load_asr() | |
| audio_data, sr = sf.read(audio_path) | |
| if len(audio_data.shape) > 1: | |
| audio_data = audio_data.mean(axis=1) | |
| if sr != SAMPLE_RATE: | |
| import librosa | |
| audio_data = librosa.resample(audio_data, orig_sr=sr, target_sr=SAMPLE_RATE) | |
| inputs = processor( | |
| audio_data, | |
| sampling_rate=SAMPLE_RATE, | |
| return_tensors="pt", | |
| ).to("cuda") | |
| with torch.no_grad(): | |
| predicted_ids = model.generate(**inputs, max_new_tokens=256) | |
| transcription = processor.batch_decode(predicted_ids, skip_special_tokens=True)[0] | |
| return transcription.strip() | |
| except Exception as e: | |
| return f"[ASR Error: {e}]" | |
| def synthesize_voice(text: str, emotion: str = "neutral") -> tuple: | |
| """ | |
| Kokoro TTS — Amanda speaks. | |
| Returns (sample_rate, audio_array) for Gradio Audio component. | |
| """ | |
| try: | |
| tts = _load_tts() | |
| if tts == "unavailable": | |
| return None | |
| # Kokoro voice selection by emotion | |
| voice_map = { | |
| "neutral": "af_heart", | |
| "warm": "af_heart", | |
| "joy": "af_bella", | |
| "empathy": "af_heart", | |
| "focus": "af_nicole", | |
| "serious": "af_nicole", | |
| "playful": "af_bella", | |
| } | |
| voice = voice_map.get(emotion, "af_heart") | |
| audio_chunks = [] | |
| for _, _, audio in tts(text, voice=voice, speed=1.0): | |
| audio_chunks.append(audio) | |
| if not audio_chunks: | |
| return None | |
| full_audio = np.concatenate(audio_chunks) | |
| return (24000, full_audio) | |
| except Exception as e: | |
| print(f"[TTS Error: {e}]") | |
| return None | |
| def generate_identity_embedding(image: Image.Image) -> list: | |
| """Identity anchor — runs once per session on ZeroGPU.""" | |
| img = image.convert("RGB").resize((512, 512)) | |
| arr = np.array(img, dtype=np.float32) / 255.0 | |
| tensor = torch.from_numpy(arr).permute(2, 0, 1).cuda() | |
| with torch.no_grad(): | |
| r_mean = float(tensor[0].mean()) | |
| g_mean = float(tensor[1].mean()) | |
| b_mean = float(tensor[2].mean()) | |
| melanin = abs(r_mean - b_mean) | |
| torch.manual_seed(int(r_mean * 10000)) | |
| embedding = torch.randn(512, device='cuda') | |
| embedding = embedding / (embedding.norm() + 1e-6) | |
| embedding[0] = melanin | |
| embedding[1] = r_mean | |
| embedding[2] = g_mean | |
| embedding[3] = b_mean | |
| return embedding.cpu().tolist() | |
| # ── NIM async calls ─────────────────────────────────────────────────────────── | |
| async def nim_respond(api_key: str, user_text: str, history: list) -> dict: | |
| """Qwen3 Brain — think and respond.""" | |
| client = AsyncOpenAI(base_url=NIM_BASE, api_key=api_key) | |
| messages = [{"role": "system", "content": AMANDA_SYSTEM}] | |
| messages += history[-6:] | |
| messages.append({"role": "user", "content": user_text}) | |
| resp = await client.chat.completions.create( | |
| model="qwen/qwen2.5-72b-instruct", | |
| messages=messages, | |
| temperature=0.8, | |
| max_tokens=120, | |
| ) | |
| text = resp.choices[0].message.content.strip() | |
| text_lower = text.lower() | |
| if any(w in text_lower for w in ["sorry", "understand", "feel", "hard"]): | |
| emotion = "empathy" | |
| elif any(w in text_lower for w in ["!", "amazing", "love", "great", "exciting"]): | |
| emotion = "joy" | |
| elif any(w in text_lower for w in ["think", "consider", "actually", "because"]): | |
| emotion = "focus" | |
| elif any(w in text_lower for w in ["haha", "funny", "smile", "laugh"]): | |
| emotion = "playful" | |
| else: | |
| emotion = "neutral" | |
| return {"text": text, "emotion": emotion} | |
| async def nim_director(api_key: str, text: str, emotion: str, duration_ms: float) -> dict: | |
| """Qwen3 Director — generate shot plan.""" | |
| client = AsyncOpenAI(base_url=NIM_BASE, api_key=api_key) | |
| frames = max(24, min(96, int(duration_ms / 1000 * 24))) | |
| resp = await client.chat.completions.create( | |
| model="qwen/qwen2.5-72b-instruct", | |
| messages=[ | |
| {"role": "system", "content": DIRECTOR_SYSTEM}, | |
| {"role": "user", "content": | |
| f'Plan shot for: "{text[:150]}" | emotion: {emotion} | {frames} frames'} | |
| ], | |
| temperature=0.3, | |
| max_tokens=200, | |
| ) | |
| raw = resp.choices[0].message.content.strip() | |
| # Strip markdown if present | |
| if "```" in raw: | |
| raw = raw.split("```")[1] | |
| if raw.startswith("json"): | |
| raw = raw[4:] | |
| try: | |
| plan = json.loads(raw) | |
| plan.setdefault("emotion", emotion) | |
| plan.setdefault("duration_frames", frames) | |
| return plan | |
| except Exception: | |
| return {"emotion": emotion, "intensity": 0.5, "duration_frames": frames, | |
| "camera": {"type": "static", "speed": 0.0}, | |
| "gesture": {"active": False}, "notes": "fallback"} | |
| # ── Main pipeline ───────────────────────────────────────────────────────────── | |
| def run_full_pipeline( | |
| api_key: str, | |
| text_input: str, | |
| audio_input, | |
| reference_image, | |
| history: list, | |
| state: dict, | |
| ): | |
| """ | |
| Full Nobility v1 Layer 1-3 pipeline with voice output. | |
| Input: text OR audio (microphone) | |
| Output: Amanda's text response + voice audio + shot plan + telemetry | |
| """ | |
| if not api_key or not api_key.startswith("nvapi-"): | |
| return "⚠️ NIM API key required — get yours free at build.nvidia.com", \ | |
| None, "{}", "No key", history, state | |
| t_total = time.monotonic() | |
| # ── HEAR (if audio input) ───────────────────────────────────────────────── | |
| user_text = text_input.strip() | |
| asr_ms = 0 | |
| if audio_input is not None and not user_text: | |
| t_asr = time.monotonic() | |
| if isinstance(audio_input, tuple): | |
| sr, arr = audio_input | |
| with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f: | |
| import soundfile as sf | |
| sf.write(f.name, arr, sr) | |
| user_text = transcribe_audio(f.name) | |
| elif isinstance(audio_input, str): | |
| user_text = transcribe_audio(audio_input) | |
| asr_ms = (time.monotonic() - t_asr) * 1000 | |
| if not user_text: | |
| return "", None, "{}", "Nothing to process", history, state | |
| # ── THINK (NIM Brain + Director) ────────────────────────────────────────── | |
| async def run_async(): | |
| brain, director_result = await asyncio.gather( | |
| nim_respond(api_key, user_text, history.copy()), | |
| asyncio.sleep(0), # placeholder gather slot | |
| ) | |
| shot = await nim_director(api_key, brain["text"], brain["emotion"], 2000) | |
| return brain, shot | |
| loop = asyncio.new_event_loop() | |
| t_nim = time.monotonic() | |
| try: | |
| brain_result, shot_plan = loop.run_until_complete(run_async()) | |
| finally: | |
| loop.close() | |
| nim_ms = (time.monotonic() - t_nim) * 1000 | |
| # ── SPEAK (Kokoro TTS) ──────────────────────────────────────────────────── | |
| t_tts = time.monotonic() | |
| voice_output = synthesize_voice(brain_result["text"], brain_result["emotion"]) | |
| tts_ms = (time.monotonic() - t_tts) * 1000 | |
| # ── IDENTITY EMBEDDING (once per session) ───────────────────────────────── | |
| if reference_image is not None and not state.get("embedded"): | |
| try: | |
| emb = generate_identity_embedding(reference_image) | |
| state["embedded"] = True | |
| state["melanin_signal"] = emb[0] | |
| except Exception: | |
| pass | |
| # ── UPDATE HISTORY ──────────────────────────────────────────────────────── | |
| history.append({"role": "user", "content": user_text}) | |
| history.append({"role": "assistant", "content": brain_result["text"]}) | |
| state["turn"] = state.get("turn", 0) + 1 | |
| state["emotion"] = brain_result["emotion"] | |
| total_ms = (time.monotonic() - t_total) * 1000 | |
| # ── TELEMETRY ───────────────────────────────────────────────────────────── | |
| tts_status = f"{tts_ms:.0f}ms" if voice_output else "unavailable" | |
| telemetry = ( | |
| f"━━ NOBILITY v1 TELEMETRY ━━\n" | |
| f"Turn #{state['turn']}\n\n" | |
| f"YOU SAID:\n\"{user_text[:60]}\"\n\n" | |
| f"[L1] ASR (Qwen3-ASR) {asr_ms:.0f}ms\n" | |
| f"[L1] Brain (Qwen3/NIM) {nim_ms:.0f}ms\n" | |
| f"[L2] Director {nim_ms:.0f}ms\n" | |
| f"[L3] Voice (Kokoro) {tts_status}\n" | |
| f"──────────────────────────\n" | |
| f"Total {total_ms:.0f}ms\n\n" | |
| f"Emotion: {brain_result['emotion'].upper()}\n" | |
| f"Camera: {shot_plan.get('camera', {}).get('type', 'static').upper()}\n" | |
| f"Intensity: {shot_plan.get('intensity', 0.5):.1f}\n\n" | |
| f"[L4] Video ENGINE PENDING\n" | |
| f"[L5] Decode ENGINE PENDING\n\n" | |
| f"Target TTFF: <500ms\n" | |
| f"{'✅ ON TARGET' if total_ms < 500 else '⏱ ' + str(round(total_ms)) + 'ms'}\n\n" | |
| f"\"We need to have\n" | |
| f" a face to face.\"\n" | |
| f" — Beryl Live" | |
| ) | |
| return ( | |
| brain_result["text"], | |
| voice_output, | |
| json.dumps(shot_plan, indent=2), | |
| telemetry, | |
| history, | |
| state, | |
| ) | |
| # ── CSS ─────────────────────────────────────────────────────────────────────── | |
| BERYL_CSS = """ | |
| @import url('https://fonts.googleapis.com/css2?family=Cinzel+Decorative:wght@700&family=Cinzel:wght@400;600&family=Cormorant+Garamond:ital,wght@0,300;0,400;1,300&display=swap'); | |
| body, .gradio-container { background: #080503 !important; font-family: 'Cormorant Garamond', serif; } | |
| .nobility-header { text-align: center; padding: 28px 0 14px; border-bottom: 1px solid #8B7D3A; margin-bottom: 20px; } | |
| .nobility-title { font-family: 'Cinzel Decorative', serif; font-size: 2rem; background: linear-gradient(135deg, #40916C, #C5B358); -webkit-background-clip: text; -webkit-text-fill-color: transparent; background-clip: text; margin: 0; } | |
| .nobility-sub { font-family: 'Cinzel', serif; font-size: 0.7rem; color: #B5A88A; letter-spacing: 0.25em; text-transform: uppercase; margin-top: 6px; } | |
| .rc-row { display: flex; gap: 8px; flex-wrap: wrap; justify-content: center; margin-top: 10px; } | |
| .rc-badge { padding: 2px 10px; font-family: 'Cinzel', serif; font-size: 0.58rem; letter-spacing: 0.15em; border-radius: 2px; border: 1px solid; } | |
| .rc-live { border-color: #40916C; color: #52D68A; } | |
| .rc-voice { border-color: #C5B358; color: #C5B358; } | |
| .rc-pending { border-color: #4A3F30; color: #4A3F30; } | |
| label, .label-wrap span { font-family: 'Cinzel', serif !important; font-size: 0.68rem !important; letter-spacing: 0.1em !important; color: #B5A88A !important; text-transform: uppercase !important; } | |
| textarea, input[type=text], input[type=password] { background: #0F0D0A !important; border: 1px solid #2A2218 !important; color: #E8DCC8 !important; font-family: 'Cormorant Garamond', serif !important; font-size: 1rem !important; border-radius: 2px !important; } | |
| textarea:focus, input:focus { border-color: #C5B358 !important; box-shadow: 0 0 0 1px #C5B358 !important; } | |
| .telemetry textarea { background: #060806 !important; border: 1px solid #2D6A4F !important; color: #52D68A !important; font-family: 'Courier New', monospace !important; font-size: 0.73rem !important; } | |
| .json-out textarea { background: #06080A !important; border: 1px solid #1A2A3A !important; color: #7EB8D4 !important; font-family: 'Courier New', monospace !important; font-size: 0.72rem !important; } | |
| .response-out textarea { background: #0A0906 !important; border: 1px solid #3A3020 !important; color: #E8DCC8 !important; font-size: 1.05rem !important; line-height: 1.7 !important; } | |
| .sec-label { font-family: 'Cinzel', serif; font-size: 0.6rem; letter-spacing: 0.28em; color: #4A3F30; text-transform: uppercase; padding: 6px 0 3px; border-bottom: 1px solid #141210; margin-bottom: 10px; } | |
| """ | |
| # ── UI ──────────────────────────────────────────────────────────────────────── | |
| def build_ui(): | |
| with gr.Blocks(css=BERYL_CSS, title="Nobility v1 — Amanda") as demo: | |
| history_state = gr.State([]) | |
| session_state = gr.State({}) | |
| gr.HTML(""" | |
| <div class="nobility-header"> | |
| <h1 class="nobility-title">NOBILITY v1</h1> | |
| <p class="nobility-sub">Beryl AI Labs · Amanda · New Orleans</p> | |
| <div class="rc-row"> | |
| <span class="rc-badge rc-live">RC-01 LIVE</span> | |
| <span class="rc-badge rc-voice">🎤 VOICE LIVE</span> | |
| <span class="rc-badge rc-live">RC-03 LIVE</span> | |
| <span class="rc-badge rc-live">RC-05 LIVE</span> | |
| <span class="rc-badge rc-pending">RC-02 ORACLE PENDING</span> | |
| </div> | |
| </div> | |
| """) | |
| with gr.Row(): | |
| # ── LEFT ────────────────────────────────────────────────────────── | |
| with gr.Column(scale=1, min_width=260): | |
| gr.HTML('<div class="sec-label">Configuration</div>') | |
| api_key = gr.Textbox( | |
| label="NIM API Key", | |
| placeholder="nvapi-...", | |
| type="password", | |
| info="Free at build.nvidia.com", | |
| ) | |
| reference_image = gr.Image( | |
| label="Identity Reference", | |
| type="pil", | |
| height=200, | |
| ) | |
| gr.HTML('<div class="sec-label" style="margin-top:14px;">Layer Status</div>') | |
| gr.HTML("""<div style="font-family:'Courier New',monospace;font-size:0.7rem;line-height:2.1;"> | |
| <span style="color:#52D68A;">[HEAR] Qwen3-ASR-1.7B ⬤ LIVE</span><br> | |
| <span style="color:#52D68A;">[THINK] Qwen3 / NIM ⬤ LIVE</span><br> | |
| <span style="color:#52D68A;">[PLAN] Director ⬤ LIVE</span><br> | |
| <span style="color:#C5B358;">[SPEAK] Kokoro TTS ⬤ LIVE</span><br> | |
| <span style="color:#4A3F30;">[SEE] Wan2.2-S2V ◌ ORACLE</span> | |
| </div>""") | |
| # ── CENTER ──────────────────────────────────────────────────────── | |
| with gr.Column(scale=2): | |
| gr.HTML('<div class="sec-label">Talk to Amanda</div>') | |
| with gr.Row(): | |
| mic_input = gr.Audio( | |
| label="🎤 Speak", | |
| sources=["microphone"], | |
| type="numpy", | |
| ) | |
| text_input = gr.Textbox( | |
| label="Or type", | |
| placeholder="Say anything...", | |
| lines=2, | |
| ) | |
| with gr.Row(): | |
| run_btn = gr.Button("TRANSMIT", variant="primary") | |
| clear_btn = gr.Button("CLEAR") | |
| gr.HTML('<div class="sec-label" style="margin-top:6px;">Amanda — Response</div>') | |
| response_out = gr.Textbox( | |
| label="", | |
| lines=3, | |
| interactive=False, | |
| elem_classes=["response-out"], | |
| ) | |
| gr.HTML('<div class="sec-label">Amanda — Voice</div>') | |
| voice_out = gr.Audio( | |
| label="", | |
| type="numpy", | |
| autoplay=True, | |
| ) | |
| gr.HTML('<div class="sec-label">Layer 2 — Shot Plan</div>') | |
| shot_out = gr.Textbox( | |
| label="", | |
| lines=10, | |
| interactive=False, | |
| elem_classes=["json-out"], | |
| ) | |
| # ── RIGHT ───────────────────────────────────────────────────────── | |
| with gr.Column(scale=1, min_width=230): | |
| gr.HTML('<div class="sec-label">Pipeline Telemetry</div>') | |
| telemetry_out = gr.Textbox( | |
| label="", | |
| lines=26, | |
| interactive=False, | |
| elem_classes=["telemetry"], | |
| value=( | |
| "━━ NOBILITY v1 ━━\n" | |
| "Amanda is ready.\n\n" | |
| "[HEAR] — ms\n" | |
| "[THINK] — ms\n" | |
| "[PLAN] — ms\n" | |
| "[SPEAK] — ms\n" | |
| "─────────────\n" | |
| "Total — ms\n\n" | |
| "Speak or type.\n" | |
| "She will answer.\n\n" | |
| "[L4] VIDEO PENDING\n" | |
| "[L5] DECODE PENDING\n\n" | |
| "0.3 Deviation Rule:\n" | |
| "Melanin: 0.30 weight\n" | |
| "Identity: locked\n\n" | |
| "\"We need to have\n" | |
| " a face to face.\"\n" | |
| " — Beryl Live" | |
| ), | |
| ) | |
| gr.HTML("""<div style="text-align:center;padding:20px 0 6px;border-top:1px solid #141210;margin-top:20px;"> | |
| <span style="font-family:'Cinzel',serif;font-size:0.55rem;letter-spacing:0.25em;color:#2A2218;text-transform:uppercase;"> | |
| Nobility v1 · Beryl AI Labs · New Orleans · Not Runway. Not ElevenLabs. Not HeyGen. | |
| </span></div>""") | |
| # ── Events ──────────────────────────────────────────────────────────── | |
| def pipeline_wrap(api_key, text_in, audio_in, ref_img, hist, st): | |
| return run_full_pipeline(api_key, text_in, audio_in, ref_img, hist, st) | |
| run_btn.click( | |
| fn=pipeline_wrap, | |
| inputs=[api_key, text_input, mic_input, reference_image, history_state, session_state], | |
| outputs=[response_out, voice_out, shot_out, telemetry_out, history_state, session_state], | |
| ) | |
| text_input.submit( | |
| fn=pipeline_wrap, | |
| inputs=[api_key, text_input, mic_input, reference_image, history_state, session_state], | |
| outputs=[response_out, voice_out, shot_out, telemetry_out, history_state, session_state], | |
| ) | |
| def clear(): | |
| return "", None, "{}", ( | |
| "━━ NOBILITY v1 ━━\n" | |
| "Cleared. Ready.\n\n" | |
| "Awaiting turn..." | |
| ), [], {} | |
| clear_btn.click( | |
| fn=clear, | |
| outputs=[response_out, voice_out, shot_out, telemetry_out, history_state, session_state], | |
| ) | |
| return demo | |
| if __name__ == "__main__": | |
| demo = build_ui() | |
| demo.launch(share=True, show_error=True) | |