From f8610a69a5a9af67d8228de09630650bb9ca07f5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 13 Mar 2025 19:41:33 -0700 Subject: [PATCH] introduce text aggregators --- CHANGELOG.md | 5 ++ src/pipecat/services/ai_services.py | 20 +++---- .../utils/text/base_text_aggregator.py | 57 +++++++++++++++++++ .../utils/text/simple_text_aggregator.py | 42 ++++++++++++++ tests/test_simple_text_aggregator.py | 29 ++++++++++ 5 files changed, 143 insertions(+), 10 deletions(-) create mode 100644 src/pipecat/utils/text/base_text_aggregator.py create mode 100644 src/pipecat/utils/text/simple_text_aggregator.py create mode 100644 tests/test_simple_text_aggregator.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 4feca2771..a48fa9c3b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added new `BaseTextAggregator`. Text aggregators are used by the TTS service + to aggregate LLM tokens and decide when the aggregated text should be pushed + to the TTS service. It also allows for the text to be manipulated while it's + being aggregated. + - Added new `UltravoxSTTService`. (see https://github.com/fixie-ai/ultravox) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 8533a23fc..1701f9829 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -45,8 +45,9 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.websocket_service import WebsocketService from pipecat.transcriptions.language import Language -from pipecat.utils.string import match_endofsentence +from pipecat.utils.text.base_text_aggregator import BaseTextAggregator from pipecat.utils.text.base_text_filter import BaseTextFilter +from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator from pipecat.utils.time import seconds_to_nanoseconds @@ -237,6 +238,9 @@ class TTSService(AIService): pause_frame_processing: bool = False, # TTS output sample rate sample_rate: Optional[int] = None, + # Text aggregator to aggregate incoming tokens and decide when to push to the TTS. + text_aggregator: Optional[BaseTextAggregator] = None, + # Text filter executed after text has been aggregated. text_filter: Optional[BaseTextFilter] = None, **kwargs, ): @@ -252,12 +256,12 @@ class TTSService(AIService): self._sample_rate = 0 self._voice_id: str = "" self._settings: Dict[str, Any] = {} + self._text_aggregator: BaseTextAggregator = text_aggregator or SimpleTextAggregator() self._text_filter: Optional[BaseTextFilter] = text_filter self._stop_frame_task: Optional[asyncio.Task] = None self._stop_frame_queue: asyncio.Queue = asyncio.Queue() - self._current_sentence: str = "" self._processing_text: bool = False @property @@ -336,8 +340,8 @@ class TTSService(AIService): # pause to avoid audio overlapping. await self._maybe_pause_frame_processing() - sentence = self._current_sentence - self._current_sentence = "" + sentence = self._text_aggregator.text + self._text_aggregator.reset() self._processing_text = False await self._push_tts_frames(sentence) if isinstance(frame, LLMFullResponseEndFrame): @@ -382,8 +386,8 @@ class TTSService(AIService): await self._stop_frame_queue.put(frame) async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): - self._current_sentence = "" self._processing_text = False + self._text_aggregator.handle_interruption() if self._text_filter: self._text_filter.handle_interruption() @@ -400,11 +404,7 @@ class TTSService(AIService): if not self._aggregate_sentences: text = frame.text else: - self._current_sentence += frame.text - eos_end_marker = match_endofsentence(self._current_sentence) - if eos_end_marker: - text = self._current_sentence[:eos_end_marker] - self._current_sentence = self._current_sentence[eos_end_marker:] + text = self._text_aggregator.aggregate(frame.text) if text: await self._push_tts_frames(text) diff --git a/src/pipecat/utils/text/base_text_aggregator.py b/src/pipecat/utils/text/base_text_aggregator.py new file mode 100644 index 000000000..452e5598e --- /dev/null +++ b/src/pipecat/utils/text/base_text_aggregator.py @@ -0,0 +1,57 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from abc import ABC, abstractmethod +from typing import Optional + + +class BaseTextAggregator(ABC): + """This is the base class for text aggregators. Text aggregators are usually + used by the TTS service to aggregate LLM tokens and decide when the + aggregated text should be pushed to the TTS service. + + Text aggregators can also be used to manipulate text while it's being + aggregated (e.g. reasoning blocks can be removed). + + """ + + @property + @abstractmethod + def text(self) -> str: + """Returns the currently aggregated text.""" + pass + + @abstractmethod + def aggregate(self, text: str) -> Optional[str]: + """Aggregates the specified text with the currently accumulated text. + + This method should be implemented to define how the new text contributes + to the aggregation process. It returns the updated aggregated text if + it's ready to be processed, or None otherwise. + + Args: + text (str): The text to be aggregated. + + Returns: + Optional[str]: The updated aggregated text or None if aggregated + text is not ready. + + """ + pass + + @abstractmethod + def handle_interruption(self): + """Handles interruptions. When an interruption occurs it is possible + that we might want to discard the aggregated text or do some internal + modifications to the aggregated text. + + """ + pass + + @abstractmethod + def reset(self): + """Clears the internally aggregated text.""" + pass diff --git a/src/pipecat/utils/text/simple_text_aggregator.py b/src/pipecat/utils/text/simple_text_aggregator.py new file mode 100644 index 000000000..9022fc25a --- /dev/null +++ b/src/pipecat/utils/text/simple_text_aggregator.py @@ -0,0 +1,42 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from typing import Optional + +from pipecat.utils.string import match_endofsentence +from pipecat.utils.text.base_text_aggregator import BaseTextAggregator + + +class SimpleTextAggregator(BaseTextAggregator): + """This is a simple text aggregator. It aggregates text until an end of + sentence is found. + + """ + + def __init__(self): + self._text = "" + + @property + def text(self) -> str: + return self._text + + def aggregate(self, text: str) -> Optional[str]: + result: Optional[str] = None + + self._text += text + + eos_end_marker = match_endofsentence(self._text) + if eos_end_marker: + result = self._text[:eos_end_marker] + self._text = self._text[eos_end_marker:] + + return result + + def handle_interruption(self): + self._text = "" + + def reset(self): + self._text = "" diff --git a/tests/test_simple_text_aggregator.py b/tests/test_simple_text_aggregator.py new file mode 100644 index 000000000..10c4c6a88 --- /dev/null +++ b/tests/test_simple_text_aggregator.py @@ -0,0 +1,29 @@ +# +# Copyright (c) 2024-2025 Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import unittest + +from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator + + +class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.aggregator = SimpleTextAggregator() + + async def test_reset_aggregations(self): + assert self.aggregator.aggregate("Hello ") == None + assert self.aggregator.text == "Hello " + self.aggregator.reset() + assert self.aggregator.text == "" + + async def test_simple_sentence(self): + assert self.aggregator.aggregate("Hello ") == None + assert self.aggregator.aggregate("Pipecat!") == "Hello Pipecat!" + assert self.aggregator.text == "" + + async def test_multiple_sentences(self): + assert self.aggregator.aggregate("Hello Pipecat! How are ") == "Hello Pipecat!" + assert self.aggregator.aggregate("you?") == " How are you?"