diff --git a/examples/foundational/05-sync-speech-and-image.py b/examples/foundational/05-sync-speech-and-image.py index f057c847c..1c2e97e8b 100644 --- a/examples/foundational/05-sync-speech-and-image.py +++ b/examples/foundational/05-sync-speech-and-image.py @@ -59,6 +59,8 @@ class MonthPrepender(FrameProcessor): self.prepend_to_next_text_frame = False async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, MonthFrame): self.most_recent_month = frame.month elif self.prepend_to_next_text_frame and isinstance(frame, TextFrame): diff --git a/examples/foundational/05a-local-sync-speech-and-image.py b/examples/foundational/05a-local-sync-speech-and-image.py index a377ebe98..7db6dc99f 100644 --- a/examples/foundational/05a-local-sync-speech-and-image.py +++ b/examples/foundational/05a-local-sync-speech-and-image.py @@ -50,6 +50,8 @@ async def main(): self.text = "" async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): self.text = frame.text await self.push_frame(frame, direction) @@ -60,6 +62,8 @@ async def main(): self.audio = bytearray() async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, AudioRawFrame): self.audio.extend(frame.audio) self.frame = AudioRawFrame( @@ -71,6 +75,8 @@ async def main(): self.frame = None async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, URLImageRawFrame): self.frame = frame diff --git a/examples/foundational/06a-image-sync.py b/examples/foundational/06a-image-sync.py index 2f5528ee4..686964502 100644 --- a/examples/foundational/06a-image-sync.py +++ b/examples/foundational/06a-image-sync.py @@ -49,6 +49,8 @@ class ImageSyncAggregator(FrameProcessor): self._waiting_image_bytes = self._waiting_image.tobytes() async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if not isinstance(frame, SystemFrame): await self.push_frame(ImageRawFrame(image=self._speaking_image_bytes, size=(1024, 1024), format=self._speaking_image_format)) await self.push_frame(frame) diff --git a/examples/foundational/11-sound-effects.py b/examples/foundational/11-sound-effects.py index 2a3e8effc..8abf3d935 100644 --- a/examples/foundational/11-sound-effects.py +++ b/examples/foundational/11-sound-effects.py @@ -60,6 +60,8 @@ for file in sound_files: class OutboundSoundEffectWrapper(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, LLMFullResponseEndFrame): await self.push_frame(sounds["ding1.wav"]) # In case anything else downstream needs it @@ -71,6 +73,8 @@ class OutboundSoundEffectWrapper(FrameProcessor): class InboundSoundEffectWrapper(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, LLMMessagesFrame): await self.push_frame(sounds["ding2.wav"]) # In case anything else downstream needs it diff --git a/examples/foundational/12-describe-video.py b/examples/foundational/12-describe-video.py index 256580c07..bde70d664 100644 --- a/examples/foundational/12-describe-video.py +++ b/examples/foundational/12-describe-video.py @@ -42,6 +42,8 @@ class UserImageRequester(FrameProcessor): self._participant_id = participant_id async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if self._participant_id and isinstance(frame, TextFrame): await self.push_frame(UserImageRequestFrame(self._participant_id), FrameDirection.UPSTREAM) await self.push_frame(frame, direction) diff --git a/examples/foundational/12a-describe-video-gemini-flash.py b/examples/foundational/12a-describe-video-gemini-flash.py index 00e168ad5..293e9a26a 100644 --- a/examples/foundational/12a-describe-video-gemini-flash.py +++ b/examples/foundational/12a-describe-video-gemini-flash.py @@ -42,6 +42,8 @@ class UserImageRequester(FrameProcessor): self._participant_id = participant_id async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if self._participant_id and isinstance(frame, TextFrame): await self.push_frame(UserImageRequestFrame(self._participant_id), FrameDirection.UPSTREAM) await self.push_frame(frame, direction) diff --git a/examples/foundational/12b-describe-video-gpt-4o.py b/examples/foundational/12b-describe-video-gpt-4o.py index c34a5a94a..19caa307c 100644 --- a/examples/foundational/12b-describe-video-gpt-4o.py +++ b/examples/foundational/12b-describe-video-gpt-4o.py @@ -42,6 +42,8 @@ class UserImageRequester(FrameProcessor): self._participant_id = participant_id async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if self._participant_id and isinstance(frame, TextFrame): await self.push_frame(UserImageRequestFrame(self._participant_id), FrameDirection.UPSTREAM) await self.push_frame(frame, direction) diff --git a/examples/foundational/12c-describe-video-anthropic.py b/examples/foundational/12c-describe-video-anthropic.py index 9a7a9a804..033af2a0a 100644 --- a/examples/foundational/12c-describe-video-anthropic.py +++ b/examples/foundational/12c-describe-video-anthropic.py @@ -42,6 +42,8 @@ class UserImageRequester(FrameProcessor): self._participant_id = participant_id async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if self._participant_id and isinstance(frame, TextFrame): await self.push_frame(UserImageRequestFrame(self._participant_id), FrameDirection.UPSTREAM) await self.push_frame(frame, direction) diff --git a/examples/foundational/13-whisper-transcription.py b/examples/foundational/13-whisper-transcription.py index 2ca8eb9fa..43d5f4840 100644 --- a/examples/foundational/13-whisper-transcription.py +++ b/examples/foundational/13-whisper-transcription.py @@ -29,6 +29,8 @@ logger.add(sys.stderr, level="DEBUG") class TranscriptionLogger(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TranscriptionFrame): print(f"Transcription: {frame.text}") diff --git a/examples/foundational/13a-whisper-local.py b/examples/foundational/13a-whisper-local.py index 366cd2cf5..6bf27aa0a 100644 --- a/examples/foundational/13a-whisper-local.py +++ b/examples/foundational/13a-whisper-local.py @@ -28,6 +28,8 @@ logger.add(sys.stderr, level="DEBUG") class TranscriptionLogger(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TranscriptionFrame): print(f"Transcription: {frame.text}") diff --git a/examples/moondream-chatbot/bot.py b/examples/moondream-chatbot/bot.py index 3e9ced259..8407d04cc 100644 --- a/examples/moondream-chatbot/bot.py +++ b/examples/moondream-chatbot/bot.py @@ -74,6 +74,8 @@ class TalkingAnimation(FrameProcessor): self._is_talking = False async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, AudioRawFrame): if not self._is_talking: await self.push_frame(talking_frame) @@ -93,6 +95,8 @@ class UserImageRequester(FrameProcessor): self.participant_id = participant_id async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if self.participant_id and isinstance(frame, TextFrame): if frame.text == user_request_answer: await self.push_frame(UserImageRequestFrame(self.participant_id), FrameDirection.UPSTREAM) @@ -107,6 +111,8 @@ class TextFilterProcessor(FrameProcessor): self.text = text async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): if frame.text != self.text: await self.push_frame(frame) @@ -116,6 +122,8 @@ class TextFilterProcessor(FrameProcessor): class ImageFilterProcessor(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if not isinstance(frame, ImageRawFrame): await self.push_frame(frame) diff --git a/examples/simple-chatbot/bot.py b/examples/simple-chatbot/bot.py index 80b60833f..778902a18 100644 --- a/examples/simple-chatbot/bot.py +++ b/examples/simple-chatbot/bot.py @@ -64,6 +64,8 @@ class TalkingAnimation(FrameProcessor): self._is_talking = False async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, AudioRawFrame): if not self._is_talking: await self.push_frame(talking_frame) diff --git a/examples/storytelling-chatbot/src/processors.py b/examples/storytelling-chatbot/src/processors.py index 18428eb72..a8b2a0980 100644 --- a/examples/storytelling-chatbot/src/processors.py +++ b/examples/storytelling-chatbot/src/processors.py @@ -52,6 +52,8 @@ class StoryImageProcessor(FrameProcessor): self._fal_service = fal_service async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, StoryImageFrame): try: async with timeout(7): @@ -86,6 +88,8 @@ class StoryProcessor(FrameProcessor): self._story = story async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, UserStoppedSpeakingFrame): # Send an app message to the UI await self.push_frame(DailyTransportMessageFrame(CUE_ASSISTANT_TURN)) diff --git a/examples/translation-chatbot/bot.py b/examples/translation-chatbot/bot.py index 38667d897..9354dc0be 100644 --- a/examples/translation-chatbot/bot.py +++ b/examples/translation-chatbot/bot.py @@ -40,6 +40,8 @@ class TranslationProcessor(FrameProcessor): self._language = language async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): context = [ { @@ -65,6 +67,8 @@ class TranslationSubtitles(FrameProcessor): # subtitles. # async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): message = { "language": self._language, diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 8eb32664c..9a5725de1 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -4,7 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from typing import Any, List, Tuple +from typing import Any, List, Mapping, Tuple from dataclasses import dataclass, field @@ -188,6 +188,7 @@ class SystemFrame(Frame): class StartFrame(SystemFrame): """This is the first frame that should be pushed down a pipeline.""" allow_interruptions: bool = False + enable_metrics: bool = False @dataclass diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py index 22ffdfdf2..ccf72bd90 100644 --- a/src/pipecat/pipeline/parallel_pipeline.py +++ b/src/pipecat/pipeline/parallel_pipeline.py @@ -20,6 +20,8 @@ class Source(FrameProcessor): self._up_queue = upstream_queue async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + match direction: case FrameDirection.UPSTREAM: await self._up_queue.put(frame) @@ -34,6 +36,8 @@ class Sink(FrameProcessor): self._down_queue = downstream_queue async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + match direction: case FrameDirection.UPSTREAM: await self.push_frame(frame, direction) @@ -90,6 +94,8 @@ class ParallelPipeline(FrameProcessor): self._down_task = loop.create_task(self._process_down_queue()) async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, StartFrame): await self._start_tasks() diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py index 2ba061a37..2cb5b45d4 100644 --- a/src/pipecat/pipeline/pipeline.py +++ b/src/pipecat/pipeline/pipeline.py @@ -19,6 +19,8 @@ class PipelineSource(FrameProcessor): self._upstream_push_frame = upstream_push_frame async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + match direction: case FrameDirection.UPSTREAM: await self._upstream_push_frame(frame, direction) @@ -33,6 +35,8 @@ class PipelineSink(FrameProcessor): self._downstream_push_frame = downstream_push_frame async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + match direction: case FrameDirection.UPSTREAM: await self.push_frame(frame, direction) @@ -61,6 +65,8 @@ class Pipeline(FrameProcessor): await self._cleanup_processors() async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if direction == FrameDirection.DOWNSTREAM: await self._source.process_frame(frame, FrameDirection.DOWNSTREAM) elif direction == FrameDirection.UPSTREAM: diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index e3e7f7e36..b1f72c096 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -28,6 +28,8 @@ class Source(FrameProcessor): self._up_queue = up_queue async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + match direction: case FrameDirection.UPSTREAM: await self._up_queue.put(frame) diff --git a/src/pipecat/processors/aggregators/gated.py b/src/pipecat/processors/aggregators/gated.py index 3c80e4641..16a7e3076 100644 --- a/src/pipecat/processors/aggregators/gated.py +++ b/src/pipecat/processors/aggregators/gated.py @@ -48,6 +48,8 @@ class GatedAggregator(FrameProcessor): self._accumulator: List[Frame] = [] async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + # We must not block system frames. if isinstance(frame, SystemFrame): await self.push_frame(frame, direction) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 289296487..fdab187bd 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -79,6 +79,8 @@ class LLMResponseAggregator(FrameProcessor): # and T2 would be dropped. async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + send_aggregation = False if isinstance(frame, self._start_frame): @@ -207,6 +209,8 @@ class LLMFullResponseAggregator(FrameProcessor): self._aggregation = "" async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): self._aggregation += frame.text elif isinstance(frame, LLMFullResponseEndFrame): diff --git a/src/pipecat/processors/aggregators/parallel_task.py b/src/pipecat/processors/aggregators/parallel_task.py index d2142f829..ce341bde5 100644 --- a/src/pipecat/processors/aggregators/parallel_task.py +++ b/src/pipecat/processors/aggregators/parallel_task.py @@ -22,6 +22,8 @@ class Source(FrameProcessor): self._up_queue = upstream_queue async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + match direction: case FrameDirection.UPSTREAM: await self._up_queue.put(frame) @@ -36,6 +38,8 @@ class Sink(FrameProcessor): self._down_queue = downstream_queue async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + match direction: case FrameDirection.UPSTREAM: await self.push_frame(frame, direction) @@ -80,6 +84,8 @@ class ParallelTask(FrameProcessor): # async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if direction == FrameDirection.UPSTREAM: # If we get an upstream frame we process it in each sink. await asyncio.gather(*[s.process_frame(frame, direction) for s in self._sinks]) diff --git a/src/pipecat/processors/aggregators/sentence.py b/src/pipecat/processors/aggregators/sentence.py index 3f323bbb5..a7992eace 100644 --- a/src/pipecat/processors/aggregators/sentence.py +++ b/src/pipecat/processors/aggregators/sentence.py @@ -33,6 +33,8 @@ class SentenceAggregator(FrameProcessor): self._aggregation = "" async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + # We ignore interim description at this point. if isinstance(frame, InterimTranscriptionFrame): return diff --git a/src/pipecat/processors/aggregators/user_response.py b/src/pipecat/processors/aggregators/user_response.py index 5ce62b859..12f3bcb93 100644 --- a/src/pipecat/processors/aggregators/user_response.py +++ b/src/pipecat/processors/aggregators/user_response.py @@ -82,6 +82,8 @@ class ResponseAggregator(FrameProcessor): # and T2 would be dropped. async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + send_aggregation = False if isinstance(frame, self._start_frame): diff --git a/src/pipecat/processors/aggregators/vision_image_frame.py b/src/pipecat/processors/aggregators/vision_image_frame.py index 45fd6756b..f0c8a9c76 100644 --- a/src/pipecat/processors/aggregators/vision_image_frame.py +++ b/src/pipecat/processors/aggregators/vision_image_frame.py @@ -30,6 +30,8 @@ class VisionImageFrameAggregator(FrameProcessor): self._describe_text = None async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): self._describe_text = frame.text elif isinstance(frame, ImageRawFrame): diff --git a/src/pipecat/processors/filters/frame_filter.py b/src/pipecat/processors/filters/frame_filter.py index 42f0e9dd8..9f2eb98c4 100644 --- a/src/pipecat/processors/filters/frame_filter.py +++ b/src/pipecat/processors/filters/frame_filter.py @@ -30,5 +30,7 @@ class FrameFilter(FrameProcessor): or isinstance(frame, SystemFrame)) async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if self._should_passthrough_frame(frame): await self.push_frame(frame, direction) diff --git a/src/pipecat/processors/filters/wake_check_filter.py b/src/pipecat/processors/filters/wake_check_filter.py index 44faa03bf..7704a5732 100644 --- a/src/pipecat/processors/filters/wake_check_filter.py +++ b/src/pipecat/processors/filters/wake_check_filter.py @@ -43,6 +43,8 @@ class WakeCheckFilter(FrameProcessor): self._wake_patterns.append(pattern) async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + try: if isinstance(frame, TranscriptionFrame): p = self._participant_states.get(frame.user_id) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 793aa8eb1..0b9c71fef 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -8,7 +8,7 @@ import asyncio from enum import Enum -from pipecat.frames.frames import ErrorFrame, Frame +from pipecat.frames.frames import ErrorFrame, Frame, StartFrame from pipecat.utils.utils import obj_count, obj_id from loguru import logger @@ -28,6 +28,18 @@ class FrameProcessor: self._next: "FrameProcessor" | None = None self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_running_loop() + # Properties + self._allow_interruptions = False + self._enable_metrics = False + + @property + def allow_interruptions(self): + return self._allow_interruptions + + @property + def enable_metrics(self): + return self._enable_metrics + async def cleanup(self): pass @@ -40,7 +52,9 @@ class FrameProcessor: return self._loop async def process_frame(self, frame: Frame, direction: FrameDirection): - pass + if isinstance(frame, StartFrame): + self._allow_interruptions = frame.allow_interruptions + self._enable_metrics = frame.enable_metrics async def push_error(self, error: ErrorFrame): await self.push_frame(error, FrameDirection.UPSTREAM) diff --git a/src/pipecat/processors/frameworks/langchain.py b/src/pipecat/processors/frameworks/langchain.py index bac39cf26..674d3aa97 100644 --- a/src/pipecat/processors/frameworks/langchain.py +++ b/src/pipecat/processors/frameworks/langchain.py @@ -39,6 +39,8 @@ class LangchainProcessor(FrameProcessor): self._participant_id = participant_id async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, LLMMessagesFrame): # Messages are accumulated by the `LLMUserResponseAggregator` in a list of messages. # The last one by the human is the one we want to send to the LLM. diff --git a/src/pipecat/processors/text_transformer.py b/src/pipecat/processors/text_transformer.py index 550b8dc1c..39fcf6b67 100644 --- a/src/pipecat/processors/text_transformer.py +++ b/src/pipecat/processors/text_transformer.py @@ -27,6 +27,8 @@ class StatelessTextTransformer(FrameProcessor): self._transform_fn = transform_fn async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): result = self._transform_fn(frame.text) if isinstance(result, Coroutine): diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index fc879d887..a46f41f35 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -106,6 +106,8 @@ class TTSService(AIService): await self.push_frame(TextFrame(text)) async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): await self._process_text_frame(frame) elif isinstance(frame, EndFrame): @@ -179,6 +181,8 @@ class STTService(AIService): async def process_frame(self, frame: Frame, direction: FrameDirection): """Processes a frame of audio data, either buffering or transcribing it.""" + await super().process_frame(frame, direction) + if isinstance(frame, CancelFrame) or isinstance(frame, EndFrame): self._wave.close() await self.push_frame(frame, direction) @@ -201,6 +205,8 @@ class ImageGenService(AIService): pass async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): await self.push_frame(frame, direction) await self.process_generator(self.run_image_gen(frame.text)) @@ -220,6 +226,8 @@ class VisionService(AIService): pass async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, VisionImageRawFrame): await self.process_generator(self.run_vision(frame)) else: diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 3e774a6bd..956941073 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -122,6 +122,8 @@ class AnthropicLLMService(LLMService): await self.push_frame(LLMFullResponseEndFrame()) async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + context = None if isinstance(frame, OpenAILLMContextFrame): diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index 81d5e756e..89ac6d83d 100644 --- a/src/pipecat/services/google.py +++ b/src/pipecat/services/google.py @@ -105,6 +105,8 @@ class GoogleLLMService(LLMService): await self.push_frame(LLMFullResponseEndFrame()) async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + context = None if isinstance(frame, OpenAILLMContextFrame): diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 4dca1a998..57bcc09d9 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -215,6 +215,8 @@ class BaseOpenAILLMService(LLMService): raise BaseException(f"Unknown return type from function callback: {type(result)}") async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + context = None if isinstance(frame, OpenAILLMContextFrame): context: OpenAILLMContext = frame.context diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 531269bec..3657727fb 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -34,7 +34,6 @@ class BaseInputTransport(FrameProcessor): self._params = params self._running = False - self._allow_interruptions = False self._executor = ThreadPoolExecutor(max_workers=5) @@ -43,11 +42,6 @@ class BaseInputTransport(FrameProcessor): self._create_push_task() async def start(self, frame: StartFrame): - # Make sure we have the latest params. Note that this transport might - # have been started on another task that might not need interruptions, - # for example. - self._allow_interruptions = frame.allow_interruptions - if self._running: return @@ -86,12 +80,13 @@ class BaseInputTransport(FrameProcessor): pass async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, CancelFrame): # We don't queue a CancelFrame since we want to stop ASAP. await self.push_frame(frame, direction) await self.stop() elif isinstance(frame, StartFrame): - self._allow_interruption = frame.allow_interruptions await self.start(frame) await self._internal_push_frame(frame, direction) elif isinstance(frame, EndFrame): @@ -128,7 +123,7 @@ class BaseInputTransport(FrameProcessor): # async def _handle_interruptions(self, frame: Frame): - if self._allow_interruptions: + if self.allow_interruptions: # Make sure we notify about interruptions quickly out-of-band if isinstance(frame, UserStartedSpeakingFrame): logger.debug("User started speaking") diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index d167919f6..e880c99d4 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -41,7 +41,6 @@ class BaseOutputTransport(FrameProcessor): self._params = params self._running = False - self._allow_interruptions = False self._executor = ThreadPoolExecutor(max_workers=5) @@ -62,11 +61,6 @@ class BaseOutputTransport(FrameProcessor): self._create_push_task() async def start(self, frame: StartFrame): - # Make sure we have the latest params. Note that this transport might - # have been started on another task that might not need interruptions, - # for example. - self._allow_interruptions = frame.allow_interruptions - if self._running: return @@ -111,6 +105,8 @@ class BaseOutputTransport(FrameProcessor): await self._sink_thread async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + # # Out-of-band frames like (CancelFrame or StartInterruptionFrame) are # pushed immediately. Other frames require order so they are put in the @@ -136,7 +132,7 @@ class BaseOutputTransport(FrameProcessor): await self._stopped_event.wait() async def _handle_interruptions(self, frame: Frame): - if not self._allow_interruptions: + if not self.allow_interruptions: return if isinstance(frame, StartInterruptionFrame): diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 893aa9400..8e42a1678 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -521,6 +521,8 @@ class DailyInputTransport(BaseInputTransport): # async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, UserImageRequestFrame): self.request_participant_image(frame.user_id) diff --git a/src/pipecat/utils/test_frame_processor.py b/src/pipecat/utils/test_frame_processor.py index f4a6674aa..2354c40e0 100644 --- a/src/pipecat/utils/test_frame_processor.py +++ b/src/pipecat/utils/test_frame_processor.py @@ -13,6 +13,8 @@ class TestFrameProcessor(FrameProcessor): super().__init__() async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + if not self.test_frames[0]: # then we've run out of required frames but the generator is still going? raise TestException(f"Oops, got an extra frame, {frame}") if isinstance(self.test_frames[0], List): diff --git a/src/pipecat/vad/silero.py b/src/pipecat/vad/silero.py index 97d0a2144..52ea159b1 100644 --- a/src/pipecat/vad/silero.py +++ b/src/pipecat/vad/silero.py @@ -94,6 +94,8 @@ class SileroVAD(FrameProcessor): # async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, AudioRawFrame): await self._analyze_audio(frame) if self._audio_passthrough: diff --git a/tests/test_langchain.py b/tests/test_langchain.py index 6b37d0987..7b32b2a9a 100644 --- a/tests/test_langchain.py +++ b/tests/test_langchain.py @@ -36,6 +36,8 @@ class TestLangchain(unittest.IsolatedAsyncioTestCase): return self.name async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + if isinstance(frame, LLMFullResponseStartFrame): self.start_collecting = True elif isinstance(frame, TextFrame) and self.start_collecting: