原理与复刻指南

0. 一分钟看懂

豆包输入法打字时用的「语音输入」，本质是客户端把麦克风音频发到字节的流式识别服务。它不需要登录、也没有对外公开 API，但它的凭据是按设备匿名下发的，所以我们可以自己复刻整条链路：

1. 豆包输入法是怎么被借用的

1.1 下载地址与安装包

官网 PC 页是单页应用，真实下载地址由一个接口给出（这是唯一官方渠道，别用第三方下载站）：

# 官网页面
https://shurufa.doubao.com/pc

# 下载信息接口（platform 可选 windows / macos / android / ios / harmony）
curl 'https://shurufa.doubao.com/api/v1/app/download_url?platform=windows'
# {"code":0,"data":{"release_date":"2026.10.10",
#   "url":"https://lf-wave.doubaocdn.com/obj/doubao-ime/app/windows/DoubaoIME_Installer_0.9.2.7_release.exe",
#   "version_code":90207,"version_name":"V0.9.2"},"msg":"success"}

Windows 包校验（本次实测）：

SHA256  d777027cc822a67dac04ba502616f6e6216a54c06681e21f8f45ef0a89cec5db
file    PE32 executable (GUI) Intel 80386, for MS Windows  # 32 位安装器

怎么把它解开

安装器是 Inno Setup 6.7.0：innoextract 1.9（官方最新版）和 master 分支（支持到 6.4.0.1）都读不了它。 可行的办法是用 Wine 32 位跑静默安装，直接把文件树取出来：

# 1) 装 32 位 wine 和虚拟显示（安装器需要窗口，无 X 会报 Invalid window handle）
dpkg --add-architecture i386 && apt-get update
apt-get install -y --no-install-recommends wine32:i386 xvfb

# 2) 静默安装到 wine 前缀（安装过程会因 TSF 注册失败回滚目录，但文件已经落盘）
export WINEPREFIX=/root/doubao-ime/wineprefix WINEARCH=win32
xvfb-run -a /usr/lib/wine/wine DoubaoIME_Installer_0.9.2.7_release.exe \
    /VERYSILENT /SUPPRESSMSGBOXES /NORESTART /LOG='C:\inno.log'

# 3) 文件就在（注意必须用 /usr/lib/wine/wine，/usr/local/bin/wine 是 wine64 跑不了 32 位）
ls "$WINEPREFIX/drive_c/Program Files/DoubaoIME/versions/v0.9.2.7/"

1.2 包结构：哪部分负责语音识别

装出来的目录（Program Files\DoubaoIME\versions\v0.9.2.7\，约 267 MB）：

从编译期路径能看出的源码结构

二进制里残留了开发机的源码路径，等于给了目录树：

1.3 识别协议

一条 WebSocket 长连接，二进制 protobuf 帧，四个步骤：

请求字段（实测得到的字段号）

StartSession 的 payload

{
  "audio_info": { "channel": 1, "format": "raw", "sample_rate": 16000 },
  "enable_punctuation": true,
  "enable_speech_rejection": false,
  "extra": {
    "did": "<device_id>",
    "input_mode": "tool",
    "strong_ddc": true,            // 强纠错
    "use_twopass_retry": true,     // 二遍重试
    "enable_asr_twopass": true,    // 流式 + 整段两遍解码
    "enable_asr_threepass": true,  // 三遍
    "disable_user_words": false,   // 允许个人词库生效
    "cell_compress_rate": 8,
    "context": "<base64>"          // 可选：chat + ime_info + hotwords（热词偏置）
  }
}

响应字段

结果 JSON（字段 7 的内容）

{
  "results": [{
    "text": "你好，这是语音识别测试",      // 本次发言的累积文本，最后一条即完整结果
    "start_time": 1.33, "end_time": 3.4,  // 秒，相对本次会话起点
    "confidence": 0.998,
    "is_interim": false,                  // false = 已定稿
    "is_vad_finished": true,              // VAD 判定这一句说完了
    "index": 0,
    "alternatives": [{ "text": "…", "words": [
        { "word": "你", "start_time": 0, "end_time": 0.34 }, …   // 逐字时间
    ]}]
  }]
}

1.4 状态码与错误处理

1.5 语言能力（实测）

原因是这个模型是 zh + en 双语（即官方宣传的"中英混输"），客户端创建识别句柄时写死的是 speech_opus + zh； 协议里没有 language 字段，所以没有"切语种"的开关。要用日语只能换官方多语种接口或在本地跑开源日语模型。

1.6 豆包的离线识别是怎么做的

这是本轮最有价值的发现。结论先给：离线识别 = 字节自研端侧推理引擎 Flute + 联网下载的模型包， 而且 Windows 0.9.2.7 里还没有这套东西（Android / macOS 有）。

模型长什么样

macOS 主程序里写死了模型的张量文件名（93 个 .bin/.dat/json），可以直接读出结构 —— 是 PANTHER Conformer 编码器 + LLM 风格解码器，CIF（连续积分激发）机制，int8 量化：

s/panther_conformer_*                     编码器
s/panther_decoder_L*                      解码器
s/panther_l0_{qkv_pre_rope,mha_raw,ffn1_gemm1,ffn2_swiglu,rmsnorm1,rmsnorm2,conv_out}.bin
up_proj.bin / down_proj.bin / v_proj.bin / swiglu.bin / o_proj ...   权重张量
*_shape.dat / input_shape.dat / config.json / token_config.json / llm_params.json

算力后端是 CPU int8（源码里有 PackBS8 format: 4x4 tiled layout matching CPU prepacked weights） 加 GPU kernel（macOS 用 Metal，源码里带 device const char* 的着色器）。体量对得上"离线模型包约 150 MB"的说法。

在线 / 离线怎么协同（SAMI 里的融合策略）

• PutOnlineCacheAudioToOffline：把在线期间缓存的音频交给整段识别重跑
• [online vs offline] words num: [%d vs %d]：两份结果对比择优
• 触发条件：asr has no network, switch to offline、asr has no result, switch to offline、recv interval time
• 时间参数：finish_wait_offline_time / offline_wait_online_time / max_wait_switch_offline_time
• 应用侧：openASRPanel canceled: network disconnected, offlineModelStatus=、Skip smart text enable prompt: reason=offline_result

1.7 离线模型能从公开渠道拿到（本项目已内置离线引擎）

本以为“只有私有引擎能吃这些模型”，结果发现字节把一批模型放在公开算法分发接口上，无需任何鉴权：

GET https://effect.snssdk.com/model/api/arithmetics?sdk_version=25.0.0
→ {"status_code":0,"data":{"arithmetics":{"cakeCook":[
     {"name":"lv_asr_support","version":"1.2",
      "file_url":{"url_list":["https://n98-lf-effect-aweme-302-ncdn.byteeffecttos.com/obj/ies.fe.effect/<md5>?module=model"],
                   "zip_url_list":["..."]}}]}}}

# 参数必须是小写 sdk_version（写 SdkVersion 会报 missing required parameter），只能 GET（POST 404）
# sdk_version 就是“面板版本”，值越大条目越多：1.0.2→3、15.0.0→875、25.0.0→1128（上限）

这条链路是怎么找到的（可复现）

1. macOS 客户端主程序里有 ASROfflineModelMetaInfoService.swift、ASROfflineModelTTNetRequest （TTNet 是字节的网络库），顺着字符串邻域找到两个 TTNet 服务名 com.wave.offline-model.metainfo-service / com.wave.offline-model.metainfo-head；
2. 同一片邻域里出现域名 https://effect.snssdk.com 与路径 /model/api/arithmetics （还有解析字段 fileDownloadURLs、installRelativePath、componentVersionKey）；
3. 裸调该路径，服务端直接报 params check error: expr_path=SdkVersion, cause=missing required parameter—— 按提示试出参数名必须是小写 sdk_version（写成 SdkVersion 反而一直报缺参数），POST 一律 404（只收 GET）；
4. 把 sdk_version 调大就能看出它是“面板版本”：1.0.2→3 个、15.0.0→875 个、25.0.0→1128 个（上限）。

