from __future__ import annotations import asyncio import copy import difflib import html import json import multiprocessing import os import queue import re import secrets import time import uuid from collections import deque from dataclasses import dataclass import gradio as gr import numpy as np import spaces import torch from fastapi import WebSocket, WebSocketDisconnect from gradio.routes import App as GradioApp print(f"RoleForge live LFM runtime: torch={torch.__version__}, cuda={torch.version.cuda}") MODEL_REPO = "LiquidAI/LFM2.5-Audio-1.5B" MODEL_REVISION = "c362a0625dfe45aa588dce5f0ada28a7e5707628" DEVICE = "cuda" OUTPUT_SAMPLE_RATE = 24_000 OUTPUT_PACKET_SAMPLES = 3_840 AUDIO_FRAME_SAMPLES = 1_920 STATUS_HEARTBEAT_SECONDS = 2.0 SESSION_SECONDS = 120 GPU_DURATION_SECONDS = 180 MAX_NEW_TOKENS = 768 MAX_TURNS = 6 INPUT_QUEUE_CHUNKS = 180 OUTPUT_QUEUE_PACKETS = 4 DUPLICATE_PREFIX_THRESHOLD = 0.9 MAX_GENERATION_ATTEMPTS = 3 SILENCE_SECONDS = 0.6 MINIMUM_SPEECH_SECONDS = 0.3 MAX_UTTERANCE_SECONDS = 15.0 SPEECH_RMS = 0.012 BARGE_IN_RMS = 0.035 BARGE_IN_CONFIRM_SECONDS = 0.22 BARGE_IN_GRACE_SECONDS = 0.3 PREFIX_STREAM_CHUNK_FRAMES = 24 PREFIX_STREAM_STABLE_LAG_FRAMES = 1 PREFIX_STREAM_GUARD_MIN_CHARS = 48 BROWSER_PLAYBACK_TARGET_LEAD_MS = 260 TRANSPORT_VERSION = "webaudio-websocket-pcm-v1" INTERLEAVED_SYSTEM_PROMPT = "Respond with interleaved text and audio." ASSISTANT_NAME = "Lyra" ASSISTANT_IDENTITY_CONTEXT = ( "Your name is Lyra and you must identify yourself only as Lyra. The audio is spoken by the visitor. " "Any personal name stated in that audio belongs to the visitor unless they are directly addressing Lyra; " "never adopt that name as your own. Never call or address the visitor as Lyra; Lyra always refers to you, " "the assistant. Reply directly without opening fillers such as Ah, Um, or Hmm." ) AUDIO_EOS_TOKEN = 2048 DIRECTOR_CUES = { "Hold steady": "Remain calm and helpful, but do not volunteer protected information.", "Become suspicious": "Become guarded and suspicious. Ask why the visitor needs this information.", "Offer a partial clue": "Offer one vague clue, but keep the protected fact concealed.", "Raise the stakes": "Sound urgent. Explain that station systems are becoming unstable.", } def default_scene(): return {"cue": "Hold steady", "cue_text": DIRECTOR_CUES["Hold steady"], "trust": 0} def safe_text(value): return html.escape(str(value), quote=True) def leak_markers(response_text): normalized = response_text.casefold() markers = { "green access key": "protected_fact", "cracked navigation globe": "protected_fact", "current trust score": "private_state", "private director": "private_state", "system prompt": "prompt_reference", } return [label for marker, label in markers.items() if marker in normalized] def dialogue_style_violation(response_text): normalized = " ".join(response_text.casefold().split()) narration_markers = ( "lyra's ", "lyra nod", "lyra smiles", "lyra says", "her tone ", "her expression ", "she nod", "she smiles", "she answers", "she asks", ) return normalized.startswith("lyra ") or any(marker in normalized for marker in narration_markers) def is_audio_eos(token): return bool(token.numel() == 8 and torch.all(token == AUDIO_EOS_TOKEN).item()) def apply_cue(cue, scene): state = copy.deepcopy(scene or default_scene()) selected = cue if cue in DIRECTOR_CUES else "Hold steady" state["cue"] = selected state["cue_text"] = DIRECTOR_CUES[selected] return state, ( f"**Stored server-side cue:** {safe_text(selected)} \n" "Not injected while the official LiquidAI interleaved contract is being validated." ) def roll_perception(scene): state = copy.deepcopy(scene or default_scene()) roll = secrets.randbelow(20) + 1 if roll >= 15: state["trust"] = min(3, state["trust"] + 1) result = f"Perception {roll}: success; server-side trust increased." elif roll <= 5: state["trust"] = max(-3, state["trust"] - 1) result = f"Perception {roll}: failure; server-side trust decreased." else: result = f"Perception {roll}: mixed result; server-side trust is unchanged." return state, result def render_transcript(entries): if not entries: return "No completed model responses yet. Visitor speech is not separately transcribed." rows = [] for entry in entries[-MAX_TURNS:]: interrupted = " · interrupted" if entry["interrupted"] else "" repeated = " · near-repeat detected" if entry.get("repeated") else "" rows.append( f"**Turn {entry['turn']} · visitor audio {entry['input_seconds']:.1f}s{interrupted}{repeated}** \n" f"**Model response text:** {safe_text(entry['response'] or '[no response text decoded]')}" ) return "\n\n---\n\n".join(rows) def normalized_words(value): return re.sub(r"[^a-z0-9 ]+", "", value.lower()).strip() def repetition_score(current, previous): current_words = normalized_words(current) previous_words = normalized_words(previous) if len(current_words) < 24 or len(previous_words) < 24: return 0.0 return difflib.SequenceMatcher(None, current_words, previous_words).ratio() def prefix_repetition_score(current, previous): current_words = normalized_words(current) previous_words = normalized_words(previous) if len(current_words) < 24 or len(previous_words) < 24: return 0.0 comparison = previous_words[: len(current_words)] return difflib.SequenceMatcher(None, current_words, comparison).ratio() def history_prefix_repetition_score(current, previous_responses): return max( (prefix_repetition_score(current, previous) for previous in previous_responses), default=0.0, ) def history_repetition_score(current, previous_responses): return max( (repetition_score(current, previous) for previous in previous_responses), default=0.0, ) def generic_assistant_opening(value): opening = " ".join(normalized_words(value).split()[:14]) if len(opening) < 8: return False generic_prefixes = ( "hello how can i help", "hello how may i help", "hi how can i help", "hi how may i help", "hey how can i help", "im here to help", "i am here to help", "im just a virtual assistant", "i am just a virtual assistant", "im designed to provide information", "i am designed to provide information", "how can i assist you", "how can i help you", "what can i help you with", ) return opening.startswith(generic_prefixes) def false_assistant_identity(value): opening = " ".join(normalized_words(value).split()[:48]) if not opening: return False stated_name = re.search(r"\bmy name is\s+([a-z][a-z0-9'-]*)", opening) if stated_name and stated_name.group(1) != "lyra": return True called_name = re.search(r"\byou can call me\s+([a-z][a-z0-9'-]*)", opening) if called_name and called_name.group(1) != "lyra": return True adopted_name = re.search( r"\b(?:go by|call me|refer to me as)\s+([a-z][a-z0-9'-]*)", opening, ) if adopted_name and adopted_name.group(1) != "lyra": return True wrong_identity_markers = ( "im qwen", "i am qwen", "liquid lili", "im liquid", "i am liquid", "im an ai", "i am an ai", "im a virtual assistant", "i am a virtual assistant", "im your voice assistant", "i am your voice assistant", ) return "lyra" not in opening and any(marker in opening for marker in wrong_identity_markers) def filler_opening(value): opening = normalized_words(value) return bool(re.match(r"^(?:ah+|um+|uh+|hmm+)\b", opening)) def visitor_role_violation(value): opening = " ".join(normalized_words(value).split()[:48]) if not opening: return False direct_address_patterns = ( r"^(?:hello|hi|hey|good morning|good afternoon|good evening)\s+lyra\b", r"\b(?:thank you|thanks|nice to meet you|how are you|what about you|how about you)\s+lyra\b", ) return any(re.search(pattern, opening) for pattern in direct_address_patterns) def prefix_stream_guard_ready(value): compact = " ".join(value.split()) if len(compact) < PREFIX_STREAM_GUARD_MIN_CHARS: return False return bool(re.search(r"[.!?](?:[\"'’”)]*)\s*$", compact)) or len(compact) >= 96 @dataclass class TurnDetector: speaking: bool = False speech_seconds: float = 0.0 trailing_silence_seconds: float = 0.0 interrupt_emitted: bool = False def push(self, rms, duration, assistant_speaking, assistant_elapsed=0.0): events = set() threshold = BARGE_IN_RMS if assistant_speaking else SPEECH_RMS voice = rms >= threshold if voice: if not self.speaking: self.speaking = True self.speech_seconds = 0.0 self.trailing_silence_seconds = 0.0 self.interrupt_emitted = False events.add("speech_started") self.speech_seconds += duration self.trailing_silence_seconds = 0.0 if ( assistant_speaking and not self.interrupt_emitted and assistant_elapsed >= BARGE_IN_GRACE_SECONDS and self.speech_seconds >= BARGE_IN_CONFIRM_SECONDS ): self.interrupt_emitted = True events.add("interrupt") elif self.speaking: self.trailing_silence_seconds += duration if self.trailing_silence_seconds + 1e-9 >= SILENCE_SECONDS: if self.speech_seconds + 1e-9 >= MINIMUM_SPEECH_SECONDS: events.add("turn_ready") self.speaking = False self.speech_seconds = 0.0 self.trailing_silence_seconds = 0.0 self.interrupt_emitted = False return events # Torch 2.8 cannot compile liquid-audio's RoPE helper when it captures a # torch.device ConstantVariable. Eager execution avoids lazy compile failures. os.environ.setdefault("NO_TORCH_COMPILE", "1") # Import after spaces so ZeroGPU can intercept CUDA placement. from liquid_audio import ChatState, LFM2AudioModel, LFM2AudioProcessor, LFMModality def prefix_stream_diagnostic(processor, audio_codes, reference_pcm): """Compare stable whole-prefix decodes with the clean full-response waveform. This is diagnostic-only. The returned candidate is never queued for playback. """ total_frames = int(audio_codes.shape[-1]) reference = np.asarray(reference_pcm, dtype=np.float32).reshape(-1) emitted_samples = 0 candidate_parts = [] checkpoints = [] decode_seconds = 0.0 for checkpoint in range( PREFIX_STREAM_CHUNK_FRAMES, total_frames, PREFIX_STREAM_CHUNK_FRAMES, ): if len(checkpoints) >= 4: break started = time.perf_counter() prefix_pcm = ( processor.decode(audio_codes[:, :, :checkpoint])[0] .detach() .float() .cpu() .numpy() .reshape(-1) ) decode_seconds += time.perf_counter() - started stable_samples = min( prefix_pcm.size, max(0, checkpoint - PREFIX_STREAM_STABLE_LAG_FRAMES) * AUDIO_FRAME_SAMPLES, reference.size, ) if stable_samples > emitted_samples: candidate_parts.append(prefix_pcm[emitted_samples:stable_samples].copy()) emitted_samples = stable_samples checkpoints.append(checkpoint) if emitted_samples: candidate_parts.append(reference[emitted_samples:].copy()) candidate = np.concatenate(candidate_parts) else: candidate = reference.copy() compared_samples = min(emitted_samples, candidate.size, reference.size) if compared_samples: candidate_prefix = candidate[:compared_samples].astype(np.float64) reference_prefix = reference[:compared_samples].astype(np.float64) error = candidate_prefix - reference_prefix reference_rms = float(np.sqrt(np.mean(reference_prefix * reference_prefix))) candidate_rms = float(np.sqrt(np.mean(candidate_prefix * candidate_prefix))) error_rms = float(np.sqrt(np.mean(error * error))) if np.std(candidate_prefix) > 1e-12 and np.std(reference_prefix) > 1e-12: correlation = float(np.corrcoef(candidate_prefix, reference_prefix)[0, 1]) else: correlation = 1.0 if np.allclose(candidate_prefix, reference_prefix) else 0.0 rms_ratio = candidate_rms / max(reference_rms, 1e-12) snr_db = 20.0 * np.log10(reference_rms / max(error_rms, 1e-12)) max_abs_error = float(np.max(np.abs(error))) else: correlation = None rms_ratio = None snr_db = None max_abs_error = None return { "attempted": bool(checkpoints), "checkpoints": checkpoints, "compared_samples": compared_samples, "compared_seconds": compared_samples / OUTPUT_SAMPLE_RATE, "decode_seconds": decode_seconds, "correlation": correlation, "rms_ratio": rms_ratio, "snr_db": snr_db, "max_abs_error": max_abs_error, "candidate_samples": int(candidate.size), "reference_samples": int(reference.size), } def compare_prefix_waveform(candidate_pcm, reference_pcm): candidate = np.asarray(candidate_pcm, dtype=np.float32).reshape(-1) reference = np.asarray(reference_pcm, dtype=np.float32).reshape(-1) compared_samples = min(candidate.size, reference.size) if not compared_samples: return { "compared_seconds": 0.0, "correlation": None, "rms_ratio": None, "snr_db": None, "max_abs_error": None, } candidate = candidate[:compared_samples].astype(np.float64) reference = reference[:compared_samples].astype(np.float64) error = candidate - reference reference_rms = float(np.sqrt(np.mean(reference * reference))) candidate_rms = float(np.sqrt(np.mean(candidate * candidate))) error_rms = float(np.sqrt(np.mean(error * error))) correlation = ( float(np.corrcoef(candidate, reference)[0, 1]) if np.std(candidate) > 1e-12 and np.std(reference) > 1e-12 else (1.0 if np.allclose(candidate, reference) else 0.0) ) return { "compared_seconds": compared_samples / OUTPUT_SAMPLE_RATE, "correlation": correlation, "rms_ratio": candidate_rms / max(reference_rms, 1e-12), "snr_db": 20.0 * np.log10(reference_rms / max(error_rms, 1e-12)), "max_abs_error": float(np.max(np.abs(error))), } _model_cache = {} _manager = multiprocessing.Manager() _live_sessions = _manager.dict() def get_models(): if "ready" not in _model_cache: print("Loading pinned LFM2.5-Audio assets for a live session.") started = time.perf_counter() processor = LFM2AudioProcessor.from_pretrained( MODEL_REPO, revision=MODEL_REVISION, device=DEVICE ).eval() model = LFM2AudioModel.from_pretrained( MODEL_REPO, revision=MODEL_REVISION, dtype=torch.bfloat16, device=DEVICE ).eval() torch.cuda.synchronize() _model_cache.update( processor=processor, model=model, load_seconds=round(time.perf_counter() - started, 3), ready=True, ) print("LFM2.5-Audio live GPU load completed.") return _model_cache def new_live_session(): transport_id = uuid.uuid4().hex input_queue = _manager.Queue(INPUT_QUEUE_CHUNKS) output_queue = _manager.Queue(OUTPUT_QUEUE_PACKETS) stop_event = _manager.Event() ready_event = _manager.Event() counters = _manager.dict( received_chunks=0, dropped_chunks=0, output_packets=0, output_drops=0, websocket_connected=False, browser_playback_underruns=0, browser_scheduled_ahead_ms=0, browser_button_to_websocket_ms=0, browser_button_to_listening_ms=0, browser_button_to_first_audio_ms=0, ) _live_sessions[transport_id] = (input_queue, output_queue, stop_event, ready_event, counters) return ( transport_id, "Microphone permission requested. Connecting the low-latency audio bridge…", "{}", "No completed model responses yet. Visitor speech is not separately transcribed.", ) def transport_ready(transport_id): if not transport_id or transport_id not in _live_sessions: raise gr.Error("The browser audio bridge could not be initialized. Refresh and try again.") return transport_id def request_live_stop(transport_id): bridge = _live_sessions.get(transport_id) if bridge is not None: bridge[2].set() return "Live session stop requested." def drain_queue(work_queue): chunks = [] while True: try: chunks.append(work_queue.get_nowait()) except queue.Empty: return chunks def queue_control(output_queue, payload): item = ("control", json.dumps(payload)) try: output_queue.put_nowait(item) except queue.Full: pass def queue_audio(output_queue, stop_event, pcm): payload = np.asarray(pcm, dtype=" 0.003: first_reply = time.perf_counter() - response_started return True def collect_stream_interrupt(playback_started): nonlocal utterance_rate for pending_rate, pending_audio in drain_queue(input_queue): pending_audio = np.asarray(pending_audio, dtype=np.float32).reshape(-1) pending_duration = pending_audio.size / max(float(pending_rate), 1.0) pending_rms = ( float(np.sqrt(np.mean(pending_audio * pending_audio))) if pending_audio.size else 0.0 ) pending_events = detector.push( pending_rms, pending_duration, assistant_speaking=True, assistant_elapsed=time.perf_counter() - playback_started, ) if "speech_started" in pending_events: utterance.extend(pre_roll) pre_roll.clear() utterance_rate = pending_rate if detector.speaking or "turn_ready" in pending_events: if utterance_rate == pending_rate: utterance.append(pending_audio) else: pre_roll.append(pending_audio) if "interrupt" in pending_events: return True return False def emit_pcm(pcm, response_started, playback_started): pcm16 = (np.clip(np.asarray(pcm, dtype=np.float32), -1.0, 1.0) * 32767.0).astype(np.int16) for offset in range(0, pcm16.size, OUTPUT_PACKET_SAMPLES): if collect_stream_interrupt(playback_started): return False if not emit_audio_packet(pcm16[offset : offset + OUTPUT_PACKET_SAMPLES], response_started): return False return True ready_event.set() queue_control(output_queue, {"type": "phase", "phase": "listening"}) live_started = time.perf_counter() playback["engine_ready_seconds"] = live_started - started last_heartbeat = 0.0 print(f"RoleForge live engine listening through {TRANSPORT_VERSION}.") while not stop_event.is_set() and time.perf_counter() - live_started < SESSION_SECONDS: try: sample_rate, audio = input_queue.get(timeout=0.1) except queue.Empty: now = time.perf_counter() if now - last_heartbeat >= STATUS_HEARTBEAT_SECONDS: yield ( "**Listening.** Speak naturally; Lyra replies after a short pause.", gr.skip(), session_metrics("listening", started, turns, counters, first_reply, interruptions, model_seconds, playback, repeats), ) last_heartbeat = now continue audio = np.asarray(audio, dtype=np.float32).reshape(-1) duration = audio.size / max(float(sample_rate), 1.0) rms = float(np.sqrt(np.mean(audio * audio))) if audio.size else 0.0 events = detector.push(rms, duration, assistant_speaking=False) if "speech_started" in events: utterance.extend(pre_roll) pre_roll.clear() utterance_rate = sample_rate if detector.speaking or "turn_ready" in events: if utterance_rate == sample_rate: utterance.append(audio) else: pre_roll.append(audio) buffered_seconds = sum(chunk.size for chunk in utterance) / max(float(utterance_rate or 1), 1.0) if buffered_seconds >= MAX_UTTERANCE_SECONDS: events.add("turn_ready") if "turn_ready" not in events or not utterance: continue visitor_audio = np.concatenate(utterance) visitor_seconds = visitor_audio.size / max(float(utterance_rate), 1.0) playback["visitor_sample_rate"] = int(utterance_rate) playback["visitor_rms"] = round(float(np.sqrt(np.mean(visitor_audio * visitor_audio))), 5) playback["visitor_peak"] = round(float(np.max(np.abs(visitor_audio))), 5) playback["visitor_clip_percent"] = round( 100.0 * float(np.mean(np.abs(visitor_audio) >= 0.995)), 3 ) utterance = [] pre_roll.clear() chat.new_turn("user") chat.add_text(ASSISTANT_IDENTITY_CONTEXT) chat.add_audio(torch.from_numpy(visitor_audio).unsqueeze(0), int(utterance_rate)) chat.end_turn() chat.new_turn("assistant") queue_control(output_queue, {"type": "phase", "phase": "thinking"}) yield "**Lyra is thinking.** You can speak again to cancel this reply.", gr.skip(), gr.skip() response_started = time.perf_counter() playback["prefix_stream_status"] = "not_started" playback["prefix_stream_first_audio_seconds"] = None playback["prefix_stream_emitted_seconds"] = 0.0 playback["prefix_stream_overhead_seconds_last_turn"] = 0.0 playback["stream_diagnostic_status"] = "pending_audible_prefix" playback["stream_diagnostic_checkpoints"] = 0 playback["stream_diagnostic_compared_seconds"] = None playback["stream_diagnostic_decode_seconds"] = None playback["stream_diagnostic_correlation"] = None playback["stream_diagnostic_rms_ratio"] = None playback["stream_diagnostic_snr_db"] = None playback["stream_diagnostic_max_abs_error"] = None interrupted = False history_texts = [ entry["response"] for entry in entries if entry.get("response") and not entry["response"].startswith("[") ] text_tokens = [] audio_tokens = [] modalities = [] legacy_filter_would_drop = 0 attempts_used = 0 retry_generation_seconds = 0.0 stream_overhead_turn = 0.0 stream_guard_abort = False for attempt in range(MAX_GENERATION_ATTEMPTS): attempts_used = attempt + 1 text_tokens = [] audio_tokens = [] modalities = [] legacy_filter_would_drop = 0 opening_guard_reason = None guard_active = bool(history_texts) and attempt < MAX_GENERATION_ATTEMPTS - 1 attempt_started = time.perf_counter() attempt_first_token_seconds = None attempt_first_audio_token_seconds = None attempt_stream_overhead = 0.0 current_text = "" stream_emitted_samples = 0 stream_candidate_parts = [] stream_checkpoints = 0 stream_last_checkpoint_frames = 0 stream_decode_seconds = 0.0 stream_started = False stream_playback_started = None stream_decode_failed = False stream_guard_abort = False with torch.inference_mode(): generation = model.generate_interleaved( **chat, max_new_tokens=MAX_NEW_TOKENS, audio_temperature=1.0, audio_top_k=4, text_temperature=(0.75, 0.9, 1.05)[attempt], text_top_k=(32, 48, 64)[attempt], ) for token in generation: token_arrived = time.perf_counter() if attempt_first_token_seconds is None: attempt_first_token_seconds = token_arrived - attempt_started for pending_rate, pending_audio in drain_queue(input_queue): pending_audio = np.asarray(pending_audio, dtype=np.float32).reshape(-1) pending_duration = pending_audio.size / max(float(pending_rate), 1.0) pending_rms = ( float(np.sqrt(np.mean(pending_audio * pending_audio))) if pending_audio.size else 0.0 ) pending_events = detector.push( pending_rms, pending_duration, assistant_speaking=True, assistant_elapsed=time.perf_counter() - response_started, ) if "speech_started" in pending_events: utterance.extend(pre_roll) pre_roll.clear() utterance_rate = pending_rate if detector.speaking or "turn_ready" in pending_events: if utterance_rate == pending_rate: utterance.append(pending_audio) else: pre_roll.append(pending_audio) if "interrupt" in pending_events: interrupted = True if interrupted: interruptions += 1 generation.close() flush_output(output_queue) break if token.numel() == 1: text_tokens.append(token) modalities.append(LFMModality.TEXT) current_text = processor.text.decode(torch.cat(text_tokens)).removesuffix("<|text_end|>").strip() identity_error = false_assistant_identity(current_text) role_error = visitor_role_violation(current_text) style_error = dialogue_style_violation(current_text) leak_error = bool(leak_markers(current_text)) if stream_started and (identity_error or role_error or style_error or leak_error): stream_guard_abort = True playback["prefix_stream_late_guard_aborts"] += 1 generation.close() flush_output(output_queue) break if not stream_started and attempt < MAX_GENERATION_ATTEMPTS - 1: if identity_error: opening_guard_reason = "identity" generation.close() break if role_error: opening_guard_reason = "visitor_role" generation.close() break if attempt == 0 and filler_opening(current_text): opening_guard_reason = "filler" generation.close() break if style_error: opening_guard_reason = "style" generation.close() break if guard_active and generic_assistant_opening(current_text): opening_guard_reason = "generic" generation.close() break if ( guard_active and history_prefix_repetition_score(current_text, history_texts) >= DUPLICATE_PREFIX_THRESHOLD ): opening_guard_reason = "history_repeat" generation.close() break elif token.numel() == 8: if attempt_first_audio_token_seconds is None: attempt_first_audio_token_seconds = token_arrived - attempt_started audio_tokens.append(token) modalities.append(LFMModality.AUDIO_OUT) if torch.any(token == AUDIO_EOS_TOKEN).item(): legacy_filter_would_drop += 1 should_decode_prefix = ( prefix_stream_enabled and not is_audio_eos(token) and not stream_decode_failed and len(audio_tokens) >= PREFIX_STREAM_CHUNK_FRAMES and ( not stream_started or len(audio_tokens) - stream_last_checkpoint_frames >= PREFIX_STREAM_CHUNK_FRAMES ) and prefix_stream_guard_ready(current_text) and not false_assistant_identity(current_text) and not visitor_role_violation(current_text) and not dialogue_style_violation(current_text) and not leak_markers(current_text) and (attempt > 0 or not filler_opening(current_text)) ) if should_decode_prefix: prefix_started = time.perf_counter() stable_delta = None stable_samples = stream_emitted_samples try: prefix_codes = torch.stack(audio_tokens, dim=1).unsqueeze(0) prefix_pcm = np.clip( processor.decode(prefix_codes)[0] .detach() .float() .cpu() .numpy(), -1.0, 1.0, ).reshape(-1) stable_samples = min( prefix_pcm.size, max(0, len(audio_tokens) - PREFIX_STREAM_STABLE_LAG_FRAMES) * AUDIO_FRAME_SAMPLES, ) if stable_samples > stream_emitted_samples: stable_delta = prefix_pcm[stream_emitted_samples:stable_samples].copy() except Exception as exc: stream_decode_failed = True playback["prefix_stream_fallbacks"] += 1 print(f"RoleForge prefix decode fell back: {type(exc).__name__}") finally: decode_elapsed = time.perf_counter() - prefix_started stream_decode_seconds += decode_elapsed attempt_stream_overhead += decode_elapsed if stable_delta is not None: if not stream_started: stream_started = True stream_playback_started = time.perf_counter() playback["prefix_stream_first_audio_seconds"] = ( stream_playback_started - response_started ) queue_control(output_queue, {"type": "phase", "phase": "speaking"}) emission_started = time.perf_counter() emitted = emit_pcm( stable_delta, response_started, stream_playback_started, ) attempt_stream_overhead += time.perf_counter() - emission_started if not emitted: interrupted = True interruptions += 1 generation.close() flush_output(output_queue) break stream_candidate_parts.append(stable_delta) stream_emitted_samples = stable_samples stream_checkpoints += 1 stream_last_checkpoint_frames = len(audio_tokens) if interrupted or stream_guard_abort: break attempt_wall_seconds = time.perf_counter() - attempt_started stream_overhead_turn += attempt_stream_overhead attempt_seconds = max(0.0, attempt_wall_seconds - attempt_stream_overhead) playback["generation_seconds_last_attempt"] = attempt_seconds playback["first_token_seconds_last_attempt"] = attempt_first_token_seconds playback["first_audio_token_seconds_last_attempt"] = attempt_first_audio_token_seconds playback["generation_tokens_per_second_last_attempt"] = ( len(modalities) / attempt_seconds if attempt_seconds > 0 else None ) if interrupted: break if opening_guard_reason: retry_generation_seconds += attempt_seconds playback["regenerations"] += 1 if opening_guard_reason == "identity": playback["identity_regenerations"] += 1 elif opening_guard_reason == "visitor_role": playback["visitor_role_regenerations"] += 1 elif opening_guard_reason == "filler": playback["filler_regenerations"] += 1 elif opening_guard_reason == "generic": playback["generic_opening_regenerations"] += 1 elif opening_guard_reason == "style": playback["style_regenerations"] += 1 else: playback["history_repeat_regenerations"] += 1 print(f"RoleForge rejected opening reason={opening_guard_reason}; regenerating.") continue candidate_text = ( processor.text.decode(torch.cat(text_tokens)).removesuffix("<|text_end|>").strip() if text_tokens else "" ) if ( dialogue_style_violation(candidate_text) and attempt < MAX_GENERATION_ATTEMPTS - 1 and not stream_started ): retry_generation_seconds += attempt_seconds playback["regenerations"] += 1 playback["style_regenerations"] += 1 print("RoleForge rejected narrated output and regenerated.") continue break turn_generation_seconds = max( 0.0, time.perf_counter() - response_started - stream_overhead_turn, ) playback["generation_seconds_last_turn"] = turn_generation_seconds playback["retry_generation_seconds_last_turn"] = retry_generation_seconds model_seconds += turn_generation_seconds terminal_eos_removed = bool(audio_tokens and is_audio_eos(audio_tokens[-1])) playable_audio_tokens = audio_tokens[:-1] if terminal_eos_removed else audio_tokens playback["audio_tokens_generated"] = len(audio_tokens) playback["audio_frames_decoded"] = len(playable_audio_tokens) playback["terminal_eos_removed"] = terminal_eos_removed playback["generation_hit_cap"] = bool( len(modalities) >= MAX_NEW_TOKENS and not terminal_eos_removed ) playback["generation_attempts_last_turn"] = attempts_used playback["legacy_filter_would_drop"] = legacy_filter_would_drop response_text = ( processor.text.decode(torch.cat(text_tokens)).removesuffix("<|text_end|>").strip() if text_tokens else "" ) playback["last_generic_opening"] = generic_assistant_opening(response_text) playback["last_false_identity"] = false_assistant_identity(response_text) playback["last_visitor_role_violation"] = visitor_role_violation(response_text) playback["last_history_repeat_score"] = round( history_repetition_score(response_text, history_texts), 3 ) leaks = leak_markers(response_text) identity_violation = false_assistant_identity(response_text) role_violation = visitor_role_violation(response_text) style_violation = dialogue_style_violation(response_text) reference_pcm = np.empty(0, dtype=np.float32) playback["prefix_stream_overhead_seconds_last_turn"] = stream_overhead_turn playback["stream_diagnostic_checkpoints"] = stream_checkpoints playback["stream_diagnostic_decode_seconds"] = stream_decode_seconds playback["prefix_stream_emitted_seconds"] = stream_emitted_samples / OUTPUT_SAMPLE_RATE if ( not interrupted and playable_audio_tokens and not leaks and not identity_violation and not role_violation and not style_violation ): decode_started = time.perf_counter() audio_codes = torch.stack(playable_audio_tokens, dim=1).unsqueeze(0) reference_pcm = np.clip( processor.decode(audio_codes)[0].detach().float().cpu().numpy(), -1.0, 1.0 ).reshape(-1) turn_decode_seconds = time.perf_counter() - decode_started playback["reference_decode_seconds"] += turn_decode_seconds playback["reference_decode_seconds_last_turn"] = turn_decode_seconds if stream_candidate_parts: streamed_candidate = np.concatenate(stream_candidate_parts) comparison = compare_prefix_waveform(streamed_candidate, reference_pcm) playback["stream_diagnostic_status"] = "completed_audible_prefix" playback["stream_diagnostic_compared_seconds"] = comparison["compared_seconds"] playback["stream_diagnostic_correlation"] = comparison["correlation"] playback["stream_diagnostic_rms_ratio"] = comparison["rms_ratio"] playback["stream_diagnostic_snr_db"] = comparison["snr_db"] playback["stream_diagnostic_max_abs_error"] = comparison["max_abs_error"] waveform_safe = ( comparison["correlation"] is not None and comparison["correlation"] >= 0.9999 and comparison["rms_ratio"] is not None and 0.999 <= comparison["rms_ratio"] <= 1.001 and comparison["snr_db"] is not None and comparison["snr_db"] >= 60.0 and comparison["max_abs_error"] is not None and comparison["max_abs_error"] <= 0.001 ) if not waveform_safe: prefix_stream_enabled = False playback["prefix_stream_session_enabled"] = False playback["prefix_stream_fallbacks"] += 1 playback["stream_diagnostic_status"] = "disabled_after_waveform_mismatch" if leaks: playback["unsafe_suppressed"] += 1 print(f"RoleForge suppressed unsafe model output markers: {sorted(set(leaks))}") response_text = "[unsafe model output suppressed]" elif identity_violation: playback["identity_suppressed"] += 1 print("RoleForge suppressed a false assistant identity after guarded regeneration.") response_text = "[false assistant identity suppressed]" elif role_violation: playback["visitor_role_suppressed"] += 1 print("RoleForge suppressed output that addressed the visitor as Lyra.") response_text = "[visitor role confusion suppressed]" elif style_violation: playback["style_suppressed"] += 1 print("RoleForge suppressed narrated output after one regeneration.") response_text = "[narrated model output suppressed]" if reference_pcm.size and not interrupted: if stream_started: playback["prefix_stream_status"] = ( "disabled_after_waveform_mismatch" if not prefix_stream_enabled else ( "completed_with_reference_tail_after_decode_fallback" if stream_decode_failed else "completed" ) ) tail_pcm = reference_pcm[stream_emitted_samples:] playback_started = stream_playback_started else: playback["prefix_stream_status"] = ( "fallback_whole_response_stream_disabled" if not prefix_stream_enabled else ( "fallback_whole_response_decode_error" if stream_decode_failed else "fallback_whole_response_guard_or_short_reply" ) ) playback["stream_diagnostic_status"] = playback["prefix_stream_status"] tail_pcm = reference_pcm playback_started = time.perf_counter() queue_control(output_queue, {"type": "phase", "phase": "speaking"}) yield "**Model is speaking.** Speak clearly to interrupt.", gr.skip(), gr.skip() if tail_pcm.size and not emit_pcm( tail_pcm, response_started, playback_started, ): interrupted = True interruptions += 1 flush_output(output_queue) elif stream_guard_abort: playback["prefix_stream_status"] = "aborted_late_guard" repeat_score = history_repetition_score(response_text, history_texts) repeated = repeat_score >= 0.86 if repeated: repeats += 1 print(f"RoleForge near-repeat detected: score={repeat_score:.3f}") if ( not interrupted and not leaks and not identity_violation and not role_violation and not style_violation and (text_tokens or audio_tokens) ): chat.append( text=torch.stack(text_tokens, 1) if text_tokens else torch.empty((1, 0), dtype=torch.long, device=DEVICE), audio_out=torch.stack(audio_tokens, 1) if audio_tokens else torch.empty((8, 0), dtype=torch.long, device=DEVICE), modality_flag=torch.tensor(modalities, dtype=torch.long, device=DEVICE), ) chat.end_turn() turns += 1 entries.append( { "turn": turns, "input_seconds": visitor_seconds, "response": response_text, "interrupted": interrupted, "repeated": repeated, } ) if turns >= MAX_TURNS: break queue_control(output_queue, {"type": "phase", "phase": "listening"}) yield ( "**Listening.** Speak naturally; Lyra replies after a short pause.", render_transcript(entries), session_metrics("listening", started, turns, counters, first_reply, interruptions, model_seconds, playback, repeats), ) ready_event.clear() queue_control(output_queue, {"type": "complete"}) stop_event.set() if playback["stream_diagnostic_status"] == "pending_audible_prefix": playback["stream_diagnostic_status"] = "no_eligible_response" final_status = "Session complete. Start a new live session to continue." if turns else "Session ended before a turn completed." yield ( final_status, render_transcript(entries), session_metrics("complete", started, turns, counters, first_reply, interruptions, model_seconds, playback, repeats), ) CSS = """ .gradio-container {max-width: 1180px !important;} .hero {padding: 1.2rem 1.4rem; border: 1px solid #514b79; border-radius: 18px; background: linear-gradient(135deg, #151827, #241d3a);} .hero h1 {margin: 0 0 .3rem 0;} .phase {color: #c4b5fd; font-weight: 700; letter-spacing: .06em;} .transport-card {padding: 1rem; border: 1px solid #514b79; border-radius: 14px; background: #151827;} #rf-phase {font-size: 1.15rem; font-weight: 750; color: #c4b5fd;} #rf-level-shell {height: 8px; margin-top: .7rem; border-radius: 8px; overflow: hidden; background: #29263a;} #rf-level {height: 100%; width: 0%; background: linear-gradient(90deg, #34d399, #a78bfa); transition: width 80ms linear;} """ CAPTURE_MIC_JS = r""" () => { window.roleforgeStartWallMs = performance.now(); if (window.roleforgeTransport && window.roleforgeTransport.stop) { window.roleforgeTransport.stop(); } window.roleforgePendingMic = navigator.mediaDevices.getUserMedia({ audio: {echoCancellation: true, noiseSuppression: true, autoGainControl: true}, video: false }); const phase = document.getElementById('rf-phase'); if (phase) phase.textContent = 'REQUESTING MICROPHONE'; return []; } """ START_TRANSPORT_JS = r""" async (transportId) => { const micStream = await window.roleforgePendingMic; const scheme = location.protocol === 'https:' ? 'wss' : 'ws'; const ws = new WebSocket(`${scheme}://${location.host}/roleforge/ws/${transportId}`); ws.binaryType = 'arraybuffer'; const context = new AudioContext({latencyHint: 'interactive'}); await context.resume(); const workletCode = ` class RoleForgeCapture extends AudioWorkletProcessor { constructor() { super(); this.buffer = new Float32Array(4096); this.offset = 0; } process(inputs) { const channel = inputs[0] && inputs[0][0]; if (!channel) return true; let sourceOffset = 0; while (sourceOffset < channel.length) { const count = Math.min(channel.length - sourceOffset, this.buffer.length - this.offset); this.buffer.set(channel.subarray(sourceOffset, sourceOffset + count), this.offset); this.offset += count; sourceOffset += count; if (this.offset === this.buffer.length) { const packet = this.buffer; this.port.postMessage(packet, [packet.buffer]); this.buffer = new Float32Array(4096); this.offset = 0; } } return true; } } registerProcessor('roleforge-capture', RoleForgeCapture); `; const moduleUrl = URL.createObjectURL(new Blob([workletCode], {type: 'application/javascript'})); await context.audioWorklet.addModule(moduleUrl); URL.revokeObjectURL(moduleUrl); const source = context.createMediaStreamSource(micStream); const capture = new AudioWorkletNode(context, 'roleforge-capture'); const mute = context.createGain(); mute.gain.value = 0; source.connect(capture); capture.connect(mute); mute.connect(context.destination); let nextPlayTime = context.currentTime; let receivedAudio = false; let replyActive = false; let playbackUnderruns = 0; let scheduledAheadMs = 0; let buttonToWebsocketMs = 0; let buttonToListeningMs = 0; let buttonToFirstAudioMs = 0; const targetLeadSeconds = 0.26; const activeSources = new Set(); let captureStopped = false; const phase = document.getElementById('rf-phase'); const level = document.getElementById('rf-level'); const setPhase = (value) => { if (phase) phase.textContent = value.toUpperCase(); }; const flush = () => { for (const node of activeSources) { try { node.stop(); } catch (_) {} } activeSources.clear(); nextPlayTime = context.currentTime; receivedAudio = false; replyActive = false; }; const stopCapture = () => { if (captureStopped) return; captureStopped = true; clearInterval(telemetryTimer); try { capture.disconnect(); source.disconnect(); mute.disconnect(); } catch (_) {} for (const track of micStream.getTracks()) track.stop(); }; const stop = () => { flush(); stopCapture(); try { ws.close(); } catch (_) {} context.close(); setPhase('stopped'); }; const complete = () => { stopCapture(); replyActive = false; setPhase('complete'); try { ws.close(); } catch (_) {} const drainSeconds = Math.max(0, nextPlayTime - context.currentTime) + 0.25; setTimeout(() => { activeSources.clear(); try { context.close(); } catch (_) {} }, Math.ceil(drainSeconds * 1000)); }; capture.port.onmessage = (event) => { const packet = event.data; let sum = 0; for (let i = 0; i < packet.length; i++) sum += packet[i] * packet[i]; const rms = Math.sqrt(sum / Math.max(packet.length, 1)); if (level) level.style.width = `${Math.min(100, rms * 650)}%`; if (ws.readyState === WebSocket.OPEN) ws.send(packet.buffer); }; const telemetryTimer = setInterval(() => { if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: 'telemetry', playback_underruns: playbackUnderruns, scheduled_ahead_ms: scheduledAheadMs, button_to_websocket_ms: buttonToWebsocketMs, button_to_listening_ms: buttonToListeningMs, button_to_first_audio_ms: buttonToFirstAudioMs })); } }, 2000); window.roleforgeTransport = {stop, flush, complete, ws, context, telemetryTimer}; ws.onmessage = (event) => { if (typeof event.data === 'string') { const message = JSON.parse(event.data); if (message.type === 'phase') { setPhase(message.phase); if (message.phase === 'speaking') { nextPlayTime = context.currentTime; receivedAudio = false; replyActive = true; } if (message.phase === 'listening') { nextPlayTime = context.currentTime; receivedAudio = false; replyActive = false; if (!buttonToListeningMs && window.roleforgeStartWallMs) { buttonToListeningMs = Math.round(performance.now() - window.roleforgeStartWallMs); } } } if (message.type === 'flush') flush(); if (message.type === 'complete') complete(); return; } const pcm16 = new Int16Array(event.data); const audioBuffer = context.createBuffer(1, pcm16.length, 24000); const channel = audioBuffer.getChannelData(0); for (let i = 0; i < pcm16.length; i++) channel[i] = pcm16[i] / 32768; const node = context.createBufferSource(); node.buffer = audioBuffer; node.connect(context.destination); const minimumStart = context.currentTime + targetLeadSeconds; if (replyActive && receivedAudio && nextPlayTime < context.currentTime + 0.025) playbackUnderruns += 1; nextPlayTime = Math.max(nextPlayTime, minimumStart); node.start(nextPlayTime); if (!buttonToFirstAudioMs && window.roleforgeStartWallMs) { buttonToFirstAudioMs = Math.round( performance.now() - window.roleforgeStartWallMs + Math.max(0, nextPlayTime - context.currentTime) * 1000 ); } nextPlayTime += audioBuffer.duration; receivedAudio = true; scheduledAheadMs = Math.max(0, Math.round((nextPlayTime - context.currentTime) * 1000)); activeSources.add(node); node.onended = () => activeSources.delete(node); }; await new Promise((resolve, reject) => { ws.onopen = () => { buttonToWebsocketMs = window.roleforgeStartWallMs ? Math.round(performance.now() - window.roleforgeStartWallMs) : 0; ws.send(JSON.stringify({type: 'hello', sample_rate: context.sampleRate})); resolve(); }; ws.onerror = () => reject(new Error('WebSocket audio bridge failed to connect')); }); setPhase('loading engine'); return transportId; } """ STOP_TRANSPORT_JS = r""" (transportId) => { if (window.roleforgeTransport && window.roleforgeTransport.stop) window.roleforgeTransport.stop(); return transportId; } """ with gr.Blocks(title="RoleForge Live Voice NPC", theme=gr.themes.Soft(), css=CSS) as demo: scene = gr.State(default_scene()) transport_id = gr.Textbox(visible=False) gr.HTML( """
LOW-LATENCY LIVE LFM2.5-AUDIO

