基于Azure Speech SDK实现.raw实时音频转文本的Python代码求助

嘿，我完全理解你作为编程新手的迷茫——刚接触Azure Speech SDK和实时音频处理确实有点无从下手。别担心，我会给你一个完整的可运行示例代码，还会把关键环节拆解清楚，帮你搞定从持续更新的.raw音频文件里实时转录文本的需求。

首先先确认几个前提：

• 你已经正确安装了Azure Speech SDK（pip install azure-cognitiveservices-speech）
• 你的.raw文件是PCM格式（这是Speech SDK默认支持的原始音频格式），并且要明确它的参数：采样率（比如16000Hz）、位深（16位）、声道数（单声道）——这些参数必须和代码里的配置一致，不然转录会出问题

下面是完整的代码示例：

import azure.cognitiveservices.speech as speechsdk
import time
import os

# 替换成你的Azure Speech资源密钥和区域
SPEECH_KEY = "你的Azure Speech资源密钥"
SPEECH_REGION = "你的资源区域，比如eastasia"

# 替换成你的.raw文件路径
RAW_FILE_PATH = "audio_stream.raw"

def raw_file_audio_pull_stream_callback():
    """
    自定义回调函数：持续从.raw文件中读取新增的音频数据
    """
    file_size = os.path.getsize(RAW_FILE_PATH)
    current_pos = 0
    
    # 持续监听文件的新增内容
    while True:
        new_file_size = os.path.getsize(RAW_FILE_PATH)
        if new_file_size > current_pos:
            # 读取新增的字节
            with open(RAW_FILE_PATH, "rb") as f:
                f.seek(current_pos)
                chunk = f.read(new_file_size - current_pos)
                current_pos = new_file_size
                yield chunk
        # 短暂休眠，避免占用过多CPU
        time.sleep(0.01)

def main():
    # 1. 配置音频格式：必须和你的.raw文件完全匹配
    # 这里示例是16kHz采样率、16位单声道PCM，根据你的实际文件调整
    audio_format = speechsdk.audio.AudioStreamFormat(
        samples_per_second=16000,
        bits_per_sample=16,
        channels=1
    )

    # 2. 创建Pull音频流，传入自定义的回调函数
    pull_stream = speechsdk.audio.PullAudioInputStream(
        stream_format=audio_format,
        pull_callback=raw_file_audio_pull_stream_callback
    )
    audio_config = speechsdk.audio.AudioConfig(stream=pull_stream)

    # 3. 初始化Speech Recognizer
    speech_config = speechsdk.SpeechConfig(subscription=SPEECH_KEY, region=SPEECH_REGION)
    # 如果需要识别中文，设置语言：speech_config.speech_recognition_language = "zh-CN"
    speech_recognizer = speechsdk.SpeechRecognizer(
        speech_config=speech_config,
        audio_config=audio_config
    )

    print("开始实时转录，按Ctrl+C停止...")

    # 4. 定义识别结果的回调函数
    def handle_recognized_result(evt):
        if evt.result.reason == speechsdk.ResultReason.RecognizedSpeech:
            print(f"转录结果: {evt.result.text}")
        elif evt.result.reason == speechsdk.ResultReason.NoMatch:
            print(f"未识别到语音: {evt.result.no_match_details}")

    # 绑定回调事件
    speech_recognizer.recognized.connect(handle_recognized_result)
    speech_recognizer.session_started.connect(lambda evt: print("识别会话已启动"))
    speech_recognizer.session_stopped.connect(lambda evt: print("识别会话已停止"))

    # 5. 启动连续识别
    speech_recognizer.start_continuous_recognition_async()

    # 保持程序运行，直到用户按下Ctrl+C
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n正在停止识别...")
        speech_recognizer.stop_continuous_recognition_async()

if __name__ == "__main__":
    main()

关键部分解释： #

1. 音频格式配置：AudioStreamFormat的参数必须和你的.raw文件完全一致——如果你的文件是8kHz采样率或者立体声，一定要修改对应的参数，否则SDK无法正确解析音频。
2. 实时文件读取回调：raw_file_audio_pull_stream_callback函数会持续检查.raw文件的大小变化，读取新增的音频数据并返回给SDK，这样就能实现“实时”处理WebSocket写入的内容。
3. PullAudioInputStream：这就是你提到的示例里的拉流模式——我们主动从文件中拉取新数据，而不是等待SDK推送，非常适合这种“文件持续被写入”的场景。
4. 连续识别模式：start_continuous_recognition_async()会让SDK持续监听音频流，不断返回转录结果，而不是只识别一次。

你需要调整的地方： #

• 替换SPEECH_KEY和SPEECH_REGION为你自己的Azure Speech资源信息（在Azure门户里可以找到）
• 修改RAW_FILE_PATH为你的.raw文件的实际路径
• 根据你的.raw文件的音频参数，调整AudioStreamFormat里的samples_per_second、bits_per_sample、channels
• 如果需要识别中文，取消注释speech_config.speech_recognition_language = "zh-CN"这一行

测试的时候，你可以先让WebSocket服务器开始往.raw文件写入音频，然后运行这个代码，就能看到实时的转录结果啦！

备注：内容来源于stack exchange，提问作者Yuuki