wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

processors: implement base process_frame(). all subsclassed should call it

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-06-06 10:53:32 -07:00
parent cf5c244487
commit e765a29ca2
39 changed files with 130 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
examples/foundational/05-sync-speech-and-image.py
View File

@@ -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):

6
examples/foundational/05a-local-sync-speech-and-image.py
View File

@@ -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

2
examples/foundational/06a-image-sync.py
View File

@@ -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)

4
examples/foundational/11-sound-effects.py
View File

@@ -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

2
examples/foundational/12-describe-video.py
View File

@@ -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)

2
examples/foundational/12a-describe-video-gemini-flash.py
View File

@@ -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)

2
examples/foundational/12b-describe-video-gpt-4o.py
View File

@@ -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)

2
examples/foundational/12c-describe-video-anthropic.py
View File

@@ -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)

2
examples/foundational/13-whisper-transcription.py
View File

@@ -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}")

2
examples/foundational/13a-whisper-local.py
View File

@@ -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}")

8
examples/moondream-chatbot/bot.py
View File

@@ -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)

2
examples/simple-chatbot/bot.py
View File

@@ -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)

4
examples/storytelling-chatbot/src/processors.py
View File

@@ -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))

4
examples/translation-chatbot/bot.py
View File

@@ -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,

3
src/pipecat/frames/frames.py
View File

@@ -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

6
src/pipecat/pipeline/parallel_pipeline.py
View File

@@ -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()

6
src/pipecat/pipeline/pipeline.py
View File

@@ -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:

2
src/pipecat/pipeline/task.py
View File

@@ -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)

2
src/pipecat/processors/aggregators/gated.py
View File

@@ -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)

4
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -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):

6
src/pipecat/processors/aggregators/parallel_task.py
View File

@@ -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])

2
src/pipecat/processors/aggregators/sentence.py
View File

@@ -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

2
src/pipecat/processors/aggregators/user_response.py
View File

@@ -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):

2
src/pipecat/processors/aggregators/vision_image_frame.py
View File

@@ -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):

2
src/pipecat/processors/filters/frame_filter.py
View File

@@ -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)

2
src/pipecat/processors/filters/wake_check_filter.py
View File

@@ -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)

18
src/pipecat/processors/frame_processor.py
View File

@@ -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)

2
src/pipecat/processors/frameworks/langchain.py
View File

@@ -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.

2
src/pipecat/processors/text_transformer.py
View File

@@ -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):

8
src/pipecat/services/ai_services.py
View File

@@ -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:

2
src/pipecat/services/anthropic.py
View File

@@ -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):

2
src/pipecat/services/google.py
View File

@@ -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):

2
src/pipecat/services/openai.py
View File

@@ -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

11
src/pipecat/transports/base_input.py
View File

@@ -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")

10
src/pipecat/transports/base_output.py
View File

@@ -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):

2
src/pipecat/transports/services/daily.py
View File

@@ -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)

2
src/pipecat/utils/test_frame_processor.py
View File

@@ -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):

2
src/pipecat/vad/silero.py
View File

@@ -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:

2
tests/test_langchain.py
View File

@@ -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: