diff --git a/compose/hf-pdocker-01/voicecoach/pipecat-agent/pipeline.py b/compose/hf-pdocker-01/voicecoach/pipecat-agent/pipeline.py new file mode 100644 index 0000000..caa8f7b --- /dev/null +++ b/compose/hf-pdocker-01/voicecoach/pipecat-agent/pipeline.py @@ -0,0 +1,467 @@ +""" +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') +MIN_FLUSH_CHARS = 40 + + +class SentenceAggregator(FrameProcessor): + """Buffers streaming TextFrames from LLM and emits one TextFrame per + complete sentence. Minimum 40 chars before flushing to prevent + single-word TTS calls.""" + + def __init__(self, **kwargs): + super().__init__(**kwargs) + self._buffer = "" + self._in_response = False + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, TextFrame): + self._in_response = True + self._buffer += frame.text + # Only emit complete sentences above the minimum threshold + while True: + match = SENTENCE_END.search(self._buffer) + if match and match.end() >= MIN_FLUSH_CHARS: + 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) + elif match and len(self._buffer) > MIN_FLUSH_CHARS * 2: + # Buffer getting long without hitting threshold — flush at sentence boundary anyway + 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 self._in_response and not isinstance(frame, (AudioRawFrame, TextFrame)): + # End of LLM response — flush whatever remains + if self._buffer.strip(): + logger.info("End flush -> TTS: %s", self._buffer.strip()[:80]) + await self.push_frame(TextFrame(text=self._buffer.strip()), direction) + self._buffer = "" + self._in_response = False + await self.push_frame(frame, direction) + else: + await self.push_frame(frame, direction) + + +# --------------------------------------------------------------------------- +# Config +# --------------------------------------------------------------------------- + +SPEACHES_URL = os.getenv("SPEACHES_URL", "http://speaches:8000") +OPENCLAW_URL = os.getenv("OPENCLAW_URL", "http://voicecoach-openclaw:3100") +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") +OPENCLAW_AGENT_ID = os.getenv("OPENCLAW_AGENT_ID", "voicecoach") + +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() + + 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=0.5, + start_secs=0.2, + stop_secs=0.4, # Fix 3: reduced from 0.8 + min_volume=0.1, + )), + 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() + + llm = OpenAILLMService( + api_key=openclaw_token, + base_url=f"{OPENCLAW_URL}/v1", + model=f"openclaw/{OPENCLAW_AGENT_ID}", + max_tokens=512, + ) + context = OpenAILLMContext(messages=[]) + context_aggregator = llm.create_context_aggregator(context) + + tts = SpeachesTTSService( + api_key="speaches", + base_url=f"{SPEACHES_URL}/v1", + model=KOKORO_MODEL, + 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)