经验：这类接口的报错比文档好用——服务端报什么字段名，就原样试什么字段名。

同一批物料里有两种封装，能力完全不同

同批还有 lv_asr_cut_mobile、online_mobile_asr_transformer_pipeline(_douyin)、 KWS 唤醒词（kws_chn_hanzi 等）、unet_denoise_44k_speech 降噪等， --list 一共能列出 24 个语音相关模型。挑法很简单：看 zip 里有没有 .onnx—— 拉取脚本会自动判断，遇到私有容器会明确告诉你“这个用不了”。

我们要的这套（lv_asr_support，解开就是完整的 ASR 流水线）：

用 onnxruntime 打开就能读出全部输入输出——这是判断“能不能自己跑”的依据：

us_cif_peak 是 CIF 的峰值序列，同时是“何时吐出一个字”的判据——和 Flute 里 ntp_cif/cif_calculator 是同一套思路。而 strategy.json 漏出了产品侧策略：

{ "dispatcher": { "ratio": 0.33, "max_counts": 5, "quality_score": 0.83 },
  "vad_strategy": { "max_slice_duration": 30, "min_slice_duration": 10 },
  "sentence_config": { "max_time_interval_ms": 300, "split_size": 16 } }

意思是：原始产品把长音频切成 10~30s 片，同一片可能用不同切点跑最多 5 次，再用 classifier 的质量分 （阈值 0.83）挑最好的一份——这也解释了包里为什么会带一个“质量分模型”。本项目简化成 停顿断句 + 单次解码 + 质量分只做展示（存在每句话的 quality 里，导出 JSON 能看到）。

输入 560 = 80 维 mel × 7（LFR m=7,n=6），和 FunASR Paraformer 同构；us_cif_peak 说明是 CIF / ntp_cif（和豆包输入法 Flute 里的模块同源），只是这里是明文 ONNX，onnxruntime 直接能跑。

# 一条命令把模型拉齐（约 310MB，走公开 CDN）
./tools/fetch_offline_asr_models.sh            # 默认取 lv_asr_support
./tools/fetch_offline_asr_models.sh --list     # 先看看有哪些语音模型

前端特征怎么对上（最容易翻车的一步）

PCM16 / 16k / 单声道
  → 80 维 log-mel：25ms 窗、10ms 步长、HTK mel（1127·ln(1+f/700)）、取 log
  → LFR(7,6)：连续 7 帧拼成 1 帧、步长 6 → 560 维（80 × 7 = 560）
  → 逐句归一化：(x − mean) / (std + 1e−10)          ← 逐句 CMVN
  → encoder → decoder → argmax → 8404 词表拼字（跳过 id 0/1/2，去掉 @@ 后缀）

2. 凭据从哪来（为什么不用登录）

豆包输入法没有任何账号体系参与语音识别，身份是「应用身份 + 设备身份」：

3. 自己写客户端

下面三份代码都是完整可运行的单文件客户端：自己注册设备、拿 app_key、跑完一次识别并打印文本。 复制走改一改就能用（asr_min.py / asr_min.js / asr_min.go 也可直接下载）。

Python

pip install websockets
python asr_min.py test.wav        # 输出：你好，这是微信语音接口测试，今天阳光很好。

#!/usr/bin/env python3
"""豆包输入法语音识别 · 最小可用客户端（Python，单文件）

原理：客户端内置 aid + 匿名设备注册 → 服务端下发 ASR app_key → 用它连 WebSocket，
      按 20ms 一帧送 16k/单声道/PCM16，边送边收识别结果。

用法: python3 asr_min.py <16k-mono-16bit.wav>
依赖: pip install websockets
"""
import asyncio
import hashlib
import json
import secrets
import sys
import time
import urllib.parse
import urllib.request
import uuid
import wave

import websockets

AID = 685343                     # 豆包输入法的 app id（取自客户端 ImeService.exe）
APP_NAME = "oime"
VERSION_CODE = 100102018
VERSION_NAME = "1.1.2"
UA = ("com.bytedance.android.doubaoime/100102018 (Linux; U; Android 16; en_US; "
      "Pixel 7 Pro; Build/BP2A.250605.031.A2)")
REGISTER_URL = "https://log.snssdk.com/service/2/device_register/"   # 头条系匿名设备注册
SETTINGS_URL = "https://is.snssdk.com/service/settings/v3/"          # 下发 asr_config.app_key
WS_URL = "wss://frontier-audio-ime-ws.doubao.com/ocean/api/v1/ws"

SAMPLE_RATE = 16000
FRAME_BYTES = 640                # 20ms × 16kHz × 2字节 = 640


# ----------------------------------------------------------------- 1. 拿凭据

def _post(url, params, body, headers):
    req = urllib.request.Request(url + "?" + urllib.parse.urlencode(params), data=body, method="POST")
    for key, value in headers.items():
        req.add_header(key, value)
    with urllib.request.urlopen(req, timeout=20) as resp:
        return json.loads(resp.read().decode())


def get_credentials():
    """返回 (device_id, app_key)。全过程匿名，不需要登录。"""
    cdid, openudid, clientudid = str(uuid.uuid4()), secrets.token_hex(8), str(uuid.uuid4())
    header = {
        "device_id": 0, "install_id": 0, "aid": AID, "app_name": APP_NAME,
        "version_code": VERSION_CODE, "version_name": VERSION_NAME,
        "manifest_version_code": VERSION_CODE, "update_version_code": VERSION_CODE,
        "channel": "official", "device_platform": "android", "os": "android", "os_api": "34",
        "os_version": "16", "device_type": "Pixel 7 Pro", "device_brand": "google",
        "device_model": "Pixel 7 Pro", "resolution": "1080*2400", "dpi": "420",
        "language": "zh", "timezone": 8, "access": "wifi", "region": "CN",
        "tz_name": "Asia/Shanghai", "tz_offset": 28800, "openudid": openudid,
        "clientudid": clientudid, "cdid": cdid,
    }
    body = json.dumps({"magic_tag": "ss_app_log", "header": header,
                       "_gen_time": int(time.time() * 1000)}).encode()
    params = {
        "device_platform": "android", "os": "android", "ssmix": "a",
        "_rticket": str(int(time.time() * 1000)), "cdid": cdid, "channel": "official",
        "aid": str(AID), "app_name": APP_NAME, "version_code": str(VERSION_CODE),
        "version_name": VERSION_NAME, "manifest_version_code": str(VERSION_CODE),
        "update_version_code": str(VERSION_CODE), "resolution": "1080*2400", "dpi": "420",
        "device_type": "Pixel 7 Pro", "device_brand": "google", "language": "zh",
        "os_api": "34", "os_version": "16", "ac": "wifi",
    }
    out = _post(REGISTER_URL, params, body,
                {"Content-Type": "application/json", "User-Agent": UA})
    device_id = str(out.get("device_id_str") or out.get("device_id") or "")
    if not device_id or device_id == "0":
        raise RuntimeError(f"设备注册失败: {out}")

    stub_body = b"body=null"
    params = {
        "device_platform": "android", "os": "android", "ssmix": "a", "channel": "official",
        "aid": str(AID), "app_name": APP_NAME, "version_code": str(VERSION_CODE),
        "version_name": VERSION_NAME, "device_id": device_id, "cdid": cdid,
        "_rticket": str(int(time.time() * 1000)),
    }
    out = _post(SETTINGS_URL, params, stub_body, {
        "User-Agent": UA,
        "x-ss-stub": hashlib.md5(stub_body).hexdigest().upper(),
        "Content-Type": "application/x-www-form-urlencoded",
    })
    app_key = (out.get("data", {}).get("settings", {}).get("asr_config", {}) or {}).get("app_key")
    if not app_key:
        raise RuntimeError(f"settings 没给 app_key: {out}")
    return device_id, app_key


