diff --git a/funasr/bin/realtime_ws.py b/funasr/bin/realtime_ws.py index a73b86a7b..8657886f0 100644 --- a/funasr/bin/realtime_ws.py +++ b/funasr/bin/realtime_ws.py @@ -33,36 +33,110 @@ logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(me logger = logging.getLogger(__name__) +def _normalize_transcript_with_positions(text): + normalized = [] + source_positions = [] + for index, char in enumerate(text): + if regex.fullmatch(r"[\p{P}\p{S}\s]", char): + continue + folded = char.casefold() + normalized.extend(folded) + source_positions.extend([index] * len(folded)) + return "".join(normalized), source_positions + + +def _normalize_transcript(text): + return _normalize_transcript_with_positions(text)[0] + + +def _merge_overlapping_transcripts(existing, incoming): + """Join adjacent rolling-window transcripts when their text overlap is clear.""" + existing_norm, existing_positions = _normalize_transcript_with_positions(existing) + incoming_norm, incoming_positions = _normalize_transcript_with_positions(incoming) + if not existing_norm or not incoming_norm: + return None + + minimum_overlap = max(4, min(len(existing_norm), len(incoming_norm)) // 4) + for overlap in range(min(len(existing_norm), len(incoming_norm)), minimum_overlap - 1, -1): + if existing_norm[-overlap:] != incoming_norm[:overlap]: + continue + existing_start = len(existing_norm) - overlap + if ( + existing_start > 0 + and existing_positions[existing_start - 1] == existing_positions[existing_start] + ): + continue + if ( + overlap < len(incoming_positions) + and incoming_positions[overlap - 1] == incoming_positions[overlap] + ): + continue + suffix_start = incoming_positions[overlap - 1] + 1 + while ( + suffix_start < len(incoming) + and regex.fullmatch(r"[\p{P}\p{S}]", incoming[suffix_start]) + and existing.endswith(incoming[suffix_start]) + ): + suffix_start += 1 + return existing + incoming[suffix_start:] + return None + + def detect_and_fix_hallucination(text, max_ngram_length=12, max_occurrences=3): """Detect repeated patterns (hallucination) and truncate to keep one occurrence.""" if not text or len(text) < max_ngram_length * 2: return text, False - cleaned = regex.sub(r'\p{P}+', '', text) + source_positions = [] + cleaned_chars = [] + for index, char in enumerate(text): + if regex.fullmatch(r'\p{P}', char): + continue + cleaned_chars.append(char) + source_positions.append(index) + cleaned = ''.join(cleaned_chars) + + def truncate_after_first(match): + def ascii_quote_count(quote, end): + count = 0 + for index in range(end): + if text[index] != quote: + continue + if ( + quote == "'" + and index > 0 + and index + 1 < len(text) + and text[index - 1].isalnum() + and text[index + 1].isalnum() + ): + continue + count += 1 + return count + + cleaned_end = match.start(1) + len(match.group(1)) + source_end = source_positions[cleaned_end - 1] + 1 + while source_end < len(text) and regex.fullmatch( + r'[\p{Pe}\p{Pf}\p{Po}\p{Pd}\p{Pc}]', text[source_end] + ): + if ( + text[source_end] in {'"', "'"} + and ascii_quote_count(text[source_end], source_end) % 2 == 0 + ): + break + source_end += 1 + return text[:source_end].rstrip(), True word_pattern = rf'(?= 0: - end_pos = text.find(repeated, pos + len(repeated)) - if end_pos >= 0: - return text[:end_pos + len(repeated)], True - return text[:len(text)//2], True + match = regex.search(word_pattern, cleaned, regex.IGNORECASE) + if match: + return truncate_after_first(match) for length in range(1, max_ngram_length): pattern = rf'(?= 0: - end_pos = text.find(repeated, pos + len(repeated)) - if end_pos >= 0: - return text[:end_pos + len(repeated)], True - return text[:len(text)//2], True + return truncate_after_first(match) return text, False @@ -401,6 +475,12 @@ class RealtimeASRSession: self.prev_text = "" self.last_partial_text = "" self.last_partial_start_ms = 0 + self.last_partial_end_ms = 0 + self.last_partial_eligible = False + self.segment_partial_text = "" + self.segment_partial_start_ms = 0 + self.segment_partial_end_ms = 0 + self.segment_partial_stable_count = 0 self.last_decode_samples = 0 self.locked_sentences = [] self.prev_seg_text = "" @@ -428,15 +508,9 @@ class RealtimeASRSession: for seg in new_confirmed: seg_text = self._decode_segment(seg) self.prev_text = "" - if not seg_text.strip(): - continue - self.locked_sentences.append({"text": seg_text, "start": int(seg[0]), "end": int(seg[1])}) - if self.spk_tracker: - s0 = int(seg[0] * self.sample_rate / 1000) - s1 = min(int(seg[1] * self.sample_rate / 1000), self.total_samples) - segment_audio = self._slice_audio(s0, s1).copy() - self.spk_tracker.assign_streaming(segment_audio, seg[0]/1000, seg[1]/1000, self.locked_sentences[-1]) - logger.info(f"Locked: [{seg[0]}-{seg[1]}ms] \"{seg_text[:40]}\"") + self._append_completed_segment(seg, seg_text) + if seg_text.strip(): + logger.info(f"Locked: [{seg[0]}-{seg[1]}ms] \"{seg_text[:40]}\"") self._compact_audio_buffer() @@ -464,6 +538,67 @@ class RealtimeASRSession: self.audio_buffer = np.array([], dtype=np.float32) self.audio_buffer_start_sample = self.total_samples + def _reset_partial_history(self): + self.segment_partial_text = "" + self.segment_partial_start_ms = 0 + self.segment_partial_end_ms = 0 + self.segment_partial_stable_count = 0 + + def _record_partial_text(self, text, start_ms, hallucinated=False): + """Build a confidence-bearing transcript across overlapping partial windows.""" + end_ms = int(self.total_samples * 1000 / self.sample_rate) + self.last_partial_end_ms = end_ms + self.last_partial_eligible = bool(text.strip()) and not hallucinated + if not text.strip() or hallucinated: + self._reset_partial_history() + return + + start_ms = int(start_ms) + incoming_norm = _normalize_transcript(text) + if not self.segment_partial_text: + self.segment_partial_text = text + self.segment_partial_start_ms = start_ms + self.segment_partial_end_ms = end_ms + self.segment_partial_stable_count = 1 + return + + history_norm = _normalize_transcript(self.segment_partial_text) + if start_ms == self.segment_partial_start_ms: + if incoming_norm == history_norm: + self.segment_partial_text = text + self.segment_partial_end_ms = end_ms + self.segment_partial_stable_count += 1 + else: + self.segment_partial_text = text + self.segment_partial_end_ms = end_ms + self.segment_partial_stable_count = 1 + return + + if start_ms < self.segment_partial_start_ms: + self.segment_partial_text = text + self.segment_partial_start_ms = start_ms + self.segment_partial_end_ms = end_ms + self.segment_partial_stable_count = 1 + return + + if start_ms > self.segment_partial_end_ms: + self.segment_partial_text = text + self.segment_partial_start_ms = start_ms + self.segment_partial_end_ms = end_ms + self.segment_partial_stable_count = 1 + return + + merged = _merge_overlapping_transcripts(self.segment_partial_text, text) + if merged is not None: + self.segment_partial_text = merged + self.segment_partial_end_ms = end_ms + self.segment_partial_stable_count = 1 + else: + self.segment_partial_text = text + self.segment_partial_start_ms = start_ms + self.segment_partial_end_ms = end_ms + self.segment_partial_stable_count = 1 + def should_decode(self): threshold = self.first_chunk_samples if not self.first_decode_done else self.chunk_samples return (self.total_samples - self.last_decode_samples) >= threshold @@ -473,10 +608,11 @@ class RealtimeASRSession: if is_final and self.endpoint_mode == "client": return self.commit() - if self.endpoint_mode == "server" and self.total_samples < self.chunk_samples: + if self.endpoint_mode == "server": if is_final: - self._release_audio_buffer() - return self._build_response(is_final) + return self._finalize_server_session() + if self.total_samples < self.first_chunk_samples: + return self._build_response(is_final) if self.endpoint_mode == "client": if self.vad.current_speech_start is None: @@ -487,25 +623,13 @@ class RealtimeASRSession: if self.total_samples - utterance_start_sample < self.first_chunk_samples: return self._build_response(is_final=False) - if is_final: - if self.vad.current_speech_start is not None: - end_ms = int(self.total_samples * 1000 / self.sample_rate) - seg = [self.vad.current_speech_start, end_ms] - seg_text = self._decode_segment(seg) - if seg_text.strip(): - self.locked_sentences.append({"text": seg_text, "start": int(seg[0]), "end": int(seg[1])}) - self.vad.current_speech_start = None - - if self.spk_tracker and self.locked_sentences: - self.locked_sentences = self.spk_tracker.finalize(self.locked_sentences) - self._release_audio_buffer() - return self._build_response(is_final) - if self.vad.current_speech_start is not None: seg_audio, partial_start_ms = self.get_partial_decode_audio() else: self.last_decode_samples = self.total_samples self.last_partial_text = "" + self.last_partial_eligible = False + self._reset_partial_history() return self._build_response(is_final) if len(seg_audio) < self.chunk_samples // 2: @@ -523,6 +647,8 @@ class RealtimeASRSession: text = _clean_asr_text(text) except Exception as e: logger.error(f"ASR error: {e}") + self.last_partial_eligible = False + self._reset_partial_history() return self._build_response(is_final) text, hallucinated = detect_and_fix_hallucination(text) @@ -532,6 +658,7 @@ class RealtimeASRSession: self.last_decode_samples = self.total_samples self.last_partial_text = text self.last_partial_start_ms = partial_start_ms + self._record_partial_text(text, partial_start_ms, hallucinated=hallucinated) if text.strip() and not self.first_decode_done: self.first_decode_done = True @@ -554,31 +681,46 @@ class RealtimeASRSession: end_ms = int(self.total_samples * 1000 / self.sample_rate) seg = [self.vad.current_speech_start, end_ms] seg_text = self._decode_segment(seg) - if seg_text.strip(): - sentence = { - "text": seg_text, - "start": int(seg[0]), - "end": int(seg[1]), - } - if self.spk_tracker: - s0 = int(seg[0] * self.sample_rate / 1000) - segment_audio = self._slice_audio(s0, self.total_samples).copy() - self.spk_tracker.assign_streaming( - segment_audio, - seg[0] / 1000, - seg[1] / 1000, - sentence, - ) - self.locked_sentences.append(sentence) + self._append_completed_segment(seg, seg_text) self.vad.current_speech_start = None - if self.spk_tracker and self.locked_sentences: - self.locked_sentences = self.spk_tracker.finalize(self.locked_sentences) + self._finalize_speakers() response = self._build_response(is_final=True) self._reset_utterance_state() return response + def _finalize_server_session(self): + duration_ms = int(self.total_samples * 1000 / self.sample_rate) + final_segments = [] + finalize_vad = getattr(self.vad, "finalize", None) + if callable(finalize_vad): + try: + final_segments = finalize_vad() or [] + except Exception as error: + logger.error(f"VAD finalization error: {error}") + + had_valid_final_segment = False + for raw_seg in final_segments: + start_ms = max(0, int(raw_seg[0])) + end_ms = min(duration_ms, int(raw_seg[1])) + if end_ms <= start_ms: + continue + had_valid_final_segment = True + seg = [start_ms, end_ms] + seg_text = self._decode_segment(seg) + self._append_completed_segment(seg, seg_text) + + if not had_valid_final_segment and self.vad.current_speech_start is not None: + seg = [int(self.vad.current_speech_start), duration_ms] + seg_text = self._decode_segment(seg) + self._append_completed_segment(seg, seg_text) + + self.vad.current_speech_start = None + self._finalize_speakers() + self._release_audio_buffer() + return self._build_response(is_final=True) + def _reset_utterance_state(self): self._release_audio_buffer() self.first_decode_done = False @@ -586,6 +728,9 @@ class RealtimeASRSession: self.prev_text = "" self.last_partial_text = "" self.last_partial_start_ms = 0 + self.last_partial_end_ms = 0 + self.last_partial_eligible = False + self._reset_partial_history() self.last_decode_samples = self.total_samples self.locked_sentences = [] self.prev_seg_text = "" @@ -604,30 +749,140 @@ class RealtimeASRSession: start_ms = int(decode_start_sample * 1000 / self.sample_rate) return self._slice_audio(decode_start_sample, self.total_samples), start_ms + def _partial_fallback_candidate(self, seg, require_stable, latest_only=False): + if latest_only: + if not self.last_partial_eligible: + return "" + text = self.last_partial_text.strip() + start_ms = self.last_partial_start_ms + end_ms = self.last_partial_end_ms + else: + if require_stable and self.segment_partial_stable_count < 2: + return "" + text = self.segment_partial_text + start_ms = self.segment_partial_start_ms + end_ms = self.segment_partial_end_ms + + end_tolerance_ms = 100 + if ( + not text + or int(start_ms) != int(seg[0]) + or abs(int(end_ms) - int(seg[1])) > end_tolerance_ms + ): + return "" + return text + + def _reconcile_completed_segment_text(self, decoded_text, seg, decode_succeeded): + """Keep a proven segment partial only when the completed decode regresses.""" + final_text, final_hallucinated = detect_and_fix_hallucination(decoded_text) + if not decode_succeeded: + partial_text = self._partial_fallback_candidate( + seg, require_stable=False, latest_only=True + ) + partial_text, _ = detect_and_fix_hallucination(partial_text) + if partial_text: + logger.warning( + "Completed segment [%d-%dms] decode failed; keeping the latest " + "full-coverage partial (%d chars)", + int(seg[0]), int(seg[1]), len(partial_text), + ) + return partial_text + return final_text + + if not final_text.strip(): + return final_text + + partial_text = self._partial_fallback_candidate( + seg, require_stable=not final_hallucinated + ) + if not partial_text: + return final_text + partial_text, partial_hallucinated = detect_and_fix_hallucination(partial_text) + final_cmp = _normalize_transcript(final_text) + partial_cmp = _normalize_transcript(partial_text) + segment_duration_ms = max(0, int(seg[1]) - int(seg[0])) + truncated_prefix = ( + segment_duration_ms >= 2000 + and len(final_cmp) >= 4 + and partial_cmp.startswith(final_cmp) + and len(partial_cmp) >= len(final_cmp) * 2 + and len(partial_cmp) - len(final_cmp) >= 8 + ) + use_partial = ( + (final_hallucinated and not partial_hallucinated) + or truncated_prefix + ) + if use_partial and partial_cmp: + reason = "hallucinated" if final_hallucinated else "truncated" + logger.warning( + "Completed segment [%d-%dms] decode was %s; " + "keeping the latest aligned partial (%d -> %d chars)", + int(seg[0]), int(seg[1]), reason, len(final_text), len(partial_text), + ) + return partial_text + return final_text + + def _clear_completed_partial(self, seg): + self.last_partial_text = "" + self.last_partial_start_ms = int(seg[1]) + self.last_partial_end_ms = int(seg[1]) + self.last_partial_eligible = False + self._reset_partial_history() + + def _append_completed_segment(self, seg, text): + if not text.strip(): + return + sentence = { + "text": text, + "start": int(seg[0]), + "end": int(seg[1]), + } + if self.spk_tracker: + s0 = int(seg[0] * self.sample_rate / 1000) + s1 = min(int(seg[1] * self.sample_rate / 1000), self.total_samples) + segment_audio = self._slice_audio(s0, s1).copy() + self.spk_tracker.assign_streaming( + segment_audio, + seg[0] / 1000, + seg[1] / 1000, + sentence, + ) + self.locked_sentences.append(sentence) + + def _finalize_speakers(self): + if self.spk_tracker and self.locked_sentences: + self.locked_sentences = self.spk_tracker.finalize(self.locked_sentences) + @torch.no_grad() def _decode_segment(self, seg): """Decode a completed VAD segment via vLLM.""" start_sample = int(seg[0] * self.sample_rate / 1000) end_sample = min(int(seg[1] * self.sample_rate / 1000), self.total_samples) seg_audio = self._slice_audio(start_sample, end_sample) - if len(seg_audio) < 1600: - return "" - audio_tensor = torch.from_numpy(seg_audio).float() - try: - results = self.vllm_engine.generate( - inputs=[audio_tensor], - hotwords=self.asr_kwargs.get("hotwords"), - language=self.asr_kwargs.get("language"), - max_new_tokens=512, - ) - text = results[0]["text"] if results else "" - text = _clean_asr_text(text) - text = _postprocess_result_text(text, self.asr_kwargs) - self.prev_seg_text = text - return text - except Exception as e: - logger.error(f"Segment decode error: {e}") - return "" + text = "" + decode_succeeded = False + if len(seg_audio) >= 1600: + audio_tensor = torch.from_numpy(seg_audio).float() + try: + results = self.vllm_engine.generate( + inputs=[audio_tensor], + hotwords=self.asr_kwargs.get("hotwords"), + language=self.asr_kwargs.get("language"), + max_new_tokens=512, + ) + text = results[0]["text"] if results else "" + text = _clean_asr_text(text) + decode_succeeded = True + except Exception as e: + logger.error(f"Segment decode error: {e}") + + text = self._reconcile_completed_segment_text( + text, seg, decode_succeeded=decode_succeeded + ) + text = _postprocess_result_text(text, self.asr_kwargs) + self.prev_seg_text = text + self._clear_completed_partial(seg) + return text def _build_response(self, is_final): duration_ms = int(self.total_samples * 1000 / self.sample_rate) @@ -672,6 +927,9 @@ class RealtimeASRSession: self.prev_text = "" self.last_partial_text = "" self.last_partial_start_ms = 0 + self.last_partial_end_ms = 0 + self.last_partial_eligible = False + self._reset_partial_history() self.last_decode_samples = 0 self.locked_sentences = [] if self.spk_tracker: diff --git a/tests/test_realtime_ws_service.py b/tests/test_realtime_ws_service.py index c15ae7160..3922427a4 100644 --- a/tests/test_realtime_ws_service.py +++ b/tests/test_realtime_ws_service.py @@ -258,6 +258,818 @@ class DummyEngine: return [{"text": "hello"}] +class FixedTextEngine(DummyEngine): + def __init__(self, text): + super().__init__() + self.text = text + + def generate(self, inputs, **kwargs): + self.input_lengths.extend(len(audio) for audio in inputs) + self.generate_kwargs.append(kwargs) + return [{"text": self.text}] + + +class FailingTextEngine(DummyEngine): + def generate(self, inputs, **kwargs): + raise RuntimeError("synthetic final decode failure") + + +class ScriptedTextEngine(DummyEngine): + def __init__(self, outputs): + super().__init__() + self.outputs = list(outputs) + + def generate(self, inputs, **kwargs): + self.input_lengths.extend(len(audio) for audio in inputs) + self.generate_kwargs.append(kwargs) + output = self.outputs.pop(0) + if isinstance(output, Exception): + raise output + return [{"text": output}] + + +class ControllableVad: + def __init__(self, sample_rate=16000): + self.sample_rate = sample_rate + self.current_speech_start = 0 + self.total_samples = 0 + self.lock_next = False + + def feed(self, audio, is_final=False): + self.total_samples += len(audio) + if not self.lock_next: + return [] + self.lock_next = False + self.current_speech_start = None + return [[0, int(self.total_samples * 1000 / self.sample_rate)]] + + def start_utterance(self, start_ms): + if self.current_speech_start is None: + self.current_speech_start = start_ms + + def reset(self): + self.current_speech_start = None + + +class PreviousBoundaryVad(ControllableVad): + def feed(self, audio, is_final=False): + self.total_samples += len(audio) + if not self.lock_next: + return [] + self.lock_next = False + self.current_speech_start = None + end_samples = self.total_samples - len(audio) + return [[0, int(end_samples * 1000 / self.sample_rate)]] + + +class FinalFlushVad: + def __init__(self, sample_rate=16000): + self.sample_rate = sample_rate + self.current_speech_start = None + self.total_samples = 0 + self.finalize_calls = 0 + + def feed(self, audio, is_final=False): + assert is_final is False + self.total_samples += len(audio) + return [] + + def finalize(self): + self.finalize_calls += 1 + return [[0, int(self.total_samples * 1000 / self.sample_rate)]] + + def reset(self): + self.current_speech_start = None + + +class RecordingSpeakerTracker: + def __init__(self): + self.assigned = [] + self.finalize_calls = 0 + + def assign_streaming(self, audio, start_sec, end_sec, sentence): + sentence["spk"] = 7 + self.assigned.append((len(audio), start_sec, end_sec)) + + def finalize(self, sentences): + self.finalize_calls += 1 + return sentences + + def reset(self): + pass + + +class OneShotSegmentVad: + def __init__(self, sample_rate=16000): + self.sample_rate = sample_rate + self.current_speech_start = 0 + self.emitted = False + + def feed(self, audio, is_final=False): + if self.emitted: + return [] + self.emitted = True + self.current_speech_start = None + return [[0, int(len(audio) * 1000 / self.sample_rate)]] + + def reset(self): + self.current_speech_start = None + + +def make_one_shot_segment_session( + module, final_text, partial_text, partial_start_ms=0, engine=None +): + sample_rate = 16000 + session = module.RealtimeASRSession( + vllm_engine=engine or FixedTextEngine(final_text), + asr_kwargs={}, + vad=OneShotSegmentVad(sample_rate), + sample_rate=sample_rate, + chunk_ms=960, + ) + session.last_partial_text = partial_text + session.last_partial_start_ms = partial_start_ms + session.last_partial_end_ms = 2000 + session.segment_partial_text = partial_text + session.segment_partial_start_ms = partial_start_ms + session.segment_partial_end_ms = 2000 + session.segment_partial_stable_count = 2 + session.last_partial_eligible = True + session.add_audio(np.zeros(2 * sample_rate, dtype=np.int16).tobytes()) + return session + + +def add_pcm(session, duration_ms, sample_rate=16000): + samples = int(duration_ms * sample_rate / 1000) + session.add_audio(np.zeros(samples, dtype=np.int16).tobytes()) + + +def decode_partial(session, duration_ms): + add_pcm(session, duration_ms, session.sample_rate) + assert session.should_decode() + return session.decode(is_final=False) + + +def test_completed_segment_keeps_aligned_partial_when_final_is_truncated_prefix(): + module = load_service_module() + partial = ( + "当然可以。目前我们已经完成了第一阶段的调查研究工作,并收集了所有必要的数据。" + "接下来我们将进入分析阶段,预计在下周完成。" + ) + + session = make_one_shot_segment_session(module, "当然可以。", partial) + + assert session.locked_sentences == [ + {"text": partial, "start": 0, "end": 2000} + ] + assert session.last_partial_text == "" + + +def test_completed_segment_keeps_clean_aligned_partial_when_final_hallucinates(): + module = load_service_module() + partial = "不好意思,上午杭高架交通比较混乱一点,我们不晓得我们有点耽误。" + repeated = "好,拜拜。" * 80 + + session = make_one_shot_segment_session(module, repeated, partial) + + assert session.locked_sentences == [ + {"text": partial, "start": 0, "end": 2000} + ] + assert session.last_partial_text == "" + + +def test_completed_segment_keeps_aligned_partial_when_final_decode_fails(): + module = load_service_module() + partial = "最终解码异常时,已经稳定显示的内容不应整句丢失。" + + session = make_one_shot_segment_session( + module, + "", + partial, + engine=FailingTextEngine(), + ) + + assert session.locked_sentences == [ + {"text": partial, "start": 0, "end": 2000} + ] + assert session.last_partial_text == "" + + +def test_completed_segment_keeps_distinct_final_correction_and_clears_partial(): + module = load_service_module() + + session = make_one_shot_segment_session( + module, + "感谢各位今天参加会议。", + "感谢各位今天参加会以。", + ) + + assert session.locked_sentences == [ + {"text": "感谢各位今天参加会议。", "start": 0, "end": 2000} + ] + assert session.last_partial_text == "" + + +def test_completed_segment_does_not_reuse_partial_from_another_window(): + module = load_service_module() + stale_partial = "当然可以。目前我们已经完成了第一阶段的调查研究工作。" + + session = make_one_shot_segment_session( + module, + "当然可以。", + stale_partial, + partial_start_ms=5000, + ) + + assert session.locked_sentences == [ + {"text": "当然可以。", "start": 0, "end": 2000} + ] + assert session.last_partial_text == "" + + +def test_hallucination_repair_handles_repeated_phrases_with_punctuation(): + module = load_service_module() + prefix = "不好意思,上午杭高架交通比较混乱一点。" + text = prefix + "好,拜拜。" * 80 + + repaired, hallucinated = module.detect_and_fix_hallucination(text) + + assert hallucinated is True + assert repaired == prefix + "好,拜拜。" + + +def test_server_stop_under_one_chunk_keeps_partial_when_final_decode_fails(): + module = load_service_module() + engine = ScriptedTextEngine( + ["已经稳定显示的短句。", RuntimeError("synthetic final decode failure")] + ) + session = module.RealtimeASRSession( + engine, + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=960, + ) + + partial = decode_partial(session, 600) + assert partial["partial"] == "已经稳定显示的短句。" + assert len(engine.generate_kwargs) == 1 + result = session.decode(is_final=True) + + assert len(engine.generate_kwargs) == 2 + assert engine.outputs == [] + assert result["sentences"] == [ + {"text": "已经稳定显示的短句。", "start": 0, "end": 600} + ] + assert session.last_partial_text == "" + + +def test_successful_empty_final_does_not_restore_old_partial(): + module = load_service_module() + partial = "会议取消。稍后另行通知。" + session = module.RealtimeASRSession( + ScriptedTextEngine([partial, partial, ""]), + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=500, + ) + + decode_partial(session, 500) + decode_partial(session, 500) + result = session.decode(is_final=True) + + assert result["sentences"] == [] + assert session.last_partial_text == "" + + +def test_short_dense_final_can_remove_a_speculative_partial_suffix(): + module = load_service_module() + partial = "会议取消。稍后另行通知。" + session = module.RealtimeASRSession( + ScriptedTextEngine([partial, partial, "会议取消。"]), + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=500, + ) + + decode_partial(session, 500) + decode_partial(session, 500) + result = session.decode(is_final=True) + + assert result["sentences"] == [ + {"text": "会议取消。", "start": 0, "end": 1000} + ] + + +def test_successful_final_can_remove_suffix_after_two_growing_partials(): + module = load_service_module() + session = module.RealtimeASRSession( + ScriptedTextEngine( + [ + "会议取消。稍后另行通知。", + "会议取消。稍后另行通知,请等待后续安排。", + "会议取消。", + ] + ), + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=1000, + endpoint_mode="client", + ) + + decode_partial(session, 1000) + decode_partial(session, 1000) + result = session.decode(is_final=True) + + assert result["sentences"] == [ + {"text": "会议取消。", "start": 0, "end": 2000} + ] + + +def test_regressed_partial_resets_confidence_before_final_correction(): + module = load_service_module() + partial = "会议取消。稍后另行通知,请等待后续安排。" + session = module.RealtimeASRSession( + ScriptedTextEngine([partial, partial, "会议取消。", "会议取消。"]), + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=1000, + ) + + decode_partial(session, 1000) + decode_partial(session, 1000) + decode_partial(session, 1000) + result = session.decode(is_final=True) + + assert result["sentences"] == [ + {"text": "会议取消。", "start": 0, "end": 3000} + ] + + +def test_vad_lock_keeps_stable_full_partial_and_assigns_speaker(): + module = load_service_module() + partial = ( + "当然可以。目前我们已经完成了第一阶段的调查研究工作,并收集了所有必要的数据。" + "接下来我们将进入分析阶段。" + ) + vad = PreviousBoundaryVad() + speaker = RecordingSpeakerTracker() + session = module.RealtimeASRSession( + ScriptedTextEngine([partial, partial, "当然可以。"]), + {}, + vad, + spk_tracker=speaker, + sample_rate=16000, + chunk_ms=1000, + ) + + decode_partial(session, 1000) + decode_partial(session, 1000) + vad.lock_next = True + add_pcm(session, 1000) + + assert session.locked_sentences == [ + {"text": partial, "start": 0, "end": 2000, "spk": 7} + ] + assert speaker.assigned == [(32000, 0.0, 2.0)] + assert speaker.finalize_calls == 0 + assert session.last_partial_text == "" + + +def test_client_commit_uses_the_same_stable_partial_reconciliation(): + module = load_service_module() + partial = ( + "当然可以。目前我们已经完成了第一阶段的调查研究工作,并收集了所有必要的数据。" + "接下来我们将进入分析阶段。" + ) + speaker = RecordingSpeakerTracker() + session = module.RealtimeASRSession( + ScriptedTextEngine([partial, partial, "当然可以。"]), + {}, + ControllableVad(), + spk_tracker=speaker, + sample_rate=16000, + chunk_ms=1000, + endpoint_mode="client", + ) + + decode_partial(session, 1000) + decode_partial(session, 1000) + result = session.decode(is_final=True) + + assert result["sentences"] == [ + {"text": partial, "start": 0, "end": 2000, "spk": 7} + ] + assert speaker.finalize_calls == 1 + assert session.last_partial_text == "" + + +def test_shifted_partial_without_full_segment_history_is_not_reused(): + module = load_service_module() + partial = "当然可以。目前我们已经完成了全部调查研究工作。" + session = module.RealtimeASRSession( + ScriptedTextEngine([partial, "当然可以。"]), + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=960, + partial_window_sec=1.5, + ) + + decode_partial(session, 2000) + result = session.decode(is_final=True) + + assert result["sentences"] == [ + {"text": "当然可以。", "start": 0, "end": 2000} + ] + + +def test_partial_missing_the_segment_tail_is_not_reused(): + module = load_service_module() + partial = "当然可以。目前我们已经完成了全部调查研究工作,并进入分析阶段。" + session = module.RealtimeASRSession( + ScriptedTextEngine([partial, partial, "当然可以。"]), + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=1000, + endpoint_mode="client", + ) + + decode_partial(session, 1000) + decode_partial(session, 1000) + add_pcm(session, 1000) + result = session.decode(is_final=True) + + assert result["sentences"] == [ + {"text": "当然可以。", "start": 0, "end": 3000} + ] + + +def test_nonoverlapping_partial_invalidates_old_full_segment_history(): + module = load_service_module() + partial = "当然可以。目前我们已经完成了全部调查研究工作,并进入分析阶段。" + session = module.RealtimeASRSession( + ScriptedTextEngine( + [ + partial, + partial, + "完全无关的新窗口内容", + "仍然无关的后续窗口内容", + "当然可以。", + ] + ), + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=500, + partial_window_sec=1.0, + endpoint_mode="client", + ) + + decode_partial(session, 500) + decode_partial(session, 500) + decode_partial(session, 500) + decode_partial(session, 500) + result = session.decode(is_final=True) + + assert result["sentences"] == [ + {"text": "当然可以。", "start": 0, "end": 2000} + ] + + +def test_hallucinated_partial_is_not_locked_when_final_decode_fails(): + module = load_service_module() + session = module.RealtimeASRSession( + ScriptedTextEngine( + ["好,拜拜。" * 80, RuntimeError("synthetic final decode failure")] + ), + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=500, + endpoint_mode="client", + ) + + partial = decode_partial(session, 500) + assert partial["partial"] == "好,拜拜。" + result = session.decode(is_final=True) + + assert result["sentences"] == [] + + +def test_hallucinated_partial_breaks_stable_partial_continuity(): + module = load_service_module() + partial = "会议取消。稍后另行通知,请等待后续安排。" + session = module.RealtimeASRSession( + ScriptedTextEngine( + [partial, "好,拜拜。" * 80, partial, "会议取消。"] + ), + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=1000, + endpoint_mode="client", + ) + + decode_partial(session, 1000) + decode_partial(session, 1000) + decode_partial(session, 1000) + result = session.decode(is_final=True) + + assert result["sentences"] == [ + {"text": "会议取消。", "start": 0, "end": 3000} + ] + + +def test_partial_decode_error_breaks_stable_partial_continuity(): + module = load_service_module() + partial = "会议取消。稍后另行通知,请等待后续安排。" + session = module.RealtimeASRSession( + ScriptedTextEngine( + [partial, RuntimeError("synthetic partial failure"), partial, "会议取消。"] + ), + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=1000, + endpoint_mode="client", + ) + + decode_partial(session, 1000) + decode_partial(session, 1000) + decode_partial(session, 1000) + result = session.decode(is_final=True) + + assert result["sentences"] == [ + {"text": "会议取消。", "start": 0, "end": 3000} + ] + + +def test_empty_partial_breaks_stable_partial_continuity(): + module = load_service_module() + partial = "会议取消。稍后另行通知,请等待后续安排。" + session = module.RealtimeASRSession( + ScriptedTextEngine([partial, "", partial, "会议取消。"]), + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=1000, + endpoint_mode="client", + ) + + decode_partial(session, 1000) + decode_partial(session, 1000) + decode_partial(session, 1000) + result = session.decode(is_final=True) + + assert result["sentences"] == [ + {"text": "会议取消。", "start": 0, "end": 3000} + ] + + +def test_partial_overlap_cannot_bridge_an_audio_coverage_gap(): + module = load_service_module() + session = module.RealtimeASRSession( + ScriptedTextEngine( + ["ABCDEFGH", "EFGHIJKLMNOP", "好,拜拜。" * 80] + ), + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=500, + partial_window_sec=1.0, + endpoint_mode="client", + ) + + decode_partial(session, 500) + decode_partial(session, 2000) + result = session.decode(is_final=True) + + assert result["sentences"] == [ + {"text": "好,拜拜。", "start": 0, "end": 2500} + ] + + +def test_partial_overlap_cannot_bridge_a_100ms_audio_gap(): + module = load_service_module() + session = module.RealtimeASRSession( + ScriptedTextEngine( + ["ABCDEFGH", "EFGHIJKLMNOP", "好,拜拜。" * 80] + ), + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=500, + partial_window_sec=0.5, + endpoint_mode="client", + ) + + decode_partial(session, 500) + decode_partial(session, 600) + result = session.decode(is_final=True) + + assert result["sentences"] == [ + {"text": "好,拜拜。", "start": 0, "end": 1100} + ] + + +def test_partial_must_start_exactly_at_segment_start_for_fallback(): + module = load_service_module() + session = module.RealtimeASRSession( + ScriptedTextEngine( + ["ABCDEFGHIJKLMNOP", "ABCDEFGHIJKLMNOP", "ABCD"] + ), + {}, + ControllableVad(), + sample_rate=16000, + chunk_ms=500, + partial_window_sec=1.9, + endpoint_mode="client", + ) + + decode_partial(session, 2000) + session.decode(is_final=False) + result = session.decode(is_final=True) + + assert result["sentences"] == [ + {"text": "ABCD", "start": 0, "end": 2000} + ] + + +def test_server_stop_assigns_speaker_to_active_segment(): + module = load_service_module() + speaker = RecordingSpeakerTracker() + session = module.RealtimeASRSession( + FixedTextEngine("已经完成的短句。"), + {}, + ControllableVad(), + spk_tracker=speaker, + sample_rate=16000, + chunk_ms=960, + ) + + add_pcm(session, 600) + result = session.decode(is_final=True) + + assert result["sentences"] == [ + {"text": "已经完成的短句。", "start": 0, "end": 600, "spk": 7} + ] + assert speaker.assigned == [(9600, 0.0, 0.6)] + assert speaker.finalize_calls == 1 + + +def test_server_stop_flushes_vad_before_decoding_final_segment(): + module = load_service_module() + vad = FinalFlushVad() + speaker = RecordingSpeakerTracker() + engine = FixedTextEngine("由 VAD final flush 确认的短句。") + session = module.RealtimeASRSession( + engine, + {}, + vad, + spk_tracker=speaker, + sample_rate=16000, + chunk_ms=960, + ) + + add_pcm(session, 600) + result = session.decode(is_final=True) + + assert vad.finalize_calls == 1 + assert len(engine.generate_kwargs) == 1 + assert result["sentences"] == [ + { + "text": "由 VAD final flush 确认的短句。", + "start": 0, + "end": 600, + "spk": 7, + } + ] + assert speaker.finalize_calls == 1 + + +def test_server_stop_finalizes_speakers_after_vad_already_locked(): + module = load_service_module() + vad = ControllableVad() + speaker = RecordingSpeakerTracker() + session = module.RealtimeASRSession( + FixedTextEngine("已经由 VAD 锁定的句子。"), + {}, + vad, + spk_tracker=speaker, + sample_rate=16000, + chunk_ms=960, + ) + + vad.lock_next = True + add_pcm(session, 600) + assert speaker.finalize_calls == 0 + result = session.decode(is_final=True) + + assert result["sentences"] == [ + { + "text": "已经由 VAD 锁定的句子。", + "start": 0, + "end": 600, + "spk": 7, + } + ] + assert speaker.finalize_calls == 1 + + +def test_overlapping_partial_windows_preserve_long_segment_on_final_regression(): + module = load_service_module() + vad = PreviousBoundaryVad() + session = module.RealtimeASRSession( + ScriptedTextEngine( + [ + "ABCD", + "ABCDEFG", + "ABCDEFGHI", + "ABCDEFGHIJ", + "DEFGHIJKLMNO", + "好,拜拜。" * 80, + ] + ), + {}, + vad, + sample_rate=16000, + chunk_ms=500, + partial_window_sec=2.0, + ) + + for _ in range(5): + decode_partial(session, 500) + vad.lock_next = True + add_pcm(session, 500) + + assert session.locked_sentences == [ + {"text": "ABCDEFGHIJKLMNO", "start": 0, "end": 2500} + ] + assert session.last_partial_text == "" + + +def test_hallucination_repair_stops_before_the_next_opening_quote(): + module = load_service_module() + prefix = "前缀:" + + repaired, hallucinated = module.detect_and_fix_hallucination( + prefix + "“好”" * 80 + ) + + assert hallucinated is True + assert repaired == prefix + "“好”" + + +def test_overlap_merge_preserves_casefold_expansion_source_positions(): + module = load_service_module() + + assert module._merge_overlapping_transcripts( + "Die Straße", "Straße ist frei" + ) == "Die Straße ist frei" + + +def test_overlap_merge_preserves_new_punctuation_after_overlap(): + module = load_service_module() + + assert module._merge_overlapping_transcripts( + "今天天气", "今天天气,真好" + ) == "今天天气,真好" + + +def test_hallucination_repair_keeps_one_ascii_quoted_phrase(): + module = load_service_module() + + repaired, hallucinated = module.detect_and_fix_hallucination( + "前缀:" + '"good"' * 80 + ) + + assert hallucinated is True + assert repaired == '前缀:"good"' + + +def test_hallucination_repair_distinguishes_apostrophe_from_ascii_quote(): + module = load_service_module() + + repaired, hallucinated = module.detect_and_fix_hallucination( + "don't " + "'good'" * 80 + ) + + assert hallucinated is True + assert repaired == "don't 'good'" + + +def test_overlap_merge_rejects_casefold_split_inside_source_character(): + module = load_service_module() + + assert module._merge_overlapping_transcripts("abcs", "abcßXYZ") is None + + def test_client_endpoint_mode_skips_fsmn_vad_model(monkeypatch): module = load_service_module() auto_model_calls = []