Select sentence tokenizer language from TTS settings

changelog/5727.added.md

@@ -0,0 +1 @@
+- TTS sentence aggregation selects a Punkt model from the configured language, with an optional `text_aggregation_language` override. Runtime language changes apply to the next generation; streaming contexts and explicitly associated RTVI observers retain the generation's tokenizer language.

docs/architecture/sentence-tokenization.md

@@ -0,0 +1,92 @@
+# Sentence tokenization
+
+Pipecat reads Punkt model parameters from its packaged `punkt_tab.zip` archive.
+Importing the text utilities does not import NLTK. Pipeline warm-up loads the
+English model in a background thread; TTS services also prepare their selected
+model while services connect. Models are cached by language. Loading requires
+neither network access nor an external NLTK data directory, and does not change
+NLTK's data search paths. The archive stays compressed on disk.
+
+## Selecting a language
+
+TTS sentence aggregation follows the service's `Settings.language`. Pipecat
+retains the input language before converting it to the provider's identifier.
+Regional variants share one Punkt model: `de-DE` selects `german`, for example,
+and `pt-BR` selects `portuguese`.
+
+For services that infer language from the text or voice, or when the text needs
+a different sentence model, pass `text_aggregation_language` to the service:
+
+```python
+tts = CartesiaTTSService(
+    api_key=api_key,
+    settings=CartesiaTTSService.Settings(voice=voice_id, language=Language.DE),
+)
+
+# An explicit override remains in effect when the TTS language changes.
+tts = OpenAITTSService(api_key=api_key, text_aggregation_language=Language.DE)
+```
+
+The available models are Czech, Danish, Dutch, English, Estonian, Finnish,
+French, German, Greek, Italian, Malayalam, Norwegian, Polish, Portuguese,
+Russian, Slovene, Spanish, Swedish, and Turkish. Unspecified, automatic, and
+unsupported languages use the English model plus Pipecat's additional
+punctuation handling, including boundaries such as `。`, `؟`, and `।`.
+This fallback does not detect the text's language automatically.
+
+Standalone callers can select the same model:
+
+```python
+aggregator = SimpleTextAggregator(language=Language.DE)
+boundary = match_endofsentence("Das ist z.B. wichtig. Weiter", language=Language.DE)
+```
+
+## Runtime updates
+
+Send a settings update between LLM generations:
+
+```python
+await worker.queue_frame(
+    TTSUpdateSettingsFrame(
+        service=tts,
+        delta=tts.Settings(language=Language.FR),
+    )
+)
+```
+
+The tokenizer language is captured at `LLMFullResponseStartFrame` and retained
+through that generation's text. A settings update arriving during aggregation
+selects the model for the next generation; it does not reinterpret or discard
+buffered text. Interruptions discard the buffer. Text streams without a start
+frame capture the language when their first text arrives, until an end frame
+or interruption. `TTSSpeakFrame` uses the configured language for its independent
+utterance.
+
+This snapshot controls sentence detection, not the provider's synthesis
+settings or the LLM's output language. Applications should coordinate the LLM
+and TTS language and send synthesis-setting updates between generations.
+
+In token aggregation mode, the TTS sentence tracker captures the model for each
+audio context. An older context retains its model while newer contexts use a
+different one. Newly selected models prepare in a background thread; the first
+generation using a model waits for preparation if it is not yet complete.
+
+## Observers
+
+RTVI's legacy bot-transcription messages perform their own sentence detection.
+Associate an observer explicitly with the relevant TTS service:
+
+```python
+observer = RTVIObserver(rtvi, tts_service=tts)
+```
+
+That observer aggregates LLM text when it reaches the selected TTS. Language
+snapshots are stored on frames separately for each service, so delayed observer
+callbacks do not read settings belonging to a newer generation. Other RTVI
+messages keep their existing sources and timing. Input transcription language
+is independent of the TTS selection.
+
+An observer without a TTS association uses English by default, or a fixed
+`text_aggregation_language` constructor argument. With multiple TTS services,
+associate the observer with the intended service; it does not infer one from
+pipeline topology.

src/pipecat/processors/frameworks/rtvi/observer.py

@@ -6,6 +6,7 @@
 
 """RTVI observer for converting pipeline frames to outgoing RTVI messages."""
 
+import asyncio
 import inspect
 import time
 import warnings
@@ -63,7 +64,7 @@
     TTFBMetricsData,
     TTSUsageMetricsData,
 )
