Merge pull request #1841 from pipecat-ai/aleix/base-text-aggregator-async

make BaseTextAggregator and BaseTextFilter functions async

CHANGELOG.md

@@ -68,6 +68,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- `BaseTextFilter` methods `filter()`, `update_settings()`,
+  `handle_interruption()` and `reset_interruption()` are now async.
+
+- `BaseTextAggregator` methods `aggregate()`, `handle_interruption()` and
+  `reset()` are now async.
+
 - The API version for `CartesiaTTSService` and `CartesiaHttpTTSService` has
   been updated. Also, the `cartesia` dependency has been updated to 2.x.
 

examples/foundational/35-pattern-pair-voice-switching.py

@@ -99,21 +99,14 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
     )
 
     # Register handler for voice switching
-    def on_voice_tag(match: PatternMatch):
+    async def on_voice_tag(match: PatternMatch):
         voice_name = match.content.strip().lower()
         if voice_name in VOICE_IDS:
-            voice_id = VOICE_IDS[voice_name]
+            # First flush any existing audio to finish the current context
-
+            await tts.flush_audio()
-            # Create task to reset the TTS context after voice change
+            # Then set the new voice
-            async def change_voice():
+            tts.set_voice(VOICE_IDS[voice_name])
-                # First flush any existing audio to finish the current context
+            logger.info(f"Switched to {voice_name} voice")
-                await tts.flush_audio()
-                # Then set the new voice
-                tts.set_voice(voice_id)
-                logger.info(f"Switched to {voice_name} voice")
-
-            # Schedule the voice change task
-            asyncio.create_task(change_voice())
         else:
             logger.warning(f"Unknown voice: {voice_name}")
 

examples/word-wrangler-gemini-live/server/bot_phone_local.py

@@ -188,7 +188,7 @@ class HostResponseTextFilter(BaseTextFilter):
         # No settings to update for this filter
         pass
 
-    def filter(self, text: str) -> str:
+    async def filter(self, text: str) -> str:
         # Remove case and whitespace for comparison
         clean_text = text.strip().upper()
 
@@ -198,10 +198,10 @@ class HostResponseTextFilter(BaseTextFilter):
 
         return text
 
-    def handle_interruption(self):
+    async def handle_interruption(self):
         self._interrupted = True
 
-    def reset_interruption(self):
+    async def reset_interruption(self):
         self._interrupted = False
 
 

examples/word-wrangler-gemini-live/server/bot_phone_twilio.py

@@ -178,7 +178,7 @@ class HostResponseTextFilter(BaseTextFilter):
         # No settings to update for this filter
         pass
 
-    def filter(self, text: str) -> str:
+    async def filter(self, text: str) -> str:
         # Remove case and whitespace for comparison
         clean_text = text.strip().upper()
 
@@ -188,10 +188,10 @@ class HostResponseTextFilter(BaseTextFilter):
 
         return text
 
-    def handle_interruption(self):
+    async def handle_interruption(self):
         self._interrupted = True
 
-    def reset_interruption(self):
+    async def reset_interruption(self):
         self._interrupted = False
 
 

src/pipecat/services/tts_service.py

@@ -157,7 +157,7 @@ class TTSService(AIService):
                 self.set_voice(value)
             elif key == "text_filter":
                 for filter in self._text_filters:
-                    filter.update_settings(value)
+                    await filter.update_settings(value)
             else:
                 logger.warning(f"Unknown setting for TTS service: {key}")
 
@@ -183,7 +183,7 @@ class TTSService(AIService):
             await self._maybe_pause_frame_processing()
 
             sentence = self._text_aggregator.text
-            self._text_aggregator.reset()
+            await self._text_aggregator.reset()
             self._processing_text = False
             await self._push_tts_frames(sentence)
             if isinstance(frame, LLMFullResponseEndFrame):
@@ -234,9 +234,9 @@ class TTSService(AIService):
 
     async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
         self._processing_text = False
