From 47e24dbe5989d9676b72a373fbd968d63e04e9c5 Mon Sep 17 00:00:00 2001 From: Praxis CI Date: Sat, 1 Aug 2026 13:00:00 +0000 Subject: [PATCH] =?UTF-8?q?feat(P01-02-01):=20service=20interfaces=20?= =?UTF-8?q?=E2=80=94=20TTSProvider,=20LLMProvider,=20Guardrail?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit server/services/base.py defines three abstract base classes with type annotations: TTSProvider (async synthesize/synthesize_all, voice_id), LLMProvider (async chat/chat_full, roleplay_model, debrief_model), Guardrail (async check, session_start_disclaimer). GuardrailVerdict/Context/TTSResult/LLMStreamChunk dataclasses carry typed metadata. server/services/registry.py resolves the active adapter from env (PRAXIS_TTS, PRAXIS_GUARDRAIL) so the pipeline never imports a concrete adapter directly — D-014/D-019/D-020 swap wiring. Verified: 'from server.services import TTSProvider, LLMProvider, Guardrail' succeeds. ---ci--- phase: 1 milestone: v0.1 plan: 02 task: 02-01 status: execute persona: backend-engineer requirements: covered: [REQ-VOICE-02, REQ-LLM-01, REQ-ORCH-02] ---/ci--- --- server/services/__init__.py | 36 +++++++ server/services/base.py | 201 ++++++++++++++++++++++++++++++++++++ server/services/registry.py | 74 +++++++++++++ 3 files changed, 311 insertions(+) create mode 100644 server/services/base.py create mode 100644 server/services/registry.py diff --git a/server/services/__init__.py b/server/services/__init__.py index e69de29..240466c 100644 --- a/server/services/__init__.py +++ b/server/services/__init__.py @@ -0,0 +1,36 @@ +"""Praxis service interfaces and adapter registry. + +Public API: + from server.services import TTSProvider, LLMProvider, Guardrail + from server.services import get_tts, get_llm, get_guardrail + +Adapters are resolved from env vars: + PRAXIS_TTS=cartesia|piper + OLLAMA_ROLEPLAY_MODEL / OLLAMA_DEBRIEF_MODEL +""" + +from __future__ import annotations + +from server.services.base import ( + Guardrail, + GuardrailContext, + GuardrailVerdict, + LLMProvider, + LLMStreamChunk, + TTSProvider, + TTSResult, +) +from server.services.registry import get_guardrail, get_llm, get_tts + +__all__ = [ + "TTSProvider", + "TTSResult", + "LLMProvider", + "LLMStreamChunk", + "Guardrail", + "GuardrailVerdict", + "GuardrailContext", + "get_tts", + "get_llm", + "get_guardrail", +] \ No newline at end of file diff --git a/server/services/base.py b/server/services/base.py new file mode 100644 index 0000000..e194f7b --- /dev/null +++ b/server/services/base.py @@ -0,0 +1,201 @@ +"""Praxis service interfaces — abstract base classes for the swappable voice-loop services. + +Per PLAN.md SLICE-02 TASK-02-01 and the D-014/D-019/D-020 swap requirements: + - TTSProvider (D-014): Cartesia (cloud) | Piper (self-hosted) + - LLMProvider (D-020): Ollama Cloud direct API (gemma4:cloud / deepseek-v4-flash:cloud) + - Guardrail (D-019): pluggable; v0.1 = Customer Service ruleset + +These ABCs are the contract the Pipecat pipeline depends on. Adapters wrap the +underlying Pipecat services (or raw APIs) so a swap requires no pipeline change. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from typing import Any, AsyncIterator, Literal + + +# ─── TTS ───────────────────────────────────────────────────────────────────── + + +@dataclass +class TTSResult: + """Result metadata from a TTS synthesis call.""" + + first_audio_ms: float | None = None + chars: int = 0 + voice_id: str | None = None + audio_format: str = "pcm_s16le" + sample_rate: int = 24000 + extra: dict[str, Any] = field(default_factory=dict) + + +class TTSProvider(ABC): + """Abstract TTS provider (D-014). + + One voice persona (D-006) for both role-play and mentor/debrief. + Selection via env var `PRAXIS_TTS=cartesia|piper`. + """ + + name: str = "abstract" + + @abstractmethod + async def synthesize(self, text: str) -> AsyncIterator[bytes]: + """Stream audio chunks (PCM s16le) for the given text. + + Yields bytes as they arrive from the upstream TTS (streaming-first). + The first yielded chunk is the first-audio byte — measure latency there. + """ + ... + # pragma: no cover — abstract + yield b"" # type: ignore[unreachable] + + @abstractmethod + async def synthesize_all(self, text: str) -> tuple[bytes, TTSResult]: + """Fully synthesize `text`, returning (audio_bytes, result_metadata). + + Convenience wrapper for the debrief path where streaming isn't required + on the critical latency path (the debrief is spoken after session end). + """ + ... + + @property + @abstractmethod + def voice_id(self) -> str: + """The configured voice persona id (D-006 — one voice).""" + ... + + +# ─── LLM ───────────────────────────────────────────────────────────────────── + + +@dataclass +class LLMStreamChunk: + """A single chunk from a streaming LLM response.""" + + content: str + is_first: bool = False + finish_reason: str | None = None + input_tokens: int | None = None + output_tokens: int | None = None + extra: dict[str, Any] = field(default_factory=dict) + + +class LLMProvider(ABC): + """Abstract LLM provider (D-020). + + Wraps Ollama Cloud direct API (https://ollama.com/v1 + bearer). Two models: + - gemma4:cloud (role-play fast path) + - deepseek-v4-flash:cloud (debrief / branch classifier, no-think mode) + """ + + name: str = "abstract" + + @abstractmethod + async def chat( + self, + messages: list[dict[str, str]], + *, + stream: bool = True, + model: str | None = None, + no_think: bool = False, + ) -> AsyncIterator[LLMStreamChunk]: + """Stream chat-completion chunks for the given messages. + + `model` overrides the provider default (e.g. deepseek-v4-flash:cloud for + the debrief). `no_think=True` requests no-think mode (deepseek-v4-flash). + """ + ... + # pragma: no cover — abstract + yield LLMStreamChunk(content="") # type: ignore[unreachable] + + @abstractmethod + async def chat_full( + self, + messages: list[dict[str, str]], + *, + model: str | None = None, + no_think: bool = False, + ) -> tuple[str, dict[str, Any]]: + """Return (full_text, usage_metadata) for non-streaming calls. + + Used by the debrief + branch classifier (offline from the voice loop). + """ + ... + + @property + @abstractmethod + def roleplay_model(self) -> str: + """The role-play fast-path model id (gemma4:cloud).""" + ... + + @property + @abstractmethod + def debrief_model(self) -> str: + """The debrief/branch-classifier model id (deepseek-v4-flash:cloud).""" + ... + + +# ─── Guardrail ─────────────────────────────────────────────────────────────── + + +@dataclass +class GuardrailVerdict: + """Verdict from a guardrail check (D-019).""" + + allowed: bool + reason: str = "" + filtered_text: str | None = None + category: str = "ok" # ok | blocked_legal | blocked_financial | blocked_medical | + # blocked_impersonation | blocked_off_role | blocked_pii + extra: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class GuardrailContext: + """Context passed to a guardrail check.""" + + role: Literal["system", "user", "assistant", "debrief"] = "user" + scenario_id: str | None = None + session_id: str | None = None + turn_seq: int | None = None + extra: dict[str, Any] = field(default_factory=dict) + + +class Guardrail(ABC): + """Abstract guardrail layer (D-019). + + Pluggable so health/electrical domains (later milestones) can inject + domain-specific rules without touching the pipeline. v0.1 ships one + implementation: CustomerServiceGuardrail (SLICE-03 TASK-03-04). + """ + + name: str = "abstract" + + @abstractmethod + async def check( + self, text: str, context: GuardrailContext | None = None + ) -> GuardrailVerdict: + """Check `text` against the ruleset; return a verdict.""" + ... + + @property + @abstractmethod + def session_start_disclaimer(self) -> str: + """The session-start disclaimer audio text (RESEARCH.md §Safety). + + Played as the first AI utterance of every session. + """ + ... + + +__all__ = [ + "TTSProvider", + "TTSResult", + "LLMProvider", + "LLMStreamChunk", + "Guardrail", + "GuardrailVerdict", + "GuardrailContext", +] \ No newline at end of file diff --git a/server/services/registry.py b/server/services/registry.py new file mode 100644 index 0000000..ec53608 --- /dev/null +++ b/server/services/registry.py @@ -0,0 +1,74 @@ +"""Adapter registry — resolves the active TTS / LLM / Guardrail from env. + +Centralizes the D-014 (TTS swap), D-020 (LLM swap), D-019 (guardrail plug) wiring +so the Pipecat pipeline never imports a concrete adapter directly. +""" + +from __future__ import annotations + +import os +from functools import lru_cache + + +def _require(key: str, *, default: str | None = None) -> str: + val = os.environ.get(key, default or "").strip() + if not val: + raise RuntimeError( + f"Required env var {key} is not set. See .env.example." + ) + return val + + +@lru_cache(maxsize=1) +def get_tts() -> "TTSProvider": # type: ignore[name-defined] + """Return the active TTSProvider based on PRAXIS_TTS (D-014).""" + # Imported lazily so importing the registry doesn't drag in Pipecat/TTS deps + # for tools that only need the interfaces. + choice = os.environ.get("PRAXIS_TTS", "cartesia").strip().lower() + if choice == "piper": + from server.tts.piper_tts import PiperTTS + + return PiperTTS() + if choice == "cartesia": + from server.tts.cartesia_tts import CartesiaTTS + + return CartesiaTTS() + raise RuntimeError( + f"Unknown PRAXIS_TTS={choice!r}; expected 'cartesia' or 'piper'." + ) + + +@lru_cache(maxsize=1) +def get_llm() -> "LLMProvider": # type: ignore[name-defined] + """Return the active LLMProvider (Ollama Cloud direct API, D-020).""" + from server.llm.ollama_cloud import OllamaCloudLLM + + return OllamaCloudLLM() + + +@lru_cache(maxsize=1) +def get_guardrail() -> "Guardrail": # type: ignore[name-defined] + """Return the active Guardrail (D-019). + + v0.1 SLICE-02 returns NoOpGuardrail; SLICE-03 swaps in CustomerServiceGuardrail. + Selection via PRAXIS_GUARDRAIL=none|customer_service (default: customer_service + once implemented; falls back to none if the ruleset isn't importable yet). + """ + choice = os.environ.get("PRAXIS_GUARDRAIL", "customer_service").strip().lower() + if choice == "none": + from server.guardrails.noop import NoOpGuardrail + + return NoOpGuardrail() + if choice == "customer_service": + try: + from server.guardrails.customer_service import CustomerServiceGuardrail + + return CustomerServiceGuardrail() + except ImportError: + # SLICE-02 fallback — real ruleset arrives in SLICE-03. + from server.guardrails.noop import NoOpGuardrail + + return NoOpGuardrail() + raise RuntimeError( + f"Unknown PRAXIS_GUARDRAIL={choice!r}; expected 'none' or 'customer_service'." + ) \ No newline at end of file