Merge pull request #3489 from pipecat-ai/mb/fix-azure-tts-punctuation-spacing

fix: AzureTTSService punctuation spacing

changelog/3489.fixed.md

@@ -0,0 +1,3 @@
+- Fixed `AzureTTSService` transcript formatting issues:
+  - Punctuation now appears without extra spaces (e.g., "Hello!" instead of "Hello !")
+  - CJK languages (Chinese, Japanese, Korean) no longer have unwanted spaces between characters

src/pipecat/services/azure/tts.py

@@ -277,6 +277,8 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
         self._started = False
         self._first_chunk = True
         self._cumulative_audio_offset: float = 0.0  # Cumulative audio duration in seconds
+        self._last_word: Optional[str] = None  # Track last word for punctuation merging
+        self._last_timestamp: Optional[float] = None  # Track last timestamp
 
     def can_generate_metrics(self) -> bool:
         """Check if this service can generate processing metrics.
@@ -346,9 +348,34 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
         await self.cancel_task(self._word_processor_task)
         self._word_processor_task = None
 
+    def _is_cjk_language(self) -> bool:
+        """Check if the configured language is CJK (Chinese, Japanese, Korean).
+
+        Returns:
+            True if the language is CJK, False otherwise.
+        """
+        language = self._settings.get("language", "").lower()
+        # Check if language starts with CJK language codes
+        return language.startswith(("zh", "ja", "ko", "cmn", "yue", "wuu"))
+
+    def _is_punctuation_only(self, text: str) -> bool:
+        """Check if text consists only of punctuation and whitespace.
+
+        Args:
+            text: Text to check.
+
+        Returns:
+            True if text is only punctuation/whitespace, False otherwise.
+        """
+        return text and all(not c.isalnum() for c in text)
+
     def _handle_word_boundary(self, evt):
         """Handle word boundary events from Azure SDK.
 
+        Azure sends punctuation as separate word boundaries, and breaks CJK text
+        into individual characters/particles. This method routes to language-specific
+        handlers to properly merge and emit word boundaries.
+
         Args:
             evt: SpeechSynthesisWordBoundaryEventArgs from Azure Speech SDK
                 containing word text and audio offset timing.
@@ -362,13 +389,75 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
         # Add cumulative offset to get absolute timestamp across sentences
         absolute_seconds = self._cumulative_audio_offset + sentence_relative_seconds
 
-        # Queue word timestamp for async processing
-        # 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")
-            # 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))
+        if not word:
+            return
+
+        # Route to language-specific handler
+        if self._is_cjk_language():
+            self._handle_cjk_word_boundary(word, absolute_seconds)
+        else:
+            self._handle_non_cjk_word_boundary(word, absolute_seconds)
+
+    def _emit_pending_word(self):
+        """Emit the currently buffered word if one exists."""
+        if self._last_word is not None:
+            self._word_boundary_queue.put_nowait((self._last_word, self._last_timestamp))
+            self._last_word = None
+            self._last_timestamp = None
+
+    def _handle_cjk_word_boundary(self, word: str, timestamp: float):
+        """Handle word boundaries for CJK languages (Chinese, Japanese, Korean).
+
+        CJK languages don't use spaces between words, so we merge characters together
+        and only emit at natural break points (punctuation or whitespace boundaries).
+        Without this logic, we don't get word output for CJK languages.
+
+        Args:
+            word: The word/character from Azure.
+            timestamp: Timestamp in seconds.
+        """
+        # First word: just store it
+        if self._last_word is None:
+            self._last_word = word
+            self._last_timestamp = timestamp
+            return
+
+        # Punctuation: merge and emit (natural break)
+        if self._is_punctuation_only(word):
+            self._last_word += word
+            self._emit_pending_word()
+            return
+
+        # Whitespace: emit before boundary, start new segment
+        if word.strip() != word:
+            self._emit_pending_word()
+            self._last_word = word
+            self._last_timestamp = timestamp
+            return
+
+        # Default: continue merging CJK characters
+        self._last_word += word
+
+    def _handle_non_cjk_word_boundary(self, word: str, timestamp: float):
+        """Handle word boundaries for non-CJK languages.
+
+        Non-CJK languages use spaces between words, so we emit each word separately
+        after merging any trailing punctuation.
+
+        Args:
+            word: The word from Azure.
+            timestamp: Timestamp in seconds.
+        """
+        # Punctuation: merge with previous word (don't emit yet)
+        if self._is_punctuation_only(word) and self._last_word is not None:
+            self._last_word += word
+            return
+
+        # Regular word: emit previous, store current
+        if self._last_word is not None:
+            self._word_boundary_queue.put_nowait((self._last_word, self._last_timestamp))
+        self._last_word = word
+        self._last_timestamp = timestamp
 
     async def _word_processor_task_handler(self):
         """Process word timestamps from the queue and call add_word_timestamps."""
@@ -397,6 +486,12 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
         Args:
             evt: Completion event from Azure Speech SDK.
         """
+        # Flush any pending word before completing
+        if self._last_word is not None:
+            self._word_boundary_queue.put_nowait((self._last_word, self._last_timestamp))
+            self._last_word = None
+            self._last_timestamp = None
+
         # Update cumulative audio offset for next sentence
         if evt.result and evt.result.audio_duration:
             self._cumulative_audio_offset += evt.result.audio_duration.total_seconds()
@@ -435,6 +530,8 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
         self._started = False
         self._first_chunk = True
         self._cumulative_audio_offset = 0.0
+        self._last_word = None
+        self._last_timestamp = None
 
     async def flush_audio(self):
         """Flush any pending audio data."""