Merge pull request #642 from pipecat-ai/aleix/input-queues-block-frames

introduce frame processor input queues block frames

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

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}!"))
+            await task.queue_frames(
-
+                [TTSSpeakFrame(f"Hello there, {participant_name}!"), EndFrame()]
-        # 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 runner.run(task)
 

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(
+    tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
+        api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        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():
+    async def say_something():
-            await asyncio.sleep(1)
+        await asyncio.sleep(1)
-            await task.queue_frame(TextFrame("Hello there!"))
+        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__":

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))
+            await task.queue_frames([LLMMessagesFrame(messages), EndFrame()])
-
-        @transport.event_handler("on_participant_left")
-        async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
 
         await runner.run(task)
 

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

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.

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:

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(
+            await self._source.queue_frame(self._initial_metrics_frame(), FrameDirection.DOWNSTREAM)
-                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))

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.
+        # Cancel the push frame task. This will stop pushing frames downstream.
-        self.__push_frame_task.cancel()
+        await self.__cancel_push_task()
-        await self.__push_frame_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:

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
+        await self.queue_frame(TTSSpeakFrame(text))
-        self._aggregate_sentences = False
-        await self.process_frame(TextFrame(text=text), FrameDirection.DOWNSTREAM)
-        self._aggregate_sentences = aggregate_sentences
-        await self.flush_audio()
 
     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
+                if frame:
-                await self.push_frame(frame)
+                    last_pts = frame.pts
+                    await self.push_frame(frame)
                 self._words_queue.task_done()
             except asyncio.CancelledError:
                 break

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

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:

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

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.

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

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