基于Azure AI Speech Service实时转录麦克风与扬声器音频的问题

核心思路 #

用双线程并行处理麦克风输入和扬声器输出的音频流，分别将两路音频推送到Azure AI Speech Service进行实时转录，确保延迟控制在5秒内。麦克风转录你已实现，扬声器部分用soundcard捕获系统输出音频，再适配Azure Speech的输入格式要求。

步骤1：配置依赖与Azure参数 #

先安装所需依赖库：

pip install azure-cognitiveservices-speech soundcard numpy

在代码中配置Azure Speech的密钥和区域：

import azure.cognitiveservices.speech as speechsdk
import soundcard as sc
import numpy as np
import threading

# Azure Speech配置
SPEECH_KEY = "你的Azure语音服务密钥"
SPEECH_REGION = "你的服务区域（比如eastasia）"

步骤2：实现扬声器音频转录函数 #

利用soundcard捕获系统扬声器输出，将音频数据转换为Azure Speech要求的PCM格式（16kHz采样率、16位单声道、小端字节序），再通过PushAudioInputStream推送到服务：

def transcribe_from_speaker():
    # 获取默认扬声器输出设备（可通过sc.all_speakers()查看所有设备）
    speaker = sc.default_speaker()
    # 设置捕获参数：16kHz采样率，单声道
    sample_rate = 16000
    channels = 1

    # 初始化Azure Speech配置
    speech_config = speechsdk.SpeechConfig(subscription=SPEECH_KEY, region=SPEECH_REGION)
    speech_config.speech_recognition_language = "zh-CN"  # 按需设置语言

    # 创建音频输入流并配置格式
    audio_format = speechsdk.audio.AudioFormat(speechsdk.AudioStreamFormat(
        sample_rate=sample_rate, bits_per_sample=16, channels=channels
    ))
    push_stream = speechsdk.audio.PushAudioInputStream(audio_format)
    audio_config = speechsdk.audio.AudioConfig(stream_input=push_stream)

    # 初始化识别器并绑定回调
    recognizer = speechsdk.SpeechRecognizer(speech_config=speech_config, audio_config=audio_config)
    def recognized_cb(evt):
        if evt.result.reason == speechsdk.ResultReason.RecognizedSpeech:
            print(f"[对方音频]：{evt.result.text}")
        elif evt.result.reason == speechsdk.ResultReason.NoMatch:
            print("[对方音频]：未识别到有效语音")
    recognizer.recognized.connect(recognized_cb)

    # 启动持续识别
    recognizer.start_continuous_recognition()

    # 循环捕获音频并推送到流
    try:
        with speaker.recorder(samplerate=sample_rate, channels=channels) as rec:
            while True:
                # 调整blocksize控制延迟，1024-2048为平衡值
                data = rec.record(numframes=1024)
                # 转换为Azure要求的16位PCM字节流
                pcm_data = (data * 32767).astype(np.int16).tobytes()
                push_stream.write(pcm_data)
    except KeyboardInterrupt:
        recognizer.stop_continuous_recognition()
        push_stream.close()

步骤3：整合麦克风转录函数并启动双线程 #

假设你已实现的transcribe_from_microphone()函数适配Azure Speech实时逻辑，直接用线程同时启动两路任务：

def transcribe_from_microphone():
    # 你的麦克风转录逻辑示例框架
    speech_config = speechsdk.SpeechConfig(subscription=SPEECH_KEY, region=SPEECH_REGION)
    speech_config.speech_recognition_language = "zh-CN"
    audio_config = speechsdk.audio.AudioConfig(use_default_microphone=True)
    recognizer = speechsdk.SpeechRecognizer(speech_config=speech_config, audio_config=audio_config)

    def recognized_cb(evt):
        if evt.result.reason == speechsdk.ResultReason.RecognizedSpeech:
            print(f"[己方音频]：{evt.result.text}")
        elif evt.result.reason == speechsdk.ResultReason.NoMatch:
            print("[己方音频]：未识别到有效语音")
    recognizer.recognized.connect(recognized_cb)
    recognizer.start_continuous_recognition()

    try:
        while True:
            pass  # 保持线程运行
    except KeyboardInterrupt:
        recognizer.stop_continuous_recognition()

if __name__ == "__main__":
    # 启动扬声器转录线程
    speaker_thread = threading.Thread(target=transcribe_from_speaker, daemon=True)
    speaker_thread.start()

    # 启动麦克风转录线程
    mic_thread = threading.Thread(target=transcribe_from_microphone, daemon=True)
    mic_thread.start()

    # 主线程等待输入退出
    input("按回车停止转录...\n")

关键注意事项 #

• 延迟控制：调整rec.record(numframes=)的数值，越小延迟越低，但CPU占用会上升，1024-2048是平衡延迟和性能的合适值。
• 设备适配：如果默认扬声器捕获失败，用print(sc.all_speakers())查看所有设备，手动指定目标设备，比如sc.get_speaker("设备名称")。
• 格式匹配：必须保证捕获的音频格式与Azure Speech配置完全一致（16kHz、16位、单声道），否则会导致转录失败或乱码。
• 线程管理：设置daemon=True让子线程随主线程退出，避免程序残留进程。

内容的提问来源于stack exchange，提问作者adityapgupta211