实时语音识别 Python SDK-智能语音交互(ISI)-阿里云帮助中心

Python SDK 可向实时语音识别服务发送音频流，并通过回调获取中间结果和句子识别结果。

前提条件

已准备 Python 3 环境。SDK 不支持 Python 2。
已获取项目 Appkey 和访问 Token。Appkey 的获取和模型配置请参见项目管理，Token 的获取请参见获取 Token。

下载安装

从GitHub获取源码，或下载NLS Python SDK 压缩包并解压。
安装构建工具 setuptools。
```
python -m pip install setuptools
```

在 SDK 根目录安装依赖和 SDK。

python -m pip install -r requirements.txt
python -m pip install .

在代码中导入 SDK。
```
import nls
```

代码示例

示例使用 Python 3.12 和 SDK 包版本 1.1.0，将 SDK 包内 tests/test1.pcm 的音频发送到实时语音识别服务。

音频为 16000 Hz、16 位、单声道 PCM。为 Appkey 对应的项目配置通用模型；使用其他音频时，选择与其语种、采样率和场景匹配的模型。配置方法请参见项目管理。
配置环境变量 NLS_APP_KEY 和 NLS_TOKEN，分别填入项目 Appkey 和访问 Token。这两个变量由示例代码读取。
将 AUDIO_FILE 设置为 test1.pcm 的本地路径。

示例通过外网地址访问服务。使用上海地域 ECS 并需要内网访问时，将 URL 改为 ws://nls-gateway-cn-shanghai-internal.aliyuncs.com/ws/v1。

import json
import os
import threading
import time

import nls

APPKEY = os.environ["NLS_APP_KEY"]
TOKEN = os.environ["NLS_TOKEN"]
URL = "wss://nls-gateway-cn-shanghai.aliyuncs.com/ws/v1"
AUDIO_FILE = "tests/test1.pcm"

ready = threading.Event()
finished = threading.Event()

def on_start(message, *args):
    ready.set()

def on_result_changed(message, *args):
    result = json.loads(message)
    print("Intermediate result:", result["payload"]["result"])

def on_sentence_end(message, *args):
    result = json.loads(message)
    print("Sentence result:", result["payload"]["result"])

def on_completed(message, *args):
    finished.set()

def on_error(message, *args):
    print("Recognition failed:", message)
    finished.set()

def on_close(*args):
    finished.set()

nls.enableTrace(False)  # 关闭 SDK 调试日志

with open(AUDIO_FILE, "rb") as audio:
    transcriber = nls.NlsSpeechTranscriber(
        url=URL,
        appkey=APPKEY,
        token=TOKEN,
        on_start=on_start,
        on_result_changed=on_result_changed,
        on_sentence_end=on_sentence_end,
        on_completed=on_completed,
        on_error=on_error,
        on_close=on_close,
    )
    try:
        transcriber.start(
            aformat="pcm",
            sample_rate=16000,
            enable_intermediate_result=True,
            enable_punctuation_prediction=True,
            enable_inverse_text_normalization=True,
        )
        if ready.is_set() and not finished.is_set():
            while not finished.is_set():
                chunk = audio.read(640)
                if not chunk:
                    break
                transcriber.send_audio(chunk)
                time.sleep(len(chunk) / (16000 * 2))
            if not finished.is_set():
                transcriber.stop()
    finally:
        transcriber.shutdown()

接口说明

实时语音识别客户端为 nls.NlsSpeechTranscriber。以下说明对应 SDK 包版本 1.1.0。

初始化（init）

创建客户端，设置凭证、网关和回调。

参数	类型	说明
`url`	str	网关 WebSocket 地址。默认值为 `wss://nls-gateway.cn-shanghai.aliyuncs.com/ws/v1`。
`appkey`	str	项目 Appkey。获取方法请参见项目管理。
`token`	str	访问 Token。获取方法请参见获取 Token。
`on_start`	callable	收到 `TranscriptionStarted` 时调用，表示识别任务已就绪。
`on_sentence_begin`	callable	收到 `SentenceBegin` 时调用，表示一个句子的识别开始。
`on_sentence_end`	callable	收到 `SentenceEnd` 时调用，消息的 `payload.result` 包含该句的完整识别结果。
`on_result_changed`	callable	收到 `TranscriptionResultChanged` 时调用，消息包含中间识别结果。需要在 `start` 中开启 `enable_intermediate_result`。
`on_completed`	callable	收到 `TranscriptionCompleted` 时调用，表示整个识别任务完成。句子结果通过 `on_sentence_end` 返回。
`on_error`	callable	收到服务端 `TaskFailed` 错误消息时调用。
`on_close`	callable	与服务端的连接关闭时调用。
`callback_args`	list	自定义参数列表。列表元素按顺序展开，传给各回调的 `*args`。

