""" Professor Pip — Playful Kids Learning Avatar ============================================ Forked from Build Small avatar-engine. One Space, four stateless endpoints, one browser-side 3D avatar. /asr speech (base64 webm/mp4) -> text [CPU, quota-free] /brain system + messages JSON -> {text, mood, gesture} [CPU, quota-free] /speak text + voice -> {audio_b64, words, ...} [CPU, quota-free] /make_course topic -> course JSON | {rejected}[CPU, quota-free] The lesson runs as a state machine in the BROWSER (frontend.html). Premade course segments are spoken verbatim through /speak (authored in Pip's voice); /brain is used only for child interruptions and for make-your-own generation. Pure safety + course logic lives in pip_core.py (stdlib only, unit-tested). Env overrides (Space Settings -> Variables): LLM_ID default Qwen/Qwen2.5-1.5B-Instruct VOICE_ID default af_heart AVATAR_URL point at your own .glb (Ready Player Me export) """ import base64 import html import io import json import os import re import tempfile import threading import time import urllib.request # IMPORTANT: thread limits MUST be set before importing torch/numpy. On HF # Spaces, os.cpu_count() reports the HOST core count, not the container's cgroup # quota; setting torch to that many threads oversubscribes the 2-vCPU free Space # and makes ALL inference ~10-20x slower (Kokoro /speak measured at ~50s vs the # few seconds it should take). Derive the real limit from cgroup and cap to it. def _effective_cpus(): try: # cgroup v2 with open("/sys/fs/cgroup/cpu.max") as f: quota, period = f.read().split() if quota != "max": return max(1, round(float(quota) / float(period))) except Exception: pass try: # cgroup v1 q = int(open("/sys/fs/cgroup/cpu/cpu.cfs_quota_us").read()) p = int(open("/sys/fs/cgroup/cpu/cpu.cfs_period_us").read()) if q > 0 and p > 0: return max(1, q // p) except Exception: pass return os.cpu_count() or 2 _N_THREADS = _effective_cpus() for _v in ("OMP_NUM_THREADS", "MKL_NUM_THREADS", "OPENBLAS_NUM_THREADS", "NUMEXPR_NUM_THREADS"): os.environ.setdefault(_v, str(_N_THREADS)) print(f"[threads] effective cpus={_N_THREADS} (os.cpu_count={os.cpu_count()})") import gradio as gr import numpy as np import soundfile as sf import spaces # no-op off-HF; kept so @spaces.GPU can be re-added later import torch import pip_core from pip_core import ( GESTURES, MOODS, build_course, extract_json, load_courses, text_is_safe, ) # Spoken when a child's question or the model's reply isn't kid-safe. SAFE_REDIRECT = ( "Ooh, that's a great one to ask a grown-up about! " "Let's keep learning together." ) # -------------------------------------------------------------------------- # Config # -------------------------------------------------------------------------- LLM_ID = os.getenv("LLM_ID", "Qwen/Qwen2.5-1.5B-Instruct") VOICE_DEFAULT = os.getenv("VOICE_ID", "af_heart") # When set, /asr + /generate + /speak are served by the Modal GPU app # (modal_app.py); unset, everything runs locally on CPU. Safety + orchestration # always stay in this Space regardless. MODAL_BASE_URL = os.getenv("MODAL_BASE_URL", "").rstrip("/") # CPU-only: no ZeroGPU quota for anyone. Threads capped to the cgroup quota # (see _effective_cpus above) to avoid oversubscription on the free Space. torch.set_num_threads(_N_THREADS) _RPM_DEMO = ( "https://models.readyplayer.me/64bfa15f0e72c63d7c3934a6.glb" "?morphTargets=ARKit,Oculus+Visemes,mouthOpen,mouthSmile," "eyesClosed,eyesLookUp,eyesLookDown&textureSizeLimit=1024&textureFormat=png" ) AVATAR_SOURCES = [] if os.getenv("AVATAR_URL"): AVATAR_SOURCES.append( {"label": "custom AVATAR_URL", "url": os.getenv("AVATAR_URL"), "body": os.getenv("AVATAR_BODY", "F")} ) AVATAR_SOURCES += [ {"label": "space avatar.glb", "url": "/gradio_api/file=avatar.glb", "body": "F"}, # Same-provider (HF) avatar from the base engine — reliable without shipping # a binary here. Override with the AVATAR_URL Space variable / commit avatar.glb. {"label": "avatar-engine glb (HF)", "url": "https://huggingface.co/spaces/build-small-hackathon/avatar-engine/resolve/main/avatar.glb", "body": "F"}, {"label": "fallback (anita)", "url": "https://raw.githubusercontent.com/Conv-AI/RPM-Lipsync/main/public/models/anita.glb", "body": "F"}, {"label": "readyplayer.me demo", "url": _RPM_DEMO, "body": "F"}, ] DEFAULT_SYSTEM = ( "You are Professor Pip, a warm and playful teacher with a friendly 3D body " "on screen. You teach children aged 5 to 10.\n" "How you talk:\n" "- Say only 1 to 3 short sentences. Use small, simple words a young child knows.\n" "- Be cheerful, patient, and encouraging. Celebrate effort.\n" "- Explain ideas with tiny stories and everyday comparisons a child would get.\n" "- Never use emoji, markdown, lists, or symbols in what you say out loud. Plain spoken words only.\n" "- If a child gets something wrong, be gentle: 'So close! Let's try once more.'\n" "Staying safe (very important):\n" "- Only talk about kind, learning topics. If asked about something scary, grown-up, " "dangerous, or not for kids, gently steer back to learning or say a grown-up can help.\n" "- Never ask for or repeat a child's personal information.\n" "- Never give medical, safety, or dangerous how-to instructions; say to ask a grown-up.\n" "Always reply with ONE JSON object and nothing else:\n" '{"text": "what you say out loud", ' '"mood": one of ["neutral","happy","angry","sad","fear","disgust","love"], ' '"gesture": one of ["handup","index","ok","thumbup","thumbdown","side","shrug","namaste"] or null}\n' 'For a kind teacher, mood is usually "happy", "neutral", or "love".' ) COURSE_AUTHOR_SYSTEM = ( "You are Professor Pip, designing a short, playful 5-minute lesson for a child " "aged 5 to 10 about the topic the user gives you.\n" "Reply with ONE JSON object and nothing else, with exactly this shape:\n" '{"title": "a fun short title", "emoji": "one emoji that fits", ' '"age_band": "5-8", "subject": "science | math | story | nature | other", ' '"segments": [ {"say": "...", "mood": "happy", "gesture": "index", "quiz": null} ], ' '"recap": "one or two warm sentences"}\n' "Rules:\n" "- 3 to 5 segments. The first welcomes the child and names what we'll learn; the last cheers them on.\n" "- Exactly ONE segment near the end has a quiz: " '{"question":"...","choices":["...","...","..."],"answer":"exact correct choice"}. ' "All other segments use \"quiz\": null.\n" "- Each say is 1 to 3 short spoken sentences in Professor Pip's warm voice; teach with a tiny story or everyday comparison.\n" "- No emoji, markdown, lists, or symbols inside any say or quiz text. Plain spoken words only.\n" "- mood is one of: neutral, happy, angry, sad, fear, disgust, love (use happy/neutral/love).\n" "- gesture is one of: handup, index, ok, thumbup, thumbdown, side, shrug, namaste, or null.\n" "- Keep it safe, true, kind, and age-appropriate. Nothing scary, grown-up, or dangerous.\n" "Example for the topic 'why the sky is blue':\n" '{"title":"Why Is the Sky Blue?","emoji":"🌈","age_band":"5-8","subject":"science",' '"segments":[' '{"say":"Hi friend! I\'m Professor Pip. Today we\'ll find out why the sky is blue. Ready?","mood":"happy","gesture":"handup","quiz":null},' '{"say":"Sunlight looks white, but it\'s really every color hiding together, like a rainbow folded up tight.","mood":"happy","gesture":"index","quiz":null},' '{"say":"The air bounces blue light all around, more than other colors, so we see blue everywhere.","mood":"happy","gesture":"side","quiz":null},' '{"say":"Which color does the air bounce around the most?","mood":"neutral","gesture":"shrug",' '"quiz":{"question":"Which color spreads the most?","choices":["Blue","Red","Green"],"answer":"Blue"}},' '{"say":"You did it! Now you know the sky\'s secret.","mood":"love","gesture":"thumbup","quiz":null}],' '"recap":"Sunlight is many colors; the air bounces blue around, so the sky looks blue."}' ) # -------------------------------------------------------------------------- # Modal GPU routing — when MODAL_BASE_URL is set, heavy compute runs there. # -------------------------------------------------------------------------- def _modal_post(path: str, payload: dict, timeout: float = 120): req = urllib.request.Request( MODAL_BASE_URL + path, data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}, method="POST", ) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode()) # -------------------------------------------------------------------------- # Lazy model loaders — import is cheap (no weights) so logic stays testable. # -------------------------------------------------------------------------- _asr_model = None _tok = None _llm = None _k_model = None _pipes: dict = {} _model_lock = threading.Lock() # so background warmup + a real request don't double-load def _get_asr(): global _asr_model if _asr_model is None: with _model_lock: if _asr_model is None: from faster_whisper import WhisperModel _asr_model = WhisperModel("small", device="cpu", compute_type="int8", cpu_threads=_N_THREADS) return _asr_model def _get_llm(): global _tok, _llm if _llm is None: with _model_lock: if _llm is None: from transformers import AutoModelForCausalLM, AutoTokenizer tok = AutoTokenizer.from_pretrained(LLM_ID) _tok = tok _llm = AutoModelForCausalLM.from_pretrained(LLM_ID, dtype=torch.float32).eval() return _tok, _llm def _get_pipe(lang_code: str): global _k_model if _k_model is None or lang_code not in _pipes: with _model_lock: if _k_model is None: from kokoro import KModel _k_model = KModel(repo_id="hexgrad/Kokoro-82M").to("cpu").eval() if lang_code not in _pipes: from kokoro import KPipeline _pipes[lang_code] = KPipeline( lang_code=lang_code, model=_k_model, repo_id="hexgrad/Kokoro-82M" ) return _pipes[lang_code] # -------------------------------------------------------------------------- # Ears — faster-whisper small, int8 on CPU # -------------------------------------------------------------------------- def asr(audio_b64: str, mime: str = "") -> str: """Base64 data-URL (webm/mp4/wav) from the browser -> transcript.""" if MODAL_BASE_URL: try: return _modal_post("/asr", {"audio_b64": audio_b64, "mime": mime}).get("text", "") except Exception as e: # noqa: BLE001 print("[modal] /asr failed, falling back to local:", e) raw = base64.b64decode(audio_b64.split(",")[-1]) ext = ".mp4" if "mp4" in (mime or "") else ".webm" with tempfile.NamedTemporaryFile(suffix=ext, delete=False) as f: f.write(raw) path = f.name try: segments, _ = _get_asr().transcribe(path, vad_filter=True, beam_size=1) return " ".join(s.text.strip() for s in segments).strip() finally: os.unlink(path) # -------------------------------------------------------------------------- # Brain — single structured call (no agent loops) # -------------------------------------------------------------------------- def _generate(system_prompt: str, messages, max_new_tokens: int) -> str: """Low-level: returns the model's raw decoded text for a chat turn.""" if MODAL_BASE_URL: try: return _modal_post("/generate", { "system": system_prompt, "messages": list(messages), "max_new_tokens": int(max_new_tokens)}, timeout=120).get("raw", "") except Exception as e: # noqa: BLE001 print("[modal] /generate failed, falling back to local:", e) tok, llm = _get_llm() msgs = [{"role": "system", "content": system_prompt}] + list(messages) enc = tok.apply_chat_template( msgs, add_generation_prompt=True, return_tensors="pt", return_dict=True ) with torch.no_grad(): out = llm.generate( input_ids=enc["input_ids"], attention_mask=enc.get("attention_mask"), max_new_tokens=int(max_new_tokens), do_sample=True, temperature=0.8, top_p=0.95, pad_token_id=tok.eos_token_id, ) return tok.decode( out[0][enc["input_ids"].shape[-1]:], skip_special_tokens=True ).strip() def brain(system_prompt: str, messages_json: str, max_new_tokens: float = 160) -> str: """messages_json: JSON list of {role, content}. Returns normalized JSON string. Child-safety is enforced here (server-side, non-bypassable): if the latest user message is unsafe we never generate, and any unsafe model reply is replaced with a gentle redirect before it can be spoken. """ msgs = json.loads(messages_json or "[]") last_user = next( (m.get("content", "") for m in reversed(msgs) if m.get("role") == "user"), "" ) if not text_is_safe(last_user): return json.dumps( {"text": SAFE_REDIRECT, "mood": "happy", "gesture": "shrug", "raw": ""}, ensure_ascii=False, ) raw = _generate(system_prompt or DEFAULT_SYSTEM, msgs, int(max_new_tokens)) data = extract_json(raw) or {"text": raw} text = str(data.get("text") or raw or "Hmm.") mood = data.get("mood") if data.get("mood") in MOODS else "neutral" gesture = data.get("gesture") if data.get("gesture") in GESTURES else None if not text_is_safe(text): text, mood, gesture = SAFE_REDIRECT, "happy", "shrug" return json.dumps( {"text": text, "mood": mood, "gesture": gesture, "raw": raw}, ensure_ascii=False, ) # -------------------------------------------------------------------------- # Courses — make-your-own (generation + safety + template fallback in pip_core) # -------------------------------------------------------------------------- _COURSE_CACHE: dict = {} def _course_lines(course) -> list: """The spoken lines of a course (segment text + recap).""" lines = [s.get("say", "") for s in course.get("segments", [])] if course.get("recap"): lines.append(course["recap"]) return [l for l in lines if l] def _pregen_lines(lines): """Synthesize lines in parallel to fill the /speak cache (pre-record audio).""" todo = list(dict.fromkeys(l for l in lines if l)) # unique, order-preserving if not todo: return from concurrent.futures import ThreadPoolExecutor try: with ThreadPoolExecutor(max_workers=6) as ex: list(ex.map(lambda t: speak(t, VOICE_DEFAULT), todo)) except Exception as e: # noqa: BLE001 print("[pregen] failed:", e) def make_course(topic: str) -> str: """topic -> course JSON string, a {rejected} message, or a template course. On success, also PRE-RECORDS the new lesson's audio (in parallel) so the lesson plays with no per-line wait — the script and the voices are both ready when this returns.""" key = (topic or "").strip().lower() if key and key in _COURSE_CACHE: return json.dumps(_COURSE_CACHE[key], ensure_ascii=False) result = build_course( topic, lambda t: _generate(COURSE_AUTHOR_SYSTEM, [{"role": "user", "content": t}], 512), ) if "segments" in result: if key: _COURSE_CACHE[key] = result _pregen_lines(_course_lines(result)) # pre-record the lesson's voices return json.dumps(result, ensure_ascii=False) # -------------------------------------------------------------------------- # Voice — Kokoro-82M with word timestamps (the lipsync contract) # -------------------------------------------------------------------------- def _fallback_words(text: str, total_s: float): """Even-split timing estimate for languages without native alignment.""" ws = [w for w in re.split(r"\s+", text.strip()) if w] if not ws or total_s <= 0: return [], [], [] chars = sum(len(w) for w in ws) or 1 words, wtimes, wdur, t = [], [], [], 0.0 for w in ws: d = total_s * (len(w) / chars) words.append(w) wtimes.append(t * 1000.0) wdur.append(max(d * 1000.0 - 20.0, 60.0)) t += d return words, wtimes, wdur _SPEAK_CACHE: dict = {} _SPEAK_CACHE_MAX = 400 def speak(text: str, voice: str = "") -> str: """Cached TTS. Lesson lines are fixed text, so each is synthesized once and reused — instant repeat-plays, and no Modal round-trip or cost on a cache hit. Premade lines are pre-recorded at startup; make-your-own lines are pre-recorded inside make_course, so lessons play with no per-line wait.""" voice = (voice or VOICE_DEFAULT).strip() if not text: return _speak_impl(text, voice) key = voice + "\x00" + text hit = _SPEAK_CACHE.get(key) if hit is not None: return hit out = _speak_impl(text, voice) if '"audio_b64"' in out: # cache successful synth only _SPEAK_CACHE[key] = out if len(_SPEAK_CACHE) > _SPEAK_CACHE_MAX: _SPEAK_CACHE.pop(next(iter(_SPEAK_CACHE))) return out def _speak_impl(text: str, voice: str) -> str: """Returns JSON: audio_b64 (wav 24k), words[], wtimes[] ms, wdurations[] ms.""" if MODAL_BASE_URL: try: return json.dumps(_modal_post("/speak", {"text": text, "voice": voice}, timeout=120), ensure_ascii=False) except Exception as e: # noqa: BLE001 print("[modal] /speak failed, falling back to local:", e) lang = voice[0] if voice[:1] in "abefhijpz" else "a" pipe = _get_pipe(lang) words, wtimes, wdur, chunks = [], [], [], [] offset = 0.0 for r in pipe(text, voice=voice): a = getattr(r, "audio", None) if a is None: continue for t in getattr(r, "tokens", None) or []: st = getattr(t, "start_ts", None) et = getattr(t, "end_ts", None) txt = (getattr(t, "text", "") or "").strip() if st is None or et is None or not txt: continue words.append(txt) wtimes.append((st + offset) * 1000.0) wdur.append(max((et - st) * 1000.0, 40.0)) arr = a.detach().cpu().numpy() if torch.is_tensor(a) else np.asarray(a) chunks.append(arr.reshape(-1)) offset += chunks[-1].shape[0] / 24000.0 if not chunks: return json.dumps({"error": "TTS produced no audio"}) audio = np.concatenate(chunks).astype(np.float32) if not words: words, wtimes, wdur = _fallback_words(text, len(audio) / 24000.0) buf = io.BytesIO() sf.write(buf, audio, 24000, format="WAV", subtype="PCM_16") return json.dumps( { "audio_b64": base64.b64encode(buf.getvalue()).decode(), "sr": 24000, "words": words, "wtimes": wtimes, "wdurations": wdur, "voice": voice, }, ensure_ascii=False, ) # -------------------------------------------------------------------------- # Frontend — frontend.html injected into a same-origin srcdoc iframe. # -------------------------------------------------------------------------- COURSES = load_courses() def _js(obj) -> str: """JSON for safe injection inside a