diff --git a/CHANGELOG.md b/CHANGELOG.md index 5396dcf42..2b7b3d7f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,24 @@ 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 `enable_metrics` to `PipelineParams`. + +- Added `MetricsFrame`. The `MetricsFrame` will report different metrics in the + system. Right now, it can report TTFB (Time To First Byte) values for + different services, that is the time spent between the arrival of a `Frame` to + the processor/service until the first `DataFrame` is pushed downstream. + +- Added TTFB metrics and debug logging for TTS services. + +### Fixed + +- Fixed PlayHT TTS service to work properly async. + ## [0.0.28] - 2024-06-05 ### Fixed 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/07d-interruptible-cartesia.py b/examples/foundational/07d-interruptible-cartesia.py index 39a77492b..283baa49a 100644 --- a/examples/foundational/07d-interruptible-cartesia.py +++ b/examples/foundational/07d-interruptible-cartesia.py @@ -20,6 +20,7 @@ from pipecat.services.openai import OpenAILLMService from pipecat.transports.services.daily import DailyParams, DailyTransport from pipecat.vad.silero import SileroVADAnalyzer + from runner import configure from loguru import logger diff --git a/examples/foundational/07e-interruptible-playht.py b/examples/foundational/07e-interruptible-playht.py new file mode 100644 index 000000000..c6c062e52 --- /dev/null +++ b/examples/foundational/07e-interruptible-playht.py @@ -0,0 +1,96 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +import os +import sys + +from pipecat.frames.frames import LLMMessagesFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantResponseAggregator, LLMUserResponseAggregator) +from pipecat.services.playht import PlayHTTTSService +from pipecat.services.openai import OpenAILLMService +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer +from pipecat.processors.logger import FrameLogger + +from runner import configure + +from loguru import logger + +from dotenv import load_dotenv +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def main(room_url: str, token): + async with aiohttp.ClientSession() as session: + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + audio_out_sample_rate=16000, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() + ) + ) + + tts = PlayHTTTSService( + user_id=os.getenv("PLAYHT_USER_ID"), + api_key=os.getenv("PLAYHT_API_KEY"), + voice_url="s3://voice-cloning-zero-shot/801a663f-efd0-4254-98d0-5c175514c3e8/jennifer/manifest.json", + ) + + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY"), + model="gpt-4o") + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) + + pipeline = Pipeline([ + transport.input(), # Transport user input + tma_in, # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + tma_out # Assistant spoken responses + ]) + + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + messages.append( + {"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + (url, token) = configure() + asyncio.run(main(url, token)) diff --git a/examples/foundational/07f-interruptible-azure-tts.py b/examples/foundational/07f-interruptible-azure-tts.py new file mode 100644 index 000000000..1770a3213 --- /dev/null +++ b/examples/foundational/07f-interruptible-azure-tts.py @@ -0,0 +1,95 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +import os +import sys + +from pipecat.frames.frames import LLMMessagesFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantResponseAggregator, LLMUserResponseAggregator) +from pipecat.services.azure import AzureTTSService +from pipecat.services.openai import OpenAILLMService +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer + + +from runner import configure + +from loguru import logger + +from dotenv import load_dotenv +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def main(room_url: str, token): + async with aiohttp.ClientSession() as session: + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + audio_out_sample_rate=16000, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() + ) + ) + + tts = AzureTTSService( + api_key=os.getenv("AZURE_SPEECH_API_KEY"), + region=os.getenv("AZURE_SPEECH_REGION"), + ) + + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY"), + model="gpt-4o") + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) + + pipeline = Pipeline([ + transport.input(), # Transport user input + tma_in, # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + tma_out # Assistant spoken responses + ]) + + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + messages.append( + {"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + (url, token) = configure() + asyncio.run(main(url, token)) diff --git a/examples/foundational/07g-interruptible-openai-tts.py b/examples/foundational/07g-interruptible-openai-tts.py new file mode 100644 index 000000000..2a45b63d8 --- /dev/null +++ b/examples/foundational/07g-interruptible-openai-tts.py @@ -0,0 +1,94 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +import os +import sys + +from pipecat.frames.frames import LLMMessagesFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantResponseAggregator, LLMUserResponseAggregator) +from pipecat.services.openai import OpenAITTSService +from pipecat.services.openai import OpenAILLMService +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer + +from runner import configure + +from loguru import logger + +from dotenv import load_dotenv +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def main(room_url: str, token): + async with aiohttp.ClientSession() as session: + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + audio_out_sample_rate=24000, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() + ) + ) + + tts = OpenAITTSService( + api_key=os.getenv("OPENAI_API_KEY"), + voice="alloy" + ) + + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY"), + model="gpt-4o") + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) + + pipeline = Pipeline([ + transport.input(), # Transport user input + tma_in, # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + tma_out # Assistant spoken responses + ]) + + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + messages.append( + {"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + (url, token) = configure() + asyncio.run(main(url, token)) 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..572246c6d 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 @@ -238,6 +239,13 @@ class StopInterruptionFrame(SystemFrame): pass +@dataclass +class MetricsFrame(SystemFrame): + """Emitted by processor that can compute metrics like latencies. + """ + ttfb: Mapping[str, float] + + # # Control frames # 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..76d98165b 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -19,6 +19,7 @@ from loguru import logger class PipelineParams(BaseModel): allow_interruptions: bool = False + enable_metrics: bool = False class Source(FrameProcessor): @@ -28,6 +29,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) @@ -87,8 +90,12 @@ class PipelineTask: raise Exception("Frames must be an iterable or async iterable") async def _process_down_queue(self): - await self._source.process_frame( - StartFrame(allow_interruptions=self._params.allow_interruptions), FrameDirection.DOWNSTREAM) + start_frame = StartFrame( + allow_interruptions=self._params.allow_interruptions, + enable_metrics=self._params.enable_metrics, + ) + await self._source.process_frame(start_frame, FrameDirection.DOWNSTREAM) + running = True should_cleanup = True while running: 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..e0a072477 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -5,10 +5,11 @@ # import asyncio +import time from enum import Enum -from pipecat.frames.frames import ErrorFrame, Frame +from pipecat.frames.frames import ErrorFrame, Frame, MetricsFrame, StartFrame from pipecat.utils.utils import obj_count, obj_id from loguru import logger @@ -28,6 +29,32 @@ 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 + + # Metrics + self._start_ttfb_time = 0 + + @property + def interruptions_allowed(self): + return self._allow_interruptions + + @property + def metrics_enabled(self): + return self._enable_metrics + + async def start_ttfb_metrics(self): + if self.metrics_enabled: + self._start_ttfb_time = time.time() + + async def stop_ttfb_metrics(self): + if self.metrics_enabled and self._start_ttfb_time > 0: + ttfb = time.time() - self._start_ttfb_time + logger.debug(f"{self.name} TTFB: {ttfb}") + await self.push_frame(MetricsFrame(ttfb={self.name: ttfb})) + self._start_ttfb_time = 0 + async def cleanup(self): pass @@ -40,7 +67,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..95b31f336 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -4,7 +4,6 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import time import base64 from pipecat.frames.frames import ( @@ -102,13 +101,16 @@ class AnthropicLLMService(LLMService): messages = self._get_messages_from_openai_context(context) - start_time = time.time() + await self.start_ttfb_metric() + response = await self._client.messages.create( messages=messages, model=self._model, max_tokens=self._max_tokens, stream=True) - logger.debug(f"Anthropic LLM TTFB: {time.time() - start_time}") + + await self.stop_ttfb_metric() + async for event in response: # logger.debug(f"Anthropic LLM event: {event}") if (event.type == "content_block_delta"): @@ -122,6 +124,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/azure.py b/src/pipecat/services/azure.py index 216d0555b..d35584303 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -11,7 +11,6 @@ import io from PIL import Image from typing import AsyncGenerator -from numpy import str_ from openai import AsyncAzureOpenAI from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, URLImageRawFrame @@ -48,6 +47,8 @@ class AzureTTSService(TTSService): async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: {text}") + await self.start_ttfb_metrics() + ssml = ( "" @@ -61,6 +62,7 @@ class AzureTTSService(TTSService): result = await asyncio.to_thread(self.speech_synthesizer.speak_ssml, (ssml)) if result.reason == ResultReason.SynthesizingAudioCompleted: + await self.stop_ttfb_metrics() # Azure always sends a 44-byte header. Strip it off. yield AudioRawFrame(audio=result.audio_data[44:], sample_rate=16000, num_channels=1) elif result.reason == ResultReason.Canceled: diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index f2d8c9b14..474600ffa 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -43,6 +43,8 @@ class CartesiaTTSService(TTSService): logger.debug(f"Generating TTS: [{text}]") try: + await self.start_ttfb_metrics() + chunk_generator = await self._client.generate( stream=True, transcript=text, @@ -52,6 +54,7 @@ class CartesiaTTSService(TTSService): ) async for chunk in chunk_generator: + await self.stop_ttfb_metrics() yield AudioRawFrame(chunk["audio"], chunk["sampling_rate"], 1) except Exception as e: logger.error(f"Cartesia exception: {e}") diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index b5901825e..aaf0c01ee 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -38,6 +38,7 @@ class DeepgramTTSService(TTSService): body = {"text": text} try: + await self.start_ttfb_metrics() async with self._aiohttp_session.post(request_url, headers=headers, json=body) as r: if r.status != 200: text = await r.text() @@ -46,6 +47,7 @@ class DeepgramTTSService(TTSService): return async for data in r.content: + await self.stop_ttfb_metrics() frame = AudioRawFrame(audio=data, sample_rate=16000, num_channels=1) yield frame except Exception as e: diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 3d602595d..717460fd3 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -47,6 +47,8 @@ class ElevenLabsTTSService(TTSService): "Content-Type": "application/json", } + await self.start_ttfb_metrics() + async with self._aiohttp_session.post(url, json=payload, headers=headers, params=querystring) as r: if r.status != 200: text = await r.text() @@ -56,5 +58,6 @@ class ElevenLabsTTSService(TTSService): async for chunk in r.content: if len(chunk) > 0: + await self.stop_ttfb_metrics() frame = AudioRawFrame(chunk, 16000, 1) yield frame diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index 81d5e756e..ffe9a9fee 100644 --- a/src/pipecat/services/google.py +++ b/src/pipecat/services/google.py @@ -1,8 +1,10 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# -import json -import os import asyncio -import time from typing import List @@ -81,9 +83,11 @@ class GoogleLLMService(LLMService): messages = self._get_messages_from_openai_context(context) - start_time = time.time() + await self.start_ttfb_metrics() + response = self._client.generate_content(messages, stream=True) - logger.debug(f"Google LLM TTFB: {time.time() - start_time}") + + await self.stop_ttfb_metrics() async for chunk in self._async_generator_wrapper(response): try: @@ -105,6 +109,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 2937eb61d..aaed7d8e7 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -3,13 +3,14 @@ # # SPDX-License-Identifier: BSD 2-Clause License # + +import aiohttp import base64 import io import json -import time + from typing import AsyncGenerator, List, Literal -import aiohttp from loguru import logger from PIL import Image @@ -94,7 +95,6 @@ class BaseOpenAILLMService(LLMService): del message["data"] del message["mime_type"] - start_time = time.time() chunks: AsyncStream[ChatCompletionChunk] = ( await self._client.chat.completions.create( model=self._model, @@ -105,8 +105,6 @@ class BaseOpenAILLMService(LLMService): ) ) - logger.debug(f"OpenAI LLM TTFB: {time.time() - start_time}") - return chunks async def _chat_completions(self, messages) -> str | None: @@ -123,6 +121,8 @@ class BaseOpenAILLMService(LLMService): arguments = "" tool_call_id = "" + await self.start_ttfb_metrics() + chunk_stream: AsyncStream[ChatCompletionChunk] = ( await self._stream_chat_completions(context) ) @@ -131,6 +131,8 @@ class BaseOpenAILLMService(LLMService): if len(chunk.choices) == 0: continue + await self.stop_ttfb_metrics() + if chunk.choices[0].delta.tool_calls: # We're streaming the LLM response to enable the fastest response times. # For text, we just yield each chunk as we receive it and count on consumers @@ -215,6 +217,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 @@ -307,6 +311,8 @@ class OpenAITTSService(TTSService): logger.debug(f"Generating TTS: [{text}]") try: + await self.start_ttfb_metrics() + async with self._client.audio.speech.with_streaming_response.create( input=text, model=self._model, @@ -320,6 +326,7 @@ class OpenAITTSService(TTSService): return async for chunk in r.iter_bytes(8192): if len(chunk) > 0: + await self.stop_ttfb_metrics() frame = AudioRawFrame(chunk, 24_000, 1) yield frame except BadRequestError as e: diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index e81cf1480..cf38e125d 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -15,8 +15,8 @@ from pipecat.services.ai_services import TTSService from loguru import logger try: - from pyht import Client from pyht.client import TTSOptions + from pyht.async_client import AsyncClient from pyht.protos.api_pb2 import Format except ModuleNotFoundError as e: logger.error(f"Exception: {e}") @@ -25,7 +25,7 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") -class PlayHTAIService(TTSService): +class PlayHTTTSService(TTSService): def __init__(self, *, api_key: str, user_id: str, voice_url: str, **kwargs): super().__init__(**kwargs) @@ -33,7 +33,7 @@ class PlayHTAIService(TTSService): self._user_id = user_id self._speech_key = api_key - self._client = Client( + self._client = AsyncClient( user_id=self._user_id, api_key=self._speech_key, ) @@ -47,28 +47,37 @@ class PlayHTAIService(TTSService): self._client.close() async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - b = bytearray() - in_header = True - for chunk in self._client.tts(text, self._options): - # skip the RIFF header. - if in_header: - b.extend(chunk) - if len(b) <= 36: - continue - else: - fh = io.BytesIO(b) - fh.seek(36) - (data, size) = struct.unpack('<4sI', fh.read(8)) - logger.debug( - f"first attempt: data: {data}, size: {hex(size)}, position: {fh.tell()}") - while data != b'data': - fh.read(size) + logger.debug(f"Generating TTS: [{text}]") + + try: + b = bytearray() + in_header = True + + await self.start_ttfb_metrics() + + playht_gen = self._client.tts( + text, + voice_engine="PlayHT2.0-turbo", + options=self._options) + + async for chunk in playht_gen: + # skip the RIFF header. + if in_header: + b.extend(chunk) + if len(b) <= 36: + continue + else: + fh = io.BytesIO(b) + fh.seek(36) (data, size) = struct.unpack('<4sI', fh.read(8)) - logger.debug( - f"subsequent data: {data}, size: {hex(size)}, position: {fh.tell()}, data != data: {data != b'data'}") - logger.debug("position: ", fh.tell()) - in_header = False - else: - if len(chunk): - frame = AudioRawFrame(chunk, 16000, 1) - yield frame + while data != b'data': + fh.read(size) + (data, size) = struct.unpack('<4sI', fh.read(8)) + in_header = False + else: + if len(chunk): + await self.stop_ttfb_metrics() + frame = AudioRawFrame(chunk, 16000, 1) + yield frame + except Exception as e: + logger.error(f"Error generating TTS: {e}") diff --git a/src/pipecat/services/whisper.py b/src/pipecat/services/whisper.py index 7884d454b..6c5dd0d1f 100644 --- a/src/pipecat/services/whisper.py +++ b/src/pipecat/services/whisper.py @@ -73,6 +73,8 @@ class WhisperSTTService(STTService): logger.error("Whisper model not available") return + await self.start_ttfb_metrics() + # Divide by 32768 because we have signed 16-bit data. audio_float = np.frombuffer(audio, dtype=np.int16).astype(np.float32) / 32768.0 @@ -83,4 +85,5 @@ class WhisperSTTService(STTService): text += f"{segment.text} " if text: + await self.stop_ttfb_metrics() yield TranscriptionFrame(text, "", int(time.time_ns() / 1000000)) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 531269bec..823ab5844 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.interruptions_allowed: # 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..4c19a0b55 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -20,6 +20,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.frames.frames import ( AudioRawFrame, CancelFrame, + MetricsFrame, SpriteFrame, StartFrame, EndFrame, @@ -41,7 +42,6 @@ class BaseOutputTransport(FrameProcessor): self._params = params self._running = False - self._allow_interruptions = False self._executor = ThreadPoolExecutor(max_workers=5) @@ -62,11 +62,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 @@ -93,6 +88,9 @@ class BaseOutputTransport(FrameProcessor): def send_message(self, frame: TransportMessageFrame): pass + def send_metrics(self, frame: MetricsFrame): + pass + def write_frame_to_camera(self, frame: ImageRawFrame): pass @@ -111,6 +109,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 +136,7 @@ class BaseOutputTransport(FrameProcessor): await self._stopped_event.wait() async def _handle_interruptions(self, frame: Frame): - if not self._allow_interruptions: + if not self.interruptions_allowed: return if isinstance(frame, StartInterruptionFrame): @@ -170,6 +170,8 @@ class BaseOutputTransport(FrameProcessor): self._set_camera_images(frame.images) elif isinstance(frame, TransportMessageFrame): self.send_message(frame) + elif isinstance(frame, MetricsFrame): + self.send_metrics(frame) else: future = asyncio.run_coroutine_threadsafe( self._internal_push_frame(frame), self.get_event_loop()) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 893aa9400..0b440181c 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -27,6 +27,7 @@ from pipecat.frames.frames import ( Frame, ImageRawFrame, InterimTranscriptionFrame, + MetricsFrame, SpriteFrame, StartFrame, TranscriptionFrame, @@ -521,6 +522,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) @@ -636,6 +639,16 @@ class DailyOutputTransport(BaseOutputTransport): def send_message(self, frame: DailyTransportMessageFrame): self._client.send_message(frame) + def send_metrics(self, frame: MetricsFrame): + ttfb = [{"name": n, "time": t} for n, t in frame.ttfb.items()] + message = DailyTransportMessageFrame(message={ + "type": "pipecat-metrics", + "metrics": { + "ttfb": ttfb + }, + }) + self._client.send_message(message) + def write_raw_audio_frames(self, frames: bytes): self._client.write_raw_audio_frames(frames) @@ -709,7 +722,7 @@ class DailyTransport(BaseTransport): # DailyTransport # - @property + @ property def participant_id(self) -> str: return self._client.participant_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: