From 379e002460b463d561b5a8b576161ccff4399d79 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dieter=20Schl=C3=BCter?= Date: Wed, 17 Jun 2026 04:37:37 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20Token-Level-LLM-Streaming=20=C3=BCber?= =?UTF-8?q?=20WebSocket=20(#4=20Ausbau)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - LLMProvider.stream: Basis-Default (Fallback über complete) + SSE-Streaming fuer OpenRouter und lokalen OpenAI-kompatiblen Provider; gemeinsamer Parser sse_delta - Orchestrator.chat_stream: LLM-Token live via on_token-Callback, danach Spoken-Adapter/Normalizer/TTS/Output; Fallback fuer Provider ohne stream - WS /ws/chat: opt-in {"stream":true} -> ack -> token* -> semantic -> audio -> done - Tests: 43 gruen (+5: SSE-Parsing, Default-Fallback, WS-Token-Flow) - Doku aktualisiert; .gitignore: *.wav (generierte Audio-Ausgaben) Co-Authored-By: Claude Opus 4.8 --- .gitignore | 3 + BEDIENUNGSANLEITUNG.md | 4 +- Docs/voice-assistant-architecture.md | 5 +- README.md | 10 ++- app/api/ws.py | 28 ++++-- app/core/orchestrator.py | 49 ++++++++++ app/providers/llm/base.py | 29 ++++++ app/providers/llm/local_openai_compatible.py | 45 +++++++++- app/providers/llm/openrouter.py | 59 ++++++++++-- tests/test_streaming.py | 94 ++++++++++++++++++++ 10 files changed, 302 insertions(+), 24 deletions(-) create mode 100644 tests/test_streaming.py diff --git a/.gitignore b/.gitignore index 5d1581d..fdc2a92 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,9 @@ config/voice-assistant.toml # Persistente Daten (SQLite-DB etc.) data/ +# Generierte Audio-Ausgaben (z. B. chat_client.py) +*.wav + # Python __pycache__/ *.py[cod] diff --git a/BEDIENUNGSANLEITUNG.md b/BEDIENUNGSANLEITUNG.md index 5a613e6..d49d4ff 100644 --- a/BEDIENUNGSANLEITUNG.md +++ b/BEDIENUNGSANLEITUNG.md @@ -272,7 +272,9 @@ Diese Erinnerungen gibt der Assistent bei jedem Chat als Kontext mit — auch oh **Echtzeit-Chat über WebSocket** (`/ws/chat`): dauerhafter Kanal, pro Nachricht `{"text": "..."}`; Antwort kommt als Event-Folge (`ack`, `semantic`, Audio, `done`). -Token per Query (`?token=…`), Gedächtnis per `?session_id=…`. +Token per Query (`?token=…`), Gedächtnis per `?session_id=…`. Mit +`{"text": "...", "stream": true}` kommt die Antwort schon während der Generierung +als `token`-Events (geringere wahrgenommene Latenz). > **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 a13744d..da60379 100644 --- a/Docs/voice-assistant-architecture.md +++ b/Docs/voice-assistant-architecture.md @@ -186,7 +186,8 @@ Device Router (strikt, Singleton); Output-Lifecycle; **Authentifizierung (Bearer-Token) + persistenter SQLite-Store für Nutzer/Sessions + Mandanten-Trennung + 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`)**; automatisierte Tests. +**WebSocket-Streaming-Chat (`/ws/chat`) inkl. Token-Level-LLM-Streaming (SSE, +opt-in via `stream:true`)**; automatisierte Tests. **Platzhalter (Gerüst):** Audio-Endpunkte (`local-default`, `bluetooth`, `mobile-ws`, `mobile-webrtc`) liefern leere Chunks — nur Auswahl/Lifecycle sind @@ -201,7 +202,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`) mit Event-Folge (ack/semantic/audio/done) ist umgesetzt. Offen: **Token-Level-LLM-Streaming**, **Audio-Eingang/Streaming-STT**, **Barge-in/Turn-Manager**, **WebRTC**. +4. **(teilweise erledigt)** Echtzeit: WebSocket-Streaming-Chat (`/ws/chat`) mit Event-Folge (ack/semantic/audio/done) **und Token-Level-LLM-Streaming (SSE, opt-in `stream:true`)** sind umgesetzt. Offen: **Audio-Streaming (chunked TTS)**, **Audio-Eingang/Streaming-STT**, **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 4ca0bcc..74be773 100644 --- a/README.md +++ b/README.md @@ -132,8 +132,14 @@ der Server streamt strukturierte Events zurück: `ack` → `semantic` → Audio → `done`. Auth (Token-Query `?token=…`), Session-Gedächtnis (`?session_id=…`) und Erinnerungen gelten wie bei `POST /api/chat`. -> Token-Level-LLM-Streaming, Audio-Eingang/Streaming-STT, Barge-in und WebRTC sind -> als nächste Increments vorgesehen (siehe Architektur-Dokument). +**Token-Streaming:** Mit `{"text": "...", "stream": true}` schickt der Server die +LLM-Antwort schon während der Generierung als `token`-Events +(`ack` → `token*` → `semantic` → Audio → `done`) — spürbar geringere wahrgenommene +Latenz. OpenRouter und der lokale OpenAI-kompatible Provider streamen via SSE; +Provider ohne Streaming liefern die komplette Antwort als ein `token`-Event. + +> Audio-Streaming (chunked TTS), Audio-Eingang/Streaming-STT, Barge-in 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 6c505bc..6570a7f 100644 --- a/app/api/ws.py +++ b/app/api/ws.py @@ -95,14 +95,28 @@ async def ws_chat( await websocket.send_json({"type": "ack", "route": route.as_dict()}) voice = msg.get("voice") or settings.openrouter_tts_voice + stream = bool(msg.get("stream")) try: - trace, audio = await orchestrator.chat_text( - text, - language=route.language, - voice=voice, - output=output, - history=llm_context, - ) + if stream: + async def on_token(delta): + await websocket.send_json({"type": "token", "text": delta}) + + trace, audio = await orchestrator.chat_stream( + text, + language=route.language, + voice=voice, + output=output, + history=llm_context, + on_token=on_token, + ) + 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)}) continue diff --git a/app/core/orchestrator.py b/app/core/orchestrator.py index 213fe3f..de4cfb6 100644 --- a/app/core/orchestrator.py +++ b/app/core/orchestrator.py @@ -105,3 +105,52 @@ class Orchestrator: ) await self._emit_to_output(audio, output) return trace, audio + + async def chat_stream( + self, + text: str, + language: str | None = None, + voice: str | None = None, + output=None, + history: list[dict] | None = None, + on_token=None, + ): + """Wie chat_text, aber die LLM-Antwort wird tokenweise gestreamt. + + `on_token(delta)` (async) wird pro Token-Delta aufgerufen. Audio/Output + werden erst nach der vollstaendigen Antwort erzeugt (TTS ist nicht streamend). + """ + trace = PipelineTrace() + trace.raw_transcript = text + trace.cleaned_transcript = await self.input_cleaner.run(text or "") + + parts: list[str] = [] + stream_fn = getattr(self.llm, "stream", None) + if stream_fn is not None: + async for delta in stream_fn(trace.cleaned_transcript or "", history=history): + parts.append(delta) + if on_token: + await on_token(delta) + else: + # Provider ohne Streaming -> komplette Antwort als ein Token. + result = await self.llm.complete(trace.cleaned_transcript or "", history=history) + parts.append(result) + if on_token: + await on_token(result) + + trace.semantic_response = "".join(parts) + if not trace.semantic_response: + raise RuntimeError("LLM returned an empty response") + + trace.spoken_response = await self.spoken_adapter.run( + trace.semantic_response, + language=language, + ) + trace.tts_ready_text = await self.tts_normalizer.run( + trace.spoken_response, + language=language, + ) + + audio = await self.tts.synthesize(trace.tts_ready_text, voice=voice) + await self._emit_to_output(audio, output) + return trace, audio diff --git a/app/providers/llm/base.py b/app/providers/llm/base.py index 5a72add..2721b00 100644 --- a/app/providers/llm/base.py +++ b/app/providers/llm/base.py @@ -1,4 +1,21 @@ +import json from abc import ABC, abstractmethod +from collections.abc import AsyncIterator + + +def sse_delta(line: str) -> str | None: + """Extrahiert das Token-Delta aus einer OpenAI-kompatiblen SSE-Zeile (oder None).""" + if not line.startswith("data:"): + return None + data = line[len("data:"):].strip() + if not data or data == "[DONE]": + return None + try: + obj = json.loads(data) + return obj["choices"][0]["delta"].get("content") + except (ValueError, KeyError, IndexError, TypeError): + return None + class LLMProvider(ABC): @abstractmethod @@ -8,3 +25,15 @@ class LLMProvider(ABC): history: list[dict] | None = None, session_id: str | None = None, ) -> str: ... + + async def stream( + self, + text: str, + history: list[dict] | None = None, + session_id: str | None = None, + ) -> AsyncIterator[str]: + """Token-Stream. Default: kein echtes Streaming -> komplette Antwort als ein Chunk. + + Provider mit SSE-Unterstuetzung ueberschreiben diese Methode. + """ + yield await self.complete(text, history=history, session_id=session_id) diff --git a/app/providers/llm/local_openai_compatible.py b/app/providers/llm/local_openai_compatible.py index 99d946b..d0bbad1 100644 --- a/app/providers/llm/local_openai_compatible.py +++ b/app/providers/llm/local_openai_compatible.py @@ -1,5 +1,9 @@ +from collections.abc import AsyncIterator + import httpx -from app.providers.llm.base import LLMProvider + +from app.providers.llm.base import LLMProvider, sse_delta + class LocalOpenAICompatibleLLM(LLMProvider): def __init__(self, base_url: str, api_key: str, model: str): @@ -7,17 +11,20 @@ class LocalOpenAICompatibleLLM(LLMProvider): self.api_key = api_key self.model = model + def _build_messages(self, text: str, history: list[dict] | None) -> list[dict]: + messages = list(history) if history else [] + messages.append({"role": "user", "content": text}) + return messages + async def complete( self, text: str, history: list[dict] | None = None, session_id: str | None = None, ) -> str: - messages = list(history) if history else [] - messages.append({"role": "user", "content": text}) payload = { "model": self.model, - "messages": messages, + "messages": self._build_messages(text, history), "temperature": 0.3, } async with httpx.AsyncClient(timeout=120) as client: @@ -29,3 +36,33 @@ class LocalOpenAICompatibleLLM(LLMProvider): response.raise_for_status() data = response.json() return data["choices"][0]["message"]["content"] + + async def stream( + self, + text: str, + history: list[dict] | None = None, + session_id: str | None = None, + ) -> AsyncIterator[str]: + payload = { + "model": self.model, + "messages": self._build_messages(text, history), + "temperature": 0.3, + "stream": True, + } + async with httpx.AsyncClient(timeout=120) as client: + async with client.stream( + "POST", + f"{self.base_url}/chat/completions", + headers={"Authorization": f"Bearer {self.api_key}"}, + json=payload, + ) as response: + if response.status_code >= 400: + body = await response.aread() + raise RuntimeError( + f"Local LLM error {response.status_code}: " + f"{body.decode(errors='replace')}" + ) + async for line in response.aiter_lines(): + delta = sse_delta(line) + if delta: + yield delta diff --git a/app/providers/llm/openrouter.py b/app/providers/llm/openrouter.py index c4143dd..369140e 100644 --- a/app/providers/llm/openrouter.py +++ b/app/providers/llm/openrouter.py @@ -1,6 +1,8 @@ +from collections.abc import AsyncIterator + import httpx -from app.providers.llm.base import LLMProvider +from app.providers.llm.base import LLMProvider, sse_delta SYSTEM_PROMPT = """ @@ -49,12 +51,7 @@ class OpenRouterLLMProvider(LLMProvider): self.api_key = (api_key or "").strip() self.model = (model or "").strip() - async def complete( - self, - text: str, - history: list[dict] | None = None, - session_id: str | None = None, - ) -> str: + def _build_messages(self, text: str, history: list[dict] | None) -> list[dict]: if not self.api_key: raise ValueError("OPENROUTER_API_KEY is empty") if not self.model: @@ -66,10 +63,17 @@ class OpenRouterLLMProvider(LLMProvider): if history: messages.extend(history) messages.append({"role": "user", "content": text.strip()}) + return messages + async def complete( + self, + text: str, + history: list[dict] | None = None, + session_id: str | None = None, + ) -> str: payload = { "model": self.model, - "messages": messages, + "messages": self._build_messages(text, history), } timeout = httpx.Timeout(connect=10.0, read=120.0, write=30.0, pool=10.0) @@ -106,3 +110,42 @@ class OpenRouterLLMProvider(LLMProvider): return str(content).strip() + async def stream( + self, + text: str, + history: list[dict] | None = None, + session_id: str | None = None, + ) -> AsyncIterator[str]: + payload = { + "model": self.model, + "messages": self._build_messages(text, history), + "stream": True, + } + timeout = httpx.Timeout(connect=10.0, read=120.0, write=30.0, pool=10.0) + + async with httpx.AsyncClient(timeout=timeout) as client: + try: + async with client.stream( + "POST", + "https://openrouter.ai/api/v1/chat/completions", + headers={ + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + }, + json=payload, + ) as response: + if response.status_code >= 400: + body = await response.aread() + raise RuntimeError( + f"OpenRouter LLM error {response.status_code}: " + f"{body.decode(errors='replace')}" + ) + async for line in response.aiter_lines(): + delta = sse_delta(line) + if delta: + yield delta + except httpx.TimeoutException as exc: + raise RuntimeError("OpenRouter LLM timeout") from exc + except httpx.HTTPError as exc: + raise RuntimeError(f"OpenRouter LLM transport error: {exc}") from exc + diff --git a/tests/test_streaming.py b/tests/test_streaming.py new file mode 100644 index 0000000..6f5a6d8 --- /dev/null +++ b/tests/test_streaming.py @@ -0,0 +1,94 @@ +import asyncio + +from fastapi.testclient import TestClient + +import app.dependencies as deps +from app.main import app +from app.providers.llm.base import sse_delta, LLMProvider + +client = TestClient(app) + + +def test_sse_delta_parsing(): + assert sse_delta('data: {"choices":[{"delta":{"content":"Hal"}}]}') == "Hal" + assert sse_delta("data: [DONE]") is None + assert sse_delta("") is None + assert sse_delta(": keep-alive") is None + assert sse_delta('data: {"choices":[{"delta":{}}]}') is None + assert sse_delta("data: nicht-json") is None + + +def test_base_stream_default_yields_full_completion(): + class P(LLMProvider): + async def complete(self, text, history=None, session_id=None): + return "ganze Antwort" + + async def run(): + return [delta async for delta in P().stream("x")] + + assert asyncio.run(run()) == ["ganze Antwort"] + + +def _install_streaming(monkeypatch, tokens): + class StreamLLM: + async def complete(self, text, history=None, session_id=None): + return "".join(tokens) + + async def stream(self, text, history=None, session_id=None): + for tok in tokens: + yield tok + + class StubTTS: + async def synthesize(self, text, voice=None, audio_format="pcm"): + return b"AUD" + + monkeypatch.setitem(deps.LLM_REGISTRY, "stream", lambda s: StreamLLM()) + monkeypatch.setitem(deps.TTS_REGISTRY, "stub", lambda s: StubTTS()) + return {"llm_provider": "stream", "tts_provider": "stub", "output_endpoint": "loopback"} + + +def test_ws_stream_emits_token_events(monkeypatch): + base = _install_streaming(monkeypatch, ["Gu", "ten ", "Tag"]) + with client.websocket_connect("/ws/chat") as ws: + ws.send_json({"text": "Hallo", "stream": True, **base}) + assert ws.receive_json()["type"] == "ack" + + tokens = [] + event = ws.receive_json() + while event["type"] == "token": + tokens.append(event["text"]) + event = ws.receive_json() + + assert tokens == ["Gu", "ten ", "Tag"] + assert event["type"] == "semantic" and event["text"] == "Guten Tag" + assert ws.receive_bytes() == b"AUD" + assert ws.receive_json()["type"] == "done" + + +def test_ws_without_stream_flag_has_no_tokens(monkeypatch): + base = _install_streaming(monkeypatch, ["a", "b"]) + with client.websocket_connect("/ws/chat") as ws: + ws.send_json({"text": "Hallo", **base}) # kein stream-Flag + assert ws.receive_json()["type"] == "ack" + assert ws.receive_json()["type"] == "semantic" # direkt, keine token-Events + + +def test_ws_stream_fallback_for_nonstreaming_llm(monkeypatch): + class OnlyComplete: + async def complete(self, text, history=None, session_id=None): + return "komplett" + + class StubTTS: + async def synthesize(self, text, voice=None, audio_format="pcm"): + return b"X" + + monkeypatch.setitem(deps.LLM_REGISTRY, "oc", lambda s: OnlyComplete()) + monkeypatch.setitem(deps.TTS_REGISTRY, "stub", lambda s: StubTTS()) + base = {"llm_provider": "oc", "tts_provider": "stub", "output_endpoint": "loopback"} + + with client.websocket_connect("/ws/chat") as ws: + ws.send_json({"text": "x", "stream": True, **base}) + assert ws.receive_json()["type"] == "ack" + token = ws.receive_json() + assert token["type"] == "token" and token["text"] == "komplett" + assert ws.receive_json()["type"] == "semantic"