Simple content filter demo

move pipecat.sync to pipecat.utils.sync

CHANGELOG.md

@@ -9,22 +9,39 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- Added `pause_frame_processing` parameter to `CartesiaTTSService` to allow
-  disabling frame processing pausing during TTS generation.
-
 - Added `wait_for_all` argument to the base `LLMService`. When enabled, this
   ensures all function calls complete before returning results to the LLM (i.e.,
   before running a new inference with those results).
 
 ### Changed
 
+- Improved interruption handling to prevent bots from repeating themselves.
+  LLM services that return multiple sentences in a single response (e.g.,
+  `GoogleLLMService`) are now split into individual sentences before being sent
+  to TTS. This ensures interruptions occur at sentence boundaries, preventing
+  the bot from repeating content after being interrupted during long responses.
+
+- Text Aggregation Improvements:
+
+  - **Breaking Change**: `BaseTextAggregator.aggregate()` now returns
+    `AsyncIterator[Aggregation]` instead of `Optional[Aggregation]`. This
+    enables the aggregator to return multiple results based on the provided
+    text.
+  - Refactored text aggregators to use inheritance: `SkipTagsAggregator` and
+    `PatternPairAggregator` now inherit from `SimpleTextAggregator`, reusing
+    the base class's sentence detection logic.
+
 - Updated `AICFilter` to use Quail STT as the default model
   (`AICModelType.QUAIL_STT`). Quail STT is optimized for human-to-machine
   interaction (e.g., voice agents, speech-to-text) and operates at a native
   sample rate of 16 kHz with fixed enhancement parameters.
 
+- Updated Deepgram logging to include Deepgram request IDs for improved debugging.
+
 ### Deprecated
 
+- Package `pipecat.sync` is deprecated, use `pipecat.utils.sync` instead.
+
 - The `noise_gate_enable` parameter in `AICFilter` is deprecated and no longer
   has any effect. Noise gating is now handled automatically by the AIC VAD
   system. Use `AICFilter.create_vad_analyzer()` for VAD functionality instead.
@@ -39,6 +56,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue where `LLMTextFrame.skip_tts` was being overwritten by LLM
+  services.
+
+- Fixed sentence aggregation to correctly handle ambiguous punctuation in
+  streaming text, such as currency ("$29.95") and abbreviations ("Mr. Smith").
+
+- Fixed bug in `PatternPairAggregator` where pattern handlers could be called
+  multiple times for `KEEP` or `AGGREGATE` patterns.
+
 - Fixed an issue in `SarvamTTSService` where the last sentence was not being
   spoken. Now, audio is flushed when the TTS services receives the
   `LLMFullResponseEndFrame` or `EndFrame`.

docs/api/conf.py

@@ -119,7 +119,6 @@ def import_core_modules():
         "pipecat.observers",
         "pipecat.runner",
         "pipecat.serializers",
-        "pipecat.sync",
         "pipecat.transcriptions",
         "pipecat.utils",
     ]

docs/api/index.rst

@@ -30,7 +30,6 @@ Quick Links
    Runner <api/pipecat.runner>
    Serializers <api/pipecat.serializers>
    Services <api/pipecat.services>
-   Sync <api/pipecat.sync>
    Transcriptions <api/pipecat.transcriptions>
    Transports <api/pipecat.transports>
-   Utils <api/pipecat.utils>
+   Utils <api/pipecat.utils>

examples/foundational/07-interruptible.py

@@ -13,12 +13,13 @@ from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
 from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.audio.vad.vad_analyzer import VADParams
-from pipecat.frames.frames import LLMRunFrame
+from pipecat.frames.frames import Frame, LLMContextFrame, 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
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
@@ -30,6 +31,44 @@ from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
 
 load_dotenv(override=True)
 
+FILTERED_WORDS = ["apple", "banana", "car"]
+
+
+class ContentFilterProcessor(FrameProcessor):
+    """Processor that filters LLMContextFrames containing specific words.
+
+    If the user's message contains any of the filtered words, the context
+    is replaced with a message indicating the assistant cannot respond.
+    """
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, LLMContextFrame):
+            # Check the last user message for filtered words
+            messages = frame.context.messages
+            if messages:
+                last_message = messages[-1]
+                content = last_message.get("content", "")
+                if isinstance(content, str):
+                    content_lower = content.lower()
+                    if any(word in content_lower for word in FILTERED_WORDS):
+                        logger.info(f"Filtered content detected: {content}")
+                        # Create a new context with a filtered response instruction
+                        filtered_context = LLMContext(
+                            messages=[
+                                {
+                                    "role": "system",
+                                    "content": "The user is asking about something you cannot give an answer about. Tell them you don't know how to respond.",
+                                }
+                            ]
+                        )
+                        await self.push_frame(LLMContextFrame(filtered_context), direction)
+                        return
+
+        await self.push_frame(frame, direction)
+
+
 # We store functions so objects (e.g. SileroVADAnalyzer) don't get
 # instantiated. The function will be called when the desired transport gets
 # selected.
@@ -76,12 +115,14 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
 
     context = LLMContext(messages)
     context_aggregator = LLMContextAggregatorPair(context)
