import whisper import os import torch import warnings import gc # Garbage Collector for memory management import tempfile import re import time from pydub import AudioSegment # 🟢 NEW: Add pydub # Suppress FP16 warnings on CPU warnings.filterwarnings("ignore", message="FP16 is not supported on CPU; using FP32 instead") class AgentInput: TURBO_MODEL_ID = "openai/whisper-large-v3-turbo" DISTIL_MODEL_ID = "distil-whisper/distil-large-v3" MMS_MODEL_ID = "facebook/mms-1b-all" QWEN_ASR_MODEL_ID = "Qwen/Qwen3-ASR-1.7B" def __init__(self, device="cpu"): print(f"👂 Agent 1 (Input) Online: Preparing Whisper on {device}...") self.device = device self.loaded_models = {} self.loaded_pipelines = {} self.turbo_disabled = False self.turbo_slow_threshold_s = float(os.environ.get("ASR_TURBO_SLOW_THRESHOLD_S", "18")) # USE 'tiny' FOR CLOUD TO PREVENT EXIT CODE 137 (OOM) # Use 'base' only if you are running on a machine with 16GB+ RAM self.model_name = "tiny" try: # Load the model and immediately collect garbage to free RAM self.model = whisper.load_model(self.model_name, device=device) self.loaded_models[self.model_name] = self.model gc.collect() print(f"✅ Whisper '{self.model_name}' model loaded. RAM optimized.") except Exception as e: print(f"⚠️ Load failed: {e}. Attempting emergency load...") # Emergency fallback to tiny if not already tried self.model = whisper.load_model("tiny", device=device) self.loaded_models["tiny"] = self.model self.model_name = "tiny" def _audio_duration_s(self, audio_path): try: return float(AudioSegment.from_file(audio_path).duration_seconds) except Exception: return None def _hint_text(self, language=None, dialect_hint=""): return f"{language or ''} {dialect_hint or ''}".strip().lower() def _is_zero_gpu_available(self): return torch.cuda.is_available() def _is_underrepresented_language(self, language=None, dialect_hint=""): hint = self._hint_text(language, dialect_hint) common_tokens = { "english", " en", "en ", "korean", " ko", "ko ", "arabic", " ar", "ar ", "chinese", "mandarin", "cantonese", " zh", "zh ", "french", " fr", "fr ", "spanish", " es", "es ", "german", "deutsch", " de", "de ", "italian", " it", "it ", "portuguese", " pt", "pt ", "hindi", " hi", "hi ", "japanese", " ja", "ja ", "russian", " ru", "ru " } if any(token in f" {hint} " for token in common_tokens): return False return bool(hint) def _expected_script(self, language=None, dialect_hint=""): lang = str(language or "").strip().lower() padded_lang = f" {lang} " if lang in {"en", "eng", "fr", "fra", "es", "spa", "de", "deu", "it", "ita", "pt", "por", "pcm", "tl", "tgl", "fil"}: return "latin" if lang in {"ko", "kor"}: return "hangul" if lang in {"ar", "ara", "ur", "urd", "fa", "fas", "prs", "ps", "pus"}: return "arabic" if lang in {"zh", "cmn", "yue", "zho"}: return "cjk" if lang in {"ja", "jpn"}: return "japanese" if lang in {"hi", "hin", "mr", "mar", "ne", "nep"}: return "devanagari" if lang in {"th", "tha"}: return "thai" hint = self._hint_text(language, dialect_hint) padded_hint = f" {hint} " if any(token in padded_hint for token in ["korean", "hangul", "satoori", "jeju", "gyeongsang", "chungcheong", "jeolla", "busan", " ko "]): return "hangul" if any(token in padded_hint for token in ["arabic", "urdu", "persian", "farsi", "dari", "pashto", " ar "]): return "arabic" if any(token in padded_hint for token in ["chinese", "mandarin", "cantonese", "yue", " zh "]): return "cjk" if any(token in padded_hint for token in ["japanese", " ja "]): return "japanese" if any(token in padded_hint for token in ["hindi", "marathi", "nepali", " hi "]): return "devanagari" if "thai" in padded_hint: return "thai" if any(token in padded_hint for token in ["english", "french", "spanish", "german", "italian", "portuguese", "tagalog", "filipino", "pidgin", "patois", " en ", " fr ", " es ", " tl "]): return "latin" if any(token in padded_lang for token in [" en ", " eng ", " fr ", " fra ", " es ", " spa "]): return "latin" return None def _looks_wrong_script(self, text, language=None, dialect_hint=""): text = str(text or "").strip() if not text: return True script = self._expected_script(language, dialect_hint) if not script: return False patterns = { "hangul": r"[\uac00-\ud7a3]", "arabic": r"[\u0600-\u06ff]", "cjk": r"[\u3400-\u9fff]", "japanese": r"[\u3040-\u30ff\u3400-\u9fff]", "devanagari": r"[\u0900-\u097f]", "thai": r"[\u0e00-\u0e7f]", "latin": r"[A-Za-z]", } expected_hits = len(re.findall(patterns[script], text)) if script == "latin": other_hits = len(re.findall(r"[\u0600-\u06ff\u0900-\u097f\u0e00-\u0e7f\u3040-\u30ff\u3400-\u9fff\uac00-\ud7a3]", text)) return expected_hits == 0 or other_hits > expected_hits return expected_hits == 0 def _mms_language_code(self, language=None, dialect_hint=""): hint = self._hint_text(language, dialect_hint) mappings = [ (["nigerian pidgin", "pidgin"], "pcm"), (["tagalog", "filipino", " tl ", " tgl ", " fil "], "tgl"), (["english", " en "], "eng"), (["korean", " ko "], "kor"), (["arabic", " ar "], "ara"), (["mandarin", "chinese", " zh "], "cmn"), (["cantonese", "yue"], "yue"), (["french", " fr "], "fra"), (["spanish", " es "], "spa"), (["german", "deutsch", " de "], "deu"), (["italian", " it "], "ita"), (["portuguese", " pt "], "por"), (["hindi", " hi "], "hin"), (["japanese", " ja "], "jpn"), (["russian", " ru "], "rus"), (["thai"], "tha"), (["vietnamese"], "vie"), (["swahili"], "swh"), (["yoruba"], "yor"), (["igbo"], "ibo"), (["hausa"], "hau"), ] padded = f" {hint} " for tokens, code in mappings: if any(token in padded for token in tokens): return code return "eng" def _normalize_manual_route(self, model_choice): choice = str(model_choice or "auto").strip().lower() aliases = { "tiny": {"engine": "whisper", "model": "tiny", "label": "Whisper tiny (Edge fastest)"}, "whisper tiny": {"engine": "whisper", "model": "tiny", "label": "Whisper tiny (Edge fastest)"}, "base": {"engine": "whisper", "model": "base", "label": "Whisper base (Edge balanced)"}, "whisper base": {"engine": "whisper", "model": "base", "label": "Whisper base (Edge balanced)"}, "small": {"engine": "whisper", "model": "small", "label": "Whisper small (Fallback)"}, "whisper-small": {"engine": "whisper", "model": "small", "label": "Whisper small (Fallback)"}, "whisper small": {"engine": "whisper", "model": "small", "label": "Whisper small (Fallback)"}, "turbo": {"engine": "hf-whisper", "model": self.TURBO_MODEL_ID, "label": "Whisper large-v3 turbo (HF Pro)"}, "large-v3-turbo": {"engine": "hf-whisper", "model": self.TURBO_MODEL_ID, "label": "Whisper large-v3 turbo (HF Pro)"}, "whisper-large-v3-turbo": {"engine": "hf-whisper", "model": self.TURBO_MODEL_ID, "label": "Whisper large-v3 turbo (HF Pro)"}, self.TURBO_MODEL_ID: {"engine": "hf-whisper", "model": self.TURBO_MODEL_ID, "label": "Whisper large-v3 turbo (HF Pro)"}, "distil": {"engine": "hf-whisper", "model": self.DISTIL_MODEL_ID, "label": "Distil-Whisper (English fast)"}, "distil-large-v3": {"engine": "hf-whisper", "model": self.DISTIL_MODEL_ID, "label": "Distil-Whisper (English fast)"}, self.DISTIL_MODEL_ID: {"engine": "hf-whisper", "model": self.DISTIL_MODEL_ID, "label": "Distil-Whisper (English fast)"}, "qwen": {"engine": "hf-whisper", "model": self.QWEN_ASR_MODEL_ID, "label": "Qwen3-ASR 1.7B (Multilingual)"}, "qwen3-asr": {"engine": "hf-whisper", "model": self.QWEN_ASR_MODEL_ID, "label": "Qwen3-ASR 1.7B (Multilingual)"}, "qwen/qwen3-asr-1.7b": {"engine": "hf-whisper", "model": self.QWEN_ASR_MODEL_ID, "label": "Qwen3-ASR 1.7B (Multilingual)"}, self.QWEN_ASR_MODEL_ID.lower(): {"engine": "hf-whisper", "model": self.QWEN_ASR_MODEL_ID, "label": "Qwen3-ASR 1.7B (Multilingual)"}, "mms": {"engine": "mms", "model": self.MMS_MODEL_ID, "label": "MMS 1B (Low-resource)"}, self.MMS_MODEL_ID: {"engine": "mms", "model": self.MMS_MODEL_ID, "label": "MMS 1B (Low-resource)"}, } return aliases.get(choice) def _auto_routes(self, language=None, dialect_hint="", audio_path=None): base = {"engine": "whisper", "model": "base", "label": "Whisper base (Edge balanced)"} tiny = {"engine": "whisper", "model": "tiny", "label": "Whisper tiny (Edge fastest)"} small = {"engine": "whisper", "model": "small", "label": "Whisper small (Fallback)"} mms = {"engine": "mms", "model": self.MMS_MODEL_ID, "label": "MMS 1B (Low-resource)"} turbo = {"engine": "hf-whisper", "model": self.TURBO_MODEL_ID, "label": "Whisper large-v3 turbo (HF Pro)"} duration_s = self._audio_duration_s(audio_path) if audio_path else None if self._is_zero_gpu_available() and not self.turbo_disabled and not (duration_s and duration_s > 18): return [turbo, base, tiny] if self._expected_script(language, dialect_hint) == "latin": return [base, tiny] if self._is_underrepresented_language(language, dialect_hint): return [base, mms, small] return [small] def resolve_model_name(self, model_choice="auto", language=None, dialect_hint="", audio_path=None): manual = self._normalize_manual_route(model_choice) if manual: return manual["model"] return self._auto_routes(language=language, dialect_hint=dialect_hint, audio_path=audio_path)[0]["model"] def _get_model(self, model_name): model_name = model_name if model_name in {"tiny", "base", "small"} else "small" if model_name in self.loaded_models: self.model_name = model_name self.model = self.loaded_models[model_name] return self.model try: print(f"🔄 Loading Whisper '{model_name}' on demand...") self.loaded_models[model_name] = whisper.load_model(model_name, device=self.device) self.model_name = model_name self.model = self.loaded_models[model_name] gc.collect() return self.model except Exception as e: print(f"⚠️ Whisper '{model_name}' unavailable ({e}); falling back to tiny.") self.model_name = "tiny" self.model = self.loaded_models.get("tiny") or whisper.load_model("tiny", device=self.device) self.loaded_models["tiny"] = self.model gc.collect() return self.model def _get_hf_pipeline(self, model_id): if model_id in self.loaded_pipelines: return self.loaded_pipelines[model_id] try: from transformers import pipeline device_index = 0 if torch.cuda.is_available() else -1 kwargs = {"model": model_id, "device": device_index} if torch.cuda.is_available(): kwargs["torch_dtype"] = torch.float16 print(f"🔄 Loading ASR pipeline '{model_id}'...") self.loaded_pipelines[model_id] = pipeline("automatic-speech-recognition", **kwargs) gc.collect() return self.loaded_pipelines[model_id] except Exception as e: raise RuntimeError(f"ASR pipeline '{model_id}' unavailable: {e}") from e def _get_mms_pipeline(self, language=None, dialect_hint=""): lang_code = self._mms_language_code(language, dialect_hint) cache_key = f"{self.MMS_MODEL_ID}:{lang_code}" if cache_key in self.loaded_pipelines: return self.loaded_pipelines[cache_key] try: from transformers import AutoProcessor, Wav2Vec2ForCTC, pipeline device_index = 0 if torch.cuda.is_available() else -1 processor = AutoProcessor.from_pretrained(self.MMS_MODEL_ID) model = Wav2Vec2ForCTC.from_pretrained(self.MMS_MODEL_ID) try: processor.tokenizer.set_target_lang(lang_code) model.load_adapter(lang_code) except Exception as e: print(f"⚠️ MMS adapter '{lang_code}' unavailable ({e}); trying English adapter.") lang_code = "eng" processor.tokenizer.set_target_lang(lang_code) model.load_adapter(lang_code) cache_key = f"{self.MMS_MODEL_ID}:{lang_code}" if torch.cuda.is_available(): model = model.to("cuda") self.loaded_pipelines[cache_key] = pipeline( "automatic-speech-recognition", model=model, tokenizer=processor.tokenizer, feature_extractor=processor.feature_extractor, device=device_index, ) gc.collect() return self.loaded_pipelines[cache_key] except Exception as e: raise RuntimeError(f"MMS ASR unavailable: {e}") from e def _transcribe_with_route(self, route, clean_path, language=None, dialect_hint=""): start = time.perf_counter() if route["engine"] == "whisper": model = self._get_model(route["model"]) result = model.transcribe(clean_path, language=language, fp16=False) elif route["engine"] == "mms": pipe = self._get_mms_pipeline(language, dialect_hint) result = pipe(clean_path) else: pipe = self._get_hf_pipeline(route["model"]) generate_kwargs = {"task": "transcribe"} if language: generate_kwargs["language"] = language result = pipe(clean_path, generate_kwargs=generate_kwargs) elapsed_s = time.perf_counter() - start if isinstance(result, dict): text = str(result.get("text", "")).strip() else: text = str(result or "").strip() if route["model"] == self.TURBO_MODEL_ID and elapsed_s > self.turbo_slow_threshold_s: self.turbo_disabled = True print(f"⚠️ Turbo route was slow ({elapsed_s:.2f}s); future Auto calls will use Whisper small.") return text, elapsed_s # 🟢 NEW HELPER: Sanitizes corrupted browser audio def _sanitize_audio(self, audio_path): try: # Try to load it regardless of format audio = AudioSegment.from_file(audio_path) # Export it to a clean, standard WAV in a temp file temp_path = os.path.join(tempfile.gettempdir(), f"clean_audio_{os.path.basename(audio_path)}.wav") audio.export(temp_path, format="wav") return temp_path except Exception as e: print(f"⚠️ Audio Sanitization Warning: {e}") return audio_path # Fallback to original if pydub fails def _should_sanitize_audio(self, sanitize_audio=True): if isinstance(sanitize_audio, str): return sanitize_audio.strip().lower() not in {"off", "false", "0", "no", "raw", "none"} return bool(sanitize_audio) def transcribe(self, audio_path, language=None, model_choice="auto", dialect_hint="", sanitize_audio=True): if not audio_path: return [{"text": "", "speaker": "SYSTEM"}] try: clean_path = self._sanitize_audio(audio_path) if self._should_sanitize_audio(sanitize_audio) else audio_path manual_route = self._normalize_manual_route(model_choice) routes = [manual_route] if manual_route else self._auto_routes(language=language, dialect_hint=dialect_hint, audio_path=clean_path) if manual_route and manual_route["engine"] != "whisper": routes.append({"engine": "whisper", "model": "small", "label": "Whisper small (Fallback)"}) transcription_text = "" used_label = "Unknown" underrepresented = self._is_underrepresented_language(language, dialect_hint) tried_mms_retry = False for route in routes: try: text, elapsed_s = self._transcribe_with_route(route, clean_path, language=language, dialect_hint=dialect_hint) used_label = route["label"] print(f"🎙️ Speech model route: requested={model_choice or 'auto'} resolved={used_label} time={elapsed_s:.2f}s") transcription_text = text if not manual_route and route["model"] == self.TURBO_MODEL_ID and elapsed_s > self.turbo_slow_threshold_s: transcription_text = "" continue needs_mms_retry = ( not manual_route and route["engine"] != "mms" and not tried_mms_retry and (underrepresented or self._looks_wrong_script(text, language, dialect_hint)) ) if needs_mms_retry: tried_mms_retry = True mms_route = {"engine": "mms", "model": self.MMS_MODEL_ID, "label": "MMS 1B (Low-resource)"} print("🔁 ASR output needs low-resource validation; retrying with MMS 1B.") mms_text, mms_elapsed = self._transcribe_with_route(mms_route, clean_path, language=language, dialect_hint=dialect_hint) if mms_text and not self._looks_wrong_script(mms_text, language, dialect_hint): transcription_text = mms_text used_label = mms_route["label"] print(f"🎙️ Speech model route: retry resolved={used_label} time={mms_elapsed:.2f}s") if transcription_text: break except Exception as route_error: print(f"⚠️ ASR route '{route['label']}' failed: {route_error}") continue gc.collect() # Force memory release # Clean up temp file if clean_path != audio_path and os.path.exists(clean_path): os.remove(clean_path) return [{"text": transcription_text, "speaker": "Speaker 1", "model": used_label}] except Exception as e: print(f"❌ Transcription Error: {e}") return [{"text": "", "speaker": "ERROR"}]