# ----------------------------------------------------------------- 2. protobuf 帧

def _varint(value):
    out = bytearray()
    while True:
        byte = value & 0x7F
        value >>= 7
        out.append(byte | 0x80 if value else byte)
        if not value:
            return bytes(out)


def encode(token="", method="", payload="", audio=b"", request_id="", state=0):
    """上游请求帧的字段号（实测得到）：
       2=token(app_key) 3=service_name 5=method 6=payload 7=audio 8=request_id 9=frame_state
    """
    body = bytearray()
    for num, val in ((2, token), (3, "ASR"), (5, method), (6, payload)):
        if val:
            raw = val.encode()
            body += _varint(num << 3 | 2) + _varint(len(raw)) + raw
    if audio:
        body += _varint(7 << 3 | 2) + _varint(len(audio)) + audio
    if request_id:
        raw = request_id.encode()
        body += _varint(8 << 3 | 2) + _varint(len(raw)) + raw
    if state:
        body += _varint(9 << 3) + _varint(state)
    return bytes(body)


def decode(data):
    """响应帧 → {字段号: [值]}。2=task_id 4=消息类型 5=状态码 6=状态文本 7=结果 JSON"""
    out, off = {}, 0

    def read_varint(buf, pos):
        val = shift = 0
        while True:
            byte = buf[pos]
            pos += 1
            val |= (byte & 0x7F) << shift
            shift += 7
            if not byte & 0x80:
                return val, pos

    while off < len(data):
        tag, off = read_varint(data, off)
        field, wire = tag >> 3, tag & 7
        if wire == 0:
            val, off = read_varint(data, off)
            out.setdefault(field, []).append(val)
        elif wire == 2:
            length, off = read_varint(data, off)
            out.setdefault(field, []).append(data[off:off + length])
            off += length
        elif wire == 1:
            off += 8
        elif wire == 5:
            off += 4
        else:
            break
    return out


def text_of(msg, field):
    val = msg.get(field)
    if not val:
        return None
    return val[0].decode("utf-8", "replace") if isinstance(val[0], bytes) else val[0]


# ----------------------------------------------------------------- 3. 跑一次识别

def is_retryable(err):
    """上游会偶发 50700000 service discovery failure（实测约 1/5 概率），重试即可。"""
    msg = str(err)
    return "service discovery" in msg or any(f"[{code}" in msg for code in range(50000000, 60000000))


async def transcribe(path, retries=4):
    """带重试的一次识别：上游抖动 / 连接断开都重试整段。"""
    last_error = None
    for attempt in range(retries):
        try:
            return await _transcribe_once(path)
        except Exception as exc:                     # noqa: BLE001
            last_error = exc
            if not is_retryable(exc) or attempt == retries - 1:
                raise
            print(f"  [重试 {attempt + 1}/{retries - 1}] {exc}", file=sys.stderr)
            await asyncio.sleep(0.6)
    raise last_error


async def _transcribe_once(path):
    with wave.open(path, "rb") as wav:
        assert (wav.getnchannels(), wav.getsampwidth(), wav.getframerate()) == (1, 2, 16000), \
            "需要 16kHz / 单声道 / 16bit 的 WAV（可用 ffmpeg -ar 16000 -ac 1 转换）"
        pcm = wav.readframes(wav.getnframes())

    device_id, app_key = get_credentials()
    url = f"{WS_URL}?aid={AID}&device_id={device_id}"
    headers = {"User-Agent": UA, "proto-version": "v2", "x-custom-keepalive": "true"}
    request_id = str(uuid.uuid4())
    text, done = "", False

    async with websockets.connect(url, additional_headers=headers,
                                  open_timeout=15, max_size=None) as ws:
        # ① StartTask：建任务
        await ws.send(encode(app_key, "StartTask", request_id=request_id))
        msg = decode(await ws.recv())
        if text_of(msg, 4) != "TaskStarted":
            raise RuntimeError(f"StartTask 失败: {text_of(msg, 4)} {msg.get(5)} {text_of(msg, 6)}")

        # ② StartSession：声明音频格式 + 识别参数
        session_payload = json.dumps({
            "audio_info": {"channel": 1, "format": "raw", "sample_rate": SAMPLE_RATE},
            "enable_punctuation": True,
            "enable_speech_rejection": False,
            "extra": {
                "did": device_id, "input_mode": "tool",
                "strong_ddc": True, "use_twopass_retry": True,
                "enable_asr_twopass": True, "enable_asr_threepass": True,
                "disable_user_words": False, "cell_compress_rate": 8,
            },
        }, ensure_ascii=False)
        await ws.send(encode(app_key, "StartSession", payload=session_payload, request_id=request_id))
        msg = decode(await ws.recv())
        if text_of(msg, 4) != "SessionStarted":
            raise RuntimeError(f"StartSession 失败: {text_of(msg, 4)} {msg.get(5)} {text_of(msg, 6)}")

        # ③ TaskRequest：20ms 一帧；首帧 1、中间 3、末帧 9
        frames = [pcm[i:i + FRAME_BYTES] for i in range(0, len(pcm), FRAME_BYTES)]
        for index, frame in enumerate(frames):
            state = 1 if index == 0 else (9 if index == len(frames) - 1 else 3)
            await ws.send(encode("", "TaskRequest", payload="{}", audio=frame,
                                 request_id=request_id, state=state))
            await asyncio.sleep(0.02)          # 按实时速度喂；也可以更快

        # ④ FinishSession：收尾
        await ws.send(encode(app_key, "FinishSession", request_id=request_id))

        while not done:
            msg = decode(await asyncio.wait_for(ws.recv(), timeout=20))
            kind, result = text_of(msg, 4), text_of(msg, 7)
            if result:
                data = json.loads(result)
                for item in data.get("results") or []:
                    if item.get("text"):
                        text = item["text"]        # 累积文本，最后一条即完整结果
            if kind in ("SessionFinished", "TaskFinished"):
                done = True
            elif kind in ("SessionFailed", "TaskFailed"):
                raise RuntimeError(f"识别失败: {msg.get(5)} {text_of(msg, 6)}")
    return text


if __name__ == "__main__":
    if len(sys.argv) < 2:
        sys.exit("用法: python3 asr_min.py <16k-mono-16bit.wav>")
    print(asyncio.run(transcribe(sys.argv[1])))

Node.js

npm i ws
node asr_min.js test.wav

#!/usr/bin/env node
/**
 * 豆包输入法语音识别 · 最小可用客户端（Node.js）
 *
 * 用法: npm i ws && node asr_min.js <16k-mono-16bit.wav>
 *
 * 为什么不用 Node 自带的全局 WebSocket：上游靠 `proto-version: v2` 这个握手头选择
 * 二进制 protobuf 协议，没有它就回 JSON；而全局 WebSocket 无法设置自定义请求头，
 * 所以这里用 ws 包（浏览器同理——这也是网页端必须走服务端中转的原因）。
 */

const fs = require("node:fs");
const crypto = require("node:crypto");
const WebSocket = require("ws");

const AID = 685343;                 // 豆包输入法的 app id（取自客户端 ImeService.exe）
const APP_NAME = "oime";
const VERSION_CODE = 100102018;
const VERSION_NAME = "1.1.2";
const UA = "com.bytedance.android.doubaoime/100102018 (Linux; U; Android 16; en_US; " +
           "Pixel 7 Pro; Build/BP2A.250605.031.A2)";
