Python SDK_一句话识别_智能语音交互-阿里云

本文介绍了如何使用阿里云智能语音服务提供的Python SDK，包括SDK的安装方法及SDK代码示例。

注意事项

在使用SDK前，请先阅读接口说明，详情请参见接口说明。

下载安装

说明：
SDK仅支持Python 3，版本要求3.4及以上，暂不支持Python 2。
请确认已安装Python包管理工具setuptools，详情请参见setuptools。如果没有安装，请使用以下命令安装：
pip install setuptools

下载Python SDK。

安装。

在SDK目录执行以下命令：

 # 打包
 python setup.py bdist_egg
 # 安装
 python setup.py install

关键接口

NlsClient：语音处理client，相当于所有语音相关处理类的factory，全局只需创建一个实例。线程安全。
SpeechRecognizer：一句话识别处理类。设置请求参数、发送请求及声音数据。非线程安全。
- start方法：建立与服务端的连接。默认参数ping_interval表示自动发送ping命令的间隔，ping_timeout表示等待ping命令响应的pong消息的超时时间，需要满足ping_interval大于ping_timeout。
- send方法：发送语音数据到服务端。
- stop方法：结束识别并关闭与服务端的连接。
- close方法：关闭与服务端的网络链接。
SpeechRecognizerCallback：回调事件函数集合的类，语音结果、异常等回调的统一入口。
- on_started方法：客户端与服务端建立连接成功的回调。
- on_result_changed方法：客户端接收到中间识别结果的回调。
- on_completed方法：客户端接收到一句话识别结束的回调，包含一句话的完整识别结果。
- on_task_failed方法：客户端接收到异常错误的回调。
- on_channel_closed方法：客户端接收到断开网络链接成功的回调。

SDK调用注意事项

NlsClient对象创建一次可以重复使用。
SpeechRecognizer对象不能重复使用，一个识别任务对应一个SpeechRecognizer对象。例如有N和音频文件则要进行N次识别任务，创建N个SpeechRecognizer对象。
SpeechRecognizerCallback对象和SpeechRecognizer对象是一一对应的，不能将一个SpeechRecognizerCallback对象设置到多个SpeechRecognizer对象中，否则不能将各识别任务区分开。

代码示例

说明：
Demo中使用的音频文件为16000Hz采样率，请在控制台中将appkey对应项目的模型设置为通用模型，以获取准确的识别效果。如果使用其他音频，请设置为支持该音频场景的模型，关于模型设置，请参见管理项目。nls-sample-16k.wav
Demo中使用了SDK内置的默认一句话识别服务的外网访问URL，如果您使用阿里云上海ECS且需要使用内网访问URL，则在创建NlsClient对象时，设置内网访问的URL。
recognizer = client.create_recognizer(callback, "ws://nls-gateway.cn-shanghai-internal.aliyuncs.com/ws/v1")

# -*- coding: utf-8 -*-
import os
import time
import threading
import ali_speech
from ali_speech.callbacks import SpeechRecognizerCallback
from ali_speech.constant import ASRFormat
from ali_speech.constant import ASRSampleRate
class MyCallback(SpeechRecognizerCallback):
    """
    构造函数的参数没有要求，可根据需要设置添加
    示例中的name参数可作为待识别的音频文件名，用于在多线程中进行区分
    """
    def __init__(self, name='default'):
        self._name = name
    def on_started(self, message):
        print('MyCallback.OnRecognitionStarted: %s' % message)
    def on_result_changed(self, message):
        print('MyCallback.OnRecognitionResultChanged: file: %s, task_id: %s, result: %s' % (
            self._name, message['header']['task_id'], message['payload']['result']))
    def on_completed(self, message):
        print('MyCallback.OnRecognitionCompleted: file: %s, task_id:%s, result:%s' % (
            self._name, message['header']['task_id'], message['payload']['result']))
    def on_task_failed(self, message):
        print('MyCallback.OnRecognitionTaskFailed: %s' % message)
    def on_channel_closed(self):
        print('MyCallback.OnRecognitionChannelClosed')
def process(client, appkey, token):
    audio_name = 'nls-sample-16k.wav'
    callback = MyCallback(audio_name)
    recognizer = client.create_recognizer(callback)
    recognizer.set_appkey(appkey)
    recognizer.set_token(token)
    recognizer.set_format(ASRFormat.PCM)
    recognizer.set_sample_rate(ASRSampleRate.SAMPLE_RATE_16K)
    recognizer.set_enable_intermediate_result(False)
    recognizer.set_enable_punctuation_prediction(True)
    recognizer.set_enable_inverse_text_normalization(True)
    try:
        ret = recognizer.start()
        if ret < 0:
            return ret
        print('sending audio...')
        with open(audio_name, 'rb') as f:
            audio = f.read(3200)
            while audio:
                ret = recognizer.send(audio)
                if ret < 0:
                    break
                time.sleep(0.1)
                audio = f.read(3200)
        recognizer.stop()
    except Exception as e:
        print(e)
    finally:
        recognizer.close()
def process_multithread(client, appkey, token, number):
    thread_list = []
    for i in range(0, number):
        thread = threading.Thread(target=process, args=(client, appkey, token))
        thread_list.append(thread)
        thread.start()
    for thread in thread_list:
        thread.join()
if __name__ == "__main__":
    client = ali_speech.NlsClient()
    # 设置输出日志信息的级别：DEBUG、INFO、WARNING、ERROR
    client.set_log_level('INFO')
    appkey = '您的appkey'
    token = '您的Token'
    process(client, appkey, token)
    # 多线程示例
    # process_multithread(client, appkey, token, 2)