From 093da817d889a1672102e42c99b3c700eae5a175 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dieter=20Schl=C3=BCter?= Date: Wed, 17 Jun 2026 05:06:58 +0200 Subject: [PATCH] feat: Barge-in/Turn-Manager und VAD-Aeusserungserkennung (#4 Ausbau) - Barge-in: Antwort-Turn als abbrechbarer asyncio.Task; {"type":"interrupt"} oder neue Eingabe bricht laufende Antwort ab -> interrupted-Event (/ws/chat + /ws/voice) - VAD (app/audio/vad.py): energie-basierte Stille-Erkennung (reines Python, int16-PCM) - /ws/voice opt-in {"type":"start","vad":true}: automatisches Aeusserungsende ohne end - Tests: 52 gruen (+5: VAD-Unit, Barge-in, VAD-Auto-Segmentierung) - Doku aktualisiert (README, BEDIENUNGSANLEITUNG, Architektur) Offen (schwere Deps/Dienste): echte partielle Live-Transkripte (Streaming-STT), WebRTC (aiortc). STT laeuft heute pro Aeusserung. Co-Authored-By: Claude Opus 4.8 --- BEDIENUNGSANLEITUNG.md | 7 +- Docs/voice-assistant-architecture.md | 7 +- README.md | 14 ++- app/api/ws.py | 144 +++++++++++++++++++-------- app/audio/vad.py | 59 +++++++++++ tests/test_realtime.py | 129 ++++++++++++++++++++++++ 6 files changed, 312 insertions(+), 48 deletions(-) create mode 100644 app/audio/vad.py create mode 100644 tests/test_realtime.py diff --git a/BEDIENUNGSANLEITUNG.md b/BEDIENUNGSANLEITUNG.md index e25b2c1..da8b369 100644 --- a/BEDIENUNGSANLEITUNG.md +++ b/BEDIENUNGSANLEITUNG.md @@ -280,7 +280,12 @@ Satz fertig ist. **Sprach-Eingang** (`/ws/voice`): Mikrofon-Audio als binäre Frames senden, dann `{"type":"end"}`. Der Server schickt ein `transcript`-Event und danach die Antwort -wie bei `/ws/chat` (`stream`/`audio_stream` im `end`-Frame möglich). +wie bei `/ws/chat` (`stream`/`audio_stream` im `end`-Frame möglich). Mit +`{"type":"start","vad":true,"format":"pcm","sample_rate":16000}` erkennt der Server +das Äußerungsende automatisch an einer Sprechpause (kein `end` nötig). + +**Unterbrechen (Barge-in):** Während der Assistent antwortet, `{"type":"interrupt"}` +senden — die laufende Antwort wird abgebrochen (`interrupted`-Event). > **Für lokale Entwicklung** ist in der mitgelieferten `.env` `AUTH_ENABLED=false` > gesetzt — dann ist kein Token nötig (anonymer Nutzer). diff --git a/Docs/voice-assistant-architecture.md b/Docs/voice-assistant-architecture.md index 2743be8..228732d 100644 --- a/Docs/voice-assistant-architecture.md +++ b/Docs/voice-assistant-architecture.md @@ -189,7 +189,8 @@ Device Router (strikt, Singleton); Output-Lifecycle; **Authentifizierung Store, fließt ins LLM)**; **Langzeit-Erinnerungen pro Nutzer (als LLM-Kontext)**; **WebSocket-Streaming-Chat (`/ws/chat`) inkl. Token-Level-LLM-Streaming (SSE, `stream:true`) und satzweisem Audio-Streaming (chunked TTS, `audio_stream:true`)**; **Sprach-Eingang -über WebSocket (`/ws/voice`: Audio rein → STT → Antwort-Pipeline)**; automatisierte Tests. +über WebSocket (`/ws/voice`: Audio rein → STT → Antwort-Pipeline) mit VAD-Aeusserungs- +erkennung und Barge-in (`interrupt`)**; automatisierte Tests. **Platzhalter (Gerüst):** Audio-Endpunkte (`local-default`, `bluetooth`, `mobile-ws`, `mobile-webrtc`) liefern leere Chunks — nur Auswahl/Lifecycle sind @@ -204,7 +205,7 @@ Reihenfolge der Weiterentwicklung: 1. **(erledigt)** Konfig- & Routing-Fundament: Profile, Device Router, Registry, Pro-Request-Override. 2. **(erledigt)** Cloud-Fundament: Bearer-Token-Auth, Mehrbenutzer, persistenter SQLite-Store, Mandanten-Trennung, dauerhafte Nutzer-Präferenzen. Offen: Skalierung auf gemeinsamen Store (Postgres/Redis) für mehrere Instanzen. 3. **(erledigt)** Konversationsgedächtnis: Kurzzeit-Gesprächsverlauf pro Session + Langzeit-Erinnerungen pro Nutzer (manuell gepflegt, als LLM-Kontext). Offen: **automatische** Extraktion/Zusammenfassung von Erinnerungen aus Gesprächen. -4. **(teilweise erledigt)** Echtzeit: WebSocket-Streaming-Chat (`/ws/chat`), **Token-Level-LLM-Streaming (SSE, `stream:true`)**, **Audio-Streaming (chunked TTS satzweise, `audio_stream:true`)** und **Audio-Eingang (`/ws/voice`: binaere Chunks -> STT -> Antwort-Pipeline)** sind umgesetzt. Offen: **partielle Live-Transkripte (Streaming-STT mit VAD)**, **Barge-in/Turn-Manager**, **WebRTC**. +4. **(weitgehend erledigt)** Echtzeit: WebSocket-Streaming-Chat (`/ws/chat`), **Token-Level-LLM-Streaming (SSE, `stream:true`)**, **Audio-Streaming (chunked TTS satzweise, `audio_stream:true`)**, **Audio-Eingang (`/ws/voice`)**, **Barge-in/Turn-Manager (`interrupt` bricht laufende Antwort ab)** und **VAD-Aeusserungserkennung (energie-basiert, opt-in)** sind umgesetzt. Offen: **echte partielle Live-Transkripte (Streaming-STT-Dienst, wortweise)** und **WebRTC (aiortc)** — beide brauchen schwere Abhaengigkeiten/Dienste. Heute laeuft STT pro Aeusserung. 5. **Resilienz:** Fallback-Policy (remote KI fällt aus → lokaler/alternativer Provider), Metriken/Tracing. 6. **Betrieb:** Kosten-/Quota-Kontrolle pro Nutzer; Notfall-/Eskalationskonzept (Senioren-Kontext). 7. **TransportRouter** als eigene lokal/remote-Achse aktivieren. @@ -227,7 +228,7 @@ voice-assistant-scaffold/ │ ├── schemas.py # Pydantic-Modelle │ ├── api/ # health, chat, speak, transcribe, devices, sessions, config, admin, me, ws │ ├── core/ # orchestrator -│ ├── audio/ # router, transport_router, endpoints/input|output/* +│ ├── audio/ # router, transport_router, vad, endpoints/input|output/* │ ├── pipeline/ # input_cleaner, spoken_response_adapter, tts_normalizer, sentence_chunker │ └── providers/ # stt/ llm/ tts/ (openrouter + lokale Stubs) ├── config/ # voice-assistant.example.toml (+ lokale .toml, gitignored) diff --git a/README.md b/README.md index 754d56d..6929aa8 100644 --- a/README.md +++ b/README.md @@ -149,8 +149,18 @@ transkribiert (STT), sendet ein `transcript`-Event und durchläuft dann dieselbe Antwort-Pipeline wie `/ws/chat` (inkl. `stream`/`audio_stream`). Damit ist Sprach-zu-Sprach-Konversation über einen Kanal möglich. -> Partielle Live-Transkripte (Streaming-STT mit VAD), Barge-in/Turn-Manager und -> WebRTC sind als nächste Increments vorgesehen (siehe Architektur-Dokument). +**VAD (automatische Äußerungserkennung):** Mit `{"type":"start","vad":true, +"sample_rate":16000,"format":"pcm"}` segmentiert der Server Äußerungen selbst anhand +von Stille (energie-basiert, reines Python) — ohne explizites `end`. Optional: +`vad_silence_ms`, `vad_threshold`. + +**Barge-in:** Eine laufende Antwort lässt sich mit `{"type":"interrupt"}` (oder durch +eine neue Eingabe) abbrechen — der Server stoppt das Streaming und meldet +`{"type":"interrupted"}`. Wichtig für natürliche Gespräche. + +> Echte **partielle Live-Transkripte** (Streaming-STT-Dienst, wortweise während des +> Sprechens) und **WebRTC** sind als nächste Increments vorgesehen (siehe +> Architektur-Dokument). Heute läuft STT pro Äußerung. ## Authentifizierung diff --git a/app/api/ws.py b/app/api/ws.py index 960c549..5fe7488 100644 --- a/app/api/ws.py +++ b/app/api/ws.py @@ -3,14 +3,18 @@ - /ws/chat : Text rein (JSON pro Turn), Antwort als Event-Folge zurueck. - /ws/voice: Audio rein (binaere Chunks + Control), Transkription -> selbe Pipeline. -Event-Folge der Antwort: ack -> [token*] -> [audio*] -> semantic -> done. +Antwort-Events: ack -> [token*] -> [audio*] -> semantic -> done. Mit {"stream":true} kommen LLM-Token live, mit {"audio_stream":true} das Audio satzweise (chunked TTS). /ws/voice sendet zuvor ein transcript-Event. -Spaeter (eigene Increments): partielle Live-Transkripte (Streaming-STT mit VAD), -Barge-in/Turn-Manager und WebRTC. +Barge-in: Ein {"type":"interrupt"}-Frame oder eine neue Eingabe bricht eine laufende +Antwort ab (-> interrupted-Event). Der Antwort-Turn laeuft als abbrechbarer Task. + +Spaeter (eigene Increments): echte partielle Live-Transkripte (Streaming-STT-Dienst), +WebRTC. """ +import asyncio import json from fastapi import APIRouter, WebSocket, WebSocketDisconnect @@ -24,6 +28,7 @@ from app.dependencies import ( resolve_output_endpoint, ) from app.store import SessionOwnershipError +from app.audio.vad import EnergyVAD router = APIRouter() @@ -126,6 +131,51 @@ async def _run_turn(websocket, store, user, session_id, route, orchestrator, out await websocket.send_json({"type": "done", "audio_format": "pcm", "sample_rate": 24000}) +async def _cancel_active(task, websocket) -> None: + """Bricht einen laufenden Antwort-Turn ab (Barge-in) und meldet 'interrupted'.""" + if task is None or task.done(): + return + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + await websocket.send_json({"type": "interrupted"}) + + +async def _chat_turn(websocket, store, user, session_id, text, options): + try: + route, orchestrator, output = await _resolve(user, session_id, options) + except SessionOwnershipError as exc: + await websocket.send_json({"type": "error", "status": 403, "detail": str(exc)}) + return + except RoutingError as exc: + await websocket.send_json({"type": "error", "status": 422, "detail": str(exc)}) + return + await _run_turn(websocket, store, user, session_id, route, orchestrator, output, text, options) + + +async def _voice_turn(websocket, store, user, session_id, audio, fmt, options): + try: + route, orchestrator, output = await _resolve(user, session_id, options) + except SessionOwnershipError as exc: + await websocket.send_json({"type": "error", "status": 403, "detail": str(exc)}) + return + except RoutingError as exc: + await websocket.send_json({"type": "error", "status": 422, "detail": str(exc)}) + return + try: + transcript = await orchestrator.stt.transcribe(audio, fmt=fmt, language=route.language) + except Exception as exc: + await websocket.send_json({"type": "error", "status": 502, "detail": str(exc)}) + return + await websocket.send_json({"type": "transcript", "text": transcript}) + if not transcript or not transcript.strip(): + await websocket.send_json({"type": "error", "detail": "empty transcript"}) + return + await _run_turn(websocket, store, user, session_id, route, orchestrator, output, transcript, options) + + @router.websocket("/ws/chat") async def ws_chat(websocket: WebSocket, session_id: str | None = None, token: str | None = None): user = _authenticate(token) @@ -134,24 +184,26 @@ async def ws_chat(websocket: WebSocket, session_id: str | None = None, token: st return await websocket.accept() store = get_store() + active = None try: while True: msg = await websocket.receive_json() + if msg.get("type") == "interrupt": + await _cancel_active(active, websocket) + active = None + continue text = (msg.get("text") or "").strip() if not text: await websocket.send_json({"type": "error", "detail": "empty text"}) continue - try: - route, orchestrator, output = await _resolve(user, session_id, msg) - except SessionOwnershipError as exc: - await websocket.send_json({"type": "error", "status": 403, "detail": str(exc)}) - continue - except RoutingError as exc: - await websocket.send_json({"type": "error", "status": 422, "detail": str(exc)}) - continue - await _run_turn(websocket, store, user, session_id, route, orchestrator, output, text, msg) + await _cancel_active(active, websocket) # Barge-in bei neuer Eingabe + active = asyncio.create_task( + _chat_turn(websocket, store, user, session_id, text, msg) + ) except WebSocketDisconnect: + if active and not active.done(): + active.cancel() return @@ -166,15 +218,33 @@ async def ws_voice(websocket: WebSocket, session_id: str | None = None, token: s audio_buffer = bytearray() fmt = "wav" + active = None + vad = None + vad_options: dict = {} + + async def _start_voice(audio: bytes, options: dict): + nonlocal active + await _cancel_active(active, websocket) # Barge-in bei neuer Aeusserung + active = asyncio.create_task( + _voice_turn(websocket, store, user, session_id, audio, fmt, options) + ) try: while True: message = await websocket.receive() if message["type"] == "websocket.disconnect": + if active and not active.done(): + active.cancel() return if message.get("bytes") is not None: audio_buffer.extend(message["bytes"]) + # VAD: Aeusserungsende automatisch erkennen (opt-in via start-Frame). + if vad is not None and vad.feed(message["bytes"]): + audio = bytes(audio_buffer) + audio_buffer.clear() + vad.reset() + await _start_voice(audio, vad_options) continue raw = message.get("text") @@ -187,9 +257,22 @@ async def ws_voice(websocket: WebSocket, session_id: str | None = None, token: s continue ctype = control.get("type") + if ctype == "interrupt": + await _cancel_active(active, websocket) + active = None + continue if ctype == "start": audio_buffer.clear() fmt = control.get("format", "wav") + if control.get("vad"): + vad = EnergyVAD( + sample_rate=control.get("sample_rate", 16000), + threshold=control.get("vad_threshold", 500.0), + silence_ms=control.get("vad_silence_ms", 700.0), + ) + vad_options = control + else: + vad = None continue if ctype != "end": continue @@ -198,35 +281,12 @@ async def ws_voice(websocket: WebSocket, session_id: str | None = None, token: s await websocket.send_json({"type": "error", "detail": "no audio received"}) continue - try: - route, orchestrator, output = await _resolve(user, session_id, control) - except SessionOwnershipError as exc: - await websocket.send_json({"type": "error", "status": 403, "detail": str(exc)}) - audio_buffer.clear() - continue - except RoutingError as exc: - await websocket.send_json({"type": "error", "status": 422, "detail": str(exc)}) - audio_buffer.clear() - continue - - try: - transcript = await orchestrator.stt.transcribe( - bytes(audio_buffer), fmt=fmt, language=route.language - ) - except Exception as exc: - await websocket.send_json({"type": "error", "status": 502, "detail": str(exc)}) - audio_buffer.clear() - continue - finally: - audio_buffer.clear() - - await websocket.send_json({"type": "transcript", "text": transcript}) - if not transcript or not transcript.strip(): - await websocket.send_json({"type": "error", "detail": "empty transcript"}) - continue - - await _run_turn( - websocket, store, user, session_id, route, orchestrator, output, transcript, control - ) + audio = bytes(audio_buffer) + audio_buffer.clear() + if vad is not None: + vad.reset() + await _start_voice(audio, control) except WebSocketDisconnect: + if active and not active.done(): + active.cancel() return diff --git a/app/audio/vad.py b/app/audio/vad.py new file mode 100644 index 0000000..c48484a --- /dev/null +++ b/app/audio/vad.py @@ -0,0 +1,59 @@ +"""Einfache energie-basierte Sprachaktivitaetserkennung (VAD). + +Reines Python (stdlib `array`), arbeitet auf s16le-PCM (mono). Erkennt das Ende +einer Aeusserung anhand andauernder Stille nach erkannter Sprache. Damit kann der +Server in /ws/voice Aeusserungen automatisch segmentieren, ohne dass der Client +ein explizites Ende-Signal schickt. + +Hinweis: Das ersetzt keinen echten Streaming-STT-Dienst (keine wortweisen +Teil-Transkripte) - es bestimmt nur die Aeusserungsgrenzen. +""" + +import array +import math + + +def rms(pcm: bytes) -> float: + """Lautstaerke (RMS) eines s16le-PCM-Puffers; 0.0 bei leerem Puffer.""" + usable = len(pcm) - (len(pcm) % 2) + if usable <= 0: + return 0.0 + samples = array.array("h") + samples.frombytes(pcm[:usable]) + if not samples: + return 0.0 + return math.sqrt(sum(s * s for s in samples) / len(samples)) + + +class EnergyVAD: + def __init__(self, sample_rate: int = 16000, threshold: float = 500.0, silence_ms: float = 700.0): + self.sample_rate = sample_rate + self.threshold = threshold + self.silence_ms = silence_ms + self._speech_started = False + self._silence_ms = 0.0 + + def feed(self, pcm: bytes) -> bool: + """Verarbeitet einen Audio-Chunk. + + Liefert True, sobald nach erkannter Sprache genug Stille (silence_ms) + vergangen ist - die Aeusserung gilt dann als beendet. + """ + level = rms(pcm) + n_samples = len(pcm) // 2 + chunk_ms = (n_samples / self.sample_rate) * 1000.0 if self.sample_rate else 0.0 + + if level >= self.threshold: + self._speech_started = True + self._silence_ms = 0.0 + return False + + if self._speech_started: + self._silence_ms += chunk_ms + if self._silence_ms >= self.silence_ms: + return True + return False + + def reset(self) -> None: + self._speech_started = False + self._silence_ms = 0.0 diff --git a/tests/test_realtime.py b/tests/test_realtime.py new file mode 100644 index 0000000..8df0b21 --- /dev/null +++ b/tests/test_realtime.py @@ -0,0 +1,129 @@ +import array +import asyncio + +from fastapi.testclient import TestClient + +import app.dependencies as deps +from app.main import app +from app.audio.vad import rms, EnergyVAD + +client = TestClient(app) + + +def _pcm(amplitude: int, n_samples: int) -> bytes: + return array.array("h", [amplitude] * n_samples).tobytes() + + +# --- VAD-Unit-Tests -------------------------------------------------------- + +def test_rms_silence_vs_loud(): + assert rms(b"") == 0.0 + assert rms(_pcm(0, 100)) == 0.0 + assert rms(_pcm(3000, 100)) > 2000 + + +def test_energy_vad_ends_after_speech_then_silence(): + vad = EnergyVAD(sample_rate=16000, threshold=500, silence_ms=300) + loud = _pcm(3000, 1600) # 100 ms Sprache + silent = _pcm(0, 1600) # 100 ms Stille + assert vad.feed(loud) is False + assert vad.feed(silent) is False # 100 ms + assert vad.feed(silent) is False # 200 ms + assert vad.feed(silent) is True # 300 ms -> Ende + + +def test_energy_vad_ignores_silence_without_speech(): + vad = EnergyVAD(sample_rate=16000, threshold=500, silence_ms=100) + for _ in range(10): + assert vad.feed(_pcm(0, 1600)) is False + + +# --- Barge-in -------------------------------------------------------------- + +def _install_slow_stream(monkeypatch): + class SlowLLM: + async def complete(self, text, history=None, session_id=None): + return "fertig" + + async def stream(self, text, history=None, session_id=None): + for i in range(200): + await asyncio.sleep(0.005) + yield f"t{i} " + + class StubTTS: + async def synthesize(self, text, voice=None, audio_format="pcm"): + return b"A" + + monkeypatch.setitem(deps.LLM_REGISTRY, "slow", lambda s: SlowLLM()) + monkeypatch.setitem(deps.TTS_REGISTRY, "stub", lambda s: StubTTS()) + return { + "llm_provider": "slow", + "tts_provider": "stub", + "output_endpoint": "loopback", + "stream": True, + } + + +def test_ws_chat_interrupt_cancels_response(monkeypatch): + base = _install_slow_stream(monkeypatch) + with client.websocket_connect("/ws/chat") as ws: + ws.send_json({"text": "Hallo", **base}) + assert ws.receive_json()["type"] == "ack" + assert ws.receive_json()["type"] == "token" # Antwort laeuft + + ws.send_json({"type": "interrupt"}) + + event = None + for _ in range(500): + event = ws.receive_json() + if event["type"] in ("interrupted", "done"): + break + assert event["type"] == "interrupted" # abgebrochen, nicht fertig + + +# --- VAD im WebSocket ------------------------------------------------------ + +def _install_voice_stubs(monkeypatch): + class STT: + async def transcribe(self, audio_bytes, fmt, language=None): + return "ok" + + class LLM: + async def complete(self, text, history=None, session_id=None): + return "antwort" + + class TTS: + async def synthesize(self, text, voice=None, audio_format="pcm"): + return b"A" + + monkeypatch.setitem(deps.STT_REGISTRY, "s", lambda x: STT()) + monkeypatch.setitem(deps.LLM_REGISTRY, "l", lambda x: LLM()) + monkeypatch.setitem(deps.TTS_REGISTRY, "t", lambda x: TTS()) + return {"stt_provider": "s", "llm_provider": "l", "tts_provider": "t", "output_endpoint": "loopback"} + + +def test_ws_voice_vad_auto_segments_utterance(monkeypatch): + opts = _install_voice_stubs(monkeypatch) + start = { + "type": "start", + "vad": True, + "sample_rate": 16000, + "vad_silence_ms": 200, + "format": "pcm", + **opts, + } + loud = _pcm(3000, 1600) # 100 ms Sprache + silent = _pcm(0, 1600) # je 100 ms Stille + + with client.websocket_connect("/ws/voice") as ws: + ws.send_json(start) + ws.send_bytes(loud) + ws.send_bytes(silent) # 100 ms + ws.send_bytes(silent) # 200 ms -> VAD-Ende, Turn startet automatisch + + transcript = ws.receive_json() + assert transcript["type"] == "transcript" and transcript["text"] == "ok" + assert ws.receive_json()["type"] == "ack" + assert ws.receive_json()["type"] == "semantic" + assert ws.receive_bytes() == b"A" + assert ws.receive_json()["type"] == "done"