deepgram: add VAD event handlers

CHANGELOG.md

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `DeepgramSTTService` now exposes two event handlers `on_speech_started` and
+  `on_utterance_end` that could be used to implement interruptions. See new
+  example `examples/foundational/07c-interruptible-deepgram-vad.py`
+
 - Added `GroqLLMService`, `GrokLLMService`, and `NimLLMService` for Groq, Grok,
   and NVIDIA NIM API integration, with an OpenAI-compatible interface.
 

examples/foundational/07c-interruptible-deepgram-vad.py

@@ -0,0 +1,105 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+
+import aiohttp
+from deepgram import LiveOptions
+from dotenv import load_dotenv
+from loguru import logger
+from runner import configure
+
+from pipecat.frames.frames import (
+    BotInterruptionFrame,
+    LLMMessagesFrame,
+    StopInterruptionFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
+)
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.services.deepgram import DeepgramSTTService, DeepgramTTSService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, _) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            None,
+            "Respond bot",
+            DailyParams(
+                audio_in_enabled=True,
+                audio_out_enabled=True,
+            ),
+        )
+
+        stt = DeepgramSTTService(
+            api_key=os.getenv("DEEPGRAM_API_KEY"),
+            live_options=LiveOptions(vad_events=True, utterance_end_ms="1000"),
+        )
+
+        tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en")
+
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                stt,  # STT
+                context_aggregator.user(),  # User responses
+                llm,  # LLM
+                tts,  # TTS
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant spoken responses
+            ]
+        )
+
+        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
+
+        @stt.event_handler("on_speech_started")
+        async def on_speech_started(stt, *args, **kwargs):
+            await task.queue_frames([BotInterruptionFrame(), UserStartedSpeakingFrame()])
+
+        @stt.event_handler("on_utterance_end")
+        async def on_utterance_end(stt, *args, **kwargs):
+            await task.queue_frames([StopInterruptionFrame(), UserStoppedSpeakingFrame()])
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            # Kick off the conversation.
+            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+            await task.queue_frames([LLMMessagesFrame(messages)])
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

@@ -43,7 +43,7 @@ azure = [ "azure-cognitiveservices-speech~=1.40.0", "openai~=1.50.2" ]
 canonical = [ "aiofiles~=24.1.0" ]
 cartesia = [ "cartesia~=1.0.13", "websockets~=13.1" ]
 daily = [ "daily-python~=0.13.0" ]
-deepgram = [ "deepgram-sdk~=3.7.3" ]
+deepgram = [ "deepgram-sdk~=3.7.7" ]
 elevenlabs = [ "websockets~=13.1" ]
 examples = [ "python-dotenv~=1.0.1", "flask~=3.0.3", "flask_cors~=4.0.1" ]
 fal = [ "fal-client~=0.4.1" ]

src/pipecat/services/deepgram.py

@@ -35,7 +35,6 @@ try:
         LiveResultResponse,
         LiveTranscriptionEvents,
         SpeakOptions,
-        logging,
     )
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
@@ -151,7 +150,10 @@ class DeepgramSTTService(STTService):
         self._connection: AsyncListenWebSocketClient = self._client.listen.asyncwebsocket.v("1")
         self._connection.on(LiveTranscriptionEvents.Transcript, self._on_message)
         if self.vad_enabled:
+            self._register_event_handler("on_speech_started")
+            self._register_event_handler("on_utterance_end")
             self._connection.on(LiveTranscriptionEvents.SpeechStarted, self._on_speech_started)
+            self._connection.on(LiveTranscriptionEvents.UtteranceEnd, self._on_utterance_end)
 
     @property
     def vad_enabled(self):
@@ -203,6 +205,10 @@ class DeepgramSTTService(STTService):
     async def _on_speech_started(self, *args, **kwargs):
         await self.start_ttfb_metrics()
         await self.start_processing_metrics()
+        await self._call_event_handler("on_speech_started", *args, **kwargs)
+
+    async def _on_utterance_end(self, *args, **kwargs):
+        await self._call_event_handler("on_utterance_end", *args, **kwargs)
 
     async def _on_message(self, *args, **kwargs):
         result: LiveResultResponse = kwargs["result"]