Merge pull request #571 from pipecat-ai/mb/add-code-filtering

Add code and table filtering option to MarkdownTextFilter

CHANGELOG.md

@@ -5,6 +5,14 @@ All notable changes to **Pipecat** will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- Added new input params to the `MarkdownTextFilter` utility. You can set
+  `filter_code` to filter code from text and `filter_tables` to filter tables
+  from text.
+
 ## [0.0.43] - 2024-10-10
 
 ### Added

src/pipecat/services/ai_services.py

@@ -294,6 +294,8 @@ class TTSService(AIService):
 
     async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
         self._current_sentence = ""
+        if self._text_filter:
+            self._text_filter.handle_interruption()
         await self.push_frame(frame, direction)
 
     async def _process_text_frame(self, frame: TextFrame):
@@ -318,6 +320,7 @@ class TTSService(AIService):
 
         await self.start_processing_metrics()
         if self._text_filter:
+            self._text_filter.reset_interruption()
             text = self._text_filter.filter(text)
         await self.process_generator(self.run_tts(text))
         await self.stop_processing_metrics()

src/pipecat/services/cartesia.py

@@ -152,7 +152,7 @@ class CartesiaTTSService(WordTTSService):
                 voice_config["__experimental_controls"]["emotion"] = self._settings["emotion"]
 
         msg = {
-            "transcript": text,
+            "transcript": text or " ",  # Text must contain at least one character
             "continue": continue_transcript,
             "context_id": self._context_id,
             "model_id": self.model_name,
@@ -271,7 +271,7 @@ class CartesiaTTSService(WordTTSService):
                 yield TTSStartedFrame()
                 self._context_id = str(uuid.uuid4())
 
-            msg = self._build_msg(text=text)
+            msg = self._build_msg(text=text or " ")  # Text must contain at least one character
 
             try:
                 await self._get_websocket().send(msg)

src/pipecat/services/openai.py

@@ -408,7 +408,7 @@ class OpenAITTSService(TTSService):
             await self.start_ttfb_metrics()
 
             async with self._client.audio.speech.with_streaming_response.create(
-                input=text,
+                input=text or " ",  # Text must contain at least one character
                 model=self.model_name,
                 voice=VALID_VOICES[self._voice_id],
                 response_format="pcm",

src/pipecat/utils/text/base_text_filter.py

@@ -16,3 +16,11 @@ class BaseTextFilter(ABC):
     @abstractmethod
     def filter(self, text: str) -> str:
         pass
+
+    @abstractmethod
+    def handle_interruption(self):
+        pass
+
+    @abstractmethod
+    def reset_interruption(self):
+        pass

src/pipecat/utils/text/markdown_text_filter.py

@@ -5,7 +5,7 @@
 #
 
 import re
-from typing import Any, Mapping
+from typing import Any, Mapping, Optional
 
 from markdown import Markdown
 from pydantic import BaseModel
@@ -22,11 +22,16 @@ class MarkdownTextFilter(BaseTextFilter):
     """
 
     class InputParams(BaseModel):
-        enable_text_filter: bool = True
+        enable_text_filter: Optional[bool] = True
+        filter_code: Optional[bool] = False
+        filter_tables: Optional[bool] = False
 
     def __init__(self, params: InputParams = InputParams(), **kwargs):
         super().__init__(**kwargs)
         self._settings = params
+        self._in_code_block = False
+        self._in_table = False
+        self._interrupted = False
 
     def update_settings(self, settings: Mapping[str, Any]):
         for key, value in settings.items():
@@ -35,27 +40,35 @@ class MarkdownTextFilter(BaseTextFilter):
 
     def filter(self, text: str) -> str:
         if self._settings.enable_text_filter:
-            # Replace newlines with spaces only when there's no text before or after
+            # Remove newlines only when there's no text before or after
-            text = re.sub(r"^\s*\n", " ", text, flags=re.MULTILINE)
+            filtered_text = re.sub(r"^\s*\n", "", text, flags=re.MULTILINE)
+
+            # Remove backticks from inline code, but not from code blocks
+            filtered_text = re.sub(r"(?<!`)`([^`\n]+)`(?!`)", r"\1", filtered_text)
 
             # Remove repeated sequences of 5 or more characters
-            text = re.sub(r"(\S)(\1{4,})", "", text)
+            filtered_text = re.sub(r"(\S)(\1{4,})", "", filtered_text)
 
             # Preserve numbered list items with a unique marker, §NUM§
-            text = re.sub(r"^(\d+\.)\s", r"§NUM§\1 ", text)
+            filtered_text = re.sub(r"^(\d+\.)\s", r"§NUM§\1 ", filtered_text)
 
             # Preserve leading/trailing spaces with a unique marker, §
             # Critical for word-by-word streaming in bot-tts-text
-            preserved_markdown = re.sub(
+            filtered_text = re.sub(
-                r"^( +)|\s+$", lambda m: "§" * len(m.group(0)), text, flags=re.MULTILINE
+                r"^( +)|\s+$", lambda m: "§" * len(m.group(0)), filtered_text, flags=re.MULTILINE
             )
 
             # Convert markdown to HTML
-            md = Markdown()
+            extension = ["tables"] if self._settings.filter_tables else []
-            html = md.convert(preserved_markdown)
+            md = Markdown(extensions=extension)
+            filtered_text = md.convert(filtered_text)
+
+            # Remove tables
+            if self._settings.filter_tables:
+                filtered_text = self.remove_tables(filtered_text)
 
             # Remove HTML tags
-            filtered_text = re.sub("<[^<]+?>", "", html)
+            filtered_text = re.sub("<[^<]+?>", "", filtered_text)
 
             # Replace HTML entities
             filtered_text = filtered_text.replace("&nbsp;", " ")
@@ -73,6 +86,10 @@ class MarkdownTextFilter(BaseTextFilter):
             filtered_text = re.sub(r"\|", "", filtered_text)
             filtered_text = re.sub(r"^\s*[-:]+\s*$", "", filtered_text, flags=re.MULTILINE)
 
+            # Remove code blocks
+            if self._settings.filter_code:
+                filtered_text = self._remove_code_blocks(filtered_text)
+
             # Restore numbered list items
             filtered_text = filtered_text.replace("§NUM§", "")
 
@@ -82,3 +99,114 @@ class MarkdownTextFilter(BaseTextFilter):
             return filtered_text
         else:
             return text
+
+    def handle_interruption(self):
+        self._interrupted = True
+        self._in_code_block = False
+        self._in_table = False
+
+    def reset_interruption(self):
+        self._interrupted = False
+
+    #
+    # Filter code
+    #
+
+    def _remove_code_blocks(self, text: str) -> str:
+        """
+        Main method to remove code blocks from the input text.
+        Handles interruptions and delegates to specific methods based on the current state.
+        """
+        if self._interrupted:
+            self._in_code_block = False
+            return text
+
+        # Pattern to match three consecutive backticks (code block delimiter)
+        code_block_pattern = r"```"
+        match = re.search(code_block_pattern, text)
+
+        if self._in_code_block:
+            return self._handle_in_code_block(match, text)
+
+        return self._handle_not_in_code_block(match, text, code_block_pattern)
+
+    def _handle_in_code_block(self, match, text):
+        """
+        Handle text when we're currently inside a code block.
+        If we find the end of the block, return text after it. Otherwise, skip the content.
+        """
+        if match:
+            self._in_code_block = False
+            end_index = match.end()
+            return text[end_index:].strip()
+        return ""  # Skip content inside code block
+
+    def _handle_not_in_code_block(self, match, text, code_block_pattern):
+        """
+        Handle text when we're not currently inside a code block.
+        Delegate to specific methods based on whether we find a code block delimiter.
+        """
+        if not match:
+            return text  # No code block found, return original text
+
+        start_index = match.start()
+        if start_index == 0 or text[:start_index].isspace():
+            return self._handle_start_of_code_block(text, start_index)
+        return self._handle_code_block_within_text(text, code_block_pattern)
+
+    def _handle_start_of_code_block(self, text, start_index):
+        """
+        Handle the case where we find the start of a code block.
+        Return any text before the code block and set the state to inside a code block.
+        """
+        self._in_code_block = True
+        return text[:start_index].strip()
+
+    def _handle_code_block_within_text(self, text, code_block_pattern):
+        """
+        Handle the case where we find a code block within the text.
+        If it's a complete code block, remove it and return surrounding text.
+        If it's the start of a code block, return text before it and set state.
+        """
+        parts = re.split(code_block_pattern, text)
+        if len(parts) > 2:
+            return (parts[0] + " " + parts[-1]).strip()
+        self._in_code_block = True
+        return parts[0].strip()
+
+    #
+    # Filter tables
+    #
+    def remove_tables(self, text: str) -> str:
+        """
+        Remove tables from the input text, handling cases where
+        both start and end tags are in the same input.
+        """
+        if self._interrupted:
+            self._in_table = False
+            return text
+
+        # Pattern to match entire table or parts of it
+        table_pattern = r"<table>.*?</table>"
+        partial_table_start = r"<table>.*"
+        partial_table_end = r".*</table>"
+
+        # Remove complete tables
+        text = re.sub(table_pattern, "", text, flags=re.DOTALL | re.IGNORECASE)
+
+        # Handle partial tables at the start
+        if self._in_table:
+            match = re.match(partial_table_end, text, re.DOTALL | re.IGNORECASE)
+            if match:
+                self._in_table = False
+                return text[match.end() :].strip()
+            else:
+                return ""  # Still inside a table, remove all content
+
+        # Handle partial tables at the end
+        match = re.search(partial_table_start, text, re.DOTALL | re.IGNORECASE)
+        if match:
+            self._in_table = True
+            return text[: match.start()].strip()
+
+        return text.strip()