#!/usr/bin/env python3 """常驻加载 VAD、说话人聚类和声纹识别模型的独立服务。""" from __future__ import annotations import asyncio import math import os import sys import tempfile import time from collections import deque from collections.abc import Mapping from pathlib import Path from typing import Any from aiohttp import web from aiohttp.web_request import FileField from dotenv import load_dotenv PROJECT_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROJECT_ROOT)) from backend.model_manifest import auxiliary_models, load_manifest, model_directory # 将辅助服务端口固定在代码变量中，服务器启动时只需执行脚本，便于部署和排查。 AUXILIARY_HOST = "0.0.0.0" AUXILIARY_PORT = 8010 # 独立启动辅助服务时也加载部署配置，确保模型目录和设备与启动器一致。 load_dotenv(PROJECT_ROOT / ".env") MODELS_DIR = Path(os.getenv("MODEL_DIR", str(PROJECT_ROOT / "models"))).resolve() AUXILIARY_DEVICE = os.getenv("AUXILIARY_DEVICE", "cuda:0") # 实时追踪器的默认参数与 FunASR HybridSpeakerTracker 保持一致。 ONLINE_SPEAKER_MATCH_THRESHOLD = float(os.getenv("FUNASR_SPEAKER_MATCH_THRESHOLD", "0.6")) ONLINE_CLUSTER_MERGE_THRESHOLD = float(os.getenv("FUNASR_SPEAKER_MERGE_THRESHOLD", "0.78")) ONLINE_MAX_SPEAKERS = max(1, int(os.getenv("FUNASR_MAX_SPEAKERS", "15"))) ONLINE_SPEAKER_HISTORY_CHUNKS = max(1, int(os.getenv("FUNASR_SPEAKER_HISTORY_CHUNKS", "128"))) MIN_ONLINE_SPEAKER_AUDIO_MS = 800 # 实时 VAD 由 WebSocket 进程加载；本服务负责加载 CAM++。 # 独立的 VAD HTTP 接口仍可按需使用。 DEFAULT_PRELOAD_KINDS = {"speaker_verification"} def _coerce_finite_float(value: object) -> float | None: """把表单或模型返回的数值安全转换为有限浮点数。""" if isinstance(value, bool): return None if isinstance(value, (int, float)): parsed = float(value) elif isinstance(value, (str, bytes)): try: parsed = float(value.strip()) except (TypeError, ValueError): return None else: return None return parsed if math.isfinite(parsed) else None def _parse_form_float(value: object, field_name: str, default: float | None = None) -> float: """解析 multipart 数值字段，避免直接把 FileField/bytes 传给 float。""" parsed = _coerce_finite_float(value) if parsed is not None: return parsed if default is not None: return default raise ValueError(f"{field_name} must be a number") def _asset_ready(path: Path, config: dict[str, Any]) -> bool: """在导入或加载模型前，先检查清单声明的文件和大小要求。""" if not path.is_dir(): return False for relative_path in config.get("required_files", []): if not (path / str(relative_path)).is_file(): return False any_files = config.get("any_files", []) if any_files and not any( file_path.is_file() for pattern in any_files for file_path in path.rglob(str(pattern)) ): return False minimum_size = int(config.get("min_total_size_bytes", 0) or 0) return not minimum_size or sum( file_path.stat().st_size for file_path in path.rglob("*") if file_path.is_file() ) >= minimum_size class AuxiliaryRuntime: """管理常驻辅助模型，并串行化 GPU 推理调用以避免显存竞争。""" def __init__(self) -> None: self.manifest = load_manifest() self.assets = auxiliary_models(self.manifest) self.models: dict[str, Any] = {} self.status: dict[str, dict[str, Any]] = {} self.inference_lock = asyncio.Lock() # 每个 WebSocket 会话独立维护聚类中心，避免不同浏览器会话互相影响。 self.speaker_clusters: dict[str, list[dict[str, Any]]] = {} # 像 FunASR 追踪器一样，为每个 WebSocket 会话保留有界的滚动窗口历史。 self.speaker_history: dict[str, dict[str, Any]] = {} self.speaker_last_seen: dict[str, float] = {} self._speaker_cluster_backend: Any | None = None self._speaker_postprocess: Any | None = None def _preload_kinds(self) -> set[str]: """读取需要在启动时加载的模型类型，默认不加载完整 diarization。""" raw = os.getenv("AUXILIARY_PRELOAD_KINDS", "") if not raw.strip(): return set(DEFAULT_PRELOAD_KINDS) # CAM++ 是必需模型；其他接口可按需增加可选模型类型。 return DEFAULT_PRELOAD_KINDS | {item.strip() for item in raw.split(",") if item.strip()} def _load_asset(self, model_id: str, config: dict[str, Any], path: Path) -> Any | None: """只加载当前运行接口需要的模型；依赖模型和对齐模型先保持本地资产就绪。""" kind = str(config.get("kind") or "") if kind in {"vad", "punctuation"}: from funasr import AutoModel # 标点模型默认在 CPU 上运行，为 CAM++ 预留显存。 device = ( os.getenv("FUNASR_PUNC_DEVICE", "cpu") if kind == "punctuation" else AUXILIARY_DEVICE ) return AutoModel( model=str(path), device=device, disable_update=True, disable_pbar=True, disable_log=True, local_files_only=True, ) if kind in {"diarization", "speaker_verification", "realtime_speaker_verification"}: from modelscope.pipelines import pipeline from modelscope.utils.constant import Tasks task = Tasks.speaker_diarization if kind == "diarization" else Tasks.speaker_verification return pipeline(task=task, model=str(path), device=AUXILIARY_DEVICE) # CAM++ 依赖模型会先确认文件已下载，后续交由对应的专用推理流程加载； # 不猜测其他模型的通用加载方式，避免意外占用显存。 return None def preload(self) -> None: """预加载核心模型；可选模型失败只记录状态，避免服务整体退出。""" failures: list[str] = [] preload_kinds = self._preload_kinds() loaded_kinds: set[str] = set() for model_id, config in self.assets.items(): kind = str(config.get("kind") or "") path = model_directory(model_id, self.manifest, MODELS_DIR) if kind == "speaker_verification" and os.getenv("CAM_MODEL_PATH"): # 后端启动器会传入选定的本地 CAM++ 目录。 path = Path(os.environ["CAM_MODEL_PATH"]).resolve() record: dict[str, Any] = {"path": str(path), "asset_ready": _asset_ready(path, config)} if kind not in preload_kinds: record["state"] = "optional_not_preloaded" if record["asset_ready"] else "optional_missing" record["preload"] = False self.status[model_id] = record continue # 清单中可能同时存在 iic/damo 两个同类型 CAM++ 资产；实时路径 # 只需一份，按清单顺序选第一个成功加载的模型，避免重复占显存。 if kind == "speaker_verification" and kind in loaded_kinds: record["state"] = "duplicate_not_preloaded" record["preload"] = False self.status[model_id] = record continue record["preload"] = True if not record["asset_ready"]: record["state"] = "missing" failures.append(model_id) self.status[model_id] = record continue try: loaded = self._load_asset(model_id, config, path) if loaded is not None: self.models[model_id] = loaded record["state"] = "loaded" loaded_kinds.add(kind) else: record["state"] = "load_error" record["error"] = "model loader returned no model" if kind == "speaker_verification": failures.append(model_id) except Exception as exc: record["state"] = "load_error" record["error"] = str(exc) failures.append(model_id) self.status[model_id] = record # 启动日志必须包含每个资产的路径、是否完整和底层异常；不能只打印 # 明确说明失败原因，避免只显示笼统的启动失败，便于远程部署时区分文件缺失和版本错误。 for model_id, record in self.status.items(): print( f"[model] {model_id}: state={record.get('state')}, " f"asset_ready={record.get('asset_ready')}, path={record.get('path')}" + (f", error={record['error']}" if record.get("error") else ""), flush=True, ) # 必需的 VAD 由 WebSocket 服务负责；此进程要求 CAM++ 可用。 required_failures = [ model_id for model_id in failures if self.assets[model_id].get("kind") == "speaker_verification" and self._speaker_embedding_model_id() is None ] if required_failures: details = "; ".join( f"{model_id} -> {self.status[model_id]['path']}" f" [{self.status[model_id].get('state')}: {self.status[model_id].get('error', 'asset missing')}]" for model_id in required_failures ) raise RuntimeError( "Auxiliary core model is missing or failed to load: " + details + ". Run python scripts/download_models.py --funasr-runtime, " "or set MODEL_DIR/CAM_MODEL_PATH to the local CAM++ asset." ) def _find_model(self, kind: str) -> Any: """按模型清单中的 kind 查找一个已经加载完成的模型。""" for model_id, config in self.assets.items(): if config.get("kind") == kind and model_id in self.models: return self.models[model_id] raise RuntimeError(f"Auxiliary model is not loaded: {kind}") def _load_optional_kind(self, kind: str) -> Any: """按需加载可选模型，例如完整 diarization 接口首次被调用时。""" for model_id, config in self.assets.items(): if config.get("kind") != kind: continue path = model_directory(model_id, self.manifest, MODELS_DIR) if not _asset_ready(path, config): raise RuntimeError(f"Auxiliary model asset is missing: {model_id} ({path})") try: loaded = self._load_asset(model_id, config, path) except Exception as exc: self.status.setdefault(model_id, {})["state"] = "load_error" self.status[model_id]["error"] = str(exc) raise RuntimeError(f"Auxiliary {kind} model failed to load: {exc}") from exc if loaded is None: raise RuntimeError(f"Auxiliary model has no loader for kind: {kind}") self.models[model_id] = loaded self.status.setdefault(model_id, {})["state"] = "loaded_on_demand" return loaded raise RuntimeError(f"Auxiliary model asset is not configured: {kind}") def _speaker_embedding_model_id(self) -> str | None: """选择实时声纹模型，并在实时模型不可用时回退到普通声纹模型。""" # CAM++ 是实时聚类的主模型；其它声纹模型不能静默替代它。 for preferred_kind in ("speaker_verification",): for model_id, config in self.assets.items(): if config.get("kind") == preferred_kind and model_id in self.models: return model_id return None async def vad(self, audio_path: str) -> Any: """使用临时音频文件执行一次串行化的 VAD 推理。""" # 实时 VAD 由 WebSocket 进程处理；此 HTTP 接口按需加载模型。 try: model = self._find_model("vad") except RuntimeError: model = self._load_optional_kind("vad") async with self.inference_lock: return await asyncio.to_thread(model.generate, input=audio_path, cache={}) async def punctuate(self, text: str) -> str: """按需加载 CT-Transformer，为已完成的 ASR 轮次添加标点。""" async with self.inference_lock: try: model = self._find_model("punctuation") except RuntimeError: # 在锁保护下加载，避免多个会话重复加载模型。 model = await asyncio.to_thread(self._load_optional_kind, "punctuation") output = await asyncio.to_thread(model.generate, input=text, cache={}) if isinstance(output, (list, tuple)) and output: output = output[0] if isinstance(output, Mapping): result = output.get("text") else: result = getattr(output, "text", None) return str(result).strip() if result else text async def diarization(self, audio_path: str) -> Any: """使用 CAM++ 对完整会话执行说话人聚类，保持跨片段的标签一致性。""" try: model = self._find_model("diarization") except RuntimeError: # 完整说话人分离不在核心启动路径中，首次调用对应接口时才加载。 model = self._load_optional_kind("diarization") async with self.inference_lock: # ModelScope 的 CAM++ 推理流程以位置参数接收音频路径；使用关键字参数 # input 在不同版本中可能被忽略或直接报错。 return await asyncio.to_thread(model, audio_path) @staticmethod def _normalize_embedding(embedding: Any) -> Any: """将声纹模型输出转成单位向量，并拒绝 NaN、无穷值和零向量。""" import numpy as np vector = np.asarray(embedding, dtype=np.float32) # 单次请求只能对应一条新声纹，不能把多个样本矩阵拼接成伪造特征。 if vector.ndim > 2 or (vector.ndim == 2 and vector.shape[0] != 1): raise RuntimeError("speaker embedding must contain exactly one vector") vector = vector.reshape(-1) if vector.size == 0 or not np.isfinite(vector).all(): raise RuntimeError("speaker embedding is empty or non-finite") norm = float(np.linalg.norm(vector)) if not np.isfinite(norm) or norm < 1e-8: raise RuntimeError("speaker embedding has zero norm") return vector / norm @staticmethod def _extract_embedding_value(result: Any) -> Any | None: """从不同 ModelScope 版本的 pipeline 返回值中提取 embedding。""" if result is None: return None # ERes2Net 推理流程在 output_emb=True 时返回 {'embs': numpy.ndarray, # 'outputs': ...}；部分版本或其他声纹推理流程使用 embedding 变体字段。 if isinstance(result, Mapping): for key in ("embs", "embedding", "embeddings", "speaker_embedding", "output_embedding"): if key in result: return AuxiliaryRuntime._extract_embedding_value(result[key]) return None # 某些 ModelScope 版本会把结果包装成带 embs/embedding 属性的对象。 for key in ("embs", "embedding", "embeddings", "speaker_embedding", "output_embedding"): value = getattr(result, key, None) if value is not None: return AuxiliaryRuntime._extract_embedding_value(value) # torch.Tensor 不能直接依赖 numpy.asarray 转成普通数组；先显式移到 CPU。 detach = getattr(result, "detach", None) if callable(detach): detached = detach() cpu = getattr(detached, "cpu", None) if callable(cpu): detached = cpu() numpy_method = getattr(detached, "numpy", None) if callable(numpy_method): return numpy_method() # 单条音频通常返回 [embedding]；递归拆开这一层，数值列表则保留为向量。 if isinstance(result, (list, tuple)) and len(result) == 1: return AuxiliaryRuntime._extract_embedding_value(result[0]) return result @staticmethod def _run_embedding_pipeline(model_pipeline: Any, audio_path: str) -> Any: """调用声纹 pipeline 的公开预处理和 embedding 输出接口。""" # ModelScope 的 ERes2Net 推理流水线要求输入为音频路径列表，并通过 # output_emb=True 返回嵌入向量；不能直接把原始 waveform Tensor 传给 # pipeline.model，否则会跳过采样率、声道数和波形预处理。 try: result = model_pipeline([audio_path], output_emb=True) except TypeError: # 兼容不支持 output_emb 参数的旧推理流水线：仍使用其自带的 # preprocess/forward，而不是直接调用内部 model，以确保输入格式一致。 preprocess = getattr(model_pipeline, "preprocess", None) forward = getattr(model_pipeline, "forward", None) if not callable(preprocess) or not callable(forward): raise RuntimeError("speaker pipeline does not expose embedding inference") result = forward(preprocess([audio_path])) embedding = AuxiliaryRuntime._extract_embedding_value(result) if embedding is None: raise RuntimeError( "speaker pipeline returned no embedding " f"(result_type={type(result).__name__})" ) return embedding def _extract_embedding_sync(self, audio_path: str) -> Any: """在工作线程中读取当前 turn，并使用已加载的声纹模型提取特征。""" import librosa audio, _ = librosa.load(audio_path, sr=16000, mono=True) audio_array = audio.reshape(-1) if audio_array.size < int(16000 * MIN_ONLINE_SPEAKER_AUDIO_MS / 1000): return None model_id = self._speaker_embedding_model_id() if model_id is None: raise RuntimeError("no loaded speaker verification model is available") model_pipeline = self.models[model_id] output = self._run_embedding_pipeline(model_pipeline, audio_path) return self._normalize_embedding(output) @staticmethod def _embedding_rows(result: Any, expected_count: int) -> Any: """兼容 ModelScope 批量 embedding 的 Tensor、数组和逐条结果格式。""" import numpy as np values = AuxiliaryRuntime._extract_embedding_value(result) if values is None: raise RuntimeError("speaker pipeline returned no batch embeddings") detach = getattr(values, "detach", None) if callable(detach): values = detach() cpu = getattr(values, "cpu", None) if callable(cpu): values = cpu() numpy_method = getattr(values, "numpy", None) if callable(numpy_method): values = numpy_method() try: array = np.asarray(values, dtype=np.float32) except (TypeError, ValueError): array = np.empty((0, 0), dtype=np.float32) if array.ndim == 3 and array.shape[0] == expected_count and array.shape[1] == 1: array = array[:, 0, :] if array.ndim == 1 and expected_count == 1: array = array.reshape(1, -1) if array.ndim == 2 and array.shape[0] == expected_count: rows = [AuxiliaryRuntime._normalize_embedding(row) for row in array] return rows # 某些 ModelScope 版本每条音频返回一个对象，而不是二维矩阵。 if isinstance(values, (list, tuple)) and len(values) == expected_count: rows = [] for item in values: value = AuxiliaryRuntime._extract_embedding_value(item) if value is None: raise RuntimeError("speaker pipeline returned an item without an embedding") rows.append(AuxiliaryRuntime._normalize_embedding(value)) return rows raise RuntimeError( "speaker pipeline batch size mismatch: " f"expected {expected_count}, received shape {getattr(array, 'shape', None)}" ) @staticmethod def _write_pcm_window(path: str, samples: Any) -> None: """把补齐后的 16kHz mono 浮点窗写成 CAM++ pipeline 可读的 PCM WAV。""" import numpy as np import wave pcm = (np.clip(samples, -1.0, 1.0) * 32767.0).astype(" list[tuple[float, float, Any]]: """按 FunASR CAM++ 的 1.5s/0.75s 重叠窗批量提取声纹。""" import librosa import numpy as np audio, _ = librosa.load(audio_path, sr=16000, mono=True) audio = np.asarray(audio, dtype=np.float32).reshape(-1) min_samples = int(16000 * MIN_ONLINE_SPEAKER_AUDIO_MS / 1000) if audio.size < min_samples: return [] window_ms = max(1500, int(os.getenv("FUNASR_SPEAKER_WINDOW_MS", "1500"))) hop_ms = max(250, int(os.getenv("FUNASR_SPEAKER_HOP_MS", "750"))) window_samples = int(16000 * window_ms / 1000) hop_samples = int(16000 * hop_ms / 1000) model_id = self._speaker_embedding_model_id() if model_id is None: raise RuntimeError("no loaded CAM++ speaker verification model is available") model_pipeline = self.models[model_id] spans: list[tuple[float, float]] = [] paths: list[str] = [] with tempfile.TemporaryDirectory(prefix="funasr-campp-windows-") as temp_dir: last_end = 0 for suggested_start in range(0, audio.size, hop_samples): end = min(suggested_start + window_samples, audio.size) if end <= last_end: break last_end = end # 与 FunASR sv_chunk 一样将尾窗右对齐，确保不足 1.5 秒的末尾窗口 # 尽量包含完整的新语音，而不是只用较短尾音再补大量零。 start = max(0, end - window_samples) chunk = audio[start:end] if chunk.size < min_samples: break # 忽略近静音窗，避免把背景底噪添加成一个新说话人。 rms = float(np.sqrt(np.mean(np.square(chunk)))) if chunk.size else 0.0 if rms >= 0.0015: padded = np.zeros(window_samples, dtype=np.float32) padded[: chunk.size] = chunk path = str(Path(temp_dir) / f"window-{len(paths):04d}.wav") self._write_pcm_window(path, padded) paths.append(path) spans.append((start * 1000 / 16000, end * 1000 / 16000)) if end >= audio.size: break if not paths: return [] embeddings = [] batch_size = max(1, min(64, int(os.getenv("FUNASR_SPEAKER_BATCH_SIZE", "16")))) for offset in range(0, len(paths), batch_size): path_batch = paths[offset : offset + batch_size] try: batch_result = model_pipeline(path_batch, output_emb=True) batch_embeddings = self._embedding_rows(batch_result, len(path_batch)) except Exception: # 有些旧推理流水线不接受多文件批次；逐窗调用以保持相同预处理路径。 batch_embeddings = [ self._normalize_embedding( self._run_embedding_pipeline(model_pipeline, path) ) for path in path_batch ] embeddings.extend(batch_embeddings) return [ (start_ms, end_ms, embedding) for (start_ms, end_ms), embedding in zip(spans, embeddings) ] def _map_cluster_centers( self, session_id: str, cluster_centers: Any, ) -> list[dict[str, Any]]: """将 FunASR 临时聚类标签映射为稳定且受上限约束的会话编号。""" import numpy as np clusters = self.speaker_clusters.setdefault(session_id, []) used_ids: set[int] = set() mapped: list[dict[str, Any]] = [] for raw_center in cluster_centers: center = self._normalize_embedding(raw_center) available = [ cluster for cluster in clusters if int(cluster["speaker_id"]) not in used_ids ] best_cluster = max( available, key=lambda cluster: float(np.dot(center, cluster["embedding"])), default=None, ) best_score = ( float(np.dot(center, best_cluster["embedding"])) if best_cluster is not None else -1.0 ) matched = ( best_cluster is not None and best_score >= ONLINE_SPEAKER_MATCH_THRESHOLD ) created = False if not matched and len(clusters) < ONLINE_MAX_SPEAKERS: speaker_id = len(clusters) clusters.append( {"speaker_id": speaker_id, "embedding": center, "count": 1} ) confidence = 0.75 strategy = "online_embedding_cluster_new" created = True else: if best_cluster is None: # 如果本轮临时聚类数超过可用编号， # 则按 FunASR 的做法回退到最近的已有身份。 best_cluster = max( clusters, key=lambda cluster: float(np.dot(center, cluster["embedding"])), ) best_score = float(np.dot(center, best_cluster["embedding"])) speaker_id = int(best_cluster["speaker_id"]) confidence = max(0.0, best_score) if matched: count = int(best_cluster["count"]) weight = 1.0 / min(count + 1, 20) best_cluster["embedding"] = self._normalize_embedding( best_cluster["embedding"] * (1.0 - weight) + center * weight ) best_cluster["count"] = count + 1 strategy = "online_embedding_cluster_match" else: # 达到配置的身份上限后，保持编号稳定， # 将未匹配的聚类分配给最近的已知中心。 strategy = "online_embedding_cluster_limit_fallback" used_ids.add(speaker_id) mapped.append( { "speaker_id": speaker_id, "speaker_confidence": round( max(0.6, min(1.0, confidence)), 3 ), "speaker_strategy": strategy, } ) return mapped def _cluster_speaker_history( self, session_id: str, ) -> tuple[list[list[float]], list[dict[str, Any]]]: """使用 FunASR 自带的后端对 CAM++ 滚动历史重新聚类。""" import numpy as np import torch from funasr.models.campplus.cluster_backend import ClusterBackend from funasr.models.campplus.utils import postprocess history = self.speaker_history[session_id] embeddings = torch.as_tensor( np.stack(list(history["embeddings"])), dtype=torch.float32, device="cpu", ) if self._speaker_cluster_backend is None: self._speaker_cluster_backend = ClusterBackend( merge_thr=ONLINE_CLUSTER_MERGE_THRESHOLD ).to("cpu") self._speaker_postprocess = postprocess # ClusterBackend 生成本轮标签；后处理还会对齐重叠区间边界， # 并在分配稳定编号前平滑过短的说话人片段。 labels = self._speaker_cluster_backend(embeddings, oracle_num=None) labels = np.asarray(labels) chunks = [ [start_ms / 1000.0, end_ms / 1000.0, None] for start_ms, end_ms in history["chunks"] ] segments, centers = self._speaker_postprocess( chunks, None, labels, embeddings, return_spk_center=True, ) stable_clusters = self._map_cluster_centers(session_id, centers) return segments, stable_clusters def _assign_embedding( self, session_id: str, embedding: Any, start_time_ms: float, end_time_ms: float, ) -> dict[str, Any]: """按照 FunASR 的 15 人策略，使用整轮嵌入进行回退分配。""" import numpy as np embedding = self._normalize_embedding(embedding) clusters = self.speaker_clusters.setdefault(session_id, []) best_cluster: dict[str, Any] | None = None best_score = -1.0 for cluster in clusters: if embedding.shape != cluster["embedding"].shape: raise RuntimeError("speaker embedding dimension changed within the session") score = float(np.dot(embedding, cluster["embedding"])) if score > best_score: best_score = score best_cluster = cluster if best_cluster is not None and best_score >= ONLINE_SPEAKER_MATCH_THRESHOLD: count = int(best_cluster["count"]) weight = 1.0 / min(count + 1, 20) best_cluster["embedding"] = self._normalize_embedding( best_cluster["embedding"] * (1.0 - weight) + embedding * weight ) best_cluster["count"] = count + 1 speaker_id = int(best_cluster["speaker_id"]) confidence = best_score strategy = "online_embedding_cluster_match" elif len(clusters) < ONLINE_MAX_SPEAKERS: speaker_id = len(clusters) clusters.append({"speaker_id": speaker_id, "embedding": embedding, "count": 1}) confidence = 0.75 strategy = "online_embedding_cluster_new" else: # 达到身份上限后，使用与 FunASR 一致的回退策略。 speaker_id = int(best_cluster["speaker_id"]) if best_cluster else 0 confidence = max(0.0, best_score) strategy = "online_embedding_cluster_limit_fallback" return { "speaker_id": speaker_id, "speaker_name": f"说话人 {speaker_id + 1}", "speaker_evidence": "fresh", "speaker_confidence": round(max(0.6, min(1.0, confidence)), 3), "speaker_strategy": strategy, "speaker_status": "confirmed", "speaker_reason": "CAM++ rolling history was clustered with FunASR's speaker backend", "start_time": start_time_ms, "end_time": end_time_ms, } async def track_speakers( self, audio_path: str, session_id: str, start_time_ms: float, end_time_ms: float, ) -> list[dict[str, Any]]: """使用 FunASR 滚动窗口聚类器追踪已完成的 VAD 轮次。""" async with self.inference_lock: now = time.monotonic() for stale_id, seen in list(self.speaker_last_seen.items()): if now - seen > 1800: self.reset_speaker_session(stale_id) self.speaker_last_seen[session_id] = now windows = await asyncio.to_thread(self._extract_window_embeddings_sync, audio_path) if session_id not in self.speaker_last_seen or not windows: return [] history = self.speaker_history.get(session_id) if history is None: history = { "chunks": deque(maxlen=ONLINE_SPEAKER_HISTORY_CHUNKS), "embeddings": deque(maxlen=ONLINE_SPEAKER_HISTORY_CHUNKS), } self.speaker_history[session_id] = history # 保留绝对时间戳，以便每轮都能结合近期轮次重新聚类。 for window_start, window_end, embedding in windows: history["chunks"].append( (start_time_ms + window_start, start_time_ms + window_end) ) history["embeddings"].append(embedding.copy()) clustered_segments, stable_clusters = self._cluster_speaker_history(session_id) results: list[dict[str, Any]] = [] for segment_start, segment_end, cluster_id in clustered_segments: segment_start_ms = max(start_time_ms, float(segment_start) * 1000.0) segment_end_ms = min(end_time_ms, float(segment_end) * 1000.0) if segment_end_ms <= segment_start_ms: continue cluster_index = int(cluster_id) if cluster_index < 0 or cluster_index >= len(stable_clusters): continue stable = stable_clusters[cluster_index] speaker_id = int(stable["speaker_id"]) results.append( { "speaker_id": speaker_id, "speaker_name": f"说话人 {speaker_id + 1}", "speaker_evidence": "fresh", "speaker_confidence": stable["speaker_confidence"], "speaker_strategy": stable["speaker_strategy"], "speaker_status": "confirmed", "speaker_reason": "CAM++ rolling history was clustered with FunASR's speaker backend", "start_time": segment_start_ms, "end_time": segment_end_ms, } ) return results async def resolve_speaker( self, audio_path: str, session_id: str, start_time_ms: float, end_time_ms: float, ) -> dict[str, Any] | None: """对一个实时 turn 提取声纹，并更新该 session 的在线聚类中心。""" async with self.inference_lock: # 异常断网时客户端可能来不及发送重置请求；过期状态会在下一次请求时回收。 now = time.monotonic() for stale_id, seen in list(self.speaker_last_seen.items()): if now - seen > 1800: self.reset_speaker_session(stale_id) self.speaker_last_seen[session_id] = now embedding = await asyncio.to_thread(self._extract_embedding_sync, audio_path) if embedding is None: return {"speaker_id": -1, "speaker_evidence": "pending", "speaker_confidence": 0.0, "speaker_status": "insufficient_audio", "speaker_reason": "音频不足 800ms，未提取声纹"} # 重置请求可能在模型线程运行期间到达；结束的会话不能被晚到结果重新创建。 if session_id not in self.speaker_last_seen: return None return self._assign_embedding( session_id, embedding, start_time_ms, end_time_ms ) def reset_speaker_session(self, session_id: str) -> None: """释放已结束 WebSocket 的聚类中心，防止长时间运行时内存增长。""" self.speaker_clusters.pop(session_id, None) self.speaker_history.pop(session_id, None) self.speaker_last_seen.pop(session_id, None) MODEL_SERVICE_KEY = web.AppKey("auxiliary_runtime", AuxiliaryRuntime) async def health_handler(request: web.Request) -> web.Response: """返回模型资产完整性和预加载状态，供 WebSocket 编排服务检查。""" runtime: AuxiliaryRuntime = request.app[MODEL_SERVICE_KEY] speaker_model_id = runtime._speaker_embedding_model_id() vad_model_id = next( (model_id for model_id, config in runtime.assets.items() if config.get("kind") == "vad" and model_id in runtime.models), None, ) # 此处的就绪状态反映实时后端使用的 CAM++ 模型。 ready = speaker_model_id is not None return web.json_response( { "ready": ready, "speaker_protocol_version": 2, "vad_model": vad_model_id, "vad_ready": vad_model_id is not None, "device": AUXILIARY_DEVICE, "speaker_embedding_model": speaker_model_id, "speaker_embedding_ready": speaker_model_id is not None, "punctuation_ready": any( config.get("kind") == "punctuation" and model_id in runtime.models for model_id, config in runtime.assets.items() ), "models": runtime.status, } ) async def vad_handler(request: web.Request) -> web.Response: """接收 WAV 文件上传，并返回 FunASR 生成的语音活动区间。""" runtime: AuxiliaryRuntime = request.app[MODEL_SERVICE_KEY] form = await request.post() upload = form.get("file") if not isinstance(upload, FileField): return web.json_response({"error": "multipart field 'file' is required"}, status=400) temp_path: str | None = None try: with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file: temp_file.write(upload.file.read()) temp_path = temp_file.name result = await runtime.vad(temp_path) return web.json_response({"segments": result}) finally: if temp_path: Path(temp_path).unlink(missing_ok=True) async def punctuation_handler(request: web.Request) -> web.Response: """本地可选 CT-Transformer 可用时返回标点结果。""" runtime: AuxiliaryRuntime = request.app[MODEL_SERVICE_KEY] try: payload = await request.json() except (ValueError, web.HTTPException): return web.json_response({"error": "request body must be JSON"}, status=400) text = payload.get("text") if isinstance(payload, dict) else None if not isinstance(text, str): return web.json_response({"error": "JSON field 'text' must be a string"}, status=400) if not text.strip(): return web.json_response({"text": text, "available": True}) try: punctuated = await runtime.punctuate(text) return web.json_response({"text": punctuated, "available": True}) except Exception as exc: # 标点模型为可选项；保持 ASR 可用，并将不可用原因告知协议桥接层。 print(f"[punctuation] unavailable: {exc}", flush=True) return web.json_response({"text": text, "available": False, "error": str(exc)}) def _normalize_diarization_segments(result: Any) -> list[dict[str, Any]]: """将不同 ModelScope 版本的聚类输出统一为 start/end/speaker 字段。""" # ModelScope 通常返回 {'text': [[start_sec, end_sec, speaker_id], ...]}； # 某些版本返回对象而不是字典，因此这里同时读取属性形式。 if isinstance(result, dict): for key in ("segments", "output", "text", "result"): candidate = result.get(key) if isinstance(candidate, list): result = candidate break else: for key in ("segments", "output", "text", "result"): candidate = getattr(result, key, None) if isinstance(candidate, list): result = candidate break if not isinstance(result, list): return [] normalized: list[dict[str, Any]] = [] for item in result: values_are_milliseconds = False if isinstance(item, dict): values_are_milliseconds = "start_time" in item or "end_time" in item start = item.get("start", item.get("start_time", item.get("begin"))) end = item.get("end", item.get("end_time", item.get("stop"))) speaker = item.get("speaker", item.get("speaker_id", item.get("label"))) elif isinstance(item, (list, tuple)) and len(item) >= 3: start, end, speaker = item[0], item[1], item[2] else: continue start_value = _coerce_finite_float(start) end_value = _coerce_finite_float(end) if start_value is None or end_value is None: continue # 列表形式的 CAM++ 时间单位为秒；明确命名为 start_time/end_time 的字段 # 字段按毫秒处理，避免用“超过多少数值”猜单位导致长录音误判。 if values_are_milliseconds: start_value /= 1000 end_value /= 1000 if end_value > start_value: normalized.append({"start_time": round(start_value * 1000, 1), "end_time": round(end_value * 1000, 1), "speaker": str(speaker)}) return normalized async def diarization_handler(request: web.Request) -> web.Response: """接收完整 WAV，返回 CAM++ 说话人聚类时间段。""" runtime: AuxiliaryRuntime = request.app[MODEL_SERVICE_KEY] form = await request.post() upload = form.get("file") if not isinstance(upload, FileField): return web.json_response({"error": "multipart field 'file' is required"}, status=400) temp_path: str | None = None try: with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file: temp_file.write(upload.file.read()) temp_path = temp_file.name result = await runtime.diarization(temp_path) return web.json_response({"segments": _normalize_diarization_segments(result)}) finally: if temp_path: Path(temp_path).unlink(missing_ok=True) async def speaker_resolve_handler(request: web.Request) -> web.Response: """接收一个实时 turn，提取声纹并返回当前会话的在线聚类标签。""" runtime: AuxiliaryRuntime = request.app[MODEL_SERVICE_KEY] form = await request.post() upload = form.get("file") if not isinstance(upload, FileField): return web.json_response({"error": "multipart field 'file' is required"}, status=400) session_id = str(form.get("session_id") or "").strip() if not session_id: return web.json_response({"error": "multipart field 'session_id' is required"}, status=400) try: # aiohttp 的 MultiDictProxy 值可能是 str、bytes 或 FileField，先转换为明确类型 # 为有限浮点数，避免静态检查告警和异常类型值进入声纹服务。 start_time_ms = _parse_form_float(form.get("start_time_ms"), "start_time_ms", default=0.0) end_time_ms = _parse_form_float(form.get("end_time_ms"), "end_time_ms", default=start_time_ms) except ValueError: return web.json_response({"error": "turn time fields must be numbers"}, status=400) temp_path: str | None = None try: with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file: temp_file.write(upload.file.read()) temp_path = temp_file.name result = await runtime.resolve_speaker( temp_path, session_id, start_time_ms, end_time_ms, ) return web.json_response(result or { "speaker_id": -1, "speaker_evidence": "pending", "speaker_confidence": 0.0, "speaker_status": "no_embedding", "speaker_reason": "当前片段未生成可用声纹", }) except Exception as exc: # 将模型推理异常返回给 WebSocket 客户端，避免客户端只能看到笼统的 500 错误。 print( f"[speaker] resolve failed: session_id={session_id}, " f"start={start_time_ms}, end={end_time_ms}, error={exc}", flush=True, ) return web.json_response({"error": str(exc)}, status=500) finally: if temp_path: Path(temp_path).unlink(missing_ok=True) async def speaker_track_handler(request: web.Request) -> web.Response: """接收一个 FunASR turn，返回经过短段平滑的 CAM++ 滑窗标签。""" runtime: AuxiliaryRuntime = request.app[MODEL_SERVICE_KEY] form = await request.post() upload = form.get("file") if not isinstance(upload, FileField): return web.json_response({"error": "multipart field 'file' is required"}, status=400) session_id = str(form.get("session_id") or "").strip() if not session_id: return web.json_response({"error": "multipart field 'session_id' is required"}, status=400) try: start_time_ms = _parse_form_float(form.get("start_time_ms"), "start_time_ms", default=0.0) end_time_ms = _parse_form_float(form.get("end_time_ms"), "end_time_ms", default=start_time_ms) except ValueError: return web.json_response({"error": "turn time fields must be numbers"}, status=400) temp_path: str | None = None try: with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file: temp_file.write(upload.file.read()) temp_path = temp_file.name segments = await runtime.track_speakers( temp_path, session_id, start_time_ms, end_time_ms ) return web.json_response({"segments": segments}) except Exception as exc: print( f"[speaker] track failed: session_id={session_id}, " f"start={start_time_ms}, end={end_time_ms}, error={exc}", flush=True, ) return web.json_response({"error": str(exc)}, status=500) finally: if temp_path: Path(temp_path).unlink(missing_ok=True) async def speaker_reset_handler(request: web.Request) -> web.Response: """释放已经结束的实时会话聚类中心。""" runtime: AuxiliaryRuntime = request.app[MODEL_SERVICE_KEY] payload = await request.json() session_id = str(payload.get("session_id") or "").strip() if isinstance(payload, dict) else "" if session_id: runtime.reset_speaker_session(session_id) return web.json_response({"ok": True}) async def create_app() -> web.Application: """创建辅助 HTTP 服务，并在服务启动前完成模型预加载。""" runtime = AuxiliaryRuntime() runtime.preload() app = web.Application(client_max_size=64 * 1024 * 1024) app[MODEL_SERVICE_KEY] = runtime app.router.add_get("/health", health_handler) app.router.add_post("/v1/vad", vad_handler) app.router.add_post("/v1/punctuation", punctuation_handler) app.router.add_post("/v1/diarization", diarization_handler) app.router.add_post("/v1/speaker/resolve", speaker_resolve_handler) app.router.add_post("/v1/speaker/track", speaker_track_handler) app.router.add_post("/v1/speaker/reset", speaker_reset_handler) return app def main() -> None: """启动宿主机上的常驻辅助模型服务。""" print(f"Auxiliary model service: http://127.0.0.1:{AUXILIARY_PORT}", flush=True) print(f"Device: {AUXILIARY_DEVICE}", flush=True) web.run_app(create_app(), host=AUXILIARY_HOST, port=AUXILIARY_PORT) if __name__ == "__main__": main()