feat: implement ASR pipeline with async batch processing, FastAPI endpoints, and client-side transcription support

asr/__init__.py

@@ -0,0 +1 @@
+# ASR package

asr/batch_workers.py

@@ -0,0 +1,256 @@
+"""
+ASR Batch Workers — Async batching infrastructure for voice transcription.
+
+Architecture (mirrors Raij/src/smart_search/batch_workers.py, audio workers only):
+  - Two asyncio.Queues: audio_en_queue and audio_ar_queue.
+  - Two worker coroutines per queue drain jobs in micro-batches.
+  - Workers offload heavy inference to a thread via run_in_executor.
+  - A shared in-memory job_store tracks job status + results.
+  - A warmup loop periodically keeps OpenMP threads alive between requests.
+    Parakeet warmup runs every PARAKEET_WARMUP_EVERY cycles (full transcribe,
+    must use the full pipeline to avoid corrupting TDT decoder state).
+    wav2vec2 warmup runs every cycle (cheap raw forward pass).
+"""
+
+import asyncio
+import time
+import uuid
+import logging
+from typing import Any
+
+from .constants import ASR_BATCH_MAX, ASR_BATCH_WINDOW_S, WARMUP_INTERVAL_S, PARAKEET_WARMUP_EVERY
+from .schemas import AudioJob
+
+logger = logging.getLogger(__name__)
+
+
+# ═══════════════════════ Job Store ════════════════════════
+
+job_store: dict[str, dict[str, Any]] = {}
+"""
+{
+    "<job_id>": {
+        "status": "pending" | "processing" | "done" | "error",
+        "result": <str transcript> | None,
+        "error": <str> | None,
+    }
+}
+"""
+
+
+def create_job() -> str:
+    """Create a new pending job and return its ID."""
+    job_id = str(uuid.uuid4())
+    job_store[job_id] = {"status": "pending", "result": None, "error": None}
+    return job_id
+
+
+# ═══════════════════════ Request-in-Flight Gate ════════════════════════
+
+_request_in_flight_count = 0
+
+
+def set_request_in_flight(active: bool):
+    """Increment/decrement in-flight counter used to gate warmup cycles."""
+    global _request_in_flight_count
+    if active:
+        _request_in_flight_count += 1
+    else:
+        _request_in_flight_count = max(0, _request_in_flight_count - 1)
+
+
+def is_request_in_flight() -> bool:
+    return _request_in_flight_count > 0
+
+
+# ═══════════════════════ Queues ════════════════════════
+
+audio_en_queue: asyncio.Queue[AudioJob] = asyncio.Queue()
+audio_ar_queue: asyncio.Queue[AudioJob] = asyncio.Queue()
+
+
+# ═══════════════════════ Workers ════════════════════════
+
+async def audio_en_worker():
+    """
+    Drains up to ASR_BATCH_MAX English audio jobs every ASR_BATCH_WINDOW_S seconds.
+    Runs one batched Parakeet transcription via run_in_executor.
+    Writes transcript into job_store and sets job.done.
+    Cleans up temp audio files after processing.
+    """
+    import os
+    from .models import transcribe_en_batch
+
+    loop = asyncio.get_event_loop()
+
+    while True:
+        first_job: AudioJob = await audio_en_queue.get()
+        batch: list[AudioJob] = [first_job]
+
+        # Collect up to (ASR_BATCH_MAX - 1) more within the time window
+        deadline = loop.time() + ASR_BATCH_WINDOW_S
+        while len(batch) < ASR_BATCH_MAX:
+            remaining = deadline - loop.time()
+            if remaining <= 0:
+                break
+            try:
+                job = await asyncio.wait_for(audio_en_queue.get(), timeout=remaining)
+                batch.append(job)
+            except asyncio.TimeoutError:
+                break
+
+        for job in batch:
+            job_store[job.job_id]["status"] = "processing"
+
+        try:
+            set_request_in_flight(True)
+            audio_paths = [job.audio_path for job in batch]
+            transcripts = await loop.run_in_executor(None, transcribe_en_batch, audio_paths)
+
+            for job, transcript in zip(batch, transcripts):
+                if not transcript.strip():
+                    job_store[job.job_id]["status"] = "error"
+                    job_store[job.job_id]["error"] = "Could not transcribe audio. Please try again and speak clearly."
+                else:
+                    job_store[job.job_id]["status"] = "done"
+                    job_store[job.job_id]["result"] = transcript
+                job.done.set()
+
+        except Exception as e:
+            logger.error(f"English ASR batch failed: {e}", exc_info=True)
+            for job in batch:
+                job_store[job.job_id]["status"] = "error"
+                job_store[job.job_id]["error"] = str(e)
+                if not job.done.is_set():
+                    job.done.set()
+        finally:
+            set_request_in_flight(False)
+            for job in batch:
+                try:
+                    os.unlink(job.audio_path)
+                except Exception:
+                    pass
+
+
+async def audio_ar_worker():
+    """
+    Drains up to ASR_BATCH_MAX Arabic audio jobs every ASR_BATCH_WINDOW_S seconds.
+    Runs one batched wav2vec2 transcription via run_in_executor.
+    Writes transcript into job_store and sets job.done.
+    Cleans up temp audio files after processing.
+    """
+    import os
+    from .models import transcribe_ar_batch
+
+    loop = asyncio.get_event_loop()
+
+    while True:
+        first_job: AudioJob = await audio_ar_queue.get()
+        batch: list[AudioJob] = [first_job]
+
+        deadline = loop.time() + ASR_BATCH_WINDOW_S
+        while len(batch) < ASR_BATCH_MAX:
+            remaining = deadline - loop.time()
+            if remaining <= 0:
+                break
+            try:
+                job = await asyncio.wait_for(audio_ar_queue.get(), timeout=remaining)
+                batch.append(job)
+            except asyncio.TimeoutError:
+                break
+
+        for job in batch:
+            job_store[job.job_id]["status"] = "processing"
+
+        try:
+            set_request_in_flight(True)
+            audio_paths = [job.audio_path for job in batch]
+            transcripts = await loop.run_in_executor(None, transcribe_ar_batch, audio_paths)
+
+            for job, transcript in zip(batch, transcripts):
+                if not transcript.strip():
+                    job_store[job.job_id]["status"] = "error"
+                    job_store[job.job_id]["error"] = "لم يتم التعرف على الصوت. الرجاء المحاولة مرة أخرى والتحدث بوضوح."
+                else:
+                    job_store[job.job_id]["status"] = "done"
+                    job_store[job.job_id]["result"] = transcript
+                job.done.set()
+
+        except Exception as e:
+            logger.error(f"Arabic ASR batch failed: {e}", exc_info=True)
+            for job in batch:
+                job_store[job.job_id]["status"] = "error"
+                job_store[job.job_id]["error"] = str(e)
+                if not job.done.is_set():
+                    job.done.set()
+        finally:
+            set_request_in_flight(False)
+            for job in batch:
+                try:
+                    os.unlink(job.audio_path)
+                except Exception:
+                    pass
+
+
+# ═══════════════════════ Warmup Loop ════════════════════════
+
+async def _asr_warmup_loop():
+    """
+    Periodically poke both ASR models to prevent OpenMP/MKL thread pool
+    spin-down during idle periods.
+
+    - wav2vec2: every WARMUP_INTERVAL_S seconds (raw forward pass, ~5-15ms)
+    - Parakeet: every PARAKEET_WARMUP_EVERY cycles (~6 min at 45s/cycle)
+      Uses full model.transcribe() to avoid corrupting TDT decoder cache.
+
+    Skipped entirely if a real request is in flight.
+    """
+    from .models import warmup_parakeet, warmup_wav2vec2
+
+    loop = asyncio.get_event_loop()
+    parakeet_cycle = 0
+
+    while True:
+        await asyncio.sleep(WARMUP_INTERVAL_S)
+        if is_request_in_flight():
+            continue
+
+        t0 = time.monotonic()
+        try:
+            await loop.run_in_executor(None, warmup_wav2vec2)
+            parakeet_cycle += 1
+            if parakeet_cycle >= PARAKEET_WARMUP_EVERY:
+                parakeet_cycle = 0
+                await loop.run_in_executor(None, warmup_parakeet)
+        except Exception as e:
+            logger.warning(f"⚠️ ASR warmup cycle error (non-fatal): {e}")
+            continue
+
+        elapsed_ms = (time.monotonic() - t0) * 1000
+        logger.info(f"🔥 ASR warmup cycle done in {elapsed_ms:.0f}ms")
+
+
+# ═══════════════════════ Startup ════════════════════════
+
+_asr_workers_started = False
+
+
+def start_asr_workers():
+    """
+    Launch all ASR async worker coroutines. Call once during app startup.
+    - 2 English audio workers (Parakeet)
+    - 2 Arabic audio workers (wav2vec2)
+    - 1 warmup loop
+    """
+    global _asr_workers_started
+    if _asr_workers_started:
+        return
+    _asr_workers_started = True
+
+    for i in range(2):
+        asyncio.create_task(audio_en_worker(), name=f"asr_en_worker_{i}")
+    for i in range(2):
+        asyncio.create_task(audio_ar_worker(), name=f"asr_ar_worker_{i}")
+    asyncio.create_task(_asr_warmup_loop(), name="asr_warmup_loop")
+
+    logger.info("✅ ASR batch workers started (2 EN + 2 AR + warmup loop)")

