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

Merge pull request #1285 from pipecat-ai/aleix/handle-stop-task-gracefully

Browse Source 
handle stop task gracefully
This commit is contained in:
Aleix Conchillo Flaqué
2025-02-25 11:25:38 -08:00
committed by GitHub
parent 12092c8715 34a50033cb
commit 4bd5e9c0a7
41 changed files with 396 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
CHANGELOG.md
View File

@@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- Added a new `StopFrame` which can be used to stop a pipeline task while
keeping the frame processors running. The frame processors could then be used
in a different pipeline. The difference between a `StopFrame` and a
`StopTaskFrame` is that, as with `EndFrame` and `EndTaskFrame`, the
`StopFrame` is pushed from the task and the `StopTaskFrame` is pushed upstream
inside the pipeline by any processor.
- Added a new `PipelineTask` parameter `observers` that replaces the previous
`PipelineParams.observers`.

7
examples/foundational/03a-local-still-frame.py
View File

@@ -18,8 +18,7 @@ from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.services.fal import FalImageGenService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.local.tk import TkLocalTransport
from pipecat.transports.local.tk import TkLocalTransport, TkTransportParams
load_dotenv(override=True)
@@ -34,7 +33,9 @@ async def main():
transport = TkLocalTransport(
tk_root,
TransportParams(camera_out_enabled=True, camera_out_width=1024, camera_out_height=1024),
TkTransportParams(
camera_out_enabled=True, camera_out_width=1024, camera_out_height=1024
),
)
imagegen = FalImageGenService(

5
examples/foundational/05a-local-sync-speech-and-image.py
View File

@@ -30,8 +30,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.cartesia import CartesiaHttpTTSService
from pipecat.services.fal import FalImageGenService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.local.tk import TkLocalTransport, TkOutputTransport
from pipecat.transports.local.tk import TkLocalTransport, TkTransportParams
load_dotenv(override=True)
@@ -152,7 +151,7 @@ async def main():
transport = TkLocalTransport(
tk_root,
TransportParams(
TkTransportParams(
audio_out_enabled=True,
camera_out_enabled=True,
camera_out_width=1024,

5
examples/foundational/09a-local-mirror.py
View File

@@ -24,8 +24,7 @@ from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.local.tk import TkLocalTransport
from pipecat.transports.local.tk import TkLocalTransport, TkTransportParams
from pipecat.transports.services.daily import DailyParams, DailyTransport
load_dotenv(override=True)
@@ -67,7 +66,7 @@ async def main():
tk_transport = TkLocalTransport(
tk_root,
TransportParams(
TkTransportParams(
audio_out_enabled=True,
camera_out_enabled=True,
camera_out_is_live=True,

8
examples/foundational/22b-natural-conversation-proposal.py
View File

@@ -138,6 +138,7 @@ class OutputGate(FrameProcessor):
self._gate_open = start_open
self._frames_buffer = []
self._notifier = notifier
self._gate_task = None
def close_gate(self):
self._gate_open = False
@@ -178,10 +179,13 @@ class OutputGate(FrameProcessor):
async def _start(self):
self._frames_buffer = []
self._gate_task = self.create_task(self._gate_task_handler())
if not self._gate_task:
self._gate_task = self.create_task(self._gate_task_handler())
async def _stop(self):
await self.cancel_task(self._gate_task)
if self._gate_task:
await self.cancel_task(self._gate_task)
self._gate_task = None
async def _gate_task_handler(self):
while True:

8
examples/foundational/22c-natural-conversation-mixed-llms.py
View File

@@ -342,6 +342,7 @@ class OutputGate(FrameProcessor):
self._gate_open = start_open
self._frames_buffer = []
self._notifier = notifier
self._gate_task = None
def close_gate(self):
self._gate_open = False
@@ -382,10 +383,13 @@ class OutputGate(FrameProcessor):
async def _start(self):
self._frames_buffer = []
self._gate_task = self.create_task(self._gate_task_handler())
if not self._gate_task:
self._gate_task = self.create_task(self._gate_task_handler())
async def _stop(self):
await self.cancel_task(self._gate_task)
if self._gate_task:
await self.cancel_task(self._gate_task)
self._gate_task = None
async def _gate_task_handler(self):
while True:

10
examples/foundational/22d-natural-conversation-gemini-audio.py
View File

@@ -25,10 +25,8 @@ from pipecat.frames.frames import (
InputAudioRawFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesFrame,
StartFrame,
StartInterruptionFrame,
StopInterruptionFrame,
SystemFrame,
TextFrame,
TranscriptionFrame,
@@ -555,6 +553,7 @@ class OutputGate(FrameProcessor):
self._notifier = notifier
self._context = context
self._transcription_buffer = user_transcription_buffer
self._gate_task = None
def close_gate(self):
self._gate_open = False
@@ -602,10 +601,13 @@ class OutputGate(FrameProcessor):
async def _start(self):
self._frames_buffer = []
self._gate_task = self.create_task(self._gate_task_handler())
if not self._gate_task:
self._gate_task = self.create_task(self._gate_task_handler())
async def _stop(self):
await self.cancel_task(self._gate_task)
if self._gate_task:
await self.cancel_task(self._gate_task)
self._gate_task = None
async def _gate_task_handler(self):
while True:

21
src/pipecat/frames/frames.py
View File

@@ -23,7 +23,7 @@ from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.clocks.base_clock import BaseClock
from pipecat.metrics.metrics import MetricsData
from pipecat.transcriptions.language import Language
from pipecat.utils.asyncio import TaskManager
from pipecat.utils.asyncio import BaseTaskManager
from pipecat.utils.time import nanoseconds_to_str
from pipecat.utils.utils import obj_count, obj_id
@@ -438,7 +438,7 @@ class StartFrame(SystemFrame):
"""This is the first frame that should be pushed down a pipeline."""
clock: BaseClock
task_manager: TaskManager
task_manager: BaseTaskManager
audio_in_sample_rate: int = 16000
audio_out_sample_rate: int = 24000
allow_interruptions: bool = False
@@ -513,9 +513,9 @@ class CancelTaskFrame(SystemFrame):
@dataclass
class StopTaskFrame(SystemFrame):
"""Indicates that a pipeline task should be stopped but that the pipeline
processors should be kept in a running state. This is normally queued from
the pipeline task.
"""This is used to notify the pipeline task that it should be stopped as
soon as possible (flushing all the queued frames) but that the pipeline
processors should be kept in a running state.
"""
@@ -722,6 +722,17 @@ class EndFrame(ControlFrame):
pass
@dataclass
class StopFrame(ControlFrame):
"""Indicates that a pipeline should be stopped but that the pipeline
processors should be kept in a running state. This is normally queued from
the pipeline task.
"""
pass
@dataclass
class LLMFullResponseStartFrame(ControlFrame):
"""Used to indicate the beginning of an LLM response. Following by one or

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

@@ -81,6 +81,8 @@ class ParallelPipeline(BasePipeline):
self._seen_ids = set()
self._endframe_counter: Dict[int, int] = {}
self._up_task = None
self._down_task = None
self._up_queue = asyncio.Queue()
self._down_queue = asyncio.Queue()
@@ -150,19 +152,30 @@ class ParallelPipeline(BasePipeline):
await self._create_tasks()
async def _stop(self):
# The up task doesn't receive an EndFrame, so we just cancel it.
await self.cancel_task(self._up_task)
# The down tasks waits for the last EndFrame sent by the internal
# pipelines.
await self._down_task
if self._up_task:
# The up task doesn't receive an EndFrame, so we just cancel it.
await self.cancel_task(self._up_task)
self._up_task = None
if self._down_task:
# The down tasks waits for the last EndFrame sent by the internal
# pipelines.
await self._down_task
self._down_task = None
async def _cancel(self):
await self.cancel_task(self._up_task)
await self.cancel_task(self._down_task)
if self._up_task:
await self.cancel_task(self._up_task)
self._up_task = None
if self._down_task:
await self.cancel_task(self._down_task)
self._down_task = None
async def _create_tasks(self):
self._up_task = self.create_task(self._process_up_queue())
self._down_task = self.create_task(self._process_down_queue())
if not self._up_task:
self._up_task = self.create_task(self._process_up_queue())
if not self._down_task:
self._down_task = self.create_task(self._process_down_queue())
async def _drain_queues(self):
while not self._up_queue.empty:

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

@@ -5,7 +5,7 @@
#
import asyncio
from typing import Any, AsyncIterable, Dict, Iterable, List
from typing import Any, AsyncIterable, Dict, Iterable, List, Optional
from loguru import logger
from pydantic import BaseModel, ConfigDict
@@ -22,6 +22,7 @@ from pipecat.frames.frames import (
HeartbeatFrame,
MetricsFrame,
StartFrame,
StopFrame,
StopTaskFrame,
)
from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData
@@ -30,7 +31,7 @@ from pipecat.pipeline.base_pipeline import BasePipeline
from pipecat.pipeline.base_task import BaseTask
from pipecat.pipeline.task_observer import TaskObserver
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.asyncio import TaskManager
from pipecat.utils.asyncio import BaseTaskManager, TaskManager
from pipecat.utils.utils import obj_count, obj_id
HEARTBEAT_SECONDS = 1.0
@@ -133,6 +134,7 @@ class PipelineTask(BaseTask):
params: PipelineParams = PipelineParams(),
observers: List[BaseObserver] = [],
clock: BaseClock = SystemClock(),
task_manager: Optional[BaseTaskManager] = None,
check_dangling_tasks: bool = True,
):
self._id: int = obj_id()
@@ -163,9 +165,9 @@ class PipelineTask(BaseTask):
# This is the heartbeat queue. When a heartbeat frame is received in the
# down queue we add it to the heartbeat queue for processing.
self._heartbeat_queue = asyncio.Queue()
# This event is used to indicate an EndFrame has been received in the
# down queue.
self._endframe_event = asyncio.Event()
# This event is used to indicate a finalize frame (e.g. EndFrame,
# StopFrame) has been received in the down queue.
self._pipeline_end_event = asyncio.Event()
self._source = PipelineTaskSource(self._up_queue)
self._source.link(pipeline)
@@ -173,7 +175,7 @@ class PipelineTask(BaseTask):
self._sink = PipelineTaskSink(self._down_queue)
pipeline.link(self._sink)
self._task_manager = TaskManager()
self._task_manager = task_manager or TaskManager()
self._observer = TaskObserver(observers=observers, task_manager=self._task_manager)
@@ -224,9 +226,12 @@ class PipelineTask(BaseTask):
"""Starts and manages the pipeline execution until completion or cancellation."""
if self.has_finished():
return
cleanup_pipeline = True
try:
push_task = await self._create_tasks()
await self._task_manager.wait_for_task(push_task)
# We have already cleaned up the pipeline inside the task.
cleanup_pipeline = False
except asyncio.CancelledError:
# We are awaiting on the push task and it might be cancelled
# (e.g. Ctrl-C). This means we will get a CancelledError here as
@@ -234,7 +239,7 @@ class PipelineTask(BaseTask):
# awaiting a task.
pass
await self._cancel_tasks()
await self._cleanup()
await self._cleanup(cleanup_pipeline)
if self._check_dangling_tasks:
self._print_dangling_tasks()
self._finished = True
@@ -305,13 +310,14 @@ class PipelineTask(BaseTask):
data.append(ProcessingMetricsData(processor=p.name, value=0.0))
return MetricsFrame(data=data)
async def _wait_for_endframe(self):
await self._endframe_event.wait()
self._endframe_event.clear()
async def _wait_for_pipeline_end(self):
await self._pipeline_end_event.wait()
self._pipeline_end_event.clear()
async def _cleanup(self):
async def _cleanup(self, cleanup_pipeline: bool):
await self._source.cleanup()
await self._pipeline.cleanup()
if cleanup_pipeline:
await self._pipeline.cleanup()
await self._sink.cleanup()
async def _process_push_queue(self):
@@ -342,18 +348,16 @@ class PipelineTask(BaseTask):
await self._source.queue_frame(self._initial_metrics_frame(), FrameDirection.DOWNSTREAM)
running = True
should_cleanup = True
cleanup_pipeline = True
while running:
frame = await self._push_queue.get()
await self._source.queue_frame(frame, FrameDirection.DOWNSTREAM)
if isinstance(frame, EndFrame):
await self._wait_for_endframe()
running = not isinstance(frame, (CancelFrame, EndFrame, StopTaskFrame))
should_cleanup = not isinstance(frame, StopTaskFrame)
if isinstance(frame, (EndFrame, StopFrame)):
await self._wait_for_pipeline_end()
running = not isinstance(frame, (CancelFrame, EndFrame, StopFrame))
cleanup_pipeline = not isinstance(frame, StopFrame)
self._push_queue.task_done()
# Cleanup only if we need to.
if should_cleanup:
await self._cleanup()
await self._cleanup(cleanup_pipeline)
async def _process_up_queue(self):
"""This is the task that processes frames coming upstream from the
@@ -371,7 +375,8 @@ class PipelineTask(BaseTask):
# Tell the task we should end right away.
await self.queue_frame(CancelFrame())
elif isinstance(frame, StopTaskFrame):
await self.queue_frame(StopTaskFrame())
# Tell the task we should stop nicely.
await self.queue_frame(StopFrame())
elif isinstance(frame, ErrorFrame):
logger.error(f"Error running app: {frame}")
if frame.fatal:
@@ -390,8 +395,8 @@ class PipelineTask(BaseTask):
"""
while True:
frame = await self._down_queue.get()
if isinstance(frame, EndFrame):
self._endframe_event.set()
if isinstance(frame, (EndFrame, StopFrame)):
self._pipeline_end_event.set()
elif isinstance(frame, HeartbeatFrame):
await self._heartbeat_queue.put(frame)
self._down_queue.task_done()

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

@@ -12,7 +12,7 @@ from attr import dataclass
from pipecat.frames.frames import Frame
from pipecat.observers.base_observer import BaseObserver
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.asyncio import TaskManager
from pipecat.utils.asyncio import BaseTaskManager
from pipecat.utils.utils import obj_count, obj_id
@@ -55,7 +55,7 @@ class TaskObserver(BaseObserver):
"""
def __init__(self, *, observers: List[BaseObserver] = [], task_manager: TaskManager):
def __init__(self, *, observers: List[BaseObserver] = [], task_manager: BaseTaskManager):
self._id: int = obj_id()
self._name: str = f"{self.__class__.__name__}#{obj_count(self)}"
self._observers = observers

8
src/pipecat/processors/aggregators/gated_openai_llm_context.py
View File

@@ -21,6 +21,7 @@ class GatedOpenAILLMContextAggregator(FrameProcessor):
self._notifier = notifier
self._start_open = start_open
self._last_context_frame = None
self._gate_task = None
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@@ -41,10 +42,13 @@ class GatedOpenAILLMContextAggregator(FrameProcessor):
await self.push_frame(frame, direction)
async def _start(self):
self._gate_task = self.create_task(self._gate_task_handler())
if not self._gate_task:
self._gate_task = self.create_task(self._gate_task_handler())
async def _stop(self):
await self.cancel_task(self._gate_task)
if self._gate_task:
await self.cancel_task(self._gate_task)
self._gate_task = None
async def _gate_task_handler(self):
while True:

11
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -246,11 +246,15 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
await super().process_frame(frame, direction)
if isinstance(frame, StartFrame):
# Push StartFrame before start(), because we want StartFrame to be
# processed by every processor before any other frame is processed.
await self.push_frame(frame, direction)
await self._start(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, EndFrame):
await self._stop(frame)
# Push EndFrame before stop(), because stop() waits on the task to
# finish and the task finishes when EndFrame is processed.
await self.push_frame(frame, direction)
await self._stop(frame)
elif isinstance(frame, CancelFrame):
await self._cancel(frame)
await self.push_frame(frame, direction)
@@ -309,7 +313,8 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
self._seen_interim_results = True
def _create_aggregation_task(self):
self._aggregation_task = self.create_task(self._aggregation_task_handler())
if not self._aggregation_task:
self._aggregation_task = self.create_task(self._aggregation_task_handler())
async def _cancel_aggregation_task(self):
if self._aggregation_task:

13
src/pipecat/processors/frame_processor.py
View File

@@ -23,7 +23,7 @@ from pipecat.frames.frames import (
)
from pipecat.metrics.metrics import LLMTokenUsage, MetricsData
from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics
from pipecat.utils.asyncio import TaskManager
from pipecat.utils.asyncio import BaseTaskManager
from pipecat.utils.utils import obj_count, obj_id
@@ -52,7 +52,7 @@ class FrameProcessor:
self._clock: Optional[BaseClock] = None
# Task Manager
self._task_manager: Optional[TaskManager] = None
self._task_manager: Optional[BaseTaskManager] = None
# Other properties
self._allow_interruptions = False
@@ -192,7 +192,7 @@ class FrameProcessor:
raise Exception(f"{self} Clock is still not initialized.")
return self._clock
def get_task_manager(self) -> TaskManager:
def get_task_manager(self) -> BaseTaskManager:
if not self._task_manager:
raise Exception(f"{self} TaskManager is still not initialized.")
return self._task_manager
@@ -240,7 +240,7 @@ class FrameProcessor:
elif isinstance(frame, StopInterruptionFrame):
self._should_report_ttfb = True
elif isinstance(frame, CancelFrame):
self._cancelling = True
await self.__cancel(frame)
async def push_error(self, error: ErrorFrame):
await self.push_frame(error, FrameDirection.UPSTREAM)
@@ -275,6 +275,11 @@ class FrameProcessor:
self.__create_input_task()
self.__create_push_task()
async def __cancel(self, frame: CancelFrame):
self._cancelling = True
await self.__cancel_input_task()
await self.__cancel_push_task()
#
# Handle interruptions
#

6
src/pipecat/processors/frameworks/rtvi.py
View File

@@ -956,8 +956,10 @@ class RTVIProcessor(FrameProcessor):
await self._pipeline.cleanup()
async def _start(self, frame: StartFrame):
self._action_task = self.create_task(self._action_task_handler())
self._message_task = self.create_task(self._message_task_handler())
if not self._action_task:
self._action_task = self.create_task(self._action_task_handler())
if not self._message_task:
self._message_task = self.create_task(self._message_task_handler())
await self._call_event_handler("on_bot_started")
async def _stop(self, frame: EndFrame):

9
src/pipecat/processors/idle_frame_processor.py
View File

@@ -30,6 +30,7 @@ class IdleFrameProcessor(FrameProcessor):
self._callback = callback
self._timeout = timeout
self._types = types
self._idle_task = None
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@@ -49,11 +50,13 @@ class IdleFrameProcessor(FrameProcessor):
self._idle_event.set()
async def cleanup(self):
await self.cancel_task(self._idle_task)
if self._idle_task:
await self.cancel_task(self._idle_task)
def _create_idle_task(self):
self._idle_event = asyncio.Event()
self._idle_task = self.create_task(self._idle_task_handler())
if not self._idle_task:
self._idle_event = asyncio.Event()
self._idle_task = self.create_task(self._idle_task_handler())
async def _idle_task_handler(self):
while True:

4
src/pipecat/processors/user_idle_processor.py
View File

@@ -102,7 +102,7 @@ class UserIdleProcessor(FrameProcessor):
def _create_idle_task(self) -> None:
"""Creates the idle task if it hasn't been created yet."""
if self._idle_task is None:
if not self._idle_task:
self._idle_task = self.create_task(self._idle_task_handler())
@property
@@ -112,7 +112,7 @@ class UserIdleProcessor(FrameProcessor):
async def _stop(self) -> None:
"""Stops and cleans up the idle monitoring task."""
if self._idle_task is not None:
if self._idle_task:
await self.cancel_task(self._idle_task)
self._idle_task = None

15
src/pipecat/services/ai_services.py
View File

@@ -273,7 +273,7 @@ class TTSService(AIService):
async def start(self, frame: StartFrame):
await super().start(frame)
self._sample_rate = self._init_sample_rate or frame.audio_out_sample_rate
if self._push_stop_frames:
if self._push_stop_frames and not self._stop_frame_task:
self._stop_frame_task = self.create_task(self._stop_frame_handler())
async def stop(self, frame: EndFrame):
@@ -487,7 +487,8 @@ class WordTTSService(TTSService):
self.reset_word_timestamps()
def _create_words_task(self):
self._words_task = self.create_task(self._words_task_handler())
if not self._words_task:
self._words_task = self.create_task(self._words_task_handler())
async def _stop_words_task(self):
if self._words_task:
@@ -663,9 +664,10 @@ class AudioContextWordTTSService(WebsocketWordTTSService):
self._create_audio_context_task()
def _create_audio_context_task(self):
self._contexts_queue = asyncio.Queue()
self._contexts: Dict[str, asyncio.Queue] = {}
self._audio_context_task = self.create_task(self._audio_context_task_handler())
if not self._audio_context_task:
self._contexts_queue = asyncio.Queue()
self._contexts: Dict[str, asyncio.Queue] = {}
self._audio_context_task = self.create_task(self._audio_context_task_handler())
async def _stop_audio_context_task(self):
if self._audio_context_task:
@@ -841,7 +843,8 @@ class SegmentedSTTService(STTService):
async def start(self, frame: StartFrame):
await super().start(frame)
(self._content, self._wave) = self._new_wave()
if not self._wave:
(self._content, self._wave) = self._new_wave()
async def stop(self, frame: EndFrame):
await super().stop(frame)

3
src/pipecat/services/assemblyai.py
View File

@@ -91,6 +91,9 @@ class AssemblyAISTTService(STTService):
AssemblyAI transcriber.
"""
if self._transcriber:
return
def on_open(session_opened: aai.RealtimeSessionOpened):
"""Callback for when the connection to AssemblyAI is opened."""
logger.info(f"{self}: Connected to AssemblyAI")

33
src/pipecat/services/azure.py
View File

@@ -535,6 +535,10 @@ class AzureTTSService(AzureBaseTTSService):
async def start(self, frame: StartFrame):
await super().start(frame)
if self._speech_config:
return
# Now self.sample_rate is properly initialized
self._speech_config = SpeechConfig(
subscription=self._api_key,
@@ -624,6 +628,10 @@ class AzureHttpTTSService(AzureBaseTTSService):
async def start(self, frame: StartFrame):
await super().start(frame)
if self._speech_config:
return
self._speech_config = SpeechConfig(
subscription=self._api_key,
region=self._region,
@@ -678,15 +686,22 @@ class AzureSTTService(STTService):
self._speech_config = SpeechConfig(subscription=api_key, region=region)
self._speech_config.speech_recognition_language = language
self._audio_stream = None
self._speech_recognizer = None
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
await self.start_processing_metrics()
self._audio_stream.write(audio)
if self._audio_stream:
self._audio_stream.write(audio)
await self.stop_processing_metrics()
yield None
async def start(self, frame: StartFrame):
await super().start(frame)
if self._audio_stream:
return
stream_format = AudioStreamFormat(samples_per_second=self.sample_rate, channels=1)
self._audio_stream = PushAudioInputStream(stream_format)
@@ -700,13 +715,21 @@ class AzureSTTService(STTService):
async def stop(self, frame: EndFrame):
await super().stop(frame)
self._speech_recognizer.stop_continuous_recognition_async()
self._audio_stream.close()
if self._speech_recognizer:
self._speech_recognizer.stop_continuous_recognition_async()
if self._audio_stream:
self._audio_stream.close()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
self._speech_recognizer.stop_continuous_recognition_async()
self._audio_stream.close()
if self._speech_recognizer:
self._speech_recognizer.stop_continuous_recognition_async()
if self._audio_stream:
self._audio_stream.close()
def _on_handle_recognized(self, event):
if event.result.reason == ResultReason.RecognizedSpeech and len(event.result.text) > 0:

6
src/pipecat/services/cartesia.py
View File

@@ -182,8 +182,8 @@ class CartesiaTTSService(AudioContextWordTTSService):
async def _connect(self):
await self._connect_websocket()
self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
if not self._receive_task:
self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
async def _disconnect(self):
if self._receive_task:
@@ -194,6 +194,8 @@ class CartesiaTTSService(AudioContextWordTTSService):
async def _connect_websocket(self):
try:
if self._websocket:
return
logger.debug("Connecting to Cartesia")
self._websocket = await websockets.connect(
f"{self._url}?api_key={self._api_key}&cartesia_version={self._cartesia_version}"

13
src/pipecat/services/elevenlabs.py
View File

@@ -214,6 +214,9 @@ class ElevenLabsTTSService(InterruptibleWordTTSService):
self._started = False
self._cumulative_time = 0
self._receive_task = None
self._keepalive_task = None
def can_generate_metrics(self) -> bool:
return True
@@ -286,8 +289,11 @@ class ElevenLabsTTSService(InterruptibleWordTTSService):
async def _connect(self):
await self._connect_websocket()
self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
self._keepalive_task = self.create_task(self._keepalive_task_handler())
if not self._receive_task:
self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
if not self._keepalive_task:
self._keepalive_task = self.create_task(self._keepalive_task_handler())
async def _disconnect(self):
if self._receive_task:
@@ -302,6 +308,9 @@ class ElevenLabsTTSService(InterruptibleWordTTSService):
async def _connect_websocket(self):
try:
if self._websocket:
return
logger.debug("Connecting to ElevenLabs")
voice_id = self._voice_id

6
src/pipecat/services/fish.py
View File

@@ -106,7 +106,8 @@ class FishAudioTTSService(InterruptibleTTSService):
async def _connect(self):
await self._connect_websocket()
self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
if not self._receive_task:
self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
async def _disconnect(self):
if self._receive_task:
@@ -117,6 +118,9 @@ class FishAudioTTSService(InterruptibleTTSService):
async def _connect_websocket(self):
try:
if self._websocket:
return
logger.debug("Connecting to Fish Audio")
headers = {"Authorization": f"Bearer {self._api_key}"}
self._websocket = await websockets.connect(self._base_url, extra_headers=headers)

19
src/pipecat/services/gladia.py
View File

@@ -172,27 +172,38 @@ class GladiaSTTService(STTService):
},
}
self._confidence = confidence
self._websocket = None
self._receive_task = None
def language_to_service_language(self, language: Language) -> Optional[str]:
return language_to_gladia_language(language)
async def start(self, frame: StartFrame):
await super().start(frame)
if self._websocket:
return
self._settings["sample_rate"] = self.sample_rate
response = await self._setup_gladia()
self._websocket = await websockets.connect(response["url"])
self._receive_task = self.create_task(self._receive_task_handler())
if not self._receive_task:
self._receive_task = self.create_task(self._receive_task_handler())
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._send_stop_recording()
await self._websocket.close()
await self.wait_for_task(self._receive_task)
if self._websocket:
await self._websocket.close()
self._websocket = None
if self._receive_task:
await self.wait_for_task(self._receive_task)
self._receive_task = None
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._websocket.close()
await self.cancel_task(self._receive_task)
if self._receive_task:
await self.cancel_task(self._receive_task)
self._receive_task = None
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
await self.start_processing_metrics()

7
src/pipecat/services/lmnt.py
View File

@@ -83,6 +83,7 @@ class LmntTTSService(InterruptibleTTSService):
"format": "raw", # Use raw format for direct PCM data
}
self._started = False
self._receive_task = None
def can_generate_metrics(self) -> bool:
return True
@@ -110,7 +111,8 @@ class LmntTTSService(InterruptibleTTSService):
async def _connect(self):
await self._connect_websocket()
self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
if not self._receive_task:
self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
async def _disconnect(self):
if self._receive_task:
@@ -122,6 +124,9 @@ class LmntTTSService(InterruptibleTTSService):
async def _connect_websocket(self):
"""Connect to LMNT websocket."""
try:
if self._websocket:
return
logger.debug("Connecting to LMNT")
# Build initial connection message

6
src/pipecat/services/playht.py
View File

@@ -159,7 +159,8 @@ class PlayHTTTSService(InterruptibleTTSService):
async def _connect(self):
await self._connect_websocket()
self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
if not self._receive_task:
self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
async def _disconnect(self):
if self._receive_task:
@@ -170,6 +171,9 @@ class PlayHTTTSService(InterruptibleTTSService):
async def _connect_websocket(self):
try:
if self._websocket:
return
logger.debug("Connecting to PlayHT")
if not self._websocket_url:

7
src/pipecat/services/rime.py
View File

@@ -165,7 +165,9 @@ class RimeTTSService(AudioContextWordTTSService):
async def _connect(self):
"""Establish websocket connection and start receive task."""
await self._connect_websocket()
self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
if not self._receive_task:
self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
async def _disconnect(self):
"""Close websocket connection and clean up tasks."""
@@ -178,6 +180,9 @@ class RimeTTSService(AudioContextWordTTSService):
async def _connect_websocket(self):
"""Connect to Rime websocket API with configured settings."""
try:
if self._websocket:
return
params = "&".join(f"{k}={v}" for k, v in self._settings.items())
url = f"{self._url}?{params}"
headers = {"Authorization": f"Bearer {self._api_key}"}

24
src/pipecat/services/riva.py
View File

@@ -166,6 +166,9 @@ class ParakeetSTTService(STTService):
self._asr_service = riva.client.ASRService(auth)
self._queue = asyncio.Queue()
self._config = None
self._thread_task = None
self._response_task = None
def can_generate_metrics(self) -> bool:
return False
@@ -173,6 +176,9 @@ class ParakeetSTTService(STTService):
async def start(self, frame: StartFrame):
await super().start(frame)
if self._config:
return
config = riva.client.StreamingRecognitionConfig(
config=riva.client.RecognitionConfig(
encoding=riva.client.AudioEncoding.LINEAR_PCM,
@@ -205,9 +211,12 @@ class ParakeetSTTService(STTService):
self._config = config
self._thread_task = self.create_task(self._thread_task_handler())
self._response_task = self.create_task(self._response_task_handler())
self._response_queue = asyncio.Queue()
if not self._thread_task:
self._thread_task = self.create_task(self._thread_task_handler())
if not self._response_task:
self._response_queue = asyncio.Queue()
self._response_task = self.create_task(self._response_task_handler())
async def stop(self, frame: EndFrame):
await super().stop(frame)
@@ -218,8 +227,13 @@ class ParakeetSTTService(STTService):
await self._stop_tasks()
async def _stop_tasks(self):
await self.cancel_task(self._thread_task)
await self.cancel_task(self._response_task)
if self._thread_task:
await self.cancel_task(self._thread_task)
self._thread_task = None
if self._response_task:
await self.cancel_task(self._response_task)
self._response_task = None
def _response_handler(self):
responses = self._asr_service.streaming_response_generator(

15
src/pipecat/services/simli.py
View File

@@ -43,14 +43,21 @@ class SimliVideoService(FrameProcessor):
self._pipecat_resampler: AudioResampler = None
self._simli_resampler = AudioResampler("s16", "mono", 16000)
self._initialized = False
self._audio_task: asyncio.Task = None
self._video_task: asyncio.Task = None
async def _start_connection(self):
await self._simli_client.Initialize()
if not self._initialized:
await self._simli_client.Initialize()
self._initialized = True
# Create task to consume and process audio and video
self._audio_task = self.create_task(self._consume_and_process_audio())
self._video_task = self.create_task(self._consume_and_process_video())
if not self._audio_task:
self._audio_task = self.create_task(self._consume_and_process_audio())
if not self._video_task:
self._video_task = self.create_task(self._consume_and_process_video())
async def _consume_and_process_audio(self):
await self._pipecat_resampler_event.wait()
@@ -117,5 +124,7 @@ class SimliVideoService(FrameProcessor):
await self._simli_client.stop()
if self._audio_task:
await self.cancel_task(self._audio_task)
self._audio_task = None
if self._video_task:
await self.cancel_task(self._video_task)
self._video_task = None

4
src/pipecat/services/xtts.py
View File

@@ -99,6 +99,10 @@ class XTTSService(TTSService):
async def start(self, frame: StartFrame):
await super().start(frame)
if self._studio_speakers:
return
async with self._aiohttp_session.get(self._settings["base_url"] + "/studio_speakers") as r:
if r.status != 200:
text = await r.text()

2
src/pipecat/transports/base_input.py
View File

@@ -74,7 +74,7 @@ class BaseInputTransport(FrameProcessor):
if self._params.audio_in_filter:
await self._params.audio_in_filter.start(self._sample_rate)
# Create audio input queue and task if needed.
if self._params.audio_in_enabled or self._params.vad_enabled:
if not self._audio_task and (self._params.audio_in_enabled or self._params.vad_enabled):
self._audio_in_queue = asyncio.Queue()
self._audio_task = self.create_task(self._audio_task_handler())

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

@@ -238,10 +238,12 @@ class BaseOutputTransport(FrameProcessor):
#
def _create_sink_tasks(self):
self._sink_queue = asyncio.Queue()
self._sink_clock_queue = asyncio.PriorityQueue()
self._sink_task = self.create_task(self._sink_task_handler())
self._sink_clock_task = self.create_task(self._sink_clock_task_handler())
if not self._sink_task:
self._sink_queue = asyncio.Queue()
self._sink_task = self.create_task(self._sink_task_handler())
if not self._sink_clock_task:
self._sink_clock_queue = asyncio.PriorityQueue()
self._sink_clock_task = self.create_task(self._sink_clock_task_handler())
async def _cancel_sink_tasks(self):
# Stop sink tasks.
@@ -358,7 +360,7 @@ class BaseOutputTransport(FrameProcessor):
def _create_camera_task(self):
# Create camera output queue and task if needed.
if self._params.camera_out_enabled:
if not self._camera_out_task and self._params.camera_out_enabled:
self._camera_out_queue = asyncio.Queue()
self._camera_out_task = self.create_task(self._camera_out_task_handler())

7
src/pipecat/transports/local/audio.py
View File

@@ -44,6 +44,9 @@ class LocalAudioInputTransport(BaseInputTransport):
async def start(self, frame: StartFrame):
await super().start(frame)
if self._in_stream:
return
self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
num_frames = int(self._sample_rate / 100) * 2 # 20ms of audio
@@ -94,6 +97,9 @@ class LocalAudioOutputTransport(BaseOutputTransport):
async def start(self, frame: StartFrame):
await super().start(frame)
if self._out_stream:
return
self._sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate
self._out_stream = self._py_audio.open(
@@ -110,6 +116,7 @@ class LocalAudioOutputTransport(BaseOutputTransport):
if self._out_stream:
self._out_stream.stop_stream()
self._out_stream.close()
self._out_stream = None
async def write_raw_audio_frames(self, frames: bytes):
if self._out_stream:

12
src/pipecat/transports/local/tk.py
View File

@@ -51,6 +51,9 @@ class TkInputTransport(BaseInputTransport):
async def start(self, frame: StartFrame):
await super().start(frame)
if self._in_stream:
return
self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
num_frames = int(self._sample_rate / 100) * 2 # 20ms of audio
@@ -70,6 +73,7 @@ class TkInputTransport(BaseInputTransport):
if self._in_stream:
self._in_stream.stop_stream()
self._in_stream.close()
self._in_stream = None
def _audio_in_callback(self, in_data, frame_count, time_info, status):
frame = InputAudioRawFrame(
@@ -86,7 +90,7 @@ class TkInputTransport(BaseInputTransport):
class TkOutputTransport(BaseOutputTransport):
_params: TkTransportParams
def __init__(self, tk_root: tk.Tk, py_audio: pyaudio.PyAudio, params: TransportParams):
def __init__(self, tk_root: tk.Tk, py_audio: pyaudio.PyAudio, params: TkTransportParams):
super().__init__(params)
self._py_audio = py_audio
self._out_stream = None
@@ -106,6 +110,9 @@ class TkOutputTransport(BaseOutputTransport):
async def start(self, frame: StartFrame):
await super().start(frame)
if self._out_stream:
return
self._sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate
self._out_stream = self._py_audio.open(
@@ -122,6 +129,7 @@ class TkOutputTransport(BaseOutputTransport):
if self._out_stream:
self._out_stream.stop_stream()
self._out_stream.close()
self._out_stream = None
async def write_raw_audio_frames(self, frames: bytes):
if self._out_stream:
@@ -145,7 +153,7 @@ class TkOutputTransport(BaseOutputTransport):
class TkLocalTransport(BaseTransport):
def __init__(self, tk_root: tk.Tk, params: TransportParams):
def __init__(self, tk_root: tk.Tk, params: TkTransportParams):
super().__init__()
self._tk_root = tk_root
self._params = params

10
src/pipecat/transports/network/fastapi_websocket.py
View File

@@ -115,15 +115,19 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
async def start(self, frame: StartFrame):
await super().start(frame)
await self._params.serializer.setup(frame)
if self._params.session_timeout:
if not self._monitor_websocket_task and self._params.session_timeout:
self._monitor_websocket_task = self.create_task(self._monitor_websocket())
await self._client.trigger_client_connected()
self._receive_task = self.create_task(self._receive_messages())
if not self._receive_task:
self._receive_task = self.create_task(self._receive_messages())
async def _stop_tasks(self):
if self._monitor_websocket_task:
await self.cancel_task(self._monitor_websocket_task)
await self.cancel_task(self._receive_task)
self._monitor_websocket_task = None
if self._receive_task:
await self.cancel_task(self._receive_task)
self._receive_task = None
async def stop(self, frame: EndFrame):
await super().stop(frame)

6
src/pipecat/transports/network/websocket_client.py
View File

@@ -30,7 +30,7 @@ from pipecat.serializers.protobuf import ProtobufFrameSerializer
from pipecat.transports.base_input import BaseInputTransport
from pipecat.transports.base_output import BaseOutputTransport
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.utils.asyncio import TaskManager
from pipecat.utils.asyncio import BaseTaskManager
class WebsocketClientParams(TransportParams):
@@ -57,12 +57,12 @@ class WebsocketClientSession:
self._callbacks = callbacks
self._transport_name = transport_name
self._task_manager: Optional[TaskManager] = None
self._task_manager: Optional[BaseTaskManager] = None
self._websocket: Optional[websockets.WebSocketClientProtocol] = None
@property
def task_manager(self) -> TaskManager:
def task_manager(self) -> BaseTaskManager:
if not self._task_manager:
raise Exception(
f"{self._transport_name}::WebsocketClientSession: TaskManager not initialized (pipeline not started?)"

10
src/pipecat/transports/network/websocket_server.py
View File

@@ -22,7 +22,6 @@ from pipecat.frames.frames import (
OutputAudioRawFrame,
StartFrame,
StartInterruptionFrame,
TextFrame,
TransportMessageFrame,
TransportMessageUrgentFrame,
)
@@ -81,22 +80,27 @@ class WebsocketServerInputTransport(BaseInputTransport):
async def start(self, frame: StartFrame):
await super().start(frame)
await self._params.serializer.setup(frame)
self._server_task = self.create_task(self._server_task_handler())
if not self._server_task:
self._server_task = self.create_task(self._server_task_handler())
async def stop(self, frame: EndFrame):
await super().stop(frame)
self._stop_server_event.set()
if self._monitor_task:
await self.cancel_task(self._monitor_task)
self._monitor_task = None
if self._server_task:
await self.wait_for_task(self._server_task)
self._server_task = None
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
if self._monitor_task:
await self.cancel_task(self._monitor_task)
self._monitor_task = None
if self._server_task:
await self.cancel_task(self._server_task)
self._server_task = None
async def _server_task_handler(self):
logger.info(f"Starting websocket server on {self._host}:{self._port}")
@@ -116,7 +120,7 @@ class WebsocketServerInputTransport(BaseInputTransport):
await self._callbacks.on_client_connected(websocket)
# Create a task to monitor the websocket connection
if self._params.session_timeout:
if not self._monitor_task and self._params.session_timeout:
self._monitor_task = self.create_task(
self._monitor_websocket(websocket, self._params.session_timeout)
)

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

@@ -43,7 +43,7 @@ from pipecat.transcriptions.language import Language
from pipecat.transports.base_input import BaseInputTransport
from pipecat.transports.base_output import BaseOutputTransport
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.utils.asyncio import TaskManager
from pipecat.utils.asyncio import BaseTaskManager
try:
from daily import CallClient, Daily, EventHandler
@@ -293,7 +293,7 @@ class DailyTransportClient(EventHandler):
self._joined_event = asyncio.Event()
self._leave_counter = 0
self._task_manager: Optional[TaskManager] = None
self._task_manager: Optional[BaseTaskManager] = None
# We use the executor to cleanup the client. We just do it from one
# place, so only one thread is really needed.
@@ -832,6 +832,9 @@ class DailyInputTransport(BaseInputTransport):
self._video_renderers = {}
# Whether we have seen a StartFrame already.
self._initialized = False
# Task that gets audio data from a device or the network and queues it
# internally to be processed.
self._audio_in_task = None
@@ -845,13 +848,19 @@ class DailyInputTransport(BaseInputTransport):
def start_audio_in_streaming(self):
# Create audio task. It reads audio frames from Daily and push them
# internally for VAD processing.
if self._params.audio_in_enabled or self._params.vad_enabled:
if not self._audio_in_task and (self._params.audio_in_enabled or self._params.vad_enabled):
logger.debug(f"Start receiving audio")
self._audio_in_task = self.create_task(self._audio_in_task_handler())
async def start(self, frame: StartFrame):
# Parent start.
await super().start(frame)
if self._initialized:
return
self._initialized = True
# Setup client.
await self._client.setup(frame)
# Join the room.
@@ -980,9 +989,18 @@ class DailyOutputTransport(BaseOutputTransport):
self._client = client
# Whether we have seen a StartFrame already.
self._initialized = False
async def start(self, frame: StartFrame):
# Parent start.
await super().start(frame)
if self._initialized:
return
self._initialized = True
# Setup client.
await self._client.setup(frame)
# Join the room.

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

@@ -27,7 +27,7 @@ from pipecat.processors.frame_processor import FrameDirection
from pipecat.transports.base_input import BaseInputTransport
from pipecat.transports.base_output import BaseOutputTransport
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.utils.asyncio import TaskManager
from pipecat.utils.asyncio import BaseTaskManager
try:
from livekit import rtc
@@ -88,7 +88,7 @@ class LiveKitTransportClient:
self._audio_tracks = {}
self._audio_queue = asyncio.Queue()
self._other_participant_has_joined = False
self._task_manager: Optional[TaskManager] = None
self._task_manager: Optional[BaseTaskManager] = None
@property
def participant_id(self) -> str:
@@ -360,7 +360,7 @@ class LiveKitInputTransport(BaseInputTransport):
await super().start(frame)
await self._client.setup(frame)
await self._client.connect()
if self._params.audio_in_enabled or self._params.vad_enabled:
if not self._audio_in_task and (self._params.audio_in_enabled or self._params.vad_enabled):
self._audio_in_task = self.create_task(self._audio_in_task_handler())
logger.info("LiveKitInputTransport started")

67
src/pipecat/utils/asyncio.py
View File

@@ -5,12 +5,76 @@
#
import asyncio
from abc import ABC, abstractmethod
from typing import Coroutine, Optional, Set
from loguru import logger
class TaskManager:
class BaseTaskManager(ABC):
@abstractmethod
def set_event_loop(self, loop: asyncio.AbstractEventLoop):
pass
@abstractmethod
def get_event_loop(self) -> asyncio.AbstractEventLoop:
pass
@abstractmethod
def create_task(self, coroutine: Coroutine, name: str) -> asyncio.Task:
"""
Creates and schedules a new asyncio Task that runs the given coroutine.
The task is added to a global set of created tasks.
Args:
loop (asyncio.AbstractEventLoop): The event loop to use for creating the task.
coroutine (Coroutine): The coroutine to be executed within the task.
name (str): The name to assign to the task for identification.
Returns:
asyncio.Task: The created task object.
"""
pass
@abstractmethod
async def wait_for_task(self, task: asyncio.Task, timeout: Optional[float] = None):
"""Wait for an asyncio.Task to complete with optional timeout handling.
This function awaits the specified asyncio.Task and handles scenarios for
timeouts, cancellations, and other exceptions. It also ensures that the task
is removed from the set of registered tasks upon completion or failure.
Args:
task (asyncio.Task): The asyncio Task to wait for.
timeout (Optional[float], optional): The maximum number of seconds
to wait for the task to complete. If None, waits indefinitely.
Defaults to None.
"""
pass
@abstractmethod
async def cancel_task(self, task: asyncio.Task, timeout: Optional[float] = None):
"""Cancels the given asyncio Task and awaits its completion with an
optional timeout.
This function removes the task from the set of registered tasks upon
completion or failure.
Args:
task (asyncio.Task): The task to be cancelled.
timeout (Optional[float]): The optional timeout in seconds to wait for the task to cancel.
"""
pass
@abstractmethod
def current_tasks(self) -> Set[asyncio.Task]:
"""Returns the list of currently created/registered tasks."""
pass
class TaskManager(BaseTaskManager):
def __init__(self) -> None:
self._tasks: Set[asyncio.Task] = set()
self._loop: Optional[asyncio.AbstractEventLoop] = None
@@ -80,6 +144,7 @@ class TaskManager:
logger.warning(f"{name}: timed out waiting for task to finish")
except asyncio.CancelledError:
logger.trace(f"{name}: unexpected task cancellation (maybe Ctrl-C?)")
raise
except Exception as e:
logger.exception(f"{name}: unexpected exception while stopping task: {e}")
finally:

2
tests/integration/integration_openai_llm.py
View File

@@ -10,7 +10,7 @@ from openai.types.chat import (
)
from pipecat.frames.frames import LLMFullResponseEndFrame, LLMFullResponseStartFrame, TextFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame, OpenAILLMService
from pipecat.utils.test_frame_processor import TestFrameProcessor