const REGISTER_URL = "https://log.snssdk.com/service/2/device_register/";
const SETTINGS_URL = "https://is.snssdk.com/service/settings/v3/";
const WS_URL = "wss://frontier-audio-ime-ws.doubao.com/ocean/api/v1/ws";

const SAMPLE_RATE = 16000;
const FRAME_BYTES = 640;            // 20ms × 16kHz × 2 字节

// ------------------------------------------------------------- 1. 拿凭据

async function postJson(url, params, body, headers) {
  const target = `${url}?${new URLSearchParams(params)}`;
  const resp = await fetch(target, { method: "POST", body, headers });
  if (!resp.ok) throw new Error(`${url} HTTP ${resp.status}: ${await resp.text()}`);
  return resp.json();
}

async function getCredentials() {
  const cdid = crypto.randomUUID();
  const openudid = crypto.randomUUID().replace(/-/g, "").slice(0, 16);
  const clientudid = crypto.randomUUID();
  const header = {
    device_id: 0, install_id: 0, aid: AID, app_name: APP_NAME,
    version_code: VERSION_CODE, version_name: VERSION_NAME,
    manifest_version_code: VERSION_CODE, update_version_code: VERSION_CODE,
    channel: "official", device_platform: "android", os: "android", os_api: "34",
    os_version: "16", device_type: "Pixel 7 Pro", device_brand: "google",
    device_model: "Pixel 7 Pro", resolution: "1080*2400", dpi: "420",
    language: "zh", timezone: 8, access: "wifi", region: "CN",
    tz_name: "Asia/Shanghai", tz_offset: 28800, openudid, clientudid, cdid,
  };
  const reg = await postJson(REGISTER_URL, {
    device_platform: "android", os: "android", ssmix: "a", _rticket: Date.now(),
    cdid, channel: "official", aid: AID, app_name: APP_NAME,
    version_code: VERSION_CODE, version_name: VERSION_NAME,
    manifest_version_code: VERSION_CODE, update_version_code: VERSION_CODE,
    resolution: "1080*2400", dpi: "420", device_type: "Pixel 7 Pro",
    device_brand: "google", language: "zh", os_api: "34", os_version: "16", ac: "wifi",
  }, JSON.stringify({ magic_tag: "ss_app_log", header, _gen_time: Date.now() }),
     { "Content-Type": "application/json", "User-Agent": UA });

  const deviceId = String(reg.device_id_str || reg.device_id || "");
  if (!deviceId || deviceId === "0") throw new Error(`设备注册失败: ${JSON.stringify(reg)}`);

  const stub = "body=null";
  const settings = await postJson(SETTINGS_URL, {
    device_platform: "android", os: "android", ssmix: "a", channel: "official",
    aid: AID, app_name: APP_NAME, version_code: VERSION_CODE, version_name: VERSION_NAME,
    device_id: deviceId, cdid, _rticket: Date.now(),
  }, stub, {
    "User-Agent": UA,
    "Content-Type": "application/x-www-form-urlencoded",
    "x-ss-stub": crypto.createHash("md5").update(stub).digest("hex").toUpperCase(),
  });
  const appKey = settings?.data?.settings?.asr_config?.app_key;
  if (!appKey) throw new Error(`settings 没给 app_key: ${JSON.stringify(settings)}`);
  return { deviceId, appKey };
}

// ------------------------------------------------------------- 2. protobuf 帧

function varint(value) {
  const out = [];
  while (true) {
    let byte = value & 0x7f;
    value = Math.floor(value / 128);
    out.push(value ? byte | 0x80 : byte);
    if (!value) return Buffer.from(out);
  }
}

function encode({ token = "", method = "", payload = "", audio = null, requestId = "", state = 0 }) {
  const parts = [];
  for (const [num, val] of [[2, token], [3, "ASR"], [5, method], [6, payload]]) {
    if (!val) continue;
    const raw = Buffer.from(val, "utf8");
    parts.push(varint((num << 3) | 2), varint(raw.length), raw);
  }
  if (audio && audio.length) {
    parts.push(varint((7 << 3) | 2), varint(audio.length), audio);
  }
  if (requestId) {
    const raw = Buffer.from(requestId, "utf8");
    parts.push(varint((8 << 3) | 2), varint(raw.length), raw);
  }
  if (state) parts.push(varint(9 << 3), varint(state));
  return Buffer.concat(parts);
}

function decode(buf) {
  const out = {};
  let off = 0;
  const readVarint = (pos) => {
    let val = 0, shift = 0;
    while (true) {
      const byte = buf[pos++];
      val |= (byte & 0x7f) << shift;
      shift += 7;
      if (!(byte & 0x80)) return [val, pos];
    }
  };
  while (off < buf.length) {
    let tag; [tag, off] = readVarint(off);
    const field = tag >> 3, wire = tag & 7;
    if (wire === 0) {
      let val; [val, off] = readVarint(off);
      (out[field] ||= []).push(val);
    } else if (wire === 2) {
      let len; [len, off] = readVarint(off);
      (out[field] ||= []).push(buf.subarray(off, off + len));
      off += len;
    } else if (wire === 1) off += 8;
    else if (wire === 5) off += 4;
    else break;
  }
  return out;
}

const textOf = (msg, field) => {
  const val = msg[field]?.[0];
  return val == null ? null : (Buffer.isBuffer(val) ? val.toString("utf8") : val);
};

// ------------------------------------------------------------- 3. 跑一次识别

function readWav(path) {
  const buf = fs.readFileSync(path);
  if (buf.toString("ascii", 0, 4) !== "RIFF") throw new Error("不是 WAV 文件");
  let off = 12;
  while (off + 8 <= buf.length) {
    const id = buf.toString("ascii", off, off + 4);
    const size = buf.readUInt32LE(off + 4);
    if (id === "fmt ") {
      const channels = buf.readUInt16LE(off + 10);
      const rate = buf.readUInt32LE(off + 12);
      const bits = buf.readUInt16LE(off + 22);
      if (channels !== 1 || rate !== SAMPLE_RATE || bits !== 16) {
        throw new Error(`需要 16kHz/单声道/16bit 的 WAV，实际 ${rate}Hz/${channels}ch/${bits}bit`);
      }
    }
    if (id === "data") {
      return buf.subarray(off + 8, off + 8 + size);
    }
    off += 8 + size + (size % 2);
  }
  throw new Error("WAV 里没有 data 段");
}

