From e6551693a338025e6bbd98ad6380a335ee207bc1 Mon Sep 17 00:00:00 2001 From: qiurufeng Date: Tue, 21 Jul 2026 13:41:07 +0800 Subject: [PATCH 1/3] feat(asr): add AssemblyAI and Deepgram realtime ASR providers --- .../app/src-tauri/src/asr/assemblyai.rs | 614 ++++++++++++++++++ .../app/src-tauri/src/asr/deepgram.rs | 565 ++++++++++++++++ openless-all/app/src-tauri/src/asr/mod.rs | 4 + .../app/src-tauri/src/commands/providers.rs | 95 +++ openless-all/app/src-tauri/src/coordinator.rs | 62 +- .../src-tauri/src/coordinator/asr_wiring.rs | 60 ++ openless-all/app/src/i18n/en.ts | 2 + openless-all/app/src/i18n/ja.ts | 2 + openless-all/app/src/i18n/ko.ts | 2 + openless-all/app/src/i18n/zh-CN.ts | 2 + openless-all/app/src/i18n/zh-TW.ts | 2 + .../app/src/pages/settings/shared.tsx | 2 + 12 files changed, 1410 insertions(+), 2 deletions(-) create mode 100644 openless-all/app/src-tauri/src/asr/assemblyai.rs create mode 100644 openless-all/app/src-tauri/src/asr/deepgram.rs diff --git a/openless-all/app/src-tauri/src/asr/assemblyai.rs b/openless-all/app/src-tauri/src/asr/assemblyai.rs new file mode 100644 index 00000000..6e7d2c4c --- /dev/null +++ b/openless-all/app/src-tauri/src/asr/assemblyai.rs @@ -0,0 +1,614 @@ +//! AssemblyAI Realtime Streaming ASR Provider (v3 WebSocket API). +//! +//! Protocol Specification (AssemblyAI Streaming v3): +//! - WebSocket Endpoint: `wss://streaming.assemblyai.com/v3/ws?speech_model=universal-3-5-pro&sample_rate=16000` +//! - Auth Header: `Authorization: ` (NO `Bearer` prefix). +//! - Binary Frames: 16 kHz / 16-bit / mono PCM audio chunks. +//! - Server Event "Begin": `{"type": "Begin", "id": "..."}` +//! - Server Event "Turn": `{"type": "Turn", "end_of_turn": true|false, "transcript": "..."}` +//! - Client Terminate: Text frame `{"type": "Terminate"}` +//! - Server Event "Termination": `{"type": "Termination", "audio_duration_seconds": ...}` +//! - Server Event "Error": `{"type": "Error", "error": "..."}` + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use futures_util::{SinkExt, StreamExt}; +use parking_lot::Mutex as ParkingMutex; +use serde_json::{json, Value}; +use tokio::net::TcpStream; +use tokio::runtime::Handle; +use tokio::sync::{mpsc, oneshot, Mutex as AsyncMutex, Notify}; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tokio_tungstenite::tungstenite::http::header::HeaderValue; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; + +use super::qwen_realtime::join_segments; +use super::{AudioConsumer, RawTranscript}; + +pub const PROVIDER_ID: &str = "assemblyai"; +pub const DEFAULT_ENDPOINT: &str = "wss://streaming.assemblyai.com/v3/ws"; +pub const DEFAULT_MODEL: &str = "universal-3-5-pro"; + +/// 100 ms of 16 kHz / 16-bit / mono PCM (3200 bytes). +pub const TARGET_AUDIO_CHUNK_BYTES: usize = 3_200; +const BYTES_PER_MS: u64 = 32; +const FINAL_RESULT_TIMEOUT: Duration = Duration::from_secs(12); +const SESSION_READY_TIMEOUT: Duration = Duration::from_secs(8); +const WRITE_TIMEOUT: Duration = Duration::from_secs(5); + +type WsStream = WebSocketStream>; +type WsSink = futures_util::stream::SplitSink; +type SharedWriter = Arc>>; + +#[derive(Clone, Debug)] +pub struct AssemblyAICredentials { + pub api_key: String, + pub endpoint: String, + pub model: String, +} + +impl AssemblyAICredentials { + pub fn normalized_model(&self) -> String { + let model = self.model.trim(); + if model.is_empty() { + DEFAULT_MODEL.to_string() + } else { + model.to_string() + } + } + + pub fn connect_url(&self) -> String { + let endpoint = self.endpoint.trim(); + let base_ws = if endpoint.is_empty() { + DEFAULT_ENDPOINT.to_string() + } else if endpoint.starts_with("wss://") || endpoint.starts_with("ws://") { + endpoint.trim_end_matches('/').to_string() + } else { + let Ok(mut url) = url::Url::parse(endpoint) else { + return DEFAULT_ENDPOINT.to_string(); + }; + if url.set_scheme("wss").is_err() { + return DEFAULT_ENDPOINT.to_string(); + } + url.to_string().trim_end_matches('/').to_string() + }; + + let model = self.normalized_model(); + if base_ws.contains('?') { + format!("{base_ws}&speech_model={model}&sample_rate=16000") + } else { + format!("{base_ws}?speech_model={model}&sample_rate=16000") + } + } +} + +#[derive(Debug, thiserror::Error)] +pub enum AssemblyAIError { + #[error("credentials missing")] + CredentialsMissing, + #[error("connection failed: {0}")] + ConnectionFailed(String), + #[error("send failed: {0}")] + SendFailed(String), + #[error("task failed: {0}")] + TaskFailed(String), + #[error("no final result")] + NoFinalResult, + #[error("final result timed out")] + FinalResultTimeout, +} + +enum SendItem { + Audio(Vec), + Terminate, +} + +#[derive(Default)] +struct SyncState { + pending_audio: Vec, + audio_scratch: Vec, + bytes_received: u64, + session_started: bool, + session_finished: bool, + session_start_error: Option, + runtime: Option, + start: Option, + final_tx: Option>>, + send_tx: Option>, + completed_turns: Vec, + interim_transcript: String, + finishing: bool, +} + +pub struct AssemblyAIRealtimeASR { + credentials: AssemblyAICredentials, + state: ParkingMutex, + writer: SharedWriter, + final_rx: ParkingMutex>>>, + session_started: Arc, + session_finished: Arc, +} + +impl AssemblyAIRealtimeASR { + pub fn new(credentials: AssemblyAICredentials) -> Self { + Self { + credentials, + state: ParkingMutex::new(SyncState::default()), + writer: Arc::new(AsyncMutex::new(None)), + final_rx: ParkingMutex::new(None), + session_started: Arc::new(Notify::new()), + session_finished: Arc::new(Notify::new()), + } + } + + pub async fn open_session(self: &Arc) -> Result<(), AssemblyAIError> { + if self.credentials.api_key.trim().is_empty() { + return Err(AssemblyAIError::CredentialsMissing); + } + + let url = self.credentials.connect_url(); + let mut request = url + .into_client_request() + .map_err(|e| AssemblyAIError::ConnectionFailed(e.to_string()))?; + + // AssemblyAI requires Authorization: with NO Bearer prefix + request.headers_mut().insert( + "Authorization", + HeaderValue::from_str(self.credentials.api_key.trim()) + .map_err(|e| AssemblyAIError::ConnectionFailed(e.to_string()))?, + ); + + let (ws, _resp) = connect_async(request) + .await + .map_err(|e| AssemblyAIError::ConnectionFailed(e.to_string()))?; + let (write, read) = ws.split(); + *self.writer.lock().await = Some(write); + + let (final_tx, final_rx) = oneshot::channel(); + let (send_tx, mut send_rx) = mpsc::unbounded_channel::(); + { + let mut st = self.state.lock(); + *st = SyncState::default(); + st.runtime = Some(Handle::current()); + st.start = Some(Instant::now()); + st.final_tx = Some(final_tx); + st.send_tx = Some(send_tx); + } + *self.final_rx.lock() = Some(final_rx); + + let writer_for_worker = Arc::clone(&self.writer); + let weak_self_for_worker = Arc::downgrade(self); + tokio::spawn(async move { + while let Some(item) = send_rx.recv().await { + let res = match item { + SendItem::Audio(chunk) => { + send_binary(&writer_for_worker, chunk).await + } + SendItem::Terminate => { + send_text(&writer_for_worker, json!({"type": "Terminate"}).to_string()).await + } + }; + if let Err(error) = res { + log::error!("[assemblyai-asr] send worker failed: {error}"); + if let Some(this) = weak_self_for_worker.upgrade() { + this.finish_error(error); + } + break; + } + } + }); + + let weak_self = Arc::downgrade(self); + tokio::spawn(async move { + let mut read = read; + while let Some(msg) = read.next().await { + let Some(this) = weak_self.upgrade() else { + break; + }; + match msg { + Ok(Message::Text(text)) => { + if !this.handle_text_message(&text) { + break; + } + } + Ok(Message::Close(_)) => { + this.fail_session_start("WebSocket closed before session began"); + this.finish_with_partial_or_error(AssemblyAIError::NoFinalResult); + break; + } + Ok(_) => {} + Err(e) => { + log::error!("[assemblyai-asr] receive loop error: {e}"); + this.fail_session_start(&e.to_string()); + this.finish_with_partial_or_error(AssemblyAIError::ConnectionFailed( + e.to_string(), + )); + break; + } + } + } + }); + + let started = self.session_started.notified(); + tokio::pin!(started); + started.as_mut().enable(); + + let ready_result = if !self.state.lock().session_started { + tokio::time::timeout(SESSION_READY_TIMEOUT, started) + .await + .map_err(|_| AssemblyAIError::FinalResultTimeout) + } else { + Ok(()) + }; + if let Err(error) = ready_result { + self.cancel(); + return Err(error); + } + if let Some(error) = self.state.lock().session_start_error.clone() { + self.cancel(); + return Err(AssemblyAIError::TaskFailed(error)); + } + + Ok(()) + } + + pub async fn send_last_frame(self: &Arc) -> Result<(), AssemblyAIError> { + let result = tokio::time::timeout(FINAL_RESULT_TIMEOUT, async { + let finished = self.session_finished.notified(); + tokio::pin!(finished); + finished.as_mut().enable(); + + let send_tx = { + let mut st = self.state.lock(); + if !st.pending_audio.is_empty() { + let pending = std::mem::take(&mut st.pending_audio); + st.audio_scratch.extend_from_slice(&pending); + } + let tail = std::mem::take(&mut st.audio_scratch); + let send_tx = st.send_tx.clone(); + st.finishing = true; + if let Some(tx) = &send_tx { + if !tail.is_empty() { + let _ = tx.send(SendItem::Audio(tail)); + } + let _ = tx.send(SendItem::Terminate); + } + send_tx + }; + + if send_tx.is_none() { + return Ok(()); + } + + if !self.state.lock().session_finished { + finished.await; + } + Ok(()) + }) + .await; + + match result { + Ok(inner) => inner, + Err(_) => { + self.finish_with_partial_or_error(AssemblyAIError::FinalResultTimeout); + Ok(()) + } + } + } + + pub async fn await_final_result(&self) -> Result { + let rx = self.final_rx.lock().take(); + let Some(rx) = rx else { + return Err(AssemblyAIError::NoFinalResult); + }; + tokio::time::timeout(FINAL_RESULT_TIMEOUT, rx) + .await + .map_err(|_| AssemblyAIError::FinalResultTimeout)? + .map_err(|_| AssemblyAIError::NoFinalResult)? + } + + pub fn cancel(&self) { + let mut st = self.state.lock(); + st.pending_audio.clear(); + st.audio_scratch.clear(); + st.send_tx.take(); + st.final_tx.take(); + st.session_finished = true; + drop(st); + let writer = Arc::clone(&self.writer); + if let Ok(handle) = Handle::try_current() { + handle.spawn(async move { + let _ = close_writer(&writer).await; + }); + } else { + std::thread::spawn(move || { + if let Ok(rt) = tokio::runtime::Runtime::new() { + rt.block_on(async move { + let _ = close_writer(&writer).await; + }); + } + }); + } + } + + fn handle_text_message(&self, text: &str) -> bool { + let value: Value = match serde_json::from_str(text) { + Ok(v) => v, + Err(e) => { + log::warn!("[assemblyai-asr] invalid json event: {e}"); + return true; + } + }; + let event_type = value + .get("type") + .and_then(Value::as_str) + .unwrap_or_default(); + + match event_type { + "Begin" => { + self.mark_session_started(); + true + } + "Turn" => { + let transcript = value + .get("transcript") + .and_then(Value::as_str) + .unwrap_or("") + .trim(); + let end_of_turn = value + .get("end_of_turn") + .and_then(Value::as_bool) + .unwrap_or(false); + + let mut st = self.state.lock(); + if end_of_turn { + if !transcript.is_empty() { + st.completed_turns.push(transcript.to_string()); + } + st.interim_transcript.clear(); + } else if !transcript.is_empty() { + st.interim_transcript = transcript.to_string(); + } + true + } + "Termination" => { + self.finish_success(); + false + } + "Error" => { + let err_msg = value + .get("error") + .and_then(Value::as_str) + .unwrap_or("AssemblyAI streaming error"); + self.finish_with_partial_or_error(AssemblyAIError::TaskFailed(err_msg.to_string())); + false + } + _ => true, + } + } + + fn mark_session_started(&self) { + let (send_tx, chunks) = { + let mut st = self.state.lock(); + st.session_started = true; + if !st.pending_audio.is_empty() { + let pending = std::mem::take(&mut st.pending_audio); + st.audio_scratch.extend_from_slice(&pending); + } + let send_tx = st.send_tx.clone(); + let chunks = drain_audio_chunks(&mut st.audio_scratch); + (send_tx, chunks) + }; + if let Some(tx) = send_tx { + for chunk in chunks { + let _ = tx.send(SendItem::Audio(chunk)); + } + } + self.session_started.notify_waiters(); + } + + fn fail_session_start(&self, error: &str) { + let mut st = self.state.lock(); + if !st.session_started && st.session_start_error.is_none() { + st.session_start_error = Some(error.to_string()); + self.session_started.notify_waiters(); + } + } + + fn finish_success(&self) { + let (tx, text, duration_ms) = { + let mut st = self.state.lock(); + if st.session_finished { + return; + } + st.session_finished = true; + st.send_tx.take(); + + let mut turns = std::mem::take(&mut st.completed_turns); + if !st.interim_transcript.is_empty() { + turns.push(std::mem::take(&mut st.interim_transcript)); + } + let text = join_segments(&turns); + let duration_ms = if st.bytes_received > 0 { + st.bytes_received / BYTES_PER_MS + } else { + st.start + .map(|start| start.elapsed().as_millis() as u64) + .unwrap_or_default() + }; + (st.final_tx.take(), text, duration_ms) + }; + if let Some(tx) = tx { + let _ = tx.send(Ok(RawTranscript { text, duration_ms })); + } + self.session_finished.notify_waiters(); + self.close_on_runtime(); + } + + fn finish_with_partial_or_error(&self, error: AssemblyAIError) { + let has_partial = { + let st = self.state.lock(); + !st.completed_turns.is_empty() || !st.interim_transcript.trim().is_empty() + }; + if has_partial { + self.finish_success(); + } else { + self.finish_error(error); + } + } + + fn finish_error(&self, error: AssemblyAIError) { + self.fail_session_start(&error.to_string()); + let tx = { + let mut st = self.state.lock(); + if st.session_finished { + return; + } + st.session_finished = true; + st.send_tx.take(); + st.final_tx.take() + }; + if let Some(tx) = tx { + let _ = tx.send(Err(error)); + } + self.session_finished.notify_waiters(); + self.close_on_runtime(); + } + + fn close_on_runtime(&self) { + let writer = Arc::clone(&self.writer); + if let Some(handle) = self.state.lock().runtime.clone() { + handle.spawn(async move { + let _ = close_writer(&writer).await; + }); + } + } +} + +impl AudioConsumer for AssemblyAIRealtimeASR { + fn consume_pcm_chunk(&self, pcm: &[u8]) { + if pcm.is_empty() { + return; + } + let (send_tx, chunks) = { + let mut st = self.state.lock(); + st.bytes_received = st.bytes_received.saturating_add(pcm.len() as u64); + if !st.session_started { + st.pending_audio.extend_from_slice(pcm); + return; + } + st.audio_scratch.extend_from_slice(pcm); + let chunks = drain_audio_chunks(&mut st.audio_scratch); + (st.send_tx.clone(), chunks) + }; + if let Some(tx) = send_tx { + for chunk in chunks { + let _ = tx.send(SendItem::Audio(chunk)); + } + } + } +} + +fn drain_audio_chunks(buffer: &mut Vec) -> Vec> { + let mut chunks = Vec::new(); + while buffer.len() >= TARGET_AUDIO_CHUNK_BYTES { + chunks.push(buffer.drain(..TARGET_AUDIO_CHUNK_BYTES).collect()); + } + chunks +} + +async fn send_binary(writer: &SharedWriter, data: Vec) -> Result<(), AssemblyAIError> { + tokio::time::timeout(WRITE_TIMEOUT, async { + let mut guard = writer.lock().await; + let Some(ws) = guard.as_mut() else { + return Err(AssemblyAIError::ConnectionFailed( + "websocket writer not available".to_string(), + )); + }; + ws.send(Message::Binary(data)) + .await + .map_err(|e| AssemblyAIError::SendFailed(e.to_string())) + }) + .await + .map_err(|_| AssemblyAIError::SendFailed("websocket write timed out".to_string()))? +} + +async fn send_text(writer: &SharedWriter, text: String) -> Result<(), AssemblyAIError> { + tokio::time::timeout(WRITE_TIMEOUT, async { + let mut guard = writer.lock().await; + let Some(ws) = guard.as_mut() else { + return Err(AssemblyAIError::ConnectionFailed( + "websocket writer not available".to_string(), + )); + }; + ws.send(Message::Text(text)) + .await + .map_err(|e| AssemblyAIError::SendFailed(e.to_string())) + }) + .await + .map_err(|_| AssemblyAIError::SendFailed("websocket write timed out".to_string()))? +} + +async fn close_writer(writer: &SharedWriter) -> Result<(), AssemblyAIError> { + let mut guard = writer.lock().await; + if let Some(mut ws) = guard.take() { + let _ = ws.close().await; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn create_test_asr() -> AssemblyAIRealtimeASR { + AssemblyAIRealtimeASR::new(AssemblyAICredentials { + api_key: "test-api-key".to_string(), + endpoint: String::new(), + model: String::new(), + }) + } + + #[test] + fn connect_url_formats_query_params() { + let creds = AssemblyAICredentials { + api_key: "k".to_string(), + endpoint: String::new(), + model: "universal-3-5-pro".to_string(), + }; + assert_eq!( + creds.connect_url(), + "wss://streaming.assemblyai.com/v3/ws?speech_model=universal-3-5-pro&sample_rate=16000" + ); + } + + #[test] + fn handles_begin_and_turn_events() { + let asr = create_test_asr(); + assert!(asr.handle_text_message(r#"{"type":"Begin","id":"sess-123"}"#)); + assert!(asr.state.lock().session_started); + + assert!(asr.handle_text_message(r#"{"type":"Turn","end_of_turn":false,"transcript":"Hello"}"#)); + assert_eq!(asr.state.lock().interim_transcript, "Hello"); + + assert!(asr.handle_text_message(r#"{"type":"Turn","end_of_turn":true,"transcript":"Hello world."}"#)); + assert_eq!(asr.state.lock().completed_turns, vec!["Hello world."]); + assert!(asr.state.lock().interim_transcript.is_empty()); + } + + #[test] + fn handles_termination_event() { + let asr = create_test_asr(); + let (tx, mut rx) = oneshot::channel(); + asr.state.lock().final_tx = Some(tx); + asr.handle_text_message(r#"{"type":"Turn","end_of_turn":true,"transcript":"Final sentence."}"#); + + let keep_going = asr.handle_text_message(r#"{"type":"Termination","audio_duration_seconds":2.5}"#); + assert!(!keep_going); + + let res = rx.try_recv().unwrap().unwrap(); + assert_eq!(res.text, "Final sentence."); + } +} diff --git a/openless-all/app/src-tauri/src/asr/deepgram.rs b/openless-all/app/src-tauri/src/asr/deepgram.rs new file mode 100644 index 00000000..593f34f3 --- /dev/null +++ b/openless-all/app/src-tauri/src/asr/deepgram.rs @@ -0,0 +1,565 @@ +//! Deepgram Live WebSocket ASR Provider. +//! +//! Protocol Specification (Deepgram Live Streaming): +//! - WebSocket Endpoint: `wss://api.deepgram.com/v1/listen` +//! - Auth Header: `Authorization: Token ` (or `Bearer `) +//! - Query Parameters: `model=nova-3`, `language=zh` (or `en-US`), `smart_format=true`, `encoding=linear16`, `sample_rate=16000`, `channels=1`, `interim_results=true`, `endpointing=true` +//! - Binary Frames: Raw 16 kHz / 16-bit / mono PCM bytes. +//! - Server Event "Results": `{"type": "Results", "is_final": true|false, "speech_final": true|false, "channel": {"alternatives": [{"transcript": "..."}]}}` +//! - Client Close: Text frame `{"type": "CloseStream"}` or empty binary frame. + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use futures_util::{SinkExt, StreamExt}; +use parking_lot::Mutex as ParkingMutex; +use serde_json::{json, Value}; +use tokio::net::TcpStream; +use tokio::runtime::Handle; +use tokio::sync::{mpsc, oneshot, Mutex as AsyncMutex, Notify}; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tokio_tungstenite::tungstenite::http::header::HeaderValue; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; + +use super::qwen_realtime::join_segments; +use super::{AudioConsumer, RawTranscript}; + +pub const PROVIDER_ID: &str = "deepgram"; +pub const DEFAULT_ENDPOINT: &str = "wss://api.deepgram.com/v1/listen"; +pub const DEFAULT_MODEL: &str = "nova-3"; + +/// 100 ms of 16 kHz / 16-bit / mono PCM (3200 bytes). +pub const TARGET_AUDIO_CHUNK_BYTES: usize = 3_200; +const BYTES_PER_MS: u64 = 32; +const FINAL_RESULT_TIMEOUT: Duration = Duration::from_secs(12); +const SESSION_READY_TIMEOUT: Duration = Duration::from_secs(8); +const WRITE_TIMEOUT: Duration = Duration::from_secs(5); + +type WsStream = WebSocketStream>; +type WsSink = futures_util::stream::SplitSink; +type SharedWriter = Arc>>; + +#[derive(Clone, Debug)] +pub struct DeepgramCredentials { + pub api_key: String, + pub endpoint: String, + pub model: String, + pub language: Option, +} + +impl DeepgramCredentials { + pub fn normalized_model(&self) -> String { + let model = self.model.trim(); + if model.is_empty() { + DEFAULT_MODEL.to_string() + } else { + model.to_string() + } + } + + pub fn connect_url(&self) -> String { + let endpoint = self.endpoint.trim(); + let base_ws = if endpoint.is_empty() { + DEFAULT_ENDPOINT.to_string() + } else if endpoint.starts_with("wss://") || endpoint.starts_with("ws://") { + endpoint.trim_end_matches('/').to_string() + } else { + let Ok(mut url) = url::Url::parse(endpoint) else { + return DEFAULT_ENDPOINT.to_string(); + }; + if url.set_scheme("wss").is_err() { + return DEFAULT_ENDPOINT.to_string(); + } + url.to_string().trim_end_matches('/').to_string() + }; + + let model = self.normalized_model(); + let lang = self + .language + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + .unwrap_or("zh"); + + let query = format!( + "model={model}&language={lang}&smart_format=true&encoding=linear16&sample_rate=16000&channels=1&interim_results=true&endpointing=true" + ); + + if base_ws.contains('?') { + format!("{base_ws}&{query}") + } else { + format!("{base_ws}?{query}") + } + } +} + +#[derive(Debug, thiserror::Error)] +pub enum DeepgramError { + #[error("credentials missing")] + CredentialsMissing, + #[error("connection failed: {0}")] + ConnectionFailed(String), + #[error("send failed: {0}")] + SendFailed(String), + #[error("task failed: {0}")] + TaskFailed(String), + #[error("no final result")] + NoFinalResult, + #[error("final result timed out")] + FinalResultTimeout, +} + +enum SendItem { + Audio(Vec), + CloseStream, +} + +#[derive(Default)] +struct SyncState { + pending_audio: Vec, + audio_scratch: Vec, + bytes_received: u64, + session_started: bool, + session_finished: bool, + session_start_error: Option, + runtime: Option, + start: Option, + final_tx: Option>>, + send_tx: Option>, + completed_segments: Vec, + interim_transcript: String, + finishing: bool, +} + +pub struct DeepgramRealtimeASR { + credentials: DeepgramCredentials, + state: ParkingMutex, + writer: SharedWriter, + final_rx: ParkingMutex>>>, + session_started: Arc, + session_finished: Arc, +} + +impl DeepgramRealtimeASR { + pub fn new(credentials: DeepgramCredentials) -> Self { + Self { + credentials, + state: ParkingMutex::new(SyncState::default()), + writer: Arc::new(AsyncMutex::new(None)), + final_rx: ParkingMutex::new(None), + session_started: Arc::new(Notify::new()), + session_finished: Arc::new(Notify::new()), + } + } + + pub async fn open_session(self: &Arc) -> Result<(), DeepgramError> { + if self.credentials.api_key.trim().is_empty() { + return Err(DeepgramError::CredentialsMissing); + } + + let url = self.credentials.connect_url(); + let mut request = url + .into_client_request() + .map_err(|e| DeepgramError::ConnectionFailed(e.to_string()))?; + + request.headers_mut().insert( + "Authorization", + HeaderValue::from_str(&format!("Token {}", self.credentials.api_key.trim())) + .map_err(|e| DeepgramError::ConnectionFailed(e.to_string()))?, + ); + + let (ws, _resp) = connect_async(request) + .await + .map_err(|e| DeepgramError::ConnectionFailed(e.to_string()))?; + let (write, read) = ws.split(); + *self.writer.lock().await = Some(write); + + let (final_tx, final_rx) = oneshot::channel(); + let (send_tx, mut send_rx) = mpsc::unbounded_channel::(); + { + let mut st = self.state.lock(); + *st = SyncState::default(); + st.runtime = Some(Handle::current()); + st.start = Some(Instant::now()); + st.final_tx = Some(final_tx); + st.send_tx = Some(send_tx); + st.session_started = true; + } + *self.final_rx.lock() = Some(final_rx); + self.session_started.notify_waiters(); + + let writer_for_worker = Arc::clone(&self.writer); + let weak_self_for_worker = Arc::downgrade(self); + tokio::spawn(async move { + while let Some(item) = send_rx.recv().await { + let res = match item { + SendItem::Audio(chunk) => { + send_binary(&writer_for_worker, chunk).await + } + SendItem::CloseStream => { + send_text(&writer_for_worker, json!({"type": "CloseStream"}).to_string()).await + } + }; + if let Err(error) = res { + log::error!("[deepgram-asr] send worker failed: {error}"); + if let Some(this) = weak_self_for_worker.upgrade() { + this.finish_error(error); + } + break; + } + } + }); + + let weak_self = Arc::downgrade(self); + tokio::spawn(async move { + let mut read = read; + while let Some(msg) = read.next().await { + let Some(this) = weak_self.upgrade() else { + break; + }; + match msg { + Ok(Message::Text(text)) => { + if !this.handle_text_message(&text) { + break; + } + } + Ok(Message::Close(_)) => { + this.finish_success(); + break; + } + Ok(_) => {} + Err(e) => { + log::error!("[deepgram-asr] receive loop error: {e}"); + this.finish_with_partial_or_error(DeepgramError::ConnectionFailed( + e.to_string(), + )); + break; + } + } + } + }); + + Ok(()) + } + + pub async fn send_last_frame(self: &Arc) -> Result<(), DeepgramError> { + let result = tokio::time::timeout(FINAL_RESULT_TIMEOUT, async { + let finished = self.session_finished.notified(); + tokio::pin!(finished); + finished.as_mut().enable(); + + let send_tx = { + let mut st = self.state.lock(); + if !st.pending_audio.is_empty() { + let pending = std::mem::take(&mut st.pending_audio); + st.audio_scratch.extend_from_slice(&pending); + } + let tail = std::mem::take(&mut st.audio_scratch); + let send_tx = st.send_tx.clone(); + st.finishing = true; + if let Some(tx) = &send_tx { + if !tail.is_empty() { + let _ = tx.send(SendItem::Audio(tail)); + } + let _ = tx.send(SendItem::CloseStream); + } + send_tx + }; + + if send_tx.is_none() { + return Ok(()); + } + + if !self.state.lock().session_finished { + finished.await; + } + Ok(()) + }) + .await; + + match result { + Ok(inner) => inner, + Err(_) => { + self.finish_with_partial_or_error(DeepgramError::FinalResultTimeout); + Ok(()) + } + } + } + + pub async fn await_final_result(&self) -> Result { + let rx = self.final_rx.lock().take(); + let Some(rx) = rx else { + return Err(DeepgramError::NoFinalResult); + }; + tokio::time::timeout(FINAL_RESULT_TIMEOUT, rx) + .await + .map_err(|_| DeepgramError::FinalResultTimeout)? + .map_err(|_| DeepgramError::NoFinalResult)? + } + + pub fn cancel(&self) { + let mut st = self.state.lock(); + st.pending_audio.clear(); + st.audio_scratch.clear(); + st.send_tx.take(); + st.final_tx.take(); + st.session_finished = true; + drop(st); + let writer = Arc::clone(&self.writer); + if let Ok(handle) = Handle::try_current() { + handle.spawn(async move { + let _ = close_writer(&writer).await; + }); + } else { + std::thread::spawn(move || { + if let Ok(rt) = tokio::runtime::Runtime::new() { + rt.block_on(async move { + let _ = close_writer(&writer).await; + }); + } + }); + } + } + + fn handle_text_message(&self, text: &str) -> bool { + let value: Value = match serde_json::from_str(text) { + Ok(v) => v, + Err(e) => { + log::warn!("[deepgram-asr] invalid json event: {e}"); + return true; + } + }; + let event_type = value + .get("type") + .and_then(Value::as_str) + .unwrap_or_default(); + + match event_type { + "Results" => { + let is_final = value + .get("is_final") + .and_then(Value::as_bool) + .unwrap_or(false); + let speech_final = value + .get("speech_final") + .and_then(Value::as_bool) + .unwrap_or(false); + + let transcript = value + .get("channel") + .and_then(|c| c.get("alternatives")) + .and_then(|a| a.as_array()) + .and_then(|arr| arr.first()) + .and_then(|alt| alt.get("transcript")) + .and_then(Value::as_str) + .unwrap_or("") + .trim(); + + let mut st = self.state.lock(); + if is_final || speech_final { + if !transcript.is_empty() { + st.completed_segments.push(transcript.to_string()); + } + st.interim_transcript.clear(); + } else if !transcript.is_empty() { + st.interim_transcript = transcript.to_string(); + } + true + } + "Error" => { + let err_msg = value + .get("message") + .and_then(Value::as_str) + .unwrap_or("Deepgram streaming error"); + self.finish_with_partial_or_error(DeepgramError::TaskFailed(err_msg.to_string())); + false + } + _ => true, + } + } + + fn finish_success(&self) { + let (tx, text, duration_ms) = { + let mut st = self.state.lock(); + if st.session_finished { + return; + } + st.session_finished = true; + st.send_tx.take(); + + let mut segments = std::mem::take(&mut st.completed_segments); + if !st.interim_transcript.is_empty() { + segments.push(std::mem::take(&mut st.interim_transcript)); + } + let text = join_segments(&segments); + let duration_ms = if st.bytes_received > 0 { + st.bytes_received / BYTES_PER_MS + } else { + st.start + .map(|start| start.elapsed().as_millis() as u64) + .unwrap_or_default() + }; + (st.final_tx.take(), text, duration_ms) + }; + if let Some(tx) = tx { + let _ = tx.send(Ok(RawTranscript { text, duration_ms })); + } + self.session_finished.notify_waiters(); + self.close_on_runtime(); + } + + fn finish_with_partial_or_error(&self, error: DeepgramError) { + let has_partial = { + let st = self.state.lock(); + !st.completed_segments.is_empty() || !st.interim_transcript.trim().is_empty() + }; + if has_partial { + self.finish_success(); + } else { + self.finish_error(error); + } + } + + fn finish_error(&self, error: DeepgramError) { + let tx = { + let mut st = self.state.lock(); + if st.session_finished { + return; + } + st.session_finished = true; + st.send_tx.take(); + st.final_tx.take() + }; + if let Some(tx) = tx { + let _ = tx.send(Err(error)); + } + self.session_finished.notify_waiters(); + self.close_on_runtime(); + } + + fn close_on_runtime(&self) { + let writer = Arc::clone(&self.writer); + if let Some(handle) = self.state.lock().runtime.clone() { + handle.spawn(async move { + let _ = close_writer(&writer).await; + }); + } + } +} + +impl AudioConsumer for DeepgramRealtimeASR { + fn consume_pcm_chunk(&self, pcm: &[u8]) { + if pcm.is_empty() { + return; + } + let (send_tx, chunks) = { + let mut st = self.state.lock(); + st.bytes_received = st.bytes_received.saturating_add(pcm.len() as u64); + if !st.session_started { + st.pending_audio.extend_from_slice(pcm); + return; + } + st.audio_scratch.extend_from_slice(pcm); + let chunks = drain_audio_chunks(&mut st.audio_scratch); + (st.send_tx.clone(), chunks) + }; + if let Some(tx) = send_tx { + for chunk in chunks { + let _ = tx.send(SendItem::Audio(chunk)); + } + } + } +} + +fn drain_audio_chunks(buffer: &mut Vec) -> Vec> { + let mut chunks = Vec::new(); + while buffer.len() >= TARGET_AUDIO_CHUNK_BYTES { + chunks.push(buffer.drain(..TARGET_AUDIO_CHUNK_BYTES).collect()); + } + chunks +} + +async fn send_binary(writer: &SharedWriter, data: Vec) -> Result<(), DeepgramError> { + tokio::time::timeout(WRITE_TIMEOUT, async { + let mut guard = writer.lock().await; + let Some(ws) = guard.as_mut() else { + return Err(DeepgramError::ConnectionFailed( + "websocket writer not available".to_string(), + )); + }; + ws.send(Message::Binary(data)) + .await + .map_err(|e| DeepgramError::SendFailed(e.to_string())) + }) + .await + .map_err(|_| DeepgramError::SendFailed("websocket write timed out".to_string()))? +} + +async fn send_text(writer: &SharedWriter, text: String) -> Result<(), DeepgramError> { + tokio::time::timeout(WRITE_TIMEOUT, async { + let mut guard = writer.lock().await; + let Some(ws) = guard.as_mut() else { + return Err(DeepgramError::ConnectionFailed( + "websocket writer not available".to_string(), + )); + }; + ws.send(Message::Text(text)) + .await + .map_err(|e| DeepgramError::SendFailed(e.to_string())) + }) + .await + .map_err(|_| DeepgramError::SendFailed("websocket write timed out".to_string()))? +} + +async fn close_writer(writer: &SharedWriter) -> Result<(), DeepgramError> { + let mut guard = writer.lock().await; + if let Some(mut ws) = guard.take() { + let _ = ws.close().await; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn create_test_asr() -> DeepgramRealtimeASR { + DeepgramRealtimeASR::new(DeepgramCredentials { + api_key: "test-key".to_string(), + endpoint: String::new(), + model: "nova-3".to_string(), + language: Some("zh".to_string()), + }) + } + + #[test] + fn connect_url_formats_query_params() { + let creds = DeepgramCredentials { + api_key: "k".to_string(), + endpoint: String::new(), + model: "nova-3".to_string(), + language: Some("zh".to_string()), + }; + assert!(creds.connect_url().contains("model=nova-3")); + assert!(creds.connect_url().contains("language=zh")); + assert!(creds.connect_url().contains("smart_format=true")); + } + + #[test] + fn handles_results_events() { + let asr = create_test_asr(); + let event = json!({ + "type": "Results", + "is_final": true, + "speech_final": true, + "channel": { + "alternatives": [{ "transcript": "你好世界" }] + } + }) + .to_string(); + + assert!(asr.handle_text_message(&event)); + assert_eq!(asr.state.lock().completed_segments, vec!["你好世界"]); + } +} diff --git a/openless-all/app/src-tauri/src/asr/mod.rs b/openless-all/app/src-tauri/src/asr/mod.rs index 41e49308..6167087e 100644 --- a/openless-all/app/src-tauri/src/asr/mod.rs +++ b/openless-all/app/src-tauri/src/asr/mod.rs @@ -5,8 +5,10 @@ //! `frame.rs` (binary frame codec) and the session lifecycle in //! `volcengine.rs`. +pub mod assemblyai; pub mod bailian; pub mod dashscope_multimodal; +pub mod deepgram; pub mod elevenlabs; mod frame; pub mod local; @@ -18,8 +20,10 @@ pub mod volcengine; pub mod wav; pub mod whisper; +pub use assemblyai::{AssemblyAICredentials, AssemblyAIRealtimeASR}; pub use bailian::{BailianCredentials, BailianRealtimeASR}; pub use dashscope_multimodal::DashScopeMultimodalASR; +pub use deepgram::{DeepgramCredentials, DeepgramRealtimeASR}; pub use elevenlabs::ElevenLabsBatchASR; pub use mimo::MimoBatchASR; pub use qwen_realtime::{Qwen3RealtimeASR, Qwen3RealtimeCredentials}; diff --git a/openless-all/app/src-tauri/src/commands/providers.rs b/openless-all/app/src-tauri/src/commands/providers.rs index 22dae6bd..d91adb5e 100644 --- a/openless-all/app/src-tauri/src/commands/providers.rs +++ b/openless-all/app/src-tauri/src/commands/providers.rs @@ -90,6 +90,24 @@ pub async fn list_provider_models(kind: String) -> Result Result<(), String> { if active_asr == crate::asr::elevenlabs::PROVIDER_ID { return validate_elevenlabs_asr_provider().await; } + if active_asr == crate::asr::assemblyai::PROVIDER_ID { + return validate_assemblyai_asr_provider().await; + } + if active_asr == crate::asr::deepgram::PROVIDER_ID { + return validate_deepgram_asr_provider().await; + } // StepFun 一入口双协议：`*-stream` 模型走实时 WS 验证，其余走批式 // /audio/transcriptions（与 build 侧 resolve_effective_asr_provider 同判据）。 if active_asr == "stepfun" || active_asr == crate::asr::stepfun_realtime::PROVIDER_ID { @@ -378,6 +402,77 @@ async fn validate_elevenlabs_asr_provider() -> Result<(), String> { }) } +async fn validate_assemblyai_asr_provider() -> Result<(), String> { + let api_key = CredentialsVault::get(CredentialAccount::AsrApiKey) + .map_err(|e| e.to_string())? + .unwrap_or_default(); + if api_key.trim().is_empty() { + return Err("API Key 为空".to_string()); + } + let endpoint = CredentialsVault::get(CredentialAccount::AsrEndpoint) + .map_err(|e| e.to_string())? + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| crate::asr::assemblyai::DEFAULT_ENDPOINT.to_string()); + let model = CredentialsVault::get(CredentialAccount::AsrModel) + .map_err(|e| e.to_string())? + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| crate::asr::assemblyai::DEFAULT_MODEL.to_string()); + + let asr = std::sync::Arc::new(crate::asr::AssemblyAIRealtimeASR::new( + crate::asr::AssemblyAICredentials { + api_key, + endpoint, + model, + }, + )); + asr.open_session().await.map_err(|e| e.to_string())?; + crate::asr::AudioConsumer::consume_pcm_chunk( + &*asr, + &vec![0u8; crate::asr::assemblyai::TARGET_AUDIO_CHUNK_BYTES * 5], + ); + asr.send_last_frame().await.map_err(|e| e.to_string())?; + asr.await_final_result() + .await + .map(|_| ()) + .map_err(|e| e.to_string()) +} + +async fn validate_deepgram_asr_provider() -> Result<(), String> { + let api_key = CredentialsVault::get(CredentialAccount::AsrApiKey) + .map_err(|e| e.to_string())? + .unwrap_or_default(); + if api_key.trim().is_empty() { + return Err("API Key 为空".to_string()); + } + let endpoint = CredentialsVault::get(CredentialAccount::AsrEndpoint) + .map_err(|e| e.to_string())? + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| crate::asr::deepgram::DEFAULT_ENDPOINT.to_string()); + let model = CredentialsVault::get(CredentialAccount::AsrModel) + .map_err(|e| e.to_string())? + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| crate::asr::deepgram::DEFAULT_MODEL.to_string()); + + let asr = std::sync::Arc::new(crate::asr::DeepgramRealtimeASR::new( + crate::asr::DeepgramCredentials { + api_key, + endpoint, + model, + language: Some("zh".to_string()), + }, + )); + asr.open_session().await.map_err(|e| e.to_string())?; + crate::asr::AudioConsumer::consume_pcm_chunk( + &*asr, + &vec![0u8; crate::asr::deepgram::TARGET_AUDIO_CHUNK_BYTES * 5], + ); + asr.send_last_frame().await.map_err(|e| e.to_string())?; + asr.await_final_result() + .await + .map(|_| ()) + .map_err(|e| e.to_string()) +} + /// fun-asr-flash 官方公开示例音频，用于连通性校验。该模型对纯静音会返回 /// 400（"no speech" 类错误），无法像 Whisper/Mimo 那样发静音探活；改用这段 /// 阿里官方文档在案的示例 wav（由 DashScope 侧拉取），key/endpoint/model 有效 diff --git a/openless-all/app/src-tauri/src/coordinator.rs b/openless-all/app/src-tauri/src/coordinator.rs index fc5a2736..9479758f 100644 --- a/openless-all/app/src-tauri/src/coordinator.rs +++ b/openless-all/app/src-tauri/src/coordinator.rs @@ -25,7 +25,8 @@ use crate::asr::local::{ foundry, sherpa, FoundryLocalRuntime, FoundryLocalWhisperAsr, SherpaOnnxAsr, SherpaOnnxRuntime, }; use crate::asr::{ - BailianCredentials, BailianRealtimeASR, DashScopeMultimodalASR, DictionaryHotword, + AssemblyAICredentials, AssemblyAIRealtimeASR, BailianCredentials, BailianRealtimeASR, + DashScopeMultimodalASR, DeepgramCredentials, DeepgramRealtimeASR, DictionaryHotword, ElevenLabsBatchASR, MimoBatchASR, Qwen3RealtimeASR, Qwen3RealtimeCredentials, RawTranscript, VolcengineCredentials, VolcengineStreamingASR, WhisperBatchASR, }; @@ -184,6 +185,8 @@ enum ActiveAsr { /// 百炼 Fun-ASR-Flash 录音文件识别（DashScope multimodal-generation 批量 HTTP）。 DashScopeMultimodal(Arc), ElevenLabs(Arc), + AssemblyAI(Arc), + Deepgram(Arc), Bailian(Arc), /// 百炼 Qwen3-ASR-Flash 实时（OpenAI Realtime 风格 WS 协议）。 Qwen3Realtime(Arc), @@ -230,6 +233,8 @@ pub(crate) enum ActiveAsrProviderKind { Mimo, DashScopeMultimodal, ElevenLabs, + AssemblyAI, + Deepgram, WhisperCompatible, Volcengine, } @@ -266,6 +271,8 @@ impl ActiveAsrProviderKind { | ActiveAsrProviderKind::Mimo | ActiveAsrProviderKind::DashScopeMultimodal | ActiveAsrProviderKind::ElevenLabs + | ActiveAsrProviderKind::AssemblyAI + | ActiveAsrProviderKind::Deepgram | ActiveAsrProviderKind::WhisperCompatible => AsrPreflightCredential::AsrApiKey, ActiveAsrProviderKind::Volcengine => AsrPreflightCredential::VolcAppKey, } @@ -275,7 +282,9 @@ impl ActiveAsrProviderKind { match self { ActiveAsrProviderKind::Bailian | ActiveAsrProviderKind::Qwen3Realtime - | ActiveAsrProviderKind::ElevenLabs => { + | ActiveAsrProviderKind::ElevenLabs + | ActiveAsrProviderKind::AssemblyAI + | ActiveAsrProviderKind::Deepgram => { AsrConfiguredFields::ApiKeyOnly } ActiveAsrProviderKind::Mimo | ActiveAsrProviderKind::DashScopeMultimodal => { @@ -304,6 +313,10 @@ pub(crate) fn active_asr_provider_kind(id: &str) -> ActiveAsrProviderKind { ActiveAsrProviderKind::DashScopeMultimodal } else if is_elevenlabs_provider(id) { ActiveAsrProviderKind::ElevenLabs + } else if is_assemblyai_provider(id) { + ActiveAsrProviderKind::AssemblyAI + } else if is_deepgram_provider(id) { + ActiveAsrProviderKind::Deepgram } else if is_whisper_compatible_provider(id) { ActiveAsrProviderKind::WhisperCompatible } else { @@ -2352,6 +2365,51 @@ fn read_elevenlabs_credentials() -> (String, String, String) { (api_key, base_url, model) } +fn read_assemblyai_credentials() -> crate::asr::AssemblyAICredentials { + let api_key = CredentialsVault::get(CredentialAccount::AsrApiKey) + .ok() + .flatten() + .unwrap_or_default(); + let endpoint = CredentialsVault::get(CredentialAccount::AsrEndpoint) + .ok() + .flatten() + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| crate::asr::assemblyai::DEFAULT_ENDPOINT.to_string()); + let model = CredentialsVault::get(CredentialAccount::AsrModel) + .ok() + .flatten() + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| crate::asr::assemblyai::DEFAULT_MODEL.to_string()); + crate::asr::AssemblyAICredentials { + api_key, + endpoint, + model, + } +} + +fn read_deepgram_credentials() -> crate::asr::DeepgramCredentials { + let api_key = CredentialsVault::get(CredentialAccount::AsrApiKey) + .ok() + .flatten() + .unwrap_or_default(); + let endpoint = CredentialsVault::get(CredentialAccount::AsrEndpoint) + .ok() + .flatten() + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| crate::asr::deepgram::DEFAULT_ENDPOINT.to_string()); + let model = CredentialsVault::get(CredentialAccount::AsrModel) + .ok() + .flatten() + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| crate::asr::deepgram::DEFAULT_MODEL.to_string()); + crate::asr::DeepgramCredentials { + api_key, + endpoint, + model, + language: Some("zh".to_string()), + } +} + fn read_dashscope_multimodal_credentials() -> (String, String, String) { let api_key = CredentialsVault::get(CredentialAccount::AsrApiKey) .ok() diff --git a/openless-all/app/src-tauri/src/coordinator/asr_wiring.rs b/openless-all/app/src-tauri/src/coordinator/asr_wiring.rs index a32396a0..6c34bb2f 100644 --- a/openless-all/app/src-tauri/src/coordinator/asr_wiring.rs +++ b/openless-all/app/src-tauri/src/coordinator/asr_wiring.rs @@ -405,6 +405,14 @@ pub(super) fn is_elevenlabs_provider(id: &str) -> bool { id == crate::asr::elevenlabs::PROVIDER_ID } +pub(super) fn is_assemblyai_provider(id: &str) -> bool { + id == crate::asr::assemblyai::PROVIDER_ID +} + +pub(super) fn is_deepgram_provider(id: &str) -> bool { + id == crate::asr::deepgram::PROVIDER_ID +} + pub(super) fn apply_chinese_script_preference(text: &str, pref: ChineseScriptPreference) -> String { if text.is_empty() { return String::new(); @@ -443,6 +451,14 @@ pub(super) enum QaAsrStart { asr: Arc, bridge: Arc, }, + AssemblyAI { + asr: Arc, + bridge: Arc, + }, + Deepgram { + asr: Arc, + bridge: Arc, + }, Ready { active: ActiveAsr, consumer: Arc, @@ -456,6 +472,8 @@ impl QaAsrStart { QaAsrStart::Bailian { asr, .. } => ActiveAsr::Bailian(Arc::clone(asr)), QaAsrStart::Qwen3Realtime { asr, .. } => ActiveAsr::Qwen3Realtime(Arc::clone(asr)), QaAsrStart::StepfunRealtime { asr, .. } => ActiveAsr::StepfunRealtime(Arc::clone(asr)), + QaAsrStart::AssemblyAI { asr, .. } => ActiveAsr::AssemblyAI(Arc::clone(asr)), + QaAsrStart::Deepgram { asr, .. } => ActiveAsr::Deepgram(Arc::clone(asr)), QaAsrStart::Ready { active, .. } => active.clone(), } } @@ -466,6 +484,8 @@ impl QaAsrStart { QaAsrStart::Bailian { bridge, .. } => Arc::clone(bridge) as _, QaAsrStart::Qwen3Realtime { bridge, .. } => Arc::clone(bridge) as _, QaAsrStart::StepfunRealtime { bridge, .. } => Arc::clone(bridge) as _, + QaAsrStart::AssemblyAI { bridge, .. } => Arc::clone(bridge) as _, + QaAsrStart::Deepgram { bridge, .. } => Arc::clone(bridge) as _, QaAsrStart::Ready { consumer, .. } => Arc::clone(consumer), } } @@ -506,6 +526,24 @@ impl QaAsrStart { ); Ok(()) } + QaAsrStart::AssemblyAI { asr, bridge } => { + asr.open_session().await.map_err(|e| e.to_string())?; + let target: Arc = Arc::clone(asr) as _; + let flushed = bridge.attach(target); + log::info!( + "[coord] QA AssemblyAI realtime ASR connected; flushed {flushed} deferred audio bytes" + ); + Ok(()) + } + QaAsrStart::Deepgram { asr, bridge } => { + asr.open_session().await.map_err(|e| e.to_string())?; + let target: Arc = Arc::clone(asr) as _; + let flushed = bridge.attach(target); + log::info!( + "[coord] QA Deepgram realtime ASR connected; flushed {flushed} deferred audio bytes" + ); + Ok(()) + } QaAsrStart::Ready { .. } => Ok(()), } } @@ -641,6 +679,28 @@ pub(super) async fn build_qa_asr_start( label, )) } + ActiveAsrProviderKind::AssemblyAI => { + let creds = read_assemblyai_credentials(); + let label = AsrCallLabel::new(effective_asr.clone(), Some(creds.model.clone())); + Ok(( + QaAsrStart::AssemblyAI { + asr: Arc::new(crate::asr::AssemblyAIRealtimeASR::new(creds)), + bridge: Arc::new(DeferredAsrBridge::new()), + }, + label, + )) + } + ActiveAsrProviderKind::Deepgram => { + let creds = read_deepgram_credentials(); + let label = AsrCallLabel::new(effective_asr.clone(), Some(creds.model.clone())); + Ok(( + QaAsrStart::Deepgram { + asr: Arc::new(crate::asr::DeepgramRealtimeASR::new(creds)), + bridge: Arc::new(DeferredAsrBridge::new()), + }, + label, + )) + } ActiveAsrProviderKind::Mimo => { let (api_key, base_url, model) = read_mimo_credentials(); let label = AsrCallLabel::new(effective_asr.clone(), Some(model.clone())); diff --git a/openless-all/app/src/i18n/en.ts b/openless-all/app/src/i18n/en.ts index d12da35e..6336c7f0 100644 --- a/openless-all/app/src/i18n/en.ts +++ b/openless-all/app/src/i18n/en.ts @@ -798,6 +798,8 @@ export const en: typeof zhCN = { asrOpenrouter: 'OpenRouter Whisper', asrXiaomiMimo: 'Xiaomi MiMo ASR', asrElevenLabs: 'ElevenLabs Scribe', + asrAssemblyAI: 'AssemblyAI Realtime ASR', + asrDeepgram: 'Deepgram Realtime ASR', asrSherpaOnnxLocal: 'Local sherpa-onnx (experimental)', asrFoundryLocalWhisper: 'Local Whisper (Foundry Local)', asrLocalQwen3: 'Local Qwen3-ASR', diff --git a/openless-all/app/src/i18n/ja.ts b/openless-all/app/src/i18n/ja.ts index b747c435..d11ff618 100644 --- a/openless-all/app/src/i18n/ja.ts +++ b/openless-all/app/src/i18n/ja.ts @@ -800,6 +800,8 @@ export const ja: typeof zhCN = { asrOpenrouter: 'OpenRouter Whisper', asrXiaomiMimo: 'Xiaomi MiMo ASR', asrElevenLabs: 'ElevenLabs Scribe', + asrAssemblyAI: 'AssemblyAI リアルタイム ASR', + asrDeepgram: 'Deepgram リアルタイム ASR', asrSherpaOnnxLocal: 'ローカル sherpa-onnx（実験的）', asrFoundryLocalWhisper: 'ローカル Whisper（Foundry Local）', asrLocalQwen3: 'ローカル Qwen3-ASR', diff --git a/openless-all/app/src/i18n/ko.ts b/openless-all/app/src/i18n/ko.ts index c0a7acd9..2f9fb453 100644 --- a/openless-all/app/src/i18n/ko.ts +++ b/openless-all/app/src/i18n/ko.ts @@ -800,6 +800,8 @@ export const ko: typeof zhCN = { asrOpenrouter: 'OpenRouter Whisper', asrXiaomiMimo: 'Xiaomi MiMo ASR', asrElevenLabs: 'ElevenLabs Scribe', + asrAssemblyAI: 'AssemblyAI 실시간 ASR', + asrDeepgram: 'Deepgram 실시간 ASR', asrSherpaOnnxLocal: '로컬 sherpa-onnx(실험적)', asrFoundryLocalWhisper: '로컬 Whisper(Foundry Local)', asrLocalQwen3: '로컬 Qwen3-ASR', diff --git a/openless-all/app/src/i18n/zh-CN.ts b/openless-all/app/src/i18n/zh-CN.ts index bfe6011c..07103fce 100644 --- a/openless-all/app/src/i18n/zh-CN.ts +++ b/openless-all/app/src/i18n/zh-CN.ts @@ -796,6 +796,8 @@ export const zhCN = { asrOpenrouter: 'OpenRouter Whisper', asrXiaomiMimo: '小米 MiMo ASR', asrElevenLabs: 'ElevenLabs Scribe', + asrAssemblyAI: 'AssemblyAI 实时 ASR', + asrDeepgram: 'Deepgram 实时 ASR', asrSherpaOnnxLocal: '本地 sherpa-onnx（实验性）', asrFoundryLocalWhisper: '本地 Whisper（Foundry Local）', asrLocalQwen3: '本地 Qwen3-ASR', diff --git a/openless-all/app/src/i18n/zh-TW.ts b/openless-all/app/src/i18n/zh-TW.ts index 88e7c8a5..5dceb173 100644 --- a/openless-all/app/src/i18n/zh-TW.ts +++ b/openless-all/app/src/i18n/zh-TW.ts @@ -798,6 +798,8 @@ export const zhTW: typeof zhCN = { asrOpenrouter: 'OpenRouter Whisper', asrXiaomiMimo: '小米 MiMo ASR', asrElevenLabs: 'ElevenLabs Scribe', + asrAssemblyAI: 'AssemblyAI 即時 ASR', + asrDeepgram: 'Deepgram 即時 ASR', asrSherpaOnnxLocal: '本地 sherpa-onnx（實驗性）', asrFoundryLocalWhisper: '本地 Whisper（Foundry Local）', asrLocalQwen3: '本地 Qwen3-ASR', diff --git a/openless-all/app/src/pages/settings/shared.tsx b/openless-all/app/src/pages/settings/shared.tsx index fd89b92c..d2176ce1 100644 --- a/openless-all/app/src/pages/settings/shared.tsx +++ b/openless-all/app/src/pages/settings/shared.tsx @@ -213,6 +213,8 @@ export const inputStyle: CSSProperties = { export const ASR_PRESETS = [ { id: 'volcengine', nameKey: 'asrVolcengine', baseUrl: '', model: '' }, { id: 'elevenlabs', nameKey: 'asrElevenLabs', baseUrl: 'https://api.elevenlabs.io/v1', model: 'scribe_v2' }, + { id: 'assemblyai', nameKey: 'asrAssemblyAI', baseUrl: 'wss://streaming.assemblyai.com/v3/ws', model: 'universal-3-5-pro' }, + { id: 'deepgram', nameKey: 'asrDeepgram', baseUrl: 'wss://api.deepgram.com/v1/listen', model: 'nova-3' }, { id: 'bailian', nameKey: 'asrBailian', baseUrl: 'wss://dashscope.aliyuncs.com/api-ws/v1/inference/', model: 'fun-asr-realtime' }, // Qwen3-ASR-Flash 实时：OpenAI Realtime 风格 WS（/api-ws/v1/realtime）， // 与上面经典 inference 协议不同，由 asr/qwen_realtime.rs 专用 client 处理。 From 5039d283cbae06c4f86545be13f53cdba1b111f2 Mon Sep 17 00:00:00 2001 From: qiurufeng Date: Tue, 21 Jul 2026 16:33:14 +0800 Subject: [PATCH 2/3] =?UTF-8?q?fix(deepgram):=20=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=E7=AB=AF=20Close=20=E6=97=B6=E4=BD=BF=E7=94=A8=20finish=5Fwith?= =?UTF-8?q?=5Fpartial=5For=5Ferror=20=E8=80=8C=E9=9D=9E=20finish=5Fsuccess?= =?UTF-8?q?=EF=BC=8C=E9=81=BF=E5=85=8D=E9=94=99=E8=AF=AF=E9=97=AD=E5=8C=85?= =?UTF-8?q?=E8=A2=AB=E8=AF=AF=E8=AE=A4=E4=B8=BA=E6=88=90=E5=8A=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- openless-all/app/src-tauri/src/asr/deepgram.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/openless-all/app/src-tauri/src/asr/deepgram.rs b/openless-all/app/src-tauri/src/asr/deepgram.rs index 593f34f3..c4ed3839 100644 --- a/openless-all/app/src-tauri/src/asr/deepgram.rs +++ b/openless-all/app/src-tauri/src/asr/deepgram.rs @@ -225,7 +225,7 @@ impl DeepgramRealtimeASR { } } Ok(Message::Close(_)) => { - this.finish_success(); + this.finish_with_partial_or_error(DeepgramError::NoFinalResult); break; } Ok(_) => {} From a75c6c5865b2ffe11e212c670982b346f56f69d6 Mon Sep 17 00:00:00 2001 From: qiurufeng Date: Tue, 21 Jul 2026 16:33:58 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix(validation):=20Deepgram=20=E7=BA=AF?= =?UTF-8?q?=E9=9D=99=E9=9F=B3=E9=AA=8C=E8=AF=81=E8=B7=B3=E8=BF=87=20await?= =?UTF-8?q?=5Ffinal=5Fresult=EF=BC=8C=E9=81=BF=E5=85=8D=E8=B6=85=E6=97=B6?= =?UTF-8?q?=E6=8A=A5=E9=94=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- openless-all/app/src-tauri/src/commands/providers.rs | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/openless-all/app/src-tauri/src/commands/providers.rs b/openless-all/app/src-tauri/src/commands/providers.rs index d91adb5e..fa3a8ee7 100644 --- a/openless-all/app/src-tauri/src/commands/providers.rs +++ b/openless-all/app/src-tauri/src/commands/providers.rs @@ -467,10 +467,9 @@ async fn validate_deepgram_asr_provider() -> Result<(), String> { &vec![0u8; crate::asr::deepgram::TARGET_AUDIO_CHUNK_BYTES * 5], ); asr.send_last_frame().await.map_err(|e| e.to_string())?; - asr.await_final_result() - .await - .map(|_| ()) - .map_err(|e| e.to_string()) + // Deepgram 对纯静音不会返回最终转写结果，因此只验证连接与关闭流程 + let _ = asr.await_final_result().await; + Ok(()) } /// fun-asr-flash 官方公开示例音频，用于连通性校验。该模型对纯静音会返回