wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add model-specific sample rates, transport example, and fix audio buffer alignment

Browse Source
This commit is contained in:
Neil Ruaro
2026-01-13 06:32:11 +09:00
parent e76a3d04f0
commit ed120d014d
4 changed files with 176 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
examples/foundational/07zb-interruptible-camb-local.py
View File

@@ -6,11 +6,8 @@
"""Camb.ai TTS example with local audio (microphone/speakers).
This example demonstrates:
- Camb.ai MARS TTS with streaming audio
- Local audio input/output (no WebRTC or Daily needed)
- TTFB metrics tracking
- End-to-end latency measurement (user speech → AI response)
This is a standalone local example for quick testing without WebRTC/Daily.
For production use with Daily/Twilio/WebRTC, see 07zb-interruptible-camb.py
Requirements:
- CAMB_API_KEY environment variable
@@ -108,7 +105,7 @@ DEFAULT_VOICE_ID = 147320
async def main(voice_id: int):
sample_rate = 48000
sample_rate = 22050 # mars-flash uses 22.05kHz
# Local audio transport - uses your microphone and speakers
# Increase audio_out_10ms_chunks for larger buffer (default is 4 = 40ms)
@@ -124,7 +121,7 @@ async def main(voice_id: int):
# Deepgram STT for speech recognition
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
# Camb.ai TTS (48kHz output)
# Camb.ai TTS
tts = CambTTSService(
api_key=os.getenv("CAMB_API_KEY"),
voice_id=voice_id,

123
examples/foundational/07zb-interruptible-camb.py Normal file
View File

@@ -0,0 +1,123 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
from dotenv import load_dotenv
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.camb.tts import CambTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
load_dotenv(override=True)
# We store functions so objects (e.g. SileroVADAnalyzer) don't get
# instantiated. The function will be called when the desired transport gets
# selected.
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info("Starting Camb.ai TTS bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CambTTSService(
api_key=os.getenv("CAMB_API_KEY"),
model="mars-flash",
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
messages = [
{
"role": "system",
"content": "You are a helpful voice assistant powered by Camb.ai text-to-speech. "
"Keep your responses concise and conversational since they will be spoken aloud. "
"Avoid special characters, emojis, or bullet points.",
},
]
context = LLMContext(messages)
context_aggregator = LLMContextAggregatorPair(context)
pipeline = Pipeline(
[
transport.input(),
stt,
context_aggregator.user(),
llm,
tts,
transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info("Client connected")
messages.append({"role": "system", "content": "Please introduce yourself to the user."})
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info("Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

70
src/pipecat/services/camb/tts.py
View File

@@ -10,11 +10,10 @@ This module provides TTS functionality using Camb.ai's MARS model family,
offering high-quality text-to-speech synthesis with streaming support.
Features:
- MARS models: mars-flash, mars-pro, mars-instruct
- MARS models: mars-flash (fast), mars-pro (high quality)
- 140+ languages supported
- Real-time streaming via official SDK
- 48kHz audio output
- Voice customization (instructions for mars-instruct)
- Model-specific sample rates: mars-pro (48kHz), mars-flash (22.05kHz)
"""
from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional
@@ -41,11 +40,17 @@ from pipecat.utils.tracing.service_decorators import traced_tts
DEFAULT_VOICE_ID = 147320
DEFAULT_LANGUAGE = "en-us"
DEFAULT_MODEL = "mars-flash" # Faster inference
DEFAULT_SAMPLE_RATE = 48000 # 48kHz
DEFAULT_TIMEOUT = 60.0 # Seconds (minimum recommended by Camb.ai)
MIN_TEXT_LENGTH = 3
MAX_TEXT_LENGTH = 3000
# Model-specific sample rates
MODEL_SAMPLE_RATES: Dict[str, int] = {
"mars-flash": 22050, # 22.05kHz
"mars-pro": 48000, # 48kHz
"mars-instruct": 22050, # 22.05kHz
}
# Gender mapping for voice listing
GENDER_MAP = {0: "Not Specified", 1: "Male", 2: "Female", 9: "Not Applicable"}
@@ -131,30 +136,23 @@ class CambTTSService(TTSService):
"""Camb.ai MARS text-to-speech service using the official SDK.
Converts text to speech using Camb.ai's MARS TTS models with support for
multiple languages. Provides custom instructions support for the mars-instruct model.
multiple languages.
All models output 48kHz audio.
Models:
- mars-flash: Fast inference, 22.05kHz output (default)
- mars-pro: High quality, 48kHz output
Example::
# Basic usage with defaults
# Basic usage with defaults (mars-flash)
tts = CambTTSService(api_key="your-api-key")
# With custom voice and model
# High quality with mars-pro
tts = CambTTSService(
api_key="your-api-key",
voice_id=12345,
model="mars-pro",
)
# mars-instruct with custom instructions
tts = CambTTSService(
api_key="your-api-key",
model="mars-instruct",
params=CambTTSService.InputParams(
user_instructions="Speak with excitement and energy"
),
)
"""
class InputParams(BaseModel):
@@ -190,10 +188,10 @@ class CambTTSService(TTSService):
Args:
api_key: Camb.ai API key for authentication.
voice_id: Voice ID to use. Defaults to DEFAULT_VOICE_ID.
model: TTS model to use. Options: "mars-flash", "mars-pro", "mars-instruct".
Defaults to DEFAULT_MODEL (mars-flash, fastest).
model: TTS model to use. Options: "mars-flash" (fast), "mars-pro" (high quality).
Defaults to DEFAULT_MODEL (mars-flash).
timeout: Request timeout in seconds. Defaults to DEFAULT_TIMEOUT (60s).
sample_rate: Audio sample rate in Hz. If None, uses DEFAULT_SAMPLE_RATE (48kHz).
sample_rate: Audio sample rate in Hz. If None, uses model-specific default.
params: Additional voice parameters. If None, uses defaults.
**kwargs: Additional arguments passed to parent TTSService.
"""
@@ -243,9 +241,9 @@ class CambTTSService(TTSService):
frame: The start frame containing initialization parameters.
"""
await super().start(frame)
# Use 48kHz sample rate if not explicitly specified
# Use model-specific sample rate if not explicitly specified
if not self._init_sample_rate:
self._sample_rate = DEFAULT_SAMPLE_RATE
self._sample_rate = MODEL_SAMPLE_RATES.get(self._model_name, 22050)
self._settings["sample_rate"] = self._sample_rate
async def _update_settings(self, settings: Mapping[str, Any]):
@@ -310,15 +308,33 @@ class CambTTSService(TTSService):
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
# Buffer for aligning chunks to 2-byte boundaries (16-bit PCM)
audio_buffer = b""
# Stream audio chunks from SDK
async for chunk in self._client.text_to_speech.tts(**tts_kwargs):
if chunk:
await self.stop_ttfb_metrics()
yield TTSAudioRawFrame(
audio=chunk,
sample_rate=self.sample_rate,
num_channels=1,
)
audio_buffer += chunk
# Only yield complete 16-bit samples (2 bytes per sample)
aligned_size = (len(audio_buffer) // 2) * 2
if aligned_size > 0:
yield TTSAudioRawFrame(
audio=audio_buffer[:aligned_size],
sample_rate=self.sample_rate,
num_channels=1,
)
audio_buffer = audio_buffer[aligned_size:]
# Yield any remaining complete samples
if len(audio_buffer) >= 2:
aligned_size = (len(audio_buffer) // 2) * 2
yield TTSAudioRawFrame(
audio=audio_buffer[:aligned_size],
sample_rate=self.sample_rate,
num_channels=1,
)
except Exception as e:
error_msg = f"Camb.ai TTS error: {e}"

11
tests/test_camb_tts.py
View File

@@ -25,8 +25,8 @@ from pipecat.frames.frames import (
)
from pipecat.services.camb.tts import (
CambTTSService,
DEFAULT_SAMPLE_RATE,
DEFAULT_VOICE_ID,
MODEL_SAMPLE_RATES,
language_to_camb_language,
)
from pipecat.tests.utils import run_test
@@ -58,7 +58,8 @@ async def test_run_camb_tts_success():
tts_service = CambTTSService(api_key="test-api-key")
# Manually set sample rate (normally done by StartFrame)
tts_service._sample_rate = DEFAULT_SAMPLE_RATE
# mars-flash uses 22.05kHz
tts_service._sample_rate = MODEL_SAMPLE_RATES["mars-flash"]
# Test run_tts directly to avoid frame count variability
text = "Hello world, this is a test."
@@ -75,9 +76,9 @@ async def test_run_camb_tts_success():
audio_frames = [f for f in frames if isinstance(f, TTSAudioRawFrame)]
assert len(audio_frames) > 0, "Should have at least one audio frame"
# Verify sample rate matches 48kHz output
# Verify sample rate matches model output (mars-flash = 22.05kHz)
for a_frame in audio_frames:
assert a_frame.sample_rate == DEFAULT_SAMPLE_RATE
assert a_frame.sample_rate == MODEL_SAMPLE_RATES["mars-flash"]
assert a_frame.num_channels == 1, "Should be mono audio"
@@ -346,7 +347,7 @@ async def test_ttfb_metrics_tracked():
MockAsyncCambAI.return_value = mock_client
tts_service = CambTTSService(api_key="test-api-key")
tts_service._sample_rate = DEFAULT_SAMPLE_RATE
tts_service._sample_rate = MODEL_SAMPLE_RATES["mars-flash"]
# Patch the metrics methods to track calls
original_start_ttfb = tts_service.start_ttfb_metrics