• • Getting Started with Speech to Text
• Pre-Recorded Audio
  • Getting Started
  • Feature Overview
  • Starter Apps
  • Tips and Tricks
• Streaming Audio
  • Conversational STT for Voice Agents (Flux)
  • Transcription (Nova-3)
  • Compare Flux to Nova-3
• Models and Languages
  • Models & Languages Overview
  • Languages Support
  • Language Detection
  • Multilingual Codeswitching
  • Model Options
  • Version
• Formatting
  • Speaker Diarization
  • Dictation
  • Filler Words
  • Measurements
  • Numerals
  • Paragraphs
  • Profanity Filtering
  • Punctuation
  • Redaction
  • Smart Formatting
  • Supported Entity Types
  • Utterances
  • Utterance Split
• Custom Vocabulary
  • Find and Replace
  • Keyterm Prompting
  • Keywords
  • Search
• Media Input Settings
  • Channels
  • Encoding
  • Multichannel
  • Sample Rate
• Results Processing
  • STT Callback
  • STT Tagging
  • Extra Metadata
• Migrating
  • Migrating From Amazon Web Services (AWS) Transcribe to Deepgram
  • Migrating From Google Speech-to-Text (STT) to Deepgram
  • Migrating From OpenAI Whisper to Deepgram
  • Migrating from AssemblyAI Speech-to-Text to Deepgram

• Important: Flux Connection Requirements
• Audio Format Requirements
• Configurable Parameters
• End-of-Turn Detection Parameters
• When to Configure These Parameters
• Using Flux: SDK vs Direct WebSocket
• Let’s Build!
• 1. Install the Deepgram SDK
• 2. Add Dependencies
• 3. Install FFMPEG on your machine
• 4. Create a .env file
• 4. Set Imports and Set Audio Stream Colors
• 5. Connect to Flux and Process Audio
• 6. Complete Code Example
• Additional Flux Demos
• Building a Voice Agent with Flux

wss://api.deepgram.com/v2/listen?model=flux-general-en

1
from deepgram import AsyncDeepgramClient
2
3
client = AsyncDeepgramClient()
4
5
# SDK automatically uses /v2/listen endpoint
6
async with client.listen.v2.connect(
7
    model="flux-general-en",
8
    encoding="linear16",
9
    sample_rate="16000"
10
) as connection:
11
    # Your code here
12
    pass

1
 # Install the Deepgram Python SDK
2
 # https://github.com/deepgram/deepgram-python-sdk
3
 pip install deepgram-sdk

1
# Install python-dotenv to protect your API key
2
pip install python-dotenv

$
touch .env

$
DEEPGRAM_API_KEY="your_deepgram_api_key"

1
import asyncio
2
import subprocess
3
from dotenv import load_dotenv
4
5
# Load environment variables from .env file
6
load_dotenv()
7
8
from deepgram import AsyncDeepgramClient
9
from deepgram.core.events import EventType
10
from deepgram.extensions.types.sockets import ListenV2SocketClientResponse
11
12
# URL for the realtime streaming audio to transcribe
13
STREAM_URL = "http://stream.live.vc.bbcmedia.co.uk/bbc_world_service"
14
15
# Terminal color codes
16
class Colors:
17
    GREEN = '\033[92m'    # 0.90-1.00
18
    YELLOW = '\033[93m'   # 0.80-0.90
19
    ORANGE = '\033[91m'   # 0.70-0.80 (using red as orange isn't standard)
20
    RED = '\033[31m'      # <=0.69
21
    RESET = '\033[0m'     # Reset to default
22
23
def get_confidence_color(confidence: float) -> str:
24
    """Return the appropriate color code based on confidence score"""
25
    if confidence >= 0.90:
26
        return Colors.GREEN
27
    elif confidence >= 0.80:
28
        return Colors.YELLOW
29
    elif confidence >= 0.70:
30
        return Colors.ORANGE
31
    else:
32
        return Colors.RED

1
async def main():
2
    """Main async function to handle URL streaming to Deepgram Flux"""
3
4
    # Create the Deepgram async client
5
    client = AsyncDeepgramClient() # The API key retrieval happens automatically in the constructor
6
7
    try:
8
        # Connect to Flux with auto-detection for streaming audio
9
        # SDK automatically connects to: wss://api.deepgram.com/v2/listen?model=flux-general-en&encoding=linear16&sample_rate=16000
10
        async with client.listen.v2.connect(
11
            model="flux-general-en",
12
            encoding="linear16",
13
            sample_rate="16000"
14
        ) as connection:
15
16
            # Define message handler function
17
            def on_message(message: ListenV2SocketClientResponse) -> None:
18
                msg_type = getattr(message, "type", "Unknown")
19
20
                # Show transcription results
21
                if hasattr(message, 'transcript') and message.transcript:
22
                    print(f"🎤 {message.transcript}")
23
24
                    # Show word-level confidence with color coding
25
                    if hasattr(message, 'words') and message.words:
26
                        colored_words = []
27
                        for word in message.words:
28
                            color = get_confidence_color(word.confidence)
29
                            colored_words.append(f"{color}{word.word}({word.confidence:.2f}){Colors.RESET}")
30
                        words_info = " | ".join(colored_words)
31
                        print(f"   📝 {words_info}")
32
                elif msg_type == "Connected":
33
                    print(f"✅ Connected to Deepgram Flux - Ready for audio!")
34
35
            # Set up event handlers
36
            connection.on(EventType.OPEN, lambda _: print("Connection opened"))
37
            connection.on(EventType.MESSAGE, on_message)
38
            connection.on(EventType.CLOSE, lambda _: print("Connection closed"))
39
            connection.on(EventType.ERROR, lambda error: print(f"Caught: {error}"))
40
41
            # Start the connection listening in background (it's already async)
42
            deepgram_task = asyncio.create_task(connection.start_listening())
43
44
            # Convert BBC stream to linear16 PCM using ffmpeg
45
            print(f"Starting to stream and convert audio from: {STREAM_URL}")
46
47
            # Use ffmpeg to convert the compressed BBC stream to linear16 PCM at 16kHz
48
            ffmpeg_cmd = [
49
                'ffmpeg',
50
                '-i', STREAM_URL,           # Input: BBC World Service stream
51
                '-f', 's16le',              # Output format: 16-bit little-endian PCM (linear16)
52
                '-ar', '16000',             # Sample rate: 16kHz
53
                '-ac', '1',                 # Channels: mono
54
                '-'                         # Output to stdout
55
            ]
56
57
            try:
58
                # Start ffmpeg process
59
                process = await asyncio.create_subprocess_exec(
60
                    *ffmpeg_cmd,
61
                    stdout=asyncio.subprocess.PIPE,
62
                    stderr=asyncio.subprocess.PIPE
63
                )
64
65
                print(f"✅ Audio conversion started (BBC → linear16 PCM)")
66
67
                # Read converted PCM data and send to Deepgram
68
                # Note: 1024 bytes = ~32ms of audio at 16kHz linear16
69
                # For optimal performance, consider using ~2560 bytes (~80ms at 16kHz)
70
                while True:
71
                    chunk = await process.stdout.read(1024)
72
                    if not chunk:
73
                        break
74
75
                    # Send converted linear16 PCM data to Flux
76
                    await connection._send(chunk)
77
78
                await process.wait()
79
80
            except Exception as e:
81
                print(f"Error during audio conversion: {e}")
82
                if 'process' in locals():
83
                    stderr = await process.stderr.read()
84
                    print(f"FFmpeg error: {stderr.decode()}")
85
86
            # Wait for Deepgram task to complete (or cancel after timeout)
87
            try:
88
                await asyncio.wait_for(deepgram_task, timeout=60)
89
            except asyncio.TimeoutError:
90
                print("Stream timeout after 60 seconds")
91
                deepgram_task.cancel()
92
93
    except Exception as e:
94
        print(f"Caught: {e}")
95
96
if __name__ == "__main__":
97
    asyncio.run(main())

1
import asyncio
2
import subprocess
3
from dotenv import load_dotenv
4
5
# Load environment variables from .env file
6
load_dotenv()
7
8
from deepgram import AsyncDeepgramClient
9
from deepgram.core.events import EventType
10
from deepgram.extensions.types.sockets import ListenV2SocketClientResponse
11
12
# URL for the realtime streaming audio to transcribe
13
STREAM_URL = "http://stream.live.vc.bbcmedia.co.uk/bbc_world_service"
14
15
# Terminal color codes
16
class Colors:
17
    GREEN = '\033[92m'    # 0.90-1.00
18
    YELLOW = '\033[93m'   # 0.80-0.90
19
    ORANGE = '\033[91m'   # 0.70-0.80 (using red as orange isn't standard)
20
    RED = '\033[31m'      # <=0.69
21
    RESET = '\033[0m'     # Reset to default
