流式转写 API

通过 WebSocket 将音频流式传输到 Convai 的 ASR 引擎，并接收实时转写。专为与 AI 角色进行低延迟、智能语音体验而设计。

此 API 仅在企业版计划中可用。

此 API 仍在开发中，目前为 Beta 功能。

流式传输实时音频输入并从 Convai 的 ASR（自动语音识别）引擎接收文本转写所需的所有相关 API 细节。

访问此端点以在基于 Convai 的应用程序和支持语音的 AI 角色中启用实时语音识别。 连接建立后，WebSocket 通道会流入音频并流出转写结果，从而实现响应迅速的对话式体验。

---

概述

基础 URL：

https://transcribe.convai.com

WebSocket 端点：

wss://transcribe.convai.com/stream

协议： 双向 —— 输入流式传输 16 位 PCM 音频，输出接收转写事件。

支持的语言： 英语

---

身份验证

请提供你的 Convai API 密钥 在初始 WebSocket 握手期间。

请求头

名称	类型	说明
CONVAI-API-KEY	字符串	你的唯一 API 密钥，可在你的 Convai 账户中找到。

替代方式（查询参数）

如果无法使用请求头身份验证：

wss://transcribe.convai.com/stream?convai-api-key=<your-api-key>

如果你的 API 密钥缺失或无效，连接将立即关闭，并返回错误事件。

---

连接会话

wss://transcribe.convai.com/stream

说明

与 Convai 的转写服务建立实时 WebSocket 连接。 会话激活后，你可以发送二进制音频帧并接收增量（transcript.partial）和最终（transcript.final）转写结果。

---

会话开始示例

{
  "type": "session.started",
  "data": {
    "session_id": "f1428g2-335...",
    "expires_at": "2025-10-13T12:00:00+00:00"
  }
}

会话关闭示例

{
  "type": "session.closed",
  "data": {}
}

---

响应（服务器事件）主体

尽管使用的是 WebSocket（而非传统的 JSON POST），消息负载仍遵循以下结构：

名称	类型	说明
type	字符串	消息类型，例如 finalize, stop，或 close.
data	对象	根据消息类型而定的可选数据字段。

---

WebSocket 事件参考

事件类型	说明	示例负载
session.started	会话成功初始化后由服务器发送。	{"type": "session.started", "data": {"session_id": "...", "expires_at": "..."}}
session.closed	表示会话已正常结束。	{"type": "session.closed", "data": {}}
transcript.partial	部分转写更新（非最终）。	{"type": "transcript.partial", "data": {"sequence_id": 1, "text": "hel", "is_final": false}}
transcript.final	带格式或不带格式的最终转写。	{"type": "transcript.final", "data": {"sequence_id": 1, "text": "Hello world.", "is_final": true, "is_formatted": true}}
error	当发生无效数据、API 密钥问题或连接问题时返回。	{"type": "error", "data": {"message": "Invalid API key."}}

---

常见数据字段

字段	类型	说明
sequence_id	整数	用于对转写消息排序的计数器。
text	字符串	到目前为止接收到的转写字符串。
is_final	布尔值	当当前话语的转写已最终确定时为 True。
is_formatted	布尔值	当已应用标点和大小写时为 True。
message_type	字符串	元数据标签，例如： Turn 或 FinalTranscript.
language_code	字符串 / null	检测到的语言。实验性功能，可以为 null。

---

流式音频要求

参数	规格
编码	PCM 16 位，小端序
声道	单声道
采样率	16 kHz
推荐帧大小	50–150 毫秒
最大帧大小	约 8 MiB

将音频数据作为 二进制 WebSocket 消息.

---

控制消息

发送基于文本的 JSON 消息来管理流：

命令	示例	说明
Finalize	{"type": "finalize"}	触发服务器发送最终转写。
Close	{"type": "close"}	正常关闭 WebSocket 会话。
Stop	{"type": "stop"}	等同于 close，可能会保留会话上下文。

控制消息必须始终以 UTF-8 编码文本.

---

错误处理

错误会以结构化 JSON 对象的形式报告。

错误消息示例

{
  "type": "error",
  "data": {
    "message": "无效的 API 密钥或未找到用户。"
  }
}

---

状态码

代码	说明
200	OK —— 请求/连接成功。
400	Bad Request —— 负载格式错误或无效。
401	Unauthorized —— API 密钥无效或缺失。
403	Forbidden —— 当前计划未被授权访问 API。
500	内部服务器错误。

---

故障排查

问题	可能原因	解决方案
401 Unauthorized	API 密钥无效或缺失。	请验证你的 CONVAI-API-KEY 请求头或查询参数。
未收到转写结果	音频格式不正确。	确保使用 PCM 16 位、单声道、16 kHz 编码。
频繁断开连接	空闲套接字或数据格式错误。	保持持续流式传输帧；实现重连逻辑。
缺少标点	未格式化的转写。	等待第二个 transcript.final 其带有 "is_formatted": true.

---

示例进展（单个话语）

"welcome"
"welcome to the"
"welcome to the museum"   <-- transcript.final
"Welcome to the museum."  <-- transcript.final（已格式化）

---

当前仅支持英语。

格式化的转写（标点/大小写）是可选的，且可能稍后才出现。

---

示例（端到端流式客户端）

下面提供了示例实现和命令，演示如何连接到流式转写 API 并执行实时转写。

