allow multiple StartFrames

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,

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)

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)

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")

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:

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

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,

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())

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()

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:

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:

src/pipecat/transports/network/websocket_server.py

@@ -22,7 +22,6 @@ from pipecat.frames.frames import (
     OutputAudioRawFrame,
     StartFrame,
     StartInterruptionFrame,
-    TextFrame,
     TransportMessageFrame,
     TransportMessageUrgentFrame,
 )

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.

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