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/transports/base_output.py b/src/pipecat/transports/base_output.py index 0ae79b9f0..afb93ebf8 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -325,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) @@ -396,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]: diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 7f872c1fb..0c83bcc13 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()