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 models deployed on AWS SageMaker endpoints via HTTP/2 bidirectional 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
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.

​

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