Merge pull request #228 from pipecat-ai/aleix/websocket-fixes

websocket fixes

CHANGELOG.md

@@ -19,6 +19,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed a couple of issues with `WebsocketServerTransport`. It needed to use
+  `push_audio_frame()` and also VAD was not working properly.
+
 - Fixed an issue that would cause LLM aggregator to fail with small
   `VADParams.stop_secs` values.
 

examples/websocket-server/bot.py

@@ -12,14 +12,14 @@ import sys
 from pipecat.frames.frames import LLMMessagesFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.task import PipelineTask
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantResponseAggregator,
     LLMUserResponseAggregator
 )
+from pipecat.services.deepgram import DeepgramSTTService
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
-from pipecat.services.whisper import WhisperSTTService
 from pipecat.transports.network.websocket_server import WebsocketServerParams, WebsocketServerTransport
 from pipecat.vad.silero import SileroVADAnalyzer
 
@@ -36,7 +36,6 @@ async def main():
     async with aiohttp.ClientSession() as session:
         transport = WebsocketServerTransport(
             params=WebsocketServerParams(
-                audio_in_enabled=True,
                 audio_out_enabled=True,
                 add_wav_header=True,
                 vad_enabled=True,
@@ -49,7 +48,7 @@ async def main():
             api_key=os.getenv("OPENAI_API_KEY"),
             model="gpt-4o")
 
-        stt = WhisperSTTService()
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
 
         tts = ElevenLabsTTSService(
             aiohttp_session=session,

src/pipecat/services/deepgram.py

@@ -25,6 +25,7 @@ from pipecat.services.ai_services import AIService, TTSService
 
 from deepgram import (
     DeepgramClient,
+    DeepgramClientOptions,
     LiveTranscriptionEvents,
     LiveOptions,
 )
@@ -92,7 +93,8 @@ class DeepgramSTTService(AIService):
 
         self._live_options = live_options
 
-        self._client = DeepgramClient(api_key)
+        self._client = DeepgramClient(
+            api_key, config=DeepgramClientOptions(options={"keepalive": "true"}))
         self._connection = self._client.listen.asynclive.v("1")
         self._connection.on(LiveTranscriptionEvents.Transcript, self._on_message)
 

src/pipecat/transports/base_input.py

@@ -70,7 +70,8 @@ class BaseInputTransport(FrameProcessor):
         return self._params.vad_analyzer
 
     def push_audio_frame(self, frame: AudioRawFrame):
-        self._audio_in_queue.put_nowait(frame)
+        if self._params.audio_in_enabled or self._params.vad_enabled:
+            self._audio_in_queue.put_nowait(frame)
 
     #
     # Frame processor

src/pipecat/transports/network/websocket_server.py

@@ -7,7 +7,6 @@
 
 import asyncio
 import io
-import queue
 import wave
 import websockets
 
@@ -53,7 +52,6 @@ class WebsocketServerInputTransport(BaseInputTransport):
 
         self._websocket: websockets.WebSocketServerProtocol | None = None
 
-        self._client_audio_queue = queue.Queue()
         self._stop_server_event = asyncio.Event()
 
     async def start(self, frame: StartFrame):
@@ -65,12 +63,6 @@ class WebsocketServerInputTransport(BaseInputTransport):
         await self._server_task
         await super().stop()
 
-    def read_next_audio_frame(self) -> AudioRawFrame | None:
-        try:
-            return self._client_audio_queue.get(timeout=1)
-        except queue.Empty:
-            return None
-
     async def _server_task_handler(self):
         logger.info(f"Starting websocket server on {self._host}:{self._port}")
         async with websockets.serve(self._client_handler, self._host, self._port) as server:
@@ -90,8 +82,8 @@ class WebsocketServerInputTransport(BaseInputTransport):
         # Handle incoming messages
         async for message in websocket:
             frame = self._params.serializer.deserialize(message)
-            if isinstance(frame, AudioRawFrame) and self._params.audio_in_enabled:
-                self._client_audio_queue.put_nowait(frame)
+            if isinstance(frame, AudioRawFrame):
+                self.push_audio_frame(frame)
             else:
                 await self._internal_push_frame(frame)