diff --git a/CHANGELOG.md b/CHANGELOG.md index 0de73ed95..f4b0f4e0e 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 +- There's now an input queue in each frame processor. When you call + `FrameProcessor.push_frame()` this will internally call + `FrameProcessor.queue_frame()` on the next processor (upstream or downstream) + and the frame will be internally queued (except system frames). Then, the + queued frames will get processed. With this input queue it is also possible + for FrameProcessors to block processing more frames by calling + `FrameProcessor.pause_processing_frames()`. The way to resume processing + frames is by calling `FrameProcessor.resume_processing_frames()`. + - Added audio filter `NoisereduceFilter`. - Introduce input transport audio filters (`BaseAudioFilter`). Audio filters can diff --git a/examples/foundational/01-say-one-thing.py b/examples/foundational/01-say-one-thing.py index 50ae1bad2..f0fec28d8 100644 --- a/examples/foundational/01-say-one-thing.py +++ b/examples/foundational/01-say-one-thing.py @@ -9,11 +9,11 @@ import aiohttp import os import sys -from pipecat.frames.frames import EndFrame, TextFrame +from pipecat.frames.frames import EndFrame, TTSSpeakFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.task import PipelineTask from pipecat.pipeline.runner import PipelineRunner -from pipecat.services.cartesia import CartesiaHttpTTSService +from pipecat.services.cartesia import CartesiaTTSService from pipecat.transports.services.daily import DailyParams, DailyTransport from runner import configure @@ -36,7 +36,7 @@ async def main(): room_url, None, "Say One Thing", DailyParams(audio_out_enabled=True) ) - tts = CartesiaHttpTTSService( + tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady ) @@ -50,12 +50,9 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): participant_name = participant.get("info", {}).get("userName", "") - await task.queue_frame(TextFrame(f"Hello there, {participant_name}!")) - - # Register an event handler to exit the application when the user leaves. - @transport.event_handler("on_participant_left") - async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.queue_frames( + [TTSSpeakFrame(f"Hello there, {participant_name}!"), EndFrame()] + ) await runner.run(task) diff --git a/examples/foundational/01a-local-audio.py b/examples/foundational/01a-local-audio.py index d39e922d7..e62ab1020 100644 --- a/examples/foundational/01a-local-audio.py +++ b/examples/foundational/01a-local-audio.py @@ -9,7 +9,7 @@ import aiohttp import os import sys -from pipecat.frames.frames import TextFrame +from pipecat.frames.frames import EndFrame, TTSSpeakFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask @@ -28,25 +28,24 @@ logger.add(sys.stderr, level="DEBUG") async def main(): - async with aiohttp.ClientSession() as session: - transport = LocalAudioTransport(TransportParams(audio_out_enabled=True)) + transport = LocalAudioTransport(TransportParams(audio_out_enabled=True)) - tts = CartesiaTTSService( - api_key=os.getenv("CARTESIA_API_KEY"), - voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady - ) + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) - pipeline = Pipeline([tts, transport.output()]) + pipeline = Pipeline([tts, transport.output()]) - task = PipelineTask(pipeline) + task = PipelineTask(pipeline) - async def say_something(): - await asyncio.sleep(1) - await task.queue_frame(TextFrame("Hello there!")) + async def say_something(): + await asyncio.sleep(1) + await task.queue_frames([TTSSpeakFrame("Hello there, how is it going!"), EndFrame()]) - runner = PipelineRunner() + runner = PipelineRunner() - await asyncio.gather(runner.run(task), say_something()) + await asyncio.gather(runner.run(task), say_something()) if __name__ == "__main__": diff --git a/examples/foundational/02-llm-say-one-thing.py b/examples/foundational/02-llm-say-one-thing.py index 891628415..ba2be607c 100644 --- a/examples/foundational/02-llm-say-one-thing.py +++ b/examples/foundational/02-llm-say-one-thing.py @@ -13,7 +13,7 @@ from pipecat.frames.frames import EndFrame, LLMMessagesFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask -from pipecat.services.cartesia import CartesiaHttpTTSService +from pipecat.services.cartesia import CartesiaTTSService from pipecat.services.openai import OpenAILLMService from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -37,7 +37,7 @@ async def main(): room_url, None, "Say One Thing From an LLM", DailyParams(audio_out_enabled=True) ) - tts = CartesiaHttpTTSService( + tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady ) @@ -57,11 +57,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - await task.queue_frame(LLMMessagesFrame(messages)) - - @transport.event_handler("on_participant_left") - async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.queue_frames([LLMMessagesFrame(messages), EndFrame()]) await runner.run(task) diff --git a/examples/patient-intake/bot.py b/examples/patient-intake/bot.py index adb3785e4..c33a2495d 100644 --- a/examples/patient-intake/bot.py +++ b/examples/patient-intake/bot.py @@ -182,7 +182,7 @@ class IntakeProcessor: } ) print(f"!!! about to await llm process frame in start prescrpitions") - await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) + await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) print(f"!!! past await process frame in start prescriptions") async def start_allergies(self, function_name, llm, context): @@ -222,7 +222,7 @@ class IntakeProcessor: "content": "Now ask the user if they have any medical conditions the doctor should know about. Once they've answered the question, call the list_conditions function.", } ) - await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) + await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) async def start_conditions(self, function_name, llm, context): print("!!! doing start conditions") @@ -261,7 +261,7 @@ class IntakeProcessor: "content": "Finally, ask the user the reason for their doctor visit today. Once they answer, call the list_visit_reasons function.", } ) - await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) + await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) async def start_visit_reasons(self, function_name, llm, context): print("!!! doing start visit reasons") @@ -270,7 +270,7 @@ class IntakeProcessor: context.add_message( {"role": "system", "content": "Now, thank the user and end the conversation."} ) - await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) + await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) async def save_data(self, function_name, tool_call_id, args, llm, context, result_callback): logger.info(f"!!! Saving data: {args}") diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py index da68212ff..ad3ebd8dd 100644 --- a/src/pipecat/pipeline/parallel_pipeline.py +++ b/src/pipecat/pipeline/parallel_pipeline.py @@ -110,13 +110,13 @@ class ParallelPipeline(BasePipeline): 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]) + await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sinks]) elif direction == FrameDirection.DOWNSTREAM: # If we get a downstream frame we process it in each source. # TODO(aleix): We are creating task for each frame. For real-time # video/audio this might be too slow. We should use an already # created task instead. - await asyncio.gather(*[s.process_frame(frame, direction) for s in self._sources]) + await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sources]) # If we get an EndFrame we stop our queue processing tasks and wait on # all the pipelines to finish. diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py index a1715570e..703f911fe 100644 --- a/src/pipecat/pipeline/pipeline.py +++ b/src/pipecat/pipeline/pipeline.py @@ -77,9 +77,9 @@ class Pipeline(BasePipeline): await super().process_frame(frame, direction) if direction == FrameDirection.DOWNSTREAM: - await self._source.process_frame(frame, FrameDirection.DOWNSTREAM) + await self._source.queue_frame(frame, FrameDirection.DOWNSTREAM) elif direction == FrameDirection.UPSTREAM: - await self._sink.process_frame(frame, FrameDirection.UPSTREAM) + await self._sink.queue_frame(frame, FrameDirection.UPSTREAM) async def _cleanup_processors(self): for p in self._processors: diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index e34ccae1b..f09013a58 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -160,19 +160,17 @@ class PipelineTask: report_only_initial_ttfb=self._params.report_only_initial_ttfb, clock=self._clock, ) - await self._source.process_frame(start_frame, FrameDirection.DOWNSTREAM) + await self._source.queue_frame(start_frame, FrameDirection.DOWNSTREAM) if self._params.enable_metrics and self._params.send_initial_empty_metrics: - await self._source.process_frame( - self._initial_metrics_frame(), FrameDirection.DOWNSTREAM - ) + await self._source.queue_frame(self._initial_metrics_frame(), FrameDirection.DOWNSTREAM) running = True should_cleanup = True while running: try: frame = await self._push_queue.get() - await self._source.process_frame(frame, FrameDirection.DOWNSTREAM) + await self._source.queue_frame(frame, FrameDirection.DOWNSTREAM) if isinstance(frame, EndFrame): await self._wait_for_endframe() running = not isinstance(frame, (StopTaskFrame, EndFrame)) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index f458f43ff..9f614ba85 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -8,6 +8,7 @@ import asyncio import inspect from enum import Enum +from typing import Awaitable, Callable, Optional from pipecat.clocks.base_clock import BaseClock from pipecat.frames.frames import ( @@ -62,6 +63,13 @@ class FrameProcessor: self._metrics = metrics or FrameProcessorMetrics() self._metrics.set_processor_name(self.name) + # Processors have an input queue. The input queue will be processed + # immediately (default) or it will block if `pause_processing_frames()` + # is called. To resume processing frames we need to call + # `resume_processing_frames()`. + self.__should_block_frames = False + self.__create_input_task() + # Every processor in Pipecat should only output frames from a single # task. This avoid problems like audio overlapping. System frames are # the exception to this rule. This create this task. @@ -126,7 +134,8 @@ class FrameProcessor: await self.stop_processing_metrics() async def cleanup(self): - pass + await self.__cancel_input_task() + await self.__cancel_push_task() def link(self, processor: "FrameProcessor"): self._next = processor @@ -145,6 +154,28 @@ class FrameProcessor: def get_clock(self) -> BaseClock: return self._clock + async def queue_frame( + self, + frame: Frame, + direction: FrameDirection = FrameDirection.DOWNSTREAM, + callback: Optional[ + Callable[["FrameProcessor", Frame, FrameDirection], Awaitable[None]] + ] = None, + ): + if isinstance(frame, SystemFrame): + # We don't want to queue system frames. + await self.process_frame(frame, direction) + else: + # We queue everything else. + await self.__input_queue.put((frame, direction, callback)) + + async def pause_processing_frames(self): + self.__should_block_frames = True + + async def resume_processing_frames(self): + self.__input_event.set() + self.__should_block_frames = False + async def process_frame(self, frame: Frame, direction: FrameDirection): if isinstance(frame, StartFrame): self._clock = frame.clock @@ -189,11 +220,16 @@ class FrameProcessor: # async def _start_interruption(self): - # Cancel the task. This will stop pushing frames downstream. - self.__push_frame_task.cancel() - await self.__push_frame_task + # Cancel the push frame task. This will stop pushing frames downstream. + await self.__cancel_push_task() - # Create a new queue and task. + # Cancel the input task. This will stop processing queued frames. + await self.__cancel_input_task() + + # Create a new input queue and task. + self.__create_input_task() + + # Create a new output queue and task. self.__create_push_task() async def _stop_interruption(self): @@ -204,17 +240,55 @@ class FrameProcessor: try: if direction == FrameDirection.DOWNSTREAM and self._next: logger.trace(f"Pushing {frame} from {self} to {self._next}") - await self._next.process_frame(frame, direction) + await self._next.queue_frame(frame, direction) elif direction == FrameDirection.UPSTREAM and self._prev: logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}") - await self._prev.process_frame(frame, direction) + await self._prev.queue_frame(frame, direction) except Exception as e: logger.exception(f"Uncaught exception in {self}: {e}") + def __create_input_task(self): + self.__input_queue = asyncio.Queue() + self.__input_frame_task = self.get_event_loop().create_task( + self.__input_frame_task_handler() + ) + self.__input_event = asyncio.Event() + + async def __cancel_input_task(self): + self.__input_frame_task.cancel() + await self.__input_frame_task + + async def __input_frame_task_handler(self): + running = True + while running: + try: + if self.__should_block_frames: + await self.__input_event.wait() + self.__input_event.clear() + + (frame, direction, callback) = await self.__input_queue.get() + + # Process the frame. + await self.process_frame(frame, direction) + + # If this frame has an associated callback, call it now. + if callback: + await callback(self, frame, direction) + + running = not isinstance(frame, EndFrame) + + self.__input_queue.task_done() + except asyncio.CancelledError: + break + 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 __cancel_push_task(self): + self.__push_frame_task.cancel() + await self.__push_frame_task + async def __push_frame_task_handler(self): running = True while running: diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 62d52f76e..f7a08802c 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -284,11 +284,7 @@ class TTSService(AIService): logger.warning(f"Unknown setting for TTS service: {key}") async def say(self, text: str): - aggregate_sentences = self._aggregate_sentences - self._aggregate_sentences = False - await self.process_frame(TextFrame(text=text), FrameDirection.DOWNSTREAM) - self._aggregate_sentences = aggregate_sentences - await self.flush_audio() + await self.queue_frame(TTSSpeakFrame(text)) async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -395,7 +391,6 @@ class WordTTSService(TTSService): def reset_word_timestamps(self): self._initial_word_timestamp = -1 - self._word_timestamps = [] async def add_word_timestamps(self, word_times: List[Tuple[str, float]]): for word, timestamp in word_times: @@ -430,7 +425,10 @@ class WordTTSService(TTSService): while True: try: (word, timestamp) = await self._words_queue.get() - if word == "LLMFullResponseEndFrame" and timestamp == 0: + if word == "Reset" and timestamp == 0: + self.reset_word_timestamps() + frame = None + elif word == "LLMFullResponseEndFrame" and timestamp == 0: frame = LLMFullResponseEndFrame() frame.pts = last_pts elif word == "TTSStoppedFrame" and timestamp == 0: @@ -439,8 +437,9 @@ class WordTTSService(TTSService): else: frame = TextFrame(word) frame.pts = self._initial_word_timestamp + timestamp - last_pts = frame.pts - await self.push_frame(frame) + if frame: + last_pts = frame.pts + await self.push_frame(frame) self._words_queue.task_done() except asyncio.CancelledError: break diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 35f91b4cd..ff7b200c1 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -14,13 +14,16 @@ from loguru import logger from pydantic.main import BaseModel from pipecat.frames.frames import ( + BotStoppedSpeakingFrame, CancelFrame, EndFrame, ErrorFrame, Frame, + LLMFullResponseEndFrame, StartFrame, StartInterruptionFrame, TTSAudioRawFrame, + TTSSpeakFrame, TTSStartedFrame, TTSStoppedFrame, ) @@ -231,7 +234,7 @@ class CartesiaTTSService(WordTTSService): # timestamp to set send context frames. self._context_id = None await self.add_word_timestamps( - [("TTSStoppedFrame", 0), ("LLMFullResponseEndFrame", 0)] + [("TTSStoppedFrame", 0), ("LLMFullResponseEndFrame", 0), ("Reset", 0)] ) elif msg["type"] == "timestamps": await self.add_word_timestamps( @@ -258,6 +261,19 @@ class CartesiaTTSService(WordTTSService): except Exception as e: logger.error(f"{self} exception: {e}") + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + # If we received a TTSSpeakFrame and the LLM response included text (it + # might be that it's only a function calling response) we pause + # processing more frames until we receive a BotStoppedSpeakingFrame. + if isinstance(frame, TTSSpeakFrame): + await self.pause_processing_frames() + elif isinstance(frame, LLMFullResponseEndFrame) and self._context_id: + await self.pause_processing_frames() + elif isinstance(frame, BotStoppedSpeakingFrame): + await self.resume_processing_frames() + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index b4dee5fa3..9b1f54351 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -13,12 +13,15 @@ from loguru import logger from pydantic import BaseModel, model_validator from pipecat.frames.frames import ( + BotStoppedSpeakingFrame, CancelFrame, EndFrame, Frame, + LLMFullResponseEndFrame, StartFrame, StartInterruptionFrame, TTSAudioRawFrame, + TTSSpeakFrame, TTSStartedFrame, TTSStoppedFrame, ) @@ -283,7 +286,20 @@ class ElevenLabsTTSService(WordTTSService): if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): self._started = False if isinstance(frame, TTSStoppedFrame): - await self.add_word_timestamps([("LLMFullResponseEndFrame", 0)]) + await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)]) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + # If we received a TTSSpeakFrame and the LLM response included text (it + # might be that it's only a function calling response) we pause + # processing more frames until we receive a BotStoppedSpeakingFrame. + if isinstance(frame, TTSSpeakFrame): + await self.pause_processing_frames() + elif isinstance(frame, LLMFullResponseEndFrame) and self._started: + await self.pause_processing_frames() + elif isinstance(frame, BotStoppedSpeakingFrame): + await self.resume_processing_frames() async def _connect(self): try: diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index 4837c058f..167181586 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -17,13 +17,16 @@ from loguru import logger from pydantic.main import BaseModel from pipecat.frames.frames import ( + BotStoppedSpeakingFrame, CancelFrame, EndFrame, ErrorFrame, Frame, + LLMFullResponseEndFrame, StartFrame, StartInterruptionFrame, TTSAudioRawFrame, + TTSSpeakFrame, TTSStartedFrame, TTSStoppedFrame, ) @@ -121,7 +124,10 @@ class PlayHTTTSService(TTSService): params: InputParams = InputParams(), **kwargs, ): - super().__init__(sample_rate=sample_rate, **kwargs) + super().__init__( + sample_rate=sample_rate, + **kwargs, + ) self._api_key = api_key self._user_id = user_id @@ -260,6 +266,19 @@ class PlayHTTTSService(TTSService): except Exception as e: logger.error(f"{self} exception in receive task: {e}") + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + # If we received a TTSSpeakFrame and the LLM response included text (it + # might be that it's only a function calling response) we pause + # processing more frames until we receive a BotStoppedSpeakingFrame. + if isinstance(frame, TTSSpeakFrame): + await self.pause_processing_frames() + elif isinstance(frame, LLMFullResponseEndFrame) and self._request_id: + await self.pause_processing_frames() + elif isinstance(frame, BotStoppedSpeakingFrame): + await self.resume_processing_frames() + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index e6c006145..afb93ebf8 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -208,12 +208,14 @@ class BaseOutputTransport(FrameProcessor): async def _bot_started_speaking(self): if not self._bot_speaking: logger.debug("Bot started speaking") + await self.push_frame(BotStartedSpeakingFrame()) await self.push_frame(BotStartedSpeakingFrame(), FrameDirection.UPSTREAM) self._bot_speaking = True async def _bot_stopped_speaking(self): if self._bot_speaking: logger.debug("Bot stopped speaking") + await self.push_frame(BotStoppedSpeakingFrame()) await self.push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM) self._bot_speaking = False @@ -323,7 +325,7 @@ class BaseOutputTransport(FrameProcessor): # async def send_image(self, frame: OutputImageRawFrame | SpriteFrame): - await self.process_frame(frame, FrameDirection.DOWNSTREAM) + await self.queue_frame(frame, FrameDirection.DOWNSTREAM) async def _draw_image(self, frame: OutputImageRawFrame): desired_size = (self._params.camera_out_width, self._params.camera_out_height) @@ -394,7 +396,7 @@ class BaseOutputTransport(FrameProcessor): # async def send_audio(self, frame: OutputAudioRawFrame): - await self.process_frame(frame, FrameDirection.DOWNSTREAM) + await self.queue_frame(frame, FrameDirection.DOWNSTREAM) def _next_audio_frame(self) -> AsyncGenerator[AudioRawFrame, None]: async def without_mixer(vad_stop_secs: float) -> AsyncGenerator[AudioRawFrame, None]: @@ -452,6 +454,7 @@ class BaseOutputTransport(FrameProcessor): # it's actually speaking. if isinstance(frame, TTSAudioRawFrame): await self._bot_started_speaking() + await self.push_frame(BotSpeakingFrame()) await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM) # Also, push frame downstream in case anyone else needs it. diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 67d551478..1eb83c370 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -890,11 +890,11 @@ class DailyTransport(BaseTransport): async def send_image(self, frame: OutputImageRawFrame | SpriteFrame): if self._output: - await self._output.process_frame(frame, FrameDirection.DOWNSTREAM) + await self._output.queue_frame(frame, FrameDirection.DOWNSTREAM) async def send_audio(self, frame: OutputAudioRawFrame): if self._output: - await self._output.process_frame(frame, FrameDirection.DOWNSTREAM) + await self._output.queue_frame(frame, FrameDirection.DOWNSTREAM) def participants(self): return self._client.participants() diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index 6b6b55c54..5d4fbdd6c 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -495,7 +495,7 @@ class LiveKitTransport(BaseTransport): async def send_audio(self, frame: OutputAudioRawFrame): if self._output: - await self._output.process_frame(frame, FrameDirection.DOWNSTREAM) + await self._output.queue_frame(frame, FrameDirection.DOWNSTREAM) def get_participants(self) -> List[str]: return self._client.get_participants()