DeepgramTTSService: re-add base_url to constructor

This commit is contained in:
Aleix Conchillo Flaqué
2025-04-16 14:48:31 -07:00
parent 31f7082d12
commit e9af585edd
4 changed files with 108 additions and 89 deletions

View File

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added ### 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 - It is now possible to disable `SoundfileMixer` when created. You can then use
`MixerEnableFrame` to dynamically enable it when necessary. `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. - `SoundfileMixer` constructor arguments need to be keywords.
### Deprecated
- `DeepgramSTTService` parameter `url` is now deprecated, use `base_url`
instead.
### Fixed ### Fixed
- Fixed a `TavusVideoService` issue that was causing audio choppiness. - Fixed a `TavusVideoService` issue that was causing audio choppiness.

View File

@@ -6,7 +6,6 @@
import os import os
import aiohttp
from dotenv import load_dotenv from dotenv import load_dotenv
from loguru import logger from loguru import logger
@@ -40,104 +39,101 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection):
), ),
) )
# Create an HTTP session stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
async with aiohttp.ClientSession() as session:
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = DeepgramTTSService( tts = DeepgramTTSService(
aiohttp_session=session, api_key=os.getenv("DEEPGRAM_API_KEY"),
api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-asteria-en",
voice="aura-asteria-en", base_url="http://0.0.0.0:8080",
base_url="http://0.0.0.0:8080/v1/speak", )
)
llm = OpenAILLMService( llm = OpenAILLMService(
# To use OpenAI # To use OpenAI
# api_key=os.getenv("OPENAI_API_KEY"), # api_key=os.getenv("OPENAI_API_KEY"),
# Or, to use a local vLLM (or similar) api server # Or, to use a local vLLM (or similar) api server
model="meta-llama/Meta-Llama-3-8B-Instruct", model="meta-llama/Meta-Llama-3-8B-Instruct",
base_url="http://0.0.0.0:8000/v1", base_url="http://0.0.0.0:8000/v1",
) )
messages = [ messages = [
{ {
"role": "system", "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.", "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) task = PipelineTask(
context_aggregator = llm.create_context_aggregator(context) pipeline,
params=PipelineParams(
allow_interruptions=True,
enable_metrics=True,
),
)
pipeline = Pipeline( # When the first participant joins, the bot should introduce itself.
[ @transport.event_handler("on_client_connected")
transport.input(), # Transport user input async def on_client_connected(transport, client):
stt, # STT logger.info(f"Client connected")
context_aggregator.user(), # Kick off the conversation.
llm, # LLM messages.append({"role": "system", "content": "Please introduce yourself to the user."})
tts, # TTS await task.queue_frames([context_aggregator.user().get_context_frame()])
transport.output(), # Transport bot output
context_aggregator.assistant(),
]
)
task = PipelineTask( # Handle "latency-ping" messages. The client will send app messages that look like
pipeline, # this:
params=PipelineParams( # { "latency-ping": { ts: <client-side timestamp> }}
allow_interruptions=True, #
enable_metrics=True, # 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")
# When the first participant joins, the bot should introduce itself. async def on_app_message(transport, message, sender):
@transport.event_handler("on_client_connected") try:
async def on_client_connected(transport, client): if "latency-ping" in message:
logger.info(f"Client connected") logger.debug(f"Received latency ping app message: {message}")
# Kick off the conversation. ts = message["latency-ping"]["ts"]
messages.append({"role": "system", "content": "Please introduce yourself to the user."}) # Send immediately
await task.queue_frames([context_aggregator.user().get_context_frame()]) transport.output().send_message(
DailyTransportMessageFrame(
# Handle "latency-ping" messages. The client will send app messages that look like message={"latency-pong-msg-handler": {"ts": ts}}, participant_id=sender
# this:
# { "latency-ping": { ts: <client-side timestamp> }}
#
# 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( # And push to the pipeline for the Daily transport.output to send
DailyTransportMessageFrame( await task.queue_frame(
message={"latency-pong-pipeline-delivery": {"ts": ts}}, DailyTransportMessageFrame(
participant_id=sender, 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") @transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client): async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected") logger.info(f"Client disconnected")
@transport.event_handler("on_client_closed") @transport.event_handler("on_client_closed")
async def on_client_closed(transport, client): async def on_client_closed(transport, client):
logger.info(f"Client closed connection") logger.info(f"Client closed connection")
await task.cancel() await task.cancel()
runner = PipelineRunner(handle_sigint=False) runner = PipelineRunner(handle_sigint=False)
await runner.run(task) await runner.run(task)
if __name__ == "__main__": if __name__ == "__main__":

View File

@@ -45,6 +45,7 @@ class DeepgramSTTService(STTService):
*, *,
api_key: str, api_key: str,
url: str = "", url: str = "",
base_url: str = "",
sample_rate: Optional[int] = None, sample_rate: Optional[int] = None,
live_options: Optional[LiveOptions] = None, live_options: Optional[LiveOptions] = None,
addons: Optional[Dict] = 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) sample_rate = sample_rate or (live_options.sample_rate if live_options else None)
super().__init__(sample_rate=sample_rate, **kwargs) 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( default_options = LiveOptions(
encoding="linear16", encoding="linear16",
language=Language.EN, language=Language.EN,
@@ -81,7 +93,7 @@ class DeepgramSTTService(STTService):
self._client = DeepgramClient( self._client = DeepgramClient(
api_key, api_key,
config=DeepgramClientOptions( config=DeepgramClientOptions(
url=url, url=base_url,
options={"keepalive": "true"}, # verbose=logging.DEBUG options={"keepalive": "true"}, # verbose=logging.DEBUG
), ),
) )

View File

@@ -18,7 +18,7 @@ from pipecat.frames.frames import (
from pipecat.services.tts_service import TTSService from pipecat.services.tts_service import TTSService
try: try:
from deepgram import DeepgramClient, SpeakOptions from deepgram import DeepgramClient, DeepgramClientOptions, SpeakOptions
except ModuleNotFoundError as e: except ModuleNotFoundError as e:
logger.error(f"Exception: {e}") logger.error(f"Exception: {e}")
logger.error("In order to use Deepgram, you need to `pip install pipecat-ai[deepgram]`.") 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, api_key: str,
voice: str = "aura-helios-en", voice: str = "aura-helios-en",
base_url: str = "",
sample_rate: Optional[int] = None, sample_rate: Optional[int] = None,
encoding: str = "linear16", encoding: str = "linear16",
**kwargs, **kwargs,
@@ -41,7 +42,9 @@ class DeepgramTTSService(TTSService):
"encoding": encoding, "encoding": encoding,
} }
self.set_voice(voice) 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: def can_generate_metrics(self) -> bool:
return True return True