from __future__ import annotations import re import time import unicodedata import numpy as np import torch import config from backend.types import StageResult _device = 'mps' if torch.backends.mps.is_available() else 'cpu' _asr_pipeline = None _qwen_model = None _qwen_tokenizer = None _livekit_model = None _livekit_tokenizer = None _livekit_im_end_id = None PROMPT_TEMPLATE = 'You are analyzing a snippet of transcribed speech from a Hindi-English (Hinglish) conversation with a voice assistant. Code-switching between Hindi and English mid-sentence is normal and not a sign of incompleteness. Filler words like "matlab", "toh", "haan", "wo kya bolte hain", "um", "uh" indicate the speaker is still thinking and has NOT completed their turn.\n\nTranscript: "{transcript}"\n\nClassify whether the speaker\'s turn is:\n- complete: the utterance is a complete thought, the speaker is done\n- incomplete: the utterance is grammatically or semantically incomplete, more is coming\n- wait: the speaker is explicitly asking for a pause (e.g. "ek second", "hold on", "wait")\n\nRespond with exactly one word: complete, incomplete, or wait.' _VALID_LABELS = ('complete', 'incomplete', 'wait') def _get_asr_pipeline(): global _asr_pipeline if _asr_pipeline is None: from transformers import pipeline _asr_pipeline = pipeline('automatic-speech-recognition', model=config.WHISPER_TINY_ID, device=_device if _device != 'mps' else -1) return _asr_pipeline def _get_qwen(): global _qwen_model, _qwen_tokenizer if _qwen_model is None: from transformers import AutoModelForCausalLM, AutoTokenizer _qwen_tokenizer = AutoTokenizer.from_pretrained(config.QWEN_LOCAL_ID) _qwen_model = AutoModelForCausalLM.from_pretrained(config.QWEN_LOCAL_ID).to(_device).eval() return (_qwen_model, _qwen_tokenizer) def _get_livekit(): global _livekit_model, _livekit_tokenizer, _livekit_im_end_id if _livekit_model is None: from transformers import AutoModelForCausalLM, AutoTokenizer _livekit_tokenizer = AutoTokenizer.from_pretrained(config.LIVEKIT_TURN_DETECTOR_ID) _livekit_model = AutoModelForCausalLM.from_pretrained(config.LIVEKIT_TURN_DETECTOR_ID).to(_device).eval() _livekit_im_end_id = _livekit_tokenizer.convert_tokens_to_ids('<|im_end|>') return (_livekit_model, _livekit_tokenizer, _livekit_im_end_id) _PUNCTUATION_RE = re.compile("[^\\w\\s'-]", re.UNICODE) _WHITESPACE_RE = re.compile('\\s+') def _normalize_for_livekit(transcript: str) -> str: text = unicodedata.normalize('NFKC', transcript).lower() text = _PUNCTUATION_RE.sub(' ', text) return _WHITESPACE_RE.sub(' ', text).strip() def transcribe(audio: np.ndarray, sample_rate: int=config.SAMPLE_RATE) -> str: asr = _get_asr_pipeline() result = asr({'raw': np.asarray(audio, dtype=np.float32), 'sampling_rate': sample_rate}) return result['text'].strip() def _parse_label(raw_text: str) -> str: lowered = raw_text.lower() for label in _VALID_LABELS: if re.search(f'\\b{label}\\b', lowered): return label return 'incomplete' @torch.inference_mode() def classify_transcript(transcript: str, temperature: float=0.2) -> dict: model, tokenizer = _get_qwen() messages = [{'role': 'user', 'content': PROMPT_TEMPLATE.format(transcript=transcript)}] prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) inputs = tokenizer(prompt, return_tensors='pt').to(_device) do_sample = temperature > 0 output_ids = model.generate(**inputs, max_new_tokens=8, do_sample=do_sample, temperature=temperature if do_sample else None, pad_token_id=tokenizer.eos_token_id) generated = output_ids[0][inputs['input_ids'].shape[1]:] raw_text = tokenizer.decode(generated, skip_special_tokens=True) label = _parse_label(raw_text) return {'verdict': label, 'raw_response': raw_text.strip()} def run(audio: np.ndarray, sample_rate: int=config.SAMPLE_RATE, temperature: float=0.2) -> StageResult: start = time.perf_counter() transcript = transcribe(audio, sample_rate) result = classify_transcript(transcript, temperature) timing_ms = (time.perf_counter() - start) * 1000 return StageResult(stage='semantic.qwen_local', timing_ms=timing_ms, output={'transcript': transcript, 'verdict': result['verdict'], 'raw_response': result['raw_response']}, available=True, provenance='real_checkpoint') @torch.inference_mode() def classify_transcript_livekit(transcript: str) -> float: model, tokenizer, im_end_id = _get_livekit() normalized = _normalize_for_livekit(transcript) prompt = f'<|im_start|><|user|>{normalized}' inputs = tokenizer(prompt, return_tensors='pt').to(_device) logits = model(**inputs).logits[0, -1, :] probs = torch.softmax(logits, dim=-1) return probs[im_end_id].item() def run_livekit(audio: np.ndarray, sample_rate: int=config.SAMPLE_RATE) -> StageResult: start = time.perf_counter() transcript = transcribe(audio, sample_rate) probability = classify_transcript_livekit(transcript) timing_ms = (time.perf_counter() - start) * 1000 return StageResult(stage='semantic.livekit_eou', timing_ms=timing_ms, output={'transcript': transcript, 'probability': probability}, available=True, provenance='real_checkpoint')