""" Voice-Endpoints für „Mit Hermes reden" (Browser-Voice + 3D-Avatar). Dünner Layer: STT/TTS werden zum Voice-Sidecar (:8650) geproxyt; der Chat geht an den Hermes-`api_server` (:8642, OpenAI-kompatibel) — denselben vollen Agenten mit Tools + geteiltem Mem0 wie CLI/Telegram. Mit stabilem `X-Hermes-Session-Id` hält die Plattform den Transcript server-seitig, daher schickt der Client je Turn nur die neue User-Nachricht. LAN-only (kein Token in der 2.0-Phase), wie die übrigen MC2-Endpoints. """ import logging import os import time import httpx from fastapi import APIRouter, File, Form, HTTPException, UploadFile from fastapi.responses import Response, StreamingResponse from pydantic import BaseModel from config import HERMES_API_KEY, HERMES_API_MODEL, HERMES_API_URL, LLAMA_SWAP_URL, VOICE_SERVICE_URL from services import announce from services.voice_metrics import Timer, get_metrics, record_stage # Per-Stage-Latenz (C2) # Injection-Schutz (Stufe 0): guard.py liegt im mcp/-Verzeichnis. Per Pfad laden (eigene MC2-Venv). import sys as _sys _GUARD_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "mcp") if _GUARD_DIR not in _sys.path: _sys.path.insert(0, _GUARD_DIR) try: from guard import wrap_untrusted except Exception: # den Voice-Pfad nie wegen des Filters lahmlegen def wrap_untrusted(text: str, label: str = "") -> str: return text log = logging.getLogger(__name__) router = APIRouter(prefix="/api") # Bildschirm-Sicht: das DEDIZIERTE Vision-Modell (Qwen3-VL-8B) beschreibt das Bild; die Beschreibung # geht als TEXT an Hermes -> Lucy behält ihr volles Hirn/Gedächtnis UND nutzt das bessere VL-Modell # (statt der schwächeren Vision der fast-MoE). Per Env abschaltbar/umstellbar. VISION_MODEL = os.environ.get("MC_VISION_MODEL", "vision") # Knappe Beschreibung = schnellere VL-Generierung UND weniger Hermes-Kontext-Bloat (B2). VISION_MAX_TOKENS = int(os.environ.get("MC_VISION_MAX_TOKENS", "280")) async def _describe_images(image_urls: list[str], hint: str) -> str: """Lässt das Vision-Modell die Screenshots (1 je Monitor) knapp beschreiben (Deutsch). Mehrere Bilder gehen in EINER Nachricht ans VL-Modell. Leerer String bei Fehler.""" multi = len(image_urls) > 1 intro = (f"Hier sind {len(image_urls)} Screenshots (je ein Monitor). Beschreibe auf Deutsch in höchstens " "5 kurzen Sätzen das Wesentliche (pro Monitor: App/Fenster, wichtige Inhalte, sichtbarer Text/Code). " "Keine Einleitung, keine Wiederholung der Frage. " if multi else "Beschreibe auf Deutsch in höchstens 5 kurzen Sätzen das Wesentliche auf diesem Screenshot " "(App/Fenster, wichtige Inhalte, sichtbarer Text/Code). Keine Einleitung. ") content: list = [{"type": "text", "text": intro + "Frage des Nutzers dazu: " + hint}] for u in image_urls: content.append({"type": "image_url", "image_url": {"url": u}}) try: # 45 s statt 120 s: Qwen3-VL braucht warm ~5 s; wenn es 45 s nicht schafft, ist etwas # kaputt und Lucy soll lieber ohne Bildschirm-Kontext antworten als ewig hängen. async with httpx.AsyncClient(timeout=httpx.Timeout(float(os.environ.get("MC_VISION_TIMEOUT", "45")), connect=5.0)) as client: r = await client.post(f"{LLAMA_SWAP_URL}/v1/chat/completions", json={ "model": VISION_MODEL, "max_tokens": VISION_MAX_TOKENS, "stream": False, "messages": [{"role": "user", "content": content}], }) r.raise_for_status() return (r.json().get("choices") or [{}])[0].get("message", {}).get("content", "").strip() except Exception as exc: log.warning("Vision-Beschreibung fehlgeschlagen: %s", exc) return "" _TIMEOUT = httpx.Timeout(120.0, connect=5.0) # Chatterbox-TTS auf CPU darf dauern class TTSIn(BaseModel): text: str engine: str = "piper" voice: str = "" language: str = "" ref_path: str = "" class ChatIn(BaseModel): text: str # die neue User-Äußerung (STT-Ergebnis) session_id: str # stabiler Voice-Faden → server-seitiger Transcript session_key: str = "" # optional: Langzeit-Memory-Scope system: str = "" # optionaler ephemerer System-Prompt (z.B. „antworte knapp/gesprochen") model: str = "" images: list[str] = [] # optionale Bildschirm-Sicht: ein data:-URL je Monitor (Lucys „Augen") class AnnounceIn(BaseModel): text: str # die Meldung (wird von Lucy gesprochen) subject: str = "" # kurze Betreffzeile (z.B. "[Update]") source: str = "" # Absender (sentry/notify/cron …) — nur fürs Log/Panel priority: str = "normal" # 'silent' = nur im Verlauf zeigen, nicht sprechen class AlarmIn(BaseModel): text: str # die Alarm-Meldung subject: str = "[Alarm]" # Betreff (Telegram-Präfix) source: str = "alarm" # Absender fürs Log/Panel (z.B. "lucy-watchdog") @router.post("/alarm") def alarm(body: AlarmIn) -> dict: """Lucy-UNABHÄNGIGER Alarm-Weg: schickt direkt auf Telegram (und legt die Meldung in den Briefkasten). Für Absender, die NICHT auf die sprechende Lucy zählen können — allen voran der PC-seitige Lucy-Watchdog, wenn die Desktop-App selbst hängt (dann nützt der Briefkasten nichts, weil niemand ihn vorliest → Telegram ist der einzige verlässliche Kanal). LAN-only wie alle MC2-Endpoints.""" text = (body.text or "").strip() if not text: raise HTTPException(400, "Leere Meldung.") subject = (body.subject or "[Alarm]").strip() try: item = announce.add(text, subject, body.source or "alarm", "normal") except ValueError as exc: raise HTTPException(400, str(exc)) announce.notify_telegram(subject, text) # best-effort Telegram (posix/bash; Windows = No-op) return {"ok": True, "item": item} @router.post("/voice/announce") def voice_announce(body: AnnounceIn) -> dict: """Meldung in den Briefkasten legen (Lucy-Proaktivität). Absender: Health-Wächter, notify.sh (Updates/Radar/Telegram-Spiegel), Hermes-cron. LAN-only wie alle MC2-Endpoints.""" try: return {"ok": True, "item": announce.add(body.text, body.subject, body.source, body.priority)} except ValueError as exc: raise HTTPException(400, str(exc)) @router.get("/voice/announcements") def voice_announcements(after: int | None = None, limit: int = 20) -> dict: """Neue Meldungen nach Cursor `after` abholen (Lucy pollt). Ohne `after` nur den aktuellen Cursor-Stand (latest) — Erststart plappert so keine alten Meldungen nach.""" return announce.list_after(after, limit) @router.get("/voice/metrics") def voice_metrics() -> dict: """Per-Stage-Latenz (STT/Vision/Chat-TTFB/TTS) — rollende Statistik, macht die Voice-Pipeline messbar (C2). Anzeige im Frontend-Overhaul (E).""" return get_metrics() @router.get("/voice/health") def voice_health() -> dict: """Erreichbarkeit des Voice-Sidecars + ob der Hermes-API-Key gesetzt ist.""" out: dict = {"sidecar": False, "hermes_key": bool(HERMES_API_KEY)} try: r = httpx.get(f"{VOICE_SERVICE_URL}/health", timeout=httpx.Timeout(5.0)) out["sidecar"] = r.status_code == 200 out["detail"] = r.json() if r.status_code == 200 else None except Exception as exc: # noqa: BLE001 out["error"] = str(exc) return out @router.get("/voice/voices") def voice_voices() -> dict: try: r = httpx.get(f"{VOICE_SERVICE_URL}/voices", timeout=httpx.Timeout(10.0)) r.raise_for_status() return r.json() except Exception as exc: # noqa: BLE001 raise HTTPException(502, f"Voice-Sidecar nicht erreichbar: {exc}") @router.post("/voice/stt") async def voice_stt(audio: UploadFile = File(...), language: str = Form(default="")) -> dict: """Mikro-Audio → Text (Proxy auf Sidecar /stt).""" data = await audio.read() if not data: raise HTTPException(400, "Leeres Audio.") files = {"audio": (audio.filename or "rec.webm", data, audio.content_type or "audio/webm")} try: async with httpx.AsyncClient(timeout=_TIMEOUT) as client: with Timer("stt"): r = await client.post(f"{VOICE_SERVICE_URL}/stt", files=files, data={"language": language}) r.raise_for_status() return r.json() except httpx.HTTPError as exc: raise HTTPException(502, f"STT fehlgeschlagen: {exc}") @router.post("/voice/turn") async def voice_turn(audio: UploadFile = File(...)) -> dict: """Semantische Turn-Detection (Smart Turn v3): war die Äußerung fertig? Proxy → Sidecar.""" data = await audio.read() if not data: raise HTTPException(400, "Leeres Audio.") files = {"audio": (audio.filename or "rec.wav", data, audio.content_type or "audio/wav")} try: async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=3.0)) as client: with Timer("turn"): r = await client.post(f"{VOICE_SERVICE_URL}/turn", files=files) r.raise_for_status() return r.json() except httpx.HTTPError as exc: # Turn-Check ist eine Optimierung — bei Ausfall lieber sofort antworten als hängen. log.warning("Turn-Check fehlgeschlagen: %s", exc) return {"complete": True, "probability": 1.0, "engine": "fallback"} @router.post("/voice/reference") async def voice_set_reference(audio: UploadFile = File(...)) -> dict: """Klon-Referenz (z.B. ElevenLabs-Erzeugnis) hochladen → Chatterbox nutzt sie. Proxy → Sidecar.""" data = await audio.read() if not data: raise HTTPException(400, "Leeres Audio.") files = {"audio": (audio.filename or "ref.wav", data, audio.content_type or "audio/mpeg")} try: async with httpx.AsyncClient(timeout=_TIMEOUT) as client: r = await client.post(f"{VOICE_SERVICE_URL}/reference", files=files) r.raise_for_status() return r.json() except httpx.HTTPError as exc: raise HTTPException(502, f"Referenz-Upload fehlgeschlagen: {exc}") @router.get("/voice/reference") def voice_get_reference() -> dict: try: r = httpx.get(f"{VOICE_SERVICE_URL}/reference", timeout=httpx.Timeout(8.0)) r.raise_for_status() return r.json() except Exception as exc: # noqa: BLE001 return {"active": False, "error": str(exc)} @router.delete("/voice/reference") def voice_clear_reference() -> dict: try: r = httpx.delete(f"{VOICE_SERVICE_URL}/reference", timeout=httpx.Timeout(8.0)) r.raise_for_status() return r.json() except httpx.HTTPError as exc: raise HTTPException(502, f"Löschen fehlgeschlagen: {exc}") @router.post("/voice/tts") async def voice_tts(body: TTSIn) -> Response: """Text → Sprache (Proxy auf Sidecar /tts), liefert WAV-Bytes.""" try: async with httpx.AsyncClient(timeout=_TIMEOUT) as client: with Timer("tts"): r = await client.post(f"{VOICE_SERVICE_URL}/tts", json=body.model_dump()) r.raise_for_status() return Response(content=r.content, media_type=r.headers.get("content-type", "audio/wav")) except httpx.HTTPError as exc: raise HTTPException(502, f"TTS fehlgeschlagen: {exc}") @router.post("/voice/chat") async def voice_chat(body: ChatIn) -> StreamingResponse: """Neue User-Äußerung → Hermes-Agent (api_server, streamend). SSE wird 1:1 durchgereicht. Mit `X-Hermes-Session-Id` hält die Plattform den Verlauf — wir senden nur die neue Nachricht. Auth per Bearer (API_SERVER_KEY); ohne Key liefert :8642 ein 401.""" if not HERMES_API_KEY: raise HTTPException(503, "HERMES_API_KEY/API_SERVER_KEY nicht gesetzt — Agent-Auth fehlt.") headers = { "Authorization": f"Bearer {HERMES_API_KEY}", "X-Hermes-Session-Id": body.session_id, } if body.session_key: headers["X-Hermes-Session-Key"] = body.session_key async def gen(): t0 = time.perf_counter() first = True first_content = True # Bildschirm-Sicht INNERHALB des Streams (C2-Fix): so startet die SSE-Antwort sofort und # der Client bekommt ein Progress-Event (-> Lucy kann eine Warte-Ansage sprechen), statt # dass der Request bis zu 120 s "tot" hängt, während das Vision-Modell beschreibt. user_text = body.text imgs = [u for u in (body.images or []) if u] if imgs: yield b'event: hermes.vision.progress\ndata: {"note": "Bildschirm wird angeschaut"}\n\n' with Timer("vision"): desc = await _describe_images(imgs, body.text) if desc: safe_desc = wrap_untrusted(desc, "BILDSCHIRM") user_text = f"[Bildschirm-Sicht — das ist gerade auf dem/den Schirm(en) zu sehen:\n{safe_desc}\n]\n\n{body.text}" messages = [] if body.system: messages.append({"role": "system", "content": body.system}) messages.append({"role": "user", "content": user_text}) payload = {"model": body.model or HERMES_API_MODEL, "messages": messages, "stream": True} # Lucys Hirn (Qwen3.6) ist ein Thinking-Modell -> für die gesprochene Assistentin Thinking AUS, # sonst generiert es tausende Reasoning-Token VOR der kurzen Antwort (gemessen: 11k Token, ~30s TTFB). # Gleiches Muster wie die fast-Spur im Gateway (gateway_proxy.py) und die Mem0-Extraktion. if os.environ.get("MC_VOICE_NO_THINK", "1") not in ("0", "false", "False"): payload["chat_template_kwargs"] = {"enable_thinking": False} try: async with httpx.AsyncClient(timeout=httpx.Timeout(None, connect=5.0)) as client: async with client.stream( "POST", f"{HERMES_API_URL}/v1/chat/completions", json=payload, headers=headers, ) as r: if r.status_code != 200: detail = (await r.aread()).decode("utf-8", "replace")[:500] yield f"data: {{\"error\": \"Hermes {r.status_code}: {detail}\"}}\n\n".encode() return async for chunk in r.aiter_raw(): if first: # Time-To-First-Byte des Hermes-Streams (Verbindungs-Overhead) record_stage("chat_ttfb", (time.perf_counter() - t0) * 1000.0) first = False # Erster CONTENT-Delta = echte Hirn-Latenz (Agent-Overhead + LLM-TTFT) — # chat_ttfb misst nur den SSE-Start (~5 ms) und ist dafür blind. if first_content and b'"content"' in chunk: record_stage("chat_first_content", (time.perf_counter() - t0) * 1000.0) first_content = False yield chunk except httpx.HTTPError as exc: yield f"data: {{\"error\": \"Verbindung zu Hermes fehlgeschlagen: {exc}\"}}\n\n".encode() return StreamingResponse(gen(), media_type="text/event-stream")