diff --git a/CHANGELOG.md b/CHANGELOG.md index f8d991f6d..0dc1057ae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `AysncFrameProcessor` and `AsyncAIService`. Some services like + `DeepgramSTTService` need to process things asynchronously. For example, audio + is sent to Deepgram but transcriptions are not returned immediately. In these + cases we still require all frames (except system frames) to be pushed + downstream from a single task. That's what `AsyncFrameProcessor` is for. It + creates a task and all frames should be pushed from that task. So, whenever a + new Deepgram transcription is ready that transcription will also be pushed + from this internal task. + - The `MetricsFrame` now includes processing metrics if metrics are enabled. The processing metrics indicate the time a processor needs to generate all its output. Note that not all processors generate these kind of metrics. diff --git a/src/pipecat/processors/async_frame_processor.py b/src/pipecat/processors/async_frame_processor.py new file mode 100644 index 000000000..db805f924 --- /dev/null +++ b/src/pipecat/processors/async_frame_processor.py @@ -0,0 +1,62 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio + +from pipecat.frames.frames import EndFrame, Frame, StartInterruptionFrame +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +class AsyncFrameProcessor(FrameProcessor): + + def __init__( + self, + name: str | None = None, + loop: asyncio.AbstractEventLoop | None = None, + **kwargs): + super().__init__(name, loop, **kwargs) + + self._create_push_task() + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, StartInterruptionFrame): + await self._handle_interruptions(frame) + + async def queue_frame( + self, + frame: Frame, + direction: FrameDirection = FrameDirection.DOWNSTREAM): + await self._push_queue.put((frame, direction)) + + async def cleanup(self): + self._push_frame_task.cancel() + await self._push_frame_task + + async def _handle_interruptions(self, frame: Frame): + # Cancel the task. This will stop pushing frames downstream. + self._push_frame_task.cancel() + await self._push_frame_task + # Push an out-of-band frame (i.e. not using the ordered push + # frame task). + await self.push_frame(frame) + # Create a new queue and task. + self._create_push_task() + + def _create_push_task(self): + self._push_queue = asyncio.Queue() + self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler()) + + async def _push_frame_task_handler(self): + running = True + while running: + try: + (frame, direction) = await self._push_queue.get() + await self.push_frame(frame, direction) + running = not isinstance(frame, EndFrame) + except asyncio.CancelledError: + break diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 0fd7cb9d1..e3667a85c 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -24,6 +24,7 @@ from pipecat.frames.frames import ( TextFrame, VisionImageRawFrame, ) +from pipecat.processors.async_frame_processor import AsyncFrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.utils.audio import calculate_audio_volume from pipecat.utils.utils import exp_smoothing @@ -60,6 +61,30 @@ class AIService(FrameProcessor): await self.push_frame(f) +class AsyncAIService(AsyncFrameProcessor): + def __init__(self, **kwargs): + super().__init__(**kwargs) + + async def start(self, frame: StartFrame): + pass + + async def stop(self, frame: EndFrame): + pass + + async def cancel(self, frame: CancelFrame): + pass + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, StartFrame): + await self.start(frame) + elif isinstance(frame, CancelFrame): + await self.cancel(frame) + elif isinstance(frame, EndFrame): + await self.stop(frame) + + class LLMService(AIService): """This class is a no-op but serves as a base class for LLM services.""" diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index f7fdbf16b..873934abb 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -24,7 +24,7 @@ from pipecat.frames.frames import ( TranscriptionFrame, URLImageRawFrame) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.ai_services import AIService, TTSService, ImageGenService +from pipecat.services.ai_services import AIService, AsyncAIService, TTSService, ImageGenService from pipecat.services.openai import BaseOpenAILLMService from loguru import logger @@ -110,7 +110,7 @@ class AzureTTSService(TTSService): logger.error(f"{self} error: {cancellation_details.error_details}") -class AzureSTTService(AIService): +class AzureSTTService(AsyncAIService): def __init__( self, *, @@ -133,18 +133,10 @@ class AzureSTTService(AIService): speech_config=speech_config, audio_config=audio_config) self._speech_recognizer.recognized.connect(self._on_handle_recognized) - # This event will be used to ignore out-of-band transcriptions while we - # are itnerrupted. - self._is_interrupted_event = asyncio.Event() - - self._create_push_task() - async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) - if isinstance(frame, StartInterruptionFrame): - await self._handle_interruptions(frame) - elif isinstance(frame, SystemFrame): + if isinstance(frame, SystemFrame): await self.push_frame(frame, direction) elif isinstance(frame, AudioRawFrame): self._audio_stream.write(frame.audio) @@ -156,44 +148,14 @@ class AzureSTTService(AIService): async def stop(self, frame: EndFrame): self._speech_recognizer.stop_continuous_recognition_async() - await self._push_queue.put((frame, FrameDirection.DOWNSTREAM)) - await self._push_frame_task async def cancel(self, frame: CancelFrame): self._speech_recognizer.stop_continuous_recognition_async() - self._push_frame_task.cancel() - await self._push_frame_task - - async def _handle_interruptions(self, frame: Frame): - # Cancel the task. This will stop pushing frames downstream. - self._push_frame_task.cancel() - await self._push_frame_task - # Push an out-of-band frame (i.e. not using the ordered push - # frame task). - await self.push_frame(frame) - # Create a new queue and task. - self._create_push_task() - - def _create_push_task(self): - self._push_queue = asyncio.Queue() - self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler()) - - async def _push_frame_task_handler(self): - running = True - while running: - try: - (frame, direction) = await self._push_queue.get() - await self.push_frame(frame, direction) - running = not isinstance(frame, EndFrame) - except asyncio.CancelledError: - break def _on_handle_recognized(self, event): if event.result.reason == ResultReason.RecognizedSpeech and len(event.result.text) > 0: - direction = FrameDirection.DOWNSTREAM frame = TranscriptionFrame(event.result.text, "", int(time.time_ns() / 1000000)) - asyncio.run_coroutine_threadsafe( - self._push_queue.put((frame, direction)), self.get_event_loop()) + asyncio.run_coroutine_threadsafe(self.queue_frame(frame), self.get_event_loop()) class AzureImageGenServiceREST(ImageGenService): diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index 799f6a6a4..4ccd8c50f 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -22,7 +22,7 @@ from pipecat.frames.frames import ( SystemFrame, TranscriptionFrame) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.ai_services import AIService, TTSService +from pipecat.services.ai_services import AIService, AsyncAIService, TTSService from loguru import logger @@ -94,7 +94,7 @@ class DeepgramTTSService(TTSService): logger.exception(f"{self} exception: {e}") -class DeepgramSTTService(AIService): +class DeepgramSTTService(AsyncAIService): def __init__(self, api_key: str, url: str = "", @@ -117,19 +117,15 @@ class DeepgramSTTService(AIService): self._connection = self._client.listen.asynclive.v("1") self._connection.on(LiveTranscriptionEvents.Transcript, self._on_message) - self._create_push_task() - async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) - if isinstance(frame, StartInterruptionFrame): - await self._handle_interruptions(frame) - elif isinstance(frame, SystemFrame): + if isinstance(frame, SystemFrame): await self.push_frame(frame, direction) elif isinstance(frame, AudioRawFrame): await self._connection.send(frame.audio) else: - await self._push_queue.put((frame, direction)) + await self.queue_frame(frame, direction) async def start(self, frame: StartFrame): if await self._connection.start(self._live_options): @@ -139,37 +135,9 @@ class DeepgramSTTService(AIService): async def stop(self, frame: EndFrame): await self._connection.finish() - await self._push_queue.put((frame, FrameDirection.DOWNSTREAM)) - await self._push_frame_task async def cancel(self, frame: CancelFrame): await self._connection.finish() - self._push_frame_task.cancel() - await self._push_frame_task - - async def _handle_interruptions(self, frame: Frame): - # Cancel the task. This will stop pushing frames downstream. - self._push_frame_task.cancel() - await self._push_frame_task - # Push an out-of-band frame (i.e. not using the ordered push - # frame task). - await self.push_frame(frame) - # Create a new queue and task. - self._create_push_task() - - def _create_push_task(self): - self._push_queue = asyncio.Queue() - self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler()) - - async def _push_frame_task_handler(self): - running = True - while running: - try: - (frame, direction) = await self._push_queue.get() - await self.push_frame(frame, direction) - running = not isinstance(frame, EndFrame) - except asyncio.CancelledError: - break async def _on_message(self, *args, **kwargs): result = kwargs["result"] @@ -177,6 +145,6 @@ class DeepgramSTTService(AIService): transcript = result.channel.alternatives[0].transcript if len(transcript) > 0: if is_final: - await self._push_queue.put((TranscriptionFrame(transcript, "", int(time.time_ns() / 1000000)), FrameDirection.DOWNSTREAM)) + await self.queue_frame(TranscriptionFrame(transcript, "", int(time.time_ns() / 1000000))) else: - await self._push_queue.put((InterimTranscriptionFrame(transcript, "", int(time.time_ns() / 1000000)), FrameDirection.DOWNSTREAM)) + await self.queue_frame(InterimTranscriptionFrame(transcript, "", int(time.time_ns() / 1000000)))