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

all frame processors are asynchrnous

Browse Source 
In this commit we make all frame processors asynchronous, that is, they have an
internal queue and they push frames using a task from that queue.
This commit is contained in:
Aleix Conchillo Flaqué
2024-09-30 10:05:56 -07:00
parent 46ac76701e
commit c8995b82e5
15 changed files with 113 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -37,7 +37,6 @@ class FrameProcessor:
*,
name: str | None = None,
metrics: FrameProcessorMetrics | None = None,
sync: bool = True,
loop: asyncio.AbstractEventLoop | None = None,
**kwargs,
):
@@ -47,7 +46,6 @@ class FrameProcessor:
self._prev: "FrameProcessor" | None = None
self._next: "FrameProcessor" | None = None
self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_running_loop()
self._sync = sync
self._event_handlers: dict = {}
@@ -66,11 +64,8 @@ class FrameProcessor:
# 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.
if not self._sync:
self.__create_push_task()
# the exception to this rule. This create this task.
self.__create_push_task()
@property
def interruptions_allowed(self):
@@ -167,7 +162,7 @@ class FrameProcessor:
await self.push_frame(error, FrameDirection.UPSTREAM)
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
if self._sync or isinstance(frame, SystemFrame):
if isinstance(frame, SystemFrame):
await self.__internal_push_frame(frame, direction)
else:
await self.__push_queue.put((frame, direction))
@@ -194,13 +189,12 @@ class FrameProcessor:
#
async def _start_interruption(self):
if not self._sync:
# Cancel the task. This will stop pushing frames downstream.
self.__push_frame_task.cancel()
await self.__push_frame_task
# Cancel the task. This will stop pushing frames downstream.
self.__push_frame_task.cancel()
await self.__push_frame_task
# Create a new queue and task.
self.__create_push_task()
# Create a new queue and task.
self.__create_push_task()
async def _stop_interruption(self):
# Nothing to do right now.

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

@@ -516,7 +516,7 @@ class RTVIProcessor(FrameProcessor):
params: RTVIProcessorParams = RTVIProcessorParams(),
**kwargs,
):
super().__init__(sync=False, **kwargs)
super().__init__(**kwargs)
self._config = config
self._params = params

2
src/pipecat/processors/gstreamer/pipeline_source.py
View File

@@ -44,7 +44,7 @@ class GStreamerPipelineSource(FrameProcessor):
clock_sync: bool = True
def __init__(self, *, pipeline: str, out_params: OutputParams = OutputParams(), **kwargs):
super().__init__(sync=False, **kwargs)
super().__init__(**kwargs)
self._out_params = out_params

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

@@ -26,7 +26,7 @@ class IdleFrameProcessor(FrameProcessor):
types: List[type] = [],
**kwargs,
):
super().__init__(sync=False, **kwargs)
super().__init__(**kwargs)
self._callback = callback
self._timeout = timeout

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

@@ -31,7 +31,7 @@ class UserIdleProcessor(FrameProcessor):
timeout: float,
**kwargs,
):
super().__init__(sync=False, **kwargs)
super().__init__(**kwargs)
self._callback = callback
self._timeout = timeout

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

