""" Pipecat pipeline -- v2 fully self-hosted. STT: Speaches (faster-whisper) via OpenAI-compatible /v1/audio/transcriptions LLM: OpenClaw VoiceCoach agent via OpenAI-compatible gateway TTS: Speaches (Kokoro 82M) via OpenAI-compatible /v1/audio/speech """ import asyncio import io import json import logging import os import uuid import wave from datetime import datetime from pathlib import Path from typing import AsyncGenerator, Optional import httpx from pipecat.frames.frames import ( AudioRawFrame, EndFrame, ErrorFrame, Frame, TextFrame, TranscriptionFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ) 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.ai_services import STTService 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__) 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", "Systran/faster-whisper-large-v3-turbo") 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 class SpeachesSTTService(STTService): """STT via Speaches OpenAI-compatible /v1/audio/transcriptions endpoint. pipecat-ai 0.0.48 does not include OpenAISTTService; this thin wrapper calls the Speaches HTTP API directly using the STTService base class. """ 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=30.0) async def set_model(self, model: str): self._model = model async def set_language(self, language): self._language = str(language) async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: if not audio: return # Wrap raw PCM in a WAV container for the API buf = io.BytesIO() with wave.open(buf, "wb") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(16000) wf.writeframes(audio) wav_bytes = buf.getvalue() try: r = await self._http.post( f"{self._base_url}/v1/audio/transcriptions", files={"file": ("audio.wav", wav_bytes, "audio/wav")}, data={"model": self._model, "language": self._language, "response_format": "json"}, ) r.raise_for_status() text = r.json().get("text", "").strip() if text: logger.info("STT transcription: %s", text[:100]) yield TranscriptionFrame(text=text, user_id="user", timestamp=datetime.utcnow().isoformat()) except Exception as e: logger.warning("Speaches STT error: %s", e) yield ErrorFrame(error=str(e)) class VocalStateInjector(FrameProcessor): """Buffers audio per utterance, sends to SpeechBrain for analysis, enriches TranscriptionFrame with [VOCAL_STATE] metadata, and saves clips for active labeling.""" 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) 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("VAD: user started speaking") 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 logger.info("VAD: user stopped speaking, buffer: %d chunks", len(self._audio_buffer)) await self.push_frame(frame, direction) elif isinstance(frame, TranscriptionFrame): wav_bytes = self._flush_audio_to_wav() self._audio_buffer = [] vocal_state = None if wav_bytes: vocal_state, analysis = await self._analyze(wav_bytes) if analysis and self._clip_worth_saving(analysis, wav_bytes): await self._save_clip(wav_bytes, analysis) enriched_text = build_user_message(frame.text, vocal_state) logger.info("Enriched transcription: %s", enriched_text[:200]) await self.push_frame( TranscriptionFrame(text=enriched_text, user_id=frame.user_id, timestamp=frame.timestamp), direction, ) else: await self.push_frame(frame, direction) 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 -- continuing without vocal state", e) return None, None def _clip_worth_saving(self, analysis: dict, wav_bytes: bytes) -> bool: duration = len(wav_bytes) / (16000 * 2) return duration >= 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)) logger.info("Saved clip: %s", clip_id) except Exception as e: logger.warning("Failed to save clip: %s", e) 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") 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.8, min_volume=0.1, )), vad_audio_passthrough=True, ), ) # STT: Speaches / faster-whisper via OpenAI-compatible API stt = SpeachesSTTService( base_url=SPEACHES_URL, model=WHISPER_MODEL, language="en", ) vocal_injector = VocalStateInjector(session_id=session_id) # LLM: OpenClaw VoiceCoach agent via OpenAI-compatible gateway 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: Speaches / Kokoro via OpenAI-compatible API tts = OpenAITTSService( api_key="speaches", base_url=f"{SPEACHES_URL}/v1", model="kokoro", voice=KOKORO_VOICE, ) pipeline = Pipeline([ transport.input(), stt, vocal_injector, context_aggregator.user(), llm, 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([TextFrame("session_start")]) @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): logger.info("Session ending: %s", session_id) await task.queue_frames([EndFrame()]) await runner.run(task)