From 81282f9c4d62746e00153b5e1080d9b20fa9316c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 11 Jun 2024 22:52:28 -0700 Subject: [PATCH 1/3] services(deepgram): keep conenction alive --- src/pipecat/services/deepgram.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index 1c3b890ee..6584775cb 100644 --- a/src/pipecat/services/deepgram.py +++ b/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) From c23b14f76870dcca35a5abf3cf64890d81628c8f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 11 Jun 2024 23:09:30 -0700 Subject: [PATCH 2/3] examples: use DeepgramSTTService in websocker-server --- examples/websocket-server/bot.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/examples/websocket-server/bot.py b/examples/websocket-server/bot.py index ba61084de..b05d644e3 100644 --- a/examples/websocket-server/bot.py +++ b/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, From 72c27215b640de635c0a61356397dc1f3c3b6350 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 11 Jun 2024 23:10:37 -0700 Subject: [PATCH 3/3] transports(websocket): use push_audio_frame() --- CHANGELOG.md | 3 +++ src/pipecat/transports/base_input.py | 3 ++- src/pipecat/transports/network/websocket_server.py | 12 ++---------- 3 files changed, 7 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0cbb495f9..c499dcc1b 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 9496cd3b4..303979816 100644 --- a/src/pipecat/transports/base_input.py +++ b/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 diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index 8456c4536..bf51732e1 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/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)