"""火山引擎(豆包)大模型流式语音识别 ASR —— V3 WebSocket 二进制协议。 文档:wss://openspeech.bytedance.com/api/v3/sauc/bigmodel(用户提供)。 对外:`recognize(pcm_bytes, rate=16000) -> text`(同步,内部 asyncio)。输入须 16k/16bit/单声道 PCM。 鉴权(旧版):X-Api-App-Key(=appid)+ X-Api-Access-Key + X-Api-Resource-Id(volc.bigasr.sauc.duration)。 帧:4B header [+seq int32(大端)] + payload_size(uint32) + payload(gzip);full-client=JSON,audio=raw。 最后一包音频用 flags=0b0011(负 seq)标记;响应里 flags 含 0b0010 即最终包。 """ from __future__ import annotations import asyncio import gzip import json import os import struct import uuid from pathlib import Path from typing import Optional from genesis.obs.logging_setup import get_logger logger = get_logger("voice.asr") WS_URL = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel" MT_FULL_CLIENT, MT_AUDIO, MT_FULL_SERVER, MT_ERROR = 0b0001, 0b0010, 0b1001, 0b1111 SER_JSON, SER_RAW = 0b0001, 0b0000 GZIP = 0b0001 FLAG_POS_SEQ = 0b0001 # 带正 seq FLAG_LAST_SEQ = 0b0011 # 带负 seq + 最后一包 def _creds() -> tuple[str, str]: appid, token = os.getenv("VOLC_APP_ID"), os.getenv("VOLC_ACCESS_TOKEN") if not (appid and token): envf = Path(__file__).resolve().parents[3] / ".env" if envf.exists(): for ln in envf.read_text(encoding="utf-8").splitlines(): ln = ln.strip() if ln.startswith("VOLC_APP_ID="): appid = appid or ln.split("=", 1)[1] elif ln.startswith("VOLC_ACCESS_TOKEN="): token = token or ln.split("=", 1)[1] if not (appid and token): raise RuntimeError("缺少 VOLC_APP_ID / VOLC_ACCESS_TOKEN(请配 .env)") return appid, token def _hdr(msg_type: int, flags: int, serial: int, compress: int = GZIP) -> bytes: return bytes([(0b0001 << 4) | 0b0001, (msg_type << 4) | flags, (serial << 4) | compress, 0]) def _full_client(params: dict) -> bytes: payload = gzip.compress(json.dumps(params, ensure_ascii=False).encode("utf-8")) return _hdr(MT_FULL_CLIENT, FLAG_POS_SEQ, SER_JSON) + struct.pack(">i", 1) \ + struct.pack(">I", len(payload)) + payload def _audio(chunk: bytes, seq: int, last: bool) -> bytes: flags = FLAG_LAST_SEQ if last else FLAG_POS_SEQ s = -seq if last else seq payload = gzip.compress(chunk) return _hdr(MT_AUDIO, flags, SER_RAW) + struct.pack(">i", s) \ + struct.pack(">I", len(payload)) + payload def _parse(data: bytes) -> dict: msg_type = (data[1] >> 4) & 0x0F flags = data[1] & 0x0F compress = data[2] & 0x0F off = 4 if msg_type == MT_ERROR: code = struct.unpack(">I", data[off:off + 4])[0]; off += 4 size = struct.unpack(">I", data[off:off + 4])[0]; off += 4 return {"error": code, "msg": data[off:off + size].decode("utf-8", "ignore")} if flags & FLAG_POS_SEQ: # 带 seq off += 4 size = struct.unpack(">I", data[off:off + 4])[0]; off += 4 payload = data[off:off + size] if compress == GZIP and payload: payload = gzip.decompress(payload) obj = json.loads(payload.decode("utf-8")) if payload else {} return {"obj": obj, "last": bool(flags & 0b0010)} async def _connect(headers): import websockets try: return await websockets.connect(WS_URL, additional_headers=headers, max_size=None) except TypeError: return await websockets.connect(WS_URL, extra_headers=headers, max_size=None) async def _recognize(pcm: bytes, rate: int, appid: str, token: str, resource_id: str) -> str: headers = {"X-Api-App-Key": appid, "X-Api-Access-Key": token, "X-Api-Resource-Id": resource_id, "X-Api-Request-Id": str(uuid.uuid4()), "X-Api-Connect-Id": str(uuid.uuid4())} params = {"user": {"uid": "genesis"}, "audio": {"format": "pcm", "rate": rate, "bits": 16, "channel": 1}, "request": {"model_name": "bigmodel", "enable_itn": True, "enable_punc": True}} chunk = max(1, int(rate * 2 * 0.2)) # 200ms 的 16bit 单声道 chunks = [pcm[i:i + chunk] for i in range(0, len(pcm), chunk)] or [b""] result = {"text": ""} ws = await _connect(headers) async def sender(): await ws.send(_full_client(params)) seq = 1 for idx, ch in enumerate(chunks): seq += 1 await ws.send(_audio(ch, seq, idx == len(chunks) - 1)) await asyncio.sleep(0.12) # 文档建议 100~200ms 发包间隔 async def receiver(): while True: r = _parse(await ws.recv()) if "error" in r: raise RuntimeError(f"ASR 错误 {r['error']}: {r['msg'][:160]}") t = (r["obj"].get("result") or {}).get("text") if t: result["text"] = t if r["last"]: break try: await asyncio.gather(sender(), receiver()) finally: await ws.close() return result["text"] def recognize(pcm: bytes, rate: int = 16000, resource_id: Optional[str] = None) -> str: """16k/16bit/单声道 PCM → 识别文本(同步)。resource_id 默认 VOLC_ASR_RESOURCE_ID 或 volc.bigasr.sauc.duration。""" appid, token = _creds() rid = resource_id or os.getenv("VOLC_ASR_RESOURCE_ID") or "volc.bigasr.sauc.duration" return asyncio.run(_recognize(pcm, rate, appid, token, rid)) async def _safe_send(ws, data) -> None: try: await ws.send(data) except Exception: pass async def relay(client_ws, rate: int = 16000, resource_id: Optional[str] = None) -> str: """全流式桥:浏览器 WS 边说边送 PCM,火山的中间识别结果实时回传浏览器(真·边说边出字)。 client_ws:服务端 websockets 连接 —— 二进制帧=一段 16k/16bit 单声道 PCM;文本帧(如 {"event":"stop"})=收尾。 每收到火山一版累计文本就回传 {"text": "...", "last": bool};返回最终文本。 """ appid, token = _creds() rid = resource_id or os.getenv("VOLC_ASR_RESOURCE_ID") or "volc.bigasr.sauc.duration" headers = {"X-Api-App-Key": appid, "X-Api-Access-Key": token, "X-Api-Resource-Id": rid, "X-Api-Request-Id": str(uuid.uuid4()), "X-Api-Connect-Id": str(uuid.uuid4())} params = {"user": {"uid": "genesis"}, "audio": {"format": "pcm", "rate": rate, "bits": 16, "channel": 1}, "request": {"model_name": "bigmodel", "enable_itn": True, "enable_punc": True}} volc = await _connect(headers) final = {"text": ""} seq = 1 try: await volc.send(_full_client(params)) async def to_volc(): nonlocal seq try: async for msg in client_ws: # 浏览器边采边送 if isinstance(msg, (bytes, bytearray)): seq += 1 await volc.send(_audio(bytes(msg), seq, last=False)) else: break # 文本控制帧 = stop except Exception: pass seq += 1 try: await volc.send(_audio(b"", seq, last=True)) # 收尾空帧(带 last 标志) except Exception: pass async def from_volc(): try: while True: r = _parse(await volc.recv()) if "error" in r: await _safe_send(client_ws, json.dumps({"error": r["msg"]})); break t = (r["obj"].get("result") or {}).get("text") if t: final["text"] = t await _safe_send(client_ws, json.dumps({"text": t or "", "last": r["last"]})) if r["last"]: break except Exception: pass await asyncio.gather(to_volc(), from_volc()) finally: try: await volc.close() except Exception: pass return final["text"]