asr/constants.py

@@ -0,0 +1,13 @@
+# ═══════════════════════ ASR Batch Worker Constants ════════════════════════
+
+# Max audio files coalesced per batch window
+ASR_BATCH_MAX = 6
+
+# Seconds to collect concurrent jobs before firing inference
+ASR_BATCH_WINDOW_S = 0.1
+
+# Lightweight wav2vec2 warmup cadence (seconds)
+WARMUP_INTERVAL_S = 45
+
+# Parakeet (full model.transcribe) warmup every N lightweight cycles (~6 min at 45s each)
+PARAKEET_WARMUP_EVERY = 8

asr/models.py

@@ -0,0 +1,287 @@
+"""
+ASR Models — English (Parakeet) and Arabic (wav2vec2) singletons.
+
+Strategy (ported from Raij/src/models.py, hotword biasing removed):
+  - Lazy-loaded singletons: models load on first call, not at import time.
+  - Warmup on load: a silent audio pass pre-JITs the computation graph so
+    the first real request has the same latency as subsequent ones.
+  - Batch inference: both transcribe_*_batch functions accept a list of
+    audio file paths and run a single forward pass.
+  - Thread lock on Parakeet: model.transcribe() is stateful (TDT decoder),
+    so we serialize all calls behind _en_model_lock.
+  - Warmup functions (warmup_parakeet / warmup_wav2vec2) are called
+    periodically by the batch worker warmup loop to prevent OpenMP
+    thread pool spin-down during idle periods.
+"""
+
+import os
+import threading
+import logging
+
+logger = logging.getLogger(__name__)
+
+# Force PyTorch path, no TensorFlow
+os.environ.setdefault("USE_TF", "0")
+os.environ.setdefault("USE_TORCH", "1")
+
+# Cap OpenMP threads — adjust if running on a GPU server with more cores
+os.environ.setdefault("OMP_NUM_THREADS", "2")
+
+import torch
+torch.set_num_threads(int(os.environ.get("OMP_NUM_THREADS", "2")))
+torch.set_num_interop_threads(1)
+
+_audio_model_en = None
+_audio_model_ar = None
+_en_model_lock = threading.Lock()
+_ar_model_lock = threading.Lock()
+
+
+# ═══════════════════════ English ASR (Parakeet) ════════════════════════
+
+def get_audio_model_en():
+    """
+    Loads nvidia/parakeet-tdt-0.6b-v2 via NeMo.
+    Runs a 1-second silence warmup to pre-JIT internal computation graphs.
+    Uses greedy_batch decoding strategy for best throughput.
+    No hotword biasing — general-purpose decoding.
+    """
+    global _audio_model_en
+    if _audio_model_en is not None:
+        return _audio_model_en
+
+    import wave
+    import tempfile
+    import nemo.collections.asr as nemo_asr
+
+    logger.info("Loading English ASR model (nvidia/parakeet-tdt-0.6b-v2)...")
+    model = nemo_asr.models.ASRModel.from_pretrained("nvidia/parakeet-tdt-0.6b-v2")
+
+    # ── Warmup: transcribe 1s of silence to pre-JIT computation graphs ──
+    warmup_path = None
+    try:
+        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
+            warmup_path = f.name
+        with wave.open(warmup_path, "wb") as wf:
+            wf.setnchannels(1)
+            wf.setsampwidth(2)
+            wf.setframerate(16000)
+            wf.writeframes(b"\x00" * 32000)  # 1s of silence at 16kHz
+        model.freeze()
+        with torch.no_grad():
+            model.transcribe([warmup_path])
+        model.unfreeze()
+        logger.info("✅ Parakeet warmup complete")
+    except Exception as e:
+        logger.warning(f"⚠️ Parakeet warmup failed (non-fatal): {e}")
+    finally:
+        if warmup_path:
+            try:
+                os.unlink(warmup_path)
+            except Exception:
+                pass
+
+    # ── Switch to greedy_batch for speed (no hotword biasing) ──
+    try:
+        from omegaconf import OmegaConf
+        decoding_cfg = OmegaConf.structured(model.cfg.decoding)
+        OmegaConf.update(decoding_cfg, "strategy", "greedy_batch")
+        if hasattr(decoding_cfg, "greedy"):
+            OmegaConf.update(decoding_cfg, "greedy.max_symbols", 5)
+        try:
+            model.change_decoding_strategy(decoding_cfg, verbose=False)
+            logger.info("✅ Parakeet decoding strategy: greedy_batch")
+        except Exception as strat_e:
+            logger.warning(f"⚠️ greedy_batch strategy failed ({strat_e}), using default")
+    except Exception as e:
+        logger.warning(f"⚠️ Parakeet decoding strategy setup failed (non-fatal): {e}")
+
+    _audio_model_en = model
+    logger.info("✅ English ASR model (Parakeet) loaded and ready")
+    return _audio_model_en
+
+
+def transcribe_en_batch(audio_paths: list[str]) -> list[str]:
+    """
+    Batch transcription for English audio using Parakeet.
+    NeMo's model.transcribe() natively handles batching internally —
+    it pads to the same length and runs a single forward pass.
+    Serialized behind _en_model_lock (Parakeet TDT decoder is stateful).
+    Returns a list of transcription strings (one per input path).
+    """
+    model = get_audio_model_en()
+    with _en_model_lock:
+        with torch.no_grad():
+            transcriptions = model.transcribe(audio_paths)
+            if isinstance(transcriptions, tuple):
+                transcriptions = transcriptions[0]
+            return [
+                (t.text if hasattr(t, "text") else str(t)).strip().rstrip(".")
+                for t in transcriptions
+            ]
+
+
+def warmup_parakeet():
+    """
+    Keeps Parakeet's OpenMP threads alive via a 0.5s silence transcription.
+    Must use model.transcribe() (not raw encoder) to avoid corrupting the
+    TDT decoder cache. No-op if model is not yet loaded.
+    """
+    if _audio_model_en is None:
+        return
+    import wave
+    import tempfile
+    model = _audio_model_en
+    warmup_path = None
+    try:
+        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
+            warmup_path = f.name
+        with wave.open(warmup_path, "wb") as wf:
+            wf.setnchannels(1)
+            wf.setsampwidth(2)
+            wf.setframerate(16000)
+            wf.writeframes(b"\x00" * 16000)  # 0.5s of silence
+        with _en_model_lock:
+            model.eval()
+            with torch.no_grad():
+                model.transcribe([warmup_path])
+    except Exception as e:
+        logger.warning(f"⚠️ Parakeet warmup error (non-fatal): {e}")
+    finally:
+        if warmup_path:
+            try:
+                os.unlink(warmup_path)
+            except Exception:
+                pass
+
+
+# ═══════════════════════ Arabic ASR (wav2vec2) ════════════════════════
+
+def get_audio_model_ar():
+    """
+    Loads IbrahimAmin/egyptian-arabic-wav2vec2-xlsr-53 via HuggingFace Transformers.
+    Runs a 0.5s dummy forward pass to warm up OpenMP threads.
+    No hotword biasing — pure greedy argmax decoding.
+    """
+    global _audio_model_ar
+    if _audio_model_ar is not None:
+        return _audio_model_ar
+
+    import numpy as np
+    from transformers import Wav2Vec2ForCTC, AutoProcessor
+
+    model_name = "IbrahimAmin/egyptian-arabic-wav2vec2-xlsr-53"
+    logger.info(f"Loading Arabic ASR model ({model_name})...")
+
+    processor = AutoProcessor.from_pretrained(model_name)
+    model = Wav2Vec2ForCTC.from_pretrained(model_name)
+    model.eval()
+
+    # ── Warmup: single forward pass on dummy audio ──
+    try:
+        dummy = np.zeros(8000, dtype=np.float32)
+        warmup_inputs = processor(
+            [dummy], sampling_rate=16000, return_tensors="pt", padding=True
+        )
+        with torch.no_grad():
+            model(**warmup_inputs)
+        logger.info("✅ Arabic ASR warmup complete")
+    except Exception as e:
+        logger.warning(f"⚠️ Arabic ASR warmup failed (non-fatal): {e}")
+
+    _audio_model_ar = {"model": model, "processor": processor, "model_name": model_name}
+    logger.info("✅ Arabic ASR model (wav2vec2) loaded and ready")
+    return _audio_model_ar
+
+
+def _load_audio_file(path: str):
+    """
+    Load an audio file to a 16kHz mono float32 numpy array.
+    Tries soundfile first (fast, no subprocess), then falls back to
+    librosa (handles more formats including webm via ffmpeg backend).
+    """
+    import numpy as np
+    try:
+        import soundfile as sf
+        data, sr = sf.read(path, dtype="float32", always_2d=False)
+        if data.ndim > 1:
+            data = data.mean(axis=1)
+        if sr != 16000:
+            import librosa
+            data = librosa.resample(data, orig_sr=sr, target_sr=16000)
+        return data.astype(np.float32)
+    except Exception:
+        import librosa
+        data, _ = librosa.load(path, sr=16000, mono=True)
+        return data.astype(np.float32)
+
+
+def transcribe_ar_batch(audio_paths: list[str]) -> list[str]:
+    """
+    Batch transcription for Arabic audio using wav2vec2.
+    Loads all audio concurrently, pads to same length,
+    runs one forward pass, and decodes via greedy argmax.
+    Returns a list of transcription strings (one per input path).
+    """
+    import numpy as np
+    from concurrent.futures import ThreadPoolExecutor
+
+    ar = get_audio_model_ar()
+    model = ar["model"]
+    processor = ar["processor"]
+
+    # Load all waveforms concurrently
+    with ThreadPoolExecutor(max_workers=min(len(audio_paths), 8)) as executor:
+        waveforms_raw = list(executor.map(_load_audio_file, audio_paths))
+
+    # Guard against empty waveforms from failed decodes
+    final_texts = [""] * len(audio_paths)
+    valid_indices: list[int] = []
+    valid_waveforms: list[np.ndarray] = []
+
+    for idx, wav in enumerate(waveforms_raw):
+        arr = np.asarray(wav, dtype=np.float32).reshape(-1)
+        if arr.size == 0:
+            logger.warning(f"⚠️ Arabic ASR: empty waveform at index {idx}, skipping")
+            continue
+        valid_indices.append(idx)
+        valid_waveforms.append(arr)
+
+    if not valid_waveforms:
+        return final_texts
+
+    inputs = processor(
+        valid_waveforms,
+        sampling_rate=16000,
+        return_tensors="pt",
+        padding=True,
+    )
+
+    with _ar_model_lock:
+        with torch.no_grad():
+            outputs = model(**inputs)
+
+    predicted_ids = torch.argmax(outputs.logits, dim=-1)
+    transcriptions = processor.batch_decode(predicted_ids)
+
+    for local_i, text in enumerate(transcriptions):
+        final_texts[valid_indices[local_i]] = text.strip().rstrip(".")
+
+    return final_texts
+
+
+def warmup_wav2vec2():
+    """
+    Lightweight raw forward pass to keep wav2vec2's OpenMP threads alive.
+    No-op if the Arabic model is not yet loaded.
+    """
+    if _audio_model_ar is None:
+        return
+    import numpy as np
+    ar = _audio_model_ar
+    dummy = np.zeros(8000, dtype=np.float32)
+    inputs = ar["processor"](
+        [dummy], sampling_rate=16000, return_tensors="pt", padding=True
+    )
+    with torch.no_grad():
+        ar["model"](**inputs)