除 on_close 外，回调的第一个参数为 JSON 格式的消息字符串，其后为展开的 callback_args。on_close 只接收展开的自定义参数。例如，callback_args=["first", "second"] 时，消息回调接收 callback(message, "first", "second")，关闭回调接收 on_close("first", "second")。

客户端参数校验、连接或调用超时等错误还可能通过异常报告，应在调用代码中处理。

start

开始实时语音识别，并同步等待任务就绪。识别就绪时触发 on_start；服务端任务失败时触发 on_error。任务就绪后可发送音频。

参数	类型	说明
`aformat`	str	音频格式，默认值为 `pcm`。支持 PCM、OPUS、OPU，其他音频格式及要求请参见实时语音识别接口说明。SDK 不进行音频编码转换，使用 OPUS 或 OPU 时需自行编码。
`sample_rate`	int	音频采样率，默认值为 16000 Hz。
`ch`	int	音频通道数，默认值为 `1`，仅支持单声道。
`enable_intermediate_result`	bool	是否返回中间结果，默认值为 `False`。
`enable_punctuation_prediction`	bool	是否在识别结果中添加标点，默认值为 `False`。
`enable_inverse_text_normalization`	bool	是否开启逆文本正则化（ITN），将中文数字转换为阿拉伯数字。默认值为 `False`。
`timeout`	int	等待任务就绪的超时时间，默认值为 10 秒。超时抛出 `StartTimeoutException`。
`ping_interval`	int 或 float	发送 Ping 的间隔，默认值为 8 秒。设置为 `0` 或 `None` 时不发送周期 Ping。
`ping_timeout`	int 或 float	发送 Ping 后等待 Pong 的超时时间，默认值为 `None`，表示不检查 Pong 超时。启用时须为正数；同时启用周期 Ping 时须小于 `ping_interval`。
`ex`	dict	额外请求参数，默认值为 `None`。字典内容以键值对形式合并到请求的 `payload` 中。请求参数请参见实时语音识别接口说明。

返回值：None。识别状态和结果通过回调返回。参数或连接错误可能抛出异常；连接关闭或任务失败也会结束等待。

stop

结束音频发送，并同步等待识别任务结束。正常完成时触发 on_completed；任务失败或连接关闭也会结束等待。

参数	类型	说明
`timeout`	int	等待任务结束的超时时间，默认值为 10 秒。超时抛出 `StopTimeoutException`。

返回值：None。

shutdown

强制关闭通过 start 创建的连接。对已创建的连接重复调用无副作用。

参数：无。返回值：None。

send_audio

发送二进制音频数据，格式须与start 的 aformat 一致。

参数	类型	说明
`pcm_data`	bytes	待发送的音频数据。使用 OPUS 或 OPU 时需自行编码，SDK 不会自动将 PCM 转换为其他格式。

返回值：None。未就绪或已结束的任务会忽略本次发送。识别就绪后，按实时速率持续发送音频。

ctrl

发送控制命令，调用形式为 ctrl(**kwargs)。

参数	类型	说明
`kwargs`	关键字参数	控制命令的参数，直接合并到 `payload` 中。传入字典时，使用 Python 的 `**` 展开语法。

在任务就绪后调用。未提供任何参数时抛出 InvalidParameter。返回值：None。

多线程和多并发

在启用全局解释器锁（GIL）的 CPython 中，同一时刻只有一个线程执行 Python 代码，某些性能导向的库可以绕过这一限制。需要充分利用多核计算资源时，推荐使用multiprocessing 或concurrent.futures.ProcessPoolExecutor。对于多个 I/O 密集型任务，多线程仍是合适的模型。

单解释器中的线程过多会增加线程切换开销，并可能导致 SDK 错误。不建议使用超过 200 个线程；需要更多并发任务时，推荐使用多进程或通过脚本创建多个解释器。