diff --git a/BEDIENUNGSANLEITUNG.md b/BEDIENUNGSANLEITUNG.md index 7436afe..e25b2c1 100644 --- a/BEDIENUNGSANLEITUNG.md +++ b/BEDIENUNGSANLEITUNG.md @@ -278,6 +278,10 @@ als `token`-Events (geringere wahrgenommene Latenz). Mit `"audio_stream": true` kommt zusätzlich das Audio satzweise (`audio`-Event + binärer Frame), sobald ein 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). + > **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 4ca8d87..2743be8 100644 --- a/Docs/voice-assistant-architecture.md +++ b/Docs/voice-assistant-architecture.md @@ -171,7 +171,8 @@ ein No-op; `LoopbackOutput` sammelt die Chunks (testbar ohne Hardware). | `POST /api/admin/users` | Nutzer anlegen (Admin-Key) → Token einmalig | | `GET /api/me` · `PUT /api/me/prefs` | aktueller Nutzer + dauerhafte Präferenzen | | `GET/POST/DELETE /api/me/memories` | Langzeit-Erinnerungen des Nutzers | -| `WS /ws/chat` | Echtzeit-Chat (Streaming-Events über WebSocket) | +| `WS /ws/chat` | Echtzeit-Chat (Text rein, Streaming-Events) | +| `WS /ws/voice` | Echtzeit-Sprache (Audio rein → STT → Antwort) | Endpunkt-/Provider-Auswahl ist über **Request-Body** (pro Aufruf), **Session** (`?session_id=…`) und **Defaults/Profil** steuerbar. Verwendete Route erscheint als @@ -187,8 +188,8 @@ Device Router (strikt, Singleton); Output-Lifecycle; **Authentifizierung + dauerhafte Nutzer-Präferenzen**; **Gesprächsgedächtnis pro Session (Verlauf im 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`)**; -automatisierte Tests. +`stream:true`) und satzweisem Audio-Streaming (chunked TTS, `audio_stream:true`)**; **Sprach-Eingang +über WebSocket (`/ws/voice`: Audio rein → STT → Antwort-Pipeline)**; automatisierte Tests. **Platzhalter (Gerüst):** Audio-Endpunkte (`local-default`, `bluetooth`, `mobile-ws`, `mobile-webrtc`) liefern leere Chunks — nur Auswahl/Lifecycle sind @@ -203,7 +204,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`)** und **Audio-Streaming (chunked TTS satzweise, `audio_stream:true`)** sind umgesetzt. Offen: **Audio-Eingang/Streaming-STT**, **Barge-in/Turn-Manager**, **WebRTC**. +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**. 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. diff --git a/README.md b/README.md index 9ac9288..754d56d 100644 --- a/README.md +++ b/README.md @@ -88,7 +88,8 @@ Aktive Konfiguration prüfen: `curl http://localhost:8080/api/config`. | `GET /api/me` | aktueller Nutzer + Präferenzen | | `PUT /api/me/prefs` | dauerhafte Routing-Präferenzen des Nutzers setzen | | `GET/POST/DELETE /api/me/memories` | Langzeit-Erinnerungen des Nutzers verwalten | -| `WS /ws/chat` | Echtzeit-Chat über WebSocket (Streaming-Events) | +| `WS /ws/chat` | Echtzeit-Chat über WebSocket (Text rein, Streaming-Events) | +| `WS /ws/voice` | Echtzeit-Sprache (Audio rein → Transkript → Antwort) | Beispiel (Sprachausgabe an den Test-Loopback, lokaler TTS-Stub): @@ -142,8 +143,14 @@ via SSE; Provider ohne Streaming liefern die komplette Antwort als ein `token`-E mit `seq` + binärer Frame) gesendet — die Ausgabe beginnt, bevor die Antwort fertig ist. `stream` und `audio_stream` lassen sich kombinieren. -> Audio-Eingang/Streaming-STT, Barge-in/Turn-Manager und WebRTC sind als nächste -> Increments vorgesehen (siehe Architektur-Dokument). +**Sprach-Eingang (`/ws/voice`):** Der Client streamt Mikrofon-Audio als binäre +Frames; ein `{"type":"end"}`-Control-Frame schließt die Äußerung ab. Der Server +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). ## Authentifizierung diff --git a/app/api/ws.py b/app/api/ws.py index 630b564..960c549 100644 --- a/app/api/ws.py +++ b/app/api/ws.py @@ -1,14 +1,18 @@ -"""WebSocket-Streaming-Chat (Echtzeit-Transport, erster Increment). +"""WebSocket-Echtzeit-Chat und -Sprache. -Etabliert einen dauerhaften, bidirektionalen Kanal: der Client schickt pro Turn -eine JSON-Nachricht, der Server streamt strukturierte Events zurueck -(ack -> semantic -> audio (binaer) -> done). Auth, Session-Gedaechtnis und -Langzeit-Erinnerungen gelten wie bei POST /api/chat. +- /ws/chat : Text rein (JSON pro Turn), Antwort als Event-Folge zurueck. +- /ws/voice: Audio rein (binaere Chunks + Control), Transkription -> selbe Pipeline. -Bewusst spaeter (eigene Increments): Token-Level-LLM-Streaming, Audio-Eingang/ -Streaming-STT, Barge-in/Interrupt und WebRTC. +Event-Folge der Antwort: 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. """ +import json + from fastapi import APIRouter, WebSocket, WebSocketDisconnect from app.config import settings @@ -23,6 +27,15 @@ from app.store import SessionOwnershipError router = APIRouter() +_OVERRIDE_KEYS = ( + "input_endpoint", + "output_endpoint", + "language", + "stt_provider", + "llm_provider", + "tts_provider", +) + def _authenticate(token: str | None): store = get_store() @@ -33,18 +46,92 @@ def _authenticate(token: str | None): return store.get_user_by_token(token) -@router.websocket("/ws/chat") -async def ws_chat( - websocket: WebSocket, - session_id: str | None = None, - token: str | None = None, -): - user = _authenticate(token) - if user is None: - # Vor accept() schliessen -> Handshake wird mit 403 abgelehnt. - await websocket.close(code=1008) +async def _resolve(user, session_id, options): + """Loest Route + Orchestrator + Output-Endpunkt auf (kann RoutingError/Ownership werfen).""" + overrides = {key: options.get(key) for key in _OVERRIDE_KEYS} + route = resolve_route(user, session_id, overrides) + orchestrator = build_orchestrator(route) + output = await resolve_output_endpoint(route) + return route, orchestrator, output + + +async def _run_turn(websocket, store, user, session_id, route, orchestrator, output, text, options): + """Faehrt einen Antwort-Turn und streamt die Events an den Client.""" + conversation = ( + store.get_recent_messages(session_id, settings.history_max_messages) + if session_id + else [] + ) + memories = store.get_memories(user.id) + llm_context = list(conversation) + if memories: + memory_text = "Was du ueber den Nutzer weisst:\n" + "\n".join( + f"- {m.content}" for m in memories + ) + llm_context = [{"role": "system", "content": memory_text}] + llm_context + + await websocket.send_json({"type": "ack", "route": route.as_dict()}) + + voice = options.get("voice") or settings.openrouter_tts_voice + stream = bool(options.get("stream")) + audio_stream = bool(options.get("audio_stream")) + + on_token = None + if stream: + async def on_token(delta): + await websocket.send_json({"type": "token", "text": delta}) + + on_audio = None + if audio_stream: + audio_seq = 0 + + async def on_audio(chunk): + nonlocal audio_seq + await websocket.send_json({"type": "audio", "seq": audio_seq}) + audio_seq += 1 + await websocket.send_bytes(chunk) + + try: + if stream or audio_stream: + trace, audio = await orchestrator.chat_stream( + text, + language=route.language, + voice=voice, + output=output, + history=llm_context, + on_token=on_token, + on_audio=on_audio, + ) + else: + trace, audio = await orchestrator.chat_text( + text, + language=route.language, + voice=voice, + output=output, + history=llm_context, + ) + except Exception as exc: + await websocket.send_json({"type": "error", "status": 502, "detail": str(exc)}) return + if session_id: + store.append_message(session_id, user.id, "user", text) + store.append_message(session_id, user.id, "assistant", trace.semantic_response) + + await websocket.send_json( + {"type": "semantic", "text": trace.semantic_response, "spoken": trace.spoken_response} + ) + if not audio_stream: + await websocket.send_bytes(audio) + await websocket.send_json({"type": "done", "audio_format": "pcm", "sample_rate": 24000}) + + +@router.websocket("/ws/chat") +async def ws_chat(websocket: WebSocket, session_id: str | None = None, token: str | None = None): + user = _authenticate(token) + if user is None: + await websocket.close(code=1008) + return await websocket.accept() store = get_store() @@ -55,103 +142,91 @@ async def ws_chat( if not text: await websocket.send_json({"type": "error", "detail": "empty text"}) continue - - overrides = { - key: msg.get(key) - for key in ( - "input_endpoint", - "output_endpoint", - "language", - "stt_provider", - "llm_provider", - "tts_provider", - ) - } - try: - route = resolve_route(user, session_id, overrides) - orchestrator = build_orchestrator(route) - output = await resolve_output_endpoint(route) - conversation = ( - store.get_recent_messages(session_id, settings.history_max_messages) - if session_id - else [] - ) - memories = store.get_memories(user.id) + 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) + except WebSocketDisconnect: + return - llm_context = list(conversation) - if memories: - memory_text = "Was du ueber den Nutzer weisst:\n" + "\n".join( - f"- {m.content}" for m in memories - ) - llm_context = [{"role": "system", "content": memory_text}] + llm_context - await websocket.send_json({"type": "ack", "route": route.as_dict()}) +@router.websocket("/ws/voice") +async def ws_voice(websocket: WebSocket, session_id: str | None = None, token: str | None = None): + user = _authenticate(token) + if user is None: + await websocket.close(code=1008) + return + await websocket.accept() + store = get_store() - voice = msg.get("voice") or settings.openrouter_tts_voice - stream = bool(msg.get("stream")) - audio_stream = bool(msg.get("audio_stream")) + audio_buffer = bytearray() + fmt = "wav" - on_token = None - if stream: - async def on_token(delta): - await websocket.send_json({"type": "token", "text": delta}) + try: + while True: + message = await websocket.receive() + if message["type"] == "websocket.disconnect": + return - on_audio = None - if audio_stream: - audio_seq = 0 - - async def on_audio(chunk): - nonlocal audio_seq - await websocket.send_json({"type": "audio", "seq": audio_seq}) - audio_seq += 1 - await websocket.send_bytes(chunk) - - try: - if stream or audio_stream: - trace, audio = await orchestrator.chat_stream( - text, - language=route.language, - voice=voice, - output=output, - history=llm_context, - on_token=on_token, - on_audio=on_audio, - ) - else: - trace, audio = await orchestrator.chat_text( - text, - language=route.language, - voice=voice, - output=output, - history=llm_context, - ) - except Exception as exc: - await websocket.send_json({"type": "error", "status": 502, "detail": str(exc)}) + if message.get("bytes") is not None: + audio_buffer.extend(message["bytes"]) continue - if session_id: - store.append_message(session_id, user.id, "user", text) - store.append_message(session_id, user.id, "assistant", trace.semantic_response) + raw = message.get("text") + if raw is None: + continue + try: + control = json.loads(raw) + except ValueError: + await websocket.send_json({"type": "error", "detail": "invalid control frame"}) + continue - await websocket.send_json( - { - "type": "semantic", - "text": trace.semantic_response, - "spoken": trace.spoken_response, - } - ) - # Bei audio_stream wurden die Audio-Chunks bereits live gesendet. - if not audio_stream: - await websocket.send_bytes(audio) - await websocket.send_json( - {"type": "done", "audio_format": "pcm", "sample_rate": 24000} + ctype = control.get("type") + if ctype == "start": + audio_buffer.clear() + fmt = control.get("format", "wav") + continue + if ctype != "end": + continue + + if not audio_buffer: + 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 ) except WebSocketDisconnect: return diff --git a/tests/test_ws.py b/tests/test_ws.py index 082e0fa..5619aa7 100644 --- a/tests/test_ws.py +++ b/tests/test_ws.py @@ -66,3 +66,51 @@ def test_ws_requires_token_when_auth_enabled(monkeypatch): with pytest.raises(WebSocketDisconnect): with client.websocket_connect("/ws/chat"): pass + + +def _install_voice_stubs(monkeypatch): + class StubSTT: + async def transcribe(self, audio_bytes, fmt, language=None): + return f"erkannt({len(audio_bytes)})" + + class StubLLM: + async def complete(self, text, history=None, session_id=None): + return f"Antwort zu {text}" + + class StubTTS: + async def synthesize(self, text, voice=None, audio_format="pcm"): + return b"VOICEAUD" + + monkeypatch.setitem(deps.STT_REGISTRY, "ss", lambda s: StubSTT()) + monkeypatch.setitem(deps.LLM_REGISTRY, "ll", lambda s: StubLLM()) + monkeypatch.setitem(deps.TTS_REGISTRY, "tt", lambda s: StubTTS()) + return { + "stt_provider": "ss", + "llm_provider": "ll", + "tts_provider": "tt", + "output_endpoint": "loopback", + } + + +def test_ws_voice_transcribes_and_answers(monkeypatch): + opts = _install_voice_stubs(monkeypatch) + with client.websocket_connect("/ws/voice?session_id=v1") as ws: + ws.send_bytes(b"PCMDATA") # 7 Bytes + ws.send_bytes(b"MORE") # 4 Bytes -> insgesamt 11 + ws.send_json({"type": "end", **opts}) + + transcript = ws.receive_json() + assert transcript["type"] == "transcript" and transcript["text"] == "erkannt(11)" + assert ws.receive_json()["type"] == "ack" + semantic = ws.receive_json() + assert semantic["type"] == "semantic" and semantic["text"] == "Antwort zu erkannt(11)" + assert ws.receive_bytes() == b"VOICEAUD" + assert ws.receive_json()["type"] == "done" + + +def test_ws_voice_empty_buffer_errors(monkeypatch): + opts = _install_voice_stubs(monkeypatch) + with client.websocket_connect("/ws/voice") as ws: + ws.send_json({"type": "end", **opts}) # kein Audio gesendet + err = ws.receive_json() + assert err["type"] == "error" and "no audio" in err["detail"]