async function transcribeOnce(path) {
  const pcm = readWav(path);
  const { deviceId, appKey } = await getCredentials();
  const requestId = crypto.randomUUID();
  const ws = new WebSocket(`${WS_URL}?aid=${AID}&device_id=${deviceId}`, {
    headers: { "User-Agent": UA, "proto-version": "v2", "x-custom-keepalive": "true" },
  });

  let text = "";
  let finished = false;
  let failure = null;
  const queue = [];
  const waiters = [];
  let closed = false;

  ws.on("message", (data) => {
    const msg = decode(Buffer.from(data));
    if (waiters.length) waiters.shift()(msg); else queue.push(msg);
  });
  ws.on("error", (err) => { failure = failure || err; closed = true; while (waiters.length) waiters.shift()(null); });
  ws.on("close", () => { closed = true; while (waiters.length) waiters.shift()(null); });

  const nextMessage = () => new Promise((resolve, reject) => {
    if (queue.length) return resolve(queue.shift());
    if (closed) return reject(new Error("连接已关闭"));
    const timer = setTimeout(() => reject(new Error("等待上游响应超时")), 25000);
    waiters.push((msg) => { clearTimeout(timer); msg ? resolve(msg) : reject(new Error("连接已关闭")); });
  });

  await new Promise((resolve, reject) => {
    ws.once("open", resolve);
    ws.once("error", () => reject(new Error("WebSocket 连接失败")));
  });

  // ① StartTask
  ws.send(encode({ token: appKey, method: "StartTask", requestId }));
  let msg = await nextMessage();
  if (textOf(msg, 4) !== "TaskStarted") {
    throw new Error(`StartTask 失败: ${textOf(msg, 4)} ${msg[5]} ${textOf(msg, 6)}`);
  }

  // ② StartSession
  const sessionPayload = JSON.stringify({
    audio_info: { channel: 1, format: "raw", sample_rate: SAMPLE_RATE },
    enable_punctuation: true,
    enable_speech_rejection: false,
    extra: {
      did: deviceId, input_mode: "tool",
      strong_ddc: true, use_twopass_retry: true,
      enable_asr_twopass: true, enable_asr_threepass: true,
      disable_user_words: false, cell_compress_rate: 8,
    },
  });
  ws.send(encode({ token: appKey, method: "StartSession", payload: sessionPayload, requestId }));
  msg = await nextMessage();
  if (textOf(msg, 4) !== "SessionStarted") {
    throw new Error(`StartSession 失败: ${textOf(msg, 4)} ${msg[5]} ${textOf(msg, 6)}`);
  }

  // ③ 边收结果边发音频
  const collector = (async () => {
    while (!finished) {
      let incoming;
      try {
        incoming = await nextMessage();
      } catch (err) { failure = failure || err; return; }
      if (!incoming) { failure = failure || new Error("连接提前关闭"); return; }
      const kind = textOf(incoming, 4);
      const result = textOf(incoming, 7);
      if (result) {
        const data = JSON.parse(result);
        for (const item of data.results || []) {
          if (item.text) text = item.text;        // 累积文本，最后一条即完整结果
        }
      }
      if (kind === "SessionFinished" || kind === "TaskFinished") finished = true;
      if (kind === "SessionFailed" || kind === "TaskFailed") {
        failure = new Error(`识别失败: ${incoming[5]} ${textOf(incoming, 6)}`);
        finished = true;
      }
    }
  })();

  for (let i = 0; i < pcm.length; i += FRAME_BYTES) {
    const frame = pcm.subarray(i, i + FRAME_BYTES);
    const state = i === 0 ? 1 : (i + FRAME_BYTES >= pcm.length ? 9 : 3);
    ws.send(encode({ method: "TaskRequest", payload: "{}", audio: frame, requestId, state }));
    await new Promise((r) => setTimeout(r, 20));         // 按实时速度喂
  }

  // ④ FinishSession
  ws.send(encode({ token: appKey, method: "FinishSession", requestId }));
  await Promise.race([collector, new Promise((r) => setTimeout(r, 25000))]);
  ws.close();
  if (failure) throw failure;
  return text;
}

const isRetryable = (err) => /service discovery|\[5\d{7}/.test(String(err));

async function transcribe(path, retries = 4) {
  for (let attempt = 0; ; attempt++) {
    try {
      return await transcribeOnce(path);
    } catch (err) {
      if (!isRetryable(err) || attempt >= retries - 1) throw err;
      console.error(`  [重试 ${attempt + 1}/${retries - 1}] ${err.message}`);
      await new Promise((r) => setTimeout(r, 600));
    }
  }
}

const file = process.argv[2];
if (!file) {
  console.error("用法: node asr_min.js <16k-mono-16bit.wav>");
  process.exit(1);
}
transcribe(file).then((text) => console.log(text))
  .catch((err) => { console.error("失败:", err.message); process.exit(2); });

Go

cd examples && go mod tidy && go run asr_min.go test.wav

// 豆包输入法语音识别 · 最小可用客户端（Go）
//
// 用法:
//   cd examples && go mod tidy && go run asr_min.go <16k-mono-16bit.wav>
//
// 依赖: github.com/gorilla/websocket
package main

import (
	"bytes"
	"crypto/md5"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/gorilla/websocket"
)

const (
	aid          = 685343 // 豆包输入法的 app id（取自客户端 ImeService.exe）
	appName      = "oime"
	versionCode  = 100102018
	versionName  = "1.1.2"
	userAgent    = "com.bytedance.android.doubaoime/100102018 (Linux; U; Android 16; en_US; Pixel 7 Pro; Build/BP2A.250605.031.A2)"
	registerURL  = "https://log.snssdk.com/service/2/device_register/"
	settingsURL  = "https://is.snssdk.com/service/settings/v3/"
	wsURL        = "wss://frontier-audio-ime-ws.doubao.com/ocean/api/v1/ws"
	sampleRate   = 16000
	frameBytes   = 640 // 20ms × 16kHz × 2 字节
	maxRetry     = 4
)

var httpClient = &http.Client{Timeout: 25 * time.Second}

// ---------------------------------------------------------------- 1. 拿凭据

func postJSON(endpoint string, params url.Values, body []byte, headers map[string]string, out any) error {
	req, err := http.NewRequest(http.MethodPost, endpoint+"?"+params.Encode(), bytes.NewReader(body))
	if err != nil {
		return err
	}
	for k, v := range headers {
		req.Header.Set(k, v)
	}
	resp, err := httpClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	raw, _ := io.ReadAll(resp.Body)
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("%s HTTP %s: %s", endpoint, resp.Status, string(raw))
	}
	return json.Unmarshal(raw, out)
}

// getCredentials 匿名注册一台设备并换取 ASR 的 app_key（不需要账号登录）
func getCredentials() (deviceID, appKey string, err error) {
	cdid := uuid.NewString()
	openudid := strings.ReplaceAll(uuid.NewString(), "-", "")[:16]
	clientudid := uuid.NewString()

	header := map[string]any{
		"device_id": 0, "install_id": 0, "aid": aid, "app_name": appName,
		"version_code": versionCode, "version_name": versionName,
		"manifest_version_code": versionCode, "update_version_code": versionCode,
		"channel": "official", "device_platform": "android", "os": "android", "os_api": "34",
		"os_version": "16", "device_type": "Pixel 7 Pro", "device_brand": "google",
		"device_model": "Pixel 7 Pro", "resolution": "1080*2400", "dpi": "420",
		"language": "zh", "timezone": 8, "access": "wifi", "region": "CN",
		"tz_name": "Asia/Shanghai", "tz_offset": 28800, "openudid": openudid,
		"clientudid": clientudid, "cdid": cdid,
	}
	body, _ := json.Marshal(map[string]any{
		"magic_tag": "ss_app_log", "header": header, "_gen_time": time.Now().UnixMilli(),
	})
	params := url.Values{
		"device_platform": {"android"}, "os": {"android"}, "ssmix": {"a"},
		"_rticket": {fmt.Sprint(time.Now().UnixMilli())}, "cdid": {cdid}, "channel": {"official"},
		"aid": {fmt.Sprint(aid)}, "app_name": {appName}, "version_code": {fmt.Sprint(versionCode)},
		"version_name": {versionName}, "manifest_version_code": {fmt.Sprint(versionCode)},
		"update_version_code": {fmt.Sprint(versionCode)}, "resolution": {"1080*2400"}, "dpi": {"420"},
		"device_type": {"Pixel 7 Pro"}, "device_brand": {"google"}, "language": {"zh"},
		"os_api": {"34"}, "os_version": {"16"}, "ac": {"wifi"},
	}
	var reg struct {
		DeviceIDStr string `json:"device_id_str"`
		DeviceID    any    `json:"device_id"`
	}
	if err = postJSON(registerURL, params, body,
		map[string]string{"Content-Type": "application/json", "User-Agent": userAgent}, &reg); err != nil {
		return "", "", err
	}
	deviceID = reg.DeviceIDStr
	if deviceID == "" || deviceID == "0" {
		deviceID = fmt.Sprint(reg.DeviceID)
	}
	if deviceID == "" || deviceID == "0" {
		return "", "", errors.New("设备注册失败：没有 device_id")
	}

	stub := []byte("body=null")
	sum := md5.Sum(stub)
	params = url.Values{
		"device_platform": {"android"}, "os": {"android"}, "ssmix": {"a"}, "channel": {"official"},
		"aid": {fmt.Sprint(aid)}, "app_name": {appName}, "version_code": {fmt.Sprint(versionCode)},
		"version_name": {versionName}, "device_id": {deviceID}, "cdid": {cdid},
		"_rticket": {fmt.Sprint(time.Now().UnixMilli())},
	}
	var settings struct {
		Data struct {
			Settings struct {
				ASRConfig struct {
					AppKey string `json:"app_key"`
				} `json:"asr_config"`
			} `json:"settings"`
		} `json:"data"`
	}
	if err = postJSON(settingsURL, params, stub, map[string]string{
		"User-Agent":       userAgent,
		"x-ss-stub":        strings.ToUpper(hex.EncodeToString(sum[:])),
		"Content-Type":     "application/x-www-form-urlencoded",
	}, &settings); err != nil {
		return "", "", err
	}
	appKey = settings.Data.Settings.ASRConfig.AppKey
	if appKey == "" {
		return "", "", errors.New("settings 没给 asr_config.app_key")
	}
	return deviceID, appKey, nil
}

