added pipeline file

app.py

@@ -0,0 +1,371 @@
+"""
+app.py — Telecalling Agent — Gradio 6 UI
+
+Layout
+──────
+┌─────────────────────────────────────────────────────┐
+│  📞 AI Telecalling Agent          [status badge]    │
+├──────────────────────┬──────────────────────────────┤
+│  🎤 LIVE CALL        │  📋 EXTRACTED DATA           │
+│  ┌────────────────┐  │  [intent markdown table]     │
+│  │ Audio stream   │  │                              │
+│  └────────────────┘  ├──────────────────────────────┤
+│  [Start] [End]       │  🤖 AGENT RESPONSE           │
+│  ┌────────────────┐  │  [spoken response box]       │
+│  │ Transcript     │  │                              │
+│  └────────────────┘  │  ✅ BOOKING CONFIRMED        │
+│                      │  [booking details box]       │
+├──────────────────────┴──────────────────────────────┤
+│  📁 CALL LOG                                        │
+│  [dataframe — recent calls]                         │
+└─────────────────────────────────────────────────────┘
+"""
+
+import logging
+import os
+import json
+
+import gradio as gr
+import numpy as np
+
+from config import APP_TITLE, APP_DESCRIPTION, SERVER_PORT, SERVER_NAME
+from pipeline.orchestrator import CallSession, PipelineUpdate
+from db import init_db
+
+# Load HuggingFace config and set token early
+try:
+    with open("hf_config.json", "r") as f:
+        hf_cfg = json.load(f)
+        hf_token = hf_cfg.get("huggingface", {}).get("hub", {}).get("token", "")
+        if hf_token and hf_token != "${HF_TOKEN}":
+            os.environ["HF_TOKEN"] = hf_token
+except (FileNotFoundError, json.JSONDecodeError) as e:
+    pass  # hf_config.json not found or invalid, use env var if set
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+# Initialize database on startup
+init_db()
+
+
+# ── CSS ───────────────────────────────────────────────────────────────────────
+
+CSS = """
+/* Global */
+.gradio-container { font-family: 'Inter', sans-serif; max-width: 1200px; }
+
+/* Status badge */
+#status-badge textarea {
+    font-size: 0.9rem;
+    font-weight: 600;
+    text-align: center;
+    border-radius: 20px;
+    padding: 4px 12px;
+    background: #f0fdf4;
+    border: 1px solid #86efac;
+    color: #166534;
+}
+
+/* Agent response */
+#agent-box textarea {
+    font-size: 1.05rem;
+    font-style: italic;
+    background: #eff6ff;
+    border: 1px solid #93c5fd;
+    border-radius: 8px;
+    color: #1e3a5f;
+    min-height: 80px;
+}
+
+/* Booking confirmed */
+#booking-box textarea {
+    background: #f0fdf4;
+    border: 1px solid #4ade80;
+    border-radius: 8px;
+    color: #14532d;
+    font-weight: 500;
+}
+
+/* Transcript */
+#transcript-box textarea {
+    font-family: monospace;
+    font-size: 0.85rem;
+    background: #1e1e2e;
+    color: #cdd6f4;
+    border-radius: 8px;
+    min-height: 180px;
+}
+
+/* VAD indicator dot */
+#vad-dot {
+    text-align: center;
+    font-size: 1.2rem;
+}
+
+/* Call buttons */
+.call-btn-start { background: #16a34a !important; color: white !important; }
+.call-btn-end   { background: #dc2626 !important; color: white !important; }
+.call-btn-reset { background: #6b7280 !important; color: white !important; }
+
+/* Intent table inside Markdown */
+#intent-panel table { width: 100%; border-collapse: collapse; font-size: 0.88rem; }
+#intent-panel th, #intent-panel td {
+    padding: 5px 10px;
+    border: 1px solid #e2e8f0;
+    text-align: left;
+}
+#intent-panel tr:nth-child(even) { background: #f8fafc; }
+
+/* Call log */
+#call-log { font-size: 0.82rem; }
+"""
+
+
+# ── UI helpers ────────────────────────────────────────────────────────────────
+
+def _format_transcript(lines: list[str]) -> str:
+    if not lines:
+        return "(waiting for speech…)"
+    return "\n".join(f"[{i+1}] {l}" for i, l in enumerate(lines))
+
+
+def _format_booking(info: dict | None) -> str:
+    if not info:
+        return ""
+    return (
+        f"✅  Booking #{info['booking_id']} confirmed!\n"
+        f"    📅  {info['date']}  🕐  {info['time']}  "
+        f"({info['duration']} min)\n"
+        f"    👤  {info['caller']}   📞  {info['type'].replace('_', ' ').title()}"
+    )
+
+
+def _call_log_rows(records: list[dict]) -> list[list]:
+    rows = []
+    for r in records:
+        ts = r.get("timestamp", "")[:16].replace("T", " ")
+        rows.append([
+            r.get("id", ""),
+            ts,
+            r.get("caller_name") or "—",
+            r.get("intent")      or "—",
+            r.get("decision")    or "—",
+            r.get("status")      or "—",
+        ])
+    return rows
+
+
+# ── Gradio App ────────────────────────────────────────────────────────────────
+
+def build_app() -> gr.Blocks:
+
+    with gr.Blocks(css=CSS, title=APP_TITLE, theme=gr.themes.Soft()) as demo:
+
+        # ── Per-session state ──────────────────────────────────────────────
+        # gr.State holds one CallSession object per browser tab.
+        session_state = gr.State(value=None)
+
+        # ── Header ─────────────────────────────────────────────────────────
+        gr.Markdown(f"# {APP_TITLE}\n_{APP_DESCRIPTION}_")
+
+        status_badge = gr.Textbox(
+            value       = "🟢 Ready — press Start Call",
+            label       = "",
+            interactive = False,
+            elem_id     = "status-badge",
+        )
+
+        # ── Main row ───────────────────────────────────────────────────────
+        with gr.Row():
+
+            # ── Left column: call controls + transcript ────────────────────
+            with gr.Column(scale=1):
+                gr.Markdown("### 🎤 Live Call")
+
+                audio_input = gr.Audio(
+                    sources    = ["microphone"],
+                    streaming  = True,
+                    type       = "numpy",
+                    label      = "Microphone (auto-detecting speech)",
+                    interactive= True,
+                    elem_id    = "audio-input",
+                )
+
+                vad_dot = gr.Markdown("⚫ _mic idle_", elem_id="vad-dot")
+
+                with gr.Row():
+                    btn_start = gr.Button(
+                        "📞 Start Call", variant="primary",
+                        elem_classes=["call-btn-start"],
+                    )
+                    btn_end = gr.Button(
+                        "📵 End Call", variant="stop",
+                        elem_classes=["call-btn-end"],
+                    )
+                    btn_reset = gr.Button(
+                        "🔄 Reset", variant="secondary",
+                        elem_classes=["call-btn-reset"],
+                    )
+
+                transcript_box = gr.Textbox(
+                    label       = "📝 Live Transcript",
+                    value       = "(waiting for speech…)",
+                    lines       = 8,
+                    max_lines   = 20,
+                    interactive = False,
+                    elem_id     = "transcript-box",
+                )
+
+            # ── Right column: intent + agent response + booking ────────────
+            with gr.Column(scale=1):
+                gr.Markdown("### 📋 Extracted Data")
+
+                intent_panel = gr.Markdown(
+                    "_No data yet — waiting for first utterance…_",
+                    elem_id = "intent-panel",
+                )
+
+                gr.Markdown("### 🤖 Agent Response")
+
+                agent_box = gr.Textbox(
+                    value       = "",
+                    label       = "",
+                    lines       = 3,
+                    interactive = False,
+                    elem_id     = "agent-box",
+                    placeholder = "Agent will respond here…",
+                )
+
+                booking_box = gr.Textbox(
+                    value       = "",
+                    label       = "📅 Booking Status",
+                    lines       = 3,
+                    interactive = False,
+                    elem_id     = "booking-box",
+                    visible     = False,
+                )
+
+        # ── Call log ───────────────────────────────────────────────────────
+        gr.Markdown("### 📁 Call Log")
+
+        call_log_table = gr.Dataframe(
+            headers     = ["ID", "Timestamp", "Caller", "Intent", "Decision", "Status"],
+            datatype    = ["number", "str", "str", "str", "str", "str"],
+            value       = [],
+            interactive = False,
+            elem_id     = "call-log",
+            row_count   = (5, "dynamic"),
+        )
+
+        # ── Helper: unpack PipelineUpdate → tuple of component values ─────
+        def _unpack(u: PipelineUpdate):
+            """Return values in the exact order of outputs lists below."""
+            vad_label = "🔴 _Speaking…_" if u.vad_speaking else "⚫ _mic idle_"
+            booking_text    = _format_booking(u.booking_confirmed)
+            booking_visible = bool(booking_text)
+            return (
+                u.status,                           # status_badge
+                vad_label,                          # vad_dot
+                _format_transcript(u.transcript_lines),  # transcript_box
+                u.intent_md,                        # intent_panel
+                u.agent_response,                   # agent_box
+                booking_text,                       # booking_box value
+                gr.update(visible=booking_visible), # booking_box visible
+                _call_log_rows(u.call_log),         # call_log_table
+            )
+
+        # ── All output components in one list (matches _unpack order) ─────
+        ALL_OUTPUTS = [
+            status_badge,
+            vad_dot,
+            transcript_box,
+            intent_panel,
+            agent_box,
+            booking_box,
+            booking_box,       # second entry → gr.update(visible=…)
+            call_log_table,
+        ]
+
+        # ── Session factory ────────────────────────────────────────────────
+        def _get_or_create_session(state):
+            if state is None:
+                state = CallSession()
+            return state
+
+        # ── Button callbacks ───────────────────────────────────────────────
+
+        def on_start(state):
+            state = _get_or_create_session(state)
+            update = state.start_call()
+            return (state, *_unpack(update))
+
+        def on_end(state):
+            state = _get_or_create_session(state)
+            update = state.end_call()
+            return (state, *_unpack(update))
+
+        def on_reset(state):
+            state = _get_or_create_session(state)
+            update = state.reset()
+            return (state, *_unpack(update))
+
+        BTN_OUTPUTS = [session_state] + ALL_OUTPUTS
+
+        btn_start.click(on_start, inputs=[session_state], outputs=BTN_OUTPUTS)
+        btn_end.click  (on_end,   inputs=[session_state], outputs=BTN_OUTPUTS)
+        btn_reset.click(on_reset, inputs=[session_state], outputs=BTN_OUTPUTS)
+
+        # ── Audio streaming callback ───────────────────────────────────────
+        # Fires every `stream_every` seconds with (sample_rate, np.ndarray).
+        # We pass the current session state in and get it back (updated).
+
+        def on_audio_stream(audio_chunk, state):
+            """
+            Called by Gradio every 0.5 s while the mic is active.
+            audio_chunk: (sample_rate: int, data: np.ndarray) | None
+            """
+            state = _get_or_create_session(state)
+
+            if not state.call_active:
+                # Return current state without processing
+                u = state._build_update()
+                return (state, *_unpack(u))
+
+            if audio_chunk is None:
+                u = state._build_update()
+                return (state, *_unpack(u))
+
+            sample_rate, audio_np = audio_chunk
+
+            # Ensure float32 mono
+            audio_np = np.array(audio_np, dtype=np.float32)
+            if audio_np.ndim == 2:
+                audio_np = audio_np.mean(axis=1)
+
+            update = state.process_audio_chunk(sample_rate, audio_np)
+            return (state, *_unpack(update))
+
+        audio_input.stream(
+            fn           = on_audio_stream,
+            inputs       = [audio_input, session_state],
+            outputs      = [session_state] + ALL_OUTPUTS,
+            stream_every = 0.5,      # seconds — half-second chunks
+            time_limit   = 3600,     # allow up to 1-hour calls
+        )
+
+    return demo
+
+
+# ── Entry point ───────────────────────────────────────────────────────────────
+
+if __name__ == "__main__":
+    app = build_app()
+    app.launch(
+        server_name = SERVER_NAME,
+        server_port = SERVER_PORT,
+        show_error  = True,
+    )

config.py

