Deepgram

​

Overview

• DeepgramSTTService for real-time speech recognition using Deepgram’s standard WebSocket API with support for interim results, language detection, and voice activity detection (VAD)
• DeepgramFluxSTTService for advanced conversational AI with Flux capabilities including intelligent turn detection, eager end-of-turn events, multilingual support (with flux-general-multi model), and enhanced speech processing for improved response timing
• DeepgramSageMakerSTTService for real-time speech recognition using Deepgram Nova models deployed on AWS SageMaker endpoints via HTTP/2 bidirectional streaming
• DeepgramFluxSageMakerSTTService for advanced conversational AI using Deepgram Flux models deployed on AWS SageMaker endpoints with native turn detection and low-latency streaming

​

Installation

uv add "pipecat-ai[deepgram]"

uv add "pipecat-ai[deepgram,sagemaker]"

​

Prerequisites

​

Deepgram Account Setup

1. Deepgram Account: Sign up at Deepgram Console
2. API Key: Generate an API key from your console dashboard
3. Model Selection: Choose from available transcription models and features

​

Required Environment Variables

• DEEPGRAM_API_KEY: Your Deepgram API key for authentication

​

AWS SageMaker Setup

1. AWS Account: With credentials configured (via environment variables, AWS CLI, or instance metadata)
2. SageMaker Endpoint: A deployed SageMaker endpoint with a Deepgram model (Nova for standard service, Flux for advanced turn detection)
3. Deepgram SDK: The Deepgram SDK may be needed for certain advanced configurations

​

DeepgramSTTService

​

Settings

​

Usage

from pipecat.services.deepgram.stt import DeepgramSTTService

stt = DeepgramSTTService(
    api_key=os.getenv("DEEPGRAM_API_KEY"),
)

​

With Custom Settings

from pipecat.services.deepgram.stt import DeepgramSTTService

stt = DeepgramSTTService(
    api_key=os.getenv("DEEPGRAM_API_KEY"),
    settings=DeepgramSTTService.Settings(
        model="nova-3-general",
        language="es",
        punctuate=True,
        smart_format=True,
    ),
)

​

Notes

• Finalize on VAD stop: When the pipeline’s VAD detects the user has stopped speaking, the service sends a finalize request to Deepgram for faster final transcript delivery.
• Multilingual support: Deepgram Nova models support many languages. The default is Language.EN (English). Set language="multi" in settings to enable multilingual transcription, which will detect and transcribe multiple languages within the same audio stream.
• Runtime settings updates: Changing settings via STTUpdateSettingsFrame triggers a reconnection with the new parameters. To avoid audio loss, reconnection is deferred until the current user turn ends (i.e., until UserStoppedSpeakingFrame is received). Audio frames arriving during the reconnect are buffered and replayed once the new connection is ready.

​

Event Handlers

​

DeepgramFluxSTTService

​

Settings

​

Usage

from pipecat.services.deepgram.flux.stt import DeepgramFluxSTTService

stt = DeepgramFluxSTTService(
    api_key=os.getenv("DEEPGRAM_API_KEY"),
)

​

With EagerEndOfTurn

from pipecat.services.deepgram.flux.stt import DeepgramFluxSTTService

stt = DeepgramFluxSTTService(
    api_key=os.getenv("DEEPGRAM_API_KEY"),
    settings=DeepgramFluxSTTService.Settings(
        eager_eot_threshold=0.5,
        eot_threshold=0.8,
        keyterm=["Pipecat", "Deepgram"],
    ),
)

​

Multilingual Support

from pipecat.services.deepgram.flux.stt import DeepgramFluxSTTService
from pipecat.transcriptions.language import Language

# Use flux-general-multi with language hints
stt = DeepgramFluxSTTService(
    api_key=os.getenv("DEEPGRAM_API_KEY"),
    settings=DeepgramFluxSTTService.Settings(
        model="flux-general-multi",
        language_hints=[Language.EN, Language.ES, Language.FR],
    ),
)

​

Updating Settings Mid-Stream

from pipecat.frames.frames import STTUpdateSettingsFrame
from pipecat.services.deepgram.flux.stt import DeepgramFluxSTTService
from pipecat.transcriptions.language import Language

# During pipeline execution, update settings without reconnecting
await worker.queue_frame(
    STTUpdateSettingsFrame(
        delta=DeepgramFluxSTTService.Settings(
            eot_threshold=0.8,
            keyterm=["Pipecat", "Deepgram"],
        )
    )
)

# Detect-then-lock: narrow language hints mid-stream
await worker.queue_frame(
    STTUpdateSettingsFrame(
        delta=DeepgramFluxSTTService.Settings(
            language_hints=[Language.ES],
        )
    )
)

​

Notes

• Turn management: Flux provides its own turn detection via StartOfTurn/EndOfTurn events and broadcasts UserStartedSpeakingFrame/UserStoppedSpeakingFrame directly. The service automatically requests ExternalUserTurnStrategies at start to avoid conflicting VAD-based turn management.
• VAD is optional: When Flux drives turn detection, a VAD (such as SileroVADAnalyzer) in your transport is not required for core functionality. Include one if you want useful STT metrics; omit it otherwise.
• On-the-fly configuration: Supports updating keyterm, eot_threshold, eager_eot_threshold, eot_timeout_ms, and language_hints mid-stream via STTUpdateSettingsFrame. These updates are sent as Configure messages over the existing WebSocket connection without requiring a reconnect.
• EagerEndOfTurn: Enabling eager_eot_threshold provides faster response times by predicting end-of-turn before it is confirmed. EagerEndOfTurn transcripts are pushed as InterimTranscriptionFrames. If the user resumes speaking, a TurnResumed event is fired.
• Multilingual support: Use the flux-general-multi model with language_hints to bias transcription toward specific languages (EN, ES, FR, DE, HI, RU, PT, JA, IT, NL). TranscriptionFrame.language reflects the detected language for each turn. Omit hints for auto-detection or pass a subset to bias toward expected languages.