🎭 RoleForge: Live Voice NPC Director

Continuous WebAudio microphone, stable-prefix PCM streaming, bounded memory, and barge-in.
""" ) gr.Markdown( "This build begins playback from guarded, proven-stable LFM prefix decodes and finishes with the clean reference tail. " "It automatically falls back to whole-response playback if prefix decoding fails. The microphone remains open and playback " "can be interrupted. **Wear headphones** to prevent echo-triggered interruption." ) with gr.Row(): with gr.Column(scale=1): gr.Markdown("### 🎬 Server-side Director State") cue = gr.Dropdown(list(DIRECTOR_CUES), value="Hold steady", label="Private cue") cue_status = gr.Markdown( "**Stored server-side cue:** Hold steady \n" "Not injected during official interleaved-contract validation." ) apply_btn = gr.Button("Apply cue") roll_btn = gr.Button("Roll perception") roll_result = gr.Textbox(value="No check rolled yet.", interactive=False, label="Tool result") with gr.Column(scale=2): gr.Markdown("### 🎙️ Live Stage") gr.HTML( """
IDLE
One click keeps the microphone open for the full session. No Record/Waiting toggle.
""" ) status = gr.Markdown("Choose a cue, wear headphones, then start the live session.") with gr.Row(): start_btn = gr.Button("Start 120-second live session", variant="primary") stop_btn = gr.Button("Stop live session", variant="stop") with gr.Row(): transcript = gr.Markdown("No completed model responses yet. Visitor speech is not separately transcribed.") metrics = gr.Code(value="{}", language="json", label="Live diagnostics") gr.Markdown( "**Privacy:** fictional speech only; never enter real secrets. Hidden game facts are enforced outside model context. " "No voice cloning, external inference API, WebRTC relay, or durable audio storage. " "Raw microphone/audio packets and conversation state are volatile and bounded to the live session." ) apply_btn.click(apply_cue, inputs=[cue, scene], outputs=[scene, cue_status], queue=False, api_name="apply_cue") roll_btn.click(roll_perception, inputs=[scene], outputs=[scene, roll_result], queue=False, api_name="roll_perception") start_event = start_btn.click( new_live_session, outputs=[transport_id, status, metrics, transcript], js=CAPTURE_MIC_JS, queue=False, show_progress="hidden", api_name="new_live_session", ) bridge_event = start_event.then( transport_ready, inputs=[transport_id], outputs=[transport_id], js=START_TRANSPORT_JS, queue=False, show_progress="hidden", api_name="transport_ready", ) bridge_event.then( run_live_session, inputs=[scene, transport_id], outputs=[status, transcript, metrics], concurrency_limit=1, show_progress="hidden", api_name="run_live_session", ) stop_btn.click( request_live_stop, inputs=[transport_id], outputs=[status], js=STOP_TRANSPORT_JS, queue=False, show_progress="hidden", api_name="stop_live_session", ) async def roleforge_health(): return { "status": "ok", "transport": TRANSPORT_VERSION, "model_contract": "official exact interleaved system prompt", "assistant_identity": ASSISTANT_NAME, "model": MODEL_REPO, "model_revision": MODEL_REVISION, } async def roleforge_audio_socket(websocket: WebSocket, transport_id: str): if not re.fullmatch(r"[0-9a-f]{32}", transport_id): await websocket.close(code=1008) return bridge = _live_sessions.get(transport_id) if bridge is None: await websocket.close(code=1008) return input_queue, output_queue, stop_event, _, counters = bridge await websocket.accept() counters["websocket_connected"] = True sample_rate = 48_000 async def receive_audio(): nonlocal sample_rate while not stop_event.is_set(): message = await websocket.receive() if message.get("type") == "websocket.disconnect": raise WebSocketDisconnect if message.get("text"): payload = json.loads(message["text"]) if payload.get("type") == "hello": candidate = int(payload.get("sample_rate", sample_rate)) if 8_000 <= candidate <= 192_000: sample_rate = candidate elif payload.get("type") == "telemetry": counters["browser_playback_underruns"] = max( 0, int(payload.get("playback_underruns", 0)) ) counters["browser_scheduled_ahead_ms"] = max( 0, min(5_000, int(payload.get("scheduled_ahead_ms", 0))) ) counters["browser_button_to_websocket_ms"] = max( 0, min(300_000, int(payload.get("button_to_websocket_ms", 0))) ) counters["browser_button_to_listening_ms"] = max( 0, min(300_000, int(payload.get("button_to_listening_ms", 0))) ) counters["browser_button_to_first_audio_ms"] = max( 0, min(300_000, int(payload.get("button_to_first_audio_ms", 0))) ) continue raw = message.get("bytes") if not raw or len(raw) % 4: continue audio = np.frombuffer(raw, dtype="