Merge pull request #83 from pragmatrix/deepgram-transcribe

Deepgram transcribe

Author: pragmatrix

.github/copilot-instructions.md

@@ -24,6 +24,9 @@
 ## Control Flow Style
 - Prefer exhaustive `match` statements for enum-based control flow instead of `if matches!(...)` shortcuts.
 
+## Streaming Services
+- Streaming transcribe/translate services drive the provider response stream and audio forwarding (plus billing) in a single `select!` loop — never a detached forwarder task — so termination and billing stay deterministic.
+
 ## Memory Promotion
 - When a durable repository-specific preference is learned during a session, write it into this file as a concise bullet if it can help future sessions.
 - Keep additions short, actionable, and scoped to coding behavior in this repository.

CONTEXT.md

@@ -23,3 +23,21 @@ These three Microsoft offerings are distinct and must not all be called "Azure".
   Note: Voice Live and Azure OpenAI Realtime are expected to converge over time;
   the `microsoft-voice-live` boundary is kept deliberately thin so the two can be
   merged later without a rewrite.
+
+- **Deepgram Flux** — Deepgram's turn-based conversational speech-to-text API
+  (`listen` v2). Unlike classic streaming recognition, it emits per-turn events
+  rather than a continuous interim/final stream. Backs the `deepgram-service`
+  crate (`DeepgramTranscribe`, registered as `deepgram-transcribe`).
+
+- **Turn** — a single contiguous span of one speaker talking, as detected by
+  Flux. A turn accumulates transcript across `Update` events and is closed by an
+  `EndOfTurn`.
+
+- **End of Turn (EOT)** — Flux's decision that the speaker has finished the
+  current turn. Governed by `eotThreshold` (confidence to close a turn) and
+  `eotTimeoutMs` (silence after which a turn is force-closed).
+
+- **Eager End of Turn** — an early, lower-confidence EOT signal (enabled by
+  `eagerEotThreshold`) that lets a downstream agent begin preparing a reply
+  before the turn is confirmed. May be retracted by a `TurnResumed` event if the
+  speaker continues.

Cargo.toml