22
23
def get_confidence_color(confidence: float) -> str:
24
    """Return the appropriate color code based on confidence score"""
25
    if confidence >= 0.90:
26
        return Colors.GREEN
27
    elif confidence >= 0.80:
28
        return Colors.YELLOW
29
    elif confidence >= 0.70:
30
        return Colors.ORANGE
31
    else:
32
        return Colors.RED
33
34
async def main():
35
    """Main async function to handle URL streaming to Deepgram Flux"""
36
37
    # Create the Deepgram async client
38
    client = AsyncDeepgramClient()
39
40
    try:
41
        # Connect to Flux with auto-detection for streaming audio
42
        # SDK automatically connects to: wss://api.deepgram.com/v2/listen?model=flux-general-en&encoding=linear16&sample_rate=16000
43
        async with client.listen.v2.connect(
44
            model="flux-general-en",
45
            encoding="linear16",
46
            sample_rate="16000"
47
        ) as connection:
48
49
            # Define message handler function
50
            def on_message(message: ListenV2SocketClientResponse) -> None:
51
                msg_type = getattr(message, "type", "Unknown")
52
53
                # Show transcription results
54
                if hasattr(message, 'transcript') and message.transcript:
55
                    print(f"🎤 {message.transcript}")
56
57
                    # Show word-level confidence with color coding
58
                    if hasattr(message, 'words') and message.words:
59
                        colored_words = []
60
                        for word in message.words:
61
                            color = get_confidence_color(word.confidence)
62
                            colored_words.append(f"{color}{word.word}({word.confidence:.2f}){Colors.RESET}")
63
                        words_info = " | ".join(colored_words)
64
                        print(f"   📝 {words_info}")
65
                elif msg_type == "Connected":
66
                    print(f"✅ Connected to Deepgram Flux - Ready for audio!")
67
68
            # Set up event handlers
69
            connection.on(EventType.OPEN, lambda _: print("Connection opened"))
70
            connection.on(EventType.MESSAGE, on_message)
71
            connection.on(EventType.CLOSE, lambda _: print("Connection closed"))
72
            connection.on(EventType.ERROR, lambda error: print(f"Caught: {error}"))
73
74
            # Start the connection listening in background (it's already async)
75
            deepgram_task = asyncio.create_task(connection.start_listening())
76
77
            # Convert BBC stream to linear16 PCM using ffmpeg
78
            print(f"Starting to stream and convert audio from: {STREAM_URL}")
79
80
            # Use ffmpeg to convert the compressed BBC stream to linear16 PCM at 16kHz
81
            ffmpeg_cmd = [
82
                'ffmpeg',
83
                '-i', STREAM_URL,           # Input: BBC World Service stream
84
                '-f', 's16le',              # Output format: 16-bit little-endian PCM (linear16)
85
                '-ar', '16000',             # Sample rate: 16kHz
86
                '-ac', '1',                 # Channels: mono
87
                '-'                         # Output to stdout
88
            ]
89
90
            try:
91
                # Start ffmpeg process
92
                process = await asyncio.create_subprocess_exec(
93
                    *ffmpeg_cmd,
94
                    stdout=asyncio.subprocess.PIPE,
95
                    stderr=asyncio.subprocess.PIPE
96
                )
97
98
                print(f"✅ Audio conversion started (BBC → linear16 PCM)")
99
100
                # Read converted PCM data and send to Deepgram
101
                # Note: 1024 bytes = ~32ms of audio at 16kHz linear16
102
                # For optimal performance, consider using ~2560 bytes (~80ms at 16kHz)
103
                while True:
104
                    chunk = await process.stdout.read(1024)
105
                    if not chunk:
106
                        break
107
108
                    # Send converted linear16 PCM data to Flux
109
                    await connection._send(chunk)
110
111
                await process.wait()
112
113
            except Exception as e:
114
                print(f"Error during audio conversion: {e}")
115
                if 'process' in locals():
116
                    stderr = await process.stderr.read()
117
                    print(f"FFmpeg error: {stderr.decode()}")
118
119
            # Wait for Deepgram task to complete (or cancel after timeout)
120
            try:
121
                await asyncio.wait_for(deepgram_task, timeout=60)
122
            except asyncio.TimeoutError:
123
                print("Stream timeout after 60 seconds")
124
                deepgram_task.cancel()
125
126
    except Exception as e:
127
        print(f"Caught: {e}")
128
129
if __name__ == "__main__":
130
    asyncio.run(main())