Feat: Per-Stage-Latenz-Metriken für die Voice-Pipeline (C2)
Neues services/voice_metrics.py (rollend, thread-safe, in-memory): misst STT, Vision-Beschreibung, Chat-TTFB (Hermes-Stream) und TTS server-seitig. voice.py instrumentiert die vier Stufen; GET /api/voice/metrics liefert avg/p50/p95/last je Stufe. Macht aus Latenz-Vermutungen Messdaten — Anzeige folgt im Frontend (E). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -11,6 +11,7 @@ LAN-only (kein Token in der 2.0-Phase), wie die übrigen MC2-Endpoints.
|
|||||||
|
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
|
import time
|
||||||
|
|
||||||
import httpx
|
import httpx
|
||||||
from fastapi import APIRouter, File, Form, HTTPException, UploadFile
|
from fastapi import APIRouter, File, Form, HTTPException, UploadFile
|
||||||
@@ -18,6 +19,7 @@ from fastapi.responses import Response, StreamingResponse
|
|||||||
from pydantic import BaseModel
|
from pydantic import BaseModel
|
||||||
|
|
||||||
from config import HERMES_API_KEY, HERMES_API_MODEL, HERMES_API_URL, LLAMA_SWAP_URL, VOICE_SERVICE_URL
|
from config import HERMES_API_KEY, HERMES_API_MODEL, HERMES_API_URL, LLAMA_SWAP_URL, VOICE_SERVICE_URL
|
||||||
|
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).
|
# Injection-Schutz (Stufe 0): guard.py liegt im mcp/-Verzeichnis. Per Pfad laden (eigene MC2-Venv).
|
||||||
import sys as _sys
|
import sys as _sys
|
||||||
@@ -86,6 +88,13 @@ class ChatIn(BaseModel):
|
|||||||
images: list[str] = [] # optionale Bildschirm-Sicht: ein data:-URL je Monitor (Lucys „Augen")
|
images: list[str] = [] # optionale Bildschirm-Sicht: ein data:-URL je Monitor (Lucys „Augen")
|
||||||
|
|
||||||
|
|
||||||
|
@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")
|
@router.get("/voice/health")
|
||||||
def voice_health() -> dict:
|
def voice_health() -> dict:
|
||||||
"""Erreichbarkeit des Voice-Sidecars + ob der Hermes-API-Key gesetzt ist."""
|
"""Erreichbarkeit des Voice-Sidecars + ob der Hermes-API-Key gesetzt ist."""
|
||||||
@@ -118,6 +127,7 @@ async def voice_stt(audio: UploadFile = File(...), language: str = Form(default=
|
|||||||
files = {"audio": (audio.filename or "rec.webm", data, audio.content_type or "audio/webm")}
|
files = {"audio": (audio.filename or "rec.webm", data, audio.content_type or "audio/webm")}
|
||||||
try:
|
try:
|
||||||
async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
|
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 = await client.post(f"{VOICE_SERVICE_URL}/stt", files=files, data={"language": language})
|
||||||
r.raise_for_status()
|
r.raise_for_status()
|
||||||
return r.json()
|
return r.json()
|
||||||
@@ -166,6 +176,7 @@ async def voice_tts(body: TTSIn) -> Response:
|
|||||||
"""Text → Sprache (Proxy auf Sidecar /tts), liefert WAV-Bytes."""
|
"""Text → Sprache (Proxy auf Sidecar /tts), liefert WAV-Bytes."""
|
||||||
try:
|
try:
|
||||||
async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
|
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 = await client.post(f"{VOICE_SERVICE_URL}/tts", json=body.model_dump())
|
||||||
r.raise_for_status()
|
r.raise_for_status()
|
||||||
return Response(content=r.content, media_type=r.headers.get("content-type", "audio/wav"))
|
return Response(content=r.content, media_type=r.headers.get("content-type", "audio/wav"))
|
||||||
@@ -190,6 +201,7 @@ async def voice_chat(body: ChatIn) -> StreamingResponse:
|
|||||||
if imgs:
|
if imgs:
|
||||||
# Bildschirm-Sicht: erst das Vision-Modell die Monitore beschreiben lassen, dann die Beschreibung
|
# Bildschirm-Sicht: erst das Vision-Modell die Monitore beschreiben lassen, dann die Beschreibung
|
||||||
# als TEXT-Kontext an Hermes (Lucy antwortet mit vollem Hirn/Gedächtnis, sieht via besserem VL-Modell).
|
# als TEXT-Kontext an Hermes (Lucy antwortet mit vollem Hirn/Gedächtnis, sieht via besserem VL-Modell).
|
||||||
|
with Timer("vision"):
|
||||||
desc = await _describe_images(imgs, body.text)
|
desc = await _describe_images(imgs, body.text)
|
||||||
if desc:
|
if desc:
|
||||||
safe_desc = wrap_untrusted(desc, "BILDSCHIRM")
|
safe_desc = wrap_untrusted(desc, "BILDSCHIRM")
|
||||||
@@ -204,6 +216,8 @@ async def voice_chat(body: ChatIn) -> StreamingResponse:
|
|||||||
headers["X-Hermes-Session-Key"] = body.session_key
|
headers["X-Hermes-Session-Key"] = body.session_key
|
||||||
|
|
||||||
async def gen():
|
async def gen():
|
||||||
|
t0 = time.perf_counter()
|
||||||
|
first = True
|
||||||
try:
|
try:
|
||||||
async with httpx.AsyncClient(timeout=httpx.Timeout(None, connect=5.0)) as client:
|
async with httpx.AsyncClient(timeout=httpx.Timeout(None, connect=5.0)) as client:
|
||||||
async with client.stream(
|
async with client.stream(
|
||||||
@@ -214,6 +228,9 @@ async def voice_chat(body: ChatIn) -> StreamingResponse:
|
|||||||
yield f"data: {{\"error\": \"Hermes {r.status_code}: {detail}\"}}\n\n".encode()
|
yield f"data: {{\"error\": \"Hermes {r.status_code}: {detail}\"}}\n\n".encode()
|
||||||
return
|
return
|
||||||
async for chunk in r.aiter_raw():
|
async for chunk in r.aiter_raw():
|
||||||
|
if first: # Time-To-First-Byte des Hermes-Streams (gefühlte Lucy-Latenz)
|
||||||
|
record_stage("chat_ttfb", (time.perf_counter() - t0) * 1000.0)
|
||||||
|
first = False
|
||||||
yield chunk
|
yield chunk
|
||||||
except httpx.HTTPError as exc:
|
except httpx.HTTPError as exc:
|
||||||
yield f"data: {{\"error\": \"Verbindung zu Hermes fehlgeschlagen: {exc}\"}}\n\n".encode()
|
yield f"data: {{\"error\": \"Verbindung zu Hermes fehlgeschlagen: {exc}\"}}\n\n".encode()
|
||||||
|
|||||||
@@ -0,0 +1,68 @@
|
|||||||
|
"""
|
||||||
|
Per-Stage-Latenz-Metriken für die Voice/Lucy-Pipeline.
|
||||||
|
|
||||||
|
Misst die Server-seitige Dauer jeder Stufe (STT, Vision-Beschreibung, Chat-TTFB, TTS) und hält
|
||||||
|
rollende Statistiken (avg/p50/p95/last) im Speicher. Macht aus Latenz-VERMUTUNGEN gemessene Fakten
|
||||||
|
— die eigentliche Voraussetzung, um gezielt zu optimieren (Stufe 5/C2 des Reviews). Anzeige im
|
||||||
|
Frontend-Overhaul (E) analog zur TokenPerformanceCard.
|
||||||
|
|
||||||
|
In-Memory + thread-safe (keine Datei-I/O — Latenz-Telemetrie ist transient, Restart = Reset).
|
||||||
|
"""
|
||||||
|
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
from collections import deque
|
||||||
|
|
||||||
|
_LOCK = threading.Lock()
|
||||||
|
_MAX = 200
|
||||||
|
_STAGES: dict[str, deque] = {}
|
||||||
|
|
||||||
|
# Bekannte Stufen (für stabile UI-Reihenfolge); unbekannte werden trotzdem erfasst.
|
||||||
|
STAGES = ("stt", "vision", "chat_ttfb", "tts")
|
||||||
|
|
||||||
|
|
||||||
|
def record_stage(stage: str, ms: float) -> None:
|
||||||
|
"""Eine gemessene Stage-Dauer (ms) verbuchen. No-op bei negativen Werten."""
|
||||||
|
if ms is None or ms < 0:
|
||||||
|
return
|
||||||
|
with _LOCK:
|
||||||
|
dq = _STAGES.get(stage)
|
||||||
|
if dq is None:
|
||||||
|
dq = _STAGES[stage] = deque(maxlen=_MAX)
|
||||||
|
dq.append(float(ms))
|
||||||
|
|
||||||
|
|
||||||
|
class Timer:
|
||||||
|
"""Context-Manager: misst die verstrichene Zeit und verbucht sie auf `stage`.
|
||||||
|
Funktioniert um `await`-Aufrufe herum (enter → await → exit)."""
|
||||||
|
|
||||||
|
def __init__(self, stage: str) -> None:
|
||||||
|
self.stage = stage
|
||||||
|
self._t0 = 0.0
|
||||||
|
|
||||||
|
def __enter__(self) -> "Timer":
|
||||||
|
self._t0 = time.perf_counter()
|
||||||
|
return self
|
||||||
|
|
||||||
|
def __exit__(self, *exc) -> None:
|
||||||
|
record_stage(self.stage, (time.perf_counter() - self._t0) * 1000.0)
|
||||||
|
|
||||||
|
|
||||||
|
def _summary(vals: list[float]) -> dict:
|
||||||
|
if not vals:
|
||||||
|
return {"count": 0}
|
||||||
|
s = sorted(vals)
|
||||||
|
n = len(s)
|
||||||
|
return {
|
||||||
|
"count": n,
|
||||||
|
"avg_ms": round(sum(s) / n, 1),
|
||||||
|
"p50_ms": round(s[n // 2], 1),
|
||||||
|
"p95_ms": round(s[min(n - 1, int(n * 0.95))], 1),
|
||||||
|
"last_ms": round(vals[-1], 1),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def get_metrics() -> dict:
|
||||||
|
"""Rollende Zusammenfassung je Stufe."""
|
||||||
|
with _LOCK:
|
||||||
|
return {stage: _summary(list(dq)) for stage, dq in _STAGES.items()}
|
||||||
Reference in New Issue
Block a user