​

Event Handlers

@stt.event_handler("on_start_of_turn")
async def on_start_of_turn(service, transcript):
    print(f"Turn started: {transcript}")

@stt.event_handler("on_end_of_turn")
async def on_end_of_turn(service, transcript):
    print(f"Turn ended: {transcript}")

@stt.event_handler("on_eager_end_of_turn")
async def on_eager_end_of_turn(service, transcript):
    print(f"Early end-of-turn prediction: {transcript}")

​

DeepgramSageMakerSTTService

​

Settings

​

Usage

from pipecat.services.deepgram.sagemaker.stt import DeepgramSageMakerSTTService

stt = DeepgramSageMakerSTTService(
    endpoint_name=os.getenv("SAGEMAKER_STT_ENDPOINT_NAME"),
    region=os.getenv("AWS_REGION"),
    settings=DeepgramSageMakerSTTService.Settings(
        model="nova-3",
        language="en",
        interim_results=True,
        punctuate=True,
    ),
)

​

Notes

• Finalize on VAD stop: Like DeepgramSTTService, the SageMaker service sends a finalize request when the pipeline’s VAD detects the user has stopped speaking.
• SageMaker deployment: Requires a Deepgram model deployed to an AWS SageMaker endpoint. See the Deepgram SageMaker deployment guide for setup instructions.
• Keepalive: Automatically sends KeepAlive messages every 5 seconds to maintain the connection during periods of silence.

​

Event Handlers

​

DeepgramFluxSageMakerSTTService

​

Settings

​

Usage

from pipecat.services.deepgram.flux.sagemaker.stt import DeepgramFluxSageMakerSTTService

stt = DeepgramFluxSageMakerSTTService(
    endpoint_name=os.getenv("SAGEMAKER_FLUX_ENDPOINT_NAME"),
    region=os.getenv("AWS_REGION"),
)

​

With Custom Settings

from pipecat.services.deepgram.flux.sagemaker.stt import DeepgramFluxSageMakerSTTService

stt = DeepgramFluxSageMakerSTTService(
    endpoint_name=os.getenv("SAGEMAKER_FLUX_ENDPOINT_NAME"),
    region=os.getenv("AWS_REGION"),
    settings=DeepgramFluxSageMakerSTTService.Settings(
        model="flux-general-en",
        eot_threshold=0.7,
        eager_eot_threshold=0.5,
        keyterm=["Pipecat", "AI"],
    ),
)

​

Updating Settings Mid-Stream

from pipecat.frames.frames import STTUpdateSettingsFrame
from pipecat.services.deepgram.flux.sagemaker.stt import DeepgramFluxSageMakerSTTService
from pipecat.transcriptions.language import Language

# Update settings without reconnecting
await worker.queue_frame(
    STTUpdateSettingsFrame(
        delta=DeepgramFluxSageMakerSTTService.Settings(
            eot_threshold=0.8,
            keyterm=["Pipecat", "Deepgram", "SageMaker"],
            language_hints=[Language.EN],
        )
    )
)

​

Notes

• Turn management: Flux provides native turn detection via StartOfTurn/EndOfTurn events and broadcasts UserStartedSpeakingFrame/UserStoppedSpeakingFrame directly. The service automatically requests ExternalUserTurnStrategies at start to avoid conflicting VAD-based turn management.
• VAD is optional: When Flux drives turn detection, a VAD (such as SileroVADAnalyzer) in your transport is not required for core functionality. Include one if you want useful STT metrics; omit it otherwise.
• On-the-fly configuration: Supports updating keyterm, eot_threshold, eager_eot_threshold, eot_timeout_ms, and language_hints mid-stream via STTUpdateSettingsFrame. These updates are sent as Configure messages over the existing HTTP/2 connection without requiring a reconnect.
• EagerEndOfTurn: Enabling eager_eot_threshold provides faster response times by predicting end-of-turn before it is confirmed. EagerEndOfTurn transcripts are pushed as InterimTranscriptionFrames. If the user resumes speaking, a TurnResumed event is fired.
• Multilingual support: Use the flux-general-multi model with language_hints to bias transcription toward specific languages (EN, ES, FR, DE, HI, RU, PT, JA, IT, NL). TranscriptionFrame.language reflects the detected language for each turn. Omit hints for auto-detection or pass a subset to bias toward expected languages.
• SageMaker deployment: Requires a Deepgram Flux model deployed to an AWS SageMaker endpoint. Unlike Nova models, Flux provides native turn detection and does not require external VAD.
• No KeepAlive needed: The Flux protocol uses a dynamic watchdog mechanism that sends silence when needed to maintain the connection (threshold: max(chunk_duration * 2, watchdog_min_timeout)), so manual KeepAlive messages are not required.

​

Event Handlers

@stt.event_handler("on_start_of_turn")
async def on_start_of_turn(service, transcript):
    print(f"Turn started: {transcript}")

@stt.event_handler("on_end_of_turn")
async def on_end_of_turn(service, transcript):
    print(f"Turn ended: {transcript}")

@stt.event_handler("on_eager_end_of_turn")
async def on_eager_end_of_turn(service, transcript):
    print(f"Early end-of-turn prediction: {transcript}")