Deploy server 2026-05-08 — latency overhaul, meeting coach AI, client-side VAD, bounded queue, runtime model switching

Dockerfile

@@ -19,9 +19,7 @@ WORKDIR $HOME/app
 # ── Install Python deps ─────────────────────────────────────────────
 COPY --chown=user pyproject.toml uv.lock* ./
 
-# Use parakeet extra
-ARG ASR_EXTRA=parakeet
-RUN uv sync --frozen --no-dev --extra ${ASR_EXTRA}
+RUN uv sync --frozen --no-dev --extra parakeet
 
 # ── Copy application code ───────────────────────────────────────────
 COPY --chown=user server/app/ ./app/
@@ -33,10 +31,8 @@ USER user
 ENV PATH="$HOME/app/.venv/bin:$PATH"
 
 # Redirect caches to /tmp (writable in HF Spaces)
-ENV TORCH_HOME=/tmp/torch_cache
 ENV HF_HOME=/tmp/hf_cache
 ENV NUMBA_CACHE_DIR=/tmp/numba_cache
-ENV MPLCONFIGDIR=/tmp/matplotlib
 
 # Defaults for HF environment
 ENV ASR_MODEL_TYPE=parakeet

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "meeting-ai-assistant"
-version = "0.1.0"
+version = "0.2.0"
 requires-python = ">=3.11, <3.12"
 dependencies = [
     "fastapi>=0.111.0",
@@ -9,20 +9,15 @@ dependencies = [
     "python-dotenv>=1.0.0",
     "websockets>=12.0",
     "numpy>=1.24.0",
-    "torch>=2.0.0",
-    "silero-vad>=4.0.0",
+    "faster-whisper>=1.0.0",
     "groq>=0.9.0",
 ]
 
-# Install with: uv sync --extra parakeet  OR  uv sync --extra whisper
 [project.optional-dependencies]
 parakeet = [
     "nemo_toolkit[asr]>=1.23.0",
     "soundfile>=0.12.0",
 ]
-whisper = [
-    "faster-whisper>=1.0.0",
-]
 
 [dependency-groups]
 dev = [

server/app/config.py

@@ -16,10 +16,10 @@ class Settings(BaseSettings):
     api_keys: str = ""
     admin_key: str = ""
     groq_api_key: str = ""
-    whisper_model: str = "base.en"
+    whisper_model: str = "base.en"   # valid: tiny.en | base.en | small.en | medium.en
     whisper_device: str = "cpu"
     whisper_compute_type: str = "float32"
-    asr_model_type: str = "parakeet"
+    asr_model_type: str = "whisper"
     vad_threshold: float = 0.5
     chunk_duration_ms: int = 300
     max_silence_ms: int = 800

server/app/main.py

@@ -21,18 +21,11 @@ logger = logging.getLogger(__name__)
 
 
 async def _load_models(app: FastAPI) -> None:
-    """Load Silero VAD (fast) then ASR engine (slow) in a background thread."""
     try:
-        from silero_vad import load_silero_vad  # deferred: requires torch
-
-        loop = asyncio.get_event_loop()
-        app.state.vad_model = await loop.run_in_executor(None, load_silero_vad)
-        logger.info("Silero VAD loaded")
-
         engine = get_asr_engine()
+        loop = asyncio.get_event_loop()
         await loop.run_in_executor(None, engine.load_model)
         app.state.asr_engine = engine
-        app.state.whisper_loaded = True
         logger.info("ASR engine loaded (%s)", type(engine).__name__)
     except Exception:
         logger.exception("Model loading failed")
@@ -41,18 +34,14 @@ async def _load_models(app: FastAPI) -> None:
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     app.state.active_sessions: dict = {}
-    app.state.whisper_loaded: bool = False
     app.state.asr_engine = None
-    app.state.vad_model = None
-
     asyncio.create_task(_load_models(app))
-
     yield
 
 
 app = FastAPI(
     title="Meeting AI Assistant",
-    version="0.1.0",
+    version="0.2.0",
     lifespan=lifespan,
 )
 

server/app/models/asr.py

@@ -33,3 +33,12 @@ def get_asr_engine() -> ASREngine:
 
     from app.models.whisper_engine import WhisperEngine
     return WhisperEngine()
+
+
+WHISPER_ALLOWLIST = ("tiny.en", "base.en", "small.en", "medium.en")
+
+
+def get_whisper_engine(model_name: str) -> ASREngine:
+    """Return a WhisperEngine configured for the given model name."""
+    from app.models.whisper_engine import WhisperEngine
+    return WhisperEngine(model_name=model_name)

server/app/models/whisper_engine.py

@@ -7,22 +7,28 @@ from app.models.asr import ASREngine, ASRResult
 
 
 class WhisperEngine(ASREngine):
-    def __init__(self) -> None:
+    def __init__(self, model_name: str | None = None) -> None:
         self._model = None
+        self._model_name = model_name or settings.whisper_model
 
     def load_model(self) -> None:
-        from faster_whisper import WhisperModel  # deferred: heavy import
-
+        from faster_whisper import WhisperModel
         self._model = WhisperModel(
-            settings.whisper_model,
+            self._model_name,
             device=settings.whisper_device,
-            compute_type=settings.whisper_compute_type,
+            compute_type="int8",
+            cpu_threads=2,
         )
 
     def transcribe_audio(self, audio: np.ndarray, sample_rate: int = 16000) -> ASRResult:
         if self._model is None:
             raise RuntimeError("WhisperEngine: model not loaded")
-        segments, _ = self._model.transcribe(audio, beam_size=1, language="en")
+        segments, _ = self._model.transcribe(
+            audio,
+            beam_size=1,
+            language="en",
+            condition_on_previous_text=False,
+        )
         text = " ".join(seg.text for seg in segments).strip()
         return ASRResult(text=text)
 

server/app/routers/websocket.py

@@ -10,31 +10,36 @@ from fastapi import APIRouter, WebSocket, WebSocketDisconnect
 
 from app.auth import verify_api_key
 from app.config import settings
+from app.models.asr import WHISPER_ALLOWLIST, get_whisper_engine
 from app.schemas.messages import (
     AuthErrorMsg,
     AuthSuccessMsg,
     ControlFrame,
+    ErrorMsg,
+    ModelChangedMsg,
     PongMsg,
     ReadyMsg,
 )
 from app.services.audio_processor import AudioProcessor
 from app.services.groq_client import GroqClient
+from app.services.session_context import SessionContext
 from app.services.transcription import TranscriptionService
 
 logger = logging.getLogger(__name__)
 router = APIRouter()
+_model_switch_lock = asyncio.Lock()
 
-SESSION_TIMEOUT_SECONDS = 7200  # 2 hours
+SESSION_TIMEOUT_SECONDS = 7200
 
 
 @router.websocket("/ws")
 async def websocket_endpoint(websocket: WebSocket) -> None:
     await websocket.accept()
     session_id: str | None = None
-    processor: AudioProcessor | None = None
+    ts_service: TranscriptionService | None = None
 
     try:
-        # ── Auth frame ────────────────────────────────────────────────────────
+        # ── Auth ──────────────────────────────────────────────────────────────
         raw = await websocket.receive_text()
         try:
             data = json.loads(raw)
@@ -68,23 +73,29 @@ async def websocket_endpoint(websocket: WebSocket) -> None:
         logger.info("WS session %s opened", session_id)
 
         # ── Per-session pipeline ──────────────────────────────────────────────
-        vad_model = getattr(websocket.app.state, "vad_model", None)
         asr_engine = getattr(websocket.app.state, "asr_engine", None)
-        # Both models must be ready; if still loading, audio frames are dropped
-        processor = AudioProcessor(vad_model=vad_model) if vad_model is not None else None
-        groq_client = GroqClient(settings.groq_api_key) if settings.groq_api_key else None
+        session_context = SessionContext()
+        processor = AudioProcessor()
+        groq_client = (
+            GroqClient(settings.groq_api_key, context=session_context)
+            if settings.groq_api_key
+            else None
+        )
         ts_service = (
             TranscriptionService(websocket=websocket, engine=asr_engine, groq_client=groq_client)
-            if asr_engine else None
+            if asr_engine
+            else None
         )
-        if processor is None:
-            logger.warning("Session %s: VAD not ready, audio will be dropped until loaded", session_id)
+        if ts_service:
+            ts_service.start()
         if groq_client is None:
-            logger.warning("Session %s: GROQ_API_KEY not set, suggestions disabled", session_id)
+            logger.warning("Session %s: GROQ_API_KEY not set — suggestions disabled", session_id)
 
         # ── Message loop ──────────────────────────────────────────────────────
         while True:
-            elapsed = (datetime.now(tz=timezone.utc) - active_sessions[session_id]).total_seconds()
+            elapsed = (
+                datetime.now(tz=timezone.utc) - active_sessions[session_id]
+            ).total_seconds()
             if elapsed > SESSION_TIMEOUT_SECONDS:
                 logger.info("Session %s exceeded 2h cap, closing", session_id)
                 break
@@ -98,15 +109,16 @@ async def websocket_endpoint(websocket: WebSocket) -> None:
             if "bytes" in message and message["bytes"] is not None:
                 if processor is not None and ts_service is not None:
                     segment = processor.process_chunk(message["bytes"])
-                    if segment is not None:
-                        asyncio.create_task(ts_service.transcribe_and_send(segment))
+                    ts_service.enqueue(segment)
                 continue
 
             if "text" in message and message["text"] is not None:
                 try:
                     ctrl = ControlFrame.model_validate_json(message["text"])
                 except Exception:
-                    logger.warning("Session %s: unrecognised frame: %s", session_id, message["text"])
+                    logger.warning(
+                        "Session %s: unrecognised frame: %s", session_id, message["text"]
+                    )
                     continue
 
                 if ctrl.type == "ping":
@@ -114,12 +126,53 @@ async def websocket_endpoint(websocket: WebSocket) -> None:
                 elif ctrl.type == "stop_session":
                     logger.info("Session %s: stop_session received", session_id)
                     break
+                elif ctrl.type == "set_model":
+                    model_name = ctrl.model or ""
+                    if model_name not in WHISPER_ALLOWLIST:
+                        await websocket.send_text(
+                            ErrorMsg(
+                                message=f"Invalid model '{model_name}'. Valid: {WHISPER_ALLOWLIST}",
+                                code="INVALID_MODEL",
+                            ).model_dump_json()
+                        )
+                    elif _model_switch_lock.locked():
+                        await websocket.send_text(
+                            ErrorMsg(
+                                message="Model switch already in progress, try again shortly",
+                                code="MODEL_SWITCH_BUSY",
+                            ).model_dump_json()
+                        )
+                    else:
+                        asyncio.create_task(
+                            _switch_model(websocket.app, model_name, websocket, _model_switch_lock)
+                        )
 
     except WebSocketDisconnect:
         logger.info("Session %s disconnected", session_id)
     finally:
-        if processor is not None:
-            processor.reset()
+        if ts_service is not None:
+            await ts_service.stop()
         active_sessions = getattr(websocket.app.state, "active_sessions", {})
         if session_id and session_id in active_sessions:
             del active_sessions[session_id]
+
+
+async def _switch_model(app, model_name: str, websocket: WebSocket, lock: asyncio.Lock) -> None:
+    async with lock:
+        await websocket.send_text(
+            ModelChangedMsg(model=model_name, status="loading").model_dump_json()
+        )
+        try:
+            engine = get_whisper_engine(model_name)
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(None, engine.load_model)
+            app.state.asr_engine = engine
+            logger.info("Model switched to %s", model_name)
+            await websocket.send_text(
+                ModelChangedMsg(model=model_name, status="ready").model_dump_json()
+            )
+        except Exception as exc:
+            logger.error("Model switch failed: %s", exc)
+            await websocket.send_text(
+                ErrorMsg(message=str(exc), code="MODEL_SWITCH_ERROR").model_dump_json()
+            )

server/app/schemas/messages.py

@@ -33,10 +33,17 @@ class TranscriptionMsg(BaseModel):
 class SuggestionMsg(BaseModel):
     type: Literal["suggestion"] = "suggestion"
     text: str
-    category: Literal["question", "insight", "action"]
+    category: Literal["answer", "question", "action", "summary", "flag"]
+    context: str = ""
     timestamp: float
 
 
+class ModelChangedMsg(BaseModel):
+    type: Literal["model_changed"] = "model_changed"
+    model: str
+    status: Literal["loading", "ready"]
+
+
 class ErrorMsg(BaseModel):
     type: Literal["error"] = "error"
     message: str
@@ -55,4 +62,5 @@ class AuthFrame(BaseModel):
 
 
 class ControlFrame(BaseModel):
-    type: Literal["start_session", "stop_session", "ping"]
+    type: Literal["start_session", "stop_session", "ping", "set_model"]
+    model: str | None = None

server/app/services/audio_processor.py

@@ -1,84 +1,14 @@
 from __future__ import annotations
 
-import logging
-from collections import deque
-
 import numpy as np
-import torch
-
-from app.config import settings
-
-logger = logging.getLogger(__name__)
 
 
 class AudioProcessor:
-    """
-    Accepts 300 ms PCM chunks (16-bit signed, 16 kHz mono) from the WebSocket,
-    runs Silero VAD on each chunk, and returns a float32 numpy array whenever a
-    complete speech segment ends (silence threshold crossed or 30-s hard cap hit).
-    """
+    """Converts raw PCM bytes to normalized float32 ndarray. VAD is handled client-side."""
 
-    SAMPLE_RATE = 16000
-    MIN_SPEECH_SAMPLES = SAMPLE_RATE // 2   # 0.5 s — ignore sub-sentence blips
-    MAX_SPEECH_SAMPLES = SAMPLE_RATE * 30   # 30 s hard cap → forced flush
-
-    def __init__(self, vad_model=None) -> None:
-        if vad_model is not None:
-            self._vad_model = vad_model
-        else:
-            from silero_vad import load_silero_vad  # deferred: requires torch
-            self._vad_model = load_silero_vad()
-        self._threshold = settings.vad_threshold
-        # How many consecutive silent chunks trigger a flush
-        self._max_silence_chunks = max(1, round(settings.max_silence_ms / settings.chunk_duration_ms))
-        self._speech_buffer: deque[float] = deque(maxlen=self.MAX_SPEECH_SAMPLES)
-        self._silence_chunks: int = 0
-
-    def process_chunk(self, pcm_bytes: bytes) -> np.ndarray | None:
-        """
-        Process one PCM frame. Returns a speech segment (float32) when speech ends,
-        otherwise None.
-        """
+    def process_chunk(self, pcm_bytes: bytes) -> np.ndarray:
         audio = np.frombuffer(pcm_bytes, dtype=np.int16).astype(np.float32) / 32768.0
-        has_speech = self._has_speech(audio)
-
-        if has_speech:
-            self._speech_buffer.extend(audio.tolist())
-            self._silence_chunks = 0
-        else:
-            self._silence_chunks += 1
-            if (
-                self._silence_chunks >= self._max_silence_chunks
-                and len(self._speech_buffer) >= self.MIN_SPEECH_SAMPLES
-            ):
-                return self._flush()
-
-        if len(self._speech_buffer) >= self.MAX_SPEECH_SAMPLES:
-            logger.warning("Speech buffer at 30-s cap, forcing flush")
-            return self._flush()
-
-        return None
+        return np.clip(audio, -1.0, 1.0)
 
     def reset(self) -> None:
-        self._speech_buffer.clear()
-        self._silence_chunks = 0
-
-    # ── private ───────────────────────────────────────────────────────────────
-
-    def _flush(self) -> np.ndarray:
-        segment = np.array(list(self._speech_buffer), dtype=np.float32)
-        self._speech_buffer.clear()
-        self._silence_chunks = 0
-        return segment
-
-    def _has_speech(self, audio: np.ndarray) -> bool:
-        from silero_vad import get_speech_timestamps  # deferred: requires torch
-
-        tensor = torch.tensor(audio)
-        stamps = get_speech_timestamps(
-            tensor,
-            self._vad_model,
-            threshold=self._threshold,
-            sampling_rate=self.SAMPLE_RATE,
-        )
-        return len(stamps) > 0
+        pass

server/app/services/groq_client.py

@@ -4,69 +4,108 @@ import asyncio
 import json
 import logging
 import re
-from collections import deque
+import time
 from dataclasses import dataclass
-from typing import Literal
+from typing import TYPE_CHECKING, Literal
+
+if TYPE_CHECKING:
+    from app.services.session_context import SessionContext
 
 logger = logging.getLogger(__name__)
 
 _SYSTEM_PROMPT = """\
-You are a real-time meeting assistant.
-Analyze the recent meeting transcript and generate ONE brief, actionable suggestion.
+You are a real-time meeting coach embedded in the participant's browser.
+Analyze the full meeting context and recent transcript, then generate ONE response.
+
+Rules:
+- If a question was asked but not answered → provide a direct, concise answer
+- If a claim needs challenging → suggest a clarifying question
+- If a decision or action item was implied → flag it explicitly
+- If a topic just resolved → offer a brief summary
+- If conversation stalled → suggest a re-engagement prompt
+
+Respond ONLY in this exact JSON format:
+{"category": "answer|question|action|summary|flag", "text": "...", "context": "one-line reason why this is relevant now"}
+
+Keep "text" under 20 words. Keep "context" under 10 words. Be specific and direct."""
 
-Respond ONLY in this exact JSON format — no other text:
-{"category": "question", "text": "..."}
+_SUMMARY_PROMPT = """\
+You are summarizing a business meeting.
 
-Categories:
-- "question": A clarifying question the participant could ask
-- "insight": A key observation or important point worth noting
-- "action": A concrete next step or task to follow up on
+Current summary:
+{current_summary}
 
-Keep the suggestion under 15 words. Be specific and direct."""
+New transcript segments:
+{new_segments}
+
+Write an updated summary under 300 words. Preserve all decisions, action items, and open questions. Use bullet points."""
 
 _MAX_RETRIES = 3
-_MAX_CONTEXT_SEGMENTS = 10
+_SUMMARY_TRIGGER = 10
+_SUGGESTION_DEBOUNCE_S = 2.0
+VALID_CATEGORIES = ("answer", "question", "action", "summary", "flag")
 
 
 @dataclass
 class SuggestionResult:
     text: str
-    category: Literal["question", "insight", "action"] = "insight"
+    category: Literal["answer", "question", "action", "summary", "flag"] = "question"
+    context: str = ""
 
 
 class GroqClient:
-    """
-    Async Groq API client with per-session transcript context, exponential
-    backoff (3 retries), and JSON-based category classification.
-    """
-
-    MODEL = "llama-3.1-8b-instant"
-
-    def __init__(self, api_key: str, _client=None) -> None:
+    MODEL = "llama-3.3-70b-versatile"
+
+    def __init__(
+        self,
+        api_key: str,
+        context: "SessionContext | None" = None,
+        _client=None,
+    ) -> None:
         if _client is not None:
             self._client = _client
         else:
-            from groq import AsyncGroq  # deferred: not needed if key absent
+            from groq import AsyncGroq
             self._client = AsyncGroq(api_key=api_key)
-        self._context: deque[str] = deque(maxlen=_MAX_CONTEXT_SEGMENTS)
+
+        if context is not None:
+            self._context = context
+        else:
+            from app.services.session_context import SessionContext
+            self._context = SessionContext()
+
+        self._last_suggestion_time: float = 0.0
+        self._background_tasks: set[asyncio.Task] = set()
 
     def add_transcription(self, text: str) -> None:
-        """Append a transcription segment to the rolling context window."""
-        if text.strip():
-            self._context.append(text.strip())
+        if not text.strip():
+            return
+        self._context.recent.append(text.strip())
+        self._context.unsummarized_count += 1
+        if self._context.unsummarized_count >= _SUMMARY_TRIGGER:
+            task = asyncio.create_task(self._update_summary())
+            self._background_tasks.add(task)
+            task.add_done_callback(self._background_tasks.discard)
+            self._context.unsummarized_count = 0
 
     async def get_suggestion(self) -> SuggestionResult | None:
-        """
-        Call Groq with the current context. Retries up to 3 times with
-        exponential backoff on rate-limit and 5xx errors.
-        Returns None on failure or empty context.
-        """
-        if not self._context:
+        if not self._context.recent:
             return None
 
-        user_prompt = "Recent transcript:\n" + "\n".join(
-            f"[{i + 1}] {seg}" for i, seg in enumerate(self._context)
-        ) + "\n\nGenerate one suggestion."
+        now = time.monotonic()
+        if now - self._last_suggestion_time < _SUGGESTION_DEBOUNCE_S:
+            return None
+        self._last_suggestion_time = now
+
+        summary_section = self._context.summary or "Meeting just started."
+        recent_transcript = "\n".join(
+            f"[{i + 1}] {seg}" for i, seg in enumerate(self._context.recent)
+        )
+        user_prompt = (
+            f"Meeting context so far:\n{summary_section}\n\n"
+            f"Recent transcript:\n{recent_transcript}\n\n"
+            "Generate one coaching response."
+        )
 
         for attempt in range(_MAX_RETRIES):
             try:
@@ -84,32 +123,49 @@ class GroqClient:
                 if result:
                     logger.info("Groq suggestion [%s]: %r", result.category, result.text)
                 return result
-
             except Exception as exc:
                 should_retry = _is_retryable(exc)
                 if not should_retry or attempt == _MAX_RETRIES - 1:
                     logger.error("Groq failed after %d attempt(s): %s", attempt + 1, exc)
                     return None
                 wait = 2 ** attempt
-                logger.warning("Groq attempt %d/%d failed (%s), retrying in %ds",
-                               attempt + 1, _MAX_RETRIES, exc, wait)
+                logger.warning("Groq retry %d/%d in %ds", attempt + 1, _MAX_RETRIES, wait)
                 await asyncio.sleep(wait)
 
         return None
 
+    async def _update_summary(self) -> None:
+        if not self._context.recent:
+            return
+        new_segments = "\n".join(f"- {seg}" for seg in self._context.recent)
+        prompt = _SUMMARY_PROMPT.format(
+            current_summary=self._context.summary or "No summary yet.",
+            new_segments=new_segments,
+        )
+        try:
+            response = await self._client.chat.completions.create(
+                model=self.MODEL,
+                messages=[{"role": "user", "content": prompt}],
+                temperature=0.3,
+                max_tokens=400,
+            )
+            new_summary = (response.choices[0].message.content or "").strip()
+            if new_summary:
+                self._context.summary = new_summary
+                logger.info("Meeting summary updated (%d chars)", len(new_summary))
+        except Exception as exc:
+            logger.warning("Summary update failed: %s", exc)
+
 
 # ── helpers ────────────────────────────────────────────────────────────────────
 
 def _parse_suggestion(raw: str) -> SuggestionResult | None:
-    """Extract JSON from LLM output; fall back to regex on malformed responses."""
-    # Try direct JSON parse first
     try:
         data = json.loads(raw.strip())
         return _build_result(data)
     except (json.JSONDecodeError, ValueError):
         pass
 
-    # Regex fallback: grab the first {...} block
     match = re.search(r"\{[^}]+\}", raw, re.DOTALL)
     if match:
         try:
@@ -126,14 +182,14 @@ def _build_result(data: dict) -> SuggestionResult | None:
     text = str(data.get("text", "")).strip()
     if not text:
         return None
-    category = data.get("category", "insight")
-    if category not in ("question", "insight", "action"):
-        category = "insight"
-    return SuggestionResult(text=text, category=category)
+    category = data.get("category", "question")
+    if category not in VALID_CATEGORIES:
+        category = "question"
+    context = str(data.get("context", "")).strip()
+    return SuggestionResult(text=text, category=category, context=context)
 
 
 def _is_retryable(exc: Exception) -> bool:
-    """Return True for rate-limit and server-side errors worth retrying."""
     try:
         from groq import APIStatusError, RateLimitError
         if isinstance(exc, RateLimitError):
@@ -142,5 +198,4 @@ def _is_retryable(exc: Exception) -> bool:
             return exc.status_code >= 500
     except ImportError:
         pass
-    # Retry on generic network errors
     return isinstance(exc, (TimeoutError, OSError))

server/app/services/session_context.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+from collections import deque
+from dataclasses import dataclass, field
+
+
+@dataclass
+class SessionContext:
+    summary: str = ""
+    recent: deque = field(default_factory=lambda: deque(maxlen=5))
+    unsummarized_count: int = 0

server/app/services/transcription.py

@@ -15,11 +15,6 @@ logger = logging.getLogger(__name__)
 
 
 class TranscriptionService:
-    """
-    Runs ASR in a thread-pool executor, sends a transcription WS event, then
-    fires a Groq suggestion request concurrently (non-blocking).
-    """
-
     def __init__(
         self,
         websocket: WebSocket,
@@ -29,11 +24,55 @@ class TranscriptionService:
         self._ws = websocket
         self._engine = engine
         self._groq = groq_client
+        self._queue: asyncio.Queue[np.ndarray] = asyncio.Queue(maxsize=3)
+        self._worker_task: asyncio.Task | None = None
+        self._pending_suggestions: set[asyncio.Task] = set()
+
+    def start(self) -> None:
+        self._worker_task = asyncio.create_task(self._worker())
+
+    async def stop(self) -> None:
+        if self._worker_task:
+            self._worker_task.cancel()
+            try:
+                await self._worker_task
+            except asyncio.CancelledError:
+                pass
+        for task in list(self._pending_suggestions):
+            task.cancel()
+        if self._pending_suggestions:
+            await asyncio.gather(*self._pending_suggestions, return_exceptions=True)
 
-    async def transcribe_and_send(self, audio: np.ndarray) -> None:
-        if not self._engine.is_loaded():
-            logger.warning("ASR engine not ready — dropping segment (%d samples)", len(audio))
+    def enqueue(self, audio: np.ndarray) -> None:
+        if not self._engine or not self._engine.is_loaded():
             return
+        if self._queue.full():
+            try:
+                self._queue.get_nowait()
+                self._queue.task_done()
+            except asyncio.QueueEmpty:
+                pass
+        try:
+            self._queue.put_nowait(audio)
+        except asyncio.QueueFull:
+            pass
+
+    async def _worker(self) -> None:
+        while True:
+            audio = await self._queue.get()
+            try:
+                await self._process(audio)
+            except Exception as exc:
+                logger.exception("Worker error: %s", exc)
+            finally:
+                self._queue.task_done()
+
+    async def _process(self, audio: np.ndarray) -> None:
+        await self._ws.send_text(
+            TranscriptionMsg(
+                text="", is_final=False, timestamp=time.time(), confidence=0.0
+            ).model_dump_json()
+        )
 
         t0 = time.perf_counter()
         try:
@@ -52,7 +91,6 @@ class TranscriptionService:
         if not result.text:
             return
 
-        # ── Send transcription immediately ────────────────────────────────────
         await self._ws.send_text(
             TranscriptionMsg(
                 text=result.text,
@@ -62,13 +100,13 @@ class TranscriptionService:
             ).model_dump_json()
         )
 
-        # ── Fire Groq suggestion concurrently (non-blocking) ─────────────────
         if self._groq is not None:
             self._groq.add_transcription(result.text)
-            asyncio.create_task(self._send_suggestion())
+            task = asyncio.create_task(self._send_suggestion())
+            self._pending_suggestions.add(task)
+            task.add_done_callback(self._pending_suggestions.discard)
 
     async def _send_suggestion(self) -> None:
-        """Fetch a suggestion from Groq and send it; never raises."""
         assert self._groq is not None
         try:
             suggestion = await self._groq.get_suggestion()
@@ -78,6 +116,7 @@ class TranscriptionService:
                 SuggestionMsg(
                     text=suggestion.text,
                     category=suggestion.category,
+                    context=suggestion.context,
                     timestamp=time.time(),
                 ).model_dump_json()
             )

uv.lock

@@ -1290,16 +1290,15 @@ wheels = [
 
 [[package]]
 name = "meeting-ai-assistant"
-version = "0.1.0"
+version = "0.2.0"
 source = { virtual = "." }
 dependencies = [
     { name = "fastapi" },
+    { name = "faster-whisper" },
     { name = "groq" },
     { name = "numpy" },
     { name = "pydantic-settings" },
     { name = "python-dotenv" },
-    { name = "silero-vad" },
-    { name = "torch" },
     { name = "uvicorn", extra = ["standard"] },
     { name = "websockets" },
 ]
@@ -1309,9 +1308,6 @@ parakeet = [
     { name = "nemo-toolkit", extra = ["asr"] },
     { name = "soundfile" },
 ]
-whisper = [
-    { name = "faster-whisper" },
-]
 
 [package.dev-dependencies]
 dev = [
@@ -1324,19 +1320,17 @@ dev = [
 [package.metadata]
 requires-dist = [
     { name = "fastapi", specifier = ">=0.111.0" },
-    { name = "faster-whisper", marker = "extra == 'whisper'", specifier = ">=1.0.0" },
+    { name = "faster-whisper", specifier = ">=1.0.0" },
     { name = "groq", specifier = ">=0.9.0" },
     { name = "nemo-toolkit", extras = ["asr"], marker = "extra == 'parakeet'", specifier = ">=1.23.0" },
     { name = "numpy", specifier = ">=1.24.0" },
     { name = "pydantic-settings", specifier = ">=2.2.0" },
     { name = "python-dotenv", specifier = ">=1.0.0" },
-    { name = "silero-vad", specifier = ">=4.0.0" },
     { name = "soundfile", marker = "extra == 'parakeet'", specifier = ">=0.12.0" },
-    { name = "torch", specifier = ">=2.0.0" },
     { name = "uvicorn", extras = ["standard"], specifier = ">=0.29.0" },
     { name = "websockets", specifier = ">=12.0" },
 ]
-provides-extras = ["parakeet", "whisper"]
+provides-extras = ["parakeet"]
 
 [package.metadata.requires-dev]
 dev = [
@@ -2580,20 +2574,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/e1/e3/c164c88b2e5ce7b24d667b9bd83589cf4f3520d97cad01534cd3c4f55fdb/setuptools-81.0.0-py3-none-any.whl", hash = "sha256:fdd925d5c5d9f62e4b74b30d6dd7828ce236fd6ed998a08d81de62ce5a6310d6", size = 1062021, upload-time = "2026-02-06T21:10:37.175Z" },
 ]
 
-[[package]]
-name = "silero-vad"
-version = "6.2.1"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "packaging" },
-    { name = "torch" },
-    { name = "torchaudio" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/32/d3/e31f526482782764aa4f70e20fd4545cf2e4a81a60b6fb0f089f6d107991/silero_vad-6.2.1.tar.gz", hash = "sha256:b23062b0e39fad17b1266fc23c1e7b4290219dbe82ce08510889e32f681f4b3b", size = 28913811, upload-time = "2026-02-24T08:41:59.329Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/0b/2b/48566f29a8b53d856ceb1994f209122749b3fda0a733a07e82047257de7a/silero_vad-6.2.1-py3-none-any.whl", hash = "sha256:09de93c4d874bb19c53e62a47dd38be5f163cedad2b5599583231f2a84ef79cb", size = 9146242, upload-time = "2026-02-24T08:41:56.955Z" },
-]
-
 [[package]]
 name = "six"
 version = "1.17.0"
@@ -2876,17 +2856,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/d1/bd/9912d30b68845256aabbb4a40aeefeef3c3b20db5211ccda653544ada4b6/torch-2.11.0-cp311-cp311-win_amd64.whl", hash = "sha256:73e24aaf8f36ab90d95cd1761208b2eb70841c2a9ca1a3f9061b39fc5331b708", size = 114519675, upload-time = "2026-03-23T18:11:52.995Z" },
 ]
 
-[[package]]
-name = "torchaudio"
-version = "2.11.0"
-source = { registry = "https://pypi.org/simple" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/94/77/0eec7f175d88f312296bd5b11c23bd58da37c1021f53da3db4df449ce3ee/torchaudio-2.11.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:492dd64645e9d0bb843e94f1d9a4d1e31426262ffc594fafecc1697df9df5eb9", size = 684142, upload-time = "2026-03-23T18:13:36.805Z" },
-    { url = "https://files.pythonhosted.org/packages/b3/f9/6f7ebe071b44592c85269762b55b63ab0a091b5f479f73544738f7564a1e/torchaudio-2.11.0-cp311-cp311-manylinux_2_28_aarch64.whl", hash = "sha256:73dab4841f94d888bc7c2aed7b5547c643edc974306919fe1adfb65d57cccf4b", size = 1626527, upload-time = "2026-03-23T18:13:39.011Z" },
-    { url = "https://files.pythonhosted.org/packages/ac/70/17408e0d154d0c894537a88dcbadc48e8ad3b6e1ef4a1dabda5d40245ee0/torchaudio-2.11.0-cp311-cp311-manylinux_2_28_x86_64.whl", hash = "sha256:1a07ec72fd6f26a588c39b5f029e0130d16bb40bc4221635580bf8fb18fcbc80", size = 1771930, upload-time = "2026-03-23T18:13:37.963Z" },
-    { url = "https://files.pythonhosted.org/packages/c9/75/b6d03fc75b409bdaec597274d1bdd4213db716ed16f6801386b31d59c551/torchaudio-2.11.0-cp311-cp311-win_amd64.whl", hash = "sha256:bb59ba4452bbbe95d75ad3ef18df9824955625f36698ce9a5998a4a9f3c1ba1d", size = 328658, upload-time = "2026-03-23T18:13:44.545Z" },
-]
-
 [[package]]
 name = "torchmetrics"
 version = "1.9.0"