@@ -144,6 +144,10 @@ class TTSService(AIService):
# if True, TTSService will push TextFrames and LLMFullResponseEndFrames,
# otherwise subclass must do it
push_text_frames: bool = True,
# if True, TTSService will push TTSStoppedFrames, otherwise subclass must do it
push_stop_frames: bool = False,
# if push_stop_frames is True, wait for this idle period before pushing TTSStoppedFrame
stop_frame_timeout_s: float = 1.0,
# TTS output sample rate
sample_rate: int = 16000,
**kwargs,
@@ -151,9 +155,15 @@ class TTSService(AIService):
super().__init__(**kwargs)
self._aggregate_sentences: bool = aggregate_sentences
self._push_text_frames: bool = push_text_frames
self._current_sentence: str = ""
self._push_stop_frames: bool = push_stop_frames
self._stop_frame_timeout_s: float = stop_frame_timeout_s
self._sample_rate: int = sample_rate
self._stop_frame_task: Optional[asyncio.Task] = None
self._stop_frame_queue: asyncio.Queue = asyncio.Queue()
self._current_sentence: str = ""
@property
def sample_rate(self) -> int:
return self._sample_rate
@@ -210,13 +220,72 @@ class TTSService(AIService):
async def set_role(self, role: str):
pass
@abstractmethod
async def flush_audio(self):
pass
# Converts the text to audio.
@abstractmethod
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
pass
async def start(self, frame: StartFrame):
await super().start(frame)
if self._push_stop_frames:
self._stop_frame_task = self.get_event_loop().create_task(self._stop_frame_handler())
async def stop(self, frame: EndFrame):
await super().stop(frame)
if self._stop_frame_task:
self._stop_frame_task.cancel()
await self._stop_frame_task
self._stop_frame_task = None
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
if self._stop_frame_task:
self._stop_frame_task.cancel()
await self._stop_frame_task
self._stop_frame_task = None
async def say(self, text: str):
await self.process_frame(TextFrame(text=text), FrameDirection.DOWNSTREAM)
await self.flush_audio()
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, TextFrame):
await self._process_text_frame(frame)
elif isinstance(frame, StartInterruptionFrame):
await self._handle_interruption(frame, direction)
elif isinstance(frame, LLMFullResponseEndFrame) or isinstance(frame, EndFrame):
sentence = self._current_sentence
self._current_sentence = ""
await self._push_tts_frames(sentence)
if isinstance(frame, LLMFullResponseEndFrame):
if self._push_text_frames:
await self.push_frame(frame, direction)
else:
await self.push_frame(frame, direction)
elif isinstance(frame, TTSSpeakFrame):
await self._push_tts_frames(frame.text)
await self.flush_audio()
elif isinstance(frame, TTSUpdateSettingsFrame):
await self._update_tts_settings(frame)
else:
await self.push_frame(frame, direction)
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
await super().push_frame(frame, direction)
if self._push_stop_frames and (
isinstance(frame, StartInterruptionFrame)
or isinstance(frame, TTSStartedFrame)
or isinstance(frame, TTSAudioRawFrame)
or isinstance(frame, TTSStoppedFrame)
):
await self._stop_frame_queue.put(frame)
async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
self._current_sentence = ""
@@ -276,88 +345,6 @@ class TTSService(AIService):
if frame.role is not None:
await self.set_role(frame.role)
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, TextFrame):
await self._process_text_frame(frame)
elif isinstance(frame, StartInterruptionFrame):
await self._handle_interruption(frame, direction)
elif isinstance(frame, LLMFullResponseEndFrame) or isinstance(frame, EndFrame):
sentence = self._current_sentence
self._current_sentence = ""
await self._push_tts_frames(sentence)
if isinstance(frame, LLMFullResponseEndFrame):
if self._push_text_frames:
await self.push_frame(frame, direction)
else:
await self.push_frame(frame, direction)
elif isinstance(frame, TTSSpeakFrame):
await self._push_tts_frames(frame.text)
elif isinstance(frame, TTSUpdateSettingsFrame):
await self._update_tts_settings(frame)
else:
await self.push_frame(frame, direction)
class AsyncTTSService(TTSService):
def __init__(
self,
# if True, TTSService will push TTSStoppedFrames, otherwise subclass must do it
push_stop_frames: bool = False,
# if push_stop_frames is True, wait for this idle period before pushing TTSStoppedFrame
stop_frame_timeout_s: float = 1.0,
**kwargs,
):
super().__init__(sync=False, **kwargs)
self._push_stop_frames: bool = push_stop_frames
self._stop_frame_timeout_s: float = stop_frame_timeout_s
self._stop_frame_task: Optional[asyncio.Task] = None
self._stop_frame_queue: asyncio.Queue = asyncio.Queue()
@abstractmethod
async def flush_audio(self):
pass
async def say(self, text: str):
await super().say(text)
await self.flush_audio()
async def start(self, frame: StartFrame):
await super().start(frame)
if self._push_stop_frames:
self._stop_frame_task = self.get_event_loop().create_task(self._stop_frame_handler())
async def stop(self, frame: EndFrame):
await super().stop(frame)
if self._stop_frame_task:
self._stop_frame_task.cancel()
await self._stop_frame_task
self._stop_frame_task = None
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
if self._stop_frame_task:
self._stop_frame_task.cancel()
await self._stop_frame_task
self._stop_frame_task = None
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, TTSSpeakFrame):
await self.flush_audio()
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
await super().push_frame(frame, direction)
if self._push_stop_frames and (
isinstance(frame, StartInterruptionFrame)
or isinstance(frame, TTSStartedFrame)
or isinstance(frame, TTSAudioRawFrame)
or isinstance(frame, TTSStoppedFrame)
):
await self._stop_frame_queue.put(frame)
async def _stop_frame_handler(self):
try:
has_started = False
@@ -378,7 +365,7 @@ class AsyncTTSService(TTSService):
pass
class AsyncWordTTSService(AsyncTTSService):
class WordTTSService(TTSService):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._initial_word_timestamp = -1

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

@@ -26,7 +26,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.transcriptions.language import Language
from pipecat.services.ai_services import AsyncWordTTSService, TTSService
from pipecat.services.ai_services import WordTTSService, TTSService
from loguru import logger
@@ -61,7 +61,7 @@ def language_to_cartesia_language(language: Language) -> str | None:
return None
class CartesiaTTSService(AsyncWordTTSService):
class CartesiaTTSService(WordTTSService):
class InputParams(BaseModel):
encoding: Optional[str] = "pcm_s16le"
sample_rate: Optional[int] = 16000

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

@@ -23,7 +23,7 @@ from pipecat.frames.frames import (
TTSStoppedFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import AsyncWordTTSService
from pipecat.services.ai_services import WordTTSService
# See .env.example for ElevenLabs configuration needed
try:
@@ -70,7 +70,7 @@ def calculate_word_times(
return word_times
class ElevenLabsTTSService(AsyncWordTTSService):
class ElevenLabsTTSService(WordTTSService):
class InputParams(BaseModel):
language: Optional[str] = None
output_format: Literal["pcm_16000", "pcm_22050", "pcm_24000", "pcm_44100"] = "pcm_16000"

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

@@ -51,7 +51,7 @@ class GladiaSTTService(STTService):
params: InputParams = InputParams(),
**kwargs,
):
super().__init__(sync=False, **kwargs)
super().__init__(**kwargs)
self._api_key = api_key
self._url = url

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

@@ -20,7 +20,7 @@ from pipecat.frames.frames import (
TTSStartedFrame,
TTSStoppedFrame,
)
from pipecat.services.ai_services import AsyncTTSService
from pipecat.services.ai_services import TTSService
from loguru import logger
@@ -35,7 +35,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
class LmntTTSService(AsyncTTSService):
class LmntTTSService(TTSService):
def __init__(
self,
*,
@@ -47,7 +47,7 @@ class LmntTTSService(AsyncTTSService):
):
# Let TTSService produce TTSStoppedFrames after a short delay of
# no activity.
super().__init__(sync=False, push_stop_frames=True, sample_rate=sample_rate, **kwargs)
super().__init__(push_stop_frames=True, sample_rate=sample_rate, **kwargs)
self._api_key = api_key
self._voice_id = voice_id

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

@@ -31,7 +31,7 @@ from loguru import logger
class BaseInputTransport(FrameProcessor):
def __init__(self, params: TransportParams, **kwargs):
super().__init__(sync=False, **kwargs)
super().__init__(**kwargs)
self._params = params

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

@@ -43,7 +43,7 @@ from pipecat.utils.time import nanoseconds_to_seconds
class BaseOutputTransport(FrameProcessor):
def __init__(self, params: TransportParams, **kwargs):
super().__init__(sync=False, **kwargs)
super().__init__(**kwargs)
self._params = params