// ---------------------------------------------------------------- 2. protobuf 帧

func putVarint(buf *bytes.Buffer, value uint64) {
	for {
		b := byte(value & 0x7f)
		value >>= 7
		if value != 0 {
			buf.WriteByte(b | 0x80)
		} else {
			buf.WriteByte(b)
			return
		}
	}
}

func putBytes(buf *bytes.Buffer, field int, raw []byte) {
	if len(raw) == 0 {
		return
	}
	putVarint(buf, uint64(field<<3|2))
	putVarint(buf, uint64(len(raw)))
	buf.Write(raw)
}

// encode 上游请求帧：2=token 3=service 5=method 6=payload 7=audio 8=request_id 9=frame_state
func encode(token, method, payload string, audio []byte, requestID string, state int) []byte {
	var buf bytes.Buffer
	putBytes(&buf, 2, []byte(token))
	putBytes(&buf, 3, []byte("ASR"))
	putBytes(&buf, 5, []byte(method))
	putBytes(&buf, 6, []byte(payload))
	putBytes(&buf, 7, audio)
	putBytes(&buf, 8, []byte(requestID))
	if state != 0 {
		putVarint(&buf, uint64(9<<3))
		putVarint(&buf, uint64(state))
	}
	return buf.Bytes()
}

func readVarint(data []byte, pos int) (uint64, int) {
	var value uint64
	var shift uint
	for {
		b := data[pos]
		pos++
		value |= uint64(b&0x7f) << shift
		shift += 7
		if b&0x80 == 0 {
			return value, pos
		}
	}
}

// decode 响应帧 → 字段号到值的映射（2=task_id 4=事件 5=状态码 6=状态文本 7=结果 JSON）
func decode(data []byte) map[int][][]byte {
	out := map[int][][]byte{}
	for off := 0; off < len(data); {
		tag, next := readVarint(data, off)
		off = next
		field, wire := int(tag>>3), int(tag&7)
		switch wire {
		case 0:
			val, next := readVarint(data, off)
			off = next
			raw := make([]byte, 8)
			binary.LittleEndian.PutUint64(raw, val)
			out[field] = append(out[field], raw)
		case 2:
			length, next := readVarint(data, off)
			off = next
			out[field] = append(out[field], data[off:off+int(length)])
			off += int(length)
		case 1:
			off += 8
		case 5:
			off += 4
		default:
			return out
		}
	}
	return out
}

func textOf(msg map[int][][]byte, field int) string {
	values := msg[field]
	if len(values) == 0 {
		return ""
	}
	return string(values[0])
}

func codeOf(msg map[int][][]byte) uint64 {
	values := msg[5]
	if len(values) == 0 {
		return 0
	}
	return binary.LittleEndian.Uint64(values[0])
}

// ---------------------------------------------------------------- 3. 跑一次识别

func readWav(path string) ([]byte, error) {
	raw, err := os.ReadFile(path)
	if err != nil {
		return nil, err
	}
	if len(raw) < 44 || string(raw[0:4]) != "RIFF" {
		return nil, errors.New("不是 WAV 文件")
	}
	for off := 12; off+8 <= len(raw); {
		id := string(raw[off : off+4])
		size := int(binary.LittleEndian.Uint32(raw[off+4 : off+8]))
		if id == "fmt " {
			channels := binary.LittleEndian.Uint16(raw[off+10 : off+12])
			rate := binary.LittleEndian.Uint32(raw[off+12 : off+16])
			bits := binary.LittleEndian.Uint16(raw[off+22 : off+24])
			if channels != 1 || rate != sampleRate || bits != 16 {
				return nil, fmt.Errorf("需要 16kHz/单声道/16bit 的 WAV，实际 %dHz/%dch/%dbit", rate, channels, bits)
			}
		}
		if id == "data" {
			end := off + 8 + size
			if end > len(raw) {
				end = len(raw)
			}
			return raw[off+8 : end], nil
		}
		off += 8 + size + size%2
	}
	return nil, errors.New("WAV 里没有 data 段")
}

type result struct {
	Results []struct {
		Text string `json:"text"`
	} `json:"results"`
}

func transcribeOnce(path string) (string, error) {
	pcm, err := readWav(path)
	if err != nil {
		return "", err
	}
	deviceID, appKey, err := getCredentials()
	if err != nil {
		return "", err
	}
	requestID := uuid.NewString()

	header := http.Header{}
	header.Set("User-Agent", userAgent)
	header.Set("proto-version", "v2")
	header.Set("x-custom-keepalive", "true")
	target := fmt.Sprintf("%s?aid=%d&device_id=%s", wsURL, aid, deviceID)
	conn, _, err := websocket.DefaultDialer.Dial(target, header)
	if err != nil {
		return "", err
	}
	defer conn.Close()

	// ① StartTask
	if err = conn.WriteMessage(websocket.BinaryMessage, encode(appKey, "StartTask", "", nil, requestID, 0)); err != nil {
		return "", err
	}
	_, data, err := conn.ReadMessage()
	if err != nil {
		return "", err
	}
	if msg := decode(data); textOf(msg, 4) != "TaskStarted" {
		return "", fmt.Errorf("StartTask 失败: %s %d %s", textOf(msg, 4), codeOf(msg), textOf(msg, 6))
	}

	// ② StartSession
	sessionPayload, _ := json.Marshal(map[string]any{
		"audio_info":              map[string]any{"channel": 1, "format": "raw", "sample_rate": sampleRate},
		"enable_punctuation":      true,
		"enable_speech_rejection": false,
		"extra": map[string]any{
			"did": deviceID, "input_mode": "tool",
			"strong_ddc": true, "use_twopass_retry": true,
			"enable_asr_twopass": true, "enable_asr_threepass": true,
			"disable_user_words": false, "cell_compress_rate": 8,
		},
	})
	if err = conn.WriteMessage(websocket.BinaryMessage,
		encode(appKey, "StartSession", string(sessionPayload), nil, requestID, 0)); err != nil {
		return "", err
	}
	if _, data, err = conn.ReadMessage(); err != nil {
		return "", err
	}
	if msg := decode(data); textOf(msg, 4) != "SessionStarted" {
		return "", fmt.Errorf("StartSession 失败: %s %d %s", textOf(msg, 4), codeOf(msg), textOf(msg, 6))
	}

	// ③ 后台收结果，主线程按 20ms 一帧送音频
	text := make(chan string, 1)
	done := make(chan error, 1)
	go func() {
		var last string
		for {
			_, data, err := conn.ReadMessage()
			if err != nil {
				done <- err
				return
			}
			msg := decode(data)
			if raw := textOf(msg, 7); raw != "" {
				var parsed result
				if json.Unmarshal([]byte(raw), &parsed) == nil {
					for _, item := range parsed.Results {
						if item.Text != "" {
							last = item.Text // 累积文本，最后一条即完整结果
						}
					}
				}
			}
			switch textOf(msg, 4) {
			case "SessionFinished", "TaskFinished":
				text <- last
				return
			case "SessionFailed", "TaskFailed":
				done <- fmt.Errorf("识别失败: %d %s", codeOf(msg), textOf(msg, 6))
				return
			}
		}
	}()

	frames := (len(pcm) + frameBytes - 1) / frameBytes
	for i := 0; i < frames; i++ {
		start := i * frameBytes
		end := start + frameBytes
		if end > len(pcm) {
			end = len(pcm)
		}
		state := 3
		if i == 0 {
			state = 1
		} else if i == frames-1 {
			state = 9
		}
		if err = conn.WriteMessage(websocket.BinaryMessage,
			encode("", "TaskRequest", "{}", pcm[start:end], requestID, state)); err != nil {
			return "", err
		}
		time.Sleep(20 * time.Millisecond) // 按实时速度喂
	}

	// ④ FinishSession
	if err = conn.WriteMessage(websocket.BinaryMessage,
		encode(appKey, "FinishSession", "", nil, requestID, 0)); err != nil {
		return "", err
	}
	select {
	case out := <-text:
		return out, nil
	case err := <-done:
		return "", err
	case <-time.After(25 * time.Second):
		return "", errors.New("等待结果超时")
	}
}