@@ -0,0 +1,71 @@
+"""
+Central configuration for the telecalling agent.
+Edit the values in this file to match your local setup.
+"""
+
+import os
+from pathlib import Path
+
+# ── Project root ──────────────────────────────────────────────────────────────
+ROOT_DIR   = Path(__file__).parent
+DATA_DIR   = ROOT_DIR / "data"
+MODELS_DIR = ROOT_DIR / "models"
+
+DATA_DIR.mkdir(exist_ok=True)
+MODELS_DIR.mkdir(exist_ok=True)
+
+# ── Hugging Face ──────────────────────────────────────────────────────────────
+# Required: Cohere Transcribe is gated — you must accept the license at
+# https://huggingface.co/CohereLabs/cohere-transcribe-03-2026
+# then set your HF token here or in the environment variable HF_TOKEN.
+HF_TOKEN = os.environ.get("HF_TOKEN", "")
+
+# ── ASR: Cohere Transcribe ────────────────────────────────────────────────────
+TRANSCRIBE_MODEL_ID = "CohereLabs/cohere-transcribe-03-2026"
+TRANSCRIBE_LANGUAGE = "en"          # ISO 639-1 — English only for now
+TRANSCRIBE_DEVICE   = "cuda:0"      # falls back to "cpu" automatically in code
+
+# ── Intent Parser: Qwen2.5-7B-Instruct (GGUF via llama-cpp-python) ────────────
+# Download command (run once):
+#   huggingface-cli download Qwen/Qwen2.5-7B-Instruct-GGUF \
+#       --include "qwen2.5-7b-instruct-q4_k_m.gguf" \
+#       --local-dir ./models
+QWEN_GGUF_PATH    = MODELS_DIR / "qwen2.5-7b-instruct-q4_k_m.gguf"
+QWEN_N_GPU_LAYERS = 20     # offload 20 transformer layers to GPU (~0.8 GB VRAM)
+                           # remaining ~15 layers run on CPU RAM
+QWEN_N_CTX        = 4096   # context window — sufficient for a call transcript
+QWEN_MAX_TOKENS   = 512    # max tokens for the structured JSON response
+QWEN_TEMPERATURE  = 0.1    # near-deterministic for structured output
+
+# ── Evaluator: MiniCPM3-4B (CPU, bitsandbytes 4-bit) ─────────────────────────
+MINICPM_MODEL_ID   = "openbmb/MiniCPM3-4B"
+MINICPM_DEVICE     = "cpu"   # runs after Qwen is done — no VRAM conflict
+MINICPM_MAX_TOKENS = 256
+
+# ── VAD: Silero VAD (ONNX, CPU) ───────────────────────────────────────────────
+VAD_SAMPLE_RATE        = 16000   # Hz — Silero and Cohere both want 16kHz
+VAD_CHUNK_MS           = 250     # ms per audio chunk fed to VAD
+VAD_CHUNK_SAMPLES      = int(VAD_SAMPLE_RATE * VAD_CHUNK_MS / 1000)  # 4000
+VAD_SILENCE_THRESHOLD  = 0.5     # Silero confidence below this → silence
+VAD_SILENCE_DURATION_S = 0.8     # seconds of silence before utterance is flushed
+VAD_MIN_SPEECH_S       = 0.5     # ignore utterances shorter than this (noise)
+
+# ── SQLite database ───────────────────────────────────────────────────────────
+DB_PATH = DATA_DIR / "calls.db"
+
+# ── Scheduling rules (evaluated by MiniCPM) ──────────────────────────────────
+# These are injected into MiniCPM's system prompt so it can reason about them.
+SCHEDULING_RULES = """
+1. Meetings can only be booked Monday–Friday, 09:00–18:00.
+2. Minimum meeting duration is 15 minutes; maximum is 120 minutes.
+3. Back-to-back meetings are not allowed — require a 15-minute gap between slots.
+4. If the caller does not provide a date or time, ask for one before confirming.
+5. If the requested slot is already booked, suggest the next available slot.
+6. Always confirm the caller's name before booking.
+"""
+
+# ── Gradio UI ─────────────────────────────────────────────────────────────────
+APP_TITLE       = "📞 AI Telecalling Agent"
+APP_DESCRIPTION = "Speak naturally — the agent will schedule your meeting automatically."
+SERVER_PORT     = 7860
+SERVER_NAME     = "0.0.0.0"   # bind to all interfaces for HF Spaces

db.py

