diff --git a/openless-all/app/src-tauri/src/asr/bailian.rs b/openless-all/app/src-tauri/src/asr/bailian.rs index 2aac2fa9..ce225581 100644 --- a/openless-all/app/src-tauri/src/asr/bailian.rs +++ b/openless-all/app/src-tauri/src/asr/bailian.rs @@ -62,6 +62,16 @@ impl BailianCredentials { } } +/// Bailian 实时 ASR 走 DashScope WebSocket 网关，接口地址只接受 ws:// 或 +/// wss://。用户容易把百炼控制台的 `https://` 兼容模式 / 专属域名地址粘进来 +/// ——那是另一套 HTTP 协议，WebSocket 握手必然失败且底层报错 +/// （"URL scheme not supported"）对用户不可读。在验证入口先拦下， +/// 前端据 `bailianEndpointSchemeInvalid` 错误码给出可操作的提示。 +pub fn endpoint_scheme_is_websocket(endpoint: &str) -> bool { + let lower = endpoint.trim().to_ascii_lowercase(); + lower.starts_with("wss://") || lower.starts_with("ws://") +} + #[derive(Debug, thiserror::Error)] pub enum BailianASRError { #[error("credentials missing")] @@ -880,6 +890,23 @@ mod tests { // ---- existing tests kept ---- + #[test] + fn endpoint_scheme_accepts_websocket_only() { + assert!(endpoint_scheme_is_websocket(DEFAULT_ENDPOINT)); + assert!(endpoint_scheme_is_websocket("ws://localhost:9000/api-ws")); + assert!(endpoint_scheme_is_websocket( + " WSS://dashscope.aliyuncs.com/api-ws/v1/inference/ " + )); + // 百炼控制台的 https 兼容模式 / 专属域名地址不是 WebSocket 网关 + assert!(!endpoint_scheme_is_websocket( + "https://llm-xxx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1" + )); + assert!(!endpoint_scheme_is_websocket( + "http://dashscope.aliyuncs.com/api-ws/v1/inference/" + )); + assert!(!endpoint_scheme_is_websocket("")); + } + #[test] fn credentials_apply_default_endpoint_and_model() { let creds = BailianCredentials { diff --git a/openless-all/app/src-tauri/src/asr/mod.rs b/openless-all/app/src-tauri/src/asr/mod.rs index e1c761c4..4eb94097 100644 --- a/openless-all/app/src-tauri/src/asr/mod.rs +++ b/openless-all/app/src-tauri/src/asr/mod.rs @@ -10,12 +10,14 @@ mod frame; pub mod local; pub mod mimo; pub mod pcm; +pub mod qwen_realtime; pub mod volcengine; pub mod wav; pub mod whisper; pub use bailian::{BailianCredentials, BailianRealtimeASR}; pub use mimo::MimoBatchASR; +pub use qwen_realtime::{Qwen3RealtimeASR, Qwen3RealtimeCredentials}; pub use volcengine::{VolcengineCredentials, VolcengineStreamingASR}; pub use whisper::WhisperBatchASR; diff --git a/openless-all/app/src-tauri/src/asr/qwen_realtime.rs b/openless-all/app/src-tauri/src/asr/qwen_realtime.rs new file mode 100644 index 00000000..f3d95e82 --- /dev/null +++ b/openless-all/app/src-tauri/src/asr/qwen_realtime.rs @@ -0,0 +1,802 @@ +//! Alibaba Cloud Bailian Qwen3-ASR-Flash realtime client. +//! +//! Speaks the OpenAI Realtime-style WebSocket protocol +//! (`/api-ws/v1/realtime?model=...`) — the protocol line `bailian.rs` left as +//! a follow-up. 与经典 `/api-ws/v1/inference` 不同：音频以 base64 JSON 事件 +//! （`input_audio_buffer.append`）发送，服务端以 `server_vad` 自动断句，每句 +//! 产生一个 `conversation.item.input_audio_transcription.completed`。 +//! +//! 2026-07 线上实测确认的关键行为： +//! - `session.finish` 会先冲刷 VAD 尚未关闭的尾段（补发 completed）再回 +//! `session.finished`，说到一半松手不会丢尾巴； +//! - 纯静音 + finish 正常返回 `session.finished`（连接检查可用，无经典协议 +//! 的 EmptyAudio 问题）； +//! - `session.update` 省略 `input_audio_transcription.language` 时自动检测语种； +//! - 业务空间专属域名（`wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com`） +//! 同样承载此路径，经典 inference 路径则只在公共网关。 + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use base64::Engine; +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::{AudioConsumer, RawTranscript}; + +pub const PROVIDER_ID: &str = "bailian-qwen3-realtime"; +pub const DEFAULT_ENDPOINT: &str = "wss://dashscope.aliyuncs.com/api-ws/v1/realtime"; +pub const DEFAULT_MODEL: &str = "qwen3-asr-flash-realtime"; + +/// 100 ms of 16 kHz / 16-bit / mono PCM，与 recorder 输出及官方示例一致。 +pub const TARGET_AUDIO_CHUNK_BYTES: usize = 3_200; +const BYTES_PER_MS: u64 = 32; +const FINAL_RESULT_TIMEOUT: Duration = Duration::from_secs(12); +/// server_vad 断句静默阈值。官方默认 400ms；取 500ms 降低说话中途换气被切断的概率。 +const VAD_SILENCE_DURATION_MS: u32 = 500; + +type WsStream = WebSocketStream>; +type WsSink = futures_util::stream::SplitSink; +type SharedWriter = Arc>>; + +#[derive(Clone, Debug)] +pub struct Qwen3RealtimeCredentials { + pub api_key: String, + pub endpoint: String, + pub model: String, +} + +impl Qwen3RealtimeCredentials { + pub fn normalized_endpoint(&self) -> String { + if self.endpoint.trim().is_empty() { + return DEFAULT_ENDPOINT.to_string(); + } + self.endpoint.trim().to_string() + } + + pub fn normalized_model(&self) -> String { + let model = self.model.trim(); + if model.is_empty() { + DEFAULT_MODEL.to_string() + } else { + model.to_string() + } + } + + /// 连接 URL：`{endpoint}?model={model}`。用户若已在 endpoint 里带了 + /// `model=` 查询参数则原样使用，不重复拼接。 + pub fn connect_url(&self) -> String { + let endpoint = self.normalized_endpoint(); + if endpoint.contains("model=") { + return endpoint; + } + let sep = if endpoint.contains('?') { '&' } else { '?' }; + format!( + "{}{}model={}", + endpoint.trim_end_matches('/'), + sep, + self.normalized_model() + ) + } +} + +#[derive(Debug, thiserror::Error)] +pub enum Qwen3ASRError { + #[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), + Finish(oneshot::Sender>), +} + +#[derive(Default)] +struct SyncState { + pending_audio: Vec, + audio_scratch: Vec, + bytes_received: u64, + session_started: bool, + session_finished: bool, + runtime: Option, + start: Option, + final_tx: Option>>, + send_tx: Option>, + /// VAD 断句后按到达顺序累积的已完成句段（completed.transcript）。 + completed_segments: Vec, + /// 当前未完成句段的最新 interim 文本；completed 到达后清空。 + /// 服务端在句段开放期把累积文本放 `stash`、精修期放 `text`，取非空者。 + partial_text: String, +} + +pub struct Qwen3RealtimeASR { + credentials: Qwen3RealtimeCredentials, + state: ParkingMutex, + writer: SharedWriter, + final_rx: ParkingMutex>>>, + session_started: Arc, +} + +impl Qwen3RealtimeASR { + pub fn new(credentials: Qwen3RealtimeCredentials) -> 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()), + } + } + + pub async fn open_session(self: &Arc) -> Result<(), Qwen3ASRError> { + if self.credentials.api_key.trim().is_empty() { + return Err(Qwen3ASRError::CredentialsMissing); + } + + let url = self.credentials.connect_url(); + let mut request = url + .into_client_request() + .map_err(|e| Qwen3ASRError::ConnectionFailed(e.to_string()))?; + request.headers_mut().insert( + "Authorization", + HeaderValue::from_str(&format!("Bearer {}", self.credentials.api_key.trim())) + .map_err(|e| Qwen3ASRError::ConnectionFailed(e.to_string()))?, + ); + request.headers_mut().insert( + "OpenAI-Beta", + HeaderValue::from_static("realtime=v1"), + ); + + let (ws, _resp) = connect_async(request) + .await + .map_err(|e| Qwen3ASRError::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); + tokio::spawn(async move { + while let Some(item) = send_rx.recv().await { + match item { + SendItem::Audio(chunk) => { + if let Err(e) = + send_text(&writer_for_worker, append_audio_message(&chunk)).await + { + log::error!("[qwen3-asr] audio frame send failed: {e}"); + } + } + SendItem::Finish(done) => { + let result = send_text(&writer_for_worker, finish_session_message()) + .await + .map_err(|e| Qwen3ASRError::SendFailed(e.to_string())); + let _ = done.send(result); + } + } + } + }); + + send_text(&self.writer, session_update_message()).await?; + + 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_with_partial_or_error(Qwen3ASRError::NoFinalResult); + break; + } + Ok(_) => {} + Err(e) => { + log::error!("[qwen3-asr] receive loop error: {e}"); + this.finish_with_partial_or_error(Qwen3ASRError::ConnectionFailed( + e.to_string(), + )); + break; + } + } + } + }); + + Ok(()) + } + + pub async fn send_last_frame(&self) -> Result<(), Qwen3ASRError> { + let started = self.session_started.notified(); + tokio::pin!(started); + started.as_mut().enable(); + let ready = { + let st = self.state.lock(); + st.session_started || st.session_finished + }; + if !ready { + tokio::time::timeout(Duration::from_secs(5), started) + .await + .map_err(|_| Qwen3ASRError::FinalResultTimeout)?; + } + let (send_tx, tail_chunks) = { + let mut st = self.state.lock(); + let send_tx = st.send_tx.clone(); + if !st.pending_audio.is_empty() { + let pending = std::mem::take(&mut st.pending_audio); + st.audio_scratch.extend_from_slice(&pending); + } + let tail = if st.audio_scratch.is_empty() { + Vec::new() + } else { + vec![std::mem::take(&mut st.audio_scratch)] + }; + (send_tx, tail) + }; + let Some(send_tx) = send_tx else { + return Ok(()); + }; + for chunk in tail_chunks { + let _ = send_tx.send(SendItem::Audio(chunk)); + } + let (done_tx, done_rx) = oneshot::channel(); + send_tx + .send(SendItem::Finish(done_tx)) + .map_err(|_| Qwen3ASRError::SendFailed("send worker closed".to_string()))?; + done_rx + .await + .map_err(|_| Qwen3ASRError::SendFailed("finish ack dropped".to_string()))? + } + + pub async fn await_final_result(&self) -> Result { + let rx = self.final_rx.lock().take(); + let Some(rx) = rx else { + return Err(Qwen3ASRError::NoFinalResult); + }; + tokio::time::timeout(FINAL_RESULT_TIMEOUT, rx) + .await + .map_err(|_| Qwen3ASRError::FinalResultTimeout)? + .map_err(|_| Qwen3ASRError::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!("[qwen3-asr] invalid json event: {e}"); + return true; + } + }; + let event = value.get("type").and_then(Value::as_str).unwrap_or_default(); + match event { + "session.created" => { + self.mark_session_started(); + true + } + "conversation.item.input_audio_transcription.text" => { + self.record_partial(&value); + true + } + "conversation.item.input_audio_transcription.completed" => { + self.record_completed(&value); + true + } + "session.finished" => { + self.finish_success(); + false + } + "error" => { + let message = value + .get("error") + .and_then(|e| e.get("message")) + .and_then(Value::as_str) + .unwrap_or("realtime session error") + .to_string(); + self.finish_with_partial_or_error(Qwen3ASRError::TaskFailed(message)); + 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 record_partial(&self, value: &Value) { + // 句段开放期服务端把累积文本放 `stash`，随后的精修 pass 放 `text`； + // 两者互斥出现，取非空者作为当前句段的 interim 文本。 + let text = value + .get("text") + .and_then(Value::as_str) + .filter(|s| !s.trim().is_empty()) + .or_else(|| { + value + .get("stash") + .and_then(Value::as_str) + .filter(|s| !s.trim().is_empty()) + }); + if let Some(text) = text { + self.state.lock().partial_text = text.trim().to_string(); + } + } + + fn record_completed(&self, value: &Value) { + let Some(transcript) = value.get("transcript").and_then(Value::as_str) else { + return; + }; + let trimmed = transcript.trim(); + let mut st = self.state.lock(); + if !trimmed.is_empty() { + st.completed_segments.push(trimmed.to_string()); + } + st.partial_text.clear(); + } + + 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); + // session.finished 前若还有未 completed 的 interim 尾巴（理论上 + // finish 会冲刷出 completed，防御性兜底），拼在最后。 + if !st.partial_text.is_empty() { + segments.push(std::mem::take(&mut st.partial_text)); + } + 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.close_on_runtime(); + } + + fn finish_with_partial_or_error(&self, error: Qwen3ASRError) { + let has_partial = { + let st = self.state.lock(); + !st.completed_segments.is_empty() || !st.partial_text.trim().is_empty() + }; + if has_partial { + // 与 Bailian / Volcengine 保持一致：连接异常但已有结果时兜底返回。 + self.finish_success(); + } else { + self.finish_error(error); + } + } + + fn finish_error(&self, error: Qwen3ASRError) { + 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.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 Qwen3RealtimeASR { + 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 +} + +/// VAD 句段拼接：CJK 之间直接相连；拉丁词之间补空格，避免英文句段黏连。 +fn join_segments(segments: &[String]) -> String { + let mut joined = String::new(); + for seg in segments.iter().map(|s| s.trim()) { + if seg.is_empty() { + continue; + } + if let (Some(prev), Some(next)) = (joined.chars().last(), seg.chars().next()) { + if prev.is_ascii_alphanumeric() && next.is_ascii_alphanumeric() { + joined.push(' '); + } + } + joined.push_str(seg); + } + joined +} + +fn session_update_message() -> String { + // language 省略 => 服务端自动检测语种（2026-07 实测可用）。 + json!({ + "type": "session.update", + "session": { + "modalities": ["text"], + "input_audio_format": "pcm", + "sample_rate": 16000, + "turn_detection": { + "type": "server_vad", + "silence_duration_ms": VAD_SILENCE_DURATION_MS, + }, + }, + }) + .to_string() +} + +fn append_audio_message(pcm: &[u8]) -> String { + json!({ + "type": "input_audio_buffer.append", + "audio": base64::engine::general_purpose::STANDARD.encode(pcm), + }) + .to_string() +} + +fn finish_session_message() -> String { + json!({ "type": "session.finish" }).to_string() +} + +async fn send_text(writer: &SharedWriter, text: String) -> Result<(), Qwen3ASRError> { + let mut guard = writer.lock().await; + let Some(ws) = guard.as_mut() else { + return Err(Qwen3ASRError::ConnectionFailed( + "websocket writer not available".to_string(), + )); + }; + ws.send(Message::Text(text)) + .await + .map_err(|e| Qwen3ASRError::SendFailed(e.to_string())) +} + +async fn close_writer(writer: &SharedWriter) -> Result<(), Qwen3ASRError> { + 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() -> Qwen3RealtimeASR { + Qwen3RealtimeASR::new(Qwen3RealtimeCredentials { + api_key: "sk-test".to_string(), + endpoint: String::new(), + model: String::new(), + }) + } + + // ---- credentials / URL ---- + + #[test] + fn credentials_apply_default_endpoint_and_model() { + let creds = Qwen3RealtimeCredentials { + api_key: "sk-test".to_string(), + endpoint: String::new(), + model: String::new(), + }; + assert_eq!(creds.normalized_endpoint(), DEFAULT_ENDPOINT); + assert_eq!(creds.normalized_model(), DEFAULT_MODEL); + assert_eq!( + creds.connect_url(), + "wss://dashscope.aliyuncs.com/api-ws/v1/realtime?model=qwen3-asr-flash-realtime" + ); + } + + #[test] + fn connect_url_keeps_existing_model_query() { + let creds = Qwen3RealtimeCredentials { + api_key: "sk-test".to_string(), + endpoint: "wss://dashscope.aliyuncs.com/api-ws/v1/realtime?model=custom".to_string(), + model: "ignored".to_string(), + }; + assert_eq!( + creds.connect_url(), + "wss://dashscope.aliyuncs.com/api-ws/v1/realtime?model=custom" + ); + } + + #[test] + fn connect_url_supports_dedicated_workspace_domain() { + let creds = Qwen3RealtimeCredentials { + api_key: "sk-test".to_string(), + endpoint: "wss://llm-xxx.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime/".to_string(), + model: String::new(), + }; + assert_eq!( + creds.connect_url(), + "wss://llm-xxx.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen3-asr-flash-realtime" + ); + } + + // ---- message builders ---- + + #[test] + fn session_update_uses_pcm_16k_and_server_vad() { + let value: Value = serde_json::from_str(&session_update_message()).unwrap(); + assert_eq!(value["type"], "session.update"); + assert_eq!(value["session"]["input_audio_format"], "pcm"); + assert_eq!(value["session"]["sample_rate"], 16000); + assert_eq!(value["session"]["turn_detection"]["type"], "server_vad"); + // language 省略走服务端自动检测 + assert!(value["session"]["input_audio_transcription"].is_null()); + } + + #[test] + fn append_message_base64_encodes_audio() { + let value: Value = serde_json::from_str(&append_audio_message(b"\x01\x02\x03")).unwrap(); + assert_eq!(value["type"], "input_audio_buffer.append"); + assert_eq!(value["audio"], "AQID"); + } + + #[test] + fn finish_message_shape() { + let value: Value = serde_json::from_str(&finish_session_message()).unwrap(); + assert_eq!(value["type"], "session.finish"); + } + + // ---- event handling ---- + + fn text_event(text: &str, stash: &str) -> Value { + json!({ + "type": "conversation.item.input_audio_transcription.text", + "text": text, + "stash": stash, + }) + } + + fn completed_event(transcript: &str) -> Value { + json!({ + "type": "conversation.item.input_audio_transcription.completed", + "transcript": transcript, + }) + } + + #[test] + fn partial_prefers_text_falls_back_to_stash() { + let asr = create_test_asr(); + // 句段开放期：text 为空、stash 有累积文本 + asr.record_partial(&text_event("", "今天")); + assert_eq!(asr.state.lock().partial_text, "今天"); + // 精修期：text 有值优先 + asr.record_partial(&text_event("今天天气", "旧stash")); + assert_eq!(asr.state.lock().partial_text, "今天天气"); + // 两者皆空不覆盖已有 partial + asr.record_partial(&text_event("", "")); + assert_eq!(asr.state.lock().partial_text, "今天天气"); + } + + #[test] + fn completed_appends_segment_and_clears_partial() { + let asr = create_test_asr(); + asr.record_partial(&text_event("", "第一句")); + asr.handle_text_message(&completed_event("第一句话说完了。").to_string()); + asr.handle_text_message(&completed_event("第二句也说完了。").to_string()); + let st = asr.state.lock(); + assert_eq!( + st.completed_segments, + vec!["第一句话说完了。", "第二句也说完了。"] + ); + assert!(st.partial_text.is_empty()); + } + + #[test] + fn finish_success_joins_segments_with_trailing_partial() { + let asr = create_test_asr(); + asr.record_completed(&completed_event("第一句。")); + asr.record_partial(&text_event("", "没说完的尾巴")); + + let (tx, mut rx) = oneshot::channel(); + { + let mut st = asr.state.lock(); + st.final_tx = Some(tx); + st.bytes_received = 32_000; + } + asr.finish_success(); + let result = rx.try_recv().unwrap().unwrap(); + assert_eq!(result.text, "第一句。没说完的尾巴"); + assert_eq!(result.duration_ms, 1_000); + } + + #[test] + fn session_finished_event_stops_loop_with_result() { + let asr = create_test_asr(); + asr.record_completed(&completed_event("你好。")); + let (tx, mut rx) = oneshot::channel(); + asr.state.lock().final_tx = Some(tx); + let keep_going = + asr.handle_text_message(&json!({"type": "session.finished"}).to_string()); + assert!(!keep_going); + assert_eq!(rx.try_recv().unwrap().unwrap().text, "你好。"); + } + + #[test] + fn error_event_with_partial_returns_partial() { + let asr = create_test_asr(); + asr.record_completed(&completed_event("已识别内容。")); + let (tx, mut rx) = oneshot::channel(); + asr.state.lock().final_tx = Some(tx); + let keep_going = asr.handle_text_message( + &json!({"type": "error", "error": {"message": "boom"}}).to_string(), + ); + assert!(!keep_going); + assert_eq!(rx.try_recv().unwrap().unwrap().text, "已识别内容。"); + } + + #[test] + fn error_event_without_partial_returns_error() { + let asr = create_test_asr(); + let (tx, mut rx) = oneshot::channel(); + asr.state.lock().final_tx = Some(tx); + asr.handle_text_message(&json!({"type": "error", "error": {"message": "boom"}}).to_string()); + let err = rx.try_recv().unwrap().unwrap_err(); + assert!(matches!(err, Qwen3ASRError::TaskFailed(m) if m == "boom")); + } + + #[test] + fn empty_session_finishes_with_empty_text() { + // 连接检查场景：纯静音无任何 completed，finish 后应返回空文本成功。 + let asr = create_test_asr(); + let (tx, mut rx) = oneshot::channel(); + asr.state.lock().final_tx = Some(tx); + asr.handle_text_message(&json!({"type": "session.finished"}).to_string()); + assert_eq!(rx.try_recv().unwrap().unwrap().text, ""); + } + + // ---- join_segments ---- + + #[test] + fn join_segments_cjk_concatenates_directly() { + let segments = vec!["第一句。".to_string(), "第二句。".to_string()]; + assert_eq!(join_segments(&segments), "第一句。第二句。"); + } + + #[test] + fn join_segments_latin_inserts_space() { + let segments = vec!["hello".to_string(), "world".to_string()]; + assert_eq!(join_segments(&segments), "hello world"); + } + + #[test] + fn join_segments_skips_empty() { + let segments = vec!["".to_string(), "内容".to_string(), " ".to_string()]; + assert_eq!(join_segments(&segments), "内容"); + } + + // ---- audio buffering ---- + + #[test] + fn audio_buffered_before_session_created() { + let asr = create_test_asr(); + asr.consume_pcm_chunk(&[0u8; 100]); + let st = asr.state.lock(); + assert_eq!(st.pending_audio.len(), 100); + assert_eq!(st.bytes_received, 100); + } + + #[test] + fn drain_audio_chunks_keeps_tail_buffered() { + let mut buffer = vec![1u8; TARGET_AUDIO_CHUNK_BYTES * 2 + 17]; + let chunks = drain_audio_chunks(&mut buffer); + assert_eq!(chunks.len(), 2); + assert_eq!(buffer.len(), 17); + } +} diff --git a/openless-all/app/src-tauri/src/commands/credentials.rs b/openless-all/app/src-tauri/src/commands/credentials.rs index d5b2980b..9c72a5de 100644 --- a/openless-all/app/src-tauri/src/commands/credentials.rs +++ b/openless-all/app/src-tauri/src/commands/credentials.rs @@ -46,7 +46,9 @@ pub(crate) fn asr_configured_for_provider(provider: &str, snap: &CredentialsSnap // 本地 ASR 不依赖云端凭据。 return true; } - if provider == crate::asr::bailian::PROVIDER_ID { + if provider == crate::asr::bailian::PROVIDER_ID + || provider == crate::asr::qwen_realtime::PROVIDER_ID + { return configured(&snap.asr_api_key); } if provider == crate::asr::mimo::PROVIDER_ID { diff --git a/openless-all/app/src-tauri/src/commands/providers.rs b/openless-all/app/src-tauri/src/commands/providers.rs index d6e29f37..31cd7a87 100644 --- a/openless-all/app/src-tauri/src/commands/providers.rs +++ b/openless-all/app/src-tauri/src/commands/providers.rs @@ -28,10 +28,29 @@ pub async fn validate_provider_credentials(kind: String) -> Result Result { if kind == "asr" && CredentialsVault::get_active_asr() == crate::asr::bailian::PROVIDER_ID { + // Bailian 实时 ASR 没有模型列表 HTTP 接口，列表只能是静态的。但直接返回 + // 硬编码列表会让「拉取模型」在 Key / endpoint 全错时也显示成功，给用户 + // "已连通"的错觉（其他厂商这颗按钮是真实带 Key GET /models）。先跑一次 + // 与「验证」相同的 WebSocket 连通性检查，语义对齐后再返回静态列表。 + validate_bailian_asr_provider().await?; return Ok(ProviderModelsResult { models: vec![crate::asr::bailian::DEFAULT_MODEL.to_string()], }); } + if kind == "asr" + && CredentialsVault::get_active_asr() == crate::asr::qwen_realtime::PROVIDER_ID + { + // 与 bailian 同理：Realtime 网关无模型列表接口，先做真实连通性检查， + // 列表为官方文档在案的稳定别名 + 快照版本。 + validate_qwen3_realtime_asr_provider().await?; + return Ok(ProviderModelsResult { + models: vec![ + crate::asr::qwen_realtime::DEFAULT_MODEL.to_string(), + "qwen3-asr-flash-realtime-2026-02-10".to_string(), + "qwen3-asr-flash-realtime-2025-10-27".to_string(), + ], + }); + } if kind == "asr" && CredentialsVault::get_active_asr() == crate::asr::mimo::PROVIDER_ID { return Ok(ProviderModelsResult { models: vec![crate::asr::mimo::DEFAULT_MODEL.to_string()], @@ -187,6 +206,9 @@ async fn validate_asr_provider() -> Result<(), String> { if active_asr == crate::asr::bailian::PROVIDER_ID { return validate_bailian_asr_provider().await; } + if active_asr == crate::asr::qwen_realtime::PROVIDER_ID { + return validate_qwen3_realtime_asr_provider().await; + } if active_asr == crate::asr::mimo::PROVIDER_ID { return validate_mimo_asr_provider().await; } @@ -229,6 +251,12 @@ async fn validate_bailian_asr_provider() -> Result<(), String> { .map_err(|e| e.to_string())? .filter(|s| !s.trim().is_empty()) .unwrap_or_else(|| crate::asr::bailian::DEFAULT_ENDPOINT.to_string()); + // 协议头先行校验：填成 https://（百炼兼容模式 / 专属域名地址）时，WebSocket + // 握手报的 "URL scheme not supported" 会被前端兜底成笼统的「操作失败」， + // 用户无从定位。这里拦下并返回专用错误码，前端映射成可操作的提示。 + if !crate::asr::bailian::endpoint_scheme_is_websocket(&endpoint) { + return Err("bailianEndpointSchemeInvalid".to_string()); + } let model = CredentialsVault::get(CredentialAccount::AsrModel) .map_err(|e| e.to_string())? .filter(|s| !s.trim().is_empty()) @@ -245,9 +273,51 @@ async fn validate_bailian_asr_provider() -> Result<(), String> { }, )); asr.open_session().await.map_err(|e| e.to_string())?; + // 验证音频必须 ≥200ms：只发 1 个 100ms 静音块时 DashScope 必然返回 + // task-failed: EmptyAudio，导致有效凭据也永远验证失败（2026-07 实测边界： + // 100ms 拒、200ms 起收）。取 500ms 留余量，与 Mimo 验证的 250ms 同量级。 + crate::asr::AudioConsumer::consume_pcm_chunk( + &*asr, + &vec![0u8; crate::asr::bailian::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_qwen3_realtime_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::qwen_realtime::DEFAULT_ENDPOINT.to_string()); + if !crate::asr::bailian::endpoint_scheme_is_websocket(&endpoint) { + return Err("qwen3EndpointSchemeInvalid".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::qwen_realtime::DEFAULT_MODEL.to_string()); + let asr = std::sync::Arc::new(crate::asr::Qwen3RealtimeASR::new( + crate::asr::Qwen3RealtimeCredentials { + api_key, + endpoint, + model, + }, + )); + asr.open_session().await.map_err(|e| e.to_string())?; + // Realtime 协议对纯静音 + finish 干净返回 session.finished（2026-07 实测）， + // 无经典协议 <200ms 必报 EmptyAudio 的问题；发 500ms 与 bailian 验证对齐。 crate::asr::AudioConsumer::consume_pcm_chunk( &*asr, - &vec![0u8; crate::asr::bailian::TARGET_AUDIO_CHUNK_BYTES], + &vec![0u8; crate::asr::qwen_realtime::TARGET_AUDIO_CHUNK_BYTES * 5], ); asr.send_last_frame().await.map_err(|e| e.to_string())?; asr.await_final_result() diff --git a/openless-all/app/src-tauri/src/coordinator.rs b/openless-all/app/src-tauri/src/coordinator.rs index 99b07d86..4bc80f66 100644 --- a/openless-all/app/src-tauri/src/coordinator.rs +++ b/openless-all/app/src-tauri/src/coordinator.rs @@ -25,8 +25,9 @@ use crate::asr::local::{ foundry, sherpa, FoundryLocalRuntime, FoundryLocalWhisperAsr, SherpaOnnxAsr, SherpaOnnxRuntime, }; use crate::asr::{ - BailianCredentials, BailianRealtimeASR, DictionaryHotword, MimoBatchASR, RawTranscript, - VolcengineCredentials, VolcengineStreamingASR, WhisperBatchASR, + BailianCredentials, BailianRealtimeASR, DictionaryHotword, MimoBatchASR, Qwen3RealtimeASR, + Qwen3RealtimeCredentials, RawTranscript, VolcengineCredentials, VolcengineStreamingASR, + WhisperBatchASR, }; use crate::combo_hotkey::{ComboHotkeyError, ComboHotkeyEvent, ComboHotkeyMonitor}; use crate::coordinator_state::{ @@ -180,6 +181,8 @@ enum ActiveAsr { Whisper(Arc), Mimo(Arc), Bailian(Arc), + /// 百炼 Qwen3-ASR-Flash 实时（OpenAI Realtime 风格 WS 协议）。 + Qwen3Realtime(Arc), #[cfg(target_os = "windows")] FoundryLocalWhisper(Arc), /// Windows sherpa-onnx 本地 ASR（offline batch + 实验 online streaming）。 @@ -208,6 +211,7 @@ fn asr_transcribe_uses_global_timeout(asr: &ActiveAsr) -> bool { #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum ActiveAsrProviderKind { Bailian, + Qwen3Realtime, Mimo, WhisperCompatible, Volcengine, @@ -216,6 +220,8 @@ enum ActiveAsrProviderKind { fn active_asr_provider_kind(id: &str) -> ActiveAsrProviderKind { if is_bailian_provider(id) { ActiveAsrProviderKind::Bailian + } else if is_qwen3_realtime_provider(id) { + ActiveAsrProviderKind::Qwen3Realtime } else if is_mimo_provider(id) { ActiveAsrProviderKind::Mimo } else if is_whisper_compatible_provider(id) { @@ -1566,6 +1572,13 @@ impl Coordinator { .map_err(|_| "重新转录超时".to_string())? .map_err(|e| e.to_string())? } + ActiveAsr::Qwen3Realtime(asr) => { + asr.send_last_frame().await.map_err(|e| e.to_string())?; + tokio::time::timeout(timeout, asr.await_final_result()) + .await + .map_err(|_| "重新转录超时".to_string())? + .map_err(|e| e.to_string())? + } ActiveAsr::Whisper(w) => tokio::time::timeout(timeout, w.transcribe()) .await .map_err(|_| "重新转录超时".to_string())? @@ -1989,6 +2002,28 @@ fn read_bailian_credentials() -> BailianCredentials { } } +fn read_qwen3_realtime_credentials() -> Qwen3RealtimeCredentials { + 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::qwen_realtime::DEFAULT_ENDPOINT.to_string()); + let model = CredentialsVault::get(CredentialAccount::AsrModel) + .ok() + .flatten() + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| crate::asr::qwen_realtime::DEFAULT_MODEL.to_string()); + Qwen3RealtimeCredentials { + api_key, + endpoint, + model, + } +} + fn read_volc_credentials() -> VolcengineCredentials { let app_id = CredentialsVault::get(CredentialAccount::VolcengineAppKey) .ok() @@ -2339,6 +2374,10 @@ mod tests { active_asr_provider_kind(crate::asr::bailian::PROVIDER_ID), ActiveAsrProviderKind::Bailian ); + assert_eq!( + active_asr_provider_kind(crate::asr::qwen_realtime::PROVIDER_ID), + ActiveAsrProviderKind::Qwen3Realtime + ); assert_eq!( active_asr_provider_kind("whisper"), ActiveAsrProviderKind::WhisperCompatible 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 8f949f40..fcbc7c5e 100644 --- a/openless-all/app/src-tauri/src/coordinator/asr_wiring.rs +++ b/openless-all/app/src-tauri/src/coordinator/asr_wiring.rs @@ -106,6 +106,7 @@ pub(super) fn ensure_asr_credentials() -> Result<(), String> { if is_whisper_compatible_provider(&active_asr) || is_bailian_provider(&active_asr) + || is_qwen3_realtime_provider(&active_asr) || is_mimo_provider(&active_asr) { let api_key = CredentialsVault::get(CredentialAccount::AsrApiKey) @@ -351,6 +352,10 @@ pub(super) fn is_bailian_provider(id: &str) -> bool { id == crate::asr::bailian::PROVIDER_ID } +pub(super) fn is_qwen3_realtime_provider(id: &str) -> bool { + id == crate::asr::qwen_realtime::PROVIDER_ID +} + pub(super) fn is_mimo_provider(id: &str) -> bool { id == crate::asr::mimo::PROVIDER_ID } @@ -385,6 +390,10 @@ pub(super) enum QaAsrStart { asr: Arc, bridge: Arc, }, + Qwen3Realtime { + asr: Arc, + bridge: Arc, + }, Ready { active: ActiveAsr, consumer: Arc, @@ -396,6 +405,7 @@ impl QaAsrStart { match self { QaAsrStart::Volcengine { asr, .. } => ActiveAsr::Volcengine(Arc::clone(asr)), QaAsrStart::Bailian { asr, .. } => ActiveAsr::Bailian(Arc::clone(asr)), + QaAsrStart::Qwen3Realtime { asr, .. } => ActiveAsr::Qwen3Realtime(Arc::clone(asr)), QaAsrStart::Ready { active, .. } => active.clone(), } } @@ -404,6 +414,7 @@ impl QaAsrStart { match self { QaAsrStart::Volcengine { bridge, .. } => Arc::clone(bridge) as _, QaAsrStart::Bailian { bridge, .. } => Arc::clone(bridge) as _, + QaAsrStart::Qwen3Realtime { bridge, .. } => Arc::clone(bridge) as _, QaAsrStart::Ready { consumer, .. } => Arc::clone(consumer), } } @@ -426,6 +437,15 @@ impl QaAsrStart { ); Ok(()) } + QaAsrStart::Qwen3Realtime { 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 Qwen3 realtime ASR connected; flushed {flushed} deferred audio bytes" + ); + Ok(()) + } QaAsrStart::Ready { .. } => Ok(()), } } @@ -515,6 +535,10 @@ pub(super) async fn build_qa_asr_start(inner: &Arc, active_asr: &str) -> asr: Arc::new(BailianRealtimeASR::new(read_bailian_credentials())), bridge: Arc::new(DeferredAsrBridge::new()), }), + ActiveAsrProviderKind::Qwen3Realtime => Ok(QaAsrStart::Qwen3Realtime { + asr: Arc::new(Qwen3RealtimeASR::new(read_qwen3_realtime_credentials())), + bridge: Arc::new(DeferredAsrBridge::new()), + }), ActiveAsrProviderKind::Mimo => { let (api_key, base_url, model) = read_mimo_credentials(); let mimo = Arc::new(MimoBatchASR::new(api_key, base_url, model)); diff --git a/openless-all/app/src-tauri/src/coordinator/dictation.rs b/openless-all/app/src-tauri/src/coordinator/dictation.rs index 01384017..c6c7caf3 100644 --- a/openless-all/app/src-tauri/src/coordinator/dictation.rs +++ b/openless-all/app/src-tauri/src/coordinator/dictation.rs @@ -1609,6 +1609,83 @@ pub(super) async fn begin_session_as( let flushed_bytes = bridge.attach(target); log::info!("[coord] Bailian ASR connected; flushed {flushed_bytes} deferred audio bytes"); finish_starting_session(inner, current_session_id).await; + } else if is_qwen3_realtime_provider(&active_asr) { + // 与 Bailian 分支同构：流式 WS 会话 + DeferredAsrBridge 缓冲开链前音频。 + let asr = Arc::new(Qwen3RealtimeASR::new(read_qwen3_realtime_credentials())); + let bridge = Arc::new(DeferredAsrBridge::new()); + let consumer: Arc = bridge.clone(); + store_asr_for_session( + inner, + current_session_id, + ActiveAsr::Qwen3Realtime(Arc::clone(&asr)), + ); + start_recorder_for_starting(inner, current_session_id, &active_asr, consumer).await?; + + if let Err(e) = asr.open_session().await { + log::error!("[coord] open Qwen3 realtime ASR session failed: {e}"); + match startup_race_status_for_starting(inner, current_session_id) { + StartupRaceStatus::StaleContinuation => { + log::info!( + "[coord] stale Qwen3 realtime ASR open_session error from session {current_session_id} — ignoring" + ); + asr.cancel(); + discard_startup_resources_for_session(inner, current_session_id); + restore_prepared_windows_ime_session(inner, current_session_id); + return Ok(()); + } + StartupRaceStatus::CancelRaced => { + asr.cancel(); + discard_startup_resources_for_session(inner, current_session_id); + restore_prepared_windows_ime_session(inner, current_session_id); + set_phase_idle_if_session_matches(inner, current_session_id); + return Ok(()); + } + StartupRaceStatus::ActiveStarting => { + asr.cancel(); + } + } + discard_startup_resources_for_session(inner, current_session_id); + emit_capsule( + inner, + CapsuleState::Error, + 0.0, + 0, + Some(format!("ASR 连接失败: {e}")), + None, + ); + restore_prepared_windows_ime_session(inner, current_session_id); + set_phase_idle_if_session_matches(inner, current_session_id); + schedule_capsule_idle(inner, CAPSULE_AUTO_HIDE_DELAY_MS); + return Err(e.to_string()); + } + match startup_race_status_for_starting(inner, current_session_id) { + StartupRaceStatus::ActiveStarting => {} + StartupRaceStatus::CancelRaced => { + log::info!( + "[coord] cancel raced during Qwen3 realtime ASR open_session — aborting begin" + ); + asr.cancel(); + discard_startup_resources_for_session(inner, current_session_id); + restore_prepared_windows_ime_session(inner, current_session_id); + set_phase_idle_if_session_matches(inner, current_session_id); + return Ok(()); + } + StartupRaceStatus::StaleContinuation => { + log::info!( + "[coord] stale Qwen3 realtime ASR open_session continuation from session {current_session_id} — ignoring" + ); + asr.cancel(); + discard_startup_resources_for_session(inner, current_session_id); + restore_prepared_windows_ime_session(inner, current_session_id); + return Ok(()); + } + } + let target: Arc = asr; + let flushed_bytes = bridge.attach(target); + log::info!( + "[coord] Qwen3 realtime ASR connected; flushed {flushed_bytes} deferred audio bytes" + ); + finish_starting_session(inner, current_session_id).await; } else if is_mimo_provider(&active_asr) { let (api_key, base_url, model) = read_mimo_credentials(); let mimo = Arc::new(MimoBatchASR::new(api_key, base_url, model)); @@ -2244,6 +2321,33 @@ pub(super) async fn end_session(inner: &Arc) -> Result<(), String> { } } } + ActiveAsr::Qwen3Realtime(asr) => { + debug_assert!(uses_global_timeout); + if let Err(e) = asr.send_last_frame().await { + log::error!("[coord] Qwen3 realtime send last frame failed: {e}"); + } + let timeout_duration = std::time::Duration::from_secs(COORDINATOR_GLOBAL_TIMEOUT_SECS); + match tokio::time::timeout(timeout_duration, asr.await_final_result()).await { + Ok(Ok(r)) => Ok(r), + Ok(Err(e)) => { + log::error!("[coord] Qwen3 realtime await final failed: {e}"); + // 关闭 WebSocket 连接，避免流式 ASR 资源泄漏 + asr.cancel(); + Err(TranscribeFail::new(format!("识别失败: {e}"), e.to_string())) + } + Err(_) => { + log::error!( + "[coord] Qwen3 realtime 全局超时 {} 秒", + COORDINATOR_GLOBAL_TIMEOUT_SECS + ); + asr.cancel(); + Err(TranscribeFail::new( + "识别超时".to_string(), + "qwen3 realtime global timeout".to_string(), + )) + } + } + } #[cfg(target_os = "windows")] ActiveAsr::FoundryLocalWhisper(local) => { debug_assert!(!uses_global_timeout); diff --git a/openless-all/app/src-tauri/src/coordinator/qa_session.rs b/openless-all/app/src-tauri/src/coordinator/qa_session.rs index 094a1c6c..a9746e07 100644 --- a/openless-all/app/src-tauri/src/coordinator/qa_session.rs +++ b/openless-all/app/src-tauri/src/coordinator/qa_session.rs @@ -242,6 +242,21 @@ pub(super) async fn transcribe_overlay_dictation_asr( } } } + ActiveAsr::Qwen3Realtime(asr) => { + debug_assert!(uses_global_timeout); + if let Err(error) = asr.send_last_frame().await { + log::error!("[coord] overlay QA: Qwen3 realtime send last frame failed: {error}"); + } + let timeout_duration = std::time::Duration::from_secs(COORDINATOR_GLOBAL_TIMEOUT_SECS); + match tokio::time::timeout(timeout_duration, asr.await_final_result()).await { + Ok(Ok(raw)) => Ok(raw), + Ok(Err(error)) => Err(error.to_string()), + Err(_) => { + asr.cancel(); + Err("qwen3 realtime global timeout".to_string()) + } + } + } ActiveAsr::Whisper(whisper) => { debug_assert!(uses_global_timeout); let timeout_duration = std::time::Duration::from_secs(COORDINATOR_GLOBAL_TIMEOUT_SECS); @@ -783,6 +798,30 @@ pub(super) async fn end_qa_session(inner: &Arc) -> Result<(), String> { } } } + ActiveAsr::Qwen3Realtime(asr) => { + debug_assert!(uses_global_timeout); + if let Err(e) = asr.send_last_frame().await { + log::error!("[coord] QA: Qwen3 realtime send last frame failed: {e}"); + } + let timeout_duration = std::time::Duration::from_secs(COORDINATOR_GLOBAL_TIMEOUT_SECS); + match tokio::time::timeout(timeout_duration, asr.await_final_result()).await { + Ok(Ok(r)) => r, + Ok(Err(e)) => { + log::error!("[coord] QA: Qwen3 realtime await final failed: {e}"); + finish_qa_with_error(inner, format!("识别失败: {e}")); + return Err(e.to_string()); + } + Err(_) => { + log::error!( + "[coord] QA: Qwen3 realtime 全局超时 {} 秒", + COORDINATOR_GLOBAL_TIMEOUT_SECS + ); + asr.cancel(); + finish_qa_with_error(inner, "识别超时".to_string()); + return Err("qwen3 realtime global timeout".to_string()); + } + } + } ActiveAsr::Whisper(w) => { debug_assert!(uses_global_timeout); let timeout_duration = std::time::Duration::from_secs(COORDINATOR_GLOBAL_TIMEOUT_SECS); diff --git a/openless-all/app/src-tauri/src/coordinator/resources.rs b/openless-all/app/src-tauri/src/coordinator/resources.rs index 03184636..30fd3bdc 100644 --- a/openless-all/app/src-tauri/src/coordinator/resources.rs +++ b/openless-all/app/src-tauri/src/coordinator/resources.rs @@ -69,6 +69,7 @@ pub(super) fn cancel_active_asr(asr: ActiveAsr) { ActiveAsr::Whisper(w) => w.cancel(), ActiveAsr::Mimo(m) => m.cancel(), ActiveAsr::Bailian(b) => b.cancel(), + ActiveAsr::Qwen3Realtime(q) => q.cancel(), #[cfg(target_os = "windows")] ActiveAsr::FoundryLocalWhisper(local) => local.cancel(), #[cfg(target_os = "windows")] diff --git a/openless-all/app/src/i18n/en.ts b/openless-all/app/src/i18n/en.ts index af5ae6da..f270b2a0 100644 --- a/openless-all/app/src/i18n/en.ts +++ b/openless-all/app/src/i18n/en.ts @@ -778,6 +778,7 @@ export const en: typeof zhCN = { custom: 'Custom', asrVolcengine: 'Volcengine bigasr', asrBailian: 'Alibaba Bailian realtime ASR', + asrBailianQwen3: 'Bailian Qwen3 Realtime ASR', asrSiliconflow: 'SiliconFlow SenseVoice', asrZhipu: 'Zhipu GLM-ASR', asrGroq: 'Groq Whisper-large-v3', @@ -837,6 +838,8 @@ export const en: typeof zhCN = { providerHttpStatus: 'Provider returned HTTP {{status}}. Check the API key permissions or endpoint.', endpointMustUseHttps: 'HTTP endpoints are allowed, but API keys and audio content may leak in transit.', endpointInvalid: 'Endpoint format is invalid.', + bailianEndpointSchemeInvalid: 'Bailian realtime ASR uses the DashScope WebSocket gateway: the endpoint must start with wss:// (default: wss://dashscope.aliyuncs.com/api-ws/v1/inference/). An https:// compatible-mode URL will not work here.', + qwen3EndpointSchemeInvalid: 'Qwen3 realtime ASR uses the DashScope Realtime WebSocket gateway: the endpoint must start with wss:// (default: wss://dashscope.aliyuncs.com/api-ws/v1/realtime). An https:// URL will not work here.', responseTooLarge: 'Provider response is too large to validate safely.', asrInvalidJson: 'ASR response is not valid JSON.', asrMissingTextField: 'ASR response is missing the text field.', diff --git a/openless-all/app/src/i18n/ja.ts b/openless-all/app/src/i18n/ja.ts index 63a64688..49ad2cd7 100644 --- a/openless-all/app/src/i18n/ja.ts +++ b/openless-all/app/src/i18n/ja.ts @@ -780,6 +780,7 @@ export const ja: typeof zhCN = { custom: 'カスタム', asrVolcengine: 'Volcengine bigasr', asrBailian: 'Alibaba Bailian リアルタイム ASR', + asrBailianQwen3: 'Bailian Qwen3 リアルタイム ASR', asrSiliconflow: 'SiliconFlow SenseVoice', asrZhipu: 'Zhipu GLM-ASR', asrGroq: 'Groq Whisper-large-v3', @@ -839,6 +840,8 @@ export const ja: typeof zhCN = { providerHttpStatus: 'サプライヤーが {{status}} を返しました。API Key 権限またはエンドポイントを確認してください。', endpointMustUseHttps: 'HTTP Endpoint も使用できますが、API Key と音声内容が通信中に漏えいする可能性があります。', endpointInvalid: 'Endpoint の形式が無効です。', + bailianEndpointSchemeInvalid: 'Bailian リアルタイム ASR は DashScope の WebSocket ゲートウェイを使用します。エンドポイントは wss:// で始まる必要があります（既定: wss://dashscope.aliyuncs.com/api-ws/v1/inference/）。https:// の互換モード URL はここでは使用できません。', + qwen3EndpointSchemeInvalid: 'Qwen3 リアルタイム ASR は DashScope Realtime WebSocket ゲートウェイを使用します。エンドポイントは wss:// で始まる必要があります（既定: wss://dashscope.aliyuncs.com/api-ws/v1/realtime）。https:// の URL はここでは使用できません。', responseTooLarge: 'サプライヤーの応答が大きすぎるため、安全のため検証を停止しました。', asrInvalidJson: 'ASR の応答が有効な JSON ではありません。', asrMissingTextField: 'ASR の応答に text フィールドがありません。', diff --git a/openless-all/app/src/i18n/ko.ts b/openless-all/app/src/i18n/ko.ts index 2b33b2ab..ce86b5a4 100644 --- a/openless-all/app/src/i18n/ko.ts +++ b/openless-all/app/src/i18n/ko.ts @@ -780,6 +780,7 @@ export const ko: typeof zhCN = { custom: '사용자 정의', asrVolcengine: 'Volcengine bigasr', asrBailian: 'Alibaba Bailian 실시간 ASR', + asrBailianQwen3: 'Bailian Qwen3 실시간 ASR', asrSiliconflow: 'SiliconFlow SenseVoice', asrZhipu: 'Zhipu GLM-ASR', asrGroq: 'Groq Whisper-large-v3', @@ -839,6 +840,8 @@ export const ko: typeof zhCN = { providerHttpStatus: '공급자가 {{status}} 를 반환했습니다. API Key 권한 또는 Endpoint 를 확인해 주세요.', endpointMustUseHttps: 'HTTP Endpoint 를 사용할 수 있지만, API Key 와 음성 내용이 전송 중 유출될 수 있습니다.', endpointInvalid: 'Endpoint 형식이 올바르지 않습니다.', + bailianEndpointSchemeInvalid: 'Bailian 실시간 ASR은 DashScope WebSocket 게이트웨이를 사용합니다. 엔드포인트는 wss://로 시작해야 합니다(기본값: wss://dashscope.aliyuncs.com/api-ws/v1/inference/). https:// 호환 모드 주소는 여기서 사용할 수 없습니다.', + qwen3EndpointSchemeInvalid: 'Qwen3 실시간 ASR은 DashScope Realtime WebSocket 게이트웨이를 사용합니다. 엔드포인트는 wss://로 시작해야 합니다(기본값: wss://dashscope.aliyuncs.com/api-ws/v1/realtime). https:// 주소는 여기서 사용할 수 없습니다.', responseTooLarge: '공급자 응답이 너무 커서 안전을 위해 검증을 중단했습니다.', asrInvalidJson: 'ASR 응답이 유효한 JSON 이 아닙니다.', asrMissingTextField: 'ASR 응답에 text 필드가 없습니다.', diff --git a/openless-all/app/src/i18n/zh-CN.ts b/openless-all/app/src/i18n/zh-CN.ts index 1207c7ac..b0e9ac2f 100644 --- a/openless-all/app/src/i18n/zh-CN.ts +++ b/openless-all/app/src/i18n/zh-CN.ts @@ -776,6 +776,7 @@ export const zhCN = { custom: '自定义', asrVolcengine: '火山引擎 bigasr', asrBailian: '阿里云百炼实时 ASR', + asrBailianQwen3: '阿里云百炼 Qwen3 实时 ASR', asrSiliconflow: '硅基流动 SenseVoice', asrZhipu: '智谱 GLM-ASR', asrGroq: 'Groq Whisper-large-v3', @@ -835,6 +836,8 @@ export const zhCN = { providerHttpStatus: '供应商接口返回 {{status}}，请检查 API Key 权限或 Endpoint。', endpointMustUseHttps: '允许使用 HTTP Endpoint，但请注意：API Key 和音频内容可能在传输中泄漏。', endpointInvalid: 'Endpoint 格式不合法。', + bailianEndpointSchemeInvalid: '百炼实时 ASR 走 DashScope WebSocket 网关，接口地址必须以 wss:// 开头（默认 wss://dashscope.aliyuncs.com/api-ws/v1/inference/）；https:// 的兼容模式地址在此不可用。', + qwen3EndpointSchemeInvalid: 'Qwen3 实时 ASR 走 DashScope Realtime WebSocket 网关，接口地址必须以 wss:// 开头（默认 wss://dashscope.aliyuncs.com/api-ws/v1/realtime）；https:// 地址在此不可用。', responseTooLarge: '供应商响应过大，已停止验证以保证安全。', asrInvalidJson: 'ASR 响应不是有效 JSON。', asrMissingTextField: 'ASR 响应缺少 text 字段。', diff --git a/openless-all/app/src/i18n/zh-TW.ts b/openless-all/app/src/i18n/zh-TW.ts index dc6b6da0..976d7100 100644 --- a/openless-all/app/src/i18n/zh-TW.ts +++ b/openless-all/app/src/i18n/zh-TW.ts @@ -778,6 +778,7 @@ export const zhTW: typeof zhCN = { custom: '自定義', asrVolcengine: '火山引擎 bigasr', asrBailian: '阿里雲百煉即時 ASR', + asrBailianQwen3: '阿里雲百煉 Qwen3 即時 ASR', asrSiliconflow: '硅基流動 SenseVoice', asrZhipu: '智譜 GLM-ASR', asrGroq: 'Groq Whisper-large-v3', @@ -837,6 +838,8 @@ export const zhTW: typeof zhCN = { providerHttpStatus: '供應商接口返回 {{status}}，請檢查 API Key 權限或 Endpoint。', endpointMustUseHttps: '允許使用 HTTP Endpoint，但請注意：API Key 和音訊內容可能在傳輸中外洩。', endpointInvalid: 'Endpoint 格式不合法。', + bailianEndpointSchemeInvalid: '百煉即時 ASR 走 DashScope WebSocket 閘道，接口地址必須以 wss:// 開頭（預設 wss://dashscope.aliyuncs.com/api-ws/v1/inference/）；https:// 的相容模式地址在此不可用。', + qwen3EndpointSchemeInvalid: 'Qwen3 即時 ASR 走 DashScope Realtime WebSocket 閘道，接口地址必須以 wss:// 開頭（預設 wss://dashscope.aliyuncs.com/api-ws/v1/realtime）；https:// 地址在此不可用。', responseTooLarge: '供應商響應過大，已停止驗證以保證安全。', asrInvalidJson: 'ASR 響應不是有效 JSON。', asrMissingTextField: 'ASR 響應缺少 text 字段。', diff --git a/openless-all/app/src/pages/settings/ProvidersSection.tsx b/openless-all/app/src/pages/settings/ProvidersSection.tsx index 3a2cff4e..bd6b5da0 100644 --- a/openless-all/app/src/pages/settings/ProvidersSection.tsx +++ b/openless-all/app/src/pages/settings/ProvidersSection.tsx @@ -155,6 +155,10 @@ const ASR_DEFAULT_RESOURCE_ID = 'volc.seedasr.sauc.duration'; const ASR_PRESETS: ReadonlyArray<{ id: AsrPresetId; nameKey: string; baseUrl: string; model: string }> = [ { id: 'volcengine', nameKey: 'asrVolcengine', baseUrl: '', model: '' }, { 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 处理。 + // 业务空间专属域名（wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime）同样可用。 + { id: 'bailian-qwen3-realtime', nameKey: 'asrBailianQwen3', baseUrl: 'wss://dashscope.aliyuncs.com/api-ws/v1/realtime', model: 'qwen3-asr-flash-realtime' }, { id: 'siliconflow', nameKey: 'asrSiliconflow', baseUrl: 'https://api.siliconflow.cn/v1', model: 'FunAudioLLM/SenseVoiceSmall' }, { id: 'zhipu', nameKey: 'asrZhipu', baseUrl: 'https://open.bigmodel.cn/api/paas/v4', model: 'glm-asr-2512' }, { id: 'groq', nameKey: 'asrGroq', baseUrl: 'https://api.groq.com/openai/v1', model: 'whisper-large-v3-turbo' }, @@ -621,6 +625,8 @@ function providerErrorMessage(error: unknown, t: ReturnType