func retryable(err error) bool {
	msg := err.Error()
	return strings.Contains(msg, "service discovery") || strings.Contains(msg, "失败: 5")
}

func transcribe(path string) (string, error) {
	var lastErr error
	for attempt := 0; attempt < maxRetry; attempt++ {
		out, err := transcribeOnce(path)
		if err == nil {
			return out, nil
		}
		lastErr = err
		if !retryable(err) {
			return "", err
		}
		fmt.Fprintf(os.Stderr, "  [重试 %d/%d] %v\n", attempt+1, maxRetry-1, err)
		time.Sleep(600 * time.Millisecond)
	}
	return "", lastErr
}

func main() {
	if len(os.Args) < 2 {
		fmt.Fprintln(os.Stderr, "用法: go run asr_min.go <16k-mono-16bit.wav>")
		os.Exit(1)
	}
	out, err := transcribe(os.Args[1])
	if err != nil {
		fmt.Fprintln(os.Stderr, "失败:", err)
		os.Exit(2)
	}
	fmt.Println(out)
}

4. 集成到你的项目

4.1 网页端为什么要中转

上游需要 3 个无法从浏览器发出的东西：自定义 WS 请求头（proto-version 等）、 设备级 app_key（不能下发到前端）、以及 HTTP 注册请求。所以浏览器方案必须是：

本 Demo 就是这么做的：前端只管录音和显示，服务端负责凭据、分帧、重试、降级、纠正词典。

4.2 本 Demo 的接口（可直接对接）

本项目的线上演示：doubao.hsfp.cn （代码就是本仓库）。下面示例里的 <你的域名> 换成你自己的部署地址即可。

WS 事件协议

// 客户端 → 服务端
二进制帧          16k 单声道 PCM16（随便怎么切，服务端会攒成 20ms 帧）
{"type":"stop"}  说完了，收尾并返回最终结果
{"type":"cancel"} 丢弃本次会话

// 服务端 → 客户端
{"type":"status","engine":"cloud|local","message":"…"}   状态/降级提示
{"type":"update","engine":"cloud","lines":[…],"text":"…"} 实时结果（每次给全量，前端整块重渲染）
{"type":"final","engine":"cloud","lines":[…],"text":"…","error":null}
{"type":"error","message":"…"}

// lines 里每一行
{"text":"你好。","start":0.0,"end":1.3,"ts":1792130303000,"final":true,
 "orig":"你号。","fixes":[["你号。","你好。"]]}   // orig/fixes 只在命中本地纠正时出现

最小前端调用示例（浏览器）

// 采集侧：AudioWorklet 把任意采样率转成 16k PCM16（完整实现在 static/worklet.js）
const ws = new WebSocket(`wss://<你的域名>/ws?engine=auto`);
ws.binaryType = "arraybuffer";

const ctx = new AudioContext({ sampleRate: 16000 });
await ctx.audioWorklet.addModule("/static/worklet.js");
const node = new AudioWorkletNode(ctx, "pcm16", { processorOptions: { targetRate: 16000 } });
ctx.createMediaStreamSource(await navigator.mediaDevices.getUserMedia({ audio: true })).connect(node);
node.connect(ctx.destination);

node.port.onmessage = (ev) => {
  if (ev.data.pcm && ws.readyState === 1) ws.send(ev.data.pcm);   // 二进制 PCM 直接发
};
ws.onmessage = (ev) => {
  const msg = JSON.parse(ev.data);
  if (msg.type === "update" || msg.type === "final") console.log(msg.text);
};
// 说完
setTimeout(() => ws.send(JSON.stringify({ type: "stop" })), 3000);

Node 后端批量转写（把服务端当上游用）

// 适合「把一批录音文件转成文字」：连本 Demo 的 /ws，喂 PCM，收最终文本
import fs from "node:fs";
import WebSocket from "ws";

async function transcribeFile(path) {
  const pcm = readWav16kMono(path);                       // 见 examples/asr_min.js 的 readWav
  const ws = new WebSocket("wss://<你的域名>/ws?engine=auto");
  return new Promise((resolve, reject) => {
    let done = null;
    ws.on("open", async () => {
      for (let i = 0; i < pcm.length; i += 640) {
        ws.send(pcm.subarray(i, i + 640));
        await new Promise((r) => setTimeout(r, 20));      // 按实时速度喂（也可更快）
      }
      ws.send(JSON.stringify({ type: "stop" }));
    });
    ws.on("message", (raw) => {
      const msg = JSON.parse(raw);
      if (msg.type === "final") { done = msg; resolve(msg.text); ws.close(); }
      if (msg.type === "error") reject(new Error(msg.message));
    });
    ws.on("close", () => { if (!done) reject(new Error("连接提前关闭")); });
  });
}

4.3 生产环境注意事项

5. 说话人分离（可选）

上游 ASR 的结果里没有 speaker 字段，所以「谁说的」得我们自己算。 默认走 online 模式：每句一定稿就提一次声纹、和已有说话人比相似度，边录边出标签； 也可以设 DOUBAO_DIAR_MODE=batch 改成"录完整段再聚类"（更准、但要等）。

这套配方与 screenpipe 一致（它的 Rust 实现同样是 pyannote 分割 + 3D-Speaker 类声纹 + 自定义聚类）； 本项目直接用 sherpa-onnx 的现成实现，省掉自己接 ONNX。

./venv/bin/pip install -r requirements-diarization.txt   # sherpa-onnx + numpy
./tools/fetch_diarization_models.sh                       # 下模型到 ./models（走 hf-mirror）

声纹库：从"1 号 2 号"升级到"这是谁"

分离只给出匿名编号，而且编号只在单次录音内有意义。想跨录音认人，就把说话人的声纹存下来做 1:N 识别： 有两种方式：① 打开 /voice 声纹管理页，念 3 段不同内容（每段 2~4 秒）， 页面会显示每段的质量和"与前几段的一致性"，保存即入库（推荐，最干净）； ② 在录音页点「说话人 N」标签、填真名并确认"记住这个声音"。 入库的是那个人每段录音的声纹向量（CAM++，192 维），存在浏览器 localStorage； 每次会话开始时随 config 发给服务端，新句子先跟声纹库比余弦（默认阈值 0.5）， 认得就用真名（页面显示 ✓ 姓名），认不出再走会话内聚类。一个人可存多例， 检索取"最像的那一例"。

