diff --git a/examples/foundational/07zb-interruptible-camb-local.py b/examples/foundational/07zb-interruptible-camb-local.py index 8d7805c86..025577792 100644 --- a/examples/foundational/07zb-interruptible-camb-local.py +++ b/examples/foundational/07zb-interruptible-camb-local.py @@ -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, diff --git a/examples/foundational/07zb-interruptible-camb.py b/examples/foundational/07zb-interruptible-camb.py new file mode 100644 index 000000000..2c9c544b3 --- /dev/null +++ b/examples/foundational/07zb-interruptible-camb.py @@ -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() diff --git a/src/pipecat/services/camb/tts.py b/src/pipecat/services/camb/tts.py index ed34005d3..5dcf19a0d 100644 --- a/src/pipecat/services/camb/tts.py +++ b/src/pipecat/services/camb/tts.py @@ -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}" diff --git a/tests/test_camb_tts.py b/tests/test_camb_tts.py index ccd5c36ae..169a1c6ee 100644 --- a/tests/test_camb_tts.py +++ b/tests/test_camb_tts.py @@ -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