Use ParallelPipeline for content filter in example 07

Run the content filter concurrently with LLM text generation
using ParallelPipeline, with a ContentFilterGate that blocks
output until the filter approves or rejects the content.

examples/foundational/07-interruptible-content-filter.py

@@ -5,6 +5,7 @@
 #
 
 import os
+from dataclasses import dataclass
 
 from dotenv import load_dotenv
 from loguru import logger
@@ -13,7 +14,16 @@ 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 Frame, LLMContextFrame, LLMRunFrame
+from pipecat.frames.frames import (
+    ControlFrame,
+    Frame,
+    LLMContextFrame,
+    LLMFullResponseEndFrame,
+    LLMRunFrame,
+    SystemFrame,
+    TTSSpeakFrame,
+)
+from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -34,11 +44,26 @@ load_dotenv(override=True)
 FILTERED_WORDS = ["apple", "banana", "car"]
 
 
-class ContentFilterProcessor(FrameProcessor):
+@dataclass
-    """Processor that filters LLMContextFrames containing specific words.
+class ContentApprovedFrame(ControlFrame):
+    """Signal frame indicating content passed the filter."""
 
-    If the user's message contains any of the filtered words, the context
+    pass
-    is replaced with a message indicating the assistant cannot respond.
+
+
+@dataclass
+class ContentRejectedFrame(ControlFrame):
+    """Signal frame indicating content was rejected by the filter."""
+
+    pass
+
+
+class ContentFilterProcessor(FrameProcessor):
+    """Checks LLMContextFrames for filtered words and emits signal frames.
+
+    Runs in one branch of a ParallelPipeline. Emits ContentApprovedFrame or
+    ContentRejectedFrame so that a downstream ContentFilterGate can decide
+    whether to let the LLM's output through.
     """
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -49,23 +74,71 @@ class ContentFilterProcessor(FrameProcessor):
             messages = frame.context.messages
             if messages:
                 last_message = messages[-1]
-                content = last_message.get("content", "")
+                content = last_message.get("content", "") if isinstance(last_message, dict) else ""
                 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
+                        await self.push_frame(ContentRejectedFrame(), direction)
-                        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
 
+            # Content is clean — approve it
+            await self.push_frame(ContentApprovedFrame(), direction)
+            return
+
+        await self.push_frame(frame, direction)
+
+
+class ContentFilterGate(FrameProcessor):
+    """Gates LLM output until the content filter signals approval or rejection.
+
+    Placed after a ParallelPipeline that runs a ContentFilterProcessor alongside
+    an LLM. Because the content filter (a fast regex check) completes before the
+    LLM's first token arrives, the signal frame always reaches this gate first.
+
+    - On ContentApprovedFrame: subsequent LLM output passes through normally.
+    - On ContentRejectedFrame: LLM output is discarded and a canned rejection
+      message is spoken instead via TTSSpeakFrame.
+
+    Note: For a production implementation with a slow content filter (e.g. an
+    external moderation API), you would add frame buffering so that LLM output
+    arriving before the filter decision is held rather than passed through.
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self._rejecting = False
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        # System frames always pass through.
+        if isinstance(frame, SystemFrame):
+            await self.push_frame(frame, direction)
+            return
+
+        # Content filter approved — LLM output will pass through normally.
+        if isinstance(frame, ContentApprovedFrame):
+            return
+
+        # Content filter rejected — suppress LLM output and speak a rejection.
+        if isinstance(frame, ContentRejectedFrame):
+            self._rejecting = True
+            await self.push_frame(
+                TTSSpeakFrame(text="I'm sorry, I can't respond to that."), direction
+            )
+            return
+
+        # LLMFullResponseEndFrame marks the end of the LLM's response.
+        # When rejecting, consume it to finish suppression.
+        if isinstance(frame, LLMFullResponseEndFrame) and self._rejecting:
+            self._rejecting = False
+            return
+
+        # While rejecting, discard all other frames (LLM text, etc.).
+        if self._rejecting:
+            return
+
         await self.push_frame(frame, direction)
 
 
@@ -97,10 +170,10 @@ transport_params = {
 async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     logger.info(f"Starting bot")
 
-    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY", ""))
 
     tts = CartesiaTTSService(
-        api_key=os.getenv("CARTESIA_API_KEY"),
+        api_key=os.getenv("CARTESIA_API_KEY", ""),
         voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
     )
 
@@ -116,15 +189,24 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     context = LLMContext(messages)
     context_aggregator = LLMContextAggregatorPair(context)
     content_filter = ContentFilterProcessor()
+    content_gate = ContentFilterGate()
 
+    # The content filter and LLM run in parallel. The content filter emits
+    # a signal frame (approved/rejected) while the LLM generates text
+    # concurrently. The gate after the ParallelPipeline blocks output until
+    # the content filter decides. TTS is placed after the gate so rejected
+    # content never reaches it.
     pipeline = Pipeline(
         [
             transport.input(),  # Transport user input
             stt,
             context_aggregator.user(),  # User responses
-            content_filter,  # Content filter
+            ParallelPipeline(
-            llm,  # LLM
+                [content_filter],  # Branch 1: content filter (emits signal frames)
-            tts,  # TTS
+                [llm],  # Branch 2: LLM text generation
+            ),
+            content_gate,  # Gates output until content filter approves
+            tts,  # TTS (only processes approved text)
             transport.output(),  # Transport bot output
             context_aggregator.assistant(),  # Assistant spoken responses
         ]