6. 离线 / 本地方案

本项目自带离线引擎（offline_engine.py）：用的是字节公开分发的 ASR 模型 （见 1.7），纯 ONNX + onnxruntime，不联网、不依赖豆包客户端。 网页上可以三档切换，方便对比效果：

每次会话的分隔线上会标出用了哪档（如 2026-10-17 03:33 · 离线 · 录音 43s / 识别到 42s），方便回溯对比。

6.1 离线引擎内部是怎么做的

它是个独立进程（默认 127.0.0.1:4272），接口形状和 FunASR 引擎完全一致， 所以 server.py 里两套本地引擎复用同一个客户端基类 ChunkEngineSession—— 换引擎只是换 URL 和“结果怎么翻译成前端要的行”。

浏览器 ──20ms PCM 帧──▶ server.py ──100ms 一批──▶ offline_engine.py
                          │                        ├─ 能量 VAD 切句
                          │                        ├─ encoder → decoder → argmax（整段）
                          ◀─ 定稿句（时间戳/质量分）─┤  punc 模型补标点
                          ◀─ 未定稿片段（每约1.2s）─┘
                          └─ 转成前端的 update / final 协议

POST /stream               → {"session_id": …}
POST /stream/{sid}/chunk   → {"sentences":[{start,end,text,quality}], "partial":"…", "audio_seconds": …}
POST /stream/{sid}/finish  → 全部句子
GET  /health               → 是否就绪 / token 数 / 是否带标点 / VAD 参数

切句与“未定稿”（体验的关键，也是踩过坑的地方）

实测效果（同一段音频跑两档）

./venv/bin/python tests/compare_engines.py     # 复现这张表（真连两个引擎跑）

怎么确认它真的没联网 / 怎么排障

curl -s http://127.0.0.1:4272/health        # 引擎自己的健康检查
tail -f /var/log/doubao-offline-asr.log     # 每句都有「定稿 1.23-3.45s（12 字，质量 0.683）」
/root/scripts/doubao-asr-web-ctl.sh offline        # 状态
/root/scripts/doubao-asr-web-ctl.sh offline-test   # 不进服务，直接用仓库音频自测

还没做的（说清楚）

• 真·端侧：现在“离线”指的是推理在本机服务器、不联网；要完全不出服务端，得把 encoder/decoder 搬进浏览器 WASM（模型 160MB+，加载与内存都考验）
• ITN：asr-itn-fst.fst 还没接（数字/单位/日期书面化）
• 热词：离线模型输入里没有 context 字段，目前只能事后同义替换；原始产品走的是“多次解码 + 质量分挑选”
• 流式编码器：现在是整段解码 + VAD 切句，做不到云端那种逐字上屏

离线引擎怎么跑

# 1) 下模型（约 310MB，公开 CDN；详见 1.7）
./tools/fetch_offline_asr_models.sh

# 2) 装依赖 + 起服务（默认监听 127.0.0.1:4272）
./venv/bin/pip install -r requirements-offline.txt
./venv/bin/python offline_engine.py                 # 前台
./venv/bin/python offline_engine.py --selftest tests/fixture-6s.wav   # 不进服务，直接自测

# 3) 让网页用上它（.env）
DOUBAO_OFFLINE_ENGINE=http://127.0.0.1:4272

# 4) 生产用 systemd（仓库外，部署环境里）
systemctl enable --now doubao-offline-asr

离线引擎的切句策略：能量 VAD（自适应底噪）+ 停顿 0.6s 断句 + 单段上限 12s； 一段话说到一半会每约 1.2s 给一次「未定稿」结果，前端表现为最后一行还在变，停顿后定稿并加标点。

另外两个开源兜底（可选）

本 Demo 的降级链（已经接好了）

云端失败时依次尝试：离线引擎 → FunASR 本地引擎（配了才用）。配置：

# .env
DOUBAO_OFFLINE_ENGINE=http://127.0.0.1:4272      # 离线引擎（本机 ONNX，默认就是它）
DOUBAO_LOCAL_ENGINE=http://127.0.0.1:4271        # 可选：FunASR 引擎（最后兜底）
DOUBAO_LOCAL_ENGINE_TOKEN=…                      # FunASR 引擎的 x-engine-token
DOUBAO_CLOUD_RETRY=2                             # 云端抖动时的重连次数
DOUBAO_CONTEXT=1                                 # 是否把热词上下文带给上游

# 选择引擎
wss://<你的域名>/ws?engine=auto      # 默认：云端优先，失败自动降级
wss://<你的域名>/ws?engine=cloud     # 只用云端（=?engine=online 等价）
wss://<你的域名>/ws?engine=offline   # 只用离线引擎（本机 ONNX）
wss://<你的域名>/ws?engine=local     # 只用 FunASR 引擎

自己搭一个本地 ASR（FunASR 最小示例）

# 引擎侧（本机 127.0.0.1:4271 在跑的就是这套）：FunASR 的流式 paraformer + VAD + 标点
from funasr import AutoModel

model = AutoModel(model="iic/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-online",
                  vad_model="iic/speech_fsmn_vad_zh-cn-16k-common-pytorch",
                  punc_model="iic/punc_ct-transformer_cn-en-common-vocab471067-large")

cache, text = {}, ""
for chunk in pcm_chunks:                    # 每块 16k/单声道/PCM16（建议 600ms）
    import numpy as np
    audio = np.frombuffer(chunk, dtype=np.int16).astype(np.float32) / 32768.0
    res = model.generate(input=audio, cache=cache, is_final=False, chunk_size=[0, 10, 5])
    text += "".join(r.get("text", "") for r in (res or []))
print(text)

要日语 / 多语种

# sherpa-onnx：下载中日英模型后直接跑（离线、无网络依赖）
pip install sherpa-onnx
# 例：模型目录里通常带 run.sh，或用官方 python api：
python -m sherpa_onnx.offline_recognizer --tokens tokens.txt --paraformer model.onnx audio.wav

7. 常见问题

Q：离线档认出来的字不对/在重复，怎么办？

先看句子后面有没有黄色小问号（低置信标记）——那说明这句的质量分低于 0.35。 离线模型对输入比云端敏感，常见原因：说话太快或停顿很多（会被合并/切分）、环境噪声大、麦克风增益过低。 自检顺序：① 用 ./offline_engine.py --selftest tests/fixture-6s.wav 确认引擎本身正常； ② 看 /var/log/doubao-offline-asr.log 里每句的质量分；③ 换成「在线」档对比，判断是音频问题还是模型问题。

Q：为什么我自己连不上，一直 50700000？

上游随机抖动（实测约 1/5），重试整段即可。如果重试也一直失败，可能是接口变了或被风控，换新设备试试、或改用本地引擎。

Q：直接用 whisper-1 模型名行不行？

不行，这不是 OpenAI 接口。字段 5 是固定的方法名，音频字段是裸 PCM/Opus，不解析 multipart。

Q：能不能不注册设备、硬编码一个 app_key？

不建议：app_key 按设备下发且会轮换，硬编码很快就失效；而且那属于拿别人的凭据。

Q：识别结果能拿到逐字时间吗？

能，在 results[].alternatives[].words[] 里（注意 end_time 偏大，需要按 1.3 的坑校正）。

Q：能做成 OpenAI 兼容的 /v1/audio/transcriptions 吗？

能，服务端包一层即可：接收 multipart → 转 16k PCM → 走本节协议 → 返回 {"text": ...}。 本 Demo 目前只提供了 WebSocket 接口，需要 HTTP 版可以照着 4.2 的协议自己加。