-from pipecat.observers.base_observer import BaseObserver, FramePushed
+from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.processors.frameworks.rtvi.frames import (
     RTVIConfigureObserverFrame,
@@ -74,10 +75,15 @@
 )
 from pipecat.processors.frameworks.rtvi.models import BotOutputTransformResult
 from pipecat.transports.base_output import BaseOutputTransport
-from pipecat.utils.string import match_endofsentence
+from pipecat.utils.string import (
+    _sent_tokenizer,
+    match_endofsentence,
+    resolve_sentence_tokenizer_language,
+)
 
 if TYPE_CHECKING:
     from pipecat.processors.frameworks.rtvi.processor import RTVIProcessor
+    from pipecat.services.tts_service import TTSService
 
 
 class RTVIFunctionCallReportLevel(StrEnum):
@@ -211,18 +217,29 @@ def __init__(
         rtvi: Optional["RTVIProcessor"] = None,
         *,
         params: RTVIObserverParams | None = None,
+        tts_service: "TTSService | None" = None,
+        text_aggregation_language: str | None = None,
         **kwargs,
     ):
         """Initialize the RTVI observer.
 
         Args:
             rtvi: The RTVI processor to push frames to.
             params: Settings to enable/disable specific messages.
+            tts_service: TTS service whose generation language drives legacy bot
+                transcription boundaries. Only LLM text processed by this service
+                contributes to those messages; other RTVI events are unaffected.
+            text_aggregation_language: Explicit sentence-detection language, used
+                when no TTS service is associated. Defaults to English.
             **kwargs: Additional arguments passed to parent class.
         """
         super().__init__(**kwargs)
         self._rtvi = rtvi
         self._params = params or RTVIObserverParams()
+        self._tts_service = tts_service
+        self._text_aggregation_language = resolve_sentence_tokenizer_language(
+            text_aggregation_language
+        )
 
         self._ignored_sources: set[FrameProcessor] = set(self._params.ignored_sources)
         self._frames_seen = set()
@@ -766,15 +783,49 @@ async def _send_aggregated_llm_text(self, frame: AggregatedTextFrame):
             tts_message = RTVI.BotTTSTextMessage(data=RTVI.TextMessageData(text=text))
             await self.send_rtvi_message(tts_message)
 
+    async def on_process_frame(self, data: FrameProcessed):
+        """Track generation language and text at the explicitly associated TTS.
+
+        Args:
+            data: Frame and processor being observed.
+        """
+        if (
+            self._tts_service is None
+            or data.processor is not self._tts_service
+            or data.processor in self._ignored_sources
+            or data.direction != FrameDirection.DOWNSTREAM
+            or not self._params.bot_llm_enabled
+        ):
+            return
+        frame = data.frame
+        language = self._tts_service.get_text_aggregation_language(frame)
+        if isinstance(frame, LLMFullResponseStartFrame):
+            if language is not None:
+                self._text_aggregation_language = language
+                await asyncio.to_thread(_sent_tokenizer, language)
+            self._bot_transcription = ""
+        elif isinstance(frame, LLMTextFrame) and not frame.skip_tts:
+            if language is not None and language != self._text_aggregation_language:
+                self._text_aggregation_language = language
+                await asyncio.to_thread(_sent_tokenizer, language)
+            await self._aggregate_bot_transcription(frame.text)
+        elif isinstance(frame, (LLMFullResponseEndFrame, InterruptionFrame)):
+            self._bot_transcription = ""
+
     async def _handle_llm_text_frame(self, frame: LLMTextFrame):
         """Handle LLM text output frames."""
         message = RTVI.BotLLMTextMessage(data=RTVI.TextMessageData(text=frame.text))
         await self.send_rtvi_message(message)
 
+        if self._tts_service is None:
+            await self._aggregate_bot_transcription(frame.text)
+
+    async def _aggregate_bot_transcription(self, text: str):
+        """Accumulate legacy bot transcription text using this generation's model."""
         # TODO (mrkb): Remove all this logic when we fully deprecate bot-transcription messages.
-        self._bot_transcription += frame.text
+        self._bot_transcription += text
 
-        if match_endofsentence(self._bot_transcription) and len(self._bot_transcription) > 0:
+        if match_endofsentence(self._bot_transcription, language=self._text_aggregation_language):
             await self.send_rtvi_message(
                 RTVI.BotTranscriptionMessage(
                     data=RTVI.TextMessageData(text=self._bot_transcription)