Qwen-Audio-3.0-ASR-Flash-Streaming/Fun-ASR-Realtime 实时语音识别 Python SDK

import os
import signal  # 用于处理键盘事件（按 Ctrl+C 终止录音）
import sys

import dashscope
import pyaudio
from dashscope.audio.asr import *

mic = None
stream = None

# 设置录音参数
sample_rate = 16000  # 采样率（Hz）
channels = 1  # 单声道
dtype = 'int16'  # 数据类型
format_pcm = 'pcm'  # 音频数据格式
block_size = 3200  # 每个缓冲区的帧数


# 实时语音识别回调
class Callback(RecognitionCallback):
  def on_open(self) -> None:
    global mic
    global stream
    print('RecognitionCallback open.')
    mic = pyaudio.PyAudio()
    stream = mic.open(format=pyaudio.paInt16,
                          channels=1,
                          rate=16000,
                          input=True)

  def on_close(self) -> None:
    global mic
    global stream
    print('RecognitionCallback close.')
    stream.stop_stream()
    stream.close()
    mic.terminate()
    stream = None
    mic = None

  def on_complete(self) -> None:
    print('RecognitionCallback completed.')  # 识别完成

  def on_error(self, result: RecognitionResult) -> None:
    print('RecognitionCallback task_id: ', result.request_id)
    print('RecognitionCallback error: ', result.message)
    # 如果音频流正在运行，停止并关闭
    if 'stream' in globals() and stream.is_active():
      stream.stop_stream()
      stream.close()
    # 强制退出程序
    sys.exit(1)

  def on_event(self, result: RecognitionResult) -> None:
    sentence = result.get_sentence()
    if 'text' in sentence:
      print('RecognitionCallback text: ', sentence['text'])
      if RecognitionResult.is_sentence_end(sentence):
        print(
          'RecognitionCallback sentence end, request_id:%s, usage:%s'
          % (result.get_request_id(), result.get_usage(sentence)))


def signal_handler(sig, frame):
  print('按下 Ctrl+C，停止识别...')
  # 停止识别
  recognition.stop()
  print('识别已停止。')
  print(
    '[Metric] requestId: {}, first package delay ms: {}, last package delay ms: {}'
    .format(
      recognition.get_last_request_id(),
      recognition.get_first_package_delay(),
      recognition.get_last_package_delay(),
    ))
  # 强制退出程序
  sys.exit(0)


# 主函数
if __name__ == '__main__':
  # 如果未配置环境变量，请将下一行替换为：dashscope.api_key = "sk-xxx"
  dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

  dashscope.base_websocket_api_url='wss://dashscope.aliyuncs.com/api-ws/v1/inference'

  # 创建识别回调
  callback = Callback()

  # 异步模式调用识别服务，可自定义识别参数，如 model、format、sample_rate
  recognition = Recognition(
    model='qwen-audio-3.0-asr-flash-streaming',
    format=format_pcm,
    # 'pcm', 'wav', 'mp3', 'opus', 'speex', 'aac', 'amr'。支持的格式请参见文档。
    sample_rate=sample_rate,
    # 支持 16000。
    semantic_punctuation_enabled=False,
    callback=callback)

  # 启动识别
  recognition.start()

  signal.signal(signal.SIGINT, signal_handler)
  print("按 Ctrl+C 停止录音和识别...")
  # 创建键盘监听，直到按下 Ctrl+C

  while True:
    if stream:
      data = stream.read(3200, exception_on_overflow=False)
      recognition.send_audio_frame(data)
    else:
      break

  recognition.stop()