#!/usr/bin/env python3 """Voice Reach V1 Space app. The default local mode is deterministic so package checks stay fast. On a Hugging Face Space, set VCW_MODEL_MODE=real for the model-backed proof path. """ from __future__ import annotations import argparse import base64 import json import os import re import subprocess import tempfile import threading import time import urllib.error import urllib.request import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles import gradio as gr from huggingface_hub import hf_hub_download from pydantic import BaseModel APP_VERSION = "v1-space-local-model-2026-06-15" SPACE_DIR = Path(__file__).resolve().parent V1_DIR = SPACE_DIR.parent if (SPACE_DIR.parent / "onboarding").exists() else SPACE_DIR IDEA_DIR = V1_DIR.parent if V1_DIR != SPACE_DIR else SPACE_DIR REPO_ROOT = V1_DIR.parents[2] if len(V1_DIR.parents) > 2 else V1_DIR def load_local_env_files(paths: tuple[Path, ...]) -> list[str]: loaded: list[str] = [] for path in paths: if not path.exists(): continue for raw_line in path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip().removeprefix("export ").strip() if key and key not in os.environ: os.environ[key] = value.strip().strip("'\"") loaded.append(str(path)) return loaded LOCAL_ENV_FILES = ( IDEA_DIR / ".env.modal.local", ) LOADED_LOCAL_ENV_FILES = load_local_env_files(LOCAL_ENV_FILES) if not os.environ.get("SPACE_ID") else [] PACKET_CANDIDATES = ( V1_DIR / "onboarding" / "metime-to.json", SPACE_DIR / "onboarding" / "metime-to.json", ) PACKET_PATH = next((path for path in PACKET_CANDIDATES if path.exists()), PACKET_CANDIDATES[0]) LEDGER_PATH = Path(os.environ.get("VCW_LEDGER_PATH", SPACE_DIR / "demo-ledger.jsonl")) TRACE_PATH = Path(os.environ.get("VCW_TRACE_PATH", "/tmp/vcw-v1-traces.jsonl")) STATIC_DIR = SPACE_DIR / "static" ASR_MODEL_ID = "nvidia/nemotron-3.5-asr-streaming-0.6b" ASR_MODEL_FILE = "nemotron-3.5-asr-streaming-0.6b.nemo" TEXT_MODEL_ID = "CohereLabs/tiny-aya-fire-GGUF" TEXT_MODEL_FILE = "tiny-aya-fire-q8_0.gguf" TEXT_MODEL_LABEL = "CohereLabs/tiny-aya-fire-GGUF:Q8_0" MODAL_ASR_MODEL_ID = ASR_MODEL_ID MODAL_TEXT_MODEL_ID = TEXT_MODEL_ID MODAL_TEXT_MODEL_LABEL = TEXT_MODEL_LABEL DEFAULT_HF_PERSONAL_BASE_URL = "https://curieous-voice-reach.hf.space" MAX_AUDIO_BYTES = int(os.environ.get("VCW_MAX_AUDIO_BYTES", str(16 * 1024 * 1024))) MAX_AUDIO_SECONDS = int(os.environ.get("VCW_MAX_AUDIO_SECONDS", "45")) SUPPORT_FIELD_CUES = { "order": { "message": ("order id", "order number", "order"), "source": ("order", "order id", "\u0911\u0930\u094d\u0921\u0930"), }, "delivery": { "message": ("delivery status", "delivery address", "delivered", "delivery"), "source": ("delivery", "delivered", "\u0921\u093f\u0932\u093f\u0935\u0930\u0940"), }, "address": { "message": ("delivery address", "address"), "source": ("address", "\u092a\u0924\u093e"), }, "payment": { "message": ("payment", "paid", "transaction"), "source": ("payment", "paid", "transaction", "\u092d\u0941\u0917\u0924\u093e\u0928", "\u092a\u0947\u092e\u0947\u0902\u091f"), }, "refund": { "message": ("refund", "refunded"), "source": ("refund", "refunded", "\u0930\u093f\u092b\u0902\u0921"), }, "phone": { "message": ("phone number", "mobile number", "call me at", "contact me at"), "source": ("phone number", "mobile number", "\u092b\u094b\u0928", "\u092e\u094b\u092c\u093e\u0907\u0932"), }, "email": { "message": ("email address", "email id"), "source": ("email address", "email id", "email", "\u0908\u092e\u0947\u0932"), }, "account": { "message": ("account id", "account number", "account"), "source": ("account", "\u0905\u0915\u093e\u0909\u0902\u091f"), }, "booking": { "message": ("booking id", "booking number", "booking", "book ", "appointment"), "source": ("booking", "book ", "appointment", "\u092c\u0941\u0915\u093f\u0902\u0917"), }, } PRESERVATION_CUES = { "antique": { "source": ("antique", "\u090f\u0902\u091f\u0940\u0915"), "message": ("antique",), }, "pricing": { "source": ("pricing", "price", "\u092a\u094d\u0930\u093e\u0907\u0938"), "message": ("pricing", "price"), }, "availability": { "source": ("availability", "available"), "message": ("availability", "available"), }, } REQUIRED_PACKET_FIELDS = { "schema_version", "site_id", "brand_name", "contact_email", "locale_defaults", "audio_eval_consent", "audio_eval_notice", "retention", "design_tokens", "copy", } REQUIRED_TOKEN_FIELDS = { "font_family", "color_background", "color_surface", "color_surface_subtle", "color_text", "color_text_muted", "color_primary", "color_primary_hover", "color_accent", "color_accent_soft", "color_border", "color_danger", "shadow_soft", "radius_container", "radius_control", "radius_button", } RUN_LEDGER: list[dict[str, Any]] = [] PIPELINE_LOCK = threading.Lock() PROCESS_LOCK = threading.Lock() PIPELINE: "ModelPipeline | None" = None TEXT_ADAPTER_LOCK = threading.Lock() TEXT_ADAPTER: "TinyAyaAdapter | None" = None class PacketError(ValueError): """Raised when the onboarding packet does not match the V1 contract.""" class SendPayload(BaseModel): site_id: str contact_email: str model_message_en: str final_message_en: str trace_id: str simulate: str | None = None def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def detect_app_host() -> str: return "hf_space" if os.environ.get("SPACE_ID") or os.environ.get("SPACE_HOST") else "local" def lifecycle_stage() -> str: return "staging" if detect_app_host() == "hf_space" else "testing" def model_mode() -> str: forced = os.environ.get("VCW_MODEL_MODE", "").strip().lower() if forced in {"real", "deterministic"}: return forced if os.environ.get("VCW_FORCE_DETERMINISTIC", "").strip().lower() in {"1", "true", "yes"}: return "deterministic" return "real" if detect_app_host() == "hf_space" else "deterministic" def runtime_switch_allowed() -> bool: return os.environ.get("VCW_ALLOW_RUNTIME_SWITCH", "").strip().lower() in {"1", "true", "yes", "on"} def selected_model_runtime(override: Any = None) -> str: requested = str(override or "").strip().lower() aliases = { "hf_hackathon_space": "hf_space", "hf_personal": "hf_personal_space", "deterministic": "none", } requested = aliases.get(requested, requested) if requested: if requested == "hf_space": return "hf_space" if requested in {"hf_personal_space", "modal"} and runtime_switch_allowed(): return requested if requested in {"hf_personal_space", "modal"}: raise HTTPException(status_code=403, detail="Model runtime switch is disabled") raise HTTPException(status_code=400, detail="Unsupported model runtime") runtime = os.environ.get("VCW_MODEL_RUNTIME", "").strip().lower() runtime = aliases.get(runtime, runtime) if runtime in {"modal", "hf_personal_space", "hf_space", "none"}: return runtime if model_mode() == "deterministic": return "none" return "hf_space" def runtime_axes(runtime: str) -> dict[str, Any]: if runtime == "modal": return { "model_backend": "modal_http", "inference_engine": "nemo+llama.cpp", "model_artifact_format": "nemo_archive+gguf", "quantization": "unquantized+q8_0", "asr_model_id": os.environ.get("APP_MODAL_ASR_MODEL_ID", MODAL_ASR_MODEL_ID), "text_model_id": os.environ.get("APP_MODAL_MODEL_ID", MODAL_TEXT_MODEL_LABEL), "fallback_used": False, } if runtime == "hf_personal_space": return { "model_backend": "hf_space_proxy", "inference_engine": "remote_space_api", "model_artifact_format": "remote_space", "quantization": "remote", "asr_model_id": ASR_MODEL_ID, "text_model_id": TEXT_MODEL_LABEL, "fallback_used": False, } if runtime == "hf_space": return { "model_backend": "nemo+llama.cpp", "inference_engine": "nemo+llama.cpp", "model_artifact_format": "nemo_archive+gguf", "quantization": "unknown+q8_0", "asr_model_id": ASR_MODEL_ID, "text_model_id": TEXT_MODEL_LABEL, "fallback_used": False, } return { "model_backend": "template", "inference_engine": "none", "model_artifact_format": "none", "quantization": "none", "asr_model_id": ASR_MODEL_ID, "text_model_id": TEXT_MODEL_LABEL, "fallback_used": True, } def runtime_availability(runtime: str) -> dict[str, Any]: if runtime == "none": return {"available": True, "reason": None} if runtime == "hf_space": if model_mode() == "deterministic": return { "available": False, "reason": "HF hackathon space is not available in local deterministic mode. Set VCW_MODEL_MODE=real on model-backed hardware.", } return {"available": True, "reason": None} if runtime == "hf_personal_space": if not runtime_switch_allowed(): return {"available": False, "reason": "Runtime switching is disabled for this Space."} if not hf_personal_base_url(): return {"available": False, "reason": "HF personal space URL is not configured."} return {"available": True, "reason": None} if runtime == "modal": if not runtime_switch_allowed(): return {"available": False, "reason": "Runtime switching is disabled for this Space."} if not modal_base_url(): return {"available": False, "reason": "Modal endpoint is not configured."} if not modal_auth_token(): return {"available": False, "reason": "Modal auth token is not configured."} return {"available": True, "reason": None} return {"available": False, "reason": "Unsupported model runtime."} def runtime_controls(default_runtime: str) -> dict[str, Any]: allow_switch = runtime_switch_allowed() selected = default_runtime if default_runtime in {"hf_space", "hf_personal_space", "modal"} else "hf_space" hf_space_availability = runtime_availability("hf_space") hf_personal_availability = runtime_availability("hf_personal_space") modal_availability = runtime_availability("modal") return { "label": "Model Runtime", "allow_switch": allow_switch, "selected": selected, "options": [ { "value": "hf_space", "label": "HF hackathon space", "note": "credit issue", "enabled": True, "available": hf_space_availability["available"], "disabled_reason": hf_space_availability["reason"], }, { "value": "hf_personal_space", "label": "HF personal space", "note": "test runtime", "enabled": allow_switch, "available": allow_switch and hf_personal_availability["available"], "disabled_reason": hf_personal_availability["reason"] if allow_switch else "Enable VCW_ALLOW_RUNTIME_SWITCH=1 to test this runtime.", }, { "value": "modal", "label": "Modal", "note": "warm GPU" if modal_availability["available"] else "test runtime", "enabled": allow_switch, "available": allow_switch and modal_availability["available"], "disabled_reason": modal_availability["reason"] if allow_switch else "Enable VCW_ALLOW_RUNTIME_SWITCH=1 to test this runtime.", }, ], } def hf_token() -> str | None: token = ( os.environ.get("HF_TOKEN") or os.environ.get("HF_TOKEN_2") or os.environ.get("HF_TOKEN_1") or os.environ.get("HUGGINGFACE_HUB_TOKEN") ) if token and not os.environ.get("HF_TOKEN"): os.environ["HF_TOKEN"] = token return token def load_packet() -> dict[str, Any]: packet = json.loads(PACKET_PATH.read_text(encoding="utf-8")) validate_packet(packet) return packet def validate_packet(packet: dict[str, Any]) -> None: missing = sorted(REQUIRED_PACKET_FIELDS - packet.keys()) if missing: raise PacketError(f"Packet missing required fields: {', '.join(missing)}") defaults = packet.get("locale_defaults", {}) modes = defaults.get("speech_modes") if modes != ["hindi", "hinglish"]: raise PacketError("Packet speech modes must be exactly ['hindi', 'hinglish']") if defaults.get("default_speech_mode") not in modes: raise PacketError("Packet default speech mode must be one of the speech modes") if defaults.get("target_output_language") != "en": raise PacketError("Packet target output language must be 'en'") if packet.get("site_id") != "metime-to": raise PacketError("V1 expects site_id='metime-to'") if packet.get("contact_email") != "founders@metime.to": raise PacketError("V1 expects contact_email='founders@metime.to'") tokens = packet.get("design_tokens", {}) missing_tokens = sorted(REQUIRED_TOKEN_FIELDS - tokens.keys()) if missing_tokens: raise PacketError(f"Packet missing design tokens: {', '.join(missing_tokens)}") copy = packet.get("copy", {}) for key in ("heading", "subheading", "record_cta", "stop_cta", "send_cta", "success"): if not copy.get(key): raise PacketError(f"Packet copy missing '{key}'") def asr_hint_for_mode(speech_mode: str) -> str: return "hi-IN" if speech_mode in {"hindi", "hinglish"} else "hi-IN" def public_packet(packet: dict[str, Any]) -> dict[str, Any]: mode = model_mode() runtime = selected_model_runtime() axes = runtime_axes(runtime) return { **packet, "runtime_controls": runtime_controls(runtime), "provenance": { "packet_path": "product/5-idea-voice-contact-widget/v1/onboarding/metime-to.json", "app_version": APP_VERSION, "lifecycle_stage": lifecycle_stage(), "app_host": detect_app_host(), "model_runtime": runtime, "model_backend": axes["model_backend"], "inference_engine": axes["inference_engine"], "model_artifact_format": axes["model_artifact_format"], "quantization": axes["quantization"], "mock_mode": "deterministic" if mode == "deterministic" else "off", "fallback_used": axes["fallback_used"], "asr_model_id": axes["asr_model_id"], "text_model_id": axes["text_model_id"], "concurrency": 1, }, } def deterministic_transcript(speech_mode: str) -> str: if speech_mode == "hindi": return "Namaste, mujhe metime.to ke baare mein founders se baat karni hai. Kripya mujhe onboarding aur pricing details bhej dijiye." return "Hi, main metime.to ke baare mein founders se connect karna chahta hoon. Please onboarding aur pricing details share kar dijiye." def deterministic_message(speech_mode: str, transcript: str) -> str: if speech_mode == "hindi": body = "I would like to connect with the founders to learn more about metime.to, including onboarding and pricing details." else: body = "I would like to connect with the founders and learn more about metime.to, especially onboarding and pricing details." return f"Hello metime.to team,\n\n{body}\n\nOriginal voice note summary: {transcript}\n\nThank you." def deterministic_model_json(speech_mode: str, transcript: str) -> dict[str, Any]: return { "message_en": deterministic_message(speech_mode, transcript), "confidence": 0.72, "needs_edit": False, "notes": [], "guardrail_flags": [], } def unsupported_support_fields(transcript: str, message: str) -> list[str]: transcript_lower = transcript.lower() message_lower = message.lower() unsupported: list[str] = [] for field, cues in SUPPORT_FIELD_CUES.items(): message_mentions = any(cue in message_lower for cue in cues["message"]) if field == "phone" and re.search(r"\b\d{7,}\b", message_lower): message_mentions = True source_mentions = any(cue in transcript_lower for cue in cues["source"]) if message_mentions and not source_mentions: unsupported.append(field) return unsupported def missing_preservation_cues(transcript: str, message: str) -> list[str]: transcript_lower = transcript.lower() message_lower = message.lower() missing: list[str] = [] for cue, patterns in PRESERVATION_CUES.items(): source_mentions = any(pattern in transcript_lower for pattern in patterns["source"]) message_mentions = any(pattern in message_lower for pattern in patterns["message"]) if source_mentions and not message_mentions: missing.append(cue) return missing def normalize_text_output(data: dict[str, Any], transcript: str) -> dict[str, Any]: message = str(data.get("message_en") or "").strip() if not message: raise ValueError("tiny_aya_empty_message") confidence_raw = data.get("confidence", 0.5) try: confidence = max(0.0, min(1.0, float(confidence_raw))) except (TypeError, ValueError): confidence = 0.5 notes_raw = data.get("notes") or [] if isinstance(notes_raw, str): notes = [notes_raw] else: notes = [str(note).strip() for note in list(notes_raw) if str(note).strip()] guardrail_flags = [str(flag).strip() for flag in list(data.get("guardrail_flags") or []) if str(flag).strip()] unsupported = unsupported_support_fields(transcript, message) if unsupported: guardrail_flags.append("unsupported_support_fields:" + ",".join(unsupported)) notes.append("draft_mentions_support_fields_not_present_in_transcript") missing_cues = missing_preservation_cues(transcript, message) if missing_cues: guardrail_flags.append("missing_preservation_cues:" + ",".join(missing_cues)) notes.append("draft_may_not_preserve_source_topic") needs_edit = bool(data.get("needs_edit")) if unsupported or missing_cues or confidence < 0.55: needs_edit = True return { "message_en": message, "confidence": confidence, "needs_edit": needs_edit, "notes": sorted(set(notes)), "guardrail_flags": sorted(set(guardrail_flags)), } def tiny_aya_messages(packet: dict[str, Any], transcript: str, speech_mode: str, target_lang: str) -> list[dict[str, str]]: payload = { "site_id": packet["site_id"], "brand_name": packet["brand_name"], "asr_transcript": transcript, "speech_mode": speech_mode, "asr_language_hint": target_lang, "target_output_language": "en", "task": "Rewrite or translate only the user's transcript into a concise English contact-form message.", "output_schema": { "message_en": "English draft preserving only the transcript's meaning", "confidence": "number from 0.0 to 1.0", "needs_edit": "boolean", "notes": "short array of review notes; empty when clean", }, } system = ( "You are a literal Hindi/Hinglish to English rewriting assistant for a website contact widget. " "Preserve the user's meaning. Do not act as customer support. Do not invent facts, fields, " "names, phone numbers, emails, order IDs, payment status, delivery outcomes, addresses, links, " "bookings, or account details. If the transcript is unrelated, unclear, or not a request, " "write a faithful English version and set needs_edit=true. Return only strict JSON." ) user = ( "Create the JSON response for this input. Use only information present in asr_transcript.\n" f"{json.dumps(payload, ensure_ascii=True, sort_keys=True)}" ) return [{"role": "system", "content": system}, {"role": "user", "content": user}] def tiny_aya_raw_prompt(messages: list[dict[str, str]]) -> str: return ( f"System:\n{messages[0]['content']}\n\n" f"User:\n{messages[1]['content']}\n\n" "JSON:" ) def append_trace(record: dict[str, Any]) -> None: redacted = json.loads(json.dumps(record, ensure_ascii=True)) RUN_LEDGER.insert(0, redacted) del RUN_LEDGER[50:] try: TRACE_PATH.parent.mkdir(parents=True, exist_ok=True) with TRACE_PATH.open("a", encoding="utf-8") as handle: handle.write(json.dumps(redacted, ensure_ascii=True) + "\n") except OSError: pass def build_trace_base(packet: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]: retention = packet.get("retention", {}) speech_mode = str(payload.get("speech_mode") or packet["locale_defaults"]["default_speech_mode"]) return { "trace_id": f"vcw_{uuid.uuid4().hex[:12]}", "created_at": utc_now(), "site_id": packet["site_id"], "speech_mode": speech_mode, "asr_language_hint": asr_hint_for_mode(speech_mode), "audio_eval_consent": bool(packet["audio_eval_consent"]), "visitor_notice_shown": bool(payload.get("visitor_notice_shown")), "raw_audio_retained": bool(retention.get("retain_raw_audio_for_eval")), "retain_transcript_for_eval": bool(retention.get("retain_transcript_for_eval")), "retain_generated_message_for_eval": bool(retention.get("retain_generated_message_for_eval")), "asr_model_id": ASR_MODEL_ID, "text_model_id": TEXT_MODEL_LABEL, "lifecycle_stage": lifecycle_stage(), "app_host": detect_app_host(), } def strip_data_url(value: str) -> str: if "," in value and value[:64].lower().startswith("data:"): return value.split(",", 1)[1] return value def suffix_for_mime(mime: str) -> str: mime = (mime or "").split(";", 1)[0].strip().lower() return { "audio/wav": ".wav", "audio/wave": ".wav", "audio/x-wav": ".wav", "audio/mpeg": ".mp3", "audio/mp3": ".mp3", "audio/webm": ".webm", "audio/ogg": ".ogg", "audio/mp4": ".m4a", }.get(mime, ".wav") def decode_audio_payload(payload: dict[str, Any]) -> tuple[Path | None, int]: audio_b64 = str(payload.get("audio_base64") or "").strip() if not audio_b64: return None, 0 raw = base64.b64decode(strip_data_url(audio_b64), validate=False) if len(raw) > MAX_AUDIO_BYTES: raise HTTPException(status_code=413, detail="Audio payload too large") mime = str(payload.get("audio_mime") or "audio/wav") suffix = suffix_for_mime(mime) source = Path(tempfile.gettempdir()) / f"vcw-audio-{uuid.uuid4().hex}{suffix}" source.write_bytes(raw) if suffix == ".wav": return source, len(raw) wav_path = source.with_suffix(".wav") command = [ "ffmpeg", "-y", "-i", str(source), "-ac", "1", "-ar", "16000", str(wav_path), ] try: subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30) except Exception as exc: raise RuntimeError(f"ffmpeg_audio_convert_failed:{type(exc).__name__}") from exc return wav_path, len(raw) def extract_text(value: Any) -> str: if isinstance(value, str): return value.strip() if isinstance(value, list) and value: return extract_text(value[0]) if isinstance(value, tuple) and value: return extract_text(value[0]) if hasattr(value, "text"): return str(value.text).strip() return str(value).strip() class NemoAsrAdapter: def __init__(self) -> None: started = time.time() hf_token() import torch import nemo.collections.asr as nemo_asr from nemo.collections.asr.parts.utils.streaming_utils import CacheAwareStreamingAudioBuffer self.torch = torch self.StreamingAudioBuffer = CacheAwareStreamingAudioBuffer self.model = nemo_asr.models.ASRModel.from_pretrained(model_name=ASR_MODEL_ID) self.device = "cuda" if torch.cuda.is_available() else "cpu" if hasattr(torch, "set_float32_matmul_precision"): torch.set_float32_matmul_precision("high") self.model = self.model.to(device=self.device, dtype=torch.float32) self.model.eval() self._configure_streaming_context() self.load_ms = round((time.time() - started) * 1000) def _configure_streaming_context(self) -> None: raw = os.environ.get("VCW_NEMO_ATT_CONTEXT_SIZE", "56,13") try: att_context_size = [int(part.strip()) for part in raw.replace("[", "").replace("]", "").split(",") if part.strip()] except ValueError: att_context_size = [] if len(att_context_size) == 2 and hasattr(self.model.encoder, "set_default_att_context_size"): self.model.encoder.set_default_att_context_size(att_context_size=att_context_size) def transcribe(self, audio_path: Path, target_lang: str) -> tuple[str, int]: started = time.time() if hasattr(self.model, "set_inference_prompt"): self.model.set_inference_prompt(target_lang) decoding = getattr(self.model, "decoding", None) if decoding is not None and hasattr(decoding, "set_strip_lang_tags"): decoding.set_strip_lang_tags(True, lang_tag_pattern=r"<[a-z]{2,3}-[A-Z]{2}>") streaming_buffer = self.StreamingAudioBuffer( model=self.model, online_normalization=False, pad_and_drop_preencoded=False, ) streaming_buffer.append_audio_file(str(audio_path), stream_id=-1) text = extract_text(self._perform_streaming(streaming_buffer)) if not text: raise RuntimeError("nemo_streaming_returned_empty_transcript") return text, round((time.time() - started) * 1000) def _perform_streaming(self, streaming_buffer: Any) -> Any: batch_size = len(streaming_buffer.streams_length) cache_last_channel, cache_last_time, cache_last_channel_len = self.model.encoder.get_initial_cache_state( batch_size=batch_size ) previous_hypotheses = None pred_out_stream = None transcribed_texts: Any = [] with self.torch.inference_mode(): for step_num, (chunk_audio, chunk_lengths) in enumerate(streaming_buffer): chunk_audio = chunk_audio.to(self.torch.float32) drop_extra = 0 if step_num != 0: drop_extra = self.model.encoder.streaming_cfg.drop_extra_pre_encoded ( pred_out_stream, transcribed_texts, cache_last_channel, cache_last_time, cache_last_channel_len, previous_hypotheses, ) = self.model.conformer_stream_step( processed_signal=chunk_audio, processed_signal_length=chunk_lengths, cache_last_channel=cache_last_channel, cache_last_time=cache_last_time, cache_last_channel_len=cache_last_channel_len, keep_all_outputs=streaming_buffer.is_buffer_empty(), previous_hypotheses=previous_hypotheses, previous_pred_out=pred_out_stream, drop_extra_pre_encoded=drop_extra, return_transcription=True, ) return transcribed_texts class TinyAyaAdapter: def __init__(self) -> None: started = time.time() from llama_cpp import Llama model_path = hf_hub_download( repo_id=TEXT_MODEL_ID, filename=TEXT_MODEL_FILE, repo_type="model", token=hf_token(), cache_dir=os.environ.get("HF_HOME") or os.environ.get("HF_MODEL_CACHE_DIR"), ) n_gpu_layers = int(os.environ.get("VCW_LLAMA_N_GPU_LAYERS", "0")) self.llm = Llama( model_path=model_path, n_ctx=int(os.environ.get("VCW_LLAMA_N_CTX", "2048")), n_threads=int(os.environ.get("VCW_LLAMA_THREADS", "4")), n_gpu_layers=n_gpu_layers, verbose=os.environ.get("VCW_LLAMA_VERBOSE", "0") in {"1", "true", "yes"}, ) self.load_ms = round((time.time() - started) * 1000) def generate(self, packet: dict[str, Any], transcript: str, speech_mode: str, target_lang: str) -> tuple[dict[str, Any], int, str]: started = time.time() messages = tiny_aya_messages(packet, transcript, speech_mode, target_lang) max_tokens = int(os.environ.get("VCW_LLAMA_MAX_TOKENS", "220")) temperature = float(os.environ.get("VCW_LLAMA_TEMPERATURE", "0.1")) top_p = float(os.environ.get("VCW_LLAMA_TOP_P", "0.95")) use_chat_completion = os.environ.get("VCW_TINY_AYA_CHAT_COMPLETION", "1").strip().lower() in {"1", "true", "yes"} raw = "" try: if not use_chat_completion: raise RuntimeError("chat_completion_disabled") result = self.llm.create_chat_completion( messages=messages, max_tokens=max_tokens, temperature=temperature, top_p=top_p, ) raw = str(result["choices"][0]["message"]["content"]).strip() except Exception: result = self.llm( tiny_aya_raw_prompt(messages), max_tokens=max_tokens, temperature=temperature, top_p=top_p, stop=["\n\nUser:", "\n\nSystem:", ""], ) raw = str(result["choices"][0]["text"]).strip() try: parsed = parse_model_json(raw, transcript) except Exception as exc: raise ValueError(f"{exc}; raw_prefix={raw[:500]!r}") from exc return parsed, round((time.time() - started) * 1000), raw def parse_model_json(raw: str, transcript: str = "") -> dict[str, Any]: text = raw.strip() try: data = json.loads(text) except json.JSONDecodeError: match = re.search(r"\{.*\}", text, re.DOTALL) if not match: raise ValueError("tiny_aya_output_not_json") data = json.loads(match.group(0)) for wrapper_key in ("output", "response", "result"): if isinstance(data.get(wrapper_key), dict): data = data[wrapper_key] break if isinstance(data.get("output_schema"), dict): nested = data["output_schema"] required = {"message_en", "confidence", "needs_edit"} if required <= set(nested): data = nested if isinstance(data.get("message"), dict) and isinstance(data["message"].get("content"), str): return parse_model_json(data["message"]["content"], transcript) if isinstance(data.get("content"), str): return parse_model_json(data["content"], transcript) required = {"message_en", "confidence", "needs_edit"} missing = required - set(data) if missing: raise ValueError(f"tiny_aya_json_missing:{','.join(sorted(missing))}") return normalize_text_output(data, transcript) class ModelPipeline: def __init__(self) -> None: self.asr = NemoAsrAdapter() self.text = TinyAyaAdapter() def run(self, packet: dict[str, Any], payload: dict[str, Any], audio_path: Path) -> dict[str, Any]: trace = build_trace_base(packet, payload) speech_mode = trace["speech_mode"] target_lang = trace["asr_language_hint"] total_started = time.time() transcript, asr_ms = self.asr.transcribe(audio_path, target_lang) if not transcript: raise RuntimeError("nemo_empty_transcript") text_json, text_ms, raw_text = self.text.generate(packet, transcript, speech_mode, target_lang) trace.update( { "model_runtime": "hf_space", "model_backend": "nemo+llama.cpp", "inference_engine": "nemo+llama.cpp", "model_artifact_format": "nemo_archive+gguf", "quantization": "unknown+q8_0", "model_id": f"{ASR_MODEL_ID}+{TEXT_MODEL_LABEL}", "fallback_used": False, "fallback_reason": None, "timings_ms": { "asr_load": self.asr.load_ms, "text_load": self.text.load_ms, "asr": asr_ms, "text": text_ms, "total": round((time.time() - total_started) * 1000), }, "asr": { "model_id": ASR_MODEL_ID, "backend": "nemo", "inference_engine": "nemo", "model_artifact_format": "nemo_archive", "quantization": "unknown", "target_lang": target_lang, "fallback_used": False, "transcript": transcript, }, "text": { "model_id": TEXT_MODEL_LABEL, "backend": "llama.cpp", "inference_engine": "llama.cpp", "model_artifact_format": "gguf", "quantization": "q8_0", "fallback_used": False, "raw_output": raw_text[:2000], "output": text_json, }, } ) return { "transcript": transcript, "model_message_en": text_json["message_en"], "text": text_json, "trace": trace, } def get_pipeline() -> ModelPipeline: global PIPELINE if PIPELINE is None: with PIPELINE_LOCK: if PIPELINE is None: PIPELINE = ModelPipeline() return PIPELINE def get_text_adapter() -> TinyAyaAdapter: global TEXT_ADAPTER if PIPELINE is not None: return PIPELINE.text if TEXT_ADAPTER is None: with TEXT_ADAPTER_LOCK: if TEXT_ADAPTER is None: TEXT_ADAPTER = TinyAyaAdapter() return TEXT_ADAPTER def modal_timeout_seconds() -> float: return float(os.environ.get("APP_MODAL_TIMEOUT_SECONDS", "180")) def modal_base_url() -> str: return os.environ.get("APP_MODAL_BASE_URL", "").strip().rstrip("/") def modal_auth_token() -> str: return os.environ.get("APP_MODAL_AUTH_TOKEN", "").strip() def modal_payload(packet: dict[str, Any], payload: dict[str, Any], trace: dict[str, Any], *, transcript: str = "") -> dict[str, Any]: audio_b64 = str(payload.get("audio_base64") or "").strip() return { "trace_id": trace["trace_id"], "site_id": packet["site_id"], "speech_mode": trace["speech_mode"], "language_hint": trace["asr_language_hint"], "asr_model_id": os.environ.get("APP_MODAL_ASR_MODEL_ID", MODAL_ASR_MODEL_ID), "text_model_id": os.environ.get("APP_MODAL_MODEL_ID", MODAL_TEXT_MODEL_ID), "transcript": transcript, "audio": { "size_bytes": int(payload.get("audio_size_bytes") or 0), "duration_ms": int(payload.get("duration_ms") or payload.get("audio_duration_ms") or 0), "content_type": str(payload.get("audio_mime") or "audio/wav"), "b64": strip_data_url(audio_b64), }, } def call_modal(payload: dict[str, Any]) -> tuple[dict[str, Any], int]: base_url = modal_base_url() token = modal_auth_token() if not base_url: raise RuntimeError("APP_MODAL_BASE_URL is not set") if not token: raise RuntimeError("APP_MODAL_AUTH_TOKEN is not set") started = time.time() request = urllib.request.Request( base_url, data=json.dumps(payload).encode("utf-8"), headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", }, method="POST", ) try: with urllib.request.urlopen(request, timeout=modal_timeout_seconds()) as response: body = response.read().decode("utf-8", errors="replace") parsed = json.loads(body) if not isinstance(parsed, dict): raise RuntimeError("modal_response_not_json_object") return parsed, round((time.time() - started) * 1000) except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace")[:500] raise RuntimeError(f"modal_http_{exc.code}:{body}") from exc def hf_personal_base_url() -> str: return os.environ.get("APP_HF_PERSONAL_BASE_URL", DEFAULT_HF_PERSONAL_BASE_URL).strip().rstrip("/") def hf_personal_auth_token() -> str: return ( os.environ.get("APP_HF_PERSONAL_TOKEN", "").strip() or os.environ.get("HF_TOKEN_2", "").strip() or os.environ.get("HF_TOKEN", "").strip() ) def call_hf_personal(path: str, payload: dict[str, Any]) -> tuple[dict[str, Any], int]: base_url = hf_personal_base_url() if not base_url: raise RuntimeError("APP_HF_PERSONAL_BASE_URL is not set") token = hf_personal_auth_token() headers = {"Content-Type": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" proxied_payload = dict(payload) proxied_payload.pop("model_runtime", None) started = time.time() request = urllib.request.Request( f"{base_url}{path}", data=json.dumps(proxied_payload).encode("utf-8"), headers=headers, method="POST", ) try: with urllib.request.urlopen(request, timeout=modal_timeout_seconds()) as response: body = response.read().decode("utf-8", errors="replace") parsed = json.loads(body) if not isinstance(parsed, dict): raise RuntimeError("hf_personal_response_not_json_object") return parsed, round((time.time() - started) * 1000) except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace")[:500] raise RuntimeError(f"hf_personal_http_{exc.code}:{body}") from exc def hf_personal_result_to_response( local_trace: dict[str, Any], result: dict[str, Any], latency_ms: int, *, text_only: bool, ) -> dict[str, Any]: remote_trace = result.get("trace") if isinstance(result.get("trace"), dict) else {} transcript = str(result.get("transcript") or remote_trace.get("transcript") or "").strip() text_json = result.get("text") if isinstance(result.get("text"), dict) else {} if not text_json: text_json = ((remote_trace.get("text") or {}).get("output") or {}) if isinstance(remote_trace.get("text"), dict) else {} text_json = normalize_text_output(text_json, transcript) local_trace.update( { "text_only_smoke": text_only, "model_runtime": "hf_personal_space", "model_backend": "hf_space_proxy", "inference_engine": "remote_space_api", "model_artifact_format": "remote_space", "quantization": "remote", "model_id": f"{hf_personal_base_url()}:{'text-smoke' if text_only else 'process'}", "fallback_used": bool(remote_trace.get("fallback_used")), "fallback_reason": remote_trace.get("fallback_reason"), "blocker": remote_trace.get("blocker"), "timings_ms": { "hf_personal_http": latency_ms, "remote_total": (remote_trace.get("timings_ms") or {}).get("total"), }, "remote_trace_id": remote_trace.get("trace_id"), "asr": remote_trace.get("asr") or { "model_id": "supplied-transcript" if text_only else ASR_MODEL_ID, "backend": "hf_space_proxy", "inference_engine": "remote_space_api", "model_artifact_format": "remote_space", "quantization": "remote", "target_lang": local_trace["asr_language_hint"], "fallback_used": False, "transcript": transcript, }, "text": remote_trace.get("text") or { "model_id": TEXT_MODEL_LABEL, "backend": "hf_space_proxy", "inference_engine": "remote_space_api", "model_artifact_format": "remote_space", "quantization": "remote", "fallback_used": bool(remote_trace.get("fallback_used")), "output": text_json, }, } ) return { "transcript": transcript, "model_message_en": text_json["message_en"], "text": text_json, "trace": local_trace, } def modal_result_to_response( packet: dict[str, Any], payload: dict[str, Any], trace: dict[str, Any], modal_result: dict[str, Any], modal_ms: int, *, text_only: bool, ) -> dict[str, Any]: asr = modal_result.get("asr") if isinstance(modal_result.get("asr"), dict) else {} text = modal_result.get("text") if isinstance(modal_result.get("text"), dict) else {} raw_output = str(text.get("raw_output") or "") transcript = str(modal_result.get("transcript") or (asr.get("output") or {}).get("transcript") or "").strip() text_output = text.get("output") if isinstance(text.get("output"), dict) else {} text_json = normalize_text_output(text_output, transcript) fallback_used = bool(modal_result.get("fallback_used")) or bool(asr.get("fallback_used")) or bool(text.get("fallback_used")) trace.update( { "text_only_smoke": text_only, "model_runtime": "modal", "model_backend": "modal_http", "inference_engine": modal_result.get("inference_engine") or "nemo+llama.cpp", "model_artifact_format": modal_result.get("model_artifact_format") or "nemo_archive+gguf", "quantization": modal_result.get("quantization") or "unquantized+q8_0", "model_id": modal_result.get("model_id") or f"{MODAL_ASR_MODEL_ID}+{MODAL_TEXT_MODEL_LABEL}", "fallback_used": fallback_used, "fallback_reason": modal_result.get("fallback_reason"), "blocker": modal_result.get("error"), "timings_ms": { "modal_http": modal_ms, "modal_total": modal_result.get("latency_ms"), "asr": asr.get("latency_ms"), "text": text.get("latency_ms"), }, "asr": { "model_id": asr.get("model_id") or ("supplied-transcript" if text_only else MODAL_ASR_MODEL_ID), "backend": asr.get("backend") or ("none" if text_only else "modal_http"), "inference_engine": asr.get("inference_engine") or ("none" if text_only else "nemo"), "model_artifact_format": asr.get("model_artifact_format") or ("none" if text_only else "nemo_archive"), "quantization": asr.get("quantization") or ("none" if text_only else "unquantized"), "target_lang": trace["asr_language_hint"], "fallback_used": bool(asr.get("fallback_used")), "transcript": transcript, }, "text": { "model_id": text.get("model_id") or MODAL_TEXT_MODEL_LABEL, "backend": text.get("backend") or "modal_http", "inference_engine": text.get("inference_engine") or "llama.cpp", "model_artifact_format": text.get("model_artifact_format") or "gguf", "quantization": text.get("quantization") or "q8_0", "fallback_used": bool(text.get("fallback_used")), "raw_output": raw_output[:2000], "output": text_json, }, } ) if fallback_used: trace["fallback_reason"] = trace["fallback_reason"] or "modal_partial_or_failed" return { "transcript": transcript, "model_message_en": text_json["message_en"], "text": text_json, "trace": trace, } def deterministic_process(packet: dict[str, Any], payload: dict[str, Any], reason: str) -> dict[str, Any]: trace = build_trace_base(packet, payload) speech_mode = trace["speech_mode"] transcript = str(payload.get("transcript") or deterministic_transcript(speech_mode)).strip() text_json = deterministic_model_json(speech_mode, transcript) trace.update( { "model_runtime": "none", "model_backend": "template", "inference_engine": "none", "model_artifact_format": "none", "quantization": "none", "model_id": "deterministic-v1-template", "mock_mode": "deterministic", "fallback_used": True, "fallback_reason": reason, "timings_ms": {"prepare": 10, "mock_asr": 1, "mock_text": 1, "total": 12}, "asr": { "model_id": ASR_MODEL_ID, "backend": "template", "inference_engine": "none", "model_artifact_format": "none", "quantization": "none", "target_lang": trace["asr_language_hint"], "fallback_used": True, "transcript": transcript, }, "text": { "model_id": TEXT_MODEL_LABEL, "backend": "template", "inference_engine": "none", "model_artifact_format": "none", "quantization": "none", "fallback_used": True, "output": text_json, }, } ) return {"transcript": transcript, "model_message_en": text_json["message_en"], "text": text_json, "trace": trace} def runtime_unavailable_response(packet: dict[str, Any], payload: dict[str, Any], runtime: str, *, text_only: bool) -> JSONResponse: availability = runtime_availability(runtime) reason = availability["reason"] or "Selected model runtime is not available." trace = build_trace_base(packet, payload) axes = runtime_axes(runtime) trace.update( { "text_only_smoke": text_only, "model_runtime": runtime, "model_backend": axes["model_backend"], "inference_engine": axes["inference_engine"], "model_artifact_format": axes["model_artifact_format"], "quantization": axes["quantization"], "model_id": f"{axes['asr_model_id']}+{axes['text_model_id']}", "fallback_used": False, "fallback_reason": None, "blocker": reason, "timings_ms": {"total": 0}, } ) append_trace(trace) return JSONResponse( { "error": "Selected model runtime is not available; deterministic output was not substituted.", "blocker": reason, "trace": trace, }, status_code=424, ) app = FastAPI(title="Voice Reach V1", version=APP_VERSION) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["GET", "POST", "OPTIONS"], allow_headers=["*"], ) app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") @app.get("/") def index() -> FileResponse: return FileResponse(SPACE_DIR / "index.html") @app.get("/health") def health() -> dict[str, Any]: mode = model_mode() runtime = selected_model_runtime() axes = runtime_axes(runtime) availability = runtime_availability(runtime) return { "ok": True, "app": "voice-reach-v1", "version": APP_VERSION, "lifecycle_stage": lifecycle_stage(), "app_host": detect_app_host(), "model_runtime": runtime, "model_backend": axes["model_backend"], "inference_engine": axes["inference_engine"], "model_artifact_format": axes["model_artifact_format"], "quantization": axes["quantization"], "model_mode": mode, "runtime_available": availability["available"], "runtime_unavailable_reason": availability["reason"], "fallback_used": axes["fallback_used"], "asr_model_id": axes["asr_model_id"], "text_model_id": axes["text_model_id"], "modal_configured": bool(modal_base_url()) if runtime == "modal" else None, "hf_personal_configured": bool(hf_personal_base_url()) if runtime == "hf_personal_space" else None, "runtime_switch_allowed": runtime_switch_allowed(), "local_env_files_loaded": [Path(path).name for path in LOADED_LOCAL_ENV_FILES], "concurrency": 1, "gradio_status_path": "/gradio", "trace_path": str(TRACE_PATH), } @app.get("/api/packet") def api_packet() -> dict[str, Any]: return public_packet(load_packet()) @app.get("/api/traces") def api_traces() -> dict[str, Any]: return {"items": RUN_LEDGER[:20]} @app.post("/api/process") async def api_process(request: Request) -> JSONResponse: packet = load_packet() payload = await request.json() speech_mode = payload.get("speech_mode", packet["locale_defaults"]["default_speech_mode"]) if speech_mode not in {"hindi", "hinglish"}: raise HTTPException(status_code=400, detail="Unsupported speech mode") runtime = selected_model_runtime(payload.get("model_runtime")) duration_ms = int(payload.get("duration_ms") or payload.get("audio_duration_ms") or 0) if duration_ms > MAX_AUDIO_SECONDS * 1000: raise HTTPException(status_code=400, detail=f"Audio exceeds {MAX_AUDIO_SECONDS} second V1 smoke limit") if payload.get("simulate") == "asr_failure": raise HTTPException(status_code=422, detail="ASR failed in deterministic test mode") if payload.get("simulate") == "text_failure": raise HTTPException(status_code=424, detail="Text generation failed in deterministic test mode") with PROCESS_LOCK: if model_mode() == "deterministic" and runtime == "none": result = deterministic_process(packet, payload, "deterministic_mode") append_trace(result["trace"]) return JSONResponse(result) if not runtime_availability(runtime)["available"]: return runtime_unavailable_response(packet, payload, runtime, text_only=False) try: if runtime == "modal": trace = build_trace_base(packet, payload) modal_result, modal_ms = call_modal(modal_payload(packet, payload, trace)) result = modal_result_to_response(packet, payload, trace, modal_result, modal_ms, text_only=False) result["trace"]["audio_size_bytes"] = int(payload.get("audio_size_bytes") or 0) result["trace"]["audio_duration_ms"] = duration_ms elif runtime == "hf_personal_space": trace = build_trace_base(packet, payload) remote_result, remote_ms = call_hf_personal("/api/process", payload) result = hf_personal_result_to_response(trace, remote_result, remote_ms, text_only=False) result["trace"]["audio_size_bytes"] = int(payload.get("audio_size_bytes") or 0) result["trace"]["audio_duration_ms"] = duration_ms elif runtime == "hf_space": audio_path, audio_size = decode_audio_payload(payload) if audio_path is None or audio_size <= 0: raise ValueError("real_model_mode_requires_audio_base64") result = get_pipeline().run(packet, payload, audio_path) result["trace"]["audio_size_bytes"] = audio_size result["trace"]["audio_duration_ms"] = duration_ms else: return runtime_unavailable_response(packet, payload, runtime, text_only=False) append_trace(result["trace"]) return JSONResponse(result) except Exception as exc: trace = build_trace_base(packet, payload) axes = runtime_axes(runtime) trace.update( { "model_runtime": runtime, "model_backend": axes["model_backend"], "inference_engine": axes["inference_engine"], "model_artifact_format": axes["model_artifact_format"], "quantization": axes["quantization"], "model_id": f"{axes['asr_model_id']}+{axes['text_model_id']}", "fallback_used": True, "fallback_reason": f"real_model_error:{type(exc).__name__}", "blocker": str(exc), "timings_ms": {}, } ) append_trace(trace) return JSONResponse( { "error": "Real model path failed; deterministic output was not substituted.", "blocker": str(exc), "trace": trace, }, status_code=424, ) @app.post("/api/text-smoke") async def api_text_smoke(request: Request) -> JSONResponse: packet = load_packet() payload = await request.json() speech_mode = payload.get("speech_mode", packet["locale_defaults"]["default_speech_mode"]) if speech_mode not in {"hindi", "hinglish"}: raise HTTPException(status_code=400, detail="Unsupported speech mode") runtime = selected_model_runtime(payload.get("model_runtime")) transcript = str(payload.get("transcript") or "").strip() if not transcript: raise HTTPException(status_code=400, detail="transcript is required") trace = build_trace_base(packet, payload) target_lang = trace["asr_language_hint"] if model_mode() == "deterministic" and runtime == "none": text_json = deterministic_model_json(speech_mode, transcript) trace.update( { "text_only_smoke": True, "model_runtime": "none", "model_backend": "template", "inference_engine": "none", "model_artifact_format": "none", "quantization": "none", "model_id": "deterministic-v1-template", "fallback_used": True, "fallback_reason": "deterministic_mode", "timings_ms": {"mock_text": 1, "total": 1}, "asr": { "model_id": "supplied-transcript", "backend": "none", "inference_engine": "none", "model_artifact_format": "none", "quantization": "none", "target_lang": target_lang, "fallback_used": False, "transcript": transcript, }, "text": { "model_id": TEXT_MODEL_LABEL, "backend": "template", "inference_engine": "none", "model_artifact_format": "none", "quantization": "none", "fallback_used": True, "output": text_json, }, } ) append_trace(trace) return JSONResponse({"transcript": transcript, "model_message_en": text_json["message_en"], "text": text_json, "trace": trace}) if not runtime_availability(runtime)["available"]: return runtime_unavailable_response(packet, payload, runtime, text_only=True) started = time.time() try: if runtime == "modal": modal_result, modal_ms = call_modal(modal_payload(packet, payload, trace, transcript=transcript)) result = modal_result_to_response(packet, payload, trace, modal_result, modal_ms, text_only=True) append_trace(result["trace"]) return JSONResponse(result) if runtime == "hf_personal_space": remote_result, remote_ms = call_hf_personal("/api/text-smoke", payload) result = hf_personal_result_to_response(trace, remote_result, remote_ms, text_only=True) append_trace(result["trace"]) return JSONResponse(result) text_json, text_ms, raw_text = get_text_adapter().generate(packet, transcript, speech_mode, target_lang) trace.update( { "text_only_smoke": True, "model_runtime": "hf_space", "model_backend": "llama.cpp", "inference_engine": "llama.cpp", "model_artifact_format": "gguf", "quantization": "q8_0", "model_id": TEXT_MODEL_LABEL, "fallback_used": False, "fallback_reason": None, "timings_ms": { "text_load": get_text_adapter().load_ms, "text": text_ms, "total": round((time.time() - started) * 1000), }, "asr": { "model_id": "supplied-transcript", "backend": "none", "inference_engine": "none", "model_artifact_format": "none", "quantization": "none", "target_lang": target_lang, "fallback_used": False, "transcript": transcript, }, "text": { "model_id": TEXT_MODEL_LABEL, "backend": "llama.cpp", "inference_engine": "llama.cpp", "model_artifact_format": "gguf", "quantization": "q8_0", "fallback_used": False, "raw_output": raw_text[:2000], "output": text_json, }, } ) append_trace(trace) return JSONResponse({"transcript": transcript, "model_message_en": text_json["message_en"], "text": text_json, "trace": trace}) except Exception as exc: axes = runtime_axes(runtime) trace.update( { "text_only_smoke": True, "model_runtime": runtime, "model_backend": axes["model_backend"], "inference_engine": axes["inference_engine"], "model_artifact_format": axes["model_artifact_format"], "quantization": axes["quantization"], "model_id": axes["text_model_id"], "fallback_used": True, "fallback_reason": f"real_text_model_error:{type(exc).__name__}", "blocker": str(exc), "timings_ms": {"total": round((time.time() - started) * 1000)}, } ) append_trace(trace) return JSONResponse( { "error": "Real text model path failed; deterministic output was not substituted.", "blocker": str(exc), "trace": trace, }, status_code=424, ) @app.post("/api/send") def api_send(payload: SendPayload) -> dict[str, Any]: if payload.simulate == "send_failure": raise HTTPException(status_code=502, detail="Send failed in deterministic test mode") final_message = payload.final_message_en.strip() model_message = payload.model_message_en.strip() if not final_message or not payload.trace_id: raise HTTPException(status_code=400, detail="final_message_en and trace_id are required") record = { "created_at": utc_now(), "site_id": payload.site_id, "contact_email": payload.contact_email, "model_message_en": model_message, "final_message_en": final_message, "user_edited": final_message != model_message, "trace_id": payload.trace_id, "delivery": "demo_ledger", "email_sent": False, "email_provider": "ledger", "fallback_used": selected_model_runtime() == "none", "model_runtime": selected_model_runtime(), } LEDGER_PATH.parent.mkdir(parents=True, exist_ok=True) with LEDGER_PATH.open("a", encoding="utf-8") as ledger: ledger.write(json.dumps(record, sort_keys=True) + "\n") return { "status": "saved", "visitor_copy": "Message saved for the demo.", "ledger_path": "product/5-idea-voice-contact-widget/v1/space/demo-ledger.jsonl", "email_sent": False, "email_provider": "ledger", "fallback_used": record["fallback_used"], "model_runtime": record["model_runtime"], } def gradio_status() -> dict[str, Any]: return {"health": health(), "packet": public_packet(load_packet())["provenance"], "recent_traces": RUN_LEDGER[:5]} def build_gradio_demo() -> gr.Blocks: with gr.Blocks(title="Voice Reach V1") as demo: gr.Markdown( """ # Voice Reach V1 The custom visitor widget is served at `/`. This status panel exists to keep the HF Space visibly Gradio-backed and to expose runtime provenance for smoke evidence. """ ) status = gr.JSON(value=gradio_status(), label="Runtime status") refresh = gr.Button("Refresh status", variant="primary") refresh.click(fn=gradio_status, inputs=None, outputs=status, concurrency_limit=1) return demo.queue(default_concurrency_limit=1) gradio_demo = build_gradio_demo() app = gr.mount_gradio_app(app, gradio_demo, path="/gradio") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run the Voice Reach V1 Space app.") parser.add_argument("--host", default="0.0.0.0") parser.add_argument("--port", default=int(os.environ.get("PORT", "7860")), type=int) parser.add_argument("--check", action="store_true", help="Validate packet and exit without loading models.") return parser.parse_args() def main() -> int: args = parse_args() packet = load_packet() if args.check: print( json.dumps( { "ok": True, "site_id": packet["site_id"], "brand_name": packet["brand_name"], "contact_email": packet["contact_email"], "speech_modes": packet["locale_defaults"]["speech_modes"], "audio_eval_consent": packet["audio_eval_consent"], "app_host": detect_app_host(), "model_mode": model_mode(), "model_runtime": selected_model_runtime(), "runtime_switch_allowed": runtime_switch_allowed(), "fallback_used": selected_model_runtime() == "none", "concurrency": 1, }, indent=2, ) ) return 0 import uvicorn uvicorn.run(app, host=args.host, port=args.port) return 0 if __name__ == "__main__": raise SystemExit(main())