-        self._text_aggregator.handle_interruption()
+        await self._text_aggregator.handle_interruption()
         for filter in self._text_filters:
-            filter.handle_interruption()
+            await filter.handle_interruption()
 
     async def _maybe_pause_frame_processing(self):
         if self._processing_text and self._pause_frame_processing:
@@ -251,7 +251,7 @@ class TTSService(AIService):
         if not self._aggregate_sentences:
             text = frame.text
         else:
-            text = self._text_aggregator.aggregate(frame.text)
+            text = await self._text_aggregator.aggregate(frame.text)
 
         if text:
             await self._push_tts_frames(text)
@@ -274,8 +274,8 @@ class TTSService(AIService):
 
         # Process all filter.
         for filter in self._text_filters:
-            filter.reset_interruption()
+            await filter.reset_interruption()
-            text = filter.filter(text)
+            text = await filter.filter(text)
 
         if text:
             await self.process_generator(self.run_tts(text))

src/pipecat/utils/text/base_text_aggregator.py

@@ -25,7 +25,7 @@ class BaseTextAggregator(ABC):
         pass
 
     @abstractmethod
-    def aggregate(self, text: str) -> Optional[str]:
+    async 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
@@ -43,7 +43,7 @@ class BaseTextAggregator(ABC):
         pass
 
     @abstractmethod
-    def handle_interruption(self):
+    async 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.
@@ -52,6 +52,6 @@ class BaseTextAggregator(ABC):
         pass
 
     @abstractmethod
-    def reset(self):
+    async def reset(self):
         """Clears the internally aggregated text."""
         pass

src/pipecat/utils/text/base_text_filter.py

@@ -10,17 +10,17 @@ from typing import Any, Mapping
 
 class BaseTextFilter(ABC):
     @abstractmethod
-    def update_settings(self, settings: Mapping[str, Any]):
+    async def update_settings(self, settings: Mapping[str, Any]):
         pass
 
     @abstractmethod
-    def filter(self, text: str) -> str:
+    async def filter(self, text: str) -> str:
         pass
 
     @abstractmethod
-    def handle_interruption(self):
+    async def handle_interruption(self):
         pass
 
     @abstractmethod
-    def reset_interruption(self):
+    async def reset_interruption(self):
         pass

src/pipecat/utils/text/markdown_text_filter.py

@@ -33,12 +33,12 @@ class MarkdownTextFilter(BaseTextFilter):
         self._in_table = False
         self._interrupted = False
 
-    def update_settings(self, settings: Mapping[str, Any]):
+    async def update_settings(self, settings: Mapping[str, Any]):
         for key, value in settings.items():
             if hasattr(self._settings, key):
                 setattr(self._settings, key, value)
 
-    def filter(self, text: str) -> str:
+    async def filter(self, text: str) -> str:
         if self._settings.enable_text_filter:
             # Remove newlines and replace with a space only when there's no text before or after
             filtered_text = re.sub(r"^\s*\n", " ", text, flags=re.MULTILINE)
@@ -104,12 +104,12 @@ class MarkdownTextFilter(BaseTextFilter):
         else:
             return text
 
-    def handle_interruption(self):
+    async def handle_interruption(self):
         self._interrupted = True
         self._in_code_block = False
         self._in_table = False
 
-    def reset_interruption(self):
+    async def reset_interruption(self):
         self._interrupted = False
 
     #

src/pipecat/utils/text/pattern_pair_aggregator.py

@@ -5,7 +5,7 @@
 #
 
 import re
-from typing import Callable, Optional, Tuple
+from typing import Awaitable, Callable, Optional, Tuple
 
 from loguru import logger
 
