Add realtime voiceprint speaker matching

README.md

@@ -100,6 +100,8 @@ TEXT_MODEL_API_KEY=none
 
 VOICEPRINT_STORE=/data/asr_voiceprints.json
 VOICEPRINT_MATCH_THRESHOLD=0.45
+REALTIME_VOICEPRINT_ENABLED=true
+REALTIME_VOICEPRINT_WINDOW_SECONDS=2.0
 ```
 
 说明：
@@ -109,6 +111,7 @@ VOICEPRINT_MATCH_THRESHOLD=0.45
 - `FUNASR_WEBSOCKET_IP` 是 Speakr 后端实时录音代理要连接的 WebSocket 地址。
 - `TARGET_WS_URL` 是 `fastapi-wss` 再往后连接的 ASR WebSocket 地址。
 - `VOICEPRINT_MATCH_THRESHOLD` 是声纹匹配阈值。值越低越容易匹配，值越高越不容易误匹配。
+- `REALTIME_VOICEPRINT_WINDOW_SECONDS` 是实时声纹匹配窗口，当前为 2 秒；识别抖动时可调到 3-5 秒。
 
 ## 运行数据
 

config.env

@@ -36,10 +36,13 @@ REALTIME_PARTIAL_INTERVAL_SECONDS=2.0
 
 # fastapi_wss legacy/local ASR-with-speaker helper
 ASR_LOCAL_URL=http://asr-test:59805/asr
-SPEAKER_DET_URL=http://voice-dected:8000/extract_embedding
+SPEAKER_DET_URL=http://voice-dected:8000/extract_embedding
-
+REALTIME_VOICEPRINT_ENABLED=true
-# Runtime
+REALTIME_VOICEPRINT_WINDOW_SECONDS=2.0
-DEVICE=cuda
+REALTIME_VOICEPRINT_MIN_SECONDS=1.2
+
+# Runtime
+DEVICE=cuda
 CUDA_VISIBLE_DEVICES=0
 NVIDIA_VISIBLE_DEVICES=7
 PYTHONUNBUFFERED=1

docker-compose.yml

@@ -193,12 +193,18 @@ services:
     environment:
       TARGET_WS_URL: ws://funasr-runtime-2pass:10095
       REDIS_URL: redis://redis:6379/0
-    volumes:
+      VOICEPRINT_STORE: /data/asr/asr_voiceprints.json
-      - ./fastapi_wss:/app
+      SPEAKER_DET_URL: http://voice-dected:8000/extract_embedding
-    ports:
+      REALTIME_VOICEPRINT_ENABLED: "true"
-      - "10095:10095"
+      REALTIME_VOICEPRINT_WINDOW_SECONDS: "2.0"
+    volumes:
+      - ./fastapi_wss:/app
+      - ./deploy/data/asr:/data/asr:ro
+    ports:
+      - "10095:10095"
     depends_on:
       - funasr-runtime-2pass
+      - voice-dected
       - redis
     command:
       - python

fastapi_wss/src/a2a_wss.py

@@ -15,14 +15,21 @@ WebSocket 智能语义分析服务
 
 import asyncio
 import json
+import math
 import os
+import time
+import uuid
+import wave
 import websockets
 import re
 import logging
+from io import BytesIO
 from contextlib import asynccontextmanager
 from pathlib import Path
 from typing import List, Optional, Dict, Set
 from dataclasses import dataclass
+from urllib import request as urllib_request
+from urllib.error import URLError, HTTPError
 from dotenv import load_dotenv
 
 try:
@@ -136,6 +143,18 @@ LLM_ROLE_HISTORY_MAX_CHARS = int(get_env("LLM_ROLE_HISTORY_MAX_CHARS", "1200"))
 # 候选说话人列表（用于身份识别 Agent）
 CANDIDATE_SPEAKERS = ["局长", "主持人", "商务专员", "市场专员", "控制要素席", "空中侦察席"]
 
+# 实时声纹匹配配置。文字实时转写不等待声纹；声纹结果稍后回填到对应句子。
+REALTIME_VOICEPRINT_ENABLED = get_env_bool("REALTIME_VOICEPRINT_ENABLED", True)
+REALTIME_VOICEPRINT_WINDOW_SECONDS = float(get_env("REALTIME_VOICEPRINT_WINDOW_SECONDS", "2.0"))
+REALTIME_VOICEPRINT_MIN_SECONDS = float(get_env("REALTIME_VOICEPRINT_MIN_SECONDS", "1.2"))
+REALTIME_VOICEPRINT_BUFFER_SECONDS = float(get_env("REALTIME_VOICEPRINT_BUFFER_SECONDS", "8.0"))
+REALTIME_VOICEPRINT_SAMPLE_RATE = int(get_env("REALTIME_VOICEPRINT_SAMPLE_RATE", "16000"))
+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"))
+SPEAKER_DET_URL = get_env("SPEAKER_DET_URL", "http://voice-dected:8000/extract_embedding")
+
 # ========================
 # 3. Redis 管理器
 # ========================
@@ -606,9 +625,7 @@ class A2AOrchestrator:
 def build_acoustic_speaker_message(data: Dict) -> Optional[Dict]:
     text = (data.get("text") or "").strip()
     speaker = (data.get("spk_name") or "").strip()
-    if not speaker or speaker.lower() == "unknown":
+    if not text or not speaker or speaker.lower() == "unknown":
-        speaker = get_env("REALTIME_DEFAULT_SPEAKER", "SPEAKER_00")
-    if not text or not speaker:
         return None
 
     segment = {
@@ -638,6 +655,193 @@ def build_acoustic_speaker_message(data: Dict) -> Optional[Dict]:
     return msg
 
 
+def build_voiceprint_speaker_message(data: Dict, speaker: str, score: float) -> Optional[Dict]:
+    text = (data.get("text") or "").strip()
+    if not text or not speaker:
+        return None
+
+    segment = {
+        "id": 0,
+        "speaker": speaker,
+        "embedding": None,
+        "seek": 0,
+        "full_text": text,
+        "start": None,
+        "end": None,
+    }
+    msg = {
+        "mode": "offline-speaker",
+        "type": "asr_with_speaker",
+        "source": "voiceprint",
+        "speaker": speaker,
+        "spk_score": score,
+        "text": text,
+        "wav_name": data.get("wav_name"),
+        "is_final": data.get("is_final", True),
+        "segments": [segment],
+    }
+    if "timestamp" in data:
+        msg["timestamp"] = data["timestamp"]
+    return msg
+
+
+class RollingPcmBuffer:
+    def __init__(self, max_seconds: float):
+        self.bytes_per_second = (
+            REALTIME_VOICEPRINT_SAMPLE_RATE
+            * REALTIME_VOICEPRINT_CHANNELS
+            * REALTIME_VOICEPRINT_SAMPLE_WIDTH
+        )
+        self.max_bytes = max(int(max_seconds * self.bytes_per_second), self.bytes_per_second)
+        self._buf = bytearray()
+        self._lock = asyncio.Lock()
+
+    async def append(self, chunk: bytes):
+        if not chunk:
+            return
+        async with self._lock:
+            self._buf.extend(chunk)
+            overflow = len(self._buf) - self.max_bytes
+            if overflow > 0:
+                del self._buf[:overflow]
+
+    async def snapshot(self, seconds: float) -> bytes:
+        want = max(0, int(seconds * self.bytes_per_second))
+        async with self._lock:
+            if want <= 0 or want >= len(self._buf):
+                return bytes(self._buf)
+            return bytes(self._buf[-want:])
+
+
+class VoiceprintMatcher:
+    def __init__(self):
+        self.store_path = Path(VOICEPRINT_STORE)
+        self.match_threshold = VOICEPRINT_MATCH_THRESHOLD
+        self._mtime = None
+        self._profiles: Dict[str, List[float]] = {}
+
+    def _load_profiles(self) -> Dict[str, List[float]]:
+        try:
+            stat = self.store_path.stat()
+        except FileNotFoundError:
+            logger.warning("实时声纹库不存在: %s", self.store_path)
+            self._profiles = {}
+            self._mtime = None
+            return self._profiles
+
+        if self._mtime == stat.st_mtime and self._profiles:
+            return self._profiles
+
+        try:
+            data = json.loads(self.store_path.read_text(encoding="utf-8"))
+        except Exception as exc:
+            logger.error("读取实时声纹库失败: %s", exc, exc_info=True)
+            self._profiles = {}
+            self._mtime = stat.st_mtime
+            return self._profiles
+
+        profiles: Dict[str, List[float]] = {}
+        containers = data.values() if isinstance(data, dict) else data
+        if isinstance(containers, dict):
+            containers = containers.values()
+        for container in containers:
+            if isinstance(container, dict) and "embedding" in container:
+                name = container.get("name") or container.get("voice_name")
+                emb = container.get("embedding")
+                if name and isinstance(emb, list):
+                    profiles[str(name)] = emb
+            elif isinstance(container, dict):
+                for name, item in container.items():
+                    if isinstance(item, dict) and isinstance(item.get("embedding"), list):
+                        profiles[str(item.get("name") or name)] = item["embedding"]
+
+        self._profiles = profiles
+        self._mtime = stat.st_mtime
+        logger.info("实时声纹库已加载: %d 个说话人", len(profiles))
+        return profiles
+
+    @staticmethod
+    def _pcm_to_wav_bytes(pcm: bytes) -> bytes:
+        bio = BytesIO()
+        with wave.open(bio, "wb") as wf:
+            wf.setnchannels(REALTIME_VOICEPRINT_CHANNELS)
+            wf.setsampwidth(REALTIME_VOICEPRINT_SAMPLE_WIDTH)
+            wf.setframerate(REALTIME_VOICEPRINT_SAMPLE_RATE)
+            wf.writeframes(pcm)
+        return bio.getvalue()
+
+    @staticmethod
+    def _cosine(a: List[float], b: List[float]) -> float:
+        if not a or not b or len(a) != len(b):
+            return -1.0
+        dot = sum(float(x) * float(y) for x, y in zip(a, b))
+        na = math.sqrt(sum(float(x) * float(x) for x in a))
+        nb = math.sqrt(sum(float(y) * float(y) for y in b))
+        if na <= 0 or nb <= 0:
+            return -1.0
+        return dot / (na * nb)
+
+    def _extract_embedding_sync(self, pcm: bytes) -> Optional[List[float]]:
+        wav_bytes = self._pcm_to_wav_bytes(pcm)
+        boundary = "----voiceprint-%s" % uuid.uuid4().hex
+        body = BytesIO()
+        body.write(f"--{boundary}\r\n".encode())
+        body.write(b'Content-Disposition: form-data; name="wav_file"; filename="realtime.wav"\r\n')
+        body.write(b"Content-Type: audio/wav\r\n\r\n")
+        body.write(wav_bytes)
+        body.write(f"\r\n--{boundary}--\r\n".encode())
+        req = urllib_request.Request(
+            SPEAKER_DET_URL,
+            data=body.getvalue(),
+            headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
+            method="POST",
+        )
+        with urllib_request.urlopen(req, timeout=8) as resp:
+            payload = json.loads(resp.read().decode("utf-8"))
+        emb = payload.get("embedding")
+        return emb if isinstance(emb, list) else None
+
+    async def match_pcm(self, pcm: bytes) -> Optional[Dict]:
+        min_bytes = int(
+            REALTIME_VOICEPRINT_MIN_SECONDS
+            * REALTIME_VOICEPRINT_SAMPLE_RATE
+            * REALTIME_VOICEPRINT_CHANNELS
+            * REALTIME_VOICEPRINT_SAMPLE_WIDTH
+        )
+        if len(pcm) < min_bytes:
+            return None
+
+        profiles = self._load_profiles()
+        if not profiles:
+            return None
+
+        try:
+            embedding = await asyncio.to_thread(self._extract_embedding_sync, pcm)
+        except (HTTPError, URLError, TimeoutError, OSError) as exc:
+            logger.warning("实时声纹提取失败: %s", exc)
+            return None
+        except Exception as exc:
+            logger.error("实时声纹提取异常: %s", exc, exc_info=True)
+            return None
+        if not embedding:
+            return None
+
+        best_name = None
+        best_score = -1.0
+        for name, profile_embedding in profiles.items():
+            score = self._cosine(embedding, profile_embedding)
+            if score > best_score:
+                best_name = name
+                best_score = score
+
+        if best_name and best_score >= self.match_threshold:
+            logger.info("实时声纹匹配: %s score=%.4f", best_name, best_score)
+            return {"speaker": best_name, "score": best_score}
+
+        logger.info("实时声纹未匹配: best=%s score=%.4f threshold=%.4f", best_name, best_score, self.match_threshold)
+        return None
+
+
 MOJIBAKE_MARKERS = ("锛", "涓", "灏", "鍒", "鏄", "鎴", "浣", "鐨", "璇", "濂", "绗")
 
 
@@ -850,6 +1054,8 @@ async def forward_audio(websocket):
 
     # 用于追踪后台任务，防止连接关闭时任务被强制销毁
     background_tasks: Set[asyncio.Task] = set()
+    audio_buffer = RollingPcmBuffer(REALTIME_VOICEPRINT_BUFFER_SECONDS)
+    voiceprint_matcher = VoiceprintMatcher() if REALTIME_VOICEPRINT_ENABLED else None
 
     try:
         # 连接到目标 ASR 服务
@@ -873,6 +1079,8 @@ async def forward_audio(websocket):
                 """
                 try:
                     async for message in websocket:
+                        if isinstance(message, (bytes, bytearray)):
+                            await audio_buffer.append(bytes(message))
                         await target_ws.send(message)
                 except websockets.exceptions.ConnectionClosed:
                     logger.info("客户端连接已关闭，停止上行转发")
@@ -897,6 +1105,27 @@ async def forward_audio(websocket):
                 except Exception as e:
                     logger.error(f"发送LLM角色结果失败: {e}", exc_info=True)
 
+            async def send_voiceprint_update(asr_data: Dict):
+                if voiceprint_matcher is None:
+                    return
+                pcm = await audio_buffer.snapshot(REALTIME_VOICEPRINT_WINDOW_SECONDS)
+                match = await voiceprint_matcher.match_pcm(pcm)
+                if not match:
+                    return
+                speaker_msg = build_voiceprint_speaker_message(
+                    asr_data,
+                    match["speaker"],
+                    match["score"],
+                )
+                if speaker_msg is None:
+                    return
+                try:
+                    await websocket.send(json.dumps(speaker_msg, ensure_ascii=False))
+                except websockets.exceptions.ConnectionClosed:
+                    logger.info("客户端连接已关闭，停止发送实时声纹结果")
+                except Exception as exc:
+                    logger.error("发送实时声纹结果失败: %s", exc, exc_info=True)
+
             async def target_to_frontend():
                 """
                 下行转发：ASR 识别结果 -> 前端
@@ -924,6 +1153,10 @@ async def forward_audio(websocket):
                                             t = asyncio.create_task(send_role_update(speaker_msg))
                                             background_tasks.add(t)
                                             t.add_done_callback(background_tasks.discard)
+                                    if voiceprint_matcher is not None:
+                                        t = asyncio.create_task(send_voiceprint_update(data))
+                                        background_tasks.add(t)
+                                        t.add_done_callback(background_tasks.discard)
                             except websockets.exceptions.ConnectionClosed:
                                 logger.info("客户端连接已关闭，停止发送增强结果")
                                 break

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

@@ -1125,7 +1125,7 @@ export class RealtimeAsrPipeline {
     let lastRenderedSpeaker = initialSpeaker;
     let lastContentChar = "";
 
-    segments.forEach((segment, index) => {
+    segments.forEach((segment) => {
       if (this.getShowSpeaker()) {
         if (
           segment.speaker &&
@@ -1153,9 +1153,6 @@ export class RealtimeAsrPipeline {
         lastContentChar
       );
       displayText += segText;
-      if (index < segments.length - 1 && !/\n\s*$/.test(displayText)) {
-        displayText += "\n\n";
-      }
 
       // 跟踪上一个 segment 的末尾非空白字符，传给下一次 handleWithTimestamp
       const trimmed = segText.replace(/\s+$/, "");