Deepgram Sdk Patterns

Deepgram SDK Patterns

Overview

Production patterns for the Deepgram speech-to-text SDK (deepgram-sdk). Covers pre-recorded transcription, live streaming, speaker diarization, and multi-language support with proper error handling.

Prerequisites

• pip install deepgram-sdk or npm install @deepgram/sdk
• DEEPGRAM_API_KEY environment variable
• Audio files or microphone access

Instructions

Step 1: Client Initialization

from deepgram import DeepgramClient, PrerecordedOptions, LiveOptions
import os

def get_deepgram_client() -> DeepgramClient:
    return DeepgramClient(os.environ["DEEPGRAM_API_KEY"])

import { createClient, DeepgramClient } from '@deepgram/sdk';
const deepgram = createClient(process.env.DEEPGRAM_API_KEY!);

Step 2: Pre-Recorded Transcription

def transcribe_file(file_path: str, language: str = "en") -> dict:
    client = get_deepgram_client()
    with open(file_path, "rb") as audio:
        response = client.listen.rest.v("1").transcribe_file(
            {"buffer": audio.read(), "mimetype": get_mimetype(file_path)},
            PrerecordedOptions(
                model="nova-2",
                language=language,
                smart_format=True,
                punctuate=True,
                diarize=True,
                utterances=True,
                paragraphs=True
            )
        )
    transcript = response.results.channels[0].alternatives[0]
    return {
        "text": transcript.transcript,
        "confidence": transcript.confidence,
        "words": [{"word": w.word, "start": w.start, "end": w.end, "speaker": getattr(w, 'speaker', None)}
                  for w in (transcript.words or [])]
    }

Step 3: Live Streaming Transcription

import asyncio

async def stream_microphone():
    client = get_deepgram_client()
    connection = client.listen.asyncwebsocket.v("1")

    async def on_message(self, result, **kwargs):
        transcript = result.channel.alternatives[0].transcript
        if transcript:
            print(f"[{result.type}] {transcript}")

    connection.on("Results", on_message)

    options = LiveOptions(
        model="nova-2",
        language="en",
        smart_format=True,
        interim_results=True,
        endpointing=300  # 300: timeout: 5 minutes
    )

    await connection.start(options)
    # Send audio chunks from microphone...
    # await connection.send(audio_bytes)
    await connection.finish()

Step 4: Batch Processing with Concurrency Control

import asyncio
from concurrent.futures import ThreadPoolExecutor

async def batch_transcribe(files: list[str], max_concurrent: int = 5) -> list:
    semaphore = asyncio.Semaphore(max_concurrent)
    results = []

    async def process_one(path):
        async with semaphore:
            loop = asyncio.get_event_loop()
            result = await loop.run_in_executor(None, transcribe_file, path)
            return {"file": path, **result}

    tasks = [process_one(f) for f in files]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return [r if not isinstance(r, Exception) else {"error": str(r)} for r in results]

Error Handling

Examples

Speaker-Labeled Transcript

result = transcribe_file("meeting.wav")
current_speaker = None
for word in result["words"]:
    if word["speaker"] != current_speaker:
        current_speaker = word["speaker"]
        print(f"\nSpeaker {current_speaker}:", end=" ")
    print(word["word"], end=" ")

Resources

• Deepgram SDK Python
• Deepgram API Docs

Output

• Configuration files or code changes applied to the project
• Validation report confirming correct implementation
• Summary of changes made and their rationale