python实现阿里语音识别asr

直接上代码

import io
import os
import time
import wave
from typing import Optional, Tuple, List

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


TAG = __name__
sample_rate = 16000

class ASRProvider():
    def __init__(self, config: dict):
        self.interface_type = InterfaceType.NON_STREAM
        self.model = config.get("model")
        self.output_dir = config.get("output_dir")
        self.seg_duration = 12800
        
        dashscope.api_key = config.get("api_key")

        # 确保输出目录存在
        os.makedirs(self.output_dir, exist_ok=True)



    async def _send_request_gummy(self, audio_data, segment_size: int) -> Optional[str]:
        """Send request to Aliyun ASR service."""
        try:
            # 创建回调对象
            callback = self.CallbackGummy(self.interface_type)

            # 初始化翻译识别聊天对象
            translator = TranslationRecognizerChat(
                model=self.model,
                format="wav",
                sample_rate=sample_rate,
                callback=callback,
            )
            translator.start()

            # 读取音频数据并分段发送
            audio_data_len = len(audio_data)
            offset = 0
            while offset < audio_data_len:
                chunk = audio_data[offset:offset + segment_size]
                if not translator.send_audio_frame(chunk):
                    print("sentence end, stop sending")
                    break
                offset += segment_size

            #logger.bind(tag=TAG).info("调用阿里云百炼语音识别前：")
            translator.stop()
            #logger.bind(tag=TAG).info("调用阿里云百炼语音识别后：")

            # 获取识别结果
            if callback.transcription_result is not None:
                return callback.transcription_result.text
            else:
                logger.bind(tag=TAG).info("XXXXX语音转换没有结果：callback.transcription_result is None")
                return None

        except Exception as e:
            logger.bind(tag=TAG).error(f"ASR request failed: {e}", exc_info=True)
            return None

    async def _send_request_paraformer(self, audio_data, segment_size: int) -> Optional[str]:
        try:
            # 创建回调对象
            callback = self.CallbackParaformer(self.interface_type)

            # 初始化翻译识别聊天对象
            recognition = Recognition(model=self.model,
                                      format='wav',
                                      sample_rate=sample_rate,
                                      callback=callback)
            recognition.start()

            # 读取音频数据并分段发送
            audio_data_len = len(audio_data)
            offset = 0
            while offset < audio_data_len:
                chunk = audio_data[offset:offset + segment_size]
                if not recognition.send_audio_frame(chunk):
                    print("sentence end, stop sending")
                    break
                offset += segment_size

            logger.bind(tag=TAG).info("调用阿里云百炼语音识别前：")
            recognition.stop()
            logger.bind(tag=TAG).info("调用阿里云百炼语音识别后：")

            # 获取识别结果
            if callback.transcription_result is not None:
                return callback.transcription_result['text']
            else:
                logger.bind(tag=TAG).info("XXXXX语音转换没有结果：callback.transcription_result is None")
                return None

        except Exception as e:
            logger.bind(tag=TAG).error(f"ASR request failed: {e}", exc_info=True)
            return None

    

    async def speech_to_text(
            self, opus_data: List[bytes], session_id: str, audio_format="opus"
    ) -> Tuple[Optional[str], Optional[str]]:
        """将语音数据转换为文本"""

        file_path = None
        try:
            # 合并所有opus数据包
            if audio_format == "pcm":
                pcm_data = opus_data
            else:
                pcm_data = self.decode_opus(opus_data)
            combined_pcm_data = b''.join(pcm_data)

            # 判断是否保存为WAV文件
            if self.delete_audio_file:
                pass
            else:
                file_path = self.save_audio_to_file(pcm_data, session_id)

            # 直接使用PCM数据
            # 计算分段大小 (单声道, 16bit, 16kHz采样率)
            size_per_sec = 1 * 2 * 16000  # nchannels * sampwidth * framerate
            segment_size = int(size_per_sec * self.seg_duration / 1000)

            # 语音识别
            start_time = time.time()

            if "paraformer" in self.model:
                text = await self._send_request_paraformer(combined_pcm_data, segment_size)
            elif "gummy" in self.model:
                text = await self._send_request_gummy(combined_pcm_data, segment_size)
            else:
                logger.bind(tag=TAG).error(f"语音识别模型不支持: {self.model}", exc_info=True)
                return "", None
            #text = await self._send_request(wav_data, segment_size)
            if text:
                logger.bind(tag=TAG).debug(
                    f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}"
                )
                return text, file_path
            return "", file_path

        except Exception as e:
            logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
            return "", file_path


    mic = None
    stream = None

    class CallbackGummy(TranslationRecognizerCallback):
        def __init__(self,interface_type):
            super().__init__()
            self.transcription_result = None
            self.interface_type = interface_type

        def on_open(self) -> None:
            print("TranslationRecognizerCallback open.")
            if self.interface_type == InterfaceType.STREAM:
                global mic
                global stream
                mic = pyaudio.PyAudio()
                stream = mic.open(
                    format=pyaudio.paInt16, channels=1, rate=sample_rate, input=True
                )



        def on_close(self) -> None:
            print("TranslationRecognizerCallback close.")
            if self.interface_type == InterfaceType.STREAM:
                global mic
                global stream
                print("TranslationRecognizerCallback close.")
                stream.stop_stream()
                stream.close()
                mic.terminate()
                stream = None
                mic = None

        def on_event(
                self,
                request_id,
                transcription_result: TranscriptionResult,
                translation_result: TranslationResult,
                usage,
        ) -> None:
            print("request id: ", request_id)
            print("usage: ", usage)
            if translation_result is not None:
                print(
                    "translation_languages: ",
                    translation_result.get_language_list(),
                )
                english_translation = translation_result.get_translation("en")
                print("sentence id: ", english_translation.sentence_id)
                print("translate to english: ", english_translation.text)
            if transcription_result is not None:
                print("sentence id: ", transcription_result.sentence_id)
                print("transcription: ", transcription_result.text)
                logger.bind(tag=TAG).info(f"阿里云百炼语音识别调用成功transcription_result.text successful - text: {transcription_result.text}")
                self.transcription_result = transcription_result

        def on_error(self, message) -> None:
            print('error: {}'.format(message))

        def on_complete(self) -> None:
            print('TranslationRecognizerCallback complete')

    class CallbackParaformer(RecognitionCallback):
        def __init__(self,interface_type):
            super().__init__()
            self.transcription_result = None
            self.interface_type = interface_type

        def on_open(self) -> None:
            print("RecognitionCallback open.")
            if self.interface_type == InterfaceType.STREAM:
                global mic
                global stream
                mic = pyaudio.PyAudio()
                stream = mic.open(format=pyaudio.paInt16,
                                  channels=1,
                                  rate=sample_rate,
                                  input=True)

        def on_close(self) -> None:
            print("RecognitionCallback close.")
            if self.interface_type == InterfaceType.STREAM:
                global mic
                global stream
                stream.stop_stream()
                stream.close()
                mic.terminate()
                stream = None
                mic = None

        def on_event(self, result: RecognitionResult) -> None:
            print('RecognitionCallback sentence: ', result.get_sentence())
            self.transcription_result = result.get_sentence()

        def on_error(self, result: RecognitionResult) -> None:
            print('error: {}'.format(result.get_sentence()))

        def on_complete(self) -> None:
            print('RecognitionCallback complete')