@@ -0,0 +1,227 @@
+"""
+Database layer — SQLite via Python's built-in sqlite3.
+Handles schema creation and all CRUD operations for the telecalling agent.
+"""
+
+import sqlite3
+import json
+import logging
+from datetime import datetime
+from contextlib import contextmanager
+from typing import Optional
+
+from config import DB_PATH
+
+logger = logging.getLogger(__name__)
+
+
+# ── Schema ────────────────────────────────────────────────────────────────────
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS calls (
+    id              INTEGER PRIMARY KEY AUTOINCREMENT,
+    timestamp       TEXT    NOT NULL,           -- ISO-8601 UTC
+    session_id      TEXT    NOT NULL,           -- UUID per Gradio session
+    raw_transcript  TEXT    NOT NULL DEFAULT '',
+    intent          TEXT,                       -- book_meeting | reschedule | cancel | ...
+    caller_name     TEXT,
+    preferred_date  TEXT,                       -- YYYY-MM-DD
+    preferred_time  TEXT,                       -- HH:MM (24h)
+    duration_mins   INTEGER,
+    participants    TEXT    DEFAULT '[]',        -- JSON array of strings
+    meeting_type    TEXT,                       -- phone | video | in_person
+    notes           TEXT,
+    confidence      REAL,                       -- 0.0–1.0 from Qwen
+    decision        TEXT,                       -- schedule | ask_followup | reject
+    reasoning       TEXT,                       -- MiniCPM's explanation
+    status          TEXT    NOT NULL DEFAULT 'open'  -- open | confirmed | cancelled
+);
+
+CREATE TABLE IF NOT EXISTS bookings (
+    id              INTEGER PRIMARY KEY AUTOINCREMENT,
+    call_id         INTEGER NOT NULL REFERENCES calls(id),
+    booked_date     TEXT    NOT NULL,           -- YYYY-MM-DD
+    booked_time     TEXT    NOT NULL,           -- HH:MM
+    duration_mins   INTEGER NOT NULL,
+    caller_name     TEXT    NOT NULL,
+    meeting_type    TEXT,
+    created_at      TEXT    NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_calls_session   ON calls(session_id);
+CREATE INDEX IF NOT EXISTS idx_bookings_date   ON bookings(booked_date);
+"""
+
+
+# ── Connection helper ─────────────────────────────────────────────────────────
+
+@contextmanager
+def get_conn():
+    """Context manager — always commits or rolls back cleanly."""
+    conn = sqlite3.connect(DB_PATH, check_same_thread=False)
+    conn.row_factory = sqlite3.Row
+    conn.execute("PRAGMA journal_mode=WAL")   # safe for concurrent Gradio threads
+    try:
+        yield conn
+        conn.commit()
+    except Exception:
+        conn.rollback()
+        raise
+    finally:
+        conn.close()
+
+
+def init_db():
+    """Create tables if they don't exist. Safe to call on every startup."""
+    with get_conn() as conn:
+        conn.executescript(SCHEMA)
+    logger.info(f"Database ready at {DB_PATH}")
+
+
+# ── Call record helpers ───────────────────────────────────────────────────────
+
+def create_call(session_id: str) -> int:
+    """
+    Insert a bare call record at the start of a session.
+    Returns the new call id so downstream steps can update it.
+    """
+    with get_conn() as conn:
+        cur = conn.execute(
+            "INSERT INTO calls (timestamp, session_id, raw_transcript) VALUES (?, ?, '')",
+            (datetime.utcnow().isoformat(), session_id)
+        )
+        return cur.lastrowid
+
+
+def append_transcript(call_id: int, new_text: str):
+    """Append a transcribed utterance to the running transcript."""
+    with get_conn() as conn:
+        conn.execute(
+            """UPDATE calls
+               SET raw_transcript = raw_transcript || '\n' || ?
+               WHERE id = ?""",
+            (new_text.strip(), call_id)
+        )
+
+
+def update_call_intent(call_id: int, parsed: dict):
+    """
+    Write Qwen's structured JSON output into the call record.
+    `parsed` is expected to match the scheduling JSON schema from config.
+    """
+    with get_conn() as conn:
+        conn.execute(
+            """UPDATE calls SET
+                intent         = :intent,
+                caller_name    = :caller_name,
+                preferred_date = :preferred_date,
+                preferred_time = :preferred_time,
+                duration_mins  = :duration_minutes,
+                participants   = :participants,
+                meeting_type   = :meeting_type,
+                notes          = :notes,
+                confidence     = :confidence
+               WHERE id = :id""",
+            {
+                "intent":           parsed.get("intent"),
+                "caller_name":      parsed.get("caller_name"),
+                "preferred_date":   parsed.get("preferred_date"),
+                "preferred_time":   parsed.get("preferred_time"),
+                "duration_minutes": parsed.get("duration_minutes"),
+                "participants":     json.dumps(parsed.get("participants", [])),
+                "meeting_type":     parsed.get("meeting_type"),
+                "notes":            parsed.get("notes"),
+                "confidence":       parsed.get("confidence"),
+                "id":               call_id,
+            }
+        )
+
+
+def update_call_decision(call_id: int, decision: str, reasoning: str):
+    """Write MiniCPM's evaluation result back to the call record."""
+    with get_conn() as conn:
+        conn.execute(
+            "UPDATE calls SET decision = ?, reasoning = ? WHERE id = ?",
+            (decision, reasoning, call_id)
+        )
+
+
+def confirm_booking(call_id: int, parsed: dict) -> int:
+    """
+    Insert a confirmed booking row and mark the call as confirmed.
+    Returns the booking id.
+    """
+    with get_conn() as conn:
+        cur = conn.execute(
+            """INSERT INTO bookings
+               (call_id, booked_date, booked_time, duration_mins,
+                caller_name, meeting_type, created_at)
+               VALUES (?, ?, ?, ?, ?, ?, ?)""",
+            (
+                call_id,
+                parsed["preferred_date"],
+                parsed["preferred_time"],
+                parsed.get("duration_minutes", 30),
+                parsed.get("caller_name", "Unknown"),
+                parsed.get("meeting_type", "phone"),
+                datetime.utcnow().isoformat(),
+            )
+        )
+        conn.execute(
+            "UPDATE calls SET status = 'confirmed' WHERE id = ?", (call_id,)
+        )
+        return cur.lastrowid
+
+
+# ── Availability check ────────────────────────────────────────────────────────
+
+def is_slot_available(date: str, time: str, duration_mins: int = 30) -> bool:
+    """
+    Returns True if no existing booking overlaps with the requested slot
+    (including the mandatory 15-minute buffer).
+    date: YYYY-MM-DD, time: HH:MM
+    """
+    from datetime import datetime, timedelta
+
+    try:
+        start = datetime.strptime(f"{date} {time}", "%Y-%m-%d %H:%M")
+        end   = start + timedelta(minutes=duration_mins + 15)   # +15 min buffer
+    except ValueError:
+        return False
+
+    with get_conn() as conn:
+        rows = conn.execute(
+            "SELECT booked_time, duration_mins FROM bookings WHERE booked_date = ?",
+            (date,)
+        ).fetchall()
+
+    for row in rows:
+        existing_start = datetime.strptime(f"{date} {row['booked_time']}", "%Y-%m-%d %H:%M")
+        existing_end   = existing_start + timedelta(minutes=row["duration_mins"] + 15)
+        # overlap check
+        if start < existing_end and end > existing_start:
+            return False
+
+    return True
+
+
+def get_booked_slots(date: str) -> list[dict]:
+    """Return all bookings for a given date for display in the UI."""
+    with get_conn() as conn:
+        rows = conn.execute(
+            """SELECT booked_time, duration_mins, caller_name, meeting_type
+               FROM bookings WHERE booked_date = ? ORDER BY booked_time""",
+            (date,)
+        ).fetchall()
+    return [dict(r) for r in rows]
+
+
+def get_recent_calls(limit: int = 20) -> list[dict]:
+    """Fetch the most recent call records for the call log panel."""
+    with get_conn() as conn:
+        rows = conn.execute(
+            """SELECT id, timestamp, caller_name, intent, decision, status
+               FROM calls ORDER BY id DESC LIMIT ?""",
+            (limit,)
+        ).fetchall()
+    return [dict(r) for r in rows]

pipeline/evaluater.py

@@ -0,0 +1,456 @@
+"""
+pipeline/evaluator.py
+
+MiniCPM3-4B scheduling evaluator.
+
+Responsibilities
+────────────────
+Takes a validated SchedulingIntent from the intent parser and decides:
+  - "schedule"      → all fields present, slot is free → confirm booking
+  - "ask_followup"  → fields missing OR slot taken → ask caller a question
+  - "reject"        → invalid request (weekend, out of hours, cancel confirmed)
+  - "end_call"      → caller said goodbye
+
+Also generates a natural spoken response the agent will display
+(and optionally TTS in future) back to the caller.
+
+Memory strategy for RTX 2050
+─────────────────────────────
+MiniCPM3-4B runs AFTER the transcriber finishes each utterance.
+They never overlap. MiniCPM3 loads in 4-bit (INT4) via bitsandbytes
+on CPU — no VRAM needed — freeing the full 4 GB for Cohere Transcribe.
+
+If bitsandbytes is unavailable (e.g. first-time setup) we fall back to
+plain float32 on CPU. Slower but always works.
+
+Why MiniCPM3-4B for this role
+───────────────────────────────
+  - Optimised for deep reasoning with structured JSON output
+  - 4B params → ~2.5 GB RAM in INT4; ~8 GB in float32
+  - Strong at rule-following (our scheduling rules inject cleanly)
+  - Fast enough on CPU for the non-latency-critical decision step
+"""
+
+import json
+import logging
+import threading
+import time
+from dataclasses import dataclass
+from typing import Optional
+
+import torch
+
+from config import (
+    MINICPM_MODEL_ID,
+    MINICPM_DEVICE,
+    MINICPM_MAX_TOKENS,
+    SCHEDULING_RULES,
+)
+from pipeline.intent_parser import SchedulingIntent
+
+logger = logging.getLogger(__name__)
+
+
+# ── Output type ───────────────────────────────────────────────────────────────
+
+@dataclass
+class EvaluationResult:
+    decision:         str            # schedule | ask_followup | reject | end_call
+    spoken_response:  str            # what the agent says back to the caller
+    reasoning:        str            # internal chain-of-thought (for DB + debug)
+    suggested_date:   Optional[str]  # YYYY-MM-DD if rescheduling to next free slot
+    suggested_time:   Optional[str]  # HH:MM if rescheduling
+
+
+# ── Prompts ───────────────────────────────────────────────────────────────────
+
+_SYSTEM_PROMPT = """You are an AI scheduling assistant on a live phone call.
+
+SCHEDULING RULES:
+{rules}
+
+Your job: given the caller's current intent data and the conversation so far,
+decide what to do next and generate a SHORT, natural spoken response.
+
+Respond ONLY with a JSON object — no prose outside the JSON:
+{{
+  "decision": "schedule" | "ask_followup" | "reject" | "end_call",
+  "spoken_response": "what you say to the caller (1-2 sentences, conversational)",
+  "reasoning": "brief internal note explaining your decision",
+  "suggested_date": "YYYY-MM-DD or null",
+  "suggested_time": "HH:MM or null"
+}}
+
+Guidelines:
+- spoken_response must sound natural on a phone call. No bullet points.
+- If asking a follow-up, ask for ONE missing field at a time.
+- If the slot is taken, suggest the next available slot from slot_conflicts.
+- If intent is "end_call" or caller says goodbye, set decision to "end_call".
+- Keep spoken_response under 30 words.
+"""
+
+_USER_PROMPT = """Intent data:
+{intent_json}
+
+Slot available: {slot_available}
+Slot conflicts (booked slots on that date): {slot_conflicts}
+Conversation history (last 6 utterances):
+{history}
+
+JSON decision:"""
+
+
+# ── Evaluator ─────────────────────────────────────────────────────────────────
+
+class Evaluator:
+    """
+    Lazy-loading MiniCPM3-4B evaluator.  CPU-only, INT4 quantised.
+    """
+
+    def __init__(self):
+        self._model     = None
+        self._tokenizer = None
+        self._lock      = threading.Lock()
+        self._loaded    = False
+
+    # ── Public ────────────────────────────────────────────────────────────────
+
+    def evaluate(
+        self,
+        intent:        SchedulingIntent,
+        utterances:    list[str],
+        slot_available: bool             = True,
+        slot_conflicts: list[dict]       = None,
+    ) -> EvaluationResult:
+        """
+        Evaluate a scheduling intent and generate a spoken response.
+
+        Parameters
+        ----------
+        intent         : validated SchedulingIntent from intent_parser
+        utterances     : full list of transcript strings so far
+        slot_available : result of db.is_slot_available()
+        slot_conflicts : result of db.get_booked_slots() for that date
+
+        Returns
+        -------
+        EvaluationResult — never raises, falls back gracefully on error
+        """
+        # Fast-path: handle terminal intents without model inference
+        fast = self._fast_path(intent)
+        if fast:
+            return fast
+
+        self._ensure_loaded()
+
+        prompt = self._build_prompt(intent, utterances, slot_available, slot_conflicts or [])
+
+        try:
+            t0  = time.perf_counter()
+            raw = self._generate(prompt)
+            elapsed = time.perf_counter() - t0
+            logger.info(f"MiniCPM3 inference in {elapsed:.2f}s — raw: {raw[:120]}")
+            return self._parse_response(raw, intent)
+
+        except Exception as exc:
+            logger.error(f"Evaluator.evaluate failed: {exc}", exc_info=True)
+            return self._fallback_result(intent)
+
+    def unload(self):
+        with self._lock:
+            if self._loaded:
+                del self._model, self._tokenizer
+                self._model     = None
+                self._tokenizer = None
+                self._loaded    = False
+                logger.info("Evaluator unloaded.")
+
+    @property
+    def is_loaded(self) -> bool:
+        return self._loaded
+
+    # ── Internal ──────────────────────────────────────────────────────────────
+
+    def _fast_path(self, intent: SchedulingIntent) -> Optional[EvaluationResult]:
+        """
+        Handle clear-cut cases without burning inference time.
+        """
+        if intent.intent == "end_call":
+            return EvaluationResult(
+                decision        = "end_call",
+                spoken_response = "Thank you for calling. Goodbye!",
+                reasoning       = "Caller indicated end of call.",
+                suggested_date  = None,
+                suggested_time  = None,
+            )
+
+        if intent.intent == "unclear" or intent.confidence < 0.15:
+            return EvaluationResult(
+                decision        = "ask_followup",
+                spoken_response = (
+                    "I'm sorry, I didn't quite catch that. "
+                    "Could you tell me what you'd like to schedule?"
+                ),
+                reasoning       = f"Intent unclear or confidence too low ({intent.confidence:.2f}).",
+                suggested_date  = None,
+                suggested_time  = None,
+            )
+
+        return None   # fall through to model inference
+
+    def _ensure_loaded(self):
+        if self._loaded:
+            return
+        with self._lock:
+            if self._loaded:
+                return
+            self._load()
+
+    def _load(self):
+        from transformers import AutoTokenizer, AutoModelForCausalLM
+
+        logger.info(f"Loading MiniCPM3-4B ({MINICPM_MODEL_ID}) on CPU…")
+        t0 = time.perf_counter()
+
+        self._tokenizer = AutoTokenizer.from_pretrained(
+            MINICPM_MODEL_ID,
+            trust_remote_code=True,
+        )
+
+        # Try INT4 via bitsandbytes first; fall back to float32
+        try:
+            from transformers import BitsAndBytesConfig
+            bnb_config = BitsAndBytesConfig(
+                load_in_4bit              = True,
+                bnb_4bit_compute_dtype    = torch.float32,
+                bnb_4bit_use_double_quant = True,
+                bnb_4bit_quant_type       = "nf4",
+            )
+            self._model = AutoModelForCausalLM.from_pretrained(
+                MINICPM_MODEL_ID,
+                quantization_config = bnb_config,
+                trust_remote_code   = True,
+                device_map          = "cpu",
+                low_cpu_mem_usage   = True,
+            )
+            logger.info("MiniCPM3 loaded in INT4 (bitsandbytes NF4)")
+
+        except (ImportError, Exception) as e:
+            logger.warning(f"bitsandbytes INT4 failed ({e}) — loading float32 on CPU")
+            self._model = AutoModelForCausalLM.from_pretrained(
+                MINICPM_MODEL_ID,
+                torch_dtype       = torch.float32,
+                trust_remote_code = True,
+                low_cpu_mem_usage = True,
+            )
+            self._model.to("cpu")
+
+        self._model.eval()
+
+        elapsed = time.perf_counter() - t0
+        logger.info(f"MiniCPM3 ready in {elapsed:.1f}s")
+        self._loaded = True
+
+    def _build_prompt(
+        self,
+        intent:         SchedulingIntent,
+        utterances:     list[str],
+        slot_available: bool,
+        slot_conflicts: list[dict],
+    ) -> str:
+        """
+        MiniCPM3 uses ChatML format identical to Qwen:
+          <|im_start|>system … <|im_end|>
+          <|im_start|>user   … <|im_end|>
+          <|im_start|>assistant
+        """
+        # Serialise intent — exclude heavy fields for prompt brevity
+        intent_dict = intent.model_dump(
+            exclude={"participants", "notes", "missing_fields"}
+        )
+        intent_dict["missing_fields"] = intent.missing_fields  # keep for context
+
+        history = "\n".join(
+            f"  [{i+1}] {u}" for i, u in enumerate(utterances[-6:])
+        ) or "  (none yet)"
+
+        conflicts_str = (
+            json.dumps(slot_conflicts, indent=2)
+            if slot_conflicts
+            else "none"
+        )
+
+        system = _SYSTEM_PROMPT.format(rules=SCHEDULING_RULES.strip())
+        user   = _USER_PROMPT.format(
+            intent_json    = json.dumps(intent_dict, indent=2),
+            slot_available = str(slot_available),
+            slot_conflicts = conflicts_str,
+            history        = history,
+        )
+
+        return (
+            f"<|im_start|>system\n{system}<|im_end|>\n"
+            f"<|im_start|>user\n{user}<|im_end|>\n"
+            f"<|im_start|>assistant\n"
+        )
+
+    def _generate(self, prompt: str) -> str:
+        inputs = self._tokenizer(
+            prompt,
+            return_tensors    = "pt",
+            truncation        = True,
+            max_length        = 3072,
+        ).to("cpu")
+
+        with torch.no_grad():
+            output_ids = self._model.generate(
+                **inputs,
+                max_new_tokens  = MINICPM_MAX_TOKENS,
+                temperature     = 0.2,
+                do_sample       = True,
+                top_p           = 0.9,
+                repetition_penalty = 1.1,
+                pad_token_id    = self._tokenizer.eos_token_id,
+            )
+
+        # Decode only the new tokens (skip the prompt)
+        new_ids = output_ids[0][inputs["input_ids"].shape[1]:]
+        return self._tokenizer.decode(new_ids, skip_special_tokens=True).strip()
+
+    def _parse_response(
+        self, raw: str, intent: SchedulingIntent
+    ) -> EvaluationResult:
+        raw = raw.strip().lstrip("```json").lstrip("```").rstrip("```").strip()
+
+        # Extract JSON block if model added preamble despite instructions
+        if "{" in raw:
+            raw = raw[raw.index("{") : raw.rindex("}") + 1]
+
+        try:
+            data = json.loads(raw)
+            decision = data.get("decision", "ask_followup")
+
+            # Sanity-check decision against intent state
+            if decision == "schedule" and intent.missing_fields:
+                logger.warning(
+                    "Model said 'schedule' but fields are missing "
+                    f"({intent.missing_fields}) — overriding to ask_followup"
+                )
+                decision = "ask_followup"
+
+            return EvaluationResult(
+                decision        = decision,
+                spoken_response = data.get("spoken_response", self._default_response(intent)),
+                reasoning       = data.get("reasoning", ""),
+                suggested_date  = data.get("suggested_date"),
+                suggested_time  = data.get("suggested_time"),
+            )
+
+        except Exception as exc:
+            logger.warning(f"Evaluator JSON parse failed: {exc} — raw: {raw[:200]}")
+            return self._fallback_result(intent)
+
+    @staticmethod
+    def _default_response(intent: SchedulingIntent) -> str:
+        if intent.missing_fields:
+            field_map = {
+                "caller_name":    "Could I get your name please?",
+                "preferred_date": "What date would you like to schedule this for?",
+                "preferred_time": "And what time works best for you?",
+            }
+            return field_map.get(
+                intent.missing_fields[0],
+                "Could you give me a bit more detail?"
+            )
+        return "Let me check availability for you."
+
+    @staticmethod
+    def _fallback_result(intent: SchedulingIntent) -> EvaluationResult:
+        return EvaluationResult(
+            decision        = "ask_followup",
+            spoken_response = "I'm sorry, could you repeat that?",
+            reasoning       = "Evaluator fallback — parse error.",
+            suggested_date  = None,
+            suggested_time  = None,
+        )
+
+
+# ── Module singleton ──────────────────────────────────────────────────────────
+
+_evaluator: Optional[Evaluator] = None
+
+
+def get_evaluator() -> Evaluator:
+    global _evaluator
+    if _evaluator is None:
+        _evaluator = Evaluator()
+    return _evaluator
+
+
+# ── Offline smoke test ────────────────────────────────────────────────────────
+
+def _smoke_test_offline():
+    logging.basicConfig(level=logging.INFO)
+    logger.info("Running Evaluator offline smoke test…")
+
+    from pipeline.intent_parser import SchedulingIntent
+
+    # 1. end_call fast-path
+    intent_end = SchedulingIntent(intent="end_call", confidence=0.9)
+    ev = Evaluator()
+    result = ev._fast_path(intent_end)
+    assert result is not None
+    assert result.decision == "end_call"
+    logger.info("  ✓ end_call fast-path")
+
+    # 2. unclear fast-path
+    intent_unclear = SchedulingIntent(intent="unclear", confidence=0.05)
+    result = ev._fast_path(intent_unclear)
+    assert result is not None and result.decision == "ask_followup"
+    logger.info("  ✓ unclear fast-path")
+
+    # 3. Normal intent — no fast-path
+    intent_book = SchedulingIntent(
+        intent="book_meeting", caller_name="Priya",
+        preferred_date="2026-06-10", preferred_time="14:00",
+        confidence=0.9,
+    ).compute_missing()
+    result = ev._fast_path(intent_book)
+    assert result is None   # should fall through to model
+    logger.info("  ✓ valid booking intent passes through to model")
+
+    # 4. Safety override — model says 'schedule' but fields missing
+    intent_partial = SchedulingIntent(
+        intent="book_meeting", caller_name="Raj", confidence=0.5
+    ).compute_missing()
+    raw_json = json.dumps({
+        "decision": "schedule",   # model hallucinated this
+        "spoken_response": "Booked!",
+        "reasoning": "test",
+        "suggested_date": None,
+        "suggested_time": None,
+    })
+    result = ev._parse_response(raw_json, intent_partial)
+    assert result.decision == "ask_followup", (
+        f"Safety override failed — got {result.decision}"
+    )
+    logger.info("  ✓ safety override: 'schedule' with missing fields → ask_followup")
+
+    # 5. Prompt build sanity
+    prompt = ev._build_prompt(intent_book, ["Hi I want to book a call"], True, [])
+    assert "<|im_start|>system" in prompt
+    assert "SCHEDULING RULES" in prompt
+    assert "slot_available" in prompt or "Slot available" in prompt
+    logger.info("  ✓ prompt structure correct")
+
+    # 6. Singleton
+    e1 = get_evaluator()
+    e2 = get_evaluator()
+    assert e1 is e2
+    logger.info("  ✓ module singleton")
+
+    logger.info("\nOffline smoke test PASSED ✓")
+
+
+if __name__ == "__main__":
+    _smoke_test_offline()

pipeline/intent_parser.py

@@ -0,0 +1,461 @@
+"""
+pipeline/intent_parser.py
+
+Qwen2.5-7B-Instruct (Q4_K_M GGUF) intent & entity extractor.
+
+Responsibilities
+────────────────
+Takes a raw transcript string from Cohere Transcribe and returns a
+validated SchedulingIntent object — structured data the evaluator and
+DB layer can act on directly.
+
+Why GGUF + llama-cpp-python
+────────────────────────────
+  - Qwen2.5-7B-Instruct in Q4_K_M needs ~4.5 GB total:
+      20 layers on RTX 2050 GPU  → ~0.8 GB VRAM
+      remaining ~15 layers on CPU RAM → ~3.7 GB RAM
+  - llama-cpp-python's grammar feature forces output to be valid JSON
+    with no post-processing hacks — zero hallucinated keys.
+
+GBNF Grammar
+────────────
+llama.cpp supports GBNF (Generative BNF) grammars that constrain the
+token sampling so the model physically cannot produce invalid JSON.
+We define a tight grammar matching exactly our SchedulingIntent schema.
+
+Download the GGUF before running (one-time, ~4.5 GB):
+  huggingface-cli download Qwen/Qwen2.5-7B-Instruct-GGUF \\
+      --include "qwen2.5-7b-instruct-q4_k_m.gguf" \\
+      --local-dir ./models
+"""
+
+import json
+import logging
+import threading
+import time
+from typing import Optional
+
+from pydantic import BaseModel, Field, field_validator
+
+from config import (
+    QWEN_GGUF_PATH,
+    QWEN_N_GPU_LAYERS,
+    QWEN_N_CTX,
+    QWEN_MAX_TOKENS,
+    QWEN_TEMPERATURE,
+)
+
+logger = logging.getLogger(__name__)
+
+
+# ── Output schema ─────────────────────────────────────────────────────────────
+
+class SchedulingIntent(BaseModel):
+    """
+    Structured output from Qwen2.5.  Every field is Optional so partial
+    information (e.g. caller gave name but no date yet) is represented
+    cleanly rather than hallucinated.
+    """
+    intent: str = Field(
+        description=(
+            "One of: book_meeting | reschedule | cancel | "
+            "check_availability | provide_info | end_call | unclear"
+        )
+    )
+    caller_name:      Optional[str] = Field(None, description="Full name of the caller")
+    preferred_date:   Optional[str] = Field(None, description="YYYY-MM-DD or null")
+    preferred_time:   Optional[str] = Field(None, description="HH:MM 24-hour or null")
+    duration_minutes: Optional[int] = Field(None, description="Meeting length in minutes")
+    participants:     list[str]     = Field(default_factory=list)
+    meeting_type:     Optional[str] = Field(None, description="phone | video | in_person")
+    notes:            Optional[str] = Field(None, description="Any extra context")
+    confidence:       float         = Field(0.0, description="0.0–1.0 extraction confidence")
+    missing_fields:   list[str]     = Field(
+        default_factory=list,
+        description="Fields still needed to complete booking"
+    )
+
+    @field_validator("intent")
+    @classmethod
+    def validate_intent(cls, v: str) -> str:
+        valid = {
+            "book_meeting", "reschedule", "cancel",
+            "check_availability", "provide_info", "end_call", "unclear"
+        }
+        return v if v in valid else "unclear"
+
+    @field_validator("preferred_date")
+    @classmethod
+    def validate_date(cls, v: Optional[str]) -> Optional[str]:
+        if v is None:
+            return None
+        import re
+        if re.match(r"^\d{4}-\d{2}-\d{2}$", v):
+            return v
+        return None   # reject malformed dates
+
+    @field_validator("preferred_time")
+    @classmethod
+    def validate_time(cls, v: Optional[str]) -> Optional[str]:
+        if v is None:
+            return None
+        import re
+        if re.match(r"^\d{2}:\d{2}$", v):
+            return v
+        return None
+
+    @field_validator("confidence")
+    @classmethod
+    def clamp_confidence(cls, v: float) -> float:
+        return max(0.0, min(1.0, float(v)))
+
+    def compute_missing(self) -> "SchedulingIntent":
+        """
+        Populate missing_fields based on what a booking actually needs.
+        Returns self for chaining.
+        """
+        needed = []
+        if self.intent == "book_meeting":
+            if not self.caller_name:    needed.append("caller_name")
+            if not self.preferred_date: needed.append("preferred_date")
+            if not self.preferred_time: needed.append("preferred_time")
+        self.missing_fields = needed
+        return self
+
+
+# ── GBNF grammar ──────────────────────────────────────────────────────────────
+# Constrains Qwen to produce ONLY valid JSON matching our schema.
+# llama.cpp samples tokens that satisfy this grammar at every step.
+
+_SCHEDULING_GRAMMAR = r"""
+root   ::= object
+value  ::= object | array | string | number | bool | null
+
+object ::=
+  "{" ws
+    "\"intent\""         ws ":" ws intent-val        ws ","  ws
+    "\"caller_name\""    ws ":" ws str-or-null        ws ","  ws
+    "\"preferred_date\"" ws ":" ws str-or-null        ws ","  ws
+    "\"preferred_time\"" ws ":" ws str-or-null        ws ","  ws
+    "\"duration_minutes\"" ws ":" ws int-or-null      ws ","  ws
+    "\"participants\""   ws ":" ws str-array          ws ","  ws
+    "\"meeting_type\""   ws ":" ws meeting-type-val   ws ","  ws
+    "\"notes\""          ws ":" ws str-or-null        ws ","  ws
+    "\"confidence\""     ws ":" ws confidence-val     ws ","  ws
+    "\"missing_fields\"" ws ":" ws str-array
+  ws "}"
+
+intent-val ::= "\"book_meeting\"" | "\"reschedule\"" | "\"cancel\"" |
+               "\"check_availability\"" | "\"provide_info\"" |
+               "\"end_call\"" | "\"unclear\""
+
+meeting-type-val ::= "\"phone\"" | "\"video\"" | "\"in_person\"" | "null"
+
+str-or-null ::= string | "null"
+int-or-null ::= integer | "null"
+
+confidence-val ::= "0" | "1" | ("0" "." [0-9]+) | ("1" "." "0"*)
+
+str-array ::= "[" ws "]" | "[" ws string (ws "," ws string)* ws "]"
+
+string ::= "\"" ([^"\\] | "\\" .)* "\""
+integer ::= "-"? [0-9]+
+number  ::= integer ("." [0-9]+)?
+bool    ::= "true" | "false"
+null    ::= "null"
+array   ::= "[" ws (value (ws "," ws value)*)? ws "]"
+ws      ::= [ \t\n\r]*
+"""
+
+
+# ── System prompt ──────────────────────────────────────────────────────────────
+
+_SYSTEM_PROMPT = """You are a scheduling assistant that extracts structured information from call transcripts.
+
+Extract ONLY what the caller explicitly said. Never invent information.
+Return a single JSON object with these exact fields:
+- intent: what the caller wants (book_meeting/reschedule/cancel/check_availability/provide_info/end_call/unclear)
+- caller_name: their name if given, else null
+- preferred_date: YYYY-MM-DD if mentioned, else null. Resolve relative dates (e.g. "tomorrow", "next Monday") to absolute dates based on today being {today}.
+- preferred_time: HH:MM 24-hour if mentioned, else null. Convert "3pm" → "15:00", "9:30 in the morning" → "09:30".
+- duration_minutes: integer if mentioned, else null. Default 30 if they say "quick meeting".
+- participants: list of names mentioned besides the caller
+- meeting_type: "phone", "video", "in_person", or null
+- notes: any other relevant context, else null
+- confidence: 0.0–1.0 reflecting how complete the information is
+- missing_fields: list of field names still needed for a complete booking
+"""
+
+_USER_PROMPT_TEMPLATE = """Transcript:
+\"\"\"{transcript}\"\"\"
+
+JSON:"""
+
+
+# ── Parser class ──────────────────────────────────────────────────────────────
+
+class IntentParser:
+    """
+    Lazy-loading Qwen2.5-7B-Instruct GGUF intent extractor.
+    Thread-safe singleton pattern — one model instance for the whole app.
+    """
+
+    def __init__(self):
+        self._llm     = None
+        self._grammar = None
+        self._lock    = threading.Lock()
+        self._loaded  = False
+
+    # ── Public ────────────────────────────────────────────────────────────────
+
+    def parse(self, transcript: str) -> SchedulingIntent:
+        """
+        Extract scheduling intent from a transcript string.
+
+        Parameters
+        ----------
+        transcript : str
+            Raw text from Cohere Transcribe (one or more utterances joined).
+
+        Returns
+        -------
+        SchedulingIntent
+            Validated Pydantic model.  Falls back to intent="unclear" on
+            any parse failure so the pipeline never crashes.
+        """
+        if not transcript or not transcript.strip():
+            return self._fallback("Empty transcript")
+
+        self._ensure_loaded()
+
+        prompt = self._build_prompt(transcript)
+
+        try:
+            t0 = time.perf_counter()
+
+            response = self._llm(
+                prompt,
+                max_tokens  = QWEN_MAX_TOKENS,
+                temperature = QWEN_TEMPERATURE,
+                stop        = ["\n}\n", "```"],   # belt-and-suspenders stop
+                grammar     = self._grammar,
+                echo        = False,
+            )
+
+            elapsed  = time.perf_counter() - t0
+            raw_text = response["choices"][0]["text"].strip()
+
+            logger.info(f"Qwen inference in {elapsed:.2f}s — raw: {raw_text[:120]}…")
+
+            return self._parse_response(raw_text, transcript)
+
+        except Exception as exc:
+            logger.error(f"IntentParser.parse failed: {exc}", exc_info=True)
+            return self._fallback(str(exc))
+
+    def parse_accumulated(self, utterances: list[str]) -> SchedulingIntent:
+        """
+        Parse the full conversation so far (list of utterance strings).
+        Joins them with newlines and runs a single inference pass.
+        Use this after each new utterance to get an updated intent state.
+        """
+        full_transcript = "\n".join(u for u in utterances if u.strip())
+        return self.parse(full_transcript)
+
+    def unload(self):
+        """Free memory — model reloads lazily on next call."""
+        with self._lock:
+            if self._loaded:
+                del self._llm
+                self._llm    = None
+                self._loaded = False
+                logger.info("IntentParser unloaded.")
+
+    @property
+    def is_loaded(self) -> bool:
+        return self._loaded
+
+    # ── Internal ──────────────────────────────────────────────────────────────
+
+    def _ensure_loaded(self):
+        if self._loaded:
+            return
+        with self._lock:
+            if self._loaded:
+                return
+            self._load()
+
+    def _load(self):
+        try:
+            from llama_cpp import Llama, LlamaGrammar
+        except ImportError:
+            raise RuntimeError(
+                "llama-cpp-python is not installed.\n"
+                "Install with CUDA support:\n"
+                "  Windows: set CMAKE_ARGS=-DGGML_CUDA=on && "
+                "pip install llama-cpp-python --no-cache-dir\n"
+                "  Linux:   CMAKE_ARGS='-DGGML_CUDA=on' "
+                "pip install llama-cpp-python --no-cache-dir"
+            )
+
+        if not QWEN_GGUF_PATH.exists():
+            raise FileNotFoundError(
+                f"GGUF not found at {QWEN_GGUF_PATH}\n"
+                "Download with:\n"
+                "  huggingface-cli download Qwen/Qwen2.5-7B-Instruct-GGUF \\\n"
+                "      --include 'qwen2.5-7b-instruct-q4_k_m.gguf' \\\n"
+                "      --local-dir ./models"
+            )
+
+        logger.info(
+            f"Loading Qwen2.5-7B Q4_K_M — "
+            f"{QWEN_N_GPU_LAYERS} layers on GPU, rest on CPU…"
+        )
+        t0 = time.perf_counter()
+
+        self._llm = Llama(
+            model_path    = str(QWEN_GGUF_PATH),
+            n_gpu_layers  = QWEN_N_GPU_LAYERS,   # 20 → ~0.8 GB VRAM
+            n_ctx         = QWEN_N_CTX,           # 4096 tokens
+            n_threads     = 6,                    # leave 2 cores for Gradio
+            n_batch       = 512,
+            verbose       = False,
+        )
+
+        self._grammar = LlamaGrammar.from_string(_SCHEDULING_GRAMMAR)
+
+        elapsed = time.perf_counter() - t0
+        logger.info(f"Qwen2.5 ready in {elapsed:.1f}s")
+        self._loaded = True
+
+    def _build_prompt(self, transcript: str) -> str:
+        """
+        Qwen2.5-Instruct uses ChatML format:
+          <|im_start|>system\n…<|im_end|>\n
+          <|im_start|>user\n…<|im_end|>\n
+          <|im_start|>assistant\n
+        The grammar then forces the assistant turn to be valid JSON.
+        """
+        from datetime import date
+        today = date.today().strftime("%Y-%m-%d")
+
+        system = _SYSTEM_PROMPT.format(today=today)
+        user   = _USER_PROMPT_TEMPLATE.format(transcript=transcript.strip())
+
+        return (
+            f"<|im_start|>system\n{system}<|im_end|>\n"
+            f"<|im_start|>user\n{user}<|im_end|>\n"
+            f"<|im_start|>assistant\n"
+        )
+
+    def _parse_response(self, raw: str, original_transcript: str) -> SchedulingIntent:
+        """
+        Parse and validate Qwen's JSON output.
+        The grammar guarantees structural validity; Pydantic validates values.
+        """
+        # Strip any accidental markdown fences
+        raw = raw.strip().lstrip("```json").lstrip("```").rstrip("```").strip()
+
+        try:
+            data   = json.loads(raw)
+            intent = SchedulingIntent(**data).compute_missing()
+            logger.info(
+                f"Parsed intent={intent.intent} "
+                f"name={intent.caller_name} "
+                f"date={intent.preferred_date} "
+                f"time={intent.preferred_time} "
+                f"confidence={intent.confidence:.2f} "
+                f"missing={intent.missing_fields}"
+            )
+            return intent
+
+        except Exception as exc:
+            logger.warning(f"JSON parse/validation failed: {exc} — raw was: {raw[:200]}")
+            return self._fallback(str(exc))
+
+    @staticmethod
+    def _fallback(reason: str) -> SchedulingIntent:
+        logger.warning(f"Returning fallback intent — reason: {reason}")
+        return SchedulingIntent(
+            intent     = "unclear",
+            confidence = 0.0,
+        ).compute_missing()
+
+
+# ── Module singleton ──────────────────────────────────────────────────────────
+
+_parser: Optional[IntentParser] = None
+
+
+def get_intent_parser() -> IntentParser:
+    global _parser
+    if _parser is None:
+        _parser = IntentParser()
+    return _parser
+
+
+# ── Offline smoke test ────────────────────────────────────────────────────────
+
+def _smoke_test_offline():
+    """Tests schema, validators, and prompt building without loading the model."""
+    logging.basicConfig(level=logging.INFO)
+    logger.info("Running IntentParser offline smoke test…")
+
+    # 1. Valid full intent
+    intent = SchedulingIntent(
+        intent           = "book_meeting",
+        caller_name      = "Priya Sharma",
+        preferred_date   = "2026-06-10",
+        preferred_time   = "14:00",
+        duration_minutes = 30,
+        participants     = ["Priya Sharma"],
+        meeting_type     = "video",
+        notes            = None,
+        confidence       = 0.95,
+    ).compute_missing()
+    assert intent.missing_fields == [], f"Expected no missing fields, got {intent.missing_fields}"
+    logger.info("  ✓ Full booking intent — no missing fields")
+
+    # 2. Partial intent — date and time missing
+    partial = SchedulingIntent(
+        intent      = "book_meeting",
+        caller_name = "Raj",
+        confidence  = 0.4,
+    ).compute_missing()
+    assert "preferred_date" in partial.missing_fields
+    assert "preferred_time" in partial.missing_fields
+    logger.info(f"  ✓ Partial intent missing fields: {partial.missing_fields}")
+
+    # 3. Invalid intent string → coerced to "unclear"
+    coerced = SchedulingIntent(intent="nonsense", confidence=0.1)
+    assert coerced.intent == "unclear"
+    logger.info("  ✓ Invalid intent string coerced to 'unclear'")
+
+    # 4. Malformed date → None
+    bad_date = SchedulingIntent(intent="book_meeting", preferred_date="June 10th")
+    assert bad_date.preferred_date is None
+    logger.info("  ✓ Malformed date rejected → None")
+
+    # 5. Confidence clamping
+    clamped = SchedulingIntent(intent="unclear", confidence=999.0)
+    assert clamped.confidence == 1.0
+    logger.info("  ✓ Confidence clamped to 1.0")
+
+    # 6. Prompt build
+    parser = IntentParser()
+    prompt = parser._build_prompt("Hi I want to book a meeting tomorrow at 3pm")
+    assert "<|im_start|>system" in prompt
+    assert "2026-" in prompt   # today injected
+    assert "tomorrow" in prompt
+    logger.info("  ✓ Prompt format correct (ChatML)")
+
+    # 7. Fallback
+    fallback = IntentParser._fallback("test")
+    assert fallback.intent == "unclear"
+    logger.info("  ✓ Fallback returns 'unclear' intent")
+
+    logger.info("\nOffline smoke test PASSED ✓")
+
+
+if __name__ == "__main__":
+    _smoke_test_offline()

pipeline/orchestrator.py

@@ -0,0 +1,441 @@
+"""
+pipeline/orchestrator.py
+
+Orchestrator for the telecalling agent pipeline.
+
+Coordinates the end-to-end flow:
+  1. VAD listener        → detects speech boundaries
+  2. Transcriber         → audio-to-text (Cohere)
+  3. Intent parser       → structured intent extraction (Qwen2.5)
+  4. Evaluator           → scheduling decision + spoken response (MiniCPM3)
+  5. Database updates    → persist call state
+
+This module is the glue between Gradio's audio stream and the ML models.
+It maintains call state across multiple audio chunks and orchestrates
+lazy-loaded model lifecycle.
+
+Usage
+─────
+    session = CallSession()
+    session.start_call()
+
+    # Called by Gradio on each audio chunk:
+    update = session.process_audio_chunk(sample_rate, audio_chunk)
+    # use update to refresh UI
+
+    session.end_call()
+"""
+
+import logging
+from dataclasses import dataclass, field
+from typing import Optional
+
+import numpy as np
+
+from config import DB_PATH
+from db import (
+    create_call,
+    append_transcript,
+    update_call_intent,
+    update_call_decision,
+    is_slot_available,
+    get_booked_slots,
+    get_recent_calls,
+)
+from pipeline.transcriber import get_transcriber
+from pipeline.intent_parser import get_intent_parser
+from pipeline.evaluater import get_evaluator
+
+logger = logging.getLogger(__name__)
+
+
+# ── Mock Transcriber — fallback when PyTorch/dependencies missing ──────────────
+
+class MockTranscriber:
+    """Fallback mock transcriber for testing without PyTorch."""
+
+    def __init__(self):
+        self.is_loaded = True
+        self.device = "mock"
+        self._call_count = 0
+
+    def transcribe(self, sample_rate: int, audio: np.ndarray) -> str:
+        """Return placeholder transcriptions for testing."""
+        self._call_count += 1
+        responses = [
+            "I'd like to book a meeting",
+            "Can we schedule for next Monday at 2 PM",
+            "Yes, that works for me",
+            "Please confirm my booking",
+            "Thank you, goodbye",
+        ]
+        return responses[self._call_count % len(responses)]
+
+    def transcribe_batch(self, utterances: list) -> list[str]:
+        return [self.transcribe(sr, audio) for sr, audio in utterances]
+
+    def unload(self):
+        pass
+
+
+def get_mock_transcriber() -> MockTranscriber:
+    """Get mock transcriber instance."""
+    return MockTranscriber()
+
+
+# ── MockVADListener — always available as fallback ────────────────────────────
+
+class MockVADListener:
+    """Fallback mock VAD for testing without onnxruntime or torch/numpy issues."""
+
+    def __init__(self):
+        self.is_speaking = False
+        self._buffer = []
+        self._chunk_count = 0
+
+    def process_chunk(self, sample_rate: int, audio):
+        """Simulate speech detection using simple chunk counting."""
+        if audio is None or len(audio) == 0:
+            return
+
+        try:
+            audio_np = np.asarray(audio, dtype=np.float32)
+        except Exception:
+            return
+
+        has_energy = float(np.max(np.abs(audio_np))) > 0.01
+
+        if has_energy:
+            self.is_speaking = True
+            self._buffer.append(audio_np)
+            self._chunk_count += 1
+
+            if self._chunk_count >= 8:
+                try:
+                    utterance = np.concatenate(self._buffer, axis=0)
+                    self._buffer = []
+                    self._chunk_count = 0
+                    self.is_speaking = False
+                    yield (sample_rate, utterance)
+                except Exception as e:
+                    logger.debug(f"MockVAD concatenate failed: {e}")
+                    self._buffer = []
+                    self._chunk_count = 0
+        else:
+            if self._buffer:
+                try:
+                    utterance = np.concatenate(self._buffer, axis=0)
+                    self._buffer = []
+                    self._chunk_count = 0
+                    self.is_speaking = False
+                    yield (sample_rate, utterance)
+                except Exception as e:
+                    logger.debug(f"MockVAD flush failed: {e}")
+                    self._buffer = []
+                    self._chunk_count = 0
+
+    def reset(self):
+        self._buffer = []
+        self._chunk_count = 0
+        self.is_speaking = False
+
+
+# ── Try to import real VADListener; fallback to mock if needed ─────────────────
+
+try:
+    from pipeline.vad_listener import VADListener
+except (ImportError, ModuleNotFoundError) as e:
+    logger.warning(f"VADListener import failed ({e}) — will use MockVADListener")
+    VADListener = MockVADListener
+
+
+# ── Output types ──────────────────────────────────────────────────────────────
+
+@dataclass
+class PipelineUpdate:
+    """Update to display in the Gradio UI after each processing step."""
+    status: str                              # status badge text
+    vad_speaking: bool                       # True if mic active
+    transcript_lines: list[str]              # accumulated utterances
+    intent_md: str                           # markdown table of intent
+    agent_response: str                      # what the agent said
+    booking_confirmed: Optional[dict]        # booking details if confirmed
+    call_log: list[dict]                     # recent call records
+
+
+@dataclass
+class _OrchestrationResponse:
+    """Internal response from pipeline processing."""
+    spoken_text: str                         # text to say back to caller
+    call_id: int                             # database row id
+    is_terminal: bool                        # True if call should end
+    interim_intent: Optional[str] = None     # intent extracted so far
+    booking_info: Optional[dict] = None      # booking details if confirmed
+
+
+# ── Session orchestrator ──────────────────────────────────────────────────────
+
+class CallSession:
+    """
+    Stateful session for one telecalling interaction in Gradio.
+
+    Lifecycle
+    ─────────
+    1. __init__() — create empty, not yet active
+    2. start_call() → activates VAD listener, creates database record
+    3. process_audio_chunk() — repeatedly as Gradio streams audio
+    4. end_call() — cleanup and reset for next call
+    5. reset() — full session reset
+
+    All methods return PipelineUpdate to refresh the UI.
+    """
+
+    def __init__(self):
+        """Initialize an inactive session."""
+        self._call_active = False
+        self._call_id = None
+        self._vad = None
+
+        # Try to load real transcriber; fall back to mock on import error
+        try:
+            self._transcriber = get_transcriber()
+        except (ImportError, ModuleNotFoundError) as e:
+            logger.warning(f"Transcriber import failed ({e}) — using MockTranscriber")
+            self._transcriber = get_mock_transcriber()
+
+        try:
+            self._parser = get_intent_parser()
+        except (ImportError, ModuleNotFoundError) as e:
+            logger.warning(f"Parser import failed ({e})")
+            self._parser = None
+
+        try:
+            self._evaluator = get_evaluator()
+        except (ImportError, ModuleNotFoundError) as e:
+            logger.warning(f"Evaluator import failed ({e})")
+            self._evaluator = None
+
+        self._utterances = []
+        self._current_intent = None
+        self._last_response = ""
+        self._booking_info = None
+
+        logger.info("CallSession created (inactive)")
+
+    @property
+    def call_active(self) -> bool:
+        """Whether a call is currently in progress."""
+        return self._call_active
+
+    # ── Lifecycle methods ──────────────────────────────────────────────────────
+
+    def start_call(self) -> PipelineUpdate:
+        """Start a new call session."""
+        self._call_active = True
+
+        # Try to initialize real VADListener; fall back to mock on error
+        try:
+            self._vad = VADListener()
+        except (RuntimeError, ValueError) as e:
+            logger.warning(f"VADListener init failed at runtime ({e}) — using MockVADListener")
+            self._vad = MockVADListener()
+
+        self._utterances = []
+        self._current_intent = None
+        self._last_response = ""
+        self._booking_info = None
+
+        self._call_id = create_call("gradio_session")
+        logger.info(f"Call started — id={self._call_id}")
+
+        return self._build_update()
+
+    def end_call(self) -> PipelineUpdate:
+        """End the current call."""
+        if not self._call_active:
+            return self._build_update()
+
+        self._call_active = False
+        if self._vad:
+            self._vad.reset()
+
+        logger.info(f"Call ended — id={self._call_id}")
+        return self._build_update()
+
+    def reset(self) -> PipelineUpdate:
+        """Full session reset."""
+        self.end_call()
+        self._call_id = None
+        self._utterances = []
+        self._current_intent = None
+        self._last_response = ""
+        self._booking_info = None
+        return self._build_update()
+
+    # ── Audio processing ──────────────────────────────────────────────────────
+
+    def process_audio_chunk(
+        self, sample_rate: int, audio: np.ndarray
+    ) -> PipelineUpdate:
+        """
+        Process one audio chunk from Gradio microphone.
+
+        If a complete utterance is detected (speech → silence), runs the
+        full pipeline: transcribe → parse intent → evaluate → update DB.
+        """
+        if not self._call_active or not self._vad or not self._call_id:
+            return self._build_update()
+
+        # ── Step 1: VAD — detect utterance boundaries ──────────────────────────
+        try:
+            utterances_detected = list(self._vad.process_chunk(sample_rate, audio))
+        except RuntimeError as e:
+            # VAD failed at runtime (e.g., torch/numpy issue) — fall back to mock
+            if not isinstance(self._vad, MockVADListener):
+                logger.warning(
+                    f"VAD process_chunk failed ({e}) — switching to MockVADListener"
+                )
+                self._vad = MockVADListener()
+                try:
+                    utterances_detected = list(self._vad.process_chunk(sample_rate, audio))
+                except Exception as e2:
+                    logger.error(f"Mock VAD also failed: {e2}")
+                    return self._build_update()
+            else:
+                logger.error(f"Mock VAD failed: {e}")
+                return self._build_update()
+
+        if not utterances_detected:
+            return self._build_update()
+
+        # ── Steps 2–5: process each utterance ──────────────────────────────────
+        for sample_rate_utt, audio_np in utterances_detected:
+            self._process_utterance(sample_rate_utt, audio_np)
+
+        return self._build_update()
+
+    # ── Internal helpers ───────────────────────────────────────────────────────
+
+    def _process_utterance(self, sample_rate: int, audio: np.ndarray) -> None:
+        """Process one complete utterance through the full pipeline."""
+        # Transcribe
+        try:
+            transcript_chunk = self._transcriber.transcribe(sample_rate, audio)
+        except (RuntimeError, ImportError) as e:
+            # Model failed to load at runtime — fall back to mock
+            if not isinstance(self._transcriber, MockTranscriber):
+                logger.warning(f"Transcriber failed ({e}) — switching to MockTranscriber")
+                self._transcriber = get_mock_transcriber()
+                transcript_chunk = self._transcriber.transcribe(sample_rate, audio)
+            else:
+                logger.error(f"MockTranscriber also failed: {e}")
+                return
+
+        if not transcript_chunk or not transcript_chunk.strip():
+            logger.debug("Transcription empty — skipping")
+            return
+
+        append_transcript(self._call_id, transcript_chunk)
+        self._utterances.append(transcript_chunk)
+        logger.info(f"Transcribed: '{transcript_chunk[:80]}…'")
+
+        # Parse intent
+        if self._parser:
+            try:
+                self._current_intent = self._parser.parse_accumulated(self._utterances)
+                update_call_intent(self._call_id, self._current_intent.model_dump())
+            except Exception as e:
+                logger.warning(f"Intent parsing failed: {e}")
+                return
+        else:
+            logger.debug("Parser not available — skipping intent parse")
+            return
+
+        # Evaluate
+        if not self._evaluator:
+            logger.debug("Evaluator not available — skipping evaluation")
+            return
+
+        slot_available = True
+        slot_conflicts = []
+
+        if (self._current_intent.preferred_date and
+            self._current_intent.preferred_time):
+            slot_available = is_slot_available(
+                self._current_intent.preferred_date,
+                self._current_intent.preferred_time,
+                self._current_intent.duration_minutes or 30,
+            )
+            if not slot_available:
+                slot_conflicts = get_booked_slots(self._current_intent.preferred_date)
+
+        try:
+            result = self._evaluator.evaluate(
+                self._current_intent,
+                self._utterances,
+                slot_available=slot_available,
+                slot_conflicts=slot_conflicts,
+            )
+        except Exception as e:
+            logger.warning(f"Evaluation failed: {e}")
+            result = None
+
+        if result:
+            update_call_decision(self._call_id, result.decision, result.reasoning)
+            self._last_response = result.spoken_response
+
+            # Handle booking confirmation
+            if result.decision == "schedule":
+                self._booking_info = {
+                    "booking_id": self._call_id,
+                    "caller": self._current_intent.caller_name or "Unknown",
+                    "date": self._current_intent.preferred_date,
+                    "time": self._current_intent.preferred_time,
+                    "duration": self._current_intent.duration_minutes or 30,
+                    "type": self._current_intent.meeting_type or "phone",
+                }
+
+            # Terminal decision — end call
+            if result.decision == "end_call":
+                self._call_active = False
+
+    def _build_update(self) -> PipelineUpdate:
+        """Build UI update from current state."""
+        status = "🟢 Active" if self._call_active else "⚫ Ready"
+        vad_speaking = self._vad.is_speaking if self._vad else False
+
+        # Format intent as markdown table
+        intent_md = self._format_intent_md()
+
+        # Call log
+        call_log = get_recent_calls(limit=10)
+
+        return PipelineUpdate(
+            status=status,
+            vad_speaking=vad_speaking,
+            transcript_lines=self._utterances,
+            intent_md=intent_md,
+            agent_response=self._last_response,
+            booking_confirmed=self._booking_info,
+            call_log=call_log,
+        )
+
+    def _format_intent_md(self) -> str:
+        """Format current intent as markdown table."""
+        if not self._current_intent:
+            return "_No data yet — waiting for first utterance…_"
+
+        intent = self._current_intent
+        rows = [
+            f"| Field | Value |",
+            f"|-------|-------|",
+            f"| Intent | `{intent.intent}` |",
+            f"| Caller | {intent.caller_name or '—'} |",
+            f"| Date | {intent.preferred_date or '—'} |",
+            f"| Time | {intent.preferred_time or '—'} |",
+            f"| Duration | {intent.duration_minutes or '—'} min |",
+            f"| Type | {intent.meeting_type or '—'} |",
+            f"| Confidence | {intent.confidence:.1%} |",
+            f"| Missing | {', '.join(intent.missing_fields) or '—'} |",
+        ]
+        return "\n".join(rows)

pipeline/transcriber.py

@@ -0,0 +1,369 @@
+"""
+pipeline/transcriber.py
+
+Cohere Transcribe wrapper for the telecalling agent.
+
+Model
+─────
+CohereLabs/cohere-transcribe-03-2026
+  - 2B parameter Conformer encoder + lightweight Transformer decoder
+  - Audio-in, text-out ASR, Apache 2.0
+  - Gated: requires HF_TOKEN with accepted licence at
+    https://huggingface.co/CohereLabs/cohere-transcribe-03-2026
+
+Memory budget on RTX 2050 (4 GB VRAM)
+──────────────────────────────────────
+  Full fp32  →  ~8 GB   ✗
+  float16    →  ~4 GB   ✗ (tight, unstable with other allocations)
+  torch_dtype=torch.float16 + device_map keeps ~2.2 GB → ✓
+
+Design decisions
+────────────────
+  - Lazy-loaded: model is not pulled from HF until the first call arrives.
+    This keeps Gradio startup fast and avoids OOM if the user never speaks.
+  - GPU-first with automatic CPU fallback: if CUDA is unavailable or VRAM
+    is insufficient the model moves to CPU transparently.
+  - Uses model.transcribe(audio_arrays=...) to avoid disk I/O entirely.
+  - compile=False by default: torch.compile causes a 30-60 s warmup on
+    first call which would break the live-call UX. Set compile=True only
+    in batch / offline mode.
+  - Thread-safe: model loading is protected by a threading.Lock so
+    concurrent Gradio sessions don't race to download.
+
+Usage
+─────
+    transcriber = Transcriber()                 # cheap — model not loaded yet
+    text = transcriber.transcribe(sr, audio_np) # loads model on first call
+    transcriber.unload()                        # free VRAM between sessions
+"""
+
+import logging
+import threading
+import time
+from typing import Optional
+
+import numpy as np
+import torch
+
+from config import (
+    TRANSCRIBE_MODEL_ID,
+    TRANSCRIBE_LANGUAGE,
+    TRANSCRIBE_DEVICE,
+    HF_TOKEN,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class Transcriber:
+    """
+    Lazy-loading wrapper around CohereLabs/cohere-transcribe-03-2026.
+
+    Thread-safe: a single instance can be shared across the whole app.
+    """
+
+    def __init__(self):
+        self._model     = None
+        self._processor = None
+        self._device    = None
+        self._lock      = threading.Lock()
+        self._loaded    = False
+
+    # ── Public API ────────────────────────────────────────────────────────────
+
+    def transcribe(self, sample_rate: int, audio: np.ndarray) -> str:
+        """
+        Transcribe a single utterance.
+
+        Parameters
+        ----------
+        sample_rate : int
+            Sample rate of `audio` (typically 16000 from VAD).
+        audio : np.ndarray
+            Mono float32 PCM in [-1.0, 1.0].
+
+        Returns
+        -------
+        str
+            Transcribed text, stripped of leading/trailing whitespace.
+            Returns "" on error so the pipeline can continue gracefully.
+        """
+        if audio is None or len(audio) == 0:
+            return ""
+
+        self._ensure_loaded()
+
+        audio = self._validate_audio(audio)
+        if audio is None:
+            return ""
+
+        try:
+            t0 = time.perf_counter()
+
+            results = self._model.transcribe(
+                processor       = self._processor,
+                audio_arrays    = [audio],        # list of np.ndarray
+                sample_rates    = [sample_rate],  # matching list of ints
+                language        = TRANSCRIBE_LANGUAGE,
+                punctuation     = True,
+                batch_size      = 1,              # one utterance at a time (live)
+                compile         = False,          # no warmup delay in live mode
+                pipeline_detokenization = False,  # not needed for batch_size=1
+            )
+
+            elapsed = time.perf_counter() - t0
+            text    = results[0].strip() if results else ""
+
+            duration = len(audio) / sample_rate
+            rtfx     = duration / elapsed if elapsed > 0 else 0
+            logger.info(
+                f"Transcribed {duration:.2f}s audio in {elapsed:.2f}s "
+                f"(RTFx {rtfx:.1f}x): '{text[:80]}{'…' if len(text)>80 else ''}'"
+            )
+            return text
+
+        except Exception as exc:
+            logger.error(f"Transcription failed: {exc}", exc_info=True)
+            return ""
+
+    def transcribe_batch(
+        self, utterances: list[tuple[int, np.ndarray]]
+    ) -> list[str]:
+        """
+        Transcribe multiple utterances in one forward pass.
+        More efficient for post-call batch processing.
+
+        Parameters
+        ----------
+        utterances : list of (sample_rate, audio_np) tuples
+
+        Returns
+        -------
+        list of str — one entry per input, "" on individual errors
+        """
+        if not utterances:
+            return []
+
+        self._ensure_loaded()
+
+        sample_rates  = [sr  for sr, _  in utterances]
+        audio_arrays  = [self._validate_audio(a) for _, a in utterances]
+
+        # Replace any None (invalid) arrays with silent arrays
+        audio_arrays = [
+            a if a is not None else np.zeros(sample_rates[i], dtype=np.float32)
+            for i, a in enumerate(audio_arrays)
+        ]
+
+        try:
+            results = self._model.transcribe(
+                processor       = self._processor,
+                audio_arrays    = audio_arrays,
+                sample_rates    = sample_rates,
+                language        = TRANSCRIBE_LANGUAGE,
+                punctuation     = True,
+                batch_size      = len(utterances),
+                compile         = False,
+                pipeline_detokenization = True,  # helps with larger batches
+            )
+            return [r.strip() for r in results]
+
+        except Exception as exc:
+            logger.error(f"Batch transcription failed: {exc}", exc_info=True)
+            return [""] * len(utterances)
+
+    def unload(self):
+        """
+        Release GPU memory.  Call at end of a call session if memory is tight.
+        Model will be reloaded lazily on the next call.
+        """
+        with self._lock:
+            if self._loaded:
+                del self._model
+                del self._processor
+                self._model     = None
+                self._processor = None
+                self._loaded    = False
+                if torch.cuda.is_available():
+                    torch.cuda.empty_cache()
+                logger.info("Transcriber unloaded — VRAM freed.")
+
+    @property
+    def is_loaded(self) -> bool:
+        return self._loaded
+
+    @property
+    def device(self) -> Optional[str]:
+        return str(self._device) if self._device else None
+
+    # ── Internal ──────────────────────────────────────────────────────────────
+
+    def _ensure_loaded(self):
+        """Load model + processor exactly once, thread-safely."""
+        if self._loaded:
+            return
+
+        with self._lock:
+            if self._loaded:   # double-checked locking
+                return
+            self._load()
+
+    def _load(self):
+        """
+        Pull model from HuggingFace and move to best available device.
+
+        Device selection for RTX 2050 (4 GB):
+          - float16 on CUDA: ~2.2 GB VRAM ← preferred
+          - float32 on CPU:  ~8.0 GB RAM  ← fallback
+        """
+        from transformers import AutoProcessor, AutoModelForSpeechSeq2Seq
+
+        logger.info(f"Loading Cohere Transcribe ({TRANSCRIBE_MODEL_ID})…")
+        t0 = time.perf_counter()
+
+        token_kwargs = {"token": HF_TOKEN} if HF_TOKEN else {}
+
+        # Processor (tokenizer + feature extractor) — tiny, load first
+        self._processor = AutoProcessor.from_pretrained(
+            TRANSCRIBE_MODEL_ID,
+            trust_remote_code=True,
+            **token_kwargs,
+        )
+
+        # Determine device & dtype
+        if torch.cuda.is_available():
+            try:
+                self._device = torch.device(TRANSCRIBE_DEVICE)
+                dtype        = torch.float16
+                logger.info(f"CUDA available — loading in float16 on {self._device}")
+            except Exception:
+                self._device = torch.device("cpu")
+                dtype        = torch.float32
+                logger.warning("CUDA device init failed — falling back to CPU float32")
+        else:
+            self._device = torch.device("cpu")
+            dtype        = torch.float32
+            logger.info("No CUDA — loading on CPU in float32 (will be slower)")
+
+        # Model weights
+        self._model = AutoModelForSpeechSeq2Seq.from_pretrained(
+            TRANSCRIBE_MODEL_ID,
+            trust_remote_code = True,
+            torch_dtype       = dtype,
+            low_cpu_mem_usage = True,   # stream weights instead of double-buffering
+            **token_kwargs,
+        ).to(self._device)
+
+        self._model.eval()
+
+        elapsed = time.perf_counter() - t0
+        logger.info(
+            f"Cohere Transcribe ready on {self._device} "
+            f"(dtype={dtype}, loaded in {elapsed:.1f}s)"
+        )
+
+        if torch.cuda.is_available():
+            allocated = torch.cuda.memory_allocated(self._device) / 1024**3
+            reserved  = torch.cuda.memory_reserved(self._device)  / 1024**3
+            logger.info(
+                f"VRAM after load — allocated: {allocated:.2f} GB, "
+                f"reserved: {reserved:.2f} GB"
+            )
+
+        self._loaded = True
+
+    @staticmethod
+    def _validate_audio(audio: np.ndarray) -> Optional[np.ndarray]:
+        """
+        Ensure audio is mono float32.  Returns None if unusable.
+        """
+        if audio is None or len(audio) == 0:
+            return None
+
+        audio = np.array(audio, dtype=np.float32)
+
+        # Stereo → mono
+        if audio.ndim == 2:
+            audio = audio.mean(axis=1)
+        elif audio.ndim != 1:
+            logger.warning(f"Unexpected audio shape {audio.shape} — skipping")
+            return None
+
+        # Normalise int16-range inputs
+        if audio.max() > 1.0 or audio.min() < -1.0:
+            audio = audio / 32768.0
+
+        audio = np.clip(audio, -1.0, 1.0)
+        return audio
+
+
+# ── Module-level singleton ─────────────────────────────────────────────────────
+# Shared across the whole app — model loaded once, reused every utterance.
+_transcriber: Optional[Transcriber] = None
+
+
+def get_transcriber() -> Transcriber:
+    """Return the module-level singleton, creating it if needed."""
+    global _transcriber
+    if _transcriber is None:
+        _transcriber = Transcriber()
+    return _transcriber
+
+
+# ── Offline smoke test (no HF token needed) ───────────────────────────────────
+
+def _smoke_test_offline():
+    """
+    Validates the audio pre-processing path without loading the model.
+    Full model load requires a HF token + accepted Cohere licence.
+    """
+    import math
+    logging.basicConfig(level=logging.INFO)
+    logger.info("Running offline smoke test (pre-processing only)…")
+
+    SR = 16000
+
+    # 1. Stereo int16 → mono float32
+    stereo_int16 = (np.random.randn(SR, 2) * 32767).astype(np.int16)
+    result = Transcriber._validate_audio(stereo_int16)
+    assert result is not None
+    assert result.ndim   == 1,        f"Expected mono, got shape {result.shape}"
+    assert result.dtype  == np.float32
+    assert result.max()  <= 1.0
+    assert result.min()  >= -1.0
+    logger.info("  ✓ Stereo int16 → mono float32 normalisation")
+
+    # 2. Already mono float32 — passthrough
+    mono_float = np.sin(2 * math.pi * 440 * np.linspace(0, 1, SR)).astype(np.float32)
+    result = Transcriber._validate_audio(mono_float)
+    assert result is not None and result.shape == (SR,)
+    logger.info("  ✓ Mono float32 passthrough")
+
+    # 3. Empty input → None
+    result = Transcriber._validate_audio(np.array([]))
+    assert result is None
+    logger.info("  ✓ Empty input → None")
+
+    # 4. Singleton pattern
+    t1 = get_transcriber()
+    t2 = get_transcriber()
+    assert t1 is t2
+    logger.info("  ✓ Module singleton")
+
+    logger.info("\nOffline smoke test PASSED ✓")
+    logger.info(
+        "\nTo run the full model test (requires HF_TOKEN + accepted licence):\n"
+        "  export HF_TOKEN=hf_...\n"
+        "  PYTHONPATH=. python3 -c \"\n"
+        "  from pipeline.transcriber import get_transcriber\n"
+        "  import numpy as np\n"
+        "  t = get_transcriber()\n"
+        "  # 2s of 440 Hz tone — expect near-empty transcription\n"
+        "  audio = np.sin(2*3.14*440*np.linspace(0,2,32000)).astype('float32')\n"
+        "  print(repr(t.transcribe(16000, audio)))\n"
+        "  \""
+    )
+
+
+if __name__ == "__main__":
+    _smoke_test_offline()

pipeline/vad_listener.py

@@ -0,0 +1,302 @@
+"""
+pipeline/vad_listener.py
+
+Silero VAD-based speech listener.
+
+Architecture
+────────────
+Gradio streams raw PCM audio from the browser microphone in small numpy
+chunks.  This module wraps that stream in a stateful VADListener that:
+
+  1. Feeds each chunk to Silero VADIterator (runs on CPU via ONNX).
+  2. Accumulates PCM frames while speech is detected.
+  3. On silence-end event, flushes the complete utterance into a
+     thread-safe queue as a (sample_rate, np.ndarray) tuple.
+  4. The orchestrator reads from that queue and passes each utterance
+     to Cohere Transcribe.
+
+Because Gradio's streaming callback fires on the main thread, all VAD
+processing is synchronous and cheap (< 1 ms per 250 ms chunk on CPU).
+No background threads are needed at this stage.
+
+Usage
+─────
+    listener = VADListener()
+
+    # Called by Gradio's streaming audio component on each chunk:
+    for utterance in listener.process_chunk(sample_rate, audio_np):
+        # utterance is (sample_rate, np.ndarray[float32]) — ready for ASR
+        transcript = transcriber.transcribe(*utterance)
+
+    # Reset between calls:
+    listener.reset()
+"""
+
+import logging
+import queue
+from typing import Generator, Optional, Tuple
+
+import numpy as np
+import torch
+
+from silero_vad import VADIterator, load_silero_vad
+
+from config import (
+    VAD_SAMPLE_RATE,
+    VAD_CHUNK_SAMPLES,
+    VAD_SILENCE_THRESHOLD,
+    VAD_SILENCE_DURATION_S,
+    VAD_MIN_SPEECH_S,
+)
+
+# Silero v6 requires exactly 512-sample windows at 16kHz (32 ms).
+# We still accept larger Gradio chunks and stride through them internally.
+VAD_WINDOW_SAMPLES = 512
+
+logger = logging.getLogger(__name__)
+
+# ── Types ─────────────────────────────────────────────────────────────────────
+Utterance = Tuple[int, np.ndarray]   # (sample_rate, float32 PCM)
+
+
+class VADListener:
+    """
+    Stateful VAD processor.  One instance per Gradio user session.
+
+    The VADIterator state machine emits:
+      {'start': sample_idx}  — speech onset detected
+      {'end':   sample_idx}  — silence detected after speech (utterance done)
+      None                   — mid-speech or mid-silence, keep accumulating
+
+    We buffer raw PCM frames during the speech window and yield a complete
+    utterance when 'end' fires.
+    """
+
+    def __init__(self):
+        logger.info("Loading Silero VAD model (ONNX, CPU)…")
+        self._model = load_silero_vad(onnx=True)     # ONNX = no PyTorch GPU mem
+
+        # min_silence_duration_ms drives the 'end' event timing.
+        # We use VAD_SILENCE_DURATION_S from config (default 0.8 s = 800 ms).
+        self._vad = VADIterator(
+            model=self._model,
+            threshold=VAD_SILENCE_THRESHOLD,
+            sampling_rate=VAD_SAMPLE_RATE,
+            min_silence_duration_ms=int(VAD_SILENCE_DURATION_S * 1000),
+            speech_pad_ms=60,       # 60 ms padding on each side — natural feel
+        )
+
+        self._speech_buffer: list[np.ndarray] = []  # accumulates speech frames
+        self._speaking: bool = False
+
+        # Internal queue — orchestrator can also pull from here if preferred
+        self._utterance_queue: queue.SimpleQueue = queue.SimpleQueue()
+
+        logger.info("Silero VAD ready.")
+
+    # ── Public interface ──────────────────────────────────────────────────────
+
+    def process_chunk(
+        self, sample_rate: int, audio: np.ndarray
+    ) -> Generator[Utterance, None, None]:
+        """
+        Accept one audio chunk from Gradio and yield zero or one utterance.
+
+        Parameters
+        ----------
+        sample_rate : int
+            Sample rate reported by Gradio (may differ from VAD_SAMPLE_RATE).
+        audio : np.ndarray
+            Raw PCM, any shape.  Will be normalised to mono float32 16 kHz.
+
+        Yields
+        ------
+        (VAD_SAMPLE_RATE, np.ndarray[float32])
+            Complete utterance, ready to be passed to Cohere Transcribe.
+        """
+        if audio is None or len(audio) == 0:
+            return
+
+        # ── 1. Normalise to mono float32 ──────────────────────────────────────
+        audio = self._to_mono_float32(audio)
+
+        # ── 2. Resample if Gradio gives a different rate ──────────────────────
+        if sample_rate != VAD_SAMPLE_RATE:
+            audio = self._resample(audio, sample_rate, VAD_SAMPLE_RATE)
+
+        # ── 3. Pad to chunk boundary (Silero needs fixed-size windows) ────────
+        audio = self._pad_to_chunk(audio)
+
+        # ── 4. Walk through 512-sample windows (Silero v6 requirement) ─────────
+        for i in range(0, len(audio), VAD_WINDOW_SAMPLES):
+            window = audio[i : i + VAD_WINDOW_SAMPLES]
+            if len(window) < VAD_WINDOW_SAMPLES:
+                break   # incomplete trailing window — skip
+
+            tensor = torch.from_numpy(window)
+            event  = self._vad(tensor)
+
+            if event is None:
+                # Mid-speech: keep buffering.  Mid-silence: do nothing.
+                if self._speaking:
+                    self._speech_buffer.append(window)
+
+            elif "start" in event:
+                # Speech onset — start accumulating
+                if not self._speaking:
+                    logger.debug(f"Speech START at sample {event['start']}")
+                    self._speaking = True
+                    self._speech_buffer = [window]  # include onset window
+                
+            elif "end" in event:
+                # Speech ended — flush if long enough
+                if self._speaking:
+                    self._speaking = False
+                    self._speech_buffer.append(window)  # include trailing window
+                    utterance = self._flush()
+                    if utterance is not None:
+                        logger.debug(
+                            f"Utterance flushed: {len(utterance[1])/VAD_SAMPLE_RATE:.2f}s"
+                        )
+                        yield utterance
+
+    def reset(self):
+        """
+        Call between calls / sessions to wipe VAD state and audio buffer.
+        """
+        self._vad.reset_states()
+        self._speech_buffer = []
+        self._speaking = False
+        # drain the queue
+        while not self._utterance_queue.empty():
+            try:
+                self._utterance_queue.get_nowait()
+            except queue.Empty:
+                break
+        logger.debug("VADListener reset.")
+
+    @property
+    def is_speaking(self) -> bool:
+        return self._speaking
+
+    # ── Internal helpers ──────────────────────────────────────────────────────
+
+    def _flush(self) -> Optional[Utterance]:
+        """
+        Concatenate buffer frames into a single utterance array.
+        Drops utterances shorter than VAD_MIN_SPEECH_S (noise / clicks).
+        """
+        if not self._speech_buffer:
+            return None
+
+        utterance = np.concatenate(self._speech_buffer, axis=0)
+        duration  = len(utterance) / VAD_SAMPLE_RATE
+
+        from config import VAD_MIN_SPEECH_S
+        if duration < VAD_MIN_SPEECH_S:
+            logger.debug(f"Utterance too short ({duration:.2f}s) — discarded.")
+            self._speech_buffer = []
+            return None
+
+        self._speech_buffer = []
+        return (VAD_SAMPLE_RATE, utterance)
+
+    @staticmethod
+    def _to_mono_float32(audio: np.ndarray) -> np.ndarray:
+        """
+        Convert any numpy audio array to mono float32 in [-1.0, 1.0].
+        Gradio typically gives int16 or float32; handles both.
+        """
+        audio = np.array(audio, dtype=np.float32)
+
+        # Stereo → mono
+        if audio.ndim == 2:
+            audio = audio.mean(axis=1)
+
+        # int16 range → float32 [-1, 1]
+        if audio.max() > 1.0 or audio.min() < -1.0:
+            audio = audio / 32768.0
+
+        # Clip to valid range
+        audio = np.clip(audio, -1.0, 1.0)
+        return audio
+
+    @staticmethod
+    def _resample(audio: np.ndarray, orig_sr: int, target_sr: int) -> np.ndarray:
+        """
+        Simple linear interpolation resample.
+        Avoids importing torchaudio here (saves CUDA context startup cost).
+        Good enough for pre-processing before VAD.
+        """
+        if orig_sr == target_sr:
+            return audio
+        ratio      = target_sr / orig_sr
+        new_length = int(len(audio) * ratio)
+        return np.interp(
+            np.linspace(0, len(audio) - 1, new_length),
+            np.arange(len(audio)),
+            audio,
+        ).astype(np.float32)
+
+    @staticmethod
+    def _pad_to_chunk(audio: np.ndarray) -> np.ndarray:
+        """
+        Zero-pad audio so its length is a multiple of VAD_CHUNK_SAMPLES.
+        Silero requires fixed-size windows.
+        """
+        remainder = len(audio) % VAD_WINDOW_SAMPLES
+        if remainder:
+            padding = VAD_WINDOW_SAMPLES - remainder
+            audio   = np.concatenate([audio, np.zeros(padding, dtype=np.float32)])
+        return audio
+
+
+# ── Standalone smoke test ─────────────────────────────────────────────────────
+
+def _smoke_test():
+    """
+    Simulates a real mic stream:
+      - 0.5 s silence
+      - 1.5 s synthetic speech-like tone (500 Hz sine)
+      - 0.9 s silence  ← should trigger 'end' and yield one utterance
+      - 0.5 s silence
+    """
+    import math
+    logging.basicConfig(level=logging.DEBUG)
+
+    SR   = VAD_SAMPLE_RATE
+    FREQ = 500            # Hz — sine at 500 Hz is reliably detected as speech
+
+    def sine_chunk(duration_s: float, frequency: float = 0) -> np.ndarray:
+        n = int(SR * duration_s)
+        if frequency == 0:
+            return np.zeros(n, dtype=np.float32)
+        t = np.linspace(0, duration_s, n, endpoint=False)
+        return (0.6 * np.sin(2 * math.pi * frequency * t)).astype(np.float32)
+
+    timeline = [
+        sine_chunk(0.5),          # silence
+        sine_chunk(1.5, FREQ),    # speech
+        sine_chunk(0.9),          # silence — should flush
+        sine_chunk(0.5),          # trailing silence
+    ]
+
+    listener = VADListener()
+    utterances_received = 0
+
+    for chunk in timeline:
+        # Feed in 250 ms sub-chunks as Gradio would
+        for start in range(0, len(chunk), VAD_CHUNK_SAMPLES):
+            sub = chunk[start : start + VAD_CHUNK_SAMPLES]
+            for utt in listener.process_chunk(SR, sub):
+                utterances_received += 1
+                duration = len(utt[1]) / utt[0]
+                print(f"  ✓ Utterance received: {duration:.2f}s at {utt[0]} Hz")
+
+    print(f"\nSmoke test complete — {utterances_received} utterance(s) flushed.")
+    assert utterances_received >= 1, "Expected at least one utterance!"
+    print("PASSED ✓")
+
+
+if __name__ == "__main__":
+    _smoke_test()

requirements.txt

@@ -0,0 +1,34 @@
+# ── Core ML / inference ───────────────────────────────────────────────────────
+torch==2.3.1
+torchaudio==2.3.1
+transformers>=4.56,<5.3,!=5.0.*,!=5.1.*   # Cohere Transcribe requirement (exact)
+accelerate>=0.30.0
+bitsandbytes>=0.43.0                       # 4-bit quant for MiniCPM on CPU
+sentencepiece
+protobuf
+
+# ── Cohere Transcribe ASR ─────────────────────────────────────────────────────
+soundfile
+librosa
+huggingface_hub
+
+# ── Qwen2.5 intent parser (llama-cpp-python with CUDA offload) ────────────────
+# Install with CUDA support — do NOT use the plain pip version.
+# Run this separately after pip install -r requirements.txt:
+#   CMAKE_ARGS="-DGGML_CUDA=on" pip install llama-cpp-python --no-cache-dir
+# Windows (MSVC):
+#   set CMAKE_ARGS=-DGGML_CUDA=on && pip install llama-cpp-python --no-cache-dir
+llama-cpp-python>=0.2.85     # plain CPU fallback — override with CUDA build above
+
+# ── VAD ───────────────────────────────────────────────────────────────────────
+silero-vad                   # pulls torch; also needs torchaudio for audio I/O
+onnxruntime>=1.16.0          # ONNX runtime for Silero VAD
+pyaudio                      # microphone access (Linux: apt install python3-pyaudio)
+numpy>=1.24.0
+
+# ── Gradio UI ─────────────────────────────────────────────────────────────────
+gradio>=4.40.0
+
+# ── Utilities ─────────────────────────────────────────────────────────────────
+python-dotenv                # optional: load HF_TOKEN from .env file
+pydantic>=2.0                # JSON schema validation for Qwen output