voicecoach/pipecat-agent/pipeline.py

"""
Main Pipecat pipeline assembly.
Connects LiveKit -> Whisper STT -> SpeechBrain -> Claude -> Piper TTS -> LiveKit.
Fully self-hosted STT/TTS (no external API dependencies except Anthropic).
"""
import asyncio
import io
import os
import logging
import wave
import httpx
from typing import Optional

from pipecat.frames.frames import (
    AudioRawFrame, TranscriptionFrame, TextFrame, EndFrame, TTSAudioRawFrame
)
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 prompt_builder import build_system_prompt, build_user_message, VocalState
from whisper_stt import LocalWhisperSTT
from piper_tts import PiperTTSClient

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")


class WhisperSTTProcessor(FrameProcessor):
    """Buffers audio from VAD and transcribes with faster-whisper on end-of-speech."""

    def __init__(self, whisper: LocalWhisperSTT, **kwargs):
        super().__init__(**kwargs)
        self._whisper = whisper
        self._audio_buffer: list[bytes] = []

    async def process_frame(self, frame, direction: FrameDirection):
        await super().process_frame(frame, direction)

        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)


class VocalStateInjector(FrameProcessor):
    """Intercepts transcriptions, fetches vocal analysis from SpeechBrain,
    and enriches the frame before passing to the LLM."""

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._current_audio: list[bytes] = []
        self._http_client = 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)
        elif isinstance(frame, TranscriptionFrame):
            vocal_state = await self._analyze_buffered_audio()
            self._current_audio = []
            enriched_text = build_user_message(frame.text, vocal_state)
            enriched_frame = TranscriptionFrame(
                text=enriched_text, user_id=frame.user_id, timestamp=frame.timestamp,
            )
            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:
            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()

            response = await self._http_client.post(
                f"{SPEECHBRAIN_URL}/analyze",
                files={"audio": ("utterance.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),
            )
        except Exception as e:
            logger.warning(f"SpeechBrain analysis failed (continuing without): {e}")
            return None


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 run_voice_coach_pipeline(
    livekit_url: str,
    livekit_token: str,
    anthropic_key: str,
    session_context: dict,
):
    """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)

    transport = LiveKitTransport(
        url=livekit_url,
        token=livekit_token,
        room_name="voicecoach",
        params=LiveKitParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
            vad_analyzer=SileroVADAnalyzer(),
        ),
    )

    whisper_stt = WhisperSTTProcessor(whisper)
    vocal_injector = VocalStateInjector()

    llm = AnthropicLLMService(
        api_key=anthropic_key,
        model=ANTHROPIC_MODEL,
        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_aggregator = llm.create_context_aggregator(context)

    piper_tts = PiperTTSProcessor(piper_client)

    pipeline = Pipeline([
        transport.input(),
        whisper_stt,
        vocal_injector,
        context_aggregator.user(),
        llm,
        piper_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(f"Participant joined: {participant['identity']}")
        await task.queue_frames([
            TextFrame("VoiceCoach connected. How are you doing today?")
        ])

    @transport.event_handler("on_participant_left")
    async def on_participant_left(transport, participant, reason):
        logger.info(f"Participant left: {participant['identity']}")
        await task.queue_frames([EndFrame()])

    try:
        await runner.run(task)
    finally:
        await piper_client.close()