"""
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 ─────────────────────────────────────────────────────────
@spaces.GPU(duration=30)
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}]"
@spaces.GPU(duration=45)
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
@spaces.GPU(duration=20)
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("""
""")
with gr.Row():
# ── LEFT ──────────────────────────────────────────────────────────
with gr.Column(scale=1, min_width=260):
gr.HTML('Configuration
')
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('Layer Status
')
gr.HTML("""
[HEAR] Qwen3-ASR-1.7B ⬤ LIVE
[THINK] Qwen3 / NIM ⬤ LIVE
[PLAN] Director ⬤ LIVE
[SPEAK] Kokoro TTS ⬤ LIVE
[SEE] Wan2.2-S2V ◌ ORACLE
""")
# ── CENTER ────────────────────────────────────────────────────────
with gr.Column(scale=2):
gr.HTML('Talk to Amanda
')
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('Amanda — Response
')
response_out = gr.Textbox(
label="",
lines=3,
interactive=False,
elem_classes=["response-out"],
)
gr.HTML('Amanda — Voice
')
voice_out = gr.Audio(
label="",
type="numpy",
autoplay=True,
)
gr.HTML('Layer 2 — Shot Plan
')
shot_out = gr.Textbox(
label="",
lines=10,
interactive=False,
elem_classes=["json-out"],
)
# ── RIGHT ─────────────────────────────────────────────────────────
with gr.Column(scale=1, min_width=230):
gr.HTML('Pipeline Telemetry
')
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("""
Nobility v1 · Beryl AI Labs · New Orleans · Not Runway. Not ElevenLabs. Not HeyGen.
""")
# ── 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)