{"owner":"QuentinFuxa","repo":"WhisperLiveKit","hasSkills":true,"hasMcp":false,"mcpConfig":null,"found":["CLAUDE.md"],"skills":{"CLAUDE.md":"# CLAUDE.md -- WhisperLiveKit\n\n## Build & Test\n\nInstall for development:\n\n```sh\npip install -e \".[test]\"\n```\n\nTest with real audio using `TestHarness` (requires models + audio files):\n\n```python\nimport asyncio\nfrom whisperlivekit import TestHarness\n\nasync def main():\n    async with TestHarness(model_size=\"base\", lan=\"en\", diarization=True) as h:\n        await h.feed(\"audio.wav\", speed=1.0)     # feed at real-time\n        await h.drain(2.0)                         # let ASR catch up\n        h.print_state()                            # see current output\n\n        await h.silence(7.0, speed=1.0)            # 7s silence\n        await h.wait_for_silence()                 # verify detection\n\n        result = await h.finish()\n        print(f\"WER: {result.wer('expected text'):.2%}\")\n        print(f\"Speakers: {result.speakers}\")\n        print(f\"Text at 3s: {result.text_at(3.0)}\")\n\nasyncio.run(main())\n```\n\n## Architecture\n\nWhisperLiveKit is a real-time speech transcription system using WebSockets.\n\n- **TranscriptionEngine** (singleton) loads models once at startup and is shared across all sessions.\n- **AudioProcessor** is created per WebSocket session. It runs an async producer-consumer pipeline: FFmpeg decodes audio, Silero VAD detects speech, the ASR backend transcribes, and results stream back to the client.\n- Two streaming policies:\n  - **LocalAgreement** (HypothesisBuffer) -- confirms tokens only when consecutive inferences agree.\n  - **SimulStreaming** (AlignAtt attention-based) -- emits tokens as soon as alignment attention is confident.\n- 6 ASR backends: WhisperASR, FasterWhisperASR, MLXWhisper, VoxtralMLX, VoxtralHF, Qwen3.\n- **SessionASRProxy** wraps the shared ASR with a per-session language override, using a lock to safely swap `original_language` during `transcribe()`.\n- **DiffTracker** implements a snapshot-then-diff protocol for bandwidth-efficient incremental WebSocket updates (opt-in via `?mode=diff`).\n\n## Key Files\n\n| File | Purpose |\n|---|---|\n| `config.py` | `WhisperLiveKitConfig` dataclass -- single source of truth for configuration |\n| `core.py` | `TranscriptionEngine` singleton, `online_factory()`, diarization/translation factories |\n| `audio_processor.py` | Per-session async pipeline (FFmpeg -> VAD -> ASR -> output) |\n| `basic_server.py` | FastAPI server: WebSocket `/asr`, REST `/v1/audio/transcriptions`, CLI `wlk` |\n| `timed_objects.py` | `ASRToken`, `Segment`, `FrontData` data structures |\n| `diff_protocol.py` | `DiffTracker` -- snapshot-then-diff WebSocket protocol |\n| `session_asr_proxy.py` | `SessionASRProxy` -- thread-safe per-session language wrapper |\n| `parse_args.py` | CLI argument parser, returns `WhisperLiveKitConfig` |\n| `test_client.py` | Headless WebSocket test client (`wlk-test`) |\n| `test_harness.py` | In-process testing harness (`TestHarness`) for real E2E testing |\n| `local_agreement/online_asr.py` | `OnlineASRProcessor` for LocalAgreement policy |\n| `simul_whisper/` | SimulStreaming policy implementation (AlignAtt) |\n\n## Key Patterns\n\n- **TranscriptionEngine** uses double-checked locking for thread-safe singleton initialization. Never create a second instance in production. Use `TranscriptionEngine.reset()` in tests only to switch backends.\n- **WhisperLiveKitConfig** dataclass is the single source of truth. Use `from_namespace()` (from argparse) or `from_kwargs()` (programmatic). `parse_args()` returns a `WhisperLiveKitConfig`, not a raw Namespace.\n- **online_factory()** in `core.py` routes to the correct online processor class based on backend and policy.\n- **FrontData.to_dict()** is the canonical output format for WebSocket messages.\n- **SessionASRProxy** uses `__getattr__` delegation -- it forwards everything except `transcribe()` to the wrapped ASR.\n- The server exposes `self.args` as a `Namespace` on `TranscriptionEngine` for backward compatibility with `AudioProcessor`.\n\n## Adding a New ASR Backend\n\n1. Create `whisperlivekit/my_backend.py` with a class implementing:\n   - `transcribe(audio, init_prompt=\"\")` -- run inference on audio array\n   - `ts_words(result)` -- extract timestamped words from result\n   - `segments_end_ts(result)` -- extract segment end timestamps\n   - `use_vad()` -- whether this backend needs external VAD\n2. Set required attributes on the class: `sep`, `original_language`, `backend_choice`, `SAMPLING_RATE`, `confidence_validation`, `tokenizer`, `buffer_trimming`, `buffer_trimming_sec`.\n3. Register in `core.py`:\n   - Add an `elif` branch in `TranscriptionEngine._do_init()` to instantiate the backend.\n   - Add a routing case in `online_factory()` to return the appropriate online processor.\n4. Add the backend choice to CLI args in `parse_args.py`.\n\n## Testing with TestHarness\n\n`TestHarness` wraps AudioProcessor in-process for full pipeline testing without a server.\n\nKey methods:\n- `feed(path, speed=1.0)` -- feed audio at controlled speed (0 = instant)\n- `silence(duration, speed=1.0)` -- inject silence (>5s triggers silence detection)\n- `drain(seconds)` -- wait for ASR to catch up without feeding audio\n- `finish(timeout)` -- signal end-of-audio, wait for pipeline to drain\n- `state` -- current `TestState` with lines, buffers, speakers, timestamps\n- `wait_for(predicate)` / `wait_for_text()` / `wait_for_silence()` / `wait_for_speakers(n)`\n- `snapshot_at(audio_time)` -- historical state at a given audio position\n- `on_update(callback)` -- register callback for each state update\n\n`TestState` provides:\n- `text`, `committed_text` -- full or committed-only transcription\n- `speakers`, `n_speakers`, `has_silence` -- speaker/silence info\n- `line_at(time_s)`, `speaker_at(time_s)`, `text_at(time_s)` -- query by timestamp\n- `lines_between(start, end)`, `text_between(start, end)` -- query by time range\n- `wer(reference)`, `wer_detailed(reference)` -- evaluation against ground truth\n- `speech_lines`, `silence_segments` -- filtered line lists\n\n## OpenAI-Compatible REST API\n\nThe server exposes an OpenAI-compatible batch transcription endpoint:\n\n```bash\n# Transcribe a file (drop-in replacement for OpenAI)\ncurl http://localhost:8000/v1/audio/transcriptions \\\n  -F file=@audio.mp3 \\\n  -F response_format=verbose_json\n\n# Works with the OpenAI Python client\nfrom openai import OpenAI\nclient = OpenAI(base_url=\"http://localhost:8000/v1\", api_key=\"unused\")\nresult = client.audio.transcriptions.create(model=\"whisper-1\", file=open(\"audio.mp3\", \"rb\"))\nprint(result.text)\n```\n\nSupported `response_format` values: `json`, `verbose_json`, `text`, `srt`, `vtt`.\nThe `model` parameter is accepted but ignored (uses the server's configured backend).\n\n## Do NOT\n\n- Do not create a second `TranscriptionEngine` instance. It is a singleton; the constructor returns the existing instance after the first call.\n- Do not modify `original_language` on the shared ASR directly. Use `SessionASRProxy` for per-session language overrides.\n- Do not assume the frontend handles diff protocol messages. Diff mode is opt-in (`?mode=diff`) and ignored by default.\n- Do not write mock-based unit tests. Use `TestHarness` with real audio for pipeline testing.\n"},"files":{"CLAUDE.md":"# CLAUDE.md -- WhisperLiveKit\n\n## Build & Test\n\nInstall for development:\n\n```sh\npip install -e \".[test]\"\n```\n\nTest with real audio using `TestHarness` (requires models + audio files):\n\n```python\nimport asyncio\nfrom whisperlivekit import TestHarness\n\nasync def main():\n    async with TestHarness(model_size=\"base\", lan=\"en\", diarization=True) as h:\n        await h.feed(\"audio.wav\", speed=1.0)     # feed at real-time\n        await h.drain(2.0)                         # let ASR catch up\n        h.print_state()                            # see current output\n\n        await h.silence(7.0, speed=1.0)            # 7s silence\n        await h.wait_for_silence()                 # verify detection\n\n        result = await h.finish()\n        print(f\"WER: {result.wer('expected text'):.2%}\")\n        print(f\"Speakers: {result.speakers}\")\n        print(f\"Text at 3s: {result.text_at(3.0)}\")\n\nasyncio.run(main())\n```\n\n## Architecture\n\nWhisperLiveKit is a real-time speech transcription system using WebSockets.\n\n- **TranscriptionEngine** (singleton) loads models once at startup and is shared across all sessions.\n- **AudioProcessor** is created per WebSocket session. It runs an async producer-consumer pipeline: FFmpeg decodes audio, Silero VAD detects speech, the ASR backend transcribes, and results stream back to the client.\n- Two streaming policies:\n  - **LocalAgreement** (HypothesisBuffer) -- confirms tokens only when consecutive inferences agree.\n  - **SimulStreaming** (AlignAtt attention-based) -- emits tokens as soon as alignment attention is confident.\n- 6 ASR backends: WhisperASR, FasterWhisperASR, MLXWhisper, VoxtralMLX, VoxtralHF, Qwen3.\n- **SessionASRProxy** wraps the shared ASR with a per-session language override, using a lock to safely swap `original_language` during `transcribe()`.\n- **DiffTracker** implements a snapshot-then-diff protocol for bandwidth-efficient incremental WebSocket updates (opt-in via `?mode=diff`).\n\n## Key Files\n\n| File | Purpose |\n|---|---|\n| `config.py` | `WhisperLiveKitConfig` dataclass -- single source of truth for configuration |\n| `core.py` | `TranscriptionEngine` singleton, `online_factory()`, diarization/translation factories |\n| `audio_processor.py` | Per-session async pipeline (FFmpeg -> VAD -> ASR -> output) |\n| `basic_server.py` | FastAPI server: WebSocket `/asr`, REST `/v1/audio/transcriptions`, CLI `wlk` |\n| `timed_objects.py` | `ASRToken`, `Segment`, `FrontData` data structures |\n| `diff_protocol.py` | `DiffTracker` -- snapshot-then-diff WebSocket protocol |\n| `session_asr_proxy.py` | `SessionASRProxy` -- thread-safe per-session language wrapper |\n| `parse_args.py` | CLI argument parser, returns `WhisperLiveKitConfig` |\n| `test_client.py` | Headless WebSocket test client (`wlk-test`) |\n| `test_harness.py` | In-process testing harness (`TestHarness`) for real E2E testing |\n| `local_agreement/online_asr.py` | `OnlineASRProcessor` for LocalAgreement policy |\n| `simul_whisper/` | SimulStreaming policy implementation (AlignAtt) |\n\n## Key Patterns\n\n- **TranscriptionEngine** uses double-checked locking for thread-safe singleton initialization. Never create a second instance in production. Use `TranscriptionEngine.reset()` in tests only to switch backends.\n- **WhisperLiveKitConfig** dataclass is the single source of truth. Use `from_namespace()` (from argparse) or `from_kwargs()` (programmatic). `parse_args()` returns a `WhisperLiveKitConfig`, not a raw Namespace.\n- **online_factory()** in `core.py` routes to the correct online processor class based on backend and policy.\n- **FrontData.to_dict()** is the canonical output format for WebSocket messages.\n- **SessionASRProxy** uses `__getattr__` delegation -- it forwards everything except `transcribe()` to the wrapped ASR.\n- The server exposes `self.args` as a `Namespace` on `TranscriptionEngine` for backward compatibility with `AudioProcessor`.\n\n## Adding a New ASR Backend\n\n1. Create `whisperlivekit/my_backend.py` with a class implementing:\n   - `transcribe(audio, init_prompt=\"\")` -- run inference on audio array\n   - `ts_words(result)` -- extract timestamped words from result\n   - `segments_end_ts(result)` -- extract segment end timestamps\n   - `use_vad()` -- whether this backend needs external VAD\n2. Set required attributes on the class: `sep`, `original_language`, `backend_choice`, `SAMPLING_RATE`, `confidence_validation`, `tokenizer`, `buffer_trimming`, `buffer_trimming_sec`.\n3. Register in `core.py`:\n   - Add an `elif` branch in `TranscriptionEngine._do_init()` to instantiate the backend.\n   - Add a routing case in `online_factory()` to return the appropriate online processor.\n4. Add the backend choice to CLI args in `parse_args.py`.\n\n## Testing with TestHarness\n\n`TestHarness` wraps AudioProcessor in-process for full pipeline testing without a server.\n\nKey methods:\n- `feed(path, speed=1.0)` -- feed audio at controlled speed (0 = instant)\n- `silence(duration, speed=1.0)` -- inject silence (>5s triggers silence detection)\n- `drain(seconds)` -- wait for ASR to catch up without feeding audio\n- `finish(timeout)` -- signal end-of-audio, wait for pipeline to drain\n- `state` -- current `TestState` with lines, buffers, speakers, timestamps\n- `wait_for(predicate)` / `wait_for_text()` / `wait_for_silence()` / `wait_for_speakers(n)`\n- `snapshot_at(audio_time)` -- historical state at a given audio position\n- `on_update(callback)` -- register callback for each state update\n\n`TestState` provides:\n- `text`, `committed_text` -- full or committed-only transcription\n- `speakers`, `n_speakers`, `has_silence` -- speaker/silence info\n- `line_at(time_s)`, `speaker_at(time_s)`, `text_at(time_s)` -- query by timestamp\n- `lines_between(start, end)`, `text_between(start, end)` -- query by time range\n- `wer(reference)`, `wer_detailed(reference)` -- evaluation against ground truth\n- `speech_lines`, `silence_segments` -- filtered line lists\n\n## OpenAI-Compatible REST API\n\nThe server exposes an OpenAI-compatible batch transcription endpoint:\n\n```bash\n# Transcribe a file (drop-in replacement for OpenAI)\ncurl http://localhost:8000/v1/audio/transcriptions \\\n  -F file=@audio.mp3 \\\n  -F response_format=verbose_json\n\n# Works with the OpenAI Python client\nfrom openai import OpenAI\nclient = OpenAI(base_url=\"http://localhost:8000/v1\", api_key=\"unused\")\nresult = client.audio.transcriptions.create(model=\"whisper-1\", file=open(\"audio.mp3\", \"rb\"))\nprint(result.text)\n```\n\nSupported `response_format` values: `json`, `verbose_json`, `text`, `srt`, `vtt`.\nThe `model` parameter is accepted but ignored (uses the server's configured backend).\n\n## Do NOT\n\n- Do not create a second `TranscriptionEngine` instance. It is a singleton; the constructor returns the existing instance after the first call.\n- Do not modify `original_language` on the shared ASR directly. Use `SessionASRProxy` for per-session language overrides.\n- Do not assume the frontend handles diff protocol messages. Diff mode is opt-in (`?mode=diff`) and ignored by default.\n- Do not write mock-based unit tests. Use `TestHarness` with real audio for pipeline testing.\n"},"items":[{"name":"CLAUDE.md","path":"CLAUDE.md","title":"CLAUDE.md","content":"# CLAUDE.md -- WhisperLiveKit\n\n## Build & Test\n\nInstall for development:\n\n```sh\npip install -e \".[test]\"\n```\n\nTest with real audio using `TestHarness` (requires models + audio files):\n\n```python\nimport asyncio\nfrom whisperlivekit import TestHarness\n\nasync def main():\n    async with TestHarness(model_size=\"base\", lan=\"en\", diarization=True) as h:\n        await h.feed(\"audio.wav\", speed=1.0)     # feed at real-time\n        await h.drain(2.0)                         # let ASR catch up\n        h.print_state()                            # see current output\n\n        await h.silence(7.0, speed=1.0)            # 7s silence\n        await h.wait_for_silence()                 # verify detection\n\n        result = await h.finish()\n        print(f\"WER: {result.wer('expected text'):.2%}\")\n        print(f\"Speakers: {result.speakers}\")\n        print(f\"Text at 3s: {result.text_at(3.0)}\")\n\nasyncio.run(main())\n```\n\n## Architecture\n\nWhisperLiveKit is a real-time speech transcription system using WebSockets.\n\n- **TranscriptionEngine** (singleton) loads models once at startup and is shared across all sessions.\n- **AudioProcessor** is created per WebSocket session. It runs an async producer-consumer pipeline: FFmpeg decodes audio, Silero VAD detects speech, the ASR backend transcribes, and results stream back to the client.\n- Two streaming policies:\n  - **LocalAgreement** (HypothesisBuffer) -- confirms tokens only when consecutive inferences agree.\n  - **SimulStreaming** (AlignAtt attention-based) -- emits tokens as soon as alignment attention is confident.\n- 6 ASR backends: WhisperASR, FasterWhisperASR, MLXWhisper, VoxtralMLX, VoxtralHF, Qwen3.\n- **SessionASRProxy** wraps the shared ASR with a per-session language override, using a lock to safely swap `original_language` during `transcribe()`.\n- **DiffTracker** implements a snapshot-then-diff protocol for bandwidth-efficient incremental WebSocket updates (opt-in via `?mode=diff`).\n\n## Key Files\n\n| File | Purpose |\n|---|---|\n| `config.py` | `WhisperLiveKitConfig` dataclass -- single source of truth for configuration |\n| `core.py` | `TranscriptionEngine` singleton, `online_factory()`, diarization/translation factories |\n| `audio_processor.py` | Per-session async pipeline (FFmpeg -> VAD -> ASR -> output) |\n| `basic_server.py` | FastAPI server: WebSocket `/asr`, REST `/v1/audio/transcriptions`, CLI `wlk` |\n| `timed_objects.py` | `ASRToken`, `Segment`, `FrontData` data structures |\n| `diff_protocol.py` | `DiffTracker` -- snapshot-then-diff WebSocket protocol |\n| `session_asr_proxy.py` | `SessionASRProxy` -- thread-safe per-session language wrapper |\n| `parse_args.py` | CLI argument parser, returns `WhisperLiveKitConfig` |\n| `test_client.py` | Headless WebSocket test client (`wlk-test`) |\n| `test_harness.py` | In-process testing harness (`TestHarness`) for real E2E testing |\n| `local_agreement/online_asr.py` | `OnlineASRProcessor` for LocalAgreement policy |\n| `simul_whisper/` | SimulStreaming policy implementation (AlignAtt) |\n\n## Key Patterns\n\n- **TranscriptionEngine** uses double-checked locking for thread-safe singleton initialization. Never create a second instance in production. Use `TranscriptionEngine.reset()` in tests only to switch backends.\n- **WhisperLiveKitConfig** dataclass is the single source of truth. Use `from_namespace()` (from argparse) or `from_kwargs()` (programmatic). `parse_args()` returns a `WhisperLiveKitConfig`, not a raw Namespace.\n- **online_factory()** in `core.py` routes to the correct online processor class based on backend and policy.\n- **FrontData.to_dict()** is the canonical output format for WebSocket messages.\n- **SessionASRProxy** uses `__getattr__` delegation -- it forwards everything except `transcribe()` to the wrapped ASR.\n- The server exposes `self.args` as a `Namespace` on `TranscriptionEngine` for backward compatibility with `AudioProcessor`.\n\n## Adding a New ASR Backend\n\n1. Create `whisperlivekit/my_backend.py` with a class implementing:\n   - `transcribe(audio, init_prompt=\"\")` -- run inference on audio array\n   - `ts_words(result)` -- extract timestamped words from result\n   - `segments_end_ts(result)` -- extract segment end timestamps\n   - `use_vad()` -- whether this backend needs external VAD\n2. Set required attributes on the class: `sep`, `original_language`, `backend_choice`, `SAMPLING_RATE`, `confidence_validation`, `tokenizer`, `buffer_trimming`, `buffer_trimming_sec`.\n3. Register in `core.py`:\n   - Add an `elif` branch in `TranscriptionEngine._do_init()` to instantiate the backend.\n   - Add a routing case in `online_factory()` to return the appropriate online processor.\n4. Add the backend choice to CLI args in `parse_args.py`.\n\n## Testing with TestHarness\n\n`TestHarness` wraps AudioProcessor in-process for full pipeline testing without a server.\n\nKey methods:\n- `feed(path, speed=1.0)` -- feed audio at controlled speed (0 = instant)\n- `silence(duration, speed=1.0)` -- inject silence (>5s triggers silence detection)\n- `drain(seconds)` -- wait for ASR to catch up without feeding audio\n- `finish(timeout)` -- signal end-of-audio, wait for pipeline to drain\n- `state` -- current `TestState` with lines, buffers, speakers, timestamps\n- `wait_for(predicate)` / `wait_for_text()` / `wait_for_silence()` / `wait_for_speakers(n)`\n- `snapshot_at(audio_time)` -- historical state at a given audio position\n- `on_update(callback)` -- register callback for each state update\n\n`TestState` provides:\n- `text`, `committed_text` -- full or committed-only transcription\n- `speakers`, `n_speakers`, `has_silence` -- speaker/silence info\n- `line_at(time_s)`, `speaker_at(time_s)`, `text_at(time_s)` -- query by timestamp\n- `lines_between(start, end)`, `text_between(start, end)` -- query by time range\n- `wer(reference)`, `wer_detailed(reference)` -- evaluation against ground truth\n- `speech_lines`, `silence_segments` -- filtered line lists\n\n## OpenAI-Compatible REST API\n\nThe server exposes an OpenAI-compatible batch transcription endpoint:\n\n```bash\n# Transcribe a file (drop-in replacement for OpenAI)\ncurl http://localhost:8000/v1/audio/transcriptions \\\n  -F file=@audio.mp3 \\\n  -F response_format=verbose_json\n\n# Works with the OpenAI Python client\nfrom openai import OpenAI\nclient = OpenAI(base_url=\"http://localhost:8000/v1\", api_key=\"unused\")\nresult = client.audio.transcriptions.create(model=\"whisper-1\", file=open(\"audio.mp3\", \"rb\"))\nprint(result.text)\n```\n\nSupported `response_format` values: `json`, `verbose_json`, `text`, `srt`, `vtt`.\nThe `model` parameter is accepted but ignored (uses the server's configured backend).\n\n## Do NOT\n\n- Do not create a second `TranscriptionEngine` instance. It is a singleton; the constructor returns the existing instance after the first call.\n- Do not modify `original_language` on the shared ASR directly. Use `SessionASRProxy` for per-session language overrides.\n- Do not assume the frontend handles diff protocol messages. Diff mode is opt-in (`?mode=diff`) and ignored by default.\n- Do not write mock-based unit tests. Use `TestHarness` with real audio for pipeline testing.\n","category":"root","tokens":1770}]}