支持gummy-chat-v1和paraformer模型，

 一、前置准备

1. 安装依赖

确保已安装 Python 3.7+，并通过 pip 安装 DashScope SDK：

pip install dashscope

2. 获取 API Key

• 登录 DashScope 控制台，在「API 密钥管理」中获取 API Key。
• 
• 若未将 API Key 配置到环境变量，需在代码中显式设置：

  import dashscope
  dashscope.api_key = "your-api-key"  # 替换为你的实际 Key

​​

一、核心定位与功能差异

​​关键区别​​：

• Gummy-Chat-v1 的语言识别是​​对话流程中的辅助功能​​，依赖文本输入或语音→文本转换；
• Paraformer-v2 的语言识别是​​语音处理的原生能力​​，直接分析音频特征，无需文本转换。

二、支持语言与方言能力对比

​​典型场景示例​​：

• Gummy-Chat-v1：用户输入“Hello，你好，今天天气怎么样？”→ 识别为“中英混合”并切换回复语言。
• Paraformer-v2：直接分析音频“今天食咗饭未啊？”→ 准确识别为“粤语”。

三、优缺点对比（语言识别视角）

​​Gummy-Chat-v1​​

✅ ​​优点​​

1. ​​多语言覆盖广​​：支持10+种主流语言，适合国际化对话场景（如跨境电商客服）。
2. ​​对话集成度高​​：语言识别结果可直接用于对话策略切换（如自动回复语言匹配）。
3. ​​开发便捷​​：通过对话API直接调用，无需额外部署语音处理模块。
4. 语音识别快

❌ ​​缺点​​

1. ​​方言识别弱​​：对粤语、川话等非标准发音识别准确率低（依赖标准化文本训练）。
2. ​​流程依赖ASR​​：若输入为语音，需先调用ASR转文本，增加延迟和计算成本。
3. ​​混合语言敏感度低​​：中英混杂场景易误判（如“这个price是多少？”可能误判为纯中文）。

​​Paraformer-v2​​

✅ ​​优点​​

1. ​​方言识别强​​：支持16种中文方言，适合方言用户密集区域（如粤语区智能硬件）。
2. ​​原生语音分析​​：直接处理音频流，无需ASR前置步骤，实时性更强（延迟<100ms）。
3. ​​多语言混合识别​​：可精准区分中英混读、中日混说等复杂场景（准确率>95%）。

❌ ​​缺点​​

1. ​​非对话模型​​：无法生成回复或理解意图，仅提供语言识别结果。
2. ​​硬件依赖​​：需配合麦克风、音频采集设备使用，不适合纯文本场景。
3. ​​噪音敏感​​：强背景音（如工地、街道）可能干扰识别效果，需降噪预处理。
4. 语音识别慢：由于增加对大量方言的识别，速度比较慢

四、性能指标与适用场景

​​典型应用案例​​：

• ​​Gummy-Chat-v1​​：跨境电商平台的智能客服，自动识别用户提问语言并切换回复语言。
• ​​Paraformer-v2​​：国际会议的实时语音转写系统，同步识别中英混说并生成多语言字幕。

五、总结与选型建议