+    content_filter = ContentFilterProcessor()
 
     pipeline = Pipeline(
         [
             transport.input(),  # Transport user input
             stt,
             context_aggregator.user(),  # User responses
+            content_filter,  # Content filter
             llm,  # LLM
             tts,  # TTS
             transport.output(),  # Transport bot output

examples/foundational/22-natural-conversation.py

@@ -28,10 +28,10 @@ from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.llm_service import LLMService
 from pipecat.services.openai.llm import OpenAIContextAggregatorPair, OpenAILLMService
-from pipecat.sync.event_notifier import EventNotifier
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
 from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+from pipecat.utils.sync.event_notifier import EventNotifier
 
 load_dotenv(override=True)
 

examples/foundational/22b-natural-conversation-proposal.py

@@ -45,11 +45,11 @@ from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.llm_service import FunctionCallParams, LLMService
 from pipecat.services.openai.llm import OpenAILLMService
-from pipecat.sync.base_notifier import BaseNotifier
-from pipecat.sync.event_notifier import EventNotifier
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
 from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+from pipecat.utils.sync.base_notifier import BaseNotifier
+from pipecat.utils.sync.event_notifier import EventNotifier
 from pipecat.utils.time import time_now_iso8601
 
 load_dotenv(override=True)

examples/foundational/22c-natural-conversation-mixed-llms.py

@@ -46,11 +46,11 @@ from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.llm_service import FunctionCallParams, LLMService
 from pipecat.services.openai.llm import OpenAILLMService
-from pipecat.sync.base_notifier import BaseNotifier
-from pipecat.sync.event_notifier import EventNotifier
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
 from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+from pipecat.utils.sync.base_notifier import BaseNotifier
+from pipecat.utils.sync.event_notifier import EventNotifier
 from pipecat.utils.time import time_now_iso8601
 
 load_dotenv(override=True)

examples/foundational/22d-natural-conversation-gemini-audio.py

@@ -47,11 +47,11 @@ from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.google.llm import GoogleLLMService
 from pipecat.services.llm_service import LLMService
-from pipecat.sync.base_notifier import BaseNotifier
-from pipecat.sync.event_notifier import EventNotifier
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
 from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+from pipecat.utils.sync.base_notifier import BaseNotifier
+from pipecat.utils.sync.event_notifier import EventNotifier
 from pipecat.utils.time import time_now_iso8601
 
 load_dotenv(override=True)

src/pipecat/extensions/ivr/ivr_navigator.py

@@ -18,8 +18,10 @@ from loguru import logger
 from pipecat.audio.dtmf.types import KeypadEntry
 from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.frames.frames import (
+    EndFrame,
     Frame,
     LLMContextFrame,
+    LLMFullResponseEndFrame,
     LLMMessagesUpdateFrame,
     LLMTextFrame,
     OutputDTMFUrgentFrame,
@@ -149,11 +151,18 @@ class IVRProcessor(FrameProcessor):
 
         elif isinstance(frame, LLMTextFrame):
             # Process text through the pattern aggregator
-            result = await self._aggregator.aggregate(frame.text)
-            if result:
+            async for result in self._aggregator.aggregate(frame.text):
                 # Push aggregated text that doesn't contain XML patterns
                 await self.push_frame(LLMTextFrame(result.text), direction)
 
+        elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)):
+            # Flush any remaining text from the aggregator
+            remaining = await self._aggregator.flush()
+            if remaining:
+                await self.push_frame(LLMTextFrame(remaining.text), direction)
+            # Push the end frame
+            await self.push_frame(frame, direction)
+
         else:
             await self.push_frame(frame, direction)
 

src/pipecat/extensions/voicemail/voicemail_detector.py

@@ -40,8 +40,8 @@ from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
 from pipecat.services.llm_service import LLMService
-from pipecat.sync.base_notifier import BaseNotifier
-from pipecat.sync.event_notifier import EventNotifier
+from pipecat.utils.sync.base_notifier import BaseNotifier
+from pipecat.utils.sync.event_notifier import EventNotifier
 
 
 class NotifierGate(FrameProcessor):

src/pipecat/frames/frames.py

@@ -330,7 +330,7 @@ class TextFrame(DataFrame):
     """
 
     text: str
-    skip_tts: bool = field(init=False)
+    skip_tts: Optional[bool] = field(init=False)
     # Whether any necessary inter-frame (leading/trailing) spaces are already
     # included in the text.
     # NOTE: Ideally this would be available at init time with a default value,
@@ -343,7 +343,7 @@ class TextFrame(DataFrame):
 
     def __post_init__(self):
         super().__post_init__()
-        self.skip_tts = False
+        self.skip_tts = None
         self.includes_inter_frame_spaces = False
         self.append_to_context = True
 
@@ -1632,22 +1632,22 @@ class LLMFullResponseStartFrame(ControlFrame):
     more TextFrames and a final LLMFullResponseEndFrame.
     """
 
-    skip_tts: bool = field(init=False)
+    skip_tts: Optional[bool] = field(init=False)
 
     def __post_init__(self):
         super().__post_init__()
-        self.skip_tts = False
+        self.skip_tts = None
 
 
 @dataclass
 class LLMFullResponseEndFrame(ControlFrame):
     """Frame indicating the end of an LLM response."""
 
-    skip_tts: bool = field(init=False)
+    skip_tts: Optional[bool] = field(init=False)
 
     def __post_init__(self):
         super().__post_init__()
-        self.skip_tts = False
+        self.skip_tts = None
 
 
 @dataclass

src/pipecat/processors/aggregators/gated_llm_context.py

@@ -9,7 +9,7 @@
 from pipecat.frames.frames import CancelFrame, EndFrame, Frame, LLMContextFrame, StartFrame
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.sync.base_notifier import BaseNotifier
+from pipecat.utils.sync.base_notifier import BaseNotifier
 
 
 class GatedLLMContextAggregator(FrameProcessor):

src/pipecat/processors/aggregators/llm_text_processor.py

@@ -83,8 +83,7 @@ class LLMTextProcessor(FrameProcessor):
         await self._text_aggregator.reset()
 
     async def _handle_llm_text(self, in_frame: LLMTextFrame):
-        aggregation = await self._text_aggregator.aggregate(in_frame.text)
-        if aggregation:
+        async for aggregation in self._text_aggregator.aggregate(in_frame.text):
             out_frame = AggregatedTextFrame(
                 text=aggregation.text,
                 aggregated_by=aggregation.type,
@@ -92,15 +91,13 @@ class LLMTextProcessor(FrameProcessor):
             out_frame.skip_tts = in_frame.skip_tts
             await self.push_frame(out_frame)
 
-    async def _handle_llm_end(self, skip_tts: bool = False):
-        # Flush any remaining aggregated text at the end of the LLM response
-        aggregation = self._text_aggregator.text
-        await self._text_aggregator.reset()
-        text = aggregation.text.strip()
-        if text:
+    async def _handle_llm_end(self, skip_tts: Optional[bool] = None):
+        # Flush any remaining text
+        remaining = await self._text_aggregator.flush()
+        if remaining:
             out_frame = AggregatedTextFrame(
-                text=text,
-                aggregated_by=aggregation.type,
+                text=remaining.text,
+                aggregated_by=remaining.type,
             )
             out_frame.skip_tts = skip_tts
             await self.push_frame(out_frame)

src/pipecat/processors/filters/wake_notifier_filter.py

@@ -10,7 +10,7 @@ from typing import Awaitable, Callable, Tuple, Type
 
 from pipecat.frames.frames import Frame
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.sync.base_notifier import BaseNotifier
+from pipecat.utils.sync.base_notifier import BaseNotifier
 
 
 class WakeNotifierFilter(FrameProcessor):

src/pipecat/services/cartesia/tts.py

@@ -234,7 +234,6 @@ class CartesiaTTSService(AudioContextWordTTSService):
         params: Optional[InputParams] = None,
         text_aggregator: Optional[BaseTextAggregator] = None,
         aggregate_sentences: Optional[bool] = True,
-        pause_frame_processing: Optional[bool] = True,
         **kwargs,
     ):
         """Initialize the Cartesia TTS service.
@@ -255,7 +254,6 @@ class CartesiaTTSService(AudioContextWordTTSService):
                     Use an LLMTextProcessor before the TTSService for custom text aggregation.
 
             aggregate_sentences: Whether to aggregate sentences within the TTSService.
-            pause_frame_processing: Whether to pause processing frames while receiving audio.
             **kwargs: Additional arguments passed to the parent service.
         """
         # Aggregating sentences still gives cleaner-sounding results and fewer
@@ -271,7 +269,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
         super().__init__(
             aggregate_sentences=aggregate_sentences,
             push_text_frames=False,
-            pause_frame_processing=pause_frame_processing,
+            pause_frame_processing=True,
             sample_rate=sample_rate,
             text_aggregator=text_aggregator,
             **kwargs,

src/pipecat/services/deepgram/flux/stt.py

@@ -244,6 +244,11 @@ class DeepgramFluxSTTService(WebsocketSTTService):
                 additional_headers={"Authorization": f"Token {self._api_key}"},
             )
 
+            headers = {
+                k: v for k, v in self._websocket.response.headers.items() if k.startswith("dg-")
+            }
+            logger.debug(f'{self}: Websocket connection initialized: {{"headers": {headers}}}')
+
             # Creating the receiver task
             if not self._receive_task:
                 self._receive_task = self.create_task(

src/pipecat/services/deepgram/stt.py

@@ -234,6 +234,13 @@ class DeepgramSTTService(STTService):
 
         if not await self._connection.start(options=self._settings, addons=self._addons):
             await self.push_error(error_msg=f"Unable to connect to Deepgram")
+        else:
+            headers = {
+                k: v
+                for k, v in self._connection._socket.response.headers.items()
+                if k.startswith("dg-")
+            }
+            logger.debug(f'{self}: Websocket connection initialized: {{"headers": {headers}}}')
 
     async def _disconnect(self):
         if await self._connection.is_connected():

src/pipecat/services/deepgram/tts.py

@@ -170,6 +170,11 @@ class DeepgramTTSService(WebsocketTTSService):
 
             self._websocket = await websocket_connect(url, additional_headers=headers)
 
+            headers = {
+                k: v for k, v in self._websocket.response.headers.items() if k.startswith("dg-")
+            }
+            logger.debug(f'{self}: Websocket connection initialized: {{"headers": {headers}}}')
+
             await self._call_event_handler("on_connected")
         except Exception as e:
             logger.error(f"{self} exception: {e}")

src/pipecat/services/llm_service.py

@@ -186,7 +186,7 @@ class LLMService(AIService):
         self._function_call_tasks: Dict[Optional[asyncio.Task], FunctionCallRunnerItem] = {}
         self._sequential_runner_task: Optional[asyncio.Task] = None
         self._tracing_enabled: bool = False
-        self._skip_tts: bool = False
+        self._skip_tts: Optional[bool] = None
 
         self._register_event_handler("on_function_calls_started")
         self._register_event_handler("on_completion_timeout")
@@ -297,7 +297,8 @@ class LLMService(AIService):
             direction: The direction of frame pushing.
         """
         if isinstance(frame, (LLMTextFrame, LLMFullResponseStartFrame, LLMFullResponseEndFrame)):
-            frame.skip_tts = self._skip_tts
+            if self._skip_tts is not None:
+                frame.skip_tts = self._skip_tts
 
         await super().push_frame(frame, direction)
 

src/pipecat/services/tts_service.py

@@ -425,16 +425,13 @@ class TTSService(AIService):
             # pause to avoid audio overlapping.
             await self._maybe_pause_frame_processing()
 
-            pending_aggregation = self._text_aggregator.text
+            # Flush any remaining text (including text waiting for lookahead)
+            remaining = await self._text_aggregator.flush()
+            if remaining:
+                await self._push_tts_frames(AggregatedTextFrame(remaining.text, remaining.type))
 
             # Reset aggregator state
-            await self._text_aggregator.reset()
             self._processing_text = False
-
-            if pending_aggregation.text:
-                await self._push_tts_frames(
-                    AggregatedTextFrame(pending_aggregation.text, pending_aggregation.type)
-                )
             if isinstance(frame, LLMFullResponseEndFrame):
                 if self._push_text_frames:
                     await self.push_frame(frame, direction)
@@ -539,17 +536,20 @@ class TTSService(AIService):
             text = frame.text
             includes_inter_frame_spaces = frame.includes_inter_frame_spaces
             aggregated_by = "token"
+
+            if text:
+                logger.trace(f"Pushing TTS frames for text: {text}, {aggregated_by}")
+                await self._push_tts_frames(
+                    AggregatedTextFrame(text, aggregated_by), includes_inter_frame_spaces
+                )
         else:
-            aggregate = await self._text_aggregator.aggregate(frame.text)
-            if aggregate:
+            async for aggregate in self._text_aggregator.aggregate(frame.text):
                 text = aggregate.text
                 aggregated_by = aggregate.type
-
-        if text:
-            logger.trace(f"Pushing TTS frames for text: {text}, {aggregated_by}")
-            await self._push_tts_frames(
-                AggregatedTextFrame(text, aggregated_by), includes_inter_frame_spaces
-            )
+                logger.trace(f"Pushing TTS frames for text: {text}, {aggregated_by}")
+                await self._push_tts_frames(
+                    AggregatedTextFrame(text, aggregated_by), includes_inter_frame_spaces
+                )
 
     async def _push_tts_frames(
         self, src_frame: AggregatedTextFrame, includes_inter_frame_spaces: Optional[bool] = False

src/pipecat/sync/base_notifier.py

@@ -6,31 +6,14 @@
 
 """Base notifier interface for Pipecat."""
 
-from abc import ABC, abstractmethod
+import warnings
 
+from pipecat.utils.sync.base_notifier import BaseNotifier
 
-class BaseNotifier(ABC):
-    """Abstract base class for notification mechanisms.
-
-    Provides a standard interface for implementing notification and waiting
-    patterns used for event coordination and signaling between components
-    in the Pipecat framework.
-    """
-
-    @abstractmethod
-    async def notify(self):
-        """Send a notification signal.
-
-        Implementations should trigger any waiting coroutines or processes
-        that are blocked on this notifier.
-        """
-        pass
-
-    @abstractmethod
-    async def wait(self):
-        """Wait for a notification signal.
-
-        Implementations should block until a notification is received
-        from the corresponding notify() call.
-        """
-        pass
+with warnings.catch_warnings():
+    warnings.simplefilter("always")
+    warnings.warn(
+        "Package pipecat.sync is deprecated, use pipecat.utils.sync instead.",
+        DeprecationWarning,
+        stacklevel=2,
+    )

src/pipecat/sync/event_notifier.py

@@ -6,40 +6,14 @@
 
 """Event-based notifier implementation using asyncio Event primitives."""
 
-import asyncio
+import warnings
 
-from pipecat.sync.base_notifier import BaseNotifier
+from pipecat.utils.sync.event_notifier import EventNotifier
 
-
-class EventNotifier(BaseNotifier):
-    """Event-based notifier using asyncio.Event for task synchronization.
-
-    Provides a simple notification mechanism where one task can signal
-    an event and other tasks can wait for that event to occur. The event
-    is automatically cleared after each wait operation.
-    """
-
-    def __init__(self):
-        """Initialize the event notifier.
-
-        Creates an internal asyncio.Event for managing notifications.
-        """
-        self._event = asyncio.Event()
-
-    async def notify(self):
-        """Signal the event to notify waiting tasks.
-
-        Sets the internal event, causing any tasks waiting on this
-        notifier to be awakened.
-        """
-        self._event.set()
-
-    async def wait(self):
-        """Wait for the event to be signaled.
-
-        Blocks until another task calls notify(). Automatically clears
-        the event after being awakened so subsequent calls will wait
-        for the next notification.
-        """
-        await self._event.wait()
-        self._event.clear()
+with warnings.catch_warnings():
+    warnings.simplefilter("always")
+    warnings.warn(
+        "Package pipecat.sync is deprecated, use pipecat.utils.sync instead.",
+        DeprecationWarning,
+        stacklevel=2,
+    )

src/pipecat/utils/sync/base_notifier.py

@@ -0,0 +1,36 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Base notifier interface for Pipecat."""
+
+from abc import ABC, abstractmethod
+
+
+class BaseNotifier(ABC):
+    """Abstract base class for notification mechanisms.
+
+    Provides a standard interface for implementing notification and waiting
+    patterns used for event coordination and signaling between components
+    in the Pipecat framework.
+    """
+
+    @abstractmethod
+    async def notify(self):
+        """Send a notification signal.
+
+        Implementations should trigger any waiting coroutines or processes
+        that are blocked on this notifier.
+        """
+        pass
+
+    @abstractmethod
+    async def wait(self):
+        """Wait for a notification signal.
+
+        Implementations should block until a notification is received
+        from the corresponding notify() call.
+        """
+        pass

src/pipecat/utils/sync/event_notifier.py

@@ -0,0 +1,45 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Event-based notifier implementation using asyncio Event primitives."""
+
+import asyncio
+
+from pipecat.utils.sync.base_notifier import BaseNotifier
+
+
+class EventNotifier(BaseNotifier):
+    """Event-based notifier using asyncio.Event for task synchronization.
+
+    Provides a simple notification mechanism where one task can signal
+    an event and other tasks can wait for that event to occur. The event
+    is automatically cleared after each wait operation.
+    """
+
+    def __init__(self):
+        """Initialize the event notifier.
+
+        Creates an internal asyncio.Event for managing notifications.
+        """
+        self._event = asyncio.Event()
+
+    async def notify(self):
+        """Signal the event to notify waiting tasks.
+
+        Sets the internal event, causing any tasks waiting on this
+        notifier to be awakened.
+        """
+        self._event.set()
+
+    async def wait(self):
+        """Wait for the event to be signaled.
+
+        Blocks until another task calls notify(). Automatically clears
+        the event after being awakened so subsequent calls will wait
+        for the next notification.
+        """
+        await self._event.wait()
+        self._event.clear()

src/pipecat/utils/text/base_text_aggregator.py

@@ -14,7 +14,7 @@ aggregated text should be sent for speech synthesis.
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from enum import Enum
-from typing import Optional
+from typing import AsyncIterator, Optional
 
 
 class AggregationType(str, Enum):
@@ -80,33 +80,43 @@ class BaseTextAggregator(ABC):
         pass
 
     @abstractmethod
-    async def aggregate(self, text: str) -> Optional[Aggregation]:
-        """Aggregate the specified text with the currently accumulated text.
+    async def aggregate(self, text: str) -> AsyncIterator[Aggregation]:
+        """Aggregate the specified text and yield completed aggregations.
 
-        This method should be implemented to define how the new text contributes
-        to the aggregation process. It returns the aggregated text and a string
-        describing how it was aggregated if it's ready to be processed,
-        or None otherwise.
+        This method processes the input text character-by-character internally
+        and yields Aggregation objects as they complete.
 
         Subclasses should implement their specific logic for:
 
-        - How to combine new text with existing accumulated text
+        - How to process text character-by-character
         - When to consider the aggregated text ready for processing
         - What criteria determine text completion (e.g., sentence boundaries)
-        - When a completion occurs, the method should return an Aggregation object
-          containing the aggregated text and its type. The text should be stripped
-          of leading/trailing whitespace so that consumers can rely on a consistent
-          format.
+        - When a completion occurs, yield an Aggregation object containing the
+          aggregated text (stripped of leading/trailing whitespace) and its type
 
         Args:
             text: The text to be aggregated.
 
+        Yields:
+            Aggregation objects as they complete. Each Aggregation consists of
+            the aggregated text (stripped of leading/trailing whitespace) and
+            a string indicating the type of aggregation (e.g., 'sentence', 'word',
+            'token', 'my_custom_aggregation').
+        """
+        pass
+        # Make this a generator to satisfy type checker
+        yield  # pragma: no cover
+
+    @abstractmethod
+    async def flush(self) -> Optional[Aggregation]:
+        """Flush any pending aggregation.
+
+        This method is called at the end of a stream (e.g., when receiving
+        LLMFullResponseEndFrame) to return any text that was buffered.
+
         Returns:
-            An Aggregation object if ready for processing, or None if more
-            text is needed before the aggregated content is ready. If an Aggregation
-            object is returned, it should consist of the updated aggregated text,
-            stripped of leading/trailing whitespace, and a string indicating the
-            type of aggregation (e.g., 'sentence', 'word', 'token', 'my_custom_aggregation').
+            An Aggregation object if there is pending text, or None if there
+            is no pending text.
         """
         pass
 

src/pipecat/utils/text/pattern_pair_aggregator.py

@@ -13,12 +13,12 @@ support for custom handlers and configurable actions for when a pattern is found
 
 import re
 from enum import Enum
-from typing import Awaitable, Callable, List, Optional, Tuple
+from typing import AsyncIterator, Awaitable, Callable, List, Optional, Tuple
 
 from loguru import logger
 
-from pipecat.utils.string import match_endofsentence
-from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator
+from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType
+from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator
 
 
 class MatchAction(Enum):
@@ -72,7 +72,7 @@ class PatternMatch(Aggregation):
         return f"PatternMatch(type={self.type}, text={self.text}, full_match={self.full_match})"
 
 
-class PatternPairAggregator(BaseTextAggregator):
+class PatternPairAggregator(SimpleTextAggregator):
     """Aggregator that identifies and processes content between pattern pairs.
 
     This aggregator buffers text until it can identify complete pattern pairs
@@ -97,9 +97,10 @@ class PatternPairAggregator(BaseTextAggregator):
         Creates an empty aggregator with no patterns or handlers registered.
         Text buffering and pattern detection will begin when text is aggregated.
         """
-        self._text = ""
+        super().__init__()
         self._patterns = {}
         self._handlers = {}
+        self._last_processed_position = 0  # Track where we last checked for complete patterns
 
     @property
     def text(self) -> Aggregation:
@@ -218,14 +219,18 @@ class PatternPairAggregator(BaseTextAggregator):
         self._handlers[type] = handler
         return self
 
-    async def _process_complete_patterns(self, text: str) -> Tuple[List[PatternMatch], str]:
-        """Process all complete pattern pairs in the text.
+    async def _process_complete_patterns(
+        self, text: str, last_processed_position: int = 0
+    ) -> Tuple[List[PatternMatch], str]:
+        """Process newly complete pattern pairs in the text.
 
-        Searches for all complete pattern pairs in the text, calls the
-        appropriate handlers, and optionally removes the matches.
+        Searches for pattern pairs that have been completed since last_processed_position,
+        calls the appropriate handlers, and optionally removes the matches.
 
         Args:
             text: The text to process for pattern matches.
+            last_processed_position: The position in text that was already processed.
+                Only patterns that end at or after this position will be processed.
 
         Returns:
             Tuple of (all_matches, processed_text) where:
@@ -259,17 +264,23 @@ class PatternPairAggregator(BaseTextAggregator):
                     content=content.strip(), type=type, full_match=full_match
                 )
 
-                # Call the appropriate handler if registered
-                if type in self._handlers:
+                # Check if this pattern was already processed
+                already_processed = match.end() <= last_processed_position
+
+                # Only call handler for newly completed patterns
+                if not already_processed and type in self._handlers:
                     try:
                         await self._handlers[type](pattern_match)
                     except Exception as e:
                         logger.error(f"Error in pattern handler for {type}: {e}")
 
-                # Remove the pattern from the text if configured
+                # Handle pattern based on action
                 if action == MatchAction.REMOVE:
-                    processed_text = processed_text.replace(full_match, "", 1)
+                    # Remove patterns are only removed once (when newly completed)
+                    if not already_processed:
+                        processed_text = processed_text.replace(full_match, "", 1)
                 else:
+                    # KEEP/AGGREGATE patterns stay in all_matches
                     all_matches.append(pattern_match)
 
         return all_matches, processed_text
@@ -305,76 +316,84 @@ class PatternPairAggregator(BaseTextAggregator):
 
         return None
 
-    async def aggregate(self, text: str) -> Optional[PatternMatch]:
+    async def aggregate(self, text: str) -> AsyncIterator[PatternMatch]:
         """Aggregate text and process pattern pairs.
 
-        This method adds the new text to the buffer, processes any complete pattern
-        pairs, and returns processed text up to sentence boundaries if possible.
-        If there are incomplete patterns (start without matching end), it will
-        continue buffering text.
+        Processes the input text character-by-character, handles pattern pairs,
+        and uses the parent's lookahead logic for sentence detection when no
+        patterns are active.
 
         Args:
-            text: New text to add to the buffer.
+            text: Text to aggregate.
 
-        Returns:
-            Processed text up to a sentence boundary, or None if more
-            text is needed to form a complete sentence or pattern.
+        Yields:
+            PatternMatch objects as patterns complete or sentences are detected.
         """
-        # Add new text to buffer
-        self._text += text
+        # Process text character by character
+        for char in text:
+            self._text += char
 
-        # Process any complete patterns in the buffer
-        patterns, processed_text = await self._process_complete_patterns(self._text)
+            # Process any newly complete patterns in the buffer
+            # Only patterns that complete after _last_processed_position will trigger handlers
+            patterns, processed_text = await self._process_complete_patterns(
+                self._text, self._last_processed_position
+            )
 
-        self._text = processed_text
+            # Update the last processed position to prevent re-processing patterns
+            # This tracks where in the buffer we've already called handlers, so we
+            # only trigger handlers once when a pattern completes
+            self._last_processed_position = len(self._text)
 
-        if len(patterns) > 0:
-            if len(patterns) > 1:
-                logger.warning(
-                    f"Multiple patterns matched: {[p.type for p in patterns]}. Only the first pattern will be returned."
+            self._text = processed_text
+
+            if len(patterns) > 0:
+                if len(patterns) > 1:
+                    logger.warning(
+                        f"Multiple patterns matched: {[p.type for p in patterns]}. Only the first pattern will be returned."
+                    )
+                # If the pattern found is set to be aggregated, return it
+                action = self._patterns[patterns[0].type].get("action", MatchAction.REMOVE)
+                if action == MatchAction.AGGREGATE:
+                    self._text = ""
+                    yield patterns[0]
+                    continue
+
+            # Check if we have incomplete patterns
+            pattern_start = self._match_start_of_pattern(self._text)
+            if pattern_start is not None:
+                # If the start pattern is at the beginning or should not be separately aggregated, continue
+                if (
+                    pattern_start[0] == 0
+                    or pattern_start[1].get("action", MatchAction.REMOVE) != MatchAction.AGGREGATE
+                ):
+                    continue
+                # For AGGREGATE patterns: yield any text before the pattern starts
+                # This ensures text doesn't get stuck in the buffer waiting for sentence
+                # boundaries when a pattern begins (e.g., "Here is code <code>..." yields "Here is code")
+                result = self._text[: pattern_start[0]]
+                self._text = self._text[pattern_start[0] :]
+                yield PatternMatch(
+                    content=result.strip(), type=AggregationType.SENTENCE, full_match=result
                 )
-            # If the pattern found is set to be aggregated, return it
-            action = self._patterns[patterns[0].type].get("action", MatchAction.REMOVE)
-            if action == MatchAction.AGGREGATE:
-                self._text = ""
-                return patterns[0]
+                continue
 
-        # Check if we have incomplete patterns
-        pattern_start = self._match_start_of_pattern(self._text)
-        if pattern_start is not None:
-            # If the start pattern is at the beginning or should not be separately aggregated, return None
-            if (
-                pattern_start[0] == 0
-                or pattern_start[1].get("action", MatchAction.REMOVE) != MatchAction.AGGREGATE
-            ):
-                return None
-            # Otherwise, strip the text up to the start pattern and return it
-            result = self._text[: pattern_start[0]]
-            self._text = self._text[pattern_start[0] :]
-            return PatternMatch(
-                content=result.strip(), type=AggregationType.SENTENCE, full_match=result
-            )
-
-        # Find sentence boundary if no incomplete patterns
-        eos_marker = match_endofsentence(self._text)
-        if eos_marker:
-            # Extract text up to the sentence boundary
-            result = self._text[:eos_marker]
-            self._text = self._text[eos_marker:]
-            return PatternMatch(
-                content=result.strip(), type=AggregationType.SENTENCE, full_match=result
-            )
-
-        # No complete sentence found yet
-        return None
+            # Use parent's lookahead logic for sentence detection
+            aggregation = await super()._check_sentence_with_lookahead(char)
+            if aggregation:
+                # Convert to PatternMatch for consistency with return type
+                yield PatternMatch(
+                    content=aggregation.text, type=aggregation.type, full_match=aggregation.text
+                )
 
     async def handle_interruption(self):
-        """Handle interruptions by clearing the buffer.
+        """Handle interruptions by clearing the buffer and pattern state.
 
         Called when an interruption occurs in the processing pipeline,
         to reset the state and discard any partially aggregated text.
         """
-        self._text = ""
+        await super().handle_interruption()
+        self._last_processed_position = 0
+        # Pattern and handler state persists across interruptions
 
     async def reset(self):
         """Clear the internally aggregated text.
@@ -382,4 +401,6 @@ class PatternPairAggregator(BaseTextAggregator):
         Resets the aggregator to its initial state, discarding any
         buffered text and clearing pattern tracking state.
         """
-        self._text = ""
+        await super().reset()
+        self._last_processed_position = 0
+        # Pattern and handler state persists across resets

src/pipecat/utils/text/simple_text_aggregator.py

@@ -11,9 +11,9 @@ until it finds an end-of-sentence marker, making it suitable for basic TTS
 text processing scenarios.
 """
 
-from typing import Optional
+from typing import AsyncIterator, Optional
 
-from pipecat.utils.string import match_endofsentence
+from pipecat.utils.string import SENTENCE_ENDING_PUNCTUATION, match_endofsentence
 from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator
 
 
@@ -31,6 +31,7 @@ class SimpleTextAggregator(BaseTextAggregator):
         Creates an empty text buffer ready to begin accumulating text tokens.
         """
         self._text = ""
+        self._needs_lookahead: bool = False
 
     @property
     def text(self) -> Aggregation:
@@ -41,30 +42,87 @@ class SimpleTextAggregator(BaseTextAggregator):
         """
         return Aggregation(text=self._text.strip(), type=AggregationType.SENTENCE)
 
-    async def aggregate(self, text: str) -> Optional[Aggregation]:
-        """Aggregate text and return completed sentences.
+    async def aggregate(self, text: str) -> AsyncIterator[Aggregation]:
+        """Aggregate text and yield completed sentences.
 
-        Adds the new text to the buffer and checks for end-of-sentence markers.
-        When a sentence boundary is found, returns the completed sentence and
-        removes it from the buffer.
+        Processes the input text character-by-character. When sentence-ending
+        punctuation is detected, it waits for non-whitespace lookahead before
+        calling NLTK. This prevents false positives like "$29." being detected
+        as a sentence when it's actually "$29.95".
 
         Args:
-            text: New text to add to the aggregation buffer.
+            text: Text to aggregate.
+
+        Yields:
+            Complete sentences as Aggregation objects.
+        """
+        # Process text character by character
+        for char in text:
+            self._text += char
+
+            # Check for sentence with lookahead
+            result = await self._check_sentence_with_lookahead(char)
+            if result:
+                yield result
+
+    async def _check_sentence_with_lookahead(self, char: str) -> Optional[Aggregation]:
+        """Check for sentence boundaries using lookahead logic.
+
+        This method implements the core sentence detection logic with lookahead.
+        When sentence-ending punctuation is detected, it waits for the next
+        non-whitespace character before calling NLTK. This disambiguates cases
+        like "$29." (not a sentence) vs "$29. Next" (sentence ends at period).
+        Whitespace alone is not meaningful lookahead since it appears in both
+        cases. Instead, the first non-whitespace character after the punctuation
+        is used to confirm the sentence boundary.
+
+        Subclasses can call this via super() to reuse the lookahead behavior
+        while adding their own logic (e.g., tag handling, pattern matching).
+
+        Args:
+            char: The most recently added character (used for lookahead check).
 
         Returns:
-            A complete sentence if an end-of-sentence marker is found,
-            or None if more text is needed to complete a sentence.
+            Aggregation if sentence found, None otherwise.
         """
-        result: Optional[str] = None
+        # If we need lookahead, check if we now have non-whitespace
+        if self._needs_lookahead:
+            # Check if the new character is non-whitespace
+            if char.strip():
+                # We have meaningful lookahead, call NLTK
+                self._needs_lookahead = False
+                eos_marker = match_endofsentence(self._text)
 
-        self._text += text
+                if eos_marker:
+                    # NLTK confirmed a sentence - return it
+                    result = self._text[:eos_marker]
+                    self._text = self._text[eos_marker:]
+                    return Aggregation(text=result, type=AggregationType.SENTENCE)
+                # No sentence found - keep accumulating
+                return None
+            # Still whitespace, keep waiting
+            return None
 
-        eos_end_marker = match_endofsentence(self._text)
-        if eos_end_marker:
-            result = self._text[:eos_end_marker]
-            self._text = self._text[eos_end_marker:]
+        # Check if we just added sentence-ending punctuation
+        if self._text and self._text[-1] in SENTENCE_ENDING_PUNCTUATION:
+            # Mark that we need lookahead (don't call NLTK yet)
+            self._needs_lookahead = True
 
-        if result:
+        return None
+
+    async def flush(self) -> Optional[Aggregation]:
+        """Flush any remaining text in the buffer.
+
+        Returns any text remaining in the buffer. This is called at the end
+        of a stream to ensure all text is processed.
+
+        Returns:
+            Any remaining text as a sentence, or None if buffer is empty.
+        """
+        if self._text:
+            # Return whatever we have in the buffer
+            result = self._text
+            await self.reset()
             return Aggregation(text=result.strip(), type=AggregationType.SENTENCE)
         return None
 
@@ -75,6 +133,7 @@ class SimpleTextAggregator(BaseTextAggregator):
         discarding any partially accumulated text.
         """
         self._text = ""
+        self._needs_lookahead = False
 
     async def reset(self):
         """Clear the internally aggregated text.
@@ -83,3 +142,4 @@ class SimpleTextAggregator(BaseTextAggregator):
         any accumulated text content.
         """
         self._text = ""
+        self._needs_lookahead = False

src/pipecat/utils/text/skip_tags_aggregator.py

@@ -11,13 +11,14 @@ between specified start/end tag pairs, ensuring that tagged content is processed
 as a unit regardless of internal punctuation.
 """
 
-from typing import Optional, Sequence
+from typing import AsyncIterator, Optional, Sequence
 
-from pipecat.utils.string import StartEndTags, match_endofsentence, parse_start_end_tags
-from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator
+from pipecat.utils.string import StartEndTags, parse_start_end_tags
+from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType
+from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator
 
 
-class SkipTagsAggregator(BaseTextAggregator):
+class SkipTagsAggregator(SimpleTextAggregator):
     """Aggregator that prevents end of sentence matching between start/end tags.
 
     This aggregator buffers text until it finds an end of sentence or a start
@@ -37,67 +38,59 @@ class SkipTagsAggregator(BaseTextAggregator):
             tags: Sequence of StartEndTags objects defining the tag pairs
                   that should prevent sentence boundary detection.
         """
-        self._text = ""
+        super().__init__()
         self._tags = tags
         self._current_tag: Optional[StartEndTags] = None
         self._current_tag_index: int = 0
 
-    @property
-    def text(self) -> Aggregation:
-        """Get the currently buffered text.
-
-        Returns:
-            The current text buffer content that hasn't been processed yet.
-        """
-        return Aggregation(text=self._text.strip(), type=AggregationType.SENTENCE)
-
-    async def aggregate(self, text: str) -> Optional[Aggregation]:
+    async def aggregate(self, text: str) -> AsyncIterator[Aggregation]:
         """Aggregate text while respecting tag boundaries.
 
-        This method adds the new text to the buffer, processes any complete
-        pattern pairs, and returns processed text up to sentence boundaries if
-        possible. If there are incomplete patterns (start without matching
-        end), it will continue buffering text.
+        Processes the input text character-by-character, updates tag state, and
+        uses the parent's lookahead logic for sentence detection when not
+        inside tags.
 
         Args:
-            text: New text to add to the buffer.
+            text: Text to aggregate.
 
-        Returns:
-            An Aggregation object containing text up to a sentence boundary and
-            marked as SENTENCE type or None if more text is needed to complete a
-            sentence or close tags.
+        Yields:
+            Aggregation objects containing text up to a sentence boundary,
+            marked as SENTENCE type.
         """
-        # Add new text to buffer
-        self._text += text
+        # Process text character by character
+        for char in text:
+            self._text += char
 
-        (self._current_tag, self._current_tag_index) = parse_start_end_tags(
-            self._text, self._tags, self._current_tag, self._current_tag_index
-        )
+            # Update tag state
+            (self._current_tag, self._current_tag_index) = parse_start_end_tags(
+                self._text, self._tags, self._current_tag, self._current_tag_index
+            )
 
-        # Find sentence boundary if no incomplete patterns
-        if not self._current_tag:
-            eos_marker = match_endofsentence(self._text)
-            if eos_marker:
-                # Extract text up to the sentence boundary
-                result = self._text[:eos_marker]
-                self._text = self._text[eos_marker:]
-                return Aggregation(text=result.strip(), type=AggregationType.SENTENCE)
+            # If inside tags, don't check for sentences
+            if self._current_tag:
+                continue
 
-        # No complete sentence found yet
-        return None
+            # Otherwise, use parent's lookahead logic for sentence detection
+            result = await super()._check_sentence_with_lookahead(char)
+            if result:
+                yield result
 
     async def handle_interruption(self):
-        """Handle interruptions by clearing the buffer.
+        """Handle interruptions by clearing the buffer and tag state.
 
         Called when an interruption occurs in the processing pipeline,
         to reset the state and discard any partially aggregated text.
         """
-        self._text = ""
+        await super().handle_interruption()
+        self._current_tag = None
+        self._current_tag_index = 0
 
     async def reset(self):
-        """Clear the internally aggregated text.
+        """Clear the internally aggregated text and tag state.
 
         Resets the aggregator to its initial state, discarding any
         buffered text.
         """
-        self._text = ""
+        await super().reset()
+        self._current_tag = None
+        self._current_tag_index = 0

tests/test_ivr_navigation.py

@@ -10,6 +10,7 @@ from unittest.mock import AsyncMock
 from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.extensions.ivr.ivr_navigator import IVRProcessor
 from pipecat.frames.frames import (
+    LLMFullResponseEndFrame,
     LLMMessagesUpdateFrame,
     LLMTextFrame,
     OutputDTMFUrgentFrame,
@@ -334,10 +335,12 @@ class TestIVRNavigation(unittest.IsolatedAsyncioTestCase):
 
         frames_to_send = [
             LLMTextFrame(text="Hello, I'm trying to reach billing."),
+            LLMFullResponseEndFrame(),
         ]
 
         expected_down_frames = [
             LLMTextFrame,  # Should pass through unchanged
+            LLMFullResponseEndFrame,
         ]
 
         expected_up_frames = [

tests/test_pattern_pair_aggregator.py

@@ -38,14 +38,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
         self.aggregator.on_pattern_match("code_pattern", self.code_handler)
 
     async def test_pattern_match_and_removal(self):
-        # First part doesn't complete the pattern
-        result = await self.aggregator.aggregate("Hello <test>pattern")
-        self.assertIsNone(result)
-        self.assertEqual(self.aggregator.text.text, "Hello <test>pattern")
-        self.assertEqual(self.aggregator.text.type, "test_pattern")
-
-        # Second part completes the pattern and includes an exclamation point
-        result = await self.aggregator.aggregate(" content</test>!")
+        text = "Hello <test>pattern content</test>!"
+        results = [result async for result in self.aggregator.aggregate(text)]
 
         # Verify the handler was called with correct PatternMatch object
         self.test_handler.assert_called_once()
@@ -55,28 +49,37 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(call_args.full_match, "<test>pattern content</test>")
         self.assertEqual(call_args.text, "pattern content")
 
-        # The exclamation point should be treated as a sentence boundary,
-        # so the result should include just text up to and including "!"
-        self.assertEqual(result.text, "Hello !")
-        self.assertEqual(result.type, "sentence")
+        # No results yet (waiting for lookahead after "!")
+        self.assertEqual(len(results), 0)
 
-        # Next sentence should be processed separately. Spaces around the sentence
-        # should be stripped in the returned Aggregation.
-        result = await self.aggregator.aggregate(" This is another sentence.")
+        # Next sentence should provide the lookahead and trigger the previous sentence
+        async for result in self.aggregator.aggregate(" This is another sentence."):
+            results.append(result)
+
+        # First result should be "Hello !" triggered by the space lookahead
+        self.assertEqual(len(results), 1)
+        self.assertEqual(results[0].text, "Hello !")
+        self.assertEqual(results[0].type, "sentence")
+
+        # Now flush to get the remaining sentence
+        result = await self.aggregator.flush()
         self.assertEqual(result.text, "This is another sentence.")
 
         # Buffer should be empty after returning a complete sentence
         self.assertEqual(self.aggregator.text.text, "")
 
     async def test_pattern_match_and_aggregate(self):
-        # First part doesn't complete the pattern
-        result = await self.aggregator.aggregate("Here is code <code>pattern")
-        self.assertEqual(result.text, "Here is code")
-        self.assertEqual(self.aggregator.text.text, "<code>pattern")
-        self.assertEqual(self.aggregator.text.type, "code_pattern")
+        text = "Here is code <code>pattern content</code> This is another sentence."
 
-        # Second part completes the pattern and includes an exclamation point
-        result = await self.aggregator.aggregate(" content</code>")
+        results = [result async for result in self.aggregator.aggregate(text)]
+
+        # First result should be "Here is code" when pattern starts
+        self.assertEqual(results[0].text, "Here is code")
+        self.assertEqual(results[0].type, "sentence")
+
+        # Second result should be the code pattern content
+        self.assertEqual(results[1].text, "pattern content")
+        self.assertEqual(results[1].type, "code_pattern")
 
         # Verify the handler was called with correct PatternMatch object
         self.code_handler.assert_called_once()
@@ -85,11 +88,9 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(call_args.type, "code_pattern")
         self.assertEqual(call_args.full_match, "<code>pattern content</code>")
         self.assertEqual(call_args.text, "pattern content")
-        self.assertEqual(result.text, "pattern content")
-        self.assertEqual(result.type, "code_pattern")
 
-        # Next sentence should be processed separately
-        result = await self.aggregator.aggregate(" This is another sentence.")
+        # Last sentence needs flush (waiting for lookahead after ".")
+        result = await self.aggregator.flush()
         self.assertEqual(result.text, "This is another sentence.")
         self.assertEqual(result.type, "sentence")
 
@@ -97,11 +98,10 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(self.aggregator.text.text, "")
 
     async def test_incomplete_pattern(self):
-        # Add text with incomplete pattern
-        result = await self.aggregator.aggregate("Hello <test>pattern content")
-
+        text = "Hello <test>pattern content"
+        results = [result async for result in self.aggregator.aggregate(text)]
         # No complete pattern yet, so nothing should be returned
-        self.assertIsNone(result)
+        self.assertEqual(len(results), 0)
 
         # The handler should not be called yet
         self.test_handler.assert_not_called()
@@ -136,9 +136,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
         self.aggregator.on_pattern_match("voice", voice_handler)
         self.aggregator.on_pattern_match("emphasis", emphasis_handler)
 
-        # Test with multiple patterns in one text block
         text = "Hello <voice>female</voice> I am <em>very</em> excited to meet you!"
-        result = await self.aggregator.aggregate(text)
+        results = [result async for result in self.aggregator.aggregate(text)]
 
         # Both handlers should be called with correct data
         voice_handler.assert_called_once()
@@ -151,6 +150,10 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(emphasis_match.type, "emphasis")
         self.assertEqual(emphasis_match.text, "very")
 
+        # With lookahead, we need to flush to get the final sentence
+        self.assertEqual(len(results), 0)  # Waiting for lookahead after "!"
+
+        result = await self.aggregator.flush()
         # Voice pattern should be removed, emphasis pattern should remain
         self.assertEqual(result.text, "Hello  I am <em>very</em> excited to meet you!")
 
@@ -158,9 +161,9 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(self.aggregator.text.text, "")
 
     async def test_handle_interruption(self):
-        # Start with incomplete pattern
-        result = await self.aggregator.aggregate("Hello <test>pattern")
-        self.assertIsNone(result)
+        text = "Hello <test>pattern"
+        results = [result async for result in self.aggregator.aggregate(text)]
+        self.assertEqual(len(results), 0)
 
         # Simulate interruption
         await self.aggregator.handle_interruption()
@@ -172,20 +175,18 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
         self.test_handler.assert_not_called()
 
     async def test_pattern_across_sentences(self):
-        # Test pattern that spans multiple sentences
-        result = await self.aggregator.aggregate("Hello <test>This is sentence one.")
-
-        # First sentence contains start of pattern but no end, so no complete pattern yet
-        self.assertIsNone(result)
-
-        # Add second part with pattern end
-        result = await self.aggregator.aggregate(" This is sentence two.</test> Final sentence.")
+        text = "Hello <test>This is sentence one. This is sentence two.</test> Final sentence."
+        results = [result async for result in self.aggregator.aggregate(text)]
 
         # Handler should be called with entire content
         self.test_handler.assert_called_once()
         call_args = self.test_handler.call_args[0][0]
         self.assertEqual(call_args.text, "This is sentence one. This is sentence two.")
 
+        # With lookahead, we need to flush to get the final sentence
+        self.assertEqual(len(results), 0)  # Waiting for lookahead after "."
+
+        result = await self.aggregator.flush()
         # Pattern should be removed, resulting in text with sentences merged
         self.assertEqual(result.text, "Hello  Final sentence.")
 

tests/test_simple_text_aggregator.py

@@ -14,22 +14,112 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase):
         self.aggregator = SimpleTextAggregator()
 
     async def test_reset_aggregations(self):
-        assert await self.aggregator.aggregate("Hello ") == None
+        text = "Hello "
+        results = [agg async for agg in self.aggregator.aggregate(text)]
+
+        # No complete sentences yet
+        assert len(results) == 0
         assert self.aggregator.text.text == "Hello"
         await self.aggregator.reset()
         assert self.aggregator.text.text == ""
 
     async def test_simple_sentence(self):
-        assert await self.aggregator.aggregate("Hello ") == None
-        aggregate = await self.aggregator.aggregate("Pipecat!")
+        text = "Hello Pipecat!"
+        results = [agg async for agg in self.aggregator.aggregate(text)]
+
+        # No complete sentences yet (waiting for lookahead after "!")
+        assert len(results) == 0
+
+        # Flush to get the pending sentence
+        aggregate = await self.aggregator.flush()
         assert aggregate.text == "Hello Pipecat!"
         assert aggregate.type == "sentence"
         assert self.aggregator.text.text == ""
 
     async def test_multiple_sentences(self):
-        aggregate = await self.aggregator.aggregate("Hello Pipecat! How are ")
-        assert aggregate.text == "Hello Pipecat!"
-        # Aggregators should strip leading/trailing spaces when returning text
-        assert self.aggregator.text.text == "How are"
-        aggregate = await self.aggregator.aggregate("you?")
-        assert aggregate.text == "How are you?"
+        text = "Hello Pipecat! How are you?"
+        results = [agg async for agg in self.aggregator.aggregate(text)]
+
+        # First sentence should be complete (lookahead from "H" confirmed it)
+        assert len(results) == 1
+        assert results[0].text == "Hello Pipecat!"
+
+        # Flush to get the pending sentence
+        result = await self.aggregator.flush()
+        assert result.text == "How are you?"
+
+    async def test_lookahead_decimal_number(self):
+        """Test that $29.95 is not split at $29."""
+        text = "Ask me for only $29.95/month."
+        results = [agg async for agg in self.aggregator.aggregate(text)]
+
+        # No complete sentences yet (waiting for lookahead after final ".")
+        assert len(results) == 0
+
+        # Can use flush() to get the pending sentence at end of stream
+        result = await self.aggregator.flush()
+        assert result.text == "Ask me for only $29.95/month."
+
+    async def test_lookahead_abbreviation(self):
+        """Test that Mr. Smith is not split at Mr."""
+        text = "Hello Mr. Smith."
+        results = [agg async for agg in self.aggregator.aggregate(text)]
+
+        # No complete sentences yet (waiting for lookahead after final ".")
+        assert len(results) == 0
+
+        # Can use flush() to get the pending sentence at end of stream
+        result = await self.aggregator.flush()
+        assert result.text == "Hello Mr. Smith."
+
+    async def test_lookahead_actual_sentence_end(self):
+        """Test that a real sentence end is detected after lookahead."""
+        text = "Hello world. Next sentence"
+        results = [agg async for agg in self.aggregator.aggregate(text)]
+
+        # First sentence should be complete (lookahead from "N" confirmed it)
+        assert len(results) == 1
+        assert results[0].text == "Hello world."
+
+    async def test_flush_pending_sentence(self):
+        """Test that flush() returns pending sentence waiting for lookahead."""
+        text = "Hello world."
+        results = [agg async for agg in self.aggregator.aggregate(text)]
+
+        # No complete sentences yet (waiting for lookahead)
+        assert len(results) == 0
+
+        # Call flush to get it
+        result = await self.aggregator.flush()
+        assert result is not None
+        assert result.text == "Hello world."
+        # Flush again should return None
+        assert await self.aggregator.flush() == None
+
+    async def test_flush_with_no_pending(self):
+        """Test that flush() returns any remaining text in buffer."""
+        text = "Hello"
+        results = [agg async for agg in self.aggregator.aggregate(text)]
+
+        # No complete sentences
+        assert len(results) == 0
+
+        result = await self.aggregator.flush()
+        # flush() now returns any remaining text, not just pending lookahead
+        assert result is not None
+        assert result.text == "Hello"
+        # Buffer should be empty after flush
+        assert self.aggregator.text.text == ""
+
+    async def test_flush_after_lookahead_confirmed(self):
+        """Test flush after lookahead has already confirmed sentence."""
+        text = "Hello. W"
+        results = [agg async for agg in self.aggregator.aggregate(text)]
+
+        # First sentence should be complete (lookahead from "W" confirmed it)
+        assert len(results) == 1
+        assert results[0].text == "Hello."
+
+        # flush() returns any remaining text (the "W" in this case)
+        result = await self.aggregator.flush()
+        assert result.text == "W"

tests/test_skip_tags_aggregator.py

@@ -17,7 +17,14 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase):
         await self.aggregator.reset()
 
         # No tags involved, aggregate at end of sentence.
-        result = await self.aggregator.aggregate("Hello Pipecat!")
+        text = "Hello Pipecat!"
+        results = [agg async for agg in self.aggregator.aggregate(text)]
+
+        # Should still be waiting for lookahead after "!"
+        self.assertEqual(len(results), 0)
+
+        # Flush to get the pending sentence
+        result = await self.aggregator.flush()
         self.assertEqual(result.text, "Hello Pipecat!")
         self.assertEqual(result.type, "sentence")
         self.assertEqual(self.aggregator.text.text, "")
@@ -26,7 +33,14 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase):
         await self.aggregator.reset()
 
         # Tags involved, avoid aggregation during tags.
-        result = await self.aggregator.aggregate("My email is <spell>foo@pipecat.ai</spell>.")
+        text = "My email is <spell>foo@pipecat.ai</spell>."
+        results = [agg async for agg in self.aggregator.aggregate(text)]
+
+        # Should still be waiting for lookahead after "."
+        self.assertEqual(len(results), 0)
+
+        # Flush to get the pending sentence
+        result = await self.aggregator.flush()
         self.assertEqual(result.text, "My email is <spell>foo@pipecat.ai</spell>.")
         self.assertEqual(result.type, "sentence")
         self.assertEqual(self.aggregator.text.text, "")
@@ -34,25 +48,17 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase):
     async def test_streaming_tags(self):
         await self.aggregator.reset()
 
-        # Tags involved, stream small chunk of texts.
-        result = await self.aggregator.aggregate("My email is <sp")
-        self.assertIsNone(result)
-        self.assertEqual(self.aggregator.text.text, "My email is <sp")
+        # Tags involved
+        text = "My email is <spell>foo.bar@pipecat.ai</spell>."
+        results = [agg async for agg in self.aggregator.aggregate(text)]
 
-        result = await self.aggregator.aggregate("ell>foo.")
-        self.assertIsNone(result)
-        self.assertEqual(self.aggregator.text.text, "My email is <spell>foo.")
-
-        result = await self.aggregator.aggregate("bar@pipecat.")
-        self.assertIsNone(result)
-        self.assertEqual(self.aggregator.text.text, "My email is <spell>foo.bar@pipecat.")
-
-        result = await self.aggregator.aggregate("ai</spe")
-        self.assertIsNone(result)
-        self.assertEqual(self.aggregator.text.text, "My email is <spell>foo.bar@pipecat.ai</spe")
+        # Should still be waiting for lookahead after "."
+        self.assertEqual(len(results), 0)
+        self.assertEqual(self.aggregator.text.text, text)
         self.assertEqual(self.aggregator.text.type, "sentence")
 
-        result = await self.aggregator.aggregate("ll>.")
-        self.assertEqual(result.text, "My email is <spell>foo.bar@pipecat.ai</spell>.")
+        # Flush to get the pending sentence
+        result = await self.aggregator.flush()
+        self.assertEqual(result.text, text)
         self.assertEqual(self.aggregator.text.text, "")
         self.assertEqual(self.aggregator.text.type, "sentence")

uv.lock

@@ -1,5 +1,5 @@
 version = 1
-revision = 2
+revision = 3
 requires-python = ">=3.10"
 resolution-markers = [
     "python_full_version >= '3.13'",
@@ -4710,7 +4710,6 @@ requires-dist = [
     { name = "numba", specifier = "==0.61.2" },
     { name = "numpy", specifier = ">=1.26.4,<3" },
     { name = "nvidia-riva-client", marker = "extra == 'nvidia'", specifier = "~=2.21.1" },
-    { name = "nvidia-riva-client", marker = "extra == 'riva'", specifier = "~=2.21.1" },
     { name = "onnxruntime", marker = "extra == 'local-smart-turn-v3'", specifier = ">=1.20.1,<2" },
     { name = "onnxruntime", marker = "extra == 'silero'", specifier = ">=1.20.1,<2" },
     { name = "openai", specifier = ">=1.74.0,<3" },
@@ -4721,6 +4720,7 @@ requires-dist = [
     { name = "opentelemetry-sdk", marker = "extra == 'tracing'", specifier = ">=1.33.0" },
     { name = "ormsgpack", marker = "extra == 'fish'", specifier = "~=1.7.0" },
     { name = "pillow", specifier = ">=11.1.0,<12" },
+    { name = "pipecat-ai", extras = ["nvidia"], marker = "extra == 'riva'" },
     { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'assemblyai'" },
     { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'asyncai'" },
     { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'aws'" },
@@ -4771,7 +4771,7 @@ requires-dist = [
     { name = "wait-for2", marker = "python_full_version < '3.12'", specifier = ">=0.4.1" },
     { name = "websockets", marker = "extra == 'websockets-base'", specifier = ">=13.1,<16.0" },
 ]
-provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "daily", "deepgram", "deepseek", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "hume", "inworld", "koala", "krisp", "langchain", "livekit", "lmnt", "local", "local-smart-turn", "local-smart-turn-v3", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "neuphonic", "nim", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "remote-smart-turn", "rime", "riva", "nvidia", "runner", "sagemaker", "sambanova", "sarvam", "sentry", "silero", "simli", "soniox", "soundfile", "speechmatics", "strands", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "websockets-base", "whisper"]
+provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "daily", "deepgram", "deepseek", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "hume", "inworld", "koala", "krisp", "langchain", "livekit", "lmnt", "local", "local-smart-turn", "local-smart-turn-v3", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "neuphonic", "noisereduce", "nvidia", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "remote-smart-turn", "rime", "riva", "runner", "sagemaker", "sambanova", "sarvam", "sentry", "silero", "simli", "soniox", "soundfile", "speechmatics", "strands", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "websockets-base", "whisper"]
 
 [package.metadata.requires-dev]
 dev = [