Alpha/InterviewAI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add context windows, web search, streamed answers, overlay move/resize; overhaul docs

Browse Source 
- Context windows (contexts/*.yaml) scope answers to a defined domain
- Optional DuckDuckGo web search behind ai.web_search.enabled (default off)
- Stream partial answers into the overlay at first-token time
- Default Whisper to local base.en (~9x faster); offline model loading
- Priority-ordered loopback detection (BlackHole > Teams device)
- Overlay: drag interior to move, edges to resize
- Stop tracking model binaries (models/ is gitignored)
- README/CLAUDE.md overhaul + tracked config.example.yaml

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
spiro-alvin-nyasimi
2026-07-03 11:22:34 +03:00
parent 138580475c
commit 08118cc650
13 changed files with 1095 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
.gitignore vendored
View File

@@ -33,12 +33,22 @@ logs/
*.log
# Configuration files with sensitive data
# (config.example.yaml is the tracked template — copy it to config.yaml)
config.yaml
config/local.yaml
config/production.yaml
*.local.yaml
*.secret.yaml
# Context windows: personal interview/meeting contexts stay private.
# Only the README and the shipped example are tracked.
contexts/*
!contexts/README.md
!contexts/python_flask_backend.yaml
# Claude Code local settings
.claude/settings.local.json
# Model files (large binaries — downloaded by setup.py / faster-whisper, not tracked)
models/

28
CLAUDE.md
View File

@@ -4,7 +4,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
## What this is
A real-time, fully on-device AI copilot for meetings (macOS-focused). **Push-to-talk** (hold Right Option) gates audio capture; it transcribes locally with **faster-whisper** (`distil-large-v3`), detects questions, and answers via a quantized **Qwen2.5-VL-7B** vision model through **llama.cpp**. The same vision model also reads **on-screen questions**: press a hotkey, drag a box, and it answers what's in the region. Answers appear in a Qt overlay hidden from screen capture. No audio or text leaves the machine.
A real-time, fully on-device AI copilot for meetings (macOS-focused). **Push-to-talk** (hold Right Option) gates audio capture; it transcribes locally with **faster-whisper** (local `models/whisper-base.en` by default — the latency-critical choice, see Latency below), detects questions, and answers via a quantized **Qwen2.5-VL-7B** vision model through **llama.cpp**, **streaming partial answers into the overlay** as they generate. The same vision model also reads **on-screen questions**: press a hotkey, drag a box, and it answers what's in the region. Answers can be scoped to a **context window** (`contexts/*.yaml`, e.g. "Flask interview") so the model stays on-topic. Answers appear in a Qt overlay hidden from screen capture (drag interior to move, edges to resize). No audio or text leaves the machine — except the **opt-in** DuckDuckGo web search (`ai.web_search.enabled`, default off).
## Commands
@@ -25,12 +25,12 @@ There is **no test suite, linter, or build step**. To test the AI loop without s
Data flow:
1. **`src/hotkeys.py`** (`HotkeyManager`) — a pynput global listener. Holding the push-to-talk key (`audio.ptt_key`, default `alt_r` = Right Option) calls `audio_listener.arm()`/`disarm()`; the screen chord (`screen.capture_key`, default `ctrl+shift+space`) calls `overlay.request_capture()`. Callbacks run on pynput's thread, so anything touching Qt must be marshaled (see below). **Requires macOS Accessibility permission.**
2. **`src/audio_listener.py`** — one `sd.RawInputStream` per source ("microphone" / "system"). In **push-to-talk mode** (default, `audio.capture_mode`), stream callbacks buffer raw PCM *only between `arm()` and `disarm()`*; on disarm the held clip per source is queued for transcription. (The original always-on `StreamSegmenter` VAD path still exists and is used only in `continuous` mode.) A single shared `WhisperModel` (config-driven `audio.whisper_model`, int8 CPU) transcribes; calls back into `MeetingAssistant.on_audio_transcript(text, timestamp, source, speaker)`.
2. **`src/audio_listener.py`** — one `sd.RawInputStream` per source ("microphone" / "system"). In **push-to-talk mode** (default, `audio.capture_mode`), stream callbacks buffer raw PCM *only between `arm()` and `disarm()`*; on disarm the held clip per source is queued for transcription. (The original always-on `StreamSegmenter` VAD path still exists and is used only in `continuous` mode; `toggle_auto_listen()` switches at runtime via Ctrl+Shift+M.) A single shared `WhisperModel` (config-driven `audio.whisper_model`, int8 CPU) transcribes — **every `transcribe()` call is serialized through `_model_lock` (faster-whisper is not thread-safe)** — then calls back into `MeetingAssistant.on_audio_transcript(text, timestamp, source, speaker)`. The loader tries local dirs first (`models/whisper-base.en` → `models/whisper-medium.en` → …) with `local_files_only` for any directory, so runtime never touches Hugging Face (blocked by the corp proxy). Loopback detection (`_find_system_device`) walks `audio.loopback_keywords` in **priority order** — a true loopback (BlackHole) beats app-specific devices ("Microsoft Teams Audio", which is Teams-only and often silent). An optional **incremental streaming transcriber** (`_StreamingTranscriber`, `audio.streaming_transcription`, default OFF) transcribes the growing clip while the key is held — see Latency for why it's off.
3. **`src/diarizer.py`** (`SpeakerDiarizer`) — the "system" loopback stream is one *mixed* channel of all remote participants, so per-utterance it computes an ECAPA-TDNN voice embedding (SpeechBrain, `models/ecapa`) and online-clusters by cosine similarity into "Person 1/2/…". Best-effort; degrades to a single "Them" if unavailable.
4. **`src/context_manager.py`** (`ContextManager`) — rolling speaker-labeled transcript (deque of last 20 utterances) used as LLM context.
5. **`src/ai_engine.py`** (`AIEngine`) — `classify()` buckets a spoken question into `math | factual | open`. **Math is solved with a safe AST evaluator (`try_solve_math`), never the LLM.** Other kinds go to llama.cpp via `create_chat_completion` (**streamed**, so a newer question interrupts an in-flight one via `interrupt_event`; all generation is serialized through `_gen_lock` since llama.cpp isn't thread-safe). The model is loaded with a `Qwen25VLChatHandler` + `mmproj` when `ai.mmproj` is present (`self.vision = True`). **`answer_from_image(path)`** sends a screenshot (base64 data-URI) + instruction as a multimodal chat message — this is the screen-question path.
5. **`src/ai_engine.py`** (`AIEngine`) — `classify()` buckets a spoken question into `math | factual | open`. **Math is solved with a safe AST evaluator (`try_solve_math`), never the LLM.** Other kinds go to llama.cpp via `create_chat_completion` (**streamed**; all generation is serialized through `_gen_lock` since llama.cpp isn't thread-safe). `_stream_chat` emits throttled partial text via an `on_partial` callback (wired by `main.py` to the overlay with a `▌` cursor) so first words show at first-token time; it prints `⏱ gen` timing per answer. **Interrupt semantics:** `interrupt()` only fires while a generation is in-flight (`_generating`), and the flag is cleared in a `finally` — a stale interrupt must never kill the *next* question (a real bug that dropped rapid follow-ups). `_system_prompt` prepends `_context_preamble()` — the active **context window** from `ContextLibrary` (`contexts/*.yaml`, hot-reloaded; `strict: true` = no out-of-scope variations). `_maybe_web_search()` (opt-in `ai.web_search`) grounds factual/open committed answers with DuckDuckGo snippets — never for math or suggested replies. The model is loaded with a `Qwen25VLChatHandler` + `mmproj` when `ai.mmproj` is present (`self.vision = True`). **`answer_from_image(path)`** sends a screenshot (base64 data-URI) + instruction as a multimodal chat message — this is the screen-question path (also context-scoped and partial-streamed).
6. **`src/region_capture.py`** (`RegionSelector` + `grab_region`) — fullscreen dim `QWidget` with a `QRubberBand`; on drag-release it maps to global coords, grabs the region via `mss` (scaling logical→physical by comparing mss size to the Qt screen size, so it's Retina-correct), writes a temp PNG, and fires `on_done(path)`. **Must be created on the Qt main thread.**
7. **`src/overlay.py`** (`InvisibleOverlay`) — PyQt5 always-on-top frameless window. Worker threads must update it only via the `_Bridge` Qt signals (`show_answer`/`show_status`/`capture_request`), never directly. `request_capture()` emits `capture_request`, whose slot runs `self.on_capture` (set by `main.py`) on the main thread — this is how the off-thread hotkey safely launches the selector. `_exclude_from_screen_capture()` sets macOS `NSWindowSharingNone` so screen-recorders/Zoom can't see it.
7. **`src/overlay.py`** (`InvisibleOverlay`) — PyQt5 always-on-top frameless window. Worker threads must update it only via the `_Bridge` Qt signals (`show_answer`/`show_status`/`capture_request`), never directly. `request_capture()` emits `capture_request`, whose slot runs `self.on_capture` (set by `main.py`) on the main thread — this is how the off-thread hotkey safely launches the selector. `_exclude_from_screen_capture()` sets macOS `NSWindowSharingNone` so screen-recorders/Zoom can't see it. The text area is a `_ResizeTextEdit`: press near an edge/corner to **resize**, press the interior and drag to **move**. Hard-won macOS quirks baked into it: fully-transparent regions of a translucent window pass clicks through (so only the opaque widget can catch presses), and `grabMouse()` never delivers the drag stream for this window — drags are tracked via `startSystemResize`/`startSystemMove` first, falling back to an **application-wide event filter**.
Threading model: Qt event loop on the main thread; the pynput hotkey listener, audio capture, transcription, terminal input, and each answer run on their own threads. Anything touching Qt from off-thread goes through the `_Bridge` signals. Answer generation is serialized/interruptible via `answer_lock` (main.py) + `interrupt_event` + `_gen_lock` (ai_engine). The region selector and screen answer are driven from `main.py._handle_screen_capture` → `_on_region_captured` → `_screen_answer_worker`.
@@ -43,10 +43,24 @@ Threading model: Qt event loop on the main thread; the pynput hotkey listener, a
## Config & assets
- **`config.yaml`** is the live config (loaded by `main.py`). `config/settings.py` is unused legacy — ignore it.
- Models under `models/` (large binaries): `models/ecapa/` (SpeechBrain diarizer), and the **Qwen2.5-VL** vision model + `mmproj` (downloaded by `setup.py` from `ggml-org/Qwen2.5-VL-7B-Instruct-GGUF`). The Whisper STT model auto-downloads by name via faster-whisper on first run (the old committed `models/whisper/` base.en dir is now only a fallback). The old `models/Qwen2.5-7B-Instruct-Q4_K_M.gguf` (text-only) is superseded by the VL model.
- **`config.yaml`** is the live config (loaded by `main.py`) and is **gitignored** — `config.example.yaml` is the tracked template; keep it in sync when adding config options. `config/settings.py` is unused legacy — ignore it.
- **`contexts/`** holds the context-window YAMLs (see `contexts/README.md`). The example `python_flask_backend.yaml` is tracked; user-personal context files are gitignored.
- Models under `models/` (large binaries, all gitignored): `models/whisper-base.en/` (default STT, downloaded directly from HF via unverified urllib because the corp proxy breaks `huggingface_hub`), `models/whisper-medium.en/` (slower/more robust alternative), `models/ecapa/` (SpeechBrain diarizer), and the **Qwen2.5-VL** vision model + `mmproj` (downloaded by `setup.py` from `ggml-org/Qwen2.5-VL-7B-Instruct-GGUF`). `models/whisper/` (bare model.bin, broken — missing tokenizer) and the old text-only Qwen GGUF are legacy.
- **`src/screen_scanner.py` is dead code** — superseded by `src/region_capture.py`. The OCR screen-scan was always disabled (no-op stub); screen questions now go through the draw-a-box vision path instead.
- **16 GB RAM is the binding constraint.** VL model (~4.7 GB) + mmproj (~1.4 GB) + Whisper + ECAPA + Qt ≈ 12–13 GB. If memory is tight, set `audio.whisper_model: medium.en`.
- **16 GB RAM is the binding constraint.** VL model (~4.7 GB) + mmproj (~1.4 GB) + Whisper + ECAPA + Qt ≈ 12–13 GB.
## Latency (measured, don't regress)
- **faster-whisper pads every `transcribe()` call to a fixed 30 s encoder window**, so per-call cost is nearly constant regardless of clip length (`medium.en` ≈ 2.3–2.9 s even for a 0.5 s tail; `base.en` ≈ 0.24–0.31 s). Consequences: the model choice, not pipeline tricks, dominates transcription latency, and incremental/streaming transcription (`_StreamingTranscriber`) barely helps on fast models while risking dropped words at chunk seams — that's why it defaults OFF.
- **Generation latency is first-token (prefill) + tokens.** Partial answers stream to the overlay (`⚡ first words` metric); the system prompt (context window) is constant per session so llama.cpp's prefix cache makes warm questions much faster than cold (~2.5 s vs ~5.6 s total observed). Keep `_user_prompt` lean — its transcript/memory budgets were deliberately trimmed.
- Console metrics: `⏱ whisper`, `⏱ gen: first token/total`, `⚡ first words on screen`.
## macOS permissions (frequent support issue)
All three go to the **host app that launches the process** (Terminal/iTerm/PyCharm), and macOS only applies them after relaunch:
- **Accessibility** — without it pynput logs `This process is not trusted!` and NO global hotkey ever fires (push-to-talk, screen grab).
- **Screen & System Audio Recording** — without it `mss` silently captures a windowless desktop (wallpaper + menu bar only), so screen questions read nothing.
- **Microphone** — audio capture.
## Platform notes

126
README.md
View File

@@ -1,20 +1,31 @@
# Meeting Assistant
A real-time, on-device AI copilot for meetings. You **hold a key to talk**, it
detects questions (even ones phrased as plain statements), and shows answers or
suggested replies in an overlay that is hidden from screen-capture. It can also
**read a question off your screen** when you drag a box around it.
A real-time, on-device AI copilot for meetings. You **hold a key to talk** (or
toggle hands-free auto-listen), it detects questions (even ones phrased as plain
statements), and **streams answers live** into an overlay that is hidden from
screen-capture. It can also **read a question off your screen** when you drag a
box around it, and scope every answer to a **context window** you define (e.g.
"Python/Flask senior backend interview").
Everything runs locally: **faster-whisper** (`distil-large-v3` by default) for
speech-to-text and a quantized **Qwen2.5-VL-7B** vision model (via `llama.cpp`)
that answers both spoken and on-screen questions. No audio or text leaves your machine.
Everything runs locally: **faster-whisper** (`base.en` by default, fully offline)
for speech-to-text and a quantized **Qwen2.5-VL-7B** vision model (via
`llama.cpp`) that answers both spoken and on-screen questions. No audio or text
leaves your machine — unless you explicitly enable the optional web search.
## What it does
- **Push-to-talk — no noise.** Audio is only captured while you **hold Right
Option (⌥)**; nothing is transcribed otherwise. Captures your microphone *and*
the other participants' audio (system output via a loopback device), tagged by
speaker (`You` / `Them`).
- **Push-to-talk or auto-listen.** Hold **Right Option (⌥)** to capture; release
to answer. Or press **Ctrl+Shift+M** to toggle hands-free auto-listen
(continuous VAD). Captures your microphone *and* the other participants' audio
(system output via a loopback device), tagged by speaker (`You` / `Them`).
- **Streams answers as they generate.** First words appear in the overlay at
first-token time (~1–3 s) with a live `▌` cursor, instead of waiting for the
full answer. The console prints latency metrics (`⏱ whisper`, `⏱ gen`,
`⚡ first words`) so you can see where time goes.
- **Context windows keep answers on-topic.** Define a scope in
`contexts/*.yaml` (e.g. a Flask interview). Every question is interpreted
inside that scope first — ask "what is STOMP" and you get the Flask-relevant
meaning first, then other variations. See [contexts/README.md](contexts/README.md).
- **Reads on-screen questions.** Press **Ctrl+Shift+Space**, drag a box over a
question (e.g. on a shared slide), and the vision model reads and answers it —
including multiple-choice, code, and math.
@@ -25,16 +36,32 @@ that answers both spoken and on-screen questions. No audio or text leaves your m
- **Auto-fills the obvious.** Arithmetic is solved instantly (no LLM); short
factual lookups get a direct answer. Open-ended questions get a concise
**suggested reply** you can read out, using the live meeting transcript as context.
- **Movable, resizable overlay.** Drag the middle of the overlay to move it;
drag any edge or corner to resize. Press **Ctrl+Shift+H** to show/hide, and
**Ctrl+Shift+S** for a meeting summary so far.
- **Optional web search (off by default).** Ground factual answers in current
information via DuckDuckGo — see Configuration. Enabling it sends the question
text off-device.
## Hearing other participants (one-time macOS setup)
Your microphone only captures *you*. To also capture what the other participants
say, the assistant reads your system audio output through a virtual loopback
device. **BlackHole** is already detected on this machine.
device. Devices are auto-detected in priority order
(`audio.loopback_keywords`): **BlackHole / Soundflower / VB-Cable / Loopback**,
then Microsoft Teams' own virtual device as a last resort.
The catch: if you send audio *only* to BlackHole, you won't hear it yourself. So
create a **Multi-Output Device** that plays to both your speakers/headphones and
BlackHole at once:
- **BlackHole (recommended — works for Google Meet, Zoom, Teams, anything):**
it mirrors *all* system audio. Install with `brew install blackhole-2ch`
(needs your admin password; reboot or restart `coreaudiod` afterwards).
- **"Microsoft Teams Audio" (zero-install experiment, Teams-only):** Teams
installs this device itself and the app will use it if no true loopback
exists — but it only ever carries Teams audio, and may be silent outside
screen-share. It will never help for Google Meet or Zoom.
The catch with BlackHole: if you send audio *only* to BlackHole, you won't hear
it yourself. So create a **Multi-Output Device** that plays to both your
speakers/headphones and BlackHole at once:
1. Open **Audio MIDI Setup** (Applications → Utilities).
2. Click **+** (bottom-left) → **Create Multi-Output Device**.
@@ -46,21 +73,29 @@ BlackHole at once:
Now meeting audio reaches both your ears and the assistant. Your microphone stays
selected as the meeting's *input*.
> Don't have BlackHole? Install with `brew install blackhole-2ch`, then re-run.
## Configuration
## Configuration (`config.yaml`)
`config.yaml` is gitignored (it may hold machine-specific/private settings) —
copy `config.example.yaml` to `config.yaml` and edit. Key options:
```yaml
audio:
source: "both" # "microphone", "system", or "both"
capture_mode: "push_to_talk" # push_to_talk (hold key) or continuous
capture_mode: "push_to_talk" # push_to_talk (hold key) or continuous (auto-listen)
ptt_key: "alt_r" # push-to-talk key (Right Option). e.g. cmd_r, f8, ctrl_r
whisper_model: "distil-large-v3" # or large-v3, medium.en, or a local dir
# base.en is ~9x faster than medium.en on CPU (~0.3s vs ~2.9s per clip) and is
# the main latency fix. Use models/whisper-medium.en for noisy audio/accents.
whisper_model: models/whisper-base.en
mic_device: null # null = auto. Or a device index / name substring.
system_device: null # null = auto-detect loopback (BlackHole). Or index / name.
system_device: null # null = auto-detect loopback. Or index / name.
loopback_keywords: ["blackhole", "soundflower", "vb-cable", "loopback", "teams audio"]
answer_sources: # which speakers trigger an answer
- "system" # other participants
- "microphone" # your own voice (handy for testing)
# Incremental transcription while the key is held. Leave off with fast models:
# faster-whisper pads every call to a fixed 30s window, so it only pays off
# with large/slow models — and can drop words at chunk seams.
streaming_transcription: false
screen:
capture_key: "ctrl+shift+space" # press, then drag a box over a question
@@ -70,32 +105,50 @@ ai:
mmproj: "mmproj-Qwen2.5-VL-7B-Instruct-f16.gguf" # required for screen reading
answer_mode: "auto_obvious" # auto_obvious | auto_all | suggest_only
user_name: "you"
# active_context: python_flask_backend # pin a context file by name (optional)
# PRIVACY: web search sends the question text to DuckDuckGo. Off by default.
web_search:
enabled: false
max_results: 3
timeout: 6.0
```
> **Tight on 16 GB RAM?** The vision model + `distil-large-v3` fit, but if memory
> gets tight set `whisper_model: medium.en`.
- `auto_obvious` — answer obvious questions (math/factual) directly; show a
*suggested reply* for open-ended ones.
- `auto_all` — generate a full answer for every detected question.
- `suggest_only` — never commit; always show a draft.
### Context windows (`contexts/`)
Each `.yaml` file defines a named scope the AI reads before answering, so it
stays on-topic without you restating the context in every question. Mark one
`active: true` (or set `ai.active_context`). Files hot-reload — edit mid-meeting
and it takes effect on the next question. `strict: true` disables the
"…then note other variations" behavior. A complete example ships in
[`contexts/python_flask_backend.yaml`](contexts/python_flask_backend.yaml).
## Run
```bash
./run.sh # macOS / Linux
./run.sh # macOS / Linux (handles venv, CA bundle, PortAudio)
# or
python main.py
```
First run downloads the models (run `python setup.py` once, ~6 GB for the vision
model; `distil-large-v3` auto-downloads on first launch).
model). The Whisper models load from local `models/whisper-*` directories, fully
offline — no Hugging Face access needed at runtime (works behind SSL-inspecting
corporate proxies).
On first launch grant three macOS permissions (System Settings → Privacy & Security):
On first launch grant three macOS permissions (System Settings → Privacy &
Security) **to the app you launch from** (Terminal, iTerm, or PyCharm):
- **Microphone** — to capture audio.
- **Screen Recording** — so the overlay can hide *itself* from capture, and so the
draw-a-box screen grab works.
- **Accessibility** — so the global push-to-talk and screen-grab hotkeys are seen.
- **Screen & System Audio Recording** — without it, screen grabs silently
capture only your wallpaper (macOS returns a windowless desktop).
- **Accessibility** — without it, the global hotkeys (push-to-talk, screen grab)
never fire; the log shows `This process is not trusted!`.
Quit and relaunch after granting — macOS applies these on restart.
You can also type a question in the terminal + Enter to test the AI directly.
@@ -103,11 +156,12 @@ You can also type a question in the terminal + Enter to test the AI directly.
| File | Role |
|------|------|
| `src/audio_listener.py` | Captures each source. Push-to-talk buffers audio only while armed; transcribes the held clip with one shared Whisper model, tags by source. |
| `src/hotkeys.py` | Global hotkeys (pynput): hold-to-talk + screen-grab chord. |
| `main.py` | Question detection + orchestration. |
| `src/ai_engine.py` | Math fast-path, classification, and Qwen2.5-VL answering for both spoken and on-screen (`answer_from_image`) questions. |
| `src/region_capture.py` | Draw-a-box fullscreen selector + region screenshot. |
| `src/audio_listener.py` | Captures each source. Push-to-talk buffers audio only while armed; transcribes with one shared Whisper model (serialized — not thread-safe), tags by source. Priority-ordered loopback detection. Optional incremental (streaming) transcription for slow models. |
| `src/hotkeys.py` | Global hotkeys (pynput): hold-to-talk, screen-grab chord, overlay toggle, summary, auto-listen toggle. |
| `main.py` | Question detection + orchestration; wires streamed partial answers into the overlay with latency metrics. |
| `src/ai_engine.py` | Math fast-path, classification, context-window injection, optional web search, and Qwen2.5-VL answering (streamed) for spoken and on-screen (`answer_from_image`) questions. Interrupts only in-flight generations so rapid follow-ups are never dropped. |
| `src/context_library.py` | Loads `contexts/*.yaml`, picks the active context, hot-reloads on file change. |
| `src/web_search.py` | Optional dependency-free DuckDuckGo search (Instant Answer + HTML results), proxy-tolerant. |
| `src/region_capture.py` | Draw-a-box fullscreen selector + Retina-correct region screenshot (freeze-frame at hotkey time). |
| `src/context_manager.py` | Rolling speaker-labeled meeting transcript. |
| `src/overlay.py` | Always-on-top overlay, hidden from screen capture. |
```
| `src/overlay.py` | Always-on-top overlay, hidden from screen capture. Drag interior to move, edges/corners to resize. |

65
config.example.yaml Normal file
View File

@@ -0,0 +1,65 @@
# Meeting Assistant configuration — copy to config.yaml and edit.
# (config.yaml is gitignored; it may hold machine-specific/private settings.)
audio:
sample_rate: 16000
chunk_duration: 2.0
vad_threshold: 0.5
# Transcription model (local dirs load fully offline — no Hugging Face access
# at runtime). base.en is ~9x faster than medium.en on CPU (~0.3s vs ~2.9s per
# clip) and is the main latency fix. Switch to models/whisper-medium.en for
# more robustness to noise/accents (slower).
whisper_model: models/whisper-base.en
language: en
# "push_to_talk" (hold the key, release to answer) or "continuous" (auto-listen).
# Toggle at runtime with ctrl+shift+m.
capture_mode: push_to_talk
ptt_key: alt_r # push-to-talk key (Right Option). e.g. cmd_r, f8, ctrl_r
source: both # "microphone", "system", or "both"
mic_device: null # null = auto. Or a device index / name substring.
system_device: null # null = auto-detect loopback. Or index / name.
# Devices to try (in priority order) as the "system" source that hears OTHER
# participants. A true loopback (BlackHole/Soundflower/VB-Cable) carries ALL
# system audio — works for Google Meet, Zoom, Teams, anything. "teams audio"
# is Microsoft Teams' own virtual device: zero-install but Teams-only, and it
# may be silent outside screen-share; treat it as an experiment.
loopback_keywords: ["blackhole", "soundflower", "vb-cable", "loopback", "teams audio"]
# Which speakers trigger an answer (others still add transcript context).
answer_sources: ["system", "microphone"]
# Incremental transcription while the push-to-talk key is held. Off by default:
# faster-whisper pads every call to a fixed 30s window, so on small/fast models
# this barely helps and can drop words at chunk seams. Enable only with a
# large/slow model where the per-call cost is worth amortizing during the hold.
streaming_transcription: false
# Bias Whisper toward your meeting's vocabulary (optional):
# transcription_prompt: "A software engineering interview about Python and Flask."
screen:
capture_key: ctrl+shift+space # press, then drag a box over an on-screen question
hotkeys:
toggle_overlay: ctrl+shift+h
meeting_summary: ctrl+shift+s
toggle_listening: ctrl+shift+m
ai:
context_window: 2048
max_history: 10
model: Qwen2.5-VL-7B-Instruct-Q4_K_M.gguf
mmproj: mmproj-Qwen2.5-VL-7B-Instruct-f16.gguf # required for screen reading
answer_mode: auto_obvious # auto_obvious | auto_all | suggest_only
user_name: you
# Pin the active context window by file name (contexts/<name>.yaml). If unset,
# the first contexts/*.yaml with `active: true` is used.
# active_context: python_flask_backend
# Optional web search to ground factual answers in current information.
# PRIVACY: when enabled, the question text is sent to DuckDuckGo — this
# breaks the otherwise fully on-device guarantee. Off by default.
web_search:
enabled: false
max_results: 3
timeout: 6.0
overlay:
position: top-right
font_size: 14

34
contexts/README.md Normal file
View File

@@ -0,0 +1,34 @@
# Context windows
Each `.yaml` file here defines a **context window** — a well-defined scope the AI
reads *before* answering, so it stays on topic instead of drifting. The active
context is prepended to every question (spoken and on-screen) before it goes to
the model.
Example: with `python_flask_backend.yaml` active, asking *"what is STOMP"* gets
the Python/Flask-relevant answer first (the WebSocket sub-protocol you'd use from
a Flask backend), then a short note on other meanings — rather than a generic or
off-topic answer.
## Format
```yaml
name: Python / Flask — Senior Backend Engineer # display name
scope: Senior backend engineering interview … # one-line summary (optional)
active: true # mark exactly one file active
strict: false # false → answer in-context first, then variations
# true → stay strictly in-scope, no variations
definition: | # required: the scope + how to interpret questions
Free-form text describing the context…
```
## Choosing the active context
1. If `config.yaml` sets `ai.active_context: <filename-without-extension>`, that
file wins (e.g. `ai.active_context: python_flask_backend`).
2. Otherwise, the first file with `active: true` is used.
3. If none match, the assistant answers without a context (default behavior).
Files are reloaded when they change on disk, so editing a context takes effect
without restarting the app. To switch contexts, set `active: true` on one file
(and `false` on the others), or set `ai.active_context` in `config.yaml`.

31
contexts/python_flask_backend.yaml Normal file
View File

@@ -0,0 +1,31 @@
name: Python / Flask — Senior Backend Engineer
scope: Senior backend engineering interview focused on Python and the Flask web framework
active: true
strict: false
definition: |
This is a senior backend engineering interview centered on Python and the
Flask web framework. Interpret every question within Python/Flask backend
engineering first, even when the question does not say so explicitly, and
answer at the depth expected of a senior engineer.
When a term also has meanings outside this scope (e.g. STOMP, CORS, WSGI,
"workers", "streams"), give the Python/Flask-relevant answer first, then
briefly note other necessary variations so the answer stays accurate.
Assume strong familiarity with and expect depth on:
- Flask app structure: application factory, blueprints, extensions, config.
- Request lifecycle: WSGI, the app/request contexts, g, before/after request.
- Concurrency & deployment: the GIL, gunicorn/uwsgi workers, gevent/eventlet,
threads vs processes, async (ASGI/Quart) trade-offs.
- Data: SQLAlchemy ORM and Core, sessions, migrations (Alembic), connection
pooling, N+1 queries, transactions.
- APIs: RESTful design, status codes, pagination, versioning, serialization
(marshmallow/pydantic), input validation, error handling.
- Auth & security: sessions vs JWT, CSRF, CORS, OWASP basics, secrets, rate
limiting.
- Reliability & performance: caching (Redis), background jobs (Celery/RQ),
idempotency, observability (logging, metrics, tracing), profiling.
- Testing & quality: pytest, fixtures, test client, mocking, coverage, CI.
Keep answers interview-appropriate: precise, technically correct, and concise,
with concrete Python/Flask examples or trade-offs where helpful.

27
main.py
View File

@@ -5,6 +5,7 @@ Meeting Assistant - Real-time AI Copilot for Meetings
import sys
import os
import time
import yaml
import threading
import logging
@@ -255,16 +256,29 @@ class MeetingAssistant:
try:
context = self.context_manager.get_context()
result = self.ai_engine.answer_question(question, context, source)
# Stream the answer into the overlay as it generates: first words
# appear at first-token time instead of after the full generation.
asked_by = speaker or ("You" if source == "microphone" else "Them")
shown_q = f"({asked_by}) {question}"
suggested_early = self.ai_engine.is_suggested(question)
t_start = time.time()
first_shown = [False]
def on_partial(text_so_far):
if not first_shown[0]:
first_shown[0] = True
print(f" ⚡ first words on screen after {time.time() - t_start:.1f}s")
self.overlay.show_answer(text_so_far + " ▌", shown_q, suggested_early)
result = self.ai_engine.answer_question(
question, context, source, on_partial=on_partial)
if not result or not result.get("text"):
return # interrupted or empty
answer = result["text"]
suggested = result.get("suggested", False)
asked_by = speaker or ("You" if source == "microphone" else "Them")
shown_q = f"({asked_by}) {question}"
self.overlay.show_answer(answer, shown_q, suggested)
self.logger.info(f"Q [{asked_by}]: {question}")
self.logger.info(f"A: {answer}")
@@ -305,7 +319,10 @@ class MeetingAssistant:
self.ai_engine.interrupt()
self.answering = True
try:
answer = self.ai_engine.answer_from_image(image_path)
answer = self.ai_engine.answer_from_image(
image_path,
on_partial=lambda t: self.overlay.show_answer(
t + " ▌", "Screen question", suggested=False))
if not answer:
return
self.overlay.show_answer(answer, "Screen question", suggested=False)

7
run.sh
View File

@@ -25,11 +25,14 @@ else
echo " Run: python3 -m venv .venv && source .venv/bin/activate && pip install -r requirements.txt"
fi
# Install PortAudio if needed (macOS)
# Install PortAudio if needed (macOS). Skip Homebrew's auto-update (slow/can fail
# behind restricted networks) and never let a brew hiccup abort the launcher —
# PortAudio is usually already present.
if [[ "$OSTYPE" == "darwin"* ]]; then
export HOMEBREW_NO_AUTO_UPDATE=1
if command -v brew &> /dev/null && ! brew list portaudio &> /dev/null 2>&1; then
echo "📦 Installing PortAudio via Homebrew..."
brew install portaudio
brew install portaudio || echo "⚠️ PortAudio install failed — continuing (may already be present)"
fi
fi

161
src/ai_engine.py
View File

@@ -1,6 +1,7 @@
import os
import re
import ast
import time
import operator
from pathlib import Path
from collections import deque
@@ -73,13 +74,35 @@ class AIEngine:
self.temperature = float(ai_cfg.get("temperature", 0.3))
self.n_ctx = int(ai_cfg.get("context_length", 4096))
# File-based context windows: read before each answer so the model
# stays scoped to the active context (e.g. a Flask interview).
from src.context_library import ContextLibrary
contexts_dir = Path(__file__).parent.parent / ai_cfg.get("contexts_dir", "contexts")
self.contexts = ContextLibrary(contexts_dir, ai_cfg.get("active_context"))
# Optional DuckDuckGo web search (off by default). When on, factual/open
# questions are grounded on fresh search snippets. PRIVACY: this sends
# the query off-device — see ai.web_search in config.yaml.
ws_cfg = ai_cfg.get("web_search", {}) if isinstance(ai_cfg.get("web_search"), dict) else {}
self.web_search_enabled = bool(ws_cfg.get("enabled", False))
self.web_search_max = int(ws_cfg.get("max_results", 3))
self.web_search_timeout = float(ws_cfg.get("timeout", 6.0))
self.model = None
self.vision = False # True once a vision (VL) model is loaded
self.memory = deque(maxlen=8)
self.interrupt_event = threading.Event()
# llama.cpp is not thread-safe; serialize all generations through one lock
self._gen_lock = threading.Lock()
# True only while a generation is inside the stream loop; interrupt()
# is a no-op otherwise so a stale flag can't kill the *next* question.
self._generating = False
self.load_model()
active = self.contexts.active()
if active:
print(f"🎯 Active context: {active.name}")
if self.web_search_enabled:
print("🌐 Web search: ON (DuckDuckGo) — queries leave this device.")
def load_model(self):
"""Load Qwen2.5-VL (vision) when an mmproj is configured, else a plain
@@ -141,8 +164,20 @@ class AIEngine:
return "factual"
return "open"
def answer_question(self, question, context, source=None):
"""Returns dict: {text, kind, suggested}. text is None if interrupted."""
def is_suggested(self, question):
"""Would this question produce a suggested reply (vs committed answer)?
Exposed so the UI can label a streaming partial correctly from token one."""
if self.answer_mode == "suggest_only":
return True
if self.answer_mode == "auto_all":
return False
return self.classify(question) == "open"
def answer_question(self, question, context, source=None, on_partial=None):
"""Returns dict: {text, kind, suggested}. text is None if interrupted.
on_partial(text_so_far) is called (throttled) as tokens stream in, so
the overlay can show the answer while it is still being generated."""
if self.interrupt_event.is_set():
self.interrupt_event.clear()
return None
@@ -162,7 +197,7 @@ class AIEngine:
if kind == "math":
text = try_solve_math(question)
else:
text = self._generate(question, context, kind, suggested)
text = self._generate(question, context, kind, suggested, on_partial)
if text is None:
return None
@@ -171,7 +206,11 @@ class AIEngine:
return {"text": text, "kind": kind, "suggested": suggested}
def interrupt(self):
self.interrupt_event.set()
# Only interrupt an in-flight generation. Setting the flag while idle
# would make the NEXT answer consume it at entry and silently die —
# dropping a rapid follow-up question instead of the stale answer.
if self._generating:
self.interrupt_event.set()
# ------------------------------------------------------------------
# Internals
@@ -180,7 +219,32 @@ class AIEngine:
def _normalize_question(self, question):
return question.strip()
_CONTEXT_LIMIT = 1800 # chars of context definition to include in the prompt
def _context_preamble(self):
"""Front-load the active context window so the model answers within its
scope. Read fresh each time, so editing a context file takes effect live."""
ctx = self.contexts.active() if getattr(self, "contexts", None) else None
if not ctx:
return ""
body = ctx.definition[: self._CONTEXT_LIMIT]
if ctx.strict:
rule = ("Answer strictly within this context and scope. Do not drift "
"outside it.")
else:
rule = ("Interpret every question within this context first, even when "
"the question does not say so. Give the in-context answer first, "
"then briefly note other necessary meanings or variations.")
scope = f" (scope: {ctx.scope})" if ctx.scope else ""
return (
f"ACTIVE CONTEXT — \"{ctx.name}\"{scope}.\n{rule}\n"
f"[Context definition]\n{body}\n\n"
)
def _system_prompt(self, kind, suggested):
return self._context_preamble() + self._base_system_prompt(kind, suggested)
def _base_system_prompt(self, kind, suggested):
if suggested:
return (
f"You are a real-time meeting copilot for {self.user_name}. "
@@ -202,18 +266,25 @@ class AIEngine:
"current question accurately and concisely (under ~70 words). Be direct."
)
def _user_prompt(self, question, context):
def _user_prompt(self, question, context, web_block=""):
# Keep the prompt lean: prefill time scales with prompt length, and on
# CPU/Metal it directly delays the first visible token. ~1200 chars of
# transcript + 3 truncated memory turns keeps follow-ups working while
# staying fast.
transcript = ""
if context:
transcript = (context.get("audio", "") or "")[-1800:].strip()
transcript = (context.get("audio", "") or "")[-1200:].strip()
# Recent Q&A thread so follow-ups resolve ("state its 4 core principles"
# after "what is Java" → the model sees the Java exchange).
memory_block = ""
if self.memory:
turns = []
for item in list(self.memory)[-5:]:
turns.append(f"Q: {item['question']}\nA: {item['response']}")
for item in list(self.memory)[-3:]:
resp = item['response']
if len(resp) > 240:
resp = resp[:240] + "…"
turns.append(f"Q: {item['question']}\nA: {resp}")
memory_block = "\n".join(turns)
parts = []
@@ -225,18 +296,52 @@ class AIEngine:
"\"those\", \"the second one\" against these earlier turns]\n"
f"{memory_block}\n"
)
if web_block:
parts.append(
"[Web search results — use these to ground your answer in current, "
"factual information; cite specifics, ignore anything irrelevant]\n"
f"{web_block}\n"
)
parts.append(f"[Current question]\n{question}")
return "\n".join(parts)
def _stream_chat(self, messages, max_tokens, multiline=False):
def _maybe_web_search(self, question, kind, suggested):
"""Best-effort DuckDuckGo lookup for committed factual/open answers.
Returns a formatted block (or "") — never raises, never blocks long."""
if not self.web_search_enabled or suggested or kind == "math":
return ""
try:
from src import web_search
results = web_search.search(
question, max_results=self.web_search_max,
timeout=self.web_search_timeout)
if results:
print(f"🌐 Web search: {len(results)} result(s) for '{question}'")
return web_search.format_for_prompt(results)
except Exception as e:
print(f"⚠️ Web search failed: {e}")
return ""
# Throttle partial-answer UI updates: often enough to feel live, rare
# enough not to flood the Qt signal queue.
_PARTIAL_INTERVAL = 0.25
def _stream_chat(self, messages, max_tokens, multiline=False, on_partial=None):
"""Run a chat completion, streaming so a new question can interrupt the
current one mid-generation. Returns cleaned text, or None if interrupted."""
current one mid-generation. Returns cleaned text, or None if interrupted.
When on_partial is set, it receives the accumulated text every
_PARTIAL_INTERVAL seconds so the UI can render the answer as it forms."""
if self.interrupt_event.is_set():
self.interrupt_event.clear()
return None
chunks = []
t0 = time.time()
first_tok = None
last_emit = 0.0
with self._gen_lock:
self._generating = True
try:
stream = self.model.create_chat_completion(
messages=messages,
@@ -253,18 +358,40 @@ class AIEngine:
delta = part["choices"][0].get("delta", {})
piece = delta.get("content")
if piece:
if first_tok is None:
first_tok = time.time() - t0
chunks.append(piece)
now = time.time()
if on_partial and now - last_emit >= self._PARTIAL_INTERVAL:
last_emit = now
try:
on_partial("".join(chunks))
except Exception:
pass # UI hiccups must never kill generation
except Exception as e:
print(f"⚠️ Generation error: {e}")
return "I couldn't process that."
finally:
self._generating = False
# An interrupt raised during this generation was aimed at this
# generation — never leave it armed for the next question.
self.interrupt_event.clear()
total = time.time() - t0
if first_tok is not None:
print(f" ⏱ gen: first token {first_tok:.1f}s · total {total:.1f}s "
f"· {len(''.join(chunks))} chars")
return self._clean("".join(chunks), multiline=multiline)
def _generate(self, question, context, kind, suggested):
def _generate(self, question, context, kind, suggested, on_partial=None):
web_block = self._maybe_web_search(question, kind, suggested)
messages = [
{"role": "system", "content": self._system_prompt(kind, suggested)},
{"role": "user", "content": self._user_prompt(question, context)},
{"role": "user", "content": self._user_prompt(question, context, web_block)},
]
# Web-grounded answers benefit from a bit more room than a bare factual.
if web_block and kind == "factual":
kind = "open" # use the longer factual->open budget below
# Spoken answers are conversational — keep them short so they come back
# fast. The prompts already target ~60-70 words; cap tokens to match
# rather than spend seconds generating up to self.max_tokens (~350).
@@ -274,13 +401,13 @@ class AIEngine:
max_tokens = min(self.max_tokens, 110)
else:
max_tokens = min(self.max_tokens, 160)
return self._stream_chat(messages, max_tokens)
return self._stream_chat(messages, max_tokens, on_partial=on_partial)
# ------------------------------------------------------------------
# Screen questions — read a screenshot and answer it directly
# ------------------------------------------------------------------
def answer_from_image(self, image_path, hint=None):
def answer_from_image(self, image_path, hint=None, on_partial=None):
"""Read the question(s) in a screenshot and answer. Returns the answer
text, or None if interrupted / unavailable."""
if not self.vision:
@@ -307,14 +434,16 @@ class AIEngine:
messages = [
{"role": "system",
"content": f"You are an expert assistant helping {self.user_name} answer "
"content": self._context_preamble()
+ f"You are an expert assistant helping {self.user_name} answer "
"a question shown on screen during a meeting."},
{"role": "user", "content": [
{"type": "text", "text": instruction},
{"type": "image_url", "image_url": {"url": data_uri}},
]},
]
text = self._stream_chat(messages, max_tokens=self.max_tokens)
text = self._stream_chat(messages, max_tokens=self.max_tokens,
on_partial=on_partial)
if text:
self.memory.append({"question": "[screen question]", "response": text,
"time": datetime.now()})

246
src/audio_listener.py
View File

@@ -176,6 +176,71 @@ class StreamSegmenter:
self._pre_frames = 0
class _StreamingTranscriber:
"""Incrementally transcribes a growing audio clip, committing words as they
fall safely behind the live edge.
Each tick re-transcribes only the uncommitted tail (plus a little left
context for accuracy) and commits new words whose end is older than
`safety` seconds from the current end — words right at the edge may still
change as more audio arrives, so they're held back. On `finalize()` the
remaining tail is committed with no safety margin. This bounds the work
done at key-release to a short tail rather than the whole clip.
"""
def __init__(self, transcribe_fn, sample_rate, safety=0.6, context=2.0, eps=0.15):
self._tx = transcribe_fn # (audio_np, word_timestamps) -> [segments]
self.sr = sample_rate
self.safety = safety
self.context = context
self.eps = eps
self.committed_words = [] # committed word strings, in order
self.committed_time = 0.0 # abs end-time (s) of last committed word
def _words_from(self, audio_np):
"""Transcribe from just before the commit point to the end; return the
context start-time and the word list (times relative to that start)."""
ctx_start = max(0.0, self.committed_time - self.context)
cs = int(ctx_start * self.sr)
chunk = audio_np[cs:]
if chunk.size < int(0.2 * self.sr): # not enough new audio to bother
return ctx_start, []
words = []
for seg in self._tx(chunk, True):
for w in (seg.words or []):
words.append(w)
return ctx_start, words
def update(self, audio_np):
"""Commit any words now safely behind the live edge. Returns text so far."""
now_end = audio_np.size / self.sr
ctx_start, words = self._words_from(audio_np)
for w in words:
abs_end = ctx_start + w.end
# Skip words already covered: committed_time is the end-time of the
# last committed word, so anything ending at/before it is a re-read
# of committed audio (the left-context overlap).
if abs_end <= self.committed_time + self.eps:
continue
if abs_end <= now_end - self.safety:
self.committed_words.append(w.word)
self.committed_time = abs_end
else:
break # reached the volatile tail
return "".join(self.committed_words).strip()
def finalize(self, audio_np):
"""Commit the remaining tail (no safety margin) and return full text."""
ctx_start, words = self._words_from(audio_np)
for w in words:
abs_end = ctx_start + w.end
if abs_end <= self.committed_time + self.eps:
continue
self.committed_words.append(w.word)
self.committed_time = abs_end
return "".join(self.committed_words).strip()
class AudioListener:
"""Captures one or more audio sources, segments speech per source, and
transcribes finished utterances with a single shared Whisper model.
@@ -241,11 +306,35 @@ class AudioListener:
self.streams = [] # open sd.RawInputStream objects
self.last_emit = {} # source -> last transcript text (de-dup)
# Whisper model is config-driven. Default to distil-large-v3: near
# large-v3 accuracy but much faster — and push-to-talk makes the
# transcription latency a non-issue. Auto-downloads/caches by name
# (honors SSL_CERT_FILE / HF_HUB_DISABLE_XET set by run.sh).
whisper_name = audio_cfg.get("whisper_model", "distil-large-v3")
# llama-style streaming transcription for push-to-talk: while the key is
# held, transcribe the growing clip incrementally and commit words that
# are safely behind the live edge, so on release only the short tail is
# left to transcribe (instead of the whole clip). Big latency win on
# long questions. faster-whisper isn't safe to call concurrently, so a
# single lock serializes every transcribe() call.
self._model_lock = threading.Lock()
# Streaming (incremental) transcription. Profiling showed faster-whisper
# pads every call to a fixed 30s encoder window, so a short tail costs
# almost as much as the whole clip — streaming barely helps on small
# models and can drop words at chunk seams. The real latency win is a
# fast model (base.en ≈ 0.3s vs medium.en ≈ 2.9s per clip). So this is
# OFF by default; turn it on only if you run a large/slow model.
self._streaming = bool(audio_cfg.get("streaming_transcription", False))
self._stream_interval = float(audio_cfg.get("streaming_interval", 1.0))
self._stream_safety = float(audio_cfg.get("streaming_safety", 0.6))
# Small left-context keeps per-tick cost low so ticks keep up with
# realtime and the commit point stays close to the live edge — which is
# what makes the post-release tail (and latency) small.
self._stream_context = float(audio_cfg.get("streaming_context", 1.0))
self._stream_transcribers = {} # source -> _StreamingTranscriber
self._stream_stop = None
self._stream_thread = None
# Whisper model is config-driven. Default to the local base.en dir:
# ~9x faster than medium.en on CPU (~0.3s vs ~2.9s per clip) with strong
# accuracy on clear speech — this is the main transcription-latency fix.
# Swap to medium.en (also local) for more robustness to noise/accents.
whisper_name = audio_cfg.get("whisper_model", "models/whisper-base.en")
cpu_threads = max(os.cpu_count() - 1, 1)
# A light domain prompt biases Whisper toward plausible vocabulary so
# short conversational clips garble less ("SOLID principles", not "solid
@@ -258,9 +347,12 @@ class AudioListener:
)
print(f"🔄 Loading Whisper model '{whisper_name}'...")
self.model = None
for attempt in (whisper_name, "models/whisper", "base"):
for attempt in (whisper_name, "models/whisper-base.en",
"models/whisper-medium.en", "models/whisper", "base"):
try:
local_only = attempt == "models/whisper"
# Any local directory loads strictly offline (no HF lookup, which
# an SSL-inspecting proxy blocks); bare names may download.
local_only = os.path.isdir(attempt)
self.model = WhisperModel(
attempt,
device="cpu",
@@ -319,9 +411,14 @@ class AudioListener:
resolved = self._resolve_device(cfg)
if resolved is not None:
return resolved
for i, dev in enumerate(sd.query_devices()):
if dev["max_input_channels"] > 0 and self._is_loopback(dev["name"]):
return i
# Keywords are in priority order: a true system-wide loopback (BlackHole)
# beats app-specific virtual devices (e.g. "Microsoft Teams Audio", which
# only carries Teams call audio — and only sometimes).
devices = sd.query_devices()
for keyword in self.loopback_keywords:
for i, dev in enumerate(devices):
if dev["max_input_channels"] > 0 and keyword in dev["name"].lower():
return i
return None
def _planned_sources(self):
@@ -412,6 +509,21 @@ class AudioListener:
for src, frames in self._preroll.items()
}
self._armed = True
# Begin incremental transcription so the clip is mostly transcribed by
# the time the key is released.
if self._streaming:
self._stream_transcribers = {
src: _StreamingTranscriber(
self._stream_transcribe_fn, self.sample_rate,
safety=self._stream_safety, context=self._stream_context)
for src in self.segmenters
}
self._stream_stop = threading.Event()
self._stream_thread = threading.Thread(
target=self._streaming_worker, daemon=True)
self._stream_thread.start()
print("\n🎙️ Listening (key held)...")
def disarm(self):
@@ -419,6 +531,19 @@ class AudioListener:
if self._auto_listen or not self._armed:
return
self._armed = False
# Stop the streaming worker (if any) and grab the per-source transcribers
# so we can finalize each clip's short remaining tail.
transcribers = {}
if self._streaming and self._stream_stop is not None:
self._stream_stop.set()
if self._stream_thread is not None:
self._stream_thread.join(timeout=2.0)
transcribers = self._stream_transcribers
self._stream_transcribers = {}
self._stream_stop = None
self._stream_thread = None
with self._ptt_lock:
buffers = self._ptt_buffers
self._ptt_buffers = {}
@@ -436,7 +561,9 @@ class AudioListener:
print(" Set your meeting app's Speaker to a Multi-Output "
"Device that includes BlackHole (README → 'Hearing other "
"participants').")
self.utterance_queue.put((source, audio))
# Carry the streaming transcriber so the loop finalizes the tail
# instead of re-transcribing the whole clip.
self.utterance_queue.put((source, audio, transcribers.get(source)))
queued = True
print("⏳ Transcribing..." if queued else " (too short — nothing captured)")
@@ -449,6 +576,8 @@ class AudioListener:
if self._auto_listen:
# entering auto-listen: drop any half-held PTT capture
self._armed = False
if self._stream_stop is not None:
self._stream_stop.set() # stop any in-flight streaming worker
with self._ptt_lock:
self._ptt_buffers = {}
print("\n🔊 Auto-listen ON — listening on its own (no key needed).")
@@ -501,6 +630,35 @@ class AudioListener:
pr.append(audio_bytes)
return _cb
# ------------------------------------------------------------------
# Streaming transcription (push-to-talk)
# ------------------------------------------------------------------
def _stream_transcribe_fn(self, audio_np, word_timestamps):
"""Run Whisper for the streamer. vad_filter is off here so word
timestamps stay aligned to the raw clip."""
return self._run_whisper(audio_np, word_timestamps=word_timestamps,
vad_filter=False)
def _streaming_worker(self):
"""While the key is held, periodically transcribe each source's growing
clip and commit stable words. Runs until disarm/mode-switch stops it."""
stop = self._stream_stop
while stop is not None and not stop.wait(self._stream_interval):
if not self._armed:
break
with self._ptt_lock:
snaps = {src: bytes(buf) for src, buf in self._ptt_buffers.items()}
for source, pcm in snaps.items():
tr = self._stream_transcribers.get(source)
if tr is None or len(pcm) < 2 * int(0.3 * self.sample_rate):
continue
audio_np = np.frombuffer(pcm, dtype=np.int16).astype(np.float32) / 32768.0
try:
tr.update(audio_np)
except Exception as e:
print(f"⚠️ streaming transcribe error [{source}]: {e}")
def _select_input_device(self):
# kept for backward compatibility
return self._find_mic_device() or 0
@@ -573,33 +731,63 @@ class AudioListener:
def _transcribe_loop(self):
while self.running:
try:
source, audio = self.utterance_queue.get(timeout=0.5)
item = self.utterance_queue.get(timeout=0.5)
except queue.Empty:
continue
source, audio = item[0], item[1]
transcriber = item[2] if len(item) > 2 else None
try:
self._transcribe(source, audio)
if transcriber is not None:
self._finalize_stream(source, audio, transcriber)
else:
self._transcribe(source, audio)
except Exception as e:
print(f"❌ Transcription error: {e}")
# On any streaming failure, fall back to a full one-shot pass.
if transcriber is not None:
try:
self._transcribe(source, audio)
except Exception as e2:
print(f"❌ Fallback transcription error: {e2}")
def _run_whisper(self, audio_np, word_timestamps=False, vad_filter=True):
"""Serialized Whisper call (faster-whisper isn't thread-safe). Returns a
list of segments."""
with self._model_lock:
segments, _info = self.model.transcribe(
audio_np,
language="en",
beam_size=1, # greedy: distil-whisper is tuned for it and
best_of=1, # it's markedly faster for conversational clips
temperature=0.0,
condition_on_previous_text=False,
initial_prompt=self._initial_prompt,
compression_ratio_threshold=1.8,
no_speech_threshold=0.6,
log_prob_threshold=-1.0,
word_timestamps=word_timestamps,
vad_filter=vad_filter,
)
return list(segments)
def _transcribe(self, source, audio):
audio_np = np.frombuffer(audio, dtype=np.int16).astype(np.float32) / 32768.0
segments, _info = self.model.transcribe(
audio_np,
language="en",
beam_size=1, # greedy: distil-whisper is tuned for it and
best_of=1, # it's markedly faster for conversational clips
temperature=0.0,
condition_on_previous_text=False,
initial_prompt=self._initial_prompt,
compression_ratio_threshold=1.8,
no_speech_threshold=0.6,
log_prob_threshold=-1.0,
word_timestamps=False,
vad_filter=True,
)
t0 = time.time()
segments = self._run_whisper(audio_np)
print(f" ⏱ whisper: {len(audio_np) / self.sample_rate:.1f}s clip "
f"in {time.time() - t0:.2f}s")
text = self._clean_text(" ".join(s.text.strip() for s in segments if s.text.strip()))
self._emit_transcript(source, audio_np, text)
def _finalize_stream(self, source, audio, transcriber):
"""Finish a push-to-talk clip that was transcribed incrementally: only
the short remaining tail still needs Whisper."""
audio_np = np.frombuffer(audio, dtype=np.int16).astype(np.float32) / 32768.0
text = self._clean_text(transcriber.finalize(audio_np))
self._emit_transcript(source, audio_np, text)
def _emit_transcript(self, source, audio_np, text):
"""Filter, de-dup, attribute a speaker, and deliver a transcript."""
if not text:
return
# drop Whisper's silence hallucinations ("Thank you.", "Bye.", "you")

116
src/context_library.py Normal file
View File

@@ -0,0 +1,116 @@
"""File-based context windows.
A "context window" is a well-defined scope (e.g. "Python / Flask — Senior
Backend Engineer interview") that the AI reads before answering, so it stays on
topic instead of drifting. Each context is one YAML file in the contexts/
folder; one is marked active. The library reloads files when they change on
disk, so editing a context takes effect without restarting the app.
YAML schema (all keys except `definition` are optional):
name: Python / Flask — Senior Backend Engineer
scope: Senior backend engineering interview focused on Python and Flask
active: true # this is the context to apply
strict: false # false → answer in-context first, then variations
definition: |
Free-form text describing the scope and how to interpret questions.
"""
import os
from pathlib import Path
import yaml
class ContextWindow:
def __init__(self, name, scope, definition, strict, active, path):
self.name = name
self.scope = scope
self.definition = definition
self.strict = strict
self.active = active
self.path = path
@property
def key(self):
"""Stable identifier used by config's ai.active_context (the filename)."""
return Path(self.path).stem
class ContextLibrary:
"""Loads context windows from a folder and exposes the active one.
`active_name` (from config ai.active_context) wins if set — it matches a
file stem or a context's `name`. Otherwise the first file with `active: true`
is used. Files are re-parsed only when their mtime changes.
"""
def __init__(self, contexts_dir, active_name=None):
self.dir = Path(contexts_dir)
self.active_name = (active_name or "").strip() or None
self._cache = {} # path -> (mtime, ContextWindow | None)
# ------------------------------------------------------------------
def _load_file(self, path):
try:
with open(path, "r") as f:
data = yaml.safe_load(f) or {}
if not isinstance(data, dict):
return None
definition = str(data.get("definition", "")).strip()
if not definition:
return None
return ContextWindow(
name=str(data.get("name", path.stem)).strip(),
scope=str(data.get("scope", "")).strip(),
definition=definition,
strict=bool(data.get("strict", False)),
active=bool(data.get("active", False)),
path=str(path),
)
except Exception as e:
print(f"⚠️ Could not parse context {path.name}: {e}")
return None
def _all(self):
"""Return the current set of ContextWindows, reparsing changed files."""
windows = []
if not self.dir.is_dir():
return windows
seen = set()
for path in sorted(self.dir.glob("*.yaml")) + sorted(self.dir.glob("*.yml")):
seen.add(str(path))
try:
mtime = path.stat().st_mtime
except OSError:
continue
cached = self._cache.get(str(path))
if cached is None or cached[0] != mtime:
window = self._load_file(path)
self._cache[str(path)] = (mtime, window)
else:
window = cached[1]
if window is not None:
windows.append(window)
# drop cache entries for files that no longer exist
for stale in [p for p in self._cache if p not in seen]:
del self._cache[stale]
return windows
def active(self):
"""The context to apply right now, or None."""
windows = self._all()
if not windows:
return None
if self.active_name:
for w in windows:
if self.active_name in (w.key, w.name):
return w
for w in windows:
if w.active:
return w
return None
def list(self):
return self._all()

179
src/overlay.py
View File

@@ -8,10 +8,182 @@ from PyQt5.QtWidgets import (
QVBoxLayout,
QLabel,
)
from PyQt5.QtCore import Qt, QTimer, QObject, pyqtSignal
from PyQt5.QtCore import Qt, QTimer, QObject, pyqtSignal, QRect, QEvent
from PyQt5.QtGui import QFont, QTextCursor, QColor
class _ResizeTextEdit(QTextEdit):
"""The overlay's text area, doubling as its drag-to-resize/move surface.
The window is frameless (no native title bar or resize border) AND
translucent, and on macOS a fully-transparent region passes clicks through —
so only this opaque widget can catch presses. Press near an edge/corner to
RESIZE; press anywhere else and drag to MOVE the window. macOS grabMouse()
doesn't deliver the drag stream for this window, so during a drag we watch
the mouse via an application-wide event filter instead.
"""
RESIZE_MARGIN = 8
def __init__(self):
super().__init__()
self.setMouseTracking(True)
self.viewport().setMouseTracking(True)
self._drag = None # None | "resize" | "move"
self._edges = (False, False, False, False) # left, top, right, bottom
self._start_geo = None
self._start_mouse = None
def _edges_at(self, pos):
m = self.RESIZE_MARGIN
vp = self.viewport()
w, h = vp.width(), vp.height()
return (pos.x() <= m, pos.y() <= m, pos.x() >= w - m, pos.y() >= h - m)
@staticmethod
def _cursor_for(edges):
left, top, right, bottom = edges
if (left and top) or (right and bottom):
return Qt.SizeFDiagCursor
if (right and top) or (left and bottom):
return Qt.SizeBDiagCursor
if left or right:
return Qt.SizeHorCursor
if top or bottom:
return Qt.SizeVerCursor
return Qt.OpenHandCursor # interior = draggable to move
def mouseMoveEvent(self, event):
if self._drag:
self._perform_drag(event.globalPos())
return
self.viewport().setCursor(self._cursor_for(self._edges_at(event.pos())))
def mousePressEvent(self, event):
if event.button() == Qt.LeftButton:
edges = self._edges_at(event.pos())
if any(edges):
# Native resize where supported; else manual via event filter.
if self._start_system_resize(edges):
event.accept()
return
self._begin_drag("resize", edges, event.globalPos())
else:
# Interior press → move the window. Native move first.
if self._start_system_move():
event.accept()
return
self.viewport().setCursor(Qt.ClosedHandCursor)
self._begin_drag("move", edges, event.globalPos())
event.accept()
return
super().mousePressEvent(event)
def _begin_drag(self, mode, edges, gpos):
self._drag = mode
self._edges = edges
self._start_geo = QRect(self.window().geometry())
self._start_mouse = gpos
QApplication.instance().installEventFilter(self)
def eventFilter(self, obj, event):
if self._drag:
et = event.type()
if et == QEvent.MouseMove:
self._perform_drag(event.globalPos())
return True
if et == QEvent.MouseButtonRelease:
self._end_drag()
return True
return super().eventFilter(obj, event)
def _perform_drag(self, gpos):
if self._drag == "move":
delta = gpos - self._start_mouse
self.window().move(self._start_geo.topLeft() + delta)
elif self._drag == "resize":
self._perform_resize(gpos)
def _end_drag(self):
if self._drag:
was = self._drag
self._drag = None
QApplication.instance().removeEventFilter(self)
if was == "move":
self.viewport().setCursor(Qt.OpenHandCursor)
def mouseReleaseEvent(self, event):
if self._drag:
self._end_drag()
event.accept()
return
super().mouseReleaseEvent(event)
def _start_system_move(self):
win = self.window()
handle = win.windowHandle() if win else None
if handle is None or not hasattr(handle, "startSystemMove"):
return False
try:
return bool(handle.startSystemMove())
except Exception:
return False
@staticmethod
def _qt_edges(edges):
left, top, right, bottom = edges
qedges = 0
if left:
qedges |= Qt.LeftEdge
if right:
qedges |= Qt.RightEdge
if top:
qedges |= Qt.TopEdge
if bottom:
qedges |= Qt.BottomEdge
return Qt.Edges(qedges)
def _start_system_resize(self, edges):
win = self.window()
handle = win.windowHandle() if win else None
if handle is None or not hasattr(handle, "startSystemResize"):
return False
try:
return bool(handle.startSystemResize(self._qt_edges(edges)))
except Exception:
return False
def _perform_resize(self, gpos):
win = self.window()
left, top, right, bottom = self._edges
geo = QRect(self._start_geo)
dx = gpos.x() - self._start_mouse.x()
dy = gpos.y() - self._start_mouse.y()
if left:
geo.setLeft(self._start_geo.left() + dx)
if right:
geo.setRight(self._start_geo.right() + dx)
if top:
geo.setTop(self._start_geo.top() + dy)
if bottom:
geo.setBottom(self._start_geo.bottom() + dy)
# Don't let a dragged edge cross past the minimum size.
minw = max(win.minimumWidth(), 1)
minh = max(win.minimumHeight(), 1)
if geo.width() < minw:
if left:
geo.setLeft(geo.right() - minw)
else:
geo.setRight(geo.left() + minw)
if geo.height() < minh:
if top:
geo.setTop(geo.bottom() - minh)
else:
geo.setBottom(geo.top() + minh)
win.setGeometry(geo)
class _Bridge(QObject):
"""Signal bridge so audio/AI threads can safely update the Qt UI."""
answer_ready = pyqtSignal(str, str, bool) # answer, question, suggested
@@ -64,6 +236,8 @@ class InvisibleOverlay:
Qt.WindowStaysOnTopHint | Qt.FramelessWindowHint | Qt.Tool
)
self.window.setAttribute(Qt.WA_TranslucentBackground)
# Let the user drag the window larger/smaller; don't let it collapse.
self.window.setMinimumSize(320, 120)
central = QWidget()
central.setStyleSheet("background: transparent;")
@@ -71,7 +245,8 @@ class InvisibleOverlay:
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
self.text_widget = QTextEdit()
# _ResizeTextEdit fills the window; its opaque edges are the resize grips.
self.text_widget = _ResizeTextEdit()
self.text_widget.setReadOnly(True)
self.text_widget.setStyleSheet("""
QTextEdit {

162
src/web_search.py Normal file
View File

@@ -0,0 +1,162 @@
"""Optional DuckDuckGo web search (off by default).
Dependency-free: uses only the standard library. Two sources are tried per
query — DuckDuckGo's Instant Answer API (a quick abstract, when available) and
the HTML results endpoint (organic results) — and merged into a short list of
{title, snippet, url} dicts the LLM can ground its answer on.
PRIVACY: enabling this sends the query text off-device to DuckDuckGo. That
breaks the app's "nothing leaves the machine" guarantee, which is why it is
gated behind ai.web_search.enabled (default false).
"""
import os
import ssl
import json
import re
import html
import urllib.parse
import urllib.request
_UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36")
def _ssl_context():
"""Honor the corporate CA bundle run.sh exports; fall back to unverified
(the app already runs behind an SSL-inspecting proxy and sets
HF_HUB_DISABLE_SSL_VERIFY, so being lenient here is consistent)."""
ca = (os.environ.get("SSL_CERT_FILE")
or os.environ.get("REQUESTS_CA_BUNDLE")
or os.environ.get("CURL_CA_BUNDLE"))
try:
if ca and os.path.exists(ca):
return ssl.create_default_context(cafile=ca)
return ssl.create_default_context()
except Exception:
return ssl._create_unverified_context()
def _is_ssl_error(exc):
"""urllib wraps the SSLError in URLError.reason — detect either form."""
if isinstance(exc, ssl.SSLError):
return True
reason = getattr(exc, "reason", None)
return isinstance(reason, ssl.SSLError)
def _get(url, timeout, data=None):
req = urllib.request.Request(url, data=data, headers={"User-Agent": _UA})
try:
with urllib.request.urlopen(req, timeout=timeout, context=_ssl_context()) as r:
return r.read().decode("utf-8", "replace")
except Exception as e:
# An SSL-inspecting proxy can present a CA that fails strict verification
# (e.g. "CA cert does not include key usage extension"). Retry once
# unverified — consistent with the app's HF_HUB_DISABLE_SSL_VERIFY.
if not _is_ssl_error(e):
return None
try:
with urllib.request.urlopen(
req, timeout=timeout, context=ssl._create_unverified_context()
) as r:
return r.read().decode("utf-8", "replace")
except Exception:
return None
def _instant_answer(query, timeout):
url = ("https://api.duckduckgo.com/?" + urllib.parse.urlencode(
{"q": query, "format": "json", "no_html": "1", "skip_disambig": "1"}))
body = _get(url, timeout)
if not body:
return None
try:
data = json.loads(body)
except Exception:
return None
abstract = (data.get("AbstractText") or "").strip()
if abstract:
return {
"title": (data.get("Heading") or query).strip(),
"snippet": abstract,
"url": (data.get("AbstractURL") or "").strip(),
}
return None
_A_RE = re.compile(
r'<a[^>]*class="result__a"[^>]*href="(?P<href>[^"]+)"[^>]*>(?P<title>.*?)</a>',
re.S)
_SNIPPET_RE = re.compile(
r'<a[^>]*class="result__snippet"[^>]*>(?P<snip>.*?)</a>', re.S)
def _strip_tags(s):
return html.unescape(re.sub(r"<[^>]+>", "", s)).strip()
def _unwrap(href):
"""DDG wraps result links as /l/?uddg=<encoded-real-url>; unwrap them."""
if href.startswith("//"):
href = "https:" + href
parsed = urllib.parse.urlparse(href)
if parsed.path.endswith("/l/") or "uddg" in parsed.query:
qs = urllib.parse.parse_qs(parsed.query)
if "uddg" in qs:
return qs["uddg"][0]
return href
def _html_results(query, timeout, limit):
data = urllib.parse.urlencode({"q": query}).encode()
body = _get("https://html.duckduckgo.com/html/", timeout, data=data)
if not body:
return []
titles = list(_A_RE.finditer(body))
snippets = list(_SNIPPET_RE.finditer(body))
out = []
for i, m in enumerate(titles[:limit]):
snip = _strip_tags(snippets[i].group("snip")) if i < len(snippets) else ""
out.append({
"title": _strip_tags(m.group("title")),
"snippet": snip,
"url": _unwrap(m.group("href")),
})
return out
def search(query, max_results=3, timeout=6.0):
"""Return up to max_results {title, snippet, url} dicts, or [] on failure.
Never raises — search is best-effort and must not break answering."""
query = (query or "").strip()
if not query:
return []
results = []
ia = _instant_answer(query, timeout)
if ia:
results.append(ia)
try:
for r in _html_results(query, timeout, max_results):
if r["title"] and not any(r["url"] == e["url"] for e in results):
results.append(r)
if len(results) >= max_results:
break
except Exception:
pass
return results[:max_results]
def format_for_prompt(results):
"""Render results as a compact block to inject into the LLM prompt."""
if not results:
return ""
lines = []
for i, r in enumerate(results, 1):
line = f"{i}. {r['title']}"
if r.get("snippet"):
line += f" — {r['snippet']}"
if r.get("url"):
line += f" ({r['url']})"
lines.append(line)
return "\n".join(lines)