468 lines
19 KiB
Python
468 lines
19 KiB
Python
"""
|
|
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)
|