@@ -12,6 +12,7 @@ members = [
     "filter-test",
     "services/aristech",
     "services/azure",
+    "services/deepgram",
     "services/elevenlabs",
     "services/google-dialog",
     "services/google-transcribe", 
@@ -37,6 +38,7 @@ google-dialog = { workspace = true }
 azure = { workspace = true }
 azure-speech = { workspace = true }
 aristech = { workspace = true }
+deepgram-service = { workspace = true }
 elevenlabs = { workspace = true }
 google-transcribe = { workspace = true }
 microsoft-voice-live = { workspace = true }
@@ -103,6 +105,7 @@ context-switch-core = { path = "core" }
 azure = { path = "services/azure" }
 playback = { path = "services/playback" }
 aristech = { path = "services/aristech" }
+deepgram-service = { path = "services/deepgram" }
 elevenlabs = { path = "services/elevenlabs" }
 google-transcribe = { path = "services/google-transcribe" }
 google-dialog = { path = "services/google-dialog" }

README.md

@@ -63,6 +63,9 @@ cargo run --example openai-dialog
 # Run generic transcribe example with Azure provider
 cargo run --example transcribe -- azure
 
+# Run generic transcribe example with Deepgram provider
+cargo run --example transcribe -- deepgram
+
 # Run generic transcribe example with ElevenLabs provider
 cargo run --example transcribe -- elevenlabs
 
@@ -108,6 +111,10 @@ AZURE_REGION=your_azure_region
 # ElevenLabs Configuration
 ELEVENLABS_API_KEY=your_elevenlabs_key
 
+# Deepgram Configuration
+DEEPGRAM_API_KEY=your_deepgram_key
+DEEPGRAM_ENDPOINT=wss://api.deepgram.com/v2/listen
+
 # Microsoft Voice Live Configuration
 MICROSOFT_VOICE_LIVE_API_KEY=your_voice_live_key
 MICROSOFT_VOICE_LIVE_ENDPOINT=wss://your-resource.services.ai.azure.com/voice-live/realtime

examples/transcribe.rs

@@ -4,18 +4,19 @@ use std::time::Duration;
 
 use anyhow::{Context, Result, bail};
 use clap::{Parser, ValueEnum};
+use tokio::select;
+use tokio::sync::mpsc::{channel, unbounded_channel};
+
 use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
 use openai_api_rs::realtime::types::{
     AzureSemanticVadConfig, EndOfUtteranceDetectionConfig, EndOfUtteranceDetectionModel,
     EndOfUtteranceThresholdLevel, TurnDetection,
 };
 use rodio::DeviceSinkBuilder;
-use tokio::select;
-use tokio::sync::mpsc::{channel, unbounded_channel};
 
 use context_switch::services::{
-    AristechTranscribe, AzureTranscribe, ElevenLabsTranscribe, GoogleTranscribe,
-    MicrosoftVoiceLiveTranscribe,
+    AristechTranscribe, AzureTranscribe, DeepgramTranscribe, ElevenLabsTranscribe,
+    GoogleTranscribe, MicrosoftVoiceLiveTranscribe,
 };
 use context_switch::{AudioConsumer, InputModality, OutputModality};
 use context_switch_core::language::Languages;
@@ -51,6 +52,8 @@ enum Provider {
     Aristech,
     #[value(name = "voice-live")]
     VoiceLive,
+    #[value(name = "deepgram")]
+    Deepgram,
 }
 
 #[tokio::main]
@@ -240,11 +243,10 @@ async fn start_conversation(
     diarization: bool,
     conversation: Conversation,
 ) -> Result<()> {
+    validate_provider_args(provider, model, region, diarization)?;
+
     match provider {
         Provider::Azure => {
-            if region.is_some() {
-                bail!("--region is only supported for the google provider");
-            }
             let params = azure::transcribe::Params {
                 endpoint: env::var("AZURE_ENDPOINT")
                     .ok()
@@ -259,12 +261,6 @@ async fn start_conversation(
             AzureTranscribe.conversation(params, conversation).await
         }
         Provider::Elevenlabs => {
-            if diarization {
-                bail!("--diarization is only supported for the azure provider");
-            }
-            if region.is_some() {
-                bail!("--region is only supported for the google provider");
-            }
             let language = Some(
                 languages
                     .single()
@@ -317,12 +313,6 @@ async fn start_conversation(
             GoogleTranscribe.conversation(params, conversation).await
         }
         Provider::Aristech => {
-            if diarization {
-                bail!("--diarization is only supported for the azure provider");
-            }
-            if region.is_some() {
-                bail!("--region is only supported for the google provider");
-            }
             let language = languages
                 .single()
                 .context("Aristech provider supports exactly one --language value")?
@@ -351,13 +341,6 @@ async fn start_conversation(
             AristechTranscribe.conversation(params, conversation).await
         }
         Provider::VoiceLive => {
-            if diarization {
-                bail!("--diarization is only supported for the azure provider");
-            }
-            if region.is_some() {
-                bail!("--region is only supported for the google provider");
-            }
-
             let language = Some(
                 languages
                     .single()
@@ -383,7 +366,7 @@ async fn start_conversation(
                     AzureSemanticVadConfig {
                         end_of_utterance_detection: Some(EndOfUtteranceDetectionConfig {
                             model: EndOfUtteranceDetectionModel::SmartEndOfTurnDetection,
-                            threshold_level: Some(EndOfUtteranceThresholdLevel::Low),
+                            threshold_level: Some(EndOfUtteranceThresholdLevel::Default),
                             timeout_ms: Some(5000),
                         }),
                         // remove_filler_words: Some(true),
@@ -396,5 +379,91 @@ async fn start_conversation(
                 .conversation(params, conversation)
                 .await
         }
+        Provider::Deepgram => {
+            let params = deepgram_service::transcribe::Params {
+                api_key: env::var("DEEPGRAM_API_KEY").expect("DEEPGRAM_API_KEY undefined"),
+                endpoint: env::var("DEEPGRAM_ENDPOINT").expect("DEEPGRAM_ENDPOINT undefined"),
+                language: languages.join_csv(),
+                profanity_filter: false,
+                keyterm: vec![],
+                turn_detection: deepgram_service::transcribe::TurnDetection::default(),
+            };
+
+            DeepgramTranscribe.conversation(params, conversation).await
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, Default)]
+struct ProviderCapabilities {
+    region: bool,
+    diarization: bool,
+    model: bool,
+}
+
+impl Provider {
+    fn capabilities(self) -> ProviderCapabilities {
+        let mut capabilities = ProviderCapabilities::default();
+
+        match self {
+            Provider::Azure => {
+                capabilities.diarization = true;
+                capabilities.model = true;
+            }
+            Provider::Deepgram => {}
+            Provider::Elevenlabs => {
+                capabilities.model = true;
+            }
+            Provider::Google => {
+                capabilities.region = true;
+                capabilities.diarization = true;
+                capabilities.model = true;
+            }
+            Provider::Aristech => {
+                capabilities.model = true;
+            }
+            Provider::VoiceLive => {
+                capabilities.model = true;
+            }
+        }
+
+        capabilities
     }
 }
+
+fn validate_capability(
+    option_name: &str,
+    is_used: bool,
+    capability: bool,
+    provider: Provider,
+) -> Result<()> {
+    if !is_used || capability {
+        return Ok(());
+    }
+
+    bail!(
+        "{option_name} is unsupported for provider '{}'",
+        provider
+            .to_possible_value()
+            .expect("Provider has a possible value")
+            .get_name()
+    )
+}
+
+fn validate_provider_args(
+    provider: Provider,
+    model: Option<&str>,
+    region: Option<&str>,
+    diarization: bool,
+) -> Result<()> {
+    let capabilities = provider.capabilities();
+
+    validate_capability("--model", model.is_some(), capabilities.model, provider)?;
+    validate_capability("--region", region.is_some(), capabilities.region, provider)?;
+    validate_capability(
+        "--diarization",
+        diarization,
+        capabilities.diarization,
+        provider,
+    )
+}

justfile

@@ -28,6 +28,9 @@ transcribe-google-latest-long:
 transcribe-voice-live-de:
     cargo run --example transcribe -- voice-live --language de-DE
 
+transcribe-deepgram-de:
+    cargo run --example transcribe -- deepgram --language de-DE
+
 transcribe-google-diarization:
     cargo run --example transcribe -- google --diarization --language de-DE --model chirp_3 --region eu
 

services/deepgram/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "deepgram-service"
+version = "0.1.0"
+edition.workspace = true
+
+[dependencies]
+context-switch-core = { workspace = true }
+
+deepgram = "0.10.0"
+
+anyhow = { workspace = true }
+async-trait = { workspace = true }
+bytes = { workspace = true }
+futures = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+tokio = { workspace = true }
+tracing = { workspace = true }

services/deepgram/src/lib.rs

@@ -0,0 +1,4 @@
+//! A Deepgram Flux speech-to-text service.
+
+pub mod transcribe;
+pub use transcribe::DeepgramTranscribe;