@@ -106,7 +106,7 @@ class PatternPairAggregator(BaseTextAggregator):
         return self
 
     def on_pattern_match(
-        self, pattern_id: str, handler: Callable[[PatternMatch], None]
+        self, pattern_id: str, handler: Callable[[PatternMatch], Awaitable[None]]
     ) -> "PatternPairAggregator":
         """Register a handler for when a pattern pair is matched.
 
@@ -124,7 +124,7 @@ class PatternPairAggregator(BaseTextAggregator):
         self._handlers[pattern_id] = handler
         return self
 
-    def _process_complete_patterns(self, text: str) -> Tuple[str, bool]:
+    async def _process_complete_patterns(self, text: str) -> Tuple[str, bool]:
         """Process all complete pattern pairs in the text.
 
         Searches for all complete pattern pairs in the text, calls the
@@ -167,7 +167,7 @@ class PatternPairAggregator(BaseTextAggregator):
                 # Call the appropriate handler if registered
                 if pattern_id in self._handlers:
                     try:
-                        self._handlers[pattern_id](pattern_match)
+                        await self._handlers[pattern_id](pattern_match)
                     except Exception as e:
                         logger.error(f"Error in pattern handler for {pattern_id}: {e}")
 
@@ -204,7 +204,7 @@ class PatternPairAggregator(BaseTextAggregator):
 
         return False
 
-    def aggregate(self, text: str) -> Optional[str]:
+    async def aggregate(self, text: str) -> Optional[str]:
         """Aggregate text and process pattern pairs.
 
         This method adds the new text to the buffer, processes any complete pattern
@@ -223,7 +223,7 @@ class PatternPairAggregator(BaseTextAggregator):
         self._text += text
 
         # Process any complete patterns in the buffer
-        processed_text, modified = self._process_complete_patterns(self._text)
+        processed_text, modified = await self._process_complete_patterns(self._text)
 
         # Only update the buffer if modifications were made
         if modified:
@@ -245,7 +245,7 @@ class PatternPairAggregator(BaseTextAggregator):
         # No complete sentence found yet
         return None
 
-    def handle_interruption(self):
+    async def handle_interruption(self):
         """Handle interruptions by clearing the buffer.
 
         Called when an interruption occurs in the processing pipeline,
@@ -253,7 +253,7 @@ class PatternPairAggregator(BaseTextAggregator):
         """
         self._text = ""
 
-    def reset(self):
+    async def reset(self):
         """Clear the internally aggregated text.
 
         Resets the aggregator to its initial state, discarding any

src/pipecat/utils/text/simple_text_aggregator.py

@@ -23,7 +23,7 @@ class SimpleTextAggregator(BaseTextAggregator):
     def text(self) -> str:
         return self._text
 
-    def aggregate(self, text: str) -> Optional[str]:
+    async def aggregate(self, text: str) -> Optional[str]:
         result: Optional[str] = None
 
         self._text += text
@@ -35,8 +35,8 @@ class SimpleTextAggregator(BaseTextAggregator):
 
         return result
 
-    def handle_interruption(self):
+    async def handle_interruption(self):
         self._text = ""
 
-    def reset(self):
+    async def reset(self):
         self._text = ""

src/pipecat/utils/text/skip_tags_aggregator.py

@@ -43,7 +43,7 @@ class SkipTagsAggregator(BaseTextAggregator):
         """
         return self._text
 
-    def aggregate(self, text: str) -> Optional[str]:
+    async def aggregate(self, text: str) -> Optional[str]:
         """Aggregate text and process pattern pairs.
 
         This method adds the new text to the buffer, processes any complete pattern
@@ -77,7 +77,7 @@ class SkipTagsAggregator(BaseTextAggregator):
         # No complete sentence found yet
         return None
 
-    def handle_interruption(self):
+    async def handle_interruption(self):
         """Handle interruptions by clearing the buffer.
 
         Called when an interruption occurs in the processing pipeline,
@@ -85,7 +85,7 @@ class SkipTagsAggregator(BaseTextAggregator):
         """
         self._text = ""
 
-    def reset(self):
+    async def reset(self):
         """Clear the internally aggregated text.
 
         Resets the aggregator to its initial state, discarding any

tests/test_markdown_text_filter.py

@@ -30,7 +30,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
             Some inline code here
         """
 
