From 3d76b30a7c1ba2653fbbaf7cc0b90772b30942fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 4 Dec 2024 17:24:27 -0800 Subject: [PATCH] deepgram: add VAD event handlers --- CHANGELOG.md | 4 + .../07c-interruptible-deepgram-vad.py | 105 ++++++++++++++++++ pyproject.toml | 2 +- src/pipecat/services/deepgram.py | 8 +- 4 files changed, 117 insertions(+), 2 deletions(-) create mode 100644 examples/foundational/07c-interruptible-deepgram-vad.py diff --git a/CHANGELOG.md b/CHANGELOG.md index de29d0c74..4020daa1a 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/examples/foundational/07c-interruptible-deepgram-vad.py b/examples/foundational/07c-interruptible-deepgram-vad.py new file mode 100644 index 000000000..40866d760 --- /dev/null +++ b/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()) diff --git a/pyproject.toml b/pyproject.toml index d487a7dd7..c1d637f32 100644 --- a/pyproject.toml +++ b/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" ] diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index d298e3e9b..f322211d8 100644 --- a/src/pipecat/services/deepgram.py +++ b/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"]