asr/routes.py

@@ -0,0 +1,90 @@
+import asyncio
+import os
+import tempfile
+import uuid
+import logging
+from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Depends
+from fastapi.responses import JSONResponse
+
+from src.dependencies import get_current_user_id
+from src.asr.schemas import AudioJob
+from src.asr.batch_workers import audio_en_queue, audio_ar_queue, job_store
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/asr", tags=["ASR"])
+
+# Allowed audio MIME types from browsers (MediaRecorder output)
+_ALLOWED_MIME_PREFIXES = ("audio/", "video/webm")  # webm is video/* but contains audio
+
+_ASR_TIMEOUT_S = 60  # max seconds to wait for a transcription result
+
+
+@router.post("/transcribe")
+async def transcribe_audio(
+    audio: UploadFile = File(..., description="Audio recording from the browser (.webm, .wav, .ogg)"),
+    language: str = Form(..., description="Language code: 'en' for English, 'ar' for Arabic"),
+    user_id: str = Depends(get_current_user_id),
+):
+    """
+    Transcribe an audio file using the selected language model:
+    - 'en': nvidia/parakeet-tdt-0.6b-v2 (NeMo)
+    - 'ar': IbrahimAmin/egyptian-arabic-wav2vec2-xlsr-53 (HuggingFace)
+
+    The audio is queued for batch inference and the response waits
+    for the transcript to be ready (up to 60 seconds).
+    """
+    if language not in ("en", "ar"):
+        raise HTTPException(400, "Invalid language. Must be 'en' or 'ar'.")
+
+    # Validate MIME type loosely (browsers vary on exact content-type for webm)
+    content_type = audio.content_type or ""
+    if not any(content_type.startswith(prefix) for prefix in _ALLOWED_MIME_PREFIXES):
+        logger.warning(f"Unexpected audio content-type: {content_type} — allowing anyway")
+
+    # Save upload to a temp file (workers clean up after processing)
+    suffix = ".webm"
+    if audio.filename:
+        _, ext = os.path.splitext(audio.filename)
+        if ext:
+            suffix = ext
+
+    tmp_path = None
+    try:
+        with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
+            tmp_path = tmp.name
+            content = await audio.read()
+            tmp.write(content)
+    except Exception as e:
+        logger.error(f"Failed to save audio upload: {e}", exc_info=True)
+        raise HTTPException(500, "Failed to save audio file.")
+
+    # Create job and queue it
+    job_id = str(uuid.uuid4())
+    job = AudioJob(job_id=job_id, audio_path=tmp_path, language=language)
+    job_store[job_id] = {"status": "pending", "result": None, "error": None}
+
+    if language == "en":
+        await audio_en_queue.put(job)
+    else:
+        await audio_ar_queue.put(job)
+
+    # Wait for the worker to finish (timeout = _ASR_TIMEOUT_S)
+    try:
+        await asyncio.wait_for(job.done.wait(), timeout=_ASR_TIMEOUT_S)
+    except asyncio.TimeoutError:
+        # Clean up the temp file if the worker hasn't done so
+        try:
+            if tmp_path and os.path.exists(tmp_path):
+                os.unlink(tmp_path)
+        except Exception:
+            pass
+        job_store.pop(job_id, None)
+        raise HTTPException(504, "Transcription timed out. Please try a shorter recording.")
+
+    entry = job_store.pop(job_id, {})
+    if entry.get("status") == "error":
+        raise HTTPException(500, entry.get("error", "Transcription failed."))
+
+    transcript = entry.get("result", "")
+    return JSONResponse({"transcript": transcript})

