From 8e0a338d969462ce610d302e6876067f7610c709 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 12 May 2026 11:15:42 -0300 Subject: [PATCH 01/16] Nvidia Sagemaker Magpie TTS service --- .../services/nvidia/sagemaker/__init__.py | 0 src/pipecat/services/nvidia/sagemaker/tts.py | 452 ++++++++++++++++++ src/pipecat/utils/text/tts_text_sanitizer.py | 132 +++++ 3 files changed, 584 insertions(+) create mode 100644 src/pipecat/services/nvidia/sagemaker/__init__.py create mode 100644 src/pipecat/services/nvidia/sagemaker/tts.py create mode 100644 src/pipecat/utils/text/tts_text_sanitizer.py diff --git a/src/pipecat/services/nvidia/sagemaker/__init__.py b/src/pipecat/services/nvidia/sagemaker/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/services/nvidia/sagemaker/tts.py b/src/pipecat/services/nvidia/sagemaker/tts.py new file mode 100644 index 000000000..e38494b77 --- /dev/null +++ b/src/pipecat/services/nvidia/sagemaker/tts.py @@ -0,0 +1,452 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""NVIDIA Magpie TTS service backed by an AWS SageMaker endpoint.""" + +import asyncio +import base64 +import json +import os +from collections.abc import AsyncGenerator +from dataclasses import dataclass +from typing import Optional + +import aioboto3 +from loguru import logger + +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + ErrorFrame, + Frame, + InterruptionFrame, + StartFrame, + TTSAudioRawFrame, +) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.aws.sagemaker.bidi_client import SageMakerBidiClient +from pipecat.services.settings import TTSSettings +from pipecat.services.tts_service import InterruptibleTTSService, TTSService +from pipecat.utils.text.tts_text_sanitizer import sanitize_text_for_tts +from pipecat.utils.tracing.service_decorators import traced_tts + + +@dataclass +class NvidiaSageMakerTTSSettings(TTSSettings): + """Settings for NvidiaSageMakerHTTPTTSService. + + Parameters: + voice: NIM voice name (e.g. ``Magpie-Multilingual.EN-US.Aria``). + language: BCP-47 language code passed to NIM (e.g. ``en-US``). + """ + + voice: str = "Magpie-Multilingual.EN-US.Aria" + language: str = "en-US" + + +class NvidiaSageMakerHTTPTTSService(TTSService): + """NVIDIA Magpie TTS service that calls a SageMaker HTTP endpoint. + + Sends each text segment to the wrapper's ``POST /invocations`` endpoint + as a JSON body and streams the raw PCM audio response back to bot + as :class:`TTSAudioRawFrame` frames. + + Example:: + + tts = NvidiaSageMakerHTTPTTSService( + endpoint_name=os.getenv("SAGEMAKER_MAGPIE_ENDPOINT_NAME"), + region=os.getenv("AWS_REGION", "us-west-2"), + settings=NvidiaSageMakerHTTPTTSService.Settings( + voice="Magpie-Multilingual.EN-US.Aria", + language="en-US", + ), + ) + """ + + Settings = NvidiaSageMakerTTSSettings + + def __init__( + self, + *, + endpoint_name: str, + region: str = "us-west-2", + sample_rate: int | None = None, + settings: NvidiaSageMakerTTSSettings | None = None, + **kwargs, + ): + """Initialize the SageMaker HTTP TTS service. + + Args: + endpoint_name: Name of the deployed SageMaker endpoint. + region: AWS region where the endpoint lives. + sample_rate: Output sample rate in Hz. Defaults to bot's pipeline rate. + params: Deprecated — use ``settings`` instead. + settings: Runtime-updatable settings (voice, language). + **kwargs: Forwarded to :class:`TTSService`. + """ + default_settings = self.Settings(model="magpie") + + if settings is not None: + default_settings.apply_update(settings) + + super().__init__( + sample_rate=sample_rate, + push_start_frame=True, + push_stop_frames=True, + settings=default_settings, + **kwargs, + ) + + self._endpoint_name = endpoint_name + self._region = region + self._client = None + self._client_ctx = None + + def can_generate_metrics(self) -> bool: + return True + + # ── Lifecycle ───────────────────────────────────────────────────────────── + + async def start(self, frame: StartFrame): + await super().start(frame) + session = aioboto3.Session( + aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"), + aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"), + region_name=self._region, + ) + self._client_ctx = session.client("sagemaker-runtime") + self._client = await self._client_ctx.__aenter__() + logger.debug(f"{self}: connected to SageMaker endpoint '{self._endpoint_name}'") + + async def _close_client(self): + if self._client_ctx is not None: + try: + await self._client_ctx.__aexit__(None, None, None) + except Exception as e: + logger.warning(f"{self}: error closing SageMaker client: {e}") + self._client_ctx = None + self._client = None + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._close_client() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._close_client() + + # ── Synthesis ───────────────────────────────────────────────────────────── + + @traced_tts + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: + """Synthesize text via SageMaker and yield a single PCM audio frame. + + Args: + text: The text to synthesize. + context_id: Pipecat audio context identifier. + + Yields: + :class:`TTSAudioRawFrame` chunks of signed 16-bit mono PCM. + """ + logger.debug(f"{self}: Generating TTS [{text}]") + + text = sanitize_text_for_tts(text) + if not text: + return + + try: + body = json.dumps( + { + "text": text, + "voice_name": self._settings.voice, + "language_code": self._settings.language, + "sample_rate_hz": self.sample_rate, + } + ) + + response = await self._client.invoke_endpoint( + EndpointName=self._endpoint_name, + ContentType="application/json", + Accept="application/octet-stream", + Body=body, + ) + + if "Body" not in response: + yield ErrorFrame(error="SageMaker TTS returned no audio stream") + return + + async for chunk in response["Body"].iter_chunks(chunk_size=self.chunk_size): + if chunk: + yield TTSAudioRawFrame( + audio=chunk, + sample_rate=self.sample_rate, + num_channels=1, + context_id=context_id, + ) + except Exception as e: + logger.error(f"{self}: SageMaker TTS error: {e}") + yield ErrorFrame(error=f"SageMaker TTS error: {e}") + finally: + await self.stop_ttfb_metrics() + + +@dataclass +class NvidiaSageMakerWSTTSSettings(TTSSettings): + """Settings for NvidiaSageMakerWebsocketTTSService. + + Parameters: + voice: NIM voice name (e.g. ``Magpie-Multilingual.EN-US.Aria``). + language: BCP-47 language code passed to NIM (e.g. ``en-US``). + """ + + voice: str = "Magpie-Multilingual.EN-US.Aria" + language: str = "en-US" + + +class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): + """NVIDIA Magpie TTS service using SageMaker bidirectional streaming. + + Maintains a persistent HTTP/2 bidi-stream session to the SageMaker endpoint + for the lifetime of the pipeline. Each text segment is sent as NIM realtime + events; audio chunks arrive asynchronously and are pushed as + :class:`TTSAudioRawFrame` frames. + + Example:: + + tts = NvidiaSageMakerWebsocketTTSService( + endpoint_name=os.getenv("SAGEMAKER_MAGPIE_ENDPOINT_NAME"), + region=os.getenv("AWS_REGION", "us-west-2"), + settings=NvidiaSageMakerWebsocketTTSService.Settings( + voice="Magpie-Multilingual.EN-US.Aria", + language="en-US", + ), + ) + """ + + Settings = NvidiaSageMakerWSTTSSettings + + def __init__( + self, + *, + endpoint_name: str, + region: str = "us-west-2", + sample_rate: int | None = None, + settings: NvidiaSageMakerWSTTSSettings | None = None, + **kwargs, + ): + default_settings = self.Settings(model="magpie") + + if settings is not None: + default_settings.apply_update(settings) + + super().__init__( + sample_rate=sample_rate, + push_start_frame=True, + push_stop_frames=True, + pause_frame_processing=True, + append_trailing_space=True, + settings=default_settings, + **kwargs, + ) + + self._endpoint_name = endpoint_name + self._region = region + self._client: SageMakerBidiClient | None = None + self._receive_task = None + self._speech_completed_event = asyncio.Event() + + def can_generate_metrics(self) -> bool: + return True + + # ── Lifecycle ───────────────────────────────────────────────────────────── + + async def start(self, frame: StartFrame): + await super().start(frame) + await self._connect() + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._disconnect() + + # ── Connection management (WebsocketService abstract interface) ──────────── + + async def _connect(self): + await super()._connect() + await self._connect_websocket() + if self._client and self._client.is_active and not self._receive_task: + self._receive_task = self.create_task(self._receive_task_handler(self._report_error)) + + async def _disconnect(self): + await super()._disconnect() + if self._receive_task: + await self.cancel_task(self._receive_task) + self._receive_task = None + await self._disconnect_websocket() + + async def _connect_websocket(self): + if self._client and self._client.is_active: + return + + logger.debug( + f"{self}: connecting to SageMaker bidi-stream endpoint '{self._endpoint_name}'" + ) + try: + self._client = SageMakerBidiClient( + endpoint_name=self._endpoint_name, + region=self._region, + model_query_string=None, + model_invocation_path=None, + ) + await self._client.start_session() + await self._send_session_config() + logger.debug(f"{self}: connected") + await self._call_event_handler("on_connected") + except Exception as e: + logger.error(f"{self}: connection error: {e}") + self._client = None + await self._call_event_handler("on_connection_error", f"{e}") + + async def _disconnect_websocket(self): + try: + if self._client and self._client.is_active: + logger.debug(f"{self}: disconnecting") + try: + await self._client.send_json({"type": "session.end"}) + except Exception as e: + logger.warning(f"{self}: error sending session.end: {e}") + await self._client.close_session() + logger.debug(f"{self}: disconnected") + except Exception as e: + logger.warning(f"{self}: error during disconnect: {e}") + finally: + self._client = None + await self._call_event_handler("on_disconnected") + + async def _verify_connection(self): + active = self._client and self._client.is_active + logger.info(f"{self}: verifying if websocket connection is active {active}") + return active + + async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): + if self._bot_speaking: + logger.debug( + f"{self}: interruption detected, sending input_text.done and waiting for speech.completed" + ) + self._disconnecting = True + self._speech_completed_event.clear() + try: + await self._client.send_json({"type": "input_text.done"}) + await asyncio.wait_for(self._speech_completed_event.wait(), timeout=5.0) + except TimeoutError: + logger.warning(f"{self}: timed out waiting for conversation.item.speech.completed") + await super()._handle_interruption(frame, direction) + + async def _receive_messages(self): + """Receive NIM JSON events and push audio frames.""" + while self._client and self._client.is_active and not self._disconnecting: + result = await self._client.receive_response() + + if self._disconnecting: + self._speech_completed_event.set() + + if result is None: + break + + if not (hasattr(result, "value") and hasattr(result.value, "bytes_")): + continue + + payload = result.value.bytes_ + if not payload: + continue + + context_id = self.get_active_audio_context_id() + + try: + msg = json.loads(payload.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + # Unexpected binary frame — treat as raw PCM + await self.push_frame( + TTSAudioRawFrame( + audio=payload, + sample_rate=self.sample_rate, + num_channels=1, + context_id=context_id, + ) + ) + continue + + event_type = msg.get("type", "") + + if event_type != "conversation.item.speech.data": + logger.debug(f"{self}: received event: {event_type}") + + if event_type == "conversation.item.speech.data": + chunk_b64 = msg.get("audio", "") + if chunk_b64: + await self.stop_ttfb_metrics() + await self.push_frame( + TTSAudioRawFrame( + audio=base64.b64decode(chunk_b64), + sample_rate=self.sample_rate, + num_channels=1, + context_id=context_id, + ) + ) + elif event_type == "error": + await self.push_error(error_msg=f"NIM error: {msg.get('message', msg)}") + # In case of error we need to reconnect, otherwise we are not going to receive audio from the TTS service anymore + break + elif event_type == "conversation.item.speech.completed": + # Need to reconnect to reset the synthesis state and be able to synthesize new text + break + + # synthesize_session.updated, input_text.committed, etc. are ignored. + + async def _send_session_config(self): + """Send synthesize_session.update to configure voice and audio params.""" + logger.debug(f"{self}: sending session config, sample_rate={self.sample_rate}") + await self._client.send_json( + { + "type": "synthesize_session.update", + "session": { + "input_text_synthesis": { + "voice_name": self._settings.voice, + "language_code": self._settings.language, + }, + "output_audio_params": { + "sample_rate_hz": self.sample_rate, + }, + }, + } + ) + + # ── Synthesis ───────────────────────────────────────────────────────────── + + @traced_tts + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: + """Send text to NIM; audio arrives asynchronously via _receive_messages.""" + logger.debug(f"{self}: Generating TTS [{text}]") + + text = sanitize_text_for_tts(text) + + logger.debug(f"{self}: sanitized text: {text}") + if not text: + return + + try: + if not self._client or not self._client.is_active: + await self._connect() + + await self._client.send_json({"type": "input_text.append", "text": text}) + await self._client.send_json({"type": "input_text.commit"}) + yield None + except Exception as e: + logger.error(f"{self}: TTS error: {e}") + yield ErrorFrame(error=f"NvidiaSageMakerWebsocketTTSService error: {e}") diff --git a/src/pipecat/utils/text/tts_text_sanitizer.py b/src/pipecat/utils/text/tts_text_sanitizer.py new file mode 100644 index 000000000..b5e78e5ce --- /dev/null +++ b/src/pipecat/utils/text/tts_text_sanitizer.py @@ -0,0 +1,132 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Utility for stripping non-speakable characters and markdown formatting from +text before it is sent to a TTS service. + +Both NvidiaSageMakerHTTPTTSService and NvidiaSageMakerWebsocketTTSService +use :func:`sanitize_text_for_tts` so the logic lives in one place. +""" + +import re + +# --------------------------------------------------------------------------- +# Emoji / symbol ranges +# --------------------------------------------------------------------------- + +_EMOJI_PATTERN = re.compile( + "[" + "\U0001f600-\U0001f64f" # Emoticons + "\U0001f300-\U0001f5ff" # Misc Symbols and Pictographs + "\U0001f680-\U0001f6ff" # Transport and Map + "\U0001f700-\U0001f77f" # Alchemical Symbols + "\U0001f780-\U0001f7ff" # Geometric Shapes Extended + "\U0001f800-\U0001f8ff" # Supplemental Arrows-C + "\U0001f900-\U0001f9ff" # Supplemental Symbols and Pictographs + "\U0001fa00-\U0001fa6f" # Chess Symbols + "\U0001fa70-\U0001faff" # Symbols and Pictographs Extended-A + "\U00002702-\U000027b0" # Dingbats + "\U0001f1e0-\U0001f1ff" # Flags (iOS) + "]+", + flags=re.UNICODE, +) + + +def sanitize_text_for_tts(text: str) -> str: + """Remove emojis and markdown formatting that should not be spoken aloud. + + Transformations applied (in order): + 1. Fenced code blocks (``` ... ```) → removed entirely + 2. Markdown headers (# / ## / …) → header text kept, # stripped + 3. Horizontal rules (--- / *** / ___) → removed + 4. Table separator rows (|---|---|) → removed + 5. Table data rows (| a | b |) → cells joined with commas + 6. Bold / italic markers (**x**, *x*, __x__, _x_) → text kept, markers stripped + 7. Blockquote markers (> …) → marker stripped, text kept + 8. Inline code backticks (`x`) → backticks stripped, text kept + 9. Emojis → removed + 10. Curly quotes → straight quotes + Em/en dashes → comma (natural pause, not a spoken symbol) + Separator hyphens ( - ) → comma + Unordered list bullets (^- , ^* ) → removed + Remaining bare * and _ → removed + Other non-speakable symbols → removed + 11. Collapse extra whitespace + + Args: + text: Raw text, potentially containing markdown and/or emoji. + + Returns: + Plain text suitable for speech synthesis. + """ + # 1. Fenced code blocks + text = re.sub(r"```[\s\S]*?```", "", text) + + # 2. Markdown headers (# Heading → Heading) + text = re.sub(r"^#{1,6}\s+", "", text, flags=re.MULTILINE) + + # 3. Horizontal rules (---, ***, ___ on their own line) + text = re.sub(r"^\s*[-*_]{3,}\s*$", "", text, flags=re.MULTILINE) + + # 4. Table separator rows |---|:---:|---| + text = re.sub(r"^\s*\|[\s\-|:]+\|\s*$", "", text, flags=re.MULTILINE) + + # 5. Table data rows | cell | cell | → cell, cell + def _table_row_to_csv(m: re.Match) -> str: + cells = [c.strip() for c in m.group(1).split("|")] + return ", ".join(c for c in cells if c) + + text = re.sub(r"^\s*\|(.+)\|\s*$", _table_row_to_csv, text, flags=re.MULTILINE) + + # 6. Bold / italic (**x**, *x*, __x__, _x_) + # Handle triple before double before single to avoid partial matches. + text = re.sub(r"\*{3}([^*]+)\*{3}", r"\1", text) + text = re.sub(r"\*{2}([^*]+)\*{2}", r"\1", text) + text = re.sub(r"\*([^*\s][^*]*[^*\s]|\S)\*", r"\1", text) + text = re.sub(r"_{3}([^_]+)_{3}", r"\1", text) + text = re.sub(r"_{2}([^_]+)_{2}", r"\1", text) + text = re.sub(r"_([^_\s][^_]*[^_\s]|\S)_", r"\1", text) + + # 7. Blockquote markers + text = re.sub(r"^\s*>\s*", "", text, flags=re.MULTILINE) + + # 8. Inline code backticks + text = re.sub(r"`([^`]*)`", r"\1", text) + + # 9. Emojis + text = _EMOJI_PATTERN.sub("", text) + + # 10. Typographic and non-speakable characters + + # Curly quotes → straight equivalents (speakable) + text = text.replace("\u2018", "'") # LEFT SINGLE QUOTATION MARK + text = text.replace("\u2019", "'") # RIGHT SINGLE QUOTATION MARK + text = text.replace("\u201c", '"') # LEFT DOUBLE QUOTATION MARK + text = text.replace("\u201d", '"') # RIGHT DOUBLE QUOTATION MARK + + # Em/en dashes → comma (they mark a pause, not a spoken symbol) + text = text.replace("\u2014", ", ") # EM DASH + text = text.replace("\u2013", ", ") # EN DASH + + # Hyphen used as a separator ( - ) → comma; keep word-hyphens (e.g. "well-known") + text = re.sub(r"(?<= )- | -(?= )", ", ", text) + + # Unordered list bullets at line start (not caught by step 3) + text = re.sub(r"^[\s]*[-*]\s+", "", text, flags=re.MULTILINE) + + # Remaining bare * and _ (e.g. orphaned markers) + text = text.replace("*", "") + text = text.replace("_", " ") + + # Other symbols that TTS engines typically misread or glitch on + text = re.sub(r"[\\|<>{}[\]~^=+#@]", "", text) + + # 11. Collapse whitespace + text = re.sub(r"\n{3,}", "\n\n", text) # no more than one blank line + text = re.sub(r"[ \t]+", " ", text) # collapse spaces/tabs + # text = text.strip() + + return text From 5dd7413c00a106d7747f74b650584b1903844f86 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 12 May 2026 11:16:00 -0300 Subject: [PATCH 02/16] Nvidia Sagemaker Nemotron ASR STT service --- src/pipecat/services/nvidia/sagemaker/stt.py | 305 +++++++++++++++++++ 1 file changed, 305 insertions(+) create mode 100644 src/pipecat/services/nvidia/sagemaker/stt.py diff --git a/src/pipecat/services/nvidia/sagemaker/stt.py b/src/pipecat/services/nvidia/sagemaker/stt.py new file mode 100644 index 000000000..6ab299bf9 --- /dev/null +++ b/src/pipecat/services/nvidia/sagemaker/stt.py @@ -0,0 +1,305 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""NVIDIA Nemotron ASR STT service backed by an AWS SageMaker bidirectional-stream endpoint. + +Uses SageMaker's HTTP/2 bidi-stream API to maintain a persistent connection to +the wrapper's /invocations-bidirectional-stream endpoint, which proxies to NIM's +realtime WebSocket. + +Audio is streamed as base64-encoded PCM16 chunks via input_audio_buffer.append +events. Transcription deltas arrive as InterimTranscriptionFrames and final +results as TranscriptionFrames. + +When the VAD detects the user has stopped speaking, input_audio_buffer.commit +is sent to trigger NIM to finalise the current utterance. +""" + +import asyncio +import base64 +import json +from collections.abc import AsyncGenerator +from dataclasses import dataclass +from typing import Optional + +from loguru import logger + +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + ErrorFrame, + Frame, + InterimTranscriptionFrame, + StartFrame, + TranscriptionFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, +) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.aws.sagemaker.bidi_client import SageMakerBidiClient +from pipecat.services.settings import STTSettings +from pipecat.services.stt_service import STTService +from pipecat.utils.time import time_now_iso8601 +from pipecat.utils.tracing.service_decorators import traced_stt + + +@dataclass +class NvidiaSageMakerWSSTTSettings(STTSettings): + """Settings for NvidiaSageMakerWebsocketSTTService. + + Parameters: + language: ISO-639-1 language code passed to NIM (e.g. ``en``). + """ + + language: str = "en-US" + + +class NvidiaSageMakerWebsocketSTTService(STTService): + """NVIDIA Nemotron ASR STT service using SageMaker bidirectional streaming. + + Maintains a persistent HTTP/2 bidi-stream session to the SageMaker endpoint + for the lifetime of the pipeline. Audio chunks are forwarded as base64-encoded + PCM16 via NIM realtime events; transcription results arrive asynchronously and + are pushed as :class:`InterimTranscriptionFrame` and :class:`TranscriptionFrame` + frames. + + Example:: + + stt = NvidiaSageMakerWebsocketSTTService( + endpoint_name=os.getenv("SAGEMAKER_ASR_ENDPOINT_NAME"), + region=os.getenv("AWS_REGION", "us-west-2"), + settings=NvidiaSageMakerWebsocketSTTService.Settings( + language="en-US", + ), + ) + """ + + Settings = NvidiaSageMakerWSSTTSettings + + def __init__( + self, + *, + endpoint_name: str, + region: str = "us-west-2", + sample_rate: int | None = None, + settings: NvidiaSageMakerWSSTTSettings | None = None, + ttfs_p99_latency: float | None = 1.5, + **kwargs, + ): + default_settings = self.Settings( + model="cache-aware-parakeet-rnnt-en-US-asr-streaming-sortformer" + ) + + if settings is not None: + default_settings.apply_update(settings) + + super().__init__( + sample_rate=sample_rate, + settings=default_settings, + ttfs_p99_latency=ttfs_p99_latency, + **kwargs, + ) + + self._endpoint_name = endpoint_name + self._region = region + self._client: SageMakerBidiClient | None = None + self._response_task: asyncio.Task | None = None + + def can_generate_metrics(self) -> bool: + return True + + # ── Lifecycle ───────────────────────────────────────────────────────────── + + async def start(self, frame: StartFrame): + await super().start(frame) + await self._connect() + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._disconnect() + + # ── Audio input ─────────────────────────────────────────────────────────── + + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: + """Send an audio chunk to NIM; transcription results arrive asynchronously. + + Each chunk is appended and immediately committed, matching the NVIDIA + reference client pattern for continuous streaming transcription. + """ + if self._client and self._client.is_active: + try: + await self._client.send_json( + { + "type": "input_audio_buffer.append", + "audio": base64.b64encode(audio).decode(), + } + ) + await self._client.send_json({"type": "input_audio_buffer.commit"}) + except Exception as e: + yield ErrorFrame(error=f"Unknown error occurred: {e}") + yield None + + # ── VAD integration ─────────────────────────────────────────────────────── + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, VADUserStartedSpeakingFrame): + logger.debug(f"{self}: VAD user started speaking") + await self.start_processing_metrics() + if isinstance(frame, VADUserStoppedSpeakingFrame): + logger.debug(f"{self}: VAD user stopped speaking") + + # ── Connection management ───────────────────────────────────────────────── + + async def _connect(self): + logger.debug( + f"{self}: connecting to SageMaker bidi-stream endpoint '{self._endpoint_name}'" + ) + try: + self._client = SageMakerBidiClient( + endpoint_name=self._endpoint_name, + region=self._region, + model_query_string=None, + model_invocation_path=None, + ) + await self._client.start_session() + await self._send_session_config() + self._response_task = self.create_task(self._process_responses()) + logger.debug(f"{self}: connected") + await self._call_event_handler("on_connected") + except Exception as e: + logger.error(f"{self}: connection error: {e}") + self._client = None + await self._call_event_handler("on_connection_error", f"{e}") + + async def _disconnect(self): + if self._response_task and not self._response_task.done(): + await self.cancel_task(self._response_task) + self._response_task = None + + if self._client and self._client.is_active: + logger.debug(f"{self}: disconnecting") + try: + await self._client.send_json({"type": "session.end"}) + except Exception as e: + logger.warning(f"{self}: error sending session.end: {e}") + await self._client.close_session() + logger.debug(f"{self}: disconnected") + + self._client = None + await self._call_event_handler("on_disconnected") + + async def _send_session_config(self): + """Send transcription_session.update to configure audio format and params. + + Specifies ``"model": "nemotron-asr-streaming"`` in ``input_audio_transcription`` so + NIM selects the correct Nemotron ASR Streaming model. + """ + logger.debug( + f"{self}: sending session config," + f" sample_rate={self.sample_rate} language={self._settings.language}" + ) + await self._client.send_json( + { + "type": "transcription_session.update", + "session": { + "input_audio_format": "pcm16", + "input_audio_params": { + "sample_rate_hz": self.sample_rate, + "num_channels": 1, + }, + "input_audio_transcription": { + "language": self._settings.language, + "model": self._settings.model, + }, + "recognition_config": { + "enable_automatic_punctuation": True, + }, + }, + } + ) + + # ── Response processing ─────────────────────────────────────────────────── + + async def _process_responses(self): + """Receive NIM JSON events and push transcription frames.""" + try: + while self._client and self._client.is_active: + result = await self._client.receive_response() + + if result is None or not ( + hasattr(result, "value") and hasattr(result.value, "bytes_") + ): + continue + + payload = result.value.bytes_ + if not payload: + continue + + try: + msg = json.loads(payload.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + continue + + event_type = msg.get("type", "") + + if event_type not in ( + "conversation.item.input_audio_transcription.delta", + "input_audio_buffer.committed", + ): + logger.debug(f"{self}: received event: {event_type}") + + if event_type == "conversation.item.input_audio_transcription.delta": + delta = msg.get("delta", "") + if delta: + logger.debug(f"{self}: received transcription delta: {delta}") + await self.push_frame( + InterimTranscriptionFrame( + delta, + self._user_id, + time_now_iso8601(), + ) + ) + + elif event_type == "conversation.item.input_audio_transcription.completed": + transcript = msg.get("transcript", "") + if transcript.strip(): + logger.debug(f"{self}: received final transcription: {transcript}") + await self.push_frame( + TranscriptionFrame( + transcript, + self._user_id, + time_now_iso8601(), + result=msg, + ) + ) + await self._handle_transcription(transcript, True) + await self.stop_processing_metrics() + + elif event_type in ( + "conversation.item.input_audio_transcription.failed", + "error", + ): + await self.push_error(error_msg=f"NIM ASR error: {msg}") + # In case of error we need to reconnect, otherwise we are not going to receive from the STT service anymore + await self._disconnect() + await self._connect() + + except asyncio.CancelledError: + logger.debug(f"{self}: response processor cancelled") + except Exception as e: + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) + finally: + logger.debug(f"{self}: response processor stopped") + + @traced_stt + async def _handle_transcription(self, transcript: str, is_final: bool, language=None): + pass From a34864d643a4dc135d3a1cae426483e3401a2e61 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 12 May 2026 11:39:52 -0300 Subject: [PATCH 03/16] Fixed ruff, pyright, and test_service_init failures --- src/pipecat/services/nvidia/sagemaker/stt.py | 55 ++++++++++--- src/pipecat/services/nvidia/sagemaker/tts.py | 83 ++++++++++++++++---- src/pipecat/utils/text/tts_text_sanitizer.py | 3 +- 3 files changed, 114 insertions(+), 27 deletions(-) diff --git a/src/pipecat/services/nvidia/sagemaker/stt.py b/src/pipecat/services/nvidia/sagemaker/stt.py index 6ab299bf9..b9a946a20 100644 --- a/src/pipecat/services/nvidia/sagemaker/stt.py +++ b/src/pipecat/services/nvidia/sagemaker/stt.py @@ -23,7 +23,6 @@ import base64 import json from collections.abc import AsyncGenerator from dataclasses import dataclass -from typing import Optional from loguru import logger @@ -51,11 +50,9 @@ class NvidiaSageMakerWSSTTSettings(STTSettings): """Settings for NvidiaSageMakerWebsocketSTTService. Parameters: - language: ISO-639-1 language code passed to NIM (e.g. ``en``). + language: ISO-639-1 language code passed to NIM (e.g. ``en-US``). """ - language: str = "en-US" - class NvidiaSageMakerWebsocketSTTService(STTService): """NVIDIA Nemotron ASR STT service using SageMaker bidirectional streaming. @@ -89,8 +86,19 @@ class NvidiaSageMakerWebsocketSTTService(STTService): ttfs_p99_latency: float | None = 1.5, **kwargs, ): + """Initialize the SageMaker WebSocket STT service. + + Args: + endpoint_name: Name of the deployed SageMaker endpoint. + region: AWS region where the endpoint lives. + sample_rate: Input sample rate in Hz. Defaults to pipeline rate. + settings: Runtime-updatable settings (language, model). + ttfs_p99_latency: Expected p99 time-to-first-segment latency in seconds. + **kwargs: Forwarded to :class:`STTService`. + """ default_settings = self.Settings( - model="cache-aware-parakeet-rnnt-en-US-asr-streaming-sortformer" + model="cache-aware-parakeet-rnnt-en-US-asr-streaming-sortformer", + language="en-US", ) if settings is not None: @@ -109,25 +117,45 @@ class NvidiaSageMakerWebsocketSTTService(STTService): self._response_task: asyncio.Task | None = None def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as this service supports metrics generation. + """ return True # ── Lifecycle ───────────────────────────────────────────────────────────── async def start(self, frame: StartFrame): + """Start the STT service and connect to the SageMaker endpoint. + + Args: + frame: The start frame containing initialization parameters. + """ await super().start(frame) await self._connect() async def stop(self, frame: EndFrame): + """Stop the STT service and disconnect from the SageMaker endpoint. + + Args: + frame: The end frame. + """ await super().stop(frame) await self._disconnect() async def cancel(self, frame: CancelFrame): + """Cancel the STT service and disconnect from the SageMaker endpoint. + + Args: + frame: The cancel frame. + """ await super().cancel(frame) await self._disconnect() # ── Audio input ─────────────────────────────────────────────────────────── - async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame | None, None]: """Send an audio chunk to NIM; transcription results arrive asynchronously. Each chunk is appended and immediately committed, matching the NVIDIA @@ -149,6 +177,12 @@ class NvidiaSageMakerWebsocketSTTService(STTService): # ── VAD integration ─────────────────────────────────────────────────────── async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process frames with VAD-specific handling for metrics lifecycle. + + Args: + frame: The frame to process. + direction: The direction of frame processing. + """ await super().process_frame(frame, direction) if isinstance(frame, VADUserStartedSpeakingFrame): @@ -167,8 +201,8 @@ class NvidiaSageMakerWebsocketSTTService(STTService): self._client = SageMakerBidiClient( endpoint_name=self._endpoint_name, region=self._region, - model_query_string=None, - model_invocation_path=None, + model_query_string="", + model_invocation_path="", ) await self._client.start_session() await self._send_session_config() @@ -207,6 +241,7 @@ class NvidiaSageMakerWebsocketSTTService(STTService): f"{self}: sending session config," f" sample_rate={self.sample_rate} language={self._settings.language}" ) + assert self._client is not None await self._client.send_json( { "type": "transcription_session.update", @@ -236,11 +271,11 @@ class NvidiaSageMakerWebsocketSTTService(STTService): result = await self._client.receive_response() if result is None or not ( - hasattr(result, "value") and hasattr(result.value, "bytes_") + hasattr(result, "value") and hasattr(result.value, "bytes_") # type: ignore[union-attr] ): continue - payload = result.value.bytes_ + payload = result.value.bytes_ # type: ignore[union-attr] if not payload: continue diff --git a/src/pipecat/services/nvidia/sagemaker/tts.py b/src/pipecat/services/nvidia/sagemaker/tts.py index e38494b77..84d5aad8f 100644 --- a/src/pipecat/services/nvidia/sagemaker/tts.py +++ b/src/pipecat/services/nvidia/sagemaker/tts.py @@ -12,7 +12,6 @@ import json import os from collections.abc import AsyncGenerator from dataclasses import dataclass -from typing import Optional import aioboto3 from loguru import logger @@ -43,9 +42,6 @@ class NvidiaSageMakerTTSSettings(TTSSettings): language: BCP-47 language code passed to NIM (e.g. ``en-US``). """ - voice: str = "Magpie-Multilingual.EN-US.Aria" - language: str = "en-US" - class NvidiaSageMakerHTTPTTSService(TTSService): """NVIDIA Magpie TTS service that calls a SageMaker HTTP endpoint. @@ -87,7 +83,11 @@ class NvidiaSageMakerHTTPTTSService(TTSService): settings: Runtime-updatable settings (voice, language). **kwargs: Forwarded to :class:`TTSService`. """ - default_settings = self.Settings(model="magpie") + default_settings = self.Settings( + model="magpie", + voice="Magpie-Multilingual.EN-US.Aria", + language="en-US", + ) if settings is not None: default_settings.apply_update(settings) @@ -106,11 +106,21 @@ class NvidiaSageMakerHTTPTTSService(TTSService): self._client_ctx = None def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as this service supports metrics generation. + """ return True # ── Lifecycle ───────────────────────────────────────────────────────────── async def start(self, frame: StartFrame): + """Start the TTS service and create the SageMaker client. + + Args: + frame: The start frame containing initialization parameters. + """ await super().start(frame) session = aioboto3.Session( aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"), @@ -131,10 +141,20 @@ class NvidiaSageMakerHTTPTTSService(TTSService): self._client = None async def stop(self, frame: EndFrame): + """Stop the TTS service and close the SageMaker client. + + Args: + frame: The end frame. + """ await super().stop(frame) await self._close_client() async def cancel(self, frame: CancelFrame): + """Cancel the TTS service and close the SageMaker client. + + Args: + frame: The cancel frame. + """ await super().cancel(frame) await self._close_client() @@ -158,6 +178,7 @@ class NvidiaSageMakerHTTPTTSService(TTSService): return try: + assert self._client is not None body = json.dumps( { "text": text, @@ -202,9 +223,6 @@ class NvidiaSageMakerWSTTSSettings(TTSSettings): language: BCP-47 language code passed to NIM (e.g. ``en-US``). """ - voice: str = "Magpie-Multilingual.EN-US.Aria" - language: str = "en-US" - class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): """NVIDIA Magpie TTS service using SageMaker bidirectional streaming. @@ -237,7 +255,20 @@ class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): settings: NvidiaSageMakerWSTTSSettings | None = None, **kwargs, ): - default_settings = self.Settings(model="magpie") + """Initialize the SageMaker WebSocket TTS service. + + Args: + endpoint_name: Name of the deployed SageMaker endpoint. + region: AWS region where the endpoint lives. + sample_rate: Output sample rate in Hz. Defaults to pipeline rate. + settings: Runtime-updatable settings (voice, language). + **kwargs: Forwarded to :class:`InterruptibleTTSService`. + """ + default_settings = self.Settings( + model="magpie", + voice="Magpie-Multilingual.EN-US.Aria", + language="en-US", + ) if settings is not None: default_settings.apply_update(settings) @@ -259,19 +290,39 @@ class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): self._speech_completed_event = asyncio.Event() def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as this service supports metrics generation. + """ return True # ── Lifecycle ───────────────────────────────────────────────────────────── async def start(self, frame: StartFrame): + """Start the TTS service and connect to the SageMaker endpoint. + + Args: + frame: The start frame containing initialization parameters. + """ await super().start(frame) await self._connect() async def stop(self, frame: EndFrame): + """Stop the TTS service and disconnect from the SageMaker endpoint. + + Args: + frame: The end frame. + """ await super().stop(frame) await self._disconnect() async def cancel(self, frame: CancelFrame): + """Cancel the TTS service and disconnect from the SageMaker endpoint. + + Args: + frame: The cancel frame. + """ await super().cancel(frame) await self._disconnect() @@ -301,8 +352,8 @@ class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): self._client = SageMakerBidiClient( endpoint_name=self._endpoint_name, region=self._region, - model_query_string=None, - model_invocation_path=None, + model_query_string="", + model_invocation_path="", ) await self._client.start_session() await self._send_session_config() @@ -335,7 +386,7 @@ class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): return active async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): - if self._bot_speaking: + if self._bot_speaking and self._client: logger.debug( f"{self}: interruption detected, sending input_text.done and waiting for speech.completed" ) @@ -359,10 +410,10 @@ class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): if result is None: break - if not (hasattr(result, "value") and hasattr(result.value, "bytes_")): + if not (hasattr(result, "value") and hasattr(result.value, "bytes_")): # type: ignore[union-attr] continue - payload = result.value.bytes_ + payload = result.value.bytes_ # type: ignore[union-attr] if not payload: continue @@ -412,6 +463,7 @@ class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): async def _send_session_config(self): """Send synthesize_session.update to configure voice and audio params.""" logger.debug(f"{self}: sending session config, sample_rate={self.sample_rate}") + assert self._client is not None await self._client.send_json( { "type": "synthesize_session.update", @@ -430,7 +482,7 @@ class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): # ── Synthesis ───────────────────────────────────────────────────────────── @traced_tts - async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame | None, None]: """Send text to NIM; audio arrives asynchronously via _receive_messages.""" logger.debug(f"{self}: Generating TTS [{text}]") @@ -444,6 +496,7 @@ class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): if not self._client or not self._client.is_active: await self._connect() + assert self._client is not None await self._client.send_json({"type": "input_text.append", "text": text}) await self._client.send_json({"type": "input_text.commit"}) yield None diff --git a/src/pipecat/utils/text/tts_text_sanitizer.py b/src/pipecat/utils/text/tts_text_sanitizer.py index b5e78e5ce..919c0946d 100644 --- a/src/pipecat/utils/text/tts_text_sanitizer.py +++ b/src/pipecat/utils/text/tts_text_sanitizer.py @@ -4,8 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Utility for stripping non-speakable characters and markdown formatting from -text before it is sent to a TTS service. +"""Utility for stripping non-speakable characters and markdown formatting from text. Both NvidiaSageMakerHTTPTTSService and NvidiaSageMakerWebsocketTTSService use :func:`sanitize_text_for_tts` so the logic lives in one place. From 6a27ed35b1ee47d33c70d95cc893c34029f3b8e8 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 12 May 2026 12:19:30 -0300 Subject: [PATCH 04/16] Fixing the Bidi client to accept None. --- src/pipecat/services/aws/sagemaker/bidi_client.py | 4 ++-- src/pipecat/services/nvidia/sagemaker/stt.py | 4 ++-- src/pipecat/services/nvidia/sagemaker/tts.py | 4 ++-- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/pipecat/services/aws/sagemaker/bidi_client.py b/src/pipecat/services/aws/sagemaker/bidi_client.py index 8d7bdeaa1..1165bb89b 100644 --- a/src/pipecat/services/aws/sagemaker/bidi_client.py +++ b/src/pipecat/services/aws/sagemaker/bidi_client.py @@ -63,8 +63,8 @@ class SageMakerBidiClient: self, endpoint_name: str, region: str, - model_invocation_path: str = "", - model_query_string: str = "", + model_invocation_path: str | None = "", + model_query_string: str | None = "", ): """Initialize the SageMaker BiDi client. diff --git a/src/pipecat/services/nvidia/sagemaker/stt.py b/src/pipecat/services/nvidia/sagemaker/stt.py index b9a946a20..33b6a574f 100644 --- a/src/pipecat/services/nvidia/sagemaker/stt.py +++ b/src/pipecat/services/nvidia/sagemaker/stt.py @@ -201,8 +201,8 @@ class NvidiaSageMakerWebsocketSTTService(STTService): self._client = SageMakerBidiClient( endpoint_name=self._endpoint_name, region=self._region, - model_query_string="", - model_invocation_path="", + model_query_string=None, + model_invocation_path=None, ) await self._client.start_session() await self._send_session_config() diff --git a/src/pipecat/services/nvidia/sagemaker/tts.py b/src/pipecat/services/nvidia/sagemaker/tts.py index 84d5aad8f..ba994b134 100644 --- a/src/pipecat/services/nvidia/sagemaker/tts.py +++ b/src/pipecat/services/nvidia/sagemaker/tts.py @@ -352,8 +352,8 @@ class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): self._client = SageMakerBidiClient( endpoint_name=self._endpoint_name, region=self._region, - model_query_string="", - model_invocation_path="", + model_query_string=None, + model_invocation_path=None, ) await self._client.start_session() await self._send_session_config() From 7f98dba925dc9c0bc00b3321aaa2e35b9c645fe9 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 12 May 2026 14:43:12 -0300 Subject: [PATCH 05/16] Changelog files for the new nvidia features. --- changelog/4464.added.2.md | 1 + changelog/4464.added.3.md | 1 + changelog/4464.added.md | 1 + 3 files changed, 3 insertions(+) create mode 100644 changelog/4464.added.2.md create mode 100644 changelog/4464.added.3.md create mode 100644 changelog/4464.added.md diff --git a/changelog/4464.added.2.md b/changelog/4464.added.2.md new file mode 100644 index 000000000..094948ff7 --- /dev/null +++ b/changelog/4464.added.2.md @@ -0,0 +1 @@ +- Added NVIDIA Magpie TTS services via AWS SageMaker: `NvidiaSageMakerHTTPTTSService` (single HTTP invocation, streams raw PCM back) and `NvidiaSageMakerWebsocketTTSService` (persistent HTTP/2 bidi-stream with full interruption support via `InterruptibleTTSService`). diff --git a/changelog/4464.added.3.md b/changelog/4464.added.3.md new file mode 100644 index 000000000..8fc7e573e --- /dev/null +++ b/changelog/4464.added.3.md @@ -0,0 +1 @@ +- Added `sanitize_text_for_tts()` utility (`pipecat.utils.text.tts_text_sanitizer`) that strips markdown formatting (fenced code blocks, headers, bold/italic, tables, blockquotes, inline code) and emojis from text before sending it to a TTS service. diff --git a/changelog/4464.added.md b/changelog/4464.added.md new file mode 100644 index 000000000..9935c47f0 --- /dev/null +++ b/changelog/4464.added.md @@ -0,0 +1 @@ +- Added `NvidiaSageMakerWebsocketSTTService` for streaming speech recognition using NVIDIA Nemotron ASR via an AWS SageMaker bidirectional-stream endpoint. Produces `InterimTranscriptionFrame` and `TranscriptionFrame` frames, is VAD-aware, and automatically reconnects on error. From c2bdc1aada0a28b494f9dd9fcf212dd84161cceb Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 12 May 2026 16:11:01 -0300 Subject: [PATCH 06/16] Fixing metrics and adding extra guard after sanitization. --- src/pipecat/services/nvidia/sagemaker/tts.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/src/pipecat/services/nvidia/sagemaker/tts.py b/src/pipecat/services/nvidia/sagemaker/tts.py index ba994b134..81ffd28bf 100644 --- a/src/pipecat/services/nvidia/sagemaker/tts.py +++ b/src/pipecat/services/nvidia/sagemaker/tts.py @@ -174,7 +174,7 @@ class NvidiaSageMakerHTTPTTSService(TTSService): logger.debug(f"{self}: Generating TTS [{text}]") text = sanitize_text_for_tts(text) - if not text: + if not text or not any(c.isalnum() for c in text): return try: @@ -199,8 +199,12 @@ class NvidiaSageMakerHTTPTTSService(TTSService): yield ErrorFrame(error="SageMaker TTS returned no audio stream") return + first_chunk = True async for chunk in response["Body"].iter_chunks(chunk_size=self.chunk_size): if chunk: + if first_chunk: + await self.stop_ttfb_metrics() + first_chunk = False yield TTSAudioRawFrame( audio=chunk, sample_rate=self.sample_rate, @@ -210,8 +214,8 @@ class NvidiaSageMakerHTTPTTSService(TTSService): except Exception as e: logger.error(f"{self}: SageMaker TTS error: {e}") yield ErrorFrame(error=f"SageMaker TTS error: {e}") - finally: - await self.stop_ttfb_metrics() + + await self.start_tts_usage_metrics(text) @dataclass @@ -489,7 +493,7 @@ class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): text = sanitize_text_for_tts(text) logger.debug(f"{self}: sanitized text: {text}") - if not text: + if not text or not any(c.isalnum() for c in text): return try: @@ -499,6 +503,7 @@ class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): assert self._client is not None await self._client.send_json({"type": "input_text.append", "text": text}) await self._client.send_json({"type": "input_text.commit"}) + await self.start_tts_usage_metrics(text) yield None except Exception as e: logger.error(f"{self}: TTS error: {e}") From 0146947b68e69bbbc570b2b041a9285b9c95eb29 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 12 May 2026 17:12:19 -0300 Subject: [PATCH 07/16] Addressing the comments left in the PR review. --- src/pipecat/services/nvidia/sagemaker/stt.py | 18 ++++++++------ src/pipecat/services/nvidia/sagemaker/tts.py | 25 ++++++-------------- 2 files changed, 18 insertions(+), 25 deletions(-) diff --git a/src/pipecat/services/nvidia/sagemaker/stt.py b/src/pipecat/services/nvidia/sagemaker/stt.py index 33b6a574f..5b95a6218 100644 --- a/src/pipecat/services/nvidia/sagemaker/stt.py +++ b/src/pipecat/services/nvidia/sagemaker/stt.py @@ -46,15 +46,15 @@ from pipecat.utils.tracing.service_decorators import traced_stt @dataclass -class NvidiaSageMakerWSSTTSettings(STTSettings): - """Settings for NvidiaSageMakerWebsocketSTTService. +class NvidiaSageMakerSTTSettings(STTSettings): + """Settings for NvidiaSageMakerSTTService. Parameters: language: ISO-639-1 language code passed to NIM (e.g. ``en-US``). """ -class NvidiaSageMakerWebsocketSTTService(STTService): +class NvidiaSageMakerSTTService(STTService): """NVIDIA Nemotron ASR STT service using SageMaker bidirectional streaming. Maintains a persistent HTTP/2 bidi-stream session to the SageMaker endpoint @@ -65,16 +65,16 @@ class NvidiaSageMakerWebsocketSTTService(STTService): Example:: - stt = NvidiaSageMakerWebsocketSTTService( + stt = NvidiaSageMakerSTTService( endpoint_name=os.getenv("SAGEMAKER_ASR_ENDPOINT_NAME"), region=os.getenv("AWS_REGION", "us-west-2"), - settings=NvidiaSageMakerWebsocketSTTService.Settings( + settings=NvidiaSageMakerSTTService.Settings( language="en-US", ), ) """ - Settings = NvidiaSageMakerWSSTTSettings + Settings = NvidiaSageMakerSTTSettings def __init__( self, @@ -82,7 +82,7 @@ class NvidiaSageMakerWebsocketSTTService(STTService): endpoint_name: str, region: str = "us-west-2", sample_rate: int | None = None, - settings: NvidiaSageMakerWSSTTSettings | None = None, + settings: NvidiaSageMakerSTTSettings | None = None, ttfs_p99_latency: float | None = 1.5, **kwargs, ): @@ -301,6 +301,8 @@ class NvidiaSageMakerWebsocketSTTService(STTService): delta, self._user_id, time_now_iso8601(), + language=self._settings.language, + result=msg, ) ) @@ -313,7 +315,9 @@ class NvidiaSageMakerWebsocketSTTService(STTService): transcript, self._user_id, time_now_iso8601(), + language=self._settings.language, result=msg, + finalized=True, ) ) await self._handle_transcription(transcript, True) diff --git a/src/pipecat/services/nvidia/sagemaker/tts.py b/src/pipecat/services/nvidia/sagemaker/tts.py index 81ffd28bf..8a0e3cd5d 100644 --- a/src/pipecat/services/nvidia/sagemaker/tts.py +++ b/src/pipecat/services/nvidia/sagemaker/tts.py @@ -35,7 +35,7 @@ from pipecat.utils.tracing.service_decorators import traced_tts @dataclass class NvidiaSageMakerTTSSettings(TTSSettings): - """Settings for NvidiaSageMakerHTTPTTSService. + """Settings for NVIDIA SageMaker TTS services. Parameters: voice: NIM voice name (e.g. ``Magpie-Multilingual.EN-US.Aria``). @@ -79,7 +79,6 @@ class NvidiaSageMakerHTTPTTSService(TTSService): endpoint_name: Name of the deployed SageMaker endpoint. region: AWS region where the endpoint lives. sample_rate: Output sample rate in Hz. Defaults to bot's pipeline rate. - params: Deprecated — use ``settings`` instead. settings: Runtime-updatable settings (voice, language). **kwargs: Forwarded to :class:`TTSService`. """ @@ -218,17 +217,7 @@ class NvidiaSageMakerHTTPTTSService(TTSService): await self.start_tts_usage_metrics(text) -@dataclass -class NvidiaSageMakerWSTTSSettings(TTSSettings): - """Settings for NvidiaSageMakerWebsocketTTSService. - - Parameters: - voice: NIM voice name (e.g. ``Magpie-Multilingual.EN-US.Aria``). - language: BCP-47 language code passed to NIM (e.g. ``en-US``). - """ - - -class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): +class NvidiaSageMakerTTSService(InterruptibleTTSService): """NVIDIA Magpie TTS service using SageMaker bidirectional streaming. Maintains a persistent HTTP/2 bidi-stream session to the SageMaker endpoint @@ -238,17 +227,17 @@ class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): Example:: - tts = NvidiaSageMakerWebsocketTTSService( + tts = NvidiaSageMakerTTSService( endpoint_name=os.getenv("SAGEMAKER_MAGPIE_ENDPOINT_NAME"), region=os.getenv("AWS_REGION", "us-west-2"), - settings=NvidiaSageMakerWebsocketTTSService.Settings( + settings=NvidiaSageMakerTTSService.Settings( voice="Magpie-Multilingual.EN-US.Aria", language="en-US", ), ) """ - Settings = NvidiaSageMakerWSTTSSettings + Settings = NvidiaSageMakerTTSSettings def __init__( self, @@ -256,7 +245,7 @@ class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): endpoint_name: str, region: str = "us-west-2", sample_rate: int | None = None, - settings: NvidiaSageMakerWSTTSSettings | None = None, + settings: NvidiaSageMakerTTSSettings | None = None, **kwargs, ): """Initialize the SageMaker WebSocket TTS service. @@ -507,4 +496,4 @@ class NvidiaSageMakerWebsocketTTSService(InterruptibleTTSService): yield None except Exception as e: logger.error(f"{self}: TTS error: {e}") - yield ErrorFrame(error=f"NvidiaSageMakerWebsocketTTSService error: {e}") + yield ErrorFrame(error=f"NvidiaSageMakerTTSService error: {e}") From bea9e4b3ba0a1615036bad5f750009b7a0de7377 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 12 May 2026 17:44:11 -0300 Subject: [PATCH 08/16] New example voice-nvidia-sagemaker.py --- env.example | 4 + examples/voice/voice-nvidia-sagemaker.py | 129 +++++++++++++++++++++++ 2 files changed, 133 insertions(+) create mode 100644 examples/voice/voice-nvidia-sagemaker.py diff --git a/env.example b/env.example index d449db6b2..6d69cc0e9 100644 --- a/env.example +++ b/env.example @@ -132,6 +132,10 @@ NOVITA_API_KEY=... # NVIDIA NVIDIA_API_KEY=... +# For a full example of how to deploy to SageMaker, see: +# https://github.com/pipecat-ai/pipecat-examples/tree/main/nvidia_sagemaker_example/deployment/aws-sagemaker-nvidia +SAGEMAKER_ASR_ENDPOINT_NAME=... +SAGEMAKER_MAGPIE_ENDPOINT_NAME=... # OpenAI OPENAI_API_KEY=... diff --git a/examples/voice/voice-nvidia-sagemaker.py b/examples/voice/voice-nvidia-sagemaker.py new file mode 100644 index 000000000..403dcfafb --- /dev/null +++ b/examples/voice/voice-nvidia-sagemaker.py @@ -0,0 +1,129 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# +# For a full example of how to deploy to SageMaker, see: +# https://github.com/pipecat-ai/pipecat-examples/tree/main/nvidia_sagemaker_example/deployment/aws-sagemaker-nvidia + +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +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, + LLMUserAggregatorParams, +) +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.nvidia.llm import NvidiaLLMService +from pipecat.services.nvidia.sagemaker.stt import NvidiaSageMakerSTTService +from pipecat.services.nvidia.sagemaker.tts import NvidiaSageMakerTTSService +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 use lambdas to defer transport parameter creation until the transport +# type is selected at runtime. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + stt = NvidiaSageMakerSTTService( + endpoint_name=os.getenv("SAGEMAKER_ASR_ENDPOINT_NAME"), + region=os.getenv("AWS_REGION", "us-west-2"), + ) + + llm = NvidiaLLMService( + api_key=os.environ["NVIDIA_API_KEY"], + settings=NvidiaLLMService.Settings( + model="meta/llama-3.3-70b-instruct", + system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.", + ), + ) + + tts = NvidiaSageMakerTTSService( + endpoint_name=os.getenv("SAGEMAKER_MAGPIE_ENDPOINT_NAME"), + region=os.getenv("AWS_REGION", "us-west-2"), + ) + + context = LLMContext() + user_aggregator, assistant_aggregator = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), + ) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + user_aggregator, # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + assistant_aggregator, # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + context.add_message( + {"role": "developer", "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(f"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() From 7984556692050113e8a408ced6253a9326c34f72 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 12 May 2026 18:00:07 -0300 Subject: [PATCH 09/16] Fixing typecheck. --- src/pipecat/services/nvidia/sagemaker/stt.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/src/pipecat/services/nvidia/sagemaker/stt.py b/src/pipecat/services/nvidia/sagemaker/stt.py index 5b95a6218..34b6d7519 100644 --- a/src/pipecat/services/nvidia/sagemaker/stt.py +++ b/src/pipecat/services/nvidia/sagemaker/stt.py @@ -39,8 +39,9 @@ from pipecat.frames.frames import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.aws.sagemaker.bidi_client import SageMakerBidiClient -from pipecat.services.settings import STTSettings +from pipecat.services.settings import STTSettings, assert_given from pipecat.services.stt_service import STTService +from pipecat.transcriptions.language import Language from pipecat.utils.time import time_now_iso8601 from pipecat.utils.tracing.service_decorators import traced_stt @@ -292,6 +293,9 @@ class NvidiaSageMakerSTTService(STTService): ): logger.debug(f"{self}: received event: {event_type}") + _lang = assert_given(self._settings.language) + language: Language | None = Language(_lang) if _lang is not None else None + if event_type == "conversation.item.input_audio_transcription.delta": delta = msg.get("delta", "") if delta: @@ -301,7 +305,7 @@ class NvidiaSageMakerSTTService(STTService): delta, self._user_id, time_now_iso8601(), - language=self._settings.language, + language=language, result=msg, ) ) @@ -315,7 +319,7 @@ class NvidiaSageMakerSTTService(STTService): transcript, self._user_id, time_now_iso8601(), - language=self._settings.language, + language=language, result=msg, finalized=True, ) From 5d6176398752babe6ae2b5b6178efb8c31d97502 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 12 May 2026 18:20:19 -0300 Subject: [PATCH 10/16] Refactoring how we are reconnecting the STT. --- src/pipecat/services/nvidia/sagemaker/stt.py | 47 +++++++++++--------- 1 file changed, 26 insertions(+), 21 deletions(-) diff --git a/src/pipecat/services/nvidia/sagemaker/stt.py b/src/pipecat/services/nvidia/sagemaker/stt.py index 34b6d7519..cf6c8c6d8 100644 --- a/src/pipecat/services/nvidia/sagemaker/stt.py +++ b/src/pipecat/services/nvidia/sagemaker/stt.py @@ -194,19 +194,31 @@ class NvidiaSageMakerSTTService(STTService): # ── Connection management ───────────────────────────────────────────────── + async def _open_client_session(self): + self._client = SageMakerBidiClient( + endpoint_name=self._endpoint_name, + region=self._region, + model_query_string=None, + model_invocation_path=None, + ) + await self._client.start_session() + await self._send_session_config() + + async def _close_client_session(self): + if self._client and self._client.is_active: + try: + await self._client.send_json({"type": "session.end"}) + except Exception as e: + logger.warning(f"{self}: error sending session.end: {e}") + await self._client.close_session() + self._client = None + async def _connect(self): logger.debug( f"{self}: connecting to SageMaker bidi-stream endpoint '{self._endpoint_name}'" ) try: - self._client = SageMakerBidiClient( - endpoint_name=self._endpoint_name, - region=self._region, - model_query_string=None, - model_invocation_path=None, - ) - await self._client.start_session() - await self._send_session_config() + await self._open_client_session() self._response_task = self.create_task(self._process_responses()) logger.debug(f"{self}: connected") await self._call_event_handler("on_connected") @@ -219,19 +231,13 @@ class NvidiaSageMakerSTTService(STTService): if self._response_task and not self._response_task.done(): await self.cancel_task(self._response_task) self._response_task = None - - if self._client and self._client.is_active: - logger.debug(f"{self}: disconnecting") - try: - await self._client.send_json({"type": "session.end"}) - except Exception as e: - logger.warning(f"{self}: error sending session.end: {e}") - await self._client.close_session() - logger.debug(f"{self}: disconnected") - - self._client = None + await self._close_client_session() await self._call_event_handler("on_disconnected") + async def _do_reconnect(self): + await self._close_client_session() + await self._open_client_session() + async def _send_session_config(self): """Send transcription_session.update to configure audio format and params. @@ -333,8 +339,7 @@ class NvidiaSageMakerSTTService(STTService): ): await self.push_error(error_msg=f"NIM ASR error: {msg}") # In case of error we need to reconnect, otherwise we are not going to receive from the STT service anymore - await self._disconnect() - await self._connect() + await self._request_reconnect() except asyncio.CancelledError: logger.debug(f"{self}: response processor cancelled") From 130bb7371cc642b1e24281c9a67485d751c09d03 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 12 May 2026 18:21:47 -0300 Subject: [PATCH 11/16] Removing sanitize_text_for_tts --- src/pipecat/services/nvidia/sagemaker/tts.py | 4 +- src/pipecat/utils/text/tts_text_sanitizer.py | 131 ------------------- 2 files changed, 1 insertion(+), 134 deletions(-) delete mode 100644 src/pipecat/utils/text/tts_text_sanitizer.py diff --git a/src/pipecat/services/nvidia/sagemaker/tts.py b/src/pipecat/services/nvidia/sagemaker/tts.py index 8a0e3cd5d..90d85bc97 100644 --- a/src/pipecat/services/nvidia/sagemaker/tts.py +++ b/src/pipecat/services/nvidia/sagemaker/tts.py @@ -479,9 +479,7 @@ class NvidiaSageMakerTTSService(InterruptibleTTSService): """Send text to NIM; audio arrives asynchronously via _receive_messages.""" logger.debug(f"{self}: Generating TTS [{text}]") - text = sanitize_text_for_tts(text) - - logger.debug(f"{self}: sanitized text: {text}") + text = text.strip() if not text or not any(c.isalnum() for c in text): return diff --git a/src/pipecat/utils/text/tts_text_sanitizer.py b/src/pipecat/utils/text/tts_text_sanitizer.py deleted file mode 100644 index 919c0946d..000000000 --- a/src/pipecat/utils/text/tts_text_sanitizer.py +++ /dev/null @@ -1,131 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""Utility for stripping non-speakable characters and markdown formatting from text. - -Both NvidiaSageMakerHTTPTTSService and NvidiaSageMakerWebsocketTTSService -use :func:`sanitize_text_for_tts` so the logic lives in one place. -""" - -import re - -# --------------------------------------------------------------------------- -# Emoji / symbol ranges -# --------------------------------------------------------------------------- - -_EMOJI_PATTERN = re.compile( - "[" - "\U0001f600-\U0001f64f" # Emoticons - "\U0001f300-\U0001f5ff" # Misc Symbols and Pictographs - "\U0001f680-\U0001f6ff" # Transport and Map - "\U0001f700-\U0001f77f" # Alchemical Symbols - "\U0001f780-\U0001f7ff" # Geometric Shapes Extended - "\U0001f800-\U0001f8ff" # Supplemental Arrows-C - "\U0001f900-\U0001f9ff" # Supplemental Symbols and Pictographs - "\U0001fa00-\U0001fa6f" # Chess Symbols - "\U0001fa70-\U0001faff" # Symbols and Pictographs Extended-A - "\U00002702-\U000027b0" # Dingbats - "\U0001f1e0-\U0001f1ff" # Flags (iOS) - "]+", - flags=re.UNICODE, -) - - -def sanitize_text_for_tts(text: str) -> str: - """Remove emojis and markdown formatting that should not be spoken aloud. - - Transformations applied (in order): - 1. Fenced code blocks (``` ... ```) → removed entirely - 2. Markdown headers (# / ## / …) → header text kept, # stripped - 3. Horizontal rules (--- / *** / ___) → removed - 4. Table separator rows (|---|---|) → removed - 5. Table data rows (| a | b |) → cells joined with commas - 6. Bold / italic markers (**x**, *x*, __x__, _x_) → text kept, markers stripped - 7. Blockquote markers (> …) → marker stripped, text kept - 8. Inline code backticks (`x`) → backticks stripped, text kept - 9. Emojis → removed - 10. Curly quotes → straight quotes - Em/en dashes → comma (natural pause, not a spoken symbol) - Separator hyphens ( - ) → comma - Unordered list bullets (^- , ^* ) → removed - Remaining bare * and _ → removed - Other non-speakable symbols → removed - 11. Collapse extra whitespace - - Args: - text: Raw text, potentially containing markdown and/or emoji. - - Returns: - Plain text suitable for speech synthesis. - """ - # 1. Fenced code blocks - text = re.sub(r"```[\s\S]*?```", "", text) - - # 2. Markdown headers (# Heading → Heading) - text = re.sub(r"^#{1,6}\s+", "", text, flags=re.MULTILINE) - - # 3. Horizontal rules (---, ***, ___ on their own line) - text = re.sub(r"^\s*[-*_]{3,}\s*$", "", text, flags=re.MULTILINE) - - # 4. Table separator rows |---|:---:|---| - text = re.sub(r"^\s*\|[\s\-|:]+\|\s*$", "", text, flags=re.MULTILINE) - - # 5. Table data rows | cell | cell | → cell, cell - def _table_row_to_csv(m: re.Match) -> str: - cells = [c.strip() for c in m.group(1).split("|")] - return ", ".join(c for c in cells if c) - - text = re.sub(r"^\s*\|(.+)\|\s*$", _table_row_to_csv, text, flags=re.MULTILINE) - - # 6. Bold / italic (**x**, *x*, __x__, _x_) - # Handle triple before double before single to avoid partial matches. - text = re.sub(r"\*{3}([^*]+)\*{3}", r"\1", text) - text = re.sub(r"\*{2}([^*]+)\*{2}", r"\1", text) - text = re.sub(r"\*([^*\s][^*]*[^*\s]|\S)\*", r"\1", text) - text = re.sub(r"_{3}([^_]+)_{3}", r"\1", text) - text = re.sub(r"_{2}([^_]+)_{2}", r"\1", text) - text = re.sub(r"_([^_\s][^_]*[^_\s]|\S)_", r"\1", text) - - # 7. Blockquote markers - text = re.sub(r"^\s*>\s*", "", text, flags=re.MULTILINE) - - # 8. Inline code backticks - text = re.sub(r"`([^`]*)`", r"\1", text) - - # 9. Emojis - text = _EMOJI_PATTERN.sub("", text) - - # 10. Typographic and non-speakable characters - - # Curly quotes → straight equivalents (speakable) - text = text.replace("\u2018", "'") # LEFT SINGLE QUOTATION MARK - text = text.replace("\u2019", "'") # RIGHT SINGLE QUOTATION MARK - text = text.replace("\u201c", '"') # LEFT DOUBLE QUOTATION MARK - text = text.replace("\u201d", '"') # RIGHT DOUBLE QUOTATION MARK - - # Em/en dashes → comma (they mark a pause, not a spoken symbol) - text = text.replace("\u2014", ", ") # EM DASH - text = text.replace("\u2013", ", ") # EN DASH - - # Hyphen used as a separator ( - ) → comma; keep word-hyphens (e.g. "well-known") - text = re.sub(r"(?<= )- | -(?= )", ", ", text) - - # Unordered list bullets at line start (not caught by step 3) - text = re.sub(r"^[\s]*[-*]\s+", "", text, flags=re.MULTILINE) - - # Remaining bare * and _ (e.g. orphaned markers) - text = text.replace("*", "") - text = text.replace("_", " ") - - # Other symbols that TTS engines typically misread or glitch on - text = re.sub(r"[\\|<>{}[\]~^=+#@]", "", text) - - # 11. Collapse whitespace - text = re.sub(r"\n{3,}", "\n\n", text) # no more than one blank line - text = re.sub(r"[ \t]+", " ", text) # collapse spaces/tabs - # text = text.strip() - - return text From b9f052079d1666e62d5c58e52b0cbdc5965d157e Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 12 May 2026 18:22:15 -0300 Subject: [PATCH 12/16] Removing sanitize_text_for_tts --- src/pipecat/services/nvidia/sagemaker/tts.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/pipecat/services/nvidia/sagemaker/tts.py b/src/pipecat/services/nvidia/sagemaker/tts.py index 90d85bc97..46ae10d4a 100644 --- a/src/pipecat/services/nvidia/sagemaker/tts.py +++ b/src/pipecat/services/nvidia/sagemaker/tts.py @@ -29,7 +29,6 @@ from pipecat.processors.frame_processor import FrameDirection from pipecat.services.aws.sagemaker.bidi_client import SageMakerBidiClient from pipecat.services.settings import TTSSettings from pipecat.services.tts_service import InterruptibleTTSService, TTSService -from pipecat.utils.text.tts_text_sanitizer import sanitize_text_for_tts from pipecat.utils.tracing.service_decorators import traced_tts From 68f265fa627d080550346ad916119dc271b08335 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 12 May 2026 18:28:14 -0300 Subject: [PATCH 13/16] Fixing ruff format. --- src/pipecat/services/nvidia/sagemaker/tts.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/services/nvidia/sagemaker/tts.py b/src/pipecat/services/nvidia/sagemaker/tts.py index 46ae10d4a..e6737a5f9 100644 --- a/src/pipecat/services/nvidia/sagemaker/tts.py +++ b/src/pipecat/services/nvidia/sagemaker/tts.py @@ -171,7 +171,7 @@ class NvidiaSageMakerHTTPTTSService(TTSService): """ logger.debug(f"{self}: Generating TTS [{text}]") - text = sanitize_text_for_tts(text) + text = text.strip() if not text or not any(c.isalnum() for c in text): return From 0740021ff47f64eb95d813664e5889ffafc687d0 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 12 May 2026 18:29:35 -0300 Subject: [PATCH 14/16] Removing changelog for sanitize_text_for_tts --- changelog/4464.added.3.md | 1 - 1 file changed, 1 deletion(-) delete mode 100644 changelog/4464.added.3.md diff --git a/changelog/4464.added.3.md b/changelog/4464.added.3.md deleted file mode 100644 index 8fc7e573e..000000000 --- a/changelog/4464.added.3.md +++ /dev/null @@ -1 +0,0 @@ -- Added `sanitize_text_for_tts()` utility (`pipecat.utils.text.tts_text_sanitizer`) that strips markdown formatting (fenced code blocks, headers, bold/italic, tables, blockquotes, inline code) and emojis from text before sending it to a TTS service. From 227ba288da75ee2f7c1d3664839777bc081a0665 Mon Sep 17 00:00:00 2001 From: Filipi da Silva Fuchter Date: Wed, 13 May 2026 06:36:45 -0400 Subject: [PATCH 15/16] Update examples/voice/voice-nvidia-sagemaker.py Co-authored-by: Mark Backman --- examples/voice/voice-nvidia-sagemaker.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/voice/voice-nvidia-sagemaker.py b/examples/voice/voice-nvidia-sagemaker.py index 403dcfafb..0256a6a55 100644 --- a/examples/voice/voice-nvidia-sagemaker.py +++ b/examples/voice/voice-nvidia-sagemaker.py @@ -54,7 +54,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") stt = NvidiaSageMakerSTTService( - endpoint_name=os.getenv("SAGEMAKER_ASR_ENDPOINT_NAME"), + endpoint_name=os.environ["SAGEMAKER_ASR_ENDPOINT_NAME"], region=os.getenv("AWS_REGION", "us-west-2"), ) From 703d23b658bf8ffb7cfa852614c8959994ef0c50 Mon Sep 17 00:00:00 2001 From: Filipi da Silva Fuchter Date: Wed, 13 May 2026 06:36:57 -0400 Subject: [PATCH 16/16] Update examples/voice/voice-nvidia-sagemaker.py Co-authored-by: Mark Backman --- examples/voice/voice-nvidia-sagemaker.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/voice/voice-nvidia-sagemaker.py b/examples/voice/voice-nvidia-sagemaker.py index 0256a6a55..ac0c6a365 100644 --- a/examples/voice/voice-nvidia-sagemaker.py +++ b/examples/voice/voice-nvidia-sagemaker.py @@ -67,7 +67,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ) tts = NvidiaSageMakerTTSService( - endpoint_name=os.getenv("SAGEMAKER_MAGPIE_ENDPOINT_NAME"), + endpoint_name=os.environ["SAGEMAKER_MAGPIE_ENDPOINT_NAME"], region=os.getenv("AWS_REGION", "us-west-2"), )