From e9af585eddfd7d928865c139a22b608edb1e0652 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 16 Apr 2025 14:48:31 -0700 Subject: [PATCH] DeepgramTTSService: re-add base_url to constructor --- CHANGELOG.md | 8 + .../16-gpu-container-local-bot.py | 168 +++++++++--------- src/pipecat/services/deepgram/stt.py | 14 +- src/pipecat/services/deepgram/tts.py | 7 +- 4 files changed, 108 insertions(+), 89 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e3976af2d..1837b4adb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- `DeepgramTTSService` accepts `base_url` argument again, allowing you to + connect to an on-prem service. + - It is now possible to disable `SoundfileMixer` when created. You can then use `MixerEnableFrame` to dynamically enable it when necessary. @@ -25,6 +28,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `SoundfileMixer` constructor arguments need to be keywords. +### Deprecated + +- `DeepgramSTTService` parameter `url` is now deprecated, use `base_url` + instead. + ### Fixed - Fixed a `TavusVideoService` issue that was causing audio choppiness. diff --git a/examples/foundational/16-gpu-container-local-bot.py b/examples/foundational/16-gpu-container-local-bot.py index 4853764fc..e4b1b34d9 100644 --- a/examples/foundational/16-gpu-container-local-bot.py +++ b/examples/foundational/16-gpu-container-local-bot.py @@ -6,7 +6,6 @@ import os -import aiohttp from dotenv import load_dotenv from loguru import logger @@ -40,104 +39,101 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): ), ) - # Create an HTTP session - async with aiohttp.ClientSession() as session: - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - tts = DeepgramTTSService( - aiohttp_session=session, - api_key=os.getenv("DEEPGRAM_API_KEY"), - voice="aura-asteria-en", - base_url="http://0.0.0.0:8080/v1/speak", - ) + tts = DeepgramTTSService( + api_key=os.getenv("DEEPGRAM_API_KEY"), + voice="aura-asteria-en", + base_url="http://0.0.0.0:8080", + ) - llm = OpenAILLMService( - # To use OpenAI - # api_key=os.getenv("OPENAI_API_KEY"), - # Or, to use a local vLLM (or similar) api server - model="meta-llama/Meta-Llama-3-8B-Instruct", - base_url="http://0.0.0.0:8000/v1", - ) + llm = OpenAILLMService( + # To use OpenAI + # api_key=os.getenv("OPENAI_API_KEY"), + # Or, to use a local vLLM (or similar) api server + model="meta-llama/Meta-Llama-3-8B-Instruct", + base_url="http://0.0.0.0:8000/v1", + ) - 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.", - }, + 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(), + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), ] + ) - context = OpenAILLMContext(messages) - context_aggregator = llm.create_context_aggregator(context) + task = PipelineTask( + pipeline, + params=PipelineParams( + allow_interruptions=True, + enable_metrics=True, + ), + ) - pipeline = Pipeline( - [ - transport.input(), # Transport user input - stt, # STT - context_aggregator.user(), - llm, # LLM - tts, # TTS - transport.output(), # Transport bot output - context_aggregator.assistant(), - ] - ) + # When the first participant joins, the bot should introduce itself. + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([context_aggregator.user().get_context_frame()]) - task = PipelineTask( - pipeline, - params=PipelineParams( - allow_interruptions=True, - enable_metrics=True, - ), - ) - - # When the first participant joins, the bot should introduce itself. - @transport.event_handler("on_client_connected") - async def on_client_connected(transport, client): - logger.info(f"Client connected") - # Kick off the conversation. - messages.append({"role": "system", "content": "Please introduce yourself to the user."}) - await task.queue_frames([context_aggregator.user().get_context_frame()]) - - # Handle "latency-ping" messages. The client will send app messages that look like - # this: - # { "latency-ping": { ts: }} - # - # We want to send an immediate pong back to the client from this handler function. - # Also, we will push a frame into the top of the pipeline and send it after the - # - @transport.event_handler("on_app_message") - async def on_app_message(transport, message, sender): - try: - if "latency-ping" in message: - logger.debug(f"Received latency ping app message: {message}") - ts = message["latency-ping"]["ts"] - # Send immediately - transport.output().send_message( - DailyTransportMessageFrame( - message={"latency-pong-msg-handler": {"ts": ts}}, participant_id=sender - ) + # Handle "latency-ping" messages. The client will send app messages that look like + # this: + # { "latency-ping": { ts: }} + # + # We want to send an immediate pong back to the client from this handler function. + # Also, we will push a frame into the top of the pipeline and send it after the + # + @transport.event_handler("on_app_message") + async def on_app_message(transport, message, sender): + try: + if "latency-ping" in message: + logger.debug(f"Received latency ping app message: {message}") + ts = message["latency-ping"]["ts"] + # Send immediately + transport.output().send_message( + DailyTransportMessageFrame( + message={"latency-pong-msg-handler": {"ts": ts}}, participant_id=sender ) - # And push to the pipeline for the Daily transport.output to send - await task.queue_frame( - DailyTransportMessageFrame( - message={"latency-pong-pipeline-delivery": {"ts": ts}}, - participant_id=sender, - ) + ) + # And push to the pipeline for the Daily transport.output to send + await task.queue_frame( + DailyTransportMessageFrame( + message={"latency-pong-pipeline-delivery": {"ts": ts}}, + participant_id=sender, ) - except Exception as e: - logger.debug(f"message handling error: {e} - {message}") + ) + except Exception as e: + logger.debug(f"message handling error: {e} - {message}") - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") - @transport.event_handler("on_client_closed") - async def on_client_closed(transport, client): - logger.info(f"Client closed connection") - await task.cancel() + @transport.event_handler("on_client_closed") + async def on_client_closed(transport, client): + logger.info(f"Client closed connection") + await task.cancel() - runner = PipelineRunner(handle_sigint=False) + runner = PipelineRunner(handle_sigint=False) - await runner.run(task) + await runner.run(task) if __name__ == "__main__": diff --git a/src/pipecat/services/deepgram/stt.py b/src/pipecat/services/deepgram/stt.py index 088b77829..7b5209e0b 100644 --- a/src/pipecat/services/deepgram/stt.py +++ b/src/pipecat/services/deepgram/stt.py @@ -45,6 +45,7 @@ class DeepgramSTTService(STTService): *, api_key: str, url: str = "", + base_url: str = "", sample_rate: Optional[int] = None, live_options: Optional[LiveOptions] = None, addons: Optional[Dict] = None, @@ -53,6 +54,17 @@ class DeepgramSTTService(STTService): sample_rate = sample_rate or (live_options.sample_rate if live_options else None) super().__init__(sample_rate=sample_rate, **kwargs) + if url: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Parameter 'url' is deprecated, use 'base_url' instead.", + DeprecationWarning, + ) + base_url = url + default_options = LiveOptions( encoding="linear16", language=Language.EN, @@ -81,7 +93,7 @@ class DeepgramSTTService(STTService): self._client = DeepgramClient( api_key, config=DeepgramClientOptions( - url=url, + url=base_url, options={"keepalive": "true"}, # verbose=logging.DEBUG ), ) diff --git a/src/pipecat/services/deepgram/tts.py b/src/pipecat/services/deepgram/tts.py index 0b32370a1..ec8a755a0 100644 --- a/src/pipecat/services/deepgram/tts.py +++ b/src/pipecat/services/deepgram/tts.py @@ -18,7 +18,7 @@ from pipecat.frames.frames import ( from pipecat.services.tts_service import TTSService try: - from deepgram import DeepgramClient, SpeakOptions + from deepgram import DeepgramClient, DeepgramClientOptions, SpeakOptions except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error("In order to use Deepgram, you need to `pip install pipecat-ai[deepgram]`.") @@ -31,6 +31,7 @@ class DeepgramTTSService(TTSService): *, api_key: str, voice: str = "aura-helios-en", + base_url: str = "", sample_rate: Optional[int] = None, encoding: str = "linear16", **kwargs, @@ -41,7 +42,9 @@ class DeepgramTTSService(TTSService): "encoding": encoding, } self.set_voice(voice) - self._deepgram_client = DeepgramClient(api_key=api_key) + + client_options = DeepgramClientOptions(url=base_url) + self._deepgram_client = DeepgramClient(api_key, config=client_options) def can_generate_metrics(self) -> bool: return True