From 3f8373f76fce3e67e7c1124352a604b0e82c1505 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 7 Jan 2026 15:39:37 -0500 Subject: [PATCH 1/4] AzureTTSService: prevent word timestamp carryover on interruption --- pyproject.toml | 2 +- src/pipecat/services/azure/tts.py | 18 ++++++++++++++++- uv.lock | 32 +++++++++++++++++++++++-------- 3 files changed, 42 insertions(+), 10 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index d662d1d84..b988eb29d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,7 +51,7 @@ assemblyai = [ "pipecat-ai[websockets-base]" ] asyncai = [ "pipecat-ai[websockets-base]" ] aws = [ "aioboto3~=15.5.0", "pipecat-ai[websockets-base]" ] aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.2.0; python_version>='3.12'" ] -azure = [ "azure-cognitiveservices-speech~=1.42.0"] +azure = [ "azure-cognitiveservices-speech~=1.44.0"] cartesia = [ "cartesia~=2.0.3", "pipecat-ai[websockets-base]" ] cerebras = [] daily = [ "daily-python~=0.23.0" ] diff --git a/src/pipecat/services/azure/tts.py b/src/pipecat/services/azure/tts.py index 309f5e6b8..32b3fa60b 100644 --- a/src/pipecat/services/azure/tts.py +++ b/src/pipecat/services/azure/tts.py @@ -411,7 +411,12 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): Args: evt: Cancellation event. """ - logger.error(f"Speech synthesis canceled: {evt.result.cancellation_details.reason}") + reason = evt.result.cancellation_details.reason + # User cancellation (from interruption) is expected, not an error + if reason == CancellationReason.CancelledByUser: + logger.debug(f"{self}: Speech synthesis canceled by user (interruption)") + else: + logger.error(f"{self}: Speech synthesis canceled: {reason}") self._audio_queue.put_nowait(None) async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): @@ -440,6 +445,17 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): """ await super()._handle_interruption(frame, direction) await self.stop_all_metrics() + + # Stop Azure synthesis to prevent more word boundaries from being added + if self._speech_synthesizer: + try: + # stop_speaking_async() returns a ResultFuture + # We need to call .get() in a thread to wait for completion + result_future = self._speech_synthesizer.stop_speaking_async() + await asyncio.to_thread(result_future.get) + except Exception as e: + logger.error(f"{self} error stopping synthesis: {e}") + # Reset cumulative audio offset on interruption self._cumulative_audio_offset = 0.0 # Clear the audio queue diff --git a/uv.lock b/uv.lock index b1df57d60..7dd5e0562 100644 --- a/uv.lock +++ b/uv.lock @@ -484,15 +484,31 @@ wheels = [ [[package]] name = "azure-cognitiveservices-speech" -version = "1.42.0" +version = "1.44.0" source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "azure-core" }, +] wheels = [ - { url = "https://files.pythonhosted.org/packages/86/1d/07fc84ab9590fae9cc66a789d1971a0e3494e605e1787879c3581c5a385a/azure_cognitiveservices_speech-1.42.0-py3-none-macosx_10_14_x86_64.whl", hash = "sha256:ad45a18ad6973a4fa2dbd4d71ded3a1a02c4dbbf13696b08f7a16f4156dddce7", size = 7420332, upload-time = "2025-01-13T22:10:18.831Z" }, - { url = "https://files.pythonhosted.org/packages/a1/72/7ebe03784b220b9adece692adc31ec6e4a1bac96c3e9d3fef511b5ca08aa/azure_cognitiveservices_speech-1.42.0-py3-none-macosx_11_0_arm64.whl", hash = "sha256:9105a64a9d83044790f4f8c9358b6ea66a7c042cbd67173db303501782e62d3f", size = 7277345, upload-time = "2025-01-13T22:10:24.23Z" }, - { url = "https://files.pythonhosted.org/packages/83/f7/9241ad7154e554730ea56271e14ad1115c278b26a81eb892eac16fabb480/azure_cognitiveservices_speech-1.42.0-py3-none-manylinux1_x86_64.whl", hash = "sha256:90890a147499239f37b0b1a5112c51820b90fa2b5adafa0df4da6cc0c211887f", size = 39727186, upload-time = "2025-01-13T22:10:01.628Z" }, - { url = "https://files.pythonhosted.org/packages/fc/fd/af607bdfa95306b13fcdeadcd48d28b80b27cc5e3b99e2bde96f6212cd3a/azure_cognitiveservices_speech-1.42.0-py3-none-manylinux2014_aarch64.whl", hash = "sha256:106fbdb165a215cada47d7e95851e0b9d2755a3f2355369bab4915ad001efe89", size = 39508123, upload-time = "2025-01-13T22:10:13.719Z" }, - { url = "https://files.pythonhosted.org/packages/17/fb/1c998efbfcb1e44f9dc4dbb8b182ea3e5287fdc167aa352aef4685e29435/azure_cognitiveservices_speech-1.42.0-py3-none-win32.whl", hash = "sha256:7d57218beec24360a8b7ce89755c2c133259e3411c233ef0a659b951e4c4c904", size = 2109807, upload-time = "2025-01-13T22:09:50.516Z" }, - { url = "https://files.pythonhosted.org/packages/52/bb/ef7a29f5717cca646be6698d80e542446a6a442be897c8f67bf93551c672/azure_cognitiveservices_speech-1.42.0-py3-none-win_amd64.whl", hash = "sha256:32076ee03b3b402a2e8841f2c21e5cd54dc3ffbf5af183426344727298c8bbd4", size = 2377971, upload-time = "2025-01-13T22:09:44.706Z" }, + { url = "https://files.pythonhosted.org/packages/0b/0d/0752835f079e8d2cc42bb634f3ccd761c8d6e9d0d46a2d6cf7b3ed8e714c/azure_cognitiveservices_speech-1.44.0-py3-none-macosx_10_14_x86_64.whl", hash = "sha256:78037a147ba72abb57e8c10b693d43a1bb029986fae0918f1f9b7d6342737bfe", size = 7492396, upload-time = "2025-05-19T15:46:11.318Z" }, + { url = "https://files.pythonhosted.org/packages/76/1d/d0ed4ec0f51303a2a532dc845eeb72c7729a3c8639b08050f3c1cd96db79/azure_cognitiveservices_speech-1.44.0-py3-none-macosx_11_0_arm64.whl", hash = "sha256:2c9b436326cd8dd82dfa88454b7b68359dfc7149e2ac9029f9bcff155ebd5c95", size = 7347577, upload-time = "2025-05-19T15:46:13.644Z" }, + { url = "https://files.pythonhosted.org/packages/89/c8/f0a4ea8bea014b912046f737e429378ceadad68258395454d62acf7f65bb/azure_cognitiveservices_speech-1.44.0-py3-none-manylinux1_x86_64.whl", hash = "sha256:e5f07fc0587067850288c17aebf33d307d2c1ef9e0b2d11d9f44bff2af400568", size = 40977193, upload-time = "2025-05-19T15:46:15.878Z" }, + { url = "https://files.pythonhosted.org/packages/6a/0d/0a0394e8102d6660afeec6b780c451401f6074b1e19f00e90785529e459e/azure_cognitiveservices_speech-1.44.0-py3-none-manylinux2014_aarch64.whl", hash = "sha256:3461e22cf04816f69a964d936218d920240f987c0656fdaaf46571529ff0f7e6", size = 40747860, upload-time = "2025-05-19T15:46:19.316Z" }, + { url = "https://files.pythonhosted.org/packages/55/ad/3b7f6eca73040821358ce01f22067446a03d876bfed41cd784291706db4c/azure_cognitiveservices_speech-1.44.0-py3-none-win32.whl", hash = "sha256:a3fe7fd67ba7db281ae490de3d71b5a22648454ec2630eb6a70797f666330586", size = 2164045, upload-time = "2025-05-19T15:46:22.373Z" }, + { url = "https://files.pythonhosted.org/packages/83/ac/f491487d7d0e25ae2929b4f07e7f9b7456feb38e65b36fb605b2c9685b10/azure_cognitiveservices_speech-1.44.0-py3-none-win_amd64.whl", hash = "sha256:77cfb5dd40733b7ccc21edc427e9fb4720997832ea8a1ba460dc94345f3588ae", size = 2422937, upload-time = "2025-05-19T15:46:23.657Z" }, +] + +[[package]] +name = "azure-core" +version = "1.37.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "requests" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/ef/83/41c9371c8298999c67b007e308a0a3c4d6a59c6908fa9c62101f031f886f/azure_core-1.37.0.tar.gz", hash = "sha256:7064f2c11e4b97f340e8e8c6d923b822978be3016e46b7bc4aa4b337cfb48aee", size = 357620, upload-time = "2025-12-11T20:05:13.518Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ee/34/a9914e676971a13d6cc671b1ed172f9804b50a3a80a143ff196e52f4c7ee/azure_core-1.37.0-py3-none-any.whl", hash = "sha256:b3abe2c59e7d6bb18b38c275a5029ff80f98990e7c90a5e646249a56630fcc19", size = 214006, upload-time = "2025-12-11T20:05:14.96Z" }, ] [[package]] @@ -4111,7 +4127,7 @@ requires-dist = [ { name = "audioop-lts", marker = "python_full_version >= '3.13'", specifier = "~=0.2.1" }, { name = "aws-sdk-bedrock-runtime", marker = "python_full_version >= '3.12' and extra == 'aws-nova-sonic'", specifier = "~=0.2.0" }, { name = "aws-sdk-sagemaker-runtime-http2", marker = "python_full_version >= '3.12' and extra == 'sagemaker'" }, - { name = "azure-cognitiveservices-speech", marker = "extra == 'azure'", specifier = "~=1.42.0" }, + { name = "azure-cognitiveservices-speech", marker = "extra == 'azure'", specifier = "~=1.44.0" }, { name = "cartesia", marker = "extra == 'cartesia'", specifier = "~=2.0.3" }, { name = "coremltools", marker = "extra == 'local-smart-turn'", specifier = ">=8.0" }, { name = "daily-python", marker = "extra == 'daily'", specifier = "~=0.23.0" }, From 95f00a3c4bea3f51c9260b47efdc6df5c43376b2 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 7 Jan 2026 15:45:30 -0500 Subject: [PATCH 2/4] AzureTTSService: Align error handling with Pipecat norms --- src/pipecat/services/azure/tts.py | 22 ++++++++-------------- 1 file changed, 8 insertions(+), 14 deletions(-) diff --git a/src/pipecat/services/azure/tts.py b/src/pipecat/services/azure/tts.py index 32b3fa60b..34c212f83 100644 --- a/src/pipecat/services/azure/tts.py +++ b/src/pipecat/services/azure/tts.py @@ -365,12 +365,9 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): # Use thread-safe queue since this is called from Azure SDK thread if word: logger.trace(f"{self}: Word boundary - '{word}' at {absolute_seconds:.2f}s") - try: - # Put in temporary queue - will be processed by async task - # Store as (word, timestamp_in_seconds) tuple - self._word_boundary_queue.put_nowait((word, absolute_seconds)) - except Exception as e: - logger.error(f"{self} error queuing word timestamp: {e}") + # Put in temporary queue - will be processed by async task + # Store as (word, timestamp_in_seconds) tuple + self._word_boundary_queue.put_nowait((word, absolute_seconds)) async def _word_processor_task_handler(self): """Process word timestamps from the queue and call add_word_timestamps.""" @@ -382,7 +379,7 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): except asyncio.CancelledError: break except Exception as e: - logger.error(f"{self} error processing word timestamp: {e}") + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) def _handle_synthesizing(self, evt): """Handle audio chunks as they arrive. @@ -416,7 +413,7 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): if reason == CancellationReason.CancelledByUser: logger.debug(f"{self}: Speech synthesis canceled by user (interruption)") else: - logger.error(f"{self}: Speech synthesis canceled: {reason}") + logger.warning(f"{self}: Speech synthesis canceled: {reason}") self._audio_queue.put_nowait(None) async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): @@ -454,7 +451,7 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): result_future = self._speech_synthesizer.stop_speaking_async() await asyncio.to_thread(result_future.get) except Exception as e: - logger.error(f"{self} error stopping synthesis: {e}") + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) # Reset cumulative audio offset on interruption self._cumulative_audio_offset = 0.0 @@ -494,9 +491,6 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): try: if self._speech_synthesizer is None: - error_msg = "Speech synthesizer not initialized." - logger.error(error_msg) - yield ErrorFrame(error=error_msg) return try: @@ -526,14 +520,14 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): yield frame except Exception as e: - logger.error(f"{self} error during synthesis: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") yield TTSStoppedFrame() self._started = False # Could add reconnection logic here if needed return except Exception as e: - logger.error(f"{self} exception: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") class AzureHttpTTSService(TTSService, AzureBaseTTSService): From 79a6adbcf36d6bd35ee5acc855458cdc94019bf7 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 7 Jan 2026 16:14:25 -0500 Subject: [PATCH 3/4] AzureTTSService: Handle first chunk only for timestamps and TTFB metrics --- src/pipecat/services/azure/tts.py | 24 +++++++++++++++++------- 1 file changed, 17 insertions(+), 7 deletions(-) diff --git a/src/pipecat/services/azure/tts.py b/src/pipecat/services/azure/tts.py index 34c212f83..2a0743626 100644 --- a/src/pipecat/services/azure/tts.py +++ b/src/pipecat/services/azure/tts.py @@ -283,6 +283,7 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): self._word_boundary_queue = asyncio.Queue() self._word_processor_task = None self._started = False + self._first_chunk = True self._cumulative_audio_offset: float = 0.0 # Cumulative audio duration in seconds async def start(self, frame: StartFrame): @@ -425,10 +426,16 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): """ await super().push_frame(frame, direction) if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): - self._started = False + self._reset_state() if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("Reset", 0)]) + def _reset_state(self): + """Reset TTS state between turns.""" + self._started = False + self._first_chunk = True + self._cumulative_audio_offset = 0.0 + async def flush_audio(self): """Flush any pending audio data.""" logger.trace(f"{self}: flushing audio") @@ -453,8 +460,8 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): except Exception as e: await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) - # Reset cumulative audio offset on interruption - self._cumulative_audio_offset = 0.0 + # Reset state on interruption + self._reset_state() # Clear the audio queue while not self._audio_queue.empty(): try: @@ -498,19 +505,23 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): await self.start_ttfb_metrics() yield TTSStartedFrame() self._started = True + self._first_chunk = True ssml = self._construct_ssml(text) self._speech_synthesizer.speak_ssml_async(ssml) await self.start_tts_usage_metrics(text) # Stream audio chunks as they arrive + while True: chunk = await self._audio_queue.get() if chunk is None: # End of stream break - await self.stop_ttfb_metrics() - await self.start_word_timestamps() + if self._first_chunk: + await self.stop_ttfb_metrics() + await self.start_word_timestamps() + self._first_chunk = False frame = TTSAudioRawFrame( audio=chunk, @@ -522,8 +533,7 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") yield TTSStoppedFrame() - self._started = False - # Could add reconnection logic here if needed + self._reset_state() return except Exception as e: From 89b36f2b25b3120755a1926d7200ecd3f09e357a Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 7 Jan 2026 16:32:32 -0500 Subject: [PATCH 4/4] AzureTTSService: Restore metrics generation --- src/pipecat/services/azure/tts.py | 24 ++++++++++++++++-------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/src/pipecat/services/azure/tts.py b/src/pipecat/services/azure/tts.py index 2a0743626..14d216bc8 100644 --- a/src/pipecat/services/azure/tts.py +++ b/src/pipecat/services/azure/tts.py @@ -144,14 +144,6 @@ class AzureBaseTTSService: self._voice_id = voice self._speech_synthesizer = None - def can_generate_metrics(self) -> bool: - """Check if this service can generate processing metrics. - - Returns: - True, as Azure TTS service supports metrics generation. - """ - return True - def language_to_service_language(self, language: Language) -> Optional[str]: """Convert a Language enum to Azure language format. @@ -286,6 +278,14 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): self._first_chunk = True self._cumulative_audio_offset: float = 0.0 # Cumulative audio duration in seconds + def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as Azure TTS service supports metrics generation. + """ + return True + async def start(self, frame: StartFrame): """Start the Azure TTS service and initialize speech synthesizer. @@ -576,6 +576,14 @@ class AzureHttpTTSService(TTSService, AzureBaseTTSService): self._speech_config = None self._speech_synthesizer = None + def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as Azure TTS service supports metrics generation. + """ + return True + async def start(self, frame: StartFrame): """Start the Azure HTTP TTS service and initialize speech synthesizer.