Python实时语音转文本：麦克风流数据处理与低延迟转写实践

import speech_recognition as sr
import time

# 初始化识别器
r = sr.Recognizer()
# 选择麦克风作为音频源
microphone = sr.Microphone()

def callback(recognizer, audio):
    """
    后台监听检测到语音后调用的回调函数。
    """
    try:
        # 使用Google Web Speech API进行识别
        # 注意：这里仍需要等待完整的音频段才能进行识别
        text = recognizer.recognize_google(audio, language="zh-CN")
        print(f"检测到语音: {text}")
        # 在这里可以添加处理识别结果的逻辑，例如检查唤醒词
        if "你好" in text:
            print("唤醒词 '你好' 被检测到！")
    except sr.UnknownValueError:
        print("无法识别音频内容")
    except sr.RequestError as e:
        print(f"请求Google Speech Recognition服务失败; {e}")

# 启动后台监听
# source: 音频源 (麦克风)
# callback: 检测到语音后调用的函数
# phrase_time_limit: 每段语音的最长持续时间，防止无限等待
print("开始后台监听...")
stop_listening = r.listen_in_background(microphone, callback, phrase_time_limit=10)

# 主程序可以继续执行其他任务
# 为了演示，这里只是简单等待
while True:
    time.sleep(0.1)
    # 可以在这里添加其他逻辑，例如检查某个条件来停止监听
    # if some_condition:
    #     stop_listening(wait_for_stop=False)
    #     break

import pyaudio
from google.cloud import speech

# ... (Google Cloud认证和客户端初始化) ...

# 音频配置
RATE = 16000  # 采样率
CHUNK = 1024  # 每次读取的音频帧数

# 创建Pyaudio流
p = pyaudio.PyAudio()
stream = p.open(format=pyaudio.paInt16,
                channels=1,
                rate=RATE,
                input=True,
                frames_per_buffer=CHUNK)

# 创建Google Cloud Speech-to-Text流式请求
config = speech.RecognitionConfig(
    encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
    sample_rate_hertz=RATE,
    language_code="zh-CN",
)
streaming_config = speech.StreamingRecognitionConfig(
    config=config,
    interim_results=True  # 启用部分结果
)

# 生成音频请求迭代器
def generate_requests():
    while True:
        data = stream.read(CHUNK)
        yield speech.StreamingRecognizeRequest(audio_content=data)

# 发送请求并处理响应
responses = client.streaming_recognize(streaming_config, generate_requests())

for response in responses