端到端流式客户端 - Python

此示例使用 Python、WebSockets 和 sounddevice 的实时麦克风输入来创建一个实时转写客户端。

要求：

• Python 3.8+
• pip install websockets sounddevice

文件： convai_stt_stream.py

Python

# 端到端流式客户端
# 也可以为其他语言创建等效客户端。

# 需要：pip install websockets sounddevice
# 运行方式：python convai_stt_stream.py

import asyncio
import json
import os
import signal
import sys

import sounddevice as sd
import websockets

API_KEY = os.getenv("CONVAI_API_KEY", "REPLACE_WITH_YOUR_KEY")
WS_URL = "wss://transcribe.convai.com/stream"

SAMPLE_RATE = 16000
CHANNELS = 1
DTYPE = "int16"          # 16 位小端 PCM
CHUNK_MS = 100           # 推荐 50–150 毫秒
BLOCKSIZE = int(SAMPLE_RATE * CHUNK_MS / 1000)  # 每个块的帧数

shutdown = asyncio.Event()

def install_signal_handlers():
    def _handler(*_):
        shutdown.set()
    for s in (signal.SIGINT, signal.SIGTERM):
        try:
            signal.signal(s, _handler)
        except Exception:
            pass

async def sender(ws, audio_q: asyncio.Queue):
    try:
        while not shutdown.is_set():
            try:
                chunk = await asyncio.wait_for(audio_q.get(), timeout=0.25)
            except asyncio.TimeoutError:
                continue
            await ws.send(chunk)
    except asyncio.CancelledError:
        pass

async def receiver(ws):
    try:
        async for msg in ws:
            print(msg)
    except asyncio.CancelledError:
        pass
    except websockets.ConnectionClosed:
        pass

async def main():
    if not API_KEY or API_KEY.startswith("REPLACE_"):
        print("请设置 CONVAI_API_KEY 环境变量或更新 API_KEY。", file=sys.stderr)
        return 1

    loop = asyncio.get_running_loop()
    audio_q: asyncio.Queue[bytes] = asyncio.Queue(maxsize=64)

    # 创建音频回调
    def make_callback(loop_):
        def _cb(indata, frames, time_info, status):
            if status:
                print(status, file=sys.stderr)
            try:
                loop_.call_soon_threadsafe(audio_q.put_nowait, bytes(indata))
            except Exception:
                pass
        return _cb

    cb = make_callback(loop)

    # 打开麦克风流
    with sd.RawInputStream(
        samplerate=SAMPLE_RATE,
        channels=CHANNELS,
        dtype=DTYPE,
        blocksize=BLOCKSIZE,
        callback=cb,
    ):
        # 使用 API 密钥作为请求头进行连接
        async with websockets.connect(
            WS_URL,
            additional_headers={"CONVAI-API-KEY": API_KEY},
            max_size=8 * 1024 * 1024,
        ) as ws:
            # 如果可用，读取初始 session.started
            try:
                msg = await asyncio.wait_for(ws.recv(), timeout=2.0)
                print(msg)
            except asyncio.TimeoutError:
                pass

            send_task = asyncio.create_task(sender(ws, audio_q))
            recv_task = asyncio.create_task(receiver(ws))

            # 等待 Ctrl+C
            await shutdown.wait()

            # 完成当前话语并排空响应
            try:
                await ws.send(json.dumps({"type": "finalize"}))
                try:
                    while True:
                        m = await asyncio.wait_for(ws.recv(), timeout=1.0)
                        print(m)
                except asyncio.TimeoutError:
                    pass
            except Exception:
                pass

            # 礼貌地关闭会话
            try:
                await ws.send(json.dumps({"type": "close"}))
            except Exception:
                pass

            # 干净地取消后台任务
            for t in (send_task, recv_task):
                t.cancel()
            for t in (send_task, recv_task):
                try:
                    await t
                except asyncio.CancelledError:
                    pass

    return 0

if __name__ == "__main__":
    install_signal_handlers()
    try:
        rc = asyncio.run(main())
    except KeyboardInterrupt:
        rc = 0
    sys.exit(rc)

运行步骤

# 1. 安装依赖
pip install websockets sounddevice

# 2. 设置你的 Convai API 密钥
export CONVAI_API_KEY=<your-api-key>
# Windows PowerShell
# $Env:CONVAI_API_KEY = "<your-api-key>"

# 3. 运行流式客户端
python convai_stt_stream.py

输出示例

{"type": "session.started", "data": {"session_id": "f1a2...", "expires_at": "..."}}
{"type": "transcript.partial", "data": {"text": "hello "}}
{"type": "transcript.final", "data": {"text": "Hello world.", "is_formatted": true}}
{"type": "session.closed", "data": {}}

cURL - 快速连通性检查

curl 并非用于 WebSocket 流式传输。请使用它来验证 HTTPS 端点是否可达，以及你的密钥是否被接受。

# 预期返回 HTTP 200 和一个简短的 ok 负载
curl -i https://transcribe.convai.com/ \
  -H "CONVAI-API-KEY: <your-api-key>"

预期响应：

{"status": "ok"}

---

结论

该 流式转写 API 通过 WebSocket 提供实时、低延迟的语音识别，实现流畅自然的 AI 交互。 通过集成此 API，你可以在游戏、助手或沉浸式 Convai 支持环境中构建响应迅速的语音体验。