TokenHub大模型服务平台

实时语音转写

平台通过 WebSocket 提供实时语音转写 API（API 编码 a2t_realtime），适用于会议记录、直播字幕、客服质检等需要持续推流、边录边识别的长时会话场景。发送 end 并收到 end=true 后，服务端会关闭连接。

  1. 支持中间结果（variable）与句末稳态结果（fixed）推送。
  2. 支持智能标点与后处理，句末结果可携带 start_time / end_time（毫秒）。
  3. 支持上下文与热词增强，提升专有名词识别率。
  4. 支持说话人分离（speaker_separate=true 时使用 Yama 引擎）。
  5. 支持 heartbeat 心跳帧，用于刷新连接空闲计时。

支持语言

1 中文（Chinese）2 英语（English）3 阿拉伯语（Arabic）
4 德语（German）5 西班牙语（Spanish）6 法语（French）
7 印尼语（Indonesian）8 日语（Japanese）9 韩语（Korean）
10 葡萄牙语（Portuguese）11 俄语（Russian）12 土耳其语（Turkish）
13 越南语（Vietnamese）14 泰语（Thai）15 意大利语（Italian）
16 粤语（Cantonese）17 长沙话（Changsha dialect）18 客家话（Hakka dialect）
19 闽南话（Hokkien）20 南昌话（Nanchang dialect）21 山西话（Shanxi dialect）
22 苏州话（Suzhou dialect）23 上海话（Shanghainese）24 济南话（Jinan dialect）
25 四川话（Sichuanese）26 武汉话（Wuhan dialect）

使用流程

  1. 在 API Key 管理 获取密钥，并写入环境变量 API_KEY。
  2. 建立 WebSocket 连接：wss://maas-api.unisound.com/v2/audio/asr/realtime?model=u2-asr（详见 实时语音转写 API）。
  3. 发送文本帧 start，业务参数位于 data 对象内。
  4. 在 start 成功之后、end 之前，循环发送二进制音频帧。
  5. 长连接场景可周期性发送 { "type": "heartbeat" } 防止空闲超时。
  6. 发送文本帧 end 结束本轮识别。
  7. 持续接收 ResultVO，直至 end=true（成功后服务端关闭连接）。

start 之前发送的音频会被丢弃。布尔参数字段请传字符串 "true" / "false"。

过程示例

1. 建立连接并发送 start

Python
"""
本示例演示实时语音转写的 WebSocket 会话。
注意：需要先将密钥写入环境变量 API_KEY；并安装 websockets：pip install websockets
"""
import asyncio
import json
import os

import websockets

API_KEY = os.environ["API_KEY"]
MODEL = "u2-asr"
WS_URL = f"wss://maas-api.unisound.com/v2/audio/asr/realtime?model={MODEL}"

START_FRAME = {
    "type": "start",
    "data": {
        "format": "pcm",
        "sample": "16k",
        "variable": "true",
        "punctuation": "true",
        "post_proc": "true",
        "speaker_separate": "false",
        "context": "产品名称是小云助手",
        "hotwords": ["小云", "转写"],
    },
}


async def run(pcm_path: str):
    headers = {"Authorization": f"Bearer {API_KEY}"}
    async with websockets.connect(WS_URL, additional_headers=headers) as ws:
        await ws.send(json.dumps(START_FRAME))
        with open(pcm_path, "rb") as f:
            while chunk := f.read(3200):
                await ws.send(chunk)
                await asyncio.sleep(0.1)  # 模拟实时推流
        await ws.send(json.dumps({"type": "end"}))
        while True:
            msg = json.loads(await ws.recv())
            print(msg)
            if msg.get("end"):
                break


if __name__ == "__main__":
    asyncio.run(run("path/to/audio.pcm"))

2. 解析识别结果

Python
# 典型中间结果（variable）
{
  "base_resp": {
    "status_code": 0,
    "status_msg": "success"
  },
  "sid": "a1b2c3d4e5f6478990abcdef12345678",
  "type": "variable",
  "text": "你好",
  "start_time": null,
  "end_time": null,
  "end": false
}

# 典型句末结果（fixed）
{
  "base_resp": {
    "status_code": 0,
    "status_msg": "success"
  },
  "sid": "a1b2c3d4e5f6478990abcdef12345678",
  "type": "fixed",
  "text": "你好世界",
  "start_time": 0,
  "end_time": 1200,
  "end": false
}

# 会话结束
{
  "base_resp": {
    "status_code": 0,
    "status_msg": "success"
  },
  "sid": "a1b2c3d4e5f6478990abcdef12345678",
  "type": "fixed",
  "text": "",
  "end": true
}

更多字段说明、引擎差异（Stream / Yama）、音频格式与错误码，请参阅 实时语音转写 API。也可在 模型体验 · 实时转写 中在线调试。