"""将未修改的腾讯演示协议转换为 FunASR 原生在线 WebSocket 协议。""" from __future__ import annotations import argparse import asyncio import json import logging import os from dataclasses import dataclass from pathlib import Path from typing import Any from uuid import uuid4 from aiohttp import WSMsgType, web from dotenv import load_dotenv try: from websockets.asyncio.client import connect as websocket_connect except ImportError: # websockets 13 之前的版本在包根目录提供相同客户端。 from websockets import connect as websocket_connect try: from .auxiliary_service import AuxiliaryModelService, AuxiliaryServiceConfig except ImportError: from auxiliary_service import AuxiliaryModelService, AuxiliaryServiceConfig PROJECT_ROOT = Path(__file__).resolve().parents[2] load_dotenv(PROJECT_ROOT / ".env") LOGGER = logging.getLogger(__name__) WEB_HOST = os.getenv("WEB_HOST", "0.0.0.0") WEB_PORT = int(os.getenv("WEB_PORT", "8082")) NATIVE_WS_URL = os.getenv("FUNASR_NATIVE_WS_URL", "ws://127.0.0.1:10095") CHUNK_SIZE = tuple( int(part.strip()) for part in os.getenv("FUNASR_CHUNK_SIZE", "0,10,5").split(",") ) if len(CHUNK_SIZE) != 3: CHUNK_SIZE = (0, 10, 5) CHUNK_INTERVAL = max(1, int(os.getenv("FUNASR_CHUNK_INTERVAL", "10"))) SAMPLE_RATE = 16000 PCM_BYTES_PER_MS = SAMPLE_RATE * 2 / 1000 FRAME_BYTES = max(2, round(60 * CHUNK_SIZE[1] / CHUNK_INTERVAL * PCM_BYTES_PER_MS)) MAX_SPEAKER_AUDIO_BYTES = 60 * SAMPLE_RATE * 2 MIN_SPEAKER_AUDIO_BYTES = int(0.8 * SAMPLE_RATE * 2) FINALIZE_TIMEOUT_SECONDS = max(30, int(os.getenv("FUNASR_FINALIZE_TIMEOUT_SECONDS", "300"))) TURN_SENTENCE_ID_STRIDE = 100 AUXILIARY_KEY = web.AppKey("auxiliary_service", AuxiliaryModelService) SESSION_REGISTRY_KEY = web.AppKey("sessions", dict) class IncrementalWavDecoder: """读取流式 PCM WAV 文件头，并逐段返回 16 kHz 单声道 PCM 负载。""" def __init__(self) -> None: self.buffer = bytearray() self.header_done = False self.format_valid = False self.data_remaining: int | None = None def feed(self, data: bytes) -> bytes: if self.header_done: if self.data_remaining is None: return data payload = data[: self.data_remaining] self.data_remaining -= len(payload) return payload self.buffer.extend(data) if len(self.buffer) < 12: return b"" if self.buffer[:4] != b"RIFF" or self.buffer[8:12] != b"WAVE": raise ValueError("WAV file must use a RIFF/WAVE container") del self.buffer[:12] while len(self.buffer) >= 8: kind = bytes(self.buffer[:4]) size = int.from_bytes(self.buffer[4:8], "little") if size > 1024 * 1024: raise ValueError("WAV header chunk is unexpectedly large") full_size = 8 + size + (size % 2) if len(self.buffer) < full_size: return b"" body = bytes(self.buffer[8 : 8 + size]) if kind == b"fmt ": if size < 16: raise ValueError("WAV fmt chunk is incomplete") fmt = ( int.from_bytes(body[0:2], "little"), int.from_bytes(body[2:4], "little"), int.from_bytes(body[4:8], "little"), int.from_bytes(body[14:16], "little"), ) if fmt != (1, 1, SAMPLE_RATE, 16): raise ValueError("WAV must be PCM16, mono, 16 kHz") self.format_valid = True if kind == b"data": if not self.format_valid or size % 2: raise ValueError("WAV must be PCM16, mono, 16 kHz") self.header_done = True self.data_remaining = size del self.buffer[:8] payload = bytes(self.buffer[:size]) del self.buffer[: min(size, len(self.buffer))] self.data_remaining -= len(payload) return payload del self.buffer[:full_size] return b"" def finish(self) -> None: if not self.header_done or self.data_remaining not in (None, 0): raise ValueError("WAV ended before its complete PCM data chunk arrived") @dataclass(frozen=True) class SpeakerJob: sentence_id: int text: str audio: bytes start_time_ms: float end_time_ms: float def split_text_by_speaker_segments( text: str, segments: list[dict[str, Any]], turn_start_ms: float, turn_end_ms: float, ) -> list[dict[str, Any]]: """按 CAM++ 时间段近似拆分 ASR 文本，并保留每段稳定说话人标签。""" usable: list[dict[str, Any]] = [] for segment in sorted( segments, key=lambda item: float(item.get("start_time", turn_start_ms)), ): try: start = max(turn_start_ms, float(segment.get("start_time", turn_start_ms))) end = min(turn_end_ms, float(segment.get("end_time", turn_end_ms))) speaker_id = int(segment.get("speaker_id", -1)) except (TypeError, ValueError): continue if end <= start or speaker_id < 0: continue speaker = { "speaker_id": speaker_id, "speaker_name": str(segment.get("speaker_name") or f"说话人 {speaker_id + 1}"), "speaker_confidence": float(segment.get("speaker_confidence") or 0), "speaker_status": str(segment.get("speaker_status") or "confirmed"), } if usable and usable[-1]["speaker"]["speaker_id"] == speaker_id: usable[-1]["end_time_ms"] = end else: usable.append( {"start_time_ms": start, "end_time_ms": end, "speaker": speaker} ) if not usable: return [] if len(usable) == 1: usable[0]["text"] = text return usable # 双重保护：服务端有最短段长滤波，桥接层也拒绝意外的短片段响应。 min_split_ms = max(1500, int(os.getenv("FUNASR_SPEAKER_MIN_SEGMENT_MS", "3000"))) while len(usable) > 1: short_index = next( ( index for index, item in enumerate(usable) if item["end_time_ms"] - item["start_time_ms"] < min_split_ms ), None, ) if short_index is None: break if short_index == 0: usable[1]["start_time_ms"] = usable[0]["start_time_ms"] del usable[0] elif short_index == len(usable) - 1: usable[-2]["end_time_ms"] = usable[-1]["end_time_ms"] del usable[-1] else: previous = usable[short_index - 1] following = usable[short_index + 1] if previous["end_time_ms"] - previous["start_time_ms"] >= ( following["end_time_ms"] - following["start_time_ms"] ): previous["end_time_ms"] = usable[short_index]["end_time_ms"] del usable[short_index] else: following["start_time_ms"] = usable[short_index]["start_time_ms"] del usable[short_index] if len(usable) == 1 or len(text) < len(usable): usable = [max(usable, key=lambda item: item["end_time_ms"] - item["start_time_ms"])] usable[0]["text"] = text return usable total_duration = sum(item["end_time_ms"] - item["start_time_ms"] for item in usable) char_start = 0 elapsed = 0.0 for index, item in enumerate(usable): elapsed += item["end_time_ms"] - item["start_time_ms"] if index == len(usable) - 1: char_end = len(text) else: char_end = round(len(text) * elapsed / total_duration) char_end = max(char_start + 1, min(char_end, len(text) - (len(usable) - index - 1))) item["text"] = text[char_start:char_end].strip() char_start = char_end return [item for item in usable if item.get("text")] class BrowserSession: """管理浏览器与原生 WebSocket 连接，并转换双方的消息格式。""" def __init__( self, browser_ws: web.WebSocketResponse, auxiliary: AuxiliaryModelService, start: dict[str, Any], voice_id: str, ) -> None: self.browser_ws = browser_ws self.auxiliary = auxiliary self.start = start self.voice_id = voice_id self.session_id = uuid4().hex self.stop_event = asyncio.Event() self.send_lock = asyncio.Lock() self.speaker_jobs: asyncio.Queue[SpeakerJob | None] = asyncio.Queue() self.wav_decoder = ( IncrementalWavDecoder() if str(start.get("source") or "mic") == "file" and Path(str(start.get("file_name") or "")).suffix.lower() == ".wav" else None ) self.speed_factor = max(0.5, min(3.0, float(start.get("speed_factor") or 1.0))) try: sentence_strategy = int(start.get("sentence_strategy", 0)) except (TypeError, ValueError): sentence_strategy = 0 self.sentence_strategy = sentence_strategy if sentence_strategy in (0, 1) else 0 self.pending_pcm = bytearray() self.turn_audio = bytearray() self.total_audio_ms = 0.0 self.turn_start_ms = 0.0 self.turn_text = "" self.sentence_id = 0 self.native_ack: dict[str, Any] | None = None self.native_error: str | None = None async def emit(self, payload: dict[str, Any]) -> None: """ASR 和 CAM++ 的完成时机不同，因此需要串行写入浏览器连接。""" async with self.send_lock: if not self.browser_ws.closed: await self.browser_ws.send_json(payload) async def emit_sentence( self, text: str, final: bool, speaker: dict[str, Any] | None = None, sentence_id: int | None = None, start_time_ms: float | None = None, end_time_ms: float | None = None, ) -> None: speaker = speaker or {} sentence = { "sentence_id": self.sentence_id if sentence_id is None else sentence_id, "sentence": text, "sentence_type": 1 if final else 0, "start_time": round(self.turn_start_ms if start_time_ms is None else start_time_ms), "end_time": round(self.total_audio_ms if end_time_ms is None else end_time_ms), # 未修改的腾讯界面使用 speaker_id 选择对应的说话人气泡。 "speaker_id": int(speaker.get("speaker_id", -1)), "speaker_name": str(speaker.get("speaker_name") or ""), "speaker_confidence": float(speaker.get("speaker_confidence") or 0), "speaker_status": str(speaker.get("speaker_status") or "pending"), } await self.emit({"type": "sentences", "sentences": [sentence]}) def align_turn_audio_to_vad(self, message: dict[str, Any]) -> None: """用原生 VAD 的绝对时间裁掉当前 turn 的前后静音样本。""" buffer_end_ms = self.turn_start_ms + len(self.turn_audio) / PCM_BYTES_PER_MS try: speech_start_ms = message.get("speech_start_ms") if speech_start_ms is not None: target = max(self.turn_start_ms, min(buffer_end_ms, float(speech_start_ms))) trim_bytes = int(round((target - self.turn_start_ms) * PCM_BYTES_PER_MS)) trim_bytes -= trim_bytes % 2 del self.turn_audio[:trim_bytes] self.turn_start_ms += trim_bytes / PCM_BYTES_PER_MS buffer_end_ms = self.turn_start_ms + len(self.turn_audio) / PCM_BYTES_PER_MS speech_end_ms = message.get("speech_end_ms") if speech_end_ms is not None: target = max(self.turn_start_ms, min(buffer_end_ms, float(speech_end_ms))) keep_bytes = int(round((target - self.turn_start_ms) * PCM_BYTES_PER_MS)) keep_bytes -= keep_bytes % 2 del self.turn_audio[keep_bytes:] except (TypeError, ValueError): LOGGER.warning("Ignoring invalid native FunASR VAD boundary: %s", message) async def send_pcm_frame(self, native_ws: Any, frame: bytes, pace_file: bool) -> None: if not frame: return if len(frame) % 2: raise ValueError("PCM16 audio ended on an incomplete sample") self.total_audio_ms += len(frame) / PCM_BYTES_PER_MS self.turn_audio.extend(frame) if len(self.turn_audio) > MAX_SPEAKER_AUDIO_BYTES: # 即使 VAD 始终没有报告端点，也要限制每轮音频的内存占用。 trim = len(self.turn_audio) - MAX_SPEAKER_AUDIO_BYTES del self.turn_audio[:trim] self.turn_start_ms += trim / PCM_BYTES_PER_MS await native_ws.send(frame) if pace_file: await asyncio.sleep(len(frame) / (SAMPLE_RATE * 2) / self.speed_factor) async def accept_audio(self, native_ws: Any, data: bytes) -> None: pcm = self.wav_decoder.feed(data) if self.wav_decoder else data if not pcm: return self.pending_pcm.extend(pcm) is_file = str(self.start.get("source") or "mic") == "file" while len(self.pending_pcm) >= FRAME_BYTES: frame = bytes(self.pending_pcm[:FRAME_BYTES]) del self.pending_pcm[:FRAME_BYTES] await self.send_pcm_frame(native_ws, frame, is_file) async def finish_audio(self, native_ws: Any) -> None: if self.wav_decoder: self.wav_decoder.finish() if self.pending_pcm: frame = bytes(self.pending_pcm) self.pending_pcm.clear() await self.send_pcm_frame( native_ws, frame, str(self.start.get("source") or "mic") == "file" ) async def read_native(self, native_ws: Any) -> None: """接收 FunASR 原生事件，并维护每段话语的中间结果缓存。""" try: while True: raw = await native_ws.recv() message = json.loads(raw) if message.get("is_end"): self.native_ack = message return if message.get("event") == "vad": self.align_turn_audio_to_vad(message) continue text = str(message.get("text") or "") if text: # FunASR 在线模式会为每个分块发送新解码出的文本。 self.turn_text += text if message.get("is_final"): final_text = self.turn_text.strip() audio = bytes(self.turn_audio) start_ms = self.turn_start_ms end_ms = start_ms + len(audio) / PCM_BYTES_PER_MS turn_sentence_id = self.sentence_id if final_text: # 先结束前端中间结果气泡；标点和 CAM++ 在独立工作线程中完成， # 不阻塞原生 WebSocket 继续读取后续音频帧。 await self.emit_sentence( final_text, final=True, sentence_id=turn_sentence_id, start_time_ms=start_ms, end_time_ms=end_ms, ) await self.speaker_jobs.put( SpeakerJob( sentence_id=turn_sentence_id, text=final_text, audio=audio, start_time_ms=start_ms, end_time_ms=end_ms, ) ) # 为同一个 VAD 轮次内可能拆出的多个气泡预留独立 ID。 self.sentence_id += TURN_SENTENCE_ID_STRIDE self.turn_text = "" self.turn_audio.clear() self.turn_start_ms = self.total_audio_ms elif self.turn_text: await self.emit_sentence(self.turn_text, final=False) except asyncio.CancelledError: raise except Exception as exc: self.native_error = str(exc) LOGGER.exception("native FunASR WebSocket closed unexpectedly") await self.emit({"type": "error", "message": f"FunASR realtime WS: {exc}"}) async def resolve_speakers(self) -> None: """按序定稿标点并用 CAM++ 滑窗恢复 turn 内说话人切换。""" while True: job = await self.speaker_jobs.get() try: if job is None: return final_text = job.text try: punctuation = await self.auxiliary.punctuate(final_text) if not punctuation.get("available", False): LOGGER.warning( "FunASR punctuation is unavailable: %s", punctuation.get("error", "model is not configured"), ) else: punctuated = str(punctuation.get("text") or "").strip() if punctuated: final_text = punctuated except Exception: LOGGER.exception("FunASR punctuation request failed; keeping raw text") subsegments: list[dict[str, Any]] = [] if len(job.audio) >= MIN_SPEAKER_AUDIO_BYTES: # 必需的 CAM++ 按 FunASR 的 1.5 秒窗口和 0.75 秒步长， # 在同一个 VAD 轮次内识别多个说话人片段。 track = getattr(self.auxiliary, "track_speakers", None) if callable(track): try: tracked = await track( job.audio, self.session_id, job.start_time_ms, job.end_time_ms, ) subsegments = split_text_by_speaker_segments( final_text, tracked, job.start_time_ms, job.end_time_ms, ) except Exception: LOGGER.exception( "CAM++ sliding-window diarization failed; falling back to whole-turn speaker" ) if not subsegments and len(job.audio) >= MIN_SPEAKER_AUDIO_BYTES: try: resolved = await self.auxiliary.resolve_speaker( job.audio, self.session_id, job.start_time_ms, job.end_time_ms, ) if resolved and int(resolved.get("speaker_id", -1)) >= 0: subsegments = [ { "text": final_text, "speaker": resolved, "start_time_ms": job.start_time_ms, "end_time_ms": job.end_time_ms, } ] except Exception: LOGGER.exception("CAM++ speaker resolution failed: voice_id=%s", self.voice_id) if not subsegments: subsegments = [ { "text": final_text, "speaker": {"speaker_id": -1}, "start_time_ms": job.start_time_ms, "end_time_ms": job.end_time_ms, } ] for index, segment in enumerate(subsegments): await self.emit_sentence( str(segment["text"]), final=True, speaker=segment.get("speaker"), sentence_id=job.sentence_id + index, start_time_ms=float(segment.get("start_time_ms", job.start_time_ms)), end_time_ms=float(segment.get("end_time_ms", job.end_time_ms)), ) finally: self.speaker_jobs.task_done() async def config_handler(_: web.Request) -> web.Response: """提供简洁的就绪状态响应，供启动器和诊断使用。""" return web.json_response( { "engine": "funasr-native-online-ws", "model": os.getenv("FUNASR_ASR_MODEL", ""), "native_ws_url": NATIVE_WS_URL, "speaker_service_url": os.getenv( "AUXILIARY_SERVICE_URL", "http://127.0.0.1:8010" ), } ) async def stop_handler(request: web.Request) -> web.Response: voice_id = request.query.get("voice_id", "").strip() if not voice_id: return web.json_response({"ok": False, "error": "missing voice_id"}, status=400) stop_event = request.app[SESSION_REGISTRY_KEY].get(voice_id) if stop_event is None: return web.json_response({"ok": False, "error": "session not found"}) stop_event.set() return web.json_response({"ok": True}) async def websocket_handler(request: web.Request) -> web.WebSocketResponse: """将腾讯前端消息桥接到 FunASR 原生实时协议。""" browser_ws = web.WebSocketResponse(max_msg_size=64 * 1024 * 1024, heartbeat=30) await browser_ws.prepare(request) session: BrowserSession | None = None native_reader: asyncio.Task[None] | None = None speaker_worker: asyncio.Task[None] | None = None voice_id = "" registered = False try: first = await browser_ws.receive() if first.type != WSMsgType.TEXT: await browser_ws.send_json({"type": "error", "message": "first message must be JSON start"}) return browser_ws start = json.loads(first.data) if not isinstance(start, dict) or start.get("type") != "start": await browser_ws.send_json({"type": "error", "message": "first message must have type=start"}) return browser_ws source = str(start.get("source") or "mic") suffix = Path(str(start.get("file_name") or "")).suffix.lower() if source == "file" and suffix not in {".pcm", ".wav"}: await browser_ws.send_json( { "type": "error", "message": "FunASR realtime mode accepts PCM or 16 kHz mono PCM WAV files.", } ) return browser_ws if source == "file" and suffix == ".pcm": start = dict(start) start["file_name"] = str(start.get("file_name") or "audio.pcm") voice_id = uuid4().hex auxiliary: AuxiliaryModelService = request.app[AUXILIARY_KEY] session = BrowserSession(browser_ws, auxiliary, start, voice_id) request.app[SESSION_REGISTRY_KEY][voice_id] = session.stop_event registered = True await session.emit({"type": "voice_id", "voice_id": voice_id}) await session.emit({"type": "start"}) # 这是 FunASR 原生 WSS 消息格式；之后每 60 毫秒发送一帧 PCM。 async with websocket_connect( NATIVE_WS_URL, subprotocols=["binary"], ping_interval=None, close_timeout=3, max_size=None, ) as native_ws: await native_ws.send( json.dumps( { "mode": "online", "chunk_size": list(CHUNK_SIZE), "chunk_interval": CHUNK_INTERVAL, "encoder_chunk_look_back": int( os.getenv("FUNASR_ENCODER_LOOK_BACK", "4") ), "decoder_chunk_look_back": int( os.getenv("FUNASR_DECODER_LOOK_BACK", "1") ), "sentence_strategy": session.sentence_strategy, "audio_fs": SAMPLE_RATE, "wav_name": voice_id, "is_speaking": True, }, ensure_ascii=False, ) ) native_reader = asyncio.create_task(session.read_native(native_ws)) speaker_worker = asyncio.create_task(session.resolve_speakers()) while not browser_ws.closed: receive_task = asyncio.create_task(browser_ws.receive()) stop_task = asyncio.create_task(session.stop_event.wait()) done, _ = await asyncio.wait( [receive_task, stop_task, native_reader], return_when=asyncio.FIRST_COMPLETED, ) if stop_task in done: receive_task.cancel() await asyncio.gather(receive_task, return_exceptions=True) break stop_task.cancel() await asyncio.gather(stop_task, return_exceptions=True) if native_reader in done: receive_task.cancel() await asyncio.gather(receive_task, return_exceptions=True) if session.native_ack is None and session.native_error is None: session.native_error = "FunASR native WebSocket ended before EOF acknowledgement" break message = await receive_task if message.type == WSMsgType.BINARY: await session.accept_audio(native_ws, bytes(message.data)) continue if message.type == WSMsgType.TEXT: try: control = json.loads(message.data) except json.JSONDecodeError: continue if isinstance(control, dict) and control.get("type") in {"eof", "stop"}: break if isinstance(control, dict) and control.get("type") == "abort": session.native_error = "session aborted by browser" break if message.type in {WSMsgType.CLOSE, WSMsgType.CLOSED, WSMsgType.ERROR}: break if session.native_error is None and not browser_ws.closed: await session.finish_audio(native_ws) # FunASR 会刷新在线缓存，并在输出最终结果后才返回确认。 await native_ws.send( json.dumps({"is_speaking": False, "is_end": True}, ensure_ascii=False) ) try: await asyncio.wait_for(native_reader, timeout=FINALIZE_TIMEOUT_SECONDS) except asyncio.TimeoutError: session.native_error = ( f"FunASR did not acknowledge end-of-input within " f"{FINALIZE_TIMEOUT_SECONDS}s" ) if speaker_worker is not None: await session.speaker_jobs.join() await session.speaker_jobs.put(None) await speaker_worker speaker_worker = None if session.native_error: await session.emit({"type": "error", "message": session.native_error}) elif session.native_ack and not session.native_ack.get("is_final", False): await session.emit( { "type": "error", "message": str(session.native_ack.get("error") or "FunASR did not finalize the stream"), } ) elif session.native_ack: await session.emit({"type": "end"}) except asyncio.CancelledError: raise except Exception as exc: LOGGER.exception("Tencent-compatible WebSocket session failed") if not browser_ws.closed: await browser_ws.send_json({"type": "error", "message": str(exc)}) finally: if native_reader is not None and not native_reader.done(): native_reader.cancel() await asyncio.gather(native_reader, return_exceptions=True) if speaker_worker is not None and not speaker_worker.done(): speaker_worker.cancel() await asyncio.gather(speaker_worker, return_exceptions=True) if registered: request.app[SESSION_REGISTRY_KEY].pop(voice_id, None) if session is not None: await session.auxiliary.reset_speaker_session(session.session_id) if not browser_ws.closed: await browser_ws.close() return browser_ws async def create_app() -> web.Application: """创建轻量协议桥接层；模型推理由 FunASR 原生服务负责。""" app = web.Application() app[AUXILIARY_KEY] = AuxiliaryModelService( AuxiliaryServiceConfig( base_url=os.getenv("AUXILIARY_SERVICE_URL", "http://127.0.0.1:8010") ) ) app[SESSION_REGISTRY_KEY] = {} async def lifecycle(application: web.Application): await application[AUXILIARY_KEY].start() try: health = await application[AUXILIARY_KEY].health() if not health.get("speaker_embedding_ready"): raise RuntimeError("CAM++ speaker service is not ready") except Exception: await application[AUXILIARY_KEY].close() raise yield await application[AUXILIARY_KEY].close() app.cleanup_ctx.append(lifecycle) app.router.add_get("/api/config", config_handler) app.router.add_get("/api/stop", stop_handler) app.router.add_get("/ws", websocket_handler) return app def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--no-browser", action="store_true") args = parser.parse_args() logging.basicConfig(level=logging.INFO) print( f"FunASR browser bridge: http://{os.getenv('WEB_DISPLAY_HOST', '127.0.0.1')}:{WEB_PORT}/api/config", flush=True, ) web.run_app(create_app(), host=WEB_HOST, port=WEB_PORT) if __name__ == "__main__": main()