asr/schemas.py

@@ -0,0 +1,11 @@
+import asyncio
+from dataclasses import dataclass, field
+
+
+@dataclass
+class AudioJob:
+    """Single audio transcription job queued for batch inference."""
+    job_id: str
+    audio_path: str          # path to the temp audio file on disk
+    language: str            # "en" | "ar"
+    done: asyncio.Event = field(default_factory=asyncio.Event)

main.py

@@ -11,6 +11,7 @@ from src.summary_generator.routes import router as summary_router
 from src.rag.routes import router as tutor_router
 from src.quiz_generator.routes import router as quiz_router
 from src.auth.routes import router as auth_router
+from src.asr.routes import router as asr_router
 from src.store import get_usage
 from src.dependencies import get_current_user_id
 from src.config import settings 
@@ -51,11 +52,25 @@ async def lifespan(app: FastAPI):
     except Exception as e:
         logger.warning(f"Embedder failed to load: {e}")
 
+    # Eagerly load ASR models so warmup runs at startup, not on first request
+    try:
+        from src.asr.models import get_audio_model_en
+        get_audio_model_en()
+    except Exception as e:
+        logger.warning(f"English ASR model failed to load: {e}")
 
+    try:
+        from src.asr.models import get_audio_model_ar
+        get_audio_model_ar()
+    except Exception as e:
+        logger.warning(f"Arabic ASR model failed to load: {e}")
 
     from src.rag.batch_workers import start_workers
     start_workers()
 
+    from src.asr.batch_workers import start_asr_workers
+    start_asr_workers()
+
     yield
 
 
@@ -87,6 +102,7 @@ app.include_router(summary_router)
 app.include_router(tutor_router)
 app.include_router(quiz_router)
 app.include_router(auth_router)
+app.include_router(asr_router)
 
 
 @app.get("/")