Fix realtime ASR preview duplication

config.env

@@ -30,7 +30,8 @@ LLM_ROLE_HISTORY_MAX_CHARS=1200
 # ASR voiceprint adapter
 VOICE_DETECTED_URL=http://voice-dected:8000
 VOICEPRINT_STORE=/data/asr_voiceprints.json
-VOICEPRINT_MATCH_THRESHOLD=0.45
+VOICEPRINT_MATCH_THRESHOLD=0.55
+VOICEPRINT_MATCH_MARGIN=0.08
 REALTIME_PARTIAL_MIN_SECONDS=1.5
 REALTIME_PARTIAL_INTERVAL_SECONDS=2.0
 

fastapi_wss/src/a2a_wss.py

@@ -153,6 +153,7 @@ REALTIME_VOICEPRINT_CHANNELS = int(get_env("REALTIME_VOICEPRINT_CHANNELS", "1"))
 REALTIME_VOICEPRINT_SAMPLE_WIDTH = int(get_env("REALTIME_VOICEPRINT_SAMPLE_WIDTH", "2"))
 VOICEPRINT_STORE = get_env("VOICEPRINT_STORE", "/data/asr/asr_voiceprints.json")
 VOICEPRINT_MATCH_THRESHOLD = float(get_env("VOICEPRINT_MATCH_THRESHOLD", "0.45"))
+VOICEPRINT_MATCH_MARGIN = float(get_env("VOICEPRINT_MATCH_MARGIN", "0.08"))
 SPEAKER_DET_URL = get_env("SPEAKER_DET_URL", "http://voice-dected:8000/extract_embedding")
 
 # ========================
@@ -637,6 +638,8 @@ def build_acoustic_speaker_message(data: Dict) -> Optional[Dict]:
         "start": None,
         "end": None,
     }
+    if data.get("speaker_ref"):
+        segment["speaker_ref"] = data["speaker_ref"]
     msg = {
         "mode": "offline-speaker",
         "type": "asr_with_speaker",
@@ -648,6 +651,8 @@ def build_acoustic_speaker_message(data: Dict) -> Optional[Dict]:
         "is_final": data.get("is_final", True),
         "segments": [segment],
     }
+    if data.get("speaker_ref"):
+        msg["speaker_ref"] = data["speaker_ref"]
     if "timestamp" in data:
         msg["timestamp"] = data["timestamp"]
     if "sentence_info" in data:
@@ -669,6 +674,8 @@ def build_voiceprint_speaker_message(data: Dict, speaker: str, score: float) ->
         "start": None,
         "end": None,
     }
+    if data.get("speaker_ref"):
+        segment["speaker_ref"] = data["speaker_ref"]
     msg = {
         "mode": "offline-speaker",
         "type": "asr_with_speaker",
@@ -680,6 +687,8 @@ def build_voiceprint_speaker_message(data: Dict, speaker: str, score: float) ->
         "is_final": data.get("is_final", True),
         "segments": [segment],
     }
+    if data.get("speaker_ref"):
+        msg["speaker_ref"] = data["speaker_ref"]
     if "timestamp" in data:
         msg["timestamp"] = data["timestamp"]
     return msg
@@ -828,17 +837,35 @@ class VoiceprintMatcher:
 
         best_name = None
         best_score = -1.0
+        second_score = -1.0
         for name, profile_embedding in profiles.items():
             score = self._cosine(embedding, profile_embedding)
             if score > best_score:
+                second_score = best_score
                 best_name = name
                 best_score = score
+            elif score > second_score:
+                second_score = score
 
-        if best_name and best_score >= self.match_threshold:
-            logger.info("实时声纹匹配: %s score=%.4f", best_name, best_score)
+        score_margin = best_score - second_score
+        if best_name and best_score >= self.match_threshold and score_margin >= VOICEPRINT_MATCH_MARGIN:
+            logger.info(
+                "实时声纹匹配: %s score=%.4f margin=%.4f",
+                best_name,
+                best_score,
+                score_margin,
+            )
             return {"speaker": best_name, "score": best_score}
 
