feat: FRIDAY Voice stack on the GPU machine; VoiceCoach history

FRIDAY Voice (speaches GPU STT/TTS + hermes agent + push-to-talk console) on .150 Docker Desktop, fronted by coach.hobbs.farm. Also captures the VoiceCoach compose/agent trees and deploy notes it replaced, which had never been committed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-11 11:19:04 -05:00
parent dddc6734d0
commit 266ea11447
29 changed files with 4494 additions and 0 deletions
@@ -0,0 +1,31 @@
+FROM python:3.11-slim
+
+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
+
+# Patch pipecat LiveKit transport bugs:
+# 1. Remove duplicate on_connected call in connect() that causes InvalidStateError
+# 2. Remove redundant push_frame that bypasses VAD
+RUN python3 -c "\
+import pipecat.transports.services.livekit as m; \
+path = m.__file__; \
+content = open(path).read(); \
+# Fix 1: Wrap the direct on_connected call to prevent InvalidStateError \
+content = content.replace( \
+    'await self._callbacks.on_connected()', \
+    'try:\n                    await self._callbacks.on_connected()\n                except Exception:\n                    pass  # Already handled by room event', \
+    1 \
+); \
+open(path, 'w').write(content); \
+print(f'Patched {path}')"
+
+COPY . .
+
+HEALTHCHECK --interval=30s --timeout=10s --start-period=30s \
+    CMD python -c "import httpx; httpx.get('http://localhost:8080/health').raise_for_status()"
+
+CMD ["python", "main.py"]
@@ -0,0 +1,113 @@
+"""
+VoiceCoach pipeline entry point -- v2.
+Runs token server + restarts pipeline immediately after each session.
+"""
+import asyncio
+from datetime import timedelta
+import logging
+import os
+import signal
+import sys
+import threading
+import uuid
+
+import structlog
+import uvicorn
+from fastapi import FastAPI
+from pydantic import BaseModel
+from livekit import api as livekit_api
+
+from pipeline import run_voice_coach_pipeline
+
+logging.basicConfig(level=logging.INFO)
+logger = structlog.get_logger()
+
+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 ---
+
+token_app = FastAPI(title="VoiceCoach Token Service")
+
+
+class TokenRequest(BaseModel):
+    identity: str = "jeremy"
+
+
+class TokenResponse(BaseModel):
+    token: str
+    url: str
+
+
+@token_app.get("/health")
+async def health():
+    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,
+    ))
+    token.identity = req.identity
+    return TokenResponse(token=token.to_jwt(), url=LIVEKIT_PUBLIC_URL)
+
+
+def run_token_server():
+    uvicorn.run(token_app, host="0.0.0.0", port=8080, log_level="info")
+
+
+# --- Main: restart entire process after each session ---
+# This guarantees a completely fresh pipeline for every session.
+# No stale transport state, no cancelled tasks, no dead queues.
+
+def run_pipeline_sync():
+    """Run one pipeline session, then exit the process.
+    Docker's restart: unless-stopped will bring us back immediately.
+    """
+    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
+    token.ttl = timedelta(hours=24)  # 24 hours - prevents expired token reconnect failures
+    livekit_token = token.to_jwt()
+
+    logger.info("Starting VoiceCoach v2 pipeline", identity=agent_identity)
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    try:
+        loop.run_until_complete(run_voice_coach_pipeline(
+            livekit_url=LIVEKIT_WS_URL,
+            livekit_token=livekit_token,
+            openclaw_token=OPENCLAW_TOKEN,
+        ))
+    except Exception as e:
+        logger.error("Pipeline ended", error=str(e))
+    finally:
+        loop.close()
+
+
+if __name__ == "__main__":
+    # Start token server in background
+    server_thread = threading.Thread(target=run_token_server, daemon=True)
+    server_thread.start()
+    logger.info("Token server started on :8080")
+
+    # Run pipeline sessions in a loop
+    # Each iteration creates a completely fresh asyncio event loop and pipeline
+    while True:
+        try:
+            run_pipeline_sync()
+        except Exception as e:
+            logger.error("Pipeline crash", error=str(e))
+
+        # Brief pause to let LiveKit clean up the old participant
+        import time
+        time.sleep(3)
+        logger.info("Restarting pipeline with fresh state...")
@@ -0,0 +1,34 @@
+"""
+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()
@@ -0,0 +1,76 @@
+"""
+Builds Claude prompts with embedded vocal state metadata.
+"""
+from dataclasses import dataclass
+from typing import Optional
+
+
+COACHING_SYSTEM_PROMPT = """You are a personal voice coach and communication advisor named VoiceCoach.
+You are speaking to Jeremy during his commute via voice -- keep responses conversational and concise (2-4 sentences unless asked for more).
+
+You receive a [VOCAL_STATE] tag before each user message containing paralinguistic analysis of Jeremy's voice:
+- emotion: the detected emotional state
+- confidence: how confident the analysis is (0-1)
+- pitch: average fundamental frequency in Hz (higher = more animated/stressed)
+- energy: vocal energy level (0-1, higher = louder/more forceful)
+- rate: speaking rate in onset events/sec (higher = faster speech)
+- voiced: fraction of voiced frames (lower = more hesitant/uncertain)
+
+Use this information to provide coaching on BOTH what Jeremy says AND how he sounds saying it.
+Important: Never quote the raw numbers. Translate them into natural observations.
+Example: "I notice some tension in your voice..." not "Your pitch was 215hz..."
+
+{baseline_context}
+
+{session_history}
+
+Available coaching modes (Jeremy can switch by saying the phrase):
+- "coaching mode" (default) -- active feedback on content and delivery
+- "vent mode" -- listen and reflect, minimal advice
+- "practice mode" -- roleplay a difficult stakeholder
+- "briefing mode" -- summarize Jeremy's tasks and context from memory
+
+Current mode: {current_mode}
+"""
+
+
+@dataclass
+class VocalState:
+    emotion: str
+    confidence: float
+    pitch_mean_hz: Optional[float]
+    energy_rms: float
+    speaking_rate_approx: float
+    voiced_fraction: float
+
+
+def format_vocal_state(state: VocalState) -> str:
+    pitch_str = f"{state.pitch_mean_hz:.0f}hz" if state.pitch_mean_hz else "unknown"
+    return (
+        f"[VOCAL_STATE "
+        f"emotion={state.emotion} "
+        f"confidence={state.confidence:.2f} "
+        f"pitch={pitch_str} "
+        f"energy={state.energy_rms:.2f} "
+        f"rate={state.speaking_rate_approx:.1f}ops "
+        f"voiced={state.voiced_fraction:.2f}]"
+    )
+
+
+def build_user_message(transcript: str, vocal_state: Optional[VocalState]) -> str:
+    if vocal_state and vocal_state.confidence > 0.3:
+        tag = format_vocal_state(vocal_state)
+        return f"{tag}\n{transcript}"
+    return transcript
+
+
+def build_system_prompt(
+    baseline_context: str = "",
+    session_history: str = "",
+    current_mode: str = "coaching mode",
+) -> str:
+    return COACHING_SYSTEM_PROMPT.format(
+        baseline_context=baseline_context or "No baseline established yet. First session: ask Jeremy to read a short neutral calibration script.",
+        session_history=session_history or "No previous session history.",
+        current_mode=current_mode,
+    )
@@ -0,0 +1,7 @@
+pipecat-ai[livekit,silero,openai]==0.0.48
+httpx==0.27.0
+python-dotenv==1.0.1
+structlog==24.4.0
+livekit-api==0.7.1
+fastapi==0.115.0
+uvicorn[standard]==0.30.0
@@ -0,0 +1,35 @@
+"""
+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