diff --git a/.env.example b/.env.example index 630617e..d951c39 100644 --- a/.env.example +++ b/.env.example @@ -118,6 +118,17 @@ QWEN3_ASR_MODEL=qwen3-asr-0.6b # ----------------------------------------------------------------------------- # 调优参数。除非你在做特定瓶颈测试，否则建议保持默认。 # ----------------------------------------------------------------------------- +# REALTIME_STREAM_CHUNK_SEC：Qwen 流式解码间隔；0.6 秒参考 FunASR 每 10 个 60ms 音频帧解码一次。 +# Qwen 每次解码会复用累计音频，缩短该值会降低显示延迟，也会增加推理开销。 +# REALTIME_STREAM_CHUNK_SEC=0.6 +# REALTIME_STREAM_MAX_PENDING_CHUNKS：Qwen 流式状态保留的待修订块数。 +# REALTIME_STREAM_MAX_PENDING_CHUNKS=3 +# REALTIME_ENABLE_VAD：启用 FSMN-VAD 流式端点和气泡时间边界。 +# REALTIME_ENABLE_VAD=true +# REALTIME_VAD_CHUNK_MS：每次送入 FSMN-VAD 的音频块长度；参考 FunASR 示例使用 200ms。 +# REALTIME_VAD_CHUNK_MS=200 +# REALTIME_VAD_PRE_ROLL_MS：检测到语音起点时保留的前置音频，避免切掉首字。 +# REALTIME_VAD_PRE_ROLL_MS=600 # ASR_BATCH_SIZE：批处理大小，表示一次并行推理的片段数。 # ASR_BATCH_SIZE=4 # ASR_ENABLE_WORD_TIMESTAMPS：是否启用字词级时间戳。 diff --git a/app/core/config.py b/app/core/config.py index f6feb65..2c8f794 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -107,7 +107,9 @@ class Settings: REALTIME_MIN_PARTIAL_SEC: float = 0.45 REALTIME_PARTIAL_EMIT_INTERVAL_SEC: float = 0.25 REALTIME_PARTIAL_WINDOW_SEC: float = 8.0 - REALTIME_STREAM_CHUNK_SEC: float = 1.2 + # Match FunASR's roughly 600 ms online decode cadence; Qwen re-decodes the + # accumulated stream, so this stays configurable for server-side tuning. + REALTIME_STREAM_CHUNK_SEC: float = 0.6 REALTIME_STREAM_MAX_PENDING_CHUNKS: int = 3 REALTIME_STREAM_WINDOW_SEC: float = 8.0 REALTIME_STREAM_STABLE_TAIL_CHARS: int = 8 @@ -119,6 +121,9 @@ class Settings: REALTIME_LONGFORM_OVERLAP_SEC: float = 1.2 REALTIME_VAD_CHECK_INTERVAL_SEC: float = 0.8 REALTIME_VAD_FINALIZE_SILENCE_SEC: float = 0.6 + REALTIME_ENABLE_VAD: bool = True + REALTIME_VAD_CHUNK_MS: int = 200 + REALTIME_VAD_PRE_ROLL_MS: int = 600 REALTIME_FINAL_SEGMENT_SOFT_LIMIT_SEC: float = 8.0 REALTIME_FINAL_SEGMENT_HARD_LIMIT_SEC: float = 12.0 REALTIME_ENABLE_DIARIZATION: bool = True @@ -383,6 +388,16 @@ class Settings: str(self.REALTIME_VAD_FINALIZE_SILENCE_SEC), ) ) + self.REALTIME_ENABLE_VAD = ( + os.getenv("REALTIME_ENABLE_VAD", str(self.REALTIME_ENABLE_VAD)).lower() + == "true" + ) + self.REALTIME_VAD_CHUNK_MS = int( + os.getenv("REALTIME_VAD_CHUNK_MS", str(self.REALTIME_VAD_CHUNK_MS)) + ) + self.REALTIME_VAD_PRE_ROLL_MS = int( + os.getenv("REALTIME_VAD_PRE_ROLL_MS", str(self.REALTIME_VAD_PRE_ROLL_MS)) + ) self.REALTIME_FINAL_SEGMENT_SOFT_LIMIT_SEC = float( os.getenv( "REALTIME_FINAL_SEGMENT_SOFT_LIMIT_SEC", diff --git a/app/services/asr/qwen3_engine.py b/app/services/asr/qwen3_engine.py index ab7e8a1..fa8b79d 100644 --- a/app/services/asr/qwen3_engine.py +++ b/app/services/asr/qwen3_engine.py @@ -131,8 +131,8 @@ def _handle_asr_error(operation: str): @dataclass class Qwen3StreamingState: internal_state: Any - chunk_size_sec: float = 1.2 - unfixed_chunk_num: int = 2 + chunk_size_sec: float = 0.6 + unfixed_chunk_num: int = 3 unfixed_token_num: int = 5 max_new_tokens: int = 32 language: Optional[str] = None @@ -576,8 +576,8 @@ class Qwen3ASREngine(BaseASREngine): if self._backend == "rust": if context: logger.debug("QwenASR Rust backend ignores streaming context hints") - chunk_size_sec = float(kwargs.get("chunk_size_sec", 1.2)) - unfixed_chunk_num = int(kwargs.get("unfixed_chunk_num", 2)) + chunk_size_sec = float(kwargs.get("chunk_size_sec", 0.6)) + unfixed_chunk_num = int(kwargs.get("unfixed_chunk_num", 3)) unfixed_token_num = int(kwargs.get("unfixed_token_num", 5)) max_new_tokens = int(kwargs.get("max_new_tokens", 32)) stream_handle = self.model.create_stream( @@ -607,8 +607,8 @@ class Qwen3ASREngine(BaseASREngine): ) return Qwen3StreamingState( internal_state=streaming_state, - chunk_size_sec=float(kwargs.get("chunk_size_sec", 1.2)), - unfixed_chunk_num=int(kwargs.get("unfixed_chunk_num", 2)), + chunk_size_sec=float(kwargs.get("chunk_size_sec", 0.6)), + unfixed_chunk_num=int(kwargs.get("unfixed_chunk_num", 3)), unfixed_token_num=int(kwargs.get("unfixed_token_num", 5)), max_new_tokens=int(kwargs.get("max_new_tokens", 32)), language=language, diff --git a/app/services/qwen3_websocket_asr.py b/app/services/qwen3_websocket_asr.py index 3aba177..948e180 100644 --- a/app/services/qwen3_websocket_asr.py +++ b/app/services/qwen3_websocket_asr.py @@ -244,6 +244,15 @@ class ConnectionContext: engine_lease: Optional[RuntimeEngineLease] = None engine: Optional[Qwen3ASREngine] = None pre_roll_audio: np.ndarray = field(default_factory=lambda: np.array([], dtype=np.float32)) + pre_roll_start_sample: int = 0 + audio_samples_received: int = 0 + segment_buffer_start_sample: int = 0 + # FSMN-VAD cache and pending samples are private to this WebSocket stream. + vad_cache: Dict[str, Any] = field(default_factory=dict) + vad_pending_audio: np.ndarray = field(default_factory=lambda: np.array([], dtype=np.float32)) + vad_speech_active: bool = False + vad_speech_start_sample: Optional[int] = None + vad_failed: bool = False # segment_audio_buffer 保留“当前断句以来”的完整音频，用于断句时提取声纹。 segment_audio_buffer: np.ndarray = field(default_factory=lambda: np.array([], dtype=np.float32)) # stream_window_buffer 保留最近窗口，供 realtime partial 在 native stream 失效时兜底重转写。 @@ -266,6 +275,7 @@ class ConnectionContext: last_partial_chunk_id: int = 0 last_partial_raw_text: str = "" last_partial_display_text: str = "" + partial_preview_text: str = "" stable_partial_prefix: str = "" pending_partial_revision_text: str = "" pending_partial_revision_rounds: int = 0 @@ -1070,7 +1080,163 @@ class Qwen3ASRService: return text[-limit:] def _enable_realtime_vad_split(self, ctx: ConnectionContext) -> bool: - return bool(ctx.params.get("enable_realtime_vad_split", False)) + return bool( + ctx.params.get( + "enable_realtime_vad_split", + getattr(settings, "REALTIME_ENABLE_VAD", True), + ) + ) + + def _enable_realtime_vad(self, ctx: ConnectionContext) -> bool: + if str(ctx.params.get("format", "pcm")).lower() == "wav": + # Uploaded WAV is delivered as one complete message, not live frames. + return False + return bool( + ctx.params.get( + "enable_realtime_vad", + getattr(settings, "REALTIME_ENABLE_VAD", True), + ) + ) and not ctx.vad_failed + + async def _warm_realtime_vad(self, ctx: ConnectionContext) -> None: + """Load the shared FSMN-VAD model before the first speech boundary is needed.""" + if not self._enable_realtime_vad(ctx): + return + try: + await run_sync(get_global_vad_model, settings.DEVICE) + except Exception as exc: + # Realtime energy gating remains available if the optional VAD model fails. + ctx.vad_failed = True + logger.warning("Realtime FSMN-VAD unavailable; using energy fallback: %s", exc) + + @staticmethod + def _generate_realtime_vad_chunks( + chunks: List[np.ndarray], + cache: Dict[str, Any], + chunk_size_ms: int, + *, + is_final: bool = False, + ) -> List[List[List[int]]]: + """Run cached FSMN-VAD inference over fixed 200 ms mono audio frames.""" + vad_model = get_global_vad_model(settings.DEVICE) + results: List[List[List[int]]] = [] + with get_vad_inference_lock(): + for index, chunk in enumerate(chunks): + output = vad_model.generate( + input=np.asarray(chunk, dtype=np.float32), + cache=cache, + is_final=is_final and index == len(chunks) - 1, + chunk_size=chunk_size_ms, + disable_pbar=True, + ) + values = output[0].get("value", []) if output else [] + results.append(values or []) + return results + + async def _push_realtime_vad_audio( + self, + ctx: ConnectionContext, + audio: np.ndarray, + ) -> List[tuple[int, int]]: + """Feed each stream's FSMN-VAD cache and return completed absolute boundaries.""" + if audio.size == 0 or not self._enable_realtime_vad(ctx): + return [] + + chunk_size_ms = max( + int(getattr(settings, "REALTIME_VAD_CHUNK_MS", 200) or 200), + 100, + ) + chunk_samples = max(int(chunk_size_ms * 16), 1600) + ctx.vad_pending_audio = np.concatenate([ctx.vad_pending_audio, audio]) + complete_count = int(ctx.vad_pending_audio.size // chunk_samples) + if complete_count <= 0: + return [] + + split_at = complete_count * chunk_samples + chunks = [ + np.asarray(ctx.vad_pending_audio[offset:offset + chunk_samples], dtype=np.float32) + for offset in range(0, split_at, chunk_samples) + ] + ctx.vad_pending_audio = np.asarray(ctx.vad_pending_audio[split_at:], dtype=np.float32) + + try: + outputs = await run_sync( + self._generate_realtime_vad_chunks, + chunks, + ctx.vad_cache, + chunk_size_ms, + ) + except Exception as exc: + ctx.vad_failed = True + ctx.vad_speech_active = False + logger.warning("Realtime FSMN-VAD inference failed; using energy fallback: %s", exc) + return [] + + completed: List[tuple[int, int]] = [] + for values in outputs: + for value in values: + if not isinstance(value, (list, tuple)) or len(value) < 2: + continue + start_ms, end_ms = int(value[0]), int(value[1]) + if start_ms >= 0: + ctx.vad_speech_start_sample = start_ms * 16 + ctx.vad_speech_active = True + if end_ms >= 0: + start_sample = ( + start_ms * 16 + if start_ms >= 0 + else ctx.vad_speech_start_sample + ) + if start_sample is not None and end_ms * 16 > start_sample: + completed.append((int(start_sample), end_ms * 16)) + ctx.vad_speech_active = False + ctx.vad_speech_start_sample = None + return completed + + async def _flush_realtime_vad(self, ctx: ConnectionContext) -> List[tuple[int, int]]: + """Flush the final short VAD frame so stop can use its last boundary.""" + if not self._enable_realtime_vad(ctx) or ctx.vad_pending_audio.size == 0: + return [] + + chunk_size_ms = max( + int(getattr(settings, "REALTIME_VAD_CHUNK_MS", 200) or 200), + 100, + ) + final_chunk = np.asarray(ctx.vad_pending_audio, dtype=np.float32) + ctx.vad_pending_audio = np.array([], dtype=np.float32) + try: + outputs = await run_sync( + self._generate_realtime_vad_chunks, + [final_chunk], + ctx.vad_cache, + chunk_size_ms, + is_final=True, + ) + except Exception as exc: + ctx.vad_failed = True + logger.warning("Realtime FSMN-VAD flush failed; using energy fallback: %s", exc) + return [] + + completed: List[tuple[int, int]] = [] + for values in outputs: + for value in values: + if not isinstance(value, (list, tuple)) or len(value) < 2: + continue + start_ms, end_ms = int(value[0]), int(value[1]) + if start_ms >= 0: + ctx.vad_speech_start_sample = start_ms * 16 + ctx.vad_speech_active = True + if end_ms >= 0: + start_sample = ( + start_ms * 16 + if start_ms >= 0 + else ctx.vad_speech_start_sample + ) + if start_sample is not None and end_ms * 16 > start_sample: + completed.append((int(start_sample), end_ms * 16)) + ctx.vad_speech_active = False + ctx.vad_speech_start_sample = None + return completed def _enable_realtime_longform(self, ctx: ConnectionContext) -> bool: return bool(ctx.params.get("enable_realtime_longform", False)) @@ -1285,25 +1451,47 @@ class Qwen3ASRService: ) return fallback, "" - def _append_pre_roll(self, ctx: ConnectionContext, audio: np.ndarray) -> None: + def _append_pre_roll( + self, + ctx: ConnectionContext, + audio: np.ndarray, + *, + start_sample: Optional[int] = None, + ) -> None: if audio.size == 0: return + if ctx.pre_roll_audio.size == 0 and start_sample is not None: + ctx.pre_roll_start_sample = int(start_sample) ctx.pre_roll_audio = np.concatenate([ctx.pre_roll_audio, audio]) max_samples = self._pre_roll_samples(ctx) if max_samples <= 0: ctx.pre_roll_audio = np.array([], dtype=np.float32) + if start_sample is not None: + ctx.pre_roll_start_sample = int(start_sample) + int(audio.size) elif ctx.pre_roll_audio.size > max_samples: - ctx.pre_roll_audio = ctx.pre_roll_audio[-max_samples:] + removed = int(ctx.pre_roll_audio.size - max_samples) + ctx.pre_roll_audio = ctx.pre_roll_audio[removed:] + ctx.pre_roll_start_sample += removed - def _start_turn(self, ctx: ConnectionContext, audio: np.ndarray) -> None: + def _start_turn( + self, + ctx: ConnectionContext, + audio: np.ndarray, + *, + start_sample: Optional[int] = None, + ) -> None: parts: List[np.ndarray] = [] + turn_start_sample = int(start_sample or 0) if ctx.pre_roll_audio.size > 0: parts.append(np.asarray(ctx.pre_roll_audio, dtype=np.float32)) + turn_start_sample = int(ctx.pre_roll_start_sample) parts.append(np.asarray(audio, dtype=np.float32)) ctx.segment_audio_buffer = ( np.concatenate(parts) if len(parts) > 1 else np.asarray(parts[0], dtype=np.float32) ) + ctx.segment_buffer_start_sample = turn_start_sample ctx.pre_roll_audio = np.array([], dtype=np.float32) + ctx.pre_roll_start_sample = int(turn_start_sample + ctx.segment_audio_buffer.size) window_samples = self._stream_window_samples() ctx.stream_window_buffer = np.asarray( ctx.segment_audio_buffer[-window_samples:], @@ -1350,6 +1538,7 @@ class Qwen3ASRService: ctx.last_partial_chunk_id = 0 ctx.last_partial_raw_text = "" ctx.last_partial_display_text = "" + ctx.partial_preview_text = "" ctx.stable_partial_prefix = "" ctx.pending_partial_revision_text = "" ctx.pending_partial_revision_rounds = 0 @@ -1828,6 +2017,20 @@ class Qwen3ASRService: stable_prefix = stripped[:-holdback_chars].rstrip() return stable_prefix or stripped + def _append_only_partial_preview( + self, + ctx: ConnectionContext, + text: str, + ) -> str: + """Expose only newly stabilized prefixes; the final event carries revisions.""" + candidate = self._partial_display_text(ctx, text) + previous = ctx.partial_preview_text + if not previous or candidate.startswith(previous): + ctx.partial_preview_text = candidate + # Qwen may rewrite its rolling snapshot. Keep the already shown prefix + # fixed until finalization so the browser never retracts earlier words. + return ctx.partial_preview_text + def _is_suspicious_segment_text( self, text: str, @@ -1924,25 +2127,22 @@ class Qwen3ASRService: if audio.size == 0: return [] - temp_path: Optional[str] = None try: - temp_path = get_speaker_registry_service().save_audio_array_to_temp( - audio, - sample_rate=16000, - ) - def _run_vad() -> list[list[int]]: vad_model = get_global_vad_model(settings.DEVICE) with get_vad_inference_lock(): - result = vad_model.generate(input=temp_path, cache={}) + # FunASR's FSMN-VAD accepts mono float32 NumPy audio directly. + result = vad_model.generate( + input=np.asarray(audio, dtype=np.float32), + cache={}, + disable_pbar=True, + ) return result[0].get("value", []) if result else [] return await run_sync(_run_vad) except Exception as exc: logger.debug("Realtime voiced-duration VAD failed: %s", exc) return None - finally: - get_speaker_registry_service().cleanup_file(temp_path) async def _split_silence_audio_by_vad( self, @@ -2389,6 +2589,8 @@ class Qwen3ASRService: *, finalized_audio_override: Optional[np.ndarray] = None, carry_audio_override: Optional[np.ndarray] = None, + segment_start_ms_override: Optional[int] = None, + segment_end_ms_override: Optional[int] = None, emit_segment_start: bool = True, ) -> None: engine = await self._ensure_engine(ctx) @@ -2500,8 +2702,18 @@ class Qwen3ASRService: is_valid = self._is_valid_committed_segment_text(segment_text, duration_sec) segment_duration_ms = int(finalized_audio.size / 16) - segment_start_ms = int(ctx.timeline_cursor_ms) - segment_end_ms = int(segment_start_ms + segment_duration_ms) + segment_start_ms = int( + ctx.timeline_cursor_ms + if segment_start_ms_override is None + else segment_start_ms_override + ) + segment_end_ms = int( + segment_start_ms + segment_duration_ms + if segment_end_ms_override is None + else segment_end_ms_override + ) + if segment_end_ms < segment_start_ms: + segment_end_ms = segment_start_ms + segment_duration_ms if is_valid: segment_payload = { @@ -2560,6 +2772,13 @@ class Qwen3ASRService: ctx.segment_index += 1 ctx.timeline_cursor_ms = segment_end_ms + if carry_audio.size > 0: + ctx.segment_buffer_start_sample += max( + int(current_audio.size - carry_audio.size), + 0, + ) + else: + ctx.segment_buffer_start_sample = int(ctx.audio_samples_received) ctx.segment_audio_buffer = np.asarray(carry_audio, dtype=np.float32) ctx.stream_window_buffer = np.asarray(carry_audio, dtype=np.float32) ctx.sentence_active = bool(carry_audio.size > 0) @@ -2574,6 +2793,52 @@ class Qwen3ASRService: np.asarray(carry_audio, dtype=np.float32), ) + async def _commit_vad_boundary( + self, + websocket: WebSocket, + ctx: ConnectionContext, + task_id: str, + boundary: tuple[int, int], + *, + reason: str = "vad_end", + emit_segment_start: bool = True, + ) -> bool: + """Trim one turn to FSMN-VAD's absolute start/end samples before commit.""" + current_audio = np.asarray(ctx.segment_audio_buffer, dtype=np.float32) + turn_start_sample = int(ctx.segment_buffer_start_sample) + start_sample = max(int(boundary[0]), turn_start_sample) + stream_end_sample = turn_start_sample + int(current_audio.size) + end_sample = min(int(boundary[1]), stream_end_sample) + if end_sample <= start_sample: + return False + + start_offset = start_sample - turn_start_sample + end_offset = end_sample - turn_start_sample + finalized_audio = np.asarray(current_audio[start_offset:end_offset], dtype=np.float32) + trailing_audio = np.asarray(current_audio[end_offset:], dtype=np.float32) + if finalized_audio.size == 0: + return False + + await self._commit_retranscribe_turn( + websocket, + ctx, + task_id, + reason, + finalized_audio_override=finalized_audio, + segment_start_ms_override=int(start_sample / 16), + segment_end_ms_override=int(end_sample / 16), + emit_segment_start=emit_segment_start, + ) + # Preserve post-end audio as pre-roll for a possible next utterance in + # the same network frame, while the bubble duration follows VAD bounds. + self._append_pre_roll( + ctx, + trailing_audio, + start_sample=end_sample, + ) + ctx.vad_speech_start_sample = None + return True + async def _send_error( self, websocket: WebSocket, @@ -2678,7 +2943,10 @@ class Qwen3ASRService: "partial_window_sec", settings.REALTIME_PARTIAL_WINDOW_SEC, ), - "pre_roll_ms": payload.get("pre_roll_ms", 240), + "pre_roll_ms": payload.get( + "pre_roll_ms", + settings.REALTIME_VAD_PRE_ROLL_MS, + ), "max_sentence_count": payload.get("max_sentence_count", 8), "max_partial_text_chars": payload.get( "max_partial_text_chars", @@ -2702,7 +2970,14 @@ class Qwen3ASRService: settings.REALTIME_ENABLE_SEGMENT_REFINE, ), "enable_realtime_longform": payload.get("enable_realtime_longform", False), - "enable_realtime_vad_split": payload.get("enable_realtime_vad_split", False), + "enable_realtime_vad": payload.get( + "enable_realtime_vad", + settings.REALTIME_ENABLE_VAD, + ), + "enable_realtime_vad_split": payload.get( + "enable_realtime_vad_split", + settings.REALTIME_ENABLE_VAD, + ), "force_stable_segment_sec": payload.get( "force_stable_segment_sec", settings.REALTIME_FORCE_STABLE_SEGMENT_SEC, @@ -2724,7 +2999,16 @@ class Qwen3ASRService: ), } + ctx.vad_cache = {} + ctx.vad_pending_audio = np.array([], dtype=np.float32) + ctx.vad_speech_active = False + ctx.vad_speech_start_sample = None + ctx.vad_failed = False + ctx.audio_samples_received = 0 + ctx.segment_buffer_start_sample = 0 + await self._ensure_engine(ctx) + await self._warm_realtime_vad(ctx) ctx.stream_window_buffer = np.array([], dtype=np.float32) self._reset_partial_state(ctx) @@ -2775,13 +3059,56 @@ class Qwen3ASRService: if audio is None: continue - has_voice = self._has_voice(audio) + audio_start_sample = int(ctx.audio_samples_received) + ctx.audio_samples_received += int(audio.size) + # The bundled demo sends a whole WAV in one frame; keep it + # intact for final ASR instead of treating it as microphone pre-roll. + bulk_audio_frame = ( + str(ctx.params.get("format", "pcm")).lower() == "wav" + or audio.size > int(1.5 * 16000) + ) + vad_boundaries = ( + [] + if str(ctx.params.get("format", "pcm")).lower() == "wav" + else await self._push_realtime_vad_audio(ctx, audio) + ) + energy_voice = self._has_voice(audio) + has_voice = ( + bulk_audio_frame + or energy_voice + or ctx.vad_speech_active + or bool(vad_boundaries) + ) if not ctx.sentence_active: if not has_voice: - self._append_pre_roll(ctx, audio) + self._append_pre_roll( + ctx, + audio, + start_sample=audio_start_sample, + ) continue - self._start_turn(ctx, audio) + if bulk_audio_frame: + self._start_turn( + ctx, + audio, + start_sample=audio_start_sample, + ) + else: + self._append_pre_roll( + ctx, + audio, + start_sample=audio_start_sample, + ) + self._start_turn( + ctx, + ( + np.array([], dtype=np.float32) + if ctx.pre_roll_audio.size > 0 + else audio + ), + start_sample=audio_start_sample, + ) engine = await self._ensure_engine(ctx) await self._push_realtime_stream_audio( ctx, @@ -2789,7 +3116,7 @@ class Qwen3ASRService: np.asarray(ctx.segment_audio_buffer, dtype=np.float32), ) else: - self._append_turn_audio(ctx, audio, has_voice=has_voice) + self._append_turn_audio(ctx, audio, has_voice=energy_voice) engine = await self._ensure_engine(ctx) await self._push_realtime_stream_audio(ctx, engine, audio) @@ -2823,7 +3150,11 @@ class Qwen3ASRService: ctx, visible_observed, ) - partial_display = self._clip_partial_text(visible_observed, ctx) + partial_display = self._append_only_partial_preview( + ctx, + visible_observed, + ) + partial_display = self._clip_partial_text(partial_display, ctx) if ( partial_display and ( @@ -2842,11 +3173,26 @@ class Qwen3ASRService: replace_snapshot=has_native_stream_snapshot, ) + if bulk_audio_frame: + partial_start_sample = ctx.segment_buffer_start_sample + partial_end_sample = ctx.audio_samples_received + else: + partial_start_sample = ctx.vad_speech_start_sample + if partial_start_sample is None and vad_boundaries: + partial_start_sample = vad_boundaries[0][0] + if partial_start_sample is None: + partial_start_sample = ctx.segment_buffer_start_sample + partial_end_sample = ( + vad_boundaries[0][1] + if vad_boundaries + else ctx.audio_samples_received + ) + sentence_payload = self._build_tencent_sentence( { "index": ctx.segment_index, - "start_ms": int(ctx.timeline_cursor_ms), - "end_ms": int(ctx.timeline_cursor_ms + (ctx.segment_audio_buffer.size / 16)), + "start_ms": int(partial_start_sample / 16), + "end_ms": int(partial_end_sample / 16), "text": partial_display, "speaker_id": -1, "sentence_type": 0, @@ -2890,13 +3236,63 @@ class Qwen3ASRService: ) continue + if vad_boundaries and ctx.sentence_active and not bulk_audio_frame: + if await self._commit_vad_boundary( + websocket, + ctx, + task_id, + vad_boundaries[0], + ): + ctx.state = ConnectionState.STREAMING + continue + silence_threshold = self._get_dynamic_silence_threshold_samples(ctx) + if self._enable_realtime_vad(ctx) and ctx.vad_speech_active: + # Give FSMN-VAD time to confirm an endpoint before the + # legacy energy-based safety fallback closes the turn. + vad_fallback_sec = max( + float(settings.REALTIME_VAD_FINALIZE_SILENCE_SEC) + 0.8, + 1.2, + ) + silence_threshold = max( + silence_threshold, + int(vad_fallback_sec * 16000), + ) if ctx.silence_samples >= silence_threshold: + if self._enable_realtime_vad(ctx): + fallback_segments = await self._run_vad_segments( + np.asarray(ctx.segment_audio_buffer, dtype=np.float32) + ) + valid_segments = [ + (int(item[0]), int(item[1])) + for item in (fallback_segments or []) + if len(item) >= 2 and int(item[1]) > int(item[0]) + ] + if valid_segments: + fallback_boundary = ( + int(ctx.segment_buffer_start_sample + valid_segments[0][0] * 16), + int(ctx.segment_buffer_start_sample + valid_segments[-1][1] * 16), + ) + if await self._commit_vad_boundary( + websocket, + ctx, + task_id, + fallback_boundary, + reason="vad_fallback", + ): + ctx.state = ConnectionState.STREAMING + continue + await self._commit_retranscribe_turn( websocket, ctx, task_id, "silence", + finalized_audio_override=np.asarray( + ctx.segment_audio_buffer, + dtype=np.float32, + ), + carry_audio_override=np.array([], dtype=np.float32), ) ctx.state = ConnectionState.STREAMING continue @@ -2959,14 +3355,41 @@ class Qwen3ASRService: task_id: str, ) -> None: try: + final_vad_boundaries = await self._flush_realtime_vad(ctx) if ctx.sentence_active and ctx.segment_audio_buffer.size > 0: - await self._commit_retranscribe_turn( + final_boundary = ( + final_vad_boundaries[-1] + if final_vad_boundaries + else None + ) + if ( + final_boundary is None + and self._enable_realtime_vad(ctx) + and ctx.vad_speech_active + and ctx.vad_speech_start_sample is not None + ): + final_boundary = ( + int(ctx.vad_speech_start_sample), + int(ctx.audio_samples_received), + ) + + if final_boundary is not None and await self._commit_vad_boundary( websocket, ctx, task_id, - "final", + final_boundary, + reason="final", emit_segment_start=False, - ) + ): + pass + else: + await self._commit_retranscribe_turn( + websocket, + ctx, + task_id, + "final", + emit_segment_start=False, + ) # Drain pending diarization before the final event; the browser closes # the socket on `end`, so later speaker updates would otherwise be lost.