diff --git a/docker-compose.yml b/docker-compose.yml index 96f5f28..f4dff14 100755 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,6 +1,6 @@ services: - # LiveKit WebRTC Server + # LiveKit WebRTC Server (unchanged) livekit-server: image: livekit/livekit-server:latest container_name: voicecoach-livekit @@ -27,7 +27,39 @@ services: max-size: "50m" max-file: "3" - # SpeechBrain Emotion Analysis (internal only) + # Speaches: self-hosted Whisper STT + Kokoro TTS (v2 - replaces Piper + in-process Whisper) + speaches: + image: ghcr.io/speaches-ai/speaches:latest-cpu + container_name: voicecoach-speaches + restart: unless-stopped + environment: + - WHISPER__MODEL=Systran/faster-whisper-large-v3-turbo + - WHISPER__INFERENCE_DEVICE=cpu + - WHISPER__COMPUTE_TYPE=int8 + - TTS__MODEL=kokoro + - TTS__DEFAULT_VOICE=af_nova + - TTS__LANGUAGE=en-us + volumes: + - /mnt/docker-data/configs/voicecoach/speaches-models:/root/.cache + networks: + - voicecoach + mem_limit: 3G + memswap_limit: 3G + labels: + - "com.centurylinklabs.watchtower.enable=false" + healthcheck: + test: ["CMD", "curl", "-sf", "http://localhost:8000/health"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 120s + logging: + driver: "json-file" + options: + max-size: "50m" + max-file: "3" + + # SpeechBrain Emotion Analysis (unchanged, internal only) speechbrain-service: build: context: ./speechbrain-service @@ -36,6 +68,7 @@ services: restart: unless-stopped volumes: - /mnt/docker-data/configs/voicecoach/speechbrain-models:/app/models + - /mnt/docker-data/configs/voicecoach/clips:/data/clips environment: - MODEL_DIR=/app/models networks: @@ -50,18 +83,26 @@ services: max-size: "50m" max-file: "3" - # Piper TTS (internal only) - piper-tts: - build: - context: ./piper-service - image: voicecoach-piper:latest - container_name: voicecoach-piper + # OpenClaw VoiceCoach Agent (v2 - LLM gateway with memory/skills) + voicecoach-openclaw: + image: margaret-openclaw:latest + container_name: voicecoach-openclaw restart: unless-stopped + user: "0:0" environment: - - PIPER_MODEL=/app/models/en_US-lessac-medium.onnx + - NODE_ENV=production + - HOME=/home/glyptic + - USER=glyptic + - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY} + - OPENCLAW_STATE_DIR=/home/glyptic/.voicecoach + - OPENCLAW_CONFIG_PATH=/home/glyptic/.voicecoach/openclaw.json + volumes: + - /mnt/infrastructure/voicecoach:/home/glyptic/.voicecoach + command: ["sh", "-c", "mkdir -p /home/glyptic && node --max-old-space-size=1500 /usr/local/lib/node_modules/openclaw/dist/index.js gateway --allow-unconfigured"] networks: - voicecoach - mem_limit: 512M + mem_limit: 2G + memswap_limit: 2G labels: - "com.centurylinklabs.watchtower.enable=false" logging: @@ -70,7 +111,7 @@ services: max-size: "50m" max-file: "3" - # Pipecat Voice Pipeline Agent + # Pipecat Voice Pipeline Agent (v2 - uses Speaches + OpenClaw gateway) pipecat-agent: build: context: ./pipecat-agent @@ -81,18 +122,25 @@ services: - .env environment: - LIVEKIT_URL=ws://livekit-server:7880 - - SPEECHBRAIN_SERVICE_URL=http://speechbrain-service:8001 - - PIPER_SERVICE_URL=http://piper-tts:8003 - - WHISPER_MODEL=medium - LIVEKIT_PUBLIC_URL=wss://voice.hobbs.farm + - SPEACHES_URL=http://speaches:8000 + - SPEECHBRAIN_URL=http://speechbrain-service:8001 + - OPENCLAW_URL=http://voicecoach-openclaw:3100 + - OPENCLAW_AGENT_ID=voicecoach + - WHISPER_MODEL=Systran/faster-whisper-large-v3-turbo + - KOKORO_VOICE=af_nova + - CLIPS_DIR=/data/clips/pending + volumes: + - /mnt/docker-data/configs/voicecoach/clips:/data/clips depends_on: - livekit-server + - speaches - speechbrain-service - - piper-tts + - voicecoach-openclaw networks: - voicecoach - mem_limit: 3G - memswap_limit: 3G + mem_limit: 1G + memswap_limit: 1G labels: - "com.centurylinklabs.watchtower.enable=false" logging: @@ -101,7 +149,7 @@ services: max-size: "50m" max-file: "3" - # Web Client (phone interface) + # Web Client (unchanged) voicecoach-web: build: context: ./web-client diff --git a/openclaw-agent/system_prompt.md b/openclaw-agent/system_prompt.md new file mode 100755 index 0000000..445e36a --- /dev/null +++ b/openclaw-agent/system_prompt.md @@ -0,0 +1,63 @@ +You are VoiceCoach, a personal communication and leadership coach for Jeremy. +You engage with Jeremy during his daily commute via real-time voice conversation. + +## Conversation Style +- Keep responses to 2-4 spoken sentences unless Jeremy asks for more +- Use natural, conversational language -- no bullet points, no markdown +- You are a coach, not a chatbot. Ask questions. Challenge thinking. Give direct feedback. + +## Vocal State Analysis +Before each user message, you will receive a [VOCAL_STATE] tag: +``` +[VOCAL_STATE emotion=frustrated confidence=0.82 pitch=215hz energy=0.71 rate=4.8ops voiced=0.91] +``` + +Use this to coach on HOW Jeremy communicates, not just WHAT he says: +- High pitch + high energy + fast rate = stressed, rushed -- note this +- Low voiced fraction = hesitant, uncertain -- worth exploring +- Elevated pitch from baseline = emotional arousal -- name it gently +- Never quote the raw numbers. Translate to natural observations. +- Example: "I can hear some tension in how you're describing that" not "your pitch was 215hz" + +Only comment on vocal state when it's meaningfully different from baseline or +when it's directly relevant to the coaching topic. Don't narrate every turn. + +## Coaching Modes +Jeremy can switch modes mid-conversation. Honor the switch immediately. + +**coaching mode** (default): Active feedback on both content and delivery. + Ask probing questions. Notice patterns. Give direct recommendations. + +**vent mode**: Listen and reflect. Minimal advice. Validate feelings. + Mirror back what you hear. Let Jeremy process out loud. + +**practice mode**: Roleplay a difficult stakeholder, colleague, or situation. + Stay in character. Be realistic -- not easy. Give feedback after. + +**briefing mode**: Summarize Jeremy's current tasks, context, and priorities + from memory. Be crisp. Flag anything time-sensitive. + +## Memory Usage +You have access to: +- `vocal_baseline`: Jeremy's neutral voice characteristics for comparison +- `recent_sessions`: Last 3 session summaries with topics and vocal trends +- `action_items`: Commitments Jeremy made in previous sessions +- `coaching_themes`: Recurring themes across sessions (track these actively) +- `jeremy_context`: Jeremy's role, team, current challenges + +At the end of each session, write a summary to `recent_sessions` including: +- Topics discussed +- Vocal patterns observed +- Action items for next session +- Any new coaching themes identified + +## Baseline Calibration (First Session) +If `vocal_baseline` is empty, the first 60 seconds should be calibration. +Say: "Before we dive in, let's do a quick baseline. Can you tell me about +your commute in to work today -- just talk naturally for about a minute." +Then write the averaged vocal readings to `vocal_baseline`. + +## Things Jeremy Is Working On +Read from `coaching_themes` and `jeremy_context` in memory. +Reference previous sessions naturally: "Last week you mentioned..." +Track whether previous action items were attempted. diff --git a/pipecat-agent/Dockerfile b/pipecat-agent/Dockerfile index 9f4590e..a21f3ee 100755 --- a/pipecat-agent/Dockerfile +++ b/pipecat-agent/Dockerfile @@ -1,24 +1,14 @@ FROM python:3.11-slim -RUN apt-get update && apt-get install -y \ - libsndfile1 \ - ffmpeg \ - && rm -rf /var/lib/apt/lists/* +RUN apt-get update && apt-get install -y libsndfile1 ffmpeg && \ + rm -rf /var/lib/apt/lists/* WORKDIR /app - COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt - -# Pre-download faster-whisper medium model at build time -RUN python -c "\ -from faster_whisper import WhisperModel; \ -WhisperModel('medium', device='cpu', compute_type='int8'); \ -print('Whisper model downloaded.')" - COPY . . HEALTHCHECK --interval=30s --timeout=10s --start-period=30s \ - CMD python -c "print('pipeline process alive')" || exit 1 + CMD python -c "import httpx; httpx.get('http://localhost:8080/health').raise_for_status()" CMD ["python", "main.py"] diff --git a/pipecat-agent/main.py b/pipecat-agent/main.py index f861856..1c6d066 100755 --- a/pipecat-agent/main.py +++ b/pipecat-agent/main.py @@ -1,11 +1,13 @@ """ -Entry point for the Pipecat VoiceCoach agent. -Runs both the token HTTP server (for web clients) and the voice pipeline. +VoiceCoach pipeline entry point -- v2. +Runs token server + pipeline with unique identity per session. """ import asyncio import logging import os import threading +import uuid +from pathlib import Path import structlog import uvicorn @@ -22,8 +24,9 @@ LIVEKIT_API_KEY = os.environ["LIVEKIT_API_KEY"] LIVEKIT_API_SECRET = os.environ["LIVEKIT_API_SECRET"] LIVEKIT_WS_URL = os.getenv("LIVEKIT_URL", "ws://livekit-server:7880") LIVEKIT_PUBLIC_URL = os.getenv("LIVEKIT_PUBLIC_URL", "wss://voice.hobbs.farm") +OPENCLAW_TOKEN = os.getenv("OPENCLAW_TOKEN", "") -# --- Token HTTP server (serves /token for web clients) --- +# --- Token HTTP server --- token_app = FastAPI(title="VoiceCoach Token Service") @@ -39,17 +42,15 @@ class TokenResponse(BaseModel): @token_app.get("/health") async def health(): - return {"status": "ok"} + return {"status": "ok", "version": "2.0"} @token_app.post("/token") async def get_token(req: TokenRequest): token = livekit_api.AccessToken(LIVEKIT_API_KEY, LIVEKIT_API_SECRET) token.with_grants(livekit_api.VideoGrants( - room_join=True, - room="voicecoach", - can_publish=True, - can_subscribe=True, + room_join=True, room="voicecoach", + can_publish=True, can_subscribe=True, )) token.identity = req.identity return TokenResponse(token=token.to_jwt(), url=LIVEKIT_PUBLIC_URL) @@ -59,35 +60,31 @@ def run_token_server(): uvicorn.run(token_app, host="0.0.0.0", port=8080, log_level="info") -# --- Main pipeline --- +# --- Main --- async def main(): - anthropic_key = os.environ["ANTHROPIC_API_KEY"] - - # Start token server in a background thread server_thread = threading.Thread(target=run_token_server, daemon=True) server_thread.start() logger.info("Token server started on :8080") - # Generate agent's own LiveKit token - token = livekit_api.AccessToken(LIVEKIT_API_KEY, LIVEKIT_API_SECRET) - token.with_grants(livekit_api.VideoGrants(room_join=True, room="voicecoach")) - token.identity = "voicecoach-agent" - livekit_token = token.to_jwt() + while True: + agent_identity = f"voicecoach-agent-{uuid.uuid4().hex[:8]}" + token = livekit_api.AccessToken(LIVEKIT_API_KEY, LIVEKIT_API_SECRET) + token.with_grants(livekit_api.VideoGrants(room_join=True, room="voicecoach")) + token.identity = agent_identity + livekit_token = token.to_jwt() - session_context = { - "baseline_context": "", - "session_history": "", - "current_mode": "coaching mode", - } + logger.info("Starting VoiceCoach v2 pipeline", identity=agent_identity) + try: + await run_voice_coach_pipeline( + livekit_url=LIVEKIT_WS_URL, + livekit_token=livekit_token, + openclaw_token=OPENCLAW_TOKEN, + ) + except Exception as e: + logger.error("Pipeline error, restarting in 35s", error=str(e)) - logger.info("Starting VoiceCoach pipeline", livekit_url=LIVEKIT_WS_URL) - await run_voice_coach_pipeline( - livekit_url=LIVEKIT_WS_URL, - livekit_token=livekit_token, - anthropic_key=anthropic_key, - session_context=session_context, - ) + await asyncio.sleep(35) if __name__ == "__main__": diff --git a/pipecat-agent/pipeline.py b/pipecat-agent/pipeline.py index 3426eeb..fb9df0f 100755 --- a/pipecat-agent/pipeline.py +++ b/pipecat-agent/pipeline.py @@ -1,157 +1,224 @@ """ -Main Pipecat pipeline assembly. -Connects LiveKit -> Whisper STT -> SpeechBrain -> Claude -> Piper TTS -> LiveKit. -Fully self-hosted STT/TTS (no external API dependencies except Anthropic). +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 os +import json import logging +import os +import uuid import wave -import httpx -from typing import Optional +from datetime import datetime +from pathlib import Path +from typing import AsyncGenerator, Optional +import httpx from pipecat.frames.frames import ( - AudioRawFrame, TranscriptionFrame, TextFrame, EndFrame, TTSAudioRawFrame + 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.services.anthropic import AnthropicLLMService -from pipecat.transports.services.livekit import LiveKitTransport, LiveKitParams from pipecat.processors.frame_processor import FrameProcessor, FrameDirection -from pipecat.vad.silero import SileroVADAnalyzer +from pipecat.services.ai_services import STTService +from pipecat.services.openai import OpenAILLMService, OpenAITTSService +from pipecat.transports.services.livekit import LiveKitTransport, LiveKitParams -from prompt_builder import build_system_prompt, build_user_message, VocalState -from whisper_stt import LocalWhisperSTT -from piper_tts import PiperTTSClient +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__) -SPEECHBRAIN_URL = os.getenv("SPEECHBRAIN_SERVICE_URL", "http://speechbrain-service:8001") -PIPER_URL = os.getenv("PIPER_SERVICE_URL", "http://piper-tts:8003") -ANTHROPIC_MODEL = "claude-sonnet-4-6" -WHISPER_MODEL = os.getenv("WHISPER_MODEL", "medium") +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 WhisperSTTProcessor(FrameProcessor): - """Buffers audio from VAD and transcribes with faster-whisper on end-of-speech.""" +class SpeachesSTTService(STTService): + """STT via Speaches OpenAI-compatible /v1/audio/transcriptions endpoint. - def __init__(self, whisper: LocalWhisperSTT, **kwargs): + 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._whisper = whisper - self._audio_buffer: list[bytes] = [] + self._base_url = base_url.rstrip("/") + self._model = model + self._language = language + self._http = httpx.AsyncClient(timeout=30.0) - async def process_frame(self, frame, direction: FrameDirection): - await super().process_frame(frame, direction) + async def set_model(self, model: str): + self._model = model - if isinstance(frame, AudioRawFrame): - self._audio_buffer.append(frame.audio) - elif isinstance(frame, EndFrame): - # VAD detected end of speech -- transcribe - if self._audio_buffer: - audio_bytes = b"".join(self._audio_buffer) - self._audio_buffer = [] - text = self._whisper.transcribe(audio_bytes) - if text.strip(): - await self.push_frame( - TranscriptionFrame(text=text, user_id="jeremy", timestamp=""), - direction, - ) - await self.push_frame(frame, direction) - else: - await self.push_frame(frame, direction) + 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): - """Intercepts transcriptions, fetches vocal analysis from SpeechBrain, - and enriches the frame before passing to the LLM.""" + """Buffers audio per utterance, sends to SpeechBrain for analysis, + enriches TranscriptionFrame with [VOCAL_STATE] metadata, + and saves clips for active labeling.""" - def __init__(self, **kwargs): + def __init__(self, session_id: str, **kwargs): super().__init__(**kwargs) - self._current_audio: list[bytes] = [] - self._http_client = httpx.AsyncClient(timeout=3.0) + 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, AudioRawFrame): - self._current_audio.append(frame.audio) + 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): - vocal_state = await self._analyze_buffered_audio() - self._current_audio = [] + 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) - enriched_frame = TranscriptionFrame( - text=enriched_text, user_id=frame.user_id, timestamp=frame.timestamp, + 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, ) - await self.push_frame(enriched_frame, direction) else: await self.push_frame(frame, direction) - async def _analyze_buffered_audio(self) -> Optional[VocalState]: - if not self._current_audio: + def _flush_audio_to_wav(self) -> Optional[bytes]: + if not self._audio_buffer: return None - try: - audio_bytes = b"".join(self._current_audio) - 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() + 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() - response = await self._http_client.post( + async def _analyze(self, wav_bytes: bytes): + try: + r = await self._http.post( f"{SPEECHBRAIN_URL}/analyze", - files={"audio": ("utterance.wav", wav_bytes, "audio/wav")}, + files={"audio": ("u.wav", wav_bytes, "audio/wav")}, ) - response.raise_for_status() - data = response.json() - return VocalState( - emotion=data.get("emotion", "neutral"), - confidence=data.get("emotion_confidence", 0.0), - pitch_mean_hz=data.get("pitch_mean_hz"), - energy_rms=data.get("energy_rms", 0.0), - speaking_rate_approx=data.get("speaking_rate_approx", 0.0), - voiced_fraction=data.get("voiced_fraction", 0.0), + 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(f"SpeechBrain analysis failed (continuing without): {e}") - return None + 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 -class PiperTTSProcessor(FrameProcessor): - """Converts text frames to audio via Piper TTS service.""" - - def __init__(self, piper_client: PiperTTSClient, **kwargs): - super().__init__(**kwargs) - self._piper = piper_client - - async def process_frame(self, frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if isinstance(frame, TextFrame): - wav_bytes = await self._piper.synthesize(frame.text) - if wav_bytes: - await self.push_frame( - TTSAudioRawFrame(audio=wav_bytes, sample_rate=22050, num_channels=1), - direction, - ) - else: - await self.push_frame(frame, direction) + 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, - anthropic_key: str, - session_context: dict, + openclaw_token: str, ): - """Assemble and run the full voice coaching pipeline.""" - - whisper = LocalWhisperSTT(model_size=WHISPER_MODEL, device="cpu", compute_type="int8") - piper_client = PiperTTSClient(base_url=PIPER_URL) + session_id = datetime.utcnow().strftime("%Y%m%d-%H%M%S") transport = LiveKitTransport( url=livekit_url, @@ -160,39 +227,52 @@ async def run_voice_coach_pipeline( params=LiveKitParams( audio_in_enabled=True, audio_out_enabled=True, - vad_analyzer=SileroVADAnalyzer(), + 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, ), ) - whisper_stt = WhisperSTTProcessor(whisper) - vocal_injector = VocalStateInjector() + # STT: Speaches / faster-whisper via OpenAI-compatible API + stt = SpeachesSTTService( + base_url=SPEACHES_URL, + model=WHISPER_MODEL, + language="en", + ) - llm = AnthropicLLMService( - api_key=anthropic_key, - model=ANTHROPIC_MODEL, + 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, ) - system_prompt = build_system_prompt( - baseline_context=session_context.get("baseline_context", ""), - session_history=session_context.get("session_history", ""), - current_mode=session_context.get("current_mode", "coaching mode"), - ) - - context = OpenAILLMContext( - messages=[{"role": "system", "content": system_prompt}] - ) + context = OpenAILLMContext(messages=[]) context_aggregator = llm.create_context_aggregator(context) - piper_tts = PiperTTSProcessor(piper_client) + # 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(), - whisper_stt, + stt, vocal_injector, context_aggregator.user(), llm, - piper_tts, + tts, transport.output(), context_aggregator.assistant(), ]) @@ -201,22 +281,16 @@ async def run_voice_coach_pipeline( 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(f"Participant joined: {participant['identity']}") - await task.queue_frames([ - TextFrame("VoiceCoach connected. How are you doing today?") - ]) + 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(f"Participant left: {participant['identity']}") + logger.info("Session ending: %s", session_id) await task.queue_frames([EndFrame()]) - try: - await runner.run(task) - finally: - await piper_client.close() + await runner.run(task) diff --git a/pipecat-agent/piper_tts.py b/pipecat-agent/piper_tts.py deleted file mode 100755 index 83493de..0000000 --- a/pipecat-agent/piper_tts.py +++ /dev/null @@ -1,34 +0,0 @@ -""" -Piper TTS client for Pipecat pipeline. -Calls the self-hosted Piper HTTP service. -""" -import io -import logging -import wave -import httpx - -logger = logging.getLogger(__name__) - - -class PiperTTSClient: - """HTTP client for the Piper TTS service.""" - - def __init__(self, base_url: str = "http://piper-tts:8003"): - self.base_url = base_url.rstrip("/") - self._client = httpx.AsyncClient(timeout=10.0) - - async def synthesize(self, text: str) -> bytes: - """Convert text to WAV audio bytes via Piper service.""" - try: - response = await self._client.post( - f"{self.base_url}/synthesize", - json={"text": text, "length_scale": 0.95, "sentence_silence": 0.15}, - ) - response.raise_for_status() - return response.content - except Exception as e: - logger.error(f"Piper TTS failed: {e}") - return b"" - - async def close(self): - await self._client.aclose() diff --git a/pipecat-agent/requirements.txt b/pipecat-agent/requirements.txt index 01d6c69..c53c6d7 100755 --- a/pipecat-agent/requirements.txt +++ b/pipecat-agent/requirements.txt @@ -1,12 +1,7 @@ pipecat-ai[livekit,silero,openai]==0.0.48 -fastapi==0.115.0 -uvicorn[standard]==0.30.0 -anthropic==0.34.0 -faster-whisper==1.0.3 -requests==2.32.3 httpx==0.27.0 python-dotenv==1.0.1 structlog==24.4.0 -numpy==1.26.4 -soundfile==0.12.1 livekit-api==0.7.1 +fastapi==0.115.0 +uvicorn[standard]==0.30.0 diff --git a/pipecat-agent/whisper_stt.py b/pipecat-agent/whisper_stt.py deleted file mode 100755 index 31f587e..0000000 --- a/pipecat-agent/whisper_stt.py +++ /dev/null @@ -1,35 +0,0 @@ -""" -Local STT service using faster-whisper. -Replaces Deepgram with fully self-hosted speech-to-text. -""" -import io -import logging -import wave -import numpy as np -from faster_whisper import WhisperModel - -logger = logging.getLogger(__name__) - - -class LocalWhisperSTT: - """Wraps faster-whisper for use in the Pipecat pipeline.""" - - def __init__(self, model_size: str = "medium", device: str = "cpu", compute_type: str = "int8"): - logger.info(f"Loading faster-whisper model: {model_size} on {device}") - self.model = WhisperModel(model_size, device=device, compute_type=compute_type) - logger.info("faster-whisper model loaded.") - - def transcribe(self, audio_bytes: bytes, sample_rate: int = 16000) -> str: - """Transcribe raw PCM audio bytes to text.""" - audio_array = np.frombuffer(audio_bytes, dtype=np.int16).astype(np.float32) / 32768.0 - - segments, info = self.model.transcribe( - audio_array, - beam_size=5, - language="en", - vad_filter=True, - vad_parameters=dict(min_silence_duration_ms=500), - ) - - text = " ".join(seg.text.strip() for seg in segments) - return text diff --git a/piper-service/main.py b/piper-service/main.py index a852be2..5696ebf 100755 --- a/piper-service/main.py +++ b/piper-service/main.py @@ -35,7 +35,7 @@ app = FastAPI(title="Piper TTS Service", lifespan=lifespan) class TTSRequest(BaseModel): text: str - speaker_id: int = 0 + speaker_id: int = None length_scale: float = 1.0 sentence_silence: float = 0.2 @@ -57,13 +57,13 @@ async def synthesize(req: TTSRequest): buf = io.BytesIO() with wave.open(buf, "wb") as wf: - voice.synthesize( - req.text, - wf, - speaker_id=req.speaker_id, - length_scale=req.length_scale, - sentence_silence=req.sentence_silence, - ) + synth_kwargs = { + "length_scale": req.length_scale, + "sentence_silence": req.sentence_silence, + } + if req.speaker_id is not None: + synth_kwargs["speaker_id"] = req.speaker_id + voice.synthesize(req.text, wf, **synth_kwargs) return Response( content=buf.getvalue(),