wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

try to use queue_frame() instead of process_frame()

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-11-06 14:18:21 -08:00
parent 5353d13151
commit 7071482583
7 changed files with 16 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
examples/patient-intake/bot.py
View File

@@ -182,7 +182,7 @@ class IntakeProcessor:
} }
) )
print(f"!!! about to await llm process frame in start prescrpitions") 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") print(f"!!! past await process frame in start prescriptions")
async def start_allergies(self, function_name, llm, context): 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.", "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): async def start_conditions(self, function_name, llm, context):
print("!!! doing start conditions") 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.", "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): async def start_visit_reasons(self, function_name, llm, context):
print("!!! doing start visit reasons") print("!!! doing start visit reasons")
@@ -270,7 +270,7 @@ class IntakeProcessor:
context.add_message( context.add_message(
{"role": "system", "content": "Now, thank the user and end the conversation."} {"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): async def save_data(self, function_name, tool_call_id, args, llm, context, result_callback):
logger.info(f"!!! Saving data: {args}") logger.info(f"!!! Saving data: {args}")

4
src/pipecat/pipeline/parallel_pipeline.py
View File

@@ -110,13 +110,13 @@ class ParallelPipeline(BasePipeline):
if direction == FrameDirection.UPSTREAM: if direction == FrameDirection.UPSTREAM:
# If we get an upstream frame we process it in each sink. # 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: elif direction == FrameDirection.DOWNSTREAM:
# If we get a downstream frame we process it in each source. # If we get a downstream frame we process it in each source.
# TODO(aleix): We are creating task for each frame. For real-time # TODO(aleix): We are creating task for each frame. For real-time
# video/audio this might be too slow. We should use an already # video/audio this might be too slow. We should use an already
# created task instead. # 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 # If we get an EndFrame we stop our queue processing tasks and wait on
# all the pipelines to finish. # all the pipelines to finish.

4
src/pipecat/pipeline/pipeline.py
View File

@@ -77,9 +77,9 @@ class Pipeline(BasePipeline):
await super().process_frame(frame, direction) await super().process_frame(frame, direction)
if direction == FrameDirection.DOWNSTREAM: if direction == FrameDirection.DOWNSTREAM:
await self._source.process_frame(frame, FrameDirection.DOWNSTREAM) await self._source.queue_frame(frame, FrameDirection.DOWNSTREAM)
elif direction == FrameDirection.UPSTREAM: 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): async def _cleanup_processors(self):
for p in self._processors: for p in self._processors:

8
src/pipecat/pipeline/task.py
View File

@@ -160,19 +160,17 @@ class PipelineTask:
report_only_initial_ttfb=self._params.report_only_initial_ttfb, report_only_initial_ttfb=self._params.report_only_initial_ttfb,
clock=self._clock, 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: 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 running = True
should_cleanup = True should_cleanup = True
while running: while running:
try: try:
frame = await self._push_queue.get() 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): if isinstance(frame, EndFrame):
await self._wait_for_endframe() await self._wait_for_endframe()
running = not isinstance(frame, (StopTaskFrame, EndFrame)) running = not isinstance(frame, (StopTaskFrame, EndFrame))

4
src/pipecat/transports/base_output.py
View File

@@ -325,7 +325,7 @@ class BaseOutputTransport(FrameProcessor):
# #
async def send_image(self, frame: OutputImageRawFrame | SpriteFrame): 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): async def _draw_image(self, frame: OutputImageRawFrame):
desired_size = (self._params.camera_out_width, self._params.camera_out_height) 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): 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]: def _next_audio_frame(self) -> AsyncGenerator[AudioRawFrame, None]:
async def without_mixer(vad_stop_secs: float) -> AsyncGenerator[AudioRawFrame, None]: async def without_mixer(vad_stop_secs: float) -> AsyncGenerator[AudioRawFrame, None]:

4
src/pipecat/transports/services/daily.py
View File

@@ -890,11 +890,11 @@ class DailyTransport(BaseTransport):
async def send_image(self, frame: OutputImageRawFrame | SpriteFrame): async def send_image(self, frame: OutputImageRawFrame | SpriteFrame):
if self._output: 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): async def send_audio(self, frame: OutputAudioRawFrame):
if self._output: if self._output:
await self._output.process_frame(frame, FrameDirection.DOWNSTREAM) await self._output.queue_frame(frame, FrameDirection.DOWNSTREAM)
def participants(self): def participants(self):
return self._client.participants() return self._client.participants()

2
src/pipecat/transports/services/livekit.py
View File

@@ -495,7 +495,7 @@ class LiveKitTransport(BaseTransport):
async def send_audio(self, frame: OutputAudioRawFrame): async def send_audio(self, frame: OutputAudioRawFrame):
if self._output: 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]: def get_participants(self) -> List[str]:
return self._client.get_participants() return self._client.get_participants()