-        result = self.filter.filter(input_text)
+        result = await self.filter.filter(input_text)
         self.assertEqual(result.strip(), expected_text.strip())
 
     async def test_space_preservation(self):
@@ -45,7 +45,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
         ]
 
         for text in input_text:
-            result = self.filter.filter(text)
+            result = await self.filter.filter(text)
             self.assertEqual(
                 len(result), len(text), f"Space preservation failed for: '{text}'\nGot: '{result}'"
             )
@@ -71,7 +71,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
         }
 
         for input_text, expected in test_cases.items():
-            result = self.filter.filter(input_text)
+            result = await self.filter.filter(input_text)
             self.assertEqual(
                 result,
                 expected,
@@ -88,7 +88,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
         2. Second item
         3. Third item with bold"""
 
-        result = self.filter.filter(input_text)
+        result = await self.filter.filter(input_text)
         self.assertEqual(
             result.strip(),
             expected.strip(),
@@ -106,7 +106,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
         }
 
         for input_text, expected in test_cases.items():
-            result = self.filter.filter(input_text)
+            result = await self.filter.filter(input_text)
             self.assertEqual(result, expected, f"HTML entity conversion failed for: '{input_text}'")
 
     async def test_asterisk_removal(self):
@@ -120,7 +120,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
         }
 
         for input_text, expected in test_cases.items():
-            result = self.filter.filter(input_text)
+            result = await self.filter.filter(input_text)
             self.assertEqual(result, expected, f"Asterisk removal failed for: '{input_text}'")
 
     async def test_newline_handling(self):
@@ -132,7 +132,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
         }
 
         for input_text, expected in test_cases.items():
-            result = self.filter.filter(input_text)
+            result = await self.filter.filter(input_text)
             self.assertEqual(
                 result, expected, f"Newline handling failed for:\n{input_text}\nGot:\n{result}"
             )
@@ -148,7 +148,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
         }
 
         for input_text, expected in test_cases.items():
-            result = self.filter.filter(input_text)
+            result = await self.filter.filter(input_text)
             self.assertEqual(
                 result,
                 expected,
@@ -166,7 +166,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
         }
 
         for input_text, expected in test_cases.items():
-            result = self.filter.filter(input_text)
+            result = await self.filter.filter(input_text)
             self.assertEqual(result, expected, f"Inline code handling failed for: '{input_text}'")
 
     async def test_simple_table_removal(self):
@@ -177,7 +177,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
 
         expected = ""
 
-        result = filter.filter(input_text)
+        result = await filter.filter(input_text)
         self.assertEqual(
             result.strip(),
             expected.strip(),
@@ -198,15 +198,15 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
         # Test with text filtering disabled
         text_with_markdown = "**bold** and *italic* with `code`"
         self.assertEqual(
-            filter.filter(text_with_markdown),
+            await filter.filter(text_with_markdown),
             text_with_markdown,
             "Disabled filter should not modify text",
         )
 
         # Enable just text filtering
-        filter.update_settings({"enable_text_filter": True})
+        await filter.update_settings({"enable_text_filter": True})
         self.assertEqual(
-            filter.filter(text_with_markdown),
+            await filter.filter(text_with_markdown),
             "bold and italic with code",
             "Enabled filter should remove markdown",
         )
@@ -217,14 +217,18 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
 
         # Initial state - formatting should be removed
         input_text = "**bold** and *italic*"
-        self.assertEqual(filter.filter(input_text), "bold and italic")
+        self.assertEqual(await filter.filter(input_text), "bold and italic")
 
         # Disable text filtering
-        filter.update_settings({"enable_text_filter": False})
+        await filter.update_settings({"enable_text_filter": False})
-        self.assertEqual(filter.filter(input_text), input_text, "Text filtering should be disabled")
+        self.assertEqual(
+            await filter.filter(input_text), input_text, "Text filtering should be disabled"
+        )
 
         # Re-enable text filtering
-        filter.update_settings({"enable_text_filter": True})
+        await filter.update_settings({"enable_text_filter": True})
         self.assertEqual(
-            filter.filter(input_text), "bold and italic", "Text filtering should be re-enabled"
+            await filter.filter(input_text),
+            "bold and italic",
+            "Text filtering should be re-enabled",
         )

tests/test_pattern_pair_aggregator.py

@@ -5,7 +5,7 @@
 #
 
 import unittest
-from unittest.mock import Mock
+from unittest.mock import AsyncMock
 
 from pipecat.utils.text.pattern_pair_aggregator import PatternMatch, PatternPairAggregator
 
@@ -13,7 +13,7 @@ from pipecat.utils.text.pattern_pair_aggregator import PatternMatch, PatternPair
 class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
     def setUp(self):
         self.aggregator = PatternPairAggregator()
-        self.test_handler = Mock()
+        self.test_handler = AsyncMock()
 
         # Add a test pattern
         self.aggregator.add_pattern_pair(
@@ -28,12 +28,12 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
 
     async def test_pattern_match_and_removal(self):
         # First part doesn't complete the pattern
-        result = self.aggregator.aggregate("Hello <test>pattern")
+        result = await self.aggregator.aggregate("Hello <test>pattern")
         self.assertIsNone(result)
         self.assertEqual(self.aggregator.text, "Hello <test>pattern")
 
         # Second part completes the pattern and includes an exclamation point
-        result = self.aggregator.aggregate(" content</test>!")
+        result = await self.aggregator.aggregate(" content</test>!")
 
         # Verify the handler was called with correct PatternMatch object
         self.test_handler.assert_called_once()
@@ -48,7 +48,7 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(result, "Hello !")
 
         # Next sentence should be processed separately
-        result = self.aggregator.aggregate(" This is another sentence.")
+        result = await self.aggregator.aggregate(" This is another sentence.")
         self.assertEqual(result, " This is another sentence.")
 
         # Buffer should be empty after returning a complete sentence
@@ -56,7 +56,7 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
 
     async def test_incomplete_pattern(self):
         # Add text with incomplete pattern
-        result = self.aggregator.aggregate("Hello <test>pattern content")
+        result = await self.aggregator.aggregate("Hello <test>pattern content")
 
         # No complete pattern yet, so nothing should be returned
         self.assertIsNone(result)
@@ -68,13 +68,13 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(self.aggregator.text, "Hello <test>pattern content")
 
         # Reset and confirm buffer is cleared
-        self.aggregator.reset()
+        await self.aggregator.reset()
         self.assertEqual(self.aggregator.text, "")
 
     async def test_multiple_patterns(self):
         # Set up multiple patterns and handlers
-        voice_handler = Mock()
+        voice_handler = AsyncMock()
-        emphasis_handler = Mock()
+        emphasis_handler = AsyncMock()
 
         self.aggregator.add_pattern_pair(
             pattern_id="voice", start_pattern="<voice>", end_pattern="</voice>", remove_match=True
@@ -92,7 +92,7 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
 
         # Test with multiple patterns in one text block
         text = "Hello <voice>female</voice> I am <em>very</em> excited to meet you!"
-        result = self.aggregator.aggregate(text)
+        result = await self.aggregator.aggregate(text)
 
         # Both handlers should be called with correct data
         voice_handler.assert_called_once()
@@ -113,11 +113,11 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
 
     async def test_handle_interruption(self):
         # Start with incomplete pattern
-        result = self.aggregator.aggregate("Hello <test>pattern")
+        result = await self.aggregator.aggregate("Hello <test>pattern")
         self.assertIsNone(result)
 
         # Simulate interruption
-        self.aggregator.handle_interruption()
+        await self.aggregator.handle_interruption()
 
         # Buffer should be cleared
         self.assertEqual(self.aggregator.text, "")
@@ -127,13 +127,13 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
 
     async def test_pattern_across_sentences(self):
         # Test pattern that spans multiple sentences
-        result = self.aggregator.aggregate("Hello <test>This is sentence one.")
+        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 = self.aggregator.aggregate(" This is sentence two.</test> Final sentence.")
+        result = await self.aggregator.aggregate(" This is sentence two.</test> Final sentence.")
 
         # Handler should be called with entire content
         self.test_handler.assert_called_once()

tests/test_simple_text_aggregator.py

@@ -14,16 +14,16 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase):
         self.aggregator = SimpleTextAggregator()
 
     async def test_reset_aggregations(self):
-        assert self.aggregator.aggregate("Hello ") == None
+        assert await self.aggregator.aggregate("Hello ") == None
         assert self.aggregator.text == "Hello "
-        self.aggregator.reset()
+        await self.aggregator.reset()
         assert self.aggregator.text == ""
 
     async def test_simple_sentence(self):
-        assert self.aggregator.aggregate("Hello ") == None
+        assert await self.aggregator.aggregate("Hello ") == None
-        assert self.aggregator.aggregate("Pipecat!") == "Hello Pipecat!"
+        assert await 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 await self.aggregator.aggregate("Hello Pipecat! How are ") == "Hello Pipecat!"
-        assert self.aggregator.aggregate("you?") == " How are you?"
+        assert await self.aggregator.aggregate("you?") == " How are you?"

tests/test_skip_tags_aggregator.py

@@ -14,41 +14,41 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase):
         self.aggregator = SkipTagsAggregator([("<spell>", "</spell>")])
 
     async def test_no_tags(self):
-        self.aggregator.reset()
+        await self.aggregator.reset()
 
         # No tags involved, aggregate at end of sentence.
-        result = self.aggregator.aggregate("Hello Pipecat!")
+        result = await self.aggregator.aggregate("Hello Pipecat!")
         self.assertEqual(result, "Hello Pipecat!")
         self.assertEqual(self.aggregator.text, "")
 
     async def test_basic_tags(self):
-        self.aggregator.reset()
+        await self.aggregator.reset()
 
         # Tags involved, avoid aggregation during tags.
-        result = self.aggregator.aggregate("My email is <spell>foo@pipecat.ai</spell>.")
+        result = await self.aggregator.aggregate("My email is <spell>foo@pipecat.ai</spell>.")
         self.assertEqual(result, "My email is <spell>foo@pipecat.ai</spell>.")
         self.assertEqual(self.aggregator.text, "")
 
     async def test_streaming_tags(self):
-        self.aggregator.reset()
+        await self.aggregator.reset()
 
         # Tags involved, stream small chunk of texts.
-        result = self.aggregator.aggregate("My email is <sp")
+        result = await self.aggregator.aggregate("My email is <sp")
         self.assertIsNone(result)
         self.assertEqual(self.aggregator.text, "My email is <sp")
 
-        result = self.aggregator.aggregate("ell>foo.")
+        result = await self.aggregator.aggregate("ell>foo.")
         self.assertIsNone(result)
         self.assertEqual(self.aggregator.text, "My email is <spell>foo.")
 
-        result = self.aggregator.aggregate("bar@pipecat.")
+        result = await self.aggregator.aggregate("bar@pipecat.")
         self.assertIsNone(result)
         self.assertEqual(self.aggregator.text, "My email is <spell>foo.bar@pipecat.")
 
-        result = self.aggregator.aggregate("ai</spe")
+        result = await self.aggregator.aggregate("ai</spe")
         self.assertIsNone(result)
         self.assertEqual(self.aggregator.text, "My email is <spell>foo.bar@pipecat.ai</spe")
 
-        result = self.aggregator.aggregate("ll>.")
+        result = await self.aggregator.aggregate("ll>.")
         self.assertEqual(result, "My email is <spell>foo.bar@pipecat.ai</spell>.")
         self.assertEqual(self.aggregator.text, "")