如何用华为云SIS服务实现麦克风音频流实时语音识别

华为云SIS服务中，提供一种实时语音识别服务的能力。它有两种接入方式：

1. 当有一个现成的音频文件时，可以用该文件模拟流式，取得实时识别的效果（这种一般用于POC效果测试）
2. 直接通过麦克风实时获取音频流，同步转写出文字（一般用于真实生产环境，如实时会议字幕）

第一步、安装SDK

可参考如下视频进行安装：https://www.bilibili.com/video/BV1XmQDYeEAp/

第二步、编写基于麦克风收音的代码，对接SDK

# -*- coding: utf-8 -*-
"""
Created on Sat Mar 15 18:40:22 2025

@author: HP
"""

import pyaudio
import numpy as np


from huaweicloud_sis.client.rasr_client import RasrClient
from huaweicloud_sis.bean.rasr_request import RasrRequest
from huaweicloud_sis.bean.callback import RasrCallBack
from huaweicloud_sis.bean.sis_config import SisConfig
import json
import os
# 鉴权参数
# 认证用的ak和sk硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全； 
# 本示例以ak和sk保存在环境变量中来实现身份验证为例，运行本示例前请先在本地环境中设置环境变量HUAWEICLOUD_SIS_AK/HUAWEICLOUD_SIS_SK/HUAWEICLOUD_SIS_PROJECT_ID。
ak = ""             # 从环境变量获取ak 参考https://support.huaweicloud.com/sdkreference-sis/sis_05_0003.html
assert ak is not None, "Please add ak in your develop environment"
sk = ""             # 从环境变量获取sk 参考https://support.huaweicloud.com/sdkreference-sis/sis_05_0003.html
assert sk is not None, "Please add sk in your develop environment"
project_id = ""     # project id 同region一一对应，参考https://support.huaweicloud.com/api-sis/sis_03_0008.html
region = 'cn-north-4'    


"""
    todo 请正确填写音频格式和模型属性字符串
    1. 音频格式一定要相匹配.
         例如音频是pcm格式，并且采样率为8k，则格式填写pcm8k16bit。
         如果返回audio_format is invalid 说明该文件格式不支持。具体支持哪些音频格式，需要参考一些api文档。

    2. 音频采样率要与属性字符串的采样率要匹配。
         例如格式选择pcm16k16bit，属性字符串却选择chinese_8k_common, 则会返回'audio_format' is not match model
"""

# 实时语音识别参数
audio_format = 'pcm16k16bit'  # 音频支持格式，如pcm16k16bit，详见api文档
property = 'chinese_16k_general'  # 属性字符串，language_sampleRate_domain, 如chinese_16k_general, 采样率要和音频一致。详见api文档


class MyCallback(RasrCallBack):
    """ 回调类，用户需要在对应方法中实现自己的逻辑，其中on_response必须重写 """

    def on_open(self):
        """ websocket连接成功会回调此函数 """
        print('websocket connect success')

    def on_start(self, message):
        """
            websocket 开始识别回调此函数
        :param message: 传入信息
        :return: -
        """
        print('webscoket start to recognize, %s' % message)

    def on_response(self, message):
        """
            websockert返回响应结果会回调此函数
        :param message: json格式
        :return: -
        """
        print(json.dumps(message, indent=2, ensure_ascii=False))

    def on_end(self, message):
        """
            websocket 结束识别回调此函数
        :param message: 传入信息
        :return: -
        """
        print('websocket is ended, %s' % message)

    def on_close(self):
        """ websocket关闭会回调此函数 """
        print('websocket is closed')

    def on_error(self, error):
        """
            websocket出错回调此函数
        :param error: 错误信息
        :return: -
        """
        print('websocket meets error, the error is %s' % error)

    def on_event(self, event):
        """
            出现事件的回调
        :param event: 事件名称
        :return: -
        """
        print('receive event %s' % event)

# 音频参数设置
SAMPLE_RATE = 16000
CHUNK = 3200
FORMAT = pyaudio.paInt16
CHANNELS = 1

# 初始化 PyAudio
p = pyaudio.PyAudio()


def rasr_example():
    """ 实时语音识别demo """
    # step1 初始化RasrClient, 暂不支持使用代理
    my_callback = MyCallback()
    config = SisConfig()
    # 设置连接超时,默认是10
    config.set_connect_timeout(10)
    # 设置读取超时, 默认是10
    config.set_read_timeout(10)
    # 设置connect lost超时，一般在普通并发下，不需要设置此值。默认是10
    config.set_connect_lost_timeout(10)
    # websocket暂时不支持使用代理
    rasr_client = RasrClient(ak=ak, sk=sk, use_aksk=True, region=region, project_id=project_id, callback=my_callback,
                             config=config)
    try:
        # step2 构造请求
        request = RasrRequest(audio_format, property)
        # 所有参数均可不设置，使用默认值
        request.set_add_punc('yes')  # 设置是否添加标点， yes or no， 默认no
        request.set_vad_head(10000)  # 设置有效头部， [0, 60000], 默认10000
        request.set_vad_tail(500)  # 设置有效尾部，[0, 3000]， 默认500
        request.set_max_seconds(30)  # 设置一句话最大长度，[1, 60], 默认30
        request.set_interim_results('yes')  # 设置是否返回中间结果，yes or no，默认no
        request.set_digit_norm('yes')  # 设置是否将语音中数字转写为阿拉伯数字，yes or no，默认yes
        # request.set_vocabulary_id('')     # 设置热词表id，若不存在则不填写，否则会报错
        request.set_need_word_info('no')  # 设置是否需要word_info，yes or no, 默认no

        # step3 选择连接模式
        # rasr_client.short_stream_connect(request)       # 流式一句话模式
        # rasr_client.sentence_stream_connect(request)    # 实时语音识别单句模式
        rasr_client.continue_stream_connect(request)  # 实时语音识别连续模式

        # use enterprise_project_Id
        # headers = {'Enterprise-Project-Id': 'your enterprise project id'}
        # rasr_client.continue_stream_connect(request, headers)

        # step4 发送音频
        rasr_client.send_start()

        # 打开麦克风流
        stream = p.open(format=FORMAT,
                        channels=CHANNELS,
                        rate=SAMPLE_RATE,
                        input=True,
                        frames_per_buffer=CHUNK)
        
        try:
            while True:
                print("Real time reading of microphone audio stream, reading 3200 bytes each time")
                data = stream.read(CHUNK)
                rasr_client.send_audio(data)  # 可选byte_len和sleep_time参数，建议使用默认值
        
        except KeyboardInterrupt:
            print("* 录音被中断")
        
        finally:
            # 停止并关闭流
            stream.stop_stream()
            stream.close()
            p.terminate()
            rasr_client.send_end()

    except Exception as e:
        print('rasr error', e)
    finally:
        # step5 关闭客户端，使用完毕后一定要关闭，否则服务端20s内没收到数据会报错并主动断开。
        rasr_client.close()


if __name__ == '__main__':
    rasr_example()

第三步、配置鉴权信息

鉴权信息包括项目ID、AK、SK。

项目ID获取可参考：https://www.bilibili.com/video/BV1SMQUYpEFK/

AKSK获取可参考：https://www.bilibili.com/video/BV1RrQDYBEeW/

获取以上信息后，在第二步的代码中补充完整

第四步、运行代码，看效果

效果可见：https://www.bilibili.com/video/BV1UVQDYoEx1/