From 98095de51eaab2d05e7bb3666c6222b47cbde49d Mon Sep 17 00:00:00 2001 From: John Vajda Date: Tue, 1 Jul 2025 15:31:41 -0600 Subject: [PATCH 1/5] implementation and unit tests --- .../clients/agent/v1/websocket/options.py | 60 ++- .../test_unit_agent_speak_provider.py | 359 ++++++++++++++++++ 2 files changed, 408 insertions(+), 11 deletions(-) create mode 100644 tests/unit_test/test_unit_agent_speak_provider.py diff --git a/deepgram/clients/agent/v1/websocket/options.py b/deepgram/clients/agent/v1/websocket/options.py index 4ef46ce9..43b2acde 100644 --- a/deepgram/clients/agent/v1/websocket/options.py +++ b/deepgram/clients/agent/v1/websocket/options.py @@ -273,26 +273,52 @@ class Agent(BaseResponse): or (isinstance(f, Think) and not f) ), ) - speak: Speak = field( - default_factory=Speak, - metadata=dataclass_config( - exclude=lambda f: f is None - or (isinstance(f, dict) and not f) - or (isinstance(f, Speak) and not f) - ), - ) + speak: Union[Speak, List[Speak]] = field(default_factory=Speak) greeting: Optional[str] = field( default=None, metadata=dataclass_config(exclude=lambda f: f is None) ) + def __post_init__(self): + """Handle conversion of dict/list data to proper Speak objects""" + # Handle listen conversion (existing pattern) + if ( + not isinstance(self.listen, Listen) + and self.listen is not None + and not (isinstance(self.listen, dict) and not self.listen) + ): + self.listen = Listen.from_dict(self.listen) + + # Handle think conversion (existing pattern) + if ( + not isinstance(self.think, Think) + and self.think is not None + and not (isinstance(self.think, dict) and not self.think) + ): + self.think = Think.from_dict(self.think) + + # Handle speak conversion (new OneOf pattern) + if isinstance(self.speak, list): + # Convert list of dicts to list of Speak objects + self.speak = [ + Speak.from_dict(item) if isinstance(item, dict) else item + for item in self.speak + ] + elif isinstance(self.speak, dict): + # Convert single dict to Speak object + self.speak = Speak.from_dict(self.speak) + # If it's already a Speak object or None, leave it as is + def __getitem__(self, key): _dict = self.to_dict() if "listen" in _dict and isinstance(_dict["listen"], dict): _dict["listen"] = Listen.from_dict(_dict["listen"]) if "think" in _dict and isinstance(_dict["think"], dict): _dict["think"] = Think.from_dict(_dict["think"]) - if "speak" in _dict and isinstance(_dict["speak"], dict): - _dict["speak"] = Speak.from_dict(_dict["speak"]) + if "speak" in _dict: + if isinstance(_dict["speak"], list): + _dict["speak"] = [Speak.from_dict(item) for item in _dict["speak"]] + elif isinstance(_dict["speak"], dict): + _dict["speak"] = Speak.from_dict(_dict["speak"]) return _dict[key] @@ -395,7 +421,19 @@ class UpdateSpeakOptions(BaseResponse): """ type: str = str(AgentWebSocketEvents.UpdateSpeak) - speak: Speak = field(default_factory=Speak) + speak: Union[Speak, List[Speak]] = field(default_factory=Speak) + + def __post_init__(self): + """Handle conversion of dict/list data to proper Speak objects""" + if isinstance(self.speak, list): + # Convert list of dicts to list of Speak objects + self.speak = [ + Speak.from_dict(item) if isinstance(item, dict) else item + for item in self.speak + ] + elif isinstance(self.speak, dict): + # Convert single dict to Speak object + self.speak = Speak.from_dict(self.speak) # InjectAgentMessage diff --git a/tests/unit_test/test_unit_agent_speak_provider.py b/tests/unit_test/test_unit_agent_speak_provider.py new file mode 100644 index 00000000..f17e4160 --- /dev/null +++ b/tests/unit_test/test_unit_agent_speak_provider.py @@ -0,0 +1,359 @@ +#!/usr/bin/env python3 +""" +Test script to verify both single provider and array provider formats work correctly +for agent speak configuration. +""" + +import json +import pytest +from deepgram.clients.agent.v1.websocket.options import ( + SettingsOptions, + UpdateSpeakOptions, + Agent, + Speak, + Provider, + Endpoint, + Header +) + + +class TestAgentSpeakSingleProvider: + """Test single provider format for agent speak configuration (backward compatibility)""" + + def test_single_provider_creation(self): + """Test creating agent speak with single provider format""" + options = SettingsOptions() + options.agent.speak.provider.type = "deepgram" + options.agent.speak.provider.model = "aura-2-thalia-en" + + # Verify the speak field is a single Speak object + assert isinstance(options.agent.speak, Speak) + assert options.agent.speak.provider.type == "deepgram" + assert options.agent.speak.provider.model == "aura-2-thalia-en" + + def test_single_provider_serialization(self): + """Test serialization of single provider format""" + options = SettingsOptions() + options.agent.speak.provider.type = "deepgram" + options.agent.speak.provider.model = "aura-2-thalia-en" + + result = options.to_dict() + speak_dict = result["agent"]["speak"] + + # Verify structure + assert isinstance(speak_dict, dict) + assert "provider" in speak_dict + assert speak_dict["provider"]["type"] == "deepgram" + assert speak_dict["provider"]["model"] == "aura-2-thalia-en" + + def test_single_provider_with_endpoint(self): + """Test single provider with custom endpoint""" + options = SettingsOptions() + options.agent.speak.provider.type = "custom" + options.agent.speak.provider.model = "custom-model" + options.agent.speak.endpoint = Endpoint() + options.agent.speak.endpoint.url = "https://custom.api.com/speak" + options.agent.speak.endpoint.headers = [ + Header(key="authorization", value="Bearer custom-token") + ] + + result = options.to_dict() + speak_dict = result["agent"]["speak"] + + assert speak_dict["provider"]["type"] == "custom" + assert speak_dict["endpoint"]["url"] == "https://custom.api.com/speak" + assert len(speak_dict["endpoint"]["headers"]) == 1 + assert speak_dict["endpoint"]["headers"][0]["key"] == "authorization" + + def test_single_provider_from_dict(self): + """Test deserialization of single provider format""" + single_data = { + "agent": { + "speak": { + "provider": {"type": "deepgram", "model": "aura-2-thalia-en"} + } + } + } + + options = SettingsOptions.from_dict(single_data) + assert isinstance(options.agent.speak, Speak) + assert options.agent.speak.provider.type == "deepgram" + assert options.agent.speak.provider.model == "aura-2-thalia-en" + + +class TestAgentSpeakArrayProvider: + """Test array provider format for agent speak configuration (new functionality)""" + + def test_array_provider_creation(self): + """Test creating agent speak with array provider format""" + # Create individual Speak objects + deepgram_speak = Speak() + deepgram_speak.provider.type = "deepgram" + deepgram_speak.provider.model = "aura-2-zeus-en" + + openai_speak = Speak() + openai_speak.provider.type = "open_ai" + openai_speak.provider.model = "tts-1" + openai_speak.provider.voice = "shimmer" + + # Create settings with array format + options = SettingsOptions() + options.agent.speak = [deepgram_speak, openai_speak] + + # Verify the speak field is a list of Speak objects + assert isinstance(options.agent.speak, list) + assert len(options.agent.speak) == 2 + assert isinstance(options.agent.speak[0], Speak) + assert isinstance(options.agent.speak[1], Speak) + + def test_array_provider_serialization(self): + """Test serialization of array provider format""" + # Create individual Speak objects + deepgram_speak = Speak() + deepgram_speak.provider.type = "deepgram" + deepgram_speak.provider.model = "aura-2-zeus-en" + + openai_speak = Speak() + openai_speak.provider.type = "open_ai" + openai_speak.provider.model = "tts-1" + openai_speak.provider.voice = "shimmer" + + # Create endpoint for OpenAI + openai_speak.endpoint = Endpoint() + openai_speak.endpoint.url = "https://api.openai.com/v1/audio/speech" + openai_speak.endpoint.headers = [ + Header(key="authorization", value="Bearer {{OPENAI_API_KEY}}") + ] + + # Create settings with array format + options = SettingsOptions() + options.agent.speak = [deepgram_speak, openai_speak] + + result = options.to_dict() + speak_array = result["agent"]["speak"] + + # Verify structure + assert isinstance(speak_array, list) + assert len(speak_array) == 2 + + # Check first provider (Deepgram) + assert speak_array[0]["provider"]["type"] == "deepgram" + assert speak_array[0]["provider"]["model"] == "aura-2-zeus-en" + + # Check second provider (OpenAI) + assert speak_array[1]["provider"]["type"] == "open_ai" + assert speak_array[1]["provider"]["model"] == "tts-1" + assert speak_array[1]["provider"]["voice"] == "shimmer" + assert speak_array[1]["endpoint"]["url"] == "https://api.openai.com/v1/audio/speech" + + def test_array_provider_from_dict(self): + """Test deserialization of array provider format""" + array_data = { + "agent": { + "speak": [ + {"provider": {"type": "deepgram", "model": "aura-2-zeus-en"}}, + {"provider": {"type": "open_ai", "model": "tts-1", "voice": "shimmer"}} + ] + } + } + + options = SettingsOptions.from_dict(array_data) + assert isinstance(options.agent.speak, list) + assert len(options.agent.speak) == 2 + assert options.agent.speak[0].provider.type == "deepgram" + assert options.agent.speak[1].provider.type == "open_ai" + assert options.agent.speak[1].provider.voice == "shimmer" + + def test_array_provider_with_multiple_endpoints(self): + """Test array provider with different endpoints""" + providers = [] + + # Provider 1: Deepgram with custom endpoint + deepgram_speak = Speak() + deepgram_speak.provider.type = "deepgram" + deepgram_speak.provider.model = "aura-2-zeus-en" + deepgram_speak.endpoint = Endpoint() + deepgram_speak.endpoint.url = "https://api.deepgram.com/v1/speak" + providers.append(deepgram_speak) + + # Provider 2: OpenAI TTS + openai_speak = Speak() + openai_speak.provider.type = "open_ai" + openai_speak.provider.model = "tts-1" + openai_speak.endpoint = Endpoint() + openai_speak.endpoint.url = "https://api.openai.com/v1/audio/speech" + providers.append(openai_speak) + + # Provider 3: Custom TTS provider + custom_speak = Speak() + custom_speak.provider.type = "custom" + custom_speak.provider.model = "custom-tts-v1" + custom_speak.endpoint = Endpoint() + custom_speak.endpoint.url = "https://custom-tts.example.com/v1/synthesize" + custom_speak.endpoint.headers = [ + Header(key="x-api-key", value="custom-api-key"), + Header(key="content-type", value="application/json") + ] + providers.append(custom_speak) + + options = SettingsOptions() + options.agent.speak = providers + + result = options.to_dict() + speak_array = result["agent"]["speak"] + + assert len(speak_array) == 3 + assert speak_array[0]["endpoint"]["url"] == "https://api.deepgram.com/v1/speak" + assert speak_array[1]["endpoint"]["url"] == "https://api.openai.com/v1/audio/speech" + assert speak_array[2]["endpoint"]["url"] == "https://custom-tts.example.com/v1/synthesize" + assert len(speak_array[2]["endpoint"]["headers"]) == 2 + + +class TestUpdateSpeakOptions: + """Test UpdateSpeakOptions with both single and array formats""" + + def test_update_speak_single_provider(self): + """Test UpdateSpeakOptions with single provider""" + speak = Speak() + speak.provider.type = "deepgram" + speak.provider.model = "aura-2-hera-en" + + update_options = UpdateSpeakOptions() + update_options.speak = speak + + result = update_options.to_dict() + + assert result["type"] == "UpdateSpeak" + assert isinstance(result["speak"], dict) + assert result["speak"]["provider"]["type"] == "deepgram" + assert result["speak"]["provider"]["model"] == "aura-2-hera-en" + + def test_update_speak_array_providers(self): + """Test UpdateSpeakOptions with array of providers""" + provider1 = Speak() + provider1.provider.type = "deepgram" + provider1.provider.model = "aura-2-hera-en" + + provider2 = Speak() + provider2.provider.type = "open_ai" + provider2.provider.model = "tts-1" + + update_options = UpdateSpeakOptions() + update_options.speak = [provider1, provider2] + + result = update_options.to_dict() + + assert result["type"] == "UpdateSpeak" + assert isinstance(result["speak"], list) + assert len(result["speak"]) == 2 + assert result["speak"][0]["provider"]["type"] == "deepgram" + assert result["speak"][1]["provider"]["type"] == "open_ai" + + +class TestAgentSpeakRoundTrip: + """Test serialization and deserialization round-trip for both formats""" + + def test_single_provider_round_trip(self): + """Test round-trip serialization/deserialization for single provider""" + # Create original + original = SettingsOptions() + original.agent.speak.provider.type = "deepgram" + original.agent.speak.provider.model = "aura-2-thalia-en" + original.agent.speak.endpoint = Endpoint() + original.agent.speak.endpoint.url = "https://api.deepgram.com/v1/speak" + + # Serialize and deserialize + serialized = original.to_dict() + restored = SettingsOptions.from_dict(serialized) + + # Verify + assert isinstance(restored.agent.speak, Speak) + assert restored.agent.speak.provider.type == "deepgram" + assert restored.agent.speak.provider.model == "aura-2-thalia-en" + assert restored.agent.speak.endpoint.url == "https://api.deepgram.com/v1/speak" + + def test_array_provider_round_trip(self): + """Test round-trip serialization/deserialization for array providers""" + # Create original + provider1 = Speak() + provider1.provider.type = "deepgram" + provider1.provider.model = "aura-2-zeus-en" + + provider2 = Speak() + provider2.provider.type = "open_ai" + provider2.provider.model = "tts-1" + provider2.endpoint = Endpoint() + provider2.endpoint.url = "https://api.openai.com/v1/audio/speech" + + original = SettingsOptions() + original.agent.speak = [provider1, provider2] + + # Serialize and deserialize + serialized = original.to_dict() + restored = SettingsOptions.from_dict(serialized) + + # Verify + assert isinstance(restored.agent.speak, list) + assert len(restored.agent.speak) == 2 + assert restored.agent.speak[0].provider.type == "deepgram" + assert restored.agent.speak[1].provider.type == "open_ai" + assert restored.agent.speak[1].endpoint.url == "https://api.openai.com/v1/audio/speech" + + +class TestAgentSpeakProviderConfig: + """Test various configuration scenarios""" + + def test_default_speak_configuration(self): + """Test default speak configuration""" + options = SettingsOptions() + + # Should default to single provider format + assert isinstance(options.agent.speak, Speak) + assert hasattr(options.agent.speak, 'provider') + + def test_arbitrary_provider_attributes(self): + """Test that arbitrary provider attributes are preserved""" + options = SettingsOptions() + options.agent.speak.provider.type = "deepgram" + options.agent.speak.provider.model = "aura-2-thalia-en" + options.agent.speak.provider.custom_attribute = "custom_value" + options.agent.speak.provider.priority = 1 + + result = options.to_dict() + provider = result["agent"]["speak"]["provider"] + + assert provider["type"] == "deepgram" + assert provider["model"] == "aura-2-thalia-en" + assert provider["custom_attribute"] == "custom_value" + assert provider["priority"] == 1 + + def test_mixed_provider_attributes_in_array(self): + """Test that different providers can have different attributes""" + provider1 = Speak() + provider1.provider.type = "deepgram" + provider1.provider.model = "aura-2-zeus-en" + provider1.provider.priority = 1 + + provider2 = Speak() + provider2.provider.type = "open_ai" + provider2.provider.model = "tts-1" + provider2.provider.voice = "shimmer" + provider2.provider.speed = 1.0 + + options = SettingsOptions() + options.agent.speak = [provider1, provider2] + + result = options.to_dict() + speak_array = result["agent"]["speak"] + + # First provider attributes + assert speak_array[0]["provider"]["type"] == "deepgram" + assert speak_array[0]["provider"]["priority"] == 1 + assert "voice" not in speak_array[0]["provider"] + + # Second provider attributes + assert speak_array[1]["provider"]["type"] == "open_ai" + assert speak_array[1]["provider"]["voice"] == "shimmer" + assert speak_array[1]["provider"]["speed"] == 1.0 + assert "priority" not in speak_array[1]["provider"] + From e34f45a393a1aea3a788343300d13b8f587a1357 Mon Sep 17 00:00:00 2001 From: John Vajda Date: Tue, 1 Jul 2025 17:47:36 -0600 Subject: [PATCH 2/5] adds readme info and example --- README.md | 50 ++++++++- examples/agent/fallback_tts/main.py | 165 ++++++++++++++++++++++++++++ 2 files changed, 214 insertions(+), 1 deletion(-) create mode 100644 examples/agent/fallback_tts/main.py diff --git a/README.md b/README.md index d2f7c6ab..cb33de8a 100644 --- a/README.md +++ b/README.md @@ -249,8 +249,10 @@ options.agent.think.provider.model = "gpt-4o-mini" options.agent.think.prompt = "You are a helpful AI assistant." options.agent.listen.provider.type = "deepgram" options.agent.listen.provider.model = "nova-3" + +# Option 1: Single TTS provider (backward compatible) options.agent.speak.provider.type = "deepgram" -options.agent.speak.provider.model ="aura-2-thalia-en" +options.agent.speak.provider.model = "aura-2-thalia-en" options.greeting = "Hello, I'm your AI assistant." @@ -261,6 +263,52 @@ connection.start(options) connection.finish() ``` +### Multiple TTS Providers (Fallback Support) + +For enhanced reliability, configure multiple TTS providers with automatic fallback: + +```python +from deepgram import ( + SettingsOptions, + Speak, + Endpoint, + Header +) + +# Primary TTS provider +primary_tts = Speak() +primary_tts.provider.type = "deepgram" +primary_tts.provider.model = "aura-2-zeus-en" + +# Fallback TTS provider +fallback_tts = Speak() +fallback_tts.provider.type = "open_ai" +fallback_tts.provider.model = "tts-1" +fallback_tts.provider.voice = "shimmer" +fallback_tts.endpoint = Endpoint() +fallback_tts.endpoint.url = "https://api.openai.com/v1/audio/speech" +fallback_tts.endpoint.headers = [ + Header(key="authorization", value="Bearer {{OPENAI_API_KEY}}") +] + +# Configure agent with fallback providers +options = SettingsOptions() +options.language = "en" +options.agent.think.provider.type = "open_ai" +options.agent.think.provider.model = "gpt-4o-mini" +options.agent.think.prompt = "You are a helpful AI assistant." +options.agent.listen.provider.type = "deepgram" +options.agent.listen.provider.model = "nova-3" + +# Option 2: Multiple TTS providers (with fallback) +options.agent.speak = [primary_tts, fallback_tts] + +options.greeting = "Hello, I'm your AI assistant with fallback TTS." + +# Start the connection +connection.start(options) +``` + This example demonstrates: - Setting up a WebSocket connection diff --git a/examples/agent/fallback_tts/main.py b/examples/agent/fallback_tts/main.py new file mode 100644 index 00000000..112b8923 --- /dev/null +++ b/examples/agent/fallback_tts/main.py @@ -0,0 +1,165 @@ +# Copyright 2024 Deepgram SDK contributors. All Rights Reserved. +# Use of this source code is governed by a MIT license that can be found in the LICENSE file. +# SPDX-License-Identifier: MIT + +""" +Example demonstrating Voice Agent with multiple TTS providers for fallback support. + +This example shows how to configure multiple TTS providers in an array format, +allowing the agent to automatically fall back to alternative providers if the +primary provider fails or is unavailable. +""" + +from deepgram.utils import verboselogs + +from deepgram import ( + DeepgramClient, + DeepgramClientOptions, + AgentWebSocketEvents, + SettingsOptions, + Speak, + Endpoint, + Header, +) + +global warning_notice +warning_notice = True + + +def main(): + try: + print("Starting Voice Agent with Fallback TTS Providers...") + + config: DeepgramClientOptions = DeepgramClientOptions( + options={ + "keepalive": "true", + "microphone_record": "true", + "speaker_playback": "true", + }, + verbose=verboselogs.SPAM, + ) + + deepgram: DeepgramClient = DeepgramClient("", config) + dg_connection = deepgram.agent.websocket.v("1") + + def on_open(self, open, **kwargs): + print(f"\n\n{open}\n\n") + + def on_binary_data(self, data, **kwargs): + global warning_notice + if warning_notice: + print("Received binary data from TTS provider") + print("Audio will be played automatically with speaker_playback=true") + warning_notice = False + + def on_welcome(self, welcome, **kwargs): + print(f"\n\n{welcome}\n\n") + + def on_settings_applied(self, settings_applied, **kwargs): + print(f"\n\nSettings Applied - Multiple TTS providers configured:\n{settings_applied}\n\n") + + def on_conversation_text(self, conversation_text, **kwargs): + print(f"\n\n{conversation_text}\n\n") + + def on_user_started_speaking(self, user_started_speaking, **kwargs): + print(f"\n\n{user_started_speaking}\n\n") + + def on_agent_thinking(self, agent_thinking, **kwargs): + print(f"\n\n{agent_thinking}\n\n") + + def on_agent_started_speaking(self, agent_started_speaking, **kwargs): + print(f"\n\n{agent_started_speaking}\n\n") + + def on_agent_audio_done(self, agent_audio_done, **kwargs): + print(f"\n\n{agent_audio_done}\n\n") + + def on_close(self, close, **kwargs): + print(f"\n\n{close}\n\n") + + def on_error(self, error, **kwargs): + print(f"\n\nError (may trigger TTS fallback): {error}\n\n") + + def on_unhandled(self, unhandled, **kwargs): + print(f"\n\n{unhandled}\n\n") + + # Register event handlers + dg_connection.on(AgentWebSocketEvents.Open, on_open) + dg_connection.on(AgentWebSocketEvents.AudioData, on_binary_data) + dg_connection.on(AgentWebSocketEvents.Welcome, on_welcome) + dg_connection.on(AgentWebSocketEvents.SettingsApplied, on_settings_applied) + dg_connection.on(AgentWebSocketEvents.ConversationText, on_conversation_text) + dg_connection.on(AgentWebSocketEvents.UserStartedSpeaking, on_user_started_speaking) + dg_connection.on(AgentWebSocketEvents.AgentThinking, on_agent_thinking) + dg_connection.on(AgentWebSocketEvents.AgentStartedSpeaking, on_agent_started_speaking) + dg_connection.on(AgentWebSocketEvents.AgentAudioDone, on_agent_audio_done) + dg_connection.on(AgentWebSocketEvents.Close, on_close) + dg_connection.on(AgentWebSocketEvents.Error, on_error) + dg_connection.on(AgentWebSocketEvents.Unhandled, on_unhandled) + + # Configure TTS providers for fallback support + print("Configuring TTS providers with fallback...") + + # Primary TTS Provider: Deepgram Aura + primary_tts = Speak() + primary_tts.provider.type = "deepgram" + primary_tts.provider.model = "aura-2-zeus-en" + + # Fallback TTS Provider: OpenAI TTS + fallback_tts = Speak() + fallback_tts.provider.type = "open_ai" + fallback_tts.provider.model = "tts-1" + fallback_tts.provider.voice = "shimmer" + + # Configure custom endpoint for OpenAI + fallback_tts.endpoint = Endpoint() + fallback_tts.endpoint.url = "https://api.openai.com/v1/audio/speech" + fallback_tts.endpoint.headers = [ + Header(key="authorization", value="Bearer {{OPENAI_API_KEY}}") + ] + + # Configure agent settings + options: SettingsOptions = SettingsOptions() + options.agent.think.provider.type = "open_ai" + options.agent.think.provider.model = "gpt-4o-mini" + options.agent.think.prompt = ( + "You are a helpful AI assistant with fallback TTS providers for reliable speech output. " + "If one provider fails, another will automatically take over." + ) + options.agent.listen.provider.type = "deepgram" + options.agent.listen.provider.model = "nova-3" + options.agent.listen.provider.keyterms = ["hello", "goodbye", "fallback"] + + # Configure multiple TTS providers (array format) + options.agent.speak = [primary_tts, fallback_tts] + + options.agent.language = "en" + options.greeting = ( + "Hello! I'm your AI assistant with fallback TTS providers. " + "I can automatically switch between Deepgram and OpenAI for reliable voice output." + ) + + print("TTS Providers configured:") + print(f"1. Primary: {primary_tts.provider.type} - {primary_tts.provider.model}") + print(f"2. Fallback: {fallback_tts.provider.type} - {fallback_tts.provider.model}") + + # Start the connection + if dg_connection.start(options) is False: + print("Failed to start connection") + return + + print("\n\n=== Voice Agent with Fallback TTS Started ===") + print("The agent is now running with primary and fallback TTS providers.") + print("If the primary provider fails, it will automatically fall back to the secondary.") + print("Press Enter to stop...\n\n") + input() + + # Close the connection + dg_connection.finish() + print("Finished - Fallback TTS example completed") + + except Exception as e: + print(f"An unexpected error occurred: {e}") + + +if __name__ == "__main__": + main() \ No newline at end of file From c1a4e80f51ae38e57c2a200238c39d73100a8712 Mon Sep 17 00:00:00 2001 From: John Vajda Date: Wed, 2 Jul 2025 11:17:16 -0600 Subject: [PATCH 3/5] refactors approach --- .../clients/agent/v1/websocket/options.py | 21 +-- examples/agent/fallback_tts/main.py | 165 ------------------ 2 files changed, 1 insertion(+), 185 deletions(-) delete mode 100644 examples/agent/fallback_tts/main.py diff --git a/deepgram/clients/agent/v1/websocket/options.py b/deepgram/clients/agent/v1/websocket/options.py index 43b2acde..e78b3c6e 100644 --- a/deepgram/clients/agent/v1/websocket/options.py +++ b/deepgram/clients/agent/v1/websocket/options.py @@ -280,33 +280,14 @@ class Agent(BaseResponse): def __post_init__(self): """Handle conversion of dict/list data to proper Speak objects""" - # Handle listen conversion (existing pattern) - if ( - not isinstance(self.listen, Listen) - and self.listen is not None - and not (isinstance(self.listen, dict) and not self.listen) - ): - self.listen = Listen.from_dict(self.listen) - - # Handle think conversion (existing pattern) - if ( - not isinstance(self.think, Think) - and self.think is not None - and not (isinstance(self.think, dict) and not self.think) - ): - self.think = Think.from_dict(self.think) - - # Handle speak conversion (new OneOf pattern) + # Handle speak conversion (OneOf pattern) if isinstance(self.speak, list): - # Convert list of dicts to list of Speak objects self.speak = [ Speak.from_dict(item) if isinstance(item, dict) else item for item in self.speak ] elif isinstance(self.speak, dict): - # Convert single dict to Speak object self.speak = Speak.from_dict(self.speak) - # If it's already a Speak object or None, leave it as is def __getitem__(self, key): _dict = self.to_dict() diff --git a/examples/agent/fallback_tts/main.py b/examples/agent/fallback_tts/main.py deleted file mode 100644 index 112b8923..00000000 --- a/examples/agent/fallback_tts/main.py +++ /dev/null @@ -1,165 +0,0 @@ -# Copyright 2024 Deepgram SDK contributors. All Rights Reserved. -# Use of this source code is governed by a MIT license that can be found in the LICENSE file. -# SPDX-License-Identifier: MIT - -""" -Example demonstrating Voice Agent with multiple TTS providers for fallback support. - -This example shows how to configure multiple TTS providers in an array format, -allowing the agent to automatically fall back to alternative providers if the -primary provider fails or is unavailable. -""" - -from deepgram.utils import verboselogs - -from deepgram import ( - DeepgramClient, - DeepgramClientOptions, - AgentWebSocketEvents, - SettingsOptions, - Speak, - Endpoint, - Header, -) - -global warning_notice -warning_notice = True - - -def main(): - try: - print("Starting Voice Agent with Fallback TTS Providers...") - - config: DeepgramClientOptions = DeepgramClientOptions( - options={ - "keepalive": "true", - "microphone_record": "true", - "speaker_playback": "true", - }, - verbose=verboselogs.SPAM, - ) - - deepgram: DeepgramClient = DeepgramClient("", config) - dg_connection = deepgram.agent.websocket.v("1") - - def on_open(self, open, **kwargs): - print(f"\n\n{open}\n\n") - - def on_binary_data(self, data, **kwargs): - global warning_notice - if warning_notice: - print("Received binary data from TTS provider") - print("Audio will be played automatically with speaker_playback=true") - warning_notice = False - - def on_welcome(self, welcome, **kwargs): - print(f"\n\n{welcome}\n\n") - - def on_settings_applied(self, settings_applied, **kwargs): - print(f"\n\nSettings Applied - Multiple TTS providers configured:\n{settings_applied}\n\n") - - def on_conversation_text(self, conversation_text, **kwargs): - print(f"\n\n{conversation_text}\n\n") - - def on_user_started_speaking(self, user_started_speaking, **kwargs): - print(f"\n\n{user_started_speaking}\n\n") - - def on_agent_thinking(self, agent_thinking, **kwargs): - print(f"\n\n{agent_thinking}\n\n") - - def on_agent_started_speaking(self, agent_started_speaking, **kwargs): - print(f"\n\n{agent_started_speaking}\n\n") - - def on_agent_audio_done(self, agent_audio_done, **kwargs): - print(f"\n\n{agent_audio_done}\n\n") - - def on_close(self, close, **kwargs): - print(f"\n\n{close}\n\n") - - def on_error(self, error, **kwargs): - print(f"\n\nError (may trigger TTS fallback): {error}\n\n") - - def on_unhandled(self, unhandled, **kwargs): - print(f"\n\n{unhandled}\n\n") - - # Register event handlers - dg_connection.on(AgentWebSocketEvents.Open, on_open) - dg_connection.on(AgentWebSocketEvents.AudioData, on_binary_data) - dg_connection.on(AgentWebSocketEvents.Welcome, on_welcome) - dg_connection.on(AgentWebSocketEvents.SettingsApplied, on_settings_applied) - dg_connection.on(AgentWebSocketEvents.ConversationText, on_conversation_text) - dg_connection.on(AgentWebSocketEvents.UserStartedSpeaking, on_user_started_speaking) - dg_connection.on(AgentWebSocketEvents.AgentThinking, on_agent_thinking) - dg_connection.on(AgentWebSocketEvents.AgentStartedSpeaking, on_agent_started_speaking) - dg_connection.on(AgentWebSocketEvents.AgentAudioDone, on_agent_audio_done) - dg_connection.on(AgentWebSocketEvents.Close, on_close) - dg_connection.on(AgentWebSocketEvents.Error, on_error) - dg_connection.on(AgentWebSocketEvents.Unhandled, on_unhandled) - - # Configure TTS providers for fallback support - print("Configuring TTS providers with fallback...") - - # Primary TTS Provider: Deepgram Aura - primary_tts = Speak() - primary_tts.provider.type = "deepgram" - primary_tts.provider.model = "aura-2-zeus-en" - - # Fallback TTS Provider: OpenAI TTS - fallback_tts = Speak() - fallback_tts.provider.type = "open_ai" - fallback_tts.provider.model = "tts-1" - fallback_tts.provider.voice = "shimmer" - - # Configure custom endpoint for OpenAI - fallback_tts.endpoint = Endpoint() - fallback_tts.endpoint.url = "https://api.openai.com/v1/audio/speech" - fallback_tts.endpoint.headers = [ - Header(key="authorization", value="Bearer {{OPENAI_API_KEY}}") - ] - - # Configure agent settings - options: SettingsOptions = SettingsOptions() - options.agent.think.provider.type = "open_ai" - options.agent.think.provider.model = "gpt-4o-mini" - options.agent.think.prompt = ( - "You are a helpful AI assistant with fallback TTS providers for reliable speech output. " - "If one provider fails, another will automatically take over." - ) - options.agent.listen.provider.type = "deepgram" - options.agent.listen.provider.model = "nova-3" - options.agent.listen.provider.keyterms = ["hello", "goodbye", "fallback"] - - # Configure multiple TTS providers (array format) - options.agent.speak = [primary_tts, fallback_tts] - - options.agent.language = "en" - options.greeting = ( - "Hello! I'm your AI assistant with fallback TTS providers. " - "I can automatically switch between Deepgram and OpenAI for reliable voice output." - ) - - print("TTS Providers configured:") - print(f"1. Primary: {primary_tts.provider.type} - {primary_tts.provider.model}") - print(f"2. Fallback: {fallback_tts.provider.type} - {fallback_tts.provider.model}") - - # Start the connection - if dg_connection.start(options) is False: - print("Failed to start connection") - return - - print("\n\n=== Voice Agent with Fallback TTS Started ===") - print("The agent is now running with primary and fallback TTS providers.") - print("If the primary provider fails, it will automatically fall back to the secondary.") - print("Press Enter to stop...\n\n") - input() - - # Close the connection - dg_connection.finish() - print("Finished - Fallback TTS example completed") - - except Exception as e: - print(f"An unexpected error occurred: {e}") - - -if __name__ == "__main__": - main() \ No newline at end of file From edc87f3a8161dc8c3ad0833edd2e740fa3719c0f Mon Sep 17 00:00:00 2001 From: John Vajda Date: Wed, 2 Jul 2025 11:26:15 -0600 Subject: [PATCH 4/5] updates readme example --- README.md | 63 ++++++++++++------------------------------------------- 1 file changed, 13 insertions(+), 50 deletions(-) diff --git a/README.md b/README.md index cb33de8a..b6374ccd 100644 --- a/README.md +++ b/README.md @@ -235,7 +235,8 @@ Configure a Voice Agent. ```python from deepgram import ( - SettingsOptions + SettingsOptions, + Speak ) # Create websocket connection @@ -250,10 +251,17 @@ options.agent.think.prompt = "You are a helpful AI assistant." options.agent.listen.provider.type = "deepgram" options.agent.listen.provider.model = "nova-3" -# Option 1: Single TTS provider (backward compatible) -options.agent.speak.provider.type = "deepgram" -options.agent.speak.provider.model = "aura-2-thalia-en" +# Configure multiple TTS providers for automatic fallback. +primary = Speak() +primary.provider.type = "deepgram" +primary.provider.model = "aura-2-zeus-en" + +fallback = Speak() +fallback.provider.type = "cartesia" +fallback.provider.model = "sonic-english" +options.agent.speak = [primary, fallback] +# Set Agent greeting options.greeting = "Hello, I'm your AI assistant." # Start the connection @@ -263,52 +271,6 @@ connection.start(options) connection.finish() ``` -### Multiple TTS Providers (Fallback Support) - -For enhanced reliability, configure multiple TTS providers with automatic fallback: - -```python -from deepgram import ( - SettingsOptions, - Speak, - Endpoint, - Header -) - -# Primary TTS provider -primary_tts = Speak() -primary_tts.provider.type = "deepgram" -primary_tts.provider.model = "aura-2-zeus-en" - -# Fallback TTS provider -fallback_tts = Speak() -fallback_tts.provider.type = "open_ai" -fallback_tts.provider.model = "tts-1" -fallback_tts.provider.voice = "shimmer" -fallback_tts.endpoint = Endpoint() -fallback_tts.endpoint.url = "https://api.openai.com/v1/audio/speech" -fallback_tts.endpoint.headers = [ - Header(key="authorization", value="Bearer {{OPENAI_API_KEY}}") -] - -# Configure agent with fallback providers -options = SettingsOptions() -options.language = "en" -options.agent.think.provider.type = "open_ai" -options.agent.think.provider.model = "gpt-4o-mini" -options.agent.think.prompt = "You are a helpful AI assistant." -options.agent.listen.provider.type = "deepgram" -options.agent.listen.provider.model = "nova-3" - -# Option 2: Multiple TTS providers (with fallback) -options.agent.speak = [primary_tts, fallback_tts] - -options.greeting = "Hello, I'm your AI assistant with fallback TTS." - -# Start the connection -connection.start(options) -``` - This example demonstrates: - Setting up a WebSocket connection @@ -914,3 +876,4 @@ project, let us know! You can either: - [Open an issue in this repository](https://github.com/deepgram/deepgram-python-sdk/issues/new) - [Join the Deepgram Github Discussions Community](https://github.com/orgs/deepgram/discussions) - [Join the Deepgram Discord Community](https://discord.gg/xWRaCDBtW4) + From dfd5a6f6b97e7f2e50a5949e199a4d616eee2a55 Mon Sep 17 00:00:00 2001 From: John Vajda Date: Wed, 2 Jul 2025 14:17:58 -0600 Subject: [PATCH 5/5] fixes markdown linting error --- README.md | 1 - 1 file changed, 1 deletion(-) diff --git a/README.md b/README.md index b6374ccd..5c3825c9 100644 --- a/README.md +++ b/README.md @@ -876,4 +876,3 @@ project, let us know! You can either: - [Open an issue in this repository](https://github.com/deepgram/deepgram-python-sdk/issues/new) - [Join the Deepgram Github Discussions Community](https://github.com/orgs/deepgram/discussions) - [Join the Deepgram Discord Community](https://discord.gg/xWRaCDBtW4) -