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, 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

pip install "pipecat-ai[deepgram]"

pip install "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.
• Deprecated vad_events: The vad_events setting is deprecated. Use Silero VAD instead.
• 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.

​

Event Handlers

@stt.event_handler("on_speech_started")
async def on_speech_started(service):
    print("User started speaking")

@stt.event_handler("on_utterance_end")
async def on_utterance_end(service):
    print("Utterance ended")

​

DeepgramFluxSTTService

​

Settings

​

Usage

from pipecat.services.deepgram.flux import DeepgramFluxSTTService

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

​

With EagerEndOfTurn

from pipecat.services.deepgram.flux 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"],
    ),
)

​

Updating Settings Mid-Stream

from pipecat.frames.frames import STTUpdateSettingsFrame
from pipecat.services.deepgram.flux import DeepgramFluxSTTSettings

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

​

Notes

• Turn management: Flux provides its own turn detection via StartOfTurn/EndOfTurn events and broadcasts UserStartedSpeakingFrame/UserStoppedSpeakingFrame directly. Use ExternalUserTurnStrategies to avoid conflicting VAD-based turn management.
• On-the-fly configuration: Supports updating keyterm, eot_threshold, eager_eot_threshold, and eot_timeout_ms 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.

​

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 DeepgramFluxSageMakerSTTSettings

# Update settings without reconnecting
await task.queue_frame(
    STTUpdateSettingsFrame(
        delta=DeepgramFluxSageMakerSTTSettings(
            eot_threshold=0.8,
            keyterm=["Pipecat", "Deepgram", "SageMaker"],
        )
    )
)

​

Notes

• Turn management: Flux provides native turn detection via StartOfTurn/EndOfTurn events and broadcasts UserStartedSpeakingFrame/UserStoppedSpeakingFrame directly. Use ExternalUserTurnStrategies to avoid conflicting VAD-based turn management.
• On-the-fly configuration: Supports updating keyterm, eot_threshold, eager_eot_threshold, and eot_timeout_ms 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.
• 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 watchdog mechanism that sends silence when needed to maintain the connection, 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}")