-        logger.info("实时声纹未匹配: best=%s score=%.4f threshold=%.4f", best_name, best_score, self.match_threshold)
+        logger.info(
+            "实时声纹未匹配: best=%s score=%.4f second=%.4f margin=%.4f threshold=%.4f required_margin=%.4f",
+            best_name,
+            best_score,
+            second_score,
+            score_margin,
+            self.match_threshold,
+            VOICEPRINT_MATCH_MARGIN,
+        )
         return None
 
 
@@ -1133,17 +1160,28 @@ async def forward_audio(websocket):
                 """
                 try:
                     async for message in target_ws:
-                        # 1. 透传 ASR 原始结果到前端（字幕显示）
+                        data = None
+                        forward_message = message
+                        if isinstance(message, str):
+                            try:
+                                data = json.loads(message)
+                            except json.JSONDecodeError:
+                                data = None
+                            if isinstance(data, dict) and data.get("mode") == "2pass-offline":
+                                data = dict(data)
+                                data.setdefault("speaker_ref", uuid.uuid4().hex)
+                                forward_message = json.dumps(data, ensure_ascii=False)
+
+                        # 1. 透传 ASR 结果到前端（字幕显示）
                         try:
-                            await websocket.send(message)
+                            await websocket.send(forward_message)
                         except websockets.exceptions.ConnectionClosed:
                             logger.info("客户端连接已关闭，停止下行转发")
                             break
 
                         # 2. 如果是文本消息，尝试触发 Agent 处理
-                        if isinstance(message, str):
+                        if isinstance(data, dict):
                             try:
-                                data = json.loads(message)
                                 # 只处理最终识别结果（2pass-offline 模式）
                                 if data.get("mode") == "2pass-offline":
                                     speaker_msg = build_acoustic_speaker_message(data)

speakr/templates/index/src/asr/realtime-asr-pipeline.js

@@ -621,7 +621,7 @@ export class RealtimeAsrPipeline {
 
   // FunASR timestamp 可能是 JSON 字符串，也可能已经是数组。这里取最后一段
   // 结束时间，用于去重、旧消息过滤和后续说话人对齐。
-  extractTimestampMs(timestampPayload) {
+  extractTimestampMs(timestampPayload) {
     if (!timestampPayload) {
       return null;
     }
@@ -645,11 +645,116 @@ export class RealtimeAsrPipeline {
         lastItem.length > 1 ? Number(lastItem[1]) : Number(lastItem[0]);
       return Number.isFinite(endOrStart) ? endOrStart : null;
     } catch (_) {
-      return null;
-    }
-  }
-
-  // 手动编辑前获取最新 ASR 时间点，用它建立旧消息屏障。
+      return null;
+    }
+  }
+
+  getSpeakerRef(parsedMessage) {
+    return parsedMessage?.speaker_ref || parsedMessage?.speakerRef || "";
+  }
+
+  normalizeSpeakerMatchText(text) {
+    return (text || "").replace(/\s+/g, "").replace(/[。！？!?，,、；;：:.]/g, "");
+  }
+
+  getCommittedRealtimeRawText() {
+    const segmentText = this.segmentList
+      .filter((item) => item.epoch === this.liveTranscriptionEpoch)
+      .map((item) => item.sourceText || item.text || "")
+      .join("");
+    const pendingText = this.pendingSentenceQueue
+      .filter((item) => item.epoch === this.liveTranscriptionEpoch)
+      .map((item) => item.sourceText || item.rawText || item.correctedText || "")
+      .join("");
+    return segmentText + pendingText;
+  }
+
+  trimCommittedRealtimePrefix(incomingText) {
+    let text = incomingText || "";
+    const committedText = this.getCommittedRealtimeRawText();
+    if (!text || committedText.length < 4) {
+      return text;
+    }
+
+    const maxOverlapLength = Math.min(180, committedText.length, text.length);
+    for (let size = maxOverlapLength; size >= 4; size--) {
+      const overlapText = committedText.slice(-size);
+      const matchIndex = text.lastIndexOf(overlapText);
+      if (matchIndex !== -1) {
+        text = text.slice(matchIndex + overlapText.length);
+        break;
+      }
+    }
+    return text;
+  }
+
+  mergeOnlinePreviewText(incomingText, parsedMessage, shouldBreak) {
+    let text = this.trimCommittedRealtimePrefix(incomingText || "");
+    if (!text.trim()) {
+      this.recTextOnline = "";
+      return;
+    }
+
+    const currentText = this.trimCommittedRealtimePrefix(this.recTextOnline || "");
+    const normalizedCurrent = this.normalizeSpeakerMatchText(currentText);
+    const normalizedIncoming = this.normalizeSpeakerMatchText(text);
+    if (
+      normalizedCurrent &&
+      normalizedIncoming &&
+      normalizedCurrent.includes(normalizedIncoming) &&
+      !normalizedIncoming.includes(normalizedCurrent)
+    ) {
+      this.recTextOnline = currentText;
+      return;
+    }
+
+    const shouldReplace =
+      parsedMessage.is_full_text === true ||
+      !currentText.trim() ||
+      (normalizedCurrent && normalizedIncoming.includes(normalizedCurrent));
+
+    if (shouldReplace) {
+      this.recTextOnline = shouldBreak ? text + "\n" : text;
+      return;
+    }
+
+    let appendText = text;
+    const maxOverlapLength = Math.min(80, currentText.length, text.length);
+    for (let size = maxOverlapLength; size >= 2; size--) {
+      if (currentText.endsWith(text.slice(0, size))) {
+        appendText = text.slice(size);
+        break;
+      }
+    }
+
+    this.recTextOnline = currentText + (shouldBreak ? appendText + "\n" : appendText);
+  }
+
+  speakerMessageMatchesItem(item, messageTimestampMs, messageText, speakerRef) {
+    if (!item) {
+      return false;
+    }
+
+    if (speakerRef || item.speakerRef) {
+      return Boolean(speakerRef && item.speakerRef && speakerRef === item.speakerRef);
+    }
+
+    if (Number.isFinite(messageTimestampMs) && Number.isFinite(item.timestampMs)) {
+      return item.timestampMs === messageTimestampMs;
+    }
+
+    const normalizedMessageText = this.normalizeSpeakerMatchText(messageText);
+    const normalizedItemText = this.normalizeSpeakerMatchText(
+      item.sourceText || item.rawText || item.text || ""
+    );
+    return Boolean(
+      normalizedMessageText &&
+        normalizedItemText &&
+        normalizedMessageText === normalizedItemText
+    );
+  }
+
+  // 手动编辑前获取最新 ASR 时间点，用它建立旧消息屏障。
   getLatestRealtimeTimestamp() {
     const latestSegmentTimestamp = this.segmentList.reduce(
       (latest, segment) =>
@@ -749,11 +854,12 @@ export class RealtimeAsrPipeline {
   }
 
   // 预览文本 = 待矫正 offline 文本 + online/offline 交接文本 + 当前 online 临时文本。
-  syncRealtimePreview() {
-    // 【兜底去重】无论前面逻辑是否产生重复标点，最终渲染前统一清理
+  syncRealtimePreview() {
+    // 【兜底去重】无论前面逻辑是否产生重复标点，最终渲染前统一清理
+    this.recTextOnline = this.trimCommittedRealtimePrefix(this.recTextOnline);
     const previewText = this._deduplicateConsecutivePunctuation(this.recTextOnline);
-    this.onPreviewTextChange(previewText);
-  }
+    this.onPreviewTextChange(previewText);
+  }
 
   // 幂等查找：同一 source/timestamp 的重复 websocket 回包不能触发第二次矫正，
   // 也不能重复落到页面。
@@ -779,7 +885,7 @@ export class RealtimeAsrPipeline {
 
   // 2pass-offline 文本进入 LLM 矫正前的唯一入口。这里负责旧消息过滤、
   // 重复回包去重、编辑重叠裁剪，以及创建 pending 队列项。
-  ensurePendingSentence(sourceText, pendingTimestamp, messageEpoch) {
+  ensurePendingSentence(sourceText, pendingTimestamp, messageEpoch, speakerRef = "") {
     const timestampMs = this.extractTimestampMs(pendingTimestamp);
     if (this.isStaleRealtimeMessage(timestampMs, messageEpoch)) {
       return null;
@@ -790,10 +896,13 @@ export class RealtimeAsrPipeline {
       sourceText,
       pendingTimestamp,
       messageEpoch
-    );
-    if (existingItem) {
-      return existingItem;
-    }
+    );
+    if (existingItem) {
+      if (speakerRef && !existingItem.speakerRef) {
+        existingItem.speakerRef = speakerRef;
+      }
+      return existingItem;
+    }
 
     const trimmedText = this.trimOfflineTextWithCarryover(sourceText || "");
     if (!trimmedText) {
@@ -810,10 +919,11 @@ export class RealtimeAsrPipeline {
       timestamp: pendingTimestamp || "",
       timestampMs,
       correctedText: "",
-      processed: false,
-      correctionRequested: false,
-      epoch: messageEpoch,
-    };
+      processed: false,
+      correctionRequested: false,
+      speakerRef,
+      epoch: messageEpoch,
+    };
 
     // 按 timestampMs 顺序插入，保持队列按音频时间有序
     // 这样 flush 时按时间顺序处理，预览也按说话顺序显示
@@ -854,6 +964,7 @@ export class RealtimeAsrPipeline {
         timestamp: item.timestamp,
         timestampMs: item.timestampMs,
         speaker: item.speaker || null,
+        speakerRef: item.speakerRef || "",
         processed: true,
         epoch: item.epoch,
       };
@@ -983,13 +1094,14 @@ export class RealtimeAsrPipeline {
         id: this.pendingSentenceSequence++,
         sourceText: remainingOnlineText,
         rawText: remainingOnlineText,
-        timestamp: "",
-        timestampMs: fallbackTimestampMs,
-        correctedText: "",
-        processed: false,
-        correctionRequested: true,
-        epoch: messageEpoch,
-      };
+        timestamp: "",
+        timestampMs: fallbackTimestampMs,
+        correctedText: "",
+        processed: false,
+        correctionRequested: true,
+        speakerRef: "",
+        epoch: messageEpoch,
+      };
       this.pendingSentenceQueue.push(pendingItem);
       this.setPendingCorrectionCount(this.pendingCorrectionCount + 1);
       this._requestCorrection(pendingItem);
@@ -1192,6 +1304,13 @@ export class RealtimeAsrPipeline {
       return;
     }
 
+    const speakerRef = this.getSpeakerRef(parsedMessage);
+    const messageText =
+      parsedMessage.text ||
+      (parsedMessage.segments &&
+        parsedMessage.segments[0] &&
+        parsedMessage.segments[0].text) ||
+      "";
     const speaker =
       (parsedMessage.segments &&
         parsedMessage.segments[0] &&
@@ -1205,11 +1324,7 @@ export class RealtimeAsrPipeline {
       if (!pending || pending.epoch !== messageEpoch || pending.speaker) {
         continue;
       }
-      if (
-        Number.isFinite(messageTimestampMs) &&
-        Number.isFinite(pending.timestampMs) &&
-        pending.timestampMs !== messageTimestampMs
-      ) {
+      if (!this.speakerMessageMatchesItem(pending, messageTimestampMs, messageText, speakerRef)) {
         continue;
       }
       pending.speaker = speaker;
@@ -1225,16 +1340,11 @@ export class RealtimeAsrPipeline {
       }
       if (
         segment.speaker === null &&
-        Number.isFinite(messageTimestampMs) &&
-        Number.isFinite(segment.timestampMs) &&
-        segment.timestampMs === messageTimestampMs
+        this.speakerMessageMatchesItem(segment, messageTimestampMs, messageText, speakerRef)
       ) {
         targetIndex = i;
         break;
       }
-      if (targetIndex === -1 && segment.speaker === null) {
-        targetIndex = i;
-      }
     }
 
     if (targetIndex !== -1) {
@@ -1257,9 +1367,10 @@ export class RealtimeAsrPipeline {
 
     const parsedMessage = JSON.parse(event.data);
     const text = "" + parsedMessage.text;
-    const mode = parsedMessage.mode;
-    const timestamp = parsedMessage.timestamp;
-    const messageEpoch = this.liveTranscriptionEpoch;
+    const mode = parsedMessage.mode;
+    const timestamp = parsedMessage.timestamp;
+    const speakerRef = this.getSpeakerRef(parsedMessage);
+    const messageEpoch = this.liveTranscriptionEpoch;
 
     if (mode === "2pass-online" || mode === "online") {
       // 低延迟预览文本可能不完整，因此不进入正式 segmentList。
@@ -1277,12 +1388,8 @@ export class RealtimeAsrPipeline {
         }
       }
 
-      if (parsedMessage.is_full_text) {
-        this.recTextOnline = shouldBreak ? text + "\n" : text;
-      } else {
-        this.recTextOnline += shouldBreak ? text + "\n" : text;
-      }
-      this.lastOnlineTimestamp = currentTimestamp;
+      this.mergeOnlinePreviewText(text, parsedMessage, shouldBreak);
+      this.lastOnlineTimestamp = currentTimestamp;
     } else if (mode === "2pass-offline" || mode === "offline") {
       // online 通道也可能回显 offline 句子，这里只作为交接预览保存。
       // 正式句子只通过 offline 通道进入矫正流程。
@@ -1290,8 +1397,12 @@ export class RealtimeAsrPipeline {
         const pendingSentence = this.ensurePendingSentence(
           text,
           timestamp,
-          messageEpoch
+          messageEpoch,
+          speakerRef
         );
+        this.recTextOnline = "";
+        this.offlineSentenceHandoffText = "";
+        this.lastOnlineTimestamp = null;
         if (
           pendingSentence &&
           !pendingSentence.processed &&
@@ -1316,10 +1427,11 @@ export class RealtimeAsrPipeline {
       return;
     }
 
-    const parsedMessage = JSON.parse(event.data);
-    const mode = parsedMessage.mode;
-    const timestamp = parsedMessage.timestamp;
-    const messageEpoch = this.liveTranscriptionEpoch;
+    const parsedMessage = JSON.parse(event.data);
+    const mode = parsedMessage.mode;
+    const timestamp = parsedMessage.timestamp;
+    const speakerRef = this.getSpeakerRef(parsedMessage);
+    const messageEpoch = this.liveTranscriptionEpoch;
     const messageTimestampMs = this.extractTimestampMs(timestamp);
 
     if (mode === "2pass-offline" || mode === "offline") {
@@ -1334,13 +1446,16 @@ export class RealtimeAsrPipeline {
         this.syncRealtimePreview();
         return;
       }
-      const pendingSentence = this.ensurePendingSentence(
-        text,
-        timestamp,
-        messageEpoch
-      );
-      this.offlineSentenceHandoffText = "";
-      this.syncRealtimePreview();
+      const pendingSentence = this.ensurePendingSentence(
+        text,
+        timestamp,
+        messageEpoch,
+        speakerRef
+      );
+      this.recTextOnline = "";
+      this.offlineSentenceHandoffText = "";
+      this.lastOnlineTimestamp = null;
+      this.syncRealtimePreview();
 
       if (
         !pendingSentence ||