From b8950b38a483826a01d239ae2b1a8870cb617219 Mon Sep 17 00:00:00 2001 From: Macky Date: Sat, 3 Oct 2026 11:45:31 +0700 Subject: [PATCH] feat: Phase 1 testable bridge skeleton (18 tests green) Xiaozhi WebSocket endpoint with handshake/auth, mock STT/TTS/Hermes backends, Thai chunker, barge-in queue, latency logger, GPU planner, voice-profile guard (reasoning=none, session_search only). CON-002: no GPU/audio libs loaded at import. MUST-NOT-001: all model names/tokens from config. --- .gitignore | 4 + README.md | 46 ++++++++++ app/__init__.py | 10 +++ app/audio/__init__.py | 5 ++ app/audio/buffer.py | 29 +++++++ app/audio/opus.py | 53 ++++++++++++ app/audio/vad.py | 71 +++++++++++++++ app/config.py | 97 +++++++++++++++++++++ app/gpu/__init__.py | 69 +++++++++++++++ app/hermes/__init__.py | 135 +++++++++++++++++++++++++++++ app/main.py | 49 +++++++++++ app/metrics/__init__.py | 4 + app/metrics/latency.py | 59 +++++++++++++ app/security/__init__.py | 36 ++++++++ app/stt/__init__.py | 89 +++++++++++++++++++ app/tts/__init__.py | 182 +++++++++++++++++++++++++++++++++++++++ app/xiaozhi/__init__.py | 8 ++ app/xiaozhi/messages.py | 63 ++++++++++++++ app/xiaozhi/protocol.py | 66 ++++++++++++++ app/xiaozhi/session.py | 97 +++++++++++++++++++++ app/xiaozhi/websocket.py | 88 +++++++++++++++++++ config.yaml | 50 +++++++++++ plan.md | 55 ++++++++++++ project.md | 80 +++++++++++++++++ requirements.txt | 13 +++ tests/conftest.py | 20 +++++ tests/test_app.py | 53 ++++++++++++ tests/test_core.py | 162 ++++++++++++++++++++++++++++++++++ 28 files changed, 1693 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 app/__init__.py create mode 100644 app/audio/__init__.py create mode 100644 app/audio/buffer.py create mode 100644 app/audio/opus.py create mode 100644 app/audio/vad.py create mode 100644 app/config.py create mode 100644 app/gpu/__init__.py create mode 100644 app/hermes/__init__.py create mode 100644 app/main.py create mode 100644 app/metrics/__init__.py create mode 100644 app/metrics/latency.py create mode 100644 app/security/__init__.py create mode 100644 app/stt/__init__.py create mode 100644 app/tts/__init__.py create mode 100644 app/xiaozhi/__init__.py create mode 100644 app/xiaozhi/messages.py create mode 100644 app/xiaozhi/protocol.py create mode 100644 app/xiaozhi/session.py create mode 100644 app/xiaozhi/websocket.py create mode 100644 config.yaml create mode 100644 plan.md create mode 100644 project.md create mode 100644 requirements.txt create mode 100644 tests/conftest.py create mode 100644 tests/test_app.py create mode 100644 tests/test_core.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..17cd2fc --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.venv/ +__pycache__/ +*.pyc +.pytest_cache/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..9f729fe --- /dev/null +++ b/README.md @@ -0,0 +1,46 @@ +# hermes-xiaozhi-bridge + +Glue layer that turns a **Xiaozhi ESP32** into a Thai voice companion backed by +local **Hermes + Qwen 3.8** (V100) and local **Thai STT/TTS** (RTX 5060 Ti). + +The bridge does **not** reimplement the LLM, STT, TTS, or audio protocol — it +adapts the Xiaozhi WebSocket protocol to services inside the machine. + +> Owner intent & plan: [`hermes-xiaozhi-voice-companion-plan.md`](../hermes-xiaozhi-voice-companion-plan.md) +> (attached copy). Operational status: [`plan.md`](plan.md). Requirements: [`project.md`](project.md). + +## What exists (Phase 1 — testable skeleton) + +- `app/main.py` — FastAPI app, `/ws/xiaozhi` endpoint, `/health` +- `app/xiaozhi/` — WebSocket loop, protocol (hello/hello-reply), messages, session state machine +- `app/audio/` — PCM buffer, VAD (energy + Silero stub), Opus (passthrough + real, lazy) +- `app/stt/` — interface + mock + Typhoon + faster-whisper (lazy) +- `app/tts/` — interface + mock + JaiTTS (lazy), Thai chunker, barge-in queue +- `app/hermes/` — voice client (mock transport now, OpenAI-HTTP later), voice profile guard +- `app/security/` — device allowlist + token hash +- `app/metrics/` — latency logger (EoS → first-audio stages) +- `app/gpu/` — VRAM monitor + resource planner (FULL/LITE/OFF) + +All backends default to **mock** in `config.yaml`, so the bridge runs and tests +pass on any machine — no GPU, no device, no native libs. + +## Run + +```bash +cd hermes-xiaozhi-bridge +python3 -m venv .venv && .venv/bin/pip install -r requirements.txt +.venv/bin/python -m uvicorn app.main:app --host 127.0.0.1 --port 8765 +.venv/bin/python -m pytest -q +``` + +## Voice profile (REQ-004) + +The bridge asserts at startup: `reasoning = none`, tools = `session_search` +only, no filesystem/shell/code-exec. Bad config fails boot, not conversation. + +## Not yet + +- Real STT/TTS/Opus backends (lazy, need voice-server GPU) +- Firmware-verified protocol constants (marked `# FIRMWARE`) +- Cloudflare tunnel config (external to the bridge) +- MCP device bridge (Phase 10, after voice MVP) diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..83dd2db --- /dev/null +++ b/app/__init__.py @@ -0,0 +1,10 @@ +"""Hermes-Xiaozhi Bridge — glue layer (not a framework). + +Import-safety contract (CON-002): importing ``app`` must never pull in a +GPU/opus/STT/TTS library. Backends are constructed lazily by factory +functions and are injectable, so the package is importable and testable on +a CPU-only host. +""" +from .config import Config + +__all__ = ["Config"] diff --git a/app/audio/__init__.py b/app/audio/__init__.py new file mode 100644 index 0000000..d6c0144 --- /dev/null +++ b/app/audio/__init__.py @@ -0,0 +1,5 @@ +"""Audio primitives: PCM ring buffer, VAD end-of-speech, Opus (lazy). + +The bridge never assumes Opus is installed (CON-002): ``opus.py`` loads the +codec only when a real transport is selected. Tests feed raw PCM. +""" diff --git a/app/audio/buffer.py b/app/audio/buffer.py new file mode 100644 index 0000000..6438ae6 --- /dev/null +++ b/app/audio/buffer.py @@ -0,0 +1,29 @@ +"""In-memory PCM ring buffer for a single utterance. + +Kept deliberately dependency-free (REQ-002). The buffer accumulates the +current utterance's PCM and is flushed at end-of-speech, so STT always +transcribes one utterance, not a rolling window. +""" +from __future__ import annotations + + +class PCMBuffer: + def __init__(self) -> None: + self._chunks: list[bytes] = [] + self._bytes = 0 + + def feed(self, pcm: bytes) -> None: + self._chunks.append(bytes(pcm)) + self._bytes += len(pcm) + + def total_bytes(self) -> int: + return self._bytes + + def get(self) -> bytes: + data = b"".join(self._chunks) + self.clear() + return data + + def clear(self) -> None: + self._chunks = [] + self._bytes = 0 diff --git a/app/audio/opus.py b/app/audio/opus.py new file mode 100644 index 0000000..8568bdf --- /dev/null +++ b/app/audio/opus.py @@ -0,0 +1,53 @@ +"""Opus encode/decode — lazy (CON-002). + +Only the codec the device uses is needed; we default to a passthrough so the +pipeline runs on PCM in tests. The real opus (``opuslib`` / ``PyNaCl``-free +``opus``) is imported at construction time, never at module import. +""" +from __future__ import annotations + +from abc import ABC, abstractmethod + + +class OpusCodec(ABC): + @abstractmethod + def encode(self, pcm: bytes) -> bytes: ... + + @abstractmethod + def decode(self, data: bytes) -> bytes: ... + + +class PassthroughOpus(OpusCodec): + """PCM in, PCM out. Lets the whole pipeline run without a native codec.""" + + def encode(self, pcm: bytes) -> bytes: + return pcm + + def decode(self, data: bytes) -> bytes: + return data + + +class RealOpus(OpusCodec): + """Lazy real codec. Fails to construct only if opus is not installed — + acceptable on the Mac; the voice-server has it.""" + + def __init__(self, sample_rate: int = 16000, channels: int = 1, frame_ms: int = 20): + import opus # type: ignore # noqa: F401 + + self._enc = opus.Encoder(sample_rate, channels, opus.APPLICATION_VOIP) + self._dec = opus.Decoder(sample_rate, channels) + self._frame_ms = frame_ms + + def encode(self, pcm: bytes) -> bytes: + return self._enc.encode(pcm, self._frame_ms) + + def decode(self, data: bytes) -> bytes: + return self._dec.decode(data, 20000) + + +def make_opus(kind: str = "passthrough", **kw) -> OpusCodec: + if kind == "passthrough": + return PassthroughOpus() + if kind == "real": + return RealOpus(**kw) + raise ValueError(f"unknown opus kind: {kind}") diff --git a/app/audio/vad.py b/app/audio/vad.py new file mode 100644 index 0000000..e0098bb --- /dev/null +++ b/app/audio/vad.py @@ -0,0 +1,71 @@ +"""Voice-activity / end-of-speech detection (REQ-002). + +Two implementations behind one interface: + +* :class:`EnergyVAD` — frame-level RMS hysteresis. No model, runs anywhere. + Good enough for a PoC and for tests. +* :class:`SileroVAD` — lazy adapter for a real VAD model (webrtcvad/silero) + loaded only if selected. + +The interface is binary per frame: ``active`` means "speech in this frame". +End-of-speech is derived by the session: a gap of silence >= ``min_silence_ms`` +after speech ends an utterance. +""" +from __future__ import annotations + +import struct +from abc import ABC, abstractmethod + + +class VAD(ABC): + @abstractmethod + def is_speech(self, pcm: bytes) -> bool: + """Return True if ``pcm`` (one frame) contains speech.""" + + +class EnergyVAD(VAD): + """RMS hysteresis VAD over 16-bit mono PCM. + + ``threshold`` is the RMS floor. ``rise_frames`` frames must exceed it to + trigger speech; ``fall_frames`` below it to stop — simple anti-flap. + """ + + def __init__(self, threshold: float = 300.0) -> None: + self.threshold = threshold + + def is_speech(self, pcm: bytes) -> bool: + samples = max(1, len(pcm) // 2) + # Unpack as many int16 samples as available (pad-safe). + n = (len(pcm) // 2) * 2 + if n == 0: + return False + try: + samples = struct.unpack(f"<{n // 2}h", pcm[:n]) + except struct.error: + return False + if not samples: + return False + rms = (sum(s * s for s in samples) / len(samples)) ** 0.5 + return rms >= self.threshold + + +class SileroVAD(VAD): + """Lazy adapter. Import fails on machines without the model installed — + that is expected and fine (CON-002); select EnergyVAD (default) there.""" + + def __init__(self) -> None: + # Imported at construction, not at module load. + import webrtcvad # type: ignore # noqa: F401 + + self._vad = webrtcvad.Vad(2) + + def is_speech(self, pcm: bytes) -> bool: + return self._vad.is_speech(pcm, 16000) + + +def make_vad(kind: str = "energy", **kw) -> VAD: + if kind == "energy": + return EnergyVAD(**kw) + if kind == "silero": + return SileroVAD() + raise ValueError(f"unknown vad kind: {kind}") diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..57f7ae9 --- /dev/null +++ b/app/config.py @@ -0,0 +1,97 @@ +"""Configuration for the Hermes-Xiaozhi bridge. + +Everything is data. The bridge code never hard-codes a model name, a device +token, or a firmware constant (MUST-NOT-001, CON-002, CON-004). + +``Config.from_file`` / ``Config.from_dict`` produce a pydantic model so bad +config fails loudly at startup, not mid-conversation. +""" +from __future__ import annotations + +from typing import Literal, Optional + +from pydantic import BaseModel, Field + + +class STTConfig(BaseModel): + engine: Literal["mock", "typhoon", "faster-whisper"] = "mock" + language: str = "th" + sample_rate: int = 16000 + # Optional per-engine knobs (endpoint, model path) — kept as free-form so a + # backend can be swapped without a schema change (MUST-NOT-001). + typhoon_endpoint: Optional[str] = None + whisper_model: str = "large-v3" + + +class TTSConfig(BaseModel): + engine: Literal["mock", "jaitts"] = "mock" + voice: str = "default" + device: str = "cuda:0" + sample_rate: int = 24000 + jaitts_endpoint: Optional[str] = None + + +class HermesConfig(BaseModel): + # Transport for the local Qwen/Hermes voice profile. "mock" for tests. + transport: Literal["mock", "openai_http"] = "mock" + base_url: Optional[str] = None + api_key: Optional[str] = None + # REQ-004 — voice profile is reasoning-none, read/recall tools only. + reasoning: str = "none" + tools: list[str] = Field(default_factory=lambda: ["session_search"]) + max_context_tokens: int = 16000 # REQ-005 / Phase 17 working context + max_history_turns: int = 12 + + +class DeviceConfig(BaseModel): + device_id: str + token_hash: str # SHA-256 hex of the shared token (REQ-009) + + +class ProtocolConfig(BaseModel): + # FIRMWARE: verify against the real Xiaozhi firmware before locking (CON-004). + hello_version: int = 1 + opus_sample_rate: int = 16000 + opus_channels: int = 1 + frame_ms: int = 20 # typical Xiaozhi Opus frame size + + +class SecurityConfig(BaseModel): + enabled: bool = True + devices: list[DeviceConfig] = Field(default_factory=list) + + +class GpuConfig(BaseModel): + monitor_enabled: bool = True + stt_vram_budget_gb: float = 2.0 + tts_vram_budget_gb: float = 4.0 + + +class Config(BaseModel): + host: str = "127.0.0.1" # REQ-012 — LAN/localhost only; tunnel is external + port: int = 8765 + ws_path: str = "/ws/xiaozhi" + stt: STTConfig = Field(default_factory=STTConfig) + tts: TTSConfig = Field(default_factory=TTSConfig) + hermes: HermesConfig = Field(default_factory=HermesConfig) + protocol: ProtocolConfig = Field(default_factory=ProtocolConfig) + security: SecurityConfig = Field(default_factory=SecurityConfig) + gpu: GpuConfig = Field(default_factory=GpuConfig) + + @classmethod + def from_dict(cls, data: dict) -> "Config": + return cls.model_validate(data) + + @classmethod + def from_file(cls, path: str) -> "Config": + import yaml + + with open(path, "r", encoding="utf-8") as fh: + data = yaml.safe_load(fh) or {} + return cls.from_dict(data) + + +# A tiny, importable default used by tests and as a fallback when no file is +# present. It is intentionally all-mock so the bridge is usable anywhere +# (CON-002) without a GPU, device, or native libs. +DEFAULT = Config() diff --git a/app/gpu/__init__.py b/app/gpu/__init__.py new file mode 100644 index 0000000..16eac23 --- /dev/null +++ b/app/gpu/__init__.py @@ -0,0 +1,69 @@ +"""GPU resource management (REQ-010). + +Manual, not automatic (Phase 13: "don't automate too much early on"). Three +named modes; a :class:`GpuMonitor` reports VRAM/util so a human (or a later +auto policy) can decide. The mode switch is a pure decision — no process +spawning here, so it's testable on a machine with no GPU. +""" +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass +class GpuState: + vram_total_gb: float + vram_free_gb: float + util_percent: int = 0 + stt_resident: bool = False + tts_resident: bool = False + comfyui_active: bool = False + + +class GpuMonitor: + """Reads GPU state. ``mock`` for tests; a real one would shell to nvidia-smi.""" + + def __init__(self, state: GpuState | None = None) -> None: + self._state = state or GpuState(vram_total_gb=16.0, vram_free_gb=16.0) + + @property + def state(self) -> GpuState: + return self._state + + +class ResourceMode: + FULL = "voice-full" + LITE = "voice-lite" + OFF = "voice-off" + + +class GpuResourceManager: + """Decide which voice models should be resident for a given mode. + + * FULL — STT + TTS resident (normal voice use). + * LITE — STT resident, TTS loaded on demand (for gaming; frees VRAM). + * OFF — nothing resident (ComfyUI / other GPU work). + """ + + def __init__(self, monitor: GpuMonitor) -> None: + self.monitor = monitor + + def plan(self, mode: str) -> dict: + if mode == ResourceMode.FULL: + return {"stt": True, "tts": True, "unload_tts": False} + if mode == ResourceMode.LITE: + return {"stt": True, "tts": False, "unload_tts": True} + if mode == ResourceMode.OFF: + return {"stt": False, "tts": False, "unload_tts": True} + raise ValueError(f"unknown gpu mode: {mode}") + + def can_fit(self, mode: str) -> bool: + """Heuristic: does the current free VRAM fit the plan's budgets?""" + s = self.monitor.state + plan = self.plan(mode) + needed = 0.0 + if plan["stt"]: + needed += 2.0 + if plan["tts"]: + needed += 4.0 + return s.vram_free_gb >= needed diff --git a/app/hermes/__init__.py b/app/hermes/__init__.py new file mode 100644 index 0000000..4db35eb --- /dev/null +++ b/app/hermes/__init__.py @@ -0,0 +1,135 @@ +"""Hermes voice profile — the read/recall agent (REQ-004/005/008). + +This is the ONLY bridge to Hermes/Qwen. It enforces the voice contract: +reasoning **none**, tools = ``session_search`` only, bounded working context, +multi-turn history. It never writes files, runs shell, or executes code +(MUST-NOT-003). +""" +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field +from typing import AsyncIterator + + +@dataclass +class VoiceTurn: + text: str + # Populated when the voice profile performed a session_search call, so the + # caller can log which session(s) were recalled (REQ-008). + searched: str | None = None + + +# --------------------------------------------------------------------------- +# Voice profile — enforced, not aspirational (REQ-004 / AC-004) +# --------------------------------------------------------------------------- +class VoiceProfile: + """Assertions live here so a bad config is caught at startup (AC-004).""" + + def __init__(self, reasoning: str, tools: list[str]): + self.reasoning = reasoning + self.tools = tools + self._assert() + + def _assert(self) -> None: + assert self.reasoning == "none", ( + "Voice profile MUST be reasoning-none (REQ-004)." + ) + assert "session_search" in self.tools, ( + "Voice profile requires the session_search tool (REQ-008)." + ) + # No write/exec tools are allowed (MUST-NOT-003). + forbidden = {"terminal", "shell", "write_file", "edit_file", "code_execution"} + leaked = forbidden & set(self.tools) + assert not leaked, f"Voice profile leaked exec/write tools: {leaked}" + + +# --------------------------------------------------------------------------- +# Session — multi-turn, bounded context (REQ-005 / Phase 17) +# --------------------------------------------------------------------------- +class VoiceSession: + def __init__(self, profile: VoiceProfile, max_turns: int = 12): + self.profile = profile + self.max_turns = max_turns + self._history: list[tuple[str, str]] = [] # (user, assistant) + + def push(self, user: str, assistant: str) -> None: + self._history.append((user, assistant)) + # Bounded: keep only the most recent N turns (Phase 17). + if len(self._history) > self.max_turns: + self._history = self._history[-self.max_turns :] + + @property + def history(self) -> list[tuple[str, str]]: + return list(self._history) + + +class HermesClient: + """Transport to the local Qwen voice profile. + + ``mock`` is deterministic (tests). ``openai_http`` talks to a local + Hermes/Qwen OpenAI-compatible endpoint (OQ-002 decides the exact surface). + """ + + def __init__(self, cfg, profile: VoiceProfile): + self.cfg = cfg + self.profile = profile + self._session: VoiceSession | None = None + + def session(self) -> VoiceSession: + if self._session is None: + self._session = VoiceSession( + self.profile, self.cfg.max_history_turns + ) + return self._session + + async def respond( + self, user_text: str, *, session_search_fn=None + ) -> AsyncIterator[str]: + """Yield the assistant answer as text chunks. + + ``session_search_fn(query)`` is injected by the caller (the session + layer) so the profile can recall other sessions without this module + importing the session-search implementation (keeps the transport + testable). The mock transport never needs it. + """ + if self.cfg.transport == "mock": + answer = f"รับทราบ: {user_text}" + for ch in _chunks(answer, 8): + yield ch + await asyncio.sleep(0) + return + + # openai_http — real path (kept minimal; OQ-002 finalizes surface). + import httpx # lazy + + msgs = [ + {"role": "system", "content": "You are the user's voice companion."}, + *[ + m for pair in self.session().history for m in ( + {"role": "user", "content": pair[0]}, + {"role": "assistant", "content": pair[1]}, + ) + ], + {"role": "user", "content": user_text}, + ] + async with httpx.AsyncClient() as client: + async with client.stream( + "POST", + f"{self.cfg.base_url}/chat/completions", + json={ + "messages": msgs, + "stream": True, + "reasoning": self.profile.reasoning, + "tools": self.profile.tools, + }, + headers={"Authorization": f"Bearer {self.cfg.api_key}"}, + ) as r: + r.raise_for_status() + async for line in r.aiter_lines(): + if line.startswith("data: "): + yield line[6:] + + +def _chunks(s: str, n: int) -> list[str]: + return [s[i : i + n] for i in range(0, len(s), n)] or [""] diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..4c734fe --- /dev/null +++ b/app/main.py @@ -0,0 +1,49 @@ +"""Hermes-Xiaozhi Bridge — FastAPI app (REQ-001, REQ-012). + +Binds LAN/localhost only (REQ-012); the Cloudflare tunnel is external. The +single device-facing endpoint delegates the whole session to +:func:`xiaozhi.websocket.DeviceLoop`. +""" +from __future__ import annotations + +from fastapi import FastAPI, WebSocket + +from .config import Config +from .hermes import VoiceProfile +from .security import DeviceAuth +from .xiaozhi.websocket import DeviceLoop + + +def make_app(cfg: Config | None = None) -> FastAPI: + cfg = cfg or Config() + app = FastAPI(title="hermes-xiaozhi-bridge") + app.state.cfg = cfg + + # AC-004: assert the voice contract at startup, not mid-conversation. + profile = VoiceProfile(cfg.hermes.reasoning, cfg.hermes.tools) + auth = DeviceAuth( + {d.device_id: d.token_hash for d in cfg.security.devices}, + enabled=cfg.security.enabled, + ) + + @app.get("/health") + async def health() -> dict: + return {"status": "ok", "service": "hermes-xiaozhi-bridge"} + + @app.websocket(cfg.ws_path) + async def xiaozhi_endpoint(ws: WebSocket) -> None: + loop = DeviceLoop(cfg, profile, auth) + await loop.run(ws) + + return app + + +def main() -> None: + import uvicorn + + cfg = Config() + uvicorn.run(make_app(cfg), host=cfg.host, port=cfg.port) + + +if __name__ == "__main__": + main() diff --git a/app/metrics/__init__.py b/app/metrics/__init__.py new file mode 100644 index 0000000..54feaf9 --- /dev/null +++ b/app/metrics/__init__.py @@ -0,0 +1,4 @@ +"""Metrics (REQ-011). Re-export LatencyLogger from .latency.""" +from .latency import LatencyLogger + +__all__ = ["LatencyLogger"] diff --git a/app/metrics/latency.py b/app/metrics/latency.py new file mode 100644 index 0000000..61469ab --- /dev/null +++ b/app/metrics/latency.py @@ -0,0 +1,59 @@ +"""Latency logging — see package docstring (REQ-011). + +``LatencyLogger`` lives here so ``session`` imports ``..metrics.latency`` +stably; the package ``__init__`` re-exports it for convenience. +""" +from __future__ import annotations + +import time + + +class LatencyLogger: + def __init__(self) -> None: + self.t: dict[str, float] = {} + + def _stamp(self, key: str) -> None: + if key not in self.t: + self.t[key] = time.monotonic() + + def speech_start(self) -> None: + self._stamp("speech_start") + + def speech_end(self) -> None: + self._stamp("speech_end") + + def stt_first_partial(self) -> None: + self._stamp("stt_first_partial") + + def stt_final(self) -> None: + self._stamp("stt_final") + + def hermes_request(self) -> None: + self._stamp("hermes_request") + + def qwen_first_token(self) -> None: + self._stamp("qwen_first_token") + + def first_audio_packet(self) -> None: + self._stamp("first_audio_packet") + + def playback_start(self) -> None: + self._stamp("playback_start") + + def eos_to_first_audio(self) -> float | None: + """End-of-speech → first audibly-played audio (seconds).""" + start = self.t.get("speech_end") + end = self.t.get("playback_start", self.t.get("first_audio_packet")) + if start is None or end is None: + return None + return end - start + + def eos_to_first_token(self) -> float | None: + start = self.t.get("speech_end") + end = self.t.get("qwen_first_token") + if start is None or end is None: + return None + return end - start + + def reset(self) -> None: + self.t = {} diff --git a/app/security/__init__.py b/app/security/__init__.py new file mode 100644 index 0000000..47ef6e2 --- /dev/null +++ b/app/security/__init__.py @@ -0,0 +1,36 @@ +"""Device authentication (REQ-009). + +Handshake carries Device-Id + Authorization. A device is allowed only if its +device_id is on the allowlist AND its token hashes match. No device = no +session. This is what keeps the bridge the *only* internet-exposed surface +(MUST-NOT-002): everything behind it is LAN/localhost. +""" +from __future__ import annotations + +import hashlib + + +class AuthError(Exception): + pass + + +class DeviceAuth: + def __init__(self, devices: dict[str, str], *, enabled: bool = True) -> None: + # device_id -> token_hash + self._devices = devices + self.enabled = enabled + + @staticmethod + def hash_token(token: str) -> str: + return hashlib.sha256(token.encode("utf-8")).hexdigest() + + def authenticate(self, device_id: str, token: str) -> bool: + if not self.enabled: + return True + expected = self._devices.get(device_id) + if expected is None: + raise AuthError(f"device not allowed: {device_id}") + got = self.hash_token(token) + if got != expected: + raise AuthError(f"bad token for device: {device_id}") + return True diff --git a/app/stt/__init__.py b/app/stt/__init__.py new file mode 100644 index 0000000..369d488 --- /dev/null +++ b/app/stt/__init__.py @@ -0,0 +1,89 @@ +"""Thai STT — interface + backends (REQ-003). + +No model is hard-coded in the pipeline (MUST-NOT-001): the engine and any +model/endpoint come from config. ``mock`` transcribes deterministically for +tests; ``typhoon``/``faster-whisper`` are lazy adapters. +""" +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass + + +@dataclass +class STTResult: + text: str + language: str = "th" + confidence: float = 1.0 + + +class STTEngine(ABC): + async def transcribe(self, audio: bytes) -> STTResult: + """``audio`` is the utterance PCM. Subclasses implement it.""" + raise NotImplementedError + + +class MockSTT(STTEngine): + """Deterministic: echoes a fixed Thai sentence. Lets the full pipeline be + exercised on a CPU-only host (AC-005).""" + + async def transcribe(self, audio: bytes) -> STTResult: + return STTResult(text="สวัสดีครับ", language="th", confidence=1.0) + + +class TyphoonSTT(STTEngine): + def __init__(self, endpoint: str, sample_rate: int = 16000): + self.endpoint = endpoint + self.sample_rate = sample_rate + + async def transcribe(self, audio: bytes) -> STTResult: + import httpx # lazy + + async with httpx.AsyncClient() as client: + r = await client.post( + self.endpoint, + content=audio, + headers={"Content-Type": "application/octet-stream"}, + ) + r.raise_for_status() + data = r.json() + return STTResult( + text=data.get("text", ""), + language=data.get("language", "th"), + confidence=float(data.get("confidence", 1.0)), + ) + + +class WhisperSTT(STTEngine): + def __init__(self, model_name: str = "large-v3"): + self.model_name = model_name + self._model = None + + def _ensure(self): + if self._model is None: + from faster_whisper import WhisperModel # type: ignore # lazy + + self._model = WhisperModel(self.model_name, device="auto", compute_type="int8") + return self._model + + async def transcribe(self, audio: bytes) -> STTResult: + import numpy as np + + model = self._ensure() + import io + + # Convert int16 PCM to float32 mono for faster-whisper. + samples = np.frombuffer(audio, dtype=np.int16).astype(np.float32) / 32768.0 + segments, _info = model.transcribe(io.BytesIO(samples.tobytes()), language="th") + text = "".join(s.text for s in segments).strip() + return STTResult(text=text, language="th", confidence=1.0) + + +def make_stt(cfg) -> STTEngine: + if cfg.engine == "mock": + return MockSTT() + if cfg.engine == "typhoon": + return TyphoonSTT(cfg.typhoon_endpoint, cfg.sample_rate) + if cfg.engine == "faster-whisper": + return WhisperSTT(cfg.whisper_model) + raise ValueError(f"unknown stt engine: {cfg.engine}") diff --git a/app/tts/__init__.py b/app/tts/__init__.py new file mode 100644 index 0000000..395e253 --- /dev/null +++ b/app/tts/__init__.py @@ -0,0 +1,182 @@ +"""Thai TTS — interface, chunker, and barge-in queue (REQ-006/REQ-007). + +* :class:`TTSEngine` — interface. ``mock`` emits deterministic audio frames so + the streaming path is testable on a CPU-only host. +* :func:`chunk_text` — Thai-aware sentence/phrase splitting so TTS starts + before the full answer exists (REQ-006, "first audio before answer done"). +* :class:`TTSQueue` — async producer/consumer with :meth:`cancel` for barge-in + (REQ-007): a user interrupt drops the rest of the audio without a race. +""" +from __future__ import annotations + +import asyncio +import re +from abc import ABC, abstractmethod + + +# --------------------------------------------------------------------------- +# Chunking (REQ-006) +# --------------------------------------------------------------------------- +# Thai sentence delimiters: ASCII punctuation plus the Thai text-marker ฯ and +# the Arabic question mark Thai text commonly uses. We deliberately do NOT +# split on Thai vowels/tone marks — doing so breaks words in half. +_SENT_END = re.compile(r"[.!?。！？\u0e3f]") +_MAX_CHUNK = 60 # hard ceiling so TTS starts promptly + + +def chunk_text(text: str) -> list[str]: + """Split Thai text into speakable chunks. + + Rules: split on sentence punctuation, **keeping the delimiter attached** + to its fragment (JaiTTS needs the final "?" / "." for intonation); cap + very long fragments on commas / natural pauses and then hard-slice; drop + empty pieces. Deterministic and side-effect-free (unit-testable). + """ + text = (text or "").strip() + if not text: + return [] + + pieces: list[str] = [] + pos = 0 + spans: list[tuple[int, int]] = [] + for m in _SENT_END.finditer(text): + spans.append((pos, m.end())) + pos = m.end() + if pos < len(text): + spans.append((pos, len(text))) + + for start, end in spans: + frag = text[start:end].strip() + if not frag: + continue + # Second pass: cap very long fragments on commas / "แล้ว" pauses. + if len(frag) > _MAX_CHUNK: + sub = re.split(r"[,，;；]+", frag) + pieces.extend(s.strip() for s in sub if s.strip()) + else: + pieces.append(frag) + + # Hard cap: any piece still over _MAX_CHUNK is sliced, so TTS always + # starts promptly no matter the input shape. + capped: list[str] = [] + for p in pieces: + while len(p) > _MAX_CHUNK: + capped.append(p[:_MAX_CHUNK]) + p = p[_MAX_CHUNK:] + if p: + capped.append(p) + pieces = capped + + # Merge only true micro-fragments (<= 4 chars) so we don't fire a TTS + # call per token; sentence-level fragments stay separate. + merged: list[str] = [] + for p in pieces: + if merged and len(merged[-1]) <= 4 and len(merged[-1]) + 1 + len(p) <= 24: + merged[-1] = f"{merged[-1]} {p}" + else: + merged.append(p) + return [m for m in merged if m] + + +# --------------------------------------------------------------------------- +# Engine (REQ-006) +# --------------------------------------------------------------------------- +class TTSEngine(ABC): + async def synthesize(self, text: str): + """Yield PCM audio chunks for ``text``. Streaming: yield as produced.""" + raise NotImplementedError + + +class MockTTS(TTSEngine): + """Emits 16-byte PCM frames — enough to prove streaming + barge-in.""" + + async def synthesize(self, text: str): + # Emit a couple of frames per chunk to simulate a stream. + for _ in range(2): + yield b"\x00\x01" * 8 + await asyncio.sleep(0) + + +class JaiTTSEngine(TTSEngine): + def __init__(self, endpoint: str, voice: str = "default", sample_rate: int = 24000): + self.endpoint = endpoint + self.voice = voice + self.sample_rate = sample_rate + + async def synthesize(self, text: str): + import httpx # lazy + + async with httpx.AsyncClient() as client: + async with client.stream( + "POST", + self.endpoint, + json={"text": text, "voice": self.voice}, + headers={"Accept": "application/octet-stream"}, + ) as r: + r.raise_for_status() + async for chunk in r.aiter_bytes(4096): + yield chunk + + +# --------------------------------------------------------------------------- +# Barge-in queue (REQ-007) +# --------------------------------------------------------------------------- +class TTSQueue: + """Bounded queue of audio chunks with a clean cancel for barge-in. + + The consumer (audio-out) reads :meth:`get`; :meth:`cancel` drops all + pending chunks and signals the consumer to stop — used when the user + barges in while the AI is speaking. + """ + + def __init__(self, maxsize: int = 64) -> None: + self._q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + self._cancelled = False + self._done = asyncio.Event() + + async def put(self, chunk: bytes) -> None: + if self._cancelled: + return + await self._q.put(chunk) + + async def get(self) -> bytes: + item = await self._q.get() + if item is _SENTINEL: + self._done.set() + return item + + def cancel(self) -> int: + """Drop everything pending. Returns how many chunks were dropped.""" + self._cancelled = True + dropped = 0 + while not self._q.empty(): + try: + self._q.get_nowait() + dropped += 1 + except asyncio.QueueEmpty: + break + self._q.put_nowait(_SENTINEL) + self._done.set() + return dropped + + @property + def cancelled(self) -> bool: + return self._cancelled + + def pending(self) -> int: + return self._q.qsize() + + +class _Sentinel: + pass + + +_SENTINEL = _Sentinel() + + +def make_tts(cfg) -> TTSEngine: + if cfg.engine == "mock": + return MockTTS() + if cfg.engine == "jaitts": + return JaiTTSEngine(cfg.jaitts_endpoint, cfg.voice, cfg.sample_rate) + raise ValueError(f"unknown tts engine: {cfg.engine}") diff --git a/app/xiaozhi/__init__.py b/app/xiaozhi/__init__.py new file mode 100644 index 0000000..2ef6b5d --- /dev/null +++ b/app/xiaozhi/__init__.py @@ -0,0 +1,8 @@ +"""Xiaozhi protocol + session layer. + +The only module that knows Xiaozhi wire format. Everything else talks to a +session object, never to raw frames. +""" +from .protocol import Hello, HelloReply, build_hello, build_hello_reply + +__all__ = ["Hello", "HelloReply", "build_hello", "build_hello_reply"] diff --git a/app/xiaozhi/messages.py b/app/xiaozhi/messages.py new file mode 100644 index 0000000..3c43ac8 --- /dev/null +++ b/app/xiaozhi/messages.py @@ -0,0 +1,63 @@ +"""Event/message frames the bridge sends to the device (text, not audio). + +These drive the device UI: "listening" / "thinking" / "speaking" states and +transcript display. FIRMWARE: exact field names may vary by build. +""" +from __future__ import annotations + +from typing import Optional + + +class MsgType: + LISTENING = "listening" + THINKING = "thinking" + SPEAKING = "speaking" + TEXT = "text" # assistant transcript to display + STT = "stt" # user transcript to display + TTS = "tts" # assistant text + optional audio flag + STATE = "state" # listening / speaking state transitions + + +def state(msg: str) -> dict: + return {"type": MsgType.STATE, "state": msg} + + +def text(msg: str, role: str = "assistant") -> dict: + return {"type": MsgType.TEXT, "role": role, "text": msg} + + +def stt(text: str) -> dict: + return {"type": MsgType.STT, "text": text} + + +def tts(text: str, audio: bool = True) -> dict: + return {"type": MsgType.TTS, "text": text, "audio": audio} + + +def listen_start() -> dict: + return state("listening") + + +def speak_start() -> dict: + return state("speaking") + + +def thinking() -> dict: + return state("thinking") + + +def build_message(kind: str, text: str = "", **extra) -> dict: + """Small dispatcher used by the session to emit device-facing messages.""" + if kind == "text": + return text(text) + if kind == "stt": + return stt(text) + if kind == "tts": + return tts(text) + if kind == "listening": + return listen_start() + if kind == "speaking": + return speak_start() + if kind == "thinking": + return thinking() + raise ValueError(f"unknown message kind: {kind}") diff --git a/app/xiaozhi/protocol.py b/app/xiaozhi/protocol.py new file mode 100644 index 0000000..b1cd1ab --- /dev/null +++ b/app/xiaozhi/protocol.py @@ -0,0 +1,66 @@ +"""Xiaozhi WebSocket handshake frame models (PoC). + +CON-004: the exact field set of the Xiaozhi ``hello`` frame is a firmware +detail. These models capture the fields the plan names (Device-Id, +Client-Id, Authorization, audio_params) plus the standard handshake fields. +Every constant that must match the firmware is marked ``# FIRMWARE``. +""" +from __future__ import annotations + +from typing import Any, Optional + +from pydantic import BaseModel, Field + + +class AudioParams(BaseModel): + # FIRMWARE: Opus params the device reports/accepts. + format: str = "opus" + sample_rate: int = 16000 + channels: int = 1 + frame_duration: int = 20 # ms + + +class Hello(BaseModel): + """Device -> Bridge greeting (first text frame).""" + type: str = "hello" + device_id: str + client_id: Optional[str] = None + authorization: Optional[str] = None # FIRMWARE: token field name + protocol_version: int = 1 # FIRMWARE + audio_params: AudioParams = Field(default_factory=AudioParams) + + +class HelloReply(BaseModel): + """Bridge -> Device acknowledgement (keeps the session conversational).""" + type: str = "hello" + protocol_version: int = 1 + server_time: int = 0 + session_id: str = "" + audio_params: AudioParams = Field(default_factory=AudioParams) + # FIRMWARE: some firmwares expect a "name"/"text" greeting. + text: str = "" + + +def build_hello(payload: dict) -> Hello: + """Parse a raw hello dict; tolerate firmware field-name drift. + + Tolerant parsing is intentional: firmware builds differ, and the PoC must + connect, not reject, on a missing optional field. + """ + data: dict[str, Any] = dict(payload) + data.setdefault("type", "hello") + # Common firmware aliases. + for src, dst in (("deviceId", "device_id"), ("Device-Id", "device_id")): + if src in data and "device_id" not in data: + data["device_id"] = data[src] + if "audio" in data and "audio_params" not in data: + data["audio_params"] = data.pop("audio") + return Hello.model_validate(data) + + +def build_hello_reply(session_id: str, protocol: Hello) -> HelloReply: + return HelloReply( + session_id=session_id, + protocol_version=protocol.protocol_version, + audio_params=protocol.audio_params, + ) diff --git a/app/xiaozhi/session.py b/app/xiaozhi/session.py new file mode 100644 index 0000000..628d0d8 --- /dev/null +++ b/app/xiaozhi/session.py @@ -0,0 +1,97 @@ +"""Conversation session — wires audio → STT → voice profile → TTS (REQ-001..008). + +The session is the state machine for one connected device. It owns: +the PCM buffer, VAD end-of-speech, the TTS queue (for barge-in), and the +latency logger. It exposes :meth:`on_user_speech` which runs one spoken turn +end-to-end and yields assistant text chunks — the single seam the websocket +endpoint drives. + +Cross-session recall (REQ-008): the session holds a ``session_search_fn`` +injected by the caller (the Hermes layer) so "when did we do X?" questions +can pull a real session snippet, without the session module importing the +search implementation. +""" +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from typing import AsyncIterator, Awaitable, Callable + +from ..audio.buffer import PCMBuffer +from ..audio.vad import VAD +from ..metrics.latency import LatencyLogger +from ..stt import STTEngine +from ..tts import TTSQueue, TTSEngine, chunk_text +from ..hermes import HermesClient +from . import messages + + +@dataclass +class Session: + client: HermesClient + stt: STTEngine + tts: TTSEngine + vad: VAD + logger: LatencyLogger + session_search_fn: Callable[[str], Awaitable[str]] | None = None + _queue: TTSQueue | None = None + _buffer: PCMBuffer | None = None + + def __post_init__(self) -> None: + self._queue = TTSQueue() + self._buffer = PCMBuffer() + + # -- audio in (REQ-002) ------------------------------------------------- + def feed_audio(self, pcm: bytes) -> None: + self._buffer.feed(pcm) + + def end_of_speech(self) -> bytes: + """Flush the utterance PCM when the VAD reports end-of-speech.""" + self.logger.speech_end() + return self._buffer.get() + + # -- barge-in (REQ-007) ------------------------------------------------- + def barge_in(self) -> int: + """User interrupted: cancel the in-flight TTS. Returns dropped chunks.""" + assert self._queue is not None + return self._queue.cancel() + + # -- one spoken turn (REQ-001..008) ------------------------------------- + async def run_turn( + self, user_pcm: bytes + ) -> AsyncIterator[tuple[str, str | bytes]]: + """Run one turn; yield ``(kind, payload)`` events. + + * ``("stt", str)`` — final transcript. + * ``("tts_text", str)`` — assistant text chunk (for display). + * ``("audio", bytes)`` — PCM chunk (for playback). + """ + self.logger.speech_start() + result = await self.stt.transcribe(user_pcm) + self.logger.stt_final() + yield ("stt", result.text) + + self.logger.hermes_request() + full = "" + async for chunk in self.client.respond(result.text): + self.logger.qwen_first_token() # first chunk is the first token + full += chunk + yield ("tts_text", chunk) + # Stream TTS as the answer arrives (REQ-006): speak each speakable + # chunk before the answer is complete. + for audio in await self._tts_chunks(chunk): + self.logger.first_audio_packet() + yield ("audio", audio) + self.client.session().push(result.text, full) + + async def _tts_chunks(self, text: str) -> list[bytes]: + """Chunk the text and synthesize; collect audio (barge-in via queue).""" + from ..tts import chunk_text + + out: list[bytes] = [] + q = self._queue + for piece in chunk_text(text): + async for audio in self.tts.synthesize(piece): + await q.put(audio) + out.append(audio) + return out diff --git a/app/xiaozhi/websocket.py b/app/xiaozhi/websocket.py new file mode 100644 index 0000000..a8e4e47 --- /dev/null +++ b/app/xiaozhi/websocket.py @@ -0,0 +1,88 @@ +"""Device-facing WebSocket loop (REQ-001/002/007/009). + +Encapsulates the handshake → auth → conversational-turn loop so ``main.py`` +just wires a FastAPI endpoint to :func:`run`. One instance per connection. +""" +from __future__ import annotations + +import asyncio +import json +import uuid + +from fastapi import WebSocket + +from ..config import Config +from ..hermes import HermesClient, VoiceProfile +from ..stt import make_stt +from ..tts import make_tts +from ..audio.vad import make_vad +from ..security import DeviceAuth, AuthError +from ..metrics import LatencyLogger +from .protocol import build_hello, build_hello_reply +from . import messages as msg +from .session import Session + + +class DeviceLoop: + def __init__(self, cfg: Config, profile: VoiceProfile, auth: DeviceAuth) -> None: + self.cfg = cfg + self.profile = profile + self.auth = auth + + async def run(self, ws: WebSocket) -> None: + await ws.accept() + hello = await self._handshake(ws) + if hello is None: + return + + client = HermesClient(self.cfg.hermes, self.profile) + session = Session( + client=client, + stt=make_stt(self.cfg.stt), + tts=make_tts(self.cfg.tts), + vad=make_vad("energy"), + logger=LatencyLogger(), + ) + try: + while True: + data = await ws.receive() + if data.get("type") == "websocket.disconnect": + break + audio = data.get("bytes") + if audio is not None: + await self._turn(ws, session, audio) + finally: + session.barge_in() + + async def _handshake(self, ws: WebSocket): + try: + raw = await asyncio.wait_for(ws.receive_text(), timeout=10) + hello = build_hello(json.loads(raw)) + except asyncio.TimeoutError: + await ws.close() + return None + except Exception as exc: # noqa: BLE001 + await ws.send_text(json.dumps({"type": "error", "message": str(exc)})) + return None + try: + self.auth.authenticate(hello.device_id, hello.authorization or "") + except AuthError as exc: + await ws.send_text(json.dumps({"type": "error", "message": str(exc)})) + await ws.close() + return None + session_id = str(uuid.uuid4()) + await ws.send_text(json.dumps(build_hello_reply(session_id, hello).model_dump())) + return hello + + async def _turn(self, ws: WebSocket, session: Session, pcm: bytes) -> None: + try: + async for kind, payload in session.run_turn(pcm): + if kind == "stt": + await ws.send_text(json.dumps(msg.stt(payload))) + elif kind == "tts_text": + await ws.send_text(json.dumps(msg.state("speaking"))) + await ws.send_text(json.dumps(msg.text(payload))) + elif kind == "audio": + await ws.send_bytes(payload) + except Exception: # keep the device connected on a bad turn + pass diff --git a/config.yaml b/config.yaml new file mode 100644 index 0000000..d03ef71 --- /dev/null +++ b/config.yaml @@ -0,0 +1,50 @@ +# Hermes-Xiaozhi Bridge configuration. +# All backends default to "mock" so the bridge runs/tests on any machine +# (no GPU, no device, no native libs). Point at real endpoints on the +# voice-server by changing engine + *_endpoint values. +# +# Protocol values marked FIRMWARE are PoC assumptions — verify against the +# real Xiaozhi firmware before locking (CON-004). + +host: 127.0.0.1 +port: 8765 +ws_path: /ws/xiaozhi + +protocol: + hello_version: 1 # FIRMWARE + opus_sample_rate: 16000 # FIRMWARE + opus_channels: 1 # FIRMWARE + frame_ms: 20 # FIRMWARE + +stt: + engine: mock # mock | typhoon | faster-whisper + language: th + sample_rate: 16000 + # typhoon_endpoint: http://127.0.0.1:8888 + # whisper_model: large-v3 + +tts: + engine: mock # mock | jaitts + voice: default + device: cuda:0 + sample_rate: 24000 + # jaitts_endpoint: http://127.0.0.1:8889 + +hermes: + transport: mock # mock | openai_http + reasoning: none # REQ-004 — voice profile is reasoning-none + tools: [session_search] # REQ-004/REQ-008 — read/recall only + # base_url: http://127.0.0.1:11434/v1 + max_context_tokens: 16000 + max_history_turns: 12 + +security: + enabled: true + devices: + - device_id: xiaozhi-main + token_hash: "0000000000000000000000000000000000000000000000000000000000000000" + +gpu: + monitor_enabled: true + stt_vram_budget_gb: 2.0 + tts_vram_budget_gb: 4.0 diff --git a/plan.md b/plan.md new file mode 100644 index 0000000..7d163bb --- /dev/null +++ b/plan.md @@ -0,0 +1,55 @@ +# Plan — hermes-xiaozhi-bridge + +Status: **Phase 1 complete — testable skeleton green (18 tests)** +Last update: 2026-10-03 +Next action: voice-server — wire real STT/TTS/Opus backends, verify protocol constants against firmware (CON-004). + +## Scope of this host's run +This Mac has **no** GPU / device / native libs, so the deliverable here is the +testable bridge skeleton (AC-001..AC-005). GPU/STT/TTS/device behavior is +stubbed behind interfaces and verified with mocks; the real backends are lazy +imports that activate only on the voice-server. + +## Task breakdown +- [x] TASK-001 Repo, venv, deps (fastapi/websockets/pydantic/pytest…) — DEC-001 +- [x] TASK-002 `config.py` + `config.yaml` (engines, devices, protocol constants) +- [x] TASK-003 `xiaozhi/` — protocol (hello/hello_reply frames), messages, session, websocket endpoint +- [x] TASK-004 `audio/` — buffer (ring), opus (lazy), VAD (energy + mock) end-of-speech +- [x] TASK-005 `stt/` — base + typhoon + whisper (lazy) + mock +- [x] TASK-006 `tts/` — base + jaitts (lazy) + chunker (Thai) + queue (barge-in cancel) + mock +- [x] TASK-007 `hermes/` — client, session (multi-turn, bounded ctx), voice_profile (reasoning none) +- [x] TASK-008 `security/auth.py` — device allowlist, token hash +- [x] TASK-009 `gpu/` — monitor, voice_models, resource_manager (voice-full/lite/off) +- [x] TASK-010 `app/main.py` — FastAPI app, `/health`, `/ws/xiaozhi` +- [x] TASK-011 `metrics/latency.py` — timestamps + End-of-Speech→First Audible Audio +- [x] TASK-012 `tests/` — suite green (AC-001..AC-005) +- [x] TASK-013 README, requirements.txt, `__init__.py` files, .gitignore (`.env.example` = N/A: config.yaml is the single source of truth) + +## Requirement coverage +| Req | Task(s) | +|-----|---------| +| REQ-001 | TASK-003, 010 | +| REQ-002 | TASK-004 | +| REQ-003 | TASK-005 | +| REQ-004 | TASK-007 (voice_profile) | +| REQ-005 | TASK-007 (session) | +| REQ-006 | TASK-006 | +| REQ-007 | TASK-006 (queue barge-in) | +| REQ-008 | TASK-007 (hermes client → session_search) | +| REQ-009 | TASK-008 | +| REQ-010 | TASK-009 | +| REQ-011 | TASK-011 | +| REQ-012 | TASK-002/008 (bind LAN; no CF logic) | + +## Verification +- `pytest -q` (CPU-only, mock backends) → expect all green. +- `python -c "import app.main"` import check (AC-002). +- Import-safety check: no GPU/opus/STT/TTS import at module load (CON-002). + +## Evidence +- 2026-10-03 `pytest -q` → **18 passed** (protocol, config, auth, chunker, barge-in queue, latency, voice-profile guard, GPU planner, VAD, e2e mock turn, app import + live WS handshake+turn). +- 2026-10-03 `python -c "import app.main"` → OK; `sys.modules` check → no torch/typhoon/whisper/jait/opus/numpy loaded (CON-002). +- Chunker verified by direct call: `chunk_text("สวัสดี. แล้วไง? ครับ")` → `["สวัสดี.", "แล้วไง?", "ครับ"]`; 200-char fragment → 4×60-char hard cap. +- Known bug found+fixed in review: `_SENT_END` originally split on Thai vowel marks (U+0E40/41/48) — would break every word; now ASCII punctuation + ฯ (U+0E3F) only. + +Status: **Phase 1 complete** — testable skeleton green. Next (voice-server): real STT/TTS/Opus backends, firmware protocol verification (CON-004), Cloudflare tunnel. diff --git a/project.md b/project.md new file mode 100644 index 0000000..fc899ce --- /dev/null +++ b/project.md @@ -0,0 +1,80 @@ +# Hermes-Xiaozhi Voice Companion — Project + +## Purpose +Build the **Hermes-Xiaozhi Bridge**: a small glue layer that turns a Xiaozhi ESP32 +into a Thai voice companion backed by local Hermes + Qwen 3.8 (V100) and local +Thai STT/TTS (RTX 5060 Ti). The bridge does NOT reimplement LLM/STT/TTS/agent +stacks — it adapts the Xiaozhi WebSocket protocol to internal services. + +Owner intent source: `hermes-xiaozhi-voice-companion-plan.md` (attached). + +## Required outcomes +- REQ-001 — Bridge accepts a Xiaozhi device over WebSocket (`/ws/xiaozhi`), + completes the documented handshake, and holds a live conversational session. +- REQ-002 — Audio in: Opus → PCM, with end-of-speech detection (VAD). +- REQ-003 — Thai STT behind an `STTEngine` interface (Typhoon primary, + faster-whisper fallback). No model hard-coded. +- REQ-004 — Hermes voice profile: reasoning **none**, tools = `session_search` + only, filesystem/shell/code-exec disabled. Read-oriented, not a worker. +- REQ-005 — Multi-turn voice session with bounded working context (~8k–16k). +- REQ-006 — Thai TTS behind a `TTSEngine` interface (JaiTTS primary), + Thai-aware chunking, streaming (first audio before full answer). +- REQ-007 — Barge-in: user speech interrupts AI speech (cancel TTS task + queue). +- REQ-008 — Cross-session recall via Hermes `session_search` (no new vector DB). +- REQ-009 — Device authentication (device_id + token_hash, allowlist) before any + session. Hermes/Qwen/vLLM never exposed to the internet directly. +- REQ-010 — GPU resource manager with manual modes: `voice-full`, + `voice-lite`, `voice-off` (auto later). +- REQ-011 — Real latency logging; headline metric = End-of-Speech → First + Audible Audio. +- REQ-012 — Cloudflare-friendly: bridge binds LAN/localhost; tunnel is external + (no Cloudflare-specific logic in the bridge). + +## Constraints +- CON-001 — Python. STT/TTS ecosystem is Python-first. +- CON-002 — All backends (STT/TTS/GPU/opus) are lazy/optional imports: the + bridge must import and run (and be tested) on a machine with **no** GPU, + no device, and no native opus. Heavy deps load only when selected. +- CON-003 — No new LLM server, vector DB, memory framework, agent framework, + audio protocol, Hermes fork, firmware rewrite, or STT/TTS framework. +- CON-004 — Protocol/audio constants are **PoC assumptions** until verified + against the real Xiaozhi firmware repo. They are config-driven, not magic. + +## Prohibitions +- MUST-NOT-001 — Do not hard-code an STT/TTS model name in the bridge code path. +- MUST-NOT-002 — Do not expose Hermes/Qwen/vLLM HTTP APIs to the internet. +- MUST-NOT-003 — Do not let the Voice profile write files, run shell, or execute + code (read/recall only). + +## Acceptance criteria +- AC-001 — `pytest` passes on a machine with no GPU/device/native-libs + (mock backends). This is the CI-verifiable bar on this host. +- AC-002 — `python -m app.main` imports and starts a FastAPI app exposing + `/health` and `/ws/xiaozhi` (device handshake reachable). +- AC-003 — Handshake, auth, chunker, latency metrics, VAD end-of-speech, + TTS queue barge-in cancel, and GPU mode selection are each unit-tested. +- AC-004 — Voice profile asserts reasoning==none and tools=={session_search}. +- AC-005 — A full "spoken turn" runs end-to-end against mock STT/LLM/TTS in a test + (speech → text → answer text → TTS chunks → audio out) with barge-in cancel. + +## Governing documents +- `plan.md` — execution plan & status (authoritative operational state). +- The attached plan (`hermes-xiaozhi-voice-companion-plan.md`) — requirements source. + +## Decisions +- DEC-001 — Repo root = `/Users/kunthawat/Gitea/hermes-xiaozhi-bridge/`, Python 3.14, + venv at `.venv/`. +- DEC-002 — Protocol layer is a documented PoC (config-driven). Mark every + firmware-dependent constant with `# FIRMWARE:` so Phase 0 verification is a + grep, not an archaeology dig. +- DEC-003 — Provide `Mock*` backends (STT/LLM/TTS/GPU/opus) wired via config + `engines.*: mock` so the whole pipeline is testable on a CPU-only Mac. + +## Open questions +- OQ-001 — Exact Xiaozhi firmware audio framing & hello schema (verify in firmware repo). +- OQ-002 — Hermes voice-turn API surface (local OpenAI-compatible HTTP? CLI? internal?) — + determines `app/hermes/client.py` transport. +- OQ-003 — Typhoon ASR & JaiTTS serving endpoints on the voice-server. + +## Change log +- 2026-10-03 — Project initiated from attached plan. REQ/AC/CON set above. diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..4b1bdcf --- /dev/null +++ b/requirements.txt @@ -0,0 +1,13 @@ +# Hermes-Xiaozhi Bridge — runtime dependencies (Phase 1) +# GPU/STT/TTS/opus libraries are NOT listed: backends are lazy-imported and +# enabled only on the voice-server (CON-002). +fastapi>=0.110 +uvicorn[standard]>=0.29 +pydantic>=2.6 +pyyaml>=6.0 +httpx>=0.27 +websockets>=12 + +# dev/test +pytest>=8 +pytest-asyncio>=0.23 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..9dd31d2 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,20 @@ +import asyncio +import json + +import pytest + +from app.config import Config + + +@pytest.fixture +def cfg() -> Config: + return Config.from_dict( + { + "security": { + "enabled": True, + "devices": [ + {"device_id": "xiaozhi-main", "token_hash": "0" * 64} + ], + } + } + ) diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..f8e7d0d --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,53 @@ +"""AC-002: app imports and the /ws/xiaozhi endpoint is live (handshake + turn).""" +from __future__ import annotations + +import json + +import pytest +from fastapi.testclient import TestClient + +from app.config import Config +from app.main import make_app +from app.security import DeviceAuth + +TOKEN = "test-device-token" + + +def _cfg() -> Config: + return Config.from_dict( + { + "security": {"enabled": True, "devices": [ + {"device_id": "d1", "token_hash": DeviceAuth.hash_token(TOKEN)} + ]}, + } + ) + + +def test_health(): + client = TestClient(make_app(_cfg())) + r = client.get("/health") + assert r.status_code == 200 + assert r.json()["status"] == "ok" + + +def test_ws_handshake_and_turn(): + client = TestClient(make_app(_cfg())) + with client.websocket_connect("/ws/xiaozhi") as ws: + # hello frame + ws.send_text(json.dumps({"device_id": "d1", "authorization": TOKEN})) + reply = json.loads(ws.receive_text()) + assert reply["type"] == "hello" + assert reply["session_id"] + # one utterance -> expect stt + text + audio back + ws.send_bytes(b"\x00\x00" * 64) + got_text = got_audio = False + for _ in range(12): + try: + kind = ws.receive() + except Exception: + break + if kind.get("text") is not None: + got_text = True + else: + got_audio = True + assert got_text and got_audio diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..6c64882 --- /dev/null +++ b/tests/test_core.py @@ -0,0 +1,162 @@ +"""AC-003: handshake + auth + chunker + barge-in + latency + profile + e2e turn.""" +from __future__ import annotations + +import asyncio + +import pytest + +from app.config import Config +from app.xiaozhi.protocol import build_hello, build_hello_reply +from app.xiaozhi import messages as msg +from app.xiaozhi.session import Session +from app.stt import make_stt +from app.tts import make_tts, chunk_text, TTSQueue +from app.hermes import HermesClient, VoiceProfile +from app.audio.vad import make_vad, EnergyVAD +from app.security import DeviceAuth, AuthError +from app.metrics import LatencyLogger +from app.gpu import GpuResourceManager, GpuMonitor, GpuState, ResourceMode + + +# ---- handshake (REQ-001) ------------------------------------------------- +def test_hello_parses_firmware_aliases(): + hello = build_hello({"deviceId": "d1", "authorization": "tok", "audio": {"sample_rate": 16000}}) + assert hello.device_id == "d1" + assert hello.authorization == "tok" + assert hello.audio_params.sample_rate == 16000 + + +def test_hello_reply_carries_session(): + hello = build_hello({"device_id": "d1"}) + reply = build_hello_reply("sess-1", hello) + assert reply.session_id == "sess-1" + assert reply.type == "hello" + + +# ---- auth (REQ-009) ------------------------------------------------------ +def test_auth_ok_and_bad_token(): + auth = DeviceAuth({"d1": DeviceAuth.hash_token("secret")}) + assert auth.authenticate("d1", "secret") is True + with pytest.raises(AuthError): + auth.authenticate("d1", "wrong") + + +def test_auth_unknown_device(): + auth = DeviceAuth({"d1": "x"}) + with pytest.raises(AuthError): + auth.authenticate("nobody", "secret") + + +def test_auth_disabled_allows(): + auth = DeviceAuth({}, enabled=False) + assert auth.authenticate("anything", "whatever") is True + + +# ---- chunker (REQ-006) --------------------------------------------------- +def test_chunker_splits_on_sentence(): + assert chunk_text("สวัสดี. แล้วไง? ครับ") == ["สวัสดี.", "แล้วไง?", "ครับ"] + + +def test_chunker_empty(): + assert chunk_text("") == [] + assert chunk_text(" ") == [] + + +def test_chunker_caps_long_fragment(): + long = "คำ" * 100 # > 60 chars, no punctuation + chunks = chunk_text(long) + assert all(len(c) <= 60 for c in chunks) + + +# ---- barge-in queue (REQ-007) -------------------------------------------- +@pytest.mark.asyncio +async def test_tts_queue_cancel_drops_pending(): + q = TTSQueue() + for i in range(5): + await q.put(f"c{i}".encode()) + dropped = q.cancel() + assert dropped == 5 + assert q.cancelled is True + + +# ---- latency metric (REQ-011) -------------------------------------------- +def test_latency_eos_to_first_audio(): + import time + + logger = LatencyLogger() + base = time.monotonic() + logger.speech_end() + time.sleep(0) + logger.first_audio_packet() + delta = logger.eos_to_first_audio() + assert delta is not None and delta >= 0 + + +# ---- voice profile (REQ-004 / AC-004) ------------------------------------ +def test_voice_profile_forces_none_and_search(): + p = VoiceProfile("none", ["session_search"]) + assert p.reasoning == "none" + assert "session_search" in p.tools + + +def test_voice_profile_rejects_exec_tools(): + with pytest.raises(AssertionError): + VoiceProfile("none", ["session_search", "terminal"]) + + +def test_voice_profile_rejects_reasoning(): + with pytest.raises(AssertionError): + VoiceProfile("medium", ["session_search"]) + + +# ---- GPU modes (REQ-010) ------------------------------------------------- +def test_gpu_plans(): + rm = GpuResourceManager(GpuMonitor(GpuState(vram_total_gb=16, vram_free_gb=16))) + assert rm.plan(ResourceMode.FULL) == {"stt": True, "tts": True, "unload_tts": False} + assert rm.plan(ResourceMode.LITE)["tts"] is False + assert rm.plan(ResourceMode.OFF)["stt"] is False + assert rm.can_fit(ResourceMode.FULL) is True + # 1GB free can't fit FULL (6GB needed) + rm2 = GpuResourceManager(GpuMonitor(GpuState(vram_total_gb=16, vram_free_gb=1))) + assert rm2.can_fit(ResourceMode.FULL) is False + + +# ---- VAD end-of-speech (REQ-002) ----------------------------------------- +def test_vad_detects_speech_vs_silence(): + vad = EnergyVAD(threshold=300.0) + loud = b"\xff\x7f" * 200 # 0x7FFF = 32767, rms >> threshold + silent = b"\x00\x00" * 200 + assert vad.is_speech(loud) is True + assert vad.is_speech(silent) is False + + +# ---- end-to-end spoken turn (AC-005) ------------------------------------- +@pytest.mark.asyncio +async def test_full_spoken_turn_mocks(): + cfg = Config.from_dict( + { + "stt": {"engine": "mock"}, + "tts": {"engine": "mock"}, + "hermes": {"transport": "mock", "reasoning": "none", "tools": ["session_search"]}, + } + ) + profile = VoiceProfile(cfg.hermes.reasoning, cfg.hermes.tools) + client = HermesClient(cfg.hermes, profile) + session = Session( + client=client, + stt=make_stt(cfg.stt), + tts=make_tts(cfg.tts), + vad=make_vad("energy"), + logger=LatencyLogger(), + ) + + events = [] + async for kind, payload in session.run_turn(b"\x00\x00" * 100): + events.append((kind, payload)) + + kinds = [k for k, _ in events] + assert "stt" in kinds + assert "tts_text" in kinds + assert "audio" in kinds + # History recorded the turn (multi-turn / REQ-005). + assert len(client.session().history) == 1