From 0aa100f127e9b363cd7fecca525a88c665392205 Mon Sep 17 00:00:00 2001 From: Bifang <915779419@qq.com> Date: Thu, 10 Sep 2026 16:00:36 +0800 Subject: [PATCH] feat: split short speaker handoff gaps --- realtime_asr_optimization_demo/README.md | 5 +++ realtime_asr_optimization_demo/server.py | 36 ++++++++++++++++- .../tests/test_server.py | 40 +++++++++++++++++++ 3 files changed, 79 insertions(+), 2 deletions(-) diff --git a/realtime_asr_optimization_demo/README.md b/realtime_asr_optimization_demo/README.md index 8b29c15..479647d 100644 --- a/realtime_asr_optimization_demo/README.md +++ b/realtime_asr_optimization_demo/README.md @@ -105,6 +105,7 @@ python server.py --no-browser "enable_native_partial_stream": true, "partial_interval_ms": 1200, "max_segment_sec": 12, + "speaker_gap_ms": 400, "display_merge": true } ``` @@ -120,6 +121,10 @@ python server.py --no-browser `sentence_id` 必须覆盖更新而不是追加。`sentence_strategy=0` 使用约 800ms 静音切句，`sentence_strategy=1` 使用约 1400ms 静音切段，更适合段落模式。 +说话人模式下，`speaker_gap_ms`（默认 400ms）会在有效语音达到 800ms 后 +提前结束短停顿话轮，用于拆开 A→B 的交接；它不触发额外声纹请求，同一说话人的 +相邻片段仍由辅助服务聚类为同一说话人。服务未就绪时该策略会自动关闭。 + ## 目录 - `server.py`：本地 HTTP 页面和 WebSocket 会话编排。 diff --git a/realtime_asr_optimization_demo/server.py b/realtime_asr_optimization_demo/server.py index f5f8e1a..acc8e5f 100644 --- a/realtime_asr_optimization_demo/server.py +++ b/realtime_asr_optimization_demo/server.py @@ -41,6 +41,10 @@ VAD_SILENCE_MS = 800 PARAGRAPH_SILENCE_MS = 1400 VAD_RMS_THRESHOLD = 450 MIN_SPEAKER_VOICE_MS = 800 +# 说话人分离开启时，用比普通 VAD 更短的静音作为候选话轮边界。 +# 这个边界只负责把 A→B 的短交接停顿拆开，不调用声纹模型，因此不会 +# 把窗口推理延迟叠加到音频帧处理路径；同一说话人的拆分片段仍由聚类合并。 +SPEAKER_GAP_MS = max(20, int(os.getenv("SPEAKER_GAP_MS", "400"))) LOGGER = logging.getLogger(__name__) @@ -148,6 +152,16 @@ class RealtimeSession: self.wav_data_remaining: int | None = None self.speaker_warning_sent = False self.speaker_enabled = self._parse_flag(start.get("speaker_diarization"), True) + # 没有辅助服务时不提前切段，避免服务不可用时把一段语音拆成许多未知片段。 + self.speaker_gap_enabled = self.speaker_enabled and auxiliary_service is not None + raw_speaker_gap = start.get("speaker_gap_ms") + if raw_speaker_gap is None: + self.speaker_gap_ms = SPEAKER_GAP_MS + else: + try: + self.speaker_gap_ms = max(20, int(float(raw_speaker_gap))) + except (TypeError, ValueError, OverflowError): + self.speaker_gap_ms = SPEAKER_GAP_MS self.input_stopped = False @staticmethod @@ -422,8 +436,20 @@ class RealtimeSession: if text: await self._emit_transcription(text, 0, self.segment_start_ms + self._duration_ms()) last_partial_bytes = len(self.segment_audio) - if self.silence_ms >= self.silence_limit_ms or self._duration_ms() >= self.max_segment_sec * 1000: - await self._commit_segment("silence" if self.silence_ms >= self.silence_limit_ms else "max_duration") + # 短交接停顿优先于普通 800/1400ms 静音切段，但必须先有 + # 至少 800ms 有效语音，避免把咳嗽、噪声或极短插话送去聚类。 + short_speaker_gap = ( + self.speaker_gap_enabled + and self.voiced_ms >= MIN_SPEAKER_VOICE_MS + and self.silence_ms >= self.speaker_gap_ms + ) + if short_speaker_gap or self.silence_ms >= self.silence_limit_ms or self._duration_ms() >= self.max_segment_sec * 1000: + reason = ( + "speaker_gap" if short_speaker_gap + else "silence" if self.silence_ms >= self.silence_limit_ms + else "max_duration" + ) + await self._commit_segment(reason) last_partial_bytes = 0 else: # 参考原 WebSocket 保留 200ms 前滚，减少首字低能量音素被裁掉。 @@ -571,6 +597,10 @@ async def websocket_handler(request: web.Request) -> web.WebSocketResponse: speaker_health_error = "辅助模型服务未就绪，请检查 /health 返回的 models 状态" except Exception as exc: speaker_health_error = f"说话人辅助服务不可用：{exc}" + if speaker_health_error: + # 辅助服务未就绪时先保留原始 VAD 切段，避免在没有声纹结果的 + # 情况下增加大量短片段；服务恢复后由新的会话重新启用。 + session.speaker_gap_enabled = False await session.emit( { "type": "start", @@ -583,6 +613,8 @@ async def websocket_handler(request: web.Request) -> web.WebSocketResponse: "speaker_diarization_enabled": session.speaker_enabled, "speaker_service_url": getattr(auxiliary_config, "base_url", None), "speaker_service_health": speaker_health, + "speaker_gap_enabled": session.speaker_gap_enabled, + "speaker_gap_ms": session.speaker_gap_ms, "sentence_strategy": session.sentence_strategy, "silence_limit_ms": session.silence_limit_ms, "display_state_supported": True, diff --git a/realtime_asr_optimization_demo/tests/test_server.py b/realtime_asr_optimization_demo/tests/test_server.py index 21af6ff..a5885b6 100644 --- a/realtime_asr_optimization_demo/tests/test_server.py +++ b/realtime_asr_optimization_demo/tests/test_server.py @@ -140,6 +140,45 @@ class WebSocketFlowTests(AioHTTPTestCase): self.assertEqual(final["sentences"][0]["end_time"], 1000) self.assertEqual(len(final["display_blocks"]), 2) + async def test_short_speaker_gap_splits_turn_before_next_speaker(self): + """说话人模式在短交接停顿处切段，保持每段独立进入在线聚类。""" + ws = await self.client.ws_connect("/ws") + await ws.send_json( + { + "type": "start", + "source": "mic", + "speaker_diarization": 1, + "speaker_gap_ms": 400, + "partial_interval_ms": 1200, + } + ) + self.assertEqual((await ws.receive_json())["type"], "start") + + voiced = b"\xe8\x03" * 16000 # 每位说话人一秒有效语音 + handoff_gap = b"\x00\x00" * 6400 # 400ms，低于原始 800ms 切段阈值 + trailing_silence = b"\x00\x00" * 12800 + await ws.send_bytes(voiced + handoff_gap + voiced + trailing_silence) + await ws.send_json({"type": "eof"}) + + events: list[dict[str, object]] = [] + while True: + message = await ws.receive_json() + events.append(message) + if message["type"] == "end": + break + + final_sentences = [ + message["sentences"][0] + for message in events + if message["type"] == "sentences" and message["sentences"][0]["sentence_type"] == 1 + ] + latest_by_id = {int(item["sentence_id"]): item for item in final_sentences} + latest = [latest_by_id[index] for index in sorted(latest_by_id)] + self.assertEqual(len(latest), 2) + self.assertEqual([item["speaker_id"] for item in latest], [0, 1]) + self.assertEqual(self.app[AUXILIARY_SERVICE_KEY].calls, 2) + await ws.close() + async def test_stop_waits_for_slow_speaker_and_includes_final_snapshot(self): """在 end 前必须收到所有声纹结果，不能复现页面原先五秒断开的行为。""" auxiliary = self.app[AUXILIARY_SERVICE_KEY] @@ -306,6 +345,7 @@ class WebSocketFlowTests(AioHTTPTestCase): self.assertEqual(start["type"], "start") self.assertFalse(start["speaker_service_health"]["ready"]) + self.assertFalse(start["speaker_gap_enabled"]) self.assertEqual(warning["type"], "speaker_warning") self.assertIn("未就绪", warning["message"]) await ws.close()