diff --git a/examples/foundational/07-interruptible-content-filter.py b/examples/foundational/07-interruptible-content-filter.py index f4c9cfb71..97e60e21c 100644 --- a/examples/foundational/07-interruptible-content-filter.py +++ b/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): - """Processor that filters LLMContextFrames containing specific words. +@dataclass +class ContentApprovedFrame(ControlFrame): + """Signal frame indicating content passed the filter.""" - If the user's message contains any of the filtered words, the context - is replaced with a message indicating the assistant cannot respond. + pass + + +@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 - 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) + await self.push_frame(ContentRejectedFrame(), 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 - llm, # LLM - tts, # TTS + ParallelPipeline( + [content_filter], # Branch 1: content filter (emits signal frames) + [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 ]