From 273692421fab018b3bff661ca281e931010cdada Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 20 Feb 2026 08:08:00 -0700 Subject: [PATCH 1/3] Add DeepgramSageMakerTTSService for Deepgram TTS on AWS SageMaker Adds a TTS service that connects to Deepgram models deployed on AWS SageMaker endpoints via HTTP/2 bidirectional streaming. Supports the Deepgram TTS protocol (Speak, Flush, Clear, Close) over the BiDi client, with interruption handling and per-turn TTFB metrics. Updates the example and env.example with separate STT/TTS endpoint names. --- env.example | 3 +- .../07c-interruptible-deepgram-sagemaker.py | 21 +- .../services/deepgram/tts_sagemaker.py | 315 ++++++++++++++++++ 3 files changed, 333 insertions(+), 6 deletions(-) create mode 100644 src/pipecat/services/deepgram/tts_sagemaker.py diff --git a/env.example b/env.example index 6e7db21e2..bc14ea0bf 100644 --- a/env.example +++ b/env.example @@ -47,7 +47,8 @@ DAILY_ROOM_URL=https://... # Deepgram DEEPGRAM_API_KEY=... -SAGEMAKER_ENDPOINT_NAME=... +SAGEMAKER_STT_ENDPOINT_NAME=... +SAGEMAKER_TTS_ENDPOINT_NAME=... # DeepSeek DEEPSEEK_API_KEY=... diff --git a/examples/foundational/07c-interruptible-deepgram-sagemaker.py b/examples/foundational/07c-interruptible-deepgram-sagemaker.py index 51a4b1bcb..f6d1c9354 100644 --- a/examples/foundational/07c-interruptible-deepgram-sagemaker.py +++ b/examples/foundational/07c-interruptible-deepgram-sagemaker.py @@ -23,8 +23,10 @@ from pipecat.processors.aggregators.llm_response_universal import ( from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.aws.llm import AWSBedrockLLMService +from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.deepgram.stt_sagemaker import DeepgramSageMakerSTTService from pipecat.services.deepgram.tts import DeepgramTTSService +from pipecat.services.deepgram.tts_sagemaker import DeepgramSageMakerTTSService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams @@ -57,12 +59,21 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): # This requires: # - AWS credentials configured (via environment variables or AWS CLI) # - A deployed SageMaker endpoint with Deepgram model - stt = DeepgramSageMakerSTTService( - endpoint_name=os.getenv("SAGEMAKER_ENDPOINT_NAME"), - region=os.getenv("AWS_REGION"), - ) + # stt = DeepgramSageMakerSTTService( + # endpoint_name=os.getenv("SAGEMAKER_STT_ENDPOINT_NAME"), + # region=os.getenv("AWS_REGION"), + # ) + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-2-andromeda-en") + # Initialize Deepgram SageMaker TTS Service + # This requires: + # - AWS credentials configured (via environment variables or AWS CLI) + # - A deployed SageMaker endpoint with Deepgram TTS model + tts = DeepgramSageMakerTTSService( + endpoint_name=os.getenv("SAGEMAKER_TTS_ENDPOINT_NAME"), + region=os.getenv("AWS_REGION"), + voice="aura-2-andromeda-en", + ) llm = AWSBedrockLLMService( aws_region=os.getenv("AWS_REGION"), diff --git a/src/pipecat/services/deepgram/tts_sagemaker.py b/src/pipecat/services/deepgram/tts_sagemaker.py new file mode 100644 index 000000000..7c04bc299 --- /dev/null +++ b/src/pipecat/services/deepgram/tts_sagemaker.py @@ -0,0 +1,315 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Deepgram text-to-speech service for AWS SageMaker. + +This module provides a Pipecat TTS service that connects to Deepgram models +deployed on AWS SageMaker endpoints. Uses HTTP/2 bidirectional streaming for +low-latency real-time speech synthesis with support for interruptions and +streaming audio output. +""" + +import asyncio +import json +from typing import AsyncGenerator, Optional + +from loguru import logger + +from pipecat.frames.frames import ( + BotStoppedSpeakingFrame, + CancelFrame, + EndFrame, + ErrorFrame, + Frame, + InterruptionFrame, + LLMFullResponseEndFrame, + StartFrame, + TTSAudioRawFrame, + TTSStartedFrame, +) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.aws.sagemaker.bidi_client import SageMakerBidiClient +from pipecat.services.tts_service import TTSService +from pipecat.utils.tracing.service_decorators import traced_tts + + +class DeepgramSageMakerTTSService(TTSService): + """Deepgram text-to-speech service for AWS SageMaker. + + Provides real-time speech synthesis using Deepgram models deployed on + AWS SageMaker endpoints. Uses HTTP/2 bidirectional streaming for low-latency + audio generation with support for interruptions via the Clear message. + + Requirements: + + - AWS credentials configured (via environment variables, AWS CLI, or instance metadata) + - A deployed SageMaker endpoint with Deepgram TTS model: https://developers.deepgram.com/docs/deploy-amazon-sagemaker + - ``pipecat-ai[sagemaker]`` installed + + Example:: + + tts = DeepgramSageMakerTTSService( + endpoint_name="my-deepgram-tts-endpoint", + region="us-east-2", + voice="aura-2-helena-en", + ) + """ + + def __init__( + self, + *, + endpoint_name: str, + region: str, + voice: str = "aura-2-helena-en", + sample_rate: Optional[int] = None, + encoding: str = "linear16", + **kwargs, + ): + """Initialize the Deepgram SageMaker TTS service. + + Args: + endpoint_name: Name of the SageMaker endpoint with Deepgram TTS model + deployed (e.g., "my-deepgram-tts-endpoint"). + region: AWS region where the endpoint is deployed (e.g., "us-east-2"). + voice: Voice model to use for synthesis. Defaults to "aura-2-helena-en". + sample_rate: Audio sample rate in Hz. If None, uses the value from StartFrame. + encoding: Audio encoding format. Defaults to "linear16". + **kwargs: Additional arguments passed to the parent TTSService. + """ + super().__init__( + sample_rate=sample_rate, + push_stop_frames=True, + pause_frame_processing=True, + append_trailing_space=True, + **kwargs, + ) + + self._endpoint_name = endpoint_name + self._region = region + self._encoding = encoding + self.set_voice(voice) + + self._client: Optional[SageMakerBidiClient] = None + self._response_task: Optional[asyncio.Task] = None + self._context_id: Optional[str] = None + self._ttfb_started: bool = False + + def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as Deepgram SageMaker TTS service supports metrics generation. + """ + return True + + async def start(self, frame: StartFrame): + """Start the Deepgram SageMaker TTS service. + + Args: + frame: The start frame containing initialization parameters. + """ + await super().start(frame) + await self._connect() + + async def stop(self, frame: EndFrame): + """Stop the Deepgram SageMaker TTS service. + + Args: + frame: The end frame. + """ + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame): + """Cancel the Deepgram SageMaker TTS service. + + Args: + frame: The cancel frame. + """ + await super().cancel(frame) + await self._disconnect() + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process frames with special handling for LLM response end. + + Args: + frame: The frame to process. + direction: The direction of frame processing. + """ + await super().process_frame(frame, direction) + + if isinstance(frame, (LLMFullResponseEndFrame, EndFrame)): + await self.flush_audio() + elif isinstance(frame, BotStoppedSpeakingFrame): + self._ttfb_started = False + + async def _connect(self): + """Connect to the SageMaker endpoint and start the BiDi session. + + Builds the Deepgram TTS query string, creates the BiDi client, + starts the streaming session, and launches a background task for processing + responses. + """ + logger.debug("Connecting to Deepgram TTS on SageMaker...") + + query_string = ( + f"model={self._voice_id}&encoding={self._encoding}&sample_rate={self.sample_rate}" + ) + + self._client = SageMakerBidiClient( + endpoint_name=self._endpoint_name, + region=self._region, + model_invocation_path="v1/speak", + model_query_string=query_string, + ) + + try: + await self._client.start_session() + + self._response_task = self.create_task(self._process_responses()) + + logger.debug("Connected to Deepgram TTS on SageMaker") + await self._call_event_handler("on_connected") + + except Exception as e: + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) + await self._call_event_handler("on_connection_error", str(e)) + + async def _disconnect(self): + """Disconnect from the SageMaker endpoint. + + Sends a Close message to Deepgram, cancels the response processing task, + and closes the BiDi session. Safe to call multiple times. + """ + if self._client and self._client.is_active: + logger.debug("Disconnecting from Deepgram TTS on SageMaker...") + + try: + await self._client.send_json({"type": "Close"}) + except Exception as e: + logger.warning(f"Failed to send Close message: {e}") + + if self._response_task and not self._response_task.done(): + await self.cancel_task(self._response_task) + + await self._client.close_session() + + logger.debug("Disconnected from Deepgram TTS on SageMaker") + await self._call_event_handler("on_disconnected") + + async def _process_responses(self): + """Process streaming responses from Deepgram TTS on SageMaker. + + Continuously receives responses from the BiDi stream. Attempts to decode + each payload as UTF-8 JSON for control messages (Flushed, Cleared, Metadata, + Warning). If decoding fails, treats the payload as raw audio bytes and pushes + a TTSAudioRawFrame downstream. + """ + try: + while self._client and self._client.is_active: + result = await self._client.receive_response() + + if result is None: + break + + if hasattr(result, "value") and hasattr(result.value, "bytes_"): + if result.value.bytes_: + payload = result.value.bytes_ + + # Try to decode as JSON control message first + try: + response_data = payload.decode("utf-8") + parsed = json.loads(response_data) + msg_type = parsed.get("type") + + if msg_type == "Metadata": + logger.trace(f"Received metadata: {parsed}") + elif msg_type == "Flushed": + logger.trace(f"Received Flushed: {parsed}") + elif msg_type == "Cleared": + logger.trace(f"Received Cleared: {parsed}") + elif msg_type == "Warning": + logger.warning( + f"{self} warning: " + f"{parsed.get('description', 'Unknown warning')}" + ) + else: + logger.debug(f"Received unknown message type: {parsed}") + + except (UnicodeDecodeError, json.JSONDecodeError): + # Not JSON — treat as raw audio bytes + await self.stop_ttfb_metrics() + frame = TTSAudioRawFrame( + payload, + self.sample_rate, + 1, + context_id=self._context_id, + ) + await self.push_frame(frame) + + except asyncio.CancelledError: + logger.debug("TTS response processor cancelled") + except Exception as e: + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) + finally: + logger.debug("TTS response processor stopped") + + async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): + """Handle interruption by sending Clear message to Deepgram. + + The Clear message will clear Deepgram's internal text buffer and stop + sending audio, allowing for a new response to be generated. + """ + await super()._handle_interruption(frame, direction) + self._ttfb_started = False + + if self._client and self._client.is_active: + try: + await self._client.send_json({"type": "Clear"}) + except Exception as e: + logger.error(f"{self} error sending Clear message: {e}") + + async def flush_audio(self): + """Flush any pending audio synthesis by sending Flush command. + + This should be called when the LLM finishes a complete response to force + generation of audio from Deepgram's internal text buffer. + """ + if self._client and self._client.is_active: + try: + await self._client.send_json({"type": "Flush"}) + except Exception as e: + logger.error(f"{self} error sending Flush message: {e}") + + @traced_tts + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: + """Generate speech from text using Deepgram TTS on SageMaker. + + Args: + text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. + + Yields: + Frame: TTSStartedFrame, then None (audio comes asynchronously via + the response processor). + """ + logger.debug(f"{self}: Generating TTS [{text}]") + + try: + if not self._ttfb_started: + await self.start_ttfb_metrics() + self._ttfb_started = True + await self.start_tts_usage_metrics(text) + + yield TTSStartedFrame(context_id=context_id) + self._context_id = context_id + + await self._client.send_json({"type": "Speak", "text": text}) + + yield None + + except Exception as e: + yield ErrorFrame(error=f"Unknown error occurred: {e}") From 62ada92188aff008e43971a5833db53afc6bb0a8 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 20 Feb 2026 08:09:57 -0700 Subject: [PATCH 2/3] Add changelog for PR #3785 --- changelog/3785.added.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/3785.added.md diff --git a/changelog/3785.added.md b/changelog/3785.added.md new file mode 100644 index 000000000..90a4172d4 --- /dev/null +++ b/changelog/3785.added.md @@ -0,0 +1 @@ +- Added `DeepgramSageMakerTTSService` for running Deepgram TTS models deployed on AWS SageMaker endpoints via HTTP/2 bidirectional streaming. Supports the Deepgram TTS protocol (Speak, Flush, Clear, Close), interruption handling, and per-turn TTFB metrics. From 82ce3ea8de2b0e0577976195c9690dbf7d41ae76 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 20 Feb 2026 08:10:41 -0700 Subject: [PATCH 3/3] Update 07c example to use DeepgramSageMakerTTSService --- .../07c-interruptible-deepgram-sagemaker.py | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/examples/foundational/07c-interruptible-deepgram-sagemaker.py b/examples/foundational/07c-interruptible-deepgram-sagemaker.py index f6d1c9354..aced7666f 100644 --- a/examples/foundational/07c-interruptible-deepgram-sagemaker.py +++ b/examples/foundational/07c-interruptible-deepgram-sagemaker.py @@ -23,9 +23,7 @@ from pipecat.processors.aggregators.llm_response_universal import ( from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.aws.llm import AWSBedrockLLMService -from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.deepgram.stt_sagemaker import DeepgramSageMakerSTTService -from pipecat.services.deepgram.tts import DeepgramTTSService from pipecat.services.deepgram.tts_sagemaker import DeepgramSageMakerTTSService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams @@ -59,11 +57,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): # This requires: # - AWS credentials configured (via environment variables or AWS CLI) # - A deployed SageMaker endpoint with Deepgram model - # stt = DeepgramSageMakerSTTService( - # endpoint_name=os.getenv("SAGEMAKER_STT_ENDPOINT_NAME"), - # region=os.getenv("AWS_REGION"), - # ) - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + stt = DeepgramSageMakerSTTService( + endpoint_name=os.getenv("SAGEMAKER_STT_ENDPOINT_NAME"), + region=os.getenv("AWS_REGION"), + ) # Initialize Deepgram SageMaker TTS Service # This requires: