jeremy/voicecoach
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: VoiceCoach v2.1 - GPU local, Ollama LLM, settings panel, token server separation

Browse Source 
- pipeline.py: switched LLM from OpenClaw/Claude to local Ollama, added settings loading, sentence aggregator, async SpeechBrain, tuned VAD
- main.py: simplified (no more token server thread, just pipeline)
- token_server.py: NEW - standalone token + settings API server
- docker-compose.yml: all services on GPU machine, added token-server service, CUDA speaches, Ollama LLM
- web-client/index.html: added settings panel with sliders
- web-client/app.js: settings management, localStorage persistence, push to server
- web-client/style.css: settings panel styles
- web-client/nginx.conf: proxy to token-server instead of pipecat-agent
- patch_livekit.py: updated with proper indentation-safe patching
This commit is contained in:
jeremy
2026-04-04 12:27:06 -05:00
parent f62772663d
commit d901208032
12 changed files with 923 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pipecat-agent/pipeline.py
+291 -111
View File
@@ -1,32 +1,39 @@
"""
Pipecat pipeline -- v2 fully self-hosted.
Pipecat pipeline -- v2.1 optimized for low latency.
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
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 AsyncGenerator, Optional
from typing import Optional
import httpx
from pipecat.frames.frames import (
AudioRawFrame, EndFrame, ErrorFrame, Frame, TextFrame, TranscriptionFrame,
UserStartedSpeakingFrame, UserStoppedSpeakingFrame,
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.ai_services import STTService
from pipecat.services.openai import OpenAILLMService, OpenAITTSService
from pipecat.transports.services.livekit import LiveKitTransport, LiveKitParams
@@ -41,78 +48,130 @@ 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')
class SentenceAggregator(FrameProcessor):
"""Buffers streaming TextFrames from LLM and emits one TextFrame per
complete sentence. This lets TTS start on the first sentence while
the LLM is still generating the rest."""
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._buffer = ""
async def process_frame(self, frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, TextFrame):
self._buffer += frame.text
while True:
match = SENTENCE_END.search(self._buffer)
if match:
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 isinstance(frame, EndFrame):
if self._buffer.strip():
logger.info("Flush -> TTS: %s", self._buffer.strip()[:80])
await self.push_frame(TextFrame(text=self._buffer.strip()), direction)
self._buffer = ""
await self.push_frame(frame, direction)
else:
# Flush on any non-text frame (LLM response end, etc.)
if self._buffer.strip() and not isinstance(frame, AudioRawFrame):
logger.info("Flush -> TTS: %s", self._buffer.strip()[:80])
await self.push_frame(TextFrame(text=self._buffer.strip()), direction)
self._buffer = ""
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")
LLM_BASE_URL = os.getenv("LLM_BASE_URL", "http://host.docker.internal:11434/v1")
LLM_MODEL = os.getenv("LLM_MODEL", "llama3.1:8b")
SPEECHBRAIN_URL = os.getenv("SPEECHBRAIN_URL", "http://speechbrain-service:8001")
WHISPER_MODEL = os.getenv("WHISPER_MODEL", "Systran/faster-whisper-large-v3-turbo")
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
class SpeachesSTTService(STTService):
"""STT via Speaches OpenAI-compatible /v1/audio/transcriptions endpoint.
# ---------------------------------------------------------------------------
# SpeachesSTTProcessor — buffers full utterance, sends once to Speaches
# ---------------------------------------------------------------------------
pipecat-ai 0.0.48 does not include OpenAISTTService; this thin wrapper
calls the Speaches HTTP API directly using the STTService base class.
"""
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=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._http = httpx.AsyncClient(timeout=60.0)
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)
@@ -120,34 +179,113 @@ class VocalStateInjector(FrameProcessor):
if isinstance(frame, UserStartedSpeakingFrame):
self._is_speaking = True
self._audio_buffer = []
logger.info("VAD: user started speaking")
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
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)
# 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("Enriched transcription: %s", enriched_text[:200])
logger.info("-> LLM: %s", enriched_text[:120])
await self.push_frame(
TranscriptionFrame(text=enriched_text, user_id=frame.user_id, timestamp=frame.timestamp),
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
@@ -180,12 +318,11 @@ class VocalStateInjector(FrameProcessor):
d,
)
except Exception as e:
logger.warning("SpeechBrain unavailable: %s -- continuing without vocal state", e)
logger.warning("SpeechBrain unavailable: %s", 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
return len(wav_bytes) / (16000 * 2) >= MIN_CLIP_DURATION
async def _save_clip(self, wav_bytes: bytes, analysis: dict):
try:
@@ -193,8 +330,7 @@ class VocalStateInjector(FrameProcessor):
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,
"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"),
@@ -203,22 +339,70 @@ class VocalStateInjector(FrameProcessor):
"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,
"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)
# ---------------------------------------------------------------------------
# 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()
# Load settings from token server
settings = {"confidence": 0.7, "min_volume": 0.3, "start_secs": 0.4,
"stop_secs": 0.5, "max_tokens": 150, "tts_voice": "af_nova"}
try:
async with httpx.AsyncClient(timeout=5.0) as client:
r = await client.get("http://token-server:8080/settings")
if r.status_code == 200:
settings.update(r.json())
logger.info("Loaded settings: %s", settings)
except Exception as e:
logger.warning("Could not load settings, using defaults: %s", e)
transport = LiveKitTransport(
url=livekit_url,
@@ -229,41 +413,38 @@ async def run_voice_coach_pipeline(
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,
confidence=float(settings["confidence"]),
start_secs=float(settings["start_secs"]),
stop_secs=float(settings["stop_secs"]),
min_volume=float(settings["min_volume"]),
)),
vad_audio_passthrough=True,
),
)
# STT: Speaches / faster-whisper via OpenAI-compatible API
stt = SpeachesSTTService(
base_url=SPEACHES_URL,
model=WHISPER_MODEL,
language="en",
)
stt = SpeachesSTTProcessor(base_url=SPEACHES_URL, model=WHISPER_MODEL, language="en")
vocal_injector = VocalStateInjector(session_id=session_id)
sentence_agg = SentenceAggregator()
from prompt_builder import build_system_prompt
system_prompt = build_system_prompt()
# 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,
api_key="ollama",
base_url=LLM_BASE_URL,
model=LLM_MODEL,
max_tokens=int(settings["max_tokens"]),
)
context = OpenAILLMContext(messages=[])
context = OpenAILLMContext(messages=[
{"role": "system", "content": system_prompt},
])
context_aggregator = llm.create_context_aggregator(context)
# TTS: Speaches / Kokoro via OpenAI-compatible API
tts = OpenAITTSService(
tts = SpeachesTTSService(
api_key="speaches",
base_url=f"{SPEACHES_URL}/v1",
model="kokoro",
voice=KOKORO_VOICE,
model=KOKORO_MODEL,
voice=settings.get("tts_voice", KOKORO_VOICE),
)
pipeline = Pipeline([
@@ -272,25 +453,24 @@ async def run_voice_coach_pipeline(
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),
)
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")])
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", session_id)
await task.queue_frames([EndFrame()])
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)