From e60b65228bb5928a26d6b4af8afa88547e1662d3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 25 Feb 2025 09:52:54 -0800 Subject: [PATCH] allow multiple StartFrames --- .../22d-natural-conversation-gemini-audio.py | 2 -- .../processors/aggregators/llm_response.py | 8 +++-- src/pipecat/services/ai_services.py | 3 +- src/pipecat/services/assemblyai.py | 3 ++ src/pipecat/services/azure.py | 33 ++++++++++++++++--- src/pipecat/services/gladia.py | 7 +++- src/pipecat/services/riva.py | 4 +++ src/pipecat/services/simli.py | 6 +++- src/pipecat/services/xtts.py | 4 +++ src/pipecat/transports/local/audio.py | 7 ++++ src/pipecat/transports/local/tk.py | 8 +++++ .../transports/network/websocket_server.py | 1 - src/pipecat/transports/services/daily.py | 18 ++++++++++ tests/integration/integration_openai_llm.py | 2 +- 14 files changed, 92 insertions(+), 14 deletions(-) diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index 0e1feee57..dde280fc1 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -25,10 +25,8 @@ from pipecat.frames.frames import ( InputAudioRawFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, - LLMMessagesFrame, StartFrame, StartInterruptionFrame, - StopInterruptionFrame, SystemFrame, TextFrame, TranscriptionFrame, diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index f25bf7644..da8ca63aa 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -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) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 594f7704d..71b1e7a92 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -840,7 +840,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) diff --git a/src/pipecat/services/assemblyai.py b/src/pipecat/services/assemblyai.py index 84113f582..94f081c5b 100644 --- a/src/pipecat/services/assemblyai.py +++ b/src/pipecat/services/assemblyai.py @@ -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") diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index fbdd08ab1..11c34214c 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -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: diff --git a/src/pipecat/services/gladia.py b/src/pipecat/services/gladia.py index 2fbf31882..f0a00b327 100644 --- a/src/pipecat/services/gladia.py +++ b/src/pipecat/services/gladia.py @@ -172,6 +172,7 @@ class GladiaSTTService(STTService): }, } self._confidence = confidence + self._websocket = None self._receive_task = None def language_to_service_language(self, language: Language) -> Optional[str]: @@ -179,6 +180,8 @@ class GladiaSTTService(STTService): 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"]) @@ -188,7 +191,9 @@ class GladiaSTTService(STTService): async def stop(self, frame: EndFrame): await super().stop(frame) await self._send_stop_recording() - await self._websocket.close() + 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 diff --git a/src/pipecat/services/riva.py b/src/pipecat/services/riva.py index 89c965c12..8d5ce79c4 100644 --- a/src/pipecat/services/riva.py +++ b/src/pipecat/services/riva.py @@ -166,6 +166,7 @@ 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 @@ -175,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, diff --git a/src/pipecat/services/simli.py b/src/pipecat/services/simli.py index 10f7ae68c..0fb490914 100644 --- a/src/pipecat/services/simli.py +++ b/src/pipecat/services/simli.py @@ -43,11 +43,15 @@ 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 if not self._audio_task: self._audio_task = self.create_task(self._consume_and_process_audio()) diff --git a/src/pipecat/services/xtts.py b/src/pipecat/services/xtts.py index 4d0f3fca6..9d2f3dd1e 100644 --- a/src/pipecat/services/xtts.py +++ b/src/pipecat/services/xtts.py @@ -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() diff --git a/src/pipecat/transports/local/audio.py b/src/pipecat/transports/local/audio.py index e7b803f2b..bc8d2dd16 100644 --- a/src/pipecat/transports/local/audio.py +++ b/src/pipecat/transports/local/audio.py @@ -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: diff --git a/src/pipecat/transports/local/tk.py b/src/pipecat/transports/local/tk.py index 73777c573..d270d6395 100644 --- a/src/pipecat/transports/local/tk.py +++ b/src/pipecat/transports/local/tk.py @@ -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( @@ -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: diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index 4d1e07230..19ebe4a45 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -22,7 +22,6 @@ from pipecat.frames.frames import ( OutputAudioRawFrame, StartFrame, StartInterruptionFrame, - TextFrame, TransportMessageFrame, TransportMessageUrgentFrame, ) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 1d31f348b..7735ae91f 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -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 @@ -852,6 +855,12 @@ class DailyInputTransport(BaseInputTransport): 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. diff --git a/tests/integration/integration_openai_llm.py b/tests/integration/integration_openai_llm.py index c788936a1..2c84c8882 100644 --- a/tests/integration/integration_openai_llm.py +++ b/tests/integration/integration_openai_llm.py @@ -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