From 7c1e2793c57aecfca2cea910d97dcf6907b8e876 Mon Sep 17 00:00:00 2001 From: shreyas-sarvam Date: Thu, 9 Oct 2025 18:26:22 +0530 Subject: [PATCH 01/20] feat: Add support for bulbul:v3 and bulbul:v3-beta --- src/pipecat/services/sarvam/tts.py | 136 ++++++++++++++++++++++------- 1 file changed, 105 insertions(+), 31 deletions(-) diff --git a/src/pipecat/services/sarvam/tts.py b/src/pipecat/services/sarvam/tts.py index a9fedcc58..5bacffdf0 100644 --- a/src/pipecat/services/sarvam/tts.py +++ b/src/pipecat/services/sarvam/tts.py @@ -20,9 +20,9 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, - InterruptionFrame, LLMFullResponseEndFrame, StartFrame, + StartInterruptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -76,17 +76,29 @@ class SarvamHttpTTSService(TTSService): Example:: - tts = SarvamTTSService( + tts = SarvamHttpTTSService( api_key="your-api-key", voice_id="anushka", model="bulbul:v2", aiohttp_session=session, - params=SarvamTTSService.InputParams( + params=SarvamHttpTTSService.InputParams( language=Language.HI, pitch=0.1, pace=1.2 ) ) + + # For bulbul v3 beta with any speaker: + tts_v3 = SarvamHttpTTSService( + api_key="your-api-key", + voice_id="speaker_name", + model="bulbul:v3, + aiohttp_session=session, + params=SarvamHttpTTSService.InputParams( + language=Language.HI, + temperature=0.8 + ) + ) """ class InputParams(BaseModel): @@ -105,6 +117,11 @@ class SarvamHttpTTSService(TTSService): pace: Optional[float] = Field(default=1.0, ge=0.3, le=3.0) loudness: Optional[float] = Field(default=1.0, ge=0.1, le=3.0) enable_preprocessing: Optional[bool] = False + temperature: Optional[float] = Field( + default=0.6, + ge=0.01, + le=1.0, + ) def __init__( self, @@ -124,7 +141,7 @@ class SarvamHttpTTSService(TTSService): api_key: Sarvam AI API subscription key. aiohttp_session: Shared aiohttp session for making requests. voice_id: Speaker voice ID (e.g., "anushka", "meera"). Defaults to "anushka". - model: TTS model to use ("bulbul:v1" or "bulbul:v2"). Defaults to "bulbul:v2". + model: TTS model to use ("bulbul:v2" or "bulbul:v3-beta" or "bulbul:v3"). Defaults to "bulbul:v2". base_url: Sarvam AI API base URL. Defaults to "https://api.sarvam.ai". sample_rate: Audio sample rate in Hz (8000, 16000, 22050, 24000). If None, uses default. params: Additional voice and preprocessing parameters. If None, uses defaults. @@ -138,15 +155,31 @@ class SarvamHttpTTSService(TTSService): self._base_url = base_url self._session = aiohttp_session - self._settings = { - "language": ( - self.language_to_service_language(params.language) if params.language else "en-IN" - ), - "pitch": params.pitch, - "pace": params.pace, - "loudness": params.loudness, - "enable_preprocessing": params.enable_preprocessing, - } + if model == "bulbul:v3-beta" or model == "bulbul:v3": + # For bulbul v3 beta, exclude pace and loudness parameters + self._settings = { + "language": ( + self.language_to_service_language(params.language) + if params.language + else "en-IN" + ), + "enable_preprocessing": params.enable_preprocessing, + "temperature": getattr(params, "temperature", 0.6), + "model": model, # Include model in settings for v3 beta + } + else: + # For bulbul v2, include all parameters including pace and loudness + self._settings = { + "language": ( + self.language_to_service_language(params.language) + if params.language + else "en-IN" + ), + "pitch": params.pitch, + "pace": params.pace, + "loudness": params.loudness, + "enable_preprocessing": params.enable_preprocessing, + } self.set_model_name(model) self.set_voice(voice_id) @@ -275,6 +308,18 @@ class SarvamTTSService(InterruptibleTTSService): pace=1.2 ) ) + + # For bulbul v3 beta with any speaker and temperature: + # Note: pace and loudness are not supported for bulbul v3 beta + tts_v3 = SarvamTTSService( + api_key="your-api-key", + voice_id="speaker_name", + model="bulbul:v3", + params=SarvamTTSService.InputParams( + language=Language.HI, + temperature=0.8 + ) + ) """ class InputParams(BaseModel): @@ -310,6 +355,14 @@ class SarvamTTSService(InterruptibleTTSService): output_audio_codec: Optional[str] = "linear16" output_audio_bitrate: Optional[str] = "128k" language: Optional[Language] = Language.EN + temperature: Optional[float] = Field( + default=0.6, + ge=0.01, + le=1.0, + description="Controls the randomness of the output for bulbul v3 beta. " + "Lower values make the output more focused and deterministic, while " + "higher values make it more random. Range: 0.01 to 1.0. Default: 0.6.", + ) def __init__( self, @@ -329,13 +382,12 @@ class SarvamTTSService(InterruptibleTTSService): Args: api_key: Sarvam API key for authenticating TTS requests. model: Identifier of the Sarvam speech model (default "bulbul:v2"). + Supports "bulbul:v2", "bulbul:v3-beta" and "bulbul:v3". voice_id: Voice identifier for synthesis (default "anushka"). url: WebSocket URL for connecting to the TTS backend (default production URL). aiohttp_session: Optional shared aiohttp session. To maintain backward compatibility. - .. deprecated:: 0.0.81 aiohttp_session is no longer used. This parameter will be removed in a future version. - aggregate_sentences: Whether to merge multiple sentences into one audio chunk (default True). sample_rate: Desired sample rate for the output audio in Hz (overrides default if set). params: Optional input parameters to override global configuration. @@ -372,21 +424,43 @@ class SarvamTTSService(InterruptibleTTSService): self.set_model_name(model) self.set_voice(voice_id) # Configuration parameters - self._settings = { - "target_language_code": ( - self.language_to_service_language(params.language) if params.language else "en-IN" - ), - "pitch": params.pitch, - "pace": params.pace, - "speaker": voice_id, - "loudness": params.loudness, - "speech_sample_rate": 0, - "enable_preprocessing": params.enable_preprocessing, - "min_buffer_size": params.min_buffer_size, - "max_chunk_length": params.max_chunk_length, - "output_audio_codec": params.output_audio_codec, - "output_audio_bitrate": params.output_audio_bitrate, - } + if model == "bulbul:v3-beta" or model == "bulbul:v3": + # For bulbul v3 beta, exclude pace and loudness parameters + self._settings = { + "target_language_code": ( + self.language_to_service_language(params.language) + if params.language + else "en-IN" + ), + "speaker": voice_id, + "speech_sample_rate": 0, + "enable_preprocessing": params.enable_preprocessing, + "min_buffer_size": params.min_buffer_size, + "max_chunk_length": params.max_chunk_length, + "output_audio_codec": params.output_audio_codec, + "output_audio_bitrate": params.output_audio_bitrate, + "temperature": getattr(params, "temperature", 0.6), + "model": model, # Include model in settings for v3 beta + } + else: + # For bulbul v2, include all parameters including pace and loudness + self._settings = { + "target_language_code": ( + self.language_to_service_language(params.language) + if params.language + else "en-IN" + ), + "pitch": params.pitch, + "pace": params.pace, + "speaker": voice_id, + "loudness": params.loudness, + "speech_sample_rate": 0, + "enable_preprocessing": params.enable_preprocessing, + "min_buffer_size": params.min_buffer_size, + "max_chunk_length": params.max_chunk_length, + "output_audio_codec": params.output_audio_codec, + "output_audio_bitrate": params.output_audio_bitrate, + } self._started = False self._receive_task = None @@ -455,7 +529,7 @@ class SarvamTTSService(InterruptibleTTSService): direction: The direction to push the frame. """ await super().push_frame(frame, direction) - if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): + if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): self._started = False async def process_frame(self, frame: Frame, direction: FrameDirection): From 9babfe9fd9cbded098d32fc787644db5089653fc Mon Sep 17 00:00:00 2001 From: shreyas-sarvam Date: Mon, 13 Oct 2025 08:54:29 +0530 Subject: [PATCH 02/20] refactor: Improve code reability and replace deprecated interruption frames --- src/pipecat/services/sarvam/tts.py | 126 ++++++++++++++--------------- 1 file changed, 62 insertions(+), 64 deletions(-) diff --git a/src/pipecat/services/sarvam/tts.py b/src/pipecat/services/sarvam/tts.py index 5bacffdf0..ae48e1ee5 100644 --- a/src/pipecat/services/sarvam/tts.py +++ b/src/pipecat/services/sarvam/tts.py @@ -20,9 +20,9 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + InterruptionFrame, LLMFullResponseEndFrame, StartFrame, - StartInterruptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -121,6 +121,9 @@ class SarvamHttpTTSService(TTSService): default=0.6, ge=0.01, le=1.0, + description="Controls the randomness of the output for bulbul v3 beta. " + "Lower values make the output more focused and deterministic, while " + "higher values make it more random. Range: 0.01 to 1.0. Default: 0.6.", ) def __init__( @@ -155,31 +158,31 @@ class SarvamHttpTTSService(TTSService): self._base_url = base_url self._session = aiohttp_session - if model == "bulbul:v3-beta" or model == "bulbul:v3": - # For bulbul v3 beta, exclude pace and loudness parameters - self._settings = { - "language": ( - self.language_to_service_language(params.language) - if params.language - else "en-IN" - ), - "enable_preprocessing": params.enable_preprocessing, - "temperature": getattr(params, "temperature", 0.6), - "model": model, # Include model in settings for v3 beta - } + # Build base settings common to all models + self._settings = { + "language": ( + self.language_to_service_language(params.language) if params.language else "en-IN" + ), + "enable_preprocessing": params.enable_preprocessing, + } + + # Add model-specific parameters + if model in ("bulbul:v3-beta", "bulbul:v3"): + self._settings.update( + { + "temperature": getattr(params, "temperature", 0.6), + "model": model, + } + ) else: - # For bulbul v2, include all parameters including pace and loudness - self._settings = { - "language": ( - self.language_to_service_language(params.language) - if params.language - else "en-IN" - ), - "pitch": params.pitch, - "pace": params.pace, - "loudness": params.loudness, - "enable_preprocessing": params.enable_preprocessing, - } + self._settings.update( + { + "pitch": params.pitch, + "pace": params.pace, + "loudness": params.loudness, + "model": model, + } + ) self.set_model_name(model) self.set_voice(voice_id) @@ -310,7 +313,7 @@ class SarvamTTSService(InterruptibleTTSService): ) # For bulbul v3 beta with any speaker and temperature: - # Note: pace and loudness are not supported for bulbul v3 beta + # Note: pace and loudness are not supported for bulbul v3 and bulbul v3 beta tts_v3 = SarvamTTSService( api_key="your-api-key", voice_id="speaker_name", @@ -386,8 +389,10 @@ class SarvamTTSService(InterruptibleTTSService): voice_id: Voice identifier for synthesis (default "anushka"). url: WebSocket URL for connecting to the TTS backend (default production URL). aiohttp_session: Optional shared aiohttp session. To maintain backward compatibility. + .. deprecated:: 0.0.81 aiohttp_session is no longer used. This parameter will be removed in a future version. + aggregate_sentences: Whether to merge multiple sentences into one audio chunk (default True). sample_rate: Desired sample rate for the output audio in Hz (overrides default if set). params: Optional input parameters to override global configuration. @@ -423,44 +428,37 @@ class SarvamTTSService(InterruptibleTTSService): self._api_key = api_key self.set_model_name(model) self.set_voice(voice_id) - # Configuration parameters - if model == "bulbul:v3-beta" or model == "bulbul:v3": - # For bulbul v3 beta, exclude pace and loudness parameters - self._settings = { - "target_language_code": ( - self.language_to_service_language(params.language) - if params.language - else "en-IN" - ), - "speaker": voice_id, - "speech_sample_rate": 0, - "enable_preprocessing": params.enable_preprocessing, - "min_buffer_size": params.min_buffer_size, - "max_chunk_length": params.max_chunk_length, - "output_audio_codec": params.output_audio_codec, - "output_audio_bitrate": params.output_audio_bitrate, - "temperature": getattr(params, "temperature", 0.6), - "model": model, # Include model in settings for v3 beta - } + # Build base settings common to all models + self._settings = { + "target_language_code": ( + self.language_to_service_language(params.language) if params.language else "en-IN" + ), + "speaker": voice_id, + "speech_sample_rate": 0, + "enable_preprocessing": params.enable_preprocessing, + "min_buffer_size": params.min_buffer_size, + "max_chunk_length": params.max_chunk_length, + "output_audio_codec": params.output_audio_codec, + "output_audio_bitrate": params.output_audio_bitrate, + } + + # Add model-specific parameters + if model in ("bulbul:v3-beta", "bulbul:v3"): + self._settings.update( + { + "temperature": getattr(params, "temperature", 0.6), + "model": model, + } + ) else: - # For bulbul v2, include all parameters including pace and loudness - self._settings = { - "target_language_code": ( - self.language_to_service_language(params.language) - if params.language - else "en-IN" - ), - "pitch": params.pitch, - "pace": params.pace, - "speaker": voice_id, - "loudness": params.loudness, - "speech_sample_rate": 0, - "enable_preprocessing": params.enable_preprocessing, - "min_buffer_size": params.min_buffer_size, - "max_chunk_length": params.max_chunk_length, - "output_audio_codec": params.output_audio_codec, - "output_audio_bitrate": params.output_audio_bitrate, - } + self._settings.update( + { + "pitch": params.pitch, + "pace": params.pace, + "loudness": params.loudness, + "model": model, + } + ) self._started = False self._receive_task = None @@ -529,7 +527,7 @@ class SarvamTTSService(InterruptibleTTSService): direction: The direction to push the frame. """ await super().push_frame(frame, direction) - if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): + if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): self._started = False async def process_frame(self, frame: Frame, direction: FrameDirection): From 9ddec0f8b491cdbe4513a494449eaaa9734dde57 Mon Sep 17 00:00:00 2001 From: nbyers-altira Date: Mon, 13 Oct 2025 10:44:25 -0400 Subject: [PATCH 03/20] is_final is not part of the segmented _handle_transcription function signature --- src/pipecat/services/riva/stt.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/services/riva/stt.py b/src/pipecat/services/riva/stt.py index d00eb4f42..f077355f4 100644 --- a/src/pipecat/services/riva/stt.py +++ b/src/pipecat/services/riva/stt.py @@ -647,7 +647,7 @@ class RivaSegmentedSTTService(SegmentedSTTService): ) transcription_found = True - await self._handle_transcription(text, True, self._language_enum) + await self._handle_transcription(text, self._language_enum) if not transcription_found: logger.debug("No transcription results found in Riva response") From cc66ac14f1699357bc2e39be9adac376721bbcac Mon Sep 17 00:00:00 2001 From: nbyers-altira Date: Mon, 13 Oct 2025 10:48:41 -0400 Subject: [PATCH 04/20] add is_final to segmented func. sig. instead so tracing is consistent --- src/pipecat/services/riva/stt.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pipecat/services/riva/stt.py b/src/pipecat/services/riva/stt.py index f077355f4..ccd166536 100644 --- a/src/pipecat/services/riva/stt.py +++ b/src/pipecat/services/riva/stt.py @@ -583,7 +583,7 @@ class RivaSegmentedSTTService(SegmentedSTTService): self._config.language_code = self._language @traced_stt - async def _handle_transcription(self, transcript: str, language: Optional[Language] = None): + async def _handle_transcription(self, transcript: str, is_final: bool, language: Optional[Language] = None): """Handle a transcription result with tracing.""" pass @@ -647,7 +647,7 @@ class RivaSegmentedSTTService(SegmentedSTTService): ) transcription_found = True - await self._handle_transcription(text, self._language_enum) + await self._handle_transcription(text, True, self._language_enum) if not transcription_found: logger.debug("No transcription results found in Riva response") From d16c36c56de1e3289594c4b2081b65c3425c0d5a Mon Sep 17 00:00:00 2001 From: dan-ince-aai Date: Wed, 15 Oct 2025 14:27:52 +0100 Subject: [PATCH 05/20] feat: add keyterms_prompt to AssemblyAI service --- src/pipecat/services/assemblyai/models.py | 2 ++ src/pipecat/services/assemblyai/stt.py | 15 ++++++++++----- 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/src/pipecat/services/assemblyai/models.py b/src/pipecat/services/assemblyai/models.py index b34ec554d..d263d113d 100644 --- a/src/pipecat/services/assemblyai/models.py +++ b/src/pipecat/services/assemblyai/models.py @@ -108,6 +108,7 @@ class AssemblyAIConnectionParams(BaseModel): end_of_turn_confidence_threshold: Confidence threshold for end-of-turn detection. min_end_of_turn_silence_when_confident: Minimum silence duration when confident about end-of-turn. max_turn_silence: Maximum silence duration before forcing end-of-turn. + keyterms_prompt: List of key terms to guide transcription. Will be JSON serialized before sending. """ sample_rate: int = 16000 @@ -117,3 +118,4 @@ class AssemblyAIConnectionParams(BaseModel): end_of_turn_confidence_threshold: Optional[float] = None min_end_of_turn_silence_when_confident: Optional[int] = None max_turn_silence: Optional[int] = None + keyterms_prompt: Optional[List[str]] = None diff --git a/src/pipecat/services/assemblyai/stt.py b/src/pipecat/services/assemblyai/stt.py index aa2fc36bc..d87261f86 100644 --- a/src/pipecat/services/assemblyai/stt.py +++ b/src/pipecat/services/assemblyai/stt.py @@ -174,11 +174,16 @@ class AssemblyAISTTService(STTService): def _build_ws_url(self) -> str: """Build WebSocket URL with query parameters using urllib.parse.urlencode.""" - params = { - k: str(v).lower() if isinstance(v, bool) else v - for k, v in self._connection_params.model_dump().items() - if v is not None - } + params = {} + for k, v in self._connection_params.model_dump().items(): + if v is not None: + if k == "keyterms_prompt": + params[k] = json.dumps(v) + elif isinstance(v, bool): + params[k] = str(v).lower() + else: + params[k] = v + if params: query_string = urlencode(params) return f"{self._api_endpoint_base_url}?{query_string}" From 41f817bf04d0c96217f2fce6adb2f0962f9cc4e5 Mon Sep 17 00:00:00 2001 From: vipyne Date: Wed, 15 Oct 2025 13:02:44 -0500 Subject: [PATCH 06/20] only import whatsapp deps if using whatsapp runner --- src/pipecat/runner/run.py | 21 ++++++++------------- 1 file changed, 8 insertions(+), 13 deletions(-) diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index a7af52e18..c5c9dc9db 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -289,19 +289,6 @@ def _add_lifespan_to_app(app: FastAPI, new_lifespan): def _setup_whatsapp_routes(app: FastAPI): """Set up WebRTC-specific routes.""" - try: - from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI - - from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection - from pipecat.transports.smallwebrtc.request_handler import ( - SmallWebRTCRequest, - SmallWebRTCRequestHandler, - ) - from pipecat.transports.whatsapp.api import WhatsAppWebhookRequest - from pipecat.transports.whatsapp.client import WhatsAppClient - except ImportError as e: - logger.error(f"WebRTC transport dependencies not installed: {e}") - return WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN") WHATSAPP_PHONE_NUMBER_ID = os.getenv("WHATSAPP_PHONE_NUMBER_ID") @@ -320,6 +307,14 @@ def _setup_whatsapp_routes(app: FastAPI): ) return + # only import WhatsApp dependencies if appropriate env vars are present + try: + from pipecat.transports.whatsapp.api import WhatsAppWebhookRequest + from pipecat.transports.whatsapp.client import WhatsAppClient + except ImportError as e: + logger.error(f"WhatsApp transport dependencies not installed: {e}") + return + # Global WhatsApp client instance whatsapp_client: Optional[WhatsAppClient] = None From 0fd5d2610484909b6162bf4ee528ca221bf20ce6 Mon Sep 17 00:00:00 2001 From: vipyne Date: Wed, 15 Oct 2025 13:16:16 -0500 Subject: [PATCH 07/20] add WHATSAPP_APP_SECRET to required whatsapp env vars --- src/pipecat/runner/run.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index c5c9dc9db..4663b115d 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -289,7 +289,6 @@ def _add_lifespan_to_app(app: FastAPI, new_lifespan): def _setup_whatsapp_routes(app: FastAPI): """Set up WebRTC-specific routes.""" - WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN") WHATSAPP_PHONE_NUMBER_ID = os.getenv("WHATSAPP_PHONE_NUMBER_ID") WHATSAPP_WEBHOOK_VERIFICATION_TOKEN = os.getenv("WHATSAPP_WEBHOOK_VERIFICATION_TOKEN") @@ -300,10 +299,17 @@ def _setup_whatsapp_routes(app: FastAPI): WHATSAPP_TOKEN, WHATSAPP_PHONE_NUMBER_ID, WHATSAPP_WEBHOOK_VERIFICATION_TOKEN, + WHATSAPP_APP_SECRET, ] ): logger.trace( - "Missing required environment variables for WhatsApp transport. Keeping it disabled." + """Missing required environment variables for WhatsApp transport: + WHATSAPP_TOKEN + WHATSAPP_PHONE_NUMBER_ID + WHATSAPP_WEBHOOK_VERIFICATION_TOKEN + WHATSAPP_APP_SECRET +Keeping it disabled. + """ ) return From 6381335346ba9622f2eced03aca07891af0c4223 Mon Sep 17 00:00:00 2001 From: vipyne Date: Thu, 16 Oct 2025 10:37:46 -0500 Subject: [PATCH 08/20] Add --whatsapp flag to runner --- CHANGELOG.md | 2 ++ src/pipecat/runner/run.py | 45 ++++++++++++++++++++++++++------------- 2 files changed, 32 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3cd715776..2b078fe80 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `--whatsapp` flag to runner to better surface WhatsApp transport logs. + - Added `on_connected` and `on_disconnected` events to TTS and STT websocket-based services. diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index 4663b115d..a3e2984e8 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -166,6 +166,7 @@ def _create_server_app( host: str = "localhost", proxy: str, esp32_mode: bool = False, + whatsapp_enabled: bool = False, folder: Optional[str] = None, ): """Create FastAPI app with transport-specific routes.""" @@ -182,7 +183,8 @@ def _create_server_app( # Set up transport-specific routes if transport_type == "webrtc": _setup_webrtc_routes(app, esp32_mode=esp32_mode, host=host, folder=folder) - _setup_whatsapp_routes(app) + if whatsapp_enabled: + _setup_whatsapp_routes(app) elif transport_type == "daily": _setup_daily_routes(app) elif transport_type in TELEPHONY_TRANSPORTS: @@ -289,32 +291,37 @@ def _add_lifespan_to_app(app: FastAPI, new_lifespan): def _setup_whatsapp_routes(app: FastAPI): """Set up WebRTC-specific routes.""" - WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN") - WHATSAPP_PHONE_NUMBER_ID = os.getenv("WHATSAPP_PHONE_NUMBER_ID") - WHATSAPP_WEBHOOK_VERIFICATION_TOKEN = os.getenv("WHATSAPP_WEBHOOK_VERIFICATION_TOKEN") WHATSAPP_APP_SECRET = os.getenv("WHATSAPP_APP_SECRET") + WHATSAPP_PHONE_NUMBER_ID = os.getenv("WHATSAPP_PHONE_NUMBER_ID") + WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN") + WHATSAPP_WEBHOOK_VERIFICATION_TOKEN = os.getenv("WHATSAPP_WEBHOOK_VERIFICATION_TOKEN") if not all( [ - WHATSAPP_TOKEN, - WHATSAPP_PHONE_NUMBER_ID, - WHATSAPP_WEBHOOK_VERIFICATION_TOKEN, WHATSAPP_APP_SECRET, + WHATSAPP_PHONE_NUMBER_ID, + WHATSAPP_TOKEN, + WHATSAPP_WEBHOOK_VERIFICATION_TOKEN, ] ): - logger.trace( + logger.error( """Missing required environment variables for WhatsApp transport: - WHATSAPP_TOKEN - WHATSAPP_PHONE_NUMBER_ID - WHATSAPP_WEBHOOK_VERIFICATION_TOKEN WHATSAPP_APP_SECRET -Keeping it disabled. + WHATSAPP_PHONE_NUMBER_ID + WHATSAPP_TOKEN + WHATSAPP_WEBHOOK_VERIFICATION_TOKEN """ ) return - # only import WhatsApp dependencies if appropriate env vars are present try: + from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI + + from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection + from pipecat.transports.smallwebrtc.request_handler import ( + SmallWebRTCRequest, + SmallWebRTCRequestHandler, + ) from pipecat.transports.whatsapp.api import WhatsAppWebhookRequest from pipecat.transports.whatsapp.client import WhatsAppClient except ImportError as e: @@ -690,6 +697,12 @@ def main(): parser.add_argument( "--verbose", "-v", action="count", default=0, help="Increase logging verbosity" ) + parser.add_argument( + "--whatsapp", + action="store_true", + default=False, + help="Ensure requried WhatsApp environment variables are present", + ) args = parser.parse_args() @@ -732,10 +745,11 @@ def main(): print() if args.esp32: print(f"🚀 Bot ready! (ESP32 mode)") - print(f" → Open http://{args.host}:{args.port}/client in your browser") + elif args.whatsapp: + print(f"🚀 Bot ready! (WhatsApp)") else: print(f"🚀 Bot ready!") - print(f" → Open http://{args.host}:{args.port}/client in your browser") + print(f" → Open http://{args.host}:{args.port}/client in your browser") print() elif args.transport == "daily": print() @@ -753,6 +767,7 @@ def main(): host=args.host, proxy=args.proxy, esp32_mode=args.esp32, + whatsapp_enabled=args.whatsapp, folder=args.folder, ) From 1b9e96c0168facfef2f704af0d329553135eda49 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 15 Oct 2025 22:37:16 -0700 Subject: [PATCH 09/20] PipelineTask: fix task cancellation issues --- CHANGELOG.md | 8 ++- src/pipecat/pipeline/runner.py | 8 ++- src/pipecat/pipeline/task.py | 99 +++++++++++++++------------------- tests/test_pipeline.py | 31 +++++------ 4 files changed, 67 insertions(+), 79 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3cd715776..c9f645589 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,12 +25,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `CartesiaSTTService` now inherits from `WebsocketSTTService`. -- Package upgrades: +# Package upgrades: - `openai` upgraded to support up to 2.x.x. - `openpipe` upgraded to support up to 5.x.x. ### Fixed +- Fixed multiple pipeline task cancellation issues. `asyncio.CancelledError` is + now handled properly in `PipelineTask` making it possible to cancel an asyncio + task that it's executing a `PipelineRunner` cleanly. Also, + `PipelineTask.cancel()` does not block anymore waiting for the `CancelFrame` + to reach the end of the pipeline (going back to the behavior in < 0.0.83). + - Fixed an issue in `ElevenLabsTTSService` and `ElevenLabsHttpTTSService` where the Flash models would split words, resulting in a space being inserted between words. diff --git a/src/pipecat/pipeline/runner.py b/src/pipecat/pipeline/runner.py index 9d82fcd88..c7e246681 100644 --- a/src/pipecat/pipeline/runner.py +++ b/src/pipecat/pipeline/runner.py @@ -70,11 +70,15 @@ class PipelineRunner(BaseObject): """ logger.debug(f"Runner {self} started running {task}") self._tasks[task.name] = task - params = PipelineTaskParams(loop=self._loop) + + # PipelineTask handles asyncio.CancelledError to shutdown the pipeline + # properly and re-raises it in case there's more cleanup to do. try: + params = PipelineTaskParams(loop=self._loop) await task.run(params) except asyncio.CancelledError: - await self._cancel() + pass + del self._tasks[task.name] # Cleanup base object. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index f6d17262e..a511db12a 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -269,6 +269,9 @@ class PipelineTask(BasePipelineTask): # StopFrame) has been received at the end of the pipeline. self._pipeline_end_event = asyncio.Event() + # This event is set when the pipeline truly finishes. + self._pipeline_finished_event = asyncio.Event() + # This is the final pipeline. It is composed of a source processor, # followed by the user pipeline, and ending with a sink processor. The # source allows us to receive and react to upstream frames, and the sink @@ -401,11 +404,7 @@ class PipelineTask(BasePipelineTask): await self.queue_frame(EndFrame()) async def cancel(self): - """Immediately stop the running pipeline. - - Cancels all running tasks and stops frame processing without - waiting for completion. - """ + """Request the running pipeline to cancel.""" if not self._finished: await self._cancel() @@ -417,51 +416,38 @@ class PipelineTask(BasePipelineTask): """ if self.has_finished(): return - cleanup_pipeline = True + + # Setup processors. + await self._setup(params) + + # Create all main tasks and wait for the main push task. This is the + # task that pushes frames to the very beginning of our pipeline (i.e. to + # our controlled source processor). + await self._create_tasks() + try: - # Setup processors. - await self._setup(params) - - # Create all main tasks and wait of the main push task. This is the - # task that pushes frames to the very beginning of our pipeline (our - # controlled source processor). - push_task = await self._create_tasks() - await push_task - - # We have already cleaned up the pipeline inside the task. - cleanup_pipeline = False - - # Pipeline has finished nicely. - self._finished = True + # Wait for pipeline to finish. + await self._wait_for_pipeline_finished() except asyncio.CancelledError: - # Raise exception back to the pipeline runner so it can cancel this - # task properly. + logger.debug(f"Pipeline task {self} got cancelled from outside...") + # We have been cancelled from outside, let's just cancel everything. + await self._cancel() + # Wait again for pipeline to finish. This time we have really + # cancelled, so it should really finish. + await self._wait_for_pipeline_finished() + # Re-raise in case there's more cleanup to do. raise finally: # We can reach this point for different reasons: # - # 1. The task has finished properly (e.g. `EndFrame`). - # 2. By calling `PipelineTask.cancel()`. - # 3. By asyncio task cancellation. - # - # Case (1) will execute the code below without issues because - # `self._finished` is true. - # - # Case (2) will execute the code below without issues because - # `self._cancelled` is true. - # - # Case (3) will raise the exception above (because we are cancelling - # the asyncio task). This will be then captured by the - # `PipelineRunner` which will call `PipelineTask.cancel()` and - # therefore becoming case (2). - if self._finished or self._cancelled: - logger.debug(f"Pipeline task {self} is finishing cleanup...") - await self._cancel_tasks() - await self._cleanup(cleanup_pipeline) - if self._check_dangling_tasks: - self._print_dangling_tasks() - self._finished = True - logger.debug(f"Pipeline task {self} has finished") + # 1. The pipeline task has finished (try case). + # 2. By an asyncio task cancellation (except case). + logger.debug(f"Pipeline task {self} is finishing...") + await self._cancel_tasks() + if self._check_dangling_tasks: + self._print_dangling_tasks() + self._finished = True + logger.debug(f"Pipeline task {self} has finished") async def queue_frame(self, frame: Frame): """Queue a single frame to be pushed down the pipeline. @@ -489,19 +475,7 @@ class PipelineTask(BasePipelineTask): if not self._cancelled: logger.debug(f"Cancelling pipeline task {self}") self._cancelled = True - cancel_frame = CancelFrame() - # Make sure everything is cleaned up downstream. This is sent - # out-of-band from the main streaming task which is what we want since - # we want to cancel right away. - await self._pipeline.queue_frame(cancel_frame) - # Wait for CancelFrame to make it through the pipeline. - await self._wait_for_pipeline_end(cancel_frame) - # Only cancel the push task, we don't want to be able to process any - # other frame after cancel. Everything else will be cancelled in - # run(). - if self._process_push_task: - await self._task_manager.cancel_task(self._process_push_task) - self._process_push_task = None + await self.queue_frame(CancelFrame()) async def _create_tasks(self): """Create and start all pipeline processing tasks.""" @@ -603,6 +577,17 @@ class PipelineTask(BasePipelineTask): self._pipeline_end_event.clear() + # We are really done. + self._pipeline_finished_event.set() + + async def _wait_for_pipeline_finished(self): + await self._pipeline_finished_event.wait() + self._pipeline_finished_event.clear() + # Make sure we wait for the main task to complete. + if self._process_push_task: + await self._process_push_task + self._process_push_task = None + async def _setup(self, params: PipelineTaskParams): """Set up the pipeline task and all processors.""" mgr_params = TaskManagerParams(loop=params.loop) diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index df4af49b9..7d9ebec6f 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -254,7 +254,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): try: await asyncio.wait_for( - asyncio.shield(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))), + task.run(PipelineTaskParams(loop=asyncio.get_event_loop())), timeout=1.0, ) except asyncio.TimeoutError: @@ -290,7 +290,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): await task.queue_frame(TextFrame(text="Hello!")) try: await asyncio.wait_for( - asyncio.shield(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))), + task.run(PipelineTaskParams(loop=asyncio.get_event_loop())), timeout=1.0, ) except asyncio.TimeoutError: @@ -301,11 +301,8 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): identity = IdentityFilter() pipeline = Pipeline([identity]) task = PipelineTask(pipeline, idle_timeout_secs=0.2) - try: - await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) - assert False - except asyncio.CancelledError: - assert True + # This shouldn't freeze, so nothing to check really. + await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) async def test_no_idle_task(self): identity = IdentityFilter() @@ -313,7 +310,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): task = PipelineTask(pipeline, idle_timeout_secs=0.2, cancel_on_idle_timeout=False) try: await asyncio.wait_for( - asyncio.shield(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))), + task.run(PipelineTaskParams(loop=asyncio.get_event_loop())), timeout=0.3, ) except asyncio.TimeoutError: @@ -332,11 +329,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): ), idle_timeout_secs=0.3, ) - try: - await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) - assert False - except asyncio.CancelledError: - assert True + await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) async def test_idle_task_event_handler_no_frames(self): identity = IdentityFilter() @@ -351,11 +344,8 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): idle_timeout = True await task.cancel() - try: - await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) - assert False - except asyncio.CancelledError: - assert idle_timeout + await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) + assert idle_timeout async def test_idle_task_event_handler_quiet_user(self): identity = IdentityFilter() @@ -416,12 +406,15 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): asyncio.create_task(delayed_frames()), ] - await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + _, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) diff_time = time.time() - start_time self.assertGreater(diff_time, sleep_time_secs * 3) + # Wait for the pending tasks to complete. + await asyncio.gather(*pending) + async def test_task_cancel_timeout(self): class CancelFilter(FrameProcessor): def __init__(self, **kwargs): From 8d8503bca7ac3a21ca9ad3eeb516cacdea648346 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 15 Oct 2025 16:20:12 -0700 Subject: [PATCH 10/20] OpenAITTSService: allow updating instructions and speed --- CHANGELOG.md | 5 +++- src/pipecat/services/openai/tts.py | 43 +++++++++++++++++++++++++----- 2 files changed, 41 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a5f114054..938b175c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added support for updating `OpenAITTSService` settings (`instructions` and + `speed`) at runtime via `TTSUpdateSettingsFrame`. + - Added `--whatsapp` flag to runner to better surface WhatsApp transport logs. - Added `on_connected` and `on_disconnected` events to TTS and STT @@ -27,7 +30,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `CartesiaSTTService` now inherits from `WebsocketSTTService`. -# Package upgrades: +- Package upgrades: - `openai` upgraded to support up to 2.x.x. - `openpipe` upgraded to support up to 5.x.x. diff --git a/src/pipecat/services/openai/tts.py b/src/pipecat/services/openai/tts.py index a67392caf..cdf0d11ac 100644 --- a/src/pipecat/services/openai/tts.py +++ b/src/pipecat/services/openai/tts.py @@ -14,6 +14,7 @@ from typing import AsyncGenerator, Dict, Literal, Optional from loguru import logger from openai import AsyncOpenAI, BadRequestError +from pydantic import BaseModel from pipecat.frames.frames import ( ErrorFrame, @@ -55,6 +56,17 @@ class OpenAITTSService(TTSService): OPENAI_SAMPLE_RATE = 24000 # OpenAI TTS always outputs at 24kHz + class InputParams(BaseModel): + """Input parameters for OpenAI TTS configuration. + + Parameters: + instructions: Instructions to guide voice synthesis behavior. + speed: Voice speed control (0.25 to 4.0, default 1.0). + """ + + instructions: Optional[str] = None + speed: Optional[float] = None + def __init__( self, *, @@ -65,6 +77,7 @@ class OpenAITTSService(TTSService): sample_rate: Optional[int] = None, instructions: Optional[str] = None, speed: Optional[float] = None, + params: Optional[InputParams] = None, **kwargs, ): """Initialize OpenAI TTS service. @@ -77,7 +90,11 @@ class OpenAITTSService(TTSService): sample_rate: Output audio sample rate in Hz. If None, uses OpenAI's default 24kHz. instructions: Optional instructions to guide voice synthesis behavior. speed: Voice speed control (0.25 to 4.0, default 1.0). + params: Optional synthesis controls (acting instructions, speed, ...). **kwargs: Additional keyword arguments passed to TTSService. + + .. deprecated:: 0.0.91 + The `instructions` and `speed` parameters are deprecated, use `InputParams` instead. """ if sample_rate and sample_rate != self.OPENAI_SAMPLE_RATE: logger.warning( @@ -86,12 +103,26 @@ class OpenAITTSService(TTSService): ) super().__init__(sample_rate=sample_rate, **kwargs) - self._speed = speed self.set_model_name(model) self.set_voice(voice) - self._instructions = instructions self._client = AsyncOpenAI(api_key=api_key, base_url=base_url) + if instructions or speed: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "The `instructions` and `speed` parameters are deprecated, use `InputParams` instead.", + DeprecationWarning, + stacklevel=2, + ) + + self._settings = { + "instructions": params.instructions if params else instructions, + "speed": params.speed if params else speed, + } + def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -144,11 +175,11 @@ class OpenAITTSService(TTSService): "response_format": "pcm", } - if self._instructions: - create_params["instructions"] = self._instructions + if self._settings["instructions"]: + create_params["instructions"] = self._settings["instructions"] - if self._speed: - create_params["speed"] = self._speed + if self._settings["speed"]: + create_params["speed"] = self._settings["speed"] async with self._client.audio.speech.with_streaming_response.create( **create_params From d12134038b1e4ba63fbc122f3f9ab5acc50d6256 Mon Sep 17 00:00:00 2001 From: shreyas-sarvam Date: Fri, 17 Oct 2025 10:07:58 +0530 Subject: [PATCH 11/20] chore: Update CHANGELOG --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 938b175c1..89078ebf1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added support for `bulbul:v3` model in `SarvamTTSService` and `SarvamHttpTTSService`. + - Added support for updating `OpenAITTSService` settings (`instructions` and `speed`) at runtime via `TTSUpdateSettingsFrame`. From d5f2dcfac0a020d8077988de71be47b210d90211 Mon Sep 17 00:00:00 2001 From: dan-ince-aai Date: Fri, 17 Oct 2025 11:32:06 +0100 Subject: [PATCH 12/20] lint --- src/pipecat/services/assemblyai/stt.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/services/assemblyai/stt.py b/src/pipecat/services/assemblyai/stt.py index d87261f86..c255cd4c4 100644 --- a/src/pipecat/services/assemblyai/stt.py +++ b/src/pipecat/services/assemblyai/stt.py @@ -183,7 +183,7 @@ class AssemblyAISTTService(STTService): params[k] = str(v).lower() else: params[k] = v - + if params: query_string = urlencode(params) return f"{self._api_endpoint_base_url}?{query_string}" From 3ba6b5565949023f0f31beac883e114e1b0df7a4 Mon Sep 17 00:00:00 2001 From: dan-ince-aai Date: Fri, 17 Oct 2025 11:38:03 +0100 Subject: [PATCH 13/20] feat: multilingual + changelog updates --- CHANGELOG.md | 4 ++++ src/pipecat/services/assemblyai/models.py | 4 ++++ 2 files changed, 8 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 424d6a420..45131e331 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `keyterms_prompt` parameter to `AssemblyAIConnectionParams`. + +- Added `speech_model` parameter to `AssemblyAIConnectionParams` to access the multilingual model. + - The runner `--folder` argument now supports downloading files from subdirectories. diff --git a/src/pipecat/services/assemblyai/models.py b/src/pipecat/services/assemblyai/models.py index d263d113d..52ea87d87 100644 --- a/src/pipecat/services/assemblyai/models.py +++ b/src/pipecat/services/assemblyai/models.py @@ -109,6 +109,7 @@ class AssemblyAIConnectionParams(BaseModel): min_end_of_turn_silence_when_confident: Minimum silence duration when confident about end-of-turn. max_turn_silence: Maximum silence duration before forcing end-of-turn. keyterms_prompt: List of key terms to guide transcription. Will be JSON serialized before sending. + speech_model: Select between English and multilingual models. Defaults to "universal-streaming-english". """ sample_rate: int = 16000 @@ -119,3 +120,6 @@ class AssemblyAIConnectionParams(BaseModel): min_end_of_turn_silence_when_confident: Optional[int] = None max_turn_silence: Optional[int] = None keyterms_prompt: Optional[List[str]] = None + speech_model: Literal["universal-streaming-english", "universal-streaming-multilingual"] = ( + "universal-streaming-english" + ) From 3e8a7cc2544b98b995917e9bddb11b5fdcdf6c7b Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 17 Oct 2025 08:57:45 -0300 Subject: [PATCH 14/20] Adding support for trickle ICE to the SmallWebRTCTransport. --- CHANGELOG.md | 2 + src/pipecat/runner/run.py | 8 ++++ .../transports/smallwebrtc/connection.py | 5 +++ .../transports/smallwebrtc/request_handler.py | 42 +++++++++++++++++++ 4 files changed, 57 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 938b175c1..6d581ccce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added support for trickle ICE to the `SmallWebRTCTransport`. + - Added support for updating `OpenAITTSService` settings (`instructions` and `speed`) at runtime via `TTSUpdateSettingsFrame`. diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index a3e2984e8..9b1d233f6 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -204,6 +204,7 @@ def _setup_webrtc_routes( from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection from pipecat.transports.smallwebrtc.request_handler import ( + SmallWebRTCPatchRequest, SmallWebRTCRequest, SmallWebRTCRequestHandler, ) @@ -256,6 +257,13 @@ def _setup_webrtc_routes( ) return answer + @app.patch("/api/offer") + async def ice_candidate(request: SmallWebRTCPatchRequest): + """Handle WebRTC new ice candidate requests.""" + logger.debug(f"Received patch request: {request}") + await small_webrtc_handler.handle_patch_request(request) + return {"status": "success"} + @asynccontextmanager async def smallwebrtc_lifespan(app: FastAPI): """Manage FastAPI application lifecycle and cleanup connections.""" diff --git a/src/pipecat/transports/smallwebrtc/connection.py b/src/pipecat/transports/smallwebrtc/connection.py index c77f4e77e..60dd7798c 100644 --- a/src/pipecat/transports/smallwebrtc/connection.py +++ b/src/pipecat/transports/smallwebrtc/connection.py @@ -689,3 +689,8 @@ class SmallWebRTCConnection(BaseObject): )() if track: track.set_enabled(signalling_message.enabled) + + async def add_ice_candidate(self, candidate): + """Handle incoming ICE candidates.""" + logger.debug(f"Adding remote candidate: {candidate}") + await self.pc.addIceCandidate(candidate) diff --git a/src/pipecat/transports/smallwebrtc/request_handler.py b/src/pipecat/transports/smallwebrtc/request_handler.py index 00d9ebb6a..b2c02a03e 100644 --- a/src/pipecat/transports/smallwebrtc/request_handler.py +++ b/src/pipecat/transports/smallwebrtc/request_handler.py @@ -14,6 +14,7 @@ from dataclasses import dataclass from enum import Enum from typing import Any, Awaitable, Callable, Dict, List, Optional +from aiortc.sdp import candidate_from_sdp from fastapi import HTTPException from loguru import logger @@ -39,6 +40,34 @@ class SmallWebRTCRequest: request_data: Optional[Any] = None +@dataclass +class IceCandidate: + """The remote ice candidate object received from the peer connection. + + Parameters: + candidate: The ice candidate patch SDP string (Session Description Protocol). + sdp_mid: The SDP mid for the candidate patch. + sdp_mline_index: The SDP mline index for the candidate patch. + """ + + candidate: str + sdp_mid: str + sdp_mline_index: int + + +@dataclass +class SmallWebRTCPatchRequest: + """Small WebRTC transport session arguments for the runner. + + Parameters: + pc_id: Identifier for the peer connection. + candidates: A list of ICE candidate patches. + """ + + pc_id: str + candidates: List[IceCandidate] + + class ConnectionMode(Enum): """Enum defining the connection handling modes.""" @@ -197,6 +226,19 @@ class SmallWebRTCRequestHandler: logger.debug(f"SmallWebRTC request details: {request}") raise + async def handle_patch_request(self, request: SmallWebRTCPatchRequest): + """Handle a SmallWebRTC patch candidate request.""" + peer_connection = self._pcs_map.get(request.pc_id) + + if not peer_connection: + raise HTTPException(status_code=404, detail="Peer connection not found") + + for c in request.candidates: + candidate = candidate_from_sdp(c.candidate) + candidate.sdpMid = c.sdp_mid + candidate.sdpMLineIndex = c.sdp_mline_index + await peer_connection.add_ice_candidate(candidate) + async def close(self): """Clear the connection map.""" coros = [pc.disconnect() for pc in self._pcs_map.values()] From 855f4842dd9f156835db322a07cd3ec1d8f57031 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 17 Oct 2025 10:10:19 -0300 Subject: [PATCH 15/20] Creating the WebRTC routes that mimic the ones provided by Pipecat Cloud. --- src/pipecat/runner/run.py | 84 +++++++++++++++++++++++++++++++++++---- 1 file changed, 77 insertions(+), 7 deletions(-) diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index 9b1d233f6..a90c96ac5 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -70,12 +70,14 @@ import asyncio import mimetypes import os import sys +import uuid from contextlib import asynccontextmanager +from http import HTTPMethod from pathlib import Path -from typing import Optional +from typing import Any, Dict, List, Optional, TypedDict import aiohttp -from fastapi.responses import FileResponse +from fastapi.responses import FileResponse, Response from loguru import logger from pipecat.runner.types import ( @@ -202,8 +204,9 @@ def _setup_webrtc_routes( try: from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI - from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection + from pipecat.transports.smallwebrtc.connection import IceServer, SmallWebRTCConnection from pipecat.transports.smallwebrtc.request_handler import ( + IceCandidate, SmallWebRTCPatchRequest, SmallWebRTCRequest, SmallWebRTCRequestHandler, @@ -212,6 +215,16 @@ def _setup_webrtc_routes( logger.error(f"WebRTC transport dependencies not installed: {e}") return + class IceConfig(TypedDict): + iceServers: List[IceServer] + + class StartBotResult(TypedDict, total=False): + sessionId: str + iceConfig: Optional[IceConfig] + + # In-memory store of active sessions: session_id -> session info + active_sessions: Dict[str, Dict[str, Any]] = {} + # Mount the frontend app.mount("/client", SmallWebRTCPrebuiltUI) @@ -264,6 +277,67 @@ def _setup_webrtc_routes( await small_webrtc_handler.handle_patch_request(request) return {"status": "success"} + @app.post("/start") + async def rtvi_start(request: Request): + """Mimic Pipecat Cloud's /start endpoint.""" + # Parse the request body + try: + request_data = await request.json() + logger.debug(f"Received request: {request_data}") + except Exception as e: + logger.error(f"Failed to parse request body: {e}") + request_data = {} + + # Store session info immediately in memory, replicate the behavior expected on Pipecat Cloud + session_id = str(uuid.uuid4()) + active_sessions[session_id] = request_data + + result: StartBotResult = {"sessionId": session_id} + if request_data.get("enableDefaultIceServers"): + result["iceConfig"] = IceConfig( + iceServers=[IceServer(urls="stun:stun.l.google.com:19302")] + ) + + return result + + @app.api_route( + "/sessions/{session_id}/{path:path}", + methods=["GET", "POST", "PUT", "PATCH", "DELETE"], + ) + async def proxy_request( + session_id: str, path: str, request: Request, background_tasks: BackgroundTasks + ): + """Mimic Pipecat Cloud's proxy.""" + active_session = active_sessions.get(session_id) + if not active_session: + return Response(content="Invalid or not-yet-ready session_id", status_code=404) + + if path.endswith("api/offer"): + # Parse the request body and convert to SmallWebRTCRequest + try: + request_data = await request.json() + if request.method == HTTPMethod.POST.value: + webrtc_request = SmallWebRTCRequest( + sdp=request_data["sdp"], + type=request_data["type"], + pc_id=request_data.get("pc_id"), + restart_pc=request_data.get("restart_pc"), + request_data=request_data, + ) + return await offer(webrtc_request, background_tasks) + elif request.method == HTTPMethod.PATCH.value: + patch_request = SmallWebRTCPatchRequest( + pc_id=request_data["pc_id"], + candidates=[IceCandidate(**c) for c in request_data.get("candidates", [])], + ) + return await ice_candidate(patch_request) + except Exception as e: + logger.error(f"Failed to parse WebRTC request: {e}") + return Response(content="Invalid WebRTC request", status_code=400) + + logger.info(f"Received request for path: {path}") + return Response(status_code=200) + @asynccontextmanager async def smallwebrtc_lifespan(app: FastAPI): """Manage FastAPI application lifecycle and cleanup connections.""" @@ -503,8 +577,6 @@ def _setup_daily_routes(app: FastAPI): else: logger.debug("No body data provided in request") - import aiohttp - from pipecat.runner.daily import configure async with aiohttp.ClientSession() as session: @@ -592,8 +664,6 @@ def _setup_telephony_routes(app: FastAPI, *, transport_type: str, proxy: str): async def _run_daily_direct(): """Run Daily bot with direct connection (no FastAPI server).""" try: - import aiohttp - from pipecat.runner.daily import configure except ImportError as e: logger.error("Daily transport dependencies not installed.") From f2c61ac9fd7aa4e626c03a56f19b4cc7ab411a13 Mon Sep 17 00:00:00 2001 From: Sam Sykes Date: Fri, 17 Oct 2025 14:34:37 +0100 Subject: [PATCH 16/20] Fix for AdditionVocabEntry without sounds_like items. --- CHANGELOG.md | 6 ++++++ pyproject.toml | 2 +- src/pipecat/services/speechmatics/stt.py | 2 +- 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d581ccce..209c9cb11 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,9 +33,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `CartesiaSTTService` now inherits from `WebsocketSTTService`. - Package upgrades: + - `openai` upgraded to support up to 2.x.x. - `openpipe` upgraded to support up to 5.x.x. +- `SpeechmaticsSTTService` updated dependencies for `speechmatics-rt>=0.5.0`. + ### Fixed - Fixed multiple pipeline task cancellation issues. `asyncio.CancelledError` is @@ -59,6 +62,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 incorrectly 16-bit aligned audio frames, potentially leading to internal errors or static audio. +- Fixed an issue in `SpeechmaticsSTTService` where `AdditionalVocabEntry` items + needed to have `sounds_like` for the session to start. + ### Other - Added foundational example `47-sentry-metrics.py`, demonstrating how to use the diff --git a/pyproject.toml b/pyproject.toml index 0d94d0fa3..65546311c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -102,7 +102,7 @@ silero = [ "onnxruntime>=1.20.1,<2" ] simli = [ "simli-ai~=0.1.10"] soniox = [ "pipecat-ai[websockets-base]" ] soundfile = [ "soundfile~=0.13.0" ] -speechmatics = [ "speechmatics-rt>=0.4.0" ] +speechmatics = [ "speechmatics-rt>=0.5.0" ] strands = [ "strands-agents>=1.9.1,<2" ] tavus=[] together = [] diff --git a/src/pipecat/services/speechmatics/stt.py b/src/pipecat/services/speechmatics/stt.py index 2c1db2a15..901edb0e8 100644 --- a/src/pipecat/services/speechmatics/stt.py +++ b/src/pipecat/services/speechmatics/stt.py @@ -620,7 +620,7 @@ class SpeechmaticsSTTService(STTService): transcription_config.additional_vocab = [ { "content": e.content, - "sounds_like": e.sounds_like, + **({"sounds_like": e.sounds_like} if e.sounds_like else {}), } for e in self._params.additional_vocab ] From a67a765783dccbf4893d535722d8d15f6c7894e9 Mon Sep 17 00:00:00 2001 From: nbyers-altira Date: Fri, 17 Oct 2025 13:49:52 -0400 Subject: [PATCH 17/20] add changelog, run linter --- CHANGELOG.md | 3 +++ src/pipecat/services/riva/stt.py | 4 +++- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d581ccce..96406e52f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed an issue in `RivaSTTService` where a runtime error occurred due to a + mismatch in the _handle_transcription method's signature. + - Fixed multiple pipeline task cancellation issues. `asyncio.CancelledError` is now handled properly in `PipelineTask` making it possible to cancel an asyncio task that it's executing a `PipelineRunner` cleanly. Also, diff --git a/src/pipecat/services/riva/stt.py b/src/pipecat/services/riva/stt.py index ccd166536..eddd3da9e 100644 --- a/src/pipecat/services/riva/stt.py +++ b/src/pipecat/services/riva/stt.py @@ -583,7 +583,9 @@ class RivaSegmentedSTTService(SegmentedSTTService): self._config.language_code = self._language @traced_stt - async def _handle_transcription(self, transcript: str, is_final: bool, language: Optional[Language] = None): + async def _handle_transcription( + self, transcript: str, is_final: bool, language: Optional[Language] = None + ): """Handle a transcription result with tracing.""" pass From a4867f61aa51497c78e2043e55f037519051c9f7 Mon Sep 17 00:00:00 2001 From: nbyers-altira Date: Fri, 17 Oct 2025 13:51:49 -0400 Subject: [PATCH 18/20] be a tad more precise in changelog --- CHANGELOG.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 96406e52f..5d8d44887 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,8 +38,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- Fixed an issue in `RivaSTTService` where a runtime error occurred due to a - mismatch in the _handle_transcription method's signature. +- Fixed an issue in `RivaSegmentedSTTService` where a runtime error occurred due + to a mismatch in the _handle_transcription method's signature. - Fixed multiple pipeline task cancellation issues. `asyncio.CancelledError` is now handled properly in `PipelineTask` making it possible to cancel an asyncio From 7ce370ccc6541d38f86d39264f7c4496e7d442a0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 17 Oct 2025 15:13:20 -0700 Subject: [PATCH 19/20] BaseOutputTransport: simplify bot speaking logic --- src/pipecat/transports/base_output.py | 125 ++++++++++++++------------ 1 file changed, 67 insertions(+), 58 deletions(-) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 9d4630100..ba508000f 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -410,6 +410,13 @@ class BaseOutputTransport(FrameProcessor): # Indicates if the bot is currently speaking. self._bot_speaking = False + # Last time a BotSpeakingFrame was pushed. + self._bot_speaking_frame_time = 0 + # How often a BotSpeakingFrame should be pushed (value should be + # lower than the audio chunks). + self._bot_speaking_frame_period = 0.2 + # Last time the bot actually spoke. + self._bot_speech_last_time = 0 self._audio_task: Optional[asyncio.Task] = None self._video_task: Optional[asyncio.Task] = None @@ -601,39 +608,71 @@ class BaseOutputTransport(FrameProcessor): async def _bot_started_speaking(self): """Handle bot started speaking event.""" - if not self._bot_speaking: - logger.debug( - f"Bot{f' [{self._destination}]' if self._destination else ''} started speaking" - ) + if self._bot_speaking: + return - downstream_frame = BotStartedSpeakingFrame() - downstream_frame.transport_destination = self._destination - upstream_frame = BotStartedSpeakingFrame() - upstream_frame.transport_destination = self._destination - await self._transport.push_frame(downstream_frame) - await self._transport.push_frame(upstream_frame, FrameDirection.UPSTREAM) + logger.debug( + f"Bot{f' [{self._destination}]' if self._destination else ''} started speaking" + ) - self._bot_speaking = True + downstream_frame = BotStartedSpeakingFrame() + downstream_frame.transport_destination = self._destination + upstream_frame = BotStartedSpeakingFrame() + upstream_frame.transport_destination = self._destination + await self._transport.push_frame(downstream_frame) + await self._transport.push_frame(upstream_frame, FrameDirection.UPSTREAM) + + self._bot_speaking = True async def _bot_stopped_speaking(self): """Handle bot stopped speaking event.""" - if self._bot_speaking: - logger.debug( - f"Bot{f' [{self._destination}]' if self._destination else ''} stopped speaking" - ) + if not self._bot_speaking: + return - downstream_frame = BotStoppedSpeakingFrame() - downstream_frame.transport_destination = self._destination - upstream_frame = BotStoppedSpeakingFrame() - upstream_frame.transport_destination = self._destination - await self._transport.push_frame(downstream_frame) - await self._transport.push_frame(upstream_frame, FrameDirection.UPSTREAM) + logger.debug( + f"Bot{f' [{self._destination}]' if self._destination else ''} stopped speaking" + ) - self._bot_speaking = False + downstream_frame = BotStoppedSpeakingFrame() + downstream_frame.transport_destination = self._destination + upstream_frame = BotStoppedSpeakingFrame() + upstream_frame.transport_destination = self._destination + await self._transport.push_frame(downstream_frame) + await self._transport.push_frame(upstream_frame, FrameDirection.UPSTREAM) - # Clean audio buffer (there could be tiny left overs if not multiple - # to our output chunk size). - self._audio_buffer = bytearray() + self._bot_speaking = False + + # Clean audio buffer (there could be tiny left overs if not multiple + # to our output chunk size). + self._audio_buffer = bytearray() + + async def _bot_currently_speaking(self): + """Handle bot speaking event.""" + await self._bot_started_speaking() + + diff_time = time.time() - self._bot_speaking_frame_time + if diff_time >= self._bot_speaking_frame_period: + await self._transport.push_frame(BotSpeakingFrame()) + await self._transport.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM) + self._bot_speaking_frame_time = time.time() + + self._bot_speech_last_time = time.time() + + async def _maybe_bot_currently_speaking(self, frame: SpeechOutputAudioRawFrame): + if not is_silence(frame.audio): + await self._bot_currently_speaking() + else: + silence_duration = time.time() - self._bot_speech_last_time + if silence_duration > BOT_VAD_STOP_SECS: + await self._bot_stopped_speaking() + + async def _handle_bot_speech(self, frame: Frame): + # TTS case. + if isinstance(frame, TTSAudioRawFrame): + await self._bot_currently_speaking() + # Speech stream case. + elif isinstance(frame, SpeechOutputAudioRawFrame): + await self._maybe_bot_currently_speaking(frame) async def _handle_frame(self, frame: Frame): """Handle various frame types with appropriate processing. @@ -641,7 +680,9 @@ class BaseOutputTransport(FrameProcessor): Args: frame: The frame to handle. """ - if isinstance(frame, OutputImageRawFrame): + if isinstance(frame, OutputAudioRawFrame): + await self._handle_bot_speech(frame) + elif isinstance(frame, OutputImageRawFrame): await self._set_video_image(frame) elif isinstance(frame, SpriteFrame): await self._set_video_images(frame.images) @@ -705,39 +746,7 @@ class BaseOutputTransport(FrameProcessor): async def _audio_task_handler(self): """Main audio processing task handler.""" - # Push a BotSpeakingFrame every 200ms, we don't really need to push it - # at every audio chunk. If the audio chunk is bigger than 200ms, push at - # every audio chunk. - TOTAL_CHUNK_MS = self._params.audio_out_10ms_chunks * 10 - BOT_SPEAKING_CHUNK_PERIOD = max(int(200 / TOTAL_CHUNK_MS), 1) - bot_speaking_counter = 0 - speech_last_speaking_time = 0 - async for frame in self._next_frame(): - # Notify the bot started speaking upstream if necessary and that - # it's actually speaking. - is_speaking = False - if isinstance(frame, TTSAudioRawFrame): - is_speaking = True - elif isinstance(frame, SpeechOutputAudioRawFrame): - if not is_silence(frame.audio): - is_speaking = True - speech_last_speaking_time = time.time() - else: - silence_duration = time.time() - speech_last_speaking_time - if silence_duration > BOT_VAD_STOP_SECS: - await self._bot_stopped_speaking() - - if is_speaking: - await self._bot_started_speaking() - if bot_speaking_counter % BOT_SPEAKING_CHUNK_PERIOD == 0: - await self._transport.push_frame(BotSpeakingFrame()) - await self._transport.push_frame( - BotSpeakingFrame(), FrameDirection.UPSTREAM - ) - bot_speaking_counter = 0 - bot_speaking_counter += 1 - # No need to push EndFrame, it's pushed from process_frame(). if isinstance(frame, EndFrame): break From 040774495022895bc5bca8442a0dfa54c2e80a87 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 17 Oct 2025 21:55:20 -0700 Subject: [PATCH 20/20] BaseOutputTransport: simplify process_frame --- src/pipecat/transports/base_output.py | 23 ++++------------------- 1 file changed, 4 insertions(+), 19 deletions(-) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index ba508000f..3edf434a9 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -293,15 +293,15 @@ class BaseOutputTransport(FrameProcessor): """ await super().process_frame(frame, direction) - # - # System frames (like InterruptionFrame) are pushed immediately. Other - # frames require order so they are put in the sink queue. - # if isinstance(frame, StartFrame): # Push StartFrame before start(), because we want StartFrame to be # processed by every processor before any other frame is processed. await self.push_frame(frame, direction) await self.start(frame) + elif isinstance(frame, EndFrame): + await self.stop(frame) + # Keep pushing EndFrame down so all the pipeline stops nicely. + await self.push_frame(frame, direction) elif isinstance(frame, CancelFrame): await self.cancel(frame) await self.push_frame(frame, direction) @@ -314,21 +314,6 @@ class BaseOutputTransport(FrameProcessor): await self.write_dtmf(frame) elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction) - # Control frames. - elif isinstance(frame, EndFrame): - await self.stop(frame) - # Keep pushing EndFrame down so all the pipeline stops nicely. - await self.push_frame(frame, direction) - elif isinstance(frame, MixerControlFrame): - await self._handle_frame(frame) - # Other frames. - elif isinstance(frame, OutputAudioRawFrame): - await self._handle_frame(frame) - elif isinstance(frame, (OutputImageRawFrame, SpriteFrame)): - await self._handle_frame(frame) - # TODO(aleix): Images and audio should support presentation timestamps. - elif frame.pts: - await self._handle_frame(frame) elif direction == FrameDirection.UPSTREAM: await self.push_frame(frame, direction) else: