""" Pipecat pipeline -- v2.1 optimized for low latency. STT: Speaches (faster-whisper) via /v1/audio/transcriptions (utterance-level) LLM: OpenClaw VoiceCoach agent via OpenAI-compatible gateway (streaming) TTS: Speaches (Kokoro 82M) via OpenAI-compatible /v1/audio/speech (sentence-level) Optimizations over v2.0: - Sentence aggregator between LLM and TTS (stream first sentence while LLM generates rest) - SpeechBrain runs async in background (non-blocking, one-turn-behind vocal state) - VAD stop_secs reduced from 0.8 to 0.4 - Whisper model default changed to large-v3-turbo (faster per-token on CPU/GPU) """ import asyncio import io import json import logging import os import re import uuid import wave from datetime import datetime from pathlib import Path from typing import Optional import httpx from pipecat.frames.frames import ( AudioRawFrame, EndFrame, TextFrame, TranscriptionFrame, TTSSpeakFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ErrorFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, ) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frame_processor import FrameProcessor, FrameDirection from pipecat.services.openai import OpenAILLMService, OpenAITTSService from pipecat.transports.services.livekit import LiveKitTransport, LiveKitParams try: from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams except ImportError: from pipecat.vad.silero import SileroVADAnalyzer from pipecat.vad.vad_analyzer import VADParams from prompt_builder import build_user_message, VocalState logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # SpeachesTTSService — Kokoro TTS via Speaches OpenAI-compatible API # --------------------------------------------------------------------------- class SpeachesTTSService(OpenAITTSService): """OpenAITTSService subclass that properly uses base_url and bypasses the hardcoded VALID_VOICES dict for Kokoro voice IDs.""" def __init__(self, *, api_key=None, base_url=None, voice="alloy", model="tts-1", sample_rate=24000, **kwargs): from pipecat.services.ai_services import TTSService TTSService.__init__(self, sample_rate=sample_rate, **kwargs) self._settings = {"sample_rate": sample_rate} self.set_model_name(model) self.set_voice(voice) from openai import AsyncOpenAI self._client = AsyncOpenAI(api_key=api_key, base_url=base_url) async def run_tts(self, text: str): logger.debug("TTS: [%s]", text[:80]) try: await self.start_ttfb_metrics() async with self._client.audio.speech.with_streaming_response.create( input=text or " ", model=self.model_name, voice=self._voice_id, response_format="pcm", ) as r: if r.status_code != 200: error = await r.text() logger.error("TTS error: %s %s", r.status_code, error) yield ErrorFrame(f"TTS error: {r.status_code}") return await self.start_tts_usage_metrics(text) yield TTSStartedFrame() async for chunk in r.iter_bytes(8192): if chunk: await self.stop_ttfb_metrics() yield TTSAudioRawFrame(chunk, 24000, 1) yield TTSStoppedFrame() except Exception as e: logger.exception("TTS error: %s", e) # --------------------------------------------------------------------------- # Sentence Aggregator — buffers LLM streaming tokens into sentences for TTS # --------------------------------------------------------------------------- SENTENCE_END = re.compile(r'[.!?]\s') class SentenceAggregator(FrameProcessor): """Buffers streaming TextFrames from LLM and emits one TextFrame per complete sentence. This lets TTS start on the first sentence while the LLM is still generating the rest.""" def __init__(self, **kwargs): super().__init__(**kwargs) self._buffer = "" async def process_frame(self, frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, TextFrame): self._buffer += frame.text while True: match = SENTENCE_END.search(self._buffer) if match: sentence = self._buffer[:match.end()].strip() self._buffer = self._buffer[match.end():] if sentence: logger.info("Sentence -> TTS: %s", sentence[:80]) await self.push_frame(TextFrame(text=sentence), direction) else: break elif isinstance(frame, EndFrame): if self._buffer.strip(): logger.info("Flush -> TTS: %s", self._buffer.strip()[:80]) await self.push_frame(TextFrame(text=self._buffer.strip()), direction) self._buffer = "" await self.push_frame(frame, direction) else: # Flush on any non-text frame (LLM response end, etc.) if self._buffer.strip() and not isinstance(frame, AudioRawFrame): logger.info("Flush -> TTS: %s", self._buffer.strip()[:80]) await self.push_frame(TextFrame(text=self._buffer.strip()), direction) self._buffer = "" await self.push_frame(frame, direction) # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- SPEACHES_URL = os.getenv("SPEACHES_URL", "http://speaches:8000") LLM_BASE_URL = os.getenv("LLM_BASE_URL", "http://host.docker.internal:11434/v1") LLM_MODEL = os.getenv("LLM_MODEL", "llama3.1:8b") SPEECHBRAIN_URL = os.getenv("SPEECHBRAIN_URL", "http://speechbrain-service:8001") WHISPER_MODEL = os.getenv("WHISPER_MODEL", "deepdml/faster-whisper-large-v3-turbo-ct2") KOKORO_MODEL = os.getenv("KOKORO_MODEL", "speaches-ai/Kokoro-82M-v1.0-ONNX-fp16") KOKORO_VOICE = os.getenv("KOKORO_VOICE", "af_nova") CLIPS_DIR = Path(os.getenv("CLIPS_DIR", "/data/clips/pending")) UNCERTAINTY_THRESHOLD = 0.60 MIN_CLIP_DURATION = 1.0 # --------------------------------------------------------------------------- # SpeachesSTTProcessor — buffers full utterance, sends once to Speaches # --------------------------------------------------------------------------- class SpeachesSTTProcessor(FrameProcessor): """Buffers audio during speech, sends complete utterance to Speaches STT.""" def __init__(self, base_url: str, model: str, language: str = "en", **kwargs): super().__init__(**kwargs) self._base_url = base_url.rstrip("/") self._model = model self._language = language self._http = httpx.AsyncClient(timeout=60.0) self._audio_buffer: list[bytes] = [] self._is_speaking = False async def process_frame(self, frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, UserStartedSpeakingFrame): self._is_speaking = True self._audio_buffer = [] logger.info("STT: user started speaking") await self.push_frame(frame, direction) elif isinstance(frame, AudioRawFrame): if self._is_speaking: self._audio_buffer.append(frame.audio) await self.push_frame(frame, direction) elif isinstance(frame, UserStoppedSpeakingFrame): self._is_speaking = False logger.info("STT: user stopped, %d chunks", len(self._audio_buffer)) if self._audio_buffer: audio_bytes = b"".join(self._audio_buffer) self._audio_buffer = [] buf = io.BytesIO() with wave.open(buf, "wb") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(16000) wf.writeframes(audio_bytes) wav_bytes = buf.getvalue() try: logger.info("STT: sending %d bytes...", len(wav_bytes)) r = await self._http.post( f"{self._base_url}/v1/audio/transcriptions", files={"file": ("utterance.wav", wav_bytes, "audio/wav")}, data={ "model": self._model, "language": self._language, "response_format": "json", "vad_filter": "false", }, ) r.raise_for_status() text = r.json().get("text", "").strip() if text: logger.info("STT: %r", text) await self.push_frame( TranscriptionFrame(text=text, user_id="jeremy", timestamp=datetime.utcnow().isoformat()), direction, ) else: logger.info("STT: empty transcription") except Exception as e: logger.warning("STT error: %s", e) await self.push_frame(frame, direction) else: await self.push_frame(frame, direction) # --------------------------------------------------------------------------- # VocalStateInjector — async SpeechBrain (non-blocking, one-turn-behind) # --------------------------------------------------------------------------- class VocalStateInjector(FrameProcessor): """Enriches TranscriptionFrame with [VOCAL_STATE] from SpeechBrain. Analysis runs in background — uses previous turn's result to avoid blocking the critical path.""" def __init__(self, session_id: str, **kwargs): super().__init__(**kwargs) self._session_id = session_id self._audio_buffer: list[bytes] = [] self._is_speaking = False self._http = httpx.AsyncClient(timeout=3.0) self._last_vocal_state: Optional[VocalState] = None async def process_frame(self, frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, UserStartedSpeakingFrame): self._is_speaking = True self._audio_buffer = [] await self.push_frame(frame, direction) elif isinstance(frame, AudioRawFrame) and self._is_speaking: self._audio_buffer.append(frame.audio) await self.push_frame(frame, direction) elif isinstance(frame, UserStoppedSpeakingFrame): self._is_speaking = False await self.push_frame(frame, direction) elif isinstance(frame, TranscriptionFrame): # Use PREVIOUS turn's vocal state (non-blocking) vocal_state = self._last_vocal_state enriched_text = build_user_message(frame.text, vocal_state) logger.info("-> LLM: %s", enriched_text[:120]) await self.push_frame( TranscriptionFrame(text=enriched_text, user_id=frame.user_id, timestamp=frame.timestamp), direction, ) # Fire background analysis for NEXT turn wav_bytes = self._flush_audio_to_wav() self._audio_buffer = [] if wav_bytes: asyncio.create_task(self._analyze_background(wav_bytes)) else: await self.push_frame(frame, direction) async def _analyze_background(self, wav_bytes: bytes): """Run SpeechBrain in background, store result for next turn.""" try: vocal_state, analysis = await self._analyze(wav_bytes) self._last_vocal_state = vocal_state if analysis and self._clip_worth_saving(analysis, wav_bytes): await self._save_clip(wav_bytes, analysis) except Exception as e: logger.warning("Background SpeechBrain failed: %s", e) def _flush_audio_to_wav(self) -> Optional[bytes]: if not self._audio_buffer: return None raw = b"".join(self._audio_buffer) buf = io.BytesIO() with wave.open(buf, "wb") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(16000) wf.writeframes(raw) return buf.getvalue() async def _analyze(self, wav_bytes: bytes): try: r = await self._http.post( f"{SPEECHBRAIN_URL}/analyze", files={"audio": ("u.wav", wav_bytes, "audio/wav")}, ) r.raise_for_status() d = r.json() return ( VocalState( emotion=d.get("emotion", "neutral"), confidence=d.get("emotion_confidence", 0.0), pitch_mean_hz=d.get("pitch_mean_hz"), energy_rms=d.get("energy_rms", 0.0), speaking_rate_approx=d.get("speaking_rate_approx", 0.0), voiced_fraction=d.get("voiced_fraction", 0.0), ), d, ) except Exception as e: logger.warning("SpeechBrain unavailable: %s", e) return None, None def _clip_worth_saving(self, analysis: dict, wav_bytes: bytes) -> bool: return len(wav_bytes) / (16000 * 2) >= MIN_CLIP_DURATION async def _save_clip(self, wav_bytes: bytes, analysis: dict): try: CLIPS_DIR.mkdir(parents=True, exist_ok=True) clip_id = f"{self._session_id}_{uuid.uuid4().hex[:8]}" (CLIPS_DIR / f"{clip_id}.wav").write_bytes(wav_bytes) meta = { "clip_id": clip_id, "session_id": self._session_id, "timestamp": datetime.utcnow().isoformat(), "duration_seconds": len(wav_bytes) / (16000 * 2), "speechbrain_emotion": analysis.get("emotion"), "speechbrain_confidence": analysis.get("emotion_confidence"), "emotion_scores": analysis.get("emotion_scores", {}), "pitch_mean_hz": analysis.get("pitch_mean_hz"), "energy_rms": analysis.get("energy_rms"), "speaking_rate_approx": analysis.get("speaking_rate_approx"), "labeled": False, "confirmed_label": None, "uncertain": analysis.get("emotion_confidence", 1.0) < UNCERTAINTY_THRESHOLD, } (CLIPS_DIR / f"{clip_id}.json").write_text(json.dumps(meta, indent=2)) except Exception as e: logger.warning("Failed to save clip: %s", e) # --------------------------------------------------------------------------- # Pre-warm # --------------------------------------------------------------------------- async def _prewarm_speaches(): """Pre-warm Whisper and Kokoro models so first request is fast.""" client = httpx.AsyncClient(timeout=120.0) try: import numpy as np sr = 16000 audio = np.zeros(sr, dtype=np.int16) buf = io.BytesIO() with wave.open(buf, "wb") as wf: wf.setnchannels(1); wf.setsampwidth(2); wf.setframerate(sr) wf.writeframes(audio.tobytes()) r = await client.post( f"{SPEACHES_URL}/v1/audio/transcriptions", files={"file": ("warmup.wav", buf.getvalue(), "audio/wav")}, data={"model": WHISPER_MODEL, "language": "en", "response_format": "json", "vad_filter": "false"}, ) logger.info("Pre-warm Whisper: %s", r.status_code) r2 = await client.post( f"{SPEACHES_URL}/v1/audio/speech", json={"model": KOKORO_MODEL, "input": "Ready.", "voice": KOKORO_VOICE}, ) logger.info("Pre-warm Kokoro: %s", r2.status_code) except Exception as e: logger.warning("Pre-warm failed: %s", e) finally: await client.aclose() # --------------------------------------------------------------------------- # Pipeline assembly # --------------------------------------------------------------------------- async def run_voice_coach_pipeline( livekit_url: str, livekit_token: str, openclaw_token: str, ): session_id = datetime.utcnow().strftime("%Y%m%d-%H%M%S") await _prewarm_speaches() # Load settings from token server settings = {"confidence": 0.7, "min_volume": 0.3, "start_secs": 0.4, "stop_secs": 0.5, "max_tokens": 150, "tts_voice": "af_nova"} try: async with httpx.AsyncClient(timeout=5.0) as client: r = await client.get("http://token-server:8080/settings") if r.status_code == 200: settings.update(r.json()) logger.info("Loaded settings: %s", settings) except Exception as e: logger.warning("Could not load settings, using defaults: %s", e) transport = LiveKitTransport( url=livekit_url, token=livekit_token, room_name="voicecoach", params=LiveKitParams( audio_in_enabled=True, audio_out_enabled=True, vad_enabled=True, vad_analyzer=SileroVADAnalyzer(params=VADParams( confidence=float(settings["confidence"]), start_secs=float(settings["start_secs"]), stop_secs=float(settings["stop_secs"]), min_volume=float(settings["min_volume"]), )), vad_audio_passthrough=True, ), ) stt = SpeachesSTTProcessor(base_url=SPEACHES_URL, model=WHISPER_MODEL, language="en") vocal_injector = VocalStateInjector(session_id=session_id) sentence_agg = SentenceAggregator() from prompt_builder import build_system_prompt system_prompt = build_system_prompt() llm = OpenAILLMService( api_key="ollama", base_url=LLM_BASE_URL, model=LLM_MODEL, max_tokens=int(settings["max_tokens"]), ) context = OpenAILLMContext(messages=[ {"role": "system", "content": system_prompt}, ]) context_aggregator = llm.create_context_aggregator(context) tts = SpeachesTTSService( api_key="speaches", base_url=f"{SPEACHES_URL}/v1", model=KOKORO_MODEL, voice=settings.get("tts_voice", KOKORO_VOICE), ) pipeline = Pipeline([ transport.input(), stt, vocal_injector, context_aggregator.user(), llm, sentence_agg, # Fix 1: aggregate sentences before TTS tts, transport.output(), context_aggregator.assistant(), ]) task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) runner = PipelineRunner() @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): logger.info("Participant joined: %s", participant) await task.queue_frames([TTSSpeakFrame("VoiceCoach connected. How are you doing today?")]) @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): logger.info("Session ending: %s (reason: %s)", session_id, reason) await task.cancel() await runner.run(task) logger